From 75b558e860d791e29c557b12f4e1767d78523403 Mon Sep 17 00:00:00 2001 From: Aatman Jain Date: Thu, 29 Oct 2026 23:34:16 +0000 Subject: [PATCH 01/11] perf(recording): overlap Windows fragmented encode readback and GPU scale Instant and studio fragmented M4S paths stage D3D readback ahead of encode, run pixel conversion through AsyncConverterPool when needed, scale mismatched capture sizes on the D3D11 video processor before readback, and align Windows instant max capture size with encode resolution to avoid redundant swscale. --- crates/enc-ffmpeg/src/mux/segmented_stream.rs | 24 ++ crates/enc-ffmpeg/src/video/h264.rs | 11 +- crates/recording/src/capture_pipeline.rs | 2 + crates/recording/src/instant_recording.rs | 6 +- .../recording/src/output_pipeline/ffmpeg.rs | 1 + .../output_pipeline/macos_fragmented_m4s.rs | 2 + .../src/output_pipeline/win_fragmented_m4s.rs | 281 ++++++++++++++---- .../src/sources/screen_capture/gpu_scale.rs | 276 +++++++++++++++++ .../src/sources/screen_capture/windows.rs | 95 ++++++ 9 files changed, 633 insertions(+), 65 deletions(-) create mode 100644 crates/recording/src/sources/screen_capture/gpu_scale.rs diff --git a/crates/enc-ffmpeg/src/mux/segmented_stream.rs b/crates/enc-ffmpeg/src/mux/segmented_stream.rs index 0ea15ad1566..8e65291bf0a 100644 --- a/crates/enc-ffmpeg/src/mux/segmented_stream.rs +++ b/crates/enc-ffmpeg/src/mux/segmented_stream.rs @@ -231,6 +231,10 @@ pub struct SegmentedVideoEncoderConfig { /// [`SegmentedVideoEncoder::queue_hw_pixel_buffer`]) before falling back /// to the software-frame path. Ignored on other platforms. pub prefer_videotoolbox_hw_input: bool, + /// When set, pixel format conversion and scaling run outside the encoder + /// (for example via [`cap_frame_converter::AsyncConverterPool`]) so + /// readback and GPU conversion can overlap encoding. + pub prefer_external_conversion: bool, } impl Default for SegmentedVideoEncoderConfig { @@ -241,6 +245,7 @@ impl Default for SegmentedVideoEncoderConfig { bpp: H264EncoderBuilder::QUALITY_BPP, output_size: None, prefer_videotoolbox_hw_input: false, + prefer_external_conversion: false, } } } @@ -333,6 +338,10 @@ impl SegmentedVideoEncoder { builder = builder.with_output_size(width, height)?; } + if config.prefer_external_conversion { + builder = builder.with_external_conversion(); + } + #[cfg(target_os = "macos")] let hw_attempt = if config.prefer_videotoolbox_hw_input { match builder.clone().build_videotoolbox_hw_input(&mut output) { @@ -480,6 +489,21 @@ impl SegmentedVideoEncoder { Ok(()) } + pub fn conversion_requirements(&self) -> crate::video::h264::ConversionRequirements { + self.encoder.conversion_requirements() + } + + pub fn queue_preconverted_frame( + &mut self, + frame: frame::Video, + timestamp: Duration, + ) -> Result<(), QueueFrameError> { + self.encoder + .queue_preconverted_frame_to(frame, timestamp, &mut self.mux)?; + self.mux.frame_queued(); + Ok(()) + } + pub fn finish(&mut self) -> Result<(), FinishError> { let encoder = &mut self.encoder; self.mux diff --git a/crates/enc-ffmpeg/src/video/h264.rs b/crates/enc-ffmpeg/src/video/h264.rs index bdf1c6f68d1..7ac2cc32e7e 100644 --- a/crates/enc-ffmpeg/src/video/h264.rs +++ b/crates/enc-ffmpeg/src/video/h264.rs @@ -1167,9 +1167,18 @@ impl H264Encoder { pub fn queue_preconverted_frame( &mut self, - mut frame: frame::Video, + frame: frame::Video, timestamp: Duration, output: &mut format::context::Output, + ) -> Result<(), QueueFrameError> { + self.queue_preconverted_frame_to(frame, timestamp, output) + } + + pub fn queue_preconverted_frame_to( + &mut self, + mut frame: frame::Video, + timestamp: Duration, + output: &mut S, ) -> Result<(), QueueFrameError> { trace!( "Encoding pre-converted frame: format={:?}, size={}x{}, expected={:?} {}x{}", diff --git a/crates/recording/src/capture_pipeline.rs b/crates/recording/src/capture_pipeline.rs index d935e0e30f2..f0ef01d982b 100644 --- a/crates/recording/src/capture_pipeline.rs +++ b/crates/recording/src/capture_pipeline.rs @@ -320,6 +320,7 @@ impl MakeCapturePipeline for screen_capture::Direct3DCapture { shared_pause_state, disk_space_callback: None, segment_tx: None, + ..Default::default() }) .await } @@ -369,6 +370,7 @@ impl MakeCapturePipeline for screen_capture::Direct3DCapture { shared_pause_state: None, disk_space_callback: None, segment_tx, + ..Default::default() }) .await } diff --git a/crates/recording/src/instant_recording.rs b/crates/recording/src/instant_recording.rs index 6c0102a11fc..1c57865fb3c 100644 --- a/crates/recording/src/instant_recording.rs +++ b/crates/recording/src/instant_recording.rs @@ -1337,7 +1337,7 @@ async fn build_instant_recording_actor( let (display, crop_bounds) = target_to_display_and_crop(&inputs.capture_target) .context("target_display_crop")?; - #[cfg(target_os = "macos")] + #[cfg(any(target_os = "macos", target_os = "windows"))] let max_capture_size = max_output_size.and_then(|max_output_size| { inputs.capture_target.physical_size().map(|size| { capture_size_constraint( @@ -1346,7 +1346,7 @@ async fn build_instant_recording_actor( ) }) }); - #[cfg(not(target_os = "macos"))] + #[cfg(target_os = "linux")] let max_capture_size = None; #[cfg(any(target_os = "macos", windows))] @@ -1530,7 +1530,7 @@ fn current_time_f64() -> f64 { .as_secs_f64() } -#[cfg(target_os = "macos")] +#[cfg(any(target_os = "macos", target_os = "windows"))] fn capture_size_constraint(input: (u32, u32), max_output_size: u32) -> (u32, u32) { let aspect_ratio = input.0 as f64 / input.1 as f64; let max_short_edge = (max_output_size as f64 / 16.0 * 9.0) as u32; diff --git a/crates/recording/src/output_pipeline/ffmpeg.rs b/crates/recording/src/output_pipeline/ffmpeg.rs index bc96016faa2..32ae5a23a85 100644 --- a/crates/recording/src/output_pipeline/ffmpeg.rs +++ b/crates/recording/src/output_pipeline/ffmpeg.rs @@ -689,6 +689,7 @@ impl SegmentedVideoMuxer { bpp: H264EncoderBuilder::QUALITY_BPP, output_size: self.output_size, prefer_videotoolbox_hw_input: false, + prefer_external_conversion: false, }; let slow_threshold_ms = frame_timing_log_threshold_ms(&self.video_config); diff --git a/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs b/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs index cd36e3bff3b..498bd55e7e8 100644 --- a/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs +++ b/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs @@ -297,6 +297,7 @@ impl MacOSFragmentedM4SMuxer { bpp: self.bpp, output_size: self.output_size, prefer_videotoolbox_hw_input: true, + prefer_external_conversion: false, }; let mut encoder = SegmentedVideoEncoder::init_write_behind( @@ -688,6 +689,7 @@ impl MacOSFragmentedM4SCameraMuxer { bpp: self.bpp, output_size: self.output_size, prefer_videotoolbox_hw_input: false, + prefer_external_conversion: false, }; let mut encoder = SegmentedVideoEncoder::init_write_behind( diff --git a/crates/recording/src/output_pipeline/win_fragmented_m4s.rs b/crates/recording/src/output_pipeline/win_fragmented_m4s.rs index b517b575903..50edcb86e46 100644 --- a/crates/recording/src/output_pipeline/win_fragmented_m4s.rs +++ b/crates/recording/src/output_pipeline/win_fragmented_m4s.rs @@ -29,6 +29,15 @@ use std::{ use tracing::*; const DEFAULT_MUXER_BUFFER_SIZE: usize = 240; +const DEFAULT_CONVERT_STAGING_DEPTH: usize = 32; +const DEFAULT_GPU_CONVERTER_WORKERS: usize = 2; + +fn get_convert_staging_depth() -> usize { + std::env::var("CAP_WIN_CONVERT_STAGING_DEPTH") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(DEFAULT_CONVERT_STAGING_DEPTH) +} fn get_muxer_buffer_size() -> usize { std::env::var("CAP_MUXER_BUFFER_SIZE") @@ -215,6 +224,7 @@ pub struct WindowsFragmentedM4SMuxer { disk_space_callback: Option, segment_tx: Option>, + prefer_external_conversion: bool, health_tx: SharedHealthSender, } @@ -228,6 +238,7 @@ pub struct WindowsFragmentedM4SMuxerConfig { pub disk_space_callback: Option, pub segment_tx: Option>, + pub prefer_external_conversion: bool, } impl Default for WindowsFragmentedM4SMuxerConfig { @@ -241,10 +252,16 @@ impl Default for WindowsFragmentedM4SMuxerConfig { shared_pause_state: None, disk_space_callback: None, segment_tx: None, + prefer_external_conversion: true, } } } +enum StagedFfmpegFrame { + Ready(ffmpeg::frame::Video, Duration), + Shutdown, +} + impl Muxer for WindowsFragmentedM4SMuxer { type Config = WindowsFragmentedM4SMuxerConfig; @@ -286,6 +303,7 @@ impl Muxer for WindowsFragmentedM4SMuxer { started: false, disk_space_callback: config.disk_space_callback, segment_tx: config.segment_tx, + prefer_external_conversion: config.prefer_external_conversion, health_tx: SharedHealthSender::new(), }; @@ -335,12 +353,14 @@ impl WindowsFragmentedM4SMuxer { sync_channel::>(buffer_size); let (ready_tx, ready_rx) = sync_channel::>(1); + let prefer_external_conversion = self.prefer_external_conversion; let encoder_config = SegmentedVideoEncoderConfig { segment_duration: self.segment_duration, preset: self.preset, bpp: self.bpp, output_size: self.output_size, prefer_videotoolbox_hw_input: false, + prefer_external_conversion, }; let mut encoder = SegmentedVideoEncoder::init_with_timed_fragments( @@ -349,6 +369,7 @@ impl WindowsFragmentedM4SMuxer { encoder_config, self.timed_fragments, )?; + let conversion_requirements = encoder.conversion_requirements(); if let Some(callback) = &self.disk_space_callback { encoder.set_disk_space_callback(callback.clone()); } @@ -362,8 +383,25 @@ impl WindowsFragmentedM4SMuxer { let health_tx = self.health_tx.clone(); let base_path = self.base_path.clone(); let pause_flag = self.pause.flag(); + let staging_depth = get_convert_staging_depth(); + let (staging_tx, staging_rx) = + sync_channel::>(staging_depth.max(4)); + let encode_encoder = encoder.clone(); + let use_gpu_converter = + prefer_external_conversion && conversion_requirements.needs_conversion; + let encode_handle = std::thread::Builder::new() + .name("win-m4s-segment-encode".to_string()) + .spawn(move || { + cap_mediafoundation_utils::thread_init(); + run_win_m4s_encode_thread( + staging_rx, + encode_encoder, + conversion_requirements, + use_gpu_converter, + ) + })?; let encoder_handle = std::thread::Builder::new() - .name("win-m4s-segment-encoder".to_string()) + .name("win-m4s-segment-readback".to_string()) .spawn(move || { cap_mediafoundation_utils::thread_init(); @@ -376,7 +414,6 @@ impl WindowsFragmentedM4SMuxer { let mut first_timestamp: Option = None; let mut slow_convert_count = 0u32; - let mut slow_encode_count = 0u32; let mut total_frames = 0u64; let mut duplicated_frames = 0u64; let mut disk_monitor = DiskSpaceMonitor::new(); @@ -393,43 +430,12 @@ impl WindowsFragmentedM4SMuxer { } }; - let encode_frame_fn = |ffmpeg_frame: ffmpeg::frame::Video, - timestamp: Duration, - slow_encode_count: &mut u32, - total_frames: &mut u64, - encoder: &Arc>| + let stage_frame = |ffmpeg_frame: ffmpeg::frame::Video, + timestamp: Duration| -> anyhow::Result<()> { - let encode_start = std::time::Instant::now(); - - match encoder.lock() { - Ok(mut enc) => { - if let Err(e) = enc.queue_frame(ffmpeg_frame, timestamp) { - warn!("Failed to encode frame: {e}"); - } - } - Err(_) => { - return Err(anyhow!( - "Encoder mutex poisoned - all subsequent frames would be lost" - )); - } - } - - let encode_elapsed_ms = encode_start.elapsed().as_millis(); - - if encode_elapsed_ms > SLOW_THRESHOLD_MS { - *slow_encode_count += 1; - if *slow_encode_count <= 5 || slow_encode_count.is_multiple_of(100) { - debug!( - elapsed_ms = encode_elapsed_ms, - count = *slow_encode_count, - "encoder.queue_frame exceeded {}ms threshold", - SLOW_THRESHOLD_MS - ); - } - } - - *total_frames += 1; - Ok(()) + staging_tx + .send(Some(StagedFfmpegFrame::Ready(ffmpeg_frame, timestamp))) + .map_err(|_| anyhow!("Windows M4S convert staging channel disconnected")) }; loop { @@ -476,17 +482,12 @@ impl WindowsFragmentedM4SMuxer { if let Ok(f) = frame.as_ffmpeg() { let normalized_ts = normalize_timestamp(ts, &mut first_timestamp); - if let Err(e) = encode_frame_fn( - f, - normalized_ts, - &mut slow_encode_count, - &mut total_frames, - &encoder_clone, - ) { - warn!("Failed to encode drained frame: {e}"); + if let Err(e) = stage_frame(f, normalized_ts) { + warn!("Failed to stage drained frame: {e}"); break; } drained += 1; + total_frames += 1; } } Ok(None) => break, @@ -525,37 +526,34 @@ impl WindowsFragmentedM4SMuxer { } let normalized_ts = normalize_timestamp(timestamp, &mut first_timestamp); - if let Err(e) = encode_frame_fn( - ffmpeg_frame, - normalized_ts, - &mut slow_encode_count, - &mut total_frames, - &encoder_clone, - ) { - error!("Encoder mutex poisoned - encoder thread likely panicked, stopping"); + if let Err(e) = stage_frame(ffmpeg_frame, normalized_ts) { + error!("Convert staging channel closed, stopping readback thread"); return Err(e); } + total_frames += 1; } + let _ = staging_tx.send(Some(StagedFfmpegFrame::Shutdown)); + drop(staging_tx); + + let encode_result = encode_handle + .join() + .map_err(|_| anyhow!("Windows M4S encode thread panicked"))?; + if total_frames > 0 { debug!( total_frames = total_frames, duplicated_frames = duplicated_frames, slow_converts = slow_convert_count, - slow_encodes = slow_encode_count, slow_convert_pct = format!( "{:.1}%", 100.0 * slow_convert_count as f64 / total_frames as f64 ), - slow_encode_pct = format!( - "{:.1}%", - 100.0 * slow_encode_count as f64 / total_frames as f64 - ), - "Windows M4S encoder timing summary (using SegmentedVideoEncoder)" + "Windows M4S readback timing summary" ); } - Ok(()) + encode_result })?; ready_rx @@ -792,6 +790,7 @@ impl WindowsFragmentedM4SCameraMuxer { bpp: self.bpp, output_size: self.output_size, prefer_videotoolbox_hw_input: false, + prefer_external_conversion: false, }; let mut encoder = SegmentedVideoEncoder::init_with_timed_fragments( @@ -1071,6 +1070,166 @@ impl AudioMuxer for WindowsFragmentedM4SCameraMuxer { } } +fn run_win_m4s_encode_thread( + staging_rx: std::sync::mpsc::Receiver>, + encoder: Arc>, + requirements: cap_enc_ffmpeg::h264::ConversionRequirements, + use_gpu_converter: bool, +) -> anyhow::Result<()> { + use cap_frame_converter::{ + AsyncConverterPool, ConversionConfig, ConvertError, ConverterPoolConfig, DropStrategy, + }; + use std::sync::atomic::{AtomicU64, Ordering}; + + const SLOW_THRESHOLD_MS: u128 = 5; + + let worker_count = std::thread::available_parallelism() + .map(|parallelism| parallelism.get().clamp(2, 4)) + .unwrap_or(DEFAULT_GPU_CONVERTER_WORKERS); + + let mut converter_pool = if use_gpu_converter { + let conversion_config = ConversionConfig::new( + requirements.input_format, + requirements.input_width, + requirements.input_height, + requirements.output_format, + requirements.output_width, + requirements.output_height, + ); + let pool_config = ConverterPoolConfig { + worker_count, + input_capacity: 120, + output_capacity: 90, + drop_strategy: DropStrategy::DropOldest, + ..Default::default() + }; + Some( + AsyncConverterPool::from_config(conversion_config, pool_config) + .map_err(|error| anyhow!("Failed to create Windows M4S converter pool: {error}"))?, + ) + } else { + None + }; + + if converter_pool.is_some() { + debug!( + worker_count = worker_count, + input = %format!( + "{:?} {}x{}", + requirements.input_format, + requirements.input_width, + requirements.input_height + ), + output = %format!( + "{:?} {}x{}", + requirements.output_format, + requirements.output_width, + requirements.output_height + ), + "Windows M4S encoder using async GPU/CPU converter pool" + ); + } + + let frame_sequence = AtomicU64::new(0); + let mut slow_encode_count = 0u32; + let mut total_frames = 0u64; + + let mut drain_converted = |pool: &mut AsyncConverterPool| -> anyhow::Result<()> { + while let Some(converted) = pool.try_recv() { + let timestamp = Duration::from_micros(converted.frame.pts().unwrap_or(0).max(0) as u64); + let encode_start = std::time::Instant::now(); + match encoder.lock() { + Ok(mut enc) => { + if let Err(error) = enc.queue_preconverted_frame(converted.frame, timestamp) { + warn!("Failed to encode converted frame: {error}"); + } + } + Err(_) => { + return Err(anyhow!( + "Encoder mutex poisoned - all subsequent frames would be lost" + )); + } + } + let encode_elapsed_ms = encode_start.elapsed().as_millis(); + if encode_elapsed_ms > SLOW_THRESHOLD_MS { + slow_encode_count += 1; + } + total_frames += 1; + } + Ok(()) + }; + + loop { + match staging_rx.recv() { + Ok(Some(StagedFfmpegFrame::Ready(mut frame, timestamp))) => { + if let Some(pool) = &mut converter_pool { + frame.set_pts(Some(timestamp.as_micros() as i64)); + match pool.submit(frame, frame_sequence.fetch_add(1, Ordering::Relaxed)) { + Ok(()) => {} + Err(ConvertError::PoolShutdown) => break, + Err(error) => { + warn!("Failed to submit frame to converter pool: {error}"); + } + } + drain_converted(pool)?; + } else { + let encode_start = std::time::Instant::now(); + match encoder.lock() { + Ok(mut enc) => { + if let Err(error) = enc.queue_frame(frame, timestamp) { + warn!("Failed to encode frame: {error}"); + } + } + Err(_) => { + return Err(anyhow!( + "Encoder mutex poisoned - all subsequent frames would be lost" + )); + } + } + let encode_elapsed_ms = encode_start.elapsed().as_millis(); + if encode_elapsed_ms > SLOW_THRESHOLD_MS { + slow_encode_count += 1; + } + total_frames += 1; + } + } + Ok(Some(StagedFfmpegFrame::Shutdown)) | Ok(None) => break, + Err(_) => break, + } + } + + if let Some(mut pool) = converter_pool { + pool.drain_with_timeout( + |converted| { + let timestamp = + Duration::from_micros(converted.frame.pts().unwrap_or(0).max(0) as u64); + if let Ok(mut enc) = encoder.lock() + && let Err(error) = enc.queue_preconverted_frame(converted.frame, timestamp) + { + warn!("Failed to encode drained converted frame: {error}"); + } else { + total_frames += 1; + } + }, + Duration::from_secs(5), + ); + } + + if total_frames > 0 { + debug!( + total_frames = total_frames, + slow_encodes = slow_encode_count, + slow_encode_pct = format!( + "{:.1}%", + 100.0 * slow_encode_count as f64 / total_frames as f64 + ), + "Windows M4S encode timing summary" + ); + } + + Ok(()) +} + #[cfg(test)] mod tests { use super::reference_video_frame; diff --git a/crates/recording/src/sources/screen_capture/gpu_scale.rs b/crates/recording/src/sources/screen_capture/gpu_scale.rs new file mode 100644 index 00000000000..5003851b212 --- /dev/null +++ b/crates/recording/src/sources/screen_capture/gpu_scale.rs @@ -0,0 +1,276 @@ +use ::windows::Win32::Graphics::Direct3D11::{ + D3D11_CPU_ACCESS_READ, D3D11_MAP_READ, D3D11_MAPPED_SUBRESOURCE, D3D11_TEXTURE2D_DESC, + D3D11_USAGE_STAGING, D3D11_VIDEO_PROCESSOR_CONTENT_DESC, D3D11_VIDEO_PROCESSOR_INPUT_VIEW_DESC, + D3D11_VIDEO_PROCESSOR_OUTPUT_VIEW_DESC, D3D11_VIDEO_PROCESSOR_STREAM, + D3D11_VPIV_DIMENSION_TEXTURE2D, D3D11_VPOV_DIMENSION_TEXTURE2D, ID3D11Device, + ID3D11DeviceContext, ID3D11Texture2D, ID3D11VideoContext, ID3D11VideoDevice, + ID3D11VideoProcessor, ID3D11VideoProcessorEnumerator, ID3D11VideoProcessorInputView, + ID3D11VideoProcessorOutputView, +}; +use ::windows::Win32::Graphics::Dxgi::Common::{ + DXGI_FORMAT, DXGI_FORMAT_B8G8R8A8_UNORM, DXGI_FORMAT_R8G8B8A8_UNORM, +}; +use ::windows::core::Interface; +use scap_direct3d::PixelFormat; +use std::mem::ManuallyDrop; +use std::ptr; + +pub struct GpuTextureScaler { + context: ID3D11DeviceContext, + video_device: ID3D11VideoDevice, + video_context: ID3D11VideoContext, + processor: ID3D11VideoProcessor, + enumerator: ID3D11VideoProcessorEnumerator, + output_texture: ID3D11Texture2D, + staging_output: ID3D11Texture2D, + pixel_format: PixelFormat, + src_width: u32, + src_height: u32, + dst_width: u32, + dst_height: u32, + row_bytes: usize, + pixel_data: Vec, +} + +impl GpuTextureScaler { + pub fn try_new( + device: &ID3D11Device, + pixel_format: PixelFormat, + src_width: u32, + src_height: u32, + dst_width: u32, + dst_height: u32, + ) -> Option { + if src_width == 0 || src_height == 0 || dst_width == 0 || dst_height == 0 { + return None; + } + + let context = unsafe { device.GetImmediateContext().ok()? }; + let video_device: ID3D11VideoDevice = device.cast().ok()?; + let video_context: ID3D11VideoContext = context.cast().ok()?; + + let content_desc = D3D11_VIDEO_PROCESSOR_CONTENT_DESC { + InputFrameFormat: + ::windows::Win32::Graphics::Direct3D11::D3D11_VIDEO_FRAME_FORMAT_PROGRESSIVE, + InputFrameRate: ::windows::Win32::Graphics::Dxgi::Common::DXGI_RATIONAL { + Numerator: 30, + Denominator: 1, + }, + InputWidth: src_width, + InputHeight: src_height, + OutputFrameRate: ::windows::Win32::Graphics::Dxgi::Common::DXGI_RATIONAL { + Numerator: 30, + Denominator: 1, + }, + OutputWidth: dst_width, + OutputHeight: dst_height, + Usage: ::windows::Win32::Graphics::Direct3D11::D3D11_VIDEO_USAGE_PLAYBACK_NORMAL, + }; + + let enumerator = unsafe { + video_device + .CreateVideoProcessorEnumerator(&content_desc) + .ok()? + }; + let processor = unsafe { video_device.CreateVideoProcessor(&enumerator, 0).ok()? }; + + let dxgi_format = pixel_format_to_dxgi(pixel_format); + let output_texture = create_default_texture(device, dst_width, dst_height, dxgi_format)?; + let staging_output = create_staging_texture(device, dst_width, dst_height, dxgi_format)?; + + let row_bytes = (dst_width * 4) as usize; + let pixel_data = vec![0u8; row_bytes * dst_height as usize]; + + Some(Self { + context, + video_device, + video_context, + processor, + enumerator, + output_texture, + staging_output, + pixel_format, + src_width, + src_height, + dst_width, + dst_height, + row_bytes, + pixel_data, + }) + } + + pub fn scale_texture(&mut self, input: &ID3D11Texture2D) -> Option<&[u8]> { + let input_view_desc = D3D11_VIDEO_PROCESSOR_INPUT_VIEW_DESC { + FourCC: 0, + ViewDimension: D3D11_VPIV_DIMENSION_TEXTURE2D, + Anonymous: + ::windows::Win32::Graphics::Direct3D11::D3D11_VIDEO_PROCESSOR_INPUT_VIEW_DESC_0 { + Texture2D: ::windows::Win32::Graphics::Direct3D11::D3D11_TEX2D_VPIV { + MipSlice: 0, + ArraySlice: 0, + }, + }, + }; + + let mut input_view: Option = None; + unsafe { + self.video_device + .CreateVideoProcessorInputView( + input, + &self.enumerator, + &input_view_desc, + Some(&mut input_view), + ) + .ok()?; + } + let input_view = input_view?; + + let output_view_desc = D3D11_VIDEO_PROCESSOR_OUTPUT_VIEW_DESC { + ViewDimension: D3D11_VPOV_DIMENSION_TEXTURE2D, + Anonymous: + ::windows::Win32::Graphics::Direct3D11::D3D11_VIDEO_PROCESSOR_OUTPUT_VIEW_DESC_0 { + Texture2D: ::windows::Win32::Graphics::Direct3D11::D3D11_TEX2D_VPOV { + MipSlice: 0, + }, + }, + }; + + let mut output_view: Option = None; + unsafe { + self.video_device + .CreateVideoProcessorOutputView( + &self.output_texture, + &self.enumerator, + &output_view_desc, + Some(&mut output_view), + ) + .ok()?; + } + let output_view = output_view?; + + let stream = D3D11_VIDEO_PROCESSOR_STREAM { + Enable: true.into(), + OutputIndex: 0, + InputFrameOrField: 0, + PastFrames: 0, + FutureFrames: 0, + ppPastSurfaces: ptr::null_mut(), + pInputSurface: std::mem::transmute_copy(&input_view), + ppFutureSurfaces: ptr::null_mut(), + ppPastSurfacesRight: ptr::null_mut(), + pInputSurfaceRight: ManuallyDrop::new(None), + ppFutureSurfacesRight: ptr::null_mut(), + }; + + unsafe { + self.video_context + .VideoProcessorBlt(&self.processor, &output_view, 0, &[stream]) + .ok()?; + + self.context + .CopyResource(&self.staging_output, &self.output_texture); + + let mut mapped = D3D11_MAPPED_SUBRESOURCE::default(); + self.context + .Map( + &self.staging_output, + 0, + D3D11_MAP_READ, + 0, + Some(&mut mapped), + ) + .ok()?; + + let src_ptr = mapped.pData as *const u8; + let src_stride = mapped.RowPitch as usize; + for row in 0..self.dst_height as usize { + let src_start = row * src_stride; + let dst_start = row * self.row_bytes; + std::ptr::copy_nonoverlapping( + src_ptr.add(src_start), + self.pixel_data.as_mut_ptr().add(dst_start), + self.row_bytes, + ); + } + + self.context.Unmap(&self.staging_output, 0); + } + + Some(self.pixel_data.as_slice()) + } + + pub fn dimensions(&self) -> (u32, u32, PixelFormat) { + (self.dst_width, self.dst_height, self.pixel_format) + } + + pub fn pixel_data_for_frame(&self) -> &[u8] { + &self.pixel_data + } +} + +fn pixel_format_to_dxgi(pixel_format: PixelFormat) -> DXGI_FORMAT { + match pixel_format { + PixelFormat::R8G8B8A8Unorm => DXGI_FORMAT_R8G8B8A8_UNORM, + PixelFormat::B8G8R8A8Unorm => DXGI_FORMAT_B8G8R8A8_UNORM, + } +} + +fn create_default_texture( + device: &ID3D11Device, + width: u32, + height: u32, + format: DXGI_FORMAT, +) -> Option { + let desc = D3D11_TEXTURE2D_DESC { + Width: width, + Height: height, + MipLevels: 1, + ArraySize: 1, + Format: format, + SampleDesc: ::windows::Win32::Graphics::Dxgi::Common::DXGI_SAMPLE_DESC { + Count: 1, + Quality: 0, + }, + Usage: ::windows::Win32::Graphics::Direct3D11::D3D11_USAGE_DEFAULT, + BindFlags: ::windows::Win32::Graphics::Direct3D11::D3D11_BIND_RENDER_TARGET.0 as u32, + CPUAccessFlags: 0, + MiscFlags: 0, + }; + unsafe { + let mut texture = None; + device + .CreateTexture2D(&desc, None, Some(&mut texture)) + .ok()?; + texture + } +} + +fn create_staging_texture( + device: &ID3D11Device, + width: u32, + height: u32, + format: DXGI_FORMAT, +) -> Option { + let desc = D3D11_TEXTURE2D_DESC { + Width: width, + Height: height, + MipLevels: 1, + ArraySize: 1, + Format: format, + SampleDesc: ::windows::Win32::Graphics::Dxgi::Common::DXGI_SAMPLE_DESC { + Count: 1, + Quality: 0, + }, + Usage: D3D11_USAGE_STAGING, + BindFlags: 0, + CPUAccessFlags: D3D11_CPU_ACCESS_READ.0 as u32, + MiscFlags: 0, + }; + unsafe { + let mut texture = None; + device + .CreateTexture2D(&desc, None, Some(&mut texture)) + .ok()?; + texture + } +} diff --git a/crates/recording/src/sources/screen_capture/windows.rs b/crates/recording/src/sources/screen_capture/windows.rs index ef8a9dbd343..83325cb1f3e 100644 --- a/crates/recording/src/sources/screen_capture/windows.rs +++ b/crates/recording/src/sources/screen_capture/windows.rs @@ -29,6 +29,9 @@ use std::{ use tokio_util::{future::FutureExt as _, sync::CancellationToken}; use tracing::*; +mod gpu_scale; +use gpu_scale::GpuTextureScaler; + // const WINDOW_DURATION: Duration = Duration::from_secs(3); // const LOG_INTERVAL: Duration = Duration::from_secs(5); @@ -185,6 +188,8 @@ struct WindowsFrameScaler { pixel_format: scap_direct3d::PixelFormat, d3d_device: ID3D11Device, state: Option, + gpu_scaler: Option, + gpu_scaler_source: Option<(u32, u32)>, } impl WindowsFrameScaler { @@ -200,10 +205,98 @@ impl WindowsFrameScaler { pixel_format, d3d_device, state: None, + gpu_scaler: None, + gpu_scaler_source: None, + } + } + + fn ensure_gpu_scaler( + &mut self, + src_width: u32, + src_height: u32, + ) -> Option<&mut GpuTextureScaler> { + let needs_reinit = self + .gpu_scaler_source + .is_none_or(|(w, h)| w != src_width || h != src_height); + + if needs_reinit { + self.gpu_scaler = GpuTextureScaler::try_new( + &self.d3d_device, + self.pixel_format, + src_width, + src_height, + self.target_width, + self.target_height, + ); + self.gpu_scaler_source = self.gpu_scaler.as_ref().map(|_| (src_width, src_height)); } + + self.gpu_scaler.as_mut() + } + + fn scale_frame_gpu(&mut self, frame: &scap_direct3d::Frame) -> Option { + let src_width = frame.width(); + let src_height = frame.height(); + let scaler = self.ensure_gpu_scaler(src_width, src_height)?; + scaler.scale_texture(frame.texture())?; + + let (width, height, pixel_format) = scaler.dimensions(); + let row_bytes = (width * 4) as usize; + let pixel_data = scaler.pixel_data_for_frame().to_vec(); + + let dxgi_format = match pixel_format { + scap_direct3d::PixelFormat::R8G8B8A8Unorm => { + ::windows::Win32::Graphics::Dxgi::Common::DXGI_FORMAT_R8G8B8A8_UNORM + } + scap_direct3d::PixelFormat::B8G8R8A8Unorm => { + ::windows::Win32::Graphics::Dxgi::Common::DXGI_FORMAT_B8G8R8A8_UNORM + } + }; + + let texture_desc = D3D11_TEXTURE2D_DESC { + Width: width, + Height: height, + MipLevels: 1, + ArraySize: 1, + Format: dxgi_format, + SampleDesc: DXGI_SAMPLE_DESC { + Count: 1, + Quality: 0, + }, + Usage: D3D11_USAGE_DEFAULT, + BindFlags: D3D11_BIND_SHADER_RESOURCE.0 as u32, + CPUAccessFlags: 0, + MiscFlags: 0, + }; + + let subresource_data = D3D11_SUBRESOURCE_DATA { + pSysMem: pixel_data.as_ptr() as *const _, + SysMemPitch: row_bytes as u32, + SysMemSlicePitch: 0, + }; + + let texture = unsafe { + let mut tex = None; + self.d3d_device + .CreateTexture2D(&texture_desc, Some(&subresource_data), Some(&mut tex)) + .ok()?; + tex? + }; + + Some(ScreenFrame::Scaled(ScaledScreenFrame { + texture, + pixel_data, + width, + height, + pixel_format, + })) } fn scale_frame(&mut self, frame: &scap_direct3d::Frame) -> Option { + if let Some(scaled) = self.scale_frame_gpu(frame) { + return Some(scaled); + } + let src_width = frame.width(); let src_height = frame.height(); @@ -657,6 +750,8 @@ fn create_d3d_capturer( ); if let Ok(mut guard) = frame_scaler.lock() { guard.state = None; + guard.gpu_scaler = None; + guard.gpu_scaler_source = None; } } ScreenFrame::Captured(frame) From 767b081a128452a8f268f47e5f411d69ac58c1c9 Mon Sep 17 00:00:00 2001 From: Aatman Jain Date: Thu, 29 Oct 2026 23:54:21 +0000 Subject: [PATCH 02/11] fix: resolve Windows gpu_scale module path and unsafe transmute Point the nested gpu_scale module at the sibling source file and wrap transmute_copy in an unsafe block for Rust 2024. --- crates/recording/src/sources/screen_capture/gpu_scale.rs | 2 +- crates/recording/src/sources/screen_capture/windows.rs | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/crates/recording/src/sources/screen_capture/gpu_scale.rs b/crates/recording/src/sources/screen_capture/gpu_scale.rs index 5003851b212..bb5df9d53fb 100644 --- a/crates/recording/src/sources/screen_capture/gpu_scale.rs +++ b/crates/recording/src/sources/screen_capture/gpu_scale.rs @@ -155,7 +155,7 @@ impl GpuTextureScaler { PastFrames: 0, FutureFrames: 0, ppPastSurfaces: ptr::null_mut(), - pInputSurface: std::mem::transmute_copy(&input_view), + pInputSurface: unsafe { std::mem::transmute_copy(&input_view) }, ppFutureSurfaces: ptr::null_mut(), ppPastSurfacesRight: ptr::null_mut(), pInputSurfaceRight: ManuallyDrop::new(None), diff --git a/crates/recording/src/sources/screen_capture/windows.rs b/crates/recording/src/sources/screen_capture/windows.rs index 83325cb1f3e..0241ba04df1 100644 --- a/crates/recording/src/sources/screen_capture/windows.rs +++ b/crates/recording/src/sources/screen_capture/windows.rs @@ -29,6 +29,7 @@ use std::{ use tokio_util::{future::FutureExt as _, sync::CancellationToken}; use tracing::*; +#[path = "gpu_scale.rs"] mod gpu_scale; use gpu_scale::GpuTextureScaler; From e077503888e06ec88a895f597010ddbb91e038fa Mon Sep 17 00:00:00 2001 From: Aatman Jain Date: Fri, 30 Oct 2026 00:01:47 +0000 Subject: [PATCH 03/11] fix(recording): ordered M4S conversion and BT.709 swscale Preserve capture sequence through the async converter pool, apply BT.709 limited-range coefficients for RGB swscale fallbacks, and cache failed GPU scaler setup per source size. --- crates/frame-converter/src/swscale.rs | 38 ++++++++++ .../src/output_pipeline/win_fragmented_m4s.rs | 75 ++++++++++++------- .../src/sources/screen_capture/windows.rs | 3 +- 3 files changed, 86 insertions(+), 30 deletions(-) diff --git a/crates/frame-converter/src/swscale.rs b/crates/frame-converter/src/swscale.rs index eafff292494..8f5f20e6f4a 100644 --- a/crates/frame-converter/src/swscale.rs +++ b/crates/frame-converter/src/swscale.rs @@ -3,6 +3,39 @@ use ffmpeg::{format::Pixel, frame, software::scaling}; use std::cell::UnsafeCell; use thread_local::ThreadLocal; +fn is_rgb(format: Pixel) -> bool { + format.descriptor().is_some_and(|descriptor| unsafe { + (*descriptor.as_ptr()).flags & ffmpeg::ffi::AV_PIX_FMT_FLAG_RGB as u64 != 0 + }) +} + +fn use_bt709_for_rgb_input( + converter: &mut scaling::Context, + input: Pixel, + output: Pixel, +) -> Result<(), ffmpeg::Error> { + if !is_rgb(input) || is_rgb(output) { + return Ok(()); + } + let result = unsafe { + let coefficients = ffmpeg::ffi::sws_getCoefficients(ffmpeg::ffi::SWS_CS_ITU709); + ffmpeg::ffi::sws_setColorspaceDetails( + converter.as_mut_ptr(), + coefficients, + 1, + coefficients, + 0, + 0, + 1 << 16, + 1 << 16, + ) + }; + if result < 0 { + return Err(ffmpeg::Error::from(result)); + } + Ok(()) +} + struct SendableContext(UnsafeCell); unsafe impl Send for SendableContext {} @@ -55,6 +88,11 @@ impl SwscaleConverter { flags, ) .map_err(|_| ConvertError::UnsupportedFormat(config.input_format, config.output_format)) + .and_then(|mut ctx| { + use_bt709_for_rgb_input(&mut ctx, config.input_format, config.output_format) + .map_err(|error| ConvertError::ConversionFailed(error.to_string()))?; + Ok(ctx) + }) } fn get_or_create_context(&self) -> Result<&SendableContext, ConvertError> { diff --git a/crates/recording/src/output_pipeline/win_fragmented_m4s.rs b/crates/recording/src/output_pipeline/win_fragmented_m4s.rs index 50edcb86e46..885ebcc2a81 100644 --- a/crates/recording/src/output_pipeline/win_fragmented_m4s.rs +++ b/crates/recording/src/output_pipeline/win_fragmented_m4s.rs @@ -1077,8 +1077,10 @@ fn run_win_m4s_encode_thread( use_gpu_converter: bool, ) -> anyhow::Result<()> { use cap_frame_converter::{ - AsyncConverterPool, ConversionConfig, ConvertError, ConverterPoolConfig, DropStrategy, + AsyncConverterPool, ConversionConfig, ConvertError, ConvertedFrame, ConverterPoolConfig, + DropStrategy, }; + use std::collections::BTreeMap; use std::sync::atomic::{AtomicU64, Ordering}; const SLOW_THRESHOLD_MS: u128 = 5; @@ -1133,28 +1135,46 @@ fn run_win_m4s_encode_thread( let frame_sequence = AtomicU64::new(0); let mut slow_encode_count = 0u32; let mut total_frames = 0u64; - - let mut drain_converted = |pool: &mut AsyncConverterPool| -> anyhow::Result<()> { - while let Some(converted) = pool.try_recv() { - let timestamp = Duration::from_micros(converted.frame.pts().unwrap_or(0).max(0) as u64); - let encode_start = std::time::Instant::now(); - match encoder.lock() { - Ok(mut enc) => { - if let Err(error) = enc.queue_preconverted_frame(converted.frame, timestamp) { - warn!("Failed to encode converted frame: {error}"); - } - } - Err(_) => { - return Err(anyhow!( - "Encoder mutex poisoned - all subsequent frames would be lost" - )); + let mut next_encode_sequence = 0u64; + let mut pending_converted: BTreeMap = BTreeMap::new(); + + let mut encode_converted = |converted: ConvertedFrame| { + let timestamp = Duration::from_micros(converted.frame.pts().unwrap_or(0).max(0) as u64); + let encode_start = std::time::Instant::now(); + match encoder.lock() { + Ok(mut enc) => { + if let Err(error) = enc.queue_preconverted_frame(converted.frame, timestamp) { + warn!("Failed to encode converted frame: {error}"); } } - let encode_elapsed_ms = encode_start.elapsed().as_millis(); - if encode_elapsed_ms > SLOW_THRESHOLD_MS { - slow_encode_count += 1; + Err(_) => { + return Err(anyhow!( + "Encoder mutex poisoned - all subsequent frames would be lost" + )); } - total_frames += 1; + } + let encode_elapsed_ms = encode_start.elapsed().as_millis(); + if encode_elapsed_ms > SLOW_THRESHOLD_MS { + slow_encode_count += 1; + } + total_frames += 1; + Ok(()) + }; + + let mut ingest_converted = |converted: ConvertedFrame| -> anyhow::Result<()> { + if converted.sequence < next_encode_sequence { + return Ok(()); + } + pending_converted.insert(converted.sequence, converted); + while let Some(converted) = pending_converted.remove(&next_encode_sequence) { + encode_converted(converted)?; + } + Ok(()) + }; + + let mut drain_converted = |pool: &mut AsyncConverterPool| -> anyhow::Result<()> { + while let Some(converted) = pool.try_recv() { + ingest_converted(converted)?; } Ok(()) }; @@ -1201,18 +1221,17 @@ fn run_win_m4s_encode_thread( if let Some(mut pool) = converter_pool { pool.drain_with_timeout( |converted| { - let timestamp = - Duration::from_micros(converted.frame.pts().unwrap_or(0).max(0) as u64); - if let Ok(mut enc) = encoder.lock() - && let Err(error) = enc.queue_preconverted_frame(converted.frame, timestamp) - { - warn!("Failed to encode drained converted frame: {error}"); - } else { - total_frames += 1; + if ingest_converted(converted).is_err() { + return; } }, Duration::from_secs(5), ); + while let Some(converted) = pending_converted.remove(&next_encode_sequence) { + if encode_converted(converted).is_err() { + break; + } + } } if total_frames > 0 { diff --git a/crates/recording/src/sources/screen_capture/windows.rs b/crates/recording/src/sources/screen_capture/windows.rs index 0241ba04df1..4ec9dc10353 100644 --- a/crates/recording/src/sources/screen_capture/windows.rs +++ b/crates/recording/src/sources/screen_capture/windows.rs @@ -29,7 +29,6 @@ use std::{ use tokio_util::{future::FutureExt as _, sync::CancellationToken}; use tracing::*; -#[path = "gpu_scale.rs"] mod gpu_scale; use gpu_scale::GpuTextureScaler; @@ -229,7 +228,7 @@ impl WindowsFrameScaler { self.target_width, self.target_height, ); - self.gpu_scaler_source = self.gpu_scaler.as_ref().map(|_| (src_width, src_height)); + self.gpu_scaler_source = Some((src_width, src_height)); } self.gpu_scaler.as_mut() From e9dd083e06e712e3885f6b0c9bfb7471565e4ebd Mon Sep 17 00:00:00 2001 From: Aatman Jain Date: Fri, 30 Oct 2026 09:32:08 +0000 Subject: [PATCH 04/11] fix(recording): resolve Windows gpu_scale module and disable scaler after runtime failure --- crates/recording/src/sources/screen_capture/windows.rs | 5 ++++- .../src/sources/screen_capture/{ => windows}/gpu_scale.rs | 4 ---- 2 files changed, 4 insertions(+), 5 deletions(-) rename crates/recording/src/sources/screen_capture/{ => windows}/gpu_scale.rs (99%) diff --git a/crates/recording/src/sources/screen_capture/windows.rs b/crates/recording/src/sources/screen_capture/windows.rs index 4ec9dc10353..c093368ecc2 100644 --- a/crates/recording/src/sources/screen_capture/windows.rs +++ b/crates/recording/src/sources/screen_capture/windows.rs @@ -238,7 +238,10 @@ impl WindowsFrameScaler { let src_width = frame.width(); let src_height = frame.height(); let scaler = self.ensure_gpu_scaler(src_width, src_height)?; - scaler.scale_texture(frame.texture())?; + if scaler.scale_texture(frame.texture()).is_none() { + self.gpu_scaler = None; + return None; + } let (width, height, pixel_format) = scaler.dimensions(); let row_bytes = (width * 4) as usize; diff --git a/crates/recording/src/sources/screen_capture/gpu_scale.rs b/crates/recording/src/sources/screen_capture/windows/gpu_scale.rs similarity index 99% rename from crates/recording/src/sources/screen_capture/gpu_scale.rs rename to crates/recording/src/sources/screen_capture/windows/gpu_scale.rs index bb5df9d53fb..ebf6022b733 100644 --- a/crates/recording/src/sources/screen_capture/gpu_scale.rs +++ b/crates/recording/src/sources/screen_capture/windows/gpu_scale.rs @@ -24,8 +24,6 @@ pub struct GpuTextureScaler { output_texture: ID3D11Texture2D, staging_output: ID3D11Texture2D, pixel_format: PixelFormat, - src_width: u32, - src_height: u32, dst_width: u32, dst_height: u32, row_bytes: usize, @@ -90,8 +88,6 @@ impl GpuTextureScaler { output_texture, staging_output, pixel_format, - src_width, - src_height, dst_width, dst_height, row_bytes, From 06e675cd3f8352fd1103796bd6527ea3ac619c5a Mon Sep 17 00:00:00 2001 From: Aatman Jain Date: Fri, 30 Oct 2026 09:32:08 +0000 Subject: [PATCH 05/11] fix(recording): reorder converted Windows M4S frames by sequence and skip lost ones --- crates/frame-converter/src/lib.rs | 3 + crates/frame-converter/src/reorder.rs | 306 ++++++++++++++++++ crates/recording/src/output_pipeline/mod.rs | 3 + .../src/output_pipeline/win_fragmented_m4s.rs | 199 +----------- .../output_pipeline/win_m4s_convert_encode.rs | 295 +++++++++++++++++ 5 files changed, 613 insertions(+), 193 deletions(-) create mode 100644 crates/frame-converter/src/reorder.rs create mode 100644 crates/recording/src/output_pipeline/win_m4s_convert_encode.rs diff --git a/crates/frame-converter/src/lib.rs b/crates/frame-converter/src/lib.rs index cbe4c581072..3c11a019cc0 100644 --- a/crates/frame-converter/src/lib.rs +++ b/crates/frame-converter/src/lib.rs @@ -7,6 +7,9 @@ pub use pool::*; mod frame_pool; pub use frame_pool::*; +mod reorder; +pub use reorder::*; + #[cfg(target_os = "macos")] mod videotoolbox; #[cfg(target_os = "macos")] diff --git a/crates/frame-converter/src/reorder.rs b/crates/frame-converter/src/reorder.rs new file mode 100644 index 00000000000..3b71a8ec3b5 --- /dev/null +++ b/crates/frame-converter/src/reorder.rs @@ -0,0 +1,306 @@ +use std::{ + collections::BTreeMap, + time::{Duration, Instant}, +}; + +pub struct SequenceReorderBuffer { + next_sequence: u64, + pending: BTreeMap, + max_pending: usize, + gap_timeout: Duration, + blocked_since: Option, + skipped: u64, + discarded: u64, +} + +impl SequenceReorderBuffer { + pub fn new(max_pending: usize, gap_timeout: Duration) -> Self { + Self { + next_sequence: 0, + pending: BTreeMap::new(), + max_pending: max_pending.max(1), + gap_timeout, + blocked_since: None, + skipped: 0, + discarded: 0, + } + } + + pub fn push(&mut self, sequence: u64, item: T) -> bool { + if sequence < self.next_sequence || self.pending.contains_key(&sequence) { + self.discarded += 1; + return false; + } + self.pending.insert(sequence, item); + true + } + + pub fn pop_ready(&mut self, now: Instant) -> Option { + if let Some(item) = self.pending.remove(&self.next_sequence) { + self.next_sequence = self.next_sequence.saturating_add(1); + self.blocked_since = None; + return Some(item); + } + + if self.pending.is_empty() { + self.blocked_since = None; + return None; + } + + let blocked_since = *self.blocked_since.get_or_insert(now); + if self.pending.len() > self.max_pending + || now.saturating_duration_since(blocked_since) >= self.gap_timeout + { + return self.pop_next_available(); + } + + None + } + + pub fn pop_next_available(&mut self) -> Option { + let (sequence, item) = self.pending.pop_first()?; + self.skipped += sequence.saturating_sub(self.next_sequence); + self.next_sequence = sequence.saturating_add(1); + self.blocked_since = None; + Some(item) + } + + pub fn len(&self) -> usize { + self.pending.len() + } + + pub fn is_empty(&self) -> bool { + self.pending.is_empty() + } + + pub fn skipped(&self) -> u64 { + self.skipped + } + + pub fn discarded(&self) -> u64 { + self.discarded + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const TIMEOUT: Duration = Duration::from_millis(250); + const JUST_UNDER_TIMEOUT: Duration = Duration::from_millis(249); + + fn buffer(max_pending: usize) -> SequenceReorderBuffer { + SequenceReorderBuffer::new(max_pending, TIMEOUT) + } + + fn drain_ready(buffer: &mut SequenceReorderBuffer, now: Instant) -> Vec { + std::iter::from_fn(|| buffer.pop_ready(now)).collect() + } + + #[test] + fn in_order_frames_are_released_immediately_and_keep_advancing() { + let now = Instant::now(); + let mut buffer = buffer(8); + + let mut released = Vec::new(); + for sequence in 0..50 { + buffer.push(sequence, sequence); + released.extend(drain_ready(&mut buffer, now)); + } + + assert_eq!(released, (0..50).collect::>()); + assert!(buffer.is_empty()); + assert_eq!(buffer.skipped(), 0); + assert_eq!(buffer.discarded(), 0); + } + + #[test] + fn out_of_order_arrival_is_released_in_sequence_order() { + let now = Instant::now(); + let mut buffer = buffer(8); + + buffer.push(2, 2); + assert_eq!(drain_ready(&mut buffer, now), Vec::::new()); + + buffer.push(0, 0); + assert_eq!(drain_ready(&mut buffer, now), vec![0]); + + buffer.push(1, 1); + assert_eq!(drain_ready(&mut buffer, now), vec![1, 2]); + + buffer.push(4, 4); + buffer.push(3, 3); + buffer.push(5, 5); + assert_eq!(drain_ready(&mut buffer, now), vec![3, 4, 5]); + assert!(buffer.is_empty()); + assert_eq!(buffer.skipped(), 0); + } + + #[test] + fn missing_frame_blocks_until_the_gap_timeout_then_is_skipped() { + let start = Instant::now(); + let mut buffer = buffer(8); + + buffer.push(0, 0); + buffer.push(2, 2); + buffer.push(3, 3); + + assert_eq!(drain_ready(&mut buffer, start), vec![0]); + assert_eq!( + drain_ready(&mut buffer, start + JUST_UNDER_TIMEOUT), + Vec::::new() + ); + assert_eq!(drain_ready(&mut buffer, start + TIMEOUT), vec![2, 3]); + assert_eq!(buffer.skipped(), 1); + assert!(buffer.is_empty()); + } + + #[test] + fn frames_after_a_skipped_gap_keep_flowing_without_waiting() { + let start = Instant::now(); + let mut buffer = buffer(8); + + buffer.push(1, 1); + assert_eq!(drain_ready(&mut buffer, start), Vec::::new()); + assert_eq!(drain_ready(&mut buffer, start + TIMEOUT), vec![1]); + + let later = start + TIMEOUT; + buffer.push(2, 2); + buffer.push(3, 3); + assert_eq!(drain_ready(&mut buffer, later), vec![2, 3]); + assert_eq!(buffer.skipped(), 1); + } + + #[test] + fn gap_timer_restarts_after_progress() { + let start = Instant::now(); + let mut buffer = buffer(8); + + buffer.push(0, 0); + buffer.push(2, 2); + assert_eq!(drain_ready(&mut buffer, start), vec![0]); + + let later = start + Duration::from_millis(240); + buffer.push(1, 1); + assert_eq!(drain_ready(&mut buffer, later), vec![1, 2]); + + buffer.push(4, 4); + assert_eq!(drain_ready(&mut buffer, later), Vec::::new()); + assert_eq!( + drain_ready(&mut buffer, later + JUST_UNDER_TIMEOUT), + Vec::::new() + ); + assert_eq!(drain_ready(&mut buffer, later + TIMEOUT), vec![4]); + assert_eq!(buffer.skipped(), 1); + } + + #[test] + fn too_many_pending_frames_skip_the_gap_without_waiting_for_the_timeout() { + let now = Instant::now(); + let mut buffer = buffer(3); + + for sequence in 2..=4 { + buffer.push(sequence, sequence); + } + assert_eq!(drain_ready(&mut buffer, now), Vec::::new()); + assert_eq!(buffer.len(), 3); + + buffer.push(5, 5); + assert_eq!(drain_ready(&mut buffer, now), vec![2, 3, 4, 5]); + assert_eq!(buffer.skipped(), 2); + assert!(buffer.is_empty()); + } + + #[test] + fn frames_arriving_after_their_gap_was_skipped_are_discarded() { + let start = Instant::now(); + let mut buffer = buffer(8); + + buffer.push(1, 1); + assert_eq!(drain_ready(&mut buffer, start + TIMEOUT), Vec::::new()); + assert_eq!(drain_ready(&mut buffer, start + TIMEOUT * 2), vec![1]); + + assert!(!buffer.push(0, 0)); + assert!(!buffer.push(1, 1)); + assert_eq!(buffer.discarded(), 2); + assert!(buffer.is_empty()); + + assert!(buffer.push(2, 2)); + assert_eq!(drain_ready(&mut buffer, start + TIMEOUT * 2), vec![2]); + } + + #[test] + fn duplicate_pending_sequence_is_rejected() { + let mut buffer = buffer(8); + + assert!(buffer.push(3, 3)); + assert!(!buffer.push(3, 30)); + assert_eq!(buffer.len(), 1); + assert_eq!(buffer.discarded(), 1); + assert_eq!(buffer.pop_next_available(), Some(3)); + } + + #[test] + fn flush_releases_everything_in_order_across_gaps() { + let now = Instant::now(); + let mut buffer = buffer(16); + + for sequence in [6, 2, 3, 9, 0] { + buffer.push(sequence, sequence); + } + assert_eq!(drain_ready(&mut buffer, now), vec![0]); + + let mut flushed = Vec::new(); + while let Some(item) = buffer.pop_next_available() { + flushed.push(item); + } + + assert_eq!(flushed, vec![2, 3, 6, 9]); + assert!(buffer.is_empty()); + assert_eq!(buffer.skipped(), 1 + 2 + 2); + assert_eq!(buffer.pop_next_available(), None); + } + + #[test] + fn lost_first_frame_does_not_stall_the_stream() { + let start = Instant::now(); + let mut buffer = buffer(8); + + for sequence in 1..=5 { + buffer.push(sequence, sequence); + } + + assert_eq!(drain_ready(&mut buffer, start), Vec::::new()); + assert_eq!( + drain_ready(&mut buffer, start + TIMEOUT), + vec![1, 2, 3, 4, 5] + ); + assert_eq!(buffer.skipped(), 1); + } + + #[test] + fn failed_conversions_scattered_through_a_long_stream_never_stall_it() { + let start = Instant::now(); + let mut buffer = buffer(16); + let mut clock = start; + let mut released = Vec::new(); + + for sequence in 0..300u64 { + if sequence % 7 == 3 { + continue; + } + buffer.push(sequence, sequence); + clock += Duration::from_millis(33); + released.extend(drain_ready(&mut buffer, clock)); + } + clock += TIMEOUT; + released.extend(drain_ready(&mut buffer, clock)); + while let Some(item) = buffer.pop_next_available() { + released.push(item); + } + + let expected: Vec = (0..300).filter(|sequence| sequence % 7 != 3).collect(); + assert_eq!(released, expected); + assert_eq!(buffer.skipped(), 300 - expected.len() as u64); + } +} diff --git a/crates/recording/src/output_pipeline/mod.rs b/crates/recording/src/output_pipeline/mod.rs index 25ecb5c52aa..2fa7df9854a 100644 --- a/crates/recording/src/output_pipeline/mod.rs +++ b/crates/recording/src/output_pipeline/mod.rs @@ -62,6 +62,9 @@ mod win_fragmented_m4s; #[cfg(windows)] pub use win_fragmented_m4s::*; +#[cfg(any(windows, test))] +mod win_m4s_convert_encode; + #[cfg(windows)] mod oop_fragmented_m4s_win; #[cfg(windows)] diff --git a/crates/recording/src/output_pipeline/win_fragmented_m4s.rs b/crates/recording/src/output_pipeline/win_fragmented_m4s.rs index 885ebcc2a81..a511b682b07 100644 --- a/crates/recording/src/output_pipeline/win_fragmented_m4s.rs +++ b/crates/recording/src/output_pipeline/win_fragmented_m4s.rs @@ -5,6 +5,7 @@ use super::core::{ use super::frame_pacing::{FramePacer, Paced}; use super::mjpeg::MjpegDecoder; use super::win::reference_video_frame; +use super::win_m4s_convert_encode::{StagedFrame, run_win_m4s_encode_thread}; use crate::{ AudioFrame, AudioMuxer, Muxer, SharedPauseState, TaskPool, VideoMuxer, output_pipeline::{NativeCameraFrame, camera_frame_to_ffmpeg}, @@ -30,7 +31,6 @@ use tracing::*; const DEFAULT_MUXER_BUFFER_SIZE: usize = 240; const DEFAULT_CONVERT_STAGING_DEPTH: usize = 32; -const DEFAULT_GPU_CONVERTER_WORKERS: usize = 2; fn get_convert_staging_depth() -> usize { std::env::var("CAP_WIN_CONVERT_STAGING_DEPTH") @@ -257,11 +257,6 @@ impl Default for WindowsFragmentedM4SMuxerConfig { } } -enum StagedFfmpegFrame { - Ready(ffmpeg::frame::Video, Duration), - Shutdown, -} - impl Muxer for WindowsFragmentedM4SMuxer { type Config = WindowsFragmentedM4SMuxerConfig; @@ -377,17 +372,15 @@ impl WindowsFragmentedM4SMuxer { encoder.set_segment_callback(tx); } let encoder = Arc::new(Mutex::new(encoder)); - let encoder_clone = encoder.clone(); let video_config = self.video_config; let health_tx = self.health_tx.clone(); let base_path = self.base_path.clone(); let pause_flag = self.pause.flag(); - let staging_depth = get_convert_staging_depth(); let (staging_tx, staging_rx) = - sync_channel::>(staging_depth.max(4)); + sync_channel::(get_convert_staging_depth().max(4)); let encode_encoder = encoder.clone(); - let use_gpu_converter = + let use_converter_pool = prefer_external_conversion && conversion_requirements.needs_conversion; let encode_handle = std::thread::Builder::new() .name("win-m4s-segment-encode".to_string()) @@ -396,8 +389,8 @@ impl WindowsFragmentedM4SMuxer { run_win_m4s_encode_thread( staging_rx, encode_encoder, - conversion_requirements, - use_gpu_converter, + &conversion_requirements, + use_converter_pool, ) })?; let encoder_handle = std::thread::Builder::new() @@ -434,7 +427,7 @@ impl WindowsFragmentedM4SMuxer { timestamp: Duration| -> anyhow::Result<()> { staging_tx - .send(Some(StagedFfmpegFrame::Ready(ffmpeg_frame, timestamp))) + .send((ffmpeg_frame, timestamp)) .map_err(|_| anyhow!("Windows M4S convert staging channel disconnected")) }; @@ -533,7 +526,6 @@ impl WindowsFragmentedM4SMuxer { total_frames += 1; } - let _ = staging_tx.send(Some(StagedFfmpegFrame::Shutdown)); drop(staging_tx); let encode_result = encode_handle @@ -1070,185 +1062,6 @@ impl AudioMuxer for WindowsFragmentedM4SCameraMuxer { } } -fn run_win_m4s_encode_thread( - staging_rx: std::sync::mpsc::Receiver>, - encoder: Arc>, - requirements: cap_enc_ffmpeg::h264::ConversionRequirements, - use_gpu_converter: bool, -) -> anyhow::Result<()> { - use cap_frame_converter::{ - AsyncConverterPool, ConversionConfig, ConvertError, ConvertedFrame, ConverterPoolConfig, - DropStrategy, - }; - use std::collections::BTreeMap; - use std::sync::atomic::{AtomicU64, Ordering}; - - const SLOW_THRESHOLD_MS: u128 = 5; - - let worker_count = std::thread::available_parallelism() - .map(|parallelism| parallelism.get().clamp(2, 4)) - .unwrap_or(DEFAULT_GPU_CONVERTER_WORKERS); - - let mut converter_pool = if use_gpu_converter { - let conversion_config = ConversionConfig::new( - requirements.input_format, - requirements.input_width, - requirements.input_height, - requirements.output_format, - requirements.output_width, - requirements.output_height, - ); - let pool_config = ConverterPoolConfig { - worker_count, - input_capacity: 120, - output_capacity: 90, - drop_strategy: DropStrategy::DropOldest, - ..Default::default() - }; - Some( - AsyncConverterPool::from_config(conversion_config, pool_config) - .map_err(|error| anyhow!("Failed to create Windows M4S converter pool: {error}"))?, - ) - } else { - None - }; - - if converter_pool.is_some() { - debug!( - worker_count = worker_count, - input = %format!( - "{:?} {}x{}", - requirements.input_format, - requirements.input_width, - requirements.input_height - ), - output = %format!( - "{:?} {}x{}", - requirements.output_format, - requirements.output_width, - requirements.output_height - ), - "Windows M4S encoder using async GPU/CPU converter pool" - ); - } - - let frame_sequence = AtomicU64::new(0); - let mut slow_encode_count = 0u32; - let mut total_frames = 0u64; - let mut next_encode_sequence = 0u64; - let mut pending_converted: BTreeMap = BTreeMap::new(); - - let mut encode_converted = |converted: ConvertedFrame| { - let timestamp = Duration::from_micros(converted.frame.pts().unwrap_or(0).max(0) as u64); - let encode_start = std::time::Instant::now(); - match encoder.lock() { - Ok(mut enc) => { - if let Err(error) = enc.queue_preconverted_frame(converted.frame, timestamp) { - warn!("Failed to encode converted frame: {error}"); - } - } - Err(_) => { - return Err(anyhow!( - "Encoder mutex poisoned - all subsequent frames would be lost" - )); - } - } - let encode_elapsed_ms = encode_start.elapsed().as_millis(); - if encode_elapsed_ms > SLOW_THRESHOLD_MS { - slow_encode_count += 1; - } - total_frames += 1; - Ok(()) - }; - - let mut ingest_converted = |converted: ConvertedFrame| -> anyhow::Result<()> { - if converted.sequence < next_encode_sequence { - return Ok(()); - } - pending_converted.insert(converted.sequence, converted); - while let Some(converted) = pending_converted.remove(&next_encode_sequence) { - encode_converted(converted)?; - } - Ok(()) - }; - - let mut drain_converted = |pool: &mut AsyncConverterPool| -> anyhow::Result<()> { - while let Some(converted) = pool.try_recv() { - ingest_converted(converted)?; - } - Ok(()) - }; - - loop { - match staging_rx.recv() { - Ok(Some(StagedFfmpegFrame::Ready(mut frame, timestamp))) => { - if let Some(pool) = &mut converter_pool { - frame.set_pts(Some(timestamp.as_micros() as i64)); - match pool.submit(frame, frame_sequence.fetch_add(1, Ordering::Relaxed)) { - Ok(()) => {} - Err(ConvertError::PoolShutdown) => break, - Err(error) => { - warn!("Failed to submit frame to converter pool: {error}"); - } - } - drain_converted(pool)?; - } else { - let encode_start = std::time::Instant::now(); - match encoder.lock() { - Ok(mut enc) => { - if let Err(error) = enc.queue_frame(frame, timestamp) { - warn!("Failed to encode frame: {error}"); - } - } - Err(_) => { - return Err(anyhow!( - "Encoder mutex poisoned - all subsequent frames would be lost" - )); - } - } - let encode_elapsed_ms = encode_start.elapsed().as_millis(); - if encode_elapsed_ms > SLOW_THRESHOLD_MS { - slow_encode_count += 1; - } - total_frames += 1; - } - } - Ok(Some(StagedFfmpegFrame::Shutdown)) | Ok(None) => break, - Err(_) => break, - } - } - - if let Some(mut pool) = converter_pool { - pool.drain_with_timeout( - |converted| { - if ingest_converted(converted).is_err() { - return; - } - }, - Duration::from_secs(5), - ); - while let Some(converted) = pending_converted.remove(&next_encode_sequence) { - if encode_converted(converted).is_err() { - break; - } - } - } - - if total_frames > 0 { - debug!( - total_frames = total_frames, - slow_encodes = slow_encode_count, - slow_encode_pct = format!( - "{:.1}%", - 100.0 * slow_encode_count as f64 / total_frames as f64 - ), - "Windows M4S encode timing summary" - ); - } - - Ok(()) -} - #[cfg(test)] mod tests { use super::reference_video_frame; diff --git a/crates/recording/src/output_pipeline/win_m4s_convert_encode.rs b/crates/recording/src/output_pipeline/win_m4s_convert_encode.rs new file mode 100644 index 00000000000..abf56add782 --- /dev/null +++ b/crates/recording/src/output_pipeline/win_m4s_convert_encode.rs @@ -0,0 +1,295 @@ +use anyhow::anyhow; +use cap_enc_ffmpeg::{h264::ConversionRequirements, segmented_stream::SegmentedVideoEncoder}; +use cap_frame_converter::{ + AsyncConverterPool, ConversionConfig, ConvertError, ConvertedFrame, ConverterPoolConfig, + DropStrategy, SequenceReorderBuffer, +}; +use std::{ + sync::{ + Arc, Mutex, + mpsc::{Receiver, RecvTimeoutError}, + }, + time::{Duration, Instant}, +}; +use tracing::*; + +const DEFAULT_CONVERTER_WORKERS: usize = 2; +const CONVERTER_INPUT_CAPACITY: usize = 120; +const CONVERTER_OUTPUT_CAPACITY: usize = 90; +const REORDER_MAX_PENDING: usize = 16; +const REORDER_GAP_TIMEOUT: Duration = Duration::from_millis(250); +const CONVERTED_POLL_INTERVAL: Duration = Duration::from_millis(10); +const POOL_DRAIN_TIMEOUT: Duration = Duration::from_secs(5); +const SLOW_ENCODE_THRESHOLD_MS: u128 = 5; + +pub(super) type StagedFrame = (ffmpeg::frame::Video, Duration); + +pub(super) fn run_win_m4s_encode_thread( + staging_rx: Receiver, + encoder: Arc>, + requirements: &ConversionRequirements, + use_converter_pool: bool, +) -> anyhow::Result<()> { + let mut converter_pool = if use_converter_pool { + Some(create_converter_pool(requirements)?) + } else { + None + }; + let mut sink = EncodeSink::new(encoder); + let mut next_sequence = 0u64; + + loop { + match staging_rx.recv_timeout(CONVERTED_POLL_INTERVAL) { + Ok((mut frame, timestamp)) => { + let Some(pool) = &converter_pool else { + sink.encode_raw(frame, timestamp)?; + continue; + }; + frame.set_pts(Some(timestamp.as_micros() as i64)); + let sequence = next_sequence; + next_sequence += 1; + match pool.submit(frame, sequence) { + Ok(()) => {} + Err(ConvertError::PoolShutdown) => break, + Err(error) => warn!("Failed to submit frame to converter pool: {error}"), + } + sink.drain_pool(pool)?; + } + Err(RecvTimeoutError::Timeout) => { + if let Some(pool) = &converter_pool { + sink.drain_pool(pool)?; + } + } + Err(RecvTimeoutError::Disconnected) => break, + } + } + + if let Some(mut pool) = converter_pool.take() { + let mut drain_error = None; + pool.drain_with_timeout( + |converted| { + if drain_error.is_none() + && let Err(error) = sink.ingest(converted) + { + drain_error = Some(error); + } + }, + POOL_DRAIN_TIMEOUT, + ); + if let Some(error) = drain_error { + return Err(error); + } + } + + sink.finish() +} + +fn create_converter_pool( + requirements: &ConversionRequirements, +) -> anyhow::Result { + let worker_count = std::thread::available_parallelism() + .map(|parallelism| parallelism.get().clamp(2, 4)) + .unwrap_or(DEFAULT_CONVERTER_WORKERS); + + let conversion_config = ConversionConfig::new( + requirements.input_format, + requirements.input_width, + requirements.input_height, + requirements.output_format, + requirements.output_width, + requirements.output_height, + ); + let pool_config = ConverterPoolConfig { + worker_count, + input_capacity: CONVERTER_INPUT_CAPACITY, + output_capacity: CONVERTER_OUTPUT_CAPACITY, + drop_strategy: DropStrategy::DropOldest, + ..Default::default() + }; + + let pool = AsyncConverterPool::from_config(conversion_config, pool_config) + .map_err(|error| anyhow!("Failed to create Windows M4S converter pool: {error}"))?; + + debug!( + worker_count = worker_count, + input = %format!( + "{:?} {}x{}", + requirements.input_format, requirements.input_width, requirements.input_height + ), + output = %format!( + "{:?} {}x{}", + requirements.output_format, requirements.output_width, requirements.output_height + ), + "Windows M4S encoder using async converter pool" + ); + + Ok(pool) +} + +struct EncodeSink { + encoder: Arc>, + reorder: SequenceReorderBuffer, + slow_encode_count: u32, + total_frames: u64, +} + +impl EncodeSink { + fn new(encoder: Arc>) -> Self { + Self { + encoder, + reorder: SequenceReorderBuffer::new(REORDER_MAX_PENDING, REORDER_GAP_TIMEOUT), + slow_encode_count: 0, + total_frames: 0, + } + } + + fn ingest(&mut self, converted: ConvertedFrame) -> anyhow::Result<()> { + self.reorder.push(converted.sequence, converted); + self.encode_ready() + } + + fn drain_pool(&mut self, pool: &AsyncConverterPool) -> anyhow::Result<()> { + while let Some(converted) = pool.try_recv() { + self.reorder.push(converted.sequence, converted); + } + self.encode_ready() + } + + fn encode_ready(&mut self) -> anyhow::Result<()> { + while let Some(converted) = self.reorder.pop_ready(Instant::now()) { + self.encode_converted(converted)?; + } + Ok(()) + } + + fn encode_converted(&mut self, converted: ConvertedFrame) -> anyhow::Result<()> { + let timestamp = Duration::from_micros(converted.frame.pts().unwrap_or(0).max(0) as u64); + self.encode_with(|encoder| encoder.queue_preconverted_frame(converted.frame, timestamp)) + } + + fn encode_raw( + &mut self, + frame: ffmpeg::frame::Video, + timestamp: Duration, + ) -> anyhow::Result<()> { + self.encode_with(|encoder| encoder.queue_frame(frame, timestamp)) + } + + fn encode_with( + &mut self, + queue: impl FnOnce(&mut SegmentedVideoEncoder) -> Result<(), E>, + ) -> anyhow::Result<()> { + let encode_start = Instant::now(); + { + let mut encoder = self.encoder.lock().map_err(|_| { + anyhow!("Encoder mutex poisoned - all subsequent frames would be lost") + })?; + if let Err(error) = queue(&mut encoder) { + warn!("Failed to encode frame: {error}"); + } + } + if encode_start.elapsed().as_millis() > SLOW_ENCODE_THRESHOLD_MS { + self.slow_encode_count += 1; + } + self.total_frames += 1; + Ok(()) + } + + fn finish(mut self) -> anyhow::Result<()> { + while let Some(converted) = self.reorder.pop_next_available() { + self.encode_converted(converted)?; + } + + if self.reorder.skipped() > 0 || self.reorder.discarded() > 0 { + warn!( + skipped = self.reorder.skipped(), + discarded = self.reorder.discarded(), + "Windows M4S converter pool lost frames before encode" + ); + } + + if self.total_frames > 0 { + debug!( + total_frames = self.total_frames, + slow_encodes = self.slow_encode_count, + slow_encode_pct = format!( + "{:.1}%", + 100.0 * self.slow_encode_count as f64 / self.total_frames as f64 + ), + "Windows M4S encode timing summary" + ); + } + + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use cap_enc_ffmpeg::{EncodedPacketStats, segmented_stream::SegmentedVideoEncoderConfig}; + use cap_media_info::{Pixel, VideoInfo}; + use std::sync::mpsc::sync_channel; + + const FRAME_COUNT: u64 = 60; + + fn bgra_frame(shade: u8) -> ffmpeg::frame::Video { + let mut frame = ffmpeg::frame::Video::new(ffmpeg::format::Pixel::BGRA, 320, 240); + frame.data_mut(0).fill(shade); + frame + } + + #[test] + fn converter_pool_keeps_encoding_after_the_first_frame() { + ffmpeg::init().unwrap(); + let temp = tempfile::tempdir().unwrap(); + let mut encoder = SegmentedVideoEncoder::init_with_timed_fragments( + temp.path().to_path_buf(), + VideoInfo { + pixel_format: Pixel::BGRA, + width: 320, + height: 240, + time_base: ffmpeg::Rational(1, 1_000_000), + frame_rate: ffmpeg::Rational(30, 1), + }, + SegmentedVideoEncoderConfig { + output_size: Some((160, 120)), + prefer_external_conversion: true, + ..Default::default() + }, + false, + ) + .unwrap(); + let requirements = encoder.conversion_requirements(); + assert!(requirements.needs_conversion); + let stats = Arc::new(EncodedPacketStats::default()); + encoder + .current_encoder_mut() + .unwrap() + .set_packet_stats(stats.clone()); + let encoder = Arc::new(Mutex::new(encoder)); + + let (staging_tx, staging_rx) = sync_channel::(8); + let encode_encoder = encoder.clone(); + let handle = std::thread::spawn(move || { + run_win_m4s_encode_thread(staging_rx, encode_encoder, &requirements, true) + }); + for index in 0..FRAME_COUNT { + staging_tx + .send(( + bgra_frame((index * 4) as u8), + Duration::from_micros(index * 33_333), + )) + .unwrap(); + } + drop(staging_tx); + handle.join().unwrap().unwrap(); + encoder.lock().unwrap().finish().unwrap(); + + let encoded = stats.snapshot().len() as u64; + assert!( + encoded >= FRAME_COUNT - FRAME_COUNT / 10, + "only {encoded} of {FRAME_COUNT} frames reached the encoder" + ); + } +} From 9931c672e418480977d1beb366ee483c05e740ae Mon Sep 17 00:00:00 2001 From: Aatman Jain Date: Fri, 30 Oct 2026 09:32:08 +0000 Subject: [PATCH 06/11] test(frame-converter): cover BT.709 swscale RGB conversion --- crates/frame-converter/src/swscale.rs | 52 +++++++++++++++++++++++++++ 1 file changed, 52 insertions(+) diff --git a/crates/frame-converter/src/swscale.rs b/crates/frame-converter/src/swscale.rs index 8f5f20e6f4a..b0b1067d0db 100644 --- a/crates/frame-converter/src/swscale.rs +++ b/crates/frame-converter/src/swscale.rs @@ -136,3 +136,55 @@ impl FrameConverter for SwscaleConverter { unsafe impl Send for SwscaleConverter {} unsafe impl Sync for SwscaleConverter {} + +#[cfg(test)] +mod tests { + use super::*; + + fn convert_red(input: Pixel, output: Pixel) -> frame::Video { + ffmpeg::init().unwrap(); + let converter = + SwscaleConverter::new(ConversionConfig::new(input, 16, 16, output, 16, 16)).unwrap(); + let mut rgb = frame::Video::new(input, 16, 16); + let stride = rgb.stride(0); + let pixel: [u8; 4] = if input == Pixel::BGRA { + [40, 40, 220, 255] + } else { + [220, 40, 40, 255] + }; + for row in 0..16 { + for col in 0..16 { + let offset = row * stride + col * 4; + rgb.data_mut(0)[offset..offset + 4].copy_from_slice(&pixel); + } + } + converter.convert(rgb).unwrap() + } + + #[test] + fn rgb_to_yuv_uses_bt709_limited_range() { + // (220, 40, 40) is Y 83.2, Cb 109.9, Cr 207.1 in BT.709 limited range; + // swscale's BT.601 default would give Y 96. + for input in [Pixel::BGRA, Pixel::RGBA] { + let nv12 = convert_red(input, Pixel::NV12); + let (y, cb, cr) = (nv12.data(0)[0], nv12.data(1)[0], nv12.data(1)[1]); + assert!( + y.abs_diff(83) <= 1 && cb.abs_diff(110) <= 1 && cr.abs_diff(207) <= 1, + "{input:?} -> NV12: y={y} cb={cb} cr={cr}" + ); + + let yuv420p = convert_red(input, Pixel::YUV420P); + let (y, cb, cr) = (yuv420p.data(0)[0], yuv420p.data(1)[0], yuv420p.data(2)[0]); + assert!( + y.abs_diff(83) <= 1 && cb.abs_diff(110) <= 1 && cr.abs_diff(207) <= 1, + "{input:?} -> YUV420P: y={y} cb={cb} cr={cr}" + ); + } + } + + #[test] + fn rgb_to_rgb_conversion_is_left_untouched() { + let rgba = convert_red(Pixel::BGRA, Pixel::RGBA); + assert_eq!(&rgba.data(0)[0..4], &[220, 40, 40, 255]); + } +} From b6870e3ba0796ba37c02c8a3829d920c00f6fc24 Mon Sep 17 00:00:00 2001 From: Aatman Jain Date: Fri, 30 Oct 2026 09:32:08 +0000 Subject: [PATCH 07/11] fix(frame-converter): use BT.709 for D3D11 RGB to YCbCr conversion --- crates/frame-converter/src/d3d11.rs | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/crates/frame-converter/src/d3d11.rs b/crates/frame-converter/src/d3d11.rs index 863d57fdcb9..3fc69511a9f 100644 --- a/crates/frame-converter/src/d3d11.rs +++ b/crates/frame-converter/src/d3d11.rs @@ -533,6 +533,16 @@ impl D3D11Converter { })? }; + if is_rgb_format(config.input_format) && !is_rgb_format(config.output_format) { + let output_color_space = + windows::Win32::Graphics::Direct3D11::D3D11_VIDEO_PROCESSOR_COLOR_SPACE { + _bitfield: YCBCR_MATRIX_BT709, + }; + unsafe { + video_context.VideoProcessorSetOutputColorSpace(&processor, &output_color_space); + } + } + let (input_texture, input_shared_handle) = create_shared_texture( &device, config.input_width, @@ -873,6 +883,15 @@ impl FrameConverter for D3D11Converter { } } +// `D3D11_VIDEO_PROCESSOR_COLOR_SPACE.YCbCr_Matrix` (bit 2): the video processor +// converts RGB to YCbCr with BT.601 unless told otherwise, while every encoder +// stream is tagged BT.709. +const YCBCR_MATRIX_BT709: u32 = 1 << 2; + +fn is_rgb_format(pixel: Pixel) -> bool { + matches!(pixel, Pixel::BGRA | Pixel::RGBA) +} + fn pixel_to_dxgi(pixel: Pixel) -> Result { match pixel { Pixel::NV12 => Ok(DXGI_FORMAT_NV12), From 4134761861f9c7f2c9eb2b89b5c640f00dd72980 Mon Sep 17 00:00:00 2001 From: Aatman Jain Date: Fri, 30 Oct 2026 10:23:07 +0000 Subject: [PATCH 08/11] fix(ci): configure Hyprland 0.57 through hyprland.lua in the Linux harness Hyprland 0.57 dropped hyprland.conf and only reads hyprland.lua, so the harness's monitor mode, session export and waybar were ignored and the output came up at vkms's preferred 1024x768. The screenshot scenario then reported screenshot_size=1024x768 instead of yes. Write the same settings as Lua and use the Lua form of the focus dispatcher. --- scripts/linux-desktop-harness/arch-up.sh | 17 ++++++++++------- .../scenario-screenshot.sh | 2 +- 2 files changed, 11 insertions(+), 8 deletions(-) diff --git a/scripts/linux-desktop-harness/arch-up.sh b/scripts/linux-desktop-harness/arch-up.sh index 6a1a8028862..f3a883510d6 100755 --- a/scripts/linux-desktop-harness/arch-up.sh +++ b/scripts/linux-desktop-harness/arch-up.sh @@ -23,13 +23,16 @@ common=(LIBSEAT_BACKEND=seatd XDG_SESSION_TYPE=wayland) case "$desktop" in hyprland) - cat >"$HOME/.config/hypr/hyprland.conf" <"$HOME/.config/hypr/hyprland.lua" <"$HOME/.config/hypr/xdph.conf" < /^(Cap|so\.cap\.Cap)$/.test(w.get_wm_class() || '') && w.showing_on_its_workspace()).forEach(w => w.activate(global.display.get_current_time_roundtrip()))" \ >/dev/null 2>&1 elif [[ -n ${HYPRLAND_INSTANCE_SIGNATURE:-} ]]; then - hyprctl dispatch focuswindow class:Cap >/dev/null 2>&1 + hyprctl dispatch "hl.dsp.focus({ window = 'class:Cap' })" >/dev/null 2>&1 fi } for _ in $(seq 60); do From bc2941c46b39880d477cedc1ba750cabfe561716 Mon Sep 17 00:00:00 2001 From: Aatman Jain Date: Fri, 30 Oct 2026 10:25:00 +0000 Subject: [PATCH 09/11] fix(ci): load videodev without linux-modules-extra for the Instant camera job The ubuntu-24.04 runner's 7.0.0-1017-azure kernel has no linux-modules-extra package, so apt-get failed before the camera test ran. That kernel ships videodev in its base modules (the desktop workflow's X11 camera step loads v4l2loopback there with only the headers), so install the extra modules only when videodev is missing and the package exists, install the headers v4l2loopback-dkms builds against, and fail clearly when videodev still cannot load. --- .github/workflows/instant-reliability.yml | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/.github/workflows/instant-reliability.yml b/.github/workflows/instant-reliability.yml index af99f5174a4..d1094d3378e 100644 --- a/.github/workflows/instant-reliability.yml +++ b/.github/workflows/instant-reliability.yml @@ -316,9 +316,19 @@ jobs: run: | set -euo pipefail sudo apt-get update -qq - # v4l2loopback needs videodev, which is in the extra kernel modules. - sudo apt-get install -y -qq "linux-modules-extra-$(uname -r)" v4l2loopback-dkms xvfb x11-utils xdotool \ - dbus ffmpeg pulseaudio pulseaudio-utils libasound2-plugins >/dev/null + packages=("linux-headers-$(uname -r)" v4l2loopback-dkms xvfb x11-utils xdotool + dbus ffmpeg pulseaudio pulseaudio-utils libasound2-plugins) + # v4l2loopback needs videodev. Newer runner kernels ship it in the + # base modules package and have no linux-modules-extra package. + extra="linux-modules-extra-$(uname -r)" + if ! sudo modprobe videodev 2>/dev/null && apt-cache show "$extra" >/dev/null 2>&1; then + packages+=("$extra") + fi + sudo apt-get install -y -qq "${packages[@]}" >/dev/null + if ! sudo modprobe videodev; then + echo "::error::videodev is not available for kernel $(uname -r), so v4l2loopback cannot load" + exit 1 + fi chmod +x "$RUNNER_TEMP/build/cap-gpui" "$RUNNER_TEMP/build/real-app-tests" sudo modprobe v4l2loopback video_nr=42 card_label="Cap Test Camera" exclusive_caps=1 sudo udevadm settle --timeout=10 From d32ac8930288644262458da27fdfd104c1d391b7 Mon Sep 17 00:00:00 2001 From: Aatman Jain Date: Fri, 30 Oct 2026 10:38:26 +0000 Subject: [PATCH 10/11] fix(verify): encode the 320x180 compositing codec case at the Web preset bitrate At 80 kbps the paravirtualised VideoToolbox HEVC encoder on the CI macOS VMs occasionally left a blurred camera code cell at margin 0.17, below the reader's 0.18 threshold. 138,240 bps is the Web export preset (0.08 bpp), still below the Social preset the verify exports use. --- crates/verify/tests/codec.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/crates/verify/tests/codec.rs b/crates/verify/tests/codec.rs index d2efa9de025..bbff4a6afe8 100644 --- a/crates/verify/tests/codec.rs +++ b/crates/verify/tests/codec.rs @@ -215,7 +215,11 @@ fn video_code_survives_compositing_scaling_rounding_bubble_and_blur() { let dir = tempfile::tempdir().unwrap(); // 640x360 is a typical verification output; 320x180 puts the screen at // a quarter of its source size and the camera code cells at ~6 px. - for (out_w, out_h, bitrate) in [(640, 360, 150_000), (320, 180, 80_000)] { + // 320x180 is encoded at the Web export preset (0.08 bpp): at 80 kbps the + // paravirtualised VideoToolbox HEVC encoder on the CI macOS VMs + // occasionally left a blurred camera cell at margin 0.17, below the + // reader's 0.18, while this host's hardware encoder read it at 0.31. + for (out_w, out_h, bitrate) in [(640, 360, 150_000), (320, 180, 138_240)] { for blur in [false, true] { let pairs: Vec<(u32, u32)> = (0..40).map(|i| (i * 3 + 100, i / 2 + 7)).collect(); let frames: Vec = pairs From ea8bf7ccbb0d7adffa2e818c865d1d94047d5d0b Mon Sep 17 00:00:00 2001 From: Aatman Jain Date: Fri, 30 Oct 2026 11:02:49 +0000 Subject: [PATCH 11/11] fix(ci): walk GIF blocks when measuring the exported GIF's length --- .../linux-desktop-harness/scenario-export.sh | 30 +++++++++++++++---- 1 file changed, 25 insertions(+), 5 deletions(-) diff --git a/scripts/linux-desktop-harness/scenario-export.sh b/scripts/linux-desktop-harness/scenario-export.sh index d73efcf77af..ec76b57331e 100755 --- a/scripts/linux-desktop-harness/scenario-export.sh +++ b/scripts/linux-desktop-harness/scenario-export.sh @@ -60,12 +60,32 @@ gif_frames_and_seconds() { import sys data = open(sys.argv[1], "rb").read() + + +def skip_sub_blocks(at): + while data[at]: + at += data[at] + 1 + return at + 1 + + if data[:6] in (b"GIF87a", b"GIF89a"): - frames, centiseconds, at = 0, 0, data.find(b"\x21\xf9\x04") - while at != -1 and at + 6 <= len(data): - frames += 1 - centiseconds += data[at + 4] | data[at + 5] << 8 - at = data.find(b"\x21\xf9\x04", at + 8) + frames, centiseconds, delay = 0, 0, 0 + flags = data[10] + at = 13 + (3 << ((flags & 7) + 1) if flags & 0x80 else 0) + while at < len(data) and data[at] != 0x3B: + if data[at] == 0x21: + if data[at + 1] == 0xF9: + delay = data[at + 4] | data[at + 5] << 8 + at = skip_sub_blocks(at + 2) + elif data[at] == 0x2C: + frames += 1 + centiseconds += delay + delay = 0 + local = data[at + 9] + at += 10 + (3 << ((local & 7) + 1) if local & 0x80 else 0) + at = skip_sub_blocks(at + 1) + else: + break print(frames, f"{centiseconds / 100:.2f}") PY }