perf(camera): reduce per-frame copies and allocations across capture backends

crates/camera-directshow/src/lib.rs

@@ -8,7 +8,12 @@ use std::{
     ops::Deref,
     os::windows::ffi::OsStringExt,
     ptr::{self, null, null_mut},
-    sync::OnceLock,
+    sync::{
+        Arc, OnceLock,
+        atomic::{AtomicBool, AtomicU64, Ordering},
+        mpsc::{Receiver, RecvTimeoutError, SyncSender, TrySendError, sync_channel},
+    },
+    thread::{self, JoinHandle},
     time::Duration,
 };
 use tracing::*;
@@ -34,6 +39,10 @@ pub fn initialize_directshow() -> windows_core::Result<()> {
     unsafe { CoInitialize(None) }.ok()
 }
 
+fn initialize_directshow_worker() {
+    let _ = unsafe { CoInitializeEx(None, COINIT_MULTITHREADED) };
+}
+
 pub trait IPinExt {
     /// # Safety
     /// Do it correctly
@@ -460,7 +469,10 @@ impl VideoInputDevice {
                 .SetFormat(&**format)
                 .map_err(StartCapturingError::Other)?;
 
-            let sink_filter = SinkFilter::new(format.clone(), callback);
+            let dropped = Arc::new(AtomicU64::new(0));
+            let (sink_filter, callback_worker) =
+                SinkFilter::new(format.clone(), callback, dropped.clone())
+                    .map_err(StartCapturingError::CallbackWorker)?;
 
             let input_sink_pin = sink_filter
                 .get_pin(0)
@@ -515,6 +527,8 @@ impl VideoInputDevice {
                 graph_builder,
                 output_capture_pin: bound.output_pin,
                 input_sink_pin,
+                callback_worker,
+                dropped_samples: dropped,
             })
         }
     }
@@ -525,9 +539,15 @@ pub struct CaptureHandle {
     graph_builder: IGraphBuilder,
     output_capture_pin: IPin,
     input_sink_pin: IPin,
+    callback_worker: CallbackWorker,
+    dropped_samples: Arc<AtomicU64>,
 }
 
 impl CaptureHandle {
+    pub fn dropped_samples(&self) -> u64 {
+        self.dropped_samples.load(Ordering::Relaxed)
+    }
+
     // Chromium: VideoCaptureDeviceWin::StopAndDeallocate
     pub fn stop_capturing(self) -> windows_core::Result<()> {
         unsafe { self.media_control.Stop() }?;
@@ -537,6 +557,8 @@ impl CaptureHandle {
             let _ = self.graph_builder.Disconnect(&self.input_sink_pin);
         }
 
+        drop(self.callback_worker);
+
         Ok(())
     }
 }
@@ -553,6 +575,8 @@ pub enum StartCapturingError {
     ConfigureGraph(windows_core::Error),
     #[error("Run: {0}")]
     Run(windows_core::Error),
+    #[error("CallbackWorker: {0}")]
+    CallbackWorker(std::io::Error),
     #[error("{0}")]
     Other(windows_core::Error),
 }
@@ -582,6 +606,11 @@ impl Iterator for VideoMediaTypesIterator<'_> {
 #[derive(Default, Debug)]
 pub struct AMMediaType(AM_MEDIA_TYPE);
 
