diff --git a/crates/media/src/lib.rs b/crates/media/src/lib.rs index 000c1624..7ef790bc 100644 --- a/crates/media/src/lib.rs +++ b/crates/media/src/lib.rs @@ -35,6 +35,9 @@ pub enum MediaError { Unsupported(String), #[error("decode error: {0}")] Decode(String), + /// The read was called off before it finished (see [`effectcraft_render::FootageSource::cancellable`]). + #[error("cancelled")] + Cancelled, } pub type Result = std::result::Result; @@ -45,6 +48,7 @@ impl From for MediaError { match e { F::Unsupported(s) => MediaError::Unsupported(s), F::Io(s) | F::Offline(s) => MediaError::Io(s), + F::Cancelled => MediaError::Cancelled, other => MediaError::Decode(other.to_string()), } } diff --git a/crates/media/src/pool.rs b/crates/media/src/pool.rs index ce980c56..a8dae92a 100644 --- a/crates/media/src/pool.rs +++ b/crates/media/src/pool.rs @@ -355,6 +355,8 @@ impl MediaPool { } match Inner::frame_at(&self.inner, footage, t, true, size) { Ok(img) => Some(img), + // nobody wants the frame any more (`cancellable`): not an error + Err(MediaError::Cancelled) => None, Err(e) => { log::warn!("media: frame of {} at {:?}: {e}", footage.path, t); None @@ -791,6 +793,12 @@ impl FootageSource for MediaPool { fn cache_budget(&self) -> Option { Some(self.budget()) } + /// A movie's decoder gives up on a seek (before it, and between samples while it decodes + /// forward from the keyframe) once `cancel` is set: a scrub's superseded frames no longer + /// hold the decoder the frame under the pointer waits for (#329). + fn cancellable(&self, cancel: &Arc, f: &mut dyn FnMut()) { + filmcraft_media::cancel::with_cancel(cancel, f) + } fn frame(&self, item: ItemId, footage: &Footage, t: Tick) -> Option> { self.frame_sized(item, footage, t, None) } diff --git a/crates/media/tests/footage.rs b/crates/media/tests/footage.rs index d26b1a1d..36e3657a 100644 --- a/crates/media/tests/footage.rs +++ b/crates/media/tests/footage.rs @@ -674,3 +674,25 @@ fn conformed_flac_audio_reads_back_like_the_decoder() { assert_eq!(pool.audio_samples(&f, at, 4800, 48_000), decoded); let _ = std::fs::remove_dir_all(&folder); } + +/// #329: a frame read that is called off (a scrub has moved on) gives up before the decoder +/// seeks and decodes forward from the keyframe, so it no longer holds the decoder the newest +/// frame waits for; the frame decodes as usual when it is wanted again. +#[test] +fn a_called_off_read_does_not_decode() { + use std::sync::atomic::AtomicBool; + let Some(p) = h264_mp4() else { return }; + let f = probe(&p).expect("probe"); + let pool = MediaPool::new(); + let t = t_of(f.frame_rate, 45); + let off = Arc::new(AtomicBool::new(true)); + let mut got = None; + pool.cancellable(&off, &mut || got = Some(pool.frame(ItemId(1), &f, t))); + assert!(got.expect("ran").is_none(), "a called-off read decoded its frame"); + assert_eq!(pool.stats().frames, 0, "nothing cached for it"); + let on = Arc::new(AtomicBool::new(false)); + let mut got = None; + pool.cancellable(&on, &mut || got = pool.frame(ItemId(1), &f, t)); + let img = got.expect("decoded once wanted again"); + assert_eq!((img.width, img.height), (1920, 1080)); +} diff --git a/crates/render/src/lib.rs b/crates/render/src/lib.rs index f4a5a98a..4f7258ac 100644 --- a/crates/render/src/lib.rs +++ b/crates/render/src/lib.rs @@ -23,6 +23,7 @@ pub mod text; pub mod three_d; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; pub use auto::{AutoKey, AutoPick}; pub use cache::{CacheStats, LayerCache, LayerStore, PrefetchStore}; @@ -85,6 +86,80 @@ pub trait FootageSource: Send + Sync { fn vector_frame(&self, _item: ItemId, _footage: &Footage, _scale: f64) -> Option> { None } + /// Run `f`, which reads frames from this source, as work that can be called off: once + /// `cancel` is set, long decodes it starts (a movie seeking back to a keyframe and decoding + /// forward) stop early and their frames come back `None`. Sources that can't stop a decode + /// just run `f`. See [`CancellableFootage`]. + fn cancellable(&self, _cancel: &Arc, f: &mut dyn FnMut()) { + f() + } + /// Reads through this source have been called off: frames may have come back `None` that + /// exist, so what was made from them must not be kept. + fn called_off(&self) -> bool { + false + } +} + +/// A render's view of a [`FootageSource`] whose frame reads can be called off with `cancel` +/// (a viewer frame nobody waits for any more, #329). Every read goes through +/// [`FootageSource::cancellable`] on its own, so reads made for this render on another thread +/// (a parallel loop) answer to this render's flag, and reads for other renders never do. +pub struct CancellableFootage<'a> { + pub inner: &'a dyn FootageSource, + pub cancel: Arc, +} + +impl CancellableFootage<'_> { + fn read(&self, f: impl FnOnce() -> Option) -> Option { + if self.cancel.load(Ordering::Relaxed) { + return None; + } + let (mut f, mut out) = (Some(f), None); + self.inner.cancellable(&self.cancel, &mut || out = f.take().and_then(|f| f())); + out + } +} + +impl FootageSource for CancellableFootage<'_> { + fn frame(&self, item: ItemId, footage: &Footage, t: Tick) -> Option> { + self.read(|| self.inner.frame(item, footage, t)) + } + fn frame_at_size(&self, item: ItemId, footage: &Footage, t: Tick, width: u32, height: u32) -> Option> { + self.read(|| self.inner.frame_at_size(item, footage, t, width, height)) + } + fn audio(&self, item: ItemId, footage: &Footage, t: Tick, frames: usize, rate: u32) -> Option> { + self.inner.audio(item, footage, t, frames, rate) + } + fn aux(&self, item: ItemId, footage: &Footage, t: Tick) -> Option> { + self.inner.aux(item, footage, t) + } + fn model(&self, item: ItemId, footage: &Footage) -> Option> { + self.inner.model(item, footage) + } + fn set_cache_budget(&self, bytes: usize) { + self.inner.set_cache_budget(bytes) + } + fn purge(&self) { + self.inner.purge() + } + fn forget(&self, path: &str) { + self.inner.forget(path) + } + fn set_conform_folder(&self, folder: Option) { + self.inner.set_conform_folder(folder) + } + fn cache_budget(&self) -> Option { + self.inner.cache_budget() + } + fn vector_frame(&self, item: ItemId, footage: &Footage, scale: f64) -> Option> { + self.inner.vector_frame(item, footage, scale) + } + fn cancellable(&self, cancel: &Arc, f: &mut dyn FnMut()) { + self.inner.cancellable(cancel, f) + } + fn called_off(&self) -> bool { + self.cancel.load(Ordering::Relaxed) || self.inner.called_off() + } } /// Footage that can be rasterised at any scale (SVG). @@ -579,6 +654,12 @@ impl<'a> Renderer<'a> { } } + /// The layer cache to keep this render's buffers in: none once its footage reads were + /// called off, as they may have come back empty ([`FootageSource::called_off`], #329). + fn cache_to_keep(&self) -> Option<&'a LayerCache> { + self.cache.filter(|_| !self.footage.called_off()) + } + /// A layer's Quality switch, limited by the precomp layers above it when /// [`RenderOpts::nested_switches`] is on. pub fn quality(&self, layer: &Layer) -> Quality { @@ -1253,7 +1334,7 @@ impl<'a> Renderer<'a> { let b = self.footage.frame(item, f, f.frame_rate.tick_of(i + 1)); if let (Some(a), Some(b)) = (a, b) { let img = blend_frames(mode, &a, &b, w as f32); - if let (Some(c), Some(k)) = (self.cache, key) { + if let (Some(c), Some(k)) = (self.cache_to_keep(), key) { c.insert(k, Arc::new(Buf { img: img.clone(), offset: [0.0; 2], scale: 1.0 })); } return Some(Arc::new(img)); @@ -1304,7 +1385,7 @@ impl<'a> Renderer<'a> { let mut nb = (*b).clone(); color::convert(&mut nb.img, &c); let nb = Arc::new(nb); - if let (Some(cache), Some(k)) = (self.cache, key) { + if let (Some(cache), Some(k)) = (self.cache_to_keep(), key) { cache.insert(k, nb.clone()); } nb @@ -1322,7 +1403,7 @@ impl<'a> Renderer<'a> { return Some((b, key)); } let flat = Arc::new(styles::Styled { passes: st.passes.iter().map(|(b, m)| ((**b).clone(), *m)).collect(), body: (*st.body).clone() }.flatten()); - if let (Some(c), Some(k)) = (self.cache, key) { + if let (Some(c), Some(k)) = (self.cache_to_keep(), key) { c.insert(k, flat.clone()); } Some((flat, key)) @@ -1353,7 +1434,7 @@ impl<'a> Renderer<'a> { } let buf = self.masked_source(ctx, layer)?; let buf = Arc::new(self.apply_effects_timed(ctx, layer, buf, false, usize::MAX, timing.map(|t| &mut t.effects))); - if let (Some(c), Some(k)) = (self.cache, key) { + if let (Some(c), Some(k)) = (self.cache_to_keep(), key) { c.insert(k, buf.clone()); } Some((buf, key)) @@ -1378,7 +1459,7 @@ impl<'a> Renderer<'a> { buf = self.effect_space_stack(ctx, layer, buf, false, effects, None, origin); } let buf = Arc::new(buf); - if let (Some(c), Some(k)) = (self.cache, key) { + if let (Some(c), Some(k)) = (self.cache_to_keep(), key) { c.insert(k, buf.clone()); } Some(buf) @@ -1410,7 +1491,7 @@ impl<'a> Renderer<'a> { } let body = Arc::new(st.body); let passes: Vec<_> = st.passes.into_iter().map(|(b, m)| (Arc::new(b), m)).collect(); - if let (Some(c), Some(k)) = (self.cache, key) { + if let (Some(c), Some(k)) = (self.cache_to_keep(), key) { c.insert(k, body.clone()); for (i, (b, _)) in passes.iter().enumerate() { c.insert(cache::derive(k, i as u64 + 1), b.clone()); @@ -2034,7 +2115,7 @@ impl<'a> Renderer<'a> { let mut foot = self.source(ctx, layer)?; let _ = masks::apply(ctx, layer, &mut foot, self.mask_blur(ctx, layer)); let foot = Arc::new(foot); - if let (Some(c), Some(k)) = (self.cache, key) { + if let (Some(c), Some(k)) = (self.cache_to_keep(), key) { c.insert(k, foot.clone()); } Some(foot) diff --git a/crates/render/src/tests.rs b/crates/render/src/tests.rs index 7e065391..fbc219d1 100644 --- a/crates/render/src/tests.rs +++ b/crates/render/src/tests.rs @@ -268,6 +268,23 @@ fn render_cached(p: &Project, cid: ItemId, t: Tick, cache: Option<&crate::LayerC r.comp_frame(cid, t) } +/// #329: a render whose footage reads were called off keeps nothing in the layer cache: a layer +/// made from a frame that came back empty (an effect's layer parameter) would be reused by the +/// next render of that time. +#[test] +fn a_called_off_render_keeps_nothing_in_the_layer_cache() { + let (p, cid) = cache_scene(); + let t = Tick::from_seconds_f64(0.5); + let cache = crate::LayerCache::default(); + let off = crate::CancellableFootage { inner: &crate::NoFootage, cancel: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true)) }; + let mut r = crate::Renderer::new(&p, &off, crate::RenderOpts::default()); + r.cache = Some(&cache); + r.comp_frame(cid, t); + assert_eq!(cache.stats().entries, 0, "a called-off render kept layers"); + render_cached(&p, cid, t, Some(&cache)); + assert!(cache.stats().entries > 0, "a render that runs to the end keeps its layers"); +} + /// [`add_effect`] for a 200×100 layer. fn add_effect_200(p: &mut Project, l: &mut effectcraft_project::Layer, id: &str, vals: &[(&str, Value)]) { add_effect(p, l, id, [200.0, 100.0], vals); diff --git a/crates/ui-egui/src/frames.rs b/crates/ui-egui/src/frames.rs index ccffeb9f..164b242a 100644 --- a/crates/ui-egui/src/frames.rs +++ b/crates/ui-egui/src/frames.rs @@ -29,7 +29,7 @@ use std::sync::{Arc, Mutex}; use effectcraft_engine::project::{Comp, ItemId, ItemKind, LayerSource, Node, Project, PropGroup}; use effectcraft_engine::render::disk_cache::{self, DiskCache}; -use effectcraft_engine::render::{ExprHost, FootageSource, LayerCache, RenderOpts, Renderer}; +use effectcraft_engine::render::{CancellableFootage, ExprHost, FootageSource, LayerCache, RenderOpts, Renderer}; use effectcraft_engine::time::Tick; use effectcraft_gpu::{DisplayFrame, Gpu}; use rayon::prelude::*; @@ -756,13 +756,22 @@ impl Frames { for j in q.jobs.iter_mut() { j.urgent = j.key == key; } + // The comp's viewer frames of other times that are rendering are called off too: a + // movie seeking for each of them held the decoder the frame under the pointer + // waited for, so it came seconds after the pointer stopped (#329). Another revision + // of the same frame (a live edit) finishes, as the viewer shows it meanwhile. + for (k, cancel) in &q.urgent_cancels { + if k.comp == key.comp && k.frame != key.frame { + cancel.store(true, Ordering::Relaxed); + } + } } if !inf.insert(key) { return; } q.seq += 1; let seq = q.seq; - q.jobs.push(Job { key, src: src.clone(), comp, t, opts, urgent, seq }); + q.jobs.push(Job { key, src: src.clone(), comp, t, opts, urgent, seq, cancel: Arc::default() }); drop(inf); drop(q); #[cfg(not(target_arch = "wasm32"))] @@ -968,6 +977,7 @@ impl Worker { q.running.insert(job.key); if job.urgent { q.urgent_running += 1; + q.urgent_cancels.push((job.key, job.cancel.clone())); } job }; @@ -977,10 +987,11 @@ impl Worker { let disk_hit = from_disk.is_some(); let ci = match from_disk { Some(f) => FrameImage::Cpu(Arc::new(egui::ColorImage::from_rgba_premultiplied([f.width as usize, f.height as usize], &f.rgba))), - None => match self.render_guarded(&job) { + None => match self.render_guarded(&job).filter(|_| !job.cancel.load(Ordering::Relaxed)) { Some(ci) => ci, + // Called off (frames it read may be missing) or failed: not cached, and not + // "in progress" forever: the viewer asks for it again. None => { - // Not "in progress" forever: the viewer asks for it again. self.release(job.key, job.urgent); return true; } @@ -1025,6 +1036,9 @@ impl Worker { q.running.remove(&key); if urgent { q.urgent_running = q.urgent_running.saturating_sub(1); + if let Some(i) = q.urgent_cancels.iter().position(|(k, _)| *k == key) { + q.urgent_cancels.swap_remove(i); + } } } if let Some(ctx) = &self.ctx { @@ -1061,7 +1075,8 @@ impl Worker { return None; } let g = job.src.gpu.as_ref()?; - let mut r = Renderer::new(&job.src.project, job.src.footage.as_ref(), job.opts); + let footage = job.footage(); + let mut r = Renderer::new(&job.src.project, &footage, job.opts); r.expr = job.src.expr.as_deref(); r.cache = Some(&job.src.layer_cache); r.accel = Some(g as &dyn effectcraft_engine::render::Accelerator); @@ -1077,7 +1092,8 @@ impl Worker { /// Render a frame; `gpu: false` keeps it off the GPU compositor. fn render_job(&self, job: &Job, gpu: bool) -> FrameImage { - let mut r = Renderer::new(&job.src.project, job.src.footage.as_ref(), job.opts); + let footage = job.footage(); + let mut r = Renderer::new(&job.src.project, &footage, job.opts); r.expr = job.src.expr.as_deref(); let gpu_allowed = gpu && !self.gpu_retired.load(Ordering::Acquire); // A failed GPU attempt may have populated intermediate entries before its scope @@ -1149,6 +1165,15 @@ struct Job { opts: RenderOpts, urgent: bool, seq: u64, + /// Set when the viewer no longer waits for this frame (see [`Frames::request_urgent`]). + cancel: Arc, +} + +impl Job { + /// The footage this job's render reads, called off with the job. + fn footage(&self) -> CancellableFootage<'_> { + CancellableFootage { inner: self.src.footage.as_ref(), cancel: self.cancel.clone() } + } } #[derive(Default)] @@ -1156,6 +1181,8 @@ struct Queue { jobs: Vec, running: HashSet, urgent_running: usize, + /// The running viewer (urgent) frames and how to call them off. + urgent_cancels: Vec<(FrameKey, Arc)>, seq: u64, } @@ -1341,6 +1368,94 @@ mod tests { assert_eq!(footage.deepest.load(Ordering::SeqCst), 1, "renders nested on a waiting thread"); } + /// #329: scrubbing a long movie, the viewer frame of every step the pointer passed started + /// rendering and seeked the movie's decoder in turn; the frame under the pointer waited for + /// all of them. A newer viewer frame of another time calls off the one rendering, which is + /// not cached (its footage may be missing). + #[test] + fn a_newer_viewer_frame_calls_off_the_one_rendering() { + use effectcraft_engine::Session; + use effectcraft_engine::project::{Footage, FootageKind}; + use effectcraft_engine::render::Image; + use effectcraft_engine::time::FrameRate; + use serde_json::json; + use std::cell::RefCell; + thread_local! { + static CANCEL: RefCell>> = const { RefCell::new(None) }; + } + /// A movie whose first frames take a long seek, which a called-off read gives up. + #[derive(Default)] + struct Seeking { + seeking: AtomicBool, + called_off: AtomicBool, + } + impl FootageSource for Seeking { + fn cancellable(&self, cancel: &Arc, f: &mut dyn FnMut()) { + let prev = CANCEL.replace(Some(cancel.clone())); + f(); + CANCEL.set(prev); + } + fn frame(&self, _: ItemId, _: &Footage, t: Tick) -> Option> { + if t < Tick::from_seconds_f64(0.25) { + self.seeking.store(true, Ordering::SeqCst); + for _ in 0..1000 { + if CANCEL.with_borrow(|c| c.as_ref().is_some_and(|c| c.load(Ordering::SeqCst))) { + self.called_off.store(true, Ordering::SeqCst); + return None; + } + std::thread::sleep(std::time::Duration::from_millis(5)); + } + } + Some(Arc::new(Image::new(4, 4))) + } + } + let mut s = Session::default(); + let cid = ItemId(s.execute("comp.new", json!({"name": "A", "width": 8, "height": 8, "duration": 10})).unwrap()["comp"].as_u64().unwrap()); + let clip = Footage { + kind: FootageKind::Video, + width: 4, + height: 4, + has_video: true, + frame_rate: FrameRate::FPS_30, + duration: Tick::from_seconds_f64(10.0), + ..Default::default() + }; + let item = Arc::make_mut(&mut s.project).add_item("clip.mov", Default::default(), None, ItemKind::Footage(clip)); + s.execute("layer.addItem", json!({"item": item.0})).unwrap(); + let footage = Arc::new(Seeking::default()); + let src = RenderSource { + project: s.project.clone(), + footage: footage.clone(), + expr: None, + layer_cache: Arc::default(), + gpu: None, + gpu_display: false, + disk: None, + }; + let f = Frames::default(); + let content = f.content_of(&s.project, s.revision, cid); + let key = |frame| FrameKey { revision: s.revision, content, comp: cid.0, frame, scale: 1000, view: 0, opts: 0 }; + f.request_urgent(&src, key(0), cid, Tick::ZERO, RenderOpts::default()); + for _ in 0..1000 { + if footage.seeking.load(Ordering::SeqCst) { + break; + } + std::thread::sleep(std::time::Duration::from_millis(2)); + } + assert!(footage.seeking.load(Ordering::SeqCst), "the first frame's render reached the footage"); + f.request_urgent(&src, key(150), cid, FrameRate::FPS_30.tick_of(150), RenderOpts::default()); + for _ in 0..1000 { + if f.inflight() == 0 { + break; + } + std::thread::sleep(std::time::Duration::from_millis(5)); + } + assert!(footage.called_off.load(Ordering::SeqCst), "the superseded frame's seek ran on"); + assert!(f.is_cached(&key(150)), "the newest frame rendered"); + assert!(!f.is_cached(&key(0)), "a called-off frame is not cached"); + assert_eq!(f.inflight(), 0, "released"); + } + /// #284: scrubbing asks for a viewer frame at every step; a newer one drops the comp's viewer /// frames still waiting (they would all render first), but not prefetch frames or another /// comp's viewer frame.