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..fbce57865d2 100644 --- a/crates/enc-ffmpeg/src/mux/segmented_stream.rs +++ b/crates/enc-ffmpeg/src/mux/segmented_stream.rs @@ -465,7 +465,15 @@ impl SegmentedVideoEncoder { pub fn queue_frame( &mut self, - frame: frame::Video, + mut frame: frame::Video, + timestamp: Duration, + ) -> Result<(), QueueFrameError> { + self.queue_frame_reusable(&mut frame, timestamp) + } + + pub fn queue_frame_reusable( + &mut self, + frame: &mut frame::Video, timestamp: Duration, ) -> Result<(), QueueFrameError> { // Encode with the frame's real capture-derived timestamp. The encoder @@ -473,7 +481,7 @@ impl SegmentedVideoEncoder { // stream restarts, dropped frames) stay in the timeline instead of // compressing it and drifting video ahead of audio. self.encoder - .queue_frame_to(frame, timestamp, &mut self.mux)?; + .queue_frame_to_reusable(frame, timestamp, &mut self.mux)?; self.mux.frame_queued(); diff --git a/crates/enc-ffmpeg/src/video/h264.rs b/crates/enc-ffmpeg/src/video/h264.rs index cbe05805f32..88489b83ae7 100644 --- a/crates/enc-ffmpeg/src/video/h264.rs +++ b/crates/enc-ffmpeg/src/video/h264.rs @@ -952,6 +952,10 @@ impl H264Encoder { let converted = converted_frame.get_or_insert_with(|| { frame::Video::new(self.output_format, self.output_width, self.output_height) }); + if unsafe { ffmpeg::ffi::av_frame_make_writable(converted.as_mut_ptr()) } < 0 { + *converted = + frame::Video::new(self.output_format, self.output_width, self.output_height); + } super::convert_or_scale(frame, converted, converter) .map_err(QueueFrameError::Converter)?; converted.set_pts(pts); @@ -972,6 +976,15 @@ impl H264Encoder { mut frame: frame::Video, timestamp: Duration, output: &mut format::context::Output, + ) -> Result<(), QueueFrameError> { + self.queue_preconverted_frame_reusable(&mut frame, timestamp, output) + } + + pub fn queue_preconverted_frame_reusable( + &mut self, + frame: &mut frame::Video, + timestamp: Duration, + output: &mut format::context::Output, ) -> Result<(), QueueFrameError> { trace!( "Encoding pre-converted frame: format={:?}, size={}x{}, expected={:?} {}x{}", @@ -983,11 +996,40 @@ impl H264Encoder { self.output_height ); + self.base.update_pts(frame, timestamp, &mut self.encoder); + self.base - .update_pts(&mut frame, timestamp, &mut self.encoder); + .send_frame(frame, output, &mut self.encoder) + .map_err(QueueFrameError::Encode)?; + + Ok(()) + } + + 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(); + if unsafe { ffmpeg::ffi::av_frame_make_writable(converted.as_mut_ptr()) } < 0 { + *converted = + frame::Video::new(self.output_format, self.output_width, self.output_height); + } + super::convert_or_scale(frame, converted, converter) + .map_err(QueueFrameError::Converter)?; + converted.set_pts(pts); + converted as &frame::Video + } else { + frame as &frame::Video + }; self.base - .send_frame(&frame, output, &mut self.encoder) + .send_frame(frame_to_send, output, &mut self.encoder) .map_err(QueueFrameError::Encode)?; Ok(()) diff --git a/crates/enc-ffmpeg/src/video/h264_packet.rs b/crates/enc-ffmpeg/src/video/h264_packet.rs index dd0fe0e8a34..b270c228dd0 100644 --- a/crates/enc-ffmpeg/src/video/h264_packet.rs +++ b/crates/enc-ffmpeg/src/video/h264_packet.rs @@ -172,6 +172,10 @@ impl H264PacketEncoder { let converted = converted_frame.get_or_insert_with(|| { frame::Video::new(self.output_format, self.output_width, self.output_height) }); + if unsafe { ffmpeg::ffi::av_frame_make_writable(converted.as_mut_ptr()) } < 0 { + *converted = + frame::Video::new(self.output_format, self.output_width, self.output_height); + } super::convert_or_scale(frame, converted, converter) .map_err(EncodePacketError::Converter)?; converted.set_pts(pts); diff --git a/crates/enc-ffmpeg/src/video/packed_yuv.rs b/crates/enc-ffmpeg/src/video/packed_yuv.rs index 2259054b307..cfd467044b8 100644 --- a/crates/enc-ffmpeg/src/video/packed_yuv.rs +++ b/crates/enc-ffmpeg/src/video/packed_yuv.rs @@ -834,6 +834,93 @@ mod tests { assert!(!packets.is_empty()); } + #[test] + fn reusable_queue_and_encode_preserve_plane_backing_and_detach_when_retained() { + ffmpeg::init().unwrap(); + if ffmpeg::codec::encoder::find_by_name("libx264").is_none() { + return; + } + + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("reusable.mp4"); + let mut output = ffmpeg::format::output_as(&path, "mp4").unwrap(); + let info = cap_media_info::VideoInfo::from_raw_ffmpeg(Pixel::BGRA, 64, 48, 30); + let mut queue_enc = super::super::h264::H264Encoder::builder(info) + .with_preset(super::super::h264::H264Preset::Ultrafast) + .with_crf(18) + .with_encoder_priority_override(&["libx264"]) + .build(&mut output) + .unwrap(); + + let mut frame = frame::Video::new(Pixel::BGRA, 64, 48); + frame.data_mut(0).fill(0x24); + let src_ptr = frame.data(0).as_ptr(); + let mut sink = crate::PacketQueue::new(ffmpeg::Rational(1, 90_000)); + for idx in 0..4u64 { + queue_enc + .queue_frame_to_reusable( + &mut frame, + std::time::Duration::from_millis(idx * 33), + &mut sink, + ) + .unwrap(); + assert_eq!(frame.data(0).as_ptr(), src_ptr); + } + queue_enc.flush_to(&mut sink).unwrap(); + assert!(!sink.take().is_empty()); + + let mut pkt_enc = super::super::h264::H264Encoder::builder(info) + .with_preset(super::super::h264::H264Preset::Ultrafast) + .with_crf(18) + .with_encoder_priority_override(&["libx264"]) + .build_standalone() + .unwrap(); + let mut converted: Option = None; + let mut packets = Vec::new(); + pkt_enc + .encode_frame_reusable( + &mut frame, + &mut converted, + std::time::Duration::ZERO, + |pkt| { + packets.push(pkt.data); + Ok(()) + }, + ) + .unwrap(); + + let mut retained = frame::Video::empty(); + let old_y_byte = { + let conv = converted.as_ref().unwrap(); + assert_eq!( + unsafe { ffmpeg::ffi::av_frame_ref(retained.as_mut_ptr(), conv.as_ptr()) }, + 0 + ); + retained.data(0)[0] + }; + + frame.data_mut(0).fill(0xee); + pkt_enc + .encode_frame_reusable( + &mut frame, + &mut converted, + std::time::Duration::from_millis(33), + |pkt| { + packets.push(pkt.data); + Ok(()) + }, + ) + .unwrap(); + assert_eq!(retained.data(0)[0], old_y_byte); + pkt_enc + .flush(|pkt| { + packets.push(pkt.data); + Ok(()) + }) + .unwrap(); + assert!(!packets.is_empty()); + } + #[test] #[ignore = "release timing bench, run with --ignored --nocapture"] fn bench_packed_yuv_conversion() { diff --git a/crates/recording/src/output_pipeline/macos.rs b/crates/recording/src/output_pipeline/macos.rs index 53e73bca231..66393a6c881 100644 --- a/crates/recording/src/output_pipeline/macos.rs +++ b/crates/recording/src/output_pipeline/macos.rs @@ -84,6 +84,14 @@ fn get_mp4_audio_finish_timeout(instant_mode: bool) -> Duration { type SharedFatalError = Arc>>; +#[derive(Clone)] +enum EncoderLoopStep { + Continue, + Break, + FinishOk, + Fatal(String), +} + fn set_fatal_error(fatal_error: &SharedFatalError, message: String) { if let Ok(mut slot) = fatal_error.lock() { if slot.is_none() { @@ -423,116 +431,138 @@ impl Muxer for AVFoundationMp4Muxer { let mut disk_monitor = DiskSpaceMonitor::new(); while let Ok(Some(msg)) = video_rx.recv() { - if let Some(ref depth) = channel_depth { - ChannelPressureTracker::on_recv(depth); - } - if fatal_error_message(&video_fatal_error).is_some() { - break; - } + let step = cidre::objc::ar_pool(|| { + let msg = msg; + if let Some(ref depth) = channel_depth { + ChannelPressureTracker::on_recv(depth); + } + if fatal_error_message(&video_fatal_error).is_some() { + return EncoderLoopStep::Break; + } - // All modes, not just instant: if the disk fills, the - // AVAssetWriter dies asynchronously mid-write and the - // non-fragmented output loses its moov (unrecoverable). - // Stopping while the writer is alive lets finish() write - // the moov and preserves the recording up to this point. - // DiskSpaceMonitor carries the platform-wide thresholds - // (warn 200MB / stop 50MB) and emits DiskSpaceLow / - // DiskSpaceExhausted so the user sees the real cause. - if let DiskSpacePollResult::Exhausted { bytes_remaining } = - disk_monitor.poll(&disk_check_path, &video_health_tx) - { - let message = format!( - "Disk space exhausted ({}MB left), stopping recording to preserve output", - bytes_remaining / (1024 * 1024) - ); - set_fatal_error(&video_fatal_error, message.clone()); - return Err(anyhow!(message)); - } + // All modes, not just instant: if the disk fills, the + // AVAssetWriter dies asynchronously mid-write and the + // non-fragmented output loses its moov (unrecoverable). + // Stopping while the writer is alive lets finish() write + // the moov and preserves the recording up to this point. + // DiskSpaceMonitor carries the platform-wide thresholds + // (warn 200MB / stop 50MB) and emits DiskSpaceLow / + // DiskSpaceExhausted so the user sees the real cause. + if let DiskSpacePollResult::Exhausted { bytes_remaining } = + disk_monitor.poll(&disk_check_path, &video_health_tx) + { + let message = format!( + "Disk space exhausted ({}MB left), stopping recording to preserve output", + bytes_remaining / (1024 * 1024) + ); + set_fatal_error(&video_fatal_error, message.clone()); + return EncoderLoopStep::Fatal(message); + } - match msg { - VideoFrameMessage::Frame(sample_buf, timestamp) => { - let mut retry_count = 0; - const MAX_RETRIES: u32 = 100; + match msg { + VideoFrameMessage::Frame(sample_buf, timestamp) => { + 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(_) => { + loop { + let mut attempt_out = None; + cidre::objc::ar_pool(|| { + attempt_out = Some(match encoder_clone.lock() { + Ok(mut encoder) => encoder + .queue_video_frame(sample_buf.clone(), timestamp) + .map_err(Some), + Err(_) => Err(None), + }); + }); + + let queue_result = match attempt_out.expect("ar_pool executed") + { + Ok(()) => Ok(()), + Err(Some(e)) => Err(e), + Err(None) => { let message = "MP4 encoder mutex poisoned".to_string(); set_fatal_error(&video_fatal_error, message.clone()); - return Err(anyhow!(message)); + return EncoderLoopStep::Fatal(message); } }; - 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; - if encoder_busy_count <= 5 || encoder_busy_count.is_multiple_of(100) { - debug!( - encoder_busy_count, - "MP4 encoder busy, frame queued after max retries" - ); + + match queue_result { + Ok(()) => break, + Err(QueueFrameError::NotReadyForMore) => { + retry_count += 1; + if retry_count >= MAX_RETRIES { + encoder_busy_count += 1; + 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; } + 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 EncoderLoopStep::Fatal(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 EncoderLoopStep::Fatal(message); + } + Err(QueueFrameError::Finished) => { + return EncoderLoopStep::FinishOk; + } + Err(e) => { + warn!("Failed to encode video frame: {e}"); 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(()), - Err(e) => { - warn!("Failed to encode video frame: {e}"); - break; } } - } - video_count_thread.fetch_add(1, std::sync::atomic::Ordering::Relaxed); - } - VideoFrameMessage::Pause => { - match encoder_clone.lock() { + video_count_thread + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } + VideoFrameMessage::Pause => match encoder_clone.lock() { Ok(mut encoder) => encoder.pause(), Err(_) => { let message = "MP4 encoder mutex poisoned".to_string(); set_fatal_error(&video_fatal_error, message.clone()); - return Err(anyhow!(message)); + return EncoderLoopStep::Fatal(message); } - } - } - VideoFrameMessage::Resume => { - match encoder_clone.lock() { + }, + VideoFrameMessage::Resume => match encoder_clone.lock() { Ok(mut encoder) => encoder.resume(), Err(_) => { let message = "MP4 encoder mutex poisoned".to_string(); set_fatal_error(&video_fatal_error, message.clone()); - return Err(anyhow!(message)); + return EncoderLoopStep::Fatal(message); } - } + }, } + + EncoderLoopStep::Continue + }); + + match step { + EncoderLoopStep::Continue => {} + EncoderLoopStep::Break => break, + EncoderLoopStep::FinishOk => return Ok(()), + EncoderLoopStep::Fatal(message) => return Err(anyhow!(message)), } } @@ -573,76 +603,106 @@ impl Muxer for AVFoundationMp4Muxer { let mut encoder_busy_count = 0u64; while let Ok(Some(msg)) = audio_rx.recv() { - if let Some(ref depth) = audio_channel_depth_thread { - ChannelPressureTracker::on_recv(depth); - } - if fatal_error_message(&audio_fatal_error).is_some() { - break; - } - - match msg { - AudioFrameMessage::Frame(frame, timestamp) => { - let mut retry_count = 0; - const MAX_RETRIES: u32 = 50; + let step = cidre::objc::ar_pool(|| { + let msg = msg; + if let Some(ref depth) = audio_channel_depth_thread { + ChannelPressureTracker::on_recv(depth); + } + if fatal_error_message(&audio_fatal_error).is_some() { + return EncoderLoopStep::Break; + } - loop { - let queue_result = { - let mut encoder = match encoder_clone.lock() { - Ok(e) => e, - Err(_) => { - let message = - "MP4 audio encoder mutex poisoned".to_string(); + match msg { + AudioFrameMessage::Frame(frame, timestamp) => { + let mut retry_count = 0; + const MAX_RETRIES: u32 = 50; + + loop { + let mut attempt_out = None; + cidre::objc::ar_pool(|| { + attempt_out = Some(match encoder_clone.lock() { + Ok(mut encoder) => encoder + .queue_audio_frame(&frame, timestamp) + .map_err(Some), + Err(_) => Err(None), + }); + }); + + let queue_result = + match attempt_out.expect("ar_pool executed") { + Ok(()) => Ok(()), + Err(Some(e)) => Err(e), + Err(None) => { + let message = + "MP4 audio encoder mutex poisoned" + .to_string(); + set_fatal_error( + &audio_fatal_error, + message.clone(), + ); + return EncoderLoopStep::Fatal(message); + } + }; + + match queue_result { + Ok(()) => break, + Err(QueueFrameError::NotReadyForMore) => { + retry_count += 1; + if retry_count >= MAX_RETRIES { + encoder_busy_count += 1; + break; + } + std::thread::sleep(Duration::from_micros(500)); + } + Err(QueueFrameError::WriterFailed(err)) => { + let total = audio_count_thread + .load(std::sync::atomic::Ordering::Relaxed); + let message = format!( + "Failed to encode audio frame: WriterFailed/{err:?} \ + (frame #{total}, ts={timestamp:?})" + ); set_fatal_error( &audio_fatal_error, message.clone(), ); - return Err(anyhow!(message)); + return EncoderLoopStep::Fatal(message); } - }; - encoder.queue_audio_frame(&frame, timestamp) - }; - - match queue_result { - Ok(()) => break, - Err(QueueFrameError::NotReadyForMore) => { - retry_count += 1; - if retry_count >= MAX_RETRIES { - encoder_busy_count += 1; + Err(QueueFrameError::Failed) => { + let total = audio_count_thread + .load(std::sync::atomic::Ordering::Relaxed); + let message = format!( + "Failed to encode audio frame: Failed \ + (frame #{total}, ts={timestamp:?})" + ); + set_fatal_error( + &audio_fatal_error, + message.clone(), + ); + return EncoderLoopStep::Fatal(message); + } + Err(QueueFrameError::Finished) => { + return EncoderLoopStep::FinishOk; + } + Err(e) => { + warn!("Failed to encode audio frame: {e}"); break; } - std::thread::sleep(Duration::from_micros(500)); - } - Err(QueueFrameError::WriterFailed(err)) => { - let total = audio_count_thread - .load(std::sync::atomic::Ordering::Relaxed); - let message = format!( - "Failed to encode audio frame: WriterFailed/{err:?} \ - (frame #{total}, ts={timestamp:?})" - ); - set_fatal_error(&audio_fatal_error, message.clone()); - return Err(anyhow!(message)); - } - Err(QueueFrameError::Failed) => { - let total = audio_count_thread - .load(std::sync::atomic::Ordering::Relaxed); - let message = format!( - "Failed to encode audio frame: Failed \ - (frame #{total}, ts={timestamp:?})" - ); - set_fatal_error(&audio_fatal_error, message.clone()); - return Err(anyhow!(message)); - } - Err(QueueFrameError::Finished) => return Ok(()), - Err(e) => { - warn!("Failed to encode audio frame: {e}"); - break; } } - } - audio_count_thread - .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + audio_count_thread + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } } + + EncoderLoopStep::Continue + }); + + match step { + EncoderLoopStep::Continue => {} + EncoderLoopStep::Break => break, + EncoderLoopStep::FinishOk => return Ok(()), + EncoderLoopStep::Fatal(message) => return Err(anyhow!(message)), } } @@ -1046,103 +1106,125 @@ impl Muxer for AVFoundationCameraMuxer { let mut disk_monitor = DiskSpaceMonitor::new(); while let Ok(Some(msg)) = video_rx.recv() { - if fatal_error_message(&video_fatal_error).is_some() { - break; - } + let step = cidre::objc::ar_pool(|| { + let msg = msg; + if fatal_error_message(&video_fatal_error).is_some() { + return EncoderLoopStep::Break; + } - // Same rationale as the screen writer above: stop while - // the AVAssetWriter is still alive so the camera file - // keeps its moov instead of dying on a failed async write - // (finish() runs because the thread exits cleanly with an - // error rather than timing out). - if let DiskSpacePollResult::Exhausted { bytes_remaining } = - disk_monitor.poll(&disk_check_path, &video_health_tx) - { - let message = format!( - "Disk space exhausted ({}MB left), stopping camera recording to preserve output", - bytes_remaining / (1024 * 1024) - ); - set_fatal_error(&video_fatal_error, message.clone()); - return Err(anyhow!(message)); - } + // Same rationale as the screen writer above: stop while + // the AVAssetWriter is still alive so the camera file + // keeps its moov instead of dying on a failed async write + // (finish() runs because the thread exits cleanly with an + // error rather than timing out). + if let DiskSpacePollResult::Exhausted { bytes_remaining } = + disk_monitor.poll(&disk_check_path, &video_health_tx) + { + let message = format!( + "Disk space exhausted ({}MB left), stopping camera recording to preserve output", + bytes_remaining / (1024 * 1024) + ); + set_fatal_error(&video_fatal_error, message.clone()); + return EncoderLoopStep::Fatal(message); + } - match msg { - CameraFrameMessage::Frame(sample_buf, timestamp) => { - let mut retry_count = 0; - const MAX_RETRIES: u32 = 100; + match msg { + CameraFrameMessage::Frame(sample_buf, timestamp) => { + 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(_) => { + loop { + let mut attempt_out = None; + cidre::objc::ar_pool(|| { + attempt_out = Some(match encoder_clone.lock() { + Ok(mut encoder) => encoder + .queue_video_frame(sample_buf.clone(), timestamp) + .map_err(Some), + Err(_) => Err(None), + }); + }); + + let queue_result = match attempt_out.expect("ar_pool executed") + { + Ok(()) => Ok(()), + Err(Some(e)) => Err(e), + Err(None) => { let message = "Camera MP4 encoder mutex poisoned".to_string(); set_fatal_error(&video_fatal_error, message.clone()); - return Err(anyhow!(message)); + return EncoderLoopStep::Fatal(message); } }; - 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; - if encoder_busy_count <= 5 || encoder_busy_count.is_multiple_of(100) { - debug!( - encoder_busy_count, - "Camera MP4 encoder busy, frame queued after max retries" - ); + + match queue_result { + Ok(()) => break, + Err(QueueFrameError::NotReadyForMore) => { + retry_count += 1; + if retry_count >= MAX_RETRIES { + encoder_busy_count += 1; + if encoder_busy_count <= 5 + || encoder_busy_count.is_multiple_of(100) + { + debug!( + encoder_busy_count, + "Camera MP4 encoder busy, frame queued after max retries" + ); + } + break; } + std::thread::sleep(Duration::from_micros(500)); + } + Err(QueueFrameError::WriterFailed(err)) => { + let message = format!( + "Failed to encode camera frame: WriterFailed/{err:?}" + ); + set_fatal_error(&video_fatal_error, message.clone()); + return EncoderLoopStep::Fatal(message); + } + Err(QueueFrameError::Failed) => { + let message = + "Failed to encode camera frame: Failed".to_string(); + set_fatal_error(&video_fatal_error, message.clone()); + return EncoderLoopStep::Fatal(message); + } + Err(QueueFrameError::Finished) => { + return EncoderLoopStep::FinishOk; + } + Err(e) => { + warn!("Failed to encode camera frame: {e}"); break; } - std::thread::sleep(Duration::from_micros(500)); - } - Err(QueueFrameError::WriterFailed(err)) => { - let message = format!( - "Failed to encode camera frame: WriterFailed/{err:?}" - ); - set_fatal_error(&video_fatal_error, message.clone()); - return Err(anyhow!(message)); - } - Err(QueueFrameError::Failed) => { - let message = "Failed to encode camera frame: Failed".to_string(); - set_fatal_error(&video_fatal_error, message.clone()); - return Err(anyhow!(message)); - } - Err(QueueFrameError::Finished) => return Ok(()), - Err(e) => { - warn!("Failed to encode camera frame: {e}"); - break; } } - } - total_frames += 1; - } - CameraFrameMessage::Pause => { - match encoder_clone.lock() { + total_frames += 1; + } + CameraFrameMessage::Pause => match encoder_clone.lock() { Ok(mut encoder) => encoder.pause(), Err(_) => { let message = "Camera MP4 encoder mutex poisoned".to_string(); set_fatal_error(&video_fatal_error, message.clone()); - return Err(anyhow!(message)); + return EncoderLoopStep::Fatal(message); } - } - } - CameraFrameMessage::Resume => { - match encoder_clone.lock() { + }, + CameraFrameMessage::Resume => match encoder_clone.lock() { Ok(mut encoder) => encoder.resume(), Err(_) => { let message = "Camera MP4 encoder mutex poisoned".to_string(); set_fatal_error(&video_fatal_error, message.clone()); - return Err(anyhow!(message)); + return EncoderLoopStep::Fatal(message); } - } + }, } + + EncoderLoopStep::Continue + }); + + match step { + EncoderLoopStep::Continue => {} + EncoderLoopStep::Break => break, + EncoderLoopStep::FinishOk => return Ok(()), + EncoderLoopStep::Fatal(message) => return Err(anyhow!(message)), } } @@ -1190,67 +1272,101 @@ impl Muxer for AVFoundationCameraMuxer { let mut encoder_busy_count = 0u64; while let Ok(Some(msg)) = audio_rx.recv() { - if let Some(ref depth) = audio_channel_depth_thread { - ChannelPressureTracker::on_recv(depth); - } - if fatal_error_message(&audio_fatal_error).is_some() { - break; - } - - match msg { - AudioFrameMessage::Frame(frame, timestamp) => { - let mut retry_count = 0; - const MAX_RETRIES: u32 = 50; + let step = cidre::objc::ar_pool(|| { + let msg = msg; + if let Some(ref depth) = audio_channel_depth_thread { + ChannelPressureTracker::on_recv(depth); + } + if fatal_error_message(&audio_fatal_error).is_some() { + return EncoderLoopStep::Break; + } - loop { - let queue_result = { - let mut encoder = match encoder_clone.lock() { - Ok(e) => e, - Err(_) => { - let message = "Camera MP4 audio encoder mutex poisoned".to_string(); - set_fatal_error(&audio_fatal_error, message.clone()); - return Err(anyhow!(message)); + match msg { + AudioFrameMessage::Frame(frame, timestamp) => { + let mut retry_count = 0; + const MAX_RETRIES: u32 = 50; + + loop { + let mut attempt_out = None; + cidre::objc::ar_pool(|| { + attempt_out = Some(match encoder_clone.lock() { + Ok(mut encoder) => encoder + .queue_audio_frame(&frame, timestamp) + .map_err(Some), + Err(_) => Err(None), + }); + }); + + let queue_result = + match attempt_out.expect("ar_pool executed") { + Ok(()) => Ok(()), + Err(Some(e)) => Err(e), + Err(None) => { + let message = + "Camera MP4 audio encoder mutex poisoned" + .to_string(); + set_fatal_error( + &audio_fatal_error, + message.clone(), + ); + return EncoderLoopStep::Fatal(message); + } + }; + + match queue_result { + Ok(()) => break, + Err(QueueFrameError::NotReadyForMore) => { + retry_count += 1; + if retry_count >= MAX_RETRIES { + encoder_busy_count += 1; + break; + } + std::thread::sleep(Duration::from_micros(500)); } - }; - encoder.queue_audio_frame(&frame, timestamp) - }; - - match queue_result { - Ok(()) => break, - Err(QueueFrameError::NotReadyForMore) => { - retry_count += 1; - if retry_count >= MAX_RETRIES { - encoder_busy_count += 1; + Err(QueueFrameError::WriterFailed(err)) => { + let message = format!( + "Failed to encode camera audio frame: WriterFailed/{err:?} \ + (frame #{total_frames}, ts={timestamp:?})" + ); + set_fatal_error( + &audio_fatal_error, + message.clone(), + ); + return EncoderLoopStep::Fatal(message); + } + Err(QueueFrameError::Failed) => { + let message = format!( + "Failed to encode camera audio frame: Failed \ + (frame #{total_frames}, ts={timestamp:?})" + ); + set_fatal_error( + &audio_fatal_error, + message.clone(), + ); + return EncoderLoopStep::Fatal(message); + } + Err(QueueFrameError::Finished) => { + return EncoderLoopStep::FinishOk; + } + Err(e) => { + warn!("Failed to encode camera audio frame: {e}"); break; } - std::thread::sleep(Duration::from_micros(500)); - } - Err(QueueFrameError::WriterFailed(err)) => { - let message = format!( - "Failed to encode camera audio frame: WriterFailed/{err:?} \ - (frame #{total_frames}, ts={timestamp:?})" - ); - set_fatal_error(&audio_fatal_error, message.clone()); - return Err(anyhow!(message)); - } - Err(QueueFrameError::Failed) => { - let message = format!( - "Failed to encode camera audio frame: Failed \ - (frame #{total_frames}, ts={timestamp:?})" - ); - set_fatal_error(&audio_fatal_error, message.clone()); - return Err(anyhow!(message)); - } - Err(QueueFrameError::Finished) => return Ok(()), - Err(e) => { - warn!("Failed to encode camera audio frame: {e}"); - break; } } - } - total_frames += 1; + total_frames += 1; + } } + + EncoderLoopStep::Continue + }); + + match step { + EncoderLoopStep::Continue => {} + EncoderLoopStep::Break => break, + EncoderLoopStep::FinishOk => return Ok(()), + EncoderLoopStep::Fatal(message) => return Err(anyhow!(message)), } } diff --git a/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs b/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs index cd36e3bff3b..5b97636856f 100644 --- a/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs +++ b/crates/recording/src/output_pipeline/macos_fragmented_m4s.rs @@ -346,114 +346,131 @@ 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; - } + let mut fatal_error: Option = None; + cidre::objc::ar_pool(|| { + let sample_buf = sample_buf; + if matches!( + disk_monitor.poll(&base_path, &health_tx), + super::core::DiskSpacePollResult::Exhausted { .. } + | super::core::DiskSpacePollResult::Stopped + ) { + disk_exhausted = true; + } + if disk_exhausted { + return; + } - 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 encoder_clone.lock() { - Ok(mut encoder) => { - if let Err(e) = encoder.queue_frame(owned_frame, timestamp) { - warn!("Failed to encode frame: {e}"); + 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_reusable(frame, timestamp) + { + warn!("Failed to encode frame: {e}"); + } + } + Err(_) => { + error!("Encoder mutex poisoned - encoder thread likely panicked, stopping"); + fatal_error = Some( + "Encoder mutex poisoned - all subsequent frames would be lost" + .to_string(), + ); + return; } } - 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 {slow_threshold_ms}ms threshold", + ); + } } } + 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"); + fatal_error = Some( + "Encoder mutex poisoned - all subsequent frames would be lost" + .to_string(), + ); + return; + } } } - } - 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; + }); + + if let Some(message) = fatal_error { + return Err(anyhow!(message)); + } } if total_frames > 0 { @@ -735,71 +752,85 @@ 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; - } + let mut fatal_error: Option = None; + cidre::objc::ar_pool(|| { + let sample_buf = sample_buf; + if matches!( + disk_monitor.poll(&base_path, &health_tx), + super::core::DiskSpacePollResult::Exhausted { .. } + | super::core::DiskSpacePollResult::Stopped + ) { + disk_exhausted = true; + } + if disk_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(); - 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"); + fatal_error = Some( + "Camera encoder mutex poisoned - all subsequent frames would be lost" + .to_string(), + ); + return; } } - 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 {slow_threshold_ms}ms threshold", + ); + } } } + Err(e) => { + warn!("Failed to convert camera frame: {e:?}"); + } } - Err(e) => { - warn!("Failed to convert camera frame: {e:?}"); - } - } - total_frames += 1; + total_frames += 1; + }); + + if let Some(message) = fatal_error { + return Err(anyhow!(message)); + } } 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..b7de4ebe451 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(frame) = self.frame.as_mut() + && 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, @@ -64,6 +69,7 @@ impl FramePool { self.frame.as_mut().expect("frame initialized above") } + #[allow(dead_code)] pub(crate) fn take_frame(&mut self) -> ffmpeg::frame::Video { self.frame.take().unwrap_or_else(|| { ffmpeg::frame::Video::new(self.pixel_format, self.width, self.height) @@ -248,4 +254,30 @@ mod tests { copy_plane_data(&plane(8, 8), &mut dest, usize::MAX / 2, 8, 8, 8); copy_plane_data(&plane(8, 8), &mut dest, usize::MAX / 2, 4, 8, 8); } + + #[test] + fn frame_pool_reuses_backing_buffer_and_detaches_when_shared() { + ffmpeg::init().unwrap(); + let mut pool = super::FramePool::new(ffmpeg::format::Pixel::BGRA, 64, 64); + + let ptr1 = { + let f = pool.get_frame(); + f.data_mut(0).fill(0x42); + f.data(0).as_ptr() + }; + let ptr2 = pool.get_frame().data(0).as_ptr(); + assert_eq!(ptr1, ptr2); + + let mut retained = ffmpeg::frame::Video::empty(); + { + let f = pool.get_frame(); + assert_eq!( + unsafe { ffmpeg::ffi::av_frame_ref(retained.as_mut_ptr(), f.as_ptr()) }, + 0 + ); + } + let next = pool.get_frame(); + next.data_mut(0).fill(0x99); + assert!(retained.data(0)[..64].iter().all(|&b| b == 0x42)); + } } diff --git a/crates/recording/src/output_pipeline/oop_fragmented_m4s.rs b/crates/recording/src/output_pipeline/oop_fragmented_m4s.rs index 9afbb643412..3da08b1ce81 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,143 +390,148 @@ 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(|| { + let sample_buf = sample_buf; + 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 subprocess_fatal || subprocess.is_exhausted() { - continue; - } + if disk_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 - ); + if subprocess_fatal || subprocess.is_exhausted() { + return; } - } - 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) - }, - ); - 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, - "OOP encode_frame 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 encode_result { - Ok(()) => { - tracker.on_frame(timestamp); - } - Err(EncodePacketError::Converter(e)) => { - warn!("OOP encoder converter failure: {e}"); - } - Err(EncodePacketError::Encode(e)) => { - if subprocess.is_exhausted() { - error!( - respawn_attempts = subprocess.respawn_attempts(), - max_consecutive_fast_failures = - subprocess.max_consecutive_fast_failures(), - "OOP subprocess respawn budget exhausted; draining further frames to disk-only fragments" - ); - subprocess_fatal = true; - } else if keyframe_emitted - && encoder_rebuild_attempts - < MAX_ENCODER_REBUILD_ATTEMPTS + 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; + if slow_encode_count <= 5 + || slow_encode_count.is_multiple_of(100) { - encoder_rebuild_attempts += 1; - warn!( - attempt = encoder_rebuild_attempts, - error = ?e, - "OOP encoder mid-stream error; attempting same-codec rebuild" + debug!( + elapsed_ms = encode_elapsed_ms, + count = slow_encode_count, + "OOP encode_frame_reusable exceeded {slow_threshold_ms}ms threshold", ); - let _ = encoder.flush(|pkt| { - if pkt.is_keyframe { - keyframe_emitted = true; - } - dispatch_packet(&mut subprocess, pkt) - }); - match rebuild_builder.clone().build_standalone() { - Ok(new_enc) => { - let old_name = encoder.codec_name().to_string(); - let new_name = new_enc.codec_name().to_string(); - info!( - old_codec = %old_name, - new_codec = %new_name, - attempt = encoder_rebuild_attempts, - "OOP encoder rebuilt mid-stream" - ); - encoder = new_enc; - health_tx.emit( - super::core::PipelineHealthEvent::EncoderRebuilt { - backend: new_name, - attempt: encoder_rebuild_attempts, - }, - ); - } - Err(rebuild_err) => { - error!( - error = ?rebuild_err, - "OOP encoder rebuild failed" - ); + } + } + + match encode_result { + Ok(()) => { + tracker.on_frame(timestamp); + } + Err(EncodePacketError::Converter(e)) => { + warn!("OOP encoder converter failure: {e}"); + } + Err(EncodePacketError::Encode(e)) => { + if subprocess.is_exhausted() { + error!( + respawn_attempts = subprocess.respawn_attempts(), + max_consecutive_fast_failures = + subprocess.max_consecutive_fast_failures(), + "OOP subprocess respawn budget exhausted; draining further frames to disk-only fragments" + ); + subprocess_fatal = true; + } else if keyframe_emitted + && encoder_rebuild_attempts + < MAX_ENCODER_REBUILD_ATTEMPTS + { + encoder_rebuild_attempts += 1; + warn!( + attempt = encoder_rebuild_attempts, + error = ?e, + "OOP encoder mid-stream error; attempting same-codec rebuild" + ); + let _ = encoder.flush(|pkt| { + if pkt.is_keyframe { + keyframe_emitted = true; + } + dispatch_packet(&mut subprocess, pkt) + }); + match rebuild_builder.clone().build_standalone() { + Ok(new_enc) => { + let old_name = encoder.codec_name().to_string(); + let new_name = new_enc.codec_name().to_string(); + info!( + old_codec = %old_name, + new_codec = %new_name, + attempt = encoder_rebuild_attempts, + "OOP encoder rebuilt mid-stream" + ); + encoder = new_enc; + converted_frame = None; + health_tx.emit( + super::core::PipelineHealthEvent::EncoderRebuilt { + backend: new_name, + attempt: encoder_rebuild_attempts, + }, + ); + } + Err(rebuild_err) => { + error!( + error = ?rebuild_err, + "OOP encoder rebuild failed" + ); + } } + } else { + warn!( + "OOP encoder transient ffmpeg error (subprocess still active)" + ); } - } else { - warn!( - "OOP encoder transient ffmpeg error (subprocess still active)" - ); } } } + Err(e) => { + warn!("OOP failed to convert frame: {e:?}"); + } } - Err(e) => { - warn!("OOP failed to convert frame: {e:?}"); - } - } - total_frames += 1; + total_frames += 1; + }); } let final_ts = last_timestamp.unwrap_or(Duration::ZERO);