diff --git a/quest/m1/README.md b/quest/m1/README.md index 19d5d1a11a7..0b1e380cd21 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -154,7 +154,6 @@ removed). - [JS FETCH timeout test](/quest/m1/js-fetch-timeout-test.md) - a JS joining FETCH still waiting on its SUBSCRIBE at the deadline is refused with TIMEOUT - [JS joins at the current group](/quest/m1/ietf-js-join.md) - a late JS moq-transport subscriber gets the group in progress whole through a joining FETCH or fill, as Rust does - [IETF fill timeout](/quest/m1/ietf-fill-timeout.md) - Rust honors native FILL_TIMEOUT to bound upstream waiting on drafts 18 and 22, alongside local content-age enforcement -- [Catalog estimates settle](/quest/m1/catalog-estimates.md) - hang quantizes jitter and delay rises, and MSF publishes once its estimates are made and then only for track changes - [IETF takeover](/quest/m1/ietf-takeover.md) - carry the shared takeover contract through the negotiated IETF epoch extension - [Identity and recovery guide](/quest/m1/identity-guide.md) - explain epochs, pinned catalogs, lazy jobs, restart and version compatibility as they exist; takeover adds its own section - [Native rung disable](/quest/m1/simulcast-rung-disable.md) - OBS stops sending the top renditions its grant cannot fund, as the Rust/FFI publishers do, and a shaped uplink through moq-ffi proves the resume diff --git a/quest/m1/catalog-estimates.md b/quest/m1/catalog-estimates.md deleted file mode 100644 index ecd245bbc1a..00000000000 --- a/quest/m1/catalog-estimates.md +++ /dev/null @@ -1,59 +0,0 @@ -# [M] Catalog estimates settle before they churn - -## Goal - -An unhinted moq-mux publisher's catalogs stop changing for small estimate -creep. Hang republishes `jitter` and `delay` only when they cross a coarse -step. The MSF catalog publishes once its estimates are made and then changes -only when the track list or a track's identity does. MSF-01 §5.3 fixes a -declared track's attributes. - -## Plan - -From [#7067](https://github.com/moq-dev/moq/issues/7067): an SDI encoder -through moq-c published 68 catalog versions in 32 s, and 16 even with every -hint from [#7068](https://github.com/moq-dev/moq/pull/7068). Of the ten that -weren't structural, eight were `delay` rising in ~8 ms steps. -[#6595](https://github.com/moq-dev/moq/pull/6595) already measures from -timestamps. `delay` still follows the cross-track write order, which tracks -runtime encode latency. It only rises, at most once per `ESTIMATE_WINDOW` -(`rs/moq-mux/src/catalog/tracks.rs:655-700`). - -Decided 2026-10-28 (maintainer): - -- **`jitter` and `delay` stay in hang.** @moq/watch, moq-audio and the TS - export use them as latency floors, and a receiver can't measure cross-track - skew itself. Hang keeps updating them in place; only a decoder identity - change publishes a new track - ([#7110](https://github.com/moq-dev/moq/pull/7110)). -- **Hang quantizes the rise.** A new `jitter` or `delay` republishes only when - it crosses the next multiple of the track's frame duration. Measurement is - unchanged. -- **MSF deduplicates.** `emit` (`rs/moq-msf` producer, `catalog/producer.rs:221-243`) - skips a group whose serialized catalog matches the previous one. -- **MSF waits for estimates, then freezes them.** MSF publishes once every - track with an unhinted estimate has closed its first group and carried - `RATE_WINDOW` (1 s, `catalog/estimate.rs:7`) of media, whichever is later. - The minimum duration matters because bitrate is measured over that window: - 20 ms audio groups would otherwise freeze it absent. Its estimated fields - then stay as first declared, and MSF republishes only for track-list or - identity changes. A track whose every estimated field is hinted publishes - at once. moq-c can hint bitrate and framerate but not jitter or delay - ([#7068](https://github.com/moq-dev/moq/pull/7068)), so its tracks wait. -- **No opt-out.** Quantizing and freezing remove the churn without new config - surface on fields players rely on. -- The HLS pointer goes away with - [time-addressed consumers](/quest/m1/time-addressing/consumers.md), so it is - out of scope here. - -Tests: a measured `delay` creeping in 1 ms steps republishes hang once per -frame-duration step. Identical MSF content publishes once. An unhinted MSF -catalog appears once the first groups close and a second of media has -passed, carries a bitrate for 20 ms audio groups, and keeps its estimates. - -Public API: none. Wire: fewer catalog groups; MSF estimate fields frozen at -first declaration. - -## Closes - -- [#7067](https://github.com/moq-dev/moq/issues/7067) - a catalog describes what a track is: keep channel measurements from churning it diff --git a/rs/moq-c/src/api.rs b/rs/moq-c/src/api.rs index 81a8dd8190b..628fb5aad0d 100644 --- a/rs/moq-c/src/api.rs +++ b/rs/moq-c/src/api.rs @@ -206,7 +206,8 @@ pub struct moq_video_init { /// /// Zero the struct and set only the `has_*` flags you want. Hints fill gaps the /// bitstream leaves (especially bitrate); a value the stream detects later wins -/// for dimensions. +/// for dimensions. The MSF catalog declares a bitrate or frame rate only when +/// it is hinted here. #[repr(C)] #[derive(Clone, Copy, Default)] pub struct moq_video_hint { diff --git a/rs/moq-ffi/src/media.rs b/rs/moq-ffi/src/media.rs index c3b5397aba8..8d8bc3e805a 100644 --- a/rs/moq-ffi/src/media.rs +++ b/rs/moq-ffi/src/media.rs @@ -228,6 +228,7 @@ pub struct MoqDatagram { /// Caller-provided video catalog fields for [`MoqVideoInit`]. /// /// Every field is optional and fills only a gap the stream leaves; a value the stream detects wins. +/// The MSF catalog declares a bitrate or frame rate only when it is hinted here. /// Publishing the catalog before the first keyframe needs at least the codec, which comes from the /// [`MoqVideoInit`] format. Audio has no equivalent: it resolves entirely from its init bytes. #[derive(Clone, Default, uniffi::Record)] diff --git a/rs/moq-mux/src/catalog/data.rs b/rs/moq-mux/src/catalog/data.rs index c62b4199327..7279138e414 100644 --- a/rs/moq-mux/src/catalog/data.rs +++ b/rs/moq-mux/src/catalog/data.rs @@ -45,7 +45,7 @@ pub(crate) struct Listing { trait Owned: Send + Sync { fn name(&self) -> &str; fn clock(&self) -> crate::Clock; - fn estimate(&mut self, estimate: super::Estimate) -> crate::Result<()>; + fn estimate(&mut self, reading: super::Reading) -> crate::Result<()>; } impl> Owned for Rendition { @@ -55,8 +55,8 @@ impl> Owned for Rendition { fn clock(&self) -> crate::Clock { Rendition::clock(self) } - fn estimate(&mut self, estimate: super::Estimate) -> crate::Result<()> { - Rendition::estimate(self, estimate) + fn estimate(&mut self, reading: super::Reading) -> crate::Result<()> { + Rendition::estimate(self, reading) } } @@ -111,6 +111,6 @@ impl Listing { // media track behind it by however fast it runs. It was written no later than now. self.estimator.order(captured.min(now)); } - self.rendition.estimate(self.estimator.estimate()) + self.rendition.estimate(self.estimator.reading()) } } diff --git a/rs/moq-mux/src/catalog/estimate.rs b/rs/moq-mux/src/catalog/estimate.rs index 31d5ab74775..bec51470bf3 100644 --- a/rs/moq-mux/src/catalog/estimate.rs +++ b/rs/moq-mux/src/catalog/estimate.rs @@ -190,6 +190,14 @@ impl Estimator { self.jitter.max = self.jitter.max.max(duration); } + /// Everything measured so far, with what a catalog rendition needs to decide when to publish it. + pub(crate) fn reading(&self) -> Reading { + Reading { + estimate: self.estimate(), + period: self.framerate.shortest, + } + } + /// Everything measured so far. /// /// Catalog-owned [`container::Producer`](crate::container::Producer) handles publish this @@ -204,6 +212,21 @@ impl Estimator { } } +/// An [`Estimate`] with the measurements that pace its publishing. +#[derive(Clone, Debug, Default)] +pub(crate) struct Reading { + pub estimate: Estimate, + /// The shortest advance between frames, which a published `jitter` or `delay` rises in whole + /// steps of when no framerate says the frame duration. + pub period: Option, +} + +impl From for Reading { + fn from(estimate: Estimate) -> Self { + Self { estimate, period: None } + } +} + /// A frame time as scale-free nanoseconds. /// /// Timestamps reaching an estimator can carry different timescales: a track normalizes them into @@ -307,10 +330,16 @@ struct Framerate { /// Frames written since the window opened. frames: u64, max: Option, + /// The shortest advance of the highest timestamp: a frame's duration, or a reordered group's + /// span of them. + shortest: Option, } impl Framerate { fn write(&mut self, ts: u128) { + if let Some(advance) = self.high.and_then(|high| elapsed(high, ts)) { + self.shortest = Some(self.shortest.map_or(advance, |shortest| shortest.min(advance))); + } self.high = Some(self.high.map_or(ts, |high| high.max(ts))); self.frames += 1; } @@ -349,6 +378,7 @@ impl Framerate { fn discontinuity(&mut self) { *self = Self { max: self.max, + shortest: self.shortest, ..Self::default() }; } diff --git a/rs/moq-mux/src/catalog/mod.rs b/rs/moq-mux/src/catalog/mod.rs index 4016f61b8e9..3dde971992e 100644 --- a/rs/moq-mux/src/catalog/mod.rs +++ b/rs/moq-mux/src/catalog/mod.rs @@ -40,6 +40,7 @@ pub use consumer::Consumer; pub use data::IntoRendition; pub(crate) use data::Listing; pub use entry::Entry; +pub(crate) use estimate::Reading; pub use estimate::{Estimate, Estimator}; pub use format::*; pub use producer::{Config, Guard, Producer}; diff --git a/rs/moq-mux/src/catalog/producer.rs b/rs/moq-mux/src/catalog/producer.rs index e0c4a0315ce..96439aa5fe1 100644 --- a/rs/moq-mux/src/catalog/producer.rs +++ b/rs/moq-mux/src/catalog/producer.rs @@ -53,6 +53,9 @@ struct State { /// The last revision [`Producer::fresh_track`] handed out per base name, so a successor never /// reuses a name. revisions: std::collections::HashMap, + + /// What the MSF catalog declares. + msf: Msf, } impl State { @@ -109,6 +112,109 @@ impl Throttle { } } +/// The MSF catalog's publish state. +/// +/// MSF declares only the estimates a publisher hinted, never measured ones: a demand-driven +/// publisher measures nothing until a viewer subscribes, and an MSF viewer finds its tracks only +/// through this catalog. MSF-01 §5.3 fixes a declared track's attributes, so each track keeps the +/// estimates it was first declared with, and the MSF catalog changes only when the track list or a +/// track's identity does. The hang catalog keeps updating its estimates in place. +struct Msf { + /// The hinted estimates of each listed rendition that measures its own, keyed like + /// [`State::owned`]. MSF declares these in place of what the catalog carries. + hints: HashMap<(std::any::TypeId, String), Hint>, + /// The estimates each MSF track was first declared with, by namespace and name. + declared: HashMap<(Option, String), Declared>, + /// The last MSF catalog written, so an identical one isn't written again. + last: Option, +} + +impl Default for Msf { + fn default() -> Self { + Self { + hints: HashMap::new(), + declared: HashMap::new(), + last: None, + } + } +} + +/// A rendition's hinted estimates, and how to write them into its config. +struct Hint { + estimate: super::Estimate, + apply: fn(&mut Catalog, &str, super::Estimate), +} + +impl Msf { + /// The MSF catalog to write for `catalog`, or `None` when it matches the last one written. + fn render(&mut self, catalog: &Catalog, base: &moq_net::Path) -> crate::Result> { + let mut msf = if self.hints.is_empty() { + to_msf(catalog, base) + } else { + let mut hinted = catalog.clone(); + for ((_, name), hint) in &self.hints { + (hint.apply)(&mut hinted, name, hint.estimate.clone()); + } + to_msf(&hinted, base) + }; + // Rebuilt from the listed tracks, so a removed track's declaration goes with it. + let mut declared = HashMap::with_capacity(msf.tracks.len()); + for track in &mut msf.tracks { + let key = (track.namespace.clone(), track.name.clone()); + let first = self.declared.remove(&key).unwrap_or_else(|| Declared::of(track)); + first.apply(track); + declared.insert(key, first); + } + self.declared = declared; + + // The MSF catalog is derived from our own types, so a serialize failure means an extension + // broke the shape; report it like any other JSON failure rather than panicking. + let json = msf.to_json().map_err(moq_json::Error::from)?; + if self.last.as_ref() == Some(&json) { + return Ok(None); + } + self.last = Some(json.clone()); + Ok(Some(json)) + } +} + +/// Write `estimate` over the estimates of the `C` config under `name`. +fn apply_hint>( + catalog: &mut Catalog, + name: &str, + estimate: super::Estimate, +) { + if let Some(config) = C::get_mut(catalog, name) { + config.set_estimate(estimate); + } +} + +/// The estimated fields an MSF track was first declared with. +struct Declared { + bitrate: Option, + framerate: Option, + jitter: Option, + delay: Option, +} + +impl Declared { + fn of(track: &moq_msf::Track) -> Self { + Self { + bitrate: track.bitrate, + framerate: track.framerate, + jitter: track.jitter, + delay: track.delay, + } + } + + fn apply(&self, track: &mut moq_msf::Track) { + track.bitrate = self.bitrate; + track.framerate = self.framerate; + track.jitter = self.jitter; + track.delay = self.delay; + } +} + /// The producer's `clock` under a root `clock` of `advertised`: re-based on a changed mapping while /// nothing has `anchored` the clock, and refused once something has. Clearing it is always refused. fn adopt(clock: crate::Clock, anchored: bool, advertised: Option) -> crate::Result { @@ -219,8 +325,8 @@ impl Outputs { std::borrow::Cow::Owned(catalog) } - /// Emit the catalog to hang and MSF. - fn emit(&mut self, catalog: &Catalog) -> crate::Result<()> { + /// Emit the catalog to hang and MSF. Neither writes a catalog identical to its last. + fn emit(&mut self, catalog: &Catalog, msf: &mut Msf) -> crate::Result<()> { let catalog = self.advertise(catalog); let catalog = catalog.as_ref(); // Every consumer refuses a catalog that fails this, so refuse it before it reaches one. @@ -229,12 +335,11 @@ impl Outputs { self.hang .update(moq_net::Timed::from(catalog).at(moq_net::Timestamp::now()))?; - // The MSF catalog is derived from our own types, so a serialize failure means an extension broke - // the shape; report it like any other JSON failure rather than panicking. `abort` below is - // what keeps the tracks consistent if this fails after the hang tracks were updated. - let msf = to_msf(catalog, &self.msf_track.broadcast().path) - .to_json() - .map_err(moq_json::Error::from)?; + // `abort` below is what keeps the tracks consistent if this fails after the hang track was + // updated. + let Some(msf) = msf.render(catalog, &self.msf_track.broadcast().path)? else { + return Ok(()); + }; let mut group = self.msf_track.append_group()?; group.write_frame(moq_net::Timestamp::now(), msf)?; group.finish()?; @@ -452,6 +557,7 @@ impl Producer { anchored, estimates: Throttle::default(), revisions: Default::default(), + msf: Msf::default(), })), hls: config.hls, max_age: config.max_age, @@ -760,28 +866,26 @@ impl Producer { /// Publish the buffered snapshot once, if buffering has just finished with a change staged. pub(super) fn flush_if_ready(&mut self) { - // Snapshot under the lock and emit outside it, since emitting writes to the catalog tracks. - let catalog = { - let mut state = take(&self.current); - let r = &state.reservations; - // Nothing was staged (e.g. every stream was ignored): stay unpublished so a later change - // still triggers the first emit, rather than publishing an empty catalog now. The track - // scan runs last, so releasing each of many reservations stays linear. - if r.published || r.reservers != 0 || !r.pending || !state.lists_track() { - return; - } - let r = &mut state.reservations; - r.pending = false; - r.published = true; - // A snapshot on the wire is a mapping a reader may have copied. - state.anchored = true; - state.estimates.fold(); - state.catalog.clone() - }; - if let Err(err) = self.outputs.emit(&catalog) { + let mut state = take(&self.current); + let r = &state.reservations; + // Nothing was staged (e.g. every stream was ignored): stay unpublished so a later change + // still triggers the first emit, rather than publishing an empty catalog now. The track + // scan runs last, so releasing each of many reservations stays linear. + if r.published || r.reservers != 0 || !r.pending || !state.lists_track() { + return; + } + let r = &mut state.reservations; + r.pending = false; + r.published = true; + // A snapshot on the wire is a mapping a reader may have copied. + state.anchored = true; + state.estimates.fold(); + // Emitted under the lock, as a guard's publish is, since the MSF state belongs to it. + let state = &mut *state; + if let Err(err) = self.outputs.emit(&state.catalog, &mut state.msf) { tracing::error!(%err, "failed to publish the catalog, aborting its tracks"); self.outputs.abort(&err); - take(&self.current).closed = Some(err); + state.closed = Some(err); } } @@ -927,7 +1031,8 @@ impl Producer { if state.closed.is_some() || !state.reservations.published { return; } - if let Err(err) = outputs.emit(&state.catalog) { + let state = &mut *state; + if let Err(err) = outputs.emit(&state.catalog, &mut state.msf) { tracing::error!(%err, "failed to publish an HLS declaration, aborting catalog tracks"); outputs.abort(&err); state.closed = Some(err); @@ -1121,7 +1226,8 @@ impl Guard<'_, E> { self.state.anchored = true; // Folded only once it is on the wire, so a failed emit leaves the estimate held. - self.outputs.emit(&self.state.catalog)?; + let state = &mut *self.state; + self.outputs.emit(&state.catalog, &mut state.msf)?; self.state.estimates.fold(); Ok(()) } @@ -1148,6 +1254,16 @@ impl Guard<'_, E> { self.publish() } + /// Declare `estimate` in MSF for the rendition holding `name` in `C`'s section, in place of + /// whatever it measures. It goes out with the next publish. + pub(super) fn hint>(&mut self, name: &str, estimate: super::Estimate) { + let hint = Hint { + estimate, + apply: apply_hint::, + }; + self.state.msf.hints.insert(owner_key::(name), hint); + } + /// Mark the rendition holding `name` in `C`'s section as listed, so it counts toward the /// initial publish's [track requirement](State::lists_track). pub(super) fn list>(&mut self, name: &str) { @@ -1167,6 +1283,7 @@ impl Guard<'_, E> { /// name is not: nothing would ever reserve it again. pub(super) fn release>(&mut self, name: &str, present: bool) { self.state.owned.remove(&owner_key::(name)); + self.state.msf.hints.remove(&owner_key::(name)); if present { C::remove(&mut self.state.catalog, name); @@ -3363,4 +3480,153 @@ mod test { v.estimate(jitter(2)).unwrap(); assert_eq!(sent(&catalog), Some(0), "the gated publish opened a window"); } + + fn delay(ms: u64) -> super::super::Reading { + super::super::Reading { + estimate: super::super::Estimate::default().with_delay(std::time::Duration::from_millis(ms)), + period: Some(std::time::Duration::from_millis(10)), + } + } + + /// A measured delay creeping a millisecond at a time republishes once per frame duration it + /// crosses, carrying the value measured then. Regression: every rise republished, once a second. + #[tokio::test(start_paused = true)] + async fn a_creeping_delay_republishes_once_per_frame() { + let (_broadcast, catalog, mut renditions) = estimating(&["v"]); + let v = &mut renditions[0]; + let initial = sent(&catalog).unwrap(); + + let mut published = Vec::new(); + for ms in 1..=35 { + // Past the once-a-second window, so only the frame step holds a rise back. + tokio::time::advance(std::time::Duration::from_secs(2)).await; + let before = sent(&catalog); + v.estimate(delay(ms)).unwrap(); + if sent(&catalog) != before { + published.push(wire(&catalog).video.renditions["v"].delay.unwrap().as_millis()); + } + } + assert_eq!(published, [10, 20, 30]); + assert_eq!(sent(&catalog), Some(initial + 3)); + assert_eq!( + catalog.snapshot().video.renditions["v"].delay, + Some(std::time::Duration::from_millis(30)), + "a rise short of a frame is not even staged" + ); + } + + /// A known framerate sets the step, whatever shorter advance the timestamps show. + #[tokio::test(start_paused = true)] + async fn a_framerate_sets_the_delay_step() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = Producer::new(&mut broadcast, Config::default()).unwrap(); + let reserved = catalog.reserve(); + let mut v = reserved.init::("v").unwrap(); + let mut config = VideoConfig::new(hang::catalog::VideoCodec::VP8); + config.framerate = Some(25.0); + v.set(config).unwrap(); + drop(reserved); + + v.estimate(delay(39)).unwrap(); + assert_eq!(wire(&catalog).video.renditions["v"].delay, None); + v.estimate(delay(40)).unwrap(); + assert_eq!( + wire(&catalog).video.renditions["v"].delay, + Some(std::time::Duration::from_millis(40)) + ); + } + + /// How many MSF catalogs have gone out on the wire. + fn msf_sent(catalog: &Producer) -> Option { + catalog.outputs.msf_track.largest().map(|position| position.group) + } + + /// The last MSF catalog on the wire. + fn msf_wire(catalog: &Producer) -> moq_msf::Catalog<()> { + let state = take(&catalog.current); + moq_msf::Catalog::from_str(state.msf.last.as_deref().expect("a published MSF catalog")).unwrap() + } + + /// A change MSF doesn't carry republishes hang but not MSF. + #[test] + fn identical_msf_content_publishes_once() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let mut catalog = Producer::new(&mut broadcast, Config::default()).unwrap(); + catalog + .mutate(|catalog| { + catalog + .video + .renditions + .insert("v".to_string(), VideoConfig::new(hang::catalog::VideoCodec::VP8)); + }) + .unwrap(); + assert_eq!(msf_sent(&catalog), Some(0)); + let hang = sent(&catalog); + + catalog + .mutate(|catalog| { + catalog.video.display = Some(hang::catalog::Display { + width: 1280, + height: 720, + }) + }) + .unwrap(); + assert_ne!(sent(&catalog), hang); + assert_eq!(msf_sent(&catalog), Some(0)); + } + + /// A demand-driven track measures nothing until a viewer subscribes, which an MSF viewer can do + /// only once MSF lists it. So an unhinted track declares MSF at once without estimates, and they + /// stay absent once hang measures them. Regression: MSF waited for the estimates. + #[tokio::test(start_paused = true)] + async fn an_unhinted_track_declares_msf_at_once() { + let (_broadcast, catalog, mut renditions) = estimating(&["v"]); + let v = &mut renditions[0]; + assert_eq!(msf_sent(&catalog), Some(0), "no media has been written"); + let track = &msf_wire(&catalog).tracks[0]; + assert_eq!((track.bitrate, track.framerate), (None, None)); + assert_eq!((track.jitter, track.delay), (None, None)); + + // 25 fps in 2 s groups. + let mut estimator = super::super::Estimator::new(); + for i in 0..=150u64 { + let ts = moq_net::Timestamp::from_micros(i * 40_000).unwrap(); + if i % 50 == 0 { + estimator.cut(Some(ts)); + } + estimator.write(ts, 1_000); + v.estimate(estimator.reading()).unwrap(); + } + let hang = &wire(&catalog).video.renditions["v"]; + assert!(hang.bitrate.is_some()); + assert_eq!(hang.framerate, Some(25.0)); + assert_eq!(msf_sent(&catalog), Some(0), "measurements leave MSF alone"); + } + + /// MSF declares the estimates a track hinted, and keeps those it first declared when a hint + /// changes. A field left unhinted stays absent whatever hang measures. + #[tokio::test(start_paused = true)] + async fn a_hinted_track_carries_its_hints() { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = Producer::new(&mut broadcast, Config::default()).unwrap(); + let reserved = catalog.reserve(); + let mut v = reserved.init::("v").unwrap(); + let mut config = VideoConfig::new(hang::catalog::VideoCodec::VP8); + config.bitrate = Some(1_000_000); + config.framerate = Some(30.0); + v.set(config).unwrap(); + drop(reserved); + assert_eq!(msf_sent(&catalog), Some(0)); + let track = &msf_wire(&catalog).tracks[0]; + assert_eq!((track.bitrate, track.framerate), (Some(1_000_000), Some(30.0))); + + v.estimate(delay(40)).unwrap(); + v.supply_bitrate(2_000_000).unwrap(); + let hang = &wire(&catalog).video.renditions["v"]; + assert_eq!(hang.delay, Some(std::time::Duration::from_millis(40))); + assert_eq!(hang.bitrate, Some(2_000_000)); + assert_eq!(msf_sent(&catalog), Some(0)); + let track = &msf_wire(&catalog).tracks[0]; + assert_eq!((track.bitrate, track.delay), (Some(1_000_000), None)); + } } diff --git a/rs/moq-mux/src/catalog/tracks.rs b/rs/moq-mux/src/catalog/tracks.rs index ff8aca1ff8b..95b1a8cf1df 100644 --- a/rs/moq-mux/src/catalog/tracks.rs +++ b/rs/moq-mux/src/catalog/tracks.rs @@ -171,7 +171,8 @@ impl RenditionConfig for hang::catalog::BinaryConfig { /// dimensions from an SPS, ...) always wins, so a detected change just updates the catalog. A /// [`codec`](Self::codec) alone is enough to publish the catalog before the first keyframe (the /// decoder config then arrives in band). Use it for fields the stream can't reveal (bitrate) or to -/// skip that wait. +/// skip that wait. The MSF catalog declares only hinted estimates (`bitrate`, `framerate`, `jitter`, +/// `delay`), never measured ones, so an MSF track carries only those set here. /// /// Video importers take one because they build the [`VideoConfig`](hang::catalog::VideoConfig) /// themselves, out of the bitstream, so the caller never holds the config to set these on. An audio @@ -367,6 +368,21 @@ impl RenditionConfig for hang::catalog::TextConfig { } } +/// How far short of a whole frame duration a value still counts as reaching it, as a fraction of the +/// period: a reorder of exactly one frame must not miss its step to rounding in the frame duration. +const STEP_SLACK: u32 = 100; + +/// `measured` if it reaches a whole `period` beyond the one `published` sits in, else `published`. +fn quantize(published: Option, measured: Option, period: Duration) -> Option { + let step = + |value: Option| value.map_or(0, |value| (value + period / STEP_SLACK).as_nanos() / period.as_nanos()); + if step(measured) > step(published) { + measured + } else { + published + } +} + /// Refuse an edit that redefines rendition `name`'s track identity in place: a publisher whose /// decoder identity changed publishes under a new name instead. fn check_identity(previous: &Catalog, next: &Catalog, name: &str) -> crate::Result<()> { @@ -549,7 +565,7 @@ pub(crate) struct Rendition> { /// SPS) holds the handle without a catalog entry, and drops without a spurious removal. present: bool, /// The fields the config carried at [`set`](Self::set): authoritative, never overwritten by - /// detection. Whatever is absent here is what detection fills. + /// detection. Whatever is absent here is what detection fills, and stays absent from MSF. supplied: Estimate, /// The last estimate handed to [`estimate`](Self::estimate), kept so a config set afterwards /// starts from whatever was already measured. @@ -632,9 +648,9 @@ impl> Rendition { /// /// Whatever [`Estimate`] fields `config` already carries are authoritative and left alone; the /// rest are filled by [`estimate`](Self::estimate), seeded with anything already measured before - /// the rendition existed (a dirty start or a B-frame reorder). A caller who wants to pre-empt - /// detection sets the field on the config, or (for a config an importer builds out of the - /// bitstream) hands the importer a hint like [`VideoHint`]. + /// the rendition existed (a dirty start or a B-frame reorder). Only the authoritative ones reach + /// the MSF catalog. A caller who wants to pre-empt detection sets the field on the config, or (for + /// a config an importer builds out of the bitstream) hands the importer a hint like [`VideoHint`]. pub(crate) fn set(&mut self, mut config: C) -> crate::Result<()> { let supplied = config.estimate(); let resolved = Self::resolved(&supplied, &self.detected); @@ -650,6 +666,9 @@ impl> Rendition { self.present = true; *guard = next; guard.list::(&self.name); + if C::detects() { + guard.hint::(&self.name, supplied.clone()); + } guard.commit()?; } self.supplied = supplied; @@ -679,15 +698,17 @@ impl> Rendition { /// Publish a measured [`Estimate`], filling only the fields the config didn't supply at /// [`set`](Self::set). /// - /// Mint the estimate from an [`Estimator`](super::Estimator), usually the one owned by a + /// Mint the reading from an [`Estimator`](super::Estimator), usually the one owned by a /// [`container::Producer`](crate::container::Producer). Call it after every write: an estimate - /// that resolves to what the catalog already carries doesn't republish it, and a `jitter` or - /// `delay` rise republishes at most once a second (see - /// [`Guard::commit_estimate`](super::Guard::commit_estimate)), going out on a later call. + /// that resolves to what the catalog already carries doesn't republish it. A `jitter` or `delay` + /// rise republishes only once it reaches the next whole frame duration above the published value, + /// and then at most once a second (see [`Guard::commit_estimate`](super::Guard::commit_estimate)), + /// going out on a later call. /// /// Calling this before [`set`](Self::set) is not wasted: the measurement is remembered and seeds /// the config once it lands. - pub(crate) fn estimate(&mut self, estimate: Estimate) -> crate::Result<()> { + pub(crate) fn estimate(&mut self, reading: impl Into) -> crate::Result<()> { + let super::Reading { estimate, period } = reading.into(); if !C::detects() { return Ok(()); } @@ -702,6 +723,18 @@ impl> Rendition { let published = config.estimate(); resolved.jitter = resolved.jitter.max(published.jitter); resolved.delay = resolved.delay.max(published.delay); + // Measurement creeps up as the encoder's latency does, a step per frame at worst; only a whole + // frame's rise is worth a republish. Until a period is known the throttle alone paces it. + let period = resolved + .framerate + .filter(|framerate| *framerate > 0.0) + .and_then(|framerate| Duration::try_from_secs_f64(framerate.recip()).ok()) + .or(period) + .filter(|period| !period.is_zero()); + if let Some(period) = period { + resolved.jitter = quantize(published.jitter, resolved.jitter, period); + resolved.delay = quantize(published.delay, resolved.delay, period); + } self.detected = estimate; if self.published.as_ref() == Some(&resolved) { // Still an observation: it releases a rise held back until its window ended. A closed @@ -757,8 +790,15 @@ impl> Rendition { let mut estimate = config.estimate(); estimate.bitrate = Some(bitrate); config.set_estimate(estimate); - self.replace(config)?; - self.supplied.bitrate = Some(bitrate); + let mut supplied = self.supplied.clone(); + supplied.bitrate = Some(bitrate); + let name = self.name.clone(); + let mut guard = self.stage(config)?; + if C::detects() { + guard.hint::(&name, supplied.clone()); + } + guard.commit()?; + self.supplied = supplied; Ok(()) } @@ -1092,7 +1132,7 @@ mod tests { let t = ts(i * 40_000); estimator.cut(Some(t)); estimator.write(t, 100_000); - rendition.estimate(estimator.estimate()).unwrap(); + rendition.estimate(estimator.reading()).unwrap(); } } diff --git a/rs/moq-mux/src/container/fmp4/import.rs b/rs/moq-mux/src/container/fmp4/import.rs index bb676a23ba0..d4bc1eecfd5 100644 --- a/rs/moq-mux/src/container/fmp4/import.rs +++ b/rs/moq-mux/src/container/fmp4/import.rs @@ -99,10 +99,10 @@ enum Rendition { } impl Rendition { - fn estimate(&mut self, estimate: crate::catalog::Estimate) -> crate::Result<()> { + fn estimate(&mut self, reading: crate::catalog::Reading) -> crate::Result<()> { match self { - Self::Video(rendition) => rendition.estimate(estimate), - Self::Audio(rendition) => rendition.estimate(estimate), + Self::Video(rendition) => rendition.estimate(reading), + Self::Audio(rendition) => rendition.estimate(reading), } } } @@ -179,9 +179,9 @@ struct Passthrough { impl Passthrough { fn publish_estimate(&mut self) -> crate::Result<()> { - let estimate = self.estimator.estimate(); - self.claim.update(&self.track.demand(), estimate.bitrate); - self.rendition.estimate(estimate) + let reading = self.estimator.reading(); + self.claim.update(&self.track.demand(), reading.estimate.bitrate); + self.rendition.estimate(reading) } /// Close the open group, so the next fragment opens one at `pending_sequence` or the next. diff --git a/rs/moq-mux/src/container/producer.rs b/rs/moq-mux/src/container/producer.rs index afc6802b9b8..7a7e65e9817 100644 --- a/rs/moq-mux/src/container/producer.rs +++ b/rs/moq-mux/src/container/producer.rs @@ -107,7 +107,7 @@ trait Rendition: Send { fn set(&mut self, config: R) -> crate::Result<()>; fn config(&self) -> crate::Result; fn replace(&mut self, config: R) -> crate::Result<()>; - fn estimate(&mut self, estimate: crate::catalog::Estimate) -> crate::Result<()>; + fn estimate(&mut self, reading: crate::catalog::Reading) -> crate::Result<()>; fn supply_bitrate(&mut self, bitrate: u64) -> crate::Result<()>; fn fresh( &self, @@ -156,8 +156,8 @@ where self.replace(config) } - fn estimate(&mut self, estimate: crate::catalog::Estimate) -> crate::Result<()> { - self.estimate(estimate) + fn estimate(&mut self, reading: crate::catalog::Reading) -> crate::Result<()> { + self.estimate(reading) } fn supply_bitrate(&mut self, bitrate: u64) -> crate::Result<()> { @@ -270,7 +270,7 @@ where fn publish_estimate(&mut self) -> crate::Result<()> { if let Some(rendition) = self.rendition.as_mut() { - rendition.estimate(self.estimator.estimate())?; + rendition.estimate(self.estimator.reading())?; } Ok(()) }