diff --git a/doc/concept/stats.md b/doc/concept/stats.md index 2ca7da9015..af17282898 100644 --- a/doc/concept/stats.md +++ b/doc/concept/stats.md @@ -122,7 +122,10 @@ a group stream. Every stats interval the relay samples each subscription's lag, the newest frame the track has produced minus the frontier, or the wall-clock time the frontier has stood still while newer media exists if that is larger. It adds the bytes the track produced during the interval to the bucket that lag -falls in, and the broadcast's subscriptions sum into one entry. +falls in, and the broadcast's subscriptions sum into one entry. A subscription +that ends between two intervals takes one last sample when it closes, so the +bytes since its previous sample are not lost, even if it opened and closed +between two ticks. | Bucket | Lag | | --- | --- | diff --git a/quest/m1/qos/README.md b/quest/m1/qos/README.md index 2028ac97d0..e7ec0b899a 100644 --- a/quest/m1/qos/README.md +++ b/quest/m1/qos/README.md @@ -39,8 +39,6 @@ on `dev`. ## Required -- [Final lag sample](/quest/m1/qos/final-lag-sample.md) - a closing - subscription records its last partial interval instead of losing it - [Lag dashboard](/quest/m1/qos/lag-dashboard.md) - the demo stats dashboard shows viewer lag percentiles and dropped media - [Starvation at frame granularity](/quest/m1/qos/starvation-frames.md) - the diff --git a/quest/m1/qos/final-lag-sample.md b/quest/m1/qos/final-lag-sample.md deleted file mode 100644 index 94086997d9..0000000000 --- a/quest/m1/qos/final-lag-sample.md +++ /dev/null @@ -1,34 +0,0 @@ -# [XS] A closing subscription takes its last lag sample - -## Goal - -When an egress subscription ends, the bytes its track produced since the last -stats tick still land in the `lag` histogram at its final lag, instead of -being lost. A subscription that opens and closes between two ticks is sampled -at least once. - -## Plan - -Lag is sampled only on `Registry::report` ticks: `Counters::sample` in -`rs/moq-net/src/stats.rs` walks weak `FrontierInner` references and prunes -the dead ones without sampling them, so a subscription's last partial -interval disappears with it. - -- The subscription guard and each in-flight group `Delivery` hold a strong - reference, so the frontier dies only once the last of them does. Its drop is - the natural place for one final `sample(now)` into the same counters. -- No double counting: `sample` advances each source's `sampled` bytes under the - frontier lock, so a tick racing the drop leaves nothing for it to count again. -- Test: a subscription that opens and closes between ticks while its track - produces lands those bytes in the histogram; one that closes mid-interval - adds exactly the bytes since its last sample. Note the behaviour in - `doc/concept/stats.md`. - -Public API: none. Wire: none, only the histogram's values change. - -Overlaps the lag-splice quest on the line branch -([#4381](https://github.com/moq-dev/moq/pull/4381)), which also changes how -`FrontierInner::sample` holds `unsampled` weight and when a spliced segment's -source stops being sampled. Land them in sequence on the same sampler, not in -parallel, and make the final drop-time sample keep any weight lag-splice -defers. diff --git a/rs/moq-net/src/model/cache.rs b/rs/moq-net/src/model/cache.rs index e5ea169a7b..c16c1579ee 100644 --- a/rs/moq-net/src/model/cache.rs +++ b/rs/moq-net/src/model/cache.rs @@ -493,8 +493,9 @@ pub(crate) struct Track { sweep: OnceLock, // What the track's groups have produced, for the egress lag sampler. Kept on the - // account because it is already the one per-track handle every group holds. - production: crate::stats::Production, + // account because it is already the one per-track handle every group holds. Shared + // so a sampler can still read the last bytes after the track is gone. + production: Arc, } impl Track { @@ -521,7 +522,7 @@ impl Track { } /// What this track's groups have produced. - pub(crate) fn production(&self) -> &crate::stats::Production { + pub(crate) fn production(&self) -> &Arc { &self.production } diff --git a/rs/moq-net/src/stats.rs b/rs/moq-net/src/stats.rs index e10f87832b..10fcbbf977 100644 --- a/rs/moq-net/src/stats.rs +++ b/rs/moq-net/src/stats.rs @@ -1208,37 +1208,38 @@ impl Registry { let mut retired = shared.retired.lock(); { let mut entries = shared.entries.lock(); - for (path, entry) in entries.iter() { - let tiers = entry.tiers.lock().expect("stats tiers poisoned"); - for (tier, counters) in tiers.iter() { + entries.retain(|path, entry| { + // With only the map's Arc left, and the map locked, no guard can reach + // the entry again. + let orphan = Arc::strong_count(entry) == 1; + let mut tiers = entry.tiers.lock().expect("stats tiers poisoned"); + tiers.retain(|tier, counters| { + // Decided before the readout, so a last bump racing this report (such + // as a frontier's final lag sample) lands in this readout or the next, + // never after the one that retires the counters. + let done = orphan && Arc::strong_count(counters) == 1; + if done { + // Pairs with the last holder's releasing decrement, so the readout + // below sees every bump it made. + std::sync::atomic::fence(Ordering::Acquire); + } counters.publisher.sample(now); + let publisher = counters.publisher.snapshot(); + let subscriber = counters.subscriber.snapshot(); + if done { + let totals = retired.traffic.entry(tier.clone()).or_default(); + totals[Role::Publisher.idx()].add(publisher); + totals[Role::Subscriber.idx()].add(subscriber); + } report.traffic.push(TrafficEntry { path: path.clone(), tier: tier.clone(), - publisher: counters.publisher.snapshot(), - subscriber: counters.subscriber.snapshot(), + publisher, + subscriber, }); - } - } - // Prune entries no guard holds anymore: with only the map's Arc - // left, no future bump can land, so the entry is done. (A guard - // created after the readout above still holds the Arc and keeps - // its entry alive.) - entries.retain(|_, entry| { - if Arc::strong_count(entry) > 1 { - return true; - } - let mut tiers = entry.tiers.lock().expect("stats tiers poisoned"); - tiers.retain(|tier, counters| { - if Arc::strong_count(counters) > 1 { - return true; - } - let totals = retired.traffic.entry(tier.clone()).or_default(); - totals[Role::Publisher.idx()].add(counters.publisher.snapshot()); - totals[Role::Subscriber.idx()].add(counters.subscriber.snapshot()); - false + !done }); - !tiers.is_empty() + !orphan || !tiers.is_empty() }); } { @@ -1829,7 +1830,11 @@ struct FrontierState { /// A track feeding a [`Frontier`], with the produced bytes already sampled. struct Source { + // Only liveness: once dead, the track can never produce again. track: Weak, + // 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, sampled: u64, } @@ -1859,20 +1864,16 @@ impl Frontier { pub(crate) fn watch(&self, track: &Arc) { let Some(inner) = &self.0 else { return }; let mut state = inner.state.lock().expect("stats frontier poisoned"); - // Only a live source can match: a dead one's address may be reused. - if state - .sources - .iter() - .any(|s| s.track.upgrade().is_some_and(|live| Arc::ptr_eq(&live, track))) - { + let production = track.production(); + if state.sources.iter().any(|s| Arc::ptr_eq(&s.production, production)) { return; } - let production = track.production(); if state.acked.is_none() { state.acked = Production::load(&production.newest); } state.sources.push(Source { track: Arc::downgrade(track), + production: production.clone(), sampled: production.bytes.load(Ordering::Relaxed), }); } @@ -1882,18 +1883,14 @@ impl Frontier { 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| { - let Some(live) = s.track.upgrade() else { return false }; - if !Arc::ptr_eq(&live, track) { + if !Arc::ptr_eq(&s.production, production) { return true; } // What it produced before the cap reached this reader: keep it for the next sample. - unsampled += live - .production() - .bytes - .load(Ordering::Relaxed) - .saturating_sub(s.sampled); + unsampled += production.bytes.load(Ordering::Relaxed).saturating_sub(s.sampled); false }); state.unsampled += unsampled; @@ -1917,11 +1914,10 @@ impl FrontierInner { let mut newest = None; let mut first: Option = None; state.sources.retain_mut(|source| { - // Only this Weak outliving the track means it can never produce again. - let Some(track) = source.track.upgrade() else { - return false; - }; - let production = track.production(); + // 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); @@ -1930,7 +1926,7 @@ impl FrontierInner { if let Some(at) = Production::load(&production.first) { first = Some(first.map_or(at, |first| first.min(at))); } - true + live }); let newest = newest?; // Opened before the track produced anything: its first frame is where the @@ -1954,6 +1950,17 @@ impl FrontierInner { } } +impl Drop for FrontierInner { + /// The last strong reference (the subscription guard or an in-flight [`Delivery`]) + /// is gone, so no tick will find this frontier again: record the bytes produced + /// since the last sample now, at the lag they last had. + fn drop(&mut self) { + if let Some((lag, weight)) = self.sample(crate::model::clock::now()) { + self.counters.publisher.lag.record(lag, weight); + } + } +} + /// One group stream written toward a peer, reported to its subscription's /// [`Frontier`]: an acknowledged FIN advances the frontier, anything else that ends /// it counts the written media as dropped on drop. Empty (no-op) off the egress side. @@ -3062,6 +3069,99 @@ mod lag_tests { assert_eq!(grew(&last, &now), vec![(5, 3 * GROUP_BYTES), (6, 3 * GROUP_BYTES)]); } + /// The egress lag bytes recorded so far, without ticking the sampler. + fn recorded(stats: &Registry) -> u64 { + stats + .snapshot() + .traffic() + .into_iter() + .find(|(_, role, _)| matches!(role, Role::Publisher)) + .map_or(0, |(_, _, traffic)| traffic.lag.total()) + } + + #[test] + fn a_subscription_closed_between_ticks_is_sampled_once() { + let mut h = Harness::new(); + let sub = h.subscribe(); + let last = h.tick(); + h.produce(); + drop(sub); + + // The tick that follows finds no frontier, yet the group still lands, at the + // 957ms it was behind when the guard dropped. + let now = h.tick(); + assert_eq!(grew(&last, &now), vec![(4, GROUP_BYTES)]); + + // Sampled exactly once: the closing tick pruned the row, and later media adds + // nothing to the retired totals. + h.produce(); + h.stats.report(&mut Report::default()); + assert_eq!(recorded(&h.stats), GROUP_BYTES); + } + + #[test] + fn a_subscription_closed_mid_interval_adds_only_the_bytes_since_its_last_sample() { + let mut h = Harness::new(); + let sub = h.subscribe(); + let mut last = h.tick(); + h.produce(); + last = { + let now = h.tick(); + assert_eq!(grew(&last, &now), vec![(4, GROUP_BYTES)]); + now + }; + h.produce(); + h.produce(); + drop(sub); + + let now = h.tick(); + assert_eq!( + grew(&last, &now).iter().map(|(_, bytes)| bytes).sum::(), + 2 * GROUP_BYTES + ); + } + + #[test] + fn an_in_flight_delivery_defers_the_final_sample_to_its_own_drop() { + let mut h = Harness::new(); + let sub = h.subscribe(); + let delivery = Harness::write(&sub, &h.produce()); + drop(sub); + // The delivery still holds the frontier, so it has not closed yet. + assert_eq!(recorded(&h.stats), 0); + + drop(delivery); + assert_eq!(recorded(&h.stats), GROUP_BYTES); + } + + #[test] + fn the_last_subscriber_samples_a_track_its_own_drop_releases() { + let mut h = Harness::new(); + let sub = h.subscribe(); + h.produce(); + // The producer is gone, so the subscriber holds the last reference to the track. + drop(h.track); + drop(h.broadcast); + + drop(sub); + assert_eq!(recorded(&h.stats), GROUP_BYTES); + } + + #[test] + fn an_in_flight_delivery_samples_a_track_that_already_closed() { + let mut h = Harness::new(); + let sub = h.subscribe(); + let delivery = Harness::write(&sub, &h.produce()); + drop(h.track); + drop(h.broadcast); + drop(sub); + assert_eq!(recorded(&h.stats), 0); + + // The track died with the subscriber; the delivery still owes its bytes. + drop(delivery); + assert_eq!(recorded(&h.stats), GROUP_BYTES); + } + #[test] fn a_closed_subscription_leaves_the_sampler() { let mut h = Harness::new();