diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml index a1ef1633de..ebec3d9268 100644 --- a/.github/workflows/nightly.yml +++ b/.github/workflows/nightly.yml @@ -219,6 +219,10 @@ jobs: if: ${{ !cancelled() }} run: nix develop --command just rs test -p moq-mux benchmark_damage_pid_fanout --run-ignored only --success-output immediate + - name: TS scheduler benchmark smoke + if: ${{ !cancelled() }} + run: nix develop --command just rs bench-ts-schedule --test + - name: Container export benchmark smoke if: ${{ !cancelled() }} run: nix develop --command just rs bench-export --test diff --git a/quest/m1/README.md b/quest/m1/README.md index e49a32b3ab..725fdd009d 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -144,7 +144,6 @@ above them, since the epoch anchor requires it. - [Encoder colour](/quest/m1/color-model.md) - every moq-video encode path signals the colour its output actually has, or refuses instead of mislabelling - [TS fidelity and recovery](/quest/m1/ts-reliability/README.md) - both modes have explicit profiles, automated validation, loss tests, and a dedicated guide - [T-STD TS export](/quest/m1/tstd/README.md) - `moq export ts` is a proper remux that passes the supported T-STD buffer models, starting with a fixed `--delay` -- [ATSC AC-3](/quest/m1/ts-atsc-ac3.md) - `export ts` carries 48 kHz ATSC AC-3 at every A/52 rate up to 640 kb/s instead of aborting from 384 kb/s up - [Open-GOP leading pictures](/quest/m1/open-gop-leading-pictures.md) - a viewer joining at a recovery point drops the leading pictures it cannot decode; continuous viewers keep them - [Audio warmup](/quest/m1/audio-warmup.md) - a viewer joining an Opus rendition mid-stream never hears the unconverged first 80 ms - [#3021](/quest/m1/3021-moq-gst-anchor-generated-media-timelines-to-wall-clock.md) - moq-gst picks the broadcast wall epoch; a restarted source is a new epoch, not a forward re-anchor diff --git a/quest/m1/ts-atsc-ac3.md b/quest/m1/ts-atsc-ac3.md deleted file mode 100644 index 111cf4f495..0000000000 --- a/quest/m1/ts-atsc-ac3.md +++ /dev/null @@ -1,52 +0,0 @@ -# [M] TS export carries 48 kHz ATSC AC-3 at every A/52 rate - -## Goal - -`moq export ts` carries 48 kHz AC-3 as ATSC (`stream_type` 0x81, with the -2,592-byte main buffer of ATSC A/53 Part 5) at every A/52 rate up to 640 kb/s -without missing a decode deadline, and the output passes `compliance.py`'s strict -`tstd` check. Today every rate from 384 kb/s up aborts the export, and 384 and -448 kb/s 5.1 is ATSC's standard main audio. The carriage stays ATSC: DVB AC-3 -(5,696 B) already fits every rate, and E-AC-3 has -[its own quest](/quest/m2/ts-eac3.md). 48 kHz is the only rate ATSC carries. -A 44.1 or 32 kHz frame that fits the buffer goes out like any other, and one -that outgrows it on its own (2,786 and 3,840 B at 640 kb/s) fails the export -with an error naming both sizes, rather than the misleading "raise the delay or -the multiplex rate" or a switch to DVB carriage. - -## Plan - -Decided (2026-10-10), planned from #5207, whose multi-PID `--headroom` arm had -to drop its AC-3 to 192 kb/s. `Schedule::admit` -(`rs/moq-mux/src/container/ts/schedule.rs`) keeps a unit's bytes in its -decoder buffer until the 25 ms slot after its due slot is past, not until it -decodes. AC-3 frames are 32 ms apart, so two frames' due slots can be one -apart, and the second must then fit beside the whole first: at 384 kb/s the -two hold 3,312 B against 2,592. The T-STD (ISO 13818-1 2.4.2.3) frees B at the -decode instant, and a 448 kb/s frame takes about 7 ms at Rx = 2 Mb/s, leaving -some 25 ms of its 32 ms spacing. The B overflow (122 %) graded on that run's -partial output is the same coarse timing, not a separate bug. - -- Free each unit's bytes at its decode instant, for every PID with a decoder - buffer (AC-3, AAC, MP2, E-AC-3, and video's EB), so the schedule follows the - standard everywhere rather than special-casing AC-3. In the slot where a - buffer frees up, the packets admitted against the freed room go after that - instant: `Slot::layout` (from #5207) gains an earliest position per PID - beside its round-robin. Chosen over admitting by sub-slots (the same idea, - coarser) and over shrinking the 25 ms PCR grid, which would cost each PID - about a tenth of its Rx to the per-slot slack, 2.5 times the clock packets, - and every PCR-timing expectation. -- A unit bigger than its PID's decoder buffer can never be admitted, so the - export fails it when it is pushed, naming the unit's size and the buffer's, - rather than waiting for its deadline to pass. Once every source has ended, - the tail still goes out late, as today. Chosen over switching such a track - to DVB carriage, which would silently change what an ATSC receiver is handed. -- [Send-ahead](/quest/m1/tstd/send-ahead.md) bounds a unit's reach by what its - decoder buffer holds and changes the same code; whichever lands second - rebases onto the other. -- Test: schedule unit tests at 384, 448, 512, and 640 kb/s frame sizes against - the 2,592 B buffer, each frame finishing by its deadline, with the existing - decoder-buffer tests moved to the exact decode instant, and a 32 kHz - 448 kb/s frame (2,688 B) failing with both sizes named. Raise the - `--headroom` arm's AC-3 (`test/ts/run.sh`) from 192 to 640 kb/s, the worst - case, so CI grades it under strict `tstd` on every run. diff --git a/quest/m1/ts-live-clock.md b/quest/m1/ts-live-clock.md index e4cc47aef5..e553ba9735 100644 --- a/quest/m1/ts-live-clock.md +++ b/quest/m1/ts-live-clock.md @@ -56,4 +56,3 @@ public API and wire changes are not expected. - [Same-instance resume](/quest/m1/ts-same-instance-resume.md) - independent full-check failure - [Transport upgrade](/quest/m1/transport-upgrade/README.md) - existing session handover and epoch contract - [Mux-rate hold #5287](https://github.com/moq-dev/moq/pull/5287) - initial rate publication, and a finished import lingering for its max-age -- [TS AC-3 scheduling](/quest/m1/ts-atsc-ac3.md) - #5306 owns decoder deadlines and unknown-rate startup overflow diff --git a/quest/m1/ts-reliability/README.md b/quest/m1/ts-reliability/README.md index 9953eae72a..3bacc6ac2e 100644 --- a/quest/m1/ts-reliability/README.md +++ b/quest/m1/ts-reliability/README.md @@ -49,9 +49,9 @@ and the MPEG-TS draft consistent in each feature change. The planning children settle catalog and clock contracts before production changes. Their output adds independently completable implementation quests to this line; an audit finding is not authorization to guess a public API. -PMT fragmentation (#5220) and initial mux-rate measurement (#5287) have -landed; PSI runs, decoder-time admission, and catalog linger remain with -their current quests. +PMT fragmentation (#5220), initial mux-rate measurement (#5287), and +decoder-time admission (#5306) have landed; PSI runs and catalog linger +remain with their current quests. ## Required diff --git a/rs/justfile b/rs/justfile index 64ea888b60..9033c2395d 100644 --- a/rs/justfile +++ b/rs/justfile @@ -363,6 +363,11 @@ bench-stats *args: bench-latency: bash sh/rs/nextest.sh cargo nextest run --locked --release -p moq-bench --run-ignored only -E 'test(benchmark_observation_contention)' --success-output immediate +# Exact TS receiver scheduling across PID and queued-unit fanout. +[positional-arguments] +bench-ts-schedule *args: + cargo bench --locked -p moq-mux --bench ts_schedule -- "$@" + # Sweep container export tracks and queued frames independently. [positional-arguments] bench-export *args: diff --git a/rs/moq-mux/Cargo.toml b/rs/moq-mux/Cargo.toml index 6dd31df234..6f04203df8 100644 --- a/rs/moq-mux/Cargo.toml +++ b/rs/moq-mux/Cargo.toml @@ -51,6 +51,10 @@ tracing-test = { workspace = true, features = ["no-env-filter"] } name = "catalog" harness = false +[[bench]] +name = "ts_schedule" +harness = false + [[bench]] name = "export" harness = false diff --git a/rs/moq-mux/benches/ts_schedule.rs b/rs/moq-mux/benches/ts_schedule.rs new file mode 100644 index 0000000000..d7ad661178 --- /dev/null +++ b/rs/moq-mux/benches/ts_schedule.rs @@ -0,0 +1,69 @@ +//! TS admission and layout across PID count and access units per PID. + +use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main}; +use std::hint::black_box; +use std::time::Duration; + +mod export { + pub const PCR_INTERVAL: Duration = Duration::from_millis(25); + use std::time::Duration; +} +#[path = "../src/container/ts/schedule.rs"] +#[allow(dead_code)] +mod schedule; + +fn packet(pid: u16) -> [u8; 188] { + let mut packet = [0xff; 188]; + packet[..4].copy_from_slice(&[0x47, (pid >> 8) as u8, pid as u8, 0x10]); + packet +} + +fn admission(c: &mut Criterion) { + let mut group = c.benchmark_group("ts_schedule"); + group.measurement_time(Duration::from_secs(1)); + group.warm_up_time(Duration::from_millis(200)); + group.sample_size(10); + for pids in [1u16, 4, 16, 64] { + for frames in [1u128, 16, 64] { + group.bench_function(BenchmarkId::new(pids.to_string(), frames), |b| { + b.iter(|| { + let mut schedule = schedule::Schedule::new(Duration::from_millis(200)); + schedule.set_rate(Some(100_000_000)); + schedule.set_clock(1); + for pid in 1..=pids { + schedule.set_buffer( + pid, + schedule::Buffer { + rate: 2_000_000, + size: Some(1_000_000), + transport: 512, + }, + ); + } + for frame in 0..frames { + for pid in 1..=pids { + let at = (999 + frame * 40) * 1_000_000; + schedule + .push( + pid, + schedule::Timing { + by: at, + decode: at + 1_000_000, + }, + packet(pid).repeat(10), + false, + ) + .unwrap(); + } + } + while let Some(slot) = schedule.next(None).unwrap() { + black_box(slot.layout(&packet(1), 100, &packet(0x1fff))); + } + }); + }); + } + } + group.finish(); +} +criterion_group!(benches, admission); +criterion_main!(benches); diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index e31341e8df..d0d9cd7ac4 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -480,6 +480,7 @@ impl Timing { Some(Buffer { rate: rate.min(level.leak), size: Some(usize::try_from(cpb / 8).unwrap_or(usize::MAX)), + transport: 512, }) } @@ -1803,8 +1804,7 @@ impl Export { es_info, }; - self.schedule.set_buffer(0, Buffer::SYSTEM); - self.schedule.set_buffer(pmt_pid, Buffer::SYSTEM); + self.schedule.set_tables(pmt_pid); self.schedule.set_clock(pcr_pid); for track in self.tracks.values() { self.liveness.register(track.pid); @@ -1978,7 +1978,15 @@ impl Export { self.schedule.set_buffer(pid, buffer); } let by = decode.as_nanos().saturating_sub(track.drain().as_nanos()); - self.schedule.push(pid, by, out, keyframe); + self.schedule.push( + pid, + super::schedule::Timing { + by, + decode: decode.as_nanos(), + }, + out, + keyframe, + )?; Ok(()) } @@ -2499,7 +2507,11 @@ fn audio_buffer(config: &AudioConfig, kind: &Kind) -> Buffer { Kind::Eac3 => (2_000_000, 12_896), _ => (2_000_000, 3_584), }; - Buffer { rate, size: Some(size) } + Buffer { + rate, + size: Some(size), + transport: 512, + } } /// The DVB AC-3 descriptor's tag (ETSI EN 300 468 6.2.1), marking AC-3 as private data. @@ -2509,8 +2521,8 @@ const AC3_DESCRIPTOR: u8 = 0x6a; const TELETEXT_DESCRIPTOR: u8 = 0x56; /// A verbatim stream's buffers, for the kinds its descriptors identify: AC-3 as DVB private -/// data (ATSC A/52 Annex A 5.4), and teletext, whose transport buffer drains at 6.75 Mb/s -/// (ETSI EN 300 472 5). Any other stream has none the schedule knows. +/// data (ATSC A/52 Annex A 5.4), and teletext, whose 480-byte transport buffer drains at +/// 6.75 Mb/s (ETSI EN 300 472 5). Any other stream has none the schedule knows. fn verbatim_buffer(kind: &Kind, descriptors: &[catalog::Descriptor]) -> Option { let Kind::Verbatim { stream_type: 0x06, @@ -2524,10 +2536,12 @@ fn verbatim_buffer(kind: &Kind, descriptors: &[catalog::Descriptor]) -> Option Some(Buffer { rate: 2_000_000, size: Some(5_696), + transport: 512, }), TELETEXT_DESCRIPTOR => Some(Buffer { rate: 6_750_000, size: None, + transport: 480, }), _ => None, }) @@ -2999,6 +3013,7 @@ mod tests { Some(Buffer { rate: 6_750_000, size: None, + transport: 480, }) ); assert_eq!(verbatim_buffer(&kind, &[descriptor(0x0a)]), None); diff --git a/rs/moq-mux/src/container/ts/schedule.rs b/rs/moq-mux/src/container/ts/schedule.rs index 6611cb2291..7c69c9318d 100644 --- a/rs/moq-mux/src/container/ts/schedule.rs +++ b/rs/moq-mux/src/container/ts/schedule.rs @@ -1,6 +1,7 @@ //! Lays a TS export's packets onto its PCR grid. -use std::collections::{HashMap, VecDeque}; +use std::cmp::Reverse; +use std::collections::{BinaryHeap, HashMap, VecDeque}; use std::time::Duration; use mpeg2ts::ts::TsPacket; @@ -25,6 +26,10 @@ const PAYLOAD: usize = TsPacket::SIZE - 4; /// this the media had an outage rather than a coarse cadence, and a dense clock history for /// a span that carried no bytes only stalls anything pacing on the asserted values. const BACKFILL: u128 = 40; +/// The longest a transport buffer may go without emptying (ISO 13818-1 2.4.2.3). +const TB_EMPTY: Duration = Duration::from_secs(1); +/// One tick of the system clock, in whole nanoseconds rounded up. +const TICK: i128 = 1_000_000_000_u128.div_ceil(SYSTEM_CLOCK) as i128; /// The system clock in Hz, the unit of a PCR. const SYSTEM_CLOCK: u128 = 27_000_000; /// A PCR wraps at 2^33 ticks of 90 kHz, each 300 ticks of the system clock. @@ -69,6 +74,9 @@ pub(super) struct Buffer { /// first packet until it decodes. `None` bounds nothing, so the PID's units go out only /// in the slot before their own. pub size: Option, + /// The transport buffer in bytes: 512 (ISO 13818-1 2.4.2.3), 480 for teletext (ETSI EN + /// 300 472 5). + pub transport: usize, } impl Buffer { @@ -76,6 +84,7 @@ impl Buffer { pub const SYSTEM: Self = Self { rate: 1_000_000, size: None, + transport: 512, }; /// The packets one slot may carry on the PID: what its transport buffer passes on within @@ -131,12 +140,22 @@ impl SystemBuffer { } } +/// The last transport arrival and the decoder removal, in nanoseconds. +#[derive(Clone, Copy)] +pub(super) struct Timing { + /// When the last packet must enter the transport buffer. + pub by: u128, + /// When the decoder removes this unit. + pub decode: u128, +} + /// One access unit's packets, with the tables muxed ahead of it, still to go out. struct Unit { /// The elementary stream it carries, whose buffers bound it. pid: u16, - /// The last slot its packets may ride: the one that ends by its decode time. + /// The coarse deadline used by release windows and unpadded admission. due: 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. start: u128, @@ -173,6 +192,18 @@ fn pid(packet: &[u8]) -> u16 { u16::from(packet[1] & 0x1f) << 8 | u16::from(packet[2]) } +fn payload(packet: &[u8]) -> &[u8] { + if packet[3] & 0x10 == 0 { + return &[]; + } + let start = if packet[3] & 0x20 != 0 { + 5 + usize::from(packet[4]) + } else { + 4 + }; + &packet[start..] +} + /// One grid slot's worth of output. pub(super) struct Slot { pub index: u128, @@ -188,10 +219,13 @@ pub(super) struct Slot { pub units: Vec, /// The most packets a slot admission lets each bounded PID carry. pub caps: HashMap, + /// Exact positions of the media packets in a padded slot, counting the opening clock. + positions: Option>, } impl Slot { - /// The whole slot, opening with its `clock` packet, interleaved by smooth weighted + /// The whole slot, opening with its `clock` packet. Padded slots use the exact + /// positions planned against receiver buffers. Unpadded slots use smooth weighted /// round-robin: each PID (the clock packet counted on its own) and the null packets take /// positions in proportion to their count, as evenly as they can, every PID keeping its /// own order. @@ -209,6 +243,20 @@ impl Slot { /// delayed PIDs can still keep within their own transport buffers ([`Self::caps`]). pub fn layout(&self, clock: &[u8], pmt_pid: u16, null: &[u8]) -> Vec { let packets = self.packets.as_chunks::<{ TsPacket::SIZE }>().0; + if let Some(positions) = &self.positions { + let mut out = Vec::with_capacity((1 + packets.len() + self.nulls) * TsPacket::SIZE); + out.extend_from_slice(clock); + for (&position, packet) in positions.iter().zip(packets) { + while out.len() / TsPacket::SIZE < position { + out.extend_from_slice(null); + } + out.extend_from_slice(packet); + } + while out.len() / TsPacket::SIZE < 1 + packets.len() + self.nulls { + out.extend_from_slice(null); + } + return out; + } let mut tables = VecDeque::new(); // Each PID's packets in mux order: the clock's PID first, then the rest as they appear. let mut lanes: Vec<(u16, VecDeque)> = vec![(pid(clock), VecDeque::new())]; @@ -330,14 +378,13 @@ impl Slot { /// Packs muxed access units onto the PCR grid. /// /// 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: no more of a PID's packets -/// in a slot than its transport buffer passes on, and no more bytes than its decoder buffer -/// has room for until the units in it decode. A unit may go out up to the `window` before +/// 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. /// /// 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 due slot fails the export rather than +/// packets, and a unit that is not complete by its exact deadline fails the export rather than /// arrive late. Each slot's PCR is the time of its own first byte at the rate, so it matches /// its byte position exactly, and how many packets a slot carries and the PCR it opens with /// are functions of the slot alone, so two exporters of a broadcast lay a slot alike. Without a rate the stream is unpadded, its PCRs on the grid, @@ -360,8 +407,13 @@ pub(super) struct Schedule { buffers: HashMap, /// The PID every slot's clock packet rides on, from [`Self::set_clock`]. clock: Option, - /// Units sent into a decoder buffer: their PID, due slot and bytes held. + /// The PMT's PID, which shares TBsys with the PAT, from [`Self::set_tables`]. + pmt: Option, + /// Units sent into a decoder buffer: their PID, decode nanoseconds and bytes held. decoding: VecDeque<(u16, u128, usize)>, + /// When each transport buffer's last admitted packet finishes leaving it, and when it + /// last started filling from empty. + transport: HashMap, /// Every source has ended, so a unit that misses its deadline is the tail going out late /// rather than an output that cannot keep up. ended: bool, @@ -378,7 +430,9 @@ impl Schedule { last: None, buffers: HashMap::new(), clock: None, + pmt: None, decoding: VecDeque::new(), + transport: HashMap::new(), ended: false, } } @@ -399,23 +453,66 @@ impl Schedule { self.buffers.insert(pid, buffer); } + /// Hold the program tables, the PAT and the PMT on `pmt_pid`, to the one TBsys they share. + pub fn set_tables(&mut self, pmt_pid: u16) { + self.buffers.insert(0, Buffer::SYSTEM); + self.buffers.insert(pmt_pid, Buffer::SYSTEM); + self.pmt = Some(pmt_pid); + } + + /// The transport buffer `pid`'s packets enter: TBsys for the reserved system PIDs (PAT, + /// CAT, TSDT, IPMP) and the PMT, else its own. + fn transport_buffer(&self, pid: u16) -> u16 { + match pid <= 3 || self.pmt == Some(pid) { + true => 0, + false => pid, + } + } + /// Count every slot's clock packet, which rides on `pid`, against that PID's transport - /// buffer, as long as the PID keeps a packet a slot for its media. + /// buffer. Unpadded admission retains at least one media packet per slot. pub fn set_clock(&mut self, pid: u16) { self.clock = Some(pid); } /// Queue an access unit on `pid` (whole TS packets, the tables muxed ahead of it first) - /// whose last byte has to arrive by `by` nanoseconds: its decode time, less what the - /// receiver needs to pass the last packet on. - pub fn push(&mut self, pid: u16, by: u128, packets: Vec, keyframe: bool) { - let due = slot(by); - self.horizon = self.horizon.max(Some(due)); + /// with its exact transport deadline and decoder removal time. + pub fn push(&mut self, pid: u16, timing: Timing, packets: Vec, keyframe: bool) -> anyhow::Result<()> { + if packets.is_empty() { + return Ok(()); + } + let due = slot(timing.by); + // An ended source relaxes deadlines, never buffer sizes: such a unit never fits. + if let Some(size) = self.buffers.get(&pid).and_then(|buffer| buffer.size) { + let on_pid = || { + packets + .as_chunks::<{ TsPacket::SIZE }>() + .0 + .iter() + .filter(|packet| self::pid(*packet) == pid) + }; + let bytes: usize = on_pid().map(|packet| payload(packet).len()).sum(); + if bytes > size { + // Only the error names the elementary size, so the PES headers are parsed here. + let headers: usize = on_pid() + .filter(|packet| packet[1] & 0x40 != 0) + .map(|packet| payload(packet)) + .filter(|body| body.starts_with(&[0, 0, 1]) && body.len() >= 9) + .map(|body| 9 + usize::from(body[8])) + .sum(); + let elementary = bytes.saturating_sub(headers); + anyhow::bail!( + "MPEG-TS access unit on PID {pid} has {elementary} payload bytes ({bytes} bytes with PES headers), exceeding its {size}-byte decoder buffer" + ); + } + } + self.horizon = self.horizon.max(Some(timing.by.div_ceil(PCR_INTERVAL.as_nanos()))); let start = self.last.map_or(due, |last| (last + 1).min(due)); self.last = Some(due); self.units.push_back(Unit { pid, due, + timing, start, packets, sent: 0, @@ -423,6 +520,7 @@ impl Schedule { late: self.next.is_some_and(|next| due < next), held: 0, }); + Ok(()) } /// Whether a unit is still to go out. @@ -443,6 +541,7 @@ impl Schedule { pub fn clear(&mut self) { self.units.clear(); self.decoding.clear(); + self.transport.clear(); self.next = None; self.horizon = None; self.last = None; @@ -498,35 +597,19 @@ impl Schedule { return Ok(None); } self.next = Some(index + 1); + if let Some(rate) = self.rate { + return self.padded(index, rate).map(Some); + } - let (take, nulls, pcr) = match self.rate { - Some(rate) => { - let allowed = (packets_before(index + 1, rate) - packets_before(index, rate)).max(1); - let take = self.admit(index, allowed as usize - 1); - if let Some(unit) = self.missed(index, &take) - && !self.ended - { - anyhow::bail!( - "MPEG-TS output missed a decode deadline on PID {} at {rate} b/s; raise the delay or the multiplex rate", - unit.pid, - ); - } - let media: usize = take.iter().sum(); - (take, allowed as usize - 1 - media, rate_pcr(index, rate)) - } - None => { - let mut take = self.admit(index, usize::MAX); - for (unit, take) in self.units.iter_mut().zip(take.iter_mut()) { - if unit.due <= index && *take < unit.remaining() { - if self.buffers.get(&unit.pid).is_some_and(|buffer| buffer.size.is_some()) { - unit.held += unit.packets_on(unit.sent + *take..unit.count(), unit.pid) * PAYLOAD; - } - *take = unit.remaining(); - } + let mut take = self.admit(index, usize::MAX); + for (unit, take) in self.units.iter_mut().zip(take.iter_mut()) { + if unit.due <= index && *take < unit.remaining() { + if self.buffers.get(&unit.pid).is_some_and(|buffer| buffer.size.is_some()) { + unit.held += unit.packets_on(unit.sent + *take..unit.count(), unit.pid) * PAYLOAD; } - (take, 0, grid_pcr(index)) + *take = unit.remaining(); } - }; + } let mut packets = Vec::new(); let mut keyframe = false; @@ -547,14 +630,14 @@ impl Schedule { unit.sent += take; } for unit in self.units.iter().filter(|unit| unit.remaining() == 0 && unit.held > 0) { - self.decoding.push_back((unit.pid, unit.due, unit.held)); + self.decoding.push_back((unit.pid, unit.timing.decode, unit.held)); } self.units.retain(|unit| unit.remaining() > 0); Ok(Some(Slot { index, - pcr: pcr as u64, + pcr: grid_pcr(index) as u64, packets, - nulls, + nulls: 0, keyframe, units, caps: self @@ -562,18 +645,227 @@ impl Schedule { .iter() .map(|(&pid, buffer)| (pid, buffer.per_slot())) .collect(), + positions: None, })) } + /// Plan the packets and null positions together, against the actual transport clock. + /// Decoder removal and the delivery deadline stay exact even between PCR grid points. + fn padded(&mut self, index: u128, rate: u64) -> anyhow::Result { + let first = packets_before(index, rate); + let count = (packets_before(index + 1, rate) - first).max(1) as usize; + let time = |position: usize| { + ((first + position as u128) * TsPacket::SIZE as u128 * 8 * 1_000_000_000 / u128::from(rate)) as i128 + - PCR_INTERVAL.as_nanos() as i128 + }; + let mut out = Slot { + index, + pcr: rate_pcr(index, rate) as u64, + packets: Vec::new(), + nulls: count - 1, + keyframe: false, + units: Vec::new(), + caps: HashMap::new(), + positions: Some(Vec::new()), + }; + // The transport buffer the clock packet rides and how long it holds it, which every + // slot's clock packet may add to a busy stretch without the chance to wait. + let mut clock = None; + if let Some(pid) = self.clock + && let Some(buffer) = self.buffers.get(&pid) + { + let duration = (TsPacket::SIZE as u128 * 8 * 1_000_000_000).div_ceil(u128::from(buffer.rate)) as i128; + let tb = self.transport_buffer(pid); + let (free, since) = self.transport.entry(tb).or_insert((time(0), time(0))); + if *free + TICK <= time(0) { + *since = time(0); + } + *free = ((*free).max(time(0)) + duration).max(time(1)); + clock = Some((tb, duration)); + } + // One head per PID preserves decode order without walking all future units at every + // transport position. Removal events keep occupancy work proportional to units. + let mut lanes: Vec> = Vec::new(); + let mut lane_by_pid = HashMap::new(); + let mut occupancy: HashMap = HashMap::new(); + let mut removals = Vec::new(); + self.decoding.retain(|&(_, decode, _)| decode as i128 > time(0)); + for &(pid, decode, held) in &self.decoding { + *occupancy.entry(pid).or_default() += held; + removals.push((decode, pid, None, held)); + } + for (i, unit) in self.units.iter().enumerate() { + let lane = *lane_by_pid.entry(unit.pid).or_insert_with(|| { + lanes.push(VecDeque::new()); + lanes.len() - 1 + }); + lanes[lane].push_back(i); + if unit.timing.decode as i128 > time(0) { + *occupancy.entry(unit.pid).or_default() += unit.held; + removals.push((unit.timing.decode, unit.pid, Some(i), 0)); + } + } + removals.sort_by_key(|event| event.0); + let mut removals = removals.into_iter().peekable(); + let mut ready: BinaryHeap> = lanes + .iter() + .enumerate() + .filter_map(|(lane, units)| { + let unit = &self.units[*units.front()?]; + (self.release(unit) <= index).then_some(Reverse((unit.timing.by, lane))) + }) + .collect(); + let mut waiting: BinaryHeap> = BinaryHeap::new(); + let mut position = 1; + while position < count { + let now = time(position); + while let Some((_, pid, unit, held)) = removals.next_if(|event| event.0 as i128 <= now) { + let held = unit.map_or(held, |i| std::mem::take(&mut self.units[i].held)); + *occupancy.get_mut(&pid).unwrap() -= held; + } + while waiting.peek().is_some_and(|Reverse((at, _))| *at <= now) { + let Reverse((_, lane)) = waiting.pop().unwrap(); + ready.push(Reverse((self.units[*lanes[lane].front().unwrap()].timing.by, lane))); + } + let mut emitted = false; + while let Some(Reverse((_, lane))) = ready.pop() { + let i = *lanes[lane].front().unwrap(); + let unit = &self.units[i]; + let packet = &unit.packets[unit.sent * TsPacket::SIZE..(unit.sent + 1) * TsPacket::SIZE]; + let pid = self::pid(packet); + let buffer = self.buffers.get(&pid); + let duration = buffer.map_or(0, |buffer| { + (TsPacket::SIZE as u128 * 8 * 1_000_000_000).div_ceil(u128::from(buffer.rate)) + }); + let tb = self.transport_buffer(pid); + let (last, since) = self.transport.get(&tb).copied().unwrap_or((now, now)); + // A receiver reads time off the PCR, to a system clock tick, so the buffer counts as + // empty only once its last packet has left a tick before this one arrives. + let empty = last + TICK <= now; + let since = if empty { now } else { since }; + let drained = last.max(now) + duration as i128; + if let Some(buffer) = buffer { + let capacity = buffer.transport as u128 * 8 * 1_000_000_000 / u128::from(buffer.rate); + if drained - now > capacity as i128 { + let ready = drained - capacity as i128; + waiting.push(Reverse((ready, lane))); + continue; + } + // The next slot's mandatory clock packet must fit even at a full media PID. + if self.clock == Some(pid) && drained > time(count) + capacity as i128 - duration as i128 { + continue; + } + } + // No byte leaves the transport buffer before it has arrived: below Rx, the + // multiplex rate decides when the packet's last byte gets there. + let done = drained.max(time(position + 1)); + // A transport buffer must empty at least once a second (2.4.2.3), so a PID kept + // busy that long waits for it to. Every later slot's clock packet that arrives + // before the buffer empties extends the stretch without the chance to wait. + let mut busy = done; + if let Some((_, duration)) = clock.filter(|&(clock, _)| clock == tb) { + let mut next = index + 1; + while busy - since <= TB_EMPTY.as_nanos() as i128 { + let start = (packets_before(next, rate) - first) as usize; + if busy + TICK <= time(start) { + break; + } + busy = (busy.max(time(start)) + duration).max(time(start + 1)); + next += 1; + } + } + if busy - since > TB_EMPTY.as_nanos() as i128 { + waiting.push(Reverse((last + TICK, lane))); + continue; + } + let held = if pid == unit.pid && buffer.is_some_and(|buffer| buffer.size.is_some()) { + payload(packet).len() + } else { + 0 + }; + if let Some(size) = buffer.and_then(|buffer| buffer.size) + && occupancy.get(&pid).copied().unwrap_or(0) + held > size + { + if let Some(event) = removals.peek() { + let ready = event.0 as i128; + waiting.push(Reverse((ready, lane))); + } + 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" + ); + } + self.transport.insert(tb, (done, since)); + out.positions.as_mut().unwrap().push(position); + out.packets.extend_from_slice(packet); + out.nulls -= 1; + let unit = &mut self.units[i]; + if unit.sent == 0 { + out.keyframe |= unit.keyframe; + if unit.packets_on(0..unit.count(), unit.pid) > 0 { + out.units.push(unit.pid); + } + } + unit.sent += 1; + if unit.timing.decode as i128 > now { + unit.held += held; + *occupancy.entry(unit.pid).or_default() += held; + } + if unit.remaining() == 0 { + lanes[lane].pop_front(); + } + if let Some(&i) = lanes[lane].front() { + let unit = &self.units[i]; + if self.release(unit) <= index { + ready.push(Reverse((unit.timing.by, lane))); + } + } + emitted = true; + break; + } + if emitted { + position += 1; + } else if let Some(Reverse((next, _))) = waiting.peek() { + // Nothing can use intervening null positions. Jump to the first packet at the + // receiver event instead of scanning the multiplex rate's worth of padding. + let absolute = (*next + PCR_INTERVAL.as_nanos() as i128).max(0) as u128; + let packet = (absolute * u128::from(rate)).div_ceil(TsPacket::SIZE as u128 * 8 * 1_000_000_000); + position = (packet.saturating_sub(first) as usize).max(position + 1); + } else { + break; + } + } + let end = time(count); + if !self.ended + && let Some(unit) = self + .units + .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 + ); + } + for unit in self.units.iter().filter(|unit| unit.remaining() == 0 && unit.held > 0) { + self.decoding.push_back((unit.pid, unit.timing.decode, unit.held)); + } + self.units.retain(|unit| unit.remaining() > 0); + Ok(out) + } + /// 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, /// or not yet released) takes nothing more this slot, so each keeps its own order. fn admit(&mut self, index: u128, mut budget: usize) -> Vec { - // A unit due in slot k decodes during slot k + 1, so its bytes leave the decoder - // buffer once that slot is past. + // Unpadded layout has no exact packet clock until the slot is assembled, so it + // retains units until their decode instant is behind the opening grid point. let mut occupancy: HashMap = HashMap::new(); - self.decoding.retain(|&(_, due, _)| due + 1 >= index); + self.decoding + .retain(|&(_, decode, _)| decode >= index.saturating_sub(1) * PCR_INTERVAL.as_nanos()); for &(pid, _, held) in &self.decoding { *occupancy.entry(pid).or_default() += held; } @@ -601,14 +893,7 @@ impl Schedule { if stopped.contains(&unit.pid) { continue; } - // Once every source has ended, a unit past its deadline goes out whatever its decoder - // buffer holds: it is late already, and one bigger than the buffer would otherwise - // never finish, holding the end of the stream open on clock packets. - let size = self - .buffers - .get(&unit.pid) - .and_then(|buffer| buffer.size) - .filter(|_| !(self.ended && unit.due < index)); + let size = self.buffers.get(&unit.pid).and_then(|buffer| buffer.size); let mut n = 0; if self.release(unit) <= index { while n < unit.remaining() && budget > 0 { @@ -646,16 +931,6 @@ impl Schedule { } take } - - /// The first unit due by slot `index` that `take` leaves incomplete, unless it was already - /// late when pushed. - fn missed(&self, index: u128, take: &[usize]) -> Option<&Unit> { - self.units - .iter() - .zip(take) - .find(|(unit, take)| !unit.late && unit.due <= index && unit.sent + **take < unit.count()) - .map(|(unit, _)| unit) - } } #[cfg(test)] @@ -671,6 +946,13 @@ mod tests { packet.repeat(count) } + fn timing(ms: u128) -> Timing { + Timing { + by: self::ms(ms), + decode: self::ms(ms), + } + } + fn ms(ms: u128) -> u128 { ms * 1_000_000 } @@ -681,6 +963,7 @@ mod tests { const VIDEO: Buffer = Buffer { rate: 100_000_000, size: Some(1_000_000), + transport: 512, }; /// Every slot until the schedule is empty: (index, packets per PID, nulls). @@ -696,13 +979,193 @@ mod tests { slots } + /// Check actual wire positions against the receiver's transport buffer, clock included. + fn checked_transport(schedule: &mut Schedule, clock: u8) -> usize { + let rate = schedule.rate.unwrap(); + let rx = schedule.buffers[&1].rate; + let mut null = unit(0, 1); + (null[1], null[2]) = (0x1f, 0xff); + let (mut fill, mut previous, mut sent) = (0.0f64, None::, 0); + while let Some(slot) = schedule.next(None).unwrap() { + sent += slot + .packets + .as_chunks::<{ TsPacket::SIZE }>() + .0 + .iter() + .filter(|packet| pid(*packet) == 1) + .count(); + for (position, packet) in slot.layout(&unit(clock, 1), 100, &null).chunks(188).enumerate() { + let at = slot.pcr as f64 / SYSTEM_CLOCK as f64 + position as f64 * 1504.0 / rate as f64; + if let Some(previous) = previous { + fill = (fill - (at - previous) * rx as f64 / 8.0).max(0.0); + } + previous = Some(at); + if pid(packet) == 1 { + fill += 188.0; + } + assert!(fill <= 512.001, "PID1 transport buffer holds {fill} bytes"); + } + } + sent + } + + /// Distinct tune-in tables keep their copies and still fit TBsys, which the PAT and PMT + /// share. + #[test] + fn system_packets_are_paced_with_media_admission() { + for pids in [3u8, 4, 8] { + for count in [10, 100] { + for rate in [25_000_000, 50_000_000, 100_000_000] { + let mut schedule = Schedule::new(Duration::from_millis(200)); + schedule.set_rate(Some(rate)); + schedule.set_clock(1); + schedule.set_tables(100); + for pid in 1..=pids { + schedule.set_buffer(u16::from(pid), VIDEO); + let packets = [unit(0, 1), unit(100, 1), unit(pid, count)].concat(); + schedule.push(u16::from(pid), timing(1_000), packets, true).unwrap(); + } + let (mut fill, mut previous, mut tables) = (0.0f64, None::, [0; 2]); + while let Some(slot) = schedule.next(None).unwrap() { + for (position, packet) in slot.layout(&unit(1, 1), 100, &[0xff; 188]).chunks(188).enumerate() { + let at = slot.pcr as f64 / SYSTEM_CLOCK as f64 + position as f64 * 1504.0 / rate as f64; + if let Some(previous) = previous { + fill = (fill - (at - previous) * 125_000.0).max(0.0); + } + previous = Some(at); + for (i, system_pid) in [0, 100].into_iter().enumerate() { + if pid(packet) == system_pid { + fill += 188.0; + tables[i] += 1; + } + } + assert!( + fill <= 512.001, + "{pids} PIDs x {count} packets at {rate}: TBsys holds {fill} bytes" + ); + } + } + assert_eq!(tables, [usize::from(pids); 2]); + } + } + } + } + + /// Sections on a reserved system PID share TBsys with the PAT and PMT. + #[test] + fn reserved_system_pids_share_tbsys_with_the_tables() { + let rate = 25_000_000; + let mut schedule = Schedule::new(Duration::from_millis(200)); + schedule.set_rate(Some(rate)); + schedule.set_clock(7); + schedule.set_buffer(7, VIDEO); + schedule.set_tables(100); + schedule.set_buffer(1, Buffer::SYSTEM); + // A CAT muxed with the tables ahead of a keyframe, as the export writes them. + let packets = [unit(0, 1), unit(100, 2), unit(1, 3), unit(7, 1)].concat(); + schedule.push(7, timing(1_000), packets, true).unwrap(); + let (mut fill, mut previous) = (0.0f64, None::); + while let Some(slot) = schedule.next(None).unwrap() { + for (position, packet) in slot.layout(&unit(7, 1), 100, &[0xff; 188]).chunks(188).enumerate() { + let at = slot.pcr as f64 / SYSTEM_CLOCK as f64 + position as f64 * 1504.0 / rate as f64; + if let Some(previous) = previous { + fill = (fill - (at - previous) * 125_000.0).max(0.0); + } + previous = Some(at); + if [0, 1, 100].contains(&pid(packet)) { + fill += 188.0; + } + assert!(fill <= 512.001, "TBsys holds {fill} bytes"); + } + } + } + + /// A teletext burst at a high multiplex rate keeps within its 480-byte transport buffer, + /// measured as the receiver sees it: each packet arriving over its time on the wire. + #[test] + fn teletext_keeps_within_its_smaller_transport_buffer() { + let rate = 50_000_000; + let teletext = Buffer { + rate: 6_750_000, + size: None, + transport: 480, + }; + let mut schedule = Schedule::new(Duration::from_millis(100)); + schedule.set_rate(Some(rate)); + schedule.set_clock(7); + schedule.set_buffer(7, VIDEO); + schedule.set_buffer(9, teletext); + schedule.push(9, timing(1_000), unit(9, 40), false).unwrap(); + let wire = 1504.0 / rate as f64; + let drain = 1504.0 / teletext.rate as f64; + let (mut leave, mut peak, mut sent) = (f64::MIN, 0.0f64, 0); + while let Some(slot) = schedule.next(None).unwrap() { + for (position, packet) in slot.layout(&unit(7, 1), 100, &[0xff; 188]).chunks(188).enumerate() { + if pid(packet) != 9 { + continue; + } + let start = slot.pcr as f64 / SYSTEM_CLOCK as f64 + position as f64 * wire; + leave = (start + wire).max(leave.max(start) + drain); + peak = peak.max((leave - start - wire) * teletext.rate as f64 / 8.0); + sent += 1; + } + } + assert_eq!(sent, 40); + assert!(peak <= 480.5, "teletext transport buffer holds {peak:.1} bytes"); + } + + /// A PID with more than a second's backlog at its Rx still lets its transport buffer + /// empty once a second, as the receiver sees it, including when the slots' mandatory + /// clock packets ride the same PID. + #[test] + fn a_busy_transport_buffer_empties_every_second() { + let rate = 10_000_000; + for (rx, count, clock) in [(1_000_000, 2_000, 7), (150_000, 150, 9), (120_000, 150, 9)] { + let slow = Buffer { + rate: rx, + size: Some(1_000_000), + transport: 512, + }; + let mut schedule = Schedule::new(Duration::from_secs(4)); + schedule.set_rate(Some(rate)); + schedule.set_clock(clock); + schedule.set_buffer(7, VIDEO); + schedule.set_buffer(9, slow); + // More than a second of packets at the PID's Rx. + schedule.push(9, timing(5_000), unit(9, count), false).unwrap(); + let wire = 1504.0 / rate as f64; + let drain = 1504.0 / rx as f64; + let (mut leave, mut since, mut longest, mut sent) = (f64::MIN, 0.0, 0.0f64, 0); + while let Some(slot) = schedule.next(None).unwrap() { + let laid = slot.layout(&unit(clock as u8, 1), 100, &[0xff; 188]); + for (position, packet) in laid.chunks(188).enumerate() { + if pid(packet) != 9 { + continue; + } + let start = slot.pcr as f64 / SYSTEM_CLOCK as f64 + position as f64 * wire; + if leave <= start { + since = start; + } + leave = (start + wire).max(leave.max(start) + drain); + longest = longest.max(leave - since); + sent += 1; + } + } + assert!(sent >= count, "{rx}: sent {sent}"); + assert!( + longest <= 1.0, + "{rx}: the transport buffer stayed busy for {longest:.4} s" + ); + } + } + /// A unit goes out a window ahead of its due slot, as soon as it is released. #[test] fn a_unit_goes_out_as_soon_as_it_is_released() { let mut schedule = Schedule::new(Duration::from_millis(100)); schedule.set_rate(Some(RATE)); schedule.set_buffer(1, VIDEO); - schedule.push(1, ms(1_000), unit(1, 3), true); + schedule.push(1, timing(1_000), unit(1, 3), true).unwrap(); let slots = drain(&mut schedule); assert_eq!(slots[0].0, 36, "a window ahead of its due slot"); assert_eq!(slots[0].1[&1], 3); @@ -717,7 +1180,7 @@ mod tests { let mut schedule = Schedule::new(Duration::from_millis(100)); schedule.set_rate(Some(RATE)); schedule.set_buffer(1, VIDEO); - schedule.push(1, ms(1_000), unit(1, 100), true); + schedule.push(1, timing(1_000), unit(1, 100), true).unwrap(); let sent: Vec<_> = drain(&mut schedule) .into_iter() .map(|(index, per_pid, _)| (index, per_pid.get(&1).copied().unwrap_or(0))) @@ -731,7 +1194,7 @@ mod tests { let mut schedule = Schedule::new(Duration::from_millis(50)); schedule.set_rate(Some(RATE)); schedule.set_buffer(1, VIDEO); - schedule.push(1, ms(1_000), unit(1, 200), true); + 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"); } @@ -743,7 +1206,7 @@ mod tests { let mut schedule = Schedule::new(Duration::from_millis(50)); schedule.set_rate(Some(RATE)); schedule.set_buffer(1, VIDEO); - schedule.push(1, ms(1_000), unit(1, 200), true); + schedule.push(1, timing(1_000), unit(1, 200), true).unwrap(); schedule.set_ended(true); let sent: usize = drain(&mut schedule) .into_iter() @@ -752,29 +1215,37 @@ mod tests { assert_eq!(sent, 200, "the whole unit goes out"); } - /// Once every source has ended, a last unit bigger than its decoder buffer still goes out, - /// late, and the stream ends rather than run on with clock packets. + /// A unit bigger than its decoder buffer fails even once every source has ended: an ended + /// source relaxes deadlines, not buffer sizes. #[test] - fn a_last_unit_bigger_than_its_buffer_ends_the_stream() { + fn an_oversized_last_unit_fails_after_the_end() { let mut schedule = Schedule::new(Duration::from_millis(100)); schedule.set_rate(Some(RATE)); // Room for ten packets. let small = Buffer { rate: 100_000_000, size: Some(10 * PAYLOAD), + transport: 512, }; schedule.set_buffer(1, small); - schedule.push(1, ms(1_000), unit(1, 11), true); schedule.set_ended(true); - let mut sent = 0; - for _ in 0..100 { - let Some(slot) = schedule.next(None).unwrap() else { - break; - }; - sent += slot.packets.len() / TsPacket::SIZE; - } - assert_eq!(sent, 11, "the whole unit goes out"); - assert!(schedule.is_empty(), "and the stream ends"); + let error = schedule.push(1, timing(1_000), unit(1, 11), true).unwrap_err(); + assert!(error.to_string().contains("decoder buffer"), "{error}"); + } + + /// Below the PID's Rx, a packet's last byte arrives only at the end of its position on the + /// wire, so a unit still arriving at its decode time misses its deadline. + #[test] + fn a_packet_still_on_the_wire_at_decode_misses_its_deadline() { + // Four positions a slot, each 6.25 ms on the wire. + let mut schedule = Schedule::new(Duration::ZERO); + schedule.set_rate(Some(4 * PACKET)); + schedule.set_buffer(1, VIDEO); + // Slot 40 carries three packets; the fourth starts at 1,006.25 ms in slot 41 and + // arrives whole at 1,012.5 ms, after the unit decodes. + schedule.push(1, timing(1_010), unit(1, 4), true).unwrap(); + let error = std::iter::from_fn(|| schedule.next(None).transpose()).find_map(Result::err); + assert!(error.is_some(), "the last byte arrives after the decode time"); } /// The unit due first goes first, whatever order the PIDs were pushed in. @@ -784,8 +1255,8 @@ mod tests { schedule.set_rate(Some(RATE)); schedule.set_buffer(1, VIDEO); schedule.set_buffer(2, VIDEO); - schedule.push(1, ms(1_100), unit(1, 39), true); - schedule.push(2, ms(1_000), unit(2, 39), false); + schedule.push(1, timing(1_100), unit(1, 39), true).unwrap(); + schedule.push(2, timing(1_000), unit(2, 39), false).unwrap(); let sent: Vec<_> = drain(&mut schedule) .into_iter() .filter(|(_, per_pid, _)| !per_pid.is_empty()) @@ -804,14 +1275,13 @@ mod tests { let audio = Buffer { rate: 2_000_000, size: Some(100_000), + transport: 512, }; schedule.set_buffer(1, audio); for at in 0..20 { - schedule.push(1, ms(1_000 + at), unit(1, 4), false); - } - for (index, per_pid, _) in drain(&mut schedule) { - assert!(per_pid.get(&1).copied().unwrap_or(0) <= 32, "slot {index}: {per_pid:?}"); + schedule.push(1, timing(1_000 + at), unit(1, 4), false).unwrap(); } + assert_eq!(checked_transport(&mut schedule, 9), 80); } /// The clock packet opening every slot rides its PID's transport buffer too, so that PID @@ -824,46 +1294,57 @@ mod tests { let audio = Buffer { rate: 2_000_000, size: Some(100_000), + transport: 512, }; schedule.set_buffer(1, audio); schedule.set_clock(1); for at in 0..20 { - schedule.push(1, ms(1_000 + at), unit(1, 4), false); + schedule.push(1, timing(1_000 + at), unit(1, 4), false).unwrap(); } - let most = drain(&mut schedule) - .into_iter() - .map(|(_, per_pid, _)| per_pid.get(&1).copied().unwrap_or(0)) - .max(); - assert_eq!(most, Some(31)); + assert_eq!(checked_transport(&mut schedule, 1), 80); } /// A PID whose transport buffer passes on a single packet a slot still carries one of its /// own beside the clock packet, so its units finish on time and the stream ends. #[test] fn a_one_packet_pid_still_progresses_beside_the_clock() { - let mut schedule = Schedule::new(Duration::from_millis(100)); + let mut schedule = Schedule::new(Duration::from_millis(500)); schedule.set_rate(Some(RATE)); // 76.8 kb/s: one packet a slot. let slow = Buffer { rate: 76_800, size: Some(100_000), + transport: 512, }; schedule.set_buffer(1, slow); schedule.set_clock(1); - schedule.push(1, ms(1_000), unit(1, 4), true); - schedule.push(1, ms(1_200), unit(1, 4), false); - let mut sent = Vec::new(); - for _ in 0..100 { - let Some(slot) = schedule.next(None).unwrap() else { - break; - }; - sent.push(slot.packets.len() / TsPacket::SIZE); - } - assert!(sent.iter().all(|&n| n <= 1), "one packet a slot: {sent:?}"); - assert_eq!(sent.iter().sum::(), 8, "both units go out"); + schedule.push(1, timing(1_000), unit(1, 4), true).unwrap(); + schedule.push(1, timing(1_400), unit(1, 4), false).unwrap(); + assert_eq!(checked_transport(&mut schedule, 1), 8, "both units go out"); assert!(schedule.is_empty(), "and the stream ends"); } + #[test] + fn an_overloaded_clock_pid_fails_instead_of_overflowing_its_transport_buffer() { + let mut schedule = Schedule::new(Duration::from_millis(100)); + schedule.set_rate(Some(RATE)); + schedule.set_clock(1); + schedule.set_buffer( + 1, + Buffer { + rate: 76_800, + size: Some(100_000), + transport: 512, + }, + ); + schedule.push(1, timing(1_000), unit(1, 4), false).unwrap(); + schedule.push(1, timing(1_200), unit(1, 4), false).unwrap(); + let error = std::iter::from_fn(|| schedule.next(None).transpose()) + .find_map(Result::err) + .unwrap(); + assert!(error.to_string().contains("decode deadline"), "{error}"); + } + /// A PID never holds more in its decoder buffer than it has: the next unit waits for the /// one before it to decode. #[test] @@ -874,17 +1355,191 @@ mod tests { let small = Buffer { rate: 100_000_000, size: Some(10 * PAYLOAD), + transport: 512, }; schedule.set_buffer(1, small); - schedule.push(1, ms(1_000), unit(1, 8), true); - schedule.push(1, ms(1_100), unit(1, 8), false); + schedule.push(1, timing(1_000), unit(1, 8), true).unwrap(); + schedule.push(1, timing(1_100), unit(1, 8), false).unwrap(); let sent: Vec<_> = drain(&mut schedule) .into_iter() .filter_map(|(index, per_pid, _)| per_pid.get(&1).map(|&n| (index, n))) .collect(); // The first unit goes at once; the second fills what is left, then the rest once the // first decodes in slot 41. - assert_eq!(sent, [(32, 8), (36, 2), (42, 6)]); + assert_eq!(sent, [(32, 8), (36, 2), (41, 6)]); + } + + /// Packetize a PES with exactly `bytes` elementary bytes and adaptation stuffing. + fn audio(bytes: usize) -> Vec { + let mut pes = vec![0; 14 + bytes]; + pes[..9].copy_from_slice(&[0, 0, 1, 0xbd, 0, 0, 0x80, 0x80, 5]); + let mut out = Vec::new(); + for (index, body) in pes.chunks(PAYLOAD).enumerate() { + let mut packet = [0xff; TsPacket::SIZE]; + packet[..4].copy_from_slice(&[0x47, if index == 0 { 0x40 } else { 0 }, 1, 0x10]); + if body.len() < PAYLOAD { + packet[3] |= 0x20; + packet[4] = (PAYLOAD - body.len() - 1) as u8; + } + packet[TsPacket::SIZE - body.len()..].copy_from_slice(body); + out.extend(packet); + } + out + } + + #[test] + fn an_oversized_ac3_frame_names_both_sizes_at_admission() { + let mut schedule = Schedule::new(Duration::from_millis(200)); + schedule.set_buffer( + 1, + Buffer { + rate: 2_000_000, + size: Some(2_592), + transport: 512, + }, + ); + let error = schedule + .push(1, timing(1_000), audio(2_688), false) + .unwrap_err() + .to_string(); + assert!(error.contains("2688"), "{error}"); + assert!(error.contains("2592"), "{error}"); + assert!(!schedule.queued(), "refusal must not partially queue a unit"); + // The 14-byte PES header waits in B with its unit (BSn includes BSoh, 2.4.2.3), so + // 2,578 elementary bytes fill the buffer exactly and 2,580 overflow it. + assert!(schedule.push(1, timing(1_000), audio(2_580), false).is_err()); + for bytes in [2_304, 2_436, 2_578] { + schedule.push(1, timing(1_000), audio(bytes), false).unwrap(); + } + } + + /// A unit with no packets, such as a dropped malformed section, has nothing to place. + #[test] + fn an_empty_unit_is_skipped() { + for rate in [Some(RATE), None] { + let mut schedule = Schedule::new(Duration::from_millis(100)); + schedule.set_rate(rate); + schedule.set_buffer(1, VIDEO); + schedule.push(1, timing(1_000), Vec::new(), false).unwrap(); + schedule.push(1, timing(1_040), unit(1, 2), true).unwrap(); + let sent: usize = drain(&mut schedule) + .iter() + .filter_map(|(_, per_pid, _)| per_pid.get(&1)) + .sum(); + assert_eq!(sent, 2, "{rate:?}"); + } + } + + #[test] + fn freed_decoder_room_is_used_after_the_exact_in_slot_instant() { + let mut schedule = Schedule::new(Duration::from_millis(200)); + schedule.set_rate(Some(25_000_000)); + schedule.set_buffer( + 1, + Buffer { + rate: 2_000_000, + size: Some(2 * PAYLOAD), + transport: 512, + }, + ); + schedule + .push( + 1, + Timing { + by: ms(1_006), + decode: ms(1_007), + }, + unit(1, 2), + false, + ) + .unwrap(); + schedule + .push( + 1, + Timing { + by: ms(1_038), + decode: ms(1_039), + }, + unit(1, 2), + false, + ) + .unwrap(); + let mut sent = Vec::new(); + while let Some(slot) = schedule.next(None).unwrap() { + for &position in slot.positions.as_ref().unwrap() { + sent.push(slot.pcr as f64 / SYSTEM_CLOCK as f64 + position as f64 * 1504.0 / 25_000_000.0); + } + } + assert_eq!(sent.len(), 4); + assert!(sent[1] < 1.007); + assert!(sent[2] >= 1.007 && sent[2] < 1.025, "exact release: {sent:?}"); + } + + /// The ATSC buffer holds each legal 48 kHz AC-3 frame until its own decode instant. + #[test] + fn atsc_ac3_frames_finish_at_every_high_bitrate() { + for bitrate in [384_000usize, 448_000, 512_000, 640_000] { + let mut schedule = Schedule::new(Duration::from_millis(200)); + schedule.set_rate(Some(25_000_000)); + schedule.set_buffer( + 1, + Buffer { + rate: 2_000_000, + size: Some(2_592), + transport: 512, + }, + ); + let bytes = bitrate / 8 * 32 / 1_000; + let packets = (bytes + 14).div_ceil(PAYLOAD); + for frame in 0..64 { + schedule + .push( + 1, + Timing { + by: ms(999 + frame * 32), + decode: ms(1_000 + frame * 32), + }, + audio(bytes), + false, + ) + .unwrap(); + } + let mut sent = 0; + let mut frames = 0; + let mut held = [0usize; 64]; + let mut free = 0.0f64; + while let Some(slot) = schedule + .next(None) + .unwrap_or_else(|error| panic!("{bitrate} b/s: {error}")) + { + for (&position, packet) in slot + .positions + .as_ref() + .unwrap() + .iter() + .zip(slot.packets.as_chunks::<{ TsPacket::SIZE }>().0) + { + let at = slot.pcr as f64 / SYSTEM_CLOCK as f64 + position as f64 * 1504.0 / 25_000_000.0; + if packet[1] & 0x40 != 0 { + frames += 1; + } + let frame = frames - 1; + free = free.max(at) + 1504.0 / 2_000_000.0; + let decode = 1.0 + frame as f64 * 0.032; + assert!(free <= decode + 1e-7, "{bitrate} frame{frame}: {free} > {decode}"); + held[frame] += payload(packet).len(); + let occupancy: usize = held + .iter() + .enumerate() + .filter(|(i, _)| 1.0 + *i as f64 * 0.032 > at) + .map(|(_, held)| held) + .sum(); + assert!(occupancy <= 2_592, "{bitrate} at{at}: decoder holds{occupancy}"); + sent += 1; + } + } + assert_eq!(sent, 64 * packets); + } } /// A PID with no decoder buffer to respect goes out only in the slot before its own. @@ -892,7 +1547,7 @@ mod tests { fn an_unbounded_pid_goes_just_before_it_is_due() { let mut schedule = Schedule::new(Duration::from_millis(100)); schedule.set_rate(Some(RATE)); - schedule.push(7, ms(1_000), unit(7, 2), false); + schedule.push(7, timing(1_000), unit(7, 2), false).unwrap(); let sent: Vec<_> = drain(&mut schedule) .into_iter() .filter_map(|(index, per_pid, _)| per_pid.get(&7).map(|&n| (index, n))) @@ -912,7 +1567,7 @@ mod tests { schedule.set_rate(Some(rate)); schedule.set_buffer(1, VIDEO); for at in 0..40 { - schedule.push(1, ms(1_000 + 25 * at), unit(1, 1), false); + schedule.push(1, timing(1_000 + 25 * at), unit(1, 1), false).unwrap(); } let mut laid = 0u128; let mut first = None; @@ -937,8 +1592,8 @@ mod tests { fn a_rate_takes_over_from_the_next_slot() { let mut schedule = Schedule::new(Duration::from_millis(100)); schedule.set_buffer(1, VIDEO); - schedule.push(1, ms(1_000), unit(1, 8), true); - schedule.push(1, ms(1_100), unit(1, 100), false); + 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()) .map(|slot| (slot.index, slot.packets.len() / TsPacket::SIZE, slot.nulls)) @@ -964,10 +1619,11 @@ mod tests { Buffer { rate: 2_000_000, size: Some(100_000), + transport: 512, }, ); - schedule.push(1, ms(1_000), unit(1, 50), true); - schedule.push(1, ms(1_100), unit(1, 4), false); + schedule.push(1, timing(1_000), unit(1, 50), true).unwrap(); + schedule.push(1, timing(1_100), unit(1, 4), false).unwrap(); let sent: Vec<_> = drain(&mut schedule) .into_iter() .map(|(index, per_pid, nulls)| (index, per_pid.get(&1).copied().unwrap_or(0), nulls)) @@ -996,6 +1652,7 @@ mod tests { keyframe: false, units: vec![], caps: HashMap::new(), + positions: None, }; let mut null = unit(0, 1); (null[1], null[2]) = (0x1f, 0xff); @@ -1033,6 +1690,7 @@ mod tests { keyframe: true, units: vec![1, 3], caps: HashMap::new(), + positions: None, }; let mut null = unit(0, 1); (null[1], null[2]) = (0x1f, 0xff); @@ -1090,7 +1748,16 @@ mod tests { packets, keyframe: true, units: vec![video.into()], - caps: HashMap::from([(video.into(), Buffer { rate: RX, size: None }.per_slot())]), + caps: HashMap::from([( + video.into(), + Buffer { + rate: RX, + size: None, + transport: 512, + } + .per_slot(), + )]), + positions: None, }; let mut null = unit(0, 1); (null[1], null[2]) = (0x1f, 0xff); @@ -1128,10 +1795,12 @@ mod tests { Buffer { rate: 7_833_600, size: None, + transport: 512, }, Buffer { rate: 8_160_000, size: None, + transport: 512, }, ); let slot = Slot { @@ -1142,6 +1811,7 @@ mod tests { keyframe: true, units: vec![1, 2], caps: HashMap::from([(1, a.per_slot()), (2, b.per_slot())]), + positions: None, }; let mut null = unit(0, 1); (null[1], null[2]) = (0x1f, 0xff); diff --git a/test/ts/run.sh b/test/ts/run.sh index 24af28a874..5ac319a7c0 100755 --- a/test/ts/run.sh +++ b/test/ts/run.sh @@ -279,14 +279,14 @@ elif [[ -n "$HRD$HEADROOM" ]]; then # A broadcast feed's audio beside it: three MPEG-1 Layer II PIDs and an AC-3 one, in a # multiplex faster than the video's Rx (10.8 Mb/s). Only then can video packets arrive # faster than its transport buffer drains, so each slot has to interleave them with - # the audio and the nulls (moq-dev/moq#5142). At 192 kb/s an AC-3 frame stays within - # a third of the 2,592-byte buffer ffmpeg's ATSC carriage gives it. + # the audio and the nulls (moq-dev/moq#5142). At 640 kb/s the AC-3 frame nearly fills + # ATSC's 2,592-byte buffer, so the next frame needs its exact decode-time release. AUDIO=(-map 0:v) for k in 1 2 3 4; do INPUTS+=(-f lavfi -i "sine=frequency=$((330 * k)):sample_rate=48000") AUDIO+=(-map "$k:a") done - AUDIO+=(-ac 2 -c:a mp2 -b:a 192k -c:a:3 ac3) + AUDIO+=(-ac 2 -c:a mp2 -b:a 192k -c:a:3 ac3 -b:a:3 640k) fi ffmpeg -y -hide_banner -loglevel error \ "${INPUTS[@]}" \