From 78fe77b049b97d205af7269fa9505d7ee66bf374 Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Sat, 25 Jul 2026 01:30:39 +0200 Subject: [PATCH] feat(host/encode): de-escalate the latency escalation once cadence holds clean MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Escalate-and-hold's missing half. The contention escalation (capture depth, then the NVENC pipelined retrieve) was permanent: one sustained overrun — even one CAUSED by the ABR overdrive's rebuild storms — cost the session its depth-1 latency and its sub-frame streaming forever, and `encode_us` reported queue depth instead of ASIC time for the rest of the session. - The leaky bucket now keeps scoring after escalation. A sustained every-frame-on-cadence run (~5 s at 120 fps) winds back one stage in reverse order: pipelined retrieve first (its rebuild restores the IO-stream binding and sub-frame chunked streaming), then capture depth back to 1. Attempts are paced by an exponential backoff (1 → 5 → 25 min, capped) — a workload that truly needs the escalation converges to keeping it, but never a permanent latch. - NVENC (Linux) implements `set_pipelined(false)`: a `want_sync` latch handled at the same drained safe point as the engage side (`maybe_disengage_async` mirrors `maybe_engage_async`); the lazy sync re-init re-arms everything and opens on an IDR. The stream loop polls until the switch lands, then re-runs the escalation warmup so the wind-back's own stall can't re-escalate it. `PUNKTFUNK_NVENC_ASYNC=1` (operator-pinned async) refuses the wind-back; the trait doc now specifies the two-way contract. - While escalated, `cadence_degraded` stays latched (bitrate climbs refused) even with the bucket drained: the headroom is spent, and climbs resuming mid-escalation would saw against it and starve the clean run. Co-Authored-By: Claude Fable 5 --- crates/pf-encode/src/enc/codec.rs | 11 ++- crates/pf-encode/src/enc/linux/nvenc_cuda.rs | 50 ++++++++++- crates/punktfunk-host/src/native/stream.rs | 92 +++++++++++++++++--- 3 files changed, 138 insertions(+), 15 deletions(-) diff --git a/crates/pf-encode/src/enc/codec.rs b/crates/pf-encode/src/enc/codec.rs index 1487b595..1b8297b4 100644 --- a/crates/pf-encode/src/enc/codec.rs +++ b/crates/pf-encode/src/enc/codec.rs @@ -352,8 +352,15 @@ pub trait Encoder: Send { /// encoder analog of the capturer depth escalation: AUs ride ~one loop tick behind (`poll` /// may return `None` while an encode is in flight) in exchange for capture/submit no longer /// serializing on the encode wait. Returns whether pipelined retrieve is (now) active; the - /// switch may be deferred to the next safe point internally. `false` from the default impl = - /// unsupported — the session loop stops asking. De-escalation is a v2 item everywhere. + /// switch may be deferred to the next safe point internally. `set_pipelined(true)` returning + /// `false` (the default impl) = unsupported — the session loop stops asking. + /// + /// `set_pipelined(false)` requests the wind-back (de-escalation, latency recovery): the + /// backend restores its sync-retrieve mode — and the latency features that mode carries + /// (IO-stream binding, sub-frame chunking) — at its next safe point, usually via a session + /// rebuild whose first frame is an IDR. The return is still "is pipelined retrieve active": + /// the caller polls until it reads `false`. Backends that never escalate return `false` + /// trivially. An operator pin (`PUNKTFUNK_NVENC_ASYNC=1`) refuses the wind-back. fn set_pipelined(&mut self, _on: bool) -> bool { false } diff --git a/crates/pf-encode/src/enc/linux/nvenc_cuda.rs b/crates/pf-encode/src/enc/linux/nvenc_cuda.rs index 0c3ef438..a7011e6e 100644 --- a/crates/pf-encode/src/enc/linux/nvenc_cuda.rs +++ b/crates/pf-encode/src/enc/linux/nvenc_cuda.rs @@ -596,6 +596,12 @@ pub struct NvencCudaEncoder { /// (escalate-and-hold, like the depth escalation); the switch itself happens at the next /// safe point via [`maybe_engage_async`](Self::maybe_engage_async). want_async: bool, + /// A de-escalation request ([`Encoder::set_pipelined(false)`]) waiting for its safe point: + /// the next drained moment tears the session down and lazily re-inits SYNC (IO-stream + /// binding and sub-frame chunking re-arm at that re-init). Distinct from `!want_async` — + /// an operator-forced async session (`PUNKTFUNK_NVENC_ASYNC=1`) also has `want_async` + /// false, and a de-escalation must never tear THAT down. + want_sync: bool, /// Boxed `CUstream` the session's IO-stream binding points at (`NvEncSetIOCudaStreams` takes /// POINTERS to `CUstream`, and this struct moves — the pointee needs a stable heap address for /// the session's lifetime). Null when stream-ordering is off; freed in `teardown` AFTER the @@ -703,6 +709,7 @@ impl NvencCudaEncoder { last_rfi_range: None, async_rt: None, want_async: false, + want_sync: false, io_stream: ptr::null_mut(), stream_ordered: false, slices: 1, @@ -736,6 +743,29 @@ impl NvencCudaEncoder { } } + /// [`maybe_engage_async`](Self::maybe_engage_async)'s inverse — wind the escalated pipelined + /// retrieve back at a safe point: nothing in flight, then a clean session rebuild whose lazy + /// SYNC re-init restores the IO-stream binding and re-arms sub-frame chunking (the two + /// latency features the escalation traded away). No-op until + /// [`want_sync`](Self::want_sync) is set and `pending` drains. + fn maybe_disengage_async(&mut self) { + if !self.want_sync || self.async_rt.is_none() || !self.pending.is_empty() { + return; + } + self.want_sync = false; + if self.inited { + // SAFETY: encode thread, `pending` empty ⇒ no encode in flight (and nothing queued + // to the retrieve thread); `teardown` joins the retrieve thread and handles exactly + // this live-session state — the next submit lazily re-inits sync. + unsafe { self.teardown() }; + tracing::info!( + "NVENC pipelined-retrieve de-escalation: rebuilding the session with the sync \ + retrieve (IO-stream binding and sub-frame chunking restored); next frame opens \ + with an IDR" + ); + } + } + /// Tear down the encode session + pooled resources. Reused on a size change and at Drop. unsafe fn teardown(&mut self) { if self.encoder.is_null() { @@ -1471,9 +1501,10 @@ impl Encoder for NvencCudaEncoder { "Linux direct-NVENC needs a CUDA frame (FramePayload::Cuda); got a CPU/dmabuf frame" ), }; - // A pending pipelined-retrieve escalation engages here, at the submit-side safe point - // (nothing in flight after the previous poll drained). + // A pending pipelined-retrieve escalation — or de-escalation — engages here, at the + // submit-side safe point (nothing in flight after the previous poll drained). self.maybe_engage_async(); + self.maybe_disengage_async(); // Re-init on a size change (the capturer can return at a different resolution after a mode // switch). Format changes (NV12↔YUV444) likewise re-init. let new_fmt = buffer_format(buf); @@ -1781,12 +1812,25 @@ impl Encoder for NvencCudaEncoder { fn set_pipelined(&mut self, on: bool) -> bool { if !on { - // v1 is escalate-and-hold (no de-escalation), mirroring the depth escalation. + // De-escalation (the v2 of escalate-and-hold): latch the wind-back intent; the + // switch itself happens at the next drained safe point + // ([`maybe_disengage_async`](Self::maybe_disengage_async)) — the caller polls + // this same method until it reports inactive. + if async_retrieve_env() == Some(true) { + // Operator pinned async on — de-escalation must not undo an explicit choice. + return self.want_async || self.async_rt.is_some(); + } + if self.want_async || self.async_rt.is_some() { + self.want_async = false; + self.want_sync = true; + self.maybe_disengage_async(); + } return self.want_async || self.async_rt.is_some(); } if async_retrieve_env() == Some(false) { return false; // operator veto: PUNKTFUNK_NVENC_ASYNC=0 means NEVER } + self.want_sync = false; // latest intent wins — cancel a pending wind-back if !self.want_async && self.async_rt.is_none() { self.want_async = true; self.maybe_engage_async(); diff --git a/crates/punktfunk-host/src/native/stream.rs b/crates/punktfunk-host/src/native/stream.rs index a84f4e0f..07019a72 100644 --- a/crates/punktfunk-host/src/native/stream.rs +++ b/crates/punktfunk-host/src/native/stream.rs @@ -1457,6 +1457,22 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option = None; + let mut deescalate_backoff = DEESCALATE_BACKOFF_START; while !stop.load(Ordering::SeqCst) && std::time::Instant::now() < deadline { // Mid-stream session switch (the box flipped Gaming↔Desktop): rebuild the WHOLE backend in // place — a different compositor at the SAME client mode — keeping the Session + send thread @@ -1794,6 +1810,7 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option { tracing::warn!(error = %format!("{e:#}"), to_kbps = new_kbps, @@ -2618,7 +2635,7 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option DEPTH_WARMUP_FRAMES { let behind = std::time::Instant::now() >= next; @@ -2627,34 +2644,89 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option 1 || pipelined_active || deescalating; // Export "encode can't hold cadence" for the control task's climb refusal. - // Stored BEFORE the escalate check: the firing iteration writes `true`, and a - // final-stage escalation freezes this whole block (the guard above goes false) - // — so an escalated session stays flagged, which is exactly right: its climb - // headroom is spent until something (a down-step, de-escalation) changes. - cadence_degraded.store(behind_score >= DEPTH_DEGRADE, Ordering::Relaxed); - if behind_score >= DEPTH_ESCALATE { + // An escalated session stays flagged even with the bucket drained: its climb + // headroom is spent, and letting climbs resume would saw against the + // escalation and starve the de-escalation clean run below. + cadence_degraded.store( + escalated || behind_score >= DEPTH_DEGRADE, + Ordering::Relaxed, + ); + if deescalating { + // A requested wind-back completes at the encoder's drained safe point — + // poll it (the call is a cheap latch check until then). + if !enc.set_pipelined(false) { + deescalating = false; + pipelined_active = false; + // Re-arm the ask: a future sustained overrun may escalate again (the + // backoff below paces how soon another wind-back may follow it). + pipeline_asked = false; + tracing::info!( + "encoder pipelined retrieve de-escalated — sync retrieve (and \ + sub-frame streaming, where armed) restored; re-monitoring cadence" + ); + // The wind-back rebuild's own stall must not re-escalate on the spot. + behind_score = 0; + depth_frames = 0; + ahead_run = 0; + } + } else if behind_score >= DEPTH_ESCALATE + && (cur_depth < max_depth || !pipeline_asked) + { if cur_depth < max_depth { cur_depth = max_depth; tracing::info!( depth = cur_depth, "IDD pipeline depth escalated — encode can't hold cadence at depth-1 \ - (GPU contention); pipelining for the rest of the session (latency \ + (GPU contention); pipelining until cadence holds clean (latency \ trade for throughput)" ); } else { pipeline_asked = true; - if enc.set_pipelined(true) { + pipelined_active = enc.set_pipelined(true); + if pipelined_active { tracing::info!( "encoder pipelined retrieve escalated — encode can't hold \ cadence and the capturer has no depth to give; the encode wait \ - moves off the loop for the rest of the session (latency trade \ + moves off the loop until cadence holds clean (latency trade \ for throughput)" ); } } // Give the action time to take effect before judging again. behind_score = 0; + ahead_run = 0; + } else if escalated { + // De-escalation: a sustained every-frame-on-cadence run at the escalated + // setting is the evidence the contention passed (a lower ABR rate, the + // game scene lightened) — wind back in reverse order, paced by the + // exponential backoff (see the consts above). + ahead_run = if behind { 0 } else { ahead_run + 1 }; + if ahead_run >= DEESCALATE_CLEAN_FRAMES + && deescalate_not_before.is_none_or(|t| std::time::Instant::now() >= t) + { + ahead_run = 0; + deescalate_not_before = + Some(std::time::Instant::now() + deescalate_backoff); + deescalate_backoff = (deescalate_backoff * 5).min(DEESCALATE_BACKOFF_MAX); + if pipelined_active { + tracing::info!( + "cadence held clean while escalated — winding the pipelined \ + retrieve back (latency recovery; costs one IDR)" + ); + deescalating = true; + } else if cur_depth > 1 { + cur_depth = 1; + tracing::info!( + depth = cur_depth, + "IDD pipeline depth de-escalated — cadence held clean at the \ + escalated depth (latency recovery)" + ); + behind_score = 0; + depth_frames = 0; + } + } } } }