From 4a89e68f9fb69ff444cc08e4064305d4760e5d2b Mon Sep 17 00:00:00 2001 From: gmrmad1 <86235830+gmrmad1@users.noreply.github.com> Date: Fri, 6 Nov 2026 03:21:11 +0530 Subject: [PATCH] Content-Aware Fill: spill the range and stream the output (#2570 phases 2-4) (#3079) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Content-Aware Fill: bit-pack holes, spill frames to disk, stream outputs (#2570 phases 2-4) * Content-Aware Fill phases 2-4: bit-pack holes, spill frames, stream outputs (clippy clean) * Address CodeRabbit review: spill safety, streaming correctness, perf, memory guard (#2570) * Address the remaining CodeRabbit review findings on Content-Aware Fill (#2570) A non-streaming fill returned empty frames, progress could jump backwards between the sink and the fill, the memory guard ignored the membrane frames still resident for the flow pass (and wasm's in-RAM copies), and the GPU fill stages ran outside an error scope, so an OOM retired the shared context instead of falling back to the CPU for that frame. 🤖 Generated with Codebuff Co-Authored-By: Codebuff * Content-Aware Fill: sample spilled frames without decoding them, check spill errors (#2570) Tracing sampled spilled frames by decoding whole frames under the shared cache lock, so parallel traces across neighbouring frames evicted each other and mostly paid a full-frame read while every other thread waited. Sample only the 2x2 bilinear support with positioned reads on an open handle instead, and report a failed spill read in the no-hole branch rather than streaming black frames as a successful fill. 🤖 Generated with Codebuff Co-Authored-By: Codebuff * Content-Aware Fill: spill membrane/propagated frames, unquantized input, surface scratch errors (#2570) Flow and propagation now write their intermediate frames to scratch spills beside the output instead of holding the whole range as float RGBA, and the input spill uses 32-bit floats whatever the project depth so the fill sees the same values as the old in-memory path. Spill reads/writes and scratch creation report through a shared error slot that run_plan surfaces instead of masking disk failures as a cancellation. Sampling uses positioned reads; the guard models the packed masks plus per-thread scratch. * Content-Aware Fill: report a membrane-spill write failure instead of cancelling (#2570) * Content-Aware Fill: include the flow fields in the memory guard (#2570) * Content-Aware Fill: scale the memory guard with workers and check scratch space (#2570) The guard's scratch term was a fixed 8 frames, which under-counted a wide machine with Lighting Correction by several times. It now models the sequential synthesis stage plus a few frames per worker (from available parallelism) and takes Lighting Correction into account. Native fills write three f32 copies of the range to scratch beside the output, so a range that cannot fit on that volume is now refused up front with the free space in the message, before any frame is rendered. Tests cover the worker/lighting budget, the scratch formula, a shared error slot between the input and secondary spills, an uncreatable spill folder, and the engine surfacing a spill failure as an error instead of a cancellation. * Content-Aware Fill: resolve a relative output folder for the scratch check (#2570) A bare relative output name has an empty parent, so the walk-up found no existing ancestor and skipped the free-space check. Resolve it against the process directory first, and cover that in the test. * Do not reformat the GPU fill tests in the spill branch (#2570) The file was already not rustfmt-clean upstream, so the formatter only added noise to a file this PR no longer needs to touch: phase 1's call-site changes are already on the base branch. * Content-Aware Fill: measure peak RSS and correct the per-worker scratch model (#2570) The ignored measure_peak_rss_of_a_fill test fills a spilled 640x360 range and reports VmHWM. On this 18-worker machine 90 frames peak at 340-470 MB across runs (35 MB with RAYON_NUM_THREADS=1) and 600 frames at 582 MB, so the scratch slope is ~7 frames a worker, not the 2 the first guard revision assumed. Scaling the summary down showed a clean linear slope at 8 and 4 workers, and Lighting Correction stayed below the flow stage's peak. Budget 8 frames a worker (9 with Lighting Correction), just above the measured top of the range, and adjust the guard test to the real numbers. At 600 frames the measured peak is 582 MB against ~11.1 GB for the old five-copies model, a 19x reduction. * Contributors: credit Madhav Agarwal in the About window Adds the self-chosen real and display name for @gmrmad1. The generated contributors.json picks it up once the PR is merged. * Address CodeRabbit: size the guard from the rayon pool; migrate GPU tests (#2570) fill_workers used available_parallelism, but the parallel fill stages run on the rayon pool, which RAYON_NUM_THREADS can size larger. Use rayon::current_num_threads so the scratch budget follows the threads that actually run; wasm keeps rayon's single-thread fallback. The GPU inpaint tests still used the pre-spill FillInput API. The earlier 'do not reformat' revert dropped their API migration along with the formatting, leaving cargo test -p effectcraft-gpu uncompilable. Restore the Frames::Memory / HoleMask / sink migration and verify the 15 inpaint tests. * Fix gate: flatten the wasm prop lookup and keep the budget wrapper test-only (#2570) * Address CodeRabbit review: spill error check, global-pool preflight, docs (#2570) * Include RAM-backed spill in the preflight and guard streaming reads (#2570) * Resolve relative spill paths, prefer stacked mounts, guard the Edge Blend sink (#2570) * Make the fill spill anonymous and test the streaming claims (#2570) The spill was a named file in the output folder that only Drop removed, so an out-of-memory kill or a force-quit left a multi-gigabyte effectcraft-fill-* file in the user's project folder, and a failed set_len after keep() leaked it too. The spill is now an anonymous temporary file (tempfile_in), which the system reclaims when its last handle closes; every read and write is positioned on that one handle (pwrite/seek_write join the existing pread/seek_read), and the folder is kept only so the propagated frames spill beside the input. Also: - Edge Blend checks for a failed read before handing a frame to the sink, so a black fallback frame never streams. - The propagation workers stop as soon as the spill records an error. - HoleMask::get/set use get()/get_mut(), so a mask whose public words are shorter than len cannot panic. New tests cover the PR's headline claims: a spilled range fills bit-identically to the same frames in RAM (returned and streamed), the sink sees frames strictly in order, a truncated spill fails with a recorded error and streams nothing, a failure mid-stream stops before the next frame, propagation stops after a failure, the spill has no name in its folder, and HoleMask round-trips at word boundaries. Co-Authored-By: Claude Opus 5.5 * Content-Aware Fill: clean up a stopped fill and count two scratch copies (#2570) - A fill that fails or is cancelled after frames started streaming deletes the PNGs it wrote, and the output folder when the fill made it and it is left empty (OutputFiles, dropped after the spill so Windows can remove the folder). - After a successful fill, run_plan checks that every frame has a file, and apply_plan/add_fill_layer return an error on an empty range or file list (a job worker's reply) instead of indexing [0]. - scratch_need charges two f32 copies (32 B/px): flow_pairs drops the membrane spill before the propagated one is created, so the third copy never coexists. - The spill-creation error carries a stable "spill: cannot make the scratch file" marker; the panel test asserted on the OS wording, which on Windows is "Cannot create a file when that file already exists". Co-Authored-By: Claude Opus 5.5 * Content-Aware Fill: a stopped fill leaves files it didn't create (#2570) A fill into a given outputDir that already holds an earlier sequence overwrites its frames as it goes (as before). The cleanup of a failed or stopped fill deleted every path it wrote, including those earlier frames, so a layer still using that sequence lost them. It now deletes only the files the fill created (CodeRabbit on #3079). Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Codebuff Co-authored-by: FredApps <153173854+FredApps@users.noreply.github.com> Co-authored-by: Claude Opus 5.5 --- Cargo.lock | 1 + crates/engine/src/cmd/content_fill.rs | 383 +++++++- crates/engine/src/system/sysinfo.rs | 107 +++ crates/engine/src/tests/panels.rs | 52 ++ crates/gpu/src/tests/inpaint.rs | 34 +- crates/raster/Cargo.toml | 1 + crates/raster/src/inpaint.rs | 1233 +++++++++++++++++++++++-- 7 files changed, 1671 insertions(+), 140 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 6ebde6757..7d58b050d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1855,6 +1855,7 @@ dependencies = [ "effectcraft-color", "effectcraft-geom", "rayon", + "tempfile", ] [[package]] diff --git a/crates/engine/src/cmd/content_fill.rs b/crates/engine/src/cmd/content_fill.rs index 04779db8b..720e780b9 100644 --- a/crates/engine/src/cmd/content_fill.rs +++ b/crates/engine/src/cmd/content_fill.rs @@ -13,7 +13,9 @@ use std::sync::Arc; use effectcraft_color::Label; use effectcraft_project::{Footage, FootageKind, ItemId, ItemKind, LayerId, LayerSource, build}; use effectcraft_raster::Image; -use effectcraft_raster::inpaint::{FillInput, FillMethod, FillOpts, dilate}; +#[cfg(not(target_arch = "wasm32"))] +use effectcraft_raster::inpaint::SpillHandle; +use effectcraft_raster::inpaint::{FillInput, FillMethod, FillOpts, Frames, HoleMask, dilate}; use effectcraft_render::{Accelerator, EvalCtx, ExprHost, FootageSource, LayerCache, Renderer}; use effectcraft_time::Tick; use serde::{Deserialize, Serialize}; @@ -170,11 +172,76 @@ pub struct FillPlan { pub dir: String, } -/// The fill holds every frame as float RGBA (16 bytes a pixel), its hole mask (1) and the filled -/// copy (16) until it finishes (#1789): when even that is more than `available`, the error to -/// report up front, instead of running out of memory partway through. +/// Transient full frames one parallel stage holds per worker: the membrane reads a frame and +/// returns a copy, the flow stage holds the neighbouring pair plus the luma pyramids, and +/// Lighting Correction adds the mismatch ring (`diff`, its mask) on top. +/// +/// Measured with `measure_peak_rss_of_a_fill` (90 frames, 640×360, Surface): peak RSS grows by +/// ~7 frames a rayon worker at the top of its run-to-run spread, and Lighting Correction stays +/// below the flow stage's peak, so 8 (9 with lighting) is a conservative constant (#2570). +fn per_worker_frames(lighting: bool) -> u64 { + if lighting { 9 } else { 8 } +} + +/// The sequential synthesis stage: the stored frame, the PatchMatch pyramid and its +/// coarse-to-fine copies, the warped previous frame, and the sink's prepared copy. +const SYNTH_FRAMES: u64 = 6; + +/// Workers a range of `frames` can keep busy (rayon never runs more tasks than frames). The +/// scratch estimate must follow the pool that actually runs the parallel stages: a synchronous +/// fill stays in the caller's pool, while an asynchronous one runs on `spawn_task`'s own OS +/// thread and so uses the global pool, whatever pool the command was called from (#2570). +fn fill_workers(frames: usize, wait: bool) -> u64 { + let pool = if wait { rayon::current_num_threads() as u64 } else { global_pool_threads() }; + pool.min(frames.max(1) as u64) +} + +/// The global Rayon pool's thread count, read from a thread outside every pool: +/// `rayon::current_num_threads()` answers for the caller's pool, not the one the background +/// task uses. Falls back to the current pool when the probe thread cannot start. +fn global_pool_threads() -> u64 { + let (tx, rx) = std::sync::mpsc::channel(); + match std::thread::Builder::new().name("fill-pool-probe".into()).spawn(move || { + let _ = tx.send(rayon::current_num_threads()); + }) { + Ok(handle) => { + let n = rx.recv().map(|n| n as u64).unwrap_or_else(|_| rayon::current_num_threads() as u64); + let _ = handle.join(); + n + } + Err(_) => rayon::current_num_threads() as u64, + } +} + +/// Native builds spill the input, the membrane-filled neighbours and the propagated copies, and +/// pack the holes to one bit per pixel, so only the two packed masks and the two flow fields +/// (~1/2 byte a pixel a frame) plus the workers' scratch stay resident (#2570). wasm has no +/// spill and keeps the input and the working copies in RAM (~32 bytes a pixel a frame). When +/// even that peak is more than `available`, report it up front instead of running out of memory +/// partway through (#1789, #2570 review). +#[cfg(test)] fn memory_shortfall(frames: usize, w: u32, h: u32, available: u64) -> Option { - let need = (frames as u64).saturating_mul(u64::from(w)).saturating_mul(u64::from(h)).saturating_mul(33); + memory_shortfall_with(frames, w, h, available, fill_workers(frames, true), false, 0) +} + +/// `workers` and `lighting` are parameters so the budget is testable: the scratch term used to +/// be a fixed 8 frames, which under-counted a wide machine by several times (#2570 review). +/// `resident` adds storage that would otherwise be spill: when the spill volume is RAM-backed, +/// its files use the same memory as the working set (#2570 review). +fn memory_shortfall_with(frames: usize, w: u32, h: u32, available: u64, workers: u64, lighting: bool, resident: u64) -> Option { + let px = u64::from(w).saturating_mul(u64::from(h)); + let per_frame = (frames as u64).saturating_mul(px); + // `holes` and the leftover propagation masks, packed to a bit a pixel each. + let masks = per_frame / 4; + // The forward and backward flow fields, ~0.125 byte a pixel each. + let flows = per_frame / 4; + // wasm keeps the input and one working copy in RAM at every stage; a RAM-backed spill + // volume counts there too. + let resident = resident.saturating_add(if cfg!(target_arch = "wasm32") { per_frame.saturating_mul(32) } else { 0 }); + // The synthesis stage is sequential; every parallel stage holds a few frames a worker. + let scratch_frames = SYNTH_FRAMES.saturating_add(per_worker_frames(lighting).saturating_mul(workers)); + let scratch = px.saturating_mul(16).saturating_mul(scratch_frames); + let need = masks.saturating_add(flows).saturating_add(resident).saturating_add(scratch); (need > available).then(|| { let gb = |b: u64| b as f64 / 1e9; format!( @@ -185,6 +252,40 @@ fn memory_shortfall(frames: usize, w: u32, h: u32, available: u64) -> Option u64 { + let bytes = (frames as u64).saturating_mul(u64::from(w)).saturating_mul(u64::from(h)).saturating_mul(32); + bytes.saturating_add(bytes / 10) +} + +/// Refuse up front when the output volume cannot hold the fill's scratch. The folder may not +/// exist yet (the spill creates it), so query the nearest existing ancestor. `None` when the +/// volume cannot be read; the write itself still reports a real error then (#2570 review). +#[cfg(not(target_arch = "wasm32"))] +fn scratch_shortfall(frames: usize, w: u32, h: u32, dir: &std::path::Path) -> Option { + let need = scratch_need(frames, w, h); + // Resolve a relative output folder against the process directory; `parent()` of a bare + // name is empty, which would otherwise skip the check. + let base = if dir.is_absolute() { dir.to_path_buf() } else { std::env::current_dir().ok()?.join(dir) }; + let mut probe = base.as_path(); + while !probe.exists() { + probe = probe.parent()?; + } + let (_, free) = crate::sysinfo::disk_space(probe)?; + (need > free).then(|| { + let gb = |b: u64| b as f64 / 1e9; + format!( + "Content-Aware Fill of {frames} frames at {w}×{h} needs up to {:.1} GB of scratch space beside the output, and {:.1} GB is free there: fill a shorter range (set the work area and choose Work Area) or choose an output folder on a larger disk", + gb(need), + gb(free) + ) + }) +} + /// The status line shown while the frames are filled. Edge Blend calls only the membrane, which /// the device backend leaves to the CPU reference, so it is not a device fill even when a `fill` /// backend is present; Object and Surface run the flow and PatchMatch on the device. @@ -192,6 +293,43 @@ fn filling_status(on_device: bool, method: FillMethod) -> &'static str { if on_device && method != FillMethod::EdgeBlend { "Filling (GPU)" } else { "Filling" } } +/// The PNGs a fill has written so far. Unless the fill finishes ([`OutputFiles::keep`]), dropping +/// it deletes the ones it created, and the output folder too when the fill made it and nothing +/// else is in it, so a failed or stopped fill leaves no partial sequence behind (#2570 review). A +/// file that was already there (an `outputDir` holding an earlier sequence) is never deleted. +struct OutputFiles { + dir: std::path::PathBuf, + /// The folder was missing before the fill (the spill creates it). + made_dir: bool, + paths: Vec, + /// Paths among `paths` that existed before the fill wrote them: not the fill's to delete. + spare: Vec, + kept: bool, +} + +impl OutputFiles { + /// The fill finished: keep its files. + fn keep(mut self) -> Vec { + self.kept = true; + std::mem::take(&mut self.paths) + } +} + +impl Drop for OutputFiles { + fn drop(&mut self) { + if self.kept { + return; + } + for path in self.paths.iter().filter(|p| !self.spare.contains(p)) { + let _ = std::fs::remove_file(path); + } + // `remove_dir` leaves a folder that still holds anything else. + if self.made_dir { + let _ = std::fs::remove_dir(&self.dir); + } + } +} + /// Run a fill: the layer's frames (rendered in passes with a job worker's GPU), the fill, the /// PNG sequence written into `plan.dir` (`Services::store_file`). Returns the files. /// @@ -224,17 +362,34 @@ pub(crate) async fn run_plan( let ctx0 = EvalCtx { expr, ..EvalCtx::new(project, cid, comp, Tick::ZERO) }; let times = &plan.times; let n = times.len() as u64; - let mut frames = Vec::with_capacity(times.len()); let mut holes = Vec::with_capacity(times.len()); + #[cfg(target_arch = "wasm32")] + let mut frames = Vec::with_capacity(times.len()); + // Declared before the spill, so on an early return the spill is closed first and the folder + // it was made in can be removed. + let dir = std::path::Path::new(&plan.dir); + let mut written = OutputFiles { dir: dir.to_path_buf(), made_dir: !dir.exists(), paths: Vec::with_capacity(times.len()), spare: Vec::new(), kept: false }; + // Spill 32-bit floats whatever the project depth: the old in-memory path kept f32 all the + // way through, so quantising here would change the fill (#2570, "no quality loss"). Scratch + // lives beside the output when one is set; the default temp folder may be RAM-backed, which + // the memory preflight accounts for when it chooses the spill folder. + #[cfg(not(target_arch = "wasm32"))] + let spill = SpillHandle::create_in(Some(dir), w as usize, h as usize, times.len(), 32) + .map_err(|e| format!("spill: cannot make the scratch file in {}: {e}", plan.dir))?; + #[cfg(not(target_arch = "wasm32"))] + let spill_probe = spill.clone(); for (k, t) in times.iter().enumerate() { let (img, hole) = crate::offload::in_passes(accel, |a| masked_frame(&r.with_accel(a), &ctx0.at(*t), layer, w, h)).await; - holes.push(dilate(&hole, w as usize, h as usize, plan.expand)); + holes.push(HoleMask::from_vec(&dilate(&hole, w as usize, h as usize, plan.expand))); + #[cfg(target_arch = "wasm32")] frames.push(img); + #[cfg(not(target_arch = "wasm32"))] + spill.write_frame(k, &img)?; if !progress(k as u64 + 1, n * 3) { return Err(crate::render_queue::CANCELLED.into()); } } - if !holes.iter().any(|h| h.iter().any(|b| *b)) { + if !holes.iter().any(|h| h.any()) { return Err("the layer has no transparent area to fill: draw a mask (Subtract) around what to remove".into()); } let mut references = vec![]; @@ -249,31 +404,94 @@ pub(crate) async fn run_plan( } let opts = FillOpts { method: FillMethod::from_name(&plan.method).unwrap_or_default(), lighting: plan.lighting, ..Default::default() }; message(filling_status(fill.is_some(), opts.method)); - let input = FillInput { frames, holes, references }; + #[cfg(target_arch = "wasm32")] + let input_frames = Frames::Memory(frames); + #[cfg(not(target_arch = "wasm32"))] + let input_frames = Frames::Spilled(spill); + let input = FillInput { frames: input_frames, holes, references }; let cpu = effectcraft_raster::inpaint::CpuFill; let backend: &dyn effectcraft_raster::inpaint::FillBackend = fill.unwrap_or(&cpu); - let out = effectcraft_raster::inpaint::fill_sequence_with(input, &opts, backend, &mut |d, t| progress(n + (d as u64 * n / t.max(1) as u64), n * 3)) - .ok_or(crate::render_queue::CANCELLED)?; - message("Writing the fill layer"); - let dir = &plan.dir; - let mut paths = vec![]; - for (k, img) in out.iter().enumerate() { - let path = std::path::Path::new(dir).join(format!("fill_{k:05}.png")).to_string_lossy().to_string(); - services.store_file(&path, &png(img)?).map_err(|e| format!("{path}: {e}"))?; - paths.push(path); - if !progress(2 * n + k as u64 + 1, n * 3) { - return Err(crate::render_queue::CANCELLED.into()); + let progress = std::rc::Rc::new(std::cell::RefCell::new(progress)); + // The fill and the sink both advance the bar: a streamed frame reports ahead of the fill's own + // progress, so both report a shared high-water mark and the bar never moves backwards. + let high = std::cell::Cell::new(0u64); + let report = |done: u64| -> bool { + let done = done.max(high.get()); + high.set(done); + (progress.borrow_mut())(done, n * 3) + }; + let paths: Vec = (0..n as usize).map(|k| dir.join(format!("fill_{k:05}.png")).to_string_lossy().to_string()).collect(); + let mut sink_error: Option = None; + let mut sink = |idx: usize, img: &Image| -> std::result::Result<(), String> { + let r = (|| -> std::result::Result<(), String> { + // A failed spilled read returns a black fallback frame; never publish it (#2570 + // review). + #[cfg(not(target_arch = "wasm32"))] + if spill_probe.has_error() { + return Err(spill_probe.take_error().unwrap_or_else(|| "spill: read failed".to_string())); + } + let existed = services.exists(&paths[idx]); + services.store_file(&paths[idx], &png(img)?).map_err(|e| format!("{}: {e}", paths[idx]))?; + written.paths.push(paths[idx].clone()); + if existed { + written.spare.push(paths[idx].clone()); + } + if !report(2 * n + idx as u64 + 1) { + return Err(crate::render_queue::CANCELLED.to_string()); + } + Ok(()) + })(); + if let Err(e) = &r + && sink_error.is_none() + { + sink_error = Some(e.clone()); + } + r + }; + let produced = + effectcraft_raster::inpaint::fill_sequence_with(input, &opts, backend, &mut |d, t| report(n + (d as u64 * n / t.max(1) as u64)), Some(&mut sink)); + #[cfg(not(target_arch = "wasm32"))] + if let Some(e) = spill_probe.take_error() { + return Err(e); + } + if produced.is_none() { + if let Some(e) = sink_error.take() { + return Err(e); + } + return Err(crate::render_queue::CANCELLED.into()); + } + let produced = produced.unwrap_or_default(); + // Non-streaming fallback (tests): write any retained images. + if !produced.is_empty() { + for (k, img) in produced.iter().enumerate() { + let path = paths[k].clone(); + services.store_file(&path, &png(img)?).map_err(|e| format!("{path}: {e}"))?; + if written.paths.len() <= k { + written.paths.push(path); + } + if !report(2 * n + k as u64 + 1) { + return Err(crate::render_queue::CANCELLED.to_string()); + } } } - Ok(paths) + // Every frame must have a file: the fill layer is built on the first one, and a short + // sequence would play the wrong frames. + if written.paths.len() != times.len() { + return Err(format!("Content-Aware Fill wrote {} of {} frames", written.paths.len(), times.len())); + } + Ok(written.keep()) } /// Add the fill layer of a finished plan (one undo step). +/// The plan comes back from a job worker, so an empty range is an error, not a crash. pub(crate) fn apply_plan(s: &mut Session, plan: &FillPlan, paths: Vec) -> Result { let importer = s.importer.clone(); - add_fill_layer(s, plan.comp, plan.layer, &plan.name, paths, (plan.width, plan.height), plan.times[0], plan.times.len(), importer) + let start = plan.times.first().copied().ok_or_else(|| bad("contentFill.generate", "the fill has no frames"))?; + add_fill_layer(s, plan.comp, plan.layer, &plan.name, paths, (plan.width, plan.height), start, plan.times.len(), importer) } +/// `contentFill.generate`: validate the layer and range, refuse a fill that cannot fit in +/// memory or scratch space, then run it (in the background unless `wait` is set). fn generate(s: &mut Session, p: &Value) -> Result { let c = "contentFill.generate"; let (cid, lid) = layer_p(s, p, c)?; @@ -307,17 +525,35 @@ fn generate(s: &mut Session, p: &Value) -> Result { } let (fa, fb) = (fr.frame_at(fr.snap_nearest(a)), fr.frame_at(b - Tick(1))); let times: Vec = (fa..=fb).map(|f| fr.tick_of(f)).collect(); - if let Some(e) = crate::sysinfo::memory().and_then(|m| memory_shortfall(times.len(), w, h, m.available)) { - return Err(bad(c, e)); - } let opts = FillOpts { method: FillMethod::from_name(&st.method).unwrap_or_default(), lighting: lighting(&st.lighting_correction).unwrap_or(0.0), ..Default::default() }; + let dir = str_p(p, "outputDir").map(str::to_string).unwrap_or_else(|| default_dir(s, &layer.name)); + // The spill lives beside the output. On a RAM-backed volume (`tmpfs`/`ramfs`, where the + // default temp dir may live) its files use the same memory as the working set, so the + // memory preflight counts them; disk-backed volumes get the space check below instead. + #[cfg(not(target_arch = "wasm32"))] + let ram_spill = if crate::sysinfo::is_memory_backed(std::path::Path::new(&dir)) { scratch_need(times.len(), w, h) } else { 0 }; + #[cfg(target_arch = "wasm32")] + let ram_spill = 0; + let wait = b_p(p, "wait").unwrap_or(false); + // `spawn_task` runs a fill inline on wasm, and `runs_inline` selects the pool it will use. + let runs_inline = wait || cfg!(target_arch = "wasm32"); + if let Some(e) = crate::sysinfo::memory() + .and_then(|m| memory_shortfall_with(times.len(), w, h, m.available, fill_workers(times.len(), runs_inline), opts.lighting > 0.0, ram_spill)) + { + return Err(bad(c, e)); + } let expand = st.alpha_expansion.round().max(0.0) as usize; let reference = st.reference_layer.filter(|r| comp.layer(*r).is_some()).map(|r| (r, Tick::from_seconds_f64(st.reference_time.unwrap_or(0.0)))); - let dir = str_p(p, "outputDir").map(str::to_string).unwrap_or_else(|| default_dir(s, &layer.name)); + // Native fills keep up to two f32 copies of the range in scratch beside the output; a volume + // that cannot hold them is refused before any frame is rendered (#2570 review). + #[cfg(not(target_arch = "wasm32"))] + if let Some(e) = scratch_shortfall(times.len(), w, h, std::path::Path::new(&dir)) { + return Err(bad(c, e)); + } let plan = FillPlan { comp: cid, layer: lid, @@ -331,7 +567,6 @@ fn generate(s: &mut Session, p: &Value) -> Result { reference, dir, }; - let wait = b_p(p, "wait").unwrap_or(false); // The browser: a job worker (its own GPU), the page keeps running. if !wait && s.offloads() { s.offload_analysis(crate::offload::WorkerJob::ContentFill { plan: Box::new(plan) }).map_err(EngineError::Other)?; @@ -377,12 +612,13 @@ fn add_fill_layer( count: usize, importer: Option>, ) -> Result { + let first = paths.first().cloned().ok_or_else(|| bad("contentFill.generate", "the fill wrote no frames"))?; let comp = s.project.comp(cid).ok_or(EngineError::NoComp)?.clone(); let src = comp.layer(lid).ok_or_else(|| bad("contentFill.generate", "the source layer is gone"))?.clone(); let fr = comp.frame_rate; let dur = fr.tick_of(count as i64); - let mut footage = importer.and_then(|i| i.probe(&paths[0]).ok()).filter(|f| f.has_video).unwrap_or_else(|| Footage { - path: paths[0].clone(), + let mut footage = importer.and_then(|i| i.probe(&first).ok()).filter(|f| f.has_video).unwrap_or_else(|| Footage { + path: first.clone(), kind: FootageKind::Sequence, width: size.0, height: size.1, @@ -468,7 +704,11 @@ pub fn specs() -> Vec { #[cfg(test)] mod tests { - use super::{filling_status, memory_shortfall}; + #[cfg(not(target_arch = "wasm32"))] + use super::scratch_need; + #[cfg(not(target_arch = "wasm32"))] + use super::scratch_shortfall; + use super::{filling_status, memory_shortfall, memory_shortfall_with}; /// Edge Blend fills with the membrane, which the device backend leaves to the CPU reference: /// it must not claim a GPU fill just because a backend is available. @@ -481,13 +721,88 @@ mod tests { assert_eq!(filling_status(false, FillMethod::Object), "Filling"); } - /// #1789: 10 s of 1080p30 (300 frames) needs at least 20 GB, which 4 GB free can't hold. + /// #1789 / #2570: a range that cannot fit is refused up front instead of OOM-ing. The + /// scratch term scales with the workers and Lighting Correction, not a fixed frame count. #[test] fn a_fill_that_cannot_fit_in_memory_is_refused_up_front() { - let e = memory_shortfall(300, 1920, 1080, 4_000_000_000).expect("refused"); + let e = memory_shortfall_with(300, 1920, 1080, 1_000_000, 8, false, 0).expect("refused"); assert!(e.contains("300 frames") && e.contains("Work Area"), "{e}"); - assert_eq!(memory_shortfall(30, 1920, 1080, 4_000_000_000), None, "a second of it fits"); + // Four workers, no lighting: the packed masks and flows plus ~38 scratch frames fit. + assert_eq!(memory_shortfall_with(30, 1920, 1080, 4_000_000_000, 4, false, 0), None); + assert_eq!(memory_shortfall_with(300, 1920, 1080, 2_000_000_000, 4, false, 0), None); + // A wide machine with Lighting Correction holds nine frames a worker: 300 frames no + // longer fit in 5 GB, which the old fixed 8-frame scratch term missed (#2570 review). + assert!(memory_shortfall_with(300, 1920, 1080, 1_000_000_000, 16, true, 0).is_some()); + assert!(memory_shortfall_with(300, 1920, 1080, 5_000_000_000, 16, true, 0).is_some()); + assert_eq!(memory_shortfall_with(300, 1920, 1080, 6_000_000_000, 16, true, 0), None); + // The two packed masks still grow with the range: a very long 1080p range cannot fit. + assert!(memory_shortfall_with(1000, 1920, 1080, 500_000_000, 8, false, 0).is_some()); + // A RAM-backed spill volume counts its bytes as resident, so the same range that fit on + // a disk-backed volume is refused (#2570 review). + assert_eq!(memory_shortfall_with(30, 1920, 1080, 2_500_000_000, 4, false, 0), None); + assert!(memory_shortfall_with(30, 1920, 1080, 2_500_000_000, 4, false, 2_000_000_000).is_some()); // Hostile sizes saturate instead of overflowing. - assert!(memory_shortfall(usize::MAX, u32::MAX, u32::MAX, u64::MAX - 1).is_some()); + assert!(memory_shortfall(usize::MAX, u32::MAX, u32::MAX, 0).is_some()); + } + + /// The native spill holds at most two f32 copies of the range beside the output at once (the + /// membrane spill is gone before the propagated one is made), plus the PNGs. + #[cfg(not(target_arch = "wasm32"))] + #[test] + fn scratch_space_is_checked_before_a_fill() { + let copies = 300u64 * 1920 * 1080 * 32; + assert_eq!(scratch_need(300, 1920, 1080), copies + copies / 10); + let tmp = std::env::temp_dir(); + // A range that cannot possibly fit on the volume is refused up front. + assert!(scratch_shortfall(usize::MAX, u32::MAX, u32::MAX, &tmp).is_some()); + // A relative folder that does not exist is resolved against the process directory. + assert!(scratch_shortfall(usize::MAX, u32::MAX, u32::MAX, std::path::Path::new("effectcraft-missing-fill-folder")).is_some()); + // A tiny fill fits. + assert_eq!(scratch_shortfall(1, 16, 16, &tmp), None); + } + + /// A fill that fails or is stopped after some frames were written deletes them, and the + /// output folder it made; a finished fill keeps its files (#2570 review). + #[cfg(not(target_arch = "wasm32"))] + #[test] + fn a_stopped_fill_removes_its_partial_output() { + use super::OutputFiles; + let root = tempfile::tempdir().unwrap(); + let write = |dir: &std::path::Path, made_dir: bool| { + std::fs::create_dir_all(dir).unwrap(); + let paths: Vec = (0..2).map(|k| dir.join(format!("fill_{k:05}.png")).to_string_lossy().to_string()).collect(); + for p in &paths { + std::fs::write(p, b"png").unwrap(); + } + OutputFiles { dir: dir.to_path_buf(), made_dir, paths, spare: Vec::new(), kept: false } + }; + // The fill made the folder: the files and the folder go. + let made = root.path().join("made"); + drop(write(&made, true)); + assert!(!made.exists(), "the partial sequence and its folder are removed"); + // The folder was already there: only the fill's own files go. + let existing = root.path().join("existing"); + drop(write(&existing, false)); + assert!(existing.exists()); + assert_eq!(std::fs::read_dir(&existing).unwrap().count(), 0); + // Something else in a folder the fill made keeps the folder. + let shared = root.path().join("shared"); + let files = write(&shared, true); + std::fs::write(shared.join("notes.txt"), b"keep").unwrap(); + drop(files); + assert_eq!(std::fs::read_dir(&shared).unwrap().count(), 1); + // A file that was there before the fill wrote over it (an earlier sequence in a given + // `outputDir`) is not the fill's to delete. + let earlier = root.path().join("earlier"); + let mut files = write(&earlier, false); + files.spare.push(files.paths[0].clone()); + drop(files); + assert!(earlier.join("fill_00000.png").exists(), "an earlier fill's frame stays"); + assert!(!earlier.join("fill_00001.png").exists(), "the fill's own frame goes"); + // A finished fill keeps its files. + let done = root.path().join("done"); + let kept = write(&done, true).keep(); + assert_eq!(kept.len(), 2); + assert!(kept.iter().all(|p| std::path::Path::new(p).exists())); } } diff --git a/crates/engine/src/system/sysinfo.rs b/crates/engine/src/system/sysinfo.rs index bfb471e92..80b51b462 100644 --- a/crates/engine/src/system/sysinfo.rs +++ b/crates/engine/src/system/sysinfo.rs @@ -207,6 +207,75 @@ pub fn disk_space(path: &std::path::Path) -> Option<(u64, u64)> { } } +/// Whether `path` is on a RAM-backed filesystem (`tmpfs`/`ramfs`), where bytes written count +/// against memory as well as space. Linux reads `/proc/self/mountinfo`; other systems report +/// `false` (their temporary folders are disk-backed by default). `path` may not exist yet: its +/// nearest existing ancestor decides, as when the fill creates its scratch folder. +pub fn is_memory_backed(path: &std::path::Path) -> bool { + #[cfg(target_os = "linux")] + { + let Some(canonical) = canonical_ancestor(path) else { return false }; + return std::fs::read_to_string("/proc/self/mountinfo").is_ok_and(|mountinfo| mountinfo_is_memory_backed(&mountinfo, &canonical)); + } + #[allow(unreachable_code)] + { + let _ = path; + false + } +} + +/// The canonical path of `path`'s nearest existing ancestor (or itself). Relative paths are +/// resolved against the current directory first, so a new relative output folder still names +/// the volume it will be created on. +#[cfg(target_os = "linux")] +fn canonical_ancestor(path: &std::path::Path) -> Option { + let absolute = if path.is_absolute() { path.to_path_buf() } else { std::env::current_dir().ok()?.join(path) }; + let mut p = absolute.as_path(); + loop { + if let Ok(canonical) = std::fs::canonicalize(p) { + return Some(canonical); + } + p = p.parent()?; + } +} + +/// Whether a `/proc/self/mountinfo` listing mounts `path` on `tmpfs` or `ramfs`; the deepest +/// matching mount point wins, and a later entry at the same mount point is the visible one. +#[cfg(target_os = "linux")] +pub(crate) fn mountinfo_is_memory_backed(mountinfo: &str, path: &std::path::Path) -> bool { + let mut best: Option<(usize, &str)> = None; + for line in mountinfo.lines() { + let Some(mount) = line.split_whitespace().nth(4) else { continue }; + let Some(fstype) = line.split(" - ").nth(1).and_then(|rest| rest.split_whitespace().next()) else { continue }; + let mount = unescape_mount(mount); + if path.starts_with(&mount) && best.is_none_or(|(longest, _)| mount.len() >= longest) { + best = Some((mount.len(), fstype)); + } + } + matches!(best, Some((_, "tmpfs" | "ramfs"))) +} + +/// Undo mountinfo's octal escapes (`\040` space, `\011` tab, `\012` newline, `\134` backslash). +#[cfg(target_os = "linux")] +fn unescape_mount(field: &str) -> String { + let mut out = String::with_capacity(field.len()); + let mut chars = field.chars().peekable(); + while let Some(c) = chars.next() { + if c == '\\' + && chars.clone().take(3).count() == 3 + && let Ok(code) = u8::from_str_radix(&chars.clone().take(3).collect::(), 8) + { + for _ in 0..3 { + chars.next(); + } + out.push(code as char); + } else { + out.push(c); + } + } + out +} + /// CPU model name. pub fn cpu_name() -> Option { #[cfg(target_os = "linux")] @@ -339,4 +408,42 @@ mod tests { assert_eq!(m.available, 4100 * 16384); assert!(m.is_low()); } + + /// The default temp folder may be a `tmpfs`; a nested disk mount under `/tmp` or a nested + /// `tmpfs` must be resolved to the deepest mount point. + #[cfg(target_os = "linux")] + #[test] + fn mountinfo_finds_memory_backed_storage() { + let mountinfo = "\ +25 1 0:22 / / rw,relatime - ext4 /dev/nvme0n1p2 rw +26 25 0:23 / /tmp rw,nosuid - tmpfs tmpfs rw +27 25 0:24 / /tmp/disk rw - ext4 /dev/sda1 rw +28 25 0:25 / /mnt/ram rw - ramfs none rw +29 25 0:26 / /media/My\\040Disk rw - ext4 /dev/sdb1 rw +"; + assert!(mountinfo_is_memory_backed(mountinfo, std::path::Path::new("/tmp/effectcraft-fill"))); + assert!(!mountinfo_is_memory_backed(mountinfo, std::path::Path::new("/tmp/disk/fill"))); + assert!(mountinfo_is_memory_backed(mountinfo, std::path::Path::new("/mnt/ram/a"))); + assert!(!mountinfo_is_memory_backed(mountinfo, std::path::Path::new("/home/fill"))); + assert!(!mountinfo_is_memory_backed(mountinfo, std::path::Path::new("/"))); + // A path that merely shares a prefix is not on that mount. + assert!(!mountinfo_is_memory_backed(mountinfo, std::path::Path::new("/tmpdisk/fill"))); + // Escaped spaces in a mount point decode before matching. + assert!(!mountinfo_is_memory_backed(mountinfo, std::path::Path::new("/media/My Disk/fill"))); + assert_eq!(unescape_mount("/media/My\\040Disk"), "/media/My Disk"); + // The later entry at the same mount point is the visible mount: a disk mount stacked on + // a tmpfs is disk-backed, and the reverse order is memory-backed (#2570 review). + let stacked_disk = "\ +30 25 0:30 / /tmp/stack rw - tmpfs tmpfs rw +31 25 0:31 / /tmp/stack rw - ext4 /dev/sdb2 rw +"; + let stacked_ram = "\ +30 25 0:30 / /tmp/stack rw - ext4 /dev/sdb2 rw +31 25 0:31 / /tmp/stack rw - tmpfs tmpfs rw +"; + assert!(!mountinfo_is_memory_backed(stacked_disk, std::path::Path::new("/tmp/stack/fill"))); + assert!(mountinfo_is_memory_backed(stacked_ram, std::path::Path::new("/tmp/stack/fill"))); + // A new relative folder resolves against the current directory before walking up. + assert!(canonical_ancestor(std::path::Path::new("no-such-relative-folder/fill")).is_some()); + } } diff --git a/crates/engine/src/tests/panels.rs b/crates/engine/src/tests/panels.rs index 622088745..547b7f100 100644 --- a/crates/engine/src/tests/panels.rs +++ b/crates/engine/src/tests/panels.rs @@ -321,6 +321,58 @@ fn content_aware_fill_removes_an_object() { assert!(s2.execute("contentFill.generate", json!({"wait": true, "outputDir": tmp("caf2").to_string_lossy()})).is_err()); } +/// A scratch folder that cannot be created surfaces as an error, not a cancelled fill or a +/// black frame (#2570 review). +#[test] +fn content_aware_fill_reports_a_spill_failure() { + let (mut s, _, _, _) = setup(48, 32, 0.2, |f| { + let mut img = Image::new(48, 32); + for y in 0..32 { + for x in 0..48 { + img.set(x, y, bg(x as f32 + f as f32, y as f32)); + } + } + img + }); + s.execute("layer.addMask", json!({"rect": [20.0, 10.0, 8.0, 8.0], "mode": "subtract"})).unwrap(); + let dir = tmp("cafspill"); + std::fs::create_dir_all(&dir).unwrap(); + let file = dir.join("not-a-folder"); + std::fs::write(&file, b"x").unwrap(); + // The fill wants to create its spill here; the parent is a file. + let out = file.join("fills"); + let e = s.execute("contentFill.generate", json!({"method": "object", "outputDir": out.to_string_lossy(), "wait": true})).unwrap_err(); + // The OS wording differs ("Not a directory" on Unix, "Cannot create a file when that file + // already exists" on Windows), so look for the fill's own marker. + let msg = e.to_string(); + assert!(msg.contains("spill: cannot make the scratch file"), "not a spill error: {msg}"); + std::fs::remove_dir_all(&dir).unwrap(); +} + +/// A fill plan that comes back without files or frames (a job worker that wrote none) is an +/// error, not a crash on the first file (#2570 review). +#[test] +fn content_aware_fill_without_files_is_an_error() { + let (mut s, cid, lid, _) = setup(48, 32, 0.2, |_| Image::filled(48, 32, [0.5, 0.5, 0.5, 1.0])); + let plan = crate::commands::content_fill::FillPlan { + comp: cid, + layer: lid, + name: "Clip".into(), + width: 48, + height: 32, + times: vec![Tick::ZERO], + method: "object".into(), + lighting: 0.0, + expand: 0, + reference: None, + dir: "Clip_Fill_1".into(), + }; + assert!(crate::commands::content_fill::apply_plan(&mut s, &plan, vec![]).is_err()); + let empty = crate::commands::content_fill::FillPlan { times: vec![], ..plan }; + assert!(crate::commands::content_fill::apply_plan(&mut s, &empty, vec!["Clip_Fill_1/fill_00000.png".into()]).is_err()); + assert_eq!(s.project.comp(cid).unwrap().layers.len(), 1, "no fill layer was added"); +} + #[test] fn content_aware_fill_runs_in_the_background() { let (mut s, _, _, _) = setup(48, 32, 0.2, |f| { diff --git a/crates/gpu/src/tests/inpaint.rs b/crates/gpu/src/tests/inpaint.rs index c6901bb45..dacda05e2 100644 --- a/crates/gpu/src/tests/inpaint.rs +++ b/crates/gpu/src/tests/inpaint.rs @@ -147,16 +147,20 @@ fn object_fill_recovers_background_and_is_stable() { holes.push(hole); gts.push(gt); } - let input = FillInput { frames, holes, references: vec![] }; - let out = fill_sequence_with(input.clone(), &FillOpts::default(), g, &mut |_, _| true).expect("fill"); + let input = FillInput { + frames: effectcraft_raster::inpaint::Frames::Memory(frames), + holes: holes.into_iter().map(|h| effectcraft_raster::inpaint::HoleMask::from_vec(&h)).collect(), + references: vec![], + }; + let out = fill_sequence_with(input.clone(), &FillOpts::default(), g, &mut |_, _| true, None).expect("fill"); for t in 0..out.len() { - let db = psnr(&out[t], >s[t], Some(&input.holes[t])); + let db = psnr(&out[t], >s[t], Some(&input.holes[t].to_vec())); assert!(db > 26.0, "frame {t} pan {pan}: {db:.1} dB"); assert_eq!(out[t].data[0][3], 1.0); } if pan == 0.0 { for t in 1..out.len() { - let both: Vec = input.holes[t].iter().zip(&input.holes[t - 1]).map(|(a, b)| *a || *b).collect(); + let both: Vec = input.holes[t].to_vec().iter().zip(input.holes[t - 1].to_vec().iter()).map(|(a, b)| *a || *b).collect(); let db = psnr(&out[t], &out[t - 1], Some(&both)); assert!(db > 30.0, "frames {}–{t} differ: {db:.1} dB", t - 1); } @@ -172,13 +176,17 @@ fn edge_blend_takes_the_cpu_membrane() { let (w, h) = (80u32, 60u32); let gt = background(w, h, 0.0); let hole = rect_hole(w, h, 30, 20, 14); - let input = FillInput { frames: vec![cut(>, &hole); 4], holes: vec![hole.clone(); 4], references: vec![] }; + let input = FillInput { + frames: effectcraft_raster::inpaint::Frames::Memory(vec![cut(>, &hole); 4]), + holes: vec![effectcraft_raster::inpaint::HoleMask::from_vec(&hole); 4], + references: vec![], + }; let opts = FillOpts { method: effectcraft_raster::inpaint::FillMethod::EdgeBlend, ..Default::default() }; - let edge = fill_sequence_with(input.clone(), &opts, g, &mut |_, _| true).expect("fill"); + let edge = fill_sequence_with(input.clone(), &opts, g, &mut |_, _| true, None).expect("fill"); for t in 1..4 { assert!(psnr(&edge[t], &edge[0], Some(&hole)) > 40.0, "frame {t}"); } - assert!(psnr(&edge[0], &membrane_fill(&input.frames[0], &hole), Some(&hole)) > 60.0); + assert!(psnr(&edge[0], &membrane_fill(&input.frames.get(0), &hole), Some(&hole)) > 60.0); } /// The dilation the level plans are built from (kept honest: the flags' target bit is `near`). @@ -299,11 +307,15 @@ fn sequence_fill_with_the_device_flow_matches_the_cpu() { frames.push(cut(>, &hole)); holes.push(hole); } - let input = FillInput { frames, holes, references: vec![] }; - let dev = fill_sequence_with(input.clone(), &FillOpts::default(), g, &mut |_, _| true).expect("gpu fill"); - let ref_cpu = fill_sequence_with(input.clone(), &FillOpts::default(), &cpu, &mut |_, _| true).expect("cpu fill"); + let input = FillInput { + frames: effectcraft_raster::inpaint::Frames::Memory(frames), + holes: holes.into_iter().map(|h| effectcraft_raster::inpaint::HoleMask::from_vec(&h)).collect(), + references: vec![], + }; + let dev = fill_sequence_with(input.clone(), &FillOpts::default(), g, &mut |_, _| true, None).expect("gpu fill"); + let ref_cpu = fill_sequence_with(input.clone(), &FillOpts::default(), &cpu, &mut |_, _| true, None).expect("cpu fill"); for t in 0..dev.len() { - let db = psnr(&dev[t], &ref_cpu[t], Some(&input.holes[t])); + let db = psnr(&dev[t], &ref_cpu[t], Some(&input.holes[t].to_vec())); assert!(db > 40.0, "frame {t} pan {pan}: device vs CPU {db:.1} dB"); } } diff --git a/crates/raster/Cargo.toml b/crates/raster/Cargo.toml index e27d55317..7cd4fbe5a 100644 --- a/crates/raster/Cargo.toml +++ b/crates/raster/Cargo.toml @@ -13,3 +13,4 @@ workspace = true effectcraft-geom = { workspace = true } effectcraft-color = { workspace = true } rayon = { workspace = true } +tempfile = { workspace = true } diff --git a/crates/raster/src/inpaint.rs b/crates/raster/src/inpaint.rs index 5625b242c..8b7127b4d 100644 --- a/crates/raster/src/inpaint.rs +++ b/crates/raster/src/inpaint.rs @@ -34,6 +34,7 @@ use rayon::prelude::*; use crate::Image; +use crate::Px; use crate::flow::{Flow, block_flow}; /// Fill method. @@ -605,13 +606,469 @@ pub fn complete_flow(mut f: Flow, hole_a: &[bool], hole_b: &[bool], w: usize, h: f } +/// A hole mask packed into 64-bit words (1 bit per pixel) to keep the per-frame cost +/// down across a whole range (#2570). +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct HoleMask { + pub words: Vec, + pub len: usize, +} + +impl HoleMask { + /// Pack one `bool` a pixel into 64-bit words. + pub fn from_vec(hole: &[bool]) -> Self { + let mut words = vec![0u64; hole.len().div_ceil(64)]; + for (i, &h) in hole.iter().enumerate() { + if h { + words[i / 64] |= 1u64 << (i % 64); + } + } + HoleMask { words, len: hole.len() } + } + /// An all-known mask of `n` pixels. + pub fn zeros(n: usize) -> Self { + HoleMask { words: vec![0u64; n.div_ceil(64)], len: n } + } + /// Whether pixel `i` is in the hole. `words` and `len` are public, so a mask whose words are + /// shorter than `len` reads the missing pixels as known instead of panicking. + pub fn get(&self, i: usize) -> bool { + i < self.len && self.words.get(i / 64).is_some_and(|w| (w >> (i % 64)) & 1 == 1) + } + /// Mark pixel `i` as missing (`true`) or known (`false`). A pixel outside the mask is ignored. + pub fn set(&mut self, i: usize, v: bool) { + if i >= self.len { + return; + } + let Some(w) = self.words.get_mut(i / 64) else { + return; + }; + if v { + *w |= 1u64 << (i % 64); + } else { + *w &= !(1u64 << (i % 64)); + } + } + /// Whether any pixel is in the hole. + pub fn any(&self) -> bool { + self.words.iter().any(|&w| w != 0) + } + /// Unpack to one `bool` a pixel. + pub fn to_vec(&self) -> Vec { + (0..self.len).map(|i| self.get(i)).collect() + } +} + +/// Where a fill's input frames live. `Memory` keeps them in RAM (tests, small ranges); +/// `Spilled` stores them on disk and reads them on demand, so a whole range costs ~0 RAM. +#[derive(Clone)] +pub enum Frames { + Memory(Vec), + Spilled(SpillHandle), +} + +impl Frames { + /// The number of frames in the range. + pub fn len(&self) -> usize { + match self { + Frames::Memory(v) => v.len(), + Frames::Spilled(h) => h.n, + } + } + /// Whether the range has no frames. + pub fn is_empty(&self) -> bool { + self.len() == 0 + } + /// Frame `k`: a clone from RAM, or a decode from the spill. + pub fn get(&self, k: usize) -> Image { + match self { + Frames::Memory(v) => v[k].clone(), + Frames::Spilled(h) => h.get(k), + } + } + /// Frame size without decoding a whole image. + pub fn size(&self) -> Option<(usize, usize)> { + match self { + Frames::Memory(v) => v.first().map(|i| (i.width as usize, i.height as usize)), + Frames::Spilled(h) => Some((h.w, h.h)), + } + } + /// Bilinear sample of one pixel. A spilled frame reads only the 2x2 support from disk, so + /// the hot trace path never decodes or clones a whole frame (#2570). + pub fn sample(&self, k: usize, x: f64, y: f64) -> Px { + match self { + Frames::Memory(v) => v[k].sample_bilinear_clamped(x, y), + Frames::Spilled(h) => h.sample(k, x, y), + } + } + /// Directory of the spill file, so a secondary spill can live beside it. + pub fn scratch_dir(&self) -> Option { + match self { + Frames::Memory(_) => None, + Frames::Spilled(h) => h.dir().map(std::path::Path::to_path_buf), + } + } + /// Whether a read or write of a spilled frame failed. + pub fn has_error(&self) -> bool { + match self { + Frames::Memory(_) => false, + Frames::Spilled(h) => h.has_error(), + } + } + /// Take the first spill failure, clearing it. + pub fn take_error(&self) -> Option { + match self { + Frames::Memory(_) => None, + Frames::Spilled(h) => h.take_error(), + } + } + /// The error slot shared by every spill in this fill, so a caller holding a clone of the + /// input handle also sees failures in the propagated-frame spill. + pub fn error_cell(&self) -> Option>>> { + match self { + Frames::Memory(_) => None, + Frames::Spilled(h) => Some(h.error_cell()), + } + } +} + +/// A spill file holding `n` frames of `w x h` pixels at `bpc` bits per channel. +/// Frames are read back on demand. The file is shared (reference-counted), so it lives until the +/// last clone is dropped (#2570 review). +#[derive(Clone)] +pub struct SpillHandle { + inner: std::sync::Arc, + pub w: usize, + pub h: usize, + pub n: usize, + pub bpc: u8, +} + +struct SpillInner { + /// The folder the spill was created in (`None`: the system temporary folder), so the fill's + /// other spills go beside it. + dir: Option, + /// An anonymous temporary file: it has no name in `dir`, and the system reclaims it when the + /// last handle closes, even when the app is killed or runs out of memory partway through a + /// fill, so a multi-gigabyte scratch file is never left in the user's folder (#2570 review). + /// Every read and write is positioned on this one handle: several trace threads read the same + /// file at once, and a shared file offset would need a lock around every read (#2570). + file: std::fs::File, + /// Shared with any secondary spill so one probe reports every scratch failure. + error: std::sync::Arc>>, +} + +impl SpillHandle { + /// The folder the spill lives in (`None`: the system temporary folder). + pub fn dir(&self) -> Option<&std::path::Path> { + self.inner.dir.as_deref() + } + + /// Bytes a pixel in the spill: RGBA `u8`, or four `f32` above 8 bits a channel. + fn bytes_per_px(&self) -> usize { + if self.bpc <= 8 { 4 } else { 16 } + } + + /// Record a scratch failure in the shared slot, keeping the first. + pub fn record_error(&self, message: String) { + if let Ok(mut e) = self.inner.error.lock() + && e.is_none() + { + *e = Some(message); + } + } + + /// The error slot shared with any secondary spill. + pub fn error_cell(&self) -> std::sync::Arc>> { + std::sync::Arc::clone(&self.inner.error) + } + + /// Whether a read or write failed (`true` when the slot is poisoned). + pub fn has_error(&self) -> bool { + self.inner.error.lock().map(|e| e.is_some()).unwrap_or(true) + } + + /// Take the first failure, clearing it. + pub fn take_error(&self) -> Option { + self.inner.error.lock().ok().and_then(|mut e| e.take()) + } + + /// Read one whole frame from the spill, reporting a short or failed read. + fn load_frame(&self, k: usize) -> Image { + let mut img = Image::new(self.w as u32, self.h as u32); + let Some(px) = self.w.checked_mul(self.h) else { + self.record_error("spill: pixel count overflow".into()); + return img; + }; + let bytes_per_px = self.bytes_per_px(); + let Some(off) = k.checked_mul(px).and_then(|v| v.checked_mul(bytes_per_px)) else { + self.record_error(format!("spill: offset overflow for frame {k}")); + return img; + }; + let Some(len) = px.checked_mul(bytes_per_px) else { + self.record_error("spill: frame size overflow".into()); + return img; + }; + let mut buf = vec![0u8; len]; + if let Err(e) = self.read_at(off as u64, &mut buf) { + self.record_error(format!("spill: {e}")); + return img; + } + if self.bpc <= 8 { + for (i, p) in img.data.iter_mut().enumerate() { + let j = i * 4; + p[0] = buf[j] as f32 / 255.0; + p[1] = buf[j + 1] as f32 / 255.0; + p[2] = buf[j + 2] as f32 / 255.0; + p[3] = buf[j + 3] as f32 / 255.0; + } + } else { + for (i, p) in img.data.iter_mut().enumerate() { + let base = i * 16; + for c in 0..4 { + let mut b = [0u8; 4]; + b.copy_from_slice(&buf[base + c * 4..base + c * 4 + 4]); + p[c] = f32::from_bits(u32::from_le_bytes(b)); + } + } + } + img + } + + /// Read a whole frame, decoding it. + pub fn get(&self, k: usize) -> Image { + self.load_frame(k) + } + + /// A positioned read: `pread`/`seek_read` use the shared handle without touching a shared + /// file offset, so tracing threads do not serialise on the file. + fn read_at(&self, off: u64, buf: &mut [u8]) -> std::io::Result<()> { + let file = &self.inner.file; + #[cfg(unix)] + { + use std::os::unix::fs::FileExt; + file.read_exact_at(buf, off) + } + #[cfg(windows)] + { + use std::os::windows::fs::FileExt; + let mut done = 0usize; + while done < buf.len() { + match file.seek_read(&mut buf[done..], off.saturating_add(done as u64)) { + Ok(0) => return Err(std::io::ErrorKind::UnexpectedEof.into()), + Ok(n) => done += n, + Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {} + Err(e) => return Err(e), + } + } + Ok(()) + } + // Elsewhere, seek the shared handle. Spills are only made natively: wasm keeps the + // propagated frames in RAM and has no file system to spill the input to. + #[cfg(not(any(unix, windows)))] + { + use std::io::{Read, Seek, SeekFrom}; + let mut f = file; + f.seek(SeekFrom::Start(off))?; + f.read_exact(buf) + } + } + + /// A positioned write on the same handle as `read_at`, so the propagation workers can store + /// their frames at once and the file never needs a name to be reopened by. + fn write_at(&self, off: u64, buf: &[u8]) -> std::io::Result<()> { + let file = &self.inner.file; + #[cfg(unix)] + { + use std::os::unix::fs::FileExt; + file.write_all_at(buf, off) + } + #[cfg(windows)] + { + use std::os::windows::fs::FileExt; + let mut done = 0usize; + while done < buf.len() { + match file.seek_write(&buf[done..], off.saturating_add(done as u64)) { + Ok(0) => return Err(std::io::ErrorKind::WriteZero.into()), + Ok(n) => done += n, + Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {} + Err(e) => return Err(e), + } + } + Ok(()) + } + #[cfg(not(any(unix, windows)))] + { + use std::io::{Seek, SeekFrom, Write}; + let mut f = file; + f.seek(SeekFrom::Start(off))?; + f.write_all(buf) + } + } + + /// One pixel decoded from `buf` at index `i` (transparent if `buf` is short). + fn px_at(&self, buf: &[u8], i: usize) -> Px { + if self.bpc <= 8 { + let j = i * 4; + [ + buf.get(j).copied().unwrap_or(0) as f32 / 255.0, + buf.get(j + 1).copied().unwrap_or(0) as f32 / 255.0, + buf.get(j + 2).copied().unwrap_or(0) as f32 / 255.0, + buf.get(j + 3).copied().unwrap_or(0) as f32 / 255.0, + ] + } else { + let base = i * 16; + let mut p = [0.0f32; 4]; + for (c, v) in p.iter_mut().enumerate() { + let at = base + c * 4; + let Some(b) = buf.get(at..at + 4).and_then(|s| <[u8; 4]>::try_from(s).ok()) else { + return [0.0; 4]; + }; + *v = f32::from_bits(u32::from_le_bytes(b)); + } + p + } + } + + /// The two horizontally adjacent pixels (`x`, `x + 1`, both clamped) of row `y` (clamped), + /// read straight from the spill file. + fn edge(&self, k: usize, y: i64, x: i64) -> Option<(Px, Px)> { + let (w, h) = (self.w, self.h); + if w == 0 || h == 0 || k >= self.n { + return None; + } + let (iw, ih) = (i64::try_from(w).ok()?, i64::try_from(h).ok()?); + let y = y.clamp(0, ih - 1) as usize; + let xa = x.clamp(0, iw - 1) as usize; + let xb = x.saturating_add(1).clamp(0, iw - 1) as usize; + let bpp = self.bytes_per_px(); + let off = k + .checked_mul(w) + .and_then(|v| v.checked_mul(h)) + .and_then(|v| v.checked_add(y.checked_mul(w)?)) + .and_then(|v| v.checked_add(xa)) + .and_then(|v| v.checked_mul(bpp)); + let Some(off) = off.and_then(|v| u64::try_from(v).ok()) else { + self.record_error("spill: sample offset overflow".into()); + return None; + }; + // One read covers both pixels unless the right edge clamps them together. A stack buffer + // keeps the per-sample allocation out of the trace loop. + let len = (xb - xa + 1) * bpp; + let mut held = [0u8; 32]; + let buf = held.get_mut(..len)?; + if let Err(e) = self.read_at(off, buf) { + self.record_error(format!("spill: {e}")); + return None; + } + let a = self.px_at(buf, 0); + let b = if xb > xa { self.px_at(buf, 1) } else { a }; + Some((a, b)) + } + + /// Bilinear sample of one pixel, reading only its 2x2 support. Matches + /// `Image::sample_bilinear_clamped` exactly, with the same edge clamping. + pub fn sample(&self, k: usize, x: f64, y: f64) -> Px { + let fx = x - 0.5; + let fy = y - 0.5; + let (x0, y0) = (fx.floor(), fy.floor()); + let (tx, ty) = ((fx - x0) as f32, (fy - y0) as f32); + // `as i64` saturates NaN and out-of-range values, as in `sample_bilinear_clamped`. + let (x0, y0) = (x0 as i64, y0 as i64); + let (Some((a, b)), Some((c, d))) = (self.edge(k, y0, x0), self.edge(k, y0.saturating_add(1), x0)) else { + return [0.0; 4]; + }; + let mut o = [0.0; 4]; + for i in 0..4 { + let top = a[i] + (b[i] - a[i]) * tx; + let bot = c[i] + (d[i] - c[i]) * tx; + o[i] = top + (bot - top) * ty; + } + o + } + + /// Write frame `k` into the spill, reporting a disk error. + pub fn write_frame(&self, k: usize, img: &Image) -> std::result::Result<(), String> { + if k >= self.n { + return Err(format!("spill: frame {k} out of range (n = {})", self.n)); + } + if img.width as usize != self.w || img.height as usize != self.h { + return Err(format!("spill: frame {k} is {}x{}, expected {}x{}", img.width, img.height, self.w, self.h)); + } + let bytes_per_px = self.bytes_per_px(); + let px = self.w.checked_mul(self.h).ok_or("overflow")?; + let off = k.checked_mul(px).and_then(|v| v.checked_mul(bytes_per_px)).ok_or("overflow")?; + let buf = if self.bpc <= 8 { + let mut buf = vec![0u8; px * 4]; + for (i, p) in img.data.iter().enumerate() { + buf[i * 4] = (p[0] * 255.0).round().clamp(0.0, 255.0) as u8; + buf[i * 4 + 1] = (p[1] * 255.0).round().clamp(0.0, 255.0) as u8; + buf[i * 4 + 2] = (p[2] * 255.0).round().clamp(0.0, 255.0) as u8; + buf[i * 4 + 3] = (p[3] * 255.0).round().clamp(0.0, 255.0) as u8; + } + buf + } else { + let mut buf = vec![0u8; px * 16]; + for (i, p) in img.data.iter().enumerate() { + let base = i * 16; + for c in 0..4 { + buf[base + c * 4..base + c * 4 + 4].copy_from_slice(&p[c].to_le_bytes()); + } + } + buf + }; + self.write_at(off as u64, &buf).map_err(|e| format!("spill: {e}")) + } + + /// Create a spill file. `dir` lets the caller keep scratch next to the output (so a + /// RAM-backed `/tmp` does not defeat the point of spilling); `None` uses the temp dir. + pub fn create_in(dir: Option<&std::path::Path>, w: usize, h: usize, n: usize, bpc: u8) -> std::io::Result { + Self::create_in_with_error(dir, w, h, n, bpc, None) + } + + /// `error`: share an existing error slot, so a secondary spill reports through the same + /// handle the caller already watches (#2570 review). + pub fn create_in_with_error( + dir: Option<&std::path::Path>, + w: usize, + h: usize, + n: usize, + bpc: u8, + error: Option>>>, + ) -> std::io::Result { + let bytes_per_px: usize = if bpc <= 8 { 4 } else { 16 }; + let total = n + .checked_mul(w) + .and_then(|v| v.checked_mul(h)) + .and_then(|v| v.checked_mul(bytes_per_px)) + .ok_or(std::io::Error::new(std::io::ErrorKind::InvalidInput, "overflow"))?; + // Anonymous, so nothing is left behind if `set_len` fails below or the app never gets + // to clean up. + let file = match dir { + Some(d) => { + std::fs::create_dir_all(d)?; + tempfile::tempfile_in(d)? + } + None => tempfile::tempfile()?, + }; + file.set_len(total as u64)?; + let error = error.unwrap_or_else(|| std::sync::Arc::new(std::sync::Mutex::new(None))); + let inner = std::sync::Arc::new(SpillInner { dir: dir.map(std::path::Path::to_path_buf), file, error }); + Ok(SpillHandle { inner, w, h, n, bpc }) + } + + /// Create a spill in the system temporary directory. + pub fn create(w: usize, h: usize, n: usize, bpc: u8) -> std::io::Result { + Self::create_in(None, w, h, n, bpc) + } +} + /// The input of a sequence fill. #[derive(Clone)] pub struct FillInput { - /// Frames (premultiplied RGBA, all the same size). - pub frames: Vec, - /// Per frame, `true` where the pixel is missing. - pub holes: Vec>, + /// Frames (premultiplied RGBA, all the same size), in RAM or spilled to disk. + pub frames: Frames, + /// Per frame, packed hole mask (`true` where the pixel is missing). + pub holes: Vec, /// Complete images for some frames: (frame index, image). pub references: Vec<(usize, Image)>, } @@ -621,7 +1078,10 @@ struct TraceContext<'a> { fwd: &'a [Flow], bwd: &'a [Flow], known: &'a (dyn Fn(usize, i64, i64) -> bool + Sync), - frames: &'a [&'a Image], + frames: &'a Frames, + ref_images: &'a [Option], + w: usize, + h: usize, max: usize, } @@ -636,7 +1096,7 @@ struct TraceStart { /// Trace a point to the nearest visible frame: (sampled colour, frames travelled). fn trace(context: &TraceContext<'_>, start: TraceStart) -> Option<([f32; 4], usize)> { let n = context.frames.len(); - let (w, h) = (context.frames[0].width as f64, context.frames[0].height as f64); + let (w, h) = (context.w as f64, context.h as f64); let (mut px, mut py) = (start.x, start.y); let mut k = start.frame as i64; for d in 1..=context.max { @@ -663,7 +1123,13 @@ fn trace(context: &TraceContext<'_>, start: TraceStart) -> Option<([f32; 4], usi .iter() .all(|&(xx, yy)| (context.known)(k as usize, xx.clamp(0, w as i64 - 1), yy.clamp(0, h as i64 - 1))); if ok { - return Some((context.frames[k as usize].sample_bilinear_clamped(px, py), d)); + let img = if let Some(r) = &context.ref_images[k as usize] { + r.sample_bilinear_clamped(px, py) + } else { + // Samples a pixel; never clones a whole spilled frame (the trace hot path). + context.frames.sample(k as usize, px, py) + }; + return Some((img, d)); } } None @@ -709,144 +1175,365 @@ pub struct CpuFill; impl FillBackend for CpuFill {} +/// Holds the propagated frames between the parallel propagation pass and the sequential +/// synthesis pass. Native builds spill them beside the input, so a whole range is never +/// resident at once (#2570); wasm keeps them in memory (it has no shared file system). +enum PropStore { + #[cfg(not(target_arch = "wasm32"))] + Spilled(SpillHandle), + #[cfg(target_arch = "wasm32")] + Memory(std::sync::Mutex>>), +} + +impl PropStore { + /// Open the propagated-frame store: spilled natively, in RAM on wasm. + fn new( + dir: Option<&std::path::Path>, + w: usize, + h: usize, + n: usize, + error: Option>>>, + ) -> std::io::Result { + #[cfg(not(target_arch = "wasm32"))] + { + // 32-bit floats: propagating quantized input would lose quality (#2570). + Ok(PropStore::Spilled(SpillHandle::create_in_with_error(dir, w, h, n, 32, error)?)) + } + #[cfg(target_arch = "wasm32")] + { + let _ = (dir, w, h, error); + Ok(PropStore::Memory(std::sync::Mutex::new((0..n).map(|_| None).collect()))) + } + } + + /// Store frame `k`, reporting a spill write failure. + fn put(&self, k: usize, img: Image) -> std::result::Result<(), String> { + match self { + #[cfg(not(target_arch = "wasm32"))] + PropStore::Spilled(h) => h.write_frame(k, &img), + #[cfg(target_arch = "wasm32")] + PropStore::Memory(v) => { + let mut v = v.lock().map_err(|_| "propagation store poisoned".to_string())?; + if let Some(slot) = v.get_mut(k) { + *slot = Some(img); + } + Ok(()) + } + } + } + + /// Read a stored frame without consuming it (a frame can be shared by two flows). + fn get(&self, k: usize) -> Image { + match self { + #[cfg(not(target_arch = "wasm32"))] + PropStore::Spilled(h) => h.get(k), + #[cfg(target_arch = "wasm32")] + PropStore::Memory(v) => v.lock().ok().and_then(|v| v.get(k).cloned()).flatten().unwrap_or_default(), + } + } + + /// Read a propagated frame once, freeing it where the backing store allows it. + fn take(&self, k: usize) -> Image { + match self { + #[cfg(not(target_arch = "wasm32"))] + PropStore::Spilled(h) => h.get(k), + #[cfg(target_arch = "wasm32")] + PropStore::Memory(v) => v.lock().ok().and_then(|mut v| v.get_mut(k).and_then(Option::take)).unwrap_or_default(), + } + } + + /// Record a failure in the spill, so the caller reports it instead of a bare cancellation. + fn record_error(&self, message: String) { + match self { + #[cfg(not(target_arch = "wasm32"))] + PropStore::Spilled(h) => h.record_error(message), + #[cfg(target_arch = "wasm32")] + PropStore::Memory(_) => { + let _ = message; + } + } + } + + /// Whether the spill failed (always `false` on wasm, which keeps frames in RAM). + fn has_error(&self) -> bool { + match self { + #[cfg(not(target_arch = "wasm32"))] + PropStore::Spilled(h) => h.has_error(), + #[cfg(target_arch = "wasm32")] + PropStore::Memory(_) => false, + } + } +} + +/// Publish a scratch failure through the input spill's shared error slot, so the caller reports +/// the disk error instead of a bare cancellation. +fn record_spill_error(cell: &Option>>>, message: String) { + if let Some(cell) = cell + && let Ok(mut e) = cell.lock() + && e.is_none() + { + *e = Some(message); + } +} + +/// Membrane-fill every frame and return the forward/backward flow for each neighbouring pair. +/// The filled frames are written to a scratch spill, so the flow pass never holds the range in +/// RAM (#2570). +#[allow(clippy::too_many_arguments)] +fn flow_pairs( + n: usize, + w: usize, + h: usize, + holes: &[HoleMask], + is_ref: &[bool], + ref_images: &[Option], + frames: &Frames, + prop_dir: Option<&std::path::Path>, + error_cell: Option>>>, + backend: &dyn FillBackend, +) -> Option> { + let store = match PropStore::new(prop_dir, w, h, n, error_cell.clone()) { + Ok(s) => s, + Err(e) => { + record_spill_error(&error_cell, format!("spill: {e}")); + return None; + } + }; + let pre = (0..n).into_par_iter().try_for_each(|t| { + let base = if is_ref[t] { ref_images[t].clone().unwrap_or_default() } else { frames.get(t) }; + let img = backend.membrane(&base, &holes[t].to_vec()); + store.put(t, img) + }); + if let Err(e) = pre { + record_spill_error(&error_cell, e); + return None; + } + if frames.has_error() { + return None; + } + let pairs: Vec<(Flow, Flow)> = (0..n.saturating_sub(1)) + .into_par_iter() + .map(|i| { + let a = store.get(i); + let b = store.get(i + 1); + let ha = holes[i].to_vec(); + let hb = holes[i + 1].to_vec(); + (backend.flow(&a, &b, &ha, &hb), backend.flow(&b, &a, &hb, &ha)) + }) + .collect(); + let store_failed = store.has_error(); + drop(store); + if store_failed || frames.has_error() { None } else { Some(pairs) } +} + /// Fill a range of frames on the CPU. `progress(done, total)` returns `false` to cancel /// (→ `None`). pub fn fill_sequence(input: &FillInput, o: &FillOpts, progress: &mut dyn FnMut(usize, usize) -> bool) -> Option> { - fill_sequence_with(input.clone(), o, &CpuFill, progress) + fill_sequence_with(input.clone(), o, &CpuFill, progress, None) } /// [`fill_sequence`] with a hardware backend for the stages it can take (see [`FillBackend`]). -pub fn fill_sequence_with(mut input: FillInput, o: &FillOpts, backend: &dyn FillBackend, progress: &mut dyn FnMut(usize, usize) -> bool) -> Option> { +pub fn fill_sequence_with( + mut input: FillInput, + o: &FillOpts, + backend: &dyn FillBackend, + progress: &mut dyn FnMut(usize, usize) -> bool, + mut sink: Option<&mut dyn FnMut(usize, &Image) -> Result<(), String>>, +) -> Option> { let n = input.frames.len(); if n == 0 { return Some(vec![]); } - let (w, h) = (input.frames[0].width as usize, input.frames[0].height as usize); + let (w, h) = input.frames.size().unwrap_or((0, 0)); + if w == 0 || h == 0 { + return Some(vec![]); + } // Steps: flows (1) + propagation (1) + per-frame synthesis (n). let total = n + 2; + let streaming = sink.is_some(); if o.method == FillMethod::EdgeBlend { let mut out = Vec::with_capacity(n); - for (i, (f, hl)) in input.frames.iter().zip(&input.holes).enumerate() { - out.push(opaque_fill(backend.membrane(f, hl), hl)); + for i in 0..n { + let f = input.frames.get(i); + // A failed read returns a black frame, which must never reach the sink. + if input.frames.has_error() { + return None; + } + let hl = input.holes[i].to_vec(); + let img = opaque_fill(backend.membrane(&f, &hl), &hl); + if let Some(cb) = sink.as_mut() + && cb(i, &img).is_err() + { + return None; + } + if !streaming { + out.push(img); + } if !progress(i + 1, n) { return None; } } + if input.frames.has_error() { + return None; + } + if streaming { + return Some(Vec::new()); + } return Some(out); } // Reference frames make their frame fully known. - let mut frames: Vec = std::mem::take(&mut input.frames); - let mut holes: Vec> = std::mem::take(&mut input.holes); + let mut holes: Vec = std::mem::take(&mut input.holes); let references = std::mem::take(&mut input.references); let mut is_ref = vec![false; n]; - for (k, img) in &references { - if *k < n && img.width as usize == w && img.height as usize == h { - frames[*k] = img.clone(); - holes[*k] = vec![false; w * h]; - is_ref[*k] = true; + let mut ref_images: Vec> = vec![None; n]; + for (k, img) in references { + if k < n && img.width as usize == w && img.height as usize == h { + ref_images[k] = Some(img); + holes[k] = HoleMask::zeros(w * h); + is_ref[k] = true; } } - // 1. Flows between membrane-filled neighbours, completed inside the holes. - let pre: Vec = frames.par_iter().zip(&holes).map(|(f, hl)| backend.membrane(f, hl)).collect(); - let pairs: Vec<(Flow, Flow)> = (0..n.saturating_sub(1)) - .into_par_iter() - .map(|i| (backend.flow(&pre[i], &pre[i + 1], &holes[i], &holes[i + 1]), backend.flow(&pre[i + 1], &pre[i], &holes[i + 1], &holes[i]))) - .collect(); + let prop_dir = input.frames.scratch_dir(); + let error_cell = input.frames.error_cell(); + // 1. Flows between membrane-filled neighbours, completed inside the holes. The + // membrane-filled frames are spilled too, so the flow pass does not hold the range. + let pairs = flow_pairs(n, w, h, &holes, &is_ref, &ref_images, &input.frames, prop_dir.as_deref(), error_cell.clone(), backend)?; let (fwd, bwd): (Vec, Vec) = pairs.into_iter().unzip(); - drop(pre); if !progress(1, total) { return None; } - // 2. Propagation (and the lighting-correction border ring). - let refs: Vec<&Image> = frames.iter().collect(); - let known = |k: usize, x: i64, y: i64| !holes[k][y as usize * w + x as usize]; - let trace_context = TraceContext { fwd: &fwd, bwd: &bwd, known: &known, frames: &refs, max: n }; + // 2. Propagation (and the lighting-correction border ring). The filled frames are + // spilled as they are produced, so only the leftover masks stay in RAM. + let store = match PropStore::new(prop_dir.as_deref(), w, h, n, error_cell.clone()) { + Ok(s) => s, + Err(e) => { + record_spill_error(&error_cell, format!("spill: {e}")); + return None; + } + }; + let known = |k: usize, x: i64, y: i64| !holes[k].get(y as usize * w + x as usize); + let trace_context = TraceContext { fwd: &fwd, bwd: &bwd, known: &known, frames: &input.frames, ref_images: &ref_images, w, h, max: n }; let ring_r = 2usize; - let mut propagated: Vec<(Image, Vec)> = (0..n) + let propagated_left: Vec = (0..n) .into_par_iter() .map(|t| { - let mut img = frames[t].clone(); - let mut left = holes[t].clone(); - if is_ref[t] || !holes[t].iter().any(|b| *b) { - return (img, left); + // Once a frame failed to spill or read the fill is lost: stop tracing the rest (the + // masks are dropped below). + if store.has_error() || input.frames.has_error() { + return HoleMask::default(); } - let sample = |i: usize| -> Option<[f32; 4]> { - let (x, y) = ((i % w) as f64 + 0.5, (i / w) as f64 + 0.5); - let f = trace(&trace_context, TraceStart { frame: t, x, y, step: 1 }); - let b = trace(&trace_context, TraceStart { frame: t, x, y, step: -1 }); - match (f, b) { - (Some((a, da)), Some((c, dc))) => { - let (wa, wc) = (1.0 / da as f32, 1.0 / dc as f32); - Some(std::array::from_fn(|k| (a[k] * wa + c[k] * wc) / (wa + wc))) + let mut img = if is_ref[t] { ref_images[t].clone().unwrap_or_default() } else { input.frames.get(t) }; + let mut left = holes[t].to_vec(); + if !is_ref[t] && holes[t].any() { + let sample = |i: usize| -> Option<[f32; 4]> { + let (x, y) = ((i % w) as f64 + 0.5, (i / w) as f64 + 0.5); + let f = trace(&trace_context, TraceStart { frame: t, x, y, step: 1 }); + let b = trace(&trace_context, TraceStart { frame: t, x, y, step: -1 }); + match (f, b) { + (Some((a, da)), Some((c, dc))) => { + let (wa, wc) = (1.0 / da as f32, 1.0 / dc as f32); + Some(std::array::from_fn(|k| (a[k] * wa + c[k] * wc) / (wa + wc))) + } + (Some((a, _)), None) | (None, Some((a, _))) => Some(a), + _ => None, } - (Some((a, _)), None) | (None, Some((a, _))) => Some(a), - _ => None, + }; + let filled: Vec<(usize, [f32; 4])> = (0..w * h).filter(|&i| holes[t].get(i)).filter_map(|i| sample(i).map(|v| (i, v))).collect(); + for (i, v) in &filled { + img.data[*i] = *v; + left[*i] = false; } - }; - let filled: Vec<(usize, [f32; 4])> = (0..w * h).filter(|&i| holes[t][i]).filter_map(|i| sample(i).map(|v| (i, v))).collect(); - for (i, v) in &filled { - img.data[*i] = *v; - left[*i] = false; - } - // Lighting correction: membrane of the border mismatch, added to propagated pixels. - if o.lighting > 0.0 && !filled.is_empty() { - let grown = dilate(&holes[t], w, h, ring_r); - let mut diff = vec![[0.0f32; 4]; w * h]; - let mut dk = vec![true; w * h]; - for i in 0..w * h { - if holes[t][i] { - dk[i] = false; - } else if grown[i] - && let Some(v) = sample(i) - { - let a = frames[t].data[i]; - diff[i] = std::array::from_fn(|c| if c < 3 { a[c] - v[c] } else { 0.0 }); + // Lighting correction: membrane of the border mismatch, added to propagated pixels. + // Non-hole pixels are untouched above, so `img` is still the original there. + if o.lighting > 0.0 && !filled.is_empty() { + let grown = dilate(&holes[t].to_vec(), w, h, ring_r); + let mut diff = vec![[0.0f32; 4]; w * h]; + let mut dk = vec![true; w * h]; + for i in 0..w * h { + if holes[t].get(i) { + dk[i] = false; + } else if grown[i] + && let Some(v) = sample(i) + { + let a = img.data[i]; + diff[i] = std::array::from_fn(|c| if c < 3 { a[c] - v[c] } else { 0.0 }); + } } - } - backend.membrane_known(&mut diff, &dk, w, h); - for (i, _) in &filled { - for c in 0..3 { - img.data[*i][c] += o.lighting * diff[*i][c]; + backend.membrane_known(&mut diff, &dk, w, h); + for (i, _) in &filled { + for c in 0..3 { + img.data[*i][c] += o.lighting * diff[*i][c]; + } } } } - (img, left) + if let Err(e) = store.put(t, img) { + store.record_error(e); + } + HoleMask::from_vec(&left) }) .collect(); - drop(refs); - drop(frames); + if input.frames.has_error() || store.has_error() { + return None; + } + drop(input); if !progress(2, total) { return None; } // 3. Synthesis of what no frame shows. let mut out: Vec> = vec![None; n]; - let first = (0..n).find(|&t| propagated[t].1.iter().any(|b| *b)); + let first = (0..n).find(|&t| propagated_left[t].any()); let Some(t0) = first else { - return Some(propagated.into_iter().zip(&holes).map(|((img, _), hl)| opaque_fill(img, hl)).collect()); + if streaming { + for t in 0..n { + let img = opaque_fill(take_checked(&store, t)?, &holes[t].to_vec()); + if !stream_frame(&mut sink, t, Some(&img), &holes) { + return None; + } + } + return Some(Vec::new()); + } + let out: Vec = (0..n).map(|t| opaque_fill(store.take(t), &holes[t].to_vec())).collect(); + if store.has_error() { + return None; + } + return Some(out); }; // Frames before t0 are complete already. - for (t, slot) in out.iter_mut().enumerate().take(t0) { - *slot = Some(std::mem::take(&mut propagated[t].0)); + for t in 0..t0 { + let img = take_checked(&store, t)?; + if streaming { + if !stream_frame(&mut sink, t, Some(&img), &holes) { + return None; + } + } else { + out[t] = Some(img); + } } let mut done = 2; - let (img0, left0) = std::mem::take(&mut propagated[t0]); - out[t0] = Some(backend.patch(&img0, &left0, None, o.patch, o.seed)); - drop((img0, left0)); + let left0 = propagated_left[t0].to_vec(); + out[t0] = Some(backend.patch(&take_checked(&store, t0)?, &left0, None, o.patch, o.seed)); + if !stream_frame(&mut sink, t0, out[t0].as_ref(), &holes) { + return None; + } done += 1; if !progress(done, total) { return None; } for t in t0 + 1..n { - // Taken, not cloned: each propagated frame is freed once its output exists. - let (img, left) = std::mem::take(&mut propagated[t]); - let (img, left) = (&img, &left); - if !left.iter().any(|b| *b) { - out[t] = Some(img.clone()); + let complete = !propagated_left[t].any(); + if complete { + out[t] = Some(take_checked(&store, t)?); } else { + let img = take_checked(&store, t)?; + let left = propagated_left[t].to_vec(); // Warp the previous result here along the backward flow (t → t − 1). // Frames after `t0` are filled in order, so the previous one is set. let Some(prev) = out[t - 1].as_ref() else { - out[t] = Some(backend.patch(img, left, None, o.patch, o.seed)); + out[t] = Some(backend.patch(&img, &left, None, o.patch, o.seed)); + if !stream_frame(&mut sink, t, out[t].as_ref(), &holes) { + return None; + } continue; }; let mut warped = img.clone(); @@ -859,15 +1546,47 @@ pub fn fill_sequence_with(mut input: FillInput, o: &FillOpts, backend: &dyn Fill } out[t] = Some(match o.method { FillMethod::Surface => warped, - _ => backend.patch(img, left, Some(&warped), o.patch, o.seed), + _ => backend.patch(&img, &left, Some(&warped), o.patch, o.seed), }); } + // Only the streaming sink has consumed the frame; a non-streaming result returns it. + if streaming && t >= 2 { + out[t - 2] = None; + } + if !stream_frame(&mut sink, t, out[t].as_ref(), &holes) { + return None; + } done += 1; if !progress(done, total) { return None; } } - Some(out.into_iter().zip(&holes).map(|(img, hl)| opaque_fill(img.unwrap_or_default(), hl)).collect()) + if store.has_error() { + return None; + } + if streaming { + for slot in out.iter_mut() { + *slot = None; + } + return Some(Vec::new()); + } + Some(out.into_iter().enumerate().map(|(i, img)| opaque_fill(img.unwrap_or_default(), &holes[i].to_vec())).collect()) +} + +/// Take frame `k` from the store, or `None` when the spill read failed. A failed read yields a +/// black frame, which must never be emitted (#2570 review). +fn take_checked(store: &PropStore, k: usize) -> Option { + let img = store.take(k); + (!store.has_error()).then_some(img) +} + +/// Hand frame `t` to the streaming sink with its filled pixels made opaque. `true` when there is +/// no sink or it accepted the frame. +fn stream_frame(sink: &mut Option<&mut dyn FnMut(usize, &Image) -> Result<(), String>>, t: usize, img: Option<&Image>, holes: &[HoleMask]) -> bool { + let (Some(cb), Some(img)) = (sink.as_mut(), img) else { + return true; + }; + cb(t, &opaque_fill(img.clone(), &holes[t].to_vec())).is_ok() } /// Filled pixels are opaque. @@ -914,6 +1633,47 @@ mod tests { (0..w * h).map(|i| (x0..x0 + s).contains(&(i % w)) && (y0..y0 + s).contains(&(i / w))).collect() } + /// Packing and unpacking a hole mask round-trips every pixel. + #[test] + fn holemask_roundtrip() { + let v: Vec = (0..1000).map(|i| i % 3 == 0).collect(); + let m = HoleMask::from_vec(&v); + assert_eq!(m.to_vec(), v); + for i in 0..v.len() { + assert_eq!(m.get(i), v[i]); + } + } + + /// Packing round-trips where the last word is full, holds one bit or is one bit short, and + /// `set` builds the same mask; a mask whose public `words` are shorter than `len` reads and + /// writes without panicking (#2570 review). + #[test] + fn holemask_roundtrips_at_word_boundaries() { + for len in [1usize, 63, 64, 65, 127, 128, 129, 191, 192] { + let v: Vec = (0..len).map(|i| i % 3 == 0 || i == len - 1).collect(); + let m = HoleMask::from_vec(&v); + assert_eq!(m.words.len(), len.div_ceil(64), "len {len}"); + assert_eq!(m.to_vec(), v, "len {len}"); + let mut s = HoleMask::zeros(len); + for (i, &h) in v.iter().enumerate() { + s.set(i, h); + } + assert_eq!(s, m, "len {len}: set builds the packed mask"); + for i in 0..len { + s.set(i, false); + } + assert!(!s.any(), "len {len}: every pixel cleared"); + // A pixel past the end is ignored. + s.set(len, true); + assert!(!s.get(len), "len {len}"); + assert!(!s.any(), "len {len}"); + } + let mut short = HoleMask { words: vec![], len: 100 }; + assert!(!short.get(70)); + short.set(70, true); + assert!(!short.any()); + } + #[test] fn membrane_matches_a_linear_ramp() { let (w, h) = (40u32, 30u32); @@ -957,10 +1717,10 @@ mod tests { let gt = background(w, h, t as f32 * pan); let hole = rect_hole(w, h, 10 + 8 * t as u32, 22, 18); frames.push(cut(>, &hole)); - holes.push(hole); + holes.push(HoleMask::from_vec(&hole)); gts.push(gt); } - (FillInput { frames, holes, references: vec![] }, gts) + (FillInput { frames: Frames::Memory(frames), holes, references: vec![] }, gts) } #[test] @@ -969,7 +1729,7 @@ mod tests { let (input, gts) = sequence(8, pan); let out = fill_sequence(&input, &FillOpts::default(), &mut |_, _| true).expect("fill"); for t in 0..out.len() { - let db = psnr(&out[t], >s[t], Some(&input.holes[t])); + let db = psnr(&out[t], >s[t], Some(&input.holes[t].to_vec())); assert!(db > 26.0, "frame {t} pan {pan}: {db:.1} dB"); assert_eq!(out[t].data[0][3], 1.0); } @@ -977,7 +1737,7 @@ mod tests { // Temporal consistency: on a static background the filled pixels agree between // frames. for t in 1..out.len() { - let both: Vec = input.holes[t].iter().zip(&input.holes[t - 1]).map(|(a, b)| *a || *b).collect(); + let both: Vec = input.holes[t].to_vec().iter().zip(input.holes[t - 1].to_vec().iter()).map(|(a, b)| *a || *b).collect(); let db = psnr(&out[t], &out[t - 1], Some(&both)); assert!(db > 30.0, "frames {}–{t} differ: {db:.1} dB", t - 1); } @@ -991,14 +1751,14 @@ mod tests { let (w, h) = (80u32, 60u32); let gt = background(w, h, 0.0); let hole = rect_hole(w, h, 30, 20, 14); - let input = FillInput { frames: vec![cut(>, &hole); 4], holes: vec![hole.clone(); 4], references: vec![] }; + let input = FillInput { frames: Frames::Memory(vec![cut(>, &hole); 4]), holes: vec![HoleMask::from_vec(&hole); 4], references: vec![] }; let surf = fill_sequence(&input, &FillOpts { method: FillMethod::Surface, ..Default::default() }, &mut |_, _| true).expect("fill"); for t in 1..4 { // Surface carries one fill through: identical frames. assert!(psnr(&surf[t], &surf[0], Some(&hole)) > 40.0); } let edge = fill_sequence(&input, &FillOpts { method: FillMethod::EdgeBlend, ..Default::default() }, &mut |_, _| true).expect("fill"); - assert!(psnr(&edge[0], &membrane_fill(&input.frames[0], &hole), Some(&hole)) > 60.0); + assert!(psnr(&edge[0], &membrane_fill(&input.frames.get(0), &hole), Some(&hole)) > 60.0); // A clean reference frame recovers the background exactly. let mut with_ref = FillInput { frames: input.frames.clone(), holes: input.holes.clone(), references: vec![(2, gt.clone())] }; let out = fill_sequence(&with_ref, &FillOpts::default(), &mut |_, _| true).expect("fill"); @@ -1014,9 +1774,13 @@ mod tests { fn lighting_correction_absorbs_a_brightness_change() { // The background brightens over time; propagated pixels come from darker frames. let (mut input, mut gts) = sequence(6, 0.0); + let mut frames = match std::mem::replace(&mut input.frames, Frames::Memory(vec![])) { + Frames::Memory(v) => v, + _ => vec![], + }; for t in 0..6 { let k = 1.0 + 0.06 * t as f32; - for img in [&mut input.frames[t], &mut gts[t]] { + for img in [&mut frames[t], &mut gts[t]] { for p in &mut img.data { for c in 0..3 { p[c] *= k; @@ -1024,9 +1788,288 @@ mod tests { } } } + input.frames = Frames::Memory(frames); let plain = fill_sequence(&input, &FillOpts::default(), &mut |_, _| true).expect("fill"); let lit = fill_sequence(&input, &FillOpts { lighting: 1.0, ..Default::default() }, &mut |_, _| true).expect("fill"); - let score = |o: &[Image]| (0..6).map(|t| psnr(&o[t], >s[t], Some(&input.holes[t]))).sum::(); + let score = |o: &[Image]| (0..6).map(|t| psnr(&o[t], >s[t], Some(&input.holes[t].to_vec()))).sum::(); assert!(score(&lit) > score(&plain), "lit {} vs plain {}", score(&lit), score(&plain)); } + + /// `SpillHandle::sample` reads only the 2x2 support; it must still agree with sampling the + /// decoded frame, including the edges and coordinates outside the image (#2570 review). + #[test] + fn spilled_samples_match_the_decoded_frame() { + let (w, h, n) = (7u32, 5u32, 3usize); + for bpc in [8u8, 32] { + let mut frames = Vec::new(); + for k in 0..n { + let mut img = Image::new(w, h); + for (i, p) in img.data.iter_mut().enumerate() { + *p = [(i % 7) as f32 / 7.0, (k * 3 + i % 5) as f32 / 5.0, 0.25, 1.0]; + } + frames.push(img); + } + let spill = SpillHandle::create(w as usize, h as usize, n, bpc).expect("spill"); + for (k, img) in frames.iter().enumerate() { + spill.write_frame(k, img).expect("write"); + } + let spilled = Frames::Spilled(spill.clone()); + assert!(!spilled.has_error()); + for k in 0..n { + let decoded = spill.get(k); + for &(x, y) in &[(0.0, 0.0), (0.5, 0.5), (3.4, 2.6), (6.5, 4.5), (7.9, 5.9), (-1.0, -1.0), (2.25, 3.75)] { + let a = spilled.sample(k, x, y); + let b = decoded.sample_bilinear_clamped(x, y); + for c in 0..4 { + assert!((a[c] - b[c]).abs() <= 1e-6, "bpc {bpc} frame {k} at ({x}, {y}) channel {c}: {} vs {}", a[c], b[c]); + } + } + } + assert!(!spilled.has_error(), "sampling a valid spill records no error"); + } + } + + /// The input's error slot is shared with the membrane and propagation spills, so one probe + /// reports any scratch failure (#2570 review). + #[test] + fn a_secondary_spill_reports_through_the_input_error_slot() { + let input = SpillHandle::create(4, 4, 2, 32).expect("input spill"); + let prop = SpillHandle::create_in_with_error(None, 4, 4, 2, 32, Some(input.error_cell())).expect("prop spill"); + prop.record_error("spill: No space left on device".into()); + assert!(input.has_error()); + assert_eq!(input.take_error().as_deref(), Some("spill: No space left on device")); + assert!(!prop.has_error(), "take clears the shared slot"); + } + + /// A spill whose scratch folder cannot be created is an error, not a silent fallback. + #[test] + fn a_spill_that_cannot_create_its_folder_is_an_error() { + let dir = tempfile::tempdir().expect("temp"); + let file = dir.path().join("not-a-folder"); + std::fs::write(&file, b"x").expect("write"); + assert!(SpillHandle::create_in(Some(&file), 4, 4, 1, 32).is_err()); + } + + /// The scratch file has no name in its folder, so a fill that is killed partway through + /// leaves nothing behind, and every read and write still reaches the frames (#2570 review). + #[test] + fn a_spill_leaves_no_file_in_its_folder() { + let dir = tempfile::tempdir().expect("temp"); + let spill = SpillHandle::create_in(Some(dir.path()), 4, 3, 2, 32).expect("spill"); + let mut img = Image::new(4, 3); + for (i, p) in img.data.iter_mut().enumerate() { + *p = [i as f32 / 12.0, 0.5, 0.25, 1.0]; + } + spill.write_frame(1, &img).expect("write"); + assert_eq!(spill.get(1).data, img.data); + assert!(!spill.has_error()); + assert_eq!(spill.dir(), Some(dir.path())); + assert_eq!(Frames::Spilled(spill.clone()).scratch_dir().as_deref(), Some(dir.path())); + // Elsewhere the file is unlinked as soon as it is made; Windows keeps a delete-on-close + // file listed until its last handle closes. + #[cfg(not(windows))] + assert_eq!(std::fs::read_dir(dir.path()).expect("list").count(), 0, "the open spill has a name in the folder"); + drop(spill); + assert_eq!(std::fs::read_dir(dir.path()).expect("list").count(), 0, "scratch left in the folder"); + } + + /// The same frames spilled to 32 bpc scratch. + fn spill_frames(frames: &Frames) -> SpillHandle { + let (w, h) = frames.size().expect("size"); + let spill = SpillHandle::create(w, h, frames.len(), 32).expect("spill"); + for k in 0..frames.len() { + spill.write_frame(k, &frames.get(k)).expect("write"); + } + spill + } + + /// Every channel of every pixel, as bits. + fn bits(img: &Image) -> Vec { + img.data.iter().flatten().map(|v| v.to_bits()).collect() + } + + /// A spilled range fills bit-identically to the same frames in RAM, streamed or returned: the + /// spill keeps 32-bit floats and samples the same 2x2 support (#2570, "no quality loss"). + #[test] + fn a_spilled_range_fills_exactly_like_one_in_memory() { + let (input, _) = sequence(6, 1.0); + for method in [FillMethod::Object, FillMethod::Surface, FillMethod::EdgeBlend] { + let opts = FillOpts { method, lighting: 0.5, ..Default::default() }; + let memory = fill_sequence(&input, &opts, &mut |_, _| true).expect("memory fill"); + let spilled = FillInput { frames: Frames::Spilled(spill_frames(&input.frames)), holes: input.holes.clone(), references: vec![] }; + let returned = fill_sequence(&spilled, &opts, &mut |_, _| true).expect("spilled fill"); + let mut streamed = Vec::new(); + let mut sink = |_: usize, img: &Image| -> Result<(), String> { + streamed.push(img.clone()); + Ok(()) + }; + let rest = fill_sequence_with(spilled, &opts, &CpuFill, &mut |_, _| true, Some(&mut sink)); + assert!(rest.is_some_and(|v| v.is_empty()), "{method:?}: the sink took every frame"); + assert_eq!((memory.len(), returned.len(), streamed.len()), (6, 6, 6), "{method:?}"); + for t in 0..6 { + assert!(bits(&returned[t]) == bits(&memory[t]), "{method:?} frame {t}: spilled fill differs from RAM"); + assert!(bits(&streamed[t]) == bits(&memory[t]), "{method:?} frame {t}: streamed fill differs from RAM"); + } + } + } + + /// The sink sees every frame once and strictly in order, whichever stage finishes it: the + /// complete frames before the first synthesised one, the synthesised run, and Edge Blend. + #[test] + fn the_sink_sees_frames_in_order() { + let (input, gts) = sequence(8, 1.0); + let mut frames: Vec = (0..8).map(|k| input.frames.get(k)).collect(); + let mut holes = input.holes.clone(); + // The first two frames are clean, so they stream ahead of the first synthesised frame. + for k in 0..2 { + frames[k] = gts[k].clone(); + holes[k] = HoleMask::zeros(frames[k].data.len()); + } + let input = FillInput { frames: Frames::Spilled(spill_frames(&Frames::Memory(frames))), holes, references: vec![] }; + for method in [FillMethod::Object, FillMethod::Surface, FillMethod::EdgeBlend] { + let mut seen = Vec::new(); + let mut sink = |t: usize, _: &Image| -> Result<(), String> { + seen.push(t); + Ok(()) + }; + let out = fill_sequence_with(input.clone(), &FillOpts { method, ..Default::default() }, &CpuFill, &mut |_, _| true, Some(&mut sink)); + assert!(out.is_some(), "{method:?}"); + assert_eq!(seen, (0..8).collect::>(), "{method:?}"); + } + } + + /// A spill that cannot be read fails the fill with the disk error in the slot the caller + /// probes, and the black fallback frame never reaches the sink (#2570 review). + #[test] + fn a_spill_read_failure_is_an_error_and_never_streamed() { + let (input, _) = sequence(4, 1.0); + let spill = spill_frames(&input.frames); + // Truncate the scratch under the fill: every read comes back short. + spill.inner.file.set_len(0).expect("truncate"); + for method in [FillMethod::Object, FillMethod::Surface, FillMethod::EdgeBlend] { + let mut seen = Vec::new(); + let mut sink = |t: usize, _: &Image| -> Result<(), String> { + seen.push(t); + Ok(()) + }; + let failed = FillInput { frames: Frames::Spilled(spill.clone()), holes: input.holes.clone(), references: vec![] }; + let out = fill_sequence_with(failed, &FillOpts { method, ..Default::default() }, &CpuFill, &mut |_, _| true, Some(&mut sink)); + assert!(out.is_none(), "{method:?}"); + assert!(seen.is_empty(), "{method:?}: streamed frames {seen:?} from an unreadable spill"); + let e = spill.take_error().expect("the read failure is recorded"); + assert!(e.starts_with("spill:"), "{method:?}: {e}"); + } + } + + /// A scratch failure while the output streams stops the fill before the next frame reaches + /// the sink, and leaves the failure for the caller to report. + #[test] + fn a_spill_failure_mid_stream_stops_before_the_next_frame() { + let (input, _) = sequence(6, 1.0); + let spill = spill_frames(&input.frames); + let cell = spill.error_cell(); + let mut seen = Vec::new(); + let mut sink = |t: usize, _: &Image| -> Result<(), String> { + seen.push(t); + if t == 2 { + *cell.lock().unwrap() = Some("spill: injected".into()); + } + Ok(()) + }; + let streamed = FillInput { frames: Frames::Spilled(spill.clone()), holes: input.holes.clone(), references: vec![] }; + let out = fill_sequence_with(streamed, &FillOpts::default(), &CpuFill, &mut |_, _| true, Some(&mut sink)); + assert!(out.is_none()); + assert_eq!(seen, vec![0, 1, 2]); + assert_eq!(spill.take_error().as_deref(), Some("spill: injected")); + } + + /// Once a propagated frame fails to spill, the other propagation workers stop instead of + /// tracing the rest of a range whose fill is already lost (#2570 review). + #[test] + fn propagation_stops_after_a_spill_failure() { + /// Fails the spill from inside the propagation stage (Lighting Correction's membrane). + struct FailInPropagation { + cell: std::sync::Arc>>, + calls: std::sync::atomic::AtomicUsize, + } + impl FillBackend for FailInPropagation { + fn membrane_known(&self, data: &mut [[f32; 4]], known: &[bool], w: usize, h: usize) { + self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + *self.cell.lock().unwrap() = Some("spill: injected".into()); + membrane_core(data, known, w, h); + } + } + let (input, _) = sequence(6, 1.0); + let spill = spill_frames(&input.frames); + let backend = FailInPropagation { cell: spill.error_cell(), calls: std::sync::atomic::AtomicUsize::new(0) }; + let input = FillInput { frames: Frames::Spilled(spill.clone()), holes: input.holes, references: vec![] }; + let opts = FillOpts { lighting: 1.0, ..Default::default() }; + // One worker, so the frames run one after another and the count is exact. + let pool = rayon::ThreadPoolBuilder::new().num_threads(1).build().unwrap(); + let out = pool.install(|| fill_sequence_with(input, &opts, &backend, &mut |_, _| true, None)); + assert!(out.is_none()); + assert_eq!(backend.calls.load(std::sync::atomic::Ordering::SeqCst), 1, "propagation kept going after the failure"); + assert_eq!(spill.take_error().as_deref(), Some("spill: injected")); + } + + /// Peak RSS of one synthetic fill, quoted in the PR. Ignored so CI does not run it. Run it + /// under `/usr/bin/time -v` and read `VmHWM` (`Maximum resident set size`): + /// + /// ```text + /// cargo test -p effectcraft-raster --lib measure_peak_rss --no-run + /// target/debug/deps/effectcraft_raster-* inpaint::tests::measure_peak_rss_of_a_fill \ + /// --ignored --exact --nocapture --test-threads=1 + /// ``` + /// + /// `RAYON_NUM_THREADS`, `MEASURE_LIGHTING=1` and `MEASURE_FRAMES` select the configuration. + /// Measured on 18 workers (2026-11-04): 90 frames of 640×360 peak at 340–470 MB across runs + /// (35 MB with one worker) and 600 frames at 582 MB, i.e. ~7 frames of scratch a worker, so + /// the guard budgets 8 (9 with Lighting Correction) (#2570). + #[cfg(not(target_arch = "wasm32"))] + #[test] + #[ignore = "memory measurement; run explicitly"] + fn measure_peak_rss_of_a_fill() { + let n: usize = std::env::var("MEASURE_FRAMES").ok().and_then(|v| v.parse().ok()).unwrap_or(90); + let (w, h) = (640u32, 360u32); + let spill = SpillHandle::create_in(None, w as usize, h as usize, n, 32).expect("spill"); + let hole = rect_hole(w, h, 240, 120, 160); + let mut holes = Vec::with_capacity(n); + for k in 0..n { + let img = background(w, h, 2.0 * k as f32); + spill.write_frame(k, &img).expect("write"); + holes.push(HoleMask::from_vec(&hole)); + } + let input = FillInput { frames: Frames::Spilled(spill), holes, references: vec![] }; + // MEASURE_LIGHTING=1 exercises the propagation ring too (one process per reading: VmHWM + // never goes down). + let lighting = std::env::var("MEASURE_LIGHTING").is_ok_and(|v| v == "1"); + let opts = FillOpts { method: FillMethod::Surface, lighting: if lighting { 1.0 } else { 0.0 }, ..Default::default() }; + let before = peak_rss_kb(); + let started = std::time::Instant::now(); + let mut sink = |_: usize, _: &Image| -> Result<(), String> { Ok(()) }; + let out = fill_sequence_with(input, &opts, &CpuFill, &mut |_, _| true, Some(&mut sink)); + eprintln!( + "measure_peak_rss: {n} frames of {w}x{h}, surface, lighting={}, {:.2}s, peak RSS {} KB -> {} KB", + if lighting { "on" } else { "off" }, + started.elapsed().as_secs_f64(), + before.unwrap_or(0), + peak_rss_kb().unwrap_or(0) + ); + assert!(out.is_some_and(|v| v.is_empty()), "the sink consumed every frame"); + } + + /// Peak resident set size of this process (`VmHWM`), in kilobytes. + #[cfg(target_os = "linux")] + fn peak_rss_kb() -> Option { + std::fs::read_to_string("/proc/self/status").ok()?.lines().find_map(|l| { + let v = l.strip_prefix("VmHWM:")?; + v.split_whitespace().next()?.parse().ok() + }) + } + + /// Peak resident set size is only measured on Linux. + #[cfg(not(target_os = "linux"))] + fn peak_rss_kb() -> Option { + None + } }