diff --git a/src/io/PipeReader.rs b/src/io/PipeReader.rs index dcbb385a9310..d5191fd715cd 100644 --- a/src/io/PipeReader.rs +++ b/src/io/PipeReader.rs @@ -214,6 +214,8 @@ bitflags::bitflags! { const KEEP_ALIVE = 1 << 10; // default true /// A read failed with a non-retry errno. Set before the bytes read ahead of the failure are delivered, so a pull from inside that delivery cannot read the fd past the error. Never cleared: `start()`, `unpause()` and `from()` keep it, and only a reader from `init()` reads again. const READ_FAILED = 1 << 11; + /// fstat says that the fd is not a FIFO and not a socket. Only for an fd that the user names (`Bun.file()`): it can be a file, or a device whose close(2) blocks, and fstat cannot tell a pty from a serial line. + const NOT_A_PIPE = 1 << 13; } } @@ -221,6 +223,11 @@ impl PosixFlags { pub(crate) const fn new() -> Self { Self::from_bits_truncate(PosixFlags::CLOSE_HANDLE.bits() | PosixFlags::KEEP_ALIVE.bits()) } + + /// Whether the fd is a pipe, a socket or the pty of `Bun.Terminal`, which `PollOrFd::close_as` closes on the calling thread. A reader knows this before it has a poll. + pub(crate) const fn is_pipe(self) -> bool { + !self.intersects(PosixFlags::MEMFD.union(PosixFlags::NOT_A_PIPE)) + } } impl PosixBufferedReader { @@ -341,12 +348,23 @@ impl PosixBufferedReader { debug_assert!(!self.flags.contains(PosixFlags::CLOSED_WITHOUT_REPORTING)); self.flags.insert(PosixFlags::CLOSED_WITHOUT_REPORTING); if self.flags.contains(PosixFlags::CLOSE_HANDLE) { - let owner = std::ptr::from_mut(self).cast::(); - self.handle.close(Some(owner), None::); + Self::close_fd(&mut self.handle, self.flags, None, None::); } } } + /// Every close of the fd of a reader. Not `PollOrFd::close`: the poll of a reader does not say what its fd is. Borrows `handle` alone, because `on_close_fn` can free the parent that embeds the reader. + fn close_fd( + handle: &mut PollOrFd, + flags: PosixFlags, + ctx: Option<*mut c_void>, + on_close_fn: Option, + ) where + F: FnOnce(*mut c_void), + { + handle.close_as(ctx, on_close_fn, true, flags.is_pipe()); + } + pub fn get_fd(&self) -> Fd { self.handle.get_fd() } @@ -387,10 +405,10 @@ impl PosixBufferedReader { // The handle is closed after the read regardless of result. // `self.handle` owns the fd; // borrow a non-owning `File` view so the temporary doesn't - // close it on drop (handle.close() below does). + // close it on drop (`close_fd` below does). let result = sys::File::borrow(&fd) .read_to_end_with_array_list(&mut self._buffer, sys::SizeHint::UnknownSize); - self.handle.close(None, None::); + Self::close_fd(&mut self.handle, self.flags, None, None::); if let Err(err) = result { bun_core::debug_warn!("error reading from memfd\n{}", err); return self.buffer(); @@ -446,7 +464,9 @@ impl PosixBufferedReader { // runs without a receiver borrow. unsafe { if (*this).flags.contains(PosixFlags::CLOSE_HANDLE) { - (*this).handle.close( + Self::close_fd( + &mut (*this).handle, + (*this).flags, Some(this.cast::()), // SAFETY: ctx is the live reader raw pointer passed above. Some(|ctx: *mut c_void| Self::done(ctx.cast::())), @@ -554,7 +574,7 @@ impl PosixBufferedReader { if !is_pollable { self.buffer().clear(); self.flags.remove(PosixFlags::IS_DONE); - self.handle.close(None, None::); + Self::close_fd(&mut self.handle, self.flags, None, None::); self.handle = PollOrFd::Fd(fd); return sys::Result::Ok(()); } diff --git a/src/io/lib.rs b/src/io/lib.rs index 8d4870736197..6cf680d1cc20 100644 --- a/src/io/lib.rs +++ b/src/io/lib.rs @@ -1881,6 +1881,11 @@ impl FilePollRef { self.inner().owner = owner; } #[inline] + #[cfg(not(windows))] + pub(crate) fn owner_tag(self) -> PollTag { + self.inner().owner.tag() + } + #[inline] pub(crate) fn deinit_force_unregister(self) { self.inner().deinit_force_unregister(); } diff --git a/src/io/pipes.rs b/src/io/pipes.rs index 5cc50a9079e6..12b3b6a7cad5 100644 --- a/src/io/pipes.rs +++ b/src/io/pipes.rs @@ -4,8 +4,8 @@ use bun_sys::Fd; #[cfg(not(windows))] use bun_sys::FdExt; -#[cfg(target_os = "macos")] -use crate::FilePollFlag; +#[cfg(not(windows))] +use crate::{FilePollFlag, PollTag}; use crate::{FilePollRef, Owner}; pub enum PollOrFd { @@ -58,35 +58,37 @@ impl PollOrFd { close_fd: bool, ) where F: FnOnce(*mut c_void), + { + #[cfg(not(windows))] + let pipe = match self { + PollOrFd::Poll(poll) => { + // Only the poll of a writer says what its fd is: a pipe, a socket or the pty of `Bun.Terminal`, or else the tty of a `FileSink`. + debug_assert!(!close_fd || poll.owner_tag() != PollTag::BufferedReader); + !poll.has_flag(FilePollFlag::Tty) + } + _ => false, + }; + #[cfg(windows)] + let pipe = false; + self.close_as(ctx, on_close_fn, close_fd, pipe); + } + + /// `close_impl` for a caller that knows whether the fd is a pipe, a socket or the pty of `Bun.Terminal`. The other end waits for the close(2) of one (a child for the end of its stdin), so it runs here, before `on_close_fn`. The close(2) of any other fd can block, so the work pool runs it later. + pub fn close_as( + &mut self, + ctx: Option<*mut c_void>, + on_close_fn: Option, + close_fd: bool, + pipe: bool, + ) where + F: FnOnce(*mut c_void), { let fd = self.get_fd(); - #[cfg(target_os = "macos")] - let mut close_async = true; - #[cfg(all(not(target_os = "macos"), not(windows)))] - let close_async = true; if matches!(self, PollOrFd::Poll(_)) { - // workaround kqueue bug. - // 1) non-blocking FIFO - // 2) open for writing only = fd 2, nonblock - // 3) open for reading only = fd 3, nonblock - // 4) write(3, "something") = 9 - // 5) read(2, buf, 9) = 9 - // 6) read(2, buf, 9) = -1 (EAGAIN) - // 7) ON ANOTHER THREAD: close(3) = 0, - // 8) kevent(2, EVFILT_READ, EV_ADD | EV_ENABLE | EV_DISPATCH, 0, 0, 0) = 0 - // 9) ??? No more events for fd 2 // Take ownership of the Box before // calling deinit_force_unregister, then leave self = Closed. let old = core::mem::replace(self, PollOrFd::Closed); if let PollOrFd::Poll(poll) = old { - #[cfg(target_os = "macos")] - { - if poll.has_flag(FilePollFlag::PollWritable) - && poll.has_flag(FilePollFlag::Nonblocking) - { - close_async = false; - } - } // Consumes the underlying allocation. poll.deinit_force_unregister(); } @@ -98,18 +100,22 @@ impl PollOrFd { // TODO: We should make this call compatible using bun.FD #[cfg(windows)] { + let _ = pipe; if close_fd { crate::closer::Closer::close(fd, bun_sys::windows::libuv::Loop::get()); } } #[cfg(not(windows))] { - if close_async && close_fd { + if close_fd && pipe { + debug_assert!(!fstat_is(fd, bun_sys::S::ISREG)); + // Not on another thread: kqueue then delivers no more events to the reader of a nonblocking FIFO. + fd.close(); + } else if close_fd { + debug_assert!(!fstat_is(fd, |mode| { + bun_sys::S::ISFIFO(mode) || bun_sys::S::ISSOCK(mode) + })); crate::closer::Closer::close(fd, ()); - } else { - if close_fd { - let _ = fd.close_allowing_bad_file_descriptor(None); - } } } if let Some(f) = on_close_fn { @@ -135,6 +141,12 @@ impl PollOrFd { } } +/// For the debug assertions of `close_as`. False when fstat fails. +#[cfg(not(windows))] +fn fstat_is(fd: Fd, is: fn(bun_sys::Mode) -> bool) -> bool { + bun_sys::fstat(fd).is_ok_and(|stat| is(stat.st_mode as bun_sys::Mode)) +} + // Sunk to `bun_io` so `FilePoll::file_type()` needs no aio→io edge; re-export // keeps the historical `bun_io::FileType` / `bun_io::pipes::FileType` paths. pub use crate::posix_event_loop::FileType; diff --git a/src/runtime/server/FileResponseStream.rs b/src/runtime/server/FileResponseStream.rs index 75b679af7d9d..7d5e43c09717 100644 --- a/src/runtime/server/FileResponseStream.rs +++ b/src/runtime/server/FileResponseStream.rs @@ -646,10 +646,27 @@ bun_io::impl_buffered_reader_parent! { impl Drop for FileResponseStream { fn drop(&mut self) { bun_output::scoped_log!(FileResponseStream, "deinit"); + let auto_close = self.auto_close.get(); // `start()` cleared CLOSE_HANDLE, so the reader's `Drop` doesn't unregister the poll. #[cfg(unix)] - self.reader_mut().handle.close_without_closing_fd(); - if self.auto_close.get() { + { + let reader = self.reader_mut(); + let fd = reader.get_fd(); + if fd != Fd::INVALID { + // `FileType::Pipe` is a character device too, so fstat says whether the other end waits for this close. + let pipe = auto_close + && reader.flags.contains(ReaderFlags::POLLABLE) + && sys::fstat(fd).is_ok_and(|stat| { + let mode = stat.st_mode as sys::Mode; + sys::S::ISFIFO(mode) || sys::S::ISSOCK(mode) + }); + reader + .handle + .close_as(None, None::, auto_close, pipe); + return; + } + } + if auto_close { #[cfg(windows)] Closer::close(self.fd.get(), bun_sys::windows::libuv::Loop::get()); #[cfg(not(windows))] diff --git a/src/runtime/webcore/FileReader.rs b/src/runtime/webcore/FileReader.rs index 5b885601454d..81fb2da3e38d 100644 --- a/src/runtime/webcore/FileReader.rs +++ b/src/runtime/webcore/FileReader.rs @@ -100,6 +100,9 @@ pub(crate) struct OpenedFileBlob { pub(crate) nonblocking: bool, #[cfg(not(windows))] pub(crate) file_type: FileType, + /// Not a FIFO and not a socket. `file_type` is a pipe for every device that can be polled. + #[cfg(not(windows))] + pub(crate) not_a_pipe: bool, } impl Default for OpenedFileBlob { @@ -110,6 +113,8 @@ impl Default for OpenedFileBlob { nonblocking: true, #[cfg(not(windows))] file_type: FileType::File, + #[cfg(not(windows))] + not_a_pipe: false, } } } @@ -224,6 +229,7 @@ impl Lazy { } else { FileType::File }; + this.not_a_pipe = this.file_type == FileType::File; // pretend it's a non-blocking pipe if it's a TTY if is_nonblocking && this.file_type != FileType::Socket { @@ -338,6 +344,9 @@ impl FileReader { .flags .set(PosixFlags::NONBLOCKING, opened.nonblocking); self.reader().flags.set(PosixFlags::POLLABLE, pollable); + self.reader() + .flags + .set(PosixFlags::NOT_A_PIPE, opened.not_a_pipe); } #[cfg(windows)] { diff --git a/test/js/bun/shell/shell-pipe-read-fault.test.ts b/test/js/bun/shell/shell-pipe-read-fault.test.ts index 594d4e8b150e..ad001b0f5730 100644 --- a/test/js/bun/shell/shell-pipe-read-fault.test.ts +++ b/test/js/bun/shell/shell-pipe-read-fault.test.ts @@ -248,10 +248,8 @@ console.log(JSON.stringify({ exitCode: r.exitCode })); // has to register its poll (the stdout/stderr readers register theirs right // after). Runs the command several times and lists the parent's fds that were // not open before: a spawn that fails part way through setup must not leave a -// stdio end open. An fd held by a registered poll is closed on the thread pool, -// so the list is polled until it drains (bounded). A command that never -// finishes shows up as a test timeout. `argv[2]` selects the single-command or -// the pipeline form. +// stdio end open. A command that never finishes shows up as a test timeout. +// `argv[2]` selects the single-command or the pipeline form. const STDIN_BLOB_FIXTURE = /* js */ ` import { $ } from "bun"; import { readdirSync, readlinkSync } from "node:fs"; @@ -265,12 +263,7 @@ await run(); const before = fds(); let last; for (let i = 0; i < 5; i++) last = await run(); -const deadline = Date.now() + 2000; -let leaked = [...fds()].filter(fd => !before.has(fd)); -while (leaked.length > 0 && Date.now() < deadline) { - await Bun.sleep(10); - leaked = [...fds()].filter(fd => !before.has(fd)); -} +const leaked = [...fds()].filter(fd => !before.has(fd)); console.log(JSON.stringify({ exitCode: last.exitCode, stderr: last.stderr.toString().trim(), leaked })); `; diff --git a/test/js/bun/spawn/spawn-pipe-read-error-leak.test.ts b/test/js/bun/spawn/spawn-pipe-read-error-leak.test.ts index 7ecf73cc7f27..1ea106616e91 100644 --- a/test/js/bun/spawn/spawn-pipe-read-error-leak.test.ts +++ b/test/js/bun/spawn/spawn-pipe-read-error-leak.test.ts @@ -99,19 +99,10 @@ async function once() { Bun.gc(true); - // The fd close goes through bun.Async.Closer which schedules close() on - // a jsc.WorkPool thread; Bun.sleep(0) doesn't synchronize with that. Poll - // a few times so a briefly-starved close thread doesn't flag a false - // positive (same pattern as spawn-stdin-pipe-fd-leak.test.ts). + // The reader closed its fd when it handled the failed read. let leaked = 0; - for (let i = 0; i < 20; i++) { - const final = snapshotFds(); - leaked = 0; - for (const [fd, link] of final) { - if (before.get(fd) !== link) leaked++; - } - if (leaked === 0) break; - await Bun.sleep(20); + for (const [fd, link] of snapshotFds()) { + if (before.get(fd) !== link) leaked++; } return leaked; } diff --git a/test/js/bun/spawn/spawn-pipe-start-error.test.ts b/test/js/bun/spawn/spawn-pipe-start-error.test.ts index 16b7a8982ab9..981147d6d233 100644 --- a/test/js/bun/spawn/spawn-pipe-start-error.test.ts +++ b/test/js/bun/spawn/spawn-pipe-start-error.test.ts @@ -224,8 +224,7 @@ if (kind !== "terminal") { globalThis.anchor.push(child); await child.exited; } -// The first spawnSync of a process creates the loop it waits on, which stays. Without pipes: theirs are closed on -// the work pool, some time after the call. +// The first spawnSync of a process creates the loop it waits on, which stays. const quiet = { cmd: ["true"], stdin: "ignore", stdout: "ignore", stderr: "ignore" }; if (kind.includes("unwatchable") || kind.startsWith("sync-") || kind.endsWith("-running")) Bun.spawnSync(quiet); const fdBaseline = openFds(); diff --git a/test/js/bun/spawn/spawn-stdin-pipe-fd-leak.test.ts b/test/js/bun/spawn/spawn-stdin-pipe-fd-leak.test.ts index e452319bf36f..fbfd279e0990 100644 --- a/test/js/bun/spawn/spawn-stdin-pipe-fd-leak.test.ts +++ b/test/js/bun/spawn/spawn-stdin-pipe-fd-leak.test.ts @@ -44,11 +44,8 @@ test.skipIf(!isPosix)("stdin: 'pipe' fd is closed on child exit without reading } await Promise.all(children.map(p => p.exited)); - // fd closes go through bun.Async.Closer on POSIX; give the close thread - // a moment — but do NOT invoke GC, since the bug was that cleanup only - // happened via Subprocess finalization. - for (let i = 0; i < 20 && countOpenFds() - baseline > N / 4; i++) await Bun.sleep(20); - + // Do NOT invoke GC: the bug was that cleanup only happened via Subprocess + // finalization. const afterExit = countOpenFds() - baseline; // Keep `children` alive across the measurement so GC finalization cannot @@ -56,8 +53,8 @@ test.skipIf(!isPosix)("stdin: 'pipe' fd is closed on child exit without reading expect(children.length).toBe(N); // Without the fix every stdin pipe (one per child) is still open here, - // so `afterExit` ≈ N. Allow slack for a few async closes still in flight. - expect(afterExit).toBeLessThan(N / 4); + // so `afterExit` ≈ N. + expect(afterExit).toBe(0); }); test("reading .stdin does not leak a native FileSink per spawn", async () => { diff --git a/test/js/bun/spawn/spawn-stdin-readable-stream-edge-cases.test.ts b/test/js/bun/spawn/spawn-stdin-readable-stream-edge-cases.test.ts index 58a9a7cbf5db..e9d448b97562 100644 --- a/test/js/bun/spawn/spawn-stdin-readable-stream-edge-cases.test.ts +++ b/test/js/bun/spawn/spawn-stdin-readable-stream-edge-cases.test.ts @@ -555,15 +555,7 @@ describe("spawn stdin ReadableStream edge cases", () => { } return pids.length; } - // Some are closed on the work pool, or by a collection. - async function leakedFds() { - const deadline = performance.now() + 2000; - while (openFds() > baseline && performance.now() < deadline) { - Bun.gc(true); - await Bun.sleep(5); - } - return openFds() - baseline; - } + const leakedFds = () => openFds() - baseline; `; async function run(code: string) { diff --git a/test/js/bun/spawn/spawn-streaming-stdin.test.ts b/test/js/bun/spawn/spawn-streaming-stdin.test.ts index fa978bb7f5a4..329561219289 100644 --- a/test/js/bun/spawn/spawn-streaming-stdin.test.ts +++ b/test/js/bun/spawn/spawn-streaming-stdin.test.ts @@ -63,13 +63,7 @@ test("spawn can write to stdin multiple chunks", async () => { remaining -= concurrency; } - // Pipe fds are closed on a background thread after their streams finish, so - // the last batch's closes may still be in flight here; wait for them to land. - let newMaxFD = getMaxFD(); - for (let i = 0; i < 100 && newMaxFD !== maxFD; i++) { - await Bun.sleep(10); - newMaxFD = getMaxFD(); - } + const newMaxFD = getMaxFD(); // assert we didn't leak any file descriptors expect(newMaxFD).toBe(maxFD); diff --git a/test/js/bun/spawn/spawn-streaming-stdout.test.ts b/test/js/bun/spawn/spawn-streaming-stdout.test.ts index 8b770854d17e..cffaf3b4b58c 100644 --- a/test/js/bun/spawn/spawn-streaming-stdout.test.ts +++ b/test/js/bun/spawn/spawn-streaming-stdout.test.ts @@ -40,28 +40,11 @@ test("spawn can read from stdout multiple chunks", async () => { await Promise.all(promises); i += concurrency; if (maxFD === -1) { - // Pipe fds are closed on a background thread after their streams finish, - // so the first batch's closes may still be in flight here; poll until the - // count settles so the baseline isn't inflated. maxFD = getMaxFD(); - for (let j = 0; j < 100; j++) { - await Bun.sleep(10); - const settled = getMaxFD(); - if (settled === maxFD) break; - maxFD = settled; - } } } - // Same race for the last batch's closes; wait for the fd count to drop back - // to the baseline before asserting. The settled baseline can still come out - // slightly high (it has no pre-spawn ground truth; #6724 moved it after the - // first batch on purpose), so ending below it is fine; only above is a leak. - let newMaxFD = getMaxFD(); - for (let i = 0; i < 100 && newMaxFD > maxFD; i++) { - await Bun.sleep(10); - newMaxFD = getMaxFD(); - } - expect(newMaxFD).toBeLessThanOrEqual(maxFD); + const newMaxFD = getMaxFD(); + expect(newMaxFD).toBe(maxFD); clearInterval(interval); await expectMaxObjectTypeCount(expect, "ReadableStream", 10); await expectMaxObjectTypeCount(expect, "ReadableStreamDefaultReader", 10); diff --git a/test/js/bun/spawn/spawnSync.test.ts b/test/js/bun/spawn/spawnSync.test.ts index 9ccf716fae0c..a0a0312fac75 100644 --- a/test/js/bun/spawn/spawnSync.test.ts +++ b/test/js/bun/spawn/spawnSync.test.ts @@ -310,9 +310,8 @@ it.skipIf(!isLinux)("a writer finalized during spawnSync leaves its fd number us let listing = Bun.spawnSync({ ...list, get maxBuffer() { Bun.gc(true); globalThis.writers = null; } }); let collectedDuringCall = 0; for (const ref of refs) if (ref.deref() === undefined) collectedDuringCall++; - // The fds are closed on the work pool. Once they are, dup() hands their numbers out again. - const deadline = performance.now() + 3000; - while (dupsOfStderr(listing) > 0 && performance.now() < deadline) listing = Bun.spawnSync(list); + // The fds were closed where the collected writers were finalized, which can be after that call listed them. + listing = Bun.spawnSync(list); const stillOpen = dupsOfStderr(listing); const writer = Bun.file(2).writer(); writer.write("registered"); diff --git a/test/js/bun/terminal/terminal.test.ts b/test/js/bun/terminal/terminal.test.ts index a83265912306..47c891aa985e 100644 --- a/test/js/bun/terminal/terminal.test.ts +++ b/test/js/bun/terminal/terminal.test.ts @@ -846,15 +846,13 @@ describe("Bun.Terminal", () => { function churn() { for (let i = 0; i < 500; i++) sink = { i, a: new Array(32).fill(i) }; } - // Collection runs the finalizer, which hands the remaining pty fds to - // a close thread; poll both conditions rather than a delay. - let fdsAfter = openFds(); - for (let i = 0; i < 50 && (collected < N || fdsAfter > fdsBefore); i++) { + // Collection runs the finalizer, which closes the remaining pty fds. + for (let i = 0; i < 50 && collected < N; i++) { churn(); Bun.gc(true); await new Promise(r => setImmediate(r)); - fdsAfter = openFds(); } + const fdsAfter = openFds(); console.log(JSON.stringify({ exits, collected, callbacksAfterExit, leakedFds: Math.max(0, fdsAfter - fdsBefore) })); `; diff --git a/test/js/bun/util/filesink.test.ts b/test/js/bun/util/filesink.test.ts index f84ba35f6e8d..32278507c370 100644 --- a/test/js/bun/util/filesink.test.ts +++ b/test/js/bun/util/filesink.test.ts @@ -4,6 +4,22 @@ import { bunEnv, bunExe, fileDescriptorLeakChecker, isLinux, isPosix, isWindows, import { mkfifo } from "mkfifo"; import { join } from "node:path"; +// For the scripts that these tests run: the open descriptors that refer to +// `target` (a path or a descriptor), found by device and inode. +const fdsOfSource = ` + function fdsOf(target) { + const { dev, ino } = typeof target === "number" ? fs.fstatSync(target) : fs.statSync(target); + const out = []; + for (let fd = 0; fd < 64; fd++) { + try { + const st = fs.fstatSync(fd); + if (st.dev === dev && st.ino === ino) out.push(fd); + } catch {} + } + return out; + } +`; + describe("FileSink", () => { const fixturesInput = [ [["abcdefghijklmnopqrstuvwxyz"], "abcdefghijklmnopqrstuvwxyz"], @@ -667,10 +683,11 @@ it("Bun.file(fd).writer() write/end under GC pressure does not crash", async () // Bun.file(fd).writer() dups the fd and registers the dup with the event loop. // When that registration fails, FileSink::setup closed the dup, but the writer -// still held it in its poll, so tearing the sink down queued a second close of -// the same fd number on the thread pool. By the time it ran the number usually -// belonged to someone else (here: the /dev/null fd opened right after), and in -// debug builds it tripped the EBADF assertion in Closer. +// still held it in its poll, so tearing the sink down closed the same fd number +// a second time. A sink on a socket closes its fd where it is torn down, so +// that second close gets EBADF, which Fd::close asserts on in debug and ASAN +// builds only. A second close that comes later closes the fd that has the +// number by then (here: the /dev/null fd opened right after), on every build. // // Linux-only: the registration failure is provoked through epoll keying its // entries on (open file, fd number), so a registration whose fd number is @@ -678,38 +695,11 @@ it("Bun.file(fd).writer() write/end under GC pressure does not crash", async () // lands on that number fails with EEXIST. kqueue drops the registration with // the fd, so this cannot be set up on macOS. it.skipIf(!isLinux)("Bun.file(fd).writer() whose registration fails closes the dup exactly once", async () => { - const fifoPath = join(tmpdirSync(), "parking.fifo"); - mkfifo(fifoPath, 0o666); const src = ` const { createSocketPair } = require("bun:internal-for-testing"); const fs = require("node:fs"); - const fifoPath = ${JSON.stringify(fifoPath)}; const [sock] = createSocketPair(); - - function fdsOf(target) { - const { dev, ino } = typeof target === "number" ? fs.fstatSync(target) : fs.statSync(target); - const out = []; - for (let fd = 0; fd < 64; fd++) { - try { - const st = fs.fstatSync(fd); - if (st.dev === dev && st.ino === ino) out.push(fd); - } catch {} - } - return out; - } - - // Park both thread-pool threads (UV_THREADPOOL_SIZE=2): each readFile opens - // the FIFO and blocks in read() until the write side is closed below, so a - // close queued on the pool by the failing writer() cannot run before we - // have reused its fd number. Holding the FIFO open read-write keeps the - // opens themselves from blocking; the loop waits for both to have opened. - const fifoWriter = fs.openSync(fifoPath, "r+"); - const parked = [1, 2].map(() => fs.promises.readFile(fifoPath)); - const parkedBy = performance.now() + 2000; - while (fdsOf(fifoPath).length < 3) { - if (performance.now() > parkedBy) throw new Error("the two readFile() calls never opened the FIFO"); - await Bun.sleep(1); - } + ${fdsOfSource} const before = fdsOf(sock); const first = Bun.file(sock).writer(); @@ -729,12 +719,8 @@ it.skipIf(!isLinux)("Bun.file(fd).writer() whose registration fails closes the d } // The failed writer() closed n; this unrelated fd gets the number back. const reused = fs.openSync("/dev/null", "r"); - - fs.closeSync(fifoWriter); - const parkedBytes = (await Promise.all(parked)).map(buf => buf.length); - // One more round trip through the now-free pool: whatever the failed - // writer() queued there has run by the time this resolves. - await fs.promises.stat("/dev/null"); + // A sink that the failed writer() left behind is torn down here. + Bun.gc(true); let reusedStillOpen = true; try { @@ -742,23 +728,485 @@ it.skipIf(!isLinux)("Bun.file(fd).writer() whose registration fails closes the d } catch { reusedStillOpen = false; } - console.log(JSON.stringify({ code, reusedIsN: reused === n, parkedBytes, reusedStillOpen })); + console.log(JSON.stringify({ code, reusedIsN: reused === n, reusedStillOpen })); // first must outlive the check: ending or collecting it closes n as well. first.unref(); `; await using proc = Bun.spawn({ cmd: [bunExe(), "-e", src], - env: { ...bunEnv, UV_THREADPOOL_SIZE: "2" }, + env: bunEnv, stderr: "pipe", }); const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); expect({ stdout: stdout.trim(), stderr, exitCode }).toEqual({ - stdout: JSON.stringify({ code: "EEXIST", reusedIsN: true, parkedBytes: [0, 0], reusedStillOpen: true }), + stdout: JSON.stringify({ code: "EEXIST", reusedIsN: true, reusedStillOpen: true }), stderr: "", exitCode: 0, }); }); +// On POSIX every pipe writer and every reader closes its descriptor through +// PollOrFd. It queued the close(2) on the thread pool and told the owner +// "closed" in the same call. The other end of a pipe sees that close only when +// it runs. So, until a pool thread was free, a child waited for the end of its +// stdin, a child that still wrote was not told that its reader was gone, and +// the child of a terminal got no SIGHUP. +// +// Each script parks both threads of the pool (UV_THREADPOOL_SIZE=2), so a close +// that waits for one of them cannot run. A row has the number of descriptors +// that are still open where the owner reports, and whether the child exits +// while the pool is parked. +// +// The rule is by the kind of the descriptor. A FIFO, a socket and the pty of +// Bun.Terminal close in place. A file, and a device that Bun.file() opened, +// still close on the pool: that close can block, and fstat cannot tell a pty +// from a serial line. +// Skipped on Windows: only the POSIX close is in place. +describe.skipIf(isWindows)("a pipe is closed when its owner reports closed", () => { + async function rowsWithParkedPool({ fifos = [], env = {} }: { fifos?: string[]; env?: typeof bunEnv }, rows: string) { + using dir = tempDir("pipe-closed", {}); + const parkingFifos = ["parking-1.fifo", "parking-2.fifo"]; + for (const fifo of [...parkingFifos, ...fifos]) mkfifo(join(String(dir), fifo), 0o666); + const src = ` + const fs = require("node:fs"); + const { join } = require("node:path"); + const dir = ${JSON.stringify(String(dir))}; + ${fdsOfSource} + function fdsWhere(test) { + const out = []; + for (let fd = 0; fd < 64; fd++) { + try { + if (test(fs.fstatSync(fd))) out.push(fd); + } catch {} + } + return out; + } + // Returns a function that counts the descriptors that were opened after + // this call and are still open: pipes (FIFOs and sockets) or terminals. + function opened(test = st => st.isFIFO() || st.isSocket()) { + const before = fdsWhere(test); + return () => fdsWhere(test).filter(fd => !before.includes(fd)).length; + } + const isTerminal = st => st.isCharacterDevice(); + async function until(condition, what) { + const deadline = performance.now() + 5000; + while (!condition()) { + if (performance.now() > deadline) throw new Error("timed out: " + what); + await Bun.sleep(1); + } + } + // Serves the file at path as the body of one response, reads the first + // bytes of it, calls afterFirstBytes() and drops the connection. Returns + // when the server has no request left: that is its report. + async function serveAndAbort(path, afterFirstBytes = () => {}) { + const server = Bun.serve({ port: 0, fetch: () => new Response(Bun.file(path)) }); + const controller = new AbortController(); + const response = await fetch("http://127.0.0.1:" + server.port + "/", { signal: controller.signal }); + const body = response.body.getReader(); + const { value } = await body.read(); + afterFirstBytes(); + controller.abort(); + await until(() => server.pendingRequests === 0, "the request is still pending"); + server.stop(true); + return new TextDecoder().decode(value); + } + + // Each readFile() opens a FIFO of its own and blocks a pool thread in + // read() until setFree() closes the write side of that FIFO. Holding the + // FIFO open read-write keeps the readFile()'s open from blocking; the + // loop waits for both to have opened. One FIFO for each thread: macOS + // reports the end to the first of two readers that wait on one FIFO and + // can leave the second one blocked. + const parking = ${JSON.stringify(parkingFifos)}.map(name => { + const path = join(dir, name); + const writer = fs.openSync(path, "r+"); + return { path, writer, read: fs.promises.readFile(path) }; + }); + const parkedBy = performance.now() + 5000; + while (parking.some(({ path }) => fdsOf(path).length < 2)) { + if (performance.now() > parkedBy) throw new Error("the two readFile() calls never opened their FIFOs"); + await Bun.sleep(1); + } + async function setFree() { + for (const { writer } of parking) fs.closeSync(writer); + await Promise.all(parking.map(({ read }) => read)); + } + + // What the first spawn and the first spawnSync open and keep is in every + // count's baseline. On macOS that is a kqueue, which fstat calls a FIFO. + await Bun.spawn({ cmd: ["true"], stdin: "ignore", stdout: "ignore", stderr: "ignore" }).exited; + Bun.spawnSync({ cmd: ["true"], stdin: "ignore", stdout: "ignore", stderr: "ignore" }); + + const rows = {}; + // By row: the exit of a child that waits for bun to close its end of a pipe. + const children = {}; + ${rows} + + // Whether each of these children has exited with the pool still parked. + // The deadline only ends a run in which one has not. + const exited = {}; + for (const row in children) children[row] = children[row].then(() => (exited[row] = true)); + let timer; + const deadline = new Promise(resolve => (timer = setTimeout(resolve, 2000))); + await Promise.race([Promise.all(Object.values(children)), deadline]); + clearTimeout(timer); + for (const row in children) rows[row].exitsWhileParked = exited[row] === true; + + await setFree(); + await Promise.all(Object.values(children)); + console.log(JSON.stringify(rows)); + `; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", src], + env: { ...bunEnv, UV_THREADPOOL_SIZE: "2", ...env }, + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + return { rows: stdout && JSON.parse(stdout), stderr, exitCode }; + } + + // cat exits when it reads the end of its stdin: that is when bun has closed + // its end of the pipe. + it("the stdin of a child process", async () => { + const result = await rowsWithParkedPool( + {}, + ` + const { spawn } = require("node:child_process"); + { + const open = opened(); + const child = Bun.spawn({ cmd: ["cat"], stdin: "pipe", stdout: "ignore", stderr: "inherit" }); + children["proc.stdin.end()"] = child.exited; + child.stdin.write("x"); + await child.stdin.end(); + rows["proc.stdin.end()"] = { open: open() }; + } + { + const open = opened(); + const child = spawn("cat", [], { stdio: ["pipe", "ignore", "inherit"] }); + children["child.stdin.end()"] = new Promise(resolve => child.on("exit", resolve)); + const closed = new Promise(resolve => child.stdin.on("close", resolve)); + child.stdin.end("x"); + await closed; + rows["child.stdin.end()"] = { open: open() }; + } + { + const open = opened(); + const child = spawn("cat", [], { stdio: ["pipe", "ignore", "inherit"] }); + children["child.stdin.destroy()"] = new Promise(resolve => child.on("exit", resolve)); + const closed = new Promise(resolve => child.stdin.on("close", resolve)); + child.stdin.destroy(); + await closed; + rows["child.stdin.destroy()"] = { open: open() }; + } + { + // Nothing reads proc.stdin: the exit of the child closes the pipe. + const open = opened(); + const child = Bun.spawn({ cmd: ["true"], stdin: "pipe", stdout: "ignore", stderr: "inherit" }); + await child.exited; + rows["proc.stdin of a child that has exited"] = { open: open() }; + } + { + // The shell writes the bytes to a pipe and then closes it. Nothing + // reports that close but the end of the command. + children["$\`cat < \${bytes}\`"] = Bun.$\`cat < \${new TextEncoder().encode("x")}\`.quiet().then(() => {}); + rows["$\`cat < \${bytes}\`"] = {}; + } + `, + ); + expect(result).toEqual({ + rows: { + "proc.stdin.end()": { open: 0, exitsWhileParked: true }, + "child.stdin.end()": { open: 0, exitsWhileParked: true }, + "child.stdin.destroy()": { open: 0, exitsWhileParked: true }, + "proc.stdin of a child that has exited": { open: 0 }, + "$`cat < ${bytes}`": { exitsWhileParked: true }, + }, + stderr: "", + exitCode: 0, + }); + }); + + // Bytes for stdin go through a pipe that bun writes and then closes. Nothing + // reports that close but the exit of the child. On Linux that pipe needs + // BUN_FEATURE_FLAG_DISABLE_MEMFD: a memfd has none. + it("bytes for the stdin of a child process", async () => { + const result = await rowsWithParkedPool( + { env: { BUN_FEATURE_FLAG_DISABLE_MEMFD: "1" } }, + ` + const child = Bun.spawn({ cmd: ["cat"], stdin: new TextEncoder().encode("x"), stdout: "ignore", stderr: "inherit" }); + children["Bun.spawn({ stdin: bytes })"] = child.exited; + rows["Bun.spawn({ stdin: bytes })"] = {}; + `, + ); + expect(result).toEqual({ + rows: { "Bun.spawn({ stdin: bytes })": { exitsWhileParked: true } }, + stderr: "", + exitCode: 0, + }); + }); + + // Here the thread that waits for the close is the one that called spawnSync. + it("a synchronous spawn", async () => { + const result = await rowsWithParkedPool( + { env: { BUN_FEATURE_FLAG_DISABLE_MEMFD: "1" } }, + ` + { + // The timeout only ends a run in which cat does not read the end of its stdin. + const open = opened(); + const { stdout, exitCode, exitedDueToTimeout = false } = Bun.spawnSync({ + cmd: ["cat"], + stdin: new TextEncoder().encode("x"), + stdout: "pipe", + stderr: "inherit", + timeout: 2000, + }); + rows["Bun.spawnSync({ stdin: bytes })"] = { stdout: stdout.toString(), exitCode, exitedDueToTimeout, open: open() }; + } + { + const open = opened(); + const { stdout, stderr } = Bun.spawnSync({ cmd: ["sh", "-c", "echo out; echo err >&2"], stdin: "ignore", stdout: "pipe", stderr: "pipe" }); + rows["Bun.spawnSync(), stdout and stderr"] = { out: stdout.toString(), err: stderr.toString(), open: open() }; + } + `, + ); + expect(result).toEqual({ + rows: { + "Bun.spawnSync({ stdin: bytes })": { stdout: "x", exitCode: 0, exitedDueToTimeout: false, open: 0 }, + "Bun.spawnSync(), stdout and stderr": { out: "out\n", err: "err\n", open: 0 }, + }, + stderr: "", + exitCode: 0, + }); + }); + + // yes writes until a write fails: that is when bun has closed its end of the + // pipe. + it("the stdout and stderr of a child process", async () => { + const result = await rowsWithParkedPool( + {}, + ` + { + const open = opened(); + const child = Bun.spawn({ cmd: ["sh", "-c", "echo out; echo err >&2"], stdin: "ignore", stdout: "pipe", stderr: "pipe" }); + const [out, err] = await Promise.all([child.stdout.text(), child.stderr.text()]); + await child.exited; + rows["proc.stdout and proc.stderr, read to the end"] = { out, err, open: open() }; + } + { + const open = opened(); + const child = Bun.spawn({ cmd: ["yes"], stdin: "ignore", stdout: "pipe", stderr: "ignore" }); + children["reader.cancel()"] = child.exited; + const reader = child.stdout.getReader(); + await reader.read(); + await reader.cancel(); + rows["reader.cancel()"] = { open: open() }; + } + { + // Nothing has read this stdout, so its reader has no poll yet. + const open = opened(); + const child = require("node:child_process").spawn("yes", [], { stdio: ["ignore", "pipe", "ignore"] }); + children["child.stdout.destroy()"] = new Promise(resolve => child.on("exit", resolve)); + child.stdout.destroy(); + rows["child.stdout.destroy()"] = { open: open() }; + } + { + // Nothing reads stdout or stderr, so neither reader has a poll when the child exits. + const open = opened(); + const child = require("node:child_process").spawn("true"); + await new Promise(resolve => child.on("close", resolve)); + rows["child_process.spawn(), nothing read"] = { open: open() }; + } + `, + ); + expect(result).toEqual({ + rows: { + "proc.stdout and proc.stderr, read to the end": { out: "out\n", err: "err\n", open: 0 }, + "reader.cancel()": { open: 0, exitsWhileParked: true }, + "child.stdout.destroy()": { open: 0, exitsWhileParked: true }, + "child_process.spawn(), nothing read": { open: 0 }, + }, + stderr: "", + exitCode: 0, + }); + }); + + it("a FIFO and a terminal", async () => { + const result = await rowsWithParkedPool( + { fifos: ["sink.fifo", "source.fifo", "body.fifo"] }, + ` + { + const path = join(dir, "sink.fifo"); + const reader = fs.openSync(path, fs.constants.O_RDONLY | fs.constants.O_NONBLOCK); + const writer = Bun.file(path).writer(); + writer.write("x"); + await writer.end(); + rows["Bun.file(fifo).writer().end()"] = { open: fdsOf(path).filter(fd => fd !== reader).length }; + fs.closeSync(reader); + } + { + const path = join(dir, "source.fifo"); + const writer = fs.openSync(path, "r+"); + const reader = Bun.file(path).stream().getReader(); + fs.writeSync(writer, "x"); + await reader.read(); + await reader.cancel(); + rows["Bun.file(fifo).stream(), reader.cancel()"] = { open: fdsOf(path).filter(fd => fd !== writer).length }; + fs.closeSync(writer); + } + { + // yes writes until the server has closed the only other descriptor of the FIFO. + const path = join(dir, "body.fifo"); + const holder = fs.openSync(path, "r+"); + const child = Bun.spawn({ cmd: ["sh", "-c", 'exec yes > "$0"', path], stdin: "ignore", stdout: "ignore", stderr: "ignore" }); + children["Bun.serve(), a FIFO as the body"] = child.exited; + await serveAndAbort(path, () => fs.closeSync(holder)); + rows["Bun.serve(), a FIFO as the body"] = { open: fdsOf(path).length }; + } + { + // cat gets SIGHUP when bun has closed every descriptor of its side of the terminal. + const open = opened(isTerminal); + const child = Bun.spawn({ cmd: ["cat"], terminal: { data() {} } }); + children["terminal.close()"] = child.exited; + child.terminal.close(); + rows["terminal.close()"] = { open: open() }; + } + `, + ); + expect(result).toEqual({ + rows: { + "Bun.file(fifo).writer().end()": { open: 0 }, + "Bun.file(fifo).stream(), reader.cancel()": { open: 0 }, + "Bun.serve(), a FIFO as the body": { open: 0, exitsWhileParked: true }, + "terminal.close()": { open: 0, exitsWhileParked: true }, + }, + stderr: "", + exitCode: 0, + }); + }); + + // Bun closes its own dup of a socket that the user gives to Bun.file(). The + // other end reads the end only when that dup was the last descriptor. + it("a socket that Bun.file() wraps", async () => { + const result = await rowsWithParkedPool( + {}, + ` + const { createSocketPair } = require("bun:internal-for-testing"); + // What the other end reads after the bytes that were written to it. + function otherEndReads(fd) { + for (;;) { + try { + if (fs.readSync(fd, Buffer.alloc(64)) === 0) return "the end"; + } catch (e) { + return e.code; + } + } + } + for (const last of [false, true]) { + const which = last ? ", the last descriptor" : ""; + { + const [socket, otherEnd] = createSocketPair(); + const open = opened(); + const reader = Bun.file(socket).stream().getReader(); + fs.writeSync(otherEnd, "x"); + await reader.read(); + if (last) fs.closeSync(socket); + await reader.cancel(); + rows["Bun.file(socket).stream(), reader.cancel()" + which] = { open: open(), otherEnd: otherEndReads(otherEnd) }; + if (!last) fs.closeSync(socket); + fs.closeSync(otherEnd); + } + { + const [socket, otherEnd] = createSocketPair(); + const open = opened(); + const writer = Bun.file(socket).writer(); + writer.write("x"); + if (last) fs.closeSync(socket); + await writer.end(); + rows["Bun.file(socket).writer().end()" + which] = { open: open(), otherEnd: otherEndReads(otherEnd) }; + if (!last) fs.closeSync(socket); + fs.closeSync(otherEnd); + } + } + `, + ); + expect(result).toEqual({ + rows: { + "Bun.file(socket).stream(), reader.cancel()": { open: 0, otherEnd: "EAGAIN" }, + "Bun.file(socket).writer().end()": { open: 0, otherEnd: "EAGAIN" }, + "Bun.file(socket).stream(), reader.cancel(), the last descriptor": { open: 0, otherEnd: "the end" }, + "Bun.file(socket).writer().end(), the last descriptor": { open: 0, otherEnd: "the end" }, + }, + stderr: "", + exitCode: 0, + }); + }); + + // The other side of the rule: a tty that Bun.file() opens is a character + // device, and so is a serial line or an input device, whose close can block. + // fstat cannot tell them apart, so these descriptors are still left to the + // pool, and each row counts the one that waits there. + // Linux-only: the path of the tty is read from /proc. + it.skipIf(!isLinux)("a tty that Bun.file() opened still closes on the pool", async () => { + const result = await rowsWithParkedPool( + {}, + ` + const terminal = new Bun.Terminal({}); + const path = fs + .readdirSync("/proc/self/fd") + .map(fd => { + try { + return fs.readlinkSync("/proc/self/fd/" + fd); + } catch { + return ""; + } + }) + .find(link => /^\\/dev\\/pts\\/\\d+$/.test(link)); + { + const before = fdsOf(path).length; + const reader = Bun.file(path).stream().getReader(); + terminal.write("x\\n"); + const { value } = await reader.read(); + await reader.cancel(); + rows["Bun.file(tty).stream(), reader.cancel()"] = { read: new TextDecoder().decode(value), open: fdsOf(path).length - before }; + } + { + // Ctrl-D on an empty line is the end of the input of a terminal. + const before = fdsOf(path).length; + const reader = Bun.file(path).stream().getReader(); + terminal.write("\\x04"); + const { done } = await reader.read(); + rows["Bun.file(tty).stream(), read to the end"] = { done, open: fdsOf(path).length - before }; + } + { + const before = fdsOf(path).length; + terminal.write("y\\n"); + const read = await serveAndAbort(path); + rows["Bun.serve(), a tty as the body"] = { read, open: fdsOf(path).length - before }; + } + { + // A new terminal: the sink has the side that bun does not make for Bun.Terminal. + const before = fdsOf("/dev/ptmx").length; + const writer = Bun.file("/dev/ptmx").writer(); + writer.write("x"); + await writer.end(); + rows["Bun.file(tty).writer().end()"] = { open: fdsOf("/dev/ptmx").length - before }; + } + terminal.close(); + `, + ); + expect(result).toEqual({ + rows: { + "Bun.file(tty).stream(), reader.cancel()": { read: "x\n", open: 1 }, + "Bun.file(tty).stream(), read to the end": { done: true, open: 1 }, + "Bun.serve(), a tty as the body": { read: "y\n", open: 1 }, + "Bun.file(tty).writer().end()": { open: 1 }, + }, + stderr: "", + exitCode: 0, + }); + }); +}); + // Skipped on Windows: the Windows FileSink writer hands bytes to uv_fs_write on // the libuv threadpool and never registers an AutoFlusher synchronously, so the // on_exit drain this suite exercises is a no-op there and every process.exit()