forked from unom/punktfunk
enricobuehler
e9a7373c76
The controller's two throughput-driven gates both compare "what the pipeline carried" against the ENCODER's target: the utilization gate asks whether a clean window actually tested that target (a calm menu proves nothing), and the never-decaying proven mark bounds how far every later climb may step. Both were fed `bytes_received`, which counts every accepted datagram — headers, FEC parity, probe filler, audio. So the figure rose with the redundancy the host adds in ANSWER to loss: at 25 % FEC the gate passed with the encoder emitting ~55 % of target, and the proven mark inherited the same inflation permanently. The signal was weakest exactly on the lossy links it exists for. Count data-shard payload separately at the reassembler's routing decision — the same place, and for the same reason, the probe counters are already stamped — and feed the ABR that. First time both gates are dimensionally honest: a media rate compared against a media target.
697 lines
42 KiB
Rust
697 lines
42 KiB
Rust
//! The blocking data-plane pump: poll the session for access units, run the adaptive-FEC
|
||
//! loss reports, the ABR controller + startup capacity probe, the jump-to-live detectors,
|
||
//! and the standing-latency bleed, and hand frames to the embedder.
|
||
|
||
use super::super::*;
|
||
use super::*;
|
||
|
||
/// Data-plane pump on a blocking thread: poll the session, hand frames to the embedder.
|
||
/// try_send drops the newest frame when the embedder lags (freshness over completeness).
|
||
/// Speed-test filler ([`FLAG_PROBE`]) is folded into the probe accumulator instead of the
|
||
/// decoder queue — it isn't video.
|
||
pub(super) struct DataPump {
|
||
pub(super) session: Session,
|
||
pub(super) frames: Arc<FrameChannel>,
|
||
pub(super) ctrl_tx: tokio::sync::mpsc::Sender<CtrlRequest>,
|
||
pub(super) shutdown: Arc<std::sync::atomic::AtomicBool>,
|
||
pub(super) probe: Arc<Mutex<ProbeState>>,
|
||
pub(super) hot_tids: Arc<Mutex<Vec<i32>>>,
|
||
pub(super) clock_offset: Arc<std::sync::atomic::AtomicI64>,
|
||
pub(super) clock_gen: Arc<AtomicU32>,
|
||
pub(super) decode_lat: Arc<Mutex<DecodeLatAcc>>,
|
||
/// Host encode-stage latency window accumulator (the ABR encode signal — see
|
||
/// [`super::super::frame_channel::EncodeLatAcc`]); fed by the datagram task.
|
||
pub(super) encode_lat: Arc<Mutex<super::super::frame_channel::EncodeLatAcc>>,
|
||
/// Accepted-mode-switch generation (control task bumps): a change resets the controller's
|
||
/// mode-scoped learned state ([`BitrateController::on_mode_switch`]).
|
||
pub(super) mode_gen: Arc<AtomicU32>,
|
||
pub(super) frames_dropped: Arc<std::sync::atomic::AtomicU64>,
|
||
pub(super) fec_recovered: Arc<std::sync::atomic::AtomicU64>,
|
||
pub(super) bitrate_ack: Arc<Mutex<Option<u32>>>,
|
||
/// Outbound decode-recovery keyframe asks, counted by the control task at its send choke
|
||
/// point; drained per report window as the ABR's recovery signal.
|
||
pub(super) recovery_kf: Arc<AtomicU32>,
|
||
/// The embedder's REQUESTED rate (0 = Automatic — the only case the ABR arms).
|
||
pub(super) bitrate_kbps: u32,
|
||
/// The rate the host actually configured (echoed in Welcome).
|
||
pub(super) resolved_bitrate_kbps: u32,
|
||
pub(super) negotiated_codec: u8,
|
||
}
|
||
|
||
impl DataPump {
|
||
pub(super) fn run(self) {
|
||
let DataPump {
|
||
mut session,
|
||
frames,
|
||
ctrl_tx,
|
||
shutdown: pump_shutdown,
|
||
probe: pump_probe,
|
||
hot_tids: pump_hot_tids,
|
||
clock_offset: pump_clock_offset,
|
||
clock_gen: pump_clock_gen,
|
||
decode_lat: pump_decode_lat,
|
||
encode_lat: pump_encode_lat,
|
||
mode_gen: pump_mode_gen,
|
||
frames_dropped,
|
||
fec_recovered,
|
||
bitrate_ack,
|
||
recovery_kf: pump_recovery_kf,
|
||
bitrate_kbps,
|
||
resolved_bitrate_kbps,
|
||
negotiated_codec,
|
||
} = self;
|
||
pin_thread_user_interactive(); // feeds the frame channel → the user-interactive video pump
|
||
register_hot_tid(&pump_hot_tids); // this thread does UDP receive + FEC reassembly — hint it
|
||
// Adaptive-FEC loss reporting: every ADAPT_REPORT_INTERVAL, report the loss observed over the
|
||
// window (shards FEC recovered, plus a bump if any frame went unrecoverable) so the host can
|
||
// size FEC to the link. Suppressed during a speed test (its FLAG_PROBE filler would skew it).
|
||
const ADAPT_REPORT_INTERVAL: Duration = Duration::from_millis(750);
|
||
let mut last_report = Instant::now();
|
||
let (
|
||
mut last_recovered,
|
||
mut last_late,
|
||
mut last_received,
|
||
mut last_dropped,
|
||
mut last_bytes,
|
||
) = (0u64, 0u64, 0u64, 0u64, 0u64);
|
||
// PUNKTFUNK_PERF: per-window pump observability — the Session's receive stage split
|
||
// (recv / decrypt / reassemble+FEC, see `Session::take_pump_perf`) and completed-AU
|
||
// inter-arrival jitter. Smoothness has no metric otherwise: jump-to-live counters only
|
||
// fire after the stream is already seconds behind.
|
||
let pump_perf_on = std::env::var("PUNKTFUNK_PERF").is_ok_and(|v| v != "0");
|
||
let mut arrivals_us: Vec<u32> = Vec::new();
|
||
let mut last_arrival: Option<Instant> = None;
|
||
// Adaptive bitrate (see `crate::abr`): armed only when the embedder asked for Automatic
|
||
// (`bitrate_kbps == 0`) and the host echoed the rate it actually configured (an old host
|
||
// echoes 0 → controller stays permanently off). Fed once per report window with the same
|
||
// deltas the LossReport uses, plus the window's mean skew-corrected one-way delay, the
|
||
// actual delivered throughput (climb gate + proven-throughput mark), and whether a
|
||
// jump-to-live flush fired.
|
||
// PyroWave sessions PIN their rate (§4.6): AIMD descent turns wavelets to mush well
|
||
// above its floor, and the climb probe's VBV reasoning doesn't apply to hard
|
||
// per-frame CBR — controller and capacity probe stay off (0 = permanently off).
|
||
let rate_pinned = negotiated_codec == crate::quic::CODEC_PYROWAVE;
|
||
// All-intra streams have no reference chains: the frame channel drains to the newest
|
||
// AU instead of strict FIFO (see `FrameChannel::set_all_intra`), so a slow consumer
|
||
// caps its standing queue at ~1 frame with zero recovery cost.
|
||
frames.set_all_intra(negotiated_codec == crate::quic::CODEC_PYROWAVE);
|
||
let mut abr = BitrateController::new(if bitrate_kbps == 0 && !rate_pinned {
|
||
resolved_bitrate_kbps
|
||
} else {
|
||
0
|
||
});
|
||
// Startup link-capacity probe (Automatic sessions): the controller's ceiling is the
|
||
// negotiated start rate — the conservative 20 Mbps default, historically a box Automatic
|
||
// could NEVER climb out of. One speed-test burst shortly after the stream settles
|
||
// measures what the link actually delivers; ×0.7 (headroom for FEC overhead + variance)
|
||
// becomes the climb ceiling and slow start does the rest. Old hosts decline (all-zero
|
||
// reply) or never answer (timeout clears the state so LossReports resume) — either way
|
||
// the ceiling stays negotiated, exactly the old behavior. PUNKTFUNK_ABR_PROBE=0 opts out.
|
||
// `PUNKTFUNK_ABR_PROBE_KBPS` lowers the burst target (unset/0/garbage → the 2 Gbps
|
||
// default): the target is deliberately far above any plausible link so the burst measures
|
||
// the link and not itself, but on links the burst DISTURBS that backfires — a constrained
|
||
// Wi-Fi link can black-hole under 2 Gbps (measured on webOS: the probe hitting the 6 s
|
||
// timeout delayed first video to 14 s, and a "successful" one still reported
|
||
// send_dropped=20211), and a 2-3 core TV client starves decoding the firehose. An
|
||
// embedder that caps its own speed test wants this capped to match.
|
||
let capacity_probe_kbps: u32 = std::env::var("PUNKTFUNK_ABR_PROBE_KBPS")
|
||
.ok()
|
||
.and_then(|v| v.trim().parse::<u32>().ok())
|
||
.filter(|&v| v > 0)
|
||
.unwrap_or(2_000_000);
|
||
const CAPACITY_PROBE_MS: u32 = 800;
|
||
const CAPACITY_PROBE_DELAY: Duration = Duration::from_secs(2);
|
||
const CAPACITY_PROBE_TIMEOUT: Duration = Duration::from_secs(6);
|
||
let mut capacity_probe_at: Option<Instant> = (bitrate_kbps == 0
|
||
&& !rate_pinned
|
||
&& resolved_bitrate_kbps > 0
|
||
&& std::env::var("PUNKTFUNK_ABR_PROBE").map_or(true, |v| v != "0"))
|
||
.then(|| Instant::now() + CAPACITY_PROBE_DELAY);
|
||
let mut capacity_probe_deadline: Option<Instant> = None;
|
||
// Edge detector + watchdog for a probe of EITHER origin (the startup capacity probe or an
|
||
// embedder speed test via `NativeClient::request_probe`). The startup path had both built
|
||
// in; the embedder path had neither, so an unanswered request wedged the report tick and a
|
||
// finished one left the ABR window anchored before the burst.
|
||
let mut was_probing = false;
|
||
// Set when a probe ends: the FIRST post-probe report window is discarded outright (no
|
||
// LossReport, no standing-latency close, no ABR feed). The `last_*` rebase below cannot
|
||
// fully clean it — probe frames still pending in the reassembler age out as
|
||
// `frames_dropped` for another LOSS_WINDOW (~120 ms) AFTER the rebase, and the burst may
|
||
// have latched `flush_in_window` — and either reads as SEVERE congestion. The 2026-07
|
||
// field report's Automatic session backed off 20→14 Mb/s one second in (exactly one
|
||
// report tick after its capacity probe) and, with slow start dead from that first
|
||
// "congestion", crawled additively for the entire match.
|
||
let mut discard_abr_window = false;
|
||
let mut probe_watchdog: Option<Instant> = None;
|
||
let (mut owd_sum_ns, mut owd_frames) = (0i128, 0u32);
|
||
let mut flush_in_window = false;
|
||
// Jump-to-live state (see the guard in the loop below): when the clock-based over-bound
|
||
// run began (`stale_since`, armed only when the skew handshake succeeded so the clocks
|
||
// are comparable), when the clock-free non-draining-queue run began (`standing_since`),
|
||
// and the last-jump instant for the shared cooldown. Wall-clock runs (T1.4), not frame
|
||
// counts — the detectors' sensitivity must not scale with fps or repeat cadence.
|
||
let mut stale_since: Option<Instant> = None;
|
||
let mut standing_since: Option<Instant> = None;
|
||
let mut last_flush: Option<Instant> = None;
|
||
// Clock-detector health: consecutive clock-triggered flushes that found no local backlog
|
||
// (see NOOP_FLUSH_DATAGRAMS). Reaching NOOP_CLOCK_FLUSHES_TO_DISARM turns the clock-based
|
||
// detector off (a clock step / upstream queue it can't fix) — until a mid-stream clock
|
||
// re-sync lands and re-arms it (`pump_clock_gen` below). The FIRST no-op flush also asks
|
||
// the control task for an immediate re-sync (via the report tick): the flush finding no
|
||
// local backlog IS the "the wall clock stepped under me" signal.
|
||
let mut noop_clock_flushes: u32 = 0;
|
||
let mut clock_detector_armed = true;
|
||
let mut resync_wanted = false;
|
||
let mut seen_clock_gen = pump_clock_gen.load(Ordering::Relaxed);
|
||
let mut seen_mode_gen = pump_mode_gen.load(Ordering::Relaxed);
|
||
// Standing-latency bleed (see StandingLatency): the third detector, for the small,
|
||
// constant, loss-free OWD elevation the two jump-to-live detectors deliberately
|
||
// tolerate (< QUEUE_HIGH frames, < FLUSH_LATENCY behind) — a sub-frame standing
|
||
// backlog, or a stale clock offset after a wall-clock step, either of which otherwise
|
||
// reads as permanent extra "network" latency for the rest of the session.
|
||
let mut standing_lat = StandingLatency::new();
|
||
while !pump_shutdown.load(Ordering::SeqCst) {
|
||
// The live host↔client offset: re-loaded every iteration so an applied mid-stream
|
||
// re-sync takes effect on the very next frame's latency math.
|
||
let clock_offset_ns = pump_clock_offset.load(Ordering::Relaxed);
|
||
// An applied re-sync invalidates the staleness run measured under the OLD offset:
|
||
// reset the counters and re-arm the clock-based detector if a step had disarmed it.
|
||
let gen = pump_clock_gen.load(Ordering::Relaxed);
|
||
if gen != seen_clock_gen {
|
||
seen_clock_gen = gen;
|
||
stale_since = None;
|
||
noop_clock_flushes = 0;
|
||
// Every OWD reading shifted with the offset — the standing-latency floor and
|
||
// any elevation measured under the old one are meaningless now. If a stale
|
||
// offset WAS the elevation, this is also the moment it gets fixed.
|
||
standing_lat.rebase();
|
||
if !clock_detector_armed {
|
||
clock_detector_armed = true;
|
||
tracing::info!("clock re-sync applied — clock-based jump-to-live re-armed");
|
||
}
|
||
}
|
||
// Mirror the reassembler's unrecoverable-drop count for the client's keyframe-recovery
|
||
// loop, and (during a speed test) the packet-level receive counters for the throughput
|
||
// measurement. Updated every iteration (not just on a produced frame) so they stay current
|
||
// through a total-loss drought where no AU completes. Cheap: a few relaxed atomic loads.
|
||
let st = session.stats();
|
||
frames_dropped.store(st.frames_dropped, Ordering::Relaxed);
|
||
fec_recovered.store(st.fec_recovered_shards, Ordering::Relaxed);
|
||
let probe_active = {
|
||
let mut p = pump_probe.lock().unwrap();
|
||
if p.active && !p.done {
|
||
// Arm edge (first mirror tick): zero the arrival stamps before the burst can
|
||
// claim them — the ProbeRequest is still queued locally (the burst starts a
|
||
// round trip later), so the reset cannot race a probe packet. `st` predates
|
||
// the reset, so the stamps mirror 0 on this tick and live values after.
|
||
let arming = p.base_bytes.is_none();
|
||
if arming {
|
||
session.reset_probe_arrivals();
|
||
}
|
||
p.rx_packets_now = st.probe_packets_received;
|
||
p.rx_bytes_now = st.probe_bytes_received;
|
||
(p.first_arrival_ns, p.last_arrival_ns) = if arming {
|
||
(0, 0)
|
||
} else {
|
||
(st.probe_first_arrival_ns, st.probe_last_arrival_ns)
|
||
};
|
||
p.base_packets.get_or_insert(st.probe_packets_received);
|
||
p.base_bytes.get_or_insert(st.probe_bytes_received);
|
||
}
|
||
p.active && !p.done
|
||
};
|
||
// A probe just ended (either kind): rebase EVERY window anchor past the burst. Its
|
||
// FLAG_PROBE filler landed in `bytes_received`/`packets_received` (session.rs counts
|
||
// every accepted datagram) but never reached the decoder, and the report tick was
|
||
// suppressed for the whole burst, so `last_*` still points before it. Without this the
|
||
// first post-burst window reads the burst rate as `actual_kbps` and poisons the ABR's
|
||
// monotone proven-throughput high-water mark — which never decays — and divides the
|
||
// window's loss by a packet count inflated with filler.
|
||
if was_probing && !probe_active {
|
||
last_recovered = st.fec_recovered_shards;
|
||
last_late = st.fec_late_shards;
|
||
last_received = st.packets_received;
|
||
last_dropped = st.frames_dropped;
|
||
last_bytes = st.media_bytes_received;
|
||
last_report = Instant::now();
|
||
discard_abr_window = true;
|
||
flush_in_window = false;
|
||
}
|
||
// Arm a watchdog on the leading edge of ANY probe, so a host that silently ignores
|
||
// `ProbeRequest` (an old build — anticipated, see the capacity-probe timeout below)
|
||
// cannot latch `active` forever and suppress the report tick for the whole session.
|
||
if !was_probing && probe_active {
|
||
let burst = Duration::from_millis(pump_probe.lock().unwrap().duration_ms as u64);
|
||
probe_watchdog = Some(Instant::now() + burst + CAPACITY_PROBE_TIMEOUT);
|
||
}
|
||
if !probe_active {
|
||
probe_watchdog = None;
|
||
} else if let Some(deadline) = probe_watchdog {
|
||
if Instant::now() >= deadline {
|
||
probe_watchdog = None;
|
||
pump_probe.lock().unwrap().active = false;
|
||
tracing::warn!(
|
||
"speed-test probe unanswered — clearing it so loss reports and ABR resume"
|
||
);
|
||
}
|
||
}
|
||
was_probing = probe_active;
|
||
// Fire the startup link-capacity probe once the stream has settled (see the constants
|
||
// above), and fold its measurement into the ABR ceiling when the result lands.
|
||
// Never steal the slot from an embedder speed test in flight: there is one `ProbeState`
|
||
// and no correlation id, so a clobber both wrecks the user's "Test connection" figure
|
||
// (its base counters get re-snapshotted mid-burst against the full-burst denominator)
|
||
// and mis-scales our own ceiling. Retry once it finishes.
|
||
if capacity_probe_at.is_some_and(|at| Instant::now() >= at) && probe_active {
|
||
capacity_probe_at = Some(Instant::now() + CAPACITY_PROBE_DELAY);
|
||
} else if capacity_probe_at.is_some_and(|at| Instant::now() >= at) {
|
||
capacity_probe_at = None;
|
||
*pump_probe.lock().unwrap() = ProbeState {
|
||
active: true,
|
||
duration_ms: CAPACITY_PROBE_MS,
|
||
..Default::default()
|
||
};
|
||
if ctrl_tx
|
||
.try_send(CtrlRequest::Probe(ProbeRequest {
|
||
target_kbps: capacity_probe_kbps,
|
||
duration_ms: CAPACITY_PROBE_MS,
|
||
}))
|
||
.is_ok()
|
||
{
|
||
capacity_probe_deadline = Some(Instant::now() + CAPACITY_PROBE_TIMEOUT);
|
||
tracing::info!(
|
||
target_kbps = capacity_probe_kbps,
|
||
duration_ms = CAPACITY_PROBE_MS,
|
||
"adaptive bitrate: startup link-capacity probe"
|
||
);
|
||
} else {
|
||
pump_probe.lock().unwrap().active = false; // ctrl queue full — skip
|
||
}
|
||
}
|
||
if let Some(deadline) = capacity_probe_deadline {
|
||
let mut p = pump_probe.lock().unwrap();
|
||
if p.done {
|
||
capacity_probe_deadline = None;
|
||
// An all-zero reply is a decline (old host / probe-less build) — keep the
|
||
// negotiated ceiling. Otherwise: delivered wire kbps × 0.7, over the
|
||
// CLIENT-measured receive interval (the host's send window closes while the
|
||
// bottleneck queue is still draining toward us, so dividing by ITS duration
|
||
// overstates the link — a 1 GbE link "measured" 1266 Mbps, and the inflated
|
||
// ceiling is permanent because set_ceiling never lowers); the host duration
|
||
// is the fallback when the burst delivered too few packets for an interval.
|
||
if p.host_duration_ms > 0 && p.delivered_bytes > 0 {
|
||
let window_ms = p.throughput_window_ms();
|
||
let delivered_kbps =
|
||
(p.delivered_bytes.saturating_mul(8) / window_ms.max(1) as u64) as u32;
|
||
let ceiling = delivered_kbps.saturating_mul(7) / 10;
|
||
abr.set_ceiling(ceiling);
|
||
tracing::info!(
|
||
delivered_kbps,
|
||
ceiling_kbps = ceiling,
|
||
client_interval_ms = p.client_interval_ms,
|
||
host_duration_ms = p.host_duration_ms,
|
||
"adaptive bitrate: link-capacity probe done — climb ceiling set"
|
||
);
|
||
} else {
|
||
tracing::info!(
|
||
"adaptive bitrate: capacity probe declined — keeping negotiated ceiling"
|
||
);
|
||
}
|
||
// Rebase the ABR window's byte anchor past the burst. (Probe filler is
|
||
// routed out of `media_bytes_received` at the reassembler, so it can no
|
||
// longer read as the burst rate on its own — but the anchor still has to
|
||
// skip the video that landed around the burst under a suppressed report
|
||
// tick, which would otherwise divide a long span's bytes by one window.)
|
||
last_bytes = st.media_bytes_received;
|
||
} else if Instant::now() >= deadline {
|
||
// The host never answered (a build that ignores ProbeRequest): clear the
|
||
// stuck-active state so LossReports resume, keep the negotiated ceiling.
|
||
p.active = false;
|
||
capacity_probe_deadline = None;
|
||
tracing::info!(
|
||
"adaptive bitrate: capacity probe timed out (old host?) — keeping negotiated ceiling"
|
||
);
|
||
}
|
||
}
|
||
if !probe_active && last_report.elapsed() >= ADAPT_REPORT_INTERVAL {
|
||
// A no-op clock flush earlier in this window suspected a wall-clock step: fire
|
||
// the mid-stream re-sync now (once — the 60 s periodic covers everything else).
|
||
if resync_wanted {
|
||
resync_wanted = false;
|
||
let _ = ctrl_tx.try_send(CtrlRequest::ClockResync);
|
||
}
|
||
// All-intra drain-to-newest skips are NOT losses (the wire delivered them) —
|
||
// surface them at debug so a slow consumer is visible without alarming the
|
||
// OSD loss counters.
|
||
let skipped = frames.take_skipped();
|
||
if skipped > 0 {
|
||
tracing::debug!(skipped, "all-intra frame channel drained to newest");
|
||
}
|
||
let discard = std::mem::take(&mut discard_abr_window);
|
||
let window_dropped = st.frames_dropped.wrapping_sub(last_dropped);
|
||
let loss_ppm = window_loss_ppm(
|
||
st.fec_recovered_shards.wrapping_sub(last_recovered),
|
||
st.fec_late_shards.wrapping_sub(last_late),
|
||
st.packets_received.wrapping_sub(last_received),
|
||
window_dropped,
|
||
);
|
||
if discard {
|
||
// Probe-tail residue (see `discard_abr_window`): a LossReport from this
|
||
// window would also spike the host's adaptive FEC off deliberate overload.
|
||
tracing::debug!(
|
||
loss_ppm,
|
||
window_dropped,
|
||
"discarding the first post-probe ABR window (probe-tail residue)"
|
||
);
|
||
} else {
|
||
let _ = ctrl_tx.try_send(CtrlRequest::Loss(LossReport { loss_ppm }));
|
||
}
|
||
// Standing-latency bleed: close the detector's window with this report's loss
|
||
// verdict and run its escalation ladder — re-sync first (free; a stale offset
|
||
// from a stepped wall clock produces exactly this signature and the applied
|
||
// re-sync rebases the floor), then a bounded flush+keyframe (drains a real
|
||
// sub-threshold standing backlog the jump-to-live thresholds tolerate), then a
|
||
// loud disarm (the path latency itself changed; nothing local fixes that).
|
||
// A discard window closes the detector as NOT-loss-free: its clean-run resets
|
||
// (conservative) and no action can fire off probe residue.
|
||
match standing_lat.on_window(!discard && loss_ppm == 0 && window_dropped == 0) {
|
||
StandingLatAction::None => {}
|
||
StandingLatAction::Resync { above_ms } => {
|
||
tracing::info!(
|
||
above_ms,
|
||
"standing latency above the session floor with zero loss — \
|
||
requesting a clock re-sync first (a stale offset reads exactly \
|
||
like this)"
|
||
);
|
||
let _ = ctrl_tx.try_send(CtrlRequest::ClockResync);
|
||
}
|
||
StandingLatAction::Bleed { above_ms } => {
|
||
// Shares the jump-to-live cooldown: an unexecuted bleed simply re-arms
|
||
// over the next windows (the detector's run rebuilds).
|
||
if last_flush.is_none_or(|t| t.elapsed() >= FLUSH_COOLDOWN) {
|
||
last_flush = Some(Instant::now());
|
||
// Deliberately NOT `flush_in_window = true`: that flag is the ABR's
|
||
// SEVERE verdict (an immediate ×0.7 back-off), and the bleed fires
|
||
// only after ~6 provably loss-free windows with a sub-25ms elevation
|
||
// the controller itself scores as fine. The bleed's effect reaches
|
||
// the ABR through the window's own honest signals (OWD/loss/decode);
|
||
// the flag stays exclusive to the jump-to-live path below.
|
||
let flushed = session.flush_backlog().unwrap_or(0);
|
||
let dropped = frames.clear();
|
||
let _ = ctrl_tx.try_send(CtrlRequest::Keyframe);
|
||
standing_lat.bled();
|
||
tracing::warn!(
|
||
above_ms,
|
||
flushed_datagrams = flushed,
|
||
dropped_frames = dropped,
|
||
"standing latency survived a clock re-sync — bled the local \
|
||
backlog (flush + keyframe)"
|
||
);
|
||
}
|
||
}
|
||
StandingLatAction::Disarm { above_ms } => {
|
||
tracing::warn!(
|
||
above_ms,
|
||
"standing latency persists after a re-sync and every bleed — not \
|
||
local, not clock; the path latency changed. Leaving it be \
|
||
(reconnect re-baselines)"
|
||
);
|
||
}
|
||
}
|
||
// Adaptive bitrate: an accepted mode switch first (it invalidates the
|
||
// mode-scoped learned state), then drain any host ack (its clamp is
|
||
// authoritative), then feed the controller this window's congestion signals; a
|
||
// decision becomes a SetBitrate on the control stream.
|
||
let mg = pump_mode_gen.load(Ordering::Relaxed);
|
||
if mg != seen_mode_gen {
|
||
seen_mode_gen = mg;
|
||
abr.on_mode_switch();
|
||
}
|
||
if let Some(acked) = bitrate_ack.lock().unwrap().take() {
|
||
abr.on_ack(acked);
|
||
}
|
||
let owd_mean_us =
|
||
(owd_frames > 0).then(|| (owd_sum_ns / owd_frames as i128 / 1000) as i64);
|
||
(owd_sum_ns, owd_frames) = (0, 0);
|
||
// Drain the embedder's decode-latency window (always, so it stays bounded even when
|
||
// the controller is disabled) → the mean feeds the decode signal; `None` when the
|
||
// embedder reported nothing this window (old embedder / no decoded frames).
|
||
let decode_mean_us = {
|
||
let mut acc = pump_decode_lat.lock().unwrap();
|
||
let (sum, count) = (acc.sum_us, acc.count);
|
||
*acc = DecodeLatAcc::default();
|
||
(count > 0).then(|| (sum / count as u64) as i64)
|
||
};
|
||
// Same drain for the host-encode window (0xCF `encode_us` via the datagram
|
||
// task) — `None` on an old host that doesn't send stage timings.
|
||
let encode_mean_us = {
|
||
let mut acc = pump_encode_lat.lock().unwrap();
|
||
let (sum, count) = (acc.sum_us, acc.count);
|
||
*acc = Default::default();
|
||
(count > 0).then(|| (sum / count as u64) as i64)
|
||
};
|
||
// Decode-recovery keyframe asks this window (counted at the control task's send
|
||
// choke point). Always drained so a discard window can't leak its count into
|
||
// the next one.
|
||
let recovery_kf_reqs = pump_recovery_kf.swap(0, Ordering::Relaxed);
|
||
// The window's ACTUAL delivered throughput — what the pipeline really carried, vs
|
||
// the target it was allowed. MEDIA bytes (data-shard payload: no headers, no FEC
|
||
// parity, no probe filler, no audio), because both consumers compare it against
|
||
// the ENCODER's target: the utilization gate asks "was the target genuinely
|
||
// tested?" and the proven mark bounds every later climb. Wire bytes answered a
|
||
// different question — they rise with the redundancy the host adds in answer to
|
||
// loss, so the gate read ~25 % high precisely on the links it exists for.
|
||
let window_ms = last_report.elapsed().as_millis().max(1) as u64;
|
||
let actual_kbps = (st
|
||
.media_bytes_received
|
||
.wrapping_sub(last_bytes)
|
||
.saturating_mul(8)
|
||
/ window_ms) as u32;
|
||
// A discard window feeds the controller NOTHING — its signals are probe-tail
|
||
// residue, and one "congestion" verdict here ends slow start for good.
|
||
let verdict = if discard {
|
||
None
|
||
} else {
|
||
abr.on_window(
|
||
Instant::now(),
|
||
window_dropped,
|
||
loss_ppm,
|
||
owd_mean_us,
|
||
decode_mean_us,
|
||
encode_mean_us,
|
||
actual_kbps,
|
||
flush_in_window,
|
||
recovery_kf_reqs,
|
||
)
|
||
};
|
||
if let Some(kbps) = verdict {
|
||
// Log the window's signals alongside the decision so an on-glass session can
|
||
// tell a decode-/encode-driven re-target (the new signals — elevated with
|
||
// loss/OWD flat) from a network-driven one.
|
||
tracing::info!(
|
||
kbps,
|
||
loss_ppm,
|
||
owd_mean_us = owd_mean_us.unwrap_or(-1),
|
||
decode_mean_us = decode_mean_us.unwrap_or(-1),
|
||
encode_mean_us = encode_mean_us.unwrap_or(-1),
|
||
actual_kbps,
|
||
flushed = flush_in_window,
|
||
recovery_kf = recovery_kf_reqs,
|
||
"adaptive bitrate: requesting encoder re-target"
|
||
);
|
||
if ctrl_tx.try_send(CtrlRequest::SetBitrate(kbps)).is_err() {
|
||
// Never reached the control task — tell the controller, or three of
|
||
// these retire it for the session as "the host never acked".
|
||
abr.on_request_dropped();
|
||
tracing::warn!(
|
||
kbps,
|
||
"adaptive bitrate: control queue full — re-target dropped"
|
||
);
|
||
}
|
||
}
|
||
flush_in_window = false;
|
||
last_report = Instant::now();
|
||
last_recovered = st.fec_recovered_shards;
|
||
last_late = st.fec_late_shards;
|
||
last_received = st.packets_received;
|
||
last_dropped = st.frames_dropped;
|
||
last_bytes = st.media_bytes_received;
|
||
if pump_perf_on {
|
||
if let Some(p) = session.take_pump_perf() {
|
||
let per_pkt_ns = |ns: u64| ns.checked_div(p.packets).unwrap_or(0);
|
||
tracing::info!(
|
||
recv_ms = p.recv_ns / 1_000_000,
|
||
decrypt_ms = p.decrypt_ns / 1_000_000,
|
||
reasm_ms = p.reasm_ns / 1_000_000,
|
||
packets = p.packets,
|
||
batches = p.batches,
|
||
pkts_per_batch = p.packets.checked_div(p.batches).unwrap_or(0),
|
||
decrypt_ns_pkt = per_pkt_ns(p.decrypt_ns),
|
||
reasm_ns_pkt = per_pkt_ns(p.reasm_ns),
|
||
"pump stage split (window)"
|
||
);
|
||
}
|
||
// Inter-arrival jitter over the window's completed AUs. `late` counts gaps
|
||
// over 2× the window median — the "a frame arrived visibly off-beat" tally.
|
||
if arrivals_us.len() >= 8 {
|
||
arrivals_us.sort_unstable();
|
||
let pct = |q: usize| arrivals_us[(arrivals_us.len() - 1) * q / 100];
|
||
let (p50, p95) = (pct(50), pct(95));
|
||
let late = arrivals_us.iter().filter(|&&d| d > p50 * 2).count();
|
||
tracing::info!(
|
||
frames = arrivals_us.len() + 1,
|
||
arrival_p50_us = p50,
|
||
arrival_p95_us = p95,
|
||
arrival_max_us = arrivals_us.last().copied().unwrap_or(0),
|
||
late,
|
||
"frame inter-arrival jitter (window)"
|
||
);
|
||
}
|
||
arrivals_us.clear();
|
||
}
|
||
}
|
||
match session.poll_frame() {
|
||
Ok(frame) => {
|
||
if frame.flags & FLAG_PROBE as u32 != 0 {
|
||
continue; // speed-test filler, not video — measured via the counters above
|
||
}
|
||
// A prefix part is not an AU arrival: the inter-arrival series, the OWD
|
||
// window and the clock-based staleness detector below all measure per-AU
|
||
// signals, so only the delivery that completes an AU feeds them (parts
|
||
// would bias OWD low and constantly reset the staleness run).
|
||
let is_au = frame.complete;
|
||
if pump_perf_on && is_au {
|
||
let now = Instant::now();
|
||
if let Some(prev) = last_arrival.replace(now) {
|
||
// 4096 ≈ 17 s at 240 fps — a stuck window can't grow it unbounded.
|
||
if arrivals_us.len() < 4096 {
|
||
arrivals_us
|
||
.push((now - prev).as_micros().min(u32::MAX as u128) as u32);
|
||
}
|
||
}
|
||
}
|
||
// Jump-to-live guard. A standing receive/hand-off queue never drains by itself —
|
||
// the pump consumes strictly in order at the arrival rate, so once behind, the
|
||
// stream stays behind for good (observed live: stuck 6–7 s). Pre-decode AUs are
|
||
// reference-chained (infinite GOP), so we can NOT drop a frame mid-stream to catch
|
||
// up; the only safe recovery is to discard the whole backlog and re-anchor decode
|
||
// on a fresh keyframe. Two independent "we're behind" signals arm it, both gated by
|
||
// FLUSH_COOLDOWN, both suspended during a speed test (the probe MEASURES a saturated
|
||
// queue; flushing would corrupt its counters):
|
||
// * clock-based — completed frames sit > FLUSH_LATENCY behind the skew-corrected
|
||
// capture clock continuously for FLUSH_AFTER. Needs the skew handshake, and
|
||
// also catches kernel/reassembler backlog the hand-off queue hasn't reached yet.
|
||
// * clock-free — the pre-decode hand-off queue stopped draining: its depth stayed
|
||
// ≥ QUEUE_HIGH (never falling to QUEUE_LOW, still high at the trip) for
|
||
// STANDING_TIME. Works with no handshake / a same-clock session (where the
|
||
// clock path is disarmed), and is the direct signal that the embedder can't
|
||
// keep up. A transient Wi-Fi clump drains within ~100 ms and never trips it.
|
||
if probe_active {
|
||
// Keep both detectors disarmed across a speed test so its (deliberately)
|
||
// saturated queue doesn't leave a primed run that fires the moment it ends.
|
||
stale_since = None;
|
||
standing_since = None;
|
||
} else {
|
||
let lat_ns = if clock_offset_ns != 0 && is_au {
|
||
now_realtime_ns() + clock_offset_ns as i128 - frame.pts_ns as i128
|
||
} else {
|
||
0
|
||
};
|
||
// Feed the adaptive-bitrate controller's OWD window (mean capture→received
|
||
// delay): rising delay under zero loss is queue growth — the pre-loss
|
||
// congestion signal. Only meaningful with a clock handshake.
|
||
if clock_offset_ns != 0 && lat_ns > 0 {
|
||
owd_sum_ns += lat_ns;
|
||
owd_frames += 1;
|
||
// The standing-latency detector rides the same signal, but off the
|
||
// window MINIMUM (robust against jitter/burst spikes — a standing
|
||
// state elevates the floor itself). Same 10 s plausibility clamp as
|
||
// the hn stats use.
|
||
if lat_ns < 10_000_000_000 {
|
||
standing_lat.note_frame(lat_ns);
|
||
}
|
||
}
|
||
if clock_detector_armed
|
||
&& clock_offset_ns != 0
|
||
&& lat_ns > FLUSH_LATENCY.as_nanos() as i128
|
||
{
|
||
stale_since.get_or_insert_with(Instant::now);
|
||
} else if is_au {
|
||
stale_since = None;
|
||
}
|
||
let depth = frames.depth();
|
||
if depth >= QUEUE_HIGH {
|
||
standing_since.get_or_insert_with(Instant::now);
|
||
} else if depth <= QUEUE_LOW {
|
||
standing_since = None;
|
||
}
|
||
// The queue trip additionally requires the depth to still be high NOW, so
|
||
// a run that started ≥ high but is hovering in the hysteresis band (a
|
||
// clump mid-drain) never fires on elapsed time alone.
|
||
let clock_behind = stale_since.is_some_and(|t| t.elapsed() >= FLUSH_AFTER);
|
||
let queue_behind = depth >= QUEUE_HIGH
|
||
&& standing_since.is_some_and(|t| t.elapsed() >= STANDING_TIME);
|
||
if (clock_behind || queue_behind)
|
||
&& last_flush.is_none_or(|t| t.elapsed() >= FLUSH_COOLDOWN)
|
||
{
|
||
stale_since = None;
|
||
standing_since = None;
|
||
last_flush = Some(Instant::now());
|
||
flush_in_window = true; // strongest "link can't hold the rate" signal
|
||
let flushed = session.flush_backlog().unwrap_or(0);
|
||
let dropped = frames.clear();
|
||
let _ = ctrl_tx.try_send(CtrlRequest::Keyframe);
|
||
tracing::warn!(
|
||
behind_ms = if clock_behind { lat_ns / 1_000_000 } else { -1 },
|
||
queue_depth = depth,
|
||
flushed_datagrams = flushed,
|
||
dropped_frames = dropped,
|
||
"receive backlog stopped draining — jumped to live (flush + keyframe)"
|
||
);
|
||
// Clock-detector health check: a clock-only trigger whose flush found
|
||
// no local backlog is a false "behind" reading (a wall-clock step, or
|
||
// an upstream queue a local flush can't drain) — repeated, it would
|
||
// cost a recovery IDR every cooldown forever. Disarm after two in a
|
||
// row; the clock-free queue detector keeps covering real backlogs.
|
||
if clock_behind
|
||
&& !queue_behind
|
||
&& flushed < NOOP_FLUSH_DATAGRAMS
|
||
&& dropped == 0
|
||
{
|
||
noop_clock_flushes += 1;
|
||
if noop_clock_flushes == 1 {
|
||
// First no-op flush = a wall-clock step is the prime
|
||
// suspect: ask for an immediate re-sync (sent on the next
|
||
// report tick). Applied, it resets these counters and
|
||
// re-arms the detector before the disarm below triggers.
|
||
resync_wanted = true;
|
||
}
|
||
if noop_clock_flushes >= NOOP_CLOCK_FLUSHES_TO_DISARM {
|
||
clock_detector_armed = false;
|
||
tracing::warn!(
|
||
"clock-based jump-to-live disarmed — its flushes found no \
|
||
local backlog (clock step or upstream queueing suspected); \
|
||
the queue-depth detector stays armed"
|
||
);
|
||
}
|
||
} else {
|
||
noop_clock_flushes = 0;
|
||
}
|
||
continue; // this frame is part of the stale past — don't render it
|
||
}
|
||
}
|
||
frames.push(frame);
|
||
}
|
||
Err(PunktfunkError::NoFrame) => {
|
||
std::thread::sleep(Duration::from_micros(300));
|
||
}
|
||
Err(_) => break,
|
||
}
|
||
}
|
||
// The pump exited (shutdown / fatal session error) — wake any consumer blocked in
|
||
// `next_frame` with a Closed signal instead of a spurious timeout (the old mpsc did this
|
||
// implicitly when the sender dropped).
|
||
frames.close();
|
||
}
|
||
}
|