diff --git a/crates/engine/src/activity.rs b/crates/engine/src/activity.rs new file mode 100644 index 000000000..654b6db44 --- /dev/null +++ b/crates/engine/src/activity.rs @@ -0,0 +1,391 @@ +//! Background tasks in flight (issue #345): imports, exports, preview builds, downloads, the face scan… Each one +//! registers here while it runs, so every frontend can show them in one place (the desktop app's activity stack) and +//! cancel the ones that can be cancelled (`activity.list`, `activity.cancel`). +//! +//! A job holds a [`TaskGuard`] for as long as it runs; dropping it — when the job ends, fails or its thread unwinds +//! from a panic — removes the row, so a dead worker never leaves a stuck bar. Jobs keep the cancel flag they already +//! check between files or photos: the guard adopts it ([`Cancel::Flag`]), so the stack's ✕, `activity.cancel` and the +//! job's own Cancel all set the same atomic. Numbers are worded by the frontend from `done`, `total` and [`Unit`]; a +//! detail set here is a name (a file, a model), never a sentence, so the frontend can translate everything else. + +use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; + +use serde::Serialize; + +/// The longest label or detail kept, in characters (a file name can be anything). +const MAX_TEXT: usize = 200; + +/// The background tasks in flight. Cheap to clone (one `Arc`), so worker threads can hold it. +#[derive(Clone, Default)] +pub struct Activity { + inner: Arc, +} + +#[derive(Default)] +struct Inner { + /// Oldest first. + tasks: Mutex>>, + next: AtomicU64, +} + +/// How a task can be cancelled. +pub enum Cancel { + /// Not at all (no ✕): the face scan, AI Denoise. + No, + /// With a flag of its own ([`TaskGuard::cancel_flag`]). + Yes, + /// With the job's existing flag. + Flag(Arc), +} + +/// How the frontend words `done` / `total`: "3 of 25", "12 of 340 MB", "40 %". +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize)] +#[serde(rename_all = "lowercase")] +pub enum Unit { + #[default] + Count, + Bytes, + Percent, +} + +impl Unit { + fn from_u8(v: u8) -> Unit { + match v { + 1 => Unit::Bytes, + 2 => Unit::Percent, + _ => Unit::Count, + } + } + + fn to_u8(self) -> u8 { + match self { + Unit::Count => 0, + Unit::Bytes => 1, + Unit::Percent => 2, + } + } +} + +/// One task as the frontends see it (`activity.list`). +#[derive(Clone, Debug, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct TaskInfo { + pub id: u64, + /// A stable id: `import`, `export`, `scan`, `previews`, `smartPreviews`, `lightroom`, `merge`, `download`, + /// `faces`, `denoise`, `findMissing`. + pub kind: &'static str, + /// English; the frontend translates it. + pub label: String, + pub done: u64, + /// 0 while the amount of work isn't known yet (an indeterminate bar). + pub total: u64, + pub unit: Unit, + /// The file, photo or model being worked on, if any. + pub detail: String, + pub cancellable: bool, + /// Cancel was asked for; the job stops at its next check. + pub cancelling: bool, + pub age_ms: u64, +} + +struct Entry { + id: u64, + kind: &'static str, + label: String, + started: web_time::Instant, + done: AtomicU64, + total: AtomicU64, + unit: AtomicU8, + detail: Mutex, + cancel: Option>, + cancellable: AtomicBool, +} + +fn lock(m: &Mutex) -> MutexGuard<'_, T> { + m.lock().unwrap_or_else(PoisonError::into_inner) +} + +fn capped(text: &str) -> String { + text.chars().take(MAX_TEXT).collect() +} + +impl Entry { + fn is_cancellable(&self) -> bool { + self.cancel.is_some() && self.cancellable.load(Ordering::Relaxed) + } + + fn is_cancelled(&self) -> bool { + self.cancel.as_ref().is_some_and(|c| c.load(Ordering::Relaxed)) + } + + fn progress(&self, done: u64, total: u64) { + self.total.store(total, Ordering::Relaxed); + self.done.store(if total > 0 { done.min(total) } else { done }, Ordering::Relaxed); + } + + fn detail(&self, text: &str) { + *lock(&self.detail) = capped(text); + } + + fn info(&self) -> TaskInfo { + TaskInfo { + id: self.id, + kind: self.kind, + label: self.label.clone(), + done: self.done.load(Ordering::Relaxed), + total: self.total.load(Ordering::Relaxed), + unit: Unit::from_u8(self.unit.load(Ordering::Relaxed)), + detail: lock(&self.detail).clone(), + cancellable: self.is_cancellable(), + cancelling: self.is_cancelled(), + age_ms: u64::try_from(self.started.elapsed().as_millis()).unwrap_or(u64::MAX), + } + } +} + +impl Activity { + /// Register a task; it is listed until the returned guard is dropped. `label` is English (the frontend + /// translates it). + pub fn start(&self, kind: &'static str, label: &str, cancel: Cancel) -> TaskGuard { + let id = self.inner.next.fetch_add(1, Ordering::Relaxed).saturating_add(1); + let cancel = match cancel { + Cancel::No => None, + Cancel::Yes => Some(Arc::new(AtomicBool::new(false))), + Cancel::Flag(flag) => Some(flag), + }; + let entry = Arc::new(Entry { + id, + kind, + label: capped(label), + started: web_time::Instant::now(), + done: AtomicU64::new(0), + total: AtomicU64::new(0), + unit: AtomicU8::new(Unit::Count.to_u8()), + detail: Mutex::new(String::new()), + cancellable: AtomicBool::new(cancel.is_some()), + cancel, + }); + lock(&self.inner.tasks).push(entry.clone()); + TaskGuard { activity: self.clone(), entry } + } + + /// Every task in flight, oldest first. + pub fn list(&self) -> Vec { + lock(&self.inner.tasks).iter().map(|e| e.info()).collect() + } + + /// Ask task `id` to stop (it stops at its next check). An unknown id (the task may have just ended) or a task + /// that can't be cancelled is an error. + pub fn cancel(&self, id: u64) -> Result<(), String> { + let tasks = lock(&self.inner.tasks); + let entry = tasks.iter().find(|e| e.id == id).ok_or_else(|| format!("no task {id}"))?; + match (&entry.cancel, entry.is_cancellable()) { + (Some(flag), true) => { + flag.store(true, Ordering::Relaxed); + Ok(()) + } + _ => Err(format!("{} can't be cancelled", entry.label)), + } + } + + /// Ask every cancellable task to stop; returns how many were asked. + pub fn cancel_all(&self) -> usize { + let tasks = lock(&self.inner.tasks); + let mut n = 0; + for e in tasks.iter().filter(|e| e.is_cancellable()) { + if let Some(flag) = &e.cancel { + flag.store(true, Ordering::Relaxed); + n += 1; + } + } + n + } + + /// Tasks that quitting would cut short: cancellable ones not already stopping. + pub fn running_cancellable(&self) -> Vec { + lock(&self.inner.tasks).iter().filter(|e| e.is_cancellable() && !e.is_cancelled()).map(|e| e.info()).collect() + } +} + +/// A task's place in the list, held by whoever owns the job; dropping it removes the row. +pub struct TaskGuard { + activity: Activity, + entry: Arc, +} + +impl TaskGuard { + pub fn id(&self) -> u64 { + self.entry.id + } + + /// `done` of `total` (`total` 0 = not known yet). `done` is kept at most `total`. + pub fn progress(&self, done: u64, total: u64) { + self.entry.progress(done, total); + } + + /// The file, photo or model being worked on (a name, not a sentence). + pub fn detail(&self, text: &str) { + self.entry.detail(text); + } + + pub fn set_unit(&self, unit: Unit) { + self.entry.unit.store(unit.to_u8(), Ordering::Relaxed); + } + + /// A step that must not be cut short (the Lightroom import's commit) turns cancelling off for its duration. + pub fn set_cancellable(&self, yes: bool) { + self.entry.cancellable.store(yes, Ordering::Relaxed); + } + + pub fn cancel_flag(&self) -> Option> { + self.entry.cancel.clone() + } + + pub fn is_cancelled(&self) -> bool { + self.entry.is_cancelled() + } + + /// For progress from another thread; it does not keep the row. + pub fn handle(&self) -> TaskHandle { + TaskHandle { entry: self.entry.clone() } + } +} + +impl Drop for TaskGuard { + fn drop(&mut self) { + let id = self.entry.id; + lock(&self.activity.inner.tasks).retain(|e| e.id != id); + } +} + +impl std::fmt::Debug for TaskGuard { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("TaskGuard").field("id", &self.entry.id).field("kind", &self.entry.kind).finish() + } +} + +/// Progress for a task from another thread (see [`TaskGuard::handle`]). +#[derive(Clone)] +pub struct TaskHandle { + entry: Arc, +} + +impl TaskHandle { + pub fn progress(&self, done: u64, total: u64) { + self.entry.progress(done, total); + } + + pub fn detail(&self, text: &str) { + self.entry.detail(text); + } + + pub fn is_cancelled(&self) -> bool { + self.entry.is_cancelled() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + + #[test] + fn guard_drop_removes_the_row() { + let a = Activity::default(); + let g = a.start("export", "Exporting", Cancel::Yes); + assert_eq!(a.list().len(), 1); + drop(g); + assert!(a.list().is_empty()); + } + + #[test] + fn guard_drop_on_panic_unwind_removes_the_row() { + let a = Activity::default(); + let a2 = a.clone(); + let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || { + let _g = a2.start("import", "Importing", Cancel::Yes); + panic!("synthetic worker panic"); + })); + assert!(r.is_err()); + assert!(a.list().is_empty()); + } + + #[test] + fn cancel_sets_the_adopted_flag() { + let a = Activity::default(); + let flag = Arc::new(AtomicBool::new(false)); + let g = a.start("export", "Exporting", Cancel::Flag(flag.clone())); + a.cancel(g.id()).unwrap(); + assert!(flag.load(Ordering::Relaxed) && g.is_cancelled()); + assert!(a.list()[0].cancelling); + } + + #[test] + fn cancel_errors_instead_of_panicking() { + let a = Activity::default(); + assert!(a.cancel(42).is_err()); + let g = a.start("faces", "Finding faces", Cancel::No); + assert!(a.cancel(g.id()).is_err()); + let h = a.start("lightroom", "Importing Lightroom catalog", Cancel::Yes); + h.set_cancellable(false); + assert!(a.cancel(h.id()).is_err()); + assert!(!h.is_cancelled()); + } + + #[test] + fn cancel_all_counts_only_cancellable() { + let a = Activity::default(); + let (_x, _y, _z) = (a.start("export", "E", Cancel::Yes), a.start("import", "I", Cancel::Yes), a.start("faces", "F", Cancel::No)); + assert_eq!(a.cancel_all(), 2); + assert!(a.running_cancellable().is_empty()); + } + + #[test] + fn progress_is_clamped_and_detail_capped() { + let a = Activity::default(); + let g = a.start("export", "E", Cancel::Yes); + g.progress(12, 10); + g.detail(&"é".repeat(300)); + let t = &a.list()[0]; + assert_eq!((t.done, t.total), (10, 10)); + assert_eq!(t.detail.chars().count(), 200); + g.progress(5, 0); // indeterminate keeps done as given + assert_eq!(a.list()[0].total, 0); + assert_eq!(a.list()[0].unit, Unit::Count); + g.set_unit(Unit::Bytes); + assert_eq!(a.list()[0].unit, Unit::Bytes); + } + + #[test] + fn ids_are_unique_and_list_is_oldest_first() { + let a = Activity::default(); + let (g1, g2) = (a.start("export", "A", Cancel::Yes), a.start("import", "B", Cancel::Yes)); + assert!(g2.id() > g1.id()); + assert_eq!(a.list().iter().map(|t| t.id).collect::>(), vec![g1.id(), g2.id()]); + } + + #[test] + fn poisoned_lock_still_works() { + let a = Activity::default(); + let a2 = a.clone(); + let _ = std::thread::spawn(move || { + let _held = a2.inner.tasks.lock().unwrap(); + panic!("synthetic panic while holding the task list"); + }) + .join(); + assert!(a.inner.tasks.is_poisoned()); + let _g = a.start("export", "E", Cancel::Yes); + assert_eq!(a.list().len(), 1); + } + + #[test] + fn handle_updates_from_another_thread() { + let a = Activity::default(); + let g = a.start("previews", "Building previews", Cancel::Yes); + let h = g.handle(); + std::thread::spawn(move || h.progress(3, 7)).join().unwrap(); + assert_eq!((a.list()[0].done, a.list()[0].total), (3, 7)); + } +} diff --git a/crates/engine/src/lib.rs b/crates/engine/src/lib.rs index 22faff70f..c507f0a0a 100644 --- a/crates/engine/src/lib.rs +++ b/crates/engine/src/lib.rs @@ -14,6 +14,7 @@ #[cfg(not(feature = "denoise"))] extern crate lightcraft_denoise_core as lightcraft_denoise; +pub mod activity; pub mod availability; mod camera_preview; pub mod camera_profiles; @@ -261,6 +262,9 @@ pub struct Session { /// The last Synchronize Folder scan, which `folder.synchronize` acts on while it is current /// (see [`Session::take_folder_changes`]). pub folder_changes: Option, + /// Background tasks in flight (imports, exports, preview builds…), for the activity stack and + /// `activity.list` / `activity.cancel`. + pub activity: activity::Activity, /// The last (or running) Build Previews. pub preview_build: Option>, /// Develop defaults applied on import (persisted in prefs.json). @@ -345,6 +349,7 @@ impl Session { before: Default::default(), import_probes: Default::default(), folder_changes: None, + activity: Default::default(), preview_build: None, import_defaults: import::ImportDefaults::default(), cache_mb: 0,