diff --git a/Cargo.toml b/Cargo.toml index 5b61f78cdd..7e30c895a7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -305,6 +305,12 @@ opt-level = 3 [profile.dev.package.sonora-simd] opt-level = 3 +# Same for the rubato resampler, whose SIMD intrinsics are all of +# `resample::tests::duration_and_samples_are_invariant_across_partitions_and_epochs`: +# 0.9 CPU-seconds unoptimized, 0.14 optimized. +[profile.dev.package.rubato] +opt-level = 3 + # The Go mirror stores native staticlibs in Git, whose per-file limit is 100 MB. # Keep all release entry points on the same compact native profile. [profile.release] diff --git a/quest/m0/ci/README.md b/quest/m0/ci/README.md index ef1a68b41b..e124a49791 100644 --- a/quest/m0/ci/README.md +++ b/quest/m0/ci/README.md @@ -160,7 +160,6 @@ The historical samples locate candidates; they are not controlled benchmarks. - [Verify the R2 Rust cache](/quest/m0/ci/mbx-r2-verify.md) - confirm the hosted Nightly jobs reuse the previous Nightly, then measure restore time and expiry - [Verify the Nix cache](/quest/m0/ci/nix-cache-verify.md) - confirm releases use `nix.moq.dev`, check its bucket sizes, and retire Cachix - [Unify dependency features](/quest/m0/ci/hakari.md) - a cargo-hakari workspace-hack so every job compiles each dependency alike and mbx hits across jobs -- [Halve the slowest tests' CPU](/quest/m0/ci/slow-tests.md) - the slowest 1% of tests take under half their 289 CPU-seconds, with coverage kept - [Release credentials off main](/quest/m0/ci/publish-release-only.md) - `publish` admits only `release` and release tags once its `workflow_run` publishers run on their tag - [The queue host keeps the queue's disk](/quest/m0/ci/queue-host-disk.md) - agent sessions can't starve the queue runner of disk, and a crashed runner restarts - [The next test-CPU cuts](/quest/m0/ci/test-cost-followups.md) - the slow-test leftovers from #6987, cut at their source diff --git a/quest/m0/ci/five-minutes.md b/quest/m0/ci/five-minutes.md index 624e4aed67..e52f9c5268 100644 --- a/quest/m0/ci/five-minutes.md +++ b/quest/m0/ci/five-minutes.md @@ -67,8 +67,8 @@ What remains for a moq-net edit, largest first: lever was measured at about 30 ms of CPU per build and dropped), or running the rs lint beside the test build in a second target directory, which trades CPU for wall time against the epic's CPU measure. Decided 2026-10-30 (user): defer that - parallel lint; measure it after the slow-tests and hakari quests land. -- The test run is the [slow tests](/quest/m0/ci/slow-tests.md) quest's. + parallel lint; measure it after the hakari quest lands (slow-tests landed in + [moq-dev/moq#6987](https://github.com/moq-dev/moq/pull/6987)). - The lint's remaining 230 s: the dev loom compile and `cargo doc`. Decided 2026-10-31 (maintainer, see the [epic](/quest/m0/ci/README.md)): the gate drops the loom compile, the Send-depth check (now its own Nightly job), the @@ -80,14 +80,13 @@ What remains for a moq-net edit, largest first: - `just test` has no root recipe: bare `just test` lists the cross-language suites. Name which command the `just test` baseline above counts before measuring it. -- Prefer the epic's levers, such as [slow tests](/quest/m0/ci/slow-tests.md), - [hakari](/quest/m0/ci/hakari.md), and cache hits, and tighter diff selection, +- Prefer the epic's levers, such as [hakari](/quest/m0/ci/hakari.md) and cache + hits, and tighter diff selection, over moving coverage. Move a check to the queue or nightly only under the epic's rule: evidence and an explicit coverage tradeoff for each. - Split any large single lever into its own quest. ## Related -- [Halve the slowest tests' CPU](/quest/m0/ci/slow-tests.md) - the largest test-time lever - [Unify dependency features](/quest/m0/ci/hakari.md) - cache hits across feature shapes - [Land main through quest land](/quest/m0/ci/land.md) - the queue gate that keeps checks moved out of the local gate diff --git a/quest/m0/ci/slow-tests.md b/quest/m0/ci/slow-tests.md deleted file mode 100644 index f2a0e50466..0000000000 --- a/quest/m0/ci/slow-tests.md +++ /dev/null @@ -1,77 +0,0 @@ -# [M] Halve the slowest tests' CPU - -## Goal - -The 103 test IDs frozen in [moq-dev/moq#6914](https://github.com/moq-dev/moq/pull/6914) -(the slowest `ceil(10,275 / 100)` of one full normal `cargo nextest` run) take under -68.92 CPU-seconds together, half their 137.83 s baseline, with every case and its -coverage preserved. Per-test CPU is the test process's user plus system time, not -nextest's wall-clock duration. Measure the same IDs on the same machine as the baseline -it is compared against; missing or failed cases are not imputed. - -## Plan - -Decided with the maintainer on 2026-10-26, from moq-dev/rust2's -[test selection report](https://github.com/moq-dev/rust2/pull/69): the slowest 1% -of tests make up 57% of what either package selection or a function-level selector -would run, so cutting them beats building a selector. - -- Prefer making a test cheaper: smaller fixtures, shared setup, simulated time instead - of real waits, fewer redundant cases. -- Moving a test to nightly only follows this epic's rule: evidence and an explicit - coverage tradeoff for each. - -### Measurement - -[moq-dev/moq#6914](https://github.com/moq-dev/moq/pull/6914) holds the frozen list with -per-test baseline and after CPU, plus the runner wrapper and command. The wrapper is a -`CARGO_TARGET__RUNNER` that records each nextest test child's `wait4` user plus -system time, run under the normal `just rs test --workspace --exclude moq-net-fuzz ---exclude moq-quic-fuzz` with zero retries. On a different machine, re-measure the -baseline source (`6f6a3ae21432a3df74d4c432451f3909da339bee`) and the candidate with the -same wrapper rather than comparing against these numbers. Unchanged tests varied -between runs on a shared host (moq-stats' `produce::tests` by up to 40%), so credit only -the tests a change touches. - -### Done so far - -[moq-dev/moq#6914](https://github.com/moq-dev/moq/pull/6914) cut the cohort from -137.83 s to 116.83 s with fixture-only changes: the two MPEG-TS reset tests in -`container::ts::import` drain on the exporter's deadlines with a retention derived from -their stimulus instead of an hour of idle output (12.98 s to 0.18 s), and the jitter -`drift` and `feed` helpers advance a `kio::clock::Deadlines` queue instead of the tokio -paused clock (10.16 s to 6.80 s across 8 tests). About 48 s remains to cut. - -Tried and dropped, so don't redo them without new evidence: - -- Caching the jitter lead median: the frame-rate benchmark stayed within noise and one - cell regressed about 6%. -- A containment shortcut for miss-verdict intersections in moq-net: sparse gap and - range sweeps were unchanged (4096 timed gaps, 2.796 ms to 2.783 ms). The broader - algorithmic fix landed in - [moq-dev/moq#6921](https://github.com/moq-dev/moq/pull/6921). -- Retaining only one phase's span in the TS reset tests: it shed the boundary and - failed the unsignalled-reset assertion. - -### Remaining cost - -After CPU of the frozen cohort by module, largest first: - -| Module | Tests | CPU s | -| --- | ---: | ---: | -| moq-mux `container::ts::export_timing_test` | 14 | 22.2 | -| moq-net `model::track::test` | 5 | 17.3 | -| moq-mux `container::ts::recovery_test` | 18 | 11.6 | -| moq-net integration `mesh_withdraw` | 6 | 9.7 | -| moq-net integration `routing_contracts::handoff` | 3 | 8.3 | -| moq-mux `jitter::tests` | 8 | 6.8 | -| moq-quic `tests::sim` | 3 | 4.6 | -| moq-stats `produce::tests` | 4 | 4.0 | -| everything else | 42 | 32.4 | - -The moq-net `model::track::test` gap tests run through the same miss judging, which -[moq-dev/moq#6921](https://github.com/moq-dev/moq/pull/6921) made incremental after this -measurement, so re-measure them first; the export timing and recovery tests are the -largest fixture-only target. - -Public API: none. diff --git a/rs/moq-audio/src/decode/consumer.rs b/rs/moq-audio/src/decode/consumer.rs index d40957e900..638a7dc7c1 100644 --- a/rs/moq-audio/src/decode/consumer.rs +++ b/rs/moq-audio/src/decode/consumer.rs @@ -2056,11 +2056,12 @@ mod tests { /// /// PCM rather than Opus only for speed: the frame term comes from the decoded /// sample count either way, and an unoptimized Opus codec takes minutes over - /// the whole corpus. [`opus_states_its_own_frame_duration`] covers Opus. + /// the whole corpus. [`opus_states_its_own_frame_duration`] covers Opus. The + /// lowest common rate keeps each packet small, since only its duration counts. async fn replay(case: &crate::jitter::tests::Case) -> Vec { use crate::jitter::tests::duration; - let rate = 48_000; + let rate = 8_000; let samples = (case.frame * rate as f64 / 1000.0) as usize; let mut catalog = hang::catalog::AudioConfig::new(hang::catalog::AudioCodec::Pcm, rate, 1); catalog.jitter = case.advertised.map(duration); diff --git a/rs/moq-cli/src/publish.rs b/rs/moq-cli/src/publish.rs index 677e1e4d0a..836e465648 100644 --- a/rs/moq-cli/src/publish.rs +++ b/rs/moq-cli/src/publish.rs @@ -1739,10 +1739,11 @@ pub(crate) mod tests { .into_iter() .map(|(sequence, frames)| (sequence, frames.expect("the recording reads every group whole"))) .collect(); - let output: Vec = recording + let output = recording .iter() - .flat_map(|(_, frames)| frames.iter().flat_map(|frame| frame.payload.iter().copied())) - .collect(); + .flat_map(|(_, frames)| frames.iter().map(|frame| &frame.payload[..])) + .collect::>() + .concat(); assert!(output.len() > KYRION.len(), "{version}: both copies crossed"); assert!( input.ends_with(&output), diff --git a/rs/moq-hls/src/import.rs b/rs/moq-hls/src/import.rs index 846fb9f2f6..890a3a259d 100644 --- a/rs/moq-hls/src/import.rs +++ b/rs/moq-hls/src/import.rs @@ -186,7 +186,9 @@ fn same_origin(entry: &Url, url: &Url) -> bool { /// Fetches playlists and segments over HTTP or from the filesystem, applying HLS byte /// ranges (and validating that the server honored them). struct Fetcher { - client: Client, + /// `None` for a `file` playlist: every fetch stays on its scheme, so it never needs one, nor + /// the system trust roots building one loads. + client: Option, /// The caller's playlist URL. A playlist is untrusted, so every fetch stays on its origin. entry: Url, } @@ -198,6 +200,10 @@ impl Fetcher { return Err(Error::ResourceRefused); } + if entry.scheme() == "file" { + return Ok(Self { client: None, entry }); + } + // An intermediate hop is already a request to its origin, so check each one before // following it. A stopped hop reaches `fetch` as a redirect response, refused there. let origin = entry.clone(); @@ -222,7 +228,10 @@ impl Fetcher { .timeout(REQUEST_TIMEOUT) .redirect(redirect) .build()?; - Ok(Self { client, entry }) + Ok(Self { + client: Some(client), + entry, + }) } /// Fetch and parse a media playlist, with the URL its relative URIs resolve against. @@ -251,7 +260,8 @@ impl Fetcher { if !same_origin(&self.entry, url) { return Err(Error::ResourceRefused); } - if url.scheme() == "file" { + // Same origin, so a file URL exactly when the entry is one. + let Some(client) = &self.client else { let path = url.to_file_path().map_err(|_| Error::InvalidFileUrl)?; let Some(range) = resource.range else { let bytes = tokio::fs::read(&path).await.map(Bytes::from).map_err(Error::from)?; @@ -265,50 +275,49 @@ impl Fetcher { .read_to_end(&mut bytes) .await .map_err(Error::from)?; - Ok(( + return Ok(( Self::validate_range_length(url, range, Bytes::from(bytes))?, url.clone(), - )) - } else { - let response = self.request(resource).send().await?; - if response.status().is_redirection() { - return Err(Error::ResourceRefused); - } - let mut response = response.error_for_status()?; - let served = response.url().clone(); - let Some(range) = resource.range else { - return Ok((response.bytes().await?, served)); - }; + )); + }; + let response = Self::request(client, resource).send().await?; + if response.status().is_redirection() { + return Err(Error::ResourceRefused); + } + let mut response = response.error_for_status()?; + let served = response.url().clone(); + let Some(range) = resource.range else { + return Ok((response.bytes().await?, served)); + }; - // An origin that ignores `Range` sends the whole resource, which would cost every - // earlier byte of the file for each segment. - if response.status() != reqwest::StatusCode::PARTIAL_CONTENT { - return Err(Error::ByteRangeIgnored { + // An origin that ignores `Range` sends the whole resource, which would cost every + // earlier byte of the file for each segment. + if response.status() != reqwest::StatusCode::PARTIAL_CONTENT { + return Err(Error::ByteRangeIgnored { + url: redact(url), + status: response.status().as_u16(), + }); + } + Self::validate_content_range(url, range, response.headers())?; + + // Stream the body so an overrun fails without buffering the excess. + let mut body = BytesMut::new(); + while let Some(chunk) = response.chunk().await? { + let read = body.len() + chunk.len(); + if !u64::try_from(read).is_ok_and(|read| read <= range.length) { + return Err(Error::ByteRangeLengthMismatch { url: redact(url), - status: response.status().as_u16(), + expected: range.length, + actual: read, }); } - Self::validate_content_range(url, range, response.headers())?; - - // Stream the body so an overrun fails without buffering the excess. - let mut body = BytesMut::new(); - while let Some(chunk) = response.chunk().await? { - let read = body.len() + chunk.len(); - if !u64::try_from(read).is_ok_and(|read| read <= range.length) { - return Err(Error::ByteRangeLengthMismatch { - url: redact(url), - expected: range.length, - actual: read, - }); - } - body.extend_from_slice(&chunk); - } - Ok((Self::validate_range_length(url, range, body.freeze())?, served)) + body.extend_from_slice(&chunk); } + Ok((Self::validate_range_length(url, range, body.freeze())?, served)) } - fn request(&self, resource: &Resource) -> reqwest::RequestBuilder { - let mut request = self.client.get(resource.url.clone()); + fn request(client: &Client, resource: &Resource) -> reqwest::RequestBuilder { + let mut request = client.get(resource.url.clone()); if let Some(range) = resource.range { let end = range.start + range.length - 1; request = request.header(reqwest::header::RANGE, format!("bytes={}-{}", range.start, end)); @@ -701,7 +710,8 @@ pub struct Import { impl Import { /// Create a new HLS import that will write into the given broadcast. /// - /// Installs rind's rustls provider as the process default unless one already is. + /// For an HTTP playlist, installs rind's rustls provider as the process default unless one + /// already is. A file playlist builds no HTTP client, so it installs none. pub fn new(broadcast: moq_net::broadcast::Producer, catalog: CatalogProducer, cfg: Config) -> Result { let fetcher = Fetcher::new(&cfg)?; Ok(Self { diff --git a/rs/moq-json/src/snapshot/encoder.rs b/rs/moq-json/src/snapshot/encoder.rs index 46548d8fcf..6c3a4112f0 100644 --- a/rs/moq-json/src/snapshot/encoder.rs +++ b/rs/moq-json/src/snapshot/encoder.rs @@ -584,6 +584,7 @@ mod test { let mut runner = TestRunner::new(PropertyConfig { cases, rng_seed: RngSeed::Fixed(u64::from(seed)), + rng_algorithm: proptest::test_runner::RngAlgorithm::XorShift, failure_persistence: None, ..Default::default() }); diff --git a/rs/moq-json/src/stream/encoder.rs b/rs/moq-json/src/stream/encoder.rs index 765d0613c4..8c9e1118c9 100644 --- a/rs/moq-json/src/stream/encoder.rs +++ b/rs/moq-json/src/stream/encoder.rs @@ -48,13 +48,20 @@ impl Pending<'_, T> { /// /// Payloads must reach the wire in order. A group has no size cap, so the encoder keeps no budget. pub struct Encoder { + /// The most bytes one record may hold. A field rather than the constant so a test can shrink it + /// to a size it can reach. + pub(super) max_frame_bytes: u64, + _marker: PhantomData, } impl Encoder { /// Create an encoder. pub fn new(_config: Config) -> Self { - Self { _marker: PhantomData } + Self { + max_frame_bytes: moq_net::group::MAX_CACHE_BYTES, + _marker: PhantomData, + } } } @@ -68,7 +75,7 @@ impl Encoder { /// ([`moq_net::group::MAX_CACHE_BYTES`]). The refused record leaves the encoder untouched. pub fn encode(&mut self, value: &T) -> Result> { let payload = Bytes::from(serde_json::to_vec(value)?); - if payload.len() as u64 > moq_net::group::MAX_CACHE_BYTES { + if payload.len() as u64 > self.max_frame_bytes { return Err(moq_net::Error::FrameTooLarge.into()); } diff --git a/rs/moq-json/src/stream/mod.rs b/rs/moq-json/src/stream/mod.rs index 992bcb7ef0..24f36ca957 100644 --- a/rs/moq-json/src/stream/mod.rs +++ b/rs/moq-json/src/stream/mod.rs @@ -380,15 +380,19 @@ mod test { matches!(result, Err(crate::Error::Net(moq_net::Error::FrameTooLarge))) } + /// A frame limit small enough to pass cheaply, standing in for moq-net's 32 MiB cache. + const BUDGET: u64 = 32 * 1024; + /// A record larger than a frame is refused before anything is written, so the log carries on: /// the next record lands and decodes. #[test] fn an_oversized_record_is_refused_and_the_log_continues() { for compression in [false, true] { let (mut producer, track) = producer_on(Config::default(), info(compression)); + producer.set_max_frame_bytes(BUDGET); producer.append(&json!({ "n": 0 })).unwrap(); - let oversized = Value::String("x".repeat(moq_net::group::MAX_CACHE_BYTES as usize)); + let oversized = Value::String("x".repeat(BUDGET as usize)); assert!(refused(producer.append(&oversized)), "compression={compression}"); producer.append(&json!({ "n": 1 })).unwrap(); diff --git a/rs/moq-json/src/stream/producer.rs b/rs/moq-json/src/stream/producer.rs index b723ceb959..e855cc520e 100644 --- a/rs/moq-json/src/stream/producer.rs +++ b/rs/moq-json/src/stream/producer.rs @@ -101,6 +101,14 @@ impl Producer { } } +#[cfg(test)] +impl Producer { + /// Shrink the frame limit to a size a test can reach. + pub(super) fn set_max_frame_bytes(&self, bytes: u64) { + self.inner.lock().unwrap().encoder.max_frame_bytes = bytes; + } +} + /// Shared publishing state behind [`Producer`]'s `Arc`. /// /// The track and the encoder are separate fields so a [`Pending`](super::Pending) record (which diff --git a/rs/moq-json/src/window/encoder.rs b/rs/moq-json/src/window/encoder.rs index e5d64a940c..e43c113c7c 100644 --- a/rs/moq-json/src/window/encoder.rs +++ b/rs/moq-json/src/window/encoder.rs @@ -158,6 +158,10 @@ pub struct Encoder { /// Frames emitted into the current group, header included. group_frames: usize, + /// The most bytes a group may hold, the cap both a header and the ops after it meet. A field + /// rather than the constant so a test can shrink it to a size it can reach. + max_group_bytes: u64, + /// Whether the next frame must be a header because a frame was lost. Kept separate from the /// window, which a resync must never discard. resync: bool, @@ -179,6 +183,7 @@ impl Encoder { start: 0, group_bytes: 0, group_frames: 0, + max_group_bytes: moq_net::group::MAX_CACHE_BYTES, resync: true, _marker: PhantomData, } @@ -257,7 +262,7 @@ impl Encoder { let payload = Bytes::from(bytes); self.group_frames += 1; self.group_bytes = self.group_bytes.saturating_add(payload.len() as u64); - if self.group_bytes > moq_net::group::MAX_CACHE_BYTES { + if self.group_bytes > self.max_group_bytes { self.reset(); return Ok(None); } @@ -293,7 +298,7 @@ impl Encoder { /// Encode the header restating the whole window and opening a new group. fn emit_header(&mut self, bytes: Vec) -> Result { let payload = Bytes::from(bytes); - if payload.len() as u64 > moq_net::group::MAX_CACHE_BYTES { + if payload.len() as u64 > self.max_group_bytes { return Err(Error::Json("window header exceeds the group cache limit".into())); } @@ -414,11 +419,15 @@ impl Encoder { mod test { use super::*; + /// A group budget small enough to fill cheaply, standing in for moq-net's 32 MiB cache. + const BUDGET: u64 = 32 * 1024; + #[test] fn an_op_that_would_evict_the_header_rolls_first() { let mut encoder = Encoder::::new(ProducerConfig::default().with_ops(true)); - let first = "a".repeat(16 * 1024 * 1024); - let next = "b".repeat(15 * 1024 * 1024); + encoder.max_group_bytes = BUDGET; + let first = "a".repeat(16 * 1024); + let next = "b".repeat(15 * 1024); let frame = encoder.push(&first).unwrap(); assert!(frame.keyframe); @@ -434,7 +443,7 @@ mod test { let frame = encoder.push(&next).unwrap(); assert!(frame.keyframe); - assert!(frame.payload.len() < moq_net::group::MAX_CACHE_BYTES as usize); + assert!(frame.payload.len() < BUDGET as usize); frame.commit(); } @@ -457,7 +466,8 @@ mod test { #[test] fn a_header_larger_than_the_group_cache_is_rejected() { let mut encoder = Encoder::::new(ProducerConfig::default()); - let record = "x".repeat(moq_net::group::MAX_CACHE_BYTES as usize); + encoder.max_group_bytes = BUDGET; + let record = "x".repeat(BUDGET as usize); let err = encoder.push(&record).err().expect("oversized header should fail"); assert!(err.to_string().contains("group cache limit")); diff --git a/rs/moq-json/src/window/properties.rs b/rs/moq-json/src/window/properties.rs index c55e46ae5f..680e904089 100644 --- a/rs/moq-json/src/window/properties.rs +++ b/rs/moq-json/src/window/properties.rs @@ -238,6 +238,7 @@ fn window_codec_property_matches_every_readers_logical_window() { let mut runner = TestRunner::new(Config { cases, rng_seed: RngSeed::Fixed(u64::from(seed)), + rng_algorithm: proptest::test_runner::RngAlgorithm::XorShift, failure_persistence: None, ..Default::default() }); diff --git a/rs/moq-mux/src/container/ts/export_timing_test.rs b/rs/moq-mux/src/container/ts/export_timing_test.rs index 4d39e22c85..1392ff3828 100644 --- a/rs/moq-mux/src/container/ts/export_timing_test.rs +++ b/rs/moq-mux/src/container/ts/export_timing_test.rs @@ -3,11 +3,15 @@ //! resumes, when the source's clock runs off the receiver's, and when two receivers render one //! broadcast for a 1+1 (SMPTE ST 2022-7) pair. //! -//! Every case runs on the paused clock, with each frame written at the instant a live source -//! would send it. A receiver's view is measured from its own output: the lead of each PES (its -//! decode time less the PCR of the slot carrying it) and the instant each slot went out. +//! Every case runs on a deadline queue the test advances itself, with each frame written at the +//! instant a live source would send it. A receiver's view is measured from its own output: the +//! lead of each PES (its decode time less the PCR of the slot carrying it) and the instant each +//! slot went out. use std::io::Cursor; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::task::{Wake, Waker}; use std::time::Duration; use bytes::{Bytes, BytesMut}; @@ -64,6 +68,11 @@ fn length_prefixed(nals: &[&[u8]]) -> Bytes { /// A live H.264 + AAC broadcast whose frames are written on the source's clock. struct Live { + /// The receivers' clock: every timer the broadcast and its legs arm waits here, and only + /// [`Self::advance`] moves it. + clock: kio::clock::Deadlines, + /// Keeps [`Self::clock`] the thread's clock, unless another broadcast's is: see [`Self::beside`]. + _entered: Option, _broadcast: moq_net::broadcast::Producer, catalog: crate::catalog::Producer, /// Written group by group, so a group can be left open: see [`Self::run`]. @@ -95,6 +104,22 @@ struct Live { impl Live { async fn new(scale: f64, audio_lag: i64) -> Self { + let clock = kio::clock::Deadlines::new(kio::clock::now()); + let entered = clock.enter(); + Self::on(clock, Some(entered), scale, audio_lag).await + } + + /// Another broadcast on this one's clock. + async fn beside(&self, scale: f64, audio_lag: i64) -> Self { + Self::on(self.clock.clone(), None, scale, audio_lag).await + } + + async fn on( + clock: kio::clock::Deadlines, + entered: Option, + scale: f64, + audio_lag: i64, + ) -> Self { let mut broadcast = moq_net::broadcast::Info::new().produce(); let consumer = broadcast.consume(); let mut catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); @@ -141,6 +166,8 @@ impl Live { Self { source: crate::source::announced(&consumer).await, + clock, + _entered: entered, _broadcast: broadcast, catalog, video, @@ -205,9 +232,9 @@ impl Live { /// Send nothing for `gap`, as a publisher that died, then resume at the live edge on the /// same timeline from the next group, as its replacement or the other half of a 1+1 pair /// would. - async fn fall_silent(&mut self, gap: Duration, legs: &mut [&mut Leg]) { + fn fall_silent(&mut self, gap: Duration, legs: &mut [&mut Leg]) { let resume = kio::clock::now() + gap; - advance(resume, legs).await; + self.advance(resume, legs); let media = (resume.duration_since(self.start).as_secs_f64() * self.scale * 1e6) as u64; self.tick = media.div_ceil(VIDEO_US * GOP) * GOP; self.audio_index = media.div_ceil(AUDIO_US); @@ -220,10 +247,13 @@ impl Live { .unwrap() .with_delay(DELAY) .with_mux_rate(self.rate); + let woken = Arc::new(Woken(AtomicBool::new(true))); Leg { export, every, next_poll: kio::clock::now(), + waiter: kio::Waiter::new(Waker::from(woken.clone())), + woken, out: Vec::new(), } } @@ -235,12 +265,12 @@ impl Live { /// still on its way when the next group starts, as when the relay's backlog for a joiner /// arrives behind the live edge. The subscription then skips it as slow once the next /// group is the delay newer, and counts a discontinuity. - async fn run(&mut self, until: u64, legs: &mut [&mut Leg], stall: impl Fn(u64) -> bool) { + fn run(&mut self, until: u64, legs: &mut [&mut Leg], stall: impl Fn(u64) -> bool) { while self.tick < until { let video = self.tick * VIDEO_US; let audio = self.audio_index * AUDIO_US; let (video_at, audio_at) = self.send_times(video, audio); - advance(video_at.min(audio_at), legs).await; + self.advance(video_at.min(audio_at), legs); if audio_at < video_at { self.send_audio(); } else { @@ -304,10 +334,10 @@ impl Live { /// Send audio alone, each frame at its instant, until media time `until` (µs), as a /// broadcast whose video has stopped. - async fn run_audio(&mut self, until: u64, legs: &mut [&mut Leg]) { + fn run_audio(&mut self, until: u64, legs: &mut [&mut Leg]) { while self.audio_index * AUDIO_US < until { let (_, at) = self.send_times(0, self.audio_index * AUDIO_US); - advance(at, legs).await; + self.advance(at, legs); self.send_audio(); for leg in legs.iter_mut() { leg.poll_now(); @@ -316,7 +346,7 @@ impl Live { } /// End the broadcast and let every leg render what it holds. - async fn finish(&mut self, legs: &mut [&mut Leg]) { + fn finish(&mut self, legs: &mut [&mut Leg]) { if let Some(mut group) = self.group.take() { group.finish().unwrap(); } @@ -324,7 +354,35 @@ impl Live { self.video.finish().unwrap(); self.audio.finish().unwrap(); for leg in legs.iter_mut() { - leg.drain().await; + leg.drain(&self.clock); + } + } + + /// Run the clock to `at`, letting every leg read whenever it would have on its own. + fn advance(&self, at: Instant, legs: &mut [&mut Leg]) { + while let Some(wake) = legs + .iter() + .filter_map(|leg| leg.wake()) + .filter(|wake| *wake <= at) + .min() + { + self.clock.advance(wake.max(self.clock.now())); + let now = kio::clock::now(); + for leg in legs.iter_mut() { + if leg.wake().is_some_and(|wake| wake <= now) { + leg.poll(); + if !leg.every.is_zero() { + while leg.next_poll <= now { + leg.next_poll += leg.every; + } + } + } + } + } + self.clock.advance(at.max(self.clock.now())); + // Read up to the instant the next frame arrives, so the arrival is timed exactly. + for leg in legs.iter_mut() { + leg.poll_now(); } } } @@ -332,13 +390,29 @@ impl Live { /// One receiver: an exporter and what it rendered, with the instant each slot went out. struct Leg { export: Export, - /// How often it reads its subscription. Zero reads after every frame the source sends; a - /// coarser cadence stands for a receiver behind a burstier network or scheduler. + /// How often it reads its subscription. Zero reads whenever its export is woken, as by each + /// frame the source sends; a coarser cadence stands for a receiver behind a burstier network + /// or scheduler. every: Duration, next_poll: Instant, + /// What its export registers to be woken by, and whether it was since the last read. + waiter: kio::Waiter, + woken: Arc, out: Vec<(Instant, Frame)>, } +struct Woken(AtomicBool); + +impl Wake for Woken { + fn wake(self: Arc) { + self.0.store(true, Ordering::Relaxed); + } + + fn wake_by_ref(self: &Arc) { + self.0.store(true, Ordering::Relaxed); + } +} + impl Leg { /// When this leg next reads on its own. fn wake(&self) -> Option { @@ -348,38 +422,41 @@ impl Leg { } } + /// Read if anything woke the export since it last read, as the task polling it would: a read + /// nothing woke finds nothing new. fn poll_now(&mut self) { - if !self.every.is_zero() { - return; + if self.every.is_zero() && self.woken.0.load(Ordering::Relaxed) { + self.poll(); } - self.poll(); } - fn poll(&mut self) { - let waiter = kio::Waiter::noop(); + /// Read everything ready now, returning whether the broadcast ended. + fn poll(&mut self) -> bool { + self.woken.0.store(false, Ordering::Relaxed); let now = kio::clock::now(); - while let std::task::Poll::Ready(frame) = self.export.poll_next(&waiter) { + while let std::task::Poll::Ready(frame) = self.export.poll_next(&self.waiter) { match frame.expect("exporter error") { Some(frame) => self.out.push((now, frame)), - None => break, + None => return true, } } + false } - /// Read to the end of the broadcast. - async fn drain(&mut self) { - loop { - self.poll(); + /// Read to the end of the broadcast, or until nothing is due and 10 ms pass with no end. + /// + /// Only `clock` fires the export's timers, so the 10 ms is an advance of it: a timeout armed + /// on it would never fire. + fn drain(&mut self, clock: &kio::clock::Deadlines) { + while !self.poll() { if let Some(deadline) = self.export.next_due() { - kio::clock::sleep_until(deadline).await; + clock.advance(deadline.max(clock.now())); continue; } - match kio::clock::timeout(Duration::from_millis(10), self.export.next()).await { - Ok(Ok(Some(frame))) => self.out.push((kio::clock::now(), frame)), - Ok(Ok(None)) => break, - Err(_) if self.export.next_due().is_none() => break, - Err(_) => {} - Ok(Err(err)) => panic!("exporter error: {err}"), + clock.advance(clock.now() + Duration::from_millis(10)); + if self.export.next_due().is_none() { + self.poll(); + break; } } } @@ -390,34 +467,6 @@ impl Leg { } } -/// Run the clock to `at`, letting every leg read whenever it would have on its own. -async fn advance(at: Instant, legs: &mut [&mut Leg]) { - while let Some(wake) = legs - .iter() - .filter_map(|leg| leg.wake()) - .filter(|wake| *wake <= at) - .min() - { - kio::clock::sleep_until(wake).await; - let now = kio::clock::now(); - for leg in legs.iter_mut() { - if leg.wake().is_some_and(|wake| wake <= now) { - leg.poll(); - if !leg.every.is_zero() { - while leg.next_poll <= now { - leg.next_poll += leg.every; - } - } - } - } - } - kio::clock::sleep_until(at).await; - // Read up to the instant the next frame arrives, so the arrival is timed exactly. - for leg in legs.iter_mut() { - leg.poll_now(); - } -} - /// A PES start as a receiver sees it. #[derive(Debug, Clone, Copy)] struct Unit { @@ -545,9 +594,9 @@ const CLOCK_SLACK: f64 = 2.0 * SLOT_MS; /// return the leg's units. async fn join(lag: i64, stall: impl Fn(u64) -> bool) -> Vec { let mut live = Live::new(1.0, lag).await; - live.run(JOIN, &mut [], |_| false).await; + live.run(JOIN, &mut [], |_| false); let mut leg = live.join(Duration::ZERO).await; - live.run(24 * GOP, &mut [&mut leg], stall).await; + live.run(24 * GOP, &mut [&mut leg], stall); assert_eq!(leg.export.dropped(), 0, "every frame arrived inside its deadline"); leg.units() } @@ -610,9 +659,9 @@ async fn a_track_sent_later_than_the_delay_loses_nothing() { for lag in [LONG_LAG, -LONG_LAG] { for join in [3 * GOP, JOIN] { let mut live = Live::new(1.0, lag).await; - live.run(join, &mut [], |_| false).await; + live.run(join, &mut [], |_| false); let mut leg = live.join(Duration::ZERO).await; - live.run(join + 20 * GOP, &mut [&mut leg], |_| false).await; + live.run(join + 20 * GOP, &mut [&mut leg], |_| false); let dropped = leg.export.dropped(); if dropped > 0 { lost.push(format!( @@ -641,10 +690,10 @@ async fn an_unmarked_step_keeps_the_export_rendering() { for lag in [0, LAGS[0], LAGS[1]] { let mut live = Live::new(1.0, lag).await; let mut leg = live.join(Duration::ZERO).await; - live.run(8 * GOP + GOP / 2, &mut [&mut leg], |_| false).await; + live.run(8 * GOP + GOP / 2, &mut [&mut leg], |_| false); live.offset += step; let stepped = kio::clock::now(); - live.run(live.tick + 8 * GOP, &mut [&mut leg], |_| false).await; + live.run(live.tick + 8 * GOP, &mut [&mut leg], |_| false); let case = format!("{} s step, lag {lag}", step as f64 / 1e6); assert_eq!(leg.export.dropped(), 0, "{case}"); @@ -683,10 +732,10 @@ async fn frames_held_through_a_silence_drop_as_late() { } let mut leg = live.join(Duration::ZERO).await; let tail = if declared { 2 } else { 5 }; - live.run(8 * GOP + tail, &mut [&mut leg], |_| false).await; - live.fall_silent(Duration::from_secs(3), &mut [&mut leg]).await; + live.run(8 * GOP + tail, &mut [&mut leg], |_| false); + live.fall_silent(Duration::from_secs(3), &mut [&mut leg]); let resumed = kio::clock::now(); - live.run(live.tick + 8 * GOP, &mut [&mut leg], |_| false).await; + live.run(live.tick + 8 * GOP, &mut [&mut leg], |_| false); let dropped = leg.export.dropped(); assert!( @@ -729,12 +778,12 @@ async fn a_cut_judges_held_video_when_the_loss_is_observed() { let mut leg = live.join(Duration::ZERO).await; // The declared clock holds only the group's first depth frames. let tail = if declared { 2 } else { 8 }; - live.run(8 * GOP + tail, &mut [&mut leg], |_| false).await; - live.fall_silent(silence, &mut [&mut leg]).await; + live.run(8 * GOP + tail, &mut [&mut leg], |_| false); + live.fall_silent(silence, &mut [&mut leg]); let observed = kio::clock::now(); live.group.take().unwrap().abort(moq_net::Error::Unroutable).unwrap(); leg.poll_now(); - live.run(live.tick + 8 * GOP, &mut [&mut leg], |_| false).await; + live.run(live.tick + 8 * GOP, &mut [&mut leg], |_| false); assert_eq!( leg.export.dropped() > 0, !silence.is_zero(), @@ -768,8 +817,8 @@ async fn a_successor_settles_held_video_across_a_skip_or_restart() { } let mut leg = live.join(Duration::ZERO).await; let tail = if declared { 2 } else { 8 }; - live.run(8 * GOP + tail, &mut [&mut leg], |_| false).await; - live.fall_silent(Duration::from_secs(3), &mut [&mut leg]).await; + live.run(8 * GOP + tail, &mut [&mut leg], |_| false); + live.fall_silent(Duration::from_secs(3), &mut [&mut leg]); if restart { live.group.take().unwrap().finish().unwrap(); let mut marker = live.video.append_group().unwrap(); @@ -786,9 +835,9 @@ async fn a_successor_settles_held_video_across_a_skip_or_restart() { } // Deliver the successor with its boundary, so no clean-end Pending flush // runs before the new frame can settle the held tail. - live.run(live.tick + 1, &mut [], |_| false).await; + live.run(live.tick + 1, &mut [], |_| false); leg.poll_now(); - live.run(live.tick + 8 * GOP, &mut [&mut leg], |_| false).await; + live.run(live.tick + 8 * GOP, &mut [&mut leg], |_| false); assert!(leg.export.dropped() > 0, "declared {declared}, restart {restart}"); } } @@ -807,11 +856,11 @@ async fn a_clean_group_end_settles_held_video_before_a_silence() { } let mut leg = live.join(Duration::ZERO).await; let tail = if declared { 2 } else { 8 }; - live.run(8 * GOP + tail, &mut [&mut leg], |_| false).await; + live.run(8 * GOP + tail, &mut [&mut leg], |_| false); live.group.take().unwrap().finish().unwrap(); leg.poll_now(); - live.fall_silent(Duration::from_secs(3), &mut [&mut leg]).await; - live.run(live.tick + 8 * GOP, &mut [&mut leg], |_| false).await; + live.fall_silent(Duration::from_secs(3), &mut [&mut leg]); + live.run(live.tick + 8 * GOP, &mut [&mut leg], |_| false); assert_eq!(leg.export.dropped(), 0, "declared {declared}"); } } @@ -824,7 +873,7 @@ async fn video_held_for_most_of_the_delay_drops_nothing() { let mut live = Live::new(1.0, lag).await; live.declare_video_jitter(DELAY * 4 / 5); let mut leg = live.join(Duration::ZERO).await; - live.run(20 * GOP, &mut [&mut leg], |_| false).await; + live.run(20 * GOP, &mut [&mut leg], |_| false); assert_eq!(leg.export.dropped(), 0, "lag {lag}: frames dropped on a clean source"); } } @@ -844,9 +893,9 @@ async fn the_video_held_at_the_end_goes_out() { } let mut leg = live.join(Duration::ZERO).await; let last = 8 * GOP + if declared { 2 } else { 5 }; - live.run(last, &mut [&mut leg], |_| false).await; - live.run_audio((last + 3 * GOP) * VIDEO_US, &mut [&mut leg]).await; - live.finish(&mut [&mut leg]).await; + live.run(last, &mut [&mut leg], |_| false); + live.run_audio((last + 3 * GOP) * VIDEO_US, &mut [&mut leg]); + live.finish(&mut [&mut leg]); assert_eq!(leg.export.dropped(), 0, "lag {lag}: frames dropped"); let video = leg.units().iter().filter(|unit| unit.pid == VIDEO_PID).count() as u64; @@ -860,22 +909,23 @@ async fn the_video_held_at_the_end_goes_out() { // turn once the walk passes the delay, and one running fast makes the buffer grow without bound. // The output's clock may follow it only as fast as 13818-1 lets a system clock change, so // reaching a source 30 ppm off takes hours (the jitter buffer's own tests run a day); these -// cases run five minutes of a source at the limit, long enough to measure its rate. +// cases run a minute of a source at the limit, long enough to measure its rate: frames arrive +// on the source's exact clock, so the measurement settles within a few steering steps. /// How far the instant each slot goes out may wander against the source over a run: what /// reaching a source at the limit costs under the slew limit (30 ppm squared over twice /// 0.075 Hz/s of 27 MHz), and a slot. const DRIFT_SLACK: Duration = Duration::from_millis(162 + 25); -/// Five minutes of media. -const DRIFT_TICKS: u64 = 300_000_000 / VIDEO_US; +/// A minute of media. +const DRIFT_TICKS: u64 = 60_000_000 / VIDEO_US; /// The spread of (slot sent − source's clock at the slot) after the first 10 s over `ticks` /// of media, the drops, and the export's stats at the end. async fn on_a_scaled_clock(scale: f64, ticks: u64) -> (Duration, u64, Stats) { let mut live = Live::new(scale, 0).await; let mut leg = live.join(Duration::ZERO).await; - live.run(ticks, &mut [&mut leg], |_| false).await; + live.run(ticks, &mut [&mut leg], |_| false); let first = leg.out.first().expect("output").1.timestamp.as_micros() as u64; let held: Vec = leg @@ -913,7 +963,7 @@ async fn a_fast_source_clock_keeps_the_delay() { /// A source 400 ppm off, past what the output's clock may follow, is measured and counted. #[tokio::test(start_paused = true)] async fn a_source_past_the_limit_is_counted() { - let (_, _, stats) = on_a_scaled_clock(1.0004, 300_000_000 / VIDEO_US).await; + let (_, _, stats) = on_a_scaled_clock(1.0004, DRIFT_TICKS).await; let drift = stats.release.drift.expect("a measurement"); assert!((drift - 400.0).abs() < 1.0, "measured {drift} ppm"); assert!(stats.release.out_of_tolerance > 0, "not counted"); @@ -929,7 +979,7 @@ async fn a_source_past_the_limit_is_counted() { async fn a_source_far_off_real_time_fails_within_seconds() { let mut live = Live::new(1.0 / 1.22, 0).await; let mut leg = live.join(Duration::ZERO).await; - live.run(10_000_000 / VIDEO_US, &mut [&mut leg], |_| false).await; + live.run(10_000_000 / VIDEO_US, &mut [&mut leg], |_| false); } // A 1+1 pair is two receivers of one broadcast whose outputs a seamless switch compares packet @@ -958,13 +1008,20 @@ fn assert_same_packets(a: &Leg, b: &Leg, from: u64) { "slot at {:?} rendered to a different size", a.timestamp ); - for (offset, (x, y)) in a.payload.iter().zip(b.payload.iter()).enumerate() { - // Only the low nibble of byte 3 is the counter. - assert!( - x == y || (offset % 188 == 3 && (x ^ y) & 0xf0 == 0), - "slot at {:?} differs outside the continuity counter: offset {offset}, {x:#04x} vs {y:#04x}", - a.timestamp, - ); + for (packet, (x, y)) in a.payload.chunks(188).zip(b.payload.chunks(188)).enumerate() { + // Only the low nibble of byte 3 is the counter. Compared a packet at a time, then + // byte by byte only to name the offset that differs. + if x.len() > 3 && x[..3] == y[..3] && (x[3] ^ y[3]) & 0xf0 == 0 && x[4..] == y[4..] { + continue; + } + for (offset, (x, y)) in x.iter().zip(y).enumerate() { + assert!( + x == y || (offset == 3 && (x ^ y) & 0xf0 == 0), + "slot at {:?} differs outside the continuity counter: offset {}, {x:#04x} vs {y:#04x}", + a.timestamp, + packet * 188 + offset, + ); + } } } } @@ -980,10 +1037,10 @@ async fn pair_at(rate: u64, scale: f64, lag: i64, ticks: u64, stall: impl Fn(u64 let mut live = Live::new(scale, lag).await; live.rate = rate; let mut a = live.join(Duration::ZERO).await; - live.run(JOIN, &mut [&mut a], stall).await; + live.run(JOIN, &mut [&mut a], stall); let mut b = live.join(Duration::from_millis(100)).await; - live.run(ticks, &mut [&mut a, &mut b], stall).await; - live.finish(&mut [&mut a, &mut b]).await; + live.run(ticks, &mut [&mut a, &mut b], stall); + live.finish(&mut [&mut a, &mut b]); let joined = b.out.first().expect("the joiner rendered").1.timestamp.as_micros() as u64; (a, b, joined + 2_000_000) } @@ -1025,10 +1082,9 @@ async fn two_legs_render_a_drifting_source_the_same_way() { async fn a_joiner_keeps_the_system_clock_in_tolerance() { const WINDOW: u64 = 30_000_000; let mut live = Live::new(1.0, 0).await; - live.run(JOIN, &mut [], |_| false).await; + live.run(JOIN, &mut [], |_| false); let mut leg = live.join(Duration::ZERO).await; - live.run(JOIN + 300_000_000 / VIDEO_US, &mut [&mut leg], |_| false) - .await; + live.run(JOIN + 300_000_000 / VIDEO_US, &mut [&mut leg], |_| false); // The instant each slot first went out, by its PCR time. let mut first = std::collections::BTreeMap::new(); @@ -1073,9 +1129,9 @@ async fn a_joiner_keeps_the_system_clock_in_tolerance() { #[tokio::test(start_paused = true)] async fn a_joiner_runs_at_the_delay_from_its_first_output() { let mut live = Live::new(1.0, DELAY.as_micros() as i64).await; - live.run(JOIN, &mut [], |_| false).await; + live.run(JOIN, &mut [], |_| false); let mut leg = live.join(Duration::ZERO).await; - live.run(JOIN + 10_000_000 / VIDEO_US, &mut [&mut leg], |_| false).await; + live.run(JOIN + 10_000_000 / VIDEO_US, &mut [&mut leg], |_| false); assert_eq!(leg.export.dropped(), 0, "every frame arrived inside its deadline"); let held: Vec = leg @@ -1104,7 +1160,7 @@ async fn live_audio_shares_its_pes() { for lag in [DELAY.as_micros() as i64, LAGS[0], LAGS[1]] { let mut live = Live::new(1.0, lag).await; let mut leg = live.join(Duration::ZERO).await; - live.run(10_000_000 / VIDEO_US, &mut [&mut leg], |_| false).await; + live.run(10_000_000 / VIDEO_US, &mut [&mut leg], |_| false); assert_eq!( leg.export.dropped(), 0, @@ -1156,10 +1212,10 @@ async fn declared_legs_author_the_same_dts() { let mut live = Live::new(1.0, 0).await; live.reorder(); let mut a = live.join(Duration::ZERO).await; - live.run(join, &mut [&mut a], stall).await; + live.run(join, &mut [&mut a], stall); let mut b = live.join(Duration::from_millis(100)).await; - live.run(12 * GOP, &mut [&mut a, &mut b], stall).await; - live.finish(&mut [&mut a, &mut b]).await; + live.run(12 * GOP, &mut [&mut a, &mut b], stall); + live.finish(&mut [&mut a, &mut b]); let (a, b) = (decode_times(&a), decode_times(&b)); let shared: Vec<_> = b.iter().filter(|(pts, _)| a.contains_key(pts)).collect(); assert!( @@ -1178,12 +1234,11 @@ async fn declared_legs_author_the_same_dts() { async fn a_stitch_keeps_slots_continuous_across_different_track_leads() { for lag in [0, LONG_LAG, -LONG_LAG] { let mut old = Live::new(1.0, lag).await; - let mut new = Live::new(1.0, lag).await; + let mut new = old.beside(1.0, lag).await; let mut leg = old.join(Duration::ZERO).await; for tick in 0..20 * GOP { - old.run(tick + 1, &mut [&mut leg], |_| false).await; - new.run(tick + 1, &mut [], |tick| (JOIN + 1..5 * GOP).contains(&tick)) - .await; + old.run(tick + 1, &mut [&mut leg], |_| false); + new.run(tick + 1, &mut [], |tick| (JOIN + 1..5 * GOP).contains(&tick)); if tick == JOIN { let catalog = new.source.catalog::<()>(Default::default()).await.unwrap(); leg.export diff --git a/rs/moq-mux/src/container/ts/fixture.rs b/rs/moq-mux/src/container/ts/fixture.rs index 1d6831c0af..39003dad1f 100644 --- a/rs/moq-mux/src/container/ts/fixture.rs +++ b/rs/moq-mux/src/container/ts/fixture.rs @@ -55,18 +55,18 @@ impl Mux { /// One packet on `pid`, its payload filled with the PID's low byte. pub(super) fn packet(&mut self, pid: u16, flags: Flags) -> &mut Self { let cc = self.cc(pid); - let mut pkt = vec![0x47, (pid >> 8) as u8 & 0x1f, pid as u8]; + let mut pkt = [pid as u8; 188]; + pkt[..3].copy_from_slice(&[0x47, (pid >> 8) as u8 & 0x1f, pid as u8]); let marked = flags.pcr.is_some() || flags.random_access || flags.discontinuity; if marked { - pkt.push(0x30 | cc); - let mut field = vec![ - u8::from(flags.discontinuity) << 7 - | u8::from(flags.random_access) << 6 - | u8::from(flags.pcr.is_some()) << 4, - ]; + pkt[3] = 0x30 | cc; + pkt[5] = u8::from(flags.discontinuity) << 7 + | u8::from(flags.random_access) << 6 + | u8::from(flags.pcr.is_some()) << 4; + pkt[4] = 1; if let Some(pcr) = flags.pcr { let (base, ext) = (pcr / 300, pcr % 300); - field.extend_from_slice(&[ + pkt[6..12].copy_from_slice(&[ (base >> 25) as u8, (base >> 17) as u8, (base >> 9) as u8, @@ -74,13 +74,11 @@ impl Mux { ((base & 1) << 7) as u8 | 0x7e | (ext >> 8) as u8, ext as u8, ]); + pkt[4] = 7; } - pkt.push(field.len() as u8); - pkt.extend_from_slice(&field); } else { - pkt.push(0x10 | cc); + pkt[3] = 0x10 | cc; } - pkt.resize(188, pid as u8); self.out.extend_from_slice(&pkt); self } diff --git a/rs/moq-mux/src/container/ts/passthrough_test.rs b/rs/moq-mux/src/container/ts/passthrough_test.rs index e35c5d6227..c2170a35ec 100644 --- a/rs/moq-mux/src/container/ts/passthrough_test.rs +++ b/rs/moq-mux/src/container/ts/passthrough_test.rs @@ -36,10 +36,11 @@ struct Imported { impl Imported { /// Every object's packets, concatenated in order. fn bytes(&self) -> Vec { - self.objects - .iter() - .flat_map(|object| object.payload.iter().copied()) - .collect() + let mut bytes = Vec::new(); + for object in &self.objects { + bytes.extend_from_slice(&object.payload); + } + bytes } fn section(&self) -> &hang::catalog::M2ts { diff --git a/rs/moq-mux/src/container/ts/recovery_test.rs b/rs/moq-mux/src/container/ts/recovery_test.rs index 005c46443b..8da9d02fc4 100644 --- a/rs/moq-mux/src/container/ts/recovery_test.rs +++ b/rs/moq-mux/src/container/ts/recovery_test.rs @@ -137,7 +137,9 @@ impl Encode { match self.video { Video::H264(x264) => args.extend(["-c:v", "libx264", "-x264-params", x264, "-profile:v", "high"]), Video::H265(params) => { - x265 = format!("{params}:log-level=error"); + // x265 sizes its own thread pools by the host's cores, ignoring `-threads`, which makes + // the encode both costlier and different from one host to the next. + x265 = format!("{params}:pools=none:frame-threads=1:log-level=error"); args.extend(["-c:v", "libx265", "-x265-params", &x265]); } Video::Mpeg2 => args.extend([ @@ -700,19 +702,14 @@ impl Ledger { struct Run { ledgers: Vec, frames: Vec, + /// The frames' payloads, concatenated. + ts: Vec, stats: stats::Export, /// The stats after each output frame, and when, from the republisher's start. samples: Vec<(Duration, stats::Export)>, } impl Run { - fn ts(&self) -> Vec { - self.frames - .iter() - .flat_map(|frame| frame.payload.iter().copied()) - .collect() - } - fn ledger(&self, video: bool) -> &Ledger { self.ledgers.iter().find(|ledger| ledger.video == video).unwrap() } @@ -960,15 +957,20 @@ async fn run(recording: &Recording, arm: &Arm) -> Run { samples.push((start.elapsed(), export.stats())); } publisher.abort(); + let mut ts = Vec::new(); + for frame in &frames { + ts.extend_from_slice(&frame.payload); + } let mut run = Run { ledgers, frames, + ts, stats: export.stats(), samples, }; // Match each PES, and each ADTS frame of an audio PES, to the unit it carries. - let mut reader = PesPacketReader::new(TsPacketReader::new(std::io::Cursor::new(run.ts()))); + let mut reader = PesPacketReader::new(TsPacketReader::new(std::io::Cursor::new(&run.ts))); while let Some(pes) = reader.read_pes_packet().unwrap() { let video = pes.header.stream_id.is_video(); assert!(video || pes.header.stream_id.is_audio(), "only audio and video go out"); @@ -1196,7 +1198,8 @@ fn decoder(video_only: bool) -> Vec<&'static str> { if !video_only { args.extend(["-map", "0:a:0", "-c:a", "pcm_s16le"]); } - args.extend(["-f", "framemd5", "pipe:1"]); + // A fast hash: MD5 over every raw picture costs about as much as the decode. + args.extend(["-f", "framehash", "-hash", "murmur3", "pipe:1"]); args } @@ -1215,7 +1218,7 @@ fn parse(output: &[u8], video_only: bool) -> Result { } let fields: Vec<&str> = line.split(',').map(str::trim).collect(); let [index, _dts, pts, duration, _size, hash] = fields[..] else { - panic!("framemd5 line {line}"); + panic!("framehash line {line}"); }; match index { "0" => { @@ -1364,7 +1367,7 @@ impl Evidence { }; let dir = std::path::Path::new(&root).join("ts-recovery").join(name); std::fs::create_dir_all(&dir).unwrap(); - std::fs::write(dir.join("export.ts"), run.ts()).unwrap(); + std::fs::write(dir.join("export.ts"), &run.ts).unwrap(); std::fs::write(dir.join("ledgers.txt"), format!("{:#?}", run.ledgers)).unwrap(); std::fs::write(dir.join("stats.txt"), format!("{:#?}", run.stats)).unwrap(); Self(Some(dir)) @@ -1397,7 +1400,7 @@ async fn recover(arm: Arm) -> Run { run.stats.release ); assert!(assert_on_time(&run.frames) > 200, "too little output to judge"); - assert_continuous(&run.ts()); + assert_continuous(&run.ts); assert_eq!( count_discontinuity(&run.frames), profile.breaks(), @@ -1406,12 +1409,12 @@ async fn recover(arm: Arm) -> Run { let reference = Reference::new(profile, run.ledger(true)); let pids = (run.pid(Class::Video), run.pid(Class::Audio)); - for (i, (ts, emitted)) in segments(&run.ts(), run.ledger(true), pids).unwrap().iter().enumerate() { + for (i, (ts, emitted)) in segments(&run.ts, run.ledger(true), pids).unwrap().iter().enumerate() { let output = decode(ts, true).unwrap_or_else(|err| panic!("segment {i}: {err}")); (reference.compare(&output, emitted)).unwrap_or_else(|err| panic!("segment {i}: {err}")); } if profile != Profile::Open { - let output = decode(&run.ts(), false).unwrap(); + let output = decode(&run.ts, false).unwrap(); let video = run.ledger(true); let emitted: Vec<&Unit> = video.units.iter().filter(|unit| unit.output).collect(); reference.compare(&output, &emitted).unwrap(); @@ -1927,7 +1930,7 @@ async fn the_graders_fail_what_they_should() { assert!(err.contains("gaps reported"), "{err}"); // The output less the tables ahead of the keyframe it resumes at. - let ts = run.ts(); + let ts = &run.ts; let pid = run.pid(Class::Video); let pids = (pid, run.pid(Class::Audio)); let packets = ts.as_chunks::<188>().0; @@ -1941,7 +1944,7 @@ async fn the_graders_fail_what_they_should() { .filter(|&(at, packet)| !((tables..resume).contains(&at) && payload(packet).0 == 0)) .flat_map(|(_, packet)| packet.iter().copied()) .collect(); - segments(&ts, &run.ledgers[video], pids).unwrap(); + segments(ts, &run.ledgers[video], pids).unwrap(); let err = segments(&stripped, &run.ledgers[video], pids).unwrap_err(); assert!(err.contains("no PAT"), "{err}"); diff --git a/rs/moq-mux/src/jitter.rs b/rs/moq-mux/src/jitter.rs index 6abf2507b9..207655ec82 100644 --- a/rs/moq-mux/src/jitter.rs +++ b/rs/moq-mux/src/jitter.rs @@ -690,20 +690,28 @@ impl Steer { where K: 'a, { - let fits: Vec<(f64, f64)> = heard + let fits: Vec<_> = heard .filter_map(|key| self.tracks.get(key)) .filter(|series| 2 * series.len() >= self.phase.len()) .filter_map(|series| { let (slope, at) = envelope(series, now)?; - Some((depth(series, slope, at, now), slope)) + Some((series, slope, at)) }) .collect(); - let best = fits.iter().map(|&(depth, _)| depth).fold(f64::INFINITY, f64::min); - let steadiest = fits - .iter() - .filter(|&&(depth, _)| depth - best <= FIT) - .map(|&(_, slope)| slope) - .min_by(|a, b| a.abs().total_cmp(&b.abs())); + // A depth only chooses between tracks, and sorting for it is most of a step's work. + let steadiest = match fits[..] { + [(_, slope, _)] => Some(slope), + _ => { + let depths: Vec<(f64, f64)> = (fits.iter()) + .map(|&(series, slope, at)| (depth(series, slope, at, now), slope)) + .collect(); + let best = depths.iter().map(|&(depth, _)| depth).fold(f64::INFINITY, f64::min); + (depths.iter()) + .filter(|&&(depth, _)| depth - best <= FIT) + .map(|&(_, slope)| slope) + .min_by(|a, b| a.abs().total_cmp(&b.abs())) + } + }; let rate = match steadiest { Some(rate) => rate, None => envelope(&self.phase, now)?.0, @@ -1945,6 +1953,11 @@ mod tests { released(buffer).into_iter().map(|(item, _)| item).collect() } + /// Let go of every frame due by now. + fn drain(buffer: &mut Buffer, waiter: &kio::Waiter) { + while buffer.poll_next(waiter).is_ready() {} + } + /// A fill reads each track's available backlog before acquisition selects a clock. #[tokio::test(start_paused = true)] async fn acquisition_observes_every_tracks_available_frames() { @@ -2507,6 +2520,7 @@ mod tests { }; let mut last: Option<(f64, Decode)> = None; let mut stepped = false; + let waiter = kio::Waiter::noop(); for k in 0..(hours * 3_600.0 * 2.0) as u64 { let decode = Duration::from_millis(k * 500); let hour = decode.as_secs_f64() / 3_600.0; @@ -2547,7 +2561,7 @@ mod tests { let slack = since(arrived, deadline) as f64 / 1e9; run.worst = run.worst.max((slack - delay.as_secs_f64()).abs()); } - due(&mut buffer); + drain(&mut buffer, &waiter); let Some(clock) = buffer.clock else { continue; @@ -2703,6 +2717,7 @@ mod tests { out_of_tolerance: 0, rates: Vec::new(), }; + let waiter = kio::Waiter::noop(); for (sent, track, decode) in frames { let arrived = start + Duration::from_secs_f64(sent); clock.advance(arrived.max(clock.now())); @@ -2726,7 +2741,7 @@ mod tests { if let Some(rate) = buffer.drift() { fed.rates.push((decode.as_secs_f64(), rate)); } - due(&mut buffer); + drain(&mut buffer, &waiter); } fed.dropped = buffer.dropped(); fed.drift = buffer.drift(); diff --git a/rs/moq-mux/src/rate.rs b/rs/moq-mux/src/rate.rs index ddccff5fca..ea5abe49a6 100644 --- a/rs/moq-mux/src/rate.rs +++ b/rs/moq-mux/src/rate.rs @@ -1348,7 +1348,7 @@ mod tests { /// invariants after every wake. mod model { use proptest::prelude::*; - use proptest::test_runner::{Config, RngSeed, TestRunner}; + use proptest::test_runner::{Config, RngAlgorithm, RngSeed, TestRunner}; use super::*; @@ -1513,6 +1513,8 @@ mod tests { let mut runner = TestRunner::new(Config { cases: 512, rng_seed: RngSeed::Fixed(6781), + // Every op forks the runner's RNG, which ChaCha makes the bulk of a debug run. + rng_algorithm: RngAlgorithm::XorShift, failure_persistence: None, ..Default::default() }); diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 7dc7a00f5d..5651943802 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -6117,7 +6117,7 @@ mod group_priority_test { async fn a_busy_track_yields_to_its_siblings() { /// `TrackServe`'s passes per poll; each starts at most one group. const BUDGET: usize = 32; - const GROUPS: u64 = 10_000; + const GROUPS: u64 = 2_000; const BATCH: u64 = 1_000; let session = SinkSession::new(crate::lite::test_transport::Log::default()).with_unacked_fin(); diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index ebc033bc06..cff982375a 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -8542,7 +8542,7 @@ mod tests { async fn a_busy_subscription_yields_to_its_siblings() { /// `RequestServe`'s passes per poll; each serves at most one group. const BUDGET: u64 = 32; - const GROUPS: u64 = 10_000; + const GROUPS: u64 = 2_000; const BATCH: u64 = 1_000; type Task = Box Poll<()>>; diff --git a/rs/moq-net/src/model/front.rs b/rs/moq-net/src/model/front.rs index f87ebf6f84..797d9742ad 100644 --- a/rs/moq-net/src/model/front.rs +++ b/rs/moq-net/src/model/front.rs @@ -1695,6 +1695,8 @@ mod tests { ]; fn walk(front: &Front, alphabet: &[Event], depth: usize, sequences: &mut usize) { + // Rendered only when an event needs it: formatting dominates the walk. + let mut before = None; for event in alphabet { // Ids are minted once per resolution, so a source can never resolve // while it is already serving; the alphabet reuses ids for brevity. @@ -1704,7 +1706,6 @@ mod tests { continue; } let mut next = front.clone(); - let before = format!("{next:?}"); let actions = next.step(event.clone()); *sequences += 1; @@ -1713,11 +1714,32 @@ mod tests { // An event that changes nothing produces nothing, except letting // go of a source the front never took. - if format!("{next:?}") == before { - assert!( - actions.iter().all(|action| matches!(action, Action::Detach { .. })), - "no-op event {event:?} produced {actions:?}" - ); + if !actions.iter().all(|action| matches!(action, Action::Detach { .. })) { + // The cheap fields settle most events; the rest compare as rendered. + let flags = |f: &Front| { + ( + f.serving, + f.serving_closing, + f.upstream, + f.resolved, + f.armed, + f.held, + f.retiring, + f.ended, + ) + }; + let moved = flags(&next) != flags(front) + || next.parked != front.parked + || next.excluded != front.excluded + || next.tracks.len() != front.tracks.len(); + if !moved { + let before = before.get_or_insert_with(|| format!("{front:?}")); + assert_ne!( + &format!("{next:?}"), + before, + "no-op event {event:?} produced {actions:?}" + ); + } } // Nothing after the end. if front.ended() { diff --git a/rs/moq-net/tests/integration/routing_contracts/handoff.rs b/rs/moq-net/tests/integration/routing_contracts/handoff.rs index 1cd9f592da..a84af6c628 100644 --- a/rs/moq-net/tests/integration/routing_contracts/handoff.rs +++ b/rs/moq-net/tests/integration/routing_contracts/handoff.rs @@ -1420,7 +1420,7 @@ fn churn(rng: &mut XorShift, freshness: Freshness) -> Net { if freshness == FLOORS { net.check().unwrap(); } - loop_free(&net); + loop_free(&net, ids); } for &(a, b) in &links { @@ -1436,8 +1436,9 @@ fn churn(rng: &mut XorShift, freshness: Freshness) -> Net { /// A node drops a feasibility distance only once every neighbor echoed its /// ROUTE_END, so an older advertisement of its own never returns as a loop. -fn loop_free(net: &Net) { - for id in (100..110).chain(200..210) { +/// `ids` are the origins' current identities; none past them exists yet. +fn loop_free(net: &Net, ids: [Id; 2]) { + for id in (100..110.min(ids[0] + 1)).chain(200..210.min(ids[1] + 1)) { for node in 0..RELAYS + 3 { let mut at = node; for hop in 0.. { diff --git a/rs/moq-net/tests/integration/routing_contracts/model.rs b/rs/moq-net/tests/integration/routing_contracts/model.rs index c297f1c2c5..12ce354d60 100644 --- a/rs/moq-net/tests/integration/routing_contracts/model.rs +++ b/rs/moq-net/tests/integration/routing_contracts/model.rs @@ -680,7 +680,11 @@ struct Front { pub struct Net { cfg: Config, nodes: Vec, - links: BTreeMap<(Node, Node), Link>, + /// `links[a][b]`: `a`'s side of its link to `b`. Dense rather than a map, + /// since selection looks a side up per route and the sweeps select constantly. + links: Vec>>, + /// Every `(subject, scope)` a configured link grants. + grants: BTreeSet<(&'static str, &'static str)>, up: BTreeSet<(Node, Node)>, /// `(node, peer)`: the identity `node` proved on its session to `peer`. A /// session proves one key, once. @@ -710,7 +714,8 @@ impl Net { ..State::default() }) .collect(), - links: BTreeMap::new(), + links: vec![vec![None; nodes]; nodes], + grants: BTreeSet::new(), up: BTreeSet::new(), proved: BTreeMap::new(), draining: BTreeSet::new(), @@ -737,8 +742,16 @@ impl Net { rule.admits(ab.cost) && rule.admits(ba.cost), "zero-cost link {a}-{b} refused" ); - self.links.insert((a, b), ab); - self.links.insert((b, a), ba); + self.links[a][b] = Some(ab); + self.links[b][a] = Some(ba); + self.grants = self + .links + .iter() + .flatten() + .flatten() + .filter_map(|link| link.grant) + .map(|grant| (grant.subject, grant.scope)) + .collect(); self.connect(a, b); } @@ -747,13 +760,18 @@ impl Net { self.link(a, b, Link::cost(cost), Link::cost(cost)); } + /// `node`'s side of its configured link to `peer`. + fn side(&self, node: Node, peer: Node) -> &Link { + self.links[node][peer].as_ref().expect("a configured link") + } + fn pair(a: Node, b: Node) -> (Node, Node) { (a.min(b), a.max(b)) } /// Open a session; each side replays what it currently advertises. pub fn connect(&mut self, a: Node, b: Node) { - assert!(self.links.contains_key(&(a, b)), "unknown link {a}-{b}"); + assert!(self.links[a][b].is_some(), "unknown link {a}-{b}"); self.up.insert(Self::pair(a, b)); for (x, y) in [(a, b), (b, a)] { if let Some(id) = self.nodes[x].proven { @@ -817,18 +835,17 @@ impl Net { /// Re-open every configured link of `node`. pub fn reconnect(&mut self, node: Node) { - let peers: Vec = self.links.keys().filter(|(a, _)| *a == node).map(|(_, b)| *b).collect(); + let peers: Vec = (0..self.nodes.len()) + .filter(|&b| self.links[node][b].is_some()) + .collect(); for peer in peers { self.connect(node, peer); } } + /// Every neighbor `node` has a session up with, in ascending order. fn peers(&self, node: Node) -> impl Iterator + '_ { - self.up.iter().filter_map(move |&(a, b)| match node { - n if n == a => Some(b), - n if n == b => Some(a), - _ => None, - }) + (0..self.nodes.len()).filter(move |&peer| self.links[node][peer].is_some() && self.connected(node, peer)) } /// `node` starts originating `id` at seqno `seqno`, proving it on every session. @@ -1169,7 +1186,7 @@ impl Net { /// The key `at` files `id` under when `from` names it, or `None` when refused. fn key(&self, at: Node, from: Node, id: Id) -> Option { - match self.links[&(at, from)].grant { + match self.side(at, from).grant { Some(grant) if self.proved.get(&(from, at)) != Some(&id) => { if id & QUALIFIED != 0 && !self.cfg.requalify { return None; @@ -1225,7 +1242,7 @@ impl Net { let Some(id) = self.key(at, from, id) else { return; }; - let scope = match self.links[&(at, from)].grant { + let scope = match self.side(at, from).grant { Some(grant) if id & QUALIFIED != 0 => Some(grant.scope), _ => None, }; @@ -1372,46 +1389,61 @@ impl Net { down_only: false, }); } - state - .rx - .get(&id)? - .iter() - .filter(|(next, entry)| !entry.held && !except.contains(next) && self.feasible(node, id, entry.advert)) - .filter_map(|(&next, entry)| { - let advert = entry.advert; - let link = self.links[&(node, next)]; - let metric = self.cfg.rule.add(advert.metric, link.cost)?; - let selected = Selected { - next: Some(next), - seqno: advert.seqno, - metric, - advertised: advert.metric, - down_only: self.cfg.export == Export::DownOnly && (link.upstream || advert.down_only), - }; - let draining = self.draining.contains(&(node, next)); - Some(((draining, Reverse(link.priority), metric, next), selected)) - }) - .min_by_key(|(rank, _)| *rank) - .map(|(_, selected)| selected) + // A plain loop rather than an iterator chain: selection dominates the + // sweeps, and unoptimized test builds pay for every adapter. + type Rank = (bool, Reverse, Metric, Node); + let mut best: Option<(Rank, Selected)> = None; + for (&next, entry) in state.rx.get(&id)? { + if entry.held || except.contains(&next) || !self.feasible(node, id, entry.advert) { + continue; + } + let advert = entry.advert; + let link = self.side(node, next); + let Some(metric) = self.cfg.rule.add(advert.metric, link.cost) else { + continue; + }; + let rank = ( + self.draining.contains(&(node, next)), + Reverse(link.priority), + metric, + next, + ); + if best.as_ref().is_some_and(|(best, _)| *best <= rank) { + continue; + } + let selected = Selected { + next: Some(next), + seqno: advert.seqno, + metric, + advertised: advert.metric, + down_only: self.cfg.export == Export::DownOnly && (link.upstream || advert.down_only), + }; + best = Some((rank, selected)); + } + best.map(|(_, selected)| selected) } /// The usable but infeasible routes `node` holds for `id`, best first. fn infeasible(&self, node: Node, id: Id) -> Vec { - let Some(routes) = self.nodes[node].rx.get(&id) else { - return Vec::new(); - }; - let mut out: Vec<(Metric, Node)> = routes - .iter() - .filter(|(_, entry)| !entry.held && !self.feasible(node, id, entry.advert)) - .filter_map(|(&next, entry)| { - let metric = self.cfg.rule.add(entry.advert.metric, self.links[&(node, next)].cost)?; - Some((metric, next)) - }) - .collect(); + let mut out: Vec<(Metric, Node)> = self.infeasible_unordered(node, id).collect(); out.sort(); out.into_iter().map(|(_, next)| next).collect() } + /// [`Net::infeasible`] with each route's metric, in neighbor order. + fn infeasible_unordered(&self, node: Node, id: Id) -> impl Iterator + '_ { + self.nodes[node] + .rx + .get(&id) + .into_iter() + .flatten() + .filter(move |(_, entry)| !entry.held && !self.feasible(node, id, entry.advert)) + .filter_map(move |(&next, entry)| { + let metric = self.cfg.rule.add(entry.advert.metric, self.side(node, next).cost)?; + Some((metric, next)) + }) + } + /// Where `node` sends a request for `id` on behalf of `from`: a feasible /// route's next hop if one avoids them, else an infeasible route's (RFC 8966 /// s3.8.1.2). @@ -1437,31 +1469,38 @@ impl Net { /// Whether `node` may follow only exportable routes for a request from `from`. fn exports_only(&self, node: Node, from: Option) -> bool { - self.cfg.recheck && from.is_some_and(|from| self.links[&(node, from)].upstream) + self.cfg.recheck && from.is_some_and(|from| self.side(node, from).upstream) } /// Whether `node` holds a reachable route to `id` but none is feasible: it /// needs the origin to advance its seqno. pub fn starved(&self, node: Node, id: Id) -> bool { - self.route(node, id).is_none() && !self.infeasible(node, id).is_empty() + self.route(node, id).is_none() && self.infeasible_unordered(node, id).next().is_some() } /// Whether `node`, with no usable route to `id`, holds a held route it could - /// recover through. A held route counts only if its last finite advertisement - /// is feasible here, so holds follow a strictly decreasing metric toward a - /// starved node and cannot hold each other up. + /// recover through. pub fn grounded(&self, node: Node, id: Id) -> bool { - self.route(node, id).is_none() - && self.nodes[node].rx.get(&id).is_some_and(|routes| { - routes - .values() - .any(|entry| entry.held && self.feasible(node, id, entry.advert)) - }) + self.route(node, id).is_none() && self.recoverable(node, id) + } + + /// Whether `node` holds a held route to `id`. One counts only if its last + /// finite advertisement is feasible here, so holds follow a strictly + /// decreasing metric toward a starved node and cannot hold each other up. + fn recoverable(&self, node: Node, id: Id) -> bool { + self.nodes[node].rx.get(&id).is_some_and(|routes| { + routes + .values() + .any(|entry| entry.held && self.feasible(node, id, entry.advert)) + }) } - /// Whether `node` keeps showing `id`'s announcements. - pub fn live(&self, node: Node, id: Id) -> bool { - self.route(node, id).is_some() || self.starved(node, id) || self.grounded(node, id) + /// Whether `node`, having selected `route` for `id`, keeps showing `id`'s + /// announcements: it is routed, [starved](Self::starved), or + /// [grounded](Self::grounded). The caller passes the route it already + /// selected, since selecting dominates the sweeps. + fn live(&self, node: Node, id: Id, route: Option) -> bool { + route.is_some() || self.infeasible_unordered(node, id).next().is_some() || self.recoverable(node, id) } /// The neighbors whose views can certify `id` at `node`: those whose last @@ -1478,18 +1517,21 @@ impl Net { }) } - /// The regions of `id` that `node`'s certifiers certify, or `None` while it - /// holds no route or a certifier has not yet answered the current interest: - /// its announcements may still trail its route. - fn certified(&self, node: Node, id: Id) -> Option> { + /// The regions of `id` that `node`'s certifiers certify, those whose route + /// is held only when `held`, or `None` while it holds no route or a + /// certifier has not yet answered the current interest: its announcements + /// may still trail its route. + fn certified(&self, node: Node, id: Id, held: bool) -> Option> { let state = &self.nodes[node]; - state.rx.get(&id)?; + let routes = state.rx.get(&id)?; let mut regions = Vec::new(); for next in self.certifiers(node, id) { if state.heard_on.get(&(next, id)) != Some(&state.interest) { return None; } - regions.extend(state.heard[&(next, id)].floors.keys()); + if held || !routes[&next].held { + regions.extend(state.heard[&(next, id)].floors.keys()); + } } Some(regions) } @@ -1506,7 +1548,7 @@ impl Net { if self.nodes[node].origins.contains_key(&id) || self.route(node, id).is_none() { return; } - let Some(regions) = self.certified(node, id) else { + let Some(regions) = self.certified(node, id, true) else { return; }; let state = &self.nodes[node]; @@ -1544,19 +1586,18 @@ impl Net { forget(&mut state.forgot, id, &remembered); } - /// The view `node` shows and forwards for `id`. - fn adopted(&self, node: Node, id: Id) -> Option<&View> { + /// The view `node` shows and forwards for `id`, having selected `route` for it. + fn adopted(&self, node: Node, id: Id, route: Option) -> Option<&View> { let state = &self.nodes[node]; if state.origins.contains_key(&id) { return state.views.get(&id); } match self.cfg.freshness { - Freshness::NextHop => self - .route(node, id) + Freshness::NextHop => route .and_then(|route| route.next) .and_then(|next| state.heard.get(&(next, id))), Freshness::Barrier | Freshness::Floors { .. } => { - self.live(node, id).then(|| state.views.get(&id)).flatten() + self.live(node, id, route).then(|| state.views.get(&id)).flatten() } } } @@ -1607,18 +1648,29 @@ impl Net { if let Some(selected) = selected && peers .iter() - .any(|&peer| !(self.links[&(node, peer)].upstream && selected.down_only)) + .any(|&peer| !(self.side(node, peer).upstream && selected.down_only)) { self.record(node, id, selected); } - let live = serves && self.live(node, id); - let view = self.adopted(node, id).cloned().unwrap_or_default(); + // Recording may have tightened the feasibility distance: select again. + let route = self.route(node, id); + let live = serves && self.live(node, id, route); + let adopted = self.adopted(node, id, route); // A sender that certifies anything answers every stream, even one // whose interest none of its regions meets, so the receiver knows it // certifies nothing there. - let answers = !view.floors.is_empty(); + let answers = adopted.is_some_and(|view| !view.floors.is_empty()); + // Each floor names its region: the stream's interest, or a scope + // below it. Peers mostly share an interest, so restrict once per interest. + let mut restricted: Vec<(&'static str, View)> = Vec::new(); + for &peer in &peers { + let watch = self.nodes[peer].watch; + if !restricted.iter().any(|(interest, _)| *interest == watch) { + restricted.push((watch, adopted.map_or_else(View::default, |view| view.restrict(watch)))); + } + } for &peer in &peers { - let upstream = self.links[&(node, peer)].upstream; + let upstream = self.side(node, peer).upstream; let state = &self.nodes[node]; let reach = match selected { Some(route) if !(upstream && route.down_only) => Some(Reach::Finite(Advert { @@ -1650,12 +1702,11 @@ impl Net { state.sent_views.remove(&(peer, id)); continue; } - // Each floor names its region: the stream's interest, or a scope below it. let watch = self.nodes[peer].watch; - let view = view.restrict(watch); + let (_, view) = restricted.iter().find(|(interest, _)| *interest == watch).unwrap(); let state = &mut self.nodes[node]; let unchanged = match state.sent_views.get(&(peer, id)) { - Some(sent) => *sent == view, + Some(sent) => sent == view, None => !answers, }; if unchanged { @@ -1668,7 +1719,7 @@ impl Net { self.stats.claim_bytes += bytes; self.queues.entry((node, peer)).or_default().push_back(Msg::Claims { id, - view, + view: view.clone(), interest: self.nodes[peer].interest, }); } @@ -1677,15 +1728,12 @@ impl Net { // identity and every neighbor echoed its ROUTE_END: no node then holds an // advertisement from this one, so no older one can return as a loop. let state = &mut self.nodes[node]; - let busy: BTreeSet = state - .rx - .keys() - .chain(state.origins.keys()) - .chain(state.sent.keys().map(|(_, id)| id)) - .chain(state.unacked.keys().map(|(_, id)| id)) - .copied() - .collect(); - state.fd.retain(|id, _| busy.contains(id)); + state.fd.retain(|id, _| { + state.rx.contains_key(id) + || state.origins.contains_key(id) + || state.sent.keys().any(|(_, sent)| sent == id) + || state.unacked.keys().any(|(_, unacked)| unacked == id) + }); self.requests(node); self.show(node); } @@ -1695,16 +1743,20 @@ impl Net { if self.cfg.requests == Requests::Off { return; } + // Every pending or starved identity; one neither pending nor starved + // falls through untouched below. let state = &self.nodes[node]; let mut ids: BTreeSet = state.pending.keys().copied().collect(); - ids.extend(state.rx.keys().filter(|id| self.starved(node, **id))); + ids.extend(state.rx.keys()); for id in ids { let state = &self.nodes[node]; if state.origins.contains_key(&id) { self.nodes[node].pending.remove(&id); continue; } - let starved = self.starved(node, id); + // Nothing below changes a selection, so one serves the whole pass. + let route = self.route(node, id); + let starved = route.is_none() && self.infeasible_unordered(node, id).next().is_some(); if starved { let want = state.fd[&id].0 + 1; match self.cfg.requests { @@ -1739,7 +1791,11 @@ impl Net { let Some(pending) = self.nodes[node].pending.get(&id).cloned() else { continue; }; - let answered = self.route(node, id).is_some_and(|route| route.seqno >= pending.seqno); + let answered = route.is_some_and(|route| route.seqno >= pending.seqno); + if answered || (pending.from.is_empty() && !starved) { + self.nodes[node].pending.remove(&id); + continue; + } // Hop by hop, a request follows the next hop even back through a // requester: next hops form no loop, one pending request per hop stops // a bounce, and no requester ever cancels, so excluding one can strand @@ -1748,17 +1804,15 @@ impl Net { Requests::HopByHop => BTreeSet::new(), _ => pending.from.clone(), }; - let target = self.target(node, id, &except); - if answered || target.is_none() || (pending.from.is_empty() && !starved) { + let Some(target) = self.target(node, id, &except) else { self.nodes[node].pending.remove(&id); continue; - } + }; let resend = match self.cfg.requests { - Requests::HopByHop => target != pending.asked, + Requests::HopByHop => Some(target) != pending.asked, _ => pending.asked.is_none(), }; if resend { - let target = target.unwrap(); self.nodes[node].pending.get_mut(&id).unwrap().asked = Some(target); self.send_request(node, target, id, pending.seqno); } @@ -1772,20 +1826,18 @@ impl Net { ids.extend(state.heard.keys().map(|(_, id)| *id)); let mut shown = BTreeMap::new(); for id in ids { - let Some(view) = self.adopted(node, id) else { + let route = self.route(node, id); + let Some(view) = self.adopted(node, id, route) else { continue; }; - let usable = self.route(node, id).is_some(); + let usable = route.is_some(); // A claim is available only through a certifier that is not held: - // another scope's usable route cannot serve it. + // another scope's usable route cannot serve it. Only claims behind a + // usable route need the certifiers. let serving: Option> = match self.cfg.freshness { - Freshness::Floors { .. } if !state.origins.contains_key(&id) => self.certified(node, id).map(|_| { - self.certifiers(node, id) - .into_iter() - .filter(|next| !state.rx[&id][next].held) - .flat_map(|next| state.heard[&(next, id)].floors.keys().copied()) - .collect() - }), + Freshness::Floors { .. } if usable && !view.items.is_empty() && !state.origins.contains_key(&id) => { + self.certified(node, id, false) + } _ => None, }; let current = state.current.get(&id); @@ -1827,11 +1879,11 @@ impl Net { // A claim no routed neighbor's region covers was let go, not ended. let uncertified = matches!(self.cfg.freshness, Freshness::Floors { .. }) && self - .certified(node, id) + .certified(node, id, true) .is_some_and(|regions| !regions.iter().any(|region| overlaps(claim, region))); let kind = if !overlaps(claim, self.nodes[node].watch) { Kind::Unwatched - } else if self.live(node, id) && !uncertified { + } else if self.live(node, id, self.route(node, id)) && !uncertified { Kind::End } else { Kind::Offline @@ -1940,12 +1992,7 @@ impl Net { if key & QUALIFIED == 0 { return None; } - let subjects: BTreeSet<&str> = self - .links - .values() - .filter_map(|link| link.grant) - .map(|grant| grant.subject) - .collect(); + let subjects: BTreeSet<&str> = self.grants.iter().map(|(subject, _)| *subject).collect(); let known: BTreeSet = self .nodes .iter() @@ -1983,19 +2030,13 @@ impl Net { } fn candidates(&self, node: Node, path: &str, from: Option) -> Vec { - let grants: BTreeSet<(&str, &str)> = self - .links - .values() - .filter_map(|link| link.grant) - .map(|grant| (grant.subject, grant.scope)) - .collect(); let mut out = Vec::new(); for (index, claim) in self.claims.iter().enumerate() { if !claim.covers(path) || self.hidden.contains(&(node, index)) { continue; } // A claim counts under its bare identity, and under a qualified key only inside that grant's scope. - let inside = grants.iter().filter(|(_, scope)| within(claim.prefix, scope)); + let inside = self.grants.iter().filter(|(_, scope)| within(claim.prefix, scope)); for id in iter::once(claim.id).chain(inside.map(|(subject, _)| qualified(subject, claim.id))) { let Some(route) = self.route_from(node, id, from) else { continue; @@ -2010,7 +2051,7 @@ impl Net { total, advertised: route.advertised, draining: route.next.is_some_and(|next| self.draining.contains(&(node, next))), - priority: route.next.map_or(0, |next| self.links[&(node, next)].priority), + priority: route.next.map_or(0, |next| self.side(node, next).priority), prefix: claim.prefix.len(), epoch: claim.epoch, down_only: route.down_only, diff --git a/rs/moq-pattern/tests/integration/pattern_exhaustive.rs b/rs/moq-pattern/tests/integration/pattern_exhaustive.rs index 27a7e4c187..cce96f37fe 100644 --- a/rs/moq-pattern/tests/integration/pattern_exhaustive.rs +++ b/rs/moq-pattern/tests/integration/pattern_exhaustive.rs @@ -148,8 +148,12 @@ fn intersect_matches_exactly_what_both_match() { for (i, a) in patterns.iter().enumerate() { for (j, b) in patterns.iter().enumerate() { let both = a.intersect(b).unwrap(); - for (k, path) in paths.iter().enumerate() { - assert_eq!(both.matches(path), table[i][k] && table[j][k], "{a} & {b} on {path:?}"); + // Below the diagonal, the commute check makes this the union `b & a` already + // matched against every path. + if i <= j { + for (k, path) in paths.iter().enumerate() { + assert_eq!(both.matches(path), table[i][k] && table[j][k], "{a} & {b} on {path:?}"); + } } assert_eq!(both, b.intersect(a).unwrap(), "{a} & {b} commutes"); assert_eq!(both.is_empty(), !a.overlaps(b), "{a} & {b} empty iff disjoint"); diff --git a/rs/moq-quic/src/tests/probe.rs b/rs/moq-quic/src/tests/probe.rs index 3937ef7cbe..3cc267dd00 100644 --- a/rs/moq-quic/src/tests/probe.rs +++ b/rs/moq-quic/src/tests/probe.rs @@ -243,9 +243,9 @@ struct Reading { } fn run(scenario: &Scenario) -> Outcome { - // Each run's packet trace is about 30 MB and takes most of its time; what it measured is - // reported once it returns - let _guard = subscribe_at(tracing::Level::INFO); + // Each run's packet trace is about 30 MB, so `RUST_LOG=trace` is slow here; what it measured + // is reported once it returns + let _guard = subscribe(); entropy::reseed(scenario.seed); let mut server_transport = transport::Config::default(); if let Some(window) = scenario.receive_window { diff --git a/rs/moq-quic/src/tests/sim.rs b/rs/moq-quic/src/tests/sim.rs index d2abb4650a..558d6076bc 100644 --- a/rs/moq-quic/src/tests/sim.rs +++ b/rs/moq-quic/src/tests/sim.rs @@ -266,9 +266,8 @@ impl Peer { }; let mut stream = conn.send_stream(id); while sending.written < sending.spec.len { - let end = sending.spec.len.min(sending.written + 65_536); - let data = payload(index, sending.written..end); - match stream.write(&mut &data[..]) { + let end = sending.spec.len.min(sending.written + PAYLOAD_MAX); + match stream.write(&mut payload(index, sending.written..end)) { Ok(written) => sending.written += written, Err(send::Error::Blocked) => break, Err(error) => panic!("writing stream {index}: {error}"), @@ -318,7 +317,7 @@ impl Peer { let end = receiving.read + chunk.bytes.len(); prop_assert!(end <= len, "stream {index} ran past its {len} bytes"); prop_assert!( - chunk.bytes == payload(index, receiving.read..end), + is_payload(index, receiving.read, &chunk.bytes), "stream {index} corrupted at {}", receiving.read ); @@ -373,9 +372,27 @@ impl Peer { } } +/// The longest range `payload` serves +const PAYLOAD_MAX: usize = 65_536; + /// The bytes at `range` of stream `index`, distinct per stream and offset -fn payload(index: usize, range: ops::Range) -> Vec { - range.map(|i| (i % 251) as u8 ^ (index as u8)).collect() +/// +/// Each stream reads one cycle of prime length from its own rotation, so streams fewer than 251 +/// apart differ at every offset. Slicing a shared buffer keeps the per-step writes and checks from +/// dominating the simulation. +fn payload(index: usize, range: ops::Range) -> &'static [u8] { + static CYCLE: std::sync::LazyLock> = + std::sync::LazyLock::new(|| (0..251 + PAYLOAD_MAX).map(|i| (i % 251) as u8).collect()); + let start = (index + range.start) % 251; + &CYCLE[start..][..range.len()] +} + +/// Whether `bytes` are stream `index`'s payload from `offset`, however long the chunk +fn is_payload(index: usize, offset: usize, bytes: &[u8]) -> bool { + bytes.chunks(PAYLOAD_MAX).enumerate().all(|(i, piece)| { + let start = offset + i * PAYLOAD_MAX; + piece == payload(index, start..start + piece.len()) + }) } fn run(case: &Case) -> Result { diff --git a/rs/moq-quic/src/tests/sim/workloads.rs b/rs/moq-quic/src/tests/sim/workloads.rs index d2eb726745..cd67f3a7bd 100644 --- a/rs/moq-quic/src/tests/sim/workloads.rs +++ b/rs/moq-quic/src/tests/sim/workloads.rs @@ -78,7 +78,7 @@ impl Transfer { _ => self.len, }; while self.written < target { - match stream.write(&mut &payload(key, self.written..target)[..]) { + match stream.write(&mut payload(key, self.written..target)) { Ok(written) => self.written += written, Err(send::Error::Blocked) => return Ok(()), Err(send::Error::Stopped(code)) if matches!(self.end, End::Stop) => { @@ -309,7 +309,7 @@ fn conn(pair: &mut Pair, handles: [ConnectionHandle; 2], side: usize) -> &mut Co } fn datagram(side: usize, index: usize) -> Vec { - let mut data = payload(side, 0..256); + let mut data = payload(side, 0..256).to_vec(); data[..4].copy_from_slice(&(index as u32).to_le_bytes()); data } diff --git a/rs/moq-quic/src/tests/util.rs b/rs/moq-quic/src/tests/util.rs index 1a94755b49..56cb672f9a 100644 --- a/rs/moq-quic/src/tests/util.rs +++ b/rs/moq-quic/src/tests/util.rs @@ -17,7 +17,7 @@ use rustls::{ client::WebPkiServerVerifier, pki_types::{CertificateDer, PrivateKeyDer}, }; -use tracing::{info_span, trace}; +use tracing::{debug_span, trace}; use super::crypto::rustls::{QuicClientConfig, QuicServerConfig, configured_provider}; use super::*; @@ -160,7 +160,7 @@ impl Pair { } pub(super) fn drive_client(&mut self) { - let span = info_span!("client"); + let span = debug_span!("client"); let _guard = span.enter(); self.client.drive(self.time, self.server.addr); for (packet, buffer) in self.client.outbound.drain(..) { @@ -192,7 +192,7 @@ impl Pair { } pub(super) fn drive_server(&mut self) { - let span = info_span!("server"); + let span = debug_span!("server"); let _guard = span.enter(); self.server.drive(self.time, self.client.addr); for (packet, buffer) in self.server.outbound.drain(..) { @@ -233,7 +233,7 @@ impl Pair { /// Just start connecting the client pub(super) fn begin_connect(&mut self, config: client::Config) -> ConnectionHandle { - let span = info_span!("client"); + let span = debug_span!("client"); let _guard = span.enter(); let (client_ch, client_conn) = self .client @@ -621,14 +621,18 @@ impl ::std::ops::DerefMut for TestEndpoint { } } +/// Logs the test's own events, and the stack's per-packet trace when `RUST_LOG` asks for it. +/// +/// Formatting every trace event costs more than the simulation it describes, and the runs are +/// deterministic, so a failure replays with `RUST_LOG=trace` to read it. The per-step side spans +/// are debug for the same reason: they only label the stack's own events. pub(super) fn subscribe() -> tracing::subscriber::DefaultGuard { - subscribe_at(tracing::Level::TRACE) -} - -/// [`subscribe`] at `level`, for a test whose trace would be too long to read -pub(super) fn subscribe_at(level: tracing::Level) -> tracing::subscriber::DefaultGuard { + let filter = tracing_subscriber::EnvFilter::builder() + .with_default_directive(tracing::Level::INFO.into()) + .from_env() + .expect("RUST_LOG"); let builder = tracing_subscriber::FmtSubscriber::builder() - .with_max_level(level) + .with_env_filter(filter) .with_writer(|| TestWriter); tracing::subscriber::set_default(builder.finish()) } diff --git a/rs/moq-relay/tests/integration/cluster_unknown.rs b/rs/moq-relay/tests/integration/cluster_unknown.rs index 97a27d7ee4..b4bca9e9e1 100644 --- a/rs/moq-relay/tests/integration/cluster_unknown.rs +++ b/rs/moq-relay/tests/integration/cluster_unknown.rs @@ -54,11 +54,13 @@ async fn spawn_relay(id: u64, connect: Vec, cluster_version: Option = (0..100).map(|seed| payload(1000, seed)).collect(); for _ in 0..2000 { - for seed in 0..100 { + for datagram in &datagrams { sender - .send_datagram(payload(1000, seed), SendOptions::default()) + .send_datagram(datagram.clone(), SendOptions::default()) .expect("the sender's keep-alive kept it open"); } tokio::time::advance(Duration::from_millis(1)).await;