From 3b7870bee1d6d02fdf75b272a9e45eec92684a08 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Tue, 6 Oct 2026 15:41:38 -0400 Subject: [PATCH] Background jobs: engine job API, progress and cancel (part of #210) (#289) * Test: retry the GPU startup sentinel check while a forked child briefly shares the lock monitor_profile tests spawn osascript in the same binary; between fork and exec the child holds a copy of the marker file descriptor, so flock reports WouldBlock and begin() returns Busy. Co-Authored-By: Claude Opus 5.5 * Background jobs: engine job API, progress and cancel (part of #210) (#289) --------- Co-authored-by: Claude Opus 5.5 --- apps/photocraft/src/gpu_startup.rs | 11 +- crates/algo/src/content_aware.rs | 73 ++- crates/algo/src/inpaint.rs | 50 +- crates/algo/src/lib.rs | 55 +- crates/algo/src/poisson.rs | 40 +- crates/algo/src/seam.rs | 59 +- crates/algo/src/tests_ext.rs | 22 + crates/automation/src/headless.rs | 17 +- crates/automation/src/rpc.rs | 42 +- crates/automation/src/server.rs | 50 +- crates/engine/src/commands.rs | 1 + crates/engine/src/edit_menu_cmds.rs | 172 ++++-- crates/engine/src/filters.rs | 74 ++- crates/engine/src/jobs.rs | 757 ++++++++++++++++++++++++ crates/engine/src/jobs_tests.rs | 333 +++++++++++ crates/engine/src/lib.rs | 58 +- crates/engine/src/photo_cmds.rs | 185 +++--- crates/engine/src/preset_import_cmds.rs | 36 +- crates/io/src/abr_map.rs | 30 +- crates/io/src/lib.rs | 23 +- crates/io/src/psd_import.rs | 44 +- crates/io/tests/cancel.rs | 62 ++ crates/raster/src/interrupt.rs | 108 ++++ crates/raster/src/lib.rs | 3 + 24 files changed, 2063 insertions(+), 242 deletions(-) create mode 100644 crates/engine/src/jobs.rs create mode 100644 crates/engine/src/jobs_tests.rs create mode 100644 crates/io/tests/cancel.rs create mode 100644 crates/raster/src/interrupt.rs diff --git a/apps/photocraft/src/gpu_startup.rs b/apps/photocraft/src/gpu_startup.rs index 1f4af46bfe..d84744deaa 100644 --- a/apps/photocraft/src/gpu_startup.rs +++ b/apps/photocraft/src/gpu_startup.rs @@ -513,7 +513,16 @@ mod tests { assert!(none.is_none()); // The process dies in the driver: the lock goes, the marker stays. drop(s); - let (prev, s) = Sentinel::begin(&dir); + // Another test in this binary spawns a process: between its fork and exec the child briefly + // shares this file's flock, so `begin` can see the lock as busy. Retry for a moment. + let (mut prev, mut s) = Sentinel::begin(&dir); + for _ in 0..50 { + if prev != Previous::Busy { + break; + } + std::thread::sleep(std::time::Duration::from_millis(10)); + (prev, s) = Sentinel::begin(&dir); + } let m = prev.crashed().cloned().expect("crash detected"); assert_eq!((m.adapter.as_str(), m.tried()), ("GPU", Some(Vulkan))); // This start succeeds: the next one finds nothing. diff --git a/crates/algo/src/content_aware.rs b/crates/algo/src/content_aware.rs index 61b1fd098e..95e03b65f3 100644 --- a/crates/algo/src/content_aware.rs +++ b/crates/algo/src/content_aware.rs @@ -17,8 +17,8 @@ //! //! Buffers are interleaved `w × h × ch` normalised floats (any model/depth); deterministic. -use crate::inpaint::{CompleteParams, complete}; -use crate::poisson::membrane_fill; +use crate::inpaint::{CompleteParams, complete_with}; +use crate::poisson::membrane_fill_with; /// Options for [`fill`]. #[derive(Clone, Debug, PartialEq)] @@ -161,11 +161,27 @@ fn box_blur(w: usize, h: usize, ch: usize, v: &[f32], r: usize) -> Vec { /// Returns the full buffer (unchanged outside the hole). Falls back to the membrane fill when /// nothing can be sampled. pub fn fill(w: usize, h: usize, ch: usize, img: &[f32], hole: &[bool], source: &[bool], opts: &FillOptions) -> Vec { + // Never cancelled; the fallback (the input unchanged) is unreachable. + fill_with(w, h, ch, img, hole, source, opts, &photocraft_raster::Interrupt::NONE).unwrap_or_else(|_| img.to_vec()) +} + +/// [`fill`] that can be cancelled (see [`crate::inpaint::complete_with`]) and reports progress. +#[allow(clippy::too_many_arguments)] +pub fn fill_with( + w: usize, + h: usize, + ch: usize, + img: &[f32], + hole: &[bool], + source: &[bool], + opts: &FillOptions, + ctl: &photocraft_raster::Interrupt, +) -> Result, photocraft_raster::Cancelled> { assert_eq!(img.len(), w * h * ch); assert_eq!(hole.len(), w * h); assert_eq!(source.len(), w * h); if !hole.iter().any(|h| *h) { - return img.to_vec(); + return Ok(img.to_vec()); } let params = CompleteParams { seed: opts.seed, ..Default::default() }; let gap = 2 * params.patch_radius + 2; @@ -208,7 +224,11 @@ pub fn fill(w: usize, h: usize, ch: usize, img: &[f32], hole: &[bool], source: & } oy += v.h + gap; } - let filled = complete(mw, mh, ch, &mimg, &unknown, ¶ms); + ctl.check()?; + // The completion is nearly all the work: it reports 0–90 %. + let cancel = || ctl.cancelled(); + let progress = |f: f32| ctl.progress(f * 0.9); + let filled = complete_with(mw, mh, ch, &mimg, &unknown, ¶ms, &photocraft_raster::Interrupt::new(&cancel, &progress))?; let mut out = img.to_vec(); match filled { Some(f) => { @@ -221,16 +241,28 @@ pub fn fill(w: usize, h: usize, ch: usize, img: &[f32], hole: &[bool], source: & } } } - None => out = membrane_fill(w, h, ch, img, hole), + None => out = membrane_fill_with(w, h, ch, img, hole, ctl)?, } + ctl.check()?; if opts.color_adaptation > 0.0 { - adapt_colors(w, h, ch, img, hole, &mut out, opts.color_adaptation); + adapt_colors(w, h, ch, img, hole, &mut out, opts.color_adaptation, ctl)?; } - out + ctl.progress(1.0); + Ok(out) } /// Pull the fill's low frequencies towards the membrane interpolation of the hole boundary. -fn adapt_colors(w: usize, h: usize, ch: usize, img: &[f32], hole: &[bool], out: &mut [f32], k: f32) { +#[allow(clippy::too_many_arguments)] +fn adapt_colors( + w: usize, + h: usize, + ch: usize, + img: &[f32], + hole: &[bool], + out: &mut [f32], + k: f32, + ctl: &photocraft_raster::Interrupt, +) -> Result<(), photocraft_raster::Cancelled> { let (mut x0, mut y0, mut x1, mut y1) = (w, h, 0, 0); for y in 0..h { for x in 0..w { @@ -240,7 +272,9 @@ fn adapt_colors(w: usize, h: usize, ch: usize, img: &[f32], hole: &[bool], out: } } let r = ((x1 - x0).max(y1 - y0) / 6).clamp(1, 32); - let membrane = membrane_fill(w, h, ch, img, hole); + let membrane = membrane_fill_with(w, h, ch, img, hole, ctl)?; + ctl.progress(0.97); + ctl.check()?; let low = box_blur(w, h, ch, out, r); for (i, &hl) in hole.iter().enumerate() { if hl { @@ -250,6 +284,7 @@ fn adapt_colors(w: usize, h: usize, ch: usize, img: &[f32], hole: &[bool], out: } } } + Ok(()) } #[cfg(test)] @@ -263,6 +298,26 @@ mod tests { (img, hole) } + #[test] + fn cancelled_fill_stops_and_progress_reaches_one() { + use std::sync::atomic::{AtomicU32, Ordering}; + let (w, h) = (60, 40); + let (img, hole) = stripes(w, h); + let source = vec![true; w * h]; + let yes = || true; + let cancelled = photocraft_raster::Interrupt::cancel_only(&yes); + assert!(fill_with(w, h, 1, &img, &hole, &source, &FillOptions::default(), &cancelled).is_err()); + let last = AtomicU32::new(0); + let no = || false; + let progress = |f: f32| { + assert!(f >= f32::from_bits(last.load(Ordering::Relaxed)), "progress went backwards"); + last.store(f.to_bits(), Ordering::Relaxed); + }; + let out = fill_with(w, h, 1, &img, &hole, &source, &FillOptions::default(), &photocraft_raster::Interrupt::new(&no, &progress)).unwrap(); + assert_eq!(f32::from_bits(last.load(Ordering::Relaxed)), 1.0); + assert_eq!(out, fill(w, h, 1, &img, &hole, &source, &FillOptions::default()), "same result as the plain fill"); + } + #[test] fn fills_only_the_hole_with_known_texture() { let (w, h) = (48, 32); diff --git a/crates/algo/src/inpaint.rs b/crates/algo/src/inpaint.rs index dca652a603..fe34d3c8f4 100644 --- a/crates/algo/src/inpaint.rs +++ b/crates/algo/src/inpaint.rs @@ -164,6 +164,8 @@ struct Solver<'a> { valid_list: Vec<(i32, i32)>, /// Target centres (patch overlaps the hole), row-major. targets: Vec, + /// Cancellation: checked per row band; a cancelled solve's output is discarded. + ctl: photocraft_raster::Interrupt<'a>, } impl Solver<'_> { @@ -211,6 +213,9 @@ impl Solver<'_> { let w = self.w; let reverse = pass % 2 == 1; let band_fn = |bi: usize, nn: &mut [(i32, i32)], co: &mut [f32]| { + if self.ctl.cancelled() { + return; + } let mut rng = Rng::new(seed ^ (bi as u64).wrapping_mul(0x2545_F491_4F6C_DD1D) ^ ((pass as u64) << 40)); let y0 = bi * BAND; let rows = nn.len() / w; @@ -292,6 +297,9 @@ impl Solver<'_> { let img = &*self.img; let row = |y: usize| -> Vec<(usize, Vec)> { let mut out = Vec::new(); + if self.ctl.cancelled() { + return out; + } for x in 0..w { if !self.hole[y * w + x] { continue; @@ -349,9 +357,24 @@ impl Solver<'_> { /// Content-aware completion of `hole` (see module docs). Returns `None` when there is no fully-known /// patch to copy from (the caller can fall back to [`membrane_fill`]). pub fn complete(w: usize, h: usize, ch: usize, img: &[f32], hole: &[bool], p: &CompleteParams) -> Option> { + // Never cancelled, so never `Err`. + complete_with(w, h, ch, img, hole, p, &photocraft_raster::Interrupt::NONE).unwrap_or(None) +} + +/// [`complete`] that can be cancelled (checked per EM iteration, PatchMatch pass and row band) +/// and reports progress (by pyramid level). `Err(Cancelled)` when `ctl` was cancelled. +pub fn complete_with( + w: usize, + h: usize, + ch: usize, + img: &[f32], + hole: &[bool], + p: &CompleteParams, + ctl: &photocraft_raster::Interrupt, +) -> Result>, photocraft_raster::Cancelled> { assert_eq!(img.len(), w * h * ch); assert_eq!(hole.len(), w * h); - let Some((bx0, by0, bx1, by1)) = hole_bbox(w, h, hole) else { return Some(img.to_vec()) }; + let Some((bx0, by0, bx1, by1)) = hole_bbox(w, h, hole) else { return Ok(Some(img.to_vec())) }; let r = p.patch_radius.max(1); let psz = 2 * r + 1; // Pyramid: shrink until the hole is a few patches across. @@ -359,6 +382,7 @@ pub fn complete(w: usize, h: usize, ch: usize, img: &[f32], hole: &[bool], p: &C let mut ext = (bx1 - bx0).max(by1 - by0); let mut exts = vec![ext]; while let Some(l) = levels.last() { + ctl.check()?; if ext <= 2 * psz || l.w / 2 < 3 * psz || l.h / 2 < 3 * psz { break; } @@ -370,7 +394,10 @@ pub fn complete(w: usize, h: usize, ch: usize, img: &[f32], hole: &[bool], p: &C let nlev = levels.len(); let mut nnf: Vec<(i32, i32)> = Vec::new(); let mut prev: Option<(usize, usize, Vec)> = None; + let total_px: usize = levels.iter().map(|l| l.w * l.h).sum(); + let mut done_px = 0usize; for li in (0..nlev).rev() { + ctl.check()?; let Level { w: lw, h: lh, img: mut limg, hole: lhole } = levels[li].clone(); // Initialise the hole: membrane at the coarsest level, upsampled result elsewhere. match &prev { @@ -393,6 +420,9 @@ pub fn complete(w: usize, h: usize, ch: usize, img: &[f32], hole: &[bool], p: &C let mut valid_list = Vec::new(); let mut targets = vec![false; lw * lh]; for y in 0..lh as i32 { + if y % 32 == 0 { + ctl.check()?; + } for x in 0..lw as i32 { let cnt = window_count(&hs, lw, lh, x - ri, y - ri, x + ri + 1, y + ri + 1); let inside = x >= ri && y >= ri && x + ri < lw as i32 && y + ri < lh as i32; @@ -406,19 +436,22 @@ pub fn complete(w: usize, h: usize, ch: usize, img: &[f32], hole: &[bool], p: &C } if valid_list.is_empty() { if li == 0 { - return None; + return Ok(None); } prev = Some((lw, lh, limg)); nnf.clear(); continue; } - let mut solver = Solver { w: lw, h: lh, ch, r: ri, img: &mut limg, hole: &lhole, valid, valid_list, targets }; + let mut solver = Solver { w: lw, h: lh, ch, r: ri, img: &mut limg, hole: &lhole, valid, valid_list, targets, ctl: *ctl }; // NNF init: upsampled from the coarser level where possible, random otherwise. let mut rng = Rng::new(p.seed ^ (li as u64) << 20); let pw_prev = prev.as_ref().map_or(0, |p| p.0); let old = std::mem::take(&mut nnf); let mut new_nnf = vec![(0i32, 0i32); lw * lh]; for y in 0..lh { + if y % 32 == 0 { + ctl.check()?; + } for x in 0..lw { if !solver.targets[y * lw + x] { continue; @@ -451,6 +484,9 @@ pub fn complete(w: usize, h: usize, ch: usize, img: &[f32], hole: &[bool], p: &C for it in 0..em { // Costs change as hole pixels are re-estimated: refresh before each PatchMatch. for (i, c) in cost.iter_mut().enumerate() { + if i % (lw * 32).max(1) == 0 { + ctl.check()?; + } if solver.targets[i] { *c = solver.dist(((i % lw) as i32, (i / lw) as i32), nnf[i], f32::INFINITY); } @@ -464,13 +500,19 @@ pub fn complete(w: usize, h: usize, ch: usize, img: &[f32], hole: &[bool], p: &C (4 * psz as i32).min(full_search) }; for pass in 0..pm_iters { + ctl.check()?; solver.patchmatch(&mut nnf, &mut cost, pass + it * pm_iters, p.seed.wrapping_add((li * 1000 + it) as u64), radius); } + ctl.check()?; solver.vote(&nnf, &cost); + ctl.check()?; } + // Finer levels cost ~4× the previous one: weight progress by pixel count. + done_px += lw * lh; + ctl.progress(done_px as f32 / total_px.max(1) as f32); prev = Some((lw, lh, limg)); } - prev.map(|(_, _, img)| img) + Ok(prev.map(|(_, _, img)| img)) } /// Proximity match: the displacement `(dx, dy)` (within `max_radius`) whose surroundings best match the diff --git a/crates/algo/src/lib.rs b/crates/algo/src/lib.rs index 46ea63750c..c41a41deaa 100644 --- a/crates/algo/src/lib.rs +++ b/crates/algo/src/lib.rs @@ -861,6 +861,20 @@ pub fn apply_in(surface: &Surface, params: &FilterParams, area: Rect, bounds: Re apply_tiled(surface, params, area.intersect(&extent), bounds, selection, auto_tile(params), Some(extent)) } +/// [`apply_in`] that can be cancelled (checked before each tile) and reports progress per tile +/// group. `None` when `ctl` was cancelled; the input surface is never modified. +pub fn apply_in_with( + surface: &Surface, + params: &FilterParams, + area: Rect, + bounds: Rect, + selection: Option<&Surface>, + extent: Rect, + ctl: &photocraft_raster::Interrupt, +) -> Option { + apply_tiled_with(surface, params, area.intersect(&extent), bounds, selection, auto_tile(params), Some(extent), ctl) +} + /// Results do not depend on the tiling, so wide-halo filters use bigger tiles to keep the /// re-read margin (and its cost) below ~2× the tile area. fn auto_tile(params: &FilterParams) -> i32 { @@ -880,9 +894,26 @@ pub fn apply_tiled( tile: i32, extent: Option, ) -> Surface { + // Never cancelled, so always `Some`; the fallback (the input unchanged) is unreachable. + apply_tiled_with(surface, params, area, bounds, selection, tile, extent, &photocraft_raster::Interrupt::NONE).unwrap_or_else(|| surface.clone()) +} + +/// [`apply_tiled`] with cancellation (checked before each tile, so a cancel takes effect within +/// one tile's work) and progress (after each group of tiles). `None` when cancelled. +#[allow(clippy::too_many_arguments)] +pub fn apply_tiled_with( + surface: &Surface, + params: &FilterParams, + area: Rect, + bounds: Rect, + selection: Option<&Surface>, + tile: i32, + extent: Option, + ctl: &photocraft_raster::Interrupt, +) -> Option { let mut out = surface.clone(); if area.is_empty() { - return out; + return Some(out); } let fmt = surface.format(); let ctx = Ctx { bounds, mode: fmt.mode, alpha: fmt.alpha }; @@ -898,7 +929,19 @@ pub fn apply_tiled( } y += tile; } + // Tiles finished so far, for progress reported per tile (from any worker thread). + let finished = std::sync::atomic::AtomicUsize::new(0); + let total = tiles.len().max(1); let run = |t: &Rect| -> (Rect, Vec) { + // Cancelled: skip the remaining tiles of the group (the result is discarded). + if ctl.cancelled() { + return (*t, Vec::new()); + } + let tick = || { + let n = finished.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1; + // The last few percent are the writes after each group. + ctl.progress(0.98 * n as f32 / total as f32); + }; let owned; let src = match (&shared, halo) { (Some(s), _) => s, @@ -931,6 +974,7 @@ pub fn apply_tiled( } } } + tick(); (*t, data) }; // Tiles are filtered in groups of about RESULT_BUDGET bytes (at least one per core) and each @@ -943,6 +987,9 @@ pub fn apply_tiled( let per_tile = (tile.max(1) as usize).pow(2) * fmt.channels() * std::mem::size_of::(); let group = (RESULT_BUDGET / per_tile.max(1)).max(threads).max(1); for chunk in tiles.chunks(group) { + if ctl.cancelled() { + return None; + } #[cfg(not(target_arch = "wasm32"))] let results: Vec<(Rect, Vec)> = { use rayon::prelude::*; @@ -950,12 +997,16 @@ pub fn apply_tiled( }; #[cfg(target_arch = "wasm32")] let results: Vec<(Rect, Vec)> = chunk.iter().map(run).collect(); + if ctl.cancelled() { + return None; + } for (t, data) in results { out.write_region(t, &data); } } out.prune(); - out + ctl.progress(1.0); + Some(out) } /// Bytes of float filter results held at once by [`apply_tiled`]. diff --git a/crates/algo/src/poisson.rs b/crates/algo/src/poisson.rs index 55b0bd8a23..1986f6b6b6 100644 --- a/crates/algo/src/poisson.rs +++ b/crates/algo/src/poisson.rs @@ -31,10 +31,13 @@ const FINE_OMEGA: f32 = 1.7; pub fn solve_membrane(w: usize, h: usize, unknown: &[bool], v: &mut [f32]) { assert_eq!(unknown.len(), w * h); assert_eq!(v.len(), w * h); - solve_rec(w, h, unknown, v, 0); + solve_rec(w, h, unknown, v, 0, &|| false); } -fn solve_rec(w: usize, h: usize, unknown: &[bool], v: &mut [f32], depth: usize) { +fn solve_rec(w: usize, h: usize, unknown: &[bool], v: &mut [f32], depth: usize, stop: &(dyn Fn() -> bool + Sync)) { + if stop() { + return; + } let n_unknown = unknown.iter().filter(|u| **u).count(); if n_unknown == 0 { return; @@ -64,7 +67,7 @@ fn solve_rec(w: usize, h: usize, unknown: &[bool], v: &mut [f32], depth: usize) } } } - solve_rec(cw, chh, &cu, &mut cv, depth + 1); + solve_rec(cw, chh, &cu, &mut cv, depth + 1, stop); // Prolong: bilinear sample of the coarse solution as the initial guess. for y in 0..h { let fy = ((y as f32 + 0.5) / 2.0 - 0.5).clamp(0.0, (chh - 1) as f32); @@ -84,7 +87,7 @@ fn solve_rec(w: usize, h: usize, unknown: &[bool], v: &mut [f32], depth: usize) v[y * w + x] = a + (b - a) * ty; } } - sor(w, h, unknown, v, FINE_ITERS, FINE_OMEGA); + sor(w, h, unknown, v, FINE_ITERS, FINE_OMEGA, stop); } else { let mean = known.iter().sum::() / known.len() as f32; for (x, u) in v.iter_mut().zip(unknown) { @@ -92,12 +95,12 @@ fn solve_rec(w: usize, h: usize, unknown: &[bool], v: &mut [f32], depth: usize) *x = mean; } } - sor(w, h, unknown, v, 2000, 1.85); + sor(w, h, unknown, v, 2000, 1.85, stop); } } /// SOR sweeps over the unknown pixels until the largest update drops below [`TOL`]. -fn sor(w: usize, h: usize, unknown: &[bool], v: &mut [f32], max_iter: usize, omega: f32) { +fn sor(w: usize, h: usize, unknown: &[bool], v: &mut [f32], max_iter: usize, omega: f32, stop: &(dyn Fn() -> bool + Sync)) { // Precompute the unknown pixels and which neighbours exist. let cells: Vec<(u32, u8)> = (0..w * h) .filter(|&i| unknown[i]) @@ -108,6 +111,9 @@ fn sor(w: usize, h: usize, unknown: &[bool], v: &mut [f32], max_iter: usize, ome }) .collect(); for _ in 0..max_iter { + if stop() { + return; + } let mut max_d = 0.0f32; for &(i, m) in &cells { let i = i as usize; @@ -229,6 +235,28 @@ pub fn membrane_fill(w: usize, h: usize, ch: usize, img: &[f32], hole: &[bool]) }) } +/// [`membrane_fill`] that stops at the next SOR sweep once `ctl` is cancelled. +pub fn membrane_fill_with( + w: usize, + h: usize, + ch: usize, + img: &[f32], + hole: &[bool], + ctl: &photocraft_raster::Interrupt, +) -> Result, photocraft_raster::Cancelled> { + ctl.check()?; + let stop = || ctl.cancelled(); + let out = per_channel(w, h, ch, |c| { + let mut v: Vec = (0..w * h).map(|i| if hole[i] { 0.0 } else { img[i * ch + c] }).collect(); + if hole.len() == w * h && v.len() == w * h { + solve_rec(w, h, hole, &mut v, 0, &stop); + } + v + }); + ctl.check()?; + Ok(out) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/algo/src/seam.rs b/crates/algo/src/seam.rs index bdd7edd3f6..2d5ad2874b 100644 --- a/crates/algo/src/seam.rs +++ b/crates/algo/src/seam.rs @@ -11,6 +11,8 @@ //! seams avoid protected pixels. Buffers are interleaved `w × h × ch` normalised floats of any //! colour model and depth (alpha is carried like any other channel). Deterministic. +use photocraft_raster::{Cancelled, Interrupt}; + /// Energy added per unit of protection: far above any gradient magnitude. const PROTECT_ENERGY: f32 = 1.0e4; @@ -97,10 +99,17 @@ fn transpose(w: usize, h: usize, ch: usize, img: &[f32]) -> Vec { /// Change the width of `img` to `new_w` by removing or duplicating vertical seams. pub fn carve_width(w: usize, h: usize, ch: usize, img: &[f32], protect: Option<&[f32]>, new_w: usize) -> Vec { + // Never cancelled; the fallback is unreachable. + carve_width_with(w, h, ch, img, protect, new_w, &Interrupt::NONE).unwrap_or_else(|_| img.to_vec()) +} + +/// [`carve_width`] that checks `ctl` before every seam and reports progress per seam. +pub fn carve_width_with(w: usize, h: usize, ch: usize, img: &[f32], protect: Option<&[f32]>, new_w: usize, ctl: &Interrupt) -> Result, Cancelled> { let new_w = new_w.max(1); if new_w == w || w == 0 || h == 0 { - return img.to_vec(); + return Ok(img.to_vec()); } + let seams = new_w.abs_diff(w).max(1) as f32; let protect_e = |e: &mut [f32], p: Option<&Vec>| { if let Some(p) = p { for (e, p) in e.iter_mut().zip(p) { @@ -112,6 +121,8 @@ pub fn carve_width(w: usize, h: usize, ch: usize, img: &[f32], protect: Option<& let (mut cur, mut cw) = (img.to_vec(), w); let mut prot = protect.map(<[f32]>::to_vec); while cw > new_w { + ctl.check()?; + ctl.progress((w - cw) as f32 / seams); let mut e = energy(cw, h, ch, &cur); protect_e(&mut e, prot.as_ref()); let seam = find_seam(cw, h, &e); @@ -119,7 +130,7 @@ pub fn carve_width(w: usize, h: usize, ch: usize, img: &[f32], protect: Option<& prot = prot.map(|p| remove_seam_1(cw, h, &p, &seam)); cw -= 1; } - return cur; + return Ok(cur); } // Enlarge in rounds of at most half the current width, each duplicating the k lowest seams. let (mut cur, mut cw) = (img.to_vec(), w); @@ -135,6 +146,8 @@ pub fn carve_width(w: usize, h: usize, ch: usize, img: &[f32], protect: Option<& if tw <= 1 { break; } + ctl.check()?; + ctl.progress((cw + (cw - tw)).saturating_sub(w) as f32 / seams); let mut e = energy(tw, h, ch, &tmp); protect_e(&mut e, tprot.as_ref()); let seam = find_seam(tw, h, &e); @@ -177,21 +190,44 @@ pub fn carve_width(w: usize, h: usize, ch: usize, img: &[f32], protect: Option<& prot = nprot; cw = nw; } - cur + Ok(cur) } /// Content-aware resize of `img` to `new_w × new_h` (width first, then height by transposing). pub fn carve(w: usize, h: usize, ch: usize, img: &[f32], protect: Option<&[f32]>, new_w: usize, new_h: usize) -> Vec { + // Never cancelled; the fallback is unreachable. + carve_with(w, h, ch, img, protect, new_w, new_h, &Interrupt::NONE).unwrap_or_else(|_| img.to_vec()) +} + +/// [`carve`] that checks `ctl` before every seam (a cancel takes effect within one seam) and +/// reports progress over both passes. +#[allow(clippy::too_many_arguments)] +pub fn carve_with( + w: usize, + h: usize, + ch: usize, + img: &[f32], + protect: Option<&[f32]>, + new_w: usize, + new_h: usize, + ctl: &Interrupt, +) -> Result, Cancelled> { let (new_w, new_h) = (new_w.max(1), new_h.max(1)); - let wide = carve_width(w, h, ch, img, protect, new_w); + // Split progress between the passes by their seam counts. + let (sw, sh) = (new_w.abs_diff(w) as f32, new_h.abs_diff(h) as f32); + let split = if sw + sh > 0.0 { sw / (sw + sh) } else { 1.0 }; + let cancel = || ctl.cancelled(); + let first = |f: f32| ctl.progress(f * split); + let wide = carve_width_with(w, h, ch, img, protect, new_w, &Interrupt::new(&cancel, &first))?; if new_h == h { - return wide; + return Ok(wide); } // Protection follows the width change by plain resampling (it only steers seams). let prot_t = protect.map(|p| transpose(new_w, h, 1, &resize_bilinear(w, h, 1, p, new_w, h))); let t = transpose(new_w, h, ch, &wide); - let tall = carve_width(h, new_w, ch, &t, prot_t.as_deref(), new_h); - transpose(new_h, new_w, ch, &tall) + let second = |f: f32| ctl.progress(split + f * (1.0 - split)); + let tall = carve_width_with(h, new_w, ch, &t, prot_t.as_deref(), new_h, &Interrupt::new(&cancel, &second))?; + Ok(transpose(new_h, new_w, ch, &tall)) } /// Plain bilinear resize (the non-content-aware part of the Amount blend). @@ -241,6 +277,15 @@ pub fn skin_mask(w: usize, h: usize, ch: usize, img: &[f32]) -> Vec { mod tests { use super::*; + #[test] + fn cancelled_carve_stops() { + let (w, h) = (40, 20); + let img = vec![0.5f32; w * h * 3]; + let yes = || true; + assert_eq!(carve_with(w, h, 3, &img, None, 30, 15, &Interrupt::cancel_only(&yes)), Err(Cancelled)); + assert_eq!(carve_with(w, h, 3, &img, None, 30, 15, &Interrupt::NONE).unwrap(), carve(w, h, 3, &img, None, 30, 15)); + } + /// A flat grey image with a salient vertical bar at x = 10..14 (red/blue columns). fn bar(w: usize, h: usize) -> Vec { let mut v = vec![0.5f32; w * h * 3]; diff --git a/crates/algo/src/tests_ext.rs b/crates/algo/src/tests_ext.rs index 9aa0345bf6..5423cc52a2 100644 --- a/crates/algo/src/tests_ext.rs +++ b/crates/algo/src/tests_ext.rs @@ -666,3 +666,25 @@ fn oil_paint_smooths_along_strokes() { let inner = Rect::new(8, 8, 32, 22); assert!(variance(&out, inner, 0) < variance(&s, inner, 0) * 0.5); } + +#[test] +fn cancelled_filters_return_none_and_progress_is_monotonic() { + use std::sync::atomic::{AtomicU32, Ordering}; + let s = pattern(SampleType::U8, R); + let p = FilterParams::GaussianBlur { radius: 3.0 }; + let area = output_area(&p, s.content_bounds(), R, None); + let yes = || true; + assert!(apply_tiled_with(&s, &p, area, R, None, 16, Some(R), &photocraft_raster::Interrupt::cancel_only(&yes)).is_none()); + let last = AtomicU32::new(0); + let calls = AtomicU32::new(0); + let no = || false; + let progress = |f: f32| { + assert!(f >= f32::from_bits(last.load(Ordering::Relaxed))); + last.store(f.to_bits(), Ordering::Relaxed); + calls.fetch_add(1, Ordering::Relaxed); + }; + let out = apply_tiled_with(&s, &p, area, R, None, 16, Some(R), &photocraft_raster::Interrupt::new(&no, &progress)).unwrap(); + assert_eq!(f32::from_bits(last.load(Ordering::Relaxed)), 1.0); + assert!(calls.load(Ordering::Relaxed) >= 1); + assert_eq!(out.read_region(R), apply_tiled(&s, &p, area, R, None, 16, Some(R)).read_region(R)); +} diff --git a/crates/automation/src/headless.rs b/crates/automation/src/headless.rs index 1c5e3baed2..97b012e770 100644 --- a/crates/automation/src/headless.rs +++ b/crates/automation/src/headless.rs @@ -185,10 +185,25 @@ impl Headless { } pub fn command_run(&mut self, id: &str, params: Value) -> Result { + self.command_start(id, params, true) + } + + /// Run a command; with `wait` false, a job-capable command (filters, Content-Aware Fill, + /// Photomerge, …) runs in the background and this returns `{"job": id}` at once (poll with + /// `jobs.list`, stop with `jobs.cancel`). Other commands finish before returning either way. + pub fn command_start(&mut self, id: &str, params: Value, wait: bool) -> Result { let params = if params.is_null() { json!({}) } else { params }; if !matches!(&self.filesystem, Filesystem::TrustedLocal) { authorize_engine_command(id, ¶ms)?; } - Ok(self.session.execute(id, params)?) + // Background jobs that finished since the last request are applied first. + self.session.poll_jobs(); + if wait { + return Ok(self.session.execute(id, params)?); + } + Ok(match self.session.start(id, params)? { + photocraft_engine::jobs::Started::Done(v) => v, + photocraft_engine::jobs::Started::Job(job) => json!({"job": job.0, "pending": true}), + }) } } diff --git a/crates/automation/src/rpc.rs b/crates/automation/src/rpc.rs index d52c364c95..5cf871b4d6 100644 --- a/crates/automation/src/rpc.rs +++ b/crates/automation/src/rpc.rs @@ -62,8 +62,12 @@ impl Headless { match method { "engine.execute" => { let id = str_of(&p, "command").ok_or_else(|| bad("engine.execute needs `command`"))?; - self.command_run(id, p.get("params").cloned().unwrap_or(Value::Null)) + let wait = p.get("wait").and_then(Value::as_bool).unwrap_or(true); + self.command_start(id, p.get("params").cloned().unwrap_or(Value::Null), wait) } + // Background jobs (#210): list (applying any that finished) and cancel. + "jobs.list" => self.command_start("jobs.list", json!({}), true), + "jobs.cancel" => self.command_start("jobs.cancel", p, true), "engine.commands" => { let all = self.command_list(); Ok(match str_of(&p, "filter").map(str::to_lowercase) { @@ -378,6 +382,42 @@ mod tests { let _ = std::fs::remove_file(path); } + #[test] + fn background_jobs_list_wait_and_cancel() { + let mut h = Headless::new(); + h.handle("doc.new", json!({"width": 600, "height": 400})).unwrap(); + h.handle("engine.execute", json!({"command": "layer.new.layer"})).unwrap(); + h.handle("engine.execute", json!({"command": "edit.fill", "params": {"color": "#808080"}})).unwrap(); + // Default: waits, the result is the command's. + let r = h.handle("engine.execute", json!({"command": "filter.noise.addNoise", "params": {"amount": 20}})).unwrap(); + assert!(r["filter"].is_object(), "{r}"); + // wait: false returns a job id at once. + let r = h.handle("engine.execute", json!({"command": "filter.blur.gaussianBlur", "params": {"radius": 40}, "wait": false})).unwrap(); + let job = r["job"].as_u64().unwrap(); + let listed = h.handle("jobs.list", json!({})).unwrap(); + assert!(listed["jobs"].as_array().unwrap().iter().any(|j| j["id"] == job), "{listed}"); + // Wait for it via jobs.list (which applies finished jobs). + let t = std::time::Instant::now(); + loop { + let l = h.handle("jobs.list", json!({})).unwrap(); + let state = l["jobs"].as_array().unwrap().iter().find(|j| j["id"] == job).map(|j| j["state"].clone()); + if state == Some(json!("done")) { + break; + } + assert!(t.elapsed().as_secs() < 60, "{l}"); + std::thread::sleep(std::time::Duration::from_millis(5)); + } + // Cancel: a second job, cancelled at once, leaves the document unchanged. + let before = h.session.active().unwrap().doc.clone(); + let r = h.handle("engine.execute", json!({"command": "filter.blur.gaussianBlur", "params": {"radius": 40}, "wait": false})).unwrap(); + let job = r["job"].as_u64().unwrap(); + h.handle("jobs.cancel", json!({"job": job})).unwrap(); + assert!(h.handle("jobs.cancel", json!({"job": "nope"})).is_err()); + let l = h.handle("jobs.list", json!({})).unwrap(); + assert_eq!(l["jobs"].as_array().unwrap().iter().find(|j| j["id"] == job).unwrap()["state"], "cancelled"); + assert!(std::sync::Arc::ptr_eq(&h.session.active().unwrap().doc, &before)); + } + #[test] fn automation_preview_rejects_oversized_full_size_and_numeric_wraparound() { let mut h = Headless::new(); diff --git a/crates/automation/src/server.rs b/crates/automation/src/server.rs index 7ac1f46fca..346ec8e08b 100644 --- a/crates/automation/src/server.rs +++ b/crates/automation/src/server.rs @@ -117,6 +117,18 @@ pub struct RunParams { /// Command parameters as a JSON object (see the `params` doc in `command_list`). #[serde(default)] pub params: Option, + /// Wait for a long command (filters, Content-Aware Fill, Photomerge, …) to finish (default + /// true). With false it runs as a background job and the result is `{"job": id}`: poll it + /// with `jobs_list`, stop it with `jobs_cancel`. + #[serde(default)] + pub wait: Option, +} + +#[derive(Debug, Deserialize, JsonSchema)] +pub struct JobCancelParams { + /// The job id from `command_run` / `jobs_list` (omit to cancel every running job). + #[serde(default)] + pub job: Option, } #[derive(Debug, Deserialize, JsonSchema)] @@ -374,7 +386,7 @@ impl PhotocraftMcp { if let Some(v) = p.name { params.insert("name".into(), v.into()); } - self.run_command("file.new".into(), Value::Object(params)).await + self.run_command("file.new".into(), Value::Object(params), true).await } #[tool(description = "Save the document. `.pcraft` is the lossless native format (incremental); other extensions \ @@ -464,9 +476,23 @@ impl PhotocraftMcp { Ok(ok_json(&Value::Array(items))) } - #[tool(description = "Run an engine command by id with JSON params (see command_list). Returns the command's JSON result.")] + #[tool(description = "Run an engine command by id with JSON params (see command_list). Returns the command's JSON result. \ + Long commands (filters, Content-Aware Fill/Scale, Photomerge, brush import) wait to finish unless `wait` is false: \ + then they run as a background job and the result is {job: id} (see jobs_list, jobs_cancel).")] async fn command_run(&self, Parameters(p): Parameters) -> Result { - self.run_command(p.id, p.params.unwrap_or_else(|| json!({}))).await + self.run_command(p.id, p.params.unwrap_or_else(|| json!({})), p.wait.unwrap_or(true)).await + } + + #[tool(description = "List background jobs: running ones with progress (0-1), message and elapsed time, then the last few \ + that ended (state done|failed|cancelled with their result or error). Finished jobs are applied first.")] + async fn jobs_list(&self) -> Result { + self.jobs_call("jobs.list", json!({})).await + } + + #[tool(description = "Cancel a background job by id (or every running job when `job` is omitted). The document is left \ + unchanged.")] + async fn jobs_cancel(&self, Parameters(p): Parameters) -> Result { + self.jobs_call("jobs.cancel", p.job.map_or_else(|| json!({}), |j| json!({"job": j}))).await } #[tool(description = "Run several engine commands in one call (fewer round trips). Returns {completed, failed, \ @@ -572,15 +598,27 @@ fn bridge_only_headless(name: &str) -> Result { } impl PhotocraftMcp { - async fn run_command(&self, id: String, params: Value) -> Result { + async fn run_command(&self, id: String, params: Value, wait: bool) -> Result { let (id2, params2) = (id.clone(), params.clone()); - if let Some(r) = self.headless_op(move |h| h.command_run(&id2, params2)).await { + if let Some(r) = self.headless_op(move |h| h.command_start(&id2, params2, wait)).await { return to_result(r); } let Some(b) = self.bridge_client() else { return Ok(no_backend()); }; - to_result(b.call("engine.execute", json!({"command": id, "params": params})).await) + to_result(b.call("engine.execute", json!({"command": id, "params": params, "wait": wait})).await) + } + + /// `jobs.list` / `jobs.cancel` in either backend. + async fn jobs_call(&self, method: &'static str, params: Value) -> Result { + let p2 = params.clone(); + if let Some(r) = self.headless_op(move |h| h.command_start(method, p2, true)).await { + return to_result(r); + } + let Some(b) = self.bridge_client() else { + return Ok(no_backend()); + }; + to_result(b.call(method, params).await) } async fn save_impl(&self, p: SaveParams) -> Result { diff --git a/crates/engine/src/commands.rs b/crates/engine/src/commands.rs index 4fdbf61cde..8be97f2078 100644 --- a/crates/engine/src/commands.rs +++ b/crates/engine/src/commands.rs @@ -991,6 +991,7 @@ fn build() -> Vec { v.extend(crate::trap_cmds::specs()); v.extend(crate::timeline_cmds::specs()); v.extend(crate::video_cmds::specs()); + v.extend(crate::jobs::specs()); v.extend(crate::wia_cmds::specs()); v.extend(crate::variables_cmds::specs()); v.extend(crate::plugin_cmds::specs()); diff --git a/crates/engine/src/edit_menu_cmds.rs b/crates/engine/src/edit_menu_cmds.rs index c3724bc324..b8018a51e3 100644 --- a/crates/engine/src/edit_menu_cmds.rs +++ b/crates/engine/src/edit_menu_cmds.rs @@ -310,7 +310,7 @@ fn rect_param(p: &Value, key: &str) -> Option { } fn content_aware_fill(s: &mut Session, p: &Value) -> Result { - use photocraft_algo::content_aware::{FillOptions, color_level, fill, rotation_level}; + use photocraft_algo::content_aware::{FillOptions, color_level, fill_with, rotation_level}; let cmd = "edit.contentAwareFill"; let id = pixel_layer(s).map_err(EngineError::Other)?; let st = s.active().ok_or(EngineError::NoDocument)?; @@ -321,14 +321,14 @@ fn content_aware_fill(s: &mut Session, p: &Value) -> Result { return Err(bad(cmd, "the selection is outside the canvas")); } let ext = hb.width().max(hb.height()) as i32; - let sampling = str_or(p, "sampling", "auto"); - let custom_mask = p.get("channel").and_then(|v| channel_mask(&doc, v)); + let sampling = str_or(p, "sampling", "auto").to_string(); + let custom_mask: Option = p.get("channel").and_then(|v| channel_mask(&doc, v)).cloned(); let custom_rect = rect_param(p, "area"); - let window = match sampling { + let window = match sampling.as_str() { "auto" => hb.inflate((ext * 3 / 4).max(32)), "rectangular" => hb.inflate(int(p, "margin").map_or(ext.max(16), |m| m.clamp(0, 100_000) as i32)), "custom" => { - let r = match (custom_rect, custom_mask) { + let r = match (custom_rect, custom_mask.as_ref()) { (Some(r), _) => r, (None, Some(m)) => m.content_bounds(), (None, None) => return Err(bad(cmd, "custom sampling needs `area` [x,y,w,h] or `channel` (alpha channel index or name)")), @@ -353,30 +353,66 @@ fn content_aware_fill(s: &mut Session, p: &Value) -> Result { let fmt = surf.format(); let n = fmt.channels(); let (w, h) = (window.width() as usize, window.height() as usize); - let img = surf.read_region(window); - let mut cover = vec![0.0f32; w * h]; - let mut hole = vec![false; w * h]; - let mut source = vec![true; w * h]; - for y in 0..h { - for x in 0..w { - let (dx, dy) = (window.x0 + x as i32, window.y0 + y as i32); - let i = y * w + x; - cover[i] = sel.sample_channel(dx, dy, 0); - hole[i] = cover[i] > 0.0; - if sampling == "custom" { - source[i] = match (custom_rect, custom_mask) { - (Some(r), _) => r.contains(dx, dy), - (None, Some(m)) => m.sample_channel(dx, dy, 0) > 0.5, - _ => true, - }; - } - } - } - let filled = fill(w, h, n, &img, &hole, &source, &opts); - let n_hole = hole.iter().filter(|h| **h).count(); let label = "Content-Aware Fill"; + // A background job when started with `Session::start` (#210): reading the window and the + // PatchMatch fill run on a worker against the document snapshot, cancellable per row band. + crate::jobs::run( + s, + label, + true, + move |ctx| { + ctx.progress(0.0, label); + let surf = doc.layer(id).and_then(|l| l.surface()).ok_or(EngineError::NoLayer(id))?; + let img = surf.read_region(window); + let mut cover = vec![0.0f32; w * h]; + let mut hole = vec![false; w * h]; + let mut source = vec![true; w * h]; + for y in 0..h { + if y % 64 == 0 { + ctx.check()?; + } + for x in 0..w { + let (dx, dy) = (window.x0 + x as i32, window.y0 + y as i32); + let i = y * w + x; + cover[i] = sel.sample_channel(dx, dy, 0); + hole[i] = cover[i] > 0.0; + if sampling == "custom" { + source[i] = match (custom_rect, custom_mask.as_ref()) { + (Some(r), _) => r.contains(dx, dy), + (None, Some(m)) => m.sample_channel(dx, dy, 0) > 0.5, + _ => true, + }; + } + } + } + ctx.check()?; + let filled = ctx.stage(0.05, 1.0, label, |ctl| fill_with(w, h, n, &img, &hole, &source, &opts, ctl)).map_err(|_| EngineError::Cancelled)?; + Ok((img, cover, hole, filled)) + }, + move |s, (img, cover, hole, filled)| apply_content_aware_fill(s, label, id, &output, fmt, window, &img, &cover, &hole, &filled), + ) +} + +/// Write a Content-Aware Fill result (the job's apply step): into the layer, a new layer or a +/// duplicate, as one undo step. +#[allow(clippy::too_many_arguments)] +fn apply_content_aware_fill( + s: &mut Session, + label: &str, + id: LayerId, + output: &str, + fmt: PixelFormat, + window: Rect, + img: &[f32], + cover: &[f32], + hole: &[bool], + filled: &[f32], +) -> Result { + let n = fmt.channels(); + let (w, h) = (window.width() as usize, window.height() as usize); + let n_hole = hole.iter().filter(|h| **h).count(); let layer = s.edit(label, |doc, active| { - let target = match output.as_str() { + let target = match output { "new" => { let mut l = Layer::raster(doc.next_layer_name("Content-Aware Fill"), PixelFormat { alpha: true, ..fmt }); let mut px = vec![0.0f32; w * h * n.max(1)]; @@ -409,7 +445,7 @@ fn content_aware_fill(s: &mut Session, p: &Value) -> Result { _ => id, }; let surf = writable_surface(doc, target)?; - let mut out = img.clone(); + let mut out = img.to_vec(); for (i, &k) in cover.iter().enumerate() { if k > 0.0 { for c in 0..n { @@ -427,7 +463,7 @@ fn content_aware_fill(s: &mut Session, p: &Value) -> Result { // ------------------------------------------------------------------ Content-Aware Scale fn content_aware_scale(s: &mut Session, p: &Value) -> Result { - use photocraft_algo::seam::{carve, resize_bilinear, skin_mask}; + use photocraft_algo::seam::{carve_with, resize_bilinear, skin_mask}; let cmd = "edit.contentAwareScale"; let id = pixel_layer(s).map_err(EngineError::Other)?; let st = s.active().ok_or(EngineError::NoDocument)?; @@ -456,38 +492,58 @@ fn content_aware_scale(s: &mut Session, p: &Value) -> Result { let amount = f32_or(p, "amount", 100.0).clamp(0.0, 100.0) / 100.0; let fmt = surf.format(); let n = fmt.channels(); - let img = surf.read_region(region); // Protection: an alpha channel and/or skin tones. - let mut protect: Option> = None; - if let Some(m) = p.get("protect").filter(|v| !v.is_null() && v.as_str() != Some("none")).and_then(|v| channel_mask(&doc, v)) { - protect = Some((0..w * h).map(|i| m.sample_channel(region.x0 + (i % w) as i32, region.y0 + (i / w) as i32, 0)).collect()); - } else if p.get("protect").is_some_and(|v| !v.is_null() && v.as_str() != Some("none")) { - return Err(bad(cmd, "`protect` must name an alpha channel (index or name) or \"none\"")); - } - if bool_or(p, "protectSkinTones", false) { - let rgba: Vec = img.chunks_exact(n).flat_map(|px| to_rgba(&fmt, px)).collect(); - let skin = skin_mask(w, h, 4, &rgba); - protect = Some(match protect { - Some(pm) => pm.iter().zip(&skin).map(|(a, b)| a.max(*b)).collect(), - None => skin, - }); - } - let carved = if amount > 0.0 { carve(w, h, n, &img, protect.as_deref(), nw, nh) } else { Vec::new() }; - let plain = if amount < 1.0 { resize_bilinear(w, h, n, &img, nw, nh) } else { Vec::new() }; - let out: Vec = match (carved.is_empty(), plain.is_empty()) { - (false, true) => carved, - (true, false) => plain, - _ => carved.iter().zip(&plain).map(|(c, l)| l + (c - l) * amount).collect(), + let protect_mask: Option = match p.get("protect").filter(|v| !v.is_null() && v.as_str() != Some("none")) { + Some(v) => Some(channel_mask(&doc, v).cloned().ok_or_else(|| bad(cmd, "`protect` must name an alpha channel (index or name) or \"none\""))?), + None => None, }; + let skin_tones = bool_or(p, "protectSkinTones", false); let dst = Rect::from_xywh(region.x0, region.y0, nw as u32, nh as u32); - s.edit("Content-Aware Scale", |doc, _| { - let surf = writable_surface(doc, id)?; - crate::pixels::clear_surface(surf, region, None); - surf.write_region(dst, &out); - surf.prune(); - Ok(()) - })?; - Ok(json!({"from": [w, h], "to": [nw, nh], "bounds": [dst.x0, dst.y0, nw, nh]})) + let label = "Content-Aware Scale"; + // A background job when started with `Session::start` (#210): seam carving runs on a worker + // against the document snapshot and checks for cancellation before every seam. + crate::jobs::run( + s, + label, + true, + move |ctx| { + ctx.progress(0.0, label); + let surf = doc.layer(id).and_then(|l| l.surface()).ok_or(EngineError::NoLayer(id))?; + let img = surf.read_region(region); + let mut protect: Option> = + protect_mask.map(|m| (0..w * h).map(|i| m.sample_channel(region.x0 + (i % w) as i32, region.y0 + (i / w) as i32, 0)).collect()); + if skin_tones { + let rgba: Vec = img.chunks_exact(n).flat_map(|px| to_rgba(&fmt, px)).collect(); + let skin = skin_mask(w, h, 4, &rgba); + protect = Some(match protect { + Some(pm) => pm.iter().zip(&skin).map(|(a, b)| a.max(*b)).collect(), + None => skin, + }); + } + ctx.check()?; + let carved = if amount > 0.0 { + ctx.stage(0.05, 0.95, label, |ctl| carve_with(w, h, n, &img, protect.as_deref(), nw, nh, ctl)).map_err(|_| EngineError::Cancelled)? + } else { + Vec::new() + }; + let plain = if amount < 1.0 { resize_bilinear(w, h, n, &img, nw, nh) } else { Vec::new() }; + Ok(match (carved.is_empty(), plain.is_empty()) { + (false, true) => carved, + (true, false) => plain, + _ => carved.iter().zip(&plain).map(|(c, l)| l + (c - l) * amount).collect::>(), + }) + }, + move |s, out| { + s.edit(label, |doc, _| { + let surf = writable_surface(doc, id)?; + crate::pixels::clear_surface(surf, region, None); + surf.write_region(dst, &out); + surf.prune(); + Ok(()) + })?; + Ok(json!({"from": [w, h], "to": [nw, nh], "bounds": [dst.x0, dst.y0, nw, nh]})) + }, + ) } // ------------------------------------------------------------------ Define Brush / Custom Shape diff --git a/crates/engine/src/filters.rs b/crates/engine/src/filters.rs index 2d1c05fec5..8fc2cf9a17 100644 --- a/crates/engine/src/filters.rs +++ b/crates/engine/src/filters.rs @@ -194,38 +194,52 @@ pub(crate) fn run_filter(s: &mut Session, id: &str, p: &Value) -> Result if let Value::Object(m) = &mut params { m.remove("__kind"); } - s.edit(&label, |doc, _| { - let selection = doc.selection.clone(); - let sel_bounds = selection.as_ref().map(photocraft_raster::Surface::content_bounds); - // Distortions centre on the selection when there is one, else the canvas. - let doc_bounds = doc.bounds(); - let bounds = sel_bounds.filter(|b| !b.is_empty()).unwrap_or(doc_bounds); - // Neighbourhood filters repeat edge pixels at the canvas edge (or past it, where the - // layer has off-canvas pixels) instead of fading in transparency, as Photoshop does. - if let Some(surf) = crate::channel_cmds::channel_surface_for_filter(doc, p)? { + let id = id.to_string(); + let p = p.clone(); + let layer_id = layer.0; + // A background job when started with `Session::start` (#210): the filter runs on a worker, + // checking for cancellation before every tile. + crate::jobs::edit_job( + s, + &label, + move |doc, _, ctx| { + let msg = fp.label().to_string(); + let filter = |surf: &photocraft_raster::Surface, fp: &FilterParams, area, bounds, sel: Option<&photocraft_raster::Surface>, extent| { + ctx.stage(0.0, 1.0, &msg, |ctl| algo::apply_in_with(surf, fp, area, bounds, sel, extent, ctl)).ok_or(EngineError::Cancelled) + }; + let selection = doc.selection.clone(); + let sel_bounds = selection.as_ref().map(photocraft_raster::Surface::content_bounds); + // Distortions centre on the selection when there is one, else the canvas. + let doc_bounds = doc.bounds(); + let bounds = sel_bounds.filter(|b| !b.is_empty()).unwrap_or(doc_bounds); + // Neighbourhood filters repeat edge pixels at the canvas edge (or past it, where the + // layer has off-canvas pixels) instead of fading in transparency, as Photoshop does. + if let Some(surf) = crate::channel_cmds::channel_surface_for_filter(doc, &p)? { + let content = surf.content_bounds(); + let area = algo::output_area(&fp, content, bounds, sel_bounds); + *surf = filter(surf, &fp, area, bounds, selection.as_ref(), doc_bounds.union(&content))?; + return Ok(fp.clone()); + } + let l = doc.layer_mut(layer).ok_or(EngineError::NoLayer(layer))?; + let mut fp = fp.clone(); + crate::filters_ext::resolve_in_layer(&mut fp, l, bounds); + let surf = match &mut l.content { + LayerContent::Raster(surf) => surf, + LayerContent::Smart(_) => { + // Non-destructive: record the filter and re-render the smart object from its source. + let sf = + SmartFilter { command: id.clone(), params: params.clone(), blend: photocraft_color::BlendMode::Normal, opacity: 1.0, visible: true }; + return crate::smart_cmds::add_smart_filter(doc, layer, sf, selection.as_ref()).map(|()| fp); + } + _ => return Err(EngineError::Other("not a pixel layer".into())), + }; let content = surf.content_bounds(); let area = algo::output_area(&fp, content, bounds, sel_bounds); - *surf = algo::apply_in(surf, &fp, area, bounds, selection.as_ref(), doc_bounds.union(&content)); - return Ok(()); - } - let l = doc.layer_mut(layer).ok_or(EngineError::NoLayer(layer))?; - crate::filters_ext::resolve_in_layer(&mut fp, l, bounds); - let surf = match &mut l.content { - LayerContent::Raster(surf) => surf, - LayerContent::Smart(_) => { - // Non-destructive: record the filter and re-render the smart object from its source. - let sf = - SmartFilter { command: id.to_string(), params: params.clone(), blend: photocraft_color::BlendMode::Normal, opacity: 1.0, visible: true }; - return crate::smart_cmds::add_smart_filter(doc, layer, sf, selection.as_ref()); - } - _ => return Err(EngineError::Other("not a pixel layer".into())), - }; - let content = surf.content_bounds(); - let area = algo::output_area(&fp, content, bounds, sel_bounds); - *surf = algo::apply_in(surf, &fp, area, bounds, selection.as_ref(), doc_bounds.union(&content)); - Ok(()) - })?; - Ok(json!({ "layer": layer.0, "filter": serde_json::to_value(&fp).unwrap_or(Value::Null) })) + *surf = filter(surf, &fp, area, bounds, selection.as_ref(), doc_bounds.union(&content))?; + Ok(fp) + }, + move |fp| json!({ "layer": layer_id, "filter": serde_json::to_value(&fp).unwrap_or(Value::Null) }), + ) } macro_rules! filter_cmd { diff --git a/crates/engine/src/jobs.rs b/crates/engine/src/jobs.rs new file mode 100644 index 0000000000..203d59b6f3 --- /dev/null +++ b/crates/engine/src/jobs.rs @@ -0,0 +1,757 @@ +//! Background jobs: long commands run on a worker thread, report progress and can be cancelled +//! (#210). +//! +//! A command opts in by doing its heavy work through [`run`] (or [`edit_job`] for a plain document +//! edit). The work closure gets a [`JobCtx`] (progress plus a cancellation flag) and runs against +//! a snapshot: documents are copy-on-write, so cloning one is O(layers). Its result is applied on +//! the UI thread as one undo step, through the same path as [`Session::execute`], so the journal, +//! history, damage and post-command hooks all behave as for a synchronous command. +//! +//! - [`Session::execute`] runs everything inline, as before: tests, the CLI and scripts see no +//! difference. +//! - [`Session::start`] runs a job-capable command on a worker thread and returns its +//! [`JobId`]; [`Session::poll_jobs`] (called every frame by the UI) applies finished jobs. +//! On wasm `start` runs inline (same API, written once). +//! - [`Session::cancel_job`] sets the flag; the algorithms check it per tile, row band or +//! iteration, and the result is discarded. The document is never touched by a worker. +//! - A panic in the worker becomes an error; the document is unchanged. +//! - While a job runs on a document, commands that would edit that document are disabled with a +//! message naming the job (queries and document-independent commands still run). + +use std::any::Any; +use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; +use std::sync::{Arc, Mutex, PoisonError, Weak}; + +use photocraft_doc::{DocId, Document, LayerId}; +use photocraft_raster::Interrupt; +use serde::Serialize; +use serde_json::{Value, json}; + +use crate::{EngineError, Result, Session}; + +/// Identifies one background job for the session's lifetime. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, serde::Deserialize)] +pub struct JobId(pub u64); + +/// What a job's work closure sees: report progress, check for cancellation. Cheap to clone; all +/// clones share the same state. +#[derive(Clone, Default)] +pub struct JobCtx { + shared: Arc, +} + +#[derive(Default)] +struct Shared { + cancel: AtomicBool, + /// `f32` bits of the progress fraction. Non-negative floats order like their bits, so + /// `fetch_max` keeps progress monotonic even with several reporting threads. + progress: AtomicU32, + message: Mutex, +} + +impl std::fmt::Debug for JobCtx { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("JobCtx").field("progress", &self.fraction()).field("cancelled", &self.cancelled()).finish() + } +} + +impl JobCtx { + pub fn new() -> Self { + Self::default() + } + + /// Report progress in `0.0..=1.0` with a short status (empty keeps the previous one). + /// Progress never goes backwards: a lower value than already reported is ignored. + pub fn progress(&self, fraction: f32, message: &str) { + if fraction.is_finite() { + self.shared.progress.fetch_max(fraction.clamp(0.0, 1.0).to_bits(), Ordering::Relaxed); + } + if !message.is_empty() { + let mut m = self.shared.message.lock().unwrap_or_else(PoisonError::into_inner); + if *m != message { + message.clone_into(&mut m); + } + } + } + + /// Has the job been cancelled? Long loops check this and stop early. + pub fn cancelled(&self) -> bool { + self.shared.cancel.load(Ordering::Relaxed) + } + + /// Ask the job to stop (idempotent). + pub fn cancel(&self) { + self.shared.cancel.store(true, Ordering::Relaxed); + } + + /// `Err(Cancelled)` once cancelled, for `?` between stages. + pub fn check(&self) -> Result<()> { + if self.cancelled() { Err(EngineError::Cancelled) } else { Ok(()) } + } + + /// Progress reported so far (`0.0..=1.0`). + pub fn fraction(&self) -> f32 { + f32::from_bits(self.shared.progress.load(Ordering::Relaxed)) + } + + /// The latest status message. + pub fn message(&self) -> String { + self.shared.message.lock().unwrap_or_else(PoisonError::into_inner).clone() + } + + /// Run `f` with an algorithm-level [`Interrupt`] whose progress maps onto `lo..hi` of this + /// job (with `message`) and whose cancellation is this job's. + pub fn stage(&self, lo: f32, hi: f32, message: &str, f: impl FnOnce(&Interrupt) -> R) -> R { + self.progress(lo, message); + let cancel = || self.cancelled(); + let progress = |p: f32| self.progress(lo + (hi - lo) * p, ""); + f(&Interrupt::new(&cancel, &progress)) + } +} + +/// Final state of a job. +#[derive(Clone, Debug, PartialEq)] +pub enum JobOutcome { + /// Applied; the command's result. + Done(Value), + /// Failed (including a panic in the worker); the document is unchanged. + Failed(String), + /// Cancelled; the result (if any) was discarded and the document is unchanged. + Cancelled, +} + +/// A job that has ended, as reported by [`Session::poll_jobs`]. +#[derive(Clone, Debug, PartialEq)] +pub struct JobEvent { + pub id: JobId, + pub command: String, + pub label: String, + pub document: Option, + pub outcome: JobOutcome, +} + +/// A job's state for the UI, the control channel and MCP (`jobs.list`). +#[derive(Clone, Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct JobInfo { + pub id: JobId, + pub command: String, + /// Shown in the progress UI ("Gaussian Blur"). + pub label: String, + /// The document the job edits (and locks), if any. + pub document: Option, + /// `running`, `done`, `failed` or `cancelled`. + pub state: &'static str, + pub progress: f32, + pub message: String, + pub elapsed_ms: f64, + /// The command's result (done) or the error (failed). + #[serde(skip_serializing_if = "Option::is_none")] + pub result: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, +} + +type Payload = Box; +type Apply = Box Result + Send>; + +/// Wall-clock milliseconds since some fixed point (0 on wasm, where `Instant` is unavailable and +/// jobs run inline anyway). +fn now_ms() -> f64 { + #[cfg(not(target_arch = "wasm32"))] + { + static START: std::sync::OnceLock = std::sync::OnceLock::new(); + START.get_or_init(std::time::Instant::now).elapsed().as_secs_f64() * 1000.0 + } + #[cfg(target_arch = "wasm32")] + { + 0.0 + } +} + +struct Running { + id: JobId, + command: String, + label: String, + params: Value, + journal: bool, + doc: Option, + /// The document as the job saw it: applying fails if it was edited meanwhile. + snapshot: Option>, + ctx: JobCtx, + started_ms: f64, + coalesce: Option, + color_restrict: Option, + rx: std::sync::mpsc::Receiver>, + apply: Option, + #[cfg(not(target_arch = "wasm32"))] + handle: Option>, +} + +impl Running { + fn info(&self) -> JobInfo { + JobInfo { + id: self.id, + command: self.command.clone(), + label: self.label.clone(), + document: self.doc, + state: "running", + progress: self.ctx.fraction(), + message: self.ctx.message(), + elapsed_ms: now_ms() - self.started_ms, + result: None, + error: None, + } + } +} + +/// The session's job table (see the module docs). +#[derive(Default)] +pub struct Jobs { + next: u64, + /// Set while [`Session::start`] runs a command: [`run`] spawns instead of running inline. + spawn: bool, + /// The job a command registered while `spawn` was set (picked up by `start`). + pending: Option, + running: Vec, + /// Ended jobs not yet collected by [`Session::poll_jobs`]. + events: Vec, + /// The last few ended jobs, for `jobs.list`. + recent: Vec, + /// Workers of cancelled jobs that may still be unwinding (native). + #[cfg(not(target_arch = "wasm32"))] + draining: Vec>, +} + +const RECENT: usize = 16; + +impl Drop for Jobs { + fn drop(&mut self) { + // Workers own snapshots only; stop them so they don't burn CPU for a closed session. + for r in self.running.iter().chain(self.pending.iter()) { + r.ctx.cancel(); + } + } +} + +impl std::fmt::Debug for Jobs { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Jobs").field("running", &self.running.len()).finish() + } +} + +/// How [`Session::start`] ended. +#[derive(Clone, Debug, PartialEq)] +pub enum Started { + /// The command ran to completion inline (it isn't a job, or this is wasm). + Done(Value), + /// The command is running in the background. + Job(JobId), +} + +/// Commands that never edit the active document, allowed while a job runs on it. +fn independent_of_document(id: &str) -> bool { + const PREFIXES: &[&str] = &["jobs.", "tools.", "view.", "window.", "help.", "brush.presets.", "gradient.presets."]; + const IDS: &[&str] = &["file.new", "file.close", "file.closeAll", "edit.preferences", "edit.colorSettings", "edit.keyboardShortcuts", "edit.menus"]; + PREFIXES.iter().any(|p| id.starts_with(p)) || IDS.contains(&id) +} + +#[cfg(not(target_arch = "wasm32"))] +fn read_path(path: &str) -> Result> { + std::fs::read(path).map_err(|e| EngineError::Other(format!("{path}: {e}"))) +} + +#[cfg(target_arch = "wasm32")] +fn read_path(path: &str) -> Result> { + Err(EngineError::Other(format!("cannot open paths on the web: {path}"))) +} + +/// Run a job-capable command's heavy part. `work` runs on a worker thread when the command was +/// started with [`Session::start`] (inline otherwise, and always on wasm); `apply` then runs on +/// the calling thread with its output and returns the command's result. With `lock_document`, +/// the active document is locked against other edits while the job runs, and applying fails if +/// it was edited anyway. +/// +/// `work` must only read what it captured (clone the document `Arc` first): it never sees the +/// session. +pub fn run( + s: &mut Session, + label: &str, + lock_document: bool, + work: impl FnOnce(&JobCtx) -> Result + Send + 'static, + apply: impl FnOnce(&mut Session, T) -> Result + Send + 'static, +) -> Result { + if !s.jobs.spawn || cfg!(target_arch = "wasm32") { + let t = work(&JobCtx::new())?; + return apply(s, t); + } + if s.jobs.pending.is_some() { + return Err(EngineError::Other(format!("`{label}` started a second background job in one command"))); + } + let (doc, snapshot) = match (lock_document, s.active()) { + (true, Some(st)) => (Some(st.doc.id), Some(Arc::downgrade(&st.doc))), + (true, None) => return Err(EngineError::NoDocument), + (false, _) => (None, None), + }; + let ctx = JobCtx::new(); + let (tx, rx) = std::sync::mpsc::channel::>(); + let erased: Apply = Box::new(move |s, any| { + let t = any.downcast::().map_err(|_| EngineError::Other("internal error: job result type mismatch".into()))?; + apply(s, *t) + }); + s.jobs.next += 1; + let id = JobId(s.jobs.next); + #[cfg(not(target_arch = "wasm32"))] + let handle = { + let wctx = ctx.clone(); + let label_w = label.to_string(); + // The work closure is moved into the thread. If spawning fails (thread limits), there is + // no way to get it back, so the error is reported instead. + let spawned = std::thread::Builder::new().name(format!("photocraft-job-{}", id.0)).spawn(move || { + let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| work(&wctx).map(|t| Box::new(t) as Payload))) + .unwrap_or_else(|_| Err(EngineError::Other(format!("`{label_w}` failed with an internal error (logged); the document is unchanged")))); + // The receiver is gone when the job was cancelled: nothing to report. + let _ = tx.send(r); + }); + Some(spawned.map_err(|e| EngineError::Other(format!("could not start a background job: {e}")))?) + }; + #[cfg(target_arch = "wasm32")] + { + // Unreachable (wasm runs inline above); keep the types used. + drop((work, tx)); + } + s.jobs.pending = Some(Running { + id, + command: String::new(), + label: label.to_string(), + params: Value::Null, + journal: true, + doc, + snapshot, + ctx, + started_ms: now_ms(), + coalesce: s.coalesce_request.clone(), + color_restrict: s.color_restrict, + rx, + apply: Some(erased), + #[cfg(not(target_arch = "wasm32"))] + handle, + }); + Ok(json!({"job": id.0})) +} + +/// A document edit whose body is the job: like [`Session::edit`] (one undo step named `label`), +/// but `f` runs on a worker against a copy of the active document when started in the +/// background. `finish` turns `f`'s output into the command's result. +pub fn edit_job( + s: &mut Session, + label: &str, + f: impl FnOnce(&mut Document, &mut Option, &JobCtx) -> Result + Send + 'static, + finish: impl FnOnce(R) -> Value + Send + 'static, +) -> Result { + if !s.jobs.spawn || cfg!(target_arch = "wasm32") { + // Inline: exactly `Session::edit`, as before jobs existed. + let ctx = JobCtx::new(); + let r = s.edit(label, |doc, active| f(doc, active, &ctx))?; + return Ok(finish(r)); + } + let st = s.active().ok_or(EngineError::NoDocument)?; + let (base, active) = (st.doc.clone(), st.active_layer); + let label_owned = label.to_string(); + run( + s, + label, + true, + move |ctx| { + let mut doc = (*base).clone(); + let mut active = active; + let r = f(&mut doc, &mut active, ctx)?; + Ok((doc, active, r)) + }, + move |s, (doc, new_active, r)| { + s.edit(&label_owned, move |d, active| { + *d = doc; + *active = new_active; + Ok(()) + })?; + Ok(finish(r)) + }, + ) +} + +/// Where a background open reads its file from. +#[derive(Clone, Debug)] +pub enum OpenSource { + /// File contents already in memory (web pickers, drops, automation). + Bytes(Arc>), + /// A path the worker reads itself (native), so even the read stays off the UI thread. + Path(String), +} + +/// The command id open jobs report (there is no engine command: opening is the shell's). +pub const OPEN_JOB: &str = "file.open"; + +impl Session { + /// Run a command, in the background when it supports it (see [`jobs`](crate::jobs)). Errors + /// like [`Session::execute`] for unknown, disabled or failing commands. + pub fn start(&mut self, id: &str, params: Value) -> Result { + self.dispatch(id, params, true) + } + + /// Open a file in the background: read (for a path) and decode on a worker, with progress + /// (per layer for PSD/PSB) and cancellation, then add the document like + /// [`Session::open_document`] (Color Settings policies apply). `name` names the document. + /// The job's result is `{document, name, warnings, color}`; on wasm it runs inline. + pub fn start_open(&mut self, name: &str, source: OpenSource) -> Result { + let label = format!("Opening {name}"); + let name_w = name.to_string(); + let name_a = name.to_string(); + self.start_job( + OPEN_JOB, + json!({"name": name}), + &label, + false, + move |ctx| { + ctx.progress(0.0, "Reading"); + let bytes = match source { + OpenSource::Bytes(b) => b, + OpenSource::Path(p) => Arc::new(read_path(&p)?), + }; + ctx.check()?; + ctx.progress(0.02, "Decoding"); + ctx.stage(0.02, 1.0, "Decoding", |ctl| photocraft_io::import_with(&name_w, &bytes, ctl)).map_err(|e| match e { + photocraft_io::IoError::Cancelled => EngineError::Cancelled, + e => EngineError::Other(e.to_string()), + }) + }, + move |s, r: photocraft_io::ImportResult| { + let (index, color) = s.open_document(r.document, None); + Ok(json!({"document": index, "name": name_a, "warnings": r.warnings, "color": color})) + }, + ) + } + + /// Start a job that isn't an engine command (opening a file, a shell operation, a test's fake + /// job): `work` runs on a worker, `apply` on this thread when [`Session::poll_jobs`] or + /// [`Session::wait_job`] picks the result up. `command` and `params` identify it in + /// [`JobInfo`]; it is not journaled. With `lock_document` the active document is locked as + /// for a command. Runs inline on wasm. + pub fn start_job( + &mut self, + command: &str, + params: Value, + label: &str, + lock_document: bool, + work: impl FnOnce(&JobCtx) -> Result + Send + 'static, + apply: impl FnOnce(&mut Session, T) -> Result + Send + 'static, + ) -> Result { + self.jobs.spawn = true; + let r = run(self, label, lock_document, work, apply); + self.jobs.spawn = false; + let pending = self.jobs.pending.take(); + match (r, pending) { + (Err(e), Some(job)) => { + job.ctx.cancel(); + Err(e) + } + (Err(e), None) => Err(e), + (Ok(_), Some(mut job)) => { + job.command = command.to_string(); + job.params = params; + job.journal = false; + let id = job.id; + self.jobs.running.push(job); + Ok(Started::Job(id)) + } + (Ok(v), None) => Ok(Started::Done(v)), + } + } + + /// Apply finished background jobs (on this thread) and return every job that ended since + /// the last call. Cheap when nothing runs; the UI calls it every frame. + pub fn poll_jobs(&mut self) -> Vec { + let mut i = 0; + while i < self.jobs.running.len() { + match self.jobs.running.get(i).map(|r| r.rx.try_recv()) { + Some(Ok(r)) => { + let job = self.jobs.running.remove(i); + self.finish_job(job, r); + } + Some(Err(std::sync::mpsc::TryRecvError::Empty)) => i += 1, + Some(Err(std::sync::mpsc::TryRecvError::Disconnected)) => { + let job = self.jobs.running.remove(i); + let label = job.label.clone(); + self.finish_job(job, Err(EngineError::Other(format!("`{label}` stopped unexpectedly; the document is unchanged")))); + } + None => break, + } + } + #[cfg(not(target_arch = "wasm32"))] + self.jobs.draining.retain(|h| !h.is_finished()); + std::mem::take(&mut self.jobs.events) + } + + /// Block until job `id` ends, apply it, and return its result (the control channel's and + /// MCP's default `wait`, and the CLI). `Err(Cancelled)` if it was cancelled. + pub fn wait_job(&mut self, id: JobId) -> Result { + if let Some(i) = self.jobs.running.iter().position(|r| r.id == id) { + let job = self.jobs.running.remove(i); + let r = job.rx.recv().unwrap_or_else(|_| Err(EngineError::Other(format!("`{}` stopped unexpectedly", job.label)))); + self.finish_job(job, r); + } + match self.jobs.recent.iter().rev().find(|j| j.id == id) { + Some(j) if j.state == "done" => Ok(j.result.clone().unwrap_or(Value::Null)), + Some(j) if j.state == "cancelled" => Err(EngineError::Cancelled), + Some(j) => Err(EngineError::Other(j.error.clone().unwrap_or_else(|| "the job failed".into()))), + None => Err(EngineError::Other(format!("no job {}", id.0))), + } + } + + /// Cancel job `id`: it stops at its next check, its result is discarded and the document + /// stays unchanged. The document is unlocked at once. False if no such job is running. + pub fn cancel_job(&mut self, id: JobId) -> bool { + let Some(i) = self.jobs.running.iter().position(|r| r.id == id) else { return false }; + let job = self.jobs.running.remove(i); + job.ctx.cancel(); + self.end_job(&job, JobOutcome::Cancelled); + #[cfg(not(target_arch = "wasm32"))] + if let Some(h) = job.handle { + self.jobs.draining.push(h); + } + true + } + + /// Cancel every running job (e.g. before quitting). + pub fn cancel_all_jobs(&mut self) { + let ids: Vec = self.jobs.running.iter().map(|r| r.id).collect(); + for id in ids { + self.cancel_job(id); + } + } + + /// Is any job running? (Cheap: the UI asks every frame.) + pub fn has_jobs(&self) -> bool { + !self.jobs.running.is_empty() + } + + /// Running job `id`. + pub fn job(&self, id: JobId) -> Option { + self.jobs.running.iter().find(|r| r.id == id).map(Running::info) + } + + /// Running jobs. + pub fn jobs(&self) -> Vec { + self.jobs.running.iter().map(Running::info).collect() + } + + /// Running jobs followed by the last few that ended (newest last), for `jobs.list`. + pub fn jobs_with_recent(&self) -> Vec { + self.jobs.recent.iter().cloned().chain(self.jobs()).collect() + } + + /// The running job that locks document `doc`, if any. + pub fn job_on(&self, doc: DocId) -> Option { + self.jobs.running.iter().find(|r| r.doc == Some(doc)).map(Running::info) + } + + /// The running job that locks the active document, if any. + pub fn active_job(&self) -> Option { + self.active().and_then(|st| self.job_on(st.doc.id)) + } + + /// Wait for background workers of cancelled jobs to exit (tests and benchmarks measure + /// cancel latency with it). Native only; returns at once on wasm. + pub fn join_cancelled_jobs(&mut self) { + #[cfg(not(target_arch = "wasm32"))] + for h in self.jobs.draining.drain(..) { + let _ = h.join(); + } + } + + /// Why command `id` can't run now because of a background job (None = no conflict). + pub(crate) fn job_conflict(&self, id: &str, journal: bool) -> Option { + if !journal || independent_of_document(id) { + return None; + } + let job = self.active_job()?; + Some(format!("“{}” is still running on this document; wait for it to finish or cancel it (Esc)", job.label)) + } + + /// Cancel the jobs editing document `doc` (it is closing). + pub(crate) fn cancel_jobs_on(&mut self, doc: DocId) { + let ids: Vec = self.jobs.running.iter().filter(|r| r.doc == Some(doc)).map(|r| r.id).collect(); + for id in ids { + self.cancel_job(id); + } + } + + /// Run the command `id`: inline, or (with `background`) as a job when it is one. + pub(crate) fn dispatch(&mut self, id: &str, params: Value, background: bool) -> Result { + let spec = crate::commands::find(id).ok_or_else(|| EngineError::UnknownCommand(id.to_string()))?; + if let Err(why) = (spec.enabled)(self) { + return Err(EngineError::Disabled(id.to_string(), why)); + } + if let Some(why) = self.job_conflict(id, spec.journal) { + return Err(EngineError::Disabled(id.to_string(), why)); + } + self.coalesce_request = params.get("coalesce").and_then(Value::as_str).map(str::to_string); + // Pixel commands follow the Channels panel target unless the caller names one. + let run_params = crate::channel_cmds::inject_target(self, id, crate::commands::inject_kind(id, params.clone())); + self.color_restrict = crate::channel_cmds::color_restriction(self, id, &run_params); + self.jobs.spawn = background; + // Last-resort guard (AGENTS.md, Never crash): a command that panics anyway fails with an + // error instead of taking the app down. `edit` only commits a document after its closure + // returns, so the documents are unchanged. + let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| (spec.run)(self, &run_params))) + .unwrap_or_else(|_| Err(EngineError::Other(format!("`{id}` failed with an internal error (logged); the document is unchanged")))); + self.jobs.spawn = false; + self.coalesce_request = None; + self.color_restrict = None; + let pending = self.jobs.pending.take(); + match (r, pending) { + (Err(e), Some(job)) => { + job.ctx.cancel(); + Err(e) + } + (Err(e), None) => Err(e), + (Ok(_), Some(mut job)) => { + job.command = id.to_string(); + job.params = params; + job.journal = spec.journal; + let jid = job.id; + self.jobs.running.push(job); + Ok(Started::Job(jid)) + } + (Ok(v), None) => { + self.after_command(id, params, spec.journal); + Ok(Started::Done(v)) + } + } + } + + /// Bookkeeping after a command (or a job's apply) succeeded. + fn after_command(&mut self, id: &str, params: Value, journal: bool) { + // A layer-mask view ends when another layer becomes active (#196). + if let Some(st) = self.active_mut() { + crate::mask_view_cmds::fix(st); + } + crate::edit_menu_cmds::after_command(self, id); + crate::automate_cmds::after_command(self, id); + self.sync_preset_store(); + if journal && !crate::brush_cmds::coalesce_journal(self, id, ¶ms) { + self.journal.push((id.to_string(), params)); + } + } + + /// Apply (or report the failure of) a job whose worker has delivered `r`. + fn finish_job(&mut self, mut job: Running, r: Result) { + let apply = job.apply.take(); + let outcome = match r { + Err(EngineError::Cancelled) => JobOutcome::Cancelled, + Err(e) => JobOutcome::Failed(e.to_string()), + Ok(_) if job.ctx.cancelled() => JobOutcome::Cancelled, + Ok(payload) => match self.apply_job(&job, apply, payload) { + Ok(v) => JobOutcome::Done(v), + Err(EngineError::Cancelled) => JobOutcome::Cancelled, + Err(e) => JobOutcome::Failed(e.to_string()), + }, + }; + self.end_job(&job, outcome); + } + + fn apply_job(&mut self, job: &Running, apply: Option, payload: Payload) -> Result { + let Running { label, doc, snapshot, coalesce, color_restrict, command, params, journal, .. } = job; + let prev_active = self.active; + let target = match doc { + Some(d) => { + let i = + self.documents().iter().position(|st| st.doc.id == *d).ok_or_else(|| EngineError::Other(format!("`{label}`: the document was closed")))?; + let unchanged = snapshot.as_ref().is_some_and(|w| self.documents().get(i).is_some_and(|st| Weak::ptr_eq(w, &Arc::downgrade(&st.doc)))); + if !unchanged { + return Err(EngineError::Other(format!("`{label}`: the document changed while it ran; the result was discarded"))); + } + self.active = Some(i); + Some(i) + } + None => None, + }; + self.coalesce_request = coalesce.clone(); + self.color_restrict = *color_restrict; + let r = match apply { + Some(f) => std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| f(self, payload))) + .unwrap_or_else(|_| Err(EngineError::Other(format!("`{label}` failed with an internal error (logged); the document is unchanged")))), + None => Err(EngineError::Other("internal error: job applied twice".into())), + }; + self.coalesce_request = None; + self.color_restrict = None; + // Keep the user's active document unless the job opened a new one. + if target.is_some() && self.active == target { + self.active = prev_active.filter(|i| *i < self.docs.len()).or(self.active); + } + let v = r?; + self.after_command(command, params.clone(), *journal); + Ok(v) + } + + fn end_job(&mut self, job: &Running, outcome: JobOutcome) { + let (state, result, error) = match &outcome { + JobOutcome::Done(v) => ("done", Some(v.clone()), None), + JobOutcome::Failed(e) => ("failed", None, Some(e.clone())), + JobOutcome::Cancelled => ("cancelled", None, None), + }; + let mut info = job.info(); + info.state = state; + info.result = result; + info.error = error; + if state == "done" { + info.progress = 1.0; + } + self.jobs.recent.push(info); + if self.jobs.recent.len() > RECENT { + let extra = self.jobs.recent.len() - RECENT; + self.jobs.recent.drain(..extra); + } + self.jobs.events.push(JobEvent { id: job.id, command: job.command.clone(), label: job.label.clone(), document: job.doc, outcome }); + } +} + +/// Engine commands for the job table (the control channel and MCP use them too). +pub fn specs() -> Vec { + use crate::commands::CommandSpec; + vec![ + CommandSpec { + id: "jobs.list", + label: "List Background Jobs", + menu: &[], + shortcut: None, + params: "{}", + enabled: |_| Ok(()), + run: |s, _| Ok(json!({"jobs": s.jobs_with_recent()})), + journal: false, + }, + CommandSpec { + id: "jobs.cancel", + label: "Cancel Background Job", + menu: &[], + shortcut: None, + params: r##"{"job":u64? (default: every running job)}"##, + enabled: |_| Ok(()), + run: |s, p| match p.get("job") { + None | Some(Value::Null) => { + let n = s.jobs.running.len(); + s.cancel_all_jobs(); + Ok(json!({"cancelled": n})) + } + Some(v) => { + let id = v.as_u64().ok_or_else(|| EngineError::BadParams { cmd: "jobs.cancel".into(), msg: "`job` must be a job id".into() })?; + if s.cancel_job(JobId(id)) { Ok(json!({"cancelled": 1})) } else { Err(EngineError::Other(format!("no running job {id}"))) } + } + }, + journal: false, + }, + ] +} + +#[cfg(test)] +#[path = "jobs_tests.rs"] +mod tests; diff --git a/crates/engine/src/jobs_tests.rs b/crates/engine/src/jobs_tests.rs new file mode 100644 index 0000000000..32dd1413b5 --- /dev/null +++ b/crates/engine/src/jobs_tests.rs @@ -0,0 +1,333 @@ +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use serde_json::json; + +use super::*; +use crate::Session; + +/// A document `w`×`h` with a textured pixel layer (so blurs change every pixel). +fn session(w: u32, h: u32) -> Session { + let mut s = Session::new(); + s.execute("file.new", json!({"width": w, "height": h})).unwrap(); + s.execute("layer.new.layer", json!({})).unwrap(); + s.edit("pattern", |doc, active| { + let surf = doc.layer_mut(active.unwrap()).unwrap().surface_mut().unwrap(); + let r = photocraft_geom::Rect::new(0, 0, w as i32, h as i32); + let mut px = Vec::with_capacity((w * h * 4) as usize); + for y in 0..h { + for x in 0..w { + let v = ((x * 7 + y * 13) % 31) as f32 / 30.0; + px.extend_from_slice(&[v, 1.0 - v, ((x / 5 + y / 3) % 2) as f32, 1.0]); + } + } + surf.write_region(r, &px); + Ok(()) + }) + .unwrap(); + s +} + +fn pixels(s: &Session) -> Vec { + let d = s.active().unwrap(); + let l = d.doc.layer(d.active_layer.unwrap()).unwrap(); + l.surface().unwrap().read_region(d.doc.bounds()) +} + +/// Poll until job `id` ends (or 60 s pass); returns its event. +fn wait_event(s: &mut Session, id: JobId) -> JobEvent { + let t = Instant::now(); + loop { + if let Some(e) = s.poll_jobs().into_iter().find(|e| e.id == id) { + return e; + } + assert!(t.elapsed() < Duration::from_secs(60), "job {id:?} did not finish"); + std::thread::sleep(Duration::from_millis(2)); + } +} + +fn job(r: Started) -> JobId { + match r { + Started::Job(id) => id, + Started::Done(v) => panic!("expected a background job, ran inline: {v}"), + } +} + +#[test] +fn a_job_applies_as_one_undo_step_and_matches_the_inline_result() { + let mut inline = session(300, 200); + inline.execute("filter.blur.gaussianBlur", json!({"radius": 4})).unwrap(); + + let mut s = session(300, 200); + let before = pixels(&s); + let steps = s.active().unwrap().history.past_len(); + let journal = s.journal.len(); + let id = job(s.start("filter.blur.gaussianBlur", json!({"radius": 4})).unwrap()); + // The canvas shows the pre-job state until the job is applied. + assert_eq!(pixels(&s), before); + let e = wait_event(&mut s, id); + let JobOutcome::Done(v) = &e.outcome else { panic!("{e:?}") }; + assert!(v["filter"].is_object(), "{v}"); + assert_eq!(e.command, "filter.blur.gaussianBlur"); + assert_eq!(s.active().unwrap().history.past_len(), steps + 1, "one undo step"); + assert_eq!(s.journal.len(), journal + 1, "journaled like execute"); + assert_eq!(s.journal.last().unwrap().0, "filter.blur.gaussianBlur"); + assert_eq!(pixels(&s), pixels(&inline), "same pixels as the synchronous command"); + assert!(s.undo()); + assert_eq!(pixels(&s), before); + assert!(s.jobs().is_empty()); +} + +#[test] +fn cancel_leaves_the_document_byte_identical() { + let mut s = session(1600, 1200); + let doc_before = s.active().unwrap().doc.clone(); + let before = pixels(&s); + let (rev, steps) = (s.active().unwrap().revision, s.active().unwrap().history.past_len()); + let id = job(s.start("filter.blur.gaussianBlur", json!({"radius": 60})).unwrap()); + assert!(s.cancel_job(id)); + let e = wait_event(&mut s, id); + assert_eq!(e.outcome, JobOutcome::Cancelled); + s.join_cancelled_jobs(); + // Nothing arrives later either. + assert!(s.poll_jobs().is_empty()); + let st = s.active().unwrap(); + assert!(Arc::ptr_eq(&st.doc, &doc_before), "the document snapshot was not replaced"); + assert_eq!((st.revision, st.history.past_len()), (rev, steps)); + assert_eq!(pixels(&s), before); + assert!(matches!(s.wait_job(id), Err(EngineError::Cancelled))); +} + +#[test] +fn a_panic_in_the_worker_becomes_an_error_and_the_document_is_unchanged() { + let mut s = session(64, 64); + let doc_before = s.active().unwrap().doc.clone(); + let id = job(s + .start_job( + "test.panics", + json!({}), + "Panicking job", + true, + |_ctx| -> Result<()> { + let v: Vec = Vec::new(); + // An out-of-bounds index: a panic inside the worker. + #[allow(clippy::indexing_slicing)] + let _ = v[usize::MAX / 2]; + Ok(()) + }, + |_s, ()| Ok(json!(null)), + ) + .unwrap()); + let e = wait_event(&mut s, id); + let JobOutcome::Failed(msg) = &e.outcome else { panic!("{e:?}") }; + assert!(msg.contains("internal error"), "{msg}"); + assert!(Arc::ptr_eq(&s.active().unwrap().doc, &doc_before)); + // The session still works. + s.execute("filter.blur.gaussianBlur", json!({"radius": 1})).unwrap(); +} + +#[test] +fn progress_is_monotonic_and_reaches_one() { + let ctx = JobCtx::new(); + ctx.progress(0.5, "half"); + ctx.progress(0.3, ""); + assert_eq!(ctx.fraction(), 0.5, "progress never goes backwards"); + assert_eq!(ctx.message(), "half"); + ctx.progress(f32::NAN, ""); + ctx.progress(7.0, ""); + assert_eq!(ctx.fraction(), 1.0); + + let mut s = session(2000, 1500); + let id = job(s.start("filter.blur.gaussianBlur", json!({"radius": 30})).unwrap()); + let mut seen = Vec::new(); + let e = loop { + if let Some(j) = s.jobs().into_iter().find(|j| j.id == id) { + seen.push(j.progress); + assert_eq!(j.label, "Gaussian Blur"); + assert_eq!(j.state, "running"); + } + if let Some(e) = s.poll_jobs().into_iter().find(|e| e.id == id) { + break e; + } + std::thread::sleep(Duration::from_millis(1)); + }; + assert!(matches!(e.outcome, JobOutcome::Done(_)), "{e:?}"); + assert!(seen.windows(2).all(|w| w[0] <= w[1]), "monotonic: {seen:?}"); + let info = s.jobs_with_recent().into_iter().find(|j| j.id == id).unwrap(); + assert_eq!((info.state, info.progress), ("done", 1.0)); +} + +/// A fake document job that runs until `gate` opens (or it is cancelled), then inverts nothing +/// but records one undo step named "Gated Job". +pub(super) fn gated_job(s: &mut Session, gate: &Arc) -> JobId { + let gate = gate.clone(); + job(s + .start_job( + "test.gated", + json!({}), + "Gated Job", + true, + move |ctx| { + while !gate.load(std::sync::atomic::Ordering::Relaxed) { + ctx.check()?; + ctx.progress(0.5, "waiting"); + std::thread::sleep(Duration::from_millis(1)); + } + Ok(()) + }, + |s, ()| { + s.edit("Gated Job", |_, _| Ok(()))?; + Ok(json!({"gated": true})) + }, + ) + .unwrap()) +} + +#[test] +fn a_running_job_locks_its_document_but_not_others() { + let mut s = session(64, 64); + let gate = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let id = gated_job(&mut s, &gate); + // Edits of the busy document are disabled with a message naming the job. + let err = s.execute("layer.new.layer", json!({})).unwrap_err(); + assert!(matches!(&err, EngineError::Disabled(_, why) if why.contains("Gated Job")), "{err}"); + assert!(!s.is_enabled("layer.new.layer")); + assert!(s.disabled_reason("filter.blur.gaussianBlur").unwrap().contains("still running")); + assert!(!s.undo(), "undo waits for the job"); + // Queries and document-independent commands still run. + s.execute("jobs.list", json!({})).unwrap(); + // A new document is free to edit while the first one is busy. + s.execute("file.new", json!({"width": 8, "height": 8})).unwrap(); + s.execute("layer.new.layer", json!({})).unwrap(); + let r = s.execute("jobs.list", json!({})).unwrap(); + assert_eq!(r["jobs"][0]["id"], json!(id.0)); + // The result applies to its own document, and the active document stays the new one. + gate.store(true, std::sync::atomic::Ordering::Relaxed); + let e = wait_event(&mut s, id); + assert!(matches!(e.outcome, JobOutcome::Done(_)), "{e:?}"); + assert_eq!(s.active_index(), Some(1)); + assert_eq!(s.documents()[0].history.past_len(), 3); +} + +#[test] +fn closing_the_document_cancels_its_job() { + let mut s = session(1600, 1200); + let id = job(s.start("filter.blur.gaussianBlur", json!({"radius": 80})).unwrap()); + s.close(0); + let e = wait_event(&mut s, id); + assert_eq!(e.outcome, JobOutcome::Cancelled); +} + +#[test] +fn jobs_cancel_command_and_bad_params() { + let mut s = session(1600, 1200); + let id = job(s.start("filter.blur.gaussianBlur", json!({"radius": 80})).unwrap()); + assert!(s.execute("jobs.cancel", json!({"job": "x"})).is_err()); + assert!(s.execute("jobs.cancel", json!({"job": 999})).is_err()); + s.execute("jobs.cancel", json!({"job": id.0})).unwrap(); + assert_eq!(wait_event(&mut s, id).outcome, JobOutcome::Cancelled); + assert_eq!(s.execute("jobs.cancel", json!({})).unwrap()["cancelled"], 0); +} + +#[test] +fn non_job_commands_still_finish_inline_with_start() { + let mut s = session(16, 16); + let r = s.start("layer.new.layer", json!({})).unwrap(); + assert!(matches!(r, Started::Done(_)), "{r:?}"); + // Errors are reported as before. + assert!(matches!(s.start("no.such.command", json!({})), Err(EngineError::UnknownCommand(_)))); +} + +#[test] +fn content_aware_fill_and_scale_run_as_jobs() { + let mut inline = session(160, 120); + let mut s = session(160, 120); + for t in [&mut inline, &mut s] { + t.execute("select.rect", json!({"x": 60, "y": 40, "width": 30, "height": 30})).unwrap(); + } + inline.execute("edit.contentAwareFill", json!({})).unwrap(); + let id = job(s.start("edit.contentAwareFill", json!({})).unwrap()); + let e = wait_event(&mut s, id); + assert!(matches!(e.outcome, JobOutcome::Done(_)), "{e:?}"); + assert_eq!(pixels(&s), pixels(&inline), "same fill as the synchronous command"); + + s.execute("select.deselect", json!({})).unwrap(); + let id = job(s.start("edit.contentAwareScale", json!({"width": 120})).unwrap()); + let e = wait_event(&mut s, id); + let JobOutcome::Done(v) = &e.outcome else { panic!("{e:?}") }; + assert_eq!(v["to"][0], 120); +} + +#[test] +fn open_runs_as_a_job_and_cancel_adds_nothing() { + let mut src = session(40, 30); + let doc = (*src.active().unwrap().doc).clone(); + let bytes = photocraft_io::export(&doc, "x.psd", &Default::default()).unwrap().bytes; + src.close(0); + let mut s = Session::new(); + let id = job(s.start_open("x.psd", OpenSource::Bytes(Arc::new(bytes.clone()))).unwrap()); + let e = wait_event(&mut s, id); + let JobOutcome::Done(v) = &e.outcome else { panic!("{e:?}") }; + assert_eq!(e.command, OPEN_JOB); + assert_eq!(v["document"], 0); + assert_eq!(s.documents().len(), 1); + assert_eq!(s.documents()[0].doc.size.width, 40); + + let id = job(s.start_open("y.psd", OpenSource::Bytes(Arc::new(bytes))).unwrap()); + s.cancel_job(id); + assert_eq!(wait_event(&mut s, id).outcome, JobOutcome::Cancelled); + assert_eq!(s.documents().len(), 1, "a cancelled open adds no document"); + + let id = job(s.start_open("missing.psd", OpenSource::Path("/no/such/file.psd".into())).unwrap()); + assert!(matches!(wait_event(&mut s, id).outcome, JobOutcome::Failed(_))); +} + +#[test] +fn wait_job_blocks_until_applied() { + let mut s = session(400, 300); + let id = job(s.start("filter.blur.gaussianBlur", json!({"radius": 3})).unwrap()); + let v = s.wait_job(id).unwrap(); + assert!(v["filter"].is_object()); + assert_eq!(s.active().unwrap().history.past_len(), 3); + // Waiting again reports the recorded result. + assert_eq!(s.wait_job(id).unwrap(), v); + assert!(s.wait_job(JobId(12345)).is_err()); +} + +/// Cancel latency on a 24 MP document (6000×4000): from `cancel_job` until the worker thread +/// has exited. Release only: debug builds are ~20× slower per tile. +#[test] +// Wall-clock timing depends on machine load (the release corpus job runs every test), so it is +// opt-in; perf_scenarios tracks the same numbers as P25/P49. +#[ignore = "timing: cargo test --release -p photocraft-engine --lib cancel_takes -- --ignored"] +fn cancel_takes_effect_within_200_ms_on_24_mp() { + for (cmd, params, select) in [("filter.blur.gaussianBlur", json!({"radius": 50}), false), ("edit.contentAwareFill", json!({}), true)] { + let mut s = session(6000, 4000); + if select { + s.execute("select.rect", json!({"x": 2000, "y": 1500, "width": 1200, "height": 900})).unwrap(); + } + let doc_before = s.active().unwrap().doc.clone(); + // Cancel at several points of the run (early, middle, late stages). + let mut worst: f64 = 0.0; + for after_ms in [20, 150, 400, 600, 800, 1000, 1300] { + let id = job(s.start(cmd, params.clone()).unwrap()); + std::thread::sleep(Duration::from_millis(after_ms)); + if s.jobs().iter().all(|j| j.id != id) || s.jobs().iter().any(|j| j.id == id && j.progress >= 1.0) { + let _ = wait_event(&mut s, id); + s.undo(); + continue; + } + let at = s.jobs().iter().find(|j| j.id == id).map(|j| j.progress).unwrap_or(0.0); + let t = Instant::now(); + assert!(s.cancel_job(id)); + s.join_cancelled_jobs(); + let ms = t.elapsed().as_secs_f64() * 1000.0; + eprintln!("{cmd}: cancelled after {after_ms} ms (at {:.0} %), latency {ms:.1} ms", at * 100.0); + worst = worst.max(ms); + assert_eq!(wait_event(&mut s, id).outcome, JobOutcome::Cancelled); + assert!(Arc::ptr_eq(&s.active().unwrap().doc, &doc_before)); + } + eprintln!("{cmd}: worst cancel latency {worst:.1} ms"); + assert!(worst < 200.0, "{cmd}: cancel took {worst:.1} ms"); + } +} diff --git a/crates/engine/src/lib.rs b/crates/engine/src/lib.rs index aa980fb7cf..dbbe92ac1b 100644 --- a/crates/engine/src/lib.rs +++ b/crates/engine/src/lib.rs @@ -39,6 +39,7 @@ pub mod gradient_fill_cmds; pub mod group_view_cmds; pub mod image_cmds; pub mod inspect; +pub mod jobs; pub mod layer_menu_cmds; pub mod layer_multi_cmds; pub mod layer_style; @@ -108,6 +109,9 @@ pub enum EngineError { NoLayer(LayerId), #[error("{0}")] Other(String), + /// A background job was cancelled (see [`jobs`]); nothing changed. + #[error("cancelled")] + Cancelled, } pub type Result = std::result::Result; @@ -265,6 +269,8 @@ pub struct Session { /// Persistent brush preset store (desktop only; `None` keeps presets session-only, as in /// headless and test sessions). See `preset_store`. pub preset_store: Option, + /// Background jobs (see [`jobs`]). + jobs: jobs::Jobs, } impl Session { @@ -315,6 +321,9 @@ impl Session { return None; } smart_cmds::on_close(self, index); + if let Some(id) = self.docs.get(index).map(|d| d.doc.id) { + self.cancel_jobs_on(id); + } let d = self.docs.remove(index); self.active = if self.docs.is_empty() { None } else { Some(index.min(self.docs.len() - 1)) }; Some(d) @@ -325,39 +334,27 @@ impl Session { /// Any command accepts an optional `"coalesce": ""` param: consecutive edits with the same /// key (and no other edit, undo or redo in between) share one history step, like Photoshop's /// single "Edit Type Layer" step for a whole typing session or one step per slider drag. + /// + /// Runs synchronously, including job-capable commands (see [`jobs`]; [`Session::start`] runs + /// those in the background). pub fn execute(&mut self, id: &str, params: Value) -> Result { - let spec = commands::find(id).ok_or_else(|| EngineError::UnknownCommand(id.to_string()))?; - if let Err(why) = (spec.enabled)(self) { - return Err(EngineError::Disabled(id.to_string(), why)); + match self.dispatch(id, params, false)? { + jobs::Started::Done(v) => Ok(v), + // Not reached: inline dispatch never starts a job. Waiting is still correct. + jobs::Started::Job(j) => self.wait_job(j), } - self.coalesce_request = params.get("coalesce").and_then(Value::as_str).map(str::to_string); - // Pixel commands follow the Channels panel target unless the caller names one. - let run_params = channel_cmds::inject_target(self, id, commands::inject_kind(id, params.clone())); - self.color_restrict = channel_cmds::color_restriction(self, id, &run_params); - // Last-resort guard (AGENTS.md, Never crash): a command that panics anyway fails with an - // error instead of taking the app down. `edit` only commits a document after its closure - // returns, so the documents are unchanged. - let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| (spec.run)(self, &run_params))) - .unwrap_or_else(|_| Err(EngineError::Other(format!("`{id}` failed with an internal error (logged); the document is unchanged")))); - self.coalesce_request = None; - self.color_restrict = None; - let r = r?; - // A layer-mask view ends when another layer becomes active (#196). - if let Some(st) = self.active_mut() { - mask_view_cmds::fix(st); - } - edit_menu_cmds::after_command(self, id); - automate_cmds::after_command(self, id); - self.sync_preset_store(); - if spec.journal && !brush_cmds::coalesce_journal(self, id, ¶ms) { - self.journal.push((id.to_string(), params)); - } - Ok(r) } /// Is the command currently runnable? (drives menu enablement) pub fn is_enabled(&self, id: &str) -> bool { - commands::find(id).is_some_and(|s| (s.enabled)(self).is_ok()) + commands::find(id).is_some_and(|s| (s.enabled)(self).is_ok() && self.job_conflict(id, s.journal).is_none()) + } + + /// Why the command can't run now (None = it can): its own precondition, or a background job + /// running on the active document. + pub fn disabled_reason(&self, id: &str) -> Option { + let Some(s) = commands::find(id) else { return Some(format!("unknown command `{id}`")) }; + (s.enabled)(self).err().or_else(|| self.job_conflict(id, s.journal)) } /// Apply an undoable edit to the active document. @@ -407,6 +404,10 @@ impl Session { } pub fn undo(&mut self) -> bool { + // A background job is computing from the current state: it must not move under it. + if self.active_job().is_some() { + return false; + } let Some(st) = self.active_mut() else { return false }; st.coalesce = None; match st.history.undo(st.doc.clone()) { @@ -422,6 +423,9 @@ impl Session { } pub fn redo(&mut self) -> bool { + if self.active_job().is_some() { + return false; + } let Some(st) = self.active_mut() else { return false }; st.coalesce = None; match st.history.redo(st.doc.clone()) { diff --git a/crates/engine/src/photo_cmds.rs b/crates/engine/src/photo_cmds.rs index 4893ea1f20..50255c9440 100644 --- a/crates/engine/src/photo_cmds.rs +++ b/crates/engine/src/photo_cmds.rs @@ -403,83 +403,114 @@ fn photomerge(s: &mut Session, p: &Value) -> Result { if srcs.len() < 2 { return Err(bad(cmd, "Photomerge needs two or more images")); } - let focal35 = p.get("focalLength").and_then(Value::as_f64).filter(|f| *f > 0.0).or_else(|| exif_focal(&srcs)); - let images: Vec<(&Surface, Rect)> = srcs.iter().map(|s| (&s.surf, Rect::new(0, 0, s.w as i32, s.h as i32))).collect(); - let al = register(&images, layout, None, geometric, focal35) - .ok_or_else(|| EngineError::Other("Photomerge couldn't find enough matching detail between the images".into()))?; - let t_reg = t0.ms() / 1000.0; - let placed: Vec = (0..srcs.len()).filter(|&i| al.placements[i].is_some()).collect(); - // Parallel to `placed`. - let placements: Vec<&Placement> = placed.iter().filter_map(|&i| al.placements[i].as_ref()).collect(); - let failed: Vec = (0..srcs.len()).filter(|&i| al.placements[i].is_none()).map(|i| srcs[i].name.clone()).collect(); - // Canvas: union of the placed images. - let (mut x0, mut y0, mut x1, mut y1) = (f64::MAX, f64::MAX, f64::MIN, f64::MIN); - for (&i, pl) in placed.iter().zip(&placements) { - let b = pl.bounds(srcs[i].w as f64, srcs[i].h as f64); - x0 = x0.min(b[0]); - y0 = y0.min(b[1]); - x1 = x1.max(b[2]); - y1 = y1.max(b[3]); - } - let (cw, ch) = ((x1 - x0).ceil(), (y1 - y0).ceil()); - if !(cw.is_finite() && ch.is_finite()) || cw > 30_000.0 || ch > 30_000.0 || cw * ch > 6.0e8 { - return Err(EngineError::Other(format!("the panorama would be {cw}×{ch} px; try another layout"))); - } - let canvas = Rect::new(0, 0, cw as i32, ch as i32); - let offset = (-x0.floor(), -y0.floor()); - let warped: Vec = { - placed - .iter() - .zip(&placements) - .map(|(&i, pl)| warp_placed(&srcs[i].surf, Rect::new(0, 0, srcs[i].w as i32, srcs[i].h as i32), (0.0, 0.0), pl, offset, Interp::Bicubic)) - .collect() - }; - let t_warp = t0.ms() / 1000.0; - let mut doc = Document::new(format!("Untitled_Panorama{}", s.documents().len() + 1), Size::new(cw as u32, ch as u32), fmt.mode, fmt.sample); - doc.icc_profile = icc; - let mut info = json!({}); - let ref_pos = placed.iter().position(|&i| i == al.reference).unwrap_or(0); - if blend { - let pls: Vec = placements.iter().map(|p| (*p).clone()).collect(); - let sizes: Vec<(usize, usize)> = placed.iter().map(|&i| (srcs[i].w, srcs[i].h)).collect(); - let order = seam_order(&warped, ref_pos); - let geo = vignette.then_some((pls.as_slice(), sizes.as_slice(), offset)); - let b = seam_blend(&warped, canvas, &order, true, true, geo); - for (k, (px, mask)) in b.layers.into_iter().enumerate() { - let mut l = Layer::new(srcs[placed[k]].name.clone(), LayerContent::Raster(px)); - let mut m = LayerMask::hide_all(); - m.surface = mask; - l.mask = Some(m); - doc.layers.push(l); - } - info = json!({"gains": b.gains, "vignette": b.vignette}); - if fill { - let filled = content_aware_fill(&b.composite, canvas); - doc.layers.push(Layer::new("Content-Aware Fill", LayerContent::Raster(filled))); - } - } else { - for (k, px) in warped.into_iter().enumerate() { - doc.layers.push(Layer::new(srcs[placed[k]].name.clone(), LayerContent::Raster(px))); - } - } - let n = doc.layers.len(); - let idx = s.add_document(doc, None); - Ok(json!({ - "document": idx, - "layout": al.layout.name(), - "reference": srcs[al.reference].name, - "focalLength35": al.focal / (srcs.iter().map(|s| s.w.max(s.h)).max().unwrap_or(1) as f64) * 36.0, - "geometricK1": al.k1, - "rms": al.rms, - "pairs": al.pairs, - "placed": placed.iter().map(|&i| srcs[i].name.clone()).collect::>(), - "failed": failed, - "width": cw as u32, - "height": ch as u32, - "layers": n, - "photometric": info, - "ms": {"register": t_reg * 1000.0, "warp": (t_warp - t_reg) * 1000.0, "total": t0.ms()}, - })) + let focal_param = p.get("focalLength").and_then(Value::as_f64).filter(|f| *f > 0.0); + let doc_name = format!("Untitled_Panorama{}", s.documents().len() + 1); + // A background job when started with `Session::start` (#210): alignment, warping and + // blending run on a worker, checking for cancellation between stages and images. + crate::jobs::run( + s, + "Photomerge", + false, + move |ctx| { + ctx.progress(0.0, "Photomerge: aligning images"); + let focal35 = focal_param.or_else(|| exif_focal(&srcs)); + let images: Vec<(&Surface, Rect)> = srcs.iter().map(|s| (&s.surf, Rect::new(0, 0, s.w as i32, s.h as i32))).collect(); + let al = register(&images, layout, None, geometric, focal35) + .ok_or_else(|| EngineError::Other("Photomerge couldn't find enough matching detail between the images".into()))?; + let t_reg = t0.ms() / 1000.0; + ctx.check()?; + ctx.progress(0.35, "Photomerge: warping images"); + let placed: Vec = (0..srcs.len()).filter(|&i| al.placements[i].is_some()).collect(); + // Parallel to `placed`. + let placements: Vec<&Placement> = placed.iter().filter_map(|&i| al.placements[i].as_ref()).collect(); + let failed: Vec = (0..srcs.len()).filter(|&i| al.placements[i].is_none()).map(|i| srcs[i].name.clone()).collect(); + // Canvas: union of the placed images. + let (mut x0, mut y0, mut x1, mut y1) = (f64::MAX, f64::MAX, f64::MIN, f64::MIN); + for (&i, pl) in placed.iter().zip(&placements) { + let b = pl.bounds(srcs[i].w as f64, srcs[i].h as f64); + x0 = x0.min(b[0]); + y0 = y0.min(b[1]); + x1 = x1.max(b[2]); + y1 = y1.max(b[3]); + } + let (cw, ch) = ((x1 - x0).ceil(), (y1 - y0).ceil()); + if !(cw.is_finite() && ch.is_finite()) || cw > 30_000.0 || ch > 30_000.0 || cw * ch > 6.0e8 { + return Err(EngineError::Other(format!("the panorama would be {cw}×{ch} px; try another layout"))); + } + let canvas = Rect::new(0, 0, cw as i32, ch as i32); + let offset = (-x0.floor(), -y0.floor()); + let warped: Vec = { + placed + .iter() + .zip(&placements) + .enumerate() + .map(|(k, (&i, pl))| { + ctx.check()?; + ctx.progress(0.35 + 0.25 * k as f32 / placed.len().max(1) as f32, ""); + Ok(warp_placed(&srcs[i].surf, Rect::new(0, 0, srcs[i].w as i32, srcs[i].h as i32), (0.0, 0.0), pl, offset, Interp::Bicubic)) + }) + .collect::>>()? + }; + let t_warp = t0.ms() / 1000.0; + ctx.check()?; + ctx.progress(0.6, "Photomerge: blending images"); + let mut doc = Document::new(doc_name, Size::new(cw as u32, ch as u32), fmt.mode, fmt.sample); + doc.icc_profile = icc; + let mut info = json!({}); + let ref_pos = placed.iter().position(|&i| i == al.reference).unwrap_or(0); + if blend { + let pls: Vec = placements.iter().map(|p| (*p).clone()).collect(); + let sizes: Vec<(usize, usize)> = placed.iter().map(|&i| (srcs[i].w, srcs[i].h)).collect(); + let order = seam_order(&warped, ref_pos); + let geo = vignette.then_some((pls.as_slice(), sizes.as_slice(), offset)); + let b = seam_blend(&warped, canvas, &order, true, true, geo); + for (k, (px, mask)) in b.layers.into_iter().enumerate() { + let mut l = Layer::new(srcs[placed[k]].name.clone(), LayerContent::Raster(px)); + let mut m = LayerMask::hide_all(); + m.surface = mask; + l.mask = Some(m); + doc.layers.push(l); + } + info = json!({"gains": b.gains, "vignette": b.vignette}); + ctx.check()?; + if fill { + ctx.progress(0.85, "Photomerge: filling edges"); + let filled = content_aware_fill(&b.composite, canvas); + doc.layers.push(Layer::new("Content-Aware Fill", LayerContent::Raster(filled))); + } + } else { + for (k, px) in warped.into_iter().enumerate() { + doc.layers.push(Layer::new(srcs[placed[k]].name.clone(), LayerContent::Raster(px))); + } + } + let n = doc.layers.len(); + ctx.check()?; + ctx.progress(1.0, ""); + Ok(( + doc, + json!({ + "document": Value::Null, + "layout": al.layout.name(), + "reference": srcs[al.reference].name, + "focalLength35": al.focal / (srcs.iter().map(|s| s.w.max(s.h)).max().unwrap_or(1) as f64) * 36.0, + "geometricK1": al.k1, + "rms": al.rms, + "pairs": al.pairs, + "placed": placed.iter().map(|&i| srcs[i].name.clone()).collect::>(), + "failed": failed, + "width": cw as u32, + "height": ch as u32, + "layers": n, + "photometric": info, + "ms": {"register": t_reg * 1000.0, "warp": (t_warp - t_reg) * 1000.0, "total": t0.ms()}, + }), + )) + }, + |s, (doc, mut info): (Document, Value)| { + info["document"] = json!(s.add_document(doc, None)); + Ok(info) + }, + ) } /// Fills the transparent areas of a composite (inside its bounding rectangle) with diff --git a/crates/engine/src/preset_import_cmds.rs b/crates/engine/src/preset_import_cmds.rs index 44a7f9a712..7799f3e7cd 100644 --- a/crates/engine/src/preset_import_cmds.rs +++ b/crates/engine/src/preset_import_cmds.rs @@ -56,13 +56,35 @@ fn unique(name: &str, taken: &[String]) -> String { fn import_abr(s: &mut Session, p: &Value) -> Result { let cmd = "brush.presets.importAbr"; - let (bytes, stem) = file_bytes(p, cmd, ".abr file")?; - let group = match p.get("group") { - Some(v) => v.as_str().map(str::trim).filter(|g| !g.is_empty()).ok_or_else(|| bad(cmd, "`group` must be a non-empty string"))?.to_string(), - None if !stem.is_empty() => stem, - None => "Imported Brushes".to_string(), - }; - let imp = photocraft_io::abr_map::read_abr(&bytes, &group).map_err(|e| bad(cmd, e))?; + let label = "Import Brushes"; + // A background job when started with `Session::start` (#210): reading and decoding the + // file run on a worker, cancellable before each brush; the library changes on apply. + let params = p.clone(); + let params_apply = p.clone(); + crate::jobs::run( + s, + label, + false, + move |ctx| { + ctx.progress(0.0, "Importing brushes"); + let p = ¶ms; + let (bytes, stem) = file_bytes(p, cmd, ".abr file")?; + let group = match p.get("group") { + Some(v) => v.as_str().map(str::trim).filter(|g| !g.is_empty()).ok_or_else(|| bad(cmd, "`group` must be a non-empty string"))?.to_string(), + None if !stem.is_empty() => stem, + None => "Imported Brushes".to_string(), + }; + ctx.check()?; + let imp = ctx + .stage(0.05, 1.0, "Importing brushes", |ctl| photocraft_io::abr_map::read_abr_with(&bytes, &group, ctl)) + .map_err(|e| if ctx.cancelled() { EngineError::Cancelled } else { bad(cmd, e) })?; + Ok((group, imp)) + }, + move |s, (group, imp)| add_abr_presets(s, ¶ms_apply, group, imp), + ) +} + +fn add_abr_presets(s: &mut Session, p: &Value, group: String, imp: photocraft_io::abr_map::AbrImport) -> Result { // Re-importing a file replaces its group instead of duplicating it. if p.get("replace").and_then(Value::as_bool).unwrap_or(true) { s.tools.presets.retain(|x| x.builtin || x.group != group); diff --git a/crates/io/src/abr_map.rs b/crates/io/src/abr_map.rs index 487f9856b0..35faa3bbf7 100644 --- a/crates/io/src/abr_map.rs +++ b/crates/io/src/abr_map.rs @@ -32,16 +32,39 @@ pub struct AbrImport { /// Parse an `.abr` file into presets in group `group` (usually the file name). pub fn read_abr(bytes: &[u8], group: &str) -> Result { + read_abr_with(bytes, group, &photocraft_raster::Interrupt::NONE) +} + +/// [`read_abr`] that checks `ctl` before each brush (and reports progress): a cancelled import +/// fails with "cancelled". +pub fn read_abr_with(bytes: &[u8], group: &str, ctl: &photocraft_raster::Interrupt) -> Result { let f = abr::parse(bytes).map_err(|e| format!("not a readable Photoshop brush file: {e}"))?; - Ok(map_file(&f, group)) + // Parsing is the small part; mapping (tip decoding) reports 10–100 %. + ctl.progress(0.1); + let cancel = || ctl.cancelled(); + let progress = |p: f32| ctl.progress(0.1 + 0.9 * p); + map_file_with(&f, group, &photocraft_raster::Interrupt::new(&cancel, &progress)).map_err(|e| e.to_string()) } /// Map a parsed file. pub fn map_file(f: &AbrFile, group: &str) -> AbrImport { + // Never cancelled, so always `Ok`. + map_file_with(f, group, &photocraft_raster::Interrupt::NONE).unwrap_or_default() +} + +/// [`map_file`] that checks `ctl` before each brush. +pub fn map_file_with(f: &AbrFile, group: &str, ctl: &photocraft_raster::Interrupt) -> Result { let mut out = AbrImport { version: f.version, warnings: f.warnings.clone(), ..Default::default() }; let mut m = Mapper { file: f, warnings: BTreeSet::new(), unknown: BTreeSet::new() }; + let total = f.legacy.len().max(f.samples.len()).max(f.presets.len()).max(1) as f32; + let step = |i: usize| -> Result<(), photocraft_raster::Cancelled> { + ctl.check()?; + ctl.progress(i as f32 / total); + Ok(()) + }; if !f.legacy.is_empty() { for (i, b) in f.legacy.iter().enumerate() { + step(i)?; let spacing = if b.spacing == 0 { 0.25 } else { f32::from(b.spacing) / 100.0 }; let base = BrushSettings { pressure_size: false, spacing, ..Default::default() }; let (brush, default_name) = match &b.tip { @@ -66,12 +89,14 @@ pub fn map_file(f: &AbrFile, group: &str) -> AbrImport { } else if f.presets.is_empty() { // Tips without a settings section (early v6 files): one preset per tip. for (i, s) in f.samples.iter().enumerate() { + step(i)?; let (tip, size) = m.tip(s); let brush = BrushSettings { pressure_size: false, spacing: 0.25, size, tip, ..Default::default() }; out.presets.push(BrushPreset { name: format!("Sampled Brush {}", i + 1), brush, builtin: false, group: group.to_string() }); } } else { for (i, d) in f.presets.iter().enumerate() { + step(i)?; let name = text(d, "Nm ").filter(|n| !n.trim().is_empty()).map(|n| n.trim().to_string()).unwrap_or_else(|| format!("Brush {}", i + 1)); match m.preset(d) { Some(brush) => out.presets.push(BrushPreset { name, brush, builtin: false, group: group.to_string() }), @@ -85,7 +110,8 @@ pub fn map_file(f: &AbrFile, group: &str) -> AbrImport { if !m.unknown.is_empty() { out.warnings.push(format!("settings without a PhotoCraft equivalent were ignored: {}", m.unknown.into_iter().collect::>().join(", "))); } - out + ctl.progress(1.0); + Ok(out) } // ------------------------------------------------------------------ descriptor helpers diff --git a/crates/io/src/lib.rs b/crates/io/src/lib.rs index 576d8a32e0..be7d50bc53 100644 --- a/crates/io/src/lib.rs +++ b/crates/io/src/lib.rs @@ -49,7 +49,7 @@ use photocraft_psd::{PsdError, PsdFile}; pub use adjust_map::ADJUSTMENT_KEYS; pub use flat::document_to_image; pub use psd_export::{PsdExportOptions, document_to_psd, document_to_psd_with}; -pub use psd_import::psd_to_document; +pub use psd_import::{psd_to_document, psd_to_document_with}; /// Errors from import/export. #[derive(Debug, thiserror::Error)] @@ -72,6 +72,9 @@ pub enum IoError { /// Camera raw decode failure. #[error("{0}")] Raw(#[from] photocraft_raw::RawError), + /// A background import was cancelled ([`import_with`]). + #[error("cancelled")] + Cancelled, } /// Result of [`import`]. @@ -109,13 +112,29 @@ pub fn is_psd(bytes: &[u8]) -> bool { /// Imports a file. PSD/PSB and camera raws are detected by magic; everything /// else is decoded with `photocraft-codecs`. pub fn import(name: &str, bytes: &[u8]) -> Result { + import_with(name, bytes, &photocraft_raster::Interrupt::NONE) +} + +/// [`import`] for a background open: checks `ctl` between stages (and per layer for PSD/PSB) and +/// reports progress. A cancelled import fails with [`IoError::Cancelled`]. +pub fn import_with(name: &str, bytes: &[u8], ctl: &photocraft_raster::Interrupt) -> Result { + ctl.check().map_err(|_| IoError::Cancelled)?; + let r = import_stages(name, bytes, ctl)?; + ctl.check().map_err(|_| IoError::Cancelled)?; + ctl.progress(1.0); + Ok(r) +} + +fn import_stages(name: &str, bytes: &[u8], ctl: &photocraft_raster::Interrupt) -> Result { // A declared native extension must reach its loader so malformed bundles retain format errors. if has_extension(name, photocraft_format::EXTENSION) || photocraft_format::is_pcraft(bytes) { return Ok(ImportResult { document: photocraft_format::load_from_bytes(bytes)?, warnings: Vec::new() }); } if is_psd(bytes) { let file = PsdFile::from_bytes(bytes)?; - let (mut document, warnings) = psd_to_document(&file); + ctl.check().map_err(|_| IoError::Cancelled)?; + ctl.progress(0.05); + let (mut document, warnings) = psd_import::psd_to_document_with(&file, ctl).ok_or(IoError::Cancelled)?; document.name = name.to_string(); return Ok(ImportResult { document, warnings }); } diff --git a/crates/io/src/psd_import.rs b/crates/io/src/psd_import.rs index 031234457e..fc5290a398 100644 --- a/crates/io/src/psd_import.rs +++ b/crates/io/src/psd_import.rs @@ -55,6 +55,11 @@ pub(crate) struct Ctx<'a> { pub txt2: Option, /// Smart-filter caches from the global `FEid`/`FXid` blocks (filter masks, by placed id). pub filter_effects: Vec, + /// Cancellation and progress for background opens (checked per layer record). + pub ctl: photocraft_raster::Interrupt<'a>, + /// Layer records decoded so far, out of `total` (progress). + pub done: usize, + pub total: usize, } fn doc_mode(m: PsdMode) -> Option { @@ -366,10 +371,19 @@ impl Ctx<'_> { fn build(&mut self, nodes: &[LayerNode]) -> Vec { let layers = self.file.layers(); + let ctl = self.ctl; nodes .iter() + // A cancelled open stops decoding; the caller discards the partial document. + .take_while(|_| !ctl.cancelled()) .map(|n| match n { - LayerNode::Layer { index } => self.layer_from_record(&layers[*index]), + LayerNode::Layer { index } => { + let l = self.layer_from_record(&layers[*index]); + self.done += 1; + // Layers are 10–95 % of an open (the merged image and channels the rest). + self.ctl.progress(0.1 + 0.85 * self.done as f32 / self.total.max(1) as f32); + l + } LayerNode::Group { index, children, .. } => { let rec = &layers[*index]; let children = self.build(children); @@ -442,6 +456,14 @@ fn parse_guides(data: &[u8], doc: &mut Document) { /// Converts a parsed PSD into a document. Never fails: problems become warnings. pub fn psd_to_document(file: &PsdFile) -> (Document, Vec) { + // Never cancelled, so always `Some`; the fallback is unreachable. + psd_to_document_with(file, &photocraft_raster::Interrupt::NONE) + .unwrap_or_else(|| (Document::new("Untitled", Size::new(1, 1), ColorMode::Rgb, SampleType::U8), Vec::new())) +} + +/// [`psd_to_document`] for a background open: checks `ctl` per layer record and reports +/// progress. `None` when cancelled. +pub fn psd_to_document_with(file: &PsdFile, ctl: &photocraft_raster::Interrupt) -> Option<(Document, Vec)> { let h = &file.header; let mut warnings = Vec::new(); let mode = doc_mode(h.color_mode).unwrap_or_else(|| { @@ -520,18 +542,32 @@ pub fn psd_to_document(file: &PsdFile) -> (Document, Vec) { dpi: doc.resolution_dpi, txt2: file.global_blocks.iter().find(|b| &b.key == b"Txt2").and_then(|b| photocraft_text::psd::parse_txt2(&b.data)), filter_effects: Vec::new(), + ctl: *ctl, + done: 0, + total: file.layers().len(), }; for b in file.global_blocks.iter().filter(|b| matches!(&b.key, b"FEid" | b"FXid")) { + // Smart-filter caches can be large: a cancelled open stops between blocks. + if ctl.cancelled() { + return None; + } match photocraft_psd::filter_effects::FilterEffects::parse(&b.data) { Ok(fx) => cx.filter_effects.extend(fx.items), Err(e) => cx.warn(format!("smart filter masks ({}) could not be read: {e}", b.key_str())), } } + if ctl.cancelled() { + return None; + } let (w, hh) = (h.width as usize, h.height as usize); let n = w * hh; let canvas = Rect::new(0, 0, h.width as i32, h.height as i32); let merged = file.decode_merged(); + if ctl.cancelled() { + return None; + } + ctl.progress(0.1); if let Err(e) = &merged { cx.warn(format!("merged image could not be decoded: {e}")); } @@ -685,5 +721,9 @@ pub fn psd_to_document(file: &PsdFile) -> (Document, Vec) { // Character and paragraph styles from the type layers' engine data. crate::text_styles_map::import(&mut doc); - (doc, cx.warnings) + if ctl.cancelled() { + return None; + } + ctl.progress(1.0); + Some((doc, cx.warnings)) } diff --git a/crates/io/tests/cancel.rs b/crates/io/tests/cancel.rs new file mode 100644 index 0000000000..efd2b665b8 --- /dev/null +++ b/crates/io/tests/cancel.rs @@ -0,0 +1,62 @@ +//! Background opens (#210): `import_with` stops when cancelled and reports progress. + +use std::sync::atomic::{AtomicU32, Ordering}; + +use photocraft_color::{ColorMode, SampleType}; +use photocraft_doc::{Document, Layer, LayerContent, Size}; +use photocraft_geom::Rect; +use photocraft_raster::{Interrupt, Surface}; + +fn layered_psd() -> Vec { + let mut d = Document::new("t", Size::new(32, 24), ColorMode::Rgb, SampleType::U8); + for i in 0..5 { + let mut s = Surface::new(d.pixel_format()); + s.fill_rect(Rect::new(i, i, 20 + i, 16 + i), &[0.2 * i as f32, 0.5, 0.7, 1.0]); + d.layers.push(Layer::new(format!("L{i}"), LayerContent::Raster(s))); + } + photocraft_io::export(&d, "t.psd", &Default::default()).unwrap().bytes +} + +#[test] +fn cancelled_import_fails_with_cancelled() { + let bytes = layered_psd(); + let yes = || true; + let r = photocraft_io::import_with("t.psd", &bytes, &Interrupt::cancel_only(&yes)); + assert!(matches!(r, Err(photocraft_io::IoError::Cancelled)), "{:?}", r.map(|_| ())); +} + +#[test] +fn import_reports_monotonic_progress_per_layer() { + let bytes = layered_psd(); + let last = AtomicU32::new(0); + let calls = AtomicU32::new(0); + let no = || false; + let progress = |f: f32| { + assert!(f >= f32::from_bits(last.load(Ordering::Relaxed)), "progress went backwards"); + last.store(f.to_bits(), Ordering::Relaxed); + calls.fetch_add(1, Ordering::Relaxed); + }; + let r = photocraft_io::import_with("t.psd", &bytes, &Interrupt::new(&no, &progress)).unwrap(); + assert_eq!(r.document.layers.len(), 5); + assert_eq!(f32::from_bits(last.load(Ordering::Relaxed)), 1.0); + assert!(calls.load(Ordering::Relaxed) >= 6, "one report per layer plus stages"); + // Same document as the plain import. + let plain = photocraft_io::import("t.psd", &bytes).unwrap(); + assert_eq!(plain.document.layers.len(), r.document.layers.len()); +} + +#[test] +fn cancelled_brush_import_fails() { + use photocraft_psd::abr::{LegacyBrush, LegacyTip, write_v12}; + let brush = |name: &str| LegacyBrush { + name: name.into(), + spacing: 10, + anti_alias: true, + tip: LegacyTip::Computed { diameter: 9, hardness: 100, angle: 0, roundness: 100 }, + }; + let bytes = write_v12(2, &[brush("a"), brush("b")], true).unwrap(); + assert_eq!(photocraft_io::abr_map::read_abr(&bytes, "g").unwrap().presets.len(), 2); + let yes = || true; + let r = photocraft_io::abr_map::read_abr_with(&bytes, "g", &Interrupt::cancel_only(&yes)); + assert_eq!(r.err().as_deref(), Some("cancelled")); +} diff --git a/crates/raster/src/interrupt.rs b/crates/raster/src/interrupt.rs new file mode 100644 index 0000000000..2a5521331b --- /dev/null +++ b/crates/raster/src/interrupt.rs @@ -0,0 +1,108 @@ +//! Cooperative cancellation and progress for long operations (background jobs, #210). +//! +//! Heavy algorithms take an [`Interrupt`] and check [`Interrupt::cancelled`] once per tile, row +//! band or iteration, returning early (typically `None`) when it is set. They may also report +//! [`Interrupt::progress`] in `0.0..=1.0`. [`Interrupt::NONE`] never cancels and ignores progress, +//! so synchronous callers pass it and get the old behaviour. +//! +//! The callbacks are `Sync`, so rayon workers can check them while filtering tiles in parallel. + +/// Cancellation check plus progress sink for one long operation. Cheap to copy. +#[derive(Clone, Copy)] +pub struct Interrupt<'a> { + cancel: &'a (dyn Fn() -> bool + Sync), + progress: &'a (dyn Fn(f32) + Sync), +} + +fn never() -> bool { + false +} + +fn ignore(_: f32) {} + +impl std::fmt::Debug for Interrupt<'_> { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Interrupt").field("cancelled", &self.cancelled()).finish() + } +} + +impl Default for Interrupt<'_> { + fn default() -> Self { + Interrupt::NONE + } +} + +impl<'a> Interrupt<'a> { + /// Never cancelled; progress is ignored. + pub const NONE: Interrupt<'static> = Interrupt { cancel: &never, progress: &ignore }; + + /// An interrupt from a cancellation check and a progress sink (`0.0..=1.0`). + pub fn new(cancel: &'a (dyn Fn() -> bool + Sync), progress: &'a (dyn Fn(f32) + Sync)) -> Self { + Self { cancel, progress } + } + + /// Only a cancellation check (progress is ignored). + pub fn cancel_only(cancel: &'a (dyn Fn() -> bool + Sync)) -> Self { + Self { cancel, progress: &ignore } + } + + /// Has the operation been cancelled? Algorithms stop as soon as they see `true`. + pub fn cancelled(&self) -> bool { + (self.cancel)() + } + + /// Report progress (`0.0..=1.0`; values outside are clamped, NaN is ignored). + pub fn progress(&self, fraction: f32) { + if fraction.is_finite() { + (self.progress)(fraction.clamp(0.0, 1.0)); + } + } + + /// `Err(Cancelled)` once cancelled, for `?` in loops. + pub fn check(&self) -> Result<(), Cancelled> { + if self.cancelled() { Err(Cancelled) } else { Ok(()) } + } +} + +/// Returned by an operation that stopped because its [`Interrupt`] was cancelled. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct Cancelled; + +impl std::fmt::Display for Cancelled { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("cancelled") + } +} + +impl std::error::Error for Cancelled {} + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; + + use super::*; + + #[test] + fn none_never_cancels() { + assert!(!Interrupt::NONE.cancelled()); + Interrupt::NONE.progress(0.5); + assert_eq!(Interrupt::NONE.check(), Ok(())); + } + + #[test] + fn reports_cancel_and_clamped_progress() { + let flag = AtomicBool::new(false); + let last = AtomicU32::new(0); + let cancel = || flag.load(Ordering::Relaxed); + let progress = |f: f32| last.store(f.to_bits(), Ordering::Relaxed); + let i = Interrupt::new(&cancel, &progress); + i.progress(2.0); + assert_eq!(f32::from_bits(last.load(Ordering::Relaxed)), 1.0); + i.progress(f32::NAN); + assert_eq!(f32::from_bits(last.load(Ordering::Relaxed)), 1.0); + assert!(!i.cancelled()); + flag.store(true, Ordering::Relaxed); + assert!(i.cancelled()); + assert_eq!(i.check(), Err(Cancelled)); + } +} diff --git a/crates/raster/src/lib.rs b/crates/raster/src/lib.rs index a80662e20c..43c2d0c870 100644 --- a/crates/raster/src/lib.rs +++ b/crates/raster/src/lib.rs @@ -7,6 +7,9 @@ #![forbid(unsafe_code)] #![deny(clippy::unwrap_used, clippy::expect_used, clippy::panic, clippy::unimplemented, clippy::todo, clippy::unreachable)] +mod interrupt; +pub use interrupt::{Cancelled, Interrupt}; + use std::collections::BTreeMap; use std::sync::Arc;