diff --git a/src/capturer/engine/linux/mod.rs b/src/capturer/engine/linux/mod.rs index 07967fb3..cb7159f7 100644 --- a/src/capturer/engine/linux/mod.rs +++ b/src/capturer/engine/linux/mod.rs @@ -5,7 +5,7 @@ use std::{ mpsc::{self, sync_channel, SyncSender}, }, thread::JoinHandle, - time::Duration, + time::{Duration, SystemTime}, }; use pipewire as pw; @@ -31,7 +31,7 @@ use pw::{ use crate::{ capturer::Options, - frame::{BGRxFrame, Frame, RGBFrame, RGBxFrame, XBGRFrame}, + frame::{BGRxFrame, Frame, RGBFrame, RGBxFrame, VideoFrame, XBGRFrame}, }; use self::{error::LinCapError, portal::ScreenCastPortal}; @@ -40,11 +40,15 @@ mod error; mod portal; static CAPTURER_STATE: AtomicU8 = AtomicU8::new(0); -static STREAM_STATE_CHANGED_TO_ERROR: AtomicBool = AtomicBool::new(false); +// Signals pipewire_capturer's main loop to stop early. Set both on a +// genuine PipeWire stream error and (see process_callback) when the frame +// receiver has disconnected -- both are "the loop should end now" +// conditions, so one flag models the intent accurately rather than two. +static STREAM_SHOULD_EXIT: AtomicBool = AtomicBool::new(false); #[derive(Clone)] struct ListenerUserData { - pub tx: mpsc::Sender, + pub tx: mpsc::SyncSender, pub format: spa::param::video::VideoInfoRaw, } @@ -85,7 +89,7 @@ fn state_changed_callback( match new { StreamState::Error(e) => { eprintln!("pipewire: State changed to error({e})"); - STREAM_STATE_CHANGED_TO_ERROR.store(true, std::sync::atomic::Ordering::Relaxed); + STREAM_SHOULD_EXIT.store(true, std::sync::atomic::Ordering::Relaxed); } _ => {} } @@ -133,35 +137,80 @@ fn process_callback(stream: &StreamRef, user_data: &mut ListenerUserData) { .to_vec() }; - if let Err(e) = match user_data.format.format() { - VideoFormat::RGBx => user_data.tx.send(Frame::RGBx(RGBxFrame { - display_time: timestamp as u64, - width: frame_size.width as i32, - height: frame_size.height as i32, - data: frame_data, - })), - VideoFormat::RGB => user_data.tx.send(Frame::RGB(RGBFrame { - display_time: timestamp as u64, - width: frame_size.width as i32, - height: frame_size.height as i32, - data: frame_data, - })), - VideoFormat::xBGR => user_data.tx.send(Frame::XBGR(XBGRFrame { - display_time: timestamp as u64, - width: frame_size.width as i32, - height: frame_size.height as i32, - data: frame_data, - })), - VideoFormat::BGRx => user_data.tx.send(Frame::BGRx(BGRxFrame { - display_time: timestamp as u64, - width: frame_size.width as i32, - height: frame_size.height as i32, - data: frame_data, - })), + // `timestamp` (from spa_meta_header.pts) is a PipeWire monotonic + // nanosecond count since an arbitrary reference — not wall-clock. + // display_time's SystemTime contract is wall-clock, so we use + // SystemTime::now() here (matches what the macOS and Windows + // engines do today). Relative frame ordering survives via + // channel-send order; sub-millisecond buffer timing is lost. + let _ = timestamp; // suppress "unused" warning until we wire pts elsewhere + let display_time = SystemTime::now(); + + // `try_send`, not `send`: this callback runs on the same thread + // that `pipewire_capturer`'s loop uses to poll CAPTURER_STATE + // between `pw_loop.iterate()` calls. A blocking `send()` on a + // bounded channel would stall this thread whenever the consumer + // falls behind, and since `LinuxCapturer::stop_capture()` joins + // this exact thread, a full channel + a paused consumer would + // make `stop_capture()` hang forever. Dropping the frame under + // backpressure (rather than blocking the producer) keeps both + // the bounded-memory guarantee and a responsive stop path. + let send_result = match user_data.format.format() { + VideoFormat::RGBx => { + user_data + .tx + .try_send(Frame::Video(VideoFrame::RGBx(RGBxFrame { + display_time, + width: frame_size.width as i32, + height: frame_size.height as i32, + data: frame_data, + }))) + } + VideoFormat::RGB => { + user_data + .tx + .try_send(Frame::Video(VideoFrame::RGB(RGBFrame { + display_time, + width: frame_size.width as i32, + height: frame_size.height as i32, + data: frame_data, + }))) + } + VideoFormat::xBGR => { + user_data + .tx + .try_send(Frame::Video(VideoFrame::XBGR(XBGRFrame { + display_time, + width: frame_size.width as i32, + height: frame_size.height as i32, + data: frame_data, + }))) + } + VideoFormat::BGRx => { + user_data + .tx + .try_send(Frame::Video(VideoFrame::BGRx(BGRxFrame { + display_time, + width: frame_size.width as i32, + height: frame_size.height as i32, + data: frame_data, + }))) + } _ => panic!("Unsupported frame format received"), - } { - eprintln!("{e}"); + }; + + if let Err(mpsc::TrySendError::Disconnected(_)) = send_result { + // The consumer (Capturer/rx) is gone, e.g. dropped without + // calling stop_capture(). Signal the main loop above to + // exit instead of spinning pw_loop.iterate() forever on a + // stream nobody will ever read from, and log it once (swap + // instead of store) rather than once per incoming frame. + if !STREAM_SHOULD_EXIT.swap(true, std::sync::atomic::Ordering::Relaxed) { + eprintln!("Frame receiver disconnected"); + } } + // TrySendError::Full is a silent intentional drop under + // backpressure — see comment above. } } else { eprintln!("Out of buffers"); @@ -173,7 +222,7 @@ fn process_callback(stream: &StreamRef, user_data: &mut ListenerUserData) { // TODO: Format negotiation fn pipewire_capturer( options: Options, - tx: mpsc::Sender, + tx: mpsc::SyncSender, ready_sender: &SyncSender, stream_id: u32, ) -> Result<(), LinCapError> { @@ -299,8 +348,8 @@ fn pipewire_capturer( // User has called Capturer::start() and we start the main loop while CAPTURER_STATE.load(std::sync::atomic::Ordering::Relaxed) == 1 - && /* If the stream state got changed to `Error`, we exit. TODO: tell user that we exited */ - !STREAM_STATE_CHANGED_TO_ERROR.load(std::sync::atomic::Ordering::Relaxed) + && /* Exit early on a PipeWire stream error or a disconnected frame receiver. TODO: tell user that we exited */ + !STREAM_SHOULD_EXIT.load(std::sync::atomic::Ordering::Relaxed) { pw_loop.iterate(Duration::from_millis(100)); } @@ -317,7 +366,7 @@ pub struct LinuxCapturer { impl LinuxCapturer { // TODO: Error handling - pub fn new(options: &Options, tx: mpsc::Sender) -> Self { + pub fn new(options: &Options, tx: mpsc::SyncSender) -> Self { let connection = dbus::blocking::Connection::new_session().expect("Failed to create dbus connection"); let stream_id = ScreenCastPortal::new(&connection) @@ -360,10 +409,10 @@ impl LinuxCapturer { } } CAPTURER_STATE.store(0, std::sync::atomic::Ordering::Relaxed); - STREAM_STATE_CHANGED_TO_ERROR.store(false, std::sync::atomic::Ordering::Relaxed); + STREAM_SHOULD_EXIT.store(false, std::sync::atomic::Ordering::Relaxed); } } -pub fn create_capturer(options: &Options, tx: mpsc::Sender) -> LinuxCapturer { +pub fn create_capturer(options: &Options, tx: mpsc::SyncSender) -> LinuxCapturer { LinuxCapturer::new(options, tx) } diff --git a/src/capturer/engine/mac/mod.rs b/src/capturer/engine/mac/mod.rs index 3136a3e6..5c0ef5e4 100644 --- a/src/capturer/engine/mac/mod.rs +++ b/src/capturer/engine/mac/mod.rs @@ -53,7 +53,14 @@ impl sc::stream::DelegateImpl for ErrorHandler { #[repr(C)] pub struct CapturerInner { - pub tx: mpsc::Sender, + pub tx: mpsc::SyncSender, + // Set once tx.try_send() observes a disconnected receiver, so later + // callbacks can skip the retain+send work entirely instead of repeating + // it with nowhere for the result to go. There's no return-an-error path + // here (unlike Windows' on_frame_arrived) to actually stop the + // sc::Stream from this callback, so this is the same "flip a shared + // flag" compromise used for the equivalent case on Linux/Windows. + disconnected: Arc, } define_obj_type!(pub Capturer + StreamOutputImpl, CapturerInner, CAPTURER); @@ -69,7 +76,27 @@ impl sc::stream::OutputImpl for Capturer { sample_buf: &mut cm::SampleBuf, kind: sc::OutputType, ) { - let _ = self.inner_mut().tx.send((sample_buf.retained(), kind)); + let inner = self.inner_mut(); + if inner + .disconnected + .load(std::sync::atomic::Ordering::Relaxed) + { + return; + } + + // try_send, not send: don't block ScreenCaptureKit's delivery queue + // when the consumer is behind — drop the sample buffer instead + // (same rationale as the Linux/Windows engines). + if let Err(mpsc::TrySendError::Disconnected(_)) = + inner.tx.try_send((sample_buf.retained(), kind)) + { + if !inner + .disconnected + .swap(true, std::sync::atomic::Ordering::Relaxed) + { + eprintln!("Frame receiver disconnected"); + } + } } } @@ -85,7 +112,7 @@ pub(crate) enum CreateCapturerError { pub(crate) fn create_capturer( options: &Options, - tx: mpsc::Sender, + tx: mpsc::SyncSender, error_flag: Arc, ) -> Result<(arc::R, arc::R, arc::R), CreateCapturerError> { // If no target is specified, capture the main display @@ -187,7 +214,10 @@ pub(crate) fn create_capturer( let error_handler = ErrorHandler::with(ErrorHandlerInner { error_flag }); let stream = sc::Stream::with_delegate(&filter, &stream_config, error_handler.as_ref()); - let capturer = CapturerInner { tx }; + let capturer = CapturerInner { + tx, + disconnected: Arc::new(AtomicBool::new(false)), + }; let queue = dispatch::Queue::serial_with_ar_pool(); diff --git a/src/capturer/engine/mod.rs b/src/capturer/engine/mod.rs index e8aa3a00..a1535c11 100644 --- a/src/capturer/engine/mod.rs +++ b/src/capturer/engine/mod.rs @@ -58,7 +58,7 @@ pub struct Engine { } impl Engine { - pub fn new(options: &Options, tx: mpsc::Sender) -> Engine { + pub fn new(options: &Options, tx: mpsc::SyncSender) -> Engine { #[cfg(target_os = "macos")] { let error_flag = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); diff --git a/src/capturer/engine/win/mod.rs b/src/capturer/engine/win/mod.rs index 227aa325..7899771f 100644 --- a/src/capturer/engine/win/mod.rs +++ b/src/capturer/engine/win/mod.rs @@ -13,7 +13,7 @@ use std::{cmp, time::Duration}; use std::{ os::windows, ptr::null_mut, - sync::mpsc::{self, Receiver, RecvTimeoutError, Sender}, + sync::mpsc::{self, Receiver, RecvTimeoutError, Sender, SyncSender}, }; use windows_capture::{ capture::{CaptureControl, Context, GraphicsCaptureApiHandler}, @@ -29,7 +29,7 @@ use windows_capture::{ #[derive(Debug)] struct Capturer { - pub tx: mpsc::Sender, + pub tx: mpsc::SyncSender, pub crop: Option, pub start_time: (i64, SystemTime), pub perf_freq: i64, @@ -116,7 +116,15 @@ impl GraphicsCaptureApiHandler for Capturer { data: raw_frame_buffer.to_vec(), }; - let _ = self.tx.send(Frame::Video(VideoFrame::BGRA(bgr_frame))); + // try_send, not send: don't block the OS capture callback + // thread when the consumer is behind — drop the frame + // instead (same rationale as the Linux engine). + match self.tx.try_send(Frame::Video(VideoFrame::BGRA(bgr_frame))) { + Ok(()) | Err(mpsc::TrySendError::Full(_)) => {} + Err(mpsc::TrySendError::Disconnected(_)) => { + return Err("frame channel disconnected".into()); + } + } } None => { // get raw frame buffer @@ -134,7 +142,12 @@ impl GraphicsCaptureApiHandler for Capturer { data: frame_data, }; - let _ = self.tx.send(Frame::Video(VideoFrame::BGRA(bgr_frame))); + match self.tx.try_send(Frame::Video(VideoFrame::BGRA(bgr_frame))) { + Ok(()) | Err(mpsc::TrySendError::Full(_)) => {} + Err(mpsc::TrySendError::Disconnected(_)) => { + return Err("frame channel disconnected".into()); + } + } } } Ok(()) @@ -172,7 +185,7 @@ impl WCStream { #[derive(Clone, Debug)] struct FlagStruct { - pub tx: mpsc::Sender, + pub tx: mpsc::SyncSender, pub crop: Option, } @@ -184,7 +197,7 @@ pub enum CreateCapturerError { pub fn create_capturer( options: &Options, - tx: mpsc::Sender, + tx: mpsc::SyncSender, ) -> Result { let target = options .target @@ -380,7 +393,7 @@ fn build_audio_stream( } fn spawn_audio_stream( - tx: Sender, + tx: SyncSender, ready_tx: Sender>, ctrl_rx: Receiver, ) { @@ -445,8 +458,12 @@ fn spawn_audio_stream( timestamp, ); - if let Err(_) = tx.send(Frame::Audio(frame)) { - return; + match tx.try_send(Frame::Audio(frame)) { + Ok(()) => {} + // Consumer is behind: drop this sample under backpressure, + // keep the audio thread alive (matches video's try_send). + Err(mpsc::TrySendError::Full(_)) => {} + Err(mpsc::TrySendError::Disconnected(_)) => return, }; } }); diff --git a/src/capturer/mod.rs b/src/capturer/mod.rs index 04a189bb..dd22f282 100644 --- a/src/capturer/mod.rs +++ b/src/capturer/mod.rs @@ -12,6 +12,11 @@ use crate::{ pub use engine::get_output_frame_size; +/// Capacity of the internal channel each platform engine sends `Frame`s +/// (or `ChannelItem`s) through. See the comment at its use site in +/// [`Capturer::build`] for why this is bounded at all. +const FRAME_CHANNEL_CAPACITY: usize = 2; + #[derive(Debug, Clone, Copy, Default)] pub enum Resolution { _480p, @@ -111,7 +116,20 @@ impl Capturer { return Err(CapturerBuildError::PermissionNotGranted); } - let (tx, rx) = mpsc::channel(); + // Bounded (not `mpsc::channel()`): each platform's capture callback + // pushes frames into this channel unconditionally, independent of + // how fast `get_next_frame()` is drained downstream. An unbounded + // channel here means any transient slowdown in the consumer + // (encode/network backpressure) grows this queue without limit -- + // confirmed via heaptrack on Linux: ~1.44GiB retained over a 25s run, + // entirely attributed to undrained `Frame` allocations queued from + // `engine::linux::process_callback`. Each engine sends with + // `try_send` (not blocking `send`) and drops the frame when this + // channel is full, so a slow consumer loses frames instead of + // growing memory unboundedly or blocking the platform capture + // thread (which on Linux would otherwise deadlock `stop_capture()`, + // since it joins that same thread). + let (tx, rx) = mpsc::sync_channel(FRAME_CHANNEL_CAPACITY); let engine = engine::Engine::new(&options, tx); Ok(Capturer { engine, rx })