diff --git a/cranelift/runtime/src/platform.rs b/cranelift/runtime/src/platform.rs index a79b0897..bf45ebd7 100644 --- a/cranelift/runtime/src/platform.rs +++ b/cranelift/runtime/src/platform.rs @@ -68,6 +68,9 @@ pub unsafe extern "C" fn pith_input() -> *mut i8 { /// Execute a command and return exit code /// +/// the child runs to completion via the process pool (see `process`), so a +/// green worker is not held for however long the command takes. +/// /// # Safety /// command must be a valid null-terminated C string #[no_mangle] @@ -87,24 +90,34 @@ pub unsafe extern "C" fn pith_exec(command: *const i8) -> i64 { cmd.args(&parts[1..]); } - if let Ok(status) = cmd.status() { - if let Some(code) = status.code() { - return code as i64; - } - return 0; + match crate::process::command_status(cmd) { + Some(status) => status.code().unwrap_or(0) as i64, + None => -1, } - -1 +} + +/// the multiplier/increment of the LCG behind `pith_random_float` (knuth's +/// MMIX constants). +const LCG_MUL: u64 = 6364136223846793005; +const LCG_INC: u64 = 1; + +fn lcg_step(s: u64) -> u64 { + s.wrapping_mul(LCG_MUL).wrapping_add(LCG_INC) } /// Random float between 0.0 and 1.0 #[no_mangle] pub extern "C" fn pith_random_float() -> f64 { - use std::num::Wrapping; - - let s = RANDOM_SEED.load(Ordering::Relaxed); - let new_s = Wrapping(s) * Wrapping(6364136223846793005) + Wrapping(1); - RANDOM_SEED.store(new_s.0, Ordering::Relaxed); - (new_s.0 >> 11) as f64 / (1u64 << 53) as f64 + // step the seed with a compare-exchange rather than a load/store pair: two + // tasks on different workers drawing at once must each claim a distinct + // step of the sequence, not both advance from the same seed and return the + // identical "random" value. relaxed ordering is enough — the value itself + // is the only payload, nothing else is published through it. + let prev = RANDOM_SEED + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |s| Some(lcg_step(s))) + .unwrap_or(0); // unreachable: the closure never returns None + let new_s = lcg_step(prev); + (new_s >> 11) as f64 / (1u64 << 53) as f64 } /// Seed the random number generator @@ -152,6 +165,50 @@ pub unsafe extern "C" fn pith_random_string(len: i64) -> *mut i8 { mod tests { use super::*; + // both random tests reseed the one process-global sequence, so they must + // not interleave with each other (cargo runs tests concurrently). + static RANDOM_TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); + + // seeding pins the whole sequence: same seed, same draws. this is the + // contract the CAS step must preserve for single-threaded callers. + #[test] + fn seeded_draws_are_reproducible() { + let _guard = RANDOM_TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner()); + pith_random_seed(42); + let first: Vec = (0..8).map(|_| pith_random_float().to_bits()).collect(); + pith_random_seed(42); + let second: Vec = (0..8).map(|_| pith_random_float().to_bits()).collect(); + assert_eq!(first, second); + } + + // the race this guards against: with a load/store seed update, two threads + // can both step from the same seed and return the identical draw. the + // compare-exchange gives every draw its own step of the sequence, so + // concurrent draws are distinct (up to the astronomically unlikely 53-bit + // output collision between different seeds). + #[test] + fn concurrent_draws_do_not_repeat() { + let _guard = RANDOM_TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner()); + pith_random_seed(7); + let threads: Vec<_> = (0..4) + .map(|_| { + std::thread::spawn(|| { + (0..2000) + .map(|_| pith_random_float().to_bits()) + .collect::>() + }) + }) + .collect(); + let mut all: Vec = threads + .into_iter() + .flat_map(|t| t.join().expect("drawing thread panicked")) + .collect(); + let total = all.len(); + all.sort_unstable(); + all.dedup(); + assert_eq!(all.len(), total, "two concurrent draws returned the same value"); + } + #[test] fn exec_rejects_null_and_invalid_utf8() { let invalid = [0xffu8, 0x00]; diff --git a/cranelift/runtime/src/process.rs b/cranelift/runtime/src/process.rs index 06ea73bf..e174c6ce 100644 --- a/cranelift/runtime/src/process.rs +++ b/cranelift/runtime/src/process.rs @@ -10,14 +10,17 @@ //! worth waiting for is pollable. the pipes are pipes, and linux will hand out //! a `pidfd` that becomes readable exactly when the child exits. so both waits //! go to the epoll reactor that already serves sockets (`netpoll`), and no -//! thread is parked on a child anywhere in the runtime. +//! green worker is parked on a spawned child anywhere in the runtime. //! -//! that also settles the question of what may run off the task's own thread: -//! nothing does. the reactor never runs pith code — it only marks a task -//! runnable, and the task resumes on the worker it was pinned to — so every -//! handle, `Bytes`, and C string here is built where it always was. - -use crate::blocking; +//! the run-to-completion calls (`process.output`, `exec`) cannot use the +//! reactor: `Command::output` and `Command::status` spawn, drain, and wait in +//! one opaque std call. those go to a small thread pool instead, the same +//! shape dns and file i/o use — see `command_output`/`command_status` and the +//! rule in `blocking` about what may cross to a pool thread. the `Command` +//! going in and the `Output` coming back are plain owned data; every handle, +//! `Bytes`, and C string is still built on the calling task's thread. + +use crate::blocking::{self, Pool}; use crate::collections::list::PithList; use crate::concurrency::scheduler::{backend, Backend}; use crate::fdio; @@ -31,6 +34,25 @@ use std::process::{Child, ChildStderr, ChildStdin, ChildStdout, Command, Stdio}; use std::sync::atomic::{AtomicI64, Ordering}; use std::sync::OnceLock; +/// the threads that run whole-child commands (`Command::output`/`status`) for +/// green tasks. a child decides for itself when it exits, so one of these can +/// be held for an unbounded time — its own small pool keeps that from queueing +/// behind (or in front of) file i/o or dns. +static POOL: Pool = Pool::new("pith-proc", 4); + +/// run `command.output()` — spawn, drain both pipes, wait — without holding a +/// green worker for the child's lifetime. off a green task it runs inline, +/// exactly as the callers always did. `None` is any spawn/read failure. +pub(crate) fn command_output(mut command: Command) -> Option { + blocking::run(&POOL, move || command.output().ok()) +} + +/// run `command.status()` — spawn and wait, stdio inherited — with the same +/// offload-or-inline split as `command_output`. +pub(crate) fn command_status(mut command: Command) -> Option { + blocking::run(&POOL, move || command.status().ok()) +} + /// which of a child's three pipes a caller wants. #[derive(Clone, Copy, PartialEq, Eq)] pub(crate) enum Pipe { @@ -232,18 +254,20 @@ pub unsafe extern "C" fn pith_process_output_argv( env_keys: PithList, env_values: PithList, ) -> i64 { - let Some(mut command) = pith_build_command(program, argv, cwd, env_keys, env_values) else { + let Some(command) = pith_build_command(program, argv, cwd, env_keys, env_values) else { return 0; }; - match command.output() { - Ok(output) => { + // the child runs to completion on a pool thread (or inline, off green); the + // pith-visible handle is built here either way. + match command_output(command) { + Some(output) => { let status = output.status.code().unwrap_or(-1) as i64; let stdout = String::from_utf8_lossy(&output.stdout).to_string(); let stderr = String::from_utf8_lossy(&output.stderr).to_string(); pith_store_process_output(status, stdout, stderr) } - Err(_) => 0, + None => 0, } } @@ -439,6 +463,26 @@ pub extern "C" fn pith_process_close(handle: i64) { mod tests { use super::*; + // unit tests never run on a green worker, so these exercise the inline arm + // of the offload split; the pool arm is the same closure on another thread. + #[test] + fn command_output_captures_both_pipes_and_the_exit_code() { + let mut cmd = Command::new("/bin/sh"); + cmd.args(["-c", "echo out; echo err >&2; exit 3"]); + let output = command_output(cmd).expect("sh should run"); + assert_eq!(output.status.code(), Some(3)); + assert_eq!(output.stdout, b"out\n"); + assert_eq!(output.stderr, b"err\n"); + } + + #[test] + fn command_status_reports_the_exit_code() { + let mut cmd = Command::new("/bin/sh"); + cmd.args(["-c", "exit 5"]); + assert_eq!(command_status(cmd).expect("sh should run").code(), Some(5)); + assert!(command_status(Command::new("/nonexistent-program")).is_none()); + } + #[test] fn invalid_process_handles_return_safe_defaults() { assert_eq!(pith_process_wait(12345), -1); diff --git a/cranelift/runtime/src/runtime_core.rs b/cranelift/runtime/src/runtime_core.rs index 415c66c9..e0b37fad 100644 --- a/cranelift/runtime/src/runtime_core.rs +++ b/cranelift/runtime/src/runtime_core.rs @@ -621,6 +621,26 @@ mod tests { } } + // structs travel between tasks, and tasks run on different worker threads, + // so the last strong owner can die on a thread the weak reader never runs + // on. the dead flag is an atomic for exactly this shape; the join gives the + // load a happens-after edge, so it must observe the death. + #[test] + fn weak_load_sees_a_release_from_another_thread() { + unsafe { + let s = pith_struct_alloc(1); + pith_struct_weak_retain(s); + assert_eq!(pith_struct_weak_load(s), s); + + std::thread::spawn(move || pith_struct_release(s)) + .join() + .expect("releasing thread panicked"); + + assert_eq!(pith_struct_weak_load(s), 0); + pith_struct_weak_release(s); + } + } + #[test] fn weak_load_on_a_dead_or_invalid_pointer_is_none() { unsafe { @@ -1689,6 +1709,15 @@ unsafe fn struct_weak(base: *mut u8) -> &'static std::sync::atomic::AtomicU32 { &*(base.add(STRUCT_OFF_WEAK) as *const std::sync::atomic::AtomicU32) } +// the dead flag must be an atomic like its neighbors: structs cross task (and +// therefore worker-thread) boundaries, so the last strong owner can set it on +// one thread while a weak reference reads it on another. the release store +// pairs with the acquire load in `pith_struct_weak_load`, so a reader that +// observes the value as dead also observes everything the destructor did. +unsafe fn struct_dead(base: *mut u8) -> &'static std::sync::atomic::AtomicU32 { + &*(base.add(STRUCT_OFF_DEAD) as *const std::sync::atomic::AtomicU32) +} + // drop one unit of weak; free the allocation when it reaches zero. this is the // single place a struct header is deallocated, so the strong path and every // weak reference funnel their final drop through here. @@ -1787,9 +1816,11 @@ pub unsafe extern "C" fn pith_struct_release(ptr: i64) { f(ptr); } crate::perf_count(&crate::PERF_STRUCT_FREES, 1); - // the value is now dead; a weak read after this returns none. drop the - // implicit weak unit, which frees the header if no weak refs remain. - (base.add(STRUCT_OFF_DEAD) as *mut u32).write(1); + // the value is now dead; a weak read after this returns none. the release + // store publishes the destructor's writes to any thread whose weak load + // sees the flag. then drop the implicit weak unit, which frees the header + // if no weak refs remain. + struct_dead(base).store(1, std::sync::atomic::Ordering::Release); struct_weak_drop(base); } @@ -1828,7 +1859,9 @@ pub unsafe extern "C" fn pith_struct_weak_load(ptr: i64) -> i64 { let Some(base) = struct_base(ptr) else { return 0; }; - if (base.add(STRUCT_OFF_DEAD) as *const u32).read() != 0 { + // acquire pairs with the release store in `pith_struct_release`: a load + // that sees the target dead also sees its destructor's effects. + if struct_dead(base).load(std::sync::atomic::Ordering::Acquire) != 0 { return 0; } ptr diff --git a/cranelift/runtime/src/utility.rs b/cranelift/runtime/src/utility.rs index ecd57365..42405893 100644 --- a/cranelift/runtime/src/utility.rs +++ b/cranelift/runtime/src/utility.rs @@ -45,6 +45,9 @@ pub unsafe extern "C" fn pith_log_error(msg: *const i8) { } /// Execute command and capture output — returns stdout as C string +/// +/// the child runs to completion via the process pool (see `process`), so a +/// green worker is not held for however long the command takes. #[no_mangle] pub unsafe extern "C" fn pith_exec_output(cmd: *const i8) -> *mut i8 { let Some(cmd_str) = cstr_str(cmd) else { @@ -52,15 +55,13 @@ pub unsafe extern "C" fn pith_exec_output(cmd: *const i8) -> *mut i8 { }; let parts: Vec<&str> = cmd_str.split_whitespace().collect(); if parts.is_empty() { - std::ptr::null_mut() - } else { - match std::process::Command::new(parts[0]) - .args(&parts[1..]) - .output() - { - Ok(output) => crate::pith_copy_bytes_to_cstring(&output.stdout), - Err(_) => std::ptr::null_mut(), - } + return std::ptr::null_mut(); + } + let mut command = std::process::Command::new(parts[0]); + command.args(&parts[1..]); + match crate::process::command_output(command) { + Some(output) => crate::pith_copy_bytes_to_cstring(&output.stdout), + None => std::ptr::null_mut(), } } diff --git a/docs/concurrency.md b/docs/concurrency.md index 9dd50e26..58fc6a46 100644 --- a/docs/concurrency.md +++ b/docs/concurrency.md @@ -216,13 +216,15 @@ matters more here than anywhere else: a wait has no bound at all, and four `sleep 3600`s would have emptied a four-thread pool. a child that has already finished by the time you wait for it is collected on the spot, with no park. -`process.output` is the exception, along with everything routed through it -(`run`, `text`, `output_checked`) and the two shell helpers `run_shell` and -`output_shell`. each of those runs a child to completion inside one call while -draining both of its pipes, and draining two pipes at once without blocking -needs a reactor wait covering more than one fd, which is not built yet. until -it is, `start` plus your own reads and `wait` yields properly; `output_ctx` is -that pattern already written out. +`process.output` — along with everything routed through it (`run`, `text`, +`output_checked`) and the two shell helpers `run_shell` and `output_shell` — +runs a child to completion inside one call while draining both of its pipes, +which the reactor cannot cover (it waits on one fd at a time). those calls go +to a small process pool of their own instead, the dns/file shape: the command +crosses to a blocking thread, the task parks, and the worker stays free for +however long the child runs. `exec` and `exec_output` take the same route. +`start` plus your own reads and `wait` still parks on the reactor directly, +with no pool thread involved. a task waits a few microseconds on-CPU before it actually parks. a read the page cache answers comes back faster than the two thread wakeups a park costs, so diff --git a/docs/limitations.md b/docs/limitations.md index 6fa42b72..8d53e5ab 100644 --- a/docs/limitations.md +++ b/docs/limitations.md @@ -107,6 +107,24 @@ correctness story: cross-module map reads, set codegen, negative float literals like `-1.0`) were re-checked and all pass; they are now pinned by regression tests (`tests/cases/test_xmod_float.pith` and friends). +- `os.set_env` after tasks have started is a libc-level race. glibc's `setenv` + and `getenv` are not synchronized against each other, and the runtime's own + pool threads read the environment behind your back — `getaddrinfo` on the + dns pool consults `RES_OPTIONS` and friends on every lookup. rust's internal + env lock only covers rust-side accesses, so a `set_env` concurrent with a + dial can crash in libc. set environment variables before spawning tasks. the + candidate fixes (snapshot the environment at startup, or make late `set_env` + write to an overlay that child processes inherit) both change observable + semantics, so this is documented rather than decided for now. +- closing an fd-backed handle races a concurrent call on the same handle. a + task parked on a socket or pipe that another task closes is woken with an + error (the reactor's close teardown), which is the common case and safe. + what remains is the standard raw-fd hazard: between one task's handle + lookup and its syscall, a close plus a new open can recycle the fd number, + and the syscall lands on the wrong fd. present on both backends and + inherited from fd semantics; closing it needs handles that carry liveness + (a generation, like task handles have) rather than raw fd numbers. do not + share a connection between tasks without coordinating its close. ## the green backend, now the default on linux @@ -134,14 +152,22 @@ thread would. on a slow network mount that reasoning does not hold and a task doing one of them holds its worker until it returns. making that decision adaptive rather than fixed is the open work. -the other is `process.output` and the calls built on it (`run`, `text`, -`output_checked`, `run_shell`, `output_shell`). each runs a child to completion -inside one runtime call and holds the worker for as long as the child lives. -draining a child's stdout and stderr at the same time without blocking needs a -reactor wait that covers more than one fd, and the reactor waits on one; adding -that is the fix. `start` plus explicit reads and `wait` yields properly today, -and `output_ctx` is that pattern already written out. `PITH_GREEN=0` is the -blunt workaround for either. +the other is `sleep`, and everything built on it. `time.delay` maps straight +to the blocking sleep, so a green task that sleeps holds its worker for the +duration — and the waiting loops built on delay hold one too: a `select` with +no ready arm sleeps a millisecond per probe until an arm fires, and the +`concurrent.after`/`ticker` workers and a context's deadline watcher do the +same. each slice is short and correctness is unaffected, but on a two-worker +host two idle selects can occupy both workers while they poll. the fix is a +real timer in the reactor — the deadline heap `netpoll` already keeps could +serve a pure-timeout wait with no fd — so a sleeping task parks the way a +socket read does. `PITH_GREEN=0` is the blunt workaround. + +(`process.output` and the calls built on it — `run`, `text`, `output_checked`, +`run_shell`, `output_shell`, plus `exec` and `exec_output` — used to be on +this list for holding a worker while a child ran to completion; they now hand +the whole spawn-drain-wait to a process pool of their own and the caller +parks, the same shape dns and file i/o use.) the calling task pays for that. a file call made from inside a task now costs a thread handoff it did not before, so a task reading a small cached file in a