diff --git a/src/runtime/webcore/blob/read_file.rs b/src/runtime/webcore/blob/read_file.rs index 8bc36a6b5805..3876937d7251 100644 --- a/src/runtime/webcore/blob/read_file.rs +++ b/src/runtime/webcore/blob/read_file.rs @@ -285,6 +285,9 @@ pub struct ReadFile { pub(crate) io_request: io::Request, #[cfg(not(windows))] pub(crate) could_block: bool, + /// FIFO vnode (not a `pipe(2)` pipe); see `bun_sys::block_until_readable`. + #[cfg(target_os = "macos")] + pub(crate) is_named_pipe: bool, pub(crate) close_after_io: bool, pub(crate) state: AtomicU8, // ClosingState } @@ -380,6 +383,8 @@ impl ReadFile { scheduled: false, }, could_block: false, + #[cfg(target_os = "macos")] + is_named_pipe: false, close_after_io: false, state: AtomicU8::new(ClosingState::Running as u8), }; @@ -465,6 +470,37 @@ impl ReadFile { } } + /// `wait_for_readable` for named pipes, on this thread; `false` if the wait itself failed. + #[cfg(target_os = "macos")] + fn block_until_readable(&mut self) -> bool { + bloblog!("ReadFile.blockUntilReadable"); + match bun_sys::block_until_readable(self.opened_fd) { + Ok(()) => true, + Err(err) => { + self.errno = Some(bun_errno::from_errno(err.errno as i32).into()); + self.system_error = Some(err.to_system_error().into()); + false + } + } + } + + /// A named pipe's whole read, as its own pool task: `JobContext::run` holds a + /// VM borrow and VM teardown waits for borrows, so waiting for a writer must + /// not happen inside it. Waiting before the first read also keeps a FIFO + /// with no writer yet from reading as empty. + #[cfg(target_os = "macos")] + fn read_named_pipe_task(task: *mut WorkPoolTask) { + // SAFETY: only reached via `WorkPoolTask::callback` with `task` = + // `&mut self.task` (intrusive) scheduled by `run_async_with_fd`; + // recover parent. + let this = unsafe { &mut *ReadFile::from_task_ptr(task) }; + if this.block_until_readable() { + this.do_read_loop(); + } else { + this.on_finish(); + } + } + /// Pick the read target: `buffer`'s spare capacity if it is at least as /// large as `stack_buffer`, otherwise `stack_buffer`; capped by /// `max_length - read_off`. Returns `(use_stack, target)` so the caller @@ -682,6 +718,11 @@ impl ReadFile { } self.could_block = !bun_sys::is_regular_file(stat.st_mode as _); + #[cfg(target_os = "macos")] + { + // pipe(2) pipes are S_IFIFO with st_dev == 0 (XNU pipe_stat). + self.is_named_pipe = bun_sys::S::ISFIFO(stat.st_mode as _) && stat.st_dev != 0; + } self.total_size = SizeType::try_from((stat.st_size as i64).max(0).min(MAX_SIZE as i64)).unwrap(); @@ -754,6 +795,16 @@ impl ReadFile { // If we immediately call read(), it will block until stdin is // readable. if self.could_block { + #[cfg(target_os = "macos")] + if self.is_named_pipe { + self.task = WorkPoolTask { + node: Default::default(), + callback: Self::read_named_pipe_task, + }; + WorkPool::schedule(&raw mut self.task); + return; + } + if bun_core::is_readable(fd) == bun_core::Pollable::NotReady { self.wait_for_readable(); return; @@ -855,6 +906,13 @@ impl ReadFile { // call. We already know it's done. && !self.read_eof) { + #[cfg(target_os = "macos")] + if self.is_named_pipe { + if self.block_until_readable() { + continue; + } + break; + } if self.could_block // If we received EOF, we can skip the poll() system // call. We already know it's done. diff --git a/src/sys/lib.rs b/src/sys/lib.rs index 8b704ae5b690..2cc3b71bedf0 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` select(2): no FD_SETSIZE limit; a set is ceil(nfds/32) 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"] @@ -7607,6 +7619,41 @@ pub fn kevent( } } +/// Blocks in `select(2)` until `fd` is readable or at EOF; retries on EINTR. +/// +/// For named pipes: XNU's kqueue filter for a FIFO (`filt_vnode_common`, which +/// `poll(2)` uses too) fires only while bytes are buffered, never for the last +/// writer closing; `fifo_select` consults the FIFO's socket, which reports that +/// close (and not a writer that has yet to connect). `pipe(2)` pipes report `EV_EOF`. +#[cfg(target_os = "macos")] +pub fn block_until_readable(fd: Fd) -> Maybe<()> { + debug_assert!(fd.is_valid()); + let index = fd.native() as usize; + let word = index / 32; + let mut read_set = vec![0u32; word + 1]; + loop { + read_set.fill(0); + read_set[word] = 1 << (index % 32); + // SAFETY: `read_set` holds the ceil(nfds / 32) words the kernel reads + // and writes back for `nfds = fd + 1`; the write and error sets and the + // timeout (wait indefinitely) may be null. + let rc = unsafe { + nocancel::select( + fd.native() + 1, + read_set.as_mut_ptr(), + core::ptr::null_mut(), + 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/util/bun-file-read.test.ts b/test/js/bun/util/bun-file-read.test.ts index 40c6309a0c13..c7785cf8738a 100644 --- a/test/js/bun/util/bun-file-read.test.ts +++ b/test/js/bun/util/bun-file-read.test.ts @@ -1,5 +1,8 @@ import { describe, expect, it } from "bun:test"; -import { tempDir } from "harness"; +import { bunEnv, bunExe, isWindows, tempDir } from "harness"; +import { mkfifo } from "mkfifo"; +import { randomBytes } from "node:crypto"; +import { closeSync, constants, openSync, writeSync } from "node:fs"; import { tmpdir } from "node:os"; import path from "node:path"; @@ -54,3 +57,221 @@ describe("Bun.file read-loop target selection", () => { expect(Bun.hash(buf)).toBe(Bun.hash(bytes.subarray(start, end))); }); }); + +// Whole-file reads of a named pipe. All but the last end with the reader +// having drained the pipe and then learning that the last writer closed; on +// macOS that EOF is invisible to kqueue and poll(2) (they only see buffered +// bytes on a FIFO), so the reader has to wait for it differently than it does +// for a pipe(2) pipe, and each of these used to leave the child blocked +// forever there. +describe.skipIf(isWindows)("reading a named pipe", () => { + function readFifoInChild(script: string, fifo: string, stdin: number | "ignore" = "ignore") { + return Bun.spawn({ + cmd: [bunExe(), "-e", script], + env: { ...bunEnv, FIFO: fifo }, + stdin, + stdout: "pipe", + stderr: "pipe", + }); + } + + // When a FIFO end gets closed is what these tests are about, so each one is + // closed explicitly at the right moment; `using` only covers the failure paths. + function openFd(file: string, flags: number | string) { + let fd = openSync(file, flags); + return { + get fd() { + return fd; + }, + close() { + if (fd !== -1) closeSync(fd); + fd = -1; + }, + [Symbol.dispose]() { + this.close(); + }, + }; + } + + // The child must already have the FIFO open for reading before a writer can + // connect to it: a non-blocking open for writing fails with ENXIO until then. + async function openWriterOnceChildIsReading(fifo: string, child: Bun.Subprocess) { + const deadline = performance.now() + 10_000; + while (true) { + try { + return openFd(fifo, constants.O_WRONLY | constants.O_NONBLOCK); + } catch (err: any) { + if (err.code !== "ENXIO") throw err; + } + const exited = child.exitCode ?? child.signalCode; + if (exited !== null || performance.now() > deadline) { + throw new Error( + `nothing opened ${fifo} for reading; child ${exited === null ? "is still running" : `exited (${exited})`}`, + ); + } + await Bun.sleep(5); + } + } + + it.concurrent("bytes() collects a payload that arrives in pieces and ends when the writer closes", async () => { + const payload = randomBytes(256 * 1024); + using dir = tempDir("bun-file-read-fifo", {}); + const fifo = path.join(String(dir), "in.fifo"); + mkfifo(fifo); + // The write end can only be opened, and written to without EPIPE, while + // some reader has the FIFO open; `holder` is that reader until the child + // has opened its own. It never reads, so every byte goes to the child. + using holder = openFd(fifo, constants.O_RDONLY | constants.O_NONBLOCK); + using writer = openFd(fifo, "w"); + await using proc = readFifoInChild( + `const bytes = await Bun.file(process.env.FIFO).bytes(); process.stdout.write(bytes.length + " " + Bun.hash(bytes));`, + fifo, + ); + const stderr = proc.stderr.text(); + // The write end is blocking, so this write only completes as the child + // drains the pipe, and closing it afterwards is what ends the child's + // read. The child cannot exit before that unless it failed; dropping + // `holder` then leaves the pipe without readers, so the blocked write + // fails with EPIPE instead of waiting forever. + const childDied = proc.exited.then(async exitCode => { + holder.close(); + throw new Error(`child exited with ${exitCode} before the payload was written: ${await stderr}`); + }); + const written = await Promise.race([Bun.write(Bun.file(writer.fd), payload), childDied]); + writer.close(); + const [stdout, stderrText, exitCode] = await Promise.all([proc.stdout.text(), stderr, proc.exited]); + + expect({ written, stdout, stderr: stderrText }).toEqual({ + written: payload.length, + stdout: `${payload.length} ${Bun.hash(payload)}`, + stderr: "", + }); + expect(exitCode).toBe(0); + }); + + it.concurrent("text() waits for a writer that connects after the read started", async () => { + using dir = tempDir("bun-file-read-fifo-late-writer", {}); + const fifo = path.join(String(dir), "late.fifo"); + mkfifo(fifo); + + await using proc = readFifoInChild(`process.stdout.write(await Bun.file(process.env.FIFO).text());`, fifo); + using writer = await openWriterOnceChildIsReading(fifo, proc); + writeSync(writer.fd, "written after the reader opened\n"); + writer.close(); + + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr }).toEqual({ stdout: "written after the reader opened\n", stderr: "" }); + expect(exitCode).toBe(0); + }); + + it.concurrent("text() resolves empty when the writer connects and closes without writing", async () => { + using dir = tempDir("bun-file-read-fifo-empty", {}); + const fifo = path.join(String(dir), "empty.fifo"); + mkfifo(fifo); + + await using proc = readFifoInChild( + `const text = await Bun.file(process.env.FIFO).text(); process.stdout.write(JSON.stringify(text));`, + fifo, + ); + using writer = await openWriterOnceChildIsReading(fifo, proc); + writer.close(); + + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr }).toEqual({ stdout: '""', stderr: "" }); + expect(exitCode).toBe(0); + }); + + it.concurrent("Bun.stdin.text() reads a FIFO inherited as stdin to EOF", async () => { + using dir = tempDir("bun-file-read-fifo-stdin", {}); + const fifo = path.join(String(dir), "stdin.fifo"); + mkfifo(fifo); + + // Same dance as above: a reader has to exist before the write end can be + // opened; here that reader becomes the child's stdin. + using readEnd = openFd(fifo, constants.O_RDONLY | constants.O_NONBLOCK); + using writer = openFd(fifo, "w"); + await using proc = readFifoInChild( + `process.stdout.write(JSON.stringify(await Bun.stdin.text()));`, + fifo, + readEnd.fd, + ); + // The child has its own descriptor for the read end now. + readEnd.close(); + writeSync(writer.fd, "stdin is a named pipe\n"); + writer.close(); + + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr }).toEqual({ stdout: JSON.stringify("stdin is a named pipe\n"), stderr: "" }); + expect(exitCode).toBe(0); + }); + + // The macOS wait is a select(2); this is the case where a plain fd_set + // would not hold the descriptor (FD_SETSIZE is 1024 there). + it.concurrent("Bun.file(fd) reads a FIFO whose descriptor number is above FD_SETSIZE", async () => { + using dir = tempDir("bun-file-read-fifo-high-fd", {}); + const fifo = path.join(String(dir), "high.fifo"); + mkfifo(fifo); + + await using proc = readFifoInChild( + `import { constants, openSync } from "node:fs"; + while (openSync("/dev/null", "r") < 1024) {} + const fd = openSync(process.env.FIFO, constants.O_RDONLY | constants.O_NONBLOCK); + const text = await Bun.file(fd).text(); + process.stdout.write(JSON.stringify({ fd, text }));`, + fifo, + ); + using writer = await openWriterOnceChildIsReading(fifo, proc); + writeSync(writer.fd, "read through a high fd\n"); + writer.close(); + + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + const { fd, text } = JSON.parse(stdout); + expect(fd).toBeGreaterThanOrEqual(1024); + expect(text).toBe("read through a high fd\n"); + expect(exitCode).toBe(0); + }); + + // A read that is waiting for a writer which never shows up must not keep + // its VM from shutting down: the wait has to happen outside the job's VM + // borrow, which terminate() waits for. + it.concurrent("terminate() completes while a worker's read is waiting on an idle writer", async () => { + using dir = tempDir("bun-file-read-fifo-worker", { + "main.fixture.ts": ` + const worker = new Worker(new URL("./worker.fixture.ts", import.meta.url).href); + const closed = new Promise(resolve => worker.addEventListener("close", resolve)); + worker.addEventListener("message", ({ data }) => console.log(data)); + // Our stdin is closed once the test has connected a writer to the FIFO. + await Bun.stdin.text(); + worker.terminate(); + await closed; + console.log("terminated"); + `, + "worker.fixture.ts": ` + const read = Bun.file(process.env.FIFO!).text(); + postMessage("reading"); + await read; + postMessage("the read finished, which it should not have"); + `, + }); + const fifo = path.join(String(dir), "idle.fifo"); + mkfifo(fifo); + + await using proc = Bun.spawn({ + cmd: [bunExe(), "main.fixture.ts"], + cwd: String(dir), + env: { ...bunEnv, FIFO: fifo }, + stdin: "pipe", + stdout: "pipe", + stderr: "pipe", + }); + // Connecting a writer proves the worker's read has the FIFO open; never + // writing to it keeps that read waiting for the rest of the test. + using _idleWriter = await openWriterOnceChildIsReading(fifo, proc); + proc.stdin.end(); + + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr }).toEqual({ stdout: "reading\nterminated\n", stderr: "" }); + expect(exitCode).toBe(0); + }); +});