perf(recording): decode each Windows MJPEG camera frame once for preview and encoder

crates/recording/src/feeds/camera.rs

@@ -1142,6 +1142,7 @@ async fn setup_camera(
 
     let (ready_tx, ready_rx) = oneshot::channel();
     let mut ready_signal = Some(ready_tx);
+    let decode_budget = crate::output_pipeline::SharedDecodeBudget::default();
 
     let capture_handle = camera
         .start_capturing(format.clone(), move |frame| {
@@ -1152,12 +1153,22 @@ async fn setup_camera(
                 cap_timestamp::PerformanceCounterTimestamp::new(frame.native().perf_counter),
             );
 
-            if native_ingress
+            let wants_native = native_ingress
                 .senders
                 .load(std::sync::atomic::Ordering::Relaxed)
-                > 0
-                && let Ok(bytes) = frame.native().bytes()
-            {
+                > 0;
+            // Until the ready signal fires the first frame must still be
+            // converted to derive VideoInfo; afterwards skip the full-frame
+            // copy entirely when nothing consumes ffmpeg frames.
+            let wants_ffmpeg = ready_signal.is_some()
+                || ffmpeg_sender_count.load(std::sync::atomic::Ordering::Acquire) > 0;
+
+            let shared_decode = (wants_native
+                && wants_ffmpeg
+                && frame.native().pixel_format == cap_camera_windows::PixelFormat::MJPEG)
+                .then(|| frame.as_ffmpeg());
+
+            if wants_native && let Ok(bytes) = frame.native().bytes() {
                 use cap_mediafoundation_utils::IMFMediaBufferExt;
                 use windows::Win32::Media::MediaFoundation::MFCreateMemoryBuffer;
 
@@ -1187,22 +1198,21 @@ async fn setup_camera(
                                 height: frame.native().height as u32,
                                 is_bottom_up: frame.native().is_bottom_up,
                                 timestamp,
+                                decoded: shared_decode
+                                    .as_ref()
+                                    .and_then(|decoded| decoded.as_ref().ok())
+                                    .and_then(|decoded| decode_budget.share(decoded)),
                             },
                         );
                     }
                 }
             }
 
-            // Until the ready signal fires the first frame must still be
-            // converted to derive VideoInfo; afterwards skip the full-frame
-            // copy entirely when nothing consumes ffmpeg frames.
-            if ready_signal.is_none()
-                && ffmpeg_sender_count.load(std::sync::atomic::Ordering::Acquire) == 0
-            {
+            if !wants_ffmpeg {
                 return;
             }
 
-            let Ok(mut ff_frame) = frame.as_ffmpeg() else {
+            let Ok(mut ff_frame) = shared_decode.unwrap_or_else(|| frame.as_ffmpeg()) else {
                 return;
             };
 

crates/recording/src/output_pipeline/win.rs

@@ -16,7 +16,7 @@ use std::{
     path::PathBuf,
     sync::{
         Arc, Mutex,
-        atomic::AtomicBool,
+        atomic::{AtomicBool, AtomicUsize, Ordering},
         mpsc::{SyncSender, TrySendError, sync_channel},
     },
     time::Duration,
@@ -610,11 +610,48 @@ pub struct NativeCameraFrame {
     pub height: u32,
     pub is_bottom_up: bool,
     pub timestamp: Timestamp,
+    pub decoded: Option<Arc<SharedDecodedFrame>>,
 }
 
 unsafe impl Send for NativeCameraFrame {}
 unsafe impl Sync for NativeCameraFrame {}
 
+pub struct SharedDecodedFrame {
+    frame: ffmpeg::frame::Video,
+    budget: SharedDecodeBudget,
+}
+
+impl Drop for SharedDecodedFrame {
+    fn drop(&mut self) {
+        self.budget.0.fetch_sub(1, Ordering::AcqRel);
+    }
+}
+
+/// Bounds how many decoded frames the capture callback can queue for the
+/// encoder, so a stalled encoder backs up on compressed MJPEG rather than on
+/// decoded frames. Frames over the limit are decoded by the encoder as before.
+#[derive(Clone, Default)]
+pub struct SharedDecodeBudget(Arc<AtomicUsize>);
+
+impl SharedDecodeBudget {
+    const MAX_IN_FLIGHT: usize = 8;
+
+    pub fn share(&self, frame: &ffmpeg::frame::Video) -> Option<Arc<SharedDecodedFrame>> {
+        if self.0.fetch_add(1, Ordering::AcqRel) >= Self::MAX_IN_FLIGHT {
+            self.0.fetch_sub(1, Ordering::AcqRel);
+            return None;
+        }
+
+        let mut shared = SharedDecodedFrame {
+            frame: ffmpeg::frame::Video::empty(),
+            budget: self.clone(),
+        };
+        let status =
+            unsafe { ffmpeg::ffi::av_frame_ref(shared.frame.as_mut_ptr(), frame.as_ptr()) };
+        (status >= 0).then(|| Arc::new(shared))
+    }
+}
+
 impl VideoFrame for NativeCameraFrame {
     fn timestamp(&self) -> Timestamp {
         self.timestamp
@@ -1354,6 +1391,9 @@ pub(crate) fn camera_frame_to_ffmpeg(
     use cap_mediafoundation_utils::IMFMediaBufferExt;
 
     if frame.pixel_format == cap_camera_windows::PixelFormat::MJPEG {
+        if let Some(decoded) = &frame.decoded {
+            return Ok(reference_video_frame(&decoded.frame));
+        }
         return decode_mjpeg_frame(frame, mjpeg_decoder);
     }
 
@@ -1959,3 +1999,143 @@ pub fn upload_mf_buffer_to_texture(
         })
     }
 }
+
+#[cfg(test)]
+mod shared_mjpeg_tests {
+    use super::*;
+    use cap_mediafoundation_utils::IMFMediaBufferExt;
+    use ffmpeg::format::Pixel;
+    use windows::Win32::Media::MediaFoundation::MFCreateMemoryBuffer;
+
+    const WIDTH: u32 = 64;
+    const HEIGHT: u32 = 48;
+
+    fn encode_mjpeg(seed: usize) -> Vec<u8> {
+        let codec = ffmpeg::codec::encoder::find(ffmpeg::codec::Id::MJPEG).unwrap();
+        let mut encoder = ffmpeg::codec::context::Context::new_with_codec(codec)
+            .encoder()
+            .video()
+            .unwrap();
+        encoder.set_width(WIDTH);
+        encoder.set_height(HEIGHT);
+        encoder.set_format(Pixel::YUVJ422P);
+        encoder.set_time_base((1, 30));
+        let mut encoder = encoder.open().unwrap();
+
+        let mut frame = ffmpeg::frame::Video::new(Pixel::YUVJ422P, WIDTH, HEIGHT);
+        for plane in 0..frame.planes() {
+            let stride = frame.stride(plane);
+            let (width, height) = (frame.plane_width(plane), frame.plane_height(plane));
+            for (y, row) in frame
+                .data_mut(plane)
+                .chunks_mut(stride)
+                .take(height as usize)
+                .enumerate()
+            {
+                for (x, pixel) in row.iter_mut().take(width as usize).enumerate() {
+                    *pixel = ((x * 3 + y * 5 + seed * 29 + plane * 71) & 0xff) as u8;
+                }
+            }
+        }
+        frame.set_pts(Some(0));
+
+        encoder.send_frame(&frame).unwrap();
+        encoder.send_eof().unwrap();
+        let mut packet = ffmpeg::Packet::empty();
+        encoder.receive_packet(&mut packet).unwrap();
+        packet.data().unwrap().to_vec()
+    }
+
+    fn mjpeg_camera_frame(
+        data: &[u8],
+        decoded: Option<Arc<SharedDecodedFrame>>,
+    ) -> NativeCameraFrame {
+        let buffer = unsafe { MFCreateMemoryBuffer(data.len() as u32) }.unwrap();
+        buffer.lock_for_write().unwrap().copy_from_slice(data);
+        unsafe { buffer.SetCurrentLength(data.len() as u32) }.unwrap();
+        #[allow(clippy::arc_with_non_send_sync)]
+        let buffer = Arc::new(Mutex::new(buffer));
+        NativeCameraFrame {
+            buffer,
+            pixel_format: cap_camera_windows::PixelFormat::MJPEG,
+            width: WIDTH,
+            height: HEIGHT,
+            is_bottom_up: false,
+            timestamp: Timestamp::Instant(std::time::Instant::now()),
+            decoded,
+        }
+    }
+
+    fn visible_planes(frame: &ffmpeg::frame::Video) -> Vec<Vec<u8>> {
+        (0..frame.planes())
+            .map(|plane| {
+                let width = frame.plane_width(plane) as usize;
+                frame
+                    .data(plane)
+                    .chunks(frame.stride(plane))
+                    .take(frame.plane_height(plane) as usize)
+                    .flat_map(|row| row[..width].iter().copied())
+                    .collect()
+            })
+            .collect()
+    }
+
+    #[test]
+    fn shared_mjpeg_frame_matches_encoder_decode_without_decoding_again() {
+        let mut preview_decoder = MjpegDecoder::default();
+        let mut shared_decoder = MjpegDecoder::default();
+        let mut fallback_decoder = MjpegDecoder::default();
+        let budget = SharedDecodeBudget::default();
+
+        for seed in 0..6 {
+            let data = encode_mjpeg(seed);
+            let preview = preview_decoder.decode(&data).unwrap();
+            let shared_frame = mjpeg_camera_frame(&data, budget.share(&preview));
+            let fallback_frame = mjpeg_camera_frame(&data, None);
+
+            let shared = camera_frame_to_ffmpeg(&shared_frame, &mut shared_decoder).unwrap();
+            let repeated =
+                camera_frame_to_ffmpeg(&shared_frame.clone(), &mut shared_decoder).unwrap();
+            let fallback = camera_frame_to_ffmpeg(&fallback_frame, &mut fallback_decoder).unwrap();
+
+            assert_eq!(shared.data(0).as_ptr(), preview.data(0).as_ptr());
+            assert_eq!(repeated.data(0).as_ptr(), preview.data(0).as_ptr());
+            assert_ne!(fallback.data(0).as_ptr(), preview.data(0).as_ptr());
+            assert_eq!(shared.format(), fallback.format());
+            assert_eq!(
+                (shared.width(), shared.height()),
+                (fallback.width(), fallback.height())
+            );
+            assert_eq!(shared.color_range(), fallback.color_range());
+            assert_eq!(shared.color_space(), fallback.color_space());
+            assert_eq!(shared.pts(), fallback.pts());
+            assert_eq!(visible_planes(&shared), visible_planes(&fallback));
+        }
+
+        assert_eq!(budget.0.load(Ordering::Acquire), 0);
+    }
+
+    #[test]
+    fn exhausted_decode_budget_leaves_decoding_to_the_encoder() {
+        let data = encode_mjpeg(3);
+        let preview = MjpegDecoder::default().decode(&data).unwrap();
+        let budget = SharedDecodeBudget::default();
+
+        let mut held: Vec<_> = (0..SharedDecodeBudget::MAX_IN_FLIGHT)
+            .map(|_| mjpeg_camera_frame(&data, budget.share(&preview)))
+            .collect();
+        assert!(held.iter().all(|frame| frame.decoded.is_some()));
+        let clones: Vec<_> = held.iter().map(NativeCameraFrame::clone).collect();
+
+        let over_budget = mjpeg_camera_frame(&data, budget.share(&preview));
+        assert!(over_budget.decoded.is_none());
+        let decoded = camera_frame_to_ffmpeg(&over_budget, &mut MjpegDecoder::default()).unwrap();
+        assert_ne!(decoded.data(0).as_ptr(), preview.data(0).as_ptr());
+        assert_eq!(visible_planes(&decoded), visible_planes(&preview));
+
+        held.pop();
+        assert!(budget.share(&preview).is_none());
+        drop(clones);
+        assert!(budget.share(&preview).is_some());
+    }
+}

crates/verify/tests/recording.rs

@@ -1143,6 +1143,7 @@ fn camera_frame(slot: u32, capture: u64) -> Option<NativeCameraFrame> {
         height: h as u32,
         is_bottom_up: false,
         timestamp: host_timestamp(capture),
+        decoded: None,
     })
 }