From b176c63580f4c0ae66bb7342880272e22080c499 Mon Sep 17 00:00:00 2001 From: t0ms Date: Thu, 15 Oct 2026 00:42:07 +0100 Subject: [PATCH 1/5] fix(ts)!: run the export one delay behind its source `moq export ts` held each frame for `--delay` and then sent it up to another `--delay` ahead of its decode time, so its output trailed the source by about twice the delay. A frame is now due at its release time: anchor + DTS + delay less its send-ahead, where the send-ahead comes out of the same delay and is capped by the PID's decoder buffer (EB from the SPS HRD or the level, B for audio). Padded video whose HRD declares a longer buffer delay than the delay fails the export, naming the `--delay` it needs. Audio still packs several frames to a PES, though B is too small to send it ahead of its due slot: the frames of a PES take its first frame's jitter key, so the PES leaves the jitter buffer whole. A PES spans at most half the delay, and AAC with a program config element starts one at each PSI repetition. The default `--delay` rises from 500 ms to 1 s, which fits a broadcast-sized CPB, and SRT egress muxes at its latency or 1 s, whichever is longer. Co-authored-by: Claude Opus 5.5 --- doc/bin/cli.md | 26 +- doc/bin/srt.md | 5 +- doc/setup/upgrade.md | 10 + quest/m1/tstd/README.md | 6 +- quest/m1/tstd/send-ahead.md | 36 --- rs/moq-cli/src/args.rs | 20 +- rs/moq-cli/src/subscribe.rs | 4 +- rs/moq-mux/benches/ts_schedule.rs | 1 + rs/moq-mux/src/codec/h264/mod.rs | 28 ++ rs/moq-mux/src/container/ts/export.rs | 281 ++++++++++++++---- rs/moq-mux/src/container/ts/export_test.rs | 64 +++- .../src/container/ts/export_timing_test.rs | 46 ++- rs/moq-mux/src/container/ts/schedule.rs | 160 +++++++--- rs/moq-mux/src/jitter.rs | 10 + rs/moq-srt/src/server.rs | 10 +- rs/moq-srt/src/ts.rs | 12 +- test/justfile | 2 +- test/ts/README.md | 6 +- test/ts/pcr-timing.py | 4 +- test/ts/run.sh | 2 +- 20 files changed, 555 insertions(+), 178 deletions(-) delete mode 100644 quest/m1/tstd/send-ahead.md diff --git a/doc/bin/cli.md b/doc/bin/cli.md index a7ca86090b0..5fa2c281466 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -127,8 +127,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 @@ -346,14 +349,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 2156a52e141..20956b09c12 100644 --- a/doc/setup/upgrade.md +++ b/doc/setup/upgrade.md @@ -387,6 +387,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/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 0952f3709cb..4a968cf5c64 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -1135,13 +1135,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(); } @@ -5461,8 +5463,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"); } @@ -6481,7 +6484,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!( @@ -7202,6 +7211,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..cb62e3d9eae 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 @@ -533,9 +547,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 +563,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, @@ -735,15 +751,13 @@ 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 last frame's due slot, and never + /// before the slot after the unit pushed ahead of it needs. 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, - }; + if unit.late { + return 0; + } + let ahead = ahead(self.buffers.get(&unit.pid), self.window); unit.settled.saturating_sub(ahead).min(unit.start) } @@ -755,23 +769,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 +794,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 +1032,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 +1104,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 +1113,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 +1233,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 +1371,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 +1409,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 +1459,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 +1494,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 +1547,7 @@ mod tests { leak: None, size: Some(10 * PAYLOAD), transport: 512, + reach: REACH, }; schedule.set_buffer(1, small); schedule.set_ended(true); @@ -1527,6 +1599,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 +1620,7 @@ mod tests { leak: None, size: Some(100_000), transport: 512, + reach: REACH, }; schedule.set_buffer(1, audio); schedule.set_clock(1); @@ -1569,6 +1643,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 +1693,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 +1741,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 +1764,7 @@ mod tests { leak: None, size: Some(100_000), transport: 512, + reach: REACH, }; schedule.set_buffer(1, slow); schedule.set_clock(1); @@ -1707,6 +1786,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 +1810,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 +1856,7 @@ mod tests { leak: None, size: Some(2_592), transport: 512, + reach: REACH, }, ); let error = schedule @@ -1820,6 +1902,7 @@ mod tests { leak: None, size: Some(2 * PAYLOAD), transport: 512, + reach: REACH, }, ); schedule @@ -1868,6 +1951,7 @@ mod tests { leak: None, size: Some(2_592), transport: 512, + reach: REACH, }, ); let bytes = bitrate / 8 * 32 / 1_000; @@ -1975,8 +2059,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 +2087,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 +2225,7 @@ mod tests { leak: None, size: None, transport: 512, + reach: Duration::ZERO, } .per_slot(0, true), )]), @@ -2183,12 +2270,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 +2450,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)]; diff --git a/rs/moq-mux/src/jitter.rs b/rs/moq-mux/src/jitter.rs index aab7f972d42..c9b0ad7e214 100644 --- a/rs/moq-mux/src/jitter.rs +++ b/rs/moq-mux/src/jitter.rs @@ -299,6 +299,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 +318,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 { 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 e4f08162eb0..35caa5bfcf8 100644 --- a/rs/moq-srt/src/ts.rs +++ b/rs/moq-srt/src/ts.rs @@ -160,10 +160,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, @@ -188,8 +192,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), @@ -209,7 +213,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 33695451923..3cf3f585de4 100644 --- a/test/justfile +++ b/test/justfile @@ -122,7 +122,7 @@ drill-sensitivity *args: # (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 0787694060f..80c929d342f 100644 --- a/test/ts/README.md +++ b/test/ts/README.md @@ -38,7 +38,7 @@ just test ts --hrd # 1080p video filling a broadcast-sized 9 Mbit C just test ts --headroom # the same with four audio PIDs, muxed above the video's Rx 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 --delay 1s # pass the exporter's --delay +just test ts --delay 2s # pass the exporter's --delay ``` `--live` swaps the analyzer, not the rig: the same round-trip runs, but the @@ -292,7 +292,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, @@ -368,7 +368,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 e81dcf6c878..55f53ca65e1 100755 --- a/test/ts/run.sh +++ b/test/ts/run.sh @@ -23,7 +23,7 @@ # ./run.sh --headroom # the same with four audio PIDs, muxed above the video's Rx # ./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 --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 From bad45d618cfffceddf94e508b3d57d8cb1a5e00b Mon Sep 17 00:00:00 2001 From: t0ms Date: Thu, 15 Oct 2026 12:50:08 +0100 Subject: [PATCH 2/5] fix(jitter): take the source's rate from its steadiest track Keyed at release times, a TS source's video is the track sent latest, and its lead wanders with how full the encoder's buffer is: by 230 ms on a 25 Mb/s broadcast capture, where the audio's keeps within 20 ms. The rate was the slope of the floors' envelope, so it followed the video's wander, reading +257 to +1,164 ppm for a source at its own rate and out of tolerance on most steps. Keep each track's own phase per step as well. The rate is the slope of the envelope of whichever heard track lies closest under its own on average; the slack now is still the floors' envelope, at that slope, so the track sent latest keeps the delay. With one track it is the floors' envelope, as before. Co-authored-by: Claude Opus 5.5 --- rs/moq-mux/src/jitter.rs | 98 +++++++++++++++++++++++++++++++++++----- 1 file changed, 87 insertions(+), 11 deletions(-) diff --git a/rs/moq-mux/src/jitter.rs b/rs/moq-mux/src/jitter.rs index c9b0ad7e214..85d4243e4fc 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,32 @@ 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 on average; 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 { @@ -604,12 +637,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, @@ -860,6 +904,11 @@ 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` `points` lie, on average. +fn depth(points: &VecDeque<(f64, f64)>, slope: f64, at: f64, now: f64) -> f64 { + points.iter().map(|&(x, y)| at - slope * (now - x) - y).sum::() / points.len() as f64 +} + #[cfg(test)] mod tests { use super::*; @@ -1423,6 +1472,9 @@ 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, } /// Feed `secs` of 10 fps frames of tracks 1 and 2 through a 500 ms buffer from a source @@ -1449,6 +1501,8 @@ mod tests { err: None, dropped: 0, last: None, + drift: None, + out_of_tolerance: 0, }; for (sent, track, decode) in frames { let arrived = start + Duration::from_secs_f64(sent); @@ -1472,6 +1526,8 @@ mod tests { due(&mut buffer); } fed.dropped = buffer.dropped(); + fed.drift = buffer.drift(); + fed.out_of_tolerance = buffer.out_of_tolerance(); fed } @@ -1565,6 +1621,26 @@ 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 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. From 2dfa32f345182b5470073605415404af28dfaf60 Mon Sep 17 00:00:00 2001 From: t0ms Date: Thu, 15 Oct 2026 17:34:40 +0100 Subject: [PATCH 3/5] fix(jitter): judge a track's depth under its envelope by the median The mean let a spell of queueing on the steady track, shorter than half the window, deepen it past the wandering track, which then measured the rate. The median is moved by such a spell no more than the envelope is. Co-authored-by: Luke Curley Co-authored-by: Claude Opus 5.5 --- rs/moq-mux/src/jitter.rs | 38 +++++++++++++++++++++++++++++++++++--- 1 file changed, 35 insertions(+), 3 deletions(-) diff --git a/rs/moq-mux/src/jitter.rs b/rs/moq-mux/src/jitter.rs index 85d4243e4fc..6715d535bc7 100644 --- a/rs/moq-mux/src/jitter.rs +++ b/rs/moq-mux/src/jitter.rs @@ -258,7 +258,8 @@ 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 on average; the phase is the floors' envelope at that slope. + /// 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, @@ -904,9 +905,11 @@ 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` `points` lie, on average. +/// 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 { - points.iter().map(|&(x, y)| at - slope * (now - x) - y).sum::() / points.len() as 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)] @@ -1475,6 +1478,8 @@ mod tests { /// 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 @@ -1503,6 +1508,7 @@ mod tests { 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); @@ -1523,6 +1529,9 @@ mod tests { break; } } + if let Some(rate) = buffer.drift() { + fed.rates.push((decode.as_secs_f64(), rate)); + } due(&mut buffer); } fed.dropped = buffer.dropped(); @@ -1641,6 +1650,29 @@ mod tests { } } + /// 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. From a44bf6515e41f5745f1bfcef12ce5d89c521bfef Mon Sep 17 00:00:00 2001 From: t0ms Date: Thu, 15 Oct 2026 22:23:42 +0100 Subject: [PATCH 4/5] perf(mux): bench the jitter buffer's steering over tracks and window Each two-second steering step measures every track heard, and each measurement sorts that track's populated window to take its median depth, so the cost grows with both track count and floors. The bench feeds a frame a second per track for as many steps as fill the window with 10, 100 or 300 floors, across 1, 4, 16 and 64 tracks. Co-authored-by: Claude Opus 5.5 --- rs/moq-mux/Cargo.toml | 4 +++ rs/moq-mux/benches/jitter.rs | 62 ++++++++++++++++++++++++++++++++++++ 2 files changed, 66 insertions(+) create mode 100644 rs/moq-mux/benches/jitter.rs 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..20af3b52a5b --- /dev/null +++ b/rs/moq-mux/benches/jitter.rs @@ -0,0 +1,62 @@ +//! The jitter buffer's clock steering swept over track count and how many floors the rate +//! window holds. +use std::task::Poll; +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); From 094ae46f9306149e6d37039dfc60782835214f36 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 15 Oct 2026 18:03:21 -0700 Subject: [PATCH 5/5] fix(ts): release a packed PES against its first frame's deadline A packed PES must be whole in B by its first frame's decode time, but its send-ahead ran from its last frame's due slot. The PES is flushed half the delay early, so video pushed ahead of it leaves its start at its own due slot, and the release collapsed to that slot. A 640 kb/s E-AC-3 PES (12.8 kB against a 12,896-byte B at 2 Mb/s) then missed its deadline at 10 Mb/s. Shared first-frame keys make the first frame's due slot the consistent bound, so the unused last-frame slot goes. Also drop an unused import from the jitter bench. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/benches/jitter.rs | 1 - rs/moq-mux/src/container/ts/schedule.rs | 48 ++++++++++++++++++++----- 2 files changed, 40 insertions(+), 9 deletions(-) diff --git a/rs/moq-mux/benches/jitter.rs b/rs/moq-mux/benches/jitter.rs index 20af3b52a5b..66ad1492baa 100644 --- a/rs/moq-mux/benches/jitter.rs +++ b/rs/moq-mux/benches/jitter.rs @@ -1,6 +1,5 @@ //! The jitter buffer's clock steering swept over track count and how many floors the rate //! window holds. -use std::task::Poll; use std::time::Duration; use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main}; diff --git a/rs/moq-mux/src/container/ts/schedule.rs b/rs/moq-mux/src/container/ts/schedule.rs index cb62e3d9eae..f6dda245013 100644 --- a/rs/moq-mux/src/container/ts/schedule.rs +++ b/rs/moq-mux/src/container/ts/schedule.rs @@ -189,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. @@ -712,7 +708,6 @@ impl Schedule { self.units.push_back(Unit { pid, due, - settled: slot(last), timing, start, packets, @@ -751,14 +746,14 @@ impl Schedule { self.ended = false; } - /// The first slot `unit` may go out in: [`ahead`] of its last frame's due slot, 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 { if unit.late { return 0; } let ahead = ahead(self.buffers.get(&unit.pid), self.window); - unit.settled.saturating_sub(ahead).min(unit.start) + unit.due.saturating_sub(ahead).min(unit.start) } /// The slot [`Self::next`] lays out next, with `known` as it would be given. @@ -2494,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(); + } + } }