From 309988a46f35ba8ca50e84c772a54561bba2ca2e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 09:32:56 -0700 Subject: [PATCH 1/8] chore: claim TS PMT split quest Co-Authored-By: GPT-6 From 94968c933f1a682f7100aca7591841204d50d57e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 09:59:59 -0700 Subject: [PATCH 2/8] fix(mux): split PAT and PMT sections across TS packets Co-Authored-By: GPT-6 --- quest/m1/README.md | 1 - quest/m1/ts-pmt-split.md | 32 -------- rs/moq-mux/src/container/ts/export.rs | 18 ++--- rs/moq-mux/src/container/ts/export_test.rs | 87 ++++++++++++++++++++++ rs/moq-mux/src/container/ts/psi.rs | 84 +++++++++++++++++++++ 5 files changed, 180 insertions(+), 42 deletions(-) delete mode 100644 quest/m1/ts-pmt-split.md diff --git a/quest/m1/README.md b/quest/m1/README.md index 25c59174d5..c4614643db 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -167,7 +167,6 @@ ladders. - [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 - [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 - [Watch decode errors](/quest/m1/watch-decode-error.md) - a WebCodecs error ends the subscription and the element reports it 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 d2e55e0501..b4c3c8a340 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -500,8 +500,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. @@ -1654,8 +1654,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, }); @@ -1729,10 +1729,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()); - self.write_packet(&mut out, Pid::PAT, None, pat)?; - self.write_packet(&mut out, pmt_pid, None, 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); } @@ -2040,7 +2040,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 136ac32139..9aa0e57f03 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -7234,3 +7234,90 @@ async fn export_stats_count_a_gap_longer_than_the_backfill() { let quiet = running.quiet.expect("the output carries a PCR"); assert!(quiet < Duration::from_millis(200), "{quiet:?}"); } + +#[tokio::test(start_paused = true)] +async fn a_large_pmt_is_split_across_packets() { + 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..24 { + 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 exporter = Export::with_ts(crate::source::announced(&broadcast.consume()), Default::default()) + .await + .unwrap(); + let ts = drain_with(exporter).await; + let mut section = Vec::new(); + let mut counters = Vec::new(); + for packet in ts.as_chunks::<188>().0 { + let pid = (u16::from(packet[1] & 0x1f) << 8) | u16::from(packet[2]); + if pid != 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; + } + } + assert!(counters.len() > 1, "PMT must span packets"); + 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(), 24); + 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"); + } +} 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. From a78ed58e62c5aaed11a7456bd41ec7eb3a32b02a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 11:48:26 -0700 Subject: [PATCH 3/8] fix(mux): pace PSI packets through the shared system buffer A PMT spanning three or more packets went out back to back with the PAT, overflowing the program's 512-byte TBsys (ISO 13818-1 2.4.2.3) at high multiplex rates. Model TBsys per slot, fed by PIDs 0-3 and the PMT PID, and space each table packet with a null (or a packet muxed ahead of it) until it fits, keeping the tables in order ahead of dependent media. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/export_test.rs | 44 +++++- rs/moq-mux/src/container/ts/schedule.rs | 162 ++++++++++++++++++--- 2 files changed, 181 insertions(+), 25 deletions(-) diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index 9e74750707..b9b6568044 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -7501,6 +7501,19 @@ async fn export_stats_count_a_gap_longer_than_the_backfill() { #[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, @@ -7509,7 +7522,7 @@ async fn a_large_pmt_is_split_across_packets() { .unwrap(); let mut config = tscat::Mpegts::default(); let mut producers = Vec::new(); - for index in 0..24 { + for index in 0..streams { let name = format!("audio{index:02}"); let track = broadcast .create_track( @@ -7542,15 +7555,32 @@ async fn a_large_pmt_is_split_across_packets() { producers.push(producer); } catalog.modify().unwrap().ext.mpegts = config; - let exporter = Export::with_ts(crate::source::announced(&broadcast.consume()), Default::default()) + let mut exporter = Export::with_ts(crate::source::announced(&broadcast.consume()), 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 ts.as_chunks::<188>().0 { - let pid = (u16::from(packet[1] & 0x1f) << 8) | u16::from(packet[2]); - if pid != 0x1000 { + for packet in packets { + if pid(packet) != 0x1000 { continue; } counters.push(packet[3] & 0x0f); @@ -7572,18 +7602,18 @@ async fn a_large_pmt_is_split_across_packets() { break; } } - assert!(counters.len() > 1, "PMT must span packets"); 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(), 24); + 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. diff --git a/rs/moq-mux/src/container/ts/schedule.rs b/rs/moq-mux/src/container/ts/schedule.rs index 81096c74ea..e0d9787ee4 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. @@ -156,15 +200,19 @@ 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 the next table would overflow it, a null + /// packet (or failing that, a packet muxed ahead of the table) takes the position instead, + /// so a PMT spanning several packets goes out at the buffer's rate. 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 { @@ -184,9 +232,8 @@ impl Slot { let 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); - out.extend_from_slice(clock); + // The media and null positions in order, `None` for a null. + let mut order: VecDeque> = VecDeque::with_capacity(total as usize); for _ in 1..total { for (credit, weight) in credit.iter_mut().zip(&weights) { *credit += weight; @@ -198,20 +245,50 @@ impl Slot { } } credit[pick] -= total; - let Some((_, lane)) = lanes.get_mut(pick) else { - out.extend_from_slice(null); - continue; - }; - 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]); + order.push_back(lanes.get_mut(pick).map(|(_, lane)| { + lane.pop_front() + .expect("a lane takes no more positions than its weight") + })); } - for table in tables { - out.extend_from_slice(&packets[table]); + + 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); + system.lay(clock); + loop { + // The next table, once the next media packet was muxed after it. + let table = tables + .front() + .copied() + .filter(|&table| order.front().is_none_or(|next| next.is_some_and(|i| table < i))); + let next = match table { + None => order.pop_front(), + Some(table) => { + let filler = || { + order + .iter() + .position(Option::is_none) + .or_else(|| order.iter().position(|next| next.is_some_and(|i| i < table))) + }; + match (!system.fits()).then(filler).flatten() { + Some(k) => order.remove(k), + // The table fits, or nothing is left to space it with. + None => { + tables.pop_front(); + system.lay(&packets[table]); + out.extend_from_slice(&packets[table]); + continue; + } + } + } + }; + let Some(next) = next else { + break; + }; + let packet = next.map_or(null, |i| &packets[i]); + system.lay(packet); + out.extend_from_slice(packet); } out } @@ -922,4 +999,53 @@ mod tests { 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 three-packet PMT go out at TBsys's rate rather than in a burst, still in + /// order ahead of the keyframe muxed after them. Nulls space them in a padded slot, and a + /// packet muxed ahead of them where the slot has no nulls. + #[test] + fn the_tables_go_out_at_the_system_buffer_rate() { + for (ahead, nulls) in [(0, 332 - 1 - 54), (40, 0)] { + let mut packets = unit(3, ahead); + packets.extend(unit(0, 1)); + for part in 0..3 { + let mut pmt = unit(100, 1); + pmt[4] = part; + packets.extend(pmt); + } + packets.extend(unit(1, 50)); + let positions = 1 + packets.len() / TsPacket::SIZE + nulls; + let slot = Slot { + index: 0, + pcr: 0, + packets, + nulls, + keyframe: true, + units: vec![1], + }; + let mut null = unit(0, 1); + (null[1], null[2]) = (0x1f, 0xff); + let laid = slot.layout(&unit(1, 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 parts: Vec = tables.iter().skip(1).map(|&k| laid[k][4]).collect(); + assert_eq!(parts, [0, 1, 2], "the PMT keeps its order"); + let keyframe = (1..laid.len()).find(|&k| pid(laid[k]) == 1).unwrap(); + assert!(tables.iter().all(|&k| k < keyframe), "the tables lead the keyframe"); + + // TBsys: 188 B in per table packet, drained at Rxsys over the slot's 25 ms. + let drain = 1_000_000.0 / 8.0 * 0.025 / positions as f64; + let (mut tb, mut peak) = (0.0f64, 0.0f64); + for packet in &laid { + tb = (tb - drain).max(0.0); + if [0, 100].contains(&pid(packet)) { + tb += 188.0; + peak = peak.max(tb); + } + } + assert!(peak <= 512.0, "TBsys peak {peak:.0} B with {ahead} packets ahead"); + } + } } From 6281a01b3281dbf8111c8831a9c96324244c502d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 12:14:39 -0700 Subject: [PATCH 4/8] test(mux): await the announced source after merging main Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/export_test.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index 911f3888ab..135bd1d646 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -7602,7 +7602,7 @@ async fn large_pmt(streams: u16, mux_rate: Option) -> usize { producers.push(producer); } catalog.modify().unwrap().ext.mpegts = config; - let mut exporter = Export::with_ts(crate::source::announced(&broadcast.consume()), Default::default()) + let mut exporter = Export::with_ts(crate::source::announced(&broadcast.consume()).await, Default::default()) .await .unwrap(); if let Some(mux_rate) = mux_rate { From c6637e8fa23aa321f0ca37831580fff1f33581ed Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 12:58:59 -0700 Subject: [PATCH 5/8] fix(mux): re-balance the slot around a PSI stall instead of pulling nulls forward Pulling the next nulls forward to space the tables stripped the gaps between a heavy video PID's packets, overflowing its transport buffer. Space a table with a null and lay the rest of the slot out afresh by smooth weighted round-robin over what remains, so every PID stays evenly spread. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/schedule.rs | 164 ++++++++++++++---------- 1 file changed, 95 insertions(+), 69 deletions(-) diff --git a/rs/moq-mux/src/container/ts/schedule.rs b/rs/moq-mux/src/container/ts/schedule.rs index e0d9787ee4..9a6397c9d4 100644 --- a/rs/moq-mux/src/container/ts/schedule.rs +++ b/rs/moq-mux/src/container/ts/schedule.rs @@ -202,8 +202,8 @@ impl Slot { /// before its first packet. /// /// The tables share one [`SystemBuffer`]. Where the next table would overflow it, a null - /// packet (or failing that, a packet muxed ahead of the table) takes the position instead, - /// so a PMT spanning several packets goes out at the buffer's rate. + /// takes the position instead, so a PMT spanning several packets goes out at the buffer's + /// rate where the slot has the nulls to spare. pub fn layout(&self, clock: &[u8], pmt_pid: u16, null: &[u8]) -> Vec { let packets = self.packets.as_chunks::<{ TsPacket::SIZE }>().0; let mut tables = VecDeque::new(); @@ -223,33 +223,18 @@ 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; - // The media and null positions in order, `None` for a null. - let mut order: VecDeque> = VecDeque::with_capacity(total as usize); - for _ in 1..total { - 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; - order.push_back(lanes.get_mut(pick).map(|(_, lane)| { - lane.pop_front() - .expect("a lane takes no more positions than its weight") - })); - } + // Positions the round-robin has still to lay. + let mut left = total - 1; let positions = 1 + packets.len() + self.nulls; let mut system = SystemBuffer::new(pmt_pid, positions); @@ -257,36 +242,65 @@ impl Slot { out.extend_from_slice(clock); system.lay(clock); loop { - // The next table, once the next media packet was muxed after it. + 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| order.front().is_none_or(|next| next.is_some_and(|i| table < i))); - let next = match table { - None => order.pop_front(), - Some(table) => { - let filler = || { - order - .iter() - .position(Option::is_none) - .or_else(|| order.iter().position(|next| next.is_some_and(|i| i < table))) - }; - match (!system.fits()).then(filler).flatten() { - Some(k) => order.remove(k), - // The table fits, or nothing is left to space it with. - None => { - tables.pop_front(); - system.lay(&packets[table]); - out.extend_from_slice(&packets[table]); - continue; - } - } + .filter(|&table| next.is_none_or(|next| next.is_some_and(|i| table < i))); + if let Some(table) = table { + // 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. + // Without a null left, the table goes out regardless. + if !system.fits() && nulls > 0 { + 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; } - }; - let Some(next) = next else { + tables.pop_front(); + system.lay(&packets[table]); + out.extend_from_slice(&packets[table]); + continue; + } + let Some(pick) = pick else { break; }; - let packet = next.map_or(null, |i| &packets[i]); + for (credit, weight) in credit.iter_mut().zip(&weights) { + *credit += weight; + } + credit[pick] -= total; + left -= 1; + let packet = match lanes.get_mut(pick) { + Some((_, lane)) => { + let i = lane + .pop_front() + .expect("a lane takes no more positions than its weight"); + &packets[i] + } + None => { + nulls -= 1; + null + } + }; system.lay(packet); out.extend_from_slice(packet); } @@ -1000,52 +1014,64 @@ mod tests { assert_eq!(laid, [1, 0, 100, 3, 0x1fff, 3, 3, 1, 0x1fff, 3]); } - /// A PAT and a three-packet PMT go out at TBsys's rate rather than in a burst, still in - /// order ahead of the keyframe muxed after them. Nulls space them in a padded slot, and a - /// packet muxed ahead of them where the slot has no nulls. + /// A PAT and a three-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 + /// still keeps within its own transport buffer. #[test] fn the_tables_go_out_at_the_system_buffer_rate() { - for (ahead, nulls) in [(0, 332 - 1 - 54), (40, 0)] { - let mut packets = unit(3, ahead); - packets.extend(unit(0, 1)); + // 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 + }; + // A 20 Mb/s slot with a small keyframe, and a 25 Mb/s slot whose video takes most of it + // alongside AC-3, MPEG audio and teletext. + let others: &[(u8, usize)] = &[(122, 9), (121, 5), (123, 5), (124, 5), (131, 1)]; + for (positions, video, count, others) in [(332, 1, 50, &[][..]), (416, 111, 300, others)] { + let mut packets = unit(0, 1); for part in 0..3 { let mut pmt = unit(100, 1); pmt[4] = part; packets.extend(pmt); } - packets.extend(unit(1, 50)); - let positions = 1 + packets.len() / TsPacket::SIZE + nulls; + 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, - nulls, keyframe: true, - units: vec![1], + units: vec![video.into()], }; let mut null = unit(0, 1); (null[1], null[2]) = (0x1f, 0xff); - let laid = slot.layout(&unit(1, 1), 100, &null); + 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 parts: Vec = tables.iter().skip(1).map(|&k| laid[k][4]).collect(); assert_eq!(parts, [0, 1, 2], "the PMT keeps its order"); - let keyframe = (1..laid.len()).find(|&k| pid(laid[k]) == 1).unwrap(); + 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"); - // TBsys: 188 B in per table packet, drained at Rxsys over the slot's 25 ms. - let drain = 1_000_000.0 / 8.0 * 0.025 / positions as f64; - let (mut tb, mut peak) = (0.0f64, 0.0f64); - for packet in &laid { - tb = (tb - drain).max(0.0); - if [0, 100].contains(&pid(packet)) { - tb += 188.0; - peak = peak.max(tb); - } - } - assert!(peak <= 512.0, "TBsys peak {peak:.0} B with {ahead} packets ahead"); + let system = peak(&laid, &[0, 100], 1_000_000.0); + assert!(system <= 512.0, "TBsys peak {system:.0} B in {positions} positions"); + // The video's Rx is 1.2 x 17,999,872 b/s, as in the full-slot test above. + let video = peak(&laid, &[video.into()], 21_599_846.0); + assert!(video <= 512.0, "video TB peak {video:.0} B in {positions} positions"); } } } From 1c7bdde58844de08062362090b06da7153a99e89 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 13:46:57 -0700 Subject: [PATCH 6/8] fix(mux): stall PSI only while delayed PIDs keep within their transport buffers No stall makes transmission time: a five-packet PMT ahead of a heavy video keyframe left the video too few positions and overflowed its transport buffer. Carry each bounded PID's per-slot transport rate on the slot, and space a table with a null only while every PID can still keep to it; past that the table goes out regardless. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/schedule.rs | 68 +++++++++++++++++++------ 1 file changed, 52 insertions(+), 16 deletions(-) diff --git a/rs/moq-mux/src/container/ts/schedule.rs b/rs/moq-mux/src/container/ts/schedule.rs index 9a6397c9d4..c7db2abbed 100644 --- a/rs/moq-mux/src/container/ts/schedule.rs +++ b/rs/moq-mux/src/container/ts/schedule.rs @@ -82,8 +82,13 @@ impl Buffer { /// the slot, less one, but at least one. Even laid out evenly, the packets bunch a little /// where one slot meets the next, and the slack drains that off. fn per_slot(&self) -> usize { + self.passes().saturating_sub(1).max(1) + } + + /// The packets its transport buffer passes on within a slot. + fn passes(&self) -> usize { let bits = u128::from(self.rate) * PCR_INTERVAL.as_nanos() / 1_000_000_000; - (bits / (TsPacket::SIZE as u128 * 8)).saturating_sub(1).max(1) as usize + (bits / (TsPacket::SIZE as u128 * 8)) as usize } } @@ -186,6 +191,8 @@ pub(super) struct Slot { pub keyframe: bool, /// The PID of each access unit whose first packet rides in the slot. pub units: Vec, + /// How many packets a slot each bounded PID's transport buffer passes on. + pub passes: HashMap, } impl Slot { @@ -203,7 +210,8 @@ impl Slot { /// /// The tables share one [`SystemBuffer`]. Where the next table would overflow it, a null /// takes the position instead, so a PMT spanning several packets goes out at the buffer's - /// rate where the slot has the nulls to spare. + /// rate, as long as the slot has nulls to spare and the delayed PIDs can still keep within + /// their own transport buffers ([`Self::passes`]). pub fn layout(&self, clock: &[u8], pmt_pid: u16, null: &[u8]) -> Vec { let packets = self.packets.as_chunks::<{ TsPacket::SIZE }>().0; let mut tables = VecDeque::new(); @@ -261,8 +269,18 @@ impl Slot { if let Some(table) = table { // 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. - // Without a null left, the table goes out regardless. - if !system.fits() && nulls > 0 { + // The stall must leave every PID the time to go no faster than its transport + // buffer passes packets on. Otherwise the table goes out regardless. + let room = || { + // Positions left after the stall, the remaining nulls among them. + let after = left as usize - 1 + tables.len(); + lanes.iter().all(|(pid, lane)| { + self.passes + .get(pid) + .is_none_or(|&passes| lane.len() * positions <= passes * after) + }) + }; + if !system.fits() && nulls > 0 && room() { nulls -= 1; system.lay(null); out.extend_from_slice(null); @@ -538,6 +556,11 @@ impl Schedule { nulls, keyframe, units, + passes: self + .buffers + .iter() + .map(|(&pid, buffer)| (pid, buffer.passes())) + .collect(), })) } @@ -971,6 +994,7 @@ mod tests { nulls: 416 - 1 - media, keyframe: false, units: vec![], + passes: HashMap::new(), }; let mut null = unit(0, 1); (null[1], null[2]) = (0x1f, 0xff); @@ -1007,6 +1031,7 @@ mod tests { nulls: 2, keyframe: true, units: vec![1, 3], + passes: HashMap::new(), }; let mut null = unit(0, 1); (null[1], null[2]) = (0x1f, 0xff); @@ -1014,9 +1039,10 @@ mod tests { assert_eq!(laid, [1, 0, 100, 3, 0x1fff, 3, 3, 1, 0x1fff, 3]); } - /// A PAT and a three-packet PMT go out at TBsys's rate rather than in a burst, spaced by + /// 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 - /// still keeps within its own transport buffer. + /// always keeps within its own transport buffer: where spacing a five-packet PMT would + /// leave the video too little time, 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 @@ -1033,12 +1059,19 @@ mod tests { } peak }; - // A 20 Mb/s slot with a small keyframe, and a 25 Mb/s slot whose video takes most of it - // alongside AC-3, MPEG audio and teletext. + // 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) in [(332, 1, 50, &[][..]), (416, 111, 300, others)] { + 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), + ] { let mut packets = unit(0, 1); - for part in 0..3 { + for part in 0..parts { let mut pmt = unit(100, 1); pmt[4] = part; packets.extend(pmt); @@ -1054,6 +1087,7 @@ mod tests { packets, keyframe: true, units: vec![video.into()], + passes: HashMap::from([(video.into(), Buffer { rate: RX, size: None }.passes())]), }; let mut null = unit(0, 1); (null[1], null[2]) = (0x1f, 0xff); @@ -1062,16 +1096,18 @@ mod tests { 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 parts: Vec = tables.iter().skip(1).map(|&k| laid[k][4]).collect(); - assert_eq!(parts, [0, 1, 2], "the PMT keeps its order"); + 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!(system <= 512.0, "TBsys peak {system:.0} B in {positions} positions"); - // The video's Rx is 1.2 x 17,999,872 b/s, as in the full-slot test above. - let video = peak(&laid, &[video.into()], 21_599_846.0); - assert!(video <= 512.0, "video TB peak {video:.0} B in {positions} positions"); + 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"); } } } From d7dc586e711f9f20c35973bb868f647ff155b5d9 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 14:20:52 -0700 Subject: [PATCH 7/8] fix(mux): hold a PSI stall to admission's per-slot cap The stall guard lent the pending tables' positions to the PIDs they delay and allowed the full transport rate without admission's one-packet slack, so video near its cap still overflowed (638 B). Compare each PID's remaining packets over the round-robin's remaining positions against the same per-slot cap admission enforces. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/schedule.rs | 43 ++++++++++++------------- 1 file changed, 21 insertions(+), 22 deletions(-) diff --git a/rs/moq-mux/src/container/ts/schedule.rs b/rs/moq-mux/src/container/ts/schedule.rs index c7db2abbed..126d08a383 100644 --- a/rs/moq-mux/src/container/ts/schedule.rs +++ b/rs/moq-mux/src/container/ts/schedule.rs @@ -82,13 +82,8 @@ impl Buffer { /// the slot, less one, but at least one. Even laid out evenly, the packets bunch a little /// where one slot meets the next, and the slack drains that off. fn per_slot(&self) -> usize { - self.passes().saturating_sub(1).max(1) - } - - /// The packets its transport buffer passes on within a slot. - fn passes(&self) -> usize { let bits = u128::from(self.rate) * PCR_INTERVAL.as_nanos() / 1_000_000_000; - (bits / (TsPacket::SIZE as u128 * 8)) as usize + (bits / (TsPacket::SIZE as u128 * 8)).saturating_sub(1).max(1) as usize } } @@ -191,8 +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, - /// How many packets a slot each bounded PID's transport buffer passes on. - pub passes: HashMap, + /// The most packets a slot admission lets each bounded PID carry. + pub caps: HashMap, } impl Slot { @@ -211,7 +206,7 @@ impl Slot { /// The tables share one [`SystemBuffer`]. Where the next table 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::passes`]). + /// 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 = VecDeque::new(); @@ -269,15 +264,17 @@ impl Slot { if let Some(table) = table { // 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 the time to go no faster than its transport - // buffer passes packets on. Otherwise the table goes out regardless. + // 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. let room = || { - // Positions left after the stall, the remaining nulls among them. - let after = left as usize - 1 + tables.len(); + // 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.passes + self.caps .get(pid) - .is_none_or(|&passes| lane.len() * positions <= passes * after) + .is_none_or(|&cap| lane.len() * positions <= cap * after) }) }; if !system.fits() && nulls > 0 && room() { @@ -556,10 +553,10 @@ impl Schedule { nulls, keyframe, units, - passes: self + caps: self .buffers .iter() - .map(|(&pid, buffer)| (pid, buffer.passes())) + .map(|(&pid, buffer)| (pid, buffer.per_slot())) .collect(), })) } @@ -994,7 +991,7 @@ mod tests { nulls: 416 - 1 - media, keyframe: false, units: vec![], - passes: HashMap::new(), + caps: HashMap::new(), }; let mut null = unit(0, 1); (null[1], null[2]) = (0x1f, 0xff); @@ -1031,7 +1028,7 @@ mod tests { nulls: 2, keyframe: true, units: vec![1, 3], - passes: HashMap::new(), + caps: HashMap::new(), }; let mut null = unit(0, 1); (null[1], null[2]) = (0x1f, 0xff); @@ -1041,8 +1038,9 @@ mod tests { /// 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 a five-packet PMT would - /// leave the video too little time, the rest of the tables go out regardless. + /// 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 @@ -1069,6 +1067,7 @@ mod tests { (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 { @@ -1087,7 +1086,7 @@ mod tests { packets, keyframe: true, units: vec![video.into()], - passes: HashMap::from([(video.into(), Buffer { rate: RX, size: None }.passes())]), + caps: HashMap::from([(video.into(), Buffer { rate: RX, size: None }.per_slot())]), }; let mut null = unit(0, 1); (null[1], null[2]) = (0x1f, 0xff); From 0abbce3dd5fad0b2220b9ef81009ea42e0a9fc6d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 14:43:40 -0700 Subject: [PATCH 8/8] fix(mux): stall PSI only ahead of a slot's first media packet Laying the slot afresh after media had gone out dropped the spacing already given the PIDs served, so tables ahead of a second keyframe bunched the first keyframe's PID (634 B). Stall only while no media packet has gone out; later tables go out as before. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/schedule.rs | 72 +++++++++++++++++++++---- 1 file changed, 62 insertions(+), 10 deletions(-) diff --git a/rs/moq-mux/src/container/ts/schedule.rs b/rs/moq-mux/src/container/ts/schedule.rs index 126d08a383..6611cb2291 100644 --- a/rs/moq-mux/src/container/ts/schedule.rs +++ b/rs/moq-mux/src/container/ts/schedule.rs @@ -203,10 +203,10 @@ impl Slot { /// 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 the next table 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`]). + /// 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 = VecDeque::new(); @@ -238,6 +238,8 @@ impl Slot { credit[0] -= total; // 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); @@ -262,11 +264,12 @@ impl Slot { .copied() .filter(|&table| next.is_none_or(|next| next.is_some_and(|i| table < i))); if let Some(table) = table { - // 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. + // 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. @@ -277,7 +280,7 @@ impl Slot { .is_none_or(|&cap| lane.len() * positions <= cap * after) }) }; - if !system.fits() && nulls > 0 && room() { + if fresh && !system.fits() && nulls > 0 && room() { nulls -= 1; system.lay(null); out.extend_from_slice(null); @@ -306,6 +309,7 @@ impl Slot { 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"); @@ -1109,4 +1113,52 @@ mod tests { 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"); + } + } }