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 + } }