diff --git a/quest/m1/ts-reliability/README.md b/quest/m1/ts-reliability/README.md index 8b797626fba..230b6ea2189 100644 --- a/quest/m1/ts-reliability/README.md +++ b/quest/m1/ts-reliability/README.md @@ -103,7 +103,6 @@ remain with their current quests. - [Access units](/quest/m1/ts-reliability/pes-access-units.md) - a PES carrying several access units publishes each at its own time - [PAT changes](/quest/m1/ts-reliability/pat-lifecycle.md) - a revised PAT never leaves a broadcast meaning another program - [Stream routes](/quest/m1/ts-reliability/stream-routes.md) - every PID gets a declared route, and what is not carried is counted -- [Program cap](/quest/m1/ts-reliability/program-cap.md) - a PAT listing every program number cannot create a broadcast per program ## Related diff --git a/quest/m1/ts-reliability/pat-lifecycle.md b/quest/m1/ts-reliability/pat-lifecycle.md index 7dd3802e6df..37d58b1e598 100644 --- a/quest/m1/ts-reliability/pat-lifecycle.md +++ b/quest/m1/ts-reliability/pat-lifecycle.md @@ -18,7 +18,9 @@ What the code does today: program to every program's `Import`; the removed program's `handle_pat` errors, `Programs::decode` propagates it, and the whole multiplex ingest stops, including unrelated programs. `Programs` sees each whole PAT first - (`Packet::pat`), so it can retire or start programs there instead. + (`Packet::pat`), so it can retire or start programs there instead. A + program still waiting for its PMT is already dropped there, or moved to + the PMT PID the later PAT names. - A PAT naming another `transport_stream_id`, or a single-program import's PAT naming another program, already refuses the import. @@ -33,8 +35,10 @@ Decided 2026-10-20 by the maintainer: If the program returns, it publishes a new epoch of its broadcast, never the retired instance's names for new content. A single-program import whose selected program leaves the PAT ends, as today. -- Open: a PAT that names a program whose PMT never arrives. Decide whether - the import refuses it or waits, and which clock bounds the wait (found in +- Open: a PAT that names a program whose PMT never arrives. `Programs` + creates a program's broadcast only once its PMT lists a stream, so such a + program waits holding only its PAT entry. Decide whether the import refuses + it instead, and which clock bounds the wait (found in [moq-dev/moq#6122](https://github.com/moq-dev/moq/pull/6122)). Rust: `rs/moq-mux` TS `Import::handle_pat` and `Programs`. diff --git a/quest/m1/ts-reliability/program-cap.md b/quest/m1/ts-reliability/program-cap.md deleted file mode 100644 index 8a27e201764..00000000000 --- a/quest/m1/ts-reliability/program-cap.md +++ /dev/null @@ -1,22 +0,0 @@ -# [XS] TS import bounds the programs a PAT can create - -## Goal - -`moq import ts --program all` creates a broadcast only for a program the input actually describes, instead of one per program number a hostile PAT names. - -Today `rs/moq-mux/src/container/ts/programs.rs` loops over every `pat.program_numbers()` and creates a broadcast, catalog producer, importer, and consumer for each. A sub-megabyte transport stream can list all 65,536 program numbers across PAT sections, so one input creates tens of thousands of live broadcasts. - -## Plan - -Decided 2026-10-22 from the security audit, with the maintainer: - -- Create a program's broadcast only once its PMT declares a supported stream, so the state scales with the PMT bytes the input carried, not with the PAT's list. `--program all` is the operator's opt-in to every described program, and an operator-selected program list narrows it; the default already refuses multi-program input. -- No new program-count constant or run budget: a cap below the PAT's domain would only refuse valid multiplexes. The session announce budget does not apply here, since it counts what a remote peer announces to us, not broadcasts `Programs` creates locally. -- Test with a synthesized PAT naming far more programs than it carries PMTs for, and assert only the described programs exist. - -Public API: none. Wire: none. - -## Related - -- [TS fidelity and recovery](/quest/m1/ts-reliability/README.md) - the TS import contract this belongs to -- [PMT lifecycle](/quest/m1/ts-reliability/pat-lifecycle.md) - the PAT handling beside it diff --git a/rs/moq-mux/benches/ts_programs.rs b/rs/moq-mux/benches/ts_programs.rs index 52be36723e5..138d738ce9a 100644 --- a/rs/moq-mux/benches/ts_programs.rs +++ b/rs/moq-mux/benches/ts_programs.rs @@ -1,6 +1,7 @@ //! `--program all` TS ingest across program count and per-program bitrate, over one second of //! a generated multiplex. Throughput is the whole input, so a parse that repeats per program -//! shows as a falling rate with program count. +//! shows as a falling rate with program count. A PAT that also names programs no PMT +//! describes shows what a program costs before it starts. use std::time::Duration; @@ -32,12 +33,13 @@ fn crc32(data: &[u8]) -> u32 { crc } -/// A long-form PSI section: `table_id`, the 16-bit extension, version 0, one section. -fn section(table_id: u8, extension: u16, body: &[u8]) -> Vec { +/// A long-form PSI section: `table_id`, the 16-bit extension, version 0, section `number` of +/// `last`. +fn section(table_id: u8, extension: u16, number: u8, last: u8, body: &[u8]) -> Vec { let length = 5 + body.len() + 4; let mut section = vec![table_id, 0xb0 | (length >> 8) as u8, length as u8]; section.extend_from_slice(&extension.to_be_bytes()); - section.extend_from_slice(&[0xc1, 0, 0]); + section.extend_from_slice(&[0xc1, number, last]); section.extend_from_slice(body); section.extend_from_slice(&crc32(§ion).to_be_bytes()); section @@ -51,6 +53,11 @@ fn video_pid(program: u16) -> u16 { 0x0200 + program } +/// The PMT PID a PAT names for an undescribed program, clear of every PID that carries data. +fn undescribed_pid(program: u16) -> u16 { + 0x1000 + program +} + #[derive(Default)] struct Mux { out: Vec, @@ -70,12 +77,14 @@ impl Mux { } fn psi(&mut self, pid: u16, section: &[u8]) { - assert!(section.len() < 184, "one packet per section"); - self.header(pid, true, false); - let end = self.out.len() + 184; - self.out.push(0); - self.out.extend_from_slice(section); - self.out.resize(end, 0xff); + let mut payload = vec![0]; + payload.extend_from_slice(section); + for (i, chunk) in payload.chunks(184).enumerate() { + self.header(pid, i == 0, false); + let end = self.out.len() + 184; + self.out.extend_from_slice(chunk); + self.out.resize(end, 0xff); + } } /// One access unit on `pid` in `packets` packets, its first carrying `pcr` (27 MHz) and a @@ -123,16 +132,27 @@ impl Mux { } /// One second of `programs` H.264 programs at `bitrate` bits per second each, interleaved -/// frame by frame, each program carrying its PCR on its video PID. -fn multiplex(programs: u16, bitrate: u64) -> Vec { +/// frame by frame, each program carrying its PCR on its video PID. The PAT also names +/// `undescribed` programs whose PMT never arrives. +fn multiplex(programs: u16, undescribed: u16, bitrate: u64) -> Vec { let packets = (bitrate / FPS / 8).div_ceil(188) as usize; let mut mux = Mux::default(); - let mut pat = Vec::new(); - for program in 1..=programs { - pat.extend_from_slice(&program.to_be_bytes()); - pat.extend_from_slice(&(0xe000 | pmt_pid(program)).to_be_bytes()); - } - let pat = section(0x00, 1, &pat); + let entries: Vec = (1..=programs) + .map(|program| (program, pmt_pid(program))) + .chain((programs + 1..=programs + undescribed).map(|program| (program, undescribed_pid(program)))) + .flat_map(|(program, pid)| { + let [a, b] = program.to_be_bytes(); + let [c, d] = (0xe000 | pid).to_be_bytes(); + [a, b, c, d] + }) + .collect(); + // A section holds at most 1021 bytes, so a long PAT spans several. + let chunks: Vec<_> = entries.chunks(250 * 4).collect(); + let last = (chunks.len() - 1) as u8; + let pat: Vec<_> = (0..) + .zip(&chunks) + .map(|(number, chunk)| section(0x00, 1, number, last, chunk)) + .collect(); let pmts: Vec<_> = (1..=programs) .map(|program| { let video = video_pid(program); @@ -140,12 +160,14 @@ fn multiplex(programs: u16, bitrate: u64) -> Vec { body.extend_from_slice(&[0xf0, 0x00, 0x1b]); body.extend_from_slice(&(0xe000 | video).to_be_bytes()); body.extend_from_slice(&[0xf0, 0x00]); - section(0x02, program, &body) + section(0x02, program, 0, 0, &body) }) .collect(); for frame in 0..FPS { if frame % PSI_EVERY == 0 { - mux.psi(0, &pat); + for pat in &pat { + mux.psi(0, pat); + } for (program, pmt) in (1..=programs).zip(&pmts) { mux.psi(pmt_pid(program), pmt); } @@ -180,26 +202,32 @@ fn programs(c: &mut Criterion) { group.warm_up_time(Duration::from_millis(300)); group.sample_size(10); for programs in [1u16, 4, 16] { - for mbps in [1u64, 4] { - let input = multiplex(programs, mbps * 1_000_000); - let (stats, _driver) = ingest(programs, &input); - assert!( - stats.streams.len() == usize::from(programs) && stats.streams.values().all(|s| s.units == FPS), - "every program delivers every frame" - ); - group.throughput(Throughput::Bytes(input.len() as u64)); - group.bench_function(BenchmarkId::new(format!("{programs}"), format!("{mbps}mbps")), |b| { - b.iter_batched( - setup, - |(mut programs, driver)| { - for chunk in input.chunks(188 * 7) { - programs.decode(chunk).unwrap(); - } - (programs, driver) - }, - BatchSize::PerIteration, - ) - }); + for undescribed in [0u16, 1000] { + for mbps in [1u64, 4] { + let input = multiplex(programs, undescribed, mbps * 1_000_000); + let (stats, _driver) = ingest(programs, &input); + assert!( + stats.streams.len() == usize::from(programs) && stats.streams.values().all(|s| s.units == FPS), + "every program delivers every frame" + ); + group.throughput(Throughput::Bytes(input.len() as u64)); + let name = match undescribed { + 0 => format!("{programs}"), + _ => format!("{programs}+{undescribed}"), + }; + group.bench_function(BenchmarkId::new(name, format!("{mbps}mbps")), |b| { + b.iter_batched( + setup, + |(mut programs, driver)| { + for chunk in input.chunks(188 * 7) { + programs.decode(chunk).unwrap(); + } + (programs, driver) + }, + BatchSize::PerIteration, + ) + }); + } } } group.finish(); diff --git a/rs/moq-mux/src/container/ts/health.rs b/rs/moq-mux/src/container/ts/health.rs index 230ba18bd34..e0ca98f5537 100644 --- a/rs/moq-mux/src/container/ts/health.rs +++ b/rs/moq-mux/src/container/ts/health.rs @@ -424,7 +424,7 @@ impl Sync { /// meaningless when spliced onto bytes that didn't follow them. Keeping one implementation /// keeps a subtle rule (which packets advance the counter, which retransmissions to ignore) /// from drifting into two, and `Continuity_count_error` reads the same verdict. -#[derive(Default)] +#[derive(Clone, Default)] pub(super) struct Continuity { /// Last continuity_counter seen on a packet with payload, to spot gaps. last_cc: Option, diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index b7a520929a1..1e8a8ec0ef8 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -297,6 +297,18 @@ impl Import { self } + /// Carry on the PMT `sections` began on `pid`, for an importer that starts mid-stream on the + /// packet after the ones `sections` read. + pub(super) fn with_pmt(mut self, pid: Pid, sections: SectionReassembler) -> Self { + self.pmt_sections.insert(pid, sections); + self + } + + /// Whether this importer reassembles the PMT sections on `pid`, and counts their CRC errors. + pub(super) fn reads_pmt(&self, pid: u16) -> bool { + Pid::new(pid).is_ok_and(|pid| self.pmt_sections.contains_key(&pid)) + } + /// A reservation on this program's timebase, so every stream shifts by its offset. fn reserve(&self) -> crate::catalog::Reserved { self.timebase.reserve() @@ -2640,7 +2652,7 @@ pub(super) fn payload(pkt: &[u8; TsPacket::SIZE]) -> Payload<'_> { /// header split, where section_length is not yet known), continuity-counter /// gaps, and adaptation-field discontinuities. Deliberately minimal: just enough to /// recover whole sections verbatim. -#[derive(Default)] +#[derive(Clone, Default)] pub(super) struct SectionReassembler { /// Bytes of the section currently being reassembled. Its 3-byte header (and /// thus section_length) may not all be present yet, so completeness is @@ -2650,9 +2662,23 @@ pub(super) struct SectionReassembler { span: Option, /// Shared with the PES path: a broken chain drops the partial either way. continuity: Continuity, + /// Whether the first section the latest packet completed began in an earlier packet. + resumed: bool, + /// Whether the partial section left assembling began in the latest packet. + fresh: bool, } impl SectionReassembler { + /// Whether the first section the latest packet completed began in an earlier packet. + pub(super) fn resumed(&self) -> bool { + self.resumed + } + + /// Whether the latest packet began the partial section left assembling. + pub(super) fn fresh(&self) -> bool { + self.fresh && !self.acc.is_empty() + } + /// Consume one 188-byte TS packet, appending every completed section to `out`. pub(super) fn push(&mut self, pkt: &[u8], out: &mut Vec>) { let mut completed = Vec::new(); @@ -2662,6 +2688,8 @@ impl SectionReassembler { fn push_at(&mut self, pkt: &[u8], source: Option<(u64, &Placement)>, out: &mut Vec<(Option, Vec)>) { let pkt: &[u8; 188] = pkt.try_into().expect("section packet must be 188 bytes"); + self.resumed = false; + self.fresh = false; match self.continuity.observe(pkt) { Continuation::Duplicate => return, // A packet flagged corrupt takes its own payload down with the partial: this is @@ -2702,7 +2730,9 @@ impl SectionReassembler { // progress (stream just joined, or a reset dropped the partial) these // bytes are an orphaned fragment, so skip straight to the pointer_field. self.acc.extend_from_slice(&payload[1..1 + ptr]); + let completed = out.len(); self.drain(out, source, end - (payload.len() - 1 - ptr) as u64); + self.resumed = out.len() > completed; } // The pointer_field is a hard section boundary: drop any leftover partial // and start the section it points to. @@ -2710,6 +2740,7 @@ impl SectionReassembler { self.acc.extend_from_slice(&payload[1 + ptr..]); self.span = source.and_then(|(_, placement)| placement.span(end - self.acc.len() as u64)); self.drain(out, source, end); + self.fresh = true; } else if !self.acc.is_empty() { // Continuation of the section in progress. A non-PUSI packet with nothing // in progress is unaligned (no pointer_field to resync on), so it is @@ -2717,7 +2748,10 @@ impl SectionReassembler { // discontinuity, or corrupt pointer dropped the partial, rather than // resuming on stray bytes that merely look like a section. self.acc.extend_from_slice(payload); + let completed = out.len(); self.drain(out, source, end); + self.resumed = out.len() > completed; + self.fresh = self.resumed; } } @@ -7084,7 +7118,7 @@ pub(super) mod test { /// One PES on `pid` carrying two whole MP2 frames: enough for the second to confirm the /// first, which is what the legacy path publishes on. - fn mp2_pes(pid: u16, cc: u8, pts: u64, fills: [u8; 2]) -> Vec { + pub(in crate::container::ts) fn mp2_pes(pid: u16, cc: u8, pts: u64, fills: [u8; 2]) -> Vec { let mut payload = mp2_frame(fills[0]); payload.extend_from_slice(&mp2_frame(fills[1])); audio_pes_packet(pid, cc, pts, &payload) @@ -7260,7 +7294,7 @@ pub(super) mod test { section } - const MP2: u8 = StreamType::Mpeg1Audio as u8; + pub(in crate::container::ts) const MP2: u8 = StreamType::Mpeg1Audio as u8; /// A PAT listing fifty programs, which takes two packets, then program 50's PMT (one /// MP2 stream on `0x0061`) and one PES on it. diff --git a/rs/moq-mux/src/container/ts/programs.rs b/rs/moq-mux/src/container/ts/programs.rs index 31e8dd9a278..d6afb0e286f 100644 --- a/rs/moq-mux/src/container/ts/programs.rs +++ b/rs/moq-mux/src/container/ts/programs.rs @@ -1,11 +1,12 @@ //! Every program of a multiplex, each imported as its own broadcast. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashMap, HashSet}; use anyhow::Context; use mpeg2ts::ts::Pid; -use super::demux::Demux; +use super::demux::{Demux, Packet}; +use super::import::SectionReassembler; use super::{Ext, Import, catalog::SI_PIDS, psi, stats}; use crate::catalog; @@ -13,9 +14,11 @@ use crate::catalog; /// own clock and catalog. /// /// Program `n` of `event.hang` publishes on `event/n.hang`, keeping the catalog suffix last so -/// format detection still sees it; a name without one gains a plain `/n`. Input ahead of that -/// PAT is dropped, as a single [`Import`] drops media ahead of its PSI. A program a later PAT -/// adds is not published; one it removes ends the import. +/// format detection still sees it; a name without one gains a plain `/n`. A program starts once +/// its PMT lists an elementary stream, so one the PAT names but the input never describes costs +/// nothing. Input ahead of that PMT is dropped, as a single [`Import`] drops media ahead of its +/// PSI; its SI arrives with the next repetition, as for an import joining mid-stream. A later PAT moves a waiting program to the PMT PID it names and drops one it no longer +/// lists. A program a later PAT adds is not published; removing a started one ends the import. /// /// The multiplex is parsed once: framing, continuity, the PAT and the SI sections are shared, /// and each program reads only the packets on the PIDs its PAT entry and PMT claim. @@ -24,7 +27,14 @@ pub struct Programs { name: moq_net::PathOwned, config: catalog::Config, demux: Demux, - /// Empty until the PAT arrives. + /// The latest whole PAT, from the first that lists a program, which seeds every program. + pat: Option, + /// The programs the first PAT lists and no PMT has started yet, by the PMT PID the latest + /// PAT names for them. + waiting: HashMap, + /// PMT sections dropped for a bad CRC on a PID no program read, by PID. + crc_errors: BTreeMap, + /// Empty until a PMT starts a program. programs: Vec, /// The programs reading each PID they claim, indexed by PID, in program order. The PAT and /// the SI PIDs go to every program instead. @@ -33,6 +43,16 @@ pub struct Programs { touched: Vec, } +/// The programs waiting for their PMT on one PID. +#[derive(Default)] +struct Waiting { + sections: SectionReassembler, + programs: HashSet, + /// Whether a program read this PID when the section still assembling began, and so counts + /// it if it is corrupt. + began_read: bool, +} + /// One program's broadcast. struct Program { /// The PAT's program number. @@ -78,6 +98,9 @@ impl Programs { name: name.as_path().to_owned(), config, demux: Demux::new(true), + pat: None, + waiting: HashMap::new(), + crc_errors: BTreeMap::new(), programs: Vec::new(), readers: vec![Vec::new(); 1 << 13], touched: Vec::new(), @@ -95,11 +118,10 @@ impl Programs { fn decode_with(&mut self, demux: &mut Demux, data: &[u8]) -> anyhow::Result<()> { demux.push(data); while let Some(packet) = demux.next() { - if self.programs.is_empty() { - match &packet.pat { - Some(pat) if !pat.program_numbers().is_empty() => self.start(pat)?, - _ => continue, - } + if let Some(pat) = &packet.pat { + self.wait(pat); + } else if !self.waiting.is_empty() { + self.describe(&packet)?; } // Every program reads the PAT, and the SI sections the parser reassembled. let everyone = packet.pid == Pid::PAT || (SI_PIDS.contains(&packet.pid) && !packet.sections.is_empty()); @@ -148,39 +170,150 @@ impl Programs { Ok(()) } - /// Create a broadcast for every program `pat` lists. - fn start(&mut self, pat: &psi::Pat) -> anyhow::Result<()> { - for program in pat.program_numbers() { - let name = program_broadcast(self.name.as_str(), program); - let mut broadcast = self - .origin - .create_broadcast(&name) - .with_context(|| format!("failed to create broadcast {name}"))?; - // TS carries undecoded elementary streams verbatim, so each catalog carries the - // `mpegts` extension, and owns its broadcast's catalog tracks. - let config = self - .config - .clone() - .with_catalog(catalog::hang::Catalog::::default()); - let catalog = catalog::Producer::new(&mut broadcast, config)?; - let import = Import::new(broadcast.clone(), catalog.reserve()).with_program(program); - // Each program is its own publisher instance for this run, so a restart - // replaces it rather than resuming into it. - let route = moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); - let listed = broadcast.consume().track(hang::Catalog::DEFAULT_NAME)?; - self.programs.push(Program { - number: program, - broadcast, - import, - catalog, - pending: Some((route, listed)), - claims: Vec::new(), - touched: false, + /// Wait for the PMT of each program the first whole PAT listing one names. A later PAT moves + /// a waiting program to the PMT PID it names, and drops one it no longer lists. + fn wait(&mut self, pat: &psi::Pat) { + // The PAT repeats several times a second, mostly unchanged. + if self.pat.as_ref() == Some(pat) { + return; + } + let mut listed = HashMap::new(); + for entry in &pat.programs { + // An importer takes a program's first entry, so a repeated number waits there alone. + if entry.program_number != 0 { + listed.entry(entry.program_number).or_insert(entry.pmt_pid.as_u16()); + } + } + if self.pat.is_none() { + if listed.is_empty() { + return; + } + for (&number, &pid) in &listed { + self.waiting.entry(pid).or_default().programs.insert(number); + } + } else { + let mut moved = Vec::new(); + self.waiting.retain(|pid, waiting| { + waiting.programs.retain(|number| match listed.get(number) { + Some(listed) if listed == pid => true, + Some(&listed) => { + moved.push((*number, listed)); + false + } + None => false, + }); + !waiting.programs.is_empty() }); + for (number, pid) in moved { + self.waiting.entry(pid).or_default().programs.insert(number); + } + } + self.pat = Some(pat.clone()); + } + + /// Read a packet on a waiting PMT PID, and start each program whose PMT it completes with a + /// stream listed. + /// + /// A started program reads this packet itself, from the reassembler as it stood before it, + /// so its own PMT path applies the PMT, counts the packet's corrupt sections, and starts its + /// clock from the PCR already seen. Of several revisions in the packet, the last decides. + fn describe(&mut self, packet: &Packet) -> anyhow::Result<()> { + let Some(waiting) = self.waiting.get_mut(&packet.pid) else { + return Ok(()); + }; + let before = waiting.sections.clone(); + let read = self.readers[usize::from(packet.pid)] + .iter() + .any(|&index| self.programs[index].import.reads_pmt(packet.pid)); + let began = waiting.began_read; + let mut sections = Vec::new(); + waiting.sections.push(packet.pkt, &mut sections); + let resumed = waiting.sections.resumed(); + if waiting.sections.fresh() { + waiting.began_read = read; + } + // Corrupt sections no program read from their start. + let mut corrupt = 0; + let mut latest: Vec = Vec::new(); + for (index, section) in sections.into_iter().enumerate() { + // Other tables may share the PID. + if section.first() != Some(&psi::PMT_TABLE_ID) { + continue; + } + if !psi::crc_ok(§ion) { + let read = if index == 0 && resumed { began } else { read }; + corrupt += u64::from(!read); + continue; + } + if let Some(pmt) = psi::Pmt::parse(§ion) + && waiting.programs.contains(&pmt.program_number) + { + latest.retain(|seen| seen.program_number != pmt.program_number); + latest.push(pmt); + } + } + latest.retain(|pmt| !pmt.streams.is_empty()); + if latest.is_empty() { + if corrupt > 0 { + *self.crc_errors.entry(packet.pid).or_default() += corrupt; + tracing::warn!(pid = packet.pid, "dropped a PSI section with a bad CRC"); + } + return Ok(()); + } + for pmt in &latest { + waiting.programs.remove(&pmt.program_number); + } + if waiting.programs.is_empty() { + self.waiting.remove(&packet.pid); + } + for pmt in latest { + self.start(pmt.program_number, packet.pid, before.clone())?; + } + // The programs just started read this packet, so they read any section it began. + if let Some(waiting) = self.waiting.get_mut(&packet.pid) { + waiting.began_read = true; } Ok(()) } + /// Create the broadcast of program `number`, its importer seeded with the PAT and carrying + /// on the PMT `sections` began on `pmt_pid`. + fn start(&mut self, number: u16, pmt_pid: u16, sections: SectionReassembler) -> anyhow::Result<()> { + let pat = self.pat.as_ref().expect("a program waits on its PAT"); + let name = program_broadcast(self.name.as_str(), number); + let mut broadcast = self + .origin + .create_broadcast(&name) + .with_context(|| format!("failed to create broadcast {name}"))?; + // TS carries undecoded elementary streams verbatim, so each catalog carries the + // `mpegts` extension, and owns its broadcast's catalog tracks. + let config = self + .config + .clone() + .with_catalog(catalog::hang::Catalog::::default()); + let catalog = catalog::Producer::new(&mut broadcast, config)?; + let pmt_pid = Pid::new(pmt_pid).expect("a packet's PID fits 13 bits"); + let mut import = Import::new(broadcast.clone(), catalog.reserve()) + .with_program(number) + .with_pmt(pmt_pid, sections); + import.handle_pat(pat)?; + // Each program is its own publisher instance for this run, so a restart + // replaces it rather than resuming into it. + let route = moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()); + let listed = broadcast.consume().track(hang::Catalog::DEFAULT_NAME)?; + self.programs.push(Program { + number, + broadcast, + import, + catalog, + pending: Some((route, listed)), + claims: Vec::new(), + touched: false, + }); + self.reclaim(self.programs.len() - 1); + Ok(()) + } + /// Move program `index` in [`Self::readers`] to the PIDs it claims now. fn reclaim(&mut self, index: usize) { let program = &mut self.programs[index]; @@ -200,7 +333,8 @@ impl Programs { /// /// The multiplex is graded once. Programs sharing a PMT PID read the same PMT sections, so /// a dropped section counts once however many read it, and programs sharing a PCR PID - /// count its faults once too. + /// count its faults once too. PMT sections dropped before any program read their PID are + /// counted here. pub fn stats(&self) -> stats::Snapshot { let mut stats = stats::Snapshot::default(); let mut crc_errors = BTreeMap::::new(); @@ -213,7 +347,7 @@ impl Programs { } errors.max(program.import.errors()); } - stats.crc_error = crc_errors.values().sum::() + self.demux.pat_crc_error(); + stats.crc_error = crc_errors.values().chain(self.crc_errors.values()).sum::() + self.demux.pat_crc_error(); errors.report(&mut stats); stats } @@ -225,7 +359,7 @@ impl Programs { /// Fails when the input ended before its PAT listed a program or any program declared a track. pub fn finish(&mut self) -> anyhow::Result> { self.demux.finish(); - anyhow::ensure!(!self.programs.is_empty(), "input ended before its PAT listed a program"); + anyhow::ensure!(self.pat.is_some(), "input ended before its PAT listed a program"); let mut announced = false; let mut finished = Vec::with_capacity(self.programs.len()); for program in &mut self.programs { @@ -266,7 +400,7 @@ mod test { use crate::catalog::hang::Container; use crate::container::Consumer; use crate::container::ts::import::test::{ - corrupt, fifty_programs, pat_section, pmt_section, psi_packets, two_programs, two_section_pat, + MP2, corrupt, fifty_programs, mp2_pes, pat_section, pmt_section, psi_packets, two_programs, two_section_pat, }; #[tokio::test(start_paused = true)] @@ -302,7 +436,7 @@ mod test { demux.push(data); while let Some(packet) = demux.next() { if let Some(pat) = packet.pat { - return Some(pat.program_numbers()); + return Some(pat.programs.iter().map(|entry| entry.program_number).collect()); } } None @@ -319,10 +453,10 @@ mod test { assert_eq!(pat_programs(&data[..100]), None); } - /// A PAT too big for one packet is read whole, and each program is handed all of it, so - /// the program listed in its second packet publishes. + /// A PAT too big for one packet is read whole, so the program listed in its second packet + /// publishes, and the forty-nine without a PMT start nothing. #[tokio::test] - async fn a_pat_spanning_two_packets_starts_every_program() { + async fn a_pat_spanning_two_packets_is_read_whole() { let origin = crate::source::produce_origin(); let mut programs = Programs::new(origin.clone(), "event.hang", catalog::Config::default()); let input = fifty_programs(); @@ -330,7 +464,7 @@ mod test { programs.decode(chunk).unwrap(); } let _finished = programs.finish().unwrap(); - assert_eq!(programs.programs.len(), 50); + assert_eq!(programs.programs.len(), 1, "only program 50 has a PMT"); let consumer = origin.consume().request_broadcast("event/50.hang", None).await.unwrap(); let catalog = hang::catalog::Catalog::<()>::subscribe(&consumer) @@ -343,8 +477,8 @@ mod test { assert_eq!(catalog.audio.renditions.len(), 1, "program 50 publishes its stream"); } - /// A corrupt PAT starts nothing and is counted; the good repetition after it starts the - /// program without counting again. + /// A corrupt PAT is counted and ignored; the good repetition after it is read without + /// counting again. #[test] fn a_corrupt_pat_starts_no_program() { let origin = crate::source::produce_origin(); @@ -356,13 +490,13 @@ mod test { let mut data = psi_packets(0, &mut cc, &[], &corrupt(pat_section(&[(1, 0x0100)]))); data.extend_from_slice(&null); programs.decode(&data).unwrap(); - assert!(programs.programs.is_empty()); + assert!(programs.pat.is_none()); assert_eq!(programs.stats().crc_error, 1); let mut data = psi_packets(0, &mut cc, &[], &pat_section(&[(1, 0x0100)])); data.extend_from_slice(&null); programs.decode(&data).unwrap(); - assert_eq!(programs.programs.len(), 1); + assert!(programs.pat.is_some()); assert_eq!(programs.stats().crc_error, 1); } @@ -392,6 +526,203 @@ mod test { assert_eq!(renditions(&origin, "event/2.hang").await, 1); } + /// A PAT naming a thousand programs starts only the two whose PMTs the input carries, even + /// where an undescribed program shares a described one's PMT PID. + #[tokio::test] + async fn only_programs_a_pmt_describes_publish() { + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin.clone(), "event.hang", catalog::Config::default()); + // Pairs of programs share a PMT PID, so 6 waits on the PID that starts 7. + let pmt_pid = |program: u16| 0x0100 + program / 2; + let mut cc = 0; + let mut data = Vec::new(); + for section in 0..4u8 { + let entries: Vec = (1..=250) + .map(|n| u16::from(section) * 250 + n) + .flat_map(|n| { + let pid = pmt_pid(n); + [(n >> 8) as u8, n as u8, 0xe0 | (pid >> 8) as u8, pid as u8] + }) + .collect(); + data.extend(psi_packets( + 0, + &mut cc, + &[], + &psi::section(0x00, 1, 0, section, 3, &entries), + )); + } + for (program, pid, fills) in [(7u16, 0x0061, [0xAA, 0xBB]), (900, 0x0071, [0xCC, 0xDD])] { + let pmt = pmt_section(program, 0, &[(MP2, pid, &[])]); + data.extend(psi_packets(pmt_pid(program), &mut 0, &[], &pmt)); + data.extend(mp2_pes(pid, 0, 90_000, fills)); + } + programs.decode(&data).unwrap(); + let _finished = programs.finish().unwrap(); + + let started: Vec<_> = programs.programs.iter().map(|program| program.number).collect(); + assert_eq!(started, [7, 900]); + assert_eq!(renditions(&origin, "event/7.hang").await, 1); + assert_eq!(renditions(&origin, "event/900.hang").await, 1); + } + + /// A later PAT moves a waiting program to its new PMT PID and drops one it no longer lists, + /// so neither the stale PMT nor the dropped program's PMT starts anything. + #[tokio::test] + async fn a_later_pat_revises_the_waiting_programs() { + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin.clone(), "event.hang", catalog::Config::default()); + let mut cc = 0; + let mut data = psi_packets(0, &mut cc, &[], &pat_section(&[(1, 0x0100), (2, 0x0101)])); + let revised = psi::section(0x00, 1, 1, 0, 0, &[0x00, 0x01, 0xe2, 0x00]); + data.extend(psi_packets(0, &mut cc, &[], &revised)); + for (pid, program, version, es) in [(0x0100, 1, 0, 0x0062), (0x0101, 2, 0, 0x0071), (0x0200, 1, 1, 0x0061)] { + data.extend(psi_packets( + pid, + &mut 0, + &[], + &pmt_section(program, version, &[(MP2, es, &[])]), + )); + } + data.extend(mp2_pes(0x0061, 0, 90_000, [0xAA, 0xBB])); + programs.decode(&data).unwrap(); + let _finished = programs.finish().unwrap(); + + let started: Vec<_> = programs.programs.iter().map(|program| program.number).collect(); + assert_eq!(started, [1]); + assert_eq!(renditions(&origin, "event/1.hang").await, 1); + } + + /// Of two revisions of a PMT in one packet, the program starts on the later. + #[tokio::test] + async fn a_program_starts_on_the_last_pmt_in_its_packet() { + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin.clone(), "event.hang", catalog::Config::default()); + let mut data = psi_packets(0, &mut 0, &[], &pat_section(&[(1, 0x0100)])); + let mut sections = pmt_section(1, 0, &[(MP2, 0x0062, &[])]); + sections.extend(pmt_section(1, 1, &[(MP2, 0x0061, &[])])); + data.extend(psi_packets(0x0100, &mut 0, &[], §ions)); + data.extend(mp2_pes(0x0061, 0, 90_000, [0xAA, 0xBB])); + programs.decode(&data).unwrap(); + let _finished = programs.finish().unwrap(); + assert_eq!(renditions(&origin, "event/1.hang").await, 1); + } + + /// A PMT spanning two packets starts its program, which carries on from the first packet + /// rather than reading the second alone. + #[tokio::test] + async fn a_pmt_spanning_two_packets_starts_its_program() { + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin.clone(), "event.hang", catalog::Config::default()); + let mut data = psi_packets(0, &mut 0, &[], &pat_section(&[(1, 0x0100)])); + let mut private = vec![0x80, 200]; + private.resize(202, 0); + let packets = psi_packets(0x0100, &mut 0, &[], &pmt_section(1, 0, &[(MP2, 0x0061, &private)])); + assert_eq!(packets.len(), 2 * 188, "the PMT spans two packets"); + data.extend(packets); + data.extend(mp2_pes(0x0061, 0, 90_000, [0xAA, 0xBB])); + programs.decode(&data).unwrap(); + let _finished = programs.finish().unwrap(); + assert_eq!(renditions(&origin, "event/1.hang").await, 1); + } + + /// A corrupt PMT counts once whether it arrives before any program reads its PID, in the + /// packet that starts one, while one program reads it beside a waiting one, or while both + /// do, including after a PAT moves a started program onto a waiting one's PID. + #[test] + fn a_corrupt_pmt_counts_once_whoever_reads_it() { + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin, "event.hang", catalog::Config::default()); + let mut null = vec![0x47, 0x1f, 0xff, 0x10]; + null.resize(188, 0xff); + let pmt = |program| pmt_section(program, 0, &[(MP2, 0x0060 + program, &[])]); + let mut data = psi_packets(0, &mut 0, &[], &pat_section(&[(1, 0x0100), (2, 0x0100)])); + let mut cc = 0; + let mut starting = corrupt(pmt(1)); + starting.extend(pmt(1)); + for section in [corrupt(pmt(1)), starting, corrupt(pmt(2)), pmt(2), corrupt(pmt(1))] { + data.extend(psi_packets(0x0100, &mut cc, &[], §ion)); + } + data.extend_from_slice(&null); + programs.decode(&data).unwrap(); + assert_eq!(programs.programs.len(), 2); + assert_eq!(programs.stats().crc_error, 4); + + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin, "event.hang", catalog::Config::default()); + let mut cc = 0; + let mut data = psi_packets(0, &mut cc, &[], &pat_section(&[(1, 0x0100), (2, 0x0200)])); + data.extend(psi_packets(0x0100, &mut 0, &[], &pmt(1))); + let moved = [0x00, 0x01, 0xe2, 0x00, 0x00, 0x02, 0xe2, 0x00]; + data.extend(psi_packets(0, &mut cc, &[], &psi::section(0x00, 1, 1, 0, 0, &moved))); + data.extend(psi_packets(0x0200, &mut 0, &[], &corrupt(pmt(2)))); + programs.decode(&data).unwrap(); + assert_eq!(programs.programs.len(), 1); + assert_eq!(programs.stats().crc_error, 1); + + // The same move landing while the corrupt section is partway in: the program it moved + // never saw that section begin, so the scan counts it. + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin, "event.hang", catalog::Config::default()); + let mut cc = 0; + let mut data = psi_packets(0, &mut cc, &[], &pat_section(&[(1, 0x0100), (2, 0x0200)])); + data.extend(psi_packets(0x0100, &mut 0, &[], &pmt(1))); + let mut private = vec![0x80, 200]; + private.resize(202, 0); + let split = psi_packets( + 0x0200, + &mut 0, + &[], + &corrupt(pmt_section(2, 0, &[(MP2, 0x0071, &private)])), + ); + assert_eq!(split.len(), 2 * 188, "the corrupt PMT spans two packets"); + data.extend_from_slice(&split[..188]); + data.extend(psi_packets(0, &mut cc, &[], &psi::section(0x00, 1, 1, 0, 0, &moved))); + data.extend_from_slice(&split[188..]); + programs.decode(&data).unwrap(); + assert_eq!(programs.stats().crc_error, 1); + + // After that move, a section start abandons the partial for another split corrupt + // section, which the moved program reads from its start, so it alone counts it. + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin, "event.hang", catalog::Config::default()); + let mut cc = 0; + let mut data = psi_packets(0, &mut cc, &[], &pat_section(&[(1, 0x0100), (2, 0x0200)])); + data.extend(psi_packets(0x0100, &mut 0, &[], &pmt(1))); + data.extend_from_slice(&split[..188]); + data.extend(psi_packets(0, &mut cc, &[], &psi::section(0x00, 1, 1, 0, 0, &moved))); + data.extend(psi_packets( + 0x0200, + &mut 1, + &[], + &corrupt(pmt_section(2, 0, &[(MP2, 0x0071, &private)])), + )); + programs.decode(&data).unwrap(); + assert_eq!(programs.stats().crc_error, 1); + + // A corrupt section begun in the packet that starts a program sharing the PID: the new + // program reads that packet, so it alone counts the section. + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin, "event.hang", catalog::Config::default()); + let mut data = psi_packets(0, &mut 0, &[], &pat_section(&[(1, 0x0100), (2, 0x0100)])); + let mut sections = pmt(1); + sections.extend(corrupt(pmt_section(2, 0, &[(MP2, 0x0071, &private)]))); + data.extend(psi_packets(0x0100, &mut 0, &[], §ions)); + programs.decode(&data).unwrap(); + assert_eq!(programs.programs.len(), 1); + assert_eq!(programs.stats().crc_error, 1); + + // A waiting program's PMT PID that a started program reads only as media: no program + // reassembles PMTs there, so the scan counts the corrupt section. + let origin = crate::source::produce_origin(); + let mut programs = Programs::new(origin, "event.hang", catalog::Config::default()); + let mut data = psi_packets(0, &mut 0, &[], &pat_section(&[(1, 0x0100), (2, 0x0061)])); + data.extend(psi_packets(0x0100, &mut 0, &[], &pmt(1))); + data.extend(psi_packets(0x0061, &mut 0, &[], &corrupt(pmt(2)))); + programs.decode(&data).unwrap(); + assert_eq!(programs.programs.len(), 1); + assert_eq!(programs.stats().crc_error, 1); + } + /// Each program announces as its own publisher instance for the run, so a restart /// replaces it rather than resuming into it. #[test] @@ -420,7 +751,7 @@ mod test { let mut packets = input.chunks(188); while programs.programs.is_empty() { programs - .decode(packets.next().expect("the PAT starts the programs")) + .decode(packets.next().expect("a PMT starts a program")) .unwrap(); } assert!( @@ -428,7 +759,7 @@ mod test { .programs .iter() .all(|program| program.broadcast.route().is_none()), - "announced on the PAT alone" + "announced on the PSI alone" ); for packet in packets { programs.decode(packet).unwrap(); @@ -505,7 +836,7 @@ mod test { null.resize(188, 0xff); data.extend_from_slice(&null); programs.decode(&data).unwrap(); - assert_eq!(programs.programs.len(), 1); + assert!(programs.pat.is_some()); assert_eq!(programs.stats().crc_error, 1); } diff --git a/rs/moq-mux/src/container/ts/psi.rs b/rs/moq-mux/src/container/ts/psi.rs index d68a912eafc..5feb2baaa7b 100644 --- a/rs/moq-mux/src/container/ts/psi.rs +++ b/rs/moq-mux/src/container/ts/psi.rs @@ -137,17 +137,6 @@ pub(super) struct Pat { pub programs: Vec, } -impl Pat { - /// The program numbers listed, without the network PID's. - pub fn program_numbers(&self) -> Vec { - self.programs - .iter() - .map(|entry| entry.program_number) - .filter(|&number| number != 0) - .collect() - } -} - /// Collects PAT sections until every section of one version has arrived. /// /// A PAT that needs more than one section is complete only once all of them are in, so a @@ -553,7 +542,8 @@ mod test { assert_eq!(pat.section(&first), None); let whole = pat.section(&second).expect("both sections are in"); assert_eq!(whole.transport_stream_id, 7); - assert_eq!(whole.program_numbers(), [1, 2]); + let numbers: Vec<_> = whole.programs.iter().map(|entry| entry.program_number).collect(); + assert_eq!(numbers, [1, 2]); // A repetition of either section repeats the whole table. assert_eq!(pat.section(&first), Some(whole));