perf(encode/nvenc-linux): LN3 — pipelined-retrieve escalation replaces the depth-1 async foot-gun

PUNKTFUNK_NVENC_ASYNC gains a real tri-state: 1 = always (as before, now
documented as ~+1 tick at depth-1), 0 = never (vetoes escalation), unset =
ADAPTIVE — off until the session loop's cadence-overrun detector escalates.

The host loop's adaptive-depth leaky bucket grows a second stage: once the
capturer's depth is maxed (Linux portal is permanently depth-1), it asks
the encoder for pipelined retrieve via the new Encoder::set_pipelined hook
(asked exactly once; default impl declines, Windows untouched).

nvenc_cuda engages at a safe point via a clean session rebuild WITHOUT the
IO-stream binding: with input==output stream bound, later stream work
waits on prior encode completions and would serialize a pipelined session
— stream-ordered submit and two-thread retrieve are mutually exclusive.
The ordered gate now also requires async_rt absence (belt-and-braces for
the runtime switch). Re-open's first frame is the standard session IDR.

On-hardware test: escalate mid-session → retrieve thread live, binding
gone, all AUs deliver, first post-escalation AU is the IDR.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

crates/pf-encode/src/enc/codec.rs

@@ -269,6 +269,15 @@ pub trait Encoder: Send {
     fn invalidate_ref_frames(&mut self, _first_frame: i64, _last_frame: i64) -> bool {
         false
     }
+    /// Escalate into a pipelined (two-thread) retrieve mode under sustained GPU contention — the
+    /// 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.
+    fn set_pipelined(&mut self, _on: bool) -> bool {
+        false
+    }
     /// Pull the next encoded AU if one is ready.
     fn poll(&mut self) -> Result<Option<EncodedFrame>>;
     /// Tear the underlying hardware encoder down and rebuild it in place, keeping the session's

crates/pf-encode/src/enc/linux/nvenc_cuda.rs

@@ -25,8 +25,11 @@
 //!     with NO per-frame `cuStreamSynchronize` and the encode orders after them on the stream —
 //!     the submit path's CPU stalls are gone even though the copy itself remains.
 //!
-//! **Two-thread retrieve** (`PUNKTFUNK_NVENC_ASYNC=1`, the same opt-in knob as the Windows
+//! **Two-thread retrieve** (`PUNKTFUNK_NVENC_ASYNC`: `1` = always, `0` = never, unset =
-//! backend — gpu-contention plan §5.B, latency plan T2.2): NVENC *async mode*
+//! **adaptive** — engaged by the session loop's contention escalation via
+//! [`Encoder::set_pipelined`] when depth-1 can't hold cadence; at depth-1 it costs ~one loop
+//! tick of latency, which is why it is not simply on. gpu-contention plan §5.B, latency plan
+//! T2.2/§7 LN3): NVENC *async mode*
 //! (`enableEncodeAsync` + completion events) is Windows-only, so the session here stays SYNC —
 //! but the NVENC guide's threading model still applies: the main thread should only *submit*
 //! while a secondary thread does the (blocking) `nvEncLockBitstream`. With the flag set, an
@@ -237,15 +240,25 @@ fn load_api() -> std::result::Result<EncodeApi, String> {
 /// bitstream/ring slot is never reused mid-encode.
 const POOL: usize = 8;
 
-/// Whether the operator asked for the two-thread retrieve (`PUNKTFUNK_NVENC_ASYNC` truthy — the
+/// The operator's `PUNKTFUNK_NVENC_ASYNC` intent (the SAME knob as the Windows backend):
-/// SAME knob as the Windows backend, so one env drives the split on either host OS). Opt-in
+/// `Some(true)` = force the two-thread retrieve from session open — note that at the Linux
-/// until on-glass validated. Unlike Windows this changes NO session parameter (Linux stays sync
+/// default pipeline depth of 1 this adds ~one loop tick of latency (the non-blocking poll's AU
-/// mode; only the blocking lock moves off the encode thread), so there is no async-rejecting
+/// rides the next tick), so it only pays under GPU contention; `Some(false)` = never (also
-/// config to fail the open.
+/// vetoes the session loop's contention escalation via [`Encoder::set_pipelined`]); `None`
+/// (unset) = adaptive — off until the session loop escalates on sustained cadence overrun.
+/// Unlike Windows this changes NO session parameter (Linux stays sync mode; only the blocking
+/// lock moves off the encode thread), so there is no async-rejecting config to fail the open.
+fn async_retrieve_env() -> Option<bool> {
+    match std::env::var("PUNKTFUNK_NVENC_ASYNC") {
+        Ok(v) if matches!(v.trim(), "1" | "true" | "yes" | "on") => Some(true),
+        Ok(v) if matches!(v.trim(), "0" | "false" | "no" | "off") => Some(false),
+        _ => None,
+    }
+}
+
+/// Operator forced the two-thread retrieve on from session open (see [`async_retrieve_env`]).
 fn async_retrieve_requested() -> bool {
-    std::env::var("PUNKTFUNK_NVENC_ASYNC")
+    async_retrieve_env() == Some(true)
-        .map(|v| matches!(v.trim(), "1" | "true" | "yes" | "on"))
-        .unwrap_or(false)
 }
 
 /// Max encodes in flight in two-thread mode (`PUNKTFUNK_NVENC_ASYNC_DEPTH`, default 4, clamped
@@ -486,6 +499,11 @@ pub struct NvencCudaEncoder {
     /// The two-thread retrieve runtime (`PUNKTFUNK_NVENC_ASYNC`) — `None` in the default
     /// single-thread mode and between sessions. Exists only `init_session`→`teardown`.
     async_rt: Option<AsyncRetrieve>,
+    /// The session loop escalated into pipelined retrieve ([`Encoder::set_pipelined`], the
+    /// contention analog of the capturer depth escalation). Sticky across session rebuilds
+    /// (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,
     /// 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
@@ -573,11 +591,35 @@ impl NvencCudaEncoder {
             split_mode: nv::NV_ENC_SPLIT_ENCODE_MODE::NV_ENC_SPLIT_DISABLE_MODE as u32,
             last_rfi_range: None,
             async_rt: None,
+            want_async: false,
             io_stream: ptr::null_mut(),
             stream_ordered: false,
         })
     }
 
+    /// Engage the escalated pipelined retrieve at a safe point: nothing in flight, and — because
+    /// a live session has its IO streams bound for stream-ordered submit, whose output-stream
+    /// semantics would make every later stream op wait on the previous encode and so serialize a
+    /// pipelined session — via a clean session rebuild WITHOUT the binding (the re-open's first
+    /// frame is the standard session-opening IDR). No-op until [`want_async`](Self::want_async)
+    /// is set and `pending` drains.
+    fn maybe_engage_async(&mut self) {
+        if !self.want_async || self.async_rt.is_some() || !self.pending.is_empty() {
+            return;
+        }
+        if self.inited {
+            // SAFETY: encode thread, `pending` empty ⇒ no encode in flight; `teardown` handles
+            // exactly this live-session state (and a torn-down encoder lazily re-inits on the
+            // next submit, which spawns the retrieve thread and skips the IO-stream arming).
+            unsafe { self.teardown() };
+            tracing::info!(
+                "NVENC pipelined-retrieve escalation: rebuilding the session without the \
+                 IO-stream binding (stream-ordered submit and two-thread retrieve are mutually \
+                 exclusive); 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() {
@@ -1035,10 +1077,11 @@ impl NvencCudaEncoder {
             self.inited = true;
             // Two-thread retrieve (T2.2): spawn the lock thread against the live session. No
             // session parameter differs — teardown/rebuild always stops it before destroy.
-            if async_retrieve_requested() {
+            if async_retrieve_requested() || self.want_async {
                 self.async_rt = Some(AsyncRetrieve::spawn(self.encoder as usize));
                 tracing::info!(
                     depth = async_inflight_cap(),
+                    escalated = self.want_async,
                     "NVENC two-thread retrieve enabled (submit thread + blocking-lock thread)"
                 );
             }
@@ -1168,6 +1211,9 @@ 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).
+        self.maybe_engage_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);
@@ -1247,7 +1293,10 @@ impl Encoder for NvencCudaEncoder {
         // `poll` (both host loops do — see `Encoder::submit`'s doc), which blocks until the encode
         // (and therefore the enqueued copy) completed. A pipelined caller (pending non-empty)
         // falls back to the blocking copy so an early-recycled source can never be read late.
-        let ordered = self.stream_ordered && self.pending.is_empty();
+        // `async_rt` must be absent too: in two-thread mode the frame may be recycled right after
+        // submit returns while the stream still holds its copy (belt-and-braces — an escalated
+        // session was rebuilt without the binding, so `stream_ordered` is false there anyway).
+        let ordered = self.stream_ordered && self.async_rt.is_none() && self.pending.is_empty();
         let t0 = std::time::Instant::now();
 
         // Copy the captured buffer into this slot's input surface before encoding it.
@@ -1455,6 +1504,21 @@ impl Encoder for NvencCudaEncoder {
         self.force_kf = true;
     }
 
+    fn set_pipelined(&mut self, on: bool) -> bool {
+        if !on {
+            // v1 is escalate-and-hold (no de-escalation), mirroring the depth escalation.
+            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
+        }
+        if !self.want_async && self.async_rt.is_none() {
+            self.want_async = true;
+            self.maybe_engage_async();
+        }
+        true
+    }
+
     fn caps(&self) -> EncoderCaps {
         EncoderCaps {
             supports_rfi: self.rfi_supported,
@@ -2090,6 +2154,75 @@ mod tests {
         );
     }
 
+    /// ON-HARDWARE (RTX box `.21`): the §7 LN3 pipelined-retrieve escalation —
+    /// `set_pipelined(true)` on a live sync session must rebuild it without the IO-stream
+    /// binding, spawn the retrieve thread on the re-open, and keep delivering AUs (the first
+    /// post-escalation AU is the re-open's session-opening IDR; pipelined `poll` is
+    /// non-blocking, so AUs may ride a later tick).
+    #[test]
+    #[ignore = "requires an NVIDIA GPU + driver — run manually on the RTX box (.21)"]
+    fn nvenc_cuda_pipelined_escalation() {
+        const W: u32 = 1280;
+        const H: u32 = 720;
+        pf_zerocopy::cuda::make_current().expect("shared CUDA context current");
+        if async_retrieve_env() == Some(false) {
+            println!("skipped: PUNKTFUNK_NVENC_ASYNC=0 vetoes the escalation");
+            return;
+        }
+        let mut enc = NvencCudaEncoder::open(
+            Codec::H265,
+            PixelFormat::Nv12,
+            W,
+            H,
+            60,
+            8_000_000,
+            true,
+            8,
+            ChromaFormat::Yuv420,
+        )
+        .expect("open NVENC CUDA session");
+        // Steady sync frames first (stream-ordered mode).
+        for i in 0..3u32 {
+            let frame = nv12_frame(W, H, i);
+            enc.submit_indexed(&frame, i).expect("submit");
+            enc.poll().expect("poll").expect("AU");
+        }
+        assert!(enc.async_rt.is_none(), "session starts sync");
+        assert!(enc.set_pipelined(true), "escalation must be accepted");
+        let mut aus = 0usize;
+        let mut first_key = false;
+        for i in 3..13u32 {
+            let frame = nv12_frame(W, H, i);
+            enc.submit_indexed(&frame, i)
+                .expect("submit post-escalation");
+            while let Some(au) = enc.poll().expect("poll") {
+                if aus == 0 {
+                    first_key = au.keyframe;
+                }
+                aus += 1;
+            }
+            std::thread::sleep(std::time::Duration::from_millis(3));
+        }
+        // Drain the pipelined tail (bounded).
+        let deadline = std::time::Instant::now() + std::time::Duration::from_millis(500);
+        while aus < 10 && std::time::Instant::now() < deadline {
+            if enc.poll().expect("poll").is_some() {
+                aus += 1;
+            }
+            std::thread::sleep(std::time::Duration::from_millis(1));
+        }
+        assert!(
+            enc.async_rt.is_some(),
+            "retrieve thread must be live after escalation"
+        );
+        assert!(
+            !enc.stream_ordered,
+            "IO-stream binding must be gone in pipelined mode"
+        );
+        assert_eq!(aus, 10, "every post-escalation frame must deliver an AU");
+        assert!(first_key, "first post-escalation AU is the re-open IDR");
+    }
+
     /// ON-HARDWARE (RTX box `.21`), MEASUREMENT probe for latency plan §7 LN1 — answers the
     /// go/no-go question for sub-frame slice output: with `PUNKTFUNK_NVENC_SLICES=4` +
     /// `PUNKTFUNK_NVENC_SUBFRAME=1`, do slices become READABLE incrementally while the frame is

crates/punktfunk-host/src/native/stream.rs

@@ -1207,6 +1207,9 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
     let mut cur_depth: usize = 1;
     let mut behind_score: u32 = 0;
     let mut depth_frames: u64 = 0;
+    // Second escalation stage (§7 LN3): once depth is maxed (or was never available — Linux),
+    // ask the encoder for pipelined retrieve exactly once. Latched whether it accepts or not.
+    let mut pipeline_asked = false;
     // ~20 net behind-frames (≈0.3 s sustained) escalates; a lone hitch decays away. Warmup skips
     // the first ~1 s so bring-up (display acquire, encoder open) never triggers it.
     const DEPTH_ESCALATE: u32 = 20;
@@ -2056,10 +2059,15 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
         // Adaptive-depth escalate signal (measured BEFORE the trailing sleep): "behind" = the
         // frame's work overran its cadence deadline `next`, so the trailing sleep would be
         // zero/negative. At depth-1 that means the synchronous poll (encode + WDDM wait) can't
-        // fit a frame interval — the contention case pipelining is for — so escalate to the
+        // fit a frame interval — the contention case pipelining is for — so escalate, and hold
-        // capturer's max and hold there. Leaky bucket + warmup skip reject one-off hitches and
+        // there. Leaky bucket + warmup skip reject one-off hitches and bring-up; no
-        // bring-up. Once escalated, `cur_depth` stays (no de-escalation in v1).
+        // de-escalation in v1. Two stages: first the CAPTURER's max depth (Windows IDD depth-2
-        if idd_adaptive_enabled() && cur_depth < max_depth {
+        // overlap); where depth can't grow (Linux portal is permanently depth-1, §7 LN3), the
+        // ENCODER's pipelined retrieve is the same trade on the other side of submit — the
+        // two-thread lock moves the encode wait off this loop so capture/submit keep cadence,
+        // at ~one tick of AU latency. `enc.set_pipelined` may decline (unsupported backend or
+        // an explicit PUNKTFUNK_NVENC_ASYNC=0); either way it is asked exactly once.
+        if idd_adaptive_enabled() && (cur_depth < max_depth || !pipeline_asked) {
             depth_frames += 1;
             if depth_frames > DEPTH_WARMUP_FRAMES {
                 let behind = std::time::Instant::now() >= next;
@@ -2069,6 +2077,7 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
                     behind_score.saturating_sub(1)
                 };
                 if behind_score >= DEPTH_ESCALATE {
+                    if cur_depth < max_depth {
                         cur_depth = max_depth;
                         tracing::info!(
                             depth = cur_depth,
@@ -2076,6 +2085,19 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
                              (GPU contention); pipelining for the rest of the session (latency \
                              trade for throughput)"
                         );
+                    } else {
+                        pipeline_asked = true;
+                        if enc.set_pipelined(true) {
+                            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 \
+                                 for throughput)"
+                            );
+                        }
+                    }
+                    // Give the action time to take effect before judging again.
+                    behind_score = 0;
                 }
             }
         }