From bbe4380b413db164be8cee8e5f1f1e5bb0dc0dd5 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Fri, 17 Jul 2026 19:31:49 +0200 Subject: [PATCH] perf(latency): T1.1 frame-driven encode trigger + T1.4 time-based flush thresholds MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit design/latency-reduction-2026-07.md tier 1, remaining halves: - T1.1: the native encode loop wakes on the capture's ACTUAL arrival instead of sampling at a free-running tick — deletes the sample-and-hold (~half a frame interval on average, a full one worst-case: ~8ms avg @60fps). New Capturer::supports_arrival_wait/wait_arrival pair (IDD-push waits its frame-ready event against the shared-header token; the PipeWire portal blocks its channel with a pending stash); backends without an arrival signal — and PUNKTFUNK_FRAME_DRIVEN=0 — keep the legacy tick bit-identically. A 0.9x-interval rate floor caps encode at ~1.11x target when the compositor outruns the session; a +0.5x-interval keepalive keeps static desktops re-encoding at 1.5x-interval cadence. Pacing deadlines re-anchor to the actual submit so they can't drift against the arrival clock. GameStream plane untouched. - T1.4: the jump-to-live detectors run on WALL-CLOCK now (STANDING_TIME / FLUSH_AFTER = 250ms) instead of 30-frame counts whose meaning scaled with fps (500ms @60 but 125ms @240 — and stretching further under T1.1's slower static-scene repeats). The queue trip also requires depth still >= high, so a hysteresis-band hover can't fire on elapsed time alone. Validated: .21 Linux 185 core + 177 host + pf-capture tests, clippy -D warnings; .133 Windows cargo check of pf-capture + punktfunk-host green. Co-Authored-By: Claude Fable 5 --- crates/pf-capture/src/lib.rs | 16 ++++++ crates/pf-capture/src/linux/mod.rs | 34 +++++++++++- crates/pf-capture/src/windows/idd_push.rs | 26 +++++++++ .../src/client/frame_channel.rs | 24 ++++---- crates/punktfunk-core/src/client/pump.rs | 55 ++++++++++--------- crates/punktfunk-host/src/native/stream.rs | 38 +++++++++++-- 6 files changed, 152 insertions(+), 41 deletions(-) diff --git a/crates/pf-capture/src/lib.rs b/crates/pf-capture/src/lib.rs index 9498c829..07f39ab8 100644 --- a/crates/pf-capture/src/lib.rs +++ b/crates/pf-capture/src/lib.rs @@ -32,6 +32,22 @@ pub trait Capturer: Send { self.next_frame().map(Some) } + /// Whether this backend can block until a frame ARRIVES ([`wait_arrival`] + /// (Self::wait_arrival)) — the frame-driven encode trigger (latency plan T1.1). `false` + /// (the default) keeps the encode loop on its legacy fixed-cadence tick for this backend. + fn supports_arrival_wait(&self) -> bool { + false + } + + /// Block until a FRESH frame is available via [`try_latest`](Self::try_latest) or + /// `deadline` passes — the encode loop's frame-driven wait (latency plan T1.1): waking on + /// the compositor's publish instead of sampling at a free-running tick deletes the + /// sample-and-hold (~half a frame interval on average). Must NOT consume the frame (the + /// loop's `try_latest` call does); backends buffer internally where the arrival channel + /// can't be peeked. Only called when [`supports_arrival_wait`](Self::supports_arrival_wait) + /// is `true`; errors surface at the following `try_latest`. + fn wait_arrival(&mut self, _deadline: std::time::Instant) {} + /// Gate expensive per-frame work so the capturer can be kept alive (reused) between /// streams without burning CPU. The portal capturer skips the de-pad copy while inactive; /// the default is a no-op (synthetic sources are produced on demand). Set `true` for the diff --git a/crates/pf-capture/src/linux/mod.rs b/crates/pf-capture/src/linux/mod.rs index 0de70ccd..ded153c8 100644 --- a/crates/pf-capture/src/linux/mod.rs +++ b/crates/pf-capture/src/linux/mod.rs @@ -35,6 +35,10 @@ use std::time::Duration; /// and no second session to conflict with). pub struct PortalCapturer { frames: Receiver, + /// A frame [`wait_arrival`](Capturer::wait_arrival) received while blocking (the channel + /// can't be peeked, so the wait must consume) — always the FIRST candidate the next + /// `try_latest`/`next_frame` considers, before draining anything newer off the channel. + pending: Option, active: Arc, /// Set true once the PipeWire stream agrees a video format. Read in [`next_frame`]'s timeout /// branch to tell "format never negotiated" (modifier/format mismatch) apart from "negotiated @@ -166,6 +170,7 @@ impl PwHandles { fn into_capturer(self, node_id: u32, keepalive: Option>) -> PortalCapturer { PortalCapturer { frames: self.frames, + pending: None, active: self.active, negotiated: self.negotiated, streaming: self.streaming, @@ -266,6 +271,9 @@ impl Capturer for PortalCapturer { self.node_id )); } + if let Some(f) = self.pending.take() { + return Ok(f); // a wait_arrival stash outranks the channel (it's older) + } let slice = Duration::from_millis(500) .min(deadline.saturating_duration_since(std::time::Instant::now())); match self.frames.recv_timeout(slice) { @@ -276,6 +284,27 @@ impl Capturer for PortalCapturer { } } + fn supports_arrival_wait(&self) -> bool { + true + } + + fn wait_arrival(&mut self, deadline: std::time::Instant) { + // Frame-driven trigger (latency plan T1.1): block on the PipeWire channel until the + // compositor delivers a frame or the deadline passes. The channel can't be peeked, so + // the received frame is stashed in `pending` — `try_latest` starts from it and still + // drains anything newer. A broken/ended stream just returns; the following + // `try_latest` surfaces the error through its existing paths. + if self.pending.is_some() || self.broken.load(Ordering::Relaxed) { + return; + } + let Some(left) = deadline.checked_duration_since(std::time::Instant::now()) else { + return; + }; + if let Ok(f) = self.frames.recv_timeout(left) { + self.pending = Some(f); + } + } + fn try_latest(&mut self) -> Result> { if self.broken.load(Ordering::Relaxed) { return Err(anyhow!( @@ -285,8 +314,9 @@ impl Capturer for PortalCapturer { )); } // Drain to the newest queued frame without blocking; `None` means the compositor - // hasn't produced a new frame since last call (static/idle desktop). - let mut latest = None; + // hasn't produced a new frame since last call (static/idle desktop). A frame + // `wait_arrival` stashed is the oldest candidate — anything on the channel is newer. + let mut latest = self.pending.take(); loop { match self.frames.try_recv() { Ok(frame) => latest = Some(frame), diff --git a/crates/pf-capture/src/windows/idd_push.rs b/crates/pf-capture/src/windows/idd_push.rs index 6626fcc6..3dfa7402 100644 --- a/crates/pf-capture/src/windows/idd_push.rs +++ b/crates/pf-capture/src/windows/idd_push.rs @@ -1625,6 +1625,32 @@ impl Capturer for IddPushCapturer { self.try_consume() } + fn supports_arrival_wait(&self) -> bool { + true + } + + fn wait_arrival(&mut self, deadline: Instant) { + // Frame-driven trigger (latency plan T1.1): wake the encode loop the moment the driver + // publishes a frame we haven't consumed. The shared-header token is the truth (state, + // not edge — the auto-reset event may have been consumed by an earlier wait); the event + // is just the wakeup, waited in ≤16 ms slices exactly like `next_frame`'s bringup wait. + loop { + let tok = frame::FrameToken::unpack(self.latest()); + if tok.generation == self.generation && u64::from(tok.seq) != self.last_seq { + return; // a fresh publish exists — `try_latest` will consume it + } + let Some(left) = deadline.checked_duration_since(Instant::now()) else { + return; + }; + let ms = (left.as_millis() as u32).clamp(1, 16); + // SAFETY: `self.event` is the live frame-ready `OwnedHandle` this capturer owns; its + // raw value (borrowed for the call, so it outlives this synchronous wait) is a valid + // auto-reset event handle. `WaitForSingleObject` only reads the handle; the bounded + // timeout keeps the wait sliced. + let _ = unsafe { WaitForSingleObject(HANDLE(self.event.as_raw_handle()), ms) }; + } + } + fn hdr_meta(&self) -> Option { // While the display is HDR we emit BT.2020 PQ (Rgb10a2) → the encoder forces HEVC Main10 + the // PQ VUI; pair that with a mastering-display SEI so any decoder tone-maps from a real grade. The diff --git a/crates/punktfunk-core/src/client/frame_channel.rs b/crates/punktfunk-core/src/client/frame_channel.rs index 2aaf1edf..99339bf5 100644 --- a/crates/punktfunk-core/src/client/frame_channel.rs +++ b/crates/punktfunk-core/src/client/frame_channel.rs @@ -15,13 +15,16 @@ pub(crate) const QUEUE_HIGH: usize = 6; /// A true standing queue never falls back to this; a clump does within a few frames. pub(crate) const QUEUE_LOW: usize = 2; -/// Consecutive frames the hand-off queue must sit ≥ [`QUEUE_HIGH`] (never dropping to [`QUEUE_LOW`]) -/// before the pump declares a standing backlog and jumps to live. ~0.5 s at 60 fps — long enough that -/// a burst/clump (which drains in a few frames) never reaches it. -pub(crate) const STANDING_FRAMES: u32 = 30; +/// How long the hand-off queue must sit ≥ [`QUEUE_HIGH`] (never dropping to [`QUEUE_LOW`], and +/// still ≥ high at the trip) before the pump declares a standing backlog and jumps to live. +/// WALL-CLOCK (latency plan T1.4) — the old 30-FRAME count made the accepted backlog scale with +/// fps (500 ms @60 but 125 ms @240) and stretched further under the frame-driven host's slower +/// static-scene repeat cadence. 250 ms sits comfortably above a Wi-Fi clump's drain time (a +/// clump empties in a few frames, ≤ ~100 ms) at any rate. +pub(crate) const STANDING_TIME: Duration = Duration::from_millis(250); /// Memory backstop on the pre-decode hand-off queue. The standing-queue detector jumps to live long -/// before this (typically ≤ QUEUE_HIGH + STANDING_FRAMES deep), and a jump already requested a +/// before this (typically ≤ QUEUE_HIGH + a [`STANDING_TIME`] run deep), and a jump already requested a /// keyframe, so on the rare path that outruns it (a wedged consumer during the flush cooldown) dropping /// the OLDEST queued AU is safe — the pending IDR re-anchors decode regardless. Purely bounds memory. const FRAME_QUEUE_HARD_CAP: usize = 90; @@ -31,14 +34,15 @@ const FRAME_QUEUE_HARD_CAP: usize = 90; /// AUs and requests a keyframe) instead of playing that far behind forever. Deliberately generous — an /// interactive stream is unusable well before 400 ms, but the bound must sit safely above the skew /// handshake's own error (≈ RTT/2) plus normal delivery jitter so a healthy stream can never trip it. -/// This is the CLOCK-BASED detector; the clock-free [`QUEUE_HIGH`]/[`STANDING_FRAMES`] detector covers +/// This is the CLOCK-BASED detector; the clock-free [`QUEUE_HIGH`]/[`STANDING_TIME`] detector covers /// same-clock and no-handshake sessions (where `clock_offset_ns == 0` disarms this one). pub(crate) const FLUSH_LATENCY: Duration = Duration::from_millis(400); -/// How many CONSECUTIVE over-bound frames arm the clock-based jump (~0.5 s at 60 fps). A genuine -/// standing queue puts EVERY frame over the bound; a one-off burst (an IDR, a Wi-Fi scan blip) clears -/// within a frame or two and never reaches the count. -pub(crate) const FLUSH_AFTER_FRAMES: u32 = 30; +/// How long frames must run CONTINUOUSLY over the [`FLUSH_LATENCY`] bound before the clock-based +/// jump fires. WALL-CLOCK (latency plan T1.4, was a 30-frame count — same fps-scaling problem as +/// the standing-queue detector). A genuine standing queue puts EVERY frame over the bound; a +/// one-off burst (an IDR, a Wi-Fi scan blip) clears within a frame or two and resets the run. +pub(crate) const FLUSH_AFTER: Duration = Duration::from_millis(250); /// Minimum spacing between jump-to-live events, so a bottleneck that instantly rebuilds the queue (a /// link/consumer that can't sustain the bitrate at all) degrades into a periodic skip + a logged diff --git a/crates/punktfunk-core/src/client/pump.rs b/crates/punktfunk-core/src/client/pump.rs index 3c528ec4..f70e4b9e 100644 --- a/crates/punktfunk-core/src/client/pump.rs +++ b/crates/punktfunk-core/src/client/pump.rs @@ -1,8 +1,8 @@ //! The client worker: QUIC handshake + control/input/datagram tasks + the blocking data-plane pump. use super::frame_channel::{ - CLOCK_RESYNC_INTERVAL, FLUSH_AFTER_FRAMES, FLUSH_COOLDOWN, FLUSH_LATENCY, - NOOP_CLOCK_FLUSHES_TO_DISARM, NOOP_FLUSH_DATAGRAMS, QUEUE_HIGH, QUEUE_LOW, STANDING_FRAMES, + CLOCK_RESYNC_INTERVAL, FLUSH_AFTER, FLUSH_COOLDOWN, FLUSH_LATENCY, + NOOP_CLOCK_FLUSHES_TO_DISARM, NOOP_FLUSH_DATAGRAMS, QUEUE_HIGH, QUEUE_LOW, STANDING_TIME, }; use super::worker::reject_from_close; use super::*; @@ -711,12 +711,13 @@ pub(super) async fn run_pump(args: WorkerArgs) { let mut capacity_probe_deadline: Option = None; let (mut owd_sum_ns, mut owd_frames) = (0i128, 0u32); let mut flush_in_window = false; - // Jump-to-live state (see the guard in the loop below): the clock-based over-bound run - // (`stale_frames`, armed only when the skew handshake succeeded so the clocks are comparable), - // the clock-free non-draining-queue run (`standing_frames`), and the last-jump instant for the - // shared cooldown. - let mut stale_frames: u32 = 0; - let mut standing_frames: u32 = 0; + // Jump-to-live state (see the guard in the loop below): when the clock-based over-bound + // run began (`stale_since`, armed only when the skew handshake succeeded so the clocks + // are comparable), when the clock-free non-draining-queue run began (`standing_since`), + // and the last-jump instant for the shared cooldown. Wall-clock runs (T1.4), not frame + // counts — the detectors' sensitivity must not scale with fps or repeat cadence. + let mut stale_since: Option = None; + let mut standing_since: Option = None; let mut last_flush: Option = None; // Clock-detector health: consecutive clock-triggered flushes that found no local backlog // (see NOOP_FLUSH_DATAGRAMS). Reaching NOOP_CLOCK_FLUSHES_TO_DISARM turns the clock-based @@ -737,7 +738,7 @@ pub(super) async fn run_pump(args: WorkerArgs) { let gen = pump_clock_gen.load(Ordering::Relaxed); if gen != seen_clock_gen { seen_clock_gen = gen; - stale_frames = 0; + stale_since = None; noop_clock_flushes = 0; if !clock_detector_armed { clock_detector_armed = true; @@ -956,18 +957,18 @@ pub(super) async fn run_pump(args: WorkerArgs) { // FLUSH_COOLDOWN, both suspended during a speed test (the probe MEASURES a saturated // queue; flushing would corrupt its counters): // * clock-based — completed frames sit > FLUSH_LATENCY behind the skew-corrected - // capture clock for FLUSH_AFTER_FRAMES straight. Needs the skew handshake, and + // capture clock continuously for FLUSH_AFTER. Needs the skew handshake, and // also catches kernel/reassembler backlog the hand-off queue hasn't reached yet. // * clock-free — the pre-decode hand-off queue stopped draining: its depth stayed - // ≥ QUEUE_HIGH (never falling to QUEUE_LOW) for STANDING_FRAMES straight. Works - // with no handshake / a same-clock session (where the clock path is disarmed), - // and is the direct signal that the embedder can't keep up. A transient Wi-Fi - // clump drains in a few frames and never reaches the count. + // ≥ QUEUE_HIGH (never falling to QUEUE_LOW, still high at the trip) for + // STANDING_TIME. Works with no handshake / a same-clock session (where the + // clock path is disarmed), and is the direct signal that the embedder can't + // keep up. A transient Wi-Fi clump drains within ~100 ms and never trips it. if probe_active { // Keep both detectors disarmed across a speed test so its (deliberately) - // saturated queue doesn't leave a primed count that fires the moment it ends. - stale_frames = 0; - standing_frames = 0; + // saturated queue doesn't leave a primed run that fires the moment it ends. + stale_since = None; + standing_since = None; } else { let lat_ns = if clock_offset_ns != 0 { now_realtime_ns() + clock_offset_ns as i128 - frame.pts_ns as i128 @@ -985,23 +986,27 @@ pub(super) async fn run_pump(args: WorkerArgs) { && clock_offset_ns != 0 && lat_ns > FLUSH_LATENCY.as_nanos() as i128 { - stale_frames += 1; + stale_since.get_or_insert_with(Instant::now); } else { - stale_frames = 0; + stale_since = None; } let depth = frames.depth(); if depth >= QUEUE_HIGH { - standing_frames += 1; + standing_since.get_or_insert_with(Instant::now); } else if depth <= QUEUE_LOW { - standing_frames = 0; + standing_since = None; } - let clock_behind = stale_frames >= FLUSH_AFTER_FRAMES; - let queue_behind = standing_frames >= STANDING_FRAMES; + // The queue trip additionally requires the depth to still be high NOW, so + // a run that started ≥ high but is hovering in the hysteresis band (a + // clump mid-drain) never fires on elapsed time alone. + let clock_behind = stale_since.is_some_and(|t| t.elapsed() >= FLUSH_AFTER); + let queue_behind = depth >= QUEUE_HIGH + && standing_since.is_some_and(|t| t.elapsed() >= STANDING_TIME); if (clock_behind || queue_behind) && last_flush.is_none_or(|t| t.elapsed() >= FLUSH_COOLDOWN) { - stale_frames = 0; - standing_frames = 0; + stale_since = None; + standing_since = None; last_flush = Some(Instant::now()); flush_in_window = true; // strongest "link can't hold the rate" signal let flushed = session.flush_backlog().unwrap_or(0); diff --git a/crates/punktfunk-host/src/native/stream.rs b/crates/punktfunk-host/src/native/stream.rs index c6029167..6143d533 100644 --- a/crates/punktfunk-host/src/native/stream.rs +++ b/crates/punktfunk-host/src/native/stream.rs @@ -193,6 +193,14 @@ fn service_probes( } } +/// T1.1 frame-driven encode trigger (latency plan): `PUNKTFUNK_FRAME_DRIVEN=0` restores the +/// legacy fixed-cadence tick everywhere (backends without an arrival wait keep it regardless — +/// see [`pf_capture::Capturer::supports_arrival_wait`]). +fn frame_driven_enabled() -> bool { + static ON: std::sync::OnceLock = std::sync::OnceLock::new(); + *ON.get_or_init(|| std::env::var("PUNKTFUNK_FRAME_DRIVEN").as_deref() != Ok("0")) +} + /// Seal one access unit and send it with MICROBURST pacing (the shared /// [`send_pacing`](crate::send_pacing) policy, native parameterization): the first `burst_cap` /// bytes go out immediately (one absorbed burst the NIC / socket tx-buffer can swallow), and @@ -1784,7 +1792,14 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option std::thread::sleep(d), - None => next = std::time::Instant::now(), + if frame_driven_enabled() && capturer.supports_arrival_wait() { + // T1.1 frame-driven trigger: instead of sleeping out the whole tick and then + // SAMPLING (which holds a frame that arrived just after the previous sample for up + // to a full interval — ~half on average), sleep only to the rate floor and then + // wake on the capture's actual arrival. The 0.9×interval floor caps the encode + // rate at ~1.11× target when the source runs faster (compositor Hz > session fps); + // the +0.5×interval keepalive keeps a static desktop re-encoding (bitrate shape, + // client liveness) at 1.5×interval cadence and bounds control-servicing latency. + let earliest = next - interval.mul_f32(0.1); + if let Some(d) = earliest.checked_duration_since(std::time::Instant::now()) { + std::thread::sleep(d); + } + capturer.wait_arrival(next + interval.mul_f32(0.5)); + } else { + match next.checked_duration_since(std::time::Instant::now()) { + Some(d) => std::thread::sleep(d), + None => next = std::time::Instant::now(), + } } } // Drain the in-flight tail (the depth-1 frames submitted but not yet polled) so the last frames still