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 c860b0cf929..abfbb7416f1 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,11 @@ impl ScreenCaptureConfig { ns::Error::with_domain(ns::ErrorDomain::os_status(), 69420, None) ); + let video_frame_counter: Arc = Arc::new(AtomicU32::new(0)); + 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,14 +176,9 @@ impl ScreenCaptureConfig { return; } - let check_skip_send = || { - cap_fail::fail_err!( - "media::sources::screen_capture::skip_send", - () - ); + cap_fail::fail_ret!("screen_capture video frame skip"); - Ok::<(), ()>(()) - }; + video_frame_count.fetch_add(1, atomic::Ordering::Relaxed); let _ = video_tx.try_send(VideoFrame { sample_buf: sample_buffer.retained(), @@ -183,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 frame skip"); let Some(audio_tx) = &mut audio_tx else { return; @@ -218,17 +217,30 @@ 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 cancel_token = CancellationToken::new(); 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(), + video_frame_counter: video_frame_counter.clone(), + cancel_token: cancel_token.clone(), + drop_guard: cancel_token.drop_guard(), + }, audio_rx.map(|rx| { SystemAudioSourceConfig( ChannelAudioSourceConfig::new(self.audio_info(), rx), @@ -297,12 +309,21 @@ 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>, + cancel_token: CancellationToken, + drop_guard: DropGuard, + video_frame_counter: Arc, +} +pub struct VideoSource { + inner: ChannelVideoSource, + capturer: Capturer, + cancel_token: CancellationToken, + video_frame_counter: Arc, + _drop_guard: DropGuard, +} impl output_pipeline::VideoSource for VideoSource { type Config = VideoSourceConfig; @@ -317,21 +338,42 @@ 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, + cancel_token: config.cancel_token, + _drop_guard: 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?; + + tokio::spawn({ + let video_frame_count = self.video_frame_counter.clone(); + async move { + loop { + tokio::time::sleep(Duration::from_secs(5)).await; + debug!( + "Captured {} frames", + video_frame_count.load(atomic::Ordering::Relaxed) + ); + } + } + .with_cancellation_token_owned(self.cancel_token.clone()) + .in_current_span() + }); Ok(()) } @@ -340,7 +382,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?; + + self.cancel_token.cancel(); Ok(()) } @@ -348,7 +396,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..f856304da51 100644 --- a/crates/recording/src/sources/screen_capture/windows.rs +++ b/crates/recording/src/sources/screen_capture/windows.rs @@ -19,9 +19,17 @@ use scap_ffmpeg::*; use scap_targets::{Display, DisplayId}; use std::{ collections::VecDeque, + sync::{ + Arc, + atomic::{self, AtomicU32}, + }, time::{Duration, Instant}, }; -use tracing::{error, info, trace}; +use tokio_util::{ + future::FutureExt as _, + sync::{CancellationToken, DropGuard}, +}; +use tracing::*; const WINDOW_DURATION: Duration = Duration::from_secs(3); const LOG_INTERVAL: Duration = Duration::from_secs(5); @@ -174,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(); @@ -199,17 +209,24 @@ 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| { - 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(); @@ -240,6 +257,21 @@ impl output_pipeline::VideoSource for VideoSource { return; }; + tokio_rt.spawn( + async move { + loop { + tokio::time::sleep(Duration::from_secs(5)).await; + debug!( + "Captured {} frames", + video_frame_counter.load(atomic::Ordering::Relaxed) + ); + } + } + .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); if let Err(ref e) = start_result { @@ -258,6 +290,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 {