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)]