From 65f9ee46dfd516a9fe37a0322b4676b155e90119 Mon Sep 17 00:00:00 2001 From: robobun Date: Sun, 26 Jul 2026 16:12:05 +0000 Subject: [PATCH 1/4] stdin: apply highwater backpressure to the pipe FileReader source FileReader.on_read_chunk exempted pollable fds from the highwater backstop ("for pipes we have to keep pulling"), so with no JS read pending the posix read loop kept draining the fd into self.buffered until EAGAIN. A single Bun.stdin.stream() read followed by an idle consumer ingested the entire pipe; process.stdin.pause() stopped delivery to JS but not the fd ingestion (RSS grew to match the feed). Drop the pollable exemption and fold !flowing into the continue decision. Make the blocking-pipe read path honor that return value (the nonblocking path already did), and have register_poll / on_poll respect IS_PAUSED so a pause issued from inside the read loop's JS re-entry is not immediately undone. on_pull re-arms the poll when it returns Pending, since the one-shot poll may be kernel-disarmed while has_pending_read() still reports the sticky registration flag. --- src/io/PipeReader.rs | 18 +++- src/runtime/webcore/FileReader.rs | 43 ++++---- test/js/node/process/process-stdin.test.ts | 115 ++++++++++++++++++++- 3 files changed, 150 insertions(+), 26 deletions(-) diff --git a/src/io/PipeReader.rs b/src/io/PipeReader.rs index e073265eb2dd..dd2238b046ff 100644 --- a/src/io/PipeReader.rs +++ b/src/io/PipeReader.rs @@ -389,6 +389,13 @@ impl PosixBufferedReader { /// embedding `self` (the shell `PipeReader` does exactly that), so the /// caller must not touch `self` again after a `false` return. pub fn register_poll(&mut self) -> bool { + // A pause issued from inside the JS re-entry of the read loop + // (on_read_chunk → microtasks → setFlowing(false) → pause()) lands + // while the loop's own `register_poll()` is still ahead on the stack; + // re-arming here would undo it. `unpause()` + `read()` re-arm. + if self.flags.contains(PosixFlags::IS_PAUSED) { + return true; + } // Hoist vtable-derived scalars and // normalize self.handle to Poll before taking the single &mut borrow, // so no raw-pointer escape is needed. @@ -503,6 +510,9 @@ impl PosixBufferedReader { } pub fn on_poll(parent: &mut PosixBufferedReader, size_hint: isize, received_hup: bool) { + if parent.flags.contains(PosixFlags::IS_PAUSED) { + return; + } let fd = parent.get_fd(); bun_sys::syslog!("onPoll({}) = {}", fd, size_hint); @@ -662,14 +672,18 @@ impl PosixBufferedReader { if streaming { // Stream this chunk and register for next cycle - let _ = parent.vtable.on_read_chunk( + if !parent.vtable.on_read_chunk( &stack_buffer[..bytes_read], if received_hup && bytes_read < stack_buffer.len() { ReadState::Eof } else { ReadState::Progress }, - ); + ) && !received_hup + && !over_budget + { + return; + } } else { parent ._buffer diff --git a/src/runtime/webcore/FileReader.rs b/src/runtime/webcore/FileReader.rs index da85a4654f4e..8553c355e28b 100644 --- a/src/runtime/webcore/FileReader.rs +++ b/src/runtime/webcore/FileReader.rs @@ -563,22 +563,6 @@ impl FileReader { true } - #[inline] - fn reader_is_pollable(&self) -> bool { - #[cfg(unix)] - { - self.reader() - .flags - .contains(bun_io::pipe_reader::PosixFlags::POLLABLE) - } - #[cfg(windows)] - { - self.reader() - .flags - .contains(bun_io::pipe_reader::WindowsFlags::POLLABLE) - } - } - pub fn on_read_chunk(&self, init_buf: &[u8], state: ReadState) -> bool { let mut buf = init_buf; bun_core::scoped_log!( @@ -809,14 +793,19 @@ impl FileReader { } } - // For pipes, we have to keep pulling or the other process will block. + // No JS read is waiting; the chunk was appended to `self.buffered`. + // Stop the read loop once the buffered bytes reach the highwater mark + // or the consumer paused (`setFlowing(false)`). A full kernel pipe + // buffer blocking the writer is the backpressure signal. `onPull` / + // `setFlowing(true)` re-derive demand and restart the reader. // SAFETY: see `reader_buffer` decl. let reader_buffer_len = unsafe { (*reader_buffer).len() }; - let ret = !matches!( - self.read_inside_on_pull.get(), - ReadDuringJSOnPullResult::Temporary(_) - ) && !(self.buffered.get().len() + reader_buffer_len >= self.highwater_mark - && !self.reader_is_pollable()); + let ret = self.flowing.get() + && !matches!( + self.read_inside_on_pull.get(), + ReadDuringJSOnPullResult::Temporary(_) + ) + && self.buffered.get().len() + reader_buffer_len < self.highwater_mark; close_if_needed!(); ret } @@ -965,6 +954,16 @@ impl FileReader { self.pending_value.with_mut(|p| p.set(&global, array)); self.pending_view.set(buffer); + // The read loop returns without re-arming the one-shot poll when + // `on_read_chunk` reports the highwater backstop, and pause() skips + // the unregister while the poll is already disarmed, so + // `has_pending_read()` (which tracks the registration flag, not the + // kernel arm state) can be true with no wakeup scheduled. The + // consumer is asking for data here, so make sure the poll is armed. + if self.flowing.get() { + self.reader().watch(); + } + bun_core::scoped_log!(FileReader, "onPull({}) = pending", buffer_len); streams::Result::Pending(self.pending.as_ptr()) diff --git a/test/js/node/process/process-stdin.test.ts b/test/js/node/process/process-stdin.test.ts index 6a64a4ffd136..c3e26ade88ed 100644 --- a/test/js/node/process/process-stdin.test.ts +++ b/test/js/node/process/process-stdin.test.ts @@ -1,5 +1,5 @@ -import { expect, test } from "bun:test"; -import { bunEnv, bunExe } from "harness"; +import { describe, expect, test } from "bun:test"; +import { bunEnv, bunExe, isASAN, isDebug, isWindows } from "harness"; test.concurrent("pipe does the right thing", async () => { // Note: Bun.spawnSync uses memfd_create on Linux for pipe, which means we see @@ -418,3 +418,114 @@ test.concurrent("pause() and resume() churn while data is in flight never destro expect(stdout.trim()).toBe(`TOTAL ${20 * 1024}`); expect(exitCode).toBe(0); }); + +// The native FileReader source over a pollable pipe used to drain the fd to +// EAGAIN regardless of JS demand, so an idle consumer still ingested the whole +// pipe into an internal buffer. The kernel pipe buffer filling up is the +// backpressure signal; these tests feed far more than that and check the +// child's resident set does not grow to match. +describe.skipIf(isWindows)("pipe backpressure", () => { + const feedMB = 40; + // With no backpressure the child buffers the whole feed (Vec growth roughly + // doubles that in RSS). With backpressure only the highwater mark plus the + // kernel pipe buffer are resident in the child. + const maxDeltaMB = isASAN || isDebug ? 24 : 16; + + async function run(consumer: string) { + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", consumer], + stdin: "pipe", + stdout: "pipe", + stderr: "pipe", + env: bunEnv, + }); + // The child exits under backpressure having accepted only a few KB, so + // the queued writes here fail with EPIPE; that is the expected outcome. + const chunk = Buffer.alloc(1024 * 1024, 0x78); + const ignoreEpipe = (e: any) => { + if (e?.code !== "EPIPE") throw e; + }; + for (let i = 0; i < feedMB; i++) { + const r = proc.stdin.write(chunk); + if (r && typeof (r as any).then === "function") (r as Promise).catch(ignoreEpipe); + } + Promise.resolve(proc.stdin.flush()).catch(ignoreEpipe); + Promise.resolve(proc.stdin.end()).catch(ignoreEpipe); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + const result = JSON.parse(stdout.trim()); + expect(exitCode).toBe(0); + return result; + } + + test.concurrent("Bun.stdin.stream(): a single read does not ingest the whole pipe", async () => { + const { first, deltaMB } = await run(` + const rd = Bun.stdin.stream().getReader(); + const c = await rd.read(); + const base = process.memoryUsage().rss; + // Give the event loop time to (incorrectly) drain the pipe. The loop is + // native (no JS on the no-pending path), so debug/ASAN overhead is small. + await new Promise(r => setTimeout(r, 1500)); + const deltaMB = Math.round((process.memoryUsage().rss - base) / 1048576); + process.stdout.write(JSON.stringify({ first: c.value?.length ?? 0, deltaMB })); + process.exit(0); + `); + expect(first).toBeGreaterThan(0); + expect(deltaMB).toBeLessThan(maxDeltaMB); + }); + + test.concurrent("process.stdin.pause() stops the fd from being read", async () => { + const { bytesAfter, deltaMB } = await run(` + let bytes = 0, pausedAt = 0; + process.stdin.on("data", chunk => { + bytes += chunk.length; + if (!pausedAt && bytes >= 1 << 20) { + pausedAt = bytes; + process.stdin.pause(); + const base = process.memoryUsage().rss; + setTimeout(() => { + const deltaMB = Math.round((process.memoryUsage().rss - base) / 1048576); + process.stdout.write(JSON.stringify({ bytesAtPause: pausedAt, bytesAfter: bytes, deltaMB })); + process.exit(0); + }, 1500); + } + }); + `); + expect(bytesAfter).toBeLessThan(feedMB * 1024 * 1024); + expect(deltaMB).toBeLessThan(maxDeltaMB); + }); + + 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. + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + const rd = Bun.stdin.stream().getReader(); + await rd.read().then(c => { globalThis.total = c.value?.length ?? 0; }); + await new Promise(r => setTimeout(r, 200)); + while (true) { + const { value, done } = await rd.read(); + if (value) total += value.length; + if (done) break; + } + process.stdout.write(String(total)); + `, + ], + stdin: "pipe", + stdout: "pipe", + stderr: "pipe", + env: bunEnv, + }); + const chunk = Buffer.alloc(64 * 1024, 0x79); + const n = 64; + for (let i = 0; i < n; i++) proc.stdin.write(chunk); + await proc.stdin.end(); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe(String(n * chunk.length)); + expect(exitCode).toBe(0); + }); +}); From 6100b0464a653d653be8270e46a8377b1fe57c8d Mon Sep 17 00:00:00 2001 From: robobun Date: Sun, 26 Jul 2026 16:50:01 +0000 Subject: [PATCH 2/4] io(posix): short-circuit register_poll when the poll is already armed on_pull's watch() re-arm and the read loop's own register_poll both land on the same one-shot poll per chunk in the streaming path, so register_with_fd was issuing an idempotent epoll_ctl(CTL_MOD) / kevent64 twice. Skip it when is_watching() already holds. --- src/io/PipeReader.rs | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/io/PipeReader.rs b/src/io/PipeReader.rs index dd2238b046ff..bde772301b6b 100644 --- a/src/io/PipeReader.rs +++ b/src/io/PipeReader.rs @@ -416,6 +416,11 @@ impl PosixBufferedReader { let Some(poll) = self.handle.get_poll_mut() else { return true; }; + if poll.is_watching() { + // Already armed kernel-side; a redundant `register_with_fd` + // issues an idempotent `epoll_ctl(CTL_MOD)` / `kevent64`. + return true; + } poll.set_owner(Owner::new(PollTag::BufferedReader, owner_ptr.cast())); if !poll.has_flag(FilePollFlag::WasEverRegistered) From be00ab70ccd2ae0b8fe36a7644c4de9b1c017bc3 Mon Sep 17 00:00:00 2001 From: robobun Date: Sun, 26 Jul 2026 16:52:07 +0000 Subject: [PATCH 3/4] trim code comments --- src/io/PipeReader.rs | 8 ++------ src/runtime/webcore/FileReader.rs | 14 +++----------- 2 files changed, 5 insertions(+), 17 deletions(-) diff --git a/src/io/PipeReader.rs b/src/io/PipeReader.rs index bde772301b6b..707292abe2f6 100644 --- a/src/io/PipeReader.rs +++ b/src/io/PipeReader.rs @@ -389,10 +389,8 @@ impl PosixBufferedReader { /// embedding `self` (the shell `PipeReader` does exactly that), so the /// caller must not touch `self` again after a `false` return. pub fn register_poll(&mut self) -> bool { - // A pause issued from inside the JS re-entry of the read loop - // (on_read_chunk → microtasks → setFlowing(false) → pause()) lands - // while the loop's own `register_poll()` is still ahead on the stack; - // re-arming here would undo it. `unpause()` + `read()` re-arm. + // pause() may land from inside on_read_chunk's JS re-entry while the + // loop's own re-arm is still ahead on the stack. if self.flags.contains(PosixFlags::IS_PAUSED) { return true; } @@ -417,8 +415,6 @@ impl PosixBufferedReader { return true; }; if poll.is_watching() { - // Already armed kernel-side; a redundant `register_with_fd` - // issues an idempotent `epoll_ctl(CTL_MOD)` / `kevent64`. return true; } poll.set_owner(Owner::new(PollTag::BufferedReader, owner_ptr.cast())); diff --git a/src/runtime/webcore/FileReader.rs b/src/runtime/webcore/FileReader.rs index 8553c355e28b..b205c96b724c 100644 --- a/src/runtime/webcore/FileReader.rs +++ b/src/runtime/webcore/FileReader.rs @@ -793,11 +793,7 @@ impl FileReader { } } - // No JS read is waiting; the chunk was appended to `self.buffered`. - // Stop the read loop once the buffered bytes reach the highwater mark - // or the consumer paused (`setFlowing(false)`). A full kernel pipe - // buffer blocking the writer is the backpressure signal. `onPull` / - // `setFlowing(true)` re-derive demand and restart the reader. + // No JS read is waiting; stop at the highwater mark. onPull restarts. // SAFETY: see `reader_buffer` decl. let reader_buffer_len = unsafe { (*reader_buffer).len() }; let ret = self.flowing.get() @@ -954,12 +950,8 @@ impl FileReader { self.pending_value.with_mut(|p| p.set(&global, array)); self.pending_view.set(buffer); - // The read loop returns without re-arming the one-shot poll when - // `on_read_chunk` reports the highwater backstop, and pause() skips - // the unregister while the poll is already disarmed, so - // `has_pending_read()` (which tracks the registration flag, not the - // kernel arm state) can be true with no wakeup scheduled. The - // consumer is asking for data here, so make sure the poll is armed. + // `has_pending_read()` tracks the registration flag, not the kernel + // arm state: the highwater backstop leaves the one-shot poll disarmed. if self.flowing.get() { self.reader().watch(); } From a772e2f50582f2ba40685adeb8113d2322b2f8fd Mon Sep 17 00:00:00 2001 From: robobun Date: Sun, 26 Jul 2026 17:09:40 +0000 Subject: [PATCH 4/4] io(posix): move the already-armed short-circuit from register_poll to watch register_poll is called from the read loop's EAGAIN path, whose epoll_ctl count the shell-pipe-read-fault tests fault-inject against. watch() is the setup/restart path (FileReader.on_pull, subprocess bring-up); skipping the redundant epoll_ctl there covers the per-chunk on_pull re-arm without shifting the read loop's call pattern. --- src/io/PipeReader.rs | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/src/io/PipeReader.rs b/src/io/PipeReader.rs index 707292abe2f6..457d7f49b40e 100644 --- a/src/io/PipeReader.rs +++ b/src/io/PipeReader.rs @@ -414,9 +414,6 @@ impl PosixBufferedReader { let Some(poll) = self.handle.get_poll_mut() else { return true; }; - if poll.is_watching() { - return true; - } poll.set_owner(Owner::new(PollTag::BufferedReader, owner_ptr.cast())); if !poll.has_flag(FilePollFlag::WasEverRegistered) @@ -465,7 +462,9 @@ impl PosixBufferedReader { } pub fn watch(&mut self) { - if self.flags.contains(PosixFlags::POLLABLE) { + if self.flags.contains(PosixFlags::POLLABLE) + && !matches!(&self.handle, PollOrFd::Poll(poll) if poll.is_watching()) + { self.register_poll(); } }