From 1e580466b562a0e0eded88f81ae4a6d0b05a6f1c Mon Sep 17 00:00:00 2001 From: Nathaniel Ledford Date: Fri, 9 Oct 2026 17:57:32 -0400 Subject: [PATCH] Engine: scan a folder for changes concurrently and in parallel The scan waits on the disk, not the CPU, so its two halves now run at once: the look for new files (which probes on its own threads) beside the checks of the library's photos (missing file, newer sidecar), which run on several threads like the import probes do. A cancelled scan stops checking. Results keep the photos' order. Co-Authored-By: Claude Opus 5.5 --- crates/engine/src/import.rs | 2 +- crates/engine/src/sync.rs | 80 +++++++++++++++++++++++++++++---- crates/engine/src/tests_sync.rs | 33 ++++++++++++++ 3 files changed, 105 insertions(+), 10 deletions(-) diff --git a/crates/engine/src/import.rs b/crates/engine/src/import.rs index 833d92e8e..124ac3903 100644 --- a/crates/engine/src/import.rs +++ b/crates/engine/src/import.rs @@ -1044,7 +1044,7 @@ fn name_family(name: &str) -> String { /// Threads for `jobs` file reads (probes, copies) at once. #[cfg(not(target_arch = "wasm32"))] -fn workers(jobs: usize) -> usize { +pub(crate) fn workers(jobs: usize) -> usize { std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4).clamp(1, 8).min(jobs.max(1)) } diff --git a/crates/engine/src/sync.rs b/crates/engine/src/sync.rs index 2012381fa..dd2b6b9b5 100644 --- a/crates/engine/src/sync.rs +++ b/crates/engine/src/sync.rs @@ -131,9 +131,20 @@ impl SyncInput { } /// Find what changed in the folder (see the module docs). Reads the disk, never the session. +/// +/// Both halves wait on the disk (a network share answers each file slowly), so they run at once: +/// the look for new files (which probes them on its own threads) beside the checks of the +/// library's photos, which run on several threads too. Stops early when `progress.cancel` is set. pub fn scan_with(input: SyncInput, progress: &ScanProgress) -> FolderChanges { let SyncInput { folder, revision, scan, photos, labels } = input; - let out = crate::import::scan_with(scan, std::slice::from_ref(&folder), progress); + let (out, checks) = std::thread::scope(|sc| { + let photos = &photos; + let labels = &labels; + let checks = sc.spawn(move || check_photos(photos, labels, progress)); + let out = crate::import::scan_with(scan, std::slice::from_ref(&folder), progress); + // (a check that panicked counts as nothing found; the scan still reports the rest) + (out, checks.join().unwrap_or_default()) + }); let mut changes = FolderChanges { path: folder, probes: out.probes, revision, ..Default::default() }; for c in out.candidates { match (&c.duplicate, &c.error) { @@ -143,20 +154,71 @@ pub fn scan_with(input: SyncInput, progress: &ScanProgress) -> FolderChanges { (None, None) => changes.new.push(c), } } - for (p, naming) in &photos { + for ((p, _), check) in photos.iter().zip(checks) { let Some(file) = crate::cmd::missing::checked_path(p) else { continue }; - if cfg!(target_arch = "wasm32") { - continue; - } - if !Path::new(file).exists() { - changes.missing.push(FolderPhoto { id: p.id.0, path: file.to_string() }); - } else if p.copy_of.is_none() && sidecar_has_news(&labels, p, *naming) { - changes.metadata.push(FolderPhoto { id: p.id.0, path: file.to_string() }); + let found = FolderPhoto { id: p.id.0, path: file.to_string() }; + match check { + Check::Missing => changes.missing.push(found), + Check::News => changes.metadata.push(found), + Check::Nothing => {} } } changes } +/// What checking one photo of the folder found. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +enum Check { + #[default] + Nothing, + Missing, + News, +} + +fn check_photo(p: &Photo, naming: SidecarNaming, labels: &Catalog) -> Check { + let Some(file) = crate::cmd::missing::checked_path(p) else { return Check::Nothing }; + if !Path::new(file).exists() { + Check::Missing + } else if p.copy_of.is_none() && sidecar_has_news(labels, p, naming) { + Check::News + } else { + Check::Nothing + } +} + +/// [`check_photo`] for each photo, in order, on several threads (each file is a wait on the +/// disk, not work for the CPU). Photos not reached before a cancel count as nothing found. +fn check_photos(photos: &[(Arc, SidecarNaming)], labels: &Catalog, progress: &ScanProgress) -> Vec { + use std::sync::atomic::Ordering::Relaxed; + if cfg!(target_arch = "wasm32") { + return vec![Check::Nothing; photos.len()]; + } + #[cfg(not(target_arch = "wasm32"))] + { + let n = crate::import::workers(photos.len()); + if n > 1 { + let next = std::sync::atomic::AtomicUsize::new(0); + let found = std::sync::Mutex::new(vec![Check::Nothing; photos.len()]); + std::thread::scope(|sc| { + for _ in 0..n { + sc.spawn(|| { + while !progress.cancel.load(Relaxed) { + let i = next.fetch_add(1, Relaxed); + let Some((p, naming)) = photos.get(i) else { break }; + let c = check_photo(p, *naming, labels); + if let Some(slot) = found.lock().unwrap_or_else(std::sync::PoisonError::into_inner).get_mut(i) { + *slot = c; + } + } + }); + } + }); + return found.into_inner().unwrap_or_else(std::sync::PoisonError::into_inner); + } + } + photos.iter().map(|(p, naming)| if progress.cancel.load(Relaxed) { Check::Nothing } else { check_photo(p, *naming, labels) }).collect() +} + /// [`scan_with`] on the calling thread. pub fn scan(s: &mut Session, path: &str) -> Result { let input = SyncInput::new(s, path)?; diff --git a/crates/engine/src/tests_sync.rs b/crates/engine/src/tests_sync.rs index 46a77fd21..3cda16b36 100644 --- a/crates/engine/src/tests_sync.rs +++ b/crates/engine/src/tests_sync.rs @@ -237,3 +237,36 @@ fn a_scan_is_stale_once_the_library_changed() { let r = s.execute("folder.synchronize", &json!({"path": dir.path("trip")})).unwrap(); assert_eq!(r["imported"], 0, "the library changed: it looked again: {r}"); } + +#[test] +fn a_cancelled_scan_stops_checking_files() { + let dir = Scratch::new("cancel"); + let mut s = library(&dir); + std::fs::remove_file(dir.path("trip/day1/b.png")).unwrap(); + let input = crate::sync::SyncInput::new(&mut s, &dir.path("trip")).unwrap(); + let progress = crate::import::ScanProgress::default(); + progress.cancel.store(true, std::sync::atomic::Ordering::Relaxed); + let c = crate::sync::scan_with(input, &progress); + assert!(c.missing.is_empty() && c.metadata.is_empty(), "{c:?}"); +} + +#[test] +fn many_photos_are_checked_in_parallel_with_the_same_answer() { + let dir = Scratch::new("many"); + for i in 0..40u8 { + write_png(&dir.path(&format!("big/{i:02}.png")), i); + } + let mut s = Session::new().with_fs(); + s.execute("library.import", &json!({"paths": [dir.path("big")]})).unwrap(); + for i in (0..40u8).step_by(3) { + std::fs::remove_file(dir.path(&format!("big/{i:02}.png"))).unwrap(); + } + for i in (1..40u8).step_by(5) { + write_sidecar(&dir.path(&format!("big/{i:02}.png")), 2, Duration::ZERO); + } + let r = scan(&mut s, &dir.path("big")); + let expect_missing: Vec = (0..40u8).step_by(3).map(|i| dir.path(&format!("big/{i:02}.png"))).collect(); + let expect_meta: Vec = (1..40u8).step_by(5).filter(|i| i % 3 != 0).map(|i| dir.path(&format!("big/{i:02}.png"))).collect(); + assert_eq!(paths(&r, "missing"), expect_missing); + assert_eq!(paths(&r, "metadata"), expect_meta); +}