diff --git a/quest/m1/bench-relay.md b/quest/m1/bench-relay.md index 1cd12ddfe9..68609da539 100644 --- a/quest/m1/bench-relay.md +++ b/quest/m1/bench-relay.md @@ -18,3 +18,9 @@ than the bench is worth, say so and stop. Reuse the scenario, the publisher and subscriber sweeps, and the delivery accounting from `rs/moq-net/benches/session.rs` so the two results line up, and the difference is the relay layer. + +Run the dash shape (`session_dash_*`) first. In the model bench, the production +point (34 peer nodes x 5 projects x 24 tracks) costs about 100 ms of one core +per 1 s stats tick, for the whole mesh. In production, moving that one session +off a relay saved 50-90% of a core on that relay alone (2026-09-25). The gap is +the relay layer or the transport, not the model. diff --git a/rs/moq-net/benches/session.rs b/rs/moq-net/benches/session.rs index 947fa10727..62485be85b 100644 --- a/rs/moq-net/benches/session.rs +++ b/rs/moq-net/benches/session.rs @@ -15,6 +15,11 @@ //! `session_join_*` times a new viewer connecting, resolving a broadcast, and //! receiving its latest group, swept over what the relays already announce. //! +//! `session_dash_*` models the moq.pro dash aggregator: one session following +//! every node's per-project stats broadcast, each announced by a peer relay, at +//! one small frame per track per tick. Throughput counts tracks, so a flat +//! elements/s across a sweep means the cost per track stays constant. +//! //! Everything runs on one current-thread runtime, so the only work measured is //! the protocol and model code, never scheduling across threads. //! @@ -41,6 +46,12 @@ const FRAMES: usize = 4; const TRACK: &str = "video"; +/// Lite 06 is what the dash session and cluster peers negotiate today. +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. @@ -151,9 +162,9 @@ fn runtime() -> tokio::runtime::Runtime { .unwrap() } -fn write_group(track: &track::Producer, payload: &Bytes) { +fn write_group(track: &track::Producer, payload: &Bytes, frames: usize) { let mut group = track.append_group().unwrap(); - for _ in 0..FRAMES { + for _ in 0..frames { group.write_frame(Timestamp::ZERO, payload.clone()).unwrap(); } group.finish().unwrap(); @@ -248,24 +259,19 @@ impl Cluster { producer } - /// Connect a viewer to `relay` and subscribe it to each of `broadcasts`. - async fn join(&mut self, relay: usize, broadcasts: impl Iterator) -> Viewer { + /// Connect a viewer session to `relay`. + async fn join(&mut self, relay: usize) -> Viewer { let origin = self.origin(CLIENT_CACHE); let mut options = MockConnectOptions::new(self.version); - options.server_publish = Some(self.relays[relay].consume()); + // Hidden included, as the dash's `.stats/` scope sees its subtree. + options.server_publish = Some(self.relays[relay].consume().with_hidden(true)); 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, + subscribers: Vec::new(), _pair: pair, - _origin: origin, + origin, } } } @@ -273,7 +279,18 @@ impl Cluster { struct Viewer { subscribers: Vec, _pair: MockPair, - _origin: origin::Producer, + origin: origin::Producer, +} + +impl Viewer { + /// Resolve `path` and subscribe to each of its `tracks`. + async fn watch(&mut self, path: &str, tracks: &[impl AsRef]) { + let broadcast = self.origin.consume().routed_broadcast(path).await.unwrap(); + for track in tracks { + self.subscribers + .push(broadcast.track(track.as_ref()).unwrap().subscribe(None).await.unwrap()); + } + } } /// A cluster with its viewers attached. @@ -313,8 +330,11 @@ impl Room { } // 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 viewer = cluster.join((viewer + 1) % shape.relays).await; + for broadcast in broadcasts { + viewer.watch(&path(broadcast), &[TRACK]).await; + } + viewers.push(viewer); } let mut room = Self { @@ -346,7 +366,7 @@ impl Room { return self.paced_round().await; } for &broadcast in &self.watched { - write_group(&self.cluster.tracks[broadcast], &self.payload); + write_group(&self.cluster.tracks[broadcast], &self.payload, FRAMES); } let mut bytes = 0; for viewer in &mut self.viewers { @@ -429,7 +449,7 @@ fn join(c: &mut Criterion, name: &str, shapes: impl IntoIterator) let cluster = Cluster::new(version, *shape).await; let payload = Bytes::from(vec![0; shape.frame]); for track in &cluster.tracks { - write_group(track, &payload); + write_group(track, &payload, FRAMES); } cluster }) @@ -446,7 +466,8 @@ fn join(c: &mut Criterion, name: &str, shapes: impl IntoIterator) for iter in 0..iters as usize { let broadcast = hosted[iter % hosted.len()]; let start = Instant::now(); - let mut viewer = cluster.join(relay, std::iter::once(broadcast)).await; + let mut viewer = cluster.join(relay).await; + viewer.watch(&path(broadcast), &[TRACK]).await; let bytes = read_group(&mut viewer.subscribers[0]).await; elapsed += start.elapsed(); assert_eq!(bytes, FRAMES * shape.frame); @@ -463,6 +484,131 @@ fn join(c: &mut Criterion, name: &str, shapes: impl IntoIterator) group.finish(); } +/// The moq.pro dash aggregator's load: one session on relay 0 following every +/// node's per-project stats broadcast. Every other relay is a node publishing +/// `.stats//node/` into its own origin, as `moq-relay`'s stats +/// producer does, so relay 0 reaches each one over a peer session. +#[derive(Clone, Copy)] +struct Dash { + /// Relays in the mesh: relay 0 hosts the dash session, the rest are nodes. + relays: usize, + /// Projects with a stats broadcast on every node. + projects: usize, + /// Tracks per stats broadcast. + tracks: usize, +} + +impl Dash { + fn total(&self) -> usize { + (self.relays - 1) * self.projects * self.tracks + } + + fn id(&self, version: &str) -> String { + format!( + "{version}/relays={}/projects={}/tracks={}/total={}", + self.relays, + self.projects, + self.tracks, + self.total() + ) + } +} + +/// A mesh of stats-publishing nodes with the dash session attached. +struct DashRoom { + _cluster: Cluster, + tracks: Vec, + _broadcasts: Vec, + dash: Viewer, + payload: Bytes, +} + +impl DashRoom { + async fn new(version: &str, shape: Dash) -> Self { + assert!(shape.relays > 1, "the dash needs a peer node"); + let mut cluster = Cluster::new( + version, + Shape { + relays: shape.relays, + publishers: 0, + viewers: 0, + ..Shape::BASE + }, + ) + .await; + + let names: Vec<_> = (0..shape.tracks).map(|track| format!("stat{track}")).collect(); + let mut tracks = Vec::new(); + let mut broadcasts = Vec::new(); + let mut paths = Vec::new(); + for node in 1..shape.relays { + for project in 0..shape.projects { + let path = format!(".stats/{project}/node/{node}"); + let broadcast = cluster.relays[node].publish(path.as_str(), Default::default()).unwrap(); + for name in &names { + tracks.push(broadcast.create_track(name.as_str(), None).unwrap()); + } + broadcasts.push(broadcast); + paths.push(path); + } + } + + let mut dash = cluster.join(0).await; + for path in &paths { + dash.watch(path, &names).await; + } + + let mut room = Self { + _cluster: cluster, + tracks, + _broadcasts: broadcasts, + dash, + payload: Bytes::from(vec![0; DASH_FRAME]), + }; + room.round().await; + room + } + + /// One stats tick: every track writes a single-frame group and the dash + /// reads it. + async fn round(&mut self) { + for track in &self.tracks { + write_group(track, &self.payload, 1); + } + let mut bytes = 0; + for subscriber in &mut self.dash.subscribers { + bytes += read_group(subscriber).await; + } + assert_eq!(bytes, self.tracks.len() * DASH_FRAME); + } + + async fn measure(&mut self, iters: u64) -> Duration { + let start = Instant::now(); + for _ in 0..iters { + self.round().await; + } + start.elapsed() + } +} + +fn dash(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_dash_{name}")); + group.sample_size(10); + for version in DASH_VERSIONS { + for shape in &shapes { + let mut room = None; + group.throughput(Throughput::Elements(shape.total() as u64)); + group.bench_function(BenchmarkId::from_parameter(shape.id(version)), |b| { + let room = room.get_or_insert_with(|| rt.block_on(DashRoom::new(version, *shape))); + b.iter_custom(|iters| rt.block_on(room.measure(iters))) + }); + } + } + group.finish(); +} + /// Print allocations per viewer-group instead of timing, to compare across revisions. fn allocations() { const ROUNDS: usize = 256; @@ -565,6 +711,47 @@ fn session(c: &mut Criterion) { }), ); join(c, "relays", [1, 2, 8].map(|relays| Shape { relays, ..base })); + + // Production (2026-09): ~34 nodes x >=5 projects x ~24 tracks. + let prod = Dash { + relays: 35, + projects: 5, + tracks: 24, + }; + // More broadcasts at a fixed track count per broadcast. + dash( + c, + "projects", + [1, 4, 16].map(|projects| Dash { + relays: 5, + projects, + ..prod + }), + ); + // More tracks per broadcast at a fixed broadcast count. + dash( + c, + "tracks", + [6, 24, 96].map(|tracks| Dash { + relays: 5, + projects: 4, + tracks, + }), + ); + // More peer sessions carrying the same 32 broadcasts, then the production + // shape. + dash( + c, + "relays", + [2, 3, 5, 9, 17, 33] + .map(|relays| Dash { + relays, + projects: 32 / (relays - 1), + ..prod + }) + .into_iter() + .chain([prod]), + ); } criterion_group!(benches, session);