mirror of
https://github.com/storytold/lightcraft.git
synced 2026-10-31 21:48:53 -06:00
Issue #345: engine task registry (activity)
A registry of the background tasks in flight on Session: a job holds a TaskGuard while it runs (dropping it, also during a panic unwind, removes the row), adopts its existing cancel flag, and reports done/total in a unit the frontend words. Groundwork for one progress stack for imports, exports, preview builds and the other long jobs. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Vyrodov
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
5436951299
commit
533a382619
@@ -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<Inner>,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct Inner {
|
||||
/// Oldest first.
|
||||
tasks: Mutex<Vec<Arc<Entry>>>,
|
||||
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<AtomicBool>),
|
||||
}
|
||||
|
||||
/// 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<String>,
|
||||
cancel: Option<Arc<AtomicBool>>,
|
||||
cancellable: AtomicBool,
|
||||
}
|
||||
|
||||
fn lock<T>(m: &Mutex<T>) -> 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<TaskInfo> {
|
||||
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<TaskInfo> {
|
||||
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<Entry>,
|
||||
}
|
||||
|
||||
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<Arc<AtomicBool>> {
|
||||
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<Entry>,
|
||||
}
|
||||
|
||||
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<_>>(), 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));
|
||||
}
|
||||
}
|
||||
@@ -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<sync::FolderChanges>,
|
||||
/// 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<std::sync::Arc<cmd::previews::PreviewBuild>>,
|
||||
/// 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,
|
||||
|
||||
Reference in New Issue
Block a user