fix(abr): probe throughput is measured over the client's receive interval
The capacity probe divided client-side bytes by the HOST's burst duration — a window wrong on both edges (base snapshotted before the burst reached the host, frozen only when the ProbeResult landed, while the host's clock stops the moment ITS send window closes, before the switch/kernel queue finishes draining toward the client). On a 1 GbE link a 2 Gbps burst target "measured" 1266 Mbps and set an 886 Mbps climb ceiling the link could never carry — permanent for the session, because set_ceiling never lowers. The reassembler now stamps probe-scoped counters (bytes, packets, first/last arrival, monotonic ns) at its FLAG_PROBE routing, so video around the burst contaminates neither numerator nor denominator; the throughput divisor is the client's first→last arrival interval, with the host duration kept as the fallback when fewer than two probe packets arrived. The user-facing speed test shares the corrected computation (ProbeOutcome/PunktfunkProbeResult layouts unchanged; elapsed_ms docs updated to the new semantics). Two guards ride along: - PUNKTFUNK_ABR_MAX_MBPS clamps inside set_ceiling — the one funnel every learned ceiling passes through — so a user cap binds no matter what any probe concludes. - The controller latches decode_cap_kbps when two CONSECUTIVE backoffs carry decode-severe evidence (deep decode excursion or jump-to-live flush) at a similar pre-backoff rate, mirroring host_cap_kbps for the client decoder: a knee below the link ceiling was a permanent 30-60 s sawtooth costing a flush + dropped-frame burst per cycle (1440p120 HEVC field case, knee ~490 Mbps). One spurious flush never latches; the cap re-probes on the CAP_REPROBE_WINDOWS clock, so it lifts when the decoder recovers. Also rights the three stale "3 Gbps" probe-clamp comments (the host constant has been 10 Gbps since MAX_PROBE_KBPS moved). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Claude Fable 5
@@ -4116,7 +4116,11 @@ pub struct PunktfunkProbeResult {
|
||||
/// Application goodput bytes / access units the host offered.
|
||||
pub host_bytes: u64,
|
||||
pub host_packets: u32,
|
||||
/// The host's measured burst duration, milliseconds (the throughput denominator).
|
||||
/// The throughput denominator, milliseconds: the client-measured burst receive interval
|
||||
/// (first → last probe-packet arrival) once `done`; the host's measured send-window
|
||||
/// duration when fewer than two probe packets arrived (no interval to measure from). The
|
||||
/// host duration alone overstates throughput — its window closes while the bottleneck
|
||||
/// queue is still draining toward the client.
|
||||
pub elapsed_ms: u32,
|
||||
/// Delivered wire throughput = `recv_bytes * 8 / elapsed_ms` (kilobits/second).
|
||||
pub throughput_kbps: u32,
|
||||
@@ -4130,7 +4134,7 @@ pub struct PunktfunkProbeResult {
|
||||
}
|
||||
|
||||
/// Start a bandwidth speed test: ask the host to burst filler over the data plane at
|
||||
/// `target_kbps` of goodput for `duration_ms` (each clamped host-side to ≤ 3 Gbps / ≤ 5 s),
|
||||
/// `target_kbps` of goodput for `duration_ms` (each clamped host-side to ≤ 10 Gbps / ≤ 5 s),
|
||||
/// *briefly pausing video*. Non-blocking — poll [`punktfunk_connection_probe_result`] until its
|
||||
/// `done` field is 1. Starting a probe resets any prior measurement.
|
||||
///
|
||||
|
||||
@@ -31,6 +31,10 @@
|
||||
//! after ~4.5 s clean, ceilinged). Changes are rate-limited (each one costs the IDR the host's
|
||||
//! rebuilt encoder opens with) and the whole controller disables itself against a host that never
|
||||
//! answers [`crate::quic::BitrateChanged`] (an older build that ignores unknown control messages).
|
||||
//! Standing limits are LEARNED rather than re-poked: two identical short host acks latch the
|
||||
//! encoder's ceiling (`host_cap_kbps`), two consecutive decode-severe backoffs at a similar rate
|
||||
//! latch the client decoder's knee (`decode_cap_kbps`) — and both re-probe slowly
|
||||
//! ([`CAP_REPROBE_WINDOWS`]) so neither latch outlives the condition that taught it.
|
||||
//!
|
||||
//! Climbs are additionally **evidence-gated**. The target is only a *promise* to the encoder —
|
||||
//! how many bits it actually emits depends on the content — so on calm content (a menu, an idle
|
||||
@@ -129,7 +133,16 @@ const ENCODE_SEVERE_US: i64 = 12_000;
|
||||
/// evidence, not a spec limit — without a re-probe, one heavy scene would cap the whole
|
||||
/// session. A still-standing limit just re-teaches itself in two short acks, which the host
|
||||
/// pre-clamps without touching the encoder — the re-probe costs no rebuild, no IDR.
|
||||
/// The [`decode cap`](BitrateController::decode_cap_kbps) re-probes on the same clock for the
|
||||
/// same reason: the decoder's knee moves with content and thermals, so its latch must not be
|
||||
/// permanent either.
|
||||
const CAP_REPROBE_WINDOWS: u32 = 80;
|
||||
/// Two consecutive decode-driven backoffs latch the
|
||||
/// [`decode cap`](BitrateController::decode_cap_kbps) only when their pre-backoff rates agree
|
||||
/// within ±1/8: the decoder's knee is a RATE, so repeated chokes at the same rate are its
|
||||
/// signature — two unrelated events (a Wi-Fi flush at 300 Mbps, a decode spike at 500) share
|
||||
/// no knee and must not teach one.
|
||||
const DECODE_CAP_SIMILAR_DIV: u32 = 8;
|
||||
/// Rolling window (in 750 ms report windows, ~30 s) whose minimum mean is the OWD baseline.
|
||||
/// Long enough to remember the uncongested floor, short enough to follow genuine path changes.
|
||||
const BASELINE_WINDOWS: usize = 40;
|
||||
@@ -137,6 +150,23 @@ const BASELINE_WINDOWS: usize = 40;
|
||||
/// predates bitrate renegotiation and going quiet for the rest of the session.
|
||||
const MAX_UNACKED: u32 = 3;
|
||||
|
||||
/// Operator escape hatch: `PUNKTFUNK_ABR_MAX_MBPS` (megabits/second, the
|
||||
/// `PUNKTFUNK_PYROWAVE_MAX_MBPS` convention) caps the climb ceiling however it is learned.
|
||||
/// The startup link-capacity probe MEASURES the ceiling, and
|
||||
/// [`set_ceiling`](BitrateController::set_ceiling)'s deliberate monotonicity makes an inflated
|
||||
/// measurement permanent for the session — a link that mis-measures (a bursty middlebox, a
|
||||
/// queue-flattered interval) needs a knob that binds regardless of what any probe claims.
|
||||
/// `PUNKTFUNK_ABR_PROBE_KBPS` is NOT that knob: it only shrinks the burst target, not what the
|
||||
/// measurement may conclude. Unset/0/garbage → no cap. Read once per controller, at
|
||||
/// construction.
|
||||
fn ceiling_cap_from_env() -> Option<u32> {
|
||||
std::env::var("PUNKTFUNK_ABR_MAX_MBPS")
|
||||
.ok()
|
||||
.and_then(|v| v.trim().parse::<u32>().ok())
|
||||
.filter(|&m| m > 0)
|
||||
.map(|m| m.saturating_mul(1_000))
|
||||
}
|
||||
|
||||
/// One decision per report window; `Some(kbps)` = send a [`crate::quic::SetBitrate`].
|
||||
pub(crate) struct BitrateController {
|
||||
/// `false` = permanently off (explicit user bitrate, an old host, or ack silence).
|
||||
@@ -147,6 +177,10 @@ pub(crate) struct BitrateController {
|
||||
/// raises it via [`set_ceiling`](Self::set_ceiling) — that measurement is what lets an
|
||||
/// Automatic session scale past its conservative start.
|
||||
ceiling_kbps: u32,
|
||||
/// The `PUNKTFUNK_ABR_MAX_MBPS` cap in kbps (see [`ceiling_cap_from_env`]), injected at
|
||||
/// construction so tests exercise the clamp without touching the process environment.
|
||||
/// `None` = no cap.
|
||||
ceiling_cap_kbps: Option<u32>,
|
||||
floor_kbps: u32,
|
||||
/// Slow start: true until the first congestion signal — clean windows DOUBLE the rate
|
||||
/// (cooldown-paced) instead of the +6 % additive step.
|
||||
@@ -178,6 +212,24 @@ pub(crate) struct BitrateController {
|
||||
short_acks: u32,
|
||||
/// Clean windows spent parked at the learned cap (the re-probe clock).
|
||||
cap_probe_windows: u32,
|
||||
/// The client-decoder rate cap, mirroring [`host_cap_kbps`](Self::host_cap_kbps) for the
|
||||
/// OTHER end of the pipe: latched when two CONSECUTIVE backoffs carried decode-severe
|
||||
/// evidence (a deep decode-latency excursion, or a jump-to-live flush — in the
|
||||
/// decoder-saturation regime the flushed backlog formed BEHIND a decoder that stopped
|
||||
/// keeping up) at a similar pre-backoff rate. Without it a decoder knee below the link
|
||||
/// ceiling is a permanent 30–60 s sawtooth: every ×0.7 backoff re-climbs toward a ceiling
|
||||
/// the decoder can't hold, and each cycle costs a flush plus a dropped-frame burst (the
|
||||
/// 1440p120 HEVC field case: knee ~490 Mbps under a ~658 Mbps ceiling). Slowly re-probed
|
||||
/// on the [`CAP_REPROBE_WINDOWS`] clock, exactly like the host cap, so a decoder that
|
||||
/// recovers (lighter content, thermal headroom) climbs again — the latch is never
|
||||
/// permanent.
|
||||
decode_cap_kbps: Option<u32>,
|
||||
/// The previous decode-driven backoff's pre-backoff rate (0 = the last backoff wasn't
|
||||
/// decode-driven): the reference the next one must land near ([`DECODE_CAP_SIMILAR_DIV`])
|
||||
/// to latch the cap — one spurious flush teaches nothing.
|
||||
decode_backoff_kbps: u32,
|
||||
/// Clean windows spent parked at the learned decode cap (its re-probe clock).
|
||||
decode_cap_probe_windows: u32,
|
||||
/// Proven throughput: the session's highest windowed ACTUAL delivered rate seen with flat
|
||||
/// decode latency — the known-good high-water mark climbs are bounded against. Never decays;
|
||||
/// shrinking capacity (thermals, a heavier scene) is the reactive decode signal's job. On
|
||||
@@ -196,10 +248,17 @@ impl BitrateController {
|
||||
/// to build a permanently-disabled controller (explicit bitrate / an old host that didn't
|
||||
/// echo one — no known ceiling to work against).
|
||||
pub(crate) fn new(start_kbps: u32) -> Self {
|
||||
Self::with_ceiling_cap(start_kbps, ceiling_cap_from_env())
|
||||
}
|
||||
|
||||
/// [`new`](Self::new) with the `PUNKTFUNK_ABR_MAX_MBPS` cap injected — the seam the unit
|
||||
/// tests use so the clamp's behavior never depends on the test process's environment.
|
||||
fn with_ceiling_cap(start_kbps: u32, ceiling_cap_kbps: Option<u32>) -> Self {
|
||||
BitrateController {
|
||||
enabled: start_kbps > 0,
|
||||
current_kbps: start_kbps,
|
||||
ceiling_kbps: start_kbps,
|
||||
ceiling_cap_kbps,
|
||||
floor_kbps: FLOOR_KBPS.min(start_kbps.max(1)),
|
||||
probing: true,
|
||||
owd_means: VecDeque::with_capacity(BASELINE_WINDOWS),
|
||||
@@ -210,6 +269,9 @@ impl BitrateController {
|
||||
short_ack_kbps: 0,
|
||||
short_acks: 0,
|
||||
cap_probe_windows: 0,
|
||||
decode_cap_kbps: None,
|
||||
decode_backoff_kbps: 0,
|
||||
decode_cap_probe_windows: 0,
|
||||
proven_kbps: 0,
|
||||
bad_windows: 0,
|
||||
clean_windows: 0,
|
||||
@@ -222,8 +284,12 @@ impl BitrateController {
|
||||
/// delivered throughput with headroom already subtracted by the caller). Without this call
|
||||
/// the ceiling stays the negotiated start rate — exactly the old behavior. Never lowers:
|
||||
/// a congested-moment measurement must not shrink authority below what was negotiated
|
||||
/// (descent is the congestion signals' job).
|
||||
/// (descent is the congestion signals' job). The `PUNKTFUNK_ABR_MAX_MBPS` cap clamps HERE
|
||||
/// — the one funnel every learned ceiling passes through — so it binds no matter how the
|
||||
/// ceiling was learned; monotonicity is precisely why the user needs it (one inflated
|
||||
/// measurement is otherwise permanent for the session).
|
||||
pub(crate) fn set_ceiling(&mut self, kbps: u32) {
|
||||
let kbps = kbps.min(self.ceiling_cap_kbps.unwrap_or(u32::MAX));
|
||||
if self.enabled && kbps > self.ceiling_kbps {
|
||||
self.ceiling_kbps = kbps;
|
||||
}
|
||||
@@ -274,11 +340,16 @@ impl BitrateController {
|
||||
|
||||
/// An accepted mode switch: the encoder's ceiling and compute knee are properties of the
|
||||
/// MODE (4K120 caps where 1080p60 never would) — drop the mode-scoped learned state. The
|
||||
/// probe-measured `ceiling_kbps` (a LINK property) survives.
|
||||
/// decoder's knee is just as mode-scoped (pixel rate drives both ends of the codec), so
|
||||
/// the decode cap goes with it. The probe-measured `ceiling_kbps` (a LINK property)
|
||||
/// survives.
|
||||
pub(crate) fn on_mode_switch(&mut self) {
|
||||
self.host_cap_kbps = None;
|
||||
self.short_acks = 0;
|
||||
self.cap_probe_windows = 0;
|
||||
self.decode_cap_kbps = None;
|
||||
self.decode_backoff_kbps = 0;
|
||||
self.decode_cap_probe_windows = 0;
|
||||
self.encode_means.clear();
|
||||
}
|
||||
|
||||
@@ -427,6 +498,30 @@ impl BitrateController {
|
||||
}
|
||||
}
|
||||
}
|
||||
// The decode cap re-probes on the same clock and for the same reason: the knee is
|
||||
// content- and thermals-dependent evidence, not a spec limit — a decoder that recovers
|
||||
// must get its headroom back, so the latch clears UPWARD through here rather than ever
|
||||
// being permanent. A still-standing knee re-latches from the next pair of
|
||||
// decode-driven backoffs.
|
||||
if let Some(cap) = self.decode_cap_kbps {
|
||||
if bad {
|
||||
self.decode_cap_probe_windows = 0;
|
||||
} else if self.current_kbps >= cap.saturating_sub(cap / 16) {
|
||||
self.decode_cap_probe_windows += 1;
|
||||
if self.decode_cap_probe_windows >= CAP_REPROBE_WINDOWS {
|
||||
self.decode_cap_probe_windows = 0;
|
||||
let lifted = cap.saturating_add(cap / 8).min(self.ceiling_kbps);
|
||||
if lifted > cap {
|
||||
tracing::debug!(
|
||||
from_kbps = cap,
|
||||
to_kbps = lifted,
|
||||
"adaptive bitrate: re-probing above the learned decode cap"
|
||||
);
|
||||
self.decode_cap_kbps = Some(lifted);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
let cooled = self
|
||||
.last_change
|
||||
.is_none_or(|t| now.duration_since(t) >= CHANGE_COOLDOWN);
|
||||
@@ -436,6 +531,31 @@ impl BitrateController {
|
||||
if (self.bad_windows >= BAD_WINDOWS_TO_DECREASE || (severe && self.bad_windows >= 1))
|
||||
&& self.current_kbps > self.floor_kbps
|
||||
{
|
||||
// Decode-cap learning (see [`decode_cap_kbps`](Self::decode_cap_kbps)): a backoff
|
||||
// with decode-severe evidence — the deep decode excursion, or the flush that
|
||||
// drained the queue behind a stalled decoder — remembers its pre-backoff rate; the
|
||||
// SECOND consecutive one at a similar rate latches that rate as the decoder's
|
||||
// knee. One event never latches (a spurious flush must stay a one-off), and a
|
||||
// backoff without decode evidence in between breaks the streak — whatever it saw,
|
||||
// it wasn't the same knee.
|
||||
if decode_severe || flushed {
|
||||
let rate = self.current_kbps;
|
||||
let similar = self.decode_backoff_kbps > 0
|
||||
&& rate.abs_diff(self.decode_backoff_kbps)
|
||||
<= self.decode_backoff_kbps / DECODE_CAP_SIMILAR_DIV;
|
||||
if similar && self.decode_cap_kbps.is_none_or(|c| rate < c) {
|
||||
tracing::info!(
|
||||
cap_kbps = rate,
|
||||
"adaptive bitrate: decode cap learned (decoder knee) — climbs stop \
|
||||
here until it lifts"
|
||||
);
|
||||
self.decode_cap_kbps = Some(rate.max(self.floor_kbps));
|
||||
self.decode_cap_probe_windows = 0;
|
||||
}
|
||||
self.decode_backoff_kbps = rate;
|
||||
} else {
|
||||
self.decode_backoff_kbps = 0;
|
||||
}
|
||||
let next = ((self.current_kbps as u64 * 7 / 10) as u32).max(self.floor_kbps);
|
||||
self.bad_windows = 0;
|
||||
return self.request(next, now);
|
||||
@@ -447,11 +567,13 @@ impl BitrateController {
|
||||
// utilized window after a long-enough clean run climbs immediately.
|
||||
let utilized =
|
||||
actual_kbps as u64 * UTILIZATION_DEN >= self.current_kbps as u64 * UTILIZATION_NUM;
|
||||
// The effective ceiling folds in the host-taught cap: the probe measured the LINK, but
|
||||
// the host's short acks measured the ENCODER — whichever binds first is the limit.
|
||||
// The effective ceiling folds in both learned caps: the probe measured the LINK, the
|
||||
// host's short acks measured the ENCODER, and the decode cap measured the CLIENT
|
||||
// DECODER — whichever binds first is the limit.
|
||||
let eff_ceiling = self
|
||||
.ceiling_kbps
|
||||
.min(self.host_cap_kbps.unwrap_or(u32::MAX));
|
||||
.min(self.host_cap_kbps.unwrap_or(u32::MAX))
|
||||
.min(self.decode_cap_kbps.unwrap_or(u32::MAX));
|
||||
let cap = eff_ceiling
|
||||
.min(self.proven_kbps.saturating_mul(PROVEN_HEADROOM_NUM) / PROVEN_HEADROOM_DEN);
|
||||
if self.current_kbps < eff_ceiling && utilized && cap > self.current_kbps {
|
||||
@@ -1447,6 +1569,243 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn env_max_mbps_caps_every_learned_ceiling() {
|
||||
// PUNKTFUNK_ABR_MAX_MBPS=50 (injected — `new` reads the env exactly once, at
|
||||
// construction): a probe "measuring" 886 Mbps (the divisor bug's field figure) must
|
||||
// not out-rank the user's cap…
|
||||
let mut c = BitrateController::with_ceiling_cap(20_000, Some(50_000));
|
||||
c.set_ceiling(886_312);
|
||||
assert_eq!(c.ceiling_kbps, 50_000);
|
||||
// …while a measurement under the cap stands untouched.
|
||||
let mut c = BitrateController::with_ceiling_cap(20_000, Some(50_000));
|
||||
c.set_ceiling(40_000);
|
||||
assert_eq!(c.ceiling_kbps, 40_000);
|
||||
// And the climb honors it: slow start doubles 20→40, the capped ceiling truncates the
|
||||
// next step to 50, then quiet — never a request past the user's limit.
|
||||
let mut c = BitrateController::with_ceiling_cap(20_000, Some(50_000));
|
||||
c.set_ceiling(886_312);
|
||||
let start = Instant::now();
|
||||
assert_eq!(run_clean(&mut c, start, 0, 1), Some(40_000));
|
||||
c.on_ack(40_000);
|
||||
assert_eq!(run_clean(&mut c, start, 2, 1), Some(50_000));
|
||||
c.on_ack(50_000);
|
||||
assert_eq!(run_clean(&mut c, start, 4, 20), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decode_cap_latches_after_two_consecutive_decode_severe_backoffs() {
|
||||
// The 1440p120 field sawtooth: a decoder knee (~500 Mbps) well under the (inflated)
|
||||
// link ceiling — nothing ever LEARNED the knee, so every re-climb ended in a flush +
|
||||
// dropped-frame burst. Establish a decode baseline on calm windows, choke twice at the
|
||||
// same rate, and the second decode-severe backoff must latch the knee.
|
||||
let mut c = BitrateController::new(500_000);
|
||||
c.set_ceiling(900_000);
|
||||
let start = Instant::now();
|
||||
// Calm baseline windows (2 Mb/s actual: unutilized, so no climb interferes).
|
||||
for i in 0..4 {
|
||||
assert_eq!(
|
||||
c.on_window(
|
||||
ticks(start, i),
|
||||
0,
|
||||
0,
|
||||
Some(10_000),
|
||||
Some(8_000),
|
||||
None,
|
||||
2_000,
|
||||
false,
|
||||
0
|
||||
),
|
||||
None
|
||||
);
|
||||
}
|
||||
// First deep decode excursion → immediate ×0.7, but ONE event must not latch.
|
||||
assert_eq!(
|
||||
c.on_window(
|
||||
ticks(start, 4),
|
||||
0,
|
||||
0,
|
||||
Some(10_000),
|
||||
Some(60_000),
|
||||
None,
|
||||
490_000,
|
||||
false,
|
||||
0
|
||||
),
|
||||
Some(350_000)
|
||||
);
|
||||
assert!(c.decode_cap_kbps.is_none());
|
||||
// Second consecutive decode-severe backoff at the same pre-backoff rate: latch.
|
||||
assert_eq!(
|
||||
c.on_window(
|
||||
ticks(start, 6),
|
||||
0,
|
||||
0,
|
||||
Some(10_000),
|
||||
Some(60_000),
|
||||
None,
|
||||
490_000,
|
||||
false,
|
||||
0
|
||||
),
|
||||
Some(350_000)
|
||||
);
|
||||
assert_eq!(c.decode_cap_kbps, Some(500_000));
|
||||
// The backoff applies; from here every climb must stop AT the knee — not the 900 Mbps
|
||||
// link ceiling the old sawtooth kept re-poking.
|
||||
c.on_ack(350_000);
|
||||
let mut max_req = 0;
|
||||
for i in 8..70 {
|
||||
if let Some(k) = c.on_window(
|
||||
ticks(start, i),
|
||||
0,
|
||||
0,
|
||||
Some(10_000),
|
||||
Some(8_000),
|
||||
None,
|
||||
1_000_000,
|
||||
false,
|
||||
0,
|
||||
) {
|
||||
assert!(k <= 500_000, "climb past the decode cap: {k}");
|
||||
max_req = max_req.max(k);
|
||||
c.on_ack(k);
|
||||
}
|
||||
}
|
||||
assert_eq!(max_req, 500_000);
|
||||
assert_eq!(c.current_kbps, 500_000);
|
||||
assert_eq!(c.decode_cap_kbps, Some(500_000));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_single_flush_or_dissimilar_backoffs_never_latch_a_decode_cap() {
|
||||
// The latch's false-positive guards. A lone jump-to-live flush (a Wi-Fi clump can
|
||||
// flush once at ANY rate) backs off but teaches nothing…
|
||||
let mut c = BitrateController::new(500_000);
|
||||
c.set_ceiling(900_000);
|
||||
let start = Instant::now();
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 0), 0, 0, None, None, None, 490_000, true, 0),
|
||||
Some(350_000)
|
||||
);
|
||||
assert!(c.decode_cap_kbps.is_none());
|
||||
c.on_ack(350_000);
|
||||
// …a LOSS-driven backoff in between breaks the streak…
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 2), 1, 0, None, None, None, 340_000, false, 0),
|
||||
Some(245_000)
|
||||
);
|
||||
assert!(c.decode_cap_kbps.is_none());
|
||||
c.on_ack(245_000);
|
||||
// …so the next flush counts as a FIRST decode event again — still no latch…
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 4), 0, 0, None, None, None, 240_000, true, 0),
|
||||
Some(171_500)
|
||||
);
|
||||
assert!(c.decode_cap_kbps.is_none());
|
||||
c.on_ack(171_500);
|
||||
// …and two consecutive decode events at DISSIMILAR rates (245 vs 171.5 Mbps — no
|
||||
// common knee) must not latch either.
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 6), 0, 0, None, None, None, 170_000, true, 0),
|
||||
Some(120_050)
|
||||
);
|
||||
assert!(c.decode_cap_kbps.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decode_cap_reprobes_after_a_sustained_clean_run() {
|
||||
// The knee is content/thermals evidence, not a spec limit: after ~60 s parked clean at
|
||||
// the latched cap, it lifts one step (+12.5 %, ceiling-bounded) — the re-probe path is
|
||||
// how the latch clears (never permanent), and a still-standing knee just re-latches
|
||||
// from the next pair of decode-driven backoffs.
|
||||
let mut c = BitrateController::new(500_000);
|
||||
c.set_ceiling(900_000);
|
||||
let start = Instant::now();
|
||||
for i in 0..4 {
|
||||
let _ = c.on_window(
|
||||
ticks(start, i),
|
||||
0,
|
||||
0,
|
||||
Some(10_000),
|
||||
Some(8_000),
|
||||
None,
|
||||
2_000,
|
||||
false,
|
||||
0,
|
||||
);
|
||||
}
|
||||
for i in [4, 6] {
|
||||
let _ = c.on_window(
|
||||
ticks(start, i),
|
||||
0,
|
||||
0,
|
||||
Some(10_000),
|
||||
Some(60_000),
|
||||
None,
|
||||
490_000,
|
||||
false,
|
||||
0,
|
||||
);
|
||||
}
|
||||
assert_eq!(c.decode_cap_kbps, Some(500_000));
|
||||
// The host's ack parks the session at the knee (its clamp is authoritative).
|
||||
c.on_ack(500_000);
|
||||
for i in 0..CAP_REPROBE_WINDOWS {
|
||||
let _ = c.on_window(
|
||||
ticks(start, 8 + i),
|
||||
0,
|
||||
0,
|
||||
Some(10_000),
|
||||
Some(8_000),
|
||||
None,
|
||||
490_000,
|
||||
false,
|
||||
0,
|
||||
);
|
||||
}
|
||||
assert_eq!(c.decode_cap_kbps, Some(500_000 + 500_000 / 8));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mode_switch_clears_the_decode_cap() {
|
||||
// A 1440p120 knee means nothing at the new mode's pixel rate — the decode cap must
|
||||
// not survive the switch (the probe-measured link ceiling does).
|
||||
let mut c = BitrateController::new(500_000);
|
||||
c.set_ceiling(900_000);
|
||||
let start = Instant::now();
|
||||
for i in 0..4 {
|
||||
let _ = c.on_window(
|
||||
ticks(start, i),
|
||||
0,
|
||||
0,
|
||||
Some(10_000),
|
||||
Some(8_000),
|
||||
None,
|
||||
2_000,
|
||||
false,
|
||||
0,
|
||||
);
|
||||
}
|
||||
for i in [4, 6] {
|
||||
let _ = c.on_window(
|
||||
ticks(start, i),
|
||||
0,
|
||||
0,
|
||||
Some(10_000),
|
||||
Some(60_000),
|
||||
None,
|
||||
490_000,
|
||||
false,
|
||||
0,
|
||||
);
|
||||
}
|
||||
assert_eq!(c.decode_cap_kbps, Some(500_000));
|
||||
c.on_mode_switch();
|
||||
assert!(c.decode_cap_kbps.is_none());
|
||||
assert_eq!(c.ceiling_kbps, 900_000);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ack_silence_disables_the_controller() {
|
||||
let mut c = BitrateController::new(20_000);
|
||||
|
||||
@@ -883,7 +883,7 @@ impl NativeClient {
|
||||
/// `target_kbps` of goodput for `duration_ms`, *briefly pausing video*. Non-blocking — the
|
||||
/// measurement accumulates in the background; poll [`NativeClient::probe_result`] until its
|
||||
/// `done` flag is set. Starting a probe resets any prior measurement. The host clamps both
|
||||
/// fields (≤ 3 Gbps, ≤ 5 s).
|
||||
/// fields (≤ 10 Gbps, ≤ 5 s).
|
||||
pub fn request_probe(&self, target_kbps: u32, duration_ms: u32) -> Result<()> {
|
||||
// Reset the accumulator so a fresh run doesn't blend into the previous one.
|
||||
*self.probe.lock().unwrap() = ProbeState {
|
||||
@@ -922,8 +922,12 @@ impl NativeClient {
|
||||
p.rx_bytes_now.saturating_sub(base_b),
|
||||
)
|
||||
};
|
||||
// The host's burst duration is the throughput denominator. bytes × 8 / ms = kilobits/second.
|
||||
let window_ms = p.host_duration_ms;
|
||||
// The throughput denominator: the client-measured receive interval once the report
|
||||
// froze one, the host's send-window duration as the fallback (see
|
||||
// `ProbeState::measured_interval_ms` for why the host window alone overstates the
|
||||
// link). Both are 0 until the report lands, so a partial read reports 0 throughput —
|
||||
// unchanged. bytes × 8 / ms = kilobits/second.
|
||||
let window_ms = p.throughput_window_ms();
|
||||
let throughput_kbps = if window_ms > 0 {
|
||||
(delivered_bytes.saturating_mul(8) / window_ms as u64) as u32
|
||||
} else {
|
||||
|
||||
@@ -1,34 +1,50 @@
|
||||
//! Speed-test probe state (`ProbeState`, pump-mirrored) and the public `ProbeOutcome`.
|
||||
|
||||
/// Accumulated state of an in-flight / finished speed test. The data-plane pump mirrors the
|
||||
/// session's packet-level receive counters here; the control task finalizes the delivered figure
|
||||
/// session's probe-scoped receive counters here; the control task finalizes the delivered figure
|
||||
/// and folds in the host's [`ProbeResult`] when it lands. Read by [`NativeClient::probe_result`].
|
||||
///
|
||||
/// Counting at the *packet* level (every delivered wire packet) — not whole reassembled probe AUs —
|
||||
/// is what makes the measurement degrade gracefully: once loss exceeds the FEC budget no AU
|
||||
/// completes, so the old AU-based count cliffed to zero even though most bytes still arrived.
|
||||
/// Counting *probe* packets only (the reassembler stamps dedicated counters at its FLAG_PROBE
|
||||
/// routing) keeps video out of the numerator: the burst pauses video, but frames already in
|
||||
/// flight land during its head, and resumed video lands between the last probe packet and the
|
||||
/// host's report — both used to inflate the all-datagram byte delta this mirrored before.
|
||||
#[derive(Default)]
|
||||
pub(crate) struct ProbeState {
|
||||
/// A probe is in progress: set by `request_probe`, cleared when the host's [`ProbeResult`]
|
||||
/// lands (a re-probe just overwrites the whole state — the latest one wins).
|
||||
pub(crate) active: bool,
|
||||
/// `session.stats()` receive counters at the burst's start (snapshotted by the pump on its first
|
||||
/// tick while active) and latest, mirrored every pump iteration.
|
||||
/// Probe-scoped receive counters (`Stats::probe_*`) at the burst's start (snapshotted by the
|
||||
/// pump on its first tick while active) and latest, mirrored every pump iteration.
|
||||
pub(crate) base_packets: Option<u64>,
|
||||
pub(crate) base_bytes: Option<u64>,
|
||||
pub(crate) rx_packets_now: u64,
|
||||
pub(crate) rx_bytes_now: u64,
|
||||
/// First / last probe-packet arrival stamps (monotonic ns, 0 = none yet), mirrored from the
|
||||
/// probe-scoped session counters. Their difference is the interval the delivered bytes
|
||||
/// actually arrived in — the honest throughput denominator (see
|
||||
/// [`measured_interval_ms`](Self::measured_interval_ms)).
|
||||
pub(crate) first_arrival_ns: u64,
|
||||
pub(crate) last_arrival_ns: u64,
|
||||
/// Delivered wire packets / plaintext bytes (header + shard), frozen when the host's report lands
|
||||
/// (so resumed video after the burst can't inflate them).
|
||||
pub(crate) delivered_packets: u64,
|
||||
pub(crate) delivered_bytes: u64,
|
||||
/// The client-measured receive interval (ms), frozen alongside the delivered figures; 0 = no
|
||||
/// usable interval (the burst delivered fewer than two probe packets) — consumers fall back
|
||||
/// to [`host_duration_ms`](Self::host_duration_ms) via
|
||||
/// [`throughput_window_ms`](Self::throughput_window_ms).
|
||||
pub(crate) client_interval_ms: u32,
|
||||
/// The host's end-of-burst report.
|
||||
pub(crate) host_goodput_bytes: u64,
|
||||
pub(crate) host_au: u32,
|
||||
/// Wire packets the host actually put on the link, and the ones its send buffer dropped.
|
||||
pub(crate) host_wire_packets: u32,
|
||||
pub(crate) host_send_dropped: u32,
|
||||
/// The host's measured burst duration (the throughput denominator).
|
||||
/// The host's measured burst duration (the throughput denominator's FALLBACK — see
|
||||
/// [`throughput_window_ms`](Self::throughput_window_ms)).
|
||||
pub(crate) host_duration_ms: u32,
|
||||
/// The host's `ProbeResult` arrived → the measurement is final.
|
||||
pub(crate) done: bool,
|
||||
@@ -39,6 +55,40 @@ pub(crate) struct ProbeState {
|
||||
pub(crate) duration_ms: u32,
|
||||
}
|
||||
|
||||
impl ProbeState {
|
||||
/// The client-measured receive interval of a finished burst, in ms: first → last
|
||||
/// probe-packet arrival, floored at 1 (a sub-ms burst divided by 0 ms would read as
|
||||
/// infinite throughput). `None` — the caller falls back to the host's duration — when
|
||||
/// fewer than two probe packets arrived or the stamps are degenerate (unset / identical /
|
||||
/// reversed): a single arrival spans no interval.
|
||||
///
|
||||
/// Why not the host's `duration_ms`: it measures the SEND window, which closes while the
|
||||
/// bottleneck (switch/kernel) queue is still draining toward the client — the tail of the
|
||||
/// bytes lands *after* it. Dividing client-side bytes by the host-side window therefore
|
||||
/// overstates the link: a 1 GbE link under a 2 Gbps burst target "measured" 1266 Mbps and
|
||||
/// handed the ABR an 886 Mbps ceiling it could never deliver — and
|
||||
/// [`set_ceiling`](crate::abr::BitrateController::set_ceiling) never lowers, so the lie
|
||||
/// was permanent for the session.
|
||||
pub(crate) fn measured_interval_ms(first_ns: u64, last_ns: u64, packets: u64) -> Option<u32> {
|
||||
if packets < 2 || first_ns == 0 || last_ns <= first_ns {
|
||||
return None;
|
||||
}
|
||||
let ms = ((last_ns - first_ns) / 1_000_000).max(1);
|
||||
Some(u32::try_from(ms).unwrap_or(u32::MAX))
|
||||
}
|
||||
|
||||
/// The throughput denominator, in ms: the client-measured receive interval when the burst
|
||||
/// produced one, else the host's send-window duration (an old measurement is better than
|
||||
/// none — and strictly conservative territory only when packets were too few to matter).
|
||||
pub(crate) fn throughput_window_ms(&self) -> u32 {
|
||||
if self.client_interval_ms > 0 {
|
||||
self.client_interval_ms
|
||||
} else {
|
||||
self.host_duration_ms
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A finished/partial speed-test measurement, returned by [`NativeClient::probe_result`].
|
||||
#[derive(Clone, Copy, Debug, Default)]
|
||||
pub struct ProbeOutcome {
|
||||
@@ -50,7 +100,11 @@ pub struct ProbeOutcome {
|
||||
/// Application goodput bytes / access units the host offered.
|
||||
pub host_bytes: u64,
|
||||
pub host_packets: u32,
|
||||
/// The burst duration the host measured, in milliseconds (the throughput denominator).
|
||||
/// The throughput denominator, in milliseconds: the client-measured receive interval
|
||||
/// (first → last probe-packet arrival) once `done`; the host's measured send-window
|
||||
/// duration when the burst delivered fewer than two probe packets (no interval to measure
|
||||
/// from). The host duration alone overstates throughput — its window closes while the
|
||||
/// bottleneck queue is still draining toward the client.
|
||||
pub elapsed_ms: u32,
|
||||
/// Delivered wire throughput = `recv_bytes * 8 / elapsed_ms` (kilobits/second). The figure to
|
||||
/// drive a [`Hello::bitrate_kbps`] choice from (allow headroom for the FEC overhead + loss).
|
||||
@@ -66,3 +120,63 @@ pub struct ProbeOutcome {
|
||||
pub wire_packets_sent: u32,
|
||||
pub send_dropped: u32,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn interval_needs_two_packets_and_a_nonzero_span() {
|
||||
// <2 packets: no interval exists — the caller must fall back to the host duration.
|
||||
assert_eq!(ProbeState::measured_interval_ms(0, 0, 0), None);
|
||||
assert_eq!(
|
||||
ProbeState::measured_interval_ms(5_000_000, 5_000_000, 1),
|
||||
None
|
||||
);
|
||||
// Two packets in the same ns / a reversed pair / an unset first stamp: same fallback.
|
||||
assert_eq!(
|
||||
ProbeState::measured_interval_ms(5_000_000, 5_000_000, 2),
|
||||
None
|
||||
);
|
||||
assert_eq!(
|
||||
ProbeState::measured_interval_ms(9_000_000, 5_000_000, 2),
|
||||
None
|
||||
);
|
||||
assert_eq!(ProbeState::measured_interval_ms(0, 5_000_000, 2), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn interval_is_floored_at_one_ms() {
|
||||
// Two packets 0.4 ms apart truncate to 0 ms — the floor keeps the division honest
|
||||
// instead of infinite.
|
||||
assert_eq!(ProbeState::measured_interval_ms(1_000, 401_000, 2), Some(1));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn interval_measures_first_to_last_arrival() {
|
||||
assert_eq!(
|
||||
ProbeState::measured_interval_ms(1_000_000, 801_000_000, 1_000),
|
||||
Some(800)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn throughput_window_falls_back_to_the_host_duration() {
|
||||
// No client interval frozen (a <2-packet burst) → the host's send window is the
|
||||
// denominator, exactly the old behavior.
|
||||
let p = ProbeState {
|
||||
host_duration_ms: 800,
|
||||
..Default::default()
|
||||
};
|
||||
assert_eq!(p.throughput_window_ms(), 800);
|
||||
// With an interval, the client measurement wins — the 1 GbE field case: the same
|
||||
// bytes over 1010 ms instead of the host's 800 ms is the difference between an
|
||||
// honest ~940 Mbps and an impossible 1266 Mbps.
|
||||
let p = ProbeState {
|
||||
client_interval_ms: 1_010,
|
||||
host_duration_ms: 800,
|
||||
..Default::default()
|
||||
};
|
||||
assert_eq!(p.throughput_window_ms(), 1_010);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -132,12 +132,24 @@ impl ControlTask {
|
||||
}
|
||||
} else if let Ok(result) = ProbeResult::decode(&msg) {
|
||||
let mut p = probe.lock().unwrap();
|
||||
// Freeze the delivered figures now (the burst is done), before resumed
|
||||
// video can inflate the packet counters.
|
||||
// Freeze the delivered figures now (the burst is done). The mirrored
|
||||
// counters are probe-scoped (stamped at the reassembler's FLAG_PROBE
|
||||
// routing), so video around the burst inflates nothing; the client's
|
||||
// first→last arrival interval is frozen with them — the denominator
|
||||
// that measures when the bytes actually ARRIVED, not when the host
|
||||
// stopped sending (its window closes while the bottleneck queue is
|
||||
// still draining this way, which is how a 1 GbE link once "measured"
|
||||
// 1266 Mbps).
|
||||
let base_p = p.base_packets.unwrap_or(p.rx_packets_now);
|
||||
let base_b = p.base_bytes.unwrap_or(p.rx_bytes_now);
|
||||
p.delivered_packets = p.rx_packets_now.saturating_sub(base_p);
|
||||
p.delivered_bytes = p.rx_bytes_now.saturating_sub(base_b);
|
||||
p.client_interval_ms = ProbeState::measured_interval_ms(
|
||||
p.first_arrival_ns,
|
||||
p.last_arrival_ns,
|
||||
p.delivered_packets,
|
||||
)
|
||||
.unwrap_or(0);
|
||||
p.host_goodput_bytes = result.bytes_sent;
|
||||
p.host_au = result.packets_sent;
|
||||
p.host_wire_packets = result.wire_packets_sent;
|
||||
@@ -151,6 +163,7 @@ impl ControlTask {
|
||||
send_dropped = result.send_dropped,
|
||||
duration_ms = result.duration_ms,
|
||||
delivered_packets = p.delivered_packets,
|
||||
client_interval_ms = p.client_interval_ms,
|
||||
"speed-test probe result"
|
||||
);
|
||||
} else if let Ok(ack) = BitrateChanged::decode(&msg) {
|
||||
|
||||
@@ -200,10 +200,23 @@ impl DataPump {
|
||||
let probe_active = {
|
||||
let mut p = pump_probe.lock().unwrap();
|
||||
if p.active && !p.done {
|
||||
p.rx_packets_now = st.packets_received;
|
||||
p.rx_bytes_now = st.bytes_received;
|
||||
p.base_packets.get_or_insert(st.packets_received);
|
||||
p.base_bytes.get_or_insert(st.bytes_received);
|
||||
// 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
|
||||
};
|
||||
@@ -280,16 +293,23 @@ impl DataPump {
|
||||
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.
|
||||
// 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 delivered_kbps = (p.delivered_bytes.saturating_mul(8)
|
||||
/ p.host_duration_ms.max(1) as u64)
|
||||
as u32;
|
||||
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 {
|
||||
|
||||
@@ -409,6 +409,27 @@ impl Reassembler {
|
||||
// can neither advance the video anchor nor be dropped as stale against it (and its aged-out
|
||||
// frames never count as `frames_dropped`, which would fire video loss recovery).
|
||||
let is_probe = hdr.user_flags & (FLAG_PROBE as u32) != 0;
|
||||
if is_probe {
|
||||
// Probe-scoped receive accounting (the speed-test numerator + denominator, see
|
||||
// `Stats::probe_first_arrival_ns`), stamped at the routing decision so video in
|
||||
// flight around the burst contaminates neither the byte count nor the arrival
|
||||
// stamps. Byte unit mirrors `bytes_received` (whole plaintext packet). The first
|
||||
// probe packet since the pump armed the probe claims the first-arrival slot (the
|
||||
// pump zeroes it before the burst can reach the host); every probe packet
|
||||
// refreshes the last-arrival stamp.
|
||||
let now_ns = crate::stats::now_monotonic_ns();
|
||||
StatsCounters::add(&stats.probe_packets_received, 1);
|
||||
StatsCounters::add(&stats.probe_bytes_received, pkt.len() as u64);
|
||||
let _ = stats.probe_first_arrival_ns.compare_exchange(
|
||||
0,
|
||||
now_ns,
|
||||
std::sync::atomic::Ordering::Relaxed,
|
||||
std::sync::atomic::Ordering::Relaxed,
|
||||
);
|
||||
stats
|
||||
.probe_last_arrival_ns
|
||||
.store(now_ns, std::sync::atomic::Ordering::Relaxed);
|
||||
}
|
||||
let win = if is_probe { probe } else { video };
|
||||
win.advance_window(
|
||||
hdr.frame_index,
|
||||
|
||||
@@ -218,6 +218,18 @@ impl Session {
|
||||
self.stats.snapshot()
|
||||
}
|
||||
|
||||
/// Re-arm the probe-scoped arrival stamps (see [`Stats::probe_first_arrival_ns`]): zero
|
||||
/// them so the NEXT burst's first packet claims the first-arrival slot. Called by the
|
||||
/// client pump when it arms a probe, strictly before the burst can have reached the host
|
||||
/// (the `ProbeRequest` is still queued locally) — so the reset cannot race a probe packet.
|
||||
/// The cumulative probe byte/packet counters are left alone: per-burst deltas come from
|
||||
/// base snapshots, the same pattern the total counters use.
|
||||
pub fn reset_probe_arrivals(&self) {
|
||||
let l = std::sync::atomic::Ordering::Relaxed;
|
||||
self.stats.probe_first_arrival_ns.store(0, l);
|
||||
self.stats.probe_last_arrival_ns.store(0, l);
|
||||
}
|
||||
|
||||
/// Wrap a packet for the wire: when encrypting, prepend the 8-byte big-endian
|
||||
/// sequence (the receiver derives the GCM nonce from it) then the ciphertext.
|
||||
/// Seal one plaintext packet into the reused `wire` buffer in place (no allocation): the wire is
|
||||
|
||||
@@ -1,6 +1,19 @@
|
||||
//! Live counters for the frame-pacing / quality logic and the web UI.
|
||||
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::OnceLock;
|
||||
use std::time::Instant;
|
||||
|
||||
/// Monotonic now, in ns since an arbitrary process-wide epoch — the basis for the probe
|
||||
/// arrival stamps below. Monotonic on purpose: the stamps are only ever DIFFERENCED on this
|
||||
/// machine (the burst's receive interval), and a wall-clock step mid-burst — the exact event
|
||||
/// the clock re-sync machinery exists for — must not corrupt the one measurement the ABR
|
||||
/// ceiling is built from, so the CLOCK_REALTIME basis `pts_ns` uses is wrong here. Floored
|
||||
/// at 1 so a stamp can never collide with the 0 = "unset" sentinel.
|
||||
pub(crate) fn now_monotonic_ns() -> u64 {
|
||||
static EPOCH: OnceLock<Instant> = OnceLock::new();
|
||||
(EPOCH.get_or_init(Instant::now).elapsed().as_nanos() as u64).max(1)
|
||||
}
|
||||
|
||||
/// Immutable snapshot, copied across the C ABI as `PunktfunkStats`.
|
||||
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
|
||||
@@ -26,11 +39,29 @@ pub struct Stats {
|
||||
pub fec_late_shards: u64,
|
||||
pub bytes_sent: u64,
|
||||
pub bytes_received: u64,
|
||||
/// Probe-scoped receive counters: wire packets / plaintext bytes carrying
|
||||
/// [`FLAG_PROBE`](crate::packet::FLAG_PROBE) (speed-test filler), counted at the
|
||||
/// reassembler's probe routing decision. `bytes_received` counts EVERY accepted datagram,
|
||||
/// so a speed-test numerator built from it inherits whatever video was in flight around
|
||||
/// the burst — these keep video out of the probe math. Deliberately NOT mirrored into the
|
||||
/// C-ABI `PunktfunkStats` (probe measurements surface via `ProbeOutcome`).
|
||||
pub probe_packets_received: u64,
|
||||
pub probe_bytes_received: u64,
|
||||
/// First / last probe-packet arrival (monotonic ns, see [`now_monotonic_ns`]; 0 = none
|
||||
/// since the last probe arm). Their difference is the burst's client-side receive
|
||||
/// interval — the honest speed-test denominator: the host's send window closes while the
|
||||
/// switch/kernel queue toward the client is still draining, so dividing client bytes by
|
||||
/// the HOST duration overstates the link (a 1 GbE link "measured" 1266 Mbps). The client
|
||||
/// pump zeroes both when it arms a probe (`Session::reset_probe_arrivals`).
|
||||
pub probe_first_arrival_ns: u64,
|
||||
pub probe_last_arrival_ns: u64,
|
||||
}
|
||||
|
||||
/// Atomic accumulators owned by a [`Session`](crate::session::Session). Snapshot to
|
||||
/// [`Stats`] for readers. `Relaxed` ordering is fine: these are monotonic counters
|
||||
/// read for display, never used to synchronize other memory.
|
||||
/// read for display, never used to synchronize other memory. (The two probe arrival
|
||||
/// stamps are the exception — slots, not counters — but they carry no synchronization
|
||||
/// duty either: they are read hundreds of ms after the last write.)
|
||||
#[derive(Default)]
|
||||
pub struct StatsCounters {
|
||||
pub frames_submitted: AtomicU64,
|
||||
@@ -44,6 +75,10 @@ pub struct StatsCounters {
|
||||
pub fec_late_shards: AtomicU64,
|
||||
pub bytes_sent: AtomicU64,
|
||||
pub bytes_received: AtomicU64,
|
||||
pub probe_packets_received: AtomicU64,
|
||||
pub probe_bytes_received: AtomicU64,
|
||||
pub probe_first_arrival_ns: AtomicU64,
|
||||
pub probe_last_arrival_ns: AtomicU64,
|
||||
}
|
||||
|
||||
impl StatsCounters {
|
||||
@@ -66,6 +101,10 @@ impl StatsCounters {
|
||||
fec_late_shards: self.fec_late_shards.load(l),
|
||||
bytes_sent: self.bytes_sent.load(l),
|
||||
bytes_received: self.bytes_received.load(l),
|
||||
probe_packets_received: self.probe_packets_received.load(l),
|
||||
probe_bytes_received: self.probe_bytes_received.load(l),
|
||||
probe_first_arrival_ns: self.probe_first_arrival_ns.load(l),
|
||||
probe_last_arrival_ns: self.probe_last_arrival_ns.load(l),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -71,7 +71,7 @@ pub(super) fn synthetic_stream(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Bounds a speed-test [`ProbeRequest`] before bursting: a 3 Gbps / 5 s ceiling keeps a probe from
|
||||
/// Bounds a speed-test [`ProbeRequest`] before bursting: a 10 Gbps / 5 s ceiling keeps a probe from
|
||||
/// monopolizing the link or stalling the stream for too long. The ceiling is set ABOVE the session
|
||||
/// bitrate cap ([`MAX_BITRATE_KBPS`], 2 Gbps) on purpose — a probe should be able to demonstrate
|
||||
/// headroom past the rate a session will actually be configured to use, so the client can pick a
|
||||
|
||||
@@ -1767,7 +1767,11 @@ typedef struct {
|
||||
// Application goodput bytes / access units the host offered.
|
||||
uint64_t host_bytes;
|
||||
uint32_t host_packets;
|
||||
// The host's measured burst duration, milliseconds (the throughput denominator).
|
||||
// The throughput denominator, milliseconds: the client-measured burst receive interval
|
||||
// (first → last probe-packet arrival) once `done`; the host's measured send-window
|
||||
// duration when fewer than two probe packets arrived (no interval to measure from). The
|
||||
// host duration alone overstates throughput — its window closes while the bottleneck
|
||||
// queue is still draining toward the client.
|
||||
uint32_t elapsed_ms;
|
||||
// Delivered wire throughput = `recv_bytes * 8 / elapsed_ms` (kilobits/second).
|
||||
uint32_t throughput_kbps;
|
||||
@@ -2846,7 +2850,7 @@ PunktfunkStatus punktfunk_connection_wants_decode_latency(const PunktfunkConnect
|
||||
|
||||
#if defined(PUNKTFUNK_FEATURE_QUIC)
|
||||
// Start a bandwidth speed test: ask the host to burst filler over the data plane at
|
||||
// `target_kbps` of goodput for `duration_ms` (each clamped host-side to ≤ 3 Gbps / ≤ 5 s),
|
||||
// `target_kbps` of goodput for `duration_ms` (each clamped host-side to ≤ 10 Gbps / ≤ 5 s),
|
||||
// *briefly pausing video*. Non-blocking — poll [`punktfunk_connection_probe_result`] until its
|
||||
// `done` field is 1. Starting a probe resets any prior measurement.
|
||||
//
|
||||
|
||||
Reference in New Issue
Block a user