diff --git a/.github/workflows/sync-tests.yml b/.github/workflows/sync-tests.yml index 1b6aa450898..1228d40e528 100644 --- a/.github/workflows/sync-tests.yml +++ b/.github/workflows/sync-tests.yml @@ -33,6 +33,8 @@ on: - "scripts/recording-reliability.py" - "scripts/recording-reliability-owned-process.cs" - "scripts/test-recording-reliability.py" + - "scripts/capbench.py" + - "scripts/test-capbench.py" - ".github/workflows/sync-tests.yml" concurrency: @@ -99,6 +101,7 @@ jobs: run: | cargo test --locked -p cap-project --lib python -B scripts/test-recording-reliability.py + python -B scripts/test-capbench.py - name: Compile recording process supervisor (Windows) if: runner.os == 'Windows' diff --git a/apps/desktop-gpui/Cargo.toml b/apps/desktop-gpui/Cargo.toml index 50fccce29ad..5fb47546dde 100644 --- a/apps/desktop-gpui/Cargo.toml +++ b/apps/desktop-gpui/Cargo.toml @@ -184,7 +184,6 @@ raw-window-handle = "0.6" tray-icon = { version = "0.21.1", default-features = false } windows-sys = { version = "0.59", features = [ "Win32_Foundation", - "Win32_Media", "Win32_Security", "Win32_System_Registry", "Win32_System_SystemInformation", diff --git a/apps/desktop-gpui/src/editor_export.rs b/apps/desktop-gpui/src/editor_export.rs index abbf6d98e15..9d9ee19a795 100644 --- a/apps/desktop-gpui/src/editor_export.rs +++ b/apps/desktop-gpui/src/editor_export.rs @@ -2863,26 +2863,6 @@ fn matches_compression(current: ExportCompression, expected: ExportCompression) std::mem::discriminant(¤t) == std::mem::discriminant(&expected) } -// The NV12 readback wait sleeps in 1 ms steps (`frame_pipeline.rs`), which the -// default 15.6 ms Windows timer resolution stretches to most of a frame each. -#[cfg(windows)] -struct TimerResolution; - -#[cfg(windows)] -impl TimerResolution { - fn hold_one_millisecond() -> Self { - unsafe { windows_sys::Win32::Media::timeBeginPeriod(1) }; - Self - } -} - -#[cfg(windows)] -impl Drop for TimerResolution { - fn drop(&mut self) { - unsafe { windows_sys::Win32::Media::timeEndPeriod(1) }; - } -} - #[allow(clippy::too_many_arguments)] async fn run_export( project_path: PathBuf, @@ -2903,9 +2883,6 @@ async fn run_export( if cancel.load(Ordering::Relaxed) { return Err("Export cancelled".into()); } - #[cfg(windows)] - let _timer_resolution = TimerResolution::hold_one_millisecond(); - let mut builder = ExporterBase::builder(project_path).with_force_ffmpeg_decoder(force); if cursor_only { builder = builder.with_config(make_cursor_only_project(project)); diff --git a/crates/editor/examples/playback-pipeline-benchmark.rs b/crates/editor/examples/playback-pipeline-benchmark.rs index 8e772d10372..bca3bc02fa4 100644 --- a/crates/editor/examples/playback-pipeline-benchmark.rs +++ b/crates/editor/examples/playback-pipeline-benchmark.rs @@ -849,6 +849,12 @@ async fn main() { .and_then(|s| s.parse().ok()) .unwrap_or(300); let force_ffmpeg = args.iter().any(|arg| arg == "--force-ffmpeg"); + // The editor holds a 1ms timer resolution while playing; this measures under the same condition. + let timer_1ms = args.iter().any(|arg| arg == "--timer-1ms"); + #[cfg(target_os = "windows")] + if timer_1ms { + unsafe { windows::Win32::Media::timeBeginPeriod(1) }; + } println!("{}", "=".repeat(60)); println!(" CAP PLAYBACK PIPELINE BENCHMARK"); @@ -858,6 +864,7 @@ async fn main() { println!("Target FPS: {fps}"); println!("Max frames: {frame_count}"); println!("Force FFmpeg: {force_ffmpeg}"); + println!("1ms timer: {timer_1ms}"); println!(); let (recording_meta, meta, project, recordings) = match load_recording(&recording_path).await { diff --git a/crates/export/src/mp4.rs b/crates/export/src/mp4.rs index 4bb938d1c3f..13d78a81e8f 100644 --- a/crates/export/src/mp4.rs +++ b/crates/export/src/mp4.rs @@ -1477,6 +1477,7 @@ async fn export_render_to_channel( let mut first_frame_data: Option = None; let mut frame_count = 0u32; let mut consecutive_timeouts = 0u32; + let mut encoder_blocked = Duration::ZERO; loop { let timeout_secs = if frame_count == 0 { @@ -1539,7 +1540,10 @@ async fn export_render_to_channel( }); } - if sender.send(export_frame).is_err() { + let send_started = std::time::Instant::now(); + let sent = sender.send(export_frame); + encoder_blocked += send_started.elapsed(); + if sent.is_err() { warn!("Encoder dropped, stopping render forwarding"); if stop_on_encoder_drop { return Err(cap_rendering::RenderingError::ImageLoadError( @@ -1568,13 +1572,32 @@ async fn export_render_to_channel( }); } - Ok::<_, cap_rendering::RenderingError>(()) + Ok::<_, cap_rendering::RenderingError>(encoder_blocked) }; + cap_rendering::loop_stats::take(); tokio::try_join!(render_future, forward_future) }; - render_result?; + let (_, encoder_blocked) = render_result?; + + // The loop_stats counters are process-wide, so an export that overlaps another one reports + // a mix of both. That is fine for profiling one export at a time; per-export numbers need + // the counters passed through the render loop instead of kept in statics. + if cap_rendering::loop_stats::enabled() { + let [decode, render, prefetch, join, render_send, frames] = + cap_rendering::loop_stats::take(); + info!( + frames, + decode_ms = decode / 1000, + render_ms = render / 1000, + prefetch_ms = prefetch / 1000, + join_ms = join / 1000, + render_send_ms = render_send / 1000, + encoder_send_ms = encoder_blocked.as_millis() as u64, + "NV12 export loop stats" + ); + } Ok(()) } diff --git a/crates/recording/src/cursor.rs b/crates/recording/src/cursor.rs index fb3ad69cc8f..1962fe3e122 100644 --- a/crates/recording/src/cursor.rs +++ b/crates/recording/src/cursor.rs @@ -770,7 +770,7 @@ fn cursor_id_for_image( cursor_id } -#[derive(Debug)] +#[derive(Debug, Clone)] struct CursorData { image: Vec, hotspot: XY, @@ -992,6 +992,23 @@ fn get_cursor_data() -> Option { return None; } + // Rendering, trimming and PNG-encoding the cursor every tick only to hash it is wasted + // while the shape is unchanged. Keyed on the handle, so a cursor scheme change that keeps + // the system cursor handles shows the old image until the shape changes; key on the + // bitmap contents instead if that ever matters. + thread_local! { + static LAST_CURSOR: std::cell::RefCell> = + const { std::cell::RefCell::new(None) }; + } + let handle = cursor_info.hCursor.0 as usize; + if let Some(cached) = LAST_CURSOR.with_borrow(|last| { + last.as_ref() + .filter(|(last_handle, _)| *last_handle == handle) + .map(|(_, data)| data.clone()) + }) { + return Some(cached); + } + // Get icon info let mut icon_info = ICONINFO::default(); if GetIconInfo(cursor_info.hCursor.into(), &mut icon_info).is_err() { @@ -1265,11 +1282,13 @@ fn get_cursor_data() -> Option { ) .ok()?; - Some(CursorData { + let data = CursorData { image: png_data, hotspot: XY::new(hotspot_x, hotspot_y), shape: CursorShape::try_from(&cursor_info.hCursor).ok(), - }) + }; + LAST_CURSOR.set(Some((handle, data.clone()))); + Some(data) } } diff --git a/crates/rendering/src/decoder/media_foundation.rs b/crates/rendering/src/decoder/media_foundation.rs index 9eed6825637..6d77001006d 100644 --- a/crates/rendering/src/decoder/media_foundation.rs +++ b/crates/rendering/src/decoder/media_foundation.rs @@ -231,9 +231,6 @@ impl MFDecoder { let video_height = decoder.height(); let matrix = super::probe_yuv_matrix(&path); - 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 { @@ -243,293 +240,106 @@ impl MFDecoder { }; let _ = ready_tx.send(Ok(init_result)); - struct PendingRequest { - frame: u32, - time: f32, - sender: oneshot::Sender, - } - - while let Ok(r) = rx.recv() { - 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().with_matrix(matrix), + )); + }; - let mut push_request = - |requested_time: f32, sender: oneshot::Sender| { - if sender.is_closed() { - return; - } - let frame = super::request_frame(requested_time, fps); - 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 = super::request_frame(time, fps); + 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().with_matrix(matrix)); - } 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 cache_min = requested_frame.saturating_sub(FRAME_CACHE_SIZE as u32 / 2); - let 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 batch_start = Instant::now(); - - 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; - } - - 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); - } - - 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); - } - - if frame_number <= requested_frame { - last_valid_frame = Some(cached); - } - - if frame_number >= requested_frame { - let frame_to_send = if frame_number == requested_frame { - cache.get(&requested_frame) - } else { - 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().with_matrix(matrix)); - } + Err(e) => { + warn!("MediaFoundation seek failed: {e}"); + requests.stopped() + } + }, + 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) } - - let readahead_target = requested_frame + readahead_frames; - if frame_number >= readahead_target || frame_number > cache_max { - break; + Ok(None) => requests.stopped(), + Err(e) => { + health.record_error(); + warn!( + consecutive_errors = health.consecutive_errors, + "MediaFoundation read_sample error: {e}" + ); + requests.stopped() } } - 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().with_matrix(matrix)); - } 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().with_matrix(matrix)); - } 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); } } }); @@ -549,21 +359,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 reached yet has to be decoded; answering with -// the cached frame before it shows a stale image at every readahead boundary. -// 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> { - cache.get(&frame).or_else(|| { - 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)] @@ -578,19 +582,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 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 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), 15), Some(&'b')); - assert_eq!(cached_frame_for(&cache, Some(15), 16), None); - assert_eq!(cached_frame_for(&cache, Some(15), 17), None); + 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))]); } } diff --git a/crates/rendering/src/frame_pipeline.rs b/crates/rendering/src/frame_pipeline.rs index 0c8461a52c0..249d222df96 100644 --- a/crates/rendering/src/frame_pipeline.rs +++ b/crates/rendering/src/frame_pipeline.rs @@ -16,7 +16,6 @@ use foreign_types::ForeignType; const GPU_BUFFER_WAIT_TIMEOUT_SECS: u64 = 10; const SOFTWARE_GPU_BUFFER_WAIT_TIMEOUT_SECS: u64 = 60; -const NV12_READBACK_SPIN: std::time::Duration = std::time::Duration::from_millis(8); /// Whether this process has selected a software (CPU rasterizer) wgpu adapter. /// Software adapters like Windows WARP legitimately take tens of seconds for the @@ -41,6 +40,50 @@ fn gpu_buffer_wait_timeout(export_deadlines: bool) -> std::time::Duration { std::time::Duration::from_secs(secs) } +enum MapWait { + Mapped, + Failed(wgpu::BufferAsyncError), + Abandoned, +} + +// Blocking on the submission's fence wakes the moment the GPU signals it. Sleep-polling +// instead rounds every wait up to the OS timer tick (~15.6ms on Windows outside +// timeBeginPeriod), which capped preview renders at ~25ms regardless of resolution. +async fn wait_for_map( + device: &wgpu::Device, + submission: &wgpu::SubmissionIndex, + rx: &mut oneshot::Receiver>, + export_deadlines: bool, + frame_number: u32, +) -> Result { + let Some(budget) = gpu_buffer_wait_timeout(export_deadlines) + .checked_sub(readback_waited(Instant::now(), frame_number)) + else { + return Ok(MapWait::Abandoned); + }; + let device = device.clone(); + let submission = submission.clone(); + let waited = tokio::time::timeout( + budget, + tokio::task::spawn_blocking(move || { + device.poll(wgpu::PollType::WaitForSubmissionIndex(submission)) + }), + ) + .await; + match waited { + Ok(Ok(result)) => { + result?; + } + _ => return Ok(MapWait::Abandoned), + } + // The wait fires the map callbacks of every submission up to this one. + Ok(match rx.try_recv() { + Ok(Ok(())) => MapWait::Mapped, + Ok(Err(error)) => MapWait::Failed(error), + Err(_) => MapWait::Abandoned, + }) +} + /// How long the readback of `frame_number` has been waited for. fn readback_waited(started: Instant, frame_number: u32) -> std::time::Duration { #[cfg(test)] @@ -862,7 +905,7 @@ impl RgbaToNv12Converter { encoder.copy_buffer_to_buffer(nv12_buffer, 0, &readback_buffer, 0, nv12_size); self.pending = Some(PendingNv12Output::Readback(PendingNv12Readback { - rx: None, + armed: None, mapped: false, buffer: readback_buffer, width, @@ -879,7 +922,7 @@ impl RgbaToNv12Converter { /// Arms the pending output after its command buffer has been submitted: /// readback mode starts the buffer map, surface mode registers a /// native queue completion signal. - pub fn after_submit(&mut self, queue: &wgpu::Queue) { + pub fn after_submit(&mut self, queue: &wgpu::Queue, submission: &wgpu::SubmissionIndex) { match self.pending.as_mut() { Some(PendingNv12Output::Readback(pending)) => { let (tx, rx) = oneshot::channel(); @@ -889,7 +932,7 @@ impl RgbaToNv12Converter { .map_async(wgpu::MapMode::Read, move |result| { let _ = tx.send(result); }); - pending.rx = Some(rx); + pending.armed = Some((rx, submission.clone())); pending.mapped = true; } #[cfg(target_os = "macos")] @@ -1085,7 +1128,10 @@ impl PendingNv12Surface { } pub struct PendingNv12Readback { - rx: Option>>, + armed: Option<( + oneshot::Receiver>, + wgpu::SubmissionIndex, + )>, mapped: bool, buffer: Arc, pub width: u32, @@ -1114,41 +1160,25 @@ impl PendingNv12Readback { device: &wgpu::Device, buffer_pool: Option<&mut NV12BufferPool>, ) -> Result { - let Some(mut rx) = self.rx.take() else { + let Some((mut rx, submission)) = self.armed.take() else { return Err(self.cancel()); }; - let start_time = Instant::now(); - let timeout_duration = gpu_buffer_wait_timeout(self.export_deadlines); - - loop { - if readback_waited(start_time, self.frame_number) > timeout_duration { - return Err(self.cancel()); - } - - match rx.try_recv() { - Ok(result) => match result { - Ok(()) => break, - Err(error) => { - self.unmap(); - return Err(RenderingError::BufferMapFailed(error)); - } - }, - Err(oneshot::error::TryRecvError::Empty) => { - device.poll(wgpu::PollType::Poll)?; - // tokio's timer ticks in whole milliseconds, so even a - // 100µs sleep outlasts a frame's GPU work and caps the - // export at the timer rate; sleep only once a frame is slow. - if start_time.elapsed() < NV12_READBACK_SPIN { - tokio::task::yield_now().await; - } else { - tokio::time::sleep(std::time::Duration::from_millis(1)).await; - } - } - Err(oneshot::error::TryRecvError::Closed) => { - return Err(self.cancel()); - } + match wait_for_map( + device, + &submission, + &mut rx, + self.export_deadlines, + self.frame_number, + ) + .await? + { + MapWait::Mapped => {} + MapWait::Failed(error) => { + self.unmap(); + return Err(RenderingError::BufferMapFailed(error)); } + MapWait::Abandoned => return Err(self.cancel()), } let buffer_slice = self.buffer.slice(..); @@ -1631,6 +1661,7 @@ impl PendingSurface { pub struct PendingReadback { readiness_phase: Option, rx: oneshot::Receiver>, + submission: wgpu::SubmissionIndex, mapped: bool, buffer: Arc, padded_bytes_per_row: u32, @@ -1665,53 +1696,29 @@ impl PendingReadback { } pub async fn wait(mut self, device: &wgpu::Device) -> Result { - let mut poll_count = 0u32; let start_time = Instant::now(); - let timeout_duration = gpu_buffer_wait_timeout(self.export_deadlines); - - loop { - if readback_waited(start_time, self.frame_number) > timeout_duration { + match wait_for_map( + device, + &self.submission, + &mut self.rx, + self.export_deadlines, + self.frame_number, + ) + .await? + { + MapWait::Mapped => {} + MapWait::Failed(error) => { + self.unmap(); + return Err(RenderingError::BufferMapFailed(error)); + } + MapWait::Abandoned => { tracing::error!( frame_number = self.frame_number, - elapsed_secs = start_time.elapsed().as_secs(), - poll_count = poll_count, - "GPU buffer mapping timed out after {}s", - timeout_duration.as_secs() + elapsed_ms = start_time.elapsed().as_millis() as u64, + "GPU buffer mapping timed out or was abandoned" ); return Err(self.cancel()); } - - match self.rx.try_recv() { - Ok(result) => match result { - Ok(()) => break, - Err(error) => { - self.unmap(); - return Err(RenderingError::BufferMapFailed(error)); - } - }, - Err(oneshot::error::TryRecvError::Empty) => { - device.poll(wgpu::PollType::Poll)?; - poll_count += 1; - if poll_count < 10 { - tokio::task::yield_now().await; - } else if poll_count < 100 { - tokio::time::sleep(std::time::Duration::from_micros(100)).await; - } else { - tokio::time::sleep(std::time::Duration::from_millis(1)).await; - } - if poll_count.is_multiple_of(10000) { - tracing::warn!( - frame_number = self.frame_number, - poll_count = poll_count, - elapsed_ms = start_time.elapsed().as_millis() as u64, - "GPU buffer mapping taking longer than expected" - ); - } - } - Err(oneshot::error::TryRecvError::Closed) => { - return Err(self.cancel()); - } - } } if let Some(phase) = &self.readiness_phase { @@ -1899,7 +1906,7 @@ impl PipelinedGpuReadback { if let Some(phase) = &phase { phase.mark("submit_start"); } - queue.submit(std::iter::once(render_encoder.finish())); + let submission = queue.submit(std::iter::once(render_encoder.finish())); if let Some(phase) = &phase { phase.mark("submit_returned"); } @@ -1917,6 +1924,7 @@ impl PipelinedGpuReadback { readiness_phase: phase, mapped: true, rx, + submission, buffer, padded_bytes_per_row, width: uniforms.output_size.0, @@ -2257,9 +2265,9 @@ pub async fn finish_encoder_nv12_pooled( })); } - queue.submit(std::iter::once(encoder.finish())); + let submission = queue.submit(std::iter::once(encoder.finish())); if submitted { - nv12_converter.after_submit(queue); + nv12_converter.after_submit(queue, &submission); } if let Some(prev) = previous { @@ -2460,8 +2468,8 @@ mod surface_output_tests { .await .expect("conversion submission") ); - queue.submit(std::iter::once(encoder.finish())); - converter.after_submit(queue); + let submission = queue.submit(std::iter::once(encoder.finish())); + converter.after_submit(queue, &submission); converter .take_pending() .expect("conversion pending") @@ -2486,8 +2494,8 @@ mod surface_output_tests { .await .expect("surface conversion") ); - queue.submit([encoder.finish()]); - converter.after_submit(&queue); + let submission = queue.submit([encoder.finish()]); + converter.after_submit(&queue, &submission); let Some(PendingNv12Output::Surface(mut pending)) = converter.take_pending() else { panic!("expected surface output"); }; diff --git a/crates/rendering/src/lib.rs b/crates/rendering/src/lib.rs index dfe7030cf43..3dd1bf0a376 100644 --- a/crates/rendering/src/lib.rs +++ b/crates/rendering/src/lib.rs @@ -259,9 +259,12 @@ fn bundled_dxc_compiler() -> wgpu::Dx12Compiler { fn bundled_dxc_paths() -> Option<(PathBuf, PathBuf)> { let executable = std::env::current_exe().ok()?; let executable_dir = executable.parent()?; - let profile_dir = (executable_dir.file_name()? == "deps") - .then(|| executable_dir.parent()) - .flatten(); + let profile_dir = matches!( + executable_dir.file_name()?.to_str(), + Some("deps" | "examples") + ) + .then(|| executable_dir.parent()) + .flatten(); [Some(executable_dir), profile_dir] .into_iter() diff --git a/crates/rendering/src/yuv_converter.rs b/crates/rendering/src/yuv_converter.rs index ce29ac62541..f7e8a798370 100644 --- a/crates/rendering/src/yuv_converter.rs +++ b/crates/rendering/src/yuv_converter.rs @@ -2022,8 +2022,8 @@ mod colour_matrix_tests { .await .expect("conversion submission") ); - queue.submit(std::iter::once(encoder.finish())); - converter.after_submit(&queue); + let submission = queue.submit(std::iter::once(encoder.finish())); + converter.after_submit(&queue, &submission); let frame = converter .take_pending() .expect("conversion pending") diff --git a/crates/scap-ffmpeg/src/direct3d.rs b/crates/scap-ffmpeg/src/direct3d.rs index fcfdd34fa6c..762af9c58b9 100644 --- a/crates/scap-ffmpeg/src/direct3d.rs +++ b/crates/scap-ffmpeg/src/direct3d.rs @@ -46,6 +46,65 @@ fn copy_frame_data( } } +struct FramePool { + key: (Pixel, u32, u32), + linesize: usize, + pool: *mut ffmpeg::ffi::AVBufferPool, +} + +impl Drop for FramePool { + fn drop(&mut self) { + // Frees the pool once every buffer still referenced by an in-flight frame is returned. + unsafe { ffmpeg::ffi::av_buffer_pool_uninit(&mut self.pool) }; + } +} + +thread_local! { + static FRAME_POOL: std::cell::RefCell> = const { std::cell::RefCell::new(None) }; +} + +// A fresh multi-megabyte allocation per frame is served by fresh OS pages on Windows, so every +// capture paid page-fault and zeroing costs several times the copy itself (~2ms per 1080p frame). +fn pooled_frame(format: Pixel, width: u32, height: u32) -> ffmpeg::frame::Video { + debug_assert!(matches!(format, Pixel::RGBA | Pixel::BGRA)); + FRAME_POOL.with_borrow_mut(|slot| { + if slot + .as_ref() + .is_none_or(|pool| pool.key != (format, width, height)) + { + *slot = None; + let linesize = (width as usize).next_multiple_of(32) * 4; + let size = + linesize * height as usize + ffmpeg::ffi::AV_INPUT_BUFFER_PADDING_SIZE as usize; + let pool = unsafe { ffmpeg::ffi::av_buffer_pool_init(size, None) }; + if pool.is_null() { + return ffmpeg::frame::Video::new(format, width, height); + } + *slot = Some(FramePool { + key: (format, width, height), + linesize, + pool, + }); + } + let pool = slot.as_ref().expect("frame pool initialized above"); + unsafe { + let buffer = ffmpeg::ffi::av_buffer_pool_get(pool.pool); + if buffer.is_null() { + return ffmpeg::frame::Video::new(format, width, height); + } + let mut frame = ffmpeg::frame::Video::empty(); + frame.set_format(format); + frame.set_width(width); + frame.set_height(height); + let raw = frame.as_mut_ptr(); + (*raw).buf[0] = buffer; + (*raw).data[0] = (*buffer).data; + (*raw).linesize[0] = pool.linesize as i32; + frame + } + }) +} + impl super::AsFFmpeg for scap_direct3d::Frame { fn as_ffmpeg(&self) -> Result { let buffer = self.as_buffer()?; @@ -59,11 +118,7 @@ impl super::AsFFmpeg for scap_direct3d::Frame { match self.pixel_format() { PixelFormat::R8G8B8A8Unorm => { - let mut ff_frame = ffmpeg::frame::Video::new( - ffmpeg::format::Pixel::RGBA, - self.width(), - self.height(), - ); + let mut ff_frame = pooled_frame(Pixel::RGBA, self.width(), self.height()); let dest_stride = ff_frame.stride(0); let dest_bytes = ff_frame.data_mut(0); @@ -80,11 +135,7 @@ impl super::AsFFmpeg for scap_direct3d::Frame { Ok(ff_frame) } PixelFormat::B8G8R8A8Unorm => { - let mut ff_frame = ffmpeg::frame::Video::new( - ffmpeg::format::Pixel::BGRA, - self.width(), - self.height(), - ); + let mut ff_frame = pooled_frame(Pixel::BGRA, self.width(), self.height()); let dest_stride = ff_frame.stride(0); let dest_bytes = ff_frame.data_mut(0); @@ -116,3 +167,31 @@ impl PixelFormatExt for PixelFormat { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn pooled_frames_recycle_released_buffers() { + let first = pooled_frame(Pixel::BGRA, 1001, 3); + assert_eq!( + (first.format(), first.width(), first.height()), + (Pixel::BGRA, 1001, 3) + ); + assert!(first.stride(0) >= 1001 * 4); + assert!(first.data(0).len() >= first.stride(0) * 3); + let held = first.data(0).as_ptr(); + + let second = pooled_frame(Pixel::BGRA, 1001, 3); + assert_ne!(second.data(0).as_ptr(), held); + + drop(first); + let third = pooled_frame(Pixel::BGRA, 1001, 3); + assert_eq!(third.data(0).as_ptr(), held); + + let resized = pooled_frame(Pixel::BGRA, 640, 2); + assert_eq!((resized.width(), resized.height()), (640, 2)); + drop((second, third)); + } +} 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 { diff --git a/scripts/capbench.py b/scripts/capbench.py new file mode 100644 index 00000000000..f270e14b661 --- /dev/null +++ b/scripts/capbench.py @@ -0,0 +1,849 @@ +#!/usr/bin/env python3 +"""Repeatable export / playback / decode / recording benchmarks for the Rust media pipeline. + +Fixture: clone CapSoftware/cap-performance-fixtures into ./performance-fixtures (as CI does). +Build the measured binaries into target/perf (release codegen; see `profile.perf` below): + + cargo build --profile perf -p cap --bin cap -p cap-test --bin cap-test -p cap-editor \\ + --example playback-pipeline-benchmark --example decode-benchmark --example editor-startup-benchmark + +`profile.perf` is not checked in; add it to .cargo/config.toml (or use --bin-dir target/release): + + [profile.perf] + inherits = "release" + codegen-units = 16 + lto = false + +Then, e.g.: + + python scripts/capbench.py run offline --out bench/base + python scripts/capbench.py run offline --variant base=bins/base --variant cand=bins/cand --out bench/ab + python scripts/capbench.py run record --screen --animate X,Y,W,H --out bench/rec + python scripts/capbench.py compare bench/base/run.json bench/cand/run.json + +On Windows each benchmark runs in a job object (whole-process-tree CPU time and peak commit) +at above-normal priority, so background load disturbs both variants less; absolute numbers are +therefore slightly optimistic. Variants are interleaved per repetition with alternating order, +each variant gets its own copy of the fixture, and comparisons report the median change with a +bootstrap 95% interval (verdicts need at least 5 runs per side). +""" +import argparse +import ctypes +import json +import os +from pathlib import Path +import platform +import random +import re +import shutil +import statistics +import subprocess +import sys +import threading +import time + +ROOT = Path(__file__).resolve().parent.parent +DEFAULT_FIXTURE = ROOT / "performance-fixtures" / "reference-recording.cap" +EXE = ".exe" if os.name == "nt" else "" + + +class Job: + def __init__(self, affinity=None): + self.handle = None + if os.name != "nt": + return + from ctypes import wintypes + + self.kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + self.kernel32.CreateJobObjectW.restype = wintypes.HANDLE + self.handle = self.kernel32.CreateJobObjectW(None, None) + if not self.handle: + raise OSError(ctypes.get_last_error(), "CreateJobObjectW failed") + limits = JobExtendedLimits() + limits.BasicLimitInformation.LimitFlags = 0x2000 | 0x20 + limits.BasicLimitInformation.PriorityClass = 0x8000 + if affinity: + limits.BasicLimitInformation.LimitFlags |= 0x10 + limits.BasicLimitInformation.Affinity = affinity + if not self.kernel32.SetInformationJobObject( + wintypes.HANDLE(self.handle), 9, ctypes.byref(limits), ctypes.sizeof(limits) + ): + raise OSError(ctypes.get_last_error(), "SetInformationJobObject failed") + + def assign(self, process): + if self.handle is None: + return + from ctypes import wintypes + + if not self.kernel32.AssignProcessToJobObject( + wintypes.HANDLE(self.handle), wintypes.HANDLE(int(process._handle)) + ): + raise OSError(ctypes.get_last_error(), "AssignProcessToJobObject failed") + + def stats(self): + if self.handle is None: + return {} + from ctypes import wintypes + + accounting = JobAccounting() + limits = JobExtendedLimits() + handle = wintypes.HANDLE(self.handle) + self.kernel32.QueryInformationJobObject( + handle, 1, ctypes.byref(accounting), ctypes.sizeof(accounting), None + ) + self.kernel32.QueryInformationJobObject( + handle, 9, ctypes.byref(limits), ctypes.sizeof(limits), None + ) + return { + "cpu_s": (accounting.TotalUserTime + accounting.TotalKernelTime) / 1e7, + "peak_commit_mb": limits.PeakJobMemoryUsed / 2**20, + "processes": accounting.TotalProcesses, + } + + def close(self): + if self.handle is not None: + from ctypes import wintypes + + self.kernel32.CloseHandle(wintypes.HANDLE(self.handle)) + self.handle = None + + +if os.name == "nt": + + class JobBasicLimits(ctypes.Structure): + _fields_ = [ + ("PerProcessUserTimeLimit", ctypes.c_int64), + ("PerJobUserTimeLimit", ctypes.c_int64), + ("LimitFlags", ctypes.c_uint32), + ("MinimumWorkingSetSize", ctypes.c_size_t), + ("MaximumWorkingSetSize", ctypes.c_size_t), + ("ActiveProcessLimit", ctypes.c_uint32), + ("Affinity", ctypes.c_size_t), + ("PriorityClass", ctypes.c_uint32), + ("SchedulingClass", ctypes.c_uint32), + ] + + class IoCounters(ctypes.Structure): + _fields_ = [(name, ctypes.c_uint64) for name in ("r", "w", "o", "rb", "wb", "ob")] + + class JobExtendedLimits(ctypes.Structure): + _fields_ = [ + ("BasicLimitInformation", JobBasicLimits), + ("IoInfo", IoCounters), + ("ProcessMemoryLimit", ctypes.c_size_t), + ("JobMemoryLimit", ctypes.c_size_t), + ("PeakProcessMemoryUsed", ctypes.c_size_t), + ("PeakJobMemoryUsed", ctypes.c_size_t), + ] + + class JobAccounting(ctypes.Structure): + _fields_ = [ + ("TotalUserTime", ctypes.c_int64), + ("TotalKernelTime", ctypes.c_int64), + ("ThisPeriodTotalUserTime", ctypes.c_int64), + ("ThisPeriodTotalKernelTime", ctypes.c_int64), + ("TotalPageFaultCount", ctypes.c_uint32), + ("TotalProcesses", ctypes.c_uint32), + ("ActiveProcesses", ctypes.c_uint32), + ("TotalTerminatedProcesses", ctypes.c_uint32), + ] + + +def run_measured(argv, timeout, env=None, cwd=None, affinity=None): + job = Job(affinity) + started = time.perf_counter() + process = subprocess.Popen( + argv, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + env=env, + cwd=cwd, + text=True, + encoding="utf-8", + errors="replace", + ) + try: + job.assign(process) + lines = [] + + def pump(): + for line in process.stdout: + lines.append((time.perf_counter() - started, line.rstrip("\n"))) + + reader = threading.Thread(target=pump, daemon=True) + reader.start() + try: + code = process.wait(timeout=timeout) + except subprocess.TimeoutExpired: + process.kill() + process.wait() + code = "timeout" + wall = time.perf_counter() - started + reader.join(5) + process.stdout.close() + return {"exit": code, "wall_s": wall, **job.stats()}, lines + finally: + job.close() + + +def float_after(lines, pattern, group=1): + regex = re.compile(pattern) + for _, line in lines: + match = regex.search(line) + if match: + return float(match.group(group)) + return None + + +def section_floats(lines, start_pattern, pattern, group=1): + regex = re.compile(pattern) + inside = False + for _, line in lines: + if re.search(start_pattern, line): + inside = True + continue + if inside: + match = regex.search(line) + if match: + return float(match.group(group)) + return None + + +def on_path(name): + path = shutil.which(name) + if path is None: + raise FileNotFoundError(f"missing {name}; put it on PATH") + return path + + +def ffprobe_video(path): + result = subprocess.run( + [ + "ffprobe", "-v", "error", "-count_packets", "-select_streams", "v:0", + "-show_entries", "stream=nb_read_packets,width,height:format=duration,size", + "-of", "json", str(path), + ], + capture_output=True, + text=True, + ) + if result.returncode != 0: + return {} + data = json.loads(result.stdout) + stream = (data.get("streams") or [{}])[0] + fmt = data.get("format", {}) + return { + "out_frames": float(stream.get("nb_read_packets", 0)), + "out_duration_s": float(fmt.get("duration", 0)), + "out_size_mb": float(fmt.get("size", 0)) / 2**20, + "out_width": float(stream.get("width", 0)), + "out_height": float(stream.get("height", 0)), + } + + +def parse_export(lines): + progress = [] + for at, line in lines: + if not line.startswith("{"): + continue + try: + event = json.loads(line) + except json.JSONDecodeError: + continue + if event.get("type") == "Progress": + progress.append((at, event.get("rendered_count", 0), event.get("total_frames", 0))) + rendering = [event for event in progress if event[1] > 0] + if not rendering: + return {} + first_at, first_rendered, _ = rendering[0] + last_at, rendered, total = rendering[-1] + metrics = {"first_frame_s": first_at, "frames": float(total or rendered)} + if last_at > first_at: + metrics["steady_fps"] = (rendered - first_rendered) / (last_at - first_at) + return metrics + + +def export_bench(fps, resolution, quality, extra=()): + def argv(ctx, out_dir): + return [ + str(ctx.bin("cap")), "export", str(ctx.fixture), "-o", str(out_dir / "export.mp4"), + "--fps", str(fps), "--resolution", resolution, "--quality", quality, "--json", *extra, + ] + + def post(ctx, out_dir, metrics): + probe = ffprobe_video(out_dir / "export.mp4") + metrics.update(probe) + if metrics.get("frames") and metrics.get("wall_s"): + metrics["export_fps"] = metrics["frames"] / metrics["wall_s"] + return out_dir / "export.mp4" + + return { + "argv": argv, + "parse": parse_export, + "post": post, + "requires": lambda ctx: on_path("ffprobe"), + "timeout": 900, + "env": {"CAP_DISABLE_ENCODER_SELF_TEST": None}, + "better": { + "wall_s": "lower", "cpu_s": "lower", "peak_commit_mb": "lower", + "export_fps": "higher", "steady_fps": "higher", "first_frame_s": "lower", + }, + } + + +def report_blocks(lines): + blocks = {} + block = stat = None + texts = [line for _, line in lines] + for index, line in enumerate(texts): + stripped = line.strip() + if index + 1 < len(texts) and set(texts[index - 1].strip()) == {"="} and set(texts[index + 1].strip()) == {"="}: + block = blocks.setdefault(stripped, {}) + stat = None + continue + if block is None: + continue + fps = re.match(r"Effective FPS:\s*([\d.]+)", stripped) + if fps: + block["fps"] = float(fps.group(1)) + elif stripped.endswith(":") and not stripped.startswith(("avg", "p50")): + stat = block.setdefault(stripped[:-1], {}) + elif stat is not None: + for key, value in re.findall(r"(avg|p50|p95|p99)=([\d.]+)ms", stripped): + stat[key] = float(value) + return blocks + + +def parse_playback_pipeline(lines): + blocks = report_blocks(lines) + decode = blocks.get("DECODE-ONLY", {}) + full = blocks.get("FULL PIPELINE - Full (1920x1080)", {}) + scrub = blocks.get("SCRUBBING (Half resolution)", {}) + metrics = { + "decode_fps": decode.get("fps"), + "pipeline_fps": full.get("fps"), + "render_p50_ms": full.get("GPU Render+Readback", {}).get("p50"), + "pipeline_p95_ms": full.get("Total (decode+render)", {}).get("p95"), + "scrub_decode_p50_ms": scrub.get("Decode", {}).get("p50"), + "scrub_p50_ms": scrub.get("Total (decode+render)", {}).get("p50"), + "scrub_p95_ms": scrub.get("Total (decode+render)", {}).get("p95"), + } + failed = sum( + int(match.group(1)) + for _, line in lines + if (match := re.match(r"(?:Decode|Render) failures:\s*(\d+)", line.strip())) + ) + if failed: + raise InvalidRun(f"playback-pipeline-benchmark reported {failed} failed frames") + return {key: value for key, value in metrics.items() if value is not None} + + +def parse_decode(lines): + metrics = { + "decoder_create_ms": float_after(lines, r"Time to create decoder:\s*([\d.]+)ms"), + "sequential_fps": section_floats(lines, r"SEQUENTIAL DECODE PERFORMANCE", r"Effective FPS:\s*([\d.]+)"), + "random_avg_ms": float_after(lines, r"Avg access time:\s*([\d.]+)ms"), + "random_p95_ms": section_floats(lines, r"RANDOM ACCESS PERFORMANCE", r"P95:\s*([\d.]+)ms"), + } + # Timings cover only the reads that succeeded, so a run with failed reads looks faster. + failed = sum( + int(match.group(2)) + for _, line in lines + if (match := re.match(r"(Frames failed|Seek failures|Failures):\s*(\d+)", line.strip())) + ) + if failed: + raise InvalidRun(f"decode-benchmark reported {failed} failed reads") + return {key: value for key, value in metrics.items() if value is not None} + + +def parse_startup(lines): + metrics = { + "startup_p50_ms": float_after(lines, r"^Startup: .*p50=([\d.]+)ms"), + "first_preview_p50_ms": float_after(lines, r"^First preview: .*p50=([\d.]+)ms"), + } + return {key: value for key, value in metrics.items() if value is not None} + + +def example_bench(name, args, parse, better, timeout=600, env=None): + def argv(ctx, out_dir): + return [str(ctx.bin(name, example=True)), *args(ctx)] + + return {"argv": argv, "parse": parse, "timeout": timeout, "env": env or {}, "better": better} + + +class InvalidRun(Exception): + pass + + +def validated_video(stdout, returncode): + try: + report = json.loads(stdout) + except json.JSONDecodeError: + raise InvalidRun(f"cap-test validate exited {returncode} without a JSON report") + if returncode != 0 or not report.get("valid"): + raise InvalidRun("invalid recording: " + "; ".join(report.get("errors") or ["no errors reported"])) + return report.get("video_info") or {} + + +def record_bench(mode, fps, seconds): + def argv(ctx, out_dir): + return [ + str(ctx.bin("cap")), "record", "--screen", ctx.screen, "--mode", mode, + "--fps", str(fps), "--duration", str(seconds), "--path", str(out_dir / "rec.cap"), + ] + + better = { + "cpu_per_rec_s": "lower", "peak_commit_mb": "lower", "delivered_fps": "higher", + "captured_ratio": "higher", "overhead_s": "lower", + } + if mode == "instant": + # cap-test validate reports no frame count or duration for instant recordings. + del better["delivered_fps"], better["captured_ratio"] + + def post(ctx, out_dir, metrics): + project = out_dir / "rec.cap" + validation = subprocess.run( + [str(ctx.bin("cap-test")), "validate", str(project), "--json"], + capture_output=True, text=True, + ) + video = validated_video(validation.stdout, validation.returncode) + frames = video.get("frame_count") + captured = video.get("duration_secs") + if frames and captured: + metrics["frames"] = float(frames) + metrics["captured_s"] = float(captured) + metrics["captured_ratio"] = float(captured) / seconds + metrics["delivered_fps"] = float(frames) / float(captured) + if metrics.get("cpu_s") is not None: + metrics["cpu_per_rec_s"] = metrics["cpu_s"] / seconds + metrics["overhead_s"] = metrics["wall_s"] - seconds + return None + + return { + "argv": argv, + "parse": lambda lines: {}, + "post": post, + "requires": lambda ctx: ctx.bin("cap-test"), + "timeout": seconds + 120, + "env": {}, + "needs_screen": True, + "better": better, + } + + +BENCHES = { + "export-1080p60": export_bench(60, "1920x1080", "maximum"), + "export-4k60": export_bench(60, "3840x2160", "maximum"), + "export-720p30": export_bench(30, "1280x720", "social"), + "playback-pipeline": example_bench( + "playback-pipeline-benchmark", + lambda ctx: ["--recording-path", str(ctx.fixture), "--fps", "60", "--frames", "300", "--force-ffmpeg"], + parse_playback_pipeline, + { + "cpu_s": "lower", "decode_fps": "higher", "pipeline_fps": "higher", "render_p50_ms": "lower", + "pipeline_p95_ms": "lower", "scrub_decode_p50_ms": "lower", "scrub_p50_ms": "lower", + "scrub_p95_ms": "lower", + }, + ), + "playback-pipeline-1ms": example_bench( + "playback-pipeline-benchmark", + lambda ctx: ["--recording-path", str(ctx.fixture), "--fps", "60", "--frames", "300", "--force-ffmpeg", "--timer-1ms"], + parse_playback_pipeline, + { + "cpu_s": "lower", "decode_fps": "higher", "pipeline_fps": "higher", "render_p50_ms": "lower", + "pipeline_p95_ms": "lower", "scrub_decode_p50_ms": "lower", "scrub_p50_ms": "lower", + "scrub_p95_ms": "lower", + }, + ), + "decode-display": example_bench( + "decode-benchmark", + lambda ctx: [ + "--video", str(ctx.fixture / "content" / "segments" / "segment-0" / "display.mp4"), + "--fps", "60", "--iterations", "50", + ], + parse_decode, + {"cpu_s": "lower", "sequential_fps": "higher", "random_avg_ms": "lower", "random_p95_ms": "lower", "decoder_create_ms": "lower"}, + ), + "editor-startup": example_bench( + "editor-startup-benchmark", + lambda ctx: ["--recording-path", str(ctx.fixture), "--runs", "3"], + parse_startup, + {"startup_p50_ms": "lower", "first_preview_p50_ms": "lower"}, + ), + "record-studio-60": record_bench("studio", 60, 10), + "record-instant-30": record_bench("instant", 30, 10), +} + +SUITES = { + "quick": ["export-1080p60", "playback-pipeline"], + "export": ["export-1080p60", "export-4k60", "export-720p30"], + "playback": ["playback-pipeline", "playback-pipeline-1ms", "decode-display", "editor-startup"], + "record": ["record-studio-60", "record-instant-30"], + "offline": ["export-1080p60", "export-4k60", "export-720p30", "playback-pipeline", "decode-display", "editor-startup"], +} +SUITES["all"] = SUITES["offline"] + SUITES["record"] + + +class Context: + def __init__(self, args): + self.source_fixture = Path(args.fixture).resolve() + self.fixture = self.source_fixture + self.bin_dir = Path(args.bin_dir).resolve() + self.screen = args.screen + self.animate = args.animate + self.affinity = int(args.affinity, 0) if args.affinity else None + self.extra_env = {} + + def bin(self, name, example=False): + path = self.bin_dir / ("examples" if example else "") / f"{name}{EXE}" + if not path.exists(): + raise FileNotFoundError(f"missing binary {path}; build it with --profile perf first") + return path + + +def bench_env(spec, ctx): + env = dict(os.environ) + env["PATH"] = str(ctx.bin_dir) + os.pathsep + env.get("PATH", "") + for key, value in spec.get("env", {}).items(): + if value is None: + env.pop(key, None) + else: + env[key] = value + env.update(ctx.extra_env) + return env + + +def start_animation(bounds): + x, y, width, height = bounds.split(",") + process = subprocess.Popen( + ["ffplay", "-hide_banner", "-loglevel", "quiet", "-noborder", "-alwaysontop", + "-left", x, "-top", y, "-x", width, "-y", height, + "-f", "lavfi", f"testsrc2=s={width}x{height}:r=60"], + stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + ) + time.sleep(1.5) + return process + + +# Job accounting, and the metrics derived from it, exist only where the job object does (Windows). +JOB_METRICS = {"cpu_s", "peak_commit_mb", "cpu_per_rec_s"} + + +def missing_metrics(better, metrics): + optional = set() if "cpu_s" in metrics else JOB_METRICS + return sorted(key for key in better if key not in metrics and key not in optional) + + +def run_once(name, spec, ctx, out_dir, keep_log): + out_dir.mkdir(parents=True, exist_ok=True) + argv = spec["argv"](ctx, out_dir) + animation = start_animation(ctx.animate) if spec.get("needs_screen") and ctx.animate else None + try: + measured, lines = run_measured(argv, spec["timeout"], env=bench_env(spec, ctx), cwd=out_dir, affinity=ctx.affinity) + finally: + if animation: + animation.kill() + animation.wait() + metrics = {key: value for key, value in measured.items() if key != "exit"} + artifact = None + code = measured["exit"] + try: + metrics.update(spec["parse"](lines)) + if code == 0 and spec.get("post"): + artifact = spec["post"](ctx, out_dir, metrics) + except InvalidRun as error: + code = "invalid" if code == 0 else code + lines.append((measured["wall_s"], f"capbench: {error}")) + missing = missing_metrics(spec["better"], metrics) + if code == 0 and missing: + code = "invalid" + lines.append((measured["wall_s"], f"capbench: no {', '.join(missing)} in this run's output")) + if keep_log or code != 0: + (out_dir / "output.log").write_text("\n".join(line for _, line in lines), encoding="utf-8") + return code, metrics, artifact + + +def build_info(bin_dir): + path = bin_dir / "build-info.txt" + return path.read_text(encoding="utf-8").strip() if path.exists() else None + + +def runner_git_info(): + def git(*args): + result = subprocess.run(["git", "-C", str(ROOT), *args], capture_output=True, text=True) + return result.stdout.strip() + + return {"commit": git("rev-parse", "HEAD"), "branch": git("rev-parse", "--abbrev-ref", "HEAD"), + "dirty": bool(git("status", "--porcelain", "--untracked-files=no"))} + + +def parse_variant_env(labels, items): + env = {label: {} for label in labels} + for item in items: + label, _, assignment = item.partition(":") + key, _, value = assignment.partition("=") + if label not in env or not key: + sys.exit(f"--env expects LABEL:KEY=VALUE with a known variant label, got {item!r}") + env[label][key] = value + return env + + +def cmd_run(args): + ctx = Context(args) + names = [] + for item in args.bench: + names.extend(SUITES.get(item, [item])) + unknown = [name for name in names if name not in BENCHES] + if unknown: + sys.exit(f"unknown benchmark(s): {', '.join(unknown)}. Known: {', '.join(BENCHES)}; suites: {', '.join(SUITES)}") + if any(BENCHES[name].get("needs_screen") for name in names) and not ctx.screen: + sys.exit("recording benchmarks need --screen (see `cap record screens`)") + if not ctx.source_fixture.exists(): + sys.exit(f"fixture not found: {ctx.source_fixture}") + + if any("=" not in item for item in args.variant): + sys.exit("--variant expects LABEL=BIN_DIR") + variants = [tuple(item.split("=", 1)) for item in args.variant] or [(args.label, args.bin_dir)] + if len({label for label, _ in variants}) != len(variants): + sys.exit("variant labels must be unique") + for label, _ in variants: + if label in ("", ".", "..") or Path(label).name != label: + sys.exit(f"label {label!r} must be a plain file name; results are written to /