diff --git a/README.md b/README.md index 037ab46..a62a9d4 100644 --- a/README.md +++ b/README.md @@ -94,8 +94,8 @@ Every panel is a tab you can drag, split, or stack. | Panel | Use it to | | --- | --- | -| Preview, Program | See the scene you are editing and the one that is live, and drag items in the preview to place them. With more than one canvas, pick the canvas they show, and that the Items panel edits, beside the preview. **Cut** puts the preview on program at once; **Transition** takes it there by the transition picked beside it, over the time set there (see [Transitions](#transitions)). **Interact** passes the mouse over the preview to the browser source under it instead, and the keyboard to the one last clicked. | -| Scenes | Pick the scene in preview, and add, rename (double-click), reorder, or remove one. A scene on program, or still leaving it in a transition, cannot be removed, even by an undo, until program cuts away from it. | +| Preview, Program | See the scene you are editing and the one that is live, and drag items in the preview to place them. With more than one canvas, pick the canvas they show, and that the Scenes and Items panels switch and edit, beside the preview; **Linked** beside Cut unlinks a further canvas (see [Canvases](#canvases)). **Cut** puts the preview on program at once; **Transition** takes it there by the transition picked beside it, over the time set there (see [Transitions](#transitions)). **Interact** passes the mouse over the preview to the browser source under it instead, and the keyboard to the one last clicked. | +| Scenes | Pick the scene in preview, and add, rename (double-click), reorder, or remove one. A scene on program on any canvas, or still leaving it in a transition, cannot be removed, even by an undo, until program cuts away from it. | | History | See every edit, and undo or redo back to any of them. | | Items | Place, show, hide, and order what the preview scene or the top layer draws on the canvas shown, back to front. Select an item to filter or copy it. | | Sources | Add a source by hand, see whether each and its own track are live and where it is shown, clone one, and remove one. Select a source to filter it. A camera or broadcast without a new picture, or a microphone or system audio without new sound, for a second reads as stalled. A browser source shows whether its page is copied on the GPU or, as a fallback, uploaded from memory (CPU), in place of a track, and **Reload** restarts its page. A media file reads as ended once it plays to its end without looping, **Play** plays it from its start, and **Reload** reads the file again. | @@ -104,8 +104,8 @@ Every panel is a tab you can drag, split, or stack. | Filters | Crop the selected source or item, add, order, and set its filters, set an item's blend mode, and copy or paste filters. | | Keyframes | Animate the selected item: add, time, ease, and remove its keyframes, and set how far each moves, resizes, crops, and filters it (see [Keyframes](#keyframes)). **Play** plays the preview scene's keyframes again. | | 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, pick the mix it records, set the canvas size and frame rate, add, resize, and remove further canvases, and see the broadcast path, whether the program is encoding, the encoder sessions open, and each canvas's renditions' encoder work. | -| Health | See how the stream reaches viewers, relay by relay and track by track, and the cause of each problem, and which recordings a crash left unfinished were repaired at launch. | +| Output | Start and stop recording, pick the canvas and the mix it records, set the canvas size and frame rate, add, resize, and remove further canvases, and see the broadcast path, whether the program is encoding, the encoder sessions open, and each canvas's renditions' and the recording's encoder work. | +| Health | See how the stream reaches viewers, relay by relay, canvas by canvas, and track by track, the cause of each problem, and which recordings a crash left unfinished were repaired at launch. | | 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. | A new source, from the Sources or Devices panel, is placed over whatever the @@ -120,11 +120,20 @@ a 1080x1920 portrait one for a vertical stream beside a 1920x1080 landscape main canvas. Add one by name in the Output panel. Each scene has its own layout on each canvas, and each canvas its own top layer; a scene without a layout on a canvas shows only that canvas's top layer there. Every canvas -shares the main canvas's frame rate, and the canvases are linked: one scene is -in preview and one on program on all of them, and a cut or transition takes -every canvas together. An item's keyframes play on every canvas from when its -scene goes on, as on the main one. A source captures or decodes once however many canvases -show it, and a canvas draws only the items it places. +shares the main canvas's frame rate. A source captures or decodes once however +many canvases show it, and a canvas draws only the items it places. Only the +canvas Preview shows draws a preview, and only while the Preview panel is +visible, so a canvas nobody looks at costs only its program. + +A further canvas starts linked: it has the main canvas's scene in preview and +on program, and a cut or transition takes it with the main canvas, by the main +canvas's transition. An item's keyframes play on it from when its scene goes +on, as on the main one. Untick **Linked** in the Program panel, while Preview +shows the canvas, to unlink it: it keeps showing what it showed, and from then +on the Scenes panel picks its preview, and Cut and Transition take it to +program, on that canvas alone, by a transition of its own, set beside them. The +canvas picker marks it unlinked. Linking it again cuts it to the main canvas's +program. Each further canvas publishes a broadcast of its own beneath the main one, at `<--broadcast>/`, with renditions and encoders of its own (see @@ -443,6 +452,8 @@ framerate = 30 width = 1080 height = 1920 top = [{ source = "logo", rect = { x = 780, y = 40, width = 280, height = 120 } }] +transition = { kind = "fade", duration = "500ms" } # unlinked, switching on its own +recorded = true # recordings hold this canvas rather than the main one [transition] kind = "slide" # fade, slide, wipe, or move @@ -548,7 +559,11 @@ ids. `[[top]]` items make up the top layer. `[canvases.]` adds a further canvas, with its `width` and `height` and its own `top` layer, and a scene's `canvases.` lists its items on that canvas; leave it out, rather than empty, for a scene that shows nothing there. -A canvas name is one segment of its broadcast path, so it holds no `/`. +A canvas name is one segment of its broadcast path, so it holds no `/`. A +`transition` unlinks the canvas, switching its scenes on its own by that +transition; without one it switches with the main canvas. `recorded = true` +has recordings hold the canvas rather than the main one, on one canvas at +most. `[mixes]` names every mix, in the order the Audio panel lists them, the one on `program`, and the one a `recording` holds, the mix on program without it; @@ -696,7 +711,7 @@ encode engine for them all. The Output panel counts the sessions open: | The program's canvas size, live | 1, or none without `--connect` | | Its half size | 1 while someone watches it | | Each further canvas's canvas size and half size | as the program's | -| The recording's own canvas size | 1 while recording | +| The recording's own size, of the canvas it holds | 1 while recording | | Each captured source or media file | 1 while on program or recorded | So a live scene showing a camera, a display, and a window holds 4 sessions, 5 @@ -715,7 +730,9 @@ where it can and software otherwise. **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 +live or not. It holds the main canvas, or the further canvas picked beside it, +even while recording: a canvas of another size goes on in a new file. 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 @@ -816,16 +833,20 @@ stuck, a source stalled (a camera or broadcast the program draws, or a microphone or system audio the mix on program plays: unmuted and not left out of it), a source the program shows or plays failed or, as a media file that does not loop, played to its end, or viewers are late, stalling, or cannot -decode a track. A relay lost while the other serves is +decode a track. A problem with a further canvas's encoder or viewers names the +canvas. A relay lost while the other serves is degraded, since viewers have nothing left to fall back to; with none serving it is unhealthy. Below that it shows: - **Relays**: each relay's state, and the protocol its session negotiated. - **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 never reach - the relay, so they are left out of the media. + connection estimates it can carry, the media every canvas's tracks write, + and the share lost and resent. Loss and headroom are graded only while + someone watches any canvas, since the relay pulls nothing otherwise. Source + tracks never reach the relay, so they are left out of the media. + +Then, for each canvas's broadcast, the main one first: + - **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 @@ -835,18 +856,20 @@ it is unhealthy. Below that it shows: - **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. + grade, and the latency from the program clock to the slowest viewer. + Latency compares the two hosts' wall clocks, so keep both synced. - **Viewers**: each viewer that reports, with its round trip, latency, and - playout delay. The studio reads 64 at most, a sample the grades rest on: it - counts the rest, and a reader one frees goes to the next viewer to arrive. + playout delay. The studio reads 64 at most per canvas, a sample the grades + rest on: it counts the rest, and a reader one frees goes to the next viewer + to arrive. + +Last, each source's own tracks, grouped by source, graded as a canvas's are +from what they write; a still image or color carries no media to grade. -The catalog asks viewers to report under `/viewers/`: a viewer -publishes `/viewers/.echo`, as `moq export --echo ` -does. The relay decides who may publish there, as for any broadcast. +Each canvas's catalog asks its viewers to report under `/viewers/`, +`` being that canvas's own path: a viewer publishes +`/viewers/.echo`, as `moq export --echo ` does. The +relay decides who may publish there, as for any broadcast. ## Development diff --git a/quest/m0/README.md b/quest/m0/README.md index 838bf5b..aaaa9b9 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -94,7 +94,6 @@ repository. - [Stinger transitions](/quest/m0/stinger.md) - a clip played over the program that switches scenes at its cut point - [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 -- [Multiple canvases](/quest/m0/multi-canvas.md) - landscape and portrait canvases composite at once from shared sources - [The join page is live](/quest/m0/join-page-live.md) - `join.studio.moq.dev` serves the join page `--join-page` defaults to - [Clap-to-measure sync delay](/quest/m0/sync-clap.md) - a clap sets a microphone's delay against its camera - [Audio filters](/quest/m0/audio-filters.md) - per-source noise suppression, gate, compressor, limiter, and plugins, configured from the studio diff --git a/quest/m0/destinations.md b/quest/m0/destinations.md index dfd3034..47e60ce 100644 --- a/quest/m0/destinations.md +++ b/quest/m0/destinations.md @@ -62,6 +62,9 @@ Decided 2026-10-22 with the maintainer: CBR) for common ingest constraints without naming platforms. The creator can override every field. +From the multi-canvas quest: each destination picks the canvas it carries and +its mix, as a recording picks both, and pushes that canvas's program. + ## Required - [Platform AAC encoders are pinned](/quest/m0/pin-aac-encode.md) - AAC on macOS and Windows diff --git a/quest/m0/multi-canvas.md b/quest/m0/multi-canvas.md deleted file mode 100644 index e7ab30d..0000000 --- a/quest/m0/multi-canvas.md +++ /dev/null @@ -1,27 +0,0 @@ -# [M] Multiple canvases - -## Goal - -The studio composites several canvases at once from shared sources, such as -a 1920x1080 landscape canvas and a 1080x1920 portrait one. Each canvas has -its own layout per scene, its own preview, and its own outputs. A source -captures or decodes once however many canvases use it. - -## Plan - -Further canvases, their layouts and top layers, linked scene switching, and a -broadcast per canvas at `<--broadcast>/` landed in -[moq-dev/studio#42](https://github.com/moq-dev/studio/pull/42). What remains: - -- Unlinking: a canvas the creator unlinks keeps its own preview and program - scene, so a scene change no longer cuts it with the others. Transitions are - per canvas, so an unlinked canvas takes its own. -- Outputs per canvas: a recording, and each destination, picks the canvas it - carries, as each picks its mix. Only the main canvas records today. -- Health per canvas: the Health panel grades each canvas's broadcast, with - viewer feedback beneath it, not only the main one's. -- Preview cost: a canvas no panel shows skips drawing its preview, so an - unseen canvas costs only its program. - -Per-canvas cost must grow only with the sources that canvas uses, never with -the whole source table; `compositor::bench` measures it. diff --git a/rs/moq-scene/src/collection.rs b/rs/moq-scene/src/collection.rs index 4f9beda..e663573 100644 --- a/rs/moq-scene/src/collection.rs +++ b/rs/moq-scene/src/collection.rs @@ -29,7 +29,8 @@ pub struct Collection { /// sources, under its own top layer. #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] pub canvases: BTreeMap, - /// How the preview is taken to program, short of a cut. + /// How the preview is taken to program, short of a cut, on the main canvas + /// and every canvas linked to it. #[serde(default)] pub transition: Transition, /// Every source, by its unique name. Scenes refer to sources by that name, @@ -99,6 +100,22 @@ impl Collection { self.scenes.iter().find(|scene| scene.name == name) } + /// The further canvas recordings hold, or `None` for the main one. + pub fn recorded(&self) -> Option<&str> { + let mut canvases = self.canvases.iter(); + canvases + .find(|(_, extra)| extra.recorded) + .map(|(name, _)| name.as_str()) + } + + /// How `canvas` takes its preview to program: an unlinked canvas's own + /// transition, or the main canvas's for it, a linked one, or one that does + /// not exist. + pub fn takes(&self, canvas: Option<&str>) -> Transition { + let extra = canvas.and_then(|name| self.canvases.get(name)); + extra.and_then(|extra| extra.transition).unwrap_or(self.transition) + } + /// The canvas `canvas` names: the main one for `None`, or a further one, /// at the main canvas's frame rate. `None` for a canvas that does not /// exist. @@ -220,6 +237,12 @@ impl Collection { .validate() .map_err(|err| Error::Invalid(format!("canvas `{name}`: {err}")))?; } + let mut recorded = self.canvases.iter().filter(|(_, extra)| extra.recorded); + if let (Some((first, _)), Some((second, _))) = (recorded.next(), recorded.next()) { + return Err(Error::Invalid(format!( + "canvases `{first}` and `{second}` are both recorded; recordings hold one canvas" + ))); + } let mut names = BTreeSet::new(); for scene in &self.scenes { @@ -463,15 +486,28 @@ pub struct Extra { /// Its top layer, as [`Collection::top`] is the main canvas's. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub top: Vec, + /// How it takes its own preview to program, short of a cut, which unlinks + /// it: it keeps a preview and program scene of its own, so a scene change + /// on the main canvas leaves it be. Without one it is linked, switching + /// scenes with the main canvas by the main canvas's transition. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub transition: Option, + /// Whether recordings hold this canvas rather than the main one. At most + /// one canvas is recorded. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub recorded: bool, } impl Extra { - /// An empty canvas of `width` by `height`. + /// An empty canvas of `width` by `height`, linked to the main canvas and + /// not recorded. pub fn new(width: u32, height: u32) -> Self { Self { width, height, top: Vec::new(), + transition: None, + recorded: false, } } @@ -2383,6 +2419,38 @@ rect = { x = 0, y = 680, width = 1280, height = 40 } ); } + /// An unlinked canvas keeps its own transition and a recorded one says + /// so in the file, and recordings hold one canvas at most. + #[test] + fn a_further_canvas_unlinks_and_records() { + let mut collection = example(); + let slide = Transition::Slide { + duration: Duration::from_millis(300), + direction: Direction::Up, + }; + collection.canvases.insert( + "tall".into(), + Extra { + transition: Some(slide), + recorded: true, + ..Extra::new(1080, 1920) + }, + ); + collection.canvases.insert("wide".into(), Extra::new(2560, 1080)); + let text = collection.to_toml().unwrap(); + assert_eq!(Collection::from_toml(&text).unwrap(), collection); + assert_eq!(collection.recorded(), Some("tall")); + assert_eq!(collection.takes(Some("tall")), slide); + assert_eq!(collection.takes(Some("wide")), collection.transition); + assert_eq!(collection.takes(None), collection.transition); + + collection.canvases.get_mut("wide").unwrap().recorded = true; + assert!( + matches!(collection.validate(), Err(Error::Invalid(_))), + "two recorded canvases" + ); + } + #[test] fn refuses_a_canvas_the_encoder_cannot_take() { for (width, height, framerate) in [(1921, 1080, 30), (0, 1080, 30), (1920, 1080, 0)] { diff --git a/rs/moq-scene/src/compositor.rs b/rs/moq-scene/src/compositor.rs index e45e017..b88819d 100644 --- a/rs/moq-scene/src/compositor.rs +++ b/rs/moq-scene/src/compositor.rs @@ -29,19 +29,20 @@ pub const FORMAT: wgpu::TextureFormat = wgpu::TextureFormat::Rgba8Unorm; /// memory. [`render`](Self::render) redraws them from the newest picture each /// source has, so a source that stalls repeats its last picture rather than /// holding up the canvas. Each source's picture is uploaded once however many -/// canvases show it, and a canvas draws only the items it places. +/// canvases show it, and a canvas draws only the items it places. Only the +/// canvases [`set_previewed`](Self::set_previewed) names draw a preview, so +/// one nobody looks at costs only its program. /// /// Pixels are blended in the gamma-encoded space sources arrive in, as a /// broadcast switcher and CSS do, so a 50% fade is a 50% mix of the stored /// values. /// -/// The canvases are linked: one scene is in preview and one on program on -/// every canvas, and a transition and each item's keyframes run on each at -/// once, on the clock [`render`](Self::render) is given, so every output drawn -/// from a program, such as the encoder and a recording, sees the same frame of -/// them. An item's keyframes start when its scene goes on program, and -/// separately when it goes into preview, for the preview, on every canvas -/// alike. +/// Each canvas switches scenes by a [`Switcher`]: the main canvas's, which a +/// further canvas shares while linked, or one of its own once the collection +/// unlinks it. Every switcher runs on the clock [`render`](Self::render) is +/// given, so every output drawn from a program, such as the encoder and a +/// recording, sees the same frame of its transition and keyframes, and linked +/// canvases switch, transition, and play keyframes as one. pub struct Compositor { device: wgpu::Device, queue: wgpu::Queue, @@ -53,17 +54,8 @@ pub struct Compositor { layers: HashMap, /// Lays out and paints overlays, started with the first one. overlays: Option, - preview_scene: Option, - program_scene: Option, - /// When the preview scene went into preview, or `None` until the next - /// render, which starts it. - preview_start: Option, - /// When the program scene went on program, as `preview_start` is. - program_start: Option, - /// How long the preview scene had been in preview as of the last render: - /// the moment of its keyframes every canvas's preview texture shows. - preview_shown: Duration, - switch: Option, + /// The main canvas's, which every linked canvas shares. + switcher: Switcher, } /// One canvas's preview and program, as of the last @@ -77,6 +69,12 @@ pub struct Output { incoming: Target, /// The scene going out, drawn whole to slide it off the canvas. outgoing: Target, + /// An unlinked canvas's own switcher; `None` while linked. + own: Option, + /// Whether the next render draws the preview. + previewing: bool, + /// Whether the last render drew it. + previewed: bool, } impl Output { @@ -88,6 +86,9 @@ impl Output { program: target("moq-scene program"), incoming: target("moq-scene incoming"), outgoing: target("moq-scene outgoing"), + own: None, + previewing: false, + previewed: false, } } @@ -96,11 +97,19 @@ impl Output { self.canvas } - /// The scene in preview. + /// The scene in preview, current only while + /// [`previewed`](Self::previewed). pub fn preview(&self) -> &wgpu::Texture { &self.preview.texture } + /// Whether the last render drew the preview, as + /// [`Compositor::set_previewed`] asks; otherwise it holds an older picture, + /// or none. + pub fn previewed(&self) -> bool { + self.previewed + } + /// The program. Created with `COPY_SRC | TEXTURE_BINDING`, what /// `moq_video::render::export::Exporter::copy` takes to hand it to an /// encoder. @@ -109,7 +118,120 @@ impl Output { } } +/// Which scene a canvas has in preview and which on program, and the +/// transition from one to the other: the main canvas's, which each linked +/// canvas shares, or an unlinked canvas's own. +/// +/// An item's keyframes start when its scene goes on program, and separately +/// when it goes into preview, for the preview. +#[derive(Clone, Default)] +pub struct Switcher { + preview_scene: Option, + program_scene: Option, + /// When the preview scene went into preview, or `None` until the next + /// render, which starts it. + preview_start: Option, + /// When the program scene went on program, as `preview_start` is. + program_start: Option, + /// How long the preview scene had been in preview as of the last render: + /// the moment of its keyframes the preview shows. + preview_shown: Duration, + switch: Option, +} + +impl Switcher { + /// The scene in preview. + pub fn preview_scene(&self) -> Option<&str> { + self.preview_scene.as_deref() + } + + /// The scene on program, or the one a transition brings in. + pub fn program_scene(&self) -> Option<&str> { + self.program_scene.as_deref() + } + + /// Whether a transition is under way. + pub fn transitioning(&self) -> bool { + self.switch.is_some() + } + + /// The scenes program draws: the one on program, and the one a transition + /// under way leaves. + pub fn on_air(&self) -> impl Iterator { + let leaving = self.switch.as_ref().and_then(|switch| switch.from.as_deref()); + self.program_scene.as_deref().into_iter().chain(leaving) + } + + /// Put the scene called `scene` in preview, or nothing, playing its + /// keyframes from the start, even if it was already there. + pub fn set_preview(&mut self, scene: Option) { + self.preview_scene = scene; + self.preview_start = None; + } + + /// How long the preview scene had been in preview as of the last + /// [`render`](Compositor::render): the moment of its items' keyframes the + /// preview shows, for [`Item::posed`]. + pub fn preview_since(&self) -> Duration { + self.preview_shown + } + + /// Put the preview on program at once, ending any transition. + pub fn cut(&mut self) { + self.switch = None; + self.program_scene = self.preview_scene.clone(); + self.program_start = None; + } + + /// Take the preview to program by `transition`, starting at `now`. + /// + /// A transition already under way lands on its scene first, so this one + /// starts from what that one was heading to. + pub fn transition(&mut self, transition: Transition, now: Instant) { + let from = std::mem::replace(&mut self.program_scene, self.preview_scene.clone()); + let from_start = self.program_start.replace(now).unwrap_or(now); + self.switch = (!transition.duration().is_zero()).then_some(Switch { + from, + from_start, + start: now, + transition, + }); + } + + /// The scenes followed by name: in preview, on program, and leaving. + fn scenes(&mut self) -> impl Iterator> { + let leaving = self.switch.as_mut().map(|switch| &mut switch.from); + [&mut self.preview_scene, &mut self.program_scene] + .into_iter() + .chain(leaving) + } + + /// Start the scenes that went on since the last render at `now`, and land + /// a transition that has run its course. + fn advance(&mut self, now: Instant) { + self.preview_shown = now.saturating_duration_since(*self.preview_start.get_or_insert(now)); + self.program_start.get_or_insert(now); + if self.switch.as_ref().is_some_and(|switch| switch.linear(now) >= 1.0) { + self.switch = None; + } + } + + /// What to draw at `now`, once [`advance`](Self::advance)d to it. + fn shot(&self, now: Instant) -> Shot<'_> { + let program = self.program_start.unwrap_or(now); + Shot { + preview: On(self.preview_scene.as_deref(), self.preview_shown), + program: On(self.program_scene.as_deref(), now.saturating_duration_since(program)), + switch: self.switch.as_ref().map(|switch| { + let from = On(switch.from.as_deref(), now.saturating_duration_since(switch.from_start)); + (switch, from, switch.linear(now)) + }), + } + } +} + /// A transition under way from `from` to the program scene. +#[derive(Clone)] struct Switch { from: Option, /// When `from` went on program. @@ -118,6 +240,14 @@ struct Switch { transition: Transition, } +impl Switch { + /// How far along it is at `now`, from 0 to 1, before easing. + fn linear(&self, now: Instant) -> f32 { + let elapsed = now.saturating_duration_since(self.start); + (elapsed.as_secs_f32() / self.transition.duration().as_secs_f32()).min(1.0) + } +} + /// A canvas-sized texture drawn into, and bound to sample. struct Target { texture: wgpu::Texture, @@ -155,8 +285,9 @@ impl effect::Library for Library<'_> { } impl Compositor { - /// A compositor drawing a main canvas of `canvas` on `device`, and no - /// further canvas until [`set_canvases`](Self::set_canvases) adds one. + /// A compositor drawing a main canvas of `canvas` on `device`, its preview + /// included, and no further canvas until + /// [`set_canvases`](Self::set_canvases) adds one. /// /// Fails if the canvas is invalid or the device cannot hold a texture that /// large. @@ -164,20 +295,17 @@ impl Compositor { fits(device, canvas)?; let effects = Effects::new(device, queue, canvas.width, canvas.height); + let mut main = Output::new(device, &effects, canvas); + main.previewing = true; Ok(Self { - main: Output::new(device, &effects, canvas), + main, extras: BTreeMap::new(), device: device.clone(), queue: queue.clone(), effects, layers: HashMap::new(), overlays: None, - preview_scene: None, - program_scene: None, - preview_start: None, - program_start: None, - preview_shown: Duration::ZERO, - switch: None, + switcher: Switcher::default(), }) } @@ -192,46 +320,80 @@ impl Compositor { /// and one `collection` no longer has is dropped. A transition under way /// carries on. /// + /// A canvas `collection` unlinks takes a switcher of its own, starting from + /// where the main canvas's is, so it goes on showing what it showed, and + /// one linked again takes the main canvas's scenes at once. + /// /// Fails, changing nothing, if a canvas is invalid or the device cannot /// hold a texture that large. pub fn set_canvases(&mut self, collection: &Collection) -> Result<(), Error> { let canvas = collection.canvas; - let extras: BTreeMap<&str, Canvas> = collection + let extras: BTreeMap<&str, (Canvas, bool)> = collection .canvases .iter() - .map(|(name, extra)| (name.as_str(), extra.canvas(canvas.framerate))) + .map(|(name, extra)| { + let linked = extra.transition.is_none(); + (name.as_str(), (extra.canvas(canvas.framerate), linked)) + }) .collect(); - for canvas in std::iter::once(&canvas).chain(extras.values()) { + for canvas in std::iter::once(&canvas).chain(extras.values().map(|(canvas, _)| canvas)) { fits(&self.device, *canvas)?; } - let width = extras.values().map(|extra| extra.width).fold(canvas.width, u32::max); - let height = extras.values().map(|extra| extra.height).fold(canvas.height, u32::max); + let width = extras + .values() + .map(|(extra, _)| extra.width) + .fold(canvas.width, u32::max); + let height = extras + .values() + .map(|(extra, _)| extra.height) + .fold(canvas.height, u32::max); self.effects.fit(width, height); let (device, effects) = (&self.device, &self.effects); let resize = |output: &mut Output, canvas: Canvas| { if output.canvas.size() == canvas.size() { output.canvas = canvas; } else { - *output = Output::new(device, effects, canvas); + let resized = Output::new(device, effects, canvas); + let Output { own, previewing, .. } = std::mem::replace(output, resized); + output.own = own; + output.previewing = previewing; } }; resize(&mut self.main, canvas); self.extras.retain(|name, _| extras.contains_key(name.as_str())); - for (name, canvas) in extras { - match self.extras.get_mut(name) { - Some(output) => resize(output, canvas), - None => { - self.extras - .insert(name.to_string(), Output::new(device, effects, canvas)); + for (name, (canvas, linked)) in extras { + let output = match self.extras.get_mut(name) { + Some(output) => { + resize(output, canvas); + output } + None => self + .extras + .entry(name.to_string()) + .or_insert_with(|| Output::new(device, effects, canvas)), + }; + if linked { + output.own = None; + } else if output.own.is_none() { + output.own = Some(self.switcher.clone()); } } Ok(()) } - /// The scene in preview on the main canvas, as of the last - /// [`render`](Self::render). + /// Draw the preview only of `canvases`, each a further one by name or the + /// main one as `None`, from the next [`render`](Self::render) on; every + /// other canvas draws its program alone. A new compositor previews its + /// main canvas. + pub fn set_previewed(&mut self, canvases: &[Option<&str>]) { + for (name, output) in self.outputs_mut() { + output.previewing = canvases.contains(&name); + } + } + + /// The main canvas's preview, as of the last [`render`](Self::render). See + /// [`Output::preview`]. pub fn preview(&self) -> &wgpu::Texture { self.main.preview() } @@ -257,56 +419,55 @@ impl Compositor { std::iter::once((None, &self.main)).chain(extras) } - /// The scene in preview. - pub fn preview_scene(&self) -> Option<&str> { - self.preview_scene.as_deref() + fn outputs_mut(&mut self) -> impl Iterator, &mut Output)> { + let extras = self + .extras + .iter_mut() + .map(|(name, output)| (Some(name.as_str()), output)); + std::iter::once((None, &mut self.main)).chain(extras) } - /// The scene on program, or the one a transition brings in. - pub fn program_scene(&self) -> Option<&str> { - self.program_scene.as_deref() + /// What switches the scenes of `canvas`, a further one by name or the main + /// one for `None`: its own once unlinked, otherwise the main canvas's, + /// which a canvas it does not draw gets too. + pub fn switcher(&self, canvas: Option<&str>) -> &Switcher { + let own = canvas.and_then(|name| self.extras.get(name)?.own.as_ref()); + own.unwrap_or(&self.switcher) } - /// Why `source`'s newest picture could not be drawn, until one is. - pub fn error(&self, source: &str) -> Option<&str> { - self.layers.get(source)?.error.as_deref() + /// [`switcher`](Self::switcher), to switch with. + pub fn switcher_mut(&mut self, canvas: Option<&str>) -> &mut Switcher { + let own = canvas.and_then(|name| self.extras.get_mut(name)?.own.as_mut()); + own.unwrap_or(&mut self.switcher) } - /// Whether a transition is under way. - pub fn transitioning(&self) -> bool { - self.switch.is_some() + /// Every switcher: the main canvas's, then each unlinked canvas's. + fn switchers_mut(&mut self) -> impl Iterator { + let own = self.extras.values_mut().filter_map(|output| output.own.as_mut()); + std::iter::once(&mut self.switcher).chain(own) } - /// The scenes program draws: the one on program, and the one a transition - /// under way leaves. + /// The scenes any canvas's program draws, as [`Switcher::on_air`] has + /// them, a scene once for each switcher drawing it. pub fn on_air(&self) -> impl Iterator { - let leaving = self.switch.as_ref().and_then(|switch| switch.from.as_deref()); - self.program_scene.as_deref().into_iter().chain(leaving) - } - - /// Put the scene called `scene` in preview, or nothing, playing its - /// keyframes from the start, even if it was already there. - pub fn set_preview(&mut self, scene: Option) { - self.preview_scene = scene; - self.preview_start = None; + let own = self.extras.values().filter_map(|output| output.own.as_ref()); + std::iter::once(&self.switcher).chain(own).flat_map(Switcher::on_air) } - /// How long the preview scene had been in preview as of the last - /// [`render`](Self::render): the moment of its items' keyframes every - /// canvas's preview shows, since the canvases are linked, for - /// [`Item::posed`]. - pub fn preview_since(&self) -> Duration { - self.preview_shown + /// Why `source`'s newest picture could not be drawn, until one is. + pub fn error(&self, source: &str) -> Option<&str> { + self.layers.get(source)?.error.as_deref() } - /// Follow each renamed scene to its new name, wherever it is: in preview, - /// on program, or leaving in a transition, and drop each removed one, so a - /// scene that later takes its name does not go on program without a cut. - /// `renames` maps each old name to its new one, or `None`, as - /// [`Collection::renames`] reads them, and each scene is looked up once, so - /// a name one scene gave up and another took is not followed twice. + /// Follow each renamed scene to its new name, wherever it is on any + /// canvas: in preview, on program, or leaving in a transition, and drop + /// each removed one, so a scene that later takes its name does not go on + /// program without a cut. `renames` maps each old name to its new one, or + /// `None`, as [`Collection::renames`] reads them, and each scene is looked + /// up once, so a name one scene gave up and another took is not followed + /// twice. pub fn rename(&mut self, renames: &BTreeMap>) { - for scene in self.scenes() { + for scene in self.switchers_mut().flat_map(Switcher::scenes) { if let Some(to) = scene.as_ref().and_then(|name| renames.get(name)) { *scene = to.clone(); } @@ -314,48 +475,20 @@ impl Compositor { } /// Drop each scene `collection` does not have from preview, program, and a - /// transition under way, such as one a reloaded file left out, so a scene - /// that later takes its name does not go on program without a cut. + /// transition under way on any canvas, such as one a reloaded file left + /// out, so a scene that later takes its name does not go on program + /// without a cut. pub fn retain(&mut self, collection: &Collection) { - for scene in self.scenes() { + for scene in self.switchers_mut().flat_map(Switcher::scenes) { if scene.as_deref().is_some_and(|name| collection.scene(name).is_none()) { *scene = None; } } } - /// The scenes followed by name: in preview, on program, and leaving. - fn scenes(&mut self) -> impl Iterator> { - let leaving = self.switch.as_mut().map(|switch| &mut switch.from); - [&mut self.preview_scene, &mut self.program_scene] - .into_iter() - .chain(leaving) - } - - /// Put the preview on program at once, ending any transition. - pub fn cut(&mut self) { - self.switch = None; - self.program_scene = self.preview_scene.clone(); - self.program_start = None; - } - - /// Take the preview to program by `transition`, starting at `now`. - /// - /// A transition already under way lands on its scene first, so this one - /// starts from what that one was heading to. - pub fn transition(&mut self, transition: Transition, now: Instant) { - let from = std::mem::replace(&mut self.program_scene, self.preview_scene.clone()); - let from_start = self.program_start.replace(now).unwrap_or(now); - self.switch = (!transition.duration().is_zero()).then_some(Switch { - from, - from_start, - start: now, - transition, - }); - } - - /// Draw preview and program on every canvas as of `now`, from the newest - /// picture each of `sources` has. + /// Draw every canvas as of `now`, from the newest picture each of + /// `sources` has: its program, and its preview if + /// [`set_previewed`](Self::set_previewed) asked for it. /// /// A source whose picture fails to draw keeps its last one and reports why /// through [`error`](Self::error); this fails only when a canvas itself @@ -363,18 +496,8 @@ impl Compositor { pub fn render(&mut self, collection: &Collection, sources: &Sources, now: Instant) -> Result<(), Error> { self.update(collection, sources, now); self.paint(collection, now); - - let since = |start: &mut Option| now.saturating_duration_since(*start.get_or_insert(now)); - let preview_since = since(&mut self.preview_start); - self.preview_shown = preview_since; - let program_since = since(&mut self.program_start); - - let linear = self.switch.as_ref().map(|switch| { - let elapsed = now.saturating_duration_since(switch.start); - (elapsed.as_secs_f32() / switch.transition.duration().as_secs_f32()).min(1.0) - }); - if linear.is_some_and(|linear| linear >= 1.0) { - self.switch = None; + for switcher in self.switchers_mut() { + switcher.advance(now); } let mut frame = self.effects.frame(); @@ -387,20 +510,19 @@ impl Compositor { size: self.main.canvas, prepared: HashMap::new(), }; - let preview = On(self.preview_scene.as_deref(), preview_since); - let program = On(self.program_scene.as_deref(), program_since); - let switch = self.switch.as_ref().zip(linear).map(|(switch, linear)| { - let from = On(switch.from.as_deref(), now.saturating_duration_since(switch.from_start)); - (switch, from, linear) - }); + let main = self.switcher.shot(now); let extras = self.extras.iter().map(|(name, output)| (Some(name.as_str()), output)); for (canvas, output) in std::iter::once((None, &self.main)).chain(extras) { draw.canvas = canvas; draw.size = output.canvas; - draw.output(output, preview, program, switch); + let shot = output.own.as_ref().map_or(main, |own| own.shot(now)); + draw.output(output, shot); } self.effects.submit(frame); + for (_, output) in self.outputs_mut() { + output.previewed = output.previewing; + } Ok(()) } @@ -527,23 +649,19 @@ impl Compositor { } } - /// The sources whose pictures preview and program draw on any canvas: - /// each visible item's, and each mask's, through clones. + /// The sources whose pictures the next render draws on any canvas: each + /// visible item's, and each mask's, through clones, of the scenes on its + /// program and, while previewed, in its preview, and of its top layer. fn shown<'a>(&self, collection: &'a Collection) -> HashSet<&'a str> { - let leaving = self.switch.as_ref().and_then(|switch| switch.from.as_deref()); - let scenes: Vec> = [self.preview_scene.as_deref(), self.program_scene.as_deref(), leaving] - .into_iter() - .flatten() - .map(|name| Some(name.to_string())) - .chain([None]) - .collect(); - let canvases = std::iter::once(None).chain(self.extras.keys().map(Some)); let mut shown = HashSet::new(); - for canvas in canvases { - for scene in &scenes { + for (canvas, output) in self.outputs() { + let switcher = output.own.as_ref().unwrap_or(&self.switcher); + let preview = switcher.preview_scene().filter(|_| output.previewing); + let scenes = preview.into_iter().chain(switcher.on_air()).map(Some).chain([None]); + for scene in scenes { let place = Place { - canvas: canvas.cloned(), - scene: scene.clone(), + canvas: canvas.map(str::to_string), + scene: scene.map(str::to_string), }; let items = collection.items(&place).unwrap_or_default(); for item in items.iter().filter(|item| item.visible) { @@ -681,6 +799,16 @@ fn fits(device: &wgpu::Device, canvas: Canvas) -> Result<(), Error> { #[derive(Clone, Copy)] struct On<'a>(Option<&'a str>, Duration); +/// What one canvas draws at a moment, as its [`Switcher`] has it. +#[derive(Clone, Copy)] +struct Shot<'a> { + preview: On<'a>, + program: On<'a>, + /// A transition under way, the scene it leaves, and how far along it is + /// before easing. + switch: Option<(&'a Switch, On<'a>, f32)>, +} + /// One render's draws onto its targets. struct Draw<'a> { collection: &'a Collection, @@ -698,19 +826,20 @@ struct Draw<'a> { } impl Draw<'_> { - /// Draw `output`'s preview, showing `preview`, and its program, showing - /// `program`, or `switch` to it from the scene it leaves, `linear` of its - /// time along. - fn output(&mut self, output: &Output, preview: On, program: On, switch: Option<(&Switch, On, f32)>) { + /// Draw `output`'s program as `shot` has it, and its preview while + /// previewing. + fn output(&mut self, output: &Output, shot: Shot) { let top = self.place(None); // The top layer goes over both outputs, and over a transition rather // than through it, so it stays on across the switch. - self.target(&output.preview); - self.scene(&output.preview, preview); - self.layer(&output.preview, top.clone(), Duration::ZERO); + if output.previewing { + self.target(&output.preview); + self.scene(&output.preview, shot.preview); + self.layer(&output.preview, top.clone(), Duration::ZERO); + } - let to = program; - match switch { + let to = shot.program; + match shot.switch { Some((switch, from, linear)) => { let progress = switch.transition.progress(linear); let canvas = Rect::fill(&self.size); @@ -1181,8 +1310,8 @@ mod tests { async fn items_draw_in_place_and_in_order() { local(async { let mut studio = Studio::new(collection()).await; - studio.compositor.set_preview(Some("red".into())); - studio.compositor.cut(); + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); + studio.compositor.switcher_mut(None).cut(); studio.render(Instant::now()); let program = studio.program(); assert_eq!(pixel(&program, 1, 1), RED); @@ -1211,8 +1340,8 @@ mod tests { ); collection.scenes[0].items.truncate(1); let mut studio = Studio::new(collection).await; - studio.compositor.set_preview(Some("red".into())); - studio.compositor.cut(); + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); + studio.compositor.switcher_mut(None).cut(); studio.render(Instant::now()); let actual = pixel(&studio.program(), 1, 1); assert!(near(actual, [128, 0, 0, 255]), "{actual:?}"); @@ -1228,14 +1357,14 @@ mod tests { local(async { let mut studio = Studio::new(collection()).await; let start = Instant::now(); - studio.compositor.set_preview(Some("red".into())); - studio.compositor.cut(); - studio.compositor.set_preview(Some("blue".into())); - studio.compositor.transition(FADE, start); - assert_eq!(studio.compositor.program_scene(), Some("blue")); + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); + studio.compositor.switcher_mut(None).cut(); + studio.compositor.switcher_mut(None).set_preview(Some("blue".into())); + studio.compositor.switcher_mut(None).transition(FADE, start); + assert_eq!(studio.compositor.switcher(None).program_scene(), Some("blue")); studio.render(start + Duration::from_millis(500)); - assert!(studio.compositor.transitioning()); + assert!(studio.compositor.switcher(None).transitioning()); let program = studio.program(); let actual = pixel(&program, 1, 1); assert!(near(actual, [128, 0, 127, 255]), "{actual:?}"); @@ -1243,7 +1372,7 @@ mod tests { assert_eq!(pixel(&studio.preview(), 1, 1), BLUE); studio.render(start + Duration::from_secs(1)); - assert!(!studio.compositor.transitioning()); + assert!(!studio.compositor.switcher(None).transitioning()); assert_eq!(pixel(&studio.program(), 1, 1), BLUE); }) .await; @@ -1281,8 +1410,8 @@ mod tests { collection.validate().unwrap(); let mut studio = Studio::new(collection).await; - studio.compositor.set_preview(Some("red".into())); - studio.compositor.cut(); + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); + studio.compositor.switcher_mut(None).cut(); studio.render(Instant::now()); let program = studio.program(); let actual = pixel(&program, 1, 6); @@ -1302,13 +1431,13 @@ mod tests { collection.top.push(Item::new("blue", CORNER)); let mut studio = Studio::new(collection).await; let start = Instant::now(); - studio.compositor.set_preview(Some("red".into())); - studio.compositor.cut(); + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); + studio.compositor.switcher_mut(None).cut(); studio.render(start); assert_eq!(pixel(&studio.program(), 1, 6), BLUE); - studio.compositor.set_preview(Some("blue".into())); - studio.compositor.transition(FADE, start); + studio.compositor.switcher_mut(None).set_preview(Some("blue".into())); + studio.compositor.switcher_mut(None).transition(FADE, start); studio.render(start + Duration::from_millis(500)); let program = studio.program(); let actual = pixel(&program, 1, 1); @@ -1355,15 +1484,15 @@ mod tests { }; let start = Instant::now(); - studio.compositor.set_preview(Some("red".into())); - studio.compositor.cut(); + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); + studio.compositor.switcher_mut(None).cut(); studio.render(start); assert_eq!(pixel(&studio.program(), 1, 1), RED); assert_eq!(tall(&mut studio), (BLUE, RED)); assert_eq!(studio.compositor.layers.len(), 2, "each source is uploaded once"); - studio.compositor.set_preview(Some("blue".into())); - studio.compositor.transition(FADE, start); + studio.compositor.switcher_mut(None).set_preview(Some("blue".into())); + studio.compositor.switcher_mut(None).transition(FADE, start); studio.render(start + Duration::from_millis(500)); let (fading, kept) = tall(&mut studio); assert!(near(fading, [0, 0, 128, 255]), "{fading:?}"); @@ -1382,6 +1511,121 @@ mod tests { .await; } + /// The 8x8 collection with a 4x8 canvas `tall` that each scene fills with + /// its own source, unlinked by `transition` unless `None`. + fn tall(transition: Option) -> Collection { + let mut collection = collection(); + collection.canvases.insert( + "tall".into(), + Extra { + transition, + ..Extra::new(4, 8) + }, + ); + let fill = Rect { + x: 0.0, + y: 0.0, + width: 4.0, + height: 8.0, + }; + for scene in &mut collection.scenes { + let source = scene.name.clone(); + scene.canvases.insert("tall".into(), vec![Item::new(source, fill)]); + } + collection.validate().unwrap(); + collection + } + + impl Studio { + /// A pixel of the 4x8 canvas `tall`'s program, or of its preview. + fn tall(&mut self, preview: bool) -> [u8; 4] { + let output = self.compositor.output(Some("tall")).unwrap(); + let texture = if preview { output.preview() } else { output.program() }.clone(); + self.read(&texture)[(4 + 1) * 4..][..4].try_into().unwrap() + } + } + + /// An unlinked canvas starts from what the main canvas shows, then cuts + /// and fades on its own while the main canvas switches without it, and + /// linked again it takes the main canvas's program at once. + #[tokio::test] + #[ignore = "needs a Vulkan device; run by just rs render-sw"] + async fn an_unlinked_canvas_switches_on_its_own() { + local(async { + let mut studio = Studio::new(tall(Some(FADE))).await; + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); + studio.compositor.switcher_mut(None).cut(); + studio.compositor.set_canvases(&studio.collection).unwrap(); + let start = Instant::now(); + studio.render(start); + assert_eq!(studio.tall(false), RED); + + studio.compositor.switcher_mut(None).set_preview(Some("blue".into())); + studio.compositor.switcher_mut(None).cut(); + studio.render(start + Duration::from_millis(100)); + assert_eq!(pixel(&studio.program(), 1, 1), BLUE); + assert_eq!(studio.tall(false), RED, "the main canvas's cut leaves it be"); + assert_eq!(studio.compositor.switcher(Some("tall")).program_scene(), Some("red")); + + let tall = studio.compositor.switcher_mut(Some("tall")); + tall.set_preview(Some("blue".into())); + tall.transition(FADE, start + Duration::from_millis(100)); + studio.render(start + Duration::from_millis(600)); + assert!(near(studio.tall(false), [128, 0, 128, 255]), "{:?}", studio.tall(false)); + assert!(!studio.compositor.switcher(None).transitioning()); + let on_air: HashSet<&str> = studio.compositor.on_air().collect(); + assert_eq!( + on_air, + HashSet::from(["red", "blue"]), + "the scene it fades from is on air" + ); + + studio + .compositor + .switcher_mut(Some("tall")) + .set_preview(Some("red".into())); + studio.compositor.switcher_mut(Some("tall")).cut(); + studio.render(start + Duration::from_millis(700)); + assert_eq!(studio.tall(false), RED); + studio.collection.canvases.get_mut("tall").unwrap().transition = None; + studio.compositor.set_canvases(&studio.collection).unwrap(); + studio.render(start + Duration::from_millis(800)); + assert_eq!( + studio.tall(false), + BLUE, + "linked again, it shows the main canvas's program" + ); + assert_eq!(studio.compositor.switcher(Some("tall")).preview_scene(), Some("blue")); + }) + .await; + } + + /// Only the canvases asked for draw a preview, and each says whether its + /// preview is current. + #[tokio::test] + #[ignore = "needs a Vulkan device; run by just rs render-sw"] + async fn only_a_previewed_canvas_draws_its_preview() { + local(async { + let mut studio = Studio::new(tall(None)).await; + studio.compositor.set_canvases(&studio.collection).unwrap(); + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); + let start = Instant::now(); + studio.render(start); + let previewed = |studio: &Studio, canvas| studio.compositor.output(canvas).unwrap().previewed(); + assert!(previewed(&studio, None) && !previewed(&studio, Some("tall"))); + assert_eq!(pixel(&studio.preview(), 1, 1), RED); + assert_eq!(studio.tall(true), [0; 4], "never drawn"); + + studio.compositor.set_previewed(&[Some("tall")]); + studio.compositor.switcher_mut(None).set_preview(Some("blue".into())); + studio.render(start + Duration::from_millis(100)); + assert!(!previewed(&studio, None) && previewed(&studio, Some("tall"))); + assert_eq!(studio.tall(true), BLUE); + assert_eq!(pixel(&studio.preview(), 1, 1), RED, "left as last drawn"); + }) + .await; + } + /// A cut lands at once, even mid-fade, and an empty program is black. #[tokio::test] #[ignore = "needs a Vulkan device; run by just rs render-sw"] @@ -1392,12 +1636,12 @@ mod tests { studio.render(start); assert_eq!(pixel(&studio.program(), 1, 1), BLACK); - studio.compositor.set_preview(Some("blue".into())); - studio.compositor.transition(FADE, start); + studio.compositor.switcher_mut(None).set_preview(Some("blue".into())); + studio.compositor.switcher_mut(None).transition(FADE, start); studio.render(start + Duration::from_millis(500)); - studio.compositor.set_preview(Some("red".into())); - studio.compositor.cut(); - assert!(!studio.compositor.transitioning()); + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); + studio.compositor.switcher_mut(None).cut(); + assert!(!studio.compositor.switcher(None).transitioning()); studio.render(start + Duration::from_millis(600)); assert_eq!(pixel(&studio.program(), 1, 1), RED); }) @@ -1454,10 +1698,10 @@ mod tests { ), ] { let start = Instant::now(); - studio.compositor.set_preview(Some("red".into())); - studio.compositor.cut(); - studio.compositor.set_preview(Some("blue".into())); - studio.compositor.transition(transition, start); + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); + studio.compositor.switcher_mut(None).cut(); + studio.compositor.switcher_mut(None).set_preview(Some("blue".into())); + studio.compositor.switcher_mut(None).transition(transition, start); studio.render(start + Duration::from_millis(500)); let program = studio.program(); let actual = [(1, 1), (1, 6), (6, 1), (6, 6)].map(|(x, y)| pixel(&program, x, y)); @@ -1502,11 +1746,14 @@ mod tests { collection.scenes[1].items = vec![Item::new("blue", CORNER), Item::new("green", tile)]; let mut studio = Studio::new(collection).await; let start = Instant::now(); - studio.compositor.set_preview(Some("red".into())); - studio.compositor.cut(); - studio.compositor.set_preview(Some("blue".into())); + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); + studio.compositor.switcher_mut(None).cut(); + studio.compositor.switcher_mut(None).set_preview(Some("blue".into())); let duration = Duration::from_secs(1); - studio.compositor.transition(Transition::Move { duration }, start); + studio + .compositor + .switcher_mut(None) + .transition(Transition::Move { duration }, start); // Eased halfway, blue is at (2, 2) and both fades are at half. studio.render(start + Duration::from_millis(500)); @@ -1521,7 +1768,7 @@ mod tests { assert!(near(actual, [64, 128, 0, 255]), "green half over half red: {actual:?}"); studio.render(start + duration); - assert!(!studio.compositor.transitioning()); + assert!(!studio.compositor.switcher(None).transitioning()); let program = studio.program(); assert_eq!(pixel(&program, 1, 6), BLUE); assert_eq!(pixel(&program, 6, 6), BLACK); @@ -1573,6 +1820,7 @@ mod tests { collection.validate().unwrap(); let mut studio = Studio::new(collection).await; studio.compositor.set_canvases(&studio.collection).unwrap(); + studio.compositor.set_previewed(&[None, Some("tall")]); let tall = |studio: &mut Studio, preview: bool| { let output = studio.compositor.output(Some("tall")).unwrap(); let texture = match preview { @@ -1585,16 +1833,16 @@ mod tests { }; let start = Instant::now(); let at = |millis| start + Duration::from_millis(millis); - studio.compositor.set_preview(Some("red".into())); - studio.compositor.cut(); - studio.compositor.set_preview(Some("blue".into())); + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); + studio.compositor.switcher_mut(None).cut(); + studio.compositor.switcher_mut(None).set_preview(Some("blue".into())); studio.render(at(0)); studio.render(at(500)); assert_eq!(tall(&mut studio, true), (BLUE, BLACK), "halfway in, in preview"); assert_eq!(pixel(&studio.preview(), 1, 6), BLUE, "the main canvas too"); - studio.compositor.cut(); + studio.compositor.switcher_mut(None).cut(); studio.render(at(500)); assert_eq!(tall(&mut studio, false), (BLACK, BLACK), "on program from the start"); assert_eq!(pixel(&studio.program(), 1, 6), BLACK, "the main canvas too"); @@ -1619,15 +1867,15 @@ mod tests { let mut studio = Studio::new(collection).await; let start = Instant::now(); let at = |millis| start + Duration::from_millis(millis); - studio.compositor.set_preview(Some("red".into())); - studio.compositor.cut(); - studio.compositor.set_preview(Some("blue".into())); + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); + studio.compositor.switcher_mut(None).cut(); + studio.compositor.switcher_mut(None).set_preview(Some("blue".into())); studio.render(at(0)); assert_eq!(pixel(&studio.preview(), 1, 6), BLACK, "off the canvas"); studio.render(at(500)); assert_eq!( - studio.compositor.preview_since(), + studio.compositor.switcher(None).preview_since(), Duration::from_millis(500), "the frame shown" ); @@ -1635,7 +1883,7 @@ mod tests { assert_eq!(pixel(&preview, 1, 6), BLUE, "halfway in"); assert_eq!(pixel(&preview, 6, 6), BLACK); - studio.compositor.cut(); + studio.compositor.switcher_mut(None).cut(); studio.render(at(500)); assert_eq!(pixel(&studio.program(), 1, 6), BLACK, "on program from the start"); assert_eq!(pixel(&studio.preview(), 1, 6), BLUE, "the preview carries on"); @@ -1649,7 +1897,7 @@ mod tests { assert_eq!(pixel(&studio.program(), 6, 6), BLUE, "holding the last keyframe"); // Picking the scene again plays it again. - studio.compositor.set_preview(Some("blue".into())); + studio.compositor.switcher_mut(None).set_preview(Some("blue".into())); studio.render(at(9000)); assert_eq!(pixel(&studio.preview(), 1, 6), BLACK); }) @@ -1673,11 +1921,11 @@ mod tests { ]; let mut studio = Studio::new(collection).await; let start = Instant::now(); - studio.compositor.set_preview(Some("blue".into())); - studio.compositor.cut(); + studio.compositor.switcher_mut(None).set_preview(Some("blue".into())); + studio.compositor.switcher_mut(None).cut(); studio.render(start); // Program has reached the jump; preview starts over before it. - studio.compositor.set_preview(Some("blue".into())); + studio.compositor.switcher_mut(None).set_preview(Some("blue".into())); studio.render(start + jump); assert_eq!(pixel(&studio.preview(), 1, 6), BLACK, "before the jump"); assert_eq!(pixel(&studio.program(), 1, 6), BLUE, "after the jump"); @@ -1697,12 +1945,12 @@ mod tests { let mut studio = Studio::new(collection).await; let start = Instant::now(); let at = |millis| start + Duration::from_millis(millis); - studio.compositor.set_preview(Some("blue".into())); - studio.compositor.cut(); - studio.compositor.set_preview(Some("red".into())); + studio.compositor.switcher_mut(None).set_preview(Some("blue".into())); + studio.compositor.switcher_mut(None).cut(); + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); studio.render(at(0)); - studio.compositor.transition(FADE, at(9000)); + studio.compositor.switcher_mut(None).transition(FADE, at(9000)); studio.render(at(9500)); let program = studio.program(); // Red has slid halfway in over the left half, blue is in place, and @@ -1732,11 +1980,14 @@ mod tests { collection.scenes[1].items = vec![invert(1.0), Item::new("red", CORNER)]; let mut studio = Studio::new(collection).await; let start = Instant::now(); - studio.compositor.set_preview(Some("red".into())); - studio.compositor.cut(); - studio.compositor.set_preview(Some("blue".into())); + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); + studio.compositor.switcher_mut(None).cut(); + studio.compositor.switcher_mut(None).set_preview(Some("blue".into())); let duration = Duration::from_secs(1); - studio.compositor.transition(Transition::Move { duration }, start); + studio + .compositor + .switcher_mut(None) + .transition(Transition::Move { duration }, start); studio.render(start); let program = studio.program(); @@ -1762,8 +2013,8 @@ mod tests { async fn a_stalled_source_repeats_its_last_picture() { local(async { let mut studio = Studio::new(collection()).await; - studio.compositor.set_preview(Some("blue".into())); - studio.compositor.cut(); + studio.compositor.switcher_mut(None).set_preview(Some("blue".into())); + studio.compositor.switcher_mut(None).cut(); studio.render(Instant::now()); assert!(studio.sources.get("blue").unwrap().take().is_none(), "drawn once"); studio.render(Instant::now()); @@ -1821,8 +2072,8 @@ mod tests { /// The program of `collection` with scene "red" cut to it, its files in `dir`. async fn drawn(collection: Collection, dir: tempfile::TempDir) -> Vec { let mut studio = Studio::with_files(collection, dir).await; - studio.compositor.set_preview(Some("red".into())); - studio.compositor.cut(); + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); + studio.compositor.switcher_mut(None).cut(); studio.render(Instant::now()); studio.program() } @@ -2279,8 +2530,8 @@ mod tests { }, ); let mut studio = Studio::new(collection).await; - studio.compositor.set_preview(Some("red".into())); - studio.compositor.cut(); + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); + studio.compositor.switcher_mut(None).cut(); studio.render(Instant::now()); let old = studio.compositor.program().clone(); @@ -2351,10 +2602,10 @@ mod tests { local(async { let mut studio = Studio::new(collection()).await; let start = Instant::now(); - studio.compositor.set_preview(Some("red".into())); - studio.compositor.cut(); - studio.compositor.set_preview(Some("blue".into())); - studio.compositor.transition(FADE, start); + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); + studio.compositor.switcher_mut(None).cut(); + studio.compositor.switcher_mut(None).set_preview(Some("blue".into())); + studio.compositor.switcher_mut(None).transition(FADE, start); // `blue` takes the name `red` gives up, which is followed once. let mut draft = studio.collection.draft("Rename"); @@ -2363,8 +2614,8 @@ mod tests { let edit = draft.finish().unwrap(); studio.compositor.rename(&studio.collection.renames(&edit)); studio.collection.apply(&edit).unwrap(); - assert_eq!(studio.compositor.preview_scene(), Some("red")); - assert_eq!(studio.compositor.program_scene(), Some("red")); + assert_eq!(studio.compositor.switcher(None).preview_scene(), Some("red")); + assert_eq!(studio.compositor.switcher(None).program_scene(), Some("red")); studio.render(start + Duration::from_millis(500)); let actual = pixel(&studio.program(), 1, 1); assert!(near(actual, [128, 0, 127, 255]), "still fading from red: {actual:?}"); @@ -2373,8 +2624,8 @@ mod tests { // dropped, so one that later takes its name is not drawn unasked. studio.collection.scenes.retain(|scene| scene.name != "red"); studio.compositor.retain(&studio.collection); - assert_eq!(studio.compositor.preview_scene(), None); - assert_eq!(studio.compositor.program_scene(), None); + assert_eq!(studio.compositor.switcher(None).preview_scene(), None); + assert_eq!(studio.compositor.switcher(None).program_scene(), None); }) .await; } @@ -2439,13 +2690,13 @@ mod tests { ); collection.scenes[1].items = vec![Item::new("overlay", Rect::fill(&collection.canvas))]; let mut studio = Studio::new(collection).await; - studio.compositor.set_preview(Some("red".into())); - studio.compositor.cut(); + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); + studio.compositor.switcher_mut(None).cut(); // Opened here, but drawn nowhere until the cut. let start = Instant::now(); studio.render(start); - studio.compositor.set_preview(Some("blue".into())); - studio.compositor.cut(); + studio.compositor.switcher_mut(None).set_preview(Some("blue".into())); + studio.compositor.switcher_mut(None).cut(); for (at, expected) in [(500, [128, 0, 128, 255]), (750, [64, 0, 191, 255]), (2000, BLUE)] { studio.render(start + Duration::from_millis(at)); @@ -2486,11 +2737,14 @@ mod tests { collection.scenes[1].items = vec![Item::new("overlay", tile(4.0, 4.0))]; let mut studio = Studio::new(collection).await; let start = Instant::now(); - studio.compositor.set_preview(Some("red".into())); - studio.compositor.cut(); - studio.compositor.set_preview(Some("blue".into())); + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); + studio.compositor.switcher_mut(None).cut(); + studio.compositor.switcher_mut(None).set_preview(Some("blue".into())); let duration = Duration::from_secs(1); - studio.compositor.transition(Transition::Move { duration }, start); + studio + .compositor + .switcher_mut(None) + .transition(Transition::Move { duration }, start); studio.render(start + Duration::from_millis(500)); let program = studio.program(); @@ -2525,12 +2779,12 @@ mod tests { let start = Instant::now(); // Paint the aside overlay, then leave it off air. - studio.compositor.set_preview(Some("red".into())); - studio.compositor.cut(); + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); + studio.compositor.switcher_mut(None).cut(); studio.render(start); assert_eq!(pixel(&studio.program(), 3, 3), [255, 255, 0, 255]); - studio.compositor.set_preview(Some("blue".into())); - studio.compositor.cut(); + studio.compositor.switcher_mut(None).set_preview(Some("blue".into())); + studio.compositor.switcher_mut(None).cut(); studio.render(start); studio.compositor.overlays.as_ref().unwrap().panic(); @@ -2540,8 +2794,8 @@ mod tests { assert!(err.contains("overlay thread stopped"), "{name}: {err}"); } - studio.compositor.set_preview(Some("red".into())); - studio.compositor.cut(); + studio.compositor.switcher_mut(None).set_preview(Some("red".into())); + studio.compositor.switcher_mut(None).cut(); studio.render(start); assert_eq!(pixel(&studio.program(), 3, 3), RED, "no frozen overlay"); }) @@ -2602,13 +2856,21 @@ mod bench { collection } - /// The mean time to draw and finish every canvas once. - async fn frame(collection: Collection) -> Duration { + /// The mean time to draw and finish every canvas once, with every canvas's + /// preview if `previews`, or only the main canvas's. + async fn frame(collection: Collection, previews: bool) -> Duration { let mut studio = Studio::new(collection).await; studio.compositor.set_canvases(&studio.collection).unwrap(); + if previews { + let names: Vec = studio.collection.canvases.keys().cloned().collect(); + let canvases: Vec> = std::iter::once(None) + .chain(names.iter().map(|name| Some(name.as_str()))) + .collect(); + studio.compositor.set_previewed(&canvases); + } let scene = studio.collection.scenes[0].name.clone(); - studio.compositor.set_preview(Some(scene)); - studio.compositor.cut(); + studio.compositor.switcher_mut(None).set_preview(Some(scene)); + studio.compositor.switcher_mut(None).cut(); let mut total = Duration::ZERO; for frame in 0..FRAMES * 2 { let start = Instant::now(); @@ -2622,18 +2884,24 @@ mod bench { } /// Sources against canvases, each further canvas placing every source - /// (dense) or two (sparse). A sparse canvas should cost about the same - /// however many sources the collection has. + /// (dense) or two (sparse), with every canvas's preview drawn or only the + /// main one's, as when the studio shows the main canvas. A sparse canvas + /// should cost about the same however many sources the collection has, + /// and an unseen one only its program. #[tokio::test] #[ignore = "a benchmark; run by hand on a GPU"] async fn sources_against_canvases() { local(async { - println!("sources canvases dense sparse"); + println!("sources canvases dense (all, main) sparse (all, main)"); for sources in [4, 16, 64] { for canvases in [1, 2, 4] { - let dense = frame(collection(sources, canvases, sources)).await; - let sparse = frame(collection(sources, canvases, 2)).await; - println!("{sources:>7} {canvases:>8} {dense:>7.2?} {sparse:>7.2?}"); + let dense = collection(sources, canvases, sources); + let sparse = collection(sources, canvases, 2); + let (dense_all, dense_main) = (frame(dense.clone(), true).await, frame(dense, false).await); + let (sparse_all, sparse_main) = (frame(sparse.clone(), true).await, frame(sparse, false).await); + println!( + "{sources:>7} {canvases:>8} {dense_all:>8.2?} {dense_main:>8.2?} {sparse_all:>8.2?} {sparse_main:>8.2?}" + ); } } }) diff --git a/rs/moq-scene/src/edit.rs b/rs/moq-scene/src/edit.rs index fb00dc9..097359c 100644 --- a/rs/moq-scene/src/edit.rs +++ b/rs/moq-scene/src/edit.rs @@ -696,8 +696,44 @@ impl Draft { }) } + /// Unlink the further canvas called `name`, to take its preview to program + /// by `transition` on its own, or link it back to the main canvas with + /// `None`. + pub fn set_extra_transition(&mut self, name: &str, transition: Option) -> Result<(), Error> { + self.update_extra(name, |extra| extra.transition = transition) + } + + /// Have recordings hold the further canvas `canvas`, or the main one for + /// `None`. + pub fn set_recorded(&mut self, canvas: Option<&str>) -> Result<(), Error> { + if let Some(name) = canvas + && !self.collection.canvases.contains_key(name) + { + return Err(conflict(format!("there is no canvas `{name}`"))); + } + let names: Vec = self.collection.canvases.keys().cloned().collect(); + for name in names { + let recorded = canvas == Some(name.as_str()); + self.update_extra(&name, |extra| extra.recorded = recorded)?; + } + Ok(()) + } + + /// Change the further canvas called `name` by `update`. + fn update_extra(&mut self, name: &str, update: impl FnOnce(&mut Extra)) -> Result<(), Error> { + let before = self.collection.canvases.get(name).cloned(); + let before = before.ok_or_else(|| conflict(format!("there is no canvas `{name}`")))?; + let mut after = before.clone(); + update(&mut after); + self.change(Change::Extra { + name: name.to_string(), + before: Some(before), + after: Some(after), + }) + } + /// Remove the further canvas called `name`, with its top layer and every - /// scene's layout on it. + /// scene's layout on it. Recordings holding it go back to the main canvas. pub fn remove_extra(&mut self, name: &str) -> Result<(), Error> { let laid: Vec<(usize, Scene)> = self .collection @@ -1438,6 +1474,37 @@ mod tests { assert_eq!(collection, start); } + /// Recording a canvas moves the pick off any other in one edit, removing + /// the recorded canvas sends recordings back to the main one, and either + /// undoes whole. + #[test] + fn the_recorded_canvas_moves_and_falls_back() { + let mut collection = collection(); + collection.canvases.insert("tall".into(), Extra::new(1080, 1920)); + collection.canvases.insert("wide".into(), Extra::new(2560, 1080)); + let mut history = History::new(); + let record = |collection: &mut Collection, history: &mut History, canvas: Option<&str>| { + let mut draft = collection.draft("Record"); + draft.set_recorded(canvas).unwrap(); + commit(collection, history, draft, None); + }; + + record(&mut collection, &mut history, Some("tall")); + assert_eq!(collection.recorded(), Some("tall")); + record(&mut collection, &mut history, Some("wide")); + assert_eq!(collection.recorded(), Some("wide")); + let mut draft = collection.draft("Remove wide"); + draft.remove_extra("wide").unwrap(); + commit(&mut collection, &mut history, draft, None); + assert_eq!(collection.recorded(), None); + assert!(collection.draft("Record").set_recorded(Some("wide")).is_err()); + + history.undo(&mut collection).unwrap(); + assert_eq!(collection.recorded(), Some("wide")); + history.undo(&mut collection).unwrap(); + assert_eq!(collection.recorded(), Some("tall")); + } + /// A renamed mix keeps its place, the program's and recordings' picks, and /// every source left out of it; a removed one takes those exclusions with /// it, sends recordings back to the mix on program, and undoes whole; and diff --git a/rs/moq-scene/src/lib.rs b/rs/moq-scene/src/lib.rs index f71e24a..62bcf54 100644 --- a/rs/moq-scene/src/lib.rs +++ b/rs/moq-scene/src/lib.rs @@ -17,10 +17,12 @@ //! compositor and sound goes to a `moq_audio::bus::Bus`, routed into its //! program mix. //! - [`Compositor`] draws the scene in preview and the scene on program into -//! textures on the caller's `wgpu` device, one pair per canvas from one -//! upload of each source's picture, cutting or running a [`Transition`] -//! between them on every canvas together, and plays each item's -//! [`Keyframe`]s from when its scene went on. `moq_video::render::export` hands the +//! textures on the caller's `wgpu` device, one [`Output`] per canvas from +//! one upload of each source's picture, its preview only while someone +//! looks. A [`Switcher`] cuts or runs a [`Transition`] between them on the +//! main canvas and every canvas linked to it together, or on an unlinked +//! canvas alone, and plays each item's [`Keyframe`]s from when its scene +//! went on. `moq_video::render::export` hands the //! 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 @@ -51,7 +53,7 @@ pub use collection::{ Audio, Blend, Canvas, Collection, Crop, Direction, Ease, Extra, Filter, Item, Keyframe, Layer, MaskMode, Mixes, Rect, Rgba, Scene, Source, Transition, Video, }; -pub use compositor::{Compositor, FORMAT}; +pub use compositor::{Compositor, FORMAT, Output, Switcher}; pub use edit::{Change, Draft, Edit, History}; pub use scale::Scaler; diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index c26b25e..f564aec 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -205,16 +205,18 @@ impl State { } /// The sources whose pictures any canvas's program draws, through clones: - /// each visible item's of the scenes on air, including one leaving in a - /// transition, and of the top layers over them, and each mask's it reads. + /// each visible item's of the scenes on air there, including one leaving + /// in a transition, and of the top layer over them, and each mask's it + /// reads. fn shown(&self) -> BTreeSet { let collection = &self.collection; - let scenes: Vec> = self.compositor.on_air().map(Some).chain([None]).collect(); - let canvases = std::iter::once(None).chain(collection.canvases.keys().map(Some)); + let compositor = &self.compositor; + let canvases = std::iter::once(None).chain(collection.canvases.keys().map(|name| Some(name.as_str()))); let items = canvases .flat_map(|canvas| { - scenes.iter().map(move |scene| moq_scene::Layer { - canvas: canvas.cloned(), + let scenes = compositor.switcher(canvas).on_air().map(Some).chain([None]); + scenes.map(move |scene| moq_scene::Layer { + canvas: canvas.map(str::to_string), scene: scene.map(str::to_string), }) }) @@ -298,6 +300,8 @@ pub struct Stats { pub rendered: AtomicU64, /// The main canvas's program video. pub video: Arc, + /// The recording's own video, of whichever canvas recordings hold. + pub recording: Arc, /// Each further canvas's program video, by the canvas's name, while it is /// drawn. canvases: Mutex>>, @@ -313,11 +317,11 @@ pub struct Stats { recording_audio_error: Mutex>, } -/// How one canvas's program video is doing. +/// How one canvas's program video, or the recording's, is doing. #[derive(Default)] pub struct VideoStats { - /// What each of its video encoders costs: the canvas size, the half size, - /// and, for the main canvas, the recording's own canvas size. + /// What each of its video encoders costs: a canvas's size and half size, + /// or, third, the recording's 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. @@ -755,9 +759,11 @@ impl Engine { browser: browser(&gpu)?, }); let mut compositor = Compositor::new(&gpu.device, &gpu.queue, collection.canvas)?; + // Before the canvases, so an unlinked one starts on the same scene. + let switcher = compositor.switcher_mut(None); + switcher.set_preview(collection.scenes.first().map(|scene| scene.name.clone())); + switcher.cut(); compositor.set_canvases(&collection)?; - compositor.set_preview(collection.scenes.first().map(|scene| scene.name.clone())); - compositor.cut(); let engine = Arc::new(Self { state: Mutex::new(State { @@ -789,6 +795,7 @@ impl Engine { let Publish { renditions, ladder, + recording, extras, audio, recorded_audio, @@ -812,14 +819,18 @@ impl Engine { // Opened before the program clock starts the sources, as the main // canvas's encoder is, so each holds its encode session first. let mut lines = Lines::new(); - let canvases: Vec<(String, Canvas)> = { + let (canvases, recorded): (Vec<(String, Canvas)>, Canvas) = { let state = engine.state.lock().unwrap(); let collection = state.collection(); let canvases = collection.canvases.keys(); - canvases + let canvases = canvases .filter_map(|name| Some((name.clone(), collection.canvas(Some(name))?))) - .collect() + .collect(); + (canvases, collection.canvas(collection.recorded()).unwrap_or(canvas)) }; + // Probed now too, so a canvas a recording cannot encode says so before + // anyone records it. It opens its session only once recorded. + let recording = Line::open(recording, recorded).await; for (name, canvas) in canvases { let video = Arc::new(VideoStats::default()); engine @@ -837,8 +848,14 @@ impl Engine { // publication alive beside a retry's. let starting = Starting(Some(engine.clone())); let program = engine.clone(); + let encoders = Encoders { + main, + recording, + extras, + lines, + }; let thread = clocked("program".into(), async move || { - program.run_program(start, epoch, main, extras, lines).await + program.run_program(start, epoch, encoders).await })?; engine.threads.lock().unwrap().push(thread); let mix = engine.clone(); @@ -1069,17 +1086,25 @@ impl Engine { .collect(); (stalled, stopped) }; - let video = &stats.video; - let canvas = &video.renditions[0]; + let video = |video: &VideoStats, failed: Option| { + let canvas = &video.renditions[Role::Canvas as usize]; + health::Video { + dropped: canvas.dropped.load(Ordering::Relaxed), + encoded: canvas.encoded.load(Ordering::Relaxed), + encoding: video.encoding.load(Ordering::Relaxed), + watched: video.watched.load(Ordering::Relaxed), + encoder: video.encoder(), + failed, + } + }; + // The main canvas's output stops with the audio too. + let main = (None, video(&stats.video, stats.error())); + let extras = stats.canvases().into_iter(); + let extras = extras.map(|(name, stats)| (Some(name), video(&stats, stats.error()))); health::Local { rendered: stats.rendered.load(Ordering::Relaxed), - dropped: canvas.dropped.load(Ordering::Relaxed), - encoded: canvas.encoded.load(Ordering::Relaxed), - encoding: video.encoding.load(Ordering::Relaxed), + canvases: std::iter::once(main).chain(extras).collect(), grants: stats.grants(), - watched: video.watched.load(Ordering::Relaxed), - encoder: video.encoder(), - failed: stats.error(), stalled, stopped, } @@ -1089,18 +1114,23 @@ impl Engine { /// canvas's program to its renditions' encoders, stamped `epoch` past /// `start`, while their gates let them 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 main canvas's - /// recording copy only while recording. + /// and the half size only while anyone watches it. The program of the + /// canvas recordings hold goes to the recording's own encoder too, which + /// encodes only while recording. /// - /// `main` encodes the main canvas, and `lines` the further canvases the - /// studio started with. Each further canvas added since gets a [`Line`] of - /// its own from `extras` once drawn, in a broadcast of its own, and each - /// ends once the collection drops its canvas. Drawing outlives the encoders: a failed - /// output stops copying its program, and the canvases keep moving. + /// See [`Encoders`] for what encodes each canvas. Drawing outlives the + /// encoders: a failed output stops copying its program, and the canvases + /// 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, mut main: Line, extras: Extras, mut lines: Lines) { + async fn run_program(&self, start: Instant, epoch: Duration, encoders: Encoders) { + let Encoders { + mut main, + mut recording, + extras, + mut lines, + } = encoders; let mut framerate = self.state.lock().unwrap().compositor.canvas().framerate; let mut clock = program_clock(start, framerate); // The encoders of lines since ended, waited on as the studio stops so @@ -1113,6 +1143,7 @@ impl Engine { // Without a sender, each encoder publishes what it was sent and // still holds, then finishes its track. retired.push(main.end(None)); + retired.push(recording.end(None)); retired.extend( lines .into_values() @@ -1124,7 +1155,8 @@ impl Engine { } return; } - let drawn: Vec<(Option, wgpu::Texture, Canvas)> = { + // Each canvas's program, by name, and the canvas recordings hold. + let (drawn, recorded): (Drawn, Option) = { let mut guard = self.state.lock().unwrap(); let state = &mut *guard; if std::mem::take(&mut state.stale) { @@ -1146,11 +1178,12 @@ impl Engine { continue; } self.program.on_program(state, shown); - state + let drawn = state .compositor .outputs() .map(|(name, output)| (name.map(str::to_string), output.program().clone(), output.canvas())) - .collect() + .collect(); + (drawn, state.collection.recorded().map(str::to_string)) }; self.stats.rendered.fetch_add(1, Ordering::Relaxed); @@ -1175,6 +1208,10 @@ impl Engine { self.stats.canvases.lock().unwrap().remove(&name); } + // The collection names only a canvas it has, which is drawn. + if let Some((_, program, canvas)) = drawn.iter().find(|(name, _, _)| *name == recorded) { + recording.feed(program, *canvas, timestamp); + } for (name, program, canvas) in drawn { let line = match &name { None => &mut main, @@ -1328,6 +1365,8 @@ struct Publish { /// The renditions of the canvas the studio started with, unless the canvas /// could not be encoded without a relay. ladder: Option, + /// Where the recording's own video goes, of whichever canvas it holds. + recording: Renditions, /// What publishes each further canvas. extras: Extras, /// The program audio's encoder. @@ -1371,7 +1410,7 @@ async fn publish( // Probed now, so a machine that cannot encode the canvas fails at startup // rather than once the program clock starts. Without a relay the studio // still composes, previews, and records audio, and a new canvas tries again. - let probe = match Probe::new(gpu, canvas, true).await { + let probe = match Probe::new(gpu, canvas, false).await { Ok(probe) => Some(probe), Err(err) if live.is_none() => { stats.video.fail(format!("{err:#}")); @@ -1392,22 +1431,14 @@ async fn publish( // both and any solicitation. let reserved = catalog.reserve(); // Viewers report back beneath the broadcast, which only a relay carries. - let viewers = match live { - Some(_) => { - let (echo, viewers) = health::echo(path)?; - let mut soliciting = catalog.clone(); - let mut section = soliciting.modify()?; - section.echo = Some(hang::catalog::Echo::new(echo)); - section.commit()?; - Some(viewers) - } - None => None, - }; + if live.is_some() { + health::solicit(&mut catalog.clone(), path)?; + } // What a recording reads the program from: its own canvas-size encoder, - // at full quality whatever the grant, and the mix it holds, 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. + // of whichever canvas it holds, at full quality whatever the grant, and + // the mix it holds, 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") @@ -1424,10 +1455,7 @@ async fn publish( gpu: gpu.clone(), broadcast: broadcast.clone(), timebase: catalog.timebase(), - recording: Some(Recording { - broadcast: recorder.clone(), - timebase: recorded.timebase(), - }), + record: false, bandwidth: bandwidth.clone(), // Live on a relay and not on demand; otherwise only a viewer asks for // the video. @@ -1437,6 +1465,18 @@ async fn publish( video: stats.video.clone(), grants: String::new(), }; + let recorded_video = Renditions { + gpu: gpu.clone(), + broadcast: recorder.clone(), + timebase: recorded.timebase(), + record: true, + bandwidth: moq_net::bandwidth::Allocator::unlimited(), + on_demand: true, + live: false, + stats: stats.clone(), + video: stats.recording.clone(), + grants: String::new(), + }; let ladder = match probe { Some(probe) => { let (ladder, opened) = renditions.start(probe)?; @@ -1478,27 +1518,21 @@ async fn publish( .announce(moq_net::origin::Route::default().with_instance(instance.clone(), 0)) .context("cannot announce the broadcast")?; - let health = live.zip(viewers).map(|(live, viewers)| { - let feedback = health::Feedback::default(); - let watching = health::Watching::default(); - tokio::spawn(log( - "viewer feedback", - feedback.clone().follow(live.reports.clone(), viewers), - )); - tokio::spawn(log( - "relay stats", - watching.clone().follow(live.reports.clone(), path.clone()), - )); - health::Publisher { - observer: hang::health::Observer::new(path.clone(), Some(instance.clone())), - catalog: graded, - clock: clock.clone(), - relays: live.links.clone(), - feedback, - watching, - sources: sources.clone(), + let health = match live { + Some(live) => { + let observer = hang::health::Observer::new(path.clone(), Some(instance.clone())); + Some(health::Publisher { + main: health::Broadcast::follow(observer, graded, &live.reports)?, + canvases: health::Canvases::new(live.reports.clone()), + clock: clock.clone(), + relays: live.links.clone(), + sources: sources.clone(), + }) } - }); + None => None, + }; + // Each further canvas is graded beside the main one, with its own viewers. + let graded = health.as_ref().map(|health| health.canvases.clone()); Ok(( Program { @@ -1523,9 +1557,11 @@ async fn publish( on_demand: renditions.on_demand, live: renditions.live, stats: stats.clone(), + graded, }, renditions, ladder, + recording: recorded_video, audio, recorded_audio, clock, @@ -1555,26 +1591,23 @@ impl Role { } } -/// Where a recording reads the program from: a broadcast the studio keeps to -/// itself, holding the recording's own canvas-size video and its own encoding -/// of the mix it holds, 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. +/// rendition its own track, since a track never changes what it carries; or +/// where the recording's own video goes, of whichever canvas it holds. #[derive(Clone)] struct Renditions { gpu: Gpu, + /// A canvas's broadcast, or the recording's: one the studio keeps to + /// itself, holding the recording's video and its own encoding of the mix + /// it holds, listed in a catalog of its own on the program's clock. broadcast: moq_net::broadcast::Producer, /// 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: the main canvas's alone. - recording: Option, + /// Whether this is the recording's: its one rendition the canvas size + /// again, at full quality whatever the grant, rather than the published + /// canvas and half sizes. + record: bool, /// The connection's bandwidth, which each published rendition follows its /// share of. bandwidth: bandwidth::Allocator, @@ -1591,6 +1624,16 @@ struct Renditions { } impl Renditions { + /// Probe for `canvas`'s renditions and [`start`](Self::start) them, waiting + /// for the canvas size's encoder to open, as at startup, so it holds its + /// encode session before any source opens one. + async fn open(&self, canvas: Canvas) -> anyhow::Result { + let probe = Probe::new(&self.gpu, canvas, self.record).await?; + let (ladder, opened) = self.start(probe)?; + opened.await.context("the encode thread stopped")??; + Ok(ladder) + } + /// 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 @@ -1626,11 +1669,7 @@ impl Renditions { config, video, } = probed; - let (broadcast, timebase) = match (role, &self.recording) { - (Role::Record, Some(recording)) => (&recording.broadcast, &recording.timebase), - (Role::Record, None) => anyhow::bail!("only the main canvas is recorded"), - _ => (&self.broadcast, &self.timebase), - }; + let (broadcast, 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", @@ -1734,17 +1773,19 @@ struct Probed { impl Probe { /// Open a throwaway encoder of each of `canvas`'s renditions on what `gpu` - /// exports, the recording's copy only if `record`, which also settles - /// whether the encoder reads the program on the GPU. The canvas size picks - /// the encoder, which the others use too. + /// exports, the published sizes or, if `record`, the recording's alone, + /// which also settles whether the encoder reads the program on the GPU. + /// The first picks the encoder, which the other uses too. async fn new(gpu: &Gpu, canvas: Canvas, record: bool) -> anyhow::Result { 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(record.then_some((Role::Record, size))); + let roles: Vec<(Role, moq_video::Size)> = match record { + true => vec![(Role::Record, size)], + false => [Role::Canvas, Role::Half] + .into_iter() + .zip(ladder::sizes(size)) + .collect(), + }; let mut kind = moq_video::encode::Kind::Auto; let mut rungs = Vec::new(); let mut refused = Vec::new(); @@ -1757,7 +1798,7 @@ impl Probe { .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), + Err(err) if role != Role::Half => return Err(err), // The program goes on without it. Err(err) => { refused.push((role, format!("{err:#}"))); @@ -1842,6 +1883,14 @@ impl Ended { } impl Line { + /// `renditions` of `canvas`, opened at once, as at startup. One whose + /// encoder cannot open draws on without video, saying why. + async fn open(renditions: Renditions, canvas: Canvas) -> Self { + let ladder = renditions.open(canvas).await; + let ladder = ladder.inspect_err(|err| renditions.video.fail(format!("{err:#}"))); + Self::new(renditions, canvas, ladder.ok(), true) + } + /// `renditions` of `canvas`: `ladder` when already open, or `None` to probe /// for them on the first feed unless `settled`, as a canvas that failed /// to open at startup is. @@ -1894,8 +1943,7 @@ impl Line { } if self.probing.is_none() && !self.settled { let ending = std::mem::take(&mut self.ending); - let record = renditions.recording.is_some(); - match Probe::spawn(renditions.gpu.clone(), self.canvas, record, ending) { + match Probe::spawn(renditions.gpu.clone(), self.canvas, renditions.record, ending) { Ok(probe) => self.probing = Some((self.canvas, probe)), Err(err) => { self.settled = true; @@ -1962,12 +2010,31 @@ impl Line { /// to start, which is not retried until the canvas is removed and added again. type Lines = BTreeMap>; +/// Each canvas's program as drawn, by name, the main one first as `None`, +/// with the canvas drawn. +type Drawn = Vec<(Option, wgpu::Texture, Canvas)>; + +/// What the program clock hands each canvas's program to. +struct Encoders { + /// The main canvas's. + main: Line, + /// The recording's own, fed whichever canvas recordings hold, which a + /// change of canvas size probes for again as any line does. + recording: Line, + /// What starts the line of a further canvas added while the studio runs. + extras: Extras, + /// The further canvases' lines, those the studio started with first. + lines: Lines, +} + /// A further canvas's broadcast, whose catalog finishes and which closes /// when dropped. struct Broadcast { publication: moq_net::origin::Publication, /// Lists the canvas's video and the main broadcast's audio. catalog: moq_scene::publish::Catalog, + /// Its grading, while live, which stops with it. + _graded: Option, } impl Drop for Broadcast { @@ -1991,6 +2058,8 @@ struct Extras { on_demand: bool, live: bool, stats: Arc, + /// Where each canvas's broadcast is graded, while live. + graded: Option, } impl Extras { @@ -2020,21 +2089,8 @@ impl Extras { video: Arc, program: &Program, ) -> anyhow::Result<(Line, Broadcast)> { - let (renditions, broadcast) = self.publish(name, video.clone(), program)?; - let opened = async { - let probe = Probe::new(&self.gpu, canvas, false).await?; - let (ladder, opened) = renditions.start(probe)?; - opened.await.context("the encode thread stopped")??; - anyhow::Ok(ladder) - }; - let ladder = match opened.await { - Ok(ladder) => Some(ladder), - Err(err) => { - video.fail(format!("{err:#}")); - None - } - }; - Ok((Line::new(renditions, canvas, ladder, true), broadcast)) + let (renditions, broadcast) = self.publish(name, video, program)?; + Ok((Line::open(renditions, canvas).await, broadcast)) } /// Publish the further canvas `name` at its own path beneath the main @@ -2079,15 +2135,27 @@ impl Extras { catalog.audio.renditions.insert(id, config); } })?; + // Its viewers report back beneath it, as the main broadcast's do. + if self.graded.is_some() { + health::solicit(&mut catalog, &path)?; + } + let instance = moq_net::Instance::random(); publication - .announce(moq_net::origin::Route::default().with_instance(moq_net::Instance::random(), 0)) + .announce(moq_net::origin::Route::default().with_instance(instance.clone(), 0)) .with_context(|| format!("cannot announce the broadcast of canvas `{name}`"))?; + let graded = match &self.graded { + Some(canvases) => { + let observer = hang::health::Observer::new(path.clone(), Some(instance)); + Some(canvases.follow(name, observer, catalog.clone())?) + } + None => None, + }; let renditions = Renditions { gpu: self.gpu.clone(), broadcast, timebase: catalog.timebase(), - recording: None, + record: false, bandwidth: self.bandwidth.clone(), on_demand: self.on_demand, live: self.live, @@ -2095,7 +2163,11 @@ impl Extras { video, grants: format!("{name}/"), }; - let broadcast = Broadcast { publication, catalog }; + let broadcast = Broadcast { + publication, + catalog, + _graded: graded, + }; Ok((renditions, broadcast)) } } @@ -2258,13 +2330,6 @@ fn connect( .unzip() } -/// Run a background reader, logging why it stopped. -async fn log(what: &'static str, task: impl Future>) { - if let Err(err) = task.await { - tracing::warn!(%err, "stopped reading {what}"); - } -} - /// What the program clock hands a rendition's encoder. enum Video { /// A copy of the program on its way to the encoder, to encode at this @@ -2631,7 +2696,7 @@ mod tests { /// Program frames the recording's own encoder encoded. fn recorded(engine: &Engine) -> u64 { - engine.stats.video.renditions[Role::Record as usize] + engine.stats.recording.renditions[Role::Record as usize] .encoded .load(Ordering::Relaxed) } @@ -2694,9 +2759,9 @@ mod tests { /// Without a relay, a recording encodes the program on its own and saves it /// as a regular MP4 holding its video and audio. The color source publishes - /// a still, which has no file of its own. A new canvas goes on in another - /// file, holding the new rendition's video. A recording holding a mix - /// other than the one on program records its audio. + /// a still, which has no file of its own. Picking a further canvas of + /// another size goes on in another file, holding its video. A recording + /// holding a mix other than the one on program records its audio. #[tokio::test(flavor = "multi_thread")] #[ignore = "needs a Vulkan device and an H.264 encoder; run by just rs render-sw"] async fn the_program_records_without_a_relay() { @@ -2722,13 +2787,17 @@ mod tests { assert_eq!(engine.stats.sessions.open(), 1); { let mut state = engine.state.lock().unwrap(); - let mut draft = state.collection().draft("Resize the canvas"); - let canvas = moq_scene::Canvas { - width: 160, - height: 90, - framerate: 15, + let mut draft = state.collection().draft("Record a portrait canvas"); + draft.set_extra("tall", 90, 160).unwrap(); + let fill = moq_scene::Rect { + x: 0.0, + y: 0.0, + width: 90.0, + height: 160.0, }; - draft.set_canvas(canvas).unwrap(); + let layer = moq_scene::Layer::scene("Scene").on(Some("tall".into())); + draft.insert_item(&layer, 0, moq_scene::Item::new("red", fill)).unwrap(); + draft.set_recorded(Some("tall")).unwrap(); state.commit(draft, None).unwrap(); } encoded(&engine, 90 + 45).await; @@ -2959,6 +3028,27 @@ mod tests { .expect("the canvas's own layout arrives"); assert_eq!(engine.stats.sessions.open(), 2, "one canvas size each"); assert!(engine.stats.canvases()["tall"].error().is_none()); + assert!(catalog.echo.is_some(), "its viewers are asked for feedback"); + // Health grades it beside the main canvas, from what it writes. + let graded = |engine: &Engine| { + let report = engine.health.lock().unwrap(); + let tall = report + .canvases + .iter() + .find(|canvas| canvas.name.as_deref() == Some("tall")); + tall.map(|tall| { + tall.tracks + .iter() + .any(|track| track.sent.is_some_and(|sent| sent.as_bps() > 0)) + }) + }; + kio::clock::timeout(TIMEOUT, async { + while graded(&engine) != Some(true) { + kio::clock::sleep(health::INTERVAL).await; + } + }) + .await + .expect("health grades the further canvas"); let track = broadcast.track(&video).unwrap().subscribe(None).await.unwrap(); { @@ -2973,6 +3063,13 @@ mod tests { .expect("the canvas's broadcast ends"); assert!(engine.stats.canvases().is_empty()); assert_eq!(engine.stats.sessions.open(), 1, "the canvas's session given back"); + kio::clock::timeout(TIMEOUT, async { + while graded(&engine).is_some() { + kio::clock::sleep(health::INTERVAL).await; + } + }) + .await + .expect("health forgets the removed canvas"); // Back again, quitting ends it as cleanly. engine.state.lock().unwrap().undo().unwrap(); @@ -3258,9 +3355,10 @@ mod tests { loop { { let report = engine.health.lock().unwrap(); - let sending = report.tracks.len() == 2 - && report - .tracks + let tracks = report.canvases.first().map_or(&[][..], |main| &main.tracks); + let sending = report.sources.is_empty() + && tracks.len() == 2 + && tracks .iter() .all(|track| track.sent.is_some_and(|sent| sent.as_bps() > 0)); let connected = matches!(&report.relays[..], [relay] if relay.state == health::State::Connected); diff --git a/rs/moq-studio/src/health.rs b/rs/moq-studio/src/health.rs index 9fd11ca..5b76e1c 100644 --- a/rs/moq-studio/src/health.rs +++ b/rs/moq-studio/src/health.rs @@ -3,12 +3,13 @@ //! //! Every [`INTERVAL`] the studio grades its stream with `hang::health`, the //! grader the rest of the stack shares, from three points of view: -//! - its own: what each track of the broadcast wrote, each relay connection's -//! state, and the serving one's round trip, estimated capacity, and loss; -//! - its viewers': the feedback each publishes under the broadcast's `echo` -//! path, which the catalog solicits (`moq export --echo` answers it); -//! - the relay's: how many sessions watch the broadcast, read from the relay's -//! stats broadcasts when it shares them with this connection. +//! - its own: what each track of every canvas's broadcast wrote, each relay +//! connection's state, and the serving one's round trip, estimated +//! capacity, and loss, which every canvas shares; +//! - its viewers': the feedback each publishes under its canvas broadcast's +//! `echo` path, which the catalog solicits (`moq export --echo` answers it); +//! - the relay's: how many sessions watch each broadcast, read from the +//! relay's stats broadcasts when it shares them with this connection. //! //! It adds what only the studio knows (program frames the encoder had no room //! for, sources that stopped delivering) and names the cause of each problem. @@ -55,15 +56,28 @@ pub struct Report { /// Each relay the broadcast goes to, the preferred first; empty when the /// studio publishes nothing. pub relays: Vec, + /// The connection every canvas's broadcast shares. pub uplink: Uplink, - pub viewers: Viewers, - /// Each track of the broadcast in catalog order, then each source's own - /// local track by source name. - pub tracks: Vec, + /// Each canvas's broadcast, the main one first; empty when the studio + /// publishes nothing. + pub canvases: Vec, + /// Each source's own local track, by source name. + pub sources: Vec, /// What is wrong, worst first, each naming its cause. pub problems: Vec, } +/// One canvas's broadcast: who watches it, and how each of its tracks reaches +/// them. +#[derive(Debug, Default)] +pub struct Canvas { + /// A further canvas's name, or `None` for the main one. + pub name: Option, + pub viewers: Viewers, + /// Each track it lists, in catalog order. + pub tracks: Vec, +} + /// One relay connection. #[derive(Debug, Clone, PartialEq)] pub struct Relay { @@ -95,7 +109,7 @@ pub struct Uplink { pub rtt: Option, /// What the congestion controller estimates the connection can carry. pub capacity: Option, - /// The media the program's tracks wrote over the last interval. Source + /// The media every canvas's tracks wrote over the last interval. Source /// tracks are left out: they stay in the studio, and never reach a relay. pub media: Option, /// Transport bytes lost per media byte written over the last interval. @@ -197,43 +211,204 @@ pub struct Problem { pub struct Local { /// Canvas frames drawn. pub rendered: u64, + /// Each canvas's program video: the main one's as `None`, then each + /// further one's by name. + pub canvases: BTreeMap, Video>, + /// Each program track's share of the connection's bandwidth, by name: a + /// further canvas's prefixed with its name and a slash. + pub grants: BTreeMap>, + /// Sources the program shows or plays that stopped delivering, and what. + pub stalled: Vec<(String, Stall)>, + /// Sources the program shows or plays that ended or failed. + pub stopped: Vec<(String, Status)>, +} + +impl Local { + /// Whether anyone subscribes to any canvas's program video. + fn watched(&self) -> bool { + self.canvases.values().any(|video| video.watched) + } +} + +/// What only the studio knows about one canvas's program video. +#[derive(Debug, Default, Clone)] +pub struct Video { /// Program frames the encoder had no room for. pub dropped: u64, /// Program frames encoded. pub encoded: u64, - /// Whether the program's canvas size should be encoding: always, or while - /// watched when on demand, and while its grant funds it. + /// Whether the canvas size should be encoding: always, or while watched + /// when on demand, and while its grant funds it. pub encoding: bool, - /// Each program track's share of the connection's bandwidth, by name. - pub grants: BTreeMap>, - /// Whether anyone subscribes to the program video. + /// Whether anyone subscribes to the video. pub watched: bool, /// The encoder backend, once one opened. pub encoder: Option, /// Why the output stopped, if it did. pub failed: Option, - /// Sources the program shows or plays that stopped delivering, and what. - pub stalled: Vec<(String, Stall)>, - /// Sources the program shows or plays that ended or failed. - pub stopped: Vec<(String, Status)>, } /// What the studio publishes, for grading its delivery. pub struct Publisher { - /// The broadcast's path and the instance it was announced with. - pub observer: Observer, - pub catalog: moq_scene::publish::Catalog, - /// The catalog's clock, which maps a track's timestamps to the wall time + /// The main canvas's broadcast. + pub main: Broadcast, + /// Each further canvas's, while it is published. + pub canvases: Canvases, + /// The catalogs' clock, which maps a track's timestamps to the wall time /// the program drew them. pub clock: moq_mux::Clock, - /// Each relay the broadcast goes to, the preferred first. + /// Each relay the broadcasts go to, the preferred first. pub relays: Vec, - pub feedback: Feedback, - pub watching: Watching, /// Where the sources keep their own tracks, for what those wrote. pub sources: moq_scene::publish::Publish, } +/// One canvas's broadcast, graded: what it lists, and what its viewers and +/// the relays say of it. +#[derive(Clone)] +pub struct Broadcast { + /// Its path and the instance it was announced with. + pub observer: Observer, + pub catalog: moq_scene::publish::Catalog, + pub feedback: Feedback, + pub watching: Watching, + /// Read for as long as any clone is held. + followers: Arc, +} + +impl Broadcast { + /// Grade the broadcast `observer` names, which `catalog` lists, reading + /// its viewers' feedback beneath it and the relays' count of who watches + /// it from `reports`, until the last clone drops. The catalog must solicit + /// that feedback, as [`solicit`] has it. + pub fn follow( + observer: Observer, + catalog: moq_scene::publish::Catalog, + reports: &moq_net::origin::Consumer, + ) -> anyhow::Result { + let (_, viewers) = echo(&observer.path)?; + let feedback = Feedback::default(); + let watching = Watching::default(); + let followers = Followers(vec![ + tokio::spawn(log( + "viewer feedback", + feedback.clone().follow(reports.clone(), viewers), + )), + tokio::spawn(log( + "relay stats", + watching.clone().follow(reports.clone(), observer.path.clone()), + )), + ]); + Ok(Self { + observer, + catalog, + feedback, + watching, + followers: Arc::new(followers), + }) + } +} + +/// Have the broadcast at `path`, which `catalog` lists, ask its viewers for +/// feedback beneath it. +pub fn solicit(catalog: &mut moq_scene::publish::Catalog, path: &moq_net::Path) -> anyhow::Result<()> { + let (echo, _) = echo(path)?; + let mut section = catalog.modify()?; + section.echo = Some(hang::catalog::Echo::new(echo)); + section.commit()?; + Ok(()) +} + +/// Tasks that stop when dropped. +struct Followers(Vec>); + +impl Drop for Followers { + fn drop(&mut self) { + for task in &self.0 { + task.abort(); + } + } +} + +/// Run a background reader, logging why it stopped. +async fn log(what: &'static str, task: impl Future>) { + if let Err(err) = task.await { + tracing::warn!(%err, "stopped reading {what}"); + } +} + +/// The further canvases' broadcasts graded, by canvas name, shared with what +/// publishes them. +#[derive(Clone)] +pub struct Canvases { + graded: Arc>>, + /// Where viewers' reports and the relays' stats arrive. + reports: moq_net::origin::Consumer, + /// Where their readers run, rather than on the thread that adds a canvas, + /// such as the program clock's. + runtime: tokio::runtime::Handle, +} + +impl Canvases { + /// Further canvases graded from what arrives on `reports`, read on the + /// tokio runtime this is called inside. + pub fn new(reports: moq_net::origin::Consumer) -> Self { + Self { + graded: Arc::default(), + reports, + runtime: tokio::runtime::Handle::current(), + } + } + + /// Grade canvas `name`'s broadcast, which `observer` names and `catalog` + /// lists, as [`Broadcast::follow`] does, until the returned handle drops. + pub fn follow( + &self, + name: &str, + observer: Observer, + catalog: moq_scene::publish::Catalog, + ) -> anyhow::Result { + let broadcast = { + let _runtime = self.runtime.enter(); + Broadcast::follow(observer, catalog, &self.reports)? + }; + let followers = broadcast.followers.clone(); + self.graded.lock().unwrap().insert(name.to_string(), broadcast); + Ok(Graded { + graded: self.graded.clone(), + name: name.to_string(), + followers, + }) + } + + fn snapshot(&self) -> Vec<(String, Broadcast)> { + let graded = self.graded.lock().unwrap(); + graded + .iter() + .map(|(name, broadcast)| (name.clone(), broadcast.clone())) + .collect() + } +} + +/// A further canvas's broadcast being graded, until dropped. +pub struct Graded { + graded: Arc>>, + name: String, + /// Its readers, which tell it from a canvas added back under its name + /// before it ended. + followers: Arc, +} + +impl Drop for Graded { + fn drop(&mut self) { + let mut graded = self.graded.lock().unwrap(); + let ours = graded.get(&self.name); + if ours.is_some_and(|broadcast| Arc::ptr_eq(&broadcast.followers, &self.followers)) { + graded.remove(&self.name); + } + } +} + /// The connection to one relay. #[derive(Clone)] pub struct Link { @@ -277,16 +452,23 @@ impl Sampler { .map(|previous| now.saturating_duration_since(previous.at)); let mut report = Report::default(); - let mut graded = Graded::default(); + // Each canvas's viewers, as graded, beside `report.canvases`. + let mut watched = Vec::new(); let (mut current, mut sources) = (None, None); let mut serving = None; if let Some(publisher) = &self.publisher { report.relays = publisher.relays.iter().map(relay).collect(); serving = report.relays.iter().position(|relay| relay.state == State::Connected); let link = serving.map(|index| &publisher.relays[index]); + let extras = publisher.canvases.snapshot().into_iter(); + let broadcasts: Vec<(Option, Broadcast)> = std::iter::once((None, publisher.main.clone())) + .chain(extras.map(|(name, broadcast)| (Some(name), broadcast))) + .collect(); + // Every canvas's tracks share the connection, so the uplink is graded + // on all of them. let ours = health::Report::new( - publisher.observer.clone(), - snapshot(publisher, link.map(|link| &link.monitor)), + publisher.main.observer.clone(), + snapshot(&broadcasts, link.map(|link| &link.monitor)), ); // Another relay's link counted its transport from another start. let previous_ours = previous @@ -294,42 +476,82 @@ impl Sampler { .filter(|previous| previous.serving == serving) .and_then(|previous| previous.report.as_ref()); let graded_ours = self.grade(&ours, previous_ours.zip(interval)); - report.uplink = uplink(&ours.snapshot, &graded_ours, local.watched); + report.uplink = uplink(&ours.snapshot, &graded_ours, local.watched()); report.uplink.relay = link.map(|link| link.endpoint.clone()); - graded.publisher = Some(graded_ours); // Graded apart from the program's, so their media stays out of the // uplink's; see [`Uplink::media`]. let mut snapshot = hang::stats::Snapshot::default(); snapshot.renditions = publisher.sources.stats(); - let theirs = health::Report::new(publisher.observer.clone(), snapshot); + let theirs = health::Report::new(publisher.main.observer.clone(), snapshot); let previous_theirs = previous.as_ref().and_then(|previous| previous.sources.as_ref()); - graded.sources = Some(self.grade(&theirs, previous_theirs.zip(interval))); - graded.viewers = publisher.feedback.grade(now, &self.thresholds); - report.viewers.unread = publisher.feedback.unread(); - report.viewers.watching = publisher.watching.get(); - let listed = Listed { - program: publisher.catalog.snapshot(), - sources: publisher.sources.catalog(), - grants: &local.grants, - }; - report.tracks = tracks(&listed, &publisher.clock, &graded); - report.viewers.reports = viewers(publisher, &graded.viewers); + let graded_theirs = self.grade(&theirs, previous_theirs.zip(interval)); + + for (name, broadcast) in broadcasts { + let graded = broadcast.feedback.grade(now, &self.thresholds); + let listed = Listed { + renditions: renditions(&broadcast.catalog.snapshot()).collect(), + prefix: prefix(name.as_deref()), + source: None, + sent: &graded_ours, + viewers: &graded, + grants: &local.grants, + }; + report.canvases.push(Canvas { + viewers: Viewers { + watching: broadcast.watching.get(), + reports: viewers(&graded, &publisher.clock), + unread: broadcast.feedback.unread(), + }, + tracks: tracks(&listed, &publisher.clock), + name, + }); + watched.push(graded); + } + let listed = publisher.sources.catalog(); + for (name, catalog) in &listed.ext.studio.sources { + let listed = Listed { + renditions: renditions(catalog).collect(), + prefix: String::new(), + source: Some(name), + sent: &graded_theirs, + viewers: &[], + grants: &local.grants, + }; + report.sources.extend(tracks(&listed, &publisher.clock)); + } current = Some(ours); sources = Some(theirs); } - let drops = previous.as_ref().map(|previous| Drops { - drawn: local.rendered.saturating_sub(previous.local.rendered), - dropped: local.dropped.saturating_sub(previous.local.dropped), - encoded: local.encoded.saturating_sub(previous.local.encoded), + let drops = previous + .as_ref() + .map(|previous| { + let drawn = local.rendered.saturating_sub(previous.local.rendered); + let canvases = local.canvases.iter(); + canvases + .filter_map(|(name, video)| { + let before = previous.local.canvases.get(name)?; + let drops = Drops { + drawn, + dropped: video.dropped.saturating_sub(before.dropped), + encoded: video.encoded.saturating_sub(before.encoded), + }; + Some((name.clone(), drops)) + }) + .collect() + }) + .unwrap_or_default(); + let seen = report.canvases.iter().zip(&watched).map(|(canvas, viewers)| Seen { + name: canvas.name.as_deref(), + viewers, + tracks: &canvas.tracks, }); report.problems = problems(&Facts { relays: &report.relays, local: &local, drops, uplink: &report.uplink, - viewers: &graded.viewers, - labels: &report.tracks, + canvases: seen.collect(), thresholds: &self.thresholds, }); @@ -376,11 +598,27 @@ fn relay(link: &Link) -> Relay { } } -/// What the studio's tracks have written, and the transport of the link -/// `monitor` watches. -fn snapshot(publisher: &Publisher, monitor: Option<&moq_tokio::connection::Monitor>) -> hang::stats::Snapshot { +/// What a further canvas's tracks are graded under, before each ID, since +/// two broadcasts can list the same one: its name and a slash, as its grants +/// are listed; nothing for the main canvas. +fn prefix(canvas: Option<&str>) -> String { + canvas.map(|name| format!("{name}/")).unwrap_or_default() +} + +/// What every canvas's tracks have written, each graded under its +/// [`prefix`], and the transport of the link `monitor` watches. +fn snapshot( + broadcasts: &[(Option, Broadcast)], + monitor: Option<&moq_tokio::connection::Monitor>, +) -> hang::stats::Snapshot { let mut snapshot = hang::stats::Snapshot::default(); - snapshot.renditions = publisher.catalog.stats(); + for (name, broadcast) in broadcasts { + let prefix = prefix(name.as_deref()); + let renditions = broadcast.catalog.stats().into_iter(); + snapshot + .renditions + .extend(renditions.map(|(id, rendition)| (format!("{prefix}{id}"), rendition))); + } if let Some(stats) = monitor.and_then(moq_tokio::connection::Monitor::stats) { snapshot.transport.rtt = stats.rtt; snapshot.transport.rate = stats.estimated_send_rate.map(Rate::as_bps); @@ -424,85 +662,120 @@ fn uplink(snapshot: &hang::stats::Snapshot, graded: &Health, watched: bool) -> U } } -/// What the studio's catalogs list now. +/// What one catalog lists now, to grade each of its tracks. struct Listed<'a> { - /// The program's broadcast. - program: hang::Catalog, - /// The sources' local one, nesting each source's catalog. - sources: hang::Catalog, + /// Each video and audio rendition, as [`renditions`] reads them. + renditions: Vec, + /// What its tracks are graded and granted under, before each ID. + prefix: String, + /// The source whose own catalog it is, or `None` for a canvas's. + source: Option<&'a String>, + /// What its tracks wrote, as graded. + sent: &'a Health, + /// Its viewers, as graded: a canvas's alone. + viewers: &'a [GradedViewer], /// Each program track's share of the connection, by name. grants: &'a BTreeMap>, } -/// Each media track `listed`, the program's and then each source's, with -/// what it sent and what its viewers received. A source's stills carry no -/// media to grade, so they are left out. -fn tracks(listed: &Listed, clock: &moq_mux::Clock, graded: &Graded) -> Vec { - let program = - renditions(&listed.program).map(|(id, label, enabled)| (None, id, label, enabled, graded.publisher.as_ref())); - let sources = listed.sources.ext.studio.sources.iter().flat_map(|(name, catalog)| { - renditions(catalog).map(move |(id, label, enabled)| (Some(name), id, label, enabled, graded.sources.as_ref())) - }); - program - .chain(sources) - .map(|(source, id, label, enabled, sent)| { - let ours = sent.and_then(|health| health.renditions.get(id)); - let interval = sent.and_then(|health| health.interval); - // A source's track is in the local broadcast, whose names can repeat - // the program's, and no relay viewer or grant reaches it. - let program = source.is_none(); - let mut track = Track { - source: source.cloned(), - id: id.clone(), - label, - granted: listed.grants.get(id).copied().flatten().filter(|_| program), - enabled, - sent: ours.and_then(|rendition| rendition.bitrate).map(Rate::from_bps), - framerate: ours - .and_then(|rendition| rendition.frames) - .zip(interval) - .map(|(frames, interval)| frames as f64 / interval.as_secs_f64()), - viewers: 0, - received: None, - verdict: Verdict::Unknown, - latency: None, - }; - for viewer in graded.viewers.iter().filter(|_| program) { - let Some(health) = &viewer.health else { continue }; - let Some(rendition) = health.renditions.get(id) else { - continue; +/// Each media track `listed`, with what it sent and what its viewers +/// received. A source's stills carry no media to grade, so they are left out. +fn tracks(listed: &Listed, clock: &moq_mux::Clock) -> Vec { + listed + .renditions + .iter() + .map( + |Rendition { + id, + label, + enabled, + main, + }| { + // One listed by reference to the main broadcast, as a further + // canvas lists the audio, is graded and granted there. + let graded = match main { + Some(track) => track.clone(), + None => format!("{}{id}", listed.prefix), }; - track.viewers += 1; - track.verdict = track.verdict.max(rendition.verdict); - if let Some(bitrate) = rendition.bitrate.map(Rate::from_bps) { - track.received = Some(track.received.map_or(bitrate, |received| received.min(bitrate))); + let ours = listed.sent.renditions.get(&graded); + let interval = listed.sent.interval; + // A source's track is in the local broadcast, whose names can repeat + // the program's, and no relay viewer or grant reaches it. + let program = listed.source.is_none(); + let mut track = Track { + source: listed.source.cloned(), + id: id.clone(), + label: label.clone(), + granted: listed.grants.get(&graded).copied().flatten().filter(|_| program), + enabled: *enabled, + sent: ours.and_then(|rendition| rendition.bitrate).map(Rate::from_bps), + framerate: ours + .and_then(|rendition| rendition.frames) + .zip(interval) + .map(|(frames, interval)| frames as f64 / interval.as_secs_f64()), + viewers: 0, + received: None, + verdict: Verdict::Unknown, + latency: None, + }; + for viewer in listed.viewers { + let Some(health) = &viewer.health else { continue }; + let Some(rendition) = health.renditions.get(id) else { + continue; + }; + track.viewers += 1; + track.verdict = track.verdict.max(rendition.verdict); + if let Some(bitrate) = rendition.bitrate.map(Rate::from_bps) { + track.received = Some(track.received.map_or(bitrate, |received| received.min(bitrate))); + } + let latency = viewer.latency(id, clock); + track.latency = track.latency.max(latency); } - let latency = viewer.latency(id, clock); - track.latency = track.latency.max(latency); - } - track - }) + track + }, + ) .collect() } +/// One video or audio rendition a catalog lists. +struct Rendition { + id: String, + /// What it carries, such as `video H.264 1920x1080`. + label: String, + enabled: bool, + /// Its track in the main broadcast, when listed there by reference: the + /// only broadcast the studio's catalogs refer to. + main: Option, +} + /// Each video and audio rendition `catalog` lists, by ID, with what it /// carries and whether it is enabled. -fn renditions(catalog: &hang::Catalog) -> impl Iterator { - let video = catalog.video.renditions.iter().map(|(id, config)| { +fn renditions(catalog: &hang::Catalog) -> impl Iterator { + let main = |id: &String, broadcast: bool, track: &Option| { + broadcast.then(|| track.clone().unwrap_or_else(|| id.clone())) + }; + let video = catalog.video.renditions.iter().map(move |(id, config)| { let size = match (config.coded_width, config.coded_height) { (Some(width), Some(height)) => format!(" {width}x{height}"), _ => String::new(), }; - (id, format!("video {}{size}", config.codec), config.enabled) + Rendition { + id: id.clone(), + label: format!("video {}{size}", config.codec), + enabled: config.enabled, + main: main(id, config.broadcast.is_some(), &config.track), + } }); - let audio = catalog.audio.renditions.iter().map(|(id, config)| { - let label = format!("audio {} {} ch", config.codec, config.channel_count); - (id, label, config.enabled) + let audio = catalog.audio.renditions.iter().map(move |(id, config)| Rendition { + id: id.clone(), + label: format!("audio {} {} ch", config.codec, config.channel_count), + enabled: config.enabled, + main: main(id, config.broadcast.is_some(), &config.track), }); video.chain(audio) } -fn viewers(publisher: &Publisher, graded: &[GradedViewer]) -> Vec { +fn viewers(graded: &[GradedViewer], clock: &moq_mux::Clock) -> Vec { graded .iter() .map(|viewer| { @@ -515,7 +788,7 @@ fn viewers(publisher: &Publisher, graded: &[GradedViewer]) -> Vec { snapshot .renditions .keys() - .filter_map(|id| viewer.latency(id, &publisher.clock)) + .filter_map(|id| viewer.latency(id, clock)) .max() }), playout: current.and_then(|snapshot| { @@ -531,16 +804,6 @@ fn viewers(publisher: &Publisher, graded: &[GradedViewer]) -> Vec { .collect() } -/// The grades of one interval. -#[derive(Default)] -struct Graded { - /// The program's tracks and the uplink. - publisher: Option, - /// The sources' own tracks. - sources: Option, - viewers: Vec, -} - /// One viewer's latest report and its grade. struct GradedViewer { name: String, @@ -573,14 +836,32 @@ struct Drops { struct Facts<'a> { relays: &'a [Relay], local: &'a Local, - drops: Option, + /// Each canvas's program frames over the last interval, once there was + /// one. + drops: BTreeMap, Drops>, uplink: &'a Uplink, - viewers: &'a [GradedViewer], - /// The tracks, for naming one a viewer reports on. - labels: &'a [Track], + /// What each canvas's viewers report, the main one first. + canvases: Vec>, thresholds: &'a Thresholds, } +/// One canvas's viewers, as graded. +struct Seen<'a> { + /// A further canvas's name, or `None` for the main one. + name: Option<&'a str>, + viewers: &'a [GradedViewer], + /// Its tracks, for naming one a viewer reports on. + tracks: &'a [Track], +} + +/// `cause` said of the canvas `canvas` names: a further one's names it. +fn on(canvas: Option<&str>, cause: String) -> String { + let Some(name) = canvas else { return cause }; + let mut chars = cause.chars(); + let first = chars.next().map(|first| first.to_lowercase().to_string()); + format!("On canvas `{name}`, {}{}", first.unwrap_or_default(), chars.as_str()) +} + /// Name each problem the facts show, worst first. fn problems(facts: &Facts<'_>) -> Vec { let mut problems = Vec::new(); @@ -614,8 +895,13 @@ fn problems(facts: &Facts<'_>) -> Vec { (State::Connected, _) => {} } } - if let Some(failed) = &local.failed { - add(Verdict::Unhealthy, format!("The output stopped: {failed}.")); + for (name, video) in &local.canvases { + if let Some(failed) = &video.failed { + add( + Verdict::Unhealthy, + on(name.as_deref(), format!("The output stopped: {failed}.")), + ); + } } for (name, stall) in &local.stalled { @@ -644,24 +930,33 @@ fn problems(facts: &Facts<'_>) -> Vec { } } - if let Some(drops) = facts.drops - && local.encoding - && local.failed.is_none() - { - let encoder = local.encoder.as_deref().unwrap_or("the encoder"); + for (name, drops) in &facts.drops { + let Some(video) = local.canvases.get(name) else { + continue; + }; + if !video.encoding || video.failed.is_some() { + continue; + } + let encoder = video.encoder.as_deref().unwrap_or("the encoder"); let share = ratio(drops.dropped, drops.drawn); let limits = &facts.thresholds.publisher.skipped; - if drops.drawn > 0 && drops.encoded == 0 && drops.dropped == 0 && local.encoder.is_some() { + if drops.drawn > 0 && drops.encoded == 0 && drops.dropped == 0 && video.encoder.is_some() { add( Verdict::Unhealthy, - format!("The encoder is stuck: {encoder} produced nothing in the last second."), + on( + name.as_deref(), + format!("The encoder is stuck: {encoder} produced nothing in the last second."), + ), ); } else if let Some(share) = share { add( grade_above(share, limits.degraded, limits.unhealthy), - format!( - "The encoder is behind: {encoder} had no room for {} of {} program frames in the last second.", - drops.dropped, drops.drawn + on( + name.as_deref(), + format!( + "The encoder is behind: {encoder} had no room for {} of {} program frames in the last second.", + drops.dropped, drops.drawn + ), ), ); } @@ -701,20 +996,22 @@ fn problems(facts: &Facts<'_>) -> Vec { ); } - viewer_problems(facts, &mut add); + for seen in &facts.canvases { + viewer_problems(uplink, seen, &mut |verdict, cause| add(verdict, on(seen.name, cause))); + } problems.sort_by_key(|problem| std::cmp::Reverse(problem.verdict)); problems } -/// What viewers' reports show, one problem per kind of trouble across them. -fn viewer_problems(facts: &Facts<'_>, add: &mut impl FnMut(Verdict, String)) { - let reporting = facts.viewers.iter().filter(|viewer| viewer.health.is_some()).count(); +/// What one canvas's viewers report, one problem per kind of trouble across +/// them, beside the `uplink` they share. +fn viewer_problems(uplink: &Uplink, seen: &Seen<'_>, add: &mut impl FnMut(Verdict, String)) { + let reporting = seen.viewers.iter().filter(|viewer| viewer.health.is_some()).count(); // A congested or lossy uplink starves every viewer, and one measured // clear of both leaves the blame with the viewers' own links. With either // unmeasured, as headroom is under a controller without an estimate, the // cause goes unnamed. - let uplink = facts.uplink; let (headroom, loss) = (uplink.grade(Input::Headroom), uplink.grade(Input::Loss)); let cause = if headroom >= Verdict::Degraded || loss >= Verdict::Degraded { ": the uplink cannot keep up" @@ -727,7 +1024,7 @@ fn viewer_problems(facts: &Facts<'_>, add: &mut impl FnMut(Verdict, String)) { // Each input's worst grade, how many viewers it troubles, and the track // the worst grade is on. let mut worst: Vec<(Input, Verdict, usize, Option<&str>)> = Vec::new(); - for viewer in facts.viewers { + for viewer in seen.viewers { let Some(health) = &viewer.health else { continue }; let transport = health.transport.evidence.iter().map(|evidence| (evidence, None)); let renditions = health.renditions.iter().flat_map(|(id, rendition)| { @@ -761,7 +1058,7 @@ fn viewer_problems(facts: &Facts<'_>, add: &mut impl FnMut(Verdict, String)) { } let track = |id: Option<&str>| { - id.and_then(|id| facts.labels.iter().find(|track| track.id == id)) + id.and_then(|id| seen.tracks.iter().find(|track| track.id == id)) .map_or_else(|| "the stream".to_string(), Track::name) }; for (input, verdict, count, id) in worst { @@ -1029,19 +1326,33 @@ fn watching(frame: &moq_stats::PrefixFrame) -> u64 { mod tests { use super::*; + /// What the main canvas alone shows: `viewers` reporting on `tracks`. fn facts<'a>(local: &'a Local, uplink: &'a Uplink, viewers: &'a [GradedViewer], tracks: &'a [Track]) -> Facts<'a> { static THRESHOLDS: std::sync::LazyLock = std::sync::LazyLock::new(Thresholds::default); Facts { relays: &[], local, - drops: None, + drops: BTreeMap::new(), uplink, - viewers, - labels: tracks, + canvases: vec![Seen { + name: None, + viewers, + tracks, + }], thresholds: &THRESHOLDS, } } + /// A studio whose main canvas's video is `main`, and the further one + /// `tall`'s, if any, `tall`. + fn local(main: Video, tall: Option