diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml index 3c1b2e07f1..1a4d3d2aaf 100644 --- a/.github/workflows/nightly.yml +++ b/.github/workflows/nightly.yml @@ -215,6 +215,10 @@ jobs: if: ${{ !cancelled() }} run: nix develop --command just rs test -p moq-mux benchmark_damage_pid_fanout --run-ignored only --success-output immediate + - name: Container export benchmark smoke + if: ${{ !cancelled() }} + run: nix develop --command just rs bench-export --test + - name: RTMP chunk-stream benchmark smoke if: ${{ !cancelled() }} run: nix develop --command just rs bench-rtmp --test diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 50c3cf8906..2c5655ee58 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -306,18 +306,24 @@ See [Authentication](/bin/relay/auth). ## Retention and latency `import --max-age` (default 30 s) tells relays how long to keep old groups -fetchable, which the [HLS gateway](/bin/hls) depends on. `export --max-delay` -(default 500 ms) is how far a stalled group may fall behind the live edge -before *this* consumer skips it. Raising the first never delays playback. +fetchable, which the [HLS gateway](/bin/hls) depends on. `export fmp4`, `export h264`, +and `export h265` take `--max-delay` (default 500 ms): how far a stalled group +may fall behind the live edge before *this* consumer skips it. Raising +`--max-age` never delays playback. At stdin EOF, `import` keeps the finished broadcast up for its max-age, so a subscriber still catching up can drain it; Ctrl-C ends that early. -`export ts` takes `--delay` (default 500 ms) instead, like an SRT receiver's +`export ts`, `export flv`, and `export mkv` take `--delay` (default 500 ms) +instead, like an SRT receiver's latency: each frame is muxed that long after its decode time, all tracks in decode order, so two exporters of one broadcast emit the same order. A frame arriving after its deadline is dropped, and video resumes at its next -keyframe. The output's PCR follows the source's clock within what ISO/IEC -13818-1 allows. `--delay 0` writes frames in arrival order and drops nothing. +keyframe. The delay also bounds how stale a group may get, so `--delay 0` writes each +frame as soon as it is read and drops none as late, but still skips a stalled +group once a newer one is ready. +FLV and MKV reject their former `--max-delay` spelling with a migration +message. TS output's PCR follows the source's clock within what ISO/IEC +13818-1 allows. `export ts --passthrough` reads the track an `import ts --passthrough` published and writes its packets back out unchanged, from its first group on, each run diff --git a/doc/bin/rtmp.md b/doc/bin/rtmp.md index 049dca8898..b5ecd9a950 100644 --- a/doc/bin/rtmp.md +++ b/doc/bin/rtmp.md @@ -38,6 +38,11 @@ and anchors the catalog clock on the first frame, so it names the wall time the push arrived. A group starting before the previous group's start, such as an encoder restarting its timestamps mid-push, ends that push with an error. +Egress uses `--max-delay` (default 500 ms) for both the FLV release delay and +the stalled-group budget. Frames go out in decode order; a late video frame +is dropped and video resumes at the next keyframe. Zero releases available +frames immediately without deadline drops. + A player that advertises enhanced-RTMP multitrack receives every rendition. Any other player receives one video rendition, the largest picture (then highest bitrate) in a codec it advertised, and one audio rendition, the highest diff --git a/doc/setup/upgrade.md b/doc/setup/upgrade.md index b916390165..99818f5879 100644 --- a/doc/setup/upgrade.md +++ b/doc/setup/upgrade.md @@ -168,13 +168,15 @@ These land with the next breaking release, not the 2026-09-23 train. the rendition's own delay added on top. `Sync.out.jitter` now always equals `Sync.out.delay`, and `"auto"` with no decoder registered resolves to 0 rather than 100 ms. -- **`moq export ts --delay` replaces `--max-age`** (#4645). Each frame is +- **`moq export ts`, `flv`, and `mkv` take `--delay`** (#4645). Each frame is written that long after its decode time, on a clock that follows the - source's; `--max-age` and `--latency-max` are refused with the new flag - named. In Rust, moq-mux's `ts::Export::with_max_age` is `with_delay`, and - `ts::stats::Export` gains the dropped-frame count (late frames, and the - video frames then dropped waiting for a keyframe), measured drift and - out-of-tolerance count beside its `streams` rows. It is no longer `Eq`. + source's; the older staleness spellings are refused with the new flag + named. In Rust, moq-mux's `ts::Export::with_max_age` and the FLV and MKV + exports' `with_max_age` are `with_delay`. moq-rtmp's FLV egress releases + frames on its `max_delay` the same way. `ts::stats::Export` gains the + dropped-frame count (late frames, and the video frames then dropped waiting + for a keyframe), measured drift and out-of-tolerance count beside its + `streams` rows. It is no longer `Eq`. - **moq-mux has no clock translators.** `clock::Anchor`, `clock::Lane`, and `SourceMap` (#4667) are gone, along with the importers' `live()`. Publish the source's own timestamps and let the catalog clock map them to wall time. @@ -232,7 +234,7 @@ These land with the next breaking release, not the 2026-09-23 train. shares a name with a publisher's retention, which keeps `max_age`. In Rust, `track::Subscription::max_age` and `with_max_age` are `max_delay` and `with_max_delay`; moq-mux's `container::Consumer::set_max_age` and the fMP4, - MKV, FLV, H.264, and H.265 exports' `with_max_age` are `set_max_delay` and + H.264, and H.265 exports' `with_max_age` are `set_max_delay` and `with_max_delay`; moq-audio's and moq-video's `decode::Options::max_age` is `max_delay`, as is moq-audio's `decode::Consumer::max_age()`; and moq-rtmp's `Play::with_max_age`, `Client::with_export_max_age`, @@ -247,8 +249,8 @@ These land with the next breaking release, not the 2026-09-23 train. `MoqVideoDecoderOutput` take `max_delay_us` (each binding in its own casing), and so do C's `moq_subscription`, `moq_audio_decoder_output`, `moq_video_decoder_output`, `moq_consume_video`, and `moq_consume_audio`. - `moq export fmp4`, `mkv`, `flv`, `h264`, `h265`, and `rtmp` take - `--max-delay`, and refuse `--max-age`. `track::Info::max_age`, + `moq export fmp4`, `h264`, `h265`, and `rtmp` take `--max-delay`, and + refuse `--max-age`. `track::Info::max_age`, `MoqTrackInfo.max_age_us`, and `moq import --max-age` are unchanged, as is the wire. - **moq-relay auth takes the client-CA answer.** `auth::Config::validate` and `init` take `client_ca: bool`, whether any listener verifies client diff --git a/quest/m1/README.md b/quest/m1/README.md index d6bc9a009c..3299f3391f 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -200,7 +200,6 @@ disabled rung strands WHEP, HLS, and moq-mux export viewers. - [moqsink first catalog](/quest/m1/gst-first-catalog.md) - a restarted `moqsink` lists every requested pad in its first catalog, so `moqsrc` resumes each on its held pad - [moqsrc A/V alignment](/quest/m1/gst-src-av-sync.md) - every `moqsrc` pad of a run shares one timestamp reference and segment base - [TS passthrough export](/quest/m1/ts-passthrough-export.md) - `export ts --passthrough` writes the `m2ts` track back byte-identical (less late drops) on a fixed delay -- [FLV and MKV export delay](/quest/m1/export-delay.md) - FLV and MKV interleave through the shared jitter buffer on a fixed delay, breaking the CLI once - [Bindings size profile](/quest/m1/ffi-size-profile.md) - a benchmark decides whether the moq-ffi builds ship at opt-level "s", which halves the dylib - [Go mirror delivery](/quest/m1/go-mirror-delivery.md) - the Go binding's staticlibs stop growing git history by ~210 MiB per release - [Dart on iOS](/quest/m1/dart-ios.md) - prove the shipped iOS native asset actually loads on a device, which no CI can diff --git a/quest/m1/export-delay.md b/quest/m1/export-delay.md deleted file mode 100644 index f82470319d..0000000000 --- a/quest/m1/export-delay.md +++ /dev/null @@ -1,29 +0,0 @@ -# [S] FLV and MKV exports release on a fixed delay - -## Goal - -`moq export flv --delay ` and `moq export mkv --delay ` interleave -tracks through the shared jitter buffer, so their tag and block order is a -function of the media, not of arrival, as the TS export's is. - -## Plan - -Today each exporter's `pick_next_track` -(`rs/moq-mux/src/container/flv/export.rs`, -`rs/moq-mux/src/container/mkv/export.rs`) takes the smallest *pending* -timestamp, so which track goes first depends on which frame has arrived. -Replace both with the jitter buffer the TS export uses -(`rs/moq-mux/src/jitter.rs`), with the same `--delay` flag and the same -late-frame drop. Update `doc/bin/cli.md`. - -Decided in the 2026-10-06 audit: `--delay` replaces the staleness flag, as it -replaced `--max-age` on TS, so one knob sets both the release delay and the -sources' staleness budget. [#4917](https://github.com/moq-dev/moq/pull/4917) -already renamed that flag to `--max-delay` (`with_max_delay` on both -exporters); this quest replaces `--max-delay` with `--delay` for flv and mkv. - -Decided 2026-10-08: FLV and MKV land together in one quest, so the CLI and the -exporters' builder API break once. - -Public API: breaking, each exporter's staleness setting becomes the delay -setting. Wire: none. diff --git a/rs/justfile b/rs/justfile index 372ccfd9cf..b34502ea0c 100644 --- a/rs/justfile +++ b/rs/justfile @@ -357,3 +357,8 @@ bench-stats *args: # Sweep latency observations against writer count and report density. bench-latency: bash sh/rs/nextest.sh cargo nextest run --locked --release -p moq-bench --run-ignored only -E 'test(benchmark_observation_contention)' --success-output immediate + +# Sweep container export tracks and queued frames independently. +[positional-arguments] +bench-export *args: + cargo bench --locked -p moq-mux --bench export -- "$@" diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs index 9afe02c023..f6a05b46b3 100644 --- a/rs/moq-cli/src/args.rs +++ b/rs/moq-cli/src/args.rs @@ -906,9 +906,11 @@ impl Export { found.flag("--name", None, "--broadcast"); } match &self.sink { - ExportSink::Fmp4(args) | ExportSink::Mkv(args) => found.extend(args.container.deprecated()), + ExportSink::Fmp4(args) => found.extend(args.container.deprecated()), + ExportSink::Mkv(args) => found.extend(args.container.deprecated()), ExportSink::Ts(args) => found.extend(args.deprecated()), - ExportSink::Flv(args) | ExportSink::H264(args) | ExportSink::H265(args) => found.extend(args.deprecated()), + ExportSink::Flv(args) => found.extend(args.deprecated()), + ExportSink::H264(args) | ExportSink::H265(args) => found.extend(args.deprecated()), ExportSink::Hls(hls) => found.extend(hls.tls.deprecated()), ExportSink::Rtmp(rtmp) => found.extend(rtmp.deprecated()), _ => {} @@ -924,11 +926,11 @@ pub enum ExportSink { /// Fragmented MP4 / CMAF to stdout. Fmp4(Fragmented), /// Matroska / WebM to stdout. - Mkv(Fragmented), + Mkv(BufferedFragmented), /// MPEG-TS to stdout. Ts(Transport), /// FLV / RTMP container to stdout. - Flv(Container), + Flv(Buffered), /// H.264 Annex-B elementary stream to stdout. H264(Container), /// H.265 Annex-B elementary stream to stdout. @@ -976,6 +978,15 @@ impl ExportSink { mux_rate: None, passthrough: false, }; + let buffered = |format, args: &Buffered| Stdout { + format, + max_delay: args.delay.into_std(), + linger: args.linger.into_std(), + stitch: false, + fragment_duration: None, + mux_rate: None, + passthrough: false, + }; Some(match self { Self::Fmp4(args) => Stdout { fragment_duration: args.fragment_duration.map(crate::duration::Duration::into_std), @@ -983,7 +994,7 @@ impl ExportSink { }, Self::Mkv(args) => Stdout { fragment_duration: args.fragment_duration.map(crate::duration::Duration::into_std), - ..container(SubscribeFormat::Mkv, &args.container) + ..buffered(SubscribeFormat::Mkv, &args.container) }, Self::Ts(args) => Stdout { format: SubscribeFormat::Ts, @@ -994,7 +1005,7 @@ impl ExportSink { mux_rate: args.mux_rate, passthrough: args.passthrough, }, - Self::Flv(args) => container(SubscribeFormat::Flv, args), + Self::Flv(args) => buffered(SubscribeFormat::Flv, args), Self::H264(args) => container(SubscribeFormat::H264, args), Self::H265(args) => container(SubscribeFormat::H265, args), _ => return None, @@ -1005,7 +1016,7 @@ impl ExportSink { /// A stdout sink's format and the options that apply to it. pub struct Stdout { pub format: SubscribeFormat, - /// The staleness budget, which `ts` also holds every frame for (`--delay`). + /// The staleness budget, also used as the TS/FLV/MKV release delay. pub max_delay: Duration, pub linger: Duration, /// Follow a replacement as a program switch (`ts` only). @@ -1016,7 +1027,7 @@ pub struct Stdout { pub passthrough: bool, } -/// Options shared by the stdout container sinks other than `ts`. +/// Staleness options for stdout sinks without a fixed release delay. #[derive(usage::Args, Clone)] #[usage(unknown_flags = "error", args_override_self = false)] pub struct Container { @@ -1051,6 +1062,56 @@ impl Container { } } +/// A stdout container with a fixed release delay. +#[derive(usage::Args, Clone)] +#[usage(unknown_flags = "error", args_override_self = false)] +pub struct Buffered { + /// How long after its decode time each frame is written (e.g. `500ms`); late frames are dropped. + /// The same delay bounds source staleness. Zero writes each frame as soon as it is read and + /// drops none as late, but still skips a stalled group once a newer one is ready. + #[usage(long, default = "500ms")] + pub delay: crate::duration::Duration, + /// Accepted only to refuse a nonzero value: only TS can mark a broadcast restart. + #[usage(long, default = "0s", hide = true)] + pub linger: crate::duration::Duration, + /// The retired staleness-only setting. + #[usage(long, hide = true)] + max_delay: Option, + /// An older spelling of the staleness setting. + #[usage(long = "max-age", hide = true)] + max_age: Option, + /// An older spelling of the staleness setting. + #[usage(long = "latency-max", hide = true)] + latency_max: Option, +} + +impl Buffered { + fn deprecated(&self) -> moq_tokio::cli::Deprecated { + let mut found = moq_tokio::cli::Deprecated::default(); + for (present, flag) in [ + (self.max_delay.is_some(), "--max-delay"), + (self.max_age.is_some(), "--max-age"), + (self.latency_max.is_some(), "--latency-max"), + ] { + if present { + found.flag(flag, None, "--delay"); + } + } + found + } +} + +/// A buffered Matroska output with an optional cluster cap. +#[derive(usage::Args, Clone)] +#[usage(unknown_flags = "error", args_override_self = false)] +pub struct BufferedFragmented { + #[usage(flatten)] + pub container: Buffered, + /// Cap the output cluster duration (e.g. `2s`); defaults to video GOPs. + #[usage(long)] + pub fragment_duration: Option, +} + /// The MPEG-TS stdout container. #[derive(usage::Args, Clone)] #[usage(unknown_flags = "error", args_override_self = false)] @@ -1108,7 +1169,7 @@ impl Transport { } } -/// The fmp4 / mkv stdout containers: [`Container`] plus a fragment cap. +/// The fMP4 stdout container with a fragment cap. #[derive(usage::Args, Clone)] #[usage(unknown_flags = "error", args_override_self = false)] pub struct Fragmented { @@ -1116,7 +1177,7 @@ pub struct Fragmented { pub container: Container, /// Cap the output fragment/cluster duration (e.g. `2s`). - /// Defaults to publisher groups for fMP4 and video GOPs for MKV. + /// Defaults to publisher groups. #[usage(long)] pub fragment_duration: Option, } @@ -1157,7 +1218,7 @@ mod tests { .unwrap() }; assert!(parse("ts", "10s").validate().is_ok()); - for format in ["fmp4", "mkv", "flv", "h264", "h265"] { + for format in ["fmp4", "h264", "h265"] { let err = parse(format, "10s").validate().unwrap_err().to_string(); assert!(err.contains("--linger"), "{format}: {err}"); assert!( @@ -1873,11 +1934,11 @@ mod tests { else { panic!("--latency-max must not start a run"); }; - assert!(err.to_string().contains("--latency-max -> --max-delay"), "{}", err); + assert!(err.to_string().contains("--latency-max -> --delay"), "{}", err); } /// A publisher's retention is `--max-age` on `import`; a subscriber's staleness budget is - /// `--max-delay` on `export`, except `export ts`, which spells it `--delay`. + /// `--max-delay` for immediate exports and `--delay` for buffered exports. #[test] fn export_staleness_is_max_delay_and_import_retention_is_max_age() { let export = |args: &[&str]| { @@ -1893,7 +1954,7 @@ mod tests { export.sink.stdout().unwrap().max_delay }; - for format in ["fmp4", "mkv", "flv", "h264", "h265"] { + for format in ["fmp4", "h264", "h265"] { assert_eq!(stdout(&[format]), Duration::from_millis(500), "{format}"); assert_eq!( stdout(&[format, "--max-delay", "2s"]), @@ -1906,6 +1967,20 @@ mod tests { assert!(err.to_string().contains("--max-age -> --max-delay"), "{format}: {err}"); } + for format in ["flv", "mkv"] { + assert_eq!(stdout(&[format]), Duration::from_millis(500)); + assert_eq!(stdout(&[format, "--delay", "2s"]), Duration::from_secs(2)); + for old in ["--max-delay", "--max-age", "--latency-max"] { + let Err(err) = export(&[format, old, "2s"]) else { + panic!("{old} must be refused") + }; + assert!( + err.to_string().contains(&format!("{old} -> --delay")), + "{format}: {err}" + ); + } + } + assert_eq!(stdout(&["ts", "--delay", "2s"]), Duration::from_secs(2)); assert!(export(&["ts", "--max-delay", "2s"]).is_err(), "ts spells it --delay"); diff --git a/rs/moq-cli/src/rtmp.rs b/rs/moq-cli/src/rtmp.rs index 7879c622e0..3be612f41a 100644 --- a/rs/moq-cli/src/rtmp.rs +++ b/rs/moq-cli/src/rtmp.rs @@ -37,8 +37,8 @@ pub struct ExportArgs { #[usage(flatten)] pub endpoint: Args, - /// How far a group may fall behind the live edge before it is skipped. RTMP is - /// unpaced, so this bounds buffering, not the wire rate. + /// FLV frame release delay and stalled-group budget. Late video resumes at + /// its next keyframe; zero releases available frames without deadline drops. #[usage(long, default = "500ms")] pub max_delay: crate::duration::Duration, diff --git a/rs/moq-cli/src/subscribe.rs b/rs/moq-cli/src/subscribe.rs index 238dcd2d7a..15ec3457d1 100644 --- a/rs/moq-cli/src/subscribe.rs +++ b/rs/moq-cli/src/subscribe.rs @@ -324,7 +324,7 @@ impl Subscribe { // shape internally (synthesizing avcC/hvcC from inline parameter sets). let stream = self.stream().await?; let mut mkv = moq_mux::container::mkv::Export::new(self.source, stream) - .with_max_delay(self.args.max_delay) + .with_delay(self.args.max_delay) .with_fragment_duration(self.args.fragment_duration); while let Some(chunk) = mkv.next().await? { @@ -480,7 +480,7 @@ impl Subscribe { // internally (synthesizing avcC from inline parameter sets). Only H.264 video // and AAC audio are supported; `fragment_duration` does not apply to FLV. let stream = self.stream().await?; - let mut flv = moq_mux::container::flv::Export::new(self.source, stream).with_max_delay(self.args.max_delay); + let mut flv = moq_mux::container::flv::Export::new(self.source, stream).with_delay(self.args.max_delay); while let Some(chunk) = flv.next().await? { stdout.write_all(&chunk).await?; diff --git a/rs/moq-mux/Cargo.toml b/rs/moq-mux/Cargo.toml index bb3364c37c..6dd31df234 100644 --- a/rs/moq-mux/Cargo.toml +++ b/rs/moq-mux/Cargo.toml @@ -50,3 +50,7 @@ tracing-test = { workspace = true, features = ["no-env-filter"] } [[bench]] name = "catalog" harness = false + +[[bench]] +name = "export" +harness = false diff --git a/rs/moq-mux/benches/export.rs b/rs/moq-mux/benches/export.rs new file mode 100644 index 0000000000..d35a1b7f66 --- /dev/null +++ b/rs/moq-mux/benches/export.rs @@ -0,0 +1,127 @@ +//! Container export cost swept over independently varied tracks and queued frames. +use std::time::Duration; + +use bytes::Bytes; +use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use hang::catalog::{AAC, AudioConfig}; +use moq_mux::catalog::hang::Container; +use moq_mux::container::{Frame, Kind}; +use moq_net::Timestamp; + +/// What keeps a prepared broadcast served; dropping it stops the origin's tasks. +struct Fixture { + /// The broadcast, its catalog and its tracks, which a reader needs kept open. + _published: Box, + _origin: moq_net::origin::Producer, + driver: tokio::task::JoinHandle, + requests: tokio::task::JoinHandle<()>, +} + +impl Drop for Fixture { + fn drop(&mut self) { + self.driver.abort(); + self.requests.abort(); + } +} + +/// A finished broadcast of `tracks` audio tracks with `frames` queued frames each. +async fn prepare(tracks: usize, frames: usize) -> (moq_mux::Source, moq_mux::catalog::Consumer, Fixture) { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let mut catalog = moq_mux::catalog::Producer::new(&mut broadcast, Default::default()).unwrap(); + let stream = moq_mux::catalog::Consumer::Hang(catalog.consume().unwrap()); + let mut publishers = Vec::new(); + for index in 0..tracks { + let track = broadcast + .create_track( + format!("audio{index}"), + catalog.track_info(hang::catalog::PRIORITY.audio), + ) + .unwrap(); + let mut config = AudioConfig::new(AAC { profile: 2 }, 44100, 2); + config.description = Some(Bytes::from_static(&[0x12, 0x10])); + publishers.push(catalog.audio(track, Container::Legacy(Kind::Audio), config).unwrap()); + } + for track in &mut publishers { + for index in 0..frames { + track + .write(Frame { + timestamp: Timestamp::from_millis(index as u64 * 20).unwrap(), + duration: None, + payload: Bytes::from_static(b"audio-frame"), + keyframe: index == 0, + }) + .unwrap(); + } + track.finish().unwrap(); + } + catalog.finish().unwrap(); + let (origin, driver) = moq_net::origin::Producer::new(Default::default()); + let driver = tokio::spawn(moq_net::time::run(driver)); + let dynamic = origin.dynamic("", Default::default()).unwrap(); + let requests = tokio::spawn(async move { + while let Ok(request) = dynamic.requested_broadcast().await { + request.accept(consumer.clone()); + } + }); + let resolved = origin.consume().request_broadcast("bench", None).await.unwrap(); + let source = moq_mux::Source::new(origin.consume(), resolved); + let fixture = Fixture { + _published: Box::new((broadcast, catalog, publishers)), + _origin: origin, + driver, + requests, + }; + (source, stream, fixture) +} + +async fn export(format: &str, source: moq_mux::Source, stream: moq_mux::catalog::Consumer) -> usize { + let mut chunks = 0; + if format == "flv" { + let mut export = moq_mux::container::flv::Export::new(source, stream).with_multitrack(true); + while let Some(chunk) = export.next().await.unwrap() { + std::hint::black_box(chunk); + chunks += 1; + } + } else { + let mut export = moq_mux::container::mkv::Export::new(source, stream).with_fragment_duration(Duration::ZERO); + while let Some(chunk) = export.next().await.unwrap() { + std::hint::black_box(chunk); + chunks += 1; + } + } + chunks +} + +fn bench_export(c: &mut Criterion) { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let mut group = c.benchmark_group("container-export"); + group.sample_size(10); + group.measurement_time(Duration::from_secs(1)); + for format in ["flv", "mkv"] { + for tracks in [1, 4, 16] { + for frames in [1, 16, 64] { + group.throughput(Throughput::Elements((tracks * frames) as u64)); + group.bench_function(BenchmarkId::new(format, format!("{tracks}x{frames}")), |b| { + // Publishing the broadcast is setup; only exporting it is measured. + b.iter_batched( + || runtime.block_on(prepare(tracks, frames)), + |(source, stream, fixture)| { + let chunks = runtime.block_on(export(format, source, stream)); + assert_eq!(chunks, 1 + tracks * frames); + fixture + }, + BatchSize::PerIteration, + ); + }); + } + } + } + group.finish(); +} + +criterion_group!(benches, bench_export); +criterion_main!(benches); diff --git a/rs/moq-mux/src/codec/video.rs b/rs/moq-mux/src/codec/video.rs index 4d00cc9f3d..7a5a64fd67 100644 --- a/rs/moq-mux/src/codec/video.rs +++ b/rs/moq-mux/src/codec/video.rs @@ -22,6 +22,21 @@ pub(crate) struct Reorder { pub period: Option<(u64, u64)>, } +impl Reorder { + /// The minimum decode delay and lookahead in 90 kHz ticks. + pub(crate) fn timing(self, framerate: Option) -> (u64, u64) { + let period = match self.period { + Some((units, scale)) => units.saturating_mul(90_000).div_ceil(scale), + None => framerate + .filter(|fps| fps.is_finite() && *fps > 0.0) + .map_or(0, |fps| (90_000.0 / fps).ceil() as u64), + }; + let delay = u64::from(self.depth).saturating_mul(period); + let lookahead = 1u64.checked_shl(self.depth).unwrap_or(u64::MAX).saturating_mul(period); + (delay, lookahead) + } +} + /// The hypothetical reference decoder a video sequence header declares: its NAL HRD's last /// schedule, the one the transport stream's decoder buffers are sized from (ISO 13818-1 /// 2.14.3.1, 2.17.2). diff --git a/rs/moq-mux/src/container/decode.rs b/rs/moq-mux/src/container/decode.rs new file mode 100644 index 0000000000..d0a88d7e92 --- /dev/null +++ b/rs/moq-mux/src/container/decode.rs @@ -0,0 +1,99 @@ +use std::collections::VecDeque; + +/// The largest supported reorder span in 90 kHz ticks. +pub(crate) const MAX_REORDER: u64 = 2 * 90_000; + +/// A video rendition's decode clock, authored from presentation timestamps. +/// +/// [`crate::container::Frame`] carries only a presentation timestamp and frames reach the muxer in decode order, +/// so a B-frame stream arrives with non-monotonic PTS and no decode time. An encoder decodes +/// the n-th frame in decode order at the n-th presentation time, a fixed reorder delay early, +/// so that is the DTS authored here: with decode-order PTS 0, 120, 40, 80 and a delay of one +/// 40 ms frame, the frames decode at -40, 0, 40, 80. It keeps the encoder's own spacing, a +/// frame's or a field's, through any switch between frame and field coding, and the delay is +/// a time, the least that keeps every DTS at or before its PTS (or the SPS's declared depth, +/// when that is more). On a source muxed that way this is its DTS, frame for frame. +/// +/// A frame's slot is settled once no frame still to come can be presented before it. A later +/// frame decodes after this one, and is presented at most the catalog `jitter` after it +/// decodes, so the slot is settled once the highest PTS read is that far past it. Frames wait +/// here until then, as long as the reordering spans and well inside the jitter buffer's delay. +/// Without a `jitter`, the SPS's depth bounds it, and so does how far a frame was presented +/// below the highest PTS read before it. A frame that still comes in below a settled slot +/// (reordering deeper than all of those) is nudged one tick past the last DTS, so the clock +/// stays strictly increasing. +/// +/// Ticks are continuous (unwrapped) 90 kHz, so the clock never wraps mid-stream; the 33-bit +/// wire wrap happens once at emission. +pub(crate) struct DecodeClock { + /// Frames read but not yet given a DTS, in decode order, with their PTS. + held: VecDeque<(T, u64)>, + /// The held frames' PTS, ascending: the slots they decode in, in turn. + slots: Vec, + /// The highest PTS read. + high: Option, + /// The furthest a frame was presented below the highest PTS read before it, plus a tick. + reach: u64, + /// The reorder delay in effect. It never shrinks: that would step the clock back. + delay: u64, + /// The last DTS handed out. + last: Option, +} + +impl Default for DecodeClock { + fn default() -> Self { + Self { + held: VecDeque::new(), + slots: Vec::new(), + high: None, + reach: 0, + delay: 0, + last: None, + } + } +} + +impl DecodeClock { + /// Hold the next frame in decode order, presented at `pts`. + pub(crate) fn push(&mut self, item: T, pts: u64) { + if let Some(high) = self.high + && pts < high + { + self.reach = self.reach.max(high - pts + 1).min(MAX_REORDER); + } + self.high = Some(self.high.map_or(pts, |high| high.max(pts))); + let at = self.slots.partition_point(|&slot| slot <= pts); + self.slots.insert(at, pts); + self.held.push_back((item, pts)); + } + + /// The next frame and its DTS, once its slot is settled `lookahead` past (the catalog + /// `jitter`), or at once when `flush`. `floor` is the least reorder delay, in ticks. + pub(crate) fn pop(&mut self, lookahead: u64, floor: u64, flush: bool) -> Option<(T, u64)> { + let high = self.high?; + let lookahead = lookahead.max(self.reach); + let settled = |slot: u64| flush || high >= slot.saturating_add(lookahead); + if !settled(*self.slots.first()?) { + return None; + } + // Every settled frame bounds the delay, so one decoding later never steps it back. + let mut delay = self.delay.max(floor); + for (&slot, (_, pts)) in self.slots.iter().zip(&self.held) { + if !settled(slot) { + break; + } + delay = delay.max(slot.saturating_sub(*pts)); + } + self.delay = delay.min(MAX_REORDER); + + let slot = self.slots.remove(0); + let (item, _) = self.held.pop_front()?; + let dts = slot.saturating_sub(self.delay); + let dts = match self.last { + Some(last) if dts <= last => last + 1, + _ => dts, + }; + self.last = Some(dts); + Some((item, dts)) + } +} diff --git a/rs/moq-mux/src/container/flv/export.rs b/rs/moq-mux/src/container/flv/export.rs index 5b52b1d063..6faa987e84 100644 --- a/rs/moq-mux/src/container/flv/export.rs +++ b/rs/moq-mux/src/container/flv/export.rs @@ -108,7 +108,9 @@ impl Flavor { pub struct Export { source: crate::Source, catalog: Option, - max_delay: std::time::Duration, + delay: std::time::Duration, + /// Frames with their tag timestamp, encoded only once released. + jitter: crate::jitter::Buffer<(u8, usize), (Frame, u32)>, /// Emit every rendition as an enhanced-RTMP multitrack track, rather than only /// the best video and audio rendition. multitrack: bool, @@ -127,7 +129,9 @@ struct FlvTrack { /// used when muxing multitrack; a single-track FLV stream carries no id. track_id: u8, source: ExportSource, - pending: Option, + pending: Option>, + /// When the source was last empty, bounding the next frame's arrival. + empty: web_async::time::Instant, finished: bool, /// The FLV payload shape (legacy CodecID vs enhanced FourCC) to mux this /// track as, fixed from its catalog codec when it's bound. @@ -178,7 +182,8 @@ impl Export { Self { source, catalog: Some(catalog), - max_delay: std::time::Duration::ZERO, + delay: std::time::Duration::ZERO, + jitter: crate::jitter::Buffer::new(std::time::Duration::ZERO), multitrack: false, video: Vec::new(), audio: Vec::new(), @@ -186,13 +191,14 @@ impl Export { } } - /// Set the max delay for each per-track source. + /// Release frames a fixed delay after their decode time, dropping late frames. /// - /// See [`Consumer`](crate::container::Consumer) for the per-track skip behavior. - /// Defaults to - /// [`std::time::Duration::ZERO`](std::time::Duration::ZERO) (skip aggressively). - pub fn with_max_delay(mut self, max_delay: std::time::Duration) -> Self { - self.max_delay = max_delay; + /// The same delay bounds each source's staleness. Zero (the default) releases each frame as + /// soon as it is read and drops none as late, but a stalled group is still skipped once a + /// newer one is ready. + pub fn with_delay(mut self, delay: std::time::Duration) -> Self { + self.delay = delay; + self.jitter = crate::jitter::Buffer::new(delay); self } @@ -227,31 +233,12 @@ impl Export { } } - // 2. Pull frames from each track into `pending`. Pre-header, drop slices - // that arrived before the track's codec config is ready: a mid-GOP joiner - // can't render them, and parking would block polling for the next SPS/PPS. - let waiting_for_header = !self.header_emitted; - for track in self.video.iter_mut().chain(self.audio.iter_mut()) { - if track.pending.is_some() || track.finished { - continue; - } - loop { - match track.source.poll_read(waiter)? { - Poll::Ready(Some(frame)) => { - if waiting_for_header && !track.source.header_ready() { - continue; - } - track.pending = Some(frame); - break; - } - Poll::Ready(None) => { - track.finished = true; - break; - } - Poll::Pending => break, - } - } - } + self.jitter.expect( + (0..self.video.len()) + .map(|i| (0, i)) + .chain((0..self.audio.len()).map(|i| (1, i))), + ); + self.fill(waiter)?; // 3. Emit the header once every track's codec config has resolved. if !self.header_emitted { @@ -268,16 +255,13 @@ impl Export { return Poll::Pending; } - // 4. Emit the smallest FLV tag timestamp as one tag. - if let Some((is_video, index)) = self.pick_next_track()? { - let track = if is_video { - &mut self.video[index] - } else { - &mut self.audio[index] - }; - let frame = track.pending.take().unwrap(); - let chunk = self.encode_frame(is_video, index, frame)?; - return Poll::Ready(Ok(Some(chunk))); + if let Poll::Ready(ready) = self.jitter.poll_next(waiter) { + let (kind, index) = ready.track; + let (frame, timestamp) = ready.item; + return Poll::Ready(Ok(Some(self.encode_frame(kind == 0, index, frame, timestamp)?))); + } + if !self.jitter.is_empty() { + return Poll::Pending; } // 5. End-of-stream once every subscribed track is drained. @@ -301,6 +285,113 @@ impl Export { !self.video.is_empty() || !self.audio.is_empty() } + #[cfg(test)] + pub(super) fn next_deadline(&self) -> Option { + self.jitter.next_deadline() + } + + /// How many frames the export holds between its sources and the output. + #[cfg(test)] + pub(super) fn buffered(&self) -> usize { + self.jitter.len() + } + + #[cfg(test)] + pub(super) fn with_replay(mut self) -> Self { + self.jitter = self.jitter.replay(); + self + } + + /// Read each track until it has a frame due, preserving ingress timing before buffering. + fn fill(&mut self, waiter: &kio::Waiter) -> anyhow::Result<()> { + for kind in [0, 1] { + let count = if kind == 0 { self.video.len() } else { self.audio.len() }; + for index in 0..count { + loop { + if self.header_emitted && self.jitter.full(&(kind, index)) { + break; + } + let track = if kind == 0 { + &mut self.video[index] + } else { + &mut self.audio[index] + }; + if !self.header_emitted && track.pending.is_some() { + break; + } + let arrival = if let Some(pending) = track.pending.take() { + pending + } else { + if track.finished { + break; + } + let frame = match track.source.poll_read(waiter)? { + Poll::Ready(Some(frame)) => frame, + Poll::Ready(None) => { + track.finished = true; + break; + } + Poll::Pending => { + track.empty = web_async::time::Instant::now(); + break; + } + }; + if !self.header_emitted && !track.source.header_ready() { + continue; + } + crate::jitter::Arrival { + arrived: track.empty, + read: web_async::time::Instant::now(), + decode: frame.timestamp, + restart: track.source.markers(), + skip: track.source.skips(), + sync: kind == 1 || frame.keyframe, + item: frame, + } + }; + if !self.header_emitted { + track.pending = Some(arrival); + break; + } + let timestamp = track.tag_timestamp(&arrival.item)?; + if track.flavor.has_composition_time() { + track.last_dts = Some(timestamp); + } + let crate::jitter::Arrival { + arrived, + read, + restart, + skip, + sync, + item, + .. + } = arrival; + if self.jitter.push( + (kind, index), + crate::jitter::Arrival { + arrived, + read, + decode: moq_net::Timestamp::from_millis(u64::from(timestamp))?, + restart, + skip, + sync, + item: (item, timestamp), + }, + )? == crate::jitter::Push::Late + { + tracing::warn!( + kind, + index, + dropped = self.jitter.dropped(), + "frame missed its deadline; dropped" + ); + } + } + } + } + Ok(()) + } + fn update_catalog(&mut self, mut catalog: Catalog) -> anyhow::Result<()> { self.source.retain_valid_media(&mut catalog); @@ -367,7 +458,7 @@ impl Export { (VideoCodec::AV1(av1), None) => Some(Bytes::copy_from_slice(&av1c_bytes(av1))), _ => None, }; - let Some(source) = ExportSource::for_video(&self.source, name, config, self.max_delay)? else { + let Some(source) = ExportSource::for_video(&self.source, name, config, self.delay)? else { continue; }; let track_id = u8::try_from(self.video.len()).context("too many FLV video tracks")?; @@ -376,6 +467,7 @@ impl Export { track_id, source, pending: None, + empty: web_async::time::Instant::now(), finished: false, flavor, fallback_description, @@ -420,7 +512,7 @@ impl Export { } let flavor = audio_flavor(config)?; ensure_legacy(&config.container, "audio", name)?; - let Some(source) = ExportSource::for_audio(&self.source, name, config, self.max_delay)? else { + let Some(source) = ExportSource::for_audio(&self.source, name, config, self.delay)? else { continue; }; let track_id = u8::try_from(self.audio.len()).context("too many FLV audio tracks")?; @@ -429,6 +521,7 @@ impl Export { track_id, source, pending: None, + empty: web_async::time::Instant::now(), finished: false, flavor, fallback_description: None, @@ -479,40 +572,15 @@ impl Export { Ok(out.freeze()) } - /// Pick the track with the smallest pending FLV tag timestamp. Returns whether - /// it's a video track and its index within that kind, or `None` if no frame is - /// pending. Video wins ties, matching the single-track interleave. - fn pick_next_track(&self) -> anyhow::Result> { - let mut best: Option<(u32, bool, usize)> = None; - for (index, track) in self.video.iter().enumerate() { - if let Some(frame) = &track.pending { - let ts = track.tag_timestamp(frame)?; - if best.is_none_or(|(b, _, _)| ts < b) { - best = Some((ts, true, index)); - } - } - } - for (index, track) in self.audio.iter().enumerate() { - if let Some(frame) = &track.pending { - let ts = frame_timestamp_ms(frame)?; - if best.is_none_or(|(b, _, _)| ts < b) { - best = Some((ts, false, index)); - } - } - } - Ok(best.map(|(_, is_video, index)| (is_video, index))) - } - - /// Encode one frame as a single FLV tag. - fn encode_frame(&mut self, is_video: bool, index: usize, frame: Frame) -> anyhow::Result { + /// Encode one frame as a single FLV tag at its authored `timestamp_ms`. + fn encode_frame(&self, is_video: bool, index: usize, frame: Frame, timestamp_ms: u32) -> anyhow::Result { let mut out = BytesMut::with_capacity(TAG_HEADER_LEN + frame.payload.len() + 8); let multitrack = self.multitrack; if is_video { - let track = &mut self.video[index]; + let track = &self.video[index]; let flavor = track.flavor; let track_id = track.track_id; let pts_ms = frame_timestamp_ms(&frame)?; - let timestamp_ms = track.tag_timestamp(&frame)?; let frame_type = if frame.keyframe { FRAME_TYPE_KEY } else { @@ -553,16 +621,12 @@ impl Export { } } } - if flavor.has_composition_time() { - track.last_dts = Some(timestamp_ms); - } body.put_slice(&frame.payload); write_tag(&mut out, TAG_VIDEO, timestamp_ms, &body)?; } else { let track = &self.audio[index]; let flavor = track.flavor; let track_id = track.track_id; - let timestamp_ms = frame_timestamp_ms(&frame)?; let mut body = BytesMut::with_capacity(7 + frame.payload.len()); if multitrack { body.put_u8((AUDIO_FORMAT_EX << 4) | AUDIO_PACKET_MULTITRACK); diff --git a/rs/moq-mux/src/container/flv/export_test.rs b/rs/moq-mux/src/container/flv/export_test.rs index 9c99b7853e..ea099bf548 100644 --- a/rs/moq-mux/src/container/flv/export_test.rs +++ b/rs/moq-mux/src/container/flv/export_test.rs @@ -135,7 +135,13 @@ async fn drain_export(mut exporter: Export, import Ok(Ok(Some(chunk))) => exported.extend_from_slice(&chunk), Ok(Ok(None)) => break, Ok(Err(e)) => panic!("exporter error: {e}"), - Err(_) => importer = None, // close the broadcast so the exporter can EOS + Err(_) => { + if let Some(deadline) = exporter.next_deadline() { + tokio::time::advance(deadline.saturating_duration_since(web_async::time::Instant::now())).await; + } else { + importer = None; // No scheduled frames remain; close the broadcast. + } + } } } drop(importer); @@ -193,7 +199,8 @@ async fn export_emits_sequence_headers_and_frames() { let exporter = open_export(crate::source::announced(&consumer).await) .await - .with_max_delay(RECORDING_MAX_DELAY); + .with_delay(RECORDING_MAX_DELAY) + .with_replay(); let exported = drain_export(exporter, importer).await; let tags = parse_tags(&exported); @@ -1223,7 +1230,10 @@ async fn drain_until_idle( Ok(Ok(Some(chunk))) => out.extend_from_slice(&chunk), Ok(Ok(None)) => return Some(Ok(())), Ok(Err(e)) => return Some(Err(e)), - Err(_) => return None, + Err(_) => { + let deadline = exporter.next_deadline()?; + tokio::time::advance(deadline.saturating_duration_since(web_async::time::Instant::now())).await; + } } } } @@ -1277,9 +1287,14 @@ async fn a_track_leaving_the_catalog_is_read_to_its_end() { let mut exporter = open_export(crate::source::announced(&consumer).await) .await - .with_max_delay(RECORDING_MAX_DELAY); + .with_delay(RECORDING_MAX_DELAY) + .with_replay(); let mut exported = Vec::new(); - assert!(drain_until_idle(&mut exporter, &mut exported).await.is_none()); + // Consume the header and the two initial frames without advancing past + // their release deadline before publishing the next live frames. + for _ in 0..3 { + exported.extend_from_slice(&exporter.next().await.unwrap().unwrap()); + } // The audio rendition retires before its tail lands; the video carries on. catalog.modify().unwrap().audio.renditions.remove(&audio_name); @@ -1314,3 +1329,146 @@ async fn drain_exporter_chunks(mut exporter: Export= delay); + + // This frame's 120ms deadline has already passed. + tokio::time::advance(std::time::Duration::from_millis(200)).await; + assert!(export.poll_next(&kio::Waiter::noop()).is_pending()); + live.track.write(raw_frame(20_000, b"expired", true)).unwrap(); + live.track.write(raw_frame(400_000, b"final", true)).unwrap(); + live.track.finish().unwrap(); + live.catalog.finish().unwrap(); + let final_frame = export.next().await.unwrap().unwrap(); + assert!(final_frame.windows(5).any(|part| part == b"final")); + assert!(!final_frame.windows(7).any(|part| part == b"expired")); + assert!(start.elapsed() >= std::time::Duration::from_millis(500)); + assert!(export.next().await.unwrap().is_none()); +} + +#[tokio::test(start_paused = true)] +async fn decode_order_is_independent_of_cross_track_arrival_order() { + use crate::container::test_util::{Live, raw_frame}; + use hang::catalog::{AAC, AudioConfig, Container}; + for reversed in [false, true] { + let mut config = AudioConfig::new(AAC { profile: 2 }, 44100, 2); + config.description = Some(Bytes::from_static(&[0x12, 0x10])); + config.container = Container::Legacy; + let mut live = Live::audio(config.clone()); + let mut other = live.add_track("other", |catalog, name| { + catalog.modify().unwrap().audio.renditions.insert(name, config); + }); + let mut export = super::Export::new(live.source().await, live.catalog_stream().await) + .with_delay(std::time::Duration::from_millis(100)) + .with_multitrack(true); + export.next().await.unwrap().unwrap(); + live.track.write(raw_frame(0, b"initial-a", true)).unwrap(); + other.write(raw_frame(0, b"initial-b", true)).unwrap(); + export.next().await.unwrap().unwrap(); + export.next().await.unwrap().unwrap(); + if reversed { + other.write(raw_frame(100_000, b"earlier", true)).unwrap(); + } else { + live.track.write(raw_frame(200_000, b"later", true)).unwrap(); + } + assert!(export.poll_next(&kio::Waiter::noop()).is_pending()); + tokio::time::advance(std::time::Duration::from_millis(20)).await; + if reversed { + live.track.write(raw_frame(200_000, b"later", true)).unwrap(); + } else { + other.write(raw_frame(100_000, b"earlier", true)).unwrap(); + } + let first = export.next().await.unwrap().unwrap(); + let second = export.next().await.unwrap().unwrap(); + assert!(first.windows(7).any(|part| part == b"earlier")); + assert!(second.windows(5).any(|part| part == b"later")); + } +} + +/// A backlog the source already holds, as after the sink stalls, goes out a frame at a time +/// as each falls due, rather than being read into the export whole first. +#[tokio::test(start_paused = true)] +async fn a_backlog_stays_in_the_source_until_due() { + use crate::container::test_util::{Live, raw_frame}; + use hang::catalog::{AAC, AudioConfig}; + for delay in [std::time::Duration::ZERO, std::time::Duration::from_millis(100)] { + let mut config = AudioConfig::new(AAC { profile: 2 }, 44100, 2); + config.description = Some(Bytes::from_static(&[0x12, 0x10])); + let mut live = Live::audio(config); + let mut export = super::Export::new(live.source().await, live.catalog_stream().await).with_delay(delay); + export.next().await.unwrap().unwrap(); + live.track.write(raw_frame(0, b"first", true)).unwrap(); + export.next().await.unwrap().unwrap(); + for index in 1..=50 { + tokio::time::advance(std::time::Duration::from_millis(20)).await; + live.track.write(raw_frame(index * 20_000, b"backlog", false)).unwrap(); + } + for _ in 1..=50 { + let tag = export.next().await.unwrap().unwrap(); + assert!(tag.windows(7).any(|part| part == b"backlog")); + assert!( + export.buffered() <= 1, + "{delay:?}: {} frames buffered", + export.buffered() + ); + } + } +} + +/// A joiner reads the retained group to its live edge to anchor the clock, then holds no +/// more than the delay of it. +#[tokio::test(start_paused = true)] +async fn a_retained_group_trims_to_the_delay() { + use crate::container::test_util::{Live, raw_frame}; + use hang::catalog::{AAC, AudioConfig}; + let mut config = AudioConfig::new(AAC { profile: 2 }, 44100, 2); + config.description = Some(Bytes::from_static(&[0x12, 0x10])); + let mut live = Live::audio(config); + for index in 0..50 { + live.track + .write(raw_frame(index * 20_000, b"retained", index == 0)) + .unwrap(); + tokio::time::advance(std::time::Duration::from_millis(20)).await; + } + let delay = std::time::Duration::from_millis(100); + let mut export = super::Export::new(live.source().await, live.catalog_stream().await).with_delay(delay); + export.next().await.unwrap().unwrap(); + export.next().await.unwrap().unwrap(); + // What is left spans no more than the delay: five 20 ms frames. + assert!(export.buffered() <= 5, "{} frames buffered", export.buffered()); +} diff --git a/rs/moq-mux/src/container/mkv/export.rs b/rs/moq-mux/src/container/mkv/export.rs index ba8a8d2f3a..69349ff0ce 100644 --- a/rs/moq-mux/src/container/mkv/export.rs +++ b/rs/moq-mux/src/container/mkv/export.rs @@ -47,7 +47,10 @@ const TIMESTAMP_SCALE_NS: u64 = 1_000_000; pub struct Export { source: crate::Source, catalog: Option, - max_delay: Duration, + delay: Duration, + jitter: crate::jitter::Buffer, + /// An already released frame waiting for the previous cluster to be returned. + held: Option<(String, Frame)>, fragment_duration: Option, tracks: HashMap, @@ -63,13 +66,53 @@ pub struct Export { } struct MkvTrack { + clock: crate::container::decode::DecodeClock>, + config: Option, + restart: u64, + reorder: Option<(u64, u64)>, source: ExportSource, - pending: Option, + pending: Option>, + empty: web_async::time::Instant, finished: bool, track_number: u64, kind: TrackKind, } +impl MkvTrack { + fn release(&mut self, name: &str, jitter: &mut crate::jitter::Buffer, flush: bool) -> Result<()> { + let Some(config) = &self.config else { return Ok(()) }; + let (floor, lookahead) = *self.reorder.get_or_insert_with(|| { + let declared = self.source.description().and_then(|description| match config.codec { + VideoCodec::H264(_) => { + crate::codec::h264::sps_reorder(crate::codec::h264::Avcc::parse(description).ok()?.sps.first()?) + } + VideoCodec::H265(_) => { + crate::codec::h265::sps_reorder(crate::codec::h265::Hvcc::parse(description).ok()?.sps.first()?) + } + _ => None, + }); + let (floor, lookahead) = declared.map_or((0, 0), |reorder| reorder.timing(config.framerate)); + let limit = crate::container::decode::MAX_REORDER; + let lookahead = lookahead + .max( + config + .jitter + .map_or(0, |jitter| (jitter.as_nanos() * 90_000 / 1_000_000_000) as u64), + ) + .min(limit); + (floor.min(limit), lookahead) + }); + while let Some((mut arrival, dts)) = self.clock.pop(lookahead, floor, flush) { + if !self.finished && !flush { + arrival.arrived = arrival.arrived.max(self.empty); + } + arrival.decode = moq_net::Timestamp::from_scale(dts, 90_000)?; + jitter.push(name.to_owned(), arrival)?; + } + Ok(()) + } +} + #[derive(Clone, Copy, PartialEq, Eq)] enum TrackKind { Video, @@ -162,7 +205,9 @@ impl Export { Self { source, catalog: Some(catalog), - max_delay: Duration::ZERO, + delay: Duration::ZERO, + jitter: crate::jitter::Buffer::new(Duration::ZERO), + held: None, fragment_duration: None, tracks: HashMap::new(), catalog_snapshot: None, @@ -171,13 +216,14 @@ impl Export { } } - /// Set the max delay for each per-track source. + /// Release frames a fixed delay after their decode time, dropping late frames. /// - /// See [`Consumer`](crate::container::Consumer) for the per-track skip behavior. - /// Defaults to - /// [`Duration::ZERO`] (skip aggressively). - pub fn with_max_delay(mut self, max_delay: Duration) -> Self { - self.max_delay = max_delay; + /// The same delay bounds each source's staleness. Zero (the default) releases each frame as + /// soon as it is read and drops none as late, but a stalled group is still skipped once a + /// newer one is ready. + pub fn with_delay(mut self, delay: Duration) -> Self { + self.delay = delay; + self.jitter = crate::jitter::Buffer::new(delay); self } @@ -214,43 +260,8 @@ impl Export { } } - // 2. Pull frames from each track into `pending`. ExportSource has - // already transformed Annex-B payloads (Avc3/Hev1) into length-prefixed - // form and absorbed any parameter-only frames before returning. - // - // Pre-header: drop slices that arrived before this track's codec config - // is ready. A receiver who subscribed mid-GOP can't render those bytes - // without the header anyway, and parking them would stop us from - // polling for the next SPS/PPS-bearing frame. - let waiting_for_header = !self.header_emitted; - for (name, track) in &mut self.tracks { - if track.pending.is_some() || track.finished { - continue; - } - loop { - match track.source.poll_read(waiter)? { - Poll::Ready(Some(frame)) => { - let geometry_ready = track.kind == TrackKind::Audio - || self - .catalog_snapshot - .as_ref() - .and_then(|catalog| catalog.video.renditions.get(name)) - .is_some_and(|config| track.source.video_geometry_ready(config)); - if waiting_for_header && (!track.source.header_ready() || !geometry_ready) { - // Drop this slice and keep polling for codec configuration or geometry. - continue; - } - track.pending = Some(frame); - break; - } - Poll::Ready(None) => { - track.finished = true; - break; - } - Poll::Pending => break, - } - } - } + self.jitter.expect(self.tracks.keys().cloned()); + self.fill(waiter)?; // 3. Before the header is emitted: keep pulling until every video // transform has produced its codec config. Frames arriving in this @@ -273,15 +284,20 @@ impl Export { return Poll::Pending; } - // 5. Pick the smallest-timestamp pending frame across tracks and route it - // through the cluster builder. - if let Some(name) = self.pick_next_track() { - let frame = self.tracks.get_mut(&name).unwrap().pending.take().unwrap(); + loop { + let (name, frame) = match self.held.take() { + Some(held) => held, + None => match self.jitter.poll_next(waiter) { + Poll::Ready(ready) => (ready.track, ready.item), + Poll::Pending => break, + }, + }; if let Some(chunk) = self.feed_frame(&name, frame)? { return Poll::Ready(Ok(Some(chunk))); } - // Frame consumed into the open cluster; loop and see if there's more to do. - return self.poll_next(waiter); + } + if !self.jitter.is_empty() { + return Poll::Pending; } // 6. End-of-stream: flush any open cluster when every subscribed track @@ -305,6 +321,93 @@ impl Export { Poll::Pending } + #[cfg(test)] + pub(super) fn next_deadline(&self) -> Option { + self.jitter.next_deadline() + } + + /// How many frames the export holds between its sources and the output. + #[cfg(test)] + pub(super) fn buffered(&self) -> usize { + self.jitter.len() + } + + #[cfg(test)] + pub(super) fn with_replay(mut self) -> Self { + self.jitter = self.jitter.replay(); + self + } + + /// Read each track until it has a frame due, preserving ingress timing before buffering. + fn fill(&mut self, waiter: &kio::Waiter) -> Result<()> { + for (name, track) in &mut self.tracks { + loop { + if self.header_emitted && self.jitter.full(name) { + break; + } + if !self.header_emitted && track.pending.is_some() { + break; + } + let arrival = if let Some(pending) = track.pending.take() { + pending + } else { + if track.finished { + break; + } + let frame = match track.source.poll_read(waiter)? { + Poll::Ready(Some(frame)) => frame, + Poll::Ready(None) => { + track.finished = true; + break; + } + Poll::Pending => { + track.empty = web_async::time::Instant::now(); + break; + } + }; + let geometry_ready = track.kind == TrackKind::Audio + || self + .catalog_snapshot + .as_ref() + .and_then(|catalog| catalog.video.renditions.get(name)) + .is_some_and(|config| track.source.video_geometry_ready(config)); + if !self.header_emitted && (!track.source.header_ready() || !geometry_ready) { + continue; + } + crate::jitter::Arrival { + arrived: track.empty, + read: web_async::time::Instant::now(), + decode: frame.timestamp, + restart: track.source.markers(), + skip: track.source.skips(), + sync: track.kind == TrackKind::Audio || frame.keyframe, + item: frame, + } + }; + if !self.header_emitted { + track.pending = Some(arrival); + break; + } + if track.kind == TrackKind::Video && !self.delay.is_zero() { + if arrival.restart != track.restart { + track.release(name, &mut self.jitter, true)?; + track.clock = Default::default(); + track.restart = arrival.restart; + } + let pts = (arrival.item.timestamp.as_nanos() * 90_000 / 1_000_000_000) as u64; + track.clock.push(arrival, pts); + track.release(name, &mut self.jitter, false)?; + } else { + self.jitter.push(name.clone(), arrival)?; + } + } + if track.finished { + track.release(name, &mut self.jitter, true)?; + } + } + Ok(()) + } + fn update_catalog(&mut self, mut catalog: Catalog) -> Result<()> { self.source.retain_valid_media(&mut catalog); @@ -329,6 +432,18 @@ impl Export { return Err(Error::HeaderRemovedTrack(name.clone()).into()); } } + for (name, config) in &catalog.video.renditions { + if let Some(track) = self.tracks.get_mut(name) { + if track + .config + .as_ref() + .is_some_and(|old| old.jitter != config.jitter || old.framerate != config.framerate) + { + track.reorder = None; + } + track.config = Some(config.clone()); + } + } self.catalog_snapshot = Some(catalog); return Ok(()); } @@ -340,14 +455,19 @@ impl Export { continue; } ensure_legacy(&config.container, "video", name)?; - let Some(source) = ExportSource::for_video(&self.source, name, config, self.max_delay)? else { + let Some(source) = ExportSource::for_video(&self.source, name, config, self.delay)? else { continue; }; self.tracks.insert( name.clone(), MkvTrack { + clock: Default::default(), + config: Some(config.clone()), + restart: 0, + reorder: None, source, pending: None, + empty: web_async::time::Instant::now(), finished: false, track_number: next_track_number, kind: TrackKind::Video, @@ -361,14 +481,19 @@ impl Export { continue; } ensure_legacy(&config.container, "audio", name)?; - let Some(source) = ExportSource::for_audio(&self.source, name, config, self.max_delay)? else { + let Some(source) = ExportSource::for_audio(&self.source, name, config, self.delay)? else { continue; }; self.tracks.insert( name.clone(), MkvTrack { + clock: Default::default(), + config: None, + restart: 0, + reorder: None, source, pending: None, + empty: web_async::time::Instant::now(), finished: false, track_number: next_track_number, kind: TrackKind::Audio, @@ -471,14 +596,6 @@ impl Export { Ok(Bytes::from(dest.into_inner())) } - fn pick_next_track(&self) -> Option { - self.tracks - .iter() - .filter_map(|(n, t)| t.pending.as_ref().map(|f| (n.clone(), f.timestamp))) - .min_by_key(|(_, ts)| *ts) - .map(|(n, _)| n) - } - /// Route an already-transformed frame through the cluster builder. Returns /// a chunk if the cluster rolled over (the returned chunk is the /// *previous* cluster; the new frame becomes the first block of a new @@ -503,7 +620,7 @@ impl Export { // A reconfiguration can leave a cluster open. Roll it out first and // retry this frame on the next poll. if let Some(open) = self.cluster.take() { - track.pending = Some(frame); + self.held = Some((name.to_string(), frame)); return Ok(Some(open.finish())); } let mut cluster = ClusterBuilder::new(frame_ticks); diff --git a/rs/moq-mux/src/container/mkv/export_test.rs b/rs/moq-mux/src/container/mkv/export_test.rs index ed46c8f3fc..2baaddb8ee 100644 --- a/rs/moq-mux/src/container/mkv/export_test.rs +++ b/rs/moq-mux/src/container/mkv/export_test.rs @@ -425,7 +425,8 @@ async fn export_emits_blocks_for_each_frame() { // Use per-frame clustering so each frame is observable as its own // Cluster chunk; batching is exercised in a dedicated test below. .with_fragment_duration(std::time::Duration::ZERO) - .with_max_delay(RECORDING_MAX_DELAY); + .with_delay(RECORDING_MAX_DELAY) + .with_replay(); let mut exported: Vec = Vec::new(); let mut importer = Some(importer); @@ -436,8 +437,11 @@ async fn export_emits_blocks_for_each_frame() { Ok(Ok(None)) => break, Ok(Err(e)) => panic!("exporter error: {e}"), Err(_) => { - // Drop the importer to close the broadcast so the exporter can EOS. - importer = None; + if let Some(deadline) = exporter.next_deadline() { + tokio::time::advance(deadline.saturating_duration_since(web_async::time::Instant::now())).await; + } else { + importer = None; + } } } } @@ -664,7 +668,8 @@ async fn export_fragment_duration_batches_blocks() { .expect("catalog consumer"); let mut exporter = crate::container::mkv::Export::new(crate::source::announced(&consumer).await, catalog_stream) .with_fragment_duration(std::time::Duration::from_secs(2)) - .with_max_delay(RECORDING_MAX_DELAY); + .with_delay(RECORDING_MAX_DELAY) + .with_replay(); let mut exported: Vec = Vec::new(); let mut importer = Some(importer); @@ -675,7 +680,11 @@ async fn export_fragment_duration_batches_blocks() { Ok(Ok(None)) => break, Ok(Err(e)) => panic!("exporter error: {e}"), Err(_) => { - importer = None; + if let Some(deadline) = exporter.next_deadline() { + tokio::time::advance(deadline.saturating_duration_since(web_async::time::Instant::now())).await; + } else { + importer = None; + } } } } @@ -881,3 +890,197 @@ fn synth_webm() -> Vec { } dest.into_inner() } + +/// A live container export holds a frame for the configured release delay. +#[tokio::test(start_paused = true)] +async fn configured_delay_holds_live_frames() { + use crate::container::test_util::{Live, raw_frame}; + use hang::catalog::{AAC, AudioConfig}; + let mut config = AudioConfig::new(AAC { profile: 2 }, 44100, 2); + config.description = Some(Bytes::from_static(&[0x12, 0x10])); + let mut live = Live::audio(config); + let mut export = super::Export::new(live.source().await, live.catalog_stream().await) + .with_delay(std::time::Duration::from_millis(100)) + .with_fragment_duration(Some(std::time::Duration::ZERO)); + assert!(export.next().await.unwrap().is_some(), "container header"); + live.track.write(raw_frame(0, &[0xde, 0xad], true)).unwrap(); + assert!( + tokio::time::timeout(std::time::Duration::from_millis(1), export.next()) + .await + .is_err(), + "the release delay must hold a live frame instead of emitting on arrival" + ); +} + +/// Release deadlines survive source EOS, and expired frames never reach the container. +#[tokio::test(start_paused = true)] +async fn delay_drops_late_frames_and_drains_the_final_frame() { + use crate::container::test_util::{Live, raw_frame}; + use hang::catalog::{AAC, AudioConfig}; + let mut config = AudioConfig::new(AAC { profile: 2 }, 44100, 2); + config.description = Some(Bytes::from_static(&[0x12, 0x10])); + let mut live = Live::audio(config); + let delay = std::time::Duration::from_millis(100); + let mut export = super::Export::new(live.source().await, live.catalog_stream().await) + .with_delay(delay) + .with_fragment_duration(std::time::Duration::ZERO); + export.next().await.unwrap().unwrap(); + let start = web_async::time::Instant::now(); + live.track.write(raw_frame(0, b"first", true)).unwrap(); + let first = export.next().await.unwrap().unwrap(); + assert!(first.windows(5).any(|part| part == b"first")); + assert!(start.elapsed() >= delay); + + // This frame's 120ms deadline has already passed. + tokio::time::advance(std::time::Duration::from_millis(200)).await; + assert!(export.poll_next(&kio::Waiter::noop()).is_pending()); + live.track.write(raw_frame(20_000, b"expired", true)).unwrap(); + live.track.write(raw_frame(400_000, b"final", true)).unwrap(); + live.track.finish().unwrap(); + live.catalog.finish().unwrap(); + let final_frame = export.next().await.unwrap().unwrap(); + assert!(final_frame.windows(5).any(|part| part == b"final")); + assert!(!final_frame.windows(7).any(|part| part == b"expired")); + assert!(start.elapsed() >= std::time::Duration::from_millis(500)); + assert!(export.next().await.unwrap().is_none()); +} + +#[tokio::test(start_paused = true)] +async fn decode_order_is_independent_of_cross_track_arrival_order() { + use crate::container::test_util::{Live, raw_frame}; + use hang::catalog::{AAC, AudioConfig, Container}; + for reversed in [false, true] { + let mut config = AudioConfig::new(AAC { profile: 2 }, 44100, 2); + config.description = Some(Bytes::from_static(&[0x12, 0x10])); + config.container = Container::Legacy; + let mut live = Live::audio(config.clone()); + let mut other = live.add_track("other", |catalog, name| { + catalog.modify().unwrap().audio.renditions.insert(name, config); + }); + let mut export = super::Export::new(live.source().await, live.catalog_stream().await) + .with_delay(std::time::Duration::from_millis(100)) + .with_fragment_duration(std::time::Duration::ZERO); + export.next().await.unwrap().unwrap(); + live.track.write(raw_frame(0, b"initial-a", true)).unwrap(); + other.write(raw_frame(0, b"initial-b", true)).unwrap(); + export.next().await.unwrap().unwrap(); + export.next().await.unwrap().unwrap(); + if reversed { + other.write(raw_frame(100_000, b"earlier", true)).unwrap(); + } else { + live.track.write(raw_frame(200_000, b"later", true)).unwrap(); + } + assert!(export.poll_next(&kio::Waiter::noop()).is_pending()); + tokio::time::advance(std::time::Duration::from_millis(20)).await; + if reversed { + live.track.write(raw_frame(200_000, b"later", true)).unwrap(); + } else { + other.write(raw_frame(100_000, b"earlier", true)).unwrap(); + } + let first = export.next().await.unwrap().unwrap(); + let second = export.next().await.unwrap().unwrap(); + assert!(first.windows(7).any(|part| part == b"earlier")); + assert!(second.windows(5).any(|part| part == b"later")); + } +} + +#[tokio::test(start_paused = true)] +async fn reordered_video_uses_decode_deadlines_but_keeps_block_pts() { + use crate::container::test_util::{Live, raw_frame, video_frame}; + use hang::catalog::{AAC, AudioConfig, Container}; + let mut live = Live::avc3(); + for config in live.catalog.modify().unwrap().video.renditions.values_mut() { + config.jitter = Some(std::time::Duration::from_millis(120)); + } + let mut audio = live.add_track("audio", |catalog, name| { + let mut config = AudioConfig::new(AAC { profile: 2 }, 44100, 2); + config.description = Some(Bytes::from_static(&[0x12, 0x10])); + config.container = Container::Legacy; + catalog.modify().unwrap().audio.renditions.insert(name, config); + }); + for (index, timestamp) in [1000, 1120, 1040, 1080].into_iter().enumerate() { + live.track.write(video_frame(timestamp * 1000, index == 0)).unwrap(); + } + audio.write(raw_frame(1_060_000, b"audio", true)).unwrap(); + live.track.finish().unwrap(); + audio.finish().unwrap(); + live.catalog.finish().unwrap(); + let mut export = super::Export::new(live.source().await, live.catalog_stream().await) + .with_delay(std::time::Duration::from_secs(5)) + .with_fragment_duration(std::time::Duration::ZERO) + .with_replay(); + export.next().await.unwrap().unwrap(); + let mut timestamps = Vec::new(); + while let Some(chunk) = export.next().await.unwrap() { + let mut cursor = Cursor::new(chunk); + let mut base = 0; + for tag in WebmIterator::new(&mut cursor, &[]) { + match tag.unwrap() { + MatroskaSpec::Timestamp(timestamp) => base = timestamp as i64, + MatroskaSpec::SimpleBlock(bytes) => { + let block = SimpleBlock::try_from(bytes.as_slice()).unwrap(); + timestamps.push(base + i64::from(block.timestamp)); + } + _ => {} + } + } + } + assert_eq!(timestamps, [1000, 1120, 1040, 1060, 1080]); +} + +/// A backlog the source already holds, as after the sink stalls, goes out a frame at a time +/// as each falls due, rather than being read into the export whole first. +#[tokio::test(start_paused = true)] +async fn a_backlog_stays_in_the_source_until_due() { + use crate::container::test_util::{Live, raw_frame}; + use hang::catalog::{AAC, AudioConfig}; + for delay in [std::time::Duration::ZERO, std::time::Duration::from_millis(100)] { + let mut config = AudioConfig::new(AAC { profile: 2 }, 44100, 2); + config.description = Some(Bytes::from_static(&[0x12, 0x10])); + let mut live = Live::audio(config); + let mut export = super::Export::new(live.source().await, live.catalog_stream().await) + .with_delay(delay) + .with_fragment_duration(std::time::Duration::ZERO); + export.next().await.unwrap().unwrap(); + live.track.write(raw_frame(0, b"first", true)).unwrap(); + export.next().await.unwrap().unwrap(); + for index in 1..=50 { + tokio::time::advance(std::time::Duration::from_millis(20)).await; + live.track.write(raw_frame(index * 20_000, b"backlog", false)).unwrap(); + } + for _ in 1..=50 { + let cluster = export.next().await.unwrap().unwrap(); + assert!(cluster.windows(7).any(|part| part == b"backlog")); + assert!( + export.buffered() <= 1, + "{delay:?}: {} frames buffered", + export.buffered() + ); + } + } +} + +/// A joiner reads the retained group to its live edge to anchor the clock, then holds no +/// more than the delay of it. +#[tokio::test(start_paused = true)] +async fn a_retained_group_trims_to_the_delay() { + use crate::container::test_util::{Live, raw_frame}; + use hang::catalog::{AAC, AudioConfig}; + let mut config = AudioConfig::new(AAC { profile: 2 }, 44100, 2); + config.description = Some(Bytes::from_static(&[0x12, 0x10])); + let mut live = Live::audio(config); + for index in 0..50 { + live.track + .write(raw_frame(index * 20_000, b"retained", index == 0)) + .unwrap(); + tokio::time::advance(std::time::Duration::from_millis(20)).await; + } + let delay = std::time::Duration::from_millis(100); + let mut export = super::Export::new(live.source().await, live.catalog_stream().await) + .with_delay(delay) + .with_fragment_duration(std::time::Duration::ZERO); + export.next().await.unwrap().unwrap(); + export.next().await.unwrap().unwrap(); + // What is left spans no more than the delay: five 20 ms frames. + assert!(export.buffered() <= 5, "{} frames buffered", export.buffered()); +} diff --git a/rs/moq-mux/src/container/mod.rs b/rs/moq-mux/src/container/mod.rs index 440859d861..b15f5d8839 100644 --- a/rs/moq-mux/src/container/mod.rs +++ b/rs/moq-mux/src/container/mod.rs @@ -15,6 +15,7 @@ use std::task::Poll; use bytes::Bytes; mod consumer; +mod decode; mod group; mod producer; mod source; diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index 7ebe9d2d40..e31341e8df 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -42,6 +42,7 @@ use crate::catalog::hang::Catalog; use crate::catalog::{CatalogFormat, Stream}; use crate::codec::video::{Hrd, Reorder}; use crate::codec::{aac, annexb, opus}; +use crate::container::decode::{DecodeClock, MAX_REORDER}; use crate::container::{ExportSource, Frame}; use crate::jitter::{self, Arrival, Push}; @@ -493,15 +494,7 @@ impl Timing { let Some(declared) = self.declared else { return (0, 0); }; - let period = match declared.period { - Some((units, scale)) => units.saturating_mul(90_000).div_ceil(scale), - None => self.framerate.map_or(0, |fps| (90_000.0 / fps).ceil() as u64), - }; - let delay = u64::from(declared.depth).saturating_mul(period); - let lookahead = 1u64 - .checked_shl(declared.depth) - .unwrap_or(u64::MAX) - .saturating_mul(period); + let (delay, lookahead) = declared.timing(self.framerate); (delay.min(MAX_REORDER), lookahead.min(MAX_REORDER)) } } @@ -2278,11 +2271,6 @@ fn pcr_packet(pid: u16, pcr: u64) -> anyhow::Result> { const PES_OPTIONAL_LEN: usize = 3 + 5; /// Extra bytes when the optional region also carries a DTS (5 DTS bytes). const PES_DTS_LEN: usize = 5; -/// Largest reorder delay or lookahead an SPS or observed reordering can raise a video decode -/// clock to (2 s, in 90 kHz ticks): far past any real reorder depth, so a corrupt SPS or a -/// timestamp stepping back within a timeline cannot hold frames or drag the DTS back without -/// bound. -const MAX_REORDER: u64 = 2 * 90_000; /// Whether `timestamp` has crossed into a later repetition slot than `last`. /// @@ -2900,101 +2888,6 @@ fn ensure_raw(container: &Container, kind: &str, name: &str) -> anyhow::Result<( } } -/// A video rendition's decode clock: a decode timestamp (DTS) for each frame from its PTS. -/// -/// [`Frame`] carries only a presentation timestamp and frames reach the muxer in decode order, -/// so a B-frame stream arrives with non-monotonic PTS and no decode time. An encoder decodes -/// the n-th frame in decode order at the n-th presentation time, a fixed reorder delay early, -/// so that is the DTS authored here: with decode-order PTS 0, 120, 40, 80 and a delay of one -/// 40 ms frame, the frames decode at -40, 0, 40, 80. It keeps the encoder's own spacing, a -/// frame's or a field's, through any switch between frame and field coding, and the delay is -/// a time, the least that keeps every DTS at or before its PTS (or the SPS's declared depth, -/// when that is more). On a source muxed that way this is its DTS, frame for frame. -/// -/// A frame's slot is settled once no frame still to come can be presented before it. A later -/// frame decodes after this one, and is presented at most the catalog `jitter` after it -/// decodes, so the slot is settled once the highest PTS read is that far past it. Frames wait -/// here until then, as long as the reordering spans and well inside the jitter buffer's delay. -/// Without a `jitter`, the SPS's depth bounds it, and so does how far a frame was presented -/// below the highest PTS read before it. A frame that still comes in below a settled slot -/// (reordering deeper than all of those) is nudged one tick past the last DTS, so the clock -/// stays strictly increasing. -/// -/// Ticks are continuous (unwrapped) 90 kHz, so the clock never wraps mid-stream; the 33-bit -/// wire wrap happens once at emission. -struct DecodeClock { - /// Frames read but not yet given a DTS, in decode order, with their PTS. - held: VecDeque<(T, u64)>, - /// The held frames' PTS, ascending: the slots they decode in, in turn. - slots: Vec, - /// The highest PTS read. - high: Option, - /// The furthest a frame was presented below the highest PTS read before it, plus a tick. - reach: u64, - /// The reorder delay in effect. It never shrinks: that would step the clock back. - delay: u64, - /// The last DTS handed out. - last: Option, -} - -impl Default for DecodeClock { - fn default() -> Self { - Self { - held: VecDeque::new(), - slots: Vec::new(), - high: None, - reach: 0, - delay: 0, - last: None, - } - } -} - -impl DecodeClock { - /// Hold the next frame in decode order, presented at `pts`. - fn push(&mut self, item: T, pts: u64) { - if let Some(high) = self.high - && pts < high - { - self.reach = self.reach.max(high - pts + 1).min(MAX_REORDER); - } - self.high = Some(self.high.map_or(pts, |high| high.max(pts))); - let at = self.slots.partition_point(|&slot| slot <= pts); - self.slots.insert(at, pts); - self.held.push_back((item, pts)); - } - - /// The next frame and its DTS, once its slot is settled `lookahead` past (the catalog - /// `jitter`), or at once when `flush`. `floor` is the least reorder delay, in ticks. - fn pop(&mut self, lookahead: u64, floor: u64, flush: bool) -> Option<(T, u64)> { - let high = self.high?; - let lookahead = lookahead.max(self.reach); - let settled = |slot: u64| flush || high >= slot.saturating_add(lookahead); - if !settled(*self.slots.first()?) { - return None; - } - // Every settled frame bounds the delay, so one decoding later never steps it back. - let mut delay = self.delay.max(floor); - for (&slot, (_, pts)) in self.slots.iter().zip(&self.held) { - if !settled(slot) { - break; - } - delay = delay.max(slot.saturating_sub(*pts)); - } - self.delay = delay.min(MAX_REORDER); - - let slot = self.slots.remove(0); - let (item, _) = self.held.pop_front()?; - let dts = slot.saturating_sub(self.delay); - let dts = match self.last { - Some(last) if dts <= last => last + 1, - _ => dts, - }; - self.last = Some(dts); - Some((item, dts)) - } -} - /// The HRD declared by the first SPS in a video rendition's avcC/hvcC. fn declared_hrd(stream_type: StreamType, description: &[u8]) -> Option { match stream_type { diff --git a/rs/moq-mux/src/jitter.rs b/rs/moq-mux/src/jitter.rs index 5f4fa61a94..ce998d2cdf 100644 --- a/rs/moq-mux/src/jitter.rs +++ b/rs/moq-mux/src/jitter.rs @@ -99,6 +99,9 @@ pub(crate) enum Push { /// /// A zero delay holds nothing and drops nothing: each frame goes out as soon as it is /// read, ordered only among the frames read together. +/// +/// A caller reads a track only while it is not [`Buffer::full`], so a backlog stays in the +/// source and goes out as it falls due, rather than being queued here whole first. pub(crate) struct Buffer { delay: Duration, /// Anchor each clock on the first frame rather than acquire it: a test exporting a @@ -712,6 +715,22 @@ impl Buffer { .min() } + /// Whether `key` has a frame waiting that is already due (any frame, without a delay), so + /// reading more of it now would only queue what the output cannot take yet. + pub fn full(&self, key: &K) -> bool { + self.tracks + .get(key) + .and_then(|track| track.queue.front()) + .is_some_and(|(_, _, deadline, _)| self.delay.is_zero() || *deadline <= Instant::now()) + } + + /// How many frames are waiting. + #[cfg(test)] + pub fn len(&self) -> usize { + let held = self.acquire.as_ref().map_or(0, |acquire| acquire.frames.len()); + held + self.tracks.values().map(|track| track.queue.len()).sum::() + } + /// Whether no frame is waiting. pub fn is_empty(&self) -> bool { self.acquire.is_none() && self.tracks.values().all(|track| track.queue.is_empty()) @@ -1440,4 +1459,24 @@ mod tests { assert_eq!(due(&mut buffer), ["a40"]); assert_eq!(buffer.dropped(), 0); } + + /// A track is full once a frame of it is due, or holds any frame without a delay, so its + /// reader leaves the rest in the source. + #[tokio::test(start_paused = true)] + async fn a_track_is_full_once_a_frame_is_due() { + let start = Instant::now(); + let mut buffer = Buffer::new(DELAY).replay(); + assert!(!buffer.full(&1)); + buffer.push(1, arrival(start, 0, "v0")).unwrap(); + assert!(!buffer.full(&1), "a frame not yet due leaves room to read ahead"); + tokio::time::advance(DELAY).await; + assert!(buffer.full(&1)); + assert!(!buffer.full(&2), "another track's due frame does not stop this one"); + assert_eq!(due(&mut buffer), ["v0"]); + assert!(!buffer.full(&1)); + + let mut buffer = Buffer::new(Duration::ZERO); + buffer.push(1, arrival(start, 0, "v0")).unwrap(); + assert!(buffer.full(&1)); + } } diff --git a/rs/moq-rtmp/src/dial.rs b/rs/moq-rtmp/src/dial.rs index 7de54d51b9..df737f0da8 100644 --- a/rs/moq-rtmp/src/dial.rs +++ b/rs/moq-rtmp/src/dial.rs @@ -162,10 +162,9 @@ impl Client { }) } - /// Set how long [`publish`](Self::publish)'s FLV muxer waits for a stalled group - /// before skipping to a newer one (the moq-level frame-drop latency). Defaults - /// to [`DEFAULT_MAX_DELAY`](crate::DEFAULT_MAX_DELAY); pass - /// [`Duration::ZERO`] to drop stale groups aggressively. + /// Set the FLV release delay and stalled-group budget for [`publish`](Self::publish). + /// Defaults to [`DEFAULT_MAX_DELAY`](crate::DEFAULT_MAX_DELAY); zero releases + /// available frames immediately without deadline drops. pub fn with_export_max_delay(mut self, max_delay: Duration) -> Self { self.export_max_delay = max_delay; self @@ -236,7 +235,7 @@ impl Client { .catalog::<()>(moq_mux::catalog::CatalogFormat::default()) .await .map_err(|e| anyhow::anyhow!("init FLV export: {e}"))?; - let mut export = FlvExport::new(source, catalog).with_max_delay(self.export_max_delay); + let mut export = FlvExport::new(source, catalog).with_delay(self.export_max_delay); let mut tags = flv::TagReader::new(); let mut buffer = [0u8; READ_BUFFER]; diff --git a/rs/moq-rtmp/src/lib.rs b/rs/moq-rtmp/src/lib.rs index 3ebb95a9db..3124966c8a 100644 --- a/rs/moq-rtmp/src/lib.rs +++ b/rs/moq-rtmp/src/lib.rs @@ -79,15 +79,11 @@ mod metadata; mod rml; mod server; -/// Default FLV muxer latency: how long a stalled group is held before skipping to -/// a newer one (the moq-level frame-drop latency). +/// Default FLV release delay and stalled-group budget. /// -/// RTMP is unpaced (tags go out as fast as the socket accepts them), so this -/// bounds buffering rather than the wire rate. Zero would drop stale groups the -/// instant a newer one arrives, which is too aggressive for RTMP's typical jitter, -/// so the default holds a couple of seconds. Override per request with -/// [`Play::with_max_delay`] / [`Client::with_export_max_delay`] or via -/// [`Config::export_max_delay`]. +/// Frames are released in decode order after this delay; late video resumes at +/// its next keyframe. Override per request with [`Play::with_max_delay`] / +/// [`Client::with_export_max_delay`] or via [`Config::export_max_delay`]. pub const DEFAULT_MAX_DELAY: Duration = Duration::from_secs(2); pub use dial::Client; diff --git a/rs/moq-rtmp/src/listen.rs b/rs/moq-rtmp/src/listen.rs index 4f614db8e4..afe497b78c 100644 --- a/rs/moq-rtmp/src/listen.rs +++ b/rs/moq-rtmp/src/listen.rs @@ -45,10 +45,9 @@ pub struct Config { /// namespace all of its streams (e.g. `live`). pub prefix: PathOwned, - /// How long a play's FLV muxer waits for a stalled group before skipping to a - /// newer one (the moq-level frame-drop latency). Defaults to - /// [`DEFAULT_MAX_DELAY`](crate::DEFAULT_MAX_DELAY); set [`Duration::ZERO`] - /// to drop stale groups aggressively. Only affects egress (plays); ingest ignores it. + /// FLV release delay and stalled-group budget for plays. + /// Defaults to [`DEFAULT_MAX_DELAY`](crate::DEFAULT_MAX_DELAY); zero releases + /// available frames immediately without deadline drops. Ingest ignores it. pub export_max_delay: Duration, /// How long relays keep a non-latest group of an ingested media track fetchable, or diff --git a/rs/moq-rtmp/src/server.rs b/rs/moq-rtmp/src/server.rs index 8b64c15d27..73096f5548 100644 --- a/rs/moq-rtmp/src/server.rs +++ b/rs/moq-rtmp/src/server.rs @@ -746,12 +746,9 @@ impl Play { self.peer } - /// Set how long the FLV muxer waits for a stalled group before skipping to a - /// newer one (the moq-level frame-drop latency). Defaults to - /// [`DEFAULT_MAX_DELAY`](crate::DEFAULT_MAX_DELAY). RTMP is unpaced (tags go out as - /// fast as the socket accepts them), so this bounds buffering, not the wire - /// rate. Pass [`Duration::ZERO`] to drop stale groups - /// aggressively. + /// Set the FLV release delay and stalled-group budget. + /// Defaults to [`DEFAULT_MAX_DELAY`](crate::DEFAULT_MAX_DELAY); zero releases + /// available frames immediately without deadline drops. pub fn with_max_delay(mut self, latency: Duration) -> Self { self.latency = latency; self @@ -838,7 +835,7 @@ impl Play { .map_err(|e| anyhow::anyhow!("init FLV export: {e}"))? .select(select); let mut export = FlvExport::new(source, stream) - .with_max_delay(self.latency) + .with_delay(self.latency) .with_multitrack(self.capabilities.multitrack); // Resolve the catalog and codec headers before Play.Start, too. Otherwise a