diff --git a/quest/m1/ts-reliability/README.md b/quest/m1/ts-reliability/README.md index d2a69c1f4b2..04d1b19d923 100644 --- a/quest/m1/ts-reliability/README.md +++ b/quest/m1/ts-reliability/README.md @@ -100,7 +100,6 @@ catalog linger (#5305). - [PAT changes](/quest/m1/ts-reliability/pat-lifecycle.md) - a running export follows a moved PMT PID - [TS workflow passes](/quest/m1/ts-reliability/validation-green.md) - the scheduled TS lanes pass on main again, with no lane loosened -- [Export SI order](/quest/m1/ts-reliability/export-si-order.md) - a TS export sends SI and media after its PMT and a PCR interval, so a hop keeps first emissions - [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 ## Related diff --git a/quest/m1/ts-reliability/export-si-order.md b/quest/m1/ts-reliability/export-si-order.md deleted file mode 100644 index a5a2026e83f..00000000000 --- a/quest/m1/ts-reliability/export-si-order.md +++ /dev/null @@ -1,27 +0,0 @@ -# [S] TS export leads with its PMT and PCR - -## Goal - -A TS export holds its first SI tables and media until the PMT and one PCR -interval are out, so an export-to-import hop keeps each table's first emission instead of -dropping it until the table repeats. - -## Plan - -Since [moq-dev/moq#6757](https://github.com/moq-dev/moq/pull/6757), TS import -admits SI only once a valid program-clock interval exists. The export writes -its first SI ahead of its PMT and PCRs, so the importer drops it. Order the -export's first output as PAT, PMT, a PCR interval, then SI. A PCR interval -here is what `si.rs` needs to place a section: one forward PCR step of at most -`MAX_INTERVAL`, not the importer's two-second catalog window. Test it with an -export-to-import round trip that sees every SI table's first emission. - -The first media PES also waits for that PCR, so a reader of our own export -never sees media it cannot place. The round trip also checks that the first -PES follows the PCR and is imported. - -Not planned (2026-10-25, maintainer): bounding SI aging against a source that -claims a huge mux rate through valid PCR steps. That needs a program-clock -floor per input byte, a new contract. - -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 34a95ef316b..0a218af1422 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -103,7 +103,9 @@ fn sanitize_mux_rate(rate: u64) -> Option { /// Use [`next`](Self::next) to pull one [`Frame`] per PCR grid slot: its `payload` /// is the TS packets belonging to that slot, stamped at the slot's boundary. The /// leading PAT/PMT rides on the first frame (so it inherits a real timestamp), and -/// is re-emitted at video keyframes and periodically for mid-stream tune-in. +/// is re-emitted at video keyframes and periodically for mid-stream tune-in. SI and +/// whole PES or sections wait until a PCR interval has followed it, since a reader +/// places them on the program clock and the PMT is what names it. /// Returns `None` when the broadcast ends. /// /// A rendition the catalog moves to another track (a new `track` or `broadcast`) goes on out @@ -2759,6 +2761,11 @@ where self.program.schedule.set_tables(pmt_pid); self.program.schedule.set_clock(pcr_pid); + // A reader of the first tables has no program clock yet. Tables that take over from + // an earlier program's flag a break in the clock it already reads instead. + if !self.program.pcr_discontinuity { + self.program.schedule.open(); + } for track in self.program.tracks.values() { self.program.liveness.register(track.pid); } @@ -3139,6 +3146,12 @@ where self.write_section(&mut out, pid, section)?; } } + // A reader drops SI it cannot place on the program clock, so while the opening's PCR + // interval may still be ahead of the unit this frame goes in, the schedule holds the SI + // for it rather than the media it would lead. + if self.program.schedule.clocking() { + self.program.schedule.push_clocked(std::mem::take(&mut out)); + } // The tables lead the unit that carries this frame, behind whatever went before it. if kind.placed() { diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index 65e1d0bb3d0..a01c4d0d558 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -126,11 +126,6 @@ where (out, exporter.stats()) } -/// An adaptation-field-only single-packet frame: the exporter's PCR carriage. -fn is_pcr_frame(frame: &Frame) -> bool { - frame.payload.len() == 188 && frame.payload[3] & 0x30 == 0x20 -} - fn assert_packet_aligned(ts: &[u8]) { assert!(!ts.is_empty(), "no TS output"); assert_eq!(ts.len() % 188, 0, "output not a whole number of 188-byte packets"); @@ -3504,6 +3499,226 @@ async fn service_layer_survives_roundtrip() { assert_eq!(program2.pmt_pid, program.pmt_pid, "PMT PID survived"); } +/// An export leads with its PAT and PMT, then a PCR interval on the clock they name, before +/// its SI: a re-import places every SI table's first emission instead of dropping it until it +/// repeats, padded or not, and every PES with it. +#[tokio::test(start_paused = true)] +async fn a_reimport_keeps_the_first_si() { + let data = include_bytes!("test_data/bbb.ts"); + let nit = make_long_section(0x40, 0x1234, 1, 0, 0, &[0xf0, 0x00]); + // Past bbb's PCRs, where its import keeps SI: bbb's own SDT precedes them. + let mut input = data.to_vec(); + input.extend_from_slice(&si_packet(0x0010, &nit)); + let mut sdt = data[..188].to_vec(); + sdt[3] = (sdt[3] & 0xf0) | ((sdt[3] + 1) & 0x0f); + input.extend_from_slice(&sdt); + let sdt = ts_sections(&sdt, 0x0011); + + // Each export reads an import of its own. + for rate in [None, Some(5_000_000)] { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let catalog = crate::catalog::Producer::new( + &mut broadcast, + crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::::default()), + ) + .unwrap(); + let mut import = crate::container::ts::Import::new(broadcast, catalog.reserve()); + import.decode(&BytesMut::from(&input[..])).unwrap(); + import.finish().unwrap(); + let units = |stats: super::stats::Snapshot| { + stats + .streams + .iter() + .filter(|(_, stream)| stream.units > 0) + .map(|(pid, stream)| (*pid, stream.units)) + .collect::>() + }; + let imported = units(import.stats()); + + let mut export = open_ts( + crate::source::announced(&consumer).await, + crate::catalog::CatalogFormat::Hang, + ) + .await + .unwrap(); + if let Some(rate) = rate { + export = export.with_mux_rate(rate); + } + let ts = drain_with(export).await; + assert_packet_aligned(&ts); + + // No SI goes out before the PMT and two PCRs after it. + let packets = ts.as_chunks::<188>().0; + let pid = |packet: &[u8; 188]| u16::from(packet[1] & 0x1f) << 8 | u16::from(packet[2]); + let pmt = packets.iter().position(|packet| pid(packet) == 0x1000).expect("a PMT"); + let clock = packets + .iter() + .enumerate() + .skip(pmt) + .filter(|(_, packet)| super::import::pcr(packet).is_some()) + .nth(1) + .expect("two PCRs after the PMT") + .0; + let first = packets + .iter() + .position(|packet| [0x0010, 0x0011].contains(&pid(packet))) + .expect("SI"); + assert!( + first > clock, + "{rate:?}: SI at packet {first}, the PMT at {pmt} and its second PCR at {clock}" + ); + + let mut broadcast2 = moq_net::broadcast::Info::new().produce(); + let consumer2 = broadcast2.consume(); + let catalog2 = crate::catalog::Producer::new( + &mut broadcast2, + crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::::default()), + ) + .unwrap(); + let mut import2 = crate::container::ts::Import::new(broadcast2, catalog2.reserve()); + import2.decode(&BytesMut::from(ts.as_ref())).unwrap(); + import2.finish().unwrap(); + let stats = import2.stats(); + for (pid, stream) in &stats.streams { + assert_eq!( + stream.unplaced, 0, + "{rate:?}: PID {pid:#x} dropped what it could not place" + ); + } + assert_eq!(units(stats), imported, "{rate:?}: every PES"); + + let si = catalog2.snapshot().ext.mpegts.si; + assert_eq!( + read_si_sections(&consumer2, &si[&0x0010][&0x40].track).await, + vec![Bytes::from(nit.clone())], + "{rate:?}: the NIT" + ); + assert_eq!( + read_si_sections(&consumer2, &si[&0x0011][&0x42].track).await, + sdt, + "{rate:?}: the SDT" + ); + } +} + +/// A gap longer than a skip after the first frame still leaves the opening's PCR interval +/// whole: the first SI follows two PCRs a reader can measure, and a re-import places it +/// instead of dropping it until it repeats, padded or not. +#[tokio::test(start_paused = true)] +async fn a_reimport_keeps_the_first_si_across_a_gap() { + for rate in [None, Some(5_000_000)] { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let mut catalog = crate::catalog::Producer::new( + &mut broadcast, + crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::::default()), + ) + .unwrap(); + + let avcc = crate::codec::h264::build_avcc(&[Bytes::from_static(SPS)], &[Bytes::from_static(PPS)]).unwrap(); + let track = broadcast + .create_track( + broadcast.unique_name(".avc1"), + hang::container::track_info(hang::catalog::PRIORITY.video), + ) + .unwrap(); + let name = track.name().to_string(); + let mut sdt_track = broadcast.create_track("0x0011-0x42.si", timed()).unwrap(); + sdt_track + .write_frame( + Timestamp::ZERO, + Bytes::from(make_long_section(0x42, 1, 0, 0, 0, &[0xaa; 8])), + ) + .unwrap(); + { + let mut guard = catalog.modify().unwrap(); + let mut cfg = VideoConfig::new(H264 { + profile: 0x64, + constraints: 0, + level: 0x1f, + inline: false, + }); + cfg.container = Container::Legacy; + cfg.description = Some(avcc); + guard.video.renditions.insert(name.clone(), cfg); + guard.ext.mpegts.si.entry(0x0011).or_default().insert( + 0x42, + tscat::SiEntry { + track: "0x0011-0x42.si".to_string(), + interval: Some(Duration::from_secs(2)), + ..Default::default() + }, + ); + } + + // Five seconds between frames, past the second a skip keeps before what comes next. + let mut producer = Producer::new(track, HangContainer::Legacy(crate::container::Kind::Data)); + for (sec, slice) in [(0, 0x65u8), (5, 0x41)] { + producer + .write(Frame { + timestamp: Timestamp::from_micros(sec * 1_000_000).unwrap(), + duration: None, + payload: length_prefixed(&[&[slice; 300]]), + keyframe: sec == 0, + }) + .unwrap(); + } + producer.finish().unwrap(); + + // No delay: each frame settles the slots before it, so the gap is skipped at once. + let mut export = open_ts( + crate::source::announced(&consumer).await, + crate::catalog::CatalogFormat::Hang, + ) + .await + .unwrap() + .with_replay(); + if let Some(rate) = rate { + export = export.with_mux_rate(rate); + } + let mut ts = BytesMut::new(); + while let Ok(frame) = tokio::time::timeout(DRAIN, export.next()).await { + let Some(frame) = frame.expect("exporter error") else { + break; + }; + ts.extend_from_slice(&frame.payload); + } + assert_packet_aligned(&ts); + + // The two PCRs between the PMT and the first SI step forward by at most a valid interval. + let packets = ts.as_chunks::<188>().0; + let pid = |packet: &[u8; 188]| u16::from(packet[1] & 0x1f) << 8 | u16::from(packet[2]); + let pmt = packets.iter().position(|packet| pid(packet) == 0x1000).expect("a PMT"); + let first = packets.iter().position(|packet| pid(packet) == 0x0011).expect("SI"); + let pcrs: Vec = packets[pmt..first].iter().filter_map(super::import::pcr).collect(); + let [.., before, last] = pcrs[..] else { + panic!("{rate:?}: {} PCRs between the PMT and the first SI", pcrs.len()); + }; + let step = (last + super::mux_rate::PCR_WRAP - before) % super::mux_rate::PCR_WRAP; + assert!( + (1..=super::mux_rate::MAX_INTERVAL).contains(&step), + "{rate:?}: the PCR steps {before} to {last} before the first SI" + ); + + let mut broadcast2 = moq_net::broadcast::Info::new().produce(); + let catalog2 = crate::catalog::Producer::new( + &mut broadcast2, + crate::catalog::Config::default().with_catalog(crate::catalog::hang::Catalog::::default()), + ) + .unwrap(); + let mut import = crate::container::ts::Import::new(broadcast2, catalog2.reserve()); + import.decode(&BytesMut::from(ts.as_ref())).unwrap(); + import.finish().unwrap(); + for (pid, stream) in &import.stats().streams { + assert_eq!( + stream.unplaced, 0, + "{rate:?}: PID {pid:#x} dropped what it could not place" + ); + } + } +} + /// Each SI PID must be re-emitted on its own interval, independently of the PSI cadence /// and of video keyframes. The fixtures are a fraction of a second long, so nothing else /// distinguishes a correct interval from "emitted once and never again": this builds a @@ -4769,9 +4984,7 @@ async fn a_selected_programs_si_survives_export() { ); } -/// The sections `ts` carries on `pid`, each whole in one packet. Read off the wire rather -/// than by a re-import: SI an export leads with precedes its PCRs, so an import drops it -/// until it repeats. +/// The sections `ts` carries on `pid`, each whole in one packet, read off the wire. fn ts_sections(ts: &[u8], pid: u16) -> Vec { ts.chunks(188) .filter(|p| (u16::from(p[1] & 0x1f) << 8 | u16::from(p[2])) == pid && p[1] & 0x40 != 0) @@ -5576,12 +5789,17 @@ async fn late_join_matches_a_running_exporter_without_video() { let (a, b) = export_twice(false).await; // The joiner's first frame leads with PAT/PMT, while the running exporter has no - // reason to repeat them there. It can spread over several PCR slices, so compare - // from the next PAT/PMT refresh, which both exporters anchor to the media grid. + // reason to repeat them there, and its SDT waits for a PCR interval after them. They can + // spread over several PCR slices, so compare from the next PAT/PMT refresh past that SDT, + // which both exporters anchor to the media grid. + let sdt = b + .iter() + .find(|frame| count_pid(std::slice::from_ref(*frame), 0x0011) > 0) + .expect("the joiner's SDT") + .timestamp; let from = b .iter() - .filter(|frame| count_pid(std::slice::from_ref(*frame), 0) > 0) - .nth(1) + .find(|frame| frame.timestamp > sdt && count_pid(std::slice::from_ref(*frame), 0) > 0) .expect("a periodic PAT/PMT refresh") .timestamp; assert_only_continuity_differs(&a, &b, from); @@ -6404,18 +6622,18 @@ async fn si_revision_after_final_media_frame_is_flushed() { ) .await .unwrap(); - let first = loop { + // v1 goes out with the media frame's slices, once the export's PCR interval is out. + let contains = |haystack: &[u8], needle: &[u8]| haystack.windows(needle.len()).any(|w| w == needle); + loop { let frame = tokio::time::timeout(DRAIN, exporter.next()) .await - .expect("the only media frame") + .expect("v1 goes out (control)") .unwrap() - .unwrap(); - if !is_pcr_frame(&frame) { - break frame; + .expect("v1 goes out (control)"); + if contains(&frame.payload, &sdt_v1) { + break; } - }; - let contains = |haystack: &[u8], needle: &[u8]| haystack.windows(needle.len()).any(|w| w == needle); - assert!(contains(&first.payload, &sdt_v1), "v1 rode the media frame (control)"); + } // The revision lands after the last media frame; only the trailing flush can // carry it. Closing the catalog lets the exporter reach end of stream. diff --git a/rs/moq-mux/src/container/ts/schedule.rs b/rs/moq-mux/src/container/ts/schedule.rs index 1976088c71e..b091c3cb0a7 100644 --- a/rs/moq-mux/src/container/ts/schedule.rs +++ b/rs/moq-mux/src/container/ts/schedule.rs @@ -30,6 +30,12 @@ const PAYLOAD: usize = TsPacket::SIZE - 4; const BACKFILL: u128 = 40; /// The longest a transport buffer may go without emptying (ISO 13818-1 2.4.2.3). const TB_EMPTY: Duration = Duration::from_secs(1); +/// How many slots after the one its first program tables finish in a timeline that +/// [opens](Schedule::open) holds what a reader places on the program clock. The reader learns +/// the clock PID from the PMT, so it times nothing by the PCR opening their slot; the next +/// slot's PCR is the first it reads, and the one after closes the first interval, which it +/// needs to place a section or whole PES. +const OPENING: u128 = 2; /// One tick of the system clock, in whole nanoseconds rounded up. const TICK: i128 = 1_000_000_000_u128.div_ceil(SYSTEM_CLOCK) as i128; /// The system clock in Hz, the unit of a PCR. @@ -589,6 +595,17 @@ pub(super) struct Schedule { seq: u64, /// 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. + opening: bool, + /// The first slot a placed unit may go out in: [`OPENING`] slots after the one the first + /// tables finish in, on a timeline that opens. + clocked: u128, + /// Packets a reader places on the program clock, held until the opening's PCR interval is + /// out ([`Self::push_clocked`]). + unclocked: Vec, + /// The unit held packets last went out in, by `seq`: what is muxed after them waits until it + /// has finished, as it rides a lane of its own. + deferred: Option, } impl Schedule { @@ -610,9 +627,41 @@ impl Schedule { ended: false, seq: 0, lead: None, + opening: false, + clocked: 0, + unclocked: Vec::new(), + deferred: None, } } + /// Open the timeline for a reader that has no program clock yet: no placed unit goes out + /// until a PCR interval has followed the first tables ([`OPENING`]), so the reader can + /// place every one. A timeline that continues a clock the reader already has needs none. + pub fn open(&mut self) { + self.opening = true; + } + + /// Whether a unit pushed now could go out before the opening's PCR interval is out, or + /// ahead of packets still held for it. + pub fn clocking(&self) -> bool { + self.opening + || !self.unclocked.is_empty() + || self.deferring() + || self.next.is_some_and(|next| next < self.clocked) + } + + /// Whether the unit held packets last went out in has some still to send. + fn deferring(&self) -> bool { + self.deferred + .is_some_and(|seq| self.units.iter().any(|unit| unit.seq == seq)) + } + + /// Hold `packets`, which a reader places on the program clock and which ride no unit of their + /// own, until the opening's PCR interval is out: they then go out first, as soon as they can. + pub fn push_clocked(&mut self, packets: Vec) { + self.unclocked.extend(packets); + } + /// Whether every source has ended, so what is queued still goes out, late if it must. /// A source that resumes (a same-instance return) holds its deadlines again. pub fn set_ended(&mut self, ended: bool) { @@ -894,25 +943,31 @@ impl Schedule { held } - /// Put in force what the tables of unit `i` carry, as their last packet goes out, if it is one. - fn enforce(&mut self, i: usize) -> bool { + /// Put in force what the tables of unit `i` carry, as their last packet goes out in slot + /// `index`, if it is one. + fn enforce(&mut self, i: usize, index: u128) -> bool { let Some(program) = &self.units[i].tables else { return false; }; self.clock = Some(program.clock); self.rate = program.rate; + if std::mem::take(&mut self.opening) { + self.clocked = self.clocked.max(index + OPENING); + } true } - /// Whether a unit is still to go out. + /// Whether a unit, or packets held for the opening's clock, are still to go out. #[cfg(test)] pub fn queued(&self) -> bool { - !self.units.is_empty() + !self.units.is_empty() || !self.unclocked.is_empty() } /// Whether nothing is left to lay out: every unit sent, and the clock past its decode time. pub fn is_empty(&self) -> bool { + // Packets held for the clock count once a slot has gone out, so slots run on to it. self.units.is_empty() + && (self.unclocked.is_empty() || self.next.is_none()) && self .horizon .is_none_or(|horizon| self.next.is_some_and(|next| next > horizon)) @@ -930,6 +985,11 @@ impl Schedule { self.last = None; self.ended = false; self.lead = None; + // An opening still pending stays: its tables never went out, so the reader has no clock + // for the next timeline to continue. + self.clocked = 0; + self.unclocked.clear(); + self.deferred = None; } /// The first slot `unit` may go out in: [`ahead`] of its due slot, and never before the @@ -939,13 +999,15 @@ impl Schedule { /// multiplex rate the one whose span holds it, where [`Self::padded`] times each packet, /// and otherwise the first that opens at or after it, as nothing times the packets /// within an unpadded slot. Slot `index` opens one slot behind its boundary ([`grid_pcr`]). + /// On a timeline that [opens](Self::open), it waits for the opening's PCR interval too. fn release(&self, unit: &Unit) -> u128 { if let Some((place, _)) = unit.place { let slot = PCR_INTERVAL.as_nanos(); - return match self.rate { + let release = match self.rate { Some(_) => place / slot + 1, None => place.div_ceil(slot) + 1, }; + return release.max(self.clocked); } if unit.late { return 0; @@ -963,7 +1025,12 @@ impl Schedule { (None, Some(first)) => Some(first), // Skip the empty slots of a long gap, keeping the second before what comes next. (Some(next), first) => { - let target = first.or(known); + let mut target = first.or(known); + // The opening's PCR interval runs slot by slot to its gate, so what it holds goes + // out on a clock the reader has measured. + if self.opening || !self.unclocked.is_empty() { + target = target.map(|target| target.min(self.clocked)); + } Some(next.max(target.map_or(0, |target| target.saturating_sub(BACKFILL)))) } } @@ -991,6 +1058,22 @@ impl Schedule { if known.is_some_and(|known| index >= known) { return Ok(None); } + // The first tables finish in the first slot unless their buffers hold them back, and a + // placed unit beside them there would go out ahead of its clock. + if self.opening && self.next.is_none() { + self.clocked = index + OPENING; + } + // Those held behind earlier ones wait for them to finish. + if !self.opening && index >= self.clocked && !self.unclocked.is_empty() && !self.deferring() { + let packets = std::mem::take(&mut self.unclocked); + let by = (index + 1) * PCR_INTERVAL.as_nanos(); + // They follow no unit, so the next one pushed keeps its own start. + let last = self.last; + let unit = self.enqueue(pid(&packets), Timing { by, decode: by }, packets, None); + unit.late = true; + self.deferred = Some(unit.seq); + self.last = last; + } self.next = Some(index + 1); if let Some(rate) = self.rate { return self.padded(index, rate).map(Some); @@ -1044,7 +1127,7 @@ impl Schedule { unit.sent += take; } let tables = (0..self.units.len()) - .filter(|&i| self.units[i].remaining() == 0 && self.enforce(i)) + .filter(|&i| self.units[i].remaining() == 0 && self.enforce(i, index)) .count(); for unit in self.units.iter().filter(|unit| unit.remaining() == 0) { self.decoding.extend(unit.leaving()); @@ -1172,7 +1255,9 @@ impl Schedule { while let Some(Reverse((by, lane))) = ready.pop() { let i = *lanes[lane].front().unwrap(); let unit = &self.units[i]; - if holding.get(&unit.pid).is_some_and(|&seq| seq < unit.seq) { + // First tables that finish past the first slot hold back what was released beside + // them. + if holding.get(&unit.pid).is_some_and(|&seq| seq < unit.seq) || self.release(unit) > index { passed.push(Reverse((by, lane))); continue; } @@ -1357,7 +1442,7 @@ impl Schedule { } if unit.remaining() == 0 { lanes[lane].pop_front(); - if self.enforce(i) { + if self.enforce(i, index) { out.tables += 1; holding = self.holding(|i| self.units[i].remaining()); } @@ -1681,6 +1766,175 @@ mod tests { assert_eq!(tables, 2); } + /// A timeline that opens holds a placed unit until a PCR interval has followed its first + /// tables, while the unit they lead goes out where it would have without the opening, padded + /// or not. + #[test] + fn an_opening_holds_placed_units_for_a_pcr_interval() { + for rate in [None, Some(RATE)] { + let first = |open: bool| { + let mut schedule = Schedule::new(Duration::ZERO); + schedule.set_rate(rate); + schedule.set_tables(100); + schedule.set_clock(1); + schedule.set_buffer(1, VIDEO); + if open { + schedule.open(); + } + assert_eq!(schedule.clocking(), open); + 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.push(1, timing(10), unit(1, 3), true).unwrap(); + let slots = drain(&mut schedule); + assert!(!schedule.clocking()); + let on = |pid: u16| { + slots + .iter() + .find(|(_, pids, _)| pids.contains_key(&pid)) + .map(|&(index, ..)| index) + }; + (on(100).unwrap(), on(1).unwrap(), on(7).unwrap()) + }; + let (tables, media, placed) = first(true); + assert_eq!(placed, tables + OPENING, "{rate:?}"); + let (_, before, early) = first(false); + assert_eq!(media, before, "{rate:?}: the media keeps its slot"); + assert!(early < placed, "{rate:?}: without the opening it goes in slot {early}"); + } + } + + /// Packets held for the opening's clock go out once its PCR interval is, without holding + /// back the unit they were muxed ahead of, padded or not. + #[test] + fn clocked_packets_follow_the_opening_without_the_media() { + 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_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.push(1, timing(10), unit(1, 3), true).unwrap(); + assert!(schedule.clocking()); + let slots = drain(&mut schedule); + assert!(!schedule.clocking()); + let on = |pid: u16| { + slots + .iter() + .find(|(_, pids, _)| pids.contains_key(&pid)) + .map(|&(index, ..)| index) + }; + assert_eq!((on(100), on(1)), (Some(0), Some(0)), "{rate:?}"); + assert_eq!(on(0x11), Some(OPENING), "{rate:?}"); + } + } + + /// A gap after the first frames skips no slot of the opening's PCR interval, so packets held + /// for it go out after consecutive PCRs rather than after a jump the reader cannot measure, + /// padded or not. Until they do, SI muxed after them is held behind them. + #[test] + fn clocked_packets_follow_the_opening_across_a_gap() { + 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_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.push(1, timing(10), unit(1, 1), true).unwrap(); + // The next frame is ten seconds out, far past a skip. + let known = Some(400); + let mut indexes = Vec::new(); + let mut held = None; + while let Some(slot) = schedule.next(known).unwrap() { + indexes.push(slot.index); + if slot + .packets + .as_chunks::<{ TsPacket::SIZE }>() + .0 + .iter() + .any(|packet| pid(packet) == 0x11) + { + held = Some(slot.index); + break; + } + // SI muxed now goes behind what is held for the same PID, never ahead of it. + assert!(schedule.clocking(), "{rate:?}: after slot {}", slot.index); + } + assert_eq!(held, Some(OPENING), "{rate:?}: slots {indexes:?}"); + assert_eq!(indexes, (0..=OPENING).collect::>(), "{rate:?}"); + assert!(!schedule.clocking()); + } + } + + /// 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] + fn clocked_packets_stay_ahead_until_they_finish() { + 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_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.push(1, timing(10), unit(1, 3), true).unwrap(); + let mut sent = 0; + let mut slots = Vec::new(); + while let Some(slot) = schedule.next(None).unwrap() { + let packets = slot.packets.as_chunks::<{ TsPacket::SIZE }>().0; + let held = packets.iter().filter(|&packet| pid(packet) == 0x11).count(); + if held > 0 { + slots.push(slot.index); + } + sent += held; + if sent < 100 { + assert!( + schedule.clocking(), + "{rate:?}: {sent} of 100 out by slot {}", + slot.index + ); + } + } + assert_eq!(sent, 100, "{rate:?}"); + assert!(!schedule.clocking(), "{rate:?}"); + if rate.is_some() { + assert!(slots.len() > 1, "the held packets span slots {slots:?}"); + } + } + } + /// The pointer byte goes just before placement, so it cannot cost a section another packet. #[test] fn placed_first_byte_accounts_for_the_section_pointer() {