From 3f2da83a6c191dfb60132973246057ebd731d117 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 09:05:48 -0700 Subject: [PATCH 1/3] fix(net): keep spliced lag exact when a route is replaced Record a retired copy's pending bytes at the edge it had, so a sample before the replacement's first frame still counts them, including when the subscription drops. Later bytes on that track do not, and they are not weighed at the replacement's lag. Unwatch the copy when let_go drops it, not when the switch caps it, so a copy the reader can still take groups from keeps weighing lag. --- quest/m1/qos/README.md | 3 - quest/m1/qos/lag-splice.md | 48 --------- quest/m1/quest-flat-lines.md | 3 +- rs/moq-net/src/model/resume.rs | 178 ++++++++++++++++++++++++++------- rs/moq-net/src/stats.rs | 120 ++++++++++++++++------ 5 files changed, 235 insertions(+), 117 deletions(-) delete mode 100644 quest/m1/qos/lag-splice.md diff --git a/quest/m1/qos/README.md b/quest/m1/qos/README.md index c85a44931f..8ab7bbd43c 100644 --- a/quest/m1/qos/README.md +++ b/quest/m1/qos/README.md @@ -44,9 +44,6 @@ on `main`. ## Required -- [Lag across a splice](/quest/m1/qos/lag-splice.md) - a route switch - neither loses pending lag weight nor keeps weighing a segment replaced - before its first frame; the line does not land until it is fixed - [Lag dashboard](/quest/m1/qos/lag-dashboard.md) - the demo stats dashboard shows viewer lag percentiles and dropped media - [Publisher timeliness](/quest/m1/qos/publisher-timeliness.md) - per diff --git a/quest/m1/qos/lag-splice.md b/quest/m1/qos/lag-splice.md deleted file mode 100644 index f7c4f583a2..0000000000 --- a/quest/m1/qos/lag-splice.md +++ /dev/null @@ -1,48 +0,0 @@ -# [S] Lag across a splice - -## Goal - -A spliced subscription's `lag` histogram stays exact across a route switch: -every byte the viewer could read is weighed once, and nothing a retired -segment produces afterwards is weighed at all, however early in its life the -segment was replaced. - -## Plan - -Decided in the 2026-10-05 audit: this quest, planned in draft -[#4381](https://github.com/moq-dev/moq/pull/4381) against the QoS line branch, -lands as a quest on `main` now that questlines are flat, and the QoS line -([#4133](https://github.com/moq-dev/moq/pull/4133)) is held until it is -fixed. The sampler exists only on the line branch, so the fix PR targets -`quest/m1/qos/README` (or lands with #4133). - -[#4298](https://github.com/moq-dev/moq/pull/4298) added the per-broadcast -viewer lag sampler (`FrontierInner::sample` in `rs/moq-net/src/stats.rs`) and -had a spliced subscription watch each segment's track -(`rs/moq-net/src/model/resume.rs`). Two Codex findings on its last round -merged unanswered, and the maintainer ruled in the 09-28 merged-PR audit that -both block the line: - -- **Pending weight is lost when the replacement has no frame yet** - ([r4113824520](https://github.com/moq-dev/moq/pull/4298#discussion_r4113824520)). - `sample` takes `state.unsampled` before it knows a source supplies `newest` - and `first`. When a capped segment folded bytes into `unsampled` and the - replacement has not produced, the `?` returns and those bytes are gone. - Keep the weight until a sample can record it, or keep the retired source's - edge alongside its weight. Still present on the line branch on 2026-10-05. -- **A segment replaced before its first frame is never unwatched** - ([r4113824522](https://github.com/moq-dev/moq/pull/4298#discussion_r4113824522)). - `unwatch` runs only when an existing segment's `end` changes. A segment - the producer replaces outright is marked `pruned` and dropped through - `retired()` without it, so if its route keeps publishing, those - unreachable bytes keep weighing this viewer's lag. Unwatch on every path - that retires or replaces a segment, not just on a cap. - -Extend the existing splice lag test (`lag_weighs_every_segment`) or add -siblings so each case fails without its fix. Rejected: two quests, since -both sit in the same splice accounting path and would conflict. - -Final lag sample (#4451) changed the same sampler; keep any weight this -quest defers in its drop-time sample. - -Public API: none. Wire: none, only the histogram's values change. diff --git a/quest/m1/quest-flat-lines.md b/quest/m1/quest-flat-lines.md index 4cfbbd4e3b..bb956c240a 100644 --- a/quest/m1/quest-flat-lines.md +++ b/quest/m1/quest-flat-lines.md @@ -57,8 +57,7 @@ Remaining: - Land each line's umbrella PR with `/quest-complete`, then delete its branch: #4403 wildcard, #4034 archive, #4039 auth, #4079 cpp, #4519 ffi-shape, - #4080 obs-moq-video, #4133 qos (held until - [Lag across a splice](/quest/m1/qos/lag-splice.md) is fixed), #4437 rs2ts, + #4080 obs-moq-video, #4133 qos, #4437 rs2ts, #4653 test-flakes-2, #4180 transport-upgrade, #4640 tstd. #4162 (audio-jitter-target) landed, and #4438 folded into rs2ts. - Fold #4255 (archive/track-timeline) into the archive branch before #4034 diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 8ff3a31084..426efed200 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -287,10 +287,9 @@ impl Copy { let mut sub = ready!(pending.poll_ok(waiter))?; sub.raise_start_to(groups.0); sub.end_at(groups.1); - // Only a copy that still serves weighs lag; see `Reader::sync`. - if self.until.is_none() { - sub.watch(&self.frontier); - } + // Weigh the copy while the reader can still take groups from it. A switch + // does not stop that; `Reader::let_go` does, once the copy is dropped. + sub.watch(&self.frontier); self.sub = Sub::Ready(sub); } let Sub::Ready(sub) = &mut self.sub else { unreachable!() }; @@ -301,6 +300,13 @@ impl Copy { fn needed(&self, serving: bool) -> bool { serving || (self.done.is_none() && Arc::strong_count(&self.lease) > 1) } + + /// Stop weighing this copy. A pending subscription was never watched. + fn unwatch(&mut self) { + if let Sub::Ready(sub) = &mut self.sub { + sub.unwatch(&self.frontier); + } + } } /// One reader's subscription to a logical track; see the module docs. @@ -370,29 +376,26 @@ impl Subscriber { self.reader().latest() } - /// Report the serving copy's track, now and after every switch, to `frontier`. + /// Report every copy still held to `frontier`, including one a switch capped but + /// the reader can still take groups from. pub(crate) fn watch(&mut self, frontier: &stats::Frontier) { let mut reader = self.reader(); reader.frontier = frontier.clone(); for copy in &mut reader.copies { copy.frontier = frontier.clone(); - if copy.until.is_none() - && let Sub::Ready(sub) = &mut copy.sub - { + if let Sub::Ready(sub) = &mut copy.sub { sub.watch(frontier); } } } - /// Stop reporting any copy's track to `frontier`, now or after a switch. + /// Stop reporting every copy's track to `frontier`. pub(crate) fn unwatch(&mut self, frontier: &stats::Frontier) { let mut reader = self.reader(); reader.frontier = stats::Frontier::default(); for copy in &mut reader.copies { copy.frontier = stats::Frontier::default(); - if copy.until.is_none() - && let Sub::Ready(sub) = &mut copy.sub - { + if let Sub::Ready(sub) = &mut copy.sub { sub.unwatch(frontier); } } @@ -481,15 +484,12 @@ impl Reader { return; } // The replaced copies finish the groups they had when replaced, and nothing newer. + // They keep weighing lag while those groups are still readable; `let_go` stops + // that once the copy is dropped. for copy in &mut self.copies { if copy.until.is_none() { copy.until = Some(self.newest.max(copy.track.latest())); copy.update(&self.mirrored); - // A replaced route's copy may live on; what it produces from now on never - // reaches this reader, so it stops weighing lag. - if let Sub::Ready(sub) = &mut copy.sub { - sub.unwatch(©.frontier); - } } } let floor = self.newest.map(Position::group); @@ -523,21 +523,30 @@ impl Reader { if copy.needed(Some(copy.generation) == serving) { return true; } - if copy.done.is_some() { - return false; - } - let until = copy.until; - let Poll::Ready(Ok(sub)) = copy.poll_ready(groups, &kio::Waiter::noop()) else { - return false; + let keep = if copy.done.is_some() { + false + } else { + let until = copy.until; + match copy.poll_ready(groups, &kio::Waiter::noop()) { + Poll::Ready(Ok(sub)) => { + // Inspect without moving either cursor: a later group poll still chooses its + // order and applies the current bounds and drift budget. Ignore the local + // cap here, since unread groups become eligible again when it rises. + sub.has_unread_group(&track::Unread { + start: groups.0.max(ordered.map_or(0, |last| last.saturating_add(1))), + end: until.and_then(|until| until.map_or(Some(0), |last| last.checked_add(1))), + delivered, + }) + } + _ => false, + } }; - // Inspect without moving either cursor: a later group poll still chooses its - // order and applies the current bounds and drift budget. Ignore the local - // cap here, since unread groups become eligible again when it rises. - sub.has_unread_group(&track::Unread { - start: groups.0.max(ordered.map_or(0, |last| last.saturating_add(1))), - end: until.and_then(|until| until.map_or(Some(0), |last| last.checked_add(1))), - delivered, - }) + // Unwatched on every retire, not only when a switch caps the copy. While the + // copy stays, the reader can still take what it already had. + if !keep { + copy.unwatch(); + } + keep }); } @@ -1286,10 +1295,10 @@ mod test { fn lag_weighs_the_serving_copy() { let registry = stats::Registry::new(stats::Config::new()); let session = registry.tier(stats::Tier::default()).session("root"); - let weight = || { + let lag = || { let mut report = stats::Report::default(); registry.report(&mut report); - report.traffic[0].publisher.lag.total() + report.traffic[0].publisher.lag }; let write = |track: &track::Producer, sequence: u64, payload: &[u8]| { let mut group = track.create_group(group::Info { sequence }).unwrap(); @@ -1312,19 +1321,116 @@ mod test { // Polling resolves the copy's subscription, which names its track to the frontier. drain(&mut sub); write(&a, 0, b"a0"); - assert_eq!(weight(), 2); + assert_eq!(lag().total(), 2); // Produced before the switch but not sampled yet: still this reader's media. write(&a, 1, b"a1"); let b = copy(); routes.serve(b.consume()); + // Retire A before B has a frame. Those bytes are recorded now, at A's edge + // (500ms behind the frame the subscription opened on), not dropped and not + // held for B's lag. drain(&mut sub); + let pending = lag(); + assert_eq!(pending.total(), 4); + assert_eq!(pending.buckets()[4], 2); + write(&b, 2, b"b2"); - assert_eq!(weight(), 6); + assert_eq!(lag().total(), 6); // The replaced route keeps publishing: not this reader's media. write(&a, 3, b"a3"); - assert_eq!(weight(), 6); + assert_eq!(lag().total(), 6); + } + + /// A copy replaced before it produces is unwatched when it is dropped, so what + /// that track writes afterwards does not move the histogram. + #[test] + fn lag_ignores_a_track_replaced_before_its_first_frame() { + let registry = stats::Registry::new(stats::Config::new()); + let session = registry.tier(stats::Tier::default()).session("root"); + let weight = || { + let mut report = stats::Report::default(); + registry.report(&mut report); + report.traffic[0].publisher.lag.total() + }; + let write = |track: &track::Producer, sequence: u64, payload: &[u8]| { + let mut group = track.create_group(group::Info { sequence }).unwrap(); + group.write_frame(ts(sequence * 500), payload).unwrap(); + group.finish().unwrap(); + }; + let drain = |sub: &mut track::Subscriber| while sub.recv_group().now_or_never().is_some() {}; + + let routes = Producer::new(); + let logical = logical(&routes); + let a = copy(); + routes.serve(a.consume()); + let mut sub = logical + .consume() + .with_stats(session.egress("demo")) + .subscribe(Subscription::default().with_max_age(Duration::from_secs(10))) + .now_or_never() + .unwrap() + .unwrap(); + drain(&mut sub); + + let b = copy(); + routes.serve(b.consume()); + drain(&mut sub); + write(&a, 0, b"retired-later"); + assert_eq!(weight(), 0); + + write(&b, 1, b"live"); + assert_eq!(weight(), 4); + } + + /// A replaced copy stays on the sampler while a group read from it is still out, + /// and what the track writes after that copy is dropped does not. + #[test] + fn lag_weighs_a_replaced_copy_until_it_is_dropped() { + let registry = stats::Registry::new(stats::Config::new()); + let session = registry.tier(stats::Tier::default()).session("root"); + let lag = || { + let mut report = stats::Report::default(); + registry.report(&mut report); + report.traffic[0].publisher.lag + }; + let write = |track: &track::Producer, sequence: u64, payload: &[u8]| { + let mut group = track.create_group(group::Info { sequence }).unwrap(); + group.write_frame(ts(sequence * 500), payload).unwrap(); + group.finish().unwrap(); + }; + let drain = |sub: &mut track::Subscriber| while sub.recv_group().now_or_never().is_some() {}; + + let routes = Producer::new(); + let logical = logical(&routes); + let a = copy(); + routes.serve(a.consume()); + let mut sub = logical + .consume() + .with_stats(session.egress("demo")) + .subscribe(Subscription::default().with_max_age(Duration::from_secs(10))) + .now_or_never() + .unwrap() + .unwrap(); + drain(&mut sub); + write(&a, 0, b"aa"); + let held = recv(&mut sub); + assert_eq!(lag().total(), 2); + + let b = copy(); + routes.serve(b.consume()); + // Let the reader notice the switch while the group is still out. + assert!(sub.recv_group().now_or_never().is_none()); + write(&a, 1, b"bbbb"); + let while_held = lag(); + assert_eq!(while_held.total(), 6); + assert_eq!(while_held.buckets()[4], 4); + + drop(held); + drain(&mut sub); + write(&a, 2, b"cccccc"); + assert_eq!(lag().total(), 6); } #[test] diff --git a/rs/moq-net/src/stats.rs b/rs/moq-net/src/stats.rs index bd3328a55f..24dd17adbd 100644 --- a/rs/moq-net/src/stats.rs +++ b/rs/moq-net/src/stats.rs @@ -1824,8 +1824,18 @@ struct FrontierState { behind: Option, /// The tracks feeding the subscription: its own, or the serving route's copy. sources: Vec, - /// Bytes an unwatched source produced since the last sample, still owed to it. - unsampled: u64, + /// Retired sources. Each keeps the bytes produced since its last sample and the + /// edge those bytes are weighed at, so a sample can record them before a + /// replacement has a frame, and so they never take the replacement's lag. + folded: Vec, +} + +/// Bytes a retired source produced before it stopped feeding the subscription, +/// weighed at the edge it had then. Later bytes on that track are not in here. +struct Folded { + weight: u64, + newest: Duration, + first: Duration, } /// A track feeding a [`Frontier`], with the produced bytes already sampled. @@ -1847,7 +1857,7 @@ impl Frontier { advanced: crate::model::clock::now(), behind: None, sources: Vec::new(), - unsampled: 0, + folded: Vec::new(), }), }); counters @@ -1879,21 +1889,29 @@ impl Frontier { } /// Remove a track that no longer feeds this subscription, such as a route's copy - /// replaced by a switch, so what it produces from then on stops counting. + /// dropped once the reader is done with it. Bytes it already produced stay, at + /// the edge it had; anything it produces afterwards does not. pub(crate) fn unwatch(&self, track: &Arc) { let Some(inner) = &self.0 else { return }; let mut state = inner.state.lock().expect("stats frontier poisoned"); let production = track.production(); - let mut unsampled = 0; - state.sources.retain(|s| { - if !Arc::ptr_eq(&s.production, production) { + let mut folded = Vec::new(); + state.sources.retain(|source| { + if !Arc::ptr_eq(&source.production, production) { return true; } - // What it produced before the cap reached this reader: keep it for the next sample. - unsampled += production.bytes.load(Ordering::Relaxed).saturating_sub(s.sampled); + // Acquire pairs with `Production::record`, so the edge covers these bytes. + let bytes = production.bytes.load(Ordering::Acquire); + let weight = bytes.saturating_sub(source.sampled); + if weight > 0 + && let Some(newest) = Production::load(&production.newest) + && let Some(first) = Production::load(&production.first) + { + folded.push(Folded { weight, newest, first }); + } false }); - state.unsampled += unsampled; + state.folded.extend(folded); } /// Start tracking one group stream written toward the peer. @@ -1905,12 +1923,47 @@ impl Frontier { } } +impl FrontierState { + /// Lag of `newest` against the acknowledged frontier, including how long that + /// frontier has stood still. Does not move the stall mark; [`Self::observe`] does. + fn lag(&self, newest: Duration, now: Instant) -> Duration { + let Some(acked) = self.acked else { + return Duration::ZERO; + }; + let behind = newest.saturating_sub(acked); + if behind.is_zero() { + return Duration::ZERO; + } + // Media time alone trusts the publisher's timestamps; a frontier standing + // still on the wall clock while there is newer media is lag too. + let since = self.behind.unwrap_or(now).max(self.advanced); + behind.max(now.duration_since(since)) + } + + /// Remember whether `newest` leaves the frontier behind, for the next sample's + /// wall-clock lag. + fn observe(&mut self, newest: Duration, now: Instant) { + let Some(acked) = self.acked else { + return; + }; + if newest.saturating_sub(acked).is_zero() { + self.behind = None; + return; + } + let since = self.behind.unwrap_or(now).max(self.advanced); + self.behind = Some(since); + } +} + impl FrontierInner { - /// One sample: the lag and the bytes produced since the last one, or `None` when - /// nothing was produced (a paused source moves nothing). + /// One sample of what the live sources produced since the last one, or `None` + /// when they produced nothing (a paused source moves nothing). Retired sources + /// are recorded here as well, each at its own edge, including when no live + /// source has a frame yet and including the sample taken on drop. fn sample(&self, now: Instant) -> Option<(Duration, u64)> { let mut state = self.state.lock().expect("stats frontier poisoned"); - let mut weight = std::mem::take(&mut state.unsampled); + let folded = std::mem::take(&mut state.folded); + let mut weight = 0; let mut newest = None; let mut first: Option = None; state.sources.retain_mut(|source| { @@ -1928,24 +1981,35 @@ impl FrontierInner { } live }); - let newest = newest?; - // Opened before the track produced anything: its first frame is where the - // subscription started. - let acked = *state.acked.get_or_insert(first?); + // Opened before anything produced: the earliest frame is where it started. + if state.acked.is_none() { + let mut seed = first; + for item in &folded { + seed = Some(seed.map_or(item.first, |seed| seed.min(item.first))); + } + state.acked = seed; + } + + // A retired source's bytes are not the replacement's. Record them even when + // the replacement has no frame, and never at the replacement's lag. + for item in &folded { + let lag = state.lag(item.newest, now); + self.counters.publisher.lag.record(lag, item.weight); + } + + let Some(newest) = newest else { + if let Some(edge) = folded.iter().map(|item| item.newest).max() { + state.observe(edge, now); + } + return None; + }; + // A live edge with no first frame and no acknowledged start cannot be placed. + state.acked?; + let lag = state.lag(newest, now); // Tracked even without weight, so a pause the peer caught up in doesn't read // as a stalled frontier once production resumes. - let behind = newest.saturating_sub(acked); - let lag = if behind.is_zero() { - state.behind = None; - Duration::ZERO - } else { - // Media time alone trusts the publisher's timestamps; a frontier standing - // still on the wall clock while there is newer media is lag too. - let since = state.behind.unwrap_or(now).max(state.advanced); - state.behind = Some(since); - behind.max(now.duration_since(since)) - }; + state.observe(newest, now); (weight > 0).then_some((lag, weight)) } } From 9b7845ad25a6741e725ad1e8568944b3ae6adf01 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 21:50:11 -0700 Subject: [PATCH 2/3] fix(net): weigh a replaced copy only up to its cap A switch caps a replaced copy at the group it had, but the copy stayed on the sampler as a whole track until dropped, so groups it produced past the cap, which this reader never receives, still added bytes and moved the edge. The sampler now counts a capped copy through a share of its track that only records groups up to the cap. Frames still written into those groups count; later groups do not. Readers capped at the same group share one tally, so a frame write costs one counter per distinct cap, and one atomic load on a track nobody capped. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/group.rs | 6 +- rs/moq-net/src/model/resume.rs | 95 ++++++++++++++++----- rs/moq-net/src/model/track.rs | 10 +++ rs/moq-net/src/stats.rs | 145 +++++++++++++++++++++++++++------ 4 files changed, 211 insertions(+), 45 deletions(-) diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index f835d333d7..5d24eac1d2 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -544,7 +544,7 @@ impl Producer { // Ingress payload: one whole frame written. self.stats.frames(1); self.stats.bytes(size); - self.cache.production().record(timestamp, size); + self.cache.production().record(self.info.sequence, timestamp, size); Ok(()) } @@ -597,7 +597,7 @@ impl Producer { state.cache += size; now = state.charge.add(size); state.stamp(frame.timestamp); - self.cache.production().record(frame.timestamp, size); + self.cache.production().record(self.info.sequence, frame.timestamp, size); state.frames.push_back(frame); } state.next_index = next_index; @@ -697,7 +697,7 @@ impl Producer { // Ingress payload: one frame opened; its bytes are counted per chunk as the // producer writes them. Production counts the declared size up front. self.stats.frames(1); - self.cache.production().record(timestamp, frame.size); + self.cache.production().record(self.info.sequence, timestamp, frame.size); Ok(frame::Info { size: frame.size, diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 426efed200..c0abaa945c 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -287,10 +287,8 @@ impl Copy { let mut sub = ready!(pending.poll_ok(waiter))?; sub.raise_start_to(groups.0); sub.end_at(groups.1); - // Weigh the copy while the reader can still take groups from it. A switch - // does not stop that; `Reader::let_go` does, once the copy is dropped. - sub.watch(&self.frontier); self.sub = Sub::Ready(sub); + self.watch(); } let Sub::Ready(sub) = &mut self.sub else { unreachable!() }; Poll::Ready(Ok(sub)) @@ -301,6 +299,18 @@ impl Copy { serving || (self.done.is_none() && Arc::strong_count(&self.lease) > 1) } + /// Weigh this copy's track while the reader can still take groups from it: all of + /// it while it serves, then only the groups up to `until`. `Reader::let_go` stops + /// that once the copy is dropped. A pending subscription is watched once it answers. + fn watch(&mut self) { + if let Sub::Ready(sub) = &mut self.sub { + sub.watch(&self.frontier); + if let Some(until) = self.until { + sub.cap(&self.frontier, until); + } + } + } + /// Stop weighing this copy. A pending subscription was never watched. fn unwatch(&mut self) { if let Sub::Ready(sub) = &mut self.sub { @@ -376,16 +386,14 @@ impl Subscriber { self.reader().latest() } - /// Report every copy still held to `frontier`, including one a switch capped but - /// the reader can still take groups from. + /// Report every copy still held to `frontier`, a replaced one only up to the group + /// its switch capped it at. pub(crate) fn watch(&mut self, frontier: &stats::Frontier) { let mut reader = self.reader(); reader.frontier = frontier.clone(); for copy in &mut reader.copies { copy.frontier = frontier.clone(); - if let Sub::Ready(sub) = &mut copy.sub { - sub.watch(frontier); - } + copy.watch(); } } @@ -483,13 +491,13 @@ impl Reader { if self.copies.last().is_some_and(|copy| copy.generation == generation) { return; } - // The replaced copies finish the groups they had when replaced, and nothing newer. - // They keep weighing lag while those groups are still readable; `let_go` stops - // that once the copy is dropped. + // The replaced copies finish the groups they had when replaced, and nothing newer, + // so only those groups keep weighing lag. for copy in &mut self.copies { if copy.until.is_none() { copy.until = Some(self.newest.max(copy.track.latest())); copy.update(&self.mirrored); + copy.watch(); } } let floor = self.newest.map(Position::group); @@ -1384,10 +1392,11 @@ mod test { assert_eq!(weight(), 4); } - /// A replaced copy stays on the sampler while a group read from it is still out, - /// and what the track writes after that copy is dropped does not. + /// A replaced copy weighs the groups it had at the switch while one is still out, + /// frames written into them included. Its later groups never reach this reader, and + /// nothing it writes after the copy is dropped counts. #[test] - fn lag_weighs_a_replaced_copy_until_it_is_dropped() { + fn lag_weighs_a_replaced_copy_up_to_its_cap() { let registry = stats::Registry::new(stats::Config::new()); let session = registry.tier(stats::Tier::default()).session("root"); let lag = || { @@ -1414,7 +1423,8 @@ mod test { .unwrap() .unwrap(); drain(&mut sub); - write(&a, 0, b"aa"); + let mut a0 = a.create_group(group::Info { sequence: 0 }).unwrap(); + a0.write_frame(ts(0), b"aa").unwrap(); let held = recv(&mut sub); assert_eq!(lag().total(), 2); @@ -1422,15 +1432,62 @@ mod test { routes.serve(b.consume()); // Let the reader notice the switch while the group is still out. assert!(sub.recv_group().now_or_never().is_none()); + // A/0 is capped in and still read; A/1 is past the cap, so neither its bytes nor + // its edge count: the frame in A/0 is weighed 300ms behind, not 500ms. + a0.write_frame(ts(300), b"cc").unwrap(); write(&a, 1, b"bbbb"); let while_held = lag(); - assert_eq!(while_held.total(), 6); - assert_eq!(while_held.buckets()[4], 4); + assert_eq!(while_held.total(), 4); + assert_eq!(while_held.buckets()[3], 2); drop(held); drain(&mut sub); - write(&a, 2, b"cccccc"); - assert_eq!(lag().total(), 6); + a0.write_frame(ts(400), b"dddd").unwrap(); + write(&a, 2, b"eeeeee"); + assert_eq!(lag().total(), 4); + } + + /// Readers capped at the same group share one count, and each still weighs every + /// frame written into it. + #[test] + fn lag_weighs_a_shared_cap_once_per_reader() { + let registry = stats::Registry::new(stats::Config::new()); + let session = registry.tier(stats::Tier::default()).session("root"); + let weight = || { + let mut report = stats::Report::default(); + registry.report(&mut report); + report.traffic[0].publisher.lag.total() + }; + + let routes = Producer::new(); + let logical = logical(&routes); + let a = copy(); + routes.serve(a.consume()); + let subscribe = || { + logical + .consume() + .with_stats(session.egress("demo")) + .subscribe(Subscription::default().with_max_age(Duration::from_secs(10))) + .now_or_never() + .unwrap() + .unwrap() + }; + let mut first = subscribe(); + let mut second = subscribe(); + // Polling resolves each copy's subscription, which names its track to the frontier. + assert!(first.recv_group().now_or_never().is_none()); + assert!(second.recv_group().now_or_never().is_none()); + let mut a0 = a.create_group(group::Info { sequence: 0 }).unwrap(); + a0.write_frame(ts(0), b"aa").unwrap(); + let _first = recv(&mut first); + let _second = recv(&mut second); + assert_eq!(weight(), 4); + + routes.serve(copy().consume()); + assert!(first.recv_group().now_or_never().is_none()); + assert!(second.recv_group().now_or_never().is_none()); + a0.write_frame(ts(100), b"cc").unwrap(); + assert_eq!(weight(), 8); } #[test] diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index b9b64065d6..3cf795cfd1 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -4051,6 +4051,16 @@ impl Subscriber { } } + /// Report only this cursor's groups up to `cap` (`None` for none) to `frontier` + /// from now on, as a replaced route's copy. A front's logical track is never a + /// route's copy, so it is simply unwatched. + pub(crate) fn cap(&mut self, frontier: &stats::Frontier, cap: Option) { + match &mut self.inner { + Inner::Plain(cursor) => frontier.cap(&cursor.state.read().cache, cap), + Inner::Resume(resume, _) => resume.unwatch(frontier), + } + } + /// Stop reporting the tracks feeding this cursor to `frontier`. pub(crate) fn unwatch(&mut self, frontier: &stats::Frontier) { match &mut self.inner { diff --git a/rs/moq-net/src/stats.rs b/rs/moq-net/src/stats.rs index 24dd17adbd..b06ba236f0 100644 --- a/rs/moq-net/src/stats.rs +++ b/rs/moq-net/src/stats.rs @@ -125,7 +125,7 @@ use std::{ fmt, sync::{ Arc, Mutex, Weak, - atomic::{AtomicU64, Ordering}, + atomic::{AtomicBool, AtomicU64, Ordering}, }, time::Duration, }; @@ -1775,6 +1775,60 @@ fn media_time(timestamp: Timestamp) -> Duration { /// frame write, whether or not the track is tagged, since its readers may be. #[derive(Default, Debug)] pub(crate) struct Production { + tally: Tally, + // The groups up to each cap a switch put on this track, since later ones never + // reach those readers. One per distinct cap, shared by every reader capped there. + shares: Mutex>>, + // Whether `shares` may be non-empty, so a track nobody capped skips the lock. + shared: AtomicBool, +} + +impl Production { + /// Record one frame of `bytes` at `timestamp` in group `sequence`. + pub(crate) fn record(&self, sequence: u64, timestamp: Timestamp, bytes: u64) { + // Shares before the track: a frame racing a new cap may go uncounted, but is + // never counted twice. See `Frontier::cap`. + if self.shared.load(Ordering::Acquire) { + let mut shares = self.shares.lock().expect("stats production poisoned"); + shares.retain(|share| { + let Some(share) = share.upgrade() else { return false }; + if share.cap.is_some_and(|cap| sequence <= cap) { + share.tally.record(timestamp, bytes); + } + true + }); + self.shared.store(!shares.is_empty(), Ordering::Release); + } + self.tally.record(timestamp, bytes); + } + + /// What this track produces from now on in groups up to `cap` (`None` for none). + fn share(&self, cap: Option) -> Arc { + let mut shares = self.shares.lock().expect("stats production poisoned"); + if let Some(share) = shares.iter().filter_map(Weak::upgrade).find(|share| share.cap == cap) { + return share; + } + // Every group produced so far is up to the cap, so the edges carry over. + let tally = Tally::default(); + tally.first.store(self.tally.first.load(Ordering::Relaxed), Ordering::Relaxed); + tally.newest.store(self.tally.newest.load(Ordering::Relaxed), Ordering::Relaxed); + let share = Arc::new(Share { cap, tally }); + shares.push(Arc::downgrade(&share)); + self.shared.store(true, Ordering::Release); + share + } +} + +/// The part of a track its capped readers can still receive: its groups up to `cap`. +#[derive(Debug)] +struct Share { + cap: Option, + tally: Tally, +} + +/// Frames produced: their payload bytes and the media-time edges they span. +#[derive(Default, Debug)] +struct Tally { // Media-time nanoseconds plus one, so zero means nothing produced yet. first: AtomicU64, newest: AtomicU64, @@ -1782,9 +1836,8 @@ pub(crate) struct Production { bytes: AtomicU64, } -impl Production { - /// Record one frame of `bytes` at `timestamp`. - pub(crate) fn record(&self, timestamp: Timestamp, bytes: u64) { +impl Tally { + fn record(&self, timestamp: Timestamp, bytes: u64) { let at = u64::try_from(media_time(timestamp).as_nanos()) .unwrap_or(u64::MAX) .saturating_add(1); @@ -1822,7 +1875,8 @@ struct FrontierState { advanced: Instant, /// When a sample first found the subscription behind since `advanced`. behind: Option, - /// The tracks feeding the subscription: its own, or the serving route's copy. + /// The tracks feeding the subscription: its own, or the route copies it still + /// reads, a replaced one only up to the group its switch capped it at. sources: Vec, /// Retired sources. Each keeps the bytes produced since its last sample and the /// edge those bytes are weighed at, so a sample can record them before a @@ -1845,9 +1899,34 @@ struct Source { // Strong, so the bytes a track produced before it died are still read, however // its last handles and this frontier's last reference race to drop. production: Arc, + // Set once a switch caps the track: only the groups this reader can still receive. + share: Option>, sampled: u64, } +impl Source { + /// What this source counts: the whole track, or its capped share. + fn tally(&self) -> &Tally { + self.share.as_ref().map_or(&self.production.tally, |share| &share.tally) + } + + /// The bytes counted since the last sample, at the edge they reached. + fn fold(&self) -> Option { + let tally = self.tally(); + // Acquire pairs with `Tally::record`, so the edge covers these bytes. + let bytes = tally.bytes.load(Ordering::Acquire); + let weight = bytes.saturating_sub(self.sampled); + if weight == 0 { + return None; + } + Some(Folded { + weight, + newest: Tally::load(&tally.newest)?, + first: Tally::load(&tally.first)?, + }) + } +} + impl Frontier { fn open(counters: &Arc) -> Self { let inner = Arc::new(FrontierInner { @@ -1879,15 +1958,43 @@ impl Frontier { return; } if state.acked.is_none() { - state.acked = Production::load(&production.newest); + state.acked = Tally::load(&production.tally.newest); } state.sources.push(Source { track: Arc::downgrade(track), production: production.clone(), - sampled: production.bytes.load(Ordering::Relaxed), + share: None, + sampled: production.tally.bytes.load(Ordering::Relaxed), }); } + /// Count only `track`'s groups up to `cap` (`None` for none) from now on: a switch + /// replaced its route, so its later groups never reach this subscription. Frames + /// still written into the earlier groups keep counting. A no-op once capped. + pub(crate) fn cap(&self, track: &Arc, cap: Option) { + let Some(inner) = &self.0 else { return }; + let mut state = inner.state.lock().expect("stats frontier poisoned"); + let production = track.production(); + let Some(source) = state + .sources + .iter_mut() + .find(|source| Arc::ptr_eq(&source.production, production)) + else { + return; + }; + if source.share.is_some() { + return; + } + // What is still unsampled all came from groups up to the cap: fold it at the + // track's edge. The track is read before the share, and a frame records into the + // share before the track, so one racing the cap is never counted twice. + let folded = source.fold(); + let share = production.share(cap); + source.sampled = share.tally.bytes.load(Ordering::Acquire); + source.share = Some(share); + state.folded.extend(folded); + } + /// Remove a track that no longer feeds this subscription, such as a route's copy /// dropped once the reader is done with it. Bytes it already produced stay, at /// the edge it had; anything it produces afterwards does not. @@ -1900,15 +2007,7 @@ impl Frontier { if !Arc::ptr_eq(&source.production, production) { return true; } - // Acquire pairs with `Production::record`, so the edge covers these bytes. - let bytes = production.bytes.load(Ordering::Acquire); - let weight = bytes.saturating_sub(source.sampled); - if weight > 0 - && let Some(newest) = Production::load(&production.newest) - && let Some(first) = Production::load(&production.first) - { - folded.push(Folded { weight, newest, first }); - } + folded.extend(source.fold()); false }); state.folded.extend(folded); @@ -1970,15 +2069,15 @@ impl FrontierInner { // Checked before reading: every writer holds the track, so a dead one has // already recorded everything it ever will, and this read is its last. let live = source.track.strong_count() > 0; - let production = &source.production; - // Acquire pairs with `Production::record`: the edges read below cover these bytes. - let bytes = production.bytes.load(Ordering::Acquire); - weight += bytes.saturating_sub(source.sampled); - source.sampled = bytes; - newest = newest.max(Production::load(&production.newest)); - if let Some(at) = Production::load(&production.first) { + let tally = source.tally(); + // Acquire pairs with `Tally::record`: the edges read below cover these bytes. + let bytes = tally.bytes.load(Ordering::Acquire); + newest = newest.max(Tally::load(&tally.newest)); + if let Some(at) = Tally::load(&tally.first) { first = Some(first.map_or(at, |first| first.min(at))); } + weight += bytes.saturating_sub(source.sampled); + source.sampled = bytes; live }); From cbc522b39024aa7795e201c2f4900fa3766e022f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 22:36:49 -0700 Subject: [PATCH 3/3] fix(net): unwatch a replaced copy at the switch again Drop the per-cap shares in favor of the base behavior: a switch unwatches the replaced copy and records its unsampled bytes at its edge then. Frames still written into a group the reader holds from it go unweighed, an approximation that keeps per-group accounting off the frame path. This also keeps a route that comes back (A to B to A) weighed: the live copy is always watched, and dropping the old copy no longer removes it. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/group.rs | 6 +- rs/moq-net/src/model/resume.rs | 176 +++++++++++++++------------------ rs/moq-net/src/model/track.rs | 10 -- rs/moq-net/src/stats.rs | 149 +++++----------------------- 4 files changed, 108 insertions(+), 233 deletions(-) diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index 5d24eac1d2..f835d333d7 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -544,7 +544,7 @@ impl Producer { // Ingress payload: one whole frame written. self.stats.frames(1); self.stats.bytes(size); - self.cache.production().record(self.info.sequence, timestamp, size); + self.cache.production().record(timestamp, size); Ok(()) } @@ -597,7 +597,7 @@ impl Producer { state.cache += size; now = state.charge.add(size); state.stamp(frame.timestamp); - self.cache.production().record(self.info.sequence, frame.timestamp, size); + self.cache.production().record(frame.timestamp, size); state.frames.push_back(frame); } state.next_index = next_index; @@ -697,7 +697,7 @@ impl Producer { // Ingress payload: one frame opened; its bytes are counted per chunk as the // producer writes them. Production counts the declared size up front. self.stats.frames(1); - self.cache.production().record(self.info.sequence, timestamp, frame.size); + self.cache.production().record(timestamp, frame.size); Ok(frame::Info { size: frame.size, diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index c0abaa945c..61e9916aa0 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -287,8 +287,11 @@ impl Copy { let mut sub = ready!(pending.poll_ok(waiter))?; sub.raise_start_to(groups.0); sub.end_at(groups.1); + // Only a copy that still serves weighs lag; see `Reader::sync`. + if self.until.is_none() { + sub.watch(&self.frontier); + } self.sub = Sub::Ready(sub); - self.watch(); } let Sub::Ready(sub) = &mut self.sub else { unreachable!() }; Poll::Ready(Ok(sub)) @@ -298,25 +301,6 @@ impl Copy { fn needed(&self, serving: bool) -> bool { serving || (self.done.is_none() && Arc::strong_count(&self.lease) > 1) } - - /// Weigh this copy's track while the reader can still take groups from it: all of - /// it while it serves, then only the groups up to `until`. `Reader::let_go` stops - /// that once the copy is dropped. A pending subscription is watched once it answers. - fn watch(&mut self) { - if let Sub::Ready(sub) = &mut self.sub { - sub.watch(&self.frontier); - if let Some(until) = self.until { - sub.cap(&self.frontier, until); - } - } - } - - /// Stop weighing this copy. A pending subscription was never watched. - fn unwatch(&mut self) { - if let Sub::Ready(sub) = &mut self.sub { - sub.unwatch(&self.frontier); - } - } } /// One reader's subscription to a logical track; see the module docs. @@ -386,24 +370,29 @@ impl Subscriber { self.reader().latest() } - /// Report every copy still held to `frontier`, a replaced one only up to the group - /// its switch capped it at. + /// Report the serving copy's track, now and after every switch, to `frontier`. pub(crate) fn watch(&mut self, frontier: &stats::Frontier) { let mut reader = self.reader(); reader.frontier = frontier.clone(); for copy in &mut reader.copies { copy.frontier = frontier.clone(); - copy.watch(); + if copy.until.is_none() + && let Sub::Ready(sub) = &mut copy.sub + { + sub.watch(frontier); + } } } - /// Stop reporting every copy's track to `frontier`. + /// Stop reporting any copy's track to `frontier`, now or after a switch. pub(crate) fn unwatch(&mut self, frontier: &stats::Frontier) { let mut reader = self.reader(); reader.frontier = stats::Frontier::default(); for copy in &mut reader.copies { copy.frontier = stats::Frontier::default(); - if let Sub::Ready(sub) = &mut copy.sub { + if copy.until.is_none() + && let Sub::Ready(sub) = &mut copy.sub + { sub.unwatch(frontier); } } @@ -491,13 +480,18 @@ impl Reader { if self.copies.last().is_some_and(|copy| copy.generation == generation) { return; } - // The replaced copies finish the groups they had when replaced, and nothing newer, - // so only those groups keep weighing lag. + // The replaced copies finish the groups they had when replaced, and nothing newer. for copy in &mut self.copies { if copy.until.is_none() { copy.until = Some(self.newest.max(copy.track.latest())); copy.update(&self.mirrored); - copy.watch(); + // A replaced route's copy may live on; what it produces from now on never + // reaches this reader, so it stops weighing lag. Frames still written into a + // group the reader holds from it go unweighed too: an approximation that + // keeps per-group accounting off the frame path. + if let Sub::Ready(sub) = &mut copy.sub { + sub.unwatch(©.frontier); + } } } let floor = self.newest.map(Position::group); @@ -531,30 +525,21 @@ impl Reader { if copy.needed(Some(copy.generation) == serving) { return true; } - let keep = if copy.done.is_some() { - false - } else { - let until = copy.until; - match copy.poll_ready(groups, &kio::Waiter::noop()) { - Poll::Ready(Ok(sub)) => { - // Inspect without moving either cursor: a later group poll still chooses its - // order and applies the current bounds and drift budget. Ignore the local - // cap here, since unread groups become eligible again when it rises. - sub.has_unread_group(&track::Unread { - start: groups.0.max(ordered.map_or(0, |last| last.saturating_add(1))), - end: until.and_then(|until| until.map_or(Some(0), |last| last.checked_add(1))), - delivered, - }) - } - _ => false, - } - }; - // Unwatched on every retire, not only when a switch caps the copy. While the - // copy stays, the reader can still take what it already had. - if !keep { - copy.unwatch(); + if copy.done.is_some() { + return false; } - keep + let until = copy.until; + let Poll::Ready(Ok(sub)) = copy.poll_ready(groups, &kio::Waiter::noop()) else { + return false; + }; + // Inspect without moving either cursor: a later group poll still chooses its + // order and applies the current bounds and drift budget. Ignore the local + // cap here, since unread groups become eligible again when it rises. + sub.has_unread_group(&track::Unread { + start: groups.0.max(ordered.map_or(0, |last| last.saturating_add(1))), + end: until.and_then(|until| until.map_or(Some(0), |last| last.checked_add(1))), + delivered, + }) }); } @@ -1351,7 +1336,7 @@ mod test { assert_eq!(lag().total(), 6); } - /// A copy replaced before it produces is unwatched when it is dropped, so what + /// A copy replaced before it produces is unwatched at the switch, so what /// that track writes afterwards does not move the histogram. #[test] fn lag_ignores_a_track_replaced_before_its_first_frame() { @@ -1392,11 +1377,11 @@ mod test { assert_eq!(weight(), 4); } - /// A replaced copy weighs the groups it had at the switch while one is still out, - /// frames written into them included. Its later groups never reach this reader, and - /// nothing it writes after the copy is dropped counts. + /// A replaced copy stops counting at the switch, even while a group read from it is + /// still out: frames written into that group afterwards go unweighed, as do its + /// later groups. What it produced before the switch still counts, at its edge. #[test] - fn lag_weighs_a_replaced_copy_up_to_its_cap() { + fn lag_stops_weighing_a_replaced_copy_at_the_switch() { let registry = stats::Registry::new(stats::Config::new()); let session = registry.tier(stats::Tier::default()).session("root"); let lag = || { @@ -1428,29 +1413,24 @@ mod test { let held = recv(&mut sub); assert_eq!(lag().total(), 2); - let b = copy(); - routes.serve(b.consume()); - // Let the reader notice the switch while the group is still out. + // Written before the switch but not sampled yet: recorded at A's edge, 300ms in. + a0.write_frame(ts(300), b"bb").unwrap(); + routes.serve(copy().consume()); assert!(sub.recv_group().now_or_never().is_none()); - // A/0 is capped in and still read; A/1 is past the cap, so neither its bytes nor - // its edge count: the frame in A/0 is weighed 300ms behind, not 500ms. - a0.write_frame(ts(300), b"cc").unwrap(); - write(&a, 1, b"bbbb"); - let while_held = lag(); - assert_eq!(while_held.total(), 4); - assert_eq!(while_held.buckets()[3], 2); + let switched = lag(); + assert_eq!(switched.total(), 4); + assert_eq!(switched.buckets()[3], 2); - drop(held); - drain(&mut sub); - a0.write_frame(ts(400), b"dddd").unwrap(); - write(&a, 2, b"eeeeee"); + a0.write_frame(ts(400), b"cccc").unwrap(); + write(&a, 1, b"dddddd"); assert_eq!(lag().total(), 4); + drop(held); } - /// Readers capped at the same group share one count, and each still weighs every - /// frame written into it. + /// A route that comes back while a group from its old copy is still out is weighed + /// again, and dropping the old copy does not silence it. #[test] - fn lag_weighs_a_shared_cap_once_per_reader() { + fn lag_weighs_a_route_that_comes_back() { let registry = stats::Registry::new(stats::Config::new()); let session = registry.tier(stats::Tier::default()).session("root"); let weight = || { @@ -1458,36 +1438,40 @@ mod test { registry.report(&mut report); report.traffic[0].publisher.lag.total() }; + let write = |track: &track::Producer, sequence: u64, payload: &[u8]| { + let mut group = track.create_group(group::Info { sequence }).unwrap(); + group.write_frame(ts(sequence * 500), payload).unwrap(); + group.finish().unwrap(); + }; + let drain = |sub: &mut track::Subscriber| while sub.recv_group().now_or_never().is_some() {}; let routes = Producer::new(); let logical = logical(&routes); let a = copy(); routes.serve(a.consume()); - let subscribe = || { - logical - .consume() - .with_stats(session.egress("demo")) - .subscribe(Subscription::default().with_max_age(Duration::from_secs(10))) - .now_or_never() - .unwrap() - .unwrap() - }; - let mut first = subscribe(); - let mut second = subscribe(); - // Polling resolves each copy's subscription, which names its track to the frontier. - assert!(first.recv_group().now_or_never().is_none()); - assert!(second.recv_group().now_or_never().is_none()); - let mut a0 = a.create_group(group::Info { sequence: 0 }).unwrap(); - a0.write_frame(ts(0), b"aa").unwrap(); - let _first = recv(&mut first); - let _second = recv(&mut second); - assert_eq!(weight(), 4); + let mut sub = logical + .consume() + .with_stats(session.egress("demo")) + .subscribe(Subscription::default().with_max_age(Duration::from_secs(10))) + .now_or_never() + .unwrap() + .unwrap(); + drain(&mut sub); + write(&a, 0, b"aa"); + let held = recv(&mut sub); + assert_eq!(weight(), 2); routes.serve(copy().consume()); - assert!(first.recv_group().now_or_never().is_none()); - assert!(second.recv_group().now_or_never().is_none()); - a0.write_frame(ts(100), b"cc").unwrap(); - assert_eq!(weight(), 8); + drain(&mut sub); + routes.serve(a.consume()); + drain(&mut sub); + write(&a, 1, b"bbbb"); + assert_eq!(weight(), 6); + + drop(held); + drain(&mut sub); + write(&a, 2, b"cccccc"); + assert_eq!(weight(), 12); } #[test] diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 3cf795cfd1..b9b64065d6 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -4051,16 +4051,6 @@ impl Subscriber { } } - /// Report only this cursor's groups up to `cap` (`None` for none) to `frontier` - /// from now on, as a replaced route's copy. A front's logical track is never a - /// route's copy, so it is simply unwatched. - pub(crate) fn cap(&mut self, frontier: &stats::Frontier, cap: Option) { - match &mut self.inner { - Inner::Plain(cursor) => frontier.cap(&cursor.state.read().cache, cap), - Inner::Resume(resume, _) => resume.unwatch(frontier), - } - } - /// Stop reporting the tracks feeding this cursor to `frontier`. pub(crate) fn unwatch(&mut self, frontier: &stats::Frontier) { match &mut self.inner { diff --git a/rs/moq-net/src/stats.rs b/rs/moq-net/src/stats.rs index b06ba236f0..5d823cff9e 100644 --- a/rs/moq-net/src/stats.rs +++ b/rs/moq-net/src/stats.rs @@ -125,7 +125,7 @@ use std::{ fmt, sync::{ Arc, Mutex, Weak, - atomic::{AtomicBool, AtomicU64, Ordering}, + atomic::{AtomicU64, Ordering}, }, time::Duration, }; @@ -1775,60 +1775,6 @@ fn media_time(timestamp: Timestamp) -> Duration { /// frame write, whether or not the track is tagged, since its readers may be. #[derive(Default, Debug)] pub(crate) struct Production { - tally: Tally, - // The groups up to each cap a switch put on this track, since later ones never - // reach those readers. One per distinct cap, shared by every reader capped there. - shares: Mutex>>, - // Whether `shares` may be non-empty, so a track nobody capped skips the lock. - shared: AtomicBool, -} - -impl Production { - /// Record one frame of `bytes` at `timestamp` in group `sequence`. - pub(crate) fn record(&self, sequence: u64, timestamp: Timestamp, bytes: u64) { - // Shares before the track: a frame racing a new cap may go uncounted, but is - // never counted twice. See `Frontier::cap`. - if self.shared.load(Ordering::Acquire) { - let mut shares = self.shares.lock().expect("stats production poisoned"); - shares.retain(|share| { - let Some(share) = share.upgrade() else { return false }; - if share.cap.is_some_and(|cap| sequence <= cap) { - share.tally.record(timestamp, bytes); - } - true - }); - self.shared.store(!shares.is_empty(), Ordering::Release); - } - self.tally.record(timestamp, bytes); - } - - /// What this track produces from now on in groups up to `cap` (`None` for none). - fn share(&self, cap: Option) -> Arc { - let mut shares = self.shares.lock().expect("stats production poisoned"); - if let Some(share) = shares.iter().filter_map(Weak::upgrade).find(|share| share.cap == cap) { - return share; - } - // Every group produced so far is up to the cap, so the edges carry over. - let tally = Tally::default(); - tally.first.store(self.tally.first.load(Ordering::Relaxed), Ordering::Relaxed); - tally.newest.store(self.tally.newest.load(Ordering::Relaxed), Ordering::Relaxed); - let share = Arc::new(Share { cap, tally }); - shares.push(Arc::downgrade(&share)); - self.shared.store(true, Ordering::Release); - share - } -} - -/// The part of a track its capped readers can still receive: its groups up to `cap`. -#[derive(Debug)] -struct Share { - cap: Option, - tally: Tally, -} - -/// Frames produced: their payload bytes and the media-time edges they span. -#[derive(Default, Debug)] -struct Tally { // Media-time nanoseconds plus one, so zero means nothing produced yet. first: AtomicU64, newest: AtomicU64, @@ -1836,8 +1782,9 @@ struct Tally { bytes: AtomicU64, } -impl Tally { - fn record(&self, timestamp: Timestamp, bytes: u64) { +impl Production { + /// Record one frame of `bytes` at `timestamp`. + pub(crate) fn record(&self, timestamp: Timestamp, bytes: u64) { let at = u64::try_from(media_time(timestamp).as_nanos()) .unwrap_or(u64::MAX) .saturating_add(1); @@ -1875,8 +1822,7 @@ struct FrontierState { advanced: Instant, /// When a sample first found the subscription behind since `advanced`. behind: Option, - /// The tracks feeding the subscription: its own, or the route copies it still - /// reads, a replaced one only up to the group its switch capped it at. + /// The tracks feeding the subscription: its own, or the serving route's copy. sources: Vec, /// Retired sources. Each keeps the bytes produced since its last sample and the /// edge those bytes are weighed at, so a sample can record them before a @@ -1899,34 +1845,9 @@ struct Source { // Strong, so the bytes a track produced before it died are still read, however // its last handles and this frontier's last reference race to drop. production: Arc, - // Set once a switch caps the track: only the groups this reader can still receive. - share: Option>, sampled: u64, } -impl Source { - /// What this source counts: the whole track, or its capped share. - fn tally(&self) -> &Tally { - self.share.as_ref().map_or(&self.production.tally, |share| &share.tally) - } - - /// The bytes counted since the last sample, at the edge they reached. - fn fold(&self) -> Option { - let tally = self.tally(); - // Acquire pairs with `Tally::record`, so the edge covers these bytes. - let bytes = tally.bytes.load(Ordering::Acquire); - let weight = bytes.saturating_sub(self.sampled); - if weight == 0 { - return None; - } - Some(Folded { - weight, - newest: Tally::load(&tally.newest)?, - first: Tally::load(&tally.first)?, - }) - } -} - impl Frontier { fn open(counters: &Arc) -> Self { let inner = Arc::new(FrontierInner { @@ -1958,46 +1879,18 @@ impl Frontier { return; } if state.acked.is_none() { - state.acked = Tally::load(&production.tally.newest); + state.acked = Production::load(&production.newest); } state.sources.push(Source { track: Arc::downgrade(track), production: production.clone(), - share: None, - sampled: production.tally.bytes.load(Ordering::Relaxed), + sampled: production.bytes.load(Ordering::Relaxed), }); } - /// Count only `track`'s groups up to `cap` (`None` for none) from now on: a switch - /// replaced its route, so its later groups never reach this subscription. Frames - /// still written into the earlier groups keep counting. A no-op once capped. - pub(crate) fn cap(&self, track: &Arc, cap: Option) { - let Some(inner) = &self.0 else { return }; - let mut state = inner.state.lock().expect("stats frontier poisoned"); - let production = track.production(); - let Some(source) = state - .sources - .iter_mut() - .find(|source| Arc::ptr_eq(&source.production, production)) - else { - return; - }; - if source.share.is_some() { - return; - } - // What is still unsampled all came from groups up to the cap: fold it at the - // track's edge. The track is read before the share, and a frame records into the - // share before the track, so one racing the cap is never counted twice. - let folded = source.fold(); - let share = production.share(cap); - source.sampled = share.tally.bytes.load(Ordering::Acquire); - source.share = Some(share); - state.folded.extend(folded); - } - /// Remove a track that no longer feeds this subscription, such as a route's copy - /// dropped once the reader is done with it. Bytes it already produced stay, at - /// the edge it had; anything it produces afterwards does not. + /// replaced by a switch. Bytes it already produced stay, at the edge it had; + /// anything it produces afterwards does not. pub(crate) fn unwatch(&self, track: &Arc) { let Some(inner) = &self.0 else { return }; let mut state = inner.state.lock().expect("stats frontier poisoned"); @@ -2007,7 +1900,15 @@ impl Frontier { if !Arc::ptr_eq(&source.production, production) { return true; } - folded.extend(source.fold()); + // Acquire pairs with `Production::record`, so the edge covers these bytes. + let bytes = production.bytes.load(Ordering::Acquire); + let weight = bytes.saturating_sub(source.sampled); + if weight > 0 + && let Some(newest) = Production::load(&production.newest) + && let Some(first) = Production::load(&production.first) + { + folded.push(Folded { weight, newest, first }); + } false }); state.folded.extend(folded); @@ -2069,15 +1970,15 @@ impl FrontierInner { // Checked before reading: every writer holds the track, so a dead one has // already recorded everything it ever will, and this read is its last. let live = source.track.strong_count() > 0; - let tally = source.tally(); - // Acquire pairs with `Tally::record`: the edges read below cover these bytes. - let bytes = tally.bytes.load(Ordering::Acquire); - newest = newest.max(Tally::load(&tally.newest)); - if let Some(at) = Tally::load(&tally.first) { - first = Some(first.map_or(at, |first| first.min(at))); - } + let production = &source.production; + // Acquire pairs with `Production::record`: the edges read below cover these bytes. + let bytes = production.bytes.load(Ordering::Acquire); weight += bytes.saturating_sub(source.sampled); source.sampled = bytes; + newest = newest.max(Production::load(&production.newest)); + if let Some(at) = Production::load(&production.first) { + first = Some(first.map_or(at, |first| first.min(at))); + } live });