diff --git a/quest/m1/moxygen/README.md b/quest/m1/moxygen/README.md index 8fae11bd13..cf8bcd1826 100644 --- a/quest/m1/moxygen/README.md +++ b/quest/m1/moxygen/README.md @@ -1,4 +1,4 @@ -# Moxygen compatibility +# [S] Moxygen compatibility ## Goal @@ -42,10 +42,6 @@ Decided in the review of Docs stay inline in the change that makes them stale. No new guide. -## Required - -- [Sparse FETCH ranges](/quest/m1/moxygen/fetch-span.md) - a FETCH costs the groups it returns, not the span of its range - ## Related - [JavaScript FETCH](/quest/m1/js-fetch.md) - the browser publisher answers a FETCH this relay forwards diff --git a/quest/m1/moxygen/fetch-span.md b/quest/m1/moxygen/fetch-span.md deleted file mode 100644 index c326968452..0000000000 --- a/quest/m1/moxygen/fetch-span.md +++ /dev/null @@ -1,27 +0,0 @@ -# [M] Sparse FETCH ranges - -## Goal - -A FETCH's cost follows the groups it can return, not the span of its range. -A track whose sequences are sparse, or whose history is long gone, answers a -wide range without spinning a worker or probing upstream once per missing -sequence. - -## Plan - -The standalone FETCH walk asks `track::Consumer::fetch_group` for every -sequence from start to the newest group, stepping over each miss. With no -fetch handler, every miss resolves at once, so a range over millions of -missing sequences runs without yielding. With a handler, a relay sends one -upstream FETCH per missing sequence. Either way a single request from a peer -buys work proportional to the newest sequence number. - -Seek the next group the track can serve instead of stepping by one. Where a -relay cannot know which groups its upstream still holds, decide what bound -or refusal is honest rather than probing blindly. Benchmark range span and -present-group count as separate axes. - -## Related - -- [Moxygen compatibility](/quest/m1/moxygen/README.md) - the line whose FETCH walk this bounds -- [Fetch without SUBSCRIBE](/quest/m1/ietf-fetch-only.md) - also changes how a relay reaches upstream for fetches diff --git a/rs/moq-net/Cargo.toml b/rs/moq-net/Cargo.toml index 967afb259d..11dfd50614 100644 --- a/rs/moq-net/Cargo.toml +++ b/rs/moq-net/Cargo.toml @@ -16,8 +16,8 @@ categories = ["multimedia", "network-programming", "web-programming"] ignored = ["getrandom"] [features] -# Exposes the wire codecs to the `fuzz/` harness and the announce bench through the -# hidden `fuzz` module. Off by default and hidden from the docs. +# Exposes the wire codecs to the `fuzz/` harness, and private paths to the announce and +# fetch benches, through the hidden `fuzz` module. Off by default and hidden from the docs. fuzz = [] [dependencies] @@ -75,6 +75,12 @@ name = "announce" harness = false required-features = ["fuzz"] +# Reaches the private FETCH walk through the hidden `fuzz` module. +[[bench]] +name = "fetch" +harness = false +required-features = ["fuzz"] + [[bench]] name = "group" harness = false diff --git a/rs/moq-net/benches/fetch.rs b/rs/moq-net/benches/fetch.rs new file mode 100644 index 0000000000..ae1e02e3af --- /dev/null +++ b/rs/moq-net/benches/fetch.rs @@ -0,0 +1,94 @@ +//! Standalone FETCH walk over a sparse track: what one range request costs a +//! publisher answering it from its own cache, with nothing upstream to ask. +//! +//! `fetch/span` holds the cached groups fixed and widens the range they are spread +//! across; `fetch/present` holds the range fixed and adds groups. The walk should +//! cost the groups it returns, so `span` stays flat and `present` grows linearly. +//! Throughput counts the groups returned. +//! +//! Run with `cargo bench -p moq-net --features fuzz --bench fetch`. + +use std::time::Duration; + +use bytes::Bytes; +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use moq_net::{Timestamp, broadcast, cache, group, track}; + +/// Range widths, from dense to sparse enough that stepping through it never ends. +const SPANS: [u64; 4] = [1 << 6, 1 << 12, 1 << 24, 1 << 40]; + +/// Cached group counts spread across the range. +const PRESENT: [u64; 4] = [1, 16, 256, 4096]; + +/// Held fixed while the other axis sweeps. +const FIXED_PRESENT: u64 = 64; +const FIXED_SPAN: u64 = 1 << 24; + +/// Keeps the ownership chain alive around the track being fetched. +struct Sparse { + _broadcast: broadcast::Producer, + _track: track::Producer, + consumer: track::Consumer, + span: u64, +} + +impl Sparse { + /// `present` one-frame groups spread evenly across `0..span`. + fn new(span: u64, present: u64) -> Self { + let mut info = broadcast::Info::default(); + // Nothing may expire or evict mid-measurement, or later samples walk fewer groups. + let config = cache::Config::default() + .with_capacity(1 << 30) + .with_expiry(Duration::from_secs(24 * 60 * 60)); + info.pool = cache::Pool::new(config); + let broadcast = broadcast::Producer::new(info); + let track = broadcast.create_track("bench", None).unwrap(); + + let step = span / present; + for sequence in (0..present).map(|i| i * step) { + let mut group = track.create_group(group::Info { sequence }).unwrap(); + group + .write_frame(Timestamp::ZERO, Bytes::from_static(&[0; 64])) + .unwrap(); + group.finish().unwrap(); + } + + Self { + consumer: track.consume(), + _broadcast: broadcast, + _track: track, + span, + } + } + + fn walk(&self) -> usize { + futures::executor::block_on(moq_net::fuzz::walk_fetch(&self.consumer, 0, self.span)).unwrap() + } +} + +fn bench_span(c: &mut Criterion) { + let mut group = c.benchmark_group("fetch/span"); + group.throughput(Throughput::Elements(FIXED_PRESENT)); + for span in SPANS { + let sparse = Sparse::new(span, FIXED_PRESENT); + assert_eq!(sparse.walk() as u64, FIXED_PRESENT); + group.bench_function(BenchmarkId::from_parameter(format!("2^{}", span.ilog2())), |b| { + b.iter(|| sparse.walk()) + }); + } + group.finish(); +} + +fn bench_present(c: &mut Criterion) { + let mut group = c.benchmark_group("fetch/present"); + for present in PRESENT { + group.throughput(Throughput::Elements(present)); + let sparse = Sparse::new(FIXED_SPAN, present); + assert_eq!(sparse.walk() as u64, present); + group.bench_function(BenchmarkId::from_parameter(present), |b| b.iter(|| sparse.walk())); + } + group.finish(); +} + +criterion_group!(benches, bench_span, bench_present); +criterion_main!(benches); diff --git a/rs/moq-net/src/fuzz.rs b/rs/moq-net/src/fuzz.rs index 66cb16c1f0..fbf7a94acc 100644 --- a/rs/moq-net/src/fuzz.rs +++ b/rs/moq-net/src/fuzz.rs @@ -7,6 +7,8 @@ //! here also lets the tests below replay them on stable, which is what turns a crash //! the fuzzer found into a regression `just check` runs. //! +//! The announce and fetch benches borrow a few `pub` helpers here for the same reason. +//! //! Compiled only under `cfg(test)` or the `fuzz` feature, so none of this is part of //! the published API. See `fuzz/README.md` for the workflow. @@ -252,6 +254,17 @@ pub fn decode_announces(mut data: &[u8], compress: bool) -> Vec { resolved } +/// Walk the whole groups `start..end` out of `track` as a standalone FETCH does, +/// returning how many it read. The fetch bench's way into the private walk. +pub async fn walk_fetch(track: &crate::track::Consumer, start: u64, end: u64) -> Result { + let start = ietf::Location { + group: start, + object: 0, + }; + let end = ietf::Location { group: end, object: 0 }; + Ok(ietf::walk_fetch(track, start, end, 0).await?.groups.len()) +} + /// Feed a lite-07 announce stream through the stateful decoder, then check that our /// encoder's compression of what it resolved reads back as the same announcements. /// diff --git a/rs/moq-net/src/ietf/mod.rs b/rs/moq-net/src/ietf/mod.rs index 944c277893..9a77d09579 100644 --- a/rs/moq-net/src/ietf/mod.rs +++ b/rs/moq-net/src/ietf/mod.rs @@ -47,6 +47,8 @@ pub use parameters::*; pub use properties::Properties; pub use publish::*; pub use publish_namespace::*; +#[cfg(any(test, feature = "fuzz"))] +pub(crate) use publisher::walk_fetch; use publisher::*; pub use request::*; pub use session::*; diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 4d5aebfd48..130b70958c 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -57,7 +57,7 @@ enum FetchPrior { /// One group of a FETCH answer, read out of the cache so a later eviction cannot /// truncate what FETCH_OK already promised. -struct FetchedGroup { +pub(crate) struct FetchedGroup { sequence: u64, /// The Object ID of the first frame. first: u64, @@ -85,8 +85,8 @@ impl FetchedGroup { } /// A FETCH range read out of a track. -struct Walked { - groups: Vec, +pub(crate) struct Walked { + pub(crate) groups: Vec, /// Why the walk stopped before the end of the range, when it did. stopped: Option, } @@ -97,12 +97,22 @@ struct Walked { /// A group that does not exist below the newest one is a hole and is skipped; one at or /// past it is the end of the track, so the walk stops there. Any other failure answers /// the whole FETCH. -async fn walk_fetch(track: &track::Consumer, start: Location, end: Location, priority: u8) -> Result { +/// +/// With nothing upstream to ask, the cache is the whole track, so a hole skips straight +/// to the next cached group: the walk costs the groups it returns, not the range's span. +pub(crate) async fn walk_fetch( + track: &track::Consumer, + start: Location, + end: Location, + priority: u8, +) -> Result { let mut groups = Vec::new(); - for sequence in start.group..=end.group { - if sequence == end.group && end.object == 0 { + let mut next = Some(start.group); + while let Some(sequence) = next { + if sequence > end.group || (sequence == end.group && end.object == 0) { break; } + next = sequence.checked_add(1); let skip = match sequence == start.group { true => start.object, @@ -116,13 +126,22 @@ async fn walk_fetch(track: &track::Consumer, start: Location, end: Location, pri let mut group = match track.fetch_group(sequence, fetch).await { Ok(group) => group, - Err(Error::NotFound) if track.latest().is_some_and(|latest| sequence < latest) => continue, - Err(err @ Error::NotFound) => { - return Ok(Walked { - groups, - stopped: Some(err), - }); - } + Err(err @ Error::NotFound) => match track.latest() { + Some(latest) if sequence < latest => { + // Capped at the newest group, whose own probe judges where the track ends. + if !track.fetches_misses() { + let cached = track.next_cached(sequence + 1); + next = Some(cached.map_or(latest, |cached| cached.min(latest))); + } + continue; + } + _ => { + return Ok(Walked { + groups, + stopped: Some(err), + }); + } + }, Err(err) => return Err(err), }; @@ -3904,6 +3923,44 @@ mod serve_tests { } } + /// With nothing upstream, a hole skips to the next cached group instead of stepping + /// through every missing sequence, which would never finish here. + #[tokio::test] + async fn a_standalone_fetch_skips_a_sparse_span() { + const FAR: u64 = 1 << 40; + for version in [Version::Draft16, Version::Draft18] { + let h = serve(version); + for sequence in [0, 5, FAR] { + let mut group = h.track.create_group(group::Info { sequence }).unwrap(); + for object in 0..2 { + group + .write_frame(timestamp(), format!("{sequence}-{object}").into_bytes()) + .unwrap(); + } + group.finish().unwrap(); + } + settle().await; + + let buf = standalone_fetch( + &h, + Location { group: 0, object: 0 }, + Location { group: FAR, object: 0 }, + GroupOrder::Ascending, + ) + .await; + let (ok, objects) = fetch_answer(buf, version); + assert_eq!( + ok.end_location, + Location { + group: FAR + 1, + object: 0 + }, + "{version}" + ); + assert_eq!(objects, pairs([0, 5, FAR]), "{version}"); + } + } + /// Read a REQUEST_ERROR (or draft-14 FETCH_ERROR) off a refused FETCH, as its code. fn fetch_refusal(mut buf: bytes::Bytes, version: Version) -> u64 { let id = u64::decode(&mut buf, version).unwrap(); diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 88948b9990..95a7e3c93d 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -738,6 +738,28 @@ impl Consumer { self.peek_below(Some(sequence)) } + /// The lowest sequence at or above `from` any segment caches; see + /// [`track::Consumer::next_cached`]. + /// + /// Segment ranges are left to the fetch, so this is a superset of what + /// [`Self::cached_group`] serves and skipping to it never passes a servable group. + pub(crate) fn next_cached(&self, from: u64) -> Option { + let segments = self.state.read().segments.clone(); + segments + .iter() + .filter_map(|segment| segment.track.next_cached(from)) + .min() + } + + /// Whether the newest segment, where fetches go, serves cache misses; see + /// [`track::Consumer::fetches_misses`]. With no segment yet, the first route to + /// serve the track may bring a handler. + pub(crate) fn fetches_misses(&self) -> bool { + // Copied out: the segment's track takes its own lock. + let track = self.state.read().segments.last().map(|segment| segment.track.clone()); + track.is_none_or(|track| track.fetches_misses()) + } + /// Walk the segments newest-first for the first one holding a group in range. /// /// Every segment, not just the newest: a takeover splices in a fresh track with no diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 46138d817c..08cf30945d 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -2579,6 +2579,31 @@ impl Consumer { } } + /// The lowest sequence at or above `from` holding a live cached group, crossing gaps + /// in the group numbering. A splice answers from whichever segment caches one first. + pub(crate) fn next_cached(&self, from: u64) -> Option { + match &self.inner { + ConsumerKind::Plain(state) => state + .read() + .lookup + .range(from..) + .find(|(_, slot)| !slot.group.is_aborted()) + .map(|(sequence, _)| *sequence), + ConsumerKind::Spliced(resume) => resume.next_cached(from), + } + } + + /// Whether a group missing from the cache can still be fetched: a [`Dynamic`] (a + /// relay's upstream) serves misses. Without one, the cache is all the track has. + /// + /// A spliced track answers for its newest segment, which is where fetches go. + pub(crate) fn fetches_misses(&self) -> bool { + match &self.inner { + ConsumerKind::Plain(state) => state.read().fetch.lock().has_handlers(), + ConsumerKind::Spliced(resume) => resume.fetches_misses(), + } + } + /// A cached group by sequence, under the same terms as [`Self::peek_latest`]. Unlike a /// fetch, a peek does not refresh the group's cache standing, so it never keeps a /// group alive over one a subscriber actually read; an aborted (evicted) group is a