diff --git a/README.md b/README.md index d5cb3f4..f69271f 100644 --- a/README.md +++ b/README.md @@ -94,7 +94,7 @@ Every panel is a tab you can drag, split, or stack. | Scenes | Pick the scene in preview, and add, rename (double-click), reorder, or remove one. A scene on program, or still fading out, cannot be removed, even by an undo, until program cuts away from it. | | History | See every edit, and undo or redo back to any of them. | | Items | Place, show, hide, and order what the preview scene or the top layer draws, back to front. Select an item to filter or copy it. | -| Sources | Add a source by hand, see whether each and its own track are live and where it is shown, clone one, and remove one. Select a source to filter it. A camera or broadcast without a new picture, or a microphone or system audio without new sound, for a second reads as stalled. A browser source shows whether its page is copied on the GPU or, as a fallback, uploaded from memory (CPU), in place of a track. | +| Sources | Add a source by hand, see whether each and its own track are live and where it is shown, clone one, and remove one. Select a source to filter it. A camera or broadcast without a new picture, or a microphone or system audio without new sound, for a second reads as stalled. A browser source shows whether its page is copied on the GPU or, as a fallback, uploaded from memory (CPU), in place of a track, and **Reload** restarts its page. | | Devices | List this machine's cameras, displays, windows, and audio inputs, and add one as a source. | | Guests | Invite a guest, show their join link as a QR code, see who joined, and take a guest out of every scene. | | Filters | Crop the selected source or item, add, order, and set its filters, set an item's blend mode, and copy or paste filters. | @@ -142,12 +142,17 @@ or not its scene is live; mute a source to leave it out. A browser source renders a web page, such as an overlay, alert box, or chat widget, at the canvas frame rate (CEF caps it at 60) over a transparent -background, and its sound joins the program audio. Each page runs in a -process of its own, so a page or browser crash fails that source, with the -reason, while the stream carries on; edit or re-add the source to restart it. -Every start is a fresh profile: no cookies or logins persist. A page opens -no other windows (`window.open` and new-tab links do nothing), and its -dropdown lists, such as a ``'s, are not drawn yet. ## Guests diff --git a/quest/m0/README.md b/quest/m0/README.md index 8f64e32..d5e84d3 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -104,7 +104,6 @@ repository. - [Audio mixes](/quest/m0/mixes.md) - per-output mixes, meters, monitoring, and sync offsets - [Audio filters](/quest/m0/audio-filters.md) - per-source noise suppression, gate, compressor, limiter, and plugins, configured from the studio - [Browser sources on macOS and Windows](/quest/m0/browser-source.md) - the out-of-process CEF pages Linux has, with shared GPU textures on the other desktops -- [Browser sources keep a profile](/quest/m0/browser-profile.md) - one persistent profile, so logins survive restarts, and a reload action - [Browser source popups](/quest/m0/browser-popups.md) - dropdowns and other in-page popups draw and take input - [Published independently](/quest/m0/publish-independently.md) - a source opts out of the program and publishes its own track for services that place it themselves - [Web studio](/quest/m0/web.md) - the full studio in wasm on WebGPU and WebCodecs, served from this repository diff --git a/quest/m0/browser-profile.md b/quest/m0/browser-profile.md deleted file mode 100644 index 16c2ea4..0000000 --- a/quest/m0/browser-profile.md +++ /dev/null @@ -1,24 +0,0 @@ -# [S] Browser sources keep a profile - -## Goal - -Browser sources share one persistent browser profile, so cookies and logins -(an alert service, a chat widget) survive a restart of the source or the -studio, as in OBS. -Each browser source also has a reload action that restarts its page without -editing the source. - -## Plan - -Decided 2026-11-02 (maintainer): one shared profile for all browser sources, -not one per source. Each page runs in its own CEF helper process today with a -fresh temporary profile, and two CEF instances cannot share a profile -directory, so sharing it means the helpers share one CEF instance or hand the -profile between them; weigh that against the crash isolation per page that the -helpers give. Keep the profile beside the studio's other per-user state, not -beside the scene file, since the scene file is meant to be shared. Raised on -[#8](https://github.com/moq-dev/studio/pull/8). - -## Related - -- [Browser sources](/quest/m0/browser-source.md) - the macOS and Windows half diff --git a/quest/m0/browser-source.md b/quest/m0/browser-source.md index 197c2b0..7d6578d 100644 --- a/quest/m0/browser-source.md +++ b/quest/m0/browser-source.md @@ -5,7 +5,8 @@ Browser sources run on macOS and Windows as they do on Linux: a URL (an overlay, alert box, or chat widget) renders into the compositor at the canvas frame rate, the creator can interact with it, its audio joins the bus, and a -crash in the browser process drops that source, never the stream. +crash drops browser sources, never the stream: a page's renderer only its own +source, the shared browser process every browser source. ## Plan @@ -20,11 +21,13 @@ Decided 2026-10-22: use CEF through `cef-rs`, desktop only. path is the fallback and shows in stats. - **Process:** CEF runs out of process. -Linux landed first (`moq_scene::browser`): one helper process per page, -which is the studio's executable started again, talking over a Unix socket -on its stdin that carries the DMA-BUF descriptors, and `cef-rs`'s -`osr_texture_import` copies each texture on the studio's device. What the -other platforms need on top: +Linux landed first (`moq_scene::browser`): one helper process every page +shares, on one persistent profile, which is the studio's executable started +again, talking over a Unix socket on its stdin that carries the DMA-BUF +descriptors, and `cef-rs`'s `osr_texture_import` copies each texture on the +studio's device. The profile lives in the XDG data directory +(`engine::browser_profile`); each platform keeps it in its own per-user data +directory instead. What the other platforms need on top: - **Windows:** a channel without Unix sockets, such as the helper's stdio pipes. The studio duplicates CEF's NT handle out of the helper with `DuplicateHandle` before `cef-rs` imports it. diff --git a/rs/moq-scene/src/browser/host.rs b/rs/moq-scene/src/browser/host.rs index ba15433..4d85c8e 100644 --- a/rs/moq-scene/src/browser/host.rs +++ b/rs/moq-scene/src/browser/host.rs @@ -1,16 +1,19 @@ -//! The helper process: CEF's browser process, rendering one page offscreen -//! and handing each picture and sound to the studio over the socket on its -//! stdin. +//! The helper process: CEF's browser process, rendering every page offscreen +//! in one profile and handing each picture and sound to the studio over the +//! socket on its stdin. -use std::cell::RefCell; +use std::cell::{Cell, RefCell}; +use std::collections::HashMap; use std::os::fd::BorrowedFd; use std::os::unix::net::UnixStream; use std::process::ExitCode; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex, mpsc}; use cef::*; -use super::protocol::{Channel, Command, Event, Format, Open, Plane, Texture}; +use super::outbox::Commands; +use super::protocol::{Channel, Command, Event, Format, Id, Open, Plane, Start, Texture}; use super::{Button, Input, Modifiers}; /// Run one of CEF's subprocesses (renderer, GPU, utility, zygote). @@ -32,13 +35,15 @@ pub(super) fn run() -> ExitCode { } } -/// What every handler shares. +/// What every page's handlers share. struct Shared { - open: Open, /// Sends from the UI thread (pictures) and CEF's audio thread (sound). writer: Mutex, /// Each [`Command::Painted`], which lets a painted picture go back to CEF. + /// Pictures are painted on the UI thread, one at a time, so one answer is + /// never another page's. painted: Mutex>, + accelerated: bool, } impl Shared { @@ -53,15 +58,27 @@ impl Shared { let _ = self.painted.lock().unwrap().recv(); } } +} + +/// One page's handlers' view of it. +struct Paged { + shared: Arc, + id: Id, + open: Open, +} +impl Paged { fn fail(&self, reason: String) { - self.send(&Event::Failed(reason), &[]); + self.shared.send(&Event::Failed(self.id, reason), &[]); } } thread_local! { - /// The page, on CEF's UI thread, where input is delivered. - static BROWSER: RefCell> = const { RefCell::new(None) }; + /// The open pages, on CEF's UI thread, where input is delivered. + static PAGES: RefCell> = RefCell::new(HashMap::new()); + /// Whether the studio is gone, so the helper quits once its last page + /// closes. + static QUITTING: Cell = const { Cell::new(false) }; } fn serve() -> Result<(), String> { @@ -73,31 +90,39 @@ fn serve() -> Result<(), String> { let reader = Channel::new(UnixStream::from(socket)); let writer = reader.try_clone().map_err(|err| format!("socket: {err}"))?; - let open = match reader.recv::() { - Ok(Some((Command::Open(open), _))) => open, - Ok(Some((command, _))) => return Err(format!("expected Open, got {command:?}")), + let start = match reader.recv::() { + Ok(Some((Command::Start(start), _))) => start, + Ok(Some((command, _))) => return Err(format!("expected Start, got {command:?}")), Ok(None) => return Ok(()), Err(err) => return Err(format!("socket: {err}")), }; let (painted_tx, painted) = mpsc::channel(); let shared = Arc::new(Shared { - open, writer: Mutex::new(writer), painted: Mutex::new(painted), + accelerated: start.accelerated, }); // The studio shows why, rather than only that the helper exited. - page(&shared, reader, painted_tx).inspect_err(|reason| shared.fail(reason.clone())) + browse(&shared, start, reader, painted_tx).inspect_err(|reason| _ = shared.send(&Event::Fatal(reason.clone()), &[])) } -/// Open the page `shared` describes and run CEF until it closes. -fn page(shared: &Arc, reader: Channel, painted_tx: mpsc::Sender<()>) -> Result<(), String> { +/// Run CEF on the profile `start` names, opening and closing pages as the +/// studio asks, until it closes the socket and the last page closes. +fn browse(shared: &Arc, start: Start, reader: Channel, painted_tx: mpsc::Sender<()>) -> Result<(), String> { let _ = api_hash(sys::CEF_API_VERSION_LAST, 0); let args = args::Args::new(); - let mut app = HelperApp::new(shared.open.render_node.clone()); + let mut app = HelperApp::new(HelperProcess::new(), start.render_node); + let profile = CefString::from(start.profile.as_str()); let settings = Settings { windowless_rendering_enabled: 1, - root_cache_path: shared.open.profile.as_str().into(), + // The global profile, kept on disk: without a cache path CEF keeps + // everything in memory. + root_cache_path: profile.clone(), + cache_path: profile, + // A login that sets a cookie without an expiry survives a restart, as + // it would in a browser left open. + persist_session_cookies: 1, log_severity: LogSeverity::WARNING, // Transparent, so an overlay shows what is under it. background_color: 0, @@ -110,163 +135,232 @@ fn page(shared: &Arc, reader: Channel, painted_tx: mpsc::Sender<()>) -> std::ptr::null_mut(), ) != 1 { - return Err("CEF failed to initialize".into()); + let notified = sys::cef_resultcode_t::CEF_RESULT_CODE_NORMAL_EXIT_PROCESS_NOTIFIED as i32; + return Err(match get_exit_code() == notified { + true => format!( + "another studio is using the browser profile {}; quit it, then reload the source", + start.profile + ), + false => "CEF failed to initialize".into(), + }); } + let commands = { + let shared = shared.clone(); + std::thread::Builder::new() + .name("browser-commands".into()) + .spawn(move || commands(shared, reader, painted_tx)) + }; + if let Err(err) = commands { + shutdown(); + return Err(format!("cannot start the helper's command thread: {err}")); + } + + run_message_loop(); + PAGES.with_borrow_mut(HashMap::clear); + // Saves the profile, such as cookies not yet written. + shutdown(); + Ok(()) +} + +/// Read the studio's commands until it closes the socket, then close every +/// page, which ends the message loop. +/// +/// What the UI thread has yet to apply waits coalesced, as in the studio, with +/// one task at a time posted to apply it, so a stalled UI thread holds what +/// the pages open need rather than a task per command. Input past the +/// backlog drops, as for a hung page. +fn commands(shared: Arc, reader: Channel, painted: mpsc::Sender<()>) { + let pending = Commands::default(); + let posted = Arc::new(AtomicBool::new(false)); + let schedule = || { + if !posted.swap(true, Ordering::AcqRel) { + let (shared, pending, posted) = (shared.clone(), pending.clone(), posted.clone()); + post(move || { + // Cleared first, so anything queued from here on posts again. + posted.store(false, Ordering::Release); + apply(&shared, pending.take()); + }); + } + }; + loop { + match reader.recv::() { + Ok(Some((Command::Painted, _))) => { + let _ = painted.send(()); + continue; + } + Ok(Some((Command::Open(id, open), _))) => pending.open(id, open), + Ok(Some((Command::Close(id), _))) => pending.close(id), + Ok(Some((Command::Input(id, input), _))) => _ = pending.input(id, input), + Ok(Some((Command::FrameRate(id, rate), _))) => _ = pending.rate(id, rate), + Ok(Some((Command::Start(_), _))) | Ok(None) | Err(_) => break, + } + schedule(); + } + // Unblocks a paint waiting on the studio. + drop(painted); + post(move || { + apply(&shared, pending.take()); + QUITTING.set(true); + let ids: Vec = PAGES.with_borrow(|pages| pages.keys().copied().collect()); + if ids.is_empty() { + quit_message_loop(); + } + for id in ids { + close(id); + } + }); +} + +/// Apply the studio's commands. Runs on the UI thread. +fn apply(shared: &Arc, batch: Vec) { + for command in batch { + match command { + Command::Open(id, open) => self::open(shared.clone(), id, open), + Command::Close(id) => close(id), + Command::Input(id, input) => deliver(id, input), + Command::FrameRate(id, rate) => retime(id, rate), + Command::Start(_) | Command::Painted => unreachable!("never pending"), + } + } +} + +/// Open page `id`. Runs on the UI thread. +fn open(shared: Arc, id: Id, open: Open) { let window = WindowInfo { windowless_rendering_enabled: 1, - shared_texture_enabled: shared.open.accelerated.into(), + shared_texture_enabled: shared.accelerated.into(), ..Default::default() }; let browser_settings = BrowserSettings { - windowless_frame_rate: shared.open.framerate as _, + windowless_frame_rate: open.framerate as _, background_color: 0, ..Default::default() }; + let url = CefString::from(open.url.as_str()); + let paged = Arc::new(Paged { shared, id, open }); let mut client = HelperClient::new( - HelperRender::new(shared.clone()), - HelperAudio::new(shared.clone()), - HelperLoad::new(shared.clone()), - HelperRequest::new(shared.clone()), - HelperLife::new(), + HelperRender::new(paged.clone()), + HelperAudio::new(paged.clone()), + HelperLoad::new(paged.clone()), + HelperRequest::new(paged.clone()), + HelperLife::new(paged.clone()), ); - let url = CefString::from(shared.open.url.as_str()); - let browser = browser_host_create_browser_sync( + match browser_host_create_browser_sync( Some(&window), Some(&mut client), Some(&url), Some(&browser_settings), None, None, - ) - .ok_or("CEF failed to create the page")?; - BROWSER.with_borrow_mut(|slot| *slot = Some(browser)); - - std::thread::Builder::new() - .name("browser-commands".into()) - .spawn(move || commands(reader, painted_tx)) - .map_err(|err| format!("cannot start the helper's command thread: {err}"))?; + ) { + Some(browser) => _ = PAGES.with_borrow_mut(|pages| pages.insert(id, browser)), + None => paged.fail("CEF failed to create the page".into()), + } +} - run_message_loop(); - BROWSER.with_borrow_mut(|slot| *slot = None); - shutdown(); - Ok(()) +/// Close page `id`, which leaves [`PAGES`] once CEF has closed it. Runs on +/// the UI thread. +fn close(id: Id) { + host(id, |host| host.close_browser(1)); } -/// Read the studio's commands until it closes the socket, then close the page, -/// which ends the message loop. -fn commands(reader: Channel, painted: mpsc::Sender<()>) { - loop { - match reader.recv::() { - Ok(Some((Command::Painted, _))) => { - let _ = painted.send(()); - } - Ok(Some((Command::Input(input), _))) => post(move || deliver(input)), - Ok(Some((Command::FrameRate(rate), _))) => post(move || retime(rate)), - Ok(Some((Command::Open(_), _))) | Ok(None) | Err(_) => break, - } +/// Run `f` on page `id`'s host, if the page is open. Runs on the UI thread. +fn host(id: Id, f: impl FnOnce(BrowserHost)) { + if let Some(host) = PAGES.with_borrow(|pages| pages.get(&id).and_then(|browser| browser.host())) { + f(host); } - // Unblocks a paint waiting on the studio. - drop(painted); - post(|| { - BROWSER.with_borrow(|browser| { - if let Some(host) = browser.as_ref().and_then(|browser| browser.host()) { - host.close_browser(1); - } - }) - }); } -/// Paint the page `rate` times a second. Runs on the UI thread. -fn retime(rate: u32) { - BROWSER.with_borrow(|browser| { - if let Some(host) = browser.as_ref().and_then(|browser| browser.host()) { - host.set_windowless_frame_rate(rate.try_into().unwrap_or(i32::MAX)); - } - }) +/// Paint page `id` `rate` times a second. Runs on the UI thread. +fn retime(id: Id, rate: u32) { + host(id, |host| { + host.set_windowless_frame_rate(rate.try_into().unwrap_or(i32::MAX)) + }); } -/// Hand one input event to the page. Runs on the UI thread. -fn deliver(input: Input) { - BROWSER.with_borrow(|browser| { - let Some(host) = browser.as_ref().and_then(|browser| browser.host()) else { - return; - }; - let mouse = |x: i32, y: i32, modifiers: Modifiers| MouseEvent { +/// Hand one input event to page `id`. Runs on the UI thread. +fn deliver(id: Id, input: Input) { + let Some(browser) = PAGES.with_borrow(|pages| pages.get(&id).cloned()) else { + return; + }; + let Some(host) = browser.host() else { + return; + }; + let mouse = |x: i32, y: i32, modifiers: Modifiers| MouseEvent { + x, + y, + modifiers: flags(modifiers), + }; + match input { + Input::Move { x, y, modifiers } => host.send_mouse_move_event(Some(&mouse(x, y, modifiers)), 0), + Input::Leave { modifiers } => host.send_mouse_move_event(Some(&mouse(0, 0, modifiers)), 1), + Input::Button { x, y, - modifiers: flags(modifiers), - }; - match input { - Input::Move { x, y, modifiers } => host.send_mouse_move_event(Some(&mouse(x, y, modifiers)), 0), - Input::Leave { modifiers } => host.send_mouse_move_event(Some(&mouse(0, 0, modifiers)), 1), - Input::Button { - x, - y, + button, + down, + clicks, + modifiers, + } => { + let button = match button { + Button::Left => MouseButtonType::LEFT, + Button::Middle => MouseButtonType::MIDDLE, + Button::Right => MouseButtonType::RIGHT, + }; + host.send_mouse_click_event( + Some(&mouse(x, y, modifiers)), button, - down, - clicks, - modifiers, - } => { - let button = match button { - Button::Left => MouseButtonType::LEFT, - Button::Middle => MouseButtonType::MIDDLE, - Button::Right => MouseButtonType::RIGHT, - }; - host.send_mouse_click_event( - Some(&mouse(x, y, modifiers)), - button, - (!down).into(), - clicks.min(3) as _, - ); - } - Input::Wheel { - x, - y, - dx, - dy, - modifiers, - } => host.send_mouse_wheel_event(Some(&mouse(x, y, modifiers)), dx, dy), - Input::Key { code, down, modifiers } => { + (!down).into(), + clicks.min(3) as _, + ); + } + Input::Wheel { + x, + y, + dx, + dy, + modifiers, + } => host.send_mouse_wheel_event(Some(&mouse(x, y, modifiers)), dx, dy), + Input::Key { code, down, modifiers } => { + let event = KeyEvent { + type_: if down { + KeyEventType::RAWKEYDOWN + } else { + KeyEventType::KEYUP + }, + modifiers: flags(modifiers), + windows_key_code: code, + ..Default::default() + }; + host.send_key_event(Some(&event)); + } + Input::Text(text) => { + for unit in text.encode_utf16() { let event = KeyEvent { - type_: if down { - KeyEventType::RAWKEYDOWN - } else { - KeyEventType::KEYUP - }, - modifiers: flags(modifiers), - windows_key_code: code, + type_: KeyEventType::CHAR, + windows_key_code: unit.into(), + character: unit, + unmodified_character: unit, ..Default::default() }; host.send_key_event(Some(&event)); } - Input::Text(text) => { - for unit in text.encode_utf16() { - let event = KeyEvent { - type_: KeyEventType::CHAR, - windows_key_code: unit.into(), - character: unit, - unmodified_character: unit, - ..Default::default() - }; - host.send_key_event(Some(&event)); - } - } - // The focused frame holds the selection, which may be in an - // iframe. - Input::Copy => { - if let Some(frame) = browser.as_ref().and_then(|browser| browser.focused_frame()) { - frame.copy(); - } + } + // The focused frame holds the selection, which may be in an + // iframe. + Input::Copy => { + if let Some(frame) = browser.focused_frame() { + frame.copy(); } - Input::Cut => { - if let Some(frame) = browser.as_ref().and_then(|browser| browser.focused_frame()) { - frame.cut(); - } + } + Input::Cut => { + if let Some(frame) = browser.focused_frame() { + frame.cut(); } - Input::Focus(focused) => host.set_focus(focused.into()), } - }); + Input::Focus(focused) => host.set_focus(focused.into()), + } } fn flags(modifiers: Modifiers) -> u32 { @@ -310,10 +404,15 @@ wrap_task! { wrap_app! { struct HelperApp { + process: BrowserProcessHandler, render_node: Option, } impl App { + fn browser_process_handler(&self) -> Option { + Some(self.process.clone()) + } + fn on_before_command_line_processing( &self, _process_type: Option<&CefString>, @@ -335,6 +434,22 @@ wrap_app! { } } +wrap_browser_process_handler! { + struct HelperProcess; + + impl BrowserProcessHandler { + // A second studio starting on this profile is refused, and told so + // by its own failure; this one opens no window for it. + fn on_already_running_app_relaunch( + &self, + _command_line: Option<&mut CommandLine>, + _current_directory: Option<&CefString>, + ) -> ::std::os::raw::c_int { + 1 + } + } +} + wrap_client! { struct HelperClient { render: RenderHandler, @@ -369,14 +484,14 @@ wrap_client! { wrap_render_handler! { struct HelperRender { - shared: Arc, + page: Arc, } impl RenderHandler { fn view_rect(&self, _browser: Option<&mut Browser>, rect: Option<&mut Rect>) { if let Some(rect) = rect { - rect.width = self.shared.open.width as _; - rect.height = self.shared.open.height as _; + rect.width = self.page.open.width as _; + rect.height = self.page.open.height as _; } } @@ -397,11 +512,12 @@ wrap_render_handler! { // this call. let data = unsafe { std::slice::from_raw_parts(buffer, len) }.to_vec(); let event = Event::Pixels { + page: self.page.id, width: width as u32, height: height as u32, data, }; - self.shared.paint(&event, &[]); + self.page.shared.paint(&event, &[]); } fn on_accelerated_paint( @@ -419,7 +535,7 @@ wrap_render_handler! { ColorType::RGBA_8888 => Format::Rgba, ColorType::BGRA_8888 => Format::Bgra, other => { - self.shared.fail(format!("CEF painted an unknown format {:?}", other.as_ref())); + self.page.fail(format!("CEF painted an unknown format {:?}", other.as_ref())); return; } }; @@ -442,14 +558,14 @@ wrap_render_handler! { // SAFETY: CEF owns these descriptors and keeps them open for this // call, which outlasts the send that duplicates them. let fds: Vec<_> = planes.iter().map(|plane| unsafe { BorrowedFd::borrow_raw(plane.fd) }).collect(); - self.shared.paint(&Event::Texture(texture), &fds); + self.page.shared.paint(&Event::Texture(self.page.id, texture), &fds); } } } wrap_audio_handler! { struct HelperAudio { - shared: Arc, + page: Arc, } impl AudioHandler { @@ -459,12 +575,12 @@ wrap_audio_handler! { params: Option<&mut AudioParameters>, ) -> ::std::os::raw::c_int { let Some(params) = params else { return 0 }; - params.channel_layout = match self.shared.open.channels { + params.channel_layout = match self.page.open.channels { 1 => ChannelLayout::LAYOUT_MONO, 2 => ChannelLayout::LAYOUT_STEREO, _ => return 0, }; - params.sample_rate = self.shared.open.sample_rate as _; + params.sample_rate = self.page.open.sample_rate as _; 1 } @@ -475,10 +591,10 @@ wrap_audio_handler! { channels: ::std::os::raw::c_int, ) { let rate = params.map(|params| params.sample_rate); - if channels as u32 != self.shared.open.channels || rate != Some(self.shared.open.sample_rate as _) { - self.shared.fail(format!( + if channels as u32 != self.page.open.channels || rate != Some(self.page.open.sample_rate as _) { + self.page.fail(format!( "the page's sound arrived as {channels} channels at {rate:?} Hz, not the bus's {} at {} Hz", - self.shared.open.channels, self.shared.open.sample_rate + self.page.open.channels, self.page.open.sample_rate )); } } @@ -493,7 +609,7 @@ wrap_audio_handler! { if data.is_null() || frames <= 0 { return; } - let channels = self.shared.open.channels as usize; + let channels = self.page.open.channels as usize; let frames = frames as usize; // SAFETY: CEF hands over one plane of `frames` samples per // channel it started the stream with, which is the bus's count. @@ -503,19 +619,19 @@ wrap_audio_handler! { let samples = (0..frames) .flat_map(|frame| planes.iter().map(move |plane| plane[frame])) .collect(); - self.shared.send(&Event::Audio(samples), &[]); + self.page.shared.send(&Event::Audio(self.page.id, samples), &[]); } fn on_audio_stream_error(&self, _browser: Option<&mut Browser>, message: Option<&CefString>) { let message = message.map(|message| message.to_string()).unwrap_or_default(); - self.shared.fail(format!("the page's sound failed: {message}")); + self.page.fail(format!("the page's sound failed: {message}")); } } } wrap_load_handler! { struct HelperLoad { - shared: Arc, + page: Arc, } impl LoadHandler { @@ -534,14 +650,14 @@ wrap_load_handler! { } let url = failed_url.map(|url| url.to_string()).unwrap_or_default(); let text = error_text.map(|text| text.to_string()).unwrap_or_default(); - self.shared.fail(format!("cannot load {url}: {text}")); + self.page.fail(format!("cannot load {url}: {text}")); } } } wrap_request_handler! { struct HelperRequest { - shared: Arc, + page: Arc, } impl RequestHandler { @@ -553,18 +669,20 @@ wrap_request_handler! { error_string: Option<&CefString>, ) { let reason = error_string.map(|reason| reason.to_string()).unwrap_or_default(); - self.shared.fail(format!("the page's renderer stopped ({error_code}): {reason}")); + self.page.fail(format!("the page's renderer stopped ({error_code}): {reason}")); } } } wrap_life_span_handler! { - struct HelperLife; + struct HelperLife { + page: Arc, + } impl LifeSpanHandler { // A source shows one page, and nothing would show a popup, so // `window.open` and `target="_blank"` open nothing. That also keeps - // the page the only browser, whose closing ends the helper. + // the studio's pages the only browsers, whose closing ends the helper. fn on_before_popup( &self, _browser: Option<&mut Browser>, @@ -585,7 +703,16 @@ wrap_life_span_handler! { } fn on_before_close(&self, _browser: Option<&mut Browser>) { - quit_message_loop(); + // A page can close itself with `window.close()`, which its source + // reports; one the studio closed has left it already. + self.page.fail("the page closed itself".into()); + let last = PAGES.with_borrow_mut(|pages| { + pages.remove(&self.page.id); + pages.is_empty() + }); + if last && QUITTING.get() { + quit_message_loop(); + } } } } diff --git a/rs/moq-scene/src/browser/import.rs b/rs/moq-scene/src/browser/import.rs index cf2442f..fe6b3d7 100644 --- a/rs/moq-scene/src/browser/import.rs +++ b/rs/moq-scene/src/browser/import.rs @@ -17,9 +17,15 @@ pub(super) fn supported(helper: &Helper) -> bool { } /// Copy the DMA-BUF `texture` describes, whose plane descriptors are `fds`, -/// into a texture of the studio's own, waiting until the copy is done. +/// into a texture of the studio's own on `device`, waiting until the copy is +/// done. #[cfg(feature = "cef")] -pub(super) fn texture(helper: &Helper, texture: &Texture, fds: Vec) -> Result { +pub(super) fn texture( + device: &wgpu::Device, + queue: &wgpu::Queue, + texture: &Texture, + fds: Vec, +) -> Result { use std::os::fd::AsRawFd; use cef::osr_texture_import::SharedTextureHandle; @@ -53,11 +59,16 @@ pub(super) fn texture(helper: &Helper, texture: &Texture, fds: Vec) -> // back, and `fds` stay open until this returns. The device is the // studio's, opened by `moq_video::render::request_device` with DMA-BUF // import, on the GPU CEF paints with. - unsafe { SharedTextureHandle::new(&info).import_texture(&helper.device, &helper.queue) } + unsafe { SharedTextureHandle::new(&info).import_texture(device, queue) } .map_err(|err| format!("cannot import the page's texture: {err}")) } #[cfg(not(feature = "cef"))] -pub(super) fn texture(_: &Helper, _: &Texture, _: Vec) -> Result { +pub(super) fn texture( + _: &wgpu::Device, + _: &wgpu::Queue, + _: &Texture, + _: Vec, +) -> Result { Err("this build cannot import shared textures (enable the `cef` feature)".into()) } diff --git a/rs/moq-scene/src/browser/mod.rs b/rs/moq-scene/src/browser/mod.rs index bf93fa1..b9b4cf4 100644 --- a/rs/moq-scene/src/browser/mod.rs +++ b/rs/moq-scene/src/browser/mod.rs @@ -2,13 +2,17 @@ //! rendered offscreen by CEF (the Chromium Embedded Framework) and drawn by the //! compositor. //! -//! Each page runs in a helper process of its own, so a crash there fails that -//! source and nothing else. The helper is the studio's own executable started -//! again, which calls [`child`] first thing in `main` to become one; CEF's own -//! subprocesses start the same way. The helper paints into a shared GPU texture -//! the studio copies on its device, or into memory the studio uploads when -//! either side cannot share textures ([`Paint`] says which). The page's sound -//! joins the program bus at the bus's rate. +//! Every page runs in one helper process, CEF's browser process, so they +//! share one persistent profile: cookies and logins survive a restart of the +//! source or the studio, and CEF lets only one process hold a profile. Each +//! page still renders in a process of its own, so a page that crashes fails +//! only its own source. A crash of the helper itself fails every page, and the +//! next page to start starts a new helper. The helper is the studio's own +//! executable started again, which calls [`child`] first thing in `main` to +//! become one; CEF's own subprocesses start the same way. The helper paints +//! into a shared GPU texture the studio copies on its device, or into memory +//! the studio uploads when either side cannot share textures ([`Paint`] says +//! which). A page's sound joins the program bus at the bus's rate. //! //! Linux only for now, and only with the `cef` feature: elsewhere a browser //! source reports that it is unsupported. @@ -21,6 +25,8 @@ use serde::{Deserialize, Serialize}; use crate::wgpu; +#[cfg(target_os = "linux")] +mod outbox; #[cfg(target_os = "linux")] mod protocol; #[cfg(target_os = "linux")] @@ -65,7 +71,8 @@ pub fn child() -> Option { } /// What browser sources need from the studio: the executable that becomes -/// their helper, and the device their pictures are copied to. +/// their helper, the profile they share, and the device their pictures are +/// copied to. Clones share one helper. #[derive(Clone)] #[cfg_attr( not(target_os = "linux"), @@ -73,19 +80,29 @@ pub fn child() -> Option { )] pub struct Helper { program: PathBuf, + profile: PathBuf, device: wgpu::Device, queue: wgpu::Queue, /// The device's DRM render node, which CEF must paint on for the studio to /// import what it paints. render_node: Option, + /// The helper, once a page has started it. + #[cfg(target_os = "linux")] + running: Arc>>>, } impl Helper { - /// Start helpers from `program`, which calls [`child`] first thing in - /// `main`, usually the running executable, and copy their pictures to - /// `device`, opened from `adapter`. + /// Start the helper from `program`, which calls [`child`] first thing in + /// `main`, usually the running executable, keep every page's cookies and + /// storage in the directory `profile`, created if missing, and copy their + /// pictures to `device`, opened from `adapter`. + /// + /// The profile holds the user's logins: keep it with their other state, + /// not beside a scene file meant to be shared. One studio at a time may + /// use it; a second's pages fail saying so. pub fn new( program: impl Into, + profile: impl Into, adapter: &wgpu::Adapter, device: &wgpu::Device, queue: &wgpu::Queue, @@ -98,9 +115,27 @@ impl Helper { .flatten(); Self { program: program.into(), + profile: profile.into(), device: device.clone(), queue: queue.clone(), render_node, + #[cfg(target_os = "linux")] + running: Default::default(), + } + } + + /// Stop the helper, failing every page, and wait for it to exit, so the + /// profile is saved and free for the next studio. The next page to start + /// starts a new helper. + pub fn stop(&self) { + #[cfg(target_os = "linux")] + { + // Held throughout, so no page starts another helper on the profile + // until this one has exited. + let mut running = self.running.lock().unwrap(); + if let Some(instance) = running.take() { + instance.stop(); + } } } } @@ -123,16 +158,15 @@ pub struct Page { #[derive(Default)] struct PageState { - /// The queue to the helper's socket, which [`run::writer`] drains. + /// The queue to the helper's socket, while the page is open. #[cfg(target_os = "linux")] - commands: Option>, - /// Whether input is being dropped because the queue is full. + commands: Option, + /// The page's number in the helper. #[cfg(target_os = "linux")] - backlogged: bool, - /// Releases that did not fit the queue, oldest first, at most one per - /// button, key, focus, and pointer, sent ahead of any newer input. + id: protocol::Id, + /// Whether input is being dropped because the backlog is full. #[cfg(target_os = "linux")] - owed: std::collections::VecDeque, + backlogged: bool, /// The canvas's frame rate, which the page paints at. #[cfg(target_os = "linux")] framerate: u32, @@ -144,91 +178,35 @@ struct PageState { #[cfg(target_os = "linux")] impl PageState { - /// Queue `input` behind any owed releases. Past the backlog, a release is - /// owed rather than dropped, so a helper that catches up never leaves the - /// page holding a button or key the studio let go of; other input drops. + /// Queue `input`, which past the backlog drops, a release aside. fn send(&mut self, input: Input) { - use std::sync::mpsc::TrySendError; - self.flush(); let Some(commands) = &self.commands else { return }; - let input = match self.owed.is_empty() { - true => match commands.try_send(protocol::Command::Input(input)) { - Ok(()) => { - self.backlogged = false; - return; + match commands.input(self.id, input) { + Ok(()) => self.backlogged = false, + Err(outbox::Refused::Full) => { + if !std::mem::replace(&mut self.backlogged, true) { + tracing::warn!("a browser source's helper is not reading input; dropping it"); } - Err(TrySendError::Full(protocol::Command::Input(input))) => input, - Err(TrySendError::Full(_)) => unreachable!("only input is sent here"), - Err(TrySendError::Disconnected(_)) => return self.disconnect(), - }, - // Newer input waits behind the owed releases. - false => input, - }; - if let Some(release) = Release::of(&input) { - self.owed.retain(|owed| Release::of(owed) != Some(release)); - self.owed.push_back(input); - } - if !std::mem::replace(&mut self.backlogged, true) { - tracing::warn!("a browser source's helper is not reading input; dropping it"); + } + Err(outbox::Refused::Gone) => self.disconnect(), } } - /// Queue a new frame rate, then the owed releases, oldest first, while the - /// queue has room. + /// Queue a new frame rate. fn flush(&mut self) { - use std::sync::mpsc::TrySendError; - let Some(commands) = &self.commands else { return }; if self.retime { - match commands.try_send(protocol::Command::FrameRate(self.framerate)) { - Ok(()) => self.retime = false, - Err(TrySendError::Full(_)) => return, - Err(TrySendError::Disconnected(_)) => return self.disconnect(), - } - } - while let Some(input) = self.owed.pop_front() { - match commands.try_send(protocol::Command::Input(input)) { - Ok(()) => {} - Err(TrySendError::Full(protocol::Command::Input(input))) => { - self.owed.push_front(input); - return; - } - Err(TrySendError::Full(_)) => unreachable!("only input is sent here"), - Err(TrySendError::Disconnected(_)) => return self.disconnect(), + if !commands.rate(self.id, self.framerate) { + return self.disconnect(); } + self.retime = false; } } - /// The helper is gone, which its source reports. + /// The page closed, or the helper is gone, which its source reports. fn disconnect(&mut self) { self.commands = None; - self.owed.clear(); - } -} - -/// What an input lets go of, which a later release of the same supersedes. -#[cfg(target_os = "linux")] -#[derive(Clone, Copy, PartialEq, Eq)] -enum Release { - Button(Button), - Key(i32), - Focus, - Pointer, -} - -#[cfg(target_os = "linux")] -impl Release { - fn of(input: &Input) -> Option { - match *input { - Input::Button { - button, down: false, .. - } => Some(Self::Button(button)), - Input::Key { code, down: false, .. } => Some(Self::Key(code)), - Input::Focus(false) => Some(Self::Focus), - Input::Leave { .. } => Some(Self::Pointer), - _ => None, - } } } @@ -391,66 +369,31 @@ fn upload( mod tests { use super::*; - fn button(down: bool) -> Input { - Input::Button { - x: 1, - y: 1, - button: Button::Left, - down, - clicks: 1, - modifiers: Modifiers::default(), - } - } - - /// Past the backlog, a release is owed rather than dropped, once per - /// button, and goes ahead of newer input as the queue makes room, so the - /// page never keeps holding what the studio let go of. - #[test] - fn a_release_past_the_backlog_is_owed() { - let (sender, queue) = std::sync::mpsc::sync_channel(4); - let page = Page::default(); - page.state.lock().unwrap().commands = Some(sender); - page.send(button(true)); - while !page.state.lock().unwrap().backlogged { - page.send(Input::Text("x".into())); - } - page.send(button(false)); - page.send(button(true)); - page.send(button(false)); - page.send(Input::Move { - x: 2, - y: 2, - modifiers: Modifiers::default(), - }); - assert_eq!(page.state.lock().unwrap().owed, [button(false)]); - - // The helper catches up, and each write makes room, as the writer's do. - let mut received = Vec::new(); - while let Ok(protocol::Command::Input(input)) = queue.try_recv() { - received.push(input); - page.state.lock().unwrap().flush(); - } - assert_eq!(received.first(), Some(&button(true))); - assert_eq!(received.last(), Some(&button(false))); - assert_eq!(received.iter().filter(|input| **input == button(false)).count(), 1); - assert!(!received.iter().any(|input| matches!(input, Input::Move { .. }))); - } - - /// A new canvas frame rate reaches the page once, even past the backlog. + /// A new canvas frame rate reaches the page once, the newest of those + /// waiting, even past the backlog. #[test] fn a_new_frame_rate_reaches_the_page() { - let (sender, queue) = std::sync::mpsc::sync_channel(1); + let commands = outbox::Commands::default(); let page = Page::default(); - page.state.lock().unwrap().commands = Some(sender); + page.state.lock().unwrap().commands = Some(commands.clone()); page.set_framerate(30); page.set_framerate(30); - assert!(matches!(queue.try_recv(), Ok(protocol::Command::FrameRate(30)))); - assert!(queue.try_recv().is_err(), "an unchanged rate is not sent again"); + assert!(matches!(commands.take()[..], [protocol::Command::FrameRate(_, 30)])); + assert!(commands.take().is_empty(), "an unchanged rate is not sent again"); - page.send(Input::Text("x".into())); + while !page.state.lock().unwrap().backlogged { + page.send(Input::Text("x".into())); + } page.set_framerate(60); - assert!(matches!(queue.try_recv(), Ok(protocol::Command::Input(_)))); - page.state.lock().unwrap().flush(); - assert!(matches!(queue.try_recv(), Ok(protocol::Command::FrameRate(60)))); + page.set_framerate(90); + let rates: Vec<_> = commands + .take() + .into_iter() + .filter_map(|command| match command { + protocol::Command::FrameRate(_, rate) => Some(rate), + _ => None, + }) + .collect(); + assert_eq!(rates, [90]); } } diff --git a/rs/moq-scene/src/browser/outbox.rs b/rs/moq-scene/src/browser/outbox.rs new file mode 100644 index 0000000..519d5df --- /dev/null +++ b/rs/moq-scene/src/browser/outbox.rs @@ -0,0 +1,306 @@ +//! Commands for pages not yet carried out, taken in batches: what the studio +//! has yet to tell the helper, which its writer takes, and what the helper +//! has yet to apply, which CEF's UI thread takes. +//! +//! Each page's open, frame rate, and close coalesce, and its pending input +//! goes with it once it closes, so what waits scales with the pages open, +//! never with how often they changed, even while the taker is stuck. A page +//! closed before its open is taken never opens. Input past the backlog drops, +//! except a release, which is owed instead, so a taker that catches up never +//! leaves a page holding a button or key that was let go. + +use std::collections::{HashMap, HashSet, VecDeque}; +use std::sync::{Arc, Condvar, Mutex}; + +use super::protocol::{Command, Id, Open}; +use super::{Button, Input}; + +/// Input waiting for the helper to read it. A helper this far behind is hung, +/// and [`Page::send`](super::Page::send) drops new input rather than wait for +/// it. +const BACKLOG: usize = 256; + +/// A handle on the commands waiting for one helper. Nothing here waits on the +/// helper. +#[derive(Clone, Default)] +pub(crate) struct Commands(Arc<(Mutex, Condvar)>); + +/// Why input was not queued. +pub(crate) enum Refused { + /// The backlog is full, so the input dropped. + Full, + /// The writer stopped, and so did the helper. + Gone, +} + +/// What an input lets go of, which a later release of the same supersedes. +#[derive(Clone, Copy, PartialEq, Eq)] +enum Release { + Button(Button), + Key(i32), + Focus, + Pointer, +} + +impl Release { + fn of(input: &Input) -> Option { + match *input { + Input::Button { + button, down: false, .. + } => Some(Self::Button(button)), + Input::Key { code, down: false, .. } => Some(Self::Key(code)), + Input::Focus(false) => Some(Self::Focus), + Input::Leave { .. } => Some(Self::Pointer), + _ => None, + } + } +} + +#[derive(Default)] +struct Outbox { + /// Pages to open, in order. + opens: Vec<(Id, Open)>, + /// Pages whose open is taken and whose close is not. + open: HashSet, + /// The newest frame rate per page. + rates: HashMap, + /// Input, at most [`BACKLOG`]. + input: VecDeque<(Id, Input)>, + /// Releases past the backlog, oldest first, at most one per page and + /// button, key, focus, and pointer, taken after the input before them. + owed: Vec<(Id, Input)>, + /// Answers owed for pictures copied. + painted: usize, + /// Open pages to close. + closes: Vec, + /// Whether the writer stopped, so nothing more is taken. + stopped: bool, +} + +impl Commands { + /// Queue `f`'s change, unless the writer stopped. + fn push(&self, f: impl FnOnce(&mut Outbox)) -> bool { + let (outbox, ready) = &*self.0; + let mut outbox = outbox.lock().unwrap(); + if outbox.stopped { + return false; + } + f(&mut outbox); + ready.notify_one(); + true + } + + /// Open page `id`. + pub(crate) fn open(&self, id: Id, open: Open) { + self.push(|outbox| outbox.opens.push((id, open))); + } + + /// Close page `id`, which drops what it has waiting, its open too if + /// that is still waiting. + pub(crate) fn close(&self, id: Id) { + self.push(|outbox| { + outbox.input.retain(|(page, _)| *page != id); + outbox.owed.retain(|(page, _)| *page != id); + outbox.rates.remove(&id); + match outbox.opens.iter().position(|(page, _)| *page == id) { + Some(at) => _ = outbox.opens.remove(at), + None if outbox.open.contains(&id) => outbox.closes.push(id), + None => {} + } + }); + } + + /// Paint page `id` at `rate`; false once the writer stopped. + pub(crate) fn rate(&self, id: Id, rate: u32) -> bool { + self.push(|outbox| _ = outbox.rates.insert(id, rate)) + } + + /// Answer the picture the helper waits on. + pub(crate) fn painted(&self) { + self.push(|outbox| outbox.painted += 1); + } + + /// Queue input for page `id`. Past the backlog, or behind an owed release, + /// a release is owed and other input drops. + pub(crate) fn input(&self, id: Id, input: Input) -> Result<(), Refused> { + let (outbox, ready) = &*self.0; + let mut outbox = outbox.lock().unwrap(); + if outbox.stopped { + return Err(Refused::Gone); + } + if outbox.owed.is_empty() && outbox.input.len() < BACKLOG { + outbox.input.push_back((id, input)); + } else if let Some(release) = Release::of(&input) { + outbox + .owed + .retain(|(page, owed)| *page != id || Release::of(owed) != Some(release)); + outbox.owed.push((id, input)); + } else { + return Err(Refused::Full); + } + ready.notify_one(); + Ok(()) + } + + /// Stop taking commands, so [`next`](Self::next) returns `None`. + pub(crate) fn stop(&self) { + let (outbox, ready) = &*self.0; + outbox.lock().unwrap().stopped = true; + ready.notify_all(); + } + + /// Wait for commands to write, or `None` once stopped. + pub(crate) fn next(&self) -> Option> { + let (outbox, ready) = &*self.0; + let mut outbox = outbox.lock().unwrap(); + loop { + if outbox.stopped { + return None; + } + let batch = outbox.take(); + if !batch.is_empty() { + return Some(batch); + } + outbox = ready.wait(outbox).unwrap(); + } + } + + /// Whether nothing waits to be taken. + #[cfg(test)] + pub(crate) fn is_empty(&self) -> bool { + let outbox = self.0.0.lock().unwrap(); + outbox.opens.is_empty() + && outbox.rates.is_empty() + && outbox.input.is_empty() + && outbox.owed.is_empty() + && outbox.painted == 0 + && outbox.closes.is_empty() + } + + /// The commands waiting, without waiting for any. + #[cfg(any(test, feature = "cef"))] + pub(crate) fn take(&self) -> Vec { + self.0.0.lock().unwrap().take() + } +} + +impl Outbox { + /// Everything waiting, each page's open ahead of anything else for it. + fn take(&mut self) -> Vec { + let mut batch = Vec::new(); + for (id, open) in self.opens.drain(..) { + self.open.insert(id); + batch.push(Command::Open(id, open)); + } + batch.extend(self.rates.drain().map(|(id, rate)| Command::FrameRate(id, rate))); + batch.extend(self.input.drain(..).map(|(id, input)| Command::Input(id, input))); + batch.extend(self.owed.drain(..).map(|(id, input)| Command::Input(id, input))); + batch.extend(std::iter::repeat_n(Command::Painted, std::mem::take(&mut self.painted))); + for id in self.closes.drain(..) { + self.open.remove(&id); + batch.push(Command::Close(id)); + } + batch + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn button(down: bool) -> Input { + Input::Button { + x: 1, + y: 1, + button: Button::Left, + down, + clicks: 1, + modifiers: Default::default(), + } + } + + /// Past the backlog, a release is owed rather than dropped, once per + /// button, and goes ahead of newer input, so the page never keeps holding + /// what was let go. + #[test] + fn a_release_past_the_backlog_is_owed() { + let commands = Commands::default(); + commands.input(0, button(true)).ok().unwrap(); + while commands.input(0, Input::Text("x".into())).is_ok() {} + for input in [button(false), button(true), button(false), Input::Text("y".into())] { + let _ = commands.input(0, input); + } + + let received: Vec = commands + .take() + .into_iter() + .map(|command| match command { + Command::Input(0, input) => input, + other => panic!("only input is queued, not {other:?}"), + }) + .collect(); + assert_eq!(received.first(), Some(&button(true))); + assert_eq!(received.last(), Some(&button(false))); + assert_eq!(received.iter().filter(|input| **input == button(false)).count(), 1); + assert!(!received.contains(&Input::Text("y".into()))); + + commands.input(0, Input::Text("z".into())).ok().unwrap(); + assert!(matches!(&commands.take()[..], [Command::Input(0, Input::Text(text))] if text == "z")); + } + + fn open(url: &str) -> Open { + Open { + url: url.into(), + ..Default::default() + } + } + + /// Reloads while the helper reads nothing leave one page waiting, not one + /// per reload: a page closed before its open is taken never opens, and its + /// input and frame rate go with it. + #[test] + fn a_page_closed_before_it_opens_never_opens() { + let commands = Commands::default(); + commands.open(0, open("first")); + assert!(matches!(commands.take()[..], [Command::Open(0, _)])); + + for id in 1..100 { + commands.open(id, open(&"x".repeat(1 << 16))); + commands.rate(id, 60); + let _ = commands.input(id, Input::Text("x".into())); + commands.close(id); + } + commands.open(100, open("last")); + commands.close(0); + + let batch = commands.take(); + assert_eq!(batch.len(), 2, "{batch:?}"); + assert!(matches!(&batch[0], Command::Open(100, open) if open.url == "last")); + assert!(matches!(batch[1], Command::Close(0))); + } + + /// A page's open goes ahead of its input and close, even in one batch, + /// and frame rates coalesce to the newest. + #[test] + fn a_batch_opens_before_anything_else() { + let commands = Commands::default(); + commands.open(1, open("page")); + commands.rate(1, 30); + commands.rate(1, 60); + let _ = commands.input(1, Input::Text("x".into())); + commands.painted(); + + let batch = commands.take(); + assert!(matches!( + batch[..], + [ + Command::Open(1, _), + Command::FrameRate(1, 60), + Command::Input(1, _), + Command::Painted + ] + )); + commands.close(1); + assert!(matches!(commands.take()[..], [Command::Close(1)])); + } +} diff --git a/rs/moq-scene/src/browser/protocol.rs b/rs/moq-scene/src/browser/protocol.rs index c84f52f..fec4d76 100644 --- a/rs/moq-scene/src/browser/protocol.rs +++ b/rs/moq-scene/src/browser/protocol.rs @@ -27,20 +27,40 @@ pub(crate) const MAX_PLANES: usize = 4; /// the other allocate. const MAX_MESSAGE: usize = (super::MAX_SIDE * super::MAX_SIDE * 4) as usize + 1024; +/// A page's number, unique among those one helper opens. +pub(crate) type Id = u64; + /// What the studio asks of the helper. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub(crate) enum Command { - /// The first message: the page to open and how to deliver it. - Open(Open), + /// The first message: how the helper runs every page. + Start(Start), + /// Open a page. + Open(Id, Open), + /// Close a page. + Close(Id), /// The last picture is copied, so the helper may let CEF reuse it. Painted, - /// Interaction with the page. - Input(Input), + /// Interaction with a page. + Input(Id, Input), /// The canvas's frame rate changed, so the page paints at this one. - FrameRate(u32), + FrameRate(Id, u32), +} + +/// How a helper runs every page it opens. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +pub(crate) struct Start { + /// Whether to paint into shared textures the studio imports, rather than + /// into memory it uploads. + pub accelerated: bool, + /// The DRM render node to paint on: the studio's GPU. + pub render_node: Option, + /// The profile every page shares: cookies, storage, and cache. CEF lets + /// one process at a time hold it. + pub profile: String, } -/// The page a helper opens. +/// A page to open. #[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] pub(crate) struct Open { pub url: String, @@ -49,14 +69,6 @@ pub(crate) struct Open { pub height: u32, /// How often the page paints: the canvas's frame rate. pub framerate: u32, - /// Whether to paint into shared textures the studio imports, rather than - /// into memory it uploads. - pub accelerated: bool, - /// The DRM render node to paint on: the studio's GPU. - pub render_node: Option, - /// An empty directory for CEF's profile, since two CEF instances cannot - /// share one. The studio removes it once the helper has exited. - pub profile: String, /// The sound the page plays is delivered at the program bus's rate and /// channel count, so it joins the bus without resampling. pub sample_rate: u32, @@ -66,17 +78,24 @@ pub(crate) struct Open { /// What the helper tells the studio. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub(crate) enum Event { - /// The page painted into a shared texture, whose plane descriptors ride + /// A page painted into a shared texture, whose plane descriptors ride /// with this message. Valid until the studio answers /// [`Command::Painted`]. - Texture(Texture), - /// The page painted into memory: `width * height` premultiplied BGRA + Texture(Id, Texture), + /// A page painted into memory: `width * height` premultiplied BGRA /// pixels, tightly packed. - Pixels { width: u32, height: u32, data: Vec }, + Pixels { + page: Id, + width: u32, + height: u32, + data: Vec, + }, /// Interleaved samples at the [`Open`] rate and channel count. - Audio(Vec), - /// The page cannot be shown, and why. - Failed(String), + Audio(Id, Vec), + /// A page cannot be shown, and why. + Failed(Id, String), + /// No page can be shown, and why. The helper exits next. + Fatal(String), } /// A shared texture's layout, CEF's `cef_accelerated_paint_info_t` minus the @@ -225,17 +244,19 @@ mod tests { }], }; let file = tempfile::tempfile().unwrap(); - helper.send(&Event::Audio(vec![0.5; 3]), &[]).unwrap(); - helper.send(&Event::Texture(texture.clone()), &[file.as_fd()]).unwrap(); - helper.send(&Event::Failed("gone".into()), &[]).unwrap(); + helper.send(&Event::Audio(1, vec![0.5; 3]), &[]).unwrap(); + helper + .send(&Event::Texture(1, texture.clone()), &[file.as_fd()]) + .unwrap(); + helper.send(&Event::Failed(1, "gone".into()), &[]).unwrap(); drop(helper); let (event, fds) = studio.recv::().unwrap().unwrap(); - assert_eq!(event, Event::Audio(vec![0.5; 3])); + assert_eq!(event, Event::Audio(1, vec![0.5; 3])); assert!(fds.is_empty()); let (event, fds) = studio.recv::().unwrap().unwrap(); - assert_eq!(event, Event::Texture(texture)); + assert_eq!(event, Event::Texture(1, texture)); assert_eq!(fds.len(), 1); let ours = rustix::fs::fstat(&fds[0]).unwrap(); let theirs = rustix::fs::fstat(&file).unwrap(); @@ -246,7 +267,7 @@ mod tests { ); let (event, fds) = studio.recv::().unwrap().unwrap(); - assert_eq!(event, Event::Failed("gone".into())); + assert_eq!(event, Event::Failed(1, "gone".into())); assert!(fds.is_empty()); assert!(studio.recv::().unwrap().is_none(), "closed"); @@ -258,6 +279,7 @@ mod tests { let (studio, helper) = pair(); let data: Vec = (0..1920 * 1080 * 4).map(|i| i as u8).collect(); let sent = Event::Pixels { + page: 1, width: 1920, height: 1080, data, diff --git a/rs/moq-scene/src/browser/run.rs b/rs/moq-scene/src/browser/run.rs index 175083b..2a1fd4c 100644 --- a/rs/moq-scene/src/browser/run.rs +++ b/rs/moq-scene/src/browser/run.rs @@ -1,28 +1,32 @@ -//! The studio's side of a browser source: start its helper, then deliver what -//! the helper paints and plays until the source stops. +//! The studio's side of browser sources: start the helper every page shares, +//! then deliver what it paints and plays to each page's source until the +//! source stops. +use std::collections::HashMap; use std::os::fd::OwnedFd; +use std::os::unix::fs::{DirBuilderExt, PermissionsExt}; use std::os::unix::net::UnixStream; use std::os::unix::process::CommandExt; use std::process::{Child, ExitStatus, Stdio}; -use std::sync::mpsc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread::JoinHandle; use moq_audio::bus::Channel as Sound; -use super::protocol::{Channel, Command, Event, Open}; +use super::outbox::Commands; +use super::protocol::{Channel, Command, Event, Id, Open, Start}; use super::{ARG, Helper, Page, Paint}; use crate::Error; use crate::source::{Mailbox, Picture, Status}; +use crate::wgpu; -/// Commands waiting for the helper to read them. A helper this far behind is -/// hung, and [`Page::send`] drops new input rather than wait for it. -const BACKLOG: usize = 256; - -/// Run the page `open` describes until its helper stops, delivering pictures -/// to `picture` and sound to `sound`. Dropping the future stops the helper. +/// Run the page `open` describes until it fails or its helper stops, +/// delivering pictures to `picture` and sound to `sound`. Dropping the future +/// closes the page. /// -/// Fills in how the helper paints and the page's frame rate: `open` names -/// only the page and the bus's format. +/// Fills in the page's frame rate: `open` names only the page and the bus's +/// format. pub(crate) async fn run( helper: &Helper, mut open: Open, @@ -31,76 +35,250 @@ pub(crate) async fn run( sound: Option, status: &kio::Producer, ) -> Result<(), Error> { - // Declared first so it is removed last, once the helper is gone. - let profile = tempfile::tempdir().map_err(|err| Error::Browser(format!("cannot create a profile: {err}")))?; - open.profile = profile.path().to_string_lossy().into_owned(); // A later change waits for the writer, which sends it after `open`. open.framerate = { let mut state = page.state.lock().unwrap(); state.retime = false; state.framerate }; - open.accelerated = super::import::supported(helper); - open.render_node = helper - .render_node - .as_ref() - .and_then(|node| node.to_str()) - .map(str::to_string); - let (mut process, events) = - spawn(helper).map_err(|err| Error::Browser(format!("cannot start {}: {err}", helper.program.display())))?; - let commands = events - .try_clone() - .map_err(|err| Error::Browser(format!("socket: {err}")))?; - let end = match commands - .send(&Command::Open(open), &[]) - .map(|()| writer(commands, page)) - { - Ok(Ok(commands)) => { - { - let mut state = page.state.lock().unwrap(); - state.commands = Some(commands.clone()); - state.flush(); - } - let end = reader(helper, events, commands, page, picture, sound, status).await; - page.state.lock().unwrap().commands = None; - end - } - Ok(Err(err)) => End::Failed(format!("cannot start the page's writer: {err}")), - Err(err) => closed(err), + let instance = helper.instance()?; + let route = Route { + page: page.clone(), + picture: picture.clone(), + sound, + status: status.clone(), + live: false, + end: kio::Producer::new(None), }; + let ended = route.end.consume(); + let opened = instance.open(open, route)?; + let reason = ended + .wait(|end: &kio::Ref<'_, Option>| match end.as_ref() { + Some(reason) => std::task::Poll::Ready(reason.clone()), + None => std::task::Poll::Pending, + }) + .await + .unwrap_or_else(|_| "the page's reader stopped".into()); + drop(opened); // A stopped page leaves the canvas rather than freezing on its last // picture, which for an alert or a ticker would show stale news. *picture.lock().unwrap() = Some(Picture::Rgba { size: moq_video::Size::new(1, 1), pixels: vec![0; 4], }); - match end { - // The helper is exiting or gone, so its status says why. - End::Closed => { - let exit = process - .exited() - .map_err(|err| Error::Browser(format!("the browser process: {err}")))?; - Err(Error::Browser(format!("the browser process exited: {exit}"))) + Err(Error::Browser(reason)) +} + +impl Helper { + /// The running helper, started if there is none or it stopped. + fn instance(&self) -> Result, Error> { + let mut running = self.running.lock().unwrap(); + if let Some(instance) = running.as_ref() + && instance.routes.lock().unwrap().ended.is_none() + { + return Ok(instance.clone()); } - End::Failed(reason) => Err(Error::Browser(reason)), + // The last one, if any, has exited, which freed the profile. + let instance = Arc::new(Instance::start(self)?); + *running = Some(instance.clone()); + Ok(instance) } } -/// Why a page stopped. -#[derive(Clone)] -enum End { - /// The helper closed the socket: it exited, or crashed. - Closed, - /// The helper reported a failure, or broke the protocol. - Failed(String), +/// The helper every page runs in. +pub(crate) struct Instance { + /// What waits for [`writer`] to write it. + commands: Commands, + routes: Arc>, + /// The next page's number. + next: AtomicU64, + /// The socket, shut down to stop the helper. + socket: Channel, + /// The thread delivering the helper's events, which ends once the helper + /// has exited. + reader: Mutex>>, } -/// What a socket error says about the helper. -fn closed(err: std::io::Error) -> End { - use std::io::ErrorKind::*; - match err.kind() { - BrokenPipe | ConnectionReset | UnexpectedEof => End::Closed, - _ => End::Failed(format!("the browser process's socket: {err}")), +/// Where each open page's events go. +#[derive(Default)] +struct Routes { + pages: HashMap, + /// Why the helper stopped, once it has: no page opens in it after. + ended: Option, +} + +/// One open page's source. +struct Route { + page: Page, + picture: Mailbox, + sound: Option, + status: kio::Producer, + /// Whether a picture has arrived. + live: bool, + /// Why the page stopped, once it has. + end: kio::Producer>, +} + +impl Route { + fn end(self, reason: String) { + if let Ok(mut end) = self.end.write() { + *end = Some(reason); + } + } +} + +impl Routes { + /// Stop the page `id`, if it is still open, saying why. + fn end(&mut self, id: Id, reason: String) { + if let Some(route) = self.pages.remove(&id) { + route.end(reason); + } + } + + /// Show `texture`, painted the way `paint` says, on page `id`. + fn show(&mut self, id: Id, texture: wgpu::Texture, paint: Paint) { + let Some(route) = self.pages.get_mut(&id) else { return }; + route.page.state.lock().unwrap().paint = Some(paint); + *route.picture.lock().unwrap() = Some(Picture::Texture(texture)); + if !std::mem::replace(&mut route.live, true) + && let Ok(mut status) = route.status.write() + { + *status = Status::Live; + } + } + + /// Play page `id`'s `samples`. + fn play(&mut self, id: Id, samples: &[f32]) { + let Some(route) = self.pages.get_mut(&id) else { return }; + if let Some(sound) = &mut route.sound + && let Err(err) = sound.write(bytemuck::cast_slice(samples)) + { + self.end(id, format!("the page's sound: {err}")); + } + } +} + +impl Instance { + /// Start a helper on `helper`'s profile. + fn start(helper: &Helper) -> Result { + let failed = |what: &str, err: std::io::Error| Error::Browser(format!("{what}: {err}")); + let profile = + std::path::absolute(&helper.profile).map_err(|err| failed("cannot find the browser profile", err))?; + // Only its owner may read the logins it keeps, even when it already + // existed; a directory another user owns refuses the change. + std::fs::DirBuilder::new() + .recursive(true) + .mode(0o700) + .create(&profile) + .and_then(|()| std::fs::set_permissions(&profile, std::fs::Permissions::from_mode(0o700))) + .map_err(|err| failed(&format!("cannot create the browser profile {}", profile.display()), err))?; + let start = Start { + accelerated: super::import::supported(helper), + render_node: helper + .render_node + .as_ref() + .and_then(|node| node.to_str()) + .map(str::to_string), + profile: profile + .to_str() + .ok_or_else(|| Error::Browser(format!("the browser profile {} is not UTF-8", profile.display())))? + .to_string(), + }; + + let (process, events) = + spawn(helper).map_err(|err| failed(&format!("cannot start {}", helper.program.display()), err))?; + let socket = events.try_clone().map_err(|err| failed("socket", err))?; + let commands = socket.try_clone().map_err(|err| failed("socket", err))?; + commands + .send(&Command::Start(start), &[]) + .map_err(|err| failed("the browser process's socket", err))?; + let routes = Arc::new(Mutex::new(Routes::default())); + let commands = writer(commands).map_err(|err| failed("cannot start the page writer", err))?; + let reader = { + // Not the helper, whose hold on this instance would keep it running + // once every other owner is gone. + let (device, queue) = (helper.device.clone(), helper.queue.clone()); + let (reading, routes) = (commands.clone(), routes.clone()); + let spawned = std::thread::Builder::new() + .name("browser-events".into()) + .spawn(move || read(&device, &queue, process, &events, &reading, &routes)); + // The writer ends with it; the helper, dropped with the closure, is + // killed. + spawned.map_err(|err| { + commands.stop(); + failed("cannot start the page reader", err) + })? + }; + Ok(Self { + commands, + routes, + next: AtomicU64::new(0), + socket, + reader: Mutex::new(Some(reader)), + }) + } + + /// Open a page, delivering its events to `route` until it is closed or + /// stops. Dropping the returned guard closes it. + fn open(self: &Arc, open: Open, route: Route) -> Result { + let id = self.next.fetch_add(1, Ordering::Relaxed); + let page = route.page.clone(); + { + let mut routes = self.routes.lock().unwrap(); + if let Some(reason) = &routes.ended { + return Err(Error::Browser(reason.clone())); + } + // Routed before it opens, so none of its events miss it. + routes.pages.insert(id, route); + } + let opened = Opened { + instance: self.clone(), + page: page.clone(), + id, + }; + // A closed queue means the writer lost the socket, so the helper is + // gone and the reader ends the page. + self.commands.open(id, open); + // After the page opens, since the helper delivers only to an open page. + let mut state = page.state.lock().unwrap(); + state.id = id; + state.commands = Some(self.commands.clone()); + state.flush(); + Ok(opened) + } + + /// Ask the helper to exit, and wait until it has. + pub(super) fn stop(&self) { + self.end(); + if let Some(reader) = self.reader.lock().unwrap().take() { + let _ = reader.join(); + } + } + + /// Ask the helper to exit, which it does once it has closed every page. + fn end(&self) { + let _ = rustix::net::shutdown(&self.socket, rustix::net::Shutdown::Write); + } +} + +impl Drop for Instance { + fn drop(&mut self) { + self.end(); + } +} + +/// An open page, closed when dropped. +struct Opened { + instance: Arc, + page: Page, + id: Id, +} + +impl Drop for Opened { + fn drop(&mut self) { + self.page.state.lock().unwrap().disconnect(); + self.instance.routes.lock().unwrap().pages.remove(&self.id); + self.instance.commands.close(self.id); } } @@ -144,65 +322,30 @@ impl Drop for Process { } } -/// Write the commands queued on the returned sender to `channel`, on a thread +/// Write the commands queued on the returned handle to `channel`, on a thread /// of its own, so a helper that stops reading blocks only that thread, never a -/// sender holding the studio's state. Each write makes room for the releases -/// `page` owes. The thread ends once every sender is dropped or the socket -/// fails, as it does when the helper is killed. -fn writer(channel: Channel, page: &Page) -> std::io::Result> { - let (sender, queue) = mpsc::sync_channel(BACKLOG); - // Weak, since the page holds a sender, which would keep the thread waiting. - let page = std::sync::Arc::downgrade(&page.state); +/// sender holding the studio's state. The thread ends once the handle is +/// stopped or a write fails, as one does when the helper exits, and ends the +/// socket on its way out. +fn writer(channel: Channel) -> std::io::Result { + let commands = Commands::default(); + let outbox = commands.clone(); std::thread::Builder::new() .name("browser-commands".into()) .spawn(move || { - for command in queue { - if channel.send(&command, &[]).is_err() { - break; - } - if let Some(page) = page.upgrade() { - page.lock().unwrap().flush(); + 'write: while let Some(batch) = outbox.next() { + for command in batch { + if channel.send(&command, &[]).is_err() { + break 'write; + } } } + // The helper exits once it reads the end, so one the studio can no + // longer reach never keeps running unheard. + outbox.stop(); + let _ = rustix::net::shutdown(&channel, rustix::net::Shutdown::Write); })?; - Ok(sender) -} - -/// Run [`deliver`] on a thread of its own, since it blocks on the socket, and -/// wait for why it ended. A thread that cannot start, or that stops without -/// saying why, fails the page with the reason rather than leaving it live. -async fn reader( - helper: &Helper, - events: Channel, - commands: mpsc::SyncSender, - page: &Page, - picture: &Mailbox, - sound: Option, - status: &kio::Producer, -) -> End { - let done = kio::Producer::new(None); - let ended = done.consume(); - let spawned = { - let (helper, page, picture, status) = (helper.clone(), page.clone(), picture.clone(), status.clone()); - std::thread::Builder::new() - .name("browser-events".into()) - .spawn(move || { - let end = deliver(&helper, &events, &commands, &page, &picture, sound, &status); - if let Ok(mut done) = done.write() { - *done = Some(end); - } - }) - }; - if let Err(err) = spawned { - return End::Failed(format!("cannot start the page's reader: {err}")); - } - let end = ended - .wait(|end: &kio::Ref<'_, Option>| match end.as_ref() { - Some(end) => std::task::Poll::Ready(end.clone()), - None => std::task::Poll::Pending, - }) - .await; - end.unwrap_or_else(|_| End::Failed("the page's reader stopped".into())) + Ok(commands) } /// Start a helper, handing it one end of a socket as its stdin. @@ -217,55 +360,83 @@ fn spawn(helper: &Helper) -> std::io::Result<(Process, Channel)> { Ok((Process { child, exit: None }, Channel::new(ours))) } -/// Read the helper's events until it closes the socket or reports a failure. +/// Deliver the helper's events until it exits or breaks the protocol, then +/// stop it and every page still open, saying why. +fn read( + device: &wgpu::Device, + queue: &wgpu::Queue, + mut process: Process, + events: &Channel, + commands: &Commands, + routes: &Mutex, +) { + let reason = match deliver(device, queue, events, commands, routes) { + // Gone, or exiting: wait for it, so the next helper finds the profile + // free. + Ok(()) => match process.exited() { + Ok(exit) => format!("the browser process exited: {exit}"), + Err(err) => format!("the browser process: {err}"), + }, + Err(reason) => { + let _ = process.stop(); + reason + } + }; + // Nothing more reaches it, so the writer ends. + commands.stop(); + let mut routes = routes.lock().unwrap(); + routes.ended = Some(reason.clone()); + for (_, route) in routes.pages.drain() { + route.end(reason.clone()); + } +} + +/// Route the helper's events to their pages until it closes the socket, or +/// fail with why it cannot go on. fn deliver( - helper: &Helper, + device: &wgpu::Device, + queue: &wgpu::Queue, events: &Channel, - commands: &mpsc::SyncSender, - page: &Page, - picture: &Mailbox, - mut sound: Option, - status: &kio::Producer, -) -> End { - let mut live = false; + commands: &Commands, + routes: &Mutex, +) -> Result<(), String> { + use std::io::ErrorKind::*; + loop { let (event, fds) = match events.recv::() { Ok(Some(received)) => received, - Ok(None) => return End::Closed, - Err(err) => return closed(err), + Ok(None) => return Ok(()), + Err(err) if matches!(err.kind(), BrokenPipe | ConnectionReset | UnexpectedEof) => return Ok(()), + Err(err) => return Err(format!("the browser process's socket: {err}")), }; - let (texture, paint) = match event { - Event::Texture(texture) => (super::import::texture(helper, &texture, fds), Paint::Gpu), - Event::Pixels { width, height, data } => ( - super::upload(&helper.device, &helper.queue, width, height, &data), - Paint::Cpu, - ), - Event::Audio(samples) => { - if let Some(sound) = &mut sound - && let Err(err) = sound.write(bytemuck::cast_slice(&samples)) - { - return End::Failed(format!("the page's sound: {err}")); - } + let (id, texture, paint) = match event { + Event::Texture(id, texture) => (id, super::import::texture(device, queue, &texture, fds), Paint::Gpu), + Event::Pixels { + page, + width, + height, + data, + } => (page, super::upload(device, queue, width, height, &data), Paint::Cpu), + Event::Audio(id, samples) => { + routes.lock().unwrap().play(id, &samples); continue; } - Event::Failed(reason) => return End::Failed(reason), - }; - // The copy is done, so the helper may hand the source back to CEF. - // Waits behind queued input, never dropped: the helper waits on it. A - // closed queue means the writer lost the socket, so the helper is gone. - if commands.send(Command::Painted).is_err() { - return End::Closed; - } - let texture = match texture { - Ok(texture) => texture, - Err(reason) => return End::Failed(reason), + Event::Failed(id, reason) => { + routes.lock().unwrap().end(id, reason); + continue; + } + Event::Fatal(reason) => return Err(reason), }; - page.state.lock().unwrap().paint = Some(paint); - *picture.lock().unwrap() = Some(Picture::Texture(texture)); - if !std::mem::replace(&mut live, true) - && let Ok(mut status) = status.write() - { - *status = Status::Live; + // The copy is done, so the helper may hand the source back to CEF, + // even for a page closed since: the helper waits on it. A closed queue + // means the writer lost the socket, so the helper is exiting, and stops + // waiting; reading on until it has exited keeps it from blocking on a + // full socket. + commands.painted(); + let mut routes = routes.lock().unwrap(); + match texture { + Ok(texture) => routes.show(id, texture, paint), + Err(reason) => routes.end(id, reason), } } } @@ -273,47 +444,51 @@ fn deliver( #[cfg(test)] mod tests { use super::*; - use crate::browser::{Input, Modifiers}; + use crate::browser::Input; /// A helper that stops reading costs dropped input, never a sender stuck - /// on a full socket while it holds the studio's state. + /// on a full socket while it holds the studio's state. Pages still open + /// and close without waiting, and reloads meanwhile leave only the newest + /// page to open once the helper reads again. #[test] - fn a_helper_that_stops_reading_never_blocks_input() { + fn a_helper_that_stops_reading_blocks_nothing() { let (ours, theirs) = UnixStream::pair().unwrap(); + let commands = writer(Channel::new(ours)).unwrap(); let page = Page::default(); - page.state.lock().unwrap().commands = Some(writer(Channel::new(ours), &page).unwrap()); - // Far more than the socket buffers and the backlog hold together. - for _ in 0..4096 { - page.send(Input::Text("x".repeat(1024))); + page.state.lock().unwrap().commands = Some(commands.clone()); + // An open far larger than the socket buffers: once the writer takes + // it, it is stuck writing it until the helper reads. + let stuck = Open { + url: "x".repeat(16 << 20), + ..Default::default() + }; + commands.open(0, stuck); + while !commands.is_empty() { + std::thread::yield_now(); } - assert!(page.state.lock().unwrap().backlogged); - drop(theirs); - } - /// Each write makes room for an owed release, so it reaches the page once - /// the helper reads again, with no newer input to carry it. - #[test] - fn the_writer_delivers_owed_releases() { - let (ours, theirs) = UnixStream::pair().unwrap(); - let page = Page::default(); - let commands = writer(Channel::new(ours), &page).unwrap(); - let release = Input::Key { - code: 65, - down: false, - modifiers: Modifiers::default(), - }; - { - let mut state = page.state.lock().unwrap(); - state.commands = Some(commands.clone()); - state.owed.push_back(release.clone()); + for _ in 0..1024 { + page.send(Input::Text("x".into())); + } + assert!(page.state.lock().unwrap().backlogged); + for id in 1..100 { + commands.open(id, Open::default()); + commands.close(id); } - commands.send(Command::Painted).unwrap(); + commands.open(100, Open::default()); let helper = Channel::new(theirs); - assert!(matches!(helper.recv::().unwrap(), Some((Command::Painted, _)))); - let Some((Command::Input(input), _)) = helper.recv().unwrap() else { - panic!("the owed release never came"); - }; - assert_eq!(input, release); + assert!(matches!( + helper.recv::().unwrap(), + Some((Command::Open(0, _), _)) + )); + loop { + match helper.recv::().unwrap() { + Some((Command::Open(100, _), _)) => break, + Some((Command::Open(id, _) | Command::Close(id), _)) => panic!("page {id} was reloaded away"), + Some(_) => {} + None => panic!("the writer stopped"), + } + } } } diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index f5ce216..5e3f820 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -122,7 +122,8 @@ pub enum Status { /// Waiting for its broadcast to be announced. Offline, /// Stopped, with the reason. A broadcast source keeps waiting for the - /// broadcast to come back; every other source stays stopped until edited. + /// broadcast to come back; every other source stays stopped until edited + /// or [restarted](Sources::restart). Failed(String), } @@ -199,6 +200,12 @@ impl Sources { drivers } + /// Stop the source called `name`, so the next [`sync`](Self::sync) starts + /// it again, as an edit would, such as to reload a web page. + pub fn restart(&mut self, name: &str) { + self.feeds.remove(name); + } + /// The running source called `name`. pub fn get(&self, name: &str) -> Option<&Feed> { self.feeds.get(name) @@ -1526,7 +1533,7 @@ mod tests { } /// Without a helper to run it in, a browser source fails rather than - /// staying blank. + /// staying blank, until a restart starts a new page. #[tokio::test] async fn a_page_without_a_helper_says_so() { local(async { @@ -1551,6 +1558,11 @@ mod tests { assert!(reason.contains("web pages") || reason.contains("Linux"), "{reason}"); assert!(chat.page().is_some_and(|page| page.paint().is_none())); assert!(chat.strip().is_some(), "a page feeds the bus"); + + let failed = chat.page().unwrap().clone(); + sources.restart("chat"); + assert_eq!(sources.sync(&collection).len(), 1, "a restart starts it again"); + assert!(!sources.get("chat").unwrap().page().unwrap().same(&failed)); }) .await; } diff --git a/rs/moq-studio/src/engine.rs b/rs/moq-studio/src/engine.rs index 6f1a0db..d8a131e 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -195,6 +195,13 @@ impl State { Ok(()) } + /// Restart the source called `name`, as an edit would, such as to reload a + /// web page. + pub fn restart(&mut self, name: &str) { + self.sources.restart(name); + self.stale = true; + } + /// After any change to the collection or its history, which the autosave /// writes once they settle. fn edited(&mut self) { @@ -606,7 +613,9 @@ impl Engine { /// publish the copies on their way to it and what it still holds, and to /// end its video track, so quitting loses none of the program's last /// pictures. The program audio ends its track too, so a recording reading - /// the program reaches the end of each. + /// the program reaches the end of each. Then stop the browser sources' + /// helper and wait for it to exit, so their profile is saved and free for + /// the next launch. pub fn stop(&self) { self.stopped.store(true, Ordering::Relaxed); let threads = std::mem::take(&mut *self.threads.lock().unwrap()); @@ -616,6 +625,11 @@ impl Engine { tracing::error!(name, "the thread panicked"); } } + // Taken out first, so the wait holds no lock the window draws with. + let browser = self.state.lock().unwrap().sources.context().browser.clone(); + if let Some(browser) = browser { + browser.stop(); + } } /// Write the collection and its history to their files unless they @@ -1350,7 +1364,8 @@ fn encode_frames( } /// What browser sources run in: this executable, which `main` hands to -/// `moq_scene::browser::child` when it is started as their helper. +/// `moq_scene::browser::child` when it is started as their helper, on the +/// user's browser profile. fn browser(gpu: &Gpu) -> anyhow::Result> { if !cfg!(feature = "cef") { return Ok(None); @@ -1358,12 +1373,28 @@ fn browser(gpu: &Gpu) -> anyhow::Result> { let program = std::env::current_exe().context("cannot find the studio's executable")?; Ok(Some(moq_scene::browser::Helper::new( program, + browser_profile()?, &gpu.adapter, &gpu.device, &gpu.queue, ))) } +/// Where browser sources keep their cookies and logins, whichever scene file +/// is open: `moq-studio/browser` in the user's data directory, as the XDG base +/// directories name it, since browser sources run on Linux only for now. +fn browser_profile() -> anyhow::Result { + let data = match std::env::var_os("XDG_DATA_HOME").map(PathBuf::from) { + // The spec ignores a relative one. + Some(data) if data.is_absolute() => data, + _ => { + let home = std::env::var_os("HOME").context("no HOME to keep the browser profile in")?; + PathBuf::from(home).join(".local/share") + } + }; + Ok(data.join("moq-studio/browser")) +} + /// Run a source on its own thread until its feed is dropped, or fail the /// source when its thread cannot start. fn spawn(driver: Driver) { diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index cb4c091..b10a2cf 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -1214,6 +1214,7 @@ impl Panels { let mut state = engine.state.lock().unwrap(); let mut remove = None; let mut clone = None; + let mut reload = None; let now = kio::clock::now(); egui::ScrollArea::vertical().show(ui, |ui| { @@ -1290,6 +1291,14 @@ impl Panels { { clone = Some(name.clone()); } + if matches!(source, Source::Browser { .. }) + && ui + .small_button("Reload") + .on_hover_text("Restart the page, keeping the cookies and logins every page shares") + .clicked() + { + reload = Some(name.clone()); + } if ui.small_button("Remove").clicked() { remove = Some(name.clone()); } @@ -1310,6 +1319,9 @@ impl Panels { } }); + if let Some(name) = reload { + state.restart(&name); + } let result = if let Some(name) = remove { // Removing the items that show it moves the ones after them. self.deselect_item(); diff --git a/rs/moq-studio/tests/browser.rs b/rs/moq-studio/tests/browser.rs index f46a7f4..661e037 100644 --- a/rs/moq-studio/tests/browser.rs +++ b/rs/moq-studio/tests/browser.rs @@ -1,8 +1,10 @@ //! Browser sources against the real helper: this package's binary, built with -//! `cef`, rendering `data:` pages with nothing fetched from the network. +//! `cef`, rendering `data:` pages, or pages served on loopback, with nothing +//! fetched from the network. #![cfg(all(target_os = "linux", feature = "cef"))] +use std::path::Path; use std::time::Duration; use kio::clock::Instant; @@ -16,7 +18,7 @@ const TIMEOUT: Duration = Duration::from_secs(20); const BLUE: [u8; 4] = [0, 0, 255, 255]; /// A studio of one scene, "Main", drawing `sources` back to front over the -/// whole of a small canvas. +/// whole of a small canvas. Dropping it stops its helper. struct Studio { collection: Collection, sources: Sources, @@ -25,12 +27,23 @@ struct Studio { queue: wgpu::Queue, /// Advances the bus, which meters each source's strip. mixer: moq_audio::bus::Driver, + helper: Helper, _origin: moq_net::origin::Producer, _dir: tempfile::TempDir, + _profile: Option, } impl Studio { + /// A studio with a fresh browser profile. async fn new(sources: &[(&str, Source)]) -> Self { + let profile = tempfile::tempdir().unwrap(); + let mut studio = Self::on(profile.path(), sources).await; + studio._profile = Some(profile); + studio + } + + /// A studio whose pages keep their cookies in `profile`. + async fn on(profile: &Path, sources: &[(&str, Source)]) -> Self { let instance = wgpu::Instance::default(); let adapter = instance .request_adapter(&wgpu::RequestAdapterOptions::default()) @@ -60,6 +73,7 @@ impl Studio { let (origin, driver) = moq_net::origin::Producer::new(Default::default()); tokio::spawn(moq_net::time::run(driver)); let (bus, mixer) = moq_audio::bus::Bus::new(Default::default()).unwrap(); + let helper = Helper::new(env!("CARGO_BIN_EXE_moq-studio"), profile, &adapter, &device, &queue); let mut sources = Sources::new(Context { program: bus.mix(), bus, @@ -67,7 +81,7 @@ impl Studio { dir: dir.path().to_path_buf(), output: moq_video::render::output(&device), publish: None, - browser: Some(Helper::new(env!("CARGO_BIN_EXE_moq-studio"), &adapter, &device, &queue)), + browser: Some(helper.clone()), }); for driver in sources.sync(&collection) { tokio::task::spawn_local(driver.run()); @@ -82,8 +96,18 @@ impl Studio { device, queue, mixer, + helper, _origin: origin, _dir: dir, + _profile: None, + } + } + + /// Restart the source `name`, as its Reload button does. + fn reload(&mut self, name: &str) { + self.sources.restart(name); + for driver in self.sources.sync(&self.collection) { + tokio::task::spawn_local(driver.run()); } } @@ -159,6 +183,12 @@ impl Studio { } } +impl Drop for Studio { + fn drop(&mut self) { + self.helper.stop(); + } +} + fn near(actual: [u8; 4], expected: [u8; 4]) -> bool { actual.iter().zip(expected).all(|(a, e)| a.abs_diff(e) <= 2) } @@ -239,8 +269,25 @@ async fn a_click_reaches_the_page() { .await; } -/// A page whose renderer crashes, or whose whole browser process does, fails -/// its own source and leaves the canvas, while everything else keeps drawing. +/// A page that closes itself fails its own source and leaves the canvas, +/// while the page beside it in the same helper keeps drawing. +#[tokio::test(flavor = "multi_thread")] +#[ignore = "needs CEF and a Vulkan device; run by just render-sw"] +async fn a_page_that_closes_itself_fails_alone() { + local(async { + let closer = page(""); + let red = page(""); + let mut studio = Studio::new(&[("red", red), ("closer", closer)]).await; + let reason = studio.until_failed("closer").await; + assert!(reason.contains("closed itself"), "{reason}"); + studio.until_center([255, 0, 0, 255], &["closer"]).await; + }) + .await; +} + +/// A page whose renderer crashes fails its own source and leaves the canvas, +/// while everything else keeps drawing. A crash of the helper every page +/// shares fails them all and nothing else, and a reload starts a new one. #[tokio::test(flavor = "multi_thread")] #[ignore = "needs CEF and a Vulkan device; run by just render-sw"] async fn a_crash_drops_only_its_source() { @@ -267,10 +314,88 @@ async fn a_crash_drops_only_its_source() { assert_eq!(studio.sources.get("blue").unwrap().status(), Status::Live); studio.until_center(BLUE, &["renderer", "browser"]).await; + + studio.reload("browser"); + studio.until_center([255, 0, 0, 255], &["renderer"]).await; }) .await; } +/// Cookies, even one without an expiry, survive a reload of the page and a +/// restart of the studio, since every page keeps them in one profile on disk. +/// A second studio on the profile while it is in use fails its pages saying +/// why, and leaves the first one's drawing. +#[tokio::test(flavor = "multi_thread")] +#[ignore = "needs CEF and a Vulkan device; run by just render-sw"] +async fn cookies_survive_a_restart() { + local(async { + // Red until it has seen its cookie, then green. + let url = serve( + "", + ) + .await; + let page = Source::Browser { + url, + width: 320, + height: 240, + }; + let profile = tempfile::tempdir().unwrap(); + let mut studio = Studio::on(profile.path(), &[("page", page.clone())]).await; + studio.until_center([255, 0, 0, 255], &[]).await; + studio.reload("page"); + studio.until_center(GREEN, &[]).await; + + let second = Studio::on(profile.path(), &[("page", page.clone())]).await; + let reason = second.until_failed("page").await; + assert!(reason.contains("another studio"), "{reason}"); + drop(second); + studio.until_center(GREEN, &[]).await; + drop(studio); + + let mut studio = Studio::on(profile.path(), &[("page", page)]).await; + studio.until_center(GREEN, &[]).await; + }) + .await; +} + +const GREEN: [u8; 4] = [0, 255, 0, 255]; + +/// Serve `html` over HTTP on loopback to every request, returning its URL, +/// since a cookie needs an origin a `data:` page lacks. +async fn serve(html: &'static str) -> String { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}/", listener.local_addr().unwrap()); + tokio::spawn(async move { + loop { + let Ok((mut stream, _)) = listener.accept().await else { + return; + }; + tokio::spawn(async move { + // The request, up to the blank line that ends its headers. + let mut request = Vec::new(); + let mut buffer = [0; 1024]; + while !request.ends_with(b"\r\n\r\n") { + match stream.read(&mut buffer).await { + Ok(0) | Err(_) => return, + Ok(read) => request.extend_from_slice(&buffer[..read]), + } + } + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: text/html\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{html}", + html.len() + ); + let _ = stream.write_all(response.as_bytes()).await; + }); + } + }); + url +} + /// The one helper this test process still runs: a child started with the /// helper's argument. fn helper() -> Option {