diff --git a/codex-rs/utils/pty/src/child.rs b/codex-rs/utils/pty/src/child.rs index a703f04a62..d9a8365e9b 100644 --- a/codex-rs/utils/pty/src/child.rs +++ b/codex-rs/utils/pty/src/child.rs @@ -1,6 +1,6 @@ //! Local child ownership and platform selection, independent of process transports. //! -//! Every child exposes Tokio stdio handles. Native children retain their PID until +//! Every child exposes Tokio stdio handles. Reap-only children retain ownership until //! reaped, including when a wait is cancelled or the async runtime shuts down. use std::io; @@ -15,6 +15,10 @@ use tokio::process::ChildStdout; #[path = "macos_child.rs"] pub(super) mod macos; +#[cfg(unix)] +#[path = "child_reaper.rs"] +pub(super) mod reaper; + /// A local subprocess with owned stdio and cancellation-safe exit handling. pub struct Child { pub(super) inner: ChildKind, @@ -25,6 +29,11 @@ pub struct Child { pub(super) enum ChildKind { Tokio(tokio::process::Child), + #[cfg(unix)] + TokioReapOnly { + child: Option, + reaper: std::sync::mpsc::Sender, + }, #[cfg(target_os = "macos")] Native(macos::NativeChild), } @@ -33,6 +42,10 @@ impl Child { pub fn id(&self) -> Option { match &self.inner { ChildKind::Tokio(child) => child.id(), + #[cfg(unix)] + ChildKind::TokioReapOnly { child, .. } => { + child.as_ref().and_then(tokio::process::Child::id) + } #[cfg(target_os = "macos")] ChildKind::Native(child) => child.id(), } @@ -43,12 +56,20 @@ impl Child { self.stdin.take(); match &mut self.inner { ChildKind::Tokio(child) => child.wait().await, + #[cfg(unix)] + ChildKind::TokioReapOnly { child, .. } => { + child + .as_mut() + .ok_or_else(|| io::Error::other("child already transferred to reaper"))? + .wait() + .await + } #[cfg(target_os = "macos")] ChildKind::Native(child) => child.wait().await, } } - /// Drain both output pipes while waiting, retaining kill-on-drop on cancellation. + /// Drain both output pipes while waiting, retaining the configured drop policy on cancellation. pub async fn wait_with_output(mut self) -> io::Result { let mut stdout = self.stdout.take(); let mut stderr = self.stderr.take(); @@ -85,6 +106,14 @@ impl Child { self.stdin.take(); match &mut self.inner { ChildKind::Tokio(child) => child.kill().await, + #[cfg(unix)] + ChildKind::TokioReapOnly { child, .. } => { + child + .as_mut() + .ok_or_else(|| io::Error::other("child already transferred to reaper"))? + .kill() + .await + } #[cfg(target_os = "macos")] ChildKind::Native(child) => child.kill().await, } @@ -94,3 +123,17 @@ impl Child { #[cfg(all(test, unix))] #[path = "child_tests.rs"] mod tests; + +impl Drop for ChildKind { + fn drop(&mut self) { + #[cfg(unix)] + if let Self::TokioReapOnly { child, reaper } = self + && let Some(child) = child.take() + && child.id().is_some() + { + // Transfer the handle, not only its PID: Tokio must not enqueue it + // into a stopped runtime's orphan queue before our worker reaps it. + let _ = reaper.send(reaper::ChildToReap::Tokio(child)); + } + } +} diff --git a/codex-rs/utils/pty/src/child_command.rs b/codex-rs/utils/pty/src/child_command.rs index 89c890ddfa..01a010e1af 100644 --- a/codex-rs/utils/pty/src/child_command.rs +++ b/codex-rs/utils/pty/src/child_command.rs @@ -2,7 +2,7 @@ //! //! The wrapped Tokio command is private: callers cannot install callbacks or //! change settings that the native backend cannot inspect. Children receive only -//! explicitly supplied environment variables and kill-on-drop. Stdio, descriptor +//! explicitly supplied environment variables and default to kill-on-drop. Stdio, descriptor //! inheritance, and compatibility fallbacks are configured independently. use std::ffi::OsStr; @@ -36,6 +36,19 @@ pub enum SpawnFallback { ReturnError, } +/// Cleanup performed when a child handle is dropped before its exit is collected. +#[derive(Clone, Copy)] +pub(crate) enum ChildDropPolicy { + /// Kill the direct child, then reap it. This is the default. + KillAndReap, + /// Reap the child when it exits, leaving termination to the caller. + #[allow( + dead_code, + reason = "Used by the pipe adapter in the next stacked change." + )] + ReapOnly, +} + /// An explicit child stdin, including a socket used for bidirectional fd transfer. pub enum ChildStdin { Piped, @@ -50,6 +63,7 @@ pub struct Command { pub(crate) descriptor_policy: DescriptorPolicy, pub(crate) fallback: SpawnFallback, pub(crate) stdin: ChildStdin, + pub(crate) drop_policy: ChildDropPolicy, #[cfg(unix)] pub(crate) arg0: Option, } @@ -69,6 +83,7 @@ impl Command { descriptor_policy: DescriptorPolicy::Inherit, fallback: SpawnFallback::Compatible, stdin: ChildStdin::Piped, + drop_policy: ChildDropPolicy::KillAndReap, #[cfg(unix)] arg0: None, } @@ -108,6 +123,20 @@ impl Command { self } + /// Choose who owns termination when the child handle is dropped. + #[allow( + dead_code, + reason = "Used by the pipe adapter in the next stacked change." + )] + pub(crate) fn drop_policy(&mut self, policy: ChildDropPolicy) -> &mut Self { + self.drop_policy = policy; + self.inner.kill_on_drop(match policy { + ChildDropPolicy::KillAndReap => true, + ChildDropPolicy::ReapOnly => false, + }); + self + } + pub fn stdin(&mut self, stdin: ChildStdin) -> &mut Self { self.stdin = stdin; self @@ -170,12 +199,30 @@ impl Command { }); } } + #[cfg(unix)] + let reaper = match self.drop_policy { + ChildDropPolicy::KillAndReap => None, + ChildDropPolicy::ReapOnly => Some(crate::child::reaper::sender()?), + }; let mut child = self.inner.spawn()?; Ok(Child { stdin: child.stdin.take(), stdout: child.stdout.take(), stderr: child.stderr.take(), - inner: ChildKind::Tokio(child), + inner: { + #[cfg(unix)] + { + match reaper { + Some(reaper) => ChildKind::TokioReapOnly { + child: Some(child), + reaper, + }, + None => ChildKind::Tokio(child), + } + } + #[cfg(not(unix))] + ChildKind::Tokio(child) + }, }) } } diff --git a/codex-rs/utils/pty/src/child_reaper.rs b/codex-rs/utils/pty/src/child_reaper.rs new file mode 100644 index 0000000000..3fb871769a --- /dev/null +++ b/codex-rs/utils/pty/src/child_reaper.rs @@ -0,0 +1,68 @@ +//! Runtime-independent cleanup for children whose termination is externally owned. +//! +//! Initialize the shared worker before launch so dropping a live child never +//! creates a thread or waits for its exit. Poll only transferred PIDs, retaining +//! ownership until reaped, so one live child cannot delay cleanup of another. + +use std::io; +#[cfg(target_os = "macos")] +use std::ptr; +use std::sync::Mutex; +use std::sync::PoisonError; +use std::sync::mpsc; +use std::time::Duration; + +/// Exclusive ownership transferred by a child handle after its caller drops it. +pub(crate) enum ChildToReap { + #[cfg(target_os = "macos")] + Native(libc::pid_t), + Tokio(tokio::process::Child), +} + +static REAPER: Mutex>> = Mutex::new(None); +const REAP_BATCH_SIZE: usize = 64; + +/// Reserve the shared reaper, returning thread creation failures before launch. +pub(crate) fn sender() -> io::Result> { + let mut reaper = REAPER.lock().unwrap_or_else(PoisonError::into_inner); + if let Some(sender) = &*reaper { + return Ok(sender.clone()); + } + let (sender, receiver) = mpsc::channel(); + std::thread::Builder::new() + .name("codex-child-reaper".into()) + .spawn(move || { + let mut pending = Vec::new(); + loop { + if pending.is_empty() { + let Ok(pid) = receiver.recv() else { return }; + pending.push(pid); + } else { + match receiver.recv_timeout(Duration::from_millis(10)) { + Ok(pid) => pending.push(pid), + Err(mpsc::RecvTimeoutError::Timeout) => {} + Err(mpsc::RecvTimeoutError::Disconnected) => return, + } + } + // Amortize scans across bursts without starving reaping when drops continue. + pending.extend(receiver.try_iter().take(REAP_BATCH_SIZE - 1)); + pending.retain_mut(|child| match child { + #[cfg(target_os = "macos")] + ChildToReap::Native(pid) => { + // SAFETY: Drop transferred exclusive ownership of this PID. + let result = unsafe { libc::waitpid(*pid, ptr::null_mut(), libc::WNOHANG) }; + result == 0 + || (result == -1 + && io::Error::last_os_error().kind() == io::ErrorKind::Interrupted) + } + ChildToReap::Tokio(child) => match child.try_wait() { + Ok(None) => true, + Ok(Some(_)) => false, + Err(error) => error.kind() == io::ErrorKind::Interrupted, + }, + }); + } + })?; + *reaper = Some(sender.clone()); + Ok(sender) +} diff --git a/codex-rs/utils/pty/src/child_tests.rs b/codex-rs/utils/pty/src/child_tests.rs index f6364733d0..0fd28e8360 100644 --- a/codex-rs/utils/pty/src/child_tests.rs +++ b/codex-rs/utils/pty/src/child_tests.rs @@ -1,4 +1,4 @@ -//! Regression coverage for output-pipe lifetimes while waiting for a child. +//! Regression coverage for output-pipe lifetimes and runtime-independent reaping. use std::future::Future; use std::future::poll_fn; @@ -66,3 +66,98 @@ async fn wait_with_output_keeps_eof_pipes_open_until_exit() -> anyhow::Result<() ); Ok(()) } + +#[test] +fn non_killing_drop_reaps_after_runtime_shutdown() -> anyhow::Result<()> { + use crate::child_command::ChildDropPolicy; + use std::time::Duration; + if std::env::var_os("CODEX_TEST_ISOLATED_REAPER").is_none() { + // Other Tokio tests must not drain this process's global orphan queue. + let output = std::process::Command::new(std::env::current_exe()?) + .args([ + "--exact", + "child::tests::non_killing_drop_reaps_after_runtime_shutdown", + "--nocapture", + ]) + .env("CODEX_TEST_ISOLATED_REAPER", "1") + .output()?; + assert!( + output.status.success(), + "isolated reaper test failed:\n{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + ); + return Ok(()); + } + for program in ["cat", "/bin/cat"] { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()?; + let mut child = runtime.block_on(async { + let mut command = crate::Command::new(program); + command.drop_policy(ChildDropPolicy::ReapOnly); + command.spawn() + })?; + let stdin = child.stdin.take(); + let pid = child.id().expect("live PID"); + let next = runtime.block_on(async { + let mut command = crate::Command::new(program); + command.drop_policy(ChildDropPolicy::ReapOnly); + command.spawn() + })?; + let next_pid = next.id().expect("live PID"); + drop(runtime); + drop(child); + drop(next); + // The shared reaper must collect the second child while the first remains alive. + wait_until_reaped(next_pid)?; + // Keep stdin open: cat must remain alive even after its owner and runtime go away. + std::thread::sleep(Duration::from_millis(50)); + let mut info = std::mem::MaybeUninit::::zeroed(); + // SAFETY: waitid only observes our child without reaping it, with writable storage. + assert_eq!( + unsafe { + libc::waitid( + libc::P_PID, + pid, + info.as_mut_ptr(), + libc::WEXITED | libc::WNOHANG | libc::WNOWAIT, + ) + }, + 0 + ); + // SAFETY: waitid succeeded and initialized the zeroed signal information. + assert_eq!(unsafe { info.assume_init().si_pid() }, 0); + drop(stdin); + wait_until_reaped(pid)?; + } + Ok(()) +} + +fn wait_until_reaped(pid: u32) -> anyhow::Result<()> { + use std::io; + use std::time::Duration; + let mut info = std::mem::MaybeUninit::::zeroed(); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + loop { + // SAFETY: WNOWAIT prevents the test from stealing the reaper's child. + if unsafe { + libc::waitid( + libc::P_PID, + pid, + info.as_mut_ptr(), + libc::WEXITED | libc::WNOHANG | libc::WNOWAIT, + ) + } == -1 + { + assert_eq!( + io::Error::last_os_error().raw_os_error(), + Some(libc::ECHILD) + ); + break; + } + anyhow::ensure!(std::time::Instant::now() < deadline, "child was not reaped"); + std::thread::sleep(Duration::from_millis(10)); + } + Ok(()) +} diff --git a/codex-rs/utils/pty/src/macos_child.rs b/codex-rs/utils/pty/src/macos_child.rs index f36a8ca3dc..5a0158b848 100644 --- a/codex-rs/utils/pty/src/macos_child.rs +++ b/codex-rs/utils/pty/src/macos_child.rs @@ -23,6 +23,10 @@ use tokio::signal::unix::Signal; use tokio::signal::unix::SignalKind; use tokio::signal::unix::signal; +use crate::child_command::ChildDropPolicy; + +use crate::child::reaper; + // libc does not expose this Apple extension. It is available since macOS 10.15, // before Codex's minimum supported macOS version (12). unsafe extern "C" { @@ -33,11 +37,12 @@ unsafe extern "C" { } /// Owns a child PID until reaping, so cancellation cannot lose or reuse it. -/// Dropping a live child kills it and reaps it independently of the Tokio runtime. +/// Drop obeys the configured kill policy and reaps independently of Tokio. pub(crate) struct NativeChild { pid: Option, status: Option, sigchld: Signal, + reaper: Option>, } impl NativeChild { @@ -74,6 +79,12 @@ impl NativeChild { .get_current_dir() .map(|cwd| c_string(cwd.as_os_str())) .transpose()?; + // Reserve the non-killing cleanup worker before creating a child. Drop + // must not need a new thread or synchronously wait for a live process. + let reaper = match request.drop_policy { + ChildDropPolicy::KillAndReap => None, + ChildDropPolicy::ReapOnly => Some(reaper::sender()?), + }; // Subscribe before spawning so a child that exits immediately cannot be missed. let sigchld = signal(SignalKind::child())?; @@ -203,6 +214,7 @@ impl NativeChild { pid: Some(pid), status: None, sigchld, + reaper, }; Ok(Some(crate::Child { inner: super::ChildKind::Native(child), @@ -278,14 +290,19 @@ impl Drop for NativeChild { fn drop(&mut self) { let _ = self.try_wait(); let Some(pid) = self.pid.take() else { return }; + if let Some(reaper) = &self.reaper { + // The shared sender keeps the worker alive for the process lifetime. + let _ = reaper.send(reaper::ChildToReap::Native(pid)); + return; + } // SAFETY: This child has not been reaped, so its PID cannot be reused. unsafe { libc::kill(pid, libc::SIGKILL); } // Drop may run during runtime shutdown. Reap independently of Tokio, - // without blocking its worker threads while the killed process exits. + // without blocking its worker threads while this process exits. if std::thread::Builder::new() - .name("mcp-child-reaper".into()) + .name("codex-child-reaper".into()) .spawn(move || reap(pid)) .is_err() {