perf(core): parallelize data import preparation

Cargo.toml

@@ -38,7 +38,7 @@ publish = false
 repository = "https://github.com/nmrtist/plotx"
 
 [workspace.dependencies]
-nmr = "=0.1.0"
+nmr = "=0.1.1"
 colorous = "1"
 num-complex = "0.4"
 nalgebra = { version = "0.35", default-features = false, features = ["std"] }

crates/core/src/state/data_import.rs

@@ -65,8 +65,8 @@ impl DataImports {
 }
 
 impl PlotxApp {
-    /// Discovery, parsing, default processing and figure preparation run on one
-    /// worker. Additional gestures queue behind it instead of multiplying RAM use.
+    /// Discovery runs off-thread; preparation uses at most four CPU workers.
+    /// Additional gestures queue behind the active batch to bound memory use.
     pub fn queue_data_import(
         &mut self,
         recent: PathBuf,
@@ -107,17 +107,15 @@ impl PlotxApp {
                             return;
                         }
                     };
-                    let total = paths.len();
-                    for path in paths {
-                        if sender.send(Event::Started(path.clone(), total)).is_err() {
-                            return;
-                        }
-                        let result = plotx_io::load_path(&path)
+                    let workers = std::thread::available_parallelism()
+                        .map_or(1, usize::from)
+                        .min(4);
+                    if !prepare_paths(&paths, &sender, workers, &|path| {
+                        plotx_io::load_path(path)
                             .map_err(|error| error.to_string())
-                            .and_then(|loaded| PreparedImport::new(loaded, request.equal_scale));
-                        if sender.send(Event::Item(path, result)).is_err() {
-                            return;
-                        }
+                            .and_then(|loaded| PreparedImport::new(loaded, request.equal_scale))
+                    }) {
+                        return;
                     }
                     // A disconnected receiver means the document was closed.
                     if sender.send(Event::Finished).is_err() {
@@ -198,3 +196,49 @@ impl PlotxApp {
         busy
     }
 }
+
+/// Small windows bound both active processing and completed results awaiting the
+/// UI. Publish in discovery order so scheduling cannot change board/undo order.
+fn prepare_paths(
+    paths: &[PathBuf],
+    sender: &mpsc::SyncSender<Event>,
+    workers: usize,
+    prepare: &(impl Fn(&std::path::Path) -> Result<PreparedImport, String> + Sync),
+) -> bool {
+    for window in paths.chunks(workers.max(1)) {
+        if sender
+            .send(Event::Started(window[0].clone(), paths.len()))
+            .is_err()
+        {
+            return false;
+        }
+        let connected = std::thread::scope(|scope| {
+            let handles: Vec<_> = window
+                .iter()
+                .map(|path| {
+                    let handle = std::thread::Builder::new()
+                        .name("data-import-prepare".into())
+                        .spawn_scoped(scope, move || prepare(path));
+                    (path, handle)
+                })
+                .collect();
+            let mut connected = true;
+            for (path, handle) in handles {
+                let result = match handle {
+                    Ok(handle) => handle.join().unwrap_or_else(|_| {
+                        Err("The import worker stopped unexpectedly; retry the import.".into())
+                    }),
+                    Err(error) => Err(format!("Could not start import worker: {error}")),
+                };
+                if connected && sender.send(Event::Item(path.clone(), result)).is_err() {
+                    connected = false;
+                }
+            }
+            connected
+        });
+        if !connected {
+            return false;
+        }
+    }
+    true
+}

crates/core/src/state/data_import_tests.rs

@@ -129,3 +129,58 @@ fn discovery_runs_off_thread_and_poll_does_not_wait_for_it() {
     release.send(()).unwrap();
     assert!(!app.poll_data_import());
 }
+
+#[test]
+fn preparation_is_parallel_bounded_ordered_and_keeps_item_errors() {
+    use std::sync::{
+        Arc, Barrier,
+        atomic::{AtomicUsize, Ordering},
+    };
+    let (sender, receiver) = mpsc::sync_channel(1);
+    let barrier = Arc::new(Barrier::new(2));
+    let running = Arc::new(AtomicUsize::new(0));
+    let peak = Arc::new(AtomicUsize::new(0));
+    let peak_worker = peak.clone();
+    let worker = std::thread::spawn(move || {
+        let paths = (0..4)
+            .map(|i| PathBuf::from(i.to_string()))
+            .collect::<Vec<_>>();
+        prepare_paths(&paths, &sender, 2, &|path| {
+            let active = running.fetch_add(1, Ordering::SeqCst) + 1;
+            peak_worker.fetch_max(active, Ordering::SeqCst);
+            barrier.wait();
+            running.fetch_sub(1, Ordering::SeqCst);
+            if path == std::path::Path::new("1") {
+                Err("bad item".into())
+            } else {
+                Ok(prepared())
+            }
+        })
+    });
+    let mut items = Vec::new();
+    while let Ok(event) = receiver.recv_timeout(std::time::Duration::from_secs(10)) {
+        if let Event::Item(path, result) = event {
+            items.push((path, result.is_ok()));
+        }
+    }
+    assert!(worker.join().unwrap());
+    assert_eq!(peak.load(Ordering::SeqCst), 2);
+    assert_eq!(
+        items,
+        vec![
+            ("0".into(), true),
+            ("1".into(), false),
+            ("2".into(), true),
+            ("3".into(), true)
+        ]
+    );
+}
+
+#[test]
+fn disconnected_import_does_not_prepare_another_window() {
+    let (sender, receiver) = mpsc::sync_channel(1);
+    drop(receiver);
+    assert!(!prepare_paths(&["unused".into()], &sender, 2, &|_| {
+        panic!("closed document must not start preparation")
+    }));
+}

docs/src/content/docs/guides/importing-data.md

@@ -48,10 +48,11 @@ extension alone. Other `.txt` files continue through table import. See the
 
 ## Follow import progress
 
-Scientific data files and acquisition folders load one dataset at a time in the
-background, so you can continue working. Completed datasets appear on the board
-without changing your current page or selection. The status bar shows the current
-file and success/failure counts; review failures in the diagnostic history.
+Scientific data files and acquisition folders load in the background, so you can
+continue working. When you import multiple datasets, they appear on the board in
+their original discovery order without changing your current page or selection.
+The status bar shows the current file and success/failure counts; review failures
+in the diagnostic history.
 Additional imports wait for the current batch. Opening, closing, or creating a
 project cancels unfinished imports for the previous project.
 

docs/src/content/docs/zh-cn/guides/importing-data.md

@@ -42,10 +42,10 @@ CasaXPS `.txt` 按结构头内容识别，而不是只看扩展名；其他 `.tx
 
 ## 查看导入进度
 
-科学数据文件与采集目录在后台逐个加载，期间可以继续操作。完成的数据集会出现在
-画板上，不会切换当前页面或选择。状态栏显示当前文件及成功、失败数量；错误可在
-诊断历史中查看。追加导入会排在当前批次之后。打开、关闭或新建项目会取消旧项目
-尚未完成的导入。
+科学数据文件与采集目录会在后台加载，期间可以继续操作。一次导入多个数据集时，
+它们会按原始发现顺序出现在画板上，不会切换当前页面或选择。状态栏显示当前文件及
+成功、失败数量；错误可在诊断历史中查看。追加导入会排在当前批次之后。打开、关闭
+或新建项目会取消旧项目尚未完成的导入。
 
 NMR 导入包含默认处理，NUS 重建可能比读取文件耗时长得多。项目文件、表格预览、
 ZIP 文件和手动补录采样表的导入使用各自的导入流程。