Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2a62fe7857 |
@@ -43,10 +43,12 @@ 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. 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),
|
||||
/// A received access unit from the feeder, ready to queue into the decoder. The `u32` is the
|
||||
/// feeder's [`NativeClient::note_frame_index`] verdict — the forward frame-index gap's WIDTH
|
||||
/// (0 = none), so the loop arms the freeze gate with the same signal and pre-credits the
|
||||
/// reassembler's later `frames_dropped` climb for the loss (the feeder already fired the RFI
|
||||
/// request).
|
||||
Au(Frame, u32),
|
||||
/// 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).
|
||||
@@ -603,7 +605,11 @@ fn feeder_loop(
|
||||
// AU's first piece (or a whole delivery), so the RFI gap detector keeps
|
||||
// counting AUs.
|
||||
let au_first = frame.part.is_none_or(|p| p.first);
|
||||
let gap = au_first && client.note_frame_index(frame.frame_index);
|
||||
let gap = if au_first {
|
||||
client.note_frame_index(frame.frame_index)
|
||||
} else {
|
||||
0
|
||||
};
|
||||
// Park the receipt stamp (keyed by the pts the codec echoes) whenever the `decode`
|
||||
// stage is consumed: the HUD, or the ABR decode signal (`measure_decode`). The
|
||||
// HUD-only `received` point + host/network split stay gated on the overlay.
|
||||
@@ -691,9 +697,12 @@ fn dispatch_event(
|
||||
match ev {
|
||||
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());
|
||||
// fold `on_decoded` (keyed by the pts the codec will echo). Credited arm: the gap width
|
||||
// pre-covers the reassembler's ~120 ms-later `frames_dropped` climb for the same loss,
|
||||
// so a fast RFI anchor that heals in between isn't re-frozen by it (the double-arm
|
||||
// race — see `ReanchorGate::arm_expecting_drops`).
|
||||
if gap > 0 {
|
||||
gate.arm_expecting_drops(Instant::now(), u64::from(gap));
|
||||
}
|
||||
// One entry per AU (parts share the pts): the completing delivery carries it.
|
||||
if f.complete {
|
||||
|
||||
@@ -222,8 +222,13 @@ pub(super) fn run_sync(
|
||||
// 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());
|
||||
// Credited arm: the gap width pre-covers the reassembler's ~120 ms-later
|
||||
// `frames_dropped` climb for the same loss, so a fast RFI anchor that heals in
|
||||
// between isn't re-frozen by it (the double-arm race — see
|
||||
// `ReanchorGate::arm_expecting_drops`).
|
||||
let gap = client.note_frame_index(frame.frame_index);
|
||||
if gap > 0 {
|
||||
gate.arm_expecting_drops(Instant::now(), u64::from(gap));
|
||||
}
|
||||
// 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.
|
||||
|
||||
@@ -774,12 +774,22 @@ public final class PunktfunkConnection {
|
||||
/// `noteFrameIndex` (the throttled RFI request); call it for every received AU. Returns false
|
||||
/// after close.
|
||||
public func noteFrameIndexGap(_ frameIndex: UInt32) -> Bool {
|
||||
noteFrameIndexGapWidth(frameIndex) > 0
|
||||
}
|
||||
|
||||
/// Like `noteFrameIndexGap`, but reports the gap's WIDTH — how many frames this arrival revealed
|
||||
/// as missing (0 = none). The post-loss re-anchor gate arms with the width
|
||||
/// (`ReanchorGate.arm(expectingDrops:)`) so the reassembler's later `framesDropped` climb for
|
||||
/// the SAME loss cannot re-freeze a stream an RFI anchor already healed (the double-arm race).
|
||||
/// Same core side effect as `noteFrameIndex` (the throttled RFI request); call it for every
|
||||
/// received AU. Returns 0 after close.
|
||||
public func noteFrameIndexGapWidth(_ frameIndex: UInt32) -> UInt32 {
|
||||
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
|
||||
guard let h = handle, !closeRequested else { return 0 }
|
||||
var width: UInt32 = 0
|
||||
_ = punktfunk_connection_note_frame_index_ex(h, frameIndex, &width)
|
||||
return width
|
||||
}
|
||||
|
||||
/// Cumulative access units the host→client reassembler dropped as unrecoverable (FEC couldn't
|
||||
|
||||
@@ -55,6 +55,16 @@ final class ReanchorGate: @unchecked Sendable {
|
||||
lock.unlock()
|
||||
}
|
||||
|
||||
/// `arm()` for a loss detected as a frame-index gap of a known width
|
||||
/// (`PunktfunkConnection.noteFrameIndexGapWidth`). Pre-credits the reassembler's later
|
||||
/// `framesDropped` climb for the same lost frames, so `poll` doesn't re-freeze a stream an
|
||||
/// RFI anchor already healed (the double-arm race — the Rust gate's docs tell the story).
|
||||
func arm(expectingDrops: UInt64) {
|
||||
lock.lock()
|
||||
punktfunk_reanchor_gate_arm_expecting_drops(ptr, expectingDrops)
|
||||
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.
|
||||
|
||||
@@ -921,7 +921,11 @@ public final class Stage2Pipeline {
|
||||
// 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() }
|
||||
// Credited arm: the gap width pre-covers the reassembler's ~120 ms-later
|
||||
// framesDropped climb for the same loss, so a fast RFI anchor that heals in
|
||||
// between isn't re-frozen by it (the double-arm race).
|
||||
let gapWidth = connection.noteFrameIndexGapWidth(au.frameIndex)
|
||||
if gapWidth > 0 { reanchorGate.arm(expectingDrops: UInt64(gapWidth)) }
|
||||
onFrame?(au)
|
||||
if let f = connection.videoCodec.formatDescription(fromKeyframe: au.data) {
|
||||
format = f // refreshed on every IDR (mode changes included)
|
||||
|
||||
@@ -100,7 +100,11 @@ final class StreamPump {
|
||||
// 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.
|
||||
// The same gap is the earliest, most precise signal to ARM the display freeze.
|
||||
if connection.noteFrameIndexGap(au.frameIndex) { gate.arm() }
|
||||
// Credited arm: the gap width pre-covers the reassembler's ~120 ms-later
|
||||
// framesDropped climb for the same loss, so a fast RFI anchor that heals in
|
||||
// between isn't re-frozen by it (the double-arm race).
|
||||
let gapWidth = connection.noteFrameIndexGapWidth(au.frameIndex)
|
||||
if gapWidth > 0 { gate.arm(expectingDrops: UInt64(gapWidth)) }
|
||||
onFrame?(au)
|
||||
let idrFormat = connection.videoCodec.formatDescription(fromKeyframe: au.data)
|
||||
if let f = idrFormat {
|
||||
|
||||
@@ -885,7 +885,12 @@ fn pump(
|
||||
Some(exp) => {
|
||||
if let Some(gap) = index_gap(exp, frame.frame_index) {
|
||||
let now = Instant::now();
|
||||
gate.arm(now);
|
||||
// Credited arm: the reassembler books these same lost frames into
|
||||
// `frames_dropped` up to ~120 ms from now; the credit keeps that
|
||||
// delayed climb from re-freezing a stream the RFI anchor healed in
|
||||
// between (the double-arm race — see
|
||||
// `ReanchorGate::arm_expecting_drops`).
|
||||
gate.arm_expecting_drops(now, u64::from(gap));
|
||||
next_expected_index = Some(frame.frame_index.wrapping_add(1));
|
||||
// The gap carries the PRECISE lost range — [first missing, newest
|
||||
// received - 1] — so this is the one recovery signal that can drive true
|
||||
|
||||
@@ -4346,7 +4346,41 @@ pub unsafe extern "C" fn punktfunk_connection_note_frame_index(
|
||||
if !gap_out.is_null() {
|
||||
// SAFETY: per the ABI contract - a caller-owned out-param, non-null on this path,
|
||||
// written once by value.
|
||||
unsafe { *gap_out = gap };
|
||||
unsafe { *gap_out = gap > 0 };
|
||||
}
|
||||
PunktfunkStatus::Ok
|
||||
})
|
||||
}
|
||||
|
||||
/// [`punktfunk_connection_note_frame_index`] with the gap WIDTH instead of a yes/no: writes to
|
||||
/// `gap_width_out` how many frames this arrival revealed as missing (0 = contiguous/straggler).
|
||||
/// A client with a post-loss display freeze passes the width to
|
||||
/// [`punktfunk_reanchor_gate_arm_expecting_drops`] so the reassembler's later `frames_dropped`
|
||||
/// climb for the SAME loss cannot re-freeze a stream an RFI anchor already healed (the double-arm
|
||||
/// race — see the gate function's doc).
|
||||
///
|
||||
/// # Safety
|
||||
/// `c` is a valid connection handle; `gap_width_out` is writable or NULL.
|
||||
#[cfg(feature = "quic")]
|
||||
#[no_mangle]
|
||||
pub unsafe extern "C" fn punktfunk_connection_note_frame_index_ex(
|
||||
c: *const PunktfunkConnection,
|
||||
frame_index: u32,
|
||||
gap_width_out: *mut u32,
|
||||
) -> PunktfunkStatus {
|
||||
guard(|| {
|
||||
// SAFETY: per the ABI contract - an opaque handle from a `*_new`/`*_pair` that the caller
|
||||
// has not yet freed, or null, which `as_mut`/`as_ref` reports as `None` and the `match`
|
||||
// here handles.
|
||||
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_width_out.is_null() {
|
||||
// SAFETY: per the ABI contract - a caller-owned out-param, non-null on this path,
|
||||
// written once by value.
|
||||
unsafe { *gap_width_out = gap };
|
||||
}
|
||||
PunktfunkStatus::Ok
|
||||
})
|
||||
@@ -4697,6 +4731,31 @@ pub unsafe extern "C" fn punktfunk_reanchor_gate_arm(g: *mut ReanchorGate) {
|
||||
});
|
||||
}
|
||||
|
||||
/// [`punktfunk_reanchor_gate_arm`] for a loss detected as a **frame-index gap**, where the caller
|
||||
/// knows how many frames the gap skipped ([`punktfunk_connection_note_frame_index_ex`]). On top of
|
||||
/// arming, the gate pre-credits the reassembler's `frames_dropped` climb those same lost frames
|
||||
/// will produce up to ~120 ms later, so [`punktfunk_reanchor_gate_poll`] does not treat that
|
||||
/// delayed bookkeeping as a SECOND loss — without the credit, a fast LTR-RFI anchor lifts the
|
||||
/// freeze between the two signals and the stale climb re-freezes a healed stream (the double-arm
|
||||
/// race). Use the plain arm for non-gap loss signals (decoder wedge/demotion). NULL is a no-op.
|
||||
///
|
||||
/// # Safety
|
||||
/// `g` is a valid gate handle.
|
||||
#[no_mangle]
|
||||
pub unsafe extern "C" fn punktfunk_reanchor_gate_arm_expecting_drops(
|
||||
g: *mut ReanchorGate,
|
||||
expected_drops: u64,
|
||||
) {
|
||||
guard_void(|| {
|
||||
// SAFETY: per the ABI contract - an opaque handle from a `*_new`/`*_pair` that the caller has
|
||||
// not yet freed, or null, which `as_mut`/`as_ref` reports as `None` and the `match` here
|
||||
// handles.
|
||||
if let Some(g) = unsafe { g.as_mut() } {
|
||||
g.arm_expecting_drops(std::time::Instant::now(), expected_drops);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// Fold one decoded frame and write to `out_present` whether to display it (`true`) or withhold it as
|
||||
/// a post-loss concealment (`false`). `flags` is the AU's `user_flags` word ([`PunktfunkFrame::flags`]):
|
||||
/// the gate reads `FLAG_SOF` (the host's IDR marker), `USER_FLAG_RECOVERY_ANCHOR` and
|
||||
|
||||
@@ -881,14 +881,20 @@ impl NativeClient {
|
||||
///
|
||||
/// 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.
|
||||
/// loop stays the backstop for when the recovery frame itself is lost. Returns the gap WIDTH —
|
||||
/// how many frames this arrival revealed as missing, `0` when none (contiguous or straggler),
|
||||
/// whether or not the RFI was throttled — so a client with a post-loss display freeze can
|
||||
/// (re-)arm it on the same signal AND pre-credit the reassembler's later `frames_dropped` climb
|
||||
/// for the same loss ([`ReanchorGate::arm_expecting_drops`] — without the credit, a fast
|
||||
/// LTR-RFI anchor lifts the freeze before the climb books the loss, and the stale climb then
|
||||
/// re-freezes the healed stream).
|
||||
///
|
||||
/// 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 {
|
||||
///
|
||||
/// [`ReanchorGate::arm_expecting_drops`]: crate::reanchor::ReanchorGate::arm_expecting_drops
|
||||
pub fn note_frame_index(&self, frame_index: u32) -> u32 {
|
||||
// Decide (and update state) under the lock; fire the request after releasing it.
|
||||
let (gap, ask) = self
|
||||
.rfi
|
||||
|
||||
@@ -32,14 +32,16 @@ pub(crate) enum RecoveryAsk {
|
||||
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, ask)`.
|
||||
/// `gap` is whether this frame revealed a forward gap (the embedder arms its post-loss display
|
||||
/// freeze on it); `ask` is the (throttled) recovery request to fire — an RFI naming the exact
|
||||
/// lost span, or a keyframe when the span exceeds [`crate::packet::RFI_MAX_RANGE`] (RFI is
|
||||
/// hopeless there: no encoder holds references that old, and a huge jump is more likely a
|
||||
/// resync — e.g. the first real AU after an old host's speed test — than a real loss). Split
|
||||
/// out from the connection so the wrapping arithmetic + [`RFI_THROTTLE`] are unit-testable
|
||||
/// without a live session (see the tests below).
|
||||
pub(crate) fn observe(&mut self, frame_index: u32, now: Instant) -> (bool, RecoveryAsk) {
|
||||
/// `gap` is how many frames this arrival revealed as missing — 0 for contiguous/straggler; the
|
||||
/// embedder arms its post-loss display freeze on a non-zero gap, and the WIDTH lets it
|
||||
/// pre-credit the reassembler's later `frames_dropped` climb for the same loss
|
||||
/// ([`crate::reanchor::ReanchorGate::arm_expecting_drops`] — the double-arm race). `ask` is the
|
||||
/// (throttled) recovery request to fire — an RFI naming the exact lost span, or a keyframe when
|
||||
/// the span exceeds [`crate::packet::RFI_MAX_RANGE`] (RFI is hopeless there: no encoder holds
|
||||
/// references that old, and a huge jump is more likely a resync — e.g. the first real AU after
|
||||
/// an old host's speed test — than a real loss). Split out from the connection so the wrapping
|
||||
/// arithmetic + [`RFI_THROTTLE`] are unit-testable without a live session (see the tests below).
|
||||
pub(crate) fn observe(&mut self, frame_index: u32, now: Instant) -> (u32, RecoveryAsk) {
|
||||
match self.next_expected {
|
||||
Some(exp) => {
|
||||
// Wrapping split at the half-space: a small positive delta is a forward gap
|
||||
@@ -47,10 +49,11 @@ impl RfiRecovery {
|
||||
let ahead = frame_index.wrapping_sub(exp);
|
||||
if ahead == 0 {
|
||||
self.next_expected = Some(frame_index.wrapping_add(1)); // contiguous
|
||||
(false, RecoveryAsk::None)
|
||||
(0, RecoveryAsk::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 recovery ask for the lost range.
|
||||
// Forward gap: [exp, frame_index-1] lost (`ahead` frames). Advance past this
|
||||
// frame so the same gap isn't re-detected, then fire a throttled recovery ask
|
||||
// for the lost range.
|
||||
self.next_expected = Some(frame_index.wrapping_add(1));
|
||||
let send = self
|
||||
.last_req
|
||||
@@ -65,15 +68,15 @@ impl RfiRecovery {
|
||||
} else {
|
||||
RecoveryAsk::Rfi(exp, frame_index.wrapping_sub(1))
|
||||
};
|
||||
(true, ask)
|
||||
(ahead, ask)
|
||||
} else {
|
||||
// Straggler behind the delivery point — leave the expectation.
|
||||
(false, RecoveryAsk::None)
|
||||
(0, RecoveryAsk::None)
|
||||
}
|
||||
}
|
||||
None => {
|
||||
self.next_expected = Some(frame_index.wrapping_add(1));
|
||||
(false, RecoveryAsk::None)
|
||||
(0, RecoveryAsk::None)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -96,7 +99,7 @@ mod rfi_recovery_tests {
|
||||
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, RecoveryAsk::None));
|
||||
assert_eq!(r.observe(100, base()), (0, RecoveryAsk::None));
|
||||
assert_eq!(r.next_expected, Some(101));
|
||||
}
|
||||
|
||||
@@ -105,9 +108,9 @@ mod rfi_recovery_tests {
|
||||
let mut r = RfiRecovery::default();
|
||||
let t = base();
|
||||
r.observe(100, t);
|
||||
assert_eq!(r.observe(101, t), (false, RecoveryAsk::None));
|
||||
assert_eq!(r.observe(102, t), (false, RecoveryAsk::None));
|
||||
assert_eq!(r.observe(103, t), (false, RecoveryAsk::None));
|
||||
assert_eq!(r.observe(101, t), (0, RecoveryAsk::None));
|
||||
assert_eq!(r.observe(102, t), (0, RecoveryAsk::None));
|
||||
assert_eq!(r.observe(103, t), (0, RecoveryAsk::None));
|
||||
assert_eq!(r.next_expected, Some(104));
|
||||
}
|
||||
|
||||
@@ -117,7 +120,7 @@ mod rfi_recovery_tests {
|
||||
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, RecoveryAsk::Rfi(101, 104)));
|
||||
assert_eq!(r.observe(105, t), (4, RecoveryAsk::Rfi(101, 104)));
|
||||
// The expectation advances past the delivered frame so the same gap can't re-fire.
|
||||
assert_eq!(r.next_expected, Some(106));
|
||||
}
|
||||
@@ -128,7 +131,7 @@ mod rfi_recovery_tests {
|
||||
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, RecoveryAsk::Rfi(101, 101)));
|
||||
assert_eq!(r.observe(102, t), (1, RecoveryAsk::Rfi(101, 101)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -137,16 +140,16 @@ mod rfi_recovery_tests {
|
||||
let t0 = base();
|
||||
r.observe(100, t0);
|
||||
// First gap fires the request and stamps the throttle.
|
||||
assert_eq!(r.observe(105, t0), (true, RecoveryAsk::Rfi(101, 104)));
|
||||
assert_eq!(r.observe(105, t0), (4, RecoveryAsk::Rfi(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, RecoveryAsk::None)
|
||||
(4, RecoveryAsk::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, RecoveryAsk::Rfi(111, 119))
|
||||
(9, RecoveryAsk::Rfi(111, 119))
|
||||
);
|
||||
}
|
||||
|
||||
@@ -158,7 +161,7 @@ mod rfi_recovery_tests {
|
||||
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, RecoveryAsk::None));
|
||||
assert_eq!(r.observe(103, t), (0, RecoveryAsk::None));
|
||||
assert_eq!(r.next_expected, Some(106));
|
||||
}
|
||||
|
||||
@@ -167,9 +170,9 @@ mod rfi_recovery_tests {
|
||||
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, RecoveryAsk::None)); // contiguous, wraps to 0
|
||||
assert_eq!(r.observe(u32::MAX, t), (0, RecoveryAsk::None)); // contiguous, wraps to 0
|
||||
assert_eq!(r.next_expected, Some(0));
|
||||
assert_eq!(r.observe(0, t), (false, RecoveryAsk::None)); // still contiguous across the wrap
|
||||
assert_eq!(r.observe(0, t), (0, RecoveryAsk::None)); // still contiguous across the wrap
|
||||
assert_eq!(r.next_expected, Some(1));
|
||||
}
|
||||
|
||||
@@ -179,7 +182,7 @@ mod rfi_recovery_tests {
|
||||
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, RecoveryAsk::Rfi(u32::MAX, 0)));
|
||||
assert_eq!(r.observe(1, t), (2, RecoveryAsk::Rfi(u32::MAX, 0)));
|
||||
assert_eq!(r.next_expected, Some(2));
|
||||
}
|
||||
|
||||
@@ -192,14 +195,14 @@ mod rfi_recovery_tests {
|
||||
// reference exists for an RFI, and the jump may be a phantom (an old host's
|
||||
// speed-test burst consuming video indexes) — ask for the IDR resync instead.
|
||||
let jump = 100 + crate::packet::RFI_MAX_RANGE + 2;
|
||||
assert_eq!(r.observe(jump, t), (true, RecoveryAsk::Keyframe));
|
||||
assert_eq!(r.observe(jump, t), (jump - 101, RecoveryAsk::Keyframe));
|
||||
// The expectation still advances past the delivered frame (no re-fire on the next one).
|
||||
assert_eq!(r.next_expected, Some(jump + 1));
|
||||
assert_eq!(r.observe(jump + 1, t), (false, RecoveryAsk::None));
|
||||
assert_eq!(r.observe(jump + 1, t), (0, RecoveryAsk::None));
|
||||
// A huge gap consumes the shared throttle too — an immediate follow-up gap stays quiet.
|
||||
assert_eq!(
|
||||
r.observe(jump + 10, t + Duration::from_millis(1)),
|
||||
(true, RecoveryAsk::None)
|
||||
(8, RecoveryAsk::None)
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -64,6 +64,26 @@ pub const REANCHOR_MARKS_TO_LIFT: u32 = 2;
|
||||
/// floor fires, so a real stall still recovers.
|
||||
pub const RECOVERY_MARK_PATIENCE: Duration = Duration::from_millis(1500);
|
||||
|
||||
/// How long a frame-index-gap arm's expected `frames_dropped` climb stays pre-credited in
|
||||
/// [`ReanchorGate::poll`]. One loss arms the gate through TWO signals: the frame-index gap the
|
||||
/// instant the AU after the loss is delivered ([`ReanchorGate::arm_expecting_drops`]), and the
|
||||
/// reassembler's `frames_dropped` climb once the lost frame ages out of its loss window (~120 ms
|
||||
/// later, and only when at least one of its packets arrived). Without the credit, a *fast* recovery
|
||||
/// — an LTR-RFI anchor typically lands within ~60 ms — lifts the freeze between the two signals,
|
||||
/// and the stale climb then re-freezes a stream that is already bit-exact healed; the host swallows
|
||||
/// the resulting keyframe request as an echo of the very RFI that healed it, so the picture stays
|
||||
/// frozen until the [`REANCHOR_FREEZE_MAX`] overdue re-ask extracts a full IDR (the field
|
||||
/// "H265 freezes on every loss, AV1 fine" signature — the slower IDR path usually lands after the
|
||||
/// climb and dodged the race).
|
||||
///
|
||||
/// Sized to cover the reassembler's 120 ms loss window plus delivery jitter with a wide margin,
|
||||
/// while staying short enough that a leftover credit (a straggler that filled the gap late, so no
|
||||
/// climb ever came; or a whole-frame vanish the reassembler never saw a packet of) cannot mask a
|
||||
/// genuinely unrelated future climb for long. A masked climb is also never silent in practice:
|
||||
/// every unrecoverable loss reveals itself as a frame-index gap on the next delivered frame, which
|
||||
/// re-arms (and re-credits) through [`ReanchorGate::arm_expecting_drops`] on its own.
|
||||
pub const DROP_CREDIT_WINDOW: Duration = Duration::from_millis(1000);
|
||||
|
||||
/// Frames skipped when `got` arrives while `expected` was the next index, or `None` if `got` is
|
||||
/// contiguous (`== expected`) or a straggler we have already passed. Frame indices are u32 counters
|
||||
/// that wrap, so the "ahead" test is a wrapping subtraction split at the half-space: a small positive
|
||||
@@ -185,6 +205,14 @@ pub struct ReanchorGate {
|
||||
/// a client stamps the decoder's decode-order watermark whenever this counter moves and
|
||||
/// discards the local recovery of anything older. Every other client ignores it.
|
||||
arms: u64,
|
||||
/// `frames_dropped` climb still expected from losses that already armed via a frame-index gap
|
||||
/// ([`Self::arm_expecting_drops`]). [`Self::poll`] consumes climbs against this before treating
|
||||
/// them as a NEW loss, so the reassembler's delayed bookkeeping of a gap-armed (and possibly
|
||||
/// already anchor-healed) loss cannot re-freeze the stream — see [`DROP_CREDIT_WINDOW`].
|
||||
drop_credit: u64,
|
||||
/// When the outstanding [`Self::drop_credit`] lapses ([`DROP_CREDIT_WINDOW`] after the latest
|
||||
/// credited arm). `None` when no credit is outstanding.
|
||||
drop_credit_expiry: Option<Instant>,
|
||||
}
|
||||
|
||||
impl ReanchorGate {
|
||||
@@ -199,6 +227,8 @@ impl ReanchorGate {
|
||||
last_dropped: frames_dropped,
|
||||
local_sei_since_arm: false,
|
||||
arms: 0,
|
||||
drop_credit: 0,
|
||||
drop_credit_expiry: None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -230,6 +260,20 @@ impl ReanchorGate {
|
||||
self.deadline = Some(now + REANCHOR_FREEZE_MAX);
|
||||
}
|
||||
|
||||
/// [`arm`](Self::arm) for a loss detected as a **frame-index gap**, where the caller knows how
|
||||
/// many frames the gap skipped. On top of arming, it pre-credits the reassembler's
|
||||
/// `frames_dropped` climb those same lost frames will produce up to ~120 ms later (its
|
||||
/// loss-window age-out), so [`poll`](Self::poll) does not treat that delayed bookkeeping as a
|
||||
/// SECOND loss. Without the credit a fast LTR-RFI anchor lifts the freeze between the two
|
||||
/// signals and the stale climb re-freezes a healed stream — the double-arm race
|
||||
/// ([`DROP_CREDIT_WINDOW`] tells the whole story). Use plain [`arm`](Self::arm) for every
|
||||
/// non-gap loss signal (decoder wedge/demotion), which has no climb to credit.
|
||||
pub fn arm_expecting_drops(&mut self, now: Instant, expected_drops: u64) {
|
||||
self.arm(now);
|
||||
self.drop_credit = self.drop_credit.saturating_add(expected_drops);
|
||||
self.drop_credit_expiry = Some(now + DROP_CREDIT_WINDOW);
|
||||
}
|
||||
|
||||
/// Fold the client's OWN recovery-point observation for one decoded frame, BEFORE handing that
|
||||
/// frame to [`on_decoded`](Self::on_decoded). Returns `true` when it lifted the freeze.
|
||||
///
|
||||
@@ -333,16 +377,36 @@ impl ReanchorGate {
|
||||
}
|
||||
|
||||
/// Periodic fold of the session's `frames_dropped` counter plus the overdue backstop. Returns
|
||||
/// `true` when the client should (throttled) request a keyframe: either the drop count climbed (a
|
||||
/// fresh unrecoverable loss — arm the freeze) or the freeze has held a full [`REANCHOR_FREEZE_MAX`]
|
||||
/// window with no re-anchor (re-ask and keep holding — NEVER resume to the concealed picture; a
|
||||
/// genuinely dead stream is the QUIC idle-timeout watchdog's job, not the gate's).
|
||||
/// `true` when the client should (throttled) request a keyframe: either the drop count climbed by
|
||||
/// more than the outstanding gap-arm credit (a fresh unrecoverable loss — arm the freeze) or the
|
||||
/// freeze has held a full [`REANCHOR_FREEZE_MAX`] window with no re-anchor (re-ask and keep
|
||||
/// holding — NEVER resume to the concealed picture; a genuinely dead stream is the QUIC
|
||||
/// idle-timeout watchdog's job, not the gate's).
|
||||
///
|
||||
/// A climb covered by [`arm_expecting_drops`](Self::arm_expecting_drops)' credit is the
|
||||
/// reassembler's delayed bookkeeping of a loss this gate already armed for — it must neither
|
||||
/// re-arm (an LTR-RFI anchor may have healed the stream in the meantime; re-freezing it is the
|
||||
/// double-arm race) nor ask again (the gap already fired the precise RFI, and if THAT recovery
|
||||
/// was lost the overdue backstop still re-asks at the [`REANCHOR_FREEZE_MAX`] deadline the
|
||||
/// gap-arm set — which is also sooner than the deadline a re-arm here would push out to).
|
||||
pub fn poll(&mut self, frames_dropped: u64, now: Instant) -> bool {
|
||||
let mut want_keyframe = false;
|
||||
if frames_dropped > self.last_dropped {
|
||||
let climb = frames_dropped - self.last_dropped;
|
||||
self.last_dropped = frames_dropped;
|
||||
self.arm(now);
|
||||
want_keyframe = true;
|
||||
if self.drop_credit_expiry.is_some_and(|e| now >= e) {
|
||||
self.drop_credit = 0;
|
||||
self.drop_credit_expiry = None;
|
||||
}
|
||||
let credited = climb.min(self.drop_credit);
|
||||
self.drop_credit -= credited;
|
||||
if self.drop_credit == 0 {
|
||||
self.drop_credit_expiry = None;
|
||||
}
|
||||
if climb > credited {
|
||||
self.arm(now);
|
||||
want_keyframe = true;
|
||||
}
|
||||
}
|
||||
if self.awaiting && self.deadline.is_some_and(|d| now >= d) {
|
||||
self.deadline = Some(now + REANCHOR_FREEZE_MAX);
|
||||
@@ -542,6 +606,84 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_rfi_anchor_is_not_refrozen_by_the_same_losss_drop_climb() {
|
||||
// The double-arm race (field: "H265 freezes on every loss, AV1 fine"): a loss arms via
|
||||
// the frame-index gap at T+10ms, the LTR-RFI anchor heals at T+60ms, and the reassembler
|
||||
// books the SAME loss into frames_dropped at ~T+130ms. The credited arm must keep that
|
||||
// stale climb from re-freezing the healed stream (and from re-asking — the host would
|
||||
// swallow the ask as an RFI echo and the picture would freeze until a forced IDR).
|
||||
let mut g = ReanchorGate::new(0);
|
||||
let t = t0();
|
||||
g.arm_expecting_drops(t + Duration::from_millis(10), 1); // gap of one lost frame + RFI
|
||||
assert_eq!(
|
||||
g.on_decoded(ANCHOR, false, t + Duration::from_millis(60)),
|
||||
GateVerdict::Present,
|
||||
"the anchor lifts"
|
||||
);
|
||||
assert!(
|
||||
!g.poll(1, t + Duration::from_millis(130)),
|
||||
"the credited climb must not ask again"
|
||||
);
|
||||
assert!(!g.is_holding(), "and must not re-freeze the healed stream");
|
||||
assert_eq!(
|
||||
g.on_decoded(0, false, t + Duration::from_millis(141)),
|
||||
GateVerdict::Present,
|
||||
"healthy P-frames keep presenting"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_climb_beyond_the_credit_is_a_fresh_loss_and_arms() {
|
||||
// The credit covers exactly the gap's frames; a bigger climb means MORE loss than the gap
|
||||
// accounted for (an interleaved partial-frame loss) — that part must still arm and ask.
|
||||
let mut g = ReanchorGate::new(0);
|
||||
let t = t0();
|
||||
g.arm_expecting_drops(t, 2);
|
||||
g.on_decoded(ANCHOR, false, t + Duration::from_millis(50)); // healed the credited loss
|
||||
assert!(
|
||||
g.poll(3, t + Duration::from_millis(130)),
|
||||
"one uncredited drop → ask"
|
||||
);
|
||||
assert!(g.is_holding(), "and re-arm for the uncredited part");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_drop_credit_expires_so_a_late_climb_still_arms() {
|
||||
// A straggler can fill the gap late (no climb ever comes) — the leftover credit must not
|
||||
// linger and mask a genuinely NEW loss later. Past DROP_CREDIT_WINDOW the credit is void.
|
||||
let mut g = ReanchorGate::new(0);
|
||||
let t = t0();
|
||||
g.arm_expecting_drops(t, 1);
|
||||
g.on_decoded(ANCHOR, false, t + Duration::from_millis(50));
|
||||
let late = t + DROP_CREDIT_WINDOW + Duration::from_millis(1);
|
||||
assert!(
|
||||
g.poll(1, late),
|
||||
"an expired credit no longer absorbs climbs"
|
||||
);
|
||||
assert!(g.is_holding());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_credited_climb_keeps_the_unhealed_freezes_original_deadline() {
|
||||
// When the recovery never arrives, consuming the climb must not silence the gate: the
|
||||
// overdue backstop still re-asks — at the deadline the GAP arm set, which is sooner than
|
||||
// the deadline a climb re-arm would have pushed out to.
|
||||
let mut g = ReanchorGate::new(0);
|
||||
let t = t0();
|
||||
g.arm_expecting_drops(t, 1); // RFI fired here; assume its anchor is lost in transit
|
||||
assert!(
|
||||
!g.poll(1, t + Duration::from_millis(130)),
|
||||
"credited climb: no early re-ask"
|
||||
);
|
||||
assert!(g.is_holding(), "still frozen — nothing healed it");
|
||||
assert!(
|
||||
g.poll(1, t + REANCHOR_FREEZE_MAX + Duration::from_millis(1)),
|
||||
"the overdue backstop still re-asks on the gap-arm's own deadline"
|
||||
);
|
||||
assert!(g.is_holding(), "and keeps holding, never resuming to gray");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_no_output_streak_trips_at_three() {
|
||||
let mut g = ReanchorGate::new(0);
|
||||
|
||||
@@ -3319,6 +3319,21 @@ PunktfunkStatus punktfunk_connection_note_frame_index(const PunktfunkConnection
|
||||
bool *gap_out);
|
||||
#endif
|
||||
|
||||
#if defined(PUNKTFUNK_FEATURE_QUIC)
|
||||
// [`punktfunk_connection_note_frame_index`] with the gap WIDTH instead of a yes/no: writes to
|
||||
// `gap_width_out` how many frames this arrival revealed as missing (0 = contiguous/straggler).
|
||||
// A client with a post-loss display freeze passes the width to
|
||||
// [`punktfunk_reanchor_gate_arm_expecting_drops`] so the reassembler's later `frames_dropped`
|
||||
// climb for the SAME loss cannot re-freeze a stream an RFI anchor already healed (the double-arm
|
||||
// race — see the gate function's doc).
|
||||
//
|
||||
// # Safety
|
||||
// `c` is a valid connection handle; `gap_width_out` is writable or NULL.
|
||||
PunktfunkStatus punktfunk_connection_note_frame_index_ex(const PunktfunkConnection *c,
|
||||
uint32_t frame_index,
|
||||
uint32_t *gap_width_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`]
|
||||
@@ -3442,6 +3457,18 @@ void punktfunk_reanchor_gate_free(ReanchorGate *g);
|
||||
// `g` is a valid gate handle.
|
||||
void punktfunk_reanchor_gate_arm(ReanchorGate *g);
|
||||
|
||||
// [`punktfunk_reanchor_gate_arm`] for a loss detected as a **frame-index gap**, where the caller
|
||||
// knows how many frames the gap skipped ([`punktfunk_connection_note_frame_index_ex`]). On top of
|
||||
// arming, the gate pre-credits the reassembler's `frames_dropped` climb those same lost frames
|
||||
// will produce up to ~120 ms later, so [`punktfunk_reanchor_gate_poll`] does not treat that
|
||||
// delayed bookkeeping as a SECOND loss — without the credit, a fast LTR-RFI anchor lifts the
|
||||
// freeze between the two signals and the stale climb re-freezes a healed stream (the double-arm
|
||||
// race). Use the plain arm for non-gap loss signals (decoder wedge/demotion). NULL is a no-op.
|
||||
//
|
||||
// # Safety
|
||||
// `g` is a valid gate handle.
|
||||
void punktfunk_reanchor_gate_arm_expecting_drops(ReanchorGate *g, uint64_t expected_drops);
|
||||
|
||||
// Fold one decoded frame and write to `out_present` whether to display it (`true`) or withhold it as
|
||||
// a post-loss concealment (`false`). `flags` is the AU's `user_flags` word ([`PunktfunkFrame::flags`]):
|
||||
// the gate reads `FLAG_SOF` (the host's IDR marker), `USER_FLAG_RECOVERY_ANCHOR` and
|
||||
|
||||
Reference in New Issue
Block a user