diff --git a/src/io/PipeReader.rs b/src/io/PipeReader.rs index 60e18aa01ea2..762e064ccd1b 100644 --- a/src/io/PipeReader.rs +++ b/src/io/PipeReader.rs @@ -4,7 +4,7 @@ use core::ptr::NonNull; use bun_sys::{self as sys, Fd}; -use crate::{EventLoopHandle, FilePollFlag, FilePollKind, FilePollRef, Owner, PollTag}; +use crate::{EventLoopHandle, FilePollKind, FilePollRef, Owner, PollTag}; // `bun.Async.Loop` — on POSIX the uws `us_loop_t`, on Windows the embedded // `uv_loop_t` (`bun_io::Loop` is the cfg-aliased nominal that picks the // right one). `BufferedReaderParent::loop_` returns this so callers in T3+ @@ -239,6 +239,10 @@ impl PosixBufferedReader { let Some(poll) = self.handle.get_poll() else { return; }; + // An unarmed poll delivers nothing; `try_register_poll` applies KEEP_ALIVE when it arms. + if value && !poll.is_watching() { + return; + } poll.set_keeping_process_alive(self.vtable.event_loop(), value); } @@ -529,9 +533,8 @@ impl PosixBufferedReader { }; poll.set_owner(Owner::new(PollTag::BufferedReader, owner_ptr.cast())); - if !poll.has_flag(FilePollFlag::WasEverRegistered) - && self.flags.contains(PosixFlags::KEEP_ALIVE) - { + // Re-applied on every arm: `pause()` unregisters, which drops it. + if self.flags.contains(PosixFlags::KEEP_ALIVE) { poll.enable_keeping_process_alive(ev); } diff --git a/src/io/posix_event_loop.rs b/src/io/posix_event_loop.rs index 1d9cfc2782be..6ff1bb4dff18 100644 --- a/src/io/posix_event_loop.rs +++ b/src/io/posix_event_loop.rs @@ -903,22 +903,23 @@ impl FilePoll { ) -> sys::Result<()> { debug_assert!(fd.native() >= 0 && fd != INVALID_FD); - if !(self.flags.contains(Flags::PollReadable) + let registered = self.flags.contains(Flags::PollReadable) || self.flags.contains(Flags::PollWritable) || self.flags.contains(Flags::PollProcess) || self.flags.contains(Flags::PollMachport) - || self.flags.contains(Flags::PollMemoryPressure)) - { - // no-op + || self.flags.contains(Flags::PollMemoryPressure); + // The `needs_rearm` skip below keeps the disarmed kernel registration, so teardown must still delete it. + let disarmed_only = !registered && self.flags.contains(Flags::NeedsRearm); + if !registered && !(disarmed_only && force_unregister) { return sys::Result::Ok(()); } - debug_assert!(fd != INVALID_FD); let watcher_fd = loop_.fd; - let both_directions = - self.flags.contains(Flags::PollReadable) && self.flags.contains(Flags::PollWritable); + let both_directions = disarmed_only + || (self.flags.contains(Flags::PollReadable) + && self.flags.contains(Flags::PollWritable)); let flag: Flags = 'brk: { - if self.flags.contains(Flags::PollReadable) { + if disarmed_only || self.flags.contains(Flags::PollReadable) { break 'brk Flags::Readable; } if self.flags.contains(Flags::PollWritable) { diff --git a/src/runtime/webcore/FileReader.rs b/src/runtime/webcore/FileReader.rs index 13ee76ec5b8f..4ef95677df64 100644 --- a/src/runtime/webcore/FileReader.rs +++ b/src/runtime/webcore/FileReader.rs @@ -655,8 +655,7 @@ impl FileReader { // No JS read is waiting; stop at the highwater mark and let onPull restart. `started` gates it: a non-lazy `Bun.spawn` pipe is already reading before any consumer attaches, and throttling then deadlocks a child alternating stdout/stderr writes. let keep_going = !self.started.get() || (self.flowing.get() && self.buffered.get().len() < self.highwater_mark); - // A completion-driven reader keeps issuing reads unless stopped; `on_pull` restarts it. - #[cfg(windows)] + // `false` only ends this read loop; `pause()` stops the reader and releases its hold on the event loop until `on_pull`. if !keep_going { self.reader().pause(); } @@ -809,6 +808,8 @@ impl FileReader { } if !self.reader().has_pending_read() && self.flowing.get() { + // A consumer is pulling again: undo the highwater pause from `on_read_chunk`. + self.reader().unpause(); // SAFETY: the reader cell is live for `self`'s lifetime; `read_into` is the raw re-entrancy-safe entry (EOF/error dispatch runs user JS). let (amount_read, state) = unsafe { IOReader::read_into(self.reader.get(), buffer) }; bun_core::scoped_log!(FileReader, "onPull({}) = {}", buffer.len(), amount_read); diff --git a/test/js/bun/spawn/spawn.test.ts b/test/js/bun/spawn/spawn.test.ts index ee4a5fdff1cd..a9003d595c48 100644 --- a/test/js/bun/spawn/spawn.test.ts +++ b/test/js/bun/spawn/spawn.test.ts @@ -807,6 +807,134 @@ describe("should not hang", () => { } }); +// What keeps a process alive whose only handle is a child's stdout reader (the +// child itself is unref'd): a read that can still deliver something does, a +// reader stopped at its highwater mark does not. Stopped, the pipe reader's +// poll is unregistered until the next pull, so it could not even observe the +// child going away. Node: readStop() at the highWaterMark leaves the handle +// inactive, and a pending read keeps it active. +describe.skipIf(isWindows)("stdout reader of an unref'd child and process lifetime", () => { + async function run(script: string) { + await using proc = spawn({ + cmd: [bunExe(), "-e", script], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(exitCode).toBe(0); + return stdout; + } + + // "saturating" fills the pipe faster than the parent reads it; "trickling" + // lets most parent reads end in EAGAIN, the path that used to re-arm the + // poll regardless of the highwater stop. + const producers = { + saturating: `const chunk = Buffer.alloc(8192, 120); for (;;) require("fs").writeSync(1, chunk);`, + trickling: `const chunk = Buffer.alloc(2048, 120); setInterval(() => require("fs").writeSync(1, chunk), 1);`, + }; + for (const [kind, producer] of Object.entries(producers)) { + it.concurrent( + `an idle reader stopped at the highwater mark does not keep the process alive (${kind} writer)`, + async () => { + const stdout = await run(` + const producer = Bun.spawn({ + cmd: [process.execPath, "-e", ${JSON.stringify(producer)}], + stdin: "ignore", + stdout: "pipe", + stderr: "inherit", + }); + const reader = producer.stdout.getReader(); + // The running child no longer counts; only its stdout reader can keep this process alive now. + producer.unref(); + let firstLength = -1; + process.on("exit", () => { + console.log(JSON.stringify({ gotFirstChunk: firstLength > 0, producerExitCode: producer.exitCode })); + producer.kill("SIGKILL"); + }); + firstLength = (await reader.read()).value.length; + // The reader stays locked and idle while the child keeps writing: the pipe + // reader fills to its highwater mark and stops, and then nothing is pending. + `); + expect(JSON.parse(stdout)).toEqual({ gotFirstChunk: true, producerExitCode: null }); + }, + ); + } + + it.concurrent("a pending read keeps the process alive until the child writes", async () => { + const child = `const fs = require("fs"); fs.readSync(0, Buffer.alloc(4)); fs.writeSync(1, "pong");`; + // Not top-level await: an unsettled entry-module promise keeps the process + // running on its own and would hide a read that does not. + const stdout = await run(` + const child = Bun.spawn({ + cmd: [process.execPath, "-e", ${JSON.stringify(child)}], + stdin: "pipe", + stdout: "pipe", + stderr: "inherit", + }); + const reader = child.stdout.getReader(); + child.unref(); + (async () => { + const pending = reader.read(); + child.stdin.write("ping"); + child.stdin.end(); + // Only this read is left to wait for. It must hold the process until "pong" arrives. + const { value } = await pending; + console.log(new TextDecoder().decode(value)); + })(); + `); + expect(stdout).toBe("pong\n"); + }); + + // node:child_process pause() stops the pipe reader (the poll is unregistered) + // and resume() arms it again; the re-armed poll must hold the loop like the + // first one did. The writer starts only once the parent is reading ("s"), + // overfills the paused Readable, and sends "END" after a pause the parent has + // to sit through idle on the pipe with nothing else pending. + it.concurrent("a child_process stdout resumed after pause() keeps the process alive again", async () => { + const writer = ` + const fs = require("fs"); + const chunk = Buffer.alloc(65536, "x"); + const waitForParent = () => fs.readSync(0, Buffer.alloc(1), 0, 1, null); + waitForParent(); + for (let i = 0; i < 16; i++) fs.writeSync(1, chunk); + waitForParent(); + setTimeout(() => fs.writeSync(1, "END"), 100); + `; + const stdout = await run(` + const { spawn } = require("node:child_process"); + const child = spawn(process.execPath, ["-e", ${JSON.stringify(writer)}], { stdio: ["pipe", "pipe", "inherit"] }); + child.unref(); + let received = 0; + let paused = false; + let drained = false; + child.stdout.on("data", chunk => { + received += chunk.length; + if (!paused) { + paused = true; + child.stdout.pause(); + const resumeWhenFull = () => { + if (child.stdout.readableLength >= child.stdout.readableHighWaterMark) { + console.log("resume"); + child.stdout.resume(); + } else { + setImmediate(resumeWhenFull); + } + }; + setImmediate(resumeWhenFull); + } else if (!drained && received >= 16 * 65536) { + drained = true; + child.stdin.end("g"); + } + }); + child.stdout.on("end", () => console.log("end " + received)); + setImmediate(() => child.stdin.write("s")); + `); + expect(stdout).toBe("resume\nend " + (16 * 65536 + 3) + "\n"); + }); +}); + describe("unref() + .exited with nothing else ref'd (Windows)", () => { // Windows: with only an unref'd uv_process_t left, uv_run() used to skip its // body and never dequeue the IOCP exit packet, so these children busy-spun diff --git a/test/js/node/process/process-stdin.test.ts b/test/js/node/process/process-stdin.test.ts index 6aa5d6a528ba..ea1cfffad6d4 100644 --- a/test/js/node/process/process-stdin.test.ts +++ b/test/js/node/process/process-stdin.test.ts @@ -305,6 +305,58 @@ test.concurrent("stdin should not allow process to exit when not paused", async expect(await proc.stderr.text()).toMatchInlineSnapshot(`""`); }); +// unref() drops stdin's hold on the event loop and ref() takes it back: after +// the pair the child has nothing else pending and must still receive every +// later byte up to EOF. +test.concurrent("stdin.ref() after unref() keeps the process alive until EOF", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + let total = 0; + process.stdin.on("data", chunk => { + if (total === 0) { + process.stdin.unref(); + process.stdin.ref(); + process.stdout.write("ready\\n"); + } + total += chunk.length; + }); + process.stdin.on("end", () => { + process.stdout.write("TOTAL " + total + "\\n"); + }); + `, + ], + stdin: "pipe", + stdout: "pipe", + stderr: "pipe", + env: bunEnv, + }); + const reader = proc.stdout.getReader(); + const decoder = new TextDecoder(); + let stdout = ""; + proc.stdin.write("x"); + while (!stdout.includes("ready\n")) { + const { value, done } = await reader.read(); + if (done) break; + stdout += decoder.decode(value); + } + // The child is idle again with only stdin to wait for; these bytes must still reach it. + const rest = Buffer.alloc(64 * 1024, "y"); + proc.stdin.write(rest); + await proc.stdin.end(); + while (true) { + const { value, done } = await reader.read(); + if (done) break; + stdout += decoder.decode(value); + } + const [stderr, exitCode] = await Promise.all([proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe(`ready\nTOTAL ${1 + rest.length}\n`); + expect(exitCode).toBe(0); +}); + test.concurrent("a throw from a 'data' listener is an uncaughtException, and stdin keeps reading", async () => { await using proc = Bun.spawn({ cmd: [ @@ -498,6 +550,70 @@ describe.skipIf(isWindows)("pipe backpressure", () => { expect(deltaMB).toBeLessThan(maxDeltaMB); }); + // Stopped at the backstop, the reader's one-shot poll is left unarmed until + // the next read, so it can observe nothing (not even the writer going away) + // and must not keep the event loop alive on its own. Node behaves the same: + // readStop() at the highWaterMark leaves the handle inactive. + test.concurrent("a reader stopped at the highwater backstop does not keep the process alive", async () => { + const { first } = await run(` + const rd = Bun.stdin.stream().getReader(); + const c = await rd.read(); + process.stdout.write(JSON.stringify({ first: (c.value?.length ?? 0) > 0 })); + // No further read and no exit(): once the backstop engages nothing is pending. + `); + expect(first).toBe(true); + }); + + // The same over an anonymous pipe (the blocking-pipe read path; Bun.spawn + // stdio above is a socketpair). The writer keeps the pipe full until the + // reader is gone, so EOF can never be what lets the reader exit. + test.concurrent("over a shell pipe, a reader stopped at the backstop does not keep the process alive", async () => { + using dir = tempDir("stdin-backstop-pipe", { + "writer.js": ` + const chunk = Buffer.alloc(65536, 0x78); + process.stdout.on("error", () => process.exit(0)); + (function pump() { + while (process.stdout.write(chunk)) {} + process.stdout.once("drain", pump); + })(); + `, + "reader.js": ` + const reader = Bun.stdin.stream().getReader(); + const { value } = await reader.read(); + process.on("exit", () => console.log("EXIT")); + console.log("FIRST " + (value.byteLength > 0)); + `, + }); + const { promise, resolve } = Promise.withResolvers<{ err: Error | null; stdout: string; stderr: string }>(); + exec( + `"${bunExe()}" writer.js | "${bunExe()}" reader.js`, + { cwd: String(dir), env: bunEnv }, + (err, stdout, stderr) => resolve({ err, stdout, stderr }), + ); + expect(await promise).toEqual({ err: null, stdout: "FIRST true\nEXIT\n", stderr: "" }); + }); + + // Stopping unregisters a fired one-shot poll without a syscall, which leaves + // its disarmed registration in the kernel. Cancelling the stream must still + // take it out: fd 0 stays open, and the next poll on it would hit EEXIST. + test.concurrent("a reader cancelled at the highwater backstop frees fd 0 for the next reader", async () => { + const { second } = await run(` + const { getEventLoopStats } = require("bun:internal-for-testing"); + // Not top-level await: an unsettled entry-module promise would keep the process alive by itself. + (async () => { + const reader = Bun.stdin.stream().getReader(); + await reader.read(); + // The parent keeps the pipe full, so the reader soon stops and lets go of the loop. + while (getEventLoopStats().loopActive) await new Promise(resolve => setImmediate(resolve)); + await reader.cancel(); + const next = await Bun.file(0).stream().getReader().read(); + process.stdout.write(JSON.stringify({ second: (next.value?.length ?? 0) > 0 })); + process.exit(0); + })(); + `); + expect(second).toBe(true); + }); + test.concurrent("reading resumes after the highwater backstop", async () => { // Stop reading long enough for the backstop to engage, then drain to EOF // and make sure every byte written by the parent is delivered.