diff --git a/apps/nescope/src/input_ipc.rs b/apps/nescope/src/input_ipc.rs index a159a644..861590a3 100644 --- a/apps/nescope/src/input_ipc.rs +++ b/apps/nescope/src/input_ipc.rs @@ -1,12 +1,16 @@ use std::io::{self, Read}; use std::os::unix::io::{AsFd, BorrowedFd}; use std::os::unix::net::UnixStream; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; use calloop::{EventSource, Interest, Mode, Poll, PostAction, Readiness, Token, TokenFactory}; pub struct InputIpcSource { stream: UnixStream, buf: Vec, + /// Set once the hub has closed its end, for the main loop to act on. + closed: Arc, } impl InputIpcSource { @@ -16,14 +20,39 @@ impl InputIpcSource { Ok(Self { stream, buf: Vec::new(), + closed: Arc::new(AtomicBool::new(false)), }) } + /// True once the hub has closed the socket. The main loop gives that a + /// grace period and then ends nescope: a compositor with no input holds + /// the GPU for nothing. + pub fn closed_flag(&self) -> Arc { + self.closed.clone() + } + pub fn try_clone(&self) -> io::Result { self.stream.try_clone() } } +impl InputIpcSource { + /// Read everything available into `buf`. `true` if the peer has closed. + fn fill(&mut self) -> io::Result { + let mut tmp = [0u8; 4096]; + loop { + match self.stream.read(&mut tmp) { + Ok(0) => return Ok(true), + Ok(n) => self.buf.extend_from_slice(&tmp[..n]), + Err(e) if e.kind() == io::ErrorKind::WouldBlock => return Ok(false), + Err(e) if e.kind() == io::ErrorKind::Interrupted => {} + Err(e) if e.kind() == io::ErrorKind::ConnectionReset => return Ok(true), + Err(e) => return Err(e), + } + } + } +} + impl AsFd for InputIpcSource { fn as_fd(&self) -> BorrowedFd<'_> { self.stream.as_fd() @@ -45,22 +74,13 @@ impl EventSource for InputIpcSource { where F: FnMut(Self::Event, &mut Self::Metadata), { - let mut tmp = [0u8; 4096]; - loop { - match self.stream.read(&mut tmp) { - Ok(0) => { - return Err(io::Error::new( - io::ErrorKind::UnexpectedEof, - "IPC socket closed", - )); - } - Ok(n) => { - self.buf.extend_from_slice(&tmp[..n]); - } - Err(e) if e.kind() == io::ErrorKind::WouldBlock => break, - Err(e) => return Err(e), - } - } + // The hub closing its end is not an error of this source: it is what + // the hub does when the box is going down (init signals every process + // at once, so the hub can be gone before our own SIGTERM is read), and + // an `Err` here is returned out of the event loop and panics `main`, + // skipping the cleanup after it. Frames already read are still + // delivered, then the source removes itself and the loop carries on. + let closed = self.fill()?; while self.buf.len() >= 2 { let payload_len = u16::from_le_bytes([self.buf[0], self.buf[1]]) as usize; @@ -73,6 +93,11 @@ impl EventSource for InputIpcSource { callback(payload, &mut ()); } + if closed { + self.closed.store(true, Ordering::Relaxed); + tracing::warn!("input IPC socket closed by the hub; no more input from it"); + return Ok(PostAction::Remove); + } Ok(PostAction::Continue) } @@ -88,3 +113,44 @@ impl EventSource for InputIpcSource { poll.unregister(self.as_fd()) } } + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Write; + + fn pair() -> (InputIpcSource, UnixStream) { + let (a, b) = UnixStream::pair().unwrap(); + a.set_nonblocking(true).unwrap(); + ( + InputIpcSource { + stream: a, + buf: Vec::new(), + closed: Arc::new(AtomicBool::new(false)), + }, + b, + ) + } + + #[test] + fn an_open_peer_with_nothing_to_say_is_not_closed() { + let (mut source, _peer) = pair(); + assert!(!source.fill().unwrap()); + } + + #[test] + fn a_closed_peer_is_closed_not_an_error() { + let (mut source, peer) = pair(); + drop(peer); + assert!(source.fill().unwrap()); + } + + #[test] + fn what_the_peer_sent_before_closing_is_kept() { + let (mut source, mut peer) = pair(); + peer.write_all(&[1, 0, 7]).unwrap(); + drop(peer); + assert!(source.fill().unwrap()); + assert_eq!(source.buf, vec![1, 0, 7]); + } +} diff --git a/apps/nescope/src/main.rs b/apps/nescope/src/main.rs index 425def56..eef9469f 100644 --- a/apps/nescope/src/main.rs +++ b/apps/nescope/src/main.rs @@ -389,11 +389,15 @@ fn main() { // ── Input IPC source ───────────────────────────────────────────────── // Connect to the neshub input socket and feed events into the - // compositor seat. Reconnection is handled in the idle callback. + // compositor seat. There is no reconnection: when the hub closes the socket + // the source removes itself, and the loop below ends nescope after a short + // grace (a signal that arrives first ends it sooner). let ipc_path = args.input_ipc.clone(); + let mut hub_closed_flag: Option> = None; match input_ipc::InputIpcSource::connect(&ipc_path) { Ok(source) => { tracing::info!("Connected to input IPC socket: {ipc_path}"); + hub_closed_flag = Some(source.closed_flag()); match source.try_clone() { Ok(write_stream) => { state.ipc_write = Some(write_stream); @@ -476,6 +480,7 @@ fn main() { // never arrives there is nothing to wait for and no game to run. let startup = std::time::Instant::now(); let mut xwayland_timed_out = false; + let mut hub_closed_at: Option = None; event_loop .run(Some(Duration::from_secs(1)), &mut data, move |data| { @@ -578,6 +583,24 @@ fn main() { } } + // ── Hub gone ────────────────────────────────────────────── + // neshub closing the input socket mid-run leaves a compositor + // nobody can drive. The grace lets a shutdown that is already + // under way (SIGTERM stops the loop itself) finish its own way; + // otherwise nescope ends, and nesinit ends the box on seeing it. + if hub_closed_flag + .as_ref() + .is_some_and(|f| f.load(std::sync::atomic::Ordering::Relaxed)) + { + let since = hub_closed_at.get_or_insert_with(std::time::Instant::now); + if since.elapsed() > Duration::from_secs(2) { + tracing::error!("input hub gone for 2 s with no shutdown — exiting"); + kill_all_children(); + data.loop_signal.stop(); + return; + } + } + // ── Poll primary process ────────────────────────────────── // The launcher (e.g. Steam's shell wrapper) may exit quickly // while the real game client stays alive as a reparented