From 9dfed648c5b9723e972cff3d04fda755570ac780 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 03:10:39 -0800 Subject: [PATCH 1/7] chore(quest): claim quest/m2/quic-realtime/bbr-delay-bound.md Quest-Claim: b1e94b01025545db From 4a0416ea593934d512d7fbcf394c1409ee22b289 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 03:36:04 -0800 Subject: [PATCH 2/7] feat(congestion): BBR3 delay bound from the forward delay's trend Bbr3Config::delay_bound reads the peer's receive timestamps through a GoogCC-style trendline (burst grouping, smoothed accumulated delay, adaptive threshold) and bounds BBR.bw at the delivery rate the queue's growth measured: ProbeBW_UP stops there, growth outside a probe sets the bound to 0.85 of it so the queue drains, and Startup and ProbeBW_UP raise it to what they deliver. Off by default; without timestamps it is BBR3. moq-quic's tests::realtime replays the cellular moq-shaper presets on the sim and tabulates queueing delay, utilization, reaction to the drop, and share beside CUBIC for each candidate; probe.rs checks native probing still raises the estimate under the bound. Co-Authored-By: Claude Opus 5.5 --- quest/m2/quic-realtime/README.md | 6 +- quest/m2/quic-realtime/bbr-delay-bound.md | 22 -- rs/justfile | 4 +- rs/moq-congestion/README.md | 2 + rs/moq-congestion/src/bbr3/delay.rs | 454 ++++++++++++++++++++++ rs/moq-congestion/src/bbr3/mod.rs | 218 ++++++++++- rs/moq-quic/src/tests/mod.rs | 1 + rs/moq-quic/src/tests/probe.rs | 48 ++- rs/moq-quic/src/tests/realtime.rs | 384 ++++++++++++++++++ rs/moq-quic/src/tests/util.rs | 42 +- 10 files changed, 1146 insertions(+), 35 deletions(-) delete mode 100644 quest/m2/quic-realtime/bbr-delay-bound.md create mode 100644 rs/moq-congestion/src/bbr3/delay.rs create mode 100644 rs/moq-quic/src/tests/realtime.rs diff --git a/quest/m2/quic-realtime/README.md b/quest/m2/quic-realtime/README.md index 2192afeb0b0..2eb3b6d0486 100644 --- a/quest/m2/quic-realtime/README.md +++ b/quest/m2/quic-realtime/README.md @@ -39,7 +39,10 @@ spike assumed no consumer; teleoperation uplinks are one): in `moq-congestion`. The moq-shaper presets `cellular-step`, `cellular-bursty` and `cellular-deep` decide it. The report states the shaper's timing jitter, measured over `near-zero`, as the noise floor of the - delay signal. + delay signal. The BBR3 delay bound (`Bbr3Config::delay_bound`) landed + first, with the harness the others join: moq-quic's `tests::realtime` + replays the three presets on the sim, and each candidate adds a + `Candidate` there (`just rs quic-sim` prints the table). - Native probing is required (decided 2026-10-25, maintainer). Every candidate implements `moq_congestion::Controller::probes` and reports `probe_inflight`, so moq-quic's native probing @@ -85,7 +88,6 @@ owns that feedback. ## Required - [GCC candidate](/quest/m2/quic-realtime/gcc.md) - libwebrtc's GoogCC vendored into `moq-congestion` and refreshed to current WebRTC -- [BBR3 delay bound](/quest/m2/quic-realtime/bbr-delay-bound.md) - BBR3 with inflight capped when the forward delay gradient shows queue growth - [SCReAMv2 candidate](/quest/m2/quic-realtime/scream.md) - the RFC 8298bis delay path ported to `moq-congestion` - [moqsink encoder feedback](/quest/m2/quic-realtime/moqsink-feedback.md) - GStreamer publishers retune their encoder to the target rate diff --git a/quest/m2/quic-realtime/bbr-delay-bound.md b/quest/m2/quic-realtime/bbr-delay-bound.md deleted file mode 100644 index fd1b1224968..00000000000 --- a/quest/m2/quic-realtime/bbr-delay-bound.md +++ /dev/null @@ -1,22 +0,0 @@ -# [M] BBR3 delay bound - -## Goal - -A BBR3 variant in `moq-congestion` caps inflight and pacing when the forward delay -gradient from receive timestamps shows queue growth, and reports its queuing -delay, throughput, and reaction time on the shared moq-shaper uplink presets. - -## Plan - -The smallest candidate: it keeps BBR3's model, startup, and fairness, and only -adds a queue-delay brake. The comparison must show whether BBR3's ProbeBW -up-cycles still push transient queues the brake cannot stop; that is the case -against it. - -- Build it as a configuration of `moq-congestion`'s BBR3, not a fork of the module, - so a loss leaves nothing to delete but the option. -- Keep BBR3's `Controller::probes` and `probe_inflight` working under the - brake, so moq-quic's native probing still raises the estimate (decided 2026-10-25, maintainer; - see the [epic](/quest/m2/quic-realtime/README.md)). Measure that a stopped - rung recovers under it. -- Unit tests drive the controller with mocked time, not wall clock. diff --git a/rs/justfile b/rs/justfile index 163e7581bd1..b7e01053189 100644 --- a/rs/justfile +++ b/rs/justfile @@ -292,7 +292,7 @@ play *args: doctest *args: bash sh/local.sh cargo test --locked --doc "$@" -# Run the moq-quic simulation's random workloads from a fresh seed, printing the settings that replay it, then the bandwidth probe's acceptance suite and the receive-timestamps comparison. +# Run the moq-quic simulation's random workloads from a fresh seed, printing the settings that replay it, then the bandwidth probe's acceptance suite, the receive-timestamps comparison, and the realtime controllers' comparison. quic-sim: #!/usr/bin/env bash set -euo pipefail @@ -307,6 +307,8 @@ quic-sim: # Forward delay against RTT as a queue-growth signal: deterministic in the simulation, then over real sockets through each cellular moq-shaper preset, one at a time so they do not add to each other's noise. bash sh/rs/nextest.sh cargo nextest run --locked --release -p moq-quic --lib --run-ignored only --no-capture -E 'test(/^tests::receive_ts::/)' bash sh/rs/nextest.sh cargo nextest run --locked --release -p moq-tokio --test integration --run-ignored only --no-capture --test-threads 1 -E 'test(/^receive_ts::/)' + # Each realtime congestion controller against BBR3 over sim replicas of the cellular presets: queueing delay, throughput, reaction to the drop, and share beside CUBIC. + bash sh/rs/nextest.sh cargo nextest run --locked --release -p moq-quic --lib --run-ignored only --no-capture -E 'test(/^tests::realtime::/)' # Prove downstream futures are Send at the default recursion limit on the # flake's pinned nightly. Nightly runs it; the land gate does not. diff --git a/rs/moq-congestion/README.md b/rs/moq-congestion/README.md index 898d009fb12..4e9ae615e0c 100644 --- a/rs/moq-congestion/README.md +++ b/rs/moq-congestion/README.md @@ -2,6 +2,8 @@ BBR3, Cubic, and Unlimited congestion controllers, consumed by moq-quic and the native transport drivers. +BBR3's optional delay bound (`Bbr3Config::delay_bound`) reads the peer's receive timestamps and brakes when the forward delay's trend shows the bottleneck queue growing. + The controllers take explicit `kio::clock::Instant` values and never drive a runtime. The `Controller` trait names no transport types: `on_sent` returns an opaque per-packet state that the transport stores and hands back with each acknowledgement or loss, and the controller paces its own sends. The code was extracted from [moq-quic](../moq-quic), under the same MIT or Apache-2.0 license. Cubic originated in quinn; BBR3 was ported from moq-dev/noq and carries the fixes listed in [moq-quic's fork history](../moq-quic/README.md). diff --git a/rs/moq-congestion/src/bbr3/delay.rs b/rs/moq-congestion/src/bbr3/delay.rs new file mode 100644 index 00000000000..22802484856 --- /dev/null +++ b/rs/moq-congestion/src/bbr3/delay.rs @@ -0,0 +1,454 @@ +//! Bottleneck queue growth, read off the forward delay that receive timestamps expose. +//! +//! The peer reports when each packet arrived, on a clock whose basis it never shares, so a +//! packet's forward delay is known only up to a constant: its trend is what carries the queue. +//! This follows libwebrtc's GoogCC delay detector (`InterArrivalDelta`, `TrendlineEstimator`, +//! and its adaptive threshold, under `modules/congestion_controller/goog_cc`), with its +//! constants: +//! +//! - Packets are grouped into bursts: those sent within [`GROUP_LENGTH`] of a group's first, and +//! those that arrive together, as a cellular scheduler's grant releases them. Per-packet delay +//! sees each grant's bunching as noise, which a group's last arrival does not. +//! - Each group's delay variation against the previous one (arrival spacing less send spacing) +//! accumulates into the forward delay, up to the unknown constant. An exponential filter +//! smooths it, and a least-squares line over the last [`WINDOW`] groups gives its trend. +//! - The trend, scaled by how many groups it has seen, is held to a threshold that adapts to +//! the path's own noise. A trend past it for longer than [`OVERUSE_TIME`], still steepening, +//! is queue growth, which stays reported until the trend falls back under the threshold. + +use std::collections::VecDeque; + +use crate::{Duration, Instant}; + +/// Packets sent within this of a group's first belong to it. +const GROUP_LENGTH: Duration = Duration::from_millis(5); +/// A packet arriving within this of the previous one, sooner than it was sent after it, was +/// released together with it. +const BURST_DELTA: Duration = Duration::from_millis(5); +/// The longest a burst of arrivals counts as one group. +const MAX_BURST: Duration = Duration::from_millis(100); +/// Arrival spacing this much past the local spacing of the ACKs that reported it means the +/// peer's clock jumped. +const ARRIVAL_OFFSET_THRESHOLD: Duration = Duration::from_secs(3); +/// Groups in a row that arrived before the previous one, after which the detector starts over. +const REORDERED_RESET: u32 = 3; +/// Groups the trend line is fit over. +const WINDOW: usize = 20; +/// The exponential filter's weight on the previous smoothed delay. +const SMOOTHING: f64 = 0.9; +/// Scales the trend before it meets the threshold. +const THRESHOLD_GAIN: f64 = 4.0; +/// The most groups the trend's scale counts. +const MIN_DELTAS: u32 = 60; +/// How long, in milliseconds, the trend must stay over the threshold to count as queue growth. +const OVERUSE_TIME: f64 = 10.0; +/// A trend this far past the threshold is a spike, such as a capacity drop, that the threshold +/// does not adapt to, in milliseconds. +const MAX_ADAPT_OFFSET: f64 = 15.0; +/// How fast the threshold rises toward a trend above it, per millisecond. +const K_UP: f64 = 0.0087; +/// How fast the threshold falls toward a trend below it, per millisecond. +const K_DOWN: f64 = 0.039; +const THRESHOLD_INITIAL: f64 = 12.5; +const THRESHOLD_MIN: f64 = 6.0; +const THRESHOLD_MAX: f64 = 600.0; +/// The most time, in milliseconds, one threshold update accounts for. +const THRESHOLD_MAX_STEP: f64 = 100.0; + +/// What the forward delay's trend says of the bottleneck queue +#[derive(Debug, Clone, Copy, Default, Eq, PartialEq)] +pub(super) enum Usage { + /// Neither growing nor shrinking past the path's noise. + #[default] + Normal, + /// Growing: the sender is past the bottleneck's rate. + Overusing, + /// Shrinking. + Underusing, +} + +/// Packets sent, or released, together +#[derive(Debug, Clone, Copy)] +struct Group { + first_send: Instant, + /// The latest send time in the group. + send: Instant, + first_arrival: Duration, + /// The last arrival. + complete: Duration, + /// When the ACK that reported the last arrival came. + acked: Instant, +} + +/// Reads queue growth from the forward delay's trend +#[derive(Debug, Clone)] +pub(super) struct Detector { + current: Option, + prev: Option, + /// Groups in a row that arrived before the one before them. + reordered: u32, + /// Group deltas seen, up to a cap. + deltas: u32, + /// The first group's arrival, in milliseconds. + first_arrival: Option, + /// The forward delay, in milliseconds, up to the peer's unknown basis. + accumulated: f64, + smoothed: f64, + /// The last [`WINDOW`] groups' arrival, since the first, and smoothed delay, in milliseconds. + history: VecDeque<(f64, f64)>, + trend: f64, + /// How long the trend has been over the threshold, in milliseconds. + overusing: Option, + overuse_count: u32, + threshold: f64, + /// The arrival, in milliseconds, of the last threshold update. + threshold_updated: Option, + usage: Usage, + /// Whether the trend has asserted growth since [`Self::take_signal`] last ran. + signaled: bool, +} + +impl Default for Detector { + fn default() -> Self { + Self { + current: None, + prev: None, + reordered: 0, + deltas: 0, + first_arrival: None, + accumulated: 0.0, + smoothed: 0.0, + history: VecDeque::with_capacity(WINDOW + 1), + trend: 0.0, + overusing: None, + overuse_count: 0, + threshold: THRESHOLD_INITIAL, + threshold_updated: None, + usage: Usage::Normal, + signaled: false, + } + } +} + +impl Detector { + /// What the trend says of the queue as of the last packet. + #[cfg(test)] + pub(super) fn usage(&self) -> Usage { + self.usage + } + + /// Whether the trend asserted growth since the last call: it crossed the threshold, or + /// stayed past it for another [`OVERUSE_TIME`] still steepening. + /// + /// [`Self::usage`] stays [`Usage::Overusing`] until the trend falls back under the threshold, + /// which the window's lag puts well after the queue stops growing. A response read from this + /// instead acts only on fresh growth. + pub(super) fn take_signal(&mut self) -> bool { + std::mem::take(&mut self.signaled) + } + + /// A packet sent at `sent` arrived at `received` on the peer's clock, as an ACK arriving at + /// `acked` reported. Packets are fed in the order they were sent within each ACK. + /// + /// equivalent to `InterArrivalDelta::ComputeDeltas`, then `TrendlineEstimator::Update` + pub(super) fn on_packet(&mut self, sent: Instant, received: Duration, acked: Instant) { + let Some(mut current) = self.current else { + self.current = Some(Group { + first_send: sent, + send: sent, + first_arrival: received, + complete: received, + acked, + }); + return; + }; + if sent < current.first_send { + // Sent before the group in progress: a late ACK for a reordered packet. + return; + } + if self.new_group(¤t, sent, received) { + if let Some(prev) = self.prev + && let Some((send_delta, arrival_delta)) = self.deltas(&prev, ¤t) + { + self.update_trend(send_delta, arrival_delta, millis(current.complete)); + } + if self.current.is_none() { + // The deltas reset the detector. + return self.on_packet(sent, received, acked); + } + self.prev = Some(current); + current = Group { + first_send: sent, + send: sent, + first_arrival: received, + complete: received, + acked, + }; + } else { + current.send = current.send.max(sent); + } + current.complete = received; + current.acked = acked; + self.current = Some(current); + } + + /// equivalent to `InterArrivalDelta::NewTimestampGroup` + fn new_group(&self, current: &Group, sent: Instant, received: Duration) -> bool { + if Self::burst(current, sent, received) { + return false; + } + sent.saturating_duration_since(current.first_send) > GROUP_LENGTH + } + + /// equivalent to `InterArrivalDelta::BelongsToBurst`: arrived on the heels of the group, + /// sooner than it was sent after it, as a batch the path released together. + fn burst(current: &Group, sent: Instant, received: Duration) -> bool { + let send_delta = sent.saturating_duration_since(current.send); + if send_delta.is_zero() { + return true; + } + let Some(arrival_delta) = received.checked_sub(current.complete) else { + return false; + }; + arrival_delta < send_delta + && arrival_delta <= BURST_DELTA + && received.saturating_sub(current.first_arrival) < MAX_BURST + } + + /// The send and arrival spacing of `current` after `prev`, in milliseconds, or `None` if the + /// pair says nothing: reordered, or across a jump in the peer's clock, which resets. + fn deltas(&mut self, prev: &Group, current: &Group) -> Option<(f64, f64)> { + let send_delta = signed_millis(current.send, prev.send); + let arrival_delta = millis(current.complete) - millis(prev.complete); + let acked_delta = signed_millis(current.acked, prev.acked); + if arrival_delta - acked_delta >= millis(ARRIVAL_OFFSET_THRESHOLD) { + *self = Self::default(); + return None; + } + if arrival_delta < 0.0 { + self.reordered += 1; + if self.reordered >= REORDERED_RESET { + *self = Self::default(); + } + return None; + } + self.reordered = 0; + Some((send_delta, arrival_delta)) + } + + /// equivalent to `TrendlineEstimator::UpdateTrendline`, at `arrival` on the peer's clock + fn update_trend(&mut self, send_delta: f64, arrival_delta: f64, arrival: f64) { + let delta = arrival_delta - send_delta; + self.deltas = (self.deltas + 1).min(1000); + let first = *self.first_arrival.get_or_insert(arrival); + self.accumulated += delta; + self.smoothed = SMOOTHING * self.smoothed + (1.0 - SMOOTHING) * self.accumulated; + self.history.push_back((arrival - first, self.smoothed)); + if self.history.len() > WINDOW { + self.history.pop_front(); + } + let mut trend = self.trend; + if self.history.len() == WINDOW { + trend = slope(&self.history).unwrap_or(trend); + } + self.detect(trend, send_delta, arrival); + } + + /// equivalent to `TrendlineEstimator::Detect` + fn detect(&mut self, trend: f64, send_delta: f64, arrival: f64) { + if self.deltas < 2 { + self.usage = Usage::Normal; + return; + } + let modified = f64::from(self.deltas.min(MIN_DELTAS)) * trend * THRESHOLD_GAIN; + if modified > self.threshold { + // Assume it has been over for half the time since the previous group. + let over = self + .overusing + .map_or(send_delta / 2.0, |over| over + send_delta); + self.overusing = Some(over); + self.overuse_count += 1; + if over > OVERUSE_TIME && self.overuse_count > 1 && trend >= self.trend { + self.overusing = Some(0.0); + self.overuse_count = 0; + self.usage = Usage::Overusing; + self.signaled = true; + } + } else if modified < -self.threshold { + self.overusing = None; + self.overuse_count = 0; + self.usage = Usage::Underusing; + } else { + self.overusing = None; + self.overuse_count = 0; + self.usage = Usage::Normal; + } + self.trend = trend; + self.update_threshold(modified, arrival); + } + + /// equivalent to `TrendlineEstimator::UpdateThreshold` + fn update_threshold(&mut self, modified: f64, arrival: f64) { + let last = *self.threshold_updated.get_or_insert(arrival); + self.threshold_updated = Some(arrival); + if modified.abs() > self.threshold + MAX_ADAPT_OFFSET { + return; + } + let k = if modified.abs() < self.threshold { + K_DOWN + } else { + K_UP + }; + let elapsed = (arrival - last).clamp(0.0, THRESHOLD_MAX_STEP); + self.threshold += k * (modified.abs() - self.threshold) * elapsed; + self.threshold = self.threshold.clamp(THRESHOLD_MIN, THRESHOLD_MAX); + } +} + +/// The least-squares slope of `points`, or `None` if they share one x +fn slope(points: &VecDeque<(f64, f64)>) -> Option { + let n = points.len() as f64; + let x = points.iter().map(|p| p.0).sum::() / n; + let y = points.iter().map(|p| p.1).sum::() / n; + let (mut num, mut den) = (0.0, 0.0); + for (px, py) in points { + num += (px - x) * (py - y); + den += (px - x).powi(2); + } + (den != 0.0).then(|| num / den) +} + +fn millis(d: Duration) -> f64 { + d.as_secs_f64() * 1e3 +} + +/// `a - b` in milliseconds, negative if `a` is earlier +fn signed_millis(a: Instant, b: Instant) -> f64 { + millis(a.saturating_duration_since(b)) - millis(b.saturating_duration_since(a)) +} + +#[cfg(test)] +mod test { + use super::*; + + /// Feeds `count` packets sent every `spacing`, each arriving `delay(i)` after it was sent, + /// acknowledged on arrival, and returns the usage after each. + fn feed( + detector: &mut Detector, + base: Instant, + count: u64, + spacing: Duration, + delay: impl Fn(u64) -> Duration, + ) -> Vec { + (0..count) + .map(|i| { + let sent = base + spacing * i as u32; + let received = sent - base + delay(i); + detector.on_packet(sent, received, sent + delay(i)); + detector.usage() + }) + .collect() + } + + #[test] + fn steady_path_is_normal() { + let mut detector = Detector::default(); + let usage = feed( + &mut detector, + Instant::now(), + 2000, + Duration::from_millis(2), + |_| Duration::from_millis(20), + ); + assert!(usage.iter().all(|&u| u == Usage::Normal)); + } + + /// A queue growing by a millisecond every 10ms, as a sender 10% over the bottleneck builds + #[test] + fn growing_queue_overuses() { + let mut detector = Detector::default(); + let base = Instant::now(); + let usage = feed(&mut detector, base, 1000, Duration::from_millis(2), |i| { + Duration::from_millis(20) + Duration::from_micros(i.saturating_sub(500) * 200) + }); + assert!(usage[..500].iter().all(|&u| u == Usage::Normal)); + let first = usage.iter().position(|&u| u == Usage::Overusing); + // Within 100ms of the queue starting to grow, 10ms of it + assert!(first.is_some_and(|i| i < 550), "first overuse at {first:?}"); + } + + /// A draining queue never reads as growth + #[test] + fn draining_queue_is_not_growth() { + let mut detector = Detector::default(); + let usage = feed( + &mut detector, + Instant::now(), + 1000, + Duration::from_millis(2), + |i| Duration::from_millis(220) - Duration::from_micros(i * 200), + ); + assert!(!usage.contains(&Usage::Overusing)); + } + + /// Arrivals bunched into 20ms grants, with no queue behind them, never read as growth + #[test] + fn grants_are_not_growth() { + let mut detector = Detector::default(); + let grant = Duration::from_millis(20); + let usage = feed( + &mut detector, + Instant::now(), + 3000, + Duration::from_millis(2), + |i| { + // Released at the next grant boundary after a 20ms path + let sent = Duration::from_millis(2 * i); + let arrival = sent + Duration::from_millis(20); + let released = arrival.as_nanos().div_ceil(grant.as_nanos()) * grant.as_nanos(); + Duration::from_nanos(released as u64) - sent + }, + ); + assert!(!usage.contains(&Usage::Overusing)); + } + + /// A jump in the peer's clock starts the detector over rather than reading as a queue + #[test] + fn clock_jump_resets() { + let mut detector = Detector::default(); + let base = Instant::now(); + let spacing = Duration::from_millis(2); + let delay = Duration::from_millis(20); + let mut usage = Vec::new(); + for i in 0..1000u32 { + let sent = base + spacing * i; + // The peer's clock jumps 10s ahead halfway, which its ACKs' arrivals do not + let jump = if i < 500 { + Duration::ZERO + } else { + Duration::from_secs(10) + }; + detector.on_packet(sent, sent - base + delay + jump, sent + delay); + usage.push(detector.usage()); + } + assert!(!usage.contains(&Usage::Overusing)); + } + + /// Packets sent before the group in progress, from a late ACK, are skipped + #[test] + fn late_packets_are_skipped() { + let mut detector = Detector::default(); + let base = Instant::now(); + feed(&mut detector, base, 100, Duration::from_millis(2), |_| { + Duration::from_millis(20) + }); + let before = detector.clone(); + detector.on_packet( + base, + Duration::from_millis(500), + base + Duration::from_secs(1), + ); + assert_eq!(detector.deltas, before.deltas); + assert_eq!(detector.usage(), Usage::Normal); + } +} diff --git a/rs/moq-congestion/src/bbr3/mod.rs b/rs/moq-congestion/src/bbr3/mod.rs index 3dbfdfa29f8..bfac766bf4d 100644 --- a/rs/moq-congestion/src/bbr3/mod.rs +++ b/rs/moq-congestion/src/bbr3/mod.rs @@ -1,5 +1,6 @@ #[cfg(test)] mod bench; +mod delay; mod max_filter; use std::any::Any; @@ -90,6 +91,11 @@ const LOSS_THRESH: f64 = 0.02; /// each round trip during which the connection detects packet loss (the value is 0.7). const BETA: f64 = 0.7; +/// The delay bound's pacing, as a share of the delivery rate while the bottleneck queue grows: +/// GoogCC's backoff from its acknowledged rate on overuse. Below the bottleneck's rate, it drains +/// the queue at 15% of that rate. +const DELAY_BETA: f64 = 0.85; + /// equivalent to BBR.Headroom: A constant specifying the multiplicative factor to apply to /// BBR.inflight_longterm when calculating a volume of free headroom to try to leave unused in the /// path (e.g. free space in the bottleneck buffer or free time slots in the bottleneck link) that @@ -523,6 +529,14 @@ pub struct Bbr3 { /// Used when computing `offload_budget` per /// . max_ack_delay: Duration, + /// Reads bottleneck queue growth from the peer's receive timestamps, when + /// [`Bbr3Config::delay_bound`] is set. + delay: Option, + /// The delay bound's ceiling on BBR.bw: the most the path delivered without its queue + /// growing. Infinite until the queue first grows, and always without the delay bound. + bw_delay: f64, + /// C.delivered when the delay bound last acted, which it does once per round trip. + delay_round_delivered: u64, } impl Bbr3 { @@ -680,6 +694,9 @@ impl Bbr3 { // when the connection supplies peer ACK-frequency parameters. ack_eliciting_threshold: 1, max_ack_delay: Duration::from_millis(25), + delay: config.delay_bound.then(delay::Detector::default), + bw_delay: f64::INFINITY, + delay_round_delivered: 0, } } @@ -823,6 +840,10 @@ impl Bbr3 { } /// equivalent to BBRUpdateACKAggregation + /// + /// The delay bound expects delivery at BBR.bw without its own `bw_delay`: pacing below the + /// delivery rate to drain a queue delivers faster than that bound while the queue lasts, + /// which is not ACK aggregation, and taking it as such inflates the window. fn update_ack_aggregation(&mut self, newly_acked: u64, now: Instant) { let interval; if let Some(extra_acked_interval_start) = self.extra_acked_interval_start { @@ -830,7 +851,11 @@ impl Bbr3 { } else { interval = Duration::from_secs(0); } - let mut expected_delivered = (self.bw * interval.as_secs_f64()) as u64; + let bw = match self.bw_delay.is_finite() { + true => self.max_bw.min(self.bw_shortterm), + false => self.bw, + }; + let mut expected_delivered = (bw * interval.as_secs_f64()) as u64; if self.extra_acked_delivered <= expected_delivered { self.extra_acked_delivered = 0; self.extra_acked_interval_start = Some(now); @@ -1316,9 +1341,11 @@ impl Bbr3 { } /// equivalent to BBRBoundBWForModel + /// + /// The delay bound's `bw_delay` bounds it too, infinite unless that is on. fn bound_bw_for_model(&mut self) { - // gives min of both f64 - self.bw = [self.max_bw, self.bw_shortterm] + // gives min of all three f64 + self.bw = [self.max_bw, self.bw_shortterm, self.bw_delay] .iter() .copied() .fold(f64::NAN, f64::min); @@ -1785,6 +1812,56 @@ impl Bbr3 { let inflight = Ord::max(inflight, self.offload_budget); Some(Ord::max(inflight, self.min_pipe_cwnd)) } + + /// The delay bound's response to the forward delay's trend showing the bottleneck queue + /// growing, at most once per round trip, so the response to one queue lands before it is + /// judged again. + /// + /// While the queue grows, the bottleneck's rate bounds the delivery rate, so the round's + /// delivery rate measures it. ProbeBW_UP has found it: `bw_delay` falls to it, and the probe + /// stops, as loss stops it, so ProbeBW_DOWN drains the probe's queue. Growth in any other + /// ProbeBW phase means the path slowed under the estimate: `bw_delay` falls to + /// [`DELAY_BETA`] of the delivery rate, so pacing drains the queue that slowing built. + /// + /// Startup, Drain, and ProbeRTT are BBR3's own: Startup ends on its plateau or on loss, and + /// Drain and ProbeRTT already send below the estimate. + fn bound_queue_growth(&mut self, now: Instant) { + let Some(rs) = self.rs else { + return; + }; + if rs.prior_delivered < self.delay_round_delivered { + return; + } + let delivery_rate = self.bw_latest.max(rs.delivery_rate); + match self.state { + BbrState::ProbeBw(ProbeBwSubstate::Up) => { + self.bw_delay = self.bw_delay.min(delivery_rate); + self.start_probe_bw_down(now); + } + BbrState::ProbeBw(_) => { + self.bw_delay = self.bw_delay.min(DELAY_BETA * delivery_rate); + } + BbrState::Startup | BbrState::Drain | BbrState::ProbeRtt => return, + } + self.bound_bw_for_model(); + self.delay_round_delivered = self.rate.delivered(); + } + + /// Raises `bw_delay` to what Startup and ProbeBW_UP deliver without the queue growing: a + /// probe the path carries lifts it a probe gain per round, as it lifts BBR.max_bw. + fn raise_bw_delay(&mut self) { + let Some(rs) = self.rs else { + return; + }; + let probing = matches!( + self.state, + BbrState::Startup | BbrState::ProbeBw(ProbeBwSubstate::Up) + ); + if probing && !rs.is_app_limited && rs.delivery_rate > self.bw_delay { + self.bw_delay = rs.delivery_rate; + self.bound_bw_for_model(); + } + } } impl Controller for Bbr3 { @@ -1843,6 +1920,11 @@ impl Controller for Bbr3 { for acked in ack.packets { self.check_recovery_done(acked.sent); newly_acked += u64::from(acked.bytes); + if let Some(delay) = &mut self.delay + && let Some(received) = acked.received + { + delay.on_packet(acked.sent, received, ack.now); + } } self.inflight = ack.in_flight; // An ACK that delivered no packet has no sample, and a finished one is never folded @@ -1857,6 +1939,12 @@ impl Controller for Bbr3 { }; self.rs = Some(rs); self.update_model_and_state(rs.prior_delivered, newly_acked, ack.app_limited, ack.now); + if let Some(grew) = self.delay.as_mut().map(delay::Detector::take_signal) { + self.raise_bw_delay(); + if grew { + self.bound_queue_growth(ack.now); + } + } self.update_control_parameters(newly_acked); } @@ -2008,9 +2096,27 @@ pub struct Bbr3Config { drain_pacing_gain: Option, pacing_margin_percent: Option, default_cwnd_gain: Option, + delay_bound: bool, } impl Bbr3Config { + /// Brake when the peer's receive timestamps show the bottleneck queue growing. + /// + /// A trend line over the forward delay, as GoogCC fits one, detects the queue growing. BBR3 + /// then bounds its bandwidth estimate, and with it the pacing rate and the window, at the + /// delivery rate the queue's growth measured: a bandwidth probe stops there, as it stops on + /// loss, and growth outside a probe sets the bound below that rate, so the queue drains. + /// Startup and ProbeBW_UP raise the bound to whatever they deliver, so the probes, native + /// probing included, still find capacity that comes back. Startup and the model are BBR3's + /// own. + /// + /// Without receive timestamps, which the transport requests separately, it does nothing. + /// Off by default. + pub fn delay_bound(&mut self, value: bool) -> &mut Self { + self.delay_bound = value; + self + } + /// Default limit on the amount of outstanding data in bytes. /// /// Recommended value: `min(10 * max_datagram_size, max(2 * max_datagram_size, 14720))` @@ -2043,6 +2149,7 @@ impl Default for Bbr3Config { drain_pacing_gain: None, pacing_margin_percent: None, default_cwnd_gain: None, + delay_bound: false, } } } @@ -2108,13 +2215,25 @@ mod test { /// Add packet `pn` to the ACK being built; a number not in flight is ignored. fn ack(&mut self, now: Instant, sent: Instant, bytes: u64, pn: u64) { + self.ack_received(now, sent, bytes, pn, None); + } + + /// [`Self::ack`], reporting when the peer received the packet. + fn ack_received( + &mut self, + now: Instant, + sent: Instant, + bytes: u64, + pn: u64, + received: Option, + ) { let _ = now; if let Some(acked) = self.sent.remove(&pn) { self.acked.push(Acked { sent, bytes: bytes as u16, packet: acked.packet, - received: None, + received, }); } } @@ -2177,6 +2296,8 @@ mod test { pn: u64, send_ns: u64, ack_ns: u64, + /// when the peer received it + recv_ns: u64, /// whether the bottleneck marked the packet CE ce: bool, } @@ -2210,6 +2331,10 @@ mod test { ce_above_ns: Option, /// the longest queueing delay a packet has met at the bottleneck max_queue_ns: u64, + /// whether each ACK reports when the peer received the packet, as receive timestamps do + timestamps: bool, + /// each packet's queueing delay at the bottleneck, by when it reached it + queues: Vec<(u64, u64)>, } impl Sim { @@ -2229,6 +2354,8 @@ mod test { btl_service_ns: (mss as f64 / bw * 1e9).round() as u64, ce_above_ns: None, max_queue_ns: 0, + timestamps: false, + queues: Vec::new(), } } @@ -2248,6 +2375,7 @@ mod test { pn: self.pn, send_ns: now_ns, ack_ns: now_ns, + recv_ns: now_ns, ce: false, }); self.pn += 1; @@ -2354,6 +2482,7 @@ mod test { let ack_ns = finish + self.ret_ns; let queue_ns = service_start - arrival; self.max_queue_ns = self.max_queue_ns.max(queue_ns); + self.queues.push((arrival, queue_ns)); let ce = self.ce_above_ns.is_some_and(|limit| queue_ns > limit); self.bbr.send( @@ -2366,6 +2495,7 @@ mod test { pn: self.pn, send_ns, ack_ns, + recv_ns: finish, ce, }); @@ -2382,7 +2512,9 @@ mod test { self.inflight -= self.mss; let now_at = self.base + Duration::from_nanos(self.now_ns); let send_at = self.base + Duration::from_nanos(p.send_ns); - self.bbr.ack(now_at, send_at, self.mss, p.pn); + let received = self.timestamps.then(|| Duration::from_nanos(p.recv_ns)); + self.bbr + .ack_received(now_at, send_at, self.mss, p.pn, received); self.bbr.end_acks(now_at, self.inflight, false); // Each ACK covers one packet, so a marked one raises the CE count. if p.ce { @@ -2431,6 +2563,7 @@ mod test { drain_pacing_gain: None, pacing_margin_percent: None, default_cwnd_gain: None, + delay_bound: false, }; let mut bbr3 = Driver::new(Arc::new(config), 2500); bbr3.pick_probe_wait(); @@ -8151,4 +8284,79 @@ mod test { assert!(marked.bbr.max_bw > 0.9 * control.bbr.max_bw); assert_eq!(marked.bbr.inflight_shortterm, u64::MAX); } + + /// A [`Sim`] whose peer reports receive timestamps, with the delay bound on or off + fn timestamped(delay_bound: bool) -> Sim { + let mut config = Bbr3Config { + probe_rng_seed: Some([7; 16]), + ..Bbr3Config::default() + }; + config.delay_bound(delay_bound); + let mut sim = Sim::new(config, 1200, LINK_BW, LINK_RTT); + sim.timestamps = true; + sim + } + + /// The `p`th percentile of the queueing delay packets met reaching the bottleneck in + /// `from..until`, in nanoseconds + fn queue_percentile(sim: &Sim, from: u64, until: u64, p: f64) -> u64 { + let mut queues: Vec = sim + .queues + .iter() + .filter(|(at, _)| (from..until).contains(at)) + .map(|&(_, queue)| queue) + .collect(); + queues.sort_unstable(); + queues[((queues.len() - 1) as f64 * p) as usize] + } + + /// The capacity drops to a quarter for 5s, then comes back. BBR3 keeps sending into the + /// queue its stale estimate builds until its window binds, while the delay bound reads the + /// queue growing and paces below the delivery rate until it drains. Its ProbeBW_UP then + /// finds the capacity again, at nearly BBR3's goodput throughout. + #[test] + fn delay_bound_drains_queue_after_capacity_drop() { + const SEC: u64 = 1_000_000_000; + let mut control = timestamped(false); + let mut bound = timestamped(true); + let mut delivered = [[0; 3]; 2]; + for (sim, delivered) in [&mut control, &mut bound].into_iter().zip(&mut delivered) { + delivered[0] = run_until(sim, 5 * SEC); + sim.btl_service_ns *= 4; + delivered[1] = run_until(sim, 10 * SEC); + sim.btl_service_ns /= 4; + delivered[2] = run_until(sim, 15 * SEC); + } + let median = |sim: &Sim| queue_percentile(sim, 5 * SEC, 10 * SEC, 0.5); + let (control_queue, bound_queue) = (median(&control), median(&bound)); + assert!( + bound_queue * 4 < control_queue, + "median queue {}ms with the bound, {}ms without", + bound_queue / MS, + control_queue / MS + ); + let [control_bytes, bound_bytes] = delivered; + for phase in 0..3 { + assert!( + bound_bytes[phase] as f64 > 0.9 * control_bytes[phase] as f64, + "phase {phase}: {bound_bytes:?} bytes delivered with the bound, {control_bytes:?} without" + ); + } + } + + /// The delay bound reads only receive timestamps: without them it changes nothing + #[test] + fn delay_bound_without_timestamps_is_bbr3() { + const SEC: u64 = 1_000_000_000; + let mut control = timestamped(false); + let mut bound = timestamped(true); + bound.timestamps = false; + for sim in [&mut control, &mut bound] { + run_until(sim, 3 * SEC); + sim.btl_service_ns *= 4; + run_until(sim, 6 * SEC); + } + assert_eq!(control.queues, bound.queues); + assert_eq!(control.bbr.window(), bound.bbr.window()); + } } diff --git a/rs/moq-quic/src/tests/mod.rs b/rs/moq-quic/src/tests/mod.rs index 1b9e2b1c338..c925469c75b 100644 --- a/rs/moq-quic/src/tests/mod.rs +++ b/rs/moq-quic/src/tests/mod.rs @@ -46,6 +46,7 @@ mod acked; mod characterize; mod probe; mod rate; +mod realtime; mod receive_ts; mod reset_stream_at; mod sim; diff --git a/rs/moq-quic/src/tests/probe.rs b/rs/moq-quic/src/tests/probe.rs index 3937ef7cbe6..c0852db8644 100644 --- a/rs/moq-quic/src/tests/probe.rs +++ b/rs/moq-quic/src/tests/probe.rs @@ -12,8 +12,10 @@ use std::collections::HashMap; +use rand::RngExt; + use super::*; -use crate::congestion::{Bbr3, CubicConfig}; +use crate::congestion::{Bbr3, Bbr3Config, CubicConfig}; /// The media's bitrate, 2 Mbit/s, which the probe has to see past const MEDIA: u64 = 250_000; @@ -87,6 +89,8 @@ struct Scenario { ack_threshold: Option, /// Bytes of bulk upload before the media starts, to learn the path's rate first warmup: usize, + /// Whether the media's BBR3 runs the delay bound, reading the receive timestamps it asks for + delay_bound: bool, } impl Default for Scenario { @@ -104,6 +108,7 @@ impl Default for Scenario { receive_window: None, ack_threshold: None, warmup: 0, + delay_bound: false, } } } @@ -267,6 +272,12 @@ fn run(scenario: &Scenario) -> Outcome { }; transport.ack_frequency = Some(ack); } + if scenario.delay_bound { + let mut bbr = Bbr3Config::default(); + bbr.delay_bound(true).probe_rng_seed(entropy::rng().random()); + transport.congestion = Arc::new(bbr); + transport.receive_timestamps = NonZeroU32::new(32); + } let mut config = client_config(); config.transport = transport; let (client_ch, server_ch) = pair.connect_with(config); @@ -680,16 +691,33 @@ fn steps() -> Scenario { /// this 100ms round trip the baseline's estimate stays at the first path's 3 Mbit/s #[test] fn rises_to_target() { + let _guard = subscribe(); + rises(&compare("rises", opening(), GATE), Duration::from_secs(5)); +} + +/// The same under the delay bound: its brake on queue growth leaves BBR3's probing to raise the +/// estimate, so a rung stopped for want of it comes back +#[test] +fn rises_to_target_under_delay_bound() { let _guard = subscribe(); let scenario = Scenario { + delay_bound: true, + ..opening() + }; + rises(&compare("rises delay bound", scenario, GATE), Duration::from_secs(5)); +} + +/// A constant bitrate over a 3 Mbit/s path with a 100ms round trip that opens up to 12 Mbit/s +/// at 5s +fn opening() -> Scenario { + Scenario { latency: Duration::from_millis(50), buffer: 170_000, keyframe: 0, rates: vec![(Duration::ZERO, 375_000), (Duration::from_secs(5), 1_500_000)], duration: Duration::from_secs(15), ..Scenario::default() - }; - rises(&compare("rises", scenario, GATE), Duration::from_secs(5)); + } } /// A path short of the target, behind a queue a second deep: each attempt that misses sits out @@ -862,6 +890,20 @@ fn steps_keyframes() { falls(&trials, Duration::from_secs(40), 375_000); } +/// Under the delay bound, as the keyframed steps without it +#[test] +#[ignore = "nightly: just rs quic-sim"] +fn steps_delay_bound() { + let _guard = subscribe(); + let scenario = Scenario { + delay_bound: true, + ..steps() + }; + let trials = compare("steps delay bound", scenario, SEEDS); + rises(&trials, Duration::from_secs(20)); + falls(&trials, Duration::from_secs(40), 375_000); +} + #[test] #[ignore = "nightly: just rs quic-sim"] fn steps_constant() { diff --git a/rs/moq-quic/src/tests/realtime.rs b/rs/moq-quic/src/tests/realtime.rs new file mode 100644 index 00000000000..9905f5a8827 --- /dev/null +++ b/rs/moq-quic/src/tests/realtime.rs @@ -0,0 +1,384 @@ +//! Realtime uplink congestion control: each candidate controller uploads as fast as it may +//! through sim replicas of moq-shaper's cellular uplink presets, and the bottleneck reports the +//! queue it kept, beside the throughput it reached and how fast it answered the uplink dropping. +//! +//! Each [`Preset`] copies its profile in `rs/moq-shaper/profiles/`: the delay each way, the +//! uplink's rate, queue, and steps, and its jitter and batches, with two changes the sim's +//! [`Link`] needs. The gaussian jitter becomes uniform in-order jitter of the same variance, and +//! a batch, which moq-shaper closes on a count or a window, becomes a scheduler grant releasing +//! whatever arrived within each window at its end. +//! +//! Every candidate runs each preset on the same seeds, alone and beside one CUBIC upload. The +//! merge gate runs one preset on one seed; the `#[ignore]`d [`compare`] table runs nightly with +//! `just rs quic-sim`. + +use std::ops::Range; + +use rand::RngExt; + +use super::*; +use crate::congestion::Bbr3Config; + +/// Packets reaching the bottleneck before this are Startup's, and count for no phase +const WARMUP: Duration = Duration::from_secs(2); +/// When every preset's uplink drops +const DROP: Duration = Duration::from_secs(10); +/// When every preset's uplink comes back +const RESTORE: Duration = Duration::from_secs(20); +/// How long each run goes on +const END: Duration = Duration::from_secs(30); +/// The span each throughput sample covers +const BIN: Duration = Duration::from_millis(100); +/// The most receive timestamps a candidate that reads them asks for per ACK +const TIMESTAMPS: u32 = 32; +/// The seeds the nightly table averages over +const SEEDS: ops::RangeInclusive = 1..=8; + +/// A moq-shaper uplink preset, replicated on the sim +#[derive(Debug, Clone, Copy)] +struct Preset { + name: &'static str, + /// The one-way delay each way + latency: Duration, + /// The uplink's in-order jitter: uniform up to this, with the variance of the preset's + /// gaussian + jitter: Duration, + /// The uplink's scheduler grants, for a preset that batches + grant: Option, + /// The uplink's rate in bytes per second, before [`DROP`] and from [`RESTORE`] + rate: u64, + /// The uplink's rate in bytes per second, from [`DROP`] until [`RESTORE`] + dropped: u64, + /// The longest a datagram waits in the uplink queue before it is dropped + queue: Duration, +} + +/// Uniform jitter with the variance of a gaussian's `sigma`: `sigma` times the root of 12 +const fn uniform(sigma_ms: u64) -> Duration { + Duration::from_micros(sigma_ms * 3464) +} + +/// 2 Mbit/s behind a 500ms queue, down to 500 kbit/s +const STEP: Preset = Preset { + name: "cellular-step", + latency: Duration::from_millis(20), + jitter: uniform(3), + grant: None, + rate: 250_000, + dropped: 62_500, + queue: Duration::from_millis(500), +}; + +/// 2 Mbit/s in 20ms grants behind a 500ms queue, down to 1 Mbit/s +const BURSTY: Preset = Preset { + name: "cellular-bursty", + latency: Duration::from_millis(20), + jitter: uniform(5), + grant: Some(Duration::from_millis(20)), + rate: 250_000, + dropped: 125_000, + queue: Duration::from_millis(500), +}; + +/// 1 Mbit/s behind a 2s queue, down to 300 kbit/s +const DEEP: Preset = Preset { + name: "cellular-deep", + latency: Duration::from_millis(30), + jitter: uniform(5), + grant: None, + rate: 125_000, + dropped: 37_500, + queue: Duration::from_secs(2), +}; + +impl Preset { + /// The bytes the uplink could carry in `range` + fn capacity(&self, range: Range) -> f64 { + let overlap = |from: Duration, until: Duration| { + let (from, until) = (from.max(range.start), until.min(range.end)); + until.saturating_sub(from).as_secs_f64() + }; + let normal = overlap(Duration::ZERO, DROP) + overlap(RESTORE, Duration::MAX); + normal * self.rate as f64 + overlap(DROP, RESTORE) * self.dropped as f64 + } +} + +/// A controller under comparison +#[derive(Clone, Copy)] +struct Candidate { + name: &'static str, + /// The transport its connection runs, with the controller and any receive timestamps it + /// reads + transport: fn() -> transport::Config, +} + +const BBR3: Candidate = Candidate { + name: "BBR3", + transport: transport::Config::default, +}; + +const DELAY_BOUND: Candidate = Candidate { + name: "BBR3 delay bound", + transport: || { + let mut bbr = Bbr3Config::default(); + bbr.delay_bound(true).probe_rng_seed(entropy::rng().random()); + transport::Config { + congestion: Arc::new(bbr), + receive_timestamps: NonZeroU32::new(TIMESTAMPS), + ..transport::Config::default() + } + }, +}; + +/// One bulk upload over the shared uplink +struct Flow { + client: ConnectionHandle, + server: ConnectionHandle, + send: StreamId, + recv: Option, + /// Bytes the receiver read in each [`BIN`] from the start + read: Vec, +} + +/// What one run measured +#[derive(Debug, Default)] +struct Outcome { + /// Each packet's arrival at the bottleneck, since the start, and its queueing delay there + queue: Vec<(Duration, Duration)>, + /// Packets the full queue dropped + dropped: u64, + /// How soon after the drop the candidate first paced at or below the dropped rate, if it did + /// before the uplink came back + reaction: Option, + /// The candidate's bytes the receiver read in each [`BIN`] + read: Vec, + /// The competing CUBIC upload's bytes the receiver read in each [`BIN`], if it ran + competitor: Vec, +} + +/// Uploads with `candidate` over `preset` on `seed`, beside a CUBIC upload if `competing` +fn run(preset: &Preset, candidate: &Candidate, seed: u64, competing: bool) -> Outcome { + entropy::reseed(seed); + let mut server = server_config(); + server.transport.limits.max_streams = NonZeroU64::new(10); + let mut pair = Pair::new(Default::default(), server); + pair.latency = preset.latency; + + let connect = |pair: &mut Pair, transport: transport::Config| { + let mut config = client_config(); + config.transport = transport; + let (client, server) = pair.connect_with(config); + let send = pair.client_streams(client).open(Dir::Uni, Default::default()).unwrap(); + Flow { + client, + server, + send, + recv: None, + read: Vec::new(), + } + }; + let mut flows = vec![connect(&mut pair, (candidate.transport)())]; + if competing { + flows.push(connect(&mut pair, cubic_transport())); + } + + let mut bottleneck = Bottleneck::timed(preset.rate, preset.queue); + bottleneck.trace = Some(Vec::new()); + pair.to_server.bottleneck = Some(bottleneck); + pair.to_server.jitter = preset.jitter; + pair.to_server.grant = preset.grant; + + const CHUNK: [u8; 16_384] = [0; 16_384]; + let start = pair.time; + let mut steps = vec![(RESTORE, preset.rate), (DROP, preset.dropped)]; + let mut reaction = None; + loop { + let now = pair.time; + if let Some(&(at, rate)) = steps.last() + && now >= start + at + { + pair.to_server.bottleneck.as_mut().unwrap().set_rate(rate); + steps.pop(); + } + for flow in &flows { + while pair.client_send(flow.client, flow.send).write(&mut &CHUNK[..]).is_ok() {} + } + + pair.drive_client(); + pair.drive_server(); + + let elapsed = now - start; + if reaction.is_none() && (DROP..RESTORE).contains(&elapsed) { + let metrics = pair.client_conn_mut(flows[0].client).congestion_state().metrics(); + if metrics.pacing_rate.is_some_and(|rate| rate <= preset.dropped) { + reaction = Some(elapsed - DROP); + } + } + + let bin = (elapsed.as_nanos() / BIN.as_nanos()) as usize; + for flow in &mut flows { + if flow.recv.is_none() { + flow.recv = pair.server_streams(flow.server).accept(Dir::Uni); + } + let Some(id) = flow.recv else { continue }; + if flow.read.len() <= bin { + flow.read.resize(bin + 1, 0); + } + let mut stream = pair.server_recv(flow.server, id); + let mut chunks = stream.read().unwrap(); + while let Ok(Some(chunk)) = chunks.next(usize::MAX) { + flow.read[bin] += chunk.bytes.len() as u64; + } + let _ = chunks.finalize(); + } + + if now >= start + END { + break; + } + let wakeups = [ + pair.client.next_wakeup(), + pair.server.next_wakeup(), + steps.last().map(|&(at, _)| start + at), + ]; + pair.time = pair.time.max(wakeups.into_iter().flatten().min().unwrap()); + } + + let bottleneck = pair.to_server.bottleneck.take().unwrap(); + let mut flows = flows.into_iter(); + Outcome { + queue: bottleneck + .trace + .unwrap() + .into_iter() + .map(|(at, delay)| (at.saturating_duration_since(start), delay)) + .collect(), + dropped: bottleneck.stats.dropped, + reaction, + read: flows.next().unwrap().read, + competitor: flows.next().map(|flow| flow.read).unwrap_or_default(), + } +} + +impl Outcome { + /// The queueing delays of packets reaching the bottleneck in `range`, sorted + fn delays(&self, range: Range) -> Vec { + let mut delays: Vec = self + .queue + .iter() + .filter(|(at, _)| range.contains(at)) + .map(|&(_, delay)| delay) + .collect(); + delays.sort_unstable(); + delays + } + + /// The `p` quantile of the queueing delay in `range` + fn queue(&self, range: Range, p: f64) -> Duration { + let delays = self.delays(range); + delays[((delays.len() - 1) as f64 * p) as usize] + } + + /// The worst queueing delay a packet met while the uplink was down, and how soon after the + /// drop it met it + fn peak(&self) -> (Duration, Duration) { + self.queue.iter().filter(|(at, _)| (DROP..RESTORE).contains(at)).fold( + (Duration::ZERO, Duration::ZERO), + |peak, &(at, delay)| match delay > peak.0 { + true => (delay, at - DROP), + false => peak, + }, + ) + } + + /// How soon after the drop the queue fell back under a quarter of its peak, if it did + /// before the uplink came back + fn drained(&self) -> Option { + let (peak, reached) = self.peak(); + self.queue + .iter() + .find(|(at, delay)| *at > DROP + reached && *at < RESTORE && *delay < peak / 4) + .map(|(at, _)| *at - DROP) + } + + /// Bytes `read` in `range`, to the [`BIN`] + fn bytes(read: &[u64], range: Range) -> f64 { + let bins = (range.start.as_nanos() / BIN.as_nanos()) as usize..(range.end.as_nanos() / BIN.as_nanos()) as usize; + read.get(bins).unwrap_or_default().iter().sum::() as f64 + } + + /// The candidate's goodput in `range`, as a share of what the uplink could carry + fn utilization(&self, preset: &Preset, range: Range) -> f64 { + Self::bytes(&self.read, range.clone()) / preset.capacity(range) + } + + /// The candidate's share of the goodput beside the CUBIC upload, after Startup + fn share(&self) -> f64 { + let ours = Self::bytes(&self.read, WARMUP..END); + ours / (ours + Self::bytes(&self.competitor, WARMUP..END)) + } +} + +/// On the gate's seed, the delay bound keeps a shorter queue than BBR3 over cellular-step, whose +/// drop BBR3 answers only once its estimate ages +#[test] +fn delay_bound_keeps_a_shorter_queue() { + let _guard = subscribe(); + let bbr3 = run(&STEP, &BBR3, 1, false); + let bound = run(&STEP, &DELAY_BOUND, 1, false); + let (bbr3, bound) = (bbr3.queue(WARMUP..END, 0.95), bound.queue(WARMUP..END, 0.95)); + assert!( + bound < bbr3, + "p95 queue {bound:?} with the delay bound, {bbr3:?} without" + ); +} + +/// The comparison's table, over every preset, candidate, and [`SEEDS`], alone and beside CUBIC +#[test] +#[ignore = "the realtime comparison, run nightly by `just rs quic-sim`"] +fn compare() { + let _guard = subscribe_at(tracing::Level::WARN); + println!( + "| preset | controller | queue p50 | p95 | p99 | p99 before the drop | utilization | after the drop | peak | reaction | drained | dropped | share beside CUBIC |" + ); + println!("|---|---|---|---|---|---|---|---|---|---|---|---|---|"); + for preset in [STEP, BURSTY, DEEP] { + for candidate in [BBR3, DELAY_BOUND] { + let alone: Vec = SEEDS.map(|seed| run(&preset, &candidate, seed, false)).collect(); + let shared: Vec = SEEDS.map(|seed| run(&preset, &candidate, seed, true)).collect(); + let ms = |f: &dyn Fn(&Outcome) -> Duration| avg(alone.iter().map(|o| f(o).as_secs_f64() * 1e3)); + println!( + "| {} | {} | {:.0}ms | {:.0}ms | {:.0}ms | {:.0}ms | {:.0}% | {:.0}% | {:.0}ms | {} | {} | {:.0} | {:.0}% |", + preset.name, + candidate.name, + ms(&|o| o.queue(WARMUP..END, 0.5)), + ms(&|o| o.queue(WARMUP..END, 0.95)), + ms(&|o| o.queue(WARMUP..END, 0.99)), + ms(&|o| o.queue(WARMUP..DROP, 0.99)), + avg(alone.iter().map(|o| o.utilization(&preset, WARMUP..END))) * 100.0, + avg(alone.iter().map(|o| o.utilization(&preset, DROP..RESTORE))) * 100.0, + ms(&|o| o.peak().0), + latency(alone.iter().map(|o| o.reaction)), + latency(alone.iter().map(Outcome::drained)), + avg(alone.iter().map(|o| o.dropped as f64)), + avg(shared.iter().map(Outcome::share)) * 100.0, + ); + } + } +} + +fn avg(values: impl Iterator) -> f64 { + let values: Vec = values.collect(); + values.iter().sum::() / values.len() as f64 +} + +/// The mean over the seeds that got there, and how many never did +fn latency(values: impl Iterator>) -> String { + let values: Vec> = values.collect(); + let hits: Vec = values.iter().flatten().map(|d| d.as_secs_f64() * 1e3).collect(); + let never = values.len() - hits.len(); + match hits.is_empty() { + true => "never".to_string(), + false if never == 0 => format!("{:.0}ms", avg(hits.into_iter())), + false => format!("{:.0}ms, never {never}", avg(hits.into_iter())), + } +} diff --git a/rs/moq-quic/src/tests/util.rs b/rs/moq-quic/src/tests/util.rs index 5af6293da71..5819872022f 100644 --- a/rs/moq-quic/src/tests/util.rs +++ b/rs/moq-quic/src/tests/util.rs @@ -315,9 +315,14 @@ pub(super) struct Bottleneck { max_queue: Duration, /// Whether packets are marked CE once the queue is half full, as a classic AQM does marks_ce: bool, + /// Whether the queue is bounded in time rather than bytes, as moq-shaper's is, so a rate + /// change keeps `max_queue` + timed: bool, /// When the link finishes serving everything queued so far busy_until: Option, pub(super) stats: QueueStats, + /// Each queued packet's arrival and queueing delay, once set to `Some` + pub(super) trace: Option>, } impl Bottleneck { @@ -329,18 +334,32 @@ impl Bottleneck { buffer_size, max_queue: buffer_size * time_per_byte, marks_ce, + timed: false, busy_until: None, stats: QueueStats::default(), + trace: None, } } - /// Serve at `bytes_per_second` from now on, keeping the queue's size in bytes + /// A link serving `bytes_per_second`, dropping what would wait in its queue past `max_queue` + /// at whatever rate it serves, as a moq-shaper rate limit does + pub(super) fn timed(bytes_per_second: u64, max_queue: Duration) -> Self { + let mut bottleneck = Self::new(bytes_per_second, 0, false); + bottleneck.max_queue = max_queue; + bottleneck.timed = true; + bottleneck + } + + /// Serve at `bytes_per_second` from now on, keeping the queue's size in bytes, or in time + /// for a [`Self::timed`] queue /// /// Packets already queued still leave when they were due, as if the link stepped once they /// had been served. pub(super) fn set_rate(&mut self, bytes_per_second: u64) { self.time_per_byte = Duration::from_nanos(1_000_000_000 / bytes_per_second); - self.max_queue = self.buffer_size * self.time_per_byte; + if !self.timed { + self.max_queue = self.buffer_size * self.time_per_byte; + } } /// Queues `bytes` arriving at `now`, returning when they leave the link and whether they @@ -353,6 +372,9 @@ impl Bottleneck { return None; } let service = bytes as u32 * self.time_per_byte; + if let Some(trace) = &mut self.trace { + trace.push((now, delay)); + } self.stats.queued += 1; self.stats.total_delay += delay; self.stats.max_delay = self.stats.max_delay.max(delay); @@ -385,8 +407,13 @@ pub(super) struct Link { pub(super) reorder: f64, /// The most a reordered arrival is held back pub(super) reorder_delay: Duration, + /// Releases in-order arrivals only at the end of each interval this long, as a cellular + /// uplink's scheduler grants them, bunching what arrived within it + pub(super) grant: Option, /// The latest in-order arrival, which jitter never moves another before last_arrival: Option, + /// When the first grant interval began + grant_epoch: Option, } impl Link { @@ -412,12 +439,23 @@ impl Link { return arrival + rng.random_range(Duration::ZERO..=self.reorder_delay); } let arrival = self.last_arrival.map_or(arrival, |last| last.max(arrival)); + let arrival = self.granted(arrival); self.last_arrival = Some(arrival); arrival }) .collect(); (marked, arrivals) } + + /// When the grant covering `arrival` releases it: the end of its interval + fn granted(&mut self, arrival: Instant) -> Instant { + let Some(grant) = self.grant else { + return arrival; + }; + let epoch = *self.grant_epoch.get_or_insert(arrival); + let intervals = (arrival - epoch).as_nanos().div_ceil(grant.as_nanos()); + epoch + Duration::from_nanos((intervals * grant.as_nanos()) as u64) + } } /// What a [`Bottleneck`] queue has seen From 4436f6e216399cacfe3fb8af8d7b47ae2935fede Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 14:03:49 -0800 Subject: [PATCH 3/7] quest: record the BBR3 delay bound decisions in the realtime epic The maintainer settled the three open decisions on 2026-11-01: the brake stays a separate bw_delay bound, its detector stays inside bbr3/, and the shared comparison runs every candidate over real moq-shaper sockets once. Co-Authored-By: Claude Opus 5.5 --- quest/m2/quic-realtime/README.md | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/quest/m2/quic-realtime/README.md b/quest/m2/quic-realtime/README.md index e5e60af9aed..adcc48f2cbc 100644 --- a/quest/m2/quic-realtime/README.md +++ b/quest/m2/quic-realtime/README.md @@ -43,6 +43,14 @@ spike assumed no consumer; teleoperation uplinks are one): first, with the harness the others join: moq-quic's `tests::realtime` replays the three presets on the sim, and each candidate adds a `Candidate` there (`just rs quic-sim` prints the table). +- BBR3 delay bound decisions, ✅ 2026-11-01 (maintainer, on + [moq-dev/moq#7544](https://github.com/moq-dev/moq/pull/7544)): the brake + is a separate `bw_delay` bound beside `max_bw` and `bw_shortterm`, which + REFILL doesn't reset and which Startup and ProbeBW_UP raise. Its GoogCC-style + trend detector stays inside `moq-congestion`'s `bbr3/`, so dropping the + candidate deletes it, rather than sharing the GCC candidate's trendline. + The candidate reported on the sim replicas only; real sockets are the + shared comparison's job. - Native probing is required (decided 2026-10-25, maintainer). Every candidate implements `moq_congestion::Controller::probes` and reports `probe_inflight`, so moq-quic's native probing @@ -65,6 +73,10 @@ Remaining work once the candidates finish (this README's own job): - Run the shared comparison on the uplink presets: queuing delay p50/p95/p99, throughput, reaction time to a capacity step down, and share against one competing CUBIC flow. Pick the winner on queuing delay. + It runs every candidate once over real moq-shaper sockets, measuring the + `near-zero` noise floor too, so a candidate's own report may stay on the + sim replicas (✅ 2026-11-01, maintainer, on + [moq-dev/moq#7544](https://github.com/moq-dev/moq/pull/7544)). - Run every candidate under both sender models in one harness: a bulk upload, and an encoder that follows `bandwidth_estimate` (the model moq-cli and moq-ffi use, and the one GoogCC is built for). Decided 2026-11-01 From 5aeeac554d30b923a8244c5e422c3b2320cda0ac Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 14:12:30 -0800 Subject: [PATCH 4/7] fix(congestion): feed the BBR3 delay detector each ACK in arrival order Fed in send order, a packet reordered within an ACK showed a negative arrival delta, which the detector discards, while the rebound after it counted as delay. Alternating 20ms and 50ms delays accumulated 30ms of false queue growth every two packets. GoogCC's DelayBasedBwe sorts each feedback by receive time, so a packet sent before the group in progress is skipped instead; the detector now does the same. Co-Authored-By: Claude Opus 5.5 --- rs/moq-congestion/src/bbr3/delay.rs | 75 +++++++++++++++++++++++++++-- rs/moq-congestion/src/bbr3/mod.rs | 8 ++- 2 files changed, 75 insertions(+), 8 deletions(-) diff --git a/rs/moq-congestion/src/bbr3/delay.rs b/rs/moq-congestion/src/bbr3/delay.rs index 22802484856..5b4051d3564 100644 --- a/rs/moq-congestion/src/bbr3/delay.rs +++ b/rs/moq-congestion/src/bbr3/delay.rs @@ -18,7 +18,7 @@ use std::collections::VecDeque; -use crate::{Duration, Instant}; +use crate::{Acked, Duration, Instant}; /// Packets sent within this of a group's first belong to it. const GROUP_LENGTH: Duration = Duration::from_millis(5); @@ -106,6 +106,8 @@ pub(super) struct Detector { usage: Usage, /// Whether the trend has asserted growth since [`Self::take_signal`] last ran. signaled: bool, + /// One ACK's arrivals and send times, sorted by arrival, kept to reuse its allocation. + batch: Vec<(Duration, Instant)>, } impl Default for Detector { @@ -126,6 +128,7 @@ impl Default for Detector { threshold_updated: None, usage: Usage::Normal, signaled: false, + batch: Vec::new(), } } } @@ -147,11 +150,32 @@ impl Detector { std::mem::take(&mut self.signaled) } + /// The packets an ACK arriving at `acked` newly acknowledged, skipping any without a receive + /// timestamp. + /// + /// They are fed in the order they arrived, as `DelayBasedBwe` sorts each feedback by + /// receive time. Fed in send order, a reordered packet would show a negative arrival delta, + /// which is discarded, while the rebound after it counts as delay, so reordering would + /// accumulate into false queue growth. In arrival order, a packet sent before the group in + /// progress is skipped instead. + pub(super) fn on_ack(&mut self, packets: &[Acked], acked: Instant) { + // Taken, since a reset in `on_packet` replaces the whole detector. + let mut batch = std::mem::take(&mut self.batch); + batch.extend(packets.iter().filter_map(|p| Some((p.received?, p.sent)))); + // Stable, so arrivals a grant released together stay in send order. + batch.sort_by_key(|&(received, _)| received); + for &(received, sent) in &batch { + self.on_packet(sent, received, acked); + } + batch.clear(); + self.batch = batch; + } + /// A packet sent at `sent` arrived at `received` on the peer's clock, as an ACK arriving at - /// `acked` reported. Packets are fed in the order they were sent within each ACK. + /// `acked` reported. Packets are fed in the order they arrived. /// /// equivalent to `InterArrivalDelta::ComputeDeltas`, then `TrendlineEstimator::Update` - pub(super) fn on_packet(&mut self, sent: Instant, received: Duration, acked: Instant) { + fn on_packet(&mut self, sent: Instant, received: Duration, acked: Instant) { let Some(mut current) = self.current else { self.current = Some(Group { first_send: sent, @@ -434,6 +458,51 @@ mod test { assert!(!usage.contains(&Usage::Overusing)); } + /// Packets reordered within an ACK, with no queue behind them, never read as growth. Fed in + /// send order, each reordered group's negative delta was discarded while the rebound after + /// it counted, adding 30ms of delay every two packets. + #[test] + fn reordered_within_an_ack_is_not_growth() { + let mut detector = Detector::default(); + let base = Instant::now(); + let spacing = Duration::from_millis(10); + let interval = Duration::from_millis(40); + // Every other packet is 30ms slower than its neighbors + let packets: Vec = (0..1000u32) + .map(|i| { + let sent = base + spacing * i; + let delay = Duration::from_millis(if i % 2 == 0 { 20 } else { 50 }); + Acked { + sent, + bytes: 1200, + packet: Default::default(), + received: Some(sent - base + delay), + } + }) + .collect(); + let mut usage = Vec::new(); + let mut signaled = false; + // Each ACK reports, in send order, whatever arrived in the 40ms before it + for ack in 1..250u32 { + let (from, to) = (interval * (ack - 1), interval * ack); + let acked: Vec = packets + .iter() + .filter(|p| p.received.is_some_and(|r| r >= from && r < to)) + .copied() + .collect(); + detector.on_ack(&acked, base + to + Duration::from_millis(20)); + usage.push(detector.usage()); + signaled |= detector.take_signal(); + } + assert!(!usage.contains(&Usage::Overusing)); + assert!(!signaled); + assert!( + detector.accumulated.abs() < 50.0, + "accumulated {}ms", + detector.accumulated + ); + } + /// Packets sent before the group in progress, from a late ACK, are skipped #[test] fn late_packets_are_skipped() { diff --git a/rs/moq-congestion/src/bbr3/mod.rs b/rs/moq-congestion/src/bbr3/mod.rs index bfac766bf4d..3f4767c5a31 100644 --- a/rs/moq-congestion/src/bbr3/mod.rs +++ b/rs/moq-congestion/src/bbr3/mod.rs @@ -1920,11 +1920,9 @@ impl Controller for Bbr3 { for acked in ack.packets { self.check_recovery_done(acked.sent); newly_acked += u64::from(acked.bytes); - if let Some(delay) = &mut self.delay - && let Some(received) = acked.received - { - delay.on_packet(acked.sent, received, ack.now); - } + } + if let Some(delay) = &mut self.delay { + delay.on_ack(ack.packets, ack.now); } self.inflight = ack.in_flight; // An ACK that delivered no packet has no sample, and a finished one is never folded From b36d5dcbd3f6ea216c202f3cc0f000c4802ecb6c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 14:32:12 -0800 Subject: [PATCH 5/7] fix(congestion): keep a zero-rate round from bounding BBR3's delay bound Queue growth read on a round that measured no rate set bw_delay to zero, pinning BBR.bw and the pacing rate there until a probe lifted it. Skip it, as update_max_bw skips a zero-rate sample. Co-Authored-By: Claude Opus 5.5 --- rs/moq-congestion/src/bbr3/mod.rs | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/rs/moq-congestion/src/bbr3/mod.rs b/rs/moq-congestion/src/bbr3/mod.rs index 8e914bc7eb6..4c470a119b3 100644 --- a/rs/moq-congestion/src/bbr3/mod.rs +++ b/rs/moq-congestion/src/bbr3/mod.rs @@ -1811,6 +1811,10 @@ impl Bbr3 { /// /// Startup, Drain, and ProbeRTT are BBR3's own: Startup ends on its plateau or on loss, and /// Drain and ProbeRTT already send below the estimate. + /// + /// A round that has measured no rate, as a zero-interval sample leaves it, bounds nothing, + /// as `update_max_bw` skips that sample: a zero bound would stop pacing until a probe the + /// path carries raised it. fn bound_queue_growth(&mut self, now: Instant) { let Some(rs) = self.rs else { return; @@ -1819,6 +1823,9 @@ impl Bbr3 { return; } let delivery_rate = self.bw_latest.max(rs.delivery_rate); + if delivery_rate <= 0.0 { + return; + } match self.state { BbrState::ProbeBw(ProbeBwSubstate::Up) => { self.bw_delay = self.bw_delay.min(delivery_rate); @@ -8324,6 +8331,26 @@ mod test { } } + /// Queue growth read on a round that measured no rate leaves the delay bound alone, rather + /// than bounding BBR.bw, and with it pacing, at zero. + #[test] + fn delay_bound_ignores_a_zero_rate() { + let mut sim = scripted(); + sim.send(0, 1); + sim.ack(10 * MS, [0], false); + sim.send(10 * MS, 1); + // A zero interval, which carries no bandwidth + sim.ack(10 * MS, [1], false); + assert_eq!(sim.bbr.rs.unwrap().delivery_rate, 0.0); + + sim.bbr.state = BbrState::ProbeBw(ProbeBwSubstate::Down); + sim.bbr.bw_latest = 0.0; + let now = sim.at(10 * MS); + sim.bbr.bound_queue_growth(now); + assert_eq!(sim.bbr.bw_delay, f64::INFINITY); + assert_eq!(sim.bbr.delay_round_delivered, 0); + } + /// The delay bound reads only receive timestamps: without them it changes nothing #[test] fn delay_bound_without_timestamps_is_bbr3() { From 597b60ed5a0d057ffd71dcf840ff834db37bd66a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 1 Nov 2026 14:51:44 -0800 Subject: [PATCH 6/7] fix(congestion): report BBR3's delay bound in its bandwidth estimate With the delay bound engaged, `Controller::metrics` still reported `max_bw`, so bandwidth followers such as an encoder kept the old rate while BBR3 paced under `bw_delay`, and the bandwidth probe judged the target reached. Report the lower of the two; `bw_delay` is infinite unless the delay bound is on, so stock BBR3 is unchanged. Co-Authored-By: Claude Opus 5.5 --- rs/moq-congestion/src/bbr3/mod.rs | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/rs/moq-congestion/src/bbr3/mod.rs b/rs/moq-congestion/src/bbr3/mod.rs index 4c470a119b3..d04b6151db6 100644 --- a/rs/moq-congestion/src/bbr3/mod.rs +++ b/rs/moq-congestion/src/bbr3/mod.rs @@ -2044,11 +2044,14 @@ impl Controller for Bbr3 { } fn metrics(&self) -> Metrics { + // The delay bound's `bw_delay` caps the estimate: it marks the path slower than `max_bw`, + // so a bandwidth follower, such as an encoder, should follow it. Infinite unless on. + let estimate = self.max_bw.min(self.bw_delay); Metrics { congestion_window: self.window(), ssthresh: None, pacing_rate: Some(self.pacing_rate.round() as u64), - bandwidth_estimate: (self.max_bw > 0.0).then_some(self.max_bw.round() as u64), + bandwidth_estimate: (estimate > 0.0).then_some(estimate.round() as u64), probe_inflight: self.probe_inflight(), } } @@ -8351,6 +8354,19 @@ mod test { assert_eq!(sim.bbr.delay_round_delivered, 0); } + /// The bandwidth estimate reports the delay bound when it is below `max_bw`, so a bandwidth + /// follower slows with the brake + #[test] + fn delay_bound_caps_the_bandwidth_estimate() { + let mut sim = scripted(); + sim.bbr.max_bw = 1_000_000.0; + assert_eq!(sim.bbr.metrics().bandwidth_estimate, Some(1_000_000)); + sim.bbr.bw_delay = 400_000.0; + assert_eq!(sim.bbr.metrics().bandwidth_estimate, Some(400_000)); + sim.bbr.bw_delay = 2_000_000.0; + assert_eq!(sim.bbr.metrics().bandwidth_estimate, Some(1_000_000)); + } + /// The delay bound reads only receive timestamps: without them it changes nothing #[test] fn delay_bound_without_timestamps_is_bbr3() { From 0d2f1de14704555ebde378d9cba5181d6092b541 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 22:47:10 -0800 Subject: [PATCH 7/7] fix(congestion): build the delay bound against main's test harness Main's subscribe() replaced subscribe_at, and moq-congestion's Controller trait is in scope in the probe tests, so the merged scenario enum is Congestion. The delay detector's two time constants are Durations, as the repository's typed units rule asks (CodeRabbit). Co-Authored-By: Claude Opus 5.5 --- rs/moq-congestion/src/bbr3/delay.rs | 12 ++++++------ rs/moq-quic/src/tests/probe.rs | 20 ++++++++++---------- rs/moq-quic/src/tests/realtime.rs | 2 +- 3 files changed, 17 insertions(+), 17 deletions(-) diff --git a/rs/moq-congestion/src/bbr3/delay.rs b/rs/moq-congestion/src/bbr3/delay.rs index 5b4051d3564..5ad2a03da8e 100644 --- a/rs/moq-congestion/src/bbr3/delay.rs +++ b/rs/moq-congestion/src/bbr3/delay.rs @@ -40,8 +40,8 @@ const SMOOTHING: f64 = 0.9; const THRESHOLD_GAIN: f64 = 4.0; /// The most groups the trend's scale counts. const MIN_DELTAS: u32 = 60; -/// How long, in milliseconds, the trend must stay over the threshold to count as queue growth. -const OVERUSE_TIME: f64 = 10.0; +/// How long the trend must stay over the threshold to count as queue growth. +const OVERUSE_TIME: Duration = Duration::from_millis(10); /// A trend this far past the threshold is a spike, such as a capacity drop, that the threshold /// does not adapt to, in milliseconds. const MAX_ADAPT_OFFSET: f64 = 15.0; @@ -52,8 +52,8 @@ const K_DOWN: f64 = 0.039; const THRESHOLD_INITIAL: f64 = 12.5; const THRESHOLD_MIN: f64 = 6.0; const THRESHOLD_MAX: f64 = 600.0; -/// The most time, in milliseconds, one threshold update accounts for. -const THRESHOLD_MAX_STEP: f64 = 100.0; +/// The most time one threshold update accounts for. +const THRESHOLD_MAX_STEP: Duration = Duration::from_millis(100); /// What the forward delay's trend says of the bottleneck queue #[derive(Debug, Clone, Copy, Default, Eq, PartialEq)] @@ -292,7 +292,7 @@ impl Detector { .map_or(send_delta / 2.0, |over| over + send_delta); self.overusing = Some(over); self.overuse_count += 1; - if over > OVERUSE_TIME && self.overuse_count > 1 && trend >= self.trend { + if over > millis(OVERUSE_TIME) && self.overuse_count > 1 && trend >= self.trend { self.overusing = Some(0.0); self.overuse_count = 0; self.usage = Usage::Overusing; @@ -323,7 +323,7 @@ impl Detector { } else { K_UP }; - let elapsed = (arrival - last).clamp(0.0, THRESHOLD_MAX_STEP); + let elapsed = (arrival - last).clamp(0.0, millis(THRESHOLD_MAX_STEP)); self.threshold += k * (modified.abs() - self.threshold) * elapsed; self.threshold = self.threshold.clamp(THRESHOLD_MIN, THRESHOLD_MAX); } diff --git a/rs/moq-quic/src/tests/probe.rs b/rs/moq-quic/src/tests/probe.rs index 6052f735636..84cf0cf9f0d 100644 --- a/rs/moq-quic/src/tests/probe.rs +++ b/rs/moq-quic/src/tests/probe.rs @@ -69,7 +69,7 @@ fn frame_size(keyframe: u64, index: u64) -> u64 { /// The congestion controller the media's connection runs #[derive(Clone, Copy)] -enum Controller { +enum Congestion { Bbr3, /// BBR3 with the delay bound, reading the receive timestamps it asks for DelayBound, @@ -104,7 +104,7 @@ struct Scenario { /// Bytes of bulk upload before the media starts, to learn the path's rate first warmup: usize, /// The media's congestion controller - controller: Controller, + congestion: Congestion, } impl Default for Scenario { @@ -122,7 +122,7 @@ impl Default for Scenario { receive_window: None, ack_threshold: None, warmup: 0, - controller: Controller::Bbr3, + congestion: Congestion::Bbr3, } } } @@ -280,15 +280,15 @@ fn run(scenario: &Scenario) -> Outcome { }; transport.ack_frequency = Some(ack); } - match scenario.controller { - Controller::Bbr3 => {} - Controller::DelayBound => { + match scenario.congestion { + Congestion::Bbr3 => {} + Congestion::DelayBound => { let mut bbr = bbr3::Config::default(); bbr.delay_bound(true).probe_rng_seed(entropy::rng().random()); transport.congestion = Arc::new(bbr); transport.receive_timestamps = NonZeroU32::new(32); } - Controller::Gcc => { + Congestion::Gcc => { transport.congestion = Arc::new(gcc::Config::default()); transport.receive_timestamps = NonZeroU32::new(32); } @@ -716,7 +716,7 @@ fn rises_to_target() { fn rises_to_target_under_delay_bound() { let _guard = subscribe(); let scenario = Scenario { - controller: Controller::DelayBound, + congestion: Congestion::DelayBound, ..opening() }; rises(&compare("rises delay bound", scenario, GATE), Duration::from_secs(5)); @@ -737,7 +737,7 @@ fn rises_to_target_under_gcc() { let scenario = Scenario { duration: Duration::from_secs(20), target, - controller: Controller::Gcc, + congestion: Congestion::Gcc, ..opening() }; let trials = compare(&format!("rises gcc {target}"), scenario, SEEDS); @@ -1043,7 +1043,7 @@ fn steps_keyframes() { fn steps_delay_bound() { let _guard = subscribe(); let scenario = Scenario { - controller: Controller::DelayBound, + congestion: Congestion::DelayBound, ..steps() }; let trials = compare("steps delay bound", scenario, SEEDS); diff --git a/rs/moq-quic/src/tests/realtime.rs b/rs/moq-quic/src/tests/realtime.rs index 829cc6d1bc3..68d5cec8107 100644 --- a/rs/moq-quic/src/tests/realtime.rs +++ b/rs/moq-quic/src/tests/realtime.rs @@ -336,7 +336,7 @@ fn delay_bound_keeps_a_shorter_queue() { #[test] #[ignore = "the realtime comparison, run nightly by `just rs quic-sim`"] fn compare() { - let _guard = subscribe_at(tracing::Level::WARN); + let _guard = subscribe(); println!( "| preset | controller | queue p50 | p95 | p99 | p99 before the drop | utilization | after the drop | peak | reaction | drained | dropped | share beside CUBIC |" );