diff --git a/src/runtime/webcore/blob/copy_file.rs b/src/runtime/webcore/blob/copy_file.rs index 427d94140e94..db5ac5deb773 100644 --- a/src/runtime/webcore/blob/copy_file.rs +++ b/src/runtime/webcore/blob/copy_file.rs @@ -383,7 +383,6 @@ impl CopyFile { } loop { - // TODO: this should use non-blocking I/O. let written: isize = match USE { TryWith::CopyFileRange => { // SAFETY: raw copy_file_range(2); both fds owned by caller, null offsets. @@ -427,6 +426,17 @@ impl CopyFile { match bun_sys::get_errno(written) { bun_sys::E::SUCCESS => {} + // A pipe-to-pipe splice is non-blocking if either pipe is, so wait on both sides. + bun_sys::E::EAGAIN => { + if let Err(err) = bun_sys::block_until_readable(src_fd) + .and_then(|()| bun_sys::block_until_writable(dest_fd)) + { + self.system_error = Some(err.to_system_error()); + return Err(bun_errno::from_errno(err.errno as i32).into()); + } + continue; + } + // XDEV: cross-device copy not supported // NOSYS: syscall not available // OPNOTSUPP: filesystem doesn't support this operation @@ -992,19 +1002,28 @@ 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) if err.is_retry() => { + bun_sys::block_until_readable(src_fd)?; + continue; + } + Err(err) => return Err(err), + }; 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) if err.is_retry() => bun_sys::block_until_writable(dest_fd)?, + Err(err) => return Err(err), } } } diff --git a/src/sys/lib.rs b/src/sys/lib.rs index d231ad3a1986..dc08619cb2f8 100644 --- a/src/sys/lib.rs +++ b/src/sys/lib.rs @@ -1411,6 +1411,8 @@ impl Tag { #[cfg(not(windows))] pub(crate) const setrlimit: Tag = Tag(106); pub const clone3: Tag = Tag(107); + #[cfg(target_os = "macos")] + pub(crate) const select: Tag = Tag(108); // `inotify_init1`/`inotify_add_watch` fold under the generic `.watch` // tag; `INotifyWatcher.rs` spells it `.inotify`. Alias to `.watch` // so the JS-facing `err.syscall == "watch"` string stays node-compatible. @@ -1418,7 +1420,7 @@ impl Tag { /// The tag name — spelling is frozen (JS-facing /// `err.syscall` string; node-compat code matches on it). pub fn name(self) -> &'static str { - const NAMES: [&str; 108] = [ + const NAMES: [&str; 109] = [ "TODO", "dup", "access", @@ -1528,6 +1530,7 @@ impl Tag { "getrlimit", "setrlimit", "clone3", + "select", ]; NAMES.get(self.0 as usize).copied().unwrap_or("unknown") } @@ -1708,6 +1711,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; + // `_DARWIN_UNLIMITED_SELECT`: no FD_SETSIZE cap; a set is ceil(nfds / 32) `u32`s. + #[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"] @@ -7630,6 +7642,86 @@ pub fn kevent( } } +/// Blocks until a read from `fd` would not return EAGAIN (there is data, EOF or an error). +#[cfg(all(unix, not(target_os = "macos")))] +pub fn block_until_readable(fd: Fd) -> Maybe<()> { + block_until(fd, posix::POLL_IN) +} + +/// Blocks until a write to `fd` would not return EAGAIN (there is room, or an error like EPIPE). +#[cfg(all(unix, not(target_os = "macos")))] +pub fn block_until_writable(fd: Fd) -> Maybe<()> { + block_until(fd, posix::POLL_OUT) +} + +/// POLLHUP / POLLERR / POLLNVAL wake this without being requested; EINTR is retried. +#[cfg(all(unix, not(target_os = "macos")))] +fn block_until(fd: Fd, events: i16) -> Maybe<()> { + debug_assert!(fd.is_valid()); + let mut fds = [posix::PollFd { + fd: fd.native(), + events, + revents: 0, + }]; + match posix::poll(&mut fds, -1) { + Ok(_) => Ok(()), + Err(err) => Err(err.with_fd(fd)), + } +} + +/// Blocks until a read from `fd` would not return EAGAIN (there is data, EOF or an error). +#[cfg(target_os = "macos")] +pub fn block_until_readable(fd: Fd) -> Maybe<()> { + select_one(fd, SelectFor::Read) +} + +/// Blocks until a write to `fd` would not return EAGAIN (there is room, or an error like EPIPE). +#[cfg(target_os = "macos")] +pub fn block_until_writable(fd: Fd) -> Maybe<()> { + select_one(fd, SelectFor::Write) +} + +#[cfg(target_os = "macos")] +#[derive(Clone, Copy)] +enum SelectFor { + Read, + Write, +} + +/// On XNU, poll(2) on a named pipe never wakes for the other end closing; select(2) does. +#[cfg(target_os = "macos")] +fn select_one(fd: Fd, what: SelectFor) -> Maybe<()> { + debug_assert!(fd.is_valid()); + let index = fd.native() as usize; + let word = index / 32; + let mut set = vec![0u32; word + 1]; + loop { + set.fill(0); + set[word] = 1 << (index % 32); + let (read_set, write_set) = match what { + SelectFor::Read => (set.as_mut_ptr(), core::ptr::null_mut()), + SelectFor::Write => (core::ptr::null_mut(), set.as_mut_ptr()), + }; + // SAFETY: `set` holds the ceil(nfds / 32) words the kernel reads and + // writes back for `nfds = fd + 1`; the other sets and the timeout + // (wait indefinitely) may be null. + let rc = unsafe { + nocancel::select( + fd.native() + 1, + read_set, + write_set, + core::ptr::null_mut(), + core::ptr::null_mut(), + ) + }; + match get_errno(rc) { + E::SUCCESS => return Ok(()), + E::EINTR => continue, + e => return Err(Error::from_code(e, Tag::select).with_fd(fd)), + } + } +} + /// `clonefile` — macOS-only CoW copy. On non-Darwin returns ENOTSUP so /// callers can fall back to `copy_file`. #[cfg(not(target_os = "macos"))] diff --git a/test/js/bun/io/bun-write-nonblocking-stdio.test.ts b/test/js/bun/io/bun-write-nonblocking-stdio.test.ts new file mode 100644 index 000000000000..0033697f061b --- /dev/null +++ b/test/js/bun/io/bun-write-nonblocking-stdio.test.ts @@ -0,0 +1,229 @@ +// O_NONBLOCK lives on the open file description, so once process.stdout has +// been used (a Worker does that while starting up) fd 1 is non-blocking for +// everything that writes to it, Bun.write(Bun.stdout, file) included; a child +// process inherits the state too. The file-to-file copy behind Bun.write is a +// blocking-style loop on a pool thread. It used to report the EAGAIN such an fd +// produces as the copy's failure instead of waiting for the fd. +// +// Bun.spawn's stdio are socketpairs, so the children run under sh to get real +// pipes on fd 0/1. Each child reports on stderr as JSON lines: {ready: } as soon as Bun.write() has been called, then the promise's +// outcome; the test supplies the input or the reader only after "ready". +import { describe, expect, it } from "bun:test"; +import { execFileSync } from "child_process"; +import fs from "fs"; +import { bunEnv, bunExe, isWindows, tempDir } from "harness"; +import { join } from "path"; + +// On Linux the copy is a splice(2)/sendfile(2) loop; the env var sends it down +// the read(2)/write(2) loop that macOS and FreeBSD always use for an fd +// destination, so the Linux lanes cover both. +const loops: [string, Record][] = [ + ["kernel copy", {}], + ["read/write loop", { BUN_CONFIG_DISABLE_COPY_FILE_RANGE: "1" }], +]; + +const copyScript = (setup: string, source: string) => ` + let prefilled = 0; + ${setup} + const copy = Bun.write(Bun.stdout, ${source}); + process.stderr.write(JSON.stringify({ ready: prefilled }) + "\\n"); + copy + .then(n => ({ resolved: n }), e => ({ rejected: e.code + " " + e.syscall })) + .then(outcome => process.stderr.write(JSON.stringify(outcome) + "\\n")); +`; + +// Makes fd 1 non-blocking the way any process.stdout use does, then fills the +// pipe behind it until the kernel refuses more: nothing can be written to it +// until the reader drains it. +const fillStdout = ` + process.stdout.write(""); + const fs = require("fs"); + const zeros = Buffer.alloc(4096); + for (;;) { + try { + prefilled += fs.writeSync(1, zeros); + } catch (e) { + if (e.code !== "EAGAIN") throw e; + break; + } + } +`; + +const shell = (pipeline: string, script: string, env: Record = {}) => + ({ + cmd: ["sh", "-c", pipeline], + env: { ...bunEnv, ...env, BUN: bunExe(), SCRIPT: script }, + stdin: "pipe", + stdout: "pipe", + stderr: "pipe", + }) as const; + +// Accumulates a stream. `until` resolves once the text so far satisfies the +// predicate, or the stream ends; `all` once the stream ends. Both resolve with +// the text so far. Concurrent waiters share one in-flight read. +function collect(stream: ReadableStream) { + const reader = stream.getReader(); + const decoder = new TextDecoder(); + let text = ""; + let pending: Promise | undefined; + const more = () => + (pending ??= reader.read().then(({ value, done }) => { + pending = undefined; + if (!done) text += decoder.decode(value, { stream: true }); + return !done; + })); + return { + async until(predicate: (text: string) => boolean) { + while (!predicate(text)) if (!(await more())) break; + return text; + }, + async all() { + while (await more()) {} + return text; + }, + }; +} +type Collected = ReturnType; + +const parseLines = (text: string) => + text + .trim() + .split("\n") + .map(line => { + try { + return JSON.parse(line); + } catch { + return line; + } + }); +const reported = (count: number) => (text: string) => text.endsWith("\n") && parseLines(text).length >= count; + +async function ready(stderr: Collected): Promise { + const text = await stderr.until(reported(1)); + if (!reported(1)(text)) throw new Error("child exited before reporting ready: " + text); + return parseLines(text)[0].ready; +} + +// The source has to be found empty, writer still attached, while the copy is +// under way. Two pieces do that: the second is sent only after the first has +// come back out of the child, so the copy's next syscall met an empty source +// and had to wait. (The broken copy reports EAGAIN instead, possibly before +// even the first piece; the second piece then just lets the pipeline finish.) +async function feedInTwoPieces(stdout: Collected, stderr: Collected, send: (piece: string) => Promise) { + await send("hel"); + await Promise.race([stdout.until(text => text.includes("hel")), stderr.until(reported(2))]); + await send("lo\n"); +} + +// Stage 1 turns two lines from sh's stdin (this test) into the two pieces. +const lateStdin = `{ read -r a; printf '%s' "$a"; read -r b; printf '%s\\n' "$b"; } | "$BUN" -e "$SCRIPT" | cat`; +const sendLine = (proc: Bun.Subprocess<"pipe", "pipe", "pipe">) => async (piece: string) => { + proc.stdin.write(piece.trimEnd() + "\n"); + await proc.stdin.flush(); +}; + +async function expectCopied(proc: Bun.Subprocess, stdout: Collected, stderr: Collected) { + const [out, err, exitCode] = await Promise.all([stdout.all(), stderr.all(), proc.exited]); + expect({ stdout: out, reports: parseLines(err) }).toEqual({ + stdout: "hello\n", + reports: [{ ready: 0 }, { resolved: 6 }], + }); + expect(exitCode).toBe(0); +} + +(isWindows ? describe.skip : describe.concurrent)("Bun.write(Bun.stdout, file) with a non-blocking stdio fd", () => { + it("waits for stdin once process.stdout has made fd 1 non-blocking", async () => { + // A pipe-to-pipe splice is non-blocking as soon as either pipe is, so it is + // the empty (blocking) stdin that used to fail the copy here. + await using proc = Bun.spawn(shell(lateStdin, copyScript(`process.stdout.write("");`, "Bun.stdin"))); + const [stdout, stderr] = [collect(proc.stdout), collect(proc.stderr)]; + await ready(stderr); + await feedInTwoPieces(stdout, stderr, sendLine(proc)); + await expectCopied(proc, stdout, stderr); + }); + + it("waits for stdin inside a Worker, where fd 1 is non-blocking from the start", async () => { + const script = ` + const { Worker } = require("node:worker_threads"); + new Worker( + 'const { parentPort } = require("node:worker_threads");' + + 'const copy = Bun.write(Bun.stdout, Bun.stdin);' + + 'parentPort.postMessage({ ready: 0 });' + + 'copy.then(n => ({ resolved: n }), e => ({ rejected: e.code + " " + e.syscall })).then(o => parentPort.postMessage(o));', + { eval: true }, + ).on("message", report => process.stderr.write(JSON.stringify(report) + "\\n")); + `; + await using proc = Bun.spawn(shell(lateStdin, script)); + const [stdout, stderr] = [collect(proc.stdout), collect(proc.stderr)]; + await ready(stderr); + await feedInTwoPieces(stdout, stderr, sendLine(proc)); + await expectCopied(proc, stdout, stderr); + }, 30_000); // Boots a second runtime inside the child; a few seconds under ASAN. + + for (const [name, env] of loops) { + // The reader of the child's stdout does not start until this test sends it + // a line over fd 3 (sh's own stdin), so the pipe the child filled up stays + // full until then; wc then counts prefill + copy. + const gatedReader = `{ read -r go <&3; exec wc -c; }`; + const size = 256 * 1024; + + async function expectDrained(proc: Bun.Subprocess<"pipe", "pipe", "pipe">, stderr: Collected, prefilled: number) { + expect(prefilled).toBeGreaterThan(0); + proc.stdin.write("go\n"); + await proc.stdin.flush(); + const [stdout, err, exitCode] = await Promise.all([proc.stdout.text(), stderr.all(), proc.exited]); + expect({ piped: Number(stdout), reports: parseLines(err) }).toEqual({ + piped: prefilled + size, + reports: [{ ready: prefilled }, { resolved: size }], + }); + expect(exitCode).toBe(0); + } + + it(`${name}: a pipe source waits for the reader of a full non-blocking stdout`, async () => { + await using proc = Bun.spawn( + shell( + `exec 3<&0; head -c ${size} /dev/zero | "$BUN" -e "$SCRIPT" | ${gatedReader}`, + copyScript(fillStdout, "Bun.stdin"), + env, + ), + ); + const stderr = collect(proc.stderr); + await expectDrained(proc, stderr, await ready(stderr)); + }); + + it(`${name}: a regular file source waits for the reader of a full non-blocking stdout`, async () => { + using dir = tempDir("bun-write-nonblocking-stdout", { "src.bin": Buffer.alloc(size, "S").toString() }); + const source = `Bun.file(${JSON.stringify(join(String(dir), "src.bin"))})`; + await using proc = Bun.spawn( + shell(`exec 3<&0; "$BUN" -e "$SCRIPT" | ${gatedReader}`, copyScript(fillStdout, source), env), + ); + const stderr = collect(proc.stderr); + await expectDrained(proc, stderr, await ready(stderr)); + }); + + it(`${name}: a non-blocking source waits for its writer`, async () => { + // The child's stdin is a FIFO read end this test opened with O_NONBLOCK; + // the description is shared, so fd 0 is non-blocking in the child + // whatever the child does. The writer is attached before the child starts. + using dir = tempDir("bun-write-nonblocking-stdin", {}); + const fifo = join(String(dir), "fifo"); + execFileSync("mkfifo", [fifo]); + const readEnd = fs.openSync(fifo, fs.constants.O_RDONLY | fs.constants.O_NONBLOCK); + const writeEnd = fs.openSync(fifo, "w"); + await using proc = Bun.spawn({ + ...shell(`"$BUN" -e "$SCRIPT" | cat`, copyScript("", "Bun.stdin"), env), + stdin: readEnd, + }); + fs.closeSync(readEnd); + const [stdout, stderr] = [collect(proc.stdout), collect(proc.stderr)]; + try { + await ready(stderr); + await feedInTwoPieces(stdout, stderr, async piece => fs.writeSync(writeEnd, piece)); + } finally { + fs.closeSync(writeEnd); + } + await expectCopied(proc, stdout, stderr); + }); + } +});