diff --git a/crates/rendering/src/decoder/avassetreader.rs b/crates/rendering/src/decoder/avassetreader.rs index af79c9560e2..97ec15b3813 100644 --- a/crates/rendering/src/decoder/avassetreader.rs +++ b/crates/rendering/src/decoder/avassetreader.rs @@ -537,7 +537,12 @@ impl AVAssetReaderDecoder { }) } - fn select_best_decoder(&mut self, requested_time: f32, is_scrubbing: bool) -> (usize, bool) { + fn select_best_decoder( + &mut self, + requested_time: f32, + is_scrubbing: bool, + must_reset: impl Fn(&DecoderInstance) -> bool, + ) -> (usize, bool) { let decoder_count = self.decoders.len(); let (best_id, _distance, needs_reset) = if is_scrubbing { self.pool_manager @@ -552,6 +557,7 @@ impl AVAssetReaderDecoder { }; let decoder_idx = best_id.min(decoder_count.saturating_sub(1)); + let needs_reset = needs_reset || self.decoders.get(decoder_idx).is_some_and(must_reset); let max_decoder_count = self .pool_manager @@ -660,7 +666,7 @@ impl AVAssetReaderDecoder { } }; let mut this = None::; - let mut keyframe_index = None; + let mut keyframe_index: Option> = None; let init_result = DecoderInitResult { width, @@ -694,6 +700,7 @@ impl AVAssetReaderDecoder { frame: u32, max_fallback_distance: u32, sender: oneshot::Sender, + retried: bool, } let mut deferred_requests = VecDeque::::new(); @@ -751,6 +758,7 @@ impl AVAssetReaderDecoder { frame, max_fallback_distance, sender, + retried: false, }); } } @@ -769,6 +777,7 @@ impl AVAssetReaderDecoder { frame, max_fallback_distance, sender, + retried: false, }); } } @@ -851,6 +860,28 @@ impl AVAssetReaderDecoder { } pending_requests = still_unfulfilled; + // A track can start after the recording does. Before its first + // frame, that frame is the answer; serving it from the cache keeps + // the reader from stepping ahead on every such request. + let stream_start_frame = keyframe_index + .as_ref() + .and_then(|index| index.keyframes().first()) + .map(|&(_, secs)| (secs * f64::from(fps)).round() as u32); + if let Some(start) = stream_start_frame + && let Some(first) = cache.get(&start).map(|cached| cached.data().clone()) + { + let mut after_start = Vec::with_capacity(pending_requests.len()); + for req in pending_requests.drain(..) { + if req.frame < start { + *last_sent_frame.borrow_mut() = Some(first.clone()); + let _ = req.sender.send(first.to_decoded_frame()); + } else { + after_start.push(req); + } + } + pending_requests = after_start; + } + if pending_requests.is_empty() { continue; } @@ -890,7 +921,20 @@ impl AVAssetReaderDecoder { } let this = this.as_mut().expect("Video reader initialized above"); - let (decoder_idx, was_reset) = this.select_best_decoder(requested_time, is_scrubbing); + // Pool positions are capped to the newest request, so a reader that + // decoded ahead can look usable for a request it has already passed. + // It can never vend that frame again, and reading on would answer + // with a stale or later frame, so it has to seek back first. Before + // the stream's first frame nothing better exists, so a seek is wasted. + let retrying = pending_requests.iter().any(|request| request.retried); + let before_stream_start = + stream_start_frame.is_some_and(|start| requested_frame < start); + let (decoder_idx, was_reset) = + this.select_best_decoder(requested_time, is_scrubbing, |decoder| { + retrying + || (!before_stream_start + && reader_passed_request(decoder.prev_vended, requested_frame)) + }); let cache_min = if was_reset { min_requested_frame.saturating_sub(FRAME_CACHE_SIZE as u32 * 2) @@ -903,9 +947,23 @@ impl AVAssetReaderDecoder { max_requested_frame + FRAME_CACHE_SIZE as u32 }; + // The last frame each other pooled reader vended is the answer for a + // request that lands in the gap after it, and the reader can never + // vend it again; keep it cached while requests alternate between + // readers. + let pinned_frames: Vec = this + .decoders + .iter() + .enumerate() + .filter(|&(index, _)| index != decoder_idx) + .filter_map(|(_, decoder)| decoder.prev_vended) + .collect(); + if was_reset { *last_sent_frame.borrow_mut() = None; - cache.retain(|&f, _| f >= cache_min && f <= cache_max); + cache.retain(|f, _| { + (*f >= cache_min && *f <= cache_max) || pinned_frames.contains(f) + }); cache_bytes = cache .values() .map(CachedFrame::byte_len) @@ -923,6 +981,7 @@ impl AVAssetReaderDecoder { // reader re-vends from the keyframe at-or-before the request, and // that frame is the true VFR hold for requests inside a pts hole. let mut hold_candidate: Option<(u32, R)> = None; + let mut retry_requests = Vec::new(); { let decoder = &mut this.decoders[decoder_idx]; @@ -950,7 +1009,8 @@ impl AVAssetReaderDecoder { let current_frame = pts_to_frame(frame.pts().value, Rational::new(1, frame.pts().scale), fps); - if let Some(prev) = decoder.prev_vended + let vended_before = decoder.prev_vended; + if let Some(prev) = vended_before && current_frame > prev + 1 { record_pts_hole(&mut pts_holes, prev, current_frame); @@ -1002,16 +1062,18 @@ impl AVAssetReaderDecoder { ) { if let Some(last_active) = &last_active_frame { - let frame_to_remove = if requested_frame > *last_active { - *cache.keys().next().unwrap() + let from_oldest = if requested_frame > *last_active { + true } else if requested_frame < *last_active { - *cache.keys().next_back().unwrap() + false } else { - let min = *cache.keys().min().unwrap(); - let max = *cache.keys().max().unwrap(); - if current_frame > max { min } else { max } + current_frame > *cache.keys().next_back().unwrap() }; - if let Some(removed) = cache.remove(&frame_to_remove) { + let victim = cache_eviction_victim(&cache, from_oldest, |frame| { + Some(frame) == vended_before || pinned_frames.contains(&frame) + }); + if let Some(removed) = victim.and_then(|frame| cache.remove(&frame)) + { cache_bytes = cache_bytes.saturating_sub(removed.byte_len()); } } else { @@ -1065,6 +1127,26 @@ impl AVAssetReaderDecoder { .flatten() .max_by_key(|d| d.number); + // The reader vends every sample in pts order, so the + // frame it vended just before this one is the newest + // at or before the request. When that frame is no + // longer held anywhere, any other answer is stale or + // from the future: retry once from a fresh seek, which + // re-vends and keeps it. + if !req.retried + && answer_misses_previous_vend( + vended_before, + req.frame, + best_before.as_ref().map(|d| d.number), + ) + { + retry_requests.push(PendingRequest { + retried: true, + ..req + }); + continue; + } + if let Some(data) = best_before { gap_hold = Some(data.clone()); *last_sent_frame.borrow_mut() = Some(data.clone()); @@ -1105,6 +1187,10 @@ impl AVAssetReaderDecoder { decoder.is_done = true; } + for request in retry_requests.into_iter().rev() { + deferred_requests.push_front(request); + } + let last_frame_in_cache = cache.keys().max().copied().unwrap_or(0); if frames_iterated > 0 { this.decoders[decoder_idx] @@ -1278,6 +1364,45 @@ impl AVAssetReaderDecoder { } } +fn reader_passed_request(prev_vended: Option, requested_frame: u32) -> bool { + prev_vended.is_some_and(|prev| prev >= requested_frame) +} + +// A request the reader steps over is answered by the frame it vended just +// before, the newest one at or before the request. Any other answer is stale. +fn answer_misses_previous_vend( + vended_before: Option, + requested_frame: u32, + answer: Option, +) -> bool { + vended_before + .filter(|&number| number <= requested_frame) + .is_some_and(|expected| answer != Some(expected)) +} + +// Drops from the end the cache is trimmed from, skipping pinned frames unless +// every frame is pinned. +fn cache_eviction_victim( + cache: &BTreeMap, + from_oldest: bool, + is_pinned: impl Fn(u32) -> bool, +) -> Option { + if from_oldest { + cache + .keys() + .copied() + .find(|&frame| !is_pinned(frame)) + .or_else(|| cache.keys().next().copied()) + } else { + cache + .keys() + .rev() + .copied() + .find(|&frame| !is_pinned(frame)) + .or_else(|| cache.keys().next_back().copied()) + } +} + fn should_grow_decoder_pool( is_scrubbing: bool, needs_reset: bool, @@ -1357,6 +1482,44 @@ mod tests { )); } + #[test] + fn a_reader_that_vended_the_request_cannot_serve_it_again() { + assert!(!reader_passed_request(None, 10)); + assert!(!reader_passed_request(Some(9), 10)); + assert!(reader_passed_request(Some(10), 10)); + assert!(reader_passed_request(Some(57), 54)); + } + + #[test] + fn only_the_frame_vended_before_a_gap_answers_requests_inside_it() { + assert!(!answer_misses_previous_vend(Some(1054), 1055, Some(1054))); + assert!(answer_misses_previous_vend(Some(1054), 1055, Some(1052))); + assert!(answer_misses_previous_vend(Some(751), 752, Some(421))); + assert!(answer_misses_previous_vend(Some(751), 752, None)); + assert!(!answer_misses_previous_vend(None, 752, Some(421))); + assert!(!answer_misses_previous_vend(Some(760), 752, Some(751))); + } + + #[test] + fn cache_eviction_skips_pinned_frames() { + let cache: BTreeMap = [415, 419, 421, 749, 751] + .into_iter() + .map(|f| (f, ())) + .collect(); + assert_eq!(cache_eviction_victim(&cache, true, |_| false), Some(415)); + assert_eq!(cache_eviction_victim(&cache, false, |_| false), Some(751)); + assert_eq!( + cache_eviction_victim(&cache, false, |f| f == 751), + Some(749) + ); + assert_eq!(cache_eviction_victim(&cache, true, |f| f < 420), Some(421)); + assert_eq!(cache_eviction_victim(&cache, true, |_| true), Some(415)); + assert_eq!( + cache_eviction_victim(&BTreeMap::::new(), true, |_| false), + None + ); + } + #[test] fn decoder_pool_grows_only_for_unserved_non_sequential_access() { assert!(should_grow_decoder_pool(true, true, 1, 5)); diff --git a/crates/rendering/src/decoder/mod.rs b/crates/rendering/src/decoder/mod.rs index 170507f3090..75bd57d1a45 100644 --- a/crates/rendering/src/decoder/mod.rs +++ b/crates/rendering/src/decoder/mod.rs @@ -671,24 +671,30 @@ pub fn pts_to_frame(pts: i64, time_base: Rational, fps: u32) -> u32 { pub const FRAME_CACHE_SIZE: usize = 90; const DEFAULT_MAX_FALLBACK_DISTANCE: u32 = 90; +/// Bounds the pts-hole map. Holes are facts about the file, and a pooled +/// reader parked past a hole never re-vends it, so the map keeps every hole a +/// realistic recording has and only evicts on pathological inputs. +const MAX_TRACKED_HOLES: usize = 1 << 14; + /// Records a pts hole discovered from a decode-order vend jump (frames vend -/// in pts order, so a jump means no samples exist in between). The map stays -/// bounded by dropping the narrowest hole — wide static-screen holds matter -/// most. +/// in pts order, so a jump means no samples exist in between). When the map +/// is full it drops the hole farthest from the one just recorded: requests +/// trail the reader, so the newest holes are the ones about to be asked for. pub(super) fn record_pts_hole( holes: &mut std::collections::BTreeMap, start: u32, end: u32, ) { - const MAX_TRACKED_HOLES: usize = 64; holes.insert(start, end); if holes.len() > MAX_TRACKED_HOLES - && let Some(narrowest) = holes - .iter() - .min_by_key(|&(&s, &e)| e.saturating_sub(s)) - .map(|(&s, _)| s) + && let (Some(&first), Some(&last)) = (holes.keys().next(), holes.keys().next_back()) { - holes.remove(&narrowest); + let farthest = if start.abs_diff(first) >= start.abs_diff(last) { + first + } else { + last + }; + holes.remove(&farthest); } } @@ -1654,6 +1660,51 @@ pub async fn spawn_decoder( mod tests { use super::*; + #[test] + fn pts_hole_tracking_keeps_the_newest_holes_when_full() { + let mut holes = std::collections::BTreeMap::new(); + let mut frame = 0u32; + for i in 0..(MAX_TRACKED_HOLES as u32 + 1000) { + let width = if i % 4 == 0 { 3 } else { 2 }; + record_pts_hole(&mut holes, frame, frame + width); + assert_eq!(holes.get(&frame), Some(&(frame + width))); + frame += width; + } + assert_eq!(holes.len(), MAX_TRACKED_HOLES); + assert!( + holes + .keys() + .all(|&start| start as usize + 3 * MAX_TRACKED_HOLES >= frame as usize) + ); + } + + #[test] + fn pts_hole_tracking_drops_the_farthest_hole_after_a_backward_seek() { + let mut holes = std::collections::BTreeMap::new(); + let last = MAX_TRACKED_HOLES as u32 - 1; + for i in 0..=last { + record_pts_hole(&mut holes, 1000 + i * 2, 1000 + i * 2 + 2); + } + record_pts_hole(&mut holes, 10, 40); + assert_eq!(holes.len(), MAX_TRACKED_HOLES); + assert_eq!(holes.get(&10), Some(&40)); + assert!(holes.contains_key(&1000)); + assert!(!holes.contains_key(&(1000 + last * 2))); + } + + #[test] + fn pts_hole_tracking_keeps_an_old_static_hold_while_later_holes_arrive() { + let mut holes = std::collections::BTreeMap::new(); + record_pts_hole(&mut holes, 1797, 2402); + let mut frame = 2402u32; + for i in 0..2000u32 { + let width = 2 + i % 7; + record_pts_hole(&mut holes, frame, frame + width); + frame += width; + } + assert_eq!(holes.get(&1797), Some(&2402)); + } + #[test] fn decoded_frame_storage_identity_matches_clones() { let frame = DecodedFrame::new_nv12(vec![0; 6], 2, 2, 2, 2); diff --git a/crates/rendering/src/yuv_converter.rs b/crates/rendering/src/yuv_converter.rs index 4bd420ddb2a..6791c1cc513 100644 --- a/crates/rendering/src/yuv_converter.rs +++ b/crates/rendering/src/yuv_converter.rs @@ -410,6 +410,25 @@ impl YuvConverterPipelines { } } +/// Conversions whose GPU work can still be pending when a new one starts. The +/// renderer waits for the previous frame only after submitting the next one, +/// and a transition frame converts twice (outgoing, then incoming), so up to +/// three earlier sources may still be sampled. +#[cfg(target_os = "macos")] +const RETAINED_IOSURFACE_SOURCES: usize = 4; + +#[cfg(target_os = "macos")] +struct RetainedImageBuf { + _image_buf: cidre::arc::R, +} + +// SAFETY: the buffer is only retained and released, never read, and +// CoreVideo reference counting is thread-safe. +#[cfg(target_os = "macos")] +unsafe impl Send for RetainedImageBuf {} +#[cfg(target_os = "macos")] +unsafe impl Sync for RetainedImageBuf {} + pub struct YuvToRgbaConverter { pipelines: Arc, y_texture: wgpu::Texture, @@ -429,6 +448,10 @@ pub struct YuvToRgbaConverter { bind_group_cache: BindGroupCache, #[cfg(target_os = "macos")] iosurface_cache: Option, + #[cfg(target_os = "macos")] + iosurface_sources: [Option; RETAINED_IOSURFACE_SOURCES], + #[cfg(target_os = "macos")] + next_iosurface_source: usize, #[cfg(target_os = "windows")] d3d11_staging_texture: Option, #[cfg(target_os = "windows")] @@ -489,6 +512,10 @@ impl YuvToRgbaConverter { bind_group_cache: BindGroupCache::new(), #[cfg(target_os = "macos")] iosurface_cache: IOSurfaceTextureCache::new(), + #[cfg(target_os = "macos")] + iosurface_sources: Default::default(), + #[cfg(target_os = "macos")] + next_iosurface_source: 0, #[cfg(target_os = "windows")] d3d11_staging_texture: None, #[cfg(target_os = "windows")] @@ -1099,6 +1126,15 @@ impl YuvToRgbaConverter { compute_pass.dispatch_workgroups(width.div_ceil(8), height.div_ceil(8), 1); } + // The Metal textures alias the IOSurface without retaining the pixel + // buffer, and the decoder recycles released buffers for new pictures. + // Hold the source until enough later conversions have started that + // the caller has waited for the GPU work that sampled it. + self.iosurface_sources[self.next_iosurface_source] = Some(RetainedImageBuf { + _image_buf: image_buf.retained(), + }); + self.next_iosurface_source = (self.next_iosurface_source + 1) % RETAINED_IOSURFACE_SOURCES; + Ok(self.current_output_view()) }