diff --git a/quest/m2/quic-realtime/README.md b/quest/m2/quic-realtime/README.md index 27f22f94d06..43d084d0511 100644 --- a/quest/m2/quic-realtime/README.md +++ b/quest/m2/quic-realtime/README.md @@ -39,7 +39,18 @@ 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 (`bbr3::Config::delay_bound`) brought + the harness: moq-quic's `tests::realtime` replays the three presets on the + sim, and the shared comparison adds the other candidates to it as + `Candidate`s (`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 @@ -84,6 +95,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 @@ -127,7 +142,6 @@ owns that feedback. ## Required -- [BBR3 delay bound](/quest/m2/quic-realtime/bbr-delay-bound.md) - BBR3 with inflight capped when the forward delay gradient shows queue growth - [io_uring runs an application's controller](/quest/m2/quic-realtime/uring-controller.md) - moq-uring takes a controller factory, so the winner installs the same way on both runtimes - [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 2a7eb5c75df..0419da5a6b6 100644 --- a/rs/justfile +++ b/rs/justfile @@ -347,6 +347,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 013502f48c5..6711872dc3f 100644 --- a/rs/moq-congestion/README.md +++ b/rs/moq-congestion/README.md @@ -2,6 +2,8 @@ BBR3, Cubic, GCC, SCReAMv2, and Unlimited congestion controllers, consumed by moq-quic and the native transport drivers. +BBR3's optional delay bound (`bbr3::Config::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. Construct controllers with `Factory::build` on an `Arc`, `Arc`, `Arc`, `Arc`, or `Arc`. `Bbr3`, `Cubic`, `Gcc`, `Scream`, `Unlimited`, and `Metrics` live at the crate root. diff --git a/rs/moq-congestion/src/bbr3/delay.rs b/rs/moq-congestion/src/bbr3/delay.rs new file mode 100644 index 00000000000..5ad2a03da8e --- /dev/null +++ b/rs/moq-congestion/src/bbr3/delay.rs @@ -0,0 +1,523 @@ +//! 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::{Acked, 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 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; +/// 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 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)] +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, + /// One ACK's arrivals and send times, sorted by arrival, kept to reuse its allocation. + batch: Vec<(Duration, Instant)>, +} + +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, + batch: Vec::new(), + } + } +} + +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) + } + + /// 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 arrived. + /// + /// equivalent to `InterArrivalDelta::ComputeDeltas`, then `TrendlineEstimator::Update` + 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 > millis(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, millis(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 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() { + 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 d725af1f02f..56c4e85ba4f 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; #[cfg(test)] @@ -88,6 +89,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 @@ -527,6 +533,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 + /// [`Config::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); @@ -1342,9 +1367,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); @@ -1817,6 +1844,63 @@ 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. + /// + /// 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; + }; + if rs.prior_delivered < self.delay_round_delivered { + 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); + 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 { @@ -1876,6 +1960,9 @@ impl Controller for Bbr3 { self.check_recovery_done(acked.sent); newly_acked += u64::from(acked.bytes); } + 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 // twice. @@ -1889,6 +1976,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); } @@ -2003,11 +2096,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(), } } @@ -2042,9 +2138,27 @@ pub struct Config { drain_pacing_gain: Option, pacing_margin_percent: Option, default_cwnd_gain: Option, + delay_bound: bool, } impl Config { + /// 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))`. @@ -2078,6 +2192,7 @@ impl Default for Config { drain_pacing_gain: None, pacing_margin_percent: None, default_cwnd_gain: None, + delay_bound: false, } } } @@ -2143,13 +2258,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, }); } } @@ -2212,6 +2339,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, } @@ -2245,6 +2374,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 { @@ -2264,6 +2397,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(), } } @@ -2283,6 +2418,7 @@ mod test { pn: self.pn, send_ns: now_ns, ack_ns: now_ns, + recv_ns: now_ns, ce: false, }); self.pn += 1; @@ -2389,6 +2525,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( @@ -2401,6 +2538,7 @@ mod test { pn: self.pn, send_ns, ack_ns, + recv_ns: finish, ce, }); @@ -2417,7 +2555,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 { @@ -2466,6 +2606,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(); @@ -8202,6 +8343,114 @@ mod test { 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 = Config { + probe_rng_seed: Some([7; 16]), + ..Config::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" + ); + } + } + + /// 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 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() { + 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()); + } + /// Each idle ACK can pass the old application-limited marker while reporting that the /// sender is still idle. The next keepalive must keep its application-limited label. #[test] diff --git a/rs/moq-quic/src/tests/mod.rs b/rs/moq-quic/src/tests/mod.rs index c3e94f9592b..3af2c95fd8d 100644 --- a/rs/moq-quic/src/tests/mod.rs +++ b/rs/moq-quic/src/tests/mod.rs @@ -45,6 +45,7 @@ mod acked; mod characterize; mod probe; mod rate; +mod realtime; mod receive_ts; mod reset_stream_at; mod scream; diff --git a/rs/moq-quic/src/tests/probe.rs b/rs/moq-quic/src/tests/probe.rs index c785f1fc29c..84cf0cf9f0d 100644 --- a/rs/moq-quic/src/tests/probe.rs +++ b/rs/moq-quic/src/tests/probe.rs @@ -18,7 +18,7 @@ use rand::{RngExt, SeedableRng}; use rand_pcg::Pcg32; use super::*; -use crate::congestion::{cubic, gcc}; +use crate::congestion::{bbr3, cubic, gcc}; /// The media's bitrate, 2 Mbit/s, which the probe has to see past const MEDIA: u64 = 250_000; @@ -67,6 +67,16 @@ fn frame_size(keyframe: u64, index: u64) -> u64 { } } +/// The congestion controller the media's connection runs +#[derive(Clone, Copy)] +enum Congestion { + Bbr3, + /// BBR3 with the delay bound, reading the receive timestamps it asks for + DelayBound, + /// GCC, reading the receive timestamps it asks for + Gcc, +} + /// One run's network and workload #[derive(Clone)] struct Scenario { @@ -93,8 +103,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 runs GCC, reading the receive timestamps it asks for, rather than BBR3 - gcc: bool, + /// The media's congestion controller + congestion: Congestion, } impl Default for Scenario { @@ -112,7 +122,7 @@ impl Default for Scenario { receive_window: None, ack_threshold: None, warmup: 0, - gcc: false, + congestion: Congestion::Bbr3, } } } @@ -270,9 +280,18 @@ fn run(scenario: &Scenario) -> Outcome { }; transport.ack_frequency = Some(ack); } - if scenario.gcc { - transport.congestion = Arc::new(gcc::Config::default()); - transport.receive_timestamps = NonZeroU32::new(32); + 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); + } + Congestion::Gcc => { + transport.congestion = Arc::new(gcc::Config::default()); + transport.receive_timestamps = NonZeroU32::new(32); + } } let mut config = client_config(); config.transport = transport; @@ -687,16 +706,20 @@ 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 { - 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() + congestion: Congestion::DelayBound, + ..opening() }; - rises(&compare("rises", scenario, GATE), Duration::from_secs(5)); + rises(&compare("rises delay bound", scenario, GATE), Duration::from_secs(5)); } /// The same under GCC, at the target and at twice it. GCC aims its probes at the target, at @@ -712,14 +735,10 @@ fn rises_to_target_under_gcc() { let opens = Duration::from_secs(5); for target in [TARGET, 2 * TARGET] { let scenario = Scenario { - latency: Duration::from_millis(50), - buffer: 170_000, - keyframe: 0, - rates: vec![(Duration::ZERO, 375_000), (opens, 1_500_000)], duration: Duration::from_secs(20), target, - gcc: true, - ..Scenario::default() + congestion: Congestion::Gcc, + ..opening() }; let trials = compare(&format!("rises gcc {target}"), scenario, SEEDS); for Trial { @@ -741,6 +760,19 @@ fn rises_to_target_under_gcc() { } } +/// 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() + } +} + /// A path short of the target, behind a queue a second deep: each attempt that misses sits out /// past the longest BBR3 waits between its own probes, and the path opening up is still found #[test] @@ -1005,6 +1037,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 { + congestion: Congestion::DelayBound, + ..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..68d5cec8107 --- /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::bbr3; + +/// 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 = bbr3::Config::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(); + 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())), + } +}