From 7cea893db5a261504d98df6ca16c9b44228c508f Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Sat, 11 Jul 2026 19:11:01 +0200 Subject: [PATCH] feat(recovery): wire LTR-RFI loss recovery into every client Centralize the client-side loss-range detector in punktfunk-core so every embedder shares one implementation instead of re-deriving the wrapping frame-index arithmetic: - NativeClient::note_frame_index(frame_index) folds each received AU (in receive order) through RfiRecovery::observe, firing a throttled RFI request for the exact lost span [first_missing, frame_index-1] on a forward gap. A host that can RFI (AMD LTR / NVENC) re-references a known-good frame instead of paying a 20-40x IDR spike; the frames_dropped-driven keyframe path stays the backstop for when the recovery frame itself is lost. - Export request_rfi + note_frame_index over the C ABI (Apple client). - Call it from the Android (hw+sw pumps), Apple (StreamPump + Stage2Pipeline via PunktfunkConnection.noteFrameIndex), and Windows in-process pumps. Linux/Deck inherit it through pf-client-core's session pump. - Split the decision into a pure RfiRecovery::observe(frame_index, now) and add 8 unit tests: arming, contiguous runs, exact lost-range, single-frame drop, the 100ms throttle (burst-suppress then re-open), reorder stragglers, and u32 wraparound (contiguous + gap-range). Co-Authored-By: Claude Opus 4.8 --- clients/android/native/src/decode.rs | 9 + .../Connection/PunktfunkConnection.swift | 14 ++ .../PunktfunkKit/Video/Stage2Pipeline.swift | 5 + .../PunktfunkKit/Video/StreamPump.swift | 5 + clients/windows/src/session.rs | 5 + crates/punktfunk-core/src/abi.rs | 64 ++++++ crates/punktfunk-core/src/client.rs | 186 ++++++++++++++++++ include/punktfunk_core.h | 37 ++++ 8 files changed, 325 insertions(+) diff --git a/clients/android/native/src/decode.rs b/clients/android/native/src/decode.rs index ecfd52b9..72cb923f 100644 --- a/clients/android/native/src/decode.rs +++ b/clients/android/native/src/decode.rs @@ -243,6 +243,11 @@ fn run_sync( if pending.is_none() { match client.next_frame(Duration::from_millis(5)) { Ok(frame) => { + // Loss recovery (RFI): feed the frame index so a forward gap fires a throttled + // reference-frame-invalidation request — an RFI-capable host (AMD LTR / NVENC) + // recovers with a cheap clean P-frame instead of a full IDR. The frames_dropped + // keyframe path below stays the backstop when the recovery frame itself is lost. + let _ = client.note_frame_index(frame.frame_index); if fed == 0 { let p = &frame.data; log::info!( @@ -1026,6 +1031,10 @@ fn feeder_loop( while !shutdown.load(Ordering::Relaxed) { match client.next_frame(Duration::from_millis(5)) { Ok(frame) => { + // Loss recovery (RFI): a forward frame-index gap fires a throttled reference-frame- + // invalidation request so an RFI-capable host recovers with a cheap clean P-frame + // instead of a full IDR (the frames_dropped keyframe path is the backstop). + let _ = client.note_frame_index(frame.frame_index); if stats.enabled() { let received_ns = now_realtime_ns(); let clock_offset = clock_offset.load(Ordering::Relaxed) as i128; diff --git a/clients/apple/Sources/PunktfunkKit/Connection/PunktfunkConnection.swift b/clients/apple/Sources/PunktfunkKit/Connection/PunktfunkConnection.swift index 627f8169..6975fa30 100644 --- a/clients/apple/Sources/PunktfunkKit/Connection/PunktfunkConnection.swift +++ b/clients/apple/Sources/PunktfunkKit/Connection/PunktfunkConnection.swift @@ -445,6 +445,20 @@ public final class PunktfunkConnection { _ = punktfunk_connection_request_keyframe(h) } + /// Feed each received AU's `frameIndex` (in receive order) so the client recovers from loss with a + /// cheap reference-frame invalidation instead of always paying for a full IDR. On a forward gap — + /// a `frameIndex` jump means the intervening frames were lost and the following AUs reference a + /// picture that never arrived — the core fires a THROTTLED RFI request for the lost range, and an + /// RFI-capable host (AMD LTR / NVENC) recovers with a clean P-frame rather than a 20-40× IDR + /// spike. Call it for every received AU; the `framesDropped`-driven `requestKeyframe()` path stays + /// the backstop for when the recovery frame itself is lost. Cheap; silently dropped after close. + public func noteFrameIndex(_ frameIndex: UInt32) { + abiLock.lock() + defer { abiLock.unlock() } + guard let h = handle, !closeRequested else { return } + _ = punktfunk_connection_note_frame_index(h, frameIndex, nil) + } + /// Cumulative access units the host→client reassembler dropped as unrecoverable (FEC couldn't /// rebuild them). The video pump polls this and calls `requestKeyframe()` when it climbs — the /// correct loss trigger under the host's infinite GOP, where unrecoverable loss yields diff --git a/clients/apple/Sources/PunktfunkKit/Video/Stage2Pipeline.swift b/clients/apple/Sources/PunktfunkKit/Video/Stage2Pipeline.swift index 68d20084..bdaa6203 100644 --- a/clients/apple/Sources/PunktfunkKit/Video/Stage2Pipeline.swift +++ b/clients/apple/Sources/PunktfunkKit/Video/Stage2Pipeline.swift @@ -388,6 +388,11 @@ public final class Stage2Pipeline { presenter.setHdrMeta(meta) } guard let au = try connection.nextAU(timeoutMs: 100) else { return true } + // Loss recovery (RFI): a forward frame-index gap fires a throttled reference- + // frame-invalidation request so an RFI-capable host (AMD LTR / NVENC) recovers + // with a cheap clean P-frame instead of a full IDR. The framesDropped-driven + // recovery below stays the backstop for when the recovery frame itself is lost. + connection.noteFrameIndex(au.frameIndex) onFrame?(au) if let f = connection.videoCodec.formatDescription(fromKeyframe: au.data) { format = f // refreshed on every IDR (mode changes included) diff --git a/clients/apple/Sources/PunktfunkKit/Video/StreamPump.swift b/clients/apple/Sources/PunktfunkKit/Video/StreamPump.swift index c19ebb7e..70b9f03b 100644 --- a/clients/apple/Sources/PunktfunkKit/Video/StreamPump.swift +++ b/clients/apple/Sources/PunktfunkKit/Video/StreamPump.swift @@ -79,6 +79,11 @@ final class StreamPump { if awaitingIDR { recovery.request() } guard let au = try connection.nextAU(timeoutMs: 100) else { return true } + // Loss recovery (RFI): a forward frame-index gap fires a throttled reference- + // frame-invalidation request so an RFI-capable host (AMD LTR / NVENC) recovers + // with a cheap clean P-frame instead of a full IDR. The framesDropped-driven + // recovery above stays the backstop for when the recovery frame itself is lost. + connection.noteFrameIndex(au.frameIndex) onFrame?(au) let idrFormat = connection.videoCodec.formatDescription(fromKeyframe: au.data) if let f = idrFormat { diff --git a/clients/windows/src/session.rs b/clients/windows/src/session.rs index 6f4c7465..033d2078 100644 --- a/clients/windows/src/session.rs +++ b/clients/windows/src/session.rs @@ -380,6 +380,11 @@ fn pump( Ok(frame) => { // The `received` point: AU fully reassembled, handed to us, before decode. let received_ns = now_ns(); + // Loss recovery (RFI): a forward frame-index gap fires a throttled reference-frame- + // invalidation request so an RFI-capable host (AMD LTR / NVENC) recovers with a cheap + // clean P-frame instead of a full IDR. The frames_dropped keyframe path below is the + // backstop for when the recovery frame itself is lost. + let _ = connector.note_frame_index(frame.frame_index); // fps = AUs received per second, Mb/s = received goodput (spec: counted at the // received point, not the decoded one). frames_n += 1; diff --git a/crates/punktfunk-core/src/abi.rs b/crates/punktfunk-core/src/abi.rs index d0dd8448..82bf497f 100644 --- a/crates/punktfunk-core/src/abi.rs +++ b/crates/punktfunk-core/src/abi.rs @@ -2405,6 +2405,70 @@ pub unsafe extern "C" fn punktfunk_connection_request_keyframe( }) } +/// Ask the host to recover from loss by **reference-frame invalidation** rather than a full IDR: +/// report the range `[first_frame, last_frame]` of access units the client can no longer trust +/// (the first missing `frame_index` through the newest received). An RFI-capable host (AMD LTR / +/// NVENC) re-references a known-good picture before `first_frame` and emits a clean P-frame tagged +/// `USER_FLAG_RECOVERY_ANCHOR` — no 20-40x IDR spike; a host that can't RFI forces an IDR instead +/// (same effect as [`punktfunk_connection_request_keyframe`]). Non-blocking, fire-and-forget; the +/// recovered frame is the only ack, so THROTTLE it exactly like the keyframe request. Prefer this +/// over the keyframe request on loss so AMD/RFI hosts avoid the spike; keep the keyframe request as +/// the backstop for when the recovery frame itself is lost. +/// +/// # Safety +/// `c` is a valid connection handle. +#[cfg(feature = "quic")] +#[no_mangle] +pub unsafe extern "C" fn punktfunk_connection_request_rfi( + c: *const PunktfunkConnection, + first_frame: u32, + last_frame: u32, +) -> PunktfunkStatus { + guard(|| { + let c = match unsafe { c.as_ref() } { + Some(c) => c, + None => return PunktfunkStatus::NullPointer, + }; + match c.inner.request_rfi(first_frame, last_frame) { + Ok(()) => PunktfunkStatus::Ok, + Err(e) => e.status(), + } + }) +} + +/// Feed each received frame's `frame_index` (the [`PunktfunkFrame::frame_index`] field, in receive +/// order) so the client recovers from loss with a cheap reference-frame invalidation instead of a +/// full IDR. On a forward gap (a `frame_index` jump = the intervening frames were lost and the +/// following AUs reference a picture that never arrived) this fires a THROTTLED +/// [`punktfunk_connection_request_rfi`] for the lost range; an RFI-capable host (AMD LTR / NVENC) +/// then recovers with a clean P-frame instead of a 20-40x IDR spike. Call it for every received +/// frame — it is cheap and idempotent, and the [`punktfunk_connection_frames_dropped`]-driven +/// keyframe request stays the backstop. Writes whether a forward gap was detected this call to +/// `gap_out` (nullable — a client with a post-loss display freeze can use it to re-arm; most +/// clients pass NULL and ignore it). +/// +/// # Safety +/// `c` is a valid connection handle; `gap_out` is writable or NULL. +#[cfg(feature = "quic")] +#[no_mangle] +pub unsafe extern "C" fn punktfunk_connection_note_frame_index( + c: *const PunktfunkConnection, + frame_index: u32, + gap_out: *mut bool, +) -> PunktfunkStatus { + guard(|| { + let c = match unsafe { c.as_ref() } { + Some(c) => c, + None => return PunktfunkStatus::NullPointer, + }; + let gap = c.inner.note_frame_index(frame_index); + if !gap_out.is_null() { + unsafe { *gap_out = gap }; + } + PunktfunkStatus::Ok + }) +} + /// Cumulative access units the host→client reassembler dropped as unrecoverable (FEC couldn't /// rebuild them). A video loop polls this and calls [`punktfunk_connection_request_keyframe`] /// when it climbs — the correct loss trigger under the host's infinite GOP, where unrecoverable diff --git a/crates/punktfunk-core/src/client.rs b/crates/punktfunk-core/src/client.rs index 9ce5eb2c..d7422bf2 100644 --- a/crates/punktfunk-core/src/client.rs +++ b/crates/punktfunk-core/src/client.rs @@ -360,6 +360,59 @@ pub struct AudioPacket { pub data: Vec, } +/// At most one client→host RFI request per this window, so a burst of frame-index gaps (a +/// full-screen pan shedding shards) can't storm the control stream. Matches the shared Vulkan pump's +/// recovery-request throttle; the host coalesces further. +const RFI_THROTTLE: Duration = Duration::from_millis(100); + +/// State for [`NativeClient::note_frame_index`] — the client-side loss-range detector shared by every +/// embedder (Android, the C-ABI Apple client, the Windows shell pump) so none re-derives the wrapping +/// frame-index arithmetic. `next_expected` is the `frame_index` expected next in receive order; +/// `last_req` throttles the RFI requests a gap fires. +#[derive(Default)] +struct RfiRecovery { + next_expected: Option, + last_req: Option, +} + +impl RfiRecovery { + /// Pure decision behind [`NativeClient::note_frame_index`]: fold one received `frame_index` (in + /// receive order) observed at `now`, advancing the expectation and returning + /// `(gap, rfi_range)`. `gap` is whether this frame revealed a forward gap; `rfi_range` is + /// `Some((first_missing, last_missing))` when a (throttled) RFI should fire for the lost span, or + /// `None` when contiguous, a straggler, or throttled. Split out from the connection so the wrapping + /// arithmetic + [`RFI_THROTTLE`] are unit-testable without a live session (see the tests below). + fn observe(&mut self, frame_index: u32, now: Instant) -> (bool, Option<(u32, u32)>) { + match self.next_expected { + Some(exp) => { + // Wrapping split at the half-space: a small positive delta is a forward gap + // (missing frames); a delta in the top half is a straggler behind us. + let ahead = frame_index.wrapping_sub(exp); + if ahead == 0 { + self.next_expected = Some(frame_index.wrapping_add(1)); // contiguous + (false, None) + } else if ahead < u32::MAX / 2 { + // Forward gap: [exp, frame_index-1] lost. Advance past this frame so the same + // gap isn't re-detected, then fire a throttled RFI for the lost range. + self.next_expected = Some(frame_index.wrapping_add(1)); + let send = self.last_req.is_none_or(|t| now.duration_since(t) >= RFI_THROTTLE); + if send { + self.last_req = Some(now); + } + let range = send.then(|| (exp, frame_index.wrapping_sub(1))); + (true, range) + } else { + (false, None) // straggler behind the delivery point — leave the expectation + } + } + None => { + self.next_expected = Some(frame_index.wrapping_add(1)); + (false, None) + } + } + } +} + pub struct NativeClient { // Each plane's receiver sits behind its own mutex so `NativeClient` is `Sync` and Rust // embedders can share one `Arc` across their plane threads (the same @@ -403,6 +456,10 @@ pub struct NativeClient { /// for the client stats HUDs (the unified spec's per-window `FEC` counter — proof FEC is /// earning its keep); readers window it by diffing successive reads. fec_recovered: Arc, + /// Client-side RFI-on-loss detector state for [`note_frame_index`](Self::note_frame_index): the + /// next `frame_index` expected in receive order + the last RFI-request time (throttle). Lets every + /// embedder share one loss-range detector instead of re-deriving the wrapping frame arithmetic. + rfi: Mutex, /// Kernel ids of the client's latency-critical native threads: the internal data-plane pump /// (UDP receive + FEC reassembly) plus any embedder plane threads registered via /// [`NativeClient::register_hot_thread`]. The Android client feeds these to an ADPF hint session @@ -689,6 +746,7 @@ impl NativeClient { worker: Some(worker), frames_dropped, fec_recovered, + rfi: Mutex::new(RfiRecovery::default()), hot_tids, clock_offset, mode: mode_slot, @@ -891,6 +949,32 @@ impl NativeClient { .map_err(|_| PunktfunkError::Closed) } + /// Feed each received AU's `frame_index` (in receive order) so the client recovers from loss with + /// a cheap reference-frame invalidation instead of always paying for a full IDR. On a **forward + /// gap** — a `frame_index` jump means the intervening frames were lost and the following AUs + /// reference a picture the decoder never got — this fires a **throttled** + /// [`request_rfi`](Self::request_rfi) for the lost range `[first_missing, frame_index-1]`. An + /// RFI-capable host (AMD LTR / NVENC) then re-references a known-good frame (a clean P-frame, no + /// 20-40x IDR spike); a host that can't RFI forces an IDR, same as the keyframe path. + /// + /// Call it for EVERY received frame; it is cheap and idempotent, and the + /// [`frames_dropped`](Self::frames_dropped)-driven [`request_keyframe`](Self::request_keyframe) + /// loop stays the backstop for when the recovery frame itself is lost. Returns `true` when a + /// forward gap was detected on this call (whether or not the RFI was throttled), so a client with + /// a post-loss display freeze can (re-)arm it on the same signal. + /// + /// This centralizes the loss-range detection so every embedder gets identical behavior. (The + /// in-process Vulkan session pump keeps its own copy because it gates a display freeze on the same + /// signal and shares one throttle across RFI + keyframe requests.) + pub fn note_frame_index(&self, frame_index: u32) -> bool { + // Decide (and update state) under the lock; fire the request after releasing it. + let (gap, rfi_range) = self.rfi.lock().unwrap().observe(frame_index, Instant::now()); + if let Some((first, last)) = rfi_range { + let _ = self.request_rfi(first, last); + } + gap + } + /// Cumulative access units the host→client reassembler dropped as unrecoverable (FEC couldn't /// rebuild them). A video loop polls this and calls [`request_keyframe`](Self::request_keyframe) /// when it increases — the correct loss trigger under infinite GOP, where unrecoverable loss @@ -1976,6 +2060,108 @@ mod host_port_tests { } } +#[cfg(test)] +mod rfi_recovery_tests { + //! The client-side loss-range detector shared by every embedder (Android, the C-ABI Apple + //! client, the Windows shell pump). `observe` is pure over `(frame_index, now)`, so the wrapping + //! frame arithmetic and the RFI throttle are exercised here without a live session. + use super::{RfiRecovery, RFI_THROTTLE}; + use std::time::{Duration, Instant}; + + // A fixed base instant; offsets model the throttle window deterministically (no sleeping). + fn base() -> Instant { + Instant::now() + } + + #[test] + fn first_frame_arms_without_a_gap() { + let mut r = RfiRecovery::default(); + // The opening frame only seeds the expectation — there is no prior frame to be missing. + assert_eq!(r.observe(100, base()), (false, None)); + assert_eq!(r.next_expected, Some(101)); + } + + #[test] + fn contiguous_frames_never_gap() { + let mut r = RfiRecovery::default(); + let t = base(); + r.observe(100, t); + assert_eq!(r.observe(101, t), (false, None)); + assert_eq!(r.observe(102, t), (false, None)); + assert_eq!(r.observe(103, t), (false, None)); + assert_eq!(r.next_expected, Some(104)); + } + + #[test] + fn forward_gap_reports_the_exact_lost_range() { + let mut r = RfiRecovery::default(); + let t = base(); + r.observe(100, t); // expecting 101 next + // 101..=104 were lost; 105 arrived. The RFI must name exactly the missing span. + assert_eq!(r.observe(105, t), (true, Some((101, 104)))); + // The expectation advances past the delivered frame so the same gap can't re-fire. + assert_eq!(r.next_expected, Some(106)); + } + + #[test] + fn single_frame_drop_names_a_unit_range() { + let mut r = RfiRecovery::default(); + let t = base(); + r.observe(100, t); + // Exactly one frame (101) lost → range is the single index [101, 101]. + assert_eq!(r.observe(102, t), (true, Some((101, 101)))); + } + + #[test] + fn throttle_suppresses_bursts_then_re_opens() { + let mut r = RfiRecovery::default(); + let t0 = base(); + r.observe(100, t0); + // First gap fires the request and stamps the throttle. + assert_eq!(r.observe(105, t0), (true, Some((101, 104)))); + // A second gap 50 ms later is still a gap, but the request is throttled away. + assert_eq!(r.observe(110, t0 + Duration::from_millis(50)), (true, None)); + // Past the window, the request re-opens for the still-accurate lost span. + assert_eq!( + r.observe(120, t0 + RFI_THROTTLE + Duration::from_millis(1)), + (true, Some((111, 119))) + ); + } + + #[test] + fn stragglers_behind_the_delivery_point_are_ignored() { + let mut r = RfiRecovery::default(); + let t = base(); + r.observe(100, t); + r.observe(105, t); // expecting 106 next + // A reordered late arrival (103, well behind 106) is neither a gap nor a request, and it + // must not rewind the expectation — otherwise the next in-order frame would false-gap. + assert_eq!(r.observe(103, t), (false, None)); + assert_eq!(r.next_expected, Some(106)); + } + + #[test] + fn wraparound_is_contiguous_across_u32_max() { + let mut r = RfiRecovery::default(); + let t = base(); + r.observe(u32::MAX - 1, t); // expecting u32::MAX next + assert_eq!(r.observe(u32::MAX, t), (false, None)); // contiguous, expectation wraps to 0 + assert_eq!(r.next_expected, Some(0)); + assert_eq!(r.observe(0, t), (false, None)); // still contiguous across the wrap + assert_eq!(r.next_expected, Some(1)); + } + + #[test] + fn gap_range_wraps_across_u32_max() { + let mut r = RfiRecovery::default(); + let t = base(); + r.observe(u32::MAX - 1, t); // expecting u32::MAX next + // u32::MAX was lost and 1 arrived → the lost span wraps: [u32::MAX, 0]. + assert_eq!(r.observe(1, t), (true, Some((u32::MAX, 0)))); + assert_eq!(r.next_expected, Some(2)); + } +} + #[cfg(test)] mod frame_channel_tests { use super::{FrameChannel, FramePop, FRAME_QUEUE_HARD_CAP}; diff --git a/include/punktfunk_core.h b/include/punktfunk_core.h index 2500aeb4..9483b9e6 100644 --- a/include/punktfunk_core.h +++ b/include/punktfunk_core.h @@ -1597,6 +1597,43 @@ PunktfunkStatus punktfunk_connection_request_mode(const PunktfunkConnection *c, PunktfunkStatus punktfunk_connection_request_keyframe(const PunktfunkConnection *c); #endif +#if defined(PUNKTFUNK_FEATURE_QUIC) +// Ask the host to recover from loss by **reference-frame invalidation** rather than a full IDR: +// report the range `[first_frame, last_frame]` of access units the client can no longer trust +// (the first missing `frame_index` through the newest received). An RFI-capable host (AMD LTR / +// NVENC) re-references a known-good picture before `first_frame` and emits a clean P-frame tagged +// `USER_FLAG_RECOVERY_ANCHOR` — no 20-40x IDR spike; a host that can't RFI forces an IDR instead +// (same effect as [`punktfunk_connection_request_keyframe`]). Non-blocking, fire-and-forget; the +// recovered frame is the only ack, so THROTTLE it exactly like the keyframe request. Prefer this +// over the keyframe request on loss so AMD/RFI hosts avoid the spike; keep the keyframe request as +// the backstop for when the recovery frame itself is lost. +// +// # Safety +// `c` is a valid connection handle. +PunktfunkStatus punktfunk_connection_request_rfi(const PunktfunkConnection *c, + uint32_t first_frame, + uint32_t last_frame); +#endif + +#if defined(PUNKTFUNK_FEATURE_QUIC) +// Feed each received frame's `frame_index` (the [`PunktfunkFrame::frame_index`] field, in receive +// order) so the client recovers from loss with a cheap reference-frame invalidation instead of a +// full IDR. On a forward gap (a `frame_index` jump = the intervening frames were lost and the +// following AUs reference a picture that never arrived) this fires a THROTTLED +// [`punktfunk_connection_request_rfi`] for the lost range; an RFI-capable host (AMD LTR / NVENC) +// then recovers with a clean P-frame instead of a 20-40x IDR spike. Call it for every received +// frame — it is cheap and idempotent, and the [`punktfunk_connection_frames_dropped`]-driven +// keyframe request stays the backstop. Writes whether a forward gap was detected this call to +// `gap_out` (nullable — a client with a post-loss display freeze can use it to re-arm; most +// clients pass NULL and ignore it). +// +// # Safety +// `c` is a valid connection handle; `gap_out` is writable or NULL. +PunktfunkStatus punktfunk_connection_note_frame_index(const PunktfunkConnection *c, + uint32_t frame_index, + bool *gap_out); +#endif + #if defined(PUNKTFUNK_FEATURE_QUIC) // Cumulative access units the host→client reassembler dropped as unrecoverable (FEC couldn't // rebuild them). A video loop polls this and calls [`punktfunk_connection_request_keyframe`]