From 6d550530fe1c6ec0c8011fe98f3485794f7aafc3 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Sun, 9 Aug 2026 00:01:08 +0200 Subject: [PATCH] =?UTF-8?q?feat(pf-capture):=20nothing=20had=20ever=20coun?= =?UTF-8?q?ted=20the=20compositor's=20buffer=20pool=20=E2=80=94=20the=20nu?= =?UTF-8?q?mber=20every=20zero-copy=20safety=20argument=20rests=20on?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wave-2 PW5 stage 1, and the one stage with no risk at all. The zero-copy capture path dups the dmabuf fd, publishes the frame, and hands the SPA buffer straight back to the producer at `.process` return — while the encode thread has not yet imported it, let alone read it. The code says so itself ("content stability across the brief import/encode window relies on the compositor's buffer-pool depth, like any zero-copy capture"). That depth is therefore load-bearing: it is the ONLY thing standing between us and the producer overwriting a buffer mid-read. And it had never been measured. Not logged, not asserted, not even requested — `build_dmabuf_buffers` set `SPA_PARAM_BUFFERS_dataType` and nothing else, so whatever the producer picked is what we got, silently. This adds the `add_buffer`/`remove_buffer` stream callbacks PipeWire has always offered and logs the count once per distinct depth: `pool_depth`, `high_water`, and the latest-frame-only `drained` count beside it. One line per session on a stable pool (`.process` runs at the capture rate — an unconditional log would be 240 lines a second of the same number), a second line if a renegotiation changes the depth. `high_water` is tracked separately from `live` because a renegotiation frees the pool before re-allocating it: any decision keyed on the live count would read that dip as "the pool shrank". `remove` saturates at zero rather than wrapping, so an unmatched remove cannot report `u32::MAX` buffers. Measurement only — no behaviour change, and no consumer of the number yet. PW5's later stages need it (a deeper encode pipeline widens the overwrite window by a full frame period), but the number is worth having regardless of whether those stages ever land: it is the answer to "is our zero-copy capture actually safe on this compositor", and until now the honest answer was "nobody knows". 3 tests pin the once-per-depth logging, the renegotiation dip, and the saturating remove. Gates green at CI parity. --- crates/pf-capture/src/linux/pipewire.rs | 121 +++++++++++++++++++++++- 1 file changed, 120 insertions(+), 1 deletion(-) diff --git a/crates/pf-capture/src/linux/pipewire.rs b/crates/pf-capture/src/linux/pipewire.rs index 317ae693..14f3091e 100644 --- a/crates/pf-capture/src/linux/pipewire.rs +++ b/crates/pf-capture/src/linux/pipewire.rs @@ -73,6 +73,10 @@ struct UserData { /// PW4 step 1: the producer-fence wait distribution, measured on this (the PipeWire loop) /// thread. Per-session, like the fall-through tally. fence_wait: FenceWaitStats, + /// PW5 step 1: the negotiated buffer-pool depth, counted from `add_buffer`/`remove_buffer`. + /// See [`PoolCensus`] — this is the number that decides whether a deeper encode pipeline is + /// safe, and until now nobody had it. + pool: PoolCensus, /// Raw-passthrough frames that silently fell through to the CPU de-pad path, by reason — see /// [`PassthroughFallbacks`]. Per-session: a fresh `UserData` is built per pipeline, so a /// compositor that starts serving dmabufs again after a rebuild gets a fresh log budget. @@ -503,6 +507,51 @@ impl FenceWaitStats { } } +/// PW5 stage 1: how many buffers the producer actually allocated for this stream. +/// +/// **Nothing in this codebase had ever counted them.** The zero-copy path dups the dmabuf fd and +/// publishes the frame while the SPA buffer is handed straight back to the producer at `.process` +/// return — so the only thing keeping capture untorn is that the producer round-robins a pool +/// deeper than our import+encode window. That depth was an unmeasured assumption; this makes it a +/// logged number, on every producer, before anything is built on it. +/// +/// `live` is maintained by the `add_buffer`/`remove_buffer` stream callbacks, which PipeWire fires +/// on the loop thread as the pool is allocated (and again, remove-then-add, on a renegotiation that +/// replaces it). There is no "pool complete" event, so the count is published from `.process`: by +/// the time the first buffer is dequeued the allocation has finished. +#[derive(Debug, Default, Clone, Copy)] +pub(super) struct PoolCensus { + /// Buffers currently in the pool (adds minus removes). + live: u32, + /// Deepest `live` seen this session — the number a depth decision must key on, since a + /// renegotiation can transiently shrink the pool to zero. + high_water: u32, + /// The `live` value already logged, so a stable pool logs exactly one line per distinct depth + /// (a renegotiation that changes the depth is worth a second line; 240 frames a second of the + /// same number is not). + logged: Option, +} + +impl PoolCensus { + fn add(&mut self) { + self.live += 1; + self.high_water = self.high_water.max(self.live); + } + + fn remove(&mut self) { + self.live = self.live.saturating_sub(1); + } + + /// Called from `.process`, once per buffer. Returns `Some(live)` the first time each distinct + /// depth is seen — the caller logs then and only then. + fn note_frame(&mut self) -> Option { + (self.logged != Some(self.live)).then(|| { + self.logged = Some(self.live); + self.live + }) + } +} + /// Per-session tally of raw-passthrough frames that fell through to the CPU path, with a one-line /// budget per distinct reason. /// @@ -1397,6 +1446,7 @@ pub fn pipewire_thread( linear_nv12_failed: false, dbg_log_n: 0, fence_wait: FenceWaitStats::default(), + pool: PoolCensus::default(), passthrough_fallbacks: PassthroughFallbacks::default(), cursor: CursorState::new(cursor_id0_hides), expect_dims: if expect_exact_dims { @@ -1495,6 +1545,11 @@ pub fn pipewire_thread( } } }) + // PW5 stage 1 — the pool census. PipeWire fires these on the loop thread as it allocates + // (and, on a renegotiation, frees then re-allocates) the stream's buffers. Counting only: + // the buffer pointer is not touched, so no lifetime question arises here. + .add_buffer(|_stream, ud, _buf| ud.pool.add()) + .remove_buffer(|_stream, ud, _buf| ud.pool.remove()) .process(|stream, ud| { // Latest-frame-only (OBS pattern): Mutter delivers buffers in bursts and recycles its // pool; an older queued buffer carries a STALE frame. Drain all queued buffers, requeue @@ -1523,6 +1578,25 @@ pub fn pipewire_thread( newest = next; drained += 1; } + // PW5 stage 1: publish the depth the producer actually negotiated, once per distinct + // value. MEASURED, not requested: `build_dmabuf_buffers` asks for a range and the + // producer picks — this line is the only place the picked number is visible. + // + // Why it matters beyond curiosity: `stream.queue_raw_buffer(newest)` at the end of this + // callback hands the buffer back while the encode thread may still be importing and + // reading its dmabuf, so content stability rests entirely on the producer not cycling + // back to this buffer before we are done with it. That window is `pool_depth` buffer + // periods wide. A pool of 2 has essentially none. + if let Some(depth) = ud.pool.note_frame() { + tracing::info!( + pool_depth = depth, + high_water = ud.pool.high_water, + drained, + "pipewire buffer pool negotiated — this is the producer's ACTUAL count \ + (add_buffer/remove_buffer), the window in which a buffer we handed back may \ + be rewritten while the encoder still reads it" + ); + } // Sacrificial-mode gate (kwin.rs `create`): until the producer renegotiates to the // expected dims, every buffer — frame AND cursor meta, whose positions are in the // doomed mode's space — belongs to the birth mode; consuming one would build the @@ -2128,7 +2202,7 @@ mod tests { use super::{ consumer_kind, resolved_capture_arm, CaptureArm, ConsumerKind, FenceWaitStats, - PassthroughFallback, PassthroughFallbacks, FENCE_WAIT_BUCKETS_US, + PassthroughFallback, PassthroughFallbacks, PoolCensus, FENCE_WAIT_BUCKETS_US, }; /// A PyroWave session is PyroWave even though it also flips `backend_is_vaapi` on (the @@ -2342,4 +2416,49 @@ mod tests { ); } } + + /// PW5 stage 1: a stable pool logs ONE line, not one per frame. `.process` runs at the capture + /// rate — an unconditional log here would be 240 lines a second of the same number. + #[test] + fn a_stable_pool_is_logged_once() { + let mut p = PoolCensus::default(); + for _ in 0..8 { + p.add(); + } + assert_eq!(p.note_frame(), Some(8)); + for _ in 0..100 { + assert_eq!(p.note_frame(), None, "the same depth must not re-log"); + } + } + + /// A renegotiation frees the pool and re-allocates it. The LIVE count therefore dips (and the + /// new depth is worth a second line), but `high_water` — the number a pipeline-depth decision + /// keys on — must not follow the dip down. + #[test] + fn a_renegotiated_pool_relogs_but_the_high_water_holds() { + let mut p = PoolCensus::default(); + for _ in 0..8 { + p.add(); + } + assert_eq!(p.note_frame(), Some(8)); + for _ in 0..8 { + p.remove(); + } + for _ in 0..4 { + p.add(); + } + assert_eq!(p.note_frame(), Some(4), "a changed depth is worth a line"); + assert_eq!(p.high_water, 8, "the deepest pool seen this session"); + } + + /// `remove_buffer` without a matching `add_buffer` must not wrap the count to `u32::MAX` — + /// a depth gate reading that would happily pipeline against a pool of zero. + #[test] + fn unmatched_removes_saturate_at_zero() { + let mut p = PoolCensus::default(); + p.remove(); + p.remove(); + assert_eq!(p.note_frame(), Some(0)); + assert_eq!(p.high_water, 0); + } }