diff --git a/src/io/PipeReader.rs b/src/io/PipeReader.rs index e073265eb2dd..457d7f49b40e 100644 --- a/src/io/PipeReader.rs +++ b/src/io/PipeReader.rs @@ -389,6 +389,11 @@ 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 { + // 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; + } // Hoist vtable-derived scalars and // normalize self.handle to Poll before taking the single &mut borrow, // so no raw-pointer escape is needed. @@ -457,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(); } } @@ -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..b205c96b724c 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,15 @@ impl FileReader { } } - // For pipes, we have to keep pulling or the other process will block. + // 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 = !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 +950,12 @@ impl FileReader { self.pending_value.with_mut(|p| p.set(&global, array)); self.pending_view.set(buffer); + // `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(); + } + 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); + }); +});