Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 70 additions & 2 deletions crates/rendering/src/decoder/avassetreader.rs
Original file line number Diff line number Diff line change
Expand Up @@ -380,6 +380,8 @@ struct DecoderInstance {
/// Lives on the instance because the reader vends in pts order between
/// resets, but request batches may be served by different pool decoders.
prev_vended: Option<u32>,
/// First frame of the uninterrupted vend run that ends at `prev_vended`.
run_start: Option<u32>,
_permit: OwnedSemaphorePermit,
}

Expand All @@ -405,6 +407,7 @@ impl DecoderInstance {
tokio_handle,
keyframe_index,
prev_vended: None,
run_start: None,
_permit: permit,
})
}
Expand All @@ -416,6 +419,7 @@ impl DecoderInstance {
self.frames_iter_valid = true;
self.health.reset_counters();
self.prev_vended = None;
self.run_start = None;
}
Err(e) => {
tracing::error!(
Expand Down Expand Up @@ -448,12 +452,21 @@ impl DecoderInstance {
self.frames_iter_valid = true;
self.health = DecoderHealth::new();
self.prev_vended = None;
self.run_start = None;
Ok(())
}

fn current_position(&self) -> f32 {
self.inner.current_position_secs()
}

fn would_serve_stale_frame(&self, frame: u32, cache: &BTreeMap<u32, CachedFrame>) -> bool {
reuse_would_serve_stale_frame(
self.run_start,
self.prev_vended,
cache.range(..=frame).next_back().map(|(&cached, _)| cached),
)
}
}

pub struct AVAssetReaderDecoder {
Expand Down Expand Up @@ -537,7 +550,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
Expand All @@ -552,6 +570,7 @@ impl AVAssetReaderDecoder {
};

let decoder_idx = best_id.min(decoder_count.saturating_sub(1));
let needs_reset = needs_reset || must_reset(&self.decoders[decoder_idx]);

let max_decoder_count = self
.pool_manager
Expand Down Expand Up @@ -890,7 +909,10 @@ 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);
let (decoder_idx, was_reset) =
this.select_best_decoder(requested_time, is_scrubbing, |decoder| {
decoder.would_serve_stale_frame(requested_frame, &cache)
});

let cache_min = if was_reset {
min_requested_frame.saturating_sub(FRAME_CACHE_SIZE as u32 * 2)
Expand Down Expand Up @@ -955,6 +977,9 @@ impl AVAssetReaderDecoder {
{
record_pts_hole(&mut pts_holes, prev, current_frame);
}
if decoder.prev_vended.is_none() {
decoder.run_start = Some(current_frame);
}
decoder.prev_vended = Some(current_frame);

let position_secs = current_frame as f32 / fps as f32;
Expand Down Expand Up @@ -1278,6 +1303,24 @@ impl AVAssetReaderDecoder {
}
}

// Requests the reader has already passed, or that fall in a gap before its
// next frame, are answered with the newest cached frame at or before them.
// That frame is only trustworthy when this reader vended it in its current
// run; an older one comes from an unrelated seek and shows stale content.
// With nothing cached at or before the request, whether the reader has
// passed it or a gap follows its evicted last frame, the only answer left is
// a later frame instead of the one on screen.
fn reuse_would_serve_stale_frame(
run_start: Option<u32>,
prev_vended: Option<u32>,
newest_cached_at_or_before: Option<u32>,
) -> bool {
let (Some(run_start), Some(_)) = (run_start, prev_vended) else {
return false;
};
newest_cached_at_or_before.is_none_or(|cached| cached < run_start)
}

fn should_grow_decoder_pool(
is_scrubbing: bool,
needs_reset: bool,
Expand Down Expand Up @@ -1357,6 +1400,31 @@ mod tests {
));
}

#[test]
fn reused_readers_never_answer_with_frames_from_an_unrelated_seek() {
assert!(!reuse_would_serve_stale_frame(None, None, Some(10)));
assert!(!reuse_would_serve_stale_frame(Some(40), Some(55), Some(48)));
assert!(!reuse_would_serve_stale_frame(Some(40), Some(55), Some(40)));
assert!(reuse_would_serve_stale_frame(Some(40), Some(55), Some(39)));
assert!(reuse_would_serve_stale_frame(Some(40), Some(55), Some(10)));
assert!(reuse_would_serve_stale_frame(Some(40), Some(48), None));
}

#[test]
fn returning_to_a_passed_hold_after_its_frame_is_evicted_resets_the_reader() {
let (run_start, held, vended) = (40, 48, 55);
assert!(!reuse_would_serve_stale_frame(
Some(run_start),
Some(vended),
Some(held)
));
assert!(reuse_would_serve_stale_frame(
Some(run_start),
Some(vended),
None
));
}

#[test]
fn decoder_pool_grows_only_for_unserved_non_sequential_access() {
assert!(should_grow_decoder_pool(true, true, 1, 5));
Expand Down
68 changes: 53 additions & 15 deletions crates/video-decode/src/avassetreader.rs
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ use tokio::runtime::Handle as TokioHandle;
#[derive(Clone)]
pub struct KeyframeIndex {
keyframes: Vec<(u32, f64)>,
time_base: ffmpeg::Rational,
fps: f64,
duration_secs: f64,
pixel_format: Option<cv::PixelFormat>,
Expand Down Expand Up @@ -271,6 +272,7 @@ impl KeyframeIndex {

Ok(Self {
keyframes,
time_base,
fps,
duration_secs,
pixel_format,
Expand Down Expand Up @@ -374,20 +376,44 @@ impl KeyframeIndex {
pub fn cached_video_info(&self) -> Option<(cv::PixelFormat, u32, u32)> {
self.pixel_format.map(|pf| (pf, self.width, self.height))
}
}

fn compute_seek_time(keyframe_index: Option<&Arc<KeyframeIndex>>, requested_time: f32) -> f32 {
if let Some(kf_index) = keyframe_index {
let keyframes = kf_index.keyframes();
// Rounded frame numbers can alias keyframes across a VFR gap and seek
// past the image that should still be held at the requested time.
let after_request =
keyframes.partition_point(|(_, time)| *time <= f64::from(requested_time));
if let Some((_, keyframe_time)) = keyframes.get(after_request.saturating_sub(1)) {
return *keyframe_time as f32;
// The reader decodes from the sync sample at or before its start time, so
// a start rounded to just below a keyframe decodes the whole previous GOP
// and vends its last frame retimed to the start. Starting on the exact
// stream tick keeps the decode on the requested keyframe.
fn exact_time(&self, secs: f64) -> cm::Time {
let numerator = i64::from(self.time_base.numerator());
let denominator = self.time_base.denominator();
if numerator <= 0 || denominator <= 0 {
return cm::Time::with_secs(secs, 100);
}
let ticks = (secs * f64::from(denominator) / numerator as f64).round() as i64;
cm::Time::new(ticks * numerator, denominator)
}
requested_time
}

fn seek_keyframe_time(keyframe_index: &KeyframeIndex, requested_time: f32) -> Option<f64> {
let keyframes = keyframe_index.keyframes();
// Rounded frame numbers can alias keyframes across a VFR gap and seek
// past the image that should still be held at the requested time.
let after_request = keyframes.partition_point(|(_, time)| *time <= f64::from(requested_time));
keyframes
.get(after_request.saturating_sub(1))
.map(|(_, time)| *time)
}

fn compute_seek_time(keyframe_index: Option<&Arc<KeyframeIndex>>, requested_time: f32) -> f32 {
keyframe_index
.and_then(|index| seek_keyframe_time(index, requested_time))
.map_or(requested_time, |time| time as f32)
}

fn reader_start_time(keyframe_index: Option<&Arc<KeyframeIndex>>, requested_time: f32) -> cm::Time {
keyframe_index
.and_then(|index| {
seek_keyframe_time(index, requested_time).map(|time| index.exact_time(time))
})
.unwrap_or_else(|| cm::Time::with_secs(requested_time as f64, 100))
}

pub struct AVAssetReaderDecoder {
Expand Down Expand Up @@ -464,7 +490,7 @@ impl AVAssetReaderDecoder {

let (track_output, reader) = Self::get_reader_track_output(
&path,
seek_time,
reader_start_time(keyframe_index.as_ref(), start_time),
&tokio_handle,
pixel_format,
width,
Expand All @@ -491,7 +517,7 @@ impl AVAssetReaderDecoder {

(self.track_output, self.reader) = Self::get_reader_track_output(
&self.path,
seek_time,
reader_start_time(self.keyframe_index.as_ref(), requested_time),
&self.tokio_handle,
self.pixel_format,
self.width,
Expand Down Expand Up @@ -533,7 +559,7 @@ impl AVAssetReaderDecoder {

fn get_reader_track_output(
path: &Path,
time: f32,
start: cm::Time,
handle: &TokioHandle,
pixel_format: cv::PixelFormat,
width: u32,
Expand All @@ -553,7 +579,7 @@ impl AVAssetReaderDecoder {
.map_err(|e| format!("AssetReader::with_asset: {e}"))?;

let time_range = cm::TimeRange {
start: cm::Time::with_secs(time as f64, 100),
start,
duration: cm::Time::infinity(),
};

Expand Down Expand Up @@ -664,6 +690,7 @@ mod tests {
fn seek_index(keyframes: Keyframes) -> Arc<KeyframeIndex> {
Arc::new(KeyframeIndex {
keyframes,
time_base: ffmpeg::Rational::new(1, 600),
fps: 2.0,
duration_secs: 8.0,
pixel_format: None,
Expand Down Expand Up @@ -696,6 +723,17 @@ mod tests {
assert_eq!(compute_seek_time(None, 3.0), 3.0);
}

#[test]
fn reader_starts_on_the_exact_keyframe_tick() {
let index = seek_index(vec![(0, 0.0), (1, 577.0 / 600.0), (3, 1147.0 / 600.0)]);
let start = reader_start_time(Some(&index), 1.5);
assert_eq!((start.value, start.scale), (577, 600));
let start = reader_start_time(Some(&index), 1.92);
assert_eq!((start.value, start.scale), (1147, 600));
let start = reader_start_time(None, 1.5);
assert_eq!((start.value, start.scale), (150, 100));
}

fn both_indexes(path: &Path) -> Option<(Option<Keyframes>, Keyframes)> {
ffmpeg::init().unwrap();
let mut input = avformat::input(path).ok()?;
Expand Down
Loading