diff --git a/quest/m1/README.md b/quest/m1/README.md index 3136b7c6bf..aaf667fc28 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -162,7 +162,6 @@ session outcomes, and embedder refusals inside session outcomes. - [Fail loud on dropped timed metadata](/quest/m1/drop-loud.md) - FLV script tags and fMP4 emsg boxes are counted and warned about instead of silently dropped, until real carriage lands - [Encoder colour](/quest/m1/color-model.md) - every moq-video encode path signals the colour its output actually has, or refuses instead of mislabelling - [T-STD TS export](/quest/m1/tstd/README.md) - `moq export ts` is a proper remux that passes the T-STD buffer model, starting with a fixed `--delay` -- [Multi-packet PMT](/quest/m1/ts-pmt-split.md) - `export ts` splits a PMT longer than one packet instead of exiting before its first packet - [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 - [Rust non-continuous signal](/quest/m1/rust-continuous.md) - the Rust container consumer reports a frame after a subscribe or discontinuity as non-continuous, like JS, for the tune-in and warmup trims - [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 diff --git a/quest/m1/ts-pmt-split.md b/quest/m1/ts-pmt-split.md deleted file mode 100644 index ddabe9001a..0000000000 --- a/quest/m1/ts-pmt-split.md +++ /dev/null @@ -1,32 +0,0 @@ -# [S] TS export writes a PMT that spans several packets - -## Goal - -`moq export ts` carries a programme whose PMT section is longer than one TS -packet (multi-language audio easily is), splitting it over consecutive -packets on the PMT PID instead of exiting before its first packet. A section -past the 1,024-byte limit (ISO 13818-1 `section_length` ≤ 1,021) fails the -export loudly. - -## Plan - -Decided (2026-10-10), importing #5110. The export hands mpeg2ts a -`TsPayload::Pmt`, which serialises the table into a single packet's payload -and fails with "failed to write whole buffer" once it overflows. mpeg2ts keeps -its section writer `pub(super)`, so it cannot be reused. - -- Build the PAT and PMT sections ourselves in - `rs/moq-mux/src/container/ts/psi.rs`, which already has the CRC and the - parser, and packetise both through the export's existing `write_section` - (pointer_field and unit start on the first packet, continuity counters - running). That leaves a single PSI write path, so mpeg2ts no longer writes - tables. Chosen over keeping the PAT on mpeg2ts (two paths) and over an - upstream mpeg2ts change (an outside party). -- Test: a synthetic export with enough audio tracks carrying language - descriptors to push the PMT past one packet. Assert that the output parses - back to the same PMT, that the continuity counters run, and that a PMT over - the limit fails loud. No external fixture. - -## Closes - -- [#5110](https://github.com/moq-dev/moq/issues/5110) - `export ts` exits before its first packet when the PMT is longer than one TS packet diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index d2faadae99..f1ff99b18b 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -515,8 +515,8 @@ impl Kind { /// The program tables plus the resolved PID layout. struct Psi { - pat: Pat, - pmt: Pmt, + pat: Bytes, + pmt: Bytes, pcr_pid: u16, /// PID the PMT rides on: the source's original (preserved from the service /// record) or the synthesized [`PMT_PID`] for a media-only source. @@ -1764,8 +1764,8 @@ impl Export { self.liveness.register(track.pid); } self.psi = Some(Psi { - pat, - pmt, + pat: super::psi::encode_pat(&pat)?, + pmt: super::psi::encode_pmt(&pmt)?, pcr_pid, pmt_pid, }); @@ -1839,12 +1839,10 @@ impl Export { if psi { let psi = self.psi.as_ref().context("PSI not built")?; let pmt_pid = psi.pmt_pid; - let pat = TsPayload::Pat(psi.pat.clone()); - let pmt = TsPayload::Pmt(psi.pmt.clone()); - let flag = self.breaks(Pid::PAT).then(|| flags(true)); - self.write_packet(&mut out, Pid::PAT, flag, pat)?; - let flag = self.breaks(pmt_pid).then(|| flags(true)); - self.write_packet(&mut out, pmt_pid, flag, pmt)?; + let pat = psi.pat.clone(); + let pmt = psi.pmt.clone(); + self.write_section(&mut out, Pid::PAT, &pat)?; + self.write_section(&mut out, pmt_pid, &pmt)?; self.last_psi = Some(frame.timestamp); } @@ -2143,7 +2141,7 @@ impl Export { Ok(()) } - /// Packetize a private section (SCTE-35 or other) verbatim. The first packet + /// Packetize a complete PSI or private section verbatim. The first packet /// carries the pointer_field plus the section start as a `Section` payload (sets /// the unit-start bit so the receiver finds the pointer_field); continuations are /// `Raw`. The section bytes are opaque, so this round-trips byte-for-byte. diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index cb6f3a916d..135bd1d646 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -7546,6 +7546,123 @@ async fn export_stats_count_a_gap_longer_than_the_backfill() { assert!(quiet < Duration::from_millis(200), "{quiet:?}"); } +#[tokio::test(start_paused = true)] +async fn a_large_pmt_is_split_across_packets() { + assert_eq!(large_pmt(24, None).await, 2); +} + +/// A PMT spanning three packets at 20 Mb/s goes out at TBsys's rate: the PAT and PMT share its +/// 512 bytes, which drain at Rxsys (ISO 13818-1 2.4.2.3). +#[tokio::test(start_paused = true)] +async fn a_large_pmt_keeps_within_the_system_buffer() { + assert_eq!(large_pmt(48, Some(20_000_000)).await, 3); +} + +/// Export `streams` language-tagged AAC streams, check the PMT reassembles across its packets +/// and keeps within TBsys at `mux_rate`, and return how many packets it spans. +async fn large_pmt(streams: u16, mux_rate: Option) -> usize { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let mut catalog = crate::catalog::Producer::new( + &mut broadcast, + crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::::default()), + ) + .unwrap(); + let mut config = tscat::Mpegts::default(); + let mut producers = Vec::new(); + for index in 0..streams { + let name = format!("audio{index:02}"); + let track = broadcast + .create_track( + name.as_str(), + hang::container::track_info(hang::catalog::PRIORITY.audio), + ) + .unwrap(); + catalog + .modify() + .unwrap() + .audio + .renditions + .insert(name.clone(), AudioConfig::new(AAC { profile: 2 }, 48_000, 2)); + let mut entry = tscat::Track::new(0x1001 + index); + entry.descriptors = vec![tscat::Descriptor { + tag: 0x0a, + data: Bytes::from_static(b"eng\0"), + }]; + config.tracks.insert(name, entry); + let mut producer = Producer::new(track, HangContainer::Legacy(Kind::Data)); + producer + .write(Frame { + timestamp: Timestamp::ZERO, + duration: None, + keyframe: true, + payload: Bytes::from_static(b"aac"), + }) + .unwrap(); + producer.finish().unwrap(); + producers.push(producer); + } + catalog.modify().unwrap().ext.mpegts = config; + let mut exporter = Export::with_ts(crate::source::announced(&broadcast.consume()).await, Default::default()) + .await + .unwrap(); + if let Some(mux_rate) = mux_rate { + exporter = exporter.with_mux_rate(mux_rate); + } + let ts = drain_with(exporter).await; + let packets = ts.as_chunks::<188>().0; + let pid = |packet: &[u8; 188]| (u16::from(packet[1] & 0x1f) << 8) | u16::from(packet[2]); + if let Some(mux_rate) = mux_rate { + // TBsys: 188 B in per table packet, drained at Rxsys over each packet's time. + let drain = 1_000_000.0 / mux_rate as f64 * 188.0; + let (mut tb, mut peak) = (0.0f64, 0.0f64); + for packet in packets { + tb = (tb - drain).max(0.0); + if [0, 0x1000].contains(&pid(packet)) { + tb += 188.0; + peak = peak.max(tb); + } + } + assert!(peak <= 512.0, "TBsys peak {peak:.0} B"); + } + let mut section = Vec::new(); + let mut counters = Vec::new(); + for packet in packets { + if pid(packet) != 0x1000 { + continue; + } + counters.push(packet[3] & 0x0f); + let offset = if packet[3] & 0x20 != 0 { + 5 + usize::from(packet[4]) + } else { + 4 + }; + if packet[1] & 0x40 != 0 { + assert!(section.is_empty()); + assert_eq!(packet[offset], 0); + section.extend_from_slice(&packet[offset + 1..]); + } else { + section.extend_from_slice(&packet[offset..]); + } + let length = 3 + (((usize::from(section[1]) & 0x0f) << 8) | usize::from(section[2])); + if section.len() >= length { + section.truncate(length); + break; + } + } + for pair in counters.windows(2) { + assert_eq!(pair[1], (pair[0] + 1) & 0x0f); + } + assert!(super::psi::crc_ok(§ion)); + let pmt = super::psi::Pmt::parse(§ion).unwrap(); + assert_eq!(pmt.streams.len(), usize::from(streams)); + for (index, stream) in pmt.streams.iter().enumerate() { + assert_eq!(stream.pid.as_u16(), 0x1001 + index as u16); + assert_eq!(stream.stream_type, 0x0f); + assert_eq!(stream.descriptors[0].data.as_ref(), b"eng\0"); + } + counters.len() +} + /// Publish a catalog-only broadcast at `live`, for a test of what follows its end. fn publish_bare( origin: &moq_net::origin::Producer, diff --git a/rs/moq-mux/src/container/ts/psi.rs b/rs/moq-mux/src/container/ts/psi.rs index 21883e7602..f3e722fc0e 100644 --- a/rs/moq-mux/src/container/ts/psi.rs +++ b/rs/moq-mux/src/container/ts/psi.rs @@ -24,6 +24,58 @@ pub(super) fn crc_ok(section: &[u8]) -> bool { CRC.checksum(body) == u32::from_be_bytes(crc.try_into().expect("four bytes")) } +/// Encode the exporter's single-program PAT independently of TS packet boundaries. +pub(super) fn encode_pat(pat: &mpeg2ts::ts::payload::Pat) -> anyhow::Result { + let mut section = vec![PAT_TABLE_ID, 0xb0, 0]; + section.extend_from_slice(&pat.transport_stream_id.to_be_bytes()); + section.extend_from_slice(&[0xc1 | (pat.version_number.as_u8() << 1), 0, 0]); + for entry in &pat.table { + section.extend_from_slice(&entry.program_num.to_be_bytes()); + section.extend_from_slice(&(0xe000 | entry.program_map_pid.as_u16()).to_be_bytes()); + } + finish_section(section) +} + +/// Encode a PMT, rejecting descriptor or table lengths that MPEG-TS cannot represent. +pub(super) fn encode_pmt(pmt: &mpeg2ts::ts::payload::Pmt) -> anyhow::Result { + let mut section = vec![PMT_TABLE_ID, 0xb0, 0]; + section.extend_from_slice(&pmt.program_num.to_be_bytes()); + section.extend_from_slice(&[0xc1 | (pmt.version_number.as_u8() << 1), 0, 0]); + section.extend_from_slice(&(0xe000 | pmt.pcr_pid.map_or(0x1fff, |pid| pid.as_u16())).to_be_bytes()); + encode_descriptors(&mut section, &pmt.program_info)?; + for stream in &pmt.es_info { + section.push(stream.stream_type as u8); + section.extend_from_slice(&(0xe000 | stream.elementary_pid.as_u16()).to_be_bytes()); + encode_descriptors(&mut section, &stream.descriptors)?; + } + finish_section(section) +} + +fn encode_descriptors(out: &mut Vec, descriptors: &[mpeg2ts::ts::Descriptor]) -> anyhow::Result<()> { + let length = descriptors + .iter() + .map(|descriptor| 2 + descriptor.data.len()) + .sum::(); + anyhow::ensure!(length <= 0x0fff, "PSI descriptor loop exceeds 4095 bytes"); + out.extend_from_slice(&(0xf000 | length as u16).to_be_bytes()); + for descriptor in descriptors { + let length = u8::try_from(descriptor.data.len()).context("PSI descriptor exceeds 255 bytes")?; + out.extend_from_slice(&[descriptor.tag, length]); + out.extend_from_slice(&descriptor.data); + } + Ok(()) +} + +fn finish_section(mut section: Vec) -> anyhow::Result { + let length = section.len() - 3 + 4; + anyhow::ensure!(length <= 1021, "PAT/PMT section exceeds 1021 bytes"); + section[1] |= (length >> 8) as u8; + section[2] = length as u8; + let crc = CRC.checksum(§ion); + section.extend_from_slice(&crc.to_be_bytes()); + Ok(section.into()) +} + /// The table_id of a PAT section. pub(super) const PAT_TABLE_ID: u8 = 0x00; /// The table_id of a PMT section. @@ -315,6 +367,38 @@ pub(super) fn section(table_id: u8, extension: u16, version: u8, number: u8, las mod test { use super::*; + #[test] + fn pmt_encoding_rejects_unrepresentable_lengths() { + use mpeg2ts::ts::payload::Pmt; + use mpeg2ts::ts::{Descriptor, VersionNumber}; + let mut pmt = Pmt { + program_num: 1, + pcr_pid: None, + version_number: VersionNumber::default(), + program_info: vec![Descriptor { + tag: 0x80, + data: vec![0; 256], + }], + es_info: Vec::new(), + }; + assert!(encode_pmt(&pmt).unwrap_err().to_string().contains("255 bytes")); + pmt.program_info = vec![ + Descriptor { + tag: 0x80, + data: vec![0; 250] + }; + 4 + ]; + let section = encode_pmt(&pmt).unwrap(); + assert!(crc_ok(§ion)); + assert_eq!(section.len(), 1024); + pmt.program_info.push(Descriptor { + tag: 0x81, + data: Vec::new(), + }); + assert!(encode_pmt(&pmt).unwrap_err().to_string().contains("1021 bytes")); + } + #[test] fn crc_matches_mpeg2() { // The single-program PAT ffmpeg writes: TSID 1, program 1 on PID 0x1000. diff --git a/rs/moq-mux/src/container/ts/schedule.rs b/rs/moq-mux/src/container/ts/schedule.rs index 81096c74ea..6611cb2291 100644 --- a/rs/moq-mux/src/container/ts/schedule.rs +++ b/rs/moq-mux/src/container/ts/schedule.rs @@ -87,6 +87,50 @@ impl Buffer { } } +/// TBsys in a receiver over one slot: the single 512-byte transport buffer that PIDs 0-3 and +/// the program's PMT share, draining at Rxsys (ISO 13818-1 2.4.2.3). +/// +/// Its fill is in bytes times the slot's positions, so each position drains a whole slot's +/// worth of Rxsys bytes and nothing rounds. It starts each slot empty: a slot's 25 ms drains +/// six times its size. +struct SystemBuffer { + pmt_pid: u16, + positions: u128, + fill: u128, +} + +impl SystemBuffer { + /// The buffer's size in bytes. + const SIZE: u128 = 512; + /// The bytes it drains a slot. + const DRAIN: u128 = Buffer::SYSTEM.rate as u128 * PCR_INTERVAL.as_nanos() / 8_000_000_000; + + fn new(pmt_pid: u16, positions: usize) -> Self { + Self { + pmt_pid, + positions: positions as u128, + fill: 0, + } + } + + fn carries(&self, pid: u16) -> bool { + pid <= 3 || pid == self.pmt_pid + } + + /// Whether a packet it carries fits at the next position. + fn fits(&self) -> bool { + self.fill.saturating_sub(Self::DRAIN) + TsPacket::SIZE as u128 * self.positions <= Self::SIZE * self.positions + } + + /// Take the next position, laid with `packet`. + fn lay(&mut self, packet: &[u8]) { + self.fill = self.fill.saturating_sub(Self::DRAIN); + if self.carries(pid(packet)) { + self.fill += TsPacket::SIZE as u128 * self.positions; + } + } +} + /// 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. @@ -142,6 +186,8 @@ pub(super) struct Slot { pub keyframe: bool, /// The PID of each access unit whose first packet rides in the slot. pub units: Vec, + /// The most packets a slot admission lets each bounded PID carry. + pub caps: HashMap, } impl Slot { @@ -156,15 +202,20 @@ impl Slot { /// tables (PAT and the PMT on `pmt_pid`) go just ahead of the first packet muxed after /// them, so they still lead the keyframe they were written for and a reader knows each PID /// before its first packet. + /// + /// The tables share one [`SystemBuffer`]. Where a table ahead of the slot's first media + /// packet would overflow it, a null takes the position instead, so a PMT spanning several + /// packets goes out at the buffer's rate, as long as the slot has nulls to spare and the + /// 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; - let mut tables = Vec::new(); + 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())]; for (i, packet) in packets.iter().enumerate() { let pid = pid(packet); if pid == 0 || pid == pmt_pid { - tables.push(i); + tables.push_back(i); } else if let Some((_, lane)) = lanes.iter_mut().find(|(p, _)| *p == pid) { lane.push_back(i); } else { @@ -175,43 +226,102 @@ impl Slot { // Every position, each lane (the nulls last) gains its weight and the one furthest // ahead takes it, paying back the total: over the slot each takes exactly its weight. // The clock takes the first. + let mut nulls = self.nulls; let mut weights: Vec = lanes .iter() .map(|(_, lane)| lane.len() as i64) - .chain([self.nulls as i64]) + .chain([nulls as i64]) .collect(); weights[0] += 1; - let total: i64 = weights.iter().sum(); + let mut total: i64 = weights.iter().sum(); let mut credit = weights.clone(); credit[0] -= total; - let mut tables = tables.into_iter().peekable(); - let mut out = Vec::with_capacity((1 + packets.len() + self.nulls) * TsPacket::SIZE); + // Positions the round-robin has still to lay. + let mut left = total - 1; + // No media packet has gone out yet, so laying the slot afresh spaces no PID worse. + let mut fresh = true; + + let positions = 1 + packets.len() + self.nulls; + let mut system = SystemBuffer::new(pmt_pid, positions); + let mut out = Vec::with_capacity(positions * TsPacket::SIZE); out.extend_from_slice(clock); - for _ in 1..total { + system.lay(clock); + loop { + let pick = (left > 0).then(|| { + let mut pick = 0; + for lane in 1..credit.len() { + if credit[lane] + weights[lane] > credit[pick] + weights[pick] { + pick = lane; + } + } + pick + }); + // The packet the round-robin lays next, `None` for a null. + let next = pick.map(|pick| lanes.get(pick).map(|(_, lane)| lane[0])); + // The next table, once the next packet was muxed after it. + let table = tables + .front() + .copied() + .filter(|&table| next.is_none_or(|next| next.is_some_and(|i| table < i))); + if let Some(table) = table { + // Ahead of the slot's first media packet, a null spaces a table that would + // overflow TBsys, and the rest of the slot is laid out afresh around it, so no + // PID's packets bunch up to make up the time. The stall must leave every PID + // spread no denser over the rest of the slot than admission holds it to over a + // whole one. Otherwise the table goes out regardless: past the first media + // packet, laying afresh would drop the spacing already given the PIDs served. + let room = || { + // The round-robin's positions left after the stall; the tables still to go + // lead the PIDs they delay, so lend them no time. + let after = left as usize - 1; + lanes.iter().all(|(pid, lane)| { + self.caps + .get(pid) + .is_none_or(|&cap| lane.len() * positions <= cap * after) + }) + }; + if fresh && !system.fits() && nulls > 0 && room() { + nulls -= 1; + system.lay(null); + out.extend_from_slice(null); + weights = lanes + .iter() + .map(|(_, lane)| lane.len() as i64) + .chain([nulls as i64]) + .collect(); + total = weights.iter().sum(); + credit = vec![0; weights.len()]; + left = total; + continue; + } + tables.pop_front(); + system.lay(&packets[table]); + out.extend_from_slice(&packets[table]); + continue; + } + let Some(pick) = pick else { + break; + }; for (credit, weight) in credit.iter_mut().zip(&weights) { *credit += weight; } - let mut pick = 0; - for lane in 1..credit.len() { - if credit[lane] > credit[pick] { - pick = lane; - } - } credit[pick] -= total; - let Some((_, lane)) = lanes.get_mut(pick) else { - out.extend_from_slice(null); - continue; + left -= 1; + let packet = match lanes.get_mut(pick) { + Some((_, lane)) => { + fresh = false; + let i = lane + .pop_front() + .expect("a lane takes no more positions than its weight"); + &packets[i] + } + None => { + nulls -= 1; + null + } }; - let i = lane - .pop_front() - .expect("a lane takes no more positions than its weight"); - while let Some(table) = tables.next_if(|&table| table < i) { - out.extend_from_slice(&packets[table]); - } - out.extend_from_slice(&packets[i]); - } - for table in tables { - out.extend_from_slice(&packets[table]); + system.lay(packet); + out.extend_from_slice(packet); } out } @@ -447,6 +557,11 @@ impl Schedule { nulls, keyframe, units, + caps: self + .buffers + .iter() + .map(|(&pid, buffer)| (pid, buffer.per_slot())) + .collect(), })) } @@ -880,6 +995,7 @@ mod tests { nulls: 416 - 1 - media, keyframe: false, units: vec![], + caps: HashMap::new(), }; let mut null = unit(0, 1); (null[1], null[2]) = (0x1f, 0xff); @@ -916,10 +1032,133 @@ mod tests { nulls: 2, keyframe: true, units: vec![1, 3], + caps: HashMap::new(), }; let mut null = unit(0, 1); (null[1], null[2]) = (0x1f, 0xff); let laid: Vec = slot.layout(&unit(1, 1), 100, &null).chunks(188).map(pid).collect(); assert_eq!(laid, [1, 0, 100, 3, 0x1fff, 3, 3, 1, 0x1fff, 3]); } + + /// A PAT and a multi-packet PMT go out at TBsys's rate rather than in a burst, spaced by + /// nulls and still in order ahead of the keyframe muxed after them. The video they delay + /// always keeps within its own transport buffer: where spacing the tables would leave the + /// video too little time (a five-packet PMT, or video near its admission cap), the rest of + /// the tables go out regardless. + #[test] + fn the_tables_go_out_at_the_system_buffer_rate() { + // The peak fill of a 512-byte transport buffer fed by `pids` and drained at `rx` b/s, over + // a 25 ms slot. + let peak = |laid: &[&[u8]], pids: &[u16], rx: f64| { + let drain = rx / 8.0 * 0.025 / laid.len() as f64; + let (mut tb, mut peak) = (0.0f64, 0.0f64); + for packet in laid { + tb = (tb - drain).max(0.0); + if pids.contains(&pid(packet)) { + tb += 188.0; + peak = peak.max(tb); + } + } + peak + }; + // The video's Rx is 1.2 x 17,999,872 b/s, as in the full-slot test above. + const RX: u64 = 21_599_846; + // A 20 Mb/s slot with a small keyframe, and 25 Mb/s slots whose video takes most of it + // alongside AC-3, MPEG audio and teletext: (positions, video PID and packets, other + // PIDs, PMT packets, whether TBsys holds). + let others: &[(u8, usize)] = &[(122, 9), (121, 5), (123, 5), (124, 5), (131, 1)]; + for (positions, video, count, others, parts, paced) in [ + (332, 1, 50, &[][..], 3, true), + (416, 111, 300, others, 3, true), + (416, 111, 300, others, 5, false), + (416, 111, 352, others, 3, false), + ] { + let mut packets = unit(0, 1); + for part in 0..parts { + let mut pmt = unit(100, 1); + pmt[4] = part; + packets.extend(pmt); + } + packets.extend(unit(video, count)); + for &(pid, count) in others { + packets.extend(unit(pid, count)); + } + let slot = Slot { + index: 0, + pcr: 0, + nulls: positions - 1 - packets.len() / TsPacket::SIZE, + packets, + keyframe: true, + units: vec![video.into()], + caps: HashMap::from([(video.into(), Buffer { rate: RX, size: None }.per_slot())]), + }; + let mut null = unit(0, 1); + (null[1], null[2]) = (0x1f, 0xff); + let laid = slot.layout(&unit(video, 1), 100, &null); + assert_eq!(laid.len(), positions * TsPacket::SIZE, "the slot keeps its size"); + let laid: Vec<&[u8]> = laid.chunks(TsPacket::SIZE).collect(); + + let tables: Vec = (0..laid.len()).filter(|&k| [0, 100].contains(&pid(laid[k]))).collect(); + let order: Vec = tables.iter().skip(1).map(|&k| laid[k][4]).collect(); + assert_eq!(order, (0..parts).collect::>(), "the PMT keeps its order"); + let keyframe = (1..laid.len()).find(|&k| pid(laid[k]) == u16::from(video)).unwrap(); + assert!(tables.iter().all(|&k| k < keyframe), "the tables lead the keyframe"); + + let system = peak(&laid, &[0, 100], 1_000_000.0); + assert!( + !paced || system <= 512.0, + "TBsys peak {system:.0} B, {parts} PMT packets" + ); + let video = peak(&laid, &[video.into()], RX as f64); + assert!(video <= 512.0, "video TB peak {video:.0} B, {parts} PMT packets"); + } + } + + /// Tables ahead of a second keyframe, after the first's PID has gone out, never stall the + /// slot: laying it afresh would drop the spacing that PID already has. + #[test] + fn tables_after_media_leave_its_spacing_alone() { + let mut packets = unit(0, 1); + packets.extend(unit(100, 1)); + packets.extend(unit(1, 125)); + packets.extend(unit(0, 1)); + packets.extend(unit(100, 1)); + packets.extend(unit(2, 114)); + let (a, b) = ( + Buffer { + rate: 7_833_600, + size: None, + }, + Buffer { + rate: 8_160_000, + size: None, + }, + ); + let slot = Slot { + index: 0, + pcr: 0, + packets, + nulls: 88, + keyframe: true, + units: vec![1, 2], + caps: HashMap::from([(1, a.per_slot()), (2, b.per_slot())]), + }; + let mut null = unit(0, 1); + (null[1], null[2]) = (0x1f, 0xff); + let laid = slot.layout(&unit(1, 1), 100, &null); + let laid: Vec<&[u8]> = laid.chunks(TsPacket::SIZE).collect(); + assert_eq!(laid.len(), 332); + for (media, buffer) in [(1, a), (2, b)] { + let drain = buffer.rate as f64 / 8.0 * 0.025 / laid.len() as f64; + let (mut tb, mut peak) = (0.0f64, 0.0f64); + for packet in &laid { + tb = (tb - drain).max(0.0); + if pid(packet) == media { + tb += 188.0; + peak = peak.max(tb); + } + } + assert!(peak <= 512.0, "PID {media} TB peak {peak:.0} B"); + } + } }