From dadb090a2c2e7c45fe9788851a0659fdd1f0fabf Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 20:16:28 -0800 Subject: [PATCH 01/19] chore(quest): claim quest/m0/encode-control.md Quest-Claim: fd8ff08aeb866861 From 599ea4b1b0637b59cd2f212f6712d08e971cac9d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 21:52:47 -0800 Subject: [PATCH 02/19] wip: ladder --- rs/moq-scene/src/ladder.rs | 486 ++++++++++++++++++++++++++++++++++++ rs/moq-scene/src/lib.rs | 6 + rs/moq-scene/src/publish.rs | 139 +++++------ rs/moq-scene/src/scale.rs | 191 ++++++++++++++ 4 files changed, 742 insertions(+), 80 deletions(-) create mode 100644 rs/moq-scene/src/ladder.rs create mode 100644 rs/moq-scene/src/scale.rs diff --git a/rs/moq-scene/src/ladder.rs b/rs/moq-scene/src/ladder.rs new file mode 100644 index 0000000..6b0877d --- /dev/null +++ b/rs/moq-scene/src/ladder.rs @@ -0,0 +1,486 @@ +//! Simulcast: one picture encoded at several sizes, each its own rendition +//! following its own share of the connection's bandwidth. +//! +//! [`sizes`] lists the rungs: the picture's own size, then smaller ones a +//! viewer on a slower link, or showing it in a corner, picks instead. Each rung +//! is two halves, so a caller can split them across threads: +//! +//! - a [`Gate`] decides, frame by frame, whether the rung encodes and at what +//! bitrate: its reservation on the connection's +//! [`Allocator`](moq_net::bandwidth::Allocator), followed through +//! [`moq_mux::rate`], and whether anyone watches its track; +//! - a [`Rendition`] does what the gate decided: it encodes, retunes, drains +//! its encoder, and flips the catalog's `enabled` for the rung. +//! +//! A rung above another disables when its grant cannot fund it, so the one +//! below keeps its share, and resumes once the grant recovers. Only a gate +//! marked `always` encodes while nobody watches. + +use moq_mux::rate; +use moq_net::bandwidth::{self, Rate}; +use moq_net::track::Demand; + +use crate::Error; + +/// Each rung's size as a fraction of the picture's, largest first. +const SCALES: [(u32, u32); 3] = [(1, 1), (2, 3), (1, 3)]; + +/// The shortest side a smaller rung may have: below it a picture is too small +/// to be worth a rendition of its own. +const MIN_SIDE: u32 = 180; + +/// Bits per pixel per second each rung reserves at most, the default moq's +/// encoders and the JS publisher use: about 4.4 Mb/s for 1080p30. +const BITS_PER_PIXEL: f64 = 0.07; + +/// How many rungs a ladder has at most, so a caller can rank its tracks below +/// every rung of another ladder. +pub const RUNGS: usize = SCALES.len(); + +/// The size an encoder takes for a picture of `size`: even, since the encoder +/// subsamples chroma. An odd picture is scaled down to it. +pub fn encodable(size: moq_video::Size) -> moq_video::Size { + moq_video::Size::new((size.width & !1).max(2), (size.height & !1).max(2)) +} + +/// The rungs a picture of `size` is encoded at, largest first: its own +/// [`encodable`] size, then each smaller one whose shorter side is at least +/// [`MIN_SIDE`]. Each keeps the picture's shape. +pub fn sizes(size: moq_video::Size) -> Vec { + let short = size.width.min(size.height); + SCALES + .iter() + .enumerate() + .filter(|(index, (num, den))| *index == 0 || short * num / den >= MIN_SIDE) + .map(|(_, scale)| scaled(size, *scale)) + .collect() +} + +/// The `index`th rung of a picture of `size`, as [`sizes`] sizes it, however +/// many rungs the ladder was opened with. +fn rung(size: moq_video::Size, index: usize) -> moq_video::Size { + scaled(size, SCALES[index]) +} + +fn scaled(size: moq_video::Size, (num, den): (u32, u32)) -> moq_video::Size { + let scale = |side: u32| (u64::from(side) * u64::from(num) / u64::from(den)) as u32; + encodable(moq_video::Size::new(scale(size.width), scale(size.height))) +} + +/// The most a rung of `size` at `framerate` sends, which it reserves and +/// encodes at while its grant allows. +pub fn ceiling(size: moq_video::Size, framerate: moq_video::FrameRate) -> Rate { + Rate::from_bps((size.pixels() as f64 * framerate.as_f64() * BITS_PER_PIXEL) as u64) +} + +/// The track priority of rung `index` of `count`, the lowest rung at `base`: +/// each rung above it ranks one lower, so the allocator funds the smaller +/// renditions first and a viewer on a slow link always has one. +pub fn priority(base: i8, index: usize, count: usize) -> i8 { + base.saturating_sub((count - 1 - index) as i8) +} + +/// Publish `rendition` on a new track of `broadcast` at `priority`, +/// registered under `reserved`. +pub fn producer( + broadcast: &moq_net::broadcast::Producer, + reserved: moq_mux::catalog::Reserved, + rendition: hang::catalog::VideoConfig, + priority: i8, +) -> Result { + // The suffix `moq_video::encode::Producer::new` derives, which takes no priority. + let suffix = match &rendition.codec { + hang::catalog::VideoCodec::H264(_) => ".avc3", + hang::catalog::VideoCodec::H265(_) => ".hev1", + other => return Err(Error::Unsupported(format!("{other} is not a codec the studio publishes"))), + }; + let track = broadcast.unique_track(suffix, reserved.track_info(priority))?; + Ok(moq_video::encode::Producer::with_track(track, reserved, rendition)?) +} + +/// What a rung does with the next frame, as its [`Gate`] decides. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Step { + /// Encode it at this bitrate. + Encode(Rate), + /// Skip it: nobody watches. The rendition stays advertised. + Idle, + /// Skip it: the grant cannot fund this rung, so it is advertised disabled + /// and viewers pick a smaller one. + Disabled, +} + +/// Decides whether one rung encodes, and how fast, from its share of the +/// connection's bandwidth and whether anyone watches it. +/// +/// Reads only cheap, synchronous state, so it can run every frame on the +/// thread that produces the frames, even while the rung encodes nothing. +pub struct Gate { + allocator: bandwidth::Allocator, + demand: Demand, + /// Held for as long as the rung publishes: dropping it hands the room back. + _reservation: bandwidth::Reservation, + grant: bandwidth::Consumer, + control: rate::Control, + always: bool, +} + +impl Gate { + /// Gate the rung publishing to `demand`'s track, reserving `ceiling` on + /// `allocator`. `lower` is the next rung down's ceiling, below which this + /// one disables; `None` for the lowest rung, which never does. An `always` + /// rung encodes while nobody watches. + pub fn new( + allocator: &bandwidth::Allocator, + demand: Demand, + ceiling: Rate, + lower: Option, + always: bool, + ) -> Self { + let reservation = allocator.reserve(&demand, ceiling); + Self { + allocator: allocator.clone(), + grant: reservation.consume(), + _reservation: reservation, + control: rate::Control::new(policy(ceiling, lower)), + demand, + always, + } + } + + /// What the rung does with the next frame. + pub fn step(&mut self) -> Step { + // A pure policy fed the current grant on every frame, rather than a + // follower woken by changes: the frames already tick this often, and a + // rung that encodes nothing still steps, so it resumes. + self.control.set_probed(self.grant.probed()); + self.control.update(self.grant.peek(), kio::clock::now()); + match self.control.target() { + rate::Target::Disabled => Step::Disabled, + rate::Target::Enabled(rate) if self.always || self.watched() => Step::Encode(rate), + rate::Target::Enabled(_) => Step::Idle, + } + } + + /// Whether anyone subscribes to the rung's track. + pub fn watched(&self) -> bool { + self.demand.wanted().is_some() + } + + /// The rung's share of the connection, as the allocator grants it: `None` + /// while the connection has no estimate. + pub fn grant(&self) -> bandwidth::Consumer { + self.grant.clone() + } + + /// Follow a rendition that now publishes to `demand`'s track at `ceiling`, + /// above a rung at `lower`: its picture changed size. + pub(crate) fn follow(&mut self, demand: Demand, ceiling: Rate, lower: Option) { + let reservation = self.allocator.reserve(&demand, ceiling); + self.grant = reservation.consume(); + self._reservation = reservation; + self.control = rate::Control::new(policy(ceiling, lower)); + self.demand = demand; + } + + /// The track this gate follows. + pub(crate) fn track(&self) -> &str { + self.demand.name() + } +} + +fn policy(ceiling: Rate, lower: Option) -> rate::Policy { + let mut policy = rate::Policy::new(ceiling); + policy.lower = lower; + policy +} + +/// What a [`Rendition`] last did. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum State { + Encoding, + Idle, + Disabled, +} + +/// One rung's track and encoder, doing what its [`Gate`] decides. +/// +/// The encoder opens on the first frame to encode and, unless held, closes +/// while the rung skips frames, so an unwatched rung holds no encoder session. +/// Ends its track cleanly when dropped, with what the encoder still held. +pub struct Rendition { + /// Taken only to abort it. + producer: Option, + config: moq_video::encode::Config, + encoder: Option, + /// Keep the encoder, and its session, while skipping frames. + hold: bool, + state: State, + /// The bitrate the gate last asked for. + target: Rate, + /// The encoder cannot retune, so it stays at its ceiling. + fixed: bool, +} + +impl Rendition { + /// Encode `config`'s pictures onto `producer`'s track. `config.bitrate` is + /// the ceiling. A `hold` rendition keeps its encoder while it skips + /// frames, and resumes with a keyframe. + pub fn new(producer: moq_video::encode::Producer, config: moq_video::encode::Config, hold: bool) -> Self { + let target = config + .bitrate + .unwrap_or_else(|| ceiling(config.size(), config.framerate)); + Self { + producer: Some(producer), + config, + encoder: None, + hold, + state: State::Idle, + target, + fixed: false, + } + } + + /// Open the encoder now rather than on the first frame, so it holds its + /// session from the start, returning the backend's name. + pub fn open(&mut self) -> Result<&str, Error> { + Ok(self.encoder()?.name()) + } + + /// The track's subscriber demand, which moves to a new track when the + /// picture changes size. + pub fn demand(&self) -> Demand { + self.producer().demand() + } + + /// The size this rendition encodes. + pub fn size(&self) -> moq_video::Size { + self.config.size() + } + + /// Do what the gate decided for the next frame. + pub fn apply(&mut self, step: Step) -> Result<(), Error> { + let state = match step { + Step::Encode(_) => State::Encoding, + Step::Idle => State::Idle, + Step::Disabled => State::Disabled, + }; + if self.state == State::Encoding && state != State::Encoding { + self.pause()?; + } + if (self.state == State::Disabled) != (state == State::Disabled) { + let enabled = state != State::Disabled; + self.producer.as_mut().expect("open until it ends").set_enabled(enabled)?; + } + if self.state != State::Encoding + && state == State::Encoding + && let Some(encoder) = &mut self.encoder + { + // A held encoder resumes where a viewer can start decoding. + encoder.cut()?; + } + self.state = state; + if let Step::Encode(target) = step { + self.target = target; + self.retune()?; + } + Ok(()) + } + + /// Encode and publish `frame`, which is [`size`](Self::size). Only after + /// an [`apply`](Self::apply) of [`Step::Encode`]. + pub fn encode(&mut self, frame: &moq_video::Frame) -> Result<(), Error> { + debug_assert_eq!(self.state, State::Encoding, "encoding a skipped frame"); + let encoded = self.encoder()?.encode(frame)?; + self.producer.as_mut().expect("open until it ends").publish(&encoded)?; + Ok(()) + } + + /// Encode pictures of `size` and `color` from the next frame on, at + /// `ceiling`. Publishes what the encoder still holds first; the next + /// encoder moves the rendition to a new track. + pub(crate) fn resize( + &mut self, + size: moq_video::Size, + color: Option, + ceiling: Rate, + ) -> Result<(), Error> { + self.drain()?; + self.config.width = size.width; + self.config.height = size.height; + self.config.color = color; + self.config.bitrate = Some(ceiling); + self.target = ceiling; + self.fixed = false; + Ok(()) + } + + /// End the track, failed with `err` so viewers see why. + pub fn abort(mut self, err: &dyn std::fmt::Display) { + if let Some(producer) = self.producer.take() { + producer.abort(moq_net::Error::Transport(err.to_string())); + } + } + + fn producer(&self) -> &moq_video::encode::Producer { + self.producer.as_ref().expect("open until it ends") + } + + /// The open encoder, opening it at the ceiling and retuning it to the + /// target if it is not. + fn encoder(&mut self) -> Result<&mut moq_video::encode::Encoder, Error> { + if self.encoder.is_none() { + self.encoder = Some(moq_video::encode::Encoder::new(&self.config)?); + self.fixed = false; + self.retune()?; + } + Ok(self.encoder.as_mut().expect("just opened")) + } + + /// Move the open encoder to the target bitrate, unless it cannot retune. + fn retune(&mut self) -> Result<(), Error> { + let Some(encoder) = self.encoder.as_mut().filter(|_| !self.fixed) else { + return Ok(()); + }; + match encoder.set_bitrate(self.target) { + Ok(()) => Ok(()), + // Not fatal: it keeps encoding at its ceiling, and only stops + // following the grant. + Err(moq_video::Error::BitrateUnsupported(name)) => { + tracing::warn!(encoder = name, "the encoder cannot follow the bandwidth grant"); + self.fixed = true; + Ok(()) + } + Err(err) => Err(err.into()), + } + } + + /// Stop encoding until the gate asks again: publish what the encoder still + /// holds, close it unless held, and mark the gap, so a viewer arriving + /// during it is not served the last group as if it were live. + fn pause(&mut self) -> Result<(), Error> { + match self.hold { + true => { + if let (Some(encoder), Some(producer)) = (&mut self.encoder, &mut self.producer) { + producer.publish(&encoder.flush()?)?; + } + } + false => self.drain()?, + } + self.producer.as_mut().expect("open until it ends").discontinuity()?; + Ok(()) + } + + /// Publish what the encoder still holds and close it, so a pipelined + /// backend's last pictures are not lost when the encoder closes. + fn drain(&mut self) -> Result<(), Error> { + if let (Some(encoder), Some(producer)) = (self.encoder.take(), &mut self.producer) { + producer.publish(&encoder.finish()?)?; + } + Ok(()) + } +} + +impl Drop for Rendition { + fn drop(&mut self) { + // A source removed or edited, or a studio closing, ends its tracks + // cleanly rather than leaving subscribers with a dropped one. + let drained = self.drain(); + match (self.producer.take(), drained) { + (Some(mut producer), Ok(())) => { + let _ = producer.finish(); + } + (Some(producer), Err(err)) => producer.abort(moq_net::Error::Transport(err.to_string())), + (None, _) => {} + } + } +} + +/// A picture's rungs, each a [`Gate`] and a [`Rendition`] on one thread. +pub(crate) struct Ladder { + rungs: Vec<(Gate, Rendition)>, + framerate: moq_video::FrameRate, +} + +impl Ladder { + /// Rungs `rungs`, largest first, as [`sizes`] lists them. + pub(crate) fn new(rungs: Vec<(Gate, Rendition)>, framerate: moq_video::FrameRate) -> Self { + Self { rungs, framerate } + } + + /// The size of the largest rung, which the others are fractions of. + pub(crate) fn size(&self) -> moq_video::Size { + self.rungs[0].1.size() + } + + /// Encode pictures of `size` and `color` from the next frame on, keeping + /// the number of rungs. + pub(crate) fn resize(&mut self, size: moq_video::Size, color: Option) -> Result<(), Error> { + for (index, (_, rendition)) in self.rungs.iter_mut().enumerate() { + let size = rung(size, index); + rendition.resize(size, color, ceiling(size, self.framerate))?; + } + Ok(()) + } + + /// Encode `frame` onto every rung its gate lets encode, scaling it to each. + pub(crate) fn encode(&mut self, frame: &moq_video::Frame) -> Result<(), Error> { + let ceilings: Vec = self + .rungs + .iter() + .map(|(_, rendition)| ceiling(rendition.size(), self.framerate)) + .collect(); + for (index, (gate, rendition)) in self.rungs.iter_mut().enumerate() { + let step = gate.step(); + rendition.apply(step)?; + if !matches!(step, Step::Encode(_)) { + continue; + } + let size = rendition.size(); + match frame.size() == size { + true => rendition.encode(frame)?, + false => rendition.encode(&frame.resize(size, &Default::default())?)?, + } + // A new size moves the rendition to a new track, whose demand the + // reservation has to follow. + let demand = rendition.demand(); + if demand.name() != gate.track() { + gate.follow(demand, ceilings[index], ceilings.get(index + 1).copied()); + } + } + Ok(()) + } + + /// End every rung's track, failed with `err`. + pub(crate) fn abort(self, err: &dyn std::fmt::Display) { + for (_, rendition) in self.rungs { + rendition.abort(err); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn rungs_keep_the_shape_down_to_a_floor() { + let size = |width, height| moq_video::Size::new(width, height); + assert_eq!( + sizes(size(1920, 1080)), + [size(1920, 1080), size(1280, 720), size(640, 360)] + ); + // A portrait picture scales by its shorter side too. + assert_eq!(sizes(size(1080, 1920)), [size(1080, 1920), size(720, 1280), size(360, 640)]); + // Rungs under the floor are dropped, and odd sides round down to even. + assert_eq!(sizes(size(641, 361)), [size(640, 360), size(426, 240)]); + assert_eq!(sizes(size(200, 150)), [size(200, 150)]); + } + + #[test] + fn smaller_rungs_rank_first() { + let base = hang::catalog::PRIORITY.video; + let ranks: Vec = (0..3).map(|index| priority(base, index, 3)).collect(); + assert_eq!(ranks, [base - 2, base - 1, base]); + } +} diff --git a/rs/moq-scene/src/lib.rs b/rs/moq-scene/src/lib.rs index 1d2e26a..ca8c17b 100644 --- a/rs/moq-scene/src/lib.rs +++ b/rs/moq-scene/src/lib.rs @@ -20,6 +20,9 @@ //! where the platform allows. //! - [`publish`] gives each local source its own track in the studio's //! broadcast, so viewers can compose the scene themselves. +//! - [`ladder`] encodes a picture at several sizes, each following its own +//! share of the connection's bandwidth, and [`Scaler`] draws the smaller +//! sizes of the program on the GPU. //! //! Every picture is drawn on the program clock the caller ticks //! [`Compositor::render`] with, from whatever each source delivered last: a @@ -30,13 +33,16 @@ pub mod clip; mod collection; mod compositor; mod edit; +pub mod ladder; pub mod publish; +mod scale; pub mod source; pub use clip::Clip; pub use collection::{Audio, Canvas, Collection, Filter, Item, Layer, Rect, Rgba, Scene, Source, Transition, Video}; pub use compositor::{Compositor, FORMAT}; pub use edit::{Change, Draft, Edit, History}; +pub use scale::Scaler; /// The `wgpu` the compositor draws with: `moq_video`'s, so its textures and /// the decoder's are the same type. diff --git a/rs/moq-scene/src/publish.rs b/rs/moq-scene/src/publish.rs index 3fbb6b8..11a40c2 100644 --- a/rs/moq-scene/src/publish.rs +++ b/rs/moq-scene/src/publish.rs @@ -7,15 +7,19 @@ //! their entries as alternatives of one picture, so a camera listed there would //! be played, or packaged for HLS, as a variant of the program. //! -//! A source encodes while it runs, whatever the demand, so a late viewer finds -//! its recent past; MoQ uploads only what a subscriber pulls. +//! A captured source encodes a [`ladder`](crate::ladder) of renditions. The +//! largest encodes while the source runs, whatever the demand, so a late viewer +//! finds its recent past; MoQ uploads only what a subscriber pulls. The smaller +//! ones encode only while watched. use std::collections::BTreeMap; use std::sync::{Arc, Mutex}; +use moq_net::bandwidth::Rate; use serde::{Deserialize, Serialize}; use crate::Error; +use crate::ladder::{self, Gate, Ladder, Rendition}; /// The studio catalog's own root section, namespaced as hang asks of /// application sections. @@ -69,8 +73,9 @@ struct Owners { impl Publish { /// Publish sources into `broadcast`, listed in `catalog`, the one serving it, - /// encoding captured video with `kind`. Each source track claims its bitrate - /// on `bandwidth`, the allocator the program's tracks claim on. + /// encoding captured video with `kind`. Each source rung reserves its + /// ceiling on `bandwidth`, the allocator the program's tracks reserve on, + /// and follows its share. /// /// Pass [`Kind::Hardware`](moq_video::encode::Kind::Hardware) on a machine /// with a hardware encoder, so a session it refuses fails that source's track @@ -138,6 +143,7 @@ impl Drop for Claim { pub(crate) struct Listing<'a> { claim: &'a Claim, kind: moq_video::encode::Kind, + bandwidth: moq_net::bandwidth::Allocator, broadcast: moq_net::broadcast::Producer, studio: Catalog, catalog: moq_mux::catalog::Producer, @@ -150,14 +156,16 @@ impl<'a> Listing<'a> { pub(crate) fn new(claim: &'a Claim, publish: &Publish) -> Result { let mut scratch = moq_net::broadcast::Info::new().produce(); // The studio's clock, so source and program timestamps share one timeline. + // No bandwidth: each rung reserves its own ceiling, which a catalog claim + // would count a second time. let config = moq_mux::catalog::Config::default() .with_clock(publish.catalog.clock()) - .with_bandwidth(publish.bandwidth.clone()) .with_hls(None); let catalog = moq_mux::catalog::Producer::new(&mut scratch, config)?; Ok(Self { claim, kind: publish.kind.clone(), + bandwidth: publish.bandwidth.clone(), broadcast: publish.broadcast.clone(), studio: publish.catalog.clone(), catalog, @@ -203,25 +211,27 @@ impl Drop for Listing<'_> { } } -/// A captured source's video, encoded frame by frame as it is shown. +/// A captured source's video, encoded frame by frame as it is shown, at each +/// rung of a [`ladder`](crate::ladder). pub(crate) struct Video { - /// Taken only to abort or finish it, as the track ends. - producer: Option, - config: moq_video::encode::Config, - /// Opened on the first frame, and reopened when the picture's size or color - /// space changes, which moves the rendition to a new track. - encoder: Option, + /// Taken only to abort it. + ladder: Option, clock: moq_mux::Clock, /// Where the capture's private timeline starts on the studio clock, in /// microseconds, taken at the first frame. epoch: Option, + /// The color space the rungs encode; a picture in another reopens them. + color: Option, } impl Video { - /// Advertise a rendition for pictures of `size`. Opens a throwaway encoder - /// to learn exactly what the real one will write, so a machine that cannot - /// encode another stream, such as a GPU at its hardware session cap, fails - /// here with the reason. + /// Advertise a rendition per rung for pictures of `size`. Opens a + /// throwaway encoder per rung to learn exactly what the real one will + /// write, so a machine that cannot encode another stream, such as a GPU at + /// its hardware session cap, fails here with the reason. + /// + /// The largest rung encodes whatever the demand, so a late viewer finds + /// its recent past; the smaller ones encode only while watched. #[cfg_attr( not(feature = "capture"), allow(dead_code, reason = "only capture sources encode video") @@ -232,22 +242,33 @@ impl Video { framerate: moq_video::FrameRate, color: Option, ) -> Result { - let size = encodable(size); - let mut config = moq_video::encode::Config::new(size.width, size.height, framerate); - config.color = color; - config.kind = listing.kind.clone(); - let rendition = config.probe().await?; - let reserved = listing - .catalog - .reserve() - .with_hls_hint(config.gop.duration(config.framerate)); - let producer = moq_video::encode::Producer::new(listing.broadcast.clone(), reserved, rendition)?; + let sizes = ladder::sizes(size); + let ceilings: Vec = sizes.iter().map(|size| ladder::ceiling(*size, framerate)).collect(); + // Below every rung of the program, which is what viewers come for. + let base = hang::catalog::PRIORITY.video.saturating_sub(ladder::RUNGS as i8); + let mut rungs = Vec::with_capacity(sizes.len()); + for (index, size) in sizes.iter().enumerate() { + let mut config = moq_video::encode::Config::new(size.width, size.height, framerate); + config.color = color; + config.kind = listing.kind.clone(); + config.bitrate = Some(ceilings[index]); + let rendition = config.probe().await?; + let reserved = listing + .catalog + .reserve() + .with_hls_hint(config.gop.duration(config.framerate)); + let priority = ladder::priority(base, index, sizes.len()); + let producer = ladder::producer(&listing.broadcast, reserved, rendition, priority)?; + let rendition = Rendition::new(producer, config, false); + let lower = ceilings.get(index + 1).copied(); + let gate = Gate::new(&listing.bandwidth, rendition.demand(), ceilings[index], lower, index == 0); + rungs.push((gate, rendition)); + } Ok(Self { - producer: Some(producer), - config, - encoder: None, + ladder: Some(Ladder::new(rungs, framerate)), clock: listing.studio.clock(), epoch: None, + color, }) } @@ -263,70 +284,28 @@ impl Video { .get_or_insert_with(|| self.clock.now().as_micros().saturating_sub(captured)); frame.timestamp = moq_net::Timestamp::from_micros((epoch + captured) as u64).map_err(moq_video::Error::from)?; - let size = encodable(frame.size()); + let ladder = self.ladder.as_mut().expect("open until it ends"); + let size = ladder::encodable(frame.size()); let color = frame.surface.color(); - if size != self.config.size() || color.is_some_and(|color| Some(color) != self.config.color) { - self.config.width = size.width; - self.config.height = size.height; - self.config.color = color; - self.drain()?; - } - let encoder = match &mut self.encoder { - Some(encoder) => encoder, - None => self.encoder.insert(moq_video::encode::Encoder::new(&self.config)?), - }; - let encoded = if frame.size() == size { - encoder.encode(frame)? - } else { - encoder.encode(&frame.resize(size, &Default::default())?)? - }; - self.producer.as_mut().expect("open until it ends").publish(&encoded)?; - Ok(()) - } - - /// Publish what the encoder still holds and close it, so a pipelined - /// backend's last pictures are not lost when the encoder is replaced or the - /// track ends. - fn drain(&mut self) -> Result<(), Error> { - if let (Some(encoder), Some(producer)) = (self.encoder.take(), &mut self.producer) { - producer.publish(&encoder.finish()?)?; + if size != ladder.size() || color.is_some_and(|color| Some(color) != self.color) { + self.color = color; + ladder.resize(size, color)?; } - Ok(()) + ladder.encode(frame) } - /// End the track, failed with `err` so viewers see why. + /// End every rung's track, failed with `err` so viewers see why. #[cfg_attr( not(feature = "capture"), allow(dead_code, reason = "only capture sources encode video") )] pub(crate) fn abort(mut self, err: &Error) { - if let Some(producer) = self.producer.take() { - producer.abort(moq_net::Error::Transport(err.to_string())); + if let Some(ladder) = self.ladder.take() { + ladder.abort(err); } } } -impl Drop for Video { - fn drop(&mut self) { - // A capture that ends, or a source removed or edited, ends its track - // cleanly rather than leaving subscribers with a dropped one. - let drained = self.drain(); - match (self.producer.take(), drained) { - (Some(mut producer), Ok(())) => { - let _ = producer.finish(); - } - (Some(producer), Err(err)) => producer.abort(moq_net::Error::Transport(err.to_string())), - (None, _) => {} - } - } -} - -/// The size an encoder takes for a picture of `size`: even, since the encoder -/// subsamples chroma. An odd picture is scaled down to it. -fn encodable(size: moq_video::Size) -> moq_video::Size { - moq_video::Size::new((size.width & !1).max(2), (size.height & !1).max(2)) -} - /// A still picture, published once as the image file it is. pub(crate) struct Still { track: moq_net::track::Producer, diff --git a/rs/moq-scene/src/scale.rs b/rs/moq-scene/src/scale.rs new file mode 100644 index 0000000..cff8861 --- /dev/null +++ b/rs/moq-scene/src/scale.rs @@ -0,0 +1,191 @@ +//! Scale a texture down on the GPU, for the smaller rungs of the program. + +use moq_video::render::wgpu; + +use crate::compositor::FORMAT; + +const SHADER: &str = r#" +@group(0) @binding(0) var image: texture_2d; +@group(0) @binding(1) var image_sampler: sampler; + +struct Varyings { + @builtin(position) position: vec4, + @location(0) uv: vec2, +} + +// One triangle covering the whole target. +@vertex fn vs_main(@builtin(vertex_index) index: u32) -> Varyings { + let corner = vec2(f32((index << 1u) & 2u), f32(index & 2u)); + var out: Varyings; + out.position = vec4(corner * vec2(2.0, -2.0) + vec2(-1.0, 1.0), 0.0, 1.0); + out.uv = corner; + return out; +} + +// Four bilinear taps a quarter of a target pixel from its center, each +// averaging four source texels: a box over the pixel's footprint, so a third +// of the size still averages every texel rather than skipping some. +@fragment fn fs_main(in: Varyings) -> @location(0) vec4 { + let offset = fwidth(in.uv) * 0.25; + let sum = textureSample(image, image_sampler, in.uv + vec2(-offset.x, -offset.y)) + + textureSample(image, image_sampler, in.uv + vec2(offset.x, -offset.y)) + + textureSample(image, image_sampler, in.uv + vec2(-offset.x, offset.y)) + + textureSample(image, image_sampler, in.uv + vec2(offset.x, offset.y)); + return sum * 0.25; +} +"#; + +/// Draws one texture into a smaller one, as [`moq_video::render::export`] +/// takes it. +/// +/// Bound to its source texture, such as the compositor's program, which keeps +/// one texture and redraws it each frame: [`scale`](Self::scale) draws whatever +/// it holds now. +pub struct Scaler { + queue: wgpu::Queue, + device: wgpu::Device, + pipeline: wgpu::RenderPipeline, + bind: wgpu::BindGroup, + output: wgpu::Texture, + view: wgpu::TextureView, +} + +impl Scaler { + /// Scale `source`, a [`FORMAT`] texture created with `TEXTURE_BINDING`, to + /// `size`. + pub fn new(device: &wgpu::Device, queue: &wgpu::Queue, source: &wgpu::Texture, size: moq_video::Size) -> Self { + let layout = device.create_bind_group_layout(&wgpu::BindGroupLayoutDescriptor { + label: Some("moq-scene scale"), + entries: &[ + wgpu::BindGroupLayoutEntry { + binding: 0, + visibility: wgpu::ShaderStages::FRAGMENT, + ty: wgpu::BindingType::Texture { + sample_type: wgpu::TextureSampleType::Float { filterable: true }, + view_dimension: wgpu::TextureViewDimension::D2, + multisampled: false, + }, + count: None, + }, + wgpu::BindGroupLayoutEntry { + binding: 1, + visibility: wgpu::ShaderStages::FRAGMENT, + ty: wgpu::BindingType::Sampler(wgpu::SamplerBindingType::Filtering), + count: None, + }, + ], + }); + let pipeline_layout = device.create_pipeline_layout(&wgpu::PipelineLayoutDescriptor { + label: Some("moq-scene scale"), + bind_group_layouts: &[Some(&layout)], + immediate_size: 0, + }); + let shader = device.create_shader_module(wgpu::ShaderModuleDescriptor { + label: Some("moq-scene scale"), + source: wgpu::ShaderSource::Wgsl(SHADER.into()), + }); + let pipeline = device.create_render_pipeline(&wgpu::RenderPipelineDescriptor { + label: Some("moq-scene scale"), + layout: Some(&pipeline_layout), + vertex: wgpu::VertexState { + module: &shader, + entry_point: Some("vs_main"), + compilation_options: Default::default(), + buffers: &[], + }, + primitive: Default::default(), + depth_stencil: None, + multisample: Default::default(), + fragment: Some(wgpu::FragmentState { + module: &shader, + entry_point: Some("fs_main"), + compilation_options: Default::default(), + targets: &[Some(wgpu::ColorTargetState { + format: FORMAT, + blend: None, + write_mask: wgpu::ColorWrites::ALL, + })], + }), + multiview_mask: None, + cache: None, + }); + let sampler = device.create_sampler(&wgpu::SamplerDescriptor { + label: Some("moq-scene scale"), + mag_filter: wgpu::FilterMode::Linear, + min_filter: wgpu::FilterMode::Linear, + ..Default::default() + }); + let source = source.create_view(&Default::default()); + let bind = device.create_bind_group(&wgpu::BindGroupDescriptor { + label: Some("moq-scene scale"), + layout: &layout, + entries: &[ + wgpu::BindGroupEntry { + binding: 0, + resource: wgpu::BindingResource::TextureView(&source), + }, + wgpu::BindGroupEntry { + binding: 1, + resource: wgpu::BindingResource::Sampler(&sampler), + }, + ], + }); + // What the exporter copies from: a texture it can sample and copy. + let output = device.create_texture(&wgpu::TextureDescriptor { + label: Some("moq-scene scaled"), + size: wgpu::Extent3d { + width: size.width, + height: size.height, + depth_or_array_layers: 1, + }, + mip_level_count: 1, + sample_count: 1, + dimension: wgpu::TextureDimension::D2, + format: FORMAT, + usage: wgpu::TextureUsages::RENDER_ATTACHMENT + | wgpu::TextureUsages::TEXTURE_BINDING + | wgpu::TextureUsages::COPY_SRC, + view_formats: &[], + }); + let view = output.create_view(&Default::default()); + Self { + queue: queue.clone(), + device: device.clone(), + pipeline, + bind, + output, + view, + } + } + + /// Draw the source as it is now, returning the texture it is drawn into. + /// Submitted before this returns, so work queued after it reads the result. + pub fn scale(&self) -> &wgpu::Texture { + let mut encoder = self.device.create_command_encoder(&wgpu::CommandEncoderDescriptor { + label: Some("moq-scene scale"), + }); + { + let mut pass = encoder.begin_render_pass(&wgpu::RenderPassDescriptor { + label: Some("moq-scene scale"), + color_attachments: &[Some(wgpu::RenderPassColorAttachment { + view: &self.view, + depth_slice: None, + resolve_target: None, + ops: wgpu::Operations { + load: wgpu::LoadOp::Clear(wgpu::Color::BLACK), + store: wgpu::StoreOp::Store, + }, + })], + depth_stencil_attachment: None, + timestamp_writes: None, + occlusion_query_set: None, + multiview_mask: None, + }); + pass.set_pipeline(&self.pipeline); + pass.set_bind_group(0, &self.bind, &[]); + pass.draw(0..3, 0..1); + } + self.queue.submit([encoder.finish()]); + &self.output + } +} From 2b4b46eb4de97bfe0d613054c65c42715f7d82db Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 07:16:11 -0800 Subject: [PATCH 03/19] feat(studio)!: simulcast ladders that follow the bandwidth grant The program and each captured source encode a ladder (the picture's size, then 2/3 and 1/3 of it); each rung reserves its ceiling, follows its grant through moq_mux::rate, disables when it cannot be funded, and encodes the smaller sizes only while watched. Catalogs no longer claim bandwidth, so no track is counted twice; the Health panel shows each track's grant. Co-Authored-By: Claude Opus 5.5 --- README.md | 56 +++-- quest/m0/README.md | 2 +- quest/m0/encode-control.md | 36 ++- quest/m0/scene-editing.md | 3 +- rs/moq-scene/src/ladder.rs | 117 ++++++++- rs/moq-scene/src/publish.rs | 62 ++++- rs/moq-studio/src/engine.rs | 460 +++++++++++++++++++++--------------- rs/moq-studio/src/health.rs | 71 ++++-- rs/moq-studio/src/ui.rs | 10 +- 9 files changed, 556 insertions(+), 261 deletions(-) diff --git a/README.md b/README.md index 36d3391..d00f535 100644 --- a/README.md +++ b/README.md @@ -39,7 +39,7 @@ Every panel is a tab you can drag, split, or stack. | Devices | List this machine's cameras, displays, windows, and audio inputs, and add one as a source. | | Filters | Add, order, and set the filters of the selected source or item, and copy or paste them. | | Audio | Set each audio source's gain and mute, and watch its meter and the program's. | -| Output | See the broadcast path, whether the program is encoding, and the encoder's work. | +| Output | See the broadcast path, whether the program is encoding, and the encoders' work over every rung. | | Health | See how the stream reaches viewers, track by track, and the cause of each problem. | A new source, from the Sources or Devices panel, is placed over whatever the @@ -157,21 +157,34 @@ before sharing. ## Output -The program publishes as one H.264 rendition at the canvas size and frame rate, -with Opus stereo audio, under the `--broadcast` path. Timestamps come from the -studio's own clock, so a source that stalls repeats its last picture instead of -holding up the stream. - -The program encodes whether or not anyone watches, so a viewer who arrives -late can fetch the recent past. `--on-demand` copies and encodes the video only +The program publishes as H.264 renditions at the canvas frame rate, with Opus +stereo audio, under the `--broadcast` path. Timestamps come from the studio's +own clock, so a source that stalls repeats its last picture instead of holding +up the stream. + +The video is a simulcast ladder: the canvas size, then two thirds and one third +of it (1080p, 720p, and 360p for a 1080p canvas), leaving out any rung under +180 pixels on its shorter side. Each rung reserves at most 0.07 bits per pixel +per frame (about 4.4 Mb/s at 1080p30) on the relay connection's bandwidth +estimate, and its encoder follows the share it is granted: down at once when +the uplink closes, back up over a few seconds when it opens. A rung whose grant +can no longer fund it, so the one below would look better, stops encoding and +is advertised as disabled in the catalog until the grant recovers, so viewers +move to a smaller one; the smallest never stops. Smaller rungs are funded +first. + +The canvas-size rung encodes whether or not anyone watches, so a viewer who +arrives late can fetch the recent past. `--on-demand` copies and encodes it only while someone watches it, which saves that work but leaves nothing from before -the first viewer; audio encodes either way. A hardware encoder that reads +the first viewer. The smaller rungs encode only while someone watches them, and +give their encode sessions back between viewers. Audio encodes either way. A +hardware encoder that reads the GPU (NVENC on Linux, VideoToolbox on macOS, Media Foundation on Windows) takes each frame there when it can read the GPU the studio draws on. When it cannot, such as NVENC while the studio draws on a laptop's integrated GPU, the studio passes over it for an encoder that reads each frame back to the CPU (the opt-in `vaapi` or `openh264` features), which the Output panel counts. -Frames the encoder cannot keep up with are dropped rather than delaying the +Frames an encoder cannot keep up with are dropped rather than delaying the stream. ### Source tracks @@ -196,18 +209,22 @@ the hang draft specifies source listings: } ``` -A captured source encodes as H.264 for as long as it is in the collection, -whether or not anyone watches, so a late viewer can fetch its recent past; the -relay uploads only what someone subscribes to. A capture that changes size or -color space moves to a new track, which its catalog entry follows. Images and +A captured source encodes as an H.264 ladder like the program's, ranked below +every rung of the program. Its own size encodes for as long as it is in the +collection, whether or not anyone watches, so a late viewer can fetch its +recent past; the relay uploads only what someone subscribes to. Its smaller +rungs encode only while watched. A capture that changes size or color space +moves each rung that encodes to a new track, which its catalog entry follows; +a rung nobody watches moves once it next encodes. Images and colors publish once, as the image file and as a one-pixel PNG. Removing or editing a source ends its track. A track carries the plain capture, before any filter, and a clone publishes none, since it shows its source's capture. Microphones and system audio do not publish tracks of their own yet, and a broadcast source is already its publisher's. -Every captured source opens an encode session as it starts, and consumer GPUs -allow only a few hardware sessions at once. When the program encodes on +Every captured source opens an encode session as it starts, and one more for +each smaller rung someone watches, and consumer GPUs allow only a few hardware +sessions at once. When the program encodes on hardware, so do the sources, and a session the GPU refuses fails that source's track with the encoder's reason in the Sources panel rather than falling back to software. The source keeps drawing in the studio's own scenes. The program @@ -242,9 +259,10 @@ stalling, or cannot decode a track. Below that it shows: `.stats` broadcasts. A relay shares those only with a connection allowed to read them, so most show "unknown". In a cluster, a relay forwarding the stream to another counts as one session. -- **Tracks**: each track's bitrate and frame rate, and, from viewers that - report, the lowest bitrate any received, its worst grade, and the latency - from the program clock to the slowest viewer. Latency compares the two +- **Tracks**: each track's share of the connection's bandwidth estimate, or + "off" for a rung its grant cannot fund, its bitrate and frame rate, and, + from viewers that report, the lowest bitrate any received, its worst grade, + and the latency from the program clock to the slowest viewer. Latency compares the two hosts' wall clocks, so keep both synced. - **Viewers**: each viewer that reports, with its round trip, latency, and playout delay. The studio reads 64 at most, a sample the grades rest on: it diff --git a/quest/m0/README.md b/quest/m0/README.md index e39b046..b6e4f71 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -85,7 +85,7 @@ repository. - [Scene editing](/quest/m0/scene-editing.md) - rename and reorder scenes, and resize the canvas while live through a new rendition - [Broadcast sources play in sync](/quest/m0/broadcast-sync.md) - subscribed broadcasts and guests play audio and video on the publisher's timestamps, with `moq play`'s delay - [Wayland display and window sources](/quest/m0/wayland-sources.md) - portal-picked sources that reopen without the picker -- [Live encode control](/quest/m0/encode-control.md) - simulcast rungs that follow the bandwidth grant, live bitrate and resolution, keyframes on request +- [Keyframes on request](/quest/m0/encode-control.md) - every rendition cuts a keyframe for a joining viewer, so the GOP can grow - [Relay failover](/quest/m0/failover.md) - the same epoch on two relays, so a relay failure does not drop the stream - [Instant replay and DVR](/quest/m0/replay.md) - replays and clips from held groups, and viewer scrubbing through the archive - [HTML/CSS overlays](/quest/m0/overlays.md) - script-free HTML and CSS graphics through Blitz natively and the DOM in the browser diff --git a/quest/m0/encode-control.md b/quest/m0/encode-control.md index 53ade6b..d17f859 100644 --- a/quest/m0/encode-control.md +++ b/quest/m0/encode-control.md @@ -1,29 +1,25 @@ -# [L] Live encode control +# [S] Keyframes on request ## Goal -`moq-studio` adapts its encode live without a restart: the program and each -source publish simulcast rungs that follow the relay's bandwidth grant, a rung -nobody watches stops encoding, and bitrate and resolution change mid-stream -through catalog updates. OBS's multitrack output does -part of this only on Windows with specific GPUs; this works on every -platform. +Each rung of the program's and every source's simulcast ladder cuts a keyframe +when a viewer joins past its newest group, so the studio can run a longer GOP +than its 2 s default at no cost to how fast a viewer starts. ## Plan -Today the studio publishes one H.264 rendition at fixed settings. Its tracks -claim their bitrate on the connection's `bandwidth::Allocator`, but nothing -follows the grant. Reuse what exists rather than adding a studio -path: the grant follower in `moq_video::encode` and `moq_mux::rate`, the -catalog's in-place updates and new renditions (`doc/concept/hang.md`). The -encode producer already cuts a keyframe when a subscriber joins -(`encode/cuts.rs`), so each rung inherits that and can run a long GOP. The -Health panel's track table already shows each track's bitrate and reporting -viewers; add each rung's grant beside them, which needs moq to expose a -track's share of the allocator. +The ladder (`rs/moq-scene/src/ladder.rs`) encodes through +`moq_video::encode::Encoder` and publishes through +`moq_video::encode::Producer`, whose join cuts are private to moq's own +`publish_capture` (`encode/cuts.rs`, `Producer::wanted`): the studio cannot see +whether a join asks for a group past the newest. This needs moq to expose that +on `encode::Producer` (or a cut-on-join helper beside it), then a moq pin bump. +With it, advertise `track::Info::refresh` on each rung's track, cut in +`Rendition::encode` when a join asks, and measure the bitrate a longer GOP saves +before picking one. ## Related -- [Simulcast rung disable](https://github.com/moq-dev/moq/blob/main/quest/m1/simulcast-rung-disable.md) - grant-driven rung disable this builds on -- [moq-audio follows the grant](https://github.com/moq-dev/moq/blob/main/quest/m2/2848-follow-the-bandwidth-grant-in-moq-audio-instead-of.md) - the audio half -- [Scene editing](/quest/m0/scene-editing.md) - canvas resize through a new rendition +- [Simulcast rung disable](https://github.com/moq-dev/moq/blob/main/quest/m1/simulcast-rung-disable.md) - the grant-driven rung disable the ladder follows +- [moq-audio follows the grant](https://github.com/moq-dev/moq/blob/main/quest/m2/2848-follow-the-bandwidth-grant-in-moq-audio-instead-of.md) - the audio half, which still only reserves +- [Scene editing](/quest/m0/scene-editing.md) - canvas resize, which has to rebuild the program's ladder diff --git a/quest/m0/scene-editing.md b/quest/m0/scene-editing.md index 53f6d10..5a95c12 100644 --- a/quest/m0/scene-editing.md +++ b/quest/m0/scene-editing.md @@ -11,7 +11,8 @@ Decided 2026-10-26 (maintainer): a resize while publishing adds a new video rendition under a new track name in a catalog update, and viewers switch to it. A track name never changes what it means (AGENTS.md), so the old track ends rather than changing resolution mid-stream. The compositor reallocates -its canvas textures and the program encoder restarts on the new size. +its canvas textures and the program's ladder (`moq_scene::ladder`) restarts +on the new size, each rung on a new track. A rename updates every reference to the scene in the collection. Reordering is the order of the scene file's list. Update diff --git a/rs/moq-scene/src/ladder.rs b/rs/moq-scene/src/ladder.rs index 6b0877d..b78ccc9 100644 --- a/rs/moq-scene/src/ladder.rs +++ b/rs/moq-scene/src/ladder.rs @@ -92,7 +92,11 @@ pub fn producer( let suffix = match &rendition.codec { hang::catalog::VideoCodec::H264(_) => ".avc3", hang::catalog::VideoCodec::H265(_) => ".hev1", - other => return Err(Error::Unsupported(format!("{other} is not a codec the studio publishes"))), + other => { + return Err(Error::Unsupported(format!( + "{other} is not a codec the studio publishes" + ))); + } }; let track = broadcast.unique_track(suffix, reserved.track_info(priority))?; Ok(moq_video::encode::Producer::with_track(track, reserved, rendition)?) @@ -215,6 +219,8 @@ pub struct Rendition { encoder: Option, /// Keep the encoder, and its session, while skipping frames. hold: bool, + /// A held encoder skipped frames, so its next frame has to be a keyframe. + resumed: bool, state: State, /// The bitrate the gate last asked for. target: Rate, @@ -235,6 +241,7 @@ impl Rendition { config, encoder: None, hold, + resumed: false, state: State::Idle, target, fixed: false, @@ -270,10 +277,13 @@ impl Rendition { } if (self.state == State::Disabled) != (state == State::Disabled) { let enabled = state != State::Disabled; - self.producer.as_mut().expect("open until it ends").set_enabled(enabled)?; + self.producer + .as_mut() + .expect("open until it ends") + .set_enabled(enabled)?; } - if self.state != State::Encoding - && state == State::Encoding + if state == State::Encoding + && std::mem::take(&mut self.resumed) && let Some(encoder) = &mut self.encoder { // A held encoder resumes where a viewer can start decoding. @@ -363,6 +373,7 @@ impl Rendition { true => { if let (Some(encoder), Some(producer)) = (&mut self.encoder, &mut self.producer) { producer.publish(&encoder.flush()?)?; + self.resumed = true; } } false => self.drain()?, @@ -471,7 +482,10 @@ mod tests { [size(1920, 1080), size(1280, 720), size(640, 360)] ); // A portrait picture scales by its shorter side too. - assert_eq!(sizes(size(1080, 1920)), [size(1080, 1920), size(720, 1280), size(360, 640)]); + assert_eq!( + sizes(size(1080, 1920)), + [size(1080, 1920), size(720, 1280), size(360, 640)] + ); // Rungs under the floor are dropped, and odd sides round down to even. assert_eq!(sizes(size(641, 361)), [size(640, 360), size(426, 240)]); assert_eq!(sizes(size(200, 150)), [size(200, 150)]); @@ -483,4 +497,97 @@ mod tests { let ranks: Vec = (0..3).map(|index| priority(base, index, 3)).collect(); assert_eq!(ranks, [base - 2, base - 1, base]); } + + /// A gate follows its share of the estimate down at once and back up over + /// the ramp, and a rung that is not `always` encodes only while watched. + #[tokio::test(start_paused = true)] + async fn a_gate_follows_its_grant_and_its_viewers() { + let broadcast = moq_net::broadcast::Info::new().produce(); + let consumer = broadcast.consume(); + let track = broadcast.create_track("video", None).unwrap(); + let estimate = bandwidth::Producer::new(); + estimate.set(Some(Rate::from_mbps(3))).unwrap(); + let allocator = bandwidth::Allocator::new(estimate.consume()); + let ceiling = Rate::from_mbps(2); + + let mut always = Gate::new(&allocator, track.demand(), ceiling, None, true); + assert_eq!(always.step(), Step::Encode(ceiling)); + estimate.set(Some(Rate::from_mbps(1))).unwrap(); + assert_eq!( + always.step(), + Step::Encode(Rate::from_mbps(1)), + "a drop applies at once" + ); + assert_eq!(always.grant().peek(), Some(Rate::from_mbps(1))); + estimate.set(Some(Rate::from_mbps(3))).unwrap(); + match always.step() { + Step::Encode(rate) => assert!(rate < ceiling, "a raise climbs: {rate:?}"), + step => panic!("{step:?}"), + } + tokio::time::advance(std::time::Duration::from_secs(5)).await; + assert_eq!(always.step(), Step::Encode(ceiling), "and reaches the ceiling"); + drop(always); + + let mut watched = Gate::new(&allocator, track.demand(), ceiling, None, false); + assert_eq!(watched.step(), Step::Idle, "nobody watches"); + let subscription = consumer.track("video").unwrap().subscribe(None).await.unwrap(); + assert_eq!(watched.step(), Step::Encode(ceiling)); + drop(subscription); + tokio::task::yield_now().await; + assert_eq!(watched.step(), Step::Idle, "the viewer left"); + } + + /// A rung its grant cannot fund is advertised disabled and encodes + /// nothing, then resumes on a keyframe; an idle one stays on offer. Either + /// gives its encoder session up, unless it holds it. + #[tokio::test] + async fn a_disabled_rung_is_advertised_off() { + for hold in [false, true] { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let catalog = moq_mux::catalog::Producer::<()>::new(&mut broadcast, Default::default()).unwrap(); + let framerate = moq_video::FrameRate::new(30, 1).unwrap(); + let size = moq_video::Size::new(64, 48); + let mut config = moq_video::encode::Config::new(size.width, size.height, framerate); + // The one encoder every host has, and one that holds no frames back. + config.kind = moq_video::encode::Kind::Named("openh264".into()); + config.bitrate = Some(ceiling(size, framerate)); + let rendition = config.probe().await.unwrap(); + let producer = producer(&broadcast, catalog.reserve(), rendition, hang::catalog::PRIORITY.video).unwrap(); + let mut rendition = Rendition::new(producer, config, hold); + let name = rendition.demand().name().to_string(); + let enabled = || catalog.snapshot().video.renditions[&name].enabled; + let sent = || catalog.stats()[&name].clone(); + let gray = |millis: u64| { + let i420 = moq_video::I420::new(size, vec![0x80; moq_video::I420::len(size).unwrap()]).unwrap(); + let timestamp = moq_net::Timestamp::from_micros(millis * 1000).unwrap(); + moq_video::Frame::new(moq_video::Surface::I420(i420), timestamp) + }; + let encode = Step::Encode(ceiling(size, framerate)); + + rendition.apply(encode).unwrap(); + for millis in [0, 33, 66] { + rendition.encode(&gray(millis)).unwrap(); + } + assert!(enabled()); + assert_eq!((sent().frames, sent().keyframes), (3, 1)); + + rendition.apply(Step::Disabled).unwrap(); + assert!(!enabled(), "viewers pick another rung"); + assert_eq!(rendition.encoder.is_some(), hold, "only a held rung keeps its session"); + + rendition.apply(encode).unwrap(); + assert!(enabled()); + for millis in [200, 233] { + rendition.encode(&gray(millis)).unwrap(); + } + assert_eq!((sent().frames, sent().keyframes), (5, 2), "it resumes on a keyframe"); + + rendition.apply(Step::Idle).unwrap(); + assert!(enabled(), "an unwatched rung stays on offer"); + assert_eq!(rendition.encoder.is_some(), hold, "only a held rung keeps its session"); + rendition.apply(encode).unwrap(); + rendition.encode(&gray(400)).unwrap(); + assert_eq!((sent().frames, sent().keyframes), (6, 3), "it resumes on a keyframe"); + } + } } diff --git a/rs/moq-scene/src/publish.rs b/rs/moq-scene/src/publish.rs index 11a40c2..38a5b19 100644 --- a/rs/moq-scene/src/publish.rs +++ b/rs/moq-scene/src/publish.rs @@ -261,7 +261,13 @@ impl Video { let producer = ladder::producer(&listing.broadcast, reserved, rendition, priority)?; let rendition = Rendition::new(producer, config, false); let lower = ceilings.get(index + 1).copied(); - let gate = Gate::new(&listing.bandwidth, rendition.demand(), ceilings[index], lower, index == 0); + let gate = Gate::new( + &listing.bandwidth, + rendition.demand(), + ceilings[index], + lower, + index == 0, + ); rungs.push((gate, rendition)); } Ok(Self { @@ -567,4 +573,58 @@ mod tests { }) .await; } + + /// A source's largest rung encodes unwatched, so a late viewer finds its + /// recent past, and a smaller one only while someone subscribes to it, + /// each listed as its own rendition. + #[tokio::test] + async fn smaller_rungs_encode_while_watched() { + local(async { + let (publish, catalog, consumer) = studio(); + let claim = publish.claim("camera"); + let listing = Listing::new(&claim, &publish).unwrap(); + let framerate = moq_video::FrameRate::new(30, 1).unwrap(); + let size = moq_video::Size::new(640, 360); + let frames = |name: &str| listing.catalog.stats().get(name).map_or(0, |sent| sent.frames); + + tokio::select! { + result = listing.mirror() => panic!("the mirror ended: {result:?}"), + () = async { + let mut video = Video::new(&listing, size, framerate, None).await.unwrap(); + let snapshot = listed(&catalog, |sources| { + sources.get("camera").is_some_and(|camera| camera.video.renditions.len() == 2) + }) + .await; + let renditions = &snapshot.ext.studio.sources["camera"].video.renditions; + let named = |width| { + let (name, _) = renditions.iter().find(|(_, config)| config.coded_width == Some(width)).unwrap(); + name.clone() + }; + let (large, small) = (named(640), named(426)); + + let mut millis = 0; + let mut encode = |video: &mut Video| { + for _ in 0..3 { + video.encode(&mut gray(size, millis)).unwrap(); + millis += 33; + } + }; + encode(&mut video); + assert!(frames(&large) > 0, "the largest rung encodes unwatched"); + assert_eq!(frames(&small), 0, "nobody watches the smaller one"); + + let watching = consumer.track(&small).unwrap().subscribe(None).await.unwrap(); + encode(&mut video); + let watched = frames(&small); + assert!(watched > 0, "a viewer starts the smaller rung"); + + drop(watching); + tokio::task::yield_now().await; + encode(&mut video); + assert_eq!(frames(&small), watched, "it stops once the viewer leaves"); + } => {} + } + }) + .await; + } } diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index d4d466b..16b766b 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -4,9 +4,11 @@ //! //! Threads: //! - **program** draws preview and program once per canvas frame and hands the -//! program to the encoder, stamped with the program clock. +//! program to each rung's encoder its gate lets encode, scaled to the rung +//! and stamped with the program clock. //! - **audio** reads the program bus on the same clock and publishes it. -//! - **encode** waits for each program copy and encodes it. +//! - **encode-***: one per rung of the program's ladder, which waits for each +//! program copy and encodes it. //! - **source-***: one per running source, since a capture stream may not //! leave the thread that opened it. //! - **autosave** writes the collection and its history to their files once @@ -21,8 +23,9 @@ use std::time::Duration; use anyhow::Context; use bytes::Bytes; use kio::clock::Instant; +use moq_scene::ladder::{self, Gate, Rendition, Step}; use moq_scene::source::{self, Driver, Sources}; -use moq_scene::{Collection, Compositor, Draft, History, wgpu}; +use moq_scene::{Collection, Compositor, Draft, History, Scaler, wgpu}; use moq_video::render::export::{Exporter, Pending, Route}; use crate::autosave::{self, Autosave}; @@ -164,7 +167,9 @@ impl State { pub struct Stats { /// Canvas frames drawn. pub rendered: AtomicU64, - /// Program frames handed to the encoder. + /// Program frames offered to an encoder, one per rung encoding it. + pub offered: AtomicU64, + /// Program frames handed to an encoder. pub copied: AtomicU64, /// Microseconds the last one waited on the GPU, and on a readback converted /// pixels. @@ -176,13 +181,13 @@ pub struct Stats { pub read_back: AtomicU64, /// Program frames not encoded because the encoder still held every copy. pub dropped: AtomicU64, - /// Program frames encoded. + /// Program frames encoded, over every rung. pub encoded: AtomicU64, /// Microseconds the last encode took. pub encode_us: AtomicU64, - /// Whether any viewer is subscribed to the video. + /// Whether any viewer is subscribed to any rung of the video. pub watched: AtomicBool, - /// The encoder backend, once one opened. + /// The encoder backend, once the largest rung's opened. pub encoder: Mutex>, /// Why the output stopped, if it did. pub error: Mutex>, @@ -199,13 +204,17 @@ impl Stats { pub struct Output { /// The broadcast path on the relay. pub path: String, - /// Whether the program video is copied and encoded only while someone - /// watches, rather than all the time. + /// Whether the program video's largest rung is copied and encoded only + /// while someone watches, rather than all the time. The smaller rungs + /// always are. pub on_demand: bool, /// The relay connection, which reconnects on its own and dials only while /// held. The Health panel reads its state. _connection: moq_tokio::Connection, publication: moq_net::origin::Publication, + /// The audio's claim on the connection's bandwidth, which it does not + /// follow: Opus at this rate is never what congests the uplink. + _audio: moq_net::bandwidth::Reservation, } impl Drop for Output { @@ -302,29 +311,50 @@ impl Engine { let (video, audio, clock, publisher) = match publish { Some(Publish { - video, + rungs, audio, - config, - exporter, clock, health, sources: _, }) => { - let zero_copy = exporter.route() == Route::Gpu; + let zero_copy = rungs[0].exporter.route() == Route::Gpu; engine.stats.zero_copy.store(zero_copy, Ordering::Relaxed); - // One more than the exporter's pool, for a pause behind a full pool. - let (frames, pending) = mpsc::sync_channel(Exporter::DEPTH + 1); - tokio::spawn(watch(video.demand(), engine.clone())); - let encoder = engine.clone(); - let (opened, open) = tokio::sync::oneshot::channel(); - std::thread::Builder::new() - .name("encode".into()) - .spawn(move || encode(&encoder, opened, pending, video, config, zero_copy))?; - // Before the program clock starts the sources, so the program holds - // its encode session even while unwatched, and a source past the - // GPU's session cap is the one that fails. - open.await.context("the encode thread stopped")??; - (Some((frames, exporter)), Some(audio), Some(clock), Some(health)) + let program = engine.state.lock().unwrap().compositor.program().clone(); + let mut video = Vec::with_capacity(rungs.len()); + for (index, rung) in rungs.into_iter().enumerate() { + let Rung { + gate, + exporter, + producer, + config, + } = rung; + // The largest rung is the canvas itself; the others draw it smaller. + let scaler = (index > 0).then(|| Scaler::new(&gpu.device, &gpu.queue, &program, exporter.size())); + // One more than the exporter's pool, for a step behind a full pool. + let (sender, pending) = mpsc::sync_channel(Exporter::DEPTH + 1); + let encoder = engine.clone(); + let size = exporter.size(); + // The largest rung holds its encoder from the start. Opened before + // the program clock starts the sources, so the program holds its + // encode session even while unwatched, and a source past the GPU's + // session cap is the one that fails. + let (opened, open) = tokio::sync::oneshot::channel(); + let opened = (index == 0).then_some(opened); + std::thread::Builder::new() + .name(format!("encode-{}p", size.height)) + .spawn(move || encode(&encoder, opened, pending, producer, config, zero_copy))?; + if index == 0 { + open.await.context("the encode thread stopped")??; + } + video.push(Feeder { + gate, + scaler, + exporter, + sender, + step: None, + }); + } + (Some(video), Some(audio), Some(clock), Some(health)) } None => (None, None, None, None), }; @@ -339,7 +369,7 @@ impl Engine { let program = engine.clone(); std::thread::Builder::new() .name("program".into()) - .spawn(move || clocked(program.run_program(start, epoch, video)))?; + .spawn(move || clocked(program.run_program(start, epoch, video.unwrap_or_default())))?; let mix = engine.clone(); std::thread::Builder::new() .name("audio".into()) @@ -386,7 +416,6 @@ impl Engine { fn local(&self, now: Instant) -> health::Local { let stats = &self.stats; let watched = stats.watched.load(Ordering::Relaxed); - let on_demand = self.output.as_ref().is_some_and(|output| output.on_demand); let stalled = { let state = self.state.lock().unwrap(); let program = state @@ -410,10 +439,9 @@ impl Engine { .collect() }; health::Local { - rendered: stats.rendered.load(Ordering::Relaxed), + offered: stats.offered.load(Ordering::Relaxed), dropped: stats.dropped.load(Ordering::Relaxed), encoded: stats.encoded.load(Ordering::Relaxed), - encoding: self.output.is_some() && (!on_demand || watched), watched, encoder: stats.encoder.lock().unwrap().clone(), failed: stats.error.lock().unwrap().clone(), @@ -422,23 +450,16 @@ impl Engine { } /// Draw a canvas frame per tick of the program clock, and hand the program - /// to the encoder, stamped `epoch` past `start`: every frame, or only while - /// anyone watches when the output is on demand. + /// to each rung's encoder, stamped `epoch` past `start`, while that rung's + /// gate lets it encode. /// - /// Drawing outlives the encoder: a failed output stops copying the program, + /// Drawing outlives the encoders: a failed rung stops copying the program, /// and the preview and program keep moving. - async fn run_program( - &self, - start: Instant, - epoch: Duration, - mut video: Option<(mpsc::SyncSender