diff --git a/changelog.d/10913-stdin-reader-restart-race.md b/changelog.d/10913-stdin-reader-restart-race.md new file mode 100644 index 0000000000..78e777013b --- /dev/null +++ b/changelog.d/10913-stdin-reader-restart-race.md @@ -0,0 +1,23 @@ +Fixed `for await (const chunk of process.stdin)` on a pipe stalling forever +part-way through the input (#10895). + +The async iterator pauses its source after every delivered chunk and resumes +it on the next pull. On `process.stdin`, `pause()` latches `STDIN_DETACHED` — +the fd-0 reader thread exits when it sees the latch at the top of its loop — +and `resume()` clears the latch and respawns the reader unless +`STDIN_READER_STARTED` says one is still running. The reader's stop decision +and its `STARTED` reset were two separate steps, so a `resume()` that landed +between them found `STARTED` still true, spawned nothing, and the old reader +then left: fd 0 had no reader while every liveness view still reported an +open, flowing stdin, and the process idled forever with input unread. One roll +per delivered chunk: near-certain at 4 MiB through a 16 KiB macOS pipe, 1 in +350 at 16 MiB on Linux. Introduced by the unified fd-0 reader (2026-09-04); +the published 0.5.1520 predates it. + +The reader's check-and-clear and the restart CAS are now atomic with respect +to each other under one lifecycle lock (never held across `read()`); the +detach exit releases the reader slot itself and disarms the drop guard. + +Tests: `crates/perry/tests/issue_10895_stdin_pipe_stall.rs` (8 MiB in 256-byte +writes, 12 rounds, red on unpatched main) and `reader_lifecycle_tests` in +`os_process_streams.rs`, which replays the interleaving deterministically. diff --git a/crates/perry-runtime/src/os_process_streams.rs b/crates/perry-runtime/src/os_process_streams.rs index e6e585f8e5..5ea11f954c 100644 --- a/crates/perry-runtime/src/os_process_streams.rs +++ b/crates/perry-runtime/src/os_process_streams.rs @@ -349,25 +349,101 @@ pub extern "C" fn js_register_stdin_reader_consumer( } } -fn ensure_stdin_reader() { +/// Serializes the reader's decision to STOP with every request to (re)START it +/// (#10895). +/// +/// `pause()` sets `STDIN_DETACHED`; the reader notices at the top of its loop +/// and exits, and `resume()` clears the latch and calls `ensure_stdin_reader`, +/// which spawns a reader only when `STDIN_READER_STARTED` is false. Those two +/// flags used to be read and written independently, so this interleaving lost +/// the restart for good: +/// +/// reader: sees `STDIN_DETACHED == true`, decides to exit +/// main: `resume()` → `STDIN_DETACHED = false`; CAS(STARTED: false→true) +/// FAILS — the dying reader has not cleared STARTED yet +/// reader: clears STARTED and is gone +/// +/// fd 0 then has no reader while every liveness view still says stdin is open +/// and flowing, so the process idles forever with input unread. The async +/// iterator pauses/resumes the source once per delivered chunk, so a piped +/// `for await (const chunk of process.stdin)` rolled this dice hundreds of +/// times per megabyte. +/// +/// Holding this lock across the reader's check-and-clear and across the +/// restart CAS makes the two atomic with respect to each other: a restart +/// request either runs entirely before the stop decision (the reader then sees +/// the cleared latch and keeps going) or entirely after it (STARTED is already +/// false, so a fresh reader is spawned). It is never held across `read()`. +static STDIN_READER_LIFECYCLE: std::sync::Mutex<()> = std::sync::Mutex::new(()); + +fn stdin_reader_lifecycle() -> std::sync::MutexGuard<'static, ()> { + STDIN_READER_LIFECYCLE + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +/// The stop half of the lifecycle handshake, over an explicit slot flag so the +/// unit tests can replay interleavings without touching the process-global +/// one. `should_stop` is evaluated UNDER the lock, and a true answer releases +/// the slot in the same step. +fn reader_slot_claim_stop( + started: &std::sync::atomic::AtomicBool, + should_stop: impl FnOnce() -> bool, +) -> bool { + let _lifecycle = stdin_reader_lifecycle(); + if should_stop() { + started.store(false, std::sync::atomic::Ordering::Release); + true + } else { + false + } +} + +/// The restart half: true when the caller now owns the (single) reader slot +/// and must spawn the reader. +fn reader_slot_claim_start(started: &std::sync::atomic::AtomicBool) -> bool { use std::sync::atomic::Ordering; - // A previous reader may have exited (EOF, error, or explicit detach); its - // drop guard resets `STDIN_READER_STARTED` to false, - // so a later `resume()`/`on(...)` can spin up a fresh reader. - if STDIN_READER_STARTED + let _lifecycle = stdin_reader_lifecycle(); + started .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) .is_ok() - { +} + +/// The reader's top-of-loop stop check. Returns true when the reader must +/// exit; STARTED has then ALREADY been cleared, under the lifecycle lock, so +/// the caller must not clear it again (a late clear would clobber the `true` +/// of a reader respawned in between and let a third one start). +fn stdin_reader_claim_stop() -> bool { + reader_slot_claim_stop(&STDIN_READER_STARTED, stdin_reader_should_stop) +} + +fn stdin_reader_claim_start() -> bool { + reader_slot_claim_start(&STDIN_READER_STARTED) +} + +fn ensure_stdin_reader() { + // A previous reader may have exited (EOF, error, or explicit detach), which + // resets `STDIN_READER_STARTED` to false so a later `resume()`/`on(...)` + // can spin up a fresh reader. The claim is atomic with a live reader's + // decision to stop (#10895). + if stdin_reader_claim_start() { std::thread::spawn(|| { use std::io::Read; - // On exit, clear STARTED so the reader can be restarted later. - struct ReaderGuard; + // On an EOF / error / panic exit, clear STARTED so the reader can + // be restarted later. The detach exit clears it itself, inside + // `stdin_reader_claim_stop`, and disarms this guard. + struct ReaderGuard { + armed: bool, + } impl Drop for ReaderGuard { fn drop(&mut self) { - STDIN_READER_STARTED.store(false, std::sync::atomic::Ordering::Release); + if self.armed { + let _lifecycle = stdin_reader_lifecycle(); + STDIN_READER_STARTED.store(false, std::sync::atomic::Ordering::Release); + } } } - let _guard = ReaderGuard; + let mut guard = ReaderGuard { armed: true }; let stdin = std::io::stdin(); let mut handle = stdin.lock(); // Read in chunks, not one byte at a time. A paste or a fast-typed @@ -387,7 +463,11 @@ fn ensure_stdin_reader() { // #9676: `stdin_reader_should_stop`, NOT `stdin_is_detached` — // an `unref()`d stdin still delivers data in Node, and reading // the liveness view here is what killed the reader for good. - if stdin_reader_should_stop() { + // #10895: the check and the STARTED reset are one step under + // the lifecycle lock, so a concurrent `resume()` can never + // find STARTED still true for a reader that is already leaving. + if stdin_reader_claim_stop() { + guard.armed = false; break; } match handle.read(&mut buf) { @@ -1266,6 +1346,88 @@ mod empty_checkpoint_tests { } } +#[cfg(test)] +mod reader_lifecycle_tests { + use super::{reader_slot_claim_start, reader_slot_claim_stop}; + use std::sync::atomic::{AtomicBool, Ordering}; + + /// #10895: replays the interleaving that stranded fd 0 without a reader — + /// the reader decides to stop, and `resume()` asks for a restart BEFORE + /// the dying reader has finished leaving. The stop decision must already + /// have released the reader slot, or the restart's claim fails and nobody + /// ever reads stdin again. + /// + /// Runs on local flags: no fd-0 reader is spawned and no process-global + /// stdin state is touched, so it cannot disturb the liveness tests. + #[test] + fn a_restart_requested_while_the_reader_is_leaving_is_not_lost() { + // A reader is running and `pause()` has latched the detach. + let started = AtomicBool::new(true); + let detached = AtomicBool::new(true); + assert!( + reader_slot_claim_stop(&started, || detached.load(Ordering::Acquire)), + "a detached reader must decide to stop" + ); + // `resume()`: clear the latch, then ask for a reader. The old reader + // has not run another instruction since its stop decision. + detached.store(false, Ordering::Release); + assert!( + reader_slot_claim_start(&started), + "restart lost: the stopping reader still held the reader slot" + ); + // The respawned reader owns the slot; a second request is a no-op. + assert!(!reader_slot_claim_start(&started)); + } + + /// The other order: `resume()` clears the latch before the reader looks. + /// The reader keeps running and no second reader may be started on fd 0. + #[test] + fn a_resume_that_beats_the_stop_check_keeps_the_one_reader() { + let started = AtomicBool::new(true); + let detached = AtomicBool::new(true); + detached.store(false, Ordering::Release); + assert!(!reader_slot_claim_start(&started)); + assert!(!reader_slot_claim_stop(&started, || detached.load(Ordering::Acquire))); + assert!(started.load(Ordering::Acquire)); + } + + /// Hammer the handshake from two threads: a "reader" that stops whenever + /// it sees the latch and a "main" that pauses/resumes. After every + /// resume the slot must be owned — by the surviving reader or by the + /// restart — never stranded. + #[test] + fn pause_resume_storm_never_strands_the_slot() { + use std::sync::Arc; + let started = Arc::new(AtomicBool::new(true)); + let detached = Arc::new(AtomicBool::new(false)); + let done = Arc::new(AtomicBool::new(false)); + let reader = { + let (started, detached, done) = (started.clone(), detached.clone(), done.clone()); + std::thread::spawn(move || { + while !done.load(Ordering::Acquire) { + // A live reader polls the latch between reads; one that + // stopped waits to be "respawned" by main's claim. + if started.load(Ordering::Acquire) { + reader_slot_claim_stop(&started, || detached.load(Ordering::Acquire)); + } + std::hint::spin_loop(); + } + }) + }; + for _ in 0..200_000 { + detached.store(true, Ordering::Release); // pause() + detached.store(false, Ordering::Release); // resume(): clear … + reader_slot_claim_start(&started); // … then ensure a reader + assert!( + started.load(Ordering::Acquire), + "resume() returned with no reader owning fd 0" + ); + } + done.store(true, Ordering::Release); + reader.join().unwrap(); + } +} + fn pump_stdin_data_chunks() { let has_bytes = STDIN_BUFFER.lock().map(|b| !b.is_empty()).unwrap_or(false); if !has_bytes { diff --git a/crates/perry/tests/issue_10895_stdin_pipe_stall.rs b/crates/perry/tests/issue_10895_stdin_pipe_stall.rs new file mode 100644 index 0000000000..f3203aa0cd --- /dev/null +++ b/crates/perry/tests/issue_10895_stdin_pipe_stall.rs @@ -0,0 +1,152 @@ +//! Regression coverage for #10895: `for await (const chunk of process.stdin)` +//! on a pipe stalled forever part-way through the input. +//! +//! The async iterator pauses its source after every delivered chunk and +//! resumes it on the next pull. On `process.stdin`, `pause()` latches +//! `STDIN_DETACHED` (the fd-0 reader thread exits when it sees it) and +//! `resume()` clears the latch and respawns the reader unless one is still +//! registered. A `resume()` that landed while the old reader was on its way +//! out found it still registered, spawned nothing, and the old reader then +//! left: no reader on fd 0, every liveness view still reporting an open, +//! flowing stdin, the process idle forever with input unread. +//! +//! It is a race, so this test is statistical by nature: many small writes +//! (each pause/resume cycle is one roll) over several rounds. On an unpatched +//! build a single 8 MiB round fed in 256-byte writes stalls roughly every +//! second time on macOS and rarely on Linux; the deterministic witness for +//! the interleaving itself is `reader_lifecycle_tests` in +//! `perry-runtime/src/os_process_streams.rs`. + +#![cfg(unix)] + +use std::io::{Read, Write}; +use std::path::{Path, PathBuf}; +use std::process::{Command, Stdio}; +use std::time::{Duration, Instant}; + +const SOURCE: &str = include_str!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/../../test-files/test_issue_10895_stdin_pipe_stall.ts" +)); + +const TOTAL_BYTES: usize = 8 * 1024 * 1024; +const WRITE_BYTES: usize = 256; +const ROUNDS: usize = 12; +const ROUND_DEADLINE: Duration = Duration::from_secs(60); + +fn perry_bin() -> PathBuf { + PathBuf::from(env!("CARGO_BIN_EXE_perry")) +} + +fn compile(dir: &Path) -> PathBuf { + let source = dir.join("stdin_pipe_stall.ts"); + let binary = dir.join("stdin_pipe_stall_bin"); + std::fs::write(&source, SOURCE).expect("write stdin fixture"); + let output = Command::new(perry_bin()) + .current_dir(dir) + .arg("compile") + .arg(&source) + .arg("-o") + .arg(&binary) + .output() + .expect("compile stdin fixture"); + assert!( + output.status.success(), + "fixture compile failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + binary +} + +/// One round: pipe `TOTAL_BYTES` in `WRITE_BYTES` writes, close stdin, and +/// require the child to report every byte before the deadline. +fn run_round(binary: &Path, round: usize) { + let mut child = Command::new(binary) + .env("PERRY_10895_DRIVE", "1") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn stdin fixture"); + let mut stdin = child.stdin.take().expect("fixture stdin"); + let writer = std::thread::spawn(move || { + let piece = [1u8; WRITE_BYTES]; + let mut left = TOTAL_BYTES; + while left > 0 { + let n = left.min(WRITE_BYTES); + // A stalled child stops draining the pipe; the kill below then + // breaks this write with EPIPE. Either way the thread ends. + if stdin.write_all(&piece[..n]).is_err() { + return left; + } + left -= n; + } + drop(stdin); + 0 + }); + + let started = Instant::now(); + let status = loop { + if let Some(status) = child.try_wait().expect("poll stdin fixture") { + break Some(status); + } + if started.elapsed() > ROUND_DEADLINE { + break None; + } + std::thread::sleep(Duration::from_millis(5)); + }; + let Some(status) = status else { + let _ = child.kill(); + let _ = child.wait(); + let unwritten = writer.join().unwrap_or(TOTAL_BYTES); + panic!( + "round {round}: piped stdin stalled — the child was still alive after {:?} with \ + {unwritten} of {TOTAL_BYTES} bytes not even accepted by the pipe (#10895)", + ROUND_DEADLINE + ); + }; + assert_eq!(writer.join().expect("writer thread"), 0); + let mut stdout = String::new(); + child + .stdout + .take() + .expect("fixture stdout") + .read_to_string(&mut stdout) + .expect("read fixture stdout"); + assert!( + status.success(), + "round {round}: exit {status:?}: {stdout:?}" + ); + assert_eq!( + stdout.trim_end(), + format!("RESULT:{TOTAL_BYTES}"), + "round {round}: not every piped byte reached the iterator" + ); +} + +#[test] +fn piped_stdin_async_iteration_reads_to_eof_every_time() { + let dir = tempfile::tempdir().expect("create fixture directory"); + let binary = compile(dir.path()); + for round in 0..ROUNDS { + run_round(&binary, round); + } +} + +/// Undriven, the fixture must leave stdin alone (the parity sweep runs it +/// with whatever stdin the caller has). +#[test] +fn undriven_fixture_does_not_wait_on_stdin() { + let dir = tempfile::tempdir().expect("create fixture directory"); + let binary = compile(dir.path()); + let output = Command::new(&binary) + .stdin(Stdio::piped()) + .output() + .expect("run undriven fixture"); + assert!(output.status.success()); + assert_eq!( + String::from_utf8_lossy(&output.stdout).trim_end(), + "RESULT:idle" + ); +} diff --git a/test-files/test_issue_10895_stdin_pipe_stall.ts b/test-files/test_issue_10895_stdin_pipe_stall.ts new file mode 100644 index 0000000000..5e1196ef91 --- /dev/null +++ b/test-files/test_issue_10895_stdin_pipe_stall.ts @@ -0,0 +1,32 @@ +// #10895: `for await (const chunk of process.stdin)` on a pipe stalled forever +// part-way through the input. The async iterator pauses the source after every +// delivered chunk and resumes it on the next pull; `pause()` makes the fd-0 +// reader thread exit and `resume()` respawns it, and a resume that landed while +// the old reader was still on its way out lost the respawn for good. +// +// Driven by `crates/perry/tests/issue_10895_stdin_pipe_stall.rs`, which pipes +// several MiB in small writes. Undriven (the parity sweep) it must not touch +// stdin at all: the sweep inherits whatever stdin the caller has, and a fixture +// that waits for EOF on a terminal never finishes. +const driven = process.env.PERRY_10895_DRIVE === "1"; + +async function main(): Promise { + if (!driven) { + console.log("RESULT:idle"); + return; + } + let total = 0; + for await (const data of process.stdin) { + const chunk: Uint8Array = data; + total += chunk.length; + // Touch the payload so a chunk that arrives with the right length but the + // wrong bytes is caught too (every byte the driver writes is 1). + if (chunk[0] !== 1 || chunk[chunk.length - 1] !== 1) { + console.log("RESULT:corrupt@" + total); + return; + } + } + console.log("RESULT:" + total); +} + +main();