From 91540f52835880e31ae17221a8b98395bb3425a8 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 24 Oct 2026 23:42:42 +0000 Subject: [PATCH 1/3] Bun.write(file, file): wait for a non-blocking fd instead of rejecting with EAGAIN The copy loops behind Bun.write(file, file) treated EAGAIN as a failure. A pipe or socket on fd 1 is non-blocking once process.stdout is used, so `await Bun.write(Bun.stdout, Bun.file(path))` rejected after a part of the file. On EAGAIN the copy now waits until the fd is ready and repeats the same syscall: - bun_sys::wait_until_readable / wait_until_writable block in poll, and in select on macOS, where poll does not wake when the other end of a named pipe closes. write_all_to_stdio uses them too. - do_copy_file_range (sendfile, splice) asks poll which end is not ready, then waits for it. A destination that reports an error gets one retry, for the syscall to report it. After that only the source ends the wait, because splice does not look at the destination while the source is empty. - read_write_loop_capped waits for the fd whose read or write failed. --- src/runtime/webcore/blob/copy_file.rs | 80 +++++- src/sys/lib.rs | 90 +++++- test/js/bun/io/bun-write.test.js | 384 ++++++++++++++++++++++++++ 3 files changed, 541 insertions(+), 13 deletions(-) diff --git a/src/runtime/webcore/blob/copy_file.rs b/src/runtime/webcore/blob/copy_file.rs index e41579856cdd..e207c18fccee 100644 --- a/src/runtime/webcore/blob/copy_file.rs +++ b/src/runtime/webcore/blob/copy_file.rs @@ -368,6 +368,7 @@ impl CopyFile { } let mut has_unset_append = false; + let mut retried_for_dest_error = false; // If they can't use copy_file_range, they probably also can't // use sendfile() or splice() @@ -474,6 +475,13 @@ impl CopyFile { ); return Err(bun_errno::from_errno(bun_sys::E::EINVAL as i32).into()); } + // EAGAIN: a pipe or socket end is non-blocking and not ready. Two regular files never are, so copy_file_range does not wait. + bun_sys::E::EAGAIN + if !matches!(USE, TryWith::CopyFileRange) + && wait_for_kernel_copy(src_fd, dest_fd, &mut retried_for_dest_error) => + { + continue; + } errno => { self.system_error = Some( bun_sys::Error { @@ -948,6 +956,48 @@ impl CopyFile { } } +/// After EAGAIN from sendfile or splice, waits for the end that is not ready. The errno does not say which end. `false` if poll fails. +#[cfg(any(target_os = "linux", target_os = "android"))] +#[cold] +fn wait_for_kernel_copy(src_fd: Fd, dest_fd: Fd, retried_for_dest_error: &mut bool) -> bool { + use bun_sys::posix::{POLL_IN, POLL_OUT, PollFd, poll}; + let mut ends = [ + PollFd { + fd: dest_fd.native(), + events: POLL_OUT, + revents: 0, + }, + PollFd { + fd: src_fd.native(), + events: POLL_IN, + revents: 0, + }, + ]; + if core::mem::take(retried_for_dest_error) { + // The retry did not report the destination's error: splice looks at the source first. poll skips a negative fd, so only the source ends this wait. + ends[0].fd = -1; + return poll(&mut ends, -1).is_ok(); + } + if poll(&mut ends, -1).is_err() { + return false; + } + let [dest, src] = &mut ends; + if dest.revents & (libc::POLLERR | libc::POLLHUP | libc::POLLNVAL) != 0 { + // One retry lets a syscall that looks at the destination first (sendfile) report the error. + *retried_for_dest_error = true; + return true; + } + if dest.revents == 0 { + src.fd = -1; + } else if src.revents == 0 { + // With no events, poll still reports a destination that closes or fails. + dest.events = 0; + } else { + return true; + } + poll(&mut ends, -1).is_ok() +} + #[cfg(any(target_os = "linux", target_os = "android"))] fn read_write_fallback( src_fd: Fd, @@ -973,6 +1023,21 @@ fn read_write_fallback( } } +/// After EAGAIN, waits until `fd` is ready. Any other error, or a wait that fails, ends the copy with `err`. +#[cfg(not(windows))] +#[cold] +fn wait_if_would_block( + err: bun_sys::Error, + wait: fn(Fd) -> bun_sys::Result<()>, + fd: Fd, +) -> bun_sys::Result<()> { + if err.is_retry() && wait(fd).is_ok() { + Ok(()) + } else { + Err(err) + } +} + #[inline(never)] // 64 KB stack buffer #[cfg(not(windows))] fn read_write_loop_capped( @@ -987,19 +1052,26 @@ fn read_write_loop_capped( let mut remaining = cap; while remaining > 0 { let want = (buf.len() as SizeType).min(remaining) as usize; - let amt = bun_sys::read(src_fd, &mut buf[..want])?; + let amt = match bun_sys::read(src_fd, &mut buf[..want]) { + Ok(amt) => amt, + Err(err) => { + wait_if_would_block(err, bun_sys::wait_until_readable, src_fd)?; + continue; + } + }; if amt == 0 { break; } remaining -= amt as SizeType; let mut slice = &buf[..amt]; while !slice.is_empty() { - match bun_sys::write(dest_fd, slice)? { - 0 => return Ok(()), - n => { + match bun_sys::write(dest_fd, slice) { + Ok(0) => return Ok(()), + Ok(n) => { *total += n as u64; slice = &slice[n..]; } + Err(err) => wait_if_would_block(err, bun_sys::wait_until_writable, dest_fd)?, } } } diff --git a/src/sys/lib.rs b/src/sys/lib.rs index 14ee170f6bb6..8b52521300e1 100644 --- a/src/sys/lib.rs +++ b/src/sys/lib.rs @@ -1672,6 +1672,15 @@ mod nocancel { ) -> isize; #[link_name = "poll$NOCANCEL"] pub(crate) fn poll(fds: *mut libc::pollfd, nfds: libc::nfds_t, timeout: c_int) -> c_int; + // The `DARWIN_EXTSN` variant has no `FD_SETSIZE` limit: a set is `nfds` bits, in 32-bit words. + #[link_name = "select$DARWIN_EXTSN$NOCANCEL"] + pub(crate) fn select( + nfds: c_int, + readfds: *mut u32, + writefds: *mut u32, + errorfds: *mut u32, + timeout: *mut libc::timeval, + ) -> c_int; // Remaining `$NOCANCEL` variants Bun links against. // safe: by-value `c_int` fd; bad fd → -1/EBADF, no UB. #[link_name = "close$NOCANCEL"] @@ -9312,22 +9321,85 @@ fn qw_set_fd(qw: &mut bun_core::output::QuietWriter, fd: Fd) { } } +/// Blocks the thread until a `read` of `fd` no longer fails with EAGAIN: there is data, end of file, or an error that the `read` then reports. +#[cfg(unix)] +#[cold] +pub fn wait_until_readable(fd: Fd) -> Maybe<()> { + wait_until(fd, posix::POLL_IN) +} + +/// Blocks the thread until a `write` to `fd` no longer fails with EAGAIN: there is room, or an error (EPIPE) that the `write` then reports. +#[cfg(unix)] +#[cold] +pub fn wait_until_writable(fd: Fd) -> Maybe<()> { + wait_until(fd, posix::POLL_OUT) +} + +#[cfg(all(unix, not(target_os = "macos")))] +fn wait_until(fd: Fd, events: i16) -> Maybe<()> { + let mut pfd = [posix::PollFd { + fd: fd.native(), + events, + revents: 0, + }]; + posix::poll(&mut pfd, -1)?; + Ok(()) +} + +/// `select`, because `poll` and kqueue on macOS do not wake when the other end of a named pipe closes. +#[cfg(target_os = "macos")] +fn wait_until(fd: Fd, events: i16) -> Maybe<()> { + const WORD_BITS: usize = u32::BITS as usize; + let Ok(bit) = usize::try_from(fd.native()) else { + return Err(Error::from_code(E::EBADF, Tag::poll)); + }; + let word = bit / WORD_BITS; + // Stdout and stderr are waited for on the crash path, so the usual fds do not allocate. + let mut inline_set = [0u32; libc::FD_SETSIZE / WORD_BITS]; + let mut heap_set; + let set: &mut [u32] = if word < inline_set.len() { + &mut inline_set + } else { + heap_set = vec![0u32; word + 1]; + &mut heap_set + }; + loop { + set[word] = 1 << (bit % WORD_BITS); + let bits = set.as_mut_ptr(); + let (readfds, writefds) = if events == posix::POLL_IN { + (bits, core::ptr::null_mut()) + } else { + (core::ptr::null_mut(), bits) + }; + // SAFETY: `set` holds the `nfds` bits, rounded up to a word, that the kernel reads and writes back. The other sets and the timeout (none: wait) can be null. + let rc = unsafe { + nocancel::select( + fd.native() + 1, + readfds, + writefds, + core::ptr::null_mut(), + core::ptr::null_mut(), + ) + }; + if rc >= 0 { + return Ok(()); + } + let errno = last_errno(); + if errno != libc::EINTR { + return Err(Error::from_code_int(errno, Tag::poll)); + } + } +} + /// Writes all of `bytes` to stdout or stderr, for a caller that has to know whether they got there. `Output` doesn't say. pub fn write_all_to_stdio(fd: Fd, mut bytes: &[u8]) -> Maybe<()> { while !bytes.is_empty() { match write(fd, bytes) { Ok(0) => return Err(Error::from_code(E::EIO, Tag::write)), Ok(n) => bytes = &bytes[n..], + // fd 1/2 are O_NONBLOCK once process.stdout/stderr exist (as in Node); wait instead of dropping output. #[cfg(unix)] - Err(e) if e.get_errno() == E::EAGAIN => { - // fd 1/2 are O_NONBLOCK once process.stdout/stderr exist (as in Node); wait instead of dropping output. - let mut pfd = [posix::PollFd { - fd: fd.native(), - events: posix::POLL_OUT, - revents: 0, - }]; - posix::poll(&mut pfd, -1)?; - } + Err(e) if e.get_errno() == E::EAGAIN => wait_until_writable(fd)?, Err(e) => return Err(e), } } diff --git a/test/js/bun/io/bun-write.test.js b/test/js/bun/io/bun-write.test.js index bcbd9deef142..01b9f00b4066 100644 --- a/test/js/bun/io/bun-write.test.js +++ b/test/js/bun/io/bun-write.test.js @@ -7,6 +7,7 @@ import { exampleSite, gcTick, isASAN, + isLinux, isMacOS, isWindows, tempDir, @@ -1784,3 +1785,386 @@ int posix_fadvise(int fd, off_t offset, off_t len, int advice) { expect(exitCode).toBe(0); }); }); + +// Bun.write(file, file) copies on a pool thread with sendfile, splice or a read/write loop. On a +// non-blocking fd those fail with EAGAIN where a blocking fd waits, and the copy has to wait too. +// Not on Windows: nothing there makes a stdio handle non-blocking, and the copy is other code. +// +// Each child reports on stderr, one JSON line each: { ready } when the copy has started, then how +// the promise settled. Bun.spawn gives a child sockets, so a child that needs a pipe runs in an +// sh pipeline. +(isWindows ? describe.skip : describe.concurrent)("Bun.write(file, file) on a non-blocking fd", () => { + const size = 1024 * 1024; + const pattern = Buffer.alloc(size, Buffer.from(Array.from({ length: 251 }, (_, i) => i))); + + // fs.promises.stat is a pool job like the copy. The first one starts a pool thread. The second + // one is queued after the copy, which gives the copy time to reach its first syscall before the + // test hears { ready }. A copy that waits passes in either order. A copy that does not wait + // fails only if it meets the fd before the test changes it. + const child = (setup, destination, source, meanwhile = "") => ` + const fs = require("fs"); + let prefilled = 0; + ${setup} + await fs.promises.stat("."); + const outcome = Bun.write(${destination}, ${source}).then( + n => ({ resolved: n }), + e => ({ rejected: e.code + " " + e.syscall }), + ); + await fs.promises.stat("."); + process.stderr.write(JSON.stringify({ ready: prefilled }) + "\\n"); + ${meanwhile} + process.stderr.write(JSON.stringify(await outcome) + "\\n"); + `; + + // The first use of process.stdout makes a pipe or socket on fd 1 non-blocking. The loop then + // fills it: nothing more fits until something reads. + const fillStdout = ` + void process.stdout; + const filler = Buffer.alloc(4096, "-"); + for (;;) { + try { + prefilled += fs.writeSync(1, filler); + } catch (e) { + if (e.code !== "EAGAIN") throw e; + break; + } + } + `; + + const sh = (pipeline, script, env = {}) => ({ + cmd: ["sh", "-c", pipeline], + env: { ...bunEnv, ...env, BUN: bunExe(), SCRIPT: script }, + stdin: "pipe", + stdout: "pipe", + stderr: "pipe", + }); + // The reader of the child's stdout starts only when the test sends a line on sh's stdin (fd 3). + const gatedReader = `{ read -r go <&3; exec cat; }`; + async function releaseReader(proc) { + proc.stdin.write("go\n"); + await proc.stdin.flush(); + } + + // Reads a stream as it arrives. `until` resolves when `done(text so far)` holds or the stream ends. + function collect(stream) { + const reader = stream.getReader(); + const decoder = new TextDecoder(); + let text = ""; + let pending; + const more = () => + (pending ??= reader.read().then(({ value, done }) => { + pending = undefined; + if (!done) text += decoder.decode(value, { stream: true }); + return !done; + })); + return { + async until(done) { + while (!done(text)) if (!(await more())) break; + return text; + }, + async all() { + while (await more()) {} + return text; + }, + }; + } + + const reports = text => + text + .split("\n") + .filter(Boolean) + .map(line => { + try { + return JSON.parse(line); + } catch { + return line; + } + }); + const reported = count => text => text.endsWith("\n") && reports(text).length >= count; + + // Resolves with the number of bytes the child put into fd 1 before the copy. + async function ready(stderr) { + const text = await stderr.until(reported(1)); + const [first] = reports(text); + if (typeof first?.ready !== "number") throw new Error("the child did not start the copy: " + text); + return first.ready; + } + + function mkfifo(dir, name) { + const fifo = join(String(dir), name); + expect(Bun.spawnSync({ cmd: ["mkfifo", fifo] }).exitCode).toBe(0); + return fifo; + } + + // Sends "hel", then "lo\n" only after "hel" came out of the copy, so the copy finds its source + // empty part-way. A copy that has already settled ends the wait too. + async function feedInTwoPieces(send, arrived, stderr) { + await send("hel"); + await Promise.race([arrived("hel"), stderr.until(reported(2))]); + await send("lo\n"); + await Promise.race([arrived("hello\n"), stderr.until(reported(2))]); + } + + function expectWholeFile(bytes, err, prefilled) { + expect({ + piped: bytes.length, + intact: pattern.equals(bytes.subarray(prefilled)), + reports: reports(err), + }).toEqual({ + piped: prefilled + size, + intact: true, + reports: [{ ready: prefilled }, { resolved: size }], + }); + } + + async function expectWholeFileThroughPipe(proc) { + const stderr = collect(proc.stderr); + const prefilled = await ready(stderr); + expect(prefilled).toBeGreaterThan(0); + await releaseReader(proc); + const [bytes, err, exitCode] = await Promise.all([proc.stdout.bytes(), stderr.all(), proc.exited]); + expectWholeFile(bytes, err, prefilled); + expect(exitCode).toBe(0); + } + + it("a file to a full stdout pipe", async () => { + using dir = tempDir("bun-write-nonblocking-pipe", { "source.bin": pattern }); + const source = `Bun.file(${JSON.stringify(join(String(dir), "source.bin"))})`; + await using proc = Bun.spawn( + sh(`exec 3<&0; "$BUN" -e "$SCRIPT" | ${gatedReader}`, child(fillStdout, "Bun.stdout", source)), + ); + await expectWholeFileThroughPipe(proc); + }); + + // Linux knows the type of fd 1 and uses sendfile for it. It knows nothing about fd 3, tries + // copy_file_range, and falls back to the read/write loop, which is the only loop on macOS. + it("a file to a full pipe on Bun.file(3)", async () => { + using dir = tempDir("bun-write-nonblocking-fd3", { "source.bin": pattern }); + const source = `Bun.file(${JSON.stringify(join(String(dir), "source.bin"))})`; + await using proc = Bun.spawn( + sh(`exec 3<&0; "$BUN" -e "$SCRIPT" 3>&1 | ${gatedReader}`, child(fillStdout, "Bun.file(3)", source)), + ); + await expectWholeFileThroughPipe(proc); + }); + + it("a stdin pipe to a full stdout pipe", async () => { + using dir = tempDir("bun-write-nonblocking-pipes", { "source.bin": pattern }); + await using proc = Bun.spawn( + sh( + `exec 3<&0; cat "$SOURCE" | "$BUN" -e "$SCRIPT" | ${gatedReader}`, + child(fillStdout, "Bun.stdout", "Bun.stdin"), + { + SOURCE: join(String(dir), "source.bin"), + }, + ), + ); + await expectWholeFileThroughPipe(proc); + }); + + // A splice between two pipes does not block if either is non-blocking, so an empty stdin fails it too. + it("a stdin pipe that has no data yet, once fd 1 is non-blocking", async () => { + // The first stage turns two lines from the test into the two pieces. + const pieces = `{ read -r a; printf '%s' "$a"; read -r b; printf '%s\\n' "$b"; }`; + await using proc = Bun.spawn( + sh(`${pieces} | "$BUN" -e "$SCRIPT" | cat`, child("void process.stdout;", "Bun.stdout", "Bun.stdin")), + ); + const [stdout, stderr] = [collect(proc.stdout), collect(proc.stderr)]; + await ready(stderr); + await feedInTwoPieces( + async piece => { + proc.stdin.write(piece.trimEnd() + "\n"); + await proc.stdin.flush(); + }, + text => stdout.until(out => out.includes(text)), + stderr, + ); + const [out, err, exitCode] = await Promise.all([stdout.all(), stderr.all(), proc.exited]); + expect({ stdout: out, reports: reports(err) }).toEqual({ + stdout: "hello\n", + reports: [{ ready: 0 }, { resolved: 6 }], + }); + expect(exitCode).toBe(0); + }); + + it("a file to a full stdout socket", async () => { + using dir = tempDir("bun-write-nonblocking-socket", { "source.bin": pattern }); + const source = `Bun.file(${JSON.stringify(join(String(dir), "source.bin"))})`; + // lazy: nothing reads the child's stdout until the test asks for it. + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", child(fillStdout, "Bun.stdout", source)], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + lazy: true, + }); + const stderr = collect(proc.stderr); + const prefilled = await ready(stderr); + expect(prefilled).toBeGreaterThan(0); + const [bytes, err, exitCode] = await Promise.all([proc.stdout.bytes(), stderr.all(), proc.exited]); + expectWholeFile(bytes, err, prefilled); + expect(exitCode).toBe(0); + }); + + // The fd is past FD_SETSIZE, where the select() that macOS waits with needs a longer fd set. + // ulimit: a process on macOS can start with fewer than 1024 fds. + it("a non-blocking FIFO source on a high fd ends the copy when its writer closes", async () => { + using dir = tempDir("bun-write-nonblocking-fifo-source", {}); + const fifo = mkfifo(dir, "source.fifo"); + // A reader has to exist before the writer can open, and a writer before the child reads. + const reader = fs.openSync(fifo, fs.constants.O_RDONLY | fs.constants.O_NONBLOCK); + const writer = fs.openSync(fifo, fs.constants.O_WRONLY); + let writerOpen = true; + try { + const open = ` + while (fs.openSync("/dev/null", "r") < 1023) {} + const source = fs.openSync(${JSON.stringify(fifo)}, fs.constants.O_RDONLY | fs.constants.O_NONBLOCK); + if (source < 1024) throw new Error("the FIFO is on fd " + source); + `; + await using proc = Bun.spawn( + sh(`ulimit -n 2048 2>/dev/null; "$BUN" -e "$SCRIPT" | cat`, child(open, "Bun.stdout", "Bun.file(source)")), + ); + const [stdout, stderr] = [collect(proc.stdout), collect(proc.stderr)]; + await ready(stderr); + await feedInTwoPieces( + async piece => void fs.writeSync(writer, piece), + text => stdout.until(out => out.includes(text)), + stderr, + ); + // The copy now waits for more. The end of the file has to wake it. + fs.closeSync(writer); + writerOpen = false; + const [out, err, exitCode] = await Promise.all([stdout.all(), stderr.all(), proc.exited]); + expect({ stdout: out, reports: reports(err) }).toEqual({ + stdout: "hello\n", + reports: [{ ready: 0 }, { resolved: 6 }], + }); + expect(exitCode).toBe(0); + } finally { + fs.closeSync(reader); + if (writerOpen) fs.closeSync(writer); + } + }); + + it("a reader that takes a part of a full FIFO and leaves fails the copy with EPIPE", async () => { + // Larger than a pipe on a kernel with 64 KiB pages, so the copy cannot finish before the reader leaves. + const big = Buffer.alloc(4 * size, pattern); + using dir = tempDir("bun-write-nonblocking-fifo-dest", { "source.bin": big }); + const source = `Bun.file(${JSON.stringify(join(String(dir), "source.bin"))})`; + const fifo = mkfifo(dir, "dest.fifo"); + const reader = fs.openSync(fifo, fs.constants.O_RDONLY | fs.constants.O_NONBLOCK); + const writer = fs.openSync(fifo, fs.constants.O_WRONLY); + const open = new Set([reader, writer]); + const close = fd => open.delete(fd) && fs.closeSync(fd); + try { + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", child(fillStdout, "Bun.stdout", source)], + env: bunEnv, + stdout: writer, + stderr: "pipe", + }); + close(writer); + const stderr = collect(proc.stderr); + const prefilled = await ready(stderr); + expect(prefilled).toBeGreaterThan(0); + // The reader takes what was in the FIFO and 8 blocks of the file, then leaves. It starts a + // process for each block, so the copy is much faster and finds the FIFO full after each one. + // Bun.spawn makes the stdin that it passes on blocking. + const blocks = Math.ceil(prefilled / 4096) + 8; + await using slowReader = Bun.spawn({ + cmd: ["sh", "-c", `i=0; while [ $i -lt ${blocks} ]; do dd bs=4096 count=1 2>/dev/null; i=$((i + 1)); done`], + env: bunEnv, + stdin: reader, + stdout: "pipe", + stderr: "inherit", + }); + close(reader); + const [taken, err, exitCode] = await Promise.all([slowReader.stdout.bytes(), stderr.all(), proc.exited]); + const copied = taken.subarray(prefilled); + expect({ + tookPartOfTheFile: copied.length > 0, + intact: big.subarray(0, copied.length).equals(copied), + reports: reports(err), + }).toEqual({ + tookPartOfTheFile: true, + intact: true, + reports: [{ ready: prefilled }, { rejected: isLinux ? "EPIPE sendfile" : "EPIPE write" }], + }); + expect(exitCode).toBe(0); + } finally { + for (const fd of open) fs.closeSync(fd); + } + }); + + // Nothing arrives on stdin, so the copy must also see that the pipe it waits for has no reader + // any more. A wait for stdin alone would never end. Linux only: the copy is a sendfile there, + // and a read/write loop first reads the empty stdin. + it.skipIf(!isLinux)("an idle stdin socket does not hide that the reader of a full stdout pipe left", async () => { + using dir = tempDir("bun-write-nonblocking-order", {}); + const gate = mkfifo(dir, "gate.fifo"); + // Linux opens a FIFO for both directions without a peer. The reader below holds the pipe until a line arrives here. + const gateFd = fs.openSync(gate, fs.constants.O_RDWR); + try { + await using proc = Bun.spawn( + sh(`"$BUN" -e "$SCRIPT" | { read -r go < "$GATE"; }`, child(fillStdout, "Bun.stdout", "Bun.stdin"), { + GATE: gate, + }), + ); + const stderr = collect(proc.stderr); + const prefilled = await ready(stderr); + expect(prefilled).toBeGreaterThan(0); + fs.writeSync(gateFd, "go\n"); + const [err, exitCode] = await Promise.all([stderr.all(), proc.exited]); + expect(reports(err)).toEqual([{ ready: prefilled }, { rejected: "EPIPE sendfile" }]); + expect(exitCode).toBe(0); + } finally { + fs.closeSync(gateFd); + } + }); + + // splice (Linux) does not report the closed stdout while stdin is empty. The copy has to sleep + // until stdin has data: a copy that asks again and again uses a whole core. + it.skipIf(!isLinux)("an idle stdin pipe after the reader of stdout left costs no CPU", async () => { + using dir = tempDir("bun-write-nonblocking-idle", {}); + const gate = mkfifo(dir, "gate.fifo"); + const gateFd = fs.openSync(gate, fs.constants.O_RDWR); + // Runs in the child once the copy has started. The sleep is the window that the CPU time is + // measured in: a debug build that waits uses about 60 ms, a copy that spins uses the window. + const measureCpu = ` + for (;;) { + try { + fs.writeSync(1, "-"); + } catch (e) { + if (e.code === "EPIPE") break; + throw e; + } + await Bun.sleep(1); + } + const before = process.cpuUsage(); + await Bun.sleep(1000); + const { user, system } = process.cpuUsage(before); + const cpuMs = Math.round((user + system) / 1000); + process.stderr.write(JSON.stringify(cpuMs < 250 ? { idle: true } : { cpuMs }) + "\\n"); + `; + try { + // The first stage keeps stdin open and empty until the test sends a line, then passes it on. + await using proc = Bun.spawn( + sh( + `{ read -r line; echo "$line"; } | "$BUN" -e "$SCRIPT" | { read -r go < "$GATE"; }`, + child("void process.stdout;", "Bun.stdout", "Bun.stdin", measureCpu), + { GATE: gate }, + ), + ); + const stderr = collect(proc.stderr); + await ready(stderr); + fs.writeSync(gateFd, "go\n"); + await stderr.until(reported(2)); + proc.stdin.write("data\n"); + await proc.stdin.flush(); + const [err, exitCode] = await Promise.all([stderr.all(), proc.exited]); + expect(reports(err)).toEqual([{ ready: 0 }, { idle: true }, { rejected: "EPIPE splice" }]); + expect(exitCode).toBe(0); + } finally { + fs.closeSync(gateFd); + } + }); +}); From 79a1f99c1b7abdf6c4e3edfe7be7130903d1a2f9 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 25 Oct 2026 17:31:24 +0000 Subject: [PATCH 2/3] Bun.write(file, file): wait inside the read/write loops, and send EAGAIN from sendfile and splice to them `bun_sys::read_retrying` and `write_retrying` wait for a non-blocking fd that is not ready, as a blocking fd does. `read_write_loop_capped` and the node:fs loop call them. `do_copy_file_range` no longer waits and repeats `sendfile` or `splice`. On EAGAIN it continues the copy in the read/write loop. This removes `wait_for_kernel_copy` and its rule for the end to wait for. The tests cover the continuation after `sendfile` and `splice` copied a part, both waits with a CPU limit, and a path destination. Co-authored-by: Dylan Conway --- src/runtime/node/node_fs.rs | 8 +- src/runtime/webcore/blob/copy_file.rs | 82 +-------- src/sys/lib.rs | 32 ++++ test/js/bun/io/bun-write.test.js | 243 ++++++++++++-------------- 4 files changed, 157 insertions(+), 208 deletions(-) diff --git a/src/runtime/node/node_fs.rs b/src/runtime/node/node_fs.rs index 14d696aae435..24c94d3e6f33 100644 --- a/src/runtime/node/node_fs.rs +++ b/src/runtime/node/node_fs.rs @@ -4880,7 +4880,7 @@ impl NodeFS { let mut broke = false; 'toplevel: while remain > 0 { let read_len = (buf.len() as u64).min(remain) as usize; - let amt = match Syscall::read(src_fd, &mut buf[..read_len]) { + let amt = match Syscall::read_retrying(src_fd, &mut buf[..read_len]) { Ok(result) => result, Err(err) => { return Err(if !src.is_empty() { @@ -4900,7 +4900,7 @@ impl NodeFS { let mut slice = &buf[..amt]; while !slice.is_empty() { - let written = match Syscall::write(dest_fd, slice) { + let written = match Syscall::write_retrying(dest_fd, slice) { Ok(result) => result, Err(err) => { return Err(if !dest.is_empty() { @@ -4919,7 +4919,7 @@ impl NodeFS { } if !broke { 'outer: loop { - let amt = match Syscall::read(src_fd, buf) { + let amt = match Syscall::read_retrying(src_fd, buf) { Ok(result) => result, Err(err) => { return Err(if !src.is_empty() { @@ -4938,7 +4938,7 @@ impl NodeFS { let mut slice = &buf[..amt]; while !slice.is_empty() { - let written = match Syscall::write(dest_fd, slice) { + let written = match Syscall::write_retrying(dest_fd, slice) { Ok(result) => result, Err(err) => { return Err(if !dest.is_empty() { diff --git a/src/runtime/webcore/blob/copy_file.rs b/src/runtime/webcore/blob/copy_file.rs index e207c18fccee..6e728b8db6de 100644 --- a/src/runtime/webcore/blob/copy_file.rs +++ b/src/runtime/webcore/blob/copy_file.rs @@ -368,7 +368,6 @@ impl CopyFile { } let mut has_unset_append = false; - let mut retried_for_dest_error = false; // If they can't use copy_file_range, they probably also can't // use sendfile() or splice() @@ -475,12 +474,9 @@ impl CopyFile { ); return Err(bun_errno::from_errno(bun_sys::E::EINVAL as i32).into()); } - // EAGAIN: a pipe or socket end is non-blocking and not ready. Two regular files never are, so copy_file_range does not wait. - bun_sys::E::EAGAIN - if !matches!(USE, TryWith::CopyFileRange) - && wait_for_kernel_copy(src_fd, dest_fd, &mut retried_for_dest_error) => - { - continue; + // EAGAIN: a pipe or socket end is non-blocking and not ready. The read/write loop waits for it. + bun_sys::E::EAGAIN if !matches!(USE, TryWith::CopyFileRange) => { + return self.fallback_read_write(remain, unknown_size, &mut total_written); } errno => { self.system_error = Some( @@ -956,48 +952,6 @@ impl CopyFile { } } -/// After EAGAIN from sendfile or splice, waits for the end that is not ready. The errno does not say which end. `false` if poll fails. -#[cfg(any(target_os = "linux", target_os = "android"))] -#[cold] -fn wait_for_kernel_copy(src_fd: Fd, dest_fd: Fd, retried_for_dest_error: &mut bool) -> bool { - use bun_sys::posix::{POLL_IN, POLL_OUT, PollFd, poll}; - let mut ends = [ - PollFd { - fd: dest_fd.native(), - events: POLL_OUT, - revents: 0, - }, - PollFd { - fd: src_fd.native(), - events: POLL_IN, - revents: 0, - }, - ]; - if core::mem::take(retried_for_dest_error) { - // The retry did not report the destination's error: splice looks at the source first. poll skips a negative fd, so only the source ends this wait. - ends[0].fd = -1; - return poll(&mut ends, -1).is_ok(); - } - if poll(&mut ends, -1).is_err() { - return false; - } - let [dest, src] = &mut ends; - if dest.revents & (libc::POLLERR | libc::POLLHUP | libc::POLLNVAL) != 0 { - // One retry lets a syscall that looks at the destination first (sendfile) report the error. - *retried_for_dest_error = true; - return true; - } - if dest.revents == 0 { - src.fd = -1; - } else if src.revents == 0 { - // With no events, poll still reports a destination that closes or fails. - dest.events = 0; - } else { - return true; - } - poll(&mut ends, -1).is_ok() -} - #[cfg(any(target_os = "linux", target_os = "android"))] fn read_write_fallback( src_fd: Fd, @@ -1023,21 +977,6 @@ fn read_write_fallback( } } -/// After EAGAIN, waits until `fd` is ready. Any other error, or a wait that fails, ends the copy with `err`. -#[cfg(not(windows))] -#[cold] -fn wait_if_would_block( - err: bun_sys::Error, - wait: fn(Fd) -> bun_sys::Result<()>, - fd: Fd, -) -> bun_sys::Result<()> { - if err.is_retry() && wait(fd).is_ok() { - Ok(()) - } else { - Err(err) - } -} - #[inline(never)] // 64 KB stack buffer #[cfg(not(windows))] fn read_write_loop_capped( @@ -1052,26 +991,19 @@ fn read_write_loop_capped( let mut remaining = cap; while remaining > 0 { let want = (buf.len() as SizeType).min(remaining) as usize; - let amt = match bun_sys::read(src_fd, &mut buf[..want]) { - Ok(amt) => amt, - Err(err) => { - wait_if_would_block(err, bun_sys::wait_until_readable, src_fd)?; - continue; - } - }; + let amt = bun_sys::read_retrying(src_fd, &mut buf[..want])?; if amt == 0 { break; } remaining -= amt as SizeType; let mut slice = &buf[..amt]; while !slice.is_empty() { - match bun_sys::write(dest_fd, slice) { - Ok(0) => return Ok(()), - Ok(n) => { + match bun_sys::write_retrying(dest_fd, slice)? { + 0 => return Ok(()), + n => { *total += n as u64; slice = &slice[n..]; } - Err(err) => wait_if_would_block(err, bun_sys::wait_until_writable, dest_fd)?, } } } diff --git a/src/sys/lib.rs b/src/sys/lib.rs index 8b52521300e1..866e025ca12c 100644 --- a/src/sys/lib.rs +++ b/src/sys/lib.rs @@ -9391,6 +9391,38 @@ fn wait_until(fd: Fd, events: i16) -> Maybe<()> { } } +/// `read` that treats `fd` as blocking even when its description has O_NONBLOCK: on EAGAIN it waits, then reads again. +#[cfg(unix)] +#[inline] +pub fn read_retrying(fd: Fd, buf: &mut [u8]) -> Maybe { + loop { + match read(fd, buf) { + Err(e) if e.is_retry() => { + if wait_until_readable(fd).is_err() { + return Err(e); + } + } + result => return result, + } + } +} + +/// The `write` counterpart of [`read_retrying`]. +#[cfg(unix)] +#[inline] +pub fn write_retrying(fd: Fd, bytes: &[u8]) -> Maybe { + loop { + match write(fd, bytes) { + Err(e) if e.is_retry() => { + if wait_until_writable(fd).is_err() { + return Err(e); + } + } + result => return result, + } + } +} + /// Writes all of `bytes` to stdout or stderr, for a caller that has to know whether they got there. `Output` doesn't say. pub fn write_all_to_stdio(fd: Fd, mut bytes: &[u8]) -> Maybe<()> { while !bytes.is_empty() { diff --git a/test/js/bun/io/bun-write.test.js b/test/js/bun/io/bun-write.test.js index 01b9f00b4066..aa58997d1c4f 100644 --- a/test/js/bun/io/bun-write.test.js +++ b/test/js/bun/io/bun-write.test.js @@ -1796,6 +1796,8 @@ int posix_fadvise(int fd, off_t offset, off_t len, int advice) { (isWindows ? describe.skip : describe.concurrent)("Bun.write(file, file) on a non-blocking fd", () => { const size = 1024 * 1024; const pattern = Buffer.alloc(size, Buffer.from(Array.from({ length: 251 }, (_, i) => i))); + // Larger than a pipe on a kernel with 64 KiB pages, so a copy into a pipe that nothing reads cannot finish. + const big = Buffer.alloc(4 * size, pattern); // fs.promises.stat is a pool job like the copy. The first one starts a pool thread. The second // one is queued after the copy, which gives the copy time to reach its first syscall before the @@ -1831,6 +1833,17 @@ int posix_fadvise(int fd, off_t offset, off_t len, int advice) { } `; + // Runs in the child while the copy waits. The sleep is the window that the CPU time is measured + // in: a copy that waits uses about 60 ms in a debug build, a copy that asks again and again uses + // the window. + const idleWhileItWaits = ` + const before = process.cpuUsage(); + await Bun.sleep(1000); + const { user, system } = process.cpuUsage(before); + const cpuMs = Math.round((user + system) / 1000); + process.stderr.write(JSON.stringify(cpuMs < 250 ? { idle: true } : { cpuMs }) + "\\n"); + `; + const sh = (pipeline, script, env = {}) => ({ cmd: ["sh", "-c", pipeline], env: { ...bunEnv, ...env, BUN: bunExe(), SCRIPT: script }, @@ -1838,12 +1851,11 @@ int posix_fadvise(int fd, off_t offset, off_t len, int advice) { stdout: "pipe", stderr: "pipe", }); + // The child as a stage of a pipeline. sh returns the status of the last stage only, so this + // stage reports the status of the child. + const bun = `{ "$BUN" -e "$SCRIPT"; echo "{\\"exit\\":$?}" >&2; }`; // The reader of the child's stdout starts only when the test sends a line on sh's stdin (fd 3). const gatedReader = `{ read -r go <&3; exec cat; }`; - async function releaseReader(proc) { - proc.stdin.write("go\n"); - await proc.stdin.flush(); - } // Reads a stream as it arrives. `until` resolves when `done(text so far)` holds or the stream ends. function collect(stream) { @@ -1896,92 +1908,80 @@ int posix_fadvise(int fd, off_t offset, off_t len, int advice) { return fifo; } - // Sends "hel", then "lo\n" only after "hel" came out of the copy, so the copy finds its source - // empty part-way. A copy that has already settled ends the wait too. - async function feedInTwoPieces(send, arrived, stderr) { - await send("hel"); - await Promise.race([arrived("hel"), stderr.until(reported(2))]); - await send("lo\n"); - await Promise.race([arrived("hello\n"), stderr.until(reported(2))]); - } - - function expectWholeFile(bytes, err, prefilled) { - expect({ - piped: bytes.length, - intact: pattern.equals(bytes.subarray(prefilled)), - reports: reports(err), - }).toEqual({ - piped: prefilled + size, - intact: true, - reports: [{ ready: prefilled }, { resolved: size }], - }); + // Sends each piece only after the one before it came out of the copy, so the copy finds its + // source empty part-way. `settled` resolves when the copy has ended: that ends the wait too. + async function feedInPieces(pieces, send, arrived, settled) { + let sent = ""; + for (const piece of pieces) { + await send(piece); + sent += piece; + await Promise.race([arrived(sent), settled]); + } } - async function expectWholeFileThroughPipe(proc) { + // The child's stdout is a pipe that nothing reads until the child has sent every report that + // comes before the outcome. The copy fills the pipe and has to wait. + async function expectWholeFileThroughPipe(proc, whileItWaits = []) { const stderr = collect(proc.stderr); - const prefilled = await ready(stderr); - expect(prefilled).toBeGreaterThan(0); - await releaseReader(proc); + await stderr.until(reported(1 + whileItWaits.length)); + proc.stdin.write("go\n"); + await proc.stdin.flush(); const [bytes, err, exitCode] = await Promise.all([proc.stdout.bytes(), stderr.all(), proc.exited]); - expectWholeFile(bytes, err, prefilled); + expect({ piped: bytes.length, intact: big.equals(bytes), reports: reports(err) }).toEqual({ + piped: big.length, + intact: true, + reports: [{ ready: 0 }, ...whileItWaits, { resolved: big.length }, { exit: 0 }], + }); expect(exitCode).toBe(0); } - it("a file to a full stdout pipe", async () => { - using dir = tempDir("bun-write-nonblocking-pipe", { "source.bin": pattern }); + // Linux: sendfile fills the pipe and then fails with EAGAIN. The read/write loop continues at + // that byte and waits. macOS has only the loop. + it("a file to a stdout pipe that fills during the copy", async () => { + using dir = tempDir("bun-write-nonblocking-pipe", { "source.bin": big }); const source = `Bun.file(${JSON.stringify(join(String(dir), "source.bin"))})`; await using proc = Bun.spawn( - sh(`exec 3<&0; "$BUN" -e "$SCRIPT" | ${gatedReader}`, child(fillStdout, "Bun.stdout", source)), + sh(`exec 3<&0; ${bun} | ${gatedReader}`, child("void process.stdout;", "Bun.stdout", source, idleWhileItWaits)), ); - await expectWholeFileThroughPipe(proc); + await expectWholeFileThroughPipe(proc, [{ idle: true }]); }); - // Linux knows the type of fd 1 and uses sendfile for it. It knows nothing about fd 3, tries - // copy_file_range, and falls back to the read/write loop, which is the only loop on macOS. - it("a file to a full pipe on Bun.file(3)", async () => { - using dir = tempDir("bun-write-nonblocking-fd3", { "source.bin": pattern }); - const source = `Bun.file(${JSON.stringify(join(String(dir), "source.bin"))})`; - await using proc = Bun.spawn( - sh(`exec 3<&0; "$BUN" -e "$SCRIPT" 3>&1 | ${gatedReader}`, child(fillStdout, "Bun.file(3)", source)), - ); - await expectWholeFileThroughPipe(proc); - }); - - it("a stdin pipe to a full stdout pipe", async () => { - using dir = tempDir("bun-write-nonblocking-pipes", { "source.bin": pattern }); + // Linux: the same for splice. + it("a stdin pipe to a stdout pipe that fills during the copy", async () => { + using dir = tempDir("bun-write-nonblocking-pipes", { "source.bin": big }); await using proc = Bun.spawn( sh( - `exec 3<&0; cat "$SOURCE" | "$BUN" -e "$SCRIPT" | ${gatedReader}`, - child(fillStdout, "Bun.stdout", "Bun.stdin"), - { - SOURCE: join(String(dir), "source.bin"), - }, + `exec 3<&0; cat "$SOURCE" | ${bun} | ${gatedReader}`, + child("void process.stdout;", "Bun.stdout", "Bun.stdin"), + { SOURCE: join(String(dir), "source.bin") }, ), ); await expectWholeFileThroughPipe(proc); }); - // A splice between two pipes does not block if either is non-blocking, so an empty stdin fails it too. - it("a stdin pipe that has no data yet, once fd 1 is non-blocking", async () => { + // A splice between two pipes does not block if either is non-blocking, so an empty stdin fails + // it too. Linux only: a read/write loop blocks in the read of stdin. + it.skipIf(!isLinux)("a stdin pipe that has no data yet, once fd 1 is non-blocking", async () => { // The first stage turns two lines from the test into the two pieces. const pieces = `{ read -r a; printf '%s' "$a"; read -r b; printf '%s\\n' "$b"; }`; await using proc = Bun.spawn( - sh(`${pieces} | "$BUN" -e "$SCRIPT" | cat`, child("void process.stdout;", "Bun.stdout", "Bun.stdin")), + sh(`${pieces} | ${bun} | cat`, child("void process.stdout;", "Bun.stdout", "Bun.stdin")), ); const [stdout, stderr] = [collect(proc.stdout), collect(proc.stderr)]; await ready(stderr); - await feedInTwoPieces( + await feedInPieces( + ["hel", "lo\n"], async piece => { proc.stdin.write(piece.trimEnd() + "\n"); await proc.stdin.flush(); }, text => stdout.until(out => out.includes(text)), - stderr, + stderr.until(reported(2)), ); const [out, err, exitCode] = await Promise.all([stdout.all(), stderr.all(), proc.exited]); expect({ stdout: out, reports: reports(err) }).toEqual({ stdout: "hello\n", - reports: [{ ready: 0 }, { resolved: 6 }], + reports: [{ ready: 0 }, { resolved: 6 }, { exit: 0 }], }); expect(exitCode).toBe(0); }); @@ -2001,7 +2001,15 @@ int posix_fadvise(int fd, off_t offset, off_t len, int advice) { const prefilled = await ready(stderr); expect(prefilled).toBeGreaterThan(0); const [bytes, err, exitCode] = await Promise.all([proc.stdout.bytes(), stderr.all(), proc.exited]); - expectWholeFile(bytes, err, prefilled); + expect({ + piped: bytes.length, + intact: pattern.equals(bytes.subarray(prefilled)), + reports: reports(err), + }).toEqual({ + piped: prefilled + size, + intact: true, + reports: [{ ready: prefilled }, { resolved: size }], + }); expect(exitCode).toBe(0); }); @@ -2021,14 +2029,16 @@ int posix_fadvise(int fd, off_t offset, off_t len, int advice) { if (source < 1024) throw new Error("the FIFO is on fd " + source); `; await using proc = Bun.spawn( - sh(`ulimit -n 2048 2>/dev/null; "$BUN" -e "$SCRIPT" | cat`, child(open, "Bun.stdout", "Bun.file(source)")), + sh(`ulimit -n 2048 2>/dev/null; ${bun} | cat`, child(open, "Bun.stdout", "Bun.file(source)", idleWhileItWaits)), ); const [stdout, stderr] = [collect(proc.stdout), collect(proc.stderr)]; - await ready(stderr); - await feedInTwoPieces( + // The first piece goes out after the child measured the wait for it. + await stderr.until(reported(2)); + await feedInPieces( + ["hel", "lo\n"], async piece => void fs.writeSync(writer, piece), text => stdout.until(out => out.includes(text)), - stderr, + stderr.until(reported(3)), ); // The copy now waits for more. The end of the file has to wake it. fs.closeSync(writer); @@ -2036,7 +2046,7 @@ int posix_fadvise(int fd, off_t offset, off_t len, int advice) { const [out, err, exitCode] = await Promise.all([stdout.all(), stderr.all(), proc.exited]); expect({ stdout: out, reports: reports(err) }).toEqual({ stdout: "hello\n", - reports: [{ ready: 0 }, { resolved: 6 }], + reports: [{ ready: 0 }, { idle: true }, { resolved: 6 }, { exit: 0 }], }); expect(exitCode).toBe(0); } finally { @@ -2046,8 +2056,6 @@ int posix_fadvise(int fd, off_t offset, off_t len, int advice) { }); it("a reader that takes a part of a full FIFO and leaves fails the copy with EPIPE", async () => { - // Larger than a pipe on a kernel with 64 KiB pages, so the copy cannot finish before the reader leaves. - const big = Buffer.alloc(4 * size, pattern); using dir = tempDir("bun-write-nonblocking-fifo-dest", { "source.bin": big }); const source = `Bun.file(${JSON.stringify(join(String(dir), "source.bin"))})`; const fifo = mkfifo(dir, "dest.fifo"); @@ -2087,7 +2095,7 @@ int posix_fadvise(int fd, off_t offset, off_t len, int advice) { }).toEqual({ tookPartOfTheFile: true, intact: true, - reports: [{ ready: prefilled }, { rejected: isLinux ? "EPIPE sendfile" : "EPIPE write" }], + reports: [{ ready: prefilled }, { rejected: "EPIPE write" }], }); expect(exitCode).toBe(0); } finally { @@ -2095,76 +2103,53 @@ int posix_fadvise(int fd, off_t offset, off_t len, int advice) { } }); - // Nothing arrives on stdin, so the copy must also see that the pipe it waits for has no reader - // any more. A wait for stdin alone would never end. Linux only: the copy is a sendfile there, - // and a read/write loop first reads the empty stdin. - it.skipIf(!isLinux)("an idle stdin socket does not hide that the reader of a full stdout pipe left", async () => { - using dir = tempDir("bun-write-nonblocking-order", {}); - const gate = mkfifo(dir, "gate.fifo"); - // Linux opens a FIFO for both directions without a peer. The reader below holds the pipe until a line arrives here. - const gateFd = fs.openSync(gate, fs.constants.O_RDWR); - try { - await using proc = Bun.spawn( - sh(`"$BUN" -e "$SCRIPT" | { read -r go < "$GATE"; }`, child(fillStdout, "Bun.stdout", "Bun.stdin"), { - GATE: gate, - }), - ); - const stderr = collect(proc.stderr); - const prefilled = await ready(stderr); - expect(prefilled).toBeGreaterThan(0); - fs.writeSync(gateFd, "go\n"); - const [err, exitCode] = await Promise.all([stderr.all(), proc.exited]); - expect(reports(err)).toEqual([{ ready: prefilled }, { rejected: "EPIPE sendfile" }]); - expect(exitCode).toBe(0); - } finally { - fs.closeSync(gateFd); - } - }); - - // splice (Linux) does not report the closed stdout while stdin is empty. The copy has to sleep - // until stdin has data: a copy that asks again and again uses a whole core. - it.skipIf(!isLinux)("an idle stdin pipe after the reader of stdout left costs no CPU", async () => { - using dir = tempDir("bun-write-nonblocking-idle", {}); - const gate = mkfifo(dir, "gate.fifo"); - const gateFd = fs.openSync(gate, fs.constants.O_RDWR); - // Runs in the child once the copy has started. The sleep is the window that the CPU time is - // measured in: a debug build that waits uses about 60 ms, a copy that spins uses the window. - const measureCpu = ` - for (;;) { - try { - fs.writeSync(1, "-"); - } catch (e) { - if (e.code === "EPIPE") break; - throw e; - } - await Bun.sleep(1); - } - const before = process.cpuUsage(); - await Bun.sleep(1000); - const { user, system } = process.cpuUsage(before); - const cpuMs = Math.round((user + system) / 1000); - process.stderr.write(JSON.stringify(cpuMs < 250 ? { idle: true } : { cpuMs }) + "\\n"); + // One socket is stdin and stdout of the child, as inetd and `socat EXEC:` start a server, so the + // use of process.stdout makes stdin non-blocking too. Bun opens a path destination itself and + // copies into it with the loop of node:fs. The file at the path is shorter than the source and + // its size is already known: the copy still goes to the end of the source. + // Linux only: on macOS fcopyfile refuses a source that is not a regular file (#14054). + it.skipIf(!isLinux)("a stdin socket to a path, once the same socket on fd 1 is non-blocking", async () => { + using dir = tempDir("bun-write-nonblocking-to-path", { "out.txt": "older" }); + const path = join(String(dir), "out.txt"); + const setup = ` + void process.stdout; + const flags = fs.readFileSync("/proc/self/fdinfo/0", "utf8").match(/^flags:\\s*([0-7]+)/m)[1]; + if (!(parseInt(flags, 8) & fs.constants.O_NONBLOCK)) throw new Error("stdin is blocking"); + const destination = Bun.file(${JSON.stringify(path)}); + if (destination.size !== 5) throw new Error("the destination has " + destination.size + " bytes"); `; + await using proc = Bun.spawn({ + cmd: ["sh", "-c", `exec "$BUN" -e "$SCRIPT" <&3 >&3 3<&-`], + env: { ...bunEnv, BUN: bunExe(), SCRIPT: child(setup, "destination", "Bun.stdin") }, + stdio: ["ignore", "ignore", "pipe", "socket-fd"], + }); + // Reading .stdio gives the fd to the caller. + const socket = proc.stdio[3]; + let socketOpen = true; try { - // The first stage keeps stdin open and empty until the test sends a line, then passes it on. - await using proc = Bun.spawn( - sh( - `{ read -r line; echo "$line"; } | "$BUN" -e "$SCRIPT" | { read -r go < "$GATE"; }`, - child("void process.stdout;", "Bun.stdout", "Bun.stdin", measureCpu), - { GATE: gate }, - ), - ); const stderr = collect(proc.stderr); - await ready(stderr); - fs.writeSync(gateFd, "go\n"); - await stderr.until(reported(2)); - proc.stdin.write("data\n"); - await proc.stdin.flush(); + let settled = false; + const outcome = stderr.until(reported(2)).finally(() => (settled = true)); + // The first piece is in the socket before the copy starts. The next two end the sized part + // of the loop and the part after it, and each finds the socket empty first. + await feedInPieces( + ["hel", "lo", " world\n"], + async piece => void fs.writeSync(socket, piece), + async text => { + while (!settled && fs.readFileSync(path, "utf8") !== text) await Bun.sleep(1); + }, + outcome, + ); + fs.closeSync(socket); + socketOpen = false; const [err, exitCode] = await Promise.all([stderr.all(), proc.exited]); - expect(reports(err)).toEqual([{ ready: 0 }, { idle: true }, { rejected: "EPIPE splice" }]); + expect({ reports: reports(err), content: fs.readFileSync(path, "utf8") }).toEqual({ + reports: [{ ready: 0 }, { resolved: 12 }], + content: "hello world\n", + }); expect(exitCode).toBe(0); } finally { - fs.closeSync(gateFd); + if (socketOpen) fs.closeSync(socket); } }); }); From 320ef461f3a42077ee47d85d1d5e170f5f796362 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 25 Oct 2026 19:43:11 +0000 Subject: [PATCH 3/3] ci: retrigger