diff --git a/quest/m1/release-perf/README.md b/quest/m1/release-perf/README.md index 1b55953452a..6b92b4e8cc9 100644 --- a/quest/m1/release-perf/README.md +++ b/quest/m1/release-perf/README.md @@ -1,4 +1,4 @@ -# Main is no slower than release +# [M] Main is no slower than release ## Goal @@ -216,20 +216,17 @@ Decided 2026-10-25 (maintainer), in [moq-dev/moq#6784](https://github.com/moq-dev/moq/pull/6784): - The sparse read slope from - [dynamic range requests](https://github.com/moq-dev/moq/pull/6171) is the - child below, and it blocks the cut. + [dynamic range requests](https://github.com/moq-dev/moq/pull/6171) blocks + the cut; [moq-dev/moq#6921](https://github.com/moq-dev/moq/pull/6921) + flattened it. - The TS schedule's growth is left to the [performance audit](/quest/m2/audit-perf.md); it does not block the cut. -- Once the child lands, re-run the relay fanout pairing on a quiet host, - recording `/proc/loadavg`, with +- Now that the sparse fix has landed, re-run the relay fanout pairing on a + quiet host, recording `/proc/loadavg`, with `BENCH_ARTIFACTS= just bench origin/release` (it also re-checks the sparse rows). Fix it if the relay CPU still rises with consistent sign; otherwise this epic is done. -## Required - -- [Sparse reads judge misses incrementally](/quest/m1/release-perf/sparse-misses.md) - `track_sparse_read` stays flat across gaps, as before range requests - ## Related - [Performance audit](/quest/m2/audit-perf.md) - compares nightly to nightly; this epic compares main to release diff --git a/quest/m1/release-perf/sparse-misses.md b/quest/m1/release-perf/sparse-misses.md deleted file mode 100644 index 7ed3dc13670..00000000000 --- a/quest/m1/release-perf/sparse-misses.md +++ /dev/null @@ -1,33 +0,0 @@ -# [M] Sparse reads judge misses incrementally - -## Goal - -A reader of a sparse track pays O(log gaps) per group handed out, as it did -before [dynamic range requests](https://github.com/moq-dev/moq/pull/6171). -`track_sparse_read` and the setup of `track_gap_flush` stay flat across their -gap sweeps. - -## Plan - -Found by the 2026-10-25 main against release run. Every hand-out goes through -`Cursor::probe` (`model/track.rs`), and `miss::Misses::judge` -(`model/track/miss.rs`) rebuilds its clipped span from the whole missable -window and re-judges every deferred gap, so each read walks every gap below -it. Instructions per write and read pair of `track_sparse_read/timed`: 152k -at 16 gaps, 2.57M at 256, and 51.4M at 4096, which is 1079x release in wall -time (2.75 ms against 2.5 us). The untimed rows match. `track_gap_flush` -builds its gaps through the same path, so its untimed setup is quadratic -(50.8G instructions at 4096 gaps). - -Judge only what changed since the last probe: the groups the read passed, -gaps the ledger resolved, and a window or budget change. A probe with -nothing new must not touch the deferred set. Keep the fills and drops -that range requests serve; the sweep, not a release ratio, is the acceptance signal. -[Drop set](/quest/m1/drop-set.md) reshapes the same state, so whichever lands -second rebases onto the other. - -Public API: none. Wire: none. - -## Related - -- [Drop set](/quest/m1/drop-set.md) - per-cursor miss judgement over the same ledger diff --git a/rs/moq-net/benches/track.rs b/rs/moq-net/benches/track.rs index 40d84446c0f..7bec7a97103 100644 --- a/rs/moq-net/benches/track.rs +++ b/rs/moq-net/benches/track.rs @@ -26,6 +26,8 @@ //! `track_sparse_read` hands out groups from a feed with every other sequence missing, //! holding a fixed number of gaps the reader still remembers: each hand-out must cost the //! same however many gaps it holds, so a walk over them shows up as a slope. +//! `track_sparse_poll` repeats a parked read without changing the feed, so revisiting +//! deferred gaps on an otherwise unchanged poll shows up independently of delivery. //! //! `track_sparse_expire` opens one new gap while one old gap expires, so a full expiry //! rescan on each arrival shows up as a slope over the gaps still within budget. @@ -1292,6 +1294,26 @@ fn bench_sparse_read(c: &mut Criterion) { group.finish(); } +fn bench_sparse_poll(c: &mut Criterion) { + let mut group = c.benchmark_group("track_sparse_poll"); + for (name, info) in [("untimed", track::Info::default()), ("timed", timed())] { + for gaps in [16u64, 256, 4096] { + group.bench_function(BenchmarkId::new(name, gaps), |b| { + b.iter_custom(|iterations| { + let mut sparse = Sparse::new(info.clone(), gaps); + let waiter = kio::Waiter::noop(); + let start = WallInstant::now(); + for _ in 0..iterations { + assert!(black_box(sparse.sub.poll_recv_group(&waiter)).is_pending()); + } + start.elapsed() + }); + }); + } + } + group.finish(); +} + /// A sparse feed that opens one gap per millisecond, keeping `gaps` within its budget. struct Expiring { sparse: Sparse, @@ -1940,6 +1962,7 @@ criterion_group!( bench_datagram_pressure, bench_parked_read, bench_sparse_read, + bench_sparse_poll, bench_sparse_expire, bench_gap_flush, bench_wall, diff --git a/rs/moq-net/src/ledger.rs b/rs/moq-net/src/ledger.rs index 8473dbc52a1..7ad416a5410 100644 --- a/rs/moq-net/src/ledger.rs +++ b/rs/moq-net/src/ledger.rs @@ -47,6 +47,44 @@ pub(crate) struct Ledger { // Index both: one due gap must not make every arrival scan all the younger ones. ages: BTreeSet<(Instant, u64)>, grace: Duration, + changes: Changes, +} + +/// Revisions of the current runs, indexed in both orders. Coalescing or removing a run +/// removes its revision too, so an idle reader never retains a history of mutations. +#[derive(Debug, Default)] +pub(crate) struct Changes { + runs: BTreeMap, + changed: BTreeMap, + revision: u64, +} + +impl Changes { + pub fn revision(&self) -> u64 { + self.revision + } + + pub fn remove(&mut self, start: u64) { + if let Some((_, revision)) = self.runs.remove(&start) { + self.changed.remove(&revision); + } + } + + pub fn insert(&mut self, run: Range) { + self.remove(run.start); + self.revision = self.revision.checked_add(1).expect("run revisions exhausted"); + self.runs.insert(run.start, (run.end, self.revision)); + self.changed.insert(self.revision, run.start); + } + + pub fn since(&self, revision: u64) -> impl Iterator> + '_ { + self.changed + .range((std::ops::Bound::Excluded(revision), std::ops::Bound::Unbounded)) + .map(|(_, start)| { + let (end, _) = self.runs.get(start).expect("a changed run stays indexed"); + *start..*end + }) + } } impl Default for Ledger { @@ -62,6 +100,7 @@ impl Ledger { runs: BTreeMap::new(), ages: BTreeSet::new(), grace, + changes: Changes::default(), } } @@ -71,6 +110,7 @@ impl Ledger { } fn remove(&mut self, start: u64) -> Run { + self.changes.remove(start); let run = self.runs.remove(&start).expect("an indexed run"); self.ages.remove(&(run.since, start)); run @@ -112,6 +152,20 @@ impl Ledger { self.runs.insert(first, Run { end, since }); } } + self.changes.insert(first..end); + } + + /// Account a live arrival, refreshing a cached incarnation even if its sequence was + /// accounted before. Its gap ages stay unchanged, but readers must revisit a refusal. + pub fn arrive(&mut self, sequence: u64, now: Instant) { + let revision = self.changes.revision(); + self.account(sequence..sequence.saturating_add(1), now); + if self.changes.revision() == revision + && let Some((&start, run)) = self.runs.range(..=sequence).next_back() + && sequence < run.end + { + self.changes.insert(start..run.end); + } } /// Fold every gap older than the grace into the runs around it. @@ -126,12 +180,9 @@ impl Ledger { break; } let run = self.remove(start); - self.runs - .range_mut(..start) - .next_back() - .expect("not the first run") - .1 - .end = run.end; + let (&lower, previous) = self.runs.range_mut(..start).next_back().expect("not the first run"); + previous.end = run.end; + self.changes.insert(lower..run.end); } } @@ -174,6 +225,10 @@ impl Ledger { .map(move |(&start, run)| start.max(groups.start)..run.end.min(groups.end)) } + pub fn changes(&self) -> &Changes { + &self.changes + } + /// Whether every group in `groups` is accounted for. pub fn covers(&self, groups: Range) -> bool { groups.is_empty() || self.run_end(groups.start).is_some_and(|end| groups.end <= end) @@ -199,6 +254,7 @@ impl Ledger { self.ages.remove(&(run.since, start)); run.since = now; self.ages.insert((now, start)); + self.changes.insert(start..run.end); } below = run.end; } @@ -254,6 +310,27 @@ impl Ledger { mod tests { use super::*; + #[test] + fn changes_keep_only_current_runs_across_coalescing_and_expiry() { + let now = Instant::ZERO; + let mut ledger = Ledger::new(GRACE); + ledger.account(0..1, now); + ledger.account(4..5, now); + ledger.account(8..9, now); + let revision = ledger.changes().revision(); + ledger.account(2..3, now); + assert_eq!(ledger.changes().since(revision).collect::>(), vec![2..3]); + let revision = ledger.changes().revision(); + ledger.account(1..4, now); + assert_eq!(ledger.changes().since(revision).collect::>(), vec![0..5]); + ledger.expire(now + GRACE); + assert_eq!(ledger.changes().since(0).collect::>(), vec![0..9]); + assert_eq!(ledger.changes.changed.len(), ledger.runs.len()); + let revision = ledger.changes().revision(); + ledger.account(2..7, now + GRACE); + assert_eq!(ledger.changes().revision(), revision, "already accounted is unchanged"); + } + /// Account `groups` and fold what aged out, as each arrival does. fn arrive(ledger: &mut Ledger, groups: Range, now: Instant) { ledger.account(groups, now); diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index e2571e368e9..18587e57a68 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -446,6 +446,7 @@ pub(crate) struct TrackState { drops: RangeMap, // Bumped on every insert into `drops`, so a cursor rescans it only after a change. drops_revision: u64, + drops_changes: ledger::Changes, // Whether the cache reflects the live feed: readers get nothing from it while it does // not, since how stale it is cannot be told; fetches still do. A track is live from @@ -886,7 +887,26 @@ impl TrackState { return; } self.cache.wakes().arrived(groups.clone()); - self.drops.insert(groups, err); + self.change_drops(groups, Some(err)); + } + + /// Keep the change index aligned with the range map, including the neighbours an + /// insertion coalesces and the two pieces a removal can leave. + fn change_drops(&mut self, groups: ops::Range, err: Option) { + let mut affected = groups.start.saturating_sub(1)..groups.end.saturating_add(1); + let old: Vec<_> = self.drops.overlapping(&affected).map(|(run, _)| run.clone()).collect(); + for run in old { + self.drops_changes.remove(run.start); + affected.start = affected.start.min(run.start); + affected.end = affected.end.max(run.end); + } + match err { + Some(err) => self.drops.insert(groups, err), + None => self.drops.remove(groups), + } + for (run, _) in self.drops.overlapping(&affected) { + self.drops_changes.insert(run.clone()); + } self.drops_revision += 1; } @@ -1775,7 +1795,7 @@ impl TrackState { /// Record that the live feed delivered `sequence`, then fold the gaps that aged out. fn record_arrival(&mut self, sequence: u64, now: crate::time::Instant) { - self.arrivals.account(sequence..sequence.saturating_add(1), now); + self.arrivals.arrive(sequence, now); // A feed still delivering keeps the gap above its last arrival open. if self.tail.is_some() { self.tail = Some(now); @@ -1797,8 +1817,7 @@ impl TrackState { /// How long a missing live group is waited for, from the max delay the track's /// subscribers ask for. fn arrival_grace(&self) -> Duration { - let max_delay = snapshot_subscription(&self.subscriptions, self.max_age_bound()) - .map_or(Duration::ZERO, |subscription| subscription.budget()); + let max_delay = clamp_max_delay(self.subscriptions.lock().max_delay(), self.max_age_bound()); ledger::grace(max_delay) } @@ -1908,8 +1927,7 @@ impl TrackState { && self.drops.contains_key(&sequence) && let Some(end) = sequence.checked_add(1) { - self.drops.remove(sequence..end); - self.drops_revision += 1; + self.change_drops(sequence..end, None); } let now = self.cache.pool().now_wall(); if visible { @@ -6384,7 +6402,17 @@ impl Cursor { misses.judge( &state, (from, revision, cap), - || below(window.missable(from), cap), + |span| { + let end = cap.map_or(*span.end(), |cap| (*span.end()).min(cap.saturating_sub(1))); + if cap == Some(0) || *span.start() > end { + return RangeInclusiveSet::new(); + } + window + .missable(*span.start()) + .take_while(|run| *run.start() <= end) + .map(|run| *run.start()..=(*run.end()).min(end)) + .collect() + }, drift, &mut given_up, waiter, @@ -8075,6 +8103,35 @@ mod test { Producer::new(Arc::new(broadcast::Info::default()), name, info) } + #[test] + fn drop_changes_follow_split_coalesced_and_recreated_runs() { + let producer = track_producer("test", Info::default()); + let indexed = || { + let state = producer.state.read(); + let mut changed: Vec<_> = state.drops_changes.since(0).collect(); + changed.sort_by_key(|run| run.start); + let current: Vec<_> = state.drops.iter().map(|(run, _)| run.clone()).collect(); + assert_eq!(changed, current, "one change entry per current drop run"); + }; + producer.drop_groups(10..20, StreamError::Old); + producer.drop_groups(20..30, StreamError::Old); + indexed(); + producer.drop_groups(15..25, StreamError::Cancel); + indexed(); + producer + .create_group(group::Info { sequence: 17 }) + .unwrap() + .finish() + .unwrap(); + indexed(); + producer.drop_groups(14..26, StreamError::Old); + indexed(); + assert_eq!( + producer.state.read().drops_changes.since(0).collect::>(), + vec![10..30] + ); + } + /// Let `duration` pass on the cache pool `producer`'s groups charge into. fn elapse(producer: &Producer, duration: Duration) { producer.broadcast.pool.step(duration); diff --git a/rs/moq-net/src/model/track/miss.rs b/rs/moq-net/src/model/track/miss.rs index ebc7083c32e..470678d5852 100644 --- a/rs/moq-net/src/model/track/miss.rs +++ b/rs/moq-net/src/model/track/miss.rs @@ -15,7 +15,7 @@ //! then the live feed may still deliver it. use std::{ - collections::BTreeSet, + collections::{BTreeMap, BTreeSet}, ops::{self, RangeInclusive}, task::Poll, time::Duration, @@ -37,10 +37,15 @@ type JudgedAt = (u64, u64, Option, Option, Option, Option, u pub(super) struct Misses { /// What the last probe judged against. A probe that sees the same skips judging. at: Option, + ledger_revision: u64, + drops_revision: u64, + budget: Option, + grace: Option, + ended: bool, /// The sequences judged already, within what the reader reads up to the newest group. judged: RangeSet, - /// Holes the live feed may still deliver, judged again on every probe. - deferred: RangeSet, + /// Holes the live feed may still deliver, revisited on a ledger change or deadline. + deferred: Deferred, /// Asked under the feed's start and refused, never asked again: given up once the feed /// gives up on them, unless they arrive first. refused: RangeSet, @@ -59,6 +64,105 @@ pub(super) struct Misses { /// A run of misses the reader gives up, for the drop cursor to name. pub(super) type GivenUp = Vec>; +/// The current deferred runs, indexed by sequence and by their next judgement deadline. +/// Splits remove the old age entry before publishing either piece; no stale timers remain. +#[derive(Default)] +struct Deferred { + runs: BTreeMap)>, + ages: BTreeSet<(crate::time::Instant, u64)>, +} + +impl Deferred { + fn put(&mut self, run: ops::Range, at: Option) { + if run.is_empty() { + return; + } + if let Some(at) = at { + self.ages.insert((at, run.start)); + } + self.runs.insert(run.start, (run.end, at)); + } + + fn take(&mut self, start: u64) -> (u64, Option) { + let (end, at) = self.runs.remove(&start).expect("an indexed deferred run"); + if let Some(at) = at { + self.ages.remove(&(at, start)); + } + (end, at) + } + + fn remove(&mut self, run: ops::Range) { + if run.is_empty() { + return; + } + let first = self + .runs + .range(..=run.start) + .next_back() + .filter(|(_, (end, _))| *end > run.start) + .map_or(run.start, |(start, _)| *start); + let starts: Vec<_> = self.runs.range(first..run.end).map(|(start, _)| *start).collect(); + for start in starts { + let (end, at) = self.take(start); + self.put(start..end.min(run.start), at); + self.put(start.max(run.end)..end, at); + } + } + + fn insert(&mut self, mut run: ops::Range, at: Option) { + if run.is_empty() { + return; + } + self.remove(run.clone()); + if let Some((&start, &(end, previous))) = self.runs.range(..run.start).next_back() + && end == run.start + && previous == at + { + self.take(start); + run.start = start; + } + if let Some(&(end, next)) = self.runs.get(&run.end) + && next == at + { + self.take(run.end); + run.end = end; + } + self.put(run, at); + } + + fn overlapping(&self, run: ops::Range) -> impl Iterator> + '_ { + let first = self + .runs + .range(..=run.start) + .next_back() + .filter(|(_, (end, _))| *end > run.start) + .map_or(run.start, |(start, _)| *start); + self.runs + .range(first..run.end) + .map(move |(&start, &(end, _))| start.max(run.start)..end.min(run.end)) + } + + fn due(&mut self, now: crate::time::Instant) -> RangeSet { + let mut due = RangeSet::new(); + while let Some(&(at, start)) = self.ages.first() { + if at > now { + break; + } + let (end, _) = self.take(start); + due.insert(start..end); + } + due + } + + fn deadline(&self) -> Option { + self.ages.first().map(|(at, _)| *at) + } + fn clear(&mut self) { + self.runs.clear(); + self.ages.clear(); + } +} + impl Misses { /// The lowest sequence still to come from a fill: asked, or answered and not handed out. pub fn lowest(&self) -> Option { @@ -250,7 +354,7 @@ impl Misses { &mut self, state: &TrackState, key: (u64, u64, Option), - target: impl FnOnce() -> RangeInclusiveSet, + mut target: impl FnMut(RangeInclusive) -> RangeInclusiveSet, drift: &Drift, given_up: &mut GivenUp, waiter: &kio::Waiter, @@ -273,57 +377,120 @@ impl Misses { state.start_sequence, state.drops_revision, ); - if self.at == Some(at) && self.deferred.is_empty() { + let now = state.cache.pool().now_wall(); + let grace = state.arrival_grace(); + let ended = state.abort.is_some() || state.is_complete(); + let terms = self.at.is_none_or(|previous| { + previous.1 != at.1 || previous.2 != at.2 || previous.4 != at.4 || previous.5 != at.5 || at.0 < previous.0 + }) || self.budget != Some(drift.budget) + || self.grace != Some(grace) + || self.ended != ended; + let ledger = state.arrivals.changes(); + let drops = &state.drops_changes; + if !terms + && self.at == Some(at) + && self.ledger_revision == ledger.revision() + && self.drops_revision == drops.revision() + && self.deferred.deadline().is_none_or(|at| at > now) + { + if let Some(at) = self.deferred.deadline() { + state.cache.watch_wall(at, waiter); + } return asks; } - self.at = Some(at); - - let low = state.live_floor.unwrap_or(0); - // A run's exclusive end must exist, so the last sequence is never judged. + let low = state.live_floor.unwrap_or(0).max(key.0); let top = newest.min(u64::MAX - 1); - let mut span = RangeSet::new(); - if low <= top { - for run in target().overlapping(&(low..=top)) { - span.insert((*run.start()).max(low)..(*run.end()).min(top) + 1); + let mut candidates = RangeSet::new(); + if terms { + let mut span = RangeSet::new(); + if low <= top { + for run in target(low..=top).iter() { + span.insert(*run.start()..*run.end() + 1); + } + } + let clip = |set: &RangeSet| -> RangeSet { + set.iter() + .flat_map(|run| { + span.overlapping(run) + .map(move |s| s.start.max(run.start)..s.end.min(run.end)) + }) + .collect() + }; + self.judged = clip(&self.judged); + self.refused = clip(&self.refused); + self.given_up = clip(&self.given_up); + for (&start, &(end, _)) in &self.deferred.runs { + for run in span.overlapping(&(start..end)) { + candidates.insert(start.max(run.start)..end.min(run.end)); + } + } + self.deferred.clear(); + for run in span.iter() { + for fresh in self.judged.gaps(run) { + candidates.insert(fresh); + } + } + } else { + // A moving read floor removes only its retired prefix. New production extends the + // judged span; neither path visits older deferred gaps. + if low > 0 { + for set in [&mut self.judged, &mut self.refused, &mut self.given_up] { + set.remove(0..low); + } + self.deferred.remove(0..low); + } + let first = self + .at + .and_then(|previous| previous.3) + .map_or(low, |previous| previous.saturating_add(1).max(low)); + if first <= top { + for run in target(first..=top).iter() { + let run = *run.start()..*run.end() + 1; + for fresh in self.judged.gaps(&run) { + candidates.insert(fresh); + } + } + } + for run in self.deferred.due(now).iter() { + candidates.insert(run.clone()); } } - // What left the span is forgotten, so it is judged afresh if it comes back, and a - // verdict behind the reader's floor stops costing it memory. - let clip = |set: &RangeSet| -> RangeSet { - set.iter() + // Current ledger runs are indexed by their last mutation. Coalescing retires the + // old entries, so a reader that was idle needs no shared mutation history. + for changed in ledger + .since(self.ledger_revision) + .chain(drops.since(self.drops_revision)) + { + for run in self.deferred.overlapping(changed.clone()) { + candidates.insert(run); + } + let arrived: Vec<_> = self + .refused + .overlapping(&changed) .flat_map(|run| { - span.overlapping(run) - .map(move |s| s.start.max(run.start)..s.end.min(run.end)) + state + .lookup + .range(run.start.max(changed.start)..run.end.min(changed.end)) + // An aborted slot is still a hole, so its refusal stands. + .filter(|(_, slot)| !slot.is_aborted()) + .map(|(&sequence, _)| sequence) }) - .collect() - }; - let judged = clip(&self.judged); - self.refused = clip(&self.refused); - self.given_up = clip(&self.given_up); - let mut candidates: RangeSet = span.iter().flat_map(|run| judged.gaps(run)).collect(); - for run in std::mem::take(&mut self.deferred).iter() { - for s in span.overlapping(run) { - candidates.insert(s.start.max(run.start)..s.end.min(run.end)); + .collect(); + for sequence in arrived { + self.refused.remove(sequence..sequence + 1); } } - self.judged = judged; + self.ledger_revision = ledger.revision(); + self.drops_revision = drops.revision(); + self.budget = Some(drift.budget); + self.grace = Some(grace); + self.ended = ended; + self.at = Some(at); for run in candidates.iter() { + self.deferred.remove(run.clone()); self.judged.insert(run.clone()); } - - // A refused group that arrived live is the feed's, and no longer refused. - let arrived: Vec = self - .refused - .iter() - .flat_map(|run| state.lookup.range(run.clone()).map(|(sequence, _)| *sequence)) - .collect(); - for sequence in arrived { - self.refused.remove(sequence..sequence + 1); - } - - let now = state.cache.pool().now_wall(); let start = state.start_sequence.unwrap_or(0); - let mut wake: Option = None; for run in candidates.iter() { for hole in holes(state, run.clone()) { for hole in self.unclaimed(state, hole) { @@ -344,17 +511,17 @@ impl Misses { if !early.is_empty() { asks.insert(early); } - self.deferred.insert(run); - wake = match (wake, folds) { - (Some(a), Some(b)) => Some(a.min(b)), - (a, b) => a.or(b), - }; + self.deferred.insert(run, folds); } } } } - for run in self.refused.iter() { - asks.remove(run.clone()); + let refused: Vec<_> = asks + .iter() + .flat_map(|ask| self.refused.overlapping(ask).cloned()) + .collect(); + for run in refused { + asks.remove(run); } for run in asks.iter() { self.deferred.remove(run.clone()); @@ -364,7 +531,7 @@ impl Misses { asks.remove(datagram.sequence..datagram.sequence.saturating_add(1)); } // A gap the feed still waits for becomes a miss when it folds. - if let Some(at) = wake { + if let Some(at) = self.deferred.deadline() { state.cache.watch_wall(at, waiter); } @@ -494,7 +661,8 @@ impl Misses { } for run in left.iter() { self.refused.insert(run.clone()); - self.deferred.insert(run.clone()); + self.deferred + .insert(run.clone(), state.arrivals.folds_at(run.start, state.arrival_grace())); } } } @@ -594,3 +762,280 @@ fn looser(ours: Option, asked: Option) -> bool { (None, _) => false, } } + +#[cfg(test)] +mod tests { + use super::super::{Servable, Subscription}; + use super::*; + + #[moq_clock::test] + async fn a_recreated_accounted_group_clears_its_earlier_fill_refusal() { + let broadcast = crate::broadcast::Info::default().produce(); + let mut track = broadcast + .create_track("recreated", crate::track::Info::default()) + .unwrap(); + let budget = Duration::from_secs(5); + let _reader = track.subscribe(Subscription::default().with_max_delay(budget)); + for sequence in [0, 4] { + track + .create_group(crate::group::Info { sequence }) + .unwrap() + .finish() + .unwrap(); + } + track.start_at(3).unwrap(); + let mut misses = Misses::default(); + let mut given_up = Vec::new(); + let waiter = kio::Waiter::noop(); + let state = track.state.read(); + let drift = Drift { + budget, + bounded: false, + edge: state.drift_edge(&Servable::ALL), + }; + let asks = misses.judge( + &state, + (0, 0, None), + |span| [span].into_iter().collect(), + &drift, + &mut given_up, + &waiter, + ); + let fills = state.fills.clone(); + drop(state); + misses.ask( + &track.state.consume(), + &fills, + &asks, + &Asking { + priority: 0, + max_delay: Some(budget), + order: Order::Ascending, + }, + &mut given_up, + ); + assert!( + misses.refused.contains(&2), + "a missing fill is refused while the live gap stays open" + ); + + // An arrival and cut happen while the cursor is idle. The drop suppresses asking + // again, and reclaiming the cut removes the cached incarnation before the probe. + track + .create_group(crate::group::Info { sequence: 2 }) + .unwrap() + .abort(Error::Cancel) + .unwrap(); + track.drop_groups(2..3, crate::StreamError::Old); + track.modify().unwrap().uncache(2); + let state = track.state.read(); + assert!(state.drops.contains_key(&2)); + assert!(!state.lookup.contains_key(&2)); + misses.judge( + &state, + (0, 0, None), + |_| panic!("the edge has not moved"), + &drift, + &mut given_up, + &waiter, + ); + assert!(misses.refused.contains(&2)); + let revision = state.arrivals.changes().revision(); + drop(state); + track + .create_group(crate::group::Info { sequence: 2 }) + .unwrap() + .finish() + .unwrap(); + let state = track.state.read(); + assert!(state.drops.is_empty(), "recreation removed the last drop run"); + assert!(state.arrivals.covers(2..3)); + assert!( + state.arrivals.changes().revision() > revision, + "the real incarnation invalidates its current run" + ); + misses.judge( + &state, + (0, 0, None), + |_| panic!("recreation extends no span"), + &drift, + &mut given_up, + &waiter, + ); + assert!( + !misses.refused.contains(&2), + "the live cached incarnation is no longer refused" + ); + assert!(given_up.is_empty(), "a live recreation is never given up"); + } + + #[moq_clock::test] + async fn an_aborted_live_group_keeps_its_earlier_fill_refusal() { + let broadcast = crate::broadcast::Info::default().produce(); + let mut track = broadcast + .create_track("aborted", crate::track::Info::default()) + .unwrap(); + let budget = Duration::from_secs(5); + let _reader = track.subscribe(Subscription::default().with_max_delay(budget)); + for sequence in [0, 4] { + track + .create_group(crate::group::Info { sequence }) + .unwrap() + .finish() + .unwrap(); + } + track.start_at(3).unwrap(); + let mut misses = Misses::default(); + let mut given_up = Vec::new(); + let waiter = kio::Waiter::noop(); + let state = track.state.read(); + let drift = Drift { + budget, + bounded: false, + edge: state.drift_edge(&Servable::ALL), + }; + let asks = misses.judge( + &state, + (0, 0, None), + |span| [span].into_iter().collect(), + &drift, + &mut given_up, + &waiter, + ); + let fills = state.fills.clone(); + drop(state); + misses.ask( + &track.state.consume(), + &fills, + &asks, + &Asking { + priority: 0, + max_delay: Some(budget), + order: Order::Ascending, + }, + &mut given_up, + ); + assert!(misses.refused.contains(&2)); + + // The live feed accounts for group 2 but aborts it without a cut, so its slot is + // still a hole: the refusal stands and the hole is given up, not asked again. + track + .create_group(crate::group::Info { sequence: 2 }) + .unwrap() + .abort(Error::Cancel) + .unwrap(); + let state = track.state.read(); + assert!(state.lookup.get(&2).is_some_and(|slot| slot.is_aborted())); + let asks = misses.judge( + &state, + (0, 0, None), + |_| panic!("the edge has not moved"), + &drift, + &mut given_up, + &waiter, + ); + assert!(!asks.contains(&2), "a refused hole is not asked again"); + assert_eq!(given_up, vec![2..3]); + } + #[moq_clock::test] + async fn unchanged_probes_and_late_arrivals_keep_other_gap_deadlines() { + let broadcast = crate::broadcast::Info::default().produce(); + let track = broadcast.create_track("sparse", crate::track::Info::default()).unwrap(); + let budget = Duration::from_secs(5); + let _reader = track.subscribe(Subscription::default().with_max_delay(budget)); + for sequence in [0, 2, 4] { + track + .create_group(crate::group::Info { sequence }) + .unwrap() + .finish() + .unwrap(); + } + let mut misses = Misses::default(); + let mut given_up = Vec::new(); + let waiter = kio::Waiter::noop(); + let state = track.state.read(); + let drift = Drift { + budget, + bounded: false, + edge: state.drift_edge(&Servable::ALL), + }; + assert!( + misses + .judge( + &state, + (0, 0, None), + |span| [span].into_iter().collect(), + &drift, + &mut given_up, + &waiter + ) + .is_empty() + ); + assert_eq!(misses.deferred.overlapping(0..5).collect::>(), vec![1..2, 3..4]); + let deadline = misses.deferred.deadline(); + assert!( + misses + .judge( + &state, + (0, 0, None), + |_| panic!("an unchanged probe rebuilds no span"), + &drift, + &mut given_up, + &waiter + ) + .is_empty() + ); + drop(state); + track + .create_group(crate::group::Info { sequence: 1 }) + .unwrap() + .finish() + .unwrap(); + let state = track.state.read(); + assert!( + misses + .judge( + &state, + (0, 0, None), + |_| panic!("a late arrival extends no span"), + &drift, + &mut given_up, + &waiter + ) + .is_empty() + ); + assert_eq!(misses.deferred.overlapping(0..5).collect::>(), vec![3..4]); + assert_eq!(misses.deferred.deadline(), deadline); + drop(state); + broadcast.info().pool.step(budget); + let state = track.state.read(); + let asks = misses.judge( + &state, + (0, 0, None), + |_| panic!("a due gap extends no span"), + &drift, + &mut given_up, + &waiter, + ); + assert!(asks.is_empty(), "the successor has aged past this reader's budget too"); + assert_eq!(given_up, vec![3..4]); + assert!(misses.deferred.runs.is_empty()); + } + + #[test] + fn deferred_splits_and_replacements_retire_the_old_age_entries() { + let mut deferred = Deferred::default(); + let early = crate::time::Instant::ZERO + Duration::from_secs(1); + let late = early + Duration::from_secs(1); + deferred.insert(0..u64::MAX, Some(early)); + deferred.insert(10..20, Some(late)); + deferred.remove(5..25); + assert_eq!(deferred.ages.len(), 2); + assert_eq!( + deferred.due(early).iter().cloned().collect::>(), + vec![0..5, 25..u64::MAX] + ); + assert!(deferred.due(late).is_empty()); + assert!(deferred.runs.is_empty()); + } +}