diff --git a/DESIGN.md b/DESIGN.md index 3fc2182..9105cb7 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -352,6 +352,10 @@ impl Loop { // processes and signals pub fn spawn(&mut self, spec: &ProcessSpec, tok: Token) -> io::Result; // handles for stdio pipes + // Node's stdio tail: extra child descriptors at fixed numbers (fork()'s IPC + // channel at 3), their parent ends written into the caller's slice. + pub fn spawn_extra(&mut self, spec: &ProcessSpec, tok: Token, + parents: &mut [Option]) -> io::Result; pub fn kill(&mut self, p: Handle, sig: Signal) -> io::Result<()>; pub fn signal_start(&mut self, sig: Signal, tok: Token) -> io::Result; @@ -373,6 +377,9 @@ pub enum OpResult { Cancelled, Closed, Stopped, Err(Error), } pub struct Error { pub kind: ErrorKind, pub os: Option } // host maps to ECONNRESET etc. + +pub struct ChildFd { pub number: u32, pub source: ChildFdSource } // number in 3..=MAX_CHILD_FD +pub enum ChildFdSource { Null, Pipe, Duplex, Handle(Handle) } // Duplex is Node's 'pipe'/'ipc' ``` ## 7. Platform design @@ -398,7 +405,7 @@ pub struct Error { pub kind: ErrorKind, pub os: Option } // host maps to - **Named pipes:** overlapped `ConnectNamedPipe`/`ReadFile`/`WriteFile`. This is the `pipe_listen`/`pipe_connect` transport (Node IPC uses named pipes on Windows). - **Stdio that isn't overlapped** (a handle inherited as a synchronous pipe or file): a dedicated reader thread per handle that posts completions. It isn't possible to reopen such a handle overlapped. libuv does the same. - **Console/TTY:** `ReadConsoleInputW` on a reader thread, with VT input and output modes (`ENABLE_VIRTUAL_TERMINAL_PROCESSING`/`_INPUT`). Resize events become a `Signal::WinCh` completion; Perry has no resize support on Windows today (`tty.rs:9–12`). -- **Processes:** `CreateProcessW` with overlapped pipe handles, a Job Object for kill-tree semantics, and `RegisterWaitForSingleObject` on the process handle to post the exit completion. +- **Processes:** `CreateProcessW` with overlapped pipe handles, a Job Object for kill-tree semantics, and `RegisterWaitForSingleObject` on the process handle to post the exit completion. Descriptors beyond stdin/stdout/stderr are published through the C run-time inherited-descriptor block in `STARTUPINFOW.lpReserved2` (count, per-descriptor flags from `GetFileType`, then the handles), plus `PROC_THREAD_ATTRIBUTE_HANDLE_LIST`. That is libuv's and Node's own convention, so a child descriptor number — `NODE_CHANNEL_FD=3` — means the same thing here as it does under Node. A child that does not use the C run-time inherits the handles but has no number for them. - **Signals:** `SetConsoleCtrlHandler` for CTRL_C, CTRL_BREAK and CTRL_CLOSE, mapped to `SIGINT`/`SIGBREAK`/`SIGHUP`. SIGTERM has no console equivalent; documented as such, matching Perry today (`os/signal.rs:441–531`). - **Wake:** `PostQueuedCompletionStatus` with a reserved completion key. - **Timer precision:** the default system tick is about 15.6 ms and `GetQueuedCompletionStatusEx` timeouts round to it. Use a **high-resolution waitable timer** (`CREATE_WAITABLE_TIMER_HIGH_RESOLUTION`, Windows 10 1803+) armed to the next deadline and associated with the port through dynamically resolved `NtAssociateWaitCompletionPacket`. `GetQueuedCompletionStatusEx` is nonalertable. Cancellation returning `STATUS_PENDING` requires dequeuing the generation-tagged timer packet before reuse. M1 prototyped both APC and NT packet routes; §15 question 3 records the NT packet decision and measured lateness. `timeBeginPeriod` is not an acceptable default because it changes the tick system-wide. @@ -455,6 +462,8 @@ Two backends, because both versions matter now: | Stdio pipes | readiness | readiness | overlapped, or reader thread | `wasi:cli` streams | `wasi:cli` streams | unsupported | | TTY | termios + readiness | termios + readiness | console API reader thread, VT modes | size only | size only | unsupported | | Child processes | pidfd / SIGCHLD | EVFILT_PROC | RegisterWaitForSingleObject + Job Object | unsupported | unsupported | unsupported | +| Child descriptors beyond stdio | one `pre_exec` `dup2` per number, sources relocated above every target first | same | C run-time inherited-descriptor block + handle-list attribute | unsupported | unsupported | unsupported | +| Child session control | `setsid` (`detached`), `setpgid` (`new_process_group`), `TIOCSCTTY` (`controlling_terminal`) | same | detached process group + Job Object; no controlling terminal | unsupported | unsupported | unsupported | | Signals | sigaction + self-pipe | EVFILT_SIGNAL | SetConsoleCtrlHandler | unsupported | unsupported | unsupported | | Files | blocking pool | blocking pool | blocking pool | `wasi:filesystem` preopens, run in `turn` | `wasi:filesystem` async preopens, run in `turn` | unsupported (no OPFS host mapping defined) | | File watch | inotify (recursion by the host) | FSEvents for directories (recursive), kqueue for files | ReadDirectoryChangesW (recursive) | unsupported | unsupported | unsupported | @@ -511,6 +520,10 @@ and work on any live socket handle, including one produced by `accept`. | Microtasks, nextTick | Perry | unchanged | | Error text and codes | Perry | Map `Error { kind, os }` to Node's `code`/`errno`/`syscall` | | Keep-alive for JS-level resources | Perry | Uses turnloop `alive()` plus its own JS timers until those move (P3) | +| Child descriptor policy | Perry | Which number is the IPC channel, and the name and value of `NODE_CHANNEL_FD`; turnloop places the descriptor, never names it | +| Program resolution and shell | Perry | `shell: true`, `execPath`, PATH policy and argv construction | +| Pty allocation | Perry | `openpty` and its termios; turnloop owns only the child-side `setsid`/`TIOCSCTTY` through `ProcessSpec::controlling_terminal` | +| Reaping a child turnloop did not spawn | Perry | turnloop reaps only its own children, always by their own identity (`waitpid(pid, …)`, never `-1`), so a host waiter in the same process keeps its own children's status. There is no `adopt_process`; see `docs/lanes/procspec.md` | | Thread-pool jobs touching JS | never | Pool jobs are `Send` Rust closures; results convert on the main thread | **Wiring for P0:** diff --git a/crates/turnloop-contract/src/bin/native_child.rs b/crates/turnloop-contract/src/bin/native_child.rs index 595de34..28b16ac 100644 --- a/crates/turnloop-contract/src/bin/native_child.rs +++ b/crates/turnloop-contract/src/bin/native_child.rs @@ -21,6 +21,12 @@ fn main() { println!("60 service timer expiries; no-spin bounds passed"); } "exit" => std::process::exit(23), + "exit-with" => std::process::exit( + args[2] + .to_str() + .and_then(|code| code.parse().ok()) + .expect("exit code"), + ), "sleep" => std::thread::sleep(Duration::from_secs(60)), "roundtrip" => { let mut stdout = std::io::stdout().lock(); @@ -89,6 +95,61 @@ fn main() { "handle" => native_handle(args.get(2).expect("pipe path")), #[cfg(any(target_vendor = "apple", target_os = "freebsd"))] "blocked-signal" => blocked_signal(), + // Both extra descriptors carry bytes in both directions. The channel + // number is read out of the environment exactly as Node reads + // NODE_CHANNEL_FD, so this proves the handoff, not a hard-coded 3. + #[cfg(any(unix, windows))] + "channel" => { + let channel = numbered_fd("NODE_CHANNEL_FD"); + let extra = numbered_fd("TURNLOOP_EXTRA_FD"); + let mut request = [0u8; 6]; + read_exact(channel, &mut request); + assert_eq!(&request, b"ping-3", "channel request"); + write_all(channel, b"pong-3"); + read_exact(extra, &mut request); + assert_eq!(&request, b"ping-4", "extra request"); + write_all(extra, b"pong-4"); + print!("ok"); + std::io::stdout().flush().expect("flush transcript"); + } + // One-way pipe, null device and an adopted parent transport. + #[cfg(any(unix, windows))] + "extra-sources" => { + let (one_way, null, adopted) = (fd_stream(3), fd_stream(4), fd_stream(5)); + write_all(one_way, b"three"); + write_all(null, b"void"); + let mut discard = [0u8; 4]; + assert_eq!(read_once(null, &mut discard), 0, "null device reads EOF"); + write_all(adopted, b"five"); + print!("ok"); + std::io::stdout().flush().expect("flush transcript"); + } + #[cfg(unix)] + "tty-session" => { + // SAFETY: queries this process only; no pointer arguments or mutation. + let (pid, group, session) = + unsafe { (libc::getpid(), libc::getpgrp(), libc::getsid(0)) }; + // SAFETY: constant device path with integer flags; -1 means no + // controlling terminal, which is the distinction under test. + let terminal = unsafe { libc::open(c"/dev/tty".as_ptr(), libc::O_RDWR) }; + let foreground = if terminal < 0 { + -1 + } else { + // SAFETY: live descriptor on this process's controlling terminal. + unsafe { libc::tcgetpgrp(terminal) } + }; + if terminal >= 0 { + // SAFETY: closing the descriptor this branch just opened. + unsafe { + libc::close(terminal); + } + } + println!("{pid}:{group}:{session}:{}", i32::from(terminal >= 0)); + assert_eq!( + foreground, group, + "child leads its terminal foreground group" + ); + } "copy" => { let mut bytes = Vec::new(); std::io::stdin() @@ -317,3 +378,115 @@ fn native_handle(path: &std::ffi::OsStr) { fn native_handle(_: &std::ffi::OsStr) { panic!("instantiate handle fixture on production native backend"); } + +/// A child descriptor as this platform's stream identity: the number itself on +/// Unix, and the C run-time's handle for it on Windows, which is precisely how +/// libuv's `uv_pipe_open` turns Node's `NODE_CHANNEL_FD` into a usable stream. +#[cfg(unix)] +type FdStream = std::os::fd::RawFd; +#[cfg(windows)] +type FdStream = windows_sys::Win32::Foundation::HANDLE; + +#[cfg(windows)] +unsafe extern "C" { + fn _get_osfhandle(fd: i32) -> isize; +} + +#[cfg(any(unix, windows))] +fn fd_stream(fd: i32) -> FdStream { + #[cfg(unix)] + { + fd + } + #[cfg(windows)] + { + // SAFETY: the C run-time owns its descriptor table; this only reads it. + let handle = unsafe { _get_osfhandle(fd) }; + assert!(handle > 0, "descriptor {fd} is absent from the C run-time"); + handle as FdStream + } +} + +#[cfg(any(unix, windows))] +fn numbered_fd(variable: &str) -> FdStream { + let value = std::env::var(variable).unwrap_or_else(|_| panic!("{variable} is unset")); + fd_stream(value.parse().expect("descriptor number")) +} + +#[cfg(any(unix, windows))] +fn read_once(stream: FdStream, buffer: &mut [u8]) -> usize { + #[cfg(unix)] + { + // SAFETY: live inherited descriptor and a writable buffer of that length. + let n = unsafe { libc::read(stream, buffer.as_mut_ptr().cast(), buffer.len()) }; + assert!(n >= 0, "read: {}", std::io::Error::last_os_error()); + n as usize + } + #[cfg(windows)] + { + use windows_sys::Win32::{Foundation::ERROR_BROKEN_PIPE, Storage::FileSystem::ReadFile}; + let mut read = 0; + // SAFETY: live inherited handle, writable buffer of the stated length, + // and a synchronous handle so no OVERLAPPED is required. + let ok = unsafe { + ReadFile( + stream, + buffer.as_mut_ptr(), + buffer.len() as u32, + &mut read, + std::ptr::null_mut(), + ) + }; + if ok == 0 { + let error = std::io::Error::last_os_error(); + assert_eq!( + error.raw_os_error(), + Some(ERROR_BROKEN_PIPE as i32), + "read: {error}" + ); + return 0; + } + read as usize + } +} + +#[cfg(any(unix, windows))] +fn read_exact(stream: FdStream, buffer: &mut [u8]) { + let mut filled = 0; + while filled < buffer.len() { + let n = read_once(stream, &mut buffer[filled..]); + assert_ne!(n, 0, "premature end of stream after {filled} bytes"); + filled += n; + } +} + +#[cfg(any(unix, windows))] +fn write_all(stream: FdStream, mut bytes: &[u8]) { + while !bytes.is_empty() { + #[cfg(unix)] + // SAFETY: live inherited descriptor and a readable buffer of that length. + let n = unsafe { libc::write(stream, bytes.as_ptr().cast(), bytes.len()) }; + #[cfg(unix)] + assert!(n > 0, "write: {}", std::io::Error::last_os_error()); + #[cfg(unix)] + let n = n as usize; + #[cfg(windows)] + let n = { + let mut wrote = 0; + // SAFETY: live inherited handle, readable buffer of the stated + // length, and a synchronous handle so no OVERLAPPED is required. + let ok = unsafe { + windows_sys::Win32::Storage::FileSystem::WriteFile( + stream, + bytes.as_ptr(), + bytes.len() as u32, + &mut wrote, + std::ptr::null_mut(), + ) + }; + assert_ne!(ok, 0, "write: {}", std::io::Error::last_os_error()); + wrote as usize + }; + bytes = &bytes[n..]; + } +} diff --git a/crates/turnloop-contract/tests/allocations.rs b/crates/turnloop-contract/tests/allocations.rs index 56d726d..0220edf 100644 --- a/crates/turnloop-contract/tests/allocations.rs +++ b/crates/turnloop-contract/tests/allocations.rs @@ -1175,6 +1175,157 @@ fn executor_steady_io_poll_and_sleep_allocate_nothing() { assert_eq!(count, 1001); } +/// Steady-state traffic on a child's extra descriptors, on both platforms. +/// +/// Spawning is resource creation and is charged to setup, like any other open. +/// What must be free is everything after it: the writes, the reads, the exit +/// and the closes of a child holding two descriptors beyond its standard three. +#[cfg(any(unix, windows))] +#[test] +fn extra_child_descriptor_traffic_allocates_nothing_after_spawn() { + const CHILDREN: usize = 8; + /// `ping-3` then `ping-4`, so one static region serves both writes. + static PINGS: [u8; 12] = *b"ping-3ping-4"; + fn ping(extra: usize) -> WriteBuf { + // SAFETY: a static region outlives every completion and the loop itself. + WriteBuf::Provided(unsafe { IoBuf::from_raw_parts(PINGS[extra * 6..].as_ptr(), 6) }) + } + /// Turn until the awaited handle has produced `want` bytes, folding every + /// other completion into the counters. Nothing here allocates. + fn collect(l: &mut Loop, out: &mut Completions, h: Handle, want: &[u8], seen: &mut [usize; 4]) { + let mut got = [0u8; 6]; + let mut filled = 0; + while filled < want.len() { + l.read(h, ReadBuf::Pooled, Token(70)).expect("read"); + let before = filled; + while filled == before { + let until = l.now() + Duration::from_secs(10); + assert!(l.now() < until); + l.turn(Timeout::Until(until), out).expect("turn"); + for c in out.drain() { + match c.result { + OpResult::Read { + n, + lease: Some(bytes), + } => { + assert_eq!(c.handle, Some(h), "only one read is in flight"); + assert!(n > 0); + got[filled..filled + n].copy_from_slice(bytes.as_slice()); + filled += n; + } + OpResult::Wrote(n) => { + assert_eq!(n, 6); + seen[0] += 1; + } + OpResult::Exited(status) => { + assert_eq!(status.code, Some(0)); + seen[1] += 1; + } + OpResult::Eof => seen[2] += 1, + OpResult::Closed => seen[3] += 1, + other => panic!("unexpected {other:?}"), + } + } + } + } + assert_eq!(filled, want.len(), "exchanges are fixed length"); + assert_eq!(&got[..filled], want, "the child answered"); + } + let mut l = Loop::new(Config::default()).expect("loop"); + let mut spec = ProcessSpec::new(env!("CARGO_BIN_EXE_native_child")); + spec.windows_hide = true; + spec.args = vec!["channel".into()]; + spec.stdio = [ProcessStdio::Null, ProcessStdio::Pipe, ProcessStdio::Null]; + spec.extra = vec![ + ChildFd { + number: 3, + source: ChildFdSource::Duplex, + }, + ChildFd { + number: 4, + source: ChildFdSource::Duplex, + }, + ]; + spec.env = vec![ + ("NODE_CHANNEL_FD".into(), "3".into()), + ("TURNLOOP_EXTRA_FD".into(), "4".into()), + ]; + let mut ends = [[None; 2]; CHILDREN]; + let children: [_; CHILDREN] = std::array::from_fn(|i| { + l.spawn_extra(&spec, Token(i as u64), &mut ends[i]) + .expect("child with extra descriptors") + }); + let mut out = Completions::default(); + // Warm the notifier, services and pooled buffers, leaving the children live. + l.turn(Timeout::Now, &mut out).expect("warm services"); + assert!(out.is_empty()); + let mut seen = [0usize; 4]; + ALLOCS.with(|n| n.set(0)); + ACTIVE.with(|v| v.set(true)); + for (i, child) in children.iter().enumerate() { + for (extra, end) in ends[i].iter().enumerate() { + let h = end.expect("parent end"); + l.write(h, ping(extra), Token(71)).expect("write ping"); + collect( + &mut l, + &mut out, + h, + if extra == 0 { b"pong-3" } else { b"pong-4" }, + &mut seen, + ); + } + collect( + &mut l, + &mut out, + child.stdout.expect("child stdout"), + b"ok", + &mut seen, + ); + } + // Every child must report its own exit before anything is closed, so a + // Cancelled watch can never stand in for a delivered status. + let until = l.now() + Duration::from_secs(20); + while seen[1] < CHILDREN { + assert!(l.now() < until, "exit deadline"); + l.turn(Timeout::Until(until), &mut out).expect("exits"); + for c in out.drain() { + match c.result { + OpResult::Exited(status) => { + assert_eq!(status.code, Some(0)); + seen[1] += 1; + } + OpResult::Eof => seen[2] += 1, + other => panic!("unexpected exit-phase completion {other:?}"), + } + } + } + for (i, child) in children.iter().enumerate() { + for h in [ends[i][0], ends[i][1], child.stdout, Some(child.handle)] { + l.close(h.expect("handle"), Token(72)).expect("close"); + } + } + while l.alive() { + assert!(l.now() < until, "teardown deadline"); + l.turn(Timeout::Until(until), &mut out).expect("teardown"); + for c in out.drain() { + match c.result { + OpResult::Eof => seen[2] += 1, + OpResult::Closed => seen[3] += 1, + other => panic!("unexpected teardown {other:?}"), + } + } + } + ACTIVE.with(|v| v.set(false)); + assert_eq!( + ALLOCS.with(Cell::get), + 0, + "extra-descriptor writes, reads, exits and closes" + ); + assert_eq!(seen[0], CHILDREN * 2, "every ping completed"); + assert_eq!(seen[1], CHILDREN, "every child exited exactly once"); + assert_eq!(seen[3], CHILDREN * 4, "every handle closed exactly once"); +} + #[cfg(unix)] #[test] fn signal_exit_and_external_notification_delivery_allocate_nothing() { diff --git a/crates/turnloop-contract/tests/process_fds.rs b/crates/turnloop-contract/tests/process_fds.rs new file mode 100644 index 0000000..acd39fa --- /dev/null +++ b/crates/turnloop-contract/tests/process_fds.rs @@ -0,0 +1,693 @@ +//! Extra child descriptors: Node's `stdio` tail, its IPC channel, and the +//! session control a pty child needs. Everything here runs on Unix and Windows +//! except the two cases marked `#[cfg]`, which are about descriptor numbers and +//! controlling terminals and have no Windows counterpart. +#![deny(unsafe_op_in_unsafe_fn)] +#![cfg(all( + not(loom), + any( + target_vendor = "apple", + target_os = "linux", + target_os = "android", + target_os = "freebsd", + target_os = "windows" + ) +))] +use std::time::Duration; +use turnloop::*; + +const CHILD: &str = env!("CARGO_BIN_EXE_native_child"); +const LIMIT: Duration = Duration::from_secs(20); + +/// Everything a turn can deliver that a descriptor exchange does not consume. +#[derive(Default)] +struct Seen { + exit: Option, + eof: usize, + wrote: usize, + cancelled: usize, + closed: usize, + /// Terminal lifecycle results in arrival order, as `(handle, Cancelled?)`. + terminals: Vec<(u64, bool)>, +} + +struct Rig { + driver: Loop, + out: Completions, + seen: Seen, + /// Bytes already delivered for a handle but not yet consumed by a caller. + /// A stream read returns whatever has arrived, so framing is the test's job. + buffered: std::collections::HashMap>, +} +impl Rig { + fn new() -> Self { + Self { + driver: Loop::new(Config::default()).expect("loop"), + out: Completions::default(), + seen: Seen::default(), + buffered: std::collections::HashMap::new(), + } + } + /// One bounded turn, routing reads into their handle's buffer. + fn turn(&mut self, deadline: Instant) { + assert!(self.driver.now() < deadline, "deadline"); + self.turn_for(Timeout::Until(deadline)); + } + /// A short turn for waiting on something the loop itself cannot observe, + /// such as a grandchild that no handle of this loop refers to. + fn settle(&mut self) { + self.turn_for(Timeout::After(Duration::from_millis(10))); + } + fn turn_for(&mut self, timeout: Timeout) { + self.driver.turn(timeout, &mut self.out).expect("turn"); + for c in self.out.drain() { + match c.result { + OpResult::Read { n, lease } => { + assert!(n > 0, "a read completion carries bytes"); + let bytes = lease.expect("pooled lease"); + let handle = c.handle.expect("a read names its handle"); + self.buffered + .entry(handle.key()) + .or_default() + .extend_from_slice(bytes.as_slice()); + } + OpResult::Wrote(n) => { + assert!(n > 0); + self.seen.wrote += 1; + } + OpResult::Eof => self.seen.eof += 1, + OpResult::Exited(status) => { + assert!(self.seen.exit.replace(status).is_none(), "exit is once"); + } + OpResult::Cancelled => { + assert!(c.op.is_some(), "a cancellation names its operation"); + assert!(c.terminal); + self.seen.cancelled += 1; + self.seen + .terminals + .push((c.handle.expect("handle").key(), true)); + } + OpResult::Closed => { + assert!(c.op.is_none(), "Closed is the handle's own result"); + assert!(c.terminal); + self.seen.closed += 1; + self.seen + .terminals + .push((c.handle.expect("handle").key(), false)); + } + other => panic!("unexpected completion {other:?}"), + } + } + } + fn have(&self, h: Handle) -> usize { + self.buffered.get(&h.key()).map_or(0, Vec::len) + } + /// Consume exactly `want` bytes from `h`, reading more when short. + fn read_exact(&mut self, h: Handle, want: usize) -> Vec { + let deadline = self.driver.now() + LIMIT; + while self.have(h) < want { + let before = self.have(h); + self.driver + .read(h, ReadBuf::Pooled, Token(90)) + .expect("submit read"); + while self.have(h) == before { + self.turn(deadline); + } + } + let rest = self.buffered.get_mut(&h.key()).expect("buffered bytes"); + rest.drain(..want).collect() + } + /// Consume one newline-terminated line, newline included. + fn read_line(&mut self, h: Handle) -> String { + let mut line = Vec::new(); + loop { + line.extend_from_slice(&self.read_exact(h, 1)); + if line.ends_with(b"\n") { + return String::from_utf8(line).expect("text line"); + } + } + } + fn write(&mut self, h: Handle, bytes: &[u8]) { + self.driver + .write(h, WriteBuf::Owned(bytes.to_vec()), Token(91)) + .expect("submit write"); + } + fn exit_status(&mut self) -> ExitStatus { + let deadline = self.driver.now() + LIMIT; + while self.seen.exit.is_none() { + self.turn(deadline); + } + self.seen.exit.expect("child exit") + } + /// Close a handle and require `ops` cancellations, then exactly one Closed, + /// in that order and only once. + fn close_completely(&mut self, h: Handle, ops: usize) { + let (cancelled, closed) = (self.seen.cancelled, self.seen.closed); + let before = self.seen.terminals.len(); + self.driver.close(h, Token(92)).expect("close"); + let deadline = self.driver.now() + LIMIT; + while self.seen.closed == closed { + self.turn(deadline); + } + assert_eq!(self.seen.closed, closed + 1, "one Closed per handle"); + assert_eq!( + self.seen.cancelled, + cancelled + ops, + "one Cancelled per outstanding operation" + ); + let mine: Vec = self.seen.terminals[before..] + .iter() + .filter(|(key, _)| *key == h.key()) + .map(|(_, cancelled)| *cancelled) + .collect(); + let mut expected = vec![true; ops]; + expected.push(false); + assert_eq!(mine, expected, "cancellations precede the single Closed"); + } +} + +fn channel_spec() -> ProcessSpec { + let mut spec = ProcessSpec::new(CHILD); + spec.windows_hide = true; + spec.args = vec!["channel".into()]; + spec.stdio = [ProcessStdio::Null, ProcessStdio::Pipe, ProcessStdio::Null]; + spec.extra = vec![ + ChildFd { + number: 3, + source: ChildFdSource::Duplex, + }, + ChildFd { + number: 4, + source: ChildFdSource::Duplex, + }, + ]; + // Exactly Node's handoff: the number lives in the environment, and the + // child finds its channel by reading it back. + spec.env = vec![ + ("NODE_CHANNEL_FD".into(), "3".into()), + ("TURNLOOP_EXTRA_FD".into(), "4".into()), + ]; + spec +} + +#[test] +fn five_descriptors_carry_bytes_both_ways_through_a_channel_handoff() { + let mut rig = Rig::new(); + let spec = channel_spec(); + let mut parents = [None; 2]; + let child = rig + .driver + .spawn_extra(&spec, Token(1), &mut parents) + .expect("spawn with extra descriptors"); + let channel = parents[0].expect("descriptor 3 parent end"); + let extra = parents[1].expect("descriptor 4 parent end"); + let stdout = child.stdout.expect("child stdout"); + assert_ne!(channel, extra); + + rig.write(channel, b"ping-3"); + assert_eq!( + rig.read_exact(channel, 6), + b"pong-3", + "descriptor 3 replies" + ); + rig.write(extra, b"ping-4"); + assert_eq!(rig.read_exact(extra, 6), b"pong-4", "descriptor 4 replies"); + assert_eq!(rig.read_exact(stdout, 2), b"ok", "child ran to completion"); + assert_eq!( + rig.exit_status(), + ExitStatus { + code: Some(0), + signal: None + } + ); + assert_eq!(rig.seen.wrote, 2, "both parent writes completed"); + for h in [channel, extra, stdout] { + rig.close_completely(h, 0); + } + rig.close_completely(child.handle, 0); + assert!(!rig.driver.alive()); +} + +/// An OS pipe created outside the loop, so both ends are ordinary synchronous +/// descriptors a child can use with plain blocking reads and writes. +fn host_pipe() -> (Detached, Detached) { + #[cfg(unix)] + { + use std::os::fd::{FromRawFd, OwnedFd}; + let mut fds = [0; 2]; + // SAFETY: writable pair of output descriptors, no other arguments. + assert_eq!(unsafe { libc::pipe(fds.as_mut_ptr()) }, 0); + // SAFETY: pipe transferred ownership of two fresh descriptors. + let (read, write) = unsafe { (OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1])) }; + ( + Detached::from_fd(read).expect("adopt read end"), + Detached::from_fd(write).expect("adopt write end"), + ) + } + #[cfg(windows)] + { + use std::os::windows::io::{FromRawHandle, OwnedHandle}; + let (mut read, mut write) = (std::ptr::null_mut(), std::ptr::null_mut()); + // SAFETY: two writable handle outputs; default attributes and buffer size. + let ok = unsafe { + windows_sys::Win32::System::Pipes::CreatePipe( + &mut read, + &mut write, + std::ptr::null(), + 0, + ) + }; + assert_ne!(ok, 0, "{}", std::io::Error::last_os_error()); + // SAFETY: CreatePipe transferred ownership of two fresh handles. + let (read, write) = unsafe { + ( + OwnedHandle::from_raw_handle(read), + OwnedHandle::from_raw_handle(write), + ) + }; + ( + Detached::from_handle(read).expect("adopt read end"), + Detached::from_handle(write).expect("adopt write end"), + ) + } +} + +#[test] +fn extra_sources_cover_one_way_pipes_the_null_device_and_adopted_transports() { + let mut rig = Rig::new(); + let (read, write) = host_pipe(); + let read = rig.driver.attach(read, Token(2)).expect("attach read end"); + let write = rig + .driver + .attach(write, Token(3)) + .expect("attach write end"); + let mut spec = ProcessSpec::new(CHILD); + spec.windows_hide = true; + spec.args = vec!["extra-sources".into()]; + spec.stdio = [ProcessStdio::Null, ProcessStdio::Pipe, ProcessStdio::Null]; + spec.extra = vec![ + ChildFd { + number: 3, + source: ChildFdSource::Pipe, + }, + ChildFd { + number: 4, + source: ChildFdSource::Null, + }, + ChildFd { + number: 5, + source: ChildFdSource::Handle(write), + }, + ]; + let mut parents = [None; 3]; + let child = rig + .driver + .spawn_extra(&spec, Token(4), &mut parents) + .expect("spawn"); + let one_way = parents[0].expect("one-way parent end"); + assert!(parents[1].is_none(), "the null device has no parent end"); + assert!(parents[2].is_none(), "an adopted transport has no new end"); + let stdout = child.stdout.expect("child stdout"); + + assert_eq!(rig.read_exact(one_way, 5), b"three", "child wrote fd 3"); + assert_eq!(rig.read_exact(read, 4), b"five", "child wrote fd 5"); + assert_eq!(rig.read_exact(stdout, 2), b"ok", "null device read as EOF"); + assert_eq!( + rig.exit_status(), + ExitStatus { + code: Some(0), + signal: None + } + ); + // The loop kept its own end of the adopted transport throughout. + assert!(rig.driver.raw_transport(write).is_ok()); + for h in [one_way, stdout, read, write] { + rig.close_completely(h, 0); + } + rig.close_completely(child.handle, 0); +} + +#[test] +fn close_orders_cancel_before_closed_for_a_child_holding_extra_descriptors() { + let mut rig = Rig::new(); + let mut spec = ProcessSpec::new(CHILD); + spec.windows_hide = true; + spec.args = vec!["sleep".into()]; + spec.stdio = [ProcessStdio::Null; 3]; + spec.extra = vec![ + ChildFd { + number: 3, + source: ChildFdSource::Duplex, + }, + ChildFd { + number: 4, + source: ChildFdSource::Pipe, + }, + ]; + let mut parents = [None; 2]; + let child = rig + .driver + .spawn_extra(&spec, Token(5), &mut parents) + .expect("spawn"); + let extras = [parents[0].expect("fd 3"), parents[1].expect("fd 4")]; + // Closing a live child terminates it and waits for reaping: its exit + // operation is the one outstanding operation, so it yields one Cancelled. + rig.close_completely(child.handle, 1); + assert!(rig.seen.exit.is_none(), "a cancelled watch reports no exit"); + #[cfg(unix)] + { + let mut status = 0; + assert_eq!( + // SAFETY: WNOHANG query of this fixture child only; ECHILD proves + // the library reaped it rather than leaving a zombie. + unsafe { libc::waitpid(child.pid as i32, &mut status, libc::WNOHANG) }, + -1 + ); + assert_eq!( + std::io::Error::last_os_error().raw_os_error(), + Some(libc::ECHILD) + ); + } + for h in extras { + rig.close_completely(h, 0); + } + assert!(!rig.driver.alive()); + rig.driver.turn(Timeout::Now, &mut rig.out).expect("quiet"); + assert!(rig.out.is_empty(), "no duplicate completions"); +} + +#[test] +fn rejected_descriptor_plans_create_nothing() { + let mut driver = Loop::new(Config::default()).expect("loop"); + let base = channel_spec(); + let refuse = |driver: &mut Loop, spec: &ProcessSpec, parents: &mut [Option]| { + let error = driver + .spawn_extra(spec, Token(6), parents) + .expect_err("rejected plan"); + assert_eq!(error.kind, ErrorKind::InvalidInput); + assert!(parents.iter().all(Option::is_none), "nothing was created"); + assert!(!driver.alive(), "no handle survived a rejected spawn"); + }; + // The parent-end slice must match the plan exactly. + refuse(&mut driver, &base, &mut [None; 1]); + refuse(&mut driver, &base, &mut [None; 3]); + // spawn() has nowhere to report parent ends. + assert_eq!( + driver.spawn(&base, Token(6)).expect_err("no sink").kind, + ErrorKind::InvalidInput + ); + assert!(!driver.alive()); + for number in [0, 1, 2, MAX_CHILD_FD + 1, u32::MAX] { + let mut spec = base.clone(); + spec.extra[1].number = number; + refuse(&mut driver, &spec, &mut [None; 2]); + } + // A repeated number would silently drop one of the two descriptors. + let mut spec = base.clone(); + spec.extra[1].number = 3; + refuse(&mut driver, &spec, &mut [None; 2]); + // Claiming a controlling terminal requires the new session that grants it. + let mut spec = base.clone(); + spec.controlling_terminal = true; + refuse(&mut driver, &spec, &mut [None; 2]); + // A handle another loop owns cannot be duplicated into a child. + let mut elsewhere = Loop::new(Config::default()).expect("second loop"); + let (read, _write) = host_pipe(); + let foreign = elsewhere.attach(read, Token(0)).expect("attach elsewhere"); + let mut spec = base.clone(); + spec.extra[1].source = ChildFdSource::Handle(foreign); + let error = driver + .spawn_extra(&spec, Token(6), &mut [None; 2]) + .expect_err("foreign handle"); + assert!(matches!( + error.kind, + ErrorKind::NotFound | ErrorKind::InvalidInput + )); + assert!(!driver.alive(), "no handle survived a rejected spawn"); +} + +/// A failed exec must be reported as a failed spawn, even when the descriptor +/// numbers under test are exactly the ones the standard library would otherwise +/// hand to its own exec-error pipe. +/// +/// `Command::spawn` creates that pipe after the standard streams and before the +/// fork, and it takes the lowest free numbers — the very ones the extra +/// descriptors' sources vacate when they are lifted above their targets. The +/// two lowest free numbers are probed and released here so that the collision +/// is the expected outcome rather than a coincidence: inherited standard streams +/// and null-device sources allocate nothing else in between. +#[cfg(unix)] +#[test] +fn a_failed_exec_is_reported_even_at_the_lowest_free_descriptor_numbers() { + use std::os::fd::{AsRawFd, FromRawFd, OwnedFd}; + let mut driver = Loop::new(Config::default()).expect("loop"); + let lowest = |from: i32| { + // SAFETY: duplicates descriptor 0 at or above `from`, nothing else. + let raw = unsafe { libc::fcntl(0, libc::F_DUPFD_CLOEXEC, from) }; + assert!(raw >= from, "{}", std::io::Error::last_os_error()); + // SAFETY: fcntl transferred ownership of a fresh descriptor. + unsafe { OwnedFd::from_raw_fd(raw) } + }; + let (first, second) = (lowest(3), lowest(4)); + let numbers = [first.as_raw_fd() as u32, second.as_raw_fd() as u32]; + drop((first, second)); + let mut spec = ProcessSpec::new("/nonexistent/turnloop-procspec-fixture"); + spec.stdio = [ProcessStdio::Inherit; 3]; + spec.extra = numbers + .iter() + .map(|number| ChildFd { + number: *number, + source: ChildFdSource::Null, + }) + .collect(); + let mut parents = [None; 2]; + let error = driver + .spawn_extra(&spec, Token(13), &mut parents) + .expect_err("a missing program cannot be executed"); + assert_eq!(error.kind, ErrorKind::NotFound, "exec failure reached us"); + assert!(parents.iter().all(Option::is_none)); + assert!(!driver.alive(), "no handle survived the failed spawn"); +} + +/// turnloop reaps only the children it owns, by their own identity. A host that +/// waits for its own child in the same process keeps that child's status, and +/// loses nothing to the loop. +#[test] +fn a_sibling_waiter_and_the_loop_keep_their_own_children() { + let mut rig = Rig::new(); + let sibling = |code: &str| { + std::process::Command::new(CHILD) + .args(["exit-with", code]) + .spawn() + .expect("sibling child") + }; + let mut spec = ProcessSpec::new(CHILD); + spec.windows_hide = true; + spec.args = vec!["exit-with".into(), "23".into()]; + spec.stdio = [ProcessStdio::Null; 3]; + spec.extra = vec![ChildFd { + number: 3, + source: ChildFdSource::Duplex, + }]; + + // The sibling exits first and stays unreaped while the loop reaps its own. + let mut first = sibling("7"); + let mut parents = [None; 1]; + let owned = rig + .driver + .spawn_extra(&spec, Token(8), &mut parents) + .expect("owned child"); + assert_eq!( + rig.exit_status(), + ExitStatus { + code: Some(23), + signal: None + }, + "the loop reported its own child's status" + ); + assert_eq!( + first.wait().expect("reap sibling").code(), + Some(7), + "the sibling's status survived the loop's reaping" + ); + rig.close_completely(parents[0].expect("fd 3"), 0); + rig.close_completely(owned.handle, 0); + + // And the other order: the host reaps first, then the loop delivers. + let mut second = sibling("11"); + assert_eq!(second.wait().expect("reap sibling").code(), Some(11)); + rig.seen.exit = None; + let mut parents = [None; 1]; + let owned = rig + .driver + .spawn_extra(&spec, Token(9), &mut parents) + .expect("owned child"); + assert_eq!( + rig.exit_status(), + ExitStatus { + code: Some(23), + signal: None + }, + "a host reap did not consume the loop's child" + ); + rig.close_completely(parents[0].expect("fd 3"), 0); + rig.close_completely(owned.handle, 0); +} + +/// A process identity pinned so that it cannot be confused with a later reuse. +struct Tracked { + #[cfg(unix)] + pid: i32, + #[cfg(windows)] + process: std::os::windows::io::OwnedHandle, +} +impl Tracked { + fn pin(pid: u32) -> Self { + #[cfg(unix)] + { + Self { pid: pid as i32 } + } + #[cfg(windows)] + { + use std::os::windows::io::FromRawHandle; + // SAFETY: the loop's owned leader keeps this descendant's identity + // alive; only synchronization access is requested. + let raw = unsafe { + windows_sys::Win32::System::Threading::OpenProcess( + windows_sys::Win32::System::Threading::PROCESS_SYNCHRONIZE, + 0, + pid, + ) + }; + assert!(!raw.is_null(), "{}", std::io::Error::last_os_error()); + Self { + // SAFETY: a successful OpenProcess transferred this handle. + process: unsafe { std::os::windows::io::OwnedHandle::from_raw_handle(raw) }, + } + } + } + fn alive(&self) -> bool { + #[cfg(unix)] + { + // SAFETY: signal 0 only probes for the process's existence. + unsafe { libc::kill(self.pid, 0) == 0 } + } + #[cfg(windows)] + { + use std::os::windows::io::AsRawHandle; + // SAFETY: owned synchronization handle and a nonblocking query. + unsafe { + windows_sys::Win32::System::Threading::WaitForSingleObject( + self.process.as_raw_handle(), + 0, + ) == windows_sys::Win32::Foundation::WAIT_TIMEOUT + } + } + } +} + +#[test] +fn a_process_group_with_extra_descriptors_still_kills_its_grandchild() { + let mut rig = Rig::new(); + let mut spec = ProcessSpec::new(CHILD); + spec.windows_hide = true; + spec.args = vec!["grandchild".into()]; + spec.stdio = [ProcessStdio::Null, ProcessStdio::Pipe, ProcessStdio::Null]; + spec.new_process_group = true; + spec.extra = vec![ChildFd { + number: 3, + source: ChildFdSource::Duplex, + }]; + let mut parents = [None; 1]; + let child = rig + .driver + .spawn_extra(&spec, Token(10), &mut parents) + .expect("leader"); + let stdout = child.stdout.expect("leader stdout"); + let text = rig.read_line(stdout); + let grandchild = Tracked::pin( + text.trim() + .strip_prefix("grandchild:") + .expect("identity") + .parse() + .expect("pid"), + ); + assert!(grandchild.alive(), "grandchild is live before the kill"); + rig.driver + .kill_group(child.handle, Signal::Kill) + .expect("kill the whole group"); + let status = rig.exit_status(); + #[cfg(unix)] + assert_eq!(status.signal, Some(libc::SIGKILL)); + #[cfg(windows)] + assert_eq!(status.code, Some(1), "TerminateJobObject exit code"); + let deadline = rig.driver.now() + LIMIT; + while grandchild.alive() { + assert!(rig.driver.now() < deadline, "grandchild outlived its group"); + rig.settle(); + } + rig.close_completely(parents[0].expect("fd 3"), 0); + rig.close_completely(stdout, 0); + rig.close_completely(child.handle, 0); +} + +#[cfg(any(target_vendor = "apple", target_os = "linux", target_os = "freebsd"))] +#[test] +fn a_detached_child_can_claim_its_stdin_as_a_controlling_terminal() { + use std::os::fd::{FromRawFd, OwnedFd}; + let (mut master, mut slave) = (0, 0); + assert_eq!( + // SAFETY: two writable descriptor outputs; null term/window settings + // select the platform defaults. + unsafe { + libc::openpty( + &mut master, + &mut slave, + std::ptr::null_mut(), + std::ptr::null_mut(), + std::ptr::null_mut(), + ) + }, + 0 + ); + // SAFETY: openpty transferred ownership of two fresh descriptors. + let (_master, slave) = unsafe { (OwnedFd::from_raw_fd(master), OwnedFd::from_raw_fd(slave)) }; + let mut rig = Rig::new(); + let terminal = rig + .driver + .attach( + Detached::from_fd(slave.try_clone().expect("dup slave")).expect("adopt slave"), + Token(11), + ) + .expect("attach slave"); + let mut spec = ProcessSpec::new(CHILD); + spec.args = vec!["tty-session".into()]; + spec.stdio = [ + ProcessStdio::Handle(terminal), + ProcessStdio::Pipe, + ProcessStdio::Null, + ]; + spec.detached = true; + spec.controlling_terminal = true; + let child = rig.driver.spawn(&spec, Token(12)).expect("pty child"); + let stdout = child.stdout.expect("child stdout"); + let text = rig.read_line(stdout); + let fields: Vec<&str> = text.trim().split(':').collect(); + assert_eq!(fields.len(), 4, "pid:group:session:terminal"); + let pid: i32 = fields[0].parse().expect("pid"); + assert_eq!(pid, child.pid as i32); + assert_eq!(fields[1], fields[0], "the child leads its own group"); + assert_eq!(fields[2], fields[0], "and its own session"); + assert_eq!(fields[3], "1", "the child has a controlling terminal"); + assert_eq!( + rig.exit_status(), + ExitStatus { + code: Some(0), + signal: None + } + ); + rig.close_completely(stdout, 0); + rig.close_completely(child.handle, 0); + rig.close_completely(terminal, 0); + drop(slave); +} diff --git a/crates/turnloop/src/backend/iocp/mod.rs b/crates/turnloop/src/backend/iocp/mod.rs index 1eaae2b..4da7082 100644 --- a/crates/turnloop/src/backend/iocp/mod.rs +++ b/crates/turnloop/src/backend/iocp/mod.rs @@ -1160,6 +1160,7 @@ unsafe impl Backend for Iocp { &mut self, handle: Handle, pipes: [Option; 3], + extra: &[Option], spec: &ProcessSpec, ) -> Result { let mut existing = [None; 3]; @@ -1168,8 +1169,19 @@ unsafe impl Backend for Iocp { existing[i] = Some(self.get(*h)?.transport.native.raw()); } } - let (child, parents) = - process::spawn(spec, existing, self.notifier.clone().ok_or_else(invalid)?)?; + let mut sources = Vec::with_capacity(spec.extra.len()); + for fd in &spec.extra { + sources.push(match fd.source { + crate::ChildFdSource::Handle(h) => Some(self.get(h)?.transport.native.raw()), + _ => None, + }); + } + let (child, parents, extra_parents) = process::spawn( + spec, + existing, + &sources, + self.notifier.clone().ok_or_else(invalid)?, + )?; let pid = child.pid; let result = (|| { for (h, parent) in pipes.into_iter().zip(parents) { @@ -1177,10 +1189,19 @@ unsafe impl Backend for Iocp { self.install(h, parent, None, None)?; } } + for (h, parent) in extra.iter().zip(extra_parents) { + if let (Some(h), Some(parent)) = (h, parent) { + self.install(*h, parent, None, None)?; + } + } child.resume() })(); if let Err(error) = result { - for h in pipes.into_iter().flatten() { + for h in pipes + .into_iter() + .flatten() + .chain(extra.iter().flatten().copied()) + { self.release(h); } return Err(error); diff --git a/crates/turnloop/src/backend/iocp/process.rs b/crates/turnloop/src/backend/iocp/process.rs index e7cb347..8f19336 100644 --- a/crates/turnloop/src/backend/iocp/process.rs +++ b/crates/turnloop/src/backend/iocp/process.rs @@ -1,5 +1,5 @@ use super::{Detached, Kind, Native, bool_result, invalid, os_error, port::owned, unsupported}; -use crate::{ExitStatus, Notifier, ProcessSpec, ProcessStdio, Result, Signal}; +use crate::{ChildFdSource, ExitStatus, Notifier, ProcessSpec, ProcessStdio, Result, Signal}; use std::{ collections::BTreeMap, ffi::{OsStr, OsString, c_void}, @@ -166,13 +166,13 @@ pub(super) fn duplicate(handle: HANDLE, inherit: bool) -> Result { // SAFETY: successful DuplicateHandle transferred unique ownership. unsafe { owned(out) }.map_err(Into::into) } -pub(super) fn null(input: bool) -> Result { +pub(super) fn null(access: u32) -> Result { let name: Vec = "NUL\0".encode_utf16().collect(); // SAFETY: valid device path; exclusive ownership of newly created handle. unsafe { owned(CreateFileW( name.as_ptr(), - if input { GENERIC_READ } else { GENERIC_WRITE }, + access, FILE_SHARE_READ | FILE_SHARE_WRITE, ptr::null(), OPEN_EXISTING, @@ -187,12 +187,30 @@ pub(super) fn stdio(index: usize, inherit: bool) -> Result { let handle = unsafe { GetStdHandle([STD_INPUT_HANDLE, STD_OUTPUT_HANDLE, STD_ERROR_HANDLE][index]) }; if handle.is_null() || handle == INVALID_HANDLE_VALUE { - let handle = null(index == 0)?; + let handle = null(read_or_write(index == 0))?; return duplicate(handle.as_raw_handle(), inherit); } duplicate(handle, inherit) } -fn pipe(index: usize) -> Result<(Detached, OwnedHandle)> { +/// Which ends of a child pipe each side may use. +#[derive(Clone, Copy, Eq, PartialEq)] +enum Direction { + /// Child stdin: the parent writes, the child reads. + ParentWrites, + /// Child stdout/stderr and a one-way extra descriptor: the child writes. + ParentReads, + /// Both ends readable and writable, as Node's stdio pipes and IPC channel are. + Duplex, +} +fn read_or_write(read: bool) -> u32 { + if read { GENERIC_READ } else { GENERIC_WRITE } +} +fn pipe(direction: Direction) -> Result<(Detached, OwnedHandle)> { + let (parent_access, child_access) = match direction { + Direction::ParentWrites => (PIPE_ACCESS_OUTBOUND, GENERIC_READ), + Direction::ParentReads => (PIPE_ACCESS_INBOUND, GENERIC_WRITE), + Direction::Duplex => (PIPE_ACCESS_DUPLEX, GENERIC_READ | GENERIC_WRITE), + }; static NEXT: AtomicU64 = AtomicU64::new(0); let name = format!( r"\\.\pipe\turnloop-stdio-{}-{}", @@ -204,13 +222,7 @@ fn pipe(index: usize) -> Result<(Detached, OwnedHandle)> { let parent = unsafe { owned(CreateNamedPipeW( name.as_ptr(), - FILE_FLAG_OVERLAPPED - | FILE_FLAG_FIRST_PIPE_INSTANCE - | if index == 0 { - PIPE_ACCESS_OUTBOUND - } else { - PIPE_ACCESS_INBOUND - }, + FILE_FLAG_OVERLAPPED | FILE_FLAG_FIRST_PIPE_INSTANCE | parent_access, PIPE_TYPE_BYTE | PIPE_WAIT | PIPE_REJECT_REMOTE_CLIENTS, 1, 65536, @@ -223,11 +235,7 @@ fn pipe(index: usize) -> Result<(Detached, OwnedHandle)> { let child = unsafe { owned(CreateFileW( name.as_ptr(), - if index == 0 { - GENERIC_READ - } else { - GENERIC_WRITE - }, + child_access, 0, ptr::null(), OPEN_EXISTING, @@ -462,35 +470,143 @@ impl Drop for Child { } } } +/// C run-time descriptor flags, as every MSVCRT/UCRT program reads them back out +/// of the inherited-descriptor block at startup. +const FOPEN: u8 = 0x01; +const FPIPE: u8 = 0x08; +const FDEV: u8 = 0x40; + +fn crt_flags(handle: HANDLE) -> u8 { + // SAFETY: live inheritable handle; the query only classifies it. + FOPEN + | match unsafe { GetFileType(handle) } { + FILE_TYPE_PIPE => FPIPE, + FILE_TYPE_CHAR => FDEV, + _ => 0, + } +} + +/// The inherited-descriptor block passed through `STARTUPINFOW.lpReserved2`. +/// +/// This is how a child gets descriptor numbers at all on Windows, and it is the +/// same convention libuv and Node use, so `NODE_CHANNEL_FD=3` names the same +/// thing in a child here as it does under Node. The layout is the C run-time's: +/// a descriptor count, then one flag byte per descriptor, then one handle per +/// descriptor, all packed without padding, so every handle is written as bytes. +/// Unused numbers below the highest one in use are present and closed. +fn inherited_block(slots: &[Option]) -> Result> { + let count = slots.len(); + let width = size_of::(); + let bytes = size_of::() + count + count * width; + // cbReserved2 is a u16; MAX_CHILD_FD keeps this far below the limit. + if u16::try_from(bytes).is_err() { + return Err(invalid()); + } + let mut block = vec![0u8; bytes]; + block[..size_of::()].copy_from_slice(&(count as i32).to_ne_bytes()); + for (i, slot) in slots.iter().enumerate() { + let (flags, handle) = match slot { + Some(handle) => ( + crt_flags(handle.as_raw_handle()), + handle.as_raw_handle() as usize, + ), + None => (0, INVALID_HANDLE_VALUE as usize), + }; + block[size_of::() + i] = flags; + let at = size_of::() + count + i * width; + block[at..at + width].copy_from_slice(&handle.to_ne_bytes()); + } + Ok(block) +} + +/// A started child, the parent ends of its standard streams, and the parent ends +/// of its extra descriptors in `ProcessSpec::extra` order. +type Spawned = (Child, [Option; 3], Vec>); + pub(super) fn spawn( spec: &ProcessSpec, existing: [Option; 3], + extra_sources: &[Option], notifier: Notifier, -) -> Result<(Child, [Option; 3])> { +) -> Result { if spec.uid.is_some() || spec.gid.is_some() || spec.program.is_empty() { return Err(unsupported()); } + if spec.controlling_terminal { + // Windows has no session/controlling-terminal concept to claim. + return Err(unsupported()); + } let application = wide(program(spec)?.as_os_str())?; let mut parents = [None, None, None]; - let mut child_ends = Vec::with_capacity(3); + let slot_count = spec + .extra + .iter() + .map(|fd| fd.number as usize + 1) + .max() + .unwrap_or(0) + .max(3); + let mut slots: Vec> = (0..slot_count).map(|_| None).collect(); for (i, option) in spec.stdio.iter().enumerate() { let handle = match option { ProcessStdio::Inherit => stdio(i, true)?, ProcessStdio::Null => { - let handle = null(i == 0)?; + let handle = null(read_or_write(i == 0))?; duplicate(handle.as_raw_handle(), true)? } ProcessStdio::Handle(_) => duplicate(existing[i].ok_or_else(invalid)?, true)?, ProcessStdio::Pipe => { - let (parent, child) = pipe(i)?; + let (parent, child) = pipe(if i == 0 { + Direction::ParentWrites + } else { + Direction::ParentReads + })?; parents[i] = Some(parent); duplicate(child.as_raw_handle(), true)? } }; - child_ends.push(handle); + slots[i] = Some(handle); + } + let mut extra_parents = Vec::with_capacity(spec.extra.len()); + for (i, fd) in spec.extra.iter().enumerate() { + let (handle, parent) = match fd.source { + ChildFdSource::Null => { + let handle = null(GENERIC_READ | GENERIC_WRITE)?; + (duplicate(handle.as_raw_handle(), true)?, None) + } + ChildFdSource::Handle(_) => ( + duplicate( + extra_sources + .get(i) + .copied() + .flatten() + .ok_or_else(invalid)?, + true, + )?, + None, + ), + ChildFdSource::Pipe => { + let (parent, child) = pipe(Direction::ParentReads)?; + (duplicate(child.as_raw_handle(), true)?, Some(parent)) + } + ChildFdSource::Duplex => { + let (parent, child) = pipe(Direction::Duplex)?; + (duplicate(child.as_raw_handle(), true)?, Some(parent)) + } + }; + let slot = slots.get_mut(fd.number as usize).ok_or_else(invalid)?; + if slot.is_some() { + return Err(invalid()); + } + *slot = Some(handle); + extra_parents.push(parent); } - let handles = std::array::from_fn::<_, 3, _>(|i| child_ends[i].as_raw_handle()); + let handles: Vec = slots + .iter() + .flatten() + .map(AsRawHandle::as_raw_handle) + .collect(); let mut attributes = Attributes::new(&handles)?; + let mut block = inherited_block(&slots)?; let mut command = Vec::new(); quote(&spec.program, &mut command)?; for arg in &spec.args { @@ -542,9 +658,17 @@ pub(super) fn spawn( } else { SW_SHOWDEFAULT } as u16; - startup.StartupInfo.hStdInput = handles[0]; - startup.StartupInfo.hStdOutput = handles[1]; - startup.StartupInfo.hStdError = handles[2]; + let standard = std::array::from_fn::<_, 3, _>(|i| { + slots[i] + .as_ref() + .map_or(ptr::null_mut(), AsRawHandle::as_raw_handle) + }); + startup.StartupInfo.hStdInput = standard[0]; + startup.StartupInfo.hStdOutput = standard[1]; + startup.StartupInfo.hStdError = standard[2]; + // The child's own descriptor numbers, including 0..2, come from here. + startup.StartupInfo.cbReserved2 = block.len() as u16; + startup.StartupInfo.lpReserved2 = block.as_mut_ptr(); startup.lpAttributeList = attributes.0.as_mut_ptr().cast(); let job = if spec.new_process_group || spec.detached { // Explicit tree control is separate from parent lifetime. Releasing this @@ -561,7 +685,8 @@ pub(super) fn spawn( // SAFETY: plain writable process output structure. let mut info: PROCESS_INFORMATION = unsafe { std::mem::zeroed() }; // SAFETY: explicit application and quoted writable command line; environment, - // directory, attribute list and exactly the listed inherited handles stay live. + // directory, attribute list, inherited-descriptor block and exactly the + // listed inherited handles all stay live across this call. bool_result(unsafe { CreateProcessW( application.as_ptr(), @@ -628,7 +753,7 @@ pub(super) fn spawn( WT_EXECUTEONLYONCE, ) })?; - Ok((child, parents)) + Ok((child, parents, extra_parents)) } #[cfg(all(test, not(loom)))] @@ -647,8 +772,8 @@ mod tests { spec.stdio = [ProcessStdio::Null; 3]; spec.windows_hide = true; spec.detached = detached; - let (mut child, _) = - spawn(&spec, [None; 3], driver.notifier()).expect("suspended child"); + let (mut child, _, _) = + spawn(&spec, [None; 3], &[], driver.notifier()).expect("suspended child"); let mut member = -1; assert_ne!( // SAFETY: both owned live handles and writable membership output. @@ -705,7 +830,7 @@ mod tests { spec.args.push("--list".into()); spec.stdio = [ProcessStdio::Null; 3]; spec.new_process_group = group; - spawn(&spec, [None; 3], driver.notifier()) + spawn(&spec, [None; 3], &[], driver.notifier()) .expect("suspended child") .0 } @@ -902,8 +1027,8 @@ mod tests { spec.windows_hide = true; spec.args.push("--list".into()); spec.stdio = [ProcessStdio::Null; 3]; - let (mut child, _) = - spawn(&spec, [None; 3], driver.notifier()).expect("suspended child"); + let (mut child, _, _) = + spawn(&spec, [None; 3], &[], driver.notifier()).expect("suspended child"); // The child cannot exit while suspended. Removing its wait now forces // the exact callback-lag window, independent of thread-pool scheduling. child.join().expect("unregister before resume"); diff --git a/crates/turnloop/src/backend/mod.rs b/crates/turnloop/src/backend/mod.rs index 4983d54..dcb17c3 100644 --- a/crates/turnloop/src/backend/mod.rs +++ b/crates/turnloop/src/backend/mod.rs @@ -338,10 +338,15 @@ pub unsafe trait Backend: Sized + 'static { fn set_notifier(&mut self, _notifier: crate::Notifier) {} /// Spawn and bind the child and requested parent pipe handles atomically. /// Failure must release all supplied handles and reap any created child. + /// + /// `extra` is parallel to `spec.extra`: a handle is supplied for every entry + /// whose source has a parent end, and `None` for the rest. The core has + /// already checked descriptor numbers and duplicates. fn spawn( &mut self, _handle: Handle, _pipes: [Option; 3], + _extra: &[Option], _spec: &crate::ProcessSpec, ) -> Result { Err(Error::new(crate::ErrorKind::Unsupported)) @@ -488,6 +493,8 @@ pub mod wasi_p3; pub use wasi_p3::WasiP3 as Platform; #[cfg(any(turnloop_backend = "kqueue", turnloop_backend = "epoll"))] mod ipc; +#[cfg(any(turnloop_backend = "kqueue", turnloop_backend = "epoll"))] +mod process; #[cfg(any(turnloop_backend = "kqueue", turnloop_backend = "epoll"))] mod services; diff --git a/crates/turnloop/src/backend/process.rs b/crates/turnloop/src/backend/process.rs new file mode 100644 index 0000000..e45b5f0 --- /dev/null +++ b/crates/turnloop/src/backend/process.rs @@ -0,0 +1,122 @@ +//! Unix child-descriptor plumbing: the extra descriptors a spec asks for, and +//! the relocation that makes the child hook's dup2 sequence order-independent. +use super::poller::last_error; +use crate::{Error, ErrorKind, Result}; +use std::os::{ + fd::{AsRawFd, FromRawFd, OwnedFd, RawFd}, + unix::net::UnixStream, +}; + +/// A descriptor on the platform null device, opened for reading and writing. +pub(super) fn null() -> Result { + // SAFETY: constant NUL-terminated device path and integer flags only. + let raw = unsafe { libc::open(c"/dev/null".as_ptr(), libc::O_RDWR | libc::O_CLOEXEC) }; + if raw < 0 { + return Err(last_error()); + } + // SAFETY: open returned a fresh exclusively owned descriptor. + Ok(unsafe { OwnedFd::from_raw_fd(raw) }) +} + +/// A connected bidirectional stream pair as `(parent end, child end)`. +/// +/// libuv creates every child pipe this way, so a child's extra descriptor +/// behaves the same here as it does under Node, including descriptor passing +/// over an IPC channel. +pub(super) fn stream_pair() -> Result<(OwnedFd, OwnedFd)> { + let (parent, child) = UnixStream::pair().map_err(Error::from)?; + Ok((OwnedFd::from(parent), OwnedFd::from(child))) +} + +/// A one-way pipe as `(parent read end, child write end)`. +/// +/// A real pipe, not a half-shut socket pair: Darwin refuses `SHUT_RD` on a +/// `socketpair` end with `ENOTCONN`, so the direction has to come from the +/// object rather than from a later call. +pub(super) fn one_way() -> Result<(OwnedFd, OwnedFd)> { + let mut fds = [0 as RawFd; 2]; + #[cfg(any(target_os = "linux", target_os = "android", target_os = "freebsd"))] + // SAFETY: writable pair of descriptor outputs and an integer flag. + let created = unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC) }; + #[cfg(not(any(target_os = "linux", target_os = "android", target_os = "freebsd")))] + // SAFETY: writable pair of descriptor outputs; close-on-exec is set below. + let created = unsafe { libc::pipe(fds.as_mut_ptr()) }; + if created < 0 { + return Err(last_error()); + } + // SAFETY: pipe transferred ownership of two fresh descriptors. + let (read, write) = unsafe { (OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1])) }; + #[cfg(not(any(target_os = "linux", target_os = "android", target_os = "freebsd")))] + for fd in [&read, &write] { + // SAFETY: live owned descriptor and integer-only fcntl arguments. + if unsafe { libc::fcntl(fd.as_raw_fd(), libc::F_SETFD, libc::FD_CLOEXEC) } < 0 { + return Err(last_error()); + } + } + Ok((read, write)) +} + +/// Move a child-side descriptor to at least `floor`, keeping close-on-exec. +/// +/// Every source then sits above every target, so the child hook can dup2 them +/// into place in any order without a later source having been overwritten by an +/// earlier target. The original descriptor is closed by the returned value's +/// replacement, never leaked. +pub(super) fn lift(fd: OwnedFd, floor: RawFd) -> Result { + if fd.as_raw_fd() >= floor { + return Ok(fd); + } + // SAFETY: live owned descriptor and an integer-only fcntl command. + let raw = unsafe { libc::fcntl(fd.as_raw_fd(), libc::F_DUPFD_CLOEXEC, floor) }; + if raw < 0 { + return Err(last_error()); + } + // SAFETY: F_DUPFD_CLOEXEC returned a fresh exclusively owned descriptor. + Ok(unsafe { OwnedFd::from_raw_fd(raw) }) +} + +/// Hold every free target number in the parent until the child has forked. +/// +/// `Command::spawn` creates its exec-error pipe after the standard streams and +/// before the fork, and the kernel gives it the lowest free numbers — which is +/// exactly what the sources vacate when they are lifted above the targets. If +/// that pipe landed on a target number, the child hook would `dup2` over it, and +/// a failed `exec` would be reported to the parent as a successful spawn. A +/// number the parent is already using cannot be handed out either, so only the +/// free ones need holding, and each is released as soon as the child exists. +pub(super) fn reserve(donor: RawFd, numbers: impl Iterator) -> Result> { + let mut held = Vec::new(); + for number in numbers { + // SAFETY: integer-only query of one descriptor's flags. + if unsafe { libc::fcntl(number, libc::F_GETFD) } >= 0 { + continue; + } + if last_error().os != Some(libc::EBADF) { + return Err(last_error()); + } + // SAFETY: live donor descriptor, lifted above every target, and a + // number this process is not using. + if unsafe { libc::dup2(donor, number) } < 0 { + return Err(last_error()); + } + // SAFETY: dup2 created this descriptor and nothing else owns it. + held.push(unsafe { OwnedFd::from_raw_fd(number) }); + // SAFETY: live owned descriptor and integer-only fcntl arguments. + if unsafe { libc::fcntl(number, libc::F_SETFD, libc::FD_CLOEXEC) } < 0 { + return Err(last_error()); + } + } + Ok(held) +} + +/// Reject a spec whose extra descriptors this platform cannot place. +pub(super) fn checked_floor(numbers: impl Iterator) -> Result { + let mut floor = 0; + for number in numbers { + let number = RawFd::try_from(number).map_err(|_| Error::new(ErrorKind::InvalidInput))?; + floor = floor.max(number); + } + floor + .checked_add(1) + .ok_or_else(|| Error::new(ErrorKind::InvalidInput)) +} diff --git a/crates/turnloop/src/backend/unix.rs b/crates/turnloop/src/backend/unix.rs index b2627f0..fa91004 100644 --- a/crates/turnloop/src/backend/unix.rs +++ b/crates/turnloop/src/backend/unix.rs @@ -15,7 +15,7 @@ use crate::{ use std::{ collections::VecDeque, net::SocketAddr, - os::fd::{AsRawFd, OwnedFd}, + os::fd::{AsRawFd, OwnedFd, RawFd}, sync::Arc, time::Duration, }; @@ -329,7 +329,13 @@ unsafe impl Backend for Unix { fn kill(&mut self, h: Handle, signal: Signal, group: bool) -> Result<()> { self.services.kill(h, signal, group) } - fn spawn(&mut self, h: Handle, pipes: [Option; 3], spec: &ProcessSpec) -> Result { + fn spawn( + &mut self, + h: Handle, + pipes: [Option; 3], + extra: &[Option], + spec: &ProcessSpec, + ) -> Result { use std::os::unix::process::CommandExt; use std::process::{Command, Stdio as ChildStdio}; // Explicit SIG_IGN/NOCLDWAIT would auto-reap children behind our ownership. @@ -343,6 +349,10 @@ unsafe impl Backend for Unix { if action.sa_sigaction == libc::SIG_IGN || action.sa_flags & libc::SA_NOCLDWAIT != 0 { return Err(Error::new(ErrorKind::InvalidInput)); } + if spec.controlling_terminal && !spec.detached { + // TIOCSCTTY succeeds only for a session leader without a terminal. + return Err(Error::new(ErrorKind::InvalidInput)); + } let mut command = Command::new(&spec.program); command.args(&spec.args); if spec.env_clear { @@ -358,19 +368,7 @@ unsafe impl Backend for Unix { if let Some(gid) = spec.gid { command.gid(gid); } - if spec.detached { - // SAFETY: the child hook only calls async-signal-safe setsid and - // constructs an OS error without allocation; it captures no state. - unsafe { - command.pre_exec(|| { - if libc::setsid() < 0 { - Err(std::io::Error::last_os_error()) - } else { - Ok(()) - } - }); - } - } else if spec.new_process_group { + if !spec.detached && spec.new_process_group { command.process_group(0); } for (i, stdio) in spec.stdio.iter().enumerate() { @@ -398,7 +396,76 @@ unsafe impl Backend for Unix { } } } - let mut child = command.spawn().map_err(Error::from)?; + // Child ends of the extra descriptors, still at whatever numbers the OS + // gave them. They are relocated above every target number first, so the + // child hook's dup2 sequence cannot overwrite a source it has not used. + let mut sources = Vec::with_capacity(spec.extra.len()); + let mut parent_ends = Vec::with_capacity(spec.extra.len()); + let floor = super::process::checked_floor(spec.extra.iter().map(|fd| fd.number))?; + for fd in &spec.extra { + let (child, parent) = match fd.source { + ChildFdSource::Null => (super::process::null()?, None), + ChildFdSource::Pipe => { + let (parent, child) = super::process::one_way()?; + (child, Some((parent, false))) + } + ChildFdSource::Duplex => { + let (parent, child) = super::process::stream_pair()?; + (child, Some((parent, true))) + } + ChildFdSource::Handle(source) => ( + self.get(source)? + .transport + .fd + .try_clone() + .map_err(Error::from)?, + None, + ), + }; + sources.push((super::process::lift(child, floor)?, fd.number as RawFd)); + parent_ends.push(parent); + } + let dups: Vec<(RawFd, RawFd)> = sources + .iter() + .map(|(child, target)| (child.as_raw_fd(), *target)) + .collect(); + let reserved = match sources.first() { + Some((donor, _)) => super::process::reserve( + donor.as_raw_fd(), + spec.extra.iter().map(|fd| fd.number as RawFd), + )?, + None => Vec::new(), + }; + let (detached, terminal) = (spec.detached, spec.controlling_terminal); + if detached || terminal || !dups.is_empty() { + // SAFETY: the child hook calls only async-signal-safe setsid, ioctl + // and dup2, reads a captured plain-integer list without allocating, + // and builds an OS error from errno alone. std runs it after the + // standard streams are in place and immediately before exec, so + // descriptor 0 is already the child's stdin and every source is + // above every target. + unsafe { + command.pre_exec(move || { + if detached && libc::setsid() < 0 { + return Err(std::io::Error::last_os_error()); + } + if terminal && libc::ioctl(0, libc::TIOCSCTTY as _, 0) < 0 { + return Err(std::io::Error::last_os_error()); + } + for &(source, target) in &dups { + // dup2 leaves the new descriptor without FD_CLOEXEC, so + // it is exactly this number that survives exec. + if libc::dup2(source, target) < 0 { + return Err(std::io::Error::last_os_error()); + } + } + Ok(()) + }); + } + } + let spawned = command.spawn(); + drop((sources, reserved)); + let mut child = spawned.map_err(Error::from)?; let pid = child.id(); let stdio: [Option; 3] = [ child.stdin.take().map(Into::into), @@ -418,10 +485,26 @@ unsafe impl Backend for Unix { self.install(handle, transport, None)?; } } + for (handle, end) in extra.iter().zip(parent_ends.drain(..)) { + if let (Some(handle), Some((fd, stream))) = (handle, end) { + let transport = if stream { + // A socket pair's own end, exactly as an accepted local + // connection is adopted: configured, not re-classified. + socket::configure(fd.as_raw_fd())?; + Detached::new(fd, Kind::Pipe) + } else { + super::ipc::classify(fd)? + }; + self.install(*handle, transport, None)?; + } + } Ok(pid) })(); if result.is_err() { - for handle in std::iter::once(h).chain(pipes.into_iter().flatten()) { + for handle in std::iter::once(h) + .chain(pipes.into_iter().flatten()) + .chain(extra.iter().flatten().copied()) + { self.release(handle); } } diff --git a/crates/turnloop/src/driver.rs b/crates/turnloop/src/driver.rs index 1a1f997..885536a 100644 --- a/crates/turnloop/src/driver.rs +++ b/crates/turnloop/src/driver.rs @@ -540,34 +540,51 @@ impl Driver { /// Spawn a child and submit its exactly-once exit operation. Closing a live /// child initiates termination, then waits through ordinary turns for reaping /// before Cancelled and Closed; a new process group enables `kill_group`. + /// + /// A spec carrying [`ProcessSpec::extra`] descriptors is `InvalidInput` here, + /// because their parent ends have nowhere to be reported; use + /// [`Driver::spawn_extra`]. pub fn spawn(&mut self, spec: &ProcessSpec, token: Token) -> Result { + self.spawn_extra(spec, token, &mut []) + } + /// Spawn a child that also receives [`ProcessSpec::extra`] descriptors. + /// + /// `parents` must have exactly one slot per [`ChildFd`], in the same order. + /// Each slot receives the parent end of that descriptor: a readable handle + /// for [`ChildFdSource::Pipe`], a readable and writable one for + /// [`ChildFdSource::Duplex`], and `None` for a source that has no parent end. + /// Every returned handle belongs to this loop and is closed like any other. + /// + /// Descriptor numbers are the child's own, run from 3 to 255, and must not + /// repeat. The child's view of them is a plain descriptor number on Unix and + /// a C run-time descriptor on Windows, so a `NODE_CHANNEL_FD`-style handoff + /// is the same environment entry on both: set it in [`ProcessSpec::env`]. + /// + /// On failure nothing is created: no child, no handle, no completion. + pub fn spawn_extra( + &mut self, + spec: &ProcessSpec, + token: Token, + parents: &mut [Option], + ) -> Result { + if parents.len() != spec.extra.len() { + return Err(Error::new(ErrorKind::InvalidInput)); + } + for (i, fd) in spec.extra.iter().enumerate() { + if !(3..=MAX_CHILD_FD).contains(&fd.number) + || spec.extra[..i].iter().any(|seen| seen.number == fd.number) + { + return Err(Error::new(ErrorKind::InvalidInput)); + } + } + if spec.controlling_terminal && !spec.detached { + // TIOCSCTTY only succeeds for a session leader without a terminal. + return Err(Error::new(ErrorKind::InvalidInput)); + } + parents.fill(None); let h = self.new_handle(Kind::Socket)?; let mut pipes = [None; 3]; - let result = (|| { - for (i, stdio) in spec.stdio.iter().enumerate() { - if let ProcessStdio::Handle(source) = stdio { - self.resource(*source)?; - } - if *stdio == ProcessStdio::Pipe { - pipes[i] = Some(self.new_handle(Kind::Socket)?); - } - } - // Reserve the terminal completion before creating an OS child. - let op = self.new_op(Some(h), token)?; - let result = self.backend.spawn(h, pipes, spec).and_then(|pid| { - self.backend.submit(Request { - op, - handle: h, - operation: Operation::ProcessExit, - })?; - Ok(pid) - }); - if result.is_err() { - self.retire(op); - self.outstanding -= 1; - } - result - })(); + let result = self.spawn_inner(h, spec, token, &mut pipes, parents); match result { Ok(pid) => Ok(Process { handle: h, @@ -577,16 +594,66 @@ impl Driver { stderr: pipes[2], }), Err(e) => { - for handle in std::iter::once(h).chain(pipes.into_iter().flatten()) { + for handle in std::iter::once(h) + .chain(pipes.into_iter().flatten()) + .chain(parents.iter().flatten().copied()) + { self.backend.release(handle); if self.handles.remove(handle.key).is_some() { self.refs -= 1; } } + parents.fill(None); Err(e) } } } + fn spawn_inner( + &mut self, + h: Handle, + spec: &ProcessSpec, + token: Token, + pipes: &mut [Option; 3], + parents: &mut [Option], + ) -> Result { + for (i, stdio) in spec.stdio.iter().enumerate() { + if let ProcessStdio::Handle(source) = stdio { + self.resource(*source)?; + } + if *stdio == ProcessStdio::Pipe { + pipes[i] = Some(self.new_handle(Kind::Socket)?); + } + } + for (slot, fd) in parents.iter_mut().zip(&spec.extra) { + match fd.source { + ChildFdSource::Handle(source) => { + self.resource(source)?; + } + ChildFdSource::Pipe | ChildFdSource::Duplex => { + *slot = Some(self.new_handle(Kind::Socket)?); + } + ChildFdSource::Null => {} + } + } + // Reserve the terminal completion before creating an OS child. + let op = self.new_op(Some(h), token)?; + let result = self + .backend + .spawn(h, *pipes, parents, spec) + .and_then(|pid| { + self.backend.submit(Request { + op, + handle: h, + operation: Operation::ProcessExit, + })?; + Ok(pid) + }); + if result.is_err() { + self.retire(op); + self.outstanding -= 1; + } + result + } /// Signal a child still owned by this loop. pub fn kill(&mut self, process: Handle, signal: Signal) -> Result<()> { self.resource(process)?; diff --git a/crates/turnloop/src/native.rs b/crates/turnloop/src/native.rs index 13945b1..596028f 100644 --- a/crates/turnloop/src/native.rs +++ b/crates/turnloop/src/native.rs @@ -48,6 +48,13 @@ pub struct ProcessSpec { pub cwd: Option, /// Child stdin, stdout and stderr, in that order. pub stdio: [ProcessStdio; 3], + /// Additional child descriptors beyond stdin/stdout/stderr, at fixed child + /// descriptor numbers. This is the tail of Node's `stdio` array: entry + /// `stdio[3]` of `['pipe','pipe','pipe','ipc']` is [`ChildFd`] number 3. + /// Empty by default; [`Driver::spawn`](crate::Driver::spawn) refuses a + /// non-empty list, because the parent ends have nowhere to go. Use + /// [`Driver::spawn_extra`](crate::Driver::spawn_extra). + pub extra: Vec, /// Unix user ID; unsupported platforms reject this option. pub uid: Option, /// Unix group ID; unsupported platforms reject this option. @@ -64,6 +71,63 @@ pub struct ProcessSpec { /// This does not unref the child or change turnloop's explicit ownership: /// closing the child or dropping its owning loop still terminates a live child. pub detached: bool, + /// Make the child's stdin its controlling terminal (Unix `TIOCSCTTY`), which + /// is what a pty child needs so that job control, `/dev/tty` and terminal + /// signals work inside it. Requires `detached`, because only a session leader + /// with no controlling terminal may claim one; without it the spawn is + /// `InvalidInput`. Windows has no equivalent and reports `Unsupported`. + /// Defaults to false. + pub controlling_terminal: bool, +} + +/// Highest child descriptor number an extra [`ChildFd`] may use. +/// +/// Node's `stdio` array indexes the child's own descriptors, so the numbers in +/// practice are small; the bound keeps the Windows inherited-descriptor block, +/// which is dense from 0 to the highest number in use, a fixed small size. +pub const MAX_CHILD_FD: u32 = 255; + +/// One additional child descriptor beyond stdin, stdout and stderr. +/// +/// The number is the child's own descriptor number, not the parent's: it is the +/// index into Node's `stdio` array, and is what a child reads out of a variable +/// such as `NODE_CHANNEL_FD`. turnloop never invents that variable; set it in +/// [`ProcessSpec::env`] alongside the descriptor. +/// +/// On Windows the child descriptor is a C run-time descriptor, published in the +/// inherited-handle block every MSVCRT/UCRT program parses at startup, exactly +/// as libuv and Node publish theirs. A child that does not use the C run-time +/// sees the handle as inherited but has no number for it. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct ChildFd { + /// Child-side descriptor number, from 3 to 255. Numbers below 3 belong to + /// [`ProcessSpec::stdio`], and a repeated number is `InvalidInput`. + pub number: u32, + /// What to place at that number. + pub source: ChildFdSource, +} + +/// Child-side configuration of one extra descriptor. +/// +/// There is deliberately no `Inherit`: inheriting the host's own descriptor +/// number would mean clearing close-on-exec on a descriptor turnloop does not +/// own. Adopt it first ([`Detached::from_fd`](crate::Detached) on Unix) and pass +/// the resulting handle. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ChildFdSource { + /// Connect to the platform null device, opened for reading and writing. + Null, + /// Create a one-way pipe that the child writes and the parent reads. The + /// parent's readable end is returned. + Pipe, + /// Create a bidirectional stream pair. The parent's end is returned and is + /// both readable and writable: a socket pair on Unix, a duplex named-pipe + /// instance on Windows. This is what Node's `'pipe'` and `'ipc'` stdio + /// entries are, and the only kind that can carry an IPC channel. + Duplex, + /// Duplicate this loop-owned transport into the child. The loop keeps its + /// own end; the child receives an independent descriptor for it. + Handle(Handle), } impl ProcessSpec { /// Launch a program with inherited environment, directory and standard streams. @@ -80,6 +144,8 @@ impl ProcessSpec { new_process_group: false, windows_hide: false, detached: false, + controlling_terminal: false, + extra: Vec::new(), } } } @@ -98,6 +164,9 @@ pub struct Process { /// Readable parent end of child stderr, if requested. pub stderr: Option, } +// Parent ends of extra descriptors are reported through the caller's slice in +// `Driver::spawn_extra`, which keeps this identity Copy and the driver free of +// a per-spawn allocation of its own. /// Portable signal names. Unsupported mappings return an error. #[derive(Clone, Copy, Debug, Eq, PartialEq)] diff --git a/docs/lanes/procspec.md b/docs/lanes/procspec.md new file mode 100644 index 0000000..e076f47 --- /dev/null +++ b/docs/lanes/procspec.md @@ -0,0 +1,323 @@ +# ProcessSpec: extra descriptors and session control — issue #38 + +Base `6c41265`, branch `lane/procspec`, macOS arm64 + Linux x86_64 + Windows CI. + +Perry's P2 migration moved child stdout/stderr, dgram and signals onto turnloop, +but child spawn and exit could not move: `ProcessSpec.stdio` is exactly three +entries, so `child_process.fork()`'s IPC channel at descriptor 3, `spawn`'s +arbitrary `stdio` array and a pty child's controlling terminal had nowhere to go. + +## 1. Decision + +**Implement (1) extra descriptors and (2) session control. Do not implement (3) +`Loop::adopt_process`.** + +### Why extra descriptors and session control + +The three things a host cannot supply from outside a spawn are the descriptor +plan, the session, and the controlling terminal: all of them happen between +`fork` and `exec`, inside the window turnloop owns. Everything else Perry's +spawn path does — program resolution, shell selection, argv, environment +composition, the *value* of `NODE_CHANNEL_FD`, pty allocation and its termios — +is policy the host already owns and keeps owning. + +Turnloop owning the child's identity is also what makes the guarantees in the +issue true. Exactly-once completion, the kill/close ordering and grandchild +cleanup all rest on the loop holding the child from before it runs: on Windows +the Job Object has to be assigned while the child is still suspended, or its +descendants are not in the job at all. + +### Why not adoption + +`Loop::adopt_process(pid, …)` was evaluated and rejected, on four grounds. + +1. **It does not deliver the guarantee it would be adopted for.** On Windows a + process must be assigned to a job *before* it creates descendants; + `AssignProcessToJobObject` on a running process leaves its existing + grandchildren outside. So an adopted process has strictly weaker tree + cleanup than a spawned one, and `kill_group` would have to be `InvalidInput` + for it. The issue requires grandchild cleanup to be kept. +2. **The portable signature does not exist.** A pid is not a safe identity on + Windows: `OpenProcess` by pid races pid reuse, so an honest API takes an + `OwnedHandle` there and a pid on Unix. That is a platform-divergent public + API in a crate whose entire point is one contract on six backends. +3. **It answers a question the descriptor work already answers.** The reason + adoption was attractive was that Perry could keep its own `pre_exec` and fd + plan. The survey of Perry's spawn paths (below) shows that plan is exactly + `setsid`, `dup2` and `TIOCSCTTY` — all three are now `ProcessSpec` fields, + so there is nothing left for the host's hook to do. +4. **A mode that exists is a decision that has not been made.** Two ways to own + a child, with different reaping rules and different tree semantics, is a + configuration matrix nobody would exercise on every platform. + +**If it is ever added, this is the reaping rule it must satisfy**, and it is +worth recording because it is the part that is easy to get wrong. turnloop must +reap an adopted child with a *targeted* wait — `waitpid(pid, …)`, never +`waitpid(-1, …)`, `wait()`, or a negative process-group pid — because a wait for +any child consumes whichever child exits first, including another subsystem's. +turnloop already satisfies this for the children it spawns: +`ChildState::reap` calls `std::process::Child::try_wait`, which is +`waitpid(pid, WNOHANG)`, and the SIGCHLD subscription only *notifies*; it never +reaps. Adoption would additionally need the converse contract stated in the API: +adopting a pid transfers reaping ownership of that pid to the loop, and the host +must not wait for it any more. Perry can honour that — the survey found exactly +one raw `waitpid` in the whole workspace (`pty/native.rs:239`, targeted and +blocking) and every other reap goes through `std::process::Child`, which is +targeted too. `a_sibling_waiter_and_the_loop_keep_their_own_children` asserts +the half that exists today, in both orders. + +### Which pre-exec behaviour turnloop owns + +Perry's current child hooks are `setsid` (detached spawns, `child_process/options.rs:149` +and `registry.rs:224`), `dup2` of the IPC socket onto the channel descriptor +(`child_process/fork.rs:282`), `dup2` + `F_SETFD` for each extra stdio entry +(`child_process/options.rs:297`, `:314`, `:325`), and, in the pty path's raw +`fork`, `setsid` + `TIOCSCTTY` + `dup2` (`pty/native.rs:198–222`). uid/gid are +delegated to std. Nothing resets signal masks or dispositions, sets `umask`, or +closes descriptor ranges. + +turnloop now owns all of it. In the order the child executes them: + +| # | step | source | +|---|---|---| +| 1 | `dup2` of stdin, stdout, stderr | std, from `ProcessSpec::stdio` | +| 2 | `setgroups`/`setgid`/`setuid` | std, from `ProcessSpec::uid`/`gid` | +| 3 | `chdir` | std, from `ProcessSpec::cwd` | +| 4 | `setpgid(0, 0)` | std, from `ProcessSpec::new_process_group` | +| 5 | empty the signal mask, `SIGPIPE` back to `SIG_DFL` | std, unconditional | +| 6 | `setsid()` | turnloop, from `ProcessSpec::detached` | +| 7 | `ioctl(0, TIOCSCTTY, 0)` | turnloop, from `ProcessSpec::controlling_terminal` | +| 8 | `dup2` of each extra descriptor onto its number | turnloop, from `ProcessSpec::extra` | + +Steps 6–8 are one hook, so their order is fixed rather than emergent. std runs +`pre_exec` closures last, immediately before `exec`, which is what makes step 7 +correct: descriptor 0 is already the child's stdin by then, so a pty child +claims the terminal it was actually given. + +The host keeps: program resolution and shell policy, argv, environment +composition including the `NODE_CHANNEL_FD` name and value, `openpty` and its +termios, and the decision of which descriptor number is the channel. + +## 2. API + +```rust +pub const MAX_CHILD_FD: u32 = 255; + +pub struct ChildFd { + pub number: u32, // 3..=MAX_CHILD_FD, unique within a spec + pub source: ChildFdSource, +} + +pub enum ChildFdSource { + Null, // the platform null device, read/write + Pipe, // one way: the child writes, the parent reads + Duplex, // both ways; Node's 'pipe' and 'ipc' entries + Handle(Handle), // duplicate a transport this loop owns +} + +pub struct ProcessSpec { + // ... unchanged fields ... + pub extra: Vec, + pub controlling_terminal: bool, +} + +impl Loop { + pub fn spawn(&mut self, spec: &ProcessSpec, token: Token) -> Result; + pub fn spawn_extra( + &mut self, + spec: &ProcessSpec, + token: Token, + parents: &mut [Option], + ) -> Result; +} +``` + +`parents` has one slot per `ChildFd`, in the same order, and receives the parent +end of each: a readable handle for `Pipe`, a readable and writable one for +`Duplex`, `None` for `Null` and `Handle`. Keeping it a caller-supplied slice is +what keeps `Process` `Copy` and keeps the driver free of a per-spawn allocation +of its own; `spawn` is `spawn_extra` with an empty slice and refuses a spec that +has extras, because their parent ends would have nowhere to go. + +There is deliberately no `Inherit` source: inheriting the host's own descriptor +number means clearing close-on-exec on a descriptor turnloop does not own. Adopt +it (`Detached::from_fd` / `from_handle`, then `attach`) and pass the handle. + +`NODE_CHANNEL_FD` is not a turnloop concept. The host writes it into +`ProcessSpec::env` next to the `ChildFd` it names, on both platforms, with the +same value. + +### Rejected plans + +All of these are `InvalidInput`, and create nothing — no child, no handle, no +completion, and `parents` is left all `None`: + +- `parents.len() != spec.extra.len()` +- a number outside `3..=MAX_CHILD_FD`, or repeated within one spec +- `spawn` (rather than `spawn_extra`) with a non-empty `extra` +- `controlling_terminal` without `detached` +- `controlling_terminal` on Windows is `Unsupported` + +## 3. Per-platform behaviour + +| | Unix (epoll/kqueue) | Windows (IOCP) | WASI 0.2/0.3, web | +|---|---|---|---| +| how the child sees a number | the descriptor number itself | a C run-time descriptor, published in `STARTUPINFOW.lpReserved2` | `Unsupported` (no processes) | +| `Duplex` | `socketpair(AF_UNIX, SOCK_STREAM)`, as libuv creates every child pipe | duplex named-pipe instance: `PIPE_ACCESS_DUPLEX` for the overlapped parent end, `GENERIC_READ\|GENERIC_WRITE` synchronous for the child | — | +| `Pipe` | `pipe2(O_CLOEXEC)`, or `pipe` + `FD_CLOEXEC` on Darwin | `PIPE_ACCESS_INBOUND` named-pipe instance | — | +| `Null` | `/dev/null`, `O_RDWR\|O_CLOEXEC` | `NUL`, `GENERIC_READ\|GENERIC_WRITE` | — | +| `Handle(h)` | `F_DUPFD_CLOEXEC` of the loop's transport | `DuplicateHandle` with inheritance | — | +| placement | one `pre_exec` hook `dup2`s each source onto its number | the inherited-descriptor block plus `PROC_THREAD_ATTRIBUTE_HANDLE_LIST` | — | +| `controlling_terminal` | `setsid` then `TIOCSCTTY` on descriptor 0 | `Unsupported` | `Unsupported` | +| grandchild cleanup | process group, `kill_group` | Job Object, `TerminateJobObject` | — | + +**Unix placement.** Sources are relocated above the highest target number with +`F_DUPFD_CLOEXEC` *before* the fork, so the child hook can `dup2` them in any +order without an earlier target having overwritten a later source. `dup2` leaves +the new descriptor without `FD_CLOEXEC`, which is exactly what makes that number +survive `exec`. Adding a hook takes std off its `posix_spawn` fast path, as +`detached` already did. + +**Unix: the target numbers are reserved across the fork.** This one is a bug +that was found and fixed during the work, and it is worth stating because it is +invisible until it bites. `Command::spawn` creates its own exec-error pipe +*after* the standard streams and immediately before the fork, with no +relocation of its own (`sys::pipe::pipe()`, or a `SOCK_SEQPACKET` pair on +Linux), so it takes the lowest free descriptor numbers — which is precisely what +the sources vacate when they are lifted above the targets. If its write end +landed on a target number, the child hook would `dup2` over it, the parent's +read would see end-of-file, and **a failed `exec` would be reported as a +successful spawn**. turnloop therefore holds every *free* target number in the +parent, with a close-on-exec duplicate, from before `Command::spawn` until the +child exists; a number the parent already uses cannot be handed to std either, +so those need nothing. `a_failed_exec_is_reported_even_at_the_lowest_free_descriptor_numbers` +probes the two lowest free numbers, releases them, uses them as the targets and +requires `NotFound`. With the reservation removed it fails, which is what makes +it a test of the reservation rather than of a coincidence. + +**Windows placement.** The inherited-descriptor block is the C run-time's own +format — a descriptor count, one flag byte per descriptor, then one handle per +descriptor, packed without padding — and it is how libuv and Node give a child +numbered descriptors at all. Flags come from `GetFileType`: `FOPEN`, plus +`FPIPE` for a pipe and `FDEV` for a character device. Numbers below the highest +one in use but not claimed are present and closed (`INVALID_HANDLE_VALUE`). +Because this is Node's own convention, `NODE_CHANNEL_FD=3` names the same thing +in a child here as it does under Node, and `_get_osfhandle(3)` in the child +returns the inherited handle. A child that does not use the C run-time inherits +the handles but has no number for them; that is the same limitation libuv has. + +**Windows `Handle(h)` caveat.** A transport the loop created itself — a named +pipe or a socket — is overlapped, and a duplicate of it in the child is +overlapped too. A child that reads it with plain +blocking calls will not work; it must use overlapped I/O, or the host must pass +a synchronous handle it adopted (`Detached::from_handle` on a `CreatePipe` end, +which is what the contract test does). libuv's `UV_INHERIT_STREAM` has the same +property. On Unix the equivalent caveat is the status flags: an adopted +transport is non-blocking, and the child's duplicate shares that file +description. + +**`Pipe` is a real pipe, not a half-shut socket pair.** Darwin refuses +`shutdown(SHUT_RD)` on a `socketpair` end with `ENOTCONN` (verified directly), +so the direction has to come from the object rather than from a later call. + +## 4. Tests + +`crates/turnloop-contract/tests/process_fds.rs`, all running on Unix and +Windows unless noted: + +- `five_descriptors_carry_bytes_both_ways_through_a_channel_handoff` — the + issue's headline case. A child with descriptors 0–4, where 3 and 4 are + `Duplex`; the parent writes `ping-3`/`ping-4` and the child answers + `pong-3`/`pong-4` on the same descriptors. The child finds descriptor 3 by + reading `NODE_CHANNEL_FD` out of its environment and 4 out of a second + variable, so the handoff is proven rather than hard-coded. Exit status, + write completions, and one `Closed` per handle are all asserted. +- `extra_sources_cover_one_way_pipes_the_null_device_and_adopted_transports` — + `Pipe` at 3, `Null` at 4, `Handle` at 5 where the handle is an OS pipe the + test created outside the loop and attached. The child writes to 3 and 5, and + reads 4 to end-of-file, which is what the null device must do. +- `close_orders_cancel_before_closed_for_a_child_holding_extra_descriptors` — + closing a live child with two extra descriptors yields exactly one + `Cancelled` then one `Closed`, in that order per handle (the rig records + terminal results in arrival order and compares the sequence, rather than only + counting them), with `op` present on the cancellation and absent on the + `Closed`. No exit is reported for the cancelled watch, the child is reaped + (`ECHILD` on Unix), the extras close independently, and a further turn + produces nothing. +- `rejected_descriptor_plans_create_nothing` — every rejection above, each + asserting `!driver.alive()` and an untouched `parents` slice afterwards. +- `a_failed_exec_is_reported_even_at_the_lowest_free_descriptor_numbers` (Unix) + — the descriptor-reservation regression test described in §3. +- `a_sibling_waiter_and_the_loop_keep_their_own_children` — a plain + `std::process::Command` child exiting 7 (then 11) beside loop-owned children + exiting 23, in both orders: the loop reports its own child's status while the + sibling is unreaped, and a host reap does not consume the loop's child. +- `a_process_group_with_extra_descriptors_still_kills_its_grandchild` — the + leader reports its grandchild's identity, the test pins it, `kill_group` + terminates the tree, and the grandchild is polled until gone. +- `a_detached_child_can_claim_its_stdin_as_a_controlling_terminal` (Unix) — + `openpty`, the slave attached and passed as stdin, `detached` + + `controlling_terminal`; the child reports that it leads its own group and + session, that `/dev/tty` opens, and that it is the terminal's foreground + group. + +`crates/turnloop-contract/tests/allocations.rs`: + +- `extra_child_descriptor_traffic_allocates_nothing_after_spawn` — eight + children with two extra `Duplex` descriptors each. Spawning is charged to + setup, as every other resource creation is; the counted window covers the + writes, the reads, the exits and the closes, and must be zero. The gate + asserts its subject ran (16 writes, 8 exits, 32 closes) rather than merely + that nothing threw, and it was sabotage-checked: a single planted + `vec![0u8; 8]` inside the window fails it. + +The fixture (`crates/turnloop-contract/src/bin/native_child.rs`) gains +`channel`, `extra-sources`, `exit-with` and `tty-session` modes. Its descriptor +I/O goes through the number on Unix and through `_get_osfhandle` on Windows, +which is exactly how `uv_pipe_open` turns Node's `NODE_CHANNEL_FD` into a +stream, so the Windows test exercises the convention rather than a turnloop +private path. + +## 5. Commands + +| command | result | +|---|---| +| `cargo fmt --all -- --check` | PASS | +| `cargo clippy --workspace --all-targets -- -D warnings -D clippy::undocumented_unsafe_blocks` (macOS arm64) | PASS | +| `cargo clippy -p turnloop -p turnloop-contract -p turnloop-io --all-targets --target x86_64-unknown-linux-gnu -- -D warnings -D clippy::undocumented_unsafe_blocks` | PASS | +| `cargo clippy … --target x86_64-pc-windows-msvc …` | PASS | +| `cargo clippy … --target wasm32-wasip2 …` | PASS | +| `cargo clippy … --target wasm32-unknown-unknown …` (web) | PASS | +| `cargo clippy --workspace --all-targets --target wasm32-wasip2 --all-features …` | UNRUN locally (no wasm C toolchain for `ring` on this host); run by CI `lint-wasm` — PASS | +| `cargo clippy … --target wasm32-wasip3 …` (nightly-2026-09-07) | UNRUN locally (toolchain not installed); run by CI `wasi (wasm32-wasip3)` and `protocol-wasi (wasm32-wasip3)` — PASS | +| `cargo test -p turnloop-contract --test process_fds -- --test-threads=1` (macOS arm64) | PASS (8/8) | +| `cargo test -p turnloop-contract --test allocations extra_child_descriptor -- --test-threads=1` (macOS arm64) | PASS | +| `cargo test --workspace --no-fail-fast -- --test-threads=1` (macOS arm64) | PASS | +| `cargo test --workspace --no-fail-fast -- --test-threads=1` (Linux x86_64, build box) | PASS for everything in this lane. One pre-existing, unrelated failure: `filesystem::permission_denied_is_reported`, because that box runs as root and root ignores the read-only mode bit; CI's Linux arms, which are not root, pass it. | +| `cargo test -p turnloop-contract --test process_fds --test allocations -- --test-threads=1` (Linux x86_64, build box, final commit) | PASS (8/8 and 16/16) | +| `bash scripts/ci/no-tokio.sh` | PASS (15 policy rows) | +| `python3 scripts/ci/soak.py` | PASS (251 locked versions, 1 active security exception) | +| `bash scripts/ci/install-wasmtime.sh` then `python3 scripts/ci/run-tests.py wasi --target wasm32-wasip2` | PASS (11 + 43 + 43 + 13 tests) | +| CI on `lane/procspec`, run `35013478648` (all four OS arms, Windows x86_64 runtime in three feature modes) | PASS | +| CI on `lane/procspec`, final run `35015001253` on `e135f79` — lint and tests on Linux x86_64 and aarch64, macOS arm64 and Windows x86_64 (each in the default, executor and all-features modes), both WASM lints, both WASI contracts, loom, Miri, the instruction gate and the protocol suites | PASS | + +### Sabotage checks + +Two gates were shown to fail when their subject is removed, rather than being +assumed to work: + +- the allocation gate, with one planted `vec![0u8; 8]` inside the counted window; +- the descriptor reservation, by deleting it and watching + `a_failed_exec_is_reported_even_at_the_lowest_free_descriptor_numbers` report + a successful spawn of a program that does not exist. + +### Not covered + +- `ChildFdSource::Handle` of an overlapped loop transport on Windows: the child + must drive it with overlapped I/O. The contract test passes a synchronous + adopted handle instead, which is the shape a host actually wants. +- Descriptor passing (`SCM_RIGHTS`) *over* an extra `Duplex` descriptor. The + transport is an `AF_UNIX` stream, so `send_handle`/`recv_handle` apply to it + like any other loop pipe, but no test exercises that combination yet; it is + what `cluster.fork()` will need. +- No `adopt_process`, by decision (§1).