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);