From ef2b37cd4164cac5cfa2862d25a881c433528516 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 09:01:10 -0700 Subject: [PATCH 1/5] chore(quest): claim bench-session Co-Authored-By: Claude Opus 5.5 From f6cfca246613631e72905bc304154b7c82483071 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 09:45:03 -0700 Subject: [PATCH 2/5] feat(moq-net): sans-IO session benchmark over a relay mesh Co-Authored-By: Claude Opus 5.5 --- quest/m1/README.md | 1 - quest/m1/bench-ci.md | 1 - quest/m1/bench-relay.md | 8 +- quest/m1/bench-session.md | 31 -- rs/moq-net/Cargo.toml | 4 + rs/moq-net/benches/session.rs | 409 ++++++++++++++++++++ rs/moq-net/tests/announce_to_serve.rs | 2 +- rs/moq-net/tests/datagram.rs | 4 +- rs/moq-net/tests/finished_broadcast_mock.rs | 2 +- rs/moq-net/tests/goaway.rs | 4 +- rs/moq-net/tests/rejoin.rs | 2 +- rs/moq-net/tests/support/harness.rs | 4 +- 12 files changed, 424 insertions(+), 48 deletions(-) delete mode 100644 quest/m1/bench-session.md create mode 100644 rs/moq-net/benches/session.rs diff --git a/quest/m1/README.md b/quest/m1/README.md index fee11fb135..cf72bbcce6 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -79,7 +79,6 @@ transport, benchmark tooling); worktrees isolate commits, not semantics. - [Benchmark regressions in CI](/quest/m1/bench-ci.md) - PRs get a non-blocking comparison of the Criterion benches they affect, and a nightly trend on main alerts on regressions - [Benchmark comparisons](/quest/m1/performance-comparisons.md) - retained evidence, repeated paired runs, and uncertainty for performance claims - [#3126](/quest/m1/3126-moq-bench-every-readme-example-fails-to-parse-and.md) - moq-bench reports per-interval latency percentiles so the ramp leaves the steady state -- [Sans-IO session bench](/quest/m1/bench-session.md) - a moq-net bench drives publisher, relay, and subscribers over the in-memory transport, swept over publishers, subscribers, and frame size - [Relay session bench](/quest/m1/bench-relay.md) - the same scenario through moq-relay's own connection handling - [Bench coverage](/quest/m1/bench-coverage.md) - Criterion targets for moq-mux containers, the hang catalog, moq-auth verification, and moq-pattern matching - [Relay profiling](/quest/m1/performance-profiles.md) - reproducible CPU and allocation captures under the existing workloads diff --git a/quest/m1/bench-ci.md b/quest/m1/bench-ci.md index 3514730d8a..b317185d53 100644 --- a/quest/m1/bench-ci.md +++ b/quest/m1/bench-ci.md @@ -50,5 +50,4 @@ a GitHub App: ## Related - [Benchmark comparisons](/quest/m1/performance-comparisons.md) - extends the same `bench/run.sh` with repeated paired rounds -- [Sans-IO session bench](/quest/m1/bench-session.md) - a low-noise end-to-end bench this job picks up - [Bench coverage](/quest/m1/bench-coverage.md) - more targets for this job to track diff --git a/quest/m1/bench-relay.md b/quest/m1/bench-relay.md index 596cddb10e..1cd12ddfe9 100644 --- a/quest/m1/bench-relay.md +++ b/quest/m1/bench-relay.md @@ -16,9 +16,5 @@ transport session without widening the public API. If the seam costs more than the bench is worth, say so and stop. Reuse the scenario, the publisher and subscriber sweeps, and the delivery -accounting from the `moq-net` bench so the two results line up, and the -difference is the relay layer. - -## Required - -- [Sans-IO session bench](/quest/m1/bench-session.md) - supplies the scenario and harness this bench reuses +accounting from `rs/moq-net/benches/session.rs` so the two results line up, and +the difference is the relay layer. diff --git a/quest/m1/bench-session.md b/quest/m1/bench-session.md deleted file mode 100644 index 82aa479d22..0000000000 --- a/quest/m1/bench-session.md +++ /dev/null @@ -1,31 +0,0 @@ -# [M] Sans-IO session benchmark in moq-net - -## Goal - -A Criterion bench in `moq-net` measures a full publishers to relay to -subscribers path with no sockets: M publishing clients, relay sessions that -forward through an origin the way `moq-relay` does, and N subscribing -clients, all over the in-memory transport. It sweeps publishers, subscribers, -and frame size independently, over lite and IETF, so a per-publisher, -per-subscriber, or per-frame cost shows as a slope and CI can compare it with -little noise. - -## Plan - -`rs/moq-net/tests/support/{mock,harness}.rs` already pairs sessions over an -in-memory WebTransport mock and runs the full handshake. Reuse it from the -bench, as `moq-uring`'s benches reuse their test support, rather than -exporting a mock from the crate. - -- Keep setup (handshakes, announce, the subscribe round trip) outside the - timed region. Time delivery of groups until every subscriber has received - every frame, and count delivered bytes so a skipped delivery can't look like - a speedup. -- Also time subscriber join, since a relay pays that per viewer. -- Report throughput in bytes and frames. -- Pick sweep points that finish quickly enough to run on every `moq-net` PR. - -## Related - -- [Relay session bench](/quest/m1/bench-relay.md) - the same shape through the real `moq-relay` handler -- [Benchmark regressions in CI](/quest/m1/bench-ci.md) - tracks this bench on PRs and nightly diff --git a/rs/moq-net/Cargo.toml b/rs/moq-net/Cargo.toml index 1dade572a1..288ac36bb2 100644 --- a/rs/moq-net/Cargo.toml +++ b/rs/moq-net/Cargo.toml @@ -84,3 +84,7 @@ harness = false [[bench]] name = "stats" harness = false + +[[bench]] +name = "session" +harness = false diff --git a/rs/moq-net/benches/session.rs b/rs/moq-net/benches/session.rs new file mode 100644 index 0000000000..6c511ff1e0 --- /dev/null +++ b/rs/moq-net/benches/session.rs @@ -0,0 +1,409 @@ +//! End-to-end session benchmarks with no sockets: publishing clients, a mesh of +//! relays that forward through their origins the way `moq-relay` does, and +//! viewing clients, all connected over the in-memory mock transport. +//! +//! `session_delivery_*` times one round: every watched broadcast writes a group +//! and every viewer reads one group per broadcast it watches. Each group sweeps +//! one or two axes of a [`Shape`] (relays, publisher sessions, broadcasts per +//! publisher, viewers, broadcasts per viewer, frame size) with the rest held +//! fixed, so a cost that grows with a table instead of the touched path shows +//! as a slope. Unwatched broadcasts stay announced and silent: their cost is the +//! route table they occupy, not a publisher writing into its own cache. +//! +//! `session_join_*` times a new viewer connecting, resolving a broadcast, and +//! receiving its latest group, swept over what the relays already announce. +//! +//! Everything runs on one current-thread runtime, so the only work measured is +//! the protocol and model code, never scheduling across threads. +//! +//! Run with `cargo bench -p moq-net --bench session`. + +#[path = "../tests/support/mod.rs"] +mod support; + +use std::time::{Duration, Instant}; + +use bytes::Bytes; +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use moq_net::{Hop, Timestamp, Version, broadcast, cache, origin, track}; +use support::harness::{MockConnectOptions, MockPair, connect_mock}; + +/// The newest moq-lite and IETF drafts. +const VERSIONS: [&str; 2] = ["moq-lite-07", "moq-transport-22"]; + +/// Frames per group, so per-frame and per-group costs both appear. +const FRAMES: usize = 4; + +const TRACK: &str = "video"; + +/// One bounded pool for every origin, as a relay configures: 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. +const CACHE_CAPACITY: u64 = 16 * 1024 * 1024; + +/// One topology: publishers and viewers spread round-robin over a full mesh of +/// relays. +#[derive(Clone, Copy)] +struct Shape { + /// Relays, each peered with every other. + relays: usize, + /// Publisher sessions. + publishers: usize, + /// Broadcasts each publisher session announces. + broadcasts: usize, + /// Viewer sessions. + viewers: usize, + /// Broadcasts each viewer subscribes to. + watch: usize, + /// Payload bytes per frame. + frame: usize, +} + +impl Shape { + /// A 16-publisher, 16-viewer room on one relay, each viewer watching one + /// broadcast with small frames, so sweeps measure per-message costs. + const BASE: Self = Self { + relays: 1, + publishers: 16, + broadcasts: 1, + viewers: 16, + watch: 1, + frame: 64, + }; + + fn total(&self) -> usize { + self.publishers * self.broadcasts + } + + /// Payload bytes every viewer reads in one round. + fn expected(&self) -> usize { + self.viewers * self.watch * FRAMES * self.frame + } + + fn id(&self, version: &str) -> String { + format!( + "{version}/relays={}/publishers={}/broadcasts={}/viewers={}/watch={}/frame={}", + self.relays, self.publishers, self.broadcasts, self.viewers, self.watch, self.frame + ) + } +} + +fn runtime() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_current_thread().enable_time().build().unwrap() +} + +fn write_group(track: &track::Producer, payload: &Bytes) { + let mut group = track.append_group().unwrap(); + for _ in 0..FRAMES { + group.write_frame(Timestamp::ZERO, payload.clone()).unwrap(); + } + group.finish().unwrap(); +} + +/// Read one whole group, returning its payload bytes so skipped work can't look +/// like a speedup. +async fn read_group(subscriber: &mut track::Subscriber) -> usize { + let mut group = subscriber.recv_group().await.unwrap().expect("track ended"); + let mut bytes = 0; + while let Some(frame) = group.read_frame().await.unwrap() { + bytes += frame.payload.len(); + } + bytes +} + +fn path(broadcast: usize) -> String { + format!("room/{broadcast}") +} + +/// Relays and publishers, before any viewer joins. Holds every handle that +/// keeps a session or broadcast alive. +struct Cluster { + version: Version, + relays: Vec, + /// One track per broadcast, indexed like [`path`]. + tracks: Vec, + _broadcasts: Vec, + _origins: Vec, + _pairs: Vec, + pool: cache::Pool, + next_hop: u64, +} + +impl Cluster { + async fn new(version: &str, shape: Shape) -> Self { + let mut this = Self { + version: version.parse().unwrap(), + relays: Vec::new(), + tracks: Vec::new(), + _broadcasts: Vec::new(), + _origins: Vec::new(), + _pairs: Vec::new(), + pool: cache::Pool::new(cache::Config::default().with_capacity(CACHE_CAPACITY)), + next_hop: 0, + }; + + for _ in 0..shape.relays { + let relay = this.origin(); + this.relays.push(relay); + } + + // Peer every pair the way a cluster dial does: each side publishes its + // whole origin, hidden broadcasts included, and subscribes into it. + for server in 0..shape.relays { + for client in 0..server { + let client = this.relays[client].clone().peer(); + let server = this.relays[server].clone().peer(); + let mut options = MockConnectOptions::new(this.version); + options.client_publish = Some(client.consume().with_hidden(true)); + options.client_subscribe = Some(client); + options.server_publish = Some(server.consume().with_hidden(true)); + options.server_subscribe = Some(server); + this._pairs.push(connect_mock(options).await); + } + } + + for publisher in 0..shape.publishers { + let origin = this.origin(); + for index in 0..shape.broadcasts { + let broadcast = origin + .publish(path(publisher * shape.broadcasts + index), Default::default()) + .unwrap(); + this.tracks.push(broadcast.create_track(TRACK, None).unwrap()); + this._broadcasts.push(broadcast); + } + + let mut options = MockConnectOptions::new(this.version); + options.client_publish = Some(origin.consume()); + options.server_subscribe = Some(this.relays[publisher % shape.relays].clone()); + this._pairs.push(connect_mock(options).await); + this._origins.push(origin); + } + + this + } + + fn origin(&mut self) -> origin::Producer { + self.next_hop += 1; + let mut config = origin::Config::new(Hop::new(self.next_hop).unwrap()); + config.pool = self.pool.clone(); + let (producer, driver) = origin::Producer::new(config); + tokio::spawn(support::harness::run(driver)); + producer + } + + /// Connect a viewer to `relay` and subscribe it to each of `broadcasts`. + async fn join(&mut self, relay: usize, broadcasts: impl Iterator) -> Viewer { + let origin = self.origin(); + let mut options = MockConnectOptions::new(self.version); + options.server_publish = Some(self.relays[relay].consume()); + options.client_subscribe = Some(origin.clone()); + let pair = connect_mock(options).await; + + let mut subscribers = Vec::new(); + for broadcast in broadcasts { + let broadcast = origin.consume().routed_broadcast(path(broadcast)).await.unwrap(); + subscribers.push(broadcast.track(TRACK).unwrap().subscribe(None).await.unwrap()); + } + + Viewer { + subscribers, + _pair: pair, + _origin: origin, + } + } +} + +struct Viewer { + subscribers: Vec, + _pair: MockPair, + _origin: origin::Producer, +} + +/// A cluster with its viewers attached. +struct Room { + cluster: Cluster, + viewers: Vec, + /// Broadcasts at least one viewer watches; only these write each round. + watched: Vec, + payload: Bytes, + expected: usize, +} + +impl Room { + async fn new(version: &str, shape: Shape) -> Self { + assert!(shape.watch <= shape.total(), "a viewer can't watch more broadcasts than exist"); + let mut cluster = Cluster::new(version, shape).await; + + // Viewer v watches a contiguous window starting at v * watch, so viewers + // spread over the broadcasts before any doubles up. + let mut viewers = Vec::new(); + let mut watched = vec![false; shape.total()]; + for viewer in 0..shape.viewers { + let broadcasts: Vec<_> = (0..shape.watch) + .map(|k| (viewer * shape.watch + k) % shape.total()) + .collect(); + for &broadcast in &broadcasts { + watched[broadcast] = true; + } + // Offset by one so a viewer lands on a different relay than the + // publisher it watches whenever there is more than one. + let relay = (viewer + 1) % shape.relays; + viewers.push(cluster.join(relay, broadcasts.into_iter()).await); + } + + let mut room = Self { + cluster, + viewers, + watched: watched.iter().enumerate().filter(|(_, w)| **w).map(|(i, _)| i).collect(), + payload: Bytes::from(vec![0; shape.frame]), + expected: shape.expected(), + }; + // Warm every path so the timed rounds skip first-group setup. + room.round().await; + room + } + + async fn round(&mut self) { + for &broadcast in &self.watched { + write_group(&self.cluster.tracks[broadcast], &self.payload); + } + let mut bytes = 0; + for viewer in &mut self.viewers { + for subscriber in &mut viewer.subscribers { + bytes += read_group(subscriber).await; + } + } + assert_eq!(bytes, self.expected); + } + + async fn measure(&mut self, iters: u64) -> Duration { + let start = Instant::now(); + for _ in 0..iters { + self.round().await; + } + start.elapsed() + } +} + +fn delivery(c: &mut Criterion, name: &str, shapes: impl IntoIterator) { + let rt = runtime(); + let shapes: Vec<_> = shapes.into_iter().collect(); + let mut group = c.benchmark_group(format!("session_delivery_{name}")); + for version in VERSIONS { + for shape in &shapes { + // Built on first call: Criterion only calls a routine its filter + // selects, and calls it again for every sample. + let mut room = None; + group.throughput(Throughput::Bytes(shape.expected() as u64)); + group.bench_function(BenchmarkId::from_parameter(shape.id(version)), |b| { + let room = room.get_or_insert_with(|| rt.block_on(Room::new(version, *shape))); + b.iter_custom(|iters| rt.block_on(room.measure(iters))) + }); + } + } + group.finish(); +} + +fn join(c: &mut Criterion, name: &str, shapes: impl IntoIterator) { + let rt = runtime(); + let shapes: Vec<_> = shapes.into_iter().collect(); + let mut group = c.benchmark_group(format!("session_join_{name}")); + for version in VERSIONS { + for shape in &shapes { + let mut cluster = None; + group.bench_function(BenchmarkId::from_parameter(shape.id(version)), |b| { + let cluster = cluster.get_or_insert_with(|| { + rt.block_on(async { + let cluster = Cluster::new(version, *shape).await; + let payload = Bytes::from(vec![0; shape.frame]); + for track in &cluster.tracks { + write_group(track, &payload); + } + cluster + }) + }); + b.iter_custom(|iters| { + rt.block_on(async { + let mut elapsed = Duration::ZERO; + for iter in 0..iters as usize { + let broadcast = iter % shape.total(); + // The last relay, so a mesh resolves across a peer. + let relay = shape.relays - 1; + let start = Instant::now(); + let mut viewer = cluster.join(relay, std::iter::once(broadcast)).await; + let bytes = read_group(&mut viewer.subscribers[0]).await; + elapsed += start.elapsed(); + assert_eq!(bytes, FRAMES * shape.frame); + // Teardown is a leaving viewer's cost, not a joining one's. + drop(viewer); + tokio::task::yield_now().await; + } + elapsed + }) + }) + }); + } + } + group.finish(); +} + +fn session(c: &mut Criterion) { + let base = Shape::BASE; + + delivery(c, "publishers", [1, 16, 256].map(|publishers| Shape { publishers, ..base })); + delivery(c, "viewers", [1, 16, 256].map(|viewers| Shape { viewers, ..base })); + delivery( + c, + "scale", + [16, 64, 256].map(|n| Shape { + publishers: n, + viewers: n, + ..base + }), + ); + // One publisher session announcing many broadcasts, most of them unwatched. + delivery( + c, + "broadcasts", + [16, 256, 4096].map(|broadcasts| Shape { + publishers: 1, + broadcasts, + ..base + }), + ); + // Each viewer watches many broadcasts over one session, like a conference. + delivery( + c, + "watch", + [1, 16, 256].map(|watch| Shape { + publishers: 1, + broadcasts: 256, + watch, + ..base + }), + ); + delivery(c, "frame", [64, 1024, 16 * 1024].map(|frame| Shape { frame, ..base })); + delivery( + c, + "relays", + [1, 2, 4, 8].map(|relays| Shape { + relays, + viewers: 64, + ..base + }), + ); + + join( + c, + "broadcasts", + [1, 64, 1024].map(|broadcasts| Shape { + publishers: 1, + broadcasts, + ..base + }), + ); + join(c, "relays", [1, 2, 8].map(|relays| Shape { relays, ..base })); +} + +criterion_group!(benches, session); +criterion_main!(benches); diff --git a/rs/moq-net/tests/announce_to_serve.rs b/rs/moq-net/tests/announce_to_serve.rs index 6eb4b57eb6..8fd27f86d8 100644 --- a/rs/moq-net/tests/announce_to_serve.rs +++ b/rs/moq-net/tests/announce_to_serve.rs @@ -119,7 +119,7 @@ async fn lifecycle(observer: Observer) -> Vec { Observer::Remote(version) => { let subscriber = produce_origin(2); let mut options = MockConnectOptions::new(version.parse::().unwrap()); - options.server_publish = Some(publisher.clone()); + options.server_publish = Some(publisher.consume()); options.client_subscribe = Some(subscriber.clone()); let pair = connect_mock(options).await; (subscriber.consume(), Some(pair)) diff --git a/rs/moq-net/tests/datagram.rs b/rs/moq-net/tests/datagram.rs index b16264be9b..4cc78b1297 100644 --- a/rs/moq-net/tests/datagram.rs +++ b/rs/moq-net/tests/datagram.rs @@ -43,7 +43,7 @@ async fn connect_datagram_track() -> Fixture { broadcast.announce(Default::default()).unwrap(); let mut options = MockConnectOptions::new("moq-lite-05".parse::().unwrap()); - options.server_publish = Some(publisher); + options.server_publish = Some(publisher.consume()); options.client_subscribe = Some(consumer_origin.clone()); let pair = connect_mock(options).await; @@ -122,7 +122,7 @@ async fn ietf_does_not_deliver_datagrams() { broadcast.announce(Default::default()).unwrap(); let mut options = MockConnectOptions::new("moq-transport-19".parse::().unwrap()); - options.server_publish = Some(publisher); + options.server_publish = Some(publisher.consume()); options.client_subscribe = Some(consumer_origin.clone()); let _pair = connect_mock(options).await; diff --git a/rs/moq-net/tests/finished_broadcast_mock.rs b/rs/moq-net/tests/finished_broadcast_mock.rs index 30fcefe404..4c987f9317 100644 --- a/rs/moq-net/tests/finished_broadcast_mock.rs +++ b/rs/moq-net/tests/finished_broadcast_mock.rs @@ -38,7 +38,7 @@ async fn round(finish_broadcast: bool) -> (Vec>, Option) let subscriber = produce_origin(2); let mut options = MockConnectOptions::new("moq-lite-05".parse::().unwrap()); - options.server_publish = Some(publisher.clone()); + options.server_publish = Some(publisher.consume()); options.client_subscribe = Some(subscriber.clone()); let pair: MockPair = connect_mock(options).await; diff --git a/rs/moq-net/tests/goaway.rs b/rs/moq-net/tests/goaway.rs index c85713fa82..bc407c2a49 100644 --- a/rs/moq-net/tests/goaway.rs +++ b/rs/moq-net/tests/goaway.rs @@ -336,7 +336,7 @@ async fn goaway_gates_new_subscribes_moq_lite_04() { let sub_origin = produce_origin(Hop::random()); let mut opts = MockConnectOptions::new(version); - opts.server_publish = Some(pub_origin.clone()); + opts.server_publish = Some(pub_origin.consume()); opts.client_subscribe = Some(sub_origin.clone()); let MockPair { client, server, .. } = connect_mock(opts).await; @@ -423,7 +423,7 @@ async fn goaway_drains_routes(version: Version) { let sub_origin = produce_origin(Hop::random()); let mut opts = MockConnectOptions::new(version); - opts.server_publish = Some(pub_origin.clone()); + opts.server_publish = Some(pub_origin.consume()); opts.client_subscribe = Some(sub_origin.clone()); let MockPair { client, server, .. } = connect_mock(opts).await; diff --git a/rs/moq-net/tests/rejoin.rs b/rs/moq-net/tests/rejoin.rs index 5ba6a80b3c..ecadbbfb18 100644 --- a/rs/moq-net/tests/rejoin.rs +++ b/rs/moq-net/tests/rejoin.rs @@ -42,7 +42,7 @@ async fn rejoin_recovers_the_group_reset_on_leave() { broadcast.announce(Default::default()).unwrap(); let mut options = MockConnectOptions::new(version.parse::().unwrap()); - options.server_publish = Some(publisher); + options.server_publish = Some(publisher.consume()); options.client_subscribe = Some(relay.clone()); let _pair = connect_mock(options).await; diff --git a/rs/moq-net/tests/support/harness.rs b/rs/moq-net/tests/support/harness.rs index 701d83fb32..4cd2cf254f 100644 --- a/rs/moq-net/tests/support/harness.rs +++ b/rs/moq-net/tests/support/harness.rs @@ -22,11 +22,11 @@ pub struct MockConnectOptions { /// The MoQ version to negotiate (determines the ALPN protocol string). pub version: Version, /// Origin whose broadcasts the client publishes to the server. - pub client_publish: Option, + pub client_publish: Option, /// Origin the client inserts remote broadcasts into. pub client_subscribe: Option, /// Origin whose broadcasts the server publishes to the client. - pub server_publish: Option, + pub server_publish: Option, /// Origin the server inserts remote broadcasts into. pub server_subscribe: Option, } From f3ae1239574b0a1d962a3aa8d5ee7aa5217a586b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 10:12:21 -0700 Subject: [PATCH 3/5] docs(quest): perf follow-ups from the session benchmark Co-Authored-By: Claude Opus 5.5 --- quest/m1/README.md | 1 + quest/m1/cache-expiry-growth.md | 24 ++++++++++++++++++++++++ quest/m1/perf/README.md | 3 +++ quest/m1/perf/announce-replay.md | 22 ++++++++++++++++++++++ quest/m1/perf/group-cost.md | 25 +++++++++++++++++++++++++ quest/m1/perf/lite-route-rescan.md | 29 +++++++++++++++++++++++++++++ rs/moq-net/benches/session.rs | 23 +++++++++++++++++++---- 7 files changed, 123 insertions(+), 4 deletions(-) create mode 100644 quest/m1/cache-expiry-growth.md create mode 100644 quest/m1/perf/announce-replay.md create mode 100644 quest/m1/perf/group-cost.md create mode 100644 quest/m1/perf/lite-route-rescan.md diff --git a/quest/m1/README.md b/quest/m1/README.md index cf72bbcce6..302ad353ab 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -87,6 +87,7 @@ transport, benchmark tooling); worktrees isolate commits, not semantics. - [Watch worker](/quest/m1/watch-worker.md) - watch playback runs in a worker onto an OffscreenCanvas, so main-thread jank never stalls video or audio - [Closure counters](/quest/m1/closure-counters.md) - a departed node's return never regresses the closure counters a consumer already saw - [RTMP interleaving](/quest/m1/rtmp-interleaving.md) - isolate partial messages before optimizing assembly copies +- [Cache expiry growth](/quest/m1/cache-expiry-growth.md) - with the default pool, relay memory plateaus at the expiry window on moq-transport as on moq-lite - [Relay memory](/quest/m1/relay-memory.md) - remeasure what an announcement costs after prefix routes - [PoP skipping](/quest/m1/pop-skipping/README.md) - short cold paths for unpopular broadcasts without losing warm backhaul dedup - [Route cost in the JS origin](/quest/m1/route-cost.md) - the browser origin ranks routes by cost and hops like Rust instead of newest-first diff --git a/quest/m1/cache-expiry-growth.md b/quest/m1/cache-expiry-growth.md new file mode 100644 index 0000000000..17cac468ed --- /dev/null +++ b/quest/m1/cache-expiry-growth.md @@ -0,0 +1,24 @@ +# [S] Cached groups expire on the IETF path + +## Goal + +With the default cache pool, a relay's memory plateaus once groups start +reaching the expiry window, on moq-transport as on moq-lite, or the cause of +continued growth is found and fixed. + +## Plan + +In `session_delivery_broadcasts` with 4096 announced broadcasts and the +default unbounded pool (30 s expiry), IETF RSS reached 1.5 GB at 10 s, 3.5 GB at +30 s, and 4.9 GB at 50 s, still climbing past the expiry window. Lite +reached about 0.36 GB by 50 s. With a 16 MiB bounded pool both stay flat, so +the retained bytes are cached groups, not a leak elsewhere. + +Reproduce with a focused test first. Unexpired groups at higher throughput, +expiry not running on a path IETF uses, or groups held outside the pool's +accounting would each explain it. Fix what is actually wrong. Consider +whether an unbounded default is the right default for an origin at all. + +## Related + +- [Relay memory](/quest/m1/relay-memory.md) - what an announcement costs in memory diff --git a/quest/m1/perf/README.md b/quest/m1/perf/README.md index b1a0a9cf4f..c855d6344d 100644 --- a/quest/m1/perf/README.md +++ b/quest/m1/perf/README.md @@ -43,6 +43,9 @@ row per io_uring worker. - [Open contract](/quest/m1/perf/uring-open-contract.md) - plan concurrent WebTransport opening and cancellation +- [Lite route rescan](/quest/m1/perf/lite-route-rescan.md) - a moq-lite session's per-group cost stops growing with the routes its peer announced +- [Announce replay](/quest/m1/perf/announce-replay.md) - the initial announce set replays in linear time, so joins don't slow with the route count +- [Group cost](/quest/m1/perf/group-cost.md) - count and cut the allocations and time spent relaying one small group to one viewer - [One enter per turn](/quest/m1/perf/uring-one-enter.md) - a parking turn pays one io_uring_enter, submits flush deferred completions, and SQEs per enter is a counter - [Run to quiescence](/quest/m1/perf/uring-quiescence.md) - a received packet's reply is staged in the same turn, under a pass budget that keeps the fairness rule - [Lock wait](/quest/m1/perf/lock-wait.md) - each worker reports time blocked on cross-worker locks, deciding whether the shared model needs work diff --git a/quest/m1/perf/announce-replay.md b/quest/m1/perf/announce-replay.md new file mode 100644 index 0000000000..db7f81b036 --- /dev/null +++ b/quest/m1/perf/announce-replay.md @@ -0,0 +1,22 @@ +# [S] Linear announce replay on subscribe + +## Goal + +A session that subscribes to announcements receives the initial set in time +linear in the number of announced broadcasts, so a viewer joining a relay +with thousands of routes connects as fast per route as one joining a handful. + +## Plan + +The moq-lite publisher's initial replay (`AnnounceRun::init` in +`rs/moq-net/src/lite/publisher.rs`) de-duplicates by scanning the pending list +for every route it drains: `initial.retain` on Lite05+, `init.contains` on +the Lite01/02 init. That is quadratic in the replay size. + +Measured with `session_join_broadcasts` (2026-09, one relay): lite join takes +199 µs with 1 announced broadcast, 517 µs with 64, and 8.9 ms with 1024, +about 8.7 µs per route at the top. IETF is 7.2 ms at 1024 with no quadratic +scan; its profile is SipHash hashing and `Path` comparison, so check whether +a faster hasher for path-keyed maps pays off on both. + +Keep the replay's order and its last-update-wins semantics. diff --git a/quest/m1/perf/group-cost.md b/quest/m1/perf/group-cost.md new file mode 100644 index 0000000000..b558d76b62 --- /dev/null +++ b/quest/m1/perf/group-cost.md @@ -0,0 +1,25 @@ +# [M] Measure and cut the fixed cost of relaying a group + +## Goal + +The fixed cost of relaying one small group to one viewer drops, measured as +allocations and time per viewer-group in the session benchmark, with no +change to what is delivered. + +## Plan + +`session_delivery_viewers` (2026-09) spends about 32 µs per viewer per +4-frame, 64-byte group over the in-memory transport, through a publisher +session, a relay origin, and a viewer session. No single function dominates. +The profile spreads it over `kio` waiter registration and parking, +`TrackState::evict_expired_scan` (4-5%), `Waiter`'s lazily allocated shared +waker (`Once::call`, about 3%, so waiters are created per poll), and +malloc/free (10-12%). + +Count allocations per viewer-group first (a counting allocator in the bench, +as `moq-json`'s allocation bench does), then remove the largest sources. A +measured no-win abandons the quest, per this line's rules. + +## Related + +- [Owned decoding copies](/quest/m1/perf/coding-decode.md) - decode-side copies are part of the same per-group cost diff --git a/quest/m1/perf/lite-route-rescan.md b/quest/m1/perf/lite-route-rescan.md new file mode 100644 index 0000000000..6591609912 --- /dev/null +++ b/quest/m1/perf/lite-route-rescan.md @@ -0,0 +1,29 @@ +# [S] Lite subscriber polls only routes with a request + +## Goal + +A moq-lite session's per-group cost stops growing with the number of +broadcasts its peer has announced: delivery over a session holding thousands +of announced routes costs what it costs with a handful, as it already does +over moq-transport. + +## Plan + +`Announced::poll_serve` (`rs/moq-net/src/lite/subscriber.rs`) polls every +attached route's `Dynamic::poll_requested_broadcast` on every driver wake, so +each incoming group pays for every route, and each pending poll registers the +driver's waiter on every idle route's list. The IETF subscriber runs one task +per route and only wakes the one that got a request. + +Measured with `cargo bench -p moq-net --bench session` (2026-09, one relay, +16 viewers each watching one broadcast): + +- `session_delivery_broadcasts`: lite 445 µs at 16 announced, 15.2 ms at + 4096; IETF flat around 458 µs. 57% of lite CPU is `WaiterList::register` + under `poll_requested_broadcast`. +- `session_delivery_scale` at 256 publishers x 256 viewers: lite 40 ms, IETF + 16.5 ms, since every viewer session holds all 256 routes. + +Wake only routes with a pending request. A task per route like IETF's or a +ready set both fit; pick by what keeps the driver's fairness rules. Land +with the before/after of those two groups. diff --git a/rs/moq-net/benches/session.rs b/rs/moq-net/benches/session.rs index 6c511ff1e0..44f740dfed 100644 --- a/rs/moq-net/benches/session.rs +++ b/rs/moq-net/benches/session.rs @@ -89,7 +89,10 @@ impl Shape { } fn runtime() -> tokio::runtime::Runtime { - tokio::runtime::Builder::new_current_thread().enable_time().build().unwrap() + tokio::runtime::Builder::new_current_thread() + .enable_time() + .build() + .unwrap() } fn write_group(track: &track::Producer, payload: &Bytes) { @@ -231,7 +234,10 @@ struct Room { impl Room { async fn new(version: &str, shape: Shape) -> Self { - assert!(shape.watch <= shape.total(), "a viewer can't watch more broadcasts than exist"); + assert!( + shape.watch <= shape.total(), + "a viewer can't watch more broadcasts than exist" + ); let mut cluster = Cluster::new(version, shape).await; // Viewer v watches a contiguous window starting at v * watch, so viewers @@ -254,7 +260,12 @@ impl Room { let mut room = Self { cluster, viewers, - watched: watched.iter().enumerate().filter(|(_, w)| **w).map(|(i, _)| i).collect(), + watched: watched + .iter() + .enumerate() + .filter(|(_, w)| **w) + .map(|(i, _)| i) + .collect(), payload: Bytes::from(vec![0; shape.frame]), expected: shape.expected(), }; @@ -350,7 +361,11 @@ fn join(c: &mut Criterion, name: &str, shapes: impl IntoIterator) fn session(c: &mut Criterion) { let base = Shape::BASE; - delivery(c, "publishers", [1, 16, 256].map(|publishers| Shape { publishers, ..base })); + delivery( + c, + "publishers", + [1, 16, 256].map(|publishers| Shape { publishers, ..base }), + ); delivery(c, "viewers", [1, 16, 256].map(|viewers| Shape { viewers, ..base })); delivery( c, From 1dfc36b7274d2a36f28b316f0682d8b593876acf Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 10:26:44 -0700 Subject: [PATCH 4/5] fix(moq-net): per-endpoint cache pools and fixed-path join samples in the session bench Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/benches/session.rs | 31 +++++++++++++++++-------------- 1 file changed, 17 insertions(+), 14 deletions(-) diff --git a/rs/moq-net/benches/session.rs b/rs/moq-net/benches/session.rs index 44f740dfed..cb8bd4559f 100644 --- a/rs/moq-net/benches/session.rs +++ b/rs/moq-net/benches/session.rs @@ -36,10 +36,11 @@ const FRAMES: usize = 4; const TRACK: &str = "video"; -/// One bounded pool for every origin, as a relay configures: 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. -const CACHE_CAPACITY: u64 = 16 * 1024 * 1024; +/// 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. +const RELAY_CACHE: u64 = 16 * 1024 * 1024; +const CLIENT_CACHE: u64 = 1024 * 1024; /// One topology: publishers and viewers spread round-robin over a full mesh of /// relays. @@ -128,7 +129,6 @@ struct Cluster { _broadcasts: Vec, _origins: Vec, _pairs: Vec, - pool: cache::Pool, next_hop: u64, } @@ -141,12 +141,11 @@ impl Cluster { _broadcasts: Vec::new(), _origins: Vec::new(), _pairs: Vec::new(), - pool: cache::Pool::new(cache::Config::default().with_capacity(CACHE_CAPACITY)), next_hop: 0, }; for _ in 0..shape.relays { - let relay = this.origin(); + let relay = this.origin(RELAY_CACHE); this.relays.push(relay); } @@ -166,7 +165,7 @@ impl Cluster { } for publisher in 0..shape.publishers { - let origin = this.origin(); + let origin = this.origin(CLIENT_CACHE); for index in 0..shape.broadcasts { let broadcast = origin .publish(path(publisher * shape.broadcasts + index), Default::default()) @@ -185,10 +184,10 @@ impl Cluster { this } - fn origin(&mut self) -> origin::Producer { + fn origin(&mut self, capacity: u64) -> origin::Producer { self.next_hop += 1; let mut config = origin::Config::new(Hop::new(self.next_hop).unwrap()); - config.pool = self.pool.clone(); + config.pool = cache::Pool::new(cache::Config::default().with_capacity(capacity)); let (producer, driver) = origin::Producer::new(config); tokio::spawn(support::harness::run(driver)); producer @@ -196,7 +195,7 @@ impl Cluster { /// Connect a viewer to `relay` and subscribe it to each of `broadcasts`. async fn join(&mut self, relay: usize, broadcasts: impl Iterator) -> Viewer { - let origin = self.origin(); + let origin = self.origin(CLIENT_CACHE); let mut options = MockConnectOptions::new(self.version); options.server_publish = Some(self.relays[relay].consume()); options.client_subscribe = Some(origin.clone()); @@ -333,13 +332,17 @@ fn join(c: &mut Criterion, name: &str, shapes: impl IntoIterator) cluster }) }); + // Broadcasts published to relay 0, joined from the last relay: every + // sample is local on one relay and crosses one peer hop on a mesh. + let hosted: Vec<_> = (0..shape.total()) + .filter(|broadcast| (broadcast / shape.broadcasts) % shape.relays == 0) + .collect(); + let relay = shape.relays - 1; b.iter_custom(|iters| { rt.block_on(async { let mut elapsed = Duration::ZERO; for iter in 0..iters as usize { - let broadcast = iter % shape.total(); - // The last relay, so a mesh resolves across a peer. - let relay = shape.relays - 1; + let broadcast = hosted[iter % hosted.len()]; let start = Instant::now(); let mut viewer = cluster.join(relay, std::iter::once(broadcast)).await; let bytes = read_group(&mut viewer.subscribers[0]).await; From d300eceacb5e12dd8274bd3cc8cc5e0a7c6aa763 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 16:18:02 -0700 Subject: [PATCH 5/5] fix(moq-net): parse moq-lite-07-wip in the session bench moq-lite-07 no longer parses. Mock publish options take a Consumer, so track_tail consumes like the other harness tests. --- rs/moq-net/benches/session.rs | 4 ++-- rs/moq-net/tests/track_tail.rs | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/rs/moq-net/benches/session.rs b/rs/moq-net/benches/session.rs index cb8bd4559f..42e65e2fd2 100644 --- a/rs/moq-net/benches/session.rs +++ b/rs/moq-net/benches/session.rs @@ -28,8 +28,8 @@ use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_m use moq_net::{Hop, Timestamp, Version, broadcast, cache, origin, track}; use support::harness::{MockConnectOptions, MockPair, connect_mock}; -/// The newest moq-lite and IETF drafts. -const VERSIONS: [&str; 2] = ["moq-lite-07", "moq-transport-22"]; +/// Newest lite draft (`moq-lite-07-wip`, opt-in) and newest IETF draft. +const VERSIONS: [&str; 2] = ["moq-lite-07-wip", "moq-transport-22"]; /// Frames per group, so per-frame and per-group costs both appear. const FRAMES: usize = 4; diff --git a/rs/moq-net/tests/track_tail.rs b/rs/moq-net/tests/track_tail.rs index 28ef3d957f..f9dd8ca201 100644 --- a/rs/moq-net/tests/track_tail.rs +++ b/rs/moq-net/tests/track_tail.rs @@ -66,7 +66,7 @@ async fn round(version: &str, late: Late) -> Outcome { let subscriber = produce_origin(2); let mut options = MockConnectOptions::new(version.parse::().unwrap()); - options.server_publish = Some(publisher.clone()); + options.server_publish = Some(publisher.consume()); options.client_subscribe = Some(subscriber.clone()); let pair = connect_mock(options).await;