From 6d804384efe516896663ba58989d51089771bca1 Mon Sep 17 00:00:00 2001 From: Marcos Daniel Rivera Gonzalez <139647784+LordOfGrillos@users.noreply.github.com> Date: Tue, 13 Oct 2026 21:39:52 -0600 Subject: [PATCH 1/2] fix(video-decode): read past Media Foundation samples that carry no frame ReadSample hands back a null sample for stream ticks (gaps in the stream), and read_sample reported those as Ok(None), the same as the end of the stream. The decoder then stopped a batch early and answered with an older frame. Read again until a sample, the end-of-stream flag or an error. --- crates/video-decode/src/media_foundation.rs | 46 +++++++++++---------- 1 file changed, 25 insertions(+), 21 deletions(-) diff --git a/crates/video-decode/src/media_foundation.rs b/crates/video-decode/src/media_foundation.rs index 45b6cf5a351..6b5755214bb 100644 --- a/crates/video-decode/src/media_foundation.rs +++ b/crates/video-decode/src/media_foundation.rs @@ -763,32 +763,36 @@ impl MediaFoundationDecoder { let mut timestamp = 0i64; let mut sample: Option = None; - unsafe { - self.source_reader - .ReadSample( - MF_SOURCE_READER_FIRST_VIDEO_STREAM.0 as u32, - 0, - Some(&mut stream_index), - Some(&mut flags), - Some(&mut timestamp), - Some(&mut sample), - ) - .map_err(|e| format!("ReadSample failed: {e:?}"))?; - } - const MF_SOURCE_READERF_ENDOFSTREAM: u32 = 0x00000001; const MF_SOURCE_READERF_ERROR: u32 = 0x00000002; - if flags & MF_SOURCE_READERF_ENDOFSTREAM != 0 { - return Ok(None); - } + // A stream tick (a gap in the stream) comes without a sample; only the end-of-stream + // flag means no samples follow. + let sample = loop { + unsafe { + self.source_reader + .ReadSample( + MF_SOURCE_READER_FIRST_VIDEO_STREAM.0 as u32, + 0, + Some(&mut stream_index), + Some(&mut flags), + Some(&mut timestamp), + Some(&mut sample), + ) + .map_err(|e| format!("ReadSample failed: {e:?}"))?; + } - if flags & MF_SOURCE_READERF_ERROR != 0 { - return Err("Stream error".to_string()); - } + if flags & MF_SOURCE_READERF_ENDOFSTREAM != 0 { + return Ok(None); + } - let Some(sample) = sample else { - return Ok(None); + if flags & MF_SOURCE_READERF_ERROR != 0 { + return Err("Stream error".to_string()); + } + + if let Some(sample) = sample.take() { + break sample; + } }; let buffer = unsafe { From ba1f2c2460491946672f293344d3369223293b7d Mon Sep 17 00:00:00 2001 From: Marcos Daniel Rivera Gonzalez <139647784+LordOfGrillos@users.noreply.github.com> Date: Tue, 13 Oct 2026 21:39:53 -0600 Subject: [PATCH 2/2] refactor(rendering): drive the Media Foundation decoder from one request state machine The decoder thread mixed batching, a mid-batch readahead slide, deferred requests and four fallback chains in one loop. The policy now lives in FrameRequests, which never touches Media Foundation: the thread takes in requests, asks it for the next step (seek, decode one sample, or wait) and does it. Tests drive it with a fake sample stream. Once the decoder moves past a requested frame, the request gets the newest sample at or below that frame. Requests are served lowest first, each with its own seek when decoding forward cannot reach it, so the "first sample after" fallback only applies to a frame before the stream's first sample. The cache keeps every sample since the last seek and evicts the oldest, so the newest sample at or below a frame is always cached. Behaviour changes: - Several requests in flight (export transitions between two segments of one recording decode both through the same decoder) each get their own frame. The highest used to be the target, and a lower one far from it got the target's frame. - A request behind the decoder with nothing cached at or below it seeks back. Within half the cache size it used to get the next decoded, later frame. - Every request slides the readahead, cache hits included, so a sequential reader no longer drains the cache while the decoder idles. - Readahead stops at its frame count, or before the next sample would evict the one answering the latest request (frames too big for the cache budget used to get a later frame); the per-batch time budget is gone. - Samples outside the old cache window are cached too, so a frame after a VFR gap past the window gets its own sample, not the one before the gap. - A failed seek answers what waits from the cache instead of reading on from the old position as if it had landed. Export output is byte-identical to the previous decoder on the reference recording. --- .../rendering/src/decoder/media_foundation.rs | 827 +++++++++++------- 1 file changed, 501 insertions(+), 326 deletions(-) diff --git a/crates/rendering/src/decoder/media_foundation.rs b/crates/rendering/src/decoder/media_foundation.rs index fa140793173..09c2f21d69d 100644 --- a/crates/rendering/src/decoder/media_foundation.rs +++ b/crates/rendering/src/decoder/media_foundation.rs @@ -230,9 +230,6 @@ impl MFDecoder { let video_width = decoder.width(); let video_height = decoder.height(); - let mut cache = BTreeMap::::new(); - let mut cache_bytes = 0usize; - let mut last_decoded_frame: Option = None; let mut health = DecoderHealthMonitor::new(); let init_result = DecoderInitResult { @@ -242,313 +239,106 @@ impl MFDecoder { }; let _ = ready_tx.send(Ok(init_result)); - struct PendingRequest { - frame: u32, - time: f32, - sender: oneshot::Sender, - } - - let mut carried: Option = None; - while let Some(r) = carried.take().or_else(|| rx.recv().ok()) { - let mut pending_requests: Vec = Vec::with_capacity(8); + let pixel_count = (video_width as u64) * (video_height as u64); + let readahead_frames = if pixel_count > 4_000_000 { + 8 + } else if pixel_count > 2_000_000 { + 15 + } else { + 30 + }; + let mut requests = FrameRequests::new(readahead_frames); + let answer = |sender: oneshot::Sender, frame: Option| { + let _ = sender.send(frame.map_or_else( + || { + DecodedFrame::new( + vec![0u8; (video_width * video_height * 4) as usize], + video_width, + video_height, + ) + }, + |frame| frame.to_decoded_frame(), + )); + }; - let mut push_request = - |requested_time: f32, sender: oneshot::Sender| { - if sender.is_closed() { - return; - } - let frame = (requested_time * fps as f32).floor() as u32; - pending_requests.push(PendingRequest { - frame, - time: requested_time, - sender, - }); + loop { + loop { + let message = match requests.next_step() { + Step::Wait => rx.recv().ok(), + _ => match rx.try_recv() { + Err(mpsc::TryRecvError::Empty) => break, + message => message.ok(), + }, }; - - match r { - VideoDecoderMessage::GetFrame(requested_time, _, sender) => { - push_request(requested_time, sender); + // Each request borrows a decoder handle, which owns the sender, so once the + // channel is gone nobody is waiting for an answer. + let Some(message) = message else { + return; + }; + let VideoDecoderMessage::GetFrame(time, _, sender) = message; + if sender.is_closed() { + continue; } - } - while let Ok(msg) = rx.try_recv() { - match msg { - VideoDecoderMessage::GetFrame(requested_time, _, sender) => { - push_request(requested_time, sender); + let frame = (time * fps as f32).floor() as u32; + match requests.request(frame, sender) { + Some((sender, cached)) => answer(sender, cached), + None => { + health.record_request(); + if health.should_warn_unhealthy() { + warn!( + name = name, + consecutive_errors = health.consecutive_errors, + texture_failures = health.consecutive_texture_read_failures, + total_decoded = health.total_frames_decoded, + "MediaFoundation decoder unhealthy, performance may degrade" + ); + } } } } - - let mut unfulfilled = Vec::with_capacity(pending_requests.len()); - for req in pending_requests.drain(..) { - if let Some(frame) = cached_frame_for(&cache, last_decoded_frame, req.frame) { - let _ = req.sender.send(frame.to_decoded_frame()); - } else if !req.sender.is_closed() { - unfulfilled.push(req); - } - } - pending_requests = unfulfilled; - - if pending_requests.is_empty() { - continue; - } - - pending_requests.sort_by_key(|r| r.frame); - - let target_request = pending_requests.pop().unwrap(); - let deferred_requests = pending_requests; - - let requested_frame = target_request.frame; - let mut sender = Some(target_request.sender); - - health.record_request(); - - if health.should_warn_unhealthy() { - warn!( - name = name, - consecutive_errors = health.consecutive_errors, - texture_failures = health.consecutive_texture_read_failures, - total_decoded = health.total_frames_decoded, - "MediaFoundation decoder unhealthy, performance may degrade" - ); - } - - let mut cache_min = requested_frame.saturating_sub(FRAME_CACHE_SIZE as u32 / 2); - let mut cache_max = requested_frame + FRAME_CACHE_SIZE as u32 / 2; - - let needs_seek = last_decoded_frame - .map(|last| { - if requested_frame <= last { - last - requested_frame > FRAME_CACHE_SIZE as u32 / 2 - } else { - requested_frame - last > FRAME_CACHE_SIZE as u32 + requests.waiting.retain(|(_, sender)| !sender.is_closed()); + + let answers = match requests.next_step() { + Step::Wait => Vec::new(), + Step::Seek(frame) => match decoder.seek(frame_to_100ns(frame, fps)) { + Ok(()) => { + requests.seeked(frame); + Vec::new() } - }) - .unwrap_or(true); - - if needs_seek { - let time_100ns = frame_to_100ns(requested_frame, fps); - if let Err(e) = decoder.seek(time_100ns) { - warn!("MediaFoundation seek failed: {e}"); - } - cache.clear(); - cache_bytes = 0; - last_decoded_frame = None; - } - - let mut last_valid_frame: Option = None; - - let pixel_count = (video_width as u64) * (video_height as u64); - let readahead_frames = if pixel_count > 4_000_000 { - 8u32 - } else if pixel_count > 2_000_000 { - 15u32 - } else { - 30u32 - }; - let max_batch_duration = if pixel_count > 4_000_000 { - Duration::from_millis(200) - } else if pixel_count > 2_000_000 { - Duration::from_millis(400) - } else { - Duration::from_millis(800) - }; - let mut batch_start = Instant::now(); - let mut readahead_target = requested_frame + readahead_frames; - - loop { - if sender.as_ref().is_some_and(|s| s.is_closed()) { - sender.take(); - break; - } - - if sender.is_none() && batch_start.elapsed() > max_batch_duration { - break; - } - - // Serve requests that arrive during readahead between samples and slide the - // window with them; otherwise a sequential reader waits out the whole batch - // and then finds the decoder idle while it consumes the cached frames. - if sender.is_none() && deferred_requests.is_empty() { - match rx.try_recv() { - Ok(VideoDecoderMessage::GetFrame(time, distance, reply)) => { - let frame = (time * fps as f32).floor() as u32; - if let Some(cached) = - cached_frame_for(&cache, last_decoded_frame, frame) - { - let _ = reply.send(cached.to_decoded_frame()); - readahead_target = - readahead_target.max(frame + readahead_frames); - cache_min = frame.saturating_sub(FRAME_CACHE_SIZE as u32 / 2); - cache_max = frame + FRAME_CACHE_SIZE as u32 / 2; - batch_start = Instant::now(); - continue; - } - carried = - Some(VideoDecoderMessage::GetFrame(time, distance, reply)); - break; - } - Err(mpsc::TryRecvError::Empty) => {} - Err(mpsc::TryRecvError::Disconnected) => break, + Err(e) => { + warn!("MediaFoundation seek failed: {e}"); + requests.stopped() } - } - - let decode_start = Instant::now(); - match decoder.read_sample() { - Ok(Some(mf_frame)) => { - let decode_time = decode_start.elapsed(); - let frame_number = pts_100ns_to_frame(mf_frame.pts, fps); - - let has_valid_zero_copy_handles = { - let null_ptr = std::ptr::null_mut(); - mf_frame.textures.y.handle.0 != null_ptr - && mf_frame.textures.uv.handle.0 != null_ptr - }; - - let nv12_data = if has_valid_zero_copy_handles { - health.record_success(decode_time); - None - } else { - match decoder.read_texture_to_cpu( - &mf_frame.textures.nv12.texture, - mf_frame.width, - mf_frame.height, - ) { - Ok(data) => { - health.record_success(decode_time); - Some(Arc::new(data)) - } - Err(e) => { - health.record_texture_read_failure(); - warn!( - "Failed to read texture to CPU for frame {frame_number}: {e}" - ); - None - } - } - }; - - let cached = CachedFrame { - number: frame_number, - _texture: has_valid_zero_copy_handles - .then(|| mf_frame.textures.nv12.texture.clone()), - _shared_handle: has_valid_zero_copy_handles - .then_some(mf_frame.textures.nv12.handle), - _y_handle: has_valid_zero_copy_handles - .then_some(mf_frame.textures.y.handle), - _uv_handle: has_valid_zero_copy_handles - .then_some(mf_frame.textures.uv.handle), - nv12_data, - width: mf_frame.width, - height: mf_frame.height, - }; - - if !has_valid_zero_copy_handles { - decoder.recycle_textures(mf_frame.textures); + }, + Step::Decode => { + let decode_start = Instant::now(); + match decoder.read_sample() { + Ok(Some(mf_frame)) => { + let cached = cache_sample( + &mut decoder, + mf_frame, + decode_start.elapsed(), + &mut health, + fps, + ); + let (number, bytes) = (cached.number, cached.estimated_bytes()); + requests.decoded(number, cached, bytes) } - - last_decoded_frame = Some(frame_number); - - if frame_number >= cache_min && frame_number <= cache_max { - let frame_bytes = cached.estimated_bytes(); - while !cache.is_empty() - && (cache.len() >= FRAME_CACHE_SIZE - || cache_bytes.saturating_add(frame_bytes) - > MAX_FRAME_CACHE_BYTES) - { - let key_to_remove = if frame_number > requested_frame { - *cache.keys().next().unwrap() - } else { - *cache.keys().next_back().unwrap() - }; - if let Some(evicted) = cache.remove(&key_to_remove) { - cache_bytes = - cache_bytes.saturating_sub(evicted.estimated_bytes()); - } - } - if let Some(replaced) = cache.insert(frame_number, cached.clone()) { - cache_bytes = - cache_bytes.saturating_sub(replaced.estimated_bytes()); - } - cache_bytes = cache_bytes.saturating_add(frame_bytes); + Ok(None) => requests.stopped(), + Err(e) => { + health.record_error(); + warn!( + consecutive_errors = health.consecutive_errors, + "MediaFoundation read_sample error: {e}" + ); + requests.stopped() } - - if frame_number <= requested_frame { - last_valid_frame = Some(cached); - } - - if frame_number > requested_frame { - let frame_to_send = last_valid_frame.as_ref().or_else(|| { - cache - .range(..=requested_frame) - .next_back() - .map(|(_, f)| f) - .or_else(|| cache.get(&frame_number)) - }); - - if let Some(frame) = frame_to_send - && let Some(s) = sender.take() - { - let _ = s.send(frame.to_decoded_frame()); - } - } - - if frame_number >= readahead_target || frame_number > cache_max { - break; - } - } - Ok(None) => { - break; - } - Err(e) => { - health.record_error(); - warn!( - consecutive_errors = health.consecutive_errors, - "MediaFoundation read_sample error: {e}" - ); - break; } } - } - - if let Some(s) = sender.take() { - let fallback = last_valid_frame.as_ref().cloned().or_else(|| { - cache - .range(..=requested_frame) - .next_back() - .or_else(|| cache.range(requested_frame..).next()) - .map(|(_, f)| f.clone()) - }); - if let Some(frame) = fallback { - let _ = s.send(frame.to_decoded_frame()); - } else { - let black_frame = DecodedFrame::new( - vec![0u8; (video_width * video_height * 4) as usize], - video_width, - video_height, - ); - let _ = s.send(black_frame); - } - } - - for req in deferred_requests { - if req.sender.is_closed() { - continue; - } - let frame_to_send = cache - .get(&req.frame) - .or_else(|| cache.range(..=req.frame).next_back().map(|(_, f)| f)) - .or(last_valid_frame.as_ref()); - if let Some(frame) = frame_to_send { - let _ = req.sender.send(frame.to_decoded_frame()); - } else { - let black_frame = DecodedFrame::new( - vec![0u8; (video_width * video_height * 4) as usize], - video_width, - video_height, - ); - let _ = req.sender.send(black_frame); - } + }; + for (sender, cached) in answers { + answer(sender, cached); } } }); @@ -568,22 +358,215 @@ fn pts_100ns_to_frame(pts_100ns: i64, fps: u32) -> u32 { super::pts_to_frame(pts_100ns, ::ffmpeg::Rational::new(1, 10_000_000), fps) } -// A frame the decoder has not moved past yet has to be decoded: answering with -// the cached frame before it shows a stale image at every readahead boundary, -// and a VFR stream can still have a later sample rounding to the same frame -// (which replaces it in the cache), so answering early would make the choice -// depend on request timing. Once the decoder is past a missing frame, the -// stream has no frame there, and the latest earlier one is the hold the video -// really shows. -fn cached_frame_for( - cache: &BTreeMap, - last_decoded: Option, - frame: u32, -) -> Option<&T> { - last_decoded - .is_some_and(|last| last > frame) - .then(|| cache.range(..=frame).next_back().map(|(_, cached)| cached)) - .flatten() +fn cache_sample( + decoder: &mut cap_video_decode::MediaFoundationDecoder, + mf_frame: cap_video_decode::MFDecodedFrame, + decode_time: Duration, + health: &mut DecoderHealthMonitor, + fps: u32, +) -> CachedFrame { + let frame_number = pts_100ns_to_frame(mf_frame.pts, fps); + + let has_valid_zero_copy_handles = { + let null_ptr = std::ptr::null_mut(); + mf_frame.textures.y.handle.0 != null_ptr && mf_frame.textures.uv.handle.0 != null_ptr + }; + + let nv12_data = if has_valid_zero_copy_handles { + health.record_success(decode_time); + None + } else { + match decoder.read_texture_to_cpu( + &mf_frame.textures.nv12.texture, + mf_frame.width, + mf_frame.height, + ) { + Ok(data) => { + health.record_success(decode_time); + Some(Arc::new(data)) + } + Err(e) => { + health.record_texture_read_failure(); + warn!("Failed to read texture to CPU for frame {frame_number}: {e}"); + None + } + } + }; + + let cached = CachedFrame { + number: frame_number, + _texture: has_valid_zero_copy_handles.then(|| mf_frame.textures.nv12.texture.clone()), + _shared_handle: has_valid_zero_copy_handles.then_some(mf_frame.textures.nv12.handle), + _y_handle: has_valid_zero_copy_handles.then_some(mf_frame.textures.y.handle), + _uv_handle: has_valid_zero_copy_handles.then_some(mf_frame.textures.uv.handle), + nv12_data, + width: mf_frame.width, + height: mf_frame.height, + }; + + if !has_valid_zero_copy_handles { + decoder.recycle_textures(mf_frame.textures); + } + cached +} + +enum Step { + Seek(u32), + Decode, + Wait, +} + +type Answer = (R, Option); + +// The decoder thread's request policy, kept apart from Media Foundation so tests can drive +// it. A request is answered with the newest sample at or below its frame once the decoder +// has moved past that frame: a VFR stream can still have a later sample rounding to the same +// frame (which replaces it), so answering earlier would make the choice depend on request +// timing. Once the decoder is past a frame the stream has no sample for, the one before it is +// the hold the video really shows. +struct FrameRequests { + // Every sample since the last seek minus the oldest evicted, so the newest entry at or + // below a frame is the stream's newest sample there. + cache: BTreeMap, + cache_bytes: usize, + waiting: Vec<(u32, R)>, + seeked_to: Option, + readahead_frames: u32, + latest_request: Option, +} + +impl FrameRequests { + fn new(readahead_frames: u32) -> Self { + Self { + cache: BTreeMap::new(), + cache_bytes: 0, + waiting: Vec::new(), + seeked_to: None, + readahead_frames, + latest_request: None, + } + } + + fn newest(&self) -> Option { + self.cache.last_key_value().map(|(&frame, _)| frame) + } + + fn newest_at(&self, frame: u32) -> Option<&T> { + self.cache + .range(..=frame) + .next_back() + .map(|(_, (sample, _))| sample) + } + + // A seek lands on the keyframe before its target, which can be any distance below it. + fn position(&self) -> Option { + self.newest().max(self.seeked_to) + } + + fn request(&mut self, frame: u32, reply: R) -> Option> { + self.latest_request = Some(frame); + if self.newest().is_some_and(|newest| newest > frame) + && let Some(sample) = self.newest_at(frame) + { + return Some((reply, Some(sample.clone()))); + } + self.waiting.push((frame, reply)); + None + } + + // The lowest waiting frame goes first, so each request gets its own sample instead of + // whatever a seek for a later one lands on. Decoding forward only reaches frames from the + // newest sample on, so one below it seeks back. + fn next_step(&self) -> Step { + let Some(target) = self.waiting.iter().map(|(frame, _)| *frame).min() else { + return if self.wants_readahead() { + Step::Decode + } else { + Step::Wait + }; + }; + let reachable = self + .newest() + .or(self.seeked_to) + .zip(self.position()) + .is_some_and(|(from, position)| { + (from..=position.saturating_add(FRAME_CACHE_SIZE as u32)).contains(&target) + }); + if reachable { + Step::Decode + } else { + Step::Seek(target) + } + } + + // Readahead stops before the next sample would evict the one answering the latest + // request: with frames too big for the cache budget, that would send every request back + // to a seek. + fn wants_readahead(&self) -> bool { + let (Some(position), Some(latest)) = (self.position(), self.latest_request) else { + return false; + }; + let full = self.cache.last_key_value().is_some_and(|(_, (_, bytes))| { + self.cache.len() >= FRAME_CACHE_SIZE || self.cache_bytes + bytes > MAX_FRAME_CACHE_BYTES + }); + let evicts_shown = full && self.cache.range(..=latest).nth(1).is_none(); + position < latest.saturating_add(self.readahead_frames) && !evicts_shown + } + + fn seeked(&mut self, frame: u32) { + self.cache.clear(); + self.cache_bytes = 0; + self.seeked_to = Some(frame); + } + + // A waiting frame the decoder jumped past has no sample at or below it since the seek, so + // it gets the first one after. + fn decoded(&mut self, frame: u32, sample: T, bytes: usize) -> Vec> { + let (passed, waiting): (Vec<_>, _) = std::mem::take(&mut self.waiting) + .into_iter() + .partition(|(requested, _)| *requested < frame); + self.waiting = waiting; + let answers = passed + .into_iter() + .map(|(requested, reply)| { + let held = self.newest_at(requested).unwrap_or(&sample); + (reply, Some(held.clone())) + }) + .collect(); + + if let Some((_, replaced)) = self.cache.remove(&frame) { + self.cache_bytes -= replaced; + } + while !self.cache.is_empty() + && (self.cache.len() >= FRAME_CACHE_SIZE + || self.cache_bytes + bytes > MAX_FRAME_CACHE_BYTES) + { + if let Some((_, (_, evicted))) = self.cache.pop_first() { + self.cache_bytes -= evicted; + } + } + self.cache.insert(frame, (sample, bytes)); + self.cache_bytes += bytes; + answers + } + + // No sample came (end of stream or a read error), so nothing reads again until the next + // request. + fn stopped(&mut self) -> Vec> { + self.latest_request = None; + std::mem::take(&mut self.waiting) + .into_iter() + .map(|(frame, reply)| { + let nearest = self.newest_at(frame).or_else(|| { + self.cache + .range(frame..) + .next() + .map(|(_, (sample, _))| sample) + }); + (reply, nearest.cloned()) + }) + .collect() + } } #[cfg(test)] @@ -598,26 +581,218 @@ mod tests { assert_eq!(pts_100ns_to_frame(10_000_000, 30), 30); } + // Samples are their index in `frames`, which holds the frame each one rounds to. + struct FakeStream { + frames: Vec, + keyframe_every: usize, + next: usize, + reads: usize, + seeks: usize, + bytes: usize, + fail_next_read: bool, + } + + impl FakeStream { + fn new(frames: impl IntoIterator, keyframe_every: usize) -> Self { + Self { + frames: frames.into_iter().collect(), + keyframe_every, + next: 0, + reads: 0, + seeks: 0, + bytes: 1, + fail_next_read: false, + } + } + + fn step( + &mut self, + requests: &mut FrameRequests, + ) -> Option>> { + match requests.next_step() { + Step::Wait => None, + Step::Seek(frame) => { + let at = self.frames.iter().rposition(|&f| f <= frame).unwrap_or(0); + self.next = at - at % self.keyframe_every; + self.seeks += 1; + assert!(self.seeks < 100, "seek loop"); + requests.seeked(frame); + Some(Vec::new()) + } + Step::Decode => { + self.reads += 1; + Some(if std::mem::take(&mut self.fail_next_read) { + requests.stopped() + } else if let Some(&frame) = self.frames.get(self.next) { + self.next += 1; + requests.decoded(frame, self.next - 1, self.bytes) + } else { + requests.stopped() + }) + } + } + } + + fn run(&mut self, requests: &mut FrameRequests) -> Vec> { + let mut answers = Vec::new(); + while let Some(step) = self.step(requests) { + answers.extend(step); + } + answers + } + } + #[test] - fn an_undecoded_frame_is_not_answered_with_the_one_before_it() { - let cache = BTreeMap::from([(14, 'a'), (15, 'b')]); - assert_eq!(cached_frame_for(&cache, Some(15), 16), None); - assert_eq!(cached_frame_for(&cache, Some(15), 17), None); + fn a_frame_is_answered_once_the_decoder_moves_past_it() { + let mut requests = FrameRequests::new(0); + let mut stream = FakeStream::new(0..20, 1); + assert!(requests.request(15, 0).is_none()); + assert_eq!(stream.run(&mut requests), [(0, Some(15))]); + assert!(requests.request(16, 1).is_none()); + assert_eq!(requests.request(15, 2), Some((2, Some(15)))); + assert_eq!(stream.run(&mut requests), [(1, Some(16))]); } #[test] - fn the_newest_decoded_frame_waits_for_a_later_sample() { - let cache = BTreeMap::from([(14, 'a'), (15, 'b')]); - assert_eq!(cached_frame_for(&cache, Some(15), 14), Some(&'a')); - assert_eq!(cached_frame_for(&cache, Some(15), 15), None); - assert_eq!(cached_frame_for(&cache, Some(16), 15), Some(&'b')); + fn colliding_vfr_samples_answer_the_same_whenever_requested() { + let frames = [0, 1, 1, 2, 3]; + + let mut early = FrameRequests::new(0); + let mut stream = FakeStream::new(frames, 10); + early.request(1, 0); + assert_eq!(stream.run(&mut early), [(0, Some(2))]); + + let mut late = FrameRequests::new(0); + let mut stream = FakeStream::new(frames, 10); + late.request(0, 0); + assert_eq!(stream.run(&mut late), [(0, Some(0))]); + assert!(late.request(1, 1).is_none()); + assert_eq!(stream.run(&mut late), [(1, Some(2))]); } #[test] fn a_frame_the_stream_skipped_holds_the_previous_one() { - let cache = BTreeMap::from([(14, 'a'), (17, 'b')]); - assert_eq!(cached_frame_for(&cache, Some(17), 15), Some(&'a')); - assert_eq!(cached_frame_for(&cache, Some(17), 16), Some(&'a')); - assert_eq!(cached_frame_for(&cache, None, 16), None); + let mut requests = FrameRequests::new(0); + let mut stream = FakeStream::new([1, 4, 5], 10); + requests.request(0, 0); + requests.request(2, 1); + requests.request(3, 2); + assert_eq!( + stream.run(&mut requests), + [(0, Some(0)), (1, Some(0)), (2, Some(0))] + ); + } + + #[test] + fn readahead_slides_with_every_request() { + let mut requests = FrameRequests::new(5); + let mut stream = FakeStream::new(0..100, 30); + requests.request(0, 0); + assert_eq!(stream.run(&mut requests), [(0, Some(0))]); + for frame in 1..40 { + let sample = frame as usize; + assert_eq!( + requests.request(frame, sample), + Some((sample, Some(sample))) + ); + assert!(stream.run(&mut requests).is_empty()); + assert_eq!(requests.position(), Some(frame + 5)); + } + assert_eq!(stream.seeks, 1); + } + + #[test] + fn readahead_never_evicts_a_frame_still_being_shown() { + let frames: Vec = (0..=10).chain(50..=90).collect(); + let mut requests = FrameRequests::new(8); + let mut stream = FakeStream::new(frames.clone(), 100); + stream.bytes = MAX_FRAME_CACHE_BYTES / 2; + for frame in 0..70 { + let mut answers: Vec<_> = requests.request(frame, 0).into_iter().collect(); + answers.extend(stream.run(&mut requests)); + assert_eq!(answers, [(0, frames.iter().rposition(|&f| f <= frame))]); + } + assert_eq!(stream.seeks, 1); + } + + #[test] + fn every_waiting_request_gets_its_own_frame() { + let mut requests = FrameRequests::new(0); + let mut stream = FakeStream::new(0..400, 30); + requests.request(300, 0); + requests.request(30, 1); + assert_eq!(stream.run(&mut requests), [(1, Some(30)), (0, Some(300))]); + assert_eq!(stream.seeks, 2); + } + + #[test] + fn a_passed_frame_missing_from_the_cache_seeks_back() { + let mut requests = FrameRequests::new(30); + let mut stream = FakeStream::new(0..200, 10); + requests.request(100, 0); + assert_eq!(stream.run(&mut requests), [(0, Some(100))]); + assert!(requests.request(90, 1).is_none()); + assert_eq!(stream.run(&mut requests), [(1, Some(90))]); + assert_eq!(stream.seeks, 2); + } + + #[test] + fn far_requests_seek_and_near_ones_decode_through() { + let mut requests = FrameRequests::new(0); + let mut stream = FakeStream::new(0..400, 30); + requests.request(10, 0); + stream.run(&mut requests); + requests.request(90, 1); + assert_eq!(stream.run(&mut requests), [(1, Some(90))]); + assert_eq!(stream.seeks, 1); + requests.request(300, 2); + assert_eq!(stream.run(&mut requests), [(2, Some(300))]); + assert_eq!(stream.seeks, 2); + } + + #[test] + fn a_seek_decodes_through_a_long_gop_without_seeking_again() { + let mut requests = FrameRequests::new(0); + let mut stream = FakeStream::new(0..400, 200); + requests.request(150, 0); + assert_eq!(stream.run(&mut requests), [(0, Some(150))]); + assert_eq!(stream.seeks, 1); + } + + #[test] + fn a_request_inside_a_seeks_preroll_decodes_through() { + let mut requests = FrameRequests::new(0); + let mut stream = FakeStream::new(0..400, 30); + requests.request(310, 0); + stream.step(&mut requests); + stream.step(&mut requests); + assert_eq!(requests.newest(), Some(300)); + requests.request(305, 1); + assert_eq!(stream.run(&mut requests), [(1, Some(305)), (0, Some(310))]); + assert_eq!(stream.seeks, 1); + } + + #[test] + fn the_end_of_the_stream_holds_its_last_frame() { + let mut requests = FrameRequests::new(30); + let mut stream = FakeStream::new(0..5, 10); + requests.request(3, 0); + assert_eq!(stream.run(&mut requests), [(0, Some(3))]); + assert!(requests.request(4, 1).is_none()); + assert!(requests.request(60, 2).is_none()); + assert_eq!(stream.run(&mut requests), [(1, Some(4)), (2, Some(4))]); + } + + #[test] + fn a_read_error_answers_what_waits_and_the_next_request_retries() { + let mut requests = FrameRequests::new(0); + let mut stream = FakeStream::new(0..10, 10); + requests.request(0, 0); + stream.run(&mut requests); + requests.request(5, 1); + stream.fail_next_read = true; + assert_eq!(stream.run(&mut requests), [(1, Some(1))]); + requests.request(5, 2); + assert_eq!(stream.run(&mut requests), [(2, Some(5))]); } }