From 8bab151e04fa1b83bbcf6be8f6bdae63959d18a0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 12:14:29 -0800 Subject: [PATCH 01/10] chore(quest): claim quest/m0/multi-canvas.md Quest-Claim: 3363157e5759aa67 From 0fb488f0b03de32d7adc3dc8a4423b7fe16fc865 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 13:15:37 -0800 Subject: [PATCH 02/10] feat(scene): further canvases with their own layouts and top layers, drawn from one upload of each source Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/clip.rs | 14 +- rs/moq-scene/src/collection.rs | 229 ++++++++++++++--- rs/moq-scene/src/compositor.rs | 458 +++++++++++++++++++++++---------- rs/moq-scene/src/edit.rs | 144 ++++++++++- rs/moq-scene/src/effect.rs | 63 ++--- rs/moq-scene/src/lib.rs | 15 +- rs/moq-studio/src/engine.rs | 2 +- rs/moq-studio/src/interact.rs | 56 ++-- rs/moq-studio/src/ui.rs | 105 ++++++-- 9 files changed, 817 insertions(+), 269 deletions(-) diff --git a/rs/moq-scene/src/clip.rs b/rs/moq-scene/src/clip.rs index 3c6398d9..d2836b17 100644 --- a/rs/moq-scene/src/clip.rs +++ b/rs/moq-scene/src/clip.rs @@ -288,7 +288,7 @@ mod tests { fn pastes_linked_within_a_collection() { let collection = collection(); let clip = Clip::items(&collection, collection.scenes[0].items.clone(), Path::new(DIR)); - let pasted = paste(&collection, &Layer::Top, &clip.to_toml().unwrap(), DIR); + let pasted = paste(&collection, &Layer::top(), &clip.to_toml().unwrap(), DIR); assert_eq!(pasted.sources, collection.sources, "no source added"); assert_eq!(pasted.top, collection.scenes[0].items); } @@ -304,7 +304,7 @@ mod tests { let mut other = Collection::new(); other.sources.insert("cam".into(), color([0, 255, 0, 255])); - let pasted = paste(&other, &Layer::Scene("Scene".into()), &text, DIR); + let pasted = paste(&other, &Layer::scene("Scene"), &text, DIR); assert_eq!(pasted.sources["cam"], color([0, 255, 0, 255]), "left alone"); assert_eq!(pasted.sources["cam 2"], Source::Camera { id: None }); assert_eq!(pasted.sources["cam faded"], Source::Clone { source: "cam 2".into() }); @@ -338,7 +338,7 @@ mod tests { .to_toml() .unwrap(); - let pasted = paste(&Collection::new(), &Layer::Top, &text, DIR); + let pasted = paste(&Collection::new(), &Layer::top(), &text, DIR); let expected = crate::Audio { exclude: ["Program".to_string()].into(), ..audio @@ -364,7 +364,7 @@ mod tests { .to_toml() .unwrap(); - let home = paste(&collection, &Layer::Top, &text, DIR); + let home = paste(&collection, &Layer::top(), &text, DIR); assert_eq!(home.sources, collection.sources, "linked to the same image"); // However the scene file spells the path. @@ -375,7 +375,7 @@ mod tests { path: "/shows/one/logo.png".into(), }, ); - let home = paste(&absolute, &Layer::Top, &text, DIR); + let home = paste(&absolute, &Layer::top(), &text, DIR); assert_eq!(home.sources, absolute.sources, "linked to the same image"); let mut other = Collection::new(); @@ -385,7 +385,7 @@ mod tests { path: "logo.png".into(), }, ); - let away = paste(&other, &Layer::Top, &text, "/shows/two"); + let away = paste(&other, &Layer::top(), &text, "/shows/two"); assert_eq!(away.sources["logo"], other.sources["logo"], "another file, left alone"); assert_eq!( away.sources["logo 2"], @@ -433,7 +433,7 @@ mod tests { let mut other = Collection::new(); other.sources.insert("matte".into(), color([0, 255, 0, 255])); - let pasted = paste(&other, &Layer::Top, &text, "/shows/two"); + let pasted = paste(&other, &Layer::top(), &text, "/shows/two"); assert_eq!(pasted.top[0].filters, [mask("matte 2")]); assert_eq!( pasted.sources["grade"], diff --git a/rs/moq-scene/src/collection.rs b/rs/moq-scene/src/collection.rs index 9c742264..93776ba9 100644 --- a/rs/moq-scene/src/collection.rs +++ b/rs/moq-scene/src/collection.rs @@ -11,8 +11,8 @@ use serde_with::{DeserializeFromStr, SerializeDisplay}; use crate::Error; -/// Everything a studio composes: the canvas, the sources, and the scenes that -/// place them. +/// Everything a studio composes: the canvases, the sources, and the scenes +/// that place them. /// /// Saved with [`to_toml`](Self::to_toml) and loaded with /// [`from_toml`](Self::from_toml). The file names devices, files, and broadcast @@ -20,8 +20,14 @@ use crate::Error; #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct Collection { - /// The output every scene is drawn onto. + /// The main canvas every scene is drawn onto, whose frame rate every + /// canvas shares. pub canvas: Canvas, + /// Further canvases by name, such as a portrait one beside a landscape + /// main canvas. Each draws its own layout of every scene from the same + /// 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. #[serde(default)] pub transition: Transition, @@ -61,6 +67,7 @@ impl Collection { pub fn new() -> Self { Self { canvas: Canvas::default(), + canvases: BTreeMap::new(), transition: Transition::default(), sources: BTreeMap::new(), mixes: Mixes::default(), @@ -91,33 +98,79 @@ impl Collection { self.scenes.iter().find(|scene| scene.name == name) } - /// The items of `layer`, or `None` for a scene that does not exist. - pub fn items(&self, layer: &Layer) -> Option<&Vec> { - match layer { - Layer::Scene(name) => self.scene(name).map(|scene| &scene.items), - Layer::Top => Some(&self.top), + /// 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. + pub fn canvas(&self, canvas: Option<&str>) -> Option { + match canvas { + None => Some(self.canvas), + Some(name) => self.canvases.get(name).map(|extra| extra.canvas(self.canvas.framerate)), } } - /// The items of `layer`, to edit. + /// The items of `layer`, or `None` for a scene or canvas that does not + /// exist. A scene with no layout on a further canvas has none there. + pub fn items(&self, layer: &Layer) -> Option<&[Item]> { + match (&layer.canvas, &layer.scene) { + (None, Some(scene)) => self.scene(scene).map(|scene| scene.items.as_slice()), + (None, None) => Some(&self.top), + (Some(canvas), scene) => { + let extra = self.canvases.get(canvas)?; + match scene { + Some(scene) => { + let layout = self.scene(scene)?.canvases.get(canvas); + Some(layout.map_or(&[], Vec::as_slice)) + } + None => Some(&extra.top), + } + } + } + } + + /// The items of `layer`, to edit. A scene's layout on a further canvas is + /// added empty, and [`prune`](Self::prune) drops it once empty again. pub(crate) fn items_mut(&mut self, layer: &Layer) -> Option<&mut Vec> { - match layer { - Layer::Scene(name) => self - .scenes - .iter_mut() - .find(|scene| &scene.name == name) - .map(|scene| &mut scene.items), - Layer::Top => Some(&mut self.top), + if let Some(canvas) = &layer.canvas + && !self.canvases.contains_key(canvas) + { + return None; + } + match (&layer.canvas, &layer.scene) { + (None, None) => Some(&mut self.top), + (Some(canvas), None) => self.canvases.get_mut(canvas).map(|extra| &mut extra.top), + (canvas, Some(name)) => { + let scene = self.scenes.iter_mut().find(|scene| &scene.name == name)?; + Some(match canvas { + None => &mut scene.items, + Some(canvas) => scene.canvases.entry(canvas.clone()).or_default(), + }) + } } } - /// Every layer: the scenes in order, then the top layer. + /// Drop the layouts on further canvases that hold no items, so a scene + /// reads the same however its layouts were emptied. + pub(crate) fn prune(&mut self) { + for scene in &mut self.scenes { + scene.canvases.retain(|_, items| !items.is_empty()); + } + } + + /// Every layer: the main canvas's scenes in order, then its top layer, + /// then each further canvas's the same way. pub fn layers(&self) -> impl Iterator { - let scenes = self - .scenes - .iter() - .map(|scene| (Layer::Scene(scene.name.clone()), scene.items.as_slice())); - scenes.chain(std::iter::once((Layer::Top, self.top.as_slice()))) + let canvases = std::iter::once(None).chain(self.canvases.keys().cloned().map(Some)); + canvases.flat_map(move |canvas| { + let scenes = self.scenes.iter().map(|scene| Some(scene.name.clone())); + scenes.chain([None]).map(move |scene| { + let layer = Layer { + canvas: canvas.clone(), + scene, + }; + let items = self.items(&layer).unwrap_or_default(); + (layer, items) + }) + }) } /// The source whose capture `name` shows: itself, or what it clones. @@ -147,6 +200,17 @@ impl Collection { /// source of the kind it reads. pub fn validate(&self) -> Result<(), Error> { self.canvas.validate()?; + for (name, extra) in &self.canvases { + if name.is_empty() || name.contains('/') { + return Err(Error::Invalid(format!( + "canvas `{name}` must be named, without a `/`, since its name names its broadcast" + ))); + } + extra + .canvas(self.canvas.framerate) + .validate() + .map_err(|err| Error::Invalid(format!("canvas `{name}`: {err}")))?; + } let mut names = BTreeSet::new(); for scene in &self.scenes { @@ -156,6 +220,16 @@ impl Collection { if !names.insert(scene.name.as_str()) { return Err(Error::Invalid(format!("two scenes are named `{}`", scene.name))); } + if let Some(canvas) = scene + .canvases + .keys() + .find(|canvas| !self.canvases.contains_key(*canvas)) + { + return Err(Error::Invalid(format!( + "scene `{}` has a layout on `{canvas}`, which is not a canvas", + scene.name + ))); + } } for (layer, items) in self.layers() { for item in items { @@ -346,6 +420,41 @@ impl Default for Canvas { } } +/// A canvas beside the main one: its size, and the items drawn over every +/// scene on it. Each scene's layout on it is the scene's own +/// [`Scene::canvases`] entry. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Extra { + /// Width in pixels. Even, since the encoder subsamples chroma. + pub width: u32, + /// Height in pixels. Even, since the encoder subsamples chroma. + pub height: u32, + /// Its top layer, as [`Collection::top`] is the main canvas's. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub top: Vec, +} + +impl Extra { + /// An empty canvas of `width` by `height`. + pub fn new(width: u32, height: u32) -> Self { + Self { + width, + height, + top: Vec::new(), + } + } + + /// This canvas drawn `framerate` times a second. + pub fn canvas(&self, framerate: u32) -> Canvas { + Canvas { + width: self.width, + height: self.height, + framerate, + } + } +} + /// How the program changes when the preview is taken to it, short of a cut. /// /// A fade runs at an even pace; the others ease in and out, so motion starts @@ -454,9 +563,13 @@ impl Direction { pub struct Scene { /// Unique among the collection's scenes. pub name: String, - /// What the scene draws, back to front. + /// What the scene draws on the main canvas, back to front. #[serde(default)] pub items: Vec, + /// What the scene draws on each further canvas, by the canvas's name, back + /// to front. A canvas without an entry shows only its top layer. + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + pub canvases: BTreeMap>, } impl Scene { @@ -465,25 +578,53 @@ impl Scene { Self { name: name.into(), items: Vec::new(), + canvases: BTreeMap::new(), } } } -/// Where a list of items lives: a scene, or the top layer over every scene. -#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum Layer { - /// The scene with this name. - Scene(String), - /// [`Collection::top`]. - Top, +/// Where a list of items lives: a scene's layout, or the top layer over every +/// scene, on one canvas. +#[derive(Clone, Debug, Default, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Layer { + /// A further canvas by name, or `None` for the main canvas. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub canvas: Option, + /// The scene, or `None` for the canvas's top layer. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub scene: Option, +} + +impl Layer { + /// The scene called `name`, on the main canvas. + pub fn scene(name: impl Into) -> Self { + Self { + canvas: None, + scene: Some(name.into()), + } + } + + /// The main canvas's top layer. + pub fn top() -> Self { + Self::default() + } + + /// The same layer on `canvas` instead. + pub fn on(self, canvas: Option) -> Self { + Self { canvas, ..self } + } } impl fmt::Display for Layer { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - match self { - Self::Scene(name) => write!(f, "scene `{name}`"), - Self::Top => write!(f, "the top layer"), + match &self.scene { + Some(name) => write!(f, "scene `{name}`")?, + None => write!(f, "the top layer")?, + } + match &self.canvas { + Some(canvas) => write!(f, " on canvas `{canvas}`"), + None => Ok(()), } } } @@ -1722,6 +1863,26 @@ rect = { x = 0, y = 680, width = 1280, height = 40 } assert!(matches!(collection.validate(), Err(Error::Invalid(_)))); } + /// A further canvas is named as a path segment, since it names its + /// broadcast, is a size the encoder can take, and is the only canvas a + /// scene lays items out on. + #[test] + fn refuses_a_further_canvas_that_cannot_be() { + for (name, width) in [("", 1080), ("a/b", 1080), ("tall", 1081)] { + let mut collection = example(); + collection.canvases.insert(name.into(), Extra::new(width, 1920)); + assert!( + matches!(collection.validate(), Err(Error::Invalid(_))), + "{name} {width}" + ); + } + let mut collection = example(); + collection.scenes[0].canvases.insert("tall".into(), Vec::new()); + assert!(matches!(collection.validate(), Err(Error::Invalid(_)))); + collection.canvases.insert("tall".into(), Extra::new(1080, 1920)); + collection.validate().unwrap(); + } + #[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 492c612b..00926499 100644 --- a/rs/moq-scene/src/compositor.rs +++ b/rs/moq-scene/src/compositor.rs @@ -1,4 +1,4 @@ -//! Draw scenes onto the canvas on the GPU: a preview, a program, and the +//! Draw scenes onto each canvas on the GPU: a preview, a program, and the //! transition between them. use std::collections::{BTreeMap, HashMap, HashSet}; @@ -16,29 +16,45 @@ use crate::source::{self, Sources}; /// `moq_video::render` produces and what its exporter hands an encoder. pub const FORMAT: wgpu::TextureFormat = wgpu::TextureFormat::Rgba8Unorm; -/// Draws a [`Collection`]'s scenes onto its canvas: one texture for the scene -/// in preview, one for the scene on program, and the [`Transition`] from one -/// to the other. +/// Draws a [`Collection`]'s scenes onto each of its canvases: one texture for +/// the scene in preview, one for the scene on program, and the [`Transition`] +/// from one to the other. /// -/// Both textures are plain [`wgpu::Texture`]s on the caller's device, so a UI +/// The textures are plain [`wgpu::Texture`]s on the caller's device, so a UI /// shows them without a copy and an encoder reads the program from the same -/// memory. [`render`](Self::render) redraws both from the newest picture each +/// 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. +/// 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. /// /// 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. /// -/// A transition runs on the clock [`render`](Self::render) is given, so every -/// output drawn from the program, such as the encoder and a recording, sees the -/// same frame of it. +/// The canvases are linked: one scene is in preview and one on program on +/// every canvas, and a transition runs 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 it. pub struct Compositor { device: wgpu::Device, queue: wgpu::Queue, - canvas: Canvas, effects: Effects, + main: Output, + extras: BTreeMap, + + layers: HashMap, + /// Lays out and paints overlays, started with the first one. + overlays: Option, + preview_scene: Option, + program_scene: Option, + switch: Option, +} + +/// One canvas's preview and program, as of the last +/// [`render`](Compositor::render). +pub struct Output { + canvas: Canvas, preview: Target, program: Target, /// The scene coming in, drawn whole so it fades, slides, or is uncovered @@ -46,13 +62,36 @@ pub struct Compositor { incoming: Target, /// The scene going out, drawn whole to slide it off the canvas. outgoing: Target, +} - layers: HashMap, - /// Lays out and paints overlays, started with the first one. - overlays: Option, - preview_scene: Option, - program_scene: Option, - switch: Option, +impl Output { + fn new(device: &wgpu::Device, effects: &Effects, canvas: Canvas) -> Self { + let target = |label| Target::new(device, effects, canvas, label); + Self { + canvas, + preview: target("moq-scene preview"), + program: target("moq-scene program"), + incoming: target("moq-scene incoming"), + outgoing: target("moq-scene outgoing"), + } + } + + /// The canvas this draws. + pub fn canvas(&self) -> Canvas { + self.canvas + } + + /// The scene in preview. + pub fn preview(&self) -> &wgpu::Texture { + &self.preview.texture + } + + /// The program. Created with `COPY_SRC | TEXTURE_BINDING`, what + /// `moq_video::render::export::Exporter::copy` takes to hand it to an + /// encoder. + pub fn program(&self) -> &wgpu::Texture { + &self.program.texture + } } /// A transition under way from `from` to the program scene. @@ -99,23 +138,20 @@ impl effect::Library for Library<'_> { } impl Compositor { - /// A compositor drawing `canvas`-sized pictures on `device`. + /// A compositor drawing a main canvas of `canvas` on `device`, 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. pub fn new(device: &wgpu::Device, queue: &wgpu::Queue, canvas: Canvas) -> Result { fits(device, canvas)?; - let effects = Effects::new(device, queue, canvas); - let target = |label| Target::new(device, &effects, canvas, label); + let effects = Effects::new(device, queue, canvas.width, canvas.height); Ok(Self { - preview: target("moq-scene preview"), - program: target("moq-scene program"), - incoming: target("moq-scene incoming"), - outgoing: target("moq-scene outgoing"), + main: Output::new(device, &effects, canvas), + extras: BTreeMap::new(), device: device.clone(), queue: queue.clone(), - canvas, effects, layers: HashMap::new(), overlays: None, @@ -125,42 +161,80 @@ impl Compositor { }) } - /// The canvas this compositor draws. + /// The main canvas. pub fn canvas(&self) -> Canvas { - self.canvas + self.main.canvas } - /// Draw `canvas` from the next [`render`](Self::render) on. A new size - /// replaces the [`preview`](Self::preview) and [`program`](Self::program) - /// textures, and every other texture sized from the canvas, with ones of - /// that size, blank until drawn; a transition under way carries on. + /// Draw `collection`'s canvases from the next [`render`](Self::render) on: + /// its main canvas and each further one. A canvas of a new size gets new + /// textures of that size, blank until drawn; a new canvas starts blank, + /// and one `collection` no longer has is dropped. A transition under way + /// carries on. /// - /// Fails, changing nothing, if the canvas is invalid or the device cannot + /// Fails, changing nothing, if a canvas is invalid or the device cannot /// hold a texture that large. - pub fn set_canvas(&mut self, canvas: Canvas) -> Result<(), Error> { - fits(&self.device, canvas)?; - self.effects.set_canvas(canvas); - if canvas.size() != self.canvas.size() { - let target = |label| Target::new(&self.device, &self.effects, canvas, label); - self.preview = target("moq-scene preview"); - self.program = target("moq-scene program"); - self.incoming = target("moq-scene incoming"); - self.outgoing = target("moq-scene outgoing"); + pub fn set_canvases(&mut self, collection: &Collection) -> Result<(), Error> { + let canvas = collection.canvas; + let extras: BTreeMap<&str, Canvas> = collection + .canvases + .iter() + .map(|(name, extra)| (name.as_str(), extra.canvas(canvas.framerate))) + .collect(); + for canvas in std::iter::once(&canvas).chain(extras.values()) { + 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); + 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); + } + }; + 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)); + } + } } - self.canvas = canvas; Ok(()) } - /// The scene in preview, as of the last [`render`](Self::render). + /// The scene in preview on the main canvas, as of the last + /// [`render`](Self::render). pub fn preview(&self) -> &wgpu::Texture { - &self.preview.texture + self.main.preview() } - /// The program output, as of the last [`render`](Self::render). Created - /// with `COPY_SRC | TEXTURE_BINDING`, what - /// `moq_video::render::export::Exporter::copy` takes to hand it to an encoder. + /// The main canvas's program, as of the last [`render`](Self::render). See + /// [`Output::program`]. pub fn program(&self) -> &wgpu::Texture { - &self.program.texture + self.main.program() + } + + /// The canvas `canvas` names, the main one for `None`, as of the last + /// [`render`](Self::render); `None` for a canvas it does not draw. + pub fn output(&self, canvas: Option<&str>) -> Option<&Output> { + match canvas { + None => Some(&self.main), + Some(name) => self.extras.get(name), + } + } + + /// Every canvas drawn, by name: the main one first, as `None`. + pub fn outputs(&self) -> impl Iterator, &Output)> { + let extras = self.extras.iter().map(|(name, output)| (Some(name.as_str()), output)); + std::iter::once((None, &self.main)).chain(extras) } /// The scene in preview. @@ -247,11 +321,11 @@ impl Compositor { }); } - /// Draw preview and program as of `now`, from the newest picture each of - /// `sources` has. + /// Draw preview and program on every canvas as of `now`, from the newest + /// picture each of `sources` has. /// /// A source whose picture fails to draw keeps its last one and reports why - /// through [`error`](Self::error); this fails only when the canvas itself + /// through [`error`](Self::error); this fails only when a canvas itself /// cannot be drawn. pub fn render(&mut self, collection: &Collection, sources: &Sources, now: Instant) -> Result<(), Error> { self.update(collection, sources, now); @@ -271,60 +345,19 @@ impl Compositor { effects: &mut self.effects, library: Library(&self.layers), frame: &mut frame, + canvas: None, + size: self.main.canvas, prepared: HashMap::new(), }; - - // The top layer goes over both outputs, and over a transition rather - // than through it, so it stays on across the switch. - draw.target(&self.preview); - draw.scene(&self.preview, self.preview_scene.as_deref()); - draw.layer(&self.preview, Place::Top); - - let to = self.program_scene.as_deref(); - match (&self.switch, linear) { - (Some(switch), Some(linear)) => { - let from = switch.from.as_deref(); - let progress = switch.transition.progress(linear); - let canvas = Rect::fill(&self.canvas); - match switch.transition { - Transition::Fade { .. } => { - draw.whole(&self.incoming, to); - draw.target(&self.program); - draw.scene(&self.program, from); - draw.place(&self.program, &self.incoming, canvas, canvas, progress); - } - // Uncovered where a slide would have moved it in. - Transition::Wipe { direction, .. } => { - draw.whole(&self.incoming, to); - draw.target(&self.program); - draw.scene(&self.program, from); - let (x, y) = offset(direction, progress, &canvas); - let (dx, dy) = offset(direction, 1.0, &canvas); - let uncovered = shifted(canvas, x - dx, y - dy); - draw.place(&self.program, &self.incoming, canvas, uncovered, 1.0); - } - Transition::Slide { direction, .. } => { - draw.whole(&self.incoming, to); - draw.whole(&self.outgoing, from); - draw.target(&self.program); - let (x, y) = offset(direction, progress, &canvas); - let (dx, dy) = offset(direction, 1.0, &canvas); - draw.place(&self.program, &self.outgoing, shifted(canvas, x, y), canvas, 1.0); - let entering = shifted(canvas, x - dx, y - dy); - draw.place(&self.program, &self.incoming, entering, canvas, 1.0); - } - Transition::Move { .. } => { - draw.target(&self.program); - draw.moving(&self.program, from, to, progress); - } - } - } - _ => { - draw.target(&self.program); - draw.scene(&self.program, to); - } + let preview = self.preview_scene.as_deref(); + let program = self.program_scene.as_deref(); + let switch = self.switch.as_ref().zip(linear); + 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); } - draw.layer(&self.program, Place::Top); self.effects.submit(frame); Ok(()) @@ -448,26 +481,31 @@ impl Compositor { } } - /// The sources whose pictures preview and program draw: each visible - /// item's, and each mask's, through clones. + /// The sources whose pictures preview and program draw on any canvas: + /// each visible item's, and each mask's, through clones. 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 = [self.preview_scene.as_deref(), self.program_scene.as_deref(), leaving] + let scenes: Vec> = [self.preview_scene.as_deref(), self.program_scene.as_deref(), leaving] .into_iter() .flatten() - .map(|name| Place::Scene(name.to_string())); + .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 place in scenes.chain([Place::Top]) { - for item in collection - .items(&place) - .into_iter() - .flatten() - .filter(|item| item.visible) - { - shown.insert(collection.capture(&item.source)); - for filter in collection.filters(item) { - if let Filter::Mask { source, .. } = filter { - shown.insert(collection.capture(source)); + for canvas in canvases { + for scene in &scenes { + let place = Place { + canvas: canvas.cloned(), + scene: scene.clone(), + }; + let items = collection.items(&place).unwrap_or_default(); + for item in items.iter().filter(|item| item.visible) { + shown.insert(collection.capture(&item.source)); + for filter in collection.filters(item) { + if let Filter::Mask { source, .. } = filter { + shown.insert(collection.capture(source)); + } } } } @@ -603,20 +641,95 @@ struct Draw<'a> { effects: &'a mut Effects, library: Library<'a>, frame: &'a mut Frame, + /// The canvas drawn now: a further one by name, or `None` for the main one. + canvas: Option<&'a str>, + /// Its size and frame rate. + size: Canvas, /// Each item, filtered once though it is drawn onto several targets. prepared: HashMap<(Place, usize), Option>, } impl Draw<'_> { + /// Draw `output`'s preview, showing `preview`, and its program, showing + /// `program`, or `switch` to it, `linear` of its time along. + fn output( + &mut self, + output: &Output, + preview: Option<&str>, + program: Option<&str>, + switch: Option<(&Switch, f32)>, + ) { + 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()); + + let to = program; + match switch { + Some((switch, linear)) => { + let from = switch.from.as_deref(); + let progress = switch.transition.progress(linear); + let canvas = Rect::fill(&self.size); + match switch.transition { + Transition::Fade { .. } => { + self.whole(&output.incoming, to); + self.target(&output.program); + self.scene(&output.program, from); + self.picture(&output.program, &output.incoming, canvas, canvas, progress); + } + // Uncovered where a slide would have moved it in. + Transition::Wipe { direction, .. } => { + self.whole(&output.incoming, to); + self.target(&output.program); + self.scene(&output.program, from); + let (x, y) = offset(direction, progress, &canvas); + let (dx, dy) = offset(direction, 1.0, &canvas); + let uncovered = shifted(canvas, x - dx, y - dy); + self.picture(&output.program, &output.incoming, canvas, uncovered, 1.0); + } + Transition::Slide { direction, .. } => { + self.whole(&output.incoming, to); + self.whole(&output.outgoing, from); + self.target(&output.program); + let (x, y) = offset(direction, progress, &canvas); + let (dx, dy) = offset(direction, 1.0, &canvas); + self.picture(&output.program, &output.outgoing, shifted(canvas, x, y), canvas, 1.0); + let entering = shifted(canvas, x - dx, y - dy); + self.picture(&output.program, &output.incoming, entering, canvas, 1.0); + } + Transition::Move { .. } => { + self.target(&output.program); + self.moving(&output.program, from, to, progress); + } + } + } + None => { + self.target(&output.program); + self.scene(&output.program, to); + } + } + self.layer(&output.program, top); + } + + /// `scene`'s layout on the canvas drawn now, or its top layer for `None`. + fn place(&self, scene: Option<&str>) -> Place { + Place { + canvas: self.canvas.map(str::to_string), + scene: scene.map(str::to_string), + } + } + /// Start drawing `target`. fn target(&mut self, target: &Target) { - self.frame.target(&target.picture); + self.frame.target(&target.picture, self.size); } /// Draw `scene`'s items onto `target`. A missing scene draws nothing. fn scene(&mut self, target: &Target, scene: Option<&str>) { if let Some(scene) = scene.filter(|name| self.collection.scene(name).is_some()) { - self.layer(target, Place::Scene(scene.to_string())); + self.layer(target, self.place(Some(scene))); } } @@ -628,8 +741,8 @@ impl Draw<'_> { /// Draw `picture`, a canvas-sized target drawn this render, onto `target` /// stretched over `at`, showing only within `within`, at `opacity`. - fn place(&mut self, target: &Target, picture: &Target, at: Rect, within: Rect, opacity: f32) { - if let Some(placed) = self.effects.place(&picture.picture, at, within, opacity) { + fn picture(&mut self, target: &Target, picture: &Target, at: Rect, within: Rect, opacity: f32) { + if let Some(placed) = self.effects.place(self.frame, &picture.picture, at, within, opacity) { self.effects.draw(self.frame, &target.texture, &placed); } } @@ -638,7 +751,7 @@ impl Draw<'_> { /// filters, and blend mode. A hidden item, a source with no picture yet, or /// a filter whose source has nothing yet draws nothing. fn layer(&mut self, target: &Target, place: Place) { - let items = self.collection.items(&place).map_or(&[][..], Vec::as_slice); + let items = self.collection.items(&place).unwrap_or_default(); for (index, item) in items.iter().enumerate().filter(|(_, item)| item.visible) { self.item(target, &place, index, item); } @@ -684,9 +797,9 @@ impl Draw<'_> { /// whose blend mode or kinds of filter differ changes look halfway too. fn moving(&mut self, target: &Target, from: Option<&str>, to: Option<&str>, progress: f32) { let collection = self.collection; - let items = |scene: Option<&str>| { - let scene = scene.and_then(|name| collection.scene(name)); - scene.map_or(&[][..], |scene| scene.items.as_slice()) + let items = |scene: Option<&str>| match scene { + Some(scene) => collection.items(&self.place(Some(scene))).unwrap_or_default(), + None => &[], }; let (leaving, entering) = (items(from), items(to)); @@ -710,7 +823,7 @@ impl Draw<'_> { slots.into_iter().map(|(to, from)| (from, to)).collect() }; - let place = Place::Scene(to.unwrap_or_default().to_string()); + let place = self.place(Some(to.unwrap_or_default())); for slot in slots { match slot { (Some(at), None) => { @@ -848,7 +961,7 @@ mod tests { use std::time::Duration; use super::*; - use crate::collection::{Blend, MaskMode, Rgba, Scene}; + use crate::collection::{Blend, Extra, MaskMode, Rgba, Scene}; use crate::source::test::{context, local, start}; /// A fade over a second. @@ -898,12 +1011,12 @@ mod tests { }; collection.scenes = vec![ Scene { - name: "red".into(), items: vec![item("red", fill), item("blue", corner)], + ..Scene::new("red") }, Scene { - name: "blue".into(), items: vec![item("blue", fill)], + ..Scene::new("blue") }, ]; collection.validate().unwrap(); @@ -952,11 +1065,11 @@ mod tests { /// The canvas texture's exact RGBA rows: no 4:2:0 round trip, so a test /// compares the stored values. fn read(&mut self, texture: &wgpu::Texture) -> Vec { - let canvas = self.compositor.canvas(); - let row = (canvas.width * 4).next_multiple_of(wgpu::COPY_BYTES_PER_ROW_ALIGNMENT); + let (width, height) = (texture.width(), texture.height()); + let row = (width * 4).next_multiple_of(wgpu::COPY_BYTES_PER_ROW_ALIGNMENT); let buffer = self.device.create_buffer(&wgpu::BufferDescriptor { label: None, - size: u64::from(row * canvas.height), + size: u64::from(row * height), usage: wgpu::BufferUsages::COPY_DST | wgpu::BufferUsages::MAP_READ, mapped_at_creation: false, }); @@ -976,7 +1089,7 @@ mod tests { self.queue.submit([encoder.finish()]); buffer.map_async(wgpu::MapMode::Read, .., |result| result.unwrap()); self.device.poll(wgpu::PollType::wait_indefinitely()).unwrap(); - let width = canvas.width as usize * 4; + let width = width as usize * 4; let mapped = buffer.slice(..).get_mapped_range().unwrap(); mapped .chunks_exact(row as usize) @@ -1152,6 +1265,66 @@ mod tests { .await; } + /// A further canvas draws its own layout of each scene and its own top + /// layer, from the same upload of each source, and cuts and fades with + /// the main canvas. A scene without a layout there shows only the top + /// layer, and a canvas the collection drops is no longer drawn. + #[tokio::test] + #[ignore = "needs a Vulkan device; run by just rs render-sw"] + async fn a_further_canvas_draws_its_own_layout() { + local(async { + let mut collection = collection(); + let mut tall = Extra::new(4, 8); + let bottom = Rect { + x: 0.0, + y: 4.0, + width: 4.0, + height: 4.0, + }; + tall.top.push(Item::new("red", bottom)); + collection.canvases.insert("tall".into(), tall); + let top = Rect { y: 0.0, ..bottom }; + collection.scenes[0] + .canvases + .insert("tall".into(), vec![Item::new("blue", top)]); + collection.validate().unwrap(); + let mut studio = Studio::new(collection).await; + studio.compositor.set_canvases(&studio.collection).unwrap(); + let tall = |studio: &mut Studio| { + let texture = studio.compositor.output(Some("tall")).unwrap().program().clone(); + let rgba = studio.read(&texture); + let pixel = |x: usize, y: usize| -> [u8; 4] { rgba[(y * 4 + x) * 4..][..4].try_into().unwrap() }; + (pixel(1, 1), pixel(1, 6)) + }; + + let start = Instant::now(); + studio.compositor.set_preview(Some("red".into())); + studio.compositor.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.render(start + Duration::from_millis(500)); + let (fading, kept) = tall(&mut studio); + assert!(near(fading, [0, 0, 128, 255]), "{fading:?}"); + assert_eq!(kept, RED, "the top layer does not fade"); + + studio.render(start + Duration::from_secs(1)); + assert_eq!(pixel(&studio.program(), 1, 1), BLUE); + assert_eq!(tall(&mut studio), (BLACK, RED), "no layout leaves the top layer"); + + studio.collection.canvases.clear(); + studio.collection.scenes[0].canvases.clear(); + studio.compositor.set_canvases(&studio.collection).unwrap(); + assert!(studio.compositor.output(Some("tall")).is_none()); + assert_eq!(studio.compositor.outputs().count(), 1); + }) + .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"] @@ -1784,9 +1957,9 @@ mod tests { width: 16, ..studio.collection.canvas }; - studio.compositor.set_canvas(wide).unwrap(); - assert_ne!(studio.compositor.program(), &old, "a new texture"); studio.collection.canvas = wide; + studio.compositor.set_canvases(&studio.collection).unwrap(); + assert_ne!(studio.compositor.program(), &old, "a new texture"); let fill = Rect::fill(&wide); studio.collection.scenes[0].items = vec![ Item::new("red", Rect { width: 8.0, ..fill }), @@ -1819,9 +1992,20 @@ mod tests { width: limit + 2, ..wide }; - assert!(matches!(studio.compositor.set_canvas(huge), Err(Error::Invalid(_)))); + let refused = |canvas: Canvas| { + let mut collection = studio.collection.clone(); + collection.canvas = canvas; + collection + }; + assert!(matches!( + studio.compositor.set_canvases(&refused(huge)), + Err(Error::Invalid(_)) + )); let empty = Canvas { width: 0, ..wide }; - assert!(matches!(studio.compositor.set_canvas(empty), Err(Error::Invalid(_)))); + assert!(matches!( + studio.compositor.set_canvases(&refused(empty)), + Err(Error::Invalid(_)) + )); assert_eq!(studio.compositor.canvas(), wide); }) .await; diff --git a/rs/moq-scene/src/edit.rs b/rs/moq-scene/src/edit.rs index b9ab7778..a1996123 100644 --- a/rs/moq-scene/src/edit.rs +++ b/rs/moq-scene/src/edit.rs @@ -14,7 +14,9 @@ use std::collections::{BTreeMap, VecDeque}; use serde::{Deserialize, Serialize}; use crate::Error; -use crate::collection::{Audio, Canvas, Collection, Filter, Item, Layer, Mixes, Scene, Source, Transition, Video}; +use crate::collection::{ + Audio, Canvas, Collection, Extra, Filter, Item, Layer, Mixes, Scene, Source, Transition, Video, +}; /// One part of a collection, before and after an edit. /// @@ -70,6 +72,14 @@ pub enum Change { Transition { before: Transition, after: Transition }, /// [`Collection::canvas`]. Canvas { before: Canvas, after: Canvas }, + /// A [`Collection::canvases`] entry. + Extra { + name: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + before: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + after: Option, + }, /// [`Collection::mixes`]. Mixes { before: Mixes, after: Mixes }, } @@ -117,6 +127,11 @@ impl Change { before: after, after: before, }, + Self::Extra { name, before, after } => Self::Extra { + name, + before: after, + after: before, + }, Self::Mixes { before, after } => Self::Mixes { before: after, after: before, @@ -132,6 +147,7 @@ impl Change { | Self::Item { .. } | Self::Transition { .. } | Self::Canvas { .. } + | Self::Extra { .. } | Self::Mixes { .. } => None, } } @@ -146,6 +162,7 @@ impl Change { Self::Item { before, after, .. } => before == after, Self::Transition { before, after } => before == after, Self::Canvas { before, after } => before == after, + Self::Extra { before, after, .. } => before == after, Self::Mixes { before, after } => before == after, } } @@ -232,6 +249,18 @@ impl Change { (Self::Canvas { before, after }, Self::Canvas { before: b, after: a }) if after == *b => { Self::Canvas { before, after: *a } } + ( + Self::Extra { name, before, after }, + Self::Extra { + name: n, + before: b, + after: a, + }, + ) if name == *n && after == *b => Self::Extra { + name, + before, + after: a.clone(), + }, (Self::Mixes { before, after }, Self::Mixes { before: b, after: a }) if after == *b => Self::Mixes { before, after: a.clone(), @@ -260,7 +289,9 @@ impl Change { let items = collection .items_mut(layer) .ok_or_else(|| conflict(format!("there is no {layer}")))?; - splice(items, *index, before, after).map_err(|err| conflict(format!("item {err} in {layer}"))) + splice(items, *index, before, after).map_err(|err| conflict(format!("item {err} in {layer}")))?; + collection.prune(); + Ok(()) } Self::Transition { before, after } => { if collection.transition != *before { @@ -276,6 +307,7 @@ impl Change { collection.canvas = *after; Ok(()) } + Self::Extra { name, before, after } => replace(&mut collection.canvases, name, before, after, "canvas"), Self::Mixes { before, after } => { if collection.mixes != *before { return Err(conflict("the mixes changed".into())); @@ -647,6 +679,49 @@ impl Draft { self.change(Change::Canvas { before, after: canvas }) } + /// Add or resize the further canvas called `name`, keeping its top layer + /// and every scene's layout on it. Items keep their place in canvas + /// pixels. + pub fn set_extra(&mut self, name: &str, width: u32, height: u32) -> Result<(), Error> { + let before = self.collection.canvases.get(name).cloned(); + let after = match &before { + Some(extra) => Extra { + width, + height, + ..extra.clone() + }, + None => Extra::new(width, height), + }; + self.change(Change::Extra { + name: name.to_string(), + before, + after: Some(after), + }) + } + + /// Remove the further canvas called `name`, with its top layer and every + /// scene's layout on it. + pub fn remove_extra(&mut self, name: &str) -> Result<(), Error> { + let laid: Vec<(usize, Scene)> = self + .collection + .scenes + .iter() + .enumerate() + .filter(|(_, scene)| scene.canvases.contains_key(name)) + .map(|(index, scene)| (index, scene.clone())) + .collect(); + for (index, mut scene) in laid { + scene.canvases.remove(name); + self.set_scene(index, scene)?; + } + let before = self.collection.canvases.get(name).cloned(); + self.change(Change::Extra { + name: name.to_string(), + before, + after: None, + }) + } + /// Set the mixes and which one the program carries, and nothing else: see /// [`rename_mix`](Self::rename_mix) and [`remove_mix`](Self::remove_mix) for /// what refers to a mix. @@ -963,7 +1038,7 @@ mod tests { } fn main() -> Layer { - Layer::Scene("Scene".into()) + Layer::scene("Scene") } /// A collection with a red source shown in the one scene. @@ -1268,6 +1343,69 @@ mod tests { assert!(matches!(draft.finish(), Err(Error::Invalid(_)))); } + /// A further canvas is added, laid out, and resized as edits; removing a + /// source takes its items off every canvas; and removing the canvas takes + /// its top layer and every scene's layout on it, undoing whole. A layout + /// emptied item by item reads as no layout, and the file round-trips. + #[test] + fn a_further_canvas_edits_and_removes_whole() { + let mut collection = collection(); + collection.sources.insert("blue".into(), color([0, 0, 255, 255])); + let mut history = History::new(); + let tall = Layer::scene("Scene").on(Some("tall".into())); + let over = Layer::top().on(Some("tall".into())); + let fill = Rect::fill(&Canvas { + width: 1080, + height: 1920, + framerate: 30, + }); + + let mut draft = collection.draft("Add a canvas"); + assert!( + draft.insert_item(&tall, 0, Item::new("red", fill)).is_err(), + "no canvas yet" + ); + draft.set_extra("tall", 1080, 1920).unwrap(); + draft.insert_item(&tall, 0, Item::new("red", fill)).unwrap(); + draft.insert_item(&tall, 1, Item::new("blue", fill)).unwrap(); + draft.insert_item(&over, 0, Item::new("blue", fill)).unwrap(); + commit(&mut collection, &mut history, draft, None); + let added = collection.clone(); + assert_eq!(collection.items(&tall).unwrap().len(), 2); + assert_eq!(collection.canvas(Some("tall")).unwrap().framerate, 30); + assert_eq!( + Collection::from_toml(&collection.to_toml().unwrap()).unwrap(), + collection + ); + + let mut draft = collection.draft("Resize"); + draft.set_extra("tall", 720, 1280).unwrap(); + commit(&mut collection, &mut history, draft, None); + assert_eq!(collection.canvases["tall"].top.len(), 1, "the top layer stays"); + + let mut draft = collection.draft("Remove blue"); + draft.remove_source("blue").unwrap(); + commit(&mut collection, &mut history, draft, None); + assert_eq!(collection.items(&tall).unwrap().len(), 1); + assert!(collection.items(&over).unwrap().is_empty()); + + let mut draft = collection.draft("Remove red there"); + draft.remove_item(&tall, 0).unwrap(); + commit(&mut collection, &mut history, draft, None); + assert!(collection.scenes[0].canvases.is_empty(), "an empty layout is dropped"); + + history.seek(&mut collection, 1).unwrap(); + assert_eq!(collection, added); + let mut draft = collection.draft("Remove the canvas"); + draft.remove_extra("tall").unwrap(); + commit(&mut collection, &mut history, draft, None); + assert!(collection.canvases.is_empty()); + assert!(collection.scenes[0].canvases.is_empty()); + assert_eq!(collection.items(&tall), None); + history.undo(&mut collection).unwrap(); + assert_eq!(collection, added); + } + /// A renamed mix keeps its place, the program's pick, and every source left /// out of it; a removed one takes those exclusions with it and undoes /// whole; and the mix the program carries stays. diff --git a/rs/moq-scene/src/effect.rs b/rs/moq-scene/src/effect.rs index b1a0ca12..70b93446 100644 --- a/rs/moq-scene/src/effect.rs +++ b/rs/moq-scene/src/effect.rs @@ -405,11 +405,14 @@ enum Command { pub(crate) struct Frame { rasters: Vec, targets: Vec, + /// The canvas of the target drawn now, which items are prepared for. + canvas: Canvas, } impl Frame { - /// Start drawing a canvas-sized target, cleared to opaque black. - pub fn target(&mut self, target: &Picture) { + /// Start drawing a target the size of `canvas`, cleared to opaque black. + pub fn target(&mut self, target: &Picture, canvas: Canvas) { + self.canvas = canvas; self.targets.push(Command::Pass { view: target.view.clone(), clear: Some(wgpu::Color::BLACK), @@ -458,7 +461,6 @@ pub(crate) trait Library { pub(crate) struct Effects { device: wgpu::Device, queue: wgpu::Queue, - canvas: Canvas, input_layout: wgpu::BindGroupLayout, extra_layout: wgpu::BindGroupLayout, @@ -475,7 +477,8 @@ pub(crate) struct Effects { none: wgpu::BindGroup, empty_2d: wgpu::TextureView, empty_3d: wgpu::TextureView, - /// What a blend mode mixes with: a copy of its target under it. + /// What a blend mode mixes with: a copy of its target under it, as wide + /// and as tall as the largest canvas. backdrop: wgpu::Texture, backdrop_view: wgpu::TextureView, @@ -485,13 +488,13 @@ pub(crate) struct Effects { used: Vec, } -/// What a blend mode mixes with, as large as the canvas. -fn backdrop(device: &wgpu::Device, canvas: Canvas) -> wgpu::Texture { +/// What a blend mode mixes with, `width` by `height`. +fn backdrop(device: &wgpu::Device, width: u32, height: u32) -> wgpu::Texture { device.create_texture(&wgpu::TextureDescriptor { label: Some("moq-scene backdrop"), size: wgpu::Extent3d { - width: canvas.width, - height: canvas.height, + width, + height, depth_or_array_layers: 1, }, mip_level_count: 1, @@ -504,7 +507,8 @@ fn backdrop(device: &wgpu::Device, canvas: Canvas) -> wgpu::Texture { } impl Effects { - pub fn new(device: &wgpu::Device, queue: &wgpu::Queue, canvas: Canvas) -> Self { + /// Effects drawn onto canvases that fit in `width` by `height`. + pub fn new(device: &wgpu::Device, queue: &wgpu::Queue, width: u32, height: u32) -> Self { let texture = |binding, sample_type, view_dimension| wgpu::BindGroupLayoutEntry { binding, visibility: wgpu::ShaderStages::FRAGMENT, @@ -622,7 +626,7 @@ impl Effects { }; let empty_2d = empty(wgpu::TextureDimension::D2, FORMAT); let empty_3d = empty(wgpu::TextureDimension::D3, wgpu::TextureFormat::Rgba32Float); - let backdrop = backdrop(device, canvas); + let backdrop = backdrop(device, width, height); let alignment = u64::from(device.limits().min_uniform_buffer_offset_alignment); let stride = (size_of::() as u64).next_multiple_of(alignment); @@ -636,7 +640,6 @@ impl Effects { none: extra_bind(device, &extra_layout, [&empty_3d, &empty_2d, &empty_2d]), device: device.clone(), queue: queue.clone(), - canvas, input_layout, extra_layout, uniform_layout, @@ -650,14 +653,13 @@ impl Effects { } } - /// Draw onto `canvas` from the next frame on, reallocating what is sized - /// from it. - pub fn set_canvas(&mut self, canvas: Canvas) { - if canvas.width != self.canvas.width || canvas.height != self.canvas.height { - self.backdrop = backdrop(&self.device, canvas); + /// Draw onto canvases that fit in `width` by `height` from the next frame + /// on, reallocating what is sized from them. + pub fn fit(&mut self, width: u32, height: u32) { + if (width, height) != (self.backdrop.width(), self.backdrop.height()) { + self.backdrop = backdrop(&self.device, width, height); self.backdrop_view = self.backdrop.create_view(&Default::default()); } - self.canvas = canvas; } /// `texture`, a source's picture or a canvas-sized target, bound to sample. @@ -816,7 +818,7 @@ impl Effects { /// nothing in `library` yet. pub fn prepare(&mut self, frame: &mut Frame, item: Item<'_>, library: &dyn Library) -> Option { let area = Area::of(item.rect); - let canvas = Area::canvas(self.canvas); + let canvas = Area::canvas(frame.canvas); if area.is_empty() { return None; } @@ -936,9 +938,9 @@ impl Effects { /// A whole canvas-sized `picture` stretched over `at`, showing only within /// `within`, at `opacity`, to draw over another: `None` if none of it /// shows on the canvas. - pub fn place(&self, picture: &Picture, at: Rect, within: Rect, opacity: f32) -> Option { + pub fn place(&self, frame: &Frame, picture: &Picture, at: Rect, within: Rect, opacity: f32) -> Option { let area = Area::of(at); - let bounds = area.intersect(&Area::of(within)).intersect(&Area::canvas(self.canvas)); + let bounds = area.intersect(&Area::of(within)).intersect(&Area::canvas(frame.canvas)); if bounds.is_empty() || opacity <= 0.0 { return None; } @@ -970,7 +972,7 @@ impl Effects { bounds, blend, } = prepared; - let (width, height) = (self.canvas.width as f32, self.canvas.height as f32); + let (width, height) = (frame.canvas.width as f32, frame.canvas.height as f32); let x = |px: f32| px / width * 2.0 - 1.0; let y = |px: f32| 1.0 - px / height * 2.0; let mut uniform = self.uniform(input, segment, item, bounds); @@ -1028,7 +1030,7 @@ impl Effects { /// `input` through `segment`, drawn into a raster covering `area`. fn raster(&mut self, frame: &mut Frame, input: &Input, segment: &Segment, item: Area, area: Area) -> Input { - let size = self.texels(&area); + let size = self.texels(frame.canvas, &area); let raster = self.acquire(size); let draw = Draw { kind: Kind::Raster, @@ -1061,7 +1063,7 @@ impl Effects { sigma: f32, ) -> Input { let mut current = self.raster(frame, input, segment, item, area); - let mut size = self.texels(&area); + let mut size = self.texels(frame.canvas, &area); let texels = |size: (u32, u32)| (size.0 as f32 / area.width(), size.1 as f32 / area.height()); // Halve until the blur is a few texels wide; each texel of a halving // averages four, so nothing aliases. @@ -1124,7 +1126,7 @@ impl Effects { offset: (f32, f32), color: Rgba, ) -> Input { - let raster = self.acquire(self.texels(&area)); + let raster = self.acquire(self.texels(frame.canvas, &area)); let uniform = Uniform { rect: FULL, input: picture.area.map(UNIT, &area), @@ -1153,14 +1155,15 @@ impl Effects { } } - /// The texels a raster covering `area` gets: one per canvas pixel, unless - /// that is more than twice the canvas or the device allows either way. - fn texels(&self, area: &Area) -> (u32, u32) { + /// The texels a raster covering `area` of `canvas` gets: one per canvas + /// pixel, unless that is more than twice the canvas or the device allows + /// either way. + fn texels(&self, canvas: Canvas, area: &Area) -> (u32, u32) { let limit = self.device.limits().max_texture_dimension_2d; - let most = |canvas: u32| (canvas * 2).min(limit) as f32; + let most = |side: u32| (side * 2).min(limit) as f32; let scale = 1f32 - .min(most(self.canvas.width) / area.width()) - .min(most(self.canvas.height) / area.height()); + .min(most(canvas.width) / area.width()) + .min(most(canvas.height) / area.height()); let texels = |length: f32| ((length * scale).ceil() as u32).max(1); (texels(area.width()), texels(area.height())) } diff --git a/rs/moq-scene/src/lib.rs b/rs/moq-scene/src/lib.rs index f5d0d77f..b09b4974 100644 --- a/rs/moq-scene/src/lib.rs +++ b/rs/moq-scene/src/lib.rs @@ -3,9 +3,9 @@ //! The UI-free half of `moq-studio`, so the same scenes can be composed by a //! windowed app, a headless renderer, or a test: //! -//! - [`Collection`] is the scene graph: a [`Canvas`], named [`Source`]s, and -//! [`Scene`]s placing them, under a top layer that stays on across scene -//! switches. It saves to a versioned TOML file that holds no credentials. +//! - [`Collection`] is the scene graph: a main [`Canvas`] and any [`Extra`] +//! ones, named [`Source`]s, and [`Scene`]s placing them on each canvas, +//! under each canvas's top layer that stays on across scene switches. It saves to a versioned TOML file that holds no credentials. //! - [`Edit`] is every change to a collection, built with a [`Draft`]. Each //! one undoes itself, so a [`History`] undoes and redoes them and saves //! beside the scene file. A [`Clip`] carries items or filters between @@ -17,8 +17,9 @@ //! 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, cutting or running a -//! [`Transition`] between them. `moq_video::render::export` hands the +//! 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. `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 @@ -44,8 +45,8 @@ pub mod source; pub use clip::Clip; pub use collection::{ - Audio, Blend, Canvas, Collection, Crop, Direction, Filter, Item, Layer, MaskMode, Mixes, Rect, Rgba, Scene, Source, - Transition, Video, + Audio, Blend, Canvas, Collection, Crop, Direction, Extra, Filter, Item, Layer, MaskMode, Mixes, Rect, Rgba, Scene, + Source, Transition, Video, }; pub use compositor::{Compositor, FORMAT}; pub use edit::{Change, Draft, Edit, History}; diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index da4dc9e9..cc3d0874 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -192,7 +192,7 @@ impl State { before.apply(edit)?; } keep_on_air(self.compositor.on_air(), &renames)?; - self.compositor.set_canvas(collection.canvas)?; + self.compositor.set_canvases(&collection)?; for renames in &renames { self.compositor.rename(renames); } diff --git a/rs/moq-studio/src/interact.rs b/rs/moq-studio/src/interact.rs index 5d287b3f..91a0fc47 100644 --- a/rs/moq-studio/src/interact.rs +++ b/rs/moq-studio/src/interact.rs @@ -6,7 +6,7 @@ use std::time::Duration; use kio::clock::Instant; use moq_scene::browser::{Button, Input, Modifiers}; -use moq_scene::{Collection, Crop, Rect, Source}; +use moq_scene::{Collection, Crop, Layer, Rect, Source}; /// How close in time and space a press must follow the last to count toward a /// double or triple click. @@ -66,19 +66,22 @@ struct Place { impl Interact { /// Turn this frame's input over `image`, which shows `scene` of - /// `collection` under its top layer, into what each page it touches should receive, by page - /// name. + /// `collection` under its top layer on `canvas`, the main one for `None`, + /// into what each page it touches should receive, by page name. pub fn update( &mut self, ui: &egui::Ui, image: egui::Rect, collection: &Collection, + canvas: Option<&str>, scene: Option<&str>, ) -> Vec<(String, Input)> { - let canvas = collection.canvas; + let Some(size) = collection.canvas(canvas) else { + return Vec::new(); + }; let on_canvas = |pos: egui::Pos2| { - let x = (pos.x - image.min.x) / image.width() * canvas.width as f32; - let y = (pos.y - image.min.y) / image.height() * canvas.height as f32; + let x = (pos.x - image.min.x) / image.width() * size.width as f32; + let y = (pos.y - image.min.y) / image.height() * size.height as f32; (x, y) }; let hit = |pos: egui::Pos2| -> Option { @@ -86,7 +89,7 @@ impl Interact { return None; } let (x, y) = on_canvas(pos); - page_at(collection, scene, x, y) + page_at(collection, canvas, scene, x, y) }; // Where `pos` falls on the captured page, even off its edges. let on_page = |capture: &Capture, pos: egui::Pos2| -> Option<(i32, i32)> { @@ -232,7 +235,7 @@ impl Interact { let scale = match unit { egui::MouseWheelUnit::Point => 1.0, egui::MouseWheelUnit::Line => 40.0, - egui::MouseWheelUnit::Page => canvas.height as f32, + egui::MouseWheelUnit::Page => size.height as f32, }; send( &page, @@ -396,12 +399,16 @@ impl Interact { /// The topmost visible browser source, or clone of one, at canvas point (`x`, /// `y`) in `scene` and the top layer over it, and the point on its page. -fn page_at(collection: &Collection, scene: Option<&str>, x: f32, y: f32) -> Option { - let scene = scene.and_then(|name| collection.scene(name)); - let items = scene.map(|scene| scene.items.as_slice()).unwrap_or_default(); - items +fn page_at(collection: &Collection, canvas: Option<&str>, scene: Option<&str>, x: f32, y: f32) -> Option { + let layer = |scene: Option<&str>| Layer { + canvas: canvas.map(str::to_string), + scene: scene.map(str::to_string), + }; + let items = |layer: Layer| collection.items(&layer).unwrap_or_default(); + let scene = scene.map(|scene| items(layer(Some(scene)))).unwrap_or_default(); + scene .iter() - .chain(&collection.top) + .chain(items(layer(None))) .rev() .filter(|item| item.visible) .find_map(|item| { @@ -587,11 +594,12 @@ mod tests { height: 800.0, }; collection.scenes = vec![Scene { - name: "Main".into(), items: vec![item("alerts", fill, true), item("chat", corner, true)], + ..Scene::new("Main") }]; - let at = - |collection: &Collection, x, y| page_at(collection, Some("Main"), x, y).map(|hit| (hit.page, hit.x, hit.y)); + let at = |collection: &Collection, x, y| { + page_at(collection, None, Some("Main"), x, y).map(|hit| (hit.page, hit.x, hit.y)) + }; assert_eq!(at(&collection, 1720.0, 680.0), Some(("chat".into(), 200, 400))); assert_eq!(at(&collection, 960.0, 540.0), Some(("alerts".into(), 400, 300))); @@ -602,7 +610,7 @@ mod tests { Some(("alerts".into(), 716, 377)), "a hidden page takes no input" ); - assert!(page_at(&collection, Some("Other"), 1.0, 1.0).is_none()); + assert!(page_at(&collection, None, Some("Other"), 1.0, 1.0).is_none()); // The top layer draws over every scene, and a clone shows its source's // page, which runs under the source's name. @@ -612,7 +620,7 @@ mod tests { collection.top = vec![item("chat copy", corner, true)]; assert_eq!(at(&collection, 1720.0, 680.0), Some(("chat".into(), 200, 400))); assert_eq!( - page_at(&collection, None, 1720.0, 680.0).map(|hit| hit.page), + page_at(&collection, None, None, 1720.0, 680.0).map(|hit| hit.page), Some("chat".into()) ); @@ -663,7 +671,7 @@ mod tests { }; let mut delivered = Vec::new(); let mut output = ctx.run_ui(input, |ui| { - delivered = interact.update(ui, image, collection, Some("Main")); + delivered = interact.update(ui, image, collection, None, Some("Main")); }); // Nothing renders, so the font atlas egui uploaded goes nowhere. output.textures_delta.clear(); @@ -695,8 +703,8 @@ mod tests { Item::new("chat", rect) }; collection.scenes = vec![Scene { - name: "Main".into(), items: vec![item(0.0), item(100.0)], + ..Scene::new("Main") }]; let ctx = egui::Context::default(); let mut interact = Interact::default(); @@ -752,8 +760,8 @@ mod tests { Item::new(source, rect) }; collection.scenes = vec![Scene { - name: "Main".into(), items: vec![item("left", 0.0), item("right", 100.0)], + ..Scene::new("Main") }]; let ctx = egui::Context::default(); let mut interact = Interact::default(); @@ -845,8 +853,8 @@ mod tests { }, ); collection.scenes = vec![Scene { - name: "Main".into(), items: vec![Item::new("chat", Rect::fill(&collection.canvas))], + ..Scene::new("Main") }]; let delivered = frame( &egui::Context::default(), @@ -903,8 +911,8 @@ mod tests { }, ); collection.scenes = vec![Scene { - name: "Main".into(), items: vec![Item::new("chat", Rect::fill(&collection.canvas))], + ..Scene::new("Main") }]; let ctx = egui::Context::default(); let mut interact = Interact::default(); @@ -958,8 +966,8 @@ mod tests { }, ); collection.scenes = vec![Scene { - name: "Main".into(), items: vec![Item::new("chat", Rect::fill(&collection.canvas))], + ..Scene::new("Main") }]; let ctx = egui::Context::default(); let mut interact = Interact::default(); diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index f31be186..dbb990c9 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -100,6 +100,7 @@ impl Studio { runtime, preview, program, + view: None, top: false, selection: None, gestures: Gestures::default(), @@ -131,6 +132,9 @@ impl Studio { } } +/// What the canvas picker calls the main canvas. +const MAIN: &str = "Main"; + /// Where the output Listen plays on is kept, per user. const SPEAKER: &str = "speaker"; @@ -152,8 +156,10 @@ impl eframe::App for Studio { let engine = self.panels.engine.clone(); let state = engine.state.lock().unwrap(); if let Some(render) = frame.wgpu_render_state() { - self.panels.preview.follow(render, state.compositor.preview()); - self.panels.program.follow(render, state.compositor.program()); + let view = state.compositor.output(self.panels.view.as_deref()); + let view = view.unwrap_or_else(|| state.compositor.output(None).expect("the main canvas")); + self.panels.preview.follow(render, view.preview()); + self.panels.program.follow(render, view.program()); } // Redraw at the canvas rate, so preview and program are live. let framerate = state.compositor.canvas().framerate; @@ -220,6 +226,9 @@ struct Panels { runtime: tokio::runtime::Handle, preview: Registered, program: Registered, + /// The canvas Preview, Program, and Items show: a further one by name, or + /// `None` for the main one. + view: Option, /// Whether the Items panel shows the top layer rather than the scene in /// preview. top: bool, @@ -599,14 +608,49 @@ impl Panels { self.renaming = None; } - /// The layer the Items panel shows, and new sources are placed in. + /// The layer the Items panel shows, and new sources are placed in, on the + /// canvas shown. fn layer(&self, state: &State) -> Result { - if self.top { - return Ok(Layer::Top); + let layer = match (self.top, state.compositor.preview_scene()) { + (true, _) => Layer::top(), + (false, Some(name)) => Layer::scene(name), + (false, None) => return Err("Pick a scene to edit what it shows.".into()), + }; + Ok(layer.on(self.view.clone())) + } + + /// The canvas Preview, Program, and Items show, and its size: the main + /// one once a further one shown is removed. + fn view(&mut self, state: &State) -> Canvas { + if let Some(canvas) = state.collection().canvas(self.view.as_deref()) { + return canvas; } - match state.compositor.preview_scene() { - Some(name) => Ok(Layer::Scene(name.to_string())), - None => Err("Pick a scene to edit what it shows.".into()), + self.view = None; + self.deselect_item(); + state.collection().canvas + } + + /// Pick which canvas Preview, Program, and Items show. + fn pick_view(&mut self, ui: &mut Ui, collection: &Collection) { + if collection.canvases.is_empty() { + return; + } + let name = |view: &Option| view.clone().unwrap_or_else(|| MAIN.to_string()); + let mut view = self.view.clone(); + egui::ComboBox::from_id_salt("view") + .selected_text(name(&view)) + .show_ui(ui, |ui| { + let views = std::iter::once(None).chain(collection.canvases.keys().cloned().map(Some)); + for option in views { + let label = name(&option); + ui.selectable_value(&mut view, option, label); + } + }) + .response + .on_hover_text("The canvas preview and program show, and the items edited are on"); + if view != self.view { + self.view = view; + self.deselect_item(); } } @@ -709,8 +753,9 @@ impl Panels { let engine = self.engine.clone(); let mut state = engine.state.lock().unwrap(); self.follow_reloads(&state); - let canvas = state.collection().canvas; + let canvas = self.view(&state); ui.horizontal(|ui| { + self.pick_view(ui, state.collection()); ui.label(format!( "Preview: {}", state.compositor.preview_scene().unwrap_or("nothing") @@ -744,7 +789,9 @@ impl Panels { let image = picture(ui, self.preview.id, canvas, egui::Sense::hover()).rect; let scene = state.compositor.preview_scene(); self.rebind(&state.sources); - let delivered = self.interact.update(ui, image, state.collection(), scene); + let delivered = self + .interact + .update(ui, image, state.collection(), self.view.as_deref(), scene); deliver(&state.sources, delivered); self.rebind(&state.sources); return; @@ -774,7 +821,7 @@ impl Panels { } let Ok(layer) = self.layer(state) else { return Ok(()) }; - let Some(items) = state.collection().items(&layer).cloned() else { + let Some(items) = state.collection().items(&layer).map(<[Item]>::to_vec) else { return Ok(()); }; let selected = match &self.selection { @@ -842,7 +889,11 @@ impl Panels { } // Drawn from the collection and selection as this frame left them. - let items = state.collection().items(&layer).cloned().unwrap_or_default(); + let items = state + .collection() + .items(&layer) + .map(<[Item]>::to_vec) + .unwrap_or_default(); let selected = match &self.selection { Some(Selection::Item(selected, index)) if *selected == layer => Some(*index), _ => None, @@ -877,7 +928,7 @@ impl Panels { fn program(&mut self, ui: &mut Ui) { let engine = self.engine.clone(); let mut state = engine.state.lock().unwrap(); - let canvas = state.collection().canvas; + let canvas = self.view(&state); let transition = state.collection().transition; let mut next = transition; let mut gesture = None; @@ -1056,7 +1107,7 @@ impl Panels { /// Keep a selected item of the scene renamed from `from` selected. fn renamed(&mut self, from: &str, to: &str) { - if let Some(Selection::Item(Layer::Scene(scene), _)) = &mut self.selection + if let Some(Selection::Item(Layer { scene: Some(scene), .. }, _)) = &mut self.selection && scene == from { *scene = to.to_string(); @@ -1121,12 +1172,12 @@ impl Panels { return; } }; + let canvas = self.view(&state); let collection = state.collection(); - let Some(items) = collection.items(&layer).cloned() else { + let Some(items) = collection.items(&layer).map(<[Item]>::to_vec) else { ui.label(format!("There is no {layer}.")); return; }; - let canvas = collection.canvas; let mut action = None; egui::ScrollArea::vertical().show(ui, |ui| { @@ -2719,7 +2770,10 @@ fn add( !state.collection().sources.contains_key(&name), "there is already a source named `{name}`" ); - let canvas = state.collection().canvas; + let canvas = layer + .as_ref() + .and_then(|layer| state.collection().canvas(layer.canvas.as_deref())); + let canvas = canvas.unwrap_or(state.collection().canvas); let place = source.has_video(); // A page or overlay, or a clone of one, lands at its own size, pixel for // pixel; anything else fills the canvas. @@ -2740,7 +2794,7 @@ fn add( draft.set_source(&name, Some(source))?; draft.set_video(&name, video)?; if place && let Some(layer) = layer { - let index = draft.collection().items(&layer).map_or(0, Vec::len); + let index = draft.collection().items(&layer).map_or(0, <[Item]>::len); draft.insert_item(&layer, index, Item::new(name.clone(), rect))?; } Ok(()) @@ -2849,17 +2903,16 @@ struct Drag { /// How far, in points on screen, the pointer reaches to an edge or a line. const REACH: f32 = 8.0; -/// The visible items of the scene in preview and the top layer, other than -/// the one at `index` of `layer`: what a dragged item snaps to. +/// The visible items of the scene in preview and the top layer on `layer`'s +/// canvas, other than the one at `index` of `layer`: what a dragged item snaps +/// to. fn others(state: &State, layer: &Layer, index: usize) -> Vec { - let scene = state - .compositor - .preview_scene() - .map(|name| Layer::Scene(name.to_string())); + let scene = state.compositor.preview_scene().map(Layer::scene); let collection = state.collection(); let mut rects = Vec::new(); - for shown in scene.into_iter().chain([Layer::Top]) { - let items = collection.items(&shown).map_or(&[][..], Vec::as_slice); + for shown in scene.into_iter().chain([Layer::top()]) { + let shown = shown.on(layer.canvas.clone()); + let items = collection.items(&shown).unwrap_or_default(); for (at, item) in items.iter().enumerate() { if item.visible && !(shown == *layer && at == index) { rects.push(item.rect); From 3b0e665cfced4af80f95f85664e30ead42e2af34 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 13:43:41 -0800 Subject: [PATCH 03/10] feat(studio): each further canvas publishes its own broadcast with its own encoders Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/engine.rs | 807 ++++++++++++++++++++++++++---------- rs/moq-studio/src/ladder.rs | 5 - rs/moq-studio/src/ui.rs | 222 +++++++--- 3 files changed, 769 insertions(+), 265 deletions(-) diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index cc3d0874..3153e53d 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -201,6 +201,23 @@ impl State { Ok(()) } + /// The visible items any canvas's program draws: of the scenes on air, and + /// of the top layers over them. + pub fn on_air(&self) -> impl Iterator { + 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)); + canvases + .flat_map(move |canvas| { + scenes.clone().into_iter().map(move |scene| moq_scene::Layer { + canvas: canvas.cloned(), + scene: scene.map(str::to_string), + }) + }) + .flat_map(|layer| collection.items(&layer).unwrap_or_default()) + .filter(|item| item.visible) + } + /// Restart the source called `name`, as an edit would, such as to reload a /// web page. pub fn restart(&mut self, name: &str) { @@ -264,8 +281,28 @@ fn keep_on_air<'a>( pub struct Stats { /// Canvas frames drawn. pub rendered: AtomicU64, - /// What each of the program's video encoders costs: the canvas size, the - /// half size, and the recording's own canvas size. + /// The main canvas's program video. + pub video: Arc, + /// Each further canvas's program video, by the canvas's name, while it is + /// drawn. + canvases: Mutex>>, + /// The video encoders open: every canvas's program and the sources'. + pub sessions: Sessions, + /// Each program track's share of the connection's bandwidth, by name: a + /// further canvas's prefixed with its own. + grants: Mutex>, + /// Why the audio stopped, if it did. + audio_error: Mutex>, + /// Why the recording's copy of the audio stopped, if it did, which leaves + /// the published audio running. + recording_audio_error: Mutex>, +} + +/// How one canvas's program video 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. pub renditions: [Load; 3], /// Whether the encoder reads the program on the GPU, rather than each frame /// being read back to the CPU. @@ -277,22 +314,13 @@ pub struct Stats { /// while on demand and unwatched, nor while a new rendition opens, nor /// while the grant cannot fund it. pub encoding: AtomicBool, - /// The video encoders open, the program's and the sources'. - pub sessions: Sessions, - /// How many sizes the program's video has now: the canvas size, and the - /// half size unless the canvas is too small or its encoder refused it. + /// How many sizes the video has now: the canvas size, and the half size + /// unless the canvas is too small or its encoder refused it. pub sizes: std::sync::atomic::AtomicUsize, - /// Each program track's share of the connection's bandwidth, by name. - grants: Mutex>, - /// The encoder backend, once the program's current rendition opened one. + /// The encoder backend, once the current rendition opened one. encoder: Mutex, /// Why the video stopped, if it did, until a new rendition starts. - video_error: Mutex>, - /// Why the audio stopped, if it did. - audio_error: Mutex>, - /// Why the recording's copy of the audio stopped, if it did, which leaves - /// the published audio running. - recording_audio_error: Mutex>, + error: Mutex>, } /// What one of the program's renditions costs. @@ -313,7 +341,7 @@ pub struct Load { pub encode_us: AtomicU64, /// Why this encoder stopped, or could not start, while the canvas size /// carries on, until the next canvas. The canvas size's own failure stops - /// the video, which [`Stats::error`] reports. + /// the video, which [`VideoStats::error`] reports. error: Mutex>, } @@ -329,23 +357,23 @@ impl Load { } } -/// The encoder backend the program's current rendition opened, if it has. +/// The encoder backend the current rendition opened, if it has. #[derive(Default)] struct EncoderName { - /// Counts the program's renditions, so one a resize retired while its - /// encoder was still opening cannot report that encoder afterwards. + /// Counts the renditions, so one a resize retired while its encoder was + /// still opening cannot report that encoder afterwards. rendition: u64, name: Option, } -impl Stats { - /// The encoder backend, once the program's current rendition opened one. +impl VideoStats { + /// The encoder backend, once the current rendition opened one. pub fn encoder(&self) -> Option { self.encoder.lock().unwrap().name.clone() } - /// The number of the program's current rendition, which its encode thread - /// reports its encoder under. + /// The number of the current rendition, which its encode thread reports + /// its encoder under. fn rendition(&self) -> u64 { self.encoder.lock().unwrap().rendition } @@ -358,7 +386,7 @@ impl Stats { encoder.name = None; } - /// Report `name` as the program's encoder, unless `rendition` was retired. + /// Report `name` as the encoder, unless `rendition` was retired. fn opened_encoder(&self, rendition: u64, name: &str) { let mut encoder = self.encoder.lock().unwrap(); if encoder.rendition == rendition { @@ -366,6 +394,23 @@ impl Stats { } } + /// Why the video stopped, if it did. + pub fn error(&self) -> Option { + self.error.lock().unwrap().clone() + } + + fn fail(&self, err: impl std::fmt::Display) { + tracing::error!(%err, "video stopped"); + *self.error.lock().unwrap() = Some(format!("video: {err}")); + } +} + +impl Stats { + /// Each further canvas's program video, by the canvas's name. + pub fn canvases(&self) -> BTreeMap> { + self.canvases.lock().unwrap().clone() + } + /// Each program track's share of the connection's bandwidth now, by name: /// `None` while the connection has no estimate. pub fn grants(&self) -> BTreeMap> { @@ -376,9 +421,9 @@ impl Stats { .collect() } - /// Why the output stopped, if it did. + /// Why the main canvas's output stopped, if it did. pub fn error(&self) -> Option { - let video = self.video_error.lock().unwrap().clone(); + let video = self.video.error(); let audio = self.audio_error.lock().unwrap().clone(); match (video, audio) { (Some(video), Some(audio)) => Some(format!("{video}; {audio}")), @@ -386,11 +431,6 @@ impl Stats { } } - fn fail_video(&self, err: impl std::fmt::Display) { - tracing::error!(%err, "video stopped"); - *self.video_error.lock().unwrap() = Some(format!("video: {err}")); - } - /// The recording's audio copy failed, which stops what recordings hold but /// not the published program. fn fail_recording_audio(&self, err: impl std::fmt::Display) { @@ -509,13 +549,9 @@ impl Program { /// Encode the local sources the program shows, from `state`, whether or /// not anything reads their tracks, so a replay of one has its past. fn show(&self, state: &State) { - let collection = &state.collection; - let scenes = state.compositor.on_air().filter_map(|scene| collection.scene(scene)); - let shown = scenes - .flat_map(|scene| &scene.items) - .chain(&collection.top) - .filter(|item| item.visible) - .map(|item| collection.capture(&item.source).to_string()) + let shown = state + .on_air() + .map(|item| state.collection.capture(&item.source).to_string()) .collect(); self.sources.show(shown); } @@ -658,6 +694,7 @@ impl Engine { browser: browser(&gpu)?, }); let mut compositor = Compositor::new(&gpu.device, &gpu.queue, collection.canvas)?; + compositor.set_canvases(&collection)?; compositor.set_preview(collection.scenes.first().map(|scene| scene.name.clone())); compositor.cut(); @@ -688,6 +725,7 @@ impl Engine { let Publish { renditions, ladder, + extras, audio, recorded_audio, clock, @@ -700,8 +738,9 @@ impl Engine { // catalog advertises for a frame is when the program drew it. let epoch = Duration::from(clock.capture(start)?); let program = engine.clone(); + let main = Line::new(renditions, canvas, ladder, true); let thread = clocked("program".into(), async move || { - program.run_program(start, epoch, renditions, ladder).await + program.run_program(start, epoch, main, extras).await })?; engine.threads.lock().unwrap().push(thread); let mix = engine.clone(); @@ -848,12 +887,7 @@ impl Engine { let stalled = { let state = self.state.lock().unwrap(); let shown: Vec<&str> = state - .compositor .on_air() - .filter_map(|name| state.collection.scene(name)) - .flat_map(|scene| &scene.items) - .chain(&state.collection.top) - .filter(|item| item.visible) .map(|item| state.collection.capture(&item.source)) .collect(); state @@ -869,67 +903,58 @@ impl Engine { }) .collect() }; - let canvas = &stats.renditions[0]; + let video = &stats.video; + let canvas = &video.renditions[0]; health::Local { rendered: stats.rendered.load(Ordering::Relaxed), dropped: canvas.dropped.load(Ordering::Relaxed), encoded: canvas.encoded.load(Ordering::Relaxed), - encoding: stats.encoding.load(Ordering::Relaxed), + encoding: video.encoding.load(Ordering::Relaxed), grants: stats.grants(), - watched: stats.watched.load(Ordering::Relaxed), - encoder: stats.encoder(), + watched: video.watched.load(Ordering::Relaxed), + encoder: video.encoder(), failed: stats.error(), stalled, } } - /// Draw a canvas frame per tick of the program clock, and hand the program - /// to each rendition's encoder, stamped `epoch` past `start`, while its gate - /// lets it encode: the canvas size every frame while live and not on - /// demand, otherwise only while anyone watches it, the half size only - /// while anyone watches it, and the recording's own only while recording. + /// Draw every canvas once per tick of the program clock, and hand each + /// 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. /// - /// A new canvas size or frame rate ends `ladder`, the renditions it was - /// drawn at, and opens others in `renditions`, since a track never changes - /// what it carries. Drawing outlives the encoders: a failed output stops copying - /// the program, and the preview and program keep moving. + /// `main` encodes the main canvas. Each further canvas gets a [`Line`] of + /// its own from `extras` once drawn, in a broadcast of its own, and ends it + /// once the collection drops it. 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, renditions: Renditions, mut ladder: Option) { - let mut canvas = self.state.lock().unwrap().compositor.canvas(); - let mut clock = program_clock(start, canvas); - // One probe at a time, with the canvas it is for, since a GPU caps its - // encode sessions. - let mut probing: Option<(Canvas, std::thread::JoinHandle>)> = None; - // The encode threads of renditions ended or failed since, whose - // sessions the next probe waits to be closed. - let mut ending = Vec::new(); - // Whether the canvas drawn has its renditions, or failed to open them, - // so a failure is not retried until the canvas changes. - let mut settled = true; - if let Some(ladder) = &ladder { - ladder.adopted(&self.stats); - } + async fn run_program(&self, start: Instant, epoch: Duration, mut main: Line, extras: Extras) { + let mut framerate = self.state.lock().unwrap().compositor.canvas().framerate; + let mut clock = program_clock(start, framerate); + // Each further canvas's line, or `None` once it failed to start, which + // is not retried until the canvas is removed and added again. + let mut lines: BTreeMap> = BTreeMap::new(); + // The encoders of lines since ended, waited on as the studio stops so + // each finishes its track. + let mut retired: Vec = Vec::new(); loop { let tick = clock.tick().await; if self.stopped.load(Ordering::Relaxed) { // Without a sender, each encoder publishes what it was sent and // still holds, then finishes its track. - ending.extend(ladder.take().into_iter().flat_map(|ladder| ladder.end(&self.stats))); - for thread in ending { - if thread.join().is_err() { - tracing::error!("an encode thread panicked"); - } - } - // It holds the threads of the renditions it waits on. - if let Some((_, probe)) = probing { - let _ = probe.join(); + retired.push(main.end()); + retired.extend(lines.into_values().flatten().map(|(line, _)| line.end())); + for ended in retired { + ended.join(); } return; } - let (program, drawn) = { + let drawn: Vec<(Option, wgpu::Texture, Canvas)> = { let mut guard = self.state.lock().unwrap(); let state = &mut *guard; if std::mem::take(&mut state.stale) { @@ -943,80 +968,52 @@ impl Engine { continue; } self.program.show(state); - (state.compositor.program().clone(), state.compositor.canvas()) + state + .compositor + .outputs() + .map(|(name, output)| (name.map(str::to_string), output.program().clone(), output.canvas())) + .collect() }; self.stats.rendered.fetch_add(1, Ordering::Relaxed); - if drawn != canvas { - if drawn.framerate != canvas.framerate { - clock = program_clock(tick + period(drawn), drawn); - } - canvas = drawn; - settled = false; - // Ended before the next open, so the two never hold encode - // sessions at once. - ending.extend(ladder.take().into_iter().flat_map(|ladder| ladder.end(&self.stats))); - self.stats.retire_encoder(); - } - // Probed beside the program clock, so drawing carries on. - if let Some((probed, probe)) = probing.take_if(|(_, probe)| probe.is_finished()) { - let probe = probe - .join() - .unwrap_or_else(|_| Err(anyhow::anyhow!("the encoder probe panicked"))); - // One for a canvas since left is dropped unlisted. - if probed == canvas { - settled = true; - match probe.and_then(|probe| renditions.start(probe)) { - Ok((started, _)) => { - started.adopted(&self.stats); - ladder = Some(started); - } - Err(err) => self.stats.fail_video(format!("{err:#}")), - } - } - } - if probing.is_none() && !settled { - let ending = std::mem::take(&mut ending); - match Probe::spawn(renditions.gpu.clone(), canvas, ending) { - Ok(probe) => probing = Some((canvas, probe)), - Err(err) => { - settled = true; - self.stats.fail_video(format!("cannot start an encoder probe: {err}")); - } - } + let timestamp = moq_net::Timestamp::try_from(epoch + tick.saturating_duration_since(start)).ok(); + // Every canvas shares the main one's frame rate. + let drawing = drawn[0].2.framerate; + if drawing != framerate { + framerate = drawing; + clock = program_clock(tick + period(framerate), framerate); } - let Some(current) = &mut ladder else { - self.stats.watched.store(false, Ordering::Relaxed); - self.stats.encoding.store(false, Ordering::Relaxed); - continue; - }; - let watched = current - .rungs - .iter() - .any(|rung| rung.role.published() && rung.gate.watched()); - self.stats.watched.store(watched, Ordering::Relaxed); - let Ok(timestamp) = moq_net::Timestamp::try_from(epoch + tick.saturating_duration_since(start)) else { - continue; - }; - let mut encoding = false; - let mut failed = Vec::new(); - for (index, rung) in current.rungs.iter_mut().enumerate() { - let role = rung.role; - match rung.feed(&self.stats.renditions[role as usize], &program, timestamp) { - Some(step) => encoding |= role == Role::Canvas && matches!(step, Step::Encode(_)), - None => failed.push((index, role)), + retired.retain(|ended| !ended.is_finished()); + let gone: Vec = lines + .keys() + .filter(|line| !drawn.iter().any(|(name, _, _)| name.as_ref() == Some(*line))) + .cloned() + .collect(); + for name in gone { + if let Some(Some((line, _))) = lines.remove(&name) { + retired.push(line.end()); } + self.stats.canvases.lock().unwrap().remove(&name); } - self.stats.encoding.store(encoding, Ordering::Relaxed); - // An encoder stops only having reported why. Without the canvas - // size the video stops; without another rendition it carries on. - if failed.iter().any(|(_, role)| *role == Role::Canvas) { - ending.extend(ladder.take().into_iter().flat_map(|ladder| ladder.end(&self.stats))); - } else { - for (index, _) in failed.into_iter().rev() { - ending.push(current.remove(index, &self.stats)); - } + + for (name, program, canvas) in drawn { + let line = match &name { + None => &mut main, + Some(name) => { + let started = lines.entry(name.clone()).or_insert_with(|| { + let video = Arc::new(VideoStats::default()); + self.stats.canvases.lock().unwrap().insert(name.clone(), video.clone()); + extras + .start(name, canvas, video.clone(), &self.program) + .inspect_err(|err| video.fail(format!("{err:#}"))) + .ok() + }); + let Some((line, _)) = started else { continue }; + line + } + }; + line.feed(&program, canvas, timestamp); } } } @@ -1131,6 +1128,8 @@ struct Publish { /// The renditions of the canvas the studio started with, unless the canvas /// could not be encoded without a relay. ladder: Option, + /// What publishes each further canvas. + extras: Extras, /// The program audio's encoder. audio: moq_audio::encode::Producer, /// The recording's own copy of it. @@ -1172,10 +1171,10 @@ 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).await { + let probe = match Probe::new(gpu, canvas, true).await { Ok(probe) => Some(probe), Err(err) if live.is_none() => { - stats.fail_video(format!("{err:#}")); + stats.video.fail(format!("{err:#}")); None } Err(err) => return Err(err), @@ -1225,16 +1224,18 @@ async fn publish( gpu: gpu.clone(), broadcast: broadcast.clone(), timebase: catalog.timebase(), - recording: Recording { + recording: Some(Recording { broadcast: recorder.clone(), timebase: recorded.timebase(), - }, + }), bandwidth: bandwidth.clone(), // Live on a relay and not on demand; otherwise only a viewer asks for // the video. on_demand: live.is_none_or(|live| live.on_demand), live: live.is_some(), stats: stats.clone(), + video: stats.video.clone(), + grants: String::new(), }; let ladder = match probe { Some(probe) => { @@ -1289,7 +1290,7 @@ async fn publish( watching.clone().follow(live.reports.clone(), path.clone()), )); health::Publisher { - observer: hang::health::Observer::new(path.clone(), Some(instance)), + observer: hang::health::Observer::new(path.clone(), Some(instance.clone())), catalog: graded, clock: clock.clone(), relays: live.links.clone(), @@ -1301,7 +1302,7 @@ async fn publish( Ok(( Program { - _origin: origin, + _origin: origin.clone(), publication, catalog: listed, clock: clock.clone(), @@ -1312,6 +1313,17 @@ async fn publish( _audio: audio_reservation, }, Publish { + extras: Extras { + origin: origin.clone(), + path: path.clone(), + clock: clock.clone(), + instance, + gpu: gpu.clone(), + bandwidth, + on_demand: renditions.on_demand, + live: renditions.live, + stats: stats.clone(), + }, renditions, ladder, audio, @@ -1361,8 +1373,8 @@ struct Renditions { /// Shared by every rendition, so each lands on the catalog's clock the same /// way and a switch keeps the video in step with the audio. timebase: moq_mux::catalog::Timebase, - /// Where the recording's own encoder writes. - recording: Recording, + /// Where the recording's own encoder writes: the main canvas's alone. + recording: Option, /// The connection's bandwidth, which each published rendition follows its /// share of. bandwidth: bandwidth::Allocator, @@ -1371,6 +1383,11 @@ struct Renditions { /// Whether the program goes out on a relay, where viewers can watch it. live: bool, stats: Arc, + /// How this canvas's video is doing. + video: Arc, + /// What each track's grant is listed under in [`Stats::grants`], before + /// its name: empty for the main canvas, and a further one's name. + grants: String, } impl Renditions { @@ -1380,10 +1397,10 @@ impl Renditions { /// recording's once recorded. fn start(&self, probe: Probe) -> anyhow::Result<(Ladder, tokio::sync::oneshot::Receiver>)> { // Before the encode threads start, which report their own failures. - *self.stats.video_error.lock().unwrap() = None; + *self.video.error.lock().unwrap() = None; for (role, load) in [Role::Half, Role::Record] .into_iter() - .map(|role| (role, &self.stats.renditions[role as usize])) + .map(|role| (role, &self.video.renditions[role as usize])) { let refused = probe.refused.iter().find(|(refused, _)| *refused == role); *load.error.lock().unwrap() = refused.map(|(_, reason)| reason.clone()); @@ -1400,7 +1417,7 @@ impl Renditions { .collect(); let (opened, open) = tokio::sync::oneshot::channel(); let mut opened = Some(opened); - let number = self.stats.rendition(); + let number = self.video.rendition(); let mut rungs = Vec::with_capacity(probe.rungs.len()); for probed in probe.rungs { let Probed { @@ -1409,8 +1426,9 @@ impl Renditions { config, video, } = probed; - let (broadcast, timebase) = match role { - Role::Record => (&self.recording.broadcast, &self.recording.timebase), + 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 reserved = timebase.reserve().with_interval(config.gop.duration(config.framerate)); @@ -1427,16 +1445,19 @@ impl Renditions { let info = reserved.track_info(priority).with_refresh(ladder::REFRESH); let track = broadcast.unique_track(suffix, info)?; let producer = moq_video::encode::Producer::with_track(track.clone(), reserved, video)?; - let gate = match role { + let (gate, grant) = match role { // Reads nothing from the connection, so it always encodes at its // ceiling, and only while recorded. - Role::Record => Gate::new( - &bandwidth::Allocator::unlimited(), - producer.demand(), - ladder::ceiling(config.size(), config.framerate), - None, - false, - ), + Role::Record => { + let gate = Gate::new( + &bandwidth::Allocator::unlimited(), + producer.demand(), + ladder::ceiling(config.size(), config.framerate), + None, + false, + ); + (gate, None) + } _ => { let gate = Gate::new( &self.bandwidth, @@ -1445,19 +1466,17 @@ impl Renditions { ceilings.get(rank + 1).copied(), role == Role::Canvas && !self.on_demand, ); - self.stats - .grants - .lock() - .unwrap() - .insert(track.name().to_string(), gate.grant()); - gate + let grant = format!("{}{}", self.grants, track.name()); + self.stats.grants.lock().unwrap().insert(grant.clone(), gate.grant()); + (gate, Some(grant)) } }; // One more than the exporter's pool, for a skip behind a full pool. let (frames, pending) = mpsc::sync_channel(Exporter::DEPTH + 1); let job = Job { - stats: self.stats.clone(), + sessions: self.stats.sessions.clone(), + video: self.video.clone(), role, rendition: (role == Role::Canvas).then_some(number), zero_copy: exporter.route() == Route::Gpu, @@ -1481,6 +1500,7 @@ impl Renditions { rungs.push(Rung { role, gate, + grant, scaler, exporter, frames, @@ -1514,16 +1534,17 @@ struct Probed { impl Probe { /// Open a throwaway encoder of each of `canvas`'s renditions on what `gpu` - /// exports, which also settles whether the encoder reads the program on - /// the GPU. The canvas size picks the encoder, which the others use too. - async fn new(gpu: &Gpu, canvas: Canvas) -> anyhow::Result { + /// 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. + 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([(Role::Record, size)]); + .chain(record.then_some((Role::Record, size))); let mut kind = moq_video::encode::Kind::Auto; let mut rungs = Vec::new(); let mut refused = Vec::new(); @@ -1559,6 +1580,7 @@ impl Probe { fn spawn( gpu: Gpu, canvas: Canvas, + record: bool, ending: Vec>, ) -> std::io::Result>> { std::thread::Builder::new().name("probe".into()).spawn(move || { @@ -1567,11 +1589,269 @@ impl Probe { let _ = thread.join(); } let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build()?; - runtime.block_on(Self::new(&gpu, canvas)) + runtime.block_on(Self::new(&gpu, canvas, record)) }) } } +/// One canvas's program video as the program clock feeds it: the renditions +/// of the canvas it last drew, the probe opening the next, and the encoders +/// ending. +struct Line { + renditions: Renditions, + ladder: Option, + /// The canvas `ladder` is for, or will be once probed. + canvas: Canvas, + /// One probe at a time, with the canvas it is for, since a GPU caps its + /// encode sessions. + probing: Option<(Canvas, std::thread::JoinHandle>)>, + /// The encode threads of renditions ended or failed since, whose sessions + /// the next probe waits to be closed. + ending: Vec>, + /// Whether the canvas drawn has its renditions, or failed to open them, so + /// a failure is not retried until the canvas changes. + settled: bool, +} + +/// What a [`Line`] leaves running once ended, to wait on. +struct Ended { + threads: Vec>, + /// Holds the encode threads it was waiting on. + probe: Option>>, +} + +impl Ended { + fn is_finished(&self) -> bool { + self.threads.iter().all(std::thread::JoinHandle::is_finished) + && self.probe.as_ref().is_none_or(std::thread::JoinHandle::is_finished) + } + + fn join(self) { + for thread in self.threads { + if thread.join().is_err() { + tracing::error!("an encode thread panicked"); + } + } + if let Some(probe) = self.probe { + let _ = probe.join(); + } + } +} + +impl Line { + /// `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. + fn new(renditions: Renditions, canvas: Canvas, ladder: Option, settled: bool) -> Self { + if let Some(ladder) = &ladder { + ladder.adopted(&renditions.video); + } + Self { + renditions, + ladder, + canvas, + probing: None, + ending: Vec::new(), + settled, + } + } + + /// Hand `program`, drawn at `drawn`, to each rendition its gate lets + /// encode, stamped `timestamp`. A new canvas size or frame rate ends the + /// renditions and probes for others, since a track never changes what it + /// carries. + fn feed(&mut self, program: &wgpu::Texture, drawn: Canvas, timestamp: Option) { + let renditions = &self.renditions; + let video = &renditions.video; + if drawn != self.canvas { + self.canvas = drawn; + self.settled = false; + // Ended before the next open, so the two never hold encode sessions + // at once. + self.ending + .extend(self.ladder.take().into_iter().flat_map(|ladder| ladder.end(renditions))); + video.retire_encoder(); + } + // Probed beside the program clock, so drawing carries on. + if let Some((probed, probe)) = self.probing.take_if(|(_, probe)| probe.is_finished()) { + let probe = probe + .join() + .unwrap_or_else(|_| Err(anyhow::anyhow!("the encoder probe panicked"))); + // One for a canvas since left is dropped unlisted. + if probed == self.canvas { + self.settled = true; + match probe.and_then(|probe| renditions.start(probe)) { + Ok((started, _)) => { + started.adopted(video); + self.ladder = Some(started); + } + Err(err) => video.fail(format!("{err:#}")), + } + } + } + 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) { + Ok(probe) => self.probing = Some((self.canvas, probe)), + Err(err) => { + self.settled = true; + video.fail(format!("cannot start an encoder probe: {err}")); + } + } + } + + let Some(current) = &mut self.ladder else { + video.watched.store(false, Ordering::Relaxed); + video.encoding.store(false, Ordering::Relaxed); + return; + }; + let watched = current + .rungs + .iter() + .any(|rung| rung.role.published() && rung.gate.watched()); + video.watched.store(watched, Ordering::Relaxed); + let Some(timestamp) = timestamp else { return }; + let mut encoding = false; + let mut failed = Vec::new(); + for (index, rung) in current.rungs.iter_mut().enumerate() { + let role = rung.role; + match rung.feed(&video.renditions[role as usize], program, timestamp) { + Some(step) => encoding |= role == Role::Canvas && matches!(step, Step::Encode(_)), + None => failed.push((index, role)), + } + } + video.encoding.store(encoding, Ordering::Relaxed); + // An encoder stops only having reported why. Without the canvas size + // the video stops; without another rendition it carries on. + if failed.iter().any(|(_, role)| *role == Role::Canvas) { + self.ending + .extend(self.ladder.take().into_iter().flat_map(|ladder| ladder.end(renditions))); + } else { + for (index, _) in failed.into_iter().rev() { + self.ending.push(current.remove(index, renditions)); + } + } + } + + /// Stop handing the renditions frames, so each encode thread publishes + /// what it holds and finishes its track. + fn end(mut self) -> Ended { + let mut threads = std::mem::take(&mut self.ending); + threads.extend( + self.ladder + .take() + .into_iter() + .flat_map(|ladder| ladder.end(&self.renditions)), + ); + self.renditions.video.watched.store(false, Ordering::Relaxed); + self.renditions.video.encoding.store(false, Ordering::Relaxed); + Ended { + threads, + probe: self.probing.map(|(_, probe)| probe), + } + } +} + +/// A further canvas's broadcast, closed 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, +} + +impl Drop for Broadcast { + fn drop(&mut self) { + self.publication.close(); + } +} + +/// What starts each further canvas's broadcast: beneath the main one's path, +/// on its clock, sharing its connection's bandwidth. +struct Extras { + origin: moq_net::origin::Producer, + path: moq_net::Path, + clock: moq_mux::Clock, + /// The main broadcast's, which its audio is listed by. + instance: moq_net::Instance, + gpu: Gpu, + bandwidth: bandwidth::Allocator, + on_demand: bool, + live: bool, + stats: Arc, +} + +impl Extras { + /// Publish the further canvas `name`, first drawn at `canvas`, at its own + /// path beneath the main broadcast's, listing the main broadcast's audio + /// by reference so it is encoded once, and the [`Line`] that encodes its + /// video, reporting in `video`. + fn start( + &self, + name: &str, + canvas: Canvas, + video: Arc, + program: &Program, + ) -> anyhow::Result<(Line, Broadcast)> { + let path = moq_net::Path::new(name) + .and_then(|name| self.path.join(&name)) + .with_context(|| format!("cannot publish canvas `{name}`"))?; + if self.live { + let (_, echo) = health::echo(&self.path)?; + anyhow::ensure!( + path != echo, + "canvas `{name}` would publish where viewers report back; rename it" + ); + } + let last = self.path.parts().last().context("the broadcast has no path")?; + let audio = moq_net::path::Relative::new(&format!("../{last}"))?; + anyhow::ensure!( + path.try_resolve(&audio).as_ref() == Some(&self.path), + "canvas `{name}` cannot refer to the main broadcast" + ); + + let publication = self + .origin + .create_broadcast(&path) + .with_context(|| format!("cannot create the broadcast of canvas `{name}`"))?; + let mut broadcast = publication.broadcast().clone(); + let config = moq_mux::catalog::Config::default() + .with_catalog(Default::default()) + .with_clock(self.clock.clone()); + let mut catalog = moq_scene::publish::Catalog::new(&mut broadcast, config)?; + let listed = program.catalog.snapshot().audio.renditions; + catalog.mutate(|catalog| { + for (id, mut config) in listed { + config.broadcast = Some(audio.clone()); + config.instance = Some(self.instance.clone()); + config.track.get_or_insert_with(|| id.clone()); + catalog.audio.renditions.insert(id, config); + } + })?; + publication + .announce(moq_net::origin::Route::default().with_instance(moq_net::Instance::random(), 0)) + .with_context(|| format!("cannot announce the broadcast of canvas `{name}`"))?; + + let renditions = Renditions { + gpu: self.gpu.clone(), + broadcast, + timebase: catalog.timebase(), + recording: None, + bandwidth: self.bandwidth.clone(), + on_demand: self.on_demand, + live: self.live, + stats: self.stats.clone(), + video, + grants: format!("{name}/"), + }; + let broadcast = Broadcast { + publication, + _catalog: catalog, + }; + Ok((Line::new(renditions, canvas, None, false), broadcast)) + } +} + /// The program video of one canvas: each of its renditions, the canvas size /// first. struct Ladder { @@ -1579,28 +1859,30 @@ struct Ladder { } impl Ladder { - /// Report these renditions as the program's from now on. - fn adopted(&self, stats: &Stats) { + /// Report these renditions as the canvas's from now on. + fn adopted(&self, video: &VideoStats) { let zero_copy = self.rungs[0].exporter.route() == Route::Gpu; - stats.zero_copy.store(zero_copy, Ordering::Relaxed); - self.count(stats); + video.zero_copy.store(zero_copy, Ordering::Relaxed); + self.count(video); } - /// Report how many sizes the program publishes. - fn count(&self, stats: &Stats) { + /// Report how many sizes the canvas publishes. + fn count(&self, video: &VideoStats) { let sizes = self.rungs.iter().filter(|rung| rung.role.published()).count(); - stats.sizes.store(sizes, Ordering::Relaxed); + video.sizes.store(sizes, Ordering::Relaxed); } /// Stop handing the renditions frames, so each encode thread encodes those /// it holds, finishes its track, and returns. - fn end(self, stats: &Stats) -> Vec> { - stats.sizes.store(0, Ordering::Relaxed); - let mut grants = stats.grants.lock().unwrap(); + fn end(self, renditions: &Renditions) -> Vec> { + renditions.video.sizes.store(0, Ordering::Relaxed); + let mut grants = renditions.stats.grants.lock().unwrap(); self.rungs .into_iter() .map(|rung| { - grants.remove(rung.gate.track()); + if let Some(grant) = &rung.grant { + grants.remove(grant); + } rung.thread }) .collect() @@ -1609,10 +1891,12 @@ impl Ladder { /// Stop handing rendition `index` frames, as [`end`](Self::end) does. /// The canvas size left alone no longer disables below the half size, /// since no viewer has a smaller one to move to. - fn remove(&mut self, index: usize, stats: &Stats) -> std::thread::JoinHandle<()> { + fn remove(&mut self, index: usize, renditions: &Renditions) -> std::thread::JoinHandle<()> { let rung = self.rungs.remove(index); - stats.grants.lock().unwrap().remove(rung.gate.track()); - self.count(stats); + if let Some(grant) = &rung.grant { + renditions.stats.grants.lock().unwrap().remove(grant); + } + self.count(&renditions.video); if rung.role == Role::Half { for canvas in self.rungs.iter_mut().filter(|rung| rung.role == Role::Canvas) { canvas.gate.alone(); @@ -1627,6 +1911,8 @@ impl Ladder { struct Rung { role: Role, gate: Gate, + /// Its grant's name in [`Stats::grants`]: a published rendition's alone. + grant: Option, /// Draws the program at the rendition's size; `None` at the canvas size. scaler: Option, exporter: Exporter, @@ -1675,15 +1961,15 @@ impl Rung { } } -/// How often the program is drawn at `canvas`'s frame rate, from `start`. -fn program_clock(start: Instant, canvas: Canvas) -> kio::clock::Interval { - let mut clock = kio::clock::interval_at(start, period(canvas)); +/// How often the program is drawn at `framerate`, from `start`. +fn program_clock(start: Instant, framerate: u32) -> kio::clock::Interval { + let mut clock = kio::clock::interval_at(start, period(framerate)); clock.set_missed_tick_behavior(kio::clock::MissedTickBehavior::Skip); clock } -fn period(canvas: Canvas) -> Duration { - Duration::from_secs(1) / canvas.framerate +fn period(framerate: u32) -> Duration { + Duration::from_secs(1) / framerate } /// Connect to every relay, publishing `origin` to each and feeding `subscribe` @@ -1745,7 +2031,9 @@ enum Video { /// What an encode thread runs. struct Job { - stats: Arc, + sessions: Sessions, + /// How its canvas's video is doing. + video: Arc, /// Which of the program's renditions it is. role: Role, /// The number of the program's renditions to report the encoder under: @@ -1766,9 +2054,9 @@ struct Encoder { track: Encoding, config: moq_video::encode::Config, joins: Joins, - /// Where to report each encoder opened as the program's, under the - /// number of the program's renditions: the canvas size's alone. - report: Option<(Arc, u64)>, + /// Where to report each encoder opened as its canvas's, under the + /// number of the canvas's renditions: the canvas size's alone. + report: Option<(Arc, u64)>, /// Keep the encoder, and its session, while skipping frames. hold: bool, /// Whether the open encoder can cut a keyframe for a joining viewer. @@ -1871,7 +2159,8 @@ impl Encoder { /// track that went away unfinished. fn encode(job: Job, opened: Option>>, video: mpsc::Receiver