From f09256f3596dd88a1a9eea0f720e10375e591fb5 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Wed, 12 Aug 2026 07:51:46 +0000 Subject: [PATCH 1/4] io: hand ReadFile/WriteFile to the io thread through the owner's pointer IoRequestLoop::schedule took &mut Request. The io thread recovers the owning ReadFile/WriteFile from exactly the pointer that was pushed, by container-of, so a pointer made from a &mut of the io_request field does not reach the owner under the aliasing rules (bun_core::container_of's contract says as much), and the reference stayed protected across the push and wake() while the io thread may already be writing the request. The three callers (wait_for_readable, wait_for_writable, FileCloser::do_close) reached it from &mut self chains, so the publishing thread kept protected references to the object on its stack while the io thread, a second pool thread, or (through io_task.finish()) the JS thread wrote to or freed it. schedule now takes *mut Request, projected from the owner. The io thread passes the popped pointer itself to the request callback, and the returned Action carries *mut Poll projected the same way, so the address registered with epoll/kqueue (which the readiness dispatch turns back into the owner) has the owner's provenance too. The callbacks take the pointer. On the pool side, the re-entries that already held the object's pointer (do_read_loop_task, do_write_loop_task, on_close_io_request) now pass it down: update, the read/write loops, on_finish, the two waits and do_close take this: *mut Self, keep their accesses statement scoped, and end with the hand-over. WriteFile::do_write no longer waits from inside; it reports WouldBlock and the loop hands the file over as its last step, where before the caller went on to read self.errno after the object had been handed to the io thread. FileCloser::update, which nothing called, is removed. The first step of a job (JobContext::run -> get_fd -> run_async_with_fd / run_with_fd) still runs under the &mut those traits pass; it now does its work first and hands over through a pointer as its last act, and the traits are left for a separate change. The WorkPool hand-overs in these files (on_ready, on_io_error, on_io_request_closed) are converted separately as well and are not touched here. A source lint bans scheduling an io request spelled from self or through a &mut reborrow; it reports the three converted sites on main. Tests cover the FIFO paths that go through the io thread: reading a FIFO by path, and Bun.write() to a non-blocking FIFO that is already full. --- src/io/lib.rs | 297 +++++++----- src/runtime/webcore/Blob.rs | 130 +++--- src/runtime/webcore/blob/read_file.rs | 425 +++++++++++------- src/runtime/webcore/blob/write_file.rs | 231 +++++++--- .../self-receiver-io-schedule.test.ts | 185 ++++++++ test/js/bun/io/bun-write.test.js | 76 ++++ test/js/bun/util/bun-file-read.test.ts | 53 ++- 7 files changed, 1007 insertions(+), 390 deletions(-) create mode 100644 test/internal/source-lints/self-receiver-io-schedule.test.ts diff --git a/src/io/lib.rs b/src/io/lib.rs index 8a9776c9a637..fc6aea655789 100644 --- a/src/io/lib.rs +++ b/src/io/lib.rs @@ -910,15 +910,35 @@ impl IoRequestLoop { unsafe { (*LOOP.get()).assume_init_ref() }.tick(); } - /// Enqueue `request` for the IO thread to pick up. Safe to call from any + /// Hand the owner of `request` over to the IO thread. Callable from any /// thread: only touches the lock-free `pending` queue and the /// async-signal-safe `waker`. This is the *only* cross-thread entry /// point — every other `IoRequestLoop` method is IO-thread-only. - pub fn schedule(request: &mut Request) { + /// + /// The push is the hand-over: from that instant the IO thread may pop the + /// request, clear `scheduled`, run its callback and (through `on_error` / + /// `on_done`) pass the owner on to a pool thread, all before this returns + /// from `wake()`. That is why this takes the pointer rather than `&mut + /// Request` (a reference argument would stay protected across the push), + /// and why the pointer has to be projected from the owner: the IO thread + /// hands this exact pointer to [`Request::callback`], which recovers the + /// owner from it by container-of, so its provenance must cover the owner, + /// which a `&mut self.io_request` reborrow's does not. + /// + /// # Safety + /// `request` is `&raw mut (*owner).io_request` of a live owner that the + /// calling thread currently holds, with `scheduled == false`. Once this is + /// called the owner belongs to the IO thread: the caller must not touch the + /// request or its owner again (not even through a reference it already + /// holds) until the IO thread hands it back. + pub unsafe fn schedule(request: *mut Request) { Self::ensure_init(); - debug_assert!(!request.scheduled); - request.scheduled = true; - let request = core::ptr::NonNull::from(request); + let request = core::ptr::NonNull::new(request).expect("io request of a live owner"); + // SAFETY: fn contract — live and ours until the push below. + unsafe { + debug_assert!(!(*request.as_ptr()).scheduled); + (*request.as_ptr()).scheduled = true; + } // SAFETY: `ONCE` above established happens-before for `load()`'s // init of `pending`/`waker`. `get_unchecked` (no owner assert) and a // pointer cast of the `repr(transparent)` `MaybeUninit` (not @@ -926,6 +946,7 @@ impl IoRequestLoop { // references formed are the `&pending` / `&waker` autorefs, which may // coexist with the IO thread's `&IoRequestLoop` held across `tick()`. // `pending.push` (lock-free MPSC) and `waker.wake` both take `&self`. + // Nothing here touches `*request` after the push. unsafe { let loop_p = LOOP.get_unchecked().cast::(); (*core::ptr::addr_of!((*loop_p).pending)).push(request); @@ -933,6 +954,26 @@ impl IoRequestLoop { } } + /// IO thread: a request just popped from `pending` is this thread's now. + /// Clears its `scheduled` bit and asks the owner what to do with it. + /// + /// The callback gets `request` as popped, i.e. the pointer `schedule` was + /// given, so the owner it recovers by container-of is in bounds of it. The + /// returned action's `poll` points into that owner, which stays this + /// thread's only until the action's `on_error` / `on_done` hands it on: + /// every use of `poll` has to come before that call. + #[cfg(not(windows))] + fn take_request(request: *mut Request) -> Action { + // SAFETY: `pending` only holds pointers that went through `schedule`, + // whose contract makes each the request field of a live owner, with + // the owner's provenance, that nobody else touches until the IO thread + // hands it on; popping it made it this thread's. + unsafe { + (*request).scheduled = false; + ((*request).callback)(request) + } + } + #[cfg(not(windows))] pub(crate) fn tick(&self) { // SAFETY: literal is NUL-terminated; len excludes the NUL. @@ -958,6 +999,19 @@ impl IoRequestLoop { } } + /// The `Readable` / `Writable` arm of [`tick_epoll`](Self::tick_epoll). + #[cfg(any(target_os = "linux", target_os = "android"))] + fn register_epoll(watcher_fd: Fd, file: &FileAction, flag: Flags) { + // SAFETY: `take_request` — the owner is ours until `on_error` hands it + // on, and this is the last use of `poll` before that. + let registered = unsafe { + Poll::register_for_epoll(file.poll, flag, file.tag, watcher_fd, true, file.fd) + }; + if let Err(err) = registered { + (file.on_error)(file.ctx, &err); + } + } + #[cfg(any(target_os = "linux", target_os = "android"))] pub(crate) fn tick_epoll(&self) { loop { @@ -967,53 +1021,34 @@ impl IoRequestLoop { let watcher_fd = self.pollfd(); loop { - let request_ptr = pending.next(); - if request_ptr.is_null() { + let request = pending.next(); + if request.is_null() { break; } - // SAFETY: pop_batch yields live nodes pushed by `schedule()`. - let request = unsafe { &mut *request_ptr }; - request.scheduled = false; - match (request.callback)(request) { - Action::Readable(readable) => { - match readable.poll.register_for_epoll( - Flags::PollReadable, - readable.tag, - watcher_fd, - true, - readable.fd, - ) { - Err(err) => { - (readable.on_error)(readable.ctx, &err); - } - Ok(()) => {} - } + match Self::take_request(request) { + Action::Readable(file) => { + Self::register_epoll(watcher_fd, &file, Flags::PollReadable) } - Action::Writable(writable) => { - match writable.poll.register_for_epoll( - Flags::PollWritable, - writable.tag, - watcher_fd, - true, - writable.fd, - ) { - Err(err) => { - (writable.on_error)(writable.ctx, &err); - } - Ok(()) => {} - } + Action::Writable(file) => { + Self::register_epoll(watcher_fd, &file, Flags::PollWritable) } Action::Close(close) => { - log!( - "close({}, registered={})", - close.fd, - close.poll.flags.contains(Flags::Registered) - ); - // Only remove from the interest list if it was previously registered. - // Otherwise, epoll gets confused. - // This state can happen if polling for readable/writable previously failed. - if close.poll.flags.contains(Flags::WasEverRegistered) { - close.poll.unregister_with_fd(watcher_fd, close.fd); + // SAFETY: `take_request` — the owner is ours until + // `on_done` hands it on; the reborrow ends with this + // block. + unsafe { + let poll = &mut *close.poll; + log!( + "close({}, registered={})", + close.fd, + poll.flags.contains(Flags::Registered) + ); + // Only remove from the interest list if it was previously registered. + // Otherwise, epoll gets confused. + // This state can happen if polling for readable/writable previously failed. + if poll.flags.contains(Flags::WasEverRegistered) { + poll.unregister_with_fd(watcher_fd, close.fd); + } } (close.on_done)(close.ctx); } @@ -1107,44 +1142,55 @@ impl IoRequestLoop { } loop { - let request_ptr = pending.next(); - if request_ptr.is_null() { + let request = pending.next(); + if request.is_null() { break; } - // SAFETY: pop_batch yields live nodes pushed by `schedule()`. - let request = unsafe { &mut *request_ptr }; - request.scheduled = false; - match (request.callback)(request) { + match Self::take_request(request) { Action::Readable(readable) => { - Poll::apply_kqueue( - ApplyAction::Readable, - readable.tag, - readable.poll, - readable.fd, - add_one(&mut events_list), - ); + // SAFETY: `take_request` — the owner stays ours + // until the kernel reports the fd (no hand-on in + // this arm). + unsafe { + Poll::apply_kqueue( + ApplyAction::Readable, + readable.tag, + readable.poll, + readable.fd, + add_one(&mut events_list), + ); + } } Action::Writable(writable) => { - Poll::apply_kqueue( - ApplyAction::Writable, - writable.tag, - writable.poll, - writable.fd, - add_one(&mut events_list), - ); - } - Action::Close(close) => { - if close.poll.flags.contains(Flags::PollReadable) - || close.poll.flags.contains(Flags::PollWritable) - { + // SAFETY: as for `Readable`. + unsafe { Poll::apply_kqueue( - ApplyAction::Cancel, - close.tag, - close.poll, - close.fd, + ApplyAction::Writable, + writable.tag, + writable.poll, + writable.fd, add_one(&mut events_list), ); } + } + Action::Close(close) => { + // SAFETY: `take_request` — the owner is ours until + // `on_done` hands it on, and this block is the last + // use of `poll` before that. + unsafe { + let flags = (*close.poll).flags; + if flags.contains(Flags::PollReadable) + || flags.contains(Flags::PollWritable) + { + Poll::apply_kqueue( + ApplyAction::Cancel, + close.tag, + close.poll, + close.fd, + add_one(&mut events_list), + ); + } + } (close.on_done)(close.ctx); } } @@ -1190,15 +1236,26 @@ impl IoRequestLoop { // ─── Request ────────────────────────────────────────────────────────────────── +/// What the IO thread does with a popped [`Request`]. Receives the pointer +/// [`IoRequestLoop::schedule`] was given (the owner's `io_request` field, with +/// the owner's provenance), so the trampoline may recover the owner from it +/// with [`IntrusiveIoRequest::from_io_request`]. `scheduled` has already been +/// cleared when this runs. +/// +/// # Safety +/// Only the IO thread may call it, with a pointer that went through +/// `schedule` and has not been handed back since. +pub type RequestCallback = unsafe fn(*mut Request) -> Action; + pub struct Request { pub next: bun_threading::Link, - pub callback: for<'a> fn(&'a mut Request) -> Action<'a>, + pub callback: RequestCallback, pub scheduled: bool, } impl Request { #[inline] - pub fn new(callback: for<'a> fn(&'a mut Request) -> Action<'a>) -> Self { + pub fn new(callback: RequestCallback) -> Self { Self { next: bun_threading::Link::new(), callback, @@ -1216,7 +1273,7 @@ impl Request { /// followed by a full fence (matches the existing pattern in /// `webcore::blob::{read_file,write_file}`). #[inline] - pub fn store_callback_seq_cst(&mut self, cb: for<'a> fn(&'a mut Request) -> Action<'a>) { + pub fn store_callback_seq_cst(&mut self, cb: RequestCallback) { // SAFETY: `callback` is a plain pointer-sized field on `self`; // volatile write prevents the compiler from reordering or eliding it. unsafe { core::ptr::write_volatile(&raw mut self.callback, cb) }; @@ -1232,7 +1289,7 @@ impl Request { /// A type that embeds an intrusive `io_request: `[`Request`] field. Declares the /// byte offset once and provides the canonical container-of recovery used by -/// every `fn(&mut Request) -> Action` io-loop trampoline. +/// every [`RequestCallback`] io-loop trampoline. /// /// Implement via [`intrusive_io_request!`]. /// @@ -1250,7 +1307,9 @@ pub unsafe trait IntrusiveIoRequest: Sized { /// # Safety /// `req` must point to the [`Request`] field at `Self::IO_REQUEST_OFFSET` /// inside a live `Self` allocation that was scheduled via that field, and - /// the pointer's provenance must cover the whole allocation. + /// the pointer's provenance must cover the whole allocation. The pointer a + /// [`RequestCallback`] receives satisfies this: it is the one + /// [`IoRequestLoop::schedule`] requires to be projected from the owner. #[inline(always)] unsafe fn from_io_request(req: *mut Request) -> *mut Self { // SAFETY: caller upholds the trait safety contract above. @@ -1261,7 +1320,7 @@ pub unsafe trait IntrusiveIoRequest: Sized { /// Implements [`IntrusiveIoRequest`] for a struct that embeds an intrusive /// `io_request: `[`Request`] field. Brings /// [`IntrusiveIoRequest::from_io_request`] into scope for the type's -/// `fn(&mut Request) -> Action` trampolines. +/// [`RequestCallback`] trampolines. #[macro_export] macro_rules! intrusive_io_request { ($ty:ty, $field:ident) => { @@ -1339,23 +1398,32 @@ pub(crate) type RequestQueue = bun_threading::UnboundedQueue; // ─── Action ─────────────────────────────────────────────────────────────────── -pub enum Action<'a> { - Readable(FileAction<'a>), - Writable(FileAction<'a>), - Close(CloseAction<'a>), +/// Returned by a [`RequestCallback`]. `poll` and `ctx` point into the +/// request's owner, which is the IO thread's until the action's `on_error` / +/// `on_done` hands it on; `poll` is a pointer (projected from the owner, like +/// the request itself) rather than a `&mut` both so the IO thread holds no +/// reference into the owner across that hand-on and because the address +/// registered with the kernel is derived from it and later turned back into +/// the owner by container-of (`__bun_io_pollable_on_ready`). +pub enum Action { + Readable(FileAction), + Writable(FileAction), + Close(CloseAction), } -pub struct FileAction<'a> { +pub struct FileAction { pub fd: Fd, - pub poll: &'a mut Poll, + /// `&raw mut (*owner).io_poll`. + pub poll: *mut Poll, pub ctx: *mut (), pub tag: PollableTag, pub on_error: fn(*mut (), &sys::Error), } -pub struct CloseAction<'a> { +pub struct CloseAction { pub fd: Fd, - pub poll: &'a mut Poll, + /// `&raw mut (*owner).io_poll`. + pub poll: *mut Poll, pub ctx: *mut (), pub tag: PollableTag, pub on_done: fn(*mut ()), @@ -1502,12 +1570,18 @@ enum ApplyAction { } impl Poll { + /// `poll` is an [`Action`]'s `poll` (see there): the address the kernel + /// hands back as `udata` is taken from it as given, so that the owner + /// recovered from it in `on_update_kqueue`'s dispatch is in bounds of it. + /// + /// # Safety + /// `poll` is the `io_poll` of a live owner the IO thread currently holds. #[cfg(any(target_os = "macos", target_os = "freebsd"))] #[inline] - pub(crate) fn apply_kqueue( + pub(crate) unsafe fn apply_kqueue( action: ApplyAction, tag: PollableTag, - poll: &mut Poll, + poll: *mut Poll, fd: Fd, kqueue_event: &mut KEvent, ) { @@ -1522,7 +1596,10 @@ impl Poll { ); let one_shot_flag = libc::EV_ONESHOT; - let udata: usize = Pollable::init(tag, std::ptr::from_mut::(poll)).ptr() as usize; + let udata: usize = Pollable::init(tag, poll).ptr() as usize; + // SAFETY: fn contract; the reborrow lasts for this call only, and the + // `udata` the kernel keeps was taken from the pointer above it. + let poll = unsafe { &mut *poll }; 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), @@ -1675,11 +1752,18 @@ impl Poll { } } + /// `poll` is an [`Action`]'s `poll` (see there): the address the kernel + /// hands back in `epoll_event.u64` is taken from it as given, so that the + /// owner recovered from it in `on_update_epoll`'s dispatch is in bounds of + /// it. + /// + /// # Safety + /// `poll` is the `io_poll` of a live owner the IO thread currently holds. #[cfg(any(target_os = "linux", target_os = "android"))] // `enumset::EnumSetType` cannot be a const generic, so `flag` is a runtime // arg. The `match` below preserves the exhaustiveness check. - pub(crate) fn register_for_epoll( - &mut self, + pub(crate) unsafe fn register_for_epoll( + poll: *mut Poll, flag: Flags, tag: PollableTag, watcher_fd: Fd, @@ -1690,11 +1774,16 @@ impl Poll { debug_assert!(fd != Fd::INVALID); + let udata = Pollable::init(tag, poll).ptr(); + // SAFETY: fn contract; the reborrow lasts for this call only, and the + // `udata` the kernel keeps was taken from the pointer above it. + let this = unsafe { &mut *poll }; + if one_shot { - self.flags.insert(Flags::OneShot); + this.flags.insert(Flags::OneShot); } - let one_shot_flag: u32 = if !self.flags.contains(Flags::OneShot) { + let one_shot_flag: u32 = if !this.flags.contains(Flags::OneShot) { 0 } else { linux::EPOLL_ONESHOT @@ -1712,11 +1801,11 @@ impl Poll { let mut event = linux::epoll_event { events: flags, - u64: Pollable::init(tag, std::ptr::from_mut::(self)).ptr(), + u64: udata, }; - let op: i32 = if self.flags.contains(Flags::WasEverRegistered) - || self.flags.contains(Flags::NeedsRearm) + let op: i32 = if this.flags.contains(Flags::WasEverRegistered) + || this.flags.contains(Flags::NeedsRearm) { linux::EPOLL_CTL_MOD } else { @@ -1740,10 +1829,10 @@ impl Poll { // Only mark if it successfully registered. // If it failed to register, we don't want to unregister it later if // it never had done so in the first place. - self.flags.insert(Flags::Registered); - self.flags.insert(Flags::WasEverRegistered); + this.flags.insert(Flags::Registered); + this.flags.insert(Flags::WasEverRegistered); - self.flags.insert(match flag { + this.flags.insert(match flag { Flags::PollReadable => Flags::PollReadable, Flags::PollProcess => { if cfg!(any(target_os = "linux", target_os = "android")) { @@ -1755,7 +1844,7 @@ impl Poll { Flags::PollWritable => Flags::PollWritable, _ => unreachable!(), }); - self.flags.remove(Flags::NeedsRearm); + this.flags.remove(Flags::NeedsRearm); Ok(()) } diff --git a/src/runtime/webcore/Blob.rs b/src/runtime/webcore/Blob.rs index 006cf841e312..01d495bcb77a 100644 --- a/src/runtime/webcore/Blob.rs +++ b/src/runtime/webcore/Blob.rs @@ -7037,17 +7037,30 @@ pub trait FileCloser: Sized { fn set_opened_fd(&mut self, fd: Fd); fn close_after_io(&self) -> bool; fn state(&self) -> &core::sync::atomic::AtomicU8; - fn io_request(&mut self) -> Option<&mut bun_io::Request>; + /// The embedded request [`do_close`](Self::do_close) hands to the io + /// thread, projected from `this` (`&raw mut (*this).io_request`, never a + /// `&mut` reborrow: `bun_io::IoRequestLoop::schedule` wants a pointer the + /// whole `Self` is reachable through, because `schedule_close` gets it + /// back and recovers `Self` from it). `None` when the type never goes + /// through the io thread (`ReadFileUV`). + /// + /// # Safety + /// `this` points at a live `Self`. + unsafe fn io_request(this: *mut Self) -> Option<*mut bun_io::Request>; fn io_poll(&mut self) -> &mut bun_io::Poll; fn task(&mut self) -> &mut bun_jsc::WorkPoolTask; - fn update(&mut self); #[cfg(windows)] fn loop_(&self) -> *mut bun_libuv_sys::uv_loop_t; - /// Intrusive backref: Rust `offset_of!` cannot name - /// fields on a trait `Self`, so each concrete impl supplies its own - /// container_of recovery (no default body). - fn schedule_close(request: &mut bun_io::Request) -> bun_io::Action<'_>; + /// The io thread's side of `do_close`: a `bun_io::RequestCallback`. + /// Intrusive backref: Rust `offset_of!` cannot name fields on a trait + /// `Self`, so each concrete impl supplies its own container_of recovery + /// (no default body). + /// + /// # Safety + /// `bun_io::RequestCallback`'s contract: io thread, `request` is the + /// pointer `do_close` scheduled. + unsafe fn schedule_close(request: *mut bun_io::Request) -> bun_io::Action; fn on_io_request_closed(this: &mut Self) { this.io_poll() @@ -7068,40 +7081,48 @@ pub trait FileCloser: Sized { /// `container_of` deref locally, so a fn-level qualifier is redundant. fn on_close_io_request(task: *mut bun_jsc::WorkPoolTask); - fn do_close(&mut self, is_allowed_to_close_fd: bool) -> bool { - // Check `close_after_io()` before `io_request()` so the immutable - // `self` reads finish before - // taking the `&mut self` borrow via `io_request()`. - if self.close_after_io() { - self.state().store( - ClosingState::Closing as u8, - core::sync::atomic::Ordering::SeqCst, - ); - if let Some(io_request) = self.io_request() { - // The io thread reads `callback` after popping from its MPSC - // queue; a plain store here is a data race. `bun_io::Request:: - // store_callback_seq_cst` lowers to a volatile write + SeqCst - // fence (Rust has no `AtomicFnPtr`). - io_request.store_callback_seq_cst(Self::schedule_close); - if !io_request.scheduled { - bun_io::IoRequestLoop::schedule(io_request); + /// Returns `true` when the close was handed to the io thread instead: + /// `*this` belongs to it from that moment (it comes back through + /// `on_close_io_request`), which is why this takes the pointer the caller + /// holds rather than `&mut self`, and why the hand-over is the last thing + /// here that touches `*this`. + /// + /// # Safety + /// `this` is a live `Self` that the calling thread currently holds. After + /// a `true` return the caller must not touch `*this` again. + unsafe fn do_close(this: *mut Self, is_allowed_to_close_fd: bool) -> bool { + // SAFETY: fn contract. Each access below is a place expression or a + // reborrow that ends with its statement; nothing follows the schedule. + unsafe { + if (*this).close_after_io() { + (*this).state().store( + ClosingState::Closing as u8, + core::sync::atomic::Ordering::SeqCst, + ); + if let Some(io_request) = Self::io_request(this) { + // The io thread reads `callback` after popping from its MPSC + // queue; a plain store here is a data race. `bun_io::Request:: + // store_callback_seq_cst` lowers to a volatile write + SeqCst + // fence (Rust has no `AtomicFnPtr`). + (*io_request).store_callback_seq_cst(Self::schedule_close); + if !(*io_request).scheduled { + bun_io::IoRequestLoop::schedule(io_request); + } + return true; } - return true; } - } - if is_allowed_to_close_fd - && self.opened_fd() != Fd::INVALID - && self.opened_fd().stdio_tag().is_none() - { - #[cfg(windows)] - bun_io::Closer::close(self.opened_fd(), self.loop_()); - #[cfg(not(windows))] - { - use bun_sys::FdExt as _; - let _ = self.opened_fd().close_allowing_bad_file_descriptor(None); + let fd = (*this).opened_fd(); + if is_allowed_to_close_fd && fd != Fd::INVALID && fd.stdio_tag().is_none() { + #[cfg(windows)] + bun_io::Closer::close(fd, (*this).loop_()); + #[cfg(not(windows))] + { + use bun_sys::FdExt as _; + let _ = fd.close_allowing_bad_file_descriptor(None); + } + (*this).set_opened_fd(Fd::INVALID); } - self.set_opened_fd(Fd::INVALID); } false @@ -7110,7 +7131,9 @@ 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. +/// an inherent `unsafe fn update(this: *mut Self)` (the pool re-entry after +/// the io thread closed the request), 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. @@ -7130,8 +7153,9 @@ macro_rules! impl_file_closer { fn state(&self) -> &::core::sync::atomic::AtomicU8 { &self.state } - fn io_request(&mut self) -> Option<&mut ::bun_io::Request> { - Some(&mut self.io_request) + unsafe fn io_request(this: *mut Self) -> Option<*mut ::bun_io::Request> { + // SAFETY: fn contract; a projection, not a reborrow. + Some(unsafe { &raw mut (*this).io_request }) } fn io_poll(&mut self) -> &mut ::bun_io::Poll { &mut self.io_poll @@ -7139,27 +7163,25 @@ macro_rules! impl_file_closer { fn task(&mut self) -> &mut ::bun_jsc::WorkPoolTask { &mut self.task } - fn update(&mut self) { - $T::update(self) - } #[cfg(windows)] fn loop_(&self) -> *mut ::bun_libuv_sys::uv_loop_t { unreachable!() } - fn schedule_close(request: &mut ::bun_io::Request) -> ::bun_io::Action<'_> { + unsafe fn schedule_close(request: *mut ::bun_io::Request) -> ::bun_io::Action { use ::bun_io::IntrusiveIoRequest as _; - // SAFETY: `request` is `&mut self.io_request` (intrusive); recover parent. - let this = unsafe { $T::from_io_request(::core::ptr::from_mut(request)) }; + // SAFETY: fn contract — `request` is the `&raw mut (*this).io_request` + // that `do_close` scheduled, so the parent is live and reachable + // through it. + let this = unsafe { $T::from_io_request(request) }; fn on_done(ctx: *mut ()) { // SAFETY: ctx is `self as *mut Self` set below. let this = unsafe { ::bun_ptr::callback_ctx::<$T>(ctx.cast()) }; <$T as crate::webcore::blob::FileCloser>::on_io_request_closed(this); } - // SAFETY: `request` is `&mut self.io_request` (intrusive), so `this` is the - // live parent; the `fd` copy and the `io_poll` field borrow are the only - // borrows formed. - let (fd, poll) = unsafe { ((*this).opened_fd, &mut (*this).io_poll) }; + // SAFETY: as above; the `fd` read ends here and `io_poll` is only + // projected, so no reference into the parent outlives this call. + let (fd, poll) = unsafe { ((*this).opened_fd, &raw mut (*this).io_poll) }; ::bun_io::Action::Close(::bun_io::CloseAction { fd, poll, @@ -7180,10 +7202,12 @@ macro_rules! impl_file_closer { // `&mut self.task` (intrusive) registered in `on_io_request_closed`; // recover parent. let this = unsafe { $T::from_task_ptr(task) }; - // SAFETY: `this` is the live parent (see above); scoped access. - unsafe { (*this).close_after_io = false }; - // SAFETY: as above; exclusive borrow scoped to the call. - $T::update(unsafe { &mut *this }); + // SAFETY: `this` is the live parent (see above) and this thread's + // until `update` hands it on, which is why it gets the pointer. + unsafe { + (*this).close_after_io = false; + $T::update(this); + } } } }; diff --git a/src/runtime/webcore/blob/read_file.rs b/src/runtime/webcore/blob/read_file.rs index 8bc36a6b5805..936da379a8ae 100644 --- a/src/runtime/webcore/blob/read_file.rs +++ b/src/runtime/webcore/blob/read_file.rs @@ -19,7 +19,7 @@ use bun_core; use bun_core::String as BunString; use bun_io as io; #[cfg(not(windows))] -use bun_io::FileAction; +use bun_io::{FileAction, IntrusiveIoRequest as _}; #[cfg(windows)] // `bun_jsc::EventLoop` is the *module*; the struct is one level deeper. use bun_jsc::event_loop::EventLoop; @@ -292,6 +292,15 @@ pub struct ReadFile { bun_threading::intrusive_work_task!(ReadFile, task); bun_io::intrusive_io_request!(ReadFile, io_request); +/// What a pool-side step of a `ReadFile` ends in; see [`ReadFile::proceed`]. +#[cfg(not(windows))] +#[derive(Clone, Copy)] +enum Next { + ReadLoop, + WaitForReadable, + Finish, +} + // The default methods on the FileOpener/FileCloser traits provide the bodies. impl FileOpener for ReadFile { fn opened_fd(&self) -> Fd { @@ -330,17 +339,57 @@ impl FileOpener for ReadFile { crate::webcore::blob::impl_file_closer!(ReadFile); impl ReadFile { - pub(crate) fn update(&mut self) { + /// Pool thread re-entry (`do_read_loop_task`, `on_close_io_request`). + /// + /// This and everything it calls take the pointer the pool handed us + /// instead of `&mut self`: each of these paths ends by handing `*this` to + /// another thread (`wait_for_readable` / `do_close` to the io thread, + /// `io_task.finish()` to the JS thread), and a `&mut self` argument would + /// still be live, and protected, on this stack while that thread writes to + /// or frees the object. Accesses are scoped to end before the hand-over. + /// + /// # Safety + /// `this` is the live `ReadFile` this thread currently holds; the caller + /// must not touch it afterwards. + pub(crate) unsafe fn update(this: *mut Self) { #[cfg(windows)] { - return; // why + let _ = this; + // Windows reads go through ReadFileUV; this is never scheduled. } #[cfg(not(windows))] { - if self.state.load(Ordering::Relaxed) == ClosingState::Closing as u8 { - self.on_finish(); - } else { - self.do_read_loop(); + // SAFETY: fn contract; the load ends before `this` is passed on. + let closing = + unsafe { (*this).state.load(Ordering::Relaxed) } == ClosingState::Closing as u8; + // SAFETY: fn contract, passed through. + unsafe { + Self::proceed( + this, + if closing { + Next::Finish + } else { + Next::ReadLoop + }, + ) + } + } + } + + /// Performs the step a pool-side function decided on once its own + /// accesses to `*this` are over. `ReadLoop` reads until it has to hand the + /// object on; the other two are the hand-overs themselves. + /// + /// # Safety + /// As [`update`](Self::update). + #[cfg(not(windows))] + unsafe fn proceed(this: *mut Self, next: Next) { + // SAFETY: fn contract, passed through. + unsafe { + match next { + Next::ReadLoop => Self::do_read_loop(this), + Next::WaitForReadable => Self::wait_for_readable(this), + Next::Finish => Self::on_finish(this), } } } @@ -433,35 +482,50 @@ impl ReadFile { unsafe { (*ctx.cast::()).on_io_error(err) } } + /// io thread: the `io::RequestCallback` installed by `wait_for_readable`. + /// + /// # Safety + /// `io::RequestCallback`'s contract: `request` is the pointer + /// `wait_for_readable` scheduled. #[cfg(not(windows))] - pub(crate) fn on_request_readable(request: &mut io::Request) -> io::Action<'_> { + pub(crate) unsafe fn on_request_readable(request: *mut io::Request) -> io::Action { bloblog!("ReadFile.onRequestReadable"); - request.scheduled = false; - // SAFETY: request points to ReadFile.io_request (intrusive field); recover parent via offset_of. - let this: &mut ReadFile = unsafe { - &mut *(bun_core::from_field_ptr!( - ReadFile, - io_request, - std::ptr::from_mut::(request) - )) - }; + // SAFETY: fn contract — `request` is `&raw mut (*this).io_request` of a + // live `ReadFile`, so the parent is reachable through it. + let this = unsafe { ReadFile::from_io_request(request) }; + // SAFETY: as above; the `fd` read ends here and `io_poll` is only + // projected, so no reference into the parent outlives this call. + let (fd, poll) = unsafe { ((*this).opened_fd, &raw mut (*this).io_poll) }; io::Action::Readable(FileAction { on_error: Self::on_io_error_thunk, - ctx: std::ptr::from_mut::(this).cast::<()>(), - fd: this.opened_fd, - poll: &mut this.io_poll, + ctx: this.cast::<()>(), + fd, + poll, tag: ReadFile::IO_TAG, }) } + /// Hands `*this` to the io thread until `opened_fd` is readable; it comes + /// back through `on_ready` / `on_io_error`. The schedule is the last access + /// (see `update` for why this is pointer-shaped). + /// + /// # Safety + /// `this` is the live `ReadFile` this thread currently holds; the caller + /// must not touch it afterwards. #[cfg(not(windows))] - pub(crate) fn wait_for_readable(&mut self) { + pub(crate) unsafe fn wait_for_readable(this: *mut Self) { bloblog!("ReadFile.waitForReadable"); - self.close_after_io = true; - self.io_request - .store_callback_seq_cst(Self::on_request_readable); - if !self.io_request.scheduled { - io::IoRequestLoop::schedule(&mut self.io_request); + // SAFETY: fn contract. The `store_callback_seq_cst` reborrow ends with + // its statement; the request is projected (not reborrowed) for + // `schedule`, which needs the whole `ReadFile` reachable through it. + unsafe { + (*this).close_after_io = true; + (*this) + .io_request + .store_callback_seq_cst(Self::on_request_readable); + if !(*this).io_request.scheduled { + io::IoRequestLoop::schedule(&raw mut (*this).io_request); + } } } @@ -627,20 +691,33 @@ impl ReadFile { self.file_store.pathlike.is_path() } + /// Ends with one of two hand-overs: `do_close` gives `*this` to the io + /// thread, otherwise `io_task.finish()` gives it to the JS thread, which + /// frees it. Pointer-shaped for that reason (see `update`). + /// + /// # Safety + /// `this` is the live `ReadFile` this thread currently holds; the caller + /// must not touch it afterwards. #[cfg(not(windows))] - fn on_finish(&mut self) { - let close_after_io = self.close_after_io; - self.size = self.buffer.len() as SizeType; + unsafe fn on_finish(this: *mut Self) { + // SAFETY: fn contract; these reads and the `is_allowed_to_close` + // reborrow end before `do_close`. + let (close_after_io, is_allowed_to_close_fd) = unsafe { + (*this).size = (*this).buffer.len() as SizeType; + ((*this).close_after_io, (*this).is_allowed_to_close()) + }; - { - if self.do_close(self.is_allowed_to_close()) { - bloblog!("ReadFile.onFinish() = deferred"); - // we have to wait for the close to finish - return; - } + // SAFETY: fn contract, passed through; on `true` the io thread owns + // `*this` and nothing below runs. + if unsafe { Self::do_close(this, is_allowed_to_close_fd) } { + bloblog!("ReadFile.onFinish() = deferred"); + // we have to wait for the close to finish + return; } if !close_after_io { - if let Some(io_task) = self.io_task.take() { + // SAFETY: fn contract — `do_close` returned `false`, so `*this` is + // still ours; the `take` ends before `finish` posts the job. + if let Some(io_task) = unsafe { (*this).io_task.take() } { bloblog!("ReadFile.onFinish() = immediately"); io_task.finish(); } @@ -700,16 +777,31 @@ impl ReadFile { } } + /// `FileOpener::get_fd`'s callback, reached from `JobContext::run`. Both + /// of those hand this frame `&mut self`, so unlike the pool re-entries + /// (`update`) a reference to the object is still live up the stack during + /// the hand-over; this frame at least finishes its own use of `self` + /// first and hands over through a pointer as its last act. #[cfg(not(windows))] fn run_async_with_fd(&mut self, fd: Fd) { + let next = self.prepare_read(fd); + let this: *mut Self = self; + // SAFETY: `this` is the live `ReadFile` `get_fd` called us with, and + // this frame does not use `self` or `this` after the call. + unsafe { Self::proceed(this, next) } + } + + /// The part of starting a read that happens before anything is handed + /// on: stat, buffer sizing, and the initial readability check. + #[cfg(not(windows))] + fn prepare_read(&mut self, fd: Fd) -> Next { if self.errno.is_some() { - self.on_finish(); - return; + return Next::Finish; } self.resolve_size_and_last_modified(fd); if self.errno.is_some() { - return self.on_finish(); + return Next::Finish; } // Special files might report a size of > 0, and be wrong. @@ -720,8 +812,7 @@ impl ReadFile { // default — `then()` reads `self.buffer` directly. self.byte_store = ByteStore::default(); - self.on_finish(); - return; + return Next::Finish; } // add an extra 16 bytes to the buffer to avoid having to resize it for trailing extra data @@ -735,8 +826,7 @@ impl ReadFile { .to_system_error() .into(), ); - self.on_finish(); - return; + return Next::Finish; } self.buffer = v; } @@ -753,146 +843,151 @@ impl ReadFile { // // If we immediately call read(), it will block until stdin is // readable. - if self.could_block { - if bun_core::is_readable(fd) == bun_core::Pollable::NotReady { - self.wait_for_readable(); - return; - } + if self.could_block && bun_core::is_readable(fd) == bun_core::Pollable::NotReady { + return Next::WaitForReadable; } - self.do_read_loop(); + Next::ReadLoop } fn do_read_loop_task(task: *mut WorkPoolTask) { // SAFETY: only reached via `WorkPoolTask::callback` with `task` = - // `&mut self.task` (intrusive) registered in `on_writable`/`init`; - // recover parent. - let this = unsafe { &mut *ReadFile::from_task_ptr(task) }; + // `&raw mut self.task` (intrusive) scheduled by `on_ready` / + // `on_io_error`; recover the parent, which is this thread's until + // `update` hands it on. + unsafe { Self::update(ReadFile::from_task_ptr(task)) } + } - this.update(); + /// # Safety + /// As [`update`](Self::update). + #[cfg(not(windows))] + unsafe fn do_read_loop(this: *mut Self) { + // SAFETY: fn contract; the reborrow ends with the call, before + // `proceed` hands the object on through `this`. + let next = unsafe { (*this).read_until_blocked() }; + // SAFETY: fn contract, passed through. + unsafe { Self::proceed(this, next) } } + /// Reads until the read is complete (`Finish`: EOF, `max_length`, or an + /// error recorded in `errno`) or the fd would block (`WaitForReadable`). + /// Never produces `ReadLoop`. #[cfg(not(windows))] - fn do_read_loop(&mut self) { - #[cfg(not(windows))] - { - // we hold a 64 KB stack buffer incase the amount of data to - // be read is greater than the reported amount - // - // 64 KB is large, but since this is running in a thread - // with it's own stack, it should have sufficient space. - // hoisted out of the loop and zero-initialized once — the - // one-time 64 KB memset is negligible next to the per-iteration - // syscall, and avoids the `MaybeUninit` → `&mut [u8]` cast (uninit - // bytes behind a `&[u8]` is technically UB even when never read). - let mut stack_buffer = [0u8; 64 * 1024]; - // `do_read` never touches `self.buffer`; move it out so the read - // target slice (which may point into its spare capacity) can be - // held as a safe `&mut [u8]` across the `&mut self` call. - let mut buffer = core::mem::take(&mut self.buffer); - while self.state.load(Ordering::Relaxed) == ClosingState::Running as u8 { - let (use_stack, buf) = Self::remaining_buffer( - &mut buffer, - &mut stack_buffer, - self.max_length, - self.read_off, - ); + fn read_until_blocked(&mut self) -> Next { + // we hold a 64 KB stack buffer incase the amount of data to + // be read is greater than the reported amount + // + // 64 KB is large, but since this is running in a thread + // with it's own stack, it should have sufficient space. + // hoisted out of the loop and zero-initialized once — the + // one-time 64 KB memset is negligible next to the per-iteration + // syscall, and avoids the `MaybeUninit` → `&mut [u8]` cast (uninit + // bytes behind a `&[u8]` is technically UB even when never read). + let mut stack_buffer = [0u8; 64 * 1024]; + // `do_read` never touches `self.buffer`; move it out so the read + // target slice (which may point into its spare capacity) can be + // held as a safe `&mut [u8]` across the `&mut self` call. + let mut buffer = core::mem::take(&mut self.buffer); + while self.state.load(Ordering::Relaxed) == ClosingState::Running as u8 { + let (use_stack, buf) = Self::remaining_buffer( + &mut buffer, + &mut stack_buffer, + self.max_length, + self.read_off, + ); - if !buf.is_empty() && self.errno.is_none() && !self.read_eof { - let mut read_amount: usize = 0; - let mut retry = false; - let continue_reading = self.do_read(buf, &mut read_amount, &mut retry); - - // We might read into the stack buffer, so we need to copy it into the heap. - if use_stack { - // `do_read` wrote `read_amount` initialized bytes at - // `stack_buffer[..read_amount]`; the stack array is live - // for this iteration. - let read = &stack_buffer[..read_amount]; - if buffer.capacity() == 0 { - // We need to allocate a new buffer - // In this case, we want to use `ensureTotalCapacityPrecise` so that it's an exact amount - // We want to avoid over-allocating incase it's a large amount of data sent in a single chunk followed by a 0 byte chunk. - buffer.reserve_exact(read.len()); - } else { - buffer.reserve(read.len()); - } - buffer.extend_from_slice(read); + if !buf.is_empty() && self.errno.is_none() && !self.read_eof { + let mut read_amount: usize = 0; + let mut retry = false; + let continue_reading = self.do_read(buf, &mut read_amount, &mut retry); + + // We might read into the stack buffer, so we need to copy it into the heap. + if use_stack { + // `do_read` wrote `read_amount` initialized bytes at + // `stack_buffer[..read_amount]`; the stack array is live + // for this iteration. + let read = &stack_buffer[..read_amount]; + if buffer.capacity() == 0 { + // We need to allocate a new buffer + // In this case, we want to use `ensureTotalCapacityPrecise` so that it's an exact amount + // We want to avoid over-allocating incase it's a large amount of data sent in a single chunk followed by a 0 byte chunk. + buffer.reserve_exact(read.len()); } else { - // record the amount of data read - // SAFETY: read() wrote `read_amount` initialized bytes into spare capacity. - unsafe { bun_core::vec::commit_spare(&mut buffer, read_amount) }; - } - // - If they DID set a max length, we should stop - // reading after that. - // - // - If they DID NOT set a max_length, then it will - // be Blob.max_size which is an impossibly large - // amount to read. - if !self.read_eof && buffer.len() >= self.max_length as usize { - break; + buffer.reserve(read.len()); } + buffer.extend_from_slice(read); + } else { + // record the amount of data read + // SAFETY: read() wrote `read_amount` initialized bytes into spare capacity. + unsafe { bun_core::vec::commit_spare(&mut buffer, read_amount) }; + } + // - If they DID set a max length, we should stop + // reading after that. + // + // - If they DID NOT set a max_length, then it will + // be Blob.max_size which is an impossibly large + // amount to read. + if !self.read_eof && buffer.len() >= self.max_length as usize { + break; + } - if !continue_reading { - // Stop reading, we errored - break; - } + if !continue_reading { + // Stop reading, we errored + break; + } - // If it's not a regular file, it might be something - // which would block on the next read. So we should - // avoid immediately reading again until the next time - // we're scheduled to read. - // - // An example of where this happens is stdin. - // - // await Bun.stdin.text(); - // - // If we immediately call read(), it will block until stdin is - // readable. - if retry - || (self.could_block - // If we received EOF, we can skip the poll() system - // call. We already know it's done. - && !self.read_eof) + // If it's not a regular file, it might be something + // which would block on the next read. So we should + // avoid immediately reading again until the next time + // we're scheduled to read. + // + // An example of where this happens is stdin. + // + // await Bun.stdin.text(); + // + // If we immediately call read(), it will block until stdin is + // readable. + if retry + || (self.could_block + // If we received EOF, we can skip the poll() system + // call. We already know it's done. + && !self.read_eof) + { + if self.could_block + // If we received EOF, we can skip the poll() system + // call. We already know it's done. + && !self.read_eof { - if self.could_block - // If we received EOF, we can skip the poll() system - // call. We already know it's done. - && !self.read_eof - { - match bun_core::is_readable(self.opened_fd) { - bun_core::Pollable::NotReady => {} - bun_core::Pollable::Ready | bun_core::Pollable::Hup => continue, - } + match bun_core::is_readable(self.opened_fd) { + bun_core::Pollable::NotReady => {} + bun_core::Pollable::Ready | bun_core::Pollable::Hup => continue, } - self.read_eof = false; - self.buffer = buffer; - self.wait_for_readable(); - - return; } - - // There can be more to read - continue; + self.read_eof = false; + self.buffer = buffer; + return Next::WaitForReadable; } - // -- We are done reading. - break; + // There can be more to read + continue; } - self.buffer = buffer; - if self.system_error.is_some() { - self.buffer = Vec::new(); // clearAndFree - } + // -- We are done reading. + break; + } + self.buffer = buffer; - // If we over-allocated by a lot, we should shrink the buffer to conserve memory. - if self.buffer.len() + 16_000 < self.buffer.capacity() { - self.buffer.shrink_to_fit(); - } - // `Bytes` is owning, and `then()` delivers `self.buffer` directly, - // so do not also stash it in `byte_store` — that would double-free. - self.on_finish(); + if self.system_error.is_some() { + self.buffer = Vec::new(); // clearAndFree } + + // If we over-allocated by a lot, we should shrink the buffer to conserve memory. + if self.buffer.len() + 16_000 < self.buffer.capacity() { + self.buffer.shrink_to_fit(); + } + // `Bytes` is owning, and `then()` delivers `self.buffer` directly, + // so do not also stash it in `byte_store` — that would double-free. + Next::Finish } } @@ -981,7 +1076,7 @@ impl<'a> FileCloser for ReadFileUV<'a> { fn state(&self) -> &AtomicU8 { unreachable!("@hasField(ReadFileUV, \"io_request\") == false") } - fn io_request(&mut self) -> Option<&mut bun_io::Request> { + unsafe fn io_request(_: *mut Self) -> Option<*mut bun_io::Request> { None } fn io_poll(&mut self) -> &mut bun_io::Poll { @@ -990,10 +1085,7 @@ impl<'a> FileCloser for ReadFileUV<'a> { fn task(&mut self) -> &mut bun_jsc::WorkPoolTask { unreachable!("@hasField(ReadFileUV, \"io_request\") == false") } - fn update(&mut self) { - unreachable!("@hasField(ReadFileUV, \"io_request\") == false") - } - fn schedule_close(_: &mut bun_io::Request) -> bun_io::Action<'_> { + unsafe fn schedule_close(_: *mut bun_io::Request) -> bun_io::Action { unreachable!("@hasField(ReadFileUV, \"io_request\") == false") } fn on_close_io_request(_: *mut bun_jsc::WorkPoolTask) { @@ -1118,7 +1210,10 @@ impl<'a> ReadFileUV<'a> { self.total_size = self.total_size.max(self.size); if needs_close { - if self.do_close(self.is_allowed_to_close()) { + let is_allowed_to_close_fd = self.is_allowed_to_close(); + // SAFETY: `self` is live; with no `io_request` (see the + // `FileCloser` impl) this only closes the fd and returns `false`. + if unsafe { Self::do_close(self, is_allowed_to_close_fd) } { // we have to wait for the close to finish return; } diff --git a/src/runtime/webcore/blob/write_file.rs b/src/runtime/webcore/blob/write_file.rs index 293d064020a5..da52f568219f 100644 --- a/src/runtime/webcore/blob/write_file.rs +++ b/src/runtime/webcore/blob/write_file.rs @@ -98,6 +98,25 @@ pub struct WriteFile { bun_threading::intrusive_work_task!(WriteFile, task); bun_io::intrusive_io_request!(WriteFile, io_request); +/// What a pool-side step of a `WriteFile` ends in; see [`WriteFile::proceed`]. +#[cfg(not(windows))] +#[derive(Clone, Copy)] +enum Next { + WriteLoop, + WaitForWritable, + Finish, +} + +/// Outcome of one `write(2)` in [`WriteFile::do_write`]. +#[cfg(not(windows))] +enum WriteStep { + Wrote(usize), + /// `EAGAIN` on a pollable fd; the caller hands the file to the io thread. + WouldBlock, + /// `errno` / `system_error` are set. + Failed, +} + // ────────────────────────────────────────────────────────────────────────── // FileOpener / FileCloser // ────────────────────────────────────────────────────────────────────────── @@ -201,15 +220,20 @@ impl WriteFile { WorkPool::schedule(&raw mut this.task); } + /// io thread: the `io::RequestCallback` installed by `wait_for_writable`. + /// + /// # Safety + /// `io::RequestCallback`'s contract: `request` is the pointer + /// `wait_for_writable` scheduled. #[cfg(not(windows))] - pub(crate) fn on_request_writable(request: &mut io::Request) -> io::Action<'_> { + pub(crate) unsafe fn on_request_writable(request: *mut io::Request) -> io::Action { bun_output::scoped_log!(WriteFile, "WriteFile.onRequestWritable()"); - request.scheduled = false; - // SAFETY: `request` points to WriteFile.io_request (intrusive); recover parent. - let this = unsafe { WriteFile::from_io_request(std::ptr::from_mut(request)) }; - // SAFETY: `request` points to WriteFile.io_request (intrusive), so `this` is the - // live parent; `fd` copy and the `io_poll` field borrow are the only borrows formed. - let (fd, poll) = unsafe { ((*this).opened_fd, &mut (*this).io_poll) }; + // SAFETY: fn contract — `request` is `&raw mut (*this).io_request` of a + // live `WriteFile`, so the parent is reachable through it. + let this = unsafe { WriteFile::from_io_request(request) }; + // SAFETY: as above; the `fd` read ends here and `io_poll` is only + // projected, so no reference into the parent outlives this call. + let (fd, poll) = unsafe { ((*this).opened_fd, &raw mut (*this).io_poll) }; io::Action::Writable(io::FileAction { on_error: Self::on_io_error, ctx: this.cast::<()>(), @@ -219,13 +243,26 @@ impl WriteFile { }) } + /// Hands `*this` to the io thread until `opened_fd` is writable; it comes + /// back through `on_ready` / `on_io_error`. The schedule is the last access + /// (see `update` for why this is pointer-shaped). + /// + /// # Safety + /// `this` is the live `WriteFile` this thread currently holds; the caller + /// must not touch it afterwards. #[cfg(not(windows))] - pub(crate) fn wait_for_writable(&mut self) { - self.close_after_io = true; - self.io_request - .store_callback_seq_cst(Self::on_request_writable); - if !self.io_request.scheduled { - io::IoRequestLoop::schedule(&mut self.io_request); + pub(crate) unsafe fn wait_for_writable(this: *mut Self) { + // SAFETY: fn contract. The `store_callback_seq_cst` reborrow ends with + // its statement; the request is projected (not reborrowed) for + // `schedule`, which needs the whole `WriteFile` reachable through it. + unsafe { + (*this).close_after_io = true; + (*this) + .io_request + .store_callback_seq_cst(Self::on_request_writable); + if !(*this).io_request.scheduled { + io::IoRequestLoop::schedule(&raw mut (*this).io_request); + } } } @@ -286,8 +323,12 @@ impl WriteFile { // reshaped for borrowck — take (off, len) here and re-derive the slice // internally so callers don't hold a borrow of self across the &mut self call. + // + // Reports `WouldBlock` rather than waiting itself: the wait hands the file to + // the io thread, so it has to be the caller's last step, not happen under + // this `&mut self` with the caller about to read `errno`. #[cfg(not(windows))] - pub(crate) fn do_write(&mut self, off: usize, len: usize, wrote: &mut usize) -> bool { + fn do_write(&mut self, off: usize, len: usize) -> WriteStep { let fd = self.opened_fd; debug_assert!(fd != Fd::INVALID); @@ -302,8 +343,8 @@ impl WriteFile { loop { match &result { bun_sys::Result::Ok(res) => { - *wrote = *res; self.total_written += *res; + return WriteStep::Wrote(*res); } bun_sys::Result::Err(err) => { if err.get_errno() == io::RETRY { @@ -312,19 +353,15 @@ impl WriteFile { // this is fine on kqueue, but not on epoll. continue; } - self.wait_for_writable(); - return false; + return WriteStep::WouldBlock; } else { self.errno = Some(bun_errno::from_errno(err.errno as i32).into()); self.system_error = Some(err.to_system_error().into()); - return false; + return WriteStep::Failed; } } } - break; } - - true } pub(crate) fn then(mut this: WriteFile, _global: &JSGlobalObject) -> Result<(), JsTerminated> { @@ -385,26 +422,56 @@ impl WriteFile { .is_path() } + /// Ends with one of two hand-overs: `do_close` gives `*this` to the io + /// thread, otherwise `io_task.finish()` gives it to the JS thread, which + /// frees it. Pointer-shaped for that reason (see `update`). + /// + /// # Safety + /// `this` is the live `WriteFile` this thread currently holds; the caller + /// must not touch it afterwards. #[cfg(not(windows))] - fn on_finish(&mut self) { + unsafe fn on_finish(this: *mut Self) { bun_output::scoped_log!(WriteFile, "WriteFile.onFinish()"); - let close_after_io = self.close_after_io; - if self.do_close(self.is_allowed_to_close()) { + // SAFETY: fn contract; the read and the `is_allowed_to_close` reborrow + // end before `do_close`. + let (close_after_io, is_allowed_to_close_fd) = + unsafe { ((*this).close_after_io, (*this).is_allowed_to_close()) }; + // SAFETY: fn contract, passed through; on `true` the io thread owns + // `*this` and nothing below runs. + if unsafe { Self::do_close(this, is_allowed_to_close_fd) } { return; } if !close_after_io { - if let Some(io_task) = self.io_task.take() { + // SAFETY: fn contract — `do_close` returned `false`, so `*this` is + // still ours; the `take` ends before `finish` posts the job. + if let Some(io_task) = unsafe { (*this).io_task.take() } { io_task.finish(); } } } + /// `FileOpener::get_fd`'s callback, reached from `JobContext::run`. Both + /// of those hand this frame `&mut self`, so unlike the pool re-entries + /// (`update`) a reference to the object is still live up the stack during + /// the hand-over; this frame at least finishes its own use of `self` + /// first and hands over through a pointer as its last act. #[cfg(not(windows))] - fn run_with_fd(&mut self, fd_: Fd) { + fn run_with_fd(&mut self, fd: Fd) { + let next = self.prepare_write(fd); + let this: *mut Self = self; + // SAFETY: `this` is the live `WriteFile` `get_fd` called us with, and + // this frame does not use `self` or `this` after the call. + unsafe { Self::proceed(this, next) } + } + + /// The part of starting a write that happens before anything is handed + /// on: deciding whether the fd can block, preallocating, and the initial + /// writability check. + #[cfg(not(windows))] + fn prepare_write(&mut self, fd_: Fd) -> Next { if fd_ == Fd::INVALID || self.errno.is_some() { - self.on_finish(); - return; + return Next::Finish; } let fd = self.opened_fd; @@ -448,8 +515,7 @@ impl WriteFile { // } if self.could_block && bun_core::is_writable(fd) == bun_core::Pollable::NotReady { - self.wait_for_writable(); - return; + return Next::WaitForWritable; } #[cfg(any(target_os = "linux", target_os = "android"))] @@ -469,38 +535,80 @@ impl WriteFile { } } - self.do_write_loop(); + Next::WriteLoop } fn do_write_loop_task(task: *mut WorkPoolTask) { - // SAFETY: only reached via `WorkPoolTask::callback` with `task` = `&mut self.task` - // (intrusive) registered in `on_writable`/`init`; recover parent. - let this = unsafe { WriteFile::from_task_ptr(task) }; - // On macOS, we use one-shot mode, so we don't need to unregister. - #[cfg(target_os = "macos")] - { - // SAFETY: `this` is the live parent (see above); scoped access. - unsafe { (*this).close_after_io = false }; + // SAFETY: only reached via `WorkPoolTask::callback` with `task` = + // `&raw mut self.task` (intrusive) scheduled by `on_ready` / + // `on_io_error`; recover the parent, which is this thread's until + // `update` hands it on. + unsafe { + let this = WriteFile::from_task_ptr(task); + // On macOS, we use one-shot mode, so we don't need to unregister. + #[cfg(target_os = "macos")] + { + (*this).close_after_io = false; + } + Self::update(this); } - // SAFETY: `this` is the live parent (see above); exclusive borrow scoped to the call. - unsafe { (*this).do_write_loop() }; - } - - pub(crate) fn update(&mut self) { - self.do_write_loop(); } - fn do_write_loop(&mut self) { + /// Pool thread re-entry (`do_write_loop_task`, `on_close_io_request`). + /// Pointer-shaped, like everything it calls, for the reason given on + /// `ReadFile::update`: every path below ends by handing `*this` to another + /// thread. + /// + /// # Safety + /// `this` is the live `WriteFile` this thread currently holds; the caller + /// must not touch it afterwards. + pub(crate) unsafe fn update(this: *mut Self) { #[cfg(windows)] { - return; // why + let _ = this; + // Windows writes go through WriteFileWindows; this is never scheduled. } #[cfg(not(windows))] - self.do_write_loop_posix(); + { + // SAFETY: fn contract, passed through. + unsafe { Self::do_write_loop(this) } + } } + /// Performs the step a pool-side function decided on once its own + /// accesses to `*this` are over. `WriteLoop` writes until it has to hand + /// the object on; the other two are the hand-overs themselves. + /// + /// # Safety + /// As [`update`](Self::update). #[cfg(not(windows))] - fn do_write_loop_posix(&mut self) { + unsafe fn proceed(this: *mut Self, next: Next) { + // SAFETY: fn contract, passed through. + unsafe { + match next { + Next::WriteLoop => Self::do_write_loop(this), + Next::WaitForWritable => Self::wait_for_writable(this), + Next::Finish => Self::on_finish(this), + } + } + } + + /// # Safety + /// As [`update`](Self::update). + #[cfg(not(windows))] + unsafe fn do_write_loop(this: *mut Self) { + // SAFETY: fn contract; the reborrow ends with the call, before + // `proceed` hands the object on through `this`. + let next = unsafe { (*this).write_until_blocked() }; + // SAFETY: fn contract, passed through. + unsafe { Self::proceed(this, next) } + } + + /// Writes until the write is complete (`Finish`: everything written, a + /// zero-length write, closing, or an error recorded in `errno`) or the fd + /// would block (`WaitForWritable`). Never produces `WriteLoop`. + #[cfg(not(windows))] + fn write_until_blocked(&mut self) -> Next { while self.state.load(Ordering::Relaxed) == ClosingState::Running as u8 { let remain_full = self.bytes_blob.shared_view(); // reshaped for borrowck — capture len/offset before mut borrow @@ -508,31 +616,22 @@ impl WriteFile { let remain_len = remain_full.len() - off; if remain_len > 0 && self.errno.is_none() { - let mut wrote: usize = 0; - let continue_writing = self.do_write(off, remain_len, &mut wrote); - if !continue_writing { - // Stop writing, we errored - if self.errno.is_some() { - self.on_finish(); - return; - } - - // Stop writing, we need to wait for it to become writable. - return; - } + let wrote = match self.do_write(off, remain_len) { + WriteStep::Wrote(wrote) => wrote, + WriteStep::WouldBlock => return Next::WaitForWritable, + WriteStep::Failed => return Next::Finish, + }; // Do not immediately attempt to write again if it's not a regular file. if self.could_block && bun_core::is_writable(self.opened_fd) == bun_core::Pollable::NotReady { - self.wait_for_writable(); - return; + return Next::WaitForWritable; } if wrote == 0 { // we are done, we received EOF - self.on_finish(); - return; + return Next::Finish; } continue; @@ -541,7 +640,7 @@ impl WriteFile { break; } - self.on_finish(); + Next::Finish } } diff --git a/test/internal/source-lints/self-receiver-io-schedule.test.ts b/test/internal/source-lints/self-receiver-io-schedule.test.ts new file mode 100644 index 000000000000..8a7deba2113a --- /dev/null +++ b/test/internal/source-lints/self-receiver-io-schedule.test.ts @@ -0,0 +1,185 @@ +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"; + +// `bun_io::IoRequestLoop::schedule` is the hand-over of a ReadFile/WriteFile +// to the io thread. Two ways of spelling its argument are banned: +// +// IoRequestLoop::schedule(&mut self.io_request) // from `self` +// IoRequestLoop::schedule(&raw mut self.io_request) +// let r = self.io_request(); ... IoRequestLoop::schedule(r) +// IoRequestLoop::schedule(&mut (*this).io_request) // a `&mut` reborrow +// +// The argument has to be the request *projected from the owner pointer the +// caller holds* (`&raw mut (*this).io_request`, or a `*mut` returned by an +// accessor that does that), for two reasons: +// +// - The io thread hands that exact pointer back to the request's callback +// (`on_request_readable` / `on_request_writable` / `schedule_close`), +// which recovers the owner from it by container-of. A `&mut` reborrow of +// the field (which is what a `&mut Request` argument or `&mut x.io_request` +// coerced to `*mut` is) only has provenance over the field's bytes, so the +// container-of walk leaves its bounds; `bun_core::container_of`'s +// contract says as much ("a `&mut field` reborrow does not suffice"). +// - The push is the moment the object changes threads: the io thread (and, +// through on_ready / on_error / on_done, a pool thread) may be writing to +// the request and its owner before `schedule` returns. A `&mut self` +// method publishing its own receiver therefore has a reference argument +// that stays protected, on this thread's stack, while another thread +// writes through the object, which both aliasing models (Tree Borrows is +// what `bun run rust:miri` uses) reject regardless of whether `self` is +// touched again. `schedule` takes `*mut Request` so that it does not do +// this itself; a caller spelling the argument from `self` reintroduces it +// one frame up. +// +// Before `schedule` took a pointer, all three callers had the shape: +// `ReadFile::wait_for_readable` and `WriteFile::wait_for_writable` +// (`schedule(&mut self.io_request)`) and `FileCloser::do_close` +// (`schedule(io_request)` with `io_request` bound from `self.io_request()`). +// The converted versions (`unsafe fn wait_for_readable(this: *mut Self)`, +// `unsafe fn do_close(this: *mut Self, ..)` in src/runtime/webcore/Blob.rs) +// are the templates; `do_write_loop_task` / `on_close_io_request` show how the +// pointer reaches them without a `&mut` being formed on the way. +// +// Scope: calls to `..IoRequestLoop::schedule(` whose argument is spelled from +// `self` (directly, or via a local bound from a `self.` expression earlier in +// the same function), or is any `&mut` expression. An argument derived from a +// reference other than `self` (`fn f(r: &mut ReadFile)` ... `&raw mut +// r.io_request`) is the same bug but outside this lint; convert it on sight. +// +// Siblings: self-receiver-reclaim.test.ts (freeing the receiver), +// fn-long-mut-reborrow.test.ts, frozen-nonnull-reborrow.test.ts. + +const root = path.resolve(import.meta.dir, "..", "..", ".."); +const rustSources = globAllSources().rust.filter(p => p.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). Same guard as +// dead-code-escapes.test.ts. +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)); +})(); + +// `IoRequestLoop::schedule(`, however the path in front is spelled, with the +// argument list open. `\s*` after the paren so a rustfmt-wrapped argument +// still matches. +const SCHEDULE = String.raw`\bIoRequestLoop::schedule\(\s*`; + +// Any `&mut ...` argument: a reborrow of the field, coerced to `*mut` at the +// call. (`&raw mut` has `raw` in between and is not matched here; it is only +// banned when its base is `self`, below.) +const MUT_REBORROW = new RegExp(SCHEDULE + String.raw`&\s*mut\b`, "g"); + +// An argument spelled from `self`: `self.` as the base of the argument, +// optionally behind `&mut` / `&raw mut` / one wrapping call such as +// `ptr::from_mut(`. +const FROM_SELF = new RegExp(SCHEDULE + String.raw`(?:[\w:]+\(\s*)?(?:&\s*(?:raw\s+)?(?:mut|const)\s+)?self\s*\.`, "g"); + +// A local bound from a `self.` expression: `let r = &mut self.io_request;`, +// `let r = self.io_request();`, `if let Some(r) = self.io_request() {`. The +// binding is then looked for as the schedule argument further down the same +// function, which ends at the next `fn` item. +const SELF_BINDING = new RegExp( + String.raw`let\s+(?:Some\(\s*)?(?:mut\s+)?(\w+)\s*\)?\s*(?::[^=;{]*)?=\s*(?:&\s*(?:raw\s+)?(?:mut|const)\s+)?self\s*\.`, + "g", +); +const FN_ITEM = /^[ \t]*(?:pub(?:\([^)]*\))?\s+)?(?:(?:const|async|unsafe|extern\s+"[^"]*")\s+)*fn\s/m; + +function scheduleOf(name: string): RegExp { + return new RegExp(SCHEDULE + name + String.raw`\s*,?\s*\)`); +} + +/** Byte offsets (into `stripped`) of every banned schedule call in one file. */ +function findBanned(stripped: string): number[] { + const hits = new Set(); + for (const m of stripped.matchAll(MUT_REBORROW)) hits.add(m.index); + for (const m of stripped.matchAll(FROM_SELF)) hits.add(m.index); + for (const binding of stripped.matchAll(SELF_BINDING)) { + const start = binding.index + binding[0].length; + const rest = stripped.slice(start); + const fnEnd = rest.search(FN_ITEM); + const body = fnEnd === -1 ? rest : rest.slice(0, fnEnd); + const call = body.search(scheduleOf(binding[1])); + if (call !== -1) hits.add(start + call); + } + return [...hits].sort((a, b) => a - b); +} + +function lineOf(text: string, offset: number): number { + return text.slice(0, offset).split("\n").length; +} + +const offenders: string[] = []; +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(); + // Strip full-line comments so prose mentions (including the in-tree comments + // describing this hazard) don't count. `[ \t]*`, not `\s*`: `\s` crosses + // newlines and would swallow blank lines, shifting the reported line numbers. + const stripped = content.replace(/^[ \t]*\/\/.*$/gm, ""); + for (const offset of findBanned(stripped)) { + offenders.push(`${source}:${lineOf(stripped, offset)}`); + } +} + +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 patterns match the banned spellings and nothing else", () => { + const banned = [ + // `wait_for_readable` / `wait_for_writable` as they were. + "io::IoRequestLoop::schedule(&mut self.io_request);", + // `FileCloser::do_close` as it was. + "if let Some(io_request) = self.io_request() {\n io_request.store_callback_seq_cst(Self::schedule_close);\n if !io_request.scheduled {\n bun_io::IoRequestLoop::schedule(io_request);\n }\n}", + // The pointer-typed argument spelled from `self` anyway. + "io::IoRequestLoop::schedule(&raw mut self.io_request);", + "bun_io::IoRequestLoop::schedule(self.io_request_ptr());", + "IoRequestLoop::schedule(core::ptr::from_mut(&mut self.io_request));", + "let request = &raw mut self.io_request;\nrequest.scheduled;\nio::IoRequestLoop::schedule(request);", + "let mut request = self.io_request();\nio::IoRequestLoop::schedule(request);", + // A `&mut` reborrow, whatever it is based on: its provenance stops at the field. + "io::IoRequestLoop::schedule(&mut (*this).io_request);", + "bun_io::IoRequestLoop::schedule(&mut *io_request);", + // rustfmt-wrapped. + "::bun_io::IoRequestLoop::schedule(\n &mut self.io_request,\n);", + "::bun_io::IoRequestLoop::schedule(\n &raw mut self.io_request,\n);", + ]; + const allowed = [ + // Projected from the pointer the caller holds: the intended shape. + "io::IoRequestLoop::schedule(&raw mut (*this).io_request);", + "if let Some(io_request) = Self::io_request(this) {\n (*io_request).store_callback_seq_cst(Self::schedule_close);\n if !(*io_request).scheduled {\n bun_io::IoRequestLoop::schedule(io_request);\n }\n}", + "let request = unsafe { Self::io_request_of(this) };\nbun_io::IoRequestLoop::schedule(request);", + // A self-derived binding that is not what gets scheduled. + "let fd = self.opened_fd;\nio::IoRequestLoop::schedule(request);", + // The binding is scheduled, but in the next function, where it is a + // raw-pointer parameter of the same name. + "let request = &mut self.io_request;\nrequest.scheduled = false;\n}\n\nunsafe fn publish(request: *mut io::Request) {\n io::IoRequestLoop::schedule(request);\n}", + // Other things called `schedule`. + "WorkPool::schedule(&raw mut self.task);", + "bun_jsc::Job::::schedule(&global.js_thread(), this, completion);", + "self.schedule(&mut self.io_request);", + ]; + expect(banned.map(s => findBanned(s).length)).toEqual(banned.map(() => 1)); + expect(allowed.map(s => findBanned(s).length)).toEqual(allowed.map(() => 0)); +}); + +test("nothing schedules an io request spelled from self or through a &mut reborrow", () => { + expect(offenders).toEqual([]); +}); diff --git a/test/js/bun/io/bun-write.test.js b/test/js/bun/io/bun-write.test.js index f02bdf4f8e17..9cac7bde4d7a 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,81 @@ const IS_UV_FS_COPYFILE_DISABLED = expect(exitCode).toBe(0); }); + // A non-blocking FIFO that is already full when Bun.write() starts: the + // WriteFile is handed to the io thread before its first write(), again each + // time the pipe fills up while the payload drains, and (on Linux) once more + // to unregister the fd when it is done. The payload is at least 256 KiB so + // Bun.write() takes the WriteFile path rather than the synchronous one. + it.skipIf(isWindows)("Bun.write(Bun.file(fd)) to a full non-blocking FIFO waits for the reader", async () => { + const size = 256 * 1024; + using dir = tempDir("bun-write-fifo", {}); + const fifo = join(String(dir), "out.fifo"); + mkfifo(fifo); + // A reader has to exist for the child's O_NONBLOCK open of the write end to + // succeed; this one never reads, the readFile below does. + const holder = fs.openSync(fifo, fs.constants.O_RDONLY | fs.constants.O_NONBLOCK); + try { + const script = ` + const fs = require("fs"); + const fd = fs.openSync(process.env.FIFO, fs.constants.O_WRONLY | fs.constants.O_NONBLOCK); + const dest = Bun.file(fd); + // fstat()s the fd: Bun.write() only polls a destination it knows is not a regular file. + dest.size; + const chunk = Buffer.alloc(4096, "P"); + let filled = 0; + try { + for (;;) filled += fs.writeSync(fd, chunk); + } catch (e) { + if (e.code !== "EAGAIN") throw e; + } + const written = Bun.write(dest, Buffer.alloc(${size}, "W")); + process.stderr.write("filled " + filled + "\\n"); + process.stdout.write(String(await written)); + `; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", script], + env: { ...bunEnv, FIFO: fifo }, + stdout: "pipe", + stderr: "pipe", + }); + const stdout = proc.stdout.text(); + const filled = Promise.withResolvers(); + const stderr = (async () => { + let text = ""; + for await (const chunk of proc.stderr) { + text += Buffer.from(chunk).toString(); + const m = /^filled (\d+)\n/m.exec(text); + if (m) filled.resolve(Number(m[1])); + } + filled.reject(new Error(`child exited before filling the FIFO: ${text}`)); + return text; + })(); + // Only start draining once the child has filled the pipe and started the write. + const prefill = await filled.promise; + const data = await fs.promises.readFile(fifo); + const [resolved, stderrText, exitCode] = await Promise.all([stdout, stderr, proc.exited]); + + expect({ + prefilled: prefill > 0, + resolved, + stderr: stderrText, + length: data.length, + prefillIntact: data.subarray(0, prefill).equals(Buffer.alloc(prefill, "P")), + payloadIntact: data.subarray(prefill).equals(Buffer.alloc(size, "W")), + }).toEqual({ + prefilled: true, + resolved: String(size), + stderr: `filled ${prefill}\n`, + length: prefill + size, + prefillIntact: true, + payloadIntact: true, + }); + expect(exitCode).toBe(0); + } finally { + fs.closeSync(holder); + } + }); + it("Bun.file(0) survives GC", async () => { for (let i = 0; i < 10; i++) { let f = Bun.file(0); diff --git a/test/js/bun/util/bun-file-read.test.ts b/test/js/bun/util/bun-file-read.test.ts index 40c6309a0c13..346f1004dbd0 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 } from "node:fs"; import { tmpdir } from "node:os"; import path from "node:path"; @@ -12,7 +15,7 @@ it("offset should work in Bun.file() #4963", async () => { expect(contents).toBe("ntents"); }); -// do_read_loop picks its read target per iteration: the 64 KB stack buffer +// ReadFile::read_until_blocked picks its read target per iteration: the 64 KB stack buffer // when self.buffer's spare capacity is smaller, otherwise the Vec's spare // capacity directly. Cover both branches plus the max_length cap so the // branch selection and the commit_spare path stay tied to the same decision. @@ -54,3 +57,49 @@ describe("Bun.file read-loop target selection", () => { expect(Bun.hash(buf)).toBe(Bun.hash(bytes.subarray(start, end))); }); }); + +// Reading a FIFO by path. The payload is several pipe buffers long, so the +// ReadFile is handed to the io thread every time it drains the pipe and has to +// wait for more, and once the writer closes it is (on Linux) handed over once +// more to unregister the fd before the fd it opened is closed. +describe.skipIf(isWindows)("Bun.file(fifo)", () => { + it("bytes() reads a pipe that is filled 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); + // Opening the write end needs a reader to exist. The holder stays open so + // the child finds a connected writer (rather than EOF) whenever it opens + // the FIFO; it never reads, so every byte goes to the child. + const holder = openSync(fifo, constants.O_RDONLY | constants.O_NONBLOCK); + let writer = openSync(fifo, "w"); + try { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `const bytes = await Bun.file(process.env.FIFO).bytes(); process.stdout.write(bytes.length + " " + Bun.hash(bytes));`, + ], + env: { ...bunEnv, FIFO: fifo }, + stdout: "pipe", + stderr: "pipe", + }); + // The write end is blocking, so this only completes as the child drains + // the pipe; closing it afterwards is what ends the child's read. + const written = await Bun.write(Bun.file(writer), payload); + closeSync(writer); + writer = -1; + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + + expect({ written, stdout, stderr }).toEqual({ + written: payload.length, + stdout: `${payload.length} ${Bun.hash(payload)}`, + stderr: "", + }); + expect(exitCode).toBe(0); + } finally { + if (writer !== -1) closeSync(writer); + closeSync(holder); + } + }); +}); From 58ed1d3fa870410951b0a638a4575a3df89a7eb6 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Wed, 12 Aug 2026 08:20:55 +0000 Subject: [PATCH 2/4] Shorten the new doc comments; fail the FIFO read test fast if the child dies The rationale for the pointer-shaped hand-over now lives on IoRequestLoop::schedule, Action and ReadFile::update; everything else points at those. In the FIFO read test the parent's own reader fd kept the blocked write from ever seeing EPIPE if the child failed, so the test would hang instead of failing. The write is now raced against the child's exit, which drops that fd and reports the child's stderr. --- src/io/lib.rs | 106 +++++++++---------------- src/runtime/webcore/Blob.rs | 48 ++++------- src/runtime/webcore/blob/read_file.rs | 96 ++++++++-------------- src/runtime/webcore/blob/write_file.rs | 91 ++++++++------------- test/js/bun/util/bun-file-read.test.ts | 32 +++++--- 5 files changed, 139 insertions(+), 234 deletions(-) diff --git a/src/io/lib.rs b/src/io/lib.rs index fc6aea655789..5e1c65765204 100644 --- a/src/io/lib.rs +++ b/src/io/lib.rs @@ -915,26 +915,19 @@ impl IoRequestLoop { /// async-signal-safe `waker`. This is the *only* cross-thread entry /// point — every other `IoRequestLoop` method is IO-thread-only. /// - /// The push is the hand-over: from that instant the IO thread may pop the - /// request, clear `scheduled`, run its callback and (through `on_error` / - /// `on_done`) pass the owner on to a pool thread, all before this returns - /// from `wake()`. That is why this takes the pointer rather than `&mut - /// Request` (a reference argument would stay protected across the push), - /// and why the pointer has to be projected from the owner: the IO thread - /// hands this exact pointer to [`Request::callback`], which recovers the - /// owner from it by container-of, so its provenance must cover the owner, - /// which a `&mut self.io_request` reborrow's does not. + /// The push is the hand-over (the IO thread may be running the callback + /// before `wake()` returns), and the callback recovers the owner from this + /// very pointer by container-of, hence `*mut` projected from the owner + /// rather than `&mut`. /// /// # Safety - /// `request` is `&raw mut (*owner).io_request` of a live owner that the - /// calling thread currently holds, with `scheduled == false`. Once this is - /// called the owner belongs to the IO thread: the caller must not touch the - /// request or its owner again (not even through a reference it already - /// holds) until the IO thread hands it back. + /// `request` is `&raw mut (*owner).io_request` of a live, unscheduled owner + /// the caller holds; the caller touches neither again until the IO thread + /// hands the owner back. pub unsafe fn schedule(request: *mut Request) { Self::ensure_init(); let request = core::ptr::NonNull::new(request).expect("io request of a live owner"); - // SAFETY: fn contract — live and ours until the push below. + // SAFETY: fn contract — ours until the push below. unsafe { debug_assert!(!(*request.as_ptr()).scheduled); (*request.as_ptr()).scheduled = true; @@ -954,20 +947,13 @@ impl IoRequestLoop { } } - /// IO thread: a request just popped from `pending` is this thread's now. - /// Clears its `scheduled` bit and asks the owner what to do with it. - /// - /// The callback gets `request` as popped, i.e. the pointer `schedule` was - /// given, so the owner it recovers by container-of is in bounds of it. The - /// returned action's `poll` points into that owner, which stays this - /// thread's only until the action's `on_error` / `on_done` hands it on: - /// every use of `poll` has to come before that call. + /// IO thread: runs the callback of a request just popped from `pending`, + /// passing the pointer through as `schedule` was given it (see [`Action`] + /// for how long the result may be used). #[cfg(not(windows))] fn take_request(request: *mut Request) -> Action { - // SAFETY: `pending` only holds pointers that went through `schedule`, - // whose contract makes each the request field of a live owner, with - // the owner's provenance, that nobody else touches until the IO thread - // hands it on; popping it made it this thread's. + // SAFETY: `pending` only holds pointers that went through `schedule` + // (live owner, owner provenance); popping one made it this thread's. unsafe { (*request).scheduled = false; ((*request).callback)(request) @@ -1002,8 +988,7 @@ impl IoRequestLoop { /// The `Readable` / `Writable` arm of [`tick_epoll`](Self::tick_epoll). #[cfg(any(target_os = "linux", target_os = "android"))] fn register_epoll(watcher_fd: Fd, file: &FileAction, flag: Flags) { - // SAFETY: `take_request` — the owner is ours until `on_error` hands it - // on, and this is the last use of `poll` before that. + // SAFETY: see `Action` — `poll` is live and ours until `on_error` below. let registered = unsafe { Poll::register_for_epoll(file.poll, flag, file.tag, watcher_fd, true, file.fd) }; @@ -1033,9 +1018,8 @@ impl IoRequestLoop { Self::register_epoll(watcher_fd, &file, Flags::PollWritable) } Action::Close(close) => { - // SAFETY: `take_request` — the owner is ours until - // `on_done` hands it on; the reborrow ends with this - // block. + // SAFETY: see `Action` — `poll` is live and ours until + // `on_done` below; the reborrow ends with the block. unsafe { let poll = &mut *close.poll; log!( @@ -1148,9 +1132,8 @@ impl IoRequestLoop { } match Self::take_request(request) { Action::Readable(readable) => { - // SAFETY: `take_request` — the owner stays ours - // until the kernel reports the fd (no hand-on in - // this arm). + // SAFETY: see `Action` — `poll` is live and ours + // (nothing in this arm hands the owner on). unsafe { Poll::apply_kqueue( ApplyAction::Readable, @@ -1174,9 +1157,8 @@ impl IoRequestLoop { } } Action::Close(close) => { - // SAFETY: `take_request` — the owner is ours until - // `on_done` hands it on, and this block is the last - // use of `poll` before that. + // SAFETY: see `Action` — `poll` is live and ours until + // `on_done` below. unsafe { let flags = (*close.poll).flags; if flags.contains(Flags::PollReadable) @@ -1236,15 +1218,12 @@ impl IoRequestLoop { // ─── Request ────────────────────────────────────────────────────────────────── -/// What the IO thread does with a popped [`Request`]. Receives the pointer -/// [`IoRequestLoop::schedule`] was given (the owner's `io_request` field, with -/// the owner's provenance), so the trampoline may recover the owner from it -/// with [`IntrusiveIoRequest::from_io_request`]. `scheduled` has already been -/// cleared when this runs. +/// Run by the IO thread with the pointer [`IoRequestLoop::schedule`] was given +/// (`scheduled` already cleared), so it may recover the owner via +/// [`IntrusiveIoRequest::from_io_request`]. /// /// # Safety -/// Only the IO thread may call it, with a pointer that went through -/// `schedule` and has not been handed back since. +/// IO thread only, with a pointer `schedule` was given and has not handed back. pub type RequestCallback = unsafe fn(*mut Request) -> Action; pub struct Request { @@ -1307,9 +1286,8 @@ pub unsafe trait IntrusiveIoRequest: Sized { /// # Safety /// `req` must point to the [`Request`] field at `Self::IO_REQUEST_OFFSET` /// inside a live `Self` allocation that was scheduled via that field, and - /// the pointer's provenance must cover the whole allocation. The pointer a - /// [`RequestCallback`] receives satisfies this: it is the one - /// [`IoRequestLoop::schedule`] requires to be projected from the owner. + /// the pointer's provenance must cover the whole allocation (true of the + /// pointer a [`RequestCallback`] receives, per [`IoRequestLoop::schedule`]). #[inline(always)] unsafe fn from_io_request(req: *mut Request) -> *mut Self { // SAFETY: caller upholds the trait safety contract above. @@ -1398,13 +1376,10 @@ pub(crate) type RequestQueue = bun_threading::UnboundedQueue; // ─── Action ─────────────────────────────────────────────────────────────────── -/// Returned by a [`RequestCallback`]. `poll` and `ctx` point into the -/// request's owner, which is the IO thread's until the action's `on_error` / -/// `on_done` hands it on; `poll` is a pointer (projected from the owner, like -/// the request itself) rather than a `&mut` both so the IO thread holds no -/// reference into the owner across that hand-on and because the address -/// registered with the kernel is derived from it and later turned back into -/// the owner by container-of (`__bun_io_pollable_on_ready`). +/// Returned by a [`RequestCallback`]. `poll` / `ctx` point into the owner, +/// which is the IO thread's only until `on_error` / `on_done` hands it on. +/// `poll` is projected from the owner like the request: the kernel hands its +/// address back and `__bun_io_pollable_on_ready` container-ofs the owner from it. pub enum Action { Readable(FileAction), Writable(FileAction), @@ -1570,12 +1545,8 @@ enum ApplyAction { } impl Poll { - /// `poll` is an [`Action`]'s `poll` (see there): the address the kernel - /// hands back as `udata` is taken from it as given, so that the owner - /// recovered from it in `on_update_kqueue`'s dispatch is in bounds of it. - /// /// # Safety - /// `poll` is the `io_poll` of a live owner the IO thread currently holds. + /// `poll` is an [`Action`]'s `poll`, live and the IO thread's. #[cfg(any(target_os = "macos", target_os = "freebsd"))] #[inline] pub(crate) unsafe fn apply_kqueue( @@ -1596,9 +1567,9 @@ impl Poll { ); let one_shot_flag = libc::EV_ONESHOT; + // `udata` comes from the pointer itself (see `Action`), not the reborrow. let udata: usize = Pollable::init(tag, poll).ptr() as usize; - // SAFETY: fn contract; the reborrow lasts for this call only, and the - // `udata` the kernel keeps was taken from the pointer above it. + // SAFETY: fn contract; the reborrow lasts for this call only. let poll = unsafe { &mut *poll }; let (filter, flags_): (i16, u16) = match action { ApplyAction::Readable => (libc::EVFILT_READ, libc::EV_ADD | one_shot_flag), @@ -1752,13 +1723,8 @@ impl Poll { } } - /// `poll` is an [`Action`]'s `poll` (see there): the address the kernel - /// hands back in `epoll_event.u64` is taken from it as given, so that the - /// owner recovered from it in `on_update_epoll`'s dispatch is in bounds of - /// it. - /// /// # Safety - /// `poll` is the `io_poll` of a live owner the IO thread currently holds. + /// `poll` is an [`Action`]'s `poll`, live and the IO thread's. #[cfg(any(target_os = "linux", target_os = "android"))] // `enumset::EnumSetType` cannot be a const generic, so `flag` is a runtime // arg. The `match` below preserves the exhaustiveness check. @@ -1774,9 +1740,9 @@ impl Poll { debug_assert!(fd != Fd::INVALID); + // `udata` comes from the pointer itself (see `Action`), not the reborrow. let udata = Pollable::init(tag, poll).ptr(); - // SAFETY: fn contract; the reborrow lasts for this call only, and the - // `udata` the kernel keeps was taken from the pointer above it. + // SAFETY: fn contract; the reborrow lasts for this call only. let this = unsafe { &mut *poll }; if one_shot { diff --git a/src/runtime/webcore/Blob.rs b/src/runtime/webcore/Blob.rs index 01d495bcb77a..4589e0e5543a 100644 --- a/src/runtime/webcore/Blob.rs +++ b/src/runtime/webcore/Blob.rs @@ -7037,12 +7037,8 @@ pub trait FileCloser: Sized { fn set_opened_fd(&mut self, fd: Fd); fn close_after_io(&self) -> bool; fn state(&self) -> &core::sync::atomic::AtomicU8; - /// The embedded request [`do_close`](Self::do_close) hands to the io - /// thread, projected from `this` (`&raw mut (*this).io_request`, never a - /// `&mut` reborrow: `bun_io::IoRequestLoop::schedule` wants a pointer the - /// whole `Self` is reachable through, because `schedule_close` gets it - /// back and recovers `Self` from it). `None` when the type never goes - /// through the io thread (`ReadFileUV`). + /// `&raw mut (*this).io_request`, as `bun_io::IoRequestLoop::schedule` + /// requires; `None` if the type never uses the io thread (`ReadFileUV`). /// /// # Safety /// `this` points at a live `Self`. @@ -7052,14 +7048,12 @@ pub trait FileCloser: Sized { #[cfg(windows)] fn loop_(&self) -> *mut bun_libuv_sys::uv_loop_t; - /// The io thread's side of `do_close`: a `bun_io::RequestCallback`. - /// Intrusive backref: Rust `offset_of!` cannot name fields on a trait - /// `Self`, so each concrete impl supplies its own container_of recovery - /// (no default body). + /// The `bun_io::RequestCallback` `do_close` installs. Intrusive backref: + /// Rust `offset_of!` cannot name fields on a trait `Self`, so each concrete + /// impl supplies its own container_of recovery (no default body). /// /// # Safety - /// `bun_io::RequestCallback`'s contract: io thread, `request` is the - /// pointer `do_close` scheduled. + /// `bun_io::RequestCallback`'s contract. unsafe fn schedule_close(request: *mut bun_io::Request) -> bun_io::Action; fn on_io_request_closed(this: &mut Self) { @@ -7081,18 +7075,14 @@ pub trait FileCloser: Sized { /// `container_of` deref locally, so a fn-level qualifier is redundant. fn on_close_io_request(task: *mut bun_jsc::WorkPoolTask); - /// Returns `true` when the close was handed to the io thread instead: - /// `*this` belongs to it from that moment (it comes back through - /// `on_close_io_request`), which is why this takes the pointer the caller - /// holds rather than `&mut self`, and why the hand-over is the last thing - /// here that touches `*this`. + /// Returns `true` when `*this` was handed to the io thread instead (it comes + /// back through `on_close_io_request`); that hand-over is the last access. /// /// # Safety - /// `this` is a live `Self` that the calling thread currently holds. After - /// a `true` return the caller must not touch `*this` again. + /// `this` is a live `Self` the caller holds; after a `true` return the + /// caller must not touch it again. unsafe fn do_close(this: *mut Self, is_allowed_to_close_fd: bool) -> bool { - // SAFETY: fn contract. Each access below is a place expression or a - // reborrow that ends with its statement; nothing follows the schedule. + // SAFETY: fn contract; every reborrow below ends with its statement. unsafe { if (*this).close_after_io() { (*this).state().store( @@ -7131,10 +7121,8 @@ 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 `unsafe fn update(this: *mut Self)` (the pool re-entry after -/// the io thread closed the request), and a [`bun_io::Tag`] variant named -/// after the type. -/// The type must also carry `bun_threading::intrusive_work_task!` and +/// an inherent `unsafe fn update(this: *mut Self)`, 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. macro_rules! impl_file_closer { @@ -7170,17 +7158,14 @@ macro_rules! impl_file_closer { unsafe fn schedule_close(request: *mut ::bun_io::Request) -> ::bun_io::Action { use ::bun_io::IntrusiveIoRequest as _; - // SAFETY: fn contract — `request` is the `&raw mut (*this).io_request` - // that `do_close` scheduled, so the parent is live and reachable - // through it. + // SAFETY: fn contract — `request` is the projection `do_close` scheduled. let this = unsafe { $T::from_io_request(request) }; fn on_done(ctx: *mut ()) { // SAFETY: ctx is `self as *mut Self` set below. let this = unsafe { ::bun_ptr::callback_ctx::<$T>(ctx.cast()) }; <$T as crate::webcore::blob::FileCloser>::on_io_request_closed(this); } - // SAFETY: as above; the `fd` read ends here and `io_poll` is only - // projected, so no reference into the parent outlives this call. + // SAFETY: as above; a field read and a projection, no reference escapes. let (fd, poll) = unsafe { ((*this).opened_fd, &raw mut (*this).io_poll) }; ::bun_io::Action::Close(::bun_io::CloseAction { fd, @@ -7202,8 +7187,7 @@ macro_rules! impl_file_closer { // `&mut self.task` (intrusive) registered in `on_io_request_closed`; // recover parent. let this = unsafe { $T::from_task_ptr(task) }; - // SAFETY: `this` is the live parent (see above) and this thread's - // until `update` hands it on, which is why it gets the pointer. + // SAFETY: as above; `update` takes the pointer because it hands it on. unsafe { (*this).close_after_io = false; $T::update(this); diff --git a/src/runtime/webcore/blob/read_file.rs b/src/runtime/webcore/blob/read_file.rs index 936da379a8ae..49bf51ddc3ad 100644 --- a/src/runtime/webcore/blob/read_file.rs +++ b/src/runtime/webcore/blob/read_file.rs @@ -339,27 +339,21 @@ impl FileOpener for ReadFile { crate::webcore::blob::impl_file_closer!(ReadFile); impl ReadFile { - /// Pool thread re-entry (`do_read_loop_task`, `on_close_io_request`). - /// - /// This and everything it calls take the pointer the pool handed us - /// instead of `&mut self`: each of these paths ends by handing `*this` to - /// another thread (`wait_for_readable` / `do_close` to the io thread, - /// `io_task.finish()` to the JS thread), and a `&mut self` argument would - /// still be live, and protected, on this stack while that thread writes to - /// or frees the object. Accesses are scoped to end before the hand-over. + /// Pool thread re-entry (`do_read_loop_task`, `on_close_io_request`). This + /// and everything below it end by handing `*this` to another thread (the io + /// thread, or the JS thread that frees it), so they carry the pointer the + /// pool gave us rather than a `&mut self` that would outlive the hand-over. /// /// # Safety - /// `this` is the live `ReadFile` this thread currently holds; the caller - /// must not touch it afterwards. + /// `this` is the live `ReadFile` this thread holds; not to be touched after. pub(crate) unsafe fn update(this: *mut Self) { #[cfg(windows)] { - let _ = this; - // Windows reads go through ReadFileUV; this is never scheduled. + let _ = this; // Reads go through ReadFileUV; never scheduled. } #[cfg(not(windows))] { - // SAFETY: fn contract; the load ends before `this` is passed on. + // SAFETY: fn contract. let closing = unsafe { (*this).state.load(Ordering::Relaxed) } == ClosingState::Closing as u8; // SAFETY: fn contract, passed through. @@ -376,9 +370,7 @@ impl ReadFile { } } - /// Performs the step a pool-side function decided on once its own - /// accesses to `*this` are over. `ReadLoop` reads until it has to hand the - /// object on; the other two are the hand-overs themselves. + /// Performs the step a `&mut self` stage decided on, after that borrow ended. /// /// # Safety /// As [`update`](Self::update). @@ -482,19 +474,16 @@ impl ReadFile { unsafe { (*ctx.cast::()).on_io_error(err) } } - /// io thread: the `io::RequestCallback` installed by `wait_for_readable`. + /// The `io::RequestCallback` installed by `wait_for_readable`. /// /// # Safety - /// `io::RequestCallback`'s contract: `request` is the pointer - /// `wait_for_readable` scheduled. + /// `io::RequestCallback`'s contract. #[cfg(not(windows))] pub(crate) unsafe fn on_request_readable(request: *mut io::Request) -> io::Action { bloblog!("ReadFile.onRequestReadable"); - // SAFETY: fn contract — `request` is `&raw mut (*this).io_request` of a - // live `ReadFile`, so the parent is reachable through it. + // SAFETY: fn contract — `request` is the projection `wait_for_readable` scheduled. let this = unsafe { ReadFile::from_io_request(request) }; - // SAFETY: as above; the `fd` read ends here and `io_poll` is only - // projected, so no reference into the parent outlives this call. + // SAFETY: as above; a field read and a projection, no reference escapes. let (fd, poll) = unsafe { ((*this).opened_fd, &raw mut (*this).io_poll) }; io::Action::Readable(FileAction { on_error: Self::on_io_error_thunk, @@ -505,19 +494,15 @@ impl ReadFile { }) } - /// Hands `*this` to the io thread until `opened_fd` is readable; it comes - /// back through `on_ready` / `on_io_error`. The schedule is the last access - /// (see `update` for why this is pointer-shaped). + /// Hands `*this` to the io thread; it comes back through `on_ready` / + /// `on_io_error`. /// /// # Safety - /// `this` is the live `ReadFile` this thread currently holds; the caller - /// must not touch it afterwards. + /// As [`update`](Self::update). #[cfg(not(windows))] pub(crate) unsafe fn wait_for_readable(this: *mut Self) { bloblog!("ReadFile.waitForReadable"); - // SAFETY: fn contract. The `store_callback_seq_cst` reborrow ends with - // its statement; the request is projected (not reborrowed) for - // `schedule`, which needs the whole `ReadFile` reachable through it. + // SAFETY: fn contract; the schedule is the last access. unsafe { (*this).close_after_io = true; (*this) @@ -691,32 +676,28 @@ impl ReadFile { self.file_store.pathlike.is_path() } - /// Ends with one of two hand-overs: `do_close` gives `*this` to the io - /// thread, otherwise `io_task.finish()` gives it to the JS thread, which - /// frees it. Pointer-shaped for that reason (see `update`). + /// Hands `*this` to the io thread (`do_close`) or to the JS thread, which + /// frees it (`io_task.finish()`). /// /// # Safety - /// `this` is the live `ReadFile` this thread currently holds; the caller - /// must not touch it afterwards. + /// As [`update`](Self::update). #[cfg(not(windows))] unsafe fn on_finish(this: *mut Self) { - // SAFETY: fn contract; these reads and the `is_allowed_to_close` - // reborrow end before `do_close`. + // SAFETY: fn contract; the reborrow ends with the block. let (close_after_io, is_allowed_to_close_fd) = unsafe { (*this).size = (*this).buffer.len() as SizeType; ((*this).close_after_io, (*this).is_allowed_to_close()) }; - // SAFETY: fn contract, passed through; on `true` the io thread owns - // `*this` and nothing below runs. + // SAFETY: fn contract, passed through. if unsafe { Self::do_close(this, is_allowed_to_close_fd) } { bloblog!("ReadFile.onFinish() = deferred"); // we have to wait for the close to finish return; } if !close_after_io { - // SAFETY: fn contract — `do_close` returned `false`, so `*this` is - // still ours; the `take` ends before `finish` posts the job. + // SAFETY: `do_close` returned `false`, so `*this` is still ours; + // `finish` is the hand-over. if let Some(io_task) = unsafe { (*this).io_task.take() } { bloblog!("ReadFile.onFinish() = immediately"); io_task.finish(); @@ -777,22 +758,18 @@ impl ReadFile { } } - /// `FileOpener::get_fd`'s callback, reached from `JobContext::run`. Both - /// of those hand this frame `&mut self`, so unlike the pool re-entries - /// (`update`) a reference to the object is still live up the stack during - /// the hand-over; this frame at least finishes its own use of `self` - /// first and hands over through a pointer as its last act. + /// `FileOpener::get_fd`'s callback. `get_fd` and `JobContext::run` still + /// pass `&mut self` down to here, so this frame's own use of it ends before + /// the hand-over. #[cfg(not(windows))] fn run_async_with_fd(&mut self, fd: Fd) { let next = self.prepare_read(fd); let this: *mut Self = self; - // SAFETY: `this` is the live `ReadFile` `get_fd` called us with, and - // this frame does not use `self` or `this` after the call. + // SAFETY: live `ReadFile` from `get_fd`; neither binding is used afterwards. unsafe { Self::proceed(this, next) } } - /// The part of starting a read that happens before anything is handed - /// on: stat, buffer sizing, and the initial readability check. + /// Stat, buffer sizing and the initial readability check. #[cfg(not(windows))] fn prepare_read(&mut self, fd: Fd) -> Next { if self.errno.is_some() { @@ -851,10 +828,8 @@ impl ReadFile { } fn do_read_loop_task(task: *mut WorkPoolTask) { - // SAFETY: only reached via `WorkPoolTask::callback` with `task` = - // `&raw mut self.task` (intrusive) scheduled by `on_ready` / - // `on_io_error`; recover the parent, which is this thread's until - // `update` hands it on. + // SAFETY: `task` is the intrusive field `on_ready` / `on_io_error` + // scheduled; the parent is this thread's until `update` hands it on. unsafe { Self::update(ReadFile::from_task_ptr(task)) } } @@ -862,16 +837,14 @@ impl ReadFile { /// As [`update`](Self::update). #[cfg(not(windows))] unsafe fn do_read_loop(this: *mut Self) { - // SAFETY: fn contract; the reborrow ends with the call, before - // `proceed` hands the object on through `this`. + // SAFETY: fn contract; the reborrow ends with the call. let next = unsafe { (*this).read_until_blocked() }; // SAFETY: fn contract, passed through. unsafe { Self::proceed(this, next) } } - /// Reads until the read is complete (`Finish`: EOF, `max_length`, or an - /// error recorded in `errno`) or the fd would block (`WaitForReadable`). - /// Never produces `ReadLoop`. + /// Reads until done (`Finish`: EOF, `max_length` or `errno`) or the fd + /// would block (`WaitForReadable`). #[cfg(not(windows))] fn read_until_blocked(&mut self) -> Next { // we hold a 64 KB stack buffer incase the amount of data to @@ -1211,8 +1184,7 @@ impl<'a> ReadFileUV<'a> { if needs_close { let is_allowed_to_close_fd = self.is_allowed_to_close(); - // SAFETY: `self` is live; with no `io_request` (see the - // `FileCloser` impl) this only closes the fd and returns `false`. + // SAFETY: `self` is live; without an `io_request` this only closes the fd. if unsafe { Self::do_close(self, is_allowed_to_close_fd) } { // we have to wait for the close to finish return; diff --git a/src/runtime/webcore/blob/write_file.rs b/src/runtime/webcore/blob/write_file.rs index da52f568219f..4b1bd20b0051 100644 --- a/src/runtime/webcore/blob/write_file.rs +++ b/src/runtime/webcore/blob/write_file.rs @@ -220,19 +220,16 @@ impl WriteFile { WorkPool::schedule(&raw mut this.task); } - /// io thread: the `io::RequestCallback` installed by `wait_for_writable`. + /// The `io::RequestCallback` installed by `wait_for_writable`. /// /// # Safety - /// `io::RequestCallback`'s contract: `request` is the pointer - /// `wait_for_writable` scheduled. + /// `io::RequestCallback`'s contract. #[cfg(not(windows))] pub(crate) unsafe fn on_request_writable(request: *mut io::Request) -> io::Action { bun_output::scoped_log!(WriteFile, "WriteFile.onRequestWritable()"); - // SAFETY: fn contract — `request` is `&raw mut (*this).io_request` of a - // live `WriteFile`, so the parent is reachable through it. + // SAFETY: fn contract — `request` is the projection `wait_for_writable` scheduled. let this = unsafe { WriteFile::from_io_request(request) }; - // SAFETY: as above; the `fd` read ends here and `io_poll` is only - // projected, so no reference into the parent outlives this call. + // SAFETY: as above; a field read and a projection, no reference escapes. let (fd, poll) = unsafe { ((*this).opened_fd, &raw mut (*this).io_poll) }; io::Action::Writable(io::FileAction { on_error: Self::on_io_error, @@ -243,18 +240,14 @@ impl WriteFile { }) } - /// Hands `*this` to the io thread until `opened_fd` is writable; it comes - /// back through `on_ready` / `on_io_error`. The schedule is the last access - /// (see `update` for why this is pointer-shaped). + /// Hands `*this` to the io thread; it comes back through `on_ready` / + /// `on_io_error`. /// /// # Safety - /// `this` is the live `WriteFile` this thread currently holds; the caller - /// must not touch it afterwards. + /// As [`update`](Self::update). #[cfg(not(windows))] pub(crate) unsafe fn wait_for_writable(this: *mut Self) { - // SAFETY: fn contract. The `store_callback_seq_cst` reborrow ends with - // its statement; the request is projected (not reborrowed) for - // `schedule`, which needs the whole `WriteFile` reachable through it. + // SAFETY: fn contract; the schedule is the last access. unsafe { (*this).close_after_io = true; (*this) @@ -323,10 +316,7 @@ impl WriteFile { // reshaped for borrowck — take (off, len) here and re-derive the slice // internally so callers don't hold a borrow of self across the &mut self call. - // - // Reports `WouldBlock` rather than waiting itself: the wait hands the file to - // the io thread, so it has to be the caller's last step, not happen under - // this `&mut self` with the caller about to read `errno`. + // Waiting is the caller's last step (a hand-over), hence `WouldBlock`. #[cfg(not(windows))] fn do_write(&mut self, off: usize, len: usize) -> WriteStep { let fd = self.opened_fd; @@ -422,52 +412,43 @@ impl WriteFile { .is_path() } - /// Ends with one of two hand-overs: `do_close` gives `*this` to the io - /// thread, otherwise `io_task.finish()` gives it to the JS thread, which - /// frees it. Pointer-shaped for that reason (see `update`). + /// Hands `*this` to the io thread (`do_close`) or to the JS thread, which + /// frees it (`io_task.finish()`). /// /// # Safety - /// `this` is the live `WriteFile` this thread currently holds; the caller - /// must not touch it afterwards. + /// As [`update`](Self::update). #[cfg(not(windows))] unsafe fn on_finish(this: *mut Self) { bun_output::scoped_log!(WriteFile, "WriteFile.onFinish()"); - // SAFETY: fn contract; the read and the `is_allowed_to_close` reborrow - // end before `do_close`. + // SAFETY: fn contract; the reborrow ends with the statement. let (close_after_io, is_allowed_to_close_fd) = unsafe { ((*this).close_after_io, (*this).is_allowed_to_close()) }; - // SAFETY: fn contract, passed through; on `true` the io thread owns - // `*this` and nothing below runs. + // SAFETY: fn contract, passed through. if unsafe { Self::do_close(this, is_allowed_to_close_fd) } { return; } if !close_after_io { - // SAFETY: fn contract — `do_close` returned `false`, so `*this` is - // still ours; the `take` ends before `finish` posts the job. + // SAFETY: `do_close` returned `false`, so `*this` is still ours; + // `finish` is the hand-over. if let Some(io_task) = unsafe { (*this).io_task.take() } { io_task.finish(); } } } - /// `FileOpener::get_fd`'s callback, reached from `JobContext::run`. Both - /// of those hand this frame `&mut self`, so unlike the pool re-entries - /// (`update`) a reference to the object is still live up the stack during - /// the hand-over; this frame at least finishes its own use of `self` - /// first and hands over through a pointer as its last act. + /// `FileOpener::get_fd`'s callback. `get_fd` and `JobContext::run` still + /// pass `&mut self` down to here, so this frame's own use of it ends before + /// the hand-over. #[cfg(not(windows))] fn run_with_fd(&mut self, fd: Fd) { let next = self.prepare_write(fd); let this: *mut Self = self; - // SAFETY: `this` is the live `WriteFile` `get_fd` called us with, and - // this frame does not use `self` or `this` after the call. + // SAFETY: live `WriteFile` from `get_fd`; neither binding is used afterwards. unsafe { Self::proceed(this, next) } } - /// The part of starting a write that happens before anything is handed - /// on: deciding whether the fd can block, preallocating, and the initial - /// writability check. + /// Blocking-ness, preallocation and the initial writability check. #[cfg(not(windows))] fn prepare_write(&mut self, fd_: Fd) -> Next { if fd_ == Fd::INVALID || self.errno.is_some() { @@ -539,10 +520,8 @@ impl WriteFile { } fn do_write_loop_task(task: *mut WorkPoolTask) { - // SAFETY: only reached via `WorkPoolTask::callback` with `task` = - // `&raw mut self.task` (intrusive) scheduled by `on_ready` / - // `on_io_error`; recover the parent, which is this thread's until - // `update` hands it on. + // SAFETY: `task` is the intrusive field `on_ready` / `on_io_error` + // scheduled; the parent is this thread's until `update` hands it on. unsafe { let this = WriteFile::from_task_ptr(task); // On macOS, we use one-shot mode, so we don't need to unregister. @@ -554,19 +533,15 @@ impl WriteFile { } } - /// Pool thread re-entry (`do_write_loop_task`, `on_close_io_request`). - /// Pointer-shaped, like everything it calls, for the reason given on - /// `ReadFile::update`: every path below ends by handing `*this` to another - /// thread. + /// Pool thread re-entry (`do_write_loop_task`, `on_close_io_request`); + /// pointer-shaped for the reason given on `ReadFile::update`. /// /// # Safety - /// `this` is the live `WriteFile` this thread currently holds; the caller - /// must not touch it afterwards. + /// `this` is the live `WriteFile` this thread holds; not to be touched after. pub(crate) unsafe fn update(this: *mut Self) { #[cfg(windows)] { - let _ = this; - // Windows writes go through WriteFileWindows; this is never scheduled. + let _ = this; // Writes go through WriteFileWindows; never scheduled. } #[cfg(not(windows))] { @@ -575,9 +550,7 @@ impl WriteFile { } } - /// Performs the step a pool-side function decided on once its own - /// accesses to `*this` are over. `WriteLoop` writes until it has to hand - /// the object on; the other two are the hand-overs themselves. + /// Performs the step a `&mut self` stage decided on, after that borrow ended. /// /// # Safety /// As [`update`](Self::update). @@ -597,16 +570,14 @@ impl WriteFile { /// As [`update`](Self::update). #[cfg(not(windows))] unsafe fn do_write_loop(this: *mut Self) { - // SAFETY: fn contract; the reborrow ends with the call, before - // `proceed` hands the object on through `this`. + // SAFETY: fn contract; the reborrow ends with the call. let next = unsafe { (*this).write_until_blocked() }; // SAFETY: fn contract, passed through. unsafe { Self::proceed(this, next) } } - /// Writes until the write is complete (`Finish`: everything written, a - /// zero-length write, closing, or an error recorded in `errno`) or the fd - /// would block (`WaitForWritable`). Never produces `WriteLoop`. + /// Writes until done (`Finish`: all written, a zero-length write, closing, + /// or `errno`) or the fd would block (`WaitForWritable`). #[cfg(not(windows))] fn write_until_blocked(&mut self) -> Next { while self.state.load(Ordering::Relaxed) == ClosingState::Running as u8 { diff --git a/test/js/bun/util/bun-file-read.test.ts b/test/js/bun/util/bun-file-read.test.ts index 346f1004dbd0..0aa017a93d6b 100644 --- a/test/js/bun/util/bun-file-read.test.ts +++ b/test/js/bun/util/bun-file-read.test.ts @@ -68,10 +68,14 @@ describe.skipIf(isWindows)("Bun.file(fifo)", () => { using dir = tempDir("bun-file-read-fifo", {}); const fifo = path.join(String(dir), "in.fifo"); mkfifo(fifo); - // Opening the write end needs a reader to exist. The holder stays open so - // the child finds a connected writer (rather than EOF) whenever it opens - // the FIFO; it never reads, so every byte goes to the child. - const holder = openSync(fifo, constants.O_RDONLY | constants.O_NONBLOCK); + // 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. + let holder = openSync(fifo, constants.O_RDONLY | constants.O_NONBLOCK); + const closeHolder = () => { + if (holder !== -1) closeSync(holder); + holder = -1; + }; let writer = openSync(fifo, "w"); try { await using proc = Bun.spawn({ @@ -84,14 +88,22 @@ describe.skipIf(isWindows)("Bun.file(fifo)", () => { stdout: "pipe", stderr: "pipe", }); - // The write end is blocking, so this only completes as the child drains - // the pipe; closing it afterwards is what ends the child's read. - const written = await Bun.write(Bun.file(writer), payload); + const stderr = proc.stderr.text(); + // The write end is blocking, so the 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 => { + closeHolder(); + throw new Error(`child exited with ${exitCode} before the payload was written: ${await stderr}`); + }); + const written = await Promise.race([Bun.write(Bun.file(writer), payload), childDied]); closeSync(writer); writer = -1; - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + const [stdout, stderrText, exitCode] = await Promise.all([proc.stdout.text(), stderr, proc.exited]); - expect({ written, stdout, stderr }).toEqual({ + expect({ written, stdout, stderr: stderrText }).toEqual({ written: payload.length, stdout: `${payload.length} ${Bun.hash(payload)}`, stderr: "", @@ -99,7 +111,7 @@ describe.skipIf(isWindows)("Bun.file(fifo)", () => { expect(exitCode).toBe(0); } finally { if (writer !== -1) closeSync(writer); - closeSync(holder); + closeHolder(); } }); }); From c7ad7e08c3b751f75644b39ebb153d74bbb407c9 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Wed, 12 Aug 2026 09:57:47 +0000 Subject: [PATCH 3/4] Skip the FIFO-by-path read test on macOS, where the read never completes The child's Bun.file(fifo).bytes() did not finish on the darwin lanes while the same setup completes on Linux; the streaming twin of this case in streams.test.js is already todo'd on macOS. The kqueue half of the change is exercised there by the Bun.stdin pipe tests; this test adds the path-opened variant, which is Linux-only for now. --- test/js/bun/util/bun-file-read.test.ts | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/test/js/bun/util/bun-file-read.test.ts b/test/js/bun/util/bun-file-read.test.ts index 0aa017a93d6b..98a68fdc5ef7 100644 --- a/test/js/bun/util/bun-file-read.test.ts +++ b/test/js/bun/util/bun-file-read.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "bun:test"; -import { bunEnv, bunExe, isWindows, tempDir } from "harness"; +import { bunEnv, bunExe, isMacOS, isWindows, tempDir } from "harness"; import { mkfifo } from "mkfifo"; import { randomBytes } from "node:crypto"; import { closeSync, constants, openSync } from "node:fs"; @@ -62,7 +62,14 @@ describe("Bun.file read-loop target selection", () => { // ReadFile is handed to the io thread every time it drains the pipe and has to // wait for more, and once the writer closes it is (on Linux) handed over once // more to unregister the fd before the fd it opened is closed. -describe.skipIf(isWindows)("Bun.file(fifo)", () => { +// +// macOS: the child never finishes this read (the test times out with the child +// still alive) while the same setup completes on Linux; reading a FIFO by path +// on macOS is also what "Bun.file() read text from pipe" in +// test/js/web/streams/streams.test.js is todo'd for. The read-side hand-overs +// are covered on macOS by the Bun.stdin pipe tests (test/regression/issue/07500, +// bun-stdin-slice.test.ts); what this adds is the path-opened variant. +describe.skipIf(isWindows || isMacOS)("Bun.file(fifo)", () => { it("bytes() reads a pipe that is filled in pieces and ends when the writer closes", async () => { const payload = randomBytes(256 * 1024); using dir = tempDir("bun-file-read-fifo", {}); From d02b34209d079bdaa4bdc1e8fe81d86937cce7b9 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Wed, 12 Aug 2026 11:05:18 +0000 Subject: [PATCH 4/4] Make the FIFO tests observe the io-thread hand-overs Both tests passed the same way whether or not the ReadFile/WriteFile ever reached the io thread. In debug builds the child now writes its WriteFile scope trace to a file (BUN_DEBUG); the parent waits for the request to show up there before feeding or draining the pipe, which makes at least one round trip certain, and afterwards checks that every request came back as a readiness callback and that the close round-tripped through the io thread. A second write-side case issues two concurrent writes on one fd: the second epoll registration fails with EEXIST, which is the registration-error hand-over, and the trace checks that the failed WriteFile still closed through the io thread before rejecting. The write test's drain also no longer blocks forever in open() if the child died after reporting the pipe full: the open is raced against the child's exit, and a lost race lends the pending open a writer so it can be closed. waitForFileToContain is added to the harness for the trace waits. --- test/harness.ts | 14 ++ test/js/bun/io/bun-write.test.js | 242 +++++++++++++++++++------ test/js/bun/util/bun-file-read.test.ts | 52 ++++-- 3 files changed, 235 insertions(+), 73 deletions(-) diff --git a/test/harness.ts b/test/harness.ts index a6c07623ec17..67c941b60f9b 100644 --- a/test/harness.ts +++ b/test/harness.ts @@ -1832,6 +1832,20 @@ export function waitForFileToExist(path: string, interval_ms: number) { } } +/** + * Resolves with the file's contents once they contain `needle` at least + * `occurrences` times. Polls without a deadline of its own (the test timeout is + * the deadline); race it against a failure signal, such as the exit of the + * process expected to write the file. + */ +export async function waitForFileToContain(path: string, needle: string, occurrences = 1): Promise { + while (true) { + const text = fs.existsSync(path) ? fs.readFileSync(path, "utf8") : ""; + if (text.split(needle).length - 1 >= occurrences) return text; + await Bun.sleep(5); + } +} + export function libcPathForDlopen() { switch (process.platform) { case "linux": diff --git a/test/js/bun/io/bun-write.test.js b/test/js/bun/io/bun-write.test.js index 9cac7bde4d7a..a8b7e462c5a2 100644 --- a/test/js/bun/io/bun-write.test.js +++ b/test/js/bun/io/bun-write.test.js @@ -7,8 +7,12 @@ import { exampleSite, gcTick, isASAN, + isDebug, + isLinux, + isMacOS, isWindows, tempDir, + waitForFileToContain, withoutAggressiveGC, } from "harness"; import { mkfifo } from "mkfifo"; @@ -538,79 +542,197 @@ const IS_UV_FS_COPYFILE_DISABLED = expect(exitCode).toBe(0); }); - // A non-blocking FIFO that is already full when Bun.write() starts: the - // WriteFile is handed to the io thread before its first write(), again each - // time the pipe fills up while the payload drains, and (on Linux) once more - // to unregister the fd when it is done. The payload is at least 256 KiB so - // Bun.write() takes the WriteFile path rather than the synchronous one. - it.skipIf(isWindows)("Bun.write(Bun.file(fd)) to a full non-blocking FIFO waits for the reader", async () => { + // Bun.write() into a non-blocking FIFO that is already full when it starts, so + // every WriteFile involved is handed to the io thread before its first + // write() (the pipe is not drained until that has happened), then again + // whenever the pipe fills up while the payload drains, and (except on macOS, + // whose one-shot kqueue registration needs no unregistering) once more at + // the end to unregister the fd. In debug builds the child's WriteFile trace + // proves those hand-overs; in release builds the logging is compiled out and + // only the outcome is checked. 256 KiB skips Bun.write()'s synchronous path + // (the full pipe would make it fall back to WriteFile anyway) and is several + // pipe buffers long. + describe.skipIf(isWindows)("Bun.write(Bun.file(fd)) into a full non-blocking FIFO", () => { const size = 256 * 1024; - using dir = tempDir("bun-write-fifo", {}); - const fifo = join(String(dir), "out.fifo"); - mkfifo(fifo); - // A reader has to exist for the child's O_NONBLOCK open of the write end to - // succeed; this one never reads, the readFile below does. - const holder = fs.openSync(fifo, fs.constants.O_RDONLY | fs.constants.O_NONBLOCK); - try { - const script = ` - const fs = require("fs"); - const fd = fs.openSync(process.env.FIFO, fs.constants.O_WRONLY | fs.constants.O_NONBLOCK); - const dest = Bun.file(fd); - // fstat()s the fd: Bun.write() only polls a destination it knows is not a regular file. - dest.size; - const chunk = Buffer.alloc(4096, "P"); - let filled = 0; + const count = (trace, marker) => trace.split(marker).length - 1; + + // Runs `writes` concurrent Bun.write()s of the payload in a child, drains + // the FIFO once the child reports the pipe full (and, in debug builds, once + // `writes` WriteFiles have reached the io thread), and returns what came + // out of the pipe and how each write settled. + async function writeIntoFullFifo(writes) { + using dir = tempDir("bun-write-fifo", {}); + const fifo = join(String(dir), "out.fifo"); + const trace = join(String(dir), "trace.log"); + mkfifo(fifo); + // A reader has to exist for the child's O_NONBLOCK open of the write end + // to succeed; this one never reads, the drain below does. + const holder = fs.openSync(fifo, fs.constants.O_RDONLY | fs.constants.O_NONBLOCK); + let opening; + try { + const script = ` + const fs = require("fs"); + const fd = fs.openSync(process.env.FIFO, fs.constants.O_WRONLY | fs.constants.O_NONBLOCK); + const dest = Bun.file(fd); + // fstat()s the fd: Bun.write() only polls a destination it knows is not a regular file. + dest.size; + const chunk = Buffer.alloc(4096, "P"); + let filled = 0; + try { + for (;;) filled += fs.writeSync(fd, chunk); + } catch (e) { + if (e.code !== "EAGAIN") throw e; + } + const payload = Buffer.alloc(${size}, "W"); + const settled = Promise.allSettled(Array.from({ length: ${writes} }, () => Bun.write(dest, payload))); + process.stderr.write("filled " + filled + "\\n"); + const results = (await settled).map(r => + r.status === "fulfilled" ? { written: r.value } : { code: r.reason.code, syscall: r.reason.syscall }, + ); + process.stdout.write(JSON.stringify(results)); + `; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", script], + // BUN_DEBUG sends the scoped debug logs to a file instead of stdout. + env: { ...bunEnv, FIFO: fifo, ...(isDebug ? { BUN_DEBUG_WriteFile: "1", BUN_DEBUG: trace } : {}) }, + stdout: "pipe", + stderr: "pipe", + }); + const stdout = proc.stdout.text(); + const filled = Promise.withResolvers(); + const stderr = (async () => { + let text = ""; + for await (const chunk of proc.stderr) { + text += Buffer.from(chunk).toString(); + const m = /^filled (\d+)\n/m.exec(text); + if (m) filled.resolve(Number(m[1])); + } + return text; + })(); + // The child cannot finish before the pipe is drained, so exiting before + // the drain has started is a failure. A blocking open of the read end + // would otherwise wait forever for a writer once the child is gone; + // after it, a dying child just ends the drain early (EOF) and the + // assertions report what arrived. + const childDied = proc.exited.then(async exitCode => { + throw new Error(`child exited with ${exitCode} before the FIFO was drained: ${await stderr}`); + }); + const prefill = await Promise.race([filled.promise, childDied]); + if (isDebug) { + await Promise.race([waitForFileToContain(trace, "WriteFile.onRequestWritable()", writes), childDied]); + } + opening = fs.promises.open(fifo, "r"); + const drain = await Promise.race([opening, childDied]); + opening = undefined; + let data; try { - for (;;) filled += fs.writeSync(fd, chunk); - } catch (e) { - if (e.code !== "EAGAIN") throw e; + data = await drain.readFile(); + } finally { + await drain.close(); } - const written = Bun.write(dest, Buffer.alloc(${size}, "W")); - process.stderr.write("filled " + filled + "\\n"); - process.stdout.write(String(await written)); - `; - await using proc = Bun.spawn({ - cmd: [bunExe(), "-e", script], - env: { ...bunEnv, FIFO: fifo }, - stdout: "pipe", - stderr: "pipe", - }); - const stdout = proc.stdout.text(); - const filled = Promise.withResolvers(); - const stderr = (async () => { - let text = ""; - for await (const chunk of proc.stderr) { - text += Buffer.from(chunk).toString(); - const m = /^filled (\d+)\n/m.exec(text); - if (m) filled.resolve(Number(m[1])); + const [stdoutText, stderrText, exitCode] = await Promise.all([stdout, stderr, proc.exited]); + let results; + try { + results = JSON.parse(stdoutText); + } catch { + results = stdoutText; } - filled.reject(new Error(`child exited before filling the FIFO: ${text}`)); - return text; - })(); - // Only start draining once the child has filled the pipe and started the write. - const prefill = await filled.promise; - const data = await fs.promises.readFile(fifo); - const [resolved, stderrText, exitCode] = await Promise.all([stdout, stderr, proc.exited]); + return { + prefill, + data, + results, + stderr: stderrText, + exitCode, + log: isDebug ? fs.readFileSync(trace, "utf8") : "", + }; + } finally { + if (opening) { + // The open lost the race above and is still waiting for a writer: + // lend it one so it completes, then close what it opened. + opening.then( + handle => handle.close(), + () => {}, + ); + fs.closeSync(fs.openSync(fifo, fs.constants.O_WRONLY | fs.constants.O_NONBLOCK)); + } + fs.closeSync(holder); + } + } - expect({ + function pipeContents(prefill, data, payloads) { + return { prefilled: prefill > 0, - resolved, - stderr: stderrText, length: data.length, prefillIntact: data.subarray(0, prefill).equals(Buffer.alloc(prefill, "P")), - payloadIntact: data.subarray(prefill).equals(Buffer.alloc(size, "W")), - }).toEqual({ + payloadsIntact: data.subarray(prefill).equals(Buffer.alloc(size * payloads, "W")), + }; + } + + it("waits for the reader and delivers the payload", async () => { + const { prefill, data, results, stderr, exitCode, log } = await writeIntoFullFifo(1); + + expect({ ...pipeContents(prefill, data, 1), results, stderr }).toEqual({ prefilled: true, - resolved: String(size), - stderr: `filled ${prefill}\n`, length: prefill + size, prefillIntact: true, - payloadIntact: true, + payloadsIntact: true, + results: [{ written: size }], + stderr: `filled ${prefill}\n`, }); expect(exitCode).toBe(0); - } finally { - fs.closeSync(holder); - } + + if (isDebug) { + const requests = count(log, "WriteFile.onRequestWritable()"); + expect(requests).toBeGreaterThanOrEqual(1); + // Every request the io thread popped came back as one readiness + // callback. Closing round-trips through the io thread as well, which + // makes on_finish run a second time; macOS clears that flag when the + // one-shot registration fires, so there it runs once. + expect({ + readies: count(log, "WriteFile.onReady()"), + ioErrors: count(log, "WriteFile.onIOError()"), + finishes: count(log, "WriteFile.onFinish()"), + }).toEqual({ readies: requests, ioErrors: 0, finishes: isMacOS ? 1 : 2 }); + } + }); + + // Two WriteFiles on one fd: the second epoll registration of the fd fails + // with EEXIST (two pollers per fd is not supported today; if that ever + // changes, this case becomes two successful writes), which is the one way + // to reach the registration-error hand-over: the io thread reports the + // error back, and the failed WriteFile still closes through the io thread + // and rejects, while the other one delivers its payload. Debug builds only, + // because only the trace makes the outcome deterministic: the pipe is not + // drained until both requests have reached the io thread, so the second + // registration always finds the first one in place. Linux only: on kqueue + // the second registration silently replaces the first instead of failing. + it.skipIf(!isLinux || !isDebug)( + "a second concurrent write is rejected with EEXIST and the first still completes", + async () => { + const { prefill, data, results, stderr, exitCode, log } = await writeIntoFullFifo(2); + + expect({ ...pipeContents(prefill, data, 1), results, stderr }).toEqual({ + prefilled: true, + length: prefill + size, + prefillIntact: true, + payloadsIntact: true, + results: expect.arrayContaining([{ written: size }, { code: "EEXIST", syscall: "epoll_ctl" }]), + stderr: `filled ${prefill}\n`, + }); + expect(results).toHaveLength(2); + expect(exitCode).toBe(0); + + const requests = count(log, "WriteFile.onRequestWritable()"); + expect(requests).toBeGreaterThanOrEqual(2); + // One request ended in the error callback instead of a readiness + // callback; both WriteFiles still closed through the io thread. + expect({ + readies: count(log, "WriteFile.onReady()"), + ioErrors: count(log, "WriteFile.onIOError()"), + finishes: count(log, "WriteFile.onFinish()"), + }).toEqual({ readies: requests - 1, ioErrors: 1, finishes: 4 }); + }, + ); }); it("Bun.file(0) survives GC", async () => { diff --git a/test/js/bun/util/bun-file-read.test.ts b/test/js/bun/util/bun-file-read.test.ts index 98a68fdc5ef7..cb91f4e80f88 100644 --- a/test/js/bun/util/bun-file-read.test.ts +++ b/test/js/bun/util/bun-file-read.test.ts @@ -1,8 +1,8 @@ import { describe, expect, it } from "bun:test"; -import { bunEnv, bunExe, isMacOS, isWindows, tempDir } from "harness"; +import { bunEnv, bunExe, isDebug, isMacOS, isWindows, tempDir, waitForFileToContain } from "harness"; import { mkfifo } from "mkfifo"; import { randomBytes } from "node:crypto"; -import { closeSync, constants, openSync } from "node:fs"; +import { closeSync, constants, openSync, readFileSync } from "node:fs"; import { tmpdir } from "node:os"; import path from "node:path"; @@ -58,10 +58,12 @@ describe("Bun.file read-loop target selection", () => { }); }); -// Reading a FIFO by path. The payload is several pipe buffers long, so the -// ReadFile is handed to the io thread every time it drains the pipe and has to -// wait for more, and once the writer closes it is (on Linux) handed over once -// more to unregister the fd before the fd it opened is closed. +// Reading a FIFO by path. Nothing is written until the child has found the pipe +// empty and handed its ReadFile to the io thread, so the read goes through at +// least one io-thread round trip, and (on Linux) through one more at the end to +// unregister the fd before the fd it opened is closed. In debug builds the +// child's ReadFile trace is what proves those hand-overs happened; in release +// builds the logging is compiled out and only the result is checked. // // macOS: the child never finishes this read (the test times out with the child // still alive) while the same setup completes on Linux; reading a FIFO by path @@ -70,10 +72,13 @@ describe("Bun.file read-loop target selection", () => { // are covered on macOS by the Bun.stdin pipe tests (test/regression/issue/07500, // bun-stdin-slice.test.ts); what this adds is the path-opened variant. describe.skipIf(isWindows || isMacOS)("Bun.file(fifo)", () => { - it("bytes() reads a pipe that is filled in pieces and ends when the writer closes", async () => { + const count = (trace: string, marker: string) => trace.split(marker).length - 1; + + it("bytes() waits for the writer, reads what it writes and ends when it closes", async () => { const payload = randomBytes(256 * 1024); using dir = tempDir("bun-file-read-fifo", {}); const fifo = path.join(String(dir), "in.fifo"); + const trace = path.join(String(dir), "trace.log"); 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 @@ -91,20 +96,25 @@ describe.skipIf(isWindows || isMacOS)("Bun.file(fifo)", () => { "-e", `const bytes = await Bun.file(process.env.FIFO).bytes(); process.stdout.write(bytes.length + " " + Bun.hash(bytes));`, ], - env: { ...bunEnv, FIFO: fifo }, + // ReadFile logs under the WriteFile scope; BUN_DEBUG sends the scoped logs to a file. + env: { ...bunEnv, FIFO: fifo, ...(isDebug ? { BUN_DEBUG_WriteFile: "1", BUN_DEBUG: trace } : {}) }, stdout: "pipe", stderr: "pipe", }); const stderr = proc.stderr.text(); - // The write end is blocking, so the 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. + // The child cannot exit before the payload is written unless it failed; + // dropping `holder` then leaves the pipe without readers, so a write + // blocked on it fails with EPIPE instead of waiting forever. const childDied = proc.exited.then(async exitCode => { closeHolder(); throw new Error(`child exited with ${exitCode} before the payload was written: ${await stderr}`); }); + if (isDebug) { + // The io thread has run the child's request: it is waiting for data that does not exist yet. + await Promise.race([waitForFileToContain(trace, "ReadFile.onRequestReadable"), childDied]); + } + // The write end is blocking, so this completes as the child drains the + // pipe; closing it afterwards is what ends the child's read. const written = await Promise.race([Bun.write(Bun.file(writer), payload), childDied]); closeSync(writer); writer = -1; @@ -116,6 +126,22 @@ describe.skipIf(isWindows || isMacOS)("Bun.file(fifo)", () => { stderr: "", }); expect(exitCode).toBe(0); + + if (isDebug) { + const log = readFileSync(trace, "utf8"); + const waits = count(log, "ReadFile.waitForReadable"); + expect(waits).toBeGreaterThanOrEqual(1); + // Every wait is one request popped by the io thread and one readiness + // callback; the close then round-trips through the io thread once + // (deferred) before the read completes (immediately). + expect({ + requests: count(log, "ReadFile.onRequestReadable"), + readies: count(log, "ReadFile.onReady"), + ioErrors: count(log, "ReadFile.onIOError"), + deferredCloses: count(log, "ReadFile.onFinish() = deferred"), + completions: count(log, "ReadFile.onFinish() = immediately"), + }).toEqual({ requests: waits, readies: waits, ioErrors: 0, deferredCloses: 1, completions: 1 }); + } } finally { if (writer !== -1) closeSync(writer); closeHolder();