From 42e07453c55179ea2d66c58b3fc082ed2778d221 Mon Sep 17 00:00:00 2001 From: tstoltz Date: Sun, 19 Apr 2026 12:26:39 +0200 Subject: [PATCH 1/9] fix(linux): use two-level Frame::Video(VideoFrame::...) enum + SystemTime display_time MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Frame enum was refactored into a two-level shape (Frame::Audio / Frame::Video(VideoFrame::*)) and the per-format structs' display_time field was changed from u64 to SystemTime, but the Linux PipeWire engine in src/capturer/engine/linux/mod.rs was not updated to match. Result: 8 compile errors on Linux, no crates.io release compiles there. This commit: - Wraps the four emit sites (RGBx / RGB / xBGR / BGRx) in Frame::Video(VideoFrame::...(...)) to match frame::Frame's current two-level shape. - Replaces `display_time: timestamp as u64` with `display_time: SystemTime::now()`, matching what the macOS and Windows engines already do. The raw PipeWire pts from spa_meta_header is monotonic ns since an arbitrary reference, not wall-clock, so it cannot be converted losslessly to SystemTime. Relative frame ordering survives via channel-send order. - Adds SystemTime to the std::time imports and VideoFrame to the crate::frame imports. `cargo check` on Linux (Ubuntu 24, pipewire 1.0, via nix develop) now succeeds — 16 warnings remaining, all pre-existing lifetime- elision nits unrelated to this fix. --- src/capturer/engine/linux/mod.rs | 37 ++++++++++++++++++++------------ 1 file changed, 23 insertions(+), 14 deletions(-) diff --git a/src/capturer/engine/linux/mod.rs b/src/capturer/engine/linux/mod.rs index 07967fb3..471458aa 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}; @@ -133,31 +133,40 @@ fn process_callback(stream: &StreamRef, user_data: &mut ListenerUserData) { .to_vec() }; + // `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(); + if let Err(e) = match user_data.format.format() { - VideoFormat::RGBx => user_data.tx.send(Frame::RGBx(RGBxFrame { - display_time: timestamp as u64, + VideoFormat::RGBx => user_data.tx.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.send(Frame::RGB(RGBFrame { - display_time: timestamp as u64, + }))), + VideoFormat::RGB => user_data.tx.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.send(Frame::XBGR(XBGRFrame { - display_time: timestamp as u64, + }))), + VideoFormat::xBGR => user_data.tx.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.send(Frame::BGRx(BGRxFrame { - display_time: timestamp as u64, + }))), + VideoFormat::BGRx => user_data.tx.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}"); From be7853437a1e485b726076e87b4536b022f9eb1f Mon Sep 17 00:00:00 2001 From: tstoltz Date: Sat, 4 Jul 2026 05:50:51 +0200 Subject: [PATCH 2/9] fix(linux): bound the internal frame channel to stop unbounded memory growth Capturer::build() wired an unbounded std::sync::mpsc::channel() between the PipeWire callback thread (engine::linux::process_callback, which copies every frame via to_vec() unconditionally, independent of whether get_next_frame() is being drained) and the consumer. Any time the consumer lags PipeWire's delivery rate even briefly, frames pile up here with zero backpressure. Confirmed via heaptrack on a real KDE-Wayland capture session: ~1.44G of 1.47G leaked in a 25.76s run traced entirely to this call site. Bounding the channel to sync_channel(2) applies real backpressure to the PipeWire iterate thread instead. Re-verified with the same repro run 5x longer (136.74s): total leaked dropped to ~77.6MB (all attributable to unrelated debug-backtrace-formatting overhead from a retry loop elsewhere), and the process_callback leak site is completely gone from the profile. Propagates the channel's Sender -> SyncSender type change through Engine::new and the Linux engine (ListenerUserData, pipewire_capturer, LinuxCapturer::new, create_capturer). mac/windows engine modules are cfg-gated out on non-matching targets so this doesn't affect them at compile time on this branch. --- src/capturer/engine/linux/mod.rs | 8 ++++---- src/capturer/engine/mod.rs | 2 +- src/capturer/mod.rs | 11 ++++++++++- 3 files changed, 15 insertions(+), 6 deletions(-) diff --git a/src/capturer/engine/linux/mod.rs b/src/capturer/engine/linux/mod.rs index 471458aa..15f3f0c5 100644 --- a/src/capturer/engine/linux/mod.rs +++ b/src/capturer/engine/linux/mod.rs @@ -44,7 +44,7 @@ static STREAM_STATE_CHANGED_TO_ERROR: AtomicBool = AtomicBool::new(false); #[derive(Clone)] struct ListenerUserData { - pub tx: mpsc::Sender, + pub tx: mpsc::SyncSender, pub format: spa::param::video::VideoInfoRaw, } @@ -182,7 +182,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> { @@ -326,7 +326,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) @@ -373,6 +373,6 @@ impl LinuxCapturer { } } -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/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/mod.rs b/src/capturer/mod.rs index 04a189bb..769f13d0 100644 --- a/src/capturer/mod.rs +++ b/src/capturer/mod.rs @@ -111,7 +111,16 @@ impl Capturer { return Err(CapturerBuildError::PermissionNotGranted); } - let (tx, rx) = mpsc::channel(); + // Bounded (not `mpsc::channel()`): the Linux engine's PipeWire + // callback thread 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: ~1.44G leaked over a 25s run, + // entirely attributed to undrained `Frame` allocations queued from + // `engine::linux::process_callback`. A small bound applies real + // backpressure to the PipeWire iterate thread instead. + let (tx, rx) = mpsc::sync_channel(2); let engine = engine::Engine::new(&options, tx); Ok(Capturer { engine, rx }) From 9ed02117f213a079ffd88943cb7183077ace340a Mon Sep 17 00:00:00 2001 From: tstoltz Date: Wed, 15 Jul 2026 20:55:19 +0200 Subject: [PATCH 3/9] fix: propagate SyncSender to win/mac + avoid blocking send on all engines The previous commit (bounding the frame channel with sync_channel(2) to fix an unbounded-memory leak on Linux) changed Capturer::build()'s and Engine::new()'s shared, non-cfg-gated tx type from mpsc::Sender to mpsc::SyncSender, but only updated the Linux engine to match. Since Engine::new() unconditionally forwards that tx into win::create_capturer(tx: mpsc::Sender) and mac::create_capturer(tx: mpsc::Sender) -- both of which still declared the old Sender type -- this was a confirmed compile break on Windows and macOS (verified by reading those files directly; mpsc::Sender and mpsc::SyncSender are distinct, non-coercible types). This repo's PR CI doesn't build those targets, so it wouldn't have been caught before merge. This commit: - Propagates mpsc::Sender/ -> mpsc::SyncSender in win/mod.rs (Capturer, FlagStruct, create_capturer, spawn_audio_stream) and mac/mod.rs (CapturerInner, create_capturer), matching the shared Engine::new() signature. - Replaces blocking `.send()` with `.try_send()` on every frame/sample emit site across all three engines (Linux, Windows, macOS), dropping the frame when the channel is full instead of blocking the capture callback thread. This matters most on Linux: process_callback runs on the same thread that pipewire_capturer's loop polls CAPTURER_STATE on, and LinuxCapturer::stop_capture() joins that thread. A blocking send() on a full bounded channel -- e.g. because the consumer paused draining get_next_frame() -- would leave that thread permanently parked inside send(), so stop_capture() would hang forever waiting for a join that can never happen. try_send() + drop-on-full preserves the bounded- memory fix's intent without introducing that deadlock. Applied the same pattern to Windows' OS capture callback and audio thread, and macOS' ScreenCaptureKit delivery callback, for the same reason (an OS- driven callback thread blocking indefinitely is not something we want to newly introduce on those platforms either). Verification: - `cargo check` on Linux (via nix-shell -p pipewire dbus pkg-config clang) passes clean: same 16 pre-existing lifetime-elision warnings as before this commit, zero new warnings or errors. - win/mod.rs and mac/mod.rs cannot be compiled or type-checked on this machine (no Windows/macOS toolchain, and their cfg-gated `mod` declarations mean rustc never even parses them on a Linux host). Verified instead by: (1) `rustfmt --check` on both files, which requires syntactically valid Rust to run at all -- both pass; (2) manually cross-referencing every mpsc type at each call site listed above against its declaration. Recommend a maintainer or CI actually builds this branch on Windows/macOS before merging. --- src/capturer/engine/linux/mod.rs | 83 +++++++++++++++++++++----------- src/capturer/engine/mac/mod.rs | 9 ++-- src/capturer/engine/win/mod.rs | 25 ++++++---- src/capturer/mod.rs | 20 +++++--- 4 files changed, 90 insertions(+), 47 deletions(-) diff --git a/src/capturer/engine/linux/mod.rs b/src/capturer/engine/linux/mod.rs index 15f3f0c5..81952b46 100644 --- a/src/capturer/engine/linux/mod.rs +++ b/src/capturer/engine/linux/mod.rs @@ -142,35 +142,64 @@ fn process_callback(stream: &StreamRef, user_data: &mut ListenerUserData) { let _ = timestamp; // suppress "unused" warning until we wire pts elsewhere let display_time = SystemTime::now(); - if let Err(e) = match user_data.format.format() { - VideoFormat::RGBx => user_data.tx.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.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.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.send(Frame::Video(VideoFrame::BGRx(BGRxFrame { - display_time, - width: frame_size.width as i32, - height: frame_size.height as i32, - data: frame_data, - }))), + // `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 { + eprintln!("Frame receiver disconnected"); } + // TrySendError::Full is a silent intentional drop under + // backpressure — see comment above. } } else { eprintln!("Out of buffers"); diff --git a/src/capturer/engine/mac/mod.rs b/src/capturer/engine/mac/mod.rs index 3136a3e6..ba23dc9c 100644 --- a/src/capturer/engine/mac/mod.rs +++ b/src/capturer/engine/mac/mod.rs @@ -53,7 +53,7 @@ impl sc::stream::DelegateImpl for ErrorHandler { #[repr(C)] pub struct CapturerInner { - pub tx: mpsc::Sender, + pub tx: mpsc::SyncSender, } define_obj_type!(pub Capturer + StreamOutputImpl, CapturerInner, CAPTURER); @@ -69,7 +69,10 @@ impl sc::stream::OutputImpl for Capturer { sample_buf: &mut cm::SampleBuf, kind: sc::OutputType, ) { - let _ = self.inner_mut().tx.send((sample_buf.retained(), kind)); + // 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). + let _ = self.inner_mut().tx.try_send((sample_buf.retained(), kind)); } } @@ -85,7 +88,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 diff --git a/src/capturer/engine/win/mod.rs b/src/capturer/engine/win/mod.rs index 227aa325..4ed12564 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,10 @@ 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). + let _ = self.tx.try_send(Frame::Video(VideoFrame::BGRA(bgr_frame))); } None => { // get raw frame buffer @@ -134,7 +137,7 @@ impl GraphicsCaptureApiHandler for Capturer { data: frame_data, }; - let _ = self.tx.send(Frame::Video(VideoFrame::BGRA(bgr_frame))); + let _ = self.tx.try_send(Frame::Video(VideoFrame::BGRA(bgr_frame))); } } Ok(()) @@ -172,7 +175,7 @@ impl WCStream { #[derive(Clone, Debug)] struct FlagStruct { - pub tx: mpsc::Sender, + pub tx: mpsc::SyncSender, pub crop: Option, } @@ -184,7 +187,7 @@ pub enum CreateCapturerError { pub fn create_capturer( options: &Options, - tx: mpsc::Sender, + tx: mpsc::SyncSender, ) -> Result { let target = options .target @@ -380,7 +383,7 @@ fn build_audio_stream( } fn spawn_audio_stream( - tx: Sender, + tx: SyncSender, ready_tx: Sender>, ctrl_rx: Receiver, ) { @@ -445,8 +448,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 769f13d0..3a24dd0d 100644 --- a/src/capturer/mod.rs +++ b/src/capturer/mod.rs @@ -111,15 +111,19 @@ impl Capturer { return Err(CapturerBuildError::PermissionNotGranted); } - // Bounded (not `mpsc::channel()`): the Linux engine's PipeWire - // callback thread 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: ~1.44G leaked over a 25s run, + // 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.44G leaked over a 25s run, // entirely attributed to undrained `Frame` allocations queued from - // `engine::linux::process_callback`. A small bound applies real - // backpressure to the PipeWire iterate thread instead. + // `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(2); let engine = engine::Engine::new(&options, tx); From 2ae046c1d873abe37e5f38eed77a73532b93d41d Mon Sep 17 00:00:00 2001 From: tstoltz Date: Wed, 15 Jul 2026 21:07:19 +0200 Subject: [PATCH 4/9] fix(linux): stop the capture loop when the frame receiver disconnects Addresses a review comment from @tembo on PR #183: previously, if the frame receiver (Capturer/rx) was dropped without calling stop_capture() -- e.g. the caller just drops the Capturer -- the PipeWire callback thread kept running indefinitely, since nothing observed the disconnect. Every subsequent frame's try_send() would hit TrySendError::Disconnected and eprintln!, spamming a log line per frame forever with no way to stop it short of process exit. Reuses the existing STREAM_STATE_CHANGED_TO_ERROR flag (already checked by pipewire_capturer's main loop condition) to make a disconnected receiver actually end the capture loop, and swaps instead of stores so the disconnect is logged exactly once. --- src/capturer/engine/linux/mod.rs | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/src/capturer/engine/linux/mod.rs b/src/capturer/engine/linux/mod.rs index 81952b46..66447a46 100644 --- a/src/capturer/engine/linux/mod.rs +++ b/src/capturer/engine/linux/mod.rs @@ -196,7 +196,14 @@ fn process_callback(stream: &StreamRef, user_data: &mut ListenerUserData) { }; if let Err(mpsc::TrySendError::Disconnected(_)) = send_result { - eprintln!("Frame receiver disconnected"); + // 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_STATE_CHANGED_TO_ERROR.swap(true, std::sync::atomic::Ordering::Relaxed) { + eprintln!("Frame receiver disconnected"); + } } // TrySendError::Full is a silent intentional drop under // backpressure — see comment above. From 9b30e91663aba65201400708f6958e73efcadf12 Mon Sep 17 00:00:00 2001 From: tstoltz Date: Wed, 15 Jul 2026 21:08:03 +0200 Subject: [PATCH 5/9] =?UTF-8?q?docs(linux):=20fix=20wording=20nit=20from?= =?UTF-8?q?=20@tembo=20review=20=E2=80=94=20retained,=20not=20leaked?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The heaptrack-measured memory was queued/undrained Frame allocations, not a true leak (it would eventually free once drained) -- "leaked" overstates it. Matches the reviewer's suggested wording. --- src/capturer/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/capturer/mod.rs b/src/capturer/mod.rs index 3a24dd0d..f9e576b6 100644 --- a/src/capturer/mod.rs +++ b/src/capturer/mod.rs @@ -116,7 +116,7 @@ impl Capturer { // 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.44G leaked over a 25s run, + // 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 From 95e1e9e7eadcf7cc1773a896b6797fe5a63c5b34 Mon Sep 17 00:00:00 2001 From: tstoltz Date: Thu, 16 Jul 2026 09:26:12 +0200 Subject: [PATCH 6/9] refactor(linux): named channel-capacity const + rename exit flag Addresses two more nits from @tembo's review round on 9b30e91: - src/capturer/mod.rs: extract the magic `2` into `FRAME_CHANNEL_CAPACITY`, so tuning it later doesn't mean hunting the call site. - src/capturer/engine/linux/mod.rs: rename `STREAM_STATE_CHANGED_TO_ERROR` to `STREAM_SHOULD_EXIT`. It's set both on a genuine PipeWire stream error and (since 2ae046c) on a disconnected frame receiver -- the old name only described the first case. Also updates the stale inline comment on the loop condition that still said "only exits on stream error". --- src/capturer/engine/linux/mod.rs | 16 ++++++++++------ src/capturer/mod.rs | 7 ++++++- 2 files changed, 16 insertions(+), 7 deletions(-) diff --git a/src/capturer/engine/linux/mod.rs b/src/capturer/engine/linux/mod.rs index 66447a46..cb7159f7 100644 --- a/src/capturer/engine/linux/mod.rs +++ b/src/capturer/engine/linux/mod.rs @@ -40,7 +40,11 @@ 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 { @@ -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); } _ => {} } @@ -201,7 +205,7 @@ fn process_callback(stream: &StreamRef, user_data: &mut ListenerUserData) { // 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_STATE_CHANGED_TO_ERROR.swap(true, std::sync::atomic::Ordering::Relaxed) { + if !STREAM_SHOULD_EXIT.swap(true, std::sync::atomic::Ordering::Relaxed) { eprintln!("Frame receiver disconnected"); } } @@ -344,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)); } @@ -405,7 +409,7 @@ 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); } } diff --git a/src/capturer/mod.rs b/src/capturer/mod.rs index f9e576b6..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, @@ -124,7 +129,7 @@ impl Capturer { // 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(2); + let (tx, rx) = mpsc::sync_channel(FRAME_CHANNEL_CAPACITY); let engine = engine::Engine::new(&options, tx); Ok(Capturer { engine, rx }) From d118c15e7d6f01431ac94d75165523b8e6db6136 Mon Sep 17 00:00:00 2001 From: Tristan Stoltz Date: Thu, 16 Jul 2026 11:18:04 +0200 Subject: [PATCH 7/9] Update src/capturer/engine/win/mod.rs Co-authored-by: tembo[bot] <208362400+tembo[bot]@users.noreply.github.com> --- src/capturer/engine/win/mod.rs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/capturer/engine/win/mod.rs b/src/capturer/engine/win/mod.rs index 4ed12564..504c97b9 100644 --- a/src/capturer/engine/win/mod.rs +++ b/src/capturer/engine/win/mod.rs @@ -119,7 +119,12 @@ impl GraphicsCaptureApiHandler for Capturer { // 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). - let _ = self.tx.try_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()); + } + } } None => { // get raw frame buffer From 32569f135ad6e8ff240fd8d167fa7ae90aef6ca5 Mon Sep 17 00:00:00 2001 From: Tristan Stoltz Date: Thu, 16 Jul 2026 11:20:52 +0200 Subject: [PATCH 8/9] Update src/capturer/engine/win/mod.rs Co-authored-by: tembo[bot] <208362400+tembo[bot]@users.noreply.github.com> --- src/capturer/engine/win/mod.rs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/capturer/engine/win/mod.rs b/src/capturer/engine/win/mod.rs index 504c97b9..7899771f 100644 --- a/src/capturer/engine/win/mod.rs +++ b/src/capturer/engine/win/mod.rs @@ -142,7 +142,12 @@ impl GraphicsCaptureApiHandler for Capturer { data: frame_data, }; - let _ = self.tx.try_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(()) From 72c78cdc9c14b7f17ea691566e239e026719313c Mon Sep 17 00:00:00 2001 From: tstoltz Date: Thu, 16 Jul 2026 17:47:29 +0200 Subject: [PATCH 9/9] fix(mac): stop doing retain+send work after the frame receiver disconnects Mirrors the fix @tembo suggested and you already applied on both Windows call sites (d118c15, 32569f1): once tx.try_send() reports TrySendError::Disconnected, don't keep repeating the (retain + send) work on every subsequent ScreenCaptureKit callback with nowhere for the result to go. Windows' on_frame_arrived can signal this by returning an Err, which windows_capture's GraphicsCaptureApiHandler treats as "stop capture". impl_stream_did_output_sample_buf has no such return channel -- it's an extern "C" callback returning (). So this uses the flag-based alternative tembo's mac comment explicitly offered ("stop the sc::Stream (or flip some shared flag)"): a new `disconnected: Arc` on CapturerInner, set once (swap, logged once) on first Disconnected, checked at the top of the callback to skip work on every later frame. Doesn't attempt to call into sc::Stream's own stop path from inside the delivery callback, since I have no way to verify that's reentrant-safe on this machine (no macOS toolchain). --- src/capturer/engine/mac/mod.rs | 31 +++++++++++++++++++++++++++++-- 1 file changed, 29 insertions(+), 2 deletions(-) diff --git a/src/capturer/engine/mac/mod.rs b/src/capturer/engine/mac/mod.rs index ba23dc9c..5c0ef5e4 100644 --- a/src/capturer/engine/mac/mod.rs +++ b/src/capturer/engine/mac/mod.rs @@ -54,6 +54,13 @@ impl sc::stream::DelegateImpl for ErrorHandler { #[repr(C)] pub struct CapturerInner { 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,10 +76,27 @@ impl sc::stream::OutputImpl for Capturer { sample_buf: &mut cm::SampleBuf, kind: sc::OutputType, ) { + 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). - let _ = self.inner_mut().tx.try_send((sample_buf.retained(), kind)); + 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"); + } + } } } @@ -190,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();