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 e41579856cdd..6e728b8db6de 100644 --- a/src/runtime/webcore/blob/copy_file.rs +++ b/src/runtime/webcore/blob/copy_file.rs @@ -474,6 +474,10 @@ 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. 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( bun_sys::Error { @@ -987,14 +991,14 @@ 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 = 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)? { + match bun_sys::write_retrying(dest_fd, slice)? { 0 => return Ok(()), n => { *total += n as u64; diff --git a/src/sys/lib.rs b/src/sys/lib.rs index 14ee170f6bb6..866e025ca12c 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,117 @@ 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)); + } + } +} + +/// `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() { 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..aa58997d1c4f 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,371 @@ 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))); + // 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 + // 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; + } + } + `; + + // 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 }, + stdin: "pipe", + 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; }`; + + // 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 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]); + } + } + + // 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); + 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]); + 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); + } + + // 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} | ${gatedReader}`, child("void process.stdout;", "Bun.stdout", source, idleWhileItWaits)), + ); + await expectWholeFileThroughPipe(proc, [{ idle: true }]); + }); + + // 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} | ${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. 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} | cat`, child("void process.stdout;", "Bun.stdout", "Bun.stdin")), + ); + const [stdout, stderr] = [collect(proc.stdout), collect(proc.stderr)]; + await ready(stderr); + 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.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 }, { exit: 0 }], + }); + 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]); + 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); + }); + + // 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} | cat`, child(open, "Bun.stdout", "Bun.file(source)", idleWhileItWaits)), + ); + const [stdout, stderr] = [collect(proc.stdout), collect(proc.stderr)]; + // 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.until(reported(3)), + ); + // 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 }, { idle: true }, { resolved: 6 }, { exit: 0 }], + }); + 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 () => { + 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: "EPIPE write" }], + }); + expect(exitCode).toBe(0); + } finally { + for (const fd of open) fs.closeSync(fd); + } + }); + + // 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 { + const stderr = collect(proc.stderr); + 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: reports(err), content: fs.readFileSync(path, "utf8") }).toEqual({ + reports: [{ ready: 0 }, { resolved: 12 }], + content: "hello world\n", + }); + expect(exitCode).toBe(0); + } finally { + if (socketOpen) fs.closeSync(socket); + } + }); +});