From 64ec5acf7bbfd7c7e5e6bb52bf43af300ee2d669 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 16:48:02 -0700 Subject: [PATCH 1/2] chore(quest): claim quest/m1/ts-reliability/psi-state.md Quest-Claim: c7756844eac9580e From 028a409fbaec6c4dfb62440cfe56a11a29276334 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 28 Oct 2026 17:12:13 -0700 Subject: [PATCH 2/2] refactor(mux): lay the TS export's tables from current state The export queued an encoded PAT and PMT with each media unit it led, its version fixed at push. Now the schedule keeps the tables as they stand (`Schedule::set_psi`) and every copy not yet begun lays them, so a table change is one state update; a copy already on its way finishes as it began. A revision takes the version after the last copy to begin, so one overtaken before any copy of it began takes none. The break flag after a switch rides the first copy to go out. A static program's output is byte-identical: every moq-mux test's export output matches main's. Co-Authored-By: Claude Opus 5.5 --- quest/m1/ts-reliability/README.md | 1 - quest/m1/ts-reliability/pat-lifecycle.md | 18 +- quest/m1/ts-reliability/psi-state.md | 29 --- rs/moq-mux/src/container/ts/export.rs | 251 ++++++++++++++++------- rs/moq-mux/src/container/ts/schedule.rs | 201 +++++++++++++++--- 5 files changed, 357 insertions(+), 143 deletions(-) delete mode 100644 quest/m1/ts-reliability/psi-state.md diff --git a/quest/m1/ts-reliability/README.md b/quest/m1/ts-reliability/README.md index 1a5fc0bc8a2..2263286d940 100644 --- a/quest/m1/ts-reliability/README.md +++ b/quest/m1/ts-reliability/README.md @@ -102,7 +102,6 @@ catalog linger (#5305). - [A timebase step restarts the TS import](/quest/m1/ts-reliability/step-restart.md) - an unflagged step or a rewind republishes under a new epoch instead of ending the import and every subscriber's export - [Export re-acquires a step](/quest/m1/ts-reliability/export-reacquire.md) - `moq export ts` re-acquires an unmarked timeline step instead of exiting on fake drift - [A restart keeps its catalog](/quest/m1/ts-reliability/restart-catalog.md) - a restarted import publishes its catalog with its first media -- [Tables from current state](/quest/m1/ts-reliability/psi-state.md) - the TS export lays its PAT and PMT from current state when their slot comes up, with byte-identical output for a static program - [PAT changes](/quest/m1/ts-reliability/pat-lifecycle.md) - a running export follows a moved PMT PID - [A demoted subscriber priority reaches the publisher](/quest/m1/ts-reliability/stitch-catch-up.md) - a mid-stream priority demotion reaches the publisher on lite-07 and moq-transport, including an old instance after an epoch switch diff --git a/quest/m1/ts-reliability/pat-lifecycle.md b/quest/m1/ts-reliability/pat-lifecycle.md index d78ca6172cf..e649d3d8778 100644 --- a/quest/m1/ts-reliability/pat-lifecycle.md +++ b/quest/m1/ts-reliability/pat-lifecycle.md @@ -19,11 +19,15 @@ export started after it. The maintainer's plan, from closing [moq-dev/moq#6301](https://github.com/moq-dev/moq/pull/6301): the export follows a new `program.pmtPid` with a new PAT, built on the table rework -that lays the PAT and PMT from the current table state when their slot is -laid instead of riding queued units, so a move is one more state change -rather than a rewrite of queued copies. That rework is its own quest, -[tables from current state](/quest/m1/ts-reliability/psi-state.md) -(planned 2026-10-28), so this quest keeps only the PMT move and its tests. +that lays the PAT and PMT from the current table state when a copy of them +begins to go out, so a move is one more state change rather than a rewrite +of queued copies. That rework has landed: the export's `Psi` holds the +tables as they stand, `Export::restate` hands them to +`Schedule::set_psi`, which every copy not yet begun lays, and a copy +already on its way finishes as it began. A move then updates `Psi::pmt_pid`, +calls `Schedule::set_tables` for the new PID, and restates; `Psi::current` +gives the PAT the next version when the PID differs from the last laid. +This quest keeps only the PMT move and its tests. The draft then says that `pmtPid` can change in place, and when a consumer rebuilding the stream moves the PMT. @@ -36,7 +40,3 @@ old one after it, and a move while every copy is still queued goes out as one PAT naming the new PID. `revision_test`'s `a_track_recorded_on_the_pmt_pid_after_the_catalog_moves_it_takes_another` then pins the move instead of the PMT staying put. - -## Required - -- [Tables from current state](/quest/m1/ts-reliability/psi-state.md) - the export lays the PAT and PMT from current state, so a move is one more state change diff --git a/quest/m1/ts-reliability/psi-state.md b/quest/m1/ts-reliability/psi-state.md deleted file mode 100644 index 857bc42adc9..00000000000 --- a/quest/m1/ts-reliability/psi-state.md +++ /dev/null @@ -1,29 +0,0 @@ -# [M] The TS export lays its tables from current state - -## Goal - -The TS export lays its PAT and PMT from the current table state when their -slot comes up, as `--sync` already does, instead of carrying queued copies -alongside media units. A static program's output is byte-identical to today. - -## Plan - -Planned 2026-10-28 (maintainer). The maintainer named this rework when -closing [moq-dev/moq#6301](https://github.com/moq-dev/moq/pull/6301): the -export's table handling from -[moq-dev/moq#6289](https://github.com/moq-dev/moq/pull/6289) queues copies of -the PAT and PMT with media units, so a table change rewrites queued copies. -Laying each table from current state when its slot is laid makes a change -one more state update, which [PAT changes](/quest/m1/ts-reliability/pat-lifecycle.md) -builds on. - -- Reuse what `--sync` already does to lay tables, rather than adding a second - path. -- Keep #6289's revision rule: a PMT revision goes out at the next version. -- Prove byte-identical output on the existing export fixtures and round - trips, and keep the T-STD schedule and the opening order from - [moq-dev/moq#7048](https://github.com/moq-dev/moq/pull/7048) unchanged. - -Rust: `rs/moq-mux` TS export (`build_psi`, the PSI schedule). - -Public API: none. Wire: none. diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index 86c84cbb12a..1084f09a5f4 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -236,8 +236,8 @@ struct Tally { pcr: (u64, bool), /// The PID of each access unit whose packets begin in the frame. units: Vec, - /// The last program tables to finish in the frame, as [`Psi::queued`] holds them. - tables: Option<(VersionNumber, psi::Pmt)>, + /// The last program tables to finish in the frame. + tables: Option, } /// A video frame waiting on a decode clock, with the codec config it was read under. @@ -972,25 +972,37 @@ impl Kind { } /// The program tables: the PAT, fixed for the program's life but for its network entry, and the -/// PMT, which lists the streams on the air. +/// PMT, which lists the streams on the air. The schedule lays them as they stand when each copy +/// begins to go out ([`Schedule::set_psi`]), so a change is one update of this state, and a +/// revision overtaken before any copy of it began takes no version. struct Psi { - pat: Bytes, /// The source's `transport_stream_id`, so the rebuilt PAT stays consistent with the carried SI. transport_stream_id: u16, - /// Whether the PAT lists the network entry, pointing at the NIT the export carries. - network: bool, /// What the PMT says now. pmt: psi::Pmt, /// PID the PMT rides on: the source's original (preserved from the service /// record) or the synthesized [`PMT_PID`] for a media-only source. pmt_pid: u16, - /// The version and PMT last handed to the schedule, which the next copy follows. - pushed: Option<(VersionNumber, psi::Pmt)>, - /// The versions handed to the schedule and not yet returned, oldest first: they go out in - /// the order they were pushed ([`Schedule::lead`]). - queued: VecDeque<(VersionNumber, psi::Pmt)>, - /// The version and PMT of the last tables returned. - out: Option<(VersionNumber, psi::Pmt)>, + /// The PMT the last tables handed to the schedule listed: the streams the next copy adds. + led: Option, + /// The tables last set on the schedule, and the PIDs they flag a break on. + stated: Option<(Laid, Vec)>, + /// The tables of the last copy to begin going out, whose version the next change follows. + laid: Option, + /// The tables of each copy begun and not yet finished, oldest first. + sending: VecDeque, + /// The last tables returned. + out: Option, +} + +/// Program tables as a copy of them goes out: the PMT at its version, and the PAT. +#[derive(Clone)] +struct Laid { + version: VersionNumber, + pmt: psi::Pmt, + /// Whether the PAT lists the network entry, pointing at the NIT the export carries. + network: bool, + pat: Bytes, } impl Psi { @@ -1004,6 +1016,37 @@ impl Psi { self.pcr_pid() .filter(|&pcr| self.pmt.streams.iter().all(|es| es.pid.as_u16() != pcr)) } + + /// The tables as they stand, listing the network entry when `network`. A PMT that differs + /// from the last laid goes out at the next version, `first` before any has; the PAT keeps its + /// version until its network entry changes, so it never runs ahead of the PMT's. + fn current(&self, network: bool, first: VersionNumber) -> anyhow::Result { + let (version, pat) = match &self.laid { + Some(laid) if laid.pmt == self.pmt && laid.network == network => return Ok(laid.clone()), + Some(laid) => { + let mut version = laid.version; + version.increment(); + (version, (laid.network == network).then(|| laid.pat.clone())) + } + None => (first, None), + }; + let pat = match pat { + Some(pat) => pat, + None => encode_pat( + self.transport_stream_id, + self.pmt.program_number, + self.pmt_pid, + network, + version, + )?, + }; + Ok(Laid { + version, + pmt: self.pmt.clone(), + network, + pat, + }) + } } /// The DVB NIT's PID (ETSI EN 300 468 5.1.3), the network PID the PAT's program 0 names. @@ -1931,7 +1974,7 @@ where .psi .as_ref() .and_then(|psi| psi.out.as_ref()) - .map(|(version, _)| *version); + .map(|laid| laid.version); next.version = out.unwrap_or(self.program.version); next.epoch = self.program.epoch; next.emitted_epoch = self.program.emitted_epoch; @@ -2666,8 +2709,9 @@ where self.program.pcr_discontinuity = true; self.program.tables.clear(); if let Some(psi) = &mut self.program.psi { - psi.queued.clear(); - psi.pushed = psi.out.clone(); + psi.sending.clear(); + psi.laid = psi.out.clone(); + psi.led = psi.out.as_ref().map(|laid| laid.pmt.clone()); } for track in self.program.tracks.values_mut() { track.pack = None; @@ -2709,11 +2753,19 @@ where /// to flag the break. The PCR PID's first packet always opens a slot with its PCR, flagged /// on its own ([`Self::lay`]). fn breaks(&mut self, pid: u16) -> bool { - let pcr = self.program.psi.as_ref().and_then(Psi::pcr_pid); - match &mut self.program.flagged { - Some(flagged) => Some(pid) != pcr && !self.program.counters.contains_key(&pid) && flagged.insert(pid), - None => false, + let owes = self.owes_break(pid); + if owes && let Some(flagged) = &mut self.program.flagged { + flagged.insert(pid); } + owes + } + + /// Whether `pid`'s next packet has to flag the break ([`Self::breaks`]), without spending it. + fn owes_break(&self, pid: u16) -> bool { + let pcr = self.program.psi.as_ref().and_then(Psi::pcr_pid); + self.program.flagged.as_ref().is_some_and(|flagged| { + Some(pid) != pcr && !self.program.counters.contains_key(&pid) && !flagged.contains(&pid) + }) } /// PID the PMT rides on: the source's original (preserved in the service record), @@ -2748,14 +2800,6 @@ where .as_ref() .map(|s| s.transport_stream_id) .unwrap_or(1); - let network = self.carries_nit(); - let pat = encode_pat( - transport_stream_id, - pmt.program_number, - pmt_pid, - network, - self.program.version, - )?; self.program.schedule.set_tables(pmt_pid); self.program.schedule.set_clock(pcr_pid); @@ -2768,13 +2812,13 @@ where self.program.liveness.register(track.pid); } self.program.psi = Some(Psi { - pat, transport_stream_id, - network, pmt, pmt_pid, - pushed: None, - queued: VecDeque::new(), + led: None, + stated: None, + laid: None, + sending: VecDeque::new(), out: None, }); self.program.psi_due = true; @@ -2889,6 +2933,7 @@ where if pmt != psi.pmt { psi.pmt = pmt; self.program.revised = true; + self.restate()?; } Ok(()) } @@ -3000,59 +3045,63 @@ where Ok(()) } - /// Hand the program tables to the schedule ahead of the next unit, if they are due. A PMT - /// that differs from the last handed over goes out under the next version. + /// Hand the program tables to the schedule ahead of the next unit, if they are due. fn lead_tables(&mut self) -> anyhow::Result<()> { if !std::mem::take(&mut self.program.psi_due) { return Ok(()); } - // A NIT acquired or lost after the tables went out revises the PAT's network entry, at - // the next version beside the PMT, so the PAT's version never runs ahead of the PMT's. - let network = self.carries_nit(); + self.restate()?; let psi = self.program.psi.as_mut().context("PSI not built")?; - let version = match &psi.pushed { - Some((version, pmt)) if *pmt == psi.pmt && psi.network == network => *version, - Some((version, _)) => { - let mut version = *version; - version.increment(); - version - } - None => self.program.version, - }; - if psi.network != network { - psi.pat = encode_pat( - psi.transport_stream_id, - psi.pmt.program_number, - psi.pmt_pid, - network, - version, - )?; - psi.network = network; - } let added: Vec = psi .pmt .streams .iter() - .filter(|es| psi.pushed.as_ref().is_none_or(|(_, pmt)| !pmt.streams.contains(es))) + .filter(|es| psi.led.as_ref().is_none_or(|pmt| !pmt.streams.contains(es))) .map(|es| es.pid.as_u16()) .collect(); - let section = super::psi::encode_pmt(&psi.pmt, version)?; - let clock = psi.pcr_pid().context("PMT without a clock")?; - psi.pushed = Some((version, psi.pmt.clone())); - psi.queued.push_back((version, psi.pmt.clone())); - let (pat, pmt_pid) = (psi.pat.clone(), psi.pmt_pid); - let mut packets = Vec::new(); - self.write_section(&mut packets, Pid::PAT, &pat)?; - self.write_section(&mut packets, pmt_pid, §ion)?; + psi.led = Some(psi.pmt.clone()); self.program.schedule.lead(super::schedule::Tables { - packets, - clock, rate: self.program.mux_rate, added, }); Ok(()) } + /// Set the program tables as they stand on the schedule, which every copy of them still to + /// begin lays. A NIT acquired or lost after the tables went out revises the PAT's network + /// entry, at the next version beside the PMT. The first copy after a switch to another + /// instance flags the break on each table PID that owes one ([`Self::breaks`]). + fn restate(&mut self) -> anyhow::Result<()> { + let network = self.carries_nit(); + let psi = self.program.psi.as_ref().context("PSI not built")?; + let tables = psi.current(network, self.program.version)?; + let pmt_pid = psi.pmt_pid; + let clock = psi.pcr_pid().context("PMT without a clock")?; + let section = super::psi::encode_pmt(&tables.pmt, tables.version)?; + let breaks: Vec = [Pid::PAT, pmt_pid] + .into_iter() + .filter(|&pid| self.owes_break(pid)) + .collect(); + let encode = |flag: &dyn Fn(u16) -> bool| -> anyhow::Result> { + let mut packets = Vec::new(); + section_packets(&mut packets, Pid::PAT, &tables.pat, flag(Pid::PAT))?; + section_packets(&mut packets, pmt_pid, §ion, flag(pmt_pid))?; + Ok(packets) + }; + let packets = encode(&|_| false)?; + let flagged = match breaks.is_empty() { + true => None, + false => Some(encode(&|pid| breaks.contains(&pid))?), + }; + self.program.schedule.set_psi(super::schedule::Psi { + packets, + flagged, + clock, + }); + self.program.psi.as_mut().expect("checked above").stated = Some((tables, breaks)); + Ok(()) + } + /// Packetize one media frame and queue it on the [`Schedule`], re-emitting PAT/PMT /// before video keyframes (and periodically) so receivers can tune in mid-stream. fn mux(&mut self, name: &str, queued: Queued) -> anyhow::Result<()> { @@ -3524,13 +3573,27 @@ where keyframe: slot.keyframe, }; let psi = self.program.psi.as_mut().expect("checked above"); - let tables = (0..slot.tables).filter_map(|_| psi.queued.pop_front()).last(); + // Each copy begun laid the tables as they stood, and the first spent the breaks they flag. + let mut spent = Vec::new(); + if slot.begun > 0 { + let (tables, breaks) = psi.stated.clone().context("program tables laid but never set")?; + psi.sending.extend(std::iter::repeat_n(tables.clone(), slot.begun)); + psi.laid = Some(tables); + spent = breaks; + } + let tables = (0..slot.tables).filter_map(|_| psi.sending.pop_front()).last(); let tally = Tally { pcr: (slot.pcr, discontinuity), units: slot.units, tables, }; self.program.queue.push_back((frame, tally)); + if !spent.is_empty() { + if let Some(flagged) = &mut self.program.flagged { + flagged.extend(spent); + } + self.restate()?; + } } Ok(()) } @@ -4536,10 +4599,58 @@ mod tests { use moq_net::Timestamp; use super::{ - Buffer, DecodeClock, Dolby, Kind, PSI_INTERVAL, PesPacket, Settle, dolby_removals, due, is_complete_section, - si_due, slot, unwrap_pts, verbatim_buffer, + Buffer, DecodeClock, Dolby, Kind, PSI_INTERVAL, PesPacket, Psi, Settle, dolby_removals, due, + is_complete_section, si_due, slot, unwrap_pts, verbatim_buffer, }; + /// The tables as they stand take the version after the last laid, whatever was set on the + /// schedule meanwhile, and the last laid's own when they come back to it. The PAT keeps its + /// version until its network entry changes. + #[test] + fn a_revision_takes_the_version_after_the_last_laid() { + use mpeg2ts::ts::{Pid, VersionNumber}; + + let pmt = |pids: &[u16]| super::psi::Pmt { + program_number: 1, + pcr_pid: Pid::new(0x100).ok(), + program_info: Vec::new(), + streams: pids + .iter() + .map(|&pid| super::psi::Es { + stream_type: 0x0f, + pid: Pid::new(pid).unwrap(), + descriptors: Vec::new(), + }) + .collect(), + }; + let mut psi = Psi { + transport_stream_id: 1, + pmt: pmt(&[0x100]), + pmt_pid: 0x20, + led: None, + stated: None, + laid: None, + sending: Default::default(), + out: None, + }; + let first = VersionNumber::from_u8(3).unwrap(); + let laid = psi.current(false, first).unwrap(); + assert_eq!(laid.version.as_u8(), 3); + psi.laid = Some(laid.clone()); + + psi.pmt = pmt(&[0x100, 0x101]); + let revised = psi.current(false, first).unwrap(); + assert_eq!(revised.version.as_u8(), 4); + assert_eq!(revised.pat, laid.pat); + + psi.pmt = pmt(&[0x100]); + assert_eq!(psi.current(false, first).unwrap().version.as_u8(), 3); + + let network = psi.current(true, first).unwrap(); + assert_eq!(network.version.as_u8(), 4); + assert_ne!(network.pat, laid.pat); + } + /// The same available video/audio backlog must choose the same start in either fill order. #[tokio::test(start_paused = true)] async fn track_fill_observes_every_available_track_before_acquisition() { diff --git a/rs/moq-mux/src/container/ts/schedule.rs b/rs/moq-mux/src/container/ts/schedule.rs index b091c3cb0a7..117b4b3541d 100644 --- a/rs/moq-mux/src/container/ts/schedule.rs +++ b/rs/moq-mux/src/container/ts/schedule.rs @@ -157,12 +157,21 @@ pub(super) struct Timing { pub decode: u128, } -/// Program tables to go out ahead of the next unit pushed ([`Schedule::lead`]). -pub(super) struct Tables { +/// The program tables as they stand ([`Schedule::set_psi`]), which each unit of them lays when +/// it begins to go out. +#[derive(Clone, PartialEq, Eq)] +pub(super) struct Psi { /// The PAT and PMT packets. pub packets: Vec, + /// The same tables flagging a break, for the first unit of them to go out after a switch to + /// another instance. + pub flagged: Option>, /// The PID every slot's clock rides once they have gone out. pub clock: u16, +} + +/// Program tables to go out ahead of the next unit pushed ([`Schedule::lead`]). +pub(super) struct Tables { /// The multiplex rate from the slot after the one they finish in. pub rate: Option, /// The PIDs they list that the tables before them did not. Their units wait for them, and @@ -431,6 +440,9 @@ pub(super) struct Slot { /// How many units of program tables finish in the slot ([`Schedule::lead`]), which go /// out in the order they were pushed. pub tables: usize, + /// How many units of program tables begin in the slot, each laying the tables as they + /// stood ([`Schedule::set_psi`]). + pub begun: usize, } impl Slot { @@ -593,6 +605,8 @@ pub(super) struct Schedule { ended: bool, /// The sequence number of the next unit pushed. seq: u64, + /// The program tables as they stand, from [`Self::set_psi`]. + psi: Option, /// Tables to go out ahead of the next unit pushed. lead: Option, /// The timeline opens ([`Self::open`]) and its first tables are still to go out. @@ -626,6 +640,7 @@ impl Schedule { mb: FxHashMap::default(), ended: false, seq: 0, + psi: None, lead: None, opening: false, clocked: 0, @@ -763,10 +778,41 @@ impl Schedule { self.units.iter().any(|unit| unit.pid == pid && unit.tables.is_none()) } - /// Send `tables` ahead of the next unit pushed, under its deadline. Program tables ride - /// one lane, so each version goes out after the one before it, and a revision is one - /// boundary: no unit of a stream they add goes out before them, and their clock PID and - /// multiplex rate take over only once they have gone out. + /// Set the program tables as they stand. Every unit of them still to begin lays them, and + /// takes over their clock PID, so a change reaches each copy not yet on its way. + pub fn set_psi(&mut self, psi: Psi) { + if self.psi.as_ref() != Some(&psi) { + self.psi = Some(psi); + self.retable(); + } + } + + /// Lay the tables as they stand in each unit of them still to begin, the first flagging + /// the break when they do ([`Psi::flagged`]). + fn retable(&mut self) { + let Some(psi) = &self.psi else { + return; + }; + let mut flagged = psi.flagged.as_ref(); + let mut units = std::mem::take(&mut self.units); + for unit in units.iter_mut().filter(|unit| unit.sent == 0) { + let Some(program) = &mut unit.tables else { + continue; + }; + program.clock = psi.clock; + let packets = flagged.take().unwrap_or(&psi.packets); + if unit.packets != *packets { + unit.packets.clone_from(packets); + unit.runs = self.runs(&unit.packets); + } + } + self.units = units; + } + + /// Send the program tables ahead of the next unit pushed, under its deadline. Program + /// tables ride one lane, so each version goes out after the one before it, and a revision + /// is one boundary: no unit of a stream they add goes out before them, and their clock PID + /// and multiplex rate take over only once they have gone out. pub fn lead(&mut self, tables: Tables) { self.lead = Some(tables); } @@ -826,11 +872,12 @@ impl Schedule { ) -> anyhow::Result<()> { if let Some(tables) = self.lead.take() { let pmt = self.pmt.context("program tables without a PMT PID")?; + let clock = self.psi.as_ref().context("program tables never set")?.clock; // Later tables wait too, so each version follows the one before it. let mut holds = tables.added; holds.extend([pid, pmt]); let program = Program { - clock: tables.clock, + clock, rate: tables.rate, holds, leads: pid, @@ -843,8 +890,9 @@ impl Schedule { }; // The unit they lead keeps its own start. let last = self.last; - self.enqueue(pmt, lead, tables.packets, Some(program)); + self.enqueue(pmt, lead, Vec::new(), Some(program)); self.last = last; + self.retable(); } if packets.is_empty() { return Ok(()); @@ -1111,12 +1159,14 @@ impl Schedule { let mut packets = Vec::new(); let mut keyframe = false; let mut units = Vec::new(); + let mut begun = 0; for (unit, take) in self.units.iter_mut().zip(take) { if take == 0 { continue; } if unit.sent == 0 { keyframe |= unit.keyframe; + begun += usize::from(unit.tables.is_some()); // Only the tables muxed ahead of a frame that wrote nothing are no access unit. if unit.tables.is_none() && unit.packets_on(0..unit.count(), unit.pid) > 0 { units.extend(std::iter::repeat_n(unit.pid, unit.frames.len().max(1))); @@ -1144,6 +1194,7 @@ impl Schedule { clocked: false, clock, tables, + begun, })) } @@ -1167,6 +1218,7 @@ impl Schedule { clocked: false, clock: self.clock, tables: 0, + begun: 0, }; let clock = self.clock.filter(|pid| self.buffers.contains_key(pid)); let mut holding = self.holding(|i| self.units[i].remaining()); @@ -1430,6 +1482,7 @@ impl Schedule { let unit = &mut self.units[i]; if unit.sent == 0 { out.keyframe |= unit.keyframe; + out.begun += usize::from(unit.tables.is_some()); if unit.tables.is_none() && unit.packets_on(0..unit.count(), unit.pid) > 0 { out.units .extend(std::iter::repeat_n(unit.pid, unit.frames.len().max(1))); @@ -1626,6 +1679,15 @@ mod tests { ms * 1_000_000 } + /// A PAT and a PMT on PID 100 of a packet each, the clock on PID 1. + fn set_psi(schedule: &mut Schedule) { + schedule.set_psi(Psi { + packets: [unit(0, 1), unit(100, 1)].concat(), + flagged: None, + clock: 1, + }); + } + /// 40 packets per slot, the clock packet included. const RATE: u64 = 40 * PACKET; /// A decoder buffer's reach past every window here, so the window bounds how far ahead a @@ -1736,11 +1798,10 @@ mod tests { let mut schedule = Schedule::new(Duration::from_millis(25)); schedule.set_rate(Some(RATE)); schedule.set_tables(100); + set_psi(&mut schedule); schedule.set_clock(1); schedule.set_buffer(1, VIDEO); let tables = || Tables { - packets: [unit(0, 1), unit(100, 1)].concat(), - clock: 1, rate: Some(RATE), added: Vec::new(), }; @@ -1776,6 +1837,7 @@ mod tests { let mut schedule = Schedule::new(Duration::ZERO); schedule.set_rate(rate); schedule.set_tables(100); + set_psi(&mut schedule); schedule.set_clock(1); schedule.set_buffer(1, VIDEO); if open { @@ -1785,12 +1847,7 @@ mod tests { let mut section = unit(7, 1); section[1] |= 0x40; schedule.push_placed(7, (0, 0), 1, section, false, Vec::new()).unwrap(); - schedule.lead(Tables { - packets: [unit(0, 1), unit(100, 1)].concat(), - clock: 1, - rate, - added: vec![1], - }); + schedule.lead(Tables { rate, added: vec![1] }); schedule.push(1, timing(10), unit(1, 3), true).unwrap(); let slots = drain(&mut schedule); assert!(!schedule.clocking()); @@ -1818,17 +1875,13 @@ mod tests { let mut schedule = Schedule::new(Duration::ZERO); schedule.set_rate(rate); schedule.set_tables(100); + set_psi(&mut schedule); schedule.set_clock(1); schedule.set_buffer(1, VIDEO); schedule.set_buffer(0x11, Buffer::SYSTEM); schedule.open(); schedule.push_clocked(unit(0x11, 1)); - schedule.lead(Tables { - packets: [unit(0, 1), unit(100, 1)].concat(), - clock: 1, - rate, - added: vec![1], - }); + schedule.lead(Tables { rate, added: vec![1] }); schedule.push(1, timing(10), unit(1, 3), true).unwrap(); assert!(schedule.clocking()); let slots = drain(&mut schedule); @@ -1853,17 +1906,13 @@ mod tests { let mut schedule = Schedule::new(Duration::ZERO); schedule.set_rate(rate); schedule.set_tables(100); + set_psi(&mut schedule); schedule.set_clock(1); schedule.set_buffer(1, VIDEO); schedule.set_buffer(0x11, Buffer::SYSTEM); schedule.open(); schedule.push_clocked(unit(0x11, 1)); - schedule.lead(Tables { - packets: [unit(0, 1), unit(100, 1)].concat(), - clock: 1, - rate, - added: vec![1], - }); + schedule.lead(Tables { rate, added: vec![1] }); schedule.push(1, timing(10), unit(1, 1), true).unwrap(); // The next frame is ten seconds out, far past a skip. let known = Some(400); @@ -1890,6 +1939,92 @@ mod tests { } } + /// A PAT and a PMT on PID 100 of `count` packets, each packet marked `mark`. + fn marked_tables(mark: u8, count: usize) -> Vec { + let mut packets = [unit(0, 1), unit(100, count)].concat(); + for packet in packets.as_chunks_mut::<{ TsPacket::SIZE }>().0 { + packet[4] = mark; + } + packets + } + + /// The mark of each table packet laid in `slot`. + fn table_marks(slot: &Slot) -> Vec { + slot.packets + .as_chunks::<{ TsPacket::SIZE }>() + .0 + .iter() + .filter(|packet| matches!(pid(*packet), 0 | 100)) + .map(|packet| packet[4]) + .collect() + } + + /// Each copy of the tables lays them as they stand when it begins: a change reaches every + /// copy still to begin, and never one already on its way, which goes out whole as it began. + #[test] + fn the_tables_lay_as_they_stand_when_they_begin() { + let mut schedule = Schedule::new(Duration::from_secs(1)); + schedule.set_rate(Some(RATE)); + schedule.set_tables(100); + schedule.set_clock(1); + schedule.set_buffer(1, VIDEO); + let psi = |mark: u8| Psi { + packets: marked_tables(mark, 12), + flagged: None, + clock: 1, + }; + schedule.set_psi(psi(1)); + for ms in [500, 600] { + schedule.lead(Tables { + rate: Some(RATE), + added: Vec::new(), + }); + schedule.push(1, timing(ms), unit(1, 1), true).unwrap(); + } + schedule.set_psi(psi(2)); + let mut marks = Vec::new(); + let mut begun = 0; + while let Some(slot) = schedule.next(None).unwrap() { + marks.extend(table_marks(&slot)); + begun += slot.begun; + // Too large for Bsys to take at once, the first copy spans slots. + if begun == 1 && slot.tables == 0 { + schedule.set_psi(psi(3)); + } + } + assert_eq!(begun, 2); + assert_eq!(marks, [[2; 13], [3; 13]].concat()); + } + + /// Only the first copy of the tables to go out after a switch flags the break. + #[test] + fn only_the_first_tables_flag_the_break() { + for rate in [None, Some(RATE)] { + let mut schedule = Schedule::new(Duration::ZERO); + schedule.set_rate(rate); + schedule.set_tables(100); + schedule.set_clock(1); + schedule.set_buffer(1, VIDEO); + schedule.set_psi(Psi { + packets: marked_tables(1, 1), + flagged: Some(marked_tables(2, 1)), + clock: 1, + }); + for ms in [10, 20] { + schedule.lead(Tables { + rate, + added: Vec::new(), + }); + schedule.push(1, timing(ms), unit(1, 1), true).unwrap(); + } + let mut marks = Vec::new(); + while let Some(slot) = schedule.next(None).unwrap() { + marks.extend(table_marks(&slot)); + } + assert_eq!(marks, [2, 2, 1, 1], "{rate:?}"); + } + } + /// Packets held for the opening that take more than one slot to go out keep SI muxed after /// them held behind them until the last has, padded or not. #[test] @@ -1898,17 +2033,13 @@ mod tests { let mut schedule = Schedule::new(Duration::ZERO); schedule.set_rate(rate); schedule.set_tables(100); + set_psi(&mut schedule); schedule.set_clock(1); schedule.set_buffer(1, VIDEO); schedule.set_buffer(0x11, Buffer::SYSTEM); schedule.open(); schedule.push_clocked(unit(0x11, 100)); - schedule.lead(Tables { - packets: [unit(0, 1), unit(100, 1)].concat(), - clock: 1, - rate, - added: vec![1], - }); + schedule.lead(Tables { rate, added: vec![1] }); schedule.push(1, timing(10), unit(1, 3), true).unwrap(); let mut sent = 0; let mut slots = Vec::new(); @@ -3087,6 +3218,7 @@ mod tests { clocked: false, clock: None, tables: 0, + begun: 0, }; let mut null = unit(0, 1); (null[1], null[2]) = (0x1f, 0xff); @@ -3127,6 +3259,7 @@ mod tests { clocked: false, clock: None, tables: 0, + begun: 0, }; let mut null = unit(0, 1); (null[1], null[2]) = (0x1f, 0xff);