diff --git a/Cargo.lock b/Cargo.lock index 1f0910a5..c1676bc2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2420,10 +2420,11 @@ dependencies = [ [[package]] name = "neshub" -version = "0.1.0" +version = "0.2.0" dependencies = [ "anyhow", "base64", + "bytes", "clap", "iroh", "libc", diff --git a/apps/nescapture/src/encode.rs b/apps/nescapture/src/encode.rs index bd4a89cc..5081c995 100644 --- a/apps/nescapture/src/encode.rs +++ b/apps/nescapture/src/encode.rs @@ -1095,9 +1095,7 @@ fn stats_sender_thread( let ca = capture_attempts.swap(0, Ordering::Relaxed); let mut buf = Vec::with_capacity(22); - nesprotocol::stats::encode_hudless_stats( - &mut buf, fps, enc_ms, dropped, pa, ca, cap_ms, - ); + nesprotocol::stats::encode_hudless_stats(&mut buf, fps, enc_ms, dropped, pa, ca, cap_ms); let _ = socket.send(&buf); } diff --git a/apps/nescope/src/focus.rs b/apps/nescope/src/focus.rs index d729fd08..cdd20a79 100644 --- a/apps/nescope/src/focus.rs +++ b/apps/nescope/src/focus.rs @@ -156,12 +156,7 @@ impl KeyboardTarget for KeyboardFocusTarget { } impl PointerTarget for KeyboardFocusTarget { - fn enter( - &self, - seat: &Seat, - data: &mut NescopeState, - event: &MotionEvent, - ) { + fn enter(&self, seat: &Seat, data: &mut NescopeState, event: &MotionEvent) { match self { Self::Window(w) => match w.underlying_surface() { WindowSurface::Wayland(w) => { @@ -175,12 +170,7 @@ impl PointerTarget for KeyboardFocusTarget { } } - fn motion( - &self, - seat: &Seat, - data: &mut NescopeState, - event: &MotionEvent, - ) { + fn motion(&self, seat: &Seat, data: &mut NescopeState, event: &MotionEvent) { match self { Self::Window(w) => match w.underlying_surface() { WindowSurface::Wayland(w) => { @@ -205,9 +195,7 @@ impl PointerTarget for KeyboardFocusTarget { WindowSurface::Wayland(w) => { PointerTarget::relative_motion(w.wl_surface(), seat, data, event) } - WindowSurface::X11(s) => { - PointerTarget::relative_motion(s, seat, data, event) - } + WindowSurface::X11(s) => PointerTarget::relative_motion(s, seat, data, event), }, Self::ProxiedX11 { proxy_surface, .. } => { PointerTarget::relative_motion(proxy_surface, seat, data, event) @@ -215,12 +203,7 @@ impl PointerTarget for KeyboardFocusTarget { } } - fn button( - &self, - seat: &Seat, - data: &mut NescopeState, - event: &ButtonEvent, - ) { + fn button(&self, seat: &Seat, data: &mut NescopeState, event: &ButtonEvent) { match self { Self::Window(w) => match w.underlying_surface() { WindowSurface::Wayland(w) => { @@ -234,17 +217,10 @@ impl PointerTarget for KeyboardFocusTarget { } } - fn axis( - &self, - seat: &Seat, - data: &mut NescopeState, - frame: AxisFrame, - ) { + fn axis(&self, seat: &Seat, data: &mut NescopeState, frame: AxisFrame) { match self { Self::Window(w) => match w.underlying_surface() { - WindowSurface::Wayland(w) => { - PointerTarget::axis(w.wl_surface(), seat, data, frame) - } + WindowSurface::Wayland(w) => PointerTarget::axis(w.wl_surface(), seat, data, frame), WindowSurface::X11(s) => PointerTarget::axis(s, seat, data, frame), }, Self::ProxiedX11 { proxy_surface, .. } => { @@ -253,13 +229,7 @@ impl PointerTarget for KeyboardFocusTarget { } } - fn leave( - &self, - seat: &Seat, - data: &mut NescopeState, - serial: Serial, - time: u32, - ) { + fn leave(&self, seat: &Seat, data: &mut NescopeState, serial: Serial, time: u32) { match self { Self::Window(w) => match w.underlying_surface() { WindowSurface::Wayland(w) => { @@ -273,21 +243,67 @@ impl PointerTarget for KeyboardFocusTarget { } } - fn gesture_swipe_begin(&self, _seat: &Seat, _data: &mut NescopeState, _event: &smithay::input::pointer::GestureSwipeBeginEvent) {} - fn gesture_swipe_update(&self, _seat: &Seat, _data: &mut NescopeState, _event: &smithay::input::pointer::GestureSwipeUpdateEvent) {} - fn gesture_swipe_end(&self, _seat: &Seat, _data: &mut NescopeState, _event: &smithay::input::pointer::GestureSwipeEndEvent) {} - fn gesture_pinch_begin(&self, _seat: &Seat, _data: &mut NescopeState, _event: &smithay::input::pointer::GesturePinchBeginEvent) {} - fn gesture_pinch_update(&self, _seat: &Seat, _data: &mut NescopeState, _event: &smithay::input::pointer::GesturePinchUpdateEvent) {} - fn gesture_pinch_end(&self, _seat: &Seat, _data: &mut NescopeState, _event: &smithay::input::pointer::GesturePinchEndEvent) {} - fn gesture_hold_begin(&self, _seat: &Seat, _data: &mut NescopeState, _event: &smithay::input::pointer::GestureHoldBeginEvent) {} - fn gesture_hold_end(&self, _seat: &Seat, _data: &mut NescopeState, _event: &smithay::input::pointer::GestureHoldEndEvent) {} + fn gesture_swipe_begin( + &self, + _seat: &Seat, + _data: &mut NescopeState, + _event: &smithay::input::pointer::GestureSwipeBeginEvent, + ) { + } + fn gesture_swipe_update( + &self, + _seat: &Seat, + _data: &mut NescopeState, + _event: &smithay::input::pointer::GestureSwipeUpdateEvent, + ) { + } + fn gesture_swipe_end( + &self, + _seat: &Seat, + _data: &mut NescopeState, + _event: &smithay::input::pointer::GestureSwipeEndEvent, + ) { + } + fn gesture_pinch_begin( + &self, + _seat: &Seat, + _data: &mut NescopeState, + _event: &smithay::input::pointer::GesturePinchBeginEvent, + ) { + } + fn gesture_pinch_update( + &self, + _seat: &Seat, + _data: &mut NescopeState, + _event: &smithay::input::pointer::GesturePinchUpdateEvent, + ) { + } + fn gesture_pinch_end( + &self, + _seat: &Seat, + _data: &mut NescopeState, + _event: &smithay::input::pointer::GesturePinchEndEvent, + ) { + } + fn gesture_hold_begin( + &self, + _seat: &Seat, + _data: &mut NescopeState, + _event: &smithay::input::pointer::GestureHoldBeginEvent, + ) { + } + fn gesture_hold_end( + &self, + _seat: &Seat, + _data: &mut NescopeState, + _event: &smithay::input::pointer::GestureHoldEndEvent, + ) { + } fn frame(&self, seat: &Seat, data: &mut NescopeState) { match self { Self::Window(w) => match w.underlying_surface() { - WindowSurface::Wayland(w) => { - PointerTarget::frame(w.wl_surface(), seat, data) - } + WindowSurface::Wayland(w) => PointerTarget::frame(w.wl_surface(), seat, data), WindowSurface::X11(s) => PointerTarget::frame(s, seat, data), }, Self::ProxiedX11 { proxy_surface, .. } => { diff --git a/apps/nescope/src/gpu_readback.rs b/apps/nescope/src/gpu_readback.rs index 14bfbb1b..acd570bc 100644 --- a/apps/nescope/src/gpu_readback.rs +++ b/apps/nescope/src/gpu_readback.rs @@ -25,10 +25,10 @@ use std::cell::{Cell, RefCell}; use std::path::PathBuf; +use smithay::backend::allocator::Buffer; use smithay::backend::allocator::dmabuf::Dmabuf; use smithay::backend::egl::{EGLContext, EGLDisplay}; use smithay::backend::renderer::gles::GlesRenderer; -use smithay::backend::allocator::Buffer; use smithay::backend::renderer::{ExportMem, ImportDma}; use smithay::utils::{Point, Rectangle, Size}; @@ -66,9 +66,10 @@ fn render_device() -> Result { .filter_map(Result::ok) .collect(); nodes.sort(); - nodes.into_iter().next().ok_or_else(|| { - "no render node found in /dev/dri; pass --render-device".to_string() - }) + nodes + .into_iter() + .next() + .ok_or_else(|| "no render node found in /dev/dri; pass --render-device".to_string()) } /// Why a read did not happen. diff --git a/apps/nescope/src/input_ipc.rs b/apps/nescope/src/input_ipc.rs index 78ab79da..a159a644 100644 --- a/apps/nescope/src/input_ipc.rs +++ b/apps/nescope/src/input_ipc.rs @@ -13,7 +13,10 @@ impl InputIpcSource { pub fn connect(path: &str) -> io::Result { let stream = UnixStream::connect(path)?; stream.set_nonblocking(true)?; - Ok(Self { stream, buf: Vec::new() }) + Ok(Self { + stream, + buf: Vec::new(), + }) } pub fn try_clone(&self) -> io::Result { @@ -46,7 +49,10 @@ impl EventSource for InputIpcSource { loop { match self.stream.read(&mut tmp) { Ok(0) => { - return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "IPC socket closed")); + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "IPC socket closed", + )); } Ok(n) => { self.buf.extend_from_slice(&tmp[..n]); diff --git a/apps/nescope/src/libinput_backend.rs b/apps/nescope/src/libinput_backend.rs index b9af5c73..b304725f 100644 --- a/apps/nescope/src/libinput_backend.rs +++ b/apps/nescope/src/libinput_backend.rs @@ -5,13 +5,9 @@ use std::os::unix::io::{AsRawFd, OwnedFd}; use std::path::Path; use smithay::reexports::input::{ - self as libinput, - event::{ - self, - keyboard::KeyboardEventTrait, - pointer::PointerScrollEvent, - }, - }; + self as libinput, + event::{self, keyboard::KeyboardEventTrait, pointer::PointerScrollEvent}, +}; use crate::input::{InputEvent, process_input}; use crate::state::NescopeState; @@ -67,7 +63,10 @@ pub fn dispatch_libinput(ctx: &mut libinput::Libinput, state: &mut NescopeState) match pev { event::PointerEvent::Motion(ev) => { process_input( - InputEvent::MouseMoveRelative { dx: ev.dx(), dy: ev.dy() }, + InputEvent::MouseMoveRelative { + dx: ev.dx(), + dy: ev.dy(), + }, state, ); } @@ -89,10 +88,8 @@ pub fn dispatch_libinput(ctx: &mut libinput::Libinput, state: &mut NescopeState) } event::PointerEvent::Button(ev) => { let button = ev.button(); - let down = matches!( - ev.button_state(), - event::pointer::ButtonState::Pressed - ); + let down = + matches!(ev.button_state(), event::pointer::ButtonState::Pressed); let ev = if down { InputEvent::MouseButtonDown { button } } else { diff --git a/apps/nescope/src/screenshot_ipc.rs b/apps/nescope/src/screenshot_ipc.rs index 3ac51919..79ad972d 100644 --- a/apps/nescope/src/screenshot_ipc.rs +++ b/apps/nescope/src/screenshot_ipc.rs @@ -45,8 +45,8 @@ pub use crate::screenshot_wire::{REQUEST_CAPTURE, encode_reply}; use smithay::desktop::{Space, Window}; use smithay::reexports::wayland_server::protocol::wl_buffer; -use smithay::wayland::seat::WaylandFocus; use smithay::wayland::compositor::{BufferAssignment, SurfaceAttributes, with_states}; +use smithay::wayland::seat::WaylandFocus; use smithay::wayland::shm::with_buffer_contents; /// Read the pixels of the frontmost mapped window. @@ -185,7 +185,6 @@ impl ScreenshotIpcSource { pub fn try_clone_writer(&self) -> io::Result { self.stream.try_clone() } - } impl AsFd for ScreenshotIpcSource { diff --git a/apps/neshub/Cargo.toml b/apps/neshub/Cargo.toml index da3ae864..48537226 100644 --- a/apps/neshub/Cargo.toml +++ b/apps/neshub/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "neshub" -version = "0.1.0" +version = "0.2.0" description = "Muxes captured video, audio and input into one QUIC connection to the client" edition.workspace = true license.workspace = true @@ -11,8 +11,7 @@ name = "neshub" path = "src/main.rs" [dependencies] -# The transport. One endpoint per box; clients dial it with a ticket. -iroh = "1.0.0" +iroh = "1.1.0" anyhow.workspace = true clap.workspace = true @@ -23,6 +22,7 @@ tokio.workspace = true tracing.workspace = true tracing-subscriber.workspace = true +bytes = "1" base64 = "0.22" uuid = { version = "1", features = ["v7"] } diff --git a/apps/neshub/src/dgram.rs b/apps/neshub/src/dgram.rs new file mode 100644 index 00000000..12ef5b3e --- /dev/null +++ b/apps/neshub/src/dgram.rs @@ -0,0 +1,224 @@ +//! Fragmenting media frames onto a connection's QUIC datagram flow. +//! +//! See `nestri_protocol::datagram` for why media left unidirectional streams and +//! what the wire format looks like. This is the sending half. + +use std::fmt; +use std::sync::Arc; +use std::sync::atomic::{AtomicU32, Ordering}; + +use bytes::BytesMut; +use iroh::endpoint::{Connection, QuicTransportConfig, SendDatagramError}; +use tracing::{debug, warn}; + +use crate::keyframe::KeyframeSender; + +use nesprotocol::MSG_DATA; +use nesprotocol::datagram::{ + DGRAM_BUFFER_BYTES, DGRAM_HDR_LEN, MAX_FRAGMENTS_PER_FRAME, MIN_DGRAM_PAYLOAD, fragment_count, + write_datagram_header, +}; +use nesprotocol::reliable::video_wants_reliable; + +/// Transport settings for an endpoint carrying media datagrams. +/// +/// See [`DGRAM_BUFFER_BYTES`] for why the defaults are not enough. +pub fn media_transport_config() -> QuicTransportConfig { + QuicTransportConfig::builder() + .datagram_send_buffer_size(DGRAM_BUFFER_BYTES) + .datagram_receive_buffer_size(Some(DGRAM_BUFFER_BYTES)) + .build() +} + +#[derive(Debug)] +pub enum SendFrameError { + /// The peer never advertised datagram support, or it is disabled locally. + /// Since v2 dropped the uni-stream media path, there is nothing to fall back + /// to and the session cannot carry media at all. + Unsupported, + /// The path currently admits less than one useful fragment. + PathTooSmall { payload: usize }, + /// The frame needs more fragments than a receiver will reassemble. + TooFragmented { fragments: usize }, + /// QUIC refused the datagram. + Quic(SendDatagramError), +} + +impl fmt::Display for SendFrameError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Unsupported => write!(f, "peer does not support QUIC datagrams"), + Self::PathTooSmall { payload } => { + write!( + f, + "path admits only {payload} bytes of payload per datagram" + ) + } + Self::TooFragmented { fragments } => write!( + f, + "frame needs {fragments} fragments, limit is {MAX_FRAGMENTS_PER_FRAME}" + ), + Self::Quic(e) => write!(f, "{e}"), + } + } +} + +impl std::error::Error for SendFrameError {} + +/// Fragments frames of one media kind onto a connection's datagram flow. +/// +/// Cloneable so the keyframe path can keep one for its fallback; both halves are +/// cheap handles onto the same connection. +#[derive(Clone)] +pub struct DatagramSender { + conn: Connection, + kind: u8, +} + +impl DatagramSender { + pub fn new(conn: Connection, kind: u8) -> Self { + Self { conn, kind } + } + + /// 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. + /// + /// A failure is reported but never retried. Retrying a real-time frame means + /// delivering it late, which is the behaviour this whole change removes. A + /// half-sent frame is abandoned where it stands; the receiver times its + /// fragments out and asks for a keyframe. + pub fn send_frame(&self, seq: u16, 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 + // between a direct path and a relay. Reading it per frame means the next + // frame is already using the new size, with no shrink-on-rejection state + // to get stuck at a pessimistic value after a transient failure. + let max = self + .conn + .max_datagram_size() + .ok_or(SendFrameError::Unsupported)?; + + let payload_size = max.saturating_sub(DGRAM_HDR_LEN); + if payload_size < MIN_DGRAM_PAYLOAD { + return Err(SendFrameError::PathTooSmall { + payload: payload_size, + }); + } + + let total = fragment_count(body.len(), payload_size); + if total > MAX_FRAGMENTS_PER_FRAME { + return Err(SendFrameError::TooFragmented { fragments: total }); + } + + // 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]; + for index in 0..total { + let start = index * payload_size; + let end = (start + payload_size).min(body.len()); + write_datagram_header(&mut header, self.kind, seq, index as u16, total as u16); + buf.extend_from_slice(&header); + buf.extend_from_slice(&body[start..end]); + } + + for index in 0..total { + let start = index * payload_size; + let end = (start + payload_size).min(body.len()); + let datagram = buf.split_to(DGRAM_HDR_LEN + (end - start)).freeze(); + // Deliberately not `send_datagram_wait`: that waits for buffer space + // under congestion, which prioritises old datagrams over new ones. + // For live media the opposite is right — drop the backlog, send the + // frame that is actually current. + self.conn + .send_datagram(datagram) + .map_err(SendFrameError::Quic)?; + } + + Ok(()) + } +} + +/// Frames arriving on `rx` are numbered and sent — deltas as datagrams, +/// keyframes on a reliable stream each when `keyframes_reliable` is set. +/// +/// A send failure does not end the loop. Datagrams are dropped by design when a +/// path is congested, and one undeliverable frame says nothing about the next; a +/// connection that has genuinely gone away ends the loop through its channel or +/// through `Connection::closed`. +/// +/// Set `keyframes_reliable` for video only. See `nestri_protocol::reliable`: +/// audio has no keyframes, and the flags byte a video frame carries at that +/// offset is unrelated data in an audio packet. +pub async fn run_datagram_writer( + conn: Connection, + kind: u8, + label: &'static str, + mut rx: tokio::sync::mpsc::UnboundedReceiver>, + relay_ms: Option>, + keyframes_reliable: bool, +) { + let sender = DatagramSender::new(conn.clone(), kind); + let keyframes = keyframes_reliable.then(|| KeyframeSender::new(conn, sender.clone())); + let mut seq: u16 = 0; + let mut body: Vec = Vec::new(); + // Datagram loss is expected and self-correcting, so failures are logged at + // debug. Losing datagram support entirely is not, and is worth a warning — + // but only the first time, since it will then be true for every frame. + let mut warned_unsupported = false; + + while let Some(payload) = rx.recv().await { + let t0 = std::time::Instant::now(); + + body.clear(); + nesprotocol::encode_frame_body(&mut body, MSG_DATA, 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 + // by design, so the time this loop spent on it says nothing. + if let Some(ref keyframes) = keyframes + && video_wants_reliable(&payload) + { + if keyframes.send(seq, &body) { + seq = seq.wrapping_add(1); + continue; + } + // Every stream slot is still busy, so this keyframe takes the lossy + // path after all. Worth recording: a keyframe on datagrams is the + // exact case this path exists to avoid, and the count is the only + // way to tell "the receiver is behind" from "the network ate it". + let (on_streams, fell_back) = keyframes.counts(); + debug!( + "{label}: keyframe {seq} falling back to datagrams ({on_streams} on streams, {fell_back} fell back)" + ); + } + + match sender.send_frame(seq, &body) { + Ok(()) => { + if let Some(ref relay) = relay_ms { + let elapsed = t0.elapsed().as_secs_f32() * 1000.0; + relay.store(elapsed.to_bits(), Ordering::Relaxed); + } + } + Err(SendFrameError::Unsupported) => { + if !warned_unsupported { + warn!("{label}: peer does not support QUIC datagrams, media cannot flow"); + warned_unsupported = true; + } + } + Err(e) => debug!("{label}: dropping frame {seq}: {e}"), + } + + seq = seq.wrapping_add(1); + } + + if let Some(ref keyframes) = keyframes { + let (on_streams, fell_back) = keyframes.counts(); + debug!("{label}: {on_streams} keyframes on streams, {fell_back} fell back to datagrams"); + } + debug!("{label} datagram writer exiting (channel closed)"); +} diff --git a/apps/neshub/src/ipc_listener.rs b/apps/neshub/src/ipc_listener.rs index 029dfe06..e61d3fd9 100644 --- a/apps/neshub/src/ipc_listener.rs +++ b/apps/neshub/src/ipc_listener.rs @@ -4,7 +4,6 @@ use std::sync::Arc; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::{UnixDatagram, UnixListener}; use tokio::sync::broadcast::error::RecvError; -use tracing::{debug, error, info, warn}; use crate::session::SessionManager; use nesprotocol::{CODEC_OPUS, STREAM_AUDIO, STREAM_VIDEO, decode_ipc_frame}; @@ -36,7 +35,7 @@ pub async fn run_video_listener(socket_path: PathBuf, session_manager: Arc { - error!( + tracing::error!( "Failed to bind video IPC socket {}: {e}", socket_path.display() ); @@ -44,7 +43,7 @@ pub async fn run_video_listener(socket_path: PathBuf, session_manager: Arc { if let Some(decoded) = decode_ipc_frame(&buf[..n]) { if decoded.stream_type != STREAM_VIDEO { - warn!( + tracing::warn!( "unexpected stream type on video socket: {}", decoded.stream_type ); @@ -68,17 +67,17 @@ pub async fn run_video_listener(socket_path: PathBuf, session_manager: Arc { - error!("video IPC recv error: {e}"); + tracing::error!("video IPC recv error: {e}"); break; } } } - info!("video IPC listener exited"); + tracing::info!("video IPC listener exited"); } pub async fn run_audio_listener(socket_path: PathBuf, session_manager: Arc) { @@ -96,7 +95,7 @@ pub async fn run_audio_listener(socket_path: PathBuf, session_manager: Arc { - error!( + tracing::error!( "Failed to bind audio IPC socket {}: {e}", socket_path.display() ); @@ -104,7 +103,7 @@ pub async fn run_audio_listener(socket_path: PathBuf, session_manager: Arc { if let Some(decoded) = decode_ipc_frame(&buf[..n]) { if decoded.stream_type != STREAM_AUDIO { - warn!( + tracing::warn!( "unexpected stream type on audio socket: {}", decoded.stream_type ); continue; } if decoded.codec != CODEC_OPUS { - warn!("unexpected codec on audio socket: {}", decoded.codec); + tracing::warn!("unexpected codec on audio socket: {}", decoded.codec); continue; } let audio_data = decoded.data.to_vec(); session_manager.broadcast_audio(audio_data).await; } else { - warn!("invalid audio IPC frame ({} bytes)", n); + tracing::warn!("invalid audio IPC frame ({} bytes)", n); } } Err(e) => { - error!("audio IPC recv error: {e}"); + tracing::error!("audio IPC recv error: {e}"); break; } } } - info!("audio IPC listener exited"); + tracing::info!("audio IPC listener exited"); } pub async fn run_input_ipc_listener( @@ -157,7 +156,7 @@ pub async fn run_input_ipc_listener( l } Err(e) => { - error!( + tracing::error!( "Failed to bind input IPC socket {}: {e}", socket_path.display() ); @@ -165,14 +164,14 @@ pub async fn run_input_ipc_listener( } }; - info!("Input IPC listening on {}", socket_path.display()); + tracing::info!("Input IPC listening on {}", socket_path.display()); loop { match listener.accept().await { Ok((stream, peer_addr)) => { - debug!(?peer_addr, "nescope connected to input IPC"); + tracing::debug!(?peer_addr, "nescope connected to input IPC"); if let Some(addr) = peer_addr.as_pathname() { - debug!(?addr, "nescope input IPC peer address"); + tracing::debug!(?addr, "nescope input IPC peer address"); } let (mut read_half, mut write_half) = stream.into_split(); @@ -188,15 +187,15 @@ pub async fn run_input_ipc_listener( frame.extend_from_slice(&len.to_le_bytes()); frame.extend_from_slice(&data); if write_half.write_all(&frame).await.is_err() { - debug!("input IPC write failed"); + tracing::debug!("input IPC write failed"); break; } } Err(RecvError::Lagged(n)) => { - warn!("input broadcast lagged by {n} messages"); + tracing::warn!("input broadcast lagged by {n} messages"); } Err(RecvError::Closed) => { - debug!("input broadcast closed"); + tracing::debug!("input broadcast closed"); break; } } @@ -238,16 +237,16 @@ pub async fn run_input_ipc_listener( _ = read_handle => {} } - debug!("nescope disconnected from input IPC"); + tracing::debug!("nescope disconnected from input IPC"); } Err(e) => { - error!("input IPC accept error: {e}"); + tracing::error!("input IPC accept error: {e}"); break; } } } - info!("input IPC listener exited"); + tracing::info!("input IPC listener exited"); } pub async fn run_stats_ipc_listener( @@ -267,7 +266,7 @@ pub async fn run_stats_ipc_listener( s } Err(e) => { - error!( + tracing::error!( "Failed to bind stats IPC socket {}: {e}", socket_path.display() ); @@ -275,7 +274,7 @@ pub async fn run_stats_ipc_listener( } }; - info!("Stats IPC listening on {}", socket_path.display()); + tracing::info!("Stats IPC listening on {}", socket_path.display()); let mut buf = vec![0u8; 256]; loop { @@ -284,10 +283,52 @@ pub async fn run_stats_ipc_listener( let _ = stats_tx.send(buf[..n].to_vec()); } Err(e) => { - error!("stats IPC recv error: {e}"); + tracing::error!("stats IPC recv error: {e}"); break; } } } - info!("stats IPC listener exited"); + + tracing::info!("stats IPC listener exited"); +} + +pub async fn run_ticket_ipc_listener(socket_path: PathBuf, ticket: crate::NestriTicket) { + if socket_path.exists() { + let _ = std::fs::remove_file(&socket_path); + } + + let listener = match UnixListener::bind(&socket_path) { + Ok(l) => { + let _ = std::fs::set_permissions( + &socket_path, + std::os::unix::fs::PermissionsExt::from_mode(0o666), + ); + l + } + Err(e) => { + tracing::error!( + "Failed to bind ticket IPC socket {}: {e}", + socket_path.display() + ); + return; + } + }; + + tracing::info!("ticket IPC listening on {}", socket_path.display()); + + loop { + match listener.accept().await { + Ok((mut stream, _)) => { + if let Err(e) = stream.write_all(format!("{ticket}\n").as_bytes()).await { + tracing::warn!("could not write ticket to IPC: {e}"); + } + } + Err(e) => { + tracing::error!("ticket IPC accept error: {e}"); + break; + } + } + } + + tracing::info!("ticket IPC listener exited"); } diff --git a/apps/neshub/src/keyframe.rs b/apps/neshub/src/keyframe.rs new file mode 100644 index 00000000..25d877d3 --- /dev/null +++ b/apps/neshub/src/keyframe.rs @@ -0,0 +1,135 @@ +//! Sending keyframes on reliable streams instead of datagrams. +//! +//! See `nesprotocol::reliable` for why keyframes are the exception to +//! "media travels as datagrams". This is the sending half; `dgram` carries +//! everything else. + +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; + +use iroh::endpoint::Connection; +use tracing::debug; + +use nesprotocol::reliable::MAX_KEYFRAME_STREAMS_IN_FLIGHT; +use nesprotocol::{STREAM_KEYFRAME, STREAM_VERSION}; + +use crate::dgram::DatagramSender; + +/// Sends keyframes, one short-lived unidirectional stream each. +pub struct KeyframeSender { + conn: Connection, + /// Where a keyframe goes when a stream cannot take it. + fallback: DatagramSender, + in_flight: Arc, + sent: Arc, + /// Keyframes that ended up on datagrams after all — refused for want of an + /// in-flight slot, or dropped back after the stream failed. + fell_back: Arc, +} + +impl KeyframeSender { + pub fn new(conn: Connection, fallback: DatagramSender) -> Self { + Self { + conn, + fallback, + in_flight: Arc::new(AtomicUsize::new(0)), + sent: Arc::new(AtomicU64::new(0)), + fell_back: Arc::new(AtomicU64::new(0)), + } + } + + /// Take responsibility for one keyframe. + /// + /// Returns false when the caller must send it as datagrams itself: every + /// in-flight slot is occupied, which means keyframes are arriving faster + /// than this connection can absorb them. Sending more stream data to a + /// receiver already that far behind would make it worse, so the frame takes + /// the lossy path rather than the queue. + /// + /// On true, the write happens on its own task. That is the whole point: + /// 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 { + if self.in_flight.load(Ordering::Relaxed) >= MAX_KEYFRAME_STREAMS_IN_FLIGHT { + self.fell_back.fetch_add(1, Ordering::Relaxed); + return false; + } + self.in_flight.fetch_add(1, Ordering::Relaxed); + + let conn = self.conn.clone(); + let fallback = self.fallback.clone(); + let in_flight = self.in_flight.clone(); + let sent = self.sent.clone(); + let fell_back = self.fell_back.clone(); + // The task outlives this call, so it needs the bytes. One copy per + // keyframe — a couple of hundred KB every keyframe interval — against + // not blocking the writer for a flow-controlled write. + let body = body.to_vec(); + + tokio::spawn(async move { + let ok = write_keyframe_stream(&conn, &body).await; + in_flight.fetch_sub(1, Ordering::Relaxed); + + if ok { + sent.fetch_add(1, Ordering::Relaxed); + return; + } + // A stream was promised and could not be delivered on. Datagrams + // 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) { + debug!("keyframe {seq}: stream failed and datagram fallback failed too: {e}"); + } + }); + + true + } + + /// Keyframes delivered on a stream, and keyframes that fell back to + /// datagrams. A rising fallback count means the receiver cannot keep up. + pub fn counts(&self) -> (u64, u64) { + ( + self.sent.load(Ordering::Relaxed), + self.fell_back.load(Ordering::Relaxed), + ) + } +} + +/// Write one keyframe on a stream of its own and close it. +/// +/// The stream carries `[1B STREAM_KEYFRAME] [1B STREAM_VERSION] [frame body]` +/// and nothing else, so finishing it is what tells the receiver where the frame +/// ends — no length prefix needed, exactly as a datagram needs none. +async fn write_keyframe_stream(conn: &Connection, body: &[u8]) -> bool { + let mut send = match conn.open_uni().await { + Ok(s) => s, + Err(e) => { + debug!("keyframe open_uni failed: {e}"); + return false; + } + }; + + if let Err(e) = send.write_all(&[STREAM_KEYFRAME, STREAM_VERSION]).await { + debug!("keyframe header write failed: {e}"); + send.reset(0u32.into()).ok(); + return false; + } + if let Err(e) = send.write_all(body).await { + debug!("keyframe body write failed: {e}"); + send.reset(0u32.into()).ok(); + return false; + } + if let Err(e) = send.finish() { + debug!("keyframe finish failed: {e}"); + return false; + } + + // Hold the in-flight slot until the receiver has actually acknowledged the + // frame, not merely until it was handed to the connection. That is what + // makes the slot count a measure of how far behind the receiver is, which + // is the only reason to bound it. + let _ = send.stopped().await; + true +} diff --git a/apps/neshub/src/main.rs b/apps/neshub/src/main.rs index 4e095dfe..20534caa 100644 --- a/apps/neshub/src/main.rs +++ b/apps/neshub/src/main.rs @@ -1,15 +1,6 @@ -//! neshub — one connection out of the box. -//! -//! Four producers inside the guest send it frames over Unix sockets: -//! nescapture (video, stats), neswire (audio), nescope (cursor). It muxes -//! them into one iroh QUIC endpoint, and fans client input back the other -//! way. That is the whole job. -//! -//! It does not know what is producing the pixels. The payload is started by -//! nesinit and neshub never learns its name, which is what lets the same -//! binary serve a game, a desktop, or something nobody here has thought of. - +mod dgram; mod ipc_listener; +mod keyframe; mod screenshot; mod session; mod ticket; @@ -20,7 +11,6 @@ use std::sync::Arc; use anyhow::Result; use clap::Parser; use iroh::endpoint::presets; -use tracing::{info, warn}; use crate::session::SessionManager; use crate::ticket::NestriTicket; @@ -95,21 +85,27 @@ struct Args { #[tokio::main] async fn main() -> Result<()> { tracing_subscriber::fmt() - .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .with_env_filter( + tracing_subscriber::EnvFilter::builder() + .with_default_directive(tracing_subscriber::filter::LevelFilter::INFO.into()) + .from_env_lossy(), + ) .init(); let args = Args::parse(); - let mut builder = iroh::Endpoint::builder(presets::N0).alpns(vec![ALPN.to_vec()]); + let mut builder = iroh::Endpoint::builder(presets::N0) + .alpns(vec![ALPN.to_vec()]) + .transport_config(crate::dgram::media_transport_config()); match args.relay.as_str() { "default" | "" => { builder = builder.relay_mode(iroh::endpoint::RelayMode::Default); - info!("using default n0-computer relays"); + tracing::info!("using default n0-computer relays"); } "none" | "off" | "disabled" => { builder = builder.relay_mode(iroh::endpoint::RelayMode::Disabled); - info!("relays disabled (direct connections only)"); + tracing::info!("relays disabled (direct connections only)"); } url => { let relay_url: iroh::RelayUrl = url.parse()?; @@ -119,14 +115,14 @@ async fn main() -> Result<()> { Arc::new(iroh::RelayConfig::new(relay_url, None)), ); builder = builder.relay_mode(iroh::endpoint::RelayMode::Custom(relay_map)); - info!("using custom relay: {url}"); + tracing::info!("using custom relay: {url}"); } } let endpoint = builder.bind().await?; let endpoint_addr = endpoint.addr(); let ep_id = endpoint_addr.id; - info!("endpoint online: {}", ep_id.fmt_short()); + 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); @@ -148,7 +144,7 @@ async fn main() -> Result<()> { while let Some(bytes) = cmd_rx.recv().await { if let Ok(sock) = std::os::unix::net::UnixDatagram::unbound() { if sock.send_to(&bytes, &cmd_path).is_err() { - warn!("nescapture cmd send failed at {}", cmd_path.display()); + tracing::warn!("nescapture cmd send failed at {}", cmd_path.display()); } } } @@ -179,13 +175,21 @@ async fn main() -> Result<()> { { let mgr = session_manager.clone(); let audio_channels = args.audio_channels as u8; - let audio_kbps = args.audio_channels * args.audio_bitrate_per_channel; + // The configured target is worth saying once, here, where it is a fact + // about this hub's arguments. It is deliberately not what gets reported + // in the stats below -- see `SessionManager::audio_bitrate_kbps`. + tracing::info!( + "audio configured for {}ch at {}kbps/channel; stats report measured ingest", + args.audio_channels, + args.audio_bitrate_per_channel + ); tokio::spawn(async move { let mut interval = tokio::time::interval(std::time::Duration::from_secs(1)); loop { interval.tick().await; let clients = mgr.client_count().await as u8; let bitrate = mgr.video_bitrate_bps(); + let audio_kbps = mgr.audio_bitrate_kbps(); let relay_ms = mgr.relay_ms(); let mut buf = Vec::with_capacity(15); nesprotocol::stats::encode_hub_stats( @@ -201,15 +205,17 @@ async fn main() -> Result<()> { }); } - // The ticket is how a client finds this endpoint. It is generated here - // because the endpoint is here, and served on a socket because the only - // thing that can carry it out of the VM is nesinit. + // ── Accept mode: generate ticket, wait for desktop-app to connect ───── let stream_name = ticket::generate_stream_name(); let ticket = NestriTicket::new(endpoint_addr, stream_name); - info!(%ticket, "endpoint ticket generated"); - ticket::serve(&args.ticket_ipc, ticket.to_string())?; + + tracing::info!("\n╔═══════════════╗"); + tracing::info!("║ NESTRI TICKET ║"); + tracing::info!("╚═══════════════╝"); + tracing::info!("{ticket}\n"); // Spawn IPC listeners + let video_ipc = args.video_ipc.clone(); let audio_ipc = args.audio_ipc.clone(); let input_ipc = args.input_ipc.clone(); @@ -237,6 +243,11 @@ async fn main() -> Result<()> { async move { ipc_listener::run_stats_ipc_listener(stats_ipc, stx).await } }); + let ticket_ipc = args.ticket_ipc.clone(); + tokio::spawn({ + async move { ipc_listener::run_ticket_ipc_listener(ticket_ipc, ticket).await } + }); + // Accept loop let ep = endpoint.clone(); let mgr = session_manager.clone(); @@ -245,7 +256,7 @@ async fn main() -> Result<()> { match incoming.await { Ok(conn) => { let remote_id = conn.remote_id(); - info!(remote = %remote_id.fmt_short(), "client connected"); + tracing::info!(remote = %remote_id.fmt_short(), "client connected"); let session = session::ClientSession::new( conn.clone(), input_broadcast_tx.clone(), @@ -261,11 +272,11 @@ async fn main() -> Result<()> { }); } Err(e) => { - warn!("incoming connection failed: {e}"); + tracing::warn!("incoming connection failed: {e}"); } } } - info!("accept loop exited"); + tracing::info!("accept loop exited"); }); // Not wired to anything today. Kept because the capture works and "show me @@ -274,18 +285,14 @@ async fn main() -> Result<()> { let _screenshots = match screenshot::listen(&args.screenshot_ipc) { Ok(connection) => Some(connection), Err(e) => { - warn!("screenshots unavailable: {e:#}"); + tracing::warn!("screenshots unavailable: {e:#}"); None } }; - // neshub outlives every session and every payload. Nothing here decides - // when the box is done -- nesinit owns that, and shuts the VM down around - // this process. - info!("neshub running"); + tracing::info!("neshub running, ctrl+c to stop"); tokio::signal::ctrl_c().await?; - - info!("shutting down.."); + tracing::info!("shutting down.."); endpoint.close().await; accept_handle.abort(); @@ -293,7 +300,7 @@ async fn main() -> Result<()> { let _ = std::fs::remove_file(&args.audio_ipc); let _ = std::fs::remove_file(&args.input_ipc); let _ = std::fs::remove_file(&args.stats_ipc); - let _ = std::fs::remove_file(&args.ticket_ipc); let _ = std::fs::remove_file("/tmp/nescapture-cmd.sock"); + let _ = std::fs::remove_file(&args.ticket_ipc); Ok(()) } diff --git a/apps/neshub/src/session.rs b/apps/neshub/src/session.rs index 785b4cb7..904de104 100644 --- a/apps/neshub/src/session.rs +++ b/apps/neshub/src/session.rs @@ -5,11 +5,14 @@ use iroh::endpoint::Connection; use tokio::sync::Mutex; use tracing::{debug, info, warn}; +use nesprotocol::datagram::{DGRAM_AUDIO, DGRAM_VIDEO}; use nesprotocol::input::{INPUT_KEY, INPUT_MOUSE_BUTTON, INPUT_MOUSE_MOVE, INPUT_MOUSE_WHEEL}; -use nesprotocol::{BIDI_INPUT, STREAM_AUDIO, STREAM_CURSOR, STREAM_STATS, STREAM_VIDEO}; -use nesprotocol::{FRAME_HDR_LEN, MSG_DATA, STREAM_VERSION, encode_frame}; +use nesprotocol::{BIDI_INPUT, STREAM_CURSOR, STREAM_STATS}; +use nesprotocol::{FRAME_HDR_LEN, STREAM_VERSION, encode_frame}; use nesprotocol::{MSG_ENCODE_SETTINGS, MSG_IDR_REQUEST, MSG_INPUT_BATCH}; +use crate::dgram::run_datagram_writer; + use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; pub struct ClientSession { @@ -36,12 +39,23 @@ impl ClientSession { let (cursor_tx, cursor_rx) = tokio::sync::mpsc::unbounded_channel::>(); let (stats_tx, stats_rx) = tokio::sync::mpsc::unbounded_channel::>(); + // Delta frames and audio go out as datagrams; cursor, stats and input + // stay on reliable streams. See `nestri_protocol::datagram` for why. + // + // Video keyframes are the exception: each goes on a reliable stream of + // its own, because a lost keyframe freezes the picture until the next + // one instead of costing a single frame. See + // `nestri_protocol::reliable`. Audio is not offered the same path — it + // has no keyframes to promote. let conn_v = conn.clone(); - let _video_task = - tokio::spawn(async move { run_video_sender(conn_v, video_rx, relay_ms).await }); + let _video_task = tokio::spawn(async move { + run_datagram_writer(conn_v, DGRAM_VIDEO, "video", video_rx, Some(relay_ms), true).await + }); let conn_a = conn.clone(); - let _audio_task = tokio::spawn(async move { run_audio_sender(conn_a, audio_rx).await }); + let _audio_task = tokio::spawn(async move { + run_datagram_writer(conn_a, DGRAM_AUDIO, "audio", audio_rx, None, false).await + }); let conn_c = conn.clone(); let _cursor_task = tokio::spawn(async move { run_cursor_sender(conn_c, cursor_rx).await }); @@ -218,131 +232,6 @@ async fn run_input_reader( debug!("input reader exiting"); } -async fn run_video_sender( - conn: Connection, - mut rx: tokio::sync::mpsc::UnboundedReceiver>, - relay_ms: Arc, -) { - loop { - let first = match rx.recv().await { - Some(data) => data, - None => { - debug!("video sender exiting (channel closed)"); - return; - } - }; - - let mut send = match conn.open_uni().await { - Ok(s) => s, - Err(e) => { - debug!("video open_uni failed: {e}"); - break; - } - }; - debug!("video uni stream opened"); - - if send - .write_all(&[STREAM_VIDEO, STREAM_VERSION]) - .await - .is_err() - { - let _ = send.finish(); - break; - } - - let mut seq: u16 = 0; - let mut buf = Vec::with_capacity(FRAME_HDR_LEN + first.len()); - encode_frame(&mut buf, MSG_DATA, seq, &first); - if send.write_all(&buf).await.is_err() { - let _ = send.finish(); - break; - } - seq = seq.wrapping_add(1); - - loop { - match rx.recv().await { - Some(bytes) => { - let t0 = std::time::Instant::now(); - buf.clear(); - encode_frame(&mut buf, MSG_DATA, seq, &bytes); - if send.write_all(&buf).await.is_err() { - break; - } - let elapsed = t0.elapsed().as_secs_f32() * 1000.0; - relay_ms.store(elapsed.to_bits(), Ordering::Relaxed); - seq = seq.wrapping_add(1); - } - None => { - let _ = send.finish(); - debug!("video sender exiting (channel closed)"); - return; - } - } - } - let _ = send.finish(); - } - debug!("video sender exiting"); -} - -async fn run_audio_sender(conn: Connection, mut rx: tokio::sync::mpsc::UnboundedReceiver>) { - loop { - let first = match rx.recv().await { - Some(data) => data, - None => { - debug!("audio sender exiting (channel closed)"); - return; - } - }; - - let mut send = match conn.open_uni().await { - Ok(s) => s, - Err(e) => { - debug!("audio open_uni failed: {e}"); - break; - } - }; - debug!("audio uni stream opened"); - - if send - .write_all(&[STREAM_AUDIO, STREAM_VERSION]) - .await - .is_err() - { - let _ = send.finish(); - break; - } - - let mut seq: u16 = 0; - let mut buf = Vec::with_capacity(FRAME_HDR_LEN + first.len()); - encode_frame(&mut buf, MSG_DATA, seq, &first); - if send.write_all(&buf).await.is_err() { - let _ = send.finish(); - break; - } - seq = seq.wrapping_add(1); - - loop { - match rx.recv().await { - Some(bytes) => { - buf.clear(); - encode_frame(&mut buf, MSG_DATA, seq, &bytes); - if send.write_all(&buf).await.is_err() { - break; - } - seq = seq.wrapping_add(1); - } - None => { - let _ = send.finish(); - debug!("audio sender exiting (channel closed)"); - return; - } - } - } - let _ = send.finish(); - } - debug!("audio sender exiting"); -} - async fn run_cursor_sender( conn: Connection, mut rx: tokio::sync::mpsc::UnboundedReceiver>, @@ -477,6 +366,8 @@ pub struct SessionManager { video_bytes: AtomicU64, last_video_bytes: AtomicU64, video_bitrate: AtomicU64, // bytes/sec + audio_bytes: AtomicU64, + last_audio_bytes: AtomicU64, relay_ms: Arc, // latest relay latency (f32 bits) } @@ -487,6 +378,8 @@ impl SessionManager { video_bytes: AtomicU64::new(0), last_video_bytes: AtomicU64::new(0), video_bitrate: AtomicU64::new(0), + audio_bytes: AtomicU64::new(0), + last_audio_bytes: AtomicU64::new(0), relay_ms: Arc::new(AtomicU32::new(0)), } } @@ -516,6 +409,11 @@ impl SessionManager { } pub async fn broadcast_audio(&self, data: Vec) { + // Counted before the early return, so the figure measures what neswire + // delivered rather than what a client happened to be around for. A hub + // with no client still knows whether audio is being produced. + self.audio_bytes + .fetch_add(data.len() as u64, Ordering::Relaxed); let sessions = self.sessions.lock().await; if sessions.is_empty() { return; @@ -557,6 +455,25 @@ impl SessionManager { self.sessions.lock().await.len() } + /// Opus actually received from neswire since the last call, in kbps. + /// + /// Measured, not configured. The reported figure used to be + /// `channels * bitrate_per_channel` straight off the hub's own command line, + /// which is a constant: it read 128 kbps whether neswire was feeding the + /// socket, feeding it silence, or had never sent a byte. A stat that cannot + /// be wrong cannot be evidence of anything. + /// + /// Like [`video_bitrate_bps`], this assumes the caller ticks once a second + /// -- the difference since the previous call *is* the per-second figure. + /// + /// [`video_bitrate_bps`]: Self::video_bitrate_bps + pub fn audio_bitrate_kbps(&self) -> u32 { + let current = self.audio_bytes.load(Ordering::Relaxed); + let last = self.last_audio_bytes.swap(current, Ordering::Relaxed); + let diff = current.saturating_sub(last); + (diff * 8 / 1000) as u32 + } + pub fn video_bitrate_bps(&self) -> u32 { let current = self.video_bytes.load(Ordering::Relaxed); let last = self.last_video_bytes.swap(current, Ordering::Relaxed); diff --git a/apps/neshub/src/ticket.rs b/apps/neshub/src/ticket.rs index 9f30cadc..0fc4e758 100644 --- a/apps/neshub/src/ticket.rs +++ b/apps/neshub/src/ticket.rs @@ -1,18 +1,7 @@ -//! The ticket: how a client finds this box. -//! -//! neshub makes one per boot and serves it on a Unix socket. It listens and -//! nesinit dials, the same way every other socket here works -- so there is no -//! race against a process that has not started yet, and no file left on disk -//! holding a live address after the box is gone. - -use std::path::Path; - -use anyhow::{Context, Result}; +use anyhow::Result; use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD}; use iroh::EndpointAddr; use serde::{Deserialize, Serialize}; -use tokio::io::AsyncWriteExt; -use tokio::net::UnixListener; /// A Nestri connection ticket. /// @@ -64,39 +53,6 @@ pub fn generate_stream_name() -> String { format!("stream-{}", uuid::Uuid::now_v7().as_simple()) } -/// Serve the ticket to anything that dials, forever. -/// -/// Returns once the socket is bound, not when someone connects: the caller has -/// an endpoint to run and a reader that never arrives is not a reason to refuse -/// clients. Every connection gets the ticket and a newline, then is dropped -- -/// there is nothing to say afterwards, and a reader blocked on more would hang. -pub fn serve(path: &Path, ticket: String) -> Result<()> { - // A stale socket file makes bind fail with EADDRINUSE, which reads as - // "something is already listening" when nothing is. - let _ = std::fs::remove_file(path); - let listener = UnixListener::bind(path) - .with_context(|| format!("could not listen on {}", path.display()))?; - tracing::info!("serving the ticket on {}", path.display()); - - tokio::spawn(async move { - let line = format!("{ticket}\n"); - loop { - match listener.accept().await { - Ok((mut stream, _)) => { - if let Err(e) = stream.write_all(line.as_bytes()).await { - tracing::warn!("could not hand over the ticket: {e}"); - } - } - Err(e) => { - tracing::warn!("accepting a ticket connection failed: {e}"); - return; - } - } - } - }); - Ok(()) -} - #[cfg(test)] mod tests { use super::*; diff --git a/apps/neswire/src/encoder.rs b/apps/neswire/src/encoder.rs index 1f5509ef..a7eedcc1 100644 --- a/apps/neswire/src/encoder.rs +++ b/apps/neswire/src/encoder.rs @@ -402,8 +402,7 @@ mod tests { let start = Instant::now(); let mut opus_buf = vec![0u8; 4000]; - send_frame(&encoder, &sender, &config, &tone(0), &start, &mut opus_buf) - .expect("send"); + send_frame(&encoder, &sender, &config, &tone(0), &start, &mut opus_buf).expect("send"); let mut buf = vec![0u8; 65536]; let n = listener.recv(&mut buf).expect("recv"); diff --git a/crates/nesprotocol/src/datagram.rs b/crates/nesprotocol/src/datagram.rs index 2eb13c2c..165d7283 100644 --- a/crates/nesprotocol/src/datagram.rs +++ b/crates/nesprotocol/src/datagram.rs @@ -193,7 +193,11 @@ mod tests { let mut framed = Vec::new(); crate::encode_frame(&mut framed, crate::MSG_DATA, 42, &[9, 8, 7]); - assert_eq!(&framed[4..], &body[..], "body is the frame minus its prefix"); + assert_eq!( + &framed[4..], + &body[..], + "body is the frame minus its prefix" + ); let (msg_type, seq, payload) = crate::decode_frame(&body).expect("body parses"); assert_eq!(msg_type, crate::MSG_DATA);