diff --git a/README.md b/README.md index a94fa40..0353a07 100644 --- a/README.md +++ b/README.md @@ -28,8 +28,8 @@ cargo run --release -p moq-studio --features capture,openh264 -- \ refuses a fixed `--connect-bind` port, since each relay's connection binds its own. Its credential rides its URL's `?jwt=`; `--connect-token` goes to the first relay only. The stream's tracks claim their bitrate on the first - relay's link alone, so only that link is probed toward the stream's rate; - no encoder adapts to either estimate. + relay's link alone, so only that link is probed toward the stream's rate, + and the video follows that link's estimate alone. - `capture` adds camera, display, window, microphone, and system-audio sources, and listening to a mix. On Linux it links ALSA; add `pipewire` for PipeWire cameras. `openh264` is the software H.264 fallback for a machine without a hardware encoder. @@ -99,7 +99,7 @@ Every panel is a tab you can drag, split, or stack. | Guests | Invite a guest, show their join link as a QR code, see who joined, and take a guest out of every scene. | | Filters | Crop the selected source or item, add, order, and set its filters, set an item's blend mode, and copy or paste filters. | | Audio | Add, rename (double-click), and remove mixes, pick the one on program and the one to listen to and on which output, and set each audio source's gain, mute, delay, and the mixes it feeds, with a meter on every mix and source. | -| Output | Start and stop recording, set the canvas size and frame rate, and see the broadcast path, whether the program is encoding, and the encoder's work. | +| Output | Start and stop recording, set the canvas size and frame rate, and see the broadcast path, whether the program is encoding, the encoder sessions open, and each rendition's encoder work. | | Health | See how the stream reaches viewers, relay by relay and track by track, and the cause of each problem. | | Replay | Replay the last moments of the program or a source in a replay source, slowed down if you like, or save them as a clip. | @@ -474,24 +474,36 @@ before sharing. ## Output -The program publishes as one H.264 rendition at the canvas size and frame rate, -with the mix on program as 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 publishes as two H.264 renditions, at the canvas size and at half +of it, at the canvas frame rate, with the mix on program as Opus stereo audio, +under the `--broadcast` path. A canvas under 360 lines has no half size. +Timestamps come from the studio's own clock, so a source that stalls repeats +its last picture instead of holding up the stream. + +Each rendition follows its share of the bandwidth the first relay's connection +estimates: a closing uplink lowers its bitrate at once, and an opening one +raises it back over a few seconds, up to 0.07 bits per pixel (about 4.4 Mb/s +for 1080p at 30 fps). The half size is funded first. When the share left for +the canvas size cannot fund it over the half size, the catalog lists it as +disabled and viewers move to the half size until the uplink recovers. A viewer +joining a track whose newest group has grown heavy asks for a keyframe, and +the encoder cuts one, so later viewers start near it. A new canvas size or frame rate, from the Output panel, an undo, or a reload, takes effect live. A track never changes what it carries, so the old -rendition's track ends and a new one is listed in its place in a catalog -update, on the same timeline, and viewers switch to it; the video pauses while -the new encoder opens. Items keep their place in canvas pixels. A canvas the -encoder cannot take stops the video, saying why, until the canvas changes +renditions' tracks end and new ones are listed in their place in a catalog +update, on the same timeline, and viewers switch to them; the video pauses +while the new encoders open. Items keep their place in canvas pixels. A canvas +the encoder cannot take stops the video, saying why, until the canvas changes again. At startup, it stops the studio when live; without `--connect`, the studio starts without video. -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 -while someone watches or records it, which saves that work but leaves nothing -from before the first viewer; audio encodes either way. A hardware encoder that reads +The canvas size encodes whether or not anyone watches, so a viewer who arrives +late can fetch the recent past. The half size encodes only while someone +watches it. `--on-demand` copies and encodes the canvas size only while someone +watches it too, which saves that work but leaves nothing from before the first +viewer; audio encodes either way. A recording has an encoder of its own (see +Recording). 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 @@ -502,57 +514,60 @@ stream. ### Source tracks -Each camera, display, window, image, and color also publishes as its own track -in the same broadcast, so a viewer can compose the scene itself. The catalog's -`video` and `audio` sections stay the program's; the sources are listed under -the `dev.moq.studio` section's `sources`, by source name, each as a hang -catalog of its own on the program's clock. This listing is provisional until -the hang draft specifies source listings: - -```json -{ - "video": { "renditions": { "0.avc3": { "codec": "avc3.42c01f", "...": "..." } } }, - "audio": { "renditions": { "audio": { "codec": "opus", "...": "..." } } }, - "dev.moq.studio": { - "sources": { - "camera": { "video": { "renditions": { "1.avc3": { "...": "..." } } } }, - "logo": { "binary": { "tracks": { "2.png": { "mode": "snapshot", "mime": "image/png" } } } } - } - } -} -``` - -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. Without `--connect`, it encodes -only while a recording reads it. A capture that changes size or -color space moves to a new track, which its catalog entry follows. 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. A browser -source or overlay publishes none, so viewers see it only in the program, 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 -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 -opens its session before any source, even with `--on-demand`, so the sources -past the cap fail rather than the program. When the -program encodes in software, a source takes hardware where it can and software -otherwise. +Each camera, display, window, image, and color also has a track of its own, so +the studio can replay or record it on its own. These tracks stay in the studio: the +program is the only video it publishes to a relay. They are listed in a local +catalog under the `dev.moq.studio` section's `sources`, by source name, each +as a hang catalog of its own on the program's clock. + +A captured source encodes its track as H.264 only while it is used: while it +is shown on program, so a replay of it has its recent past, or while a +recording reads its track. Otherwise it holds no encoder. A display or window +that goes into use while its screen stands still, and so delivers nothing new, +restarts its capture, so its track starts with the picture as it is. A capture that +changes size or color space moves to a new track, which its catalog entry +follows. Images and colors keep their picture 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 has none, since it shows its +source's capture. Microphones and system audio have no tracks of their own yet, +nor does a browser source or an overlay, and a broadcast source is already its +publisher's. + +Every encoder holds an encode session, and consumer GPUs allow only a few +hardware sessions at once: 8 on a GeForce card, while a base Apple M4 has one +encode engine for them all. The Output panel counts the sessions open: + +| Encoder | Sessions | +|---|---| +| The program's canvas size, live | 1, or none without `--connect` | +| Its half size | 1 while someone watches it | +| The recording's own canvas size | 1 while recording | +| Each captured source | 1 while on program or recorded | + +So a live scene showing a camera, a display, and a window holds 4 sessions, 5 +with the half size watched, and 6 while recording too. 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 opens its session before any +source, even with `--on-demand`, so the sources past the cap fail rather than +the program. When the program encodes in software, a source takes hardware +where it can and software otherwise. ### Recording -**Start recording** in the Output panel saves the program to -` .mp4` in the `--recordings` directory, live -or not. With **each source too**, every camera, display, and window that -publishes a track of its own records into a file of its own beside it, +**Start recording** in the Output panel saves the program, at the canvas size, +to ` .mp4` in the `--recordings` directory, +live or not. The recording has an encoder of its own, opened only while it +records, at full quality whatever the uplink grants the live stream, as OBS's +advanced output mode records apart from streaming. It costs one more encode +session and one more copy of each program frame while recording. With +**each source too**, every camera, display, and window records its +own track into a file of its own beside it, ` .mp4`, holding the capture before -any filter. Broadcast sources, guests included, publish no track of the -studio's, so they get no file of their own. A source added while recording +any filter; a source being recorded encodes whether or not it is on program. +Broadcast sources, guests included, have no track of the studio's, so they get +no file of their own. A source added while recording starts its file then, and one removed finishes its file then. A new canvas size or frame rate, or a capture that changes size or color space, moves the video to a new track, so the program or that source goes on in a new file, @@ -560,14 +575,15 @@ numbered after the first, since players show only one video track. A file starts with up to 2 s from before it opened, so a recording starts a little early, and a track that goes on into the next file, such as the program audio, repeats up to 2 s of the last. A -recording reads the broadcast the way a viewer does, so it shares the live -encoders and starts and stops without touching what the relay serves. Without -`--connect`, or with `--on-demand`, the program encodes only while something -records or watches it. +recording reads the studio's own broadcasts, never what the relay serves, so it +starts and stops without touching the live stream. The mix on program is +encoded a second time for it while recording, so the file holds the same mix +viewers hear. Without `--connect`, only the +recording's encoder encodes the program. Each file plays while it grows, so a crash leaves a playable fragmented MP4, and becomes a regular MP4 when the recording stops or the studio quits. -Quitting first lets the program's encoders publish what they still hold, and +Quitting first lets the recording's encoders publish what they still hold, and the program's file keeps it. A file nothing reached before the recording stopped is removed. @@ -602,16 +618,18 @@ one. That is usually a fraction of a second, but it can reach back as far as the keyframe for a slow or deeply reordered source. The panel says where once it is written, shortly after the click. -Nothing records in the background for this. The program and each source's -track keep their last 30 seconds for late viewers anyway, and a replay copies -from those, so a replay needs `--connect` and goes back at most that far. With -`--on-demand`, the program has no past from before someone watched it. A +Nothing records in the background for this. The program's tracks keep their +last 30 seconds for late viewers anyway, and each source's track while it is on +program or recorded, and a replay copies from those, so a replay needs +`--connect` and goes back at most that far. With `--on-demand`, the program has +no past from before someone watched it, and a source has none from before it +was last on program or recorded; a replay of nothing is refused, saying why. A broadcast source replays from what the studio received of it, and only over a relay connection whose protocol version can address a track by time; an older one refuses with the reason in the panel. An image, a color, or a web page has no past to replay: a page publishes no track of its own, so replay the program -to show it. A replay source publishes each replay as its own track while it plays, -like a camera, so viewers composing the scene themselves see it too. +to show it. A replay source on program encodes each replay as its own track +while it plays, like a camera. ## Health @@ -630,18 +648,20 @@ it is unhealthy. Below that it shows: - **Uplink**: the round trip to the first connected relay, the bandwidth the connection estimates it can carry, the media the program's tracks write, and the share lost and resent. Loss and headroom are graded only while someone - watches, since the relay pulls nothing otherwise. Source tracks are left out - of the media, since they encode whether or not anyone pulls them. + watches, since the relay pulls nothing otherwise. Source tracks never reach + the relay, so they are left out of the media. - **Watching**: the sessions subscribed to the broadcast, from the relays' `.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. Two relays whose stats use the same node name publish them at the same path, so only one of them counts: the first relay's while it has them. -- **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. The program's tracks come - first, then each source's own tracks, grouped by source; a still image or +- **Tracks**: each track's share of the connection's bandwidth (**Granted**, + or "off" for a rendition the share 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. The + program's tracks come first, then each source's own tracks, grouped by + source; a still image or color carries no media to grade. Latency compares the two hosts' wall clocks, so keep both synced. - **Viewers**: each viewer that reports, with its round trip, latency, and diff --git a/quest/m0/README.md b/quest/m0/README.md index 4eea230..e02b0c6 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -93,7 +93,6 @@ repository. - [Audio source tracks](/quest/m0/audio-source-tracks.md) - microphones and system audio get their own tracks, for isolated recording and replay - [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 and stingers](/quest/m0/animation.md) - keyframed item properties with easing, and stinger transitions - [ML video effects](/quest/m0/effects.md) - background removal and other ML effects, baked into their source's track - [Destinations](/quest/m0/destinations.md) - RTMP, RTMPS, and WHIP push with per-destination codecs and shared encoders, on a UI-free egress library a hosted service can reuse diff --git a/quest/m0/encode-control.md b/quest/m0/encode-control.md deleted file mode 100644 index dfc71f6..0000000 --- a/quest/m0/encode-control.md +++ /dev/null @@ -1,28 +0,0 @@ -# [L] Live encode control - -## 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. - -## 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. - -## 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 diff --git a/quest/m0/publish-independently.md b/quest/m0/publish-independently.md index 7f20082..a656aeb 100644 --- a/quest/m0/publish-independently.md +++ b/quest/m0/publish-independently.md @@ -14,7 +14,7 @@ Decided 2026-11-03 (maintainer, on [moq-dev/studio#21](https://github.com/moq-dev/studio/pull/21)): sources are not published to the relay by default, since every encoding costs an encoder session and consumer GPUs have few. A source published independently gets one -rendition, as sources do in the encode ladder, while the program keeps its +rendition, as every source track does, while the program keeps its two sizes. It counts as used, so it encodes for as long as it is published ([Audio source tracks](/quest/m0/audio-source-tracks.md) states the rule). @@ -25,5 +25,4 @@ user does, so `README.md` documents them. ## Related -- [Live encode control](/quest/m0/encode-control.md) - the encode ladder that makes source tracks local by default - [Viewer-side composition](/quest/m1/compose/README.md) - the m1 epic that composes from independently published sources diff --git a/quest/m0/record-mix.md b/quest/m0/record-mix.md index 038cdf3..bb5ecaa 100644 --- a/quest/m0/record-mix.md +++ b/quest/m0/record-mix.md @@ -13,19 +13,16 @@ Split from [Audio mixes](https://github.com/moq-dev/studio/pull/32), which landed the mixes, routing, meters, delay, listening, and the program's pick (`[mixes] program`). -A recording reads the studio's broadcast the way a viewer does -(`rs/moq-studio/src/record.rs`), so a mix other than the program's needs an -Opus track of its own. Keep that track off the relay: a mix the creator left -off the stream must not be fetchable from it, as source tracks stay local -after [Live encode control](/quest/m0/encode-control.md). Encode it only while -a recording reads it, and record the program's video beside it in one file. +A recording reads the program from a broadcast the studio keeps off the relay +(`Recording` in `rs/moq-studio/src/engine.rs`, from +[moq-dev/studio#21](https://github.com/moq-dev/studio/pull/21)), beside the +recording's own canvas-size encoder: its Opus track is a second encoding of +the mix on program, made only while a recording reads it. Point that encoding +at the mix the recording picks instead, so a mix the creator left off the +stream is never fetchable from the relay. The pick lives in the scene file beside `[mixes] program`, and the Output panel shows it. Update `README.md`. -## Required - -- [Live encode control](/quest/m0/encode-control.md) - the local broadcast a mix track stays off the relay in - ## Related - [Destinations](/quest/m0/destinations.md) - each destination picks a mix the same way diff --git a/rs/moq-scene/src/lib.rs b/rs/moq-scene/src/lib.rs index 4cbd969..f5d0d77 100644 --- a/rs/moq-scene/src/lib.rs +++ b/rs/moq-scene/src/lib.rs @@ -19,9 +19,10 @@ //! - [`Compositor`] draws the scene in preview and the scene on program into //! textures on the caller's `wgpu` device, cutting or running a //! [`Transition`] between them. `moq_video::render::export` hands the -//! program to an encoder, on the GPU where the platform allows. -//! - [`publish`] gives each local source its own track in the studio's -//! broadcast, so viewers can compose the scene themselves. +//! program to an encoder, on the GPU where the platform allows, and a +//! [`Scaler`] draws it smaller for a smaller rendition. +//! - [`publish`] gives each local source its own track, in a broadcast the +//! studio keeps to itself, to replay or record the source on its own. //! //! Every picture is drawn on the program clock the caller ticks //! [`Compositor::render`] with, from whatever each source delivered last: a @@ -38,6 +39,7 @@ mod lut; mod overlay; pub mod publish; pub mod replay; +mod scale; pub mod source; pub use clip::Clip; @@ -47,6 +49,7 @@ pub use collection::{ }; 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 2de68f1..8ee7ee2 100644 --- a/rs/moq-scene/src/publish.rs +++ b/rs/moq-scene/src/publish.rs @@ -1,43 +1,43 @@ -//! Each local source's own track in the studio's broadcast. +//! Each local source's own track, kept in a broadcast of the studio's own. //! -//! A source publishes what it shows beside the program, so a viewer can -//! compose the scene itself. Its renditions are listed in a hang catalog of its -//! own, nested by source name under the studio catalog's [`SECTION`] section. -//! The root `video` and `audio` sections stay the program's alone: hang reads -//! 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's track is for the studio's own use, such as replaying the source +//! or recording it on its own, and is never announced to a relay: the program +//! is the only video the studio publishes, which keeps it within the encode +//! sessions a GPU allows. Its renditions are listed in a hang catalog of its +//! own, nested by source name under the local catalog's [`SECTION`] section. //! -//! A source encodes while it runs, whatever the demand, so a late viewer finds -//! its recent past; MoQ uploads only what a subscriber pulls. On -//! [demand](Publish::on_demand), it encodes only while subscribed. +//! A source encodes only while it is used: while it shows on program +//! ([`Publish::show`]), so a replay of it has its recent past, or while +//! something in the studio subscribes to its track, such as a recording. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use serde::{Deserialize, Serialize}; use crate::Error; -/// The studio catalog's own root section, namespaced as hang asks of +/// The local catalog's own root section, namespaced as hang asks of /// application sections. pub const SECTION: &str = "dev.moq.studio"; -/// The application sections of the studio catalog. +/// The application sections of the local catalog. #[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] pub struct Ext { - /// The [`SECTION`] section, left off the wire while nothing is published. + /// The [`SECTION`] section, left off the wire while nothing is listed. #[serde(rename = "dev.moq.studio", default, skip_serializing_if = "Section::is_empty")] pub studio: Section, } impl moq_mux::catalog::hang::CatalogExt for Ext {} -/// What the studio lists beside the program. +/// What the local catalog lists. #[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct Section { - /// Each publishing source's catalog, by source name. Its tracks are in the - /// studio's broadcast, and its timestamps are on the studio catalog's clock. + /// Each source's catalog, by source name. Its tracks are in the local + /// broadcast, and its timestamps are on the program's clock. #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] pub sources: BTreeMap, } @@ -48,21 +48,49 @@ impl Section { } } -/// The studio catalog. +/// The catalog of the program's broadcast, or of the sources' local one. pub type Catalog = moq_mux::catalog::Producer; -/// Where sources publish: the studio's broadcast and its catalog. +/// How many video encoders are open. Each holds a session on a hardware +/// encoder, which a GPU caps, such as at 8 on a GeForce card. +#[derive(Clone, Debug, Default)] +pub struct Sessions(Arc); + +impl Sessions { + /// How many encoders are open now. + pub fn open(&self) -> usize { + self.0.load(Ordering::Relaxed) + } + + fn hold(&self) -> Session { + self.0.fetch_add(1, Ordering::Relaxed); + Session(self.clone()) + } +} + +/// One open encoder's place in its [`Sessions`], given back when dropped. +struct Session(Sessions); + +impl Drop for Session { + fn drop(&mut self) { + (self.0).0.fetch_sub(1, Ordering::Relaxed); + } +} + +/// Where sources publish: a broadcast the studio keeps to itself, and its +/// catalog. #[derive(Clone)] pub struct Publish { broadcast: moq_net::broadcast::Producer, catalog: Catalog, - bandwidth: moq_net::bandwidth::Allocator, kind: moq_video::encode::Kind, - on_demand: bool, + sessions: Sessions, owners: Arc>, + /// The sources on program, which each source's video follows. + shown: Arc>>, } -/// Which [`Claim`] owns each source name's entry in the studio catalog, and +/// Which [`Claim`] owns each source name's entry in the local catalog, and /// each open [`Listing`]'s catalog by its claim. #[derive(Default)] struct Owners { @@ -72,38 +100,51 @@ 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. + /// Keep sources' tracks in a new local broadcast stamped on `clock`, the + /// program's, encoding captured video with `kind` and counting its open + /// encoders in `sessions`. /// /// 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 /// with the reason rather than quietly loading the CPU with a software one. - pub fn new( - broadcast: moq_net::broadcast::Producer, - catalog: Catalog, - bandwidth: moq_net::bandwidth::Allocator, - kind: moq_video::encode::Kind, - ) -> Self { - Self { + pub fn new(clock: moq_mux::Clock, kind: moq_video::encode::Kind, sessions: Sessions) -> Result { + let mut broadcast = moq_net::broadcast::Info::new().produce(); + let config = moq_mux::catalog::Config::default() + .with_catalog(Default::default()) + .with_clock(clock); + let catalog = Catalog::new(&mut broadcast, config)?; + Ok(Self { broadcast, catalog, - bandwidth, kind, - on_demand: false, + sessions, owners: Default::default(), - } + shown: Default::default(), + }) } - /// Encode each source's video only while someone subscribes to it, for a - /// studio whose only readers are its own recordings, which never fetch a - /// past they were not reading. - pub fn on_demand(mut self) -> Self { - self.on_demand = true; - self + /// The local broadcast, to read a source's track from. + pub fn consume(&self) -> moq_net::broadcast::Consumer { + self.broadcast.consume() + } + + /// What the local catalog lists now. + pub fn catalog(&self) -> hang::Catalog { + self.catalog.snapshot() + } + + /// Encode the sources named in `shown`, those on program, whether or not + /// anything reads their tracks; any other encodes only while read. + pub fn show(&self, shown: BTreeSet) { + // Only a change wakes the sources, since this runs every frame. + self.shown.send_if_modified(|current| { + let changed = *current != shown; + *current = shown; + changed + }); } - /// What each source track the studio catalog lists has written, keyed by + /// What each source track the local catalog lists has written, keyed by /// rendition ID, as [`Catalog::stats`](moq_mux::catalog::Producer::stats) /// reports the program's. The broadcast names every track uniquely, so all /// sources share one map. A replaced source's tracks drop out with its @@ -114,7 +155,7 @@ impl Publish { listed.flat_map(|catalog| catalog.stats()).collect() } - /// Claim `name`'s entry in the studio catalog for a source that is opening, + /// Claim `name`'s entry in the local catalog for a source that is opening, /// taking it from any older source of that name. pub(crate) fn claim(&self, name: &str) -> Claim { let mut owners = self.owners.lock().unwrap(); @@ -129,7 +170,7 @@ impl Publish { } } -/// A source's hold on its name's entry in the studio catalog, until a newer +/// A source's hold on its name's entry in the local catalog, until a newer /// source of that name claims it. /// /// An edited source's replacement starts before the old one stops, and their @@ -157,33 +198,33 @@ impl Drop for Claim { } } -/// One source's catalog, listed in the studio's for as long as this lives and -/// its claim holds. +/// One source's catalog, listed in the local one for as long as this lives +/// and its claim holds. pub(crate) struct Listing<'a> { claim: &'a Claim, kind: moq_video::encode::Kind, - on_demand: bool, + sessions: Sessions, + shown: tokio::sync::watch::Receiver>, broadcast: moq_net::broadcast::Producer, studio: Catalog, catalog: moq_mux::catalog::Producer, /// Where the nested catalog writes its own track, which nobody reads: its - /// content reaches viewers through the studio catalog instead. + /// content reaches viewers through the local catalog instead. _scratch: moq_net::broadcast::Producer, } 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. - let config = moq_mux::catalog::Config::default() - .with_clock(publish.catalog.clock()) - .with_bandwidth(publish.bandwidth.clone()); + // The program's clock, so source and program timestamps share one timeline. + let config = moq_mux::catalog::Config::default().with_clock(publish.catalog.clock()); let catalog = moq_mux::catalog::Producer::new(&mut scratch, config)?; claim.owners.lock().unwrap().catalogs.insert(claim.id, catalog.clone()); Ok(Self { claim, kind: publish.kind.clone(), - on_demand: publish.on_demand, + sessions: publish.sessions.clone(), + shown: publish.shown.subscribe(), broadcast: publish.broadcast.clone(), studio: publish.catalog.clone(), catalog, @@ -191,14 +232,14 @@ impl<'a> Listing<'a> { }) } - /// Copy every update of this source's catalog into the studio's, until + /// Copy every update of this source's catalog into the local one, until /// either catalog closes. pub(crate) async fn mirror(&self) -> Result<(), Error> { let mut updates = self.catalog.consume()?; let mut studio = self.studio.clone(); while let Some(mut catalog) = updates.next().await? { // Media timing is the root's: the one clock, and the join interval an - // HLS export of the program numbers its segments on. + // HLS export numbers its segments on. catalog.media = Default::default(); // The claim is checked inside the mutation, so a newer source's // writes, made under the same lock, are never overwritten. @@ -222,7 +263,7 @@ impl<'a> Listing<'a> { impl Drop for Listing<'_> { fn drop(&mut self) { self.claim.owners.lock().unwrap().catalogs.remove(&self.claim.id); - // Fails only once the studio catalog closed, which took the entry with it. + // Fails only once the local catalog closed, which took the entry with it. let _ = self.studio.mutate(|root| { if self.claim.owned() { root.ext.studio.sources.remove(&self.claim.name); @@ -243,6 +284,8 @@ pub trait Encode { fn finish(self) -> Result, moq_video::Error>; /// [`Encoder::cut`](moq_video::encode::Encoder::cut). fn cut(&mut self) -> Result<(), moq_video::Error>; + /// [`Encoder::set_bitrate`](moq_video::encode::Encoder::set_bitrate). + fn set_bitrate(&mut self, bitrate: moq_net::bandwidth::Rate) -> Result<(), moq_video::Error>; } impl Encode for moq_video::encode::Encoder { @@ -261,6 +304,10 @@ impl Encode for moq_video::encode::Encoder { fn cut(&mut self) -> Result<(), moq_video::Error> { self.cut() } + + fn set_bitrate(&mut self, bitrate: moq_net::bandwidth::Rate) -> Result<(), moq_video::Error> { + self.set_bitrate(bitrate) + } } /// An encoder and the video track it publishes into. @@ -273,15 +320,19 @@ impl Encode for moq_video::encode::Encoder { pub struct Encoding { /// Taken only to abort or finish it, as the track ends. producer: Option, - encoder: Option, + /// The open encoder, and its place among the open sessions. + encoder: Option<(E, Session)>, + sessions: Sessions, } impl Encoding { - /// Publish into `producer`, with no encoder open yet. - pub fn new(producer: moq_video::encode::Producer) -> Self { + /// Publish into `producer`, with no encoder open yet, counting each one + /// opened in `sessions`. + pub fn new(producer: moq_video::encode::Producer, sessions: Sessions) -> Self { Self { producer: Some(producer), encoder: None, + sessions, } } @@ -290,16 +341,21 @@ impl Encoding { self.encoder.is_some() } - /// Whether anyone subscribes to the track. - pub(crate) fn wanted(&self) -> bool { - let producer = self.producer.as_ref().expect("open until it ends"); - producer.demand().wanted().is_some() + /// Whether anything reads the track. + pub fn watched(&self) -> bool { + self.demand().wanted().is_some() + } + + /// Who reads the track, which moves with it when a new encoder writes + /// another kind of picture. + pub fn demand(&self) -> moq_net::track::Demand { + self.producer.as_ref().expect("open until it ends").demand() } /// Encode with `encoder` from here on, after draining the one it replaces. pub fn open(&mut self, encoder: E) -> Result<(), Error> { self.drain()?; - self.encoder = Some(encoder); + self.encoder = Some((encoder, self.sessions.hold())); Ok(()) } @@ -309,24 +365,52 @@ impl Encoding { /// /// When no encoder is [open](Self::open). pub fn encode(&mut self, frame: &moq_video::Frame) -> Result<(), Error> { - let encoder = self.encoder.as_mut().expect("an encoder is open"); + let (encoder, _) = self.encoder.as_mut().expect("an encoder is open"); let encoded = encoder.encode(frame)?; self.producer.as_mut().expect("open until it ends").publish(&encoded)?; Ok(()) } + /// Open a group with the next picture, as a keyframe. + /// + /// # Panics + /// + /// When no encoder is [open](Self::open). + pub fn cut(&mut self) -> Result<(), Error> { + let (encoder, _) = self.encoder.as_mut().expect("an encoder is open"); + Ok(encoder.cut()?) + } + + /// Encode at `bitrate` from the next picture on. + /// + /// # Panics + /// + /// When no encoder is [open](Self::open). + pub fn set_bitrate(&mut self, bitrate: moq_net::bandwidth::Rate) -> Result<(), Error> { + let (encoder, _) = self.encoder.as_mut().expect("an encoder is open"); + Ok(encoder.set_bitrate(bitrate)?) + } + + /// Advertise whether the track is encoding, as the catalog's `enabled`, + /// so viewers pick another rendition while it is not. + pub fn set_enabled(&mut self, enabled: bool) -> Result<(), Error> { + let producer = self.producer.as_mut().expect("open until it ends"); + producer.set_enabled(enabled)?; + Ok(()) + } + /// Close the encoder, publishing what it still holds. pub fn drain(&mut self) -> Result<(), Error> { - if let (Some(encoder), Some(producer)) = (self.encoder.take(), &mut self.producer) { + if let (Some((encoder, _session)), Some(producer)) = (self.encoder.take(), &mut self.producer) { producer.publish(&encoder.finish()?)?; } Ok(()) } - /// Mark a gap in the track after [draining](Self::drain) the encoder, so - /// the next one [opened](Self::open) starts with a keyframe, even on a - /// backend that cannot [cut](Encode::cut). - pub(crate) fn discontinuity(&mut self) -> Result<(), Error> { + /// Mark a gap in the track after [draining](Self::drain) the encoder, which + /// gives its session back, so the next one [opened](Self::open) starts with + /// a keyframe, even on a backend that cannot [cut](Encode::cut). + pub fn discontinuity(&mut self) -> Result<(), Error> { self.drain()?; self.producer.as_mut().expect("open until it ends").discontinuity()?; Ok(()) @@ -338,7 +422,7 @@ impl Encoding { /// is a keyframe, so a viewer arriving after the gap starts there. pub fn pause(&mut self) -> Result<(), Error> { let producer = self.producer.as_mut().expect("open until it ends"); - if let Some(encoder) = &mut self.encoder { + if let Some((encoder, _)) = &mut self.encoder { producer.publish(&encoder.flush()?)?; encoder.cut()?; } @@ -380,25 +464,26 @@ impl Drop for Encoding { } } -/// A captured or replayed source's video, encoded frame by frame as it is shown. -pub(crate) struct Video { +/// A captured or replayed source's video, encoded frame by frame as it is +/// shown, while it is used. +pub(crate) struct Video<'a> { track: Encoding, + claim: &'a Claim, + /// The sources on program. + shown: tokio::sync::watch::Receiver>, config: moq_video::encode::Config, - /// Encodes only while the track is subscribed. - on_demand: bool, clock: moq_mux::Clock, /// Where the capture's private timeline starts on the studio clock, in /// microseconds, taken at the first frame. epoch: Option, } -impl Video { +impl<'a> Video<'a> { /// 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. + /// encode the picture at all fails here with the reason. pub(crate) async fn new( - listing: &Listing<'_>, + listing: &Listing<'a>, size: moq_video::Size, framerate: moq_video::FrameRate, color: Option, @@ -414,18 +499,20 @@ impl Video { .with_interval(config.gop.duration(config.framerate)); let producer = moq_video::encode::Producer::new(listing.broadcast.clone(), reserved, rendition)?; Ok(Self { - track: Encoding::new(producer), + track: Encoding::new(producer, listing.sessions.clone()), + claim: listing.claim, + shown: listing.shown.clone(), config, - on_demand: listing.on_demand, clock: listing.studio.clock(), epoch: None, }) } - /// Encode and publish `frame`, restamping it onto the studio clock, unless - /// it is on demand and nobody subscribes. The encoder opens on the first - /// frame it encodes, and reopens when the picture's size or color space - /// changes, which moves the rendition to a new track. + /// Restamp `frame` onto the program's clock, and encode and publish it + /// while the source is used: shown on program, or its track read. The + /// encoder opens on the first frame used and closes while none is, marking + /// the gap, and reopens when the picture's size or color space changes, + /// which moves the rendition to a new track. pub(crate) fn encode(&mut self, frame: &mut moq_video::Frame) -> Result<(), Error> { let captured = frame.timestamp.as_micros(); let epoch = *self @@ -433,13 +520,8 @@ 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)?; - if self.on_demand && !self.track.wanted() { - // The next subscriber starts on a fresh encoder's keyframe, and the - // session is free meanwhile; the gap is marked so nobody reads across it. - if self.track.is_open() { - self.track.discontinuity()?; - } - return Ok(()); + if !self.used() { + return self.release(); } let size = encodable(frame.size()); @@ -461,6 +543,60 @@ impl Video { } } + /// Whether the source is used: shown on program, or its track read. + fn used(&self) -> bool { + self.shown.borrow().contains(&self.claim.name) || self.track.watched() + } + + /// Wait until whether the source is used disagrees with whether its + /// encoder is open, so a capture that delivers nothing new, such as a + /// still screen, can act without waiting on a picture: [`release`] the + /// encoder when it is open and nothing uses the source, or bring a picture + /// when it is closed and the source goes into use. + /// + /// [`release`]: Self::release + #[cfg_attr( + not(any(feature = "capture", test)), + expect(dead_code, reason = "only a capture waits on it") + )] + pub(crate) async fn settle(&self) { + let mut demand = self.track.demand(); + let mut shown = self.shown.clone(); + shown.mark_unchanged(); + while self.used() == self.track.is_open() { + // The first look at the demand reports it as it is now. + tokio::select! { + changed = shown.changed() => if changed.is_err() { std::future::pending::<()>().await }, + wanted = demand.wanted_changed() => if wanted.is_err() { std::future::pending::<()>().await }, + } + } + } + + /// Whether the encoder is open. + #[cfg_attr(not(feature = "capture"), expect(dead_code, reason = "only a capture asks"))] + pub(crate) fn is_open(&self) -> bool { + self.track.is_open() + } + + /// Take pictures from a reopened capture, whose private timeline starts + /// over: the next picture maps it onto the program's clock afresh. + #[cfg_attr( + not(any(feature = "capture", test)), + expect(dead_code, reason = "only a capture reopens") + )] + pub(crate) fn reopened(&mut self) { + self.epoch = None; + } + + /// Close the encoder while nothing uses the source, marking the gap, which + /// gives the encode session back. + pub(crate) fn release(&mut self) -> Result<(), Error> { + if self.track.is_open() && !self.used() { + self.track.discontinuity()?; + } + Ok(()) + } + /// Mark a break, such as a gap in a replay: what follows does not continue /// what came before, and a viewer arriving during it waits for what follows /// rather than being served the last picture as if it were live. @@ -533,18 +669,15 @@ mod tests { use crate::source::test::{context, local, start, until, until_time, video}; use crate::source::{Sources, Status}; - /// A studio broadcast and catalog to publish into, and a reader of it. + /// Where sources publish, its catalog, and a reader of its broadcast. fn studio() -> (Publish, Catalog, moq_net::broadcast::Consumer) { - let mut broadcast = moq_net::broadcast::Info::new().produce(); - let consumer = broadcast.consume(); - let config = moq_mux::catalog::Config::default().with_catalog(Default::default()); - let catalog = Catalog::new(&mut broadcast, config).unwrap(); - let bandwidth = moq_net::bandwidth::Allocator::unlimited(); - let publish = Publish::new(broadcast, catalog.clone(), bandwidth, moq_video::encode::Kind::Auto); + let clock = moq_mux::Clock::new(); + let publish = Publish::new(clock, moq_video::encode::Kind::Auto, Sessions::default()).unwrap(); + let (catalog, consumer) = (publish.catalog.clone(), publish.consume()); (publish, catalog, consumer) } - /// Yield until `f` holds of the studio catalog's source listings. + /// Yield until `f` holds of the local catalog's source listings. async fn listed(catalog: &Catalog, f: impl Fn(&BTreeMap) -> bool) -> hang::Catalog { for _ in 0..1000 { let snapshot = catalog.snapshot(); @@ -650,10 +783,12 @@ mod tests { moq_video::Frame::new(moq_video::Surface::I420(i420), timestamp) } - /// A captured picture encodes into its source's own listing, and one that - /// changes size reopens the encoder rather than failing the track. + /// A captured picture encodes into its source's own listing only while the + /// source is used, shown on program or its track read, holding an encode + /// session only then, and one that changes size reopens the encoder rather + /// than failing the track. #[tokio::test] - async fn a_video_follows_its_pictures_size() { + async fn a_video_encodes_while_used() { local(async { let (publish, catalog, consumer) = studio(); let claim = publish.claim("camera"); @@ -665,42 +800,55 @@ mod tests { let width = renditions.next()?.coded_width; renditions.next().is_none().then_some(width).flatten() }; + let written = || publish.stats().values().map(|rendition| rendition.frames).sum::(); tokio::select! { result = listing.mirror() => panic!("the mirror ended: {result:?}"), () = async { let small = moq_video::Size::new(64, 48); let mut video = Video::new(&listing, small, framerate, None).await.unwrap(); - for millis in [0, 33, 66] { + for millis in [0, 33] { video.encode(&mut gray(small, millis)).unwrap(); } + assert_eq!((written(), publish.sessions.open()), (0, 0), "unused"); let snapshot = listed(&catalog, |sources| width(sources) == Some(64)).await; assert!(snapshot.video.renditions.is_empty(), "no source is a program rendition"); + publish.show(["camera".to_string()].into()); + for millis in [66, 100, 133] { + video.encode(&mut gray(small, millis)).unwrap(); + } + assert_eq!((written(), publish.sessions.open()), (3, 1), "on program"); + let large = moq_video::Size::new(80, 60); - for millis in [100, 133] { + for millis in [166, 200] { video.encode(&mut gray(large, millis)).unwrap(); } let snapshot = listed(&catalog, |sources| width(sources) == Some(80)).await; + assert_eq!(publish.sessions.open(), 1, "the resize closed the old encoder"); - // Removing the source ends its track cleanly. + publish.show(Default::default()); + video.encode(&mut gray(large, 233)).unwrap(); + assert_eq!((written(), publish.sessions.open()), (5, 0), "off program"); + + // A reader, such as a recording, uses it too. let name = snapshot.ext.studio.sources["camera"].video.renditions.keys().next().unwrap(); let mut track = consumer.track(name).unwrap().subscribe(None).await.unwrap(); + video.encode(&mut gray(large, 266)).unwrap(); + assert_eq!((written(), publish.sessions.open()), (6, 1), "read"); + + // Removing the source ends its track cleanly. drop(video); while let Some(mut group) = track.recv_group().await.unwrap() { while group.read_frame().await.unwrap().is_some() {} } + assert_eq!(publish.sessions.open(), 0); } => {} } // Every frame written counts for the Health panel, while the - // studio catalog lists the source. - let stats = publish.stats(); - assert_eq!( - stats.values().map(|rendition| rendition.frames).sum::(), - 5, - "{stats:?}" - ); + // local catalog lists the source. + assert_eq!(written(), 6); // A replacement takes the listing, and the old tracks with it. let _newer = publish.claim("camera"); assert!(publish.stats().is_empty()); @@ -710,13 +858,57 @@ mod tests { .await; } - /// An on-demand source encodes only while its track is subscribed, holding - /// no encoder between subscribers. + /// A source leaving program gives its encode session back without waiting + /// on a picture, since a still screen delivers none, and one going on + /// program while closed says so, so its capture can bring a picture. + #[tokio::test(start_paused = true)] + async fn a_video_follows_its_use_without_a_picture() { + local(async { + let (publish, _catalog, _consumer) = studio(); + let claim = publish.claim("display"); + let listing = Listing::new(&claim, &publish).unwrap(); + let framerate = moq_video::FrameRate::new(30, 1).unwrap(); + let size = moq_video::Size::new(64, 48); + let mut video = Video::new(&listing, size, framerate, None).await.unwrap(); + let wait = std::time::Duration::from_secs(1); + + assert!( + kio::clock::timeout(wait, video.settle()).await.is_err(), + "unused and closed" + ); + publish.show(["display".to_string()].into()); + kio::clock::timeout(wait, video.settle()) + .await + .expect("on program without a picture, so the capture brings one"); + video.encode(&mut gray(size, 0)).unwrap(); + assert_eq!(publish.sessions.open(), 1); + assert!(kio::clock::timeout(wait, video.settle()).await.is_err(), "on program"); + + publish.show(Default::default()); + kio::clock::timeout(wait, video.settle()).await.expect("off program"); + video.release().unwrap(); + assert_eq!(publish.sessions.open(), 0); + assert!(kio::clock::timeout(wait, video.settle()).await.is_err(), "closed"); + + // A capture reopened a minute on starts its own timeline over, which + // lands on the program's clock now rather than a minute back. + kio::clock::sleep(std::time::Duration::from_secs(60)).await; + video.reopened(); + publish.show(["display".to_string()].into()); + let now = publish.catalog.clock().now(); + let mut picture = gray(size, 0); + video.encode(&mut picture).unwrap(); + assert!(picture.timestamp >= now, "{:?} before {now:?}", picture.timestamp); + }) + .await; + } + + /// A source off program encodes only while its track is subscribed, such + /// as by a recording, holding no encoder between subscribers. #[tokio::test] - async fn an_on_demand_video_encodes_while_subscribed() { + async fn an_unshown_video_encodes_while_subscribed() { local(async { let (publish, catalog, consumer) = studio(); - let publish = publish.on_demand(); let claim = publish.claim("camera"); let listing = Listing::new(&claim, &publish).unwrap(); let framerate = moq_video::FrameRate::new(30, 1).unwrap(); @@ -798,6 +990,10 @@ mod tests { fn cut(&mut self) -> Result<(), moq_video::Error> { self.encoder.cut() } + + fn set_bitrate(&mut self, bitrate: moq_net::bandwidth::Rate) -> Result<(), moq_video::Error> { + self.encoder.set_bitrate(bitrate) + } } /// A pipelined encoder's last pictures land before a pause's gap rather @@ -825,7 +1021,7 @@ mod tests { let container = moq_mux::catalog::hang::Container::try_from(&rendition).unwrap(); let mut reader = moq_mux::container::Consumer::new(subscriber, container); - let mut track = Encoding::new(producer); + let mut track = Encoding::new(producer, Sessions::default()); let encoder = moq_video::encode::Encoder::new(&config).unwrap(); track .open(Pipelined { @@ -929,13 +1125,15 @@ mod tests { .await; } - /// A replay source publishes each replay as it shows, so viewers composing - /// the scene themselves see it too, gaps included. + /// A replay source on program encodes each replay onto its own track as it + /// shows, gaps included. #[tokio::test(start_paused = true)] async fn a_replay_is_published_as_it_plays() { local(async { let dir = tempfile::tempdir().unwrap(); let (publish, catalog, consumer) = studio(); + // On program, so its track encodes. + publish.show(["replay".to_string()].into()); let (mut context, _origin) = context(dir.path()); context.publish = Some(publish); let mut collection = Collection::new(); diff --git a/rs/moq-scene/src/replay.rs b/rs/moq-scene/src/replay.rs index dffd666..5469a7f 100644 --- a/rs/moq-scene/src/replay.rs +++ b/rs/moq-scene/src/replay.rs @@ -66,6 +66,10 @@ pub struct Take { pub video: Option<(String, VideoConfig)>, /// The audio rendition, under its catalog name. pub audio: Option<(String, AudioConfig)>, + /// How much of the span the video's pictures cover: less where the + /// rendition was not encoding, gaps inside the span included, but not for + /// pictures it dropped in between. + pub covered: Duration, /// Where the span starts on the copied broadcast's timeline. pub start: moq_net::Timestamp, /// Where it ends. @@ -82,8 +86,9 @@ impl Take { /// at the live edge may stop a frame or two short. /// /// Refuses a target with neither rendition, one served from another - /// broadcast, a length of zero or over [`MAX_LENGTH`], a broadcast that has - /// described neither rendition yet, and, over a wire that cannot address a + /// broadcast, a length of zero or over [`MAX_LENGTH`], a broadcast that holds + /// nothing of the span, such as one that has described neither rendition + /// yet or encoded nothing then, and, over a wire that cannot address a /// received track by time, a broadcast from a relay. pub fn new(target: &Target, length: Duration) -> Result { Self::cut(target, length, Cut::Decodable) @@ -116,25 +121,39 @@ impl Take { let mut broadcast = moq_net::broadcast::Info::new().produce(); let mut tracks = Vec::new(); let mut video = None; + let mut covered = Duration::ZERO; if let Some((name, config)) = &target.video { reference(config)?; - if let Some(track) = copy(target, &broadcast, Reference::track(config, name), start, cut)? { + let container = moq_mux::catalog::hang::Container::try_from(config)?; + let track = Reference::track(config, name); + if let Some((track, held)) = copy(target, &broadcast, track, start, cut, &container)? { tracks.push(track); - video = Some((name.clone(), config.clone())); + covered = held; + // What the take holds plays, even from a rendition the + // publisher has since stopped encoding. + let mut config = config.clone(); + config.enabled = true; + video = Some((name.clone(), config)); } } let mut audio = None; if let Some((name, config)) = &target.audio { reference(config)?; - if let Some(track) = copy(target, &broadcast, Reference::track(config, name), start, cut)? { + let container = moq_mux::catalog::hang::Container::try_from(config)?; + let track = Reference::track(config, name); + if let Some((track, _)) = copy(target, &broadcast, track, start, cut, &container)? { tracks.push(track); - audio = Some((name.clone(), config.clone())); + let mut config = config.clone(); + config.enabled = true; + audio = Some((name.clone(), config)); } } if tracks.is_empty() { - return Err(Error::Unsupported( - "the broadcast has described none of its tracks yet".into(), - )); + return Err(Error::Unsupported(format!( + "nothing of the last {} s is held: the broadcast encoded nothing then, or has not described its \ + tracks yet", + length.as_secs_f32() + ))); } // Listed so an export of the take finds its renditions. let mut listing = moq_mux::catalog::Producer::new(&mut broadcast, Default::default())?; @@ -150,6 +169,7 @@ impl Take { _listing: listing, video, audio, + covered, start, end, }) @@ -183,15 +203,18 @@ enum Cut { } /// Copy the span of `name` in `target`, as far as it is readable now, into a -/// track of `broadcast`, and end that track. `None` while its publisher has -/// not described the track, so nothing of it is held. +/// track of `broadcast`, and end that track, with how much of the span its +/// media covers. `None` when it holds no media of the span, only the markers +/// of a publisher's gaps if anything, or while its publisher has not +/// described the track. `container` reads its frames. fn copy( target: &Target, broadcast: &moq_net::broadcast::Producer, name: &str, start: moq_net::Timestamp, cut: Cut, -) -> Result, Error> { + container: &impl moq_mux::container::Container, +) -> Result, Error> { let range = moq_net::track::Range { start: moq_net::track::Anchor::Time(start), end: Some(moq_net::track::Anchor::Time(target.end)), @@ -212,14 +235,29 @@ fn copy( // Never stale: the take keeps its groups for as long as it lives. let info = source.info().clone().with_max_age(None); let mut track = broadcast.create_track(name, info)?; - match copy_held(&mut source, &mut track, target.end, cut) { - Ok(()) => track.finish()?, + let covered = match copy_held(&mut source, &mut track, start..=target.end, cut, container) { + Ok(Some(covered)) => covered, + Ok(None) => { + let _ = track.abort(moq_net::Error::Cancel); + return Ok(None); + } Err(err) => { let _ = track.clone().abort(moq_net::Error::Cancel); return Err(err); } + }; + track.finish()?; + Ok(Some((track, covered))) +} + +/// Whether `frame` is only the marker a publisher writes at a gap: one media +/// frame with nothing in it, as `container` reads it. A frame it cannot read +/// counts as media, since only the publisher's marker is known to be empty. +fn marker(frame: &moq_net::frame::Frame, container: &impl moq_mux::container::Container) -> bool { + match container.decode(frame.clone()) { + Ok(frames) => !frames.is_empty() && frames.iter().all(|frame| frame.payload.is_empty()), + Err(_) => false, } - Ok(Some(track)) } /// A group's frames as far as they were readable, and how reading it ended. @@ -229,13 +267,16 @@ struct Held { } /// Copy every group and frame `source` has ready, up to `end` as `cut` says, -/// into `track`. +/// into `track`, returning how much of `span` its media covers, or +/// `None` without media anywhere. fn copy_held( source: &mut moq_net::track::Subscriber, track: &mut moq_net::track::Producer, - end: moq_net::Timestamp, + span: std::ops::RangeInclusive, cut: Cut, -) -> Result<(), Error> { + container: &impl moq_mux::container::Container, +) -> Result, Error> { + let end = *span.end(); // Never woken: anything not ready now is left out. let now = kio::Waiter::noop(); // Groups come in the order they reached the studio, which over a session @@ -286,6 +327,39 @@ fn copy_held( } } + // The time inside the span the media covers, from each picture to the + // next. A gap's marker or a missing group breaks it, so the time a + // rendition was not encoding counts for nothing, while a rendition that + // dropped pictures still covers the time between those it kept. Its media + // may all be from before the span, as a clip cut back to its keyframe is. + let mut media = false; + let mut covered = Duration::ZERO; + let mut last: Option<(u64, Duration)> = None; + for (&sequence, held) in &held { + // Groups follow on only when no sequence is missing between them. + last = last.filter(|(previous, _)| sequence <= previous + 1); + for frame in &held.frames { + if marker(frame, container) { + last = None; + continue; + } + media = true; + let Some(at) = frame.timestamp.filter(|at| span.contains(at)).map(Duration::from) else { + continue; + }; + // Decode order can show a picture before the one decoded ahead of + // it, so only time past the newest shown counts. + last = match last { + Some((_, newest)) if at <= newest => Some((sequence, newest)), + Some((_, newest)) => { + covered += at - newest; + Some((sequence, at)) + } + None => Some((sequence, at)), + }; + } + } + // Groups the span covers that are not here are gaps, not groups to come. let copied: Vec = held.into_keys().collect(); for pair in copied.windows(2) { @@ -293,7 +367,7 @@ fn copy_held( track.drop_groups(pair[0] + 1..pair[1], moq_net::StreamError::NotFound); } } - Ok(()) + Ok(media.then_some(covered)) } #[cfg(test)] @@ -385,7 +459,12 @@ mod tests { #[tokio::test(start_paused = true)] async fn a_take_holds_its_span() { let (broadcast, _track) = broadcast(10, None); - let take = Take::new(&target(&broadcast, ts(7_500)), Duration::from_secs(3)).unwrap(); + let mut target = target(&broadcast, ts(7_500)); + // Disabled since, such as for lack of bandwidth, which leaves what it + // held playable. + target.video.as_mut().unwrap().1.enabled = false; + let take = Take::new(&target, Duration::from_secs(3)).unwrap(); + assert!(take.video.as_ref().unwrap().1.enabled); assert_eq!(take.start, ts(4_500)); assert_eq!( read(&take).await, @@ -500,6 +579,78 @@ mod tests { assert!(matches!(result, Err(Error::Unsupported(_)))); } + /// A take says how much of its span holds pictures, and one whose span + /// holds only the marker of a gap, such as a rendition the grant disabled, + /// is refused as holding nothing. + #[test] + fn a_take_of_a_gap_holds_no_pictures() { + use moq_mux::container::Container; + + let (broadcast, track) = broadcast(3, None); + let take = Take::new(&target(&broadcast, ts(2_900)), Duration::from_secs(2)).unwrap(); + // The span starts at 0.900, past group 0's pictures. + assert_eq!(take.covered, Duration::from_millis(1_666), "from 1.000 to 2.666"); + + // The marker a publisher writes as it stops encoding. + let config = target(&broadcast, ts(0)).video.unwrap().1; + let container = moq_mux::catalog::hang::Container::try_from(&config).unwrap(); + let mut group = track.create_group(moq_net::group::Info { sequence: 3 }).unwrap(); + let gap = moq_mux::container::Frame { + timestamp: ts(3_500), + payload: bytes::Bytes::new(), + keyframe: false, + duration: None, + }; + container.write(&mut group, &[gap]).unwrap(); + group.finish().unwrap(); + let result = Take::new(&target(&broadcast, ts(3_900)), Duration::from_millis(300)); + assert!(matches!(result, Err(Error::Unsupported(_))), "{:?}", result.err()); + } + + /// Coverage is time, not pictures: a rendition at half the frame rate + /// throughout covers more than one at the full rate that stopped partway, + /// and a missing group breaks it. + #[test] + fn coverage_counts_time_not_pictures() { + let take = |seconds: std::ops::Range, per_second: u64| { + let broadcast = moq_net::broadcast::Info::new().produce(); + let track = broadcast + .create_track("video", hang::container::track_info(hang::catalog::PRIORITY.video)) + .unwrap(); + for second in seconds { + let at: Vec = (0..per_second) + .map(|index| second * 1000 + index * 1000 / per_second) + .collect(); + write(&track, second, &at); + } + let take = Take::new(&target(&broadcast, ts(9_999)), Duration::from_secs(10)).unwrap(); + (take.covered, broadcast, track) + }; + let (slow, ..) = take(0..10, 15); + let (stopped, ..) = take(4..10, 30); + assert!(slow > stopped, "{slow:?} over {stopped:?}"); + assert_eq!(stopped, Duration::from_millis(5_966), "from 4.000 to 9.966"); + + // A group missing from the middle covers nothing. + let broadcast = moq_net::broadcast::Info::new().produce(); + let track = broadcast + .create_track("video", hang::container::track_info(hang::catalog::PRIORITY.video)) + .unwrap(); + write(&track, 0, &second(0)); + write(&track, 2, &second(2)); + let take = Take::new(&target(&broadcast, ts(2_999)), Duration::from_secs(3)).unwrap(); + assert_eq!(take.covered, Duration::from_millis(666 + 666)); + } + + /// A take of a track that wrote nothing in its span, such as a source that + /// was not encoding, is refused rather than played as nothing. + #[test] + fn a_take_of_nothing_is_refused() { + let (broadcast, _track) = broadcast(0, None); + let result = Take::new(&target(&broadcast, ts(7_500)), Duration::from_secs(3)); + assert!(matches!(result, Err(Error::Unsupported(_))), "{:?}", result.err()); + } + #[test] fn refuses_a_span_it_cannot_hold() { let (broadcast, _track) = broadcast(1, None); diff --git a/rs/moq-scene/src/scale.rs b/rs/moq-scene/src/scale.rs new file mode 100644 index 0000000..ae4cdf5 --- /dev/null +++ b/rs/moq-scene/src/scale.rs @@ -0,0 +1,192 @@ +//! Scale a texture down on the GPU, for the program's smaller rendition. + +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. +pub struct Scaler { + queue: wgpu::Queue, + device: wgpu::Device, + pipeline: wgpu::RenderPipeline, + layout: wgpu::BindGroupLayout, + sampler: wgpu::Sampler, + output: wgpu::Texture, + view: wgpu::TextureView, +} + +impl Scaler { + /// Scale [`FORMAT`] textures to `size`. + pub fn new(device: &wgpu::Device, queue: &wgpu::Queue, 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() + }); + // 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, + layout, + sampler, + output, + view, + } + } + + /// Draw `source`, a [`FORMAT`] texture created with `TEXTURE_BINDING`, as + /// it is now, returning the texture it is drawn into. Submitted before this + /// returns, so work queued after it reads the result. + /// + /// Bound per call rather than once, since the compositor replaces its + /// program texture when the canvas changes size. + pub fn scale(&self, source: &wgpu::Texture) -> &wgpu::Texture { + let source = source.create_view(&Default::default()); + let bind = self.device.create_bind_group(&wgpu::BindGroupDescriptor { + label: Some("moq-scene scale"), + layout: &self.layout, + entries: &[ + wgpu::BindGroupEntry { + binding: 0, + resource: wgpu::BindingResource::TextureView(&source), + }, + wgpu::BindGroupEntry { + binding: 1, + resource: wgpu::BindingResource::Sampler(&self.sampler), + }, + ], + }); + 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, &bind, &[]); + pass.draw(0..3, 0..1); + } + self.queue.submit([encoder.finish()]); + &self.output + } +} diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index 9466a7a..96ad01c 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -6,7 +6,8 @@ //! dropped, which happens when the source is removed or edited. //! //! With [`Context::publish`] set, each local source that shows a picture also -//! publishes it as its own track, web pages aside; see [`crate::publish`]. +//! publishes it as its own track in the studio's local broadcast, web pages +//! aside; see [`crate::publish`]. use std::collections::BTreeMap; use std::future::Future; @@ -406,7 +407,7 @@ pub struct Driver { plays: Option>, } -/// Where a source's own track goes, its claim on the studio catalog's listing +/// Where a source's own track goes, its claim on the local catalog's listing /// under its name, and what the track is doing. struct Track { publish: Publish, @@ -461,7 +462,7 @@ impl Driver { .ok() } - /// Keep the studio catalog listing what `listing` publishes. A catalog that + /// Keep the local catalog listing what `listing` publishes. A catalog that /// closes fails the track, and the source keeps running. async fn mirror(&self, listing: Option<&Listing<'_>>) { if let Some(listing) = listing @@ -586,14 +587,51 @@ impl Driver { None => None, }; + // A display or window delivers nothing while its screen stands still, + // so going into use reopens its capture, which starts with the picture + // as it is. A camera delivers its next picture within a frame anyway. + let still = matches!(source, Source::Display { .. } | Source::Window { .. }); let result = async { - while let Some(mut frame) = stream.read().await? { + // Whether a change of use is still to be acted on: not again until + // a picture arrives, since a closed encoder opens only on one. + let mut settling = true; + loop { + let settled = async { + match &track { + Some(video) if settling => video.settle().await, + _ => std::future::pending().await, + } + }; + // Leaving use gives the encode session back at once, and going + // into use brings a picture, even while the capture delivers + // nothing new. + let frame = tokio::select! { + frame = stream.read() => frame?, + () = settled => { + let Some(video) = &mut track else { continue }; + if video.is_open() { + if let Err(err) = video.release() { + self.fail_encoding(&mut track, err); + } + } else { + settling = false; + if still { + // Closed before the next opens, so the device is + // captured once. + drop(stream); + stream = moq_video::capture::open(&config).await?; + video.reopened(); + } + } + continue; + } + }; + settling = true; + let Some(mut frame) = frame else { break }; if let Some(video) = &mut track && let Err(err) = video.encode(&mut frame) { - tracing::warn!(source = %self.name, %err, "cannot encode the source"); - self.fail_track(Status::Failed(err.to_string())); - track.take().expect("encoding").abort(&err); + self.fail_encoding(&mut track, err); } self.show(Picture::Frame(frame)); } @@ -606,6 +644,17 @@ impl Driver { result } + /// Fail the source's own track with `err`, ending it, while the source + /// keeps drawing. + #[cfg(feature = "capture")] + fn fail_encoding(&self, track: &mut Option>, err: Error) { + tracing::warn!(source = %self.name, %err, "cannot encode the source"); + self.fail_track(Status::Failed(err.to_string())); + if let Some(video) = track.take() { + video.abort(&err); + } + } + #[cfg(feature = "capture")] async fn listen(&self, source: moq_audio::capture::Source) -> Result<(), Error> { let strip = self.strip.clone().expect("audio sources have a strip"); @@ -857,7 +906,7 @@ impl Driver { /// Ends once the whole span has played. /// /// With `listing`, the pictures are also encoded onto the source's own - /// track as they show, so viewers composing the scene see the replay too. + /// track as they show, while it is used. async fn replay(&self, play: Play, listing: Option<&Listing<'_>>) -> Result<(), Error> { set(&self.status, Status::Live); let take = &play.take; @@ -911,8 +960,8 @@ impl Driver { self.show(Picture::Frame(frame)); } // A stretch the broadcast was offline, paused, or lost - // replays as the gap it was, for viewers composing the - // scene too. Past the end, the replay is over anyway. + // replays as the gap it was, on its own track too. Past + // the end, the replay is over anyway. moq_video::decode::Read::Gap { timestamp, .. } | moq_video::decode::Read::End { end: timestamp } => { kio::clock::sleep_until(at(timestamp.clamp(take.start, take.end))).await; diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 6c10d3f..da4dc9e 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -1,16 +1,19 @@ //! Everything behind the window: the GPU, the sources, the program clock that //! draws the canvas and mixes the audio, and the broadcast it encodes into: -//! the program, and each local source's own track beside it. The relays serve -//! that broadcast, and recordings read it. +//! the program, at the canvas size and at half of it ([`crate::ladder`]). The +//! relays serve that broadcast, and recordings read it. Each local source's own +//! track stays in a broadcast of the studio's own, which recordings read too. //! //! 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 rendition's encoder its gate lets encode, scaled to the +//! rendition and stamped with the program clock. //! - **audio** advances the audio bus, and with it every mix, on the same //! clock. -//! - **audio-encode** encodes the mix on program into the program audio. +//! - **audio-encode** encodes the mix on program into the program audio, and +//! **audio-record** into the recording's own copy while a recording reads it. //! - **encode**: one per program rendition, which waits for each program copy -//! and encodes it. A new canvas size or frame rate starts a new rendition. +//! and encodes it. A new canvas size or frame rate starts new renditions. //! - Recordings write their files on the tokio runtime. //! - **source-***: one per running source, since a capture stream may not //! leave the thread that opened it. @@ -27,12 +30,14 @@ use std::time::Duration; use anyhow::Context; use bytes::Bytes; use kio::clock::Instant; -use moq_scene::publish::Encoding; +use moq_net::bandwidth; +use moq_scene::publish::{Encoding, Sessions}; use moq_scene::source::{self, Driver, Sources}; -use moq_scene::{Canvas, Collection, Compositor, Draft, Edit, History, wgpu}; +use moq_scene::{Canvas, Collection, Compositor, Draft, Edit, History, Scaler, wgpu}; use moq_video::render::export::{Exporter, Pending, Route}; use crate::autosave::{self, Autosave}; +use crate::ladder::{self, Gate, Joins, Step}; use crate::{health, mix, record}; /// The GPU everything shares: sources upload to it, the compositor draws on @@ -259,34 +264,69 @@ fn keep_on_air<'a>( pub struct Stats { /// Canvas frames drawn. pub rendered: AtomicU64, - /// Program frames handed to the encoder. - pub copied: AtomicU64, - /// Microseconds the last one waited on the GPU, and on a readback converted - /// pixels. - pub copy_us: AtomicU64, + /// What each of the program's video encoders costs: the canvas size, the + /// half size, and the recording's own canvas size. + pub renditions: [Load; 3], /// Whether the encoder reads the program on the GPU, rather than each frame /// being read back to the CPU. pub zero_copy: AtomicBool, - /// Program frames read back to the CPU for the encoder. - pub read_back: AtomicU64, - /// Program frames not encoded because the encoder still held every copy. - pub dropped: AtomicU64, - /// Program frames encoded. - pub encoded: AtomicU64, - /// Microseconds the last encode took. - pub encode_us: AtomicU64, - /// Whether anyone subscribes to the program video: a viewer through the - /// relay, or a recording. + /// Whether anyone subscribes to any of the video's renditions: a viewer + /// through the relay, or a recording. pub watched: AtomicBool, - /// Whether the program is handed to an encoder now: not while on demand - /// and unwatched, nor while a new rendition opens. + /// Whether the program is handed to the canvas size's encoder now: not + /// while on demand and unwatched, nor while a new rendition opens, nor + /// while the grant cannot fund it. pub encoding: AtomicBool, + /// The video encoders open, the program's and the sources'. + pub sessions: Sessions, + /// How many sizes the program's video has now: the canvas size, and the + /// half size unless the canvas is too small or its encoder refused it. + pub sizes: std::sync::atomic::AtomicUsize, + /// Each program track's share of the connection's bandwidth, by name. + grants: Mutex>, /// The encoder backend, once the program's current rendition opened one. encoder: Mutex, /// Why the video stopped, if it did, until a new rendition starts. video_error: Mutex>, /// Why the audio stopped, if it did. audio_error: Mutex>, + /// Why the recording's copy of the audio stopped, if it did, which leaves + /// the published audio running. + recording_audio_error: Mutex>, +} + +/// What one of the program's renditions costs. +#[derive(Default)] +pub struct Load { + /// Program frames handed to its encoder. + pub copied: AtomicU64, + /// Microseconds the last one waited on the GPU, and on a readback converted + /// pixels. + pub copy_us: AtomicU64, + /// Program frames read back to the CPU for its encoder. + pub read_back: AtomicU64, + /// Program frames not encoded because its encoder still held every copy. + pub dropped: AtomicU64, + /// Program frames encoded. + pub encoded: AtomicU64, + /// Microseconds the last encode took. + pub encode_us: AtomicU64, + /// Why this encoder stopped, or could not start, while the canvas size + /// carries on, until the next canvas. The canvas size's own failure stops + /// the video, which [`Stats::error`] reports. + error: Mutex>, +} + +impl Load { + /// Why this encoder stopped, if it did; see [`Load::error`]. + pub fn error(&self) -> Option { + self.error.lock().unwrap().clone() + } + + fn fail(&self, err: impl std::fmt::Display) { + tracing::warn!(%err, "a program rendition stopped"); + *self.error.lock().unwrap() = Some(err.to_string()); + } } /// The encoder backend the program's current rendition opened, if it has. @@ -326,6 +366,16 @@ impl Stats { } } + /// Each program track's share of the connection's bandwidth now, by name: + /// `None` while the connection has no estimate. + pub fn grants(&self) -> BTreeMap> { + let grants = self.grants.lock().unwrap(); + grants + .iter() + .map(|(name, grant)| (name.clone(), grant.peek())) + .collect() + } + /// Why the output stopped, if it did. pub fn error(&self) -> Option { let video = self.video_error.lock().unwrap().clone(); @@ -341,6 +391,18 @@ impl Stats { *self.video_error.lock().unwrap() = Some(format!("video: {err}")); } + /// The recording's audio copy failed, which stops what recordings hold but + /// not the published program. + fn fail_recording_audio(&self, err: impl std::fmt::Display) { + tracing::error!(%err, "the recording's audio stopped"); + *self.recording_audio_error.lock().unwrap() = Some(format!("{err}")); + } + + /// Why the recording's copy of the audio stopped, if it did. + pub fn recording_audio_error(&self) -> Option { + self.recording_audio_error.lock().unwrap().clone() + } + fn fail_audio(&self, err: impl std::fmt::Display) { tracing::error!(%err, "audio stopped"); *self.audio_error.lock().unwrap() = Some(format!("audio: {err}")); @@ -351,8 +413,9 @@ impl Stats { pub struct Output { /// The broadcast path on the relays. pub path: String, - /// Whether the program video is copied and encoded only while someone - /// watches or records it, rather than all the time. + /// Whether the program video's canvas size is copied and encoded only + /// while someone watches it, rather than all the time. The half size + /// always is. pub on_demand: bool, /// A connection to each relay, which reconnects on its own and dials only /// while held. The Health panel reads their states. @@ -368,14 +431,26 @@ pub struct Relay { /// The program's broadcast, open for as long as the engine runs. pub struct Program { - /// Where the broadcast is announced: the relay connections publish it, - /// and recordings read it. - origin: moq_net::origin::Producer, + /// Where the broadcast is announced, which the relay connections publish. + _origin: moq_net::origin::Producer, publication: moq_net::origin::Publication, /// What the broadcast lists and the clock its timestamps are on, for a /// replay of the program or a source. catalog: moq_scene::publish::Catalog, clock: moq_mux::Clock, + /// Where the local sources keep their own tracks, on the same clock. + sources: moq_scene::publish::Publish, + /// What a recording reads the program from, on the same clock. + recording: moq_net::origin::Publication, + /// The recording broadcast's catalog, held for as long as the engine runs, + /// so a recording still reading as the studio stops finds it open. + _recorded: moq_scene::publish::Catalog, + /// The origin the studio's own broadcasts are read through, which no relay + /// connection carries. + local: moq_net::origin::Producer, + /// 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: bandwidth::Reservation, } impl Drop for Program { @@ -385,22 +460,37 @@ impl Drop for Program { } impl Program { - /// What a replay of the program copies: its video and audio, up to now. - pub fn program(&self) -> moq_scene::replay::Target { + /// What a replay of the program could copy: its video at each size, + /// largest first, with its audio, up to now. + pub fn program(&self) -> Vec { let catalog = self.catalog.snapshot(); - self.target( - catalog.video.renditions.into_iter().next(), - catalog.audio.renditions.into_iter().next(), - ) + let audio = catalog.audio.renditions.into_iter().next(); + let mut video: Vec<_> = catalog.video.renditions.into_iter().collect(); + video.sort_by_key(|(_, config)| { + let pixels = u64::from(config.coded_width.unwrap_or(0)) * u64::from(config.coded_height.unwrap_or(0)); + std::cmp::Reverse(pixels) + }); + match video.is_empty() { + true => vec![self.target(None, audio)], + false => video + .into_iter() + .map(|video| self.target(Some(video), audio.clone())) + .collect(), + } } /// What a replay of the local source `name` copies: the video of its own /// track, up to now, or `None` while it publishes no video. pub fn source(&self, name: &str) -> Option { - let mut catalog = self.catalog.snapshot(); + let mut catalog = self.sources.catalog(); let listed = catalog.ext.studio.sources.remove(name)?; let video = listed.video.renditions.into_iter().next()?; - Some(self.target(Some(video), None)) + Some(moq_scene::replay::Target { + broadcast: self.sources.consume(), + video: Some(video), + audio: None, + end: self.clock.now(), + }) } fn target( @@ -415,6 +505,20 @@ impl Program { end: self.clock.now(), } } + + /// Encode the local sources the program shows, from `state`, whether or + /// not anything reads their tracks, so a replay of one has its past. + fn show(&self, state: &State) { + let collection = &state.collection; + let scenes = state.compositor.on_air().filter_map(|scene| collection.scene(scene)); + let shown = scenes + .flat_map(|scene| &scene.items) + .chain(&collection.top) + .filter(|item| item.visible) + .map(|item| collection.capture(&item.source).to_string()) + .collect(); + self.sources.show(shown); + } } /// The running studio. @@ -461,7 +565,7 @@ pub struct Launch { pub relays: Vec, /// The broadcast path the program is published at. pub broadcast: String, - /// Whether the program video encodes only while someone watches or records. + /// Whether the program video encodes only while someone watches. pub on_demand: bool, /// The directory recordings are saved in. pub recordings: PathBuf, @@ -523,12 +627,19 @@ impl Engine { connections, links, reports, + on_demand, }; (Some(live), None) }; let (program, publish) = publish(&gpu, &stats, origin, &path, canvas, live.as_ref()).await?; - let source = moq_mux::Source::new(program.origin.consume(), program.publication.broadcast().consume()); - let recorder = record::Recorder::new(source, &recordings, autosave.path())?; + // Nothing in either local broadcast refers to another. + let local = |broadcast: moq_net::broadcast::Consumer| moq_mux::Source::new(program.local.consume(), broadcast); + let recorder = record::Recorder::new( + local(program.recording.broadcast().consume()), + local(program.sources.consume()), + &recordings, + autosave.path(), + )?; let output = live.map(|live| Output { path: path.to_string(), on_demand, @@ -576,8 +687,9 @@ impl Engine { }); let Publish { renditions, - rendition, + ladder, audio, + recorded_audio, clock, health, sources: _, @@ -589,7 +701,7 @@ impl Engine { let epoch = Duration::from(clock.capture(start)?); let program = engine.clone(); let thread = clocked("program".into(), async move || { - program.run_program(start, epoch, renditions, rendition).await + program.run_program(start, epoch, renditions, ladder).await })?; engine.threads.lock().unwrap().push(thread); let mix = engine.clone(); @@ -597,14 +709,20 @@ impl Engine { let rate = config.sample_rate; let thread = clocked("audio".into(), async move || mix.run_audio(start, config, mixer).await)?; engine.threads.lock().unwrap().push(thread); - // Its own thread, so an encoder that falls behind skips its own audio - // rather than holding up the bus. - let encode = engine.clone(); - let follow = engine.program_audio.follow(); - let thread = clocked("audio-encode".into(), async move || { - encode.encode_audio(epoch, rate, follow, audio).await - })?; - engine.threads.lock().unwrap().push(thread); + // Each on its own thread, so an encoder that falls behind skips its own + // audio rather than holding up the bus or the other. + let audio = [ + ("audio-encode", Audio::Program(audio)), + ("audio-record", Audio::Recording(recorded_audio)), + ]; + for (name, audio) in audio { + let encode = engine.clone(); + let follow = engine.program_audio.follow(); + let thread = clocked(name.into(), async move || { + encode.encode_audio(epoch, rate, follow, audio).await + })?; + engine.threads.lock().unwrap().push(thread); + } let save = engine.clone(); clocked("autosave".into(), async move || save.autosave.run(&save.state).await)?; tokio::spawn(engine.clone().grade(health::Sampler::new(health))); @@ -751,11 +869,13 @@ impl Engine { }) .collect() }; + let canvas = &stats.renditions[0]; health::Local { rendered: stats.rendered.load(Ordering::Relaxed), - dropped: stats.dropped.load(Ordering::Relaxed), - encoded: stats.encoded.load(Ordering::Relaxed), + dropped: canvas.dropped.load(Ordering::Relaxed), + encoded: canvas.encoded.load(Ordering::Relaxed), encoding: stats.encoding.load(Ordering::Relaxed), + grants: stats.grants(), watched: stats.watched.load(Ordering::Relaxed), encoder: stats.encoder(), failed: stats.error(), @@ -764,39 +884,32 @@ 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 while live and - /// not on demand, otherwise only while anyone watches or records it. + /// to each rendition's encoder, stamped `epoch` past `start`, while its gate + /// lets it encode: the canvas size every frame while live and not on + /// demand, otherwise only while anyone watches it, the half size only + /// while anyone watches it, and the recording's own only while recording. /// - /// A new canvas size or frame rate ends `rendition`, the one it was drawn - /// at, and opens another in `renditions`, since a track never changes what - /// it carries. Drawing outlives the encoder: a failed output stops copying + /// A new canvas size or frame rate ends `ladder`, the renditions it was + /// drawn at, and opens others in `renditions`, since a track never changes + /// what it carries. Drawing outlives the encoders: a failed output stops copying /// the program, and the preview and program keep moving. /// /// Runs until the studio [stops](Self::stop), and returns once every /// rendition's encode thread has finished its track. - async fn run_program( - &self, - start: Instant, - epoch: Duration, - renditions: Renditions, - mut rendition: Option, - ) { + async fn run_program(&self, start: Instant, epoch: Duration, renditions: Renditions, mut ladder: Option) { let mut canvas = self.state.lock().unwrap().compositor.canvas(); let mut clock = program_clock(start, canvas); - // Live on a relay and not on demand; otherwise a recording or a viewer - // asks for the video. - let always = self.output.as_ref().is_some_and(|output| !output.on_demand); // One probe at a time, with the canvas it is for, since a GPU caps its // encode sessions. let mut probing: Option<(Canvas, std::thread::JoinHandle>)> = None; // The encode threads of renditions ended or failed since, whose // sessions the next probe waits to be closed. let mut ending = Vec::new(); - // Whether the canvas drawn has its rendition, or failed to open one, so - // a failure is not retried until the canvas changes. + // Whether the canvas drawn has its renditions, or failed to open them, + // so a failure is not retried until the canvas changes. let mut settled = true; - if let Some(rendition) = &rendition { - rendition.adopted(&self.stats); + if let Some(ladder) = &ladder { + ladder.adopted(&self.stats); } loop { @@ -804,7 +917,7 @@ impl Engine { if self.stopped.load(Ordering::Relaxed) { // Without a sender, each encoder publishes what it was sent and // still holds, then finishes its track. - ending.extend(rendition.take().map(Rendition::end)); + ending.extend(ladder.take().into_iter().flat_map(|ladder| ladder.end(&self.stats))); for thread in ending { if thread.join().is_err() { tracing::error!("an encode thread panicked"); @@ -829,6 +942,7 @@ impl Engine { tracing::error!(%err, "cannot draw the canvas"); continue; } + self.program.show(state); (state.compositor.program().clone(), state.compositor.canvas()) }; self.stats.rendered.fetch_add(1, Ordering::Relaxed); @@ -839,9 +953,9 @@ impl Engine { } canvas = drawn; settled = false; - // Ended before the next opens, so the two never hold encode + // Ended before the next open, so the two never hold encode // sessions at once. - ending.extend(rendition.take().map(Rendition::end)); + ending.extend(ladder.take().into_iter().flat_map(|ladder| ladder.end(&self.stats))); self.stats.retire_encoder(); } // Probed beside the program clock, so drawing carries on. @@ -855,7 +969,7 @@ impl Engine { match probe.and_then(|probe| renditions.start(probe)) { Ok((started, _)) => { started.adopted(&self.stats); - rendition = Some(started); + ladder = Some(started); } Err(err) => self.stats.fail_video(format!("{err:#}")), } @@ -872,41 +986,36 @@ impl Engine { } } - let watched = rendition - .as_ref() - .is_some_and(|rendition| rendition.watched.load(Ordering::Relaxed)); - self.stats.watched.store(watched, Ordering::Relaxed); - let wanted = rendition.is_some() && (always || watched); - self.stats.encoding.store(wanted, Ordering::Relaxed); - let Some(current) = &mut rendition else { continue }; - let was = std::mem::replace(&mut current.encoding, wanted); - if !wanted { - // Behind every frame already sent, so the break lands after them. - // A send fails only once the encoder stopped, having reported why. - if was && current.frames.send(Video::Pause).is_err() { - ending.extend(rendition.take().map(Rendition::end)); - } + let Some(current) = &mut ladder else { + self.stats.watched.store(false, Ordering::Relaxed); + self.stats.encoding.store(false, Ordering::Relaxed); continue; - } + }; + let watched = current + .rungs + .iter() + .any(|rung| rung.role.published() && rung.gate.watched()); + self.stats.watched.store(watched, Ordering::Relaxed); let Ok(timestamp) = moq_net::Timestamp::try_from(epoch + tick.saturating_duration_since(start)) else { continue; }; - match current.exporter.copy(&program, timestamp) { - // The pool bounds the frames in flight, so this waits only behind - // a pause the encoder has not reached yet. - Ok(Some(pending)) => { - if current.frames.send(Video::Frame(pending)).is_err() { - ending.extend(rendition.take().map(Rendition::end)); - } - } - Ok(None) => { - self.stats.dropped.fetch_add(1, Ordering::Relaxed); + let mut encoding = false; + let mut failed = Vec::new(); + for (index, rung) in current.rungs.iter_mut().enumerate() { + let role = rung.role; + match rung.feed(&self.stats.renditions[role as usize], &program, timestamp) { + Some(step) => encoding |= role == Role::Canvas && matches!(step, Step::Encode(_)), + None => failed.push((index, role)), } - // The canvas and the exporter share a size and format, so this is - // the GPU failing: the video stops, and the canvas keeps drawing. - Err(err) => { - let _ = current.frames.send(Video::Failed(err)); - ending.extend(rendition.take().map(Rendition::end)); + } + self.stats.encoding.store(encoding, Ordering::Relaxed); + // An encoder stops only having reported why. Without the canvas + // size the video stops; without another rendition it carries on. + if failed.iter().any(|(_, role)| *role == Role::Canvas) { + ending.extend(ladder.take().into_iter().flat_map(|ladder| ladder.end(&self.stats))); + } else { + for (index, _) in failed.into_iter().rev() { + ending.push(current.remove(index, &self.stats)); } } } @@ -942,19 +1051,37 @@ impl Engine { } } - /// Encode the mix `follow` reads into the program audio, `producer`, - /// stamped `epoch` past the bus's start at `rate`, until the bus stops, and - /// finish the track. - async fn encode_audio( - &self, - epoch: Duration, - rate: u32, - mut follow: mix::Follow, - mut producer: moq_audio::encode::Producer, - ) { + /// Encode the mix `follow` reads into `audio`, stamped `epoch` past the + /// bus's start at `rate`, until the bus stops, and finish the track. + /// + /// The recording's copy encodes only while a recording reads it, and marks + /// the gap as the last one stops, so the next starts on the program clock + /// rather than where the samples left off. + async fn encode_audio(&self, epoch: Duration, rate: u32, mut follow: mix::Follow, audio: Audio) { + let (mut producer, recording) = match audio { + Audio::Program(producer) => (producer, false), + Audio::Recording(producer) => (producer, true), + }; + // The recording's copy failing leaves the published audio running, so + // it is the recording's to report. + let fail = |err: moq_audio::Error| match recording { + false => self.stats.fail_audio(err), + true => self.stats.fail_recording_audio(err), + }; let rate = u128::from(rate); let mut next = None; + let mut wanted = false; while let Some(block) = follow.read().await { + let was = std::mem::replace(&mut wanted, !recording || producer.demand().wanted().is_some()); + if !wanted { + if was && let Err(err) = producer.discontinuity() { + fail(err); + producer.abort(moq_net::StreamError::Internal.into()); + return; + } + next = None; + continue; + } // Audio this encoder fell too far behind to read leaves a gap in // the track rather than pulling what follows early. if next.is_some_and(|next| next != block.position) { @@ -968,46 +1095,57 @@ impl Engine { }; let data: Vec = block.data.iter().flat_map(|sample| sample.to_le_bytes()).collect(); if let Err(err) = producer.write(&moq_audio::Frame::new(Bytes::from(data), timestamp)) { - self.stats.fail_audio(err); - // Viewers see the audio fail rather than go away unfinished. + fail(err); + // Readers see the audio fail rather than go away unfinished. producer.abort(moq_net::StreamError::Internal.into()); return; } } if let Err(err) = producer.finish() { - self.stats.fail_audio(err); + fail(err); } } } +/// One encoder of the mix on program: the published program audio, or the +/// recording's own copy. +enum Audio { + Program(moq_audio::encode::Producer), + Recording(moq_audio::encode::Producer), +} + /// The relay connections the program goes live on, the preferred first, and /// the origin their viewers' reports and the relays' stats arrive on. struct Live { connections: Vec, links: Vec, reports: moq_net::origin::Consumer, + /// The canvas size encodes only while someone watches it. + on_demand: bool, } /// The encoders behind the program's broadcast, handed to their threads. struct Publish { - /// Where the program video is published, a rendition at a time. + /// Where the program video is published, a canvas at a time. renditions: Renditions, - /// The rendition of the canvas the studio started with, unless the canvas + /// The renditions of the canvas the studio started with, unless the canvas /// could not be encoded without a relay. - rendition: Option, + ladder: Option, + /// The program audio's encoder. audio: moq_audio::encode::Producer, + /// The recording's own copy of it. + recorded_audio: moq_audio::encode::Producer, /// The catalog's clock, which the codec producers write timestamps /// against verbatim. clock: moq_mux::Clock, /// What the health sampler grades, while live on a relay. health: Option, - /// Where each local source publishes its own track. + /// Where each local source keeps its own track. sources: moq_scene::publish::Publish, } /// Announce the program broadcast at `path` on `origin` under one instance, -/// which `live`'s relays publish when there are any. The broadcast also -/// carries each local source's own track. +/// which `live`'s relays publish when there are any. async fn publish( gpu: &Gpu, stats: &Arc, @@ -1018,16 +1156,16 @@ async fn publish( ) -> anyhow::Result<(Program, Publish)> { let publication = origin.create_broadcast(path).context("cannot create the broadcast")?; let mut broadcast = publication.broadcast().clone(); - // Each track claims its bitrate on the first relay's estimate, which that - // connection then probes to fund: the bandwidth the stream is granted. An - // allocator spans one connection, so the failover's link is never probed. + // Each track reserves its ceiling on the first relay's estimate, which + // that connection then probes to fund: the bandwidth the stream is granted. + // An allocator spans one connection, so the failover's link is never + // probed. The catalog claims nothing itself, which would count each track + // twice. Without a relay nothing divides any bandwidth. let bandwidth = match live { Some(live) => live.connections[0].allocator(), None => moq_net::bandwidth::Allocator::unlimited(), }; - let catalog = moq_mux::catalog::Config::default() - .with_bandwidth(bandwidth.clone()) - .with_catalog(Default::default()); + let catalog = moq_mux::catalog::Config::default().with_catalog(Default::default()); let catalog = moq_scene::publish::Catalog::new(&mut broadcast, catalog)?; let clock = catalog.clock(); @@ -1046,16 +1184,11 @@ async fn publish( // session the GPU refuses fails that source loudly instead of falling to // software. When the program could not use hardware, such as NVENC beside an // integrated GPU it cannot read, a source may still reach it from the CPU. - let kind = match probe.as_ref().map(|probe| &probe.config.kind) { + let kind = match probe.as_ref().map(|probe| &probe.rungs[0].config.kind) { Some(moq_video::encode::Kind::Named(name)) if name == "openh264" => moq_video::encode::Kind::Auto, _ => moq_video::encode::Kind::Hardware, }; - let mut sources = moq_scene::publish::Publish::new(broadcast.clone(), catalog.clone(), bandwidth, kind); - // Without a relay only a recording reads a source's track, and nothing - // fetches a past it was not recording. - if live.is_none() { - sources = sources.on_demand(); - } + let sources = moq_scene::publish::Publish::new(clock.clone(), kind, stats.sessions.clone())?; // Held until the video and audio are listed, so the first catalog carries // both and any solicitation. let reserved = catalog.reserve(); @@ -1071,21 +1204,45 @@ async fn publish( } None => None, }; + // What a recording reads the program from: its own canvas-size encoder, + // at full quality whatever the grant, and the program audio, in a + // broadcast no relay sees: announced on an origin of its own, which no + // relay connection carries, so a recording reads it as it would the + // published program. + let local = moq_tokio::origin::spawn(); + let recording = local + .create_broadcast("recording") + .context("cannot create the recording broadcast")?; + recording + .announce(Default::default()) + .context("cannot announce the recording broadcast")?; + let mut recorder = recording.broadcast().clone(); + let config = moq_mux::catalog::Config::default() + .with_catalog(Default::default()) + .with_clock(clock.clone()); + let recorded = moq_scene::publish::Catalog::new(&mut recorder, config)?; let renditions = Renditions { gpu: gpu.clone(), broadcast: broadcast.clone(), timebase: catalog.timebase(), + recording: Recording { + broadcast: recorder.clone(), + timebase: recorded.timebase(), + }, + bandwidth: bandwidth.clone(), + // Live on a relay and not on demand; otherwise only a viewer asks for + // the video. + on_demand: live.is_none_or(|live| live.on_demand), + live: live.is_some(), stats: stats.clone(), - runtime: tokio::runtime::Handle::current(), }; - let rendition = match probe { + let ladder = match probe { Some(probe) => { - let (rendition, opened) = renditions.start(probe)?; + let (ladder, opened) = renditions.start(probe)?; // 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. + // its encode session even while unwatched. match opened.await.context("the encode thread stopped")? { - Ok(()) => Some(rendition), + Ok(()) => Some(ladder), // Its encode thread reported why. Err(_) if live.is_none() => None, Err(err) => return Err(err), @@ -1103,7 +1260,14 @@ async fn publish( options.config.bitrate = Some(AUDIO_BITRATE); let graded = catalog.clone(); let listed = catalog.clone(); + let recorded_audio = moq_audio::encode::Producer::new(&mut recorder, recorded.clone(), input.clone(), &options)?; let audio = moq_audio::encode::Producer::new(&mut broadcast, catalog, input, &options)?; + let audio_reservation = bandwidth.reserve(&audio.demand(), AUDIO_BITRATE); + stats + .grants + .lock() + .unwrap() + .insert(audio.track_name().to_string(), audio_reservation.consume()); drop(reserved); // One instance on every relay, so a viewer moving between them resumes @@ -1137,15 +1301,21 @@ async fn publish( Ok(( Program { - origin, + _origin: origin, publication, catalog: listed, clock: clock.clone(), + sources: sources.clone(), + recording, + _recorded: recorded, + local, + _audio: audio_reservation, }, Publish { renditions, - rendition, + ladder, audio, + recorded_audio, clock, health, sources, @@ -1153,8 +1323,37 @@ async fn publish( )) } -/// Where a canvas's program video is published: one rendition at a time, -/// each its own track, since a track never changes what it carries. +/// What one of the program's video encoders is for, and its place in +/// [`Stats::renditions`]. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum Role { + /// The canvas size, published. + Canvas = 0, + /// Half of it, published. + Half = 1, + /// The canvas size again for a recording, in the studio's own recording + /// broadcast, at full quality whatever the bandwidth grant. + Record = 2, +} + +impl Role { + /// Whether viewers on a relay see it. + fn published(self) -> bool { + self != Self::Record + } +} + +/// Where a recording reads the program from: a broadcast the studio keeps to +/// itself, holding the recording's own canvas-size video and a copy of the +/// program audio, listed in a catalog of its own on the program's clock. +#[derive(Clone)] +struct Recording { + broadcast: moq_net::broadcast::Producer, + timebase: moq_mux::catalog::Timebase, +} + +/// Where a canvas's program video is published, a canvas at a time, each +/// rendition its own track, since a track never changes what it carries. #[derive(Clone)] struct Renditions { gpu: Gpu, @@ -1162,71 +1361,197 @@ struct Renditions { /// Shared by every rendition, so each lands on the catalog's clock the same /// way and a switch keeps the video in step with the audio. timebase: moq_mux::catalog::Timebase, + /// Where the recording's own encoder writes. + recording: Recording, + /// The connection's bandwidth, which each published rendition follows its + /// share of. + bandwidth: bandwidth::Allocator, + /// Whether the canvas size encodes only while someone watches. + on_demand: bool, + /// Whether the program goes out on a relay, where viewers can watch it. + live: bool, stats: Arc, - /// Where each rendition's viewers are followed. - runtime: tokio::runtime::Handle, } impl Renditions { - /// List `probe`'s rendition in the catalog under a new track, and start its - /// encode thread, which reports on the receiver once its encoder opens. - fn start(&self, probe: Probe) -> anyhow::Result<(Rendition, tokio::sync::oneshot::Receiver>)> { - let Probe { - exporter, - config, - video, - } = probe; - // Before its encode thread starts, which reports its own failure. + /// List each of `probe`'s renditions in its catalog under a new track, and + /// start their encode threads. The receiver reports once the canvas size's + /// encoder opens; the half size opens its own once watched, and the + /// recording's once recorded. + fn start(&self, probe: Probe) -> anyhow::Result<(Ladder, tokio::sync::oneshot::Receiver>)> { + // Before the encode threads start, which report their own failures. *self.stats.video_error.lock().unwrap() = None; - let reserved = self - .timebase - .reserve() - .with_interval(config.gop.duration(config.framerate)); - let producer = moq_video::encode::Producer::new(self.broadcast.clone(), reserved, video)?; - let watched = Arc::new(AtomicBool::new(false)); - self.runtime.spawn(watch(producer.demand(), watched.clone())); - - // One more than the exporter's pool, for a pause behind a full pool. - let (frames, pending) = mpsc::sync_channel(Exporter::DEPTH + 1); - let zero_copy = exporter.route() == Route::Gpu; - let stats = self.stats.clone(); + for (role, load) in [Role::Half, Role::Record] + .into_iter() + .map(|role| (role, &self.stats.renditions[role as usize])) + { + let refused = probe.refused.iter().find(|(refused, _)| *refused == role); + *load.error.lock().unwrap() = refused.map(|(_, reason)| reason.clone()); + if let Some((_, reason)) = refused { + tracing::warn!(?role, %reason, "the encoder refused a rendition"); + } + } + // The published sizes, largest first, each funded over the next. + let ceilings: Vec<_> = probe + .rungs + .iter() + .filter(|probed| probed.role.published()) + .map(|probed| ladder::ceiling(probed.config.size(), probed.config.framerate)) + .collect(); let (opened, open) = tokio::sync::oneshot::channel(); - let number = stats.rendition(); - let thread = std::thread::Builder::new() - .name("encode".into()) - .spawn(move || encode(&stats, number, opened, pending, producer, config, zero_copy))?; - let rendition = Rendition { - exporter, - frames, - watched, - encoding: false, - thread, - }; - Ok((rendition, open)) + let mut opened = Some(opened); + let number = self.stats.rendition(); + let mut rungs = Vec::with_capacity(probe.rungs.len()); + for probed in probe.rungs { + let Probed { + role, + exporter, + config, + video, + } = probed; + let (broadcast, timebase) = match role { + Role::Record => (&self.recording.broadcast, &self.recording.timebase), + _ => (&self.broadcast, &self.timebase), + }; + let reserved = timebase.reserve().with_interval(config.gop.duration(config.framerate)); + let suffix = match &video.codec { + hang::catalog::VideoCodec::H264(_) => ".avc3", + hang::catalog::VideoCodec::H265(_) => ".hev1", + other => anyhow::bail!("{other} is not a codec the studio publishes"), + }; + let rank = role as usize; + let priority = match role { + Role::Record => hang::catalog::PRIORITY.video, + _ => ladder::priority(rank, ceilings.len()), + }; + let info = reserved.track_info(priority).with_refresh(ladder::REFRESH); + let track = broadcast.unique_track(suffix, info)?; + let producer = moq_video::encode::Producer::with_track(track.clone(), reserved, video)?; + let gate = match role { + // Reads nothing from the connection, so it always encodes at its + // ceiling, and only while recorded. + Role::Record => Gate::new( + &bandwidth::Allocator::unlimited(), + producer.demand(), + ladder::ceiling(config.size(), config.framerate), + None, + false, + ), + _ => { + let gate = Gate::new( + &self.bandwidth, + producer.demand(), + ceilings[rank], + ceilings.get(rank + 1).copied(), + role == Role::Canvas && !self.on_demand, + ); + self.stats + .grants + .lock() + .unwrap() + .insert(track.name().to_string(), gate.grant()); + gate + } + }; + + // One more than the exporter's pool, for a skip behind a full pool. + let (frames, pending) = mpsc::sync_channel(Exporter::DEPTH + 1); + let job = Job { + stats: self.stats.clone(), + role, + rendition: (role == Role::Canvas).then_some(number), + zero_copy: exporter.route() == Route::Gpu, + producer, + track, + config, + // The canvas size holds its session while it skips frames. + hold: role == Role::Canvas, + }; + // Live, the canvas size holds its session from the start, so a + // source past the GPU's session cap is the one that fails. Without + // a relay nobody can watch it, so it opens nothing. + let opened = match role { + Role::Canvas if self.live => opened.take(), + _ => None, + }; + let thread = std::thread::Builder::new() + .name("encode".into()) + .spawn(move || encode(job, opened, pending))?; + let scaler = (role == Role::Half).then(|| Scaler::new(&self.gpu.device, &self.gpu.queue, exporter.size())); + rungs.push(Rung { + role, + gate, + scaler, + exporter, + frames, + skipped: None, + thread, + }); + } + if let Some(opened) = opened { + let _ = opened.send(Ok(())); + } + Ok((Ladder { rungs }, open)) } } -/// An encoder proven to take this GPU's export of a canvas, before its -/// rendition is listed. +/// Encoders proven to take this GPU's export of a canvas for each of its +/// renditions, before they are listed. struct Probe { + /// The canvas size first. + rungs: Vec, + /// Why the encoder refused a rendition the program does without. + refused: Vec<(Role, String)>, +} + +/// One rendition of a [`Probe`]. +struct Probed { + role: Role, exporter: Exporter, config: moq_video::encode::Config, video: hang::catalog::VideoConfig, } impl Probe { - /// Open a throwaway encoder of `canvas` on what `gpu` exports, which also - /// settles whether the encoder reads the program on the GPU. + /// Open a throwaway encoder of each of `canvas`'s renditions on what `gpu` + /// exports, which also settles whether the encoder reads the program on + /// the GPU. The canvas size picks the encoder, which the others use too. async fn new(gpu: &Gpu, canvas: Canvas) -> anyhow::Result { - let mut config = moq_video::encode::Config::new(canvas.width, canvas.height, canvas.framerate()); - let (exporter, video) = Exporter::new(&gpu.device, &gpu.queue, &mut config) - .await - .with_context(|| format!("cannot encode a {}x{} canvas", canvas.width, canvas.height))?; - Ok(Self { - exporter, - config, - video, - }) + let framerate = canvas.framerate(); + let size = moq_video::Size::new(canvas.width, canvas.height); + let sizes = ladder::sizes(size); + let roles = [Role::Canvas, Role::Half] + .into_iter() + .zip(sizes) + .chain([(Role::Record, size)]); + let mut kind = moq_video::encode::Kind::Auto; + let mut rungs = Vec::new(); + let mut refused = Vec::new(); + for (role, size) in roles { + let mut config = moq_video::encode::Config::new(size.width, size.height, framerate); + config.bitrate = Some(ladder::ceiling(size, framerate)); + config.kind = kind.clone(); + let probed = Exporter::new(&gpu.device, &gpu.queue, &mut config) + .await + .with_context(|| format!("cannot encode a {}x{} canvas", size.width, size.height)); + let (exporter, video) = match probed { + Ok(probed) => probed, + Err(err) if role == Role::Canvas => return Err(err), + // The program goes on without it. + Err(err) => { + refused.push((role, format!("{err:#}"))); + continue; + } + }; + kind = config.kind.clone(); + rungs.push(Probed { + role, + exporter, + config, + video, + }); + } + Ok(Self { rungs, refused }) } /// [`new`](Self::new) on a thread of its own, off the program clock, once @@ -1247,29 +1572,106 @@ impl Probe { } } -/// One rendition of the program video: the exporter the program clock copies -/// into, and the encode thread behind it. -struct Rendition { - exporter: Exporter, - frames: mpsc::SyncSender