diff --git a/crates/rendering/src/decoder/mod.rs b/crates/rendering/src/decoder/mod.rs index 419dc595644..186ad71f131 100644 --- a/crates/rendering/src/decoder/mod.rs +++ b/crates/rendering/src/decoder/mod.rs @@ -727,6 +727,10 @@ pub(crate) const EXPORT_STALL_LIMIT: Duration = Duration::from_secs(120); enum FrameDeadline { /// Playback: a missed deadline returns no frame so the caller can move on. Drop, + /// The first frame after a seek or for a paused preview. Nothing asks for + /// it again, so on macOS a missed deadline switches to software decoding + /// at once, which then answers it. + Initial, /// Offline export: a missed deadline keeps waiting for the same reply. Wait, } @@ -754,6 +758,60 @@ struct MacOsDecoderFallback { path: PathBuf, fps: u32, decoder: tokio::sync::OnceCell>, + stall: Mutex, +} + +#[cfg(target_os = "macos")] +#[derive(Default)] +struct PrimaryStall { + timed_out_rounds: u32, + last_timeout: Option, +} + +#[cfg(target_os = "macos")] +impl MacOsDecoderFallback { + /// A single playback timeout is a busy machine, not a dead reader: only a + /// reader that misses this many playback rounds in a row without + /// delivering any frame is replaced by software decoding. + const MAX_TIMED_OUT_ROUNDS: u32 = 3; + + fn new(name: &'static str, path: PathBuf, fps: u32) -> Self { + Self { + name, + path, + fps, + decoder: tokio::sync::OnceCell::new(), + stall: Mutex::new(PrimaryStall::default()), + } + } + + fn primary_answered(&self) { + *self + .stall + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = PrimaryStall::default(); + } + + /// Playback keeps many requests in flight, and one slow decode times them + /// out together; only a request sent after the previous timeout starts a + /// new round. + fn primary_stalled(&self, sent_at: tokio::time::Instant) -> bool { + let mut stall = self + .stall + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if stall.last_timeout.is_none_or(|last| sent_at >= last) { + stall.timed_out_rounds += 1; + stall.last_timeout = Some(tokio::time::Instant::now()); + } + stall.timed_out_rounds >= Self::MAX_TIMED_OUT_ROUNDS + } +} + +enum FrameReply { + Frame(DecodedFrame), + NoFrame, + TimedOut, } impl AsyncVideoDecoderHandle { @@ -789,7 +847,7 @@ impl AsyncVideoDecoderHandle { Self::INITIAL_SEEK_TIMEOUT_MS, self.max_fallback_distance .min(Self::INITIAL_MAX_FALLBACK_DISTANCE), - FrameDeadline::Drop, + FrameDeadline::Initial, ) .await .ok() @@ -836,12 +894,24 @@ impl AsyncVideoDecoderHandle { Err(_) => Ok(None), }; } - let primary = self - .request_frame(time, timeout_ms, max_fallback_distance, deadline) - .await; - if let Ok(Some(frame)) = primary { - return Ok(Some(frame)); - } + let sent_at = tokio::time::Instant::now(); + let (fallback_reason, primary) = match self + .send_frame_request(time, timeout_ms, max_fallback_distance, deadline) + .await + { + Ok(FrameReply::Frame(frame)) => { + fallback.primary_answered(); + return Ok(Some(frame)); + } + Ok(FrameReply::TimedOut) + if deadline == FrameDeadline::Drop && !fallback.primary_stalled(sent_at) => + { + return Ok(None); + } + Ok(FrameReply::TimedOut) => ("AVAssetReader stopped answering", Ok(None)), + Ok(FrameReply::NoFrame) => ("AVAssetReader returned no frame", Ok(None)), + Err(stalled) => ("AVAssetReader stalled", Err(stalled)), + }; let result = fallback .decoder .get_or_init(|| async { @@ -858,8 +928,7 @@ impl AsyncVideoDecoderHandle { false, ) .await?; - decoder.status.fallback_reason = - Some("AVAssetReader returned no frame".to_string()); + decoder.status.fallback_reason = Some(fallback_reason.to_string()); Ok(decoder) }) .await; @@ -886,6 +955,21 @@ impl AsyncVideoDecoderHandle { max_fallback_distance: u32, deadline: FrameDeadline, ) -> Result, DecodeStalled> { + self.send_frame_request(time, timeout_ms, max_fallback_distance, deadline) + .await + .map(|reply| match reply { + FrameReply::Frame(frame) => Some(frame), + FrameReply::NoFrame | FrameReply::TimedOut => None, + }) + } + + async fn send_frame_request( + &self, + time: f32, + timeout_ms: u64, + max_fallback_distance: u32, + deadline: FrameDeadline, + ) -> Result { let (tx, mut rx) = tokio::sync::oneshot::channel(); let adjusted_time = self.get_time(time); @@ -902,13 +986,14 @@ impl AsyncVideoDecoderHandle { time = adjusted_time, "decoder thread is gone; frame request dropped" ); - return Ok(None); + return Ok(FrameReply::NoFrame); } let started = tokio::time::Instant::now(); loop { match tokio::time::timeout(Duration::from_millis(timeout_ms), &mut rx).await { - Ok(result) => return Ok(result.ok()), + Ok(Ok(frame)) => return Ok(FrameReply::Frame(frame)), + Ok(Err(_)) => return Ok(FrameReply::NoFrame), Err(_) if deadline == FrameDeadline::Wait => { let waited = started.elapsed(); if waited >= EXPORT_STALL_LIMIT { @@ -932,7 +1017,7 @@ impl AsyncVideoDecoderHandle { timeout_ms = timeout_ms, "Frame decode request timed out" ); - return Ok(None); + return Ok(FrameReply::TimedOut); } } } @@ -1515,12 +1600,11 @@ pub async fn spawn_decoder( status, max_fallback_distance: DEFAULT_MAX_FALLBACK_DISTANCE, #[cfg(target_os = "macos")] - fallback: Some(Arc::new(MacOsDecoderFallback { + fallback: Some(Arc::new(MacOsDecoderFallback::new( name, - path: path.clone(), + path.clone(), fps, - decoder: tokio::sync::OnceCell::new(), - })), + ))), #[cfg(target_os = "macos")] sequential_access: Some(sequential_access), }) @@ -2448,15 +2532,22 @@ mod managed_worker_tests { fallback_reason: None, }, max_fallback_distance: 2, - fallback: Some(Arc::new(MacOsDecoderFallback { - name: "screen", - path: directory.path().to_path_buf(), - fps: 30, - decoder: tokio::sync::OnceCell::new(), - })), + fallback: Some(Arc::new(MacOsDecoderFallback::new( + "screen", + directory.path().to_path_buf(), + 30, + ))), sequential_access: None, }; let cloned = decoder.clone(); + for _ in 1..MacOsDecoderFallback::MAX_TIMED_OUT_ROUNDS { + let (first, second) = tokio::join!( + decoder.get_frame_with_timeout(0.0, 200, 2, FrameDeadline::Drop), + cloned.get_frame_with_timeout(0.0, 200, 2, FrameDeadline::Drop) + ); + assert!(matches!((first, second), (Ok(None), Ok(None)))); + assert_eq!(decoder.decoder_type(), DecoderType::AVAssetReader); + } let (first, second) = tokio::join!( decoder.get_frame_with_timeout(0.0, 200, 2, FrameDeadline::Drop), cloned.get_frame_with_timeout(0.0, 200, 2, FrameDeadline::Drop) @@ -2468,14 +2559,20 @@ mod managed_worker_tests { ); assert_eq!(decoder.decoder_type(), DecoderType::FFmpegSoftware); assert_eq!(cloned.decoder_type(), DecoderType::FFmpegSoftware); - assert!(decoder.fallback_reason().is_some()); + assert_eq!( + decoder.fallback_reason(), + Some("AVAssetReader stopped answering") + ); let (reference, join) = ordinary_worker(directory.path().to_path_buf(), 0.25, false).await; assert_eq!( frame_pixels(&first), frame_pixels(&reference.get_frame_initial(0.0).await.unwrap()) ); let pending_before = receiver.try_iter().count(); - assert_eq!(pending_before, 2); + assert_eq!( + pending_before, + 2 * MacOsDecoderFallback::MAX_TIMED_OUT_ROUNDS as usize + ); assert!(decoder.get_frame_initial(0.5).await.is_some()); assert_eq!(receiver.try_iter().count(), 0); drop(reference); diff --git a/crates/rendering/src/decoder/model/mod.rs b/crates/rendering/src/decoder/model/mod.rs index 291012dbdd1..d3be6de1c6d 100644 --- a/crates/rendering/src/decoder/model/mod.rs +++ b/crates/rendering/src/decoder/model/mod.rs @@ -1094,6 +1094,7 @@ fn run_suite(decoder: Decoder) { fn avassetreader_answers_match_the_reference_model() { // Engine verify runs only this test of the module on macOS. avassetreader_answers_stay_exact_when_its_reader_fails(); + avassetreader_switches_to_software_only_when_its_reader_stops_answering(); run_suite(Decoder::Avf); avassetreader_scrubbing_backwards_reuses_the_frames_below_each_seek(); avassetreader_stops_reading_only_once_every_request_is_abandoned(); @@ -1156,6 +1157,181 @@ fn avassetreader_answers_stay_exact_when_its_reader_fails() { } } +/// A busy machine can miss one playback deadline without anything being +/// wrong with the reader: that request gets no frame, and hardware decoding +/// stays. A reader that answers without a frame, misses every playback +/// deadline round after round, or misses the deadline of a first frame that +/// nothing will ask for again is replaced by software decoding. +#[cfg(target_os = "macos")] +#[test] +fn avassetreader_switches_to_software_only_when_its_reader_stops_answering() { + use super::{FrameDeadline, MacOsDecoderFallback}; + + let script = Script { + time_base: (1, 30), + frames: (0..120) + .map(|pts| scripted::ScriptFrame { + pts, + keyframe: pts % 30 == 0, + }) + .collect(), + last_duration: 1, + }; + let runtime = runtime(); + let time = |frame: u32| frame as f32 / 30.0; + let slow_vend = Duration::from_millis(1500); + let deadline_ms = 250; + + let registration = scripted::register(script.clone()); + let path = registration.path().to_path_buf(); + runtime.block_on(async { + let handle = spawn_decoder("model", path, 30, 0.0, false) + .await + .unwrap() + .with_max_fallback_distance(2); + let mut frame = 0; + for episode in 0..3 { + registration.set_vend_delay(slow_vend); + for _ in 1..MacOsDecoderFallback::MAX_TIMED_OUT_ROUNDS { + frame += 1; + let late = handle + .get_frame_with_timeout(time(frame), deadline_ms, 2, FrameDeadline::Drop) + .await; + assert!( + matches!(late, Ok(None)), + "slow episode {episode}: request {frame} beat a {deadline_ms} ms deadline" + ); + assert_eq!(handle.decoder_type(), DecoderType::AVAssetReader); + } + registration.set_vend_delay(Duration::ZERO); + frame += 1; + assert_eq!( + handle + .get_frame_initial(time(frame)) + .await + .as_ref() + .and_then(frame_identity), + Some(frame), + "after slow episode {episode}" + ); + assert_eq!( + handle.decoder_type(), + DecoderType::AVAssetReader, + "slow episode {episode} switched to {}", + handle.decoder_type() + ); + } + + registration.set_vend_delay(slow_vend); + for round in 1..=MacOsDecoderFallback::MAX_TIMED_OUT_ROUNDS { + frame += 1; + let request = + handle.get_frame_with_timeout(time(frame), deadline_ms, 2, FrameDeadline::Drop); + if round < MacOsDecoderFallback::MAX_TIMED_OUT_ROUNDS { + assert!( + matches!(request.await, Ok(None)), + "stalled round {round}: request {frame} beat a {deadline_ms} ms deadline" + ); + assert_eq!(handle.decoder_type(), DecoderType::AVAssetReader); + continue; + } + // The reader is still inside the vend it started with the slow + // delay in round 1, so lifting the delay now lets only the + // software decoder answer this request within its deadline. + let lift_delay = async { + tokio::time::sleep(Duration::from_millis(deadline_ms / 2)).await; + registration.set_vend_delay(Duration::ZERO); + }; + let (answered, ()) = tokio::join!(request, lift_delay); + assert_eq!( + answered.unwrap().as_ref().and_then(frame_identity), + Some(frame), + "the request that switched to software decoding" + ); + } + assert_eq!(handle.decoder_type(), DecoderType::FFmpegSoftware); + assert_eq!( + handle.fallback_reason(), + Some("AVAssetReader stopped answering") + ); + frame += 1; + assert_eq!( + handle + .get_frame(time(frame)) + .await + .as_ref() + .and_then(frame_identity), + Some(frame) + ); + }); + + let registration = scripted::register(script.clone()); + let path = registration.path().to_path_buf(); + runtime.block_on(async { + let handle = spawn_decoder("model", path, 30, 0.0, false) + .await + .unwrap() + .with_max_fallback_distance(2); + assert_eq!( + handle + .get_frame_initial(time(5)) + .await + .as_ref() + .and_then(frame_identity), + Some(5) + ); + registration.set_vend_delay(slow_vend); + let late = handle + .get_frame_with_timeout(time(6), deadline_ms, 2, FrameDeadline::Drop) + .await; + assert!( + matches!(late, Ok(None)), + "request 6 beat a {deadline_ms} ms deadline" + ); + assert_eq!(handle.decoder_type(), DecoderType::AVAssetReader); + let seek = handle.get_frame_with_timeout(time(10), deadline_ms, 2, FrameDeadline::Initial); + let lift_delay = async { + tokio::time::sleep(Duration::from_millis(deadline_ms / 2)).await; + registration.set_vend_delay(Duration::ZERO); + }; + let (answered, ()) = tokio::join!(seek, lift_delay); + assert_eq!( + answered.unwrap().as_ref().and_then(frame_identity), + Some(10), + "the first timed-out initial request" + ); + assert_eq!(handle.decoder_type(), DecoderType::FFmpegSoftware); + assert_eq!( + handle.fallback_reason(), + Some("AVAssetReader stopped answering") + ); + }); + + let registration = scripted::register(script); + registration.fail_avf_read_at(40); + registration.fail_avf_read_at(40); + let path = registration.path().to_path_buf(); + runtime.block_on(async { + let handle = spawn_decoder("model", path, 30, 0.0, false) + .await + .unwrap() + .with_max_fallback_distance(2); + assert_eq!( + handle + .get_frame(time(45)) + .await + .as_ref() + .and_then(frame_identity), + Some(45) + ); + assert_eq!(handle.decoder_type(), DecoderType::FFmpegSoftware); + assert_eq!( + handle.fallback_reason(), + Some("AVAssetReader returned no frame") + ); + }); +} + #[test] fn ffmpeg_answers_match_the_reference_model() { run_suite(Decoder::Ffmpeg);