diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 9f2ca99e294..d127621494c 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -131,8 +131,11 @@ Other video waits for its first keyframe. MPEG-TS export pads to the source's constant mux rate when the catalog recorded one, or to `--mux-rate`, on a constant-rate schedule an IRD or groomer can lock to. Each frame goes out as early as the receiver's buffers admit, up -to `--delay` ahead of its decode time on the output's clock, so the output trails the source by twice -the delay; a frame that cannot arrive in time at the rate fails the export. +to `--delay` ahead of its decode time and no further than its decoder buffer +holds. That send-ahead comes out of the delay, so the output trails a TS source +by `--delay`. A frame that cannot arrive in time at the rate fails the export, +as does video whose HRD declares a longer buffer delay than `--delay`; the +error names the `--delay` it needs. ## Play @@ -353,14 +356,17 @@ may fall behind the live edge before *this* consumer skips it. Raising 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`, `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 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. +`export ts`, `export flv`, and `export mkv` take `--delay` instead, like an SRT +receiver's latency, all tracks in decode order, so two exporters of one broadcast +emit the same order. FLV and MKV mux each frame that long after its decode time +(default 500 ms). TS output runs that far behind the source (default 1 s): the +send-ahead that fills a receiver's buffers comes out of the same budget, and 1 s +fits a broadcast-sized CPB. A source that sends no track ahead of its decode +time, as RTMP and browser publishers do, trails by the video's send-ahead on +top. A frame arriving after its deadline is dropped, and video resumes at its +next 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. The output follows the source's clock within what ISO/IEC 13818-1 allows a TS PCR. A source further off fails the export, naming its rate, once diff --git a/doc/bin/srt.md b/doc/bin/srt.md index 3188573aef0..7f7ee1efc6a 100644 --- a/doc/bin/srt.md +++ b/doc/bin/srt.md @@ -44,7 +44,8 @@ moq --connect https://relay.example.com/anon --broadcast event.hang import srt - ``` `--latency` (default 500 ms) sets the SRT receive buffer and doubles as the -export's jitter buffer, like `export ts --delay` in the [CLI](/bin/cli#export): -a frame arriving later than that is dropped. A `--connect` URL needs a +export's delay, like `export ts --delay` in the [CLI](/bin/cli#export), but no +less than that flag's 1 s default: the send-ahead that fills a receiver's +buffers comes out of the delay. A frame arriving later than that is dropped. A `--connect` URL needs a `streamid` query or a path; a listener bridges one `--broadcast` and ignores the stream id it is offered. The library is [`moq-srt`](https://docs.rs/moq-srt). diff --git a/doc/setup/upgrade.md b/doc/setup/upgrade.md index 2cfe780628a..dd5c605fdc7 100644 --- a/doc/setup/upgrade.md +++ b/doc/setup/upgrade.md @@ -404,6 +404,16 @@ These land with the next breaking release, not the 2026-09-23 train. `next()` and `for await` over a snapshot consumer rather than relying on type errors. On an untimed track nothing marks a group stale, so a `next()` reader that falls behind replays the whole backlog where it used to skip it. +- **`moq export ts --delay` defaults to 1 s and includes the send-ahead.** Each + frame still goes out up to the delay ahead of its decode time, now no further + than its decoder buffer holds, and that comes out of the same budget: the + output trails a TS source by the delay instead of twice it. With a multiplex + rate, video whose HRD declares a longer buffer delay than `--delay` stops the + export with the delay it needs named, so a broadcast feed run at + `--delay 500ms` needs `--delay 1s`. A source that sends every track at its + decode time (RTMP, a browser) trails by the delay plus the video's + send-ahead: 2 s at the new default, where `--delay 500ms` keeps the 1 s it + had. moq-srt egress runs at its SRT latency or 1 s, whichever is longer. ## Wire diff --git a/quest/m1/tstd/README.md b/quest/m1/tstd/README.md index 5060f62cdd3..21d68223bac 100644 --- a/quest/m1/tstd/README.md +++ b/quest/m1/tstd/README.md @@ -19,7 +19,7 @@ first. Measured (2026-10-01, #4645): with the fixed-delay jitter buffer, per-PID admission against each PID's T-STD buffers, and PCRs at their byte position, a clean-path round trip passes the strict T-STD check and TSDuck's ±500 ns -pcrverify at the default 500 ms delay, for the generated clip at 10 Mb/s and +pcrverify at the then-default 500 ms delay, for the generated clip at 10 Mb/s and 2 Mb/s and for a 1080p encode filling a 9 Mbit CPB (`just test ts --hrd`, after t0ms's recipe). A unit that cannot arrive by its DTS fails the export rather than arrive late. t0ms graded it on a CNN broadcast capture (PAFF H.264, MP2, @@ -27,8 +27,7 @@ DVB AC-3, teletext, SCTE-35; 2026-10-02): at 1 s on loopback, across hosts and under the #4613 rig at 0 % and 1 % loss, every track passed T-STD and `compliance.py` with no late drops, once the clock anchored on the track sent latest. At 500 ms a 540 s run still misses a video deadline about 157 s in, -which [send-ahead](/quest/m1/tstd/send-ahead.md)'s 1 s default addresses, and -10 % loss is still to pass. +which the 1 s default `--delay` addresses, and 10 % loss is still to pass. This README owns the remaining end-to-end proof: make the #4613 netem rig (10% loss, a redistributable or reproducible ~10 Mb/s broadcast-style TS) @@ -49,7 +48,6 @@ MB/AAC controls. Do not relabel the current subset a full standards check. ## Required -- [Send-ahead within the delay](/quest/m1/tstd/send-ahead.md) - total lag is `--delay`, send-ahead included, with a 1 s default - [PSI pacing](/quest/m1/tstd/psi-runs.md) - padded output paces the PAT and PMT so TBsys and Bsys never overflow ## Related diff --git a/quest/m1/tstd/send-ahead.md b/quest/m1/tstd/send-ahead.md deleted file mode 100644 index ddaa61b589e..00000000000 --- a/quest/m1/tstd/send-ahead.md +++ /dev/null @@ -1,36 +0,0 @@ -# [M] TS export lag stays within the delay, send-ahead included - -## Goal - -`moq export ts` output trails the source by `--delay`, not twice it. The -send-ahead window that pre-loads the receiver's buffers comes out of the same -budget, capped at what the decoder buffer can hold. The default becomes 1 s, -so broadcast-sized CPBs fit out of the box. - -## Plan - -Decided (2026-10-01). Found in the fixed-delay release -(#4645): frames are held until anchor + DTS + delay, and the schedule then -sends each unit up to another delay ahead of its DTS, so a mid-group joiner -trails the source by two delays. t0ms measured about 1.7 s of delivery at -500 ms: the two delays plus the join and the source's own send-ahead, so -carving alone should cut it by about one delay, not to 0.5 s. - -- One budget. A frame's release deadline is anchor + DTS + delay − its - send-ahead. The send-ahead is at most what the PID's decoder buffer can hold - ahead of its DTS (EB from the SPS HRD or the level, B per 13818-1); sending - further ahead only adds latency (0.93 s of reach on `hrd9m` against a 2 s - delay). -- The default `--delay` rises from 500 ms to 1 s. Broadcast CPBs need - 0.6-1 s of send-ahead (t0ms's CNN capture, `hrd9m`). A source that needs - more than the delay allows fails loud and names the `--delay` it needs. - Update `doc/bin/cli.md`, moq-srt egress (which passes its SRT latency as - the delay), and every example invocation. -- Test: a mid-group joiner's every slot goes out `delay` after the source sent - it (the existing `a_joiner_runs_at_the_delay_from_its_first_output`, tightened - from two delays to one). `hrd9m` passes strict `tstd` at the 1 s default, - and a delay below its buffer delay fails loud. -- Prove the latency change with generated or redistributable fixtures and - automated assertions in CI. A third-party capture re-grade can supplement - that evidence, but is not a completion dependency or a substitute for the - automated test. diff --git a/rs/moq-cli/src/args.rs b/rs/moq-cli/src/args.rs index c64d63788fd..d89b157ba2b 100644 --- a/rs/moq-cli/src/args.rs +++ b/rs/moq-cli/src/args.rs @@ -1052,7 +1052,8 @@ impl ExportSink { /// A stdout sink's format and the options that apply to it. pub struct Stdout { pub format: SubscribeFormat, - /// The staleness budget, also used as the TS/FLV/MKV release delay. + /// The staleness budget, also the FLV/MKV release delay and how far behind the source `ts` + /// runs (`--delay`). pub max_delay: Duration, pub linger: Duration, /// Follow a replacement as a program switch (`ts` only). @@ -1152,10 +1153,11 @@ pub struct BufferedFragmented { #[derive(usage::Args, Clone)] #[usage(unknown_flags = "error", args_override_self = false)] pub struct Transport { - /// How long after its decode time each frame is written (e.g. `500ms`), like an SRT - /// receiver's latency. A frame that arrives later than this is dropped, and a stalled - /// group is skipped once it falls half this behind the newest content. - #[usage(long, default = "500ms")] + /// How far behind the source the output runs (e.g. `2s`), like an SRT receiver's latency, + /// including how far ahead of its decode time each frame is sent. A frame that arrives + /// later than this is dropped, and a stalled group is skipped once it falls half this + /// behind the newest content. + #[usage(long, default = "1s")] pub delay: crate::duration::Duration, /// How long to wait for the same publisher instance to come back once it ends (e.g. `10s`), @@ -1193,13 +1195,13 @@ pub struct Transport { impl Transport { fn deprecated(&self) -> moq_tokio::cli::Deprecated { - const HOLDS: &str = "it also holds every frame that long after its decode time"; + const BEHIND: &str = "it also sets how far behind the source the output runs"; let mut found = moq_tokio::cli::Deprecated::default(); if self.max_age.is_some() { - found.changed("--max-age", None, "--delay", HOLDS); + found.changed("--max-age", None, "--delay", BEHIND); } if self.latency_max.is_some() { - found.changed("--latency-max", None, "--delay", HOLDS); + found.changed("--latency-max", None, "--delay", BEHIND); } found } @@ -1280,7 +1282,7 @@ mod tests { }; export.sink.stdout().unwrap().max_delay }; - assert_eq!(stdout(&[]), Duration::from_millis(500)); + assert_eq!(stdout(&[]), Duration::from_secs(1)); assert_eq!(stdout(&["--delay", "2s"]), Duration::from_secs(2)); for old in ["--max-age", "--latency-max"] { let Err(err) = parse(&[old, "1s"]) else { diff --git a/rs/moq-cli/src/subscribe.rs b/rs/moq-cli/src/subscribe.rs index b6f7555d30a..9f9579d76f1 100644 --- a/rs/moq-cli/src/subscribe.rs +++ b/rs/moq-cli/src/subscribe.rs @@ -177,8 +177,8 @@ pub struct SubscribeArgs { /// The format to write to stdout. pub format: SubscribeFormat, - /// How far playback may drift from the live edge before skipping groups. TS also - /// holds every frame this long after its decode time (`--delay`). + /// How far playback may drift from the live edge before skipping groups. TS also runs + /// this far behind the source (`--delay`). pub max_delay: Duration, /// How long to wait for the broadcast to come back after it ends (TS only). diff --git a/rs/moq-mux/Cargo.toml b/rs/moq-mux/Cargo.toml index 6f04203df88..4d5fe054524 100644 --- a/rs/moq-mux/Cargo.toml +++ b/rs/moq-mux/Cargo.toml @@ -58,3 +58,7 @@ harness = false [[bench]] name = "export" harness = false + +[[bench]] +name = "jitter" +harness = false diff --git a/rs/moq-mux/benches/jitter.rs b/rs/moq-mux/benches/jitter.rs new file mode 100644 index 00000000000..66ad1492baa --- /dev/null +++ b/rs/moq-mux/benches/jitter.rs @@ -0,0 +1,61 @@ +//! The jitter buffer's clock steering swept over track count and how many floors the rate +//! window holds. +use std::time::Duration; + +use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main}; +use moq_net::Timestamp; +use web_async::time::Instant; + +#[path = "../src/jitter.rs"] +#[allow(dead_code)] +mod jitter; + +/// Every track sends a frame a second as it decodes, for as many two-second steering steps as +/// fill the rate window with `floors`, each step measuring every track heard. +async fn steer(tracks: u16, floors: u64) -> u64 { + let start = Instant::now(); + let mut buffer = jitter::Buffer::new(Duration::from_millis(500)); + buffer.expect(1..=tracks); + let waiter = kio::Waiter::noop(); + for second in 0..floors * 2 { + let at = start + Duration::from_secs(second); + tokio::time::advance(at.saturating_duration_since(Instant::now())).await; + for track in 1..=tracks { + let arrival = jitter::Arrival { + arrived: at, + read: at, + decode: Timestamp::from_millis(second * 1_000).unwrap(), + restart: 0, + skip: 0, + sync: true, + item: (), + }; + buffer.push(track, arrival).unwrap(); + } + while buffer.poll_next(&waiter).is_ready() {} + } + buffer.dropped() +} + +fn steering(c: &mut Criterion) { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_time() + .start_paused(true) + .build() + .unwrap(); + let mut group = c.benchmark_group("jitter_steering"); + group.measurement_time(Duration::from_secs(1)); + group.warm_up_time(Duration::from_millis(200)); + group.sample_size(10); + for tracks in [1u16, 4, 16, 64] { + for floors in [10u64, 100, 300] { + group.bench_function(BenchmarkId::new(tracks.to_string(), floors), |b| { + b.iter(|| assert_eq!(runtime.block_on(steer(tracks, floors)), 0)); + }); + } + } + group.finish(); +} + +criterion_group!(benches, steering); +criterion_main!(benches); diff --git a/rs/moq-mux/benches/ts_schedule.rs b/rs/moq-mux/benches/ts_schedule.rs index 09e693a0b1d..72b8f3b68a3 100644 --- a/rs/moq-mux/benches/ts_schedule.rs +++ b/rs/moq-mux/benches/ts_schedule.rs @@ -38,6 +38,7 @@ fn admission(c: &mut Criterion) { leak: None, size: Some(1_000_000), transport: 512, + reach: Duration::from_secs(1), }, ); } diff --git a/rs/moq-mux/src/codec/h264/mod.rs b/rs/moq-mux/src/codec/h264/mod.rs index 773708a498a..634a2c2c21e 100644 --- a/rs/moq-mux/src/codec/h264/mod.rs +++ b/rs/moq-mux/src/codec/h264/mod.rs @@ -747,6 +747,12 @@ pub(crate) mod fixtures { 0x67, 0xf4, 0x00, 0x15, 0x91, 0x9d, 0x02, 0x82, 0x1f, 0x89, 0xc0, 0x44, 0x00, 0x00, 0x03, 0x00, 0x04, 0x00, 0x00, 0x03, 0x00, 0xca, 0x7c, 0x50, 0xaa, 0x80, ]; + /// `bframes=1:force-cfr=1:nal-hrd=cbr:bitrate=9000:vbv-maxrate=9000:vbv-bufsize=9000`: a + /// contribution encoder's 9 Mbit CBR NAL HRD at 9 Mb/s, one second of buffer delay. + pub(crate) const SPS_HRD: &[u8] = &[ + 0x67, 0x64, 0x00, 0x1e, 0xac, 0xe4, 0x10, 0x9b, 0x01, 0x10, 0x00, 0x00, 0x03, 0x00, 0x10, 0x00, 0x00, 0x03, + 0x03, 0x2e, 0x04, 0x00, 0x00, 0x89, 0x54, 0x40, 0x00, 0x11, 0x2a, 0x8e, 0x4c, 0x30, 0x07, 0x8b, 0x16, 0x89, + ]; pub(crate) const PPS: &[u8] = &[0x68, 0xeb, 0xe3, 0xcb, 0x22, 0xc0]; } @@ -804,6 +810,28 @@ mod tests { assert_eq!(sps_framerate(&[0x67, 0x42, 0xc0, 0x1f, 0xde]), None); } + #[test] + fn sps_hrd_reads_the_nal_hrd() { + use crate::codec::video::{Hrd, Reorder}; + + assert_eq!( + sps_hrd(fixtures::SPS_HRD), + Some(Hrd { + bit_rate: 9_000_000, + cpb_size: 9_000_000 + }) + ); + // The bitstream restriction follows the HRD in the VUI, so it still reads past it. + assert_eq!( + sps_reorder(fixtures::SPS_HRD), + Some(Reorder { + depth: 1, + period: Some((2, 50)) + }) + ); + assert_eq!(sps_hrd(fixtures::SPS_IPB), None); + } + fn annexb_frame(nals: &[&[u8]]) -> Bytes { let mut buf = BytesMut::new(); for nal in nals { diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index 412536e27a5..bbd4398298f 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -11,9 +11,9 @@ //! position agrees with the values, and a pacing caller releases each slot at //! the instant it asserts. Video is carried as Annex-B, audio as ADTS AAC. //! -//! Every frame is muxed a fixed delay after its decode time ([`Export::with_delay`]), -//! so tracks interleave in `(DTS, PID)` order whatever the arrival skew between them, -//! and the output keeps the source's pace. +//! Every frame is muxed a fixed delay after its release time, its decode time less its +//! send-ahead ([`Export::with_delay`]), so tracks interleave in one order whatever the +//! arrival skew between them, and the output keeps the source's pace. //! //! Video flows through `ExportSource`, which normalizes every H.264/H.265 //! source to length-prefixed NALU plus a resolved avcC/hvcC (parsing in-band @@ -219,12 +219,20 @@ struct Queued { frame: Frame, /// When the frame decodes: its DTS, else its PTS. decode: Timestamp, + /// Where the jitter buffer orders it ([`Track::release_key`]). + key: Timestamp, /// Authored decode timestamp, as [`PesUnit::dts`]. dts: Option, /// The avcC/hvcC the frame was read under, for video. description: Option, + /// The PES it may share ([`Track::pack_start`]). + pes: Option, } +/// The PES an audio frame may share ([`Track::pack_start`]): the place of its first frame on +/// the media timeline, and its [`Track::psi_slot`]. +type PesStart = (u128, Option); + struct Track { source: ExportSource, /// The first frame, held until the program tables are built. @@ -254,6 +262,9 @@ struct Track { ac3_carry: Option, /// The audio frames muxed into a PES that has yet to go out. pack: Option, + /// The last frame's decode time, its jitter buffer key ([`Self::release_key`]) and the PES + /// it may share. + released: Option<(Timestamp, Timestamp, Option)>, } /// Shared program state needed while reading one track. @@ -281,6 +292,7 @@ impl Track { buffer: None, ac3_carry: None, pack: None, + released: None, } } @@ -450,6 +462,43 @@ impl Track { } } + /// The bytes a PES of the track has for its frames, and how many frames of `len` bytes + /// that holds: B holds the PES header with the frames (ISO 13818-1 2.4.2.3). + fn fit(&self, len: usize) -> (usize, u128) { + let size = self + .buffer() + .and_then(|buffer| buffer.size) + .unwrap_or(0) + .saturating_sub(PES_PREFIX_LEN + PES_OPTIONAL_LEN); + (size, (size / len.max(1)).max(1) as u128) + } + + /// The PSI repetition slot a PES of the track keeps to, if any. AAC with a program config + /// element repeats it in the first frame it muxes after the tables, which go out on + /// [`due`]'s grid; a PES that spanned a boundary of the grid could be muxed after the tables + /// and still go out ahead of them, where a receiver tuning in at the tables misses it. + fn psi_slot(&self, timestamp: Timestamp) -> Option { + matches!(&self.kind, Kind::Aac { config, .. } if config.program_config.is_some()) + .then(|| slot(timestamp, PSI_INTERVAL)) + } + + /// The PES [`Export::pack`] may cut a frame of `payload` presented at `timestamp` into under + /// `delay`, if it packs the frame: as many frames as B holds from a place on the media + /// timeline that is a multiple of them, cut again where they would span more than half the + /// delay and at each [`Self::psi_slot`]. + fn pack_start(&self, payload: &[u8], timestamp: Timestamp, delay: Duration) -> Option { + let (samples, rate) = self.samples(payload).filter(|_| !delay.is_zero())?; + let len = match self.kind { + Kind::Aac { .. } => adts::MIN_HEADER_LEN + payload.len(), + _ => payload.len(), + }; + let place = place(timestamp, samples, rate); + let behind = place % self.fit(len).1; + let frame = 1_000_000_000 * u128::from(samples) / u128::from(rate); + let spans = (delay / 2).as_nanos() / frame.max(1) + 1; + Some((place - behind % spans, self.psi_slot(timestamp))) + } + /// How long a receiver takes to pass one of the track's packets on, so its last packet has /// to arrive that long before it decodes. A packet's bytes reach a decoder buffer only once /// they have drained through the T-STD buffers ahead of it (ISO 13818-1 2.4.2); without it @@ -459,6 +508,55 @@ impl Track { self.buffer().unwrap_or(Buffer::SYSTEM).drain() } + /// How far ahead of its decode time the track's units may go out under `delay`: as far + /// as [`schedule::ahead`] lets them, and nothing without a delay. + fn ahead(&self, delay: Duration) -> Duration { + if delay.is_zero() { + return Duration::ZERO; + } + let slots = schedule::ahead(self.buffer().as_ref(), schedule::slot(delay.as_nanos())); + PCR_INTERVAL * u32::try_from(slots).unwrap_or(u32::MAX) + } + + /// Where the jitter buffer orders a frame decoding at `decode` under `delay`: at its release + /// time, [`Self::ahead`] before its decode time, so it reaches the schedule as the first slot + /// it may ride is laid, and a slot goes out the delay after the source's clock rather than a + /// delay later still. Keys run a delay past release times, which no send-ahead exceeds, so a + /// frame near the start of the timeline is not cut short of its send-ahead. The clock anchors + /// on the track sent latest against these keys, so every track keeps the whole delay to + /// arrive in. A send-ahead that grew (a new SPS declaring a larger CPB) releases no frame + /// before the one ahead of it. + /// + /// An audio frame that shares a PES with the frame released before it takes that frame's + /// key, so the whole PES is in hand when its first slot is laid; the first frame released + /// of a PES (a joiner's may not be the PES's first) keeps its own. That costs no latency. + /// A TS source sends a PES's frames together, so they lose nothing. One that sends audio + /// at its decode time sends the later frames behind their key, so the clock still anchors + /// on the first, and they have that much less of the delay to arrive in. + fn release_key( + &mut self, + decode: Timestamp, + delay: Duration, + frame: &Frame, + ) -> anyhow::Result<(Timestamp, Option)> { + let lead = delay.saturating_sub(self.ahead(delay)); + let key = Duration::from(decode) + .checked_add(lead) + .context("MPEG-TS decode time past the end of the timeline")?; + let mut key = Timestamp::try_from(key)?; + let start = self.pack_start(&frame.payload, frame.timestamp, delay); + if let Some((last, previous, shared)) = self.released + && decode >= last + { + key = match start.is_some() && shared == start { + true => previous, + false => key.max(previous), + }; + } + self.released = Some((decode, key, start)); + Ok((key, start)) + } + /// Queue every held video frame whose DTS is settled, or all of them when `flush`. /// /// A held frame is ready no earlier than the frame read since the source was last found @@ -499,18 +597,28 @@ impl Track { let decode = dts .and_then(|ticks| Timestamp::from_scale(ticks, 90_000).ok()) .unwrap_or(frame.timestamp); + let (key, pes) = self.release_key(decode, jitter.delay(), &frame)?; + // Replaying anchors on the first frame, and the tracks arrive together rather than each + // sent ahead by its send-ahead: judged as arriving that much earlier, as a live source + // sends it, every track's first frames are on the clock whichever anchors it. + let ahead = match jitter.replaying() { + true => (Duration::from(decode) + jitter.delay()).saturating_sub(Duration::from(key)), + false => Duration::ZERO, + }; let arrival = Arrival { - arrived, - read, - decode, + arrived: arrived - ahead, + read: read - ahead, + decode: key, restart, skip, sync: frame.keyframe || !matches!(self.kind, Kind::Video(_)), item: Queued { frame, decode, + key, dts, description, + pes, }, }; if jitter.push(self.pid, arrival)? == Push::Late { @@ -559,6 +667,7 @@ impl Timing { /// The receiver's buffers (H.222.0 2.14.3.1, 2.17.2): EB is the HRD's CPB, else the /// level's; the transport buffer drains at Rx, scaled from the HRD's BitRate or else the /// level's; and MB passes payload on into EB at the level's Rbx whatever the HRD declares. + /// EB reaches the HRD's buffer delay ahead, else what the level's CPB holds at its top rate. fn buffer(&self) -> Option { let level = self.level?; let (bit_rate, cpb) = match self.hrd { @@ -570,9 +679,16 @@ impl Timing { leak: Some(level.leak), size: Some(usize::try_from(cpb / 8).unwrap_or(usize::MAX)), transport: 512, + reach: self.buffer_delay().unwrap_or_else(|| fill(level.cpb, level.leak)), }) } + /// How long the HRD's CPB takes to fill at its rate: how far ahead of its decode time the + /// source may send a picture, so the export has to as well. + fn buffer_delay(&self) -> Option { + self.hrd.map(|hrd| fill(hrd.cpb_size, hrd.bit_rate)) + } + /// The least reorder delay and lookahead the SPS declares, in ticks, so two exporters agree /// on them from their first keyframe whatever reordering each has seen. /// @@ -679,6 +795,8 @@ struct Pack { unit: PesUnit, /// When the first frame decodes, the PES's deadline. decode: Timestamp, + /// The PES its frames share ([`Track::pack_start`]). + start: Option, /// The frames' elementary stream bytes. payload: Vec, /// Each frame's decode nanoseconds and bytes, which leave the decoder buffer as it decodes. @@ -959,23 +1077,33 @@ where self } - /// Mux each frame this long after its decode time, like an SRT receiver's TSBPD. + /// Mux each frame this long after its release time, like an SRT receiver's TSBPD: its + /// decode time less its send-ahead. + /// + /// The send-ahead comes out of the same budget. A frame may go out up to the delay + /// ahead of its decode time, as far as its PID's decoder buffer holds and as early as the + /// receiver's buffers admit, so a heavy passage rides the slots before it. Its release + /// time is that far before its decode time, so it is muxed just as the first slot it may + /// ride is laid, and each slot goes out the delay after its own time on the source's + /// clock rather than a delay later still. /// /// The clock is acquired before anything goes out: frames are held until a track starts - /// a new group, or for the delay at most, and the freshest of them sets the clock, so a - /// joiner runs at the delay from its first output. The output then keeps the source's - /// pace and muxes every track in `(DTS, PID)` order whatever the arrival skew between - /// them. A frame that arrives after its deadline is dropped, and a video track that - /// dropped one resumes at its next keyframe. A source whose clock runs further off ours than - /// 13818-1 lets the PCR follow fails the export once it has used half the delay, where a - /// brief network stall only drops what it made late. A stalled group is skipped once it falls - /// half the delay behind the newest content (see [`Consumer`](crate::container::Consumer)), - /// so the group after it still arrives with the other half to spare. + /// a new group, or for the delay at most, and the freshest of them against its release + /// time sets the clock, so a joiner runs at the delay from its first output and every + /// frame has the whole delay to arrive in. A source that sends its video its send-ahead + /// early, as a TS mux does, is trailed by the delay; one that sends every track at its + /// decode time, by the delay and the video's send-ahead. The output then keeps the + /// source's pace and muxes every track in `(release time, PID)` order whatever the + /// arrival skew between them. A frame that arrives after its deadline is dropped, and a + /// video track that dropped one resumes at its next keyframe. A source whose clock runs + /// further off ours than 13818-1 lets the PCR follow fails the export once it has used half + /// the delay, where a brief network stall only drops what it made late. A stalled group is + /// skipped once it falls half the delay behind the newest content (see + /// [`Consumer`](crate::container::Consumer)), so the group after it still arrives with + /// the other half to spare. /// - /// It is also how far ahead of its decode time a frame may go out, as early as the - /// receiver's buffers for its PID admit, so a heavy passage rides the slots before it - /// and the output trails the source by twice the delay. With a multiplex rate, a frame - /// that cannot arrive by its decode time fails the export. Defaults to + /// With a multiplex rate, a frame that cannot arrive by its decode time fails the export, + /// as does video whose HRD declares a buffer delay past the delay. Defaults to /// [`Duration::ZERO`], which holds nothing and sends each frame by its decode time. pub fn with_delay(mut self, delay: Duration) -> Self { self.delay = delay; @@ -1132,7 +1260,7 @@ where .context("frame for an unknown PID")?; self.last_timestamp = Some(ready.item.frame.timestamp); let decode = ready.item.decode.as_nanos(); - self.lay_before(decode)?; + self.lay_before(ready.item.key.as_nanos())?; self.mux(&name, ready.item)?; if self.delay.is_zero() { // Without a clock to lay slots on, a frame settles every slot before its own: a @@ -1161,6 +1289,12 @@ where } } } + // A PES's frames leave the jitter buffer together, under its first frame's key + // ([`Track::release_key`]), so with none due its pack is whole, and without slots + // running on an empty schedule nothing else would send it on before it is due. + if self.schedule.is_empty() { + self.flush_packs(|_, _| true)?; + } self.lay_due(waiter)?; if let Some(out) = self.pop() { return Poll::Ready(Ok(Some(out))); @@ -1895,6 +2029,8 @@ where decode, dts, description, + pes, + .. } = queued; let is_video = matches!(self.tracks.get(name).context("missing track")?.kind, Kind::Video(_)); // Refresh PSI at keyframes or after the interval lapses. @@ -2035,8 +2171,8 @@ where }; // Only a delay leaves the frames time to wait for the ones after them. let track = self.tracks.get(name).context("missing track")?; - if let Some(samples) = track.samples(&es_payload).filter(|_| !self.delay.is_zero()) { - return self.pack(name, unit, decode, out, es_payload, samples); + if let Some((samples, rate)) = track.samples(&es_payload).filter(|_| !self.delay.is_zero()) { + return self.pack(name, unit, decode, out, es_payload, (samples, rate, pes)); } self.flush(name)?; self.write_pes(&mut out, &unit, &es_payload)?; @@ -2049,12 +2185,14 @@ where /// it, and leave the tables muxed ahead of it to ride the next unit queued. A PES carries /// frames while they run on from its PTS without a gap and fit the PID's decoder buffer /// whole with its header, so it arrives by its first frame's decode time as one frame - /// would, and they span at most half the delay, so the PES has at least that long to go - /// out in. + /// would, and they span at most half the delay: its frames leave the jitter buffer with + /// its first ([`Track::release_key`]), so each keeps at least that much of the delay to + /// arrive in. /// - /// A PES starts at every frame whose place on the media timeline, counted in frames, is a - /// multiple of the frames the buffer holds, so exporters of the broadcast that joined at - /// different frames still cut the same PESs from the first such frame, like [`due`]. + /// A PES carries only frames that left the jitter buffer together ([`Track::pack_start`]), + /// so it is whole however often the exporter is polled, and exporters of the broadcast that + /// joined at different frames still cut the same PESs from the first that starts one, like + /// [`due`]. fn pack( &mut self, name: &str, @@ -2062,24 +2200,17 @@ where decode: Timestamp, tables: Vec, payload: Vec, - (samples, rate): (u64, u32), + (samples, rate, start): (u64, u32, Option), ) -> anyhow::Result<()> { let budget = self.budget().as_nanos(); let track = self.tracks.get(name).context("missing track")?; - // B holds the PES header with the frames (ISO 13818-1 2.4.2.3). - let size = track - .buffer() - .and_then(|buffer| buffer.size) - .unwrap_or(0) - .saturating_sub(PES_PREFIX_LEN + PES_OPTIONAL_LEN); - let fit = (size / payload.len().max(1)).max(1) as u128; - // Rounded, since a frame need not last whole microseconds: AAC's 1,024 samples at 48 kHz. - let frame = 1_000_000 * u128::from(samples); - let place = (unit.timestamp.as_micros() * u128::from(rate) + frame / 2) / frame; + let (size, fit) = track.fit(payload.len()); + let place = place(unit.timestamp, samples, rate); // A PTS is a whole 90 kHz tick, so the frames run on within one of where they end. - let joins = !place.is_multiple_of(fit) + let joins = start.is_some() && track.pack.as_ref().is_some_and(|pack| { - pack.rate == rate + pack.start == start + && pack.rate == rate && pack.payload.len() + payload.len() <= size && decode.as_nanos().saturating_sub(pack.decode.as_nanos()) <= budget && pack.end().abs_diff(unit.timestamp.as_micros()) <= 12 @@ -2092,6 +2223,7 @@ where let pack = track.pack.get_or_insert(Pack { unit, decode, + start, payload: Vec::new(), frames: Vec::new(), samples: 0, @@ -2100,9 +2232,8 @@ where pack.frames.push((decode.as_nanos(), payload.len())); pack.payload.extend_from_slice(&payload); pack.samples += samples; - // The next frame starts a PES of its own anyway, and without slots running on an - // empty schedule, nothing would send this one on before it is due. - if (place + 1).is_multiple_of(fit) || self.schedule.is_empty() { + // The next frame starts a PES of its own anyway. + if (place + 1).is_multiple_of(fit) { self.flush(name)?; } Ok(()) @@ -2168,6 +2299,19 @@ where if let Some(buffer) = track.buffer() { self.schedule.set_buffer(pid, buffer); } + // A padded output cannot burst a picture late, so a CPB that the source may fill to + // its declared buffer delay needs that much send-ahead, all of it out of the delay. + if self.mux_rate.is_some() + && let Some(needs) = track.timing.buffer_delay() + { + let slots = schedule::slot(needs.as_nanos()); + anyhow::ensure!( + slots <= schedule::slot(self.delay.as_nanos()), + "MPEG-TS video on PID {pid} declares a {needs:.3?} decoder buffer delay (its HRD's CPB over its bit rate), past what a {:?} delay sends it ahead; raise the delay to {:?}", + self.delay, + PCR_INTERVAL * u32::try_from(slots).unwrap_or(u32::MAX), + ); + } let by = decode.as_nanos().saturating_sub(track.drain().as_nanos()); self.schedule.push_frames( pid, @@ -2190,21 +2334,29 @@ where /// When the next grid slot is due on the jitter buffer's clock: once every frame that could /// ride it has gone out of the jitter buffer, so its contents depend on the frames alone, /// never on when they arrived, and the output keeps the clock's pace while media is queued - /// or still to decode. `None` without a delay, where frames settle the slots instead. + /// (in the schedule or still in the jitter buffer) or still to decode. `None` without a + /// delay, where frames settle the slots instead. fn slot_due(&self) -> Option<(u128, web_async::time::Instant)> { let (known, settled) = self.slot_settled()?; let settled = Timestamp::from_micros(u64::try_from(settled / 1_000).ok()?).ok()?; Some((known, self.jitter.at(settled)?)) } - /// The `known` that settles the next grid slot, and the decode time, in nanoseconds, by - /// which every frame that could ride it has gone out of the jitter buffer. + /// The `known` that settles the next grid slot, and the jitter buffer key, in nanoseconds, + /// by which every frame that could ride it has gone out of the jitter buffer. Keys run a + /// delay past release times ([`Track::release_key`]). fn slot_settled(&self) -> Option<(u128, u128)> { - if self.delay.is_zero() || self.psi.is_none() || (self.schedule.is_empty() && !self.packing()) { + if self.delay.is_zero() + || self.psi.is_none() + || (self.schedule.is_empty() && !self.packing() && self.jitter.is_empty()) + { return None; } let (_, known) = self.schedule.upcoming()?; - Some((known, known * PCR_INTERVAL.as_nanos() + self.max_drain().as_nanos())) + Some(( + known, + known * PCR_INTERVAL.as_nanos() + self.max_drain().as_nanos() + self.delay.as_nanos(), + )) } /// Whether a track holds frames packed for a PES still to go out. @@ -2212,10 +2364,11 @@ where self.tracks.values().any(|track| track.pack.is_some()) } - /// Lay out every grid slot settled before a frame decoding at `decode` goes out, so frames - /// and slots follow each other in the same order however often the exporter is polled. - fn lay_before(&mut self, decode: u128) -> anyhow::Result<()> { - while let Some((known, _)) = self.slot_settled().filter(|&(_, settled)| settled < decode) { + /// Lay out every grid slot settled before a frame keyed `key` in the jitter buffer goes + /// out, so frames and slots follow each other in the same order however often the + /// exporter is polled. + fn lay_before(&mut self, key: u128) -> anyhow::Result<()> { + while let Some((known, _)) = self.slot_settled().filter(|&(_, settled)| settled < key) { self.lay(Some(known))?; } Ok(()) @@ -2596,6 +2749,14 @@ fn adaptation_size(af: &AdaptationField) -> usize { 2 + if af.pcr.is_some() { 6 } else { 0 } } +/// Where an audio frame of `samples` at `rate` presented at `timestamp` sits on the media +/// timeline, counted in frames. Rounded, since a frame need not last whole microseconds: +/// AAC's 1,024 samples at 48 kHz. +fn place(timestamp: Timestamp, samples: u64, rate: u32) -> u128 { + let frame = 1_000_000 * u128::from(samples); + (timestamp.as_micros() * u128::from(rate) + frame / 2) / frame +} + /// The 33-bit wire timestamp field (90 kHz). DTS and PTS both wrap into it. const TS_TIMESTAMP_MASK: u64 = (1 << 33) - 1; @@ -2726,8 +2887,16 @@ fn video_level(config: &VideoConfig, name: &str) -> Level { }) } +/// How long `bits` take to arrive at `rate` bits per second: how far ahead of its decode +/// time a decoder buffer that size, filling at that rate, holds a unit. +fn fill(bits: u64, rate: u64) -> Duration { + let nanos = u128::from(bits) * 1_000_000_000 / u128::from(rate.max(1)); + Duration::from_nanos(u64::try_from(nanos).unwrap_or(u64::MAX)) +} + /// An audio stream's buffers in a receiver (H.222.0 2.4.2.3): by channel count for ADTS -/// AAC (and Opus, which borrows them), 2 Mb/s and the codec's main buffer for the rest. +/// AAC (and Opus, which borrows them), 2 Mb/s and the codec's main buffer for the rest. B +/// reaches as far ahead as it takes to fill at Rx. fn audio_buffer(config: &AudioConfig, kind: &Kind) -> Buffer { let by_channels = || match config.channel_count { 0..=2 => (2_000_000, 3_584), @@ -2748,6 +2917,7 @@ fn audio_buffer(config: &AudioConfig, kind: &Kind) -> Buffer { leak: None, size: Some(size), transport: 512, + reach: fill(size as u64 * 8, rate), } } @@ -2775,12 +2945,14 @@ fn verbatim_buffer(kind: &Kind, descriptors: &[catalog::Descriptor]) -> Option Some(Buffer { rate: 6_750_000, leak: None, size: None, transport: 480, + reach: Duration::ZERO, }), _ => None, }) @@ -3254,6 +3426,7 @@ mod tests { leak: None, size: None, transport: 480, + reach: Duration::ZERO, }) ); assert_eq!(verbatim_buffer(&kind, &[descriptor(0x0a)]), None); diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index ebae31dce9c..e77cd97475e 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -1137,13 +1137,15 @@ async fn export_pcr_wraps_below_zero_at_start() { .insert(track.name().to_string(), cfg); } let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data)); + // One group, so the exporter joining at the live edge reads it from the frame decoding in + // slot 0: audio goes out no earlier than its own slot. for i in 0..4u64 { producer .write(Frame { timestamp: Timestamp::from_micros(i * 20_000).unwrap(), duration: None, payload: Bytes::from_static(&[0x01, 0x02, 0x03, 0x04]), - keyframe: true, + keyframe: i == 0, }) .unwrap(); } @@ -5463,8 +5465,9 @@ async fn tune_in_does_not_wait_on_dropped_audio() { rig.video(tick); } out.extend(poll_frames(&mut export)); - // Past the keyframe's deadline, and far enough on that a later frame settles its slot. - tokio::time::advance(delay + Duration::from_micros(4 * VIDEO_US)).await; + // Past the audio's deadline, a delay on from the video's since audio is not sent ahead, + // and far enough on that a later frame settles its slot. + tokio::time::advance(2 * delay + Duration::from_micros(4 * VIDEO_US)).await; out.extend(poll_frames(&mut export)); assert!(!out.is_empty(), "the tune-in waited on audio it had dropped"); } @@ -6483,7 +6486,13 @@ async fn pcr_stays_among_the_bytes_across_reordered_tracks() { .with_delay(RECORDING_MAX_AGE); let frames = drain_frames(&mut exporter).await; let pcrs = media_pcrs(&frames); - assert!(pcrs.len() > 50, "expected the full feed, got {} PCRs", pcrs.len()); + assert_eq!(exporter.dropped(), 0, "expected the full feed"); + // The joiner reads the newest group, 400 ms of video sent at most its 375 ms reach early. + assert!( + pcrs.len() > 15, + "expected a clock across the group, got {} PCRs", + pcrs.len() + ); let gaps: Vec = pcrs.windows(2).map(|w| w[1].0 - w[0].0).collect(); assert_eq!( @@ -7204,6 +7213,53 @@ async fn reordered_video_loses_nothing_on_a_clean_path() { assert_on_time(&out); } +/// Padded video whose HRD declares a longer buffer delay than the export's fails with its +/// first picture, naming the delay it needs, rather than once the source fills its CPB and a +/// picture misses its deadline. At that delay it runs. +#[tokio::test(start_paused = true)] +async fn padded_video_needs_the_delay_its_hrd_declares() { + use crate::codec::h264::fixtures::{PPS, SPS_HRD}; + + let waiter = kio::Waiter::noop(); + for (delay, refused) in [(Duration::from_millis(500), true), (Duration::from_secs(1), false)] { + let mut rig = Reordered::new(SPS_HRD, PPS, &[]).await; + let mut export = rig.export(delay).await.with_mux_rate(2_000_000); + let start = tokio::time::Instant::now(); + let mut out = Vec::new(); + let mut failed = None; + 'run: for tick in 0..TICKS / 2 { + tokio::time::sleep_until(start + Duration::from_micros(tick * VIDEO_US)).await; + rig.video(tick); + while rig.audio_before(tick + 1) {} + while let std::task::Poll::Ready(frame) = export.poll_next(&waiter) { + match frame { + Ok(Some(frame)) => out.push(frame), + Ok(None) => break, + Err(err) => { + failed = Some(err.to_string()); + break 'run; + } + } + } + } + if refused { + let err = failed.expect("a 500 ms delay cannot send a 1 s CPB far enough ahead"); + assert!(err.contains("raise the delay to 1s"), "{err}"); + continue; + } + assert_eq!(failed, None, "a 1 s delay sends a 1 s CPB far enough ahead"); + rig.finish(); + out.extend(drain_frames(&mut export).await); + assert_eq!(export.dropped(), 0, "a clean path drops nothing"); + assert_eq!( + video_timing(&out, ..).len() as u64, + TICKS / 2, + "every video frame went out" + ); + assert_on_time(&out); + } +} + /// Two exporters of a broadcast padded to a multiplex rate, one joining a second late, lay the /// same packets once the joiner's first group is out: how many packets a slot carries, the PCR /// it opens with and which units ride it are functions of the media, not of when either 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 0b9ad6f115f..44b043eb723 100644 --- a/rs/moq-mux/src/container/ts/export_timing_test.rs +++ b/rs/moq-mux/src/container/ts/export_timing_test.rs @@ -897,12 +897,13 @@ async fn a_joiner_keeps_the_system_clock_in_tolerance() { } /// A receiver that joins mid-group runs at the delay from its first output: every slot goes -/// out a fixed lag after the source sent its media (the delay before release, and the delay -/// the schedule sends ahead of the decode time), rather than that plus how old the group it -/// joined on was. +/// out the delay after a TS source sent it, rather than that plus how old the group it joined +/// on was, or plus the send-ahead again. The source sends its video as far ahead as the +/// export does (the level's CPB reaches past the delay, so the whole delay) and its audio at +/// its decode time, which is the source's clock. #[tokio::test(start_paused = true)] async fn a_joiner_runs_at_the_delay_from_its_first_output() { - let mut live = Live::new(1.0, 0).await; + let mut live = Live::new(1.0, DELAY.as_micros() as i64).await; live.run(JOIN, &mut [], |_| false).await; let mut leg = live.join(Duration::ZERO).await; live.run(JOIN + 10_000_000 / VIDEO_US, &mut [&mut leg], |_| false).await; @@ -911,14 +912,47 @@ async fn a_joiner_runs_at_the_delay_from_its_first_output() { let held: Vec = leg .out .iter() - .map(|(sent, frame)| sent.saturating_duration_since(live.sent(frame.timestamp.as_micros() as u64))) + .map(|(sent, frame)| { + let media = frame.timestamp.as_micros() as u64; + sent.saturating_duration_since(live.send_times(media, media).1) + }) .collect(); assert!(held.len() > 100, "too little output to judge: {}", held.len()); let slot = Duration::from_millis(SLOT_MS as u64); for (i, held) in held.iter().enumerate() { assert!( - *held >= 2 * DELAY && *held <= 2 * DELAY + 2 * slot, + *held >= DELAY && *held <= DELAY + 2 * slot, "slot {i} went out {held:?} after the source sent it, against a {DELAY:?} delay" ); } } + +/// Live audio shares PESs as replayed audio does, though each frame is released only its +/// send-ahead before it decodes: the frames of a PES go out with its first, however late +/// against its video the source sends them, even at their decode time. +#[tokio::test(start_paused = true)] +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; + assert_eq!( + leg.export.dropped(), + 0, + "lag {lag}: every frame arrived inside its deadline" + ); + + let audio: Vec = units(&leg.out) + .iter() + .filter(|unit| unit.pid == AUDIO_PID) + .map(|unit| unit.decode) + .collect(); + assert!(audio.len() > 10, "lag {lag}: too little audio to judge"); + // 187 bytes with the ADTS header: 19 to the PID's 3,584-byte buffer with the PES + // header, cut again past half the delay, which 12 frames span. + for pair in audio.windows(2) { + let frames = ((pair[1] - pair[0]) as f64 / at90(AUDIO_US) as f64).round() as u64; + assert!(matches!(frames, 12 | 7), "lag {lag}: a PES carried {frames} frames"); + } + } +} diff --git a/rs/moq-mux/src/container/ts/schedule.rs b/rs/moq-mux/src/container/ts/schedule.rs index 5d6092ec0bd..f6dda245013 100644 --- a/rs/moq-mux/src/container/ts/schedule.rs +++ b/rs/moq-mux/src/container/ts/schedule.rs @@ -41,6 +41,16 @@ pub(super) fn slot(nanos: u128) -> u128 { nanos / PCR_INTERVAL.as_nanos() } +/// How many slots ahead of its due slot a unit on a PID with `buffer` may go out, within a +/// `window` of slots: as far as its decoder buffer reaches, or only the slot before its own +/// where no decoder buffer bounds it. +pub(super) fn ahead(buffer: Option<&Buffer>, window: u128) -> u128 { + match buffer { + Some(buffer) if buffer.size.is_some() => window.min(slot(buffer.reach.as_nanos())), + _ => 1, + } +} + /// One slot in system clock ticks. const SLOT_TICKS: u128 = PCR_INTERVAL.as_nanos() * SYSTEM_CLOCK / 1_000_000_000; @@ -79,6 +89,9 @@ pub(super) struct Buffer { /// The transport buffer in bytes: 512 (ISO 13818-1 2.4.2.3), 480 for teletext (ETSI EN /// 300 472 5). pub transport: usize, + /// How far ahead of its decode time the decoder buffer holds a unit: its size over the + /// rate it fills at. Sending further ahead only adds latency. + pub reach: Duration, } impl Buffer { @@ -88,6 +101,7 @@ impl Buffer { leak: None, size: None, transport: 512, + reach: Duration::ZERO, }; /// How long the receiver takes to pass one packet on into the decoder buffer: through the @@ -175,10 +189,6 @@ struct Unit { pid: u16, /// The coarse deadline used by release windows and unpadded admission. due: u128, - /// The due slot of its last frame. A PES that packs several is pushed once that frame is - /// muxed, so its window runs from there, as one frame's runs from when it is pushed: every - /// slot it may ride is laid out after it is queued, whenever the caller lays them. - settled: u128, timing: Timing, /// The slot after the unit pushed before it, which it may go out from even without a /// window: nothing pushed later can be due sooner. @@ -533,9 +543,11 @@ impl Slot { /// /// Each slot's packets go earliest deadline first across PIDs, each PID in its own order, and /// each unit as soon as its PID's buffers in a receiver admit it. Padded output checks -/// transport occupancy and decoder removal at exact packet positions. A unit may go out up to the `window` before -/// its due slot, which keeps that far ahead of a heavy passage, so the output trails the -/// decode times by the window. +/// transport occupancy and decoder removal at exact packet positions. A unit may go out as far +/// ahead of its due slot as its PID's decoder buffer reaches, within the `window`, which keeps +/// that far ahead of a heavy passage ([`ahead`]). A slot is laid once no unit still to be +/// pushed could go out in it, so a caller that pushes each unit that far ahead of its due slot +/// has the output trail the decode times by no more than the window. /// /// With a multiplex rate each slot carries exactly what the rate allows, padded with null /// packets, and a unit that is not complete by its exact deadline fails the export rather than @@ -547,7 +559,7 @@ impl Slot { pub(super) struct Schedule { /// In push order, which is each PID's decode order. units: VecDeque, - /// Slots before its due slot a unit may go out: the export's delay. + /// Slots before its due slot a unit may go out at most: the export's delay. window: u128, /// The multiplex rate, in bits per second. rate: Option, @@ -696,7 +708,6 @@ impl Schedule { self.units.push_back(Unit { pid, due, - settled: slot(last), timing, start, packets, @@ -735,16 +746,14 @@ impl Schedule { self.ended = false; } - /// The first slot `unit` may go out in: a window ahead of its last frame's due slot, or - /// only the slot before it for a PID whose decoder buffer bounds nothing, and never before - /// the slot after the unit pushed ahead of it needs. + /// The first slot `unit` may go out in: [`ahead`] of its due slot, and never before the + /// slot after the unit pushed ahead of it needs. A packed PES is due by its first frame. fn release(&self, unit: &Unit) -> u128 { - let ahead = match self.buffers.get(&unit.pid).and_then(|buffer| buffer.size) { - _ if unit.late => return 0, - Some(_) => self.window, - None => 1, - }; - unit.settled.saturating_sub(ahead).min(unit.start) + if unit.late { + return 0; + } + let ahead = ahead(self.buffers.get(&unit.pid), self.window); + unit.due.saturating_sub(ahead).min(unit.start) } /// The slot [`Self::next`] lays out next, with `known` as it would be given. @@ -755,23 +764,23 @@ impl Schedule { (None, Some(first)) => Some(first), // Skip the empty slots of a long gap, keeping the second before what comes next. (Some(next), first) => { - let target = first.or(known.map(|known| known.saturating_sub(self.window))); + let target = first.or(known); Some(next.max(target.map_or(0, |target| target.saturating_sub(BACKFILL)))) } } } - /// The next slot to lay out, and the first slot a unit still to be pushed could be due - /// in once every unit that may ride it is known: the `known` that settles it. + /// The next slot to lay out, and the first slot a unit still to be pushed could go out in + /// once every unit that may ride it is known: the `known` that settles it. pub fn upcoming(&self) -> Option<(u128, u128)> { let index = self.index(None)?; - Some((index, index + self.window + 1)) + Some((index, index + 1)) } /// Lay out the next slot, if it is settled. /// - /// `known` is the first slot a unit still to be pushed could be due in, which settles - /// every slot whose window ends before it; `None` means nothing more is coming, so every + /// `known` is the first slot a unit still to be pushed could go out in ([`Self::release`]), + /// which settles every slot before it; `None` means nothing more is coming, so every /// queued unit goes out and the clock runs on to the last decode time. pub fn next(&mut self, known: Option) -> anyhow::Result> { if known.is_none() && self.is_empty() { @@ -780,7 +789,7 @@ impl Schedule { let Some(index) = self.index(known) else { return Ok(None); }; - if known.is_some_and(|known| index + self.window >= known) { + if known.is_some_and(|known| index >= known) { return Ok(None); } self.next = Some(index + 1); @@ -1018,9 +1027,7 @@ impl Schedule { continue; } if !self.ended && !unit.late && pid == unit.pid && done > unit.timing.decode as i128 { - anyhow::bail!( - "MPEG-TS output missed a decode deadline on PID {pid} at {rate} b/s; raise the delay or the multiplex rate" - ); + return Err(self.missed(pid, rate)); } self.transport.insert(tb, (arrived, since)); if let Some((leaked, _)) = leak { @@ -1092,10 +1099,7 @@ impl Schedule { .iter() .find(|unit| !unit.late && unit.timing.by as i128 <= end && unit.remaining() > 0) { - anyhow::bail!( - "MPEG-TS output missed a decode deadline on PID {} at {rate} b/s; raise the delay or the multiplex rate", - unit.pid - ); + return Err(self.missed(unit.pid, rate)); } for unit in self.units.iter().filter(|unit| unit.remaining() == 0) { self.decoding.extend(unit.leaving()); @@ -1104,6 +1108,27 @@ impl Schedule { Ok(out) } + /// The error for a unit on `pid` that misses its decode deadline at `rate`, naming what + /// would carry it: the delay at which it goes as far ahead as its decoder buffer holds, + /// where the window cut it short, else a faster multiplex. + fn missed(&self, pid: u16, rate: u64) -> anyhow::Error { + let reach = self + .buffers + .get(&pid) + .filter(|buffer| buffer.size.is_some()) + .map(|buffer| slot(buffer.reach.as_nanos())) + .filter(|&reach| reach > self.window); + match reach { + Some(reach) => anyhow::anyhow!( + "MPEG-TS output missed a decode deadline on PID {pid} at {rate} b/s; raise the delay to {:?}, which sends it as far ahead as its decoder buffer holds, or the multiplex rate", + PCR_INTERVAL * u32::try_from(reach).unwrap_or(u32::MAX) + ), + None => anyhow::anyhow!( + "MPEG-TS output missed a decode deadline on PID {pid} at {rate} b/s, sent as far ahead as its decoder buffer holds; raise the multiplex rate" + ), + } + } + /// How many of each unit's packets slot `index` carries, at most `budget` in all. /// /// Earliest deadline first across PIDs. A PID whose unit stops short (its buffers full, @@ -1203,12 +1228,16 @@ mod tests { /// 40 packets per slot, the clock packet included. const RATE: u64 = 40 * PACKET; + /// A decoder buffer's reach past every window here, so the window bounds how far ahead a + /// unit goes. + const REACH: Duration = Duration::from_secs(1); /// A video-like buffer: far more than a slot passes on, and a large decoder buffer. const VIDEO: Buffer = Buffer { rate: 100_000_000, leak: None, size: Some(1_000_000), transport: 512, + reach: REACH, }; /// Every slot until the schedule is empty: (index, packets per PID, nulls). @@ -1337,6 +1366,7 @@ mod tests { leak: None, size: None, transport: 480, + reach: Duration::ZERO, }; let mut schedule = Schedule::new(Duration::from_millis(100)); schedule.set_rate(Some(rate)); @@ -1374,6 +1404,7 @@ mod tests { leak: None, size: Some(1_000_000), transport: 512, + reach: Duration::from_secs(4), }; let mut schedule = Schedule::new(Duration::from_secs(4)); schedule.set_rate(Some(rate)); @@ -1423,6 +1454,27 @@ mod tests { assert_eq!(clock, [36, 37, 38, 39, 40], "the clock runs on to its decode time"); } + /// However long the window, a unit goes out no further ahead of its due slot than its + /// decoder buffer reaches. + #[test] + fn a_unit_goes_out_no_further_ahead_than_its_buffer_reaches() { + let mut schedule = Schedule::new(Duration::from_millis(500)); + schedule.set_rate(Some(RATE)); + schedule.set_buffer( + 1, + Buffer { + reach: Duration::from_millis(100), + ..VIDEO + }, + ); + schedule.push(1, timing(1_000), unit(1, 3), true).unwrap(); + assert_eq!( + drain(&mut schedule)[0].0, + 36, + "the buffer's reach ahead of its due slot" + ); + } + /// A burst bigger than a slot fills the slots from its release on. #[test] fn a_burst_fills_the_slots_from_its_release() { @@ -1437,15 +1489,29 @@ mod tests { assert_eq!(sent, [(36, 39), (37, 39), (38, 22), (39, 0), (40, 0)]); } - /// A burst the window cannot carry by its due slot fails the export. + /// A burst the window cannot carry by its due slot fails the export, naming the delay that + /// sends it as far ahead as its decoder buffer holds. Once the buffer rather than the + /// window stops it, only a faster multiplex helps. #[test] fn a_burst_too_big_for_the_window_fails() { - let mut schedule = Schedule::new(Duration::from_millis(50)); - schedule.set_rate(Some(RATE)); - schedule.set_buffer(1, VIDEO); - schedule.push(1, timing(1_000), unit(1, 200), true).unwrap(); - let err = std::iter::from_fn(|| schedule.next(None).transpose()).find_map(Result::err); - assert!(err.is_some(), "a burst past the window must fail the export"); + for (window, reach, fix) in [ + (50, 1_000, "raise the delay to 1s"), + (1_000, 50, "raise the multiplex rate"), + ] { + let mut schedule = Schedule::new(Duration::from_millis(window)); + schedule.set_rate(Some(RATE)); + schedule.set_buffer( + 1, + Buffer { + reach: Duration::from_millis(reach), + ..VIDEO + }, + ); + schedule.push(1, timing(1_000), unit(1, 200), true).unwrap(); + let err = std::iter::from_fn(|| schedule.next(None).transpose()).find_map(Result::err); + let err = err.expect("a burst past the window must fail the export").to_string(); + assert!(err.contains(fix), "window {window} ms, reach {reach} ms: {err}"); + } } /// Once every source has ended, a unit the window cannot carry by its due slot is the @@ -1476,6 +1542,7 @@ mod tests { leak: None, size: Some(10 * PAYLOAD), transport: 512, + reach: REACH, }; schedule.set_buffer(1, small); schedule.set_ended(true); @@ -1527,6 +1594,7 @@ mod tests { leak: None, size: Some(100_000), transport: 512, + reach: REACH, }; schedule.set_buffer(1, audio); for at in 0..20 { @@ -1547,6 +1615,7 @@ mod tests { leak: None, size: Some(100_000), transport: 512, + reach: REACH, }; schedule.set_buffer(1, audio); schedule.set_clock(1); @@ -1569,6 +1638,7 @@ mod tests { leak: Some(RBX as u64), size: Some(3_750_000), transport: 512, + reach: REACH, }; let rate = 26_000_000; let mut schedule = Schedule::new(Duration::from_millis(200)); @@ -1618,12 +1688,14 @@ mod tests { leak: Some(RBX), size: Some(3_750_000), transport: 512, + reach: Duration::from_secs(2), }; let audio = Buffer { rate: 2_000_000, leak: None, size: Some(3_584), transport: 512, + reach: Duration::from_secs(2), }; let rate = 26_000_000; let mut schedule = Schedule::new(Duration::from_secs(2)); @@ -1664,6 +1736,7 @@ mod tests { leak: Some(24_000_000), size: None, transport: 512, + reach: Duration::ZERO, }; assert_eq!(second(leak), 16_304); // Rx binds: 20 Mb/s of 188-byte packets, 13,297.9 a second, less 40 clock packets. @@ -1686,6 +1759,7 @@ mod tests { leak: None, size: Some(100_000), transport: 512, + reach: REACH, }; schedule.set_buffer(1, slow); schedule.set_clock(1); @@ -1707,6 +1781,7 @@ mod tests { leak: None, size: Some(100_000), transport: 512, + reach: REACH, }, ); // 76.8 kb/s passes on five packets in the window, the slots' PCRs among them. @@ -1730,6 +1805,7 @@ mod tests { leak: None, size: Some(10 * PAYLOAD), transport: 512, + reach: REACH, }; schedule.set_buffer(1, small); schedule.push(1, timing(1_000), unit(1, 8), true).unwrap(); @@ -1775,6 +1851,7 @@ mod tests { leak: None, size: Some(2_592), transport: 512, + reach: REACH, }, ); let error = schedule @@ -1820,6 +1897,7 @@ mod tests { leak: None, size: Some(2 * PAYLOAD), transport: 512, + reach: REACH, }, ); schedule @@ -1868,6 +1946,7 @@ mod tests { leak: None, size: Some(2_592), transport: 512, + reach: REACH, }, ); let bytes = bitrate / 8 * 32 / 1_000; @@ -1975,8 +2054,9 @@ mod tests { schedule.set_buffer(1, VIDEO); schedule.push(1, timing(1_000), unit(1, 8), true).unwrap(); schedule.push(1, timing(1_100), unit(1, 100), false).unwrap(); - // Unpadded, both go out as soon as they are released, a window ahead. - let first: Vec<_> = std::iter::from_fn(|| schedule.next(Some(slot(ms(1_100)))).unwrap()) + // Unpadded, both go out as soon as they are released, a window ahead: the slots before + // the second unit's release are settled. + let first: Vec<_> = std::iter::from_fn(|| schedule.next(Some(slot(ms(1_000)))).unwrap()) .map(|slot| (slot.index, slot.packets.len() / TsPacket::SIZE, slot.nulls)) .collect(); assert_eq!(first, [(36, 8, 0), (37, 0, 0), (38, 0, 0), (39, 0, 0)]); @@ -2002,6 +2082,7 @@ mod tests { leak: None, size: Some(100_000), transport: 512, + reach: REACH, }, ); schedule.push(1, timing(1_000), unit(1, 50), true).unwrap(); @@ -2139,6 +2220,7 @@ mod tests { leak: None, size: None, transport: 512, + reach: Duration::ZERO, } .per_slot(0, true), )]), @@ -2183,12 +2265,14 @@ mod tests { leak: None, size: None, transport: 512, + reach: Duration::ZERO, }, Buffer { rate: 8_160_000, leak: None, size: None, transport: 512, + reach: Duration::ZERO, }, ); let slot = Slot { @@ -2361,6 +2445,7 @@ mod tests { leak: None, size: Some(1_600), transport: 512, + reach: REACH, }, ); let frames = vec![(ms(1_000), 500), (ms(1_024), 500), (ms(1_048), 500)]; @@ -2404,4 +2489,41 @@ mod tests { starts[1] ); } + + /// A packed PES must be whole by its first frame's decode time, so its send-ahead runs + /// from that frame's due slot, not its last's: here the PES needs over two slots at Rx. + #[test] + fn a_packed_pes_goes_out_ahead_of_its_first_frame() { + let mut schedule = Schedule::new(Duration::from_secs(1)); + schedule.set_rate(Some(10_000_000)); + schedule.set_buffer( + 1, + Buffer { + rate: 2_000_000, + leak: None, + size: Some(12_896), + transport: 512, + reach: Duration::from_nanos(12_896 * 8 * 1_000_000_000 / 2_000_000), + }, + ); + schedule.set_buffer(2, VIDEO); + // Video pushed ahead of it starts the PES no earlier than its own due slot. + schedule.push(2, timing(3_160), unit(2, 2), true).unwrap(); + let frames = (0..5).map(|i| (ms(3_200 + 32 * i), 2_560)).collect(); + schedule + .push_frames( + 1, + Timing { + by: ms(3_199), + decode: ms(3_200), + }, + audio(12_800), + false, + frames, + ) + .unwrap(); + while schedule.queued() { + schedule.next(Some(u128::MAX)).unwrap(); + } + } } diff --git a/rs/moq-mux/src/jitter.rs b/rs/moq-mux/src/jitter.rs index aab7f972d42..6715d535bc7 100644 --- a/rs/moq-mux/src/jitter.rs +++ b/rs/moq-mux/src/jitter.rs @@ -88,11 +88,14 @@ pub(crate) enum Push { /// The clock follows the source's: a source clock running slower than ours would make /// every frame late in the end, and a faster one would hold more and more. So the clock /// measures the source's rate and runs its decode timeline at it, holding the freshest -/// frames of the track sent latest a delay. Each [`STEER`] of our time gives a floor, the -/// most slack a frame of that track arrived with (the least across tracks of each one's -/// most, a track not heard placed by its lead), which queueing and retransmission only ever -/// lower; the upper envelope of the floors over [`RATE_WINDOW`] gives the rate (its slope) -/// and the slack now, so a spell of queueing shorter than half the window moves neither. +/// frames of the track sent latest a delay. Each [`STEER`] of our time gives each track it +/// hears its most slack, which queueing and retransmission only ever lower, and a floor, +/// the least of those (a track not heard placed by its lead). Over [`RATE_WINDOW`] the +/// rate is the slope of one track's upper envelope, the track whose slack keeps closest to +/// it: a TS mux sends its video ahead by however full the encoder's buffer is, so the +/// video's slack wanders by hundreds of ms where the audio's keeps within a few. The slack +/// now is the floors' upper envelope at that slope, so the track sent latest keeps the +/// delay, and a spell of queueing shorter than half the window moves neither. /// The backlog a network stall releases is read at once, so it gives one floor, its /// freshest frame's: stepped on decode time it would give one per step of the media it /// held, and early on, before the window fills, those would read as a source far slower @@ -224,8 +227,11 @@ struct Steer { /// This step's floor. floor: Option>, /// Each step's floor, as its decode time and phase in seconds, over the last - /// [`RATE_WINDOW`]. Its slope is how much faster the source's clock runs than ours. + /// [`RATE_WINDOW`]. phase: VecDeque<(f64, f64)>, + /// Each track's own phase at each step that heard it, over the same window. The slope of + /// the steadiest's envelope is how much faster the source's clock runs than ours. + tracks: BTreeMap>, /// How far the drift has moved the deadlines, in seconds. steered: f64, /// The source's clock rate off ours, as last measured: positive runs fast. @@ -241,6 +247,7 @@ impl Default for Steer { Self { floor: None, phase: VecDeque::new(), + tracks: BTreeMap::new(), steered: 0.0, source: None, startup: None, @@ -248,6 +255,33 @@ impl Default for Steer { } } +impl Steer { + /// The source's rate and the floor's phase now. The rate is the slope of the envelope of + /// whichever track in `heard`, of those heard in at least half the floors, lies closest + /// under its own at the median, so a spell of queueing on it shorter than half the window + /// does not hand the rate to another; the phase is the floors' envelope at that slope. + fn measure<'a>(&self, heard: impl Iterator, now: f64) -> Option<(f64, f64)> + where + K: 'a, + { + let steadiest = 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)) + }) + .min_by(|a, b| a.0.total_cmp(&b.0)) + .map(|(_, slope)| slope); + let rate = match steadiest { + Some(rate) => rate, + None => envelope(&self.phase, now)?.0, + }; + let at = self.phase.iter().map(|&(x, y)| y + rate * (now - x)).reduce(f64::max)?; + Some((rate, at)) + } +} + /// Each track's frame with the most slack in a step: the one queued least on its way. The /// step's floor is the track with the least of those, the one sent latest against its decode time. struct Floor { @@ -299,6 +333,11 @@ impl Buffer { } } + /// How long past its decode time each frame is held. + pub fn delay(&self) -> Duration { + self.delay + } + /// The tracks an acquisition waits to hear from, two delays at most, before it anchors: any /// of them may be the one sent latest against its decode time. pub fn expect(&mut self, keys: impl IntoIterator) { @@ -313,6 +352,11 @@ impl Buffer { self } + /// Whether each clock anchors on its first frame ([`Self::replay`]). + pub fn replaying(&self) -> bool { + self.replay + } + /// Queue a frame, or drop it if it cannot make its deadline. Fails once the source's clock /// runs further off ours than the clock may follow. pub fn push(&mut self, key: K, arrival: Arrival) -> anyhow::Result { @@ -594,12 +638,23 @@ impl Buffer { } } self.steer.steered += elapsed * clock.drift; - let phases = &mut self.steer.phase; - phases.push_back((at, phase)); - while phases.front().is_some_and(|&(at, _)| now - at > RATE_WINDOW) { - phases.pop_front(); + let window = |series: &mut VecDeque<(f64, f64)>| { + while series.front().is_some_and(|&(at, _)| now - at > RATE_WINDOW) { + series.pop_front(); + } + }; + self.steer.phase.push_back((at, phase)); + window(&mut self.steer.phase); + for (key, &(track, decode)) in &floor.tracks { + self.steer + .tracks + .entry(key.clone()) + .or_default() + .push_back((decode, track)); } - let Some((rate, at)) = envelope(phases, now) else { + self.steer.tracks.values_mut().for_each(window); + self.steer.tracks.retain(|_, series| !series.is_empty()); + let Some((rate, at)) = self.steer.measure(floor.tracks.keys(), now) else { self.clock = Some(Clock { anchor, base: arrival.decode, @@ -850,6 +905,13 @@ fn envelope(points: &VecDeque<(f64, f64)>, now: f64) -> Option<(f64, f64)> { Some((slope, y1 + slope * (now - x1))) } +/// How far below the line of `slope` that is at `at` at `now` the median of `points` lies. +fn depth(points: &VecDeque<(f64, f64)>, slope: f64, at: f64, now: f64) -> f64 { + let mut below: Vec = points.iter().map(|&(x, y)| at - slope * (now - x) - y).collect(); + below.sort_by(f64::total_cmp); + below[below.len() / 2] +} + #[cfg(test)] mod tests { use super::*; @@ -1413,6 +1475,11 @@ mod tests { dropped: u64, /// What the last push did. last: Option, + /// The source's rate the buffer measured at the end, and how often it was out of tolerance. + drift: Option, + out_of_tolerance: u64, + /// The rate measured after each push, by decode time in seconds. + rates: Vec<(f64, f64)>, } /// Feed `secs` of 10 fps frames of tracks 1 and 2 through a 500 ms buffer from a source @@ -1439,6 +1506,9 @@ mod tests { err: None, dropped: 0, last: None, + drift: None, + out_of_tolerance: 0, + rates: Vec::new(), }; for (sent, track, decode) in frames { let arrived = start + Duration::from_secs_f64(sent); @@ -1459,9 +1529,14 @@ mod tests { break; } } + if let Some(rate) = buffer.drift() { + fed.rates.push((decode.as_secs_f64(), rate)); + } due(&mut buffer); } fed.dropped = buffer.dropped(); + fed.drift = buffer.drift(); + fed.out_of_tolerance = buffer.out_of_tolerance(); fed } @@ -1555,6 +1630,49 @@ mod tests { } } + /// The track sent latest, sent ahead by a lead that wanders as a TS mux's video does with + /// how full the encoder's buffer is, here by 200 ms over 20 minutes, does not move the rate: + /// it is measured on the track sent as it decodes, whose slack keeps to its envelope, so a + /// source within the limit is never out of tolerance and the estimate settles at its rate. + #[tokio::test(start_paused = true)] + async fn a_wandering_track_sent_latest_does_not_move_the_rate() { + for ppm in [-20.0, 0.0, 20.0] { + let fed = feed(ppm, 3_600.0, |track, sent| match track { + 1 => Some(sent + 0.2 - 0.1 * (sent * std::f64::consts::TAU / 1_200.0).cos()), + _ => Some(sent), + }) + .await; + assert!(fed.err.is_none(), "{ppm} ppm: {:?}", fed.err); + assert_eq!(fed.dropped, 0, "{ppm} ppm"); + assert_eq!(fed.out_of_tolerance, 0, "{ppm} ppm"); + let drift = fed.drift.expect("a measurement") * 1e6; + assert!((drift + ppm).abs() < 0.5, "{ppm} ppm: measured {drift} ppm"); + } + } + + /// A spell of queueing on the steady track, four minutes of 200 ms more, does not hand the + /// rate to the wandering one: a track's swing is the median of how far its slack lies under + /// its envelope, which a spell shorter than half the window does not move. + #[tokio::test(start_paused = true)] + async fn a_spell_on_the_steady_track_keeps_the_rate_on_it() { + let spell = 1_500.0..1_740.0; + for ppm in [-20.0, 0.0, 20.0] { + let fed = feed(ppm, 2_400.0, |track, sent| match track { + 1 => Some(sent + 0.2 - 0.1 * (sent * std::f64::consts::TAU / 900.0).cos()), + _ => Some(sent + if spell.contains(&sent) { 0.2 } else { 0.0 }), + }) + .await; + assert!(fed.err.is_none(), "{ppm} ppm: {:?}", fed.err); + let worst = fed + .rates + .iter() + .filter(|(at, _)| *at > RATE_WINDOW + 60.0) + .map(|(_, rate)| (rate * 1e6 + ppm).abs()) + .fold(0.0, f64::max); + assert!(worst < 0.5, "{ppm} ppm: measured up to {worst} ppm off"); + } + } + /// A joiner is handed its group at once. The clock is anchored on the freshest frame, so /// what is older than the delay is dropped without counting as late, and the next group /// goes out a delay after it arrived. diff --git a/rs/moq-srt/src/server.rs b/rs/moq-srt/src/server.rs index 9757e69645c..92e3a81ae58 100644 --- a/rs/moq-srt/src/server.rs +++ b/rs/moq-srt/src/server.rs @@ -184,8 +184,8 @@ pub struct Server { _listener: SrtListener, incoming: SrtIncoming, local_addr: SocketAddr, - /// The negotiated SRT receive latency, reused as the egress skip threshold on - /// each [`Subscribe`] (see [`crate::ts::Subscriber::new`]). + /// The negotiated SRT receive latency, reused as the egress delay (at least a + /// second) on each [`Subscribe`] (see [`crate::ts::Subscriber::new`]). latency: Duration, } @@ -193,8 +193,8 @@ impl Server { /// Bind an SRT listener on `addr` (SRT has no well-known port; 9000 is common). /// /// `latency` is the SRT receive latency, negotiated at handshake time; pass - /// `None` for a sensible default (500ms). It doubles as the egress skip - /// threshold for [`Subscribe`] requests. + /// `None` for a sensible default (500ms). It doubles as the egress delay for + /// [`Subscribe`] requests, but no less than a second. pub async fn bind(addr: SocketAddr, latency: impl Into>) -> Result { let latency = latency.into().unwrap_or(DEFAULT_LATENCY); @@ -276,7 +276,7 @@ struct Pending { /// fields out of it (e.g. a token in `u=` or a custom key). stream_id: Option, peer: SocketAddr, - /// The SRT receive latency, reused as the egress skip threshold on a subscribe. + /// The SRT receive latency, reused as the egress delay (at least a second) on a subscribe. latency: Duration, /// Retention declared on the media tracks an ingest mints, or `None` for hang's own /// default. Override with [`Publish::with_max_age`]. diff --git a/rs/moq-srt/src/ts.rs b/rs/moq-srt/src/ts.rs index c2a25048f7a..7ccaa59ace9 100644 --- a/rs/moq-srt/src/ts.rs +++ b/rs/moq-srt/src/ts.rs @@ -161,10 +161,14 @@ impl Publisher { } } +/// The least delay an egress muxes at, `moq export ts`'s default: the send-ahead that fills a +/// receiver's buffers comes out of the delay, and a broadcast-sized CPB needs a second of it. +const MIN_DELAY: Duration = Duration::from_secs(1); + /// How an egress treats its broadcast ending or being replaced. #[derive(Clone, Copy, Debug, Default)] pub(crate) struct Options { - /// The muxer's jitter-buffer delay ([`Subscriber::new`]). + /// The muxer's delay, at least [`MIN_DELAY`] ([`Subscriber::new`]). pub latency: Duration, /// How long a return or stitched replacement has to supply its first catalog. pub linger: Duration, @@ -189,8 +193,8 @@ pub struct Subscriber { impl Subscriber { /// Resolve the broadcast at `path` in the origin and prepare to mux it to TS. /// - /// `options.latency` is the muxer's jitter-buffer delay: each frame is muxed that long after - /// its decode time, a frame arriving later is dropped, and a stalled group is + /// `options.latency` is the muxer's delay, but no less than [`MIN_DELAY`]: the output runs + /// that far behind the source, a frame arriving later is dropped, and a stalled group is /// skipped after half of it. We reuse the locally configured SRT receive latency for /// it, the same budget an SRT hop gives a packet. It's the configured value, /// not the handshake result (srt-tokio doesn't expose the negotiated latency), @@ -210,7 +214,7 @@ impl Subscriber { let broadcast = origin.request_broadcast(path, None).await?; let source = moq_mux::Source::new(origin.clone(), broadcast); let catalog = source.catalog::(moq_mux::catalog::CatalogFormat::Hang).await?; - let export = ts::Export::new(source, catalog).with_delay(options.latency); + let export = ts::Export::new(source, catalog).with_delay(options.latency.max(MIN_DELAY)); let follower = ts::Follower::new(export)? .with_linger(options.linger) .with_stitch(options.stitch); diff --git a/test/justfile b/test/justfile index 8b3860825bd..9dc104843b0 100644 --- a/test/justfile +++ b/test/justfile @@ -129,7 +129,7 @@ drill-pr mode base="": # (use a real capture), `--live` (grade PCR release timing and byte position off # the pipe instead, which a capture cannot carry; nightly runs this arm), # `--open-gop` (an open-GOP clip whose leading pictures must survive), `--hrd` -# (1080p video filling a broadcast-sized CPB), `--delay 1s` (the exporter's). See +# (1080p video filling a broadcast-sized CPB), `--delay 2s` (the exporter's). See # ts/README.md. # Check the subscriber's MPEG-TS output for IRD compliance. diff --git a/test/ts/README.md b/test/ts/README.md index af49738fb62..3ae16c10b81 100644 --- a/test/ts/README.md +++ b/test/ts/README.md @@ -39,7 +39,7 @@ just test ts --headroom # the same with four audio PIDs, muxed above the just test ts --leak # 1080p video with no HRD, filling its level's 24 Mb/s leak just test ts --full # the same with two audio PIDs, filling its multiplex with no nulls just test ts --fragment-pes # legal headers split across packets, with FFmpeg/TSDuck grading -just test ts --delay 1s # pass the exporter's --delay +just test ts --delay 2s # pass the exporter's --delay ``` `--fragment-pes` shifts generated audio/video PES bursts into their following @@ -298,7 +298,7 @@ the total and reports the rate over the tail of the sample beside it. A sender that buffers builds a standing lag once and then runs at the media rate: the lag is a constant offset no receiver can see, and it cannot grow past the latency budget the sender is allowed to hold, so set `--drift-ms` to that budget -(`export ts --delay`, 500 ms by default). A pipe that is not running at the +(`export ts --delay`, 1 s by default). A pipe that is not running at the media rate never stops accumulating and so breaches any fixed bound given a long enough sample. The tail rate is what tells the two apart, and it needs a sample longer than the lag takes to build: measured against the grid-sliced exporter, @@ -374,7 +374,7 @@ outgrow a 6 kB slot, so the exporter has to spread each over the slots before it DTS; CI runs that too (`just test ts --bitrate 2000000`). `--hrd` goes further: a 1080p encode with a 9 Mbit NAL HRD kept near full by noise, the shape of a contribution encoder's output, which sends pictures most of a second ahead of their -decode time and loads the decoder buffer past 60 % at the default delay (the recipe +decode time and fills the decoder buffer to its declared size at the default delay (the recipe moq-dev/moq#4645 graded with; CI runs it too). Every other clip muxes at or below the video's Rx, where its transport buffer cannot fill. `--headroom` puts that video beside three MPEG-1 Layer II PIDs and an AC-3 one in a 12.5 Mb/s multiplex, above diff --git a/test/ts/pcr-timing.py b/test/ts/pcr-timing.py index 6d0491b57d0..69cb6f435cb 100755 --- a/test/ts/pcr-timing.py +++ b/test/ts/pcr-timing.py @@ -670,10 +670,10 @@ def main(): ap.add_argument( "--drift-ms", type=float, - default=500.0, + default=1000.0, help="bound on accumulated release drift, being the standing lag the sender may " "hold; set it to the sender's latency budget (moq export ts --delay, " - "itself 500ms by default)", + "itself 1s by default)", ) ap.add_argument( "--rate-ppm", diff --git a/test/ts/run.sh b/test/ts/run.sh index 7df732f7144..8756789e816 100755 --- a/test/ts/run.sh +++ b/test/ts/run.sh @@ -24,7 +24,7 @@ # ./run.sh --leak # 1080p video with no HRD, filling its level's 24 Mb/s leak # ./run.sh --full # the same with two audio PIDs, filling its multiplex with no nulls # ./run.sh --fragment-pes # split generated clear PES headers across packets -# ./run.sh --delay 1s # pass the exporter's --delay +# ./run.sh --delay 2s # pass the exporter's --delay # `--live` swaps the analyzer, not the rig. compliance.py grades a captured file # on the stream's own PCR clock, which is the right basis for the IRD model it