+// The media type owns its format block and is immutable after construction;
+// the callback worker only reads it.
+unsafe impl Send for AMMediaType {}
+unsafe impl Sync for AMMediaType {}
+
 impl AMMediaType {
     pub fn new(typ: &AM_MEDIA_TYPE) -> Self {
         Self(unsafe { copy_media_type(typ) })
@@ -593,6 +622,32 @@ impl AMMediaType {
         std::mem::forget(wrapper);
         inner
     }
+
+    fn matches_media_type(&self, other: &AM_MEDIA_TYPE) -> bool {
+        let a = &self.0;
+        let b = other;
+
+        if a.majortype != b.majortype
+            || a.subtype != b.subtype
+            || a.formattype != b.formattype
+            || a.cbFormat != b.cbFormat
+        {
+            return false;
+        }
+
+        if a.cbFormat == 0 {
+            return true;
+        }
+
+        if a.pbFormat.is_null() || b.pbFormat.is_null() {
+            return a.pbFormat == b.pbFormat;
+        }
+
+        unsafe {
+            std::slice::from_raw_parts(a.pbFormat, a.cbFormat as usize)
+                == std::slice::from_raw_parts(b.pbFormat, b.cbFormat as usize)
+        }
+    }
 }
 
 impl Deref for AMMediaType {
@@ -628,13 +683,22 @@ pub struct SinkFilter {
 }
 
 impl SinkFilter {
-    pub fn new(desired_media_type: AMMediaType, callback: SinkCallback) -> ComObject<Self> {
+    pub(crate) fn new(
+        desired_media_type: AMMediaType,
+        callback: SinkCallback,
+        dropped: Arc<AtomicU64>,
+    ) -> std::io::Result<(ComObject<Self>, CallbackWorker)> {
+        let (callback_tx, callback_rx) = sync_channel(CALLBACK_QUEUE_DEPTH);
+        let callback_worker =
+            CallbackWorker::spawn(callback, callback_rx, Arc::new(AtomicBool::new(false)))?;
+
         let this = SinkFilter {
             state: RefCell::new(State_Stopped),
             owning_graph: RefCell::new(None),
             input_pin: SinkInputPin {
                 desired_media_type,
-                callback: RefCell::new(callback),
+                callback_tx,
+                dropped,
                 current_media_type: Default::default(),
                 connected_pin: Default::default(),
                 owner: Default::default(),
@@ -646,7 +710,7 @@ impl SinkFilter {
         // SAFETY: SinkFilter always implements IBaseFilter
         *this.input_pin.owner.borrow_mut() = Some(this.cast::<IBaseFilter>().unwrap());
 
-        this
+        Ok((this, callback_worker))
     }
 
     fn no_of_pins(&self) -> u32 {
@@ -814,16 +878,73 @@ pub struct CallbackData<'a> {
     pub perf_counter: i64,
 }
 
-pub type SinkCallback = Box<dyn FnMut(CallbackData)>;
+pub type SinkCallback = Box<dyn FnMut(CallbackData) + Send>;
+
+const CALLBACK_QUEUE_DEPTH: usize = 4;
+
+struct OwnedCallbackData {
+    sample: IMediaSample,
+    media_type: Arc<AMMediaType>,
+    timestamp: Duration,
+    perf_counter: i64,
+}
+
+unsafe impl Send for OwnedCallbackData {}
+
+struct CallbackWorker {
+    stop: Arc<AtomicBool>,
+    handle: Option<JoinHandle<()>>,
+}
+
+impl CallbackWorker {
+    fn spawn(
+        mut callback: SinkCallback,
+        samples: Receiver<OwnedCallbackData>,
+        stop: Arc<AtomicBool>,
+    ) -> std::io::Result<Self> {
+        let worker_stop = stop.clone();
+        let handle = thread::Builder::new()
+            .name("dshow-camera-callback".to_string())
+            .spawn(move || {
+                initialize_directshow_worker();
+                while !worker_stop.load(Ordering::Acquire) {
+                    match samples.recv_timeout(Duration::from_millis(100)) {
+                        Ok(data) => callback(CallbackData {
+                            sample: &data.sample,
+                            media_type: data.media_type.as_ref(),
+                            timestamp: data.timestamp,
+                            perf_counter: data.perf_counter,
+                        }),
+                        Err(RecvTimeoutError::Timeout) => {}
+                        Err(RecvTimeoutError::Disconnected) => break,
+                    }
+                }
+            })?;
+
+        Ok(Self {
+            stop,
+            handle: Some(handle),
+        })
+    }
+}
+
+impl Drop for CallbackWorker {
+    fn drop(&mut self) {
+        self.stop.store(true, Ordering::Release);
+        if let Some(handle) = self.handle.take() {
+            let _ = handle.join();
+        }
+    }
+}
 
 #[implement(IPin, IMemInputPin)]
 struct SinkInputPin {
     desired_media_type: AMMediaType,
-    #[allow(unused)]
-    current_media_type: RefCell<AMMediaType>,
+    current_media_type: RefCell<Arc<AMMediaType>>,
     connected_pin: RefCell<Option<IPin>>,
     owner: RefCell<Option<IBaseFilter>>,
-    callback: RefCell<SinkCallback>,
+    callback_tx: SyncSender<OwnedCallbackData>,
+    dropped: Arc<AtomicU64>,
 }
 
 // impl SinkInputPin {
@@ -890,7 +1011,7 @@ impl IPin_Impl for SinkInputPin_Impl {
         };
         *connected_pin = Some(pconnector.clone());
         self.current_media_type
-            .replace(AMMediaType::new(unsafe { &*pmt }));
+            .replace(Arc::new(AMMediaType::new(unsafe { &*pmt })));
         S_OK.ok()
     }
 
@@ -917,7 +1038,7 @@ impl IPin_Impl for SinkInputPin_Impl {
             .as_ref()
             .ok_or(VFW_E_NOT_CONNECTED)?;
 
-        unsafe { *pmt = self.current_media_type.borrow().clone().into_inner() };
+        unsafe { *pmt = (**self.current_media_type.borrow()).clone().into_inner() };
 
         Ok(())
     }
@@ -1020,16 +1141,16 @@ impl IMemInputPin_Impl for SinkInputPin_Impl {
             return Ok(());
         };
 
-        unsafe {
-            if let Ok(new_media_type) = psample.GetMediaType()
+        let media_type = {
+            let mut current = self.current_media_type.borrow_mut();
+            if let Ok(new_media_type) = unsafe { psample.GetMediaType() }
                 && !new_media_type.is_null()
+                && !current.matches_media_type(unsafe { &*new_media_type })
             {
-                self.current_media_type
-                    .replace(AMMediaType::new(&*new_media_type));
+                *current = Arc::new(AMMediaType::new(unsafe { &*new_media_type }));
             }
-        }
-
-        let media_type = self.current_media_type.borrow();
+            Arc::clone(&*current)
+        };
 
         let length = unsafe { psample.GetActualDataLength() };
 
@@ -1046,12 +1167,19 @@ impl IMemInputPin_Impl for SinkInputPin_Impl {
 
         unsafe { psample.GetTime(&mut start_time, &mut end_time) }?;
 
-        (self.callback.borrow_mut())(CallbackData {
-            sample: psample,
-            media_type: &media_type,
+        let data = OwnedCallbackData {
+            sample: psample.clone(),
+            media_type,
             timestamp: Duration::from_micros(start_time as u64 / 10),
             perf_counter,
-        });
+        };
+
+        if let Err(TrySendError::Full(_)) = self.callback_tx.try_send(data) {
+            let dropped = self.dropped.fetch_add(1, Ordering::Relaxed) + 1;
+            if dropped == 1 || dropped.is_multiple_of(100) {
+                tracing::warn!(dropped, "DirectShow camera callback queue full");
+            }
+        }
 
         Ok(())
     }

crates/camera-ffmpeg/Cargo.toml

@@ -16,9 +16,13 @@ cidre = { workspace = true }
 cap-camera-avfoundation = { path = "../camera-avfoundation" }
 
 [target.'cfg(windows)'.dependencies]
-windows = { workspace = true }
+windows = { workspace = true, features = [
+    "Win32_Foundation",
+    "Win32_Media_MediaFoundation",
+] }
 windows-core = { workspace = true }
 cap-camera-windows = { path = "../camera-windows" }
+cap-mediafoundation-utils = { path = "../mediafoundation-utils" }
 
 [dev-dependencies]
 inquire = "0.7.5"