diff --git a/quest/m2/cluster-routing/routes/README.md b/quest/m2/cluster-routing/routes/README.md index 26e66f7f56..71d383ddbb 100644 --- a/quest/m2/cluster-routing/routes/README.md +++ b/quest/m2/cluster-routing/routes/README.md @@ -542,7 +542,7 @@ benchmarks. - [Route layer acceptance](/quest/m2/cluster-routing/routes/acceptance.md) - forwarding fixtures ported and the model deleted, incident fixtures, end-to-end checks, and benchmarks - [Firefox tabs prove their route key](/quest/m2/cluster-routing/routes/browser-proof.md) - a current Firefox tab opens a lite-07 session with a route-key proof the relay verifies, once neqo's exporter follows the draft - [Playwright bundles Chromium's spec exporter](/quest/m2/cluster-routing/routes/playwright-pin.md) - the toolchain's Playwright bundles Chromium 158+, so the route key lane proves the Chromium tab -- [Showing an identity's claims costs the claims that changed](/quest/m2/cluster-routing/routes/show-cost.md) - lite-07 session setup and teardown grow linearly with broadcasts, not quadratically per identity +- [The route node behaves the same whoever creates it](/quest/m2/cluster-routing/routes/node-creation-order.md) - creating an origin's lite-07 route node from SETUP instead of the route layer changes nothing observable ## Related diff --git a/quest/m2/cluster-routing/routes/show-cost.md b/quest/m2/cluster-routing/routes/show-cost.md deleted file mode 100644 index 2e1867b188..0000000000 --- a/quest/m2/cluster-routing/routes/show-cost.md +++ /dev/null @@ -1,32 +0,0 @@ -# [M] Showing an identity's claims costs the claims that changed - -## Goal - -A lite-07 session's setup and teardown cost grows linearly with the -broadcasts it carries, as lite-06's does. One identity holding many claims no -longer makes `show` in `rs/moq-net/src/route/mod.rs` quadratic. The session -bench (`rs/moq-net/benches/session`) and the route bench -(`rs/moq-net/benches/routes.rs`) show it, swept over both axes as the -repository's fan-out rule requires. - -## Plan - -Planned 2026-11-09 (maintainer, /quest-complete follow-up of -[moq-dev/moq#8352](https://github.com/moq-dev/moq/pull/8352)). ✅ m2, in this -epic (rejected: m4, dropping it). - -#8352's session bench measured lite-07 setup at 4096 broadcasts at 4.9 s of -CPU, against 0.5 s for lite-06, and teardown the same. It traced this to -showing one identity's claims: every change re-derives what the identity -shows from all of its claims. That ran on main after -[moq-dev/moq#8293](https://github.com/moq-dev/moq/pull/8293) (routes presented -incrementally), but before -[moq-dev/moq#8374](https://github.com/moq-dev/moq/pull/8374) (a relay folds -received route batches by the claims they change). - -Re-measure on current main first. If #8374 already made it linear, record the -numbers in the PR and delete this quest. Otherwise keep what `show` derives up -to date per changed claim instead of recomputing it, and add a route-bench -case swept over claims per identity and claims changed. - -Public API: none expected. Wire: none. diff --git a/rs/moq-net/benches/session/mod.rs b/rs/moq-net/benches/session/mod.rs index 80b5b0327c..f484d8f4c5 100644 --- a/rs/moq-net/benches/session/mod.rs +++ b/rs/moq-net/benches/session/mod.rs @@ -43,6 +43,10 @@ //! `session_join_*` times a new viewer connecting, resolving a broadcast, and //! receiving its latest group, swept over what the relays already announce. //! +//! `session_announce` times one broadcast announced and withdrawn beside many a publisher +//! already announces, until every viewer session saw both, swept over those broadcasts and +//! the viewers: a change's cost grows with the viewers it reaches, never with the claims held. +//! //! `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 @@ -601,7 +605,8 @@ struct Cluster { /// One datagram track per broadcast, indexed like [`path`], for datagram shapes only. datagrams: Vec, _broadcasts: Vec, - _origins: Vec, + /// Each publisher's origin, indexed like `publisher`. + origins: Vec, _pairs: Vec, next_hop: u64, /// The relay and publisher caches; each viewer owns its own. @@ -621,7 +626,7 @@ impl Cluster { tracks: Vec::new(), datagrams: Vec::new(), _broadcasts: Vec::new(), - _origins: Vec::new(), + origins: Vec::new(), _pairs: Vec::new(), next_hop: 0, pools: Vec::new(), @@ -689,7 +694,7 @@ impl Cluster { let target = &this.relays[publisher % shape.relays]; options.server_subscribe = Some(B::origin(target).clone()); this._pairs.push(B::connect(options, target, false).await); - this._origins.push(origin); + this.origins.push(origin); } this @@ -1477,6 +1482,69 @@ fn join(c: &mut Criterion, name: &str, shapes: impl IntoIterator(c: &mut Criterion) { + let mut group = c.benchmark_group(format!("{}session_announce", B::PREFIX)); + for version in VERSIONS { + for (broadcasts, viewers) in ANNOUNCE { + let rt = runtime(); + let mut room = None; + let id = format!("{version}/broadcasts={broadcasts}/viewers={viewers}"); + group.bench_function(BenchmarkId::from_parameter(id), |b| { + let (cluster, routes, next) = room.get_or_insert_with(|| { + rt.block_on(async { + let shape = Shape { + publishers: 1, + broadcasts, + viewers, + ..Shape::BASE + }; + let mut cluster = Cluster::::new(version, shape, false).await; + let mut routes = Vec::new(); + for _ in 0..viewers { + let viewer = cluster.join(0, false).await; + let mut announced = viewer.origin.consume().routes(); + for _ in 0..broadcasts { + let event = announced.next().await.expect("announced"); + assert_eq!(event.kind, origin::RouteKind::Start); + } + routes.push((viewer, announced)); + } + (cluster, routes, broadcasts) + }) + }); + b.iter_custom(|iters| { + rt.block_on(async { + let start = Instant::now(); + for _ in 0..iters { + // A fresh path each time: a broadcast's identity never comes back. + let route = cluster.origins[0] + .route(path(*next), origin::RouteOptions::default()) + .unwrap(); + *next += 1; + for (_, announced) in routes.iter_mut() { + let event = announced.next().await.expect("announced"); + assert_eq!(event.kind, origin::RouteKind::Start); + } + drop(route); + for (_, announced) in routes.iter_mut() { + let event = announced.next().await.expect("withdrawn"); + assert_eq!(event.kind, origin::RouteKind::End); + } + } + start.elapsed() + }) + }) + }); + } + } + 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 @@ -2101,6 +2169,8 @@ pub(crate) fn session(c: &mut Criterion) { join::(c, "relays", [1, 2, 8].map(|relays| Shape { relays, ..base })); + announce::(c); + churn::(c, "sessions", [1, 16, 128].map(|sessions| Churn { sessions, held: 16 })); churn::( c, diff --git a/rs/moq-net/src/lite/publisher/routed.rs b/rs/moq-net/src/lite/publisher/routed.rs index 3a72ece523..3e4fd8f5fb 100644 --- a/rs/moq-net/src/lite/publisher/routed.rs +++ b/rs/moq-net/src/lite/publisher/routed.rs @@ -39,6 +39,8 @@ struct Sent { reach: BTreeMap, /// Every claim the node sent for it, named as the handle names it, before filtering. claims: BTreeMap, + /// How many of `claims` the stream's link never carries (see [`sends`]). + withheld: usize, /// The node's floor for those claims. floor: Option, /// Whether the peer was sent a floor for it yet. @@ -49,6 +51,17 @@ impl Sent { fn reach(&self) -> Option { self.reach.values().next().copied() } + + /// The node's claim at `path` is now `claim`, which `link` may withhold. + fn set(&mut self, link: &route::Link, path: Path, claim: Option) { + let withheld = claim.as_ref().is_some_and(|claim| !sends(link, claim)); + let old = match claim { + Some(claim) => self.claims.insert(path, claim), + None => self.claims.remove(&path), + }; + self.withheld += usize::from(withheld); + self.withheld -= usize::from(old.is_some_and(|old| !sends(link, &old))); + } } /// A node stream the answer reads. @@ -339,6 +352,7 @@ impl Answer { shown: 0, reach: BTreeMap::from([(source, reach)]), claims: BTreeMap::new(), + withheld: 0, floor: None, floored: false, }; @@ -377,7 +391,7 @@ impl Answer { self.echo(id, source); let sent = self.routes.get_mut(&id).expect("matched above"); for path in &paths { - sent.claims.remove(path); + sent.set(&self.link, path.clone(), None); } for path in paths { self.sync(stream, id, path)?; @@ -502,10 +516,7 @@ impl Answer { None if self.origin.shadows(&path) => continue, None => path, }; - match claim { - Some(claim) => sent.claims.insert(path.clone(), claim), - None => sent.claims.remove(&path), - }; + sent.set(&self.link, path.clone(), claim); paths.push(path); } sent.floor = floor; @@ -521,10 +532,8 @@ impl Answer { let certified = self.certified(); // Like the grant, the upstream horizon and the proxy preference leave claims out // that a floor would certify absent. - let hides = (self.routes.get(&id).into_iter()) - .flat_map(|sent| sent.claims.values()) - .any(|claim| !sends(&self.link, claim)); let sent = self.routes.get_mut(&id).expect("only a held route has a batch"); + let hides = sent.withheld > 0; let Some(wire) = sent.wire else { return Ok(()); }; diff --git a/rs/moq-net/src/lite/subscriber/routed.rs b/rs/moq-net/src/lite/subscriber/routed.rs index d1b55c5bd6..a2cef69462 100644 --- a/rs/moq-net/src/lite/subscriber/routed.rs +++ b/rs/moq-net/src/lite/subscriber/routed.rs @@ -6,7 +6,7 @@ //! the origin, minting a source per request that names the claim's origin node. use std::{ - collections::{BTreeMap, BTreeSet, HashMap}, + collections::{BTreeMap, HashMap}, task::Poll, }; @@ -22,8 +22,8 @@ use super::{AnnouncedRoute, Subscriber}; struct WireRoute { node: route::Id, reach: route::Reach, - /// The Announce IDs of its live claims. - claims: BTreeSet, + /// The Announce ID of each of its live claims, by path. + claims: BTreeMap, /// How its claims differ from `committed`, by path, so what the stream keeps /// between floors is bounded by its claims rather than the messages it read. pending: BTreeMap>, @@ -175,7 +175,7 @@ impl Routed { WireRoute { node, reach, - claims: BTreeSet::new(), + claims: BTreeMap::new(), pending: BTreeMap::new(), committed: None, }, @@ -186,7 +186,7 @@ impl Routed { Msg::RouteHold { id } => self.reach(id, route::Reach::Held)?, Msg::RouteEnd { id } => { let route = self.routes.remove(&id).ok_or(Error::ProtocolViolation)?; - for claim in route.claims { + for claim in route.claims.into_values() { decoder.end(claim)?; self.claims.remove(&claim); self.held.pop(); @@ -271,17 +271,12 @@ impl Routed { /// A claim the peer started under `route`, assigned Announce ID `id`. fn start(&mut self, id: u64, route: u64, path: Path, claim: node::Claim) -> Result<(), Error> { - let claims = &self.claims; let wire = self.routes.get_mut(&route).ok_or(Error::ProtocolViolation)?; - if wire - .claims - .iter() - .any(|other| claims.get(other).is_some_and(|other| other.path == path)) - { + if wire.claims.contains_key(&path) { return Err(Error::ProtocolViolation); } self.held.push(self.slots.acquire()?); - wire.claims.insert(id); + wire.claims.insert(path.clone(), id); wire.change(path.clone(), Some(claim.clone())); self.claims.insert(id, WireClaim { route, path, claim }); Ok(()) @@ -292,7 +287,7 @@ impl Routed { let claim = self.claims.remove(&id).ok_or(Error::ProtocolViolation)?; self.held.pop(); let route = self.routes.get_mut(&claim.route).ok_or(Error::ProtocolViolation)?; - route.claims.remove(&id); + route.claims.remove(&claim.path); route.change(claim.path, None); Ok(()) }