mirror of
https://github.com/storytold/lightcraft.git
synced 2026-10-24 10:09:01 -06:00
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 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
5345f0a3cf
commit
1e580466b5
@@ -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))
|
||||
}
|
||||
|
||||
|
||||
@@ -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<Photo>, SidecarNaming)], labels: &Catalog, progress: &ScanProgress) -> Vec<Check> {
|
||||
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<FolderChanges> {
|
||||
let input = SyncInput::new(s, path)?;
|
||||
|
||||
@@ -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<String> = (0..40u8).step_by(3).map(|i| dir.path(&format!("big/{i:02}.png"))).collect();
|
||||
let expect_meta: Vec<String> = (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);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user