shifathosn
/
filmcraft
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

#480: exports no longer stop for good when a clip's decoding uses rayon 

A GopCache decodes with its lock held, and that decoding starts rayon work
(rotating iPhone frames, cropping, slice-threaded decoders). On the global pool
a worker that took part of it could, waiting inside it, pick up another render
job asking the same cache for a frame and block on the lock, while the lock
holder waited for the work on that worker's stack: exports of rotated iPhone
HEVC clips stopped making frames in about half the runs.

The locked decoding now runs on a separate decode pool, whose threads never wait
for a cache's lock; a caller that is a rayon worker waits on a plain thread so it
does no other rayon work while holding the lock. The cancel/catch-up/draft/
background hints travel with it (filmcraft_media::cancel::Context). Intra-only
streams crop and rotate before taking the lock again.

Second version of #662 (reverted): the regression test's stand-in decoder summed
with overflow, which panics in debug builds; it now wraps explicitly.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
MrCloudy2 2026-10-10 17:14:34 +02:00
parent 4ecdb6ea71
commit f79a51dd24
4 changed files with 211 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@ -1830,6 +1830,7 @@ dependencies = [
"filmcraft-vp9",
"image",
"log",
"rayon",
"serde_json",
"symphonia",
"thiserror 2.0.21",

1
crates/codecs/Cargo.toml
View File

@ -31,6 +31,7 @@ filmcraft-av1 = { workspace = true }
filmcraft-apv = { workspace = true }
filmcraft-aac = { workspace = true }
filmcraft-opus = { workspace = true }
rayon = { workspace = true }
thiserror = { workspace = true }
log = { workspace = true }
# Decode timing counters (`gop_stats`); std's Instant panics on wasm32.

164
crates/codecs/src/gop.rs
View File

@ -109,8 +109,9 @@ impl std::ops::Sub for GopStats {
}
}
/// A container's video sample table, in decode (file) order.
pub trait VideoSamples {
/// A container's video sample table, in decode (file) order. `Sync`: a cache decodes on its own
/// threads ([`isolated`]).
pub trait VideoSamples: Sync {
fn count(&self) -> usize;
/// Presentation timestamp of sample `i` (track units).
fn pts(&self, i: usize) -> i64;
@ -437,6 +438,57 @@ const MIN_FRAMES: usize = 64;
/// rather than restarting at the sync sample.
const CONTINUE_THROUGH: usize = 48;
/// Threads that run the caches' decoding (see [`isolated`]); `None` where none can be made.
#[cfg(not(target_arch = "wasm32"))]
fn decode_pool() -> Option<&'static rayon::ThreadPool> {
static POOL: std::sync::OnceLock<Option<rayon::ThreadPool>> = std::sync::OnceLock::new();
POOL.get_or_init(|| {
let n = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1);
rayon::ThreadPoolBuilder::new().num_threads(n).thread_name(|i| format!("decode-{i}")).build().ok()
})
.as_ref()
}
/// Run a cache's decoding, its lock held by the caller, on [`decode_pool`], the caller waiting
/// without doing other rayon work meanwhile (#480).
///
/// That decoding starts rayon work (a decoder's slices, cropping and rotating the frames). On the
/// pool of the render jobs that ask the cache for frames, a worker that took part of it could,
/// waiting inside it, pick up another of those jobs and block on the lock: the lock holder then
/// waits for its work on the blocked worker's stack, and that worker for the lock, for good (an
/// export of rotated iPhone clips stopped making frames). On the decode pool that work only ever
/// runs beside other caches' decoding, which never waits for a cache's lock. A caller that is a
/// rayon worker would also run other jobs of its pool while it waits for the decode pool, some
/// of them asking for a cache's lock while it holds this one: it waits on a plain thread instead.
fn isolated<T: Send>(f: impl FnOnce() -> crate::Result<T> + Send) -> crate::Result<T> {
let ctx = filmcraft_media::cancel::Context::current();
let job = move || ctx.run(f);
#[cfg(not(target_arch = "wasm32"))]
if let Some(pool) = decode_pool() {
if rayon::current_thread_index().is_none() {
return pool.install(job);
}
let slot = Mutex::new(Some(job));
let waited = std::thread::scope(|scope| {
let spawned = std::thread::Builder::new().name("decode-wait".into()).spawn_scoped(scope, || {
let job = slot.lock().unwrap_or_else(PoisonError::into_inner).take();
job.map(|job| pool.install(job))
});
spawned.map(|h| h.join())
});
return match waited {
Ok(Ok(Some(r))) => r,
Ok(Err(_)) => Err(CodecError::Decode("internal error while decoding".into())),
// no thread to wait on: decode here, as before
Ok(Ok(None)) | Err(_) => match slot.into_inner().unwrap_or_else(PoisonError::into_inner) {
Some(job) => job(),
None => Err(CodecError::Decode("internal error while decoding".into())),
},
};
}
job()
}
impl GopCache {
pub fn new(explicit_color: Option<ColorInfo>) -> Self {
// unit tests count decoder restarts: each of their caches gets a pool of its own
@ -494,7 +546,11 @@ impl GopCache {
}
fn store(&self, st: &mut State, pts: i64, f: VideoFrame, draft: bool) {
let f = self.finish(f);
self.store_finished(st, pts, self.finish(f), draft);
}
/// Cache a frame [`Self::finish`] has been applied to.
fn store_finished(&self, st: &mut State, pts: i64, f: VideoFrame, draft: bool) {
st.last_used = Instant::now();
if draft {
st.drafts.insert(pts);
@ -616,6 +672,22 @@ impl GopCache {
if st.intra {
return self.intra_frame(st, s, i, want_pts);
}
// with the lock held: see `isolated`
let st = &mut *st;
isolated(|| self.decode_until(st, s, i, want_pts, late_before, draft))
}
/// Decode (the lock held) until the frame with `want_pts` (sample `i`) is cached; return it.
fn decode_until(
&self,
st: &mut State,
s: &dyn VideoSamples,
i: usize,
want_pts: i64,
late_before: Option<i64>,
draft: bool,
) -> crate::Result<Arc<VideoFrame>> {
let n = s.count();
let mut key = decode_start(s, i);
// Continue the running decoder when it has passed the wanted sample's sync sample and
// either has not reached the sample yet or has been fed it without outputting it yet
@ -669,14 +741,14 @@ impl GopCache {
let out = timed(|| decoder.decode(&data, s.pts(k)))?;
st.next += 1;
DECODED.fetch_add(1, Ordering::Relaxed);
self.store_output(&mut st, out);
self.store_output(st, out);
if st.cached(want_pts, draft).is_some() {
break;
}
}
if st.cached(want_pts, draft).is_none() {
let out = st.decoder.as_mut().map(|d| timed(|| d.flush())).unwrap_or_default();
self.store_output(&mut st, out);
self.store_output(st, out);
st.next = usize::MAX;
}
// nearest decoded frame at or before the wanted pts (robust to decoder pts quirks)
@ -761,14 +833,17 @@ impl GopCache {
DECODED.fetch_add(1, Ordering::Relaxed);
timed(|| dec.decode(&data, want_pts))
});
// crop and rotate before taking the lock: that work runs on rayon (see `isolated`)
let res = res.map(|out| {
count_frames(&out);
out.into_iter().map(|d| (d.pts, self.finish(d.frame), d.draft)).collect::<Vec<_>>()
});
let mut st = self.shared.state.lock().unwrap_or_else(|e| e.into_inner());
if st.spare.len() < 16 {
st.spare.push(dec);
}
let res = res?;
count_frames(&res);
for d in res {
self.store(&mut st, d.pts, d.frame, d.draft);
for (pts, frame, draft) in res? {
self.store_finished(&mut st, pts, frame, draft);
}
st.at_or_before(want_pts, true).ok_or_else(|| CodecError::Decode("frame not produced".into()))
}
@ -900,6 +975,77 @@ mod tests {
assert!(s.decodes.load(Ordering::Relaxed) <= 60 + 24 + 1);
}
/// A decoder that, like a slice-threaded one, splits each picture's work over rayon.
struct ParDec(Box<dyn VideoDecoder>);
impl VideoDecoder for ParDec {
fn decode(&mut self, sample: &[u8], pts: i64) -> crate::Result<Vec<DecodedFrame>> {
use rayon::prelude::*;
let rows: u64 = (0..64u32)
.into_par_iter()
.with_max_len(1)
.map(|r| (0..20_000u64).fold(u64::from(r), |a, x| a.wrapping_mul(31).wrapping_add(x)))
.reduce(|| 0, u64::wrapping_add);
std::hint::black_box(rows);
self.0.decode(sample, pts)
}
fn flush(&mut self) -> Vec<DecodedFrame> {
self.0.flush()
}
fn reset(&mut self) {
self.0.reset()
}
fn name(&self) -> &str {
"par"
}
}
struct ParSamples(Samples);
impl VideoSamples for ParSamples {
fn count(&self) -> usize {
self.0.count()
}
fn pts(&self, i: usize) -> i64 {
self.0.pts(i)
}
fn sync_before(&self, i: usize) -> usize {
self.0.sync_before(i)
}
fn sample_at(&self, t: i64) -> Option<usize> {
self.0.sample_at(t)
}
fn read(&self, i: usize) -> crate::Result<Vec<u8>> {
self.0.read(i)
}
fn make_decoder(&self) -> crate::Result<Box<dyn VideoDecoder>> {
Ok(Box::new(ParDec(self.0.make_decoder()?)))
}
}
/// #480: render jobs on rayon asking one source for frames while its decoding (a decoder's
/// slices, rotating the frames) runs on rayon too. A worker that took part of that work and,
/// waiting inside it, picked up another frame request blocked on the cache's lock for good:
/// an export of rotated iPhone clips stopped making frames.
#[test]
fn rayon_frame_requests_while_decoding_on_rayon_never_deadlock() {
use rayon::prelude::*;
let s = Arc::new(ParSamples(samples(240, 30, 4, false)));
let (tx, rx) = std::sync::mpsc::channel();
std::thread::spawn(move || {
for round in 0..40usize {
// export-like: many frame workers, frames of several GOPs of a fresh source at once
let c = GopCache::new(None).with_rotation(1);
(0..96usize).into_par_iter().with_max_len(1).for_each(|k| {
let i = (k * 37 + round * 53) % 240;
assert_eq!(index_of(&c.frame(&*s, i as i64 * 1000).expect("frame")), i);
});
}
let _ = tx.send(());
});
assert!(rx.recv_timeout(Duration::from_secs(120)).is_ok(), "frame requests deadlocked");
}
#[test]
fn running_decoder_continues_through_a_nearby_sync_sample() {
let s = samples(200, 30, 0, false);

54
crates/media/src/cancel.rs
View File

@ -96,10 +96,64 @@ pub fn cancelled() -> bool {
CURRENT.with(|c| c.borrow().as_ref().is_some_and(|f| f.load(Ordering::Relaxed)))
}
/// The hints of the work running on this thread ([`with_cancel`], [`with_catch_up`],
/// [`with_draft`], [`with_background`]), to carry them to another thread doing part of that work.
#[derive(Clone, Default)]
pub struct Context {
cancel: Option<Arc<AtomicBool>>,
catch_up: Option<Tick>,
draft: bool,
background: bool,
}
impl Context {
/// This thread's hints.
pub fn current() -> Self {
Context { cancel: CURRENT.with(|c| c.borrow().clone()), catch_up: catch_up(), draft: draft(), background: background() }
}
/// Run `f` with these hints, restoring this thread's however `f` ends (a pool thread must
/// not keep another job's hints after a panic).
pub fn run<R>(self, f: impl FnOnce() -> R) -> R {
struct Restore(Context);
impl Drop for Restore {
fn drop(&mut self) {
let prev = std::mem::take(&mut self.0);
CURRENT.with(|c| *c.borrow_mut() = prev.cancel);
CATCH_UP.with(|c| c.set(prev.catch_up));
DRAFT.with(|c| c.set(prev.draft));
BACKGROUND.with(|c| c.set(prev.background));
}
}
let _restore = Restore(Context::current());
CURRENT.with(|c| *c.borrow_mut() = self.cancel);
CATCH_UP.with(|c| c.set(self.catch_up));
DRAFT.with(|c| c.set(self.draft));
BACKGROUND.with(|c| c.set(self.background));
f()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn context_carries_the_hints_to_another_thread_and_restores_its_own() {
let flag = Arc::new(AtomicBool::new(true));
let ctx = with_cancel(&flag, || with_draft(true, || with_background(true, || with_catch_up(Some(Tick(5)), Context::current))));
std::thread::scope(|s| {
s.spawn(|| {
let seen = with_draft(false, || ctx.clone().run(|| (cancelled(), draft(), background(), catch_up())));
assert_eq!(seen, (true, true, true, Some(Tick(5))));
// a panic inside leaves this thread's own hints in place
let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| ctx.clone().run(|| panic!("job"))));
assert!(r.is_err());
assert_eq!((cancelled(), draft(), background(), catch_up()), (false, false, false, None));
});
});
}
#[test]
fn flag_is_scoped_to_the_closure() {
let flag = Arc::new(AtomicBool::new(false));