From 42e07453c55179ea2d66c58b3fc082ed2778d221 Mon Sep 17 00:00:00 2001 From: tstoltz Date: Sun, 19 Apr 2026 12:26:39 +0200 Subject: [PATCH 1/2] 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/2] 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 })