diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml index 8a9822c43f..5f5761107a 100644 --- a/.github/workflows/nightly.yml +++ b/.github/workflows/nightly.yml @@ -18,7 +18,7 @@ name: Nightly # the wheel builds in a manylinux container, the Kotlin/Swift/Go/Dart packages # link the FFI on every platform) and nothing notices until the tag is pushed. # -# `audio-quality` is here because of what it measures. It plays sixteen rows of +# `audio-quality` is here because of what it measures. It plays twenty-four rows of # audio in real time, a minute each, which no merge gate should carry, and every # number it grades is a timing one that moves with whatever else the runner is # doing. A per-PR arm would fail on unrelated load often enough to be ignored; @@ -207,7 +207,7 @@ jobs: audio-quality: name: Audio quality runs-on: ubuntu-24.04-arm - # Sixteen rows at a minute each, plus the relay, CLI, and shaper built from source. + # Twenty-four rows at a minute each, plus the relay, CLI, and shaper built from source. timeout-minutes: 60 steps: - name: Free disk space diff --git a/Cargo.lock b/Cargo.lock index b397cdaf8f..01366893eb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3036,6 +3036,16 @@ version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "15cdd26707701c53297e2fa6afb323d55fbc1d0810c3aec078ae3ef0424c3c15" +[[package]] +name = "humantime-serde" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57a3db5ea5923d99402c94e9feb261dc5ee9b4efa158b0315f788cf549cc200c" +dependencies = [ + "humantime", + "serde", +] + [[package]] name = "hybrid-array" version = "0.4.15" @@ -4716,8 +4726,12 @@ dependencies = [ "anyhow", "clap", "humantime", + "humantime-serde", "rand 0.10.3", + "serde", + "serde_json", "tokio", + "toml 1.1.6+spec-1.1.0", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 789ecbbfd0..8665f822e1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -124,6 +124,7 @@ hex = "0.4" # HMAC-SHA256 for the mDNS membership proofs (moq-tokio's `mdns` feature). hmac = "0.13" humantime = "2.3" +humantime-serde = "1.1.1" io-uring = "0.7.14" jsonwebtoken = "11" kio = { version = "0.6.1", path = "rs/kio" } diff --git a/quest/m0/audio-quality-harness/README.md b/quest/m0/audio-quality-harness/README.md index a02c524c59..42ac07ff3d 100644 --- a/quest/m0/audio-quality-harness/README.md +++ b/quest/m0/audio-quality-harness/README.md @@ -1,4 +1,4 @@ -# Audio quality harness +# [S] Audio quality harness ## Goal @@ -23,14 +23,14 @@ are. The native lane moved to m1 as a standalone quest (decided in the same budgets and metric schema so the two implementations can be compared rather than merely both passing. -The browser lane has landed as `test/audio-quality/`, upstreamed from the -reporter's fork on #3477 (`fperex/moq`, branch `debug-findings-solution`) on -its own, without the fork's player, estimator, or shaper changes. Real arrival -traces have landed beside it in `test/audio-quality/traces/`, graded as replay -rows through the player's container consumer and rings, and replayed by the -[jitter target's watch quest](/quest/m0/audio-jitter-target/watch.md). That -leaves the bursty and step profiles, which need shaper features the fork has -and `main` does not. +The browser lane and its opt-in shaper profiles have landed in `test/audio-quality/`, +upstreamed from the reporter's fork on #3477 (`fperex/moq`, branch +`debug-findings-solution`), without the fork's player or estimator changes. +The matrix includes bursty and mid-run step paths, and mild/wide jitter preserves +packet order. Real arrival traces have landed beside it in +`test/audio-quality/traces/`, graded as replay rows through the player's +container consumer and rings, and replayed by the [jitter target's watch +quest](/quest/m0/audio-jitter-target/watch.md). Jitter comes from the seeded userspace UDP shaper the transport drills run under (`rs/moq-shaper`, documented in `test/drill/README.md`), not from a fake @@ -56,10 +56,6 @@ nightly runner's first runs, since nightly only runs `main`'s code. Tightening the auto rows belongs to the [jitter target's watch quest](/quest/m0/audio-jitter-target/watch.md). -## Required - -- [Shaper profiles](/quest/m0/audio-quality-harness/shaper-profiles.md) - the bursty and mid-run step profiles, once moq-shaper can express them - ## Related - [Native audio quality](/quest/m1/audio-quality-native.md) - the same profiles and budgets through `moq play` on a dummy device diff --git a/quest/m0/audio-quality-harness/shaper-profiles.md b/quest/m0/audio-quality-harness/shaper-profiles.md deleted file mode 100644 index 71536c518d..0000000000 --- a/quest/m0/audio-quality-harness/shaper-profiles.md +++ /dev/null @@ -1,34 +0,0 @@ -# [M] The audio quality matrix grades a bursty path and a mid-run step - -## Goal - -`just test audio-quality` gains the two profiles `moq-shaper` cannot express -today: `bursty`, which releases datagrams in batches (the flush-span shape from -#3477, seven datagrams per 160 ms window), and `step`, which changes the -profile part-way through a row and forces the adaptive target to move. Both get -budgets in `test/audio-quality/budgets.json` and run nightly with the rest. - -## Plan - -The browser lane landed on the shaper as it is: one uniform profile both ways, -fixed for the life of the process. Its `mild` and `wide` profiles spread -arrivals with uniform jitter, which also reorders datagrams whenever the -jitter exceeds their spacing, so `wide` measures reordering as much as spread. - -The reporter's fork (`fperex/moq`, branch `debug-findings-solution`, -`rs/moq-shaper`) already has what is missing, as separate commits: batches that -release datagrams together, steps that change a profile mid-run, an -order-preserving gaussian jitter model, named profiles loaded from TOML, and a -JSON report of the counters. Upstream the batch, step, and order-preserving -jitter changes with the fork author's credit, keeping each behind the options -the drills do not set so their lane is unchanged. Skip TCP passthrough: the -harness answers the page's certificate fetch itself. - -Then add `bursty` and `step` to `run.sh`, refuse a `--duration` that ends -before the step, and switch `mild` and `wide` to the order-preserving jitter so -their names mean spread alone. Re-record the budgets for every row whose profile -changed. - -## Related - -- [Audio jitter target](/quest/m0/audio-jitter-target/README.md) - the estimator the step profile exercises diff --git a/quest/m1/audio-quality-native.md b/quest/m1/audio-quality-native.md index 11373ae103..232cc61deb 100644 --- a/quest/m1/audio-quality-native.md +++ b/quest/m1/audio-quality-native.md @@ -45,7 +45,3 @@ Standalone in m1 rather than a child of the m0 [Audio quality harness](/quest/m0/audio-quality-harness/README.md) line (decided in the 2026-09-28 quest audit): nothing in m0 waits on it. The native jitter target it grades is done on the jitter target line. - -## Related - -- [Shaper profiles](/quest/m0/audio-quality-harness/shaper-profiles.md) - the bursty and step profiles both lanes should run diff --git a/rs/moq-shaper/Cargo.toml b/rs/moq-shaper/Cargo.toml index a923d0f2f1..ebeb0190d7 100644 --- a/rs/moq-shaper/Cargo.toml +++ b/rs/moq-shaper/Cargo.toml @@ -20,5 +20,12 @@ doc = false anyhow = { workspace = true } clap = { workspace = true } humantime = { workspace = true } +humantime-serde = { workspace = true } rand = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } tokio = { workspace = true, features = ["macros", "net", "rt-multi-thread", "signal", "sync", "time"] } +toml = { workspace = true } + +[dev-dependencies] +tokio = { workspace = true, features = ["test-util"] } diff --git a/rs/moq-shaper/README.md b/rs/moq-shaper/README.md new file mode 100644 index 0000000000..af9e215f0a --- /dev/null +++ b/rs/moq-shaper/README.md @@ -0,0 +1,175 @@ +# moq-shaper + +A seeded userspace UDP impairment relay, so a test can put an impaired path in front of a server. + +The shaper binds one socket, forwards every datagram to a target after a profile's treatment, and +forwards the replies back. A client dials the shaper instead of the server and reaches it through a +path with loss, a rate limit, delay, jitter, and reordering. QUIC is indifferent to the extra hop, so +this impairs a real transport with no capabilities and nothing touching the host's network, on macOS +and Linux alike. Kernel impairment (`netem` in a network namespace) is Linux only, needs +`CAP_NET_ADMIN`, and changes the host's network for everything running on it. + +It is a datagram relay, not an HTTP interceptor or a TCP proxy, neither of which can impair QUIC. + +## Usage + +```bash +moq-relay --listen 127.0.0.1:4443 & + +moq-shaper --listen 127.0.0.1:4444 --target 127.0.0.1:4443 \ + --delay 20ms --jitter 5ms --loss 0.05 --reorder 0.02 --rate 10000000 --seed 1 +``` + +Point the client at the shaper's port. The flags build one profile, applied both ways. The seed and +profile print at start, and the counters at exit (Ctrl-C or SIGTERM), which exits nonzero if the +profile never acted. + +| Flag | What it does | +| --- | --- | +| `--listen` | The address clients send to. | +| `--target` | The address every datagram is forwarded to. | +| `--seed` | Seeds every decision; the profile's own, or random, when omitted. | +| `--delay` | The base one-way delay. | +| `--jitter` | How far the delay varies: the most either way, or the sigma of a gaussian. | +| `--jitter-model` | `uniform` (the default) or `gaussian`. See [Jitter](#jitter). | +| `--loss` | The probability that a datagram is dropped. | +| `--reorder` | The probability that a datagram skips the delay, overtaking those in flight. | +| `--rate`, `--burst`, `--queue` | A token bucket: bits per second, the bytes it lets out ahead after an idle spell (1500), and the longest a datagram waits before it is dropped (100ms). | +| `--batch`, `--batch-window` | Hold datagrams until this many wait, or the window closes, then release them together. See [Batches](#batches). | +| `--shared` | Every client shares one link each way. See [Shared path](#shared-path). | +| `--profile` | A built-in profile's name, or a profile TOML file, in place of the flags that shape the path. See [Profiles](#profiles). | +| `--report` | Write the profile, seed and counters to this file as JSON at exit. See [Report](#report). | +| `--report-interval` | Print the same JSON as one line on stdout this often. | + +As a library, `Shaper::bind` takes a `Config`: where to listen and forward, the seed, and a `Profile` +per direction. It also takes a `Setup`, which is a `Config` plus the opt-in options below, each off by +default. `Shaper::verify` fails when the shaper stopped forwarding, when an impairment the profile +configures never acted and the traffic makes that silence implausible, or when a direction carried +traffic but no datagram saw one of its phases, before the first step or after one. The relay's drills +(`rs/moq-relay/tests/drills.rs`, described in `test/drill/README.md`) run every scenario through it. + +## What a seed does and does not fix + +Every decision comes from the seed: each link draws loss, reorder and jitter from its own stream, in the +same order for every datagram whether or not the knob is on. Replay a run with the same seed and the +same datagrams are dropped, the same ones skip the delay, and each draws the same jitter. + +The seed does not make the kernel's delivery clock reproducible: when a held datagram actually lands is +the host's scheduling. A test asserting on a replay should assert on the decisions, not on arrival times. + +## Jitter + +The default, `uniform`, draws each datagram's delay on its own within `jitter` either way of `delay`, so a +later datagram can overtake an earlier one. A drill wants that, and `jitter` may not exceed `delay`. + +`gaussian` treats `jitter` as a sigma, clamps the draw at zero, and never lets a datagram leave before the +one in front of it: queueing delay on a FIFO path, which stretches and compresses the spacing and never +changes the order. The sigma may exceed `delay`, but a nonzero `jitter` still needs a nonzero `delay`. Only `reorder` overtakes. This matters beyond realism: QUIC reads an overtake as loss, +retransmits and backs off, so a run meant to measure a jittery path ends up measuring the congestion +response. On a 5ms delay with 5ms of jitter, reordering alone dragged a receiver's audio buffer from +120ms to nearly two seconds. + +## Shared path + +Each client flow gets a link of its own each way by default: its own seeded stream, rate limit and queue, +so its decisions do not depend on anyone else's traffic. + +With `shared`, every flow takes one link each way, the way clients behind one access link do. A browser +page with a WebTransport session of its own and another in its audio worker is two flows on one path: +shared, a gaussian path keeps the order across both, a batch fills from both, and one rate limit serves +both. A flow's treatment then depends on how its datagrams interleave with the others'. + +## Batches + +A paced hop, a Wi-Fi access point or a cellular scheduler bunches traffic: nothing moves for a while, then +several datagrams arrive at once. A batch holds datagrams until `count` are waiting, or until `window` +after the first arrived, then sends them all when the latest would have left. No delay or jitter produces +that clump, and it is what a receiver's jitter estimate has to cope with. A `count` below two is refused, +since the datagram that fills a batch never waits. + +## Steps + +A step changes a direction's delay, jitter, loss, reorder or rate limit once the run reaches `at`, counted +from when forwarding started. It changes only what it names, so a later step puts one knob back. A rate +step keeps the old bucket's credit, or its debt, clipped to the new one, and a limit that was not there +starts full. A step can add or change a rate limit, never remove one. A step that leaves the profile as +it was, or comes at zero, is refused. Steps come from a profile file. +A direction that carried traffic fails `verify` unless some datagram saw each phase: the profile before +the first step and after every step. Traffic that starts late, ends early, or goes quiet across a step +never saw the path change. + +## Profiles + +A profile file names both directions' treatment, the seed, and whether the path is shared. `up` is from +the client to the target, `down` the way back. Every field of a direction is optional and off when left +out, and unknown fields are refused. + +```toml +name = "example" +seed = 7 +shared = true + +[up] +delay = "5ms" +jitter = "5ms" +jitter_model = "gaussian" # required once the delay varies, through jitter or steps +loss = 0.02 +reorder = 0.01 # skips the delay, so it needs one + +[up.rate] +bits_per_second = 2000000 +burst = 12500 +queue = "250ms" + +[up.batch] +count = 7 +window = "160ms" + +[[up.steps]] # at 30s the delay grows, +at = "30s" +delay = "60ms" + +[[up.steps]] # and at 60s it recovers +at = "60s" +delay = "5ms" + +[down] +delay = "5ms" +``` + +`--profile` takes a file's path, or one of the built-ins in `profiles/`, embedded in the binary. The +audio-quality lane (`test/audio-quality/`) uses `bursty` and `step`, plus its own flag profiles. + +| Name | What it models | +| --- | --- | +| `near-zero` | The control. The shaper is in the path but treats nothing. | +| `mild` | A healthy wired LAN: 5ms delay, 5ms sigma, in order. | +| `bursty` | A paced hop: seven datagrams per 160ms window, released together. | +| `step` | 5ms for thirty seconds, then 60ms, with no recovery. | + +Each built-in is a shared path, as a browser page's sessions on one host would be. + +## Counters + +`Shaper::stats` counts what each direction did, summed over every client. + +| Counter | Meaning | +| --- | --- | +| `packets` | Datagrams received. | +| `lost` | Datagrams dropped by `loss`. | +| `overflowed` | Datagrams dropped because the rate limit's queue was full. | +| `throttled` | Datagrams that waited for the rate limit. | +| `delayed` | Datagrams given a nonzero delay, by the profile or by a batch holding them. | +| `reordered` | Datagrams sent ahead of the delay, overtaking any still in flight. | + +## Report + +`--report ` writes `{"profile", "seed", "up", "down"}` at exit, before the verdict, so a run that +fails it still leaves its counters behind. `profile` is the profile's name, or null when the flags built +it, and `up` and `down` hold the counters above. `--report-interval` prints the same object as one line on +stdout each period, which also survives a shaper killed before it could write the file. + +A profile that silently did nothing turns an impaired run into an unimpaired pass. `verify` catches that +for the shaper's own exit; a harness grading a run should also check the report, where `near-zero` is the +only profile for which a `delayed` of zero is the right answer. Keep the name, the seed and the counters +with the run's artifacts: the seed is what turns a failing run into one that can be looked at again. diff --git a/rs/moq-shaper/profiles/bursty.toml b/rs/moq-shaper/profiles/bursty.toml new file mode 100644 index 0000000000..307c8f2f4e --- /dev/null +++ b/rs/moq-shaper/profiles/bursty.toml @@ -0,0 +1,16 @@ +# A paced hop that bunches traffic: nothing moves until seven datagrams are +# waiting or the window closes, and then they all leave together. +# +# This is the shape a receiver's jitter estimate has to see, and no delay-and- +# spread profile produces it: the datagrams arrive together rather than late. +name = "bursty" +seed = 3 +shared = true + +[up.batch] +count = 7 +window = "160ms" + +[down.batch] +count = 7 +window = "160ms" diff --git a/rs/moq-shaper/profiles/mild.toml b/rs/moq-shaper/profiles/mild.toml new file mode 100644 index 0000000000..de101bb5bf --- /dev/null +++ b/rs/moq-shaper/profiles/mild.toml @@ -0,0 +1,15 @@ +# A healthy wired LAN: a few milliseconds of delay with the same spread again, +# queued in order. +name = "mild" +seed = 2 +shared = true + +[up] +delay = "5ms" +jitter = "5ms" +jitter_model = "gaussian" + +[down] +delay = "5ms" +jitter = "5ms" +jitter_model = "gaussian" diff --git a/rs/moq-shaper/profiles/near-zero.toml b/rs/moq-shaper/profiles/near-zero.toml new file mode 100644 index 0000000000..627f619487 --- /dev/null +++ b/rs/moq-shaper/profiles/near-zero.toml @@ -0,0 +1,7 @@ +# The control: the shaper sits in the path but treats nothing. +# +# Every other profile is graded against this one, and a harness asserting that an +# impairment applied needs a profile where `delayed == 0` is the right answer. +name = "near-zero" +seed = 1 +shared = true diff --git a/rs/moq-shaper/profiles/step.toml b/rs/moq-shaper/profiles/step.toml new file mode 100644 index 0000000000..bdd0170d2f --- /dev/null +++ b/rs/moq-shaper/profiles/step.toml @@ -0,0 +1,24 @@ +# A path that gets worse mid-run: 5ms for the first thirty seconds, then 60ms. +# +# An estimator that only ever converges downward passes every steady profile and +# fails this one. It never recovers, which is what makes it the baseline a +# recovery profile is read against. +name = "step" +seed = 4 +shared = true + +[up] +delay = "5ms" +jitter_model = "gaussian" + +[[up.steps]] +at = "30s" +delay = "60ms" + +[down] +delay = "5ms" +jitter_model = "gaussian" + +[[down.steps]] +at = "30s" +delay = "60ms" diff --git a/rs/moq-shaper/src/lib.rs b/rs/moq-shaper/src/lib.rs index 730d3a9814..5da562780e 100644 --- a/rs/moq-shaper/src/lib.rs +++ b/rs/moq-shaper/src/lib.rs @@ -13,33 +13,43 @@ //! A profile that silently did nothing would turn an impaired run into an //! unimpaired pass, so [`Shaper::verify`] fails when an impairment the profile //! configures never acted and the traffic makes that silence implausible. +//! +//! A [`Setup`] adds opt-in options to a [`Config`]: a jitter +//! model that keeps the order, one link shared by every client, batches, steps, +//! and named profiles loaded as a [`Preset`]. The README says why each exists. use std::{ cmp::Reverse, - collections::{BinaryHeap, HashMap, hash_map}, + collections::{BTreeSet, BinaryHeap, HashMap, hash_map}, fmt, net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}, sync::{ - Arc, OnceLock, + Arc, Mutex, OnceLock, atomic::{AtomicU64, Ordering}, }, time::Duration, }; +mod preset; + +pub use preset::Preset; + use anyhow::Context; use rand::{RngExt, SeedableRng, rngs::Xoshiro256PlusPlus}; +use serde::{Deserialize, Serialize}; use tokio::{net::UdpSocket, sync::mpsc, task::JoinSet, time::Instant}; /// How one direction of the path treats each datagram. /// /// A datagram is first subject to `loss`, then waits for the `rate` limit, then -/// takes `delay` plus or minus `jitter` to arrive, unless `reorder` sends it -/// ahead of everything still in flight. +/// takes `delay` plus or minus `jitter`, drawn as its [`Jitter`] model says, to +/// arrive, unless `reorder` sends it ahead of everything still in flight. #[derive(Clone, Debug, Default, PartialEq)] pub struct Profile { /// The base one-way delay. pub delay: Duration, - /// The most a datagram's delay varies from `delay`, uniformly either way. + /// How far a datagram's delay varies from `delay`: the most either way when + /// [`Jitter::Uniform`], the sigma when [`Jitter::Gaussian`]. pub jitter: Duration, /// The probability that a datagram is dropped. pub loss: f64, @@ -50,7 +60,7 @@ pub struct Profile { } impl Profile { - fn validate(&self) -> anyhow::Result<()> { + fn validate(&self, model: Jitter) -> anyhow::Result<()> { anyhow::ensure!( (0.0..=1.0).contains(&self.loss), "loss {} is not a probability", @@ -66,12 +76,19 @@ impl Profile { "reorder {} needs a delay to overtake", self.reorder ); + // A gaussian clamps its draw at zero instead, as a queue cannot run early. anyhow::ensure!( - self.jitter <= self.delay, + model == Jitter::Gaussian || self.jitter <= self.delay, "jitter {:?} exceeds delay {:?}, which would need a negative delay", self.jitter, self.delay ); + // With no delay to vary, `verify` has nothing to hold the jitter to. + anyhow::ensure!( + self.jitter.is_zero() || !self.delay.is_zero(), + "jitter {:?} needs a delay to vary", + self.jitter + ); if let Some(rate) = &self.rate { anyhow::ensure!(rate.bits_per_second > 0, "a rate limit of zero passes nothing"); } @@ -79,6 +96,138 @@ impl Profile { } } +/// How a direction draws each datagram's jitter. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum Jitter { + /// Uniform within `jitter` either way of `delay`, drawn for each datagram on + /// its own, so a later datagram can overtake an earlier one. + #[default] + Uniform, + /// A gaussian with `jitter` as its sigma, clamped at zero, that never leaves + /// before the datagram in front: it varies the spacing, never the order. + /// + /// This is queueing delay on a FIFO path. A jitter that overtakes hands QUIC + /// a gap it can only read as loss, so the run measures its congestion + /// response rather than the jitter. + Gaussian, +} + +/// Hold datagrams, then release them together, the way a paced hop bunches them. +/// +/// A batch closes once `count` datagrams are waiting, or `window` after the +/// first of them arrived, and everything in it leaves when the latest would. +/// That clump is what a receiver's jitter estimate sees on such a path, and no +/// delay or jitter produces it: the datagrams arrive together rather than late. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Batch { + /// How many datagrams close a batch early. + pub count: usize, + /// How long a batch waits for that many. + #[serde(with = "humantime_serde")] + pub window: Duration, +} + +/// A change to one direction's profile once the run reaches `at`. +/// +/// A step changes only what it names, so a later step puts one knob back +/// without restating the rest. It can add or change a rate limit, never remove +/// one. +#[derive(Clone, Debug, Default, PartialEq, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Step { + /// How far into the run the change happens. + #[serde(with = "humantime_serde")] + pub at: Duration, + /// The delay from then on. + #[serde(default, with = "humantime_serde")] + pub delay: Option, + /// The jitter from then on. + #[serde(default, with = "humantime_serde")] + pub jitter: Option, + /// The loss from then on. + #[serde(default)] + pub loss: Option, + /// The reorder from then on. + #[serde(default)] + pub reorder: Option, + /// The rate limit from then on. + #[serde(default)] + pub rate: Option, +} + +impl Step { + /// Change what this step names in `profile`. + fn apply(&self, profile: &mut Profile) { + if let Some(delay) = self.delay { + profile.delay = delay; + } + if let Some(jitter) = self.jitter { + profile.jitter = jitter; + } + if let Some(loss) = self.loss { + profile.loss = loss; + } + if let Some(reorder) = self.reorder { + profile.reorder = reorder; + } + if let Some(rate) = &self.rate { + profile.rate = Some(rate.clone()); + } + } +} + +/// One direction's opt-in options beyond its [`Profile`]. The default adds nothing. +#[derive(Clone, Debug, Default, PartialEq)] +pub struct Options { + /// How the profile's `jitter` is drawn. + pub jitter_model: Jitter, + /// Hold datagrams and release them together, if at all. + pub batch: Option, + /// Changes to the profile part-way through the run, earliest first. + pub steps: Vec, +} + +impl Options { + /// Check the options, and `profile` as they draw it, after every step too. + fn validate(&self, profile: &Profile) -> anyhow::Result<()> { + profile.validate(self.jitter_model)?; + if let Some(batch) = &self.batch { + // The datagram that fills a batch leaves at once, so one of one holds nothing. + anyhow::ensure!(batch.count > 1, "a batch of {} never holds anything", batch.count); + anyhow::ensure!(!batch.window.is_zero(), "a batch with no window never holds anything"); + } + + let mut profile = profile.clone(); + let mut previous: Option = None; + for step in &self.steps { + // The profile is what the run opens with, so a step at zero would + // replace it before any datagram saw it. + anyhow::ensure!( + !step.at.is_zero(), + "a step at zero replaces the profile; change the profile instead" + ); + // A link only ever looks at the next step due, so one out of order + // would be skipped without a word. + anyhow::ensure!( + previous.is_none_or(|previous| step.at > previous), + "the step at {:?} comes after the one at {previous:?}; steps go in order", + step.at + ); + // A step that leaves the profile as it was thinks it changes and does not. + let before = profile.clone(); + step.apply(&mut profile); + anyhow::ensure!(profile != before, "the step at {:?} changes nothing", step.at); + profile + .validate(self.jitter_model) + .with_context(|| format!("after the step at {:?}", step.at))?; + previous = Some(step.at); + } + Ok(()) + } +} + /// Each impairment a profile draws for: its name, how likely it is to act on /// any one datagram (zero when unconfigured), and the count showing it did. /// @@ -113,14 +262,29 @@ fn unapplied(config: &Config, stats: &Stats) -> Vec<&'static str> { .collect() } +/// Whether the batches `setup` configures implausibly never held a datagram. +/// +/// A batch holds every datagram but the one that fills it, so it acts on at +/// least `1 - 1 / count` of them. `batched` counts only what a batch held, not +/// what the profile delayed, so a batch that never changed a departure is caught +/// even behind a delay. +fn unbatched(setup: &Setup, stats: &Stats, batched: u64) -> bool { + let chance = |options: &Options| options.batch.map_or(0.0, |batch| 1.0 - 1.0 / batch.count as f64); + let silence = (1.0 - chance(&setup.up)).powf(stats.up.packets as f64) + * (1.0 - chance(&setup.down)).powf(stats.down.packets as f64); + batched == 0 && silence < IMPLAUSIBLE +} + /// A token-bucket rate limit with a bounded queue behind it. -#[derive(Clone, Debug, PartialEq)] +#[derive(Clone, Debug, PartialEq, Deserialize)] +#[serde(deny_unknown_fields)] pub struct Rate { /// The sustained rate. pub bits_per_second: u64, /// How many bytes may go out ahead of the sustained rate after an idle spell. pub burst: u64, /// The longest a datagram waits for the bucket; one that would wait longer is dropped. + #[serde(with = "humantime_serde")] pub queue: Duration, } @@ -146,8 +310,40 @@ pub struct Config { pub down: Profile, } +/// A [`Config`] plus the opt-in options it has no field for. +/// +/// Every option defaults to off, so a setup made from a config shapes exactly +/// as that config does. +#[derive(Clone, Debug)] +pub struct Setup { + /// Where to listen and forward, the seed, and each direction's profile. + pub config: Config, + /// Every client shares one link each way, the way clients behind one access + /// link do, rather than each getting its own. + /// + /// Shared, a jitter that keeps the order keeps it across clients, and one + /// seeded stream and one rate limit serve them all, so a client's treatment + /// depends on how its datagrams interleave with the others'. + pub shared: bool, + /// The options from a client toward the target. + pub up: Options, + /// The options from the target back toward a client. + pub down: Options, +} + +impl From for Setup { + fn from(config: Config) -> Self { + Self { + config, + shared: false, + up: Options::default(), + down: Options::default(), + } + } +} + /// What one direction did, summed over every client. -#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize)] pub struct Counters { /// Datagrams received. pub packets: u64, @@ -157,7 +353,7 @@ pub struct Counters { pub overflowed: u64, /// Datagrams that waited for the rate limit. pub throttled: u64, - /// Datagrams given a nonzero delay. + /// Datagrams given a nonzero delay, by the profile or by a batch holding them. pub delayed: u64, /// Datagrams sent ahead of the delay, overtaking any still in flight. pub reordered: u64, @@ -174,7 +370,7 @@ impl fmt::Display for Counters { } /// What both directions did. -#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize)] pub struct Stats { /// From clients toward the target. pub up: Counters, @@ -191,7 +387,7 @@ impl fmt::Display for Stats { /// A running shaper. Dropping it stops forwarding. pub struct Shaper { addr: SocketAddr, - config: Config, + setup: Setup, tally: Arc<[Tally; 2]>, /// Why forwarding stopped, if it did. failed: Arc>, @@ -200,9 +396,11 @@ pub struct Shaper { impl Shaper { /// Bind the listening socket and start forwarding. - pub async fn bind(config: Config) -> anyhow::Result { - config.up.validate().context("invalid up profile")?; - config.down.validate().context("invalid down profile")?; + pub async fn bind(setup: impl Into) -> anyhow::Result { + let setup = setup.into(); + let config = &setup.config; + setup.up.validate(&config.up).context("invalid up profile")?; + setup.down.validate(&config.down).context("invalid down profile")?; let listen = UdpSocket::bind(config.bind) .await @@ -212,11 +410,11 @@ impl Shaper { let tally = Arc::new([Tally::default(), Tally::default()]); let failed = Arc::new(OnceLock::new()); let task = tokio::spawn({ - let config = config.clone(); + let setup = setup.clone(); let tally = tally.clone(); let failed = failed.clone(); async move { - if let Err(err) = run(Arc::new(listen), config, tally).await { + if let Err(err) = run(Arc::new(listen), setup, tally).await { let _ = failed.set(format!("{err:#}")); } } @@ -224,7 +422,7 @@ impl Shaper { Ok(Self { addr, - config, + setup, tally, failed, task, @@ -238,7 +436,7 @@ impl Shaper { /// The configuration this shaper runs, including its seed. pub fn config(&self) -> &Config { - &self.config + &self.setup.config } /// What the shaper has done so far. @@ -249,18 +447,49 @@ impl Shaper { } } - /// Fail unless the shaper is still forwarding and every impairment the - /// profile configures acted on some datagram, in either direction. + /// Fail unless the shaper is still forwarding, every impairment the profile + /// configures acted on some datagram, and some datagram saw every phase of + /// a direction with steps, before the first and after each. /// /// An impairment is only held to that once the traffic makes its silence - /// implausible: a short run can see no loss, but never no delay. + /// implausible: a short run can see no loss, but never no delay. The phases + /// are held to it once their direction carried anything: traffic that all + /// came on one side of a step never saw the path change. pub fn verify(&self) -> anyhow::Result { let stats = self.stats(); if let Some(err) = self.failed.get() { anyhow::bail!("the shaper stopped forwarding: {err} ({stats})"); } - let missing = unapplied(&self.config, &stats); + let mut missing: Vec = unapplied(&self.setup.config, &stats) + .into_iter() + .map(String::from) + .collect(); + let batched = self + .tally + .iter() + .map(|tally| tally.batched.load(Ordering::Relaxed)) + .sum(); + if unbatched(&self.setup, &stats, batched) { + missing.push("batch".to_string()); + } + for (name, options, tally, counters) in [ + ("up", &self.setup.up, &self.tally[UP], &stats.up), + ("down", &self.setup.down, &self.tally[DOWN], &stats.down), + ] { + let Some(first) = options.steps.first().filter(|_| counters.packets > 0) else { + continue; + }; + let phases = tally.phases.lock().unwrap(); + if !phases.contains(&0) { + missing.push(format!("the {name} profile before its step at {:?}", first.at)); + } + for (index, step) in options.steps.iter().enumerate() { + if !phases.contains(&(index + 1)) { + missing.push(format!("the {name} step at {:?}", step.at)); + } + } + } anyhow::ensure!( missing.is_empty(), "the profile never applied {} ({stats}), so this run was not impaired as configured", @@ -287,6 +516,10 @@ struct Tally { throttled: AtomicU64, delayed: AtomicU64, reordered: AtomicU64, + /// The phases some link treated a datagram in: 0 before the first step, n after the nth. + phases: Mutex>, + /// Datagrams a batch held past when they would otherwise have left. + batched: AtomicU64, } impl Tally { @@ -309,12 +542,22 @@ fn bump(counter: &AtomicU64) { /// Accept datagrams from clients, giving each client its own flow. /// /// A flow is a socket of its own toward the target, so the target sees one -/// address per client just as it would without the shaper in the way. -async fn run(listen: Arc, config: Config, tally: Arc<[Tally; 2]>) -> anyhow::Result<()> { - let mut flows = HashMap::::new(); +/// address per client just as it would without the shaper in the way. Each +/// flow takes a link of its own each way, unless the path is shared. +async fn run(listen: Arc, setup: Setup, tally: Arc<[Tally; 2]>) -> anyhow::Result<()> { + let config = &setup.config; + let mut flows = HashMap::::new(); let mut tasks = JoinSet::new(); let mut buf = vec![0u8; u16::MAX as usize]; + // What a step's `at` counts from. + let start = Instant::now(); + + // A shared path is one link each way, drawing the streams the first flow would. + let shared = setup + .shared + .then(|| [UP, DOWN].map(|direction| link(&mut tasks, &setup, direction, 0, start, &tally))); + loop { let (size, from) = tokio::select! { res = listen.recv_from(&mut buf) => res.context("receive from a client")?, @@ -322,31 +565,71 @@ async fn run(listen: Arc, config: Config, tally: Arc<[Tally; 2]>) -> res.context("flow task panicked")??; continue; } + }; let now = Instant::now(); - let stream = 2 * flows.len() as u64; - let link = match flows.entry(from) { + let number = flows.len() as u64; + let flow = match flows.entry(from) { hash_map::Entry::Occupied(entry) => entry.into_mut(), hash_map::Entry::Vacant(entry) => { let upstream = Arc::new(bind_toward(config.target).await?); - let (up_tx, up_rx) = mpsc::unbounded_channel(); - tasks.spawn(deliver(up_rx, upstream.clone(), config.target)); - - let (down_tx, down_rx) = mpsc::unbounded_channel(); - tasks.spawn(deliver(down_rx, listen.clone(), from)); - - let down = Link::new(&config.down, config.seed, stream + 1, now, down_tx); - tasks.spawn(reply(upstream, config.target, down, tally.clone())); - - entry.insert(Link::new(&config.up, config.seed, stream, now, up_tx)) + let [up, down] = match &shared { + Some(links) => links.clone(), + None => [UP, DOWN].map(|direction| link(&mut tasks, &setup, direction, number, start, &tally)), + }; + tasks.spawn(reply( + upstream.clone(), + config.target, + down, + listen.clone(), + from, + tally.clone(), + )); + + entry.insert(Flow { upstream, up }) } }; - link.push(now, buf[..size].to_vec(), &tally[UP]); + let datagram = buf[..size].to_vec(); + let mut up = flow.up.lock().expect("link poisoned"); + up.push(now, datagram, &flow.upstream, config.target, &tally[UP]); } } +/// One client: its socket toward the target, and the link it sends through. +struct Flow { + upstream: Arc, + up: Arc>, +} + +/// One direction of the flow numbered `flow` as a link, and the task delivering +/// what it treats. +fn link( + tasks: &mut JoinSet>, + setup: &Setup, + direction: usize, + flow: u64, + start: Instant, + tally: &Arc<[Tally; 2]>, +) -> Arc> { + let (profile, options) = match direction { + UP => (&setup.config.up, &setup.up), + _ => (&setup.config.down, &setup.down), + }; + let (queue, queued) = mpsc::unbounded_channel(); + tasks.spawn(deliver(queued, options.batch, tally.clone(), direction)); + let stream = 2 * flow + direction as u64; + Arc::new(Mutex::new(Link::new( + profile, + options, + setup.config.seed, + stream, + start, + queue, + ))) +} + /// A socket that can reach `target`: loopback for a loopback target, so the /// shaper never opens a port beyond the host when it does not need to. async fn bind_toward(target: SocketAddr) -> anyhow::Result { @@ -361,11 +644,13 @@ async fn bind_toward(target: SocketAddr) -> anyhow::Result { .with_context(|| format!("bind a socket toward {target}")) } -/// Feed what the target sends a flow into that flow's return link. +/// Feed what the target sends a flow into the link back toward its client. async fn reply( socket: Arc, target: SocketAddr, - mut link: Link, + link: Arc>, + listen: Arc, + client: SocketAddr, tally: Arc<[Tally; 2]>, ) -> anyhow::Result<()> { let mut buf = vec![0u8; u16::MAX as usize]; @@ -375,105 +660,261 @@ async fn reply( if from != target { continue; } - link.push(Instant::now(), buf[..size].to_vec(), &tally[DOWN]); + let datagram = buf[..size].to_vec(); + let mut down = link.lock().expect("link poisoned"); + down.push(Instant::now(), datagram, &listen, client, &tally[DOWN]); } } -/// Send each treated datagram at its departure time. +/// Send each treated datagram at its departure time, after its batch if the +/// link has one. async fn deliver( - mut queue: mpsc::UnboundedReceiver<(Instant, Vec)>, - socket: Arc, - dest: SocketAddr, + mut queue: mpsc::UnboundedReceiver, + batch: Option, + tally: Arc<[Tally; 2]>, + direction: usize, ) -> anyhow::Result<()> { // Ordered by departure, then by arrival, so ties keep their order. let mut pending = BinaryHeap::>::new(); - let mut datagrams = HashMap::>::new(); + let mut parcels = HashMap::::new(); let mut sequence = 0u64; + let mut held = Held::default(); + let mut ready = Vec::new(); loop { let next = pending.peek().map(|Reverse((at, _))| *at); + let wake = next.into_iter().chain(held.closes).min(); tokio::select! { item = queue.recv() => { // The link is gone, so the flow is too. - let Some((at, datagram)) = item else { return Ok(()) }; - pending.push(Reverse((at, sequence))); - datagrams.insert(sequence, datagram); - sequence += 1; + let Some(parcel) = item else { return Ok(()) }; + match batch { + Some(batch) => held.hold(batch, parcel, &mut ready, &tally[direction]), + None => ready.push(parcel), + } + } + _ = tokio::time::sleep_until(wake.unwrap_or_else(Instant::now)), if wake.is_some() => { + // The window closes a batch that never filled. + if let Some(closes) = held.closes.filter(|&closes| closes <= Instant::now()) { + held.release(closes, &mut ready, &tally[direction]); + } } - _ = tokio::time::sleep_until(next.unwrap_or_else(Instant::now)), if next.is_some() => { - let now = Instant::now(); - while let Some(&Reverse((at, id))) = pending.peek() { - if at > now { - break; - } - pending.pop(); - let datagram = datagrams.remove(&id).expect("queued datagram"); - // A send error is the path losing the datagram, the way a - // network does when the far end is gone (a killed relay, say), - // so it is not the shaper failing. - let _ = socket.send_to(&datagram, dest).await; + } + for parcel in ready.drain(..) { + pending.push(Reverse((parcel.at, sequence))); + parcels.insert(sequence, parcel); + sequence += 1; + } + + // Whatever is due leaves now. The timer rounds up to the next + // millisecond, which would delay even a datagram given no delay. + let now = Instant::now(); + while let Some(&Reverse((at, id))) = pending.peek() { + if at > now { + break; + } + pending.pop(); + let parcel = parcels.remove(&id).expect("queued datagram"); + // A send error is the path losing the datagram, the way a network + // does when the far end is gone (a killed relay, say), so it is not + // the shaper failing. + let _ = parcel.socket.send_to(&parcel.datagram, parcel.dest).await; + } + } +} + +/// The datagrams a batch is holding, and when its window closes on them. +#[derive(Default)] +struct Held { + parcels: Vec, + closes: Option, +} + +impl Held { + /// Hold `parcel`, releasing the batch into `ready` once it is full. + fn hold(&mut self, batch: Batch, parcel: Parcel, ready: &mut Vec, tally: &Tally) { + let arrived = parcel.arrived; + // The window may have closed before its timer fired; the batch it closed + // leaves without this datagram, which starts the next one. + if let Some(closes) = self.closes.filter(|&closes| closes <= arrived) { + self.release(closes, ready, tally); + } + self.closes.get_or_insert(arrived + batch.window); + self.parcels.push(parcel); + if self.parcels.len() >= batch.count { + self.release(arrived, ready, tally); + } + } + + /// Release the batch, closed at `closed`, into `ready`: every datagram leaves + /// when the latest would, and none before the batch closed. + fn release(&mut self, closed: Instant, ready: &mut Vec, tally: &Tally) { + self.closes = None; + let Some(latest) = self.parcels.iter().map(|parcel| parcel.at).max() else { + return; + }; + let at = latest.max(closed); + for mut parcel in self.parcels.drain(..) { + if at > parcel.at { + bump(&tally.batched); + // A hold is a delay, counted once whichever stage gave it. + if !parcel.delayed { + bump(&tally.delayed); } } + parcel.at = at; + ready.push(parcel); } } } -/// One direction of one flow: the decisions and the state they depend on. +/// A treated datagram: when it arrived and leaves, and the socket and address +/// it leaves by. +struct Parcel { + arrived: Instant, + at: Instant, + /// Whether the link already counted it as delayed. + delayed: bool, + datagram: Vec, + socket: Arc, + dest: SocketAddr, +} + +/// One direction of one flow, or of every flow on a shared path: the decisions +/// and the state they depend on. struct Link { profile: Profile, + jitter: Jitter, rng: Xoshiro256PlusPlus, /// When the token bucket would be full again, as in GCRA. full_at: Instant, - queue: mpsc::UnboundedSender<(Instant, Vec)>, + /// When the latest in-order datagram leaves, which the next one never beats. + floor: Instant, + /// When the run started, which a step's `at` counts from. + start: Instant, + /// Every step, earliest first. + steps: Vec, + /// How many of `steps` have applied. + stepped: usize, + /// Whether a datagram was treated since the latest step applied. + treated: bool, + queue: mpsc::UnboundedSender, } impl Link { - /// Each direction of each flow draws from its own stream of the one seed, - /// so a flow's decisions do not depend on how its datagrams interleave with - /// anyone else's. Flows are numbered in the order they first send, since a - /// client's ephemeral port is no identity across runs. + /// Each link draws from its own stream of the one seed, so a flow's + /// decisions do not depend on how its datagrams interleave with anyone + /// else's, unless the path is shared. Flows are numbered in the order they + /// first send, since a client's ephemeral port is no identity across runs. fn new( profile: &Profile, + options: &Options, seed: u64, stream: u64, - now: Instant, - queue: mpsc::UnboundedSender<(Instant, Vec)>, + start: Instant, + queue: mpsc::UnboundedSender, ) -> Self { Self { profile: profile.clone(), + jitter: options.jitter_model, rng: Xoshiro256PlusPlus::seed_from_u64(seed ^ stream.wrapping_mul(0x9E37_79B9_7F4A_7C15)), - // Not the moment the link is built, which is after `now`: that would - // make the first datagram queue behind a bucket still refilling. - full_at: now, + // Full since the run started, never since the link was built, which + // is after its first datagram arrived: that would make the datagram + // queue behind a bucket still refilling. + full_at: start, + floor: start, + start, + steps: options.steps.clone(), + stepped: 0, + treated: false, queue, } } - fn push(&mut self, now: Instant, datagram: Vec, tally: &Tally) { + /// Apply every step the run has reached by `now`, in order, and record the + /// phase a datagram arriving then sees. + fn step(&mut self, now: Instant, tally: &Tally) { + while let Some(step) = self.steps.get(self.stepped) { + // A step past what the clock can hold is one the run never reaches. + let Some(at) = self.start.checked_add(step.at).filter(|at| *at <= now) else { + break; + }; + if let Some(rate) = &step.rate { + // Credit accrues at the new rate from the scheduled transition, + // even when no datagram arrived at that instant. + self.full_at = self.refilled(rate, at); + } + step.apply(&mut self.profile); + self.stepped += 1; + self.treated = false; + } + // Applying a step is no evidence it acted, so record the phase a datagram sees. + if !self.treated { + self.treated = true; + tally.phases.lock().unwrap().insert(self.stepped); + } + } + + /// When `rate`'s bucket would be full again if it took over at `now`. + /// + /// A limit that was not there starts full, the way the run started. One + /// that was keeps its credit, or its debt, clipped to the new bucket. + fn refilled(&self, rate: &Rate, now: Instant) -> Instant { + let Some(old) = &self.profile.rate else { + return now; + }; + let owed = self.full_at.saturating_duration_since(now).as_secs_f64() * old.bits_per_second as f64 / 8.0; + let credit = (old.burst as f64 - owed).min(rate.burst as f64); + now + Duration::from_secs_f64((rate.burst as f64 - credit) * 8.0 / rate.bits_per_second as f64) + } + + /// Treat a datagram and queue it to leave by `socket` for `dest`. + fn push(&mut self, now: Instant, datagram: Vec, socket: &Arc, dest: SocketAddr, tally: &Tally) { + let Some((at, delayed)) = self.treat(now, datagram.len(), tally) else { + return; + }; + // The delivery task only ends once this link is dropped. + let _ = self.queue.send(Parcel { + arrived: now, + at, + delayed, + datagram, + socket: socket.clone(), + dest, + }); + } + + /// When a datagram of `size` bytes arriving `now` leaves, and whether it was + /// counted as delayed, or `None` when it is dropped. + fn treat(&mut self, now: Instant, size: usize, tally: &Tally) -> Option<(Instant, bool)> { + self.step(now, tally); bump(&tally.packets); // Every draw happens for every datagram, whatever the profile, so one // knob's outcome never shifts the stream another knob draws from. let lose = self.rng.random::() < self.profile.loss; let skip = self.rng.random::() < self.profile.reorder; - let spread = self.rng.random::() * 2.0 - 1.0; + let spread = match self.jitter { + Jitter::Uniform => self.rng.random::() * 2.0 - 1.0, + Jitter::Gaussian => gaussian(&mut self.rng), + }; if lose { bump(&tally.lost); - return; + return None; } let mut depart = now; if let Some(rate) = &self.profile.rate { // GCRA: a datagram may leave once the backlog, itself included, is // within the burst allowance. - let full_at = self.full_at.max(now) + rate.cost(datagram.len() as u64); + let full_at = self.full_at.max(now) + rate.cost(size as u64); let start = full_at .checked_sub(rate.cost(rate.burst)) .map_or(now, |start| start.max(now)); if start - now > rate.queue { bump(&tally.overflowed); - return; + return None; } if start > now { bump(&tally.throttled); @@ -482,19 +923,39 @@ impl Link { depart = start; } + let mut delayed = false; if skip { bump(&tally.reordered); + } else if self.jitter == Jitter::Gaussian { + // A queue: nothing leaves before the datagram in front of it. + let wait = self.profile.delay.as_secs_f64() + self.profile.jitter.as_secs_f64() * spread; + let leave = (depart + Duration::from_secs_f64(wait.max(0.0))).max(self.floor); + // Counted like the uniform model whenever a delay is configured, so + // the chance `verify` holds it to is the same. + delayed = !self.profile.delay.is_zero() || leave > depart; + self.floor = leave; + depart = leave; } else if !self.profile.delay.is_zero() { let jitter = self.profile.jitter.as_secs_f64() * spread; depart += Duration::from_secs_f64(self.profile.delay.as_secs_f64() + jitter); + delayed = true; + } + if delayed { bump(&tally.delayed); } - // The delivery task only ends once this link is dropped. - let _ = self.queue.send((depart, datagram)); + Some((depart, delayed)) } } +/// A standard normal sample by Box-Muller, which always takes exactly two draws, +/// so the jitter never shifts the stream the next datagram's knobs draw from. +fn gaussian(rng: &mut Xoshiro256PlusPlus) -> f64 { + let u1 = rng.random::().max(f64::MIN_POSITIVE); + let u2 = rng.random::(); + (-2.0 * u1.ln()).sqrt() * (std::f64::consts::TAU * u2).cos() +} + #[cfg(test)] mod tests { use super::*; @@ -552,6 +1013,22 @@ mod tests { assert_eq!(stats.down.packets, 50); } + #[tokio::test(start_paused = true)] + async fn an_undelayed_datagram_waits_for_no_timer() { + let (_shaper, client) = setup(1, Profile::default(), Profile::default()).await; + + // Off a millisecond boundary, where a timer would round up to the next one. + tokio::time::advance(Duration::from_micros(500)).await; + let start = Instant::now(); + client.send(b"ping").await.unwrap(); + let mut buf = [0u8; 4]; + client.recv(&mut buf).await.unwrap(); + + // A paused clock only moves when every task waits on a timer, so a round + // trip that took any time slept on one. + assert_eq!(start.elapsed(), Duration::ZERO); + } + #[tokio::test] async fn the_seed_reproduces_the_losses() { let lossy = Profile { @@ -724,4 +1201,875 @@ mod tests { }; refused(undelayed, "needs a delay").await; } + + /// An echo server, a shaper `setup` builds from a config aimed at it, and a + /// client dialing the shaper. + async fn shaped(setup: impl FnOnce(Config) -> Setup) -> (Shaper, UdpSocket) { + let echo = UdpSocket::bind(LOCALHOST).await.unwrap(); + let target = echo.local_addr().unwrap(); + tokio::spawn(async move { + let mut buf = vec![0u8; u16::MAX as usize]; + loop { + let (size, from) = echo.recv_from(&mut buf).await.unwrap(); + echo.send_to(&buf[..size], from).await.unwrap(); + } + }); + + let config = Config { + bind: LOCALHOST, + target, + seed: 3, + up: Profile::default(), + down: Profile::default(), + }; + let shaper = Shaper::bind(setup(config)).await.unwrap(); + + let client = UdpSocket::bind(LOCALHOST).await.unwrap(); + client.connect(shaper.addr()).await.unwrap(); + (shaper, client) + } + + /// Push `count` numbered datagrams through one link, `spacing` apart, and + /// return each id with its departure, in the order they would be sent. + fn departures( + profile: &Profile, + options: &Options, + count: u32, + spacing: Duration, + ) -> (Vec<(u32, Instant)>, Counters) { + let (queue, _) = mpsc::unbounded_channel(); + let now = Instant::now(); + let mut link = Link::new(profile, options, 7, 0, now, queue); + let tally = Tally::default(); + let mut sent: Vec<(u32, Instant)> = (0..count) + .filter_map(|id| Some((id, link.treat(now + spacing * id, 4, &tally)?.0))) + .collect(); + // Ties leave in arrival order, which is id order here. + sent.sort_by_key(|&(id, at)| (at, id)); + (sent, tally.snapshot()) + } + + const GAUSSIAN: Options = Options { + jitter_model: Jitter::Gaussian, + batch: None, + steps: Vec::new(), + }; + + #[tokio::test] + async fn only_the_gaussian_model_keeps_the_order() { + let jittery = Profile { + delay: Duration::from_millis(20), + jitter: Duration::from_millis(10), + ..Default::default() + }; + + let (_uniform, client) = shaped(|config| { + Config { + up: jittery.clone(), + ..config + } + .into() + }) + .await; + let got = round_trip(&client, 100).await; + let mut sorted = got.clone(); + sorted.sort(); + assert_eq!(sorted, (0..100).collect::>(), "jitter lost datagrams"); + assert_ne!(got, sorted, "uniform jitter never overtook, so this proves nothing"); + + let (gaussian, client) = shaped(|config| Setup { + up: GAUSSIAN, + ..Config { up: jittery, ..config }.into() + }) + .await; + let got = round_trip(&client, 100).await; + assert_eq!(got, (0..100).collect::>(), "gaussian jitter reordered"); + + let stats = gaussian.verify().unwrap(); + assert_eq!(stats.up.reordered, 0, "{stats}"); + assert_eq!(stats.up.delayed, 100, "{stats}"); + } + + #[test] + fn gaussian_jitter_never_leaves_before_it_arrived() { + // A sigma far past the delay, so most draws would be negative unclamped. + let wide = Profile { + delay: Duration::from_millis(1), + jitter: Duration::from_millis(50), + ..Default::default() + }; + let now = Instant::now(); + let (sent, counters) = departures(&wide, &GAUSSIAN, 1000, Duration::ZERO); + assert_eq!(sent.len(), 1000); + assert!(sent.iter().all(|&(_, at)| at >= now)); + assert!(counters.delayed > 0, "{counters}"); + } + + #[test] + fn gaussian_jitter_alone_never_changes_the_order() { + // Ten times the arrival spacing as sigma, so an independent draw per + // datagram would shuffle nearly all of them. + let jittery = Profile { + delay: Duration::from_millis(5), + jitter: Duration::from_millis(50), + ..Default::default() + }; + let (sent, counters) = departures(&jittery, &GAUSSIAN, 2000, Duration::from_millis(5)); + let ids: Vec = sent.iter().map(|&(id, _)| id).collect(); + assert_eq!(ids, (0..2000).collect::>(), "gaussian jitter reordered"); + assert_eq!(counters.reordered, 0); + } + + #[test] + fn gaussian_jitter_still_varies_the_spacing() { + let spacing = Duration::from_millis(20); + let jittery = Profile { + delay: Duration::from_millis(5), + jitter: Duration::from_millis(5), + ..Default::default() + }; + let (sent, _) = departures(&jittery, &GAUSSIAN, 2000, spacing); + + // Keeping the order is not pacing the path: a datagram that drew more + // than the one in front falls further behind, and one that drew less + // closes up against it. + let gaps: Vec = sent.windows(2).map(|pair| pair[1].1 - pair[0].1).collect(); + assert!(gaps.iter().any(|&gap| gap < spacing), "nothing closed up"); + assert!(gaps.iter().any(|&gap| gap > spacing), "nothing fell behind"); + } + + #[test] + fn a_reorder_still_overtakes_gaussian_jitter() { + let shuffled = Profile { + delay: Duration::from_millis(5), + jitter: Duration::from_millis(50), + reorder: 0.05, + ..Default::default() + }; + let (sent, counters) = departures(&shuffled, &GAUSSIAN, 2000, Duration::from_millis(5)); + let overtaken = sent.windows(2).filter(|pair| pair[0].0 > pair[1].0).count(); + assert!(overtaken > 0, "a reorder never overtook"); + assert!(counters.reordered > 0, "{counters}"); + } + + #[test] + fn the_seed_reproduces_the_gaussian_decisions() { + let everything = Profile { + delay: Duration::from_millis(5), + jitter: Duration::from_millis(5), + loss: 0.05, + reorder: 0.05, + ..Default::default() + }; + let (first, first_counters) = departures(&everything, &GAUSSIAN, 1000, Duration::from_millis(1)); + let (second, second_counters) = departures(&everything, &GAUSSIAN, 1000, Duration::from_millis(1)); + + // The same ids survive and leave at the same offsets from their start. + let offsets = |sent: &[(u32, Instant)]| { + let start = sent.iter().map(|&(_, at)| at).min().unwrap(); + sent.iter().map(|&(id, at)| (id, at - start)).collect::>() + }; + assert_eq!(offsets(&first), offsets(&second)); + assert_eq!(first_counters, second_counters); + assert!(first_counters.lost > 0 && first_counters.reordered > 0); + } + + #[tokio::test] + async fn a_gaussian_sigma_may_exceed_the_delay() { + // The uniform model refuses this; a gaussian clamps its draw at zero. + let wide = Profile { + delay: Duration::from_millis(5), + jitter: Duration::from_millis(10), + ..Default::default() + }; + let (shaper, client) = shaped(|config| Setup { + up: GAUSSIAN, + ..Config { up: wide, ..config }.into() + }) + .await; + assert_eq!(round_trip(&client, 20).await, (0..20).collect::>()); + shaper.verify().unwrap(); + + // With no delay, `verify` could never tell the jitter acted. + let undelayed = Profile { + jitter: Duration::from_millis(5), + ..Default::default() + }; + let err = Shaper::bind(Setup { + up: GAUSSIAN, + ..Config { + bind: LOCALHOST, + target: LOCALHOST, + seed: 0, + up: undelayed, + down: Profile::default(), + } + .into() + }) + .await + .err() + .expect("accepted gaussian jitter with no delay"); + assert!(format!("{err:#}").contains("needs a delay to vary"), "{err:#}"); + } + + /// Two clients take turns sending `count` numbered datagrams up the path + /// `setup` builds, and what reaches the target comes back, in order, with + /// how long after the first send it did. + async fn two_clients(setup: impl FnOnce(Config) -> Setup, count: u32) -> Vec<(u32, Duration)> { + let target = UdpSocket::bind(LOCALHOST).await.unwrap(); + let config = Config { + bind: LOCALHOST, + target: target.local_addr().unwrap(), + seed: 5, + up: Profile::default(), + down: Profile::default(), + }; + let shaper = Shaper::bind(setup(config)).await.unwrap(); + + let clients = [ + UdpSocket::bind(LOCALHOST).await.unwrap(), + UdpSocket::bind(LOCALHOST).await.unwrap(), + ]; + let start = Instant::now(); + for id in 0..count { + let client = &clients[id as usize % 2]; + client.send_to(&id.to_be_bytes(), shaper.addr()).await.unwrap(); + } + + let mut got = Vec::new(); + let mut buf = [0u8; 4]; + while let Ok(Ok(_)) = tokio::time::timeout(Duration::from_millis(500), target.recv_from(&mut buf)).await { + got.push((u32::from_be_bytes(buf), start.elapsed())); + } + got + } + + #[tokio::test] + async fn a_shared_path_keeps_the_order_across_clients() { + let jittery = Profile { + delay: Duration::from_millis(20), + jitter: Duration::from_millis(10), + ..Default::default() + }; + let order = |shared: bool| { + let up = jittery.clone(); + async move { + let got = two_clients( + |config| Setup { + shared, + up: GAUSSIAN, + ..Config { up, ..config }.into() + }, + 200, + ) + .await; + got.into_iter().map(|(id, _)| id).collect::>() + } + }; + + // Separate links keep each client's order, but not the order between them. + let apart = order(false).await; + let mut sorted = apart.clone(); + sorted.sort(); + assert_eq!(sorted, (0..200).collect::>(), "a client lost datagrams"); + assert_ne!( + apart, sorted, + "separate links kept the order anyway, so this proves nothing" + ); + for parity in 0..2 { + assert!(apart.iter().filter(|&&id| id % 2 == parity).is_sorted()); + } + + assert_eq!( + order(true).await, + (0..200).collect::>(), + "a shared path reordered" + ); + } + + /// Send `count` numbered datagrams `spacing` apart, and collect what echoes + /// back with when it did, alongside when each was sent. + async fn paced(client: &UdpSocket, count: u32, spacing: Duration) -> (Vec, Vec<(u32, Instant)>) { + let send = async { + let mut sent = Vec::new(); + for id in 0..count { + sent.push(Instant::now()); + client.send(&id.to_be_bytes()).await.unwrap(); + tokio::time::sleep(spacing).await; + } + sent + }; + let receive = async { + let mut got = Vec::new(); + let mut buf = [0u8; 4]; + while let Ok(Ok(_)) = tokio::time::timeout(Duration::from_millis(500), client.recv(&mut buf)).await { + got.push((u32::from_be_bytes(buf), Instant::now())); + } + got + }; + tokio::join!(send, receive) + } + + fn batched(count: usize, window: Duration) -> Options { + Options { + batch: Some(Batch { count, window }), + ..Default::default() + } + } + + #[tokio::test] + async fn a_batch_releases_on_count_and_on_the_window() { + let socket = Arc::new(UdpSocket::bind(LOCALHOST).await.unwrap()); + let ms = |ms| Duration::from_millis(ms); + let now = Instant::now(); + let parcel = |at: Instant| Parcel { + arrived: at, + at, + delayed: false, + datagram: Vec::new(), + socket: socket.clone(), + dest: LOCALHOST, + }; + let batch = Batch { + count: 3, + window: ms(160), + }; + let tally = Tally::default(); + let mut held = Held::default(); + let mut ready = Vec::new(); + + held.hold(batch, parcel(now), &mut ready, &tally); + held.hold(batch, parcel(now + ms(10)), &mut ready, &tally); + assert!(ready.is_empty(), "a partial batch leaked"); + assert_eq!( + held.closes, + Some(now + ms(160)), + "the window counts from the first arrival" + ); + + held.hold(batch, parcel(now + ms(20)), &mut ready, &tally); + let leaves: Vec = ready.drain(..).map(|parcel| parcel.at).collect(); + assert_eq!( + leaves, + [now + ms(20); 3], + "a full batch leaves together, as the latest would" + ); + assert_eq!( + tally.snapshot().delayed, + 2, + "the datagram that filled it waited for nothing" + ); + + // The window closes a batch that never fills, and nothing in it leaves sooner. + held.hold(batch, parcel(now + ms(30)), &mut ready, &tally); + let closes = held.closes.unwrap(); + assert_eq!(closes, now + ms(190)); + held.release(closes, &mut ready, &tally); + assert_eq!(ready.iter().map(|parcel| parcel.at).collect::>(), [closes]); + assert_eq!(tally.snapshot().delayed, 3); + ready.clear(); + + // A datagram that arrives after the window closed, before its timer fired, + // starts the next batch rather than joining the closed one. + held.hold(batch, parcel(now + ms(200)), &mut ready, &tally); + let closes = held.closes.unwrap(); + held.hold(batch, parcel(closes + ms(1)), &mut ready, &tally); + assert_eq!(ready.iter().map(|parcel| parcel.at).collect::>(), [closes]); + assert_eq!(held.parcels.len(), 1, "the late datagram joined the closed batch"); + assert_eq!(held.closes, Some(closes + ms(1) + ms(160))); + assert_eq!(tally.snapshot().delayed, 4); + assert_eq!(tally.batched.load(Ordering::Relaxed), 4); + + // Behind a delay longer than the window, a lone datagram leaves when the + // delay says: the batch held nothing, whatever `delayed` counts. + ready.clear(); + let mut held = Held::default(); + let tally = Tally::default(); + let late = Parcel { + at: now + ms(1000), + delayed: true, + ..parcel(now) + }; + held.hold(batch, late, &mut ready, &tally); + held.release(held.closes.unwrap(), &mut ready, &tally); + assert_eq!(ready[0].at, now + ms(1000)); + assert_eq!(tally.batched.load(Ordering::Relaxed), 0); + } + + #[tokio::test] + async fn a_batch_releases_datagrams_together() { + let (shaper, client) = shaped(|config| Setup { + up: batched(7, Duration::from_millis(160)), + ..config.into() + }) + .await; + + // Three full batches, each filled well inside its window. + let (_, got) = paced(&client, 21, Duration::from_millis(10)).await; + assert_eq!( + got.iter().map(|&(id, _)| id).collect::>(), + (0..21).collect::>() + ); + + let batches: Vec<&[(u32, Instant)]> = got.chunks(7).collect(); + for batch in &batches { + let spread = batch[6].1 - batch[0].1; + assert!( + spread < Duration::from_millis(15), + "a batch arrived spread over {spread:?}" + ); + } + for pair in batches.windows(2) { + let gap = pair[1][0].1 - pair[0][0].1; + assert!(gap > Duration::from_millis(40), "two batches arrived {gap:?} apart"); + } + + let stats = shaper.verify().unwrap(); + assert_eq!( + stats.up.delayed, 18, + "every datagram but the last of each batch waits: {stats}" + ); + } + + #[tokio::test] + async fn the_window_releases_a_batch_that_never_fills() { + let (shaper, client) = shaped(|config| Setup { + up: batched(100, Duration::from_millis(80)), + ..config.into() + }) + .await; + + let (sent, got) = paced(&client, 3, Duration::ZERO).await; + assert_eq!(got.len(), 3); + let held = got[0].1 - sent[0]; + assert!(held >= Duration::from_millis(80), "the batch left after only {held:?}"); + assert_eq!(shaper.verify().unwrap().up.delayed, 3); + } + + #[tokio::test] + async fn a_shared_batch_fills_from_every_client() { + // Three datagrams from each of two clients, into batches of six: only a + // shared path fills one, and separate links wait out the window. + let window = Duration::from_millis(300); + for shared in [true, false] { + let got = two_clients( + |config| Setup { + shared, + up: batched(6, window), + ..config.into() + }, + 6, + ) + .await; + assert_eq!(got.len(), 6); + let last = got.iter().map(|&(_, after)| after).max().unwrap(); + assert_eq!( + last < window / 2, + shared, + "shared {shared}: the batch left after {last:?}" + ); + } + } + + #[tokio::test] + async fn a_batch_needs_a_count_and_a_window() { + let refused = |options: Options, why: &'static str| async move { + let config = Config { + bind: LOCALHOST, + target: LOCALHOST, + seed: 0, + up: Profile::default(), + down: Profile::default(), + }; + let err = Shaper::bind(Setup { + up: options, + ..config.into() + }) + .await + .err() + .unwrap_or_else(|| panic!("accepted a batch where {why}")); + assert!(format!("{err:#}").contains(why), "{err:#}"); + }; + refused(batched(0, Duration::from_millis(10)), "a batch of 0 never holds").await; + refused(batched(1, Duration::from_millis(10)), "a batch of 1 never holds").await; + refused(batched(7, Duration::ZERO), "no window never holds").await; + } + + #[test] + fn a_batch_that_never_held_anything_is_unapplied() { + let config = Config { + bind: LOCALHOST, + target: LOCALHOST, + seed: 0, + up: Profile::default(), + down: Profile::default(), + }; + let setup = Setup { + up: batched(7, Duration::from_millis(160)), + ..config.into() + }; + let quiet = |packets| Stats { + up: Counters { + packets, + ..Default::default() + }, + ..Default::default() + }; + + // (1/7)^2 is 2%: two datagrams through batches of seven prove nothing. + assert!(!unbatched(&setup, &quiet(2), 0)); + // (1/7)^10 is 4e-9: ten datagrams a batch never held means no batch. + assert!(unbatched(&setup, &quiet(10), 0)); + assert!(!unbatched(&setup, &quiet(10), 6)); + + // A delay the profile gave is not a batch holding anything. + let mut delayed = quiet(10); + delayed.up.delayed = 10; + assert!(unbatched(&setup, &delayed, 0)); + } + + /// How long after `at` each of `sizes` leaves one link, fed at the same instant. + fn owed(link: &mut Link, at: Instant, sizes: &[usize]) -> Vec { + let tally = Tally::default(); + sizes + .iter() + .map(|&size| link.treat(at, size, &tally).expect("dropped").0 - at) + .collect() + } + + fn stepped(steps: Vec) -> Options { + Options { + steps, + ..Default::default() + } + } + + #[test] + fn each_step_applies_at_its_own_time() { + let ms = Duration::from_millis; + let profile = Profile { + delay: ms(5), + ..Default::default() + }; + let options = stepped(vec![ + Step { + at: ms(100), + delay: Some(ms(60)), + ..Default::default() + }, + Step { + at: ms(200), + delay: Some(ms(5)), + ..Default::default() + }, + ]); + let (queue, _) = mpsc::unbounded_channel(); + let start = Instant::now(); + let mut link = Link::new(&profile, &options, 7, 0, start, queue); + + assert_eq!(owed(&mut link, start, &[16]), [ms(5)]); + assert_eq!(owed(&mut link, start + ms(100), &[16]), [ms(60)]); + // The step back restores what the run opened with. + assert_eq!(owed(&mut link, start + ms(200), &[16]), [ms(5)]); + } + + #[test] + fn a_loss_step_turns_the_loss_on_and_off_again() { + let secs = Duration::from_secs; + let options = stepped(vec![ + Step { + at: secs(30), + loss: Some(1.0), + ..Default::default() + }, + Step { + at: secs(60), + loss: Some(0.0), + ..Default::default() + }, + ]); + let (queue, _) = mpsc::unbounded_channel(); + let start = Instant::now(); + let mut link = Link::new(&Profile::default(), &options, 7, 0, start, queue); + let tally = Tally::default(); + let mut lost = |at: Instant| (0..100).filter(|_| link.treat(at, 16, &tally).is_none()).count(); + + assert_eq!(lost(start), 0, "the profile opens clean"); + assert_eq!(lost(start + secs(30)), 100, "the step lost nothing"); + assert_eq!(lost(start + secs(60)), 0, "the step back never cleared"); + } + + #[test] + fn a_rate_step_narrows_the_bucket_and_widens_it_again() { + let secs = Duration::from_secs; + let rate = |bytes_per_second: u64| Rate { + bits_per_second: bytes_per_second * 8, + burst: 100, + queue: secs(10), + }; + let profile = Profile { + rate: Some(rate(4000)), + ..Default::default() + }; + let options = stepped(vec![ + Step { + at: secs(30), + rate: Some(rate(1000)), + ..Default::default() + }, + Step { + at: secs(60), + rate: Some(rate(4000)), + ..Default::default() + }, + ]); + let (queue, _) = mpsc::unbounded_channel(); + let start = Instant::now(); + let mut link = Link::new(&profile, &options, 7, 0, start, queue); + + // The bucket is full at each stretch, so the first 100 bytes leave at + // once and the next 100 owe one bucket's refill at the rate in force. + let ms = Duration::from_millis; + assert_eq!(owed(&mut link, start, &[100, 100]), [ms(0), ms(25)]); + assert_eq!(owed(&mut link, start + secs(30), &[100, 100]), [ms(0), ms(100)]); + assert_eq!(owed(&mut link, start + secs(60), &[100, 100]), [ms(0), ms(25)]); + } + + #[test] + fn a_rate_step_keeps_the_debt_it_takes_over() { + let ms = Duration::from_millis; + let rate = |bytes_per_second: u64| Rate { + bits_per_second: bytes_per_second * 8, + burst: 100, + queue: Duration::from_secs(10), + }; + let profile = Profile { + rate: Some(rate(1000)), + ..Default::default() + }; + // Halfway through a 300 byte backlog at 1000 bytes a second, the rate doubles. + let options = stepped(vec![Step { + at: ms(100), + rate: Some(rate(2000)), + ..Default::default() + }]); + let (queue, _) = mpsc::unbounded_channel(); + let start = Instant::now(); + let mut link = Link::new(&profile, &options, 7, 0, start, queue); + assert_eq!(owed(&mut link, start, &[100, 100, 100]), [ms(0), ms(100), ms(200)]); + + // 100 bytes of the backlog are still owed, and now drain in 50ms, so + // a fresh 100 bytes leave once those and themselves are paid. + assert_eq!(owed(&mut link, start + ms(100), &[100]), [ms(100)]); + + // A quiet link still changes rates at the step, not at its next arrival. + let (queue, _) = mpsc::unbounded_channel(); + let mut link = Link::new(&profile, &options, 7, 0, start, queue); + owed(&mut link, start, &[100, 100, 100]); + assert_eq!(owed(&mut link, start + ms(200), &[100]), [ms(0)]); + } + + #[tokio::test] + async fn a_step_gets_worse_part_way_through() { + let ms = Duration::from_millis; + let (shaper, client) = shaped(|config| Setup { + up: stepped(vec![Step { + at: ms(150), + delay: Some(ms(60)), + ..Default::default() + }]), + ..Config { + up: Profile { + delay: ms(5), + ..Default::default() + }, + ..config + } + .into() + }) + .await; + + let (sent, got) = paced(&client, 40, ms(10)).await; + assert_eq!(got.len(), 40); + let median = |range: std::ops::Range| { + let mut latency: Vec = got[range].iter().map(|&(id, at)| at - sent[id as usize]).collect(); + latency.sort(); + latency[latency.len() / 2] + }; + + // The first ten leave well inside the first 150ms, the last fifteen well after. + let (before, after) = (median(0..10), median(25..40)); + assert!( + after > before + ms(30), + "median latency went from {before:?} to {after:?}" + ); + shaper.verify().unwrap(); + } + + #[tokio::test] + async fn a_step_the_run_never_reached_is_unapplied() { + let (shaper, client) = shaped(|config| Setup { + down: stepped(vec![Step { + at: Duration::from_secs(60), + delay: Some(Duration::from_millis(60)), + ..Default::default() + }]), + ..config.into() + }) + .await; + assert_eq!(round_trip(&client, 10).await, (0..10).collect::>()); + + let err = shaper.verify().expect_err("a run that ended before its step passed"); + assert!(format!("{err:#}").contains("the down step at 60s"), "{err:#}"); + } + + #[tokio::test(start_paused = true)] + async fn a_step_no_datagram_saw_is_unapplied() { + let ms = Duration::from_millis; + let (shaper, client) = shaped(|config| Setup { + up: stepped(vec![ + Step { + at: ms(100), + delay: Some(ms(60)), + ..Default::default() + }, + Step { + at: ms(200), + delay: Some(ms(5)), + ..Default::default() + }, + ]), + ..Config { + up: Profile { + delay: ms(5), + ..Default::default() + }, + ..config + } + .into() + }) + .await; + + // One datagram before the first step and one after the second, so the + // 60ms phase passes with nothing in it. + let mut buf = [0u8; 4]; + client.send(&0u32.to_be_bytes()).await.unwrap(); + client.recv(&mut buf).await.unwrap(); + tokio::time::sleep(ms(250)).await; + client.send(&1u32.to_be_bytes()).await.unwrap(); + client.recv(&mut buf).await.unwrap(); + + let err = shaper.verify().expect_err("a run that skipped a step passed"); + let err = format!("{err:#}"); + assert!(err.contains("the up step at 100ms"), "{err}"); + assert!(!err.contains("the up step at 200ms"), "{err}"); + } + + #[tokio::test(start_paused = true)] + async fn a_profile_no_datagram_saw_before_its_step_is_unapplied() { + let ms = Duration::from_millis; + let (shaper, client) = shaped(|config| Setup { + up: stepped(vec![Step { + at: ms(100), + delay: Some(ms(60)), + ..Default::default() + }]), + ..Config { + up: Profile { + delay: ms(5), + ..Default::default() + }, + ..config + } + .into() + }) + .await; + + // The first datagram comes after the step, so the 5ms phase passes with nothing in it. + tokio::time::sleep(ms(250)).await; + let mut buf = [0u8; 4]; + client.send(&0u32.to_be_bytes()).await.unwrap(); + client.recv(&mut buf).await.unwrap(); + + let err = shaper.verify().expect_err("a run that started after its step passed"); + let err = format!("{err:#}"); + assert!(err.contains("the up profile before its step at 100ms"), "{err}"); + assert!(!err.contains("the up step at"), "{err}"); + } + + #[test] + fn a_step_past_the_clock_never_applies() { + let options = stepped(vec![Step { + at: Duration::MAX, + delay: Some(Duration::from_millis(60)), + ..Default::default() + }]); + let (queue, _) = mpsc::unbounded_channel(); + let start = Instant::now(); + let mut link = Link::new(&Profile::default(), &options, 7, 0, start, queue); + assert_eq!(owed(&mut link, start + Duration::from_secs(1), &[16]), [Duration::ZERO]); + } + + #[tokio::test] + async fn steps_that_would_be_skipped_or_do_nothing_are_refused() { + let refused = |steps: Vec, why: &'static str| async move { + let config = Config { + bind: LOCALHOST, + target: LOCALHOST, + seed: 0, + up: Profile { + delay: Duration::from_millis(20), + ..Default::default() + }, + down: Profile::default(), + }; + let err = Shaper::bind(Setup { + down: stepped(steps), + ..config.into() + }) + .await + .err() + .unwrap_or_else(|| panic!("accepted steps where {why}")); + assert!(format!("{err:#}").contains(why), "{err:#}"); + }; + let at = |secs| Step { + at: Duration::from_secs(secs), + loss: Some(0.1), + ..Default::default() + }; + + refused(vec![at(60), at(30)], "steps go in order").await; + refused(vec![at(0)], "a step at zero").await; + refused( + vec![Step { + at: Duration::from_secs(30), + ..Default::default() + }], + "changes nothing", + ) + .await; + // Restating the value already in force changes nothing either. + refused( + vec![ + at(30), + Step { + at: Duration::from_secs(60), + loss: Some(0.1), + ..Default::default() + }, + ], + "the step at 60s changes nothing", + ) + .await; + // Uniform jitter past the delay, reached by a step rather than at the start. + refused( + vec![Step { + at: Duration::from_secs(30), + jitter: Some(Duration::from_millis(10)), + ..Default::default() + }], + "exceeds delay", + ) + .await; + } } diff --git a/rs/moq-shaper/src/main.rs b/rs/moq-shaper/src/main.rs index 2a2841e101..c16eb30578 100644 --- a/rs/moq-shaper/src/main.rs +++ b/rs/moq-shaper/src/main.rs @@ -1,10 +1,13 @@ //! Put a seeded, impaired UDP path in front of a server. //! -//! Both directions get the same profile. The seed and profile print at start, -//! and the counters at exit, which fails if the profile never acted. +//! Both directions get the same profile from the flags, unless `--profile` +//! names a profile file, which sets each direction on its own. The seed and +//! profile print at start, and the counters at exit, which fails if the profile +//! never acted. -use std::{net::SocketAddr, time::Duration}; +use std::{net::SocketAddr, path::PathBuf, time::Duration}; +use anyhow::Context; use clap::Parser; #[derive(Parser)] @@ -16,15 +19,28 @@ struct Args { /// The address every datagram is forwarded to. #[arg(long)] target: SocketAddr, - /// Seeds every treatment decision; random when omitted. + /// Seeds every treatment decision; the profile's own, or random, when omitted. #[arg(long)] seed: Option, + /// A built-in profile's name, or a profile TOML file, in place of the flags + /// that shape the path. + #[arg( + long, + conflicts_with_all = [ + "delay", "jitter", "jitter_model", "loss", "reorder", "rate", "burst", "queue", "batch", "batch_window", + "shared", + ], + )] + profile: Option, /// The base one-way delay, e.g. `20ms`. #[arg(long, default_value = "0s", value_parser = humantime::parse_duration)] delay: Duration, - /// The most the delay varies, uniformly either way. + /// How far the delay varies: the most either way, or the sigma of a gaussian. #[arg(long, default_value = "0s", value_parser = humantime::parse_duration)] jitter: Duration, + /// How the jitter is drawn. + #[arg(long, value_enum, default_value_t = JitterModel::Uniform)] + jitter_model: JitterModel, /// The probability that a datagram is dropped. #[arg(long, default_value_t = 0.0)] loss: f64, @@ -40,12 +56,103 @@ struct Args { /// The longest a datagram waits for the rate limit before it is dropped. #[arg(long, default_value = "100ms", value_parser = humantime::parse_duration, requires = "rate")] queue: Duration, + /// Hold datagrams until this many are waiting, then release them together. + #[arg(long, requires = "batch_window")] + batch: Option, + /// The longest a batch waits to fill before it leaves anyway. + #[arg(long, value_parser = humantime::parse_duration, requires = "batch")] + batch_window: Option, + /// Every client shares one link each way, instead of each getting its own. + #[arg(long)] + shared: bool, + /// Write the profile, seed and counters to this file as JSON at exit. + #[arg(long)] + report: Option, + /// Print the same JSON as one line on stdout this often. + #[arg(long, value_parser = humantime::parse_duration)] + report_interval: Option, +} + +#[derive(Clone, Copy, clap::ValueEnum)] +enum JitterModel { + /// Uniform either way of the delay, per datagram, so datagrams can overtake each other. + Uniform, + /// A gaussian clamped at zero that never lets a datagram overtake the one in front. + Gaussian, } #[tokio::main] async fn main() -> anyhow::Result<()> { let args = Args::parse(); + anyhow::ensure!( + args.report_interval.is_none_or(|period| !period.is_zero()), + "the report interval must be positive" + ); + + let (name, setup) = match &args.profile { + Some(profile) => { + let preset = moq_shaper::Preset::load(profile)?; + let mut setup = preset.setup(args.listen, args.target); + if let Some(seed) = args.seed { + setup.config.seed = seed; + } + (Some(preset.name), setup) + } + None => (None, flags(&args)), + }; + + let treatment = match &name { + Some(name) => format!("profile {name}"), + None if setup.shared || setup.up != moq_shaper::Options::default() => format!( + "profile {:?}, options {:?}, shared {}", + setup.config.up, setup.up, setup.shared + ), + None => format!("profile {:?}", setup.config.up), + }; + let shaper = moq_shaper::Shaper::bind(setup).await?; + let config = shaper.config(); + println!( + "shaper: {} -> {}, seed {}, {treatment}", + shaper.addr(), + config.target, + config.seed + ); + + let report = || Report { + profile: name.as_deref(), + seed: config.seed, + stats: shaper.stats(), + }; + + // The first tick is a full period in, so a line reports traffic, not the start. + let mut interval = args + .report_interval + .map(|period| tokio::time::interval_at(tokio::time::Instant::now() + period, period)); + let shutdown = shutdown(); + tokio::pin!(shutdown); + loop { + tokio::select! { + res = &mut shutdown => { + res?; + break; + } + _ = tick(&mut interval) => println!("{}", serde_json::to_string(&report())?), + } + } + + // Before the verdict, so a run that fails it still leaves its counters behind. + if let Some(path) = &args.report { + std::fs::write(path, serde_json::to_vec(&report())?).with_context(|| format!("write {}", path.display()))?; + } + + let stats = shaper.verify()?; + println!("shaper: {stats}"); + Ok(()) +} + +/// The setup the flags describe: one profile and one set of options, both ways. +fn flags(args: &Args) -> moq_shaper::Setup { let profile = moq_shaper::Profile { delay: args.delay, jitter: args.jitter, @@ -57,29 +164,50 @@ async fn main() -> anyhow::Result<()> { queue: args.queue, }), }; - let shaper = moq_shaper::Shaper::bind(moq_shaper::Config { + let config = moq_shaper::Config { bind: args.listen, target: args.target, seed: args.seed.unwrap_or_else(rand::random), up: profile.clone(), down: profile, - }) - .await?; - - let config = shaper.config(); - println!( - "shaper: {} -> {}, seed {}, profile {:?}", - shaper.addr(), - config.target, - config.seed, - config.up - ); + }; + let options = moq_shaper::Options { + jitter_model: match args.jitter_model { + JitterModel::Uniform => moq_shaper::Jitter::Uniform, + JitterModel::Gaussian => moq_shaper::Jitter::Gaussian, + }, + batch: args + .batch + .zip(args.batch_window) + .map(|(count, window)| moq_shaper::Batch { count, window }), + ..Default::default() + }; + moq_shaper::Setup { + shared: args.shared, + up: options.clone(), + down: options, + ..config.into() + } +} - shutdown().await?; +/// What a run did, as the JSON a harness reads back. +#[derive(serde::Serialize)] +struct Report<'a> { + /// The profile's name, or null when the flags built it. + profile: Option<&'a str>, + seed: u64, + #[serde(flatten)] + stats: moq_shaper::Stats, +} - let stats = shaper.verify()?; - println!("shaper: {stats}"); - Ok(()) +/// The next report tick, or never without an interval. +async fn tick(interval: &mut Option) { + match interval { + Some(interval) => { + interval.tick().await; + } + None => std::future::pending().await, + } } /// Wait for Ctrl-C, or SIGTERM where there is one, which is how a harness stops a child. diff --git a/rs/moq-shaper/src/preset.rs b/rs/moq-shaper/src/preset.rs new file mode 100644 index 0000000000..e8cd8095a9 --- /dev/null +++ b/rs/moq-shaper/src/preset.rs @@ -0,0 +1,248 @@ +use std::{net::SocketAddr, time::Duration}; + +use anyhow::Context; +use serde::Deserialize; + +use crate::{Batch, Config, Jitter, Options, Profile, Rate, Setup, Step}; + +/// The profiles shipped under `profiles/`, embedded so a binary needs no data files. +const BUILTIN: &[(&str, &str)] = &[ + ("near-zero", include_str!("../profiles/near-zero.toml")), + ("mild", include_str!("../profiles/mild.toml")), + ("bursty", include_str!("../profiles/bursty.toml")), + ("step", include_str!("../profiles/step.toml")), +]; + +/// A named, seeded treatment of both directions, loaded from a TOML profile. +#[derive(Clone, Debug)] +pub struct Preset { + /// The name a run records its counters under. + pub name: String, + seed: u64, + shared: bool, + up: (Profile, Options), + down: (Profile, Options), +} + +impl Preset { + /// Load a built-in profile by name, or a profile file by path. + pub fn load(name_or_path: &str) -> anyhow::Result { + let text = match BUILTIN.iter().find(|(name, _)| *name == name_or_path) { + Some((_, text)) => (*text).to_string(), + // An unknown name is almost always a typo, so say what the choices were. + None => std::fs::read_to_string(name_or_path).with_context(|| { + format!( + "no built-in profile or readable file named {name_or_path:?}; the built-ins are {}", + Self::names().collect::>().join(", ") + ) + })?, + }; + + let file: File = toml::from_str(&text).with_context(|| format!("parse profile {name_or_path:?}"))?; + file.preset() + .with_context(|| format!("invalid profile {name_or_path:?}")) + } + + /// The built-in profiles' names. + pub fn names() -> impl Iterator { + BUILTIN.iter().map(|(name, _)| *name) + } + + /// This preset listening on `bind` and forwarding to `target`, with its own seed. + pub fn setup(&self, bind: SocketAddr, target: SocketAddr) -> Setup { + Setup { + config: Config { + bind, + target, + seed: self.seed, + up: self.up.0.clone(), + down: self.down.0.clone(), + }, + shared: self.shared, + up: self.up.1.clone(), + down: self.down.1.clone(), + } + } +} + +/// A profile file. +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct File { + name: String, + seed: u64, + #[serde(default)] + shared: bool, + #[serde(default)] + up: Side, + #[serde(default)] + down: Side, +} + +impl File { + fn preset(self) -> anyhow::Result { + anyhow::ensure!(!self.name.is_empty(), "the profile has no name"); + Ok(Preset { + up: self.up.split().context("up")?, + down: self.down.split().context("down")?, + name: self.name, + seed: self.seed, + shared: self.shared, + }) + } +} + +/// One direction of a profile file: its [`Profile`] and [`Options`] in one table. +#[derive(Default, Deserialize)] +#[serde(default, deny_unknown_fields)] +struct Side { + #[serde(with = "humantime_serde")] + delay: Duration, + #[serde(with = "humantime_serde")] + jitter: Duration, + jitter_model: Option, + loss: f64, + reorder: f64, + rate: Option, + batch: Option, + steps: Vec, +} + +impl Side { + fn split(self) -> anyhow::Result<(Profile, Options)> { + // Whether datagrams can overtake each other depends on the model as soon + // as their delays differ, so a file has to say which it means rather + // than inherit whichever is the default. + let varies = !self.jitter.is_zero() + || self + .steps + .iter() + .any(|step| step.delay.is_some() || step.jitter.is_some_and(|jitter| !jitter.is_zero())); + anyhow::ensure!( + self.jitter_model.is_some() || !varies, + "the delay varies, so name a jitter_model: \"uniform\" lets a datagram overtake the one in front, \"gaussian\" never does" + ); + + let profile = Profile { + delay: self.delay, + jitter: self.jitter, + loss: self.loss, + reorder: self.reorder, + rate: self.rate, + }; + let options = Options { + jitter_model: self.jitter_model.unwrap_or_default(), + batch: self.batch, + steps: self.steps, + }; + options.validate(&profile)?; + Ok((profile, options)) + } +} + +#[cfg(test)] +mod tests { + use std::net::{IpAddr, Ipv4Addr}; + + use super::*; + + const LOCALHOST: SocketAddr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); + + #[test] + fn builtins_parse_and_are_named_after_their_file() { + for name in Preset::names() { + let preset = Preset::load(name).unwrap_or_else(|err| panic!("{name}: {err:#}")); + assert_eq!(preset.name, name); + } + } + + #[test] + fn builtins_treat_the_path_except_the_control() { + for name in Preset::names() { + let preset = Preset::load(name).unwrap(); + let untreated = (Profile::default(), Options::default()); + let idle = preset.up == untreated && preset.down == untreated; + assert_eq!(idle, name == "near-zero", "{name} treats the wrong amount of traffic"); + } + } + + #[test] + fn the_step_builtin_steps_both_directions_at_thirty_seconds() { + let setup = Preset::load("step").unwrap().setup(LOCALHOST, LOCALHOST); + for (which, profile, options) in [ + ("up", &setup.config.up, &setup.up), + ("down", &setup.config.down, &setup.down), + ] { + assert_eq!(profile.delay, Duration::from_millis(5), "{which}"); + assert_eq!(options.jitter_model, Jitter::Gaussian, "{which}"); + assert_eq!(options.steps.len(), 1, "{which}"); + assert_eq!(options.steps[0].at, Duration::from_secs(30), "{which}"); + assert_eq!(options.steps[0].delay, Some(Duration::from_millis(60)), "{which}"); + } + } + + #[test] + fn an_unknown_name_lists_the_builtins() { + let err = format!("{:#}", Preset::load("nope").unwrap_err()); + assert!(err.contains("near-zero"), "{err}"); + } + + #[test] + fn an_unknown_field_is_refused() { + // A field the schema does not know, like a rate in bytes, is refused rather than ignored. + let Err(err) = toml::from_str::("name = \"x\"\nseed = 1\n[up.rate]\nbytes_per_second = 1000\n") else { + panic!("a rate in bytes parsed"); + }; + assert!(err.to_string().contains("bytes_per_second"), "{err}"); + } + + #[test] + fn a_varying_delay_has_to_name_its_model() { + let parse = |text: &str| toml::from_str::(text).unwrap().preset(); + + let err = parse("name = \"x\"\nseed = 1\n[up]\ndelay = \"5ms\"\njitter = \"5ms\"\n").unwrap_err(); + assert!(format!("{err:#}").contains("jitter_model"), "{err:#}"); + + let err = parse("name = \"x\"\nseed = 1\n[up]\ndelay = \"5ms\"\n[[up.steps]]\nat = \"1s\"\ndelay = \"9ms\"\n") + .unwrap_err(); + assert!(format!("{err:#}").contains("jitter_model"), "{err:#}"); + + // A fixed delay cannot reorder anything, so it needs no model. + parse("name = \"x\"\nseed = 1\n[up]\ndelay = \"5ms\"\n").unwrap(); + parse("name = \"x\"\nseed = 1\n[up]\ndelay = \"5ms\"\njitter = \"5ms\"\njitter_model = \"uniform\"\n").unwrap(); + } + + #[test] + fn a_profile_file_loads_by_path() { + let path = std::env::temp_dir().join(format!("moq-shaper-preset-{}.toml", std::process::id())); + std::fs::write( + &path, + "name = \"file\"\nseed = 9\n[down.batch]\ncount = 3\nwindow = \"50ms\"\n", + ) + .unwrap(); + let preset = Preset::load(path.to_str().unwrap()); + std::fs::remove_file(&path).unwrap(); + + let setup = preset.unwrap().setup(LOCALHOST, LOCALHOST); + assert_eq!(setup.config.seed, 9); + assert_eq!( + setup.down.batch, + Some(Batch { + count: 3, + window: Duration::from_millis(50) + }) + ); + } + + #[tokio::test] + async fn the_builtins_all_bind() { + for name in Preset::names() { + // Binding proves the profile is usable, not just parseable. + let preset = Preset::load(name).unwrap(); + let shaper = crate::Shaper::bind(preset.setup(LOCALHOST, LOCALHOST)) + .await + .unwrap_or_else(|err| panic!("{name}: {err:#}")); + shaper.verify().unwrap(); + } + } +} diff --git a/rs/moq-shaper/tests/cli.rs b/rs/moq-shaper/tests/cli.rs new file mode 100644 index 0000000000..8c15351647 --- /dev/null +++ b/rs/moq-shaper/tests/cli.rs @@ -0,0 +1,87 @@ +//! The binary end to end: a profile by name and the +//! counters it reports while it runs and when it stops. +#![cfg(unix)] + +use std::{ + io::{BufRead, BufReader}, + net::{SocketAddr, UdpSocket}, + process::{Command, Stdio}, + time::Duration, +}; + +#[test] +fn a_profile_run_reports_its_counters() { + let target = UdpSocket::bind("127.0.0.1:0").unwrap(); + let target_addr = target.local_addr().unwrap(); + + let report = std::env::temp_dir().join(format!("moq-shaper-report-{}.json", std::process::id())); + let mut shaper = Command::new(env!("CARGO_BIN_EXE_moq-shaper")) + .args(["--listen", "127.0.0.1:0", "--target", &target_addr.to_string()]) + .args(["--profile", "mild", "--seed", "7"]) + .args(["--report-interval", "100ms", "--report", report.to_str().unwrap()]) + .stdout(Stdio::piped()) + .spawn() + .unwrap(); + let mut stdout = BufReader::new(shaper.stdout.take().unwrap()); + + // The first line says where it listens, and under what. + let mut line = String::new(); + stdout.read_line(&mut line).unwrap(); + assert!(line.ends_with("seed 7, profile mild\n"), "{line}"); + let addr: SocketAddr = line.split_whitespace().nth(1).unwrap().parse().unwrap(); + + let client = UdpSocket::bind("127.0.0.1:0").unwrap(); + for id in 0..50u32 { + client.send_to(&id.to_be_bytes(), addr).unwrap(); + } + target.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); + let mut buf = [0u8; 4]; + for _ in 0..50 { + target.recv_from(&mut buf).expect("a datagram never arrived"); + } + + // A line each interval, once the traffic has been counted. + let interim = loop { + line.clear(); + stdout.read_line(&mut line).unwrap(); + let json: serde_json::Value = serde_json::from_str(&line).unwrap(); + if json["up"]["packets"] == 50 { + break json; + } + }; + assert_eq!(interim["profile"], "mild"); + + let status = Command::new("kill") + .args(["-TERM", &shaper.id().to_string()]) + .status() + .unwrap(); + assert!(status.success()); + assert!(shaper.wait().unwrap().success(), "the profile never acted"); + + let written = std::fs::read(&report).expect("no report at exit"); + std::fs::remove_file(&report).unwrap(); + let json: serde_json::Value = serde_json::from_slice(&written).unwrap(); + assert_eq!(json["profile"], "mild"); + assert_eq!(json["seed"], 7); + assert_eq!(json["up"]["packets"], 50); + assert_eq!(json["up"]["delayed"], 50); + assert_eq!(json["up"]["lost"], 0); +} + +#[test] +fn a_zero_report_interval_is_refused_before_binding() { + let result = Command::new(env!("CARGO_BIN_EXE_moq-shaper")) + .args([ + "--listen", + "127.0.0.1:0", + "--target", + "127.0.0.1:9", + "--report-interval", + "0s", + ]) + .output() + .unwrap(); + assert!(!result.status.success()); + assert!(result.stdout.is_empty(), "bound before refusing the interval"); + assert!(String::from_utf8_lossy(&result.stderr).contains("report interval must be positive")); +} diff --git a/test/audio-quality/README.md b/test/audio-quality/README.md index d351a84bba..60bdb96da6 100644 --- a/test/audio-quality/README.md +++ b/test/audio-quality/README.md @@ -15,7 +15,7 @@ just test audio-quality --enforce # fail on the bud just test audio-quality --replays # the replays only, in seconds ``` -The matrix is codec x profile x ring: 16 browser rows, about 20 minutes, plus 6 [replay +The matrix is codec x profile x ring: 24 browser rows, about 30 minutes, plus 6 [replay rows](#traces) of recorded arrivals. `--profiles`, `--rings`, and `--codecs` take comma-separated lists; `--seed` replays a given impairment. The table prints either way, and only `--enforce` (which the nightly job passes) turns a breach into a failed run. @@ -35,17 +35,19 @@ decides which ring ran. | Profile | Path | Delay | | --- | --- | --- | | `near-zero` | through the shaper, untreated | auto | -| `mild` | 5 ms one way, plus or minus 5 ms uniform | auto | -| `wide` | 40 ms one way, plus or minus 40 ms uniform | auto | +| `mild` | 5 ms one way, 5 ms Gaussian sigma, in order | auto | +| `wide` | 40 ms one way, 40 ms Gaussian sigma, in order | auto | +| `bursty` | batches of seven datagrams, released within 160 ms | auto | +| `step` | 5 ms one way, stepping to 60 ms at 30 s | auto | | `fixed-250` | through the shaper, untreated | a fixed 250 ms | `fixed-250` is the control: every other row adapts, so without it the whole matrix could pass while a fixed preset regressed, and a fixed preset is what many viewers land on. Loss, reorder, and rate limits stay off: the buffer's job is absorbing arrival spread, and congestion response would make a -failure hard to attribute. The shaper's uniform jitter does reorder datagrams whose spacing is -smaller than it, so `wide` measures some reordering along with the spread. The bursty and mid-run -step profiles need shaper features that are not here yet -([quest](../../quest/m0/audio-quality-harness/shaper-profiles.md)). +failure hard to attribute. The `mild` and `wide` profiles preserve datagram order and share one path across the page's sessions. +`bursty` and `step` use the shaper's named profiles on that same shared path. The step is timed from +shaper startup, before the page connects; the driver gives up on a row with no audio 20 s in, so the +measured window starts before the step, and `--duration` must exceed 30 s so playback spans it. The AAC arm keeps ffmpeg's default PES packing, whose multi-frame bursts are the flush-span shape the reporter measured on the public relay (#3477). The source is a sine tone rather than a film, so a @@ -179,7 +181,7 @@ replay row crosses no shaper, so the `shaper` void does not apply to it. | `ring` | `crossOriginIsolated` does not match the row's ring, so the other ring ran. | | `clock` | The AudioContext clock was unreadable or ran more than 1% off the viewer's. | | `window` | No audio after the warmup. | -| `driver` | The page never reached a session or first audio within 30 s, or the driver threw. | +| `driver` | The page never reached a session and first audio within 20 s, or the driver threw. | ## Budgets @@ -220,9 +222,9 @@ clients/js/ The harness is the reporter's, from #3477: fperex built it on their fork `fperex/moq` (branches `debug/rt-audio` and `debug-findings-solution`), with the black-box probe, the analyzer's skip-ahead rule, the Playwright matrix over `moq-shaper`, the budget file graded under `--enforce`, and the -nightly job. This lane upstreams it on its own, without the fork's player, estimator, or shaper -changes: the underrun count comes from a tap on the output rather than a counter in the player, and -the profiles are the ones the shaper on `main` can express. +nightly job. The harness and opt-in shaper features preserve that author's credit. The player and estimator +changes remain separate: the underrun count comes from a tap on the output rather than a counter in the player, and +the profiles now include the fork's batch, step, and order-preserving jitter treatments. ## Not covered diff --git a/test/audio-quality/budgets.json b/test/audio-quality/budgets.json index 6d212be1f5..7a9bb8f4c3 100644 --- a/test/audio-quality/budgets.json +++ b/test/audio-quality/budgets.json @@ -1,6 +1,34 @@ { - "note": "Measured 2026-09-28 on a shared Linux workstation: two full 60 s runs and one 20 s run, seed 7. Each ceiling is twice the worse run, rounded up to two significant figures, with a floor (5 underruns/min, 20 gap ms/min, 1 skip/min, 100 discarded ms/min, 1% stalled, 1% silence) so one runner hiccup does not fail a clean row; target_ms_p95 is the worst measured plus one 20 ms bucket. The fixed-250 control measured zero on every counter. Every auto row records today's shallow target (#3477: 43 ms for Opus, underrunning thousands of quanta a minute), so its ceilings catch a regression past that floor and nothing better; the jitter target work is expected to tighten them. Not yet measured on the nightly runner, which has no audio device; re-record from its first runs. The replay rows are deterministic, so their ceilings are exactly what they measured on 2026-09-29, with no headroom: any regression fails them, and an improvement should be tightened in.", + "note": "Measured 2026-09-28 on a shared Linux workstation: two full 60 s runs and one 20 s run, seed 7. Each ceiling is twice the worse run, rounded up to two significant figures, with a floor (5 underruns/min, 20 gap ms/min, 1 skip/min, 100 discarded ms/min, 1% stalled, 1% silence) so one runner hiccup does not fail a clean row; target_ms_p95 is the worst measured plus one 20 ms bucket. The fixed-250 control measured zero on every counter. Every auto row records today's shallow target (#3477: 43 ms for Opus, underrunning thousands of quanta a minute), so its ceilings catch a regression past that floor and nothing better; the jitter target work is expected to tighten them. Not yet measured on the nightly runner, which has no audio device; re-record from its first runs. Updated 2026-09-29: mild and wide now use shared order-preserving Gaussian jitter; bursty and step are new. Those 16 rows were re-recorded for 60 s each, seed 7, on the same shared Linux workstation with concurrent bounded builds. Their ceilings use twice the measurement rounded upward to two significant figures and the same floors above; target_ms_p95 adds one 20 ms bucket. Other rows retain their previous measurements. These remain current-player regression ceilings, not target audio quality; re-record on the nightly runner before tightening them. The replay rows are deterministic, so their ceilings are exactly what they measured on 2026-09-29, with no headroom: any regression fails them, and an improvement should be tightened in.", "rows": [ + { + "runtime": "chromium", + "codec": "aac", + "rate": 44100, + "profile": "bursty", + "ring": "isolated", + "underruns_per_min": 1100.0, + "underrun_ms_per_min": 3000.0, + "skip_aheads_per_min": 6.6, + "discarded_ms_per_min": 500.0, + "stalled_share": 0.01, + "silence_share": 0.05, + "target_ms_p95": 404.8 + }, + { + "runtime": "chromium", + "codec": "aac", + "rate": 44100, + "profile": "bursty", + "ring": "plain", + "underruns_per_min": 740.0, + "underrun_ms_per_min": 2200.0, + "skip_aheads_per_min": 4.4, + "discarded_ms_per_min": 350.0, + "stalled_share": 0.01, + "silence_share": 0.034, + "target_ms_p95": 368.5 + }, { "runtime": "chromium", "codec": "aac", @@ -35,13 +63,13 @@ "rate": 44100, "profile": "mild", "ring": "isolated", - "underruns_per_min": 270, - "underrun_ms_per_min": 780, - "skip_aheads_per_min": 2.2, - "discarded_ms_per_min": 160, + "underruns_per_min": 320.0, + "underrun_ms_per_min": 890.0, + "skip_aheads_per_min": 1.0, + "discarded_ms_per_min": 100.0, "stalled_share": 0.01, - "silence_share": 0.012, - "target_ms_p95": 210 + "silence_share": 0.014, + "target_ms_p95": 211 }, { "runtime": "chromium", @@ -49,13 +77,13 @@ "rate": 44100, "profile": "mild", "ring": "plain", - "underruns_per_min": 570, - "underrun_ms_per_min": 1700, - "skip_aheads_per_min": 2.2, - "discarded_ms_per_min": 120, + "underruns_per_min": 340.0, + "underrun_ms_per_min": 970.0, + "skip_aheads_per_min": 1.0, + "discarded_ms_per_min": 100.0, "stalled_share": 0.01, - "silence_share": 0.026, - "target_ms_p95": 210 + "silence_share": 0.014, + "target_ms_p95": 206 }, { "runtime": "chromium", @@ -89,15 +117,43 @@ "runtime": "chromium", "codec": "aac", "rate": 44100, - "profile": "wide", + "profile": "step", "ring": "isolated", - "underruns_per_min": 430, - "underrun_ms_per_min": 1200, + "underruns_per_min": 970.0, + "underrun_ms_per_min": 2800.0, "skip_aheads_per_min": 6.6, - "discarded_ms_per_min": 330, + "discarded_ms_per_min": 380.0, "stalled_share": 0.01, - "silence_share": 0.018, - "target_ms_p95": 250 + "silence_share": 0.042, + "target_ms_p95": 207.3 + }, + { + "runtime": "chromium", + "codec": "aac", + "rate": 44100, + "profile": "step", + "ring": "plain", + "underruns_per_min": 450.0, + "underrun_ms_per_min": 1200.0, + "skip_aheads_per_min": 2.2, + "discarded_ms_per_min": 120.0, + "stalled_share": 0.01, + "silence_share": 0.02, + "target_ms_p95": 212.3 + }, + { + "runtime": "chromium", + "codec": "aac", + "rate": 44100, + "profile": "wide", + "ring": "isolated", + "underruns_per_min": 140.0, + "underrun_ms_per_min": 360.0, + "skip_aheads_per_min": 2.2, + "discarded_ms_per_min": 140.0, + "stalled_share": 0.01, + "silence_share": 0.01, + "target_ms_p95": 319.8 }, { "runtime": "chromium", @@ -105,13 +161,41 @@ "rate": 44100, "profile": "wide", "ring": "plain", - "underruns_per_min": 540, - "underrun_ms_per_min": 1500, - "skip_aheads_per_min": 4.4, - "discarded_ms_per_min": 260, + "underruns_per_min": 330.0, + "underrun_ms_per_min": 940.0, + "skip_aheads_per_min": 1.0, + "discarded_ms_per_min": 100.0, "stalled_share": 0.01, - "silence_share": 0.03, - "target_ms_p95": 250 + "silence_share": 0.018, + "target_ms_p95": 308.5 + }, + { + "runtime": "chromium", + "codec": "opus", + "rate": 48000, + "profile": "bursty", + "ring": "isolated", + "underruns_per_min": 4800.0, + "underrun_ms_per_min": 13000.0, + "skip_aheads_per_min": 8.8, + "discarded_ms_per_min": 490.0, + "stalled_share": 0.018, + "silence_share": 0.2, + "target_ms_p95": 128 + }, + { + "runtime": "chromium", + "codec": "opus", + "rate": 48000, + "profile": "bursty", + "ring": "plain", + "underruns_per_min": 2900.0, + "underrun_ms_per_min": 7100.0, + "skip_aheads_per_min": 6.6, + "discarded_ms_per_min": 470.0, + "stalled_share": 0.028, + "silence_share": 0.13, + "target_ms_p95": 139.3 }, { "runtime": "chromium", @@ -147,12 +231,12 @@ "rate": 48000, "profile": "mild", "ring": "isolated", - "underruns_per_min": 4600, - "underrun_ms_per_min": 12000, - "skip_aheads_per_min": 1, - "discarded_ms_per_min": 100, + "underruns_per_min": 1300.0, + "underrun_ms_per_min": 2600.0, + "skip_aheads_per_min": 1.0, + "discarded_ms_per_min": 100.0, "stalled_share": 0.01, - "silence_share": 0.19, + "silence_share": 0.026, "target_ms_p95": 63 }, { @@ -161,12 +245,12 @@ "rate": 48000, "profile": "mild", "ring": "plain", - "underruns_per_min": 4800, - "underrun_ms_per_min": 12000, - "skip_aheads_per_min": 2.2, - "discarded_ms_per_min": 130, - "stalled_share": 0.028, - "silence_share": 0.22, + "underruns_per_min": 1400.0, + "underrun_ms_per_min": 2900.0, + "skip_aheads_per_min": 1.0, + "discarded_ms_per_min": 100.0, + "stalled_share": 0.01, + "silence_share": 0.032, "target_ms_p95": 63 }, { @@ -197,19 +281,47 @@ "silence_share": 0.19, "target_ms_p95": 63 }, + { + "runtime": "chromium", + "codec": "opus", + "rate": 48000, + "profile": "step", + "ring": "isolated", + "underruns_per_min": 1300.0, + "underrun_ms_per_min": 2700.0, + "skip_aheads_per_min": 1.0, + "discarded_ms_per_min": 100.0, + "stalled_share": 0.01, + "silence_share": 0.034, + "target_ms_p95": 63 + }, + { + "runtime": "chromium", + "codec": "opus", + "rate": 48000, + "profile": "step", + "ring": "plain", + "underruns_per_min": 2500.0, + "underrun_ms_per_min": 5700.0, + "skip_aheads_per_min": 2.2, + "discarded_ms_per_min": 100.0, + "stalled_share": 0.01, + "silence_share": 0.084, + "target_ms_p95": 63 + }, { "runtime": "chromium", "codec": "opus", "rate": 48000, "profile": "wide", "ring": "isolated", - "underruns_per_min": 8100, - "underrun_ms_per_min": 22000, - "skip_aheads_per_min": 29, - "discarded_ms_per_min": 1700, - "stalled_share": 0.56, - "silence_share": 0.73, - "target_ms_p95": 82 + "underruns_per_min": 2200.0, + "underrun_ms_per_min": 5600.0, + "skip_aheads_per_min": 2.2, + "discarded_ms_per_min": 100.0, + "stalled_share": 0.01, + "silence_share": 0.092, + "target_ms_p95": 150.5 }, { "runtime": "chromium", @@ -217,13 +329,13 @@ "rate": 48000, "profile": "wide", "ring": "plain", - "underruns_per_min": 6700, - "underrun_ms_per_min": 17000, - "skip_aheads_per_min": 18, - "discarded_ms_per_min": 1300, - "stalled_share": 0.49, - "silence_share": 0.75, - "target_ms_p95": 82 + "underruns_per_min": 2300.0, + "underrun_ms_per_min": 5700.0, + "skip_aheads_per_min": 4.4, + "discarded_ms_per_min": 200.0, + "stalled_share": 0.01, + "silence_share": 0.09, + "target_ms_p95": 154.3 }, { "runtime": "replay", diff --git a/test/audio-quality/clients/js/driver.ts b/test/audio-quality/clients/js/driver.ts index 3fec53e203..e640a584f9 100644 --- a/test/audio-quality/clients/js/driver.ts +++ b/test/audio-quality/clients/js/driver.ts @@ -57,8 +57,12 @@ if ( process.exit(2); } -/** How long the session and the first audio each get before the row is abandoned. */ -const STARTUP_MS = 30_000; +/** + * How long the session and the first audio get, together, before the row is abandoned. It stays short + * of the step profile's change at 30s, so a step row always measures audio from before the change. + */ +const STARTUP_MS = 20_000; +const startupDeadline = Date.now() + STARTUP_MS; /** How often the probe is drained into the ndjson. */ const DRAIN_MS = 1000; @@ -98,8 +102,7 @@ const refuse = (assertion: string, detail: string) => { /** Poll the page until `ready` returns a value, or throw naming what never happened. */ async function waitFor(page: Page, what: string, ready: () => T | undefined): Promise { - const deadline = Date.now() + STARTUP_MS; - while (Date.now() < deadline) { + while (Date.now() < startupDeadline) { const value = await page.evaluate(ready); if (value !== undefined && value !== null) return value; await sleep(200); diff --git a/test/audio-quality/clients/js/src/run.test.ts b/test/audio-quality/clients/js/src/run.test.ts new file mode 100644 index 0000000000..13bda94a13 --- /dev/null +++ b/test/audio-quality/clients/js/src/run.test.ts @@ -0,0 +1,34 @@ +import { expect, test } from "bun:test"; +import { fileURLToPath } from "node:url"; + +const script = fileURLToPath(new URL("../../../run.sh", import.meta.url)); + +test("a step row must play beyond the profile change", () => { + for (const duration of ["20", "30"]) { + const result = Bun.spawnSync(["bash", script, "--profiles", "step", "--duration", duration, "--list"]); + expect(result.exitCode).toBe(2); + expect(result.stderr.toString()).toContain("must exceed the step at 30s"); + } + const result = Bun.spawnSync(["bash", script, "--profiles", "step", "--duration", "31", "--list"]); + expect(result.exitCode).toBe(0); + expect(result.stdout.toString()).toContain("chromium-opus-48000-step-plain"); +}); + +test("short steady rows remain supported", () => { + const result = Bun.spawnSync(["bash", script, "--profiles", "mild", "--duration", "20", "--list"]); + expect(result.exitCode).toBe(0); +}); + +test("the nightly matrix includes both new profiles", () => { + const result = Bun.spawnSync(["bash", script, "--list"]); + expect(result.exitCode).toBe(0); + const rows = result.stdout + .toString() + .trim() + .split("\n") + .filter((row) => row.startsWith("chromium-")); + expect(rows).toHaveLength(24); + for (const profile of ["bursty", "step"]) { + expect(rows.filter((row) => row.includes(`-${profile}-`))).toHaveLength(4); + } +}); diff --git a/test/audio-quality/run.sh b/test/audio-quality/run.sh index 1416a24dcd..7a71b3368a 100755 --- a/test/audio-quality/run.sh +++ b/test/audio-quality/run.sh @@ -15,7 +15,7 @@ source "$AQ_DIR/../lib/harness.sh" # Captured before the parse consumes it, so the rerun command carries every flag. RERUN="$(harness_env AQ_PROFILE RELAY_BIN)just test audio-quality$(harness_argv "$@")" -ALL_PROFILES=(near-zero mild wide fixed-250) +ALL_PROFILES=(near-zero mild wide bursty step fixed-250) ALL_RINGS=(isolated plain) ALL_CODECS=(opus aac) @@ -122,7 +122,13 @@ valid() { echo "error: unknown $name '$want' (known: $*)" >&2 exit 2 } -for p in "${PROFILES[@]}"; do valid "$p" profile "${ALL_PROFILES[@]}" "${ALL_TRACES[@]}"; done +for p in "${PROFILES[@]}"; do + valid "$p" profile "${ALL_PROFILES[@]}" "${ALL_TRACES[@]}" + if [[ "$p" == step ]] && ((DURATION <= 30)); then + echo "error: --duration must exceed the step at 30s" >&2 + exit 2 + fi +done for r in "${RINGS[@]}"; do valid "$r" ring "${ALL_RINGS[@]}"; done for c in "${CODECS[@]}"; do valid "$c" codec "${ALL_CODECS[@]}"; done @@ -152,8 +158,9 @@ delay_of() { shaper_of() { case "$1" in near-zero | fixed-250) ;; - mild) echo "--delay 5ms --jitter 5ms" ;; - wide) echo "--delay 40ms --jitter 40ms" ;; + mild) echo "--delay 5ms --jitter 5ms --jitter-model gaussian --shared" ;; + wide) echo "--delay 40ms --jitter 40ms --jitter-model gaussian --shared" ;; + bursty | step) echo "--profile $1" ;; esac } diff --git a/test/justfile b/test/justfile index 7a8c76c30d..ed16fc6191 100644 --- a/test/justfile +++ b/test/justfile @@ -77,7 +77,7 @@ drill *args: # Plays a tone in headless Chromium through moq-shaper and grades what the listener # heard (underruns, skip-aheads, silence, the resolved delay) against -# audio-quality/budgets.json. The full matrix is 16 rows at 60 s; nightly runs it with +# audio-quality/budgets.json. The full matrix is 24 rows at 60 s; nightly runs it with # --enforce. Narrow it with `--profiles mild --codecs opus --rings plain --duration 20`. # `--replays` grades only the recorded traces, in seconds, as PRs touching the player do. # See audio-quality/README.md.