diff --git a/rs/libmoq/src/video.rs b/rs/libmoq/src/video.rs index 72f2bf26b..491a0d0c3 100644 --- a/rs/libmoq/src/video.rs +++ b/rs/libmoq/src/video.rs @@ -140,7 +140,9 @@ impl Video { // Flatten to CPU bytes outside the lock (a GPU frame downloads here), // then hold the lock only to buffer it; release before the callback. let frame = VideoFrame { - timestamp_us: frame.timestamp_us, + // The C ABI carries microseconds; the decoded frame's Timestamp is + // constrained to a QUIC VarInt, so the microsecond value fits a u64. + timestamp_us: frame.timestamp.as_micros() as u64, width: frame.width, height: frame.height, data: frame.into_i420()?, diff --git a/rs/moq-transcode/src/rung.rs b/rs/moq-transcode/src/rung.rs index d85660979..10a1709a7 100644 --- a/rs/moq-transcode/src/rung.rs +++ b/rs/moq-transcode/src/rung.rs @@ -253,7 +253,11 @@ async fn transcode_group_inner( demand: Option<&moq_net::track::Demand>, ) -> Result { let mut first = true; - let mut last_timestamp = 0u64; + // The latest presentation time seen, tracked so a one-shot group can stamp any + // packets the encoder still holds at the end. `None` until the first frame: + // `Timestamp` compares by raw value at a fixed scale, so there's no scale-neutral + // zero to seed it with (the source's scale isn't known until a frame arrives). + let mut last_timestamp: Option = None; loop { let frames = match demand { @@ -268,12 +272,9 @@ async fn transcode_group_inner( }; for frame in frames { - let timestamp: u64 = frame - .timestamp - .as_micros() - .try_into() - .map_err(|_| moq_net::TimeOverflow)?; - last_timestamp = last_timestamp.max(timestamp); + let timestamp = frame.timestamp; + // All source frames share one scale, so this `max` never crosses scales. + last_timestamp = Some(last_timestamp.map_or(timestamp, |last| last.max(timestamp))); // A group opens on a keyframe by construction, so the first frame is // an IDR. The low-level `Container::read` the transcoder uses does not @@ -289,20 +290,18 @@ async fn transcode_group_inner( } } - if demand.is_none() { - // One-shot group: drain whatever the encoder still buffers. + if let (None, Some(last_timestamp)) = (demand, last_timestamp) { + // One-shot group: drain whatever the encoder still buffers, stamping it with + // the last presentation time we saw. No frames read means nothing to drain. write(output, pipeline.finish(last_timestamp)?)?; } Ok(true) } /// Append encoded packets to the output group in the legacy hang framing. -fn write(output: &mut moq_net::group::Producer, packets: Vec<(u64, Bytes)>) -> Result<(), Error> { +fn write(output: &mut moq_net::group::Producer, packets: Vec<(moq_net::Timestamp, Bytes)>) -> Result<(), Error> { for (timestamp, payload) in packets { - let frame = hang::container::Frame { - timestamp: moq_net::Timestamp::from_micros(timestamp)?, - payload, - }; + let frame = hang::container::Frame { timestamp, payload }; frame.encode(output)?; } Ok(()) @@ -348,11 +347,16 @@ impl Pipeline { } /// Transcode one container payload into zero or more encoded packets, each - /// paired with its presentation timestamp in microseconds. - fn process(&mut self, payload: &Bytes, timestamp: u64, keyframe: bool) -> Result, Error> { + /// paired with its presentation timestamp. + fn process( + &mut self, + payload: &Bytes, + timestamp: moq_net::Timestamp, + keyframe: bool, + ) -> Result, Error> { let mut packets = Vec::new(); for raw in self.decoder.decode(payload, timestamp, keyframe)? { - let raw_timestamp = raw.timestamp_us; + let raw_timestamp = raw.timestamp; let encoded = if (raw.width, raw.height) == (self.width, self.height) { // Already at the rung size (the decoder scaled): feed the frame // through as-is, keeping a GPU frame on the GPU. @@ -370,7 +374,7 @@ impl Pipeline { } /// Drain the encoder, pairing any buffered packets with `timestamp`. - fn finish(&mut self, timestamp: u64) -> Result, Error> { + fn finish(&mut self, timestamp: moq_net::Timestamp) -> Result, Error> { Ok(self .encoder .finish()? diff --git a/rs/moq-video/src/decode/backend/mediafoundation.rs b/rs/moq-video/src/decode/backend/mediafoundation.rs index 4dd5fcfc3..bc3e58eee 100644 --- a/rs/moq-video/src/decode/backend/mediafoundation.rs +++ b/rs/moq-video/src/decode/backend/mediafoundation.rs @@ -29,6 +29,7 @@ use std::mem::ManuallyDrop; use std::ptr; use bytes::Bytes; +use moq_net::Timestamp; use windows::Win32::Graphics::Direct3D11::{ID3D11Device, ID3D11Texture2D}; use windows::Win32::Media::MediaFoundation::{ IMF2DBuffer, IMFDXGIBuffer, IMFDXGIDeviceManager, IMFMediaType, IMFSample, IMFTransform, MF_E_NO_MORE_TYPES, @@ -353,7 +354,7 @@ impl MediaFoundation { } impl Backend for MediaFoundation { - fn decode(&mut self, access_unit: Bytes, timestamp_us: u64, _keyframe: bool) -> Result, Error> { + fn decode(&mut self, access_unit: Bytes, timestamp: Timestamp, _keyframe: bool) -> Result, Error> { let mut out = Vec::new(); let sample = self.build_sample(&access_unit)?; @@ -383,7 +384,7 @@ impl Backend for MediaFoundation { Ok(out .into_iter() .map(|i420| Decoded { - timestamp_us, + timestamp, frame: Frame::I420(i420), }) .collect()) diff --git a/rs/moq-video/src/decode/backend/mod.rs b/rs/moq-video/src/decode/backend/mod.rs index 8488eaf7a..7e4eb5272 100644 --- a/rs/moq-video/src/decode/backend/mod.rs +++ b/rs/moq-video/src/decode/backend/mod.rs @@ -15,6 +15,7 @@ //! fallback below the hardware path. use bytes::Bytes; +use moq_net::Timestamp; use super::decoder::{Config, Kind}; use crate::Error; @@ -50,10 +51,10 @@ impl Codec { /// One decoded picture: the raw frame plus its presentation timestamp. pub(crate) struct Decoded { - /// Presentation timestamp in microseconds. Backends that decode one-in - /// one-out echo the input timestamp; NVDEC threads timestamps through its - /// parser, so they survive decoder delay and frame reordering. - pub timestamp_us: u64, + /// Presentation timestamp. Backends that decode one-in one-out echo the input + /// timestamp; NVDEC threads timestamps through its parser, so they survive + /// decoder delay and frame reordering. + pub timestamp: Timestamp, /// The decoded picture: CPU I420, or a GPU frame the encode side can consume /// without a CPU round trip. pub frame: Frame, @@ -63,11 +64,10 @@ pub(crate) struct Decoded { /// or more decoded frames (zero while the decoder is still buffering, e.g. before /// the first keyframe's parameter sets). pub(crate) trait Backend: Send { - /// Decode one Annex-B access unit stamped with its presentation time in - /// microseconds. `keyframe` marks a keyframe (parameter sets are inline ahead - /// of it). Takes an owned [`Bytes`] so a backend can split out NAL slices - /// without copying. - fn decode(&mut self, access_unit: Bytes, timestamp_us: u64, keyframe: bool) -> Result, Error>; + /// Decode one Annex-B access unit stamped with its presentation `timestamp`. + /// `keyframe` marks a keyframe (parameter sets are inline ahead of it). Takes an + /// owned [`Bytes`] so a backend can split out NAL slices without copying. + fn decode(&mut self, access_unit: Bytes, timestamp: Timestamp, keyframe: bool) -> Result, Error>; /// The decoder name in use, e.g. `"videotoolbox"` (for logging). fn name(&self) -> &str; diff --git a/rs/moq-video/src/decode/backend/nvdec.rs b/rs/moq-video/src/decode/backend/nvdec.rs index a0eeefa0f..346f35666 100644 --- a/rs/moq-video/src/decode/backend/nvdec.rs +++ b/rs/moq-video/src/decode/backend/nvdec.rs @@ -26,6 +26,7 @@ use std::sync::Arc; use bytes::Bytes; use cudarc::driver::CudaContext; +use moq_net::Timestamp; use moq_nvenc::cuvid; use moq_nvenc::sys::cuviddec::{ CUVIDDECODECAPS, CUVIDDECODECREATEINFO, CUVIDPICPARAMS, CUVIDPROCPARAMS, CUvideodecoder, cudaVideoChromaFormat, @@ -161,7 +162,7 @@ impl Nvdec { } impl Backend for Nvdec { - fn decode(&mut self, access_unit: Bytes, timestamp_us: u64, _keyframe: bool) -> Result, Error> { + fn decode(&mut self, access_unit: Bytes, timestamp: Timestamp, _keyframe: bool) -> Result, Error> { // The parser callbacks (decoder create, decode) and the map/copy below // all need the CUDA context current on this thread. self.state @@ -177,7 +178,8 @@ impl Backend for Nvdec { | (CUvideopacketflags::CUVID_PKT_ENDOFPICTURE as c_ulong), payload_size: access_unit.len() as c_ulong, payload: access_unit.as_ptr(), - timestamp: timestamp_us as i64, + // cuvid's clock rate is microseconds (set at parser creation). + timestamp: timestamp.as_micros() as i64, }; // SAFETY: parser and packet are valid; the payload outlives the call @@ -384,7 +386,7 @@ impl State { Ok(Decoded { // Timestamps rode the parser from `decode` (microseconds, unsigned). - timestamp_us: disp.timestamp.max(0) as u64, + timestamp: Timestamp::from_micros(disp.timestamp.max(0) as u64).unwrap_or(Timestamp::ZERO), frame: Frame::Cuda(copied?), }) } @@ -512,10 +514,11 @@ mod tests { let rgba = gradient_rgba(w, h); let mut out = Vec::new(); for i in 0..10u64 { + let timestamp = Timestamp::from_micros(i * 33_333).unwrap(); for packet in encoder.encode_rgba(&rgba, w, h, i == 0).unwrap() { - for decoded in decoder.decode(packet, i * 33_333, i == 0).unwrap() { + for decoded in decoder.decode(packet, timestamp, i == 0).unwrap() { let i420 = decoded.frame.to_i420().unwrap().into_owned(); - out.push((decoded.timestamp_us, i420)); + out.push((decoded.timestamp.as_micros() as u64, i420)); } } } @@ -646,8 +649,9 @@ mod tests { let mut decoder = decoder; let mut frames = Vec::new(); for i in 0..10u64 { + let timestamp = Timestamp::from_micros(i * 33_333).unwrap(); for packet in source.encode_rgba(&rgba, w, h, i == 0).unwrap() { - frames.extend(decoder.decode(packet, i * 33_333, i == 0).unwrap()); + frames.extend(decoder.decode(packet, timestamp, i == 0).unwrap()); } } frames @@ -686,7 +690,10 @@ mod tests { .unwrap(); let mut last = None; for (i, packet) in packets.into_iter().enumerate() { - for out in check.decode(packet, i as u64, i == 0).unwrap() { + for out in check + .decode(packet, Timestamp::from_micros(i as u64).unwrap(), i == 0) + .unwrap() + { last = Some(out.frame.to_i420().unwrap().into_owned()); } } diff --git a/rs/moq-video/src/decode/backend/openh264.rs b/rs/moq-video/src/decode/backend/openh264.rs index bc74f0d96..daa4e5fda 100644 --- a/rs/moq-video/src/decode/backend/openh264.rs +++ b/rs/moq-video/src/decode/backend/openh264.rs @@ -5,6 +5,7 @@ //! inline ahead of each keyframe) and returns packed I420. use bytes::Bytes; +use moq_net::Timestamp; use openh264::OpenH264API; use openh264::decoder::{Decoder, DecoderConfig}; use openh264::formats::YUVSource; @@ -33,7 +34,7 @@ impl Openh264 { } impl Backend for Openh264 { - fn decode(&mut self, access_unit: Bytes, timestamp_us: u64, _keyframe: bool) -> Result, Error> { + fn decode(&mut self, access_unit: Bytes, timestamp: Timestamp, _keyframe: bool) -> Result, Error> { let decoded = self .decoder .decode(&access_unit) @@ -64,7 +65,7 @@ impl Backend for Openh264 { ); // openh264 is one-in one-out, so the input timestamp is the output's. Ok(vec![Decoded { - timestamp_us, + timestamp, frame: Frame::I420(frame), }]) } diff --git a/rs/moq-video/src/decode/backend/videotoolbox.rs b/rs/moq-video/src/decode/backend/videotoolbox.rs index fc536ee64..7f411a13e 100644 --- a/rs/moq-video/src/decode/backend/videotoolbox.rs +++ b/rs/moq-video/src/decode/backend/videotoolbox.rs @@ -22,6 +22,7 @@ use std::ptr::{self, NonNull}; use bytes::Bytes; use moq_mux::codec::annexb::NalIterator; +use moq_net::Timestamp; use objc2_core_foundation::{CFDictionary, CFNumber, CFNumberType, CFRetained, CFString}; use objc2_core_media::{ CMBlockBuffer, CMFormatDescription, CMSampleBuffer, CMTime, CMVideoFormatDescriptionCreateFromH264ParameterSets, @@ -166,7 +167,7 @@ impl VideoToolbox { } impl Backend for VideoToolbox { - fn decode(&mut self, access_unit: Bytes, timestamp_us: u64, _keyframe: bool) -> Result, Error> { + fn decode(&mut self, access_unit: Bytes, timestamp: Timestamp, _keyframe: bool) -> Result, Error> { // Split the Annex-B access unit, pull out any parameter sets, and gather // the slices into length-prefixed (4-byte) form. `NalIterator` yields the // parameter-set NALs as zero-copy `Bytes` (sub-slices of `access_unit`), so @@ -230,7 +231,7 @@ impl Backend for VideoToolbox { Ok(std::mem::take(&mut self.sink.frames) .into_iter() .map(|i420| Decoded { - timestamp_us, + timestamp, frame: Frame::I420(i420), }) .collect()) diff --git a/rs/moq-video/src/decode/consumer.rs b/rs/moq-video/src/decode/consumer.rs index 4efea5cab..01308b7c4 100644 --- a/rs/moq-video/src/decode/consumer.rs +++ b/rs/moq-video/src/decode/consumer.rs @@ -61,15 +61,10 @@ impl Consumer { let Some(mux_frame) = self.track.read().await? else { return Ok(None); }; - let timestamp_us: u64 = mux_frame - .timestamp - .as_micros() - .try_into() - .map_err(|_| moq_net::TimeOverflow)?; self.pending.extend( self.decoder - .decode(&mux_frame.payload, timestamp_us, mux_frame.keyframe)?, + .decode(&mux_frame.payload, mux_frame.timestamp, mux_frame.keyframe)?, ); } } diff --git a/rs/moq-video/src/decode/decoder.rs b/rs/moq-video/src/decode/decoder.rs index 88531e96e..25ee94345 100644 --- a/rs/moq-video/src/decode/decoder.rs +++ b/rs/moq-video/src/decode/decoder.rs @@ -12,6 +12,7 @@ use std::time::Duration; use bytes::Bytes; use hang::catalog::{VideoCodec, VideoConfig}; use moq_mux::codec::{annexb, h264, h265}; +use moq_net::Timestamp; use super::Frame; use super::backend::{self, Backend, Codec}; @@ -141,9 +142,12 @@ impl Decoder { self.backend.name() } - /// Decode one container frame, returning zero or more raw frames stamped with - /// `timestamp_us` (the container frame's presentation time in microseconds). - pub fn decode(&mut self, payload: &Bytes, timestamp_us: u64, keyframe: bool) -> Result, Error> { + /// Decode one container frame, returning zero or more raw frames. `timestamp` is + /// this frame's presentation time; it rides through the decoder and comes back on + /// each output frame, so a reordering decoder (B-frames) stamps every picture + /// with its own presentation time rather than this access unit's. With no + /// reordering the two coincide. + pub fn decode(&mut self, payload: &Bytes, timestamp: Timestamp, keyframe: bool) -> Result, Error> { // Wait for the first keyframe: a decoder started mid-GOP can't decode // delta frames, and the parameter sets ride along with the keyframe. if !self.got_keyframe { @@ -167,10 +171,10 @@ impl Decoder { Ok(self .backend - .decode(annexb, timestamp_us, keyframe)? + .decode(annexb, timestamp, keyframe)? .into_iter() .map(|decoded| Frame { - timestamp_us: decoded.timestamp_us, + timestamp: decoded.timestamp, width: decoded.frame.width(), height: decoded.frame.height(), inner: decoded.frame, @@ -181,6 +185,8 @@ impl Decoder { #[cfg(test)] mod tests { + use moq_net::Timestamp; + use super::backend::{self, Codec}; use crate::encode::{Config as EncodeConfig, Encoder, Kind as EncodeKind}; use crate::frame::I420; @@ -220,8 +226,10 @@ mod tests { let mut decoded = Vec::new(); for i in 0..10u64 { let keyframe = i == 0; + // Distinct, spread-apart timestamps so a round-tripped value is unambiguous. + let timestamp = Timestamp::from_micros(i * 33_333).unwrap(); for packet in encoder.encode_rgba(&frame, 320, 240, keyframe).unwrap() { - decoded.extend(decoder.decode(packet, i * 33_333, keyframe).unwrap()); + decoded.extend(decoder.decode(packet, timestamp, keyframe).unwrap()); } } @@ -229,6 +237,19 @@ mod tests { for out in &decoded { assert_gray(&out.frame.to_i420().unwrap(), 320, 240); } + + // The timestamp rides through the codec and comes back on each picture. These + // backends don't reorder, so it returns in feed order: strictly increasing and + // drawn from the values we fed. + let micros: Vec = decoded.iter().map(|d| d.timestamp.as_micros()).collect(); + assert!( + micros.windows(2).all(|w| w[0] < w[1]), + "decoded timestamps not strictly increasing: {micros:?}" + ); + assert!( + micros.iter().all(|&t| t % 33_333 == 0 && t < 333_330), + "decoded timestamp outside the fed set: {micros:?}" + ); } /// A decoder config selecting one backend by kind. diff --git a/rs/moq-video/src/decode/mod.rs b/rs/moq-video/src/decode/mod.rs index 8130e9f8c..2f8c0ff41 100644 --- a/rs/moq-video/src/decode/mod.rs +++ b/rs/moq-video/src/decode/mod.rs @@ -11,6 +11,7 @@ //! yields [`Error::UnsupportedCodec`](crate::Error). use bytes::Bytes; +use moq_net::Timestamp; use crate::Error; @@ -32,8 +33,10 @@ pub use decoder::{Config, Decoder, Kind}; /// (the zero-copy transcode path), while [`into_i420`](Self::into_i420) /// downloads it. pub struct Frame { - /// Presentation timestamp in microseconds (from the container). - pub timestamp_us: u64, + /// Presentation timestamp, carried through from the container. It rides out of + /// the decoder with each picture, so a reordered frame (B-frames) keeps its own + /// time rather than the input access unit's. + pub timestamp: Timestamp, /// Frame width in pixels (even). pub width: u32, /// Frame height in pixels (even). diff --git a/rs/moq-video/src/encode/backend/nvenc.rs b/rs/moq-video/src/encode/backend/nvenc.rs index 831cde322..a0e4830a2 100644 --- a/rs/moq-video/src/encode/backend/nvenc.rs +++ b/rs/moq-video/src/encode/backend/nvenc.rs @@ -516,8 +516,9 @@ mod tests { // decoder latency or frame reordering. let mut decoded = None; for i in 0..10u64 { + let timestamp = moq_net::Timestamp::from_micros(i * 33_333).unwrap(); for packet in encoder.encode_rgba(&rgba, w, h, i == 0).unwrap() { - for out in decoder.decode(packet, i * 33_333, i == 0).unwrap() { + for out in decoder.decode(packet, timestamp, i == 0).unwrap() { decoded = Some(out.frame.to_i420().unwrap().into_owned()); } }