diff --git a/README.md b/README.md index 0cf243c..39f56b5 100644 --- a/README.md +++ b/README.md @@ -13,8 +13,9 @@ cargo run --release -p moq-studio --features capture,openh264 -- \ show.toml --connect https://cdn.moq.dev/anon --broadcast my-show ``` -- `show.toml` is the scene collection. It is created if it does not exist, and - changes are saved to it a second after you stop editing, with no save step. +- `show.toml` is the scene collection. If it does not exist yet, + [guided setup](#first-launch) proposes one and creates it. Changes are saved + to it a second after you stop editing, with no save step. - Without `--connect`, the studio composes and previews but publishes nothing, and broadcast sources stay offline. - `capture` adds camera, display, window, microphone, and system-audio sources. @@ -30,6 +31,44 @@ cargo run --release -p moq-studio --features capture,openh264 -- \ - The `--connect-*`, `--quic-*`, and `--log-level` flags are [moq-cli's](https://doc.moq.dev/bin/cli); see `moq-studio --help`. +## First launch + +Opening a scene file that does not exist yet starts guided setup instead of +the studio. It runs three checks, and a check that fails shows its reason: + +- **Devices:** the cameras, displays, and audio inputs this machine has. +- **Encoders:** for each canvas setup can propose (1080p and 720p at 60 or 30 + fps, and 360p at 30 fps), the encoder the program would pick, whether it + reads the GPU, the bitrate it encodes at, and how long a frame takes over a + short burst of moving frames. An encoder keeps up when a frame takes at most + three quarters of the frame period, leaving the rest for sources and the + compositor. +- **Uplink:** with `--connect`, setup publishes a probe at the `--broadcast` + path for about six seconds and subscribes to it through the relay, so the + relay pulls it. The congestion controller's send-rate estimate is the + measurement. A connection that reports no estimate, such as WebSocket, fails + the check. + +Setup then proposes a scene named Main and a canvas, explaining each choice. +The scene has the first display filling the canvas, the first camera in the +bottom-right corner over it (or filling the canvas alone), and the first audio +input. Swap any of them for another device or leave it out. The canvas is the +largest one whose encoder keeps up and whose video and audio fit in two thirds +of the measured uplink, the most the Health panel grades healthy. Without a +measurement, it is the default 1080p at 30 fps. If no encoder keeps up, it is +the canvas closest to keeping up. + +Nothing is written until you start the studio, so quitting during setup runs +it again on the next launch. **Start with this scene** and **Start with an +empty scene** both create the scene file at once, the second with the chosen +canvas and no sources. The proposed scene is the first edit in the history, so +undoing it leaves that empty scene. + +Setup writes an ordinary scene file. Change its sources and scenes in their +panels, and its canvas in the file, which takes a restart. The Output panel +shows the canvas in use, and the Health panel shows how the uplink holds up +once you are live. + ## Panels Every panel is a tab you can drag, split, or stack. @@ -44,7 +83,7 @@ Every panel is a tab you can drag, split, or stack. | Devices | List this machine's cameras, displays, windows, and audio inputs, and add one as a source. | | Filters | Add, order, and set the filters of the selected source or item, and copy or paste them. | | Audio | Set each audio source's gain and mute, and watch its meter and the program's. | -| Output | See the broadcast path, whether the program is encoding, and the encoder's work. | +| Output | See the broadcast path, whether the program is encoding, the canvas, and the encoder's work. | | Health | See how the stream reaches viewers, track by track, and the cause of each problem. | A new source, from the Sources or Devices panel, is placed over whatever the diff --git a/justfile b/justfile index 2bd9ae7..a40fff9 100644 --- a/justfile +++ b/justfile @@ -48,7 +48,7 @@ render-sw: icd=("$mesa"/share/vulkan/icd.d/lvp_icd.*.json) export LD_LIBRARY_PATH="$loader/lib" VK_DRIVER_FILES="${icd[0]}" cargo nextest run --locked -p moq-scene --lib --run-ignored only -E 'test(/^compositor::tests::/)' - cargo nextest run --locked -p moq-studio --features openh264 --run-ignored only -E 'test(=engine::tests::the_program_reaches_a_relay)' + cargo nextest run --locked -p moq-studio --features openh264 --run-ignored only -E 'test(=engine::tests::the_program_reaches_a_relay) | test(=setup::tests::times_every_canvas)' # Move the moq submodule to moq's latest main (or a given commit) and refresh Cargo.lock. upgrade-moq rev="origin/main": diff --git a/quest/m0/README.md b/quest/m0/README.md index 842be04..813af37 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -96,7 +96,6 @@ repository. - [Animation and transitions](/quest/m0/animation.md) - move, slide, wipe, and stinger transitions, and keyframed item properties - [Video effects](/quest/m0/effects.md) - a curated effect set with CSS meanings, and ML effects baked into their source - [Snapping and guides](/quest/m0/snapping.md) - preview moves snap to edges, centers, and other items -- [Guided setup](/quest/m0/guided-setup.md) - first launch detects devices, measures the uplink, and proposes settings - [Destinations](/quest/m0/destinations.md) - RTMP, RTMPS, and WHIP push with per-destination codecs and shared encoders, on a UI-free egress library a hosted service can reuse - [Multiple canvases](/quest/m0/multi-canvas.md) - landscape and portrait canvases composite at once from shared sources - [Guests](/quest/m0/guests.md) - remote guests and phone cameras join through a moq-room link or QR code and appear as sources diff --git a/quest/m0/guided-setup.md b/quest/m0/guided-setup.md deleted file mode 100644 index ced4e99..0000000 --- a/quest/m0/guided-setup.md +++ /dev/null @@ -1,19 +0,0 @@ -# [M] Guided setup - -## Goal - -The first launch of `moq-studio` gets a creator live without reading -settings: it detects cameras, microphones, and displays, measures the uplink, -probes which encoders work on this machine, and proposes a scene and output -settings in plain language. Every setting it chooses can be changed later and -explains itself. - -## Plan - -Twitch Studio did this before it was shut down, and "settings are too -complicated" is a top OBS complaint. Reuse the probes that already exist -(the encoder choice `moq_video::render::export` makes at setup, the bandwidth -grant) rather than a separate benchmark. The grant alone does not measure -the uplink, so setup runs a short probe publish to the configured relay and -reads the congestion controller's bandwidth estimate. A probe that fails is -reported, not hidden. diff --git a/rs/moq-studio/src/autosave.rs b/rs/moq-studio/src/autosave.rs index 6adde96..9a1c7c0 100644 --- a/rs/moq-studio/src/autosave.rs +++ b/rs/moq-studio/src/autosave.rs @@ -118,7 +118,8 @@ struct Written { impl Autosave { /// Read the scene file at `path` and its history, or start a new - /// collection if there is no scene file yet; the first save creates it. + /// collection if there is no scene file yet; the first save creates it, and + /// fails rather than replace one another program created meanwhile. pub fn open(path: PathBuf) -> anyhow::Result<(Self, Loaded)> { let (collection, written) = match std::fs::read_to_string(&path) { Ok(text) => { @@ -151,6 +152,12 @@ impl Autosave { &self.path } + /// Whether the scene file exists: it was read when opened, or a save has + /// created it since. + pub fn exists(&self) -> bool { + self.written.lock().unwrap().collection.is_some() + } + /// Where the studio records its edits to the collection. pub fn edits(&self) -> Edits { self.edits.clone() @@ -171,11 +178,13 @@ impl Autosave { self.edits.0.lock().take(); let result = serialize().and_then(|text| { if written.collection.as_ref() != Some(&text.collection) { - write(&self.path, &text.collection).with_context(|| format!("writing {}", self.path.display()))?; + let new = written.collection.is_none(); + write(&self.path, &text.collection, new).with_context(|| format!("writing {}", self.path.display()))?; written.collection = Some(text.collection); } if written.history != text.history { - write(&self.history, &text.history).with_context(|| format!("writing {}", self.history.display()))?; + write(&self.history, &text.history, false) + .with_context(|| format!("writing {}", self.history.display()))?; written.history = text.history; } Ok(()) @@ -263,9 +272,13 @@ fn load_history(path: &Path, collection: Collection) -> anyhow::Result { /// Replace the file at `path` with `text` all at once, so a crash leaves /// either the old file or the new one, never part of either. -fn write(path: &Path, text: &str) -> std::io::Result<()> { +/// +/// A `new` file must not exist yet: one another program created since the +/// studio found none is left alone rather than replaced. +fn write(path: &Path, text: &str, new: bool) -> std::io::Result<()> { let path = resolve(path)?; let existing = match std::fs::metadata(&path) { + Ok(_) if new => return Err(created_since()), Ok(metadata) => { // Refused as a write in place would be, so a read-only scene file // is not replaced. @@ -302,13 +315,27 @@ fn write(path: &Path, text: &str) -> std::io::Result<()> { // On disk before the rename, so a power loss cannot leave the new name // pointing at an empty file. file.as_file().sync_all()?; - file.persist(&path)?; + match new { + // Fails if the file appeared since the check above. + true => file.persist_noclobber(&path).map_err(|err| match err.error.kind() { + std::io::ErrorKind::AlreadyExists => created_since(), + _ => err.error, + })?, + false => file.persist(&path)?, + }; // And the rename itself, so a power loss cannot undo the save. #[cfg(unix)] std::fs::File::open(dir)?.sync_all()?; Ok(()) } +fn created_since() -> std::io::Error { + std::io::Error::new( + std::io::ErrorKind::AlreadyExists, + "another program created it since the studio started; restart the studio to open it", + ) +} + /// The file a chain of symlinks at `path` ends in, which need not exist yet, /// so a save replaces that file and every link survives. fn resolve(path: &Path) -> std::io::Result { @@ -391,6 +418,21 @@ mod tests { ); } + /// A scene file another program creates after the studio found none, such + /// as a second studio finishing setup first, is never replaced. + #[test] + fn leaves_a_file_created_since_it_found_none() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("show.toml"); + let (autosave, Loaded { collection: opened, .. }) = Autosave::open(path.clone()).unwrap(); + + std::fs::write(&path, "elsewhere").unwrap(); + let err = format!("{:#}", autosave.save(serialize(&opened)).unwrap_err()); + assert!(err.contains("restart the studio"), "{err}"); + assert_eq!(std::fs::read_to_string(&path).unwrap(), "elsewhere"); + assert!(autosave.error().is_some(), "shown until a restart"); + } + /// A new file gets the permissions any new file would, not a temporary /// file's owner-only ones. #[cfg(unix)] diff --git a/rs/moq-studio/src/devices.rs b/rs/moq-studio/src/devices.rs new file mode 100644 index 0000000..faa9f41 --- /dev/null +++ b/rs/moq-studio/src/devices.rs @@ -0,0 +1,83 @@ +//! The capture devices this machine has, for the Devices panel and setup. + +use moq_scene::Source; + +/// A device the user can add as a source. +#[derive(Clone, Debug)] +pub struct Pick { + pub name: String, + pub source: Source, +} + +/// One listing of every kind of device, each of which can fail on its own. +pub struct Listing { + pub cameras: Result, String>, + pub displays: Result, String>, + pub windows: Result, String>, + /// Microphones and other inputs, plus system audio where it can be captured. + pub audio: Result, String>, +} + +impl Listing { + /// Each kind with its heading, in the order the Devices panel shows them. + pub fn kinds(&mut self) -> [(&'static str, &mut Result, String>); 4] { + [ + ("Cameras", &mut self.cameras), + ("Displays", &mut self.displays), + ("Windows", &mut self.windows), + ("Audio inputs", &mut self.audio), + ] + } +} + +#[cfg(feature = "capture")] +pub async fn list() -> Listing { + fn picks(result: Result, impl std::fmt::Display>, pick: impl Fn(T) -> Pick) -> Result, String> { + result + .map(|devices| devices.into_iter().map(pick).collect()) + .map_err(|err| err.to_string()) + } + + let cameras = picks(moq_video::capture::cameras().await, |camera| Pick { + name: camera.name, + source: Source::Camera { id: Some(camera.id) }, + }); + let displays = picks(moq_video::capture::displays().await, |display| Pick { + name: display.name, + source: Source::Display { id: Some(display.id) }, + }); + let windows = picks(moq_video::capture::windows().await, |window| Pick { + name: format!("{} ({})", window.title, window.app), + source: Source::Window { id: window.id }, + }); + let mut audio = picks(moq_audio::capture::devices().await, |device| Pick { + name: format!("{} ({})", device.name, device.host), + source: Source::Microphone { id: Some(device.id) }, + }); + if cfg!(target_os = "macos") + && let Ok(audio) = &mut audio + { + audio.push(Pick { + name: "System audio".into(), + source: Source::SystemAudio, + }); + } + + Listing { + cameras, + displays, + windows, + audio, + } +} + +#[cfg(not(feature = "capture"))] +pub async fn list() -> Listing { + let unsupported = || Err("this build cannot capture; rebuild with `--features capture`".to_string()); + Listing { + cameras: unsupported(), + displays: unsupported(), + windows: unsupported(), + audio: unsupported(), + } +} diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index da8ab11..0f1ad79 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -229,27 +229,41 @@ pub struct Engine { } /// The audio the program publishes. -const AUDIO_BITRATE: moq_net::bandwidth::Rate = moq_net::bandwidth::Rate::from_kbps(128); +pub const AUDIO_BITRATE: moq_net::bandwidth::Rate = moq_net::bandwidth::Rate::from_kbps(128); /// How often the audio thread reads the bus. Short enough that a read is never /// close to the bus's buffering, long enough to cost nothing. const AUDIO_PERIOD: Duration = Duration::from_millis(10); +/// Everything starting the engine takes besides the scene file, which guided +/// setup holds back until the creator has settled on a collection. +#[derive(Clone)] +pub struct Launch { + pub gpu: Gpu, + /// The relay to publish to, if any. + pub connect: Option<(moq_tokio::Client, url::Url)>, + /// The broadcast path the program is published at. + pub broadcast: String, + /// Whether the program video encodes only while someone watches. + pub on_demand: bool, +} + impl Engine { - /// Open the `loaded` collection and history on `gpu`, start the program - /// clock, and save both to `autosave`'s files as they change. - /// With `connect`, publish the program at `broadcast` on that relay, - /// encoding its video only while someone watches when `on_demand`. + /// Open the `loaded` collection and history on `launch.gpu`, start the + /// program clock, and save both to `autosave`'s files as they change, + /// starting at once. + /// With `launch.connect`, publish the program at `launch.broadcast` on that + /// relay, encoding its video only while someone watches when + /// `launch.on_demand`. /// /// Call inside the tokio runtime that serves the relay connection. - pub async fn start( - gpu: Gpu, - autosave: Autosave, - loaded: autosave::Loaded, - connect: Option<(moq_tokio::Client, url::Url)>, - broadcast: &str, - on_demand: bool, - ) -> anyhow::Result> { + pub async fn start(launch: Launch, autosave: Autosave, loaded: autosave::Loaded) -> anyhow::Result> { + let Launch { + gpu, + connect, + broadcast, + on_demand, + } = launch; let bus = moq_audio::bus::Bus::new(Default::default())?; let subscribe = moq_tokio::origin::spawn(); let sources = subscribe.consume(); @@ -262,7 +276,7 @@ impl Engine { let (output, origin, publish) = match connect { Some((client, url)) => { - let (output, publish) = publish(&gpu, client, url, subscribe, broadcast, canvas, on_demand).await?; + let (output, publish) = publish(&gpu, client, url, subscribe, &broadcast, canvas, on_demand).await?; (Some(output), None, Some(publish)) } None => (None, Some(subscribe), None), @@ -859,16 +873,19 @@ mod tests { filters: Vec::new(), }); - let gpu = Gpu::new().await.expect("a GPU"); + let launch = Launch { + gpu: Gpu::new().await.expect("a GPU"), + connect: Some((client, fixture.url.clone())), + broadcast: "studio".into(), + on_demand: false, + }; let (autosave, _) = Autosave::open(dir.path().join("show.toml")).unwrap(); - let connect = Some((client, fixture.url.clone())); - let on_demand = false; let loaded = autosave::Loaded { collection, history: History::new(), dropped: 0, }; - let engine = Engine::start(gpu, autosave, loaded, connect, "studio", on_demand) + let engine = Engine::start(launch, autosave, loaded) .await .expect("the studio starts"); assert!( diff --git a/rs/moq-studio/src/main.rs b/rs/moq-studio/src/main.rs index 2f8e4e8..6afdca5 100644 --- a/rs/moq-studio/src/main.rs +++ b/rs/moq-studio/src/main.rs @@ -5,21 +5,25 @@ //! window around them and the program clock that publishes the result. use std::path::PathBuf; -use std::sync::Arc; +use std::sync::{Arc, OnceLock}; use anyhow::Context; mod autosave; +mod devices; mod engine; mod health; +mod setup; mod ui; +mod uplink; #[derive(usage_rs::Cli, Clone)] #[usage(unknown_flags = "error", args_override_self = false)] #[usage(name = "moq-studio")] #[usage(completion, settings)] pub struct Config { - /// The scene collection to open, saved as it changes. Created if it does not exist. + /// The scene collection to open, saved as it changes. If it does not exist, + /// guided setup proposes one and creates it. #[usage(value_hint = usage_rs::ValueHint::FilePath, extensions("toml"))] pub scenes: PathBuf, @@ -67,21 +71,37 @@ fn main() -> anyhow::Result<()> { // The QUIC client binds its socket on the current runtime. let _runtime = runtime.enter(); let gpu = pollster::block_on(engine::Gpu::new())?; - let (autosave, loaded) = autosave::Autosave::open(config.scenes.clone())?; let connect = match config.client.url.clone() { Some(url) => Some((config.client.clone().init(config.quic.clone())?, url)), None => None, }; - let engine = runtime.block_on(engine::Engine::start( - gpu.clone(), - autosave, - loaded, + let launch = engine::Launch { + gpu: gpu.clone(), connect, - &config.broadcast, - config.on_demand, - ))?; + broadcast: config.broadcast.clone(), + on_demand: config.on_demand, + }; + let exists = config + .scenes + .try_exists() + .with_context(|| format!("reading {}", config.scenes.display()))?; + let app = match exists { + true => { + let (autosave, loaded) = autosave::Autosave::open(config.scenes.clone())?; + App::Starting(runtime.block_on(engine::Engine::start(launch, autosave, loaded))?) + } + // Nothing is saved until setup starts the studio, so quitting before then + // leaves no file and the next launch runs setup again. + false => App::Setup(Box::new(setup::Setup::new( + launch, + config.scenes.clone(), + runtime.handle().clone(), + )?)), + }; - let ran = run(engine.clone(), gpu, runtime.handle().clone()); + let started = Arc::new(OnceLock::new()); + let ran = run(app, gpu, runtime.handle().clone(), started.clone()); + let Some(engine) = started.get() else { return ran }; // The edits since the last autosave, so quitting loses none. let saved = engine.save(); if let (Err(err), Err(_)) = (&saved, &ran) { @@ -91,8 +111,48 @@ fn main() -> anyhow::Result<()> { saved } +/// What the window shows. +enum App { + /// First launch: guided setup, until it starts the studio. + Setup(Box), + /// An engine whose studio is built once the window can draw it. + Starting(Arc), + Studio(Box), +} + +/// The window's contents, and what building the studio takes. +struct Window { + app: App, + render: eframe::egui_wgpu::RenderState, + runtime: tokio::runtime::Handle, + /// The engine once the studio is built, for the save on quit. + started: Arc>>, +} + +impl eframe::App for Window { + fn ui(&mut self, ui: &mut egui::Ui, frame: &mut eframe::Frame) { + let engine = match &mut self.app { + App::Studio(studio) => return studio.ui(ui, frame), + App::Setup(setup) => match setup.ui(ui) { + Some(engine) => engine, + None => return, + }, + App::Starting(engine) => engine.clone(), + }; + let _ = self.started.set(engine.clone()); + let studio = ui::Studio::new(&self.render, engine, self.runtime.clone()); + self.app = App::Studio(Box::new(studio)); + ui.ctx().request_repaint(); + } +} + /// Show the window until it is closed. -fn run(engine: Arc, gpu: engine::Gpu, runtime: tokio::runtime::Handle) -> anyhow::Result<()> { +fn run( + app: App, + gpu: engine::Gpu, + runtime: tokio::runtime::Handle, + started: Arc>>, +) -> anyhow::Result<()> { let options = eframe::NativeOptions { viewport: egui::ViewportBuilder::default() .with_title("MoQ Studio") @@ -114,7 +174,18 @@ fn run(engine: Arc, gpu: engine::Gpu, runtime: tokio::runtime::H eframe::run_native( "moq-studio", options, - Box::new(move |cc| Ok(Box::new(ui::Studio::new(cc, engine, runtime)?))), + Box::new(move |cc| { + let render = cc + .wgpu_render_state + .clone() + .context("the window is not drawn with wgpu")?; + Ok(Box::new(Window { + app, + render, + runtime, + started, + })) + }), ) .map_err(|err| anyhow::anyhow!("{err}")) .context("the window failed") diff --git a/rs/moq-studio/src/setup.rs b/rs/moq-studio/src/setup.rs new file mode 100644 index 0000000..103f0be --- /dev/null +++ b/rs/moq-studio/src/setup.rs @@ -0,0 +1,1048 @@ +//! Guided setup: on first launch, find this machine's devices, the encoders +//! that work on it, and the uplink to the relay, then propose a scene and a +//! canvas in plain language before the studio starts. +//! +//! Every probe that fails is shown with its reason, and the proposal says what +//! it assumed instead. + +use std::path::PathBuf; +use std::sync::Arc; +use std::sync::mpsc; +use std::time::Duration; + +use anyhow::Context; +use egui::Ui; +use moq_net::bandwidth::Rate; +use moq_scene::{Canvas, Collection, Draft, Edit, History, Item, Rect, Scene, wgpu}; + +use crate::autosave::{self, Autosave}; +use crate::devices::{self, Listing, Pick}; +use crate::engine::{self, Engine, Launch}; +use crate::health::rate; +use crate::uplink; + +/// The canvases setup can propose, best first. Each encodes at its own default +/// bitrate, so a smaller one is what fits a slower uplink. +const CANVASES: [Canvas; 5] = [ + canvas(1920, 1080, 60), + canvas(1920, 1080, 30), + canvas(1280, 720, 60), + canvas(1280, 720, 30), + canvas(640, 360, 30), +]; + +const fn canvas(width: u32, height: u32, framerate: u32) -> Canvas { + Canvas { + width, + height, + framerate, + } +} + +/// The share of the measured uplink the program may use: as much as keeps the +/// Health panel's headroom grade healthy, so a proposal is not flagged the +/// moment it goes live. The rest absorbs the estimate's error and whatever else +/// shares the connection. +fn headroom() -> f64 { + 1.0 / hang::health::Thresholds::default().transport.headroom.degraded +} + +/// Frames timed per canvas once its encoder opens: enough to average out one +/// slow frame, few enough that setup stays quick. +const BURST: u32 = 15; + +/// The share of each frame's period that copying and encoding it may take, so +/// the encoder still keeps up while sources and the compositor share the machine. +const PACE: f64 = 0.75; + +/// How far the timed picture pans each frame, so the encoder searches motion as +/// it does for live content instead of skipping a still picture. +const PAN: u32 = 4; + +/// The scene setup proposes. +const SCENE: &str = "Main"; + +/// How one canvas encodes on this machine, as the program's own encoder choice +/// settles it. +#[derive(Clone, Debug, PartialEq)] +pub struct Encoding { + /// The encoder backend, such as `nvenc` or `openh264`. + pub encoder: String, + /// Whether it reads the canvas on the GPU rather than a copy on the CPU. + pub gpu: bool, + /// The video bitrate it encodes at. + pub bitrate: Rate, + /// How long copying and encoding a frame took, on average over a burst. + pub frame: Duration, +} + +impl Encoding { + /// What the program sends: this video plus the audio. + fn total(&self) -> Rate { + Rate::from_bps(self.bitrate.as_bps() + engine::AUDIO_BITRATE.as_bps()) + } + + /// Whether the encoder keeps up with `canvas` in real time, within [`PACE`]. + fn keeps_up(&self, canvas: Canvas) -> bool { + self.frame <= budget(canvas) + } + + /// How many times its [`budget`] a frame of `canvas` took, scaled to compare + /// canvases without floats. + fn load(&self, canvas: Canvas) -> u128 { + self.frame.as_nanos() * u128::from(canvas.framerate) + } +} + +/// The time a frame of `canvas` may take to copy and encode. +fn budget(canvas: Canvas) -> Duration { + (Duration::from_secs(1) / canvas.framerate).mul_f64(PACE) +} + +/// Each of [`CANVASES`], with how it encodes or why it cannot. +pub type Encoders = Vec<(Canvas, Result)>; + +/// The canvases setup may propose: those whose encoder keeps up, or the one +/// closest to keeping up when none does, best first. +fn candidates(encoders: &Encoders) -> Vec<(Canvas, &Encoding)> { + let encodable = encoders + .iter() + .filter_map(|(canvas, result)| Some((*canvas, result.as_ref().ok()?))); + let fast: Vec<_> = encodable + .clone() + .filter(|(canvas, encoding)| encoding.keeps_up(*canvas)) + .collect(); + match fast.is_empty() { + true => encodable + .min_by_key(|(canvas, encoding)| encoding.load(*canvas)) + .into_iter() + .collect(), + false => fast, + } +} + +/// Open the exporter the program would for each canvas, the same probe that +/// picks its encoder at startup, and time a burst through what it settled on. +async fn probe_encoders(gpu: engine::Gpu) -> Encoders { + let picture = Picture::new(&gpu); + let mut probed = Vec::new(); + for canvas in CANVASES { + let result = probe_encoder(&gpu, &picture, canvas) + .await + .map_err(|err| format!("{err:#}")); + probed.push((canvas, result)); + } + probed +} + +async fn probe_encoder(gpu: &engine::Gpu, picture: &Picture, canvas: Canvas) -> anyhow::Result { + use moq_video::render::export::{Exporter, Route}; + + let mut config = moq_video::encode::Config::new(canvas.width, canvas.height, canvas.framerate()); + let (mut exporter, rendition) = Exporter::new(&gpu.device, &gpu.queue, &mut config).await?; + let bitrate = rendition.bitrate.context("the encoder reports no bitrate")?; + let encoder = match &config.kind { + moq_video::encode::Kind::Named(name) => name.clone(), + other => format!("{other:?}"), + }; + + // The program's own path: a canvas-sized texture copied through the + // exporter, waited on, and encoded, as its encode thread does. + let started = kio::clock::now(); + let target = gpu.device.create_texture(&wgpu::TextureDescriptor { + label: Some("setup burst"), + size: extent(canvas.width, canvas.height), + mip_level_count: 1, + sample_count: 1, + dimension: wgpu::TextureDimension::D2, + format: wgpu::TextureFormat::Rgba8Unorm, + usage: wgpu::TextureUsages::COPY_SRC | wgpu::TextureUsages::COPY_DST | wgpu::TextureUsages::TEXTURE_BINDING, + view_formats: &[], + }); + let mut session = moq_video::encode::Encoder::new(&config)?; + let period = Duration::from_secs(1) / canvas.framerate; + let frame = time(budget(canvas), async |index| { + picture.pan(gpu, &target, index); + let timestamp = moq_net::Timestamp::try_from(period * index)?; + let pending = match exporter.copy(&target, timestamp)? { + Some(pending) => pending, + // Fed back to back, as the program never feeds it, a pipelined + // encoder can hold every slot; drained, it gives them back. + None => { + session.flush()?; + exporter + .copy(&target, timestamp)? + .context("the encoder holds every frame")? + } + }; + session.encode(&pending.wait()?)?; + // So the time covers the frames a pipelined encoder still holds. + if index == BURST { + session.flush()?; + } + Ok(()) + }) + .await?; + tracing::info!( + canvas = describe(canvas), + encoder, + ?frame, + burst = ?started.elapsed(), + "timed the encoder" + ); + + Ok(Encoding { + encoder, + gpu: exporter.route() == Route::Gpu, + bitrate: Rate::from_bps(bitrate), + frame, + }) +} + +/// How long a frame takes through `encode`, on average over [`BURST`] frames, +/// calling it with each index from 0 through [`BURST`]. +/// +/// Frame 0 is not timed: it is a keyframe, and the encoder's first frame. +/// The burst stops as soon as it is over `budget` a frame in total, since its +/// average can no longer come in under it, so a slow encoder costs no more +/// time than a fast one. +async fn time(budget: Duration, mut encode: impl AsyncFnMut(u32) -> anyhow::Result<()>) -> anyhow::Result { + encode(0).await?; + let started = kio::clock::now(); + for index in 1..=BURST { + encode(index).await?; + let elapsed = started.elapsed(); + if elapsed > budget * BURST { + return Ok(elapsed / index); + } + } + Ok(started.elapsed() / BURST) +} + +/// A picture as large as the largest canvas plus a burst's panning, uploaded +/// once and copied from for every canvas. +struct Picture(wgpu::Texture); + +impl Picture { + fn new(gpu: &engine::Gpu) -> Self { + let width = CANVASES.iter().map(|canvas| canvas.width).max().unwrap_or_default() + PAN * BURST; + let height = CANVASES.iter().map(|canvas| canvas.height).max().unwrap_or_default(); + let texture = gpu.device.create_texture(&wgpu::TextureDescriptor { + label: Some("setup picture"), + size: extent(width, height), + mip_level_count: 1, + sample_count: 1, + dimension: wgpu::TextureDimension::D2, + format: wgpu::TextureFormat::Rgba8Unorm, + usage: wgpu::TextureUsages::COPY_SRC | wgpu::TextureUsages::COPY_DST, + view_formats: &[], + }); + // Gradients crossed with a pattern: detail to search, unlike a flat fill, + // without the cost of noise, which no camera or screen produces. + let pixels: Vec = (0..height) + .flat_map(|y| (0..width).flat_map(move |x| [x as u8, y as u8, (x ^ y) as u8, 0xff])) + .collect(); + gpu.queue.write_texture( + texture.as_image_copy(), + &pixels, + wgpu::TexelCopyBufferLayout { + offset: 0, + bytes_per_row: Some(width * 4), + rows_per_image: Some(height), + }, + extent(width, height), + ); + Self(texture) + } + + /// Draw frame `index` of a burst into `target`: the picture, panned [`PAN`] + /// pixels a frame. + fn pan(&self, gpu: &engine::Gpu, target: &wgpu::Texture, index: u32) { + let mut commands = gpu.device.create_command_encoder(&Default::default()); + commands.copy_texture_to_texture( + wgpu::TexelCopyTextureInfo { + texture: &self.0, + mip_level: 0, + origin: wgpu::Origin3d { + x: index * PAN, + y: 0, + z: 0, + }, + aspect: wgpu::TextureAspect::All, + }, + target.as_image_copy(), + extent(target.width(), target.height()), + ); + gpu.queue.submit([commands.finish()]); + } +} + +fn extent(width: u32, height: u32) -> wgpu::Extent3d { + wgpu::Extent3d { + width, + height, + depth_or_array_layers: 1, + } +} + +/// The canvas to propose: the best candidate whose program fits the uplink's +/// [`headroom`], or the cheapest candidate when none fits. Without a +/// measurement it is the default canvas when that is a candidate. +fn pick(encoders: &Encoders, uplink: Option) -> Canvas { + let candidates = candidates(encoders); + let Some(uplink) = uplink else { + let default = Canvas::default(); + return candidates + .iter() + .map(|(canvas, _)| *canvas) + .find(|canvas| *canvas == default) + .or_else(|| candidates.first().map(|(canvas, _)| *canvas)) + .unwrap_or(default); + }; + let budget = uplink.scaled(headroom()); + candidates + .iter() + .find(|(_, encoding)| encoding.total() <= budget) + .or_else(|| candidates.iter().min_by_key(|(_, encoding)| encoding.total())) + .map(|(canvas, _)| *canvas) + .unwrap_or_default() +} + +/// Why `canvas` suits this machine, in plain words. +fn explain(canvas: Canvas, encoders: &Encoders, uplink: &Result) -> String { + let name = describe(canvas); + let encoding = match encoders.iter().find(|(probed, _)| *probed == canvas) { + Some((_, Ok(encoding))) => encoding, + Some((_, Err(err))) => { + return format!("{name} cannot be encoded on this machine, so publishing it fails: {err}"); + } + None => return format!("{name} was not probed."), + }; + let route = match encoding.gpu { + true => "on the GPU", + false => "from a copy read back to the CPU each frame", + }; + let mut text = format!( + "{name}: {} encodes it {route} at about {}, {} with audio.", + encoding.encoder, + rate(encoding.bitrate), + rate(encoding.total()) + ); + if !encoding.keeps_up(canvas) { + text.push_str(&format!( + " A frame took {}, more than the {} keeping up at {} fps allows, so expect dropped frames.", + millis(encoding.frame), + millis(budget(canvas)), + canvas.framerate + )); + } + let uplink = match uplink { + Ok(uplink) => *uplink, + Err(_) => { + text.push_str(" The uplink was not measured, so check that it carries that."); + return text; + } + }; + let share = encoding.total().as_bps() as f64 / uplink.as_bps().max(1) as f64; + text.push_str(&format!( + " That is {}% of the {} uplink measured to the relay", + (share * 100.0).round(), + rate(uplink) + )); + match share <= headroom() { + true => text.push('.'), + false => text.push_str(&format!( + ", more than the {}% it should use. Expect dropped frames.", + (headroom() * 100.0).round() + )), + } + text +} + +fn describe(canvas: Canvas) -> String { + format!("{}x{} at {} fps", canvas.width, canvas.height, canvas.framerate) +} + +fn millis(duration: Duration) -> String { + format!("{:.1} ms", duration.as_secs_f64() * 1e3) +} + +/// The devices and canvas the creator settles on, each picked by index into +/// the listing. +#[derive(Clone, Debug, Default, PartialEq)] +struct Choices { + display: Option, + camera: Option, + audio: Option, + canvas: Canvas, +} + +/// The first of each kind of device, on `canvas`. +fn propose(listing: &Listing, canvas: Canvas) -> Choices { + let first = |devices: &Result, String>| devices.as_ref().ok().filter(|d| !d.is_empty()).map(|_| 0); + Choices { + display: first(&listing.displays), + camera: first(&listing.cameras), + audio: first(&listing.audio), + canvas, + } +} + +/// What setup starts the studio with: a new collection on the chosen canvas +/// and, unless `empty`, the proposal as its first edit, so the history holds it +/// and undoing it leaves the empty collection. +fn loaded(listing: &Listing, choices: &Choices, empty: bool) -> anyhow::Result { + let mut collection = Collection::new(); + collection.canvas = choices.canvas; + let mut history = History::new(); + if !empty { + let edit = proposal(&collection, listing, choices)?; + collection.apply(&edit)?; + history.record(edit, None); + } + Ok(autosave::Loaded { + collection, + history, + dropped: 0, + }) +} + +/// The edit `choices` describe to the new `collection`: its scene becomes +/// [`SCENE`], with the display filling the canvas and the camera over its +/// corner (or filling the canvas alone), and the audio input is added. +fn proposal(collection: &Collection, listing: &Listing, choices: &Choices) -> Result { + let chosen = |devices: &Result, String>, index: Option| { + devices.as_ref().ok().and_then(|devices| devices.get(index?)).cloned() + }; + let display = chosen(&listing.displays, choices.display); + let camera = chosen(&listing.cameras, choices.camera); + let audio = chosen(&listing.audio, choices.audio); + + let mut draft = collection.draft(format!("Set up scene `{SCENE}`")); + let mut scene = Scene::new(SCENE); + let fill = Rect::fill(&choices.canvas); + let corner = corner(&choices.canvas); + let placed = [ + (display.as_ref(), fill), + (camera.as_ref(), if display.is_some() { corner } else { fill }), + ]; + for (pick, rect) in placed { + let Some(pick) = pick else { continue }; + let name = add(&mut draft, pick)?; + scene.items.push(Item::new(name, rect)); + } + if let Some(pick) = &audio { + add(&mut draft, pick)?; + } + draft.set_scene(0, scene)?; + draft.finish() +} + +/// Add `pick` under its device name, numbered past any source already called that. +fn add(draft: &mut Draft, pick: &Pick) -> Result { + let name = draft.collection().unused_name(&pick.name); + draft.set_source(&name, Some(pick.source.clone()))?; + Ok(name) +} + +/// A quarter of the canvas wide in its bottom-right corner, 16:9 like most +/// cameras, inset by a fortieth of the width. +fn corner(canvas: &Canvas) -> Rect { + let width = canvas.width as f32 / 4.0; + let height = width * 9.0 / 16.0; + let inset = canvas.width as f32 / 40.0; + Rect { + x: canvas.width as f32 - width - inset, + y: canvas.height as f32 - height - inset, + width, + height, + } +} + +/// A result being worked out in the background. +enum Task { + Running(mpsc::Receiver), + Done(T), +} + +impl Task { + /// Run `task` in `runtime` on a thread of its own, since encoder probes and + /// the engine's start block while they open encoders. Not on the runtime's + /// blocking pool, which dropping the runtime waits out, so a probe that never + /// ends, such as one waiting on a camera permission prompt, cannot hold the + /// studio open once its window closes. + fn spawn(runtime: &tokio::runtime::Handle, task: F) -> std::io::Result + where + F: FnOnce() -> Fut + Send + 'static, + Fut: Future, + { + let (send, recv) = mpsc::channel(); + let runtime = runtime.clone(); + std::thread::Builder::new().name("setup".into()).spawn(move || { + let _ = send.send(runtime.block_on(task())); + })?; + Ok(Self::Running(recv)) + } + + /// The result, once it arrived. + /// + /// Panics if the task did, rather than waiting forever on a result that + /// will never come. + fn poll(&mut self) -> Option<&T> { + if let Self::Running(recv) = self { + match recv.try_recv() { + Ok(done) => *self = Self::Done(done), + Err(mpsc::TryRecvError::Empty) => {} + Err(mpsc::TryRecvError::Disconnected) => panic!("a setup task stopped without a result"), + } + } + match self { + Self::Running(_) => None, + Self::Done(done) => Some(done), + } + } +} + +/// The setup screen. +pub struct Setup { + runtime: tokio::runtime::Handle, + launch: Launch, + /// The scene file, which does not exist until setup starts the studio. + path: PathBuf, + devices: Task, + encoders: Task, + /// Started once the encoders are known, since they set what to probe up to. + uplink: Option>>, + /// Filled in once every probe is done, then edited by the creator. + choices: Option, + starting: Option>>>, + /// Why the last start failed. + error: Option, +} + +impl Setup { + /// Start probing for `launch`, to create the scene file at `path`. + pub fn new(launch: Launch, path: PathBuf, runtime: tokio::runtime::Handle) -> std::io::Result { + let devices = Task::spawn(&runtime, devices::list)?; + let gpu = launch.gpu.clone(); + let encoders = Task::spawn(&runtime, move || probe_encoders(gpu))?; + Ok(Self { + runtime, + launch, + path, + devices, + encoders, + uplink: None, + choices: None, + starting: None, + error: None, + }) + } + + /// Draw the screen; the engine once the creator has started the studio. + pub fn ui(&mut self, ui: &mut Ui) -> Option> { + self.advance(); + if self.choices.is_none() || self.starting.is_some() { + ui.ctx().request_repaint_after(Duration::from_millis(100)); + } + if let Some(starting) = &mut self.starting + && starting.poll().is_some() + { + let Some(Task::Done(result)) = self.starting.take() else { + unreachable!("polled done") + }; + match result { + Ok(engine) => return Some(engine), + Err(err) => self.error = Some(format!("{err:#}")), + } + } + + let mut start = None; + egui::CentralPanel::default().show(ui, |ui| { + egui::ScrollArea::vertical().show(ui, |ui| { + ui.heading("Set up MoQ Studio"); + ui.label( + "Setup checks this machine and proposes a first scene and canvas. Change anything here, \ + or later: sources and scenes in their panels, the canvas in the scene file.", + ); + ui.separator(); + self.findings(ui); + ui.separator(); + start = self.proposal(ui); + }); + }); + if let Some(empty) = start { + self.start(empty); + } + None + } + + /// Collect finished probes, start the uplink probe once the encoders are + /// known, and propose once everything is in. + fn advance(&mut self) { + self.devices.poll(); + if self.uplink.is_none() + && let Some(encoders) = self.encoders.poll() + { + let ceiling = candidates(encoders) + .into_iter() + .map(|(_, encoding)| encoding.total()) + .max() + .map(|top| top.scaled(1.0 / headroom())); + let connect = self + .launch + .connect + .clone() + .map(|(client, url)| (client, url, self.launch.broadcast.clone())); + self.uplink = Some(match (connect, ceiling) { + (None, _) => Task::Done(Err("no relay is configured; start with --connect to measure it".into())), + (Some(_), None) => Task::Done(Err("nothing encodes on this machine, so nothing would be sent".into())), + (Some((client, url, path)), Some(ceiling)) => { + let measured = Task::spawn(&self.runtime, move || async move { + uplink::measure(&client, url, &path, ceiling, uplink::WINDOW) + .await + .map_err(|err| format!("{err:#}")) + }); + measured.unwrap_or_else(|err| Task::Done(Err(format!("cannot start the probe: {err}")))) + } + }); + } + if let Some(uplink) = &mut self.uplink { + uplink.poll(); + } + if self.choices.is_none() + && let (Task::Done(listing), Task::Done(encoders), Some(Task::Done(uplink))) = + (&self.devices, &self.encoders, &self.uplink) + { + let canvas = pick(encoders, uplink.as_ref().ok().copied()); + self.choices = Some(propose(listing, canvas)); + } + } + + fn findings(&mut self, ui: &mut Ui) { + ui.strong("This machine"); + egui::Grid::new("findings").num_columns(2).show(ui, |ui| { + ui.label("Devices"); + match &mut self.devices { + Task::Running(_) => { + ui.spinner(); + } + Task::Done(listing) => { + ui.vertical(|ui| { + for (kind, devices) in listing.kinds() { + match devices { + Ok(devices) => ui.label(format!("{kind}: {}", devices.len())), + Err(err) => ui.colored_label(ui.visuals().error_fg_color, format!("{kind}: {err}")), + }; + } + }); + } + } + ui.end_row(); + + ui.label("Encoders"); + match &self.encoders { + Task::Running(_) => { + ui.horizontal(|ui| { + ui.spinner(); + ui.label("timing a short burst through each canvas's encoder"); + }); + } + Task::Done(encoders) => { + ui.vertical(|ui| { + for (canvas, result) in encoders { + match result { + Ok(encoding) => { + let text = format!( + "{}: {}{}, {}, {} a frame", + describe(*canvas), + encoding.encoder, + if encoding.gpu { " on the GPU" } else { " from the CPU" }, + rate(encoding.bitrate), + millis(encoding.frame), + ); + match encoding.keeps_up(*canvas) { + true => ui.label(text), + false => ui.colored_label( + ui.visuals().warn_fg_color, + format!("{text}, too slow to keep up"), + ), + } + } + Err(err) => ui.colored_label( + ui.visuals().error_fg_color, + format!("{}: {err}", describe(*canvas)), + ), + }; + } + }); + } + } + ui.end_row(); + + ui.label("Uplink"); + match &self.uplink { + None => { + ui.weak("measured once the encoders are known"); + } + Some(Task::Running(_)) => { + ui.horizontal(|ui| { + ui.spinner(); + ui.label(format!( + "publishing a probe to the relay for about {} s", + uplink::WINDOW.as_secs() + )); + }); + } + Some(Task::Done(Ok(uplink))) => { + ui.label(format!("{} to the relay", rate(*uplink))); + } + Some(Task::Done(Err(err))) => { + ui.colored_label(ui.visuals().error_fg_color, format!("not measured: {err}")); + } + } + ui.end_row(); + }); + } + + /// Draw the proposal; `Some(empty)` once the creator starts the studio, with + /// `empty` when they turned the proposal down. + fn proposal(&mut self, ui: &mut Ui) -> Option { + let (Some(choices), Task::Done(listing), Task::Done(encoders), Some(Task::Done(uplink))) = + (&mut self.choices, &self.devices, &self.encoders, &self.uplink) + else { + ui.label("Proposing once every check is done."); + return None; + }; + + let idle = self.starting.is_none(); + ui.strong(format!("Proposed scene: {SCENE}")); + ui.add_enabled_ui(idle, |ui| { + egui::Grid::new("proposal").num_columns(3).show(ui, |ui| { + let has_display = device(ui, "Display", "display", &listing.displays, &mut choices.display); + ui.label(match (has_display, choices.camera.is_some()) { + (true, _) => "Fills the canvas, so viewers see your screen.", + (false, true) => "No display, so the camera fills the canvas.", + (false, false) => "No display.", + }); + ui.end_row(); + + let has_camera = device(ui, "Camera", "camera", &listing.cameras, &mut choices.camera); + ui.label(match (has_camera, choices.display.is_some()) { + (true, true) => "Sits in the bottom-right corner, a quarter of the canvas wide, over the display.", + (true, false) => "Fills the canvas.", + (false, _) => "No camera.", + }); + ui.end_row(); + + let has_audio = device(ui, "Audio", "audio", &listing.audio, &mut choices.audio); + ui.label(match has_audio { + true => "Mixed into the program. Set its level in the Audio panel.", + false => "No input, so the program is silent unless a source carries sound.", + }); + ui.end_row(); + + ui.label("Canvas"); + egui::ComboBox::from_id_salt("canvas") + .selected_text(describe(choices.canvas)) + .show_ui(ui, |ui| { + for (canvas, _) in encoders { + ui.selectable_value(&mut choices.canvas, *canvas, describe(*canvas)); + } + }); + ui.label(explain(choices.canvas, encoders, uplink)); + ui.end_row(); + }); + }); + + if let Some(error) = &self.error { + ui.colored_label(ui.visuals().error_fg_color, error); + } + if !idle { + ui.spinner(); + return None; + } + let mut start = None; + ui.horizontal(|ui| { + if ui.button("Start with this scene").clicked() { + start = Some(false); + } + if ui.button("Start with an empty scene").clicked() { + start = Some(true); + } + }); + start + } + + /// Start the studio with the proposal, or with no sources on the chosen + /// canvas when `empty`. Either way its autosave creates the scene file at + /// once. + fn start(&mut self, empty: bool) { + let (Task::Done(listing), Some(choices)) = (&self.devices, &self.choices) else { + return; + }; + let loaded = match loaded(listing, choices, empty) { + Ok(loaded) => loaded, + Err(err) => { + self.error = Some(format!("{err:#}")); + return; + } + }; + // Kept, so a start that fails can be retried with other choices. + let launch = self.launch.clone(); + let path = self.path.clone(); + let starting = Task::spawn(&self.runtime, move || async move { + // Setup proposes only for a missing file, so one that appeared since, + // such as another studio's, is opened by a restart, not overwritten. + // The autosave's first save refuses one that appears after this. + let (autosave, _) = Autosave::open(path)?; + anyhow::ensure!( + !autosave.exists(), + "{} was created during setup; restart the studio to open it", + autosave.path().display() + ); + Engine::start(launch, autosave, loaded).await + }); + match starting { + Ok(starting) => { + self.error = None; + self.starting = Some(starting); + } + Err(err) => self.error = Some(format!("cannot start the studio: {err}")), + } + } +} + +/// A row picking one of `devices` or none, or saying why there is none to +/// pick. Whether one is picked. +fn device(ui: &mut Ui, label: &str, id: &str, devices: &Result, String>, choice: &mut Option) -> bool { + ui.label(label); + match devices { + Ok(devices) if !devices.is_empty() => { + let selected = choice + .and_then(|index| devices.get(index)) + .map_or("None", |pick| pick.name.as_str()); + egui::ComboBox::from_id_salt(id) + .selected_text(selected) + .show_ui(ui, |ui| { + ui.selectable_value(choice, None, "None"); + for (index, pick) in devices.iter().enumerate() { + ui.selectable_value(choice, Some(index), &pick.name); + } + }); + } + Ok(_) => { + ui.weak("none found"); + } + Err(err) => { + ui.colored_label(ui.visuals().error_fg_color, "cannot list") + .on_hover_text(err); + } + } + choice.is_some() +} + +#[cfg(test)] +mod tests { + use moq_scene::Source; + + use super::*; + + /// [`CANVASES`] each encoding at its default bitrate and keeping up, or + /// failing where `None`. + fn encoders(kbps: [Option; 5]) -> Encoders { + CANVASES + .into_iter() + .zip(kbps) + .map(|(canvas, kbps)| { + let result = kbps + .map(|kbps| Encoding { + encoder: "test".into(), + gpu: true, + bitrate: Rate::from_kbps(kbps), + frame: Duration::from_millis(1), + }) + .ok_or_else(|| "refused".to_string()); + (canvas, result) + }) + .collect() + } + + /// [`encoders`] at their default bitrates, each frame taking `millis`. + fn timed(millis: [u64; 5]) -> Encoders { + let mut encoders = encoders(DEFAULTS); + for ((_, result), millis) in encoders.iter_mut().zip(millis) { + result.as_mut().unwrap().frame = Duration::from_millis(millis); + } + encoders + } + + const DEFAULTS: [Option; 5] = [Some(8709), Some(4354), Some(3870), Some(1935), Some(484)]; + + #[test] + fn picks_the_best_canvas_within_the_headroom() { + let encoders = encoders(DEFAULTS); + // 4.354 + 0.128 Mbps is just inside two thirds of 6.8 Mbps, where the + // Health panel's headroom grade turns degraded. + assert_eq!(pick(&encoders, Some(Rate::from_kbps(6800))), CANVASES[1]); + assert_eq!(pick(&encoders, Some(Rate::from_kbps(6700))), CANVASES[2]); + assert_eq!(pick(&encoders, Some(Rate::from_mbps(100))), CANVASES[0]); + } + + #[test] + fn skips_canvases_whose_encoder_cannot_keep_up() { + // 60 fps leaves 12.5 ms a frame, and 30 fps 25 ms. + let encoders = timed([20, 20, 12, 12, 5]); + assert_eq!(pick(&encoders, Some(Rate::from_mbps(100))), CANVASES[1]); + assert_eq!(pick(&encoders, None), Canvas::default()); + let encoders = timed([20, 30, 12, 12, 5]); + assert_eq!(pick(&encoders, Some(Rate::from_mbps(100))), CANVASES[2]); + assert_eq!(pick(&encoders, None), CANVASES[2], "the default does not keep up"); + } + + #[test] + fn proposes_the_closest_to_keeping_up_when_none_does() { + let encoders = timed([100, 80, 60, 40, 30]); + assert_eq!(pick(&encoders, Some(Rate::from_mbps(100))), CANVASES[4]); + assert_eq!(pick(&encoders, None), CANVASES[4]); + } + + /// Times `time` over an encoder whose frames take `first` and then `rest` + /// on the paused clock, with how many frames it was handed. + async fn burst(budget: Duration, first: Duration, rest: Duration) -> (Duration, u32) { + let mut calls = 0; + let frame = time(budget, async |index| { + calls += 1; + kio::clock::sleep(if index == 0 { first } else { rest }).await; + Ok(()) + }) + .await + .unwrap(); + (frame, calls) + } + + #[tokio::test(start_paused = true)] + async fn times_every_frame_but_the_first() { + let budget = Duration::from_millis(25); + let (frame, calls) = burst(budget, Duration::from_secs(1), Duration::from_millis(10)).await; + assert_eq!(frame, Duration::from_millis(10)); + assert_eq!(calls, BURST + 1); + } + + /// A slow encoder is cut off once it cannot average under the budget, so + /// it takes no longer than one that keeps up. + #[tokio::test(start_paused = true)] + async fn stops_a_burst_that_falls_behind() { + let budget = Duration::from_millis(25); + let started = kio::clock::now(); + let (frame, calls) = burst(budget, Duration::ZERO, Duration::from_millis(50)).await; + assert_eq!(frame, Duration::from_millis(50)); + // Over the 375 ms the whole burst may take, at the eighth timed frame. + assert_eq!(calls, 9); + assert!(started.elapsed() <= budget * BURST + Duration::from_millis(50)); + } + + /// The burst runs on a real device and encoder, through the program's own + /// copy path. + #[tokio::test(flavor = "multi_thread")] + #[ignore = "needs a Vulkan device and an H.264 encoder; run by just render-sw"] + async fn times_every_canvas() { + let gpu = engine::Gpu::new().await.expect("a GPU"); + for (canvas, result) in probe_encoders(gpu).await { + let encoding = result.unwrap_or_else(|err| panic!("{}: {err}", describe(canvas))); + assert!(encoding.frame > Duration::ZERO, "{}", describe(canvas)); + } + } + + #[test] + fn falls_back_to_the_cheapest_canvas_that_encodes() { + let encoders = encoders([Some(8709), Some(4354), Some(3870), Some(1935), None]); + assert_eq!(pick(&encoders, Some(Rate::from_kbps(300))), CANVASES[3]); + } + + #[test] + fn skips_canvases_that_do_not_encode() { + let encoders = encoders([None, Some(4354), Some(3870), Some(1935), Some(484)]); + assert_eq!(pick(&encoders, Some(Rate::from_mbps(100))), CANVASES[1]); + } + + #[test] + fn keeps_the_default_canvas_without_a_measurement() { + assert_eq!(pick(&encoders(DEFAULTS), None), Canvas::default()); + assert_eq!( + pick(&encoders([None; 5]), Some(Rate::from_mbps(100))), + Canvas::default() + ); + } + + fn listing(cameras: &[&str], displays: &[&str], audio: &[&str]) -> Listing { + let picks = |names: &[&str], source: fn(String) -> Source| { + Ok(names + .iter() + .map(|name| Pick { + name: name.to_string(), + source: source(format!("{name}-id")), + }) + .collect()) + }; + Listing { + cameras: picks(cameras, |id| Source::Camera { id: Some(id) }), + displays: picks(displays, |id| Source::Display { id: Some(id) }), + windows: Ok(Vec::new()), + audio: picks(audio, |id| Source::Microphone { id: Some(id) }), + } + } + + #[test] + fn the_camera_sits_over_the_display() { + let listing = listing(&["Webcam"], &["Screen"], &["Mic"]); + let choices = propose(&listing, CANVASES[3]); + let autosave::Loaded { + mut collection, + mut history, + .. + } = loaded(&listing, &choices, false).unwrap(); + collection.to_toml().expect("a valid collection"); + + let items = &collection.scene(SCENE).expect("the scene").items; + let placed: Vec<_> = items.iter().map(|item| item.source.as_str()).collect(); + assert_eq!(placed, ["Screen", "Webcam"], "back to front"); + assert_eq!(items[0].rect, Rect::fill(&CANVASES[3])); + let corner = items[1].rect; + assert!( + corner.x > 640.0 && corner.y > 360.0, + "in the bottom-right quarter: {corner:?}" + ); + assert!(corner.x + corner.width < 1280.0 && corner.y + corner.height < 720.0); + assert!(collection.sources.contains_key("Mic"), "the input is a source"); + + // One edit, which undoes to what starting empty opens. + assert_eq!(history.entries().len(), 1); + history.undo(&mut collection).unwrap(); + assert_eq!(collection, loaded(&listing, &choices, true).unwrap().collection); + assert!(collection.sources.is_empty()); + assert_eq!(collection.canvas, CANVASES[3]); + } + + #[test] + fn the_camera_alone_fills_the_canvas() { + let listing = listing(&["Webcam"], &["Screen"], &[]); + let mut choices = propose(&listing, Canvas::default()); + choices.display = None; + let collection = loaded(&listing, &choices, false).unwrap().collection; + let items = &collection.scene(SCENE).expect("the scene").items; + assert_eq!(items.len(), 1); + assert_eq!(items[0].rect, Rect::fill(&Canvas::default())); + } + + #[test] + fn devices_sharing_a_name_get_their_own_sources() { + let listing = listing(&["Built-in"], &["Built-in"], &["Built-in"]); + let collection = loaded(&listing, &propose(&listing, Canvas::default()), false) + .unwrap() + .collection; + let names: Vec<_> = collection.sources.keys().map(String::as_str).collect(); + assert_eq!(names, ["Built-in", "Built-in 2", "Built-in 3"]); + collection.to_toml().expect("a valid collection"); + } +} diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index 74e76e9..5b98810 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -10,6 +10,7 @@ use egui_dock::{DockArea, DockState, NodeIndex, Style, TabViewer}; use moq_scene::source::{Feed, Status}; use moq_scene::{Clip, Collection, Draft, Filter, Item, Layer, Rect, Scene, Source, Video, wgpu}; +use crate::devices::{self, Listing, Pick}; use crate::engine::{Engine, State}; use crate::health::{self, Verdict}; @@ -54,16 +55,7 @@ pub struct Studio { } impl Studio { - pub fn new( - cc: &eframe::CreationContext<'_>, - engine: Arc, - runtime: tokio::runtime::Handle, - ) -> anyhow::Result { - let render = cc - .wgpu_render_state - .as_ref() - .context("the window is not drawn with wgpu")?; - + pub fn new(render: &eframe::egui_wgpu::RenderState, engine: Arc, runtime: tokio::runtime::Handle) -> Self { // The compositor's textures live as long as the engine, so registering // them once is enough: egui samples whatever they hold each frame. let (preview, program, dropped) = { @@ -88,7 +80,7 @@ impl Studio { let [_, sources] = tree.split_right(items, 0.4, vec![Tab::Sources, Tab::Devices]); tree.split_right(sources, 0.5, vec![Tab::Filters, Tab::Health, Tab::Audio, Tab::Output]); - Ok(Self { + Self { dock, panels: Panels { engine, @@ -104,7 +96,7 @@ impl Studio { devices: Devices::default(), message: (dropped > 0).then(|| Err(dropped_message(dropped))), }, - }) + } } } @@ -877,7 +869,8 @@ impl Panels { fn devices(&mut self, ui: &mut Ui) { self.devices.poll(); if ui.button("Refresh").clicked() { - self.devices.refresh(&self.runtime); + let refreshed = self.devices.refresh(&self.runtime).context("cannot list devices"); + self.check(refreshed); } let Some(picked) = self.devices.ui(ui) else { return }; let mut state = self.engine.state.lock().unwrap(); @@ -1044,6 +1037,13 @@ impl Panels { ui.end_row(); } } + let canvas = self.engine.state.lock().unwrap().collection().canvas; + ui.label("Canvas"); + ui.label(format!( + "{}x{} at {} fps, set in the scene file and read at startup", + canvas.width, canvas.height, canvas.framerate + )); + ui.end_row(); ui.label("Drawn"); ui.label(format!("{} frames", count(&stats.rendered))); ui.end_row(); @@ -1485,12 +1485,6 @@ impl AddSource { } } -/// A device the user can add as a source. -struct Pick { - name: String, - source: Source, -} - /// The capture devices this machine has, listed in the background. #[derive(Default)] struct Devices { @@ -1498,22 +1492,20 @@ struct Devices { pending: Option>, } -/// One listing of every kind of device, each of which can fail on its own. -struct Listing { - kinds: Vec<(&'static str, Result, String>)>, -} - impl Devices { - fn refresh(&mut self, runtime: &tokio::runtime::Handle) { + fn refresh(&mut self, runtime: &tokio::runtime::Handle) -> std::io::Result<()> { let (send, recv) = std::sync::mpsc::channel(); - self.pending = Some(recv); // The capture backends' listing futures are `!Send` on macOS, as their // capture futures are (which is why each source runs on its own thread), - // so poll the listing on a blocking thread rather than spawn it. + // so poll the listing on a thread rather than spawn it. Not the runtime's + // blocking pool, which dropping the runtime waits out, so a listing stuck + // on a permission prompt cannot hold the studio open once it closes. let handle = runtime.clone(); - runtime.spawn_blocking(move || { - let _ = send.send(handle.block_on(list())); - }); + std::thread::Builder::new().name("devices".into()).spawn(move || { + let _ = send.send(handle.block_on(devices::list())); + })?; + self.pending = Some(recv); + Ok(()) } fn poll(&mut self) { @@ -1535,8 +1527,8 @@ impl Devices { let mut picked = None; egui::ScrollArea::vertical() .show(ui, |ui| { - for (kind, devices) in &mut listing.kinds { - ui.strong(*kind); + for (kind, devices) in listing.kinds() { + ui.strong(kind); match devices { Ok(devices) if devices.is_empty() => { ui.weak("none"); @@ -1564,56 +1556,3 @@ impl Devices { .inner } } - -#[cfg(feature = "capture")] -async fn list() -> Listing { - fn picks(result: Result, impl std::fmt::Display>, pick: impl Fn(T) -> Pick) -> Result, String> { - result - .map(|devices| devices.into_iter().map(pick).collect()) - .map_err(|err| err.to_string()) - } - - let cameras = picks(moq_video::capture::cameras().await, |camera| Pick { - name: camera.name, - source: Source::Camera { id: Some(camera.id) }, - }); - let displays = picks(moq_video::capture::displays().await, |display| Pick { - name: display.name, - source: Source::Display { id: Some(display.id) }, - }); - let windows = picks(moq_video::capture::windows().await, |window| Pick { - name: format!("{} ({})", window.title, window.app), - source: Source::Window { id: window.id }, - }); - let mut microphones = picks(moq_audio::capture::devices().await, |device| Pick { - name: format!("{} ({})", device.name, device.host), - source: Source::Microphone { id: Some(device.id) }, - }); - if cfg!(target_os = "macos") - && let Ok(microphones) = &mut microphones - { - microphones.push(Pick { - name: "System audio".into(), - source: Source::SystemAudio, - }); - } - - Listing { - kinds: vec![ - ("Cameras", cameras), - ("Displays", displays), - ("Windows", windows), - ("Audio inputs", microphones), - ], - } -} - -#[cfg(not(feature = "capture"))] -async fn list() -> Listing { - Listing { - kinds: vec![( - "Devices", - Err("this build cannot capture; rebuild with `--features capture`".into()), - )], - } -} diff --git a/rs/moq-studio/src/uplink.rs b/rs/moq-studio/src/uplink.rs new file mode 100644 index 0000000..f054f9e --- /dev/null +++ b/rs/moq-studio/src/uplink.rs @@ -0,0 +1,191 @@ +//! Measure the uplink to the relay, for guided setup. +//! +//! The bandwidth grant alone measures nothing: a congestion controller only +//! learns what the path carries while something is sent. So this publishes a +//! short probe broadcast, subscribes to it back through the relay so the relay +//! pulls it, reserves the most setup could propose so the transport probes up +//! to that, and reads the congestion controller's send-rate estimate. + +use std::time::Duration; + +use anyhow::Context; +use bytes::Bytes; +use moq_net::bandwidth::Rate; + +/// How long the estimate is watched once the probe flows. The transport's probe +/// reaches its target within about five seconds of the path opening up. +pub const WINDOW: Duration = Duration::from_secs(6); + +/// The least filler sent per second, until the estimate says more fits. +const FLOOR: Rate = Rate::from_kbps(500); + +/// Filler frames per second. +const RATE: u32 = 30; + +/// The probe's one track. +const TRACK: &str = "probe"; + +/// Publish a probe at `path` on the relay at `url` for `window` once it flows, +/// probing up to `ceiling`, and return the send-rate estimate it ends with. +/// +/// Fails, rather than guessing, when either connection fails, the relay never +/// delivers the probe back, or the transport reports no estimate (WebSocket +/// has none). +pub async fn measure( + client: &moq_tokio::Client, + url: url::Url, + path: &str, + ceiling: Rate, + window: Duration, +) -> anyhow::Result { + let publish = moq_tokio::origin::spawn(); + let publication = publish + .create_broadcast(path) + .context("cannot create the probe broadcast")?; + let track = publication + .broadcast() + .create_track(TRACK, None) + .context("cannot create the probe track")?; + publication + .announce(moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint())) + .context("cannot announce the probe")?; + + // One-shot, so a relay that refuses either side ends the probe with its reason + // instead of being redialed. + let client = client.clone().with_reconnect(false); + let publisher = client.clone().with_publisher(&publish).connect(url.clone()); + let subscribe = moq_tokio::origin::spawn(); + let viewer = client.with_subscriber(subscribe.clone()).connect(url); + + // The transport probes up to the reservations' sum, so the estimate can rise + // past what the filler sends. + let reservation = publisher.allocator().reserve(&track.demand(), ceiling); + + let measured = async { + // The window starts when the probe comes back, so a slow dial does not eat it. + let Ok(received) = kio::clock::timeout(window, receive(subscribe.consume(), path)).await else { + anyhow::bail!(match publisher.connected() && viewer.connected() { + true => format!("the relay did not deliver the probe back within {} s", window.as_secs()), + false => format!("cannot connect to the relay within {} s", window.as_secs()), + }); + }; + let mut groups = received?; + // Drain the probe so the relay keeps pulling it; only the window ends this. + let drained: Result, _> = kio::clock::timeout(window, async { + loop { + groups.recv_group().await?.context("the probe ended early")?; + } + }) + .await; + if let Ok(Err(err)) = drained { + return Err(err); + } + // The controller's estimate is already a maximum over its recent rounds. + publisher + .send_bandwidth() + .peek() + .context("the relay connection reports no bandwidth estimate, as over WebSocket") + }; + + let result = tokio::select! { + result = publisher.closed() => Err(anyhow::anyhow!("the relay connection ended: {}", describe(result))), + result = viewer.closed() => Err(anyhow::anyhow!("the probe's viewer connection ended: {}", describe(result))), + Err(err) = fill(track, &reservation, ceiling) => Err(err), + result = measured => result, + }; + publication.close(); + result +} + +/// Wait for the probe through the relay, and subscribe to its track. +async fn receive(origin: moq_net::origin::Consumer, path: &str) -> anyhow::Result { + let broadcast = origin.follow(path, None)?.available().await?; + let mut track = broadcast.track(TRACK)?.subscribe(None).await?; + // Wait for the first group, so the window measures data that flows. + track.recv_group().await?.context("the probe ended before it began")?; + Ok(track) +} + +/// Send filler at the rate the estimate grants, between [`FLOOR`] and `ceiling`. +/// +/// Returns only on an error, since the probe ends by dropping it. +async fn fill( + track: moq_net::track::Producer, + reservation: &moq_net::bandwidth::Reservation, + ceiling: Rate, +) -> anyhow::Result<()> { + let largest = (ceiling.as_bps() / 8 / u64::from(RATE)) as usize; + let filler = Bytes::from(vec![0u8; largest.max(1)]); + let mut clock = kio::clock::interval(Duration::from_secs(1) / RATE); + clock.set_missed_tick_behavior(kio::clock::MissedTickBehavior::Skip); + let mut group = None; + let mut frames = 0; + + loop { + clock.tick().await; + if frames % RATE == 0 { + if let Some(mut done) = group.take() { + moq_net::group::Producer::finish(&mut done)?; + } + group = Some(track.append_group()?); + } + frames += 1; + let rate = reservation.rate().unwrap_or(FLOOR).clamp(FLOOR, ceiling); + let size = (rate.as_bps() / 8 / u64::from(RATE)) as usize; + let group = group.as_mut().expect("opened above"); + group.write_frame(None, filler.slice(..size.clamp(1, filler.len())))?; + } +} + +fn describe(result: moq_tokio::Result<()>) -> String { + match result { + Ok(()) => "closed".into(), + Err(err) => err.to_string(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The probe reaches a relay and comes back with the congestion + /// controller's estimate. + #[tokio::test(flavor = "multi_thread")] + async fn measures_the_uplink_to_a_relay() { + let _ = moq_tokio::crypto::install_default(); + let fixture = moq_relay::test_relay().await.expect("test relay"); + let ready = fixture.relay.ready(); + tokio::spawn(fixture.relay.run()); + ready.wait().await.expect("relay ready"); + + let mut client = moq_tokio::connect::Config::default(); + client.tls.fingerprint = vec![fixture.fingerprint.clone()]; + let client = client.init(Default::default()).expect("client"); + + let rate = measure( + &client, + fixture.url.clone(), + "studio", + Rate::from_mbps(10), + Duration::from_secs(2), + ) + .await + .expect("a measurement"); + assert!(rate > Rate::ZERO, "an estimate of {rate:?}"); + } + + /// A relay that cannot be reached fails the probe rather than reporting a rate. + #[tokio::test(flavor = "multi_thread")] + async fn reports_an_unreachable_relay() { + let _ = moq_tokio::crypto::install_default(); + let client = moq_tokio::connect::Config::default() + .init(Default::default()) + .expect("client"); + // Nothing listens on the discard port. + let url = url::Url::parse("https://127.0.0.1:9").unwrap(); + let err = measure(&client, url, "studio", Rate::from_mbps(10), Duration::from_secs(2)) + .await + .expect_err("no relay answers"); + assert!(err.to_string().contains("connect"), "names the connection: {err:#}"); + } +}