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 95962a21737..ee069d2968a 100644 --- a/apps/desktop-gpui/Cargo.toml +++ b/apps/desktop-gpui/Cargo.toml @@ -185,7 +185,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..85274d8d710 100644 --- a/crates/rendering/src/decoder/media_foundation.rs +++ b/crates/rendering/src/decoder/media_foundation.rs @@ -249,7 +249,8 @@ impl MFDecoder { sender: oneshot::Sender, } - while let Ok(r) = rx.recv() { + 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 mut push_request = @@ -315,8 +316,8 @@ impl MFDecoder { ); } - 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 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| { @@ -355,7 +356,8 @@ impl MFDecoder { } else { Duration::from_millis(800) }; - let batch_start = Instant::now(); + 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()) { @@ -367,6 +369,34 @@ impl MFDecoder { 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 = super::request_frame(time, fps); + if let Some(cached) = + cached_frame_for(&cache, last_decoded_frame, frame) + { + let _ = + reply.send(cached.to_decoded_frame().with_matrix(matrix)); + 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, + } + } + let decode_start = Instant::now(); match decoder.read_sample() { Ok(Some(mf_frame)) => { @@ -451,18 +481,14 @@ impl MFDecoder { 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 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() @@ -471,7 +497,6 @@ impl MFDecoder { } } - let readahead_target = requested_frame + readahead_frames; if frame_number >= readahead_target || frame_number > cache_max { break; } @@ -549,21 +574,22 @@ 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. +// 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> { - cache.get(&frame).or_else(|| { - last_decoded - .is_some_and(|last| last > frame) - .then(|| cache.range(..=frame).next_back().map(|(_, cached)| cached)) - .flatten() - }) + last_decoded + .is_some_and(|last| last > frame) + .then(|| cache.range(..=frame).next_back().map(|(_, cached)| cached)) + .flatten() } #[cfg(test)] @@ -581,11 +607,18 @@ mod tests { #[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); } + #[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')); + } + #[test] fn a_frame_the_stream_skipped_holds_the_previous_one() { let cache = BTreeMap::from([(14, 'a'), (17, 'b')]); diff --git a/crates/rendering/src/frame_pipeline.rs b/crates/rendering/src/frame_pipeline.rs index cf502820034..458687ab387 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 @@ -890,7 +889,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, @@ -907,7 +906,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(); @@ -917,7 +916,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")] @@ -1113,7 +1112,10 @@ impl PendingNv12Surface { } pub struct PendingNv12Readback { - rx: Option>>, + armed: Option<( + oneshot::Receiver>, + wgpu::SubmissionIndex, + )>, mapped: bool, buffer: Arc, pub width: u32, @@ -1142,41 +1144,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()); - } + let completed = wait_for_submission( + device, + &submission, + Instant::now(), + self.frame_number, + gpu_buffer_wait_timeout(self.export_deadlines), + ) + .await; + match (completed, rx.try_recv()) { + (true, Ok(Ok(()))) => {} + (_, Ok(Err(error))) => { + self.unmap(); + return Err(RenderingError::BufferMapFailed(error)); } + _ => return Err(self.cancel()), } let buffer_slice = self.buffer.slice(..); @@ -2250,9 +2236,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 { @@ -2453,8 +2439,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") @@ -2479,8 +2465,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 c939d016210..6abde058f9f 100644 --- a/crates/rendering/src/lib.rs +++ b/crates/rendering/src/lib.rs @@ -260,9 +260,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/scripts/capbench.py b/scripts/capbench.py new file mode 100644 index 00000000000..55fd98085be --- /dev/null +++ b/scripts/capbench.py @@ -0,0 +1,866 @@ +#!/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: + raise InvalidRun(f"ffprobe could not read {path.name}: {result.stderr.strip()[:200]}") + data = json.loads(result.stdout) + stream = (data.get("streams") or [{}])[0] + fmt = data.get("format", {}) + if not int(stream.get("nb_read_packets", 0)): + raise InvalidRun(f"ffprobe found no video frames in {path.name}") + 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_problem(probe, expected_frames, resolution): + if not probe.get("out_frames"): + return "export has no readable video frames" + if expected_frames and probe["out_frames"] != expected_frames: + return f"export has {probe['out_frames']:.0f} frames, expected {expected_frames:.0f}" + max_width, max_height = (float(side) for side in resolution.split("x")) + width, height = probe.get("out_width", 0), probe.get("out_height", 0) + if not (0 < width <= max_width and 0 < height <= max_height): + return f"export is {width:.0f}x{height:.0f}, outside {resolution}" + return None + + +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") + problem = export_problem(probe, metrics.get("frames"), resolution) + if problem: + raise InvalidRun(problem) + 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 /