diff --git a/crates/camera/src/macos.rs b/crates/camera/src/macos.rs index fece38486f1..467e554e45d 100644 --- a/crates/camera/src/macos.rs +++ b/crates/camera/src/macos.rs @@ -5,9 +5,6 @@ use cidre::*; use objc2_av_foundation::*; pub(super) fn list_cameras_impl() -> impl Iterator { - // ar_pool: called from pool-less tokio threads on a polling cadence; the - // unique_id/localized_name accessors autorelease NSStrings that would - // otherwise accumulate for the process lifetime. objc::ar_pool(|| { let devices = cap_camera_avfoundation::list_video_devices(); devices @@ -24,38 +21,44 @@ pub(super) fn list_cameras_impl() -> impl Iterator { impl CameraInfo { pub(super) fn formats_impl(&self) -> Option> { - let device = find_device(self)?; - - let mut ret = vec![]; - - for format in device.formats().iter() { - let desc = format.format_desc(); - let width = desc.dims().width as u32; - let height = desc.dims().height as u32; - let pixel_format = fourcc_display(desc.media_sub_type()); - - for fr_range in format.video_supported_frame_rate_ranges().iter() { - // SAFETY: trust me bro it crashes on intel mac otherwise - let fr_range = unsafe { - &*(fr_range as *const av::capture::device::FrameRateRange) - .cast::() - }; - - let min = unsafe { fr_range.minFrameDuration() }; - - ret.push(Format { - native: format.retained(), - info: FormatInfo { - width, - height, - frame_rate: min.timescale as f32 / min.value as f32, - }, - pixel_format: Some(pixel_format.clone()), - }) + objc::ar_pool(|| { + let device = find_device(self)?; + let mut ret = vec![]; + + for format in device.formats().iter() { + append_supported_formats(&mut ret, format); } + + Some(ret) + }) + } +} + +fn append_supported_formats(ret: &mut Vec, format: &av::capture::device::Format) { + let desc = format.format_desc(); + let width = desc.dims().width as u32; + let height = desc.dims().height as u32; + let pixel_format = fourcc_display(desc.media_sub_type()); + + for fr_range in format.video_supported_frame_rate_ranges().iter() { + // Cast through AVFrameRateRange to avoid Intel macOS selector dispatch crashes. + let fr_range = unsafe { + &*(fr_range as *const av::capture::device::FrameRateRange).cast::() + }; + let min = unsafe { fr_range.minFrameDuration() }; + if min.value <= 0 { + continue; } - Some(ret) + ret.push(Format { + native: format.retained(), + info: FormatInfo { + width, + height, + frame_rate: min.timescale as f32 / min.value as f32, + }, + pixel_format: Some(pixel_format.clone()), + }); } } @@ -94,17 +97,19 @@ fn avfoundation_session_lifecycle_guard() -> std::sync::MutexGuard<'static, ()> } fn find_device(info: &CameraInfo) -> Option> { - let devices = list_video_devices(); - devices - .iter() - .find( - |d| match (ModelID::from_avfoundation(d).as_ref(), info.model_id()) { - (Some(a), Some(b)) => a == b, - (None, None) => d.unique_id().to_string() == info.device_id(), - _ => false, - }, - ) - .map(|v| v.retained()) + objc::ar_pool(|| { + let devices = list_video_devices(); + devices + .iter() + .find( + |d| match (ModelID::from_avfoundation(d).as_ref(), info.model_id()) { + (Some(a), Some(b)) => a == b, + (None, None) => d.unique_id().to_string() == info.device_id(), + _ => false, + }, + ) + .map(|v| v.retained()) + }) } fn fourcc_display(fourcc: u32) -> String { @@ -112,10 +117,7 @@ fn fourcc_display(fourcc: u32) -> String { cidre::four_cc_to_str(&mut bytes).to_string() } -/// The min frame duration of the format's frame-rate range matching the -/// selected rate. Formats are enumerated per frame-rate range, so without -/// this the device free-runs at the format's default rate (often the -/// fastest range, e.g. 60fps) while the pipeline assumes the selected one. +// Match the selected frame rate's minimum frame duration so the sensor does not free-run at 60fps. fn min_frame_duration_for_rate( format: &av::capture::device::Format, frame_rate: f32, @@ -123,7 +125,6 @@ fn min_frame_duration_for_rate( let mut best: Option<(f32, cm::Time)> = None; for fr_range in format.video_supported_frame_rate_ranges().iter() { - // SAFETY: trust me bro it crashes on intel mac otherwise let fr_range = unsafe { &*(fr_range as *const av::capture::device::FrameRateRange).cast::() }; @@ -143,23 +144,12 @@ fn min_frame_duration_for_rate( best.and_then(|(rate_diff, duration)| (rate_diff < 1.0).then_some(duration)) } -pub(super) fn start_capturing_impl( - camera: &CameraInfo, - format: Format, - mode: CaptureMode, +fn build_sample_delegate( mut callback: impl FnMut(CapturedFrame) + 'static, -) -> Result { - let mut device = find_device(camera) - .ok_or(StartCapturingError::DeviceNotFound)? - .retained(); - - let input = - av::capture::DeviceInput::with_device(&device).map_err(AVFoundationError::Static)?; - - let queue = dispatch::Queue::new(); +) -> arc::R { let mut missing_image_bufs: u64 = 0; - let delegate = - CallbackOutputDelegate::with(CallbackOutputDelegateInner::new(Box::new(move |data| { + CallbackOutputDelegate::with(CallbackOutputDelegateInner::new(Box::new(move |data| { + objc::ar_pool(|| { if data.sample_buf.image_buf().is_none() { missing_image_bufs += 1; if missing_image_bufs == 1 || missing_image_bufs.is_multiple_of(100) { @@ -169,113 +159,141 @@ pub(super) fn start_capturing_impl( ); } return; - }; + } callback(CapturedFrame { native: NativeCapturedFrame(data.sample_buf.retained()), timestamp: data.timestamp, }); - }))); + }); + }))) +} + +fn configure_native_pixel_format(output: &mut av::CaptureVideoDataOutput, format: &Format) { + let pixel_format = format.native().format_desc().media_sub_type(); + let is_available = output + .available_video_cv_pixel_formats() + .iter() + .any(|n| n.as_u32() == pixel_format); + + if !is_available { + tracing::warn!( + pixel_format = %fourcc_display(pixel_format), + "Native camera pixel format not offered by video output, using default conversion" + ); + return; + } - let mut output = av::capture::VideoDataOutput::new(); - let mut session = av::capture::Session::new(); - let mut added_input = false; + let video_settings = ns::Dictionary::with_keys_values( + &[cv::pixel_buffer_keys::pixel_format().as_ns()], + &[ns::Number::with_u32(pixel_format).as_id_ref()], + ); + if let Err(err) = output.set_video_settings(Some(video_settings.as_ref())) { + tracing::warn!( + pixel_format = %fourcc_display(pixel_format), + "Failed to request native camera pixel format, using default conversion: {err}" + ); + } +} - session.configure(|s| { - if s.can_add_input(&input) { - s.add_input(&input); - added_input = true; - } else { - return; - } +struct SessionStartContext<'a> { + device: &'a mut arc::R, + session: &'a mut arc::R, + format: &'a Format, + mode: CaptureMode, +} - s.add_output(&output); - }); +fn start_avfoundation_session(ctx: SessionStartContext<'_>) -> Result<(), StartCapturingError> { + let _session_lifecycle_guard = avfoundation_session_lifecycle_guard(); - if !added_input { - return Err(AVFoundationError::Message( - "Failed to add camera input to AVFoundation session".to_string(), - ) - .into()); + if ctx.mode == CaptureMode::Compatibility { + ctx.session.start_running(); + return Ok(()); } - if mode == CaptureMode::Native { - let pixel_format = format.native().format_desc().media_sub_type(); - let is_available = output - .available_video_cv_pixel_formats() - .iter() - .any(|n| n.as_u32() == pixel_format); + // Hold the device config lock across start_running so macOS does not reset the active format. + let mut lock = ctx + .device + .config_lock() + .map_err(AVFoundationError::Retained)?; + lock.set_active_format(ctx.format.native()); - if is_available { - let video_settings = ns::Dictionary::with_keys_values( - &[cv::pixel_buffer_keys::pixel_format().as_ns()], - &[ns::Number::with_u32(pixel_format).as_id_ref()], - ); - if let Err(err) = output.set_video_settings(Some(video_settings.as_ref())) { - tracing::warn!( - pixel_format = %fourcc_display(pixel_format), - "Failed to request native camera pixel format, using default conversion: {err}" - ); - } - } else { + if let Some(duration) = + min_frame_duration_for_rate(ctx.format.native(), ctx.format.frame_rate()) + { + if let Err(err) = lock.set_active_video_min_frame_duration(duration) { tracing::warn!( - pixel_format = %fourcc_display(pixel_format), - "Native camera pixel format not offered by video output, using default conversion" + frame_rate = ctx.format.frame_rate(), + "Failed to set camera min frame duration: {err}" ); } + } else { + tracing::warn!( + frame_rate = ctx.format.frame_rate(), + "No frame-rate range matches the selected camera rate, using device default" + ); } - output.set_sample_buf_delegate(Some(delegate.as_ref()), Some(&queue)); - - { - let _session_lifecycle_guard = avfoundation_session_lifecycle_guard(); - - match mode { - CaptureMode::Native => { - // The device config must stay locked while running starts, - // otherwise start_running can overwrite the active format on macOS - // https://stackoverflow.com/questions/36689578/avfoundation-capturing-video-with-custom-resolution - let mut _lock = device.config_lock().map_err(AVFoundationError::Retained)?; - - _lock.set_active_format(format.native()); - - // Setting the active format resets the frame durations to the - // format's defaults, which can be a faster rate than the - // selected one (the device would then deliver e.g. 60fps - // while the pipeline records it as 30fps, stretching the - // recording). Cap delivery at the selected rate; leave the - // max duration alone so low-light rate reduction still works. - if let Some(duration) = - min_frame_duration_for_rate(format.native(), format.frame_rate()) - { - if let Err(err) = _lock.set_active_video_min_frame_duration(duration) { - tracing::warn!( - frame_rate = format.frame_rate(), - "Failed to set camera min frame duration: {err}" - ); - } - } else { - tracing::warn!( - frame_rate = format.frame_rate(), - "No frame-rate range matches the selected camera rate, using device default" - ); - } + ctx.session.start_running(); + Ok(()) +} - session.start_running(); - } - CaptureMode::Compatibility => { - session.start_running(); +pub(super) fn start_capturing_impl( + camera: &CameraInfo, + format: Format, + mode: CaptureMode, + callback: impl FnMut(CapturedFrame) + 'static, +) -> Result { + objc::ar_pool(|| { + let mut device = find_device(camera) + .ok_or(StartCapturingError::DeviceNotFound)? + .retained(); + + let input = + av::capture::DeviceInput::with_device(&device).map_err(AVFoundationError::Static)?; + let queue = dispatch::Queue::new(); + let delegate = build_sample_delegate(callback); + + let mut output = av::capture::VideoDataOutput::new(); + let mut session = av::capture::Session::new(); + let mut added_input = false; + + session.configure(|s| { + if s.can_add_input(&input) { + s.add_input(&input); + s.add_output(&output); + added_input = true; } + }); + + if !added_input { + return Err(AVFoundationError::Message( + "Failed to add camera input to AVFoundation session".to_string(), + ) + .into()); } - } - Ok(AVFoundationRecordingHandle { - _delegate: delegate, - session, - output, - input, - _device: device, - is_torn_down: false, + if mode == CaptureMode::Native { + configure_native_pixel_format(&mut output, &format); + } + + output.set_sample_buf_delegate(Some(delegate.as_ref()), Some(&queue)); + + start_avfoundation_session(SessionStartContext { + device: &mut device, + session: &mut session, + format: &format, + mode, + })?; + + Ok(AVFoundationRecordingHandle { + _delegate: delegate, + session, + output, + input, + _device: device, + is_torn_down: false, + }) }) } @@ -300,16 +318,18 @@ impl AVFoundationRecordingHandle { } self.is_torn_down = true; - let _session_lifecycle_guard = avfoundation_session_lifecycle_guard(); - - self.session.stop_running(); - self.output - .set_sample_buf_delegate::(None, None); - let output = self.output.clone(); - let input = self.input.clone(); - self.session.configure(move |s| { - s.remove_output(&output); - s.remove_input(&input); + objc::ar_pool(|| { + let _session_lifecycle_guard = avfoundation_session_lifecycle_guard(); + + self.session.stop_running(); + self.output + .set_sample_buf_delegate::(None, None); + let output = self.output.clone(); + let input = self.input.clone(); + self.session.configure(move |s| { + s.remove_output(&output); + s.remove_input(&input); + }); }); } } diff --git a/crates/frame-converter/src/pool.rs b/crates/frame-converter/src/pool.rs index ee8ac5a499f..fff7b93ade1 100644 --- a/crates/frame-converter/src/pool.rs +++ b/crates/frame-converter/src/pool.rs @@ -1,5 +1,5 @@ use crate::{ConversionConfig, ConvertError, FrameConverter, VideoFramePool, create_converter}; -use ffmpeg::frame; +use ffmpeg::{format::Pixel, frame}; use std::{ sync::{ Arc, @@ -10,6 +10,23 @@ use std::{ }; use tracing::{debug, info, warn}; +const CONVERTER_POOL_BYTE_BUDGET: usize = 64 * 1024 * 1024; + +pub(crate) fn estimate_frame_bytes(format: Pixel, width: u32, height: u32) -> usize { + let pixels = (width as usize).saturating_mul(height as usize); + let (num, den) = match format { + Pixel::NV12 | Pixel::YUV420P | Pixel::YUVJ420P => (3, 2), + Pixel::YUYV422 | Pixel::UYVY422 | Pixel::YUV422P => (2, 1), + Pixel::RGB24 | Pixel::BGR24 => (3, 1), + _ => (4, 1), + }; + pixels + .saturating_mul(num) + .checked_div(den) + .unwrap_or(0) + .max(1) +} + pub struct ConvertedFrame { pub frame: frame::Video, pub sequence: u64, @@ -23,6 +40,7 @@ pub struct InputFrame { pub submit_time: Instant, } +#[derive(Clone, Debug)] pub struct ConverterPoolConfig { pub worker_count: usize, pub input_capacity: usize, @@ -43,6 +61,64 @@ impl Default for ConverterPoolConfig { } } +impl ConverterPoolConfig { + pub fn bounded_for_conversion( + &self, + conversion: &ConversionConfig, + ) -> Result { + if self.worker_count == 0 || self.output_capacity == 0 { + return Err(ConvertError::ConversionFailed( + "Converter pool requires at least one worker and one output slot".to_string(), + )); + } + + let in_bytes = estimate_frame_bytes( + conversion.input_format, + conversion.input_width, + conversion.input_height, + ); + let out_bytes = estimate_frame_bytes( + conversion.output_format, + conversion.output_width, + conversion.output_height, + ); + let worker_frame_bytes = in_bytes.saturating_add(out_bytes); + if worker_frame_bytes.saturating_add(out_bytes) > CONVERTER_POOL_BYTE_BUDGET { + return Err(ConvertError::ConversionFailed( + "Conversion frame sizes exceed converter pool byte budget".to_string(), + )); + } + + let max_workers_with_out = + CONVERTER_POOL_BYTE_BUDGET.saturating_sub(out_bytes) / worker_frame_bytes; + let max_workers = + (CONVERTER_POOL_BYTE_BUDGET / 2 / worker_frame_bytes).clamp(1, max_workers_with_out); + let worker_count = self.worker_count.clamp(1, max_workers); + + let remaining_after_workers = CONVERTER_POOL_BYTE_BUDGET + .saturating_sub(worker_count.saturating_mul(worker_frame_bytes)); + let input_headroom = remaining_after_workers.saturating_sub(out_bytes); + let min_input = usize::from(input_headroom >= in_bytes && self.input_capacity > 0); + let max_input = (input_headroom / 2 / in_bytes).max(min_input); + let input_capacity = self.input_capacity.clamp(min_input, max_input); + + let remaining_for_output = remaining_after_workers + .saturating_sub(input_capacity.saturating_mul(in_bytes)); + let max_out_slots = remaining_for_output / out_bytes; + let max_output = (max_out_slots / 2).max(1); + let output_capacity = self.output_capacity.clamp(1, max_output); + let max_pool = max_out_slots.saturating_sub(output_capacity); + + Ok(Self { + worker_count, + input_capacity, + output_capacity, + drop_strategy: self.drop_strategy, + frame_pool_capacity: self.frame_pool_capacity.min(max_pool), + }) + } +} + #[derive(Clone, Copy, Debug)] pub enum DropStrategy { DropOldest, @@ -58,38 +134,38 @@ pub struct ConverterPoolStats { pub struct AsyncConverterPool { input_tx: Option>, + input_rx: flume::Receiver, output_rx: flume::Receiver, workers: Vec>, shutdown: Arc, stats: Arc, + drop_strategy: DropStrategy, frame_pool: Option>, } +#[derive(Default)] struct PoolStats { frames_received: AtomicU64, frames_converted: AtomicU64, frames_dropped: AtomicU64, } -impl Default for PoolStats { - fn default() -> Self { - Self { - frames_received: AtomicU64::new(0), - frames_converted: AtomicU64::new(0), - frames_dropped: AtomicU64::new(0), - } - } +struct PoolBuildSpec { + factory: F, + converter_name: String, + config: ConverterPoolConfig, + frame_pool: Option>, } impl AsyncConverterPool { pub fn new(converter: Arc, config: ConverterPoolConfig) -> Self { let converter_name = converter.name().to_string(); - Self::new_with_factory( - move |_| Ok(Arc::clone(&converter)), - &converter_name, + Self::new_with_spec(PoolBuildSpec { + factory: move |_| Ok(Arc::clone(&converter)), + converter_name, config, - None, - ) + frame_pool: None, + }) .expect("Factory using pre-created converter should not fail") } @@ -97,12 +173,13 @@ impl AsyncConverterPool { conversion_config: ConversionConfig, pool_config: ConverterPoolConfig, ) -> Result { + let bounded_config = pool_config.bounded_for_conversion(&conversion_config)?; let first_converter = create_converter(conversion_config.clone())?; let converter_name = first_converter.name().to_string(); - let frame_pool = if pool_config.frame_pool_capacity > 0 { + let frame_pool = if bounded_config.frame_pool_capacity > 0 { Some(VideoFramePool::new( - pool_config.frame_pool_capacity, + bounded_config.frame_pool_capacity, conversion_config.output_format, conversion_config.output_width, conversion_config.output_height, @@ -111,115 +188,108 @@ impl AsyncConverterPool { None }; - Self::new_with_factory( - move |worker_id| { + Self::new_with_spec(PoolBuildSpec { + factory: move |worker_id| { if worker_id == 0 { Ok(Arc::clone(&first_converter)) } else { create_converter(conversion_config.clone()) } }, - &converter_name, - pool_config, + converter_name, + config: bounded_config, frame_pool, - ) + }) } - fn new_with_factory( - factory: F, - converter_name: &str, - config: ConverterPoolConfig, - frame_pool: Option>, - ) -> Result + fn new_with_spec(spec: PoolBuildSpec) -> Result where F: Fn(usize) -> Result, ConvertError> + Send + Sync + 'static, { - let (input_tx, input_rx) = flume::bounded(config.input_capacity); - let (output_tx, output_rx) = flume::bounded(config.output_capacity); + let (input_tx, input_rx) = flume::bounded(spec.config.input_capacity); + let (output_tx, output_rx) = flume::bounded(spec.config.output_capacity); let shutdown = Arc::new(AtomicBool::new(false)); let stats = Arc::new(PoolStats::default()); - let mut workers = Vec::with_capacity(config.worker_count); - let factory = Arc::new(factory); + let mut workers = Vec::with_capacity(spec.config.worker_count); + let factory = Arc::new(spec.factory); - for worker_id in 0..config.worker_count { + for worker_id in 0..spec.config.worker_count { let factory = Arc::clone(&factory); - let input_rx = input_rx.clone(); - let output_tx = output_tx.clone(); - let shutdown = Arc::clone(&shutdown); - let stats = Arc::clone(&stats); - let drop_strategy = config.drop_strategy; - let frame_pool = frame_pool.clone(); + let worker_ctx = WorkerContext { + worker_id, + input_rx: input_rx.clone(), + output_tx: output_tx.clone(), + shutdown: Arc::clone(&shutdown), + stats: Arc::clone(&stats), + drop_strategy: spec.config.drop_strategy, + frame_pool: spec.frame_pool.clone(), + }; let handle = thread::Builder::new() .name(format!("converter-worker-{worker_id}")) - .spawn(move || { - let converter = match factory(worker_id) { - Ok(c) => c, - Err(e) => { - warn!("Worker {} failed to create converter: {}", worker_id, e); - return; - } - }; - worker_loop(WorkerContext { - worker_id, - converter, - input_rx, - output_tx, - shutdown, - stats, - drop_strategy, - frame_pool, - }); + .spawn(move || match factory(worker_id) { + Ok(converter) => worker_loop(worker_ctx, converter), + Err(e) => warn!("Worker {worker_id} failed to create converter: {e}"), }) .expect("Failed to spawn converter worker thread"); workers.push(handle); } + let pooling_suffix = if spec.frame_pool.is_some() { + " with frame pooling" + } else { + "" + }; info!( "AsyncConverterPool started with {} workers using {} converter{}", - config.worker_count, - converter_name, - if frame_pool.is_some() { - " with frame pooling" - } else { - "" - } + spec.config.worker_count, spec.converter_name, pooling_suffix ); Ok(Self { input_tx: Some(input_tx), + input_rx, output_rx, workers, shutdown, stats, - frame_pool: frame_pool.clone(), + drop_strategy: spec.config.drop_strategy, + frame_pool: spec.frame_pool, }) } + fn record_input_drop(&self) { + let dropped = self.stats.frames_dropped.fetch_add(1, Ordering::Relaxed) + 1; + if dropped.is_multiple_of(30) { + warn!("Converter pool input full, dropped {dropped} frames so far"); + } + } + pub fn submit(&self, frame: frame::Video, sequence: u64) -> Result<(), ConvertError> { let Some(input_tx) = &self.input_tx else { return Err(ConvertError::PoolShutdown); }; self.stats.frames_received.fetch_add(1, Ordering::Relaxed); - let submit_time = Instant::now(); - - match input_tx.try_send(InputFrame { + let candidate = InputFrame { frame, sequence, - submit_time, - }) { + submit_time: Instant::now(), + }; + + match input_tx.try_send(candidate) { Ok(()) => Ok(()), - Err(flume::TrySendError::Full(_)) => { - self.stats.frames_dropped.fetch_add(1, Ordering::Relaxed); - let dropped = self.stats.frames_dropped.load(Ordering::Relaxed); - if dropped.is_multiple_of(30) { - warn!( - "Converter pool input full, dropped {} frames so far", - dropped - ); + Err(flume::TrySendError::Full(rejected)) => { + if matches!(self.drop_strategy, DropStrategy::DropOldest) { + if self.input_rx.try_recv().is_ok() { + self.record_input_drop(); + } + if input_tx.try_send(rejected).is_err() { + self.record_input_drop(); + } + } else { + self.record_input_drop(); } Ok(()) } @@ -264,6 +334,13 @@ impl AsyncConverterPool { self.frame_pool.as_ref() } + fn pending_frame_count(&self) -> u64 { + let stats = self.stats(); + stats + .frames_received + .saturating_sub(stats.frames_converted + stats.frames_dropped) + } + pub fn drain_with_timeout( &mut self, mut frame_handler: impl FnMut(ConvertedFrame), @@ -271,21 +348,17 @@ impl AsyncConverterPool { ) -> usize { self.input_tx.take(); - let deadline = std::time::Instant::now() + timeout; + let deadline = Instant::now() + timeout; let mut drained = 0; - while std::time::Instant::now() < deadline { + while Instant::now() < deadline { match self.output_rx.recv_timeout(Duration::from_millis(50)) { Ok(frame) => { frame_handler(frame); drained += 1; } Err(flume::RecvTimeoutError::Timeout) => { - let stats = self.stats(); - let pending = stats - .frames_received - .saturating_sub(stats.frames_converted + stats.frames_dropped); - if pending == 0 { + if self.pending_frame_count() == 0 { break; } } @@ -330,7 +403,6 @@ impl AsyncConverterPool { struct WorkerContext { worker_id: usize, - converter: Arc, input_rx: flume::Receiver, output_tx: flume::Sender, shutdown: Arc, @@ -339,93 +411,107 @@ struct WorkerContext { frame_pool: Option>, } -fn worker_loop(ctx: WorkerContext) { - debug!( - "Converter worker {} started{}", - ctx.worker_id, - if ctx.frame_pool.is_some() { - " with frame pooling" - } else { - "" +impl WorkerContext { + fn convert_input( + &self, + converter: &dyn FrameConverter, + input_frame: frame::Video, + ) -> Result { + let Some(ref pool) = self.frame_pool else { + return converter.convert(input_frame); + }; + + let mut output = pool.get(); + match converter.convert_into(input_frame, &mut output) { + Ok(()) => Ok(output), + Err(e) => { + pool.put(output); + Err(e) + } } - ); - let mut local_converted = 0u64; - let mut local_errors = 0u64; + } - while !ctx.shutdown.load(Ordering::Relaxed) { - match ctx.input_rx.recv_timeout(Duration::from_millis(100)) { - Ok(input) => { - let sequence = input.sequence; - let submit_time = input.submit_time; - let convert_start = Instant::now(); - - let result = if let Some(ref pool) = ctx.frame_pool { - let mut output = pool.get(); - match ctx.converter.convert_into(input.frame, &mut output) { - Ok(()) => Ok(output), - Err(e) => { - pool.put(output); - Err(e) - } - } - } else { - ctx.converter.convert(input.frame) - }; - - match result { - Ok(converted) => { - let conversion_duration = convert_start.elapsed(); - local_converted += 1; - ctx.stats.frames_converted.fetch_add(1, Ordering::Relaxed); - - match ctx.output_tx.try_send(ConvertedFrame { - frame: converted, - sequence, - submit_time, - conversion_duration, - }) { - Ok(()) => {} - Err(flume::TrySendError::Full(dropped_frame)) => { - match ctx.drop_strategy { - DropStrategy::DropOldest | DropStrategy::DropNewest => { - ctx.stats.frames_dropped.fetch_add(1, Ordering::Relaxed); - } - } - if let Some(ref pool) = ctx.frame_pool { - pool.put(dropped_frame.frame); - } - } - Err(flume::TrySendError::Disconnected(dropped_frame)) => { - if let Some(ref pool) = ctx.frame_pool { - pool.put(dropped_frame.frame); - } - break; - } - } - } - Err(e) => { - local_errors += 1; - if local_errors % 10 == 1 { - warn!( - "Worker {}: conversion error (#{} total): {}", - ctx.worker_id, local_errors, e - ); - } + fn recycle_frame(&self, frame: frame::Video) { + if let Some(ref pool) = self.frame_pool { + pool.put(frame); + } + } + + fn deliver_converted(&self, converted_frame: ConvertedFrame) -> bool { + match self.output_tx.try_send(converted_frame) { + Ok(()) => true, + Err(flume::TrySendError::Full(dropped_frame)) => { + match self.drop_strategy { + DropStrategy::DropOldest | DropStrategy::DropNewest => { + self.stats.frames_dropped.fetch_add(1, Ordering::Relaxed); } } + self.recycle_frame(dropped_frame.frame); + true + } + Err(flume::TrySendError::Disconnected(dropped_frame)) => { + self.recycle_frame(dropped_frame.frame); + false + } + } + } + + fn process_one( + &self, + converter: &dyn FrameConverter, + input: InputFrame, + counters: &mut (u64, u64), + ) -> bool { + let convert_start = Instant::now(); + match self.convert_input(converter, input.frame) { + Ok(converted) => { + counters.0 += 1; + self.stats.frames_converted.fetch_add(1, Ordering::Relaxed); + self.deliver_converted(ConvertedFrame { + frame: converted, + sequence: input.sequence, + submit_time: input.submit_time, + conversion_duration: convert_start.elapsed(), + }) } - Err(flume::RecvTimeoutError::Timeout) => { - continue; + Err(e) => { + counters.1 += 1; + if counters.1 % 10 == 1 { + warn!( + "Worker {}: conversion error (#{} total): {}", + self.worker_id, counters.1, e + ); + } + true } - Err(flume::RecvTimeoutError::Disconnected) => { - break; + } + } +} + +fn worker_loop(ctx: WorkerContext, converter: Arc) { + let pooling_label = if ctx.frame_pool.is_some() { + " with frame pooling" + } else { + "" + }; + debug!("Converter worker {} started{}", ctx.worker_id, pooling_label); + let mut counters = (0u64, 0u64); + + while !ctx.shutdown.load(Ordering::Relaxed) { + match ctx.input_rx.recv_timeout(Duration::from_millis(100)) { + Ok(input) => { + if !ctx.process_one(converter.as_ref(), input, &mut counters) { + break; + } } + Err(flume::RecvTimeoutError::Timeout) => continue, + Err(flume::RecvTimeoutError::Disconnected) => break, } } debug!( "Converter worker {} finished: {} converted, {} errors", - ctx.worker_id, local_converted, local_errors + ctx.worker_id, counters.0, counters.1 ); } @@ -434,3 +520,43 @@ impl Drop for AsyncConverterPool { self.do_shutdown(); } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn bounds_high_res_converter_pool_queues_within_memory_budget() { + let conversion = ConversionConfig::new(Pixel::BGRA, 3840, 2160, Pixel::NV12, 3840, 2160); + let requested = ConverterPoolConfig { + worker_count: 4, + input_capacity: 120, + output_capacity: 90, + drop_strategy: DropStrategy::DropOldest, + frame_pool_capacity: 16, + }; + let bounded = requested + .bounded_for_conversion(&conversion) + .expect("4K BGRA -> NV12 should fit within converter pool budget"); + let in_bytes = estimate_frame_bytes(Pixel::BGRA, 3840, 2160); + let out_bytes = estimate_frame_bytes(Pixel::NV12, 3840, 2160); + let retained_bytes = (bounded.input_capacity + bounded.worker_count) * in_bytes + + (bounded.output_capacity + bounded.frame_pool_capacity + bounded.worker_count) + * out_bytes; + assert!(bounded.worker_count >= 1); + assert!(bounded.output_capacity >= 1); + assert!(retained_bytes <= CONVERTER_POOL_BYTE_BUDGET); + } + + #[test] + fn rejects_conversions_exceeding_converter_pool_byte_budget() { + let requested = ConverterPoolConfig::default(); + let oversized_worker = + ConversionConfig::new(Pixel::BGRA, 4096, 4096, Pixel::NV12, 4096, 4096); + assert!(requested.bounded_for_conversion(&oversized_worker).is_err()); + + let oversized_output = + ConversionConfig::new(Pixel::RGB24, 3840, 2160, Pixel::BGRA, 3840, 2160); + assert!(requested.bounded_for_conversion(&oversized_output).is_err()); + } +} diff --git a/crates/frame-converter/src/videotoolbox.rs b/crates/frame-converter/src/videotoolbox.rs index 8d4d8e28d87..7a95e4b9abe 100644 --- a/crates/frame-converter/src/videotoolbox.rs +++ b/crates/frame-converter/src/videotoolbox.rs @@ -95,8 +95,60 @@ struct SessionHandle(VTPixelTransferSessionRef); unsafe impl Send for SessionHandle {} +struct OwnedPixelBuffer(CVPixelBufferRef); + +unsafe impl Send for OwnedPixelBuffer {} + +impl OwnedPixelBuffer { + fn as_ptr(&self) -> CVPixelBufferRef { + self.0 + } +} + +impl Drop for OwnedPixelBuffer { + fn drop(&mut self) { + if !self.0.is_null() { + unsafe { + CVPixelBufferRelease(self.0); + } + } + } +} + +#[derive(Clone, Copy)] +struct PlaneCopyRegion { + src_ptr: *const u8, + src_stride: usize, + dst_ptr: *mut u8, + dst_stride: usize, + height: usize, +} + +unsafe fn copy_strided_plane(region: PlaneCopyRegion) { + if region.src_stride == region.dst_stride { + unsafe { + ptr::copy_nonoverlapping( + region.src_ptr, + region.dst_ptr, + region.height * region.src_stride, + ); + } + return; + } + + let copy_len = region.src_stride.min(region.dst_stride); + for row in 0..region.height { + unsafe { + let src_row = region.src_ptr.add(row * region.src_stride); + let dst_row = region.dst_ptr.add(row * region.dst_stride); + ptr::copy_nonoverlapping(src_row, dst_row, copy_len); + } + } +} + pub struct VideoToolboxConverter { session: Mutex, + reusable_output_buffer: Mutex>, input_format: Pixel, input_cv_format: u32, output_format: Pixel, @@ -146,6 +198,7 @@ impl VideoToolboxConverter { Ok(Self { session: Mutex::new(SessionHandle(session)), + reusable_output_buffer: Mutex::new(None), input_format: config.input_format, input_cv_format, output_format: config.output_format, @@ -162,7 +215,7 @@ impl VideoToolboxConverter { fn create_input_pixel_buffer( &self, input: &frame::Video, - ) -> Result { + ) -> Result { let mut pixel_buffer: CVPixelBufferRef = ptr::null_mut(); let base_address = input.data(0).as_ptr() as *mut c_void; @@ -189,10 +242,14 @@ impl VideoToolboxConverter { ))); } - Ok(pixel_buffer) + Ok(OwnedPixelBuffer(pixel_buffer)) } - fn create_output_pixel_buffer(&self) -> Result { + fn take_or_create_output_pixel_buffer(&self) -> Result { + if let Some(cached) = self.reusable_output_buffer.lock().take() { + return Ok(cached); + } + let mut pixel_buffer: CVPixelBufferRef = ptr::null_mut(); let status = unsafe { @@ -212,7 +269,14 @@ impl VideoToolboxConverter { ))); } - Ok(pixel_buffer) + Ok(OwnedPixelBuffer(pixel_buffer)) + } + + fn recycle_output_pixel_buffer(&self, buffer: OwnedPixelBuffer) { + let mut slot = self.reusable_output_buffer.lock(); + if slot.is_none() { + *slot = Some(buffer); + } } fn copy_output_to_frame( @@ -220,57 +284,35 @@ impl VideoToolboxConverter { pixel_buffer: CVPixelBufferRef, output: &mut frame::Video, ) -> Result<(), ConvertError> { - unsafe { - let lock_status = CVPixelBufferLockBaseAddress(pixel_buffer, 0); - if lock_status != K_CV_RETURN_SUCCESS { - return Err(ConvertError::ConversionFailed(format!( - "CVPixelBufferLockBaseAddress failed: {lock_status}" - ))); - } + let lock_status = unsafe { CVPixelBufferLockBaseAddress(pixel_buffer, 0) }; + if lock_status != K_CV_RETURN_SUCCESS { + return Err(ConvertError::ConversionFailed(format!( + "CVPixelBufferLockBaseAddress failed: {lock_status}" + ))); } unsafe { let plane_count = CVPixelBufferGetPlaneCount(pixel_buffer); if plane_count == 0 { - let src_ptr = CVPixelBufferGetBaseAddress(pixel_buffer); - let src_stride = CVPixelBufferGetBytesPerRow(pixel_buffer); - let height = CVPixelBufferGetHeight(pixel_buffer); - let dst_stride = output.stride(0); - - let dst_data = output.data_mut(0); - let dst_ptr = dst_data.as_mut_ptr(); - - if src_stride == dst_stride { - ptr::copy_nonoverlapping(src_ptr, dst_ptr, height * src_stride); - } else { - let copy_len = src_stride.min(dst_stride); - for row in 0..height { - let src_row = src_ptr.add(row * src_stride); - let dst_row = dst_ptr.add(row * dst_stride); - ptr::copy_nonoverlapping(src_row, dst_row, copy_len); - } - } + let region = PlaneCopyRegion { + src_ptr: CVPixelBufferGetBaseAddress(pixel_buffer), + src_stride: CVPixelBufferGetBytesPerRow(pixel_buffer), + dst_ptr: output.data_mut(0).as_mut_ptr(), + dst_stride: output.stride(0), + height: CVPixelBufferGetHeight(pixel_buffer), + }; + copy_strided_plane(region); } else { for plane in 0..plane_count { - let src_ptr = CVPixelBufferGetBaseAddressOfPlane(pixel_buffer, plane); - let src_stride = CVPixelBufferGetBytesPerRowOfPlane(pixel_buffer, plane); - let height = CVPixelBufferGetHeightOfPlane(pixel_buffer, plane); - let dst_stride = output.stride(plane); - - let dst_data = output.data_mut(plane); - let dst_ptr = dst_data.as_mut_ptr(); - - if src_stride == dst_stride { - ptr::copy_nonoverlapping(src_ptr, dst_ptr, height * src_stride); - } else { - let copy_len = src_stride.min(dst_stride); - for row in 0..height { - let src_row = src_ptr.add(row * src_stride); - let dst_row = dst_ptr.add(row * dst_stride); - ptr::copy_nonoverlapping(src_row, dst_row, copy_len); - } - } + let region = PlaneCopyRegion { + src_ptr: CVPixelBufferGetBaseAddressOfPlane(pixel_buffer, plane), + src_stride: CVPixelBufferGetBytesPerRowOfPlane(pixel_buffer, plane), + dst_ptr: output.data_mut(plane).as_mut_ptr(), + dst_stride: output.stride(plane), + height: CVPixelBufferGetHeightOfPlane(pixel_buffer, plane), + }; + copy_strided_plane(region); } } @@ -317,23 +359,22 @@ impl FrameConverter for VideoToolboxConverter { } let input_buffer = self.create_input_pixel_buffer(&input)?; - let output_buffer = self.create_output_pixel_buffer()?; + let output_buffer = self.take_or_create_output_pixel_buffer()?; let status = { let session_guard = self.session.lock(); unsafe { - VTPixelTransferSessionTransferImage(session_guard.0, input_buffer, output_buffer) + VTPixelTransferSessionTransferImage( + session_guard.0, + input_buffer.as_ptr(), + output_buffer.as_ptr(), + ) } }; - unsafe { - CVPixelBufferRelease(input_buffer); - } + drop(input_buffer); if status != 0 { - unsafe { - CVPixelBufferRelease(output_buffer); - } return Err(ConvertError::ConversionFailed(format!( "VTPixelTransferSessionTransferImage failed: {status}" ))); @@ -345,12 +386,9 @@ impl FrameConverter for VideoToolboxConverter { ); } - self.copy_output_to_frame(output_buffer, output)?; + self.copy_output_to_frame(output_buffer.as_ptr(), output)?; output.set_pts(input.pts()); - - unsafe { - CVPixelBufferRelease(output_buffer); - } + self.recycle_output_pixel_buffer(output_buffer); Ok(()) } diff --git a/crates/recording/src/benchmark.rs b/crates/recording/src/benchmark.rs index 6da4a285eba..91deb0ac58d 100644 --- a/crates/recording/src/benchmark.rs +++ b/crates/recording/src/benchmark.rs @@ -6,6 +6,17 @@ use std::{ time::{Duration, Instant}, }; +const MAX_TIMING_SAMPLES: usize = 4_096; + +fn push_bounded_sample(samples: &RwLock>, value_ns: u64) { + let mut guard = samples.write().unwrap(); + if guard.len() >= MAX_TIMING_SAMPLES { + let drain_count = MAX_TIMING_SAMPLES / 4; + guard.drain(..drain_count); + } + guard.push(value_ns); +} + #[derive(Debug, Clone)] pub struct FrameTiming { pub receive_time: Instant, @@ -82,22 +93,16 @@ impl PipelineMetrics { pub fn record_frame_converted(&self, duration: Duration) { self.frames_converted.fetch_add(1, Ordering::Relaxed); - self.conversion_times_ns - .write() - .unwrap() - .push(duration.as_nanos() as u64); + push_bounded_sample(&self.conversion_times_ns, duration.as_nanos() as u64); } pub fn record_frame_encoded(&self, encode_duration: Duration, pipeline_latency: Duration) { self.frames_encoded.fetch_add(1, Ordering::Relaxed); - self.encode_times_ns - .write() - .unwrap() - .push(encode_duration.as_nanos() as u64); - self.pipeline_latencies_ns - .write() - .unwrap() - .push(pipeline_latency.as_nanos() as u64); + push_bounded_sample(&self.encode_times_ns, encode_duration.as_nanos() as u64); + push_bounded_sample( + &self.pipeline_latencies_ns, + pipeline_latency.as_nanos() as u64, + ); } pub fn record_dropped_input(&self) { @@ -155,6 +160,44 @@ pub struct MetricsSnapshot { pub duration: Option, } +fn print_timing_section(title: &str, samples_ns: &[u64]) { + println!("\n--- {title} ---"); + if let Some(avg) = average_duration(samples_ns) { + println!(" Average: {avg:?}"); + } + for (label, pct) in [("P50: ", 50.0), ("P95: ", 95.0), ("P99: ", 99.0)] { + if let Some(val) = percentile_duration(samples_ns, pct) { + println!(" {label} {val:?}"); + } + } + if let Some(max) = max_duration(samples_ns) { + println!(" Max: {max:?}"); + } +} + +fn average_duration(samples_ns: &[u64]) -> Option { + if samples_ns.is_empty() { + return None; + } + let sum: u64 = samples_ns.iter().sum(); + Some(Duration::from_nanos(sum / samples_ns.len() as u64)) +} + +fn max_duration(samples_ns: &[u64]) -> Option { + samples_ns.iter().max().map(|&ns| Duration::from_nanos(ns)) +} + +fn timing_json_object(samples_ns: &[u64]) -> serde_json::Value { + let to_ms = |d: Option| d.map(|v| v.as_secs_f64() * 1000.0); + serde_json::json!({ + "avg": to_ms(average_duration(samples_ns)), + "p50": to_ms(percentile_duration(samples_ns, 50.0)), + "p95": to_ms(percentile_duration(samples_ns, 95.0)), + "p99": to_ms(percentile_duration(samples_ns, 99.0)), + "max": to_ms(max_duration(samples_ns)), + }) +} + impl MetricsSnapshot { pub fn total_frames_dropped(&self) -> u64 { self.frames_dropped_input + self.frames_dropped_output + self.frames_dropped_conversion @@ -176,36 +219,15 @@ impl MetricsSnapshot { } pub fn avg_conversion_time(&self) -> Option { - if self.conversion_times_ns.is_empty() { - None - } else { - let sum: u64 = self.conversion_times_ns.iter().sum(); - Some(Duration::from_nanos( - sum / self.conversion_times_ns.len() as u64, - )) - } + average_duration(&self.conversion_times_ns) } pub fn avg_encode_time(&self) -> Option { - if self.encode_times_ns.is_empty() { - None - } else { - let sum: u64 = self.encode_times_ns.iter().sum(); - Some(Duration::from_nanos( - sum / self.encode_times_ns.len() as u64, - )) - } + average_duration(&self.encode_times_ns) } pub fn avg_pipeline_latency(&self) -> Option { - if self.pipeline_latencies_ns.is_empty() { - None - } else { - let sum: u64 = self.pipeline_latencies_ns.iter().sum(); - Some(Duration::from_nanos( - sum / self.pipeline_latencies_ns.len() as u64, - )) - } + average_duration(&self.pipeline_latencies_ns) } pub fn percentile_encode_time(&self, p: f64) -> Option { @@ -221,24 +243,29 @@ impl MetricsSnapshot { } pub fn max_encode_time(&self) -> Option { - self.encode_times_ns - .iter() - .max() - .map(|&ns| Duration::from_nanos(ns)) + max_duration(&self.encode_times_ns) } pub fn max_conversion_time(&self) -> Option { - self.conversion_times_ns - .iter() - .max() - .map(|&ns| Duration::from_nanos(ns)) + max_duration(&self.conversion_times_ns) } pub fn max_pipeline_latency(&self) -> Option { - self.pipeline_latencies_ns - .iter() - .max() - .map(|&ns| Duration::from_nanos(ns)) + max_duration(&self.pipeline_latencies_ns) + } + + fn print_assessment(&self) { + println!("\n--- Performance Assessment ---"); + let can_sustain_30fps = self.effective_fps() >= 29.5; + let low_drop_rate = self.drop_rate() < 1.0; + let acceptable_latency = self + .percentile_pipeline_latency(95.0) + .is_none_or(|l| l < Duration::from_millis(100)); + + let verdict = |ok: bool| if ok { "YES ✓" } else { "NO ✗" }; + println!(" Can sustain 30 FPS: {}", verdict(can_sustain_30fps)); + println!(" Low drop rate (<1%): {}", verdict(low_drop_rate)); + println!(" Acceptable latency: {}", verdict(acceptable_latency)); } pub fn print_report(&self) { @@ -261,85 +288,10 @@ impl MetricsSnapshot { println!(" Total Dropped: {}", self.total_frames_dropped()); println!(" Drop Rate: {:.2}%", self.drop_rate()); - println!("\n--- Conversion Timing ---"); - if let Some(avg) = self.avg_conversion_time() { - println!(" Average: {avg:?}"); - } - if let Some(p50) = self.percentile_conversion_time(50.0) { - println!(" P50: {p50:?}"); - } - if let Some(p95) = self.percentile_conversion_time(95.0) { - println!(" P95: {p95:?}"); - } - if let Some(p99) = self.percentile_conversion_time(99.0) { - println!(" P99: {p99:?}"); - } - if let Some(max) = self.max_conversion_time() { - println!(" Max: {max:?}"); - } - - println!("\n--- Encoding Timing ---"); - if let Some(avg) = self.avg_encode_time() { - println!(" Average: {avg:?}"); - } - if let Some(p50) = self.percentile_encode_time(50.0) { - println!(" P50: {p50:?}"); - } - if let Some(p95) = self.percentile_encode_time(95.0) { - println!(" P95: {p95:?}"); - } - if let Some(p99) = self.percentile_encode_time(99.0) { - println!(" P99: {p99:?}"); - } - if let Some(max) = self.max_encode_time() { - println!(" Max: {max:?}"); - } - - println!("\n--- Total Pipeline Latency ---"); - if let Some(avg) = self.avg_pipeline_latency() { - println!(" Average: {avg:?}"); - } - if let Some(p50) = self.percentile_pipeline_latency(50.0) { - println!(" P50: {p50:?}"); - } - if let Some(p95) = self.percentile_pipeline_latency(95.0) { - println!(" P95: {p95:?}"); - } - if let Some(p99) = self.percentile_pipeline_latency(99.0) { - println!(" P99: {p99:?}"); - } - if let Some(max) = self.max_pipeline_latency() { - println!(" Max: {max:?}"); - } - - println!("\n--- Performance Assessment ---"); - let can_sustain_30fps = self.effective_fps() >= 29.5; - let low_drop_rate = self.drop_rate() < 1.0; - let acceptable_latency = self - .percentile_pipeline_latency(95.0) - .map(|l| l < Duration::from_millis(100)) - .unwrap_or(true); - - println!( - " Can sustain 30 FPS: {}", - if can_sustain_30fps { - "YES ✓" - } else { - "NO ✗" - } - ); - println!( - " Low drop rate (<1%): {}", - if low_drop_rate { "YES ✓" } else { "NO ✗" } - ); - println!( - " Acceptable latency: {}", - if acceptable_latency { - "YES ✓" - } else { - "NO ✗" - } - ); + print_timing_section("Conversion Timing", &self.conversion_times_ns); + print_timing_section("Encoding Timing", &self.encode_times_ns); + print_timing_section("Total Pipeline Latency", &self.pipeline_latencies_ns); + self.print_assessment(); println!("\n=== End Report ===\n"); } @@ -358,27 +310,9 @@ impl MetricsSnapshot { "drop_rate_percent": self.drop_rate(), "effective_fps": self.effective_fps(), }, - "conversion_ms": { - "avg": self.avg_conversion_time().map(|d| d.as_secs_f64() * 1000.0), - "p50": self.percentile_conversion_time(50.0).map(|d| d.as_secs_f64() * 1000.0), - "p95": self.percentile_conversion_time(95.0).map(|d| d.as_secs_f64() * 1000.0), - "p99": self.percentile_conversion_time(99.0).map(|d| d.as_secs_f64() * 1000.0), - "max": self.max_conversion_time().map(|d| d.as_secs_f64() * 1000.0), - }, - "encoding_ms": { - "avg": self.avg_encode_time().map(|d| d.as_secs_f64() * 1000.0), - "p50": self.percentile_encode_time(50.0).map(|d| d.as_secs_f64() * 1000.0), - "p95": self.percentile_encode_time(95.0).map(|d| d.as_secs_f64() * 1000.0), - "p99": self.percentile_encode_time(99.0).map(|d| d.as_secs_f64() * 1000.0), - "max": self.max_encode_time().map(|d| d.as_secs_f64() * 1000.0), - }, - "pipeline_latency_ms": { - "avg": self.avg_pipeline_latency().map(|d| d.as_secs_f64() * 1000.0), - "p50": self.percentile_pipeline_latency(50.0).map(|d| d.as_secs_f64() * 1000.0), - "p95": self.percentile_pipeline_latency(95.0).map(|d| d.as_secs_f64() * 1000.0), - "p99": self.percentile_pipeline_latency(99.0).map(|d| d.as_secs_f64() * 1000.0), - "max": self.max_pipeline_latency().map(|d| d.as_secs_f64() * 1000.0), - }, + "conversion_ms": timing_json_object(&self.conversion_times_ns), + "encoding_ms": timing_json_object(&self.encode_times_ns), + "pipeline_latency_ms": timing_json_object(&self.pipeline_latencies_ns), }) .to_string() } diff --git a/crates/recording/src/memory_profiling.rs b/crates/recording/src/memory_profiling.rs index 430c1a74aa8..801dba3c4d7 100644 --- a/crates/recording/src/memory_profiling.rs +++ b/crates/recording/src/memory_profiling.rs @@ -294,6 +294,78 @@ impl CpuTracker { Some(curr.cpu_percent_since(prev)) } + fn compute_interval_percents(&self) -> Vec { + self.samples + .windows(2) + .map(|pair| pair[1].cpu_percent_since(&pair[0])) + .collect() + } + + fn print_timeline_summary(&self, cpu_percents: &[f64]) { + if cpu_percents.is_empty() { + return; + } + let max_cpu = cpu_percents.iter().copied().fold(0.0_f64, f64::max); + let min_cpu = cpu_percents.iter().copied().fold(f64::MAX, f64::min); + + let mut sorted = cpu_percents.to_vec(); + sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal)); + let p50 = sorted[sorted.len() / 2]; + let p95_idx = ((sorted.len() as f64) * 0.95) as usize; + let p95 = sorted[p95_idx.min(sorted.len() - 1)]; + + println!("\nCPU % timeline (per sample interval):"); + println!(" Min: {min_cpu:.1}%"); + println!(" P50: {p50:.1}%"); + println!(" P95: {p95:.1}%"); + println!(" Max: {max_cpu:.1}%"); + } + + fn print_thread_summary(&self) { + let max_threads = self + .samples + .iter() + .map(|s| s.thread_count) + .max() + .unwrap_or(0); + let min_threads = self + .samples + .iter() + .map(|s| s.thread_count) + .min() + .unwrap_or(0); + let last_threads = self.samples.last().map(|s| s.thread_count).unwrap_or(0); + + println!("\nThreads:"); + println!(" Current: {last_threads}"); + println!(" Min: {min_threads}"); + println!(" Max: {max_threads}"); + + if max_threads > min_threads + 10 { + println!(" *** Thread count variance is high — possible thread leak ***"); + } + } + + fn print_cpu_assessment(&self, avg_per_core_pct: f64, cpu_percents: &[f64]) { + println!("\n=== CPU Assessment ==="); + let max_cpu_per_core = cpu_percents.iter().copied().fold(0.0_f64, f64::max) + / self.num_cores as f64; + + if avg_per_core_pct < 10.0 { + println!(" CPU usage: LOW ({avg_per_core_pct:.1}% per core avg)"); + } else if avg_per_core_pct < 30.0 { + println!(" CPU usage: MODERATE ({avg_per_core_pct:.1}% per core avg)"); + } else { + println!( + " CPU usage: HIGH ({avg_per_core_pct:.1}% per core avg) — investigate hot paths" + ); + } + + if max_cpu_per_core > 80.0 { + println!(" CPU spikes: Peak {max_cpu_per_core:.0}% per core — may cause frame drops"); + } + } + pub fn print_report(&self) { if self.samples.len() < 2 { println!("Not enough CPU samples to analyze"); @@ -323,62 +395,10 @@ impl CpuTracker { println!("CPU time (total): {total_cpu_secs:.2}s"); println!("Average CPU: {avg_cpu_pct:.1}% ({avg_per_core_pct:.1}% per core)"); - let mut cpu_percents: Vec = Vec::new(); - for i in 1..self.samples.len() { - cpu_percents.push(self.samples[i].cpu_percent_since(&self.samples[i - 1])); - } - - if !cpu_percents.is_empty() { - let max_cpu = cpu_percents.iter().cloned().fold(0.0_f64, f64::max); - let min_cpu = cpu_percents.iter().cloned().fold(f64::MAX, f64::min); - - let mut sorted = cpu_percents.clone(); - sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal)); - let p50 = sorted[sorted.len() / 2]; - let p95_idx = ((sorted.len() as f64) * 0.95) as usize; - let p95 = sorted[p95_idx.min(sorted.len() - 1)]; - - println!("\nCPU % timeline (per sample interval):"); - println!(" Min: {min_cpu:.1}%"); - println!(" P50: {p50:.1}%"); - println!(" P95: {p95:.1}%"); - println!(" Max: {max_cpu:.1}%"); - } - - let thread_counts: Vec = self.samples.iter().map(|s| s.thread_count).collect(); - let max_threads = thread_counts.iter().copied().max().unwrap_or(0); - let min_threads = thread_counts.iter().copied().min().unwrap_or(0); - let last_threads = thread_counts.last().copied().unwrap_or(0); - - println!("\nThreads:"); - println!(" Current: {last_threads}"); - println!(" Min: {min_threads}"); - println!(" Max: {max_threads}"); - - if max_threads > min_threads + 10 { - println!(" *** Thread count variance is high — possible thread leak ***"); - } - - println!("\n=== CPU Assessment ==="); - let max_cpu_per_core = if !cpu_percents.is_empty() { - cpu_percents.iter().cloned().fold(0.0_f64, f64::max) / self.num_cores as f64 - } else { - 0.0 - }; - - if avg_per_core_pct < 10.0 { - println!(" CPU usage: LOW ({avg_per_core_pct:.1}% per core avg)"); - } else if avg_per_core_pct < 30.0 { - println!(" CPU usage: MODERATE ({avg_per_core_pct:.1}% per core avg)"); - } else { - println!( - " CPU usage: HIGH ({avg_per_core_pct:.1}% per core avg) — investigate hot paths" - ); - } - - if max_cpu_per_core > 80.0 { - println!(" CPU spikes: Peak {max_cpu_per_core:.0}% per core — may cause frame drops"); - } + let cpu_percents = self.compute_interval_percents(); + self.print_timeline_summary(&cpu_percents); + self.print_thread_summary(); + self.print_cpu_assessment(avg_per_core_pct, &cpu_percents); } } @@ -453,15 +473,10 @@ impl MemoryTracker { return; } + let growth = self.analyze_growth(); let first = &self.samples[0]; let last = &self.samples[self.samples.len() - 1]; let duration_secs = last.0.as_secs_f64() - first.0.as_secs_f64(); - let growth = last.1.primary_metric() - first.1.primary_metric(); - let growth_rate = if duration_secs > 0.0 { - growth / duration_secs - } else { - 0.0 - }; println!("Memory Timeline:"); println!( @@ -493,20 +508,14 @@ impl MemoryTracker { println!("Duration: {duration_secs:.1}s"); println!("Start: {:.1} MB", first.1.primary_metric()); println!("End: {:.1} MB", last.1.primary_metric()); - println!("Growth: {growth:.1} MB"); + println!("Growth: {:.1} MB", growth.total_mb); println!( "Rate: {:.2} MB/s ({:.1} MB/min)", - growth_rate, - growth_rate * 60.0 - ); - println!( - "Peak: {:.1} MB", - self.samples - .iter() - .map(|(_, s)| s.primary_metric()) - .fold(0.0_f64, |a, b| a.max(b)) + growth.rate_mb_per_sec, + growth.rate_mb_per_sec * 60.0 ); - println!("\nVerdict: {}", classify_leak(growth_rate)); + println!("Peak: {:.1} MB", growth.peak_mb); + println!("\nVerdict: {}", classify_leak(growth.rate_mb_per_sec)); } } diff --git a/crates/recording/src/sources/camera.rs b/crates/recording/src/sources/camera.rs index d1c16f06529..7db844baa97 100644 --- a/crates/recording/src/sources/camera.rs +++ b/crates/recording/src/sources/camera.rs @@ -10,11 +10,50 @@ use std::sync::{ Arc, atomic::{AtomicBool, AtomicU64, Ordering}, }; -use std::time::Duration; +use std::time::{Duration, Instant}; use tokio::sync::oneshot; const CAMERA_SENDER_ATTACH_TIMEOUT: Duration = Duration::from_millis(1500); +const CAMERA_CHANNEL_BYTE_BUDGET: usize = 64 * 1024 * 1024; +const MIN_CAMERA_CHANNEL_FRAMES: usize = 1; +const MAX_CAMERA_CHANNEL_FRAMES: usize = 64; + +pub(crate) fn estimate_video_frame_bytes(info: &VideoInfo) -> usize { + let pixels = (info.width as usize).saturating_mul(info.height as usize); + let bpp_num_den: (usize, usize) = match info.pixel_format { + ffmpeg::format::Pixel::NV12 + | ffmpeg::format::Pixel::YUV420P + | ffmpeg::format::Pixel::YUVJ420P => (3, 2), + ffmpeg::format::Pixel::YUYV422 + | ffmpeg::format::Pixel::UYVY422 + | ffmpeg::format::Pixel::YUV422P => (2, 1), + ffmpeg::format::Pixel::RGB24 | ffmpeg::format::Pixel::BGR24 => (3, 1), + _ => (4, 1), + }; + pixels + .saturating_mul(bpp_num_den.0) + .checked_div(bpp_num_den.1) + .unwrap_or(0) + .max(1) +} + +pub(crate) fn bounded_camera_channel_capacity(info: &VideoInfo) -> usize { + let frame_bytes = estimate_video_frame_bytes(info); + (CAMERA_CHANNEL_BYTE_BUDGET / frame_bytes) + .clamp(MIN_CAMERA_CHANNEL_FRAMES, MAX_CAMERA_CHANNEL_FRAMES) +} + +#[derive(Clone, Copy)] +struct ScalingSpec { + src_width: u32, + src_height: u32, + src_format: ffmpeg::format::Pixel, + dst_width: u32, + dst_height: u32, + dst_format: ffmpeg::format::Pixel, +} + struct CameraFrameScaler { context: ffmpeg::software::scaling::Context, source_width: u32, @@ -24,28 +63,21 @@ struct CameraFrameScaler { unsafe impl Send for CameraFrameScaler {} impl CameraFrameScaler { - fn new( - src_width: u32, - src_height: u32, - src_format: ffmpeg::format::Pixel, - dst_width: u32, - dst_height: u32, - dst_format: ffmpeg::format::Pixel, - ) -> anyhow::Result { + fn new(spec: ScalingSpec) -> anyhow::Result { let context = ffmpeg::software::scaling::Context::get( - src_format, - src_width, - src_height, - dst_format, - dst_width, - dst_height, + spec.src_format, + spec.src_width, + spec.src_height, + spec.dst_format, + spec.dst_width, + spec.dst_height, ffmpeg::software::scaling::Flags::BILINEAR, )?; Ok(Self { context, - source_width: src_width, - source_height: src_height, + source_width: spec.src_width, + source_height: spec.src_height, }) } @@ -61,6 +93,209 @@ impl CameraFrameScaler { } } +struct CameraLoopRuntime { + original_info: VideoInfo, + video_tx: mpsc::Sender, + stall_health_tx: crate::output_pipeline::HealthSender, + stopped: Arc, + scaled_frame_count: Arc, + scaler: Option, + frame_count: u64, + sent_count: u64, + dropped_count: u64, + start: Instant, +} + +impl CameraLoopRuntime { + fn ensure_scaler(&mut self, frame: &ffmpeg::frame::Video) -> bool { + let frame_width = frame.width(); + let frame_height = frame.height(); + let needs_new_scaler = self + .scaler + .as_ref() + .is_none_or(|s| !s.matches_source(frame_width, frame_height)); + + if !needs_new_scaler { + return true; + } + + let spec = ScalingSpec { + src_width: frame_width, + src_height: frame_height, + src_format: frame.format(), + dst_width: self.original_info.width, + dst_height: self.original_info.height, + dst_format: self.original_info.pixel_format, + }; + + match CameraFrameScaler::new(spec) { + Ok(new_scaler) => { + tracing::info!( + src_width = frame_width, + src_height = frame_height, + dst_width = self.original_info.width, + dst_height = self.original_info.height, + "Camera source: created scaler for dimension change" + ); + self.scaler = Some(new_scaler); + true + } + Err(e) => { + tracing::warn!( + src_width = frame_width, + src_height = frame_height, + error = %e, + "Camera source: failed to create scaler, dropping frame" + ); + self.dropped_count += 1; + false + } + } + } + + fn normalize_frame_dimensions(&mut self, frame: &mut FFmpegVideoFrame) -> bool { + let width_matches = frame.inner.width() == self.original_info.width; + let height_matches = frame.inner.height() == self.original_info.height; + + if width_matches && height_matches { + if self.scaler.take().is_some() { + let total_scaled = self.scaled_frame_count.load(Ordering::Relaxed); + tracing::info!( + total_scaled, + "Camera source: dimensions restored to original, removing scaler" + ); + } + return true; + } + + if !self.ensure_scaler(&frame.inner) { + return false; + } + + let Some(scaler) = self.scaler.as_mut() else { + return true; + }; + + match scaler.scale(&frame.inner) { + Ok(scaled) => { + frame.inner = scaled; + self.scaled_frame_count.fetch_add(1, Ordering::Relaxed); + true + } + Err(e) => { + if self.dropped_count.is_multiple_of(30) { + tracing::warn!(error = %e, "Camera source: scale failed, dropping frame"); + } + self.dropped_count += 1; + false + } + } + } + + fn forward_frame(&mut self, mut frame: FFmpegVideoFrame) -> bool { + self.frame_count += 1; + if !self.normalize_frame_dimensions(&mut frame) { + return true; + } + + let outcome = send_with_stall_budget_futures( + &mut self.video_tx, + frame, + "camera-video", + &self.stall_health_tx, + ); + self.record_send_outcome(outcome) + } + + fn record_send_outcome(&mut self, outcome: StallSendOutcome) -> bool { + match outcome { + StallSendOutcome::Sent => { + self.sent_count += 1; + if self.sent_count.is_multiple_of(300) { + let total_scaled = self.scaled_frame_count.load(Ordering::Relaxed); + tracing::debug!( + sent_count = self.sent_count, + dropped_count = self.dropped_count, + total_scaled, + elapsed = ?self.start.elapsed(), + "Camera source stats" + ); + } + true + } + StallSendOutcome::StalledAndDropped { .. } => { + self.dropped_count += 1; + if self.dropped_count.is_multiple_of(30) { + tracing::warn!( + dropped_count = self.dropped_count, + "Camera source: encoder can't keep up" + ); + } + true + } + StallSendOutcome::Disconnected => { + tracing::debug!( + sent_count = self.sent_count, + dropped_count = self.dropped_count, + "Camera source: pipeline closed" + ); + false + } + } + } + + async fn run(mut self, rx: flume::Receiver, stop_rx: oneshot::Receiver<()>) { + tracing::debug!( + original_width = self.original_info.width, + original_height = self.original_info.height, + "Camera source task started" + ); + let mut stop_rx = stop_rx.fuse(); + + loop { + if self.stopped.load(Ordering::Relaxed) { + tracing::debug!("Camera source: stop flag set, exiting"); + break; + } + + let keep_running = tokio::select! { + biased; + _ = &mut stop_rx => { + tracing::debug!("Camera source: received stop signal"); + false + } + result = rx.recv_async() => match result { + Ok(frame) => self.forward_frame(frame), + Err(e) => { + tracing::debug!( + frame_count = self.frame_count, + elapsed = ?self.start.elapsed(), + error = %e, + "Camera feed disconnected (rx closed)" + ); + false + } + } + }; + + if !keep_running { + break; + } + } + + drop(self.video_tx); + let total_scaled = self.scaled_frame_count.load(Ordering::Relaxed); + tracing::info!( + frame_count = self.frame_count, + sent_count = self.sent_count, + dropped_count = self.dropped_count, + total_scaled, + elapsed = ?self.start.elapsed(), + "Camera source finished" + ); + } +} + pub struct Camera { #[allow(dead_code)] feed_lock: Arc, @@ -82,14 +317,11 @@ impl VideoSource for Camera { where Self: Sized, { - let (tx, rx) = flume::bounded(256); let stall_health_tx = ctx.health_tx().clone(); let full_res = feed_lock.hold_full_resolution(); - let original_video_info = *feed_lock.video_info(); - let original_width = original_video_info.width; - let original_height = original_video_info.height; - let original_format = original_video_info.pixel_format; + let channel_capacity = bounded_camera_channel_capacity(&original_video_info); + let (tx, rx) = flume::bounded(channel_capacity); tokio::time::timeout( CAMERA_SENDER_ATTACH_TIMEOUT, @@ -101,173 +333,20 @@ impl VideoSource for Camera { let (stop_tx, stop_rx) = oneshot::channel(); let stopped = Arc::new(AtomicBool::new(false)); - let stopped_clone = stopped.clone(); - let scaled_frame_count = Arc::new(AtomicU64::new(0)); - let scaled_count_clone = scaled_frame_count.clone(); - - crate::output_pipeline::spawn_capture_task(async move { - tracing::debug!( - original_width, - original_height, - "Camera source task started" - ); - let mut frame_count: u64 = 0; - let mut sent_count: u64 = 0; - let mut dropped_count: u64 = 0; - let start = std::time::Instant::now(); - let mut video_tx = video_tx; - let mut stop_rx = stop_rx.fuse(); - let mut scaler: Option = None; - - loop { - if stopped_clone.load(Ordering::Relaxed) { - tracing::debug!("Camera source: stop flag set, exiting"); - break; - } + let runtime = CameraLoopRuntime { + original_info: original_video_info, + video_tx, + stall_health_tx, + stopped: stopped.clone(), + scaled_frame_count: Arc::new(AtomicU64::new(0)), + scaler: None, + frame_count: 0, + sent_count: 0, + dropped_count: 0, + start: Instant::now(), + }; - tokio::select! { - biased; - _ = &mut stop_rx => { - tracing::debug!("Camera source: received stop signal"); - break; - } - result = rx.recv_async() => { - match result { - Ok(mut frame) => { - frame_count += 1; - - let frame_width = frame.inner.width(); - let frame_height = frame.inner.height(); - - if frame_width != original_width || frame_height != original_height { - let needs_new_scaler = scaler - .as_ref() - .is_none_or(|s| !s.matches_source(frame_width, frame_height)); - - if needs_new_scaler { - let frame_format = frame.inner.format(); - match CameraFrameScaler::new( - frame_width, - frame_height, - frame_format, - original_width, - original_height, - original_format, - ) { - Ok(new_scaler) => { - tracing::info!( - src_width = frame_width, - src_height = frame_height, - dst_width = original_width, - dst_height = original_height, - "Camera source: created scaler for dimension change" - ); - scaler = Some(new_scaler); - } - Err(e) => { - tracing::warn!( - src_width = frame_width, - src_height = frame_height, - error = %e, - "Camera source: failed to create scaler, dropping frame" - ); - dropped_count += 1; - continue; - } - } - } - - if let Some(s) = &mut scaler { - match s.scale(&frame.inner) { - Ok(scaled) => { - frame.inner = scaled; - scaled_count_clone.fetch_add(1, Ordering::Relaxed); - } - Err(e) => { - if dropped_count.is_multiple_of(30) { - tracing::warn!( - error = %e, - "Camera source: scale failed, dropping frame" - ); - } - dropped_count += 1; - continue; - } - } - } - } else if scaler.is_some() { - let total_scaled = scaled_count_clone.load(Ordering::Relaxed); - tracing::info!( - total_scaled, - "Camera source: dimensions restored to original, removing scaler" - ); - scaler = None; - } - - match send_with_stall_budget_futures( - &mut video_tx, - frame, - "camera-video", - &stall_health_tx, - ) { - StallSendOutcome::Sent => { - sent_count += 1; - if sent_count.is_multiple_of(300) { - let total_scaled = scaled_count_clone.load(Ordering::Relaxed); - tracing::debug!( - sent_count, - dropped_count, - total_scaled, - elapsed = ?start.elapsed(), - "Camera source stats" - ); - } - } - StallSendOutcome::StalledAndDropped { .. } => { - dropped_count += 1; - if dropped_count.is_multiple_of(30) { - tracing::warn!( - dropped_count, - "Camera source: encoder can't keep up" - ); - } - } - StallSendOutcome::Disconnected => { - tracing::debug!( - sent_count, - dropped_count, - "Camera source: pipeline closed" - ); - break; - } - } - } - Err(e) => { - tracing::debug!( - frame_count, - elapsed = ?start.elapsed(), - error = %e, - "Camera feed disconnected (rx closed)" - ); - break; - } - } - } - } - } - - drop(video_tx); - - let total_scaled = scaled_count_clone.load(Ordering::Relaxed); - tracing::info!( - frame_count, - sent_count, - dropped_count, - total_scaled, - elapsed = ?start.elapsed(), - "Camera source finished" - ); - }); + crate::output_pipeline::spawn_capture_task(runtime.run(rx, stop_rx)); Ok(Self { feed_lock, @@ -294,3 +373,37 @@ impl VideoSource for Camera { .boxed() } } + +#[cfg(test)] +mod tests { + use super::*; + use cap_media_info::Pixel; + + #[test] + fn bounds_4k_bgra_camera_channel_within_memory_budget() { + let info = VideoInfo { + pixel_format: Pixel::BGRA, + width: 3840, + height: 2160, + time_base: ffmpeg::Rational(1, 30), + frame_rate: ffmpeg::Rational(30, 1), + }; + let capacity = bounded_camera_channel_capacity(&info); + let retained_bytes = capacity * estimate_video_frame_bytes(&info); + assert!(retained_bytes <= CAMERA_CHANNEL_BYTE_BUDGET); + } + + #[test] + fn bounds_1080p_nv12_camera_channel_by_byte_budget() { + let info = VideoInfo { + pixel_format: Pixel::NV12, + width: 1920, + height: 1080, + time_base: ffmpeg::Rational(1, 30), + frame_rate: ffmpeg::Rational(30, 1), + }; + let capacity = bounded_camera_channel_capacity(&info); + assert!(capacity >= MIN_CAMERA_CHANNEL_FRAMES); + assert!(capacity <= MAX_CAMERA_CHANNEL_FRAMES); + } +}