From afe35de2d66f1a7071a1f3828cd6fb4a2957644f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 5 Nov 2026 08:31:29 -0800 Subject: [PATCH 1/6] chore(quest): claim quest/m0/multi-canvas.md Quest-Claim: 903e184462767cb6 From d72fa1b9eb5b831175de740ee8ecc0a6c2e71c94 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 5 Nov 2026 09:08:56 -0800 Subject: [PATCH 2/6] wip: unlink canvases, preview only what is seen, record any canvas, grade each canvas Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/collection.rs | 72 ++- rs/moq-scene/src/compositor.rs | 768 +++++++++++++++++++++---------- rs/moq-scene/src/edit.rs | 69 ++- rs/moq-scene/src/lib.rs | 12 +- rs/moq-studio/src/engine.rs | 331 ++++++++------ rs/moq-studio/src/health.rs | 802 +++++++++++++++++++++++---------- rs/moq-studio/src/ui.rs | 411 +++++++++++------ rs/moq-studio/tests/browser.rs | 4 +- 8 files changed, 1706 insertions(+), 763 deletions(-) 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..2d23215 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,22 @@ 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::default(), + 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 = live.zip(health.as_ref()); + let graded = graded.map(|(live, health)| (health.canvases.clone(), live.reports.clone())); Ok(( Program { @@ -1523,9 +1558,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 +1592,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 +1625,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 +1670,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 +1774,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 +1799,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 +1884,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 +1944,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 +2011,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 +2059,9 @@ struct Extras { on_demand: bool, live: bool, stats: Arc, + /// Where each canvas's broadcast is graded, and the origin its viewers' + /// reports and the relays' stats arrive on, while live. + graded: Option<(health::Canvases, moq_net::origin::Consumer)>, } impl Extras { @@ -2020,21 +2091,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 +2137,28 @@ 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, reports)) => { + let observer = hang::health::Observer::new(path.clone(), Some(instance)); + let broadcast = health::Broadcast::follow(observer, catalog.clone(), reports)?; + Some(canvases.insert(name, broadcast)) + } + 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 +2166,11 @@ impl Extras { video, grants: format!("{name}/"), }; - let broadcast = Broadcast { publication, catalog }; + let broadcast = Broadcast { + publication, + catalog, + _graded: graded, + }; Ok((renditions, broadcast)) } } @@ -2258,13 +2333,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 +2699,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) } @@ -3258,9 +3326,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..d405cb9 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,168 @@ 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, Default)] +pub struct Canvases(Arc>>); + +impl Canvases { + /// Grade `broadcast` as canvas `name`'s until the returned handle drops. + pub fn insert(&self, name: &str, broadcast: Broadcast) -> Graded { + self.0.lock().unwrap().insert(name.to_string(), broadcast); + Graded { + canvases: self.clone(), + name: name.to_string(), + } + } + + fn snapshot(&self) -> Vec<(String, Broadcast)> { + let canvases = self.0.lock().unwrap(); + canvases + .iter() + .map(|(name, broadcast)| (name.clone(), broadcast.clone())) + .collect() + } +} + +/// A further canvas's broadcast being graded, until dropped. +pub struct Graded { + canvases: Canvases, + name: String, +} + +impl Drop for Graded { + fn drop(&mut self) { + self.canvases.0.lock().unwrap().remove(&self.name); + } +} + /// The connection to one relay. #[derive(Clone)] pub struct Link { @@ -277,16 +416,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 +440,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 +562,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,39 +626,41 @@ 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<(String, String, bool)>, + /// 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); +/// 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(|(id, label, enabled)| { + let graded = format!("{}{id}", listed.prefix); + 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 = source.is_none(); + let program = listed.source.is_none(); let mut track = Track { - source: source.cloned(), + source: listed.source.cloned(), id: id.clone(), - label, - granted: listed.grants.get(id).copied().flatten().filter(|_| program), - enabled, + 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) @@ -467,7 +671,7 @@ fn tracks(listed: &Listed, clock: &moq_mux::Clock, graded: &Graded) -> Vec Vec(catalog: &hang::Catalog) -> impl Iterator { +fn renditions(catalog: &hang::Catalog) -> impl Iterator { let video = catalog.video.renditions.iter().map(|(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) + (id.clone(), format!("video {}{size}", config.codec), config.enabled) }); let audio = catalog.audio.renditions.iter().map(|(id, config)| { let label = format!("audio {} {} ch", config.codec, config.channel_count); - (id, label, config.enabled) + (id.clone(), label, config.enabled) }); 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 +719,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 +735,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 +767,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 +826,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 +861,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 +927,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 +955,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 +989,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 +1257,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