diff --git a/docs/v2/specs/std-process.md b/docs/v2/specs/std-process.md index 1a13273..c78e40f 100644 --- a/docs/v2/specs/std-process.md +++ b/docs/v2/specs/std-process.md @@ -1,43 +1,71 @@ # std/process Spec -Spawn a subprocess, run it to completion, and read its exit code plus -captured stdout and stderr. Optionally feed data to the child's stdin. The -primitives bind the raven-runtime C ABI (backed by `std::process::Command`); -the wrappers add the Result/Error model and the `Output` value type in pure +Spawn a subprocess and either run it to completion (`run`) or keep it +running while its output streams in (`start`). The primitives bind the +raven-runtime C ABI (backed by `std::process::Command`); the wrappers add +the Result/Error model and the `Output` and `Child` value types in pure Raven. ## Import ```rust -import std/process { run, run_with_input } +import std/process { run, run_with_input, start, Child } ``` -The `success` method on `Output` comes in with the type and needs no -separate selector. +The methods on `Output` and `Child` come in with the types and need no +separate selectors. ## Surface ```rust struct Output { code: Int, stdout: String, stderr: String } +struct Child { id: Int } fun run(program: String, args: List) -> Result fun run_with_input(program: String, args: List, input: String) -> Result +fun start(program: String, args: List) -> Result impl Output { fun success(self) -> Bool // code == 0 } + +impl Child { + fun read_stdout(self) -> String // drained since the last read, "" when nothing new + fun read_stderr(self) -> String + fun poll(self) -> Option // None while running, Some(code) once exited + fun running(self) -> Bool + fun wait(self) -> Int // block (polling) until exit + fun kill(self) -> Bool + fun free(self) // drop the runtime entry; does not kill +} ``` `run` spawns `program` with `args`, feeds it no stdin, waits for it to finish, and returns its captured output. `run_with_input` is the same but -writes `input` to the child's stdin (which is then closed). - -## Run-to-completion model - -There is no streaming in v2.0. A run executes the child to completion in a -single runtime call that captures the whole of stdout and stderr into an -owned result before returning. A caller that needs incremental output is -not served by this surface. +writes `input` to the child's stdin (which is then closed). `start` spawns +the child and returns immediately with a handle. (`spawn` is the goroutine +keyword, hence `start`.) + +## Two capture models + +A `run` executes the child to completion in a single runtime call that +captures the whole of stdout and stderr into an owned result before +returning. + +A `start` leaves the child running: runtime reader threads pump its stdout +and stderr into buffers as they arrive, and each `read_stdout` / +`read_stderr` call drains what accumulated since the previous read. Each +buffer is capped at 8 MB: a child that outruns its reader keeps only the +most recent bytes, so an undrained pipe cannot grow memory without bound +(keep draining to see everything). `poll` try-waits without blocking +(`running` is its Bool shorthand); `wait` loops poll with a short +scheduler-friendly sleep so other goroutines proceed. `kill` terminates the +child (killing an already-exited child reports success). `free` drops the +registry entry without killing, so a caller that wants the child stopped +must `kill` first; reaping is still guaranteed, since a freed child whose +exit was not yet observed is handed to a background waiter that collects it +when it exits, never leaving a zombie. The child's stdin is null: it sees +end of input immediately and does not inherit the parent's terminal. ## Argument encoding (NUL-joined) @@ -57,6 +85,12 @@ wrappers read the three fields by id through the extractors, then free the entry, so a caller never sees the id. Ids start at 1; an id of 0 is the spawn-failure sentinel paired with a set last-error. +Started children live in a second registry sharing the same id sequence +(so an id is unique across both): the entry holds the live process handle, +the two output buffers the reader threads fill, and the exit code once +observed, cached so polling after exit stays cheap. `poll` reports `-2` +over the FFI while the child runs; the wrapper maps that to `None`. + ## Non-zero exit is not an error A child that spawns and runs but exits with a non-zero code is a normal, diff --git a/examples/v2/spawn_child.rv b/examples/v2/spawn_child.rv new file mode 100644 index 0000000..21bfdfb --- /dev/null +++ b/examples/v2/spawn_child.rv @@ -0,0 +1,27 @@ +// golden:skip - the child half of the std/process spawn smoke tests. Prints +// "early" immediately, idles long enough for the parent to observe it +// running (and to drain the first line while it still runs), then prints +// "late" and exits 3. The idle defaults to 600ms; the parent's kill test +// passes a large one as the first argument to make the child effectively +// hang until killed. +import std/io { println } +import std/string +import std/sync { sleep_millis } +import std/env { args, exit } + +fun main() { + println("early") + let idle = 600 + let a = args() + if a.len() > 1 { + match a[1].parse_int() { + Some(ms) -> { + idle = ms + }, + None -> {}, + } + } + sleep_millis(idle) + println("late") + exit(3) +} diff --git a/examples/v2/spawn_parent.rv b/examples/v2/spawn_parent.rv new file mode 100644 index 0000000..26dde2f --- /dev/null +++ b/examples/v2/spawn_parent.rv @@ -0,0 +1,85 @@ +// golden:skip - std/process spawn parent, checked in codegen_smoke.rs +// (process_spawn_streams_and_kills). Drives the whole streaming surface +// against the child in RAVEN_PROC_CHILD (spawn_child.rv): +// +// 1. Streaming: start the child, drain stdout until "early" arrives, and +// prove the child is still running at that moment ("streamed-early"). +// Then wait for exit and print the code (3) and whether the full output +// also carried "late" ("got-late"). +// 2. Kill: start the child with a 20s idle, kill it while it runs, and +// print "killed" when the exit code comes back non-zero. +// 3. Spawn failure: a nonexistent program takes the Err path. +// +// Expected stdout: +// streamed-early +// 3 +// got-late +// killed +// spawn failed +import std/io { println } +import std/process { Child, start } +import std/sync { sleep_millis } +import std/env { get_env } + +fun main() { + let child_path = get_env("RAVEN_PROC_CHILD") + + // 1. Streaming while the child runs. + let none: List = [] + match start(child_path, none) { + Ok(c) -> { + let seen = "" + let live = false + let tries = 0 + while !seen.contains("early") && tries < 400 { + seen = seen.concat(c.read_stdout()) + if seen.contains("early") && c.running() { + live = true + } + sleep_millis(10) + tries = tries + 1 + } + if live { + println("streamed-early") + } + let code = c.wait() + println("${code}") + seen = seen.concat(c.read_stdout()) + if seen.contains("late") { + println("got-late") + } + c.free() + }, + Err(e) -> { + println("spawn failed early: ${e.message()}") + }, + } + + // 2. Kill a long-running child (a 20s idle passed as its argument). + match start(child_path, ["20000"]) { + Ok(c) -> { + // Let it start, then stop it. + sleep_millis(100) + c.kill() + let code = c.wait() + if code != 0 { + println("killed") + } + c.free() + }, + Err(e) -> { + println("spawn failed kill: ${e.message()}") + }, + } + + // 3. The spawn-failure Err path. + match start("definitely-not-a-real-program-12345", none) { + Ok(c) -> { + println("unexpectedly spawned") + c.free() + }, + Err(e) -> { + println("spawn failed") + }, + } +} diff --git a/raven-runtime/src/lib.rs b/raven-runtime/src/lib.rs index 4e85800..efa6b60 100644 --- a/raven-runtime/src/lib.rs +++ b/raven-runtime/src/lib.rs @@ -57,7 +57,7 @@ use std::net::{TcpListener, TcpStream, ToSocketAddrs}; use std::process; use std::slice; use std::sync::atomic::{AtomicI64, Ordering}; -use std::sync::{Mutex, OnceLock}; +use std::sync::{Arc, Mutex, OnceLock}; use std::time::Duration; /// Ignore `SIGPIPE` once, process-wide, on Unix. Writing to a pipe or socket @@ -210,6 +210,63 @@ fn process_next_id() -> i64 { NEXT_ID.fetch_add(1, Ordering::Relaxed) } +/// A live spawned child. Unlike `ProcResult` (a finished child's capture), +/// this holds the running process: reader threads append its stdout and +/// stderr to the shared buffers as they arrive, `raven_process_poll` +/// try-waits it, and `raven_process_kill` terminates it. `code` caches the +/// exit status once observed so poll stays cheap after exit. +struct SpawnEntry { + child: process::Child, + stdout: Arc>>, + stderr: Arc>>, + code: Option, +} + +/// The process-wide spawned-child registry. Ids come from `process_next_id` +/// (shared with the finished-child registry, so an id is unique across +/// both); 0 stays the spawn-failure sentinel. +fn spawn_registry() -> &'static Mutex> { + static REGISTRY: OnceLock>> = OnceLock::new(); + REGISTRY.get_or_init(|| Mutex::new(HashMap::new())) +} + +/// Drain a spawned child's accumulated output buffer into a fresh Vec. +fn spawn_drain(buf: &Arc>>) -> Vec { + let mut guard = buf.lock().unwrap(); + std::mem::take(&mut *guard) +} + +/// Cap on each spawned child's accumulated output buffer. A child that +/// outruns its reader keeps only the most recent bytes; the oldest are +/// dropped, so an undrained pipe cannot grow memory without bound. +const SPAWN_BUFFER_MAX: usize = 8 * 1024 * 1024; + +/// Append pumped bytes to a spawned child's buffer, dropping the oldest +/// bytes past `SPAWN_BUFFER_MAX`. +fn spawn_append(buf: &Arc>>, bytes: &[u8]) { + let mut guard = buf.lock().unwrap(); + guard.extend_from_slice(bytes); + if guard.len() > SPAWN_BUFFER_MAX { + let excess = guard.len() - SPAWN_BUFFER_MAX; + guard.drain(..excess); + } +} + +/// Pump one child pipe into a shared buffer until end of stream. Runs on a +/// plain OS thread outside the collector: it only touches the Arc'd buffer, +/// never a GC object. +fn spawn_pump(mut pipe: R, buf: Arc>>) { + std::thread::spawn(move || { + let mut chunk = [0u8; 8192]; + loop { + match pipe.read(&mut chunk) { + Ok(0) | Err(_) => break, + Ok(n) => spawn_append(&buf, &chunk[..n]), + } + } + }); +} + /// The process-wide compiled-regex registry keyed by an incrementing id. /// Ids start at 1; 0 is the failure sentinel that pairs with a set /// last-error. @@ -2459,6 +2516,165 @@ pub extern "C" fn raven_process_free(id: i64) { process_registry().lock().unwrap().remove(&id); } +/// Spawn `program` with `args_nul_joined` (same encoding as +/// `raven_process_run`) WITHOUT waiting: the child runs with a null stdin +/// while reader threads pump its stdout and stderr into buffers that +/// `raven_process_read_stdout`/`_stderr` drain incrementally. Returns the +/// spawn id, or 0 on a spawn failure with the last-error slot set. +/// +/// # Safety +/// +/// `program` and `args_nul_joined` must be valid +/// `raven_string_from_bytes`-built `String`s. +#[no_mangle] +pub extern "C" fn raven_process_spawn( + program: *const object::String, + args_nul_joined: *const object::String, +) -> i64 { + ignore_sigpipe(); + let (Some(program), Some(args_joined)) = (env_name(program), env_name(args_nul_joined)) else { + process_set_error("program or args is not valid UTF-8".to_string()); + return 0; + }; + let args: Vec<&str> = if args_joined.is_empty() { + Vec::new() + } else { + args_joined.split('\0').skip(1).collect() + }; + + let mut command = process::Command::new(program); + command.args(&args); + command.stdin(process::Stdio::null()); + command.stdout(process::Stdio::piped()); + command.stderr(process::Stdio::piped()); + + let mut child = match command.spawn() { + Ok(c) => c, + Err(e) => { + process_set_error(e.to_string()); + return 0; + } + }; + + let stdout = Arc::new(Mutex::new(Vec::new())); + let stderr = Arc::new(Mutex::new(Vec::new())); + if let Some(pipe) = child.stdout.take() { + spawn_pump(pipe, Arc::clone(&stdout)); + } + if let Some(pipe) = child.stderr.take() { + spawn_pump(pipe, Arc::clone(&stderr)); + } + + let id = process_next_id(); + spawn_registry().lock().unwrap().insert( + id, + SpawnEntry { + child, + stdout, + stderr, + code: None, + }, + ); + process_clear_error(); + id +} + +/// Whatever the spawned child `id` wrote to stdout since the last read, +/// drained. Empty when nothing new arrived or the id is unknown. +#[no_mangle] +pub extern "C" fn raven_process_read_stdout(id: i64) -> *mut object::String { + // Copy out before allocating; the allocation may collect and must not + // happen under a registry lock (see `raven_http_status_text`). + let out: Vec = { + let registry = spawn_registry().lock().unwrap(); + registry + .get(&id) + .map(|e| spawn_drain(&e.stdout)) + .unwrap_or_default() + }; + object::raven_string_from_bytes(out.as_ptr(), out.len()) +} + +/// Whatever the spawned child `id` wrote to stderr since the last read, +/// drained. Empty when nothing new arrived or the id is unknown. +#[no_mangle] +pub extern "C" fn raven_process_read_stderr(id: i64) -> *mut object::String { + let err: Vec = { + let registry = spawn_registry().lock().unwrap(); + registry + .get(&id) + .map(|e| spawn_drain(&e.stderr)) + .unwrap_or_default() + }; + object::raven_string_from_bytes(err.as_ptr(), err.len()) +} + +/// Poll the spawned child `id` without blocking: -2 while it is still +/// running, its exit code once it has exited (a child terminated by a +/// signal with no code maps to -1), and -1 for an unknown id. The observed +/// code is cached so polling after exit stays cheap. +#[no_mangle] +pub extern "C" fn raven_process_poll(id: i64) -> i64 { + let mut registry = spawn_registry().lock().unwrap(); + let Some(entry) = registry.get_mut(&id) else { + return -1; + }; + if let Some(code) = entry.code { + return code; + } + match entry.child.try_wait() { + Ok(Some(status)) => { + let code = status.code().map(|c| c as i64).unwrap_or(-1); + entry.code = Some(code); + code + } + Ok(None) => -2, + Err(_) => -1, + } +} + +/// Kill the spawned child `id`. Returns 1 when the signal was delivered (or +/// the child had already exited), 0 for an unknown id or a delivery +/// failure. The reader threads finish on their own as the pipes close. +#[no_mangle] +pub extern "C" fn raven_process_kill(id: i64) -> i64 { + let mut registry = spawn_registry().lock().unwrap(); + let Some(entry) = registry.get_mut(&id) else { + return 0; + }; + if entry.code.is_some() { + return 1; + } + match entry.child.kill() { + Ok(()) => 1, + // On both Unix and Windows killing an already-exited child reports + // an error; that is success for the caller's purposes. + Err(_) => match entry.child.try_wait() { + Ok(Some(_)) => 1, + _ => 0, + }, + } +} + +/// Drop the spawned child `id` from the registry. The child is NOT killed: +/// freeing a running child leaves it running detached (kill first to stop +/// it). Reaping is still guaranteed: a child whose exit was already +/// observed by poll has been reaped by that try_wait, and any other child +/// is handed to a background waiter thread that collects it when it exits, +/// so a freed child never lingers as a zombie. An unknown id is a no-op. +#[no_mangle] +pub extern "C" fn raven_process_spawn_free(id: i64) { + let entry = spawn_registry().lock().unwrap().remove(&id); + if let Some(mut entry) = entry { + if entry.code.is_some() { + return; + } + std::thread::spawn(move || { + let _ = entry.child.wait(); + }); + } +} + #[cfg(test)] mod tests { use super::*; @@ -2470,6 +2686,20 @@ mod tests { assert_eq!(size_of::(), 16); } + #[test] + fn spawn_buffer_caps_at_max_keeping_the_tail() { + // A child that outruns its reader must not grow memory without + // bound: the buffer holds at most SPAWN_BUFFER_MAX bytes and the + // oldest are dropped, so the newest output survives. + let buf = Arc::new(Mutex::new(Vec::new())); + spawn_append(&buf, &vec![1u8; SPAWN_BUFFER_MAX]); + spawn_append(&buf, &[2u8; 100]); + let drained = spawn_drain(&buf); + assert_eq!(drained.len(), SPAWN_BUFFER_MAX); + assert!(drained[drained.len() - 100..].iter().all(|&b| b == 2)); + assert_eq!(drained[0], 1); + } + #[test] fn random_entropy_never_repeats_within_a_process() { // The call counter must make successive seeds distinct even when diff --git a/stdlib/std/process.rv b/stdlib/std/process.rv index 6a3449a..51f2101 100644 --- a/stdlib/std/process.rv +++ b/stdlib/std/process.rv @@ -1,8 +1,16 @@ // std/process: spawn subprocesses over the raven-runtime C ABI, backed by -// std::process::Command. A child runs to completion in one runtime call -// that captures its exit code, stdout, and stderr into a registry entry -// keyed by an opaque integer id; the wrappers pull those out and free the -// id. A non-zero exit is NOT an Err: the caller inspects `code`. Only a +// std::process::Command. Two modes share one registry-id scheme: +// +// - `run` waits: the child runs to completion in one runtime call that +// captures its exit code, stdout, and stderr; the wrappers pull those out +// and free the id. +// - `start` streams: the child starts without waiting (`spawn` is the +// goroutine keyword, so the function is `start`), runtime reader threads +// accumulate its output as it arrives, and the Child handle drains it +// incrementally (read_stdout/read_stderr), polls or waits for the exit +// code, and can kill the child. +// +// A non-zero exit is NOT an Err: the caller inspects `code`. Only a // spawn failure (for example the program is not found) takes the Err path // through the runtime's thread-local last-error slot, tagged with kind // "process". @@ -15,6 +23,7 @@ import std/error import std/string +import std/sync { sleep_millis } struct Output { code: Int, @@ -29,6 +38,91 @@ extern "C" { fun raven_process_stdout(id: Int) -> String fun raven_process_stderr(id: Int) -> String fun raven_process_free(id: Int) + fun raven_process_spawn(program: String, args_nul_joined: String) -> Int + fun raven_process_read_stdout(id: Int) -> String + fun raven_process_read_stderr(id: Int) -> String + fun raven_process_poll(id: Int) -> Int + fun raven_process_kill(id: Int) -> Int + fun raven_process_spawn_free(id: Int) +} + +// A child started with `start`, still possibly running. Output streams in: +// read_stdout and read_stderr drain whatever arrived since the last read, +// poll and running report status without blocking, wait blocks until exit, +// and kill stops it. Call free when done with the handle; freeing does not +// kill a running child. +struct Child { + id: Int, +} + +impl Child { + // Whatever the child wrote to stdout since the last read, "" when + // nothing new has arrived. + fun read_stdout(self) -> String { + return raven_process_read_stdout(self.id) + } + + // Whatever the child wrote to stderr since the last read. + fun read_stderr(self) -> String { + return raven_process_read_stderr(self.id) + } + + // The exit code once the child has exited, None while it runs. A child + // terminated by a signal with no code reports -1. + fun poll(self) -> Option { + let c = raven_process_poll(self.id) + if c == 0 - 2 { + return None + } + return Some(c) + } + + fun running(self) -> Bool { + return raven_process_poll(self.id) == 0 - 2 + } + + // Block until the child exits and return its code. Polls with a short + // sleep, so other goroutines keep running. + fun wait(self) -> Int { + let done = false + let code = 0 - 1 + while !done { + match self.poll() { + Some(c) -> { + code = c + done = true + }, + None -> { + sleep_millis(5) + }, + } + } + return code + } + + // Stop the child. True when the kill was delivered or it had already + // exited. + fun kill(self) -> Bool { + return raven_process_kill(self.id) == 1 + } + + // Drop the runtime entry for this child. Does not kill a running child; + // the runtime still reaps it when it exits. + fun free(self) { + raven_process_spawn_free(self.id) + } +} + +// Start `program` with `args` without waiting for it. Output accumulates as +// the child runs; drain it incrementally with read_stdout/read_stderr. On a +// spawn failure returns Err tagged with kind "process". +fun start(program: String, args: List) -> Result { + validate_args(args)? + let id = raven_process_spawn(program, join_args(args)) + if id == 0 { + return Err(Error { kind: "process", message: raven_process_last_error() }) + } + return Ok(Child { id }) } impl Output { diff --git a/tests/codegen_smoke.rs b/tests/codegen_smoke.rs index 4946255..1cb48c6 100644 --- a/tests/codegen_smoke.rs +++ b/tests/codegen_smoke.rs @@ -1963,6 +1963,41 @@ fn proc_argv_rejects_nul_arg() { ); } +#[test] +fn process_spawn_streams_and_kills() { + let Some(runtime) = supported_runtime() else { + return; + }; + // The streaming half of std/process end to end. The child + // (spawn_child.rv) prints "early", idles, prints "late", exits 3. The + // parent (spawn_parent.rv) proves the output streams (it drains "early" + // while the child is still running), waits for the code, kills a + // long-idling second child, and takes the Err path on a nonexistent + // program. See the expected transcript in spawn_parent.rv. + let child = build_example_binary("spawn_child.rv", &runtime); + let parent = build_example_binary("spawn_parent.rv", &runtime); + + let output = Command::new(&parent.binary) + .env("RAVEN_PROC_CHILD", &child.binary) + .output() + .expect("run spawn_parent binary"); + let stdout = String::from_utf8_lossy(&output.stdout).into_owned(); + let stderr = String::from_utf8_lossy(&output.stderr).into_owned(); + cleanup(&child.tmp); + cleanup(&parent.tmp); + assert!( + output.status.success(), + "spawn_parent binary exited non zero: status={:?} stderr={}", + output.status, + stderr + ); + assert_eq!( + stdout, "streamed-early\n3\ngot-late\nkilled\nspawn failed\n", + "unexpected stdout for spawn_parent: {:?}", + stdout + ); +} + #[test] fn proc_binary_io_compiles_and_runs() { let Some(runtime) = supported_runtime() else {