From f889516ef4ca26454de83a1299cbd7b16d2e79cd Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Sat, 17 Oct 2026 13:36:44 +0000 Subject: [PATCH] feat: window streams in the box (nescope, nescapture, neshub) (#423) The box half of phase 1 of window streams: two games in one box, each its own stream. Wire formats are nesprotocol's `window` module (#421). ## What each app does now **nescope** - Gives every toplevel an id on first map: 0 for the first window, never reused, none past 65535. - Tells the hub, on the existing input socket, when a window opens, changes title or size, and closes, in nesprotocol's window encoding. It also sends a box-internal `MSG_WINDOW_PID` saying which process owns the window: the Wayland client, or `_NET_WM_PID` for Xwayland. The launch id comes from that process's `NESTRI_LAUNCH_ID`. - Handles `INPUT_WINDOW_FOCUS` by raising and focusing that window, and `INPUT_WINDOW_OUTPUT_SIZE` per window. The old `INPUT_OUTPUT_SIZE` means window 0. There is still one output: a window's size applies while that window has focus and is remembered otherwise. On every input connection, input resets to window 0. - Before cursor updates, sends `MSG_CURSOR_WINDOW` when the window they describe changes. **nescapture** - Tags each video IPC frame with its pid in a trailer (`IPC_FLAG_PID` in the reserved byte). An older reader never sees it. - Listens on `/tmp/nescapture-cmd-.sock`. It still binds the shared `/tmp/nescapture-cmd.sock` and relays from it, so whatever sends there still reaches the newest encoder, as before. The shared socket was the only resource two capturing processes fought over: the second one took it, and commands went only to that one. The video and stats sockets are send-only datagram sockets and are safe to share. Capture slots and devices are per process. **neshub** - Joins frames to windows by pid. An untagged frame is window 0. A pid with no window falls back to the newest window of its launch, then to window 0 while no other window is open. A pid whose window closed belongs to nobody. - A connection that never subscribes gets the box's primary picture in the old encoding. The primary picture is the frames of the first encoder to draw, until it has been silent for a second. A subscribed connection gets only its window's frames, windowed for any window but 0, with per-window seq. PyroWave stays window 0 only. - `MSG_WINDOW_SUBSCRIBE` switches the connection first, then resets the writer's seq and resync gate in line with the frames, then requests a keyframe. The first subscribe opens `STREAM_WINDOWS` on the video connection, starting with `OPENED` (and the icon, if known) for every open window. Later lifecycle events go to every subscribed connection. - Input to nescope carries `INPUT_WINDOW_FOCUS` only when the window changes, and the hub does no merging across tags. A gamepad message moves focus only when its window differs from the current one and no other client sent keyboard or pointer input in the last second. Focus tags move keyboard and pointer focus only: nescope neither raises the window nor resizes the output for one. The output changes size only on an explicit `INPUT_WINDOW_OUTPUT_SIZE` from the window that owns it, last writer wins with 500 ms hysteresis between windows. A client's `INPUT_OUTPUT_SIZE` is rewritten to `INPUT_WINDOW_OUTPUT_SIZE`, except for window 0. - A cursor update goes to the clients subscribed to the window it describes. Clients that never subscribed get every update, as before. - When nescope reconnects (a new compositor), the hub sends `CLOSED` for every window and every subscription goes dead until the client subscribes again. - `DatagramSender` sizes fragments with `max_fragment_payload` and writes headers with `write_video_datagram_header`. - Commands go to every known encoder on its own socket, or to the shared socket while none is known (as today). An encoder that appears later is sent the commands so far (the latest of each message type) and then a keyframe request. The first encoder is not sent them again. - A keyframe on a client's loss or subscribe goes only to that client's window's encoder (the primary's, for a client that never subscribed). - Known limit: bitrate is box-wide. One controller sets one target for every window's encoder, so one client's congestion sets every window's rate. Per-window rate is follow-up work. - An X11 unmap is `CLOSED` for the hub, and a re-map is `OPENED` with the same id. nescope no longer exits when its first game exits; nesinit owns box power-off. The hub never reads `/proc//environ` (it runs as another uid): nescapture sends its launch id. ## Shipping nescope and neshub ship together in the guest image and must not ship apart. ## Compatibility One game with an old client sends the same bytes as before. The client never subscribes, so it gets the primary picture, which is that game, as `DGRAM_VIDEO`/`MSG_DATA` with an unchanged seq. It gets no windows stream. Cursor updates reach it as before. Focus tags appear only if the game's window is not 0, and window 0 resizes use the old form. Mixed versions also work: - An old nescapture is window 0. - An old nescope never sends window messages. Every frame is then window 0, or the primary picture. - An old hub sends no tags, which nescope reads as window 0, and it still reaches nescapture through the shared socket. ## Tests `cargo test` on nesprotocol (122), nescope (17+5), nescapture (161), neshub (74), all passing. New tests: - The window table: joining pids to windows, the launch fallback, retired pids, the primary picture, the snapshot. - Focus tagging and resize rewriting. - The command fan-out. - nescope's id registry (first window 0, ids never reused, refused past 65535) and per-window focus and sizes. - The IPC pid trailer. - Two QUIC end-to-end tests: - Two tagged encoders: an unsubscribed client sees only the first game in the old encoding. After subscribing to window 1 it sees only the second game, windowed, with seq starting again, and a windows stream that starts with both `OPENED` events. - An untagged encoder gives the old stream. Not run on a box: no neslet, nestri-host or box was touched. --- apps/nescapture/src/encode.rs | 98 ++++- apps/nescope/src/handlers.rs | 33 +- apps/nescope/src/input.rs | 2 +- apps/nescope/src/main.rs | 38 +- apps/nescope/src/state.rs | 205 ++++++++++ apps/nescope/src/windows.rs | 440 +++++++++++++++++++++ apps/neshub/src/capture_cmd.rs | 235 +++++++++++ apps/neshub/src/dgram.rs | 84 ++-- apps/neshub/src/e2e_tests.rs | 181 ++++++++- apps/neshub/src/ipc_listener.rs | 85 +++- apps/neshub/src/keyframe.rs | 4 +- apps/neshub/src/main.rs | 79 +++- apps/neshub/src/session.rs | 478 ++++++++++++++++++++++- apps/neshub/src/windows.rs | 651 +++++++++++++++++++++++++++++++ crates/nesprotocol/src/lib.rs | 75 ++++ crates/nesprotocol/src/window.rs | 86 +++- 16 files changed, 2686 insertions(+), 88 deletions(-) create mode 100644 apps/nescope/src/windows.rs create mode 100644 apps/neshub/src/capture_cmd.rs create mode 100644 apps/neshub/src/windows.rs diff --git a/apps/nescapture/src/encode.rs b/apps/nescapture/src/encode.rs index 566062a7..3b41e512 100644 --- a/apps/nescapture/src/encode.rs +++ b/apps/nescapture/src/encode.rs @@ -56,7 +56,7 @@ use std::time::Instant; use nesprotocol::{ CODEC_AV1, CODEC_H264, CODEC_H265, CODEC_KEEP, CODEC_PYROWAVE, ClientCaps, FLAG_KEYFRAME, FLAG_RECONFIG, MSG_CLIENT_CAPS, MSG_ENCODE_SETTINGS, MSG_IDR_REQUEST, STREAM_VIDEO, - decode_client_caps, decode_encode_settings_ext, encode_ipc_frame, + decode_client_caps, decode_encode_settings_ext, encode_ipc_frame_from, }; use pixelforge::{ Codec, ColorConverter, ColorConverterConfig, ColorRange, ColorSpec, EncodeBitDepth, @@ -131,6 +131,74 @@ pub fn effective_colorspace(vk_colorspace: u32, declared: Option) -> u32 { vk_colorspace } +/// Relay commands sent to the shared command socket onto this process's own, +/// ref(d-0085). +/// +/// Whatever still sends to the shared path (a compositor's colour message, a +/// hub before it has heard from any encoder) reaches one encoder through it. +/// Taken only if nobody holds it: a live socket is another game's, and +/// unlinking it would cut that game off. One relay per process, however many +/// times the pipeline is rebuilt, so no thread is left behind. Best effort: +/// a process that cannot have it still has its own socket. +fn relay_legacy_commands(own: std::path::PathBuf) { + use std::os::unix::net::UnixDatagram; + use std::sync::atomic::{AtomicBool, Ordering}; + static RELAYING: AtomicBool = AtomicBool::new(false); + if RELAYING.swap(true, Ordering::AcqRel) { + return; + } + let shared = std::path::Path::new(nesprotocol::CAPTURE_CMD_SOCKET); + let live = UnixDatagram::unbound() + .and_then(|probe| probe.connect(shared)) + .is_ok(); + if live { + log::info!("{} belongs to another encoder; leaving it", shared.display()); + RELAYING.store(false, Ordering::Release); + return; + } + let _ = std::fs::remove_file(shared); + let sock = match UnixDatagram::bind(shared) { + Ok(s) => s, + Err(e) => { + log::warn!("shared cmd socket bind failed: {e}"); + RELAYING.store(false, Ordering::Release); + return; + } + }; + let _ = std::fs::set_permissions(shared, std::os::unix::fs::PermissionsExt::from_mode(0o666)); + let spawned = std::thread::Builder::new() + .name("nescapture-cmd-relay".into()) + .spawn(move || { + let Ok(out) = UnixDatagram::unbound() else { + return; + }; + let mut buf = [0u8; 128]; + while let Ok(n) = sock.recv(&mut buf) { + if let Err(e) = out.send_to(&buf[..n], &own) { + log::debug!("relaying a shared-socket command failed: {e}"); + } + } + }); + if let Err(e) = spawned { + log::warn!("spawn cmd relay: {e}"); + } +} + +/// `STREAM_ENCODER_HELLO` for this process: its pid and `NESTRI_LAUNCH_ID`. +fn encoder_hello() -> Vec { + let launch = std::env::var("NESTRI_LAUNCH_ID").unwrap_or_default(); + encode_ipc_frame_from( + std::process::id(), + nesprotocol::STREAM_ENCODER_HELLO, + 0, + 0, + 0, + 0, + 0, + launch.as_bytes(), + ) +} + /// A surface colour from the compositor, as a Vulkan colour space. /// /// Mapped into the swapchain's own vocabulary so there is one thing to reason @@ -1132,7 +1200,13 @@ impl PipelineHandle { thread::Builder::new() .name("nescapture-idr".into()) .spawn(move || { - let cmd_path = std::path::PathBuf::from("/tmp/nescapture-cmd.sock"); + // A socket of this process's own, ref(d-0085): with two games + // in one box, one shared path meant the second encoder took it + // from the first, and every keyframe request and bitrate change + // after that reached only the newer game. + let cmd_path = std::path::PathBuf::from(nesprotocol::capture_cmd_socket( + std::process::id(), + )); let _ = std::fs::remove_file(&cmd_path); let sock = match std::os::unix::net::UnixDatagram::bind(&cmd_path) { Ok(s) => { @@ -1148,6 +1222,10 @@ impl PipelineHandle { return; } }; + // The shared path too, relayed onto this one, so a hub or a + // compositor that predates per-process sockets still reaches the + // newest encoder exactly as it did. + relay_legacy_commands(cmd_path.clone()); let mut buf = [0u8; 128]; loop { match sock.recv(&mut buf) { @@ -2940,7 +3018,10 @@ fn ipc_send_thread( if reconfig { flags |= FLAG_RECONFIG; } - let ipc_frame = encode_ipc_frame( + // Tagged with this process, so the hub can tell which + // window the frame is, ref(d-0085). + let ipc_frame = encode_ipc_frame_from( + std::process::id(), STREAM_VIDEO, protocol_codec, flags, @@ -2949,7 +3030,16 @@ fn ipc_send_thread( height, &pkt.data, ); - (vec![ipc_frame], pkt.is_key_frame, pkt.data.len()) + // The launch this encoder draws for goes ahead of every + // keyframe, ref(d-0085): the hub cannot read this + // process's environment, and a keyframe is where a late + // hub starts anyway. + let messages = if pkt.is_key_frame { + vec![encoder_hello(), ipc_frame] + } else { + vec![ipc_frame] + }; + (messages, pkt.is_key_frame, pkt.data.len()) } Encoded::Pyro(frame) => { let enc_ms = (frame.stats.gpu_ns() / 1e6) as f32; diff --git a/apps/nescope/src/handlers.rs b/apps/nescope/src/handlers.rs index 075ec57f..c75212df 100644 --- a/apps/nescope/src/handlers.rs +++ b/apps/nescope/src/handlers.rs @@ -158,6 +158,20 @@ impl XdgShellHandler for NescopeState { &mut self.xdg_shell_state } + fn title_changed(&mut self, surface: ToplevelSurface) { + // The one place a toplevel's title is read after it was opened, not + // every commit. + let root = surface.wl_surface().clone(); + let window = self + .space + .elements() + .find(|w| w.wl_surface().is_some_and(|s| *s == root)) + .cloned(); + if let Some(window) = window { + self.window_retitled(&window); + } + } + fn new_toplevel(&mut self, surface: ToplevelSurface) { // Tell the toplevel to fill the entire virtual output. surface.with_pending_state(|state| { @@ -167,11 +181,13 @@ impl XdgShellHandler for NescopeState { let window = Window::new_wayland_window(surface); self.space.map_element(window.clone(), (0, 0), false); + self.window_opened(&window); self.set_keyboard_focus_to_window(&window); tracing::debug!("New XDG toplevel"); } fn toplevel_destroyed(&mut self, surface: ToplevelSurface) { + self.window_closed(&crate::state::WinKey::Wayland(surface.wl_surface().clone())); // Remove the dead window from the space. let elem = self .space @@ -467,7 +483,9 @@ impl XwmHandler for CalloopData { return; } let win = Window::new_x11_window(window); - self.state.space.map_element(win, (0, 0), true); + self.state.space.map_element(win.clone(), (0, 0), true); + // An id on first map; a window unmapped and mapped again keeps it. + self.state.window_opened(&win); self.state.determine_and_apply_focus(); } @@ -493,6 +511,9 @@ impl XwmHandler for CalloopData { if !window.is_override_redirect() { let _ = window.set_mapped(false); + // Closed for the hub; mapped again, it opens under the same id. + self.state + .window_unmapped(&crate::state::WinKey::X11(window.window_id())); } if was_focused { @@ -501,7 +522,10 @@ impl XwmHandler for CalloopData { } } - fn destroyed_window(&mut self, _xwm: XwmId, _window: X11Surface) {} + fn destroyed_window(&mut self, _xwm: XwmId, window: X11Surface) { + self.state + .window_closed(&crate::state::WinKey::X11(window.window_id())); + } fn configure_request( &mut self, @@ -543,11 +567,14 @@ impl XwmHandler for CalloopData { } } - fn property_notify(&mut self, _xwm: XwmId, _window: X11Surface, property: WmWindowProperty) { + fn property_notify(&mut self, _xwm: XwmId, window: X11Surface, property: WmWindowProperty) { // Only recalculate focus when the window class changes (steam_app_* detection). if matches!(property, WmWindowProperty::Class) { self.state.determine_and_apply_focus(); } + if matches!(property, WmWindowProperty::Title) { + self.state.window_retitled(&Window::new_x11_window(window)); + } } fn resize_request(&mut self, _: XwmId, _: X11Surface, _: u32, _: ResizeEdge) {} diff --git a/apps/nescope/src/input.rs b/apps/nescope/src/input.rs index ae02abff..f68e7bec 100644 --- a/apps/nescope/src/input.rs +++ b/apps/nescope/src/input.rs @@ -340,7 +340,7 @@ pub fn decode_wire_event(data: &[u8]) -> Option { } // Not input: the IPC source hands it to the output before this. DecodedInput::OutputSize { .. } => None, - // Box-internal window messages; nescope does not act on them yet. + // Box-internal window messages: the IPC source handles them before this. DecodedInput::WindowOutputSize { .. } | DecodedInput::WindowFocus { .. } => None, } } diff --git a/apps/nescope/src/main.rs b/apps/nescope/src/main.rs index 39717eaa..425def56 100644 --- a/apps/nescope/src/main.rs +++ b/apps/nescope/src/main.rs @@ -70,6 +70,7 @@ mod libinput_backend; //mod screenshot_ipc; //mod screenshot_wire; mod state; +mod windows; mod xwm; use crate::input::{decode_wire_event, process_input}; @@ -176,8 +177,8 @@ struct Args { /// Game command — everything after '--'. /// - /// **Optional.** With one, nescope launches it and exits when it and its - /// windows are gone — a wrapper around a single game. Without one, nescope + /// **Optional.** With one, nescope launches it and exits once it has + /// exited and no window of any launch is left. Without one, nescope /// is a plain compositor: it comes up, publishes its displays and waits, /// and whatever wants to draw connects to it. /// @@ -397,6 +398,7 @@ fn main() { Ok(write_stream) => { state.ipc_write = Some(write_stream); state.cursor_image_sent = false; // re-send on reconnect + state.hub_connected(); } Err(e) => { tracing::warn!("Failed to clone IPC write stream: {e}"); @@ -404,12 +406,25 @@ fn main() { } loop_handle .insert_source(source, move |payload, _, data| { - if let Some(nesprotocol::input::DecodedInput::OutputSize { width, height }) = - nesprotocol::input::decode_input_event(&payload) - { - data.state.request_output_size(width, height); - } else if let Some(event) = decode_wire_event(&payload) { - process_input(event, &mut data.state); + use nesprotocol::input::DecodedInput; + match nesprotocol::input::decode_input_event(&payload) { + // The old form is window 0's, ref(d-0085). + Some(DecodedInput::OutputSize { width, height }) => { + data.state.window_output_size(0, width, height) + } + Some(DecodedInput::WindowOutputSize { + window, + width, + height, + }) => data.state.window_output_size(window, width, height), + Some(DecodedInput::WindowFocus { window }) => { + data.state.focus_window_from_hub(window) + } + _ => { + if let Some(event) = decode_wire_event(&payload) { + process_input(event, &mut data.state); + } + } } }) .expect("Failed to register input IPC source"); @@ -587,8 +602,11 @@ fn main() { } // ── Auto-exit after all windows are gone ────────────────── - // Wait 5 s after the last mapped window disappears to give - // any lingering save-game / cleanup processes time to finish. + // Only once this nescope's own game has exited AND no window of + // any launch is left: later launches draw on this compositor, so + // their windows keep it up, ref(d-0085). The 5 s wait gives + // lingering save-game / cleanup processes time to finish. A box + // with one game ends exactly as before. if data.state.game_launched && data.game_process.is_none() { let has_windows = data.state.space.elements().next().is_some(); if !has_windows { diff --git a/apps/nescope/src/state.rs b/apps/nescope/src/state.rs index b2b04c61..622dcd04 100644 --- a/apps/nescope/src/state.rs +++ b/apps/nescope/src/state.rs @@ -72,6 +72,25 @@ fn fit_output(want: (u32, u32), max: (u32, u32)) -> (u32, u32) { (side(w, max.0), side(h, max.1)) } +/// What a window is known by for its id: its Wayland surface, or for an +/// Xwayland window its X11 id (the surface comes and goes with mapping). +#[derive(Debug, Clone, PartialEq)] +pub enum WinKey { + Wayland(WlSurface), + X11(u32), +} + +impl WinKey { + pub fn of(window: &Window) -> Option { + if let Some(x11) = window.x11_surface() { + return Some(WinKey::X11(x11.window_id())); + } + window + .toplevel() + .map(|t| WinKey::Wayland(t.wl_surface().clone())) + } +} + pub struct CalloopData { pub state: NescopeState, pub display: Display, @@ -205,6 +224,14 @@ pub struct NescopeState { pub presentation_seq: u64, + // ── Windows, ref(d-0085) ────────────────────────────────────────────── + /// Every toplevel's id, and what the hub has been told about it. + pub windows: crate::windows::Registry, + /// Which window input is for, as the hub's focus tags say. + pub window_input: crate::windows::WindowInput, + /// The window the cursor updates last sent describe. + last_cursor_window: nesprotocol::window::WindowId, + // ── Dimensions + frame rate ─────────────────────────────────────────── pub width: u32, pub height: u32, @@ -343,6 +370,9 @@ impl NescopeState { last_sent_cursor_status: 0xFF, last_pointer_activity: std::time::Instant::now(), presentation_seq: 0, + windows: Default::default(), + window_input: Default::default(), + last_cursor_window: 0, width, height, launch_size: (width, height), @@ -502,6 +532,11 @@ impl NescopeState { self.output .change_current_state(Some(mode), None, None, None); + let size = (clamp16(width), clamp16(height)); + for m in self.windows.resize_all(size) { + self.send_box_message(&m); + } + for window in self.space.elements().cloned().collect::>() { if let Some(tl) = window.toplevel() { tl.with_pending_state(|s| { @@ -783,8 +818,157 @@ impl NescopeState { send_frames_surface_tree(root, &output, now, None, |_, _| Some(output.clone())); } + // ----------------------------------------------------------------------- + // Windows, ref(d-0085) + // ----------------------------------------------------------------------- + + /// One box-internal message to the hub on the input socket. + pub(crate) fn send_box_message(&mut self, message: &[u8]) { + use std::io::Write; + let Some(ipc) = self.ipc_write.as_mut() else { + return; + }; + let mut frame = Vec::with_capacity(2 + message.len()); + frame.extend_from_slice(&(message.len() as u16).to_le_bytes()); + frame.extend_from_slice(message); + let _ = ipc.write_all(&frame); + let _ = ipc.flush(); + } + + /// A toplevel was mapped: give it an id, once, and tell the hub. + pub(crate) fn window_opened(&mut self, window: &Window) { + let Some(key) = WinKey::of(window) else { + return; + }; + let (pid, title, app_id) = if let Some(x11) = window.x11_surface() { + // The Wayland client of an Xwayland window is Xwayland itself; + // the game says who it is in _NET_WM_PID. + (x11.pid(), x11.title(), x11.class()) + } else if let Some(toplevel) = window.toplevel() { + use smithay::reexports::wayland_server::Resource as _; + let pid = toplevel + .wl_surface() + .client() + .and_then(|c| c.get_credentials(&self.display_handle).ok()) + .and_then(|c| u32::try_from(c.pid).ok()); + let (title, app_id) = toplevel_title(toplevel.wl_surface()); + (pid, title, app_id) + } else { + return; + }; + let new = crate::windows::NewWindow { + pid, + title, + app_id, + size: (clamp16(self.width), clamp16(self.height)), + }; + if let Some(messages) = + self.windows + .open(key, new, crate::windows::launch_of_process) + { + for m in messages { + self.send_box_message(&m); + } + } + } + + /// A toplevel was destroyed. + pub(crate) fn window_closed(&mut self, key: &WinKey) { + let id = self.windows.id_of(key); + if let Some(m) = self.windows.close(key) { + self.send_box_message(&m); + } + if let Some(id) = id { + self.window_input.forget(id); + } + } + + /// An X11 window was unmapped: closed for the hub, its id kept. + pub(crate) fn window_unmapped(&mut self, key: &WinKey) { + if let Some(m) = self.windows.unmap(key) { + self.send_box_message(&m); + } + } + + /// A window's title may have changed. + pub(crate) fn window_retitled(&mut self, window: &Window) { + let Some(key) = WinKey::of(window) else { + return; + }; + let title = match window.x11_surface() { + Some(x11) => x11.title(), + None => match window.toplevel() { + Some(t) => toplevel_title(t.wl_surface()).0, + None => return, + }, + }; + if let Some(m) = self.windows.retitle(&key, &title) { + self.send_box_message(&m); + } + } + + /// The hub's input connection is new: input is window 0's until it says + /// otherwise, and it hears every window there is. + pub(crate) fn hub_connected(&mut self) { + self.window_input.reset(); + let size = (clamp16(self.width), clamp16(self.height)); + for m in self.windows.announce_all(size) { + self.send_box_message(&m); + } + } + + /// The hub says input from here on is `id`'s: give it keyboard focus. + /// Nothing is raised and the output keeps its size: a fullscreen game + /// must not lose focus, or the output change size, because a client + /// looked at another window. Size follows only an explicit request, + /// ref(d-0085). + pub(crate) fn focus_window_from_hub(&mut self, id: nesprotocol::window::WindowId) { + if !self.window_input.focus(id) { + return; + } + let key = self.windows.key_of(id).cloned(); + let window = key.and_then(|key| { + self.space + .elements() + .find(|w| WinKey::of(w).as_ref() == Some(&key)) + .cloned() + }); + match window { + Some(window) => self.set_keyboard_focus_to_window(&window), + None => tracing::debug!("focus for window {id}, which is not mapped"), + } + } + + /// A client's window is `width`x`height`, for window `id`. + pub(crate) fn window_output_size( + &mut self, + id: nesprotocol::window::WindowId, + width: u16, + height: u16, + ) { + if let Some((w, h)) = + self.window_input + .size(id, width, height, std::time::Instant::now()) + { + self.request_output_size(w, h); + } + } + pub(crate) fn send_cursor_update(&mut self) { use std::io::Write; + // Cursor updates describe the window input is going to; the hub sends + // them to that window's clients. Said only when it changes, so a box + // that only ever has window 0 sends what it always did. + let current = self.window_input.current(); + if current != self.last_cursor_window && self.ipc_write.is_some() { + let mut m = Vec::with_capacity(3); + nesprotocol::window::encode_cursor_window(&mut m, current); + self.send_box_message(&m); + self.last_cursor_window = current; + // Those clients may never have had the image or the state. + self.cursor_image_sent = false; + self.last_sent_cursor_status = 0xFF; + } let ipc = match self.ipc_write.as_mut() { Some(s) => s, None => return, @@ -859,6 +1043,27 @@ impl NescopeState { // Helpers // --------------------------------------------------------------------------- +fn clamp16(v: u32) -> u16 { + v.min(u32::from(u16::MAX)) as u16 +} + +/// An xdg toplevel's title and app id, empty where the client set none. +fn toplevel_title(surface: &WlSurface) -> (String, String) { + smithay::wayland::compositor::with_states(surface, |states| { + states + .data_map + .get::() + .and_then(|d| d.lock().ok()) + .map(|d| { + ( + d.title.clone().unwrap_or_default(), + d.app_id.clone().unwrap_or_default(), + ) + }) + .unwrap_or_default() + }) +} + fn intern_atom( conn: &smithay::reexports::x11rb::rust_connection::RustConnection, name: &[u8], diff --git a/apps/nescope/src/windows.rs b/apps/nescope/src/windows.rs new file mode 100644 index 00000000..acf20b8b --- /dev/null +++ b/apps/nescope/src/windows.rs @@ -0,0 +1,440 @@ +//! Window ids and what the hub is told about each window, ref(d-0085). +//! +//! Every toplevel gets an id when it is first mapped: `0` for the first window +//! of this compositor, then counting up, never reused, and none past 65,535. +//! The hub hears each window's lifecycle on the input socket in nesprotocol's +//! window encoding, and which process owns it, which is how it joins an +//! encoder's frames (tagged with its process) to the window they draw. +//! +//! This module holds no compositor objects beyond an opaque key, so the rules +//! can be tested without one. + + +use nesprotocol::window::{WindowEvent, WindowId, encode_event_message, encode_window_pid}; + +/// One window as the hub knows it. +#[derive(Debug, Clone)] +pub struct Tracked { + pub key: K, + pub id: WindowId, + pub pid: Option, + /// False while an X11 window is unmapped: the hub has been told it closed, + /// and a map again opens it under the same id. + mapped: bool, + launch_id: String, + title: String, + app_id: String, +} + +/// The windows this compositor has given ids to. +#[derive(Debug)] +pub struct Registry { + /// The next id to mint; past `u16::MAX` none are left. + next: u32, + windows: Vec>, + refused: bool, +} + +impl Default for Registry { + fn default() -> Self { + Self { + next: 0, + windows: Vec::new(), + refused: false, + } + } +} + +/// What the compositor knows about a window when it maps. +pub struct NewWindow { + pub pid: Option, + pub title: String, + pub app_id: String, + pub size: (u16, u16), +} + +impl Registry { + /// Give a newly mapped window an id, and return the messages that tell the + /// hub. A window unmapped earlier is opened again under its old id. `None` + /// for a window that is already open, or when ids have run out for this + /// run. + pub fn open( + &mut self, + key: K, + new: NewWindow, + launch_of: impl Fn(u32) -> Option, + ) -> Option>> { + if let Some(w) = self.windows.iter_mut().find(|w| w.key == key) { + if w.mapped { + return None; + } + w.mapped = true; + w.pid = new.pid.or(w.pid); + w.title = new.title; + w.app_id = new.app_id; + tracing::info!(window = w.id, "window opened again"); + return Some(announce(w, new.size)); + } + let Ok(id) = WindowId::try_from(self.next) else { + if !self.refused { + tracing::error!("this compositor has minted every window id; refusing new windows"); + self.refused = true; + } + return None; + }; + self.next += 1; + let launch_id = new.pid.and_then(launch_of).unwrap_or_default(); + let tracked = Tracked { + key, + id, + pid: new.pid, + mapped: true, + launch_id, + title: new.title, + app_id: new.app_id, + }; + let messages = announce(&tracked, new.size); + tracing::info!( + window = id, + pid = ?tracked.pid, + launch = %tracked.launch_id, + app_id = %tracked.app_id, + "window opened" + ); + self.windows.push(tracked); + Some(messages) + } + + /// A window went for good. The message for the hub, if it was open. + pub fn close(&mut self, key: &K) -> Option> { + let at = self.windows.iter().position(|w| w.key == *key)?; + let gone = self.windows.remove(at); + tracing::info!(window = gone.id, "window closed"); + gone.mapped.then(|| closed(gone.id)) + } + + /// An X11 window was unmapped: closed for the hub, but its id is kept for + /// when it is mapped again. + pub fn unmap(&mut self, key: &K) -> Option> { + let w = self.windows.iter_mut().find(|w| w.key == *key)?; + if !w.mapped { + return None; + } + w.mapped = false; + tracing::info!(window = w.id, "window unmapped"); + Some(closed(w.id)) + } + + /// A window's title is `title` now. The message for the hub if it changed. + pub fn retitle(&mut self, key: &K, title: &str) -> Option> { + let w = self.windows.iter_mut().find(|w| w.key == *key)?; + if w.title == title { + return None; + } + w.title = title.to_string(); + if !w.mapped { + return None; + } + let mut m = Vec::new(); + encode_event_message( + &mut m, + &WindowEvent::Title { + window: w.id, + title: w.title.clone(), + }, + ); + Some(m) + } + + /// Every window is now `size`: one output, which every window fills. + pub fn resize_all(&self, size: (u16, u16)) -> Vec> { + self.windows + .iter() + .filter(|w| w.mapped) + .map(|w| { + let mut m = Vec::new(); + encode_event_message( + &mut m, + &WindowEvent::Size { + window: w.id, + width: size.0, + height: size.1, + }, + ); + m + }) + .collect() + } + + /// Everything the hub needs to know of every window, for a hub that has + /// just connected. + pub fn announce_all(&self, size: (u16, u16)) -> Vec> { + self.windows + .iter() + .filter(|w| w.mapped) + .flat_map(|w| announce(w, size)) + .collect() + } + + pub fn id_of(&self, key: &K) -> Option { + self.windows.iter().find(|w| w.key == *key).map(|w| w.id) + } + + pub fn key_of(&self, id: WindowId) -> Option<&K> { + self.windows + .iter() + .find(|w| w.id == id && w.mapped) + .map(|w| &w.key) + } +} + +fn closed(window: WindowId) -> Vec { + let mut m = Vec::new(); + encode_event_message(&mut m, &WindowEvent::Closed { window }); + m +} + +fn announce(w: &Tracked, size: (u16, u16)) -> Vec> { + let mut out = Vec::with_capacity(2); + let mut m = Vec::new(); + encode_event_message( + &mut m, + &WindowEvent::Opened { + window: w.id, + width: size.0, + height: size.1, + launch_id: w.launch_id.clone(), + title: w.title.clone(), + app_id: w.app_id.clone(), + }, + ); + out.push(m); + if let Some(pid) = w.pid { + let mut m = Vec::new(); + encode_window_pid(&mut m, w.id, pid); + out.push(m); + } + out +} + +/// Read a process's `NESTRI_LAUNCH_ID`, which the launch put in its +/// environment and its children inherit. +pub fn launch_of_process(pid: u32) -> Option { + let environ = std::fs::read(format!("/proc/{pid}/environ")).ok()?; + environ + .split(|b| *b == 0) + .find_map(|kv| kv.strip_prefix(b"NESTRI_LAUNCH_ID=")) + .and_then(|v| String::from_utf8(v.to_vec()).ok()) +} + +/// How long the output keeps the size one window gave it before another +/// window may change it. +pub const OUTPUT_HYSTERESIS: std::time::Duration = std::time::Duration::from_millis(500); + +/// Which window input is for, and who may size the output. +/// +/// The hub tags input with focus messages; until it sends one, and again +/// after every reconnect, input is window `0`'s, which is what a hub that +/// predates tags means. Focus moves keyboard and pointer input and nothing +/// else. With one output for every window, only an explicit size request from +/// the window that has focus changes it; when several windows ask, the last +/// writer wins, but not within [`OUTPUT_HYSTERESIS`] of another window's +/// change, so two clients of different sizes do not fight over it, +/// ref(d-0085). +#[derive(Debug, Default)] +pub struct WindowInput { + current: WindowId, + last_resize: Option<(WindowId, std::time::Instant)>, +} + +impl WindowInput { + /// The input connection is new: back to window `0`. + pub fn reset(&mut self) { + self.current = 0; + } + + pub fn current(&self) -> WindowId { + self.current + } + + /// `window`'s client is `width`x`height`. The size to apply now, if any: + /// only for the window that owns the output, and not within + /// [`OUTPUT_HYSTERESIS`] of a change another window made. + pub fn size( + &mut self, + window: WindowId, + width: u16, + height: u16, + now: std::time::Instant, + ) -> Option<(u16, u16)> { + if window != self.current { + return None; + } + if let Some((who, at)) = self.last_resize + && who != window + && now.duration_since(at) < OUTPUT_HYSTERESIS + { + return None; + } + self.last_resize = Some((window, now)); + Some((width, height)) + } + + /// Input from here on is `window`'s. Whether that is a change. + pub fn focus(&mut self, window: WindowId) -> bool { + if window == self.current { + return false; + } + self.current = window; + true + } + + pub fn forget(&mut self, window: WindowId) { + if self.last_resize.is_some_and(|(w, _)| w == window) { + self.last_resize = None; + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use nesprotocol::window::{MSG_WINDOW_PID, decode_window_pid}; + + fn new(pid: Option, title: &str) -> NewWindow { + NewWindow { + pid, + title: title.into(), + app_id: "app".into(), + size: (1920, 1080), + } + } + + fn decode(m: &[u8]) -> WindowEvent { + WindowEvent::decode(m[0], &m[1..]).expect("an event") + } + + #[test] + fn ids_start_at_zero_and_are_never_reused() { + let mut r = Registry::::default(); + let launch = |pid: u32| Some(format!("launch_{pid}")); + let a = r.open(10, new(Some(1), "a"), launch).unwrap(); + assert_eq!( + decode(&a[0]), + WindowEvent::Opened { + window: 0, + width: 1920, + height: 1080, + launch_id: "launch_1".into(), + title: "a".into(), + app_id: "app".into(), + } + ); + assert_eq!(a[1][0], MSG_WINDOW_PID); + assert_eq!(decode_window_pid(&a[1][1..]), Some((0, 1))); + assert!( + r.open(10, new(Some(1), "a"), launch).is_none(), + "once per window" + ); + let b = r.open(11, new(None, "b"), launch).unwrap(); + assert_eq!(b.len(), 1, "no owner to name"); + assert_eq!(decode(&b[0]).window(), 1); + assert_eq!( + decode(&r.close(&10).unwrap()), + WindowEvent::Closed { window: 0 } + ); + assert!(r.close(&10).is_none()); + let c = r.open(12, new(None, "c"), launch).unwrap(); + assert_eq!(decode(&c[0]).window(), 2, "0 is not given out again"); + assert_eq!(r.key_of(2), Some(&12)); + assert_eq!(r.id_of(&11), Some(1)); + } + + #[test] + fn ids_run_out_rather_than_wrap() { + let mut r = Registry:: { + next: u32::from(u16::MAX), + ..Default::default() + }; + assert!(r.open(1, new(None, ""), |_| None).is_some()); + assert!(r.open(2, new(None, ""), |_| None).is_none()); + assert!(r.open(3, new(None, ""), |_| None).is_none()); + } + + #[test] + fn titles_and_sizes() { + let mut r = Registry::::default(); + r.open(1, new(None, "a"), |_| None); + assert!(r.retitle(&1, "a").is_none(), "unchanged says nothing"); + assert_eq!( + decode(&r.retitle(&1, "b").unwrap()), + WindowEvent::Title { + window: 0, + title: "b".into() + } + ); + r.open(2, new(None, "c"), |_| None); + let sizes: Vec<_> = r + .resize_all((1280, 720)) + .iter() + .map(|m| decode(m)) + .collect(); + assert_eq!(sizes.len(), 2); + assert_eq!( + sizes[1], + WindowEvent::Size { + window: 1, + width: 1280, + height: 720 + } + ); + assert_eq!(r.announce_all((1, 1)).len(), 2); + } + + #[test] + fn an_unmapped_window_is_closed_for_the_hub_and_keeps_its_id() { + let mut r = Registry::::default(); + let launch = |_: u32| None; + r.open(7, new(Some(1), "a"), launch).unwrap(); + assert_eq!(decode(&r.unmap(&7).unwrap()), WindowEvent::Closed { window: 0 }); + assert!(r.unmap(&7).is_none()); + assert!(r.retitle(&7, "b").is_none(), "nothing to tell while closed"); + assert!(r.announce_all((1, 1)).is_empty()); + let again = r.open(7, new(Some(1), "c"), launch).unwrap(); + assert_eq!(decode(&again[0]).window(), 0); + assert!(matches!(decode(&again[0]), WindowEvent::Opened { .. })); + // Destroyed while unmapped: nothing more to say. + r.unmap(&7); + assert!(r.close(&7).is_none()); + assert_eq!(decode(&r.open(8, new(None, "d"), launch).unwrap()[0]).window(), 1); + } + + #[test] + fn focus_moves_input_only_and_sizing_is_the_focused_windows() { + use std::time::{Duration, Instant}; + let t = Instant::now(); + let mut i = WindowInput::default(); + // A hub that predates tags: every resize is window 0's, applied. + assert_eq!(i.size(0, 1280, 720, t), Some((1280, 720))); + // Another window's resize is not applied, and not remembered. + assert_eq!(i.size(3, 800, 600, t), None); + assert!(i.focus(3)); + assert!(!i.focus(3)); + // Focus alone changed no size; the window must ask. + // Within the hysteresis of window 0's change: held off. + assert_eq!(i.size(3, 1024, 768, t + Duration::from_millis(100)), None); + assert_eq!( + i.size(3, 1024, 768, t + Duration::from_millis(500)), + Some((1024, 768)) + ); + // The same window may resize at once. + assert_eq!( + i.size(3, 1100, 800, t + Duration::from_millis(510)), + Some((1100, 800)) + ); + assert!(i.focus(0)); + assert_eq!(i.size(0, 640, 480, t + Duration::from_millis(600)), None); + i.reset(); + assert_eq!(i.current(), 0); + } +} diff --git a/apps/neshub/src/capture_cmd.rs b/apps/neshub/src/capture_cmd.rs new file mode 100644 index 00000000..32cf9d55 --- /dev/null +++ b/apps/neshub/src/capture_cmd.rs @@ -0,0 +1,235 @@ +//! Where commands for the encoders go, ref(d-0085). +//! +//! Every game in the box runs an encoder of its own, listening on a socket +//! named for its process. The hub learns an encoder's process id from the +//! first frame it tags, and from then on sends every command to every encoder +//! it knows. Until it knows one, commands go to the shared socket, exactly as +//! they did when a box had one encoder: that is the path for an encoder that +//! predates tagging, and for every command sent before the first frame. +//! +//! An encoder that turns up later has missed what the others were told (the +//! codec every client decodes, the bitrate), so it is told again: the latest +//! command of each type, then a keyframe request. A keyframe request for one +//! client goes only to the encoder of the window that client is shown. The first encoder is not: every +//! command before it went to the shared socket, which reached it, and +//! repeating a codec change to an encoder already on that codec would restart +//! its stream for nothing. A box with one game therefore gets exactly the +//! commands it always did. + +use std::collections::BTreeSet; +use std::path::PathBuf; + +use nesprotocol::MSG_IDR_REQUEST; + +/// Marks a hub-internal command meant for one encoder: `[TO_ENCODER][4B pid] +/// [command]`. Stripped before anything is sent; never leaves the hub. +const TO_ENCODER: u8 = 0xF0; + +/// `cmd` for the encoder in `pid` alone, or for every encoder when `None`. +pub fn to_encoder(pid: Option, cmd: &[u8]) -> Vec { + match pid { + Some(pid) => { + let mut out = Vec::with_capacity(5 + cmd.len()); + out.push(TO_ENCODER); + out.extend_from_slice(&pid.to_le_bytes()); + out.extend_from_slice(cmd); + out + } + None => cmd.to_vec(), + } +} + +/// A keyframe request for the encoder in `pid`, or every encoder. +pub fn keyframe_request(pid: Option) -> Vec { + to_encoder(pid, &[MSG_IDR_REQUEST]) +} + +/// Split a queued command into its target and the command itself. +pub fn target_of(cmd: Vec) -> (Option, Vec) { + if cmd.len() >= 5 && cmd[0] == TO_ENCODER { + let pid = u32::from_le_bytes([cmd[1], cmd[2], cmd[3], cmd[4]]); + (Some(pid), cmd[5..].to_vec()) + } else { + (None, cmd) + } +} + +#[derive(Debug, Default)] +pub struct Fanout { + encoders: BTreeSet, + /// The latest command of each type, in the order they were last sent. + history: Vec>, + /// Whether any encoder has been known: only the very first is spared the + /// replay, since only it was reached by the shared socket. + ever_known: bool, +} + +impl Fanout { + /// Where a command goes now: every known encoder's own socket, or the + /// shared one while none is known. + pub fn destinations(&self) -> Vec { + if self.encoders.is_empty() { + return vec![PathBuf::from(nesprotocol::CAPTURE_CMD_SOCKET)]; + } + self.encoders + .iter() + .map(|pid| PathBuf::from(nesprotocol::capture_cmd_socket(*pid))) + .collect() + } + + /// Where a command for `target` goes: that encoder alone when it is + /// known, otherwise wherever an untargeted one would. + pub fn destinations_for(&self, target: Option) -> Vec { + match target { + Some(pid) if self.encoders.contains(&pid) => { + vec![PathBuf::from(nesprotocol::capture_cmd_socket(pid))] + } + _ => self.destinations(), + } + } + + /// Record a command as sent to every encoder, so a later one can be told + /// it. Only the latest of each type is kept. + pub fn sent(&mut self, cmd: &[u8]) { + let Some(&kind) = cmd.first() else { + return; + }; + if kind == MSG_IDR_REQUEST { + return; + } + self.history.retain(|c| c[0] != kind); + self.history.push(cmd.to_vec()); + } + + /// A frame came from `pid`. Returns what to send it if it is new to the + /// hub (or was forgotten), `None` if it was known. + pub fn encoder(&mut self, pid: u32) -> Option>> { + if !self.encoders.insert(pid) { + return None; + } + if !std::mem::replace(&mut self.ever_known, true) { + return Some(Vec::new()); + } + let mut replay = self.history.clone(); + replay.push(vec![MSG_IDR_REQUEST]); + Some(replay) + } + + /// The encoder in `pid` has gone: its socket refuses. + pub fn forget(&mut self, pid: u32) { + self.encoders.remove(&pid); + } +} + +/// Whether a send error means the encoder's process is gone, rather than a +/// passing condition (a full buffer) that the next command may not meet. +pub fn encoder_is_gone(e: &std::io::Error) -> bool { + matches!( + e.kind(), + std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::NotFound + ) +} + +/// The process id in an encoder's own socket path. +pub fn pid_of(path: &std::path::Path) -> Option { + path.file_name()? + .to_str()? + .strip_prefix("nescapture-cmd-")? + .strip_suffix(".sock")? + .parse() + .ok() +} + +#[cfg(test)] +mod tests { + use super::*; + + const SETTINGS: &[u8] = &[nesprotocol::MSG_ENCODE_SETTINGS, 1, 2, 3]; + + #[test] + fn one_untagged_encoder_is_the_shared_socket_as_before() { + let mut f = Fanout::default(); + assert_eq!( + f.destinations(), + vec![PathBuf::from("/tmp/nescapture-cmd.sock")] + ); + f.sent(SETTINGS); + assert_eq!( + f.destinations(), + vec![PathBuf::from("/tmp/nescapture-cmd.sock")] + ); + } + + #[test] + fn the_first_encoder_is_not_told_twice() { + let mut f = Fanout::default(); + f.sent(SETTINGS); + assert_eq!(f.encoder(10), Some(vec![])); + assert_eq!(f.encoder(10), None); + assert_eq!( + f.destinations(), + vec![PathBuf::from("/tmp/nescapture-cmd-10.sock")] + ); + } + + #[test] + fn a_later_encoder_hears_the_latest_of_each_type_then_a_keyframe_request() { + let mut f = Fanout::default(); + f.sent(SETTINGS); + f.encoder(10); + f.sent(&[MSG_IDR_REQUEST]); + let caps = [nesprotocol::MSG_CLIENT_CAPS, 1]; + let bitrate = [nesprotocol::MSG_ENCODE_SETTINGS, 9]; + f.sent(&caps); + f.sent(&bitrate); + assert_eq!( + f.encoder(20), + Some(vec![caps.to_vec(), bitrate.to_vec(), vec![MSG_IDR_REQUEST]]) + ); + assert_eq!(f.destinations().len(), 2); + f.forget(10); + assert_eq!( + f.destinations(), + vec![PathBuf::from("/tmp/nescapture-cmd-20.sock")] + ); + // Forgotten, then heard from again: learnt again, and told again. + assert_eq!(f.encoder(10).map(|r| r.len()), Some(3)); + } + + #[test] + fn a_targeted_command_goes_to_its_encoder_alone() { + let mut f = Fanout::default(); + f.encoder(10); + f.encoder(20); + let (target, cmd) = target_of(keyframe_request(Some(20))); + assert_eq!(cmd, vec![MSG_IDR_REQUEST]); + assert_eq!( + f.destinations_for(target), + vec![PathBuf::from("/tmp/nescapture-cmd-20.sock")] + ); + // Unknown target, or none: everyone. + assert_eq!(f.destinations_for(Some(30)).len(), 2); + assert_eq!(target_of(keyframe_request(None)), (None, vec![MSG_IDR_REQUEST])); + assert!(encoder_is_gone(&std::io::Error::from(std::io::ErrorKind::ConnectionRefused))); + assert!(!encoder_is_gone(&std::io::Error::from(std::io::ErrorKind::WouldBlock))); + } + + #[test] + fn a_second_encoder_before_any_command_gets_a_keyframe_request() { + let mut f = Fanout::default(); + assert_eq!(f.encoder(10), Some(vec![])); + assert_eq!(f.encoder(11), Some(vec![vec![MSG_IDR_REQUEST]])); + } + + #[test] + fn pid_from_path() { + assert_eq!( + pid_of(std::path::Path::new("/tmp/nescapture-cmd-42.sock")), + Some(42) + ); + assert_eq!( + pid_of(std::path::Path::new("/tmp/nescapture-cmd.sock")), + None + ); + } +} diff --git a/apps/neshub/src/dgram.rs b/apps/neshub/src/dgram.rs index be6f2015..7d401279 100644 --- a/apps/neshub/src/dgram.rs +++ b/apps/neshub/src/dgram.rs @@ -14,10 +14,13 @@ use tracing::{debug, warn}; use crate::keyframe::KeyframeSender; use crate::pyro::PyroSender; -use nesprotocol::MSG_DATA; use nesprotocol::datagram::{ - DGRAM_BUFFER_BYTES, DGRAM_HDR_LEN, DatagramHeader, MAX_FRAGMENTS_PER_FRAME, MIN_DGRAM_PAYLOAD, - fragment_count, write_datagram_header, + DGRAM_BUFFER_BYTES, DGRAM_HDR_LEN, DGRAM_VIDEO, DatagramHeader, MAX_FRAGMENTS_PER_FRAME, + MIN_DGRAM_PAYLOAD, fragment_count, write_datagram_header, +}; +use nesprotocol::window::{ + DGRAM_WINDOW_HDR_LEN, WINDOW_DEFAULT, WindowId, max_fragment_payload, + write_video_datagram_header, }; use nesprotocol::pyrowave::PyroFrame; use nesprotocol::reliable::video_wants_reliable; @@ -30,8 +33,16 @@ use nesprotocol::reliable::video_wants_reliable; #[derive(Debug, Clone)] pub enum MediaItem { /// A hardware-codec frame or an audio packet, laid out as the hub - /// broadcasts it. + /// broadcasts it. Video sent this way is window `0`, in the encoding every + /// client has always read. Coded(Bytes), + /// A hardware-codec frame of another window, ref(d-0085). + Windowed(WindowId, Bytes), + /// The connection now shows another window: number from zero again and + /// forget any wait for a keyframe, since none of it describes the new + /// window. Sent through the same queue as the frames so it lands between + /// the last frame of the old window and the first of the new. + Resubscribe, /// A whole PyroWave frame, which travels by rules of its own. PyroWave(Arc), } @@ -101,7 +112,12 @@ impl DatagramSender { /// Fragment one frame body and send the pieces. /// /// `body` is `[1B type][2B seq][payload]` — the frame layout the stream path - /// used, minus the length prefix a datagram does not need. + /// used, minus the length prefix a datagram does not need — or, for a + /// window other than `0`, the windowed body. + /// + /// `window` is the video window the datagrams belong to: its header is two + /// bytes longer for any window but `0`, so the fragment size follows it. + /// Always `0` for audio. /// /// A failure is reported but never retried. Retrying a real-time frame means /// delivering it late, which is the behaviour this whole change removes. The @@ -109,7 +125,7 @@ impl DatagramSender { /// hold of the connection lock, so a path that shrinks mid-frame rejects the /// frame rather than leaving half of it on the wire for the receiver to time /// out. - pub fn send_frame(&self, seq: u16, body: &[u8]) -> Result<(), SendFrameError> { + pub fn send_frame(&self, seq: u16, window: WindowId, body: &[u8]) -> Result<(), SendFrameError> { // Ask QUIC what fits right now rather than assuming. The estimate moves // over a connection's life as path MTU discovery runs, and it can shrink // when a path changes underneath us — which for iroh includes migrating @@ -121,7 +137,13 @@ impl DatagramSender { .max_datagram_size() .ok_or(SendFrameError::Unsupported)?; - let payload_size = max.saturating_sub(DGRAM_HDR_LEN); + let windowed = self.kind == DGRAM_VIDEO && window != WINDOW_DEFAULT; + let header_len = if windowed { DGRAM_WINDOW_HDR_LEN } else { DGRAM_HDR_LEN }; + let payload_size = if self.kind == DGRAM_VIDEO { + max_fragment_payload(window, max) + } else { + max.saturating_sub(DGRAM_HDR_LEN) + }; if payload_size < MIN_DGRAM_PAYLOAD { return Err(SendFrameError::PathTooSmall { payload: payload_size, @@ -136,25 +158,27 @@ impl DatagramSender { // Lay every fragment down in one allocation, then hand out slices of it. // `BytesMut::split_to` gives each datagram an owned `Bytes` that shares // this buffer, so a 300-fragment keyframe costs one allocation, not 300. - let mut buf = BytesMut::with_capacity(total * DGRAM_HDR_LEN + body.len()); - let mut header = [0u8; DGRAM_HDR_LEN]; + let mut buf = BytesMut::with_capacity(total * header_len + body.len()); + let mut header = [0u8; DGRAM_WINDOW_HDR_LEN]; // One reading for the whole frame: every fragment is handed over in the // single call below, so they all leave this end at the same moment. let sent_us = nesprotocol::clock::monotonic_us(); for index in 0..total { let start = index * payload_size; let end = (start + payload_size).min(body.len()); - write_datagram_header( - &mut header, - &DatagramHeader { - kind: self.kind, - seq, - index: index as u16, - total: total as u16, - sent_us, - }, - ); - buf.extend_from_slice(&header); + let h = DatagramHeader { + kind: self.kind, + seq, + index: index as u16, + total: total as u16, + sent_us, + }; + if windowed { + write_video_datagram_header(&mut header, &h, window); + } else { + write_datagram_header(&mut header, &h); + } + buf.extend_from_slice(&header[..header_len]); buf.extend_from_slice(&body[start..end]); } @@ -162,7 +186,7 @@ impl DatagramSender { .map(|index| { let start = index * payload_size; let end = (start + payload_size).min(body.len()); - buf.split_to(DGRAM_HDR_LEN + (end - start)).freeze() + buf.split_to(header_len + (end - start)).freeze() }) .collect(); @@ -288,8 +312,14 @@ pub async fn run_datagram_writer( while let Some(item) = rx.recv().await { let t0 = std::time::Instant::now(); - let payload = match item { - MediaItem::Coded(payload) => payload, + let (window, payload) = match item { + MediaItem::Coded(payload) => (WINDOW_DEFAULT, payload), + MediaItem::Windowed(window, payload) => (window, payload), + MediaItem::Resubscribe => { + seq = 0; + resync = ResyncGate::default(); + continue; + } // No keyframe stream and no resync gate: every PyroWave frame // stands alone, so there is nothing to resynchronise and no frame // worth a reliable stream more than another. @@ -306,7 +336,9 @@ pub async fn run_datagram_writer( } }; body.clear(); - nesprotocol::encode_frame_body(&mut body, MSG_DATA, seq, &payload); + // Window 0 is exactly the `MSG_DATA` body every client reads; audio + // is always window 0. + nesprotocol::window::encode_video_frame_body(&mut body, window, seq, &payload); // A keyframe goes on a stream of its own when one will take it. The // relay timing below is not recorded for it: the write is asynchronous @@ -352,7 +384,7 @@ pub async fn run_datagram_writer( if let Some(ref keyframes) = keyframes && video_wants_reliable(&payload) { - if keyframes.send(seq, &body) { + if keyframes.send(seq, window, &body) { seq = seq.wrapping_add(1); continue; } @@ -366,7 +398,7 @@ pub async fn run_datagram_writer( ); } - match sender.send_frame(seq, &body) { + match sender.send_frame(seq, window, &body) { Ok(()) => { if let Some(ref relay) = relay_ms { let elapsed = t0.elapsed().as_secs_f32() * 1000.0; diff --git a/apps/neshub/src/e2e_tests.rs b/apps/neshub/src/e2e_tests.rs index 915024dd..3feb6d43 100644 --- a/apps/neshub/src/e2e_tests.rs +++ b/apps/neshub/src/e2e_tests.rs @@ -85,9 +85,12 @@ async fn rig_with(client_buffer: Option) -> Rig { ))); // Commands go nowhere here: no client in these tests states codecs. let (cmd_tx, _cmd_rx) = tokio::sync::mpsc::unbounded_channel(); + let (encoder_tx, _encoder_rx) = tokio::sync::mpsc::unbounded_channel(); async move { let _cmd_rx = _cmd_rx; - crate::ipc_listener::run_video_listener(path, mgr, controller, cmd_tx).await + let _encoder_rx = _encoder_rx; + crate::ipc_listener::run_video_listener(path, mgr, controller, cmd_tx, encoder_tx, Default::default()) + .await } }); @@ -399,3 +402,179 @@ async fn a_new_carrier_starts_from_live_media() { "the first audio a new carrier sent was produced before it attached" ); } + +/// Two encoders, two windows, through the hub for real, ref(d-0085). +mod windows { + use super::*; + use nesprotocol::window::{ + DGRAM_VIDEO_WINDOW, STREAM_WINDOWS, WindowEvent, decode_video_datagram, + decode_video_frame_body, encode_event_message, encode_window_pid, + }; + use nesprotocol::{CODEC_H264, FLAG_KEYFRAME, MSG_DATA, STREAM_VIDEO, encode_ipc_frame_from}; + + const GAME_A: u32 = 100; + const GAME_B: u32 = 200; + + async fn draw(rig: &Rig, pid: u32, n: u8, key: bool) { + let flags = if key { FLAG_KEYFRAME } else { 0 }; + let tag = (pid / 100) as u8; + let m = encode_ipc_frame_from(pid, STREAM_VIDEO, CODEC_H264, flags, 1, 64, 64, &[tag, n]); + rig.sender.send(&m).await.expect("ipc send"); + tokio::time::sleep(Duration::from_millis(5)).await; + } + + /// `(raw kind byte, window in the datagram, window in the body, seq, + /// which game drew it, frame number)` for every video datagram that + /// arrives before the line goes quiet. + async fn received(conn: &Connection) -> Vec<(u8, u16, u16, u16, u8, u8)> { + let mut out = Vec::new(); + while let Ok(Ok(d)) = + tokio::time::timeout(Duration::from_millis(300), conn.read_datagram()).await + { + let (h, window, fragment) = decode_video_datagram(&d).expect("video"); + assert_eq!((h.index, h.total), (0, 1), "small frames are one datagram"); + let (body_window, seq, payload) = decode_video_frame_body(fragment).expect("body"); + let (_, data) = nesprotocol::decode_video_payload(payload).expect("payload"); + if window == 0 { + assert_eq!(fragment[0], MSG_DATA, "window 0 keeps the old body"); + } + out.push((d[0], window, body_window, seq, data[0], data[1])); + } + out + } + + async fn tell(rig: &Rig, f: impl Fn(&mut Vec)) { + let mut m = Vec::new(); + f(&mut m); + rig.mgr.on_nescope(m).await; + } + + fn opened(window: u16, launch: &str) -> WindowEvent { + WindowEvent::Opened { + window, + width: 64, + height: 64, + launch_id: launch.into(), + title: launch.into(), + app_id: String::new(), + } + } + + #[tokio::test(flavor = "multi_thread")] + async fn an_old_client_sees_the_first_game_and_a_subscriber_its_window() { + let rig = rig().await; + tell(&rig, |m| encode_event_message(m, &opened(0, "launch_a"))).await; + tell(&rig, |m| encode_event_message(m, &opened(1, "launch_b"))).await; + tell(&rig, |m| encode_window_pid(m, 0, GAME_A)).await; + tell(&rig, |m| encode_window_pid(m, 1, GAME_B)).await; + + // Never subscribed: the first game to draw, in the old encoding. + draw(&rig, GAME_A, 0, true).await; + draw(&rig, GAME_B, 0, true).await; + for n in 1..6 { + draw(&rig, GAME_A, n, false).await; + draw(&rig, GAME_B, n, false).await; + } + let got = received(&rig.client).await; + assert_eq!(got.len(), 5, "{got:?}"); + for (i, (kind, window, body_window, seq, game, n)) in got.iter().enumerate() { + assert_eq!((*kind, *window, *body_window), (0, 0, 0)); + assert_eq!((*game, *n), (1, i as u8 + 1), "only game A, in order"); + assert_eq!(*seq, i as u16 + 1, "seq 0 was the keyframe's"); + } + + // Subscribed to window 1: game B only, windowed, numbered afresh. + let (idr, mut idr_rx) = tokio::sync::mpsc::unbounded_channel(); + let me = rig._client_ep.addr().id; + rig.mgr.subscribe(me, 1, &idr).await; + // Only game B, whose window this is, is asked for a keyframe. + assert_eq!( + idr_rx.try_recv().ok(), + Some(crate::capture_cmd::keyframe_request(Some(GAME_B))) + ); + draw(&rig, GAME_B, 10, true).await; + for n in 11..15 { + draw(&rig, GAME_A, n, false).await; + draw(&rig, GAME_B, n, false).await; + } + let got = received(&rig.client).await; + assert_eq!(got.len(), 4, "{got:?}"); + for (i, (kind, window, body_window, seq, game, n)) in got.iter().enumerate() { + assert_eq!((*kind, *window, *body_window), (DGRAM_VIDEO_WINDOW, 1, 1)); + assert_eq!((*game, *n), (2, 11 + i as u8)); + assert_eq!(*seq, i as u16 + 1); + } + + // Its windows stream starts with the box as it stands. + let events = tokio::time::timeout(Duration::from_secs(2), async { + loop { + let mut recv = rig.client.accept_uni().await.expect("uni"); + let mut head = [0u8; 2]; + recv.read_exact(&mut head).await.expect("head"); + if head[0] != STREAM_WINDOWS { + continue; + } + let mut events = Vec::new(); + for _ in 0..2 { + let mut len = [0u8; 4]; + recv.read_exact(&mut len).await.expect("len"); + let mut frame = vec![0u8; u32::from_le_bytes(len) as usize]; + recv.read_exact(&mut frame).await.expect("frame"); + events.push(WindowEvent::decode(frame[0], &frame[3..]).expect("event")); + } + return events; + } + }) + .await + .expect("a windows stream"); + assert_eq!(events, vec![opened(0, "launch_a"), opened(1, "launch_b")]); + } + + #[tokio::test(flavor = "multi_thread")] + async fn a_primary_switch_starts_with_a_keyframe_and_a_fresh_sequence() { + let rig = rig().await; + draw(&rig, GAME_A, 0, true).await; + for n in 1..4 { + draw(&rig, GAME_A, n, false).await; + } + let got = received(&rig.client).await; + assert!(got.iter().all(|g| g.4 == 1), "{got:?}"); + // A stops; B draws deltas for long enough to take over. + let start = Instant::now(); + let mut n = 0u8; + while start.elapsed() < Duration::from_millis(1400) { + draw(&rig, GAME_B, n, false).await; + n = n.wrapping_add(1); + tokio::time::sleep(Duration::from_millis(15)).await; + } + // Deltas from the new encoder are held until its keyframe. + draw(&rig, GAME_B, 200, true).await; + for n in 201..204 { + draw(&rig, GAME_B, n, false).await; + } + let got = received(&rig.client).await; + assert_eq!(got.len(), 3, "only the deltas after B's keyframe: {got:?}"); + for (i, g) in got.iter().enumerate() { + assert_eq!((g.0, g.4, g.5), (0, 2, 201 + i as u8)); + assert_eq!(g.3, i as u16 + 1, "numbered afresh after the keyframe's 0"); + } + } + + #[tokio::test(flavor = "multi_thread")] + async fn one_untagged_game_is_the_old_stream() { + // An encoder that predates tagging, a nescope that predates windows. + let rig = rig().await; + let send = |n: u8, key: bool| { + let flags = if key { FLAG_KEYFRAME } else { 0 }; + nesprotocol::encode_ipc_frame(STREAM_VIDEO, CODEC_H264, flags, 1, 64, 64, &[1, n]) + }; + rig.sender.send(&send(0, true)).await.unwrap(); + for n in 1..4 { + tokio::time::sleep(Duration::from_millis(5)).await; + rig.sender.send(&send(n, false)).await.unwrap(); + } + let got = received(&rig.client).await; + assert_eq!(got.len(), 3); + assert!(got.iter().all(|g| (g.0, g.1) == (0, 0))); + } +} diff --git a/apps/neshub/src/ipc_listener.rs b/apps/neshub/src/ipc_listener.rs index 1e9aae9f..99511e0b 100644 --- a/apps/neshub/src/ipc_listener.rs +++ b/apps/neshub/src/ipc_listener.rs @@ -22,11 +22,19 @@ fn set_recv_buffer(socket: &UnixDatagram, size: libc::c_int) { } } +/// Encoder processes the video listener has reported to the command +/// forwarder, ref(d-0085). +pub type KnownEncoders = Arc>>; + pub async fn run_video_listener( socket_path: PathBuf, session_manager: Arc, controller: Arc>, cmd_tx: tokio::sync::mpsc::UnboundedSender>, + // Each encoder process the first time it tags a frame, so commands can + // reach it on its own socket, ref(d-0085). + encoder_tx: tokio::sync::mpsc::UnboundedSender, + known_encoders: KnownEncoders, ) { if socket_path.exists() { let _ = std::fs::remove_file(&socket_path); @@ -60,6 +68,13 @@ pub async fn run_video_listener( match socket.recv(&mut buf).await { Ok(n) => { if let Some(decoded) = decode_ipc_frame(&buf[..n]) { + if decoded.stream_type == nesprotocol::STREAM_ENCODER_HELLO { + if let Some(pid) = decoded.pid { + let launch = String::from_utf8_lossy(decoded.data).into_owned(); + session_manager.note_encoder_launch(pid, launch); + } + continue; + } if decoded.stream_type == STREAM_PYROWAVE { let now_ms = started.elapsed().as_millis() as u64; match pyrowave.push(&decoded, now_ms) { @@ -82,6 +97,12 @@ pub async fn run_video_listener( ); continue; } + if let Some(pid) = decoded.pid + && known_encoders.lock().is_ok_and(|mut k| k.insert(pid)) + { + tracing::info!("encoder in process {pid} is drawing"); + let _ = encoder_tx.send(pid); + } // An encoder on a codec nobody here decodes is told again. session_manager .reassert_codec(decoded.codec, &controller, &cmd_tx) @@ -97,7 +118,9 @@ pub async fn run_video_listener( decoded.data, ); - session_manager.broadcast_video(frame_data).await; + session_manager + .broadcast_video(frame_data, decoded.pid) + .await; } else { tracing::warn!("invalid video IPC frame ({} bytes)", n); } @@ -171,9 +194,33 @@ pub async fn run_audio_listener(socket_path: PathBuf, session_manager: Arc bool { + use nesprotocol::window::*; + matches!( + kind, + nesprotocol::input::CURSOR_UPDATE + | nesprotocol::input::CURSOR_IMAGE + | MSG_WINDOW_OPENED + | MSG_WINDOW_TITLE + | MSG_WINDOW_ICON + | MSG_WINDOW_SIZE + | MSG_WINDOW_CLOSED + | MSG_WINDOW_PID + | MSG_CURSOR_WINDOW + ) +} + +/// nescope's socket: input to it, cursor, windows and stats from it. +/// +/// `cursor_tx` carries everything about the cursor and the windows, in order, +/// and an empty message each time nescope connects: a new connection is a new +/// compositor, whose window ids are not the last one's. pub async fn run_input_ipc_listener( socket_path: PathBuf, - input_tx: tokio::sync::broadcast::Sender>, + input_tx: tokio::sync::broadcast::Sender, cursor_tx: tokio::sync::mpsc::UnboundedSender>, nescope_stats_tx: tokio::sync::mpsc::UnboundedSender>, ) { @@ -210,19 +257,25 @@ pub async fn run_input_ipc_listener( let (mut read_half, mut write_half) = stream.into_split(); let mut input_rx = input_tx.subscribe(); - let _c_tx = cursor_tx.clone(); + let _ = cursor_tx.send(Vec::new()); let write_handle = tokio::spawn(async move { - loop { + // Fresh per connection: nescope starts every connection + // on window 0. + let mut tagger = crate::windows::FocusTagger::default(); + 'events: loop { match input_rx.recv().await { - Ok(data) => { - let len = data.len() as u16; - let mut frame = Vec::with_capacity(2 + data.len()); - frame.extend_from_slice(&len.to_le_bytes()); - frame.extend_from_slice(&data); - if write_half.write_all(&frame).await.is_err() { - tracing::debug!("input IPC write failed"); - break; + Ok((client, window, event)) => { + let now = std::time::Instant::now(); + for data in tagger.messages(client, window, &event, now) { + let len = data.len() as u16; + let mut frame = Vec::with_capacity(2 + data.len()); + frame.extend_from_slice(&len.to_le_bytes()); + frame.extend_from_slice(&data); + if write_half.write_all(&frame).await.is_err() { + tracing::debug!("input IPC write failed"); + break 'events; + } } } Err(RecvError::Lagged(n)) => { @@ -252,11 +305,9 @@ pub async fn run_input_ipc_listener( if read_half.read_exact(&mut payload).await.is_err() { break; } - // Route: cursor updates start with 0x80, stats start with 0x00-0x02 - if !payload.is_empty() - && (payload[0] == nesprotocol::input::CURSOR_UPDATE - || payload[0] == nesprotocol::input::CURSOR_IMAGE) - { + // Route: cursor and window messages in order on one + // channel; stats start with 0x00-0x02. + if payload.first().is_some_and(|k| is_cursor_or_window(*k)) { let _ = c_tx.send(payload); } else { let _ = ns_tx.send(payload); diff --git a/apps/neshub/src/keyframe.rs b/apps/neshub/src/keyframe.rs index a0d8cd4d..b774b104 100644 --- a/apps/neshub/src/keyframe.rs +++ b/apps/neshub/src/keyframe.rs @@ -50,7 +50,7 @@ impl KeyframeSender { /// stream writes are flow-controlled, so a receiver with no window left /// would otherwise hold up every delta frame queued behind this keyframe — /// and the deltas are the frames with an expiry date. - pub fn send(&self, seq: u16, body: &[u8]) -> bool { + pub fn send(&self, seq: u16, window: nesprotocol::window::WindowId, body: &[u8]) -> bool { if self.in_flight.load(Ordering::Relaxed) >= MAX_KEYFRAME_STREAMS_IN_FLIGHT { self.fell_back.fetch_add(1, Ordering::Relaxed); return false; @@ -79,7 +79,7 @@ impl KeyframeSender { // are a worse path for a keyframe, but they beat dropping the one // frame everything after it is predicted from. fell_back.fetch_add(1, Ordering::Relaxed); - if let Err(e) = fallback.send_frame(seq, &body) { + if let Err(e) = fallback.send_frame(seq, window, &body) { debug!("keyframe {seq}: stream failed and datagram fallback failed too: {e}"); } }); diff --git a/apps/neshub/src/main.rs b/apps/neshub/src/main.rs index b4716bf1..17baae06 100644 --- a/apps/neshub/src/main.rs +++ b/apps/neshub/src/main.rs @@ -11,6 +11,8 @@ mod rate; mod screenshot; mod session; mod ticket; +mod windows; +mod capture_cmd; use std::net::{Ipv4Addr, SocketAddrV4}; use std::path::PathBuf; @@ -234,7 +236,7 @@ async fn main() -> Result<()> { tracing::info!("endpoint online: {}", ep_id.fmt_short()); // Input broadcast channel: input reader -> input IPC listener -> nescope - let (input_broadcast_tx, _) = tokio::sync::broadcast::channel::>(256); + let (input_broadcast_tx, _) = tokio::sync::broadcast::channel::(256); // Cursor channel: IPC listener (read side) -> client sessions -> desktop-app let (cursor_tx, mut cursor_rx) = tokio::sync::mpsc::unbounded_channel::>(); @@ -255,15 +257,69 @@ async fn main() -> Result<()> { // IDR / encode settings command channel: input reader → nescapture let (cmd_tx, mut cmd_rx) = tokio::sync::mpsc::unbounded_channel::>(); + // Encoder processes as the video listener first hears from them. + let (encoder_tx, mut encoder_rx) = tokio::sync::mpsc::unbounded_channel::(); + // Encoders the video listener has reported. One the forwarder finds gone + // is taken out, so its next frame reports it again. + let known_encoders: ipc_listener::KnownEncoders = Default::default(); { + let known = known_encoders.clone(); let mgr = session_manager.clone(); tokio::spawn(async move { - let cmd_path = std::path::PathBuf::from("/tmp/nescapture-cmd.sock"); - while let Some(mut bytes) = cmd_rx.recv().await { + let cmd_path = std::path::PathBuf::from(nesprotocol::CAPTURE_CMD_SOCKET); + let mut fanout = capture_cmd::Fanout::default(); + loop { + let bytes = tokio::select! { + Some(pid) = encoder_rx.recv() => { + // Every encoder the hub knows hears every command, on + // its own socket; a new one first hears what it + // missed, ref(d-0085). + for mut cmd in fanout.encoder(pid).unwrap_or_default() { + fill_pyrowave_packet_size(&mut cmd, mgr.pyro_packet_size().await); + let path = nesprotocol::capture_cmd_socket(pid); + if let Ok(sock) = std::os::unix::net::UnixDatagram::unbound() + && let Err(e) = sock.send_to(&cmd, &path) + { + tracing::debug!("telling encoder {pid} what it missed: {e}"); + } + } + continue; + } + cmd = cmd_rx.recv() => match cmd { + Some(cmd) => cmd, + None => break, + }, + }; + // A keyframe request for one window goes to that window's + // encoder alone; everything else to all of them. + let (target, mut bytes) = capture_cmd::target_of(bytes); fill_pyrowave_packet_size(&mut bytes, mgr.pyro_packet_size().await); let Ok(sock) = std::os::unix::net::UnixDatagram::unbound() else { continue; }; + let destinations = fanout.destinations_for(target); + if target.is_none() { + fanout.sent(&bytes); + } + if destinations.first() != Some(&cmd_path) { + for path in destinations { + match sock.send_to(&bytes, &path) { + Ok(_) => {} + // Refused or absent: that process has gone. + Err(e) if capture_cmd::encoder_is_gone(&e) => { + tracing::info!("encoder at {} is gone: {e}", path.display()); + if let Some(pid) = capture_cmd::pid_of(&path) { + fanout.forget(pid); + if let Ok(mut known) = known.lock() { + known.remove(&pid); + } + } + } + Err(e) => tracing::debug!("command to {}: {e}", path.display()), + } + } + continue; + } // The encoder's socket is not there until the game draws its // first frame, and a client can connect and say what it decodes // before that. A command sent then used to be lost -- the codec @@ -292,15 +348,16 @@ async fn main() -> Result<()> { }); } - // Spawn cursor relay + // Cursor and window relay { let mgr = session_manager.clone(); tokio::spawn(async move { while let Some(data) = cursor_rx.recv().await { - mgr.broadcast_cursor(data).await; + mgr.on_nescope(data).await; } }); } + tokio::spawn(session_manager.clone().serve_subscriptions(cmd_tx.clone())); // Spawn nescope stats relay { @@ -455,7 +512,17 @@ async fn main() -> Result<()> { tokio::spawn({ let (mgr, controller, cmd_tx) = (session_manager.clone(), controller.clone(), cmd_tx.clone()); - async move { ipc_listener::run_video_listener(video_ipc, mgr, controller, cmd_tx).await } + async move { + ipc_listener::run_video_listener( + video_ipc, + mgr, + controller, + cmd_tx, + encoder_tx, + known_encoders, + ) + .await + } }); tokio::spawn({ let mgr = session_manager.clone(); diff --git a/apps/neshub/src/session.rs b/apps/neshub/src/session.rs index 5f7b7838..f8d2aac8 100644 --- a/apps/neshub/src/session.rs +++ b/apps/neshub/src/session.rs @@ -20,12 +20,41 @@ use nesprotocol::{ use crate::control::{Controller, PathView}; use crate::negotiation::{Negotiator, effective_caps}; +use crate::capture_cmd::keyframe_request; use crate::dgram::{MediaItem, run_datagram_writer}; +use crate::windows::{Source, Subscription, WindowTable, route, wants_cursor}; +use nesprotocol::window::{MSG_WINDOW_SUBSCRIBE, STREAM_WINDOWS, WINDOW_DEFAULT, WindowEvent, WindowId}; use bytes::Bytes; use nesprotocol::pyrowave::PyroFrame; use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}; +/// Input on its way to the box, ref(d-0085): which client sent it (its +/// gamepad session number), the window it belongs to, and the event. An empty +/// event asks only that the window be focused, which the box's side grants +/// only when no other client is typing; see `windows::FocusTagger`. +pub type TaggedInput = (u32, WindowId, Vec); + +/// What a client's input reader needs to say which window its input is for. +#[derive(Clone)] +struct InputWindow { + subscription: Arc>, + /// The window of the box's primary picture, which is what a client that + /// never subscribed is looking at. + primary: Arc, +} + +impl InputWindow { + /// `None` while subscribed to a window that no longer exists. + fn current(&self) -> Option { + match *self.subscription.lock().ok()? { + Subscription::Legacy => Some(self.primary.load(Ordering::Relaxed) as WindowId), + Subscription::Window(w) => Some(w), + Subscription::Torn => None, + } + } +} + /// One client, across the several connections it opens. /// /// A connection is the unit congestion control, pacing and the datagram send @@ -71,7 +100,18 @@ pub struct ClientSession { /// The rate a PyroWave request that names none runs at. pyrowave_kbps: u32, relay_ms: Arc, - input_broadcast: tokio::sync::broadcast::Sender>, + input_broadcast: tokio::sync::broadcast::Sender, + /// Which window this client is shown, ref(d-0085). + subscription: Arc>, + /// The encoder whose frames this client was last sent (0: none yet), so + /// its keyframe requests go to that encoder alone. + encoder_pid: Arc, + primary_window: Arc, + /// Where a subscribe request goes: the manager, which holds the windows. + subscribe_tx: tokio::sync::mpsc::UnboundedSender<(iroh::EndpointId, WindowId)>, + /// Lifecycle events for this client's windows stream, `[type][payload]`. + send_windows: tokio::sync::mpsc::UnboundedSender>, + pending_windows: Option>>, /// This client's number on the gamepad socket, so the box's side can tell /// two clients' controllers apart. Assigned by the manager, never reused /// within one hub's lifetime. @@ -115,8 +155,12 @@ impl Drop for ClientSession { impl ClientSession { /// A client with no connections yet. They attach as they are accepted. + #[allow(clippy::too_many_arguments)] pub fn new( - input_broadcast: tokio::sync::broadcast::Sender>, + input_broadcast: tokio::sync::broadcast::Sender, + primary_window: Arc, + encoder_pid: u32, + subscribe_tx: tokio::sync::mpsc::UnboundedSender<(iroh::EndpointId, WindowId)>, gamepad_session: u32, gamepad_tx: tokio::sync::broadcast::Sender>, relay_ms: Arc, @@ -131,7 +175,14 @@ impl ClientSession { let (audio_tx, audio_rx) = tokio::sync::mpsc::unbounded_channel::(); let (cursor_tx, cursor_rx) = tokio::sync::mpsc::unbounded_channel::>(); let (stats_tx, stats_rx) = tokio::sync::mpsc::unbounded_channel::>(); + let (windows_tx, windows_rx) = tokio::sync::mpsc::unbounded_channel::>(); Self { + subscription: Arc::new(std::sync::Mutex::new(Subscription::Legacy)), + encoder_pid: Arc::new(AtomicU32::new(encoder_pid)), + primary_window, + subscribe_tx, + send_windows: windows_tx, + pending_windows: Some(windows_rx), video_conn: None, other_conns: Vec::new(), awaiting_keyframe: Arc::new(AtomicBool::new(false)), @@ -189,7 +240,9 @@ impl ClientSession { // says so, a round trip later; and deltas are held back until // it is sent, as for any client waiting on one. self.awaiting_keyframe.store(true, Ordering::Relaxed); - let _ = self.idr_cmd_tx.send(vec![MSG_IDR_REQUEST]); + let _ = self + .idr_cmd_tx + .send(keyframe_request(pid_or_all(&self.encoder_pid))); // The client says how each frame arrived, back on this same // connection; the rate is decided on each one as it comes. self.tasks.push(tokio::spawn(run_feedback_reader( @@ -198,6 +251,12 @@ impl ClientSession { self.controller.clone(), self.idr_cmd_tx.clone(), ))); + // Lifecycle events, on a stream opened only once there is one + // to send, which is only once this client has subscribed. + if let Some(windows) = self.pending_windows.take() { + self.tasks + .push(tokio::spawn(run_windows_sender(conn.clone(), windows))); + } let relay_ms = self.relay_ms.clone(); let awaiting = self.awaiting_keyframe.clone(); let withheld = self.withheld.clone(); @@ -254,8 +313,12 @@ impl ClientSession { session: self.gamepad_session, to_box: self.gamepad_tx.clone(), }; + let window = InputWindow { + subscription: self.subscription.clone(), + primary: self.primary_window.clone(), + }; self.tasks.push(tokio::spawn(async move { - run_input_reader(conn, broadcast, gamepad, feedback).await + run_input_reader(conn, broadcast, window, gamepad, feedback).await })); } Carrier::Control => { @@ -286,9 +349,16 @@ impl ClientSession { client: self.client, negotiator: self.negotiator.clone(), }; + let subscribe = ( + self.client, + self.subscribe_tx.clone(), + self.encoder_pid.clone(), + ); self.tasks.push(tokio::spawn(async move { - run_control_reader(conn, idr, reports, controller, awaiting, pyrowave, caps) - .await + run_control_reader( + conn, idr, reports, controller, awaiting, pyrowave, caps, subscribe, + ) + .await })); } } @@ -402,6 +472,60 @@ impl ClientSession { debug!("failed to send stats data: {e}"); } } + + /// Which window this client is shown. + pub fn subscription(&self) -> Subscription { + self.subscription + .lock() + .map(|s| *s) + .unwrap_or(Subscription::Legacy) + } + + /// Show this client another window, and return what it was shown before. + /// + /// In the order the protocol asks for: the switch first, so the next frame + /// routed to this client is the new window's, then the writer's numbering + /// and resynchronisation reset in line with the frames, then a keyframe + /// requested, since nothing the client holds predicts the new window. A + /// keyframe requested before the switch could arrive for the old one. + pub fn subscribe( + &self, + to: Subscription, + encoder: Option, + idr_cmd_tx: &tokio::sync::mpsc::UnboundedSender>, + ) -> Subscription { + let before = match self.subscription.lock() { + Ok(mut s) => std::mem::replace(&mut *s, to), + Err(_) => return Subscription::Legacy, + }; + if before == to { + return before; + } + self.restart_picture(); + self.encoder_pid.store(encoder.unwrap_or(0), Ordering::Relaxed); + if matches!(to, Subscription::Window(_)) { + let _ = idr_cmd_tx.send(keyframe_request(encoder)); + } + before + } + + /// What this client is shown is now another encoder's picture: number from + /// zero, hold deltas until a keyframe, and forget what it reported about + /// the old one. The caller asks for the keyframe. + pub fn restart_picture(&self) { + if self.pending_video.is_none() { + let _ = self.send_video.send(MediaItem::Resubscribe); + } + self.awaiting_keyframe.store(true, Ordering::Relaxed); + let _ = self.take_report(); + } + + /// Queue one lifecycle message, `[type][payload]`, for the windows stream. + pub fn send_window_message(&self, message: Vec) { + if let Err(e) = self.send_windows.send(message) { + debug!("failed to queue a window event: {e}"); + } + } } #[allow(clippy::too_many_arguments)] @@ -541,6 +665,20 @@ fn split_input_events(payload: &[u8]) -> Vec> { out } +/// How often a client's gamepad may ask for its window's focus. Only a bound +/// on the asking: a request for the window that already has focus changes +/// nothing. +const GAMEPAD_FOCUS_ASK: std::time::Duration = std::time::Duration::from_millis(100); + +/// The encoder a client's keyframe requests are for, or every encoder when +/// it has not been sent anyone's frames yet. +fn pid_or_all(pid: &AtomicU32) -> Option { + match pid.load(Ordering::Relaxed) { + 0 => None, + pid => Some(pid), + } +} + /// Where one client's gamepad messages go. struct Gamepads { session: u32, @@ -554,25 +692,50 @@ struct Gamepads { /// controller is and what to make of it is the box's side to decide, not the /// hub's. The stream's return half carries that side's feedback -- rumble -- /// back. +/// +/// Each event goes to the box with the window this client is shown, which is +/// where the box's focus tags come from, ref(d-0085). A gamepad message also +/// asks for that window's focus: the pads reach the game by another route, +/// and a game reads them only while it has focus. Asked at most every +/// [`GAMEPAD_FOCUS_ASK`]; whether it is granted is the tagger's decision. async fn run_input_reader( conn: Connection, - input_broadcast: tokio::sync::broadcast::Sender>, + input_broadcast: tokio::sync::broadcast::Sender, + window: InputWindow, gamepads: Gamepads, feedback: tokio::sync::mpsc::UnboundedReceiver>, ) { let outbound = Some((MSG_GAMEPAD_FEEDBACK, feedback)); + let mut focus_asked: Option = None; run_framed_reader(conn, BIDI_INPUT, "input", outbound, |msg_type, payload| { let input_broadcast = input_broadcast.clone(); let to_box = gamepads.to_box.clone(); let session = gamepads.session; + let target = window.current(); + let mut ask_focus = None; + if msg_type == MSG_GAMEPAD + && let Some(w) = target + { + let now = std::time::Instant::now(); + if focus_asked.is_none_or(|at| now.duration_since(at) >= GAMEPAD_FOCUS_ASK) { + focus_asked = Some(now); + ask_focus = Some(w); + } + } async move { match msg_type { MSG_INPUT_BATCH => { + // Nothing to deliver to: the window this client was shown + // went with its compositor. + let Some(w) = target else { return }; for event in split_input_events(&payload) { - let _ = input_broadcast.send(event); + let _ = input_broadcast.send((session, w, event)); } } MSG_GAMEPAD => { + if let Some(w) = ask_focus { + let _ = input_broadcast.send((session, w, Vec::new())); + } let mut frame = Vec::with_capacity(6 + payload.len()); nesprotocol::gamepad::encode_ipc(&mut frame, session, &payload); // An error is nobody listening: the box's gamepad side is @@ -633,6 +796,7 @@ struct CapsSink { } /// Everything the client says that is not an input event. +#[allow(clippy::too_many_arguments)] async fn run_control_reader( conn: Connection, idr_cmd_tx: tokio::sync::mpsc::UnboundedSender>, @@ -641,6 +805,13 @@ async fn run_control_reader( awaiting_keyframe: Arc, pyrowave: PyroWaveGate, caps_sink: CapsSink, + // Who this is, where subscribe requests go, and the encoder this + // client's keyframe requests are for. + subscribe: ( + iroh::EndpointId, + tokio::sync::mpsc::UnboundedSender<(iroh::EndpointId, WindowId)>, + Arc, + ), ) { run_framed_reader(conn, BIDI_CONTROL, "control", None, |msg_type, payload| { let idr_cmd_tx = idr_cmd_tx.clone(); @@ -649,8 +820,16 @@ async fn run_control_reader( let awaiting_keyframe = awaiting_keyframe.clone(); let pyrowave = pyrowave.clone(); let caps_sink = caps_sink.clone(); + let subscribe = subscribe.clone(); async move { match msg_type { + MSG_WINDOW_SUBSCRIBE => match nesprotocol::window::decode_subscribe(&payload) { + Some(window) => { + info!("client subscribes to window {window}"); + let _ = subscribe.1.send((subscribe.0, window)); + } + None => debug!("unreadable window subscribe ({} bytes)", payload.len()), + }, MSG_IDR_REQUEST => { // Client-side rate limited to one every two seconds, which // is still far too often for info when a struggling @@ -660,7 +839,9 @@ async fn run_control_reader( // undecodable and starves the keyframe that would fix it. // See `dgram::ResyncGate`. awaiting_keyframe.store(true, Ordering::Relaxed); - let _ = idr_cmd_tx.send(vec![MSG_IDR_REQUEST]); + // To the encoder of the window this client is shown, + // ref(d-0085): the other windows are fine. + let _ = idr_cmd_tx.send(keyframe_request(pid_or_all(&subscribe.2))); } MSG_CLIENT_CAPS => { // Passed through untouched, and in particular *without* @@ -771,6 +952,45 @@ async fn run_control_reader( .await; } +/// The windows stream, ref(d-0085): `[STREAM_WINDOWS][STREAM_VERSION]` then +/// one frame per lifecycle event. Opened on the first event, so a client that +/// never subscribes never sees it. +async fn run_windows_sender( + conn: Connection, + mut rx: tokio::sync::mpsc::UnboundedReceiver>, +) { + let Some(first) = rx.recv().await else { + return; + }; + let mut send = match conn.open_uni().await { + Ok(s) => s, + Err(e) => { + debug!("windows open_uni failed: {e}"); + return; + } + }; + if send + .write_all(&[STREAM_WINDOWS, STREAM_VERSION]) + .await + .is_err() + { + return; + } + let mut buf = Vec::new(); + let mut next = Some(first); + while let Some(message) = next { + if let Some((&msg_type, payload)) = message.split_first() { + buf.clear(); + encode_frame(&mut buf, msg_type, 0, payload); + if send.write_all(&buf).await.is_err() { + return; + } + } + next = rx.recv().await; + } + let _ = send.finish(); +} + async fn run_cursor_sender( conn: Connection, mut rx: tokio::sync::mpsc::UnboundedReceiver>, @@ -976,12 +1196,34 @@ pub struct SessionManager { pyrowave_kbps: u32, /// What every connected client decodes, and so what the encoder may pick. negotiator: Arc, + /// The box's windows and which encoder draws which, ref(d-0085). + windows: std::sync::Mutex, + /// The window of the primary picture, for input from clients that never + /// subscribed. + primary_window: Arc, + /// The window the cursor updates describe, as nescope last said. + cursor_window: AtomicU32, + subscribe_tx: tokio::sync::mpsc::UnboundedSender<(iroh::EndpointId, WindowId)>, + subscribe_rx: + std::sync::Mutex>>, + /// The encoder of the primary picture (0: untagged or none yet). + primary_pid: Arc, + /// Where commands for the encoders go; set by `serve_subscriptions`. + cmd_tx: std::sync::OnceLock>>, } impl SessionManager { pub fn new(pyrowave_kbps: u32) -> Self { let (gamepad_tx, _) = tokio::sync::broadcast::channel(256); + let (subscribe_tx, subscribe_rx) = tokio::sync::mpsc::unbounded_channel(); Self { + windows: std::sync::Mutex::new(WindowTable::default()), + primary_window: Arc::new(AtomicU32::new(0)), + cursor_window: AtomicU32::new(0), + subscribe_tx, + subscribe_rx: std::sync::Mutex::new(Some(subscribe_rx)), + primary_pid: Arc::new(AtomicU32::new(0)), + cmd_tx: std::sync::OnceLock::new(), sessions: Arc::new(Mutex::new(HashMap::new())), gamepad_tx, next_gamepad_session: AtomicU32::new(1), @@ -1011,7 +1253,7 @@ impl SessionManager { id: iroh::EndpointId, carrier: Carrier, conn: Connection, - input_broadcast: tokio::sync::broadcast::Sender>, + input_broadcast: tokio::sync::broadcast::Sender, idr_cmd_tx: tokio::sync::mpsc::UnboundedSender>, controller: Arc>, ) { @@ -1020,6 +1262,9 @@ impl SessionManager { let session = sessions.entry(id).or_insert_with(|| { ClientSession::new( input_broadcast, + self.primary_window.clone(), + self.primary_pid.load(Ordering::Relaxed), + self.subscribe_tx.clone(), self.next_gamepad_session.fetch_add(1, Ordering::Relaxed), self.gamepad_tx.clone(), self.relay_ms.clone(), @@ -1143,7 +1388,8 @@ impl SessionManager { .unwrap_or((0, 0, 0)) } - pub async fn broadcast_video(&self, data: Vec) { + /// Hand a frame from `source` to every client shown its window. + pub async fn broadcast_video(&self, data: Vec, source: Source) { self.note_pipeline_delay(&data); // Counted before the early return, like audio, so the figure measures // what the encoder produced rather than what a client happened to be @@ -1156,10 +1402,202 @@ impl SessionManager { self.video_delta_bytes .fetch_add(data.len() as u64, Ordering::Relaxed); } - let data = MediaItem::Coded(Bytes::from(data)); + let (window, note) = { + let mut windows = self.windows.lock().unwrap_or_else(|e| e.into_inner()); + let window = windows.resolve(source); + if let (Some(w), Some(pid)) = (window, source) { + windows.note_drawn(w, pid); + } + let note = windows.note_primary(source, std::time::Instant::now()); + if note.primary { + self.primary_window.store( + u32::from(window.unwrap_or(WINDOW_DEFAULT)), + Ordering::Relaxed, + ); + self.primary_pid.store(source.unwrap_or(0), Ordering::Relaxed); + } + (window, note) + }; + let data = Bytes::from(data); + let sessions = self.sessions.lock().await; + if note.switched { + // Clients that never subscribed are now shown another encoder: + // nothing they hold predicts it, ref(d-0085). + info!("primary picture is now encoder {source:?}"); + let mut any = false; + for session in sessions.values() { + if session.subscription() == Subscription::Legacy { + session.restart_picture(); + session + .encoder_pid + .store(source.unwrap_or(0), Ordering::Relaxed); + any = true; + } + } + if any && let Some(tx) = self.cmd_tx.get() { + let _ = tx.send(keyframe_request(source)); + } + } + for session in sessions.values() { + match route(session.subscription(), window, note.primary) { + Some(WINDOW_DEFAULT) => session.send_video_frame(MediaItem::Coded(data.clone())), + Some(w) => session.send_video_frame(MediaItem::Windowed(w, data.clone())), + None => {} + } + } + } + + /// Encoder `pid` said which launch it draws for, ref(d-0085). + pub fn note_encoder_launch(&self, pid: u32, launch: String) { + self.windows + .lock() + .unwrap_or_else(|e| e.into_inner()) + .note_launch(pid, launch); + } + + /// One message from nescope on the input socket that is not stats: + /// cursor updates and the box's windows, in the order nescope sent them. + /// An empty one means nescope has (re)connected. + pub async fn on_nescope(&self, message: Vec) { + use nesprotocol::window::{MSG_CURSOR_WINDOW, MSG_WINDOW_PID, decode_window_pid}; + let Some(&kind) = message.first() else { + self.compositor_restarted().await; + return; + }; + match kind { + nesprotocol::input::CURSOR_UPDATE | nesprotocol::input::CURSOR_IMAGE => { + self.broadcast_cursor(message).await + } + MSG_CURSOR_WINDOW => match nesprotocol::window::decode_subscribe(&message[1..]) { + Some(w) => self.cursor_window.store(u32::from(w), Ordering::Relaxed), + None => debug!("unreadable cursor window"), + }, + MSG_WINDOW_PID => match decode_window_pid(&message[1..]) { + Some((window, pid)) => { + debug!("window {window} is process {pid}"); + self.windows + .lock() + .unwrap_or_else(|e| e.into_inner()) + .bind_pid(window, pid); + } + None => debug!("unreadable window pid"), + }, + _ => match WindowEvent::decode(kind, &message[1..]) { + Some(event) => self.window_event(event, message).await, + None => debug!("unknown message {kind:#04x} from nescope"), + }, + } + } + + /// Record a lifecycle event and pass it to every client with a windows + /// stream. + async fn window_event(&self, event: WindowEvent, message: Vec) { + match &event { + WindowEvent::Opened { window, launch_id, title, .. } => { + info!("window {window} opened: launch {launch_id:?}, {title:?}") + } + WindowEvent::Closed { window } => info!("window {window} closed"), + _ => {} + } + self.windows + .lock() + .unwrap_or_else(|e| e.into_inner()) + .apply(&event); let sessions = self.sessions.lock().await; for session in sessions.values() { - session.send_video_frame(data.clone()); + if session.subscription() != Subscription::Legacy { + session.send_window_message(message.clone()); + } + } + } + + /// nescope connected again, so the box's compositor is a new one and every + /// window id before now means nothing: say each window closed, and end + /// every subscription to one. + async fn compositor_restarted(&self) { + let closed = self + .windows + .lock() + .unwrap_or_else(|e| e.into_inner()) + .reset(); + self.cursor_window.store(0, Ordering::Relaxed); + if closed.is_empty() { + return; + } + info!("compositor restarted; {} window(s) gone", closed.len()); + let sessions = self.sessions.lock().await; + for session in sessions.values() { + let sub = session.subscription(); + if sub == Subscription::Legacy { + continue; + } + for window in &closed { + let mut m = Vec::new(); + nesprotocol::window::encode_event_message( + &mut m, + &WindowEvent::Closed { window: *window }, + ); + session.send_window_message(m); + } + if let Subscription::Window(_) = sub + && let Ok(mut s) = session.subscription.lock() + { + *s = Subscription::Torn; + } + } + } + + /// A client asked to be shown `window`. + /// + /// The first subscription opens its windows stream, starting with every + /// window already open so a late client learns the box as it stands. + pub async fn subscribe( + &self, + client: iroh::EndpointId, + window: WindowId, + idr_cmd_tx: &tokio::sync::mpsc::UnboundedSender>, + ) { + let snapshot = self + .windows + .lock() + .unwrap_or_else(|e| e.into_inner()) + .snapshot(); + let sessions = self.sessions.lock().await; + let Some(session) = sessions.get(&client) else { + return; + }; + let encoder = self + .windows + .lock() + .unwrap_or_else(|e| e.into_inner()) + .encoder_of(window); + let before = session.subscribe(Subscription::Window(window), encoder, idr_cmd_tx); + if before == Subscription::Legacy { + for event in snapshot { + let mut m = Vec::new(); + nesprotocol::window::encode_event_message(&mut m, &event); + session.send_window_message(m); + } + } + } + + /// Serve subscribe requests from every client's control reader. Run once, + /// on a task of its own. + pub async fn serve_subscriptions( + self: Arc, + idr_cmd_tx: tokio::sync::mpsc::UnboundedSender>, + ) { + let _ = self.cmd_tx.set(idr_cmd_tx.clone()); + let rx = self + .subscribe_rx + .lock() + .unwrap_or_else(|e| e.into_inner()) + .take(); + let Some(mut rx) = rx else { + return; + }; + while let Some((client, window)) = rx.recv().await { + self.subscribe(client, window, &idr_cmd_tx).await; } } @@ -1174,8 +1612,14 @@ impl SessionManager { .fetch_add(frame.data.len() as u64, Ordering::Relaxed); let item = MediaItem::PyroWave(Arc::new(frame)); let sessions = self.sessions.lock().await; + // PyroWave has no windowed form: it is window 0's, ref(d-0085). for session in sessions.values() { - session.send_video_frame(item.clone()); + if matches!( + session.subscription(), + Subscription::Legacy | Subscription::Window(WINDOW_DEFAULT) + ) { + session.send_video_frame(item.clone()); + } } } @@ -1217,8 +1661,12 @@ impl SessionManager { if sessions.is_empty() { return; } + // To the clients shown the window the cursor describes, ref(d-0085). + let window = self.cursor_window.load(Ordering::Relaxed) as WindowId; for session in sessions.values() { - session.send_cursor_data(data.clone()); + if wants_cursor(session.subscription(), window) { + session.send_cursor_data(data.clone()); + } } } diff --git a/apps/neshub/src/windows.rs b/apps/neshub/src/windows.rs new file mode 100644 index 00000000..f71a862c --- /dev/null +++ b/apps/neshub/src/windows.rs @@ -0,0 +1,651 @@ +//! Which window a frame, a cursor update or an input event belongs to, +//! ref(d-0085). +//! +//! The box tells the hub about windows on the input socket: nescope mints the +//! ids and says which process owns each one. Encoders tag their frames with +//! their process id. This module joins the two, decides which connection is +//! shown what, and tags input on its way back to the box. +//! +//! The compatibility rule everything here follows: a connection that never +//! subscribes is shown what a client saw before windows existed, in the old +//! encoding. That is the box's *primary* picture: one encoder's frames, +//! replaced only after it has gone silent, by an encoder that has been drawing +//! for a while, with a keyframe and a fresh sequence on every switch. With one +//! game in the box it is that game, whatever window id nescope gave it, and +//! the bytes are the ones the client always received. + +use std::collections::{BTreeMap, HashMap}; +use std::time::{Duration, Instant}; + +use nesprotocol::input::{INPUT_OUTPUT_SIZE, encode_window_focus, encode_window_output_size}; +use nesprotocol::window::{WINDOW_DEFAULT, WindowEvent, WindowId}; + +/// Who drew a frame: an encoder's process id, or `None` for an encoder that +/// predates tagging, which is the box's first window. +pub type Source = Option; + +/// How long the primary encoder may go without a frame before another one +/// takes its place for connections that never subscribed. +pub const PRIMARY_SILENCE: Duration = Duration::from_secs(1); + +/// How long an encoder must have been drawing, without a gap of +/// [`PRIMARY_SILENCE`], before it may become the primary picture. A game that +/// draws one frame on its way out does not take the picture. +pub const PRIMARY_SETTLE: Duration = Duration::from_millis(250); + +/// One encoder's recent drawing, for choosing the primary picture. +#[derive(Debug, Clone, Copy)] +struct Drawing { + /// When its current run of frames began. + since: Instant, + last: Instant, + /// The order encoders were first heard from: higher is newer. + order: u64, +} + +/// What [`WindowTable::note_primary`] decided about one frame. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct PrimaryNote { + /// The frame is the primary picture. + pub primary: bool, + /// The primary picture just moved to this frame's encoder: its watchers + /// need a keyframe and a fresh sequence. + pub switched: bool, +} + +/// What a connection is shown. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Subscription { + /// Never subscribed: the box's primary picture, encoded as window `0`. + Legacy, + /// One window, by id. + Window(WindowId), + /// Subscribed to a window of a compositor that has since gone. Its ids + /// mean nothing now, so nothing is shown until the client subscribes again. + Torn, +} + +/// The window a frame is sent as to a connection with `sub`, or `None` if +/// that connection is not shown it. `window` is the frame's window, if known; +/// `primary` is whether it is the primary picture. +pub fn route(sub: Subscription, window: Option, primary: bool) -> Option { + match sub { + Subscription::Legacy => primary.then_some(WINDOW_DEFAULT), + Subscription::Window(w) => (window == Some(w)).then_some(w), + Subscription::Torn => None, + } +} + +/// Whether a connection with `sub` is sent a cursor update describing +/// `cursor_window`. A connection that never subscribed gets every update, as +/// it always did. +pub fn wants_cursor(sub: Subscription, cursor_window: WindowId) -> bool { + match sub { + Subscription::Legacy => true, + Subscription::Window(w) => w == cursor_window, + Subscription::Torn => false, + } +} + +/// A window as it stands, enough to tell a late subscriber about it. +#[derive(Debug, Clone)] +struct OpenWindow { + opened: WindowEvent, + icon: Option, +} + +/// The box's windows, their owners, and which encoder is primary. +#[derive(Debug, Default)] +pub struct WindowTable { + open: BTreeMap, + pids: HashMap, + /// Processes whose window has closed. Their last frames belong to nobody, + /// not to whichever window is left. + retired: std::collections::HashSet, + /// The launch each encoder said it draws for, ref(d-0085). The encoder + /// says so itself: the hub runs as another user and cannot read a game's + /// environment. + launch_of: HashMap, + /// The encoder that last drew each window, so a keyframe request can go to + /// that encoder alone. + drawn_by: HashMap, + drawing: HashMap, + next_order: u64, + primary: Option, +} + +impl WindowTable { + /// Take in a lifecycle event from the box. + pub fn apply(&mut self, event: &WindowEvent) { + match event { + WindowEvent::Opened { window, .. } => { + self.open.insert( + *window, + OpenWindow { + opened: event.clone(), + icon: None, + }, + ); + } + WindowEvent::Title { window, title } => { + if let Some(WindowEvent::Opened { title: t, .. }) = + self.open.get_mut(window).map(|w| &mut w.opened) + { + *t = title.clone(); + } + } + WindowEvent::Size { + window, + width, + height, + } => { + if let Some(WindowEvent::Opened { + width: w, + height: h, + .. + }) = self.open.get_mut(window).map(|w| &mut w.opened) + { + (*w, *h) = (*width, *height); + } + } + WindowEvent::Icon { window, .. } => { + if let Some(open) = self.open.get_mut(window) { + open.icon = Some(event.clone()); + } + } + WindowEvent::Closed { window } => { + self.open.remove(window); + let retired = &mut self.retired; + self.pids.retain(|pid, w| { + let keep = w != window; + if !keep { + retired.insert(*pid); + } + keep + }); + } + } + } + + /// Window `window` is owned by process `pid`. + pub fn bind_pid(&mut self, window: WindowId, pid: u32) { + self.pids.insert(pid, window); + self.retired.remove(&pid); + } + + /// Encoder `pid` draws for launch `launch` (empty: it did not say). + pub fn note_launch(&mut self, pid: u32, launch: String) { + if launch.is_empty() { + self.launch_of.remove(&pid); + } else { + self.launch_of.insert(pid, launch); + } + } + + /// Encoder `pid` drew a frame of `window`. + pub fn note_drawn(&mut self, window: WindowId, pid: u32) { + self.drawn_by.insert(window, pid); + } + + /// The encoder that last drew `window`. + pub fn encoder_of(&self, window: WindowId) -> Option { + self.drawn_by.get(&window).copied() + } + + /// The encoder of the primary picture, when it is a tagged one. + pub fn primary_pid(&self) -> Option { + self.primary.flatten() + } + + /// Every open window, as the events that tell a late subscriber about it. + pub fn snapshot(&self) -> Vec { + let mut out = Vec::new(); + for open in self.open.values() { + out.push(open.opened.clone()); + out.extend(open.icon.clone()); + } + out + } + + /// The compositor went: forget everything, and say which windows closed. + pub fn reset(&mut self) -> Vec { + let closed = self.open.keys().copied().collect(); + *self = Self { + primary: self.primary, + drawing: std::mem::take(&mut self.drawing), + next_order: self.next_order, + launch_of: std::mem::take(&mut self.launch_of), + ..Self::default() + }; + closed + } + + /// Which window a frame from `source` is. + /// + /// An untagged encoder is window `0`. A tagged one is the window its + /// process owns; failing that, the newest window of the launch the process + /// belongs to (a renderer process drawing for a window another process of + /// the same launch owns); failing that, window `0` while the box has no + /// other window. Otherwise nobody can say, and the frame is shown only as + /// the primary picture, if it is that. + pub fn resolve(&mut self, source: Source) -> Option { + let Some(pid) = source else { + return Some(WINDOW_DEFAULT); + }; + if let Some(window) = self.pids.get(&pid) { + return Some(*window); + } + if self.retired.contains(&pid) { + return None; + } + if let Some(launch) = self.launch_of.get(&pid) { + let newest = self.open.iter().rev().find_map(|(id, w)| match &w.opened { + WindowEvent::Opened { launch_id, .. } if launch_id == launch => Some(*id), + _ => None, + }); + if newest.is_some() { + return newest; + } + } + self.open + .keys() + .all(|w| *w == WINDOW_DEFAULT) + .then_some(WINDOW_DEFAULT) + } + + /// Note a frame from `source` and decide whether it is the primary + /// picture. + /// + /// The first encoder to draw is primary, and stays so while it keeps + /// drawing. Only once it has been silent for [`PRIMARY_SILENCE`] may + /// another take over, and only one that has drawn for [`PRIMARY_SETTLE`]; + /// of those, the newest encoder, which is the most recent launch. The + /// first choice is not a switch: nobody has been shown anything yet. + pub fn note_primary(&mut self, source: Source, now: Instant) -> PrimaryNote { + let order = &mut self.next_order; + let d = self.drawing.entry(source).or_insert_with(|| { + *order += 1; + Drawing { + since: now, + last: now, + order: *order, + } + }); + if now.duration_since(d.last) >= PRIMARY_SILENCE { + d.since = now; + } + d.last = now; + + let not = PrimaryNote { + primary: false, + switched: false, + }; + let Some(current) = self.primary else { + self.primary = Some(source); + return PrimaryNote { + primary: true, + switched: false, + }; + }; + if current == source { + return PrimaryNote { + primary: true, + switched: false, + }; + } + let current_silent = self + .drawing + .get(¤t) + .is_none_or(|c| now.duration_since(c.last) >= PRIMARY_SILENCE); + if !current_silent { + return not; + } + let best = self + .drawing + .iter() + .filter(|(_, d)| { + now.duration_since(d.last) < PRIMARY_SILENCE + && now.duration_since(d.since) >= PRIMARY_SETTLE + }) + .max_by_key(|(_, d)| d.order) + .map(|(s, _)| *s); + if best != Some(source) { + return not; + } + self.primary = Some(source); + PrimaryNote { + primary: true, + switched: true, + } + } +} + +/// Tags input on its way to the box with the window it belongs to. +/// +/// One per connection to nescope, starting at window `0`, which is what +/// nescope assumes on every new connection. A focus tag goes out only when the +/// window changes, so a box where every event is window `0`'s sees exactly the +/// bytes it always did. +#[derive(Debug, Default)] +pub struct FocusTagger { + last: WindowId, + /// Who last sent a keyboard or pointer event, and when. + typing: Option<(u32, Instant)>, +} + +/// How long a client's keyboard or pointer input keeps another client's +/// gamepad from taking focus. +pub const TYPING_HOLDS_FOCUS: Duration = Duration::from_secs(1); + +impl FocusTagger { + /// The messages to write for one input event from `client`, for + /// `window`. An empty event asks only for focus: a gamepad message, whose + /// pads reach the game by another route and are read only by the focused + /// window. It moves focus only to a different window, and not while + /// another client has typed or pointed in the last [`TYPING_HOLDS_FOCUS`]. + pub fn messages( + &mut self, + client: u32, + window: WindowId, + event: &[u8], + now: Instant, + ) -> Vec> { + let mut out = Vec::with_capacity(2); + if event.first() == Some(&INPUT_OUTPUT_SIZE) && event.len() >= 5 { + // Carries its window itself and moves no focus. Window 0 keeps + // the old form, which a compositor that predates windows reads. + if window == WINDOW_DEFAULT { + out.push(event.to_vec()); + } else { + let width = u16::from_le_bytes([event[1], event[2]]); + let height = u16::from_le_bytes([event[3], event[4]]); + let mut m = Vec::with_capacity(7); + encode_window_output_size(&mut m, window, width, height); + out.push(m); + } + return out; + } + if event.is_empty() { + let held = self.typing.is_some_and(|(who, at)| { + who != client && now.duration_since(at) < TYPING_HOLDS_FOCUS + }); + if held { + return out; + } + } else { + self.typing = Some((client, now)); + } + if window != self.last { + let mut tag = Vec::with_capacity(3); + encode_window_focus(&mut tag, window); + out.push(tag); + self.last = window; + } + if !event.is_empty() { + out.push(event.to_vec()); + } + out + } +} + +#[cfg(test)] +mod tests { + use super::*; + use nesprotocol::input::{ + DecodedInput, decode_input_event, encode_key_event, encode_mouse_move, encode_output_size, + }; + + fn opened(window: WindowId, launch: &str) -> WindowEvent { + WindowEvent::Opened { + window, + width: 1920, + height: 1080, + launch_id: launch.into(), + title: format!("w{window}"), + app_id: String::new(), + } + } + + + #[test] + fn an_untagged_encoder_is_window_zero() { + let mut t = WindowTable::default(); + assert_eq!(t.resolve(None), Some(0)); + t.apply(&opened(0, "a")); + t.apply(&opened(1, "b")); + assert_eq!(t.resolve(None), Some(0)); + } + + #[test] + fn a_tagged_encoder_is_the_window_its_process_owns() { + let mut t = WindowTable::default(); + t.apply(&opened(0, "a")); + t.apply(&opened(1, "b")); + t.bind_pid(0, 100); + t.bind_pid(1, 200); + assert_eq!(t.resolve(Some(100)), Some(0)); + assert_eq!(t.resolve(Some(200)), Some(1)); + t.apply(&WindowEvent::Closed { window: 1 }); + // Its window gone: its last frames are nobody's, even with only + // window 0 left. + assert_eq!(t.resolve(Some(200)), None); + } + + #[test] + fn an_unknown_process_is_found_by_its_launch() { + let mut t = WindowTable::default(); + t.apply(&opened(0, "steam")); + t.apply(&opened(1, "launch_b")); + t.apply(&opened(2, "launch_b")); + t.note_launch(300, "launch_b".into()); + assert_eq!(t.resolve(Some(300)), Some(2), "the newest of its launch"); + assert_eq!(t.resolve(Some(301)), None); + } + + #[test] + fn with_one_window_any_encoder_is_window_zero() { + // One game, nescope that predates pid messages: as it always was. + let mut t = WindowTable::default(); + assert_eq!(t.resolve(Some(4242)), Some(0)); + t.apply(&opened(0, "")); + assert_eq!(t.resolve(Some(4242)), Some(0)); + } + + fn ms(t0: Instant, ms: u64) -> Instant { + t0 + Duration::from_millis(ms) + } + + #[test] + fn the_first_encoder_to_draw_is_primary_until_it_stops() { + let mut t = WindowTable::default(); + let t0 = Instant::now(); + let yes = PrimaryNote { primary: true, switched: false }; + let no = PrimaryNote { primary: false, switched: false }; + assert_eq!(t.note_primary(Some(1), t0), yes, "the first is no switch"); + assert_eq!(t.note_primary(Some(2), ms(t0, 10)), no); + assert_eq!(t.note_primary(Some(1), ms(t0, 500)), yes); + assert_eq!(t.note_primary(Some(2), ms(t0, 1400)), no, "1 is not silent yet"); + // 1 silent, but 2 only began this run at 1400: not settled. + assert_eq!(t.note_primary(Some(2), ms(t0, 1600)), no); + assert_eq!( + t.note_primary(Some(2), ms(t0, 1700)), + PrimaryNote { primary: true, switched: true } + ); + assert_eq!(t.note_primary(Some(1), ms(t0, 1710)), no); + assert_eq!(t.primary_pid(), Some(2)); + } + + #[test] + fn a_fresh_encoder_waits_to_settle_and_the_newest_wins() { + let mut t = WindowTable::default(); + let t0 = Instant::now(); + t.note_primary(Some(1), t0); + // 2 and 3 draw in turn from 100 ms; 1 has stopped. + for at in (100..1000).step_by(20) { + assert!(!t.note_primary(Some(2), ms(t0, at)).primary); + assert!(!t.note_primary(Some(3), ms(t0, at + 10)).primary); + } + // 1 silent, both settled: 2's frame does not take it, 3 is newer. + assert!(!t.note_primary(Some(2), ms(t0, 1000)).primary); + assert!(t.note_primary(Some(3), ms(t0, 1010)).switched); + + // A newcomer that has drawn under 250 ms is passed over. + let mut t = WindowTable::default(); + t.note_primary(Some(1), t0); + assert!(!t.note_primary(Some(4), ms(t0, 1100)).primary, "not settled"); + assert!(t.note_primary(Some(4), ms(t0, 1360)).switched); + } + + #[test] + fn alternating_bursts_do_not_flap_faster_than_the_silence() { + let mut t = WindowTable::default(); + let t0 = Instant::now(); + let mut switches = Vec::new(); + for frame in 0..1000u64 { + let at = frame * 16; + let who = if (at / 1500) % 2 == 0 { 1 } else { 2 }; + if t.note_primary(Some(who), ms(t0, at)).switched { + switches.push(at); + } + } + assert!(!switches.is_empty()); + for w in switches.windows(2) { + assert!(w[1] - w[0] >= 1000, "switched at {} then {}", w[0], w[1]); + } + } + + #[test] + fn routing() { + use Subscription::*; + assert_eq!( + route(Legacy, Some(3), true), + Some(0), + "old encoding for an old client" + ); + assert_eq!(route(Legacy, Some(0), false), None); + assert_eq!(route(Window(3), Some(3), false), Some(3)); + assert_eq!(route(Window(0), Some(0), true), Some(0)); + assert_eq!(route(Window(3), Some(4), true), None); + assert_eq!(route(Window(3), None, true), None); + assert_eq!(route(Torn, Some(0), true), None); + assert!(wants_cursor(Legacy, 7)); + assert!(wants_cursor(Window(7), 7)); + assert!(!wants_cursor(Window(7), 0)); + assert!(!wants_cursor(Torn, 0)); + } + + #[test] + fn snapshot_reflects_the_box_as_it_stands() { + let mut t = WindowTable::default(); + t.apply(&opened(0, "a")); + t.apply(&opened(1, "b")); + t.apply(&WindowEvent::Title { + window: 1, + title: "Paused".into(), + }); + t.apply(&WindowEvent::Size { + window: 1, + width: 1280, + height: 720, + }); + t.apply(&WindowEvent::Icon { + window: 1, + width: 1, + height: 1, + rgba: vec![1, 2, 3, 4], + }); + t.apply(&WindowEvent::Closed { window: 0 }); + let snap = t.snapshot(); + assert_eq!(snap.len(), 2); + assert_eq!( + snap[0], + WindowEvent::Opened { + window: 1, + width: 1280, + height: 720, + launch_id: "b".into(), + title: "Paused".into(), + app_id: String::new(), + } + ); + assert!(matches!(snap[1], WindowEvent::Icon { window: 1, .. })); + assert_eq!(t.reset(), vec![1]); + assert!(t.snapshot().is_empty()); + } + + #[test] + fn one_window_input_is_the_old_bytes() { + let mut tagger = FocusTagger::default(); + let mut key = Vec::new(); + encode_key_event(&mut key, true, 30); + let mut size = Vec::new(); + encode_output_size(&mut size, 1280, 720); + assert_eq!(tagger.messages(1, 0, &key, Instant::now()), vec![key.clone()]); + assert_eq!(tagger.messages(1, 0, &size, Instant::now()), vec![size.clone()]); + assert!(tagger.messages(1, 0, &[], Instant::now()).is_empty()); + } + + #[test] + fn a_change_of_window_is_tagged_once() { + let mut tagger = FocusTagger::default(); + let mut mv = Vec::new(); + encode_mouse_move(&mut mv, 1, 1); + let out = tagger.messages(1, 2, &mv, Instant::now()); + assert_eq!(out.len(), 2); + assert!(matches!( + decode_input_event(&out[0]), + Some(DecodedInput::WindowFocus { window: 2 }) + )); + assert_eq!(out[1], mv); + assert_eq!( + tagger.messages(1, 2, &mv, Instant::now()), + vec![mv.clone()], + "no tag while unchanged" + ); + // Back to window 0: tagged, since the box is now on 2. + assert_eq!(tagger.messages(1, 0, &mv, Instant::now()).len(), 2); + // Focus alone. + let out = tagger.messages(1, 5, &[], Instant::now()); + assert_eq!(out.len(), 1); + assert!(tagger.messages(1, 5, &[], Instant::now()).is_empty()); + } + + #[test] + fn a_gamepad_does_not_take_focus_from_someone_typing() { + let mut tagger = FocusTagger::default(); + let t0 = Instant::now(); + let mut key = Vec::new(); + encode_key_event(&mut key, true, 30); + // Client 1 types into window 0; client 2's pad asks for window 4. + tagger.messages(1, 0, &key, t0); + assert!(tagger.messages(2, 4, &[], t0 + Duration::from_millis(500)).is_empty()); + // A second of quiet: it may. + assert_eq!(tagger.messages(2, 4, &[], t0 + Duration::from_millis(1100)).len(), 1); + // Asking for the window that has focus changes nothing. + assert!(tagger.messages(2, 4, &[], t0 + Duration::from_millis(1200)).is_empty()); + // The typist's own pad is not held back by its own typing. + tagger.messages(1, 4, &key, t0 + Duration::from_millis(1300)); + assert_eq!(tagger.messages(1, 0, &[], t0 + Duration::from_millis(1310)).len(), 1); + } + + #[test] + fn a_window_resize_is_rewritten_except_for_window_zero() { + let mut tagger = FocusTagger::default(); + let mut size = Vec::new(); + encode_output_size(&mut size, 1280, 720); + let out = tagger.messages(1, 3, &size, Instant::now()); + assert_eq!( + out.len(), + 1, + "a resize carries its window and moves no focus" + ); + assert!(matches!( + decode_input_event(&out[0]), + Some(DecodedInput::WindowOutputSize { + window: 3, + width: 1280, + height: 720 + }) + )); + } +} diff --git a/crates/nesprotocol/src/lib.rs b/crates/nesprotocol/src/lib.rs index 5ec0f0aa..15180c9a 100644 --- a/crates/nesprotocol/src/lib.rs +++ b/crates/nesprotocol/src/lib.rs @@ -173,6 +173,13 @@ pub const STREAM_KEYFRAME: u8 = 5; /// One message of a PyroWave frame, on the nescapture→hub video socket only. /// A frame is several of these: see the [`pyrowave`] module. pub const STREAM_PYROWAVE: u8 = 7; +/// An encoder introducing itself, on the nescapture→hub video socket only, +/// ref(d-0085): a pid-tagged IPC frame ([`IPC_FLAG_PID`]) whose data is the +/// encoder process's `NESTRI_LAUNCH_ID` (UTF-8, possibly empty). Sent before +/// every keyframe, so a hub that started late or lost one still learns it. +/// The encoder runs inside the game and can read its own environment; the +/// hub runs as another user and cannot. +pub const STREAM_ENCODER_HELLO: u8 = 9; // ── Bidi stream types (desktop ↔ hub over QUIC bidi) ─────────── @@ -224,6 +231,47 @@ pub const FLAG_RECONFIG: u8 = 0x02; // stream reconfiguration (codec/bit-depth c /// [`clock::monotonic_us`]; for audio, the packet's first sample on the audio /// graph's clock, in 48 kHz frames. Both are the producer's to know and the /// hub's to carry; neither is ever converted into the other. +/// Set in an IPC frame's reserved byte when the producer's process id follows +/// the data as `[4B pid LE]`, ref(d-0085). +/// +/// Several games in one box each run their own encoder, and the hub tells their +/// frames apart by process. A reader that predates this ignores the reserved +/// byte and reads exactly `len` bytes of data, so the trailer is invisible to +/// it; a frame without the flag is from a producer that predates it, which the +/// hub treats as the box's first window. +pub const IPC_FLAG_PID: u8 = 0x01; + +/// Where an encoder listens for commands when one process per box was all +/// there was. Every encoder still binds it, so a hub that predates per-process +/// sockets keeps reaching the newest one, as it always did. +pub const CAPTURE_CMD_SOCKET: &str = "/tmp/nescapture-cmd.sock"; + +/// Where the encoder in process `pid` listens for commands, ref(d-0085). One +/// socket per process, so two games in one box are not told each other's +/// keyframe requests, and neither steals the other's socket. +pub fn capture_cmd_socket(pid: u32) -> String { + format!("/tmp/nescapture-cmd-{pid}.sock") +} + +/// [`encode_ipc_frame`], tagged with the producer's process id. See +/// [`IPC_FLAG_PID`]. +#[allow(clippy::too_many_arguments)] +pub fn encode_ipc_frame_from( + pid: u32, + stream_type: u8, + codec: u8, + flags: u8, + stamp: u64, + width: u16, + height: u16, + data: &[u8], +) -> Vec { + let mut buf = encode_ipc_frame(stream_type, codec, flags, stamp, width, height, data); + buf[7] |= IPC_FLAG_PID; + buf.extend_from_slice(&pid.to_le_bytes()); + buf +} + pub fn encode_ipc_frame( stream_type: u8, codec: u8, @@ -260,6 +308,8 @@ pub struct DecodedIpcFrame<'a> { pub width: u16, pub height: u16, pub data: &'a [u8], + /// The producer's process id, when it said ([`IPC_FLAG_PID`]). + pub pid: Option, } /// Parse a raw IPC datagram. Returns None if magic doesn't match or buffer is too short. @@ -282,7 +332,15 @@ pub fn decode_ipc_frame(buf: &[u8]) -> Option> { return None; } + let end = IPC_HEADER_LEN + data_len; + let pid = if buf[7] & IPC_FLAG_PID != 0 { + Some(u32::from_le_bytes(buf.get(end..end + 4)?.try_into().ok()?)) + } else { + None + }; + Some(DecodedIpcFrame { + pid, stream_type, codec, flags, @@ -1258,6 +1316,23 @@ mod media_control_tests { mod media_payload_tests { use super::*; + #[test] + fn an_ipc_frame_carries_its_producer_compatibly() { + let plain = encode_ipc_frame(STREAM_VIDEO, CODEC_AV1, FLAG_KEYFRAME, 7, 640, 360, &[1, 2, 3]); + let tagged = + encode_ipc_frame_from(4242, STREAM_VIDEO, CODEC_AV1, FLAG_KEYFRAME, 7, 640, 360, &[1, 2, 3]); + assert_eq!(decode_ipc_frame(&plain).unwrap().pid, None); + let d = decode_ipc_frame(&tagged).unwrap(); + assert_eq!((d.pid, d.data, d.flags), (Some(4242), &[1u8, 2, 3][..], FLAG_KEYFRAME)); + // What an older reader sees: the same header apart from the reserved + // byte, the same data, and bytes after it that it never reads. + assert_eq!(&tagged[..7], &plain[..7]); + assert_eq!(&tagged[8..plain.len()], &plain[8..]); + // A flag promising a pid that is not there is not a frame. + assert!(decode_ipc_frame(&tagged[..tagged.len() - 1]).is_none()); + assert_eq!(capture_cmd_socket(12), "/tmp/nescapture-cmd-12.sock"); + } + #[test] fn an_ipc_frame_carries_its_stamp_whole() { // A u64 of microseconds, so a stamp past 2^32 must come back intact: diff --git a/crates/nesprotocol/src/window.rs b/crates/nesprotocol/src/window.rs index c239c4ee..ce6fe993 100644 --- a/crates/nesprotocol/src/window.rs +++ b/crates/nesprotocol/src/window.rs @@ -16,9 +16,14 @@ //! module existed, which is the whole compatibility story: //! //! - **Old client, new hub.** An old client never sends [`MSG_WINDOW_SUBSCRIBE`], -//! so the hub treats it as subscribed to window `0`, sends it only window -//! `0`'s frames, and those are byte-identical to what it always received. It -//! is never sent a [`STREAM_WINDOWS`] stream either: the hub opens one only +//! so the hub sends it the box's *primary picture* in window `0`'s +//! encoding, byte-identical to what it always received. The primary picture +//! is one encoder's frames: the first to draw, replaced only after it has +//! been silent for a second by one that has drawn for at least 250 ms +//! (preferring the newest), with a keyframe and a fresh sequence on every +//! switch. It is usually window `0` but need not be: a launcher window that +//! never draws can take id `0`. With one game it is that game. Such a +//! client is never sent a [`STREAM_WINDOWS`] stream: the hub opens one only //! for a client that has subscribed. //! - **New client, old hub.** The old hub logs [`MSG_WINDOW_SUBSCRIBE`] as an //! unknown control message and carries on, sends window `0` in the old @@ -89,6 +94,16 @@ //! fragments with [`max_fragment_payload`] for the window it is sending, never //! from `DGRAM_HDR_LEN` alone, or the datagram outgrows the path. //! +//! # Known limits (phase 1) +//! +//! - **One bitrate for the box.** Every window's encoder is given the same +//! target, decided by the one controller, so one client's congestion sets +//! every window's rate. Per-window rate is follow-up work. A keyframe +//! request does go only to the encoder of the requesting client's window. +//! - **nescope and neshub ship together** in the guest image and must not ship +//! apart: a nescope that sends window messages to a hub that predates them +//! is not a supported pairing, even though both ends tolerate it. +//! //! # Lifetime of ids and subscriptions //! //! Ids do not wrap within a run of a box: a box that has minted 65,535 window @@ -429,6 +444,46 @@ impl WindowEvent { } } +// ── Box-internal (nescope → hub) ──────────────────────────────────────────── +// +// Carried on the input socket beside the cursor messages, each as one +// length-prefixed message whose first byte is its type, the same way the +// lifecycle events above travel there. Never sent to a client. + +/// Window `[2B]` is owned by process `[4B pid]`: how the hub joins an encoder's +/// frames, tagged with its process, to the window they draw. +pub const MSG_WINDOW_PID: u8 = 0x28; + +/// The cursor messages after this one describe the pointer as window `[2B]` +/// sees it, until the next. A compositor that never sends it describes +/// window `0`. +pub const MSG_CURSOR_WINDOW: u8 = 0x29; + +/// `[MSG_WINDOW_PID][2B window][4B pid]`. +pub fn encode_window_pid(buf: &mut Vec, window: WindowId, pid: u32) { + buf.push(MSG_WINDOW_PID); + buf.extend_from_slice(&window.to_le_bytes()); + buf.extend_from_slice(&pid.to_le_bytes()); +} + +/// The payload after the type byte of [`MSG_WINDOW_PID`]. +pub fn decode_window_pid(payload: &[u8]) -> Option<(WindowId, u32)> { + let mut r = Reader(payload); + Some((r.u16()?, r.u32()?)) +} + +/// `[MSG_CURSOR_WINDOW][2B window]`. +pub fn encode_cursor_window(buf: &mut Vec, window: WindowId) { + buf.push(MSG_CURSOR_WINDOW); + buf.extend_from_slice(&window.to_le_bytes()); +} + +/// A lifecycle event as one box-internal message: `[type][payload]`. +pub fn encode_event_message(buf: &mut Vec, event: &WindowEvent) { + buf.push(event.msg_type()); + event.encode_payload(buf); +} + fn put_str(buf: &mut Vec, s: &str) { let mut end = s.len().min(u16::MAX as usize); while !s.is_char_boundary(end) { @@ -723,6 +778,31 @@ mod tests { assert_eq!(decode_subscribe(&[4, 0, 0xEE]), Some(4)); } + #[test] + fn box_internal_messages_round_trip() { + let mut b = Vec::new(); + encode_window_pid(&mut b, 3, 0x0102_0304); + assert_eq!(b, [MSG_WINDOW_PID, 3, 0, 4, 3, 2, 1]); + assert_eq!(decode_window_pid(&b[1..]), Some((3, 0x0102_0304))); + assert_eq!(decode_window_pid(&b[1..6]), None); + + let mut b = Vec::new(); + encode_cursor_window(&mut b, 0x0201); + assert_eq!(b, [MSG_CURSOR_WINDOW, 1, 2]); + assert_eq!(decode_subscribe(&b[1..]), Some(0x0201)); + + let ev = WindowEvent::Closed { window: 9 }; + let mut b = Vec::new(); + encode_event_message(&mut b, &ev); + assert_eq!(WindowEvent::decode(b[0], &b[1..]), Some(ev)); + + // None of them can be mistaken for what already shares the socket. + for t in [MSG_WINDOW_PID, MSG_CURSOR_WINDOW, MSG_WINDOW_OPENED, MSG_WINDOW_CLOSED] { + assert!(t != crate::input::CURSOR_UPDATE && t != crate::input::CURSOR_IMAGE); + assert!(t > crate::stats::STATS_HUB); + } + } + #[test] #[should_panic] #[cfg(debug_assertions)]