Merge pull request #784 from vangel76/fix-329-seek-latency

#329: scrubbing a long movie: superseded viewer frames call off their decodes

# Conflicts:
#	crates/media/tests/footage.rs

crates/media/src/lib.rs

@@ -37,6 +37,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<T> = std::result::Result<T, MediaError>;
@@ -47,6 +50,7 @@ impl From<filmcraft_media::MediaError> 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()),
         }
     }

crates/media/src/pool.rs

@@ -357,6 +357,8 @@ impl MediaPool {
         }
         match Inner::frame_at(&self.inner, footage, t, true, size, float) {
             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
@@ -806,6 +808,12 @@ impl FootageSource for MediaPool {
     fn cache_budget(&self) -> Option<usize> {
         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<AtomicBool>, f: &mut dyn FnMut()) {
+        filmcraft_media::cancel::with_cancel(cancel, f)
+    }
     fn frame(&self, item: ItemId, footage: &Footage, t: Tick) -> Option<Arc<Image>> {
         self.frame_sized(item, footage, t, None, FloatCurve::Srgb)
     }

crates/media/tests/footage.rs

@@ -967,3 +967,25 @@ fn cineon_still_shows_code_values() {
         assert!(effectcraft_media::probe_bytes("cut.cin", cut).is_err(), "{n}");
     }
 }
+
+/// #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));
+}

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};
@@ -110,6 +111,80 @@ pub trait FootageSource: Send + Sync {
     fn vector_frame(&self, _item: ItemId, _footage: &Footage, _scale: f64) -> Option<Arc<Image>> {
         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<AtomicBool>, 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<AtomicBool>,
+}
+
+impl CancellableFootage<'_> {
+    fn read<R>(&self, f: impl FnOnce() -> Option<R>) -> Option<R> {
+        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<Arc<Image>> {
+        self.read(|| self.inner.frame(item, footage, t))
+    }
+    fn frame_at_size(&self, item: ItemId, footage: &Footage, t: Tick, width: u32, height: u32) -> Option<Arc<Image>> {
+        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<Vec<f32>> {
+        self.inner.audio(item, footage, t, frames, rate)
+    }
+    fn aux(&self, item: ItemId, footage: &Footage, t: Tick) -> Option<Arc<effectcraft_raster::AuxChannels>> {
+        self.inner.aux(item, footage, t)
+    }
+    fn model(&self, item: ItemId, footage: &Footage) -> Option<Arc<effectcraft_model::Model>> {
+        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<std::path::PathBuf>) {
+        self.inner.set_conform_folder(folder)
+    }
+    fn cache_budget(&self) -> Option<usize> {
+        self.inner.cache_budget()
+    }
+    fn vector_frame(&self, item: ItemId, footage: &Footage, scale: f64) -> Option<Arc<Image>> {
+        self.inner.vector_frame(item, footage, scale)
+    }
+    fn cancellable(&self, cancel: &Arc<AtomicBool>, 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).
@@ -604,6 +679,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 {
@@ -1279,7 +1360,7 @@ impl<'a> Renderer<'a> {
                 let b = self.footage.frame_with(item, f, f.frame_rate.tick_of(i + 1), None, float);
                 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));
@@ -1327,7 +1408,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
@@ -1345,7 +1426,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))
@@ -1376,7 +1457,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))
@@ -1401,7 +1482,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)
@@ -1433,7 +1514,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());
@@ -2057,7 +2138,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)

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);

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<AtomicBool>,
+}
+
+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<Job>,
     running: HashSet<FrameKey>,
     urgent_running: usize,
+    /// The running viewer (urgent) frames and how to call them off.
+    urgent_cancels: Vec<(FrameKey, Arc<AtomicBool>)>,
     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<Option<Arc<AtomicBool>>> = 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<AtomicBool>, f: &mut dyn FnMut()) {
+                let prev = CANCEL.replace(Some(cancel.clone()));
+                f();
+                CANCEL.set(prev);
+            }
+            fn frame(&self, _: ItemId, _: &Footage, t: Tick) -> Option<Arc<Image>> {
+                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.