From cf169457c43abbf4fce979f2f1ea7e27106940b3 Mon Sep 17 00:00:00 2001 From: Brendan Allan Date: Fri, 17 Oct 2025 10:32:08 +0800 Subject: [PATCH 1/8] log number of captured frames --- .../src/sources/screen_capture/macos.rs | 102 ++++++++++++++---- .../src/sources/screen_capture/windows.rs | 25 +++++ 2 files changed, 104 insertions(+), 23 deletions(-) diff --git a/crates/recording/src/sources/screen_capture/macos.rs b/crates/recording/src/sources/screen_capture/macos.rs index c860b0cf929..112ec9c5638 100644 --- a/crates/recording/src/sources/screen_capture/macos.rs +++ b/crates/recording/src/sources/screen_capture/macos.rs @@ -8,12 +8,19 @@ use crate::{ }; use anyhow::{Context, anyhow}; use cidre::*; -use futures::{FutureExt, channel::mpsc, future::BoxFuture}; -use std::sync::{ - Arc, - atomic::{self, AtomicBool}, +use futures::{FutureExt as _, channel::mpsc, future::BoxFuture}; +use std::{ + sync::{ + Arc, + atomic::{self, AtomicBool, AtomicU32}, + }, + time::Duration, }; use tokio::sync::broadcast; +use tokio_util::{ + future::FutureExt as _, + sync::{CancellationToken, DropGuard}, +}; use tracing::debug; #[derive(Debug)] @@ -149,8 +156,27 @@ impl ScreenCaptureConfig { ns::Error::with_domain(ns::ErrorDomain::os_status(), 69420, None) ); + let video_frame_counter: Arc = Arc::new(AtomicU32::new(0)); + let cancel_token = CancellationToken::new(); + + tokio::spawn({ + let video_frame_count = video_frame_counter.clone(); + async move { + loop { + tokio::time::sleep(Duration::from_secs(3)); + debug!( + "Captured {} frames", + video_frame_count.load(atomic::Ordering::Relaxed) + ); + } + } + .with_cancellation_token_owned(cancel_token.clone()) + .in_current_span() + }); + let builder = scap_screencapturekit::Capturer::builder(content_filter, settings) .with_output_sample_buf_cb({ + let video_frame_count = video_frame_counter.clone(); move |frame| { let sample_buffer = frame.sample_buf(); @@ -166,6 +192,8 @@ impl ScreenCaptureConfig { return; } + video_frame_count.fetch_add(1, atomic::Ordering::Relaxed); + let check_skip_send = || { cap_fail::fail_err!( "media::sources::screen_capture::skip_send", @@ -218,17 +246,27 @@ impl ScreenCaptureConfig { } } }) - .with_stop_with_err_cb(move |_, err| { - let _ = error_tx.send(err.retained()); + .with_stop_with_err_cb({ + let video_frame_count = video_frame_counter.clone(); + move |_, err| { + debug!( + "Capturer stopping after creating {} video frames", + video_frame_count.load(atomic::Ordering::Relaxed) + ); + + let _ = error_tx.send(err.retained()); + } }); let capturer = Capturer::new(Arc::new(builder.build()?)); Ok(( - VideoSourceConfig( - ChannelVideoSourceConfig::new(self.video_info, video_rx), - capturer.clone(), - error_rx.resubscribe(), - ), + VideoSourceConfig { + inner: ChannelVideoSourceConfig::new(self.video_info, video_rx), + capturer: capturer.clone(), + error_rx: error_rx.resubscribe(), + drop_guard: cancel_token.drop_guard(), + video_frame_counter: video_frame_counter.clone(), + }, audio_rx.map(|rx| { SystemAudioSourceConfig( ChannelAudioSourceConfig::new(self.audio_info(), rx), @@ -297,12 +335,19 @@ impl Capturer { } } -pub struct VideoSourceConfig( - ChannelVideoSourceConfig, - Capturer, - broadcast::Receiver>, -); -pub struct VideoSource(ChannelVideoSource, Capturer); +pub struct VideoSourceConfig { + inner: ChannelVideoSourceConfig, + capturer: Capturer, + error_rx: broadcast::Receiver>, + drop_guard: DropGuard, + video_frame_counter: Arc, +} +pub struct VideoSource { + inner: ChannelVideoSource, + capturer: Capturer, + drop_guard: Option, + video_frame_counter: Arc, +} impl output_pipeline::VideoSource for VideoSource { type Config = VideoSourceConfig; @@ -317,21 +362,26 @@ impl output_pipeline::VideoSource for VideoSource { Self: Sized, { ctx.tasks().spawn("screen-capture", async move { - if let Ok(err) = config.2.recv().await { + if let Ok(err) = config.error_rx.recv().await { return Err(anyhow!("{err}")); } Ok(()) }); - ChannelVideoSource::setup(config.0, video_tx, ctx) + ChannelVideoSource::setup(config.inner, video_tx, ctx) .await - .map(|source| Self(source, config.1)) + .map(|source| Self { + inner: source, + capturer: config.capturer, + drop_guard: Some(config.drop_guard), + video_frame_counter: config.video_frame_counter, + }) } fn start(&mut self) -> BoxFuture<'_, anyhow::Result<()>> { async move { - self.1.start().await?; + self.capturer.start().await?; Ok(()) } @@ -340,7 +390,13 @@ impl output_pipeline::VideoSource for VideoSource { fn stop(&mut self) -> BoxFuture<'_, anyhow::Result<()>> { async move { - self.1.stop().await?; + debug!( + "Capturer stopping after creating {} video frames", + self.video_frame_counter.load(atomic::Ordering::Relaxed) + ); + self.capturer.stop().await?; + + drop(self.drop_guard.take()); Ok(()) } @@ -348,7 +404,7 @@ impl output_pipeline::VideoSource for VideoSource { } fn video_info(&self) -> VideoInfo { - self.0.video_info() + self.inner.video_info() } } diff --git a/crates/recording/src/sources/screen_capture/windows.rs b/crates/recording/src/sources/screen_capture/windows.rs index 05b020212f2..034c894b1cf 100644 --- a/crates/recording/src/sources/screen_capture/windows.rs +++ b/crates/recording/src/sources/screen_capture/windows.rs @@ -19,8 +19,13 @@ use scap_ffmpeg::*; use scap_targets::{Display, DisplayId}; use std::{ collections::VecDeque, + sync::atomic, time::{Duration, Instant}, }; +use tokio_util::{ + future::FutureExt as _, + sync::{CancellationToken, DropGuard}, +}; use tracing::{error, info, trace}; const WINDOW_DURATION: Duration = Duration::from_secs(3); @@ -199,10 +204,14 @@ impl output_pipeline::VideoSource for VideoSource { } }; + let video_frame_counter: Arc = Arc::new(AtomicU32::new(0)); + let cancel_token = CancellationToken::new(); + let res = scap_direct3d::Capturer::new( capture_item, settings, move |frame| { + video_frame_counter.fetch_add(1, atomic::Ordering::Relaxed); let timestamp = frame.inner().SystemRelativeTime()?; let timestamp = Timestamp::PerformanceCounter( PerformanceCounterTimestamp::new(timestamp.Duration), @@ -240,6 +249,22 @@ impl output_pipeline::VideoSource for VideoSource { return; }; + let drop_guard = cancel_token.drop_guard(); + tokio::spawn({ + let video_frame_count = video_frame_counter.clone(); + async move { + loop { + tokio::time::sleep(Duration::from_secs(3)); + debug!( + "Captured {} frames", + video_frame_count.load(atomic::Ordering::Relaxed) + ); + } + } + .with_cancellation_token_owned(cancel_token.clone()) + .in_current_span() + }); + trace!("Starting D3D capturer"); let start_result = capturer.start().map_err(Into::into); if let Err(ref e) = start_result { From f4e9ed990f49ca01e4840d41689bc25ef898ebe8 Mon Sep 17 00:00:00 2001 From: Brendan Allan Date: Fri, 17 Oct 2025 10:41:01 +0800 Subject: [PATCH 2/8] work properly --- .../src/sources/screen_capture/macos.rs | 45 ++++++++++--------- .../src/sources/screen_capture/windows.rs | 4 +- 2 files changed, 28 insertions(+), 21 deletions(-) diff --git a/crates/recording/src/sources/screen_capture/macos.rs b/crates/recording/src/sources/screen_capture/macos.rs index 112ec9c5638..d536b7376a9 100644 --- a/crates/recording/src/sources/screen_capture/macos.rs +++ b/crates/recording/src/sources/screen_capture/macos.rs @@ -157,22 +157,6 @@ impl ScreenCaptureConfig { ); let video_frame_counter: Arc = Arc::new(AtomicU32::new(0)); - let cancel_token = CancellationToken::new(); - - tokio::spawn({ - let video_frame_count = video_frame_counter.clone(); - async move { - loop { - tokio::time::sleep(Duration::from_secs(3)); - debug!( - "Captured {} frames", - video_frame_count.load(atomic::Ordering::Relaxed) - ); - } - } - .with_cancellation_token_owned(cancel_token.clone()) - .in_current_span() - }); let builder = scap_screencapturekit::Capturer::builder(content_filter, settings) .with_output_sample_buf_cb({ @@ -258,14 +242,17 @@ impl ScreenCaptureConfig { } }); + let cancel_token = CancellationToken::new(); let capturer = Capturer::new(Arc::new(builder.build()?)); + Ok(( VideoSourceConfig { inner: ChannelVideoSourceConfig::new(self.video_info, video_rx), capturer: capturer.clone(), error_rx: error_rx.resubscribe(), - drop_guard: cancel_token.drop_guard(), video_frame_counter: video_frame_counter.clone(), + cancel_token: cancel_token.clone(), + drop_guard: cancel_token.drop_guard(), }, audio_rx.map(|rx| { SystemAudioSourceConfig( @@ -339,14 +326,16 @@ pub struct VideoSourceConfig { inner: ChannelVideoSourceConfig, capturer: Capturer, error_rx: broadcast::Receiver>, + cancel_token: CancellationToken, drop_guard: DropGuard, video_frame_counter: Arc, } pub struct VideoSource { inner: ChannelVideoSource, capturer: Capturer, - drop_guard: Option, + cancel_token: CancellationToken, video_frame_counter: Arc, + _drop_guard: DropGuard, } impl output_pipeline::VideoSource for VideoSource { @@ -374,7 +363,8 @@ impl output_pipeline::VideoSource for VideoSource { .map(|source| Self { inner: source, capturer: config.capturer, - drop_guard: Some(config.drop_guard), + cancel_token: config.cancel_token, + _drop_guard: config.drop_guard, video_frame_counter: config.video_frame_counter, }) } @@ -383,6 +373,21 @@ impl output_pipeline::VideoSource for VideoSource { async move { self.capturer.start().await?; + tokio::spawn({ + let video_frame_count = self.video_frame_counter.clone(); + async move { + loop { + tokio::time::sleep(Duration::from_secs(3)).await; + debug!( + "Captured {} frames", + video_frame_count.load(atomic::Ordering::Relaxed) + ); + } + } + .with_cancellation_token_owned(self.cancel_token.clone()) + .in_current_span() + }); + Ok(()) } .boxed() @@ -396,7 +401,7 @@ impl output_pipeline::VideoSource for VideoSource { ); self.capturer.stop().await?; - drop(self.drop_guard.take()); + self.cancel_token.cancel(); Ok(()) } diff --git a/crates/recording/src/sources/screen_capture/windows.rs b/crates/recording/src/sources/screen_capture/windows.rs index 034c894b1cf..25d9f1d4bab 100644 --- a/crates/recording/src/sources/screen_capture/windows.rs +++ b/crates/recording/src/sources/screen_capture/windows.rs @@ -254,7 +254,7 @@ impl output_pipeline::VideoSource for VideoSource { let video_frame_count = video_frame_counter.clone(); async move { loop { - tokio::time::sleep(Duration::from_secs(3)); + tokio::time::sleep(Duration::from_secs(3)).await; debug!( "Captured {} frames", video_frame_count.load(atomic::Ordering::Relaxed) @@ -283,6 +283,8 @@ impl output_pipeline::VideoSource for VideoSource { if reply.send(capturer.stop().map_err(Into::into)).is_err() { return; } + + drop(drop_guard) }); ctx.tasks().spawn("d3d-capture", async move { From 86f7a0703e1d1a26919ba1077e7061b16f0bde17 Mon Sep 17 00:00:00 2001 From: Brendan Allan Date: Fri, 17 Oct 2025 10:48:02 +0800 Subject: [PATCH 3/8] cleanup fail_err --- crates/fail/src/lib.rs | 20 +++++++++++++++++++ .../src/sources/screen_capture/macos.rs | 19 +++--------------- .../src/sources/screen_capture/windows.rs | 2 +- 3 files changed, 24 insertions(+), 17 deletions(-) diff --git a/crates/fail/src/lib.rs b/crates/fail/src/lib.rs index de8f2038ad8..edfaef0f1b1 100644 --- a/crates/fail/src/lib.rs +++ b/crates/fail/src/lib.rs @@ -57,6 +57,26 @@ macro_rules! fail_err { } }; } +#[macro_export] +macro_rules! fail_ret { + ($name:literal) => { + #[cfg(debug_assertions)] + { + const NAME: &'static str = concat!(env!("CARGO_PKG_NAME"), "::", $name); + + $crate::private::inventory::submit! { + $crate::Fail { name: NAME } + } + + let should_fail = $crate::private::should_fail(NAME); + + if should_fail { + eprintln!("Purposely returned at '{NAME}'"); + return; + } + } + }; +} #[doc(hidden)] pub mod private { diff --git a/crates/recording/src/sources/screen_capture/macos.rs b/crates/recording/src/sources/screen_capture/macos.rs index d536b7376a9..f3f60ce3a75 100644 --- a/crates/recording/src/sources/screen_capture/macos.rs +++ b/crates/recording/src/sources/screen_capture/macos.rs @@ -178,14 +178,7 @@ impl ScreenCaptureConfig { video_frame_count.fetch_add(1, atomic::Ordering::Relaxed); - let check_skip_send = || { - cap_fail::fail_err!( - "media::sources::screen_capture::skip_send", - () - ); - - Ok::<(), ()>(()) - }; + cap_fail::fail_ret!("screen_capture vidoe skip"); let _ = video_tx.try_send(VideoFrame { sample_buf: sample_buffer.retained(), @@ -195,13 +188,7 @@ impl ScreenCaptureConfig { scap_screencapturekit::Frame::Audio(_) => { use ffmpeg::ChannelLayout; - let res = || { - cap_fail::fail_err!("screen_capture audio skip", ()); - Ok::<(), ()>(()) - }; - if res().is_err() { - return; - } + cap_fail::fail_ret!("screen_capture audio skip"); let Some(audio_tx) = &mut audio_tx else { return; @@ -377,7 +364,7 @@ impl output_pipeline::VideoSource for VideoSource { let video_frame_count = self.video_frame_counter.clone(); async move { loop { - tokio::time::sleep(Duration::from_secs(3)).await; + tokio::time::sleep(Duration::from_secs(5)).await; debug!( "Captured {} frames", video_frame_count.load(atomic::Ordering::Relaxed) diff --git a/crates/recording/src/sources/screen_capture/windows.rs b/crates/recording/src/sources/screen_capture/windows.rs index 25d9f1d4bab..d01350c4a07 100644 --- a/crates/recording/src/sources/screen_capture/windows.rs +++ b/crates/recording/src/sources/screen_capture/windows.rs @@ -254,7 +254,7 @@ impl output_pipeline::VideoSource for VideoSource { let video_frame_count = video_frame_counter.clone(); async move { loop { - tokio::time::sleep(Duration::from_secs(3)).await; + tokio::time::sleep(Duration::from_secs(5)).await; debug!( "Captured {} frames", video_frame_count.load(atomic::Ordering::Relaxed) From 0a4d99ccf9136d9eea071a8b27b3d048cfc01a38 Mon Sep 17 00:00:00 2001 From: Brendan Allan Date: Fri, 17 Oct 2025 10:52:15 +0800 Subject: [PATCH 4/8] improve naming --- crates/recording/src/sources/screen_capture/macos.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/recording/src/sources/screen_capture/macos.rs b/crates/recording/src/sources/screen_capture/macos.rs index f3f60ce3a75..abfbb7416f1 100644 --- a/crates/recording/src/sources/screen_capture/macos.rs +++ b/crates/recording/src/sources/screen_capture/macos.rs @@ -176,9 +176,9 @@ impl ScreenCaptureConfig { return; } - video_frame_count.fetch_add(1, atomic::Ordering::Relaxed); + cap_fail::fail_ret!("screen_capture video frame skip"); - cap_fail::fail_ret!("screen_capture vidoe skip"); + video_frame_count.fetch_add(1, atomic::Ordering::Relaxed); let _ = video_tx.try_send(VideoFrame { sample_buf: sample_buffer.retained(), @@ -188,7 +188,7 @@ impl ScreenCaptureConfig { scap_screencapturekit::Frame::Audio(_) => { use ffmpeg::ChannelLayout; - cap_fail::fail_ret!("screen_capture audio skip"); + cap_fail::fail_ret!("screen_capture audio frame skip"); let Some(audio_tx) = &mut audio_tx else { return; From 3c2fd1972f53836696c048a2b8f00f5dcd2c69c2 Mon Sep 17 00:00:00 2001 From: Brendan Allan Date: Fri, 17 Oct 2025 10:53:48 +0800 Subject: [PATCH 5/8] fix windows --- crates/recording/src/sources/screen_capture/windows.rs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/crates/recording/src/sources/screen_capture/windows.rs b/crates/recording/src/sources/screen_capture/windows.rs index d01350c4a07..081123be932 100644 --- a/crates/recording/src/sources/screen_capture/windows.rs +++ b/crates/recording/src/sources/screen_capture/windows.rs @@ -19,14 +19,17 @@ use scap_ffmpeg::*; use scap_targets::{Display, DisplayId}; use std::{ collections::VecDeque, - sync::atomic, + sync::{ + Arc, + atomic::{self, AtomicU32}, + }, time::{Duration, Instant}, }; use tokio_util::{ future::FutureExt as _, sync::{CancellationToken, DropGuard}, }; -use tracing::{error, info, trace}; +use tracing::*; const WINDOW_DURATION: Duration = Duration::from_secs(3); const LOG_INTERVAL: Duration = Duration::from_secs(5); From 07a7a4b129cfb1ff6fab429a6b6fdf63054ce712 Mon Sep 17 00:00:00 2001 From: Brendan Allan Date: Fri, 17 Oct 2025 12:27:18 +0800 Subject: [PATCH 6/8] fix windows --- .../src/sources/screen_capture/windows.rs | 24 ++++++++++--------- 1 file changed, 13 insertions(+), 11 deletions(-) diff --git a/crates/recording/src/sources/screen_capture/windows.rs b/crates/recording/src/sources/screen_capture/windows.rs index 081123be932..5bc63c7e8d7 100644 --- a/crates/recording/src/sources/screen_capture/windows.rs +++ b/crates/recording/src/sources/screen_capture/windows.rs @@ -213,15 +213,18 @@ impl output_pipeline::VideoSource for VideoSource { let res = scap_direct3d::Capturer::new( capture_item, settings, - move |frame| { - video_frame_counter.fetch_add(1, atomic::Ordering::Relaxed); - let timestamp = frame.inner().SystemRelativeTime()?; - let timestamp = Timestamp::PerformanceCounter( - PerformanceCounterTimestamp::new(timestamp.Duration), - ); - let _ = video_tx.try_send(VideoFrame { frame, timestamp }); - - Ok(()) + { + let video_frame_counter = video_frame_counter.clone(); + move |frame| { + video_frame_counter.fetch_add(1, atomic::Ordering::Relaxed); + let timestamp = frame.inner().SystemRelativeTime()?; + let timestamp = Timestamp::PerformanceCounter( + PerformanceCounterTimestamp::new(timestamp.Duration), + ); + let _ = video_tx.try_send(VideoFrame { frame, timestamp }); + + Ok(()) + } }, { let mut error_tx = error_tx.clone(); @@ -252,9 +255,7 @@ impl output_pipeline::VideoSource for VideoSource { return; }; - let drop_guard = cancel_token.drop_guard(); tokio::spawn({ - let video_frame_count = video_frame_counter.clone(); async move { loop { tokio::time::sleep(Duration::from_secs(5)).await; @@ -267,6 +268,7 @@ impl output_pipeline::VideoSource for VideoSource { .with_cancellation_token_owned(cancel_token.clone()) .in_current_span() }); + let drop_guard = cancel_token.drop_guard(); trace!("Starting D3D capturer"); let start_result = capturer.start().map_err(Into::into); From 3d1c50018a74acb96e9c8f5e97c5e6185da2b122 Mon Sep 17 00:00:00 2001 From: Brendan Allan Date: Fri, 17 Oct 2025 12:35:42 +0800 Subject: [PATCH 7/8] whoops --- crates/recording/src/sources/screen_capture/windows.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/recording/src/sources/screen_capture/windows.rs b/crates/recording/src/sources/screen_capture/windows.rs index 5bc63c7e8d7..61531e8ad73 100644 --- a/crates/recording/src/sources/screen_capture/windows.rs +++ b/crates/recording/src/sources/screen_capture/windows.rs @@ -261,7 +261,7 @@ impl output_pipeline::VideoSource for VideoSource { tokio::time::sleep(Duration::from_secs(5)).await; debug!( "Captured {} frames", - video_frame_count.load(atomic::Ordering::Relaxed) + video_frame_counter.load(atomic::Ordering::Relaxed) ); } } From 5069f247210de54483010760b67b3fe26491e741 Mon Sep 17 00:00:00 2001 From: Brendan Allan Date: Fri, 17 Oct 2025 12:57:45 +0800 Subject: [PATCH 8/8] properly fix windows --- crates/recording/src/sources/screen_capture/windows.rs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/crates/recording/src/sources/screen_capture/windows.rs b/crates/recording/src/sources/screen_capture/windows.rs index 61531e8ad73..f856304da51 100644 --- a/crates/recording/src/sources/screen_capture/windows.rs +++ b/crates/recording/src/sources/screen_capture/windows.rs @@ -182,6 +182,8 @@ impl output_pipeline::VideoSource for VideoSource { let (mut error_tx, mut error_rx) = mpsc::channel(1); let (ctrl_tx, ctrl_rx) = std::sync::mpsc::sync_channel::(1); + let tokio_rt = tokio::runtime::Handle::current(); + ctx.tasks().spawn_thread("d3d-capture-thread", move || { cap_mediafoundation_utils::thread_init(); @@ -255,8 +257,8 @@ impl output_pipeline::VideoSource for VideoSource { return; }; - tokio::spawn({ - async move { + tokio_rt.spawn( + async move { loop { tokio::time::sleep(Duration::from_secs(5)).await; debug!( @@ -267,7 +269,7 @@ impl output_pipeline::VideoSource for VideoSource { } .with_cancellation_token_owned(cancel_token.clone()) .in_current_span() - }); + ); let drop_guard = cancel_token.drop_guard(); trace!("Starting D3D capturer");