Add a reap-only drop policy for child processes (#47603)

## What changed

Add a crate-private `ChildDropPolicy` to choose whether dropping a child kills it or only reaps it after exit. Keep `KillAndReap` as the default.

For `ReapOnly` on Unix, transfer dropped children to a shared reaper thread that operates independently of the Tokio runtime. Support both Tokio children and native macOS children, without blocking drop or letting a live child delay reaping others.

## Testing

Add an isolated regression test verifying that reap-only children survive handle drop after runtime shutdown, are reaped after exit, and do not prevent other exited children from being reaped.

GitOrigin-RevId: 5ec639597dba7f3b61694d6d465991e054ae3397
This commit is contained in:
Charlie Marsh
2026-09-23 17:59:23 +00:00
committed by copyberry
parent 2210190435
commit 897b7ae285
5 changed files with 278 additions and 8 deletions
+45 -2
View File
@@ -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<tokio::process::Child>,
reaper: std::sync::mpsc::Sender<reaper::ChildToReap>,
},
#[cfg(target_os = "macos")]
Native(macos::NativeChild),
}
@@ -33,6 +42,10 @@ impl Child {
pub fn id(&self) -> Option<u32> {
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<std::process::Output> {
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));
}
}
}
+49 -2
View File
@@ -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<OsString>,
}
@@ -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)
},
})
}
}
+68
View File
@@ -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<Option<mpsc::Sender<ChildToReap>>> = 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<mpsc::Sender<ChildToReap>> {
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)
}
+96 -1
View File
@@ -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::<libc::siginfo_t>::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::<libc::siginfo_t>::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(())
}
+20 -3
View File
@@ -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<libc::pid_t>,
status: Option<ExitStatus>,
sigchld: Signal,
reaper: Option<std::sync::mpsc::Sender<reaper::ChildToReap>>,
}
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()
{