diff --git a/corcovado/src/sys/unix/epoll.rs b/corcovado/src/sys/unix/epoll.rs index 19161940ea..8186cf52f8 100644 --- a/corcovado/src/sys/unix/epoll.rs +++ b/corcovado/src/sys/unix/epoll.rs @@ -268,8 +268,19 @@ impl Events { } pub fn push_event(&mut self, event: Event) { + // Unlike registration interests, injected events must retain HUP and + // error readiness. The kernel reports these flags automatically for + // file descriptors, but user-space registrations have no kernel event. + let readiness = UnixReady::from(event.readiness()); + let mut events = ioevent_to_epoll(event.readiness(), PollOpt::empty()); + if readiness.is_hup() { + events |= EPOLLHUP as u32; + } + if readiness.is_error() { + events |= EPOLLERR as u32; + } self.events.push(libc::epoll_event { - events: ioevent_to_epoll(event.readiness(), PollOpt::empty()), + events, u64: usize::from(event.token()) as u64, }); } @@ -297,3 +308,23 @@ pub fn millis(duration: Duration) -> u64 { .saturating_mul(MILLIS_PER_SEC) .saturating_add(millis as u64) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn injected_events_preserve_hangup_and_error_readiness() { + for readiness in [ + Ready::from(UnixReady::hup()), + Ready::from(UnixReady::error()), + Ready::readable() | UnixReady::hup() | UnixReady::error(), + ] { + let mut events = Events::with_capacity(1); + events.push_event(Event::new(readiness, Token(42))); + let event = events.get(0).unwrap(); + assert_eq!(event.readiness(), readiness); + assert_eq!(event.token(), Token(42)); + } + } +} diff --git a/frontends/rioterm/src/application.rs b/frontends/rioterm/src/application.rs index eb9602b5c3..d464130413 100644 --- a/frontends/rioterm/src/application.rs +++ b/frontends/rioterm/src/application.rs @@ -664,7 +664,7 @@ impl ApplicationHandler for Application<'_> { if self.config.confirm_before_quit { route.confirm_quit(); } else { - route.quit(); + event_loop.exit(); } } } @@ -2036,6 +2036,10 @@ impl ApplicationHandler for Application<'_> { .. } => { if route.has_key_wait(&key_event, &mut self.router.clipboard) { + if route.quit_requested { + event_loop.exit(); + return; + } if route.path != RoutePath::Terminal && key_event.state == ElementState::Released { diff --git a/frontends/rioterm/src/context/mod.rs b/frontends/rioterm/src/context/mod.rs index 446d2fc2a1..1a2785a9d4 100644 --- a/frontends/rioterm/src/context/mod.rs +++ b/frontends/rioterm/src/context/mod.rs @@ -51,6 +51,8 @@ pub struct Context { pub main_fd: Arc, #[cfg(not(target_os = "windows"))] pub shell_pid: u32, + #[cfg(not(target_os = "windows"))] + child_terminator: teletypewriter::ChildTerminator, pub rich_text_id: usize, pub dimension: ContextDimension, pub title: ContextTitle, @@ -69,13 +71,12 @@ impl Drop for Context { fn drop(&mut self) { // Shutdown the terminal's PTY. let _ = self.messenger.channel.send(Msg::Shutdown); - - // `create_dead_context` uses 1 as a placeholder PID, so guard against - // signalling init (1) or our own process group (0). + // Also hang up synchronously: quit paths call process::exit + // right after dropping routes, before the reader thread can run + // its shutdown escalation. The handle is a no-op once the child + // was reaped, so no stale PID is ever signaled. #[cfg(not(target_os = "windows"))] - if self.shell_pid > 1 { - teletypewriter::kill_pid(self.shell_pid as i32); - } + let _ = self.child_terminator.hangup(); } } @@ -190,6 +191,8 @@ pub fn create_dead_context( main_fd: Arc::new(-1), #[cfg(not(target_os = "windows"))] shell_pid: 1, + #[cfg(not(target_os = "windows"))] + child_terminator: teletypewriter::ChildTerminator::retired(), messenger: Messenger::new(sender), renderable_content: RenderableContent::new(Cursor::default()), terminal, @@ -341,6 +344,8 @@ impl ContextManager { let main_fd = pty.child.id.clone(); #[cfg(not(target_os = "windows"))] let shell_pid = *pty.child.pid.clone() as u32; + #[cfg(not(target_os = "windows"))] + let child_terminator = pty.child.terminator(); #[cfg(target_os = "windows")] { @@ -382,6 +387,8 @@ impl ContextManager { main_fd, #[cfg(not(target_os = "windows"))] shell_pid, + #[cfg(not(target_os = "windows"))] + child_terminator, messenger, terminal, rich_text_id, diff --git a/frontends/rioterm/src/router/mod.rs b/frontends/rioterm/src/router/mod.rs index dc1f624db6..2a0be62e33 100644 --- a/frontends/rioterm/src/router/mod.rs +++ b/frontends/rioterm/src/router/mod.rs @@ -53,6 +53,10 @@ pub struct Route<'a> { pub assistant: assistant::Assistant, pub path: RoutePath, pub window: RouteWindow<'a>, + /// Set by `quit`; the application answers it with an event loop + /// exit so `exiting` drops every route (hanging up each PTY child) + /// before the process exits. + pub quit_requested: bool, } impl Route<'_> { @@ -67,6 +71,7 @@ impl Route<'_> { assistant, path, window, + quit_requested: false, } } } @@ -253,7 +258,10 @@ impl Route<'_> { #[inline] pub fn quit(&mut self) { - std::process::exit(0); + // A direct process::exit here would skip every destructor: no + // Msg::Shutdown, no hangup, and PTY children ignoring the + // kernel's HUP-on-master-close would be orphaned. + self.quit_requested = true; } #[inline] @@ -659,6 +667,7 @@ impl Router<'_> { window, path: RoutePath::Terminal, assistant: Assistant::new(), + quit_requested: false, }; if let Some(err) = &self.propagated_report { @@ -696,6 +705,7 @@ impl Router<'_> { window, path: RoutePath::Terminal, assistant: Assistant::new(), + quit_requested: false, }, ); self.quake_window_id = Some(id); @@ -728,6 +738,7 @@ impl Router<'_> { window, path: RoutePath::Terminal, assistant: Assistant::new(), + quit_requested: false, }, ); } diff --git a/librio/src/lib.rs b/librio/src/lib.rs index 4ce2067906..fda8134f9e 100644 --- a/librio/src/lib.rs +++ b/librio/src/lib.rs @@ -356,6 +356,8 @@ pub struct Surface { shell_pid: u32, #[cfg(all(feature = "pty", not(target_os = "windows")))] main_fd: std::os::fd::RawFd, + #[cfg(all(feature = "pty", not(target_os = "windows")))] + child_terminator: teletypewriter::ChildTerminator, #[cfg(feature = "pty")] _io_thread: std::thread::JoinHandle<( Machine, @@ -511,6 +513,8 @@ impl Surface { let shell_pid = pty.child_watcher().pid().map(|pid| pid.get()).unwrap_or(0); #[cfg(not(target_os = "windows"))] let main_fd = *pty.child.id; + #[cfg(not(target_os = "windows"))] + let child_terminator = pty.child.terminator(); let machine = Machine::new( Arc::clone(&terminal), @@ -535,6 +539,8 @@ impl Surface { shell_pid, #[cfg(not(target_os = "windows"))] main_fd, + #[cfg(not(target_os = "windows"))] + child_terminator, _io_thread: io_thread, }) } @@ -1024,11 +1030,12 @@ impl Surface { } /// The pid of the program this surface spawned (the shell, or the - /// configured `shell` program). On unix it is a session leader, so a - /// host that must take the whole process tree down on teardown can - /// `killpg` it: dropping the surface only hangs up the pty and signals - /// this pid. On Windows it is the conpty child's process id (terminate - /// it with `TerminateProcess`/taskkill); 0 if the pid was unavailable. + /// configured `shell` program), for identity and diagnostics. Do not + /// signal it on teardown: dropping the surface already hangs up the + /// process group, and the reader thread escalates to SIGKILL and + /// reaps, so a host-side killpg would race that escalation. On + /// Windows it is the conpty child's process id; 0 if the pid was + /// unavailable. #[cfg(feature = "pty")] pub fn child_pid(&self) -> u32 { self.shell_pid @@ -1186,8 +1193,12 @@ impl Surface { impl Drop for Surface { fn drop(&mut self) { let _ = self.channel.send(Msg::Shutdown); + // Also hang up synchronously: an embedder may exit the process + // right after dropping the surface, before the reader thread can + // run its shutdown escalation. The handle is a no-op once the + // child was reaped, so no stale PID is ever signaled. #[cfg(not(target_os = "windows"))] - teletypewriter::kill_pid(self.shell_pid as i32); + let _ = self.child_terminator.hangup(); } } diff --git a/rio-vt/README.md b/rio-vt/README.md index c098ca72f6..50a4fbc559 100644 --- a/rio-vt/README.md +++ b/rio-vt/README.md @@ -134,6 +134,20 @@ automatically. See [`librio/src/lib.rs`](../librio/src/lib.rs) for a complete, working setup (`Crosswords::new` + `Machine::new` + `teletypewriter`), which is also the reference consumer of this crate. +On child exit, the reader drains available output across parsing budgets and +flushes pending synchronized updates before publishing `ChildExited` and +`CloseTerminal`. It does not wait for descendants holding the slave open: +a read that would block ends the drain, and continuous descendant output is +limited to 100 ms between parsing batches. Terminal-lock waits and parsing +can extend that interval. + +Send `Msg::Shutdown` to close a running PTY. On Unix, shutdown sends SIGHUP, +allows a 100 ms grace period, then uses SIGKILL if necessary and waits to +reap the child. Repeated shutdown preserves its exit status and does not +signal a reaped child. The final reap can take longer for a process stuck +in the kernel; shutdown is not a strict wall-clock deadline. Embedders +should let the PTY owner manage termination instead of signaling a saved PID. + ## Pull-based rendering `rio-vt` does not draw anything. A frontend reads terminal state on demand: diff --git a/rio-vt/src/performer/mod.rs b/rio-vt/src/performer/mod.rs index ad6d682c8a..d7d2acf927 100644 --- a/rio-vt/src/performer/mod.rs +++ b/rio-vt/src/performer/mod.rs @@ -2,6 +2,9 @@ pub mod handler; mod osc; pub mod parser; +#[cfg(all(test, feature = "pty"))] +mod tests; + #[cfg(feature = "pty")] use crate::crosswords::Crosswords; #[cfg(feature = "pty")] @@ -63,6 +66,20 @@ const READ_CHUNK: usize = 65536; #[cfg(feature = "pty")] const MAX_LOCKED_READ: usize = READ_CHUNK * 4; +#[cfg(feature = "pty")] +#[derive(Debug, PartialEq, Eq)] +enum ReadOutcome { + Idle, + Closed, + Budget, +} + +#[cfg(feature = "pty")] +enum ExitReason { + Shutdown, + ChildExited(Option), +} + // Guards the pairing that once regressed: a MAX_LOCKED_READ below // READ_CHUNK ends the burst loop after a single partial chunk. #[cfg(feature = "pty")] @@ -220,9 +237,10 @@ where /// throughput; the confirming read that ends a burst costs a /// single `EAGAIN`. #[inline] - fn pty_read(&mut self, state: &mut State, buf: &mut [u8]) -> io::Result<()> { + fn pty_read(&mut self, state: &mut State, buf: &mut [u8]) -> io::Result { let mut unprocessed = 0; let mut processed = 0; + let mut result = Ok(ReadOutcome::Budget); // Reserve the next terminal lock for PTY reading. let _terminal_lease = Some(self.terminal.lease()); @@ -231,19 +249,27 @@ where loop { // Read from the PTY. let cap = (unprocessed + READ_CHUNK).min(buf.len()); - match self.pty.reader().read(&mut buf[unprocessed..cap]) { - // This is received on Windows/macOS when no more data is readable from the PTY. - Ok(0) if unprocessed == 0 => break, - Ok(got) => unprocessed += got, - Err(err) => match err.kind() { - ErrorKind::Interrupted | ErrorKind::WouldBlock => { - // Go back to mio if we're caught up on parsing and the PTY would block. - if unprocessed == 0 { - break; - } - } - _ => return Err(err), - }, + let stopped = match self.pty.reader().read(&mut buf[unprocessed..cap]) { + Ok(0) => { + result = Ok(ReadOutcome::Closed); + true + } + Ok(got) => { + unprocessed += got; + false + } + Err(err) if err.kind() == ErrorKind::Interrupted => continue, + Err(err) if err.kind() == ErrorKind::WouldBlock => { + result = Ok(ReadOutcome::Idle); + true + } + Err(err) => { + result = Err(err); + true + } + }; + if stopped && unprocessed == 0 { + break; } // Attempt to lock the terminal. @@ -251,7 +277,7 @@ where Some(terminal) => terminal, None => terminal.insert(match self.terminal.try_lock_unfair() { // Force block if we are at the buffer size limit. - None if unprocessed >= READ_BUFFER_SIZE => { + None if stopped || unprocessed == buf.len() => { self.terminal.lock_unfair() } None => continue, @@ -266,7 +292,7 @@ where unprocessed = 0; // Assure we're not blocking the terminal too long unnecessarily. - if processed >= MAX_LOCKED_READ { + if stopped || processed >= MAX_LOCKED_READ { break; } } @@ -286,7 +312,7 @@ where } } - Ok(()) + result } /// Drain the channel. @@ -347,178 +373,215 @@ where spawn_named("PTY reader", move || { let mut state = State::default(); let mut buf = [0u8; READ_BUFFER_SIZE]; + let reason = self.run(&mut state, &mut buf); + self.finish(&mut state, &mut buf, reason); + (self, state) + }) + } - let mut tokens = (0..).map(Into::into); - - // The channel is drained to empty on every wakeup, which clears - // its readiness and re-arms the next edge transition, so plain - // edge (no oneshot, no re-registration) is enough. Level would go - // through the readiness queue's re-enqueue path, which is much - // more expensive per wakeup. - let channel_token = tokens.next().unwrap(); - self.poll - .register( - &self.receiver.rx, - channel_token, - Ready::readable(), - PollOpt::edge(), - ) - .unwrap(); - - // The PTY is level-triggered: pty_read may stop before draining - // the fd (MAX_LOCKED_READ), which would lose an edge, and level - // registrations stay armed so no re-registration is needed after - // each event. The write interest must be dropped as soon as the - // write queue drains or the poll would keep waking up for the - // writable PTY. - let poll_opts = PollOpt::level(); - - // Register TTY through EventedRW interface. - self.pty - .register(&self.poll, &mut tokens, Ready::readable(), poll_opts) - .unwrap(); - - let mut events = Events::with_capacity(1024); - let mut last_interest = Ready::readable(); - - 'event_loop: loop { - // Wakeup the event loop when a synchronized update timeout was reached. - let handler = state.parser.sync_timeout(); - let timeout = handler - .sync_timeout() - .map(|st| st.saturating_duration_since(Instant::now())); - - events.clear(); - if let Err(err) = self.poll.poll(&mut events, timeout) { - match err.kind() { - ErrorKind::Interrupted => continue, - _ => { - error!("Event loop polling error: {err}"); - break 'event_loop; - } - } - } - - // Handle synchronized update timeout. - if events.is_empty() && self.receiver.peek().is_none() { - let mut terminal = self.terminal.lock(); - state.parser.stop_sync(&mut *terminal); - - // Notify renderer if damage available and no event in flight - if !terminal.damage_event_in_flight - && terminal.peek_damage_event().is_some() - { - terminal.damage_event_in_flight = true; - self.event_proxy.send_event( - RioEvent::TerminalDamaged(self.route_id), - self.window_id, - ); - } - - continue; - } + fn run(&mut self, state: &mut State, buf: &mut [u8]) -> io::Result { + let mut tokens = (0..).map(Into::into); + + // The channel is drained to empty on every wakeup, which clears + // its readiness and re-arms the next edge transition, so plain + // edge (no oneshot, no re-registration) is enough. Level would go + // through the readiness queue's re-enqueue path, which is much + // more expensive per wakeup. + let channel_token = tokens.next().unwrap(); + self.poll.register( + &self.receiver.rx, + channel_token, + Ready::readable(), + PollOpt::edge(), + )?; + + // The PTY is level-triggered: pty_read may stop before draining + // the fd (MAX_LOCKED_READ), which would lose an edge, and level + // registrations stay armed so no re-registration is needed after + // each event. The write interest must be dropped as soon as the + // write queue drains or the poll would keep waking up for the + // writable PTY. + let poll_opts = PollOpt::level(); + + // Register TTY through EventedRW interface. + self.pty + .register(&self.poll, &mut tokens, Ready::readable(), poll_opts)?; + + let mut events = Events::with_capacity(1024); + let mut last_interest = Ready::readable(); - // Handle channel events, if there are any. - if !self.drain_recv_channel(&mut state) { - break; + loop { + // Wakeup the event loop when a synchronized update timeout was reached. + let handler = state.parser.sync_timeout(); + let timeout = handler + .sync_timeout() + .map(|st| st.saturating_duration_since(Instant::now())); + + events.clear(); + if let Err(err) = self.poll.poll(&mut events, timeout) { + match err.kind() { + ErrorKind::Interrupted => continue, + _ => return Err(err), } + } - for event in events.iter() { - match event.token() { - // Channel messages were already drained above. - token if token == channel_token => (), - token if token == self.pty.child_event_token() => { - if let Some(teletypewriter::ChildEvent::Exited(status)) = - self.pty.next_child_event() - { - // In the future allow configure exit - // if self.hold { - // With hold enabled, make sure the PTY is drained. - // let _ = self.pty_read(&mut state, &mut buf); - // } else { - // // Without hold, shutdown the terminal. - // self.terminal.lock().exit(); - // } - - // Drain whatever the child wrote before it - // exited so short-lived commands don't lose - // their final output. - if let Err(err) = self.pty_read(&mut state, &mut buf) { - tracing::debug!( - "PTY drain after child exit failed: {err}" - ); - } + // Handle synchronized update timeout. + if events.is_empty() && self.receiver.peek().is_none() { + let mut terminal = self.terminal.lock(); + state.parser.stop_sync(&mut *terminal); - self.event_proxy.send_event( - RioEvent::ChildExited(self.route_id, status), - self.window_id, - ); + // Notify renderer if damage available and no event in flight + if !terminal.damage_event_in_flight + && terminal.peek_damage_event().is_some() + { + terminal.damage_event_in_flight = true; + self.event_proxy.send_event( + RioEvent::TerminalDamaged(self.route_id), + self.window_id, + ); + } - self.terminal.lock().exit(); + continue; + } - self.event_proxy - .send_event(RioEvent::Render, self.window_id); + // Handle channel events, if there are any. + if !self.drain_recv_channel(state) { + return Ok(ExitReason::Shutdown); + } - break 'event_loop; - } + for event in events.iter() { + match event.token() { + // Channel messages were already drained above. + token if token == channel_token => (), + token if token == self.pty.child_event_token() => { + if let Some(teletypewriter::ChildEvent::Exited(status)) = + self.pty.next_child_event() + { + return Ok(ExitReason::ChildExited(status)); } + } - token - if token == self.pty.read_token() - || token == self.pty.write_token() => - { - #[cfg(unix)] - if UnixReady::from(event.readiness()).is_hup() { - // Don't try to do I/O on a dead PTY. - continue; - } - if event.readiness().is_readable() { - if let Err(err) = self.pty_read(&mut state, &mut buf) { - // On Linux, a `read` on the master side of a PTY can fail - // with `EIO` if the client side hangs up. In that case, - // just loop back round for the inevitable `Exited` event. - #[cfg(target_os = "linux")] - if err.raw_os_error() == Some(libc::EIO) { - continue; - } - - error!( - "Error reading from PTY in event loop: {}", - err - ); - break 'event_loop; + token + if token == self.pty.read_token() + || token == self.pty.write_token() => + { + #[cfg(unix)] + let hung_up = UnixReady::from(event.readiness()).is_hup(); + #[cfg(not(unix))] + let hung_up = false; + // HUP can accompany unread final output. + if event.readiness().is_readable() || hung_up { + if let Err(err) = self.pty_read(state, buf) { + // On Linux, a `read` on the master side of a PTY can fail + // with `EIO` if the client side hangs up. In that case, + // just loop back round for the inevitable `Exited` event. + #[cfg(target_os = "linux")] + if err.raw_os_error() == Some(libc::EIO) { + continue; } - } - if event.readiness().is_writable() { - if let Err(err) = self.pty_write(&mut state) { - error!("Error writing to PTY in event loop: {}", err); - break 'event_loop; - } + return Err(err); } } - _ => (), + + if !hung_up && event.readiness().is_writable() { + self.pty_write(state)?; + } } + _ => (), } + } - // Update the PTY registration when write interest changed. - let mut interest = Ready::readable(); - if state.needs_write() { - interest.insert(Ready::writable()); - } - if interest != last_interest { - self.pty - .reregister(&self.poll, interest, poll_opts) - .unwrap(); - last_interest = interest; - } + // Update the PTY registration when write interest changed. + let mut interest = Ready::readable(); + if state.needs_write() { + interest.insert(Ready::writable()); } + if interest != last_interest { + self.pty.reregister(&self.poll, interest, poll_opts)?; + last_interest = interest; + } + } + } - // The evented instances are not dropped here so deregister them explicitly. - let _ = self.poll.deregister(&self.receiver.rx); - let _ = self.pty.deregister(&self.poll); + /// All reader exits, including partial registration failures, finalize here. + fn finish( + &mut self, + state: &mut State, + buf: &mut [u8], + reason: io::Result, + ) { + if matches!(reason, Ok(ExitReason::ChildExited(_))) { + // Do not wait for descendants to close the slave. Drain across + // parsing budgets, limiting continuously available output to 100 ms + // between batches (terminal lock waits can extend this interval). + let deadline = Instant::now() + std::time::Duration::from_millis(100); + // Windows Closed means the ConPTY ring is momentarily empty + // (never WouldBlock, and no real EOF exists until the drain's + // own shutdown closes the pseudoconsole), while the pump + // thread delivers final output after the exit event. Retry it + // through a short quiet window, refreshed by each batch of + // data, so a close is not taxed the full deadline. Unix + // Closed is a real EOF and ends the drain at once. + let quiet = std::time::Duration::from_millis(25); + let mut quiet_deadline = Instant::now() + quiet; + loop { + match self.pty_read(state, buf) { + Ok(ReadOutcome::Budget) if Instant::now() < deadline => { + quiet_deadline = Instant::now() + quiet; + continue; + } + Ok(ReadOutcome::Closed) + if cfg!(windows) + && Instant::now() < deadline + && Instant::now() < quiet_deadline => + { + std::thread::sleep(std::time::Duration::from_millis(2)); + continue; + } + Err(err) => tracing::debug!("PTY final drain: {err}"), + _ => (), + } + break; + } + } - (self, state) - }) + // Flush exactly once before publishing any exit notification. + let pending_sync = state.parser.sync_bytes_count() > 0; + state.parser.stop_sync(&mut *self.terminal.lock()); + match &reason { + Ok(ExitReason::ChildExited(status)) => { + self.event_proxy.send_event( + RioEvent::ChildExited(self.route_id, *status), + self.window_id, + ); + self.terminal.lock().exit(); + } + // A reader failure also closes the terminal, since shutdown() + // below kills the child. The close is driven by exit() (it + // emits CloseTerminal, the event frontends act on); the + // ChildExited alongside is informational for embedders. + // Removing the exit() would leave a dead pane open forever. + Err(err) => { + error!("PTY reader failed: {err}"); + self.event_proxy.send_event( + RioEvent::ChildExited(self.route_id, None), + self.window_id, + ); + self.terminal.lock().exit(); + } + Ok(ExitReason::Shutdown) => (), + } + if pending_sync || !matches!(reason, Ok(ExitReason::Shutdown)) { + self.event_proxy + .send_event(RioEvent::Render, self.window_id); + } + if let Err(err) = self.pty.shutdown() { + error!("PTY shutdown failed: {err}"); + } + // These objects are retained in the returned Machine. Deregistration is + // best effort because setup may have failed before registering them. + let _ = self.poll.deregister(&self.receiver.rx); + let _ = self.pty.deregister(&self.poll); } } diff --git a/rio-vt/src/performer/tests.rs b/rio-vt/src/performer/tests.rs new file mode 100644 index 0000000000..b11fe8511c --- /dev/null +++ b/rio-vt/src/performer/tests.rs @@ -0,0 +1,342 @@ +use super::*; +use crate::ansi::CursorShape; +use crate::crosswords::pos::{Column, Line}; +use crate::crosswords::CrosswordsSize; +use crate::event::VoidListener; +use std::sync::mpsc; +use std::time::Duration; +use teletypewriter::{ChildEvent, EventedPty, ProcessReadWrite, WinsizeBuilder}; + +struct TestPty { + bytes: io::Cursor>, + ending: Option, + exhausted: Option>, + writer: Vec, + child: Option, + registration_error: bool, + read_event: bool, + shutdown_calls: usize, + deregister_calls: usize, + raw_error: Option, +} + +impl Read for TestPty { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + let got = self.bytes.read(buf)?; + if got != 0 { + return Ok(got); + } + if let Some(sender) = self.exhausted.take() { + sender.send(()).unwrap(); + } + if let Some(code) = self.raw_error { + return Err(io::Error::from_raw_os_error(code)); + } + match self.ending { + Some(kind) => Err(io::Error::from(kind)), + None => Ok(0), + } + } +} + +impl ProcessReadWrite for TestPty { + type Reader = Self; + type Writer = Vec; + fn reader(&mut self) -> &mut Self { + self + } + fn writer(&mut self) -> &mut Self::Writer { + &mut self.writer + } + fn read_token(&self) -> corcovado::Token { + 1.into() + } + fn write_token(&self) -> corcovado::Token { + 1.into() + } + fn set_winsize(&mut self, _: WinsizeBuilder) -> io::Result<()> { + Ok(()) + } + fn register( + &mut self, + poll: &corcovado::Poll, + _: &mut dyn Iterator, + _: Ready, + _: PollOpt, + ) -> io::Result<()> { + if self.registration_error { + return Err(io::Error::from(ErrorKind::Other)); + } + if let Some(child) = &self.child { + #[cfg(unix)] + let interest = Ready::readable() | Ready::from(UnixReady::hup()); + #[cfg(not(unix))] + let interest = Ready::readable(); + poll.register( + child, + if self.read_event { + self.read_token() + } else { + self.child_event_token() + }, + interest, + PollOpt::edge(), + )?; + } + Ok(()) + } + fn reregister( + &mut self, + _: &corcovado::Poll, + _: Ready, + _: PollOpt, + ) -> io::Result<()> { + Ok(()) + } + fn deregister(&mut self, _: &corcovado::Poll) -> io::Result<()> { + self.deregister_calls += 1; + Ok(()) + } +} + +impl EventedPty for TestPty { + fn shutdown(&mut self) -> io::Result<()> { + self.shutdown_calls += 1; + Ok(()) + } + fn child_event_token(&self) -> corcovado::Token { + 2.into() + } + fn next_child_event(&mut self) -> Option { + Some(ChildEvent::Exited(Some(0))) + } +} + +fn machine(bytes: Vec, ending: Option) -> Machine { + let terminal = Crosswords::new( + CrosswordsSize::new(80, 24), + CursorShape::Block, + VoidListener, + WindowId::from(0), + 0, + 0, + ); + Machine::new( + Arc::new(FairMutex::new(terminal)), + TestPty { + bytes: io::Cursor::new(bytes), + ending, + exhausted: None, + writer: Vec::new(), + child: None, + registration_error: false, + read_event: false, + shutdown_calls: 0, + deregister_calls: 0, + raw_error: None, + }, + VoidListener, + WindowId::from(0), + 0, + ) + .unwrap() +} + +fn first_line(machine: &Machine, len: usize) -> String { + let terminal = machine.terminal.lock(); + (0..len) + .map(|col| terminal.grid[Line(0)][Column(col)].c()) + .collect() +} + +// Holding the terminal until the confirming read guarantees the first batch +// cannot be parsed before EOF/error. No timing assumptions or sleeps are needed. +fn read_with_contended_terminal(ending: Option, raw_error: Option) { + let mut machine = machine(b"final output".to_vec(), ending); + machine.pty.raw_error = raw_error; + let terminal = Arc::clone(&machine.terminal); + let guard = terminal.lock(); + let (sender, receiver) = mpsc::channel(); + machine.pty.exhausted = Some(sender); + let worker = std::thread::spawn(move || { + let result = + machine.pty_read(&mut State::default(), &mut vec![0; READ_BUFFER_SIZE]); + (machine, result) + }); + let exhausted = receiver.recv_timeout(Duration::from_secs(5)); + drop(guard); + let (machine, result) = worker.join().unwrap(); + exhausted.unwrap(); + if let Some(code) = raw_error { + assert_eq!(result.unwrap_err().raw_os_error(), Some(code)); + } else { + match ending { + Some(kind) => assert_eq!(result.unwrap_err().kind(), kind), + None => assert_eq!(result.unwrap(), ReadOutcome::Closed), + } + } + assert_eq!(first_line(&machine, 12), "final output"); +} + +#[test] +fn eof_preserves_buffered_output_under_lock_contention() { + read_with_contended_terminal(None, None); +} + +#[test] +fn read_error_preserves_buffered_output_under_lock_contention() { + read_with_contended_terminal(Some(ErrorKind::Other), None); +} + +#[test] +fn final_output_can_be_drained_across_multiple_parse_budgets() { + // NUL padding consumes parse budgets without scrolling the final marker. + let mut bytes = vec![0; MAX_LOCKED_READ * 2 + 1]; + bytes.extend_from_slice(b"final output"); + let mut machine = machine(bytes, None); + let mut state = State::default(); + let mut buf = vec![0; READ_BUFFER_SIZE]; + assert_eq!( + machine.pty_read(&mut state, &mut buf).unwrap(), + ReadOutcome::Budget + ); + assert_eq!( + machine.pty_read(&mut state, &mut buf).unwrap(), + ReadOutcome::Budget + ); + assert_eq!( + machine.pty_read(&mut state, &mut buf).unwrap(), + ReadOutcome::Closed + ); + assert_eq!(first_line(&machine, 12), "final output"); +} + +#[test] +fn child_exit_drains_multiple_budgets_and_finishes_pending_sync() { + let mut bytes = vec![0; MAX_LOCKED_READ * 2 + 1]; + bytes.extend_from_slice(b"\x1b[?2026hfinal output"); + // WouldBlock models a descendant retaining the slave after the child exits. + let mut machine = machine(bytes, Some(ErrorKind::WouldBlock)); + let (registration, readiness) = corcovado::Registration::new2(); + machine.pty.child = Some(registration); + readiness.set_readiness(Ready::readable()).unwrap(); + let (machine, state) = machine.spawn().join().unwrap(); + assert_eq!(first_line(&machine, 12), "final output"); + assert_eq!(state.parser.sync_bytes_count(), 0); + assert_eq!(machine.pty.shutdown_calls, 1); + assert_eq!(machine.pty.deregister_calls, 1); +} + +#[cfg(unix)] +#[test] +fn eio_preserves_buffered_output_under_lock_contention() { + read_with_contended_terminal(None, Some(libc::EIO)); +} + +#[test] +fn registration_failure_shuts_down_pty() { + let mut machine = machine(Vec::new(), None); + machine.pty.registration_error = true; + let (machine, _) = machine.spawn().join().unwrap(); + assert_eq!(machine.pty.shutdown_calls, 1); + assert_eq!(machine.pty.deregister_calls, 1); +} + +#[test] +fn read_failure_shuts_down_pty_and_finishes_pending_sync() { + let mut machine = + machine(b"\x1b[?2026hfinal output".to_vec(), Some(ErrorKind::Other)); + let (registration, readiness) = corcovado::Registration::new2(); + machine.pty.child = Some(registration); + machine.pty.read_event = true; + readiness.set_readiness(Ready::readable()).unwrap(); + let (machine, state) = machine.spawn().join().unwrap(); + assert_eq!(machine.pty.shutdown_calls, 1); + assert_eq!(machine.pty.deregister_calls, 1); + assert_eq!(first_line(&machine, 12), "final output"); + assert_eq!(state.parser.sync_bytes_count(), 0); +} + +#[derive(Clone)] +struct ExitObserver { + terminal: Arc>>>>, + observed: mpsc::Sender, +} + +impl EventListener for ExitObserver { + fn send_event(&self, event: RioEvent, _: WindowId) { + if matches!(event, RioEvent::ChildExited(..)) { + let terminal = self.terminal.lock().unwrap().upgrade().unwrap(); + let terminal = terminal.lock(); + let text = (0..12) + .map(|col| terminal.grid[Line(0)][Column(col)].c()) + .collect(); + self.observed.send(text).unwrap(); + } + } +} + +#[test] +fn pending_sync_is_visible_before_child_exited_notification() { + let (observed, receiver) = mpsc::channel(); + let observer = ExitObserver { + terminal: Arc::new(std::sync::Mutex::new(std::sync::Weak::new())), + observed, + }; + let terminal = Arc::new(FairMutex::new(Crosswords::new( + CrosswordsSize::new(80, 24), + CursorShape::Block, + observer.clone(), + WindowId::from(0), + 0, + 0, + ))); + *observer.terminal.lock().unwrap() = Arc::downgrade(&terminal); + let mut pty = machine( + b"\x1b[?2026hfinal output".to_vec(), + Some(ErrorKind::WouldBlock), + ) + .pty; + let (registration, readiness) = corcovado::Registration::new2(); + pty.child = Some(registration); + readiness.set_readiness(Ready::readable()).unwrap(); + let machine = Machine::new(terminal, pty, observer, WindowId::from(0), 0).unwrap(); + let worker = machine.spawn(); + assert_eq!( + receiver.recv_timeout(Duration::from_secs(5)).unwrap(), + "final output" + ); + worker.join().unwrap(); +} + +#[cfg(unix)] +#[test] +fn hangup_without_readable_readiness_drains_residual_output() { + let mut machine = machine(b"final output".to_vec(), Some(ErrorKind::Other)); + let (registration, readiness) = corcovado::Registration::new2(); + machine.pty.child = Some(registration); + machine.pty.read_event = true; + readiness.set_readiness(UnixReady::hup().into()).unwrap(); + let shutdown = machine.channel(); + let (completed, receiver) = mpsc::channel(); + let worker = machine.spawn(); + std::thread::spawn(move || completed.send(worker.join().unwrap()).unwrap()); + let result = receiver.recv_timeout(Duration::from_secs(5)); + if result.is_err() { + // Unblock the worker even if HUP handling regresses. + shutdown.send(Msg::Shutdown).unwrap(); + } + let (machine, _) = result.expect("HUP must be handled without readable readiness"); + assert_eq!(first_line(&machine, 12), "final output"); + assert_eq!(machine.pty.shutdown_calls, 1); + assert_eq!(machine.pty.deregister_calls, 1); +} + +#[test] +fn explicit_shutdown_finalizes_once() { + let machine = machine(Vec::new(), None); + machine.channel().send(Msg::Shutdown).unwrap(); + let (machine, _) = machine.spawn().join().unwrap(); + assert_eq!(machine.pty.shutdown_calls, 1); + assert_eq!(machine.pty.deregister_calls, 1); +} diff --git a/teletypewriter/src/lib.rs b/teletypewriter/src/lib.rs index 89f75814f4..c67fdcb409 100644 --- a/teletypewriter/src/lib.rs +++ b/teletypewriter/src/lib.rs @@ -54,6 +54,15 @@ pub enum ChildEvent { } pub trait EventedPty: ProcessReadWrite { + /// Optional hook for releasing child-process resources when the I/O loop stops. + /// + /// The default does nothing; success does not guarantee child termination. + /// Unix terminates and reaps its child here. Windows keeps its existing + /// cleanup on drop, when the ConPTY backend closes the pseudoconsole. + fn shutdown(&mut self) -> std::io::Result<()> { + Ok(()) + } + fn child_event_token(&self) -> corcovado::Token; /// Tries to retrieve an event. diff --git a/teletypewriter/src/unix/child.rs b/teletypewriter/src/unix/child.rs new file mode 100644 index 0000000000..aae271a86e --- /dev/null +++ b/teletypewriter/src/unix/child.rs @@ -0,0 +1,571 @@ +use super::TIOCSWINSZ; +use crate::{ChildEvent, Winsize, WinsizeBuilder}; +use std::io; +use std::ops::Deref; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +/// Grace between SIGHUP and SIGKILL. Shell HUP traps and history +/// flushes routinely exceed 100ms; kitty and ghostty never escalate +/// at all, so err on the long side. +const HANGUP_GRACE: Duration = Duration::from_secs(1); +/// Bound on reaping after SIGKILL. A child in uninterruptible sleep +/// survives SIGKILL; an unbounded wait would hold the lifecycle mutex +/// (and any Drop running it) forever. The child stays `Running` on +/// timeout so a later poll or terminate can finish the reap. +const KILL_REAP_TIMEOUT: Duration = Duration::from_secs(1); + +#[derive(Debug)] +pub struct Child { + pub id: Arc, + pub pid: Arc, + #[allow(dead_code)] + ptsname: String, + #[allow(dead_code)] + process: Option, + lifecycle: Arc>, +} + +/// A retired child has no PID that can accidentally be signaled after reuse. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ChildLifecycle { + Running(libc::pid_t), + /// SIGHUP was delivered (a Drop-side hangup or a started terminate), + /// so terminate does not signal it again: HUP traps doing + /// non-idempotent work must run once per close. + HungUp(libc::pid_t), + /// SIGKILL was sent but the reap timed out (uninterruptible sleep); + /// later attempts only re-poll instead of repeating the escalation. + Killed(libc::pid_t), + Exited(Option), +} + +impl ChildLifecycle { + fn wait(&mut self, options: libc::c_int) -> io::Result { + let (Self::Running(pid) | Self::HungUp(pid) | Self::Killed(pid)) = *self else { + return Ok(*self); + }; + loop { + let mut status = 0; + let result = unsafe { libc::waitpid(pid, &mut status, options) }; + if result == pid { + *self = Self::Exited(Some(status)); + return Ok(*self); + } + if result == 0 { + return Ok(*self); + } + let error = io::Error::last_os_error(); + match error.raw_os_error() { + Some(libc::EINTR) => continue, + Some(libc::ECHILD) => { + *self = Self::Exited(None); + return Ok(*self); + } + _ => return Err(error), + } + } + } + + fn reap_within(&mut self, timeout: Duration) -> io::Result { + let deadline = Instant::now() + timeout; + loop { + if let Self::Exited(_) = self.wait(libc::WNOHANG)? { + return Ok(true); + } + if Instant::now() >= deadline { + return Ok(false); + } + std::thread::sleep(Duration::from_millis(5)); + } + } + + fn hangup(&mut self) -> io::Result<()> { + let Self::Running(pid) = self.wait(libc::WNOHANG)? else { + return Ok(()); + }; + signal(pid, libc::SIGHUP)?; + *self = Self::HungUp(pid); + Ok(()) + } + + fn terminate(&mut self) -> io::Result<()> { + let pid = match self.wait(libc::WNOHANG)? { + Self::Exited(_) => return Ok(()), + Self::Killed(pid) => { + // The earlier SIGKILL may not have landed (single-pid + // EPERM is not tolerated, but killpg EPERM is), so + // re-send it, then re-poll only briefly: Killed is + // reached after a full reap timeout, and this arm can + // run from Drop on a UI thread. + signal(pid, libc::SIGKILL)?; + return if self.reap_within(Duration::from_millis(5))? { + Ok(()) + } else { + Err(timed_out()) + }; + } + Self::Running(pid) => { + signal(pid, libc::SIGHUP)?; + *self = Self::HungUp(pid); + pid + } + Self::HungUp(pid) => pid, + }; + if self.reap_within(HANGUP_GRACE)? { + return Ok(()); + } + signal(pid, libc::SIGKILL)?; + *self = Self::Killed(pid); + if self.reap_within(KILL_REAP_TIMEOUT)? { + return Ok(()); + } + Err(timed_out()) + } +} + +fn timed_out() -> io::Error { + io::Error::new(io::ErrorKind::TimedOut, "child not reaped after SIGKILL") +} + +/// Signal the child's process group so descendants sharing it are +/// reached (forkpty makes the child a session leader). Fall back to +/// the pid when the child still shares our own group: pre-setsid, or +/// a directly spawned process, where killpg would signal us too. +fn signal(pid: libc::pid_t, signal: libc::c_int) -> io::Result<()> { + let pgid = unsafe { libc::getpgid(pid) }; + let grouped = pgid > 0 && pgid != unsafe { libc::getpgrp() }; + let result = if grouped { + unsafe { libc::killpg(pgid, signal) } + } else { + unsafe { libc::kill(pid, signal) } + }; + if result == -1 { + let error = io::Error::last_os_error(); + let tolerated = match error.raw_os_error() { + Some(libc::ESRCH) => true, + // BSD killpg reports EPERM when any group member cannot be + // signaled (macOS setuid login(1) wrapper, sudo children), + // even though the signal reached the others (ghostty#2273). + // Failing here would skip escalation and reaping, leaking a + // zombie. Group form only: for a single pid EPERM means + // nothing was signaled, and Linux killpg likewise only errs + // when no member was signaled. + Some(libc::EPERM) => { + grouped + && cfg!(any( + target_os = "macos", + target_os = "freebsd", + target_os = "openbsd", + target_os = "netbsd", + target_os = "dragonfly", + )) + } + _ => false, + }; + if !tolerated { + return Err(error); + } + } + Ok(()) +} + +/// A cloneable handle sharing the child's lifecycle state, for owners +/// that outlive the `Pty` (which moves into the reader thread). Going +/// through the lifecycle keeps the reaped-PID guarantee: a retired +/// child is never signaled. +#[derive(Debug, Clone)] +pub struct ChildTerminator(Arc>); + +impl ChildTerminator { + /// A handle with no child; every operation is a no-op. + pub fn retired() -> Self { + Self(Arc::new(Mutex::new(ChildLifecycle::Exited(None)))) + } + + /// Send SIGHUP without waiting or escalating. For Drop on threads + /// that cannot block (the reader thread's shutdown escalation may + /// never run when the process exits right after). Contention is + /// retried briefly: a poll_exit holder clears in microseconds, and + /// this hangup may be the only signal a quitting process delivers. + /// A hold outlasting the retry means terminate() is escalating and + /// SIGHUP already went out; blocking on it would stall the caller + /// for the whole grace period. + pub fn hangup(&self) -> io::Result<()> { + use std::sync::TryLockError; + let deadline = Instant::now() + Duration::from_millis(10); + loop { + match self.0.try_lock() { + Ok(mut lifecycle) => return lifecycle.hangup(), + Err(TryLockError::Poisoned(poisoned)) => { + return poisoned.into_inner().hangup(); + } + Err(TryLockError::WouldBlock) if Instant::now() < deadline => { + std::thread::yield_now(); + } + Err(TryLockError::WouldBlock) => return Ok(()), + } + } + } +} + +impl Child { + pub(super) fn new( + fd: libc::c_int, + pid: libc::pid_t, + ptsname: String, + process: Option, + ) -> Self { + assert!(pid > 0); + Self { + id: Arc::new(fd), + pid: Arc::new(pid), + ptsname, + process, + lifecycle: Arc::new(Mutex::new(ChildLifecycle::Running(pid))), + } + } + + pub fn terminator(&self) -> ChildTerminator { + ChildTerminator(self.lifecycle.clone()) + } + + pub(super) fn poll_exit(&self) -> io::Result> { + match self + .lifecycle + .lock() + .unwrap_or_else(|e| e.into_inner()) + .wait(libc::WNOHANG)? + { + ChildLifecycle::Running(_) + | ChildLifecycle::HungUp(_) + | ChildLifecycle::Killed(_) => Ok(None), + ChildLifecycle::Exited(status) => Ok(Some(ChildEvent::Exited(status))), + } + } + + /// The tcgetwinsize function fills in the winsize structure pointed to by + /// gws with values that represent the size of the terminal window for which + /// fd provides an open file descriptor. If no error occurs tcgetwinsize() + /// returns zero (0). + /// The tcsetwinsize function sets the terminal window size, for the terminal + /// referenced by fd, to the sizes from the winsize structure pointed to by + /// sws. If no error occurs tcsetwinsize() returns zero (0). + /// The winsize structure, defined in , contains (at least) the + /// following four fields + /// unsigned short ws_row; /* Number of rows, in characters */ + /// unsigned short ws_col; /* Number of columns, in characters */ + /// unsigned short ws_xpixel; /* Width, in pixels */ + /// unsigned short ws_ypixel; /* Height, in pixels */ + /// If the actual window size of the controlling terminal of a process + /// changes, the process is sent a SIGWINCH signal. See signal(7). Note + /// simply changing the sizes using tcsetwinsize() does not necessarily + /// change the actual window size, and if not, will not generate a SIGWINCH. + pub fn set_winsize(&self, winsize_builder: WinsizeBuilder) -> io::Result<()> { + let winsize: Winsize = winsize_builder.build(); + match unsafe { libc::ioctl(**self, TIOCSWINSZ, &winsize as *const _) } { + -1 => Err(io::Error::last_os_error()), + _ => Ok(()), + } + } + + /// Return the child’s exit status if it has already exited. If the child is still running, return Ok(None). + /// If another reaper consumed the status, return an error on every call. + /// https://linux.die.net/man/2/waitpid + pub fn waitpid(&self) -> Result, String> { + match self.poll_exit().map_err(|error| error.to_string())? { + None => Ok(None), + Some(ChildEvent::Exited(Some(status))) => Ok(Some(status)), + Some(ChildEvent::Exited(None)) => { + Err(io::Error::from_raw_os_error(libc::ECHILD).to_string()) + } + } + } + + /// Hang up the child, then force termination after a short grace period, + /// and reap it. Repeated calls preserve the original exit status. + pub fn terminate(&self) -> io::Result<()> { + self.lifecycle + .lock() + .unwrap_or_else(|e| e.into_inner()) + .terminate() + } +} + +impl Deref for Child { + type Target = libc::c_int; + fn deref(&self) -> &libc::c_int { + &self.id + } +} + +impl Drop for Child { + fn drop(&mut self) { + if let Err(error) = self.terminate() { + tracing::warn!(%error, "failed to terminate PTY child"); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::unix::create_pty_with_spawn; + use crate::EventedPty; + use libc::waitpid; + use std::io::Error; + use std::io::Read; + use std::process::{Command, Stdio}; + + fn child(script: &str) -> Child { + let mut process = Command::new("/bin/sh") + .args(["-c", script]) + .stdout(Stdio::piped()) + .spawn() + .unwrap(); + // The script announces that its signal disposition is installed. + let mut ready = [0]; + process + .stdout + .take() + .unwrap() + .read_exact(&mut ready) + .unwrap(); + let pid = process.id() as libc::pid_t; + Child::new(-1, pid, String::new(), Some(process)) + } + + fn assert_reaped(pid: libc::pid_t) { + let mut status = 0; + assert_eq!(unsafe { waitpid(pid, &mut status, libc::WNOHANG) }, -1); + assert_eq!(Error::last_os_error().raw_os_error(), Some(libc::ECHILD)); + } + + #[test] + fn natural_exit_status_survives_repeated_wait_and_termination() { + let child = child("printf r; exit 23"); + let pid = *child.pid; + let deadline = Instant::now() + Duration::from_secs(5); + let status = loop { + if let Some(status) = child.waitpid().unwrap() { + break status; + } + assert!(Instant::now() < deadline); + std::thread::sleep(Duration::from_millis(5)); + }; + assert_eq!(libc::WEXITSTATUS(status), 23); + child.terminate().unwrap(); + assert_eq!(child.waitpid().unwrap(), Some(status)); + drop(child); + assert_reaped(pid); + } + + #[test] + fn ignored_hangup_is_escalated_and_reaped_even_if_public_pid_changes() { + let mut child = child("trap '' HUP; printf r; while :; do :; done"); + let pid = *child.pid; + child.pid = Arc::new(0); + child.terminate().unwrap(); + let status = child.waitpid().unwrap().unwrap(); + assert!(libc::WIFSIGNALED(status)); + assert_eq!(libc::WTERMSIG(status), libc::SIGKILL); + child.terminate().unwrap(); + drop(child); + assert_reaped(pid); + } + + #[test] + fn pty_natural_exit_emits_status_once() { + let mut pty = create_pty_with_spawn( + Some("/bin/sh"), + vec!["-c".into(), "exit 29".into()], + &None, + None, + 80, + 24, + 0, + 0, + ) + .unwrap(); + let pid = *pty.child.pid; + let deadline = Instant::now() + Duration::from_secs(5); + let status = loop { + if let Some(ChildEvent::Exited(Some(status))) = pty.next_child_event() { + break status; + } + assert!( + Instant::now() < deadline, + "PTY exit event was not delivered" + ); + std::thread::sleep(Duration::from_millis(5)); + }; + assert!(libc::WIFEXITED(status)); + assert_eq!(libc::WEXITSTATUS(status), 29); + // A subsequent SIGCHLD must not emit the cached status a second time. + unsafe { + libc::raise(libc::SIGCHLD); + } + assert!(pty.next_child_event().is_none()); + pty.shutdown().unwrap(); + assert_eq!(pty.child.waitpid().unwrap(), Some(status)); + drop(pty); + assert_reaped(pid); + } + + #[test] + fn externally_reaped_pty_emits_unknown_status_once() { + let mut pty = create_pty_with_spawn( + Some("/bin/sh"), + vec!["-c".into(), "exit 0".into()], + &None, + None, + 80, + 24, + 0, + 0, + ) + .unwrap(); + pty.child.process.as_mut().unwrap().wait().unwrap(); + assert!(pty.child.waitpid().is_err()); + assert!(pty.child.waitpid().is_err()); + // Ensure the signal queue has an event even if the original SIGCHLD + // arrived while the external reaper was consuming the status. + unsafe { + libc::raise(libc::SIGCHLD); + } + let deadline = Instant::now() + Duration::from_secs(5); + loop { + if let Some(event) = pty.next_child_event() { + assert_eq!(event, ChildEvent::Exited(None)); + break; + } + assert!( + Instant::now() < deadline, + "PTY exit event was not delivered" + ); + std::thread::sleep(Duration::from_millis(5)); + } + unsafe { + libc::raise(libc::SIGCHLD); + } + assert!(pty.next_child_event().is_none()); + pty.shutdown().unwrap(); + assert!(pty.child.waitpid().is_err()); + } + + #[test] + fn failed_spawn_closes_pty_descriptors() { + const ISOLATED: &str = "RIO_TEST_FAILED_PTY_SPAWN"; + if std::env::var_os(ISOLATED).is_none() { + // Descriptor counts are process-wide, so run outside parallel tests. + let status = Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "unix::child::tests::failed_spawn_closes_pty_descriptors", + ]) + .env(ISOLATED, "1") + .status() + .unwrap(); + assert!(status.success()); + return; + } + let failed_spawn = || { + create_pty_with_spawn( + Some("/definitely-missing-rio-test-shell"), + vec![], + &None, + None, + 80, + 24, + 0, + 0, + ) + }; + // Warm up any process-global signal machinery before taking a baseline. + assert!(failed_spawn().is_err()); + let descriptors = || { + (0..1024) + .filter(|fd| unsafe { libc::fcntl(*fd, libc::F_GETFD) != -1 }) + .collect::>() + }; + let before = descriptors(); + for _ in 0..8 { + assert!(failed_spawn().is_err()); + } + assert_eq!(descriptors(), before); + } + + #[test] + fn reaped_child_never_signals_a_reused_pid() { + let mut original = child("printf r; exit 0"); + { + let mut lifecycle = original.lifecycle.lock().unwrap(); + lifecycle.wait(0).unwrap(); + } + let sentinel = child("trap - HUP; printf r; while :; do :; done"); + // Simulate PID reuse deterministically after reaping the original. + original.pid = sentinel.pid.clone(); + original.terminate().unwrap(); + drop(original); + // A signal delivery may be asynchronous; allow it to become observable. + std::thread::sleep(Duration::from_millis(50)); + assert_eq!(sentinel.waitpid().unwrap(), None); + sentinel.terminate().unwrap(); + } + + #[test] + fn terminator_hangup_after_reap_is_noop() { + let child = child("printf r; exit 0"); + let handle = child.terminator(); + child.lifecycle.lock().unwrap().wait(0).unwrap(); + handle.hangup().unwrap(); + assert!(child.waitpid().is_ok()); + } + + #[test] + fn terminator_hangup_delivers_sighup_without_reaping() { + let child = child("trap 'exit 17' HUP; printf r; while :; do :; done"); + child.terminator().hangup().unwrap(); + let deadline = Instant::now() + Duration::from_secs(5); + let status = loop { + if let Some(status) = child.waitpid().unwrap() { + break status; + } + assert!(Instant::now() < deadline); + std::thread::sleep(Duration::from_millis(5)); + }; + assert!(libc::WIFEXITED(status)); + assert_eq!(libc::WEXITSTATUS(status), 17); + } + + #[test] + fn graceful_hangup_preserves_exit_status() { + let child = child("trap 'exit 17' HUP; printf r; while :; do :; done"); + child.terminate().unwrap(); + let status = child.waitpid().unwrap().unwrap(); + assert!(libc::WIFEXITED(status)); + assert_eq!(libc::WEXITSTATUS(status), 17); + } + + #[test] + fn drop_reaps_running_child() { + let child = child("trap 'exit 17' HUP; printf r; while :; do :; done"); + let pid = *child.pid; + drop(child); + assert_reaped(pid); + } + + #[test] + fn externally_reaped_child_is_retired() { + let mut child = child("printf r; exit 0"); + child.process.as_mut().unwrap().wait().unwrap(); + assert!(child.waitpid().is_err()); + assert_eq!( + *child.lifecycle.lock().unwrap(), + ChildLifecycle::Exited(None) + ); + assert!(child.waitpid().is_err()); + assert_eq!(child.poll_exit().unwrap(), Some(ChildEvent::Exited(None))); + child.terminate().unwrap(); + } +} diff --git a/teletypewriter/src/unix/mod.rs b/teletypewriter/src/unix/mod.rs index 35a8e2d36d..b7de5d4fe7 100644 --- a/teletypewriter/src/unix/mod.rs +++ b/teletypewriter/src/unix/mod.rs @@ -1,8 +1,10 @@ #![cfg(unix)] +mod child; #[cfg(target_os = "macos")] mod macos; mod signals; +pub use child::{Child, ChildTerminator}; extern crate libc; @@ -24,7 +26,6 @@ use std::os::unix::process::CommandExt; use std::path::PathBuf; use std::process::{Command, Stdio}; use std::ptr; -use std::sync::Arc; #[cfg(all(target_os = "linux", not(target_env = "musl")))] const TIOCSWINSZ: libc::c_ulong = 0x5414; @@ -52,12 +53,6 @@ extern "C" { winsize: *const Winsize, ) -> libc::pid_t; - fn waitpid( - pid: libc::pid_t, - status: *mut libc::c_int, - options: libc::c_int, - ) -> libc::pid_t; - fn ptsname(fd: *mut libc::c_int) -> *mut libc::c_char; } @@ -141,6 +136,7 @@ pub struct Pty { token: corcovado::Token, signals_token: corcovado::Token, signals: Signals, + child_event_emitted: bool, } impl Deref for Pty { @@ -562,6 +558,11 @@ pub fn create_pty_with_spawn( ))); } + // Own both descriptors before any fallible setup so every error path + // closes them, including command and signal registration failures. + let file = unsafe { File::from_raw_fd(main) }; + let owned_child = unsafe { OwnedFd::from_raw_fd(child) }; + let user = match ShellUser::from_env() { Ok(data) => data, Err(..) => ShellUser { @@ -649,10 +650,7 @@ pub fn create_pty_with_spawn( } // Setup child stdin/stdout/stderr as child fd of PTY. - // Ownership of fd is transferred to the Stdio structs and will be closed by them at the end of - // this scope. (It is not an issue that the fd is closed three times since File::drop ignores - // error on libc::close.). - let owned_child = unsafe { OwnedFd::from_raw_fd(child) }; + // Each Stdio owns a distinct descriptor and closes it when dropped. builder.stdin(owned_child.try_clone()?); builder.stderr(owned_child.try_clone()?); @@ -704,26 +702,26 @@ pub fn create_pty_with_spawn( } // Prepare signal handling before spawning child. - let signals = - Signals::new([sigconsts::SIGCHLD]).expect("error preparing signal handling"); + let signals = Signals::new([sigconsts::SIGCHLD])?; match builder.spawn() { Ok(child_process) => { + let ptsname: String = tty_ptsname(main).unwrap_or_else(|_| "".to_string()); + let child_unix = Child::new( + main, + child_process.id() as libc::pid_t, + ptsname, + Some(child_process), + ); + unsafe { set_nonblocking(main); } - let ptsname: String = tty_ptsname(main).unwrap_or_else(|_| "".to_string()); - let child_unix = Child { - id: Arc::new(main), - ptsname, - pid: Arc::new(child_process.id().try_into().unwrap()), - process: Some(child_process), - }; - Ok(Pty { child: child_unix, - file: unsafe { File::from_raw_fd(main) }, + child_event_emitted: false, + file, token: corcovado::Token::from(0), signals, signals_token: corcovado::Token::from(0), @@ -792,6 +790,8 @@ pub fn create_pty_with_fork( tracing::info!("fork {:?}", shell_program); + let signals = Signals::new([sigconsts::SIGCHLD])?; + match unsafe { forkpty( &mut main as *mut _, @@ -803,32 +803,27 @@ pub fn create_pty_with_fork( 0 => { set_child_envs(envs); default_shell_command(shell_program, args); - Err(Error::other(format!( - "forkpty has reach unreachable with {shell_program}" - ))) + // Never return into the terminal application in the forked child + // when exec fails, or run the parent's destructors there. + unsafe { libc::_exit(127) } } id if id > 0 => { + let file = unsafe { File::from_raw_fd(main) }; // TODO: Currently we fork the process and don't wait to know if led to failure // Whenever it happens it will just simply shut down the teletyperwriter // In the future add an option to check before release the method let ptsname: String = tty_ptsname(main).unwrap_or_else(|_| "".to_string()); - let child = Child { - id: Arc::new(main), - ptsname, - pid: Arc::new(id), - process: None, - }; + let child = Child::new(main, id, ptsname, None); unsafe { set_nonblocking(main); } - let signals = Signals::new([sigconsts::SIGCHLD]) - .expect("error preparing signal handling"); Ok(Pty { child, + child_event_emitted: false, signals, - file: unsafe { File::from_raw_fd(main) }, + file, token: corcovado::Token(0), signals_token: corcovado::Token(0), }) @@ -865,82 +860,6 @@ unsafe fn set_nonblocking(fd: libc::c_int) { assert_eq!(res, 0); } -#[derive(Debug)] -pub struct Child { - pub id: Arc, - pub pid: Arc, - #[allow(dead_code)] - ptsname: String, - #[allow(dead_code)] - process: Option, -} - -impl Child { - /// The tcgetwinsize function fills in the winsize structure pointed to by - /// gws with values that represent the size of the terminal window for which - /// fd provides an open file descriptor. If no error occurs tcgetwinsize() - /// returns zero (0). - /// The tcsetwinsize function sets the terminal window size, for the terminal - /// referenced by fd, to the sizes from the winsize structure pointed to by - /// sws. If no error occurs tcsetwinsize() returns zero (0). - /// The winsize structure, defined in , contains (at least) the - /// following four fields - /// unsigned short ws_row; /* Number of rows, in characters */ - /// unsigned short ws_col; /* Number of columns, in characters */ - /// unsigned short ws_xpixel; /* Width, in pixels */ - /// unsigned short ws_ypixel; /* Height, in pixels */ - /// If the actual window size of the controlling terminal of a process - /// changes, the process is sent a SIGWINCH signal. See signal(7). Note - /// simply changing the sizes using tcsetwinsize() does not necessarily - /// change the actual window size, and if not, will not generate a SIGWINCH. - pub fn set_winsize(&self, winsize_builder: WinsizeBuilder) -> io::Result<()> { - let winsize: Winsize = winsize_builder.build(); - match unsafe { libc::ioctl(**self, TIOCSWINSZ, &winsize as *const _) } { - -1 => Err(io::Error::last_os_error()), - _ => Ok(()), - } - } - - /// Return the child’s exit status if it has already exited. If the child is still running, return Ok(None). - /// https://linux.die.net/man/2/waitpid - pub fn waitpid(&self) -> Result, String> { - let mut status = 0 as libc::c_int; - // If WNOHANG was specified in options and there were no children in a waitable state, then waitid() returns 0 immediately and the state of the siginfo_t structure pointed to by infop is unspecified. To distinguish this case from that where a child was in a waitable state, zero out the si_pid field before the call and check for a nonzero value in this field after the call returns. - let res = - unsafe { waitpid(*self.pid, &mut status as *mut libc::c_int, libc::WNOHANG) }; - if res <= -1 { - return Err(String::from("error")); - } - - if res == 0 && status == 0 { - return Ok(None); - } - - Ok(Some(status)) - } -} - -pub fn kill_pid(pid: i32) { - unsafe { - libc::kill(pid, libc::SIGHUP); - } -} - -impl Deref for Child { - type Target = libc::c_int; - fn deref(&self) -> &libc::c_int { - &self.id - } -} - -impl Drop for Child { - fn drop(&mut self) { - unsafe { - libc::kill(*self.pid, libc::SIGHUP); - } - } -} - pub fn command_per_pid(pid: libc::pid_t) -> String { let current_process_name = Command::new("ps") .arg("-p") @@ -957,21 +876,23 @@ pub fn command_per_pid(pid: libc::pid_t) -> String { } impl EventedPty for Pty { + fn shutdown(&mut self) -> io::Result<()> { + self.child.terminate() + } + #[inline] fn next_child_event(&mut self) -> Option { + if self.child_event_emitted { + return None; + } self.signals.pending().next().and_then(|signal| { if signal != sigconsts::SIGCHLD { return None; } - match self.child.waitpid() { - Err(_e) => { - // std::process::exit(1); - None - } - Ok(None) => None, - Ok(Some(status)) => Some(ChildEvent::Exited(Some(status))), - } + let event = self.child.poll_exit().ok().flatten()?; + self.child_event_emitted = true; + Some(event) }) }