From 6e20b6a2ca3d9ca5c91615206c2247611adfd129 Mon Sep 17 00:00:00 2001 From: Johann Date: Sun, 1 Nov 2026 16:13:48 -0300 Subject: [PATCH 1/2] perf(recording): drain macOS autorelease pools during camera capture and muxing (fixes #109) - Wrap CameraInfo::formats_impl in objc::ar_pool to prevent autoreleased NSStrings and format descriptions from accumulating - Configure AVCaptureVideoDataOutput callback queue with dispatch::Queue::serial_with_ar_pool() (matching scap-screencapturekit) so camera sample buffer delegate blocks release autoreleased CoreMedia/CoreVideo objects on every frame - Wrap AVFoundationMp4Muxer, MacOSFragmentedM4SMuxer, and OutOfProcessFragmentedM4SMuxer frame processing/conversion inside cidre::objc::ar_pool to eliminate unbounded memory growth during long recordings Closes #109 --- crates/camera/src/macos.rs | 64 +++---- crates/recording/src/output_pipeline/macos.rs | 110 ++++++------ .../output_pipeline/macos_fragmented_m4s.rs | 163 +++++++++--------- .../src/output_pipeline/oop_fragmented_m4s.rs | 3 +- 4 files changed, 179 insertions(+), 161 deletions(-) diff --git a/crates/camera/src/macos.rs b/crates/camera/src/macos.rs index fece38486f1..128e0f2e395 100644 --- a/crates/camera/src/macos.rs +++ b/crates/camera/src/macos.rs @@ -24,38 +24,40 @@ 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() { + 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()), + }) + } } - } - Some(ret) + Some(ret) + }) } } @@ -156,7 +158,7 @@ pub(super) fn start_capturing_impl( let input = av::capture::DeviceInput::with_device(&device).map_err(AVFoundationError::Static)?; - let queue = dispatch::Queue::new(); + let queue = dispatch::Queue::serial_with_ar_pool(); let mut missing_image_bufs: u64 = 0; let delegate = CallbackOutputDelegate::with(CallbackOutputDelegateInner::new(Box::new(move |data| { diff --git a/crates/recording/src/output_pipeline/macos.rs b/crates/recording/src/output_pipeline/macos.rs index 0fd2e282a9b..d31ef429ce7 100644 --- a/crates/recording/src/output_pipeline/macos.rs +++ b/crates/recording/src/output_pipeline/macos.rs @@ -457,65 +457,73 @@ impl Muxer for AVFoundationMp4Muxer { let mut retry_count = 0; const MAX_RETRIES: u32 = 100; - loop { - let queue_result = { - let mut encoder = match encoder_clone.lock() { - Ok(e) => e, - Err(_) => { - let message = "MP4 encoder mutex poisoned".to_string(); - set_fatal_error(&video_fatal_error, message.clone()); - return Err(anyhow!(message)); - } + let step_result = cidre::objc::ar_pool(|| { + loop { + let queue_result = { + let mut encoder = match encoder_clone.lock() { + Ok(e) => e, + Err(_) => { + let message = "MP4 encoder mutex poisoned".to_string(); + set_fatal_error(&video_fatal_error, message.clone()); + return Err(anyhow!(message)); + } + }; + encoder.queue_video_frame(sample_buf.clone(), timestamp) }; - encoder.queue_video_frame(sample_buf.clone(), timestamp) - }; - match queue_result { - Ok(()) => break, - Err(QueueFrameError::NotReadyForMore) => { - retry_count += 1; - if retry_count >= MAX_RETRIES { - encoder_busy_count += 1; - video_busy_thread.store( - encoder_busy_count, - std::sync::atomic::Ordering::Relaxed, - ); - if encoder_busy_count <= 5 || encoder_busy_count.is_multiple_of(100) { - debug!( + match queue_result { + Ok(()) => break Ok(None), + Err(QueueFrameError::NotReadyForMore) => { + retry_count += 1; + if retry_count >= MAX_RETRIES { + encoder_busy_count += 1; + video_busy_thread.store( encoder_busy_count, - "MP4 encoder busy, frame queued after max retries" + std::sync::atomic::Ordering::Relaxed, ); + if encoder_busy_count <= 5 || encoder_busy_count.is_multiple_of(100) { + debug!( + encoder_busy_count, + "MP4 encoder busy, frame queued after max retries" + ); + } + break Ok(None); } - break; + std::thread::sleep(Duration::from_micros(500)); + } + Err(QueueFrameError::WriterFailed(err)) => { + let total = video_count_thread + .load(std::sync::atomic::Ordering::Relaxed); + let message = format!( + "Failed to encode video frame: WriterFailed/{err:?} \ + (frame #{total}, ts={timestamp:?})" + ); + set_fatal_error(&video_fatal_error, message.clone()); + return Err(anyhow!(message)); + } + Err(QueueFrameError::Failed) => { + let total = video_count_thread + .load(std::sync::atomic::Ordering::Relaxed); + let message = format!( + "Failed to encode video frame: Failed \ + (frame #{total}, ts={timestamp:?})" + ); + set_fatal_error(&video_fatal_error, message.clone()); + return Err(anyhow!(message)); + } + Err(QueueFrameError::Finished) => return Ok(Some(())), + Err(e) => { + warn!("Failed to encode video frame: {e}"); + break Ok(None); } - std::thread::sleep(Duration::from_micros(500)); - } - Err(QueueFrameError::WriterFailed(err)) => { - let total = video_count_thread - .load(std::sync::atomic::Ordering::Relaxed); - let message = format!( - "Failed to encode video frame: WriterFailed/{err:?} \ - (frame #{total}, ts={timestamp:?})" - ); - set_fatal_error(&video_fatal_error, message.clone()); - return Err(anyhow!(message)); - } - Err(QueueFrameError::Failed) => { - let total = video_count_thread - .load(std::sync::atomic::Ordering::Relaxed); - let message = format!( - "Failed to encode video frame: Failed \ - (frame #{total}, ts={timestamp:?})" - ); - set_fatal_error(&video_fatal_error, message.clone()); - return Err(anyhow!(message)); - } - Err(QueueFrameError::Finished) => return Ok(()), - Err(e) => { - warn!("Failed to encode video frame: {e}"); - break; } } + }); + + match step_result { + Ok(Some(())) => return Ok(()), + Ok(None) => {} + Err(err) => return Err(err), } video_count_thread.fetch_add(1, std::sync::atomic::Ordering::Relaxed); diff --git a/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs b/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs index a0390e30141..0b5e0c4f64d 100644 --- a/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs +++ b/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs @@ -370,100 +370,107 @@ impl MacOSFragmentedM4SMuxer { continue; } - if let Some(frame_pool) = frame_pool.as_mut() { - let convert_start = std::time::Instant::now(); - let frame = frame_pool.get_frame(); - let fill_result = fill_frame_from_sample_buf(&sample_buf, frame); - let convert_elapsed_ms = convert_start.elapsed().as_millis(); - - if convert_elapsed_ms > slow_threshold_ms { - slow_convert_count += 1; - if slow_convert_count <= 5 || slow_convert_count.is_multiple_of(100) { - debug!( - elapsed_ms = convert_elapsed_ms, - count = slow_convert_count, - "fill_frame_from_sample_buf exceeded {}ms threshold", - slow_threshold_ms - ); + let step_result = cidre::objc::ar_pool(|| -> Result<(), anyhow::Error> { + if let Some(frame_pool) = frame_pool.as_mut() { + let convert_start = std::time::Instant::now(); + let frame = frame_pool.get_frame(); + let fill_result = fill_frame_from_sample_buf(&sample_buf, frame); + let convert_elapsed_ms = convert_start.elapsed().as_millis(); + + if convert_elapsed_ms > slow_threshold_ms { + slow_convert_count += 1; + if slow_convert_count <= 5 || slow_convert_count.is_multiple_of(100) { + debug!( + elapsed_ms = convert_elapsed_ms, + count = slow_convert_count, + "fill_frame_from_sample_buf exceeded {}ms threshold", + slow_threshold_ms + ); + } } - } - match fill_result { - Ok(()) => { - let encode_start = std::time::Instant::now(); - let owned_frame = frame_pool.take_frame(); + match fill_result { + Ok(()) => { + let encode_start = std::time::Instant::now(); + let owned_frame = frame_pool.take_frame(); - match encoder_clone.lock() { - Ok(mut encoder) => { - if let Err(e) = encoder.queue_frame(owned_frame, timestamp) { - warn!("Failed to encode frame: {e}"); + match encoder_clone.lock() { + Ok(mut encoder) => { + if let Err(e) = encoder.queue_frame(owned_frame, timestamp) { + warn!("Failed to encode frame: {e}"); + } + } + Err(_) => { + error!("Encoder mutex poisoned - encoder thread likely panicked, stopping"); + return Err(anyhow!("Encoder mutex poisoned - all subsequent frames would be lost")); } } - Err(_) => { - error!("Encoder mutex poisoned - encoder thread likely panicked, stopping"); - 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 - ); + 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 + ); + } } } + Err(e) => { + warn!("Failed to convert frame: {e:?}"); + } } - Err(e) => { - warn!("Failed to convert frame: {e:?}"); - } - } - } else { - let encode_start = std::time::Instant::now(); - - let queued = match sample_buf.image_buf() { - Some(image_buf) => { - let ptr = (std::ptr::from_ref(image_buf) - as *const std::ffi::c_void) - .cast_mut(); - match encoder_clone.lock() { - Ok(mut encoder) => { - unsafe { encoder.queue_hw_pixel_buffer(ptr, timestamp) } - .map_err(|e| anyhow!("{e}")) - } - Err(_) => { - error!("Encoder mutex poisoned - encoder thread likely panicked, stopping"); - return Err(anyhow!("Encoder mutex poisoned - all subsequent frames would be lost")); + } else { + let encode_start = std::time::Instant::now(); + + let queued = match sample_buf.image_buf() { + Some(image_buf) => { + let ptr = (std::ptr::from_ref(image_buf) + as *const std::ffi::c_void) + .cast_mut(); + match encoder_clone.lock() { + Ok(mut encoder) => { + unsafe { encoder.queue_hw_pixel_buffer(ptr, timestamp) } + .map_err(|e| anyhow!("{e}")) + } + Err(_) => { + error!("Encoder mutex poisoned - encoder thread likely panicked, stopping"); + return Err(anyhow!("Encoder mutex poisoned - all subsequent frames would be lost")); + } } } + None => Err(anyhow!("sample buffer has no image buffer")), + }; + + if let Err(e) = queued { + warn!("Failed to encode frame: {e}"); } - None => Err(anyhow!("sample buffer has no image buffer")), - }; - if let Err(e) = queued { - warn!("Failed to encode frame: {e}"); - } + let encode_elapsed_ms = encode_start.elapsed().as_millis(); - 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_hw_pixel_buffer exceeded {}ms threshold", - slow_threshold_ms - ); + 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_hw_pixel_buffer exceeded {}ms threshold", + slow_threshold_ms + ); + } } } + Ok(()) + }); + + if let Err(err) = step_result { + return Err(err); } total_frames += 1; diff --git a/crates/recording/src/output_pipeline/oop_fragmented_m4s.rs b/crates/recording/src/output_pipeline/oop_fragmented_m4s.rs index 9afbb643412..d2c7a872284 100644 --- a/crates/recording/src/output_pipeline/oop_fragmented_m4s.rs +++ b/crates/recording/src/output_pipeline/oop_fragmented_m4s.rs @@ -411,7 +411,8 @@ impl OutOfProcessFragmentedM4SMuxer { let convert_start = std::time::Instant::now(); let frame = frame_pool.get_frame(); - let fill_result = fill_frame_from_sample_buf(&sample_buf, frame); + let fill_result = + cidre::objc::ar_pool(|| fill_frame_from_sample_buf(&sample_buf, frame)); let convert_elapsed_ms = convert_start.elapsed().as_millis(); if convert_elapsed_ms > slow_threshold_ms { From 5b6337a23a163c6e0f7436a5626082674a332287 Mon Sep 17 00:00:00 2001 From: Johann Date: Fri, 6 Nov 2026 21:49:36 +0000 Subject: [PATCH 2/2] fix(recording): satisfy cidre ar_pool Clone bound by returning unit from closures --- crates/recording/src/output_pipeline/macos.rs | 28 +++++++++++++------ .../output_pipeline/macos_fragmented_m4s.rs | 10 ++++--- .../src/output_pipeline/oop_fragmented_m4s.rs | 6 ++-- 3 files changed, 30 insertions(+), 14 deletions(-) diff --git a/crates/recording/src/output_pipeline/macos.rs b/crates/recording/src/output_pipeline/macos.rs index d31ef429ce7..7dba784546d 100644 --- a/crates/recording/src/output_pipeline/macos.rs +++ b/crates/recording/src/output_pipeline/macos.rs @@ -457,7 +457,8 @@ impl Muxer for AVFoundationMp4Muxer { let mut retry_count = 0; const MAX_RETRIES: u32 = 100; - let step_result = cidre::objc::ar_pool(|| { + let mut step_result = Ok(None); + cidre::objc::ar_pool(|| { loop { let queue_result = { let mut encoder = match encoder_clone.lock() { @@ -465,14 +466,18 @@ impl Muxer for AVFoundationMp4Muxer { Err(_) => { let message = "MP4 encoder mutex poisoned".to_string(); set_fatal_error(&video_fatal_error, message.clone()); - return Err(anyhow!(message)); + step_result = Err(anyhow!(message)); + return; } }; encoder.queue_video_frame(sample_buf.clone(), timestamp) }; match queue_result { - Ok(()) => break Ok(None), + Ok(()) => { + step_result = Ok(None); + break; + } Err(QueueFrameError::NotReadyForMore) => { retry_count += 1; if retry_count >= MAX_RETRIES { @@ -487,7 +492,8 @@ impl Muxer for AVFoundationMp4Muxer { "MP4 encoder busy, frame queued after max retries" ); } - break Ok(None); + step_result = Ok(None); + break; } std::thread::sleep(Duration::from_micros(500)); } @@ -499,7 +505,8 @@ impl Muxer for AVFoundationMp4Muxer { (frame #{total}, ts={timestamp:?})" ); set_fatal_error(&video_fatal_error, message.clone()); - return Err(anyhow!(message)); + step_result = Err(anyhow!(message)); + return; } Err(QueueFrameError::Failed) => { let total = video_count_thread @@ -509,12 +516,17 @@ impl Muxer for AVFoundationMp4Muxer { (frame #{total}, ts={timestamp:?})" ); set_fatal_error(&video_fatal_error, message.clone()); - return Err(anyhow!(message)); + step_result = Err(anyhow!(message)); + return; + } + Err(QueueFrameError::Finished) => { + step_result = Ok(Some(())); + return; } - Err(QueueFrameError::Finished) => return Ok(Some(())), Err(e) => { warn!("Failed to encode video frame: {e}"); - break Ok(None); + step_result = Ok(None); + break; } } } diff --git a/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs b/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs index 0b5e0c4f64d..53b158a31fc 100644 --- a/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs +++ b/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs @@ -370,7 +370,8 @@ impl MacOSFragmentedM4SMuxer { continue; } - let step_result = cidre::objc::ar_pool(|| -> Result<(), anyhow::Error> { + let mut step_result = Ok(()); + cidre::objc::ar_pool(|| { if let Some(frame_pool) = frame_pool.as_mut() { let convert_start = std::time::Instant::now(); let frame = frame_pool.get_frame(); @@ -402,7 +403,8 @@ impl MacOSFragmentedM4SMuxer { } Err(_) => { error!("Encoder mutex poisoned - encoder thread likely panicked, stopping"); - return Err(anyhow!("Encoder mutex poisoned - all subsequent frames would be lost")); + step_result = Err(anyhow!("Encoder mutex poisoned - all subsequent frames would be lost")); + return; } } @@ -441,7 +443,8 @@ impl MacOSFragmentedM4SMuxer { } Err(_) => { error!("Encoder mutex poisoned - encoder thread likely panicked, stopping"); - return Err(anyhow!("Encoder mutex poisoned - all subsequent frames would be lost")); + step_result = Err(anyhow!("Encoder mutex poisoned - all subsequent frames would be lost")); + return; } } } @@ -466,7 +469,6 @@ impl MacOSFragmentedM4SMuxer { } } } - Ok(()) }); if let Err(err) = step_result { diff --git a/crates/recording/src/output_pipeline/oop_fragmented_m4s.rs b/crates/recording/src/output_pipeline/oop_fragmented_m4s.rs index d2c7a872284..46d8c892472 100644 --- a/crates/recording/src/output_pipeline/oop_fragmented_m4s.rs +++ b/crates/recording/src/output_pipeline/oop_fragmented_m4s.rs @@ -411,8 +411,10 @@ impl OutOfProcessFragmentedM4SMuxer { let convert_start = std::time::Instant::now(); let frame = frame_pool.get_frame(); - let fill_result = - cidre::objc::ar_pool(|| fill_frame_from_sample_buf(&sample_buf, frame)); + let mut fill_result = Ok(()); + cidre::objc::ar_pool(|| { + fill_result = fill_frame_from_sample_buf(&sample_buf, frame); + }); let convert_elapsed_ms = convert_start.elapsed().as_millis(); if convert_elapsed_ms > slow_threshold_ms {