diff --git a/crates/codecs/src/gop.rs b/crates/codecs/src/gop.rs --- a/crates/codecs/src/gop.rs +++ b/crates/codecs/src/gop.rs @@ -293,6 +293,70 @@ struct Pool { struct Shared { state: Mutex, pool: Arc, + /// The pts each request is waiting for (one entry per request: its job's cancellation flag, + /// None when it cannot be cancelled), registered before the request takes `state`'s lock. + /// Eviction keeps these frames: with several render jobs on one source, the one that holds + /// the lock decodes past the pictures the others wait for, and evicting those made each of + /// them seek back and decode their GOP again. A request whose job was cancelled (a frame a + /// scrub has left) stays registered until it returns but no longer counts ([`live`]). + /// Lock order: `state`, then `waiting` (never the other way round). + waiting: Mutex>>, +} + +/// A registered request's cancellation flag. +type Waiter = Option>; + +/// Whether any of the requests waiting for a frame still wants it (not cancelled). +fn live(w: &[Waiter]) -> bool { + w.iter().any(|c| !c.as_ref().is_some_and(|c| c.load(Ordering::Relaxed))) +} + +/// A request's entry in [`Shared::waiting`], removed on drop (every return path). +struct Waiting<'a> { + shared: &'a Shared, + pts: i64, + cancel: Waiter, +} + +impl<'a> Waiting<'a> { + /// Register a request for `pts`, under this thread's cancellation flag. + fn register(shared: &'a Shared, pts: i64) -> Self { + let cancel = filmcraft_media::cancel::current_flag(); + shared.waiting.lock().unwrap_or_else(PoisonError::into_inner).entry(pts).or_default().push(cancel.clone()); + Self { shared, pts, cancel } + } +} + +impl Drop for Waiting<'_> { + fn drop(&mut self) { + let mut w = self.shared.waiting.lock().unwrap_or_else(PoisonError::into_inner); + if let Some(v) = w.get_mut(&self.pts) { + // this request's entry (any entry with the same flag, or with none, is equivalent) + let same = |c: &Waiter| match (c, &self.cancel) { + (Some(a), Some(b)) => Arc::ptr_eq(a, b), + (None, None) => true, + _ => false, + }; + if let Some(i) = v.iter().position(same) { + v.swap_remove(i); + } + if v.is_empty() { + w.remove(&self.pts); + } + } + } +} + +/// The cached frame to evict next: the one farthest from `pivot` at either end of `frames` that +/// is neither `pivot`, `newest` nor waited for (`wanted`). None when every frame is one of those. +fn eviction_victim(frames: &BTreeMap, pivot: i64, newest: i64, wanted: &BTreeMap>) -> Option { + let keep = |k: &i64| *k == pivot || *k == newest || wanted.get(k).is_some_and(|w| live(w)); + let low = frames.keys().find(|k| !keep(k)).copied(); + let high = frames.keys().rev().find(|k| !keep(k)).copied(); + match (low, high) { + (Some(l), Some(h)) => Some(if pivot.saturating_sub(l) > h.saturating_sub(pivot) { l } else { h }), + (l, h) => l.or(h), + } } impl Drop for Shared { @@ -455,8 +519,10 @@ const MIN_FRAMES: usize = 64; const FRAME_CACHE_CEILING: usize = 384 << 20; /// A running decoder up to this many samples before the wanted sample's sync sample keeps going -/// rather than restarting at the sync sample. -const CONTINUE_THROUGH: usize = 48; +/// rather than restarting at the sync sample (a restart costs the reorder delay again). Further +/// back it restarts, unless another request waits for a frame in between: decoding a whole GOP +/// nobody wants (a decoder that fell behind the playhead) costs more than the restart. +const CONTINUE_THROUGH: usize = 8; /// Threads that run the caches' decoding (see [`isolated`]); `None` where none can be made. #[cfg(not(target_arch = "wasm32"))] @@ -533,6 +599,7 @@ impl GopCache { background: false, }), pool: pool.clone(), + waiting: Mutex::new(BTreeMap::new()), }); pool.register(&shared); Self { shared, explicit_color, rotation: 0, crop: None, budget: FRAME_CACHE_CEILING } @@ -595,16 +662,12 @@ impl GopCache { // Evict frames far from the request (pin), not the newest decode. A frame-threaded // decoder can emit a burst past the playhead; pivoting on those pts used to drop the // wanted picture and walk the GOP again. + // Frames other requests wait for are kept as well (bounded by the number of requests). let pivot = if st.pin != i64::MIN { st.pin } else { pts }; + // (held, not copied, through the loop: lock order is state, then waiting) + let wanted = self.shared.waiting.lock().unwrap_or_else(PoisonError::into_inner); while st.bytes > budget && st.frames.len() > 2 { - let (Some(&first), Some(&last)) = (st.frames.keys().next(), st.frames.keys().next_back()) else { break }; - let mut victim = if pivot - first > last - pivot { first } else { last }; - if victim == pivot || victim == pts { - victim = if victim == first { last } else { first }; - if victim == pivot || victim == pts { - break; - } - } + let Some(victim) = eviction_victim(&st.frames, pivot, pts, &wanted) else { break }; if let Some(v) = st.frames.remove(&victim) { st.drafts.remove(&victim); st.bytes -= v.byte_size(); @@ -613,6 +676,7 @@ impl GopCache { filmcraft_frame::pool::recycle(v); } } + drop(wanted); let pool = &self.shared.pool; if st.bytes >= before { pool.bytes.fetch_add(st.bytes - before, Ordering::Relaxed); @@ -667,6 +731,7 @@ impl GopCache { if DecodingGuard::active(me) { return self.private_frame(s, i, want_pts, n); } + let _waiting = Waiting::register(&self.shared, want_pts); let draft = filmcraft_media::cancel::draft(); // Idle caches' decoders released below. Declared before the lock guard, so it is dropped // after the guard on every return path: dropping a decoder can block (a hardware session @@ -734,7 +799,11 @@ impl GopCache { let running = st.next != usize::MAX && st.next <= n && st.start <= key; // Decoding on through a short stretch into the next GOP is cheaper than a restart, and // playback wants those frames anyway. - let near = st.next <= key && key - st.next <= CONTINUE_THROUGH; + let wanted_on_the_way = || { + let w = self.shared.waiting.lock().unwrap_or_else(PoisonError::into_inner); + w.range(..s.pts(key)).rev().find(|(_, v)| live(v)).is_some_and(|(&p, _)| p > st.out_max) + }; + let near = st.next <= key && (key - st.next <= CONTINUE_THROUGH || wanted_on_the_way()); let continuing = running && (st.next > key || near) && (i >= st.next || (st.start <= i && want_pts > st.out_max)); if !continuing { SEEKS.fetch_add(1, Ordering::Relaxed); @@ -767,7 +836,13 @@ impl GopCache { } let k = st.next; let data = s.read(k)?; - if k != i && s.pts(k) < late && st.decoder.as_ref().is_some_and(|d| d.is_disposable(&data)) { + // a late disposable picture is skipped, unless another request waits for it (lock + // order: state, then waiting) + if k != i + && s.pts(k) < late + && st.decoder.as_ref().is_some_and(|d| d.is_disposable(&data)) + && !self.shared.waiting.lock().unwrap_or_else(PoisonError::into_inner).get(&s.pts(k)).is_some_and(|v| live(v)) + { st.next += 1; SKIPPED.fetch_add(1, Ordering::Relaxed); continue; diff --git a/crates/codecs/src/gop/tests.rs b/crates/codecs/src/gop/tests.rs --- a/crates/codecs/src/gop/tests.rs +++ b/crates/codecs/src/gop/tests.rs @@ -237,6 +237,67 @@ fn running_decoder_continues_through_a_nearby_sync_sample() { assert!(s.decodes.load(Ordering::Relaxed) <= 41 + 1); } +#[test] +fn a_decoder_far_behind_restarts_at_the_wanted_gop() { + // sync samples every 30: the decoder stands at 26, the wanted frame 70 is two GOPs on. + // Decoding 26..70 (44 pictures nobody asked for) used to be preferred over a restart at 60. + let s = samples(200, 30, 0, false); + let c = GopCache::new(None); + assert_eq!(index_of(&c.frame(&s, 25_000).expect("frame")), 25); + let before = s.decodes.load(Ordering::Relaxed); + assert_eq!(index_of(&c.frame(&s, 70_000).expect("frame")), 70); + assert_eq!(s.resets.load(Ordering::Relaxed), 2, "restarted at the sync sample"); + assert!(s.decodes.load(Ordering::Relaxed) - before <= 12, "decoded {}", s.decodes.load(Ordering::Relaxed) - before); +} + +#[test] +fn a_decoder_far_behind_continues_when_a_frame_on_the_way_is_waited_for() { + // as above, but another request waits for frame 40 (between the decoder at 26 and the + // sync sample at 60): decoding on to 70 serves it too, so there is no restart + let s = samples(200, 30, 0, false); + let c = GopCache::new(None); + assert_eq!(index_of(&c.frame(&s, 25_000).expect("frame")), 25); + let _other = Waiting::register(&c.shared, 40_000); + assert_eq!(index_of(&c.frame(&s, 70_000).expect("frame")), 70); + assert_eq!(s.resets.load(Ordering::Relaxed), 1, "continued instead of restarting at 60"); +} + +#[test] +fn a_cancelled_request_does_not_keep_a_decoder_far_behind_going() { + // a scrub leaves superseded requests registered until they return: one waiting for frame + // 40 whose job was cancelled must not make the decoder walk 26..70 instead of restarting + let s = samples(200, 30, 0, false); + let c = GopCache::new(None); + assert_eq!(index_of(&c.frame(&s, 25_000).expect("frame")), 25); + let flag = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let _stale = filmcraft_media::cancel::with_cancel(&flag, || Waiting::register(&c.shared, 40_000)); + flag.store(true, Ordering::Relaxed); + let before = s.decodes.load(Ordering::Relaxed); + assert_eq!(index_of(&c.frame(&s, 70_000).expect("frame")), 70); + assert_eq!(s.resets.load(Ordering::Relaxed), 2, "restarted at the sync sample"); + assert!(s.decodes.load(Ordering::Relaxed) - before <= 12, "decoded {}", s.decodes.load(Ordering::Relaxed) - before); +} + +#[test] +fn eviction_keeps_the_frames_requests_wait_for() { + let frames: BTreeMap = (0..10).map(|k| (k, ())).collect(); + let none = BTreeMap::new(); + // pivot 9: the oldest frame goes first, as before + assert_eq!(eviction_victim(&frames, 9, 9, &none), Some(0)); + // frames 0..=2 are waited for by other render jobs: the next oldest goes instead + let wanted: BTreeMap> = [(0, vec![None]), (1, vec![None]), (2, vec![None, None])].into_iter().collect(); + assert_eq!(eviction_victim(&frames, 9, 9, &wanted), Some(3)); + // nothing left that may go + let all: BTreeMap> = (0..10).map(|k| (k, vec![None])).collect(); + assert_eq!(eviction_victim(&frames, 9, 9, &all), None); + // a frame only cancelled requests wait for may go + let gone = Some(Arc::new(std::sync::atomic::AtomicBool::new(true))); + let stale: BTreeMap> = [(0, vec![gone])].into_iter().collect(); + assert_eq!(eviction_victim(&frames, 9, 9, &stale), Some(0)); + // the farther end wins, whichever side it is on + assert_eq!(eviction_victim(&frames, 1, 1, &none), Some(9)); +} + #[test] fn intra_frames_decode_in_parallel_once_each() { let s = Arc::new(samples(64, 1, 0, true)); diff --git a/crates/media/src/cancel.rs b/crates/media/src/cancel.rs --- a/crates/media/src/cancel.rs +++ b/crates/media/src/cancel.rs @@ -91,6 +91,12 @@ pub fn with_cancel(flag: &Arc, f: impl FnOnce() -> R) -> R { r } +/// This thread's cancellation flag ([`with_cancel`]), to tell later whether the work that is +/// running now was cancelled (None: it cannot be). +pub fn current_flag() -> Option> { + CURRENT.with(|c| c.borrow().clone()) +} + /// Whether the work running on this thread has been cancelled. pub fn cancelled() -> bool { CURRENT.with(|c| c.borrow().as_ref().is_some_and(|f| f.load(Ordering::Relaxed))) diff --git a/docs/architecture.md b/docs/architecture.md --- a/docs/architecture.md +++ b/docs/architecture.md @@ -245,6 +245,11 @@ file ──► codecs (MP4/MOV, MKV, audio) demux + decode, GOP-aware see every frame of the GOP. Sequential playback reuses the decoder. Decoders implement `codecs::VideoDecoder`. `register_video_decoder` puts a factory in front of the built-in ones, so a hardware decoder can take precedence. The GOP cache never holds its lock while decoding. + Several frame jobs can share one decoder: each request registers the frame it waits for, and + eviction keeps those frames, so a job decoding past them does not make the others seek back and + decode their GOP again. A decoder that has not reached the wanted frame's sync sample and is more + than 8 samples short of it restarts there instead of decoding forward, unless another request + waits for a frame in between; one already past that sync sample carries on. - **Hardware decoding.** The apps (desktop, CLI / headless MCP, bench) call `filmcraft_platform::register()` at startup, which on macOS registers a VideoToolbox factory for `avcC` / `hvcC` streams (8 / 10-bit, 4:2:0 and 4:2:2; elsewhere it does nothing), and on Windows a