diff --git a/quest/m0/nightly/README.md b/quest/m0/nightly/README.md index eb79b121e0d..3d0c5dea508 100644 --- a/quest/m0/nightly/README.md +++ b/quest/m0/nightly/README.md @@ -30,13 +30,16 @@ This README's own work, after the children: macOS `receive_credit` timeout has no owner since its quest was deleted in [moq-dev/moq#6732](https://github.com/moq-dev/moq/pull/6732), so plan one under [test flakes](/quest/m0/test-flakes-2/README.md) if it recurs, the paired bench - threshold to [bench CI](/quest/m1/bench-ci.md), and the moq-net bench - panics in "Rust bench targets" to - [moq-net's benches run to completion](/quest/m1/bench-test-mode.md). + threshold to [bench CI](/quest/m1/bench-ci.md). The moq-net benchmark + fixtures were repaired in [moq-dev/moq#7940](https://github.com/moq-dev/moq/pull/7940). The receive-windows benchmark's overcommitted cells, skipped since [moq-dev/moq#7838](https://github.com/moq-dev/moq/pull/7838), moved to [m1](/quest/m1/receive-windows-evicted.md) on 2026-11-04 (maintainer). +Decided 2026-11-05 (maintainer, /quest-plan): catch the next moq-net bench +break in Nightly's measured coverage only. Do not add a test-mode bench run +to the PR gate; the full run costs about 17 minutes per PR. + Public API: none. Wire: none. ## Required diff --git a/quest/m1/README.md b/quest/m1/README.md index f96e4ef9f14..35aec931bc7 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -174,7 +174,6 @@ their end report, so an embedder's drain alert pages on real cuts. - [Main is no slower than release](/quest/m1/release-perf/README.md) - every benchmark where main is slower than release is fixed or accepted with a reason, checked by a final paired run before the cut - [Calibrate the paired nightly benchmarks](/quest/m1/bench-ci.md) - A/A and deliberately slowed Nightly runs set the paired comparison's threshold, so its regression alerts can be trusted - [moq-json's table/deltas bench measures the same work every iteration](/quest/m1/bench-json-deltas.md) - the `table/deltas/1%` cases stop flipping between snapshot and delta timings, so their trend history is usable -- [moq-net's benches run to completion](/quest/m1/bench-test-mode.md) - the `origin`, `session` and `memory` benches stop panicking in test mode and Nightly's measured run, and the maintainer picks where CI catches the next break - [Publish moq-bench to crates.io](/quest/m1/bench-publish.md) - `cargo install moq-bench` installs one binary, with the host sampler as `moq-bench host`, first as 0.1.0 - [Catalog contributions become includes on the root's timeline](/quest/m1/catalog-include.md) - rename the unreleased `contributions` map before the cut and state that included broadcasts share the root's timeline - [Unbounded groups](/quest/m1/unbounded-groups/README.md) - groups and frames lose their size caps; a group keeps a 32 MB tail, and readers behind it replay through paced flights or refresh diff --git a/quest/m1/bench-test-mode.md b/quest/m1/bench-test-mode.md deleted file mode 100644 index 0c0952cfcc9..00000000000 --- a/quest/m1/bench-test-mode.md +++ /dev/null @@ -1,59 +0,0 @@ -# [S] moq-net's benches run to completion - -## Goal - -Every moq-net bench target runs to completion, both in Criterion's test mode -(`cargo test -p moq-net --features fuzz --benches`) and in Nightly's measured -run, so a bench broken by a behaviour change fails where someone sees it. -Done when that command passes on main, Nightly's "Rust bench targets" job gets -past its `criterion net` step, and the PR records where CI catches the next -such break. - -## Plan - -Planned 2026-11-04 with the maintainer. While iterating -[moq-dev/moq#7729](https://github.com/moq-dev/moq/pull/7729) (authored tracks -default to zero max age, merged 2026-11-04), its agent ran the Rust benches in -test mode: `group` panicked under the new default and was fixed there, while -`origin`, `session` and `track` in `rs/moq-net/benches/` panicked on main -before the change. - -Nightly already runs these targets measured: `sh/gh/nightly.sh`'s `rust-bench` -step runs `criterion net --no-fail-fast -p moq-net --features fuzz --bench '*'`. -Scheduled run -[41656613559](https://github.com/moq-dev/moq/actions/runs/41656613559) -(2026-11-03, before #7729) failed it on four targets: - -- `origin` (`origin.rs:574` and `:576`, `winner changed`) in - `origin/reprice_duplicate/1d_1s_cost2`. -- `session` (`session/mod.rs:371` and `:492`, `Evicted`) in - `session_delivery_burst/moq-transport-22/...`. -- `memory` (`memory.rs:194`, `Evicted`) in `memory/streams=16/size=1048576`. -- `track` (`track.rs:1718` and `:1730`) in `track_range_widening/1r/64`, fixed - by [moq-dev/moq#7840](https://github.com/moq-dev/moq/pull/7840) (merged - 2026-11-04). - -Start by re-running every moq-net bench target on main, in test mode and in a -short measured run, since #7729's zero max-age default may have broken more. -Fix each at its cause: a bench whose expectation predates a deliberate -behaviour change is updated; a bench exposing a product bug gets the product -fix. `origin` and `session` need `fuzz`, and Cargo skips a bench whose -`required-features` are off without a word, so a test-mode run without it -passes while they're broken. - -Where CI catches the next break is the maintainer's call; ask in the PR. A -test-mode run in Nightly adds nothing, since the measured run covers the same -targets. Today the PR gate builds benches with their features but leaves them -unrun (`rs/justfile` above the `test` recipe; `.config/nextest.toml`'s -`default-filter = "not kind(bench)"`). Running them there, which would have -caught these before merge, reverses the -[Nightly epic](/quest/m0/nightly/README.md)'s 2026-10-24 decision that the -gate keeps skipping benches and a green Nightly catches them within a day. -Measure the test-mode run's cost before asking. - -Public API: none. Wire: none. - -## Related - -- [Nightly passes on main](/quest/m0/nightly/README.md) - this quest owns the moq-net bench panics in Nightly's "Rust bench targets" job -- [Calibrate the paired nightly benchmarks](/quest/m1/bench-ci.md) - calibrates the measurements; this quest only keeps the bench executables correct diff --git a/rs/moq-net/benches/memory.rs b/rs/moq-net/benches/memory.rs index 73d84991de7..fd5361c1022 100644 --- a/rs/moq-net/benches/memory.rs +++ b/rs/moq-net/benches/memory.rs @@ -100,6 +100,8 @@ fn now() -> moq_net::time::Instant { /// One publisher and one subscriber session over an in-memory pair. struct Link { + pools: Vec, + cache_now: moq_net::time::Instant, tracks: Vec, subscribers: Vec, payload: Bytes, @@ -110,12 +112,12 @@ struct Link { impl Link { async fn new(streams: usize, size: usize) -> Self { + let mut pools = Vec::new(); let version: Version = VERSION.parse().unwrap(); - let origin = |hop: u64| { + let mut origin = |hop: u64| { let mut config = origin::Config::new(Hop::new(hop).unwrap()); - // Hold one round's groups, with room for the handover, and no more. - config.pool = - cache::Pool::new(cache::Config::default().with_capacity((4 * streams * FRAMES * size) as u64)); + config.pool = cache::Pool::new(cache::Config::default().with_expiry(cache::DEFAULT_EXPIRY)); + pools.push(config.pool.clone()); let (producer, driver) = origin::Producer::new(config); tokio::spawn(moq_net::time::run(driver)); producer @@ -171,6 +173,8 @@ impl Link { } let mut link = Self { + pools, + cache_now: now(), tracks, subscribers, payload: Bytes::from(vec![0; size]), @@ -180,6 +184,7 @@ impl Link { }; // Warm every stream so the timed rounds skip first-group setup. for _ in 0..4 { + link.tick(); link.round().await; } link @@ -206,13 +211,31 @@ impl Link { bytes } + // Advance the cache/model clock between rounds to expire read history. + fn tick(&mut self) { + self.cache_now += cache::DEFAULT_EXPIRY; + for pool in &self.pools { + pool.poll(self.cache_now, &kio::Waiter::noop()); + } + } + async fn measure(&mut self, iters: u64) -> Duration { let expected = self.tracks.len() * FRAMES * self.payload.len(); - let start = Instant::now(); + let counting = std::env::var_os("MEMORY_ALLOCS").is_some(); + let mut elapsed = Duration::ZERO; for _ in 0..iters { + self.tick(); + if counting { + COUNTING.store(true, Ordering::Relaxed); + } + let start = Instant::now(); assert_eq!(self.round().await, expected); + elapsed += start.elapsed(); + if counting { + COUNTING.store(false, Ordering::Relaxed); + } } - start.elapsed() + elapsed } } @@ -225,9 +248,7 @@ fn allocations() { let mut link = rt.block_on(Link::new(streams, size)); ALLOCATIONS.store(0, Ordering::Relaxed); ALLOCATED.store(0, Ordering::Relaxed); - COUNTING.store(true, Ordering::Relaxed); rt.block_on(link.measure(ROUNDS as u64)); - COUNTING.store(false, Ordering::Relaxed); let frames = (ROUNDS * streams * FRAMES) as f64; let allocations = ALLOCATIONS.load(Ordering::Relaxed) as f64 / frames; diff --git a/rs/moq-net/benches/origin.rs b/rs/moq-net/benches/origin.rs index 506f86071a0..4fa1bbef981 100644 --- a/rs/moq-net/benches/origin.rs +++ b/rs/moq-net/benches/origin.rs @@ -564,6 +564,7 @@ fn bench_reprice_duplicate(c: &mut Criterion) { ) .unwrap() .with_hidden(true) + .covering() .routes() }) .collect(); diff --git a/rs/moq-net/benches/session/mod.rs b/rs/moq-net/benches/session/mod.rs index 761f544dc61..1c50fba8cce 100644 --- a/rs/moq-net/benches/session/mod.rs +++ b/rs/moq-net/benches/session/mod.rs @@ -34,9 +34,11 @@ //! one datagram per track per round across publisher and viewer axes, then larger payloads //! and cadences, on the lite versions (moq-transport carries no datagram tracks). Throughput //! counts delivered datagrams, each checked for its sequence and payload, so a lost one fails. -//! `SESSION_EXPIRY_ONLY=1` uses the standalone 30s-expiry pool and reports groups -//! written plus charged cache bytes every 10s during long profiling runs. Each round -//! advances the cache clock by a second, so the expiry window spans 30 rounds. +//! Delivery uses expiry-only pools: history expires between rounds outside timing, +//! so byte pressure cannot shed unread frames during a batch. +//! `SESSION_EXPIRY_ONLY=1` advances the cache clock by a second instead of the 30s +//! expiry window, retaining 30 rounds for profiling. It reports groups written and +//! charged cache bytes every 10s during long runs. //! //! `session_join_*` times a new viewer connecting, resolving a broadcast, and //! receiving its latest group, swept over what the relays already announce. @@ -123,9 +125,7 @@ const DASH_VERSIONS: [&str; 2] = ["moq-lite-06", "moq-transport-22"]; /// A stats frame: a small JSON snapshot. const DASH_FRAME: usize = 128; -/// Each endpoint gets its own bounded pool, as a relay configures one per -/// process: without a byte target, every hop keeps each group for the whole -/// expiry window and memory grows with the run length instead of the shape. +/// Initial budgets for join and churn fixtures. Delivery rooms use expiry-only pools. const RELAY_CACHE: u64 = 16 * 1024 * 1024; const CLIENT_CACHE: u64 = 1024 * 1024; @@ -517,7 +517,7 @@ pub(crate) trait Backend { type Relay: Clone; const PREFIX: &'static str; - fn relay(id: u64, capacity: u64) -> (Self::Relay, cache::Pool); + fn relay(id: u64, capacity: Option) -> (Self::Relay, cache::Pool); fn origin(relay: &Self::Relay) -> &origin::Producer; async fn connect(options: MockConnectOptions, relay: &Self::Relay, hidden: bool) -> Pair; } @@ -529,7 +529,7 @@ impl Backend for Model { type Relay = origin::Producer; const PREFIX: &'static str = ""; - fn relay(id: u64, capacity: u64) -> (Self::Relay, cache::Pool) { + fn relay(id: u64, capacity: Option) -> (Self::Relay, cache::Pool) { make_origin(id, capacity) } @@ -566,9 +566,10 @@ impl Drop for Pair { } } -fn make_origin(id: u64, capacity: u64) -> (origin::Producer, cache::Pool) { +fn make_origin(id: u64, capacity: Option) -> (origin::Producer, cache::Pool) { let mut config = origin::Config::new(Hop::new(id).unwrap()); - let pool = cache::Pool::new(if std::env::var_os("SESSION_EXPIRY_ONLY").is_some() { + let profiling = std::env::var_os("SESSION_EXPIRY_ONLY").is_some(); + let pool = cache::Pool::new(if capacity.is_none() || profiling { cache::Config::default().with_expiry(cache::DEFAULT_EXPIRY) } else { cache::Config::default().with_capacity(capacity) @@ -582,6 +583,7 @@ fn make_origin(id: u64, capacity: u64) -> (origin::Producer, cache::Pool) { /// Relays and publishers, before any viewer joins. Holds every handle that /// keeps a session or broadcast alive. struct Cluster { + expiry: bool, version: Version, window: usize, credit: Option, @@ -601,8 +603,9 @@ struct Cluster { } impl Cluster { - async fn new(version: &str, shape: Shape) -> Self { + async fn new(version: &str, shape: Shape, expiry: bool) -> Self { let mut this = Self { + expiry, version: version.parse().unwrap(), window: shape.window, credit: shape.credit, @@ -620,7 +623,7 @@ impl Cluster { for _ in 0..shape.relays { this.next_hop += 1; - let (relay, pool) = B::relay(this.next_hop, RELAY_CACHE); + let (relay, pool) = B::relay(this.next_hop, (!expiry).then_some(RELAY_CACHE)); this.relays.push(relay); this.pools.push(pool); } @@ -678,7 +681,7 @@ impl Cluster { fn origin(&mut self, capacity: u64) -> (origin::Producer, cache::Pool) { self.next_hop += 1; - make_origin(self.next_hop, capacity) + make_origin(self.next_hop, (!self.expiry).then_some(capacity)) } /// Connect a viewer session to `relay`. @@ -810,7 +813,7 @@ impl Room { ); assert!(!shape.paced || shape.burst == 1, "a paced round writes one frame"); assert!(!shape.fetch || !shape.paced, "fetch rounds read finished groups"); - let mut cluster = Cluster::::new(version, shape).await; + let mut cluster = Cluster::::new(version, shape, true).await; let mut viewers = Vec::new(); let mut watched = vec![false; shape.total()]; @@ -879,12 +882,6 @@ impl Room { room.writing.resize_with(room.watched.len(), || None); room.reading.resize_with(shape.viewers * shape.watch, || None); - // First measure the complete batch without eviction. The delivery benchmark must - // fit its working set; cache-pressure loss is a different workload. - let capacities: Vec<_> = room.pools().map(cache::Pool::capacity).collect(); - for pool in room.pools() { - pool.resize(None); - } let blocked_before: usize = room .viewers .iter() @@ -898,11 +895,6 @@ impl Room { room.tick(); room.round().await; } - for (pool, capacity) in room.pools().zip(capacities) { - // Keep the old and current batch resident during the handover, including - // bookkeeping rather than estimating a group from payload bytes alone. - pool.resize(capacity.map(|capacity| capacity.max(pool.used() * 2))); - } // Exercise turnover as well as the empty cache, including Criterion's one-pass // nightly smoke mode. for _ in 0..4 { @@ -945,11 +937,15 @@ impl Room { self.cluster.pools.iter().chain(viewers) } - /// Unpaced rounds otherwise collapse many complete batches into one cache tick, - /// where yesterday's read outranks today's unread write. Sample a new recency - /// epoch per round without sleeping or changing protocol time. + /// Advance the cache/model clock outside timing to expire old batches. + /// Each round completes its reads before this step; pending read deadlines + /// share this clock too. fn tick(&mut self) { - self.cache_now += Duration::from_secs(1); + self.cache_now += if std::env::var_os("SESSION_EXPIRY_ONLY").is_some() { + Duration::from_secs(1) + } else { + cache::DEFAULT_EXPIRY + }; for pool in self.pools() { pool.poll(self.cache_now, &kio::Waiter::noop()); } @@ -1315,6 +1311,7 @@ pub(crate) fn range_updates(c: &mut Criterion) { publishers: 1, ..Shape::BASE }, + false, ) .await; let mut viewers = Vec::new(); @@ -1407,7 +1404,7 @@ fn join(c: &mut Criterion, name: &str, shapes: impl IntoIterator::new(version, *shape).await; + let cluster = Cluster::::new(version, *shape, false).await; let payload = Bytes::from(vec![0; shape.frame]); for track in &cluster.tracks { for sequence in 0..shape.history { @@ -1503,6 +1500,7 @@ impl DashRoom { viewers: 0, ..Shape::BASE }, + false, ) .await; @@ -1666,6 +1664,7 @@ impl ChurnRoom { viewers: 0, ..Shape::BASE }, + false, ) .await; let published = &cluster._broadcasts[0]; @@ -1881,7 +1880,7 @@ pub(crate) fn withdrawal(c: &mut Criterion) { publishers: 0, ..Shape::BASE }; - let cluster = Cluster::::new("moq-lite-06", shape).await; + let cluster = Cluster::::new("moq-lite-06", shape, false).await; let mut announced = B::origin(cluster.relays.last().unwrap()).consume().routes(); let live: Vec<_> = (0..broadcasts) .map(|i| { diff --git a/rs/moq-relay/benches/session.rs b/rs/moq-relay/benches/session.rs index d918a61cef7..bfa7b429a41 100644 --- a/rs/moq-relay/benches/session.rs +++ b/rs/moq-relay/benches/session.rs @@ -68,10 +68,10 @@ impl Backend for Relay { type Relay = Node; const PREFIX: &'static str = "relay_"; - fn relay(id: u64, capacity: u64) -> (Node, cache::Pool) { + fn relay(id: u64, capacity: Option) -> (Node, cache::Pool) { let mut config = moq_relay::cache::Config::default(); if std::env::var_os("SESSION_EXPIRY_ONLY").is_none() { - config.capacity = Some(capacity.to_string()); + config.capacity = capacity.map(|capacity| capacity.to_string()); } let cache = config.init().unwrap(); let pool = cache.pool.clone();