diff --git a/src/mux/codec/adts.rs b/src/mux/codec/adts.rs index 468fb6d..30ed307 100644 --- a/src/mux/codec/adts.rs +++ b/src/mux/codec/adts.rs @@ -112,10 +112,14 @@ impl CodecParser for AdtsParser { } self.tally.record_kept(); + // One PES is one unit here, so the unit's first byte is in THIS packet + // and its facts are this packet's -- the same rule the buffering + // parsers apply through `PesBuf::front`, with nothing carried over. + let facts = super::pesbuf::PesFacts::of(pes); vec![Frame { - discontinuity: pes.discontinuity, + discontinuity: facts.discontinuity, coding: None, - source: None, + source: facts.source, pts_ns, keyframe: true, data: pes.data.clone(), diff --git a/src/mux/codec/dvdsub.rs b/src/mux/codec/dvdsub.rs index c96c838..441ec76 100644 --- a/src/mux/codec/dvdsub.rs +++ b/src/mux/codec/dvdsub.rs @@ -25,7 +25,10 @@ pub struct DvdSubParser { /// Pre-formatted VobSub .idx palette header for codec_private. codec_data: Option>, /// In-progress SPU reassembly: (head PTS in ns, declared SPU_size, bytes). - pending: Option<(i64, usize, Vec)>, + /// The SPU being accumulated, with the facts of the PES that STARTED it. + /// An SPU spans PES packets, so its timestamp and source offset are the + /// opening packet's — the same rule every other buffering parser applies. + pending: Option<(super::pesbuf::PesFacts, usize, Vec)>, } impl DvdSubParser { @@ -41,11 +44,12 @@ impl DvdSubParser { fn take_if_complete(&mut self, force: bool) -> Option { let (_, size, buf) = self.pending.as_ref()?; if force || buf.len() >= *size { - let (pts_ns, _, data) = self.pending.take().unwrap(); + let (facts, _, data) = self.pending.take().unwrap(); + let pts_ns = facts.presentation_ns().unwrap_or(0); return Some(Frame { discontinuity: false, coding: None, - source: None, + source: facts.source, pts_ns, keyframe: true, data, @@ -106,7 +110,7 @@ impl CodecParser for DvdSubParser { out.push(Frame { discontinuity: false, coding: None, - source: None, + source: super::pesbuf::PesFacts::of(pes).source, pts_ns, keyframe: true, data: pes.data.clone(), @@ -120,7 +124,7 @@ impl CodecParser for DvdSubParser { out.push(Frame { discontinuity: false, coding: None, - source: None, + source: super::pesbuf::PesFacts::of(pes).source, pts_ns, keyframe: true, data: pes.data.clone(), @@ -133,7 +137,7 @@ impl CodecParser for DvdSubParser { if buf.len() > MAX_SPU_BYTES { buf.truncate(MAX_SPU_BYTES); } - self.pending = Some((pts_ns, declared, buf)); + self.pending = Some((super::pesbuf::PesFacts::of(pes), declared, buf)); if let Some(frame) = self.take_if_complete(false) { out.push(frame); } @@ -671,7 +675,17 @@ mod tests { let f = parser.parse(&make_pes(vec![0x00, 0x10, 0xAA], None)); assert!(f.is_empty(), "incomplete sized segment held, not emitted"); assert!(parser.pending.is_some(), "started a new pending SPU"); - assert_eq!(parser.pending.as_ref().unwrap().0, 0, "pts 0 (no PTS)"); + assert_eq!( + parser + .pending + .as_ref() + .unwrap() + .0 + .presentation_ns() + .unwrap_or(0), + 0, + "pts 0 (no PTS)" + ); } #[test] diff --git a/src/mux/codec/mod.rs b/src/mux/codec/mod.rs index 321a2ed..1b9a3c6 100644 --- a/src/mux/codec/mod.rs +++ b/src/mux/codec/mod.rs @@ -34,6 +34,11 @@ pub mod mpeg2; pub mod mpegaudio; /// HDMV PGS (Presentation Graphics Stream) subtitle parser. pub mod pgs; + +/// One accumulation buffer for parsers that assemble access units across PES +/// packets, so a unit's timestamp and its source offset always come from the +/// packet that carried its first byte -- and from the SAME packet. +pub(crate) mod pesbuf; /// Display-order PTS reconstruction for sparse-PTS program-stream video. pub(crate) mod reorder; /// Shared MPEG/Annex-B start-code scanning helpers. diff --git a/src/mux/codec/pesbuf.rs b/src/mux/codec/pesbuf.rs new file mode 100644 index 0000000..32e153a --- /dev/null +++ b/src/mux/codec/pesbuf.rs @@ -0,0 +1,295 @@ +//! One accumulation buffer for parsers that assemble access units across PES +//! packets. +//! +//! A buffering parser has to answer the same question for every unit it emits: +//! *which PES contributed this unit's FIRST byte?* Its timestamp comes from +//! that PES, and so does the source byte offset that identifies which clip of a +//! multi-clip title the unit belongs to. The trailing PES packets that complete +//! the unit carry their own, later, values which must not override it. +//! +//! That question was answered three different ways. DTS kept a deque of +//! `(offset, pts)` markers and took the one covering offset 0 — correct. AC-3 +//! kept a single carry-over timestamp. TrueHD kept its own. None of them +//! carried the source offset at all, so provenance existed only for video, and +//! a title whose clip marks could not be read from timestamps alone had nine +//! audio and subtitle tracks with nothing to place them by. +//! +//! Three spellings of one rule is how they drifted, so this is the one place it +//! lives. The buffer owns the bytes AND the marks, and returns a PES's facts +//! together — a parser cannot take the timestamp from one PES and the source +//! from another, because it does not assemble them itself. + +use super::PesPacket; +use super::pts_to_ns; +use crate::pes::SourcePos; + +/// What a PES contributes to the bytes it carried. +/// +/// Returned as a unit so a caller cannot mix fields from different packets. +/// The timestamps are carried RAW, as the packet had them. This type answers +/// *which packet* a unit's facts come from — the question that was being +/// answered three different ways. How a given codec derives a timestamp from +/// that packet stays the codec's business: DVD subtitles read `pts` only and +/// fall back to 0, most audio takes `pts.or(dts)`. Deriving it here would have +/// changed those semantics silently while fixing provenance. +#[derive(Debug, Clone, Copy, Default, PartialEq)] +pub(crate) struct PesFacts { + /// Presentation timestamp in 90kHz ticks, as carried. + pub pts: Option, + /// Decode timestamp in 90kHz ticks, as carried. + pub dts: Option, + /// Byte offset of this PES's first ES byte within the title's feed — what + /// identifies the clip a frame came from. `None` when the demuxer was fed + /// without a base offset. + pub source: Option, + /// Packets for this stream were lost before this PES. + pub discontinuity: bool, +} + +impl PesFacts { + /// The facts a PES packet carries, read straight off it. + /// + /// Every parser reads its frame's timestamp, source and discontinuity from + /// a `PesFacts` — never off a `PesPacket` field directly — so the three can + /// never be taken from different packets. Parsers that assemble units + /// across packets get theirs from [`PesBuf::front`]; this is the same value + /// for a parser whose unit begins in the packet it is handed. + pub(crate) fn of(pes: &PesPacket) -> Self { + Self { + pts: pes.pts, + dts: pes.dts, + source: pes.source, + discontinuity: pes.discontinuity, + } + } + + /// This unit's presentation time in nanoseconds — the ONE derivation. + /// + /// PTS and DTS are not two spellings of one value: PTS is when to display, + /// DTS is when to decode, and for a stream that REORDERS (video carrying + /// B-frames) they differ, so reading DTS as a presentation time would be + /// wrong. Reordering is handled by the video path, which reconstructs + /// display order rather than calling this. + /// + /// For everything that reaches here — audio and subtitles — there is no + /// reordering, so DTS *is* the presentation time, and falling back to it is + /// reading the same value from whichever field the packet used. A fallback + /// for a missing field, not a second rule: dvdsub read `pts` alone and + /// returned 0 for a packet that carried only DTS. + pub(crate) fn presentation_ns(&self) -> Option { + self.pts.or(self.dts).map(pts_to_ns) + } +} + +/// Bytes accumulated across PES packets, each byte attributable to the packet +/// that carried it. +pub(crate) struct PesBuf { + buf: Vec, + /// `(offset of this PES's first byte within `buf`, its facts)`, ascending. + /// Offsets are relative to the current front and are rebased on `drain`. + marks: std::collections::VecDeque<(usize, PesFacts)>, +} + +impl PesBuf { + pub(crate) fn with_capacity(n: usize) -> Self { + Self { + buf: Vec::with_capacity(n), + marks: std::collections::VecDeque::new(), + } + } + + /// Append a PES's payload, recording where its bytes begin. + /// + /// A PES carrying no payload records nothing: it contributed no byte, so it + /// can never be the answer to "which packet carried the byte at offset N", + /// and admitting a zero-length mark would let it shadow the packet that + /// actually did. + pub(crate) fn push(&mut self, pes: &PesPacket) { + if pes.data.is_empty() { + return; + } + self.marks.push_back((self.buf.len(), PesFacts::of(pes))); + self.buf.extend_from_slice(&pes.data); + } + + /// Append raw bytes attributed to the SAME PES as the bytes already at the + /// end of the buffer. For a parser that rewrites or re-frames payload + /// in-place rather than appending a packet verbatim. + pub(crate) fn push_bytes(&mut self, data: &[u8]) { + self.buf.extend_from_slice(data); + } + + /// The facts of the PES that carried the byte at `off`. + /// + /// The last mark at or before `off`: bytes belong to the most recent packet + /// that started at or before them. Defaults when the buffer holds bytes + /// that predate any mark (a parser that seeded it directly). + pub(crate) fn facts_at(&self, off: usize) -> PesFacts { + let mut found = PesFacts::default(); + for &(at, facts) in &self.marks { + if at > off { + break; + } + found = facts; + } + found + } + + /// The facts of the PES that carried the byte at the FRONT of the buffer — + /// the first byte of the access unit a parser is about to emit, which is + /// the whole point of this type. + pub(crate) fn front(&self) -> PesFacts { + self.facts_at(0) + } + + pub(crate) fn as_slice(&self) -> &[u8] { + &self.buf + } + + pub(crate) fn len(&self) -> usize { + self.buf.len() + } + + pub(crate) fn is_empty(&self) -> bool { + self.buf.is_empty() + } + + /// Consume `n` bytes from the front, rebasing the marks onto the new front. + /// + /// The mark covering the new front byte is RETAINED at offset 0 even though + /// its packet started earlier: those bytes are still that packet's. Dropping + /// it would attribute the remainder of a straddling unit to whichever packet + /// happened to start next — precisely the misattribution this type exists to + /// prevent, and it would land at a clip boundary, the one place it matters. + pub(crate) fn drain(&mut self, n: usize) { + let n = n.min(self.buf.len()); + if n == 0 { + return; + } + self.buf.drain(..n); + let covering = self.facts_at(n); + self.marks.retain(|&(at, _)| at > n); + for m in &mut self.marks { + m.0 -= n; + } + if self.marks.front().map(|&(at, _)| at) != Some(0) { + self.marks.push_front((0, covering)); + } + } + + pub(crate) fn clear(&mut self) { + self.buf.clear(); + self.marks.clear(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn pes(data: &[u8], pts: Option, byte: Option) -> PesPacket { + PesPacket { + pid: 0x1100, + pts, + dts: None, + data: data.to_vec(), + source: byte.map(SourcePos::at_byte), + discontinuity: false, + } + } + + /// The point of the type: a unit assembled from two packets is attributed + /// to the one that carried its FIRST byte, not the one that completed it. + #[test] + fn a_unit_spanning_two_packets_belongs_to_the_packet_it_started_in() { + let mut b = PesBuf::with_capacity(64); + b.push(&pes(&[1, 2, 3], Some(90_000), Some(1_000))); + b.push(&pes(&[4, 5, 6], Some(180_000), Some(2_000))); + let f = b.front(); + assert_eq!(f.presentation_ns(), Some(pts_to_ns(90_000))); + assert_eq!(f.source.unwrap().byte, 1_000, "the packet it STARTED in"); + } + + /// And the timestamp and the source come from the SAME packet — the + /// property that cannot hold when each is derived separately. + #[test] + fn timestamp_and_source_always_come_from_one_packet() { + let mut b = PesBuf::with_capacity(64); + b.push(&pes(&[1, 2], Some(90_000), Some(10))); + b.push(&pes(&[3, 4], Some(180_000), Some(20))); + b.push(&pes(&[5, 6], Some(270_000), Some(30))); + for off in 0..6usize { + let f = b.facts_at(off); + let want = match off { + 0 | 1 => (Some(pts_to_ns(90_000)), 10), + 2 | 3 => (Some(pts_to_ns(180_000)), 20), + _ => (Some(pts_to_ns(270_000)), 30), + }; + assert_eq!( + (f.presentation_ns(), f.source.unwrap().byte), + want, + "offset {off}" + ); + } + } + + /// Draining a completed unit must leave the REMAINDER attributed to the + /// packet that carried it, not to whichever packet starts next. + #[test] + fn draining_keeps_the_remainder_attributed_to_its_own_packet() { + let mut b = PesBuf::with_capacity(64); + b.push(&pes(&[1, 2, 3, 4], Some(90_000), Some(1_000))); + b.push(&pes(&[5, 6], Some(180_000), Some(2_000))); + // Emit a 2-byte unit: bytes 3 and 4 are still the FIRST packet's. + b.drain(2); + assert_eq!(b.front().source.unwrap().byte, 1_000); + // Emit those two: now the front is genuinely the second packet's. + b.drain(2); + assert_eq!(b.front().source.unwrap().byte, 2_000); + assert_eq!(b.front().presentation_ns(), Some(pts_to_ns(180_000))); + } + + /// A payload-less PES contributed no byte, so it must not shadow the packet + /// that did — otherwise the next unit takes a timestamp from a packet whose + /// bytes are not in it. + #[test] + fn an_empty_packet_does_not_claim_the_next_units_bytes() { + let mut b = PesBuf::with_capacity(64); + b.push(&pes(&[1, 2], Some(90_000), Some(1_000))); + b.push(&pes(&[], Some(180_000), Some(2_000))); + assert_eq!(b.front().source.unwrap().byte, 1_000); + assert_eq!(b.len(), 2, "an empty packet adds no bytes"); + } + + /// Draining everything and refilling must not resurrect a stale mark. + #[test] + fn a_fully_drained_buffer_takes_its_next_packets_facts() { + let mut b = PesBuf::with_capacity(64); + b.push(&pes(&[1, 2], Some(90_000), Some(1_000))); + b.drain(2); + assert!(b.is_empty()); + b.push(&pes(&[9], Some(450_000), Some(9_000))); + assert_eq!(b.front().source.unwrap().byte, 9_000); + assert_eq!(b.front().presentation_ns(), Some(pts_to_ns(450_000))); + } + + /// The two ways a parser can obtain facts must agree when the unit begins + /// in the packet just handed over — otherwise "one pattern" is two. + #[test] + fn a_unit_starting_in_this_packet_reads_the_same_either_way() { + let p = pes(&[1, 2, 3], Some(90_000), Some(4_242)); + let mut b = PesBuf::with_capacity(16); + b.push(&p); + assert_eq!(b.front(), PesFacts::of(&p)); + } + + /// Over-draining is clamped rather than panicking: a parser that + /// mis-sizes a unit must not take the process down. + #[test] + fn draining_past_the_end_is_clamped() { + let mut b = PesBuf::with_capacity(16); + b.push(&pes(&[1, 2], Some(90_000), Some(1))); + b.drain(99); + assert!(b.is_empty()); + } +} diff --git a/src/mux/codec/pgs.rs b/src/mux/codec/pgs.rs index 867a397..6e95d9b 100644 --- a/src/mux/codec/pgs.rs +++ b/src/mux/codec/pgs.rs @@ -214,7 +214,12 @@ impl ForcedTracker { /// Stateful parser that collapses PGS display/clear PCS pairs into /// duration-bearing Matroska frames. Implements [`CodecParser`]. pub struct PgsParser { - pending: Option<(i64, Vec)>, + /// The display set being accumulated, with the facts of the PES that + /// STARTED it. A set spans PES packets — it opens on a display PCS and + /// closes on the next one — so its timestamp and its source offset are the + /// opening packet's, never the closing packet's. Same rule the other + /// buffering parsers get from `PesBuf::front`. + pending: Option<(super::pesbuf::PesFacts, Vec)>, } impl Default for PgsParser { @@ -234,12 +239,13 @@ impl PgsParser { /// closes or replaces it), clamped to >= 0. Shared by the clear-PCS and /// replace-PCS arms so the Frame shape stays in one place. fn emit_pending(&mut self, end_pts_ns: i64) -> Option { - let (start_pts, data) = self.pending.take()?; + let (facts, data) = self.pending.take()?; + let start_pts = facts.presentation_ns().unwrap_or(0); let duration = end_pts_ns.saturating_sub(start_pts).max(0) as u64; Some(Frame { discontinuity: false, coding: None, - source: None, + source: facts.source, pts_ns: start_pts, keyframe: true, data, @@ -271,12 +277,12 @@ impl CodecParser for PgsParser { return self .pending .take() - .map(|(start_pts, data)| { + .map(|(facts, data)| { vec![Frame { discontinuity: false, coding: None, - source: None, - pts_ns: start_pts, + source: facts.source, + pts_ns: facts.presentation_ns().unwrap_or(0), keyframe: true, data, duration_ns: None, @@ -300,11 +306,11 @@ impl CodecParser for PgsParser { Some(0) => { let frame = match pts { Some(end) => self.emit_pending(end), - None => self.pending.take().map(|(start_pts, data)| Frame { + None => self.pending.take().map(|(facts, data)| Frame { discontinuity: false, coding: None, - source: None, - pts_ns: start_pts, + source: facts.source, + pts_ns: facts.presentation_ns().unwrap_or(0), keyframe: true, data, duration_ns: None, @@ -318,17 +324,24 @@ impl CodecParser for PgsParser { Some(_) => match pts { Some(start) => { out.extend(self.emit_pending(start)); - self.pending = Some((start, pes.data.clone())); + // The set's facts are THIS packet's — the one that opened + // it. `start` is that packet's PTS by construction. + self.pending = Some((super::pesbuf::PesFacts::of(pes), pes.data.clone())); + debug_assert_eq!( + super::pesbuf::PesFacts::of(pes).presentation_ns(), + Some(start), + "the opening packet's PTS is the set's start" + ); } // A display PCS with no PTS has an unknown start time. Don't // store it with a 0 sentinel (wrong start, absurd duration). // Flush any prior pending undurated and skip storing this one. None => { - out.extend(self.pending.take().map(|(start_pts, data)| Frame { + out.extend(self.pending.take().map(|(facts, data)| Frame { discontinuity: false, coding: None, - source: None, - pts_ns: start_pts, + source: facts.source, + pts_ns: facts.presentation_ns().unwrap_or(0), keyframe: true, data, duration_ns: None, @@ -379,11 +392,11 @@ impl CodecParser for PgsParser { // until end of file, which is exactly the desired behavior for // the final on-screen subtitle (see the module doc). match self.pending.take() { - Some((start_pts, data)) => vec![Frame { + Some((facts, data)) => vec![Frame { discontinuity: false, coding: None, - source: None, - pts_ns: start_pts, + source: facts.source, + pts_ns: facts.presentation_ns().unwrap_or(0), keyframe: true, data, duration_ns: None,