unom/punktfunk
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases 23 Wiki Activity

feat(client): freeze-until-reanchor loss recovery on Android + Apple via shared core gate

Browse Source 
After unrecoverable loss the host keeps sending delta frames that reference a
picture the client never received; hardware decoders conceal these as gray/
garbage with a success status. Linux already withheld them and held the last
good frame until a proven clean re-anchor — this brings that behavior to the
Android and Apple clients.

Extract the Linux pump's freeze state machine into a shared `ReanchorGate` in
punktfunk-core (reanchor.rs, 18 tests) exposed over the C ABI (ABI v6, additive —
no wire change) for the Swift clients. Migrate the Linux/Deck pump
(pf-client-core) onto it as the parity proof (no-op refactor). Then wire:

- Android (decode.rs, both sync + async loops): arm on the frame-index gap, a
  pts-keyed flag map carries the wire flags to the output-buffer release, fold
  the gate per drained output, gate.poll replaces the dropped-climb block.
- Apple Stage2Pipeline (default): arm on a gap (new noteFrameIndexGap), withhold
  at the ring-submit seam (CAMetalLayer holds its last drawable), poll
  framesDropped, fold VT decode errors through the no-output streak.
- Apple StreamPump (stage-1): fold at enqueue, withhold via
  kCMSampleAttachmentKey_DoNotDisplay so the layer keeps decoding (reference
  chain intact) but holds the last displayed frame.
- Apple VideoDecoder: thread the AU's wire flags to the async decode callback via
  a retained FrameContext refcon (replaces the receivedNs bit-pattern scalar).

Lifts only on a proven re-anchor (IDR / RFI anchor / 2nd recovery mark) with a
500 ms backstop so a lost re-anchor can never freeze forever. Apple: swift build
clean, 123/123 tests pass (incl. VideoToolboxRoundTripTests). On-glass
loss-injection validation still owed.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Enrico Bühler
2026-07-13 01:21:25 +02:00
parent cd701a9594
commit 8a18e130a2
11 changed files with 1104 additions and 380 deletions
Download Patch File Download Diff File Expand all files Collapse all files
clients/android/native/src/decode.rs
+133 -46
View File
@@ -15,6 +15,7 @@ use ndk::media::media_format::MediaFormat;
use ndk::native_window::NativeWindow;
use punktfunk_core::client::NativeClient;
use punktfunk_core::error::PunktfunkError;
use punktfunk_core::reanchor::{GateVerdict, ReanchorGate};
use punktfunk_core::session::Frame;
use std::collections::VecDeque;
use std::ffi::c_void;
@@ -208,9 +209,15 @@ fn run_sync(
// pressure the AU stays parked here instead of being dropped (a drop forces a keyframe
// round-trip) and we only pop the next one once it's queued.
let mut pending: Option<Frame> = None;
// Loss recovery: watch the host→client unrecoverable-drop count and ask for an IDR when it
// climbs.
let mut last_dropped = client.frames_dropped();
// Freeze-until-reanchor: the shared post-loss gate ([`punktfunk_core::reanchor::ReanchorGate`]).
// Armed on a frame-index gap or a dropped-count climb, it withholds the decoder's concealed output
// (released WITHOUT rendering — the SurfaceView keeps the last rendered frame on glass) until a
// proven clean re-anchor lifts it: an IDR (wire FLAG_SOF), an RFI anchor, or the 2nd recovery mark.
// `last_kf_req` throttles the keyframe intents it emits; `recovery_flags` carries each AU's
// user_flags from feed to present (keyed by the codec-echoed pts) so `on_decoded` reads the
// re-anchor signalling the platform decoder doesn't expose.
let mut gate = ReanchorGate::new(client.frames_dropped());
let mut recovery_flags: VecDeque<(u64, u32)> = VecDeque::new();
let mut last_kf_req: Option<Instant> = None;
// Skew-corrected latency stats (spec: design/stats-unification.md) use the negotiated
// host-minus-client clock offset (0 if the host didn't answer the skew handshake — then the
@@ -245,9 +252,18 @@ fn run_sync(
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);
// recovers with a cheap clean P-frame instead of a full IDR. The same forward gap
// arms the freeze gate so the decoder's concealment is held off the screen until the
// recovery re-anchors. The frames_dropped keyframe path below stays the backstop.
if client.note_frame_index(frame.frame_index) {
gate.arm(Instant::now());
}
// Park this AU's re-anchor flags for the present side (keyed by the pts the codec
// echoes on the output buffer) — unconditional, unlike the HUD's `in_flight` map.
recovery_flags.push_back((frame.pts_ns / 1000, frame.flags));
if recovery_flags.len() > IN_FLIGHT_CAP {
recovery_flags.pop_front();
}
if fed == 0 {
let p = &frame.data;
log::info!(
@@ -336,6 +352,8 @@ fn run_sync(
&mut in_flight,
clock_offset.load(Ordering::Relaxed),
&tracker,
&mut gate,
&mut recovery_flags,
);
rendered += r;
discarded += d;
@@ -375,21 +393,19 @@ fn run_sync(
work_accum_ns = 0;
}
// Loss recovery: under infinite GOP the only recovery keyframe is one we request. The
// reassembler drops unrecoverable AUs (frames_dropped); the decoder then conceals the
// reference-missing delta frames that follow and renders them without error, so keying off
// a decode error rarely fires. Request an IDR when the drop count climbs, throttled — the
// decode stays wedged for several frames until the IDR lands, so requesting every frame
// would flood the control stream.
let dropped = client.frames_dropped();
if dropped > last_dropped {
last_dropped = dropped;
let now = Instant::now();
if last_kf_req.is_none_or(|t| now.duration_since(t) >= Duration::from_millis(100)) {
last_kf_req = Some(now);
let _ = client.request_keyframe();
log::debug!("decode: requested keyframe (loss recovery, dropped={dropped})");
}
// Loss recovery + overdue backstop, folded through the gate. Under infinite GOP the only
// recovery keyframe is one we request; the reassembler drops unrecoverable AUs (frames_dropped)
// and the decoder then conceals the reference-missing deltas and renders them without error, so
// a decode-error trigger rarely fires — the gate arms the freeze on the drop-count climb
// instead. An overdue freeze (held REANCHOR_FREEZE_MAX with no clean re-anchor) re-asks while it
// keeps holding: never resume to gray — a dead stream is the QUIC idle-timeout watchdog's job.
let now = Instant::now();
if gate.poll(client.frames_dropped(), now)
&& last_kf_req.is_none_or(|t| now.duration_since(t) >= Duration::from_millis(100))
{
last_kf_req = Some(now);
let _ = client.request_keyframe();
log::debug!("decode: requested keyframe (loss recovery / overdue re-anchor)");
}
}
@@ -707,8 +723,10 @@ struct OutputReady {
/// internal looper thread) push the codec ones; the feeder thread pushes `Au`. Each carries only
/// owned/`Copy` data so the callback closures satisfy the `Send` bound and never touch the codec.
enum DecodeEvent {
/// A received access unit from the feeder, ready to queue into the decoder.
Au(Frame),
/// A received access unit from the feeder, ready to queue into the decoder. The `bool` is the
/// feeder's [`NativeClient::note_frame_index`] verdict — `true` when this AU revealed a forward
/// frame-index gap, so the loop arms the freeze gate (the feeder already fired the RFI request).
Au(Frame, bool),
/// An input buffer slot freed (index) — we can queue an AU into it.
InputAvailable(usize),
/// A decoded frame is ready (buffer index + echoed pts + the callback-time `decoded` stamp).
@@ -894,7 +912,12 @@ fn run_async(
let mut discarded: u64 = 0;
// AUs larger than the codec input buffer, dropped whole (see `feed`/`feed_ready`).
let mut oversized_dropped: u64 = 0;
let mut last_dropped = client.frames_dropped();
// Freeze-until-reanchor gate (see the sync loop for the rationale). Armed on a frame-index gap
// (the feeder's Au verdict), a parked-AU overflow drop, a dropped-count climb, or a recoverable
// codec error; `recovery_flags` carries each AU's user_flags from `dispatch_event` (feed) to
// `present_ready` (present), keyed by the codec-echoed pts.
let mut gate = ReanchorGate::new(client.frames_dropped());
let mut recovery_flags: VecDeque<(u64, u32)> = VecDeque::new();
let mut last_kf_req: Option<Instant> = None;
// Productive (dispatch+feed+present) time between displayed frames; reported to ADPF once one is
// presented. The blocking event wait is excluded (idle, not work) — same accounting as the sync loop.
@@ -920,6 +943,8 @@ fn run_async(
&mut ready,
&mut fmt_dirty,
&mut fatal,
&mut gate,
&mut recovery_flags,
));
}
// Coalesce every other event already queued into this one work pass — correct newest-only
@@ -932,6 +957,8 @@ fn run_async(
&mut ready,
&mut fmt_dirty,
&mut fatal,
&mut gate,
&mut recovery_flags,
));
}
stats.note_skipped(aus_dropped); // parked-AU overflow drops are client-side skips too
@@ -956,6 +983,8 @@ fn run_async(
&tracker,
&mut rendered,
&mut discarded,
&mut gate,
&mut recovery_flags,
);
work_accum_ns += work_t0.elapsed().as_nanos() as i64;
@@ -987,17 +1016,19 @@ fn run_async(
log::info!("decode: fed={fed} rendered={rendered} discarded={discarded}");
}
}
// Loss recovery: request an IDR when the reassembler's unrecoverable-drop count climbs (or we
// dropped a parked AU on overflow), throttled so a multi-frame recovery gap doesn't flood the
// control stream.
let dropped = client.frames_dropped();
if dropped > last_dropped || aus_dropped > 0 {
last_dropped = dropped;
let now = Instant::now();
if last_kf_req.is_none_or(|t| now.duration_since(t) >= Duration::from_millis(100)) {
last_kf_req = Some(now);
let _ = client.request_keyframe();
}
// Loss recovery + overdue backstop, folded through the gate. A parked-AU overflow drop is itself
// a loss, so it arms the freeze directly; the gate's `poll` then arms on a dropped-count climb
// and re-asks on an overdue freeze. All keyframe intents route through the shared 100 ms
// throttle so a multi-frame recovery gap can't flood the control stream.
let now = Instant::now();
if aus_dropped > 0 {
gate.arm(now);
}
if (gate.poll(client.frames_dropped(), now) || aus_dropped > 0)
&& last_kf_req.is_none_or(|t| now.duration_since(t) >= Duration::from_millis(100))
{
last_kf_req = Some(now);
let _ = client.request_keyframe();
}
}
@@ -1033,8 +1064,9 @@ fn feeder_loop(
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);
// instead of a full IDR (the frames_dropped keyframe path is the backstop). The gap
// verdict rides the Au event so the decode loop arms its freeze gate on the same signal.
let gap = 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;
@@ -1067,7 +1099,7 @@ fn feeder_loop(
}
}
}
if ev_tx.send(DecodeEvent::Au(frame)).is_err() {
if ev_tx.send(DecodeEvent::Au(frame, gap)).is_err() {
break; // the decode loop is gone
}
}
@@ -1079,6 +1111,7 @@ fn feeder_loop(
/// Route one [`DecodeEvent`] into the loop's working sets. Returns `true` only when a parked AU was
/// dropped on overflow (the caller then requests a keyframe).
#[allow(clippy::too_many_arguments)] // two call sites; the freeze gate + flag map are threaded in
fn dispatch_event(
ev: DecodeEvent,
pending_aus: &mut VecDeque<Frame>,
@@ -1086,9 +1119,20 @@ fn dispatch_event(
ready: &mut Vec<OutputReady>,
fmt_dirty: &mut bool,
fatal: &mut bool,
gate: &mut ReanchorGate,
recovery_flags: &mut VecDeque<(u64, u32)>,
) -> bool {
match ev {
DecodeEvent::Au(f) => {
DecodeEvent::Au(f, gap) => {
// A forward frame-index gap arms the freeze; park this AU's flags for the present side to
// fold `on_decoded` (keyed by the pts the codec will echo).
if gap {
gate.arm(Instant::now());
}
recovery_flags.push_back((f.pts_ns / 1000, f.flags));
if recovery_flags.len() > IN_FLIGHT_CAP {
recovery_flags.pop_front();
}
pending_aus.push_back(f);
if pending_aus.len() > FRAME_PARK_CAP {
pending_aus.pop_front(); // sustained overflow — drop oldest, signal a keyframe request
@@ -1109,6 +1153,10 @@ fn dispatch_event(
DecodeEvent::Error { fatal: f } => {
if f {
*fatal = true;
} else {
// A recoverable/transient codec error is a decode hiccup on a broken reference chain —
// arm the freeze so the concealed output it recovers into is held off the screen.
gate.arm(Instant::now());
}
}
}
@@ -1180,6 +1228,8 @@ fn present_ready(
tracker: &DisplayTracker,
rendered: &mut u64,
discarded: &mut u64,
gate: &mut ReanchorGate,
recovery_flags: &mut VecDeque<(u64, u32)>,
) {
if ready.is_empty() {
return;
@@ -1192,10 +1242,16 @@ fn present_ready(
note_decoded_pts(stats, &mut g, clock_offset, o.pts_us, o.decoded_ns);
}
}
// Fold EVERY output through the gate in pts (== decode) order — even the ones newest-wins discards —
// so the two-mark re-anchor count stays correct; the newest's verdict decides whether it reaches
// glass (`false` = withheld concealment; the SurfaceView keeps the last rendered frame frozen on).
let now = Instant::now();
let last = ready.len() - 1;
let mut skipped: u64 = 0;
for (i, o) in ready.drain(..).enumerate() {
let render = i == last;
let flags = take_flags(recovery_flags, o.pts_us);
let present = gate.on_decoded(flags, false, now) == GateVerdict::Present;
let render = i == last && present;
match codec.release_output_buffer_by_index(o.index, render) {
Ok(()) if render => {
*rendered += 1;
@@ -1215,7 +1271,7 @@ fn present_ready(
}
}
}
stats.note_skipped(skipped); // HUD `skipped` counter (newest-wins drops); no-op while hidden
stats.note_skipped(skipped); // HUD `skipped` counter (newest-wins + held-off drops); no-op hidden
}
/// React to an output-format change by signalling the stream's HDR dataspace on the Surface (SDR
@@ -1411,19 +1467,28 @@ fn drain(
in_flight: &mut VecDeque<(u64, i128)>,
clock_offset: i64,
tracker: &DisplayTracker,
gate: &mut ReanchorGate,
recovery_flags: &mut VecDeque<(u64, u32)>,
) -> (u64, u64) {
// Newest ready buffer so far (presented after the loop) with its HUD metadata —
// `Some((pts_us, decoded_ns))` only while the HUD is visible (the stamp read is gated).
// `Some((pts_us, decoded_ns))` only while the HUD is visible. `held_present` is the freeze gate's
// verdict for that newest buffer (`false` = a post-loss concealment to withhold).
let mut held: Option<(OutputBuffer<'_>, Option<(u64, i128)>)> = None;
let mut held_present = true;
let mut discarded: u64 = 0;
let mut wait = first_wait;
loop {
match codec.dequeue_output_buffer(wait) {
Ok(DequeuedOutputBufferInfoResult::Buffer(buf)) => {
wait = Duration::ZERO; // only the first dequeue may block
// Fold every dequeued frame through the gate in pts (== decode) order — even the ones
// the newest-wins policy discards — so the two-mark re-anchor count stays correct; the
// verdict of the newest (last folded) buffer decides whether it reaches glass.
let pts_us = buf.info().presentation_time_us().max(0) as u64;
let flags = take_flags(recovery_flags, pts_us);
held_present = gate.on_decoded(flags, false, Instant::now()) == GateVerdict::Present;
let meta = if stats.enabled() {
// The dequeue IS the sync loop's decoded-availability instant.
let pts_us = buf.info().presentation_time_us().max(0) as u64;
let decoded_ns = now_realtime_ns();
note_decoded_pts(stats, in_flight, clock_offset, pts_us, decoded_ns);
Some((pts_us, decoded_ns))
@@ -1469,16 +1534,19 @@ fn drain(
}
}
}
// Present the newest ready frame, if any, and park its metadata for the render callback.
// Present the newest ready frame — UNLESS the gate is withholding it as a post-loss concealment,
// in which case release it without rendering (the SurfaceView keeps the last rendered frame frozen
// on glass) and count it as a discard rather than a display.
let mut rendered = 0;
if let Some((buf, meta)) = held {
match codec.release_output_buffer(buf, true) {
Ok(()) => {
match codec.release_output_buffer(buf, held_present) {
Ok(()) if held_present => {
rendered = 1;
if let Some((pts_us, decoded_ns)) = meta {
tracker.note_rendered(pts_us, decoded_ns);
}
}
Ok(()) => discarded += 1, // held off the screen — awaiting a clean re-anchor
Err(e) => log::warn!("decode: release_output_buffer: {e}"),
}
}
@@ -1520,6 +1588,25 @@ fn note_decoded_pts(
stats.note_decoded(e2e_us, decode_us);
}
/// The AU `user_flags` for a decoded output, keyed by the echoed `presentationTimeUs`. Recovery
/// signalling (FLAG_SOF IDR marker / RECOVERY_ANCHOR / RECOVERY_POINT) rides the AU's flags, which are
/// only in scope at feed time — so the feed side parks `(pts_us, flags)` here and the present side
/// looks them up to fold [`ReanchorGate::on_decoded`]. Decode order == input order (low-latency, no
/// B-frames), so this evicts entries older than `pts_us` as it goes; a miss (probe filler, or an entry
/// aged past the cap) reads `0` — no recovery flags, decoded normally.
fn take_flags(map: &mut VecDeque<(u64, u32)>, pts_us: u64) -> u32 {
while let Some(&(p, f)) = map.front() {
if p > pts_us {
break; // future frame — leave it for its own output buffer
}
map.pop_front();
if p == pts_us {
return f;
}
}
0
}
/// Map the decoder's reported output colour to a BT.2020 HDR dataspace, or `None` for SDR. The
/// integer values are the Android MediaFormat colour constants the NDK shares: COLOR_TRANSFER
/// ST2084 = 6 (PQ/HDR10), HLG = 7; COLOR_RANGE FULL = 1, LIMITED = 2 (the host encodes limited).
clients/apple/Sources/PunktfunkKit/Connection/PunktfunkConnection.swift
+15
View File
@@ -450,6 +450,21 @@ public final class PunktfunkConnection {
_ = punktfunk_connection_note_frame_index(h, frameIndex, nil)
}
/// Like `noteFrameIndex`, but also reports whether the core saw a FORWARD frame-index gap — the
/// signal that intervening frames were lost and the following AUs reference a picture that never
/// arrived. The post-loss re-anchor gate arms its display freeze on a gap (the earliest, most
/// precise loss trigger — ahead of the `framesDropped` climb). Same core side effect as
/// `noteFrameIndex` (the throttled RFI request); call it for every received AU. Returns false
/// after close.
public func noteFrameIndexGap(_ frameIndex: UInt32) -> Bool {
abiLock.lock()
defer { abiLock.unlock() }
guard let h = handle, !closeRequested else { return false }
var gap = false
_ = punktfunk_connection_note_frame_index(h, frameIndex, &gap)
return gap
}
/// 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
clients/apple/Sources/PunktfunkKit/Video/ReanchorGate.swift
+99
View File
@@ -0,0 +1,99 @@
// Swift wrapper around the punktfunk-core C ABI's post-loss re-anchor gate
// (`punktfunk_reanchor_gate_*`, ABI v6). The shared Rust gate (crates/punktfunk-core/src/reanchor.rs)
// is what the Linux/Windows desktop pump and the Android client use directly; the Swift clients reach
// it across the C ABI so the freeze-until-reanchor policy is defined ONCE for every platform.
//
// Why a freeze at all: after unrecoverable loss the host keeps sending delta frames that reference a
// picture the client never got. Hardware decoders (VideoToolbox included) don't reliably error on
// that — they CONCEAL, returning a gray/garbage frame with a success status. Presenting those is the
// visible "gray flash with motion" of the loss reports. The gate withholds concealed frames and holds
// the last good picture on glass until a PROVEN clean re-anchor lands — an IDR (wire `FLAG_SOF`), an
// RFI recovery anchor (`USER_FLAG_RECOVERY_ANCHOR`), or the 2nd of two intra-refresh recovery marks
// (`USER_FLAG_RECOVERY_POINT`) — with a bounded backstop so a lost re-anchor can never freeze forever.
// See punktfunk-planning design/client-reanchor-freeze-parity.md.
//
// Threading: one gate per session. Its calls arrive from two threads — the pump thread (`arm` on a
// frame-index gap / a submit failure, `poll` per iteration) and a VideoToolbox decode thread
// (`onDecoded` per decoded frame, `onNoOutput` on a decode error). The raw Rust gate is a plain
// struct behind an opaque pointer with no internal synchronization, so every call is serialized under
// `lock` here — the calls are cheap field updates, so contention is negligible. `@unchecked Sendable`:
// the lock enforces the contract.
import Foundation
import PunktfunkCore
final class ReanchorGate: @unchecked Sendable {
private let lock = NSLock()
/// The opaque `ReanchorGate *`. `var` so `reseed` can swap it at session start. Never NULL
/// (`punktfunk_reanchor_gate_new` never returns NULL).
private var ptr: OpaquePointer
/// Seed the baseline with the connection's current `framesDropped` so the first `poll` doesn't
/// read the session's starting drop count as a fresh loss.
init(framesDropped: UInt64) {
ptr = punktfunk_reanchor_gate_new(framesDropped)
}
deinit { punktfunk_reanchor_gate_free(ptr) }
/// Re-anchor the drop-count baseline to `framesDropped` for a (re)started session. The gate is
/// created in the pipeline's init (before a connection exists, seeded 0); `start` calls this once
/// the live connection's count is known so a mid-life connection's non-zero baseline isn't
/// mistaken for loss on the first poll.
func reseed(framesDropped: UInt64) {
lock.lock()
defer { lock.unlock() }
punktfunk_reanchor_gate_free(ptr)
ptr = punktfunk_reanchor_gate_new(framesDropped)
}
/// Arm the freeze: a loss was detected (a frame-index gap, or a decoder wedge). Zeroes the
/// recovery-mark count and (re)sets the backstop deadline.
func arm() {
lock.lock()
punktfunk_reanchor_gate_arm(ptr)
lock.unlock()
}
/// Fold one decoded frame. `flags` is the AU's wire `user_flags`. Returns true to PRESENT the
/// frame, false to WITHHOLD it as a post-loss concealment (hold the last good picture). Pass
/// `decoderKeyframe: false` — VideoToolbox doesn't flag IDRs, so the wire `FLAG_SOF` covers it.
func onDecoded(flags: UInt32, decoderKeyframe: Bool = false) -> Bool {
lock.lock()
defer { lock.unlock() }
var present = false
_ = punktfunk_reanchor_gate_on_decoded(ptr, flags, decoderKeyframe, &present)
return present
}
/// A received AU produced no decoded frame (a VideoToolbox decode error). Returns true when the
/// no-output streak has tripped (the gate armed the freeze) and the caller should — throttled —
/// request a keyframe.
func onNoOutput() -> Bool {
lock.lock()
defer { lock.unlock() }
var requestKf = false
_ = punktfunk_reanchor_gate_on_no_output(ptr, &requestKf)
return requestKf
}
/// Periodic fold of the session's `framesDropped` plus the overdue backstop. Returns true when the
/// caller should — throttled — request a keyframe (a drop-count climb armed a fresh freeze, or the
/// freeze is overdue and re-asks while it keeps holding).
func poll(framesDropped: UInt64) -> Bool {
lock.lock()
defer { lock.unlock() }
var requestKf = false
_ = punktfunk_reanchor_gate_poll(ptr, framesDropped, &requestKf)
return requestKf
}
/// Whether the gate is currently withholding concealed frames (frozen on the last good picture).
var isHolding: Bool {
lock.lock()
defer { lock.unlock() }
var holding = false
_ = punktfunk_reanchor_gate_is_holding(ptr, &holding)
return holding
}
}
clients/apple/Sources/PunktfunkKit/Video/Stage2Pipeline.swift
+25 -6
View File
@@ -259,6 +259,10 @@ public final class Stage2Pipeline {
private let endToEndMeter: LatencyMeter?
private let displayMeter: LatencyMeter?
private let recovery = KeyframeRecovery()
/// Post-loss freeze-until-reanchor gate (shared core policy via the C ABI). Created here seeded 0;
/// `start` reseeds it to the live connection's drop count. Captured by the decoder callbacks
/// (which withhold concealed frames) and driven by the pump (arm on a gap, poll per iteration).
private let gate = ReanchorGate(framesDropped: 0)
private var token = StopFlag()
private var offsetNs: Int64 = 0
/// Signalled when the pump thread exits, so `stop()` can join it (bounded) before `decoder.reset()`
@@ -306,21 +310,29 @@ public final class Stage2Pipeline {
let ring = ring
let recovery = recovery
let renderSignal = renderSignal
let gate = gate
self.decoder = VideoDecoder(
onDecoded: { frame in
// Decode stage = received→decoded, both client CLOCK_REALTIME (offset 0 — no
// skew applies). Stamped at decode completion, so it covers every decoded frame,
// including ones the newest-wins ring drops before present.
// including ones the re-anchor gate withholds or the newest-wins ring drops.
decodeMeter?.record(
ptsNs: UInt64(frame.receivedNs), atNs: frame.decodedNs, offsetNs: 0)
// Freeze-until-reanchor: WITHHOLD a decoder-concealed post-loss frame (the gray/
// garbage VideoToolbox returns Ok for a reference-missing delta) — don't submit it,
// so the CAMetalLayer keeps its last good drawable on glass. The gate lifts (returns
// present) on a proven clean re-anchor (IDR / RFI anchor / 2nd recovery mark) or the
// bounded backstop. decoderKeyframe=false: VT doesn't flag IDRs, the wire FLAG_SOF does.
guard gate.onDecoded(flags: frame.flags) else { return }
ring.submit(frame)
// FRAME ARRIVAL is the render trigger (never the display link — see the header).
renderSignal.signal()
},
// Async decode failure (a bad P-frame referencing a lost/corrupt IDR): the pump resets to
// re-gate on the next IDR, and we ask the host to send one now (infinite GOP — it wouldn't
// Async decode failure (a bad P-frame referencing a lost/corrupt IDR): fold it into the
// gate's no-output streak (which arms the freeze after a short run, matching the desktop),
// and when that trips ask the host for a fresh IDR now (infinite GOP — it wouldn't
// otherwise come soon). Throttled in KeyframeRecovery.
onDecodeError: { _ in recovery.request() })
onDecodeError: { _ in if gate.onNoOutput() { recovery.request() } })
}
/// Start pulling AUs into the decoder. MAIN THREAD. `onFrame` fires per AU at receipt (the
@@ -334,6 +346,7 @@ public final class Stage2Pipeline {
) {
offsetNs = connection.clockOffsetNs
recovery.bind(connection) // arm host-keyframe recovery for this session
gate.reseed(framesDropped: connection.framesDropped()) // baseline the freeze to this session
token = StopFlag() // fresh token per start — a stop is permanent (like StreamPump)
// Configure the decoder's chroma + the layer's initial colorimetry before the first frame. The
@@ -348,6 +361,7 @@ public final class Stage2Pipeline {
let recovery = recovery
let presenter = presenter
let pumpStopped = pumpStopped
let reanchorGate = gate
let thread = Thread {
defer { pumpStopped.signal() } // let stop() join the pump (bounded) before decoder.reset()
var format: CMVideoFormatDescription?
@@ -379,6 +393,9 @@ public final class Stage2Pipeline {
awaitingIDR = true
}
if awaitingIDR { recovery.request() }
// Freeze backstop: a drop-count climb arms the gate (in case the frame-index gap
// below was itself lost), and an overdue freeze re-asks for the re-anchor.
if reanchorGate.poll(framesDropped: dropped) { recovery.request() }
// Drain HDR mastering metadata (0xCE) and hand it to the PRESENTER (→ CAEDRMetadata).
// Polled UNCONDITIONALLY (not gated on connection.isHDR, the fixed Welcome flag): the
// host sends 0xCE only for HDR, INCLUDING a mid-session SDR→HDR transition (a game
@@ -391,8 +408,10 @@ public final class Stage2Pipeline {
// 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)
// recovery above stays the backstop for when the recovery frame itself is lost.
// The same gap is the earliest, most precise signal to ARM the display freeze —
// the following concealed frames are withheld until a clean re-anchor.
if connection.noteFrameIndexGap(au.frameIndex) { reanchorGate.arm() }
onFrame?(au)
if let f = connection.videoCodec.formatDescription(fromKeyframe: au.data) {
format = f // refreshed on every IDR (mode changes included)
clients/apple/Sources/PunktfunkKit/Video/StreamPump.swift
+33 -1
View File
@@ -28,6 +28,11 @@ final class StreamPump {
// Coalesced host keyframe requests (100 ms throttle — see KeyframeRecovery).
let recovery = KeyframeRecovery()
recovery.bind(connection)
// Post-loss freeze-until-reanchor (shared core policy via the C ABI). Stage-1 has no per-frame
// decode callback, so the gate is folded at ENQUEUE (from the AU's wire flags): a withheld
// frame is still enqueued but flagged DoNotDisplay so the layer's decoder keeps the reference
// chain fed while the last GOOD picture stays on glass — until a clean re-anchor lifts it.
let gate = ReanchorGate(framesDropped: connection.framesDropped())
// The layer is non-Sendable but its enqueue/flush are documented thread-safe, and after
// this point only the pump thread drives it — assert that so the @Sendable Thread closure
// may capture it.
@@ -77,13 +82,17 @@ final class StreamPump {
awaitingIDR = true
}
if awaitingIDR { recovery.request() }
// Freeze backstop: a drop-count climb arms the gate (should the frame-index gap
// below be lost too), and an overdue freeze re-asks for the re-anchor.
if gate.poll(framesDropped: dropped) { 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)
// The same gap is the earliest, most precise signal to ARM the display freeze.
if connection.noteFrameIndexGap(au.frameIndex) { gate.arm() }
onFrame?(au)
let idrFormat = connection.videoCodec.formatDescription(fromKeyframe: au.data)
if let f = idrFormat {
@@ -107,6 +116,7 @@ final class StreamPump {
// delta into a failed layer can't recover it.
if !wasFailed { pumpLog.warning("video: display layer .failed — flushing + re-anchoring") }
layer.flush()
gate.arm() // a wedged decoder is a loss — freeze until the re-anchor
if idrFormat == nil {
format = nil
awaitingIDR = true
@@ -117,6 +127,13 @@ final class StreamPump {
let sample = connection.videoCodec.sampleBuffer(au: au, format: f),
!token.isStopped // don't enqueue a stale frame after a restart
else { return true }
// Freeze-until-reanchor: while holding, WITHHOLD this concealed post-loss frame by
// flagging it DoNotDisplay — the layer still decodes it (keeping the reference
// chain fed) but shows the last GOOD picture until a clean re-anchor lifts the
// gate. Folded from the AU's wire flags (stage-1 has no decode callback).
if !gate.onDecoded(flags: au.flags) {
StreamPump.setDoNotDisplay(sample)
}
layer.enqueue(sample)
return true
} catch {
@@ -133,6 +150,21 @@ final class StreamPump {
thread.start()
}
/// Flag a sample decode-but-don't-display (`kCMSampleAttachmentKey_DoNotDisplay`). Used to
/// withhold decoder-concealed post-loss frames while the re-anchor gate holds: the layer keeps
/// its reference chain fed without flipping the frozen picture. No-op if the attachments array
/// can't be materialized (then the frame just displays — the freeze degrades to the old behavior).
private static func setDoNotDisplay(_ sample: CMSampleBuffer) {
guard let attachments = CMSampleBufferGetSampleAttachmentsArray(
sample, createIfNecessary: true), CFArrayGetCount(attachments) > 0
else { return }
let dict = unsafeBitCast(CFArrayGetValueAtIndex(attachments, 0), to: CFMutableDictionary.self)
CFDictionarySetValue(
dict,
Unmanaged.passUnretained(kCMSampleAttachmentKey_DoNotDisplay).toOpaque(),
Unmanaged.passUnretained(kCFBooleanTrue).toOpaque())
}
/// Stop pumping (≤ one poll timeout). Does not close the connection.
func stop() {
token.stop()
clients/apple/Sources/PunktfunkKit/Video/VideoDecoder.swift
+38 -11
View File
@@ -27,19 +27,40 @@ public struct ReadyFrame: @unchecked Sendable {
/// True when the stream is HDR (BT.2020 PQ): the buffer is 10-bit P010 and the presenter must
/// configure EDR + BT.2020 PQ output. Derived from the decoded buffer's pixel format.
public let isHDR: Bool
/// The AU's wire `user_flags` (`AccessUnit.flags`), threaded through the decode via the frame
/// context so the re-anchor gate can classify this decoded frame (IDR / RFI anchor / recovery
/// mark) at present time — the async decode callback has no other access to it. 0 when unknown.
public let flags: UInt32
}
/// Per-frame context threaded through the VideoToolbox frame refcon: the AU's receipt instant (for
/// the decode-stage meter) and its wire `user_flags` (for the re-anchor gate). Retained across the
/// async decode and reclaimed exactly once — by the output callback for every frame VideoToolbox
/// accepts, or by `decode`'s error branch for a frame `DecodeFrame` rejected outright (the callback
/// then never fires). A tiny per-frame allocation, the price of smuggling two values (a 64-bit
/// instant plus the flags) through the single `void*` a bit-pattern scalar can't hold.
private final class FrameContext {
let receivedNs: Int64
let flags: UInt32
init(receivedNs: Int64, flags: UInt32) {
self.receivedNs = receivedNs
self.flags = flags
}
}
/// The C output callback can't capture context, so VideoToolbox hands it the refcon we set at
/// session creation — a pointer back to the owning `VideoDecoder`. The per-frame refcon carries
/// the AU's `receivedNs` as a pointer bit pattern (a scalar smuggled through the C void*, never
/// dereferenced) so the decode stage can be computed against decode-completion.
/// session creation — a pointer back to the owning `VideoDecoder`. The per-frame refcon is the
/// retained `FrameContext` set at submit; reclaim it here (balancing `passRetained`) and unpack the
/// AU's receipt instant (for the decode stage) and wire flags (for the re-anchor gate).
private let decoderOutputCallback: VTDecompressionOutputCallback = {
refcon, frameRefcon, status, _, imageBuffer, pts, _ in
guard let refcon else { return }
let receivedNs = frameRefcon.map { Int64(Int(bitPattern: $0)) } ?? 0
let ctx = frameRefcon.map { Unmanaged<FrameContext>.fromOpaque($0).takeRetainedValue() }
Unmanaged<VideoDecoder>.fromOpaque(refcon)
.takeUnretainedValue()
.handleDecoded(status: status, imageBuffer: imageBuffer, pts: pts, receivedNs: receivedNs)
.handleDecoded(
status: status, imageBuffer: imageBuffer, pts: pts,
receivedNs: ctx?.receivedNs ?? 0, flags: ctx?.flags ?? 0)
}
/// Owns a `VTDecompressionSession` rebuilt whenever the format description changes (every IDR /
@@ -117,16 +138,21 @@ public final class VideoDecoder: @unchecked Sendable {
let sample = codec.sampleBuffer(au: au, format: newFormat)
else { lock.unlock(); return false }
var infoOut = VTDecodeInfoFlags()
// The AU's receipt instant + wire flags ride through as a retained context; the output
// callback reclaims it. Retain immediately before submit so no early return can leak it.
let ctx = FrameContext(receivedNs: au.receivedNs, flags: au.flags)
let refcon = Unmanaged.passRetained(ctx).toOpaque()
let status = VTDecompressionSessionDecodeFrame(
session,
sampleBuffer: sample,
flags: [._EnableAsynchronousDecompression],
// The AU's receipt instant rides through as a bit pattern (nil for 0 — the output
// callback maps that back to 0); the callback needs it to stamp the decode stage.
frameRefcon: UnsafeMutableRawPointer(bitPattern: Int(au.receivedNs)),
frameRefcon: refcon,
infoFlagsOut: &infoOut)
lock.unlock()
if status != noErr {
// DecodeFrame rejected the frame outright — the output callback will NOT fire, so
// reclaim the context here (balancing passRetained) to avoid leaking it.
Unmanaged<FrameContext>.fromOpaque(refcon).release()
onDecodeError(status)
return false
}
@@ -231,9 +257,10 @@ public final class VideoDecoder: @unchecked Sendable {
}
/// VT thread. Stamp decode-completion and enqueue, or report the error. `receivedNs` is the
/// AU's receipt instant threaded through the frame refcon (0 = unknown).
/// AU's receipt instant and `flags` its wire `user_flags`, both threaded through the frame refcon
/// (0 = unknown).
fileprivate func handleDecoded(
status: OSStatus, imageBuffer: CVImageBuffer?, pts: CMTime, receivedNs: Int64
status: OSStatus, imageBuffer: CVImageBuffer?, pts: CMTime, receivedNs: Int64, flags: UInt32
) {
guard status == noErr, let imageBuffer else {
onDecodeError(status)
@@ -259,6 +286,6 @@ public final class VideoDecoder: @unchecked Sendable {
onDecoded(
ReadyFrame(
ptsNs: ptsNs, receivedNs: receivedNs, decodedNs: decodedNs,
pixelBuffer: imageBuffer, isHDR: isHDR))
pixelBuffer: imageBuffer, isHDR: isHDR, flags: flags))
}
}