From ac94c7da2f03a0270a8207d614d23180a40e7ff5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 20:06:23 -0800 Subject: [PATCH 1/8] chore(quest): claim quest/m0/browser-profile.md Quest-Claim: 2be9ea4d3ce5a0e6 From 52680d510351aecd53c6ffc168198054d9d24d51 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 22:35:59 -0800 Subject: [PATCH 2/8] feat(scene): browser sources share one persistent profile, and reload Every page runs in one helper, CEF's browser process, on a profile kept in the user's XDG data directory, so cookies and logins survive a reload and a studio restart. CEF lets only one process hold a profile, so per-page helpers could not share one; each page still renders in its own renderer process. A crash of the helper fails every page, and the next page to start starts a new one. The studio stops the helper on quit so the profile is saved and free for a relaunch, and a second studio on the same profile fails its pages saying so. Sources gains a Reload button for browser sources, which restarts the page. Co-Authored-By: Claude Opus 5.5 --- README.md | 19 +- quest/m0/README.md | 1 - quest/m0/browser-profile.md | 24 -- quest/m0/browser-source.md | 12 +- rs/moq-scene/src/browser/host.rs | 414 +++++++++++++--------- rs/moq-scene/src/browser/mod.rs | 81 +++-- rs/moq-scene/src/browser/protocol.rs | 76 ++-- rs/moq-scene/src/browser/run.rs | 509 +++++++++++++++++++-------- rs/moq-scene/src/source.rs | 16 +- rs/moq-studio/src/engine.rs | 38 +- rs/moq-studio/src/main.rs | 2 + rs/moq-studio/src/ui.rs | 12 + rs/moq-studio/tests/browser.rs | 130 ++++++- 13 files changed, 925 insertions(+), 409 deletions(-) delete mode 100644 quest/m0/browser-profile.md diff --git a/README.md b/README.md index 7bba27a..82703c9 100644 --- a/README.md +++ b/README.md @@ -92,7 +92,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 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 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. | @@ -140,12 +140,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 64c5905..1eb6192 100644 --- a/quest/m0/README.md +++ b/quest/m0/README.md @@ -99,7 +99,6 @@ repository. - [Audio filters](/quest/m0/audio-filters.md) - per-source noise suppression, gate, compressor, limiter, and plugins, configured from the studio - [Recording](/quest/m0/recording.md) - each canvas's program, and optionally each source, recorded to disk alongside live outputs - [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 - [Web studio](/quest/m0/web.md) - the full studio in wasm on WebGPU and WebCodecs, served from this repository - [Virtual camera](/quest/m0/vcam/README.md) - a canvas's program exposed as a system camera on Linux, Windows, and macOS 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..39250d5 100644 --- a/quest/m0/browser-source.md +++ b/quest/m0/browser-source.md @@ -20,11 +20,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..8f8fdd6 100644 --- a/rs/moq-scene/src/browser/host.rs +++ b/rs/moq-scene/src/browser/host.rs @@ -1,8 +1,9 @@ -//! 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; @@ -10,7 +11,7 @@ use std::sync::{Arc, Mutex, mpsc}; use cef::*; -use super::protocol::{Channel, Command, Event, Format, Open, Plane, Texture}; +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 +33,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 +56,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 +88,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 +133,202 @@ 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. +fn commands(shared: Arc, reader: Channel, painted: mpsc::Sender<()>) { + loop { + match reader.recv::() { + Ok(Some((Command::Painted, _))) => { + let _ = painted.send(()); + } + Ok(Some((Command::Open(id, open), _))) => { + let shared = shared.clone(); + post(move || self::open(shared, id, open)); + } + Ok(Some((Command::Close(id), _))) => post(move || close(id)), + Ok(Some((Command::Input(id, input), _))) => post(move || deliver(id, input)), + Ok(Some((Command::FrameRate(id, rate), _))) => post(move || retime(id, rate)), + Ok(Some((Command::Start(_), _))) | Ok(None) | Err(_) => break, + } + } + // Unblocks a paint waiting on the studio. + drop(painted); + post(|| { + 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); + } + }); +} + +/// 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(id), ); - 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 +372,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 +402,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 +452,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 +480,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 +503,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 +526,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 +543,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 +559,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 +577,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 +587,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 +618,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 +637,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 { + id: Id, + } 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 +671,13 @@ wrap_life_span_handler! { } fn on_before_close(&self, _browser: Option<&mut Browser>) { - quit_message_loop(); + let last = PAGES.with_borrow_mut(|pages| { + pages.remove(&self.id); + pages.is_empty() + }); + if last && QUITTING.get() { + quit_message_loop(); + } } } } diff --git a/rs/moq-scene/src/browser/mod.rs b/rs/moq-scene/src/browser/mod.rs index bf93fa1..97fa646 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. @@ -65,7 +69,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 +78,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 +113,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,9 +156,13 @@ 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, which [`run::writer`] drains, while + /// the page is open. #[cfg(target_os = "linux")] commands: Option>, + /// The page's number in the helper. + #[cfg(target_os = "linux")] + id: protocol::Id, /// Whether input is being dropped because the queue is full. #[cfg(target_os = "linux")] backlogged: bool, @@ -153,12 +190,12 @@ impl PageState { 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)) { + true => match commands.try_send(protocol::Command::Input(self.id, input)) { Ok(()) => { self.backlogged = false; return; } - Err(TrySendError::Full(protocol::Command::Input(input))) => input, + Err(TrySendError::Full(protocol::Command::Input(_, input))) => input, Err(TrySendError::Full(_)) => unreachable!("only input is sent here"), Err(TrySendError::Disconnected(_)) => return self.disconnect(), }, @@ -181,16 +218,16 @@ impl PageState { let Some(commands) = &self.commands else { return }; if self.retime { - match commands.try_send(protocol::Command::FrameRate(self.framerate)) { + match commands.try_send(protocol::Command::FrameRate(self.id, 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)) { + match commands.try_send(protocol::Command::Input(self.id, input)) { Ok(()) => {} - Err(TrySendError::Full(protocol::Command::Input(input))) => { + Err(TrySendError::Full(protocol::Command::Input(_, input))) => { self.owed.push_front(input); return; } @@ -200,7 +237,7 @@ impl PageState { } } - /// 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(); @@ -426,7 +463,7 @@ mod tests { // 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() { + while let Ok(protocol::Command::Input(_, input)) = queue.try_recv() { received.push(input); page.state.lock().unwrap().flush(); } @@ -444,13 +481,13 @@ mod tests { page.state.lock().unwrap().commands = Some(sender); page.set_framerate(30); page.set_framerate(30); - assert!(matches!(queue.try_recv(), Ok(protocol::Command::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"); page.send(Input::Text("x".into())); page.set_framerate(60); - assert!(matches!(queue.try_recv(), Ok(protocol::Command::Input(_)))); + assert!(matches!(queue.try_recv(), Ok(protocol::Command::Input(..)))); page.state.lock().unwrap().flush(); - assert!(matches!(queue.try_recv(), Ok(protocol::Command::FrameRate(60)))); + assert!(matches!(queue.try_recv(), Ok(protocol::Command::FrameRate(_, 60)))); } } 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..f4e45b8 100644 --- a/rs/moq-scene/src/browser/run.rs +++ b/rs/moq-scene/src/browser/run.rs @@ -1,28 +1,35 @@ -//! 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; 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, mpsc}; +use std::thread::JoinHandle; use moq_audio::bus::Channel as Sound; -use super::protocol::{Channel, Command, Event, Open}; +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 +38,247 @@ 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 { + /// Queued commands, which [`writer`] writes. + commands: mpsc::SyncSender, + 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, + /// Pages closed while the command queue was full, which [`writer`] closes + /// as the queue makes room. + closing: Vec, + /// 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. + std::fs::DirBuilder::new() + .recursive(true) + .mode(0o700) + .create(&profile) + .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, routes.clone()).map_err(|err| failed("cannot start the page writer", err))?; + let reader = { + let (helper, commands, routes) = (helper.clone(), commands.clone(), routes.clone()); + std::thread::Builder::new() + .name("browser-events".into()) + .spawn(move || read(&helper, process, &events, &commands, &routes)) + .map_err(|err| 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. + let _ = self.commands.send(Command::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(); + let mut routes = self.instance.routes.lock().unwrap(); + routes.pages.remove(&self.id); + // Never waits on the helper: a full queue closes it as it drains. + if let Err(mpsc::TrySendError::Full(_)) = self.instance.commands.try_send(Command::Close(self.id)) { + routes.closing.push(self.id); + } } } @@ -146,65 +324,41 @@ impl Drop for Process { /// Write the commands queued on the returned sender 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> { +/// sender holding the studio's state. Each write makes room for the pages +/// `routes` closed and the releases their pages owe. The thread ends once +/// every sender is dropped or a write fails, as one does when the helper +/// exits, and ends the socket on its way out. +fn writer(channel: Channel, routes: Arc>) -> 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); std::thread::Builder::new() .name("browser-commands".into()) .spawn(move || { - for command in queue { + 'write: for command in queue { if channel.send(&command, &[]).is_err() { break; } - if let Some(page) = page.upgrade() { - page.lock().unwrap().flush(); + // Taken first, so a closing page never waits on a socket write. + let (closing, pages): (Vec, Vec) = { + let mut routes = routes.lock().unwrap(); + let pages = routes.pages.values().map(|route| route.page.clone()).collect(); + (std::mem::take(&mut routes.closing), pages) + }; + for id in closing { + if channel.send(&Command::Close(id), &[]).is_err() { + break 'write; + } + } + for page in pages { + page.state.lock().unwrap().flush(); } } + // The helper exits once it reads the end, so one the studio can no + // longer reach never keeps running unheard. + 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())) -} - /// Start a helper, handing it one end of a socket as its stdin. fn spawn(helper: &Helper) -> std::io::Result<(Process, Channel)> { let (ours, theirs) = UnixStream::pair()?; @@ -217,55 +371,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( + helper: &Helper, + mut process: Process, + events: &Channel, + commands: &mpsc::SyncSender, + routes: &Mutex, +) { + let reason = match deliver(helper, 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 + } + }; + 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, events: &Channel, commands: &mpsc::SyncSender, - page: &Page, - picture: &Mailbox, - mut sound: Option, - status: &kio::Producer, -) -> End { - let mut live = false; + 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 } => ( + let (id, texture, paint) = match event { + Event::Texture(id, texture) => (id, super::import::texture(helper, &texture, fds), Paint::Gpu), + Event::Pixels { + page, + width, + height, + data, + } => ( + page, 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}")); - } + 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. Waits behind queued input, never + // dropped: 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. + let _ = commands.send(Command::Painted); + let mut routes = routes.lock().unwrap(); + match texture { + Ok(texture) => routes.show(id, texture, paint), + Err(reason) => routes.end(id, reason), } } } @@ -275,13 +457,33 @@ mod tests { use super::*; use crate::browser::{Input, Modifiers}; + /// An open page, as [`Instance::open`] routes one, with a writer to + /// `ours`. + fn routed(ours: UnixStream) -> (Page, mpsc::SyncSender, Arc>) { + let routes = Arc::new(Mutex::new(Routes::default())); + let commands = writer(Channel::new(ours), routes.clone()).unwrap(); + let page = Page::default(); + page.state.lock().unwrap().commands = Some(commands.clone()); + routes.lock().unwrap().pages.insert( + 0, + Route { + page: page.clone(), + picture: Default::default(), + sound: None, + status: kio::Producer::new(Status::Starting), + live: false, + end: kio::Producer::new(None), + }, + ); + (page, commands, routes) + } + /// A helper that stops reading costs dropped input, never a sender stuck /// on a full socket while it holds the studio's state. #[test] fn a_helper_that_stops_reading_never_blocks_input() { let (ours, theirs) = UnixStream::pair().unwrap(); - let page = Page::default(); - page.state.lock().unwrap().commands = Some(writer(Channel::new(ours), &page).unwrap()); + let (page, _commands, _routes) = routed(ours); // Far more than the socket buffers and the backlog hold together. for _ in 0..4096 { page.send(Input::Text("x".repeat(1024))); @@ -290,28 +492,29 @@ mod tests { 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. + /// Each write makes room for an owed release and a page closed past the + /// backlog, so both reach the helper once it reads again, with nothing + /// newer to carry them. #[test] - fn the_writer_delivers_owed_releases() { + fn the_writer_delivers_owed_releases_and_closes() { let (ours, theirs) = UnixStream::pair().unwrap(); - let page = Page::default(); - let commands = writer(Channel::new(ours), &page).unwrap(); + let (page, commands, routes) = routed(ours); 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()); - } + page.state.lock().unwrap().owed.push_back(release.clone()); + routes.lock().unwrap().closing.push(7); commands.send(Command::Painted).unwrap(); let helper = Channel::new(theirs); assert!(matches!(helper.recv::().unwrap(), Some((Command::Painted, _)))); - let Some((Command::Input(input), _)) = helper.recv().unwrap() else { + assert!(matches!( + helper.recv::().unwrap(), + Some((Command::Close(7), _)) + )); + let Some((Command::Input(0, input), _)) = helper.recv().unwrap() else { panic!("the owed release never came"); }; assert_eq!(input, release); diff --git a/rs/moq-scene/src/source.rs b/rs/moq-scene/src/source.rs index c6c1773..6aeaab7 100644 --- a/rs/moq-scene/src/source.rs +++ b/rs/moq-scene/src/source.rs @@ -102,7 +102,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), } @@ -177,6 +178,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) @@ -1434,7 +1441,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 { @@ -1459,6 +1466,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 fd4d57e..b4b74c0 100644 --- a/rs/moq-studio/src/engine.rs +++ b/rs/moq-studio/src/engine.rs @@ -15,7 +15,7 @@ //! - The UI runs on the main thread and edits [`State`] under its lock. use std::collections::BTreeMap; -use std::path::Path; +use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex, mpsc}; use std::time::Duration; @@ -191,6 +191,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) { @@ -565,6 +572,16 @@ impl Engine { self.autosave.reload(&self.state) } + /// 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_browsers(&self) { + // 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 over what the files hold now. pub fn keep(&self) -> anyhow::Result<()> { self.autosave.keep(&self.state) @@ -1204,7 +1221,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); @@ -1212,12 +1230,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/main.rs b/rs/moq-studio/src/main.rs index fbdc3c8..bd16d2d 100644 --- a/rs/moq-studio/src/main.rs +++ b/rs/moq-studio/src/main.rs @@ -172,6 +172,8 @@ fn main() -> anyhow::Result { }; // The edits since the last autosave, so quitting loses none. let saved = engine.save(); + // So a quick relaunch finds the browser profile saved and free. + engine.stop_browsers(); if let (Err(err), Err(_)) = (&saved, &ran) { tracing::error!(error = format!("{err:#}"), "cannot save on quit"); } diff --git a/rs/moq-studio/src/ui.rs b/rs/moq-studio/src/ui.rs index 038c0ed..17d7a61 100644 --- a/rs/moq-studio/src/ui.rs +++ b/rs/moq-studio/src/ui.rs @@ -1210,6 +1210,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| { @@ -1286,6 +1287,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()); } @@ -1306,6 +1315,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 aede51b..8758b5c 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, @@ -24,12 +26,23 @@ struct Studio { device: wgpu::Device, queue: wgpu::Queue, bus: moq_audio::bus::Bus, + 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()) @@ -59,13 +72,14 @@ impl Studio { let (origin, driver) = moq_net::origin::Producer::new(Default::default()); tokio::spawn(moq_net::time::run(driver)); let bus = 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 { bus: bus.clone(), origin: origin.consume(), 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()); @@ -80,8 +94,18 @@ impl Studio { device, queue, bus, + 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()); } } @@ -157,6 +181,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) } @@ -238,8 +268,9 @@ 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 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() { @@ -266,10 +297,99 @@ 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. +#[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; + drop(studio); + + let mut studio = Studio::on(profile.path(), &[("page", page)]).await; + studio.until_center(GREEN, &[]).await; + }) + .await; +} + +/// A second studio on a profile in use fails its pages saying why, and +/// leaves the first studio's pages drawing. +#[tokio::test(flavor = "multi_thread")] +#[ignore = "needs CEF and a Vulkan device; run by just render-sw"] +async fn a_profile_in_use_fails_the_second_studio() { + local(async { + let profile = tempfile::tempdir().unwrap(); + let red = page(""); + let mut first = Studio::on(profile.path(), &[("page", red.clone())]).await; + first.until_center([255, 0, 0, 255], &[]).await; + + let second = Studio::on(profile.path(), &[("page", red)]).await; + let reason = second.until_failed("page").await; + assert!(reason.contains("another studio"), "{reason}"); + first.until_center([255, 0, 0, 255], &[]).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 { From a84759f608f6403ff0c61c5f65b79272016e4b74 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 3 Nov 2026 23:31:34 -0800 Subject: [PATCH 3/8] fix(scene): a deferred close follows its open, and a dropped helper stops A page closed past the backlog closed ahead of its queued Open, leaving an orphan page. The writer now closes a deferred page only once its Open is written, and drops an Open whose page already closed. The reader held a Helper clone, whose slot kept the instance alive, so dropping every owner never ended the socket. It now holds only the device and queue. Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/browser/import.rs | 19 +++++-- rs/moq-scene/src/browser/run.rs | 91 ++++++++++++++++++++++-------- 2 files changed, 84 insertions(+), 26 deletions(-) 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/run.rs b/rs/moq-scene/src/browser/run.rs index f4e45b8..2746848 100644 --- a/rs/moq-scene/src/browser/run.rs +++ b/rs/moq-scene/src/browser/run.rs @@ -199,10 +199,13 @@ impl Instance { let routes = Arc::new(Mutex::new(Routes::default())); let commands = writer(commands, routes.clone()).map_err(|err| failed("cannot start the page writer", err))?; let reader = { - let (helper, commands, routes) = (helper.clone(), commands.clone(), routes.clone()); + // 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 (commands, routes) = (commands.clone(), routes.clone()); std::thread::Builder::new() .name("browser-events".into()) - .spawn(move || read(&helper, process, &events, &commands, &routes)) + .spawn(move || read(&device, &queue, process, &events, &commands, &routes)) .map_err(|err| failed("cannot start the page reader", err))? }; Ok(Self { @@ -333,7 +336,23 @@ fn writer(channel: Channel, routes: Arc>) -> std::io::Result { + let mut routes = routes.lock().unwrap(); + // Closed while its Open waited: it never opens. + if let Some(at) = routes.closing.iter().position(|closing| closing == id) { + routes.closing.swap_remove(at); + continue; + } + open.insert(*id); + } + Command::Close(id) => _ = open.remove(id), + _ => {} + } if channel.send(&command, &[]).is_err() { break; } @@ -341,7 +360,15 @@ fn writer(channel: Channel, routes: Arc>) -> std::io::Result, Vec) = { let mut routes = routes.lock().unwrap(); let pages = routes.pages.values().map(|route| route.page.clone()).collect(); - (std::mem::take(&mut routes.closing), pages) + let mut closing = Vec::new(); + routes.closing.retain(|id| match open.remove(id) { + true => { + closing.push(*id); + false + } + false => true, + }); + (closing, pages) }; for id in closing { if channel.send(&Command::Close(id), &[]).is_err() { @@ -374,13 +401,14 @@ fn spawn(helper: &Helper) -> std::io::Result<(Process, Channel)> { /// Deliver the helper's events until it exits or breaks the protocol, then /// stop it and every page still open, saying why. fn read( - helper: &Helper, + device: &wgpu::Device, + queue: &wgpu::Queue, mut process: Process, events: &Channel, commands: &mpsc::SyncSender, routes: &Mutex, ) { - let reason = match deliver(helper, events, commands, routes) { + 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() { @@ -402,7 +430,8 @@ fn read( /// 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, routes: &Mutex, @@ -417,17 +446,13 @@ fn deliver( Err(err) => return Err(format!("the browser process's socket: {err}")), }; let (id, texture, paint) = match event { - Event::Texture(id, texture) => (id, super::import::texture(helper, &texture, fds), Paint::Gpu), + Event::Texture(id, texture) => (id, super::import::texture(device, queue, &texture, fds), Paint::Gpu), Event::Pixels { page, width, height, data, - } => ( - page, - super::upload(&helper.device, &helper.queue, width, height, &data), - Paint::Cpu, - ), + } => (page, super::upload(device, queue, width, height, &data), Paint::Cpu), Event::Audio(id, samples) => { routes.lock().unwrap().play(id, &samples); continue; @@ -492,31 +517,53 @@ mod tests { drop(theirs); } - /// Each write makes room for an owed release and a page closed past the - /// backlog, so both reach the helper once it reads again, with nothing - /// newer to carry them. + /// 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_and_closes() { + fn the_writer_delivers_owed_releases() { let (ours, theirs) = UnixStream::pair().unwrap(); - let (page, commands, routes) = routed(ours); + let (page, commands, _routes) = routed(ours); let release = Input::Key { code: 65, down: false, modifiers: Modifiers::default(), }; page.state.lock().unwrap().owed.push_back(release.clone()); - routes.lock().unwrap().closing.push(7); commands.send(Command::Painted).unwrap(); let helper = Channel::new(theirs); assert!(matches!(helper.recv::().unwrap(), Some((Command::Painted, _)))); - assert!(matches!( - helper.recv::().unwrap(), - Some((Command::Close(7), _)) - )); let Some((Command::Input(0, input), _)) = helper.recv().unwrap() else { panic!("the owed release never came"); }; assert_eq!(input, release); } + + /// A page closed past the backlog closes once the next write makes room, + /// and never ahead of its Open: one closed while its Open waited never + /// opens. + #[test] + fn a_deferred_close_follows_its_open() { + let (ours, theirs) = UnixStream::pair().unwrap(); + let routes = Arc::new(Mutex::new(Routes::default())); + let commands = writer(Channel::new(ours), routes.clone()).unwrap(); + let helper = Channel::new(theirs); + + routes.lock().unwrap().closing.push(2); + commands.send(Command::Open(1, Open::default())).unwrap(); + commands.send(Command::Open(2, Open::default())).unwrap(); + assert!(matches!( + helper.recv::().unwrap(), + Some((Command::Open(1, _), _)) + )); + + routes.lock().unwrap().closing.push(1); + commands.send(Command::Painted).unwrap(); + assert!(matches!(helper.recv::().unwrap(), Some((Command::Painted, _)))); + assert!(matches!( + helper.recv::().unwrap(), + Some((Command::Close(1), _)) + )); + assert!(routes.lock().unwrap().closing.is_empty()); + } } From 835eec2bddd51b775efda9907cd3505814108547 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 03:24:57 -0800 Subject: [PATCH 4/8] fix(scene): a page that closes itself fails, and opening never waits A page that called window.close() left its source waiting on a helper that stays up for the other pages. The helper now reports every page CEF closes, which the studio ignores for one it closed itself. Opening a page waited on a command queue that input from a hung helper could fill, so removing a starting source could hang its thread. Only input now counts against the backlog; opens, closes, frame rates, and paint answers always queue, in order, which also retires the deferred close bookkeeping. Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/browser/host.rs | 9 +- rs/moq-scene/src/browser/mod.rs | 59 ++++----- rs/moq-scene/src/browser/run.rs | 214 ++++++++++++++++--------------- rs/moq-studio/tests/browser.rs | 16 +++ 4 files changed, 164 insertions(+), 134 deletions(-) diff --git a/rs/moq-scene/src/browser/host.rs b/rs/moq-scene/src/browser/host.rs index 8f8fdd6..275a155 100644 --- a/rs/moq-scene/src/browser/host.rs +++ b/rs/moq-scene/src/browser/host.rs @@ -212,7 +212,7 @@ fn open(shared: Arc, id: Id, open: Open) { HelperAudio::new(paged.clone()), HelperLoad::new(paged.clone()), HelperRequest::new(paged.clone()), - HelperLife::new(id), + HelperLife::new(paged.clone()), ); match browser_host_create_browser_sync( Some(&window), @@ -644,7 +644,7 @@ wrap_request_handler! { wrap_life_span_handler! { struct HelperLife { - id: Id, + page: Arc, } impl LifeSpanHandler { @@ -671,8 +671,11 @@ wrap_life_span_handler! { } fn on_before_close(&self, _browser: Option<&mut Browser>) { + // 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.id); + pages.remove(&self.page.id); pages.is_empty() }); if last && QUITTING.get() { diff --git a/rs/moq-scene/src/browser/mod.rs b/rs/moq-scene/src/browser/mod.rs index 97fa646..a091eed 100644 --- a/rs/moq-scene/src/browser/mod.rs +++ b/rs/moq-scene/src/browser/mod.rs @@ -156,17 +156,16 @@ pub struct Page { #[derive(Default)] struct PageState { - /// The queue to the helper's socket, which [`run::writer`] drains, while - /// the page is open. + /// The queue to the helper's socket, while the page is open. #[cfg(target_os = "linux")] - commands: Option>, + commands: Option, /// The page's number in the helper. #[cfg(target_os = "linux")] id: protocol::Id, - /// Whether input is being dropped because the queue is full. + /// Whether input is being dropped because the backlog is full. #[cfg(target_os = "linux")] backlogged: bool, - /// Releases that did not fit the queue, oldest first, at most one per + /// Releases that did not fit the backlog, oldest first, at most one per /// button, key, focus, and pointer, sent ahead of any newer input. #[cfg(target_os = "linux")] owed: std::collections::VecDeque, @@ -185,19 +184,16 @@ impl PageState { /// 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. 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(self.id, input)) { + true => match commands.input(self.id, input) { Ok(()) => { self.backlogged = false; return; } - Err(TrySendError::Full(protocol::Command::Input(_, input))) => input, - Err(TrySendError::Full(_)) => unreachable!("only input is sent here"), - Err(TrySendError::Disconnected(_)) => return self.disconnect(), + Err(run::Refused::Full(input)) => input, + Err(run::Refused::Gone) => return self.disconnect(), }, // Newer input waits behind the owed releases. false => input, @@ -212,27 +208,23 @@ impl PageState { } /// Queue a new frame rate, then the owed releases, oldest first, while the - /// queue has room. + /// backlog has room. 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.id, self.framerate)) { - Ok(()) => self.retime = false, - Err(TrySendError::Full(_)) => return, - Err(TrySendError::Disconnected(_)) => return self.disconnect(), + if !commands.send(protocol::Command::FrameRate(self.id, self.framerate)) { + return self.disconnect(); } + self.retime = false; } while let Some(input) = self.owed.pop_front() { - match commands.try_send(protocol::Command::Input(self.id, input)) { + match commands.input(self.id, input) { Ok(()) => {} - Err(TrySendError::Full(protocol::Command::Input(_, input))) => { + Err(run::Refused::Full(input)) => { self.owed.push_front(input); return; } - Err(TrySendError::Full(_)) => unreachable!("only input is sent here"), - Err(TrySendError::Disconnected(_)) => return self.disconnect(), + Err(run::Refused::Gone) => return self.disconnect(), } } } @@ -444,9 +436,9 @@ mod tests { /// 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 (commands, queue) = run::Commands::new(); let page = Page::default(); - page.state.lock().unwrap().commands = Some(sender); + page.state.lock().unwrap().commands = Some(commands.clone()); page.send(button(true)); while !page.state.lock().unwrap().backlogged { page.send(Input::Text("x".into())); @@ -463,7 +455,11 @@ mod tests { // 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() { + while let Ok(command) = queue.try_recv() { + commands.wrote(&command); + let protocol::Command::Input(_, input) = command else { + panic!("only input is queued"); + }; received.push(input); page.state.lock().unwrap().flush(); } @@ -476,18 +472,19 @@ mod tests { /// A new canvas frame rate reaches the page once, even past the backlog. #[test] fn a_new_frame_rate_reaches_the_page() { - let (sender, queue) = std::sync::mpsc::sync_channel(1); + let (commands, queue) = run::Commands::new(); let page = Page::default(); - page.state.lock().unwrap().commands = Some(sender); + page.state.lock().unwrap().commands = Some(commands); 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"); - 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)))); + let last = queue.try_iter().last(); + assert!(matches!(last, Some(protocol::Command::FrameRate(_, 60)))); } } diff --git a/rs/moq-scene/src/browser/run.rs b/rs/moq-scene/src/browser/run.rs index 2746848..3d97896 100644 --- a/rs/moq-scene/src/browser/run.rs +++ b/rs/moq-scene/src/browser/run.rs @@ -8,22 +8,81 @@ use std::os::unix::fs::DirBuilderExt; use std::os::unix::net::UnixStream; use std::os::unix::process::CommandExt; use std::process::{Child, ExitStatus, Stdio}; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex, mpsc}; use std::thread::JoinHandle; use moq_audio::bus::Channel as Sound; use super::protocol::{Channel, Command, Event, Id, Open, Start}; -use super::{ARG, Helper, Page, Paint}; +use super::{ARG, Helper, Input, 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. +/// Input waiting for the helper to read it. A helper this far behind is hung, +/// and [`Page::send`] drops new input rather than wait for it. const BACKLOG: usize = 256; +/// The queue to the helper's socket, which [`writer`] drains. Only input +/// counts against the [`BACKLOG`], so a hung helper costs dropped input +/// rather than memory, while what keeps pages and paints in step (opens, +/// closes, frame rates, and answers to paints), few and needed, always +/// queues, in order, without waiting. +#[derive(Clone)] +pub(crate) struct Commands { + sender: mpsc::Sender, + /// Input queued and not yet written. + input: Arc, +} + +/// Why input was not queued. +pub(crate) enum Refused { + /// The backlog is full, and the input is handed back. + Full(Input), + /// The writer is gone, and so is the helper. + Gone, +} + +impl Commands { + /// A queue, and the end [`writer`] drains. + pub(crate) fn new() -> (Self, mpsc::Receiver) { + let (sender, queue) = mpsc::channel(); + let commands = Self { + sender, + input: Default::default(), + }; + (commands, queue) + } + + /// Queue `command`, which is not input; false once the writer is gone. + pub(crate) fn send(&self, command: Command) -> bool { + self.sender.send(command).is_ok() + } + + /// Queue input for page `id`, unless the backlog is full. + pub(crate) fn input(&self, id: Id, input: Input) -> Result<(), Refused> { + if self.input.fetch_add(1, Ordering::Relaxed) >= BACKLOG { + self.input.fetch_sub(1, Ordering::Relaxed); + return Err(Refused::Full(input)); + } + self.sender.send(Command::Input(id, input)).map_err(|_| Refused::Gone) + } + + /// Note that `command`, taken off the queue, is written. + fn written(input: &AtomicUsize, command: &Command) { + if matches!(command, Command::Input(..)) { + input.fetch_sub(1, Ordering::Relaxed); + } + } + + /// Note that `command` is written, as [`writer`] does. + #[cfg(test)] + pub(crate) fn wrote(&self, command: &Command) { + Self::written(&self.input, command); + } +} + /// 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. @@ -91,7 +150,7 @@ impl Helper { /// The helper every page runs in. pub(crate) struct Instance { /// Queued commands, which [`writer`] writes. - commands: mpsc::SyncSender, + commands: Commands, routes: Arc>, /// The next page's number. next: AtomicU64, @@ -106,9 +165,6 @@ pub(crate) struct Instance { #[derive(Default)] struct Routes { pages: HashMap, - /// Pages closed while the command queue was full, which [`writer`] closes - /// as the queue makes room. - closing: Vec, /// Why the helper stopped, once it has: no page opens in it after. ended: Option, } @@ -237,7 +293,7 @@ impl Instance { }; // A closed queue means the writer lost the socket, so the helper is // gone and the reader ends the page. - let _ = self.commands.send(Command::Open(id, open)); + self.commands.send(Command::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; @@ -276,12 +332,8 @@ struct Opened { impl Drop for Opened { fn drop(&mut self) { self.page.state.lock().unwrap().disconnect(); - let mut routes = self.instance.routes.lock().unwrap(); - routes.pages.remove(&self.id); - // Never waits on the helper: a full queue closes it as it drains. - if let Err(mpsc::TrySendError::Full(_)) = self.instance.commands.try_send(Command::Close(self.id)) { - routes.closing.push(self.id); - } + self.instance.routes.lock().unwrap().pages.remove(&self.id); + self.instance.commands.send(Command::Close(self.id)); } } @@ -325,56 +377,31 @@ impl Drop for Process { } } -/// Write the commands queued on the returned sender to `channel`, on a thread +/// Write the commands queued on the returned queue 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 pages -/// `routes` closed and the releases their pages owe. The thread ends once -/// every sender is dropped or a write fails, as one does when the helper -/// exits, and ends the socket on its way out. -fn writer(channel: Channel, routes: Arc>) -> std::io::Result> { - let (sender, queue) = mpsc::sync_channel(BACKLOG); +/// sender holding the studio's state. Each write makes room for the releases +/// `routes`' pages owe. The thread ends once every sender is dropped or a +/// write fails, as one does when the helper exits, and ends the socket on its +/// way out. +fn writer(channel: Channel, routes: Arc>) -> std::io::Result { + let (commands, queue) = Commands::new(); + let input = commands.input.clone(); std::thread::Builder::new() .name("browser-commands".into()) .spawn(move || { - // Pages whose Open is written and whose Close is not, since a page - // closed past the backlog closes only after it opens. - let mut open = std::collections::HashSet::new(); - 'write: for command in queue { - match &command { - Command::Open(id, _) => { - let mut routes = routes.lock().unwrap(); - // Closed while its Open waited: it never opens. - if let Some(at) = routes.closing.iter().position(|closing| closing == id) { - routes.closing.swap_remove(at); - continue; - } - open.insert(*id); - } - Command::Close(id) => _ = open.remove(id), - _ => {} - } + for command in queue { if channel.send(&command, &[]).is_err() { break; } - // Taken first, so a closing page never waits on a socket write. - let (closing, pages): (Vec, Vec) = { - let mut routes = routes.lock().unwrap(); - let pages = routes.pages.values().map(|route| route.page.clone()).collect(); - let mut closing = Vec::new(); - routes.closing.retain(|id| match open.remove(id) { - true => { - closing.push(*id); - false - } - false => true, - }); - (closing, pages) - }; - for id in closing { - if channel.send(&Command::Close(id), &[]).is_err() { - break 'write; - } - } + Commands::written(&input, &command); + // Taken first, so no page waits on a socket write. + let pages: Vec = routes + .lock() + .unwrap() + .pages + .values() + .map(|route| route.page.clone()) + .collect(); for page in pages { page.state.lock().unwrap().flush(); } @@ -383,7 +410,7 @@ fn writer(channel: Channel, routes: Arc>) -> std::io::Result, + commands: &Commands, routes: &Mutex, ) { let reason = match deliver(device, queue, events, commands, routes) { @@ -433,7 +460,7 @@ fn deliver( device: &wgpu::Device, queue: &wgpu::Queue, events: &Channel, - commands: &mpsc::SyncSender, + commands: &Commands, routes: &Mutex, ) -> Result<(), String> { use std::io::ErrorKind::*; @@ -464,11 +491,11 @@ fn deliver( Event::Fatal(reason) => return Err(reason), }; // The copy is done, so the helper may hand the source back to CEF, - // even for a page closed since. Waits behind queued input, never - // dropped: 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. - let _ = commands.send(Command::Painted); + // 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.send(Command::Painted); let mut routes = routes.lock().unwrap(); match texture { Ok(texture) => routes.show(id, texture, paint), @@ -484,7 +511,7 @@ mod tests { /// An open page, as [`Instance::open`] routes one, with a writer to /// `ours`. - fn routed(ours: UnixStream) -> (Page, mpsc::SyncSender, Arc>) { + fn routed(ours: UnixStream) -> (Page, Commands, Arc>) { let routes = Arc::new(Mutex::new(Routes::default())); let commands = writer(Channel::new(ours), routes.clone()).unwrap(); let page = Page::default(); @@ -504,17 +531,32 @@ mod tests { } /// 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. A page still opens + /// and closes past the backlog without waiting, in order, so removing a + /// starting page never hangs on the helper. #[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 (page, _commands, _routes) = routed(ours); + let (page, commands, _routes) = routed(ours); // Far more than the socket buffers and the backlog hold together. for _ in 0..4096 { page.send(Input::Text("x".repeat(1024))); } assert!(page.state.lock().unwrap().backlogged); - drop(theirs); + commands.send(Command::Open(1, Open::default())); + commands.send(Command::Close(1)); + + let helper = Channel::new(theirs); + let mut pages = Vec::new(); + while pages.len() < 2 { + match helper.recv::().unwrap() { + Some((Command::Open(id, _), _)) => pages.push(("open", id)), + Some((Command::Close(id), _)) => pages.push(("close", id)), + Some(_) => {} + None => panic!("the writer stopped"), + } + } + assert_eq!(pages, [("open", 1), ("close", 1)]); } /// Each write makes room for an owed release, so it reaches the page once @@ -529,7 +571,7 @@ mod tests { modifiers: Modifiers::default(), }; page.state.lock().unwrap().owed.push_back(release.clone()); - commands.send(Command::Painted).unwrap(); + commands.send(Command::Painted); let helper = Channel::new(theirs); assert!(matches!(helper.recv::().unwrap(), Some((Command::Painted, _)))); @@ -538,32 +580,4 @@ mod tests { }; assert_eq!(input, release); } - - /// A page closed past the backlog closes once the next write makes room, - /// and never ahead of its Open: one closed while its Open waited never - /// opens. - #[test] - fn a_deferred_close_follows_its_open() { - let (ours, theirs) = UnixStream::pair().unwrap(); - let routes = Arc::new(Mutex::new(Routes::default())); - let commands = writer(Channel::new(ours), routes.clone()).unwrap(); - let helper = Channel::new(theirs); - - routes.lock().unwrap().closing.push(2); - commands.send(Command::Open(1, Open::default())).unwrap(); - commands.send(Command::Open(2, Open::default())).unwrap(); - assert!(matches!( - helper.recv::().unwrap(), - Some((Command::Open(1, _), _)) - )); - - routes.lock().unwrap().closing.push(1); - commands.send(Command::Painted).unwrap(); - assert!(matches!(helper.recv::().unwrap(), Some((Command::Painted, _)))); - assert!(matches!( - helper.recv::().unwrap(), - Some((Command::Close(1), _)) - )); - assert!(routes.lock().unwrap().closing.is_empty()); - } } diff --git a/rs/moq-studio/tests/browser.rs b/rs/moq-studio/tests/browser.rs index a3c7bb8..661e037 100644 --- a/rs/moq-studio/tests/browser.rs +++ b/rs/moq-studio/tests/browser.rs @@ -269,6 +269,22 @@ async fn a_click_reaches_the_page() { .await; } +/// 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. From ea8f5dcb9ce07bfd30dfcada2a3123ca8e69e987 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 06:15:44 -0800 Subject: [PATCH 5/8] fix(scene): coalesce each page's pending commands, so reloads cannot pile up While the helper read nothing, every Reload queued another Open and Close, URL included, which the helper replayed once it recovered. What the studio has yet to tell the helper now waits in an outbox that coalesces each page's open, frame rate, and close: a page closed before its open is written never opens and drops its pending input, so what waits scales with the pages open. Input stays bounded by the backlog. The browser profile is also made owner-only when it already exists, and the macOS and Windows quest states the shared helper's crash scope. Co-Authored-By: Claude Opus 5.5 --- quest/m0/browser-source.md | 3 +- rs/moq-scene/src/browser/mod.rs | 58 +++++--- rs/moq-scene/src/browser/outbox.rs | 216 +++++++++++++++++++++++++++++ rs/moq-scene/src/browser/run.rs | 136 ++++++------------ 4 files changed, 296 insertions(+), 117 deletions(-) create mode 100644 rs/moq-scene/src/browser/outbox.rs diff --git a/quest/m0/browser-source.md b/quest/m0/browser-source.md index 39250d5..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 diff --git a/rs/moq-scene/src/browser/mod.rs b/rs/moq-scene/src/browser/mod.rs index a091eed..fc8f4c5 100644 --- a/rs/moq-scene/src/browser/mod.rs +++ b/rs/moq-scene/src/browser/mod.rs @@ -25,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")] @@ -158,7 +160,7 @@ pub struct Page { struct PageState { /// The queue to the helper's socket, while the page is open. #[cfg(target_os = "linux")] - commands: Option, + commands: Option, /// The page's number in the helper. #[cfg(target_os = "linux")] id: protocol::Id, @@ -192,8 +194,8 @@ impl PageState { self.backlogged = false; return; } - Err(run::Refused::Full(input)) => input, - Err(run::Refused::Gone) => return self.disconnect(), + Err(outbox::Refused::Full(input)) => input, + Err(outbox::Refused::Gone) => return self.disconnect(), }, // Newer input waits behind the owed releases. false => input, @@ -212,7 +214,7 @@ impl PageState { fn flush(&mut self) { let Some(commands) = &self.commands else { return }; if self.retime { - if !commands.send(protocol::Command::FrameRate(self.id, self.framerate)) { + if !commands.rate(self.id, self.framerate) { return self.disconnect(); } self.retime = false; @@ -220,11 +222,11 @@ impl PageState { while let Some(input) = self.owed.pop_front() { match commands.input(self.id, input) { Ok(()) => {} - Err(run::Refused::Full(input)) => { + Err(outbox::Refused::Full(input)) => { self.owed.push_front(input); return; } - Err(run::Refused::Gone) => return self.disconnect(), + Err(outbox::Refused::Gone) => return self.disconnect(), } } } @@ -436,7 +438,7 @@ mod tests { /// page never keeps holding what the studio let go of. #[test] fn a_release_past_the_backlog_is_owed() { - let (commands, queue) = run::Commands::new(); + let commands = outbox::Commands::default(); let page = Page::default(); page.state.lock().unwrap().commands = Some(commands.clone()); page.send(button(true)); @@ -453,14 +455,19 @@ mod tests { }); assert_eq!(page.state.lock().unwrap().owed, [button(false)]); - // The helper catches up, and each write makes room, as the writer's do. + // The helper catches up, and each batch makes room, as the writer's do. let mut received = Vec::new(); - while let Ok(command) = queue.try_recv() { - commands.wrote(&command); - let protocol::Command::Input(_, input) = command else { - panic!("only input is queued"); - }; - received.push(input); + loop { + let batch = commands.take(); + if batch.is_empty() { + break; + } + for command in batch { + let protocol::Command::Input(_, input) = command else { + panic!("only input is queued"); + }; + received.push(input); + } page.state.lock().unwrap().flush(); } assert_eq!(received.first(), Some(&button(true))); @@ -469,22 +476,31 @@ mod tests { 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 (commands, queue) = run::Commands::new(); + let commands = outbox::Commands::default(); let page = Page::default(); - page.state.lock().unwrap().commands = Some(commands); + 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"); while !page.state.lock().unwrap().backlogged { page.send(Input::Text("x".into())); } page.set_framerate(60); - let last = queue.try_iter().last(); - assert!(matches!(last, Some(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..d2a04c6 --- /dev/null +++ b/rs/moq-scene/src/browser/outbox.rs @@ -0,0 +1,216 @@ +//! What the studio has yet to tell the helper, which the writer takes in +//! batches. +//! +//! 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 helper reads nothing. +//! A page closed before its open was written never opens. + +use std::collections::{HashMap, HashSet, VecDeque}; +use std::sync::{Arc, Condvar, Mutex}; + +use super::Input; +use super::protocol::{Command, Id, Open}; + +/// 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, and the input is handed back. + Full(Input), + /// The writer stopped, and so did the helper. + Gone, +} + +#[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)>, + /// 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.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`, unless the backlog is full. + 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.input.len() >= BACKLOG { + return Err(Refused::Full(input)); + } + outbox.input.push_back((id, input)); + 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(); + } + } + + /// The commands waiting, without waiting for any. + #[cfg(test)] + 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(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 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/run.rs b/rs/moq-scene/src/browser/run.rs index 3d97896..bba8e6d 100644 --- a/rs/moq-scene/src/browser/run.rs +++ b/rs/moq-scene/src/browser/run.rs @@ -4,85 +4,23 @@ use std::collections::HashMap; use std::os::fd::OwnedFd; -use std::os::unix::fs::DirBuilderExt; +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::atomic::{AtomicU64, AtomicUsize, Ordering}; -use std::sync::{Arc, Mutex, 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::outbox::Commands; use super::protocol::{Channel, Command, Event, Id, Open, Start}; -use super::{ARG, Helper, Input, Page, Paint}; +use super::{ARG, Helper, Page, Paint}; use crate::Error; use crate::source::{Mailbox, Picture, Status}; use crate::wgpu; -/// Input waiting for the helper to read it. A helper this far behind is hung, -/// and [`Page::send`] drops new input rather than wait for it. -const BACKLOG: usize = 256; - -/// The queue to the helper's socket, which [`writer`] drains. Only input -/// counts against the [`BACKLOG`], so a hung helper costs dropped input -/// rather than memory, while what keeps pages and paints in step (opens, -/// closes, frame rates, and answers to paints), few and needed, always -/// queues, in order, without waiting. -#[derive(Clone)] -pub(crate) struct Commands { - sender: mpsc::Sender, - /// Input queued and not yet written. - input: Arc, -} - -/// Why input was not queued. -pub(crate) enum Refused { - /// The backlog is full, and the input is handed back. - Full(Input), - /// The writer is gone, and so is the helper. - Gone, -} - -impl Commands { - /// A queue, and the end [`writer`] drains. - pub(crate) fn new() -> (Self, mpsc::Receiver) { - let (sender, queue) = mpsc::channel(); - let commands = Self { - sender, - input: Default::default(), - }; - (commands, queue) - } - - /// Queue `command`, which is not input; false once the writer is gone. - pub(crate) fn send(&self, command: Command) -> bool { - self.sender.send(command).is_ok() - } - - /// Queue input for page `id`, unless the backlog is full. - pub(crate) fn input(&self, id: Id, input: Input) -> Result<(), Refused> { - if self.input.fetch_add(1, Ordering::Relaxed) >= BACKLOG { - self.input.fetch_sub(1, Ordering::Relaxed); - return Err(Refused::Full(input)); - } - self.sender.send(Command::Input(id, input)).map_err(|_| Refused::Gone) - } - - /// Note that `command`, taken off the queue, is written. - fn written(input: &AtomicUsize, command: &Command) { - if matches!(command, Command::Input(..)) { - input.fetch_sub(1, Ordering::Relaxed); - } - } - - /// Note that `command` is written, as [`writer`] does. - #[cfg(test)] - pub(crate) fn wrote(&self, command: &Command) { - Self::written(&self.input, command); - } -} - /// 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. @@ -149,7 +87,7 @@ impl Helper { /// The helper every page runs in. pub(crate) struct Instance { - /// Queued commands, which [`writer`] writes. + /// What waits for [`writer`] to write it. commands: Commands, routes: Arc>, /// The next page's number. @@ -226,11 +164,13 @@ impl Instance { 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. + // 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), @@ -293,7 +233,7 @@ impl Instance { }; // A closed queue means the writer lost the socket, so the helper is // gone and the reader ends the page. - self.commands.send(Command::Open(id, open)); + 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; @@ -333,7 +273,7 @@ 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.send(Command::Close(self.id)); + self.instance.commands.close(self.id); } } @@ -377,23 +317,24 @@ impl Drop for Process { } } -/// Write the commands queued on the returned queue 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 -/// `routes`' pages owe. The thread ends once every sender is dropped or a -/// write fails, as one does when the helper exits, and ends the socket on its -/// way out. +/// sender holding the studio's state. Each batch written makes room for the +/// releases `routes`' pages owe. 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, routes: Arc>) -> std::io::Result { - let (commands, queue) = Commands::new(); - let input = commands.input.clone(); + 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; + 'write: while let Some(batch) = outbox.next() { + for command in batch { + if channel.send(&command, &[]).is_err() { + break 'write; + } } - Commands::written(&input, &command); // Taken first, so no page waits on a socket write. let pages: Vec = routes .lock() @@ -408,6 +349,7 @@ fn writer(channel: Channel, routes: Arc>) -> std::io::Result routes.show(id, texture, paint), @@ -531,32 +475,34 @@ mod tests { } /// A helper that stops reading costs dropped input, never a sender stuck - /// on a full socket while it holds the studio's state. A page still opens - /// and closes past the backlog without waiting, in order, so removing a - /// starting page never hangs on the helper. + /// 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_blocks_nothing() { let (ours, theirs) = UnixStream::pair().unwrap(); let (page, commands, _routes) = routed(ours); - // Far more than the socket buffers and the backlog hold together. + // Far more than the socket buffers and the backlog hold together, so + // the writer is stuck on the socket. for _ in 0..4096 { page.send(Input::Text("x".repeat(1024))); } assert!(page.state.lock().unwrap().backlogged); - commands.send(Command::Open(1, Open::default())); - commands.send(Command::Close(1)); + for id in 1..100 { + commands.open(id, Open::default()); + commands.close(id); + } + commands.open(100, Open::default()); let helper = Channel::new(theirs); - let mut pages = Vec::new(); - while pages.len() < 2 { + loop { match helper.recv::().unwrap() { - Some((Command::Open(id, _), _)) => pages.push(("open", id)), - Some((Command::Close(id), _)) => pages.push(("close", id)), + Some((Command::Open(100, _), _)) => break, + Some((Command::Open(id, _) | Command::Close(id), _)) => panic!("page {id} was reloaded away"), Some(_) => {} None => panic!("the writer stopped"), } } - assert_eq!(pages, [("open", 1), ("close", 1)]); } /// Each write makes room for an owed release, so it reaches the page once @@ -571,7 +517,7 @@ mod tests { modifiers: Modifiers::default(), }; page.state.lock().unwrap().owed.push_back(release.clone()); - commands.send(Command::Painted); + commands.painted(); let helper = Channel::new(theirs); assert!(matches!(helper.recv::().unwrap(), Some((Command::Painted, _)))); From e3e8a55b1cdd5d28007da002653162b8c36049da Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 06:25:49 -0800 Subject: [PATCH 6/8] fix(scene): the helper coalesces what CEF's UI thread has yet to apply The helper posted a UI task per command, so a stalled UI thread held one task, and one URL, per reload, and replayed pages the studio had closed. It now keeps the same coalescing outbox the studio does, with one task at a time posted to apply it. Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/browser/host.rs | 48 +++++++++++++++++++++++++----- rs/moq-scene/src/browser/outbox.rs | 11 +++---- 2 files changed, 46 insertions(+), 13 deletions(-) diff --git a/rs/moq-scene/src/browser/host.rs b/rs/moq-scene/src/browser/host.rs index 275a155..4d85c8e 100644 --- a/rs/moq-scene/src/browser/host.rs +++ b/rs/moq-scene/src/browser/host.rs @@ -7,10 +7,12 @@ 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::outbox::Commands; use super::protocol::{Channel, Command, Event, Format, Id, Open, Plane, Start, Texture}; use super::{Button, Input, Modifiers}; @@ -163,25 +165,42 @@ fn browse(shared: &Arc, start: Start, reader: Channel, painted_tx: mpsc: /// 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), _))) => { - let shared = shared.clone(); - post(move || self::open(shared, id, open)); - } - Ok(Some((Command::Close(id), _))) => post(move || close(id)), - Ok(Some((Command::Input(id, input), _))) => post(move || deliver(id, input)), - Ok(Some((Command::FrameRate(id, rate), _))) => post(move || retime(id, rate)), + 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(|| { + post(move || { + apply(&shared, pending.take()); QUITTING.set(true); let ids: Vec = PAGES.with_borrow(|pages| pages.keys().copied().collect()); if ids.is_empty() { @@ -193,6 +212,19 @@ fn commands(shared: Arc, reader: Channel, painted: mpsc::Sender<()>) { }); } +/// 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 { diff --git a/rs/moq-scene/src/browser/outbox.rs b/rs/moq-scene/src/browser/outbox.rs index d2a04c6..b37be90 100644 --- a/rs/moq-scene/src/browser/outbox.rs +++ b/rs/moq-scene/src/browser/outbox.rs @@ -1,10 +1,11 @@ -//! What the studio has yet to tell the helper, which the writer takes in -//! batches. +//! 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 helper reads nothing. -//! A page closed before its open was written never opens. +//! never with how often they changed, even while the taker is stuck. A page +//! closed before its open is taken never opens. use std::collections::{HashMap, HashSet, VecDeque}; use std::sync::{Arc, Condvar, Mutex}; @@ -129,7 +130,7 @@ impl Commands { } /// The commands waiting, without waiting for any. - #[cfg(test)] + #[cfg(any(test, feature = "cef"))] pub(crate) fn take(&self) -> Vec { self.0.0.lock().unwrap().take() } From 52abb05c22a280bfd035b1981531c7706473d711 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 06:38:33 -0800 Subject: [PATCH 7/8] fix(scene): owe releases in the outbox, so neither queue drops one A release past the helper's own backlog dropped, though the studio had already sent it, leaving the page holding a key once its UI thread caught up. Owing releases now lives in the outbox both sides share: past the backlog a release is owed, once per page and button, key, focus, and pointer, and taken after the input before it. The studio's page state and writer no longer track owed releases. Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/browser/mod.rs | 125 +++-------------------------- rs/moq-scene/src/browser/outbox.rs | 93 +++++++++++++++++++-- rs/moq-scene/src/browser/run.rs | 71 +++------------- 3 files changed, 103 insertions(+), 186 deletions(-) diff --git a/rs/moq-scene/src/browser/mod.rs b/rs/moq-scene/src/browser/mod.rs index fc8f4c5..b9b4cf4 100644 --- a/rs/moq-scene/src/browser/mod.rs +++ b/rs/moq-scene/src/browser/mod.rs @@ -167,10 +167,6 @@ struct PageState { /// Whether input is being dropped because the backlog is full. #[cfg(target_os = "linux")] backlogged: bool, - /// Releases that did not fit the backlog, oldest first, at most one per - /// button, key, focus, and pointer, sent ahead of any newer input. - #[cfg(target_os = "linux")] - owed: std::collections::VecDeque, /// The canvas's frame rate, which the page paints at. #[cfg(target_os = "linux")] framerate: u32, @@ -182,35 +178,22 @@ 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) { self.flush(); let Some(commands) = &self.commands else { return }; - let input = match self.owed.is_empty() { - true => match commands.input(self.id, 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(outbox::Refused::Full(input)) => input, - Err(outbox::Refused::Gone) => 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 - /// backlog has room. + /// Queue a new frame rate. fn flush(&mut self) { let Some(commands) = &self.commands else { return }; if self.retime { @@ -219,47 +202,11 @@ impl PageState { } self.retime = false; } - while let Some(input) = self.owed.pop_front() { - match commands.input(self.id, input) { - Ok(()) => {} - Err(outbox::Refused::Full(input)) => { - self.owed.push_front(input); - return; - } - Err(outbox::Refused::Gone) => return self.disconnect(), - } - } } /// 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, - } } } @@ -422,60 +369,6 @@ 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 commands = outbox::Commands::default(); - let page = Page::default(); - page.state.lock().unwrap().commands = Some(commands.clone()); - 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 batch makes room, as the writer's do. - let mut received = Vec::new(); - loop { - let batch = commands.take(); - if batch.is_empty() { - break; - } - for command in batch { - let protocol::Command::Input(_, input) = command else { - panic!("only input is queued"); - }; - 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, the newest of those /// waiting, even past the backlog. #[test] diff --git a/rs/moq-scene/src/browser/outbox.rs b/rs/moq-scene/src/browser/outbox.rs index b37be90..f48c3c7 100644 --- a/rs/moq-scene/src/browser/outbox.rs +++ b/rs/moq-scene/src/browser/outbox.rs @@ -5,13 +5,15 @@ //! 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. +//! 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::Input; 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 @@ -25,12 +27,35 @@ pub(crate) struct Commands(Arc<(Mutex, Condvar)>); /// Why input was not queued. pub(crate) enum Refused { - /// The backlog is full, and the input is handed back. - Full(Input), + /// 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. @@ -41,6 +66,9 @@ struct Outbox { 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. @@ -72,6 +100,7 @@ impl Commands { 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), @@ -91,17 +120,24 @@ impl Commands { self.push(|outbox| outbox.painted += 1); } - /// Queue input for page `id`, unless the backlog is full. + /// 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.input.len() >= BACKLOG { - return Err(Refused::Full(input)); + 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); } - outbox.input.push_back((id, input)); ready.notify_one(); Ok(()) } @@ -146,6 +182,7 @@ impl Outbox { } 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); @@ -159,6 +196,46 @@ impl Outbox { 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(), diff --git a/rs/moq-scene/src/browser/run.rs b/rs/moq-scene/src/browser/run.rs index bba8e6d..11b9d76 100644 --- a/rs/moq-scene/src/browser/run.rs +++ b/rs/moq-scene/src/browser/run.rs @@ -193,7 +193,7 @@ impl Instance { .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, routes.clone()).map_err(|err| failed("cannot start the page writer", err))?; + 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. @@ -319,11 +319,10 @@ impl Drop for Process { /// 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 batch written makes room for the -/// releases `routes`' pages owe. 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, routes: Arc>) -> std::io::Result { +/// 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() @@ -335,17 +334,6 @@ fn writer(channel: Channel, routes: Arc>) -> std::io::Result = routes - .lock() - .unwrap() - .pages - .values() - .map(|route| route.page.clone()) - .collect(); - for page in pages { - page.state.lock().unwrap().flush(); - } } // The helper exits once it reads the end, so one the studio can no // longer reach never keeps running unheard. @@ -451,28 +439,7 @@ fn deliver( #[cfg(test)] mod tests { use super::*; - use crate::browser::{Input, Modifiers}; - - /// An open page, as [`Instance::open`] routes one, with a writer to - /// `ours`. - fn routed(ours: UnixStream) -> (Page, Commands, Arc>) { - let routes = Arc::new(Mutex::new(Routes::default())); - let commands = writer(Channel::new(ours), routes.clone()).unwrap(); - let page = Page::default(); - page.state.lock().unwrap().commands = Some(commands.clone()); - routes.lock().unwrap().pages.insert( - 0, - Route { - page: page.clone(), - picture: Default::default(), - sound: None, - status: kio::Producer::new(Status::Starting), - live: false, - end: kio::Producer::new(None), - }, - ); - (page, commands, routes) - } + 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. Pages still open @@ -481,7 +448,9 @@ mod tests { #[test] fn a_helper_that_stops_reading_blocks_nothing() { let (ours, theirs) = UnixStream::pair().unwrap(); - let (page, commands, _routes) = routed(ours); + let commands = writer(Channel::new(ours)).unwrap(); + let page = Page::default(); + page.state.lock().unwrap().commands = Some(commands.clone()); // Far more than the socket buffers and the backlog hold together, so // the writer is stuck on the socket. for _ in 0..4096 { @@ -504,26 +473,4 @@ mod tests { } } } - - /// 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, commands, _routes) = routed(ours); - let release = Input::Key { - code: 65, - down: false, - modifiers: Modifiers::default(), - }; - page.state.lock().unwrap().owed.push_back(release.clone()); - commands.painted(); - - let helper = Channel::new(theirs); - assert!(matches!(helper.recv::().unwrap(), Some((Command::Painted, _)))); - let Some((Command::Input(0, input), _)) = helper.recv().unwrap() else { - panic!("the owed release never came"); - }; - assert_eq!(input, release); - } } From a3f6458e5d97dfeabbd25547f6dc7ef9f0e6e24a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 4 Nov 2026 08:17:34 -0800 Subject: [PATCH 8/8] fix(scene): stop the writer when the reader cannot start, and pin the stalled-writer test A reader thread that failed to spawn left the writer waiting forever in the outbox, holding its thread and socket; that path now stops it. The stalled-helper test assumed the input flood had blocked the writer, which the backlog does not prove. It now has the writer take an open larger than the socket buffers first, so the writer is stuck before the reloads are queued. Co-Authored-By: Claude Opus 5.5 --- rs/moq-scene/src/browser/outbox.rs | 12 +++++++++++ rs/moq-scene/src/browser/run.rs | 34 +++++++++++++++++++++++------- 2 files changed, 38 insertions(+), 8 deletions(-) diff --git a/rs/moq-scene/src/browser/outbox.rs b/rs/moq-scene/src/browser/outbox.rs index f48c3c7..519d5df 100644 --- a/rs/moq-scene/src/browser/outbox.rs +++ b/rs/moq-scene/src/browser/outbox.rs @@ -165,6 +165,18 @@ impl Commands { } } + /// 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 { diff --git a/rs/moq-scene/src/browser/run.rs b/rs/moq-scene/src/browser/run.rs index 11b9d76..2a1fd4c 100644 --- a/rs/moq-scene/src/browser/run.rs +++ b/rs/moq-scene/src/browser/run.rs @@ -198,11 +198,16 @@ impl Instance { // 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 (commands, routes) = (commands.clone(), routes.clone()); - std::thread::Builder::new() + let (reading, routes) = (commands.clone(), routes.clone()); + let spawned = std::thread::Builder::new() .name("browser-events".into()) - .spawn(move || read(&device, &queue, process, &events, &commands, &routes)) - .map_err(|err| failed("cannot start the page reader", err))? + .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, @@ -451,10 +456,19 @@ mod tests { let commands = writer(Channel::new(ours)).unwrap(); let page = Page::default(); page.state.lock().unwrap().commands = Some(commands.clone()); - // Far more than the socket buffers and the backlog hold together, so - // the writer is stuck on the socket. - for _ in 0..4096 { - page.send(Input::Text("x".repeat(1024))); + // 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(); + } + + for _ in 0..1024 { + page.send(Input::Text("x".into())); } assert!(page.state.lock().unwrap().backlogged); for id in 1..100 { @@ -464,6 +478,10 @@ mod tests { commands.open(100, Open::default()); let helper = Channel::new(theirs); + assert!(matches!( + helper.recv::().unwrap(), + Some((Command::Open(0, _), _)) + )); loop { match helper.recv::().unwrap() { Some((Command::Open(100, _), _)) => break,