From 5b61f8c46b9963414da6d9203a707c081a038d21 Mon Sep 17 00:00:00 2001 From: EduardoAndradeMachado <119344097+EduardoAndradeMachado@users.noreply.github.com> Date: Fri, 23 Oct 2026 21:32:47 -0300 Subject: [PATCH 1/3] engine, codecs: open an item's media source once under parallel first requests MediaPool::resolve read the cache, opened the file on a miss and only then stored the source, so the parallel frame workers of an export, which ask for a clip's first frames at the same moment, each opened the file: one source, one GOP cache and one hardware decoder per worker, each walking the GOP for a single frame and then dropped. perf.stats showed one hardware session and one seek per renderer for every clip an export entered. The pool now holds a lock per item while that item is being opened: the first worker opens, the others wait, re-check the cache and share the source. Items already in the pool take the read-locked path as before, different items still open in parallel, and the same gate covers proxies. A proxy that fails to open is not tried again for that item for a second, so the requests waiting on the gate fall back to the original instead of each paying for the failure in turn. The open runs on a pool of its own (gop::isolated_open, beside the decode pool of #480): an opener that decodes as it opens (an EXR still) starts rayon work, and a frame worker waiting inside that work on its own pool could pick up another frame job of the batch, which would block on the gate it holds. Without that pool (and on the web build) the open runs on the asking thread without the gate, as before, and keeps the thread's deferred-read mark. On a 20 s GPU export of a 1080p H.264 timeline (3 media items, 600 frames, NVENC): hardware sessions 10 -> 3, seeks 11 -> 4-6, samples decoded 1436 -> 926-1173, export 7.6-8.1 s -> 7.1-7.3 s; output byte-identical. --- Cargo.lock | 1 + crates/codecs/src/gop.rs | 121 +++++++++--- crates/engine/Cargo.toml | 1 + crates/engine/src/media_pool.rs | 340 ++++++++++++++++++++++++++++++-- 4 files changed, 419 insertions(+), 44 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 9ff01577bf..d3494e62bd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1943,6 +1943,7 @@ dependencies = [ "filmcraft-tts", "image", "log", + "rayon", "serde", "serde_json", "thiserror 2.0.21", diff --git a/crates/codecs/src/gop.rs b/crates/codecs/src/gop.rs index bcfd4f7ec9..9d6cee1e1d 100644 --- a/crates/codecs/src/gop.rs +++ b/crates/codecs/src/gop.rs @@ -456,15 +456,27 @@ 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. +/// A pool of one thread per core named `-`; `None` where none can be made. +#[cfg(not(target_arch = "wasm32"))] +fn pool_per_core(name: &'static str) -> Option { + let n = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1); + rayon::ThreadPoolBuilder::new().num_threads(n).thread_name(move |i| format!("{name}-{i}")).build().ok() +} + +/// Threads that run the caches' decoding (see [`isolated`]). #[cfg(not(target_arch = "wasm32"))] fn decode_pool() -> Option<&'static rayon::ThreadPool> { static POOL: std::sync::OnceLock> = 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() + POOL.get_or_init(|| pool_per_core("decode")).as_ref() +} + +/// Threads that open sources (see [`isolated_open`]): apart from [`decode_pool`], so an open +/// that reads a slow file, or decodes a frame through a cache while it opens, never holds up +/// the caches' decoding. +#[cfg(not(target_arch = "wasm32"))] +fn open_pool() -> Option<&'static rayon::ThreadPool> { + static POOL: std::sync::OnceLock> = std::sync::OnceLock::new(); + POOL.get_or_init(|| pool_per_core("open")).as_ref() } /// Run a cache's decoding, its lock held by the caller, on [`decode_pool`], the caller waiting @@ -479,32 +491,85 @@ fn decode_pool() -> Option<&'static rayon::ThreadPool> { /// 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(f: impl FnOnce() -> crate::Result + Send) -> crate::Result { + #[cfg(not(target_arch = "wasm32"))] + { + isolated_on(decode_pool(), "decode-wait", "decoding", f) + } + #[cfg(target_arch = "wasm32")] + { + filmcraft_media::cancel::Context::current().run(f) + } +} + +/// Run a source's opening, a lock on that source held by the caller, on [`open_pool`], the caller +/// waiting without doing other rayon work meanwhile, for the reason [`isolated`] gives: an opener +/// that decodes as it opens (an EXR still) starts rayon work, and the frame worker holding the +/// lock, waiting inside that work, could pick up another frame job that blocks on the lock. The +/// open pool's work never waits on such a lock. An open started from an open (a job of the open +/// pool) runs right there: waiting for the pool from inside it could leave every one of its +/// threads waiting for the others. +pub fn isolated_open(f: impl FnOnce() -> crate::Result + Send) -> crate::Result { + #[cfg(not(target_arch = "wasm32"))] + { + let pool = open_pool(); + if pool.and_then(|p| p.current_thread_index()).is_some() { + return filmcraft_media::cancel::Context::current().run(f); + } + isolated_on(pool, "open-wait", "opening", f) + } + #[cfg(target_arch = "wasm32")] + { + filmcraft_media::cancel::Context::current().run(f) + } +} + +/// Whether [`isolated_open`] runs its job on the open pool. Without it (the pool could not be +/// made; the web build, which has no threads) the job runs on the calling thread. +pub fn opens_apart() -> bool { + #[cfg(not(target_arch = "wasm32"))] + { + open_pool().is_some() + } + #[cfg(target_arch = "wasm32")] + { + false + } +} + +/// Run `f` on `pool` with the caller's cancel hints, the caller waiting on a plain thread named +/// `wait` when it is a rayon worker (so it runs none of its pool's jobs meanwhile); `f` runs on +/// the calling thread when there is no pool or no thread to wait on. `what` names the work in the +/// error for a job that panicked. +#[cfg(not(target_arch = "wasm32"))] +fn isolated_on( + pool: Option<&'static rayon::ThreadPool>, + wait: &'static str, + what: &'static str, + f: impl FnOnce() -> crate::Result + Send, +) -> crate::Result { 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()) + let Some(pool) = pool else { return job() }; + 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(wait.into()).spawn_scoped(scope, || { + let job = slot.lock().unwrap_or_else(PoisonError::into_inner).take(); + job.map(|job| pool.install(job)) }); - 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())), - }, - }; + spawned.map(|h| h.join()) + }); + match waited { + Ok(Ok(Some(r))) => r, + Ok(Err(_)) => Err(CodecError::Decode(format!("internal error while {what}"))), + // no thread to wait on: run here, as before + Ok(Ok(None)) | Err(_) => match slot.into_inner().unwrap_or_else(PoisonError::into_inner) { + Some(job) => job(), + None => Err(CodecError::Decode(format!("internal error while {what}"))), + }, } - job() } impl GopCache { diff --git a/crates/engine/Cargo.toml b/crates/engine/Cargo.toml index d39450794c..2488b63e5f 100644 --- a/crates/engine/Cargo.toml +++ b/crates/engine/Cargo.toml @@ -51,3 +51,4 @@ filmcraft-testkit = { workspace = true } filmcraft-dng = { workspace = true, features = ["writer"] } filmcraft-isobmff = { workspace = true } image = { version = "0.25", default-features = false, features = ["png", "gif"] } +rayon = { workspace = true } diff --git a/crates/engine/src/media_pool.rs b/crates/engine/src/media_pool.rs index 23278317fa..c5f017ccbb 100644 --- a/crates/engine/src/media_pool.rs +++ b/crates/engine/src/media_pool.rs @@ -50,6 +50,10 @@ pub struct MediaPool { /// item → (reference key, source) sources: RwLock>, proxies: RwLock>, + /// Per item: the lock held while its source is being opened (see [`Self::opening`]). + opening: Mutex>>>, + /// Per item: the proxy key that last failed to open, and when (see [`PROXY_RETRY`]). + proxy_failed: Mutex>, offline: Arc>>, use_proxies: AtomicBool, /// Per item: video frames that failed in a row in the monitors (see [`Self::monitor_source`]). @@ -67,6 +71,8 @@ impl Default for MediaPool { Self { sources: RwLock::new(HashMap::new()), proxies: RwLock::new(HashMap::new()), + opening: Mutex::new(HashMap::new()), + proxy_failed: Mutex::new(HashMap::new()), offline: Arc::new(RwLock::new(HashMap::new())), use_proxies: AtomicBool::new(false), failures: Arc::default(), @@ -76,6 +82,24 @@ impl Default for MediaPool { } } +/// How long a proxy that failed to open is not tried again for the same item. The requests that +/// wait for one item's open (an export batch, playback prefetch) take turns: without it, a proxy +/// that takes long to fail (a share that times out) would make each of them pay that time in turn. +const PROXY_RETRY: std::time::Duration = std::time::Duration::from_secs(1); + +/// `open()`, and whether a read in it was deferred ([`filmcraft_media::pending`]), leaving the +/// thread's own mark as it was: where the open runs on the asking thread (the web build), a mark +/// an earlier layer of the same frame set stays, so the frame is still known to be incomplete. +fn open_noting_pending(open: impl FnOnce() -> T) -> (T, bool) { + let before = filmcraft_media::pending::take(); + let opened = open(); + let during = filmcraft_media::pending::take(); + if before { + filmcraft_media::pending::mark(); + } + (opened, during) +} + /// Largest file [`MediaPool::probe_file`] reads whole when no streaming reader takes it (WAV, /// MP3 and other byte-opened formats, or an unsupported container). pub const PROBE_WHOLE_FILE_MAX: u64 = 64 * 1024 * 1024; @@ -177,9 +201,49 @@ impl MediaPool { pub fn remove(&self, item: ItemId) { self.sources.write().unwrap_or_else(|e| e.into_inner()).remove(&item); self.proxies.write().unwrap_or_else(|e| e.into_inner()).remove(&item); + let mut opening = self.opening.lock().unwrap_or_else(|e| e.into_inner()); + if opening.get(&item).is_some_and(|g| Arc::strong_count(g) == 1) { + opening.remove(&item); + } + drop(opening); + self.proxy_failed.lock().unwrap_or_else(|e| e.into_inner()).remove(&item); self.offline.write().unwrap_or_else(|e| e.into_inner()).remove(&item); } + /// The lock a thread holds while it opens `item`'s source (full-resolution or proxy). Workers + /// that ask for an item the pool does not hold yet at the same moment (an export's first + /// frames, each on its own thread) wait here instead of each opening the file, with its own + /// decoder and GOP cache, and then share the one source. `None` when the open would run on + /// the calling thread (see [`Self::open_apart`]): a frame worker waiting inside the opener's + /// rayon work could block on the lock it holds, so such a thread opens without it. + fn opening(&self, item: ItemId) -> Option>> { + filmcraft_codecs::gop::opens_apart().then(|| self.opening.lock().unwrap_or_else(|e| e.into_inner()).entry(item).or_default().clone()) + } + + /// Run `open` (a source's opening, under the item's gate) off the caller's rayon pool. An + /// opener that decodes at open (an EXR still) starts rayon work; a frame worker waiting inside + /// that work would pick up another frame job of its pool, which blocks on the gate the worker + /// holds (#480, [`filmcraft_codecs::gop::isolated_open`]). A read the opener could not serve + /// yet is marked on this thread, as if it had opened here. + fn open_apart(open: impl FnOnce() -> Result + Send) -> Result { + match filmcraft_codecs::gop::isolated_open(|| Ok(open_noting_pending(open))) { + Ok((opened, pending)) => { + if pending { + filmcraft_media::pending::mark(); + } + opened + } + Err(e) => Err(MediaError::Io(format!("the opener stopped: {e}"))), + } + } + + /// The cached full-resolution source of `item` when it was opened under `key` (or without one). + fn cached_under(&self, item: ItemId, key: &str) -> Option { + let sources = self.sources.read().unwrap_or_else(|e| e.into_inner()); + let (k, s) = sources.get(&item)?; + (k.is_empty() || k == key).then(|| s.clone()) + } + /// Record the media and proxy paths of `p`, loaded from the document at `document` (None: one /// without a file), as stored paths: they are opened only as [`crate::doc_path::resolve`] /// allows. A path already recorded keeps the document it came from first. @@ -232,6 +296,8 @@ impl MediaPool { pub fn clear(&self) { self.sources.write().unwrap_or_else(|e| e.into_inner()).clear(); self.proxies.write().unwrap_or_else(|e| e.into_inner()).clear(); + self.opening.lock().unwrap_or_else(|e| e.into_inner()).retain(|_, g| Arc::strong_count(g) > 1); + self.proxy_failed.lock().unwrap_or_else(|e| e.into_inner()).clear(); self.offline.write().unwrap_or_else(|e| e.into_inner()).clear(); } @@ -319,10 +385,13 @@ impl MediaPool { return Some(s); } let key = media_key(m); - if let Some((k, s)) = self.sources.read().unwrap_or_else(|e| e.into_inner()).get(&item) - && (k.is_empty() || *k == key) - { - return Some(s.clone()); + if let Some(s) = self.cached_under(item, &key) { + return Some(s); + } + let gate = self.opening(item); + let _opening = gate.as_ref().map(|g| g.lock().unwrap_or_else(|e| e.into_inner())); + if let Some(s) = self.cached_under(item, &key) { + return Some(s); } let src: SharedSource = match &m.media { MediaRef::Generator(g) => { @@ -334,12 +403,14 @@ impl MediaPool { self.set_offline(item, OfflineReason::MadeOffline, path, "made offline"); Arc::new(SlateSource::new(m.info.clone(), &file_name(path), OfflineReason::MadeOffline)) } - MediaRef::File { path } => match self.follow_media(path).and_then(|file| { - if m.info.kind == filmcraft_media::MediaKind::ImageSequence { - open_image_sequence(&file, services, m.info.frame_rate(), &it.name) - } else { - self.open_file(&file, services) - } + MediaRef::File { path } => match Self::open_apart(|| { + self.follow_media(path).and_then(|file| { + if m.info.kind == filmcraft_media::MediaKind::ImageSequence { + open_image_sequence(&file, services, m.info.frame_rate(), &it.name) + } else { + self.open_file(&file, services) + } + }) }) { Ok(s) => { self.offline.write().unwrap_or_else(|e| e.into_inner()).remove(&item); @@ -367,12 +438,24 @@ impl MediaPool { fn proxy_source(&self, item: ItemId, path: &str, m: &MediaClip, services: &dyn Services) -> Option { let key = format!("proxy:{path}"); - if let Some((k, s)) = self.proxies.read().unwrap_or_else(|e| e.into_inner()).get(&item) - && *k == key - { - return Some(s.clone()); + let cached = || { + let proxies = self.proxies.read().unwrap_or_else(|e| e.into_inner()); + proxies.get(&item).filter(|(k, _)| *k == key).map(|(_, s)| s.clone()) + }; + if let Some(s) = cached() { + return Some(s); } - match self.follow_media(path).and_then(|file| self.open_file(&file, services)) { + let gate = self.opening(item); + let _opening = gate.as_ref().map(|g| g.lock().unwrap_or_else(|e| e.into_inner())); + if let Some(s) = cached() { + return Some(s); + } + let failed_lately = + self.proxy_failed.lock().unwrap_or_else(|e| e.into_inner()).get(&item).is_some_and(|(k, at)| *k == key && at.elapsed() < PROXY_RETRY); + if failed_lately { + return None; + } + match Self::open_apart(|| self.follow_media(path).and_then(|file| self.open_file(&file, services))) { Ok(s) => { // a proxy with fewer audio streams than the original (say one track of an OBS // recording's seven) still stands in for the picture; the audio comes from the @@ -380,19 +463,23 @@ impl MediaPool { let audio_from = if s.info().audio_streams.len() < m.info.audio_streams.len() { log::warn!("proxy has fewer audio streams than the original: {path}; audio from full-resolution media"); match &m.media { - MediaRef::File { path: original } if !m.offline => self.follow_media(original).and_then(|file| self.open_file(&file, services)).ok(), + MediaRef::File { path: original } if !m.offline => { + Self::open_apart(|| self.follow_media(original).and_then(|file| self.open_file(&file, services))).ok() + } _ => None, } } else { None }; let src: SharedSource = Arc::new(ProxySource { proxy: s, info: m.info.clone(), audio_from }); + self.proxy_failed.lock().unwrap_or_else(|e| e.into_inner()).remove(&item); self.proxies.write().unwrap_or_else(|e| e.into_inner()).insert(item, (key, src.clone())); Some(src) } Err(e) => { // a missing proxy falls back to the full-resolution media log::warn!("proxy offline: {path}: {e}"); + self.proxy_failed.lock().unwrap_or_else(|e| e.into_inner()).insert(item, (key, web_time::Instant::now())); None } } @@ -715,6 +802,227 @@ mod tests { std::fs::remove_dir_all(&dir).unwrap(); } + /// A host whose reader takes 50 ms to open any path (paths starting with `missing` fail after + /// that), counting the opens, how many ran at once and the threads they ran on. + struct SlowReader { + bytes: Arc<[u8]>, + opens: std::sync::atomic::AtomicUsize, + opening: std::sync::atomic::AtomicUsize, + most_at_once: std::sync::atomic::AtomicUsize, + /// Wait (up to 10 s) for a second open in progress before finishing. + wait_for_company: bool, + /// Run a pending rayon job of the opening thread's pool while opening, as an opener that + /// decodes on rayon (an EXR still) does while it waits for its own work. + yield_to_rayon: bool, + threads: Mutex>, + } + impl SlowReader { + fn new(bytes: Vec) -> Self { + Self { + bytes: bytes.into(), + opens: Default::default(), + opening: Default::default(), + most_at_once: Default::default(), + wait_for_company: false, + yield_to_rayon: false, + threads: Default::default(), + } + } + } + impl Services for SlowReader { + fn read_file(&self, path: &str) -> std::io::Result> { + Err(std::io::Error::other(format!("{path}: read whole"))) + } + fn write_file(&self, _path: &str, _data: &[u8]) -> std::io::Result<()> { + Ok(()) + } + fn reader(&self, path: &str) -> Option> { + self.opens.fetch_add(1, Ordering::SeqCst); + let at_once = self.opening.fetch_add(1, Ordering::SeqCst) + 1; + self.most_at_once.fetch_max(at_once, Ordering::SeqCst); + self.threads.lock().unwrap().push(std::thread::current().name().unwrap_or("").to_string()); + if self.yield_to_rayon { + rayon::yield_now(); + } + let started = web_time::Instant::now(); + while self.wait_for_company && self.opening.load(Ordering::SeqCst) < 2 && started.elapsed() < std::time::Duration::from_secs(10) { + std::thread::sleep(std::time::Duration::from_millis(1)); + } + self.most_at_once.fetch_max(self.opening.load(Ordering::SeqCst), Ordering::SeqCst); + std::thread::sleep(std::time::Duration::from_millis(50)); + self.opening.fetch_sub(1, Ordering::SeqCst); + if path.starts_with("missing") { + return Some(Err(std::io::Error::new(std::io::ErrorKind::NotFound, path.to_string()))); + } + Some(Ok(Arc::new(filmcraft_media::reader::MemReader(self.bytes.clone())))) + } + } + + /// A project with a WAV item per `(path, proxy path)`, and the reader that serves them all. + fn wav_items(items: &[(&str, Option<&str>)]) -> (Project, Vec, SlowReader) { + let services = SlowReader::new(filmcraft_media::wav::write_wav16(&[0.0, 0.5, -0.5, 0.25], 2, 48_000)); + let info = MediaPool::default().open_file("a.wav", &services).unwrap().info().clone(); + services.opens.store(0, Ordering::SeqCst); + services.threads.lock().unwrap().clear(); + let mut project = Project::new("race"); + let ids = items + .iter() + .map(|(path, proxy)| { + project.add_item( + path, + filmcraft_project::Label::Iris, + ItemKind::Media(MediaClip { + media: MediaRef::File { path: path.to_string() }, + info: info.clone(), + interpret: Default::default(), + mark_in: None, + mark_out: None, + markers: vec![], + offline: false, + proxy: proxy.map(|p| MediaRef::File { path: p.to_string() }), + identity: None, + }), + None, + ) + }) + .collect(); + (project, ids, services) + } + + /// `requests` frame jobs on a rayon pool of `workers` threads named `frame-`, as an export + /// renders a batch, each resolving `item`. Fails when they have not all finished within 30 s + /// (workers stuck for good are left behind). + fn ask_from_frame_workers( + pool: &Arc, + project: &Arc, + item: ItemId, + services: &Arc, + workers: usize, + requests: usize, + ) -> Vec { + use rayon::prelude::*; + let (pool, project, services) = (pool.clone(), project.clone(), services.clone()); + let (tx, rx) = std::sync::mpsc::channel(); + let asking = std::thread::spawn(move || { + let frames = rayon::ThreadPoolBuilder::new().num_threads(workers).thread_name(|n| format!("frame-{n}")).build().expect("a frame pool"); + let sources: Vec = frames.install(|| { + (0..requests).into_par_iter().with_max_len(1).map(|_| pool.source_for(&project, item, &*services).expect("the item opens")).collect() + }); + let _ = tx.send(sources); + }); + match rx.recv_timeout(std::time::Duration::from_secs(30)) { + Ok(sources) => sources, + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => panic!("the frame workers are stuck: deadlock"), + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => match asking.join() { + Err(panic) => std::panic::resume_unwind(panic), + Ok(()) => panic!("the frame workers ended without their sources"), + }, + } + } + + /// An export's frame workers ask for a clip's first frames at the same moment, before the + /// pool holds its source: the file is opened once and every worker gets that one source. + #[test] + fn parallel_first_requests_open_an_item_once() { + let (project, ids, services) = wav_items(&[("a.wav", None)]); + let (project, services, pool) = (Arc::new(project), Arc::new(services), Arc::new(MediaPool::default())); + let sources = ask_from_frame_workers(&pool, &project, ids[0], &services, 8, 8); + assert_eq!(services.opens.load(Ordering::SeqCst), 1, "one open per item, however many ask at once"); + assert!(sources.iter().all(|s| Arc::ptr_eq(s, &sources[0])), "every worker shares the one source"); + assert_eq!(pool.open_sources(), 1); + } + + /// The same for an item read through its proxy. + #[test] + fn parallel_first_requests_open_a_proxy_once() { + let (project, ids, services) = wav_items(&[("a.wav", Some("a-proxy.wav"))]); + let (project, services, pool) = (Arc::new(project), Arc::new(services), Arc::new(MediaPool::default())); + pool.set_use_proxies(true); + let sources = ask_from_frame_workers(&pool, &project, ids[0], &services, 8, 8); + assert_eq!(services.opens.load(Ordering::SeqCst), 1, "one open of the proxy, however many ask at once"); + assert!(sources.iter().all(|s| Arc::ptr_eq(s, &sources[0])), "every worker shares the one proxy source"); + assert_eq!(pool.open_sources(), 1, "the proxy, not the full-resolution file"); + assert!(pool.cached(ids[0]).is_none(), "the full-resolution file was not opened"); + } + + /// A proxy that fails to open is tried once by the requests waiting for it, which then read + /// the full-resolution file: they do not each pay for the failure in turn. + #[test] + fn parallel_first_requests_try_a_missing_proxy_once() { + let (project, ids, services) = wav_items(&[("a.wav", Some("missing-proxy.wav"))]); + let (project, services, pool) = (Arc::new(project), Arc::new(services), Arc::new(MediaPool::default())); + pool.set_use_proxies(true); + let sources = ask_from_frame_workers(&pool, &project, ids[0], &services, 8, 8); + assert_eq!(services.opens.load(Ordering::SeqCst), 2, "the proxy once, then the full-resolution file once"); + assert!(sources.iter().all(|s| Arc::ptr_eq(s, &sources[0])), "every worker shares the full-resolution source"); + } + + /// Waiting is per item: two items asked for at once open side by side. + #[test] + fn different_items_open_in_parallel() { + if std::thread::available_parallelism().map_or(1, |n| n.get()) < 2 { + eprintln!("skipped: one core, the open pool has a single thread"); + return; + } + let (project, ids, mut services) = wav_items(&[("a.wav", None), ("b.wav", None)]); + services.wait_for_company = true; + let pool = MediaPool::default(); + let start = std::sync::Barrier::new(2); + std::thread::scope(|scope| { + for item in &ids { + scope.spawn(|| { + start.wait(); + pool.full_res_source(&project, *item, &services).expect("the item opens"); + }); + } + }); + assert_eq!(services.opens.load(Ordering::SeqCst), 2); + assert_eq!(services.most_at_once.load(Ordering::SeqCst), 2, "the two opens overlapped"); + } + + /// Opens run on the open pool, never on the frame workers that ask for them. + #[test] + fn opens_run_on_the_open_pool() { + let (project, ids, services) = wav_items(&[("a.wav", Some("a-proxy.wav"))]); + let (project, services, pool) = (Arc::new(project), Arc::new(services), Arc::new(MediaPool::default())); + ask_from_frame_workers(&pool, &project, ids[0], &services, 4, 4); + pool.set_use_proxies(true); + ask_from_frame_workers(&pool, &project, ids[0], &services, 4, 4); + let threads = services.threads.lock().unwrap().clone(); + assert_eq!(threads.len(), 2, "one open of the file, one of the proxy"); + assert!(threads.iter().all(|t| t.starts_with("open-")), "opened on {threads:?}"); + } + + /// A frame worker holds an item's gate while its open runs rayon work (an EXR still decodes as + /// it opens). On the frame worker's own pool, that work would let the worker pick up another + /// frame job of the batch, which waits for the gate the worker holds, for good. + #[test] + fn an_opener_running_rayon_work_never_deadlocks_the_frame_workers() { + let (project, ids, mut services) = wav_items(&[("a.wav", None)]); + services.yield_to_rayon = true; + let (project, services, pool) = (Arc::new(project), Arc::new(services), Arc::new(MediaPool::default())); + let sources = ask_from_frame_workers(&pool, &project, ids[0], &services, 2, 32); + assert_eq!(sources.len(), 32); + assert_eq!(services.opens.load(Ordering::SeqCst), 1); + } + + /// Opening an item keeps a deferred-read mark the thread already had (an earlier layer of the + /// frame whose data is still being fetched), and reports one the open itself made. Where the + /// open runs on the asking thread (the web build) the frame would otherwise be encoded without + /// that layer. + #[test] + fn opening_keeps_the_threads_deferred_read_mark() { + let _ = filmcraft_media::pending::take(); + filmcraft_media::pending::mark(); + let ((), during) = open_noting_pending(|| ()); + assert!(!during, "the open deferred nothing"); + assert!(filmcraft_media::pending::take(), "the earlier mark stays"); + + let ((), during) = open_noting_pending(filmcraft_media::pending::mark); + assert!(during, "the open's own deferred read is reported"); + assert!(!filmcraft_media::pending::take(), "and left to the caller to mark"); + } + /// A file whose first frame decodes and whose later frames need a decoder FilmCraft lacks. struct PartlyDecodable(MediaInfo); impl MediaSource for PartlyDecodable { From e0ec58e3c9b2d54a237e28cfa5f30e83696a8414 Mon Sep 17 00:00:00 2001 From: EduardoAndradeMachado <119344097+EduardoAndradeMachado@users.noreply.github.com> Date: Fri, 23 Oct 2026 21:32:47 -0300 Subject: [PATCH 2/3] docs: one media source per item under parallel export requests performance.md: the race, the gate and its open pool, and the before/after counters and time of GPU exports. architecture.md: the sources paragraph. --- docs/architecture.md | 5 +++- docs/performance.md | 69 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 73 insertions(+), 1 deletion(-) diff --git a/docs/architecture.md b/docs/architecture.md index f8995621b1..1acd444aa3 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -225,7 +225,10 @@ file ──► codecs (MP4/MOV, MKV, audio) demux + decode, GOP-aware see - **Sources.** `media::MediaSource` yields `video_frame(FrameRequest)` and `audio(start, frames, rate)` in media time. Sources are `Send + Sync` and shared by monitors, thumbnails, playback and export. The engine's `MediaPool` creates one per project item, lazily, - through registered openers (`codecs::openers()`: MP4/MOV, MKV/WebM, audio files). + through registered openers (`codecs::openers()`: MP4/MOV, MKV/WebM, audio files); threads that + ask for an item not opened yet at the same time (an export's parallel frame workers) wait for the + one open instead of each opening a source of their own. The open runs on a pool of its own + (`codecs::gop::isolated_open`), not on the asking thread's rayon pool nor on the decode pool. - **Pixel aspect ratio.** Demuxers report it in `VideoStreamInfo::par` (MP4/MOV `pasp`, Matroska display size, MXF aspect ratio, MPEG sequence header / VUI); Interpret Footage's `Interpretation::par` overrides it (`clip.interpretFootage {pixelAspect}`), and diff --git a/docs/performance.md b/docs/performance.md index 65310442ad..ff48c05bfa 100644 --- a/docs/performance.md +++ b/docs/performance.md @@ -344,6 +344,75 @@ the encoder's hand-off). Only NVENC, whose encode is light on the CPU, comes clo encoder waits for the renderer), as expected. 1080p exports use 70-160 MB more (the second batch of 16 frames); 4K exports use less than before because the batch is smaller (8 for SDR, 2 for HDR instead of 16). +## Results (export: a clip's source opened once under parallel first requests, before → after) + +`MediaPool::resolve` looked the item up in the cache, opened the file on a miss and then stored the +source. The frames of an export batch are rendered in parallel (one rayon worker per frame, up to +four GPU renderers), so a clip's first frames reached the pool at the same moment, every worker +missed and every worker opened the file: one source, one GOP cache and one decoder each, which +each walked the GOP from the sync sample to deliver a single frame. Only the last source stored +survived; the others were dropped with their decoders. On a timeline of H.264 camera clips this +showed in `perf.stats` as one hardware session and one seek per renderer for each clip the export +entered, not one per clip, and as samples decoded far above the frames the export asked for. + +Now the pool holds a lock per item while a source is being opened (`MediaPool::opening`): the first +worker opens the file, the others wait on that lock, find the source stored and share it. An item +already in the pool takes the read-locked lookup as before, different items still open in +parallel, and the same gate covers proxies. A proxy that fails to open is not tried again for that +item for a second (`PROXY_RETRY`), so the requests waiting on the gate fall back to the +full-resolution file instead of each paying for the failure in turn. The open itself runs off the +asking thread, on a pool of its own (`codecs::gop::isolated_open`, one thread per core, beside the decode +pool of #480): an opener that decodes at open (an EXR still) starts rayon work, and a frame worker +waiting inside that work on its own pool could pick up another frame job of the export, which +would block on the gate the worker holds. Opens do not run on the decode pool, so a slow file +never holds up the caches' decoding. Where that pool cannot be made, and on the web build, which +has no threads, the open runs on the asking thread without the gate, as before; a deferred-read +mark (`media::pending`) the thread already had stays set across it. The tests +(`parallel_first_requests_open_an_item_once` and its neighbours) run the requests as frame jobs on +a rayon pool against a host whose reader takes 50 ms and records the opens and the threads they +run on: one open per item and per proxy (8 before), a missing proxy tried once (9 before), two +items open side by side, opens on the open pool, an opener that runs a pending rayon job while it +opens finishes the batch (stuck for good with the open run on the frame worker), and an open +keeps the thread's deferred-read mark. + +Measured (Xeon E5-2680 v4 14C/28T, RTX 5060, Windows 11, release CLI of this change against +`main` at the same base, `gpuRendering` and `hardwareEncoding` auto, 1080p30 H.264 + AAC output, +a timeline of 1080p H.264 camera clips at 30 and 59.94 fps with 0.2 s dip-to-black transitions, +`perf.stats decode` after the export, two alternated rounds on an otherwise idle machine; the +counters vary with scheduling between runs, so the ranges over the rounds are given): + +| Range | Build | HW sessions | seeks | samples decoded | `decodeMs` | +|---|---|---|---|---|---| +| 5 s (1 media item, 150 frames) | before | 4 | 5 | 621 | 2904–2942 | +| 5 s | after | 1 | 2 | 348–349 | 682–694 | +| 20 s (3 media items, 600 frames) | before | 10 | 11 | 1436–1438 | 5501–5531 | +| 20 s | after | 3 | 4–6 | 926–1173 | 1973–2342 | + +`RAYON_NUM_THREADS=1` (one frame at a time, which cannot race) gave the 5 s range on `main` 1 +session, 2 seeks and 348 samples: the after row. After the change the export opens exactly one +source per media item (3 for 3 items) and seeks once per clip entered (4 clips in the 20 s range); +the round with 6 seeks had a late frame request behind the one decoder's position, which costs a +seek whatever the number of sources. The samples decoded above the frames requested that remain +(926 for 600 frames) are the GOP walks at the clips' in points and the 59.94 fps sources decoded +for 30 fps output; they are not duplicate work. The output files are byte-identical before and +after (same frames, same encoder input order), checked with `cmp` on every export below. + +Time (CLI process from start to exit, the export's own `seconds`, CPU time of the process, GPU +utilisation averaged over the run from `nvidia-smi dmon`): + +| Range | Build | wall | export | process CPU | GPU | +|---|---|---|---|---|---| +| 5 s | before | 4.6 s, 4.0 s | 2.8 s, 2.7 s | 10.4 s, 9.9 s | 11 %, 24 % | +| 5 s | after | 3.9 s, 3.5 s | 2.2 s, 2.2 s | 9.4 s, 10.2 s | 13 %, 26 % | +| 20 s | before | 9.4 s, 9.0 s | 8.1 s, 7.6 s | 35.5 s, 38.7 s | 33 %, 30 % | +| 20 s | after | 8.7 s, 8.5 s | 7.3 s, 7.1 s | 35.4 s, 36.0 s | 30 %, 27 % | + +The export is 20 % shorter over 5 s (one clip entered, where the duplicate GOP walks were most of +the decode) and 8 % over 20 s. The duplicate decoders ran on the hardware decoder in parallel with +the renderers, so removing them shortens the decode work (`decodeMs` −60 to −77 %) more than the +export itself; process CPU time is the same within noise (rendering and the NVENC hand-off +dominate it, and the duplicate decoding was mostly hardware time). + ## Results (HW3: VideoToolbox hardware H.265 (HEVC) encoding, against hardware H.264) Apple M1 (8 cores, 16 GB), 2026-10-07, single runs on an otherwise idle machine, release build. Both From 3049ee0eea9a7df8684900f074d427c1e1e2ebc3 Mon Sep 17 00:00:00 2001 From: EduardoAndradeMachado <119344097+EduardoAndradeMachado@users.noreply.github.com> Date: Fri, 23 Oct 2026 21:44:13 -0300 Subject: [PATCH 3/3] =?UTF-8?q?docs:=20decodeMs=20reduction=20starts=20at?= =?UTF-8?q?=20=E2=88=9258=20%=20(review)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The lowest round of the 20 s export went from 5531 to 2342 ms, −57.7 %. --- docs/performance.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/performance.md b/docs/performance.md index ff48c05bfa..2f323375b1 100644 --- a/docs/performance.md +++ b/docs/performance.md @@ -409,7 +409,7 @@ utilisation averaged over the run from `nvidia-smi dmon`): The export is 20 % shorter over 5 s (one clip entered, where the duplicate GOP walks were most of the decode) and 8 % over 20 s. The duplicate decoders ran on the hardware decoder in parallel with -the renderers, so removing them shortens the decode work (`decodeMs` −60 to −77 %) more than the +the renderers, so removing them shortens the decode work (`decodeMs` −58 to −77 %) more than the export itself; process CPU time is the same within noise (rendering and the NVENC hand-off dominate it, and the duplicate decoding was mostly hardware time).