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/enc-ffmpeg/src/mux/segmented_stream.rs b/crates/enc-ffmpeg/src/mux/segmented_stream.rs index 0ea15ad1566..6f6d75f826c 100644 --- a/crates/enc-ffmpeg/src/mux/segmented_stream.rs +++ b/crates/enc-ffmpeg/src/mux/segmented_stream.rs @@ -480,6 +480,19 @@ impl SegmentedVideoEncoder { Ok(()) } + pub fn queue_frame_reusable( + &mut self, + frame: &mut frame::Video, + timestamp: Duration, + ) -> Result<(), QueueFrameError> { + self.encoder + .queue_frame_to_reusable(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 01db20997a7..c6a93749ad4 100644 --- a/crates/enc-ffmpeg/src/video/h264.rs +++ b/crates/enc-ffmpeg/src/video/h264.rs @@ -990,6 +990,40 @@ impl H264Encoder { self.queue_frame_to(frame, timestamp, output) } + pub fn queue_frame_to_reusable( + &mut self, + frame: &mut frame::Video, + timestamp: Duration, + output: &mut S, + ) -> Result<(), QueueFrameError> { + self.base.update_pts(frame, timestamp, &mut self.encoder); + + let frame_to_send = if let Some(converter) = &mut self.converter { + let pts = frame.pts(); + let converted = self.converted_frame_pool.as_mut().unwrap(); + let ret = unsafe { ffmpeg::ffi::av_frame_make_writable(converted.as_mut_ptr()) }; + if ret < 0 { + return Err(QueueFrameError::Encode(ffmpeg::Error::from(ret))); + } + super::convert_or_scale(frame, converted, converter) + .map_err(QueueFrameError::Converter)?; + converted.set_pts(pts); + converted as &frame::Video + } else { + frame as &frame::Video + }; + let frame_to_send = self + .hw_upload + .frame(frame_to_send) + .map_err(QueueFrameError::Encode)?; + + self.base + .send_frame(frame_to_send, output, &mut self.encoder) + .map_err(QueueFrameError::Encode)?; + + Ok(()) + } + pub fn queue_frame_to( &mut self, mut frame: frame::Video, @@ -1002,6 +1036,10 @@ impl H264Encoder { let frame_to_send = if let Some(converter) = &mut self.converter { let pts = frame.pts(); let converted = self.converted_frame_pool.as_mut().unwrap(); + let ret = unsafe { ffmpeg::ffi::av_frame_make_writable(converted.as_mut_ptr()) }; + if ret < 0 { + return Err(QueueFrameError::Encode(ffmpeg::Error::from(ret))); + } super::convert_or_scale(&frame, converted, converter) .map_err(QueueFrameError::Converter)?; converted.set_pts(pts); @@ -1032,9 +1070,20 @@ impl H264Encoder { let frame_to_send = if let Some(converter) = &mut self.converter { let pts = frame.pts(); + if converted_frame.as_ref().is_some_and(|f| { + f.format() != self.output_format + || f.width() != self.output_width + || f.height() != self.output_height + }) { + *converted_frame = None; + } let converted = converted_frame.get_or_insert_with(|| { frame::Video::new(self.output_format, self.output_width, self.output_height) }); + let ret = unsafe { ffmpeg::ffi::av_frame_make_writable(converted.as_mut_ptr()) }; + if ret < 0 { + return Err(QueueFrameError::Encode(ffmpeg::Error::from(ret))); + } super::convert_or_scale(frame, converted, converter) .map_err(QueueFrameError::Converter)?; converted.set_pts(pts); diff --git a/crates/enc-ffmpeg/src/video/h264_packet.rs b/crates/enc-ffmpeg/src/video/h264_packet.rs index 9f702f1a23f..98d2c409a9a 100644 --- a/crates/enc-ffmpeg/src/video/h264_packet.rs +++ b/crates/enc-ffmpeg/src/video/h264_packet.rs @@ -143,6 +143,10 @@ impl H264PacketEncoder { let converted = self.converted_frame_pool.as_mut().expect( "converted_frame_pool present whenever converter is Some (invariant preserved by from_opened)", ); + let ret = unsafe { ffmpeg::ffi::av_frame_make_writable(converted.as_mut_ptr()) }; + if ret < 0 { + return Err(EncodePacketError::Encode(ffmpeg::Error::from(ret))); + } super::convert_or_scale(&frame, converted, converter) .map_err(EncodePacketError::Converter)?; converted.set_pts(pts); @@ -176,9 +180,20 @@ impl H264PacketEncoder { let frame_to_send = if let Some(converter) = &mut self.converter { let pts = frame.pts(); + if converted_frame.as_ref().is_some_and(|f| { + f.format() != self.output_format + || f.width() != self.output_width + || f.height() != self.output_height + }) { + *converted_frame = None; + } let converted = converted_frame.get_or_insert_with(|| { frame::Video::new(self.output_format, self.output_width, self.output_height) }); + let ret = unsafe { ffmpeg::ffi::av_frame_make_writable(converted.as_mut_ptr()) }; + if ret < 0 { + return Err(EncodePacketError::Encode(ffmpeg::Error::from(ret))); + } super::convert_or_scale(frame, converted, converter) .map_err(EncodePacketError::Converter)?; converted.set_pts(pts); diff --git a/crates/recording/src/output_pipeline/macos.rs b/crates/recording/src/output_pipeline/macos.rs index 53e73bca231..fa233e4206e 100644 --- a/crates/recording/src/output_pipeline/macos.rs +++ b/crates/recording/src/output_pipeline/macos.rs @@ -455,7 +455,7 @@ impl Muxer for AVFoundationMp4Muxer { const MAX_RETRIES: u32 = 100; loop { - let queue_result = { + let queue_result = cidre::objc::ar_pool(|| { let mut encoder = match encoder_clone.lock() { Ok(e) => e, Err(_) => { @@ -464,8 +464,8 @@ impl Muxer for AVFoundationMp4Muxer { return Err(anyhow!(message)); } }; - encoder.queue_video_frame(sample_buf.clone(), timestamp) - }; + Ok(encoder.queue_video_frame(sample_buf.clone(), timestamp)) + })?; match queue_result { Ok(()) => break, @@ -586,7 +586,7 @@ impl Muxer for AVFoundationMp4Muxer { const MAX_RETRIES: u32 = 50; loop { - let queue_result = { + let queue_result = cidre::objc::ar_pool(|| { let mut encoder = match encoder_clone.lock() { Ok(e) => e, Err(_) => { @@ -599,8 +599,8 @@ impl Muxer for AVFoundationMp4Muxer { return Err(anyhow!(message)); } }; - encoder.queue_audio_frame(&frame, timestamp) - }; + Ok(encoder.queue_audio_frame(&frame, timestamp)) + })?; match queue_result { Ok(()) => break, @@ -1072,7 +1072,7 @@ impl Muxer for AVFoundationCameraMuxer { const MAX_RETRIES: u32 = 100; loop { - let queue_result = { + let queue_result = cidre::objc::ar_pool(|| { let mut encoder = match encoder_clone.lock() { Ok(e) => e, Err(_) => { @@ -1082,8 +1082,8 @@ impl Muxer for AVFoundationCameraMuxer { return Err(anyhow!(message)); } }; - encoder.queue_video_frame(sample_buf.clone(), timestamp) - }; + Ok(encoder.queue_video_frame(sample_buf.clone(), timestamp)) + })?; match queue_result { Ok(()) => break, @@ -1203,7 +1203,7 @@ impl Muxer for AVFoundationCameraMuxer { const MAX_RETRIES: u32 = 50; loop { - let queue_result = { + let queue_result = cidre::objc::ar_pool(|| { let mut encoder = match encoder_clone.lock() { Ok(e) => e, Err(_) => { @@ -1212,8 +1212,8 @@ impl Muxer for AVFoundationCameraMuxer { return Err(anyhow!(message)); } }; - encoder.queue_audio_frame(&frame, timestamp) - }; + Ok(encoder.queue_audio_frame(&frame, timestamp)) + })?; match queue_result { Ok(()) => break, diff --git a/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs b/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs index cd36e3bff3b..5716c6fd576 100644 --- a/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs +++ b/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs @@ -346,114 +346,116 @@ impl MacOSFragmentedM4SMuxer { let slow_threshold_ms = frame_timing_log_threshold_ms(&video_config); while let Ok(Some((sample_buf, timestamp))) = video_rx.recv() { - if matches!( - disk_monitor.poll(&base_path, &health_tx), - super::core::DiskSpacePollResult::Exhausted { .. } - | super::core::DiskSpacePollResult::Stopped - ) { - disk_exhausted = true; - } - if disk_exhausted { - continue; - } + cidre::objc::ar_pool(|| -> anyhow::Result<()> { + if matches!( + disk_monitor.poll(&base_path, &health_tx), + super::core::DiskSpacePollResult::Exhausted { .. } + | super::core::DiskSpacePollResult::Stopped + ) { + disk_exhausted = true; + } + if disk_exhausted { + return Ok(()); + } - 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 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 - ); + 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(); - 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_reusable(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_reusable 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")), - }; + None => Err(anyhow!("sample buffer has no image buffer")), + }; - if let Err(e) = queued { - warn!("Failed to encode frame: {e}"); - } + 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 + ); + } } } - } - total_frames += 1; + total_frames += 1; + Ok(()) + })?; } if total_frames > 0 { @@ -735,71 +737,73 @@ impl MacOSFragmentedM4SCameraMuxer { let slow_threshold_ms = frame_timing_log_threshold_ms(&video_config); while let Ok(Some((sample_buf, timestamp))) = video_rx.recv() { - if matches!( - disk_monitor.poll(&base_path, &health_tx), - super::core::DiskSpacePollResult::Exhausted { .. } - | super::core::DiskSpacePollResult::Stopped - ) { - disk_exhausted = true; - } - if disk_exhausted { - continue; - } + cidre::objc::ar_pool(|| -> anyhow::Result<()> { + if matches!( + disk_monitor.poll(&base_path, &health_tx), + super::core::DiskSpacePollResult::Exhausted { .. } + | super::core::DiskSpacePollResult::Stopped + ) { + disk_exhausted = true; + } + if disk_exhausted { + return Ok(()); + } + + 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(); - 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, - "Camera fill_frame_from_sample_buf exceeded {}ms threshold", - slow_threshold_ms - ); + 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, + "Camera 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(); - match encoder_clone.lock() { - Ok(mut encoder) => { - if let Err(e) = encoder.queue_frame(owned_frame, timestamp) { - warn!("Failed to encode camera frame: {e}"); + match encoder_clone.lock() { + Ok(mut encoder) => { + if let Err(e) = encoder.queue_frame_reusable(frame, timestamp) { + warn!("Failed to encode camera frame: {e}"); + } + } + Err(_) => { + error!("Camera encoder mutex poisoned - encoder thread likely panicked, stopping"); + return Err(anyhow!("Camera encoder mutex poisoned - all subsequent frames would be lost")); } } - Err(_) => { - error!("Camera encoder mutex poisoned - encoder thread likely panicked, stopping"); - return Err(anyhow!("Camera encoder mutex poisoned - all subsequent frames would be lost")); - } - } - 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, - "Camera encoder.queue_frame 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, + "Camera encoder.queue_frame_reusable exceeded {}ms threshold", + slow_threshold_ms + ); + } } } + Err(e) => { + warn!("Failed to convert camera frame: {e:?}"); + } } - Err(e) => { - warn!("Failed to convert camera frame: {e:?}"); - } - } - total_frames += 1; + total_frames += 1; + Ok(()) + })?; } if total_frames > 0 { diff --git a/crates/recording/src/output_pipeline/macos_frame_convert.rs b/crates/recording/src/output_pipeline/macos_frame_convert.rs index 552f428f500..92dbd61e434 100644 --- a/crates/recording/src/output_pipeline/macos_frame_convert.rs +++ b/crates/recording/src/output_pipeline/macos_frame_convert.rs @@ -54,6 +54,11 @@ impl FramePool { } pub(crate) fn get_frame(&mut self) -> &mut ffmpeg::frame::Video { + if let Some(ref mut frame) = self.frame + && unsafe { ffmpeg::ffi::av_frame_make_writable(frame.as_mut_ptr()) } < 0 + { + self.frame = None; + } if self.frame.is_none() { self.frame = Some(ffmpeg::frame::Video::new( self.pixel_format, @@ -238,6 +243,25 @@ mod tests { } } + #[test] + fn frame_pool_reuses_backing_buffer_when_writable_and_detaches_when_shared() { + let mut pool = super::FramePool::new(ffmpeg::format::Pixel::NV12, 64, 64); + let first_ptr = pool.get_frame().data(0).as_ptr(); + let second_ptr = pool.get_frame().data(0).as_ptr(); + assert_eq!( + first_ptr, second_ptr, + "writable pool frame should reuse backing buffer" + ); + + let retained = pool.get_frame().clone(); + let third_ptr = pool.get_frame().data(0).as_ptr(); + assert_ne!( + first_ptr, third_ptr, + "shared pool frame should allocate a writable replacement buffer via av_frame_make_writable" + ); + drop(retained); + } + #[test] fn empty_and_degenerate_planes_do_not_panic() { let mut dest = vec![0; 64]; diff --git a/crates/recording/src/output_pipeline/oop_fragmented_m4s.rs b/crates/recording/src/output_pipeline/oop_fragmented_m4s.rs index 9afbb643412..e2cff1d141e 100644 --- a/crates/recording/src/output_pipeline/oop_fragmented_m4s.rs +++ b/crates/recording/src/output_pipeline/oop_fragmented_m4s.rs @@ -374,6 +374,7 @@ impl OutOfProcessFragmentedM4SMuxer { let pixel_format = ffmpeg_pixel_format_for_cap(video_config.pixel_format); let mut frame_pool = FramePool::new(pixel_format, video_config.width, video_config.height); + let mut converted_frame: Option = None; let mut encoder = encoder; let mut subprocess = subprocess; @@ -389,59 +390,60 @@ impl OutOfProcessFragmentedM4SMuxer { let slow_threshold_ms = frame_timing_log_threshold_ms(&video_config); while let Ok(Some((sample_buf, timestamp))) = video_rx.recv() { - last_timestamp = Some(timestamp); + cidre::objc::ar_pool(|| { + last_timestamp = Some(timestamp); - match disk_monitor.poll(&base_path, &health_tx) { - super::core::DiskSpacePollResult::Exhausted { .. } => { - disk_exhausted = true; - } - super::core::DiskSpacePollResult::Stopped => { - disk_exhausted = true; + match disk_monitor.poll(&base_path, &health_tx) { + super::core::DiskSpacePollResult::Exhausted { .. } => { + disk_exhausted = true; + } + super::core::DiskSpacePollResult::Stopped => { + disk_exhausted = true; + } + _ => {} } - _ => {} - } - if disk_exhausted { - continue; - } + if disk_exhausted { + return; + } - if subprocess_fatal || subprocess.is_exhausted() { - continue; - } + if subprocess_fatal || subprocess.is_exhausted() { + return; + } - 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, - "OOP fill_frame_from_sample_buf exceeded {}ms threshold", - slow_threshold_ms - ); + 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, + "OOP fill_frame_from_sample_buf exceeded {}ms threshold", + slow_threshold_ms + ); + } } - } - match fill_result { - Ok(()) => { - let owned_frame = frame_pool.take_frame(); - let encode_start = std::time::Instant::now(); - let subprocess_ref = &mut subprocess; - let keyframe_emitted_ref = &mut keyframe_emitted; - let encode_result = encoder.encode_frame( - owned_frame, - timestamp, - |pkt| { - if pkt.is_keyframe { - *keyframe_emitted_ref = true; - } - dispatch_packet(subprocess_ref, pkt) - }, - ); + match fill_result { + Ok(()) => { + let encode_start = std::time::Instant::now(); + let subprocess_ref = &mut subprocess; + let keyframe_emitted_ref = &mut keyframe_emitted; + let encode_result = encoder.encode_frame_reusable( + frame, + &mut converted_frame, + timestamp, + |pkt| { + if pkt.is_keyframe { + *keyframe_emitted_ref = true; + } + dispatch_packet(subprocess_ref, pkt) + }, + ); let encode_elapsed_ms = encode_start.elapsed().as_millis(); if encode_elapsed_ms > slow_threshold_ms { slow_encode_count += 1; @@ -498,6 +500,7 @@ impl OutOfProcessFragmentedM4SMuxer { "OOP encoder rebuilt mid-stream" ); encoder = new_enc; + converted_frame = None; health_tx.emit( super::core::PipelineHealthEvent::EncoderRebuilt { backend: new_name, @@ -523,9 +526,10 @@ impl OutOfProcessFragmentedM4SMuxer { Err(e) => { warn!("OOP failed to convert frame: {e:?}"); } - } + } - total_frames += 1; + total_frames += 1; + }); } let final_ts = last_timestamp.unwrap_or(Duration::ZERO);