From aaa5ed1bb7dac656adbc55d0f8a2d7311dfb27d8 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 14:42:03 -0800 Subject: [PATCH 1/8] chore(quest): claim quest/m0/guided-setup.md Quest-Claim: 579298a7a5ac44a3 From cc186e521267ba07a0d1407919ddc57e84b7648c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 14:54:39 -0800 Subject: [PATCH 2/8] feat(studio): guided setup on first launch Co-Authored-By: Claude Opus 5.5 --- README.md | 34 +- quest/m0/README.md | 1 - quest/m0/guided-setup.md | 19 - rs/moq-studio/src/devices.rs | 83 +++++ rs/moq-studio/src/engine.rs | 57 ++- rs/moq-studio/src/main.rs | 78 +++- rs/moq-studio/src/setup.rs | 694 +++++++++++++++++++++++++++++++++++ rs/moq-studio/src/ui.rs | 94 +---- rs/moq-studio/src/uplink.rs | 183 +++++++++ 9 files changed, 1113 insertions(+), 130 deletions(-) delete mode 100644 quest/m0/guided-setup.md create mode 100644 rs/moq-studio/src/devices.rs create mode 100644 rs/moq-studio/src/setup.rs create mode 100644 rs/moq-studio/src/uplink.rs diff --git a/README.md b/README.md index 6da3900..59a0e6f 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,8 @@ 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 on the first save. +- `show.toml` is the scene collection. If it does not exist yet, + [guided setup](#first-launch) proposes one and writes it there. - 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. @@ -24,6 +25,35 @@ 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, and the bitrate it encodes at. +- **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 video and audio fit in three quarters of the measured +uplink. Without a measurement, it is the default 1080p at 30 fps. +**Start with this scene** writes the scene file and opens the studio. +**Start with an empty scene** opens it with nothing and writes no file, so +setup runs again on the next launch unless you save first. + +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. + ## Panels Every panel is a tab you can drag, split, or stack. @@ -36,7 +66,7 @@ Every panel is a tab you can drag, split, or stack. | Sources | Add a source by hand, see whether each is live, and remove one. | | Devices | List this machine's cameras, displays, windows, and audio inputs, and add one as a source. | | Audio | Set each audio source's gain and mute, and watch its meter and the program's. | -| Output | See the relay connection, whether anyone is watching, and the encoder's work. | +| Output | See the relay connection, whether anyone is watching, the canvas, and the encoder's work. | A broadcast source subscribes through the same relay connection, plays the best video and audio its catalog lists, switches when the catalog changes, and diff --git a/quest/m0/README.md b/quest/m0/README.md index 653a3c0..a82744c 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -94,7 +94,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/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 9cceccc..5d60718 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -186,26 +186,42 @@ 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 collection, which guided +/// setup holds until it has proposed one. +#[derive(Clone)] +pub struct Launch { + pub gpu: Gpu, + /// The scene file. + pub path: PathBuf, + /// 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 collection at `path` on `gpu` and start the program clock. - /// With `connect`, publish the program at `broadcast` on that relay, - /// encoding its video only while someone watches when `on_demand`. + /// Open `collection` from `launch.path` on `launch.gpu` and start the + /// program clock. 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, - path: PathBuf, - collection: Collection, - connect: Option<(moq_tokio::Client, url::Url)>, - broadcast: &str, - on_demand: bool, - ) -> anyhow::Result> { + pub async fn start(launch: Launch, collection: Collection) -> anyhow::Result> { + let Launch { + gpu, + path, + connect, + broadcast, + on_demand, + } = launch; let bus = moq_audio::bus::Bus::new(Default::default())?; let subscribe = moq_tokio::origin::spawn(); let dir = path.parent().map(Path::to_path_buf).unwrap_or_default(); @@ -222,7 +238,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), @@ -636,13 +652,14 @@ mod tests { visible: true, }); - let gpu = Gpu::new().await.expect("a GPU"); - let path = dir.path().join("show.toml"); - let connect = Some((client, fixture.url.clone())); - let on_demand = false; - let engine = Engine::start(gpu, path, collection, connect, "studio", on_demand) - .await - .expect("the studio starts"); + let launch = Launch { + gpu: Gpu::new().await.expect("a GPU"), + path: dir.path().join("show.toml"), + connect: Some((client, fixture.url.clone())), + broadcast: "studio".into(), + on_demand: false, + }; + let engine = Engine::start(launch, collection).await.expect("the studio starts"); let broadcast = kio::clock::timeout(TIMEOUT, relay.follow("studio", None).unwrap().available()) .await diff --git a/rs/moq-studio/src/main.rs b/rs/moq-studio/src/main.rs index 9527525..d053539 100644 --- a/rs/moq-studio/src/main.rs +++ b/rs/moq-studio/src/main.rs @@ -9,15 +9,19 @@ use std::sync::Arc; use anyhow::Context; +mod devices; mod engine; +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. Created on the first save if it does not exist. + /// The scene collection to open. If it does not exist, guided setup proposes + /// one and writes it here. #[usage(value_hint = usage_rs::ValueHint::FilePath, extensions("toml"))] pub scenes: PathBuf, @@ -65,25 +69,69 @@ 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 collection = engine::load(&config.scenes)?; 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(), - config.scenes.clone(), - collection, + let launch = engine::Launch { + gpu: gpu.clone(), + path: config.scenes.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 collection = engine::load(&config.scenes)?; + App::Starting(runtime.block_on(engine::Engine::start(launch, collection))?) + } + false => App::Setup(Box::new(setup::Setup::new(launch, runtime.handle().clone()))), + }; - run(engine, gpu, runtime.handle().clone()) + run(app, gpu, runtime.handle().clone()) +} + +/// 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, +} + +impl eframe::App for Window { + fn ui(&mut self, ui: &mut egui::Ui, frame: &mut eframe::Frame) { + let started = match &mut self.app { + App::Studio(studio) => return studio.ui(ui, frame), + App::Setup(setup) => match setup.ui(ui) { + Some(started) => started, + None => return, + }, + App::Starting(engine) => setup::Started { + engine: engine.clone(), + message: None, + }, + }; + let studio = ui::Studio::new(&self.render, started.engine, self.runtime.clone(), started.message); + 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) -> anyhow::Result<()> { let options = eframe::NativeOptions { viewport: egui::ViewportBuilder::default() .with_title("MoQ Studio") @@ -105,7 +153,13 @@ 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 })) + }), ) .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..c19ef47 --- /dev/null +++ b/rs/moq-studio/src/setup.rs @@ -0,0 +1,694 @@ +//! 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::sync::Arc; +use std::sync::mpsc; +use std::time::Duration; + +use egui::Ui; +use moq_net::bandwidth::Rate; +use moq_scene::{Canvas, Collection, Item, Rect, Scene}; + +use crate::devices::{self, Listing, Pick}; +use crate::engine::{self, Engine, Launch}; +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. The rest absorbs the +/// estimate's error and whatever else shares the connection. +const HEADROOM: f64 = 0.75; + +/// 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, +} + +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()) + } +} + +/// Each of [`CANVASES`], with how it encodes or why it cannot. +pub type Encoders = Vec<(Canvas, Result)>; + +/// Open the exporter the program would for each canvas, the same probe that +/// picks its encoder at startup, and keep what it settled on. +async fn probe_encoders(gpu: engine::Gpu) -> Encoders { + let mut probed = Vec::new(); + for canvas in CANVASES { + let mut config = moq_video::encode::Config::new(canvas.width, canvas.height, canvas.framerate()); + let result = moq_video::render::export::Exporter::new(&gpu.device, &gpu.queue, &mut config) + .await + .map_err(|err| err.to_string()) + .and_then(|(exporter, rendition)| { + let encoder = match &config.kind { + moq_video::encode::Kind::Named(name) => name.clone(), + other => format!("{other:?}"), + }; + let bitrate = rendition.bitrate.ok_or("the encoder reports no bitrate")?; + Ok(Encoding { + encoder, + gpu: exporter.route() == moq_video::render::export::Route::Gpu, + bitrate: Rate::from_bps(bitrate), + }) + }); + probed.push((canvas, result)); + } + probed +} + +/// The canvas to propose: the best one that encodes and whose program fits the +/// uplink's [`HEADROOM`], or the cheapest that encodes when none fits. Without a +/// measurement it is the default canvas when that encodes. +fn pick(encoders: &Encoders, uplink: Option) -> Canvas { + let encodable = || encoders.iter().filter_map(|(canvas, result)| Some((*canvas, result.as_ref().ok()?))); + let Some(uplink) = uplink else { + let default = Canvas::default(); + return encodable() + .map(|(canvas, _)| canvas) + .find(|canvas| *canvas == default) + .or_else(|| encodable().next().map(|(canvas, _)| canvas)) + .unwrap_or(default); + }; + let budget = uplink.scaled(HEADROOM); + encodable() + .find(|(_, encoding)| encoding.total() <= budget) + .or_else(|| encodable().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 => "reading it on the GPU", + false => "from a copy read back to the CPU each frame", + }; + let cost = format!( + "{} encodes it with {route} at about {}, {} with audio", + encoding.encoder, + mbps(encoding.bitrate), + mbps(encoding.total()) + ); + let uplink = match uplink { + Ok(uplink) => *uplink, + Err(_) => return format!("{name}: {cost}. The uplink was not measured, so check that it carries that."), + }; + let share = encoding.total().as_bps() as f64 / uplink.as_bps().max(1) as f64; + let percent = (share * 100.0).round(); + if share <= HEADROOM { + format!( + "{name}: {cost}, {percent}% of the {} uplink measured to the relay.", + mbps(uplink) + ) + } else { + format!( + "{name}: {cost}, {percent}% of the {} uplink measured to the relay, more than the {}% it should use. Expect dropped frames.", + mbps(uplink), + (HEADROOM * 100.0).round() + ) + } +} + +fn describe(canvas: Canvas) -> String { + format!("{}x{} at {} fps", canvas.width, canvas.height, canvas.framerate) +} + +fn mbps(rate: Rate) -> String { + format!("{:.1} Mbps", rate.as_bps() as f64 / 1e6) +} + +/// 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, + } +} + +/// The collection `choices` describe: the display filling the canvas, the +/// camera over its corner (or filling the canvas alone), and the audio input. +fn collection(listing: &Listing, choices: &Choices) -> Collection { + 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 collection = Collection::new(); + collection.canvas = choices.canvas; + 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 collection, pick); + scene.items.push(Item { + source: name, + rect, + visible: true, + }); + } + if let Some(pick) = &audio { + add(&mut collection, pick); + } + collection.scenes = vec![scene]; + collection +} + +/// Add `pick` under its device name, numbered past any source already called that. +fn add(collection: &mut Collection, pick: &Pick) -> String { + let name = std::iter::once(pick.name.clone()) + .chain((2..).map(|n| format!("{} {n}", pick.name))) + .find(|name| !collection.sources.contains_key(name)) + .expect("an unused name"); + collection.sources.insert(name.clone(), pick.source.clone()); + 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, + } +} + +/// What the studio starts with once setup is done: its engine, and the line for +/// its status bar. +pub struct Started { + pub engine: Arc, + pub message: Option>, +} + +/// A result being worked out in the background. +enum Task { + Running(mpsc::Receiver), + Done(T), +} + +impl Task { + /// Run `task` on a blocking thread of `runtime`, since encoder probes and the + /// engine's start block while they open encoders. + fn spawn(runtime: &tokio::runtime::Handle, task: F) -> Self + where + F: FnOnce() -> Fut + Send + 'static, + Fut: Future, + { + let (send, recv) = mpsc::channel(); + let handle = runtime.clone(); + runtime.spawn_blocking(move || { + let _ = send.send(handle.block_on(task())); + }); + Self::Running(recv) + } + + /// The result, once it arrived. + fn poll(&mut self) -> Option<&T> { + if let Self::Running(recv) = self + && let Ok(done) = recv.try_recv() + { + *self = Self::Done(done); + } + match self { + Self::Running(_) => None, + Self::Done(done) => Some(done), + } + } +} + +/// The setup screen. +pub struct Setup { + runtime: tokio::runtime::Handle, + launch: Launch, + 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`. + pub fn new(launch: Launch, runtime: tokio::runtime::Handle) -> Self { + let devices = Task::spawn(&runtime, devices::list); + let gpu = launch.gpu.clone(); + let encoders = Task::spawn(&runtime, move || probe_encoders(gpu)); + Self { + runtime, + launch, + devices, + encoders, + uplink: None, + choices: None, + starting: None, + error: None, + } + } + + /// Draw the screen; what the studio starts with once the creator starts it. + 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(started) => return Some(started), + 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 = encoders + .iter() + .filter_map(|(_, result)| result.as_ref().ok()) + .map(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)) => Task::spawn(&self.runtime, move || async move { + uplink::measure(&client, url, &path, ceiling, uplink::WINDOW) + .await + .map_err(|err| format!("{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.spinner(); + } + Task::Done(encoders) => { + ui.vertical(|ui| { + for (canvas, result) in encoders { + match result { + Ok(encoding) => ui.label(format!( + "{}: {}{}, {}", + describe(*canvas), + encoding.encoder, + if encoding.gpu { " on the GPU" } else { " from the CPU" }, + mbps(encoding.bitrate) + )), + 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(rate))) => { + ui.label(format!("{} to the relay", mbps(*rate))); + } + 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; + }; + + ui.strong(format!("Proposed scene: {SCENE}")); + 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 { + true => "Fills the canvas, so viewers see your screen.", + false => "No display, so the camera fills the canvas.", + }); + 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 self.starting.is_some() { + 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, writing it to the scene file, or + /// with an empty collection when `empty`, which writes nothing yet. + fn start(&mut self, empty: bool) { + let (Task::Done(listing), Some(choices)) = (&self.devices, &self.choices) else { + return; + }; + let proposed = (!empty).then(|| collection(listing, choices)); + // Kept, so a start that fails can be retried with other choices. + let launch = self.launch.clone(); + self.error = None; + self.starting = Some(Task::spawn(&self.runtime, move || async move { + let path = launch.path.clone(); + let engine = Engine::start(launch, proposed.clone().unwrap_or_default()).await?; + let message = proposed.map(|_| { + let saved = engine.state.lock().unwrap().save(); + saved + .map(|()| format!("set up; saved {}", path.display())) + .map_err(|err| format!("{err:#}")) + }); + Ok(Started { engine, message }) + })); + } +} + +/// 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, 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), + }) + .ok_or_else(|| "refused".to_string()); + (canvas, result) + }) + .collect() + } + + 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 three quarters of 6 Mbps. + assert_eq!(pick(&encoders, Some(Rate::from_mbps(6))), CANVASES[1]); + assert_eq!(pick(&encoders, Some(Rate::from_kbps(5900))), CANVASES[2]); + assert_eq!(pick(&encoders, Some(Rate::from_mbps(100))), CANVASES[0]); + } + + #[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 collection = collection(&listing, &choices); + 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"); + } + + #[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 = collection(&listing, &choices); + 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 = collection(&listing, &propose(&listing, Canvas::default())); + 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 e1c4f92..6aad662 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -4,12 +4,12 @@ use std::sync::Arc; use std::sync::atomic::Ordering; use std::time::Duration; -use anyhow::Context; use egui::{Color32, Ui}; use egui_dock::{DockArea, DockState, NodeIndex, Style, TabViewer}; use moq_scene::source::Status; use moq_scene::{Item, Rect, Scene, Source, wgpu}; +use crate::devices::{self, Listing, Pick}; use crate::engine::{Engine, State}; /// The panels, each a tab the user can move, split, or stack. @@ -47,15 +47,13 @@ pub struct Studio { } impl Studio { + /// The studio over `engine`, with `message` in the status bar. pub fn new( - cc: &eframe::CreationContext<'_>, + render: &eframe::egui_wgpu::RenderState, engine: Arc, runtime: tokio::runtime::Handle, - ) -> anyhow::Result { - let render = cc - .wgpu_render_state - .as_ref() - .context("the window is not drawn with wgpu")?; + message: Option>, + ) -> Self { // The compositor's textures live as long as the engine, so registering // them once is enough: egui samples whatever they hold each frame. @@ -80,7 +78,7 @@ impl Studio { let [_, sources] = tree.split_right(items, 0.4, vec![Tab::Sources, Tab::Devices]); tree.split_right(sources, 0.5, vec![Tab::Audio, Tab::Output]); - Ok(Self { + Self { dock, panels: Panels { engine, @@ -91,9 +89,9 @@ impl Studio { new_item: None, add: AddSource::default(), devices: Devices::default(), - message: None, + message, }, - }) + } } } @@ -518,6 +516,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(); @@ -772,12 +777,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 { @@ -785,17 +784,12 @@ 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) { let (send, recv) = std::sync::mpsc::channel(); self.pending = Some(recv); runtime.spawn(async move { - let _ = send.send(list().await); + let _ = send.send(devices::list().await); }); } @@ -818,8 +812,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"); @@ -848,55 +842,3 @@ impl Devices { } } -#[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..74a724c --- /dev/null +++ b/rs/moq-studio/src/uplink.rs @@ -0,0 +1,183 @@ +//! 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:#}"); + } +} From e34d1d9472f4a0c297ea09df0a93745a3dad77b1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 14:59:19 -0800 Subject: [PATCH 3/8] fix(studio): fail loudly on a dead setup task and lock choices while starting Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/setup.rs | 125 +++++++++++++++++++++--------------- rs/moq-studio/src/ui.rs | 2 - rs/moq-studio/src/uplink.rs | 16 +++-- 3 files changed, 84 insertions(+), 59 deletions(-) diff --git a/rs/moq-studio/src/setup.rs b/rs/moq-studio/src/setup.rs index c19ef47..a1cceb2 100644 --- a/rs/moq-studio/src/setup.rs +++ b/rs/moq-studio/src/setup.rs @@ -94,7 +94,11 @@ async fn probe_encoders(gpu: engine::Gpu) -> Encoders { /// uplink's [`HEADROOM`], or the cheapest that encodes when none fits. Without a /// measurement it is the default canvas when that encodes. fn pick(encoders: &Encoders, uplink: Option) -> Canvas { - let encodable = || encoders.iter().filter_map(|(canvas, result)| Some((*canvas, result.as_ref().ok()?))); + let encodable = || { + encoders + .iter() + .filter_map(|(canvas, result)| Some((*canvas, result.as_ref().ok()?))) + }; let Some(uplink) = uplink else { let default = Canvas::default(); return encodable() @@ -116,7 +120,9 @@ fn explain(canvas: Canvas, encoders: &Encoders, uplink: &Result) - 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}"), + 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 { @@ -267,11 +273,16 @@ impl Task { } /// 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 - && let Ok(done) = recv.try_recv() - { - *self = Self::Done(done); + 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, @@ -474,46 +485,50 @@ impl Setup { return None; }; + let idle = self.starting.is_none(); ui.strong(format!("Proposed scene: {SCENE}")); - 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 { - true => "Fills the canvas, so viewers see your screen.", - false => "No display, so the camera fills the canvas.", - }); - 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(); + 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_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(); + 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(); - 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)); - } + 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.label(explain(choices.canvas, encoders, uplink)); - ui.end_row(); + 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 self.starting.is_some() { + if !idle { ui.spinner(); return None; } @@ -555,23 +570,21 @@ impl Setup { /// 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 { +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); - } - }); + 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"); @@ -634,7 +647,10 @@ mod tests { #[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()); + assert_eq!( + pick(&encoders([None; 5]), Some(Rate::from_mbps(100))), + Canvas::default() + ); } fn listing(cameras: &[&str], displays: &[&str], audio: &[&str]) -> Listing { @@ -667,7 +683,10 @@ mod tests { 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 > 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"); } diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index 6aad662..0d95805 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -54,7 +54,6 @@ impl Studio { runtime: tokio::runtime::Handle, message: Option>, ) -> 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) = { @@ -841,4 +840,3 @@ impl Devices { .inner } } - diff --git a/rs/moq-studio/src/uplink.rs b/rs/moq-studio/src/uplink.rs index 74a724c..f054f9e 100644 --- a/rs/moq-studio/src/uplink.rs +++ b/rs/moq-studio/src/uplink.rs @@ -39,7 +39,9 @@ pub async fn measure( window: Duration, ) -> anyhow::Result { let publish = moq_tokio::origin::spawn(); - let publication = publish.create_broadcast(path).context("cannot create the probe broadcast")?; + let publication = publish + .create_broadcast(path) + .context("cannot create the probe broadcast")?; let track = publication .broadcast() .create_track(TRACK, None) @@ -160,9 +162,15 @@ mod tests { 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"); + 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:?}"); } From 98187f4a1fef3835ffa8cb38e6721d4fa2925863 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 15:02:03 -0800 Subject: [PATCH 4/8] fix(studio): keep a setup proposal unsaved until its write lands Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/setup.rs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/rs/moq-studio/src/setup.rs b/rs/moq-studio/src/setup.rs index a1cceb2..c901fde 100644 --- a/rs/moq-studio/src/setup.rs +++ b/rs/moq-studio/src/setup.rs @@ -558,8 +558,11 @@ impl Setup { let path = launch.path.clone(); let engine = Engine::start(launch, proposed.clone().unwrap_or_default()).await?; let message = proposed.map(|_| { - let saved = engine.state.lock().unwrap().save(); - saved + let mut state = engine.state.lock().unwrap(); + // Unsaved until the write lands, so a failed one still shows. + state.edit(); + state + .save() .map(|()| format!("set up; saved {}", path.display())) .map_err(|err| format!("{err:#}")) }); From 58671c87d0af3a5384ae6e675dc499430da5efa4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 19:13:03 -0800 Subject: [PATCH 5/8] feat(studio): propose only canvases whose encoder keeps up Each canvas's encoder probe now times a 15-frame burst of a panning picture through the program's own copy and encode path, and setup proposes only canvases that copy and encode a frame within three quarters of its period. When none keeps up, it proposes the one closest to keeping up. A burst that falls behind stops early, so a slow encoder costs no more time than a fast one. The uplink headroom now follows the Health panel's degraded threshold, so a proposal is not graded degraded the moment it goes live, and rates use the Health panel's formatting. Co-Authored-By: Claude Opus 5.5 --- README.md | 11 +- justfile | 2 +- rs/moq-studio/src/setup.rs | 438 ++++++++++++++++++++++++++++++------- 3 files changed, 373 insertions(+), 78 deletions(-) diff --git a/README.md b/README.md index d4613f7..e74ce8b 100644 --- a/README.md +++ b/README.md @@ -34,7 +34,10 @@ 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, and the bitrate it encodes at. + 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 @@ -45,8 +48,10 @@ 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 video and audio fit in three quarters of the measured -uplink. Without a measurement, it is the default 1080p at 30 fps. +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 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/rs/moq-studio/src/setup.rs b/rs/moq-studio/src/setup.rs index 215d794..4a09981 100644 --- a/rs/moq-studio/src/setup.rs +++ b/rs/moq-studio/src/setup.rs @@ -13,11 +13,12 @@ use std::time::Duration; use anyhow::Context; use egui::Ui; use moq_net::bandwidth::Rate; -use moq_scene::{Canvas, Collection, Item, Rect, Scene}; +use moq_scene::{Canvas, Collection, Item, Rect, Scene, wgpu}; use crate::autosave::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 @@ -38,9 +39,25 @@ const fn canvas(width: u32, height: u32, framerate: u32) -> Canvas { } } -/// The share of the measured uplink the program may use. The rest absorbs the -/// estimate's error and whatever else shares the connection. -const HEADROOM: f64 = 0.75; +/// 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"; @@ -55,6 +72,8 @@ pub struct Encoding { 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 { @@ -62,59 +81,230 @@ impl Encoding { 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 keep what it settled on. +/// 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 mut config = moq_video::encode::Config::new(canvas.width, canvas.height, canvas.framerate()); - let result = moq_video::render::export::Exporter::new(&gpu.device, &gpu.queue, &mut config) + let result = probe_encoder(&gpu, &picture, canvas) .await - .map_err(|err| err.to_string()) - .and_then(|(exporter, rendition)| { - let encoder = match &config.kind { - moq_video::encode::Kind::Named(name) => name.clone(), - other => format!("{other:?}"), - }; - let bitrate = rendition.bitrate.ok_or("the encoder reports no bitrate")?; - Ok(Encoding { - encoder, - gpu: exporter.route() == moq_video::render::export::Route::Gpu, - bitrate: Rate::from_bps(bitrate), - }) - }); + .map_err(|err| format!("{err:#}")); probed.push((canvas, result)); } probed } -/// The canvas to propose: the best one that encodes and whose program fits the -/// uplink's [`HEADROOM`], or the cheapest that encodes when none fits. Without a -/// measurement it is the default canvas when that encodes. -fn pick(encoders: &Encoders, uplink: Option) -> Canvas { - let encodable = || { - encoders - .iter() - .filter_map(|(canvas, result)| Some((*canvas, result.as_ref().ok()?))) +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 encodable() - .map(|(canvas, _)| canvas) + return candidates + .iter() + .map(|(canvas, _)| *canvas) .find(|canvas| *canvas == default) - .or_else(|| encodable().next().map(|(canvas, _)| canvas)) + .or_else(|| candidates.first().map(|(canvas, _)| *canvas)) .unwrap_or(default); }; - let budget = uplink.scaled(HEADROOM); - encodable() + let budget = uplink.scaled(headroom()); + candidates + .iter() .find(|(_, encoding)| encoding.total() <= budget) - .or_else(|| encodable().min_by_key(|(_, encoding)| encoding.total())) - .map(|(canvas, _)| canvas) + .or_else(|| candidates.iter().min_by_key(|(_, encoding)| encoding.total())) + .map(|(canvas, _)| *canvas) .unwrap_or_default() } @@ -129,41 +319,52 @@ fn explain(canvas: Canvas, encoders: &Encoders, uplink: &Result) - None => return format!("{name} was not probed."), }; let route = match encoding.gpu { - true => "reading it on the GPU", + true => "on the GPU", false => "from a copy read back to the CPU each frame", }; - let cost = format!( - "{} encodes it with {route} at about {}, {} with audio", + let mut text = format!( + "{name}: {} encodes it {route} at about {}, {} with audio.", encoding.encoder, - mbps(encoding.bitrate), - mbps(encoding.total()) + 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(_) => return format!("{name}: {cost}. The uplink was not measured, so check that it carries that."), + 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; - let percent = (share * 100.0).round(); - if share <= HEADROOM { - format!( - "{name}: {cost}, {percent}% of the {} uplink measured to the relay.", - mbps(uplink) - ) - } else { - format!( - "{name}: {cost}, {percent}% of the {} uplink measured to the relay, more than the {}% it should use. Expect dropped frames.", - mbps(uplink), - (HEADROOM * 100.0).round() - ) + 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 mbps(rate: Rate) -> String { - format!("{:.1} Mbps", rate.as_bps() as f64 / 1e6) +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 @@ -368,12 +569,11 @@ impl Setup { if self.uplink.is_none() && let Some(encoders) = self.encoders.poll() { - let ceiling = encoders - .iter() - .filter_map(|(_, result)| result.as_ref().ok()) - .map(Encoding::total) + let ceiling = candidates(encoders) + .into_iter() + .map(|(_, encoding)| encoding.total()) .max() - .map(|top| top.scaled(1.0 / HEADROOM)); + .map(|top| top.scaled(1.0 / headroom())); let connect = self .launch .connect @@ -425,19 +625,32 @@ impl Setup { ui.label("Encoders"); match &self.encoders { Task::Running(_) => { - ui.spinner(); + 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) => ui.label(format!( - "{}: {}{}, {}", - describe(*canvas), - encoding.encoder, - if encoding.gpu { " on the GPU" } else { " from the CPU" }, - mbps(encoding.bitrate) - )), + 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)), @@ -463,8 +676,8 @@ impl Setup { )); }); } - Some(Task::Done(Ok(rate))) => { - ui.label(format!("{} to the relay", mbps(*rate))); + 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}")); @@ -609,7 +822,8 @@ mod tests { use super::*; - /// [`CANVASES`] each encoding at its default bitrate, or failing where `None`. + /// [`CANVASES`] each encoding at its default bitrate and keeping up, or + /// failing where `None`. fn encoders(kbps: [Option; 5]) -> Encoders { CANVASES .into_iter() @@ -620,6 +834,7 @@ mod tests { encoder: "test".into(), gpu: true, bitrate: Rate::from_kbps(kbps), + frame: Duration::from_millis(1), }) .ok_or_else(|| "refused".to_string()); (canvas, result) @@ -627,17 +842,92 @@ mod tests { .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 three quarters of 6 Mbps. - assert_eq!(pick(&encoders, Some(Rate::from_mbps(6))), CANVASES[1]); - assert_eq!(pick(&encoders, Some(Rate::from_kbps(5900))), CANVASES[2]); + // 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]); From 2532532ef88e8998317bf3f9fc8210c95fd0cc7a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 19:25:16 -0800 Subject: [PATCH 6/8] fix(studio): keep the chosen canvas when setup starts empty Co-Authored-By: Claude Opus 5.5 --- README.md | 4 ++-- rs/moq-studio/src/setup.rs | 11 ++++++++--- 2 files changed, 10 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index e74ce8b..127cd84 100644 --- a/README.md +++ b/README.md @@ -55,8 +55,8 @@ 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 nothing in -it. +empty scene** both create the scene file at once, the second with the chosen +canvas and no sources. 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 diff --git a/rs/moq-studio/src/setup.rs b/rs/moq-studio/src/setup.rs index 4a09981..72b8be8 100644 --- a/rs/moq-studio/src/setup.rs +++ b/rs/moq-studio/src/setup.rs @@ -756,14 +756,19 @@ impl Setup { start } - /// Start the studio with the proposal, or with an empty collection when - /// `empty`. Either way its autosave creates the scene file at once. + /// 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 collection = match empty { - true => Collection::new(), + true => { + let mut collection = Collection::new(); + collection.canvas = choices.canvas; + collection + } false => collection(listing, choices), }; // Kept, so a start that fails can be retried with other choices. From c3e9c7b31095d0a429242c4e3d874d60df785b9a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 21:54:32 -0800 Subject: [PATCH 7/8] fix(studio): check for a scene file that appeared during setup through the autosave The autosave's own read is the check, and its first save refuses a file created after it, so no window remains where setup could replace one. Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/autosave.rs | 9 ++++++++- rs/moq-studio/src/setup.rs | 10 ++++------ 2 files changed, 12 insertions(+), 7 deletions(-) diff --git a/rs/moq-studio/src/autosave.rs b/rs/moq-studio/src/autosave.rs index 7e0d801..9a1c7c0 100644 --- a/rs/moq-studio/src/autosave.rs +++ b/rs/moq-studio/src/autosave.rs @@ -152,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() @@ -177,7 +183,8 @@ impl Autosave { written.collection = Some(text.collection); } if written.history != text.history { - write(&self.history, &text.history, false).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(()) diff --git a/rs/moq-studio/src/setup.rs b/rs/moq-studio/src/setup.rs index 6c7a7d1..ca39789 100644 --- a/rs/moq-studio/src/setup.rs +++ b/rs/moq-studio/src/setup.rs @@ -789,15 +789,13 @@ impl Setup { self.starting = Some(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. - let exists = path - .try_exists() - .with_context(|| format!("reading {}", path.display()))?; + // The autosave's first save refuses one that appears after this. + let (autosave, _) = Autosave::open(path)?; anyhow::ensure!( - !exists, + !autosave.exists(), "{} was created during setup; restart the studio to open it", - path.display() + autosave.path().display() ); - let (autosave, _) = Autosave::open(path)?; Engine::start(launch, autosave, loaded).await })); } From 90efa11a39abb95126a99654f90784bf08905532 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 06:29:42 -0800 Subject: [PATCH 8/8] fix(studio): run setup's probes on threads the runtime does not wait for Dropping the runtime waits out its blocking pool, so a probe stuck on a camera permission prompt held the studio open after its window closed. Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/main.rs | 2 +- rs/moq-studio/src/setup.rs | 54 +++++++++++++++++++++++--------------- 2 files changed, 34 insertions(+), 22 deletions(-) diff --git a/rs/moq-studio/src/main.rs b/rs/moq-studio/src/main.rs index b166847..6afdca5 100644 --- a/rs/moq-studio/src/main.rs +++ b/rs/moq-studio/src/main.rs @@ -96,7 +96,7 @@ fn main() -> anyhow::Result<()> { launch, config.scenes.clone(), runtime.handle().clone(), - ))), + )?)), }; let started = Arc::new(OnceLock::new()); diff --git a/rs/moq-studio/src/setup.rs b/rs/moq-studio/src/setup.rs index ca39789..103f0be 100644 --- a/rs/moq-studio/src/setup.rs +++ b/rs/moq-studio/src/setup.rs @@ -466,19 +466,22 @@ enum Task { } impl Task { - /// Run `task` on a blocking thread of `runtime`, since encoder probes and the - /// engine's start block while they open encoders. - fn spawn(runtime: &tokio::runtime::Handle, task: F) -> Self + /// 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 handle = runtime.clone(); - runtime.spawn_blocking(move || { - let _ = send.send(handle.block_on(task())); - }); - Self::Running(recv) + 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. @@ -519,11 +522,11 @@ pub struct Setup { impl Setup { /// Start probing for `launch`, to create the scene file at `path`. - pub fn new(launch: Launch, path: PathBuf, runtime: tokio::runtime::Handle) -> Self { - let devices = Task::spawn(&runtime, devices::list); + 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)); - Self { + let encoders = Task::spawn(&runtime, move || probe_encoders(gpu))?; + Ok(Self { runtime, launch, path, @@ -533,7 +536,7 @@ impl Setup { choices: None, starting: None, error: None, - } + }) } /// Draw the screen; the engine once the creator has started the studio. @@ -594,11 +597,14 @@ impl Setup { 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)) => Task::spawn(&self.runtime, move || async move { - uplink::measure(&client, url, &path, ceiling, uplink::WINDOW) - .await - .map_err(|err| format!("{err:#}")) - }), + (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 { @@ -785,8 +791,7 @@ impl Setup { // Kept, so a start that fails can be retried with other choices. let launch = self.launch.clone(); let path = self.path.clone(); - self.error = None; - self.starting = Some(Task::spawn(&self.runtime, move || async move { + 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. @@ -797,7 +802,14 @@ impl Setup { 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}")), + } } }