diff --git a/src/io/lib.rs b/src/io/lib.rs index 8a9776c9a63..7c83974ff0b 100644 --- a/src/io/lib.rs +++ b/src/io/lib.rs @@ -1117,8 +1117,7 @@ impl IoRequestLoop { match (request.callback)(request) { Action::Readable(readable) => { Poll::apply_kqueue( - ApplyAction::Readable, - readable.tag, + ApplyAction::Readable(readable.tag), readable.poll, readable.fd, add_one(&mut events_list), @@ -1126,8 +1125,7 @@ impl IoRequestLoop { } Action::Writable(writable) => { Poll::apply_kqueue( - ApplyAction::Writable, - writable.tag, + ApplyAction::Writable(writable.tag), writable.poll, writable.fd, add_one(&mut events_list), @@ -1137,13 +1135,7 @@ impl IoRequestLoop { if close.poll.flags.contains(Flags::PollReadable) || close.poll.flags.contains(Flags::PollWritable) { - Poll::apply_kqueue( - ApplyAction::Cancel, - close.tag, - close.poll, - close.fd, - add_one(&mut events_list), - ); + close.poll.unregister_kqueue(self.pollfd(), close.fd); } (close.on_done)(close.ctx); } @@ -1160,8 +1152,6 @@ impl IoRequestLoop { c_int::try_from(change_count).expect("int cast"), // The same array may be used for the changelist and eventlist. events_list.as_mut_ptr(), - // we set 0 here so that if we get an error on - // registration, it becomes errno c_int::try_from(capacity).expect("int cast"), core::ptr::null(), ); @@ -1357,7 +1347,6 @@ pub struct CloseAction<'a> { pub fd: Fd, pub poll: &'a mut Poll, pub ctx: *mut (), - pub tag: PollableTag, pub on_done: fn(*mut ()), } @@ -1496,8 +1485,10 @@ pub type FlagsSet = enumset::EnumSet; #[cfg(any(target_os = "macos", target_os = "freebsd"))] #[derive(PartialEq, Eq, Clone, Copy)] enum ApplyAction { - Readable, - Writable, + /// One-shot `EV_ADD` whose events dispatch to the owner the tag names. + Readable(PollableTag), + Writable(PollableTag), + /// `EV_DELETE` of whatever the poll still has armed; dispatches to nobody. Cancel, } @@ -1506,7 +1497,6 @@ impl Poll { #[inline] pub(crate) fn apply_kqueue( action: ApplyAction, - tag: PollableTag, poll: &mut Poll, fd: Fd, kqueue_event: &mut KEvent, @@ -1514,23 +1504,31 @@ impl Poll { log!( "register({}, {})", match action { - ApplyAction::Readable => "readable", - ApplyAction::Writable => "writable", + ApplyAction::Readable(_) => "readable", + ApplyAction::Writable(_) => "writable", ApplyAction::Cancel => "cancel", }, fd ); let one_shot_flag = libc::EV_ONESHOT; - let udata: usize = Pollable::init(tag, std::ptr::from_mut::(poll)).ptr() as usize; - let (filter, flags_): (i16, u16) = match action { - ApplyAction::Readable => (libc::EVFILT_READ, libc::EV_ADD | one_shot_flag), - ApplyAction::Writable => (libc::EVFILT_WRITE, libc::EV_ADD | one_shot_flag), + let poll_ptr = std::ptr::from_mut::(poll); + let (filter, flags_, udata): (i16, u16, usize) = match action { + ApplyAction::Readable(tag) => ( + libc::EVFILT_READ, + libc::EV_ADD | one_shot_flag, + Pollable::init(tag, poll_ptr).ptr() as usize, + ), + ApplyAction::Writable(tag) => ( + libc::EVFILT_WRITE, + libc::EV_ADD | one_shot_flag, + Pollable::init(tag, poll_ptr).ptr() as usize, + ), ApplyAction::Cancel => { if poll.flags.contains(Flags::PollReadable) { - (libc::EVFILT_READ, libc::EV_DELETE) + (libc::EVFILT_READ, libc::EV_DELETE, 0) } else if poll.flags.contains(Flags::PollWritable) { - (libc::EVFILT_WRITE, libc::EV_DELETE) + (libc::EVFILT_WRITE, libc::EV_DELETE, 0) } else { unreachable!() } @@ -1561,10 +1559,10 @@ impl Poll { } match action { - ApplyAction::Readable => { + ApplyAction::Readable(_) => { poll.flags.insert(Flags::PollReadable); } - ApplyAction::Writable => { + ApplyAction::Writable(_) => { poll.flags.insert(Flags::PollWritable); } ApplyAction::Cancel => { @@ -1602,6 +1600,23 @@ impl Poll { self.flags.remove(Flags::Registered); } + /// Not batched into the next `kevent()`: the caller hands the owner off right after this. + #[cfg(any(target_os = "macos", target_os = "freebsd"))] + pub(crate) fn unregister_kqueue(&mut self, watcher_fd: Fd, fd: Fd) { + let mut change: KEvent = bun_core::ffi::zeroed(); + Poll::apply_kqueue(ApplyAction::Cancel, self, fd, &mut change); + // nevents = 0: apply only. ENOENT (one-shot already fired) and EBADF + // (fd already closed) are the usual outcomes; nothing to do for either. + let _ = kevent_call( + watcher_fd.native(), + &raw const change, + 1, + core::ptr::null_mut(), + 0, + core::ptr::null(), + ); + } + #[cfg(any(target_os = "macos", target_os = "freebsd"))] pub(crate) fn on_update_kqueue(event: KEvent) { #[cfg(target_os = "macos")] @@ -1620,7 +1635,10 @@ impl Poll { // CYCLEBREAK: owner (ReadFile/WriteFile) is T6; dispatch via link-time // `extern "Rust"` defined in `bun_runtime::dispatch`. The // container_of(io_poll) recovery happens there. - if event.flags == libc::EV_ERROR { + // + // xnu ORs EV_ERROR into the flags of a rejected change (EV_ADD|EV_ONESHOT + // |EV_ERROR); FreeBSD replaces them. Only the bit is common to both. + if (event.flags & libc::EV_ERROR) != 0 { log!("error({}) = {}", event.ident, event.data); // SAFETY: poll is the `io_poll` field of a live owner; link-time // extern body matches on `tag`. diff --git a/src/runtime/webcore/Blob.rs b/src/runtime/webcore/Blob.rs index 006cf841e31..4834e2169ca 100644 --- a/src/runtime/webcore/Blob.rs +++ b/src/runtime/webcore/Blob.rs @@ -7031,8 +7031,6 @@ pub trait FileOpener: Sized { // TODO: move to bun_sys? pub trait FileCloser: Sized { - const IO_TAG: bun_io::Tag; - fn opened_fd(&self) -> Fd; fn set_opened_fd(&mut self, fd: Fd); fn close_after_io(&self) -> bool; @@ -7109,15 +7107,13 @@ pub trait FileCloser: Sized { } /// Implements [`FileCloser`] for a task struct with the standard field set -/// (`opened_fd`, `close_after_io`, `state`, `io_request`, `io_poll`, `task`), -/// an inherent `update()`, and a [`bun_io::Tag`] variant named after the type. -/// The type must also carry `bun_threading::intrusive_work_task!` and -/// `bun_io::intrusive_io_request!`, which provide the parent-pointer recovery -/// used by the two trampolines. +/// (`opened_fd`, `close_after_io`, `state`, `io_request`, `io_poll`, `task`) +/// and an inherent `update()`. The type must also carry +/// `bun_threading::intrusive_work_task!` and `bun_io::intrusive_io_request!`, +/// which provide the parent-pointer recovery used by the two trampolines. macro_rules! impl_file_closer { ($T:ident) => { impl crate::webcore::blob::FileCloser for $T { - const IO_TAG: ::bun_io::Tag = ::bun_io::Tag::$T; fn opened_fd(&self) -> ::bun_sys::Fd { self.opened_fd } @@ -7164,7 +7160,6 @@ macro_rules! impl_file_closer { fd, poll, ctx: this.cast::<()>(), - tag: ::IO_TAG, on_done, }) } diff --git a/src/runtime/webcore/blob/read_file.rs b/src/runtime/webcore/blob/read_file.rs index 8bc36a6b580..18c0bcafa0d 100644 --- a/src/runtime/webcore/blob/read_file.rs +++ b/src/runtime/webcore/blob/read_file.rs @@ -960,7 +960,6 @@ impl<'a> FileOpener for ReadFileUV<'a> { #[cfg(windows)] impl<'a> FileCloser for ReadFileUV<'a> { - const IO_TAG: bun_io::Tag = bun_io::Tag::ReadFile; fn opened_fd(&self) -> Fd { self.opened_fd } diff --git a/test/internal/source-lints/kevent-ev-error-equality.test.ts b/test/internal/source-lints/kevent-ev-error-equality.test.ts new file mode 100644 index 00000000000..2291e7794da --- /dev/null +++ b/test/internal/source-lints/kevent-ev-error-equality.test.ts @@ -0,0 +1,144 @@ +import { file } from "bun"; +import { expect, test } from "bun:test"; +import { realpathSync } from "fs"; +import path from "path"; +import { globAllSources } from "../../../scripts/glob-sources.ts"; + +// A kevent changelist entry the kernel rejects is handed back in the eventlist +// with EV_ERROR set in `flags` and the errno in `data`. How it is "set" differs +// between the two kqueue kernels we build for: +// +// xnu (bsd/kern/kern_event.c, kevent_register) kev->flags |= EV_ERROR; +// FreeBSD (sys/kern/kern_event.c, kqueue_kevent) kevp->flags = EV_ERROR; +// +// so on macOS the reply to a rejected `EV_ADD|EV_ONESHOT` reads 0x4011, not +// 0x4000, and `flags == EV_ERROR` is false for it: the io request loop +// (src/io/lib.rs) dispatched such a reply as a ready event, and FilePoll +// (src/io/posix_event_loop.rs) used to swallow its change errors the same way +// (#31701). Both kernels agree on the bit, so that is what gets tested: +// +// flags == EV_ERROR / flags != EV::ERROR -> (flags & EV_ERROR) != 0 +// +// Both of the io layer's kqueue users inherited the equality from the Zig +// original; this keeps it from coming back in either. Scope is the Rust tree +// because that is what .github/workflows/source-lints.yml runs this for; the C +// kqueue code in packages/bun-usockets is outside its triggers. + +const root = path.resolve(import.meta.dir, "..", "..", ".."); +const rustSources = globAllSources().rust.filter(abs => abs.endsWith(".rs")); + +// Only scan files tracked in HEAD (a `git stash` round-trip can leave stray +// `.rs` files in the working tree; CI runs on a clean checkout). +const tracked: Set | null = (() => { + const r = Bun.spawnSync({ + cmd: ["git", "-C", root, "ls-tree", "-r", "--name-only", "-z", "HEAD"], + stdout: "pipe", + stderr: "ignore", + }); + if (!r.success) return null; + return new Set(r.stdout.toString().split("\0").filter(Boolean)); +})(); + +// The constant in any of its spellings: `EV_ERROR`, `libc::EV_ERROR`, +// `bun_sys::darwin::EV::ERROR`, ... +const CONSTANT = String.raw`(?:\w+::)*EV(?:_ERROR|::ERROR)\b`; +// `x & EV_ERROR`: the bit test. `(?(); +let scanned = 0; +for (const abs of rustSources) { + const source = path.relative(root, abs).replaceAll(path.sep, "/"); + // `src/cli` is a symlink into `src/runtime/cli`; count each file once under + // its canonical path. + if (path.relative(root, realpathSync(abs)).replaceAll(path.sep, "/") !== source) continue; + if (tracked !== null && !tracked.has(source)) continue; + scanned++; + const content = await file(abs).text(); + if (!content.includes("EV_ERROR") && !content.includes("EV::ERROR")) continue; + for (const [index, line] of content.split("\n").entries()) { + if (hasBitTest(line)) filesWithBitTests.add(source); + if (comparesWholeWord(line)) offenders.push(`${source}:${index + 1}: ${line.trim()}`); + } +} + +test("scans a non-empty set of tracked Rust sources", () => { + // Guards against the tracked/realpath filters above over-firing and leaving + // nothing to scan, which would make the ban below pass vacuously. + expect(scanned).toBeGreaterThan(0); +}); + +test("the scan still sees the bit tests that are known to be in the tree", () => { + // Named files rather than a count: if the file set or the comment handling + // ever stops seeing one of these, the ban below would be vacuous for it. + // The io request loop's event dispatch, FilePoll's register/unregister and + // the process reaper's kevent loop. + expect([...filesWithBitTests].sort()).toEqual( + expect.arrayContaining(["src/io/lib.rs", "src/io/posix_event_loop.rs", "src/spawn/process.rs"]), + ); +}); + +test("the pattern recognizes the spellings it claims to", () => { + const banned = [ + "if event.flags == libc::EV_ERROR {", + "if changelist[0].flags == EV::ERROR {", + "if ev.flags != bun_sys::darwin::EV::ERROR {", + "if libc::EV_ERROR == event.flags {", + "let failed = event.flags == (EV::ERROR);", + "if event.flags == EV_ERROR && event.data != 0 {", + // A bit test elsewhere on the line does not excuse a whole-word compare. + "if (event.flags & libc::EV_ERROR) != 0 && event.flags == libc::EV_ERROR {", + "let whole = flags == EV::ERROR || (flags & EV::ERROR) == EV::ERROR;", + ]; + // Bit tests: allowed, and what the liveness check above counts. + const bitTests = [ + "if (event.flags & libc::EV_ERROR) != 0 {", + "if (changelist[0].flags & EV::ERROR) != 0 && changelist[0].data != 0 {", + "if (changelist[i].flags & EV::ERROR) == 0 || changelist[i].data == 0 {", + "if (event.flags & EV_ERROR) == EV_ERROR {", + "if (event.flags & libc::EV_ERROR) != libc::EV_ERROR {", + "if r.flags & libc::EV_ERROR == 0 || r.data == 0 {", + "if event.data != 0 && event.flags & EV_ERROR != 0 {", + "let is_error = event.flags & EV::ERROR != 0;", + "let rejected = (kev.flags & EV::ERROR) != 0; // not kev.flags == EV::ERROR", + ]; + // Neither: mentions of the constant that are not tests of a flags word. + const neither = [ + "pub const ERROR: u16 = libc::EV_ERROR;", + " // xnu ORs EV_ERROR in, so `flags == EV_ERROR` is the bug this comment is about", + ]; + expect(banned.filter(s => !comparesWholeWord(s))).toEqual([]); + expect(bitTests.filter(s => !hasBitTest(s) || comparesWholeWord(s))).toEqual([]); + expect(neither.filter(s => hasBitTest(s) || comparesWholeWord(s))).toEqual([]); +}); + +test("kevent EV_ERROR is tested as a bit, never compared against the whole flags word", () => { + expect(offenders).toEqual([]); +}); diff --git a/test/js/bun/io/bun-write.test.js b/test/js/bun/io/bun-write.test.js index f02bdf4f8e1..193a2726939 100644 --- a/test/js/bun/io/bun-write.test.js +++ b/test/js/bun/io/bun-write.test.js @@ -11,6 +11,7 @@ import { tempDir, withoutAggressiveGC, } from "harness"; +import { mkfifo } from "mkfifo"; import path, { join } from "path"; let i = 0; @@ -537,6 +538,46 @@ const IS_UV_FS_COPYFILE_DISABLED = expect(exitCode).toBe(0); }); + // Bun.write(Bun.file(fd), bytes) and Bun.file(fd).text() on a pipe run on + // the thread pool and, whenever poll() says the pipe is not ready, park on + // the io thread (src/io/lib.rs IoRequestLoop: epoll on Linux, kqueue on + // macOS) until it is. The reader below starts on an empty FIFO and the + // writer has several times more than any platform's pipe buffer holds, so + // both of them take that detour, repeatedly, before this resolves. + it.skipIf(isWindows)("Bun.write and Bun.file(fd).text() on a non-blocking FIFO wait for each other", async () => { + using dir = tempDir("bun-write-fifo", {}); + const fifo = join(String(dir), "data.fifo"); + mkfifo(fifo); + // O_RDWR: opening a FIFO for both directions never blocks waiting for a + // peer, and keeps a writer attached so the reader cannot see EOF early. + // O_NONBLOCK: a full pipe reports EAGAIN to the writer instead of blocking + // a pool thread. Two separate fds because epoll registers an fd only once, + // and the reader and the writer each register their own. + const flags = fs.constants.O_RDWR | fs.constants.O_NONBLOCK; + const readFd = fs.openSync(fifo, flags); + const writeFd = fs.openSync(fifo, flags); + try { + // 256 KiB is also the size from which Bun.write() always goes to the + // thread pool instead of first trying a synchronous write. + const payload = Buffer.alloc(256 * 1024, "0123456789abcdef\n").toString(); + + const reading = Bun.file(readFd).slice(0, payload.length).text(); + + const destination = Bun.file(writeFd); + // Bun.write() only waits for an fd destination to become writable once + // the fd's type has been resolved; reading .size does that up front. + destination.size; + const writing = Bun.write(destination, payload); + + const [text, written] = await Promise.all([reading, writing]); + expect({ written, length: text.length }).toEqual({ written: payload.length, length: payload.length }); + expect(text).toBe(payload); + } finally { + fs.closeSync(writeFd); + fs.closeSync(readFd); + } + }); + it("Bun.file(0) survives GC", async () => { for (let i = 0; i < 10; i++) { let f = Bun.file(0);