From ca9bc3f49241ef678abcf1ddc83301b4622dbf71 Mon Sep 17 00:00:00 2001 From: Jiekang Tian Date: Mon, 21 Sep 2026 13:20:15 +0800 Subject: [PATCH] perf(core): parallelize data import preparation --- Cargo.lock | 4 +- Cargo.toml | 2 +- crates/core/src/state/data_import.rs | 68 +++++++++++++++---- crates/core/src/state/data_import_tests.rs | 55 +++++++++++++++ .../src/content/docs/guides/importing-data.md | 9 +-- .../docs/zh-cn/guides/importing-data.md | 8 +-- 6 files changed, 123 insertions(+), 23 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 6550ded..a928e69 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3423,9 +3423,9 @@ dependencies = [ [[package]] name = "nmr" -version = "0.1.0" +version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5749f9cb01773d2df50a3824c4591af5367da58daeaf9dd199d051a1358fda9c" +checksum = "a82e45f821d8269a728af7928832bb4a0a76fbd5df960ea785397edd07e275bd" dependencies = [ "num-complex", "rustfft", diff --git a/Cargo.toml b/Cargo.toml index f5b22ac..bc2a0b1 100644 --- a/Cargo.toml +++ b/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"] } diff --git a/crates/core/src/state/data_import.rs b/crates/core/src/state/data_import.rs index 8dd3c93..dfc312e 100644 --- a/crates/core/src/state/data_import.rs +++ b/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, + workers: usize, + prepare: &(impl Fn(&std::path::Path) -> Result + 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 +} diff --git a/crates/core/src/state/data_import_tests.rs b/crates/core/src/state/data_import_tests.rs index 503e8a4..1c82257 100644 --- a/crates/core/src/state/data_import_tests.rs +++ b/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::>(); + 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") + })); +} diff --git a/docs/src/content/docs/guides/importing-data.md b/docs/src/content/docs/guides/importing-data.md index 29b8e40..d542d8e 100644 --- a/docs/src/content/docs/guides/importing-data.md +++ b/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. diff --git a/docs/src/content/docs/zh-cn/guides/importing-data.md b/docs/src/content/docs/zh-cn/guides/importing-data.md index 1e7d686..5dd3c2a 100644 --- a/docs/src/content/docs/zh-cn/guides/importing-data.md +++ b/docs/src/content/docs/zh-cn/guides/importing-data.md @@ -42,10 +42,10 @@ CasaXPS `.txt` 按结构头内容识别，而不是只看扩展名；其他 `.tx ## 查看导入进度 -科学数据文件与采集目录在后台逐个加载，期间可以继续操作。完成的数据集会出现在 -画板上，不会切换当前页面或选择。状态栏显示当前文件及成功、失败数量；错误可在 -诊断历史中查看。追加导入会排在当前批次之后。打开、关闭或新建项目会取消旧项目 -尚未完成的导入。 +科学数据文件与采集目录会在后台加载，期间可以继续操作。一次导入多个数据集时， +它们会按原始发现顺序出现在画板上，不会切换当前页面或选择。状态栏显示当前文件及 +成功、失败数量；错误可在诊断历史中查看。追加导入会排在当前批次之后。打开、关闭 +或新建项目会取消旧项目尚未完成的导入。 NMR 导入包含默认处理，NUS 重建可能比读取文件耗时长得多。项目文件、表格预览、 ZIP 文件和手动补录采样表的导入使用各自的导入流程。