From ea49628b86f8c5c4bb925a007938aaaddecd1912 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 14:42:13 -0800 Subject: [PATCH 1/5] chore(quest): claim quest/m0/vcam/linux.md Quest-Claim: 131a812fdb42e147 From 81c621d5ff85dcdc51b117eb55447abb6ea7ba9f Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 15:09:47 -0800 Subject: [PATCH 2/5] feat(studio): a Linux virtual camera, as a PipeWire node or a v4l2loopback device The Output panel's Virtual camera toggle shows the program as a camera: a PipeWire Video/Source node with the Camera role (the `pipewire` feature), falling back to an existing v4l2loopback device. The program is read back for it only while an app reads the camera. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/nightly.yml | 35 ++- Cargo.lock | 2 + Cargo.toml | 1 + README.md | 23 +- justfile | 12 +- quest/m0/vcam/README.md | 8 +- quest/m0/vcam/linux.md | 7 - rs/moq-studio/Cargo.toml | 16 +- rs/moq-studio/src/engine.rs | 106 ++++++++- rs/moq-studio/src/main.rs | 1 + rs/moq-studio/src/ui.rs | 33 +++ rs/moq-studio/src/vcam.rs | 341 ++++++++++++++++++++++++++++ rs/moq-studio/src/vcam/pipewire.rs | 343 +++++++++++++++++++++++++++++ rs/moq-studio/src/vcam/v4l2.rs | 150 +++++++++++++ sh/pipewire-session.sh | 46 ++++ 15 files changed, 1103 insertions(+), 21 deletions(-) delete mode 100644 quest/m0/vcam/linux.md create mode 100644 rs/moq-studio/src/vcam.rs create mode 100644 rs/moq-studio/src/vcam/pipewire.rs create mode 100644 rs/moq-studio/src/vcam/v4l2.rs create mode 100755 sh/pipewire-session.sh diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml index 8a4c8d2c..f401c0f4 100644 --- a/.github/workflows/nightly.yml +++ b/.github/workflows/nightly.yml @@ -1,7 +1,9 @@ name: Nightly -# The gates too slow or too specialized for a pull request: compositing and -# publishing through wgpu on Mesa's software Vulkan driver. +# The gates too slow or too specialized for a pull request: compositing, +# publishing, and the virtual camera through wgpu on Mesa's software Vulkan +# driver, the camera in a private PipeWire session; and the virtual camera's +# v4l2loopback fallback, a module load no pull request job should make. permissions: contents: read @@ -34,3 +36,32 @@ jobs: - name: Render run: nix develop --command just render-sw + + v4l2loopback: + name: v4l2loopback camera + runs-on: ubuntu-24.04 + timeout-minutes: 30 + env: + NEXTEST_PROFILE: ci + + steps: + - name: Checkout + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + submodules: true + + - uses: ./.github/actions/nix + + - name: Load v4l2loopback + # Out of tree, so DKMS builds it for the runner's kernel. + run: | + sudo apt-get update + sudo apt-get install -y "linux-headers-$(uname -r)" v4l2loopback-dkms + sudo modprobe v4l2loopback exclusive_caps=1 + sudo udevadm settle + # The runner user is not in the video group. + sudo chmod a+rw /dev/video* + + - name: Test + run: nix develop --command just v4l2loopback diff --git a/Cargo.lock b/Cargo.lock index c3f6f1d3..136c563b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3465,7 +3465,9 @@ dependencies = [ "moq-relay", "moq-scene", "moq-tokio", + "moq-v4l", "moq-video", + "pipewire", "pollster", "tempfile", "tokio", diff --git a/Cargo.toml b/Cargo.toml index feecb17b..5e5d1a90 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -31,6 +31,7 @@ moq-net = { path = "moq/rs/moq-net" } moq-relay = { path = "moq/rs/moq-relay", default-features = false } moq-scene = { path = "rs/moq-scene" } moq-tokio = { path = "moq/rs/moq-tokio", default-features = false } +moq-v4l = { path = "moq/rs/moq-v4l" } # Default features off, as in moq: each consumer opts into native codecs and # rendering explicitly. moq-video = { path = "moq/rs/moq-video", default-features = false } diff --git a/README.md b/README.md index 6da3900a..0bab5d55 100644 --- a/README.md +++ b/README.md @@ -17,8 +17,9 @@ cargo run --release -p moq-studio --features capture,openh264 -- \ - 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. - On Linux it links ALSA; add `pipewire` for PipeWire cameras. `openh264` - is the software H.264 fallback for a machine without a hardware encoder. + On Linux it links ALSA; add `pipewire` for PipeWire cameras and the + [virtual camera](#virtual-camera)'s PipeWire node. `openh264` is the + software H.264 fallback for a machine without a hardware encoder. - On Linux, display and window sources capture X11 and XWayland only; a Wayland screen or window cannot be picked yet. - The `--connect-*`, `--quic-*`, and `--log-level` flags are @@ -36,7 +37,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, and the encoder's work, and turn the virtual camera on. | A broadcast source subscribes through the same relay connection, plays the best video and audio its catalog lists, switches when the catalog changes, and @@ -117,6 +118,22 @@ macOS. Every other source, every source on Windows or on a GPU that cannot import, and every still image delivers its frames from the CPU, which the compositor uploads. +## Virtual camera + +The **Virtual camera** toggle in the Output panel shows the program as a +camera that video-call and conferencing apps can pick, at the canvas size and +frame rate. It stays on until you turn it off or quit; it is not saved in the +scene file. The program is copied for the camera only while an app reads it. + +- **Linux:** a PipeWire camera named "MoQ Studio" (needs the `pipewire` + feature), which browsers and other PipeWire-aware apps list beside your + other cameras. Without PipeWire, the studio writes to the first + [v4l2loopback](https://github.com/v4l2loopback/v4l2loopback) device that + takes the canvas size, for apps that open `/dev/video*` directly. The + studio never loads that module, since it needs root: + `sudo modprobe v4l2loopback exclusive_caps=1 card_label="MoQ Studio"`. +- **macOS and Windows:** not available yet. + ## Development Use the Nix dev shell (`direnv allow`, or `nix develop --command ...`) so diff --git a/justfile b/justfile index 2bd9ae7b..8085afcb 100644 --- a/justfile +++ b/justfile @@ -1,7 +1,7 @@ set shell := ["bash", "-euo", "pipefail", "-c"] # The device and codec features off by default, which every check covers too. -features := "moq-scene/capture,moq-studio/capture,moq-studio/openh264" +features := "moq-scene/capture,moq-studio/capture,moq-studio/openh264,moq-studio/pipewire" # List the available development commands. default: @@ -38,8 +38,8 @@ fix: nixfmt flake.nix cargo fmt --all -# Composite scenes and publish a program through wgpu on Mesa's software Vulkan -# driver, so no GPU is needed. +# Composite scenes, publish a program, and show it as a PipeWire camera through +# wgpu on Mesa's software Vulkan driver, so no GPU is needed. render-sw: #!/usr/bin/env bash set -euo pipefail @@ -49,6 +49,12 @@ render-sw: 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)' + sh/pipewire-session.sh cargo nextest run --locked -p moq-studio --features pipewire --run-ignored only -E 'test(=engine::tests::the_virtual_camera_reaches_a_pipewire_app)' + +# Show the program through a v4l2loopback device, which needs the module loaded +# (`sudo modprobe v4l2loopback exclusive_caps=1`) and the device writable. +v4l2loopback: + cargo nextest run --locked -p moq-studio --run-ignored only -E 'test(=vcam::v4l2::tests::a_v4l2_app_reads_the_program)' # 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/vcam/README.md b/quest/m0/vcam/README.md index f77993ed..dfacdefb 100644 --- a/quest/m0/vcam/README.md +++ b/quest/m0/vcam/README.md @@ -18,8 +18,14 @@ Decided 2026-10-22: - **macOS:** a CoreMediaIO Camera Extension, which needs signing. - **Order:** the first backend to land adds the studio toggle. +The Linux backend landed the toggle: **Virtual camera** in the Output panel, +for the one canvas, not saved in the scene file. Each OS backend is a +`vcam::backend` in `rs/moq-studio`, fed tightly packed RGBA program frames on +the `vcam` thread only while an app reads the camera. With +[multiple canvases](/quest/m0/multi-canvas.md), the toggle moves to each +canvas's outputs. + ## Required -- [Linux virtual camera](/quest/m0/vcam/linux.md) - a PipeWire video node carrying the program - [Windows virtual camera](/quest/m0/vcam/windows.md) - a Media Foundation virtual camera on Windows 11 - [macOS virtual camera](/quest/m0/vcam/macos.md) - a CoreMediaIO Camera Extension inside the app bundle diff --git a/quest/m0/vcam/linux.md b/quest/m0/vcam/linux.md deleted file mode 100644 index c5c22c6c..00000000 --- a/quest/m0/vcam/linux.md +++ /dev/null @@ -1,7 +0,0 @@ -# [M] Linux virtual camera - -## Goal - -The studio publishes a canvas's program as a PipeWire video node that -PipeWire-aware apps (browsers and conferencing apps) list as a camera. It -falls back to an existing v4l2loopback device when one is present. diff --git a/rs/moq-studio/Cargo.toml b/rs/moq-studio/Cargo.toml index 8a0312d7..a71e1dcf 100644 --- a/rs/moq-studio/Cargo.toml +++ b/rs/moq-studio/Cargo.toml @@ -28,10 +28,10 @@ nvidia = ["moq-video/nvidia"] openh264 = ["moq-video/openh264"] # Intel/AMD VAAPI hardware codecs (Linux); its bindgen needs libclang. vaapi = ["moq-video/vaapi"] -# PipeWire cameras, drawn without a copy, and audio through the sound server. -# Wayland screens and windows are not reachable yet: display and window sources -# capture X11 only. -pipewire = ["capture", "moq-scene/pipewire"] +# PipeWire cameras, drawn without a copy, audio through the sound server, and +# the virtual camera as a PipeWire node. Wayland screens and windows are not +# reachable yet: display and window sources capture X11 only. +pipewire = ["capture", "moq-scene/pipewire", "dep:pipewire"] [dependencies] anyhow = { workspace = true, features = ["backtrace"] } @@ -55,6 +55,14 @@ tracing = { workspace = true } url = { workspace = true } usage-rs = { workspace = true, features = ["completions", "config"] } +[target.'cfg(target_os = "linux")'.dependencies] +# The virtual camera's v4l2loopback fallback. Pure Rust with its kernel bindings +# checked in, and already in the graph through moq-video's capture. +moq-v4l = { workspace = true } +# The virtual camera's PipeWire node, behind `pipewire`. moq-video links the +# same release; `v0_3_34` adds the trigger a driving stream paces frames with. +pipewire = { version = "0.10", optional = true, features = ["v0_3_34"] } + [dev-dependencies] moq-relay = { workspace = true, features = ["test-support"] } tempfile = { workspace = true } diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 9ccecccb..1c8a67a6 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -6,6 +6,8 @@ //! program to the encoder, stamped with the program clock. //! - **audio** reads the program bus on the same clock and publishes it. //! - **encode** waits for each program copy and encodes it. +//! - **vcam** waits for each program copy the virtual camera takes, while it is +//! on, and hands it to the camera's backend. //! - **source-***: one per running source, since a capture stream may not //! leave the thread that opened it. //! - The UI runs on the main thread and edits [`State`] under its lock. @@ -67,6 +69,8 @@ pub struct State { pub collection: Collection, pub sources: Sources, pub compositor: Compositor, + /// The canvas's virtual camera, while it is on. + camera: Option, /// Edited since the sources were last synced. stale: bool, /// Edited since the last save or load. @@ -85,6 +89,11 @@ impl State { self.unsaved } + /// How the virtual camera is doing, or `None` while it is off. + pub fn camera(&self) -> Option { + self.camera.as_ref().map(crate::vcam::Camera::status) + } + pub fn save(&mut self) -> anyhow::Result<()> { let text = self.collection.to_toml()?; std::fs::write(&self.path, text).with_context(|| format!("writing {}", self.path.display()))?; @@ -176,6 +185,7 @@ impl Drop for Output { /// The running studio. pub struct Engine { + gpu: Gpu, pub state: Mutex, pub stats: Stats, pub bus: moq_audio::bus::Bus, @@ -229,11 +239,13 @@ impl Engine { }; let engine = Arc::new(Self { + gpu, state: Mutex::new(State { path, collection, sources, compositor, + camera: None, stale: true, unsaved: false, }), @@ -284,9 +296,25 @@ impl Engine { Ok(engine) } + /// Turn the canvas's virtual camera on or off. Turning it on fails, saying + /// why, when this machine has no way to publish one. + pub fn set_camera(&self, on: bool) -> anyhow::Result<()> { + let canvas = self.state.lock().unwrap().collection.canvas; + // Opened and closed outside the lock, so the program clock does not + // wait on the camera's backend. + let camera = match on { + true => Some(crate::vcam::Camera::open(&self.gpu.device, &self.gpu.queue, canvas)?), + false => None, + }; + let previous = std::mem::replace(&mut self.state.lock().unwrap().camera, camera); + drop(previous); + Ok(()) + } + /// Draw a canvas frame per tick of the program clock, and hand the program /// to the encoder, stamped `epoch` past `start`: every frame, or only while - /// anyone watches when the output is on demand. + /// anyone watches when the output is on demand. The virtual camera, while + /// on, takes every frame an app reads. /// /// Drawing outlives the encoder: a failed output stops copying the program, /// and the preview and program keep moving. @@ -317,6 +345,9 @@ impl Engine { tracing::error!(%err, "cannot draw the canvas"); continue; } + if let Some(camera) = &mut state.camera { + camera.offer(state.compositor.program()); + } state.compositor.program().clone() }; self.stats.rendered.fetch_add(1, Ordering::Relaxed); @@ -712,4 +743,77 @@ mod tests { .await .expect("sound arrives"); } + + /// The virtual camera reaches an app that opens PipeWire cameras as the + /// picture the studio composed. + #[cfg(all(target_os = "linux", feature = "pipewire"))] + #[tokio::test(flavor = "multi_thread")] + #[ignore = "needs a PipeWire session and a Vulkan device; run by just render-sw"] + async fn the_virtual_camera_reaches_a_pipewire_app() { + let dir = tempfile::tempdir().unwrap(); + let mut collection = Collection::new(); + collection.canvas = moq_scene::Canvas { + width: 320, + height: 180, + framerate: 30, + }; + collection.sources.insert( + "red".into(), + moq_scene::Source::Color { + color: moq_scene::Rgba([255, 0, 0, 255]), + }, + ); + collection.scenes[0].items.push(moq_scene::Item { + source: "red".into(), + rect: moq_scene::Rect::fill(&collection.canvas), + visible: true, + }); + + let gpu = Gpu::new().await.expect("a GPU"); + let path = dir.path().join("show.toml"); + let engine = Engine::start(gpu, path, collection, None, "studio", false) + .await + .expect("the studio starts"); + engine.set_camera(true).expect("the camera turns on"); + let status = engine.state.lock().unwrap().camera().expect("a camera"); + assert_eq!(status.backend, "PipeWire"); + assert!(!status.reading, "nothing reads the camera yet"); + + // Opened the way an app opens a PipeWire camera: by role and class. + let mut config = moq_video::capture::Config::default(); + config.source = moq_video::capture::Source::Camera(Some("pipewire:moq-studio".into())); + config.width = Some(320); + config.height = Some(180); + let mut stream = kio::clock::timeout(TIMEOUT, moq_video::capture::open(&config)) + .await + .expect("the camera opens") + .expect("a camera stream"); + assert_eq!((stream.width(), stream.height()), (320, 180)); + // The canvas is black until the color source delivers its first + // picture, so wait for the composed one rather than take the first. + kio::clock::timeout(TIMEOUT, async { + loop { + let frame = stream.read().await.expect("capture").expect("the camera ended"); + let rgba = frame.surface.to_rgba(&Default::default()).expect("rgba"); + let center = (90 * 320 + 160) * 4; + let [r, g, b, _] = rgba.data()[center..center + 4] else { + unreachable!() + }; + // The reader converts to I420, which shifts a primary a little. + if r > 200 && g < 60 && b < 60 { + return; + } + } + }) + .await + .expect("the composed picture arrives"); + let status = engine.state.lock().unwrap().camera().expect("a camera"); + assert!(status.reading, "the camera does not know an app reads it"); + assert!(status.sent > 0); + assert_eq!(status.error, None); + + drop(stream); + engine.set_camera(false).expect("the camera turns off"); + assert!(engine.state.lock().unwrap().camera().is_none()); + } } diff --git a/rs/moq-studio/src/main.rs b/rs/moq-studio/src/main.rs index 95275251..87923e4d 100644 --- a/rs/moq-studio/src/main.rs +++ b/rs/moq-studio/src/main.rs @@ -11,6 +11,7 @@ use anyhow::Context; mod engine; mod ui; +mod vcam; #[derive(usage_rs::Cli, Clone)] #[usage(unknown_flags = "error", args_override_self = false)] diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index e1c4f92d..9cc6d848 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -559,6 +559,39 @@ impl Panels { if let Some(error) = stats.error.lock().unwrap().as_ref() { ui.colored_label(ui.visuals().error_fg_color, error); } + + ui.separator(); + self.camera(ui); + } + + /// The canvas's virtual camera: a toggle, and how it is doing while on. + fn camera(&mut self, ui: &mut Ui) { + let camera = self.engine.state.lock().unwrap().camera(); + let mut on = camera.is_some(); + if ui + .checkbox(&mut on, "Virtual camera") + .on_hover_text("Show the program as a camera that video-call apps can pick") + .changed() + { + let result = self.engine.set_camera(on); + let message = if on { "virtual camera on" } else { "virtual camera off" }; + self.report(result.map(|()| message.to_string())); + } + let Some(camera) = camera else { return }; + egui::Grid::new("camera").show(ui, |ui| { + ui.label("Camera"); + ui.label(&camera.backend); + ui.end_row(); + ui.label("Apps"); + ui.label(if camera.reading { "reading" } else { "none; not copying" }); + ui.end_row(); + ui.label("Sent"); + ui.label(format!("{} frames, {} dropped", camera.sent, camera.dropped)); + ui.end_row(); + }); + if let Some(error) = &camera.error { + ui.colored_label(ui.visuals().error_fg_color, error); + } } } diff --git a/rs/moq-studio/src/vcam.rs b/rs/moq-studio/src/vcam.rs new file mode 100644 index 00000000..8c0a4f75 --- /dev/null +++ b/rs/moq-studio/src/vcam.rs @@ -0,0 +1,341 @@ +//! The program as a system camera, for video-call and conferencing apps that +//! offer no ingest. +//! +//! On Linux the camera is a PipeWire video node (with the `pipewire` feature), +//! which PipeWire-aware apps such as browsers list as a camera. Without +//! PipeWire it is an existing v4l2loopback device, since that kernel module +//! needs root to load and so is never the first choice. +//! +//! The program clock offers every canvas frame; [`Camera::offer`] copies it +//! into a staging buffer only while an app reads the camera, and the `vcam` +//! thread waits for the copy and hands the pixels to the backend. + +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, mpsc}; + +use anyhow::Context; +use moq_scene::{Canvas, wgpu}; + +#[cfg(all(target_os = "linux", feature = "pipewire"))] +mod pipewire; +#[cfg(target_os = "linux")] +mod v4l2; + +/// Copies in flight at once: one being handed over, one being copied, and one +/// spare, as the encoder's export keeps. +const DEPTH: usize = 3; + +/// A canvas's program, published as a camera other apps can open. +/// +/// Dropping it removes the camera. +pub struct Camera { + /// Where the camera lives, for the Output panel. + backend: String, + readback: Readback, + /// `None` once dropping. + frames: Option>, + worker: Option>, + shared: Arc, +} + +/// What the camera's threads report to the program clock and the UI. +#[derive(Default)] +struct Shared { + /// Whether an app reads the camera, so frames are worth copying. + reading: AtomicBool, + /// Frames handed to the backend. + sent: AtomicU64, + /// Frames not copied because every staging buffer was still in flight. + dropped: AtomicU64, + /// Set with `error`, so the program clock checks it without a lock. + failed: AtomicBool, + /// Why the camera stopped, if it did. + error: Mutex>, +} + +impl Shared { + fn fail(&self, err: impl std::fmt::Display) { + tracing::error!(%err, "virtual camera stopped"); + self.error.lock().unwrap().get_or_insert_with(|| err.to_string()); + self.failed.store(true, Ordering::Relaxed); + } + + fn failed(&self) -> bool { + self.failed.load(Ordering::Relaxed) + } + + /// Whether a frame is worth copying: an app reads the camera, which has + /// not stopped. + fn wanted(&self) -> bool { + self.reading.load(Ordering::Relaxed) && !self.failed() + } +} + +/// How the camera is doing, for the Output panel. +pub struct Status { + /// Where the camera lives, such as a PipeWire node or a `/dev/videoN` path. + pub backend: String, + /// Whether an app reads the camera. + pub reading: bool, + pub sent: u64, + pub dropped: u64, + /// Why the camera stopped, if it did. + pub error: Option, +} + +/// Where the camera's frames go, on the `vcam` thread. +trait Sink: Send { + /// Show one picture: tightly packed RGBA rows at the canvas size. + fn write(&mut self, rgba: &[u8]) -> anyhow::Result<()>; +} + +impl Camera { + /// Publish `canvas`-sized programs drawn on `device` as a camera. + /// + /// Fails, naming why each backend refused, when this machine has no way to + /// publish one. + pub fn open(device: &wgpu::Device, queue: &wgpu::Queue, canvas: Canvas) -> anyhow::Result { + let size = moq_video::Size::new(canvas.width, canvas.height); + let shared = Arc::new(Shared::default()); + let (sink, backend) = backend(size, canvas.framerate, &shared)?; + tracing::info!(%backend, "virtual camera on"); + + let (frames, pending) = mpsc::sync_channel(DEPTH); + let worker = { + let device = device.clone(); + let shared = shared.clone(); + std::thread::Builder::new() + .name("vcam".into()) + .spawn(move || run(&device, pending, sink, &shared)) + .context("cannot start the virtual camera thread")? + }; + Ok(Self { + backend, + readback: Readback::new(device, queue, size), + frames: Some(frames), + worker: Some(worker), + shared, + }) + } + + /// Copy `program` to the camera, if an app reads it. + /// + /// The copy is submitted before this returns, so the caller draws into + /// `program` again freely. Never waits on the GPU. + pub fn offer(&mut self, program: &wgpu::Texture) { + if !self.shared.wanted() { + return; + } + let Some(frames) = &self.frames else { return }; + let Some(staged) = self.readback.copy(program) else { + self.shared.dropped.fetch_add(1, Ordering::Relaxed); + return; + }; + // The pool holds no more buffers than the channel, so it is never full; + // a send fails only once the thread stopped, having said why. + let _ = frames.try_send(staged); + } + + pub fn status(&self) -> Status { + Status { + backend: self.backend.clone(), + reading: self.shared.wanted(), + sent: self.shared.sent.load(Ordering::Relaxed), + dropped: self.shared.dropped.load(Ordering::Relaxed), + error: self.shared.error.lock().unwrap().clone(), + } + } +} + +impl Drop for Camera { + fn drop(&mut self) { + // Ends the thread, which removes the camera before this returns, so + // turning it straight back on finds the device free. + self.frames.take(); + if let Some(worker) = self.worker.take() { + let _ = worker.join(); + } + } +} + +/// The first backend that takes a `size` camera at `framerate`. +#[cfg(target_os = "linux")] +fn backend(size: moq_video::Size, framerate: u32, shared: &Arc) -> anyhow::Result<(Box, String)> { + #[cfg(feature = "pipewire")] + let pipewire = match pipewire::Node::open(size, framerate, shared.clone()) { + Ok(node) => return Ok((Box::new(node), "PipeWire".to_string())), + Err(err) => { + tracing::warn!(err = %format!("{err:#}"), "no PipeWire camera, trying v4l2loopback"); + format!("{err:#}") + } + }; + #[cfg(not(feature = "pipewire"))] + let pipewire = "this build has no `pipewire` feature".to_string(); + + match v4l2::Loopback::open(size, framerate) { + Ok(loopback) => { + let backend = loopback.name(); + // The kernel keeps no count of readers to wait for. + shared.reading.store(true, Ordering::Relaxed); + Ok((Box::new(loopback), backend)) + } + Err(err) => anyhow::bail!("no virtual camera. PipeWire: {pipewire}. v4l2loopback: {err:#}"), + } +} + +#[cfg(not(target_os = "linux"))] +fn backend(size: moq_video::Size, framerate: u32, shared: &Arc) -> anyhow::Result<(Box, String)> { + let _ = (size, framerate, shared); + anyhow::bail!("the virtual camera is not available on this platform yet") +} + +/// Hand each copy to `sink` as it lands, until the camera is dropped or fails. +fn run(device: &wgpu::Device, pending: mpsc::Receiver, mut sink: Box, shared: &Shared) { + let mut rgba = Vec::new(); + while let Ok(staged) = pending.recv() { + if let Err(err) = staged.read(device, &mut rgba).and_then(|()| sink.write(&rgba)) { + shared.fail(format!("{err:#}")); + return; + } + shared.sent.fetch_add(1, Ordering::Relaxed); + } +} + +/// Copies the program into a small pool of staging buffers the CPU can map. +struct Readback { + device: wgpu::Device, + queue: wgpu::Queue, + size: moq_video::Size, + /// Bytes per row in a staging buffer, padded to the copy alignment. + row: u32, + pool: Arc>, +} + +/// Staging buffers not held by a [`Staged`]. +#[derive(Default)] +struct Pool { + free: Vec, + /// Buffers in existence, up to [`DEPTH`]. + allocated: usize, +} + +impl Readback { + fn new(device: &wgpu::Device, queue: &wgpu::Queue, size: moq_video::Size) -> Self { + Self { + device: device.clone(), + queue: queue.clone(), + size, + row: (size.width * 4).next_multiple_of(wgpu::COPY_BYTES_PER_ROW_ALIGNMENT), + pool: Arc::default(), + } + } + + /// Start copying `texture`, or `None` when every buffer is in flight. + fn copy(&mut self, texture: &wgpu::Texture) -> Option { + let mut pool = self.pool.lock().unwrap(); + let buffer = match pool.free.pop() { + Some(buffer) => buffer, + None if pool.allocated < DEPTH => { + pool.allocated += 1; + self.device.create_buffer(&wgpu::BufferDescriptor { + label: Some("moq-studio vcam"), + size: u64::from(self.row) * u64::from(self.size.height), + usage: wgpu::BufferUsages::COPY_DST | wgpu::BufferUsages::MAP_READ, + mapped_at_creation: false, + }) + } + None => return None, + }; + drop(pool); + + let mut encoder = self.device.create_command_encoder(&wgpu::CommandEncoderDescriptor { + label: Some("moq-studio vcam"), + }); + encoder.copy_texture_to_buffer( + texture.as_image_copy(), + wgpu::TexelCopyBufferInfo { + buffer: &buffer, + layout: wgpu::TexelCopyBufferLayout { + offset: 0, + bytes_per_row: Some(self.row), + rows_per_image: Some(self.size.height), + }, + }, + wgpu::Extent3d { + width: self.size.width, + height: self.size.height, + depth_or_array_layers: 1, + }, + ); + let submission = self.queue.submit([encoder.finish()]); + Some(Staged { + buffer: Some(buffer), + submission, + size: self.size, + row: self.row, + pool: self.pool.clone(), + mapping: false, + }) + } +} + +/// One copy in flight. Dropping it returns its buffer to the pool. +struct Staged { + buffer: Option, + submission: wgpu::SubmissionIndex, + size: moq_video::Size, + row: u32, + pool: Arc>, + /// Set while a mapping was requested and not yet undone, so a read that + /// failed partway does not hand a mapped buffer to the next copy. + mapping: bool, +} + +impl Staged { + /// Block until the copy lands, then fill `rgba` with tightly packed rows. + fn read(mut self, device: &wgpu::Device, rgba: &mut Vec) -> anyhow::Result<()> { + let buffer = self.buffer.as_ref().expect("held until drop"); + let (send, recv) = mpsc::channel(); + self.mapping = true; + buffer.map_async(wgpu::MapMode::Read, .., move |result| { + let _ = send.send(result); + }); + device + .poll(wgpu::PollType::Wait { + submission_index: Some(self.submission.clone()), + timeout: None, + }) + .map_err(|err| anyhow::anyhow!("waiting for the program copy: {err}"))?; + recv.recv() + .context("the program copy was dropped")? + .context("mapping the program copy")?; + + let width = self.size.width as usize * 4; + rgba.clear(); + { + let mapped = buffer + .slice(..) + .get_mapped_range() + .map_err(|err| anyhow::anyhow!("reading the program copy: {err}"))?; + for row in mapped.chunks_exact(self.row as usize) { + rgba.extend_from_slice(&row[..width]); + } + } + buffer.unmap(); + self.mapping = false; + Ok(()) + } +} + +impl Drop for Staged { + fn drop(&mut self) { + let Some(buffer) = self.buffer.take() else { return }; + let mut pool = self.pool.lock().unwrap(); + // One a failed read left mapped is let go and replaced. + if self.mapping { + pool.allocated -= 1; + } else { + pool.free.push(buffer); + } + } +} diff --git a/rs/moq-studio/src/vcam/pipewire.rs b/rs/moq-studio/src/vcam/pipewire.rs new file mode 100644 index 00000000..c7a1d28a --- /dev/null +++ b/rs/moq-studio/src/vcam/pipewire.rs @@ -0,0 +1,343 @@ +//! The camera as a PipeWire video node. +//! +//! A `Video/Source` node with the `Camera` role is what PipeWire-aware apps +//! (browsers through the camera portal, conferencing apps) list as a camera. +//! It offers one format, the canvas's RGBA at its size and frame rate, so the +//! program reaches an app without a conversion here. +//! +//! PipeWire objects stay on the thread that made them, so the node runs its +//! own main loop. The `vcam` thread leaves the newest picture in a [`Slot`] +//! and wakes the loop; the loop copies it into a PipeWire buffer. + +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, mpsc}; + +use anyhow::Context; +use moq_video::Size; +use pipewire as pw; +use pw::spa; +use spa::param::format::{FormatProperties, MediaSubtype, MediaType}; +use spa::param::video::{VideoFormat, VideoInfoRaw}; + +use super::{Shared, Sink}; + +/// The node's name, which `pw-cli` and pickers that show ids print. +const NAME: &str = "moq-studio"; + +/// What apps list the camera as. +const DESCRIPTION: &str = "MoQ Studio"; + +pub struct Node { + slot: Arc, + wake: pw::channel::Sender, + thread: Option>, + shared: Arc, +} + +/// The newest picture, handed from the `vcam` thread to the node's loop. +#[derive(Default)] +struct Slot { + picture: Mutex>, + /// A wake is on its way to the loop, so another is not sent: the channel + /// never holds more than one. + woken: AtomicBool, +} + +enum Message { + /// A new picture is in the slot. + Picture, + Quit, +} + +impl Node { + /// Add a `size` camera node at `framerate` to the session's PipeWire. + /// + /// Fails when there is no PipeWire session to add it to. + pub fn open(size: Size, framerate: u32, shared: Arc) -> anyhow::Result { + let slot = Arc::new(Slot::default()); + let (wake, messages) = pw::channel::channel(); + let (ready, opened) = mpsc::channel(); + let thread = { + let slot = slot.clone(); + let shared = shared.clone(); + std::thread::Builder::new() + .name("vcam-pipewire".into()) + .spawn(move || { + if let Err(err) = run(size, framerate, slot, shared, messages, &ready) { + let _ = ready.send(Err(format!("{err:#}"))); + } + }) + .context("cannot start the PipeWire thread")? + }; + match opened.recv() { + Ok(Ok(())) => Ok(Self { + slot, + wake, + thread: Some(thread), + shared, + }), + Ok(Err(err)) => { + let _ = thread.join(); + Err(anyhow::anyhow!(err)) + } + Err(_) => { + let _ = thread.join(); + anyhow::bail!("the PipeWire thread stopped") + } + } + } +} + +impl Sink for Node { + fn write(&mut self, rgba: &[u8]) -> anyhow::Result<()> { + if self.shared.failed() { + anyhow::bail!("the PipeWire node stopped"); + } + { + let mut picture = self.slot.picture.lock().unwrap(); + picture.clear(); + picture.extend_from_slice(rgba); + } + if !self.slot.woken.swap(true, Ordering::AcqRel) && self.wake.send(Message::Picture).is_err() { + anyhow::bail!("the PipeWire node stopped"); + } + Ok(()) + } +} + +impl Drop for Node { + fn drop(&mut self) { + let _ = self.wake.send(Message::Quit); + if let Some(thread) = self.thread.take() { + let _ = thread.join(); + } + } +} + +/// Add the node and run it until it is dropped or PipeWire ends it, saying on +/// `ready` once it is added. Fails only before that. +fn run( + size: Size, + framerate: u32, + slot: Arc, + shared: Arc, + messages: pw::channel::Receiver, + ready: &mpsc::Sender>, +) -> anyhow::Result<()> { + pw::init(); + let mainloop = pw::main_loop::MainLoopRc::new(None).context("PipeWire main loop")?; + let context = pw::context::ContextRc::new(&mainloop, None).context("PipeWire context")?; + let core = context.connect_rc(None).context("connecting to PipeWire")?; + let stream = pw::stream::StreamRc::new( + core, + NAME, + pw::properties::properties! { + *pw::keys::MEDIA_TYPE => "Video", + *pw::keys::MEDIA_CATEGORY => "Capture", + *pw::keys::MEDIA_CLASS => "Video/Source", + *pw::keys::MEDIA_ROLE => "Camera", + *pw::keys::NODE_NAME => NAME, + *pw::keys::NODE_DESCRIPTION => DESCRIPTION, + }, + ) + .context("PipeWire stream")?; + + let stride = size.width * 4; + let _listener = stream + .add_local_listener::<()>() + .state_changed({ + let mainloop = mainloop.downgrade(); + let shared = shared.clone(); + move |_, _, _, state| { + tracing::debug!(?state, "virtual camera node"); + // Streaming once an app links to the node, paused when it leaves. + let reading = matches!(state, pw::stream::StreamState::Streaming); + shared.reading.store(reading, Ordering::Relaxed); + // Our own teardown quits the loop before anything disconnects, so + // either of these is PipeWire ending the node. + let ended = match state { + pw::stream::StreamState::Error(err) => Some(format!("PipeWire: {err}")), + pw::stream::StreamState::Unconnected => Some("the PipeWire session ended".to_string()), + _ => None, + }; + if let Some(ended) = ended { + shared.fail(ended); + if let Some(mainloop) = mainloop.upgrade() { + mainloop.quit(); + } + } + } + }) + .param_changed({ + let shared = shared.clone(); + move |stream, _, id, param| { + let Some(param) = param else { return }; + if id != spa::param::ParamType::Format.as_raw() { + return; + } + // The offer is fixed, so the format an app picks can only be it. + let mut info = VideoInfoRaw::new(); + if info.parse(param).is_err() || info.format() != VideoFormat::RGBA { + shared.fail(format!("an app negotiated {:?}, not RGBA", info.format())); + return; + } + let buffers = buffers(size, stride); + let Some(buffers) = spa::pod::Pod::from_bytes(&buffers) else { + return; + }; + if let Err(err) = stream.update_params(&mut [buffers]) { + shared.fail(format!("PipeWire buffers: {err}")); + } + } + }) + .process({ + let slot = slot.clone(); + let shared = shared.clone(); + move |stream, _| { + let picture = slot.picture.lock().unwrap(); + // Nothing to show before the first picture lands. + if picture.is_empty() { + return; + } + // Every buffer is with the app; it gets the next picture instead. + let Some(mut buffer) = stream.dequeue_buffer() else { + return; + }; + let Some(data) = buffer.datas_mut().first_mut() else { + return; + }; + let Some(target) = data.data().and_then(|target| target.get_mut(..picture.len())) else { + // Sized by our own buffers param, so this is PipeWire + // handing over something else. + shared.fail("PipeWire handed over a buffer smaller than a picture"); + return; + }; + target.copy_from_slice(&picture); + let chunk = data.chunk_mut(); + *chunk.offset_mut() = 0; + *chunk.size_mut() = picture.len() as u32; + *chunk.stride_mut() = stride as i32; + } + }) + .register() + .context("PipeWire stream listener")?; + + let format = format(size, framerate); + let format = spa::pod::Pod::from_bytes(&format).context("the PipeWire format offer")?; + // Driving, so the node runs on the program clock: each picture triggers a + // graph cycle that hands it to whichever app reads the camera. No + // autoconnect: an app links to the node when it opens the camera. + stream + .connect( + spa::utils::Direction::Output, + None, + pw::stream::StreamFlags::DRIVER | pw::stream::StreamFlags::MAP_BUFFERS, + &mut [format], + ) + .context("adding the PipeWire node")?; + + let _messages = messages.attach(mainloop.loop_(), { + let mainloop = mainloop.downgrade(); + let slot = slot.clone(); + let stream = stream.downgrade(); + move |message| match message { + Message::Picture => { + slot.woken.store(false, Ordering::Release); + let Some(stream) = stream.upgrade() else { return }; + // Another driver runs the graph and takes the newest picture on + // its own cycle; only a driving node triggers one. + if stream.is_driving() + && let Err(err) = stream.trigger_process() + { + tracing::debug!(%err, "cannot trigger the virtual camera"); + } + } + Message::Quit => { + if let Some(mainloop) = mainloop.upgrade() { + mainloop.quit(); + } + } + } + }); + + if ready.send(Ok(())).is_err() { + return Ok(()); + } + mainloop.run(); + Ok(()) +} + +/// The one format the node offers: `size` RGBA at `framerate`. +fn format(size: Size, framerate: u32) -> Vec { + let object = spa::pod::object!( + spa::utils::SpaTypes::ObjectParamFormat, + spa::param::ParamType::EnumFormat, + spa::pod::property!(FormatProperties::MediaType, Id, MediaType::Video), + spa::pod::property!(FormatProperties::MediaSubtype, Id, MediaSubtype::Raw), + spa::pod::property!(FormatProperties::VideoFormat, Id, VideoFormat::RGBA), + spa::pod::property!( + FormatProperties::VideoSize, + Rectangle, + spa::utils::Rectangle { + width: size.width, + height: size.height, + } + ), + spa::pod::property!( + FormatProperties::VideoFramerate, + Fraction, + spa::utils::Fraction { + num: framerate, + denom: 1, + } + ), + ); + serialize(spa::pod::Value::Object(object)) +} + +/// Buffers sized for one `size` RGBA picture in shared memory. +fn buffers(size: Size, stride: u32) -> Vec { + let mem_ptr = 1 << spa::buffer::DataType::MemPtr.as_raw(); + let mem_fd = 1 << spa::buffer::DataType::MemFd.as_raw(); + let object = spa::pod::Object { + type_: spa::utils::SpaTypes::ObjectParamBuffers.as_raw(), + id: spa::param::ParamType::Buffers.as_raw(), + properties: vec![ + spa::pod::Property::new( + spa::sys::SPA_PARAM_BUFFERS_buffers, + spa::pod::Value::Choice(spa::pod::ChoiceValue::Int(spa::utils::Choice( + spa::utils::ChoiceFlags::empty(), + spa::utils::ChoiceEnum::Range { + default: 4, + min: 2, + max: 8, + }, + ))), + ), + spa::pod::Property::new(spa::sys::SPA_PARAM_BUFFERS_blocks, spa::pod::Value::Int(1)), + spa::pod::Property::new( + spa::sys::SPA_PARAM_BUFFERS_size, + spa::pod::Value::Int((stride * size.height) as i32), + ), + spa::pod::Property::new(spa::sys::SPA_PARAM_BUFFERS_stride, spa::pod::Value::Int(stride as i32)), + spa::pod::Property::new( + spa::sys::SPA_PARAM_BUFFERS_dataType, + spa::pod::Value::Choice(spa::pod::ChoiceValue::Int(spa::utils::Choice( + spa::utils::ChoiceFlags::empty(), + spa::utils::ChoiceEnum::Flags { + default: mem_fd | mem_ptr, + flags: Vec::new(), + }, + ))), + ), + ], + }; + serialize(spa::pod::Value::Object(object)) +} + +fn serialize(value: spa::pod::Value) -> Vec { + spa::pod::serialize::PodSerializer::serialize(std::io::Cursor::new(Vec::new()), &value) + .expect("serializing a pod into memory cannot fail") + .0 + .into_inner() +} diff --git a/rs/moq-studio/src/vcam/v4l2.rs b/rs/moq-studio/src/vcam/v4l2.rs new file mode 100644 index 00000000..6d80dc21 --- /dev/null +++ b/rs/moq-studio/src/vcam/v4l2.rs @@ -0,0 +1,150 @@ +//! The fallback camera: an existing v4l2loopback device, written to as I420. +//! +//! v4l2loopback is a kernel module, so the studio never creates a device; it +//! takes the first one it finds that accepts the program's size. + +use std::io::Write; +use std::path::PathBuf; + +use anyhow::Context; +use moq_v4l::video::Output; +use moq_video::{I420, Size}; + +use super::Sink; + +/// The driver name v4l2loopback devices report. +const DRIVER: &str = "v4l2 loopback"; + +/// I420, which every V4L2 camera reader takes. +const FOURCC: &[u8; 4] = b"YU12"; + +pub struct Loopback { + device: moq_v4l::Device, + path: PathBuf, + size: Size, +} + +impl Loopback { + /// Open the first v4l2loopback device that takes `size` I420 frames. + pub fn open(size: Size, framerate: u32) -> anyhow::Result { + let mut refused = Vec::new(); + for path in devices()? { + let Ok(device) = moq_v4l::Device::with_path(&path) else { + continue; + }; + match device.query_caps() { + Ok(caps) if caps.driver == DRIVER => {} + _ => continue, + } + match configure(&device, size, framerate) { + Ok(()) => return Ok(Self { device, path, size }), + Err(err) => refused.push(format!("{}: {err:#}", path.display())), + } + } + match refused.is_empty() { + true => anyhow::bail!("no device; load the v4l2loopback module to create one"), + false => anyhow::bail!("no device takes the program ({})", refused.join("; ")), + } + } + + pub fn name(&self) -> String { + self.path.display().to_string() + } +} + +impl Sink for Loopback { + fn write(&mut self, rgba: &[u8]) -> anyhow::Result<()> { + let i420 = moq_video::Surface::rgba(rgba, self.size)?.into_i420()?; + // v4l2loopback takes one frame per write, so a short one would split it. + let written = self + .device + .write(i420.data()) + .with_context(|| format!("writing to {}", self.path.display()))?; + anyhow::ensure!( + written == i420.data().len(), + "{} took {written} of a frame's {} bytes", + self.path.display(), + i420.data().len() + ); + Ok(()) + } +} + +/// Every `/dev/videoN`, in number order. +fn devices() -> anyhow::Result> { + let mut devices: Vec<(u32, PathBuf)> = std::fs::read_dir("/dev") + .context("listing /dev")? + .filter_map(|entry| { + let path = entry.ok()?.path(); + let number = path.file_name()?.to_str()?.strip_prefix("video")?.parse().ok()?; + Some((number, path)) + }) + .collect(); + devices.sort(); + Ok(devices.into_iter().map(|(_, path)| path).collect()) +} + +/// Set `device` to take `size` I420 frames at `framerate`, refusing any other +/// layout it answers with. +fn configure(device: &moq_v4l::Device, size: Size, framerate: u32) -> anyhow::Result<()> { + let fourcc = moq_v4l::FourCC::new(FOURCC); + let mut want = moq_v4l::Format::new(size.width, size.height, fourcc); + want.stride = size.width; + let len = I420::len(size)?; + want.size = len as u32; + let set = device.set_format(&want).context("setting the format")?; + anyhow::ensure!( + set.width == size.width && set.height == size.height && set.fourcc == fourcc, + "it took {}x{} {} rather than {size} {}", + set.width, + set.height, + set.fourcc, + fourcc + ); + anyhow::ensure!( + set.stride == size.width && set.size as usize == len, + "it pads I420 rows to {} bytes", + set.stride + ); + device + .set_params(&moq_v4l::video::output::Parameters::with_fps(framerate)) + .context("setting the frame rate")?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use std::io::Read; + + use super::*; + + /// A V4L2 app reading the loopback device gets the picture written to it. + #[test] + #[ignore = "needs a v4l2loopback device; run by just v4l2loopback"] + fn a_v4l2_app_reads_the_program() { + let size = Size::new(320, 180); + let mut loopback = Loopback::open(size, 30).expect("a v4l2loopback device"); + let red = [255, 0, 0, 255].repeat((size.width * size.height) as usize); + // A reader opens the device once a writer has set it up, as an app does. + loopback.write(&red).expect("write"); + let mut reader = moq_v4l::Device::with_path(&loopback.path).expect("the reader opens"); + let mut frame = vec![0; I420::len(size).unwrap()]; + loopback.write(&red).expect("write"); + let read = reader.read(&mut frame).expect("read"); + assert_eq!(read, frame.len()); + + let i420 = I420::new(size, frame).expect("an I420 frame"); + let center = |plane: &[u8], width: u32, height: u32| plane[(height / 2 * width + width / 2) as usize]; + let (y, u, v) = ( + center(i420.y(), size.width, size.height), + center(i420.u(), size.width / 2, size.height / 2), + center(i420.v(), size.width / 2, size.height / 2), + ); + // Red, in any matrix: dim luma, blue-difference below gray, and + // red-difference well above it. + assert!( + y < 100 && u < 128 && v > 200, + "the reader got y={y} u={u} v={v}, not red" + ); + } +} diff --git a/sh/pipewire-session.sh b/sh/pipewire-session.sh new file mode 100755 index 00000000..c235a82e --- /dev/null +++ b/sh/pipewire-session.sh @@ -0,0 +1,46 @@ +#!/usr/bin/env bash +# Run a command inside a private PipeWire session: a PipeWire daemon and the +# WirePlumber session manager that links streams, on a runtime directory of +# their own, so a test reaches neither the desktop's session nor needs one. +set -euo pipefail + +# The dev shell carries PipeWire for linking; the session manager comes from the +# same nixpkgs. +wireplumber=$(nix build --inputs-from . --no-link --print-out-paths 'nixpkgs#wireplumber^out') + +# Only the packages' own configuration, never the host's, which can name +# components these builds do not ship. +pipewire_share=$(dirname "$(command -v pipewire)")/../share +export PIPEWIRE_CONFIG_DIR=$pipewire_share/pipewire +export WIREPLUMBER_CONFIG_DIR=$wireplumber/share/wireplumber +export WIREPLUMBER_DATA_DIR=$wireplumber/share/wireplumber + +runtime=$(mktemp -d) +export XDG_RUNTIME_DIR=$runtime +unset PIPEWIRE_REMOTE DBUS_SESSION_BUS_ADDRESS +pids=() +cleanup() { + if ((${#pids[@]})); then + kill "${pids[@]}" 2>/dev/null || true + wait "${pids[@]}" 2>/dev/null || true + fi + rm -rf "$runtime" +} +trap cleanup EXIT + +pipewire & +pids+=("$!") +# The socket appears once the daemon listens; a client started before it fails. +until [[ -S $runtime/pipewire-0 ]]; do + kill -0 "${pids[0]}" + sleep 0.1 +done +"$wireplumber/bin/wireplumber" & +pids+=("$!") +# WirePlumber connects as the session manager once it has loaded its policy. +until pw-cli ls Client 2>/dev/null | grep -q 'application.name = "WirePlumber"'; do + kill -0 "${pids[1]}" + sleep 0.1 +done + +"$@" From fc0d1c12ca75c17751e5ab0ec58e744db641e2a8 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 2 Nov 2026 15:13:22 -0800 Subject: [PATCH 3/5] fix(studio): report the PipeWire camera added only once PipeWire adds it Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/vcam/pipewire.rs | 32 +++++++++++++++++++++++------- 1 file changed, 25 insertions(+), 7 deletions(-) diff --git a/rs/moq-studio/src/vcam/pipewire.rs b/rs/moq-studio/src/vcam/pipewire.rs index c7a1d28a..24f6e51f 100644 --- a/rs/moq-studio/src/vcam/pipewire.rs +++ b/rs/moq-studio/src/vcam/pipewire.rs @@ -9,6 +9,8 @@ //! own main loop. The `vcam` thread leaves the newest picture in a [`Slot`] //! and wakes the loop; the loop copies it into a PipeWire buffer. +use std::cell::RefCell; +use std::rc::Rc; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex, mpsc}; @@ -63,7 +65,10 @@ impl Node { std::thread::Builder::new() .name("vcam-pipewire".into()) .spawn(move || { - if let Err(err) = run(size, framerate, slot, shared, messages, &ready) { + let ready = Rc::new(RefCell::new(Some(ready))); + if let Err(err) = run(size, framerate, slot, shared, messages, ready.clone()) + && let Some(ready) = ready.take() + { let _ = ready.send(Err(format!("{err:#}"))); } }) @@ -114,15 +119,19 @@ impl Drop for Node { } } +/// Whoever waits for the node to be added, until it is or PipeWire refuses it. +type Ready = Rc>>>>; + /// Add the node and run it until it is dropped or PipeWire ends it, saying on -/// `ready` once it is added. Fails only before that. +/// `ready` whether PipeWire added it. A failure before then goes to `ready` +/// rather than `shared`, so the caller can still fall back. fn run( size: Size, framerate: u32, slot: Arc, shared: Arc, messages: pw::channel::Receiver, - ready: &mpsc::Sender>, + ready: Ready, ) -> anyhow::Result<()> { pw::init(); let mainloop = pw::main_loop::MainLoopRc::new(None).context("PipeWire main loop")?; @@ -158,10 +167,22 @@ fn run( let ended = match state { pw::stream::StreamState::Error(err) => Some(format!("PipeWire: {err}")), pw::stream::StreamState::Unconnected => Some("the PipeWire session ended".to_string()), + // Paused first once PipeWire has added the node. + pw::stream::StreamState::Paused => { + if let Some(ready) = ready.take() { + let _ = ready.send(Ok(())); + } + None + } _ => None, }; if let Some(ended) = ended { - shared.fail(ended); + match ready.take() { + Some(ready) => { + let _ = ready.send(Err(ended)); + } + None => shared.fail(ended), + } if let Some(mainloop) = mainloop.upgrade() { mainloop.quit(); } @@ -260,9 +281,6 @@ fn run( } }); - if ready.send(Ok(())).is_err() { - return Ok(()); - } mainloop.run(); Ok(()) } From 1c3ffc55ab73dc1d0d187d57ed8f639332e1f53e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 19:11:47 -0800 Subject: [PATCH 4/5] fix(studio): build the merged virtual camera One Linux dependency table, the camera's Output panel row after the canvases, and the lockfile entries for moq-v4l and pipewire. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 2 ++ rs/moq-studio/Cargo.toml | 14 ++++++-------- rs/moq-studio/src/ui.rs | 5 +++-- 3 files changed, 11 insertions(+), 10 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 9047d301..e87b2ec4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4370,8 +4370,10 @@ dependencies = [ "moq-scene", "moq-stats", "moq-tokio", + "moq-v4l", "moq-video", "mp4-atom", + "pipewire", "pollster 1.0.1", "qrcodegen", "raw-window-handle", diff --git a/rs/moq-studio/Cargo.toml b/rs/moq-studio/Cargo.toml index a90d874c..6736a4f7 100644 --- a/rs/moq-studio/Cargo.toml +++ b/rs/moq-studio/Cargo.toml @@ -92,6 +92,12 @@ winit = { version = "0.30", default-features = false } # PipeWire capture uses, so the two share one D-Bus stack. ashpd = { version = "0.13", default-features = false, features = ["global_shortcuts", "tokio"] } futures-util = "0.3" +# The virtual camera's v4l2loopback fallback. Pure Rust with its kernel bindings +# checked in, and already in the graph through moq-video's capture. +moq-v4l = { workspace = true } +# The virtual camera's PipeWire node, behind `pipewire`. moq-video links the +# same release; `v0_3_34` adds the trigger a driving stream paces frames with. +pipewire = { version = "0.10", optional = true, features = ["v0_3_34"] } # Whether the window is on Wayland, where only the portal hears keys globally. raw-window-handle = "0.6" # Which keysyms the X11 layout puts on each key, so a hotkey grabs the key @@ -103,14 +109,6 @@ x11rb = "0.13" # registers the key pressed. windows-sys = { version = "0.61", features = ["Win32_UI_Input_KeyboardAndMouse"] } -[target.'cfg(target_os = "linux")'.dependencies] -# The virtual camera's v4l2loopback fallback. Pure Rust with its kernel bindings -# checked in, and already in the graph through moq-video's capture. -moq-v4l = { workspace = true } -# The virtual camera's PipeWire node, behind `pipewire`. moq-video links the -# same release; `v0_3_34` adds the trigger a driving stream paces frames with. -pipewire = { version = "0.10", optional = true, features = ["v0_3_34"] } - [dev-dependencies] # A clock that moves wall time too, which renewed tokens' expiries read. kio = { workspace = true, features = ["manual"] } diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index d4559a78..4a7248b2 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -2652,8 +2652,6 @@ impl Panels { if let Some(error) = stats.recording_audio_error() { ui.colored_label(ui.visuals().warn_fg_color, format!("recording audio off: {error}")); } - ui.separator(); - self.camera(ui); for (name, video) in stats.canvases() { ui.separator(); @@ -2673,6 +2671,9 @@ impl Panels { ui.colored_label(ui.visuals().error_fg_color, error); } } + + ui.separator(); + self.camera(ui); } /// The main canvas's virtual camera: a toggle, and how it is doing while on. From 63720dcc0f6c3270367d83a4b12929be3d78c740 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 19:13:09 -0800 Subject: [PATCH 5/5] style(studio): format the camera's picture row Co-Authored-By: Claude Opus 5.5 --- rs/moq-studio/src/ui.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index 4a7248b2..84cfc4b0 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -2696,7 +2696,10 @@ impl Panels { ui.end_row(); ui.label("Picture"); let canvas = camera.canvas; - ui.label(format!("{}x{} at {} fps", canvas.width, canvas.height, canvas.framerate)); + ui.label(format!( + "{}x{} at {} fps", + canvas.width, canvas.height, canvas.framerate + )); ui.end_row(); ui.label("Apps"); ui.label(if camera.reading { "reading" } else { "none; not copying" });