diff --git a/rs/hang/src/container/frame.rs b/rs/hang/src/container/frame.rs index 77c7ae8955..e1c1382093 100644 --- a/rs/hang/src/container/frame.rs +++ b/rs/hang/src/container/frame.rs @@ -174,7 +174,7 @@ mod test { #[test] fn track_info_uses_container_timescale() { - assert_eq!(track_info(crate::catalog::PRIORITY.video).timescale, TIMESCALE); + assert_eq!(track_info(crate::catalog::PRIORITY.video).timescale, Some(TIMESCALE)); } #[test] @@ -186,7 +186,7 @@ mod test { // Retimescaling for a container that carries the source's own scale keeps it, since that // is the shape that would otherwise reach for `Info::default()` and lose the retention. let at = track_info(crate::catalog::PRIORITY.video).with_timescale(Timescale::MILLI); - assert_eq!(at.timescale, Timescale::MILLI); + assert_eq!(at.timescale, Some(Timescale::MILLI)); assert_eq!(at.max_age, Some(MAX_AGE)); } diff --git a/rs/moq-ffi/src/producer.rs b/rs/moq-ffi/src/producer.rs index 59de359329..def763bde8 100644 --- a/rs/moq-ffi/src/producer.rs +++ b/rs/moq-ffi/src/producer.rs @@ -65,7 +65,7 @@ impl TryFrom<&moq_net::track::Info> for MoqTrackInfo { Ok(Self { priority: info.priority, max_age_us, - timescale: Some(info.timescale.as_u64()), + timescale: info.timescale.map(|scale| scale.as_u64()), }) } } diff --git a/rs/moq-flate/src/stream/mod.rs b/rs/moq-flate/src/stream/mod.rs index 0327947c6c..88b8fae587 100644 --- a/rs/moq-flate/src/stream/mod.rs +++ b/rs/moq-flate/src/stream/mod.rs @@ -190,7 +190,7 @@ mod test { /// allocating 32 MB to provoke it. Borrowed from moq-json's stream tests. fn rejecting_track() -> moq_net::track::Producer { let mut info = moq_net::track::Info::default(); - info.timescale = moq_net::Timescale::new((1u64 << 62) - 1).unwrap(); + info.timescale = Some(moq_net::Timescale::new((1u64 << 62) - 1).unwrap()); moq_net::broadcast::Info::new() .produce() diff --git a/rs/moq-gst/src/sink/pad.rs b/rs/moq-gst/src/sink/pad.rs index b606283142..e587df1626 100644 --- a/rs/moq-gst/src/sink/pad.rs +++ b/rs/moq-gst/src/sink/pad.rs @@ -1248,7 +1248,7 @@ mod tests { .subscribe(None) .await .expect("subscribe to the opaque track"); - assert_eq!(subscriber.info().timescale, moq_net::Timescale::MICRO); + assert_eq!(subscriber.info().timescale, Some(moq_net::Timescale::MICRO)); assert_eq!( subscriber.info().max_age, Some(std::time::Duration::from_secs(5)), diff --git a/rs/moq-json/src/snapshot/mod.rs b/rs/moq-json/src/snapshot/mod.rs index 761ccaf009..bef3c89be3 100644 --- a/rs/moq-json/src/snapshot/mod.rs +++ b/rs/moq-json/src/snapshot/mod.rs @@ -704,7 +704,7 @@ mod test { /// allocating 32 MB to provoke it. fn rejecting_track() -> moq_net::track::Producer { let mut info = moq_net::track::Info::default(); - info.timescale = moq_net::Timescale::new((1u64 << 62) - 1).unwrap(); + info.timescale = Some(moq_net::Timescale::new((1u64 << 62) - 1).unwrap()); moq_net::broadcast::Info::new() .produce() diff --git a/rs/moq-json/src/stream/mod.rs b/rs/moq-json/src/stream/mod.rs index 5c2e217fa3..1a74355043 100644 --- a/rs/moq-json/src/stream/mod.rs +++ b/rs/moq-json/src/stream/mod.rs @@ -237,7 +237,7 @@ mod test { /// allocating 32 MB to provoke it. fn rejecting_track() -> moq_net::track::Producer { let mut info = moq_net::track::Info::default(); - info.timescale = moq_net::Timescale::new((1u64 << 62) - 1).unwrap(); + info.timescale = Some(moq_net::Timescale::new((1u64 << 62) - 1).unwrap()); moq_net::broadcast::Info::new() .produce() diff --git a/rs/moq-json/src/window/mod.rs b/rs/moq-json/src/window/mod.rs index d0c461b3b0..eb9003e284 100644 --- a/rs/moq-json/src/window/mod.rs +++ b/rs/moq-json/src/window/mod.rs @@ -91,7 +91,7 @@ mod test { /// A track whose timestamp conversion rejects every frame after its group is published. fn rejecting_track() -> moq_net::track::Producer { let mut info = moq_net::track::Info::default(); - info.timescale = moq_net::Timescale::new((1u64 << 62) - 1).unwrap(); + info.timescale = Some(moq_net::Timescale::new((1u64 << 62) - 1).unwrap()); moq_net::broadcast::Info::new() .produce() diff --git a/rs/moq-mux/src/catalog/producer.rs b/rs/moq-mux/src/catalog/producer.rs index 698d1fd4d2..fc45c011e3 100644 --- a/rs/moq-mux/src/catalog/producer.rs +++ b/rs/moq-mux/src/catalog/producer.rs @@ -1132,11 +1132,11 @@ mod test { let info = catalog.track_info(hang::catalog::PRIORITY.video); assert_eq!(info.max_age, Some(std::time::Duration::from_secs(3))); - assert_eq!(info.timescale, hang::container::TIMESCALE); + assert_eq!(info.timescale, Some(hang::container::TIMESCALE)); let at = info.with_timescale(moq_net::Timescale::MILLI); assert_eq!(at.max_age, Some(std::time::Duration::from_secs(3))); - assert_eq!(at.timescale, moq_net::Timescale::MILLI); + assert_eq!(at.timescale, Some(moq_net::Timescale::MILLI)); // Every handle mints under the same policy, whatever order it was taken in: the codec // paths hold a reservation and the container paths hold a clone. diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 856fca169b..02be03c05d 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -671,7 +671,7 @@ where None => track::Position::after_group(end.group), }); let _ = track.update(subscription); - let timescale = msg.properties_wanted.then(|| track.info().timescale); + let timescale = track.info().timescale.filter(|_| msg.properties_wanted); // Draft-20 replaced joining FETCH with subscription fills. Older drafts save // the same boundary used by the subscription so the two streams never overlap. @@ -716,7 +716,7 @@ where // We serve the newest group first, matching moq-lite. true => ietf::Properties { max_cache_duration: track.info().max_age, - timescale: Some(track.info().timescale), + timescale: track.info().timescale, priority: Some(super::priority::to_wire(track.info().priority)), group_order: Some(GroupOrder::Descending), }, @@ -3279,7 +3279,9 @@ mod serve_tests { fn serve(version: Version) -> Serve { let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); let broadcast = origin.publish("room", crate::origin::Route::default()).unwrap(); - let track = broadcast.create_track("video", None).unwrap(); + // Timed, so served objects carry a Timestamp. + let info = track::Info::default().with_timescale(crate::Timescale::MILLI); + let track = broadcast.create_track("video", info).unwrap(); let session = ScriptedSession::per_stream(vec![Vec::new()]); let log = session.log.clone(); diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index e1b6bd34eb..462622b0e3 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -1996,8 +1996,9 @@ where priority, largest, } = accepted; + // Untimed unless SUBSCRIBE_OK declared a TIMESCALE. let info = track::Info::default() - .with_timescale(Timescale::MICRO) + .with_timescale(timescale) .with_max_age(max_age) .with_priority(super::priority::from_wire(priority.unwrap_or(128))); let (mut track, dynamic) = match resumed { @@ -3504,7 +3505,7 @@ where // have set it. FETCH_OK carries the same Track Properties, so the retention window // comes from it. let info = track::Info::default() - .with_timescale(Timescale::MICRO) + .with_timescale(ok.properties.timescale) .with_max_age(ok.properties.max_cache_duration); // Joined fetches still count until they pick the accepted group up from the cache. let joined = request.result.clone(); diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 861b82abf1..63337dc05c 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -1202,7 +1202,7 @@ impl Request for TrackInfoServe { writer.buffer(&lite::TrackInfo { priority: info.priority, max_age: info.max_age, - timescale: info.timescale, + timescale: info.scale(), })?; Poll::Ready(Ok(ControlFlow::Break(()))) } @@ -1295,7 +1295,7 @@ impl Request for SubscribeServe { // "don't emit the prefix" (the frames still carry timestamps in the // model, just not on this wire). let timescale = if shared.version.has_track_stream() { - Some(track.info().timescale) + Some(track.info().scale()) } else { None }; diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index ea801d25c8..360ae632e8 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -3876,7 +3876,7 @@ impl kio::Task for TrackServeRun { // Lite05 carries per-frame timestamps on the wire at this scale; // `Some` tells the ingest to decode them instead of stamping // local receive time. - let timescale = Some(info.timescale); + let timescale = info.timescale; self.state = TrackRunState::Serve(ServeLoop::new(&self.serve, request, info, timescale)); } Err(err) => { diff --git a/rs/moq-net/src/lite/track.rs b/rs/moq-net/src/lite/track.rs index a49e6d1dce..d4ee07270e 100644 --- a/rs/moq-net/src/lite/track.rs +++ b/rs/moq-net/src/lite/track.rs @@ -203,7 +203,7 @@ mod test { let info = TrackInfo { priority: info.priority, max_age: info.max_age, - timescale: info.timescale, + timescale: info.scale(), }; let mut buf = Vec::new(); info.encode(&mut buf, Version::Lite05).unwrap(); diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index 0c9f9edd75..126de710e1 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -26,6 +26,8 @@ use std::task::{Poll, ready}; use crate::{Error, IntoBytes, Result, Timestamp}; +pub use super::timing::group::{Timed, Untimed}; + /// Maximum total size of frames in a group. /// /// A write that would exceed this aborts the group with [`Error::GroupTooLarge`]. @@ -461,9 +463,9 @@ impl Producer { self.info } - /// The parent track's timescale. + /// The scale this group's frame timestamps are stored at; milliseconds on an untimed track. pub fn timescale(&self) -> Timescale { - self.track.timescale + self.track.scale() } /// Start the group at frame `index` rather than 0, so the first frame written lands @@ -506,7 +508,7 @@ impl Producer { /// a presentation time, pass [`Timestamp::now`] explicitly. pub fn write_frame(&mut self, timestamp: Timestamp, data: B) -> Result<()> { let timestamp = timestamp - .convert(self.track.timescale) + .convert(self.track.scale()) .map_err(|_| Error::TimestampMismatch)?; let payload = data.into_bytes(); if payload.len() as u64 > MAX_CACHE_BYTES { @@ -561,7 +563,7 @@ impl Producer { for frame in frames.filled() { frame .timestamp - .convert(self.track.timescale) + .convert(self.track.scale()) .map_err(|_| Error::TimestampMismatch)?; if frame.payload.len() as u64 > MAX_CACHE_BYTES { return Err(Error::FrameTooLarge); @@ -590,7 +592,7 @@ impl Producer { for mut frame in frames.drain() { frame.timestamp = frame .timestamp - .convert(self.track.timescale) + .convert(self.track.scale()) .expect("timestamp scale checked above"); let size = frame.payload.len() as u64; state.cache += size; @@ -654,7 +656,7 @@ impl Producer { fn open_frame(&mut self, frame: frame::Info, buf: &FrameBuf) -> Result { let timestamp = frame .timestamp - .convert(self.track.timescale) + .convert(self.track.scale()) .map_err(|_| Error::TimestampMismatch)?; if frame.size > MAX_CACHE_BYTES { return Err(Error::FrameTooLarge); @@ -1435,9 +1437,9 @@ impl Consumer { self.cursor.state.poll_closed(waiter) } - /// The parent track's timescale. + /// The scale this group's frame timestamps are stored at; milliseconds on an untimed track. pub fn timescale(&self) -> Timescale { - self.track.timescale + self.track.scale() } /// The index of the next frame this consumer will return. diff --git a/rs/moq-net/src/model/mod.rs b/rs/moq-net/src/model/mod.rs index a2e6a3b414..deeee6b961 100644 --- a/rs/moq-net/src/model/mod.rs +++ b/rs/moq-net/src/model/mod.rs @@ -4,6 +4,7 @@ pub mod cache; pub mod frame; pub mod group; pub mod track; +pub mod typed; // The origin + announce subsystem shares one implementation (a route table). // It stays in a single private module and is surfaced as two curated public @@ -22,6 +23,7 @@ mod resume; mod subscription; mod time; mod timed; +mod timing; mod weak_cache; #[cfg(test)] diff --git a/rs/moq-net/src/model/timed.rs b/rs/moq-net/src/model/timed.rs index 4cb354f7d9..50651491a0 100644 --- a/rs/moq-net/src/model/timed.rs +++ b/rs/moq-net/src/model/timed.rs @@ -1,59 +1,21 @@ -use bytes::Bytes; - use crate::Timestamp; -/// A value to publish, and optionally when it was captured. +/// A value and the time it was captured, written to a timed track. /// -/// `T` is the clock the time is on: a raw [`Timestamp`] by default, or whatever a higher layer -/// maps onto one. Converts from a bare payload (bytes, or a `&V` to serialize), so a producer -/// taking one still accepts the plain value, stamped when written. -#[derive(Debug, Clone, Copy)] -pub struct Timed { +/// An untimed track takes the bare value instead, so a stamp can be neither forgotten on a +/// timed track nor invented on an untimed one. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Timed

{ /// The value to publish. pub value: P, - /// When the value was captured. `None` stamps it when written. - pub at: Option, -} - -impl Timed { - /// Set when the value was captured. - pub fn at(mut self, at: T) -> Self { - self.at = Some(at); - self - } -} - -impl, T> From for Timed { - fn from(value: B) -> Self { - Self { - value: value.into(), - at: None, - } - } + /// When the value was captured, at any scale; the track converts it to its own. + pub at: Timestamp, } -impl<'a, V, T> From<&'a V> for Timed<&'a V, T> { - fn from(value: &'a V) -> Self { - Self { value, at: None } - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn a_bare_payload_converts_untimed() { - let timed: Timed = vec![1u8, 2].into(); - assert_eq!(timed.value, Bytes::from_static(&[1, 2])); - assert_eq!(timed.at, None); - - let stamped = Timed::from(&b"x"[..]).at(Timestamp::from_millis(5).unwrap()); - assert_eq!(stamped.at, Some(Timestamp::from_millis(5).unwrap())); - - let value = 7u32; - let timed: Timed<&u32, std::time::Instant> = (&value).into(); - assert_eq!(*timed.value, 7); +impl

Timed

{ + /// Pair a value with when it was captured. + pub fn new(value: P, at: Timestamp) -> Self { + Self { value, at } } } diff --git a/rs/moq-net/src/model/timing.rs b/rs/moq-net/src/model/timing.rs new file mode 100644 index 0000000000..786d186a26 --- /dev/null +++ b/rs/moq-net/src/model/timing.rs @@ -0,0 +1,453 @@ +//! Typed handles that make a track all-timed or all-untimed. +//! +//! Readers learn the timing at runtime, when the subscription is accepted, through +//! [`track::Subscriber::timing`](crate::track::Subscriber::timing). Writers pick it when +//! they create the track. +//! +//! Variant A (this module): the timescale is a runtime value in [`track::Info`]. The +//! same `Timed`/`Untimed` handles wrap a writer (`Timed`) or a reader +//! (`Timed`, the default), so the timing is in the type and the scale +//! is in the value. Variant B ([`crate::typed`]) puts the scale in the writer's type too. +//! +//! [`track::Info`]: crate::track::Info + +/// Typed track handles, re-exported from [`crate::track`]. +pub(super) mod track { + use std::task::Poll; + + use bytes::Bytes; + + use crate::frame::Frame; + use crate::{Error, IntoBytes, Result, Timescale, Timestamp, group, track}; + + use super::group::{Timed as TimedGroup, Untimed as UntimedGroup}; + + /// A subscribed track, split by whether its frames carry timestamps. + pub enum Timing { + /// Every frame and datagram carries a timestamp. + Timed(Timed), + /// No frame or datagram carries a timestamp. + Untimed(Untimed), + } + + /// A timed track: every frame and datagram carries a timestamp at [`Self::timescale`]. + /// + /// Reads through a [`track::Subscriber`] (the default) or writes through a + /// [`track::Producer`]. + pub struct Timed { + inner: H, + timescale: Timescale, + } + + /// An untimed track: frames and datagrams are bare payloads. + /// + /// Reads through a [`track::Subscriber`] (the default) or writes through a + /// [`track::Producer`]. + pub struct Untimed { + inner: H, + } + + impl Timed { + /// Units per second for every timestamp on this track. + pub fn timescale(&self) -> Timescale { + self.timescale + } + } + + impl track::Subscriber { + /// Split this subscriber by the track's timing, as the publisher declared it. + pub fn timing(self) -> Timing { + match self.info().timescale { + Some(timescale) => Timing::Timed(Timed { inner: self, timescale }), + None => Timing::Untimed(Untimed { inner: self }), + } + } + } + + impl track::Producer { + /// Write this track as timed, taking a [`crate::Timed`] per frame. + /// + /// Refused with [`Error::TimestampMismatch`] when [`track::Info::timescale`] is `None`. + pub fn timed(self) -> Result> { + let timescale = self.info().timescale.ok_or(Error::TimestampMismatch)?; + Ok(Timed { inner: self, timescale }) + } + + /// Write this track as untimed, taking a bare payload per frame. + /// + /// Refused with [`Error::TimestampMismatch`] when [`track::Info::timescale`] is set. + pub fn untimed(self) -> Result> { + if self.info().timescale.is_some() { + return Err(Error::TimestampMismatch); + } + Ok(Untimed { inner: self }) + } + } + + impl Timed { + /// The track's [`track::Info`]. + pub fn info(&self) -> &track::Info { + self.inner.info() + } + + /// Poll for the next group in arrival order. + pub fn poll_recv_group(&mut self, waiter: &kio::Waiter) -> Poll>> { + self.inner + .poll_recv_group(waiter) + .map_ok(|group| group.map(TimedGroup::new)) + } + + /// Receive the next group in arrival order. + pub async fn recv_group(&mut self) -> Result> { + kio::wait(|waiter| self.poll_recv_group(waiter)).await + } + + /// Poll for the next datagram in arrival order. + pub fn poll_recv_datagram(&mut self, waiter: &kio::Waiter) -> Poll>> { + self.inner.poll_recv_datagram(waiter).map_ok(|datagram| { + datagram.map(|datagram| Frame { + timestamp: datagram.timestamp, + payload: datagram.payload, + }) + }) + } + + /// Receive the next datagram in arrival order. + pub async fn recv_datagram(&mut self) -> Result> { + kio::wait(|waiter| self.poll_recv_datagram(waiter)).await + } + } + + impl Untimed { + /// The track's [`track::Info`]. + pub fn info(&self) -> &track::Info { + self.inner.info() + } + + /// Poll for the next group in arrival order. + pub fn poll_recv_group(&mut self, waiter: &kio::Waiter) -> Poll>> { + self.inner + .poll_recv_group(waiter) + .map_ok(|group| group.map(UntimedGroup::new)) + } + + /// Receive the next group in arrival order. + pub async fn recv_group(&mut self) -> Result> { + kio::wait(|waiter| self.poll_recv_group(waiter)).await + } + + /// Poll for the next datagram's payload in arrival order. + pub fn poll_recv_datagram(&mut self, waiter: &kio::Waiter) -> Poll>> { + self.inner + .poll_recv_datagram(waiter) + .map_ok(|datagram| datagram.map(|datagram| datagram.payload)) + } + + /// Receive the next datagram's payload in arrival order. + pub async fn recv_datagram(&mut self) -> Result> { + kio::wait(|waiter| self.poll_recv_datagram(waiter)).await + } + } + + impl Timed { + /// The track's name, unique within its broadcast. + pub fn name(&self) -> &str { + self.inner.name() + } + + /// Create a group with the next sequence number. + pub fn append_group(&self) -> Result> { + Ok(TimedGroup::new(self.inner.append_group()?)) + } + + /// Create a group with the given sequence number. + pub fn create_group(&self, info: group::Info) -> Result> { + Ok(TimedGroup::new(self.inner.create_group(info)?)) + } + + /// Write a single-frame group. + pub fn write_frame(&mut self, frame: crate::Timed) -> Result<()> { + self.inner.write_frame(frame.at, frame.value) + } + + /// Append a datagram with the next sequence number, returning the sequence. + pub fn append_datagram(&mut self, datagram: crate::Timed) -> Result { + self.inner.append_datagram(datagram.at, datagram.value) + } + + /// End the track after its last group. + pub fn finish(&self) -> Result<()> { + self.inner.finish() + } + + /// A cheap, cloneable handle to subscribe with. + pub fn consume(&self) -> track::Consumer { + self.inner.consume() + } + } + + impl Untimed { + /// The track's name, unique within its broadcast. + pub fn name(&self) -> &str { + self.inner.name() + } + + /// Create a group with the next sequence number. + pub fn append_group(&self) -> Result> { + Ok(UntimedGroup::new(self.inner.append_group()?)) + } + + /// Create a group with the given sequence number. + pub fn create_group(&self, info: group::Info) -> Result> { + Ok(UntimedGroup::new(self.inner.create_group(info)?)) + } + + /// Write a single-frame group. + pub fn write_frame(&mut self, payload: B) -> Result<()> { + self.inner.write_frame(Timestamp::now(), payload) + } + + /// Append a datagram with the next sequence number, returning the sequence. + pub fn append_datagram(&mut self, payload: B) -> Result { + self.inner.append_datagram(Timestamp::now(), payload) + } + + /// End the track after its last group. + pub fn finish(&self) -> Result<()> { + self.inner.finish() + } + + /// A cheap, cloneable handle to subscribe with. + pub fn consume(&self) -> track::Consumer { + self.inner.consume() + } + } +} + +/// Typed group handles, re-exported from [`crate::group`]. +pub(super) mod group { + use std::task::Poll; + + use bytes::Bytes; + + use crate::frame::{self, Frame}; + use crate::{Error, IntoBytes, Result, Timescale, Timestamp, group}; + + /// A group of a timed track: every frame carries a timestamp. + /// + /// Reads through a [`group::Consumer`] (the default) or writes through a + /// [`group::Producer`]. + pub struct Timed { + inner: H, + } + + /// A group of an untimed track: frames are bare payloads. + /// + /// Reads through a [`group::Consumer`] (the default) or writes through a + /// [`group::Producer`]. + pub struct Untimed { + inner: H, + } + + impl Timed { + // Only a typed track hands these out, having checked the timing. + pub(super) fn new(inner: H) -> Self { + Self { inner } + } + } + + impl Untimed { + pub(super) fn new(inner: H) -> Self { + Self { inner } + } + } + + impl> std::ops::Deref for Timed { + type Target = group::Info; + + fn deref(&self) -> &group::Info { + &self.inner + } + } + + impl> std::ops::Deref for Untimed { + type Target = group::Info; + + fn deref(&self) -> &group::Info { + &self.inner + } + } + + impl Timed { + /// Units per second for every timestamp in this group. + pub fn timescale(&self) -> Timescale { + self.inner.timescale() + } + + /// Poll for the next frame, timestamp and payload. + pub fn poll_read_frame(&mut self, waiter: &kio::Waiter) -> Poll>> { + self.inner.poll_read_frame(waiter) + } + + /// Read the next frame, timestamp and payload. + pub async fn read_frame(&mut self) -> Result> { + self.inner.read_frame().await + } + + /// Fill `out` with every frame that is ready, without blocking. + pub fn poll_read_frames( + &mut self, + waiter: &kio::Waiter, + out: &mut frame::Buffer, + ) -> Poll> { + self.inner.poll_read_frames(waiter, out) + } + + /// Fill `out` with every frame that is ready, waiting for at least one. + pub async fn read_frames<'a, const N: usize>( + &mut self, + out: &'a mut frame::Buffer, + ) -> Result<&'a mut [Frame]> { + self.inner.read_frames(out).await + } + + /// Return the next frame for chunked reading. + pub async fn next_frame(&mut self) -> Result> { + self.inner.next_frame().await + } + } + + impl Untimed { + /// Poll for the next frame's payload. + pub fn poll_read_frame(&mut self, waiter: &kio::Waiter) -> Poll>> { + self.inner + .poll_read_frame(waiter) + .map_ok(|frame| frame.map(|frame| frame.payload)) + } + + /// Read the next frame's payload. + pub async fn read_frame(&mut self) -> Result> { + kio::wait(|waiter| self.poll_read_frame(waiter)).await + } + } + + impl Timed { + /// Write a whole frame. + pub fn write_frame(&mut self, frame: crate::Timed) -> Result<()> { + self.inner.write_frame(frame.at, frame.value) + } + + /// Open a frame of `size` bytes captured `at`, to write in chunks. + pub fn create_frame(&mut self, size: u64, at: Timestamp) -> Result> { + self.inner.create_frame(frame::Info { size, timestamp: at }) + } + + /// Write a batch of frames at once, draining `frames`. + pub fn write_frames(&mut self, frames: &mut frame::Buffer) -> Result<()> { + self.inner.write_frames(frames) + } + + /// Mark the group complete. + pub fn finish(&self) -> Result<()> { + self.inner.finish() + } + + /// Abort the group with `err`. + pub fn abort(self, err: Error) -> Result<()> { + self.inner.abort(err) + } + } + + impl Untimed { + /// Write a whole frame. + pub fn write_frame(&mut self, payload: B) -> Result<()> { + self.inner.write_frame(Timestamp::now(), payload) + } + + /// Open a frame of `size` bytes, to write in chunks. + pub fn create_frame(&mut self, size: u64) -> Result> { + self.inner.create_frame(frame::Info { + size, + timestamp: Timestamp::now(), + }) + } + + /// Mark the group complete. + pub fn finish(&self) -> Result<()> { + self.inner.finish() + } + + /// Abort the group with `err`. + pub fn abort(self, err: Error) -> Result<()> { + self.inner.abort(err) + } + } +} + +#[cfg(test)] +mod tests { + use bytes::Bytes; + + use crate::{Error, Timed, Timescale, Timestamp, broadcast, track}; + + fn ms(value: u64) -> Timestamp { + Timestamp::from_millis(value).unwrap() + } + + #[tokio::test] + async fn a_timed_track_reads_timestamps() { + let broadcast = broadcast::Info::default().produce(); + let info = track::Info::default().with_timescale(Timescale::MICRO); + let mut video = broadcast.create_track("video", info).unwrap().timed().unwrap(); + let consumer = video.consume(); + + video.write_frame(Timed::new(&b"key"[..], ms(2))).unwrap(); + video.append_datagram(Timed::new(&b"dg"[..], ms(3))).unwrap(); + + let track::Timing::Timed(mut reader) = consumer.subscribe(None).await.unwrap().timing() else { + panic!("a track with a timescale reads timed"); + }; + assert_eq!(reader.timescale(), Timescale::MICRO); + + let mut group = reader.recv_group().await.unwrap().unwrap(); + let frame = group.read_frame().await.unwrap().unwrap(); + assert_eq!(frame.timestamp, Timestamp::from_micros(2_000).unwrap()); + assert_eq!(frame.payload, Bytes::from_static(b"key")); + + let datagram = reader.recv_datagram().await.unwrap().unwrap(); + assert_eq!(datagram.timestamp.value(), 3_000); + } + + #[tokio::test] + async fn an_untimed_track_reads_bare_payloads() { + let broadcast = broadcast::Info::default().produce(); + let mut chat = broadcast.create_track("chat", None).unwrap().untimed().unwrap(); + let consumer = chat.consume(); + + let mut group = chat.append_group().unwrap(); + group.write_frame(&b"hi"[..]).unwrap(); + group.finish().unwrap(); + chat.append_datagram(&b"dg"[..]).unwrap(); + + let track::Timing::Untimed(mut reader) = consumer.subscribe(None).await.unwrap().timing() else { + panic!("a track without a timescale reads untimed"); + }; + let mut group = reader.recv_group().await.unwrap().unwrap(); + assert_eq!(group.read_frame().await.unwrap().unwrap(), Bytes::from_static(b"hi")); + assert_eq!( + reader.recv_datagram().await.unwrap().unwrap(), + Bytes::from_static(b"dg") + ); + } + + #[test] + fn the_writer_must_match_the_track() { + let broadcast = broadcast::Info::default().produce(); + let untimed = broadcast.create_track("chat", None).unwrap(); + assert!(matches!(untimed.timed(), Err(Error::TimestampMismatch))); + + let info = track::Info::default().with_timescale(Timescale::MILLI); + let timed = broadcast.create_track("video", info).unwrap(); + assert!(matches!(timed.untimed(), Err(Error::TimestampMismatch))); + } +} diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 95d0acae27..0ca9915fd2 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -20,6 +20,7 @@ use super::{Datagram, Requests}; use super::Cap; pub use super::subscription::{Position, Subscription}; +pub use super::timing::track::{Timed, Timing, Untimed}; use std::{ collections::{BTreeMap, VecDeque}, @@ -87,13 +88,14 @@ pub(crate) const IDLE_LINGER: Duration = Duration::from_secs(30); #[derive(Clone, Debug)] #[non_exhaustive] pub struct Info { - /// Units per second for per-frame timestamps on this track. + /// Units per second for per-frame timestamps, or `None` for an untimed track. /// - /// Every track is timed; this defaults to [`Timescale::MILLI`]. On Lite05+ it is - /// reported in TRACK_INFO and the publisher zigzag-delta encodes per-frame - /// timestamps at this scale on the wire. Protocols whose wire can't carry it - /// (pre-Lite05 moq-lite, IETF moq-transport) fall back to local monotonic milliseconds. - pub timescale: Timescale, + /// Timedness is a property of the whole track: a timed track stamps every frame and + /// datagram, an untimed one stamps none. Defaults to `None`, so a publisher opts into + /// timing. A receiver learns it when the track is accepted: TRACK_INFO on Lite05+, the + /// TIMESCALE property on moq-transport. Wires that can't declare it (pre-Lite05, + /// IETF drafts 14-16) arrive untimed. + pub timescale: Option, /// How far behind the live edge a group may fall, in media timestamps, before it /// is stale. The newest group is always retained. /// @@ -125,7 +127,7 @@ pub struct Info { impl Default for Info { fn default() -> Self { Self { - timescale: Timescale::default(), + timescale: None, max_age: None, priority: DEFAULT_PRIORITY, } @@ -133,15 +135,23 @@ impl Default for Info { } impl Info { - /// Set the per-frame timestamp scale, returning `self` for chaining. - /// - /// Defaults to [`Timescale::MILLI`]. On Lite05+ this scale is reported in TRACK_INFO - /// and used to encode per-frame timestamps on the wire. - pub fn with_timescale(mut self, timescale: Timescale) -> Self { - self.timescale = timescale; + /// Make the track timed at `timescale`, returning `self` for chaining. + pub fn with_timescale(mut self, timescale: impl Into>) -> Self { + self.timescale = timescale.into(); self } + /// The scale frames are stored and encoded at: the track's own, or milliseconds + /// for an untimed track. + // + // Mock-up stand-in: the untyped handles still store a timestamp on every frame, so + // an untimed track keeps a send-time stamp internally (what Lite05/06 put on the + // wire anyway). Typed readers never see it. #4822's internal `Option` + // replaces this. + pub(crate) fn scale(&self) -> Timescale { + self.timescale.unwrap_or(Timescale::MILLI) + } + /// Set how old a non-latest group may get before eviction, returning `self` for chaining. pub fn with_max_age(mut self, max_age: impl Into>) -> Self { self.max_age = max_age.into(); @@ -1380,6 +1390,11 @@ impl Producer { &self.name } + /// The properties this track was created with. + pub(crate) fn info(&self) -> &Info { + &self.info + } + /// The parent broadcast this track belongs to. pub fn broadcast(&self) -> &broadcast::Info { &self.broadcast @@ -1445,7 +1460,7 @@ impl Producer { let meter = self.stats.meter(); let mut state = self.modify()?; // Normalize into the track's timescale, like frames (see `group::Producer::create_frame`). - let timescale = state.info.as_ref().unwrap().timescale; + let timescale = state.info.as_ref().unwrap().scale(); let timestamp = timestamp.convert(timescale).map_err(|_| Error::TimestampMismatch)?; let sequence = match state.max_sequence { Some(s) => s.checked_add(1).ok_or(coding::BoundsExceeded)?, @@ -1488,7 +1503,7 @@ impl Producer { let meter = self.stats.meter(); let mut state = self.modify()?; // Normalize into the track's timescale, like frames (see `group::Producer::create_frame`). - let timescale = state.info.as_ref().unwrap().timescale; + let timescale = state.info.as_ref().unwrap().scale(); let timestamp = timestamp.convert(timescale).map_err(|_| Error::TimestampMismatch)?; if let Some(fin) = state.final_sequence && sequence >= fin diff --git a/rs/moq-net/src/model/typed.rs b/rs/moq-net/src/model/typed.rs new file mode 100644 index 0000000000..6638ff5e47 --- /dev/null +++ b/rs/moq-net/src/model/typed.rs @@ -0,0 +1,257 @@ +//! Variant B mock-up: the writer's timing and timescale as a type parameter. +//! +//! A writer names its timing when it creates the track: [`Untimed`], or a scale such as +//! [`Milli`], [`Micro`], or [`Hz<90_000>`](Hz). The track declares that scale, and +//! the writer's methods take a [`crate::Timed`] or a bare payload to match, so neither the +//! timedness nor the declared scale can disagree with the writes. +//! +//! Readers are unchanged from variant A: a received track's timescale is only learned at +//! runtime, so [`track::Subscriber::timing`](crate::track::Subscriber::timing) still splits +//! it. Relays keep the untyped [`crate::track::Producer`], whose scale is runtime state. +//! +//! Landing this variant would make the core handles generic (`track::Producer`) rather +//! than wrapping them; the wrappers here keep both variants compiling side by side. + +use std::marker::PhantomData; +use std::sync::Arc; + +use crate::{Error, IntoBytes, Result, Timescale, Timestamp, broadcast}; + +mod sealed { + pub trait Sealed {} +} + +/// A writer's timing, fixed when it creates the track. +pub trait Timing: sealed::Sealed + Send + Sync + 'static { + /// The timescale the track declares, or `None` for an untimed track. + const TIMESCALE: Option; +} + +/// A timing whose frames carry timestamps. +pub trait Scale: Timing {} + +/// No timestamps: frames are bare payloads. +pub struct Untimed; + +/// `N` timestamp units per second. +pub struct Hz; + +/// Millisecond timestamps. +pub type Milli = Hz<1_000>; + +/// Microsecond timestamps. +pub type Micro = Hz<1_000_000>; + +/// Nanosecond timestamps. +pub type Nano = Hz<1_000_000_000>; + +impl sealed::Sealed for Untimed {} +impl Timing for Untimed { + const TIMESCALE: Option = None; +} + +impl sealed::Sealed for Hz {} +impl Timing for Hz { + // An invalid `N` (zero, or past the varint range) fails to compile. + const TIMESCALE: Option = match Timescale::new(N) { + Ok(scale) => Some(scale), + Err(_) => panic!("invalid timescale"), + }; +} +impl Scale for Hz {} + +impl broadcast::Producer { + /// Produce a track whose writes are typed by `T`, which also sets its timescale. + /// + /// Refused with [`Error::TimestampMismatch`] when `info` declares a different timescale. + pub fn create_typed( + &self, + name: impl Into>, + info: impl Into>, + ) -> Result> { + let mut info = info.into().unwrap_or_default(); + if info.timescale.is_some() && info.timescale != T::TIMESCALE { + return Err(Error::TimestampMismatch); + } + info.timescale = T::TIMESCALE; + Ok(track::Producer { + inner: self.create_track(name, info)?, + _timing: PhantomData, + }) + } +} + +/// Typed track writers. +pub mod track { + use super::*; + + /// Writes a track whose timing is `T`. + pub struct Producer { + pub(super) inner: crate::track::Producer, + pub(super) _timing: PhantomData, + } + + impl Producer { + /// The track's name, unique within its broadcast. + pub fn name(&self) -> &str { + self.inner.name() + } + + /// Create a group with the next sequence number. + pub fn append_group(&self) -> Result> { + Ok(super::group::Producer::new(self.inner.append_group()?)) + } + + /// Create a group with the given sequence number. + pub fn create_group(&self, info: crate::group::Info) -> Result> { + Ok(super::group::Producer::new(self.inner.create_group(info)?)) + } + + /// End the track after its last group. + pub fn finish(&self) -> Result<()> { + self.inner.finish() + } + + /// A cheap, cloneable handle to subscribe with. + pub fn consume(&self) -> crate::track::Consumer { + self.inner.consume() + } + } + + impl Producer { + /// Write a single-frame group. + pub fn write_frame(&mut self, frame: crate::Timed) -> Result<()> { + self.inner.write_frame(frame.at, frame.value) + } + + /// Append a datagram with the next sequence number, returning the sequence. + pub fn append_datagram(&mut self, datagram: crate::Timed) -> Result { + self.inner.append_datagram(datagram.at, datagram.value) + } + } + + impl Producer { + /// Write a single-frame group. + pub fn write_frame(&mut self, payload: B) -> Result<()> { + self.inner.write_frame(Timestamp::now(), payload) + } + + /// Append a datagram with the next sequence number, returning the sequence. + pub fn append_datagram(&mut self, payload: B) -> Result { + self.inner.append_datagram(Timestamp::now(), payload) + } + } +} + +/// Typed group writers. +pub mod group { + use super::*; + use crate::frame; + + /// Writes a group of a track whose timing is `T`. + pub struct Producer { + inner: crate::group::Producer, + _timing: PhantomData, + } + + impl Producer { + pub(super) fn new(inner: crate::group::Producer) -> Self { + Self { + inner, + _timing: PhantomData, + } + } + + /// Mark the group complete. + pub fn finish(&self) -> Result<()> { + self.inner.finish() + } + + /// Abort the group with `err`. + pub fn abort(self, err: Error) -> Result<()> { + self.inner.abort(err) + } + } + + impl std::ops::Deref for Producer { + type Target = crate::group::Info; + + fn deref(&self) -> &crate::group::Info { + &self.inner + } + } + + impl Producer { + /// Write a whole frame. + pub fn write_frame(&mut self, frame: crate::Timed) -> Result<()> { + self.inner.write_frame(frame.at, frame.value) + } + + /// Open a frame of `size` bytes captured `at`, to write in chunks. + pub fn create_frame(&mut self, size: u64, at: Timestamp) -> Result> { + self.inner.create_frame(frame::Info { size, timestamp: at }) + } + } + + impl Producer { + /// Write a whole frame. + pub fn write_frame(&mut self, payload: B) -> Result<()> { + self.inner.write_frame(Timestamp::now(), payload) + } + + /// Open a frame of `size` bytes, to write in chunks. + pub fn create_frame(&mut self, size: u64) -> Result> { + self.inner.create_frame(frame::Info { + size, + timestamp: Timestamp::now(), + }) + } + } +} + +#[cfg(test)] +mod tests { + use bytes::Bytes; + + use super::*; + use crate::{Timed, track as model}; + + #[tokio::test] + async fn the_type_sets_the_timescale() { + let broadcast = broadcast::Info::default().produce(); + let mut video = broadcast.create_typed::>("video", None).unwrap(); + video + .write_frame(Timed::new(&b"key"[..], Timestamp::from_millis(1).unwrap())) + .unwrap(); + + let model::Timing::Timed(mut reader) = video.consume().subscribe(None).await.unwrap().timing() else { + panic!("a scaled writer makes a timed track"); + }; + assert_eq!(reader.timescale(), Timescale::new(90_000).unwrap()); + let mut group = reader.recv_group().await.unwrap().unwrap(); + assert_eq!(group.read_frame().await.unwrap().unwrap().timestamp.value(), 90); + } + + #[tokio::test] + async fn an_untimed_writer_makes_an_untimed_track() { + let broadcast = broadcast::Info::default().produce(); + let mut chat = broadcast.create_typed::("chat", None).unwrap(); + chat.write_frame(&b"hi"[..]).unwrap(); + + let model::Timing::Untimed(mut reader) = chat.consume().subscribe(None).await.unwrap().timing() else { + panic!("an untimed writer makes an untimed track"); + }; + let mut group = reader.recv_group().await.unwrap().unwrap(); + assert_eq!(group.read_frame().await.unwrap().unwrap(), Bytes::from_static(b"hi")); + } + + #[test] + fn a_conflicting_info_is_refused() { + let broadcast = broadcast::Info::default().produce(); + let info = model::Info::default().with_timescale(Timescale::MILLI); + assert!(matches!( + broadcast.create_typed::("video", info), + Err(Error::TimestampMismatch) + )); + } +} diff --git a/rs/moq-net/tests/timing.rs b/rs/moq-net/tests/timing.rs new file mode 100644 index 0000000000..e10fd2260b --- /dev/null +++ b/rs/moq-net/tests/timing.rs @@ -0,0 +1,65 @@ +//! A track's timing survives the wire: readers learn it when the subscription is accepted. +//! +//! Wires that can't declare a timescale (pre-Lite05) deliver every track untimed; Lite05 +//! and Lite06 always declare one, so an untimed track goes out timed; moq-transport +//! declares TIMESCALE only for a timed track. + +mod support; + +use std::time::Duration; + +use moq_net::{Hop, Timescale, Version, track}; +use support::harness::{MockConnectOptions, connect_mock}; + +const TEST_TIMEOUT: Duration = Duration::from_secs(10); + +fn produce_origin(hop: u64) -> moq_net::origin::Producer { + let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(Hop::new(hop).unwrap())); + tokio::spawn(support::harness::run(driver)); + producer +} + +/// The scale each track arrives with: `(timed track, untimed track)`. +async fn received(version: &str) -> (Option, Option) { + let publisher = produce_origin(1); + let consumer_origin = produce_origin(2); + + let broadcast = publisher.create_broadcast("room").unwrap(); + let info = track::Info::default().with_timescale(Timescale::MICRO); + let _video = broadcast.create_track("video", info).unwrap().timed().unwrap(); + let _chat = broadcast.create_track("chat", None).unwrap().untimed().unwrap(); + broadcast.announce(Default::default()).unwrap(); + + let mut options = MockConnectOptions::new(version.parse::().unwrap()); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(consumer_origin.clone()); + let _pair = connect_mock(options).await; + + let consumer = consumer_origin.consume(); + consumer.routed("room").await.unwrap(); + let remote = consumer.request_broadcast("room").await.unwrap(); + + let mut scales = Vec::new(); + for name in ["video", "chat"] { + let subscriber = remote.track(name).unwrap().subscribe(None).await.unwrap(); + scales.push(match subscriber.timing() { + track::Timing::Timed(timed) => Some(timed.timescale()), + track::Timing::Untimed(_) => None, + }); + } + (scales[0], scales[1]) +} + +#[tokio::test] +async fn timing_is_learned_from_the_wire() { + tokio::time::timeout(TEST_TIMEOUT, async { + assert_eq!(received("moq-lite-03").await, (None, None)); + assert_eq!( + received("moq-lite-05").await, + (Some(Timescale::MICRO), Some(Timescale::MILLI)) + ); + assert_eq!(received("moq-transport-17").await, (Some(Timescale::MICRO), None)); + }) + .await + .expect("timed out"); +}