diff --git a/src/jsc/CallFrame.rs b/src/jsc/CallFrame.rs index c654b79b2815..b80b2bccd672 100644 --- a/src/jsc/CallFrame.rs +++ b/src/jsc/CallFrame.rs @@ -3,7 +3,6 @@ use core::ffi::{c_uint, c_void}; use crate::virtual_machine::VirtualMachine; use crate::{JSGlobalObject, JSValue}; -use bun_collections::IntegerBitSet; #[cfg(debug_assertions)] use bun_core::ZStr; @@ -229,6 +228,10 @@ pub struct CallerSrcLoc { /// This is an advanced iterator struct which is used by various APIs. In /// Node.fs, `will_be_async` is set to true which allows string/path APIs to /// know if they have to do threadsafe clones. +/// +/// It never roots anything: while the host call runs, the arguments are kept +/// alive by the caller's frame; whatever must outlive the call takes its own +/// hold (`to_thread_safe`). pub struct ArgumentsSlice<'a> { /// Backing storage for the remaining-args view. Both [`Self::init`] and /// [`Self::init_async`] borrow — `all: &'a [JSValue]` already ties this @@ -241,7 +244,6 @@ pub struct ArgumentsSlice<'a> { remaining_start: usize, pub vm: &'a VirtualMachine, pub all: &'a [JSValue], - pub(crate) protected: IntegerBitSet<32>, pub will_be_async: bool, } @@ -252,40 +254,12 @@ impl<'a> ArgumentsSlice<'a> { &self.remaining_buf[self.remaining_start..] } - pub(crate) fn unprotect(&mut self) { - let mut iter = self.protected.iterator::(); - while let Some(i) = iter.next() { - self.all[i].unprotect(); - } - self.protected = IntegerBitSet::<32>::init_empty(); - } - - pub fn protect_eat(&mut self) { - if self.remaining().is_empty() { - return; - } - // `remaining_buf.len() == all.len()` for both init variants, so - // `all.len() - remaining().len()` reduces to `remaining_start`. - let index = self.all.len() - self.remaining().len(); - self.protected.set(index); - self.all[index].protect(); - self.eat(); - } - - pub fn protect_eat_next(&mut self) -> Option { - if self.remaining().is_empty() { - return None; - } - self.next_eat() - } - pub fn init(vm: &'a VirtualMachine, slice: &'a [JSValue]) -> ArgumentsSlice<'a> { ArgumentsSlice { remaining_buf: Cow::Borrowed(slice), remaining_start: 0, vm, all: slice, - protected: IntegerBitSet::<32>::init_empty(), will_be_async: false, } } @@ -314,12 +288,6 @@ impl<'a> ArgumentsSlice<'a> { } } -impl<'a> Drop for ArgumentsSlice<'a> { - fn drop(&mut self) { - self.unprotect(); - } -} - // `CallFrame`/`VM`/`JSGlobalObject` are opaque `UnsafeCell`-backed ZST handles; // `&T` is ABI-identical to non-null `*const T`. Out-params are exclusive `&mut` // to plain `#[repr(C)]` PODs. `describeFrame` returns a raw C string that the diff --git a/src/jsc/VmHandle.rs b/src/jsc/VmHandle.rs index c8a03dc8db22..4c367e3ee64e 100644 --- a/src/jsc/VmHandle.rs +++ b/src/jsc/VmHandle.rs @@ -409,6 +409,19 @@ impl VmHandle { self.0.state() == State::Open } + /// Is the calling thread this VM's thread? Any thread. Compares handle + /// identity, not the `VirtualMachine` address: this `Shared` outlives its + /// VM, so a later VM allocated at the same address is still told apart. + #[inline] + pub fn is_current_thread(&self) -> bool { + match VirtualMachine::get_or_null() { + // SAFETY: the thread-local is this thread's live VM (cleared + // before a worker's VM is freed). + Some(vm) => Arc::ptr_eq(&unsafe { &*vm }.handle_ref().0, &self.0), + None => false, + } + } + pub(crate) fn tickets_outstanding(&self) -> u32 { self.0.tickets.load(Ordering::SeqCst) } diff --git a/src/jsc/array_buffer.rs b/src/jsc/array_buffer.rs index 69947bfbd5a1..1bf4d2f71428 100644 --- a/src/jsc/array_buffer.rs +++ b/src/jsc/array_buffer.rs @@ -1062,6 +1062,123 @@ unsafe impl bun_ptr::ExternalSharedDescriptor for JSCArrayBuffer { } } +unsafe extern "C" { + safe fn JSC__JSValue__retainPinnedArrayBuffer( + value: JSValue, + out_ptr: &mut *const u8, + out_len: &mut usize, + ) -> *mut JSCArrayBuffer; + safe fn JSC__ArrayBuffer__releasePinned(self_: &JSCArrayBuffer); +} + +/// The byte range of a JS `ArrayBuffer` or view, kept alive and in place by +/// one ref + one pin held directly on its `JSC::ArrayBuffer` — the refcounted +/// owner of the storage, not a GC cell. The JS object may be collected while +/// this lives, and releasing touches no `JSCell`, so it may drop inside a GC +/// finalizer. +/// +/// That refcount is not atomic and the owning VM's collector also touches it, +/// so the release must happen on that VM's thread. `slice()` is fine from +/// anywhere; a `Drop` anywhere else (another VM's thread letting go of a +/// shared `Blob` store, a pool thread) is posted back to the owning VM. +pub struct PinnedArrayBuffer { + owner: ptr::NonNull, + ptr: *const u8, + len: usize, + vm: crate::VmHandle, +} + +impl PinnedArrayBuffer { + /// JS thread. `None` if `value` is not a buffer/view or is detached. A + /// view that has no `ArrayBuffer` yet gets one (see + /// `retainPinnedArrayBuffer`). + pub fn retain(value: JSValue) -> Option { + let mut ptr: *const u8 = ptr::null(); + let mut len = 0usize; + let owner = ptr::NonNull::new(JSC__JSValue__retainPinnedArrayBuffer( + value, &mut ptr, &mut len, + ))?; + Some(Self { + owner, + ptr, + len, + vm: crate::virtual_machine::VirtualMachine::get().handle(), + }) + } + + #[inline] + pub fn slice(&self) -> &[u8] { + if self.len == 0 { + return &[]; + } + // SAFETY: `ptr[..len]` lies inside the storage `owner` keeps allocated + // (ref) and undetachable (pin) until `Drop`. + unsafe { core::slice::from_raw_parts(self.ptr, self.len) } + } +} + +impl Drop for PinnedArrayBuffer { + fn drop(&mut self) { + if self.vm.is_current_thread() { + JSC__ArrayBuffer__releasePinned(JSCArrayBuffer::opaque_ref(self.owner.as_ptr())); + } else { + release_on_owning_thread(self.owner, &self.vm); + } + } +} + +/// Hand a ref taken on `vm`'s thread back to that thread to release. If the +/// VM has already closed, its heap went with it and the `ArrayBuffer` (whose +/// wrapper `Weak` lived in that heap) must not be touched again: it is left +/// unreleased. +#[cold] +fn release_on_owning_thread(owner: ptr::NonNull, vm: &crate::VmHandle) { + use bun_event_loop::ConcurrentTask::ConcurrentTask; + use bun_event_loop::ManagedTask::ManagedTask; + + struct Release { + owner: ptr::NonNull, + vm: crate::VmHandle, + } + impl Drop for Release { + // Runs on the owning thread (task run, or freed unrun while that VM + // drains) — or, if the post was refused, right here: then do nothing. + fn drop(&mut self) { + if self.vm.is_current_thread() { + JSC__ArrayBuffer__releasePinned(JSCArrayBuffer::opaque_ref(self.owner.as_ptr())); + } + } + } + fn run(this: *mut Release) -> bun_event_loop::JsResult<()> { + // SAFETY: `this` is the box handed to `new_owned` below; `run` is its + // only consumer on this path (`ManagedTask::run` does not free `ctx`). + drop(unsafe { bun_core::heap::take(this) }); + Ok(()) + } + + let ctx = bun_core::heap::into_raw(Box::new(Release { + owner, + vm: vm.clone(), + })); + let task = ConcurrentTask::create(ManagedTask::new_owned(ctx, run)); + match vm.post(crate::LoopKind::Regular, task) { + crate::Posted::Queued => { + bun_core::scoped_log!( + ArrayBuffer, + "pinned ArrayBuffer released off its VM's thread: posted back" + ); + } + crate::Posted::Refused(task) => { + bun_core::scoped_log!( + ArrayBuffer, + "pinned ArrayBuffer outlived its VM: left unreleased" + ); + // SAFETY: refused ⇒ not queued anywhere; ours to free. + unsafe { ConcurrentTask::release_refused(task) }; + } + } +} + impl JSCArrayBuffer { pub fn as_array_buffer(&mut self) -> ArrayBuffer { let mut out = core::mem::MaybeUninit::::uninit(); diff --git a/src/jsc/bindings/bindings.cpp b/src/jsc/bindings/bindings.cpp index b024bc42ca6c..631a670de9f2 100644 --- a/src/jsc/bindings/bindings.cpp +++ b/src/jsc/bindings/bindings.cpp @@ -3548,6 +3548,49 @@ CPP_DECL void JSC__JSValue__unpinArrayBuffer(JSC::EncodedJSValue v) buf->unpin(); } +// Hold the backing store itself rather than the JS object: one ref + one pin on +// the `JSC::ArrayBuffer` (refcounted, not a GC cell). The wrapper may then be +// collected while the bytes stay allocated and in place, and releasing touches +// no JSCell — so a native holder with no root to lean on, destroyed from a GC +// finalizer (a Blob store) or at an async op's completion, can release inline. +// This VM's thread only: the refcount is not atomic (the Rust holder, +// `PinnedArrayBuffer`, posts an off-thread release back here). +// +// Unlike `pinStorage`, a bufferless view is given its ArrayBuffer here +// (OversizeTypedArray: adopted in place, no byte copy) because there is no +// caller root keeping the view alive. `out_ptr`/`out_len` are the view's byte +// range, read after that so they point into the storage the ArrayBuffer owns. +CPP_DECL JSC::ArrayBuffer* JSC__JSValue__retainPinnedArrayBuffer(JSC::EncodedJSValue v, const uint8_t** out_ptr, size_t* out_len) +{ + auto value = JSC::JSValue::decode(v); + JSC::ArrayBuffer* buf = nullptr; + if (auto* jb = dynamicDowncast(value)) { + buf = jb->impl(); + if (!buf || buf->isDetached()) + return nullptr; + *out_ptr = static_cast(buf->data()); + *out_len = buf->byteLength(); + } else if (auto* view = dynamicDowncast(value); view && !view->isDetached()) { + buf = view->possiblySharedBuffer(); + if (!buf) + return nullptr; + *out_ptr = static_cast(view->vector()); + *out_len = view->byteLength(); + } else { + return nullptr; + } + buf->ref(); + if (!buf->isShared()) + buf->pin(); + return buf; +} +CPP_DECL void JSC__ArrayBuffer__releasePinned(JSC::ArrayBuffer* buf) +{ + if (!buf->isShared()) + buf->unpin(); + buf->deref(); +} + // Borrow `v`'s byte storage for off-thread reading. Splits out only the // `FastTypedArray` case from `pinArrayBuffer`, because that's the one mode // where `possiblySharedBuffer()` actually COPIES data diff --git a/src/jsc/lib.rs b/src/jsc/lib.rs index 605f0c9e9e16..59157c2cf008 100644 --- a/src/jsc/lib.rs +++ b/src/jsc/lib.rs @@ -191,7 +191,7 @@ pub use self::js_value::{ // and is wired into `event_loop::tick` directly at link time. No fn-pointer // hook is re-exported from the crate root. pub use self::array_buffer::{ - ArrayBuffer, BinaryType, JSCArrayBuffer, MarkedArrayBuffer, TypedArrayType, + ArrayBuffer, BinaryType, JSCArrayBuffer, MarkedArrayBuffer, PinnedArrayBuffer, TypedArrayType, }; pub use self::console_object as ConsoleObject; pub use self::console_object::Formatter; diff --git a/src/jsc/node_path.rs b/src/jsc/node_path.rs index e00df5bb73c2..8b0339f6c437 100644 --- a/src/jsc/node_path.rs +++ b/src/jsc/node_path.rs @@ -11,7 +11,7 @@ use bun_core::{SliceWithUnderlyingString, ZigStringSlice}; use bun_ptr::cow_slice::CowSlice; use bun_sys::Fd; -use crate::array_buffer::MarkedArrayBuffer; +use crate::array_buffer::{MarkedArrayBuffer, PinnedArrayBuffer}; // ────────────────────────────────────────────────────────────────────────── // RAII for `protect()`/`unprotect()` pairs taken by `to_thread_safe()`. @@ -24,8 +24,8 @@ use crate::array_buffer::MarkedArrayBuffer; // every early return between `to_thread_safe` and the manual cleanup. // ────────────────────────────────────────────────────────────────────────── -/// Undo the `JSValue::protect()` calls taken by [`to_thread_safe`]( -/// PathLike::to_thread_safe) (or an `args::*` type's `to_thread_safe`). +/// Undo the `JSValue::protect()` calls taken by an `args::*` type's +/// `to_thread_safe` (e.g. `StringOrBuffer::Buffer`). /// /// Implementations release **only** the JS-GC protect refcount — owned Rust /// payloads (Vec, `SliceWithUnderlyingString`, …) are freed by the type's own @@ -100,7 +100,15 @@ impl Default for ThreadSafe { /// `node.PathLike`. pub enum PathLike { String(CowSlice), + /// A JS buffer borrowed for the duration of one call: the argument keeps + /// the cell alive and the pin keeps its storage in place. Anything held + /// past the call goes through [`PathLike::to_thread_safe`] first. Buffer(MarkedArrayBuffer), + /// A `Buffer` after `to_thread_safe`: the same bytes, kept by a ref + pin + /// on the backing `JSC::ArrayBuffer` instead of by the JS cell, so it can + /// be released from a GC finalizer (a `Blob` store's path) or another + /// thread as well as from an async op's completion. + PinnedBuffer(PinnedArrayBuffer), SliceWithUnderlyingString(SliceWithUnderlyingString), ThreadsafeString(SliceWithUnderlyingString), EncodedSlice(ZigStringSlice), @@ -133,6 +141,10 @@ impl Clone for PathLike { owns_buffer: false, pinned: false, }), + // Non-owning, as for `Buffer`: whoever clones a retained path also + // holds what retains it (the `Store`) for at least as long, and a + // borrow can be made and dropped on any thread. + Self::PinnedBuffer(b) => Self::String(CowSlice::init_unchecked(b.slice(), false)), Self::SliceWithUnderlyingString(s) => { // `dupe_ref()` alone leaves `utf8` empty (lib.rs:1603) — a // cloned PathLike would then return b"" from `slice()`. Clone @@ -171,9 +183,8 @@ impl Drop for PathLike { Self::SliceWithUnderlyingString(s) | Self::ThreadsafeString(s) => { core::mem::take(s).deinit(); } - // `ZigStringSlice` releases its WTF ref / owned buffer in its own - // `Drop`. - Self::EncodedSlice(_) => {} + // `PinnedArrayBuffer` / `ZigStringSlice` release in their own `Drop`. + Self::PinnedBuffer(_) | Self::EncodedSlice(_) => {} } } } @@ -189,6 +200,7 @@ impl PathLike { match self { Self::String(s) => s.slice(), Self::Buffer(b) => b.slice(), + Self::PinnedBuffer(b) => b.slice(), Self::SliceWithUnderlyingString(s) | Self::ThreadsafeString(s) => s.slice(), Self::EncodedSlice(s) => s.slice(), } @@ -198,20 +210,18 @@ impl PathLike { match self { Self::String(s) => s.length(), Self::Buffer(b) => b.slice().len(), + Self::PinnedBuffer(b) => b.slice().len(), Self::SliceWithUnderlyingString(_) | Self::ThreadsafeString(_) => 0, Self::EncodedSlice(s) => s.slice().len(), } } - /// Promote any borrowed-JS - /// payload to a thread-safe representation. For `Buffer` the variant is - /// kept and the backing JS value is `protect()`ed (paired with - /// [`Unprotect::unprotect`]); the discriminant is preserved so callers - /// matching on `Buffer` after this call see the same shape. - /// - /// Prefer [`Self::into_thread_safe`] which returns a [`ThreadSafe`] guard; - /// this in-place form exists for nested calls from container types' - /// `to_thread_safe`. + /// Promote any payload that is only valid for the current call into one + /// that can be held past it, read from a work-pool thread, and released on + /// the JS thread wherever the holder happens to die — an async op's + /// completion, or a `Blob` store dropped from its cell's GC finalizer. + /// Zero-copy: a `Buffer` keeps borrowing the same bytes, now owned via the + /// backing store (`PinnedBuffer`) rather than the JS object. pub fn to_thread_safe(&mut self) { match self { Self::SliceWithUnderlyingString(s) => { @@ -220,23 +230,27 @@ impl PathLike { *self = Self::ThreadsafeString(owned); } Self::Buffer(b) => { - b.buffer.value.protect(); + // Dropping the `Buffer` arm afterwards releases its own pin. + *self = match PinnedArrayBuffer::retain(b.buffer.value) { + Some(pinned) => Self::PinnedBuffer(pinned), + // Detached: there are no bytes to keep. + None => Self::default(), + }; } - Self::String(_) | Self::ThreadsafeString(_) | Self::EncodedSlice(_) => {} + Self::String(_) + | Self::PinnedBuffer(_) + | Self::ThreadsafeString(_) + | Self::EncodedSlice(_) => {} } } } impl Unprotect for PathLike { - /// JS-side half of cleanup — undo - /// the `protect()` taken by [`Self::to_thread_safe`] / - /// `ArgumentsSlice::protect_eat`. Owned payloads are released by `Drop`. + /// Nothing to release: [`Self::to_thread_safe`] holds the backing store, + /// not a `protect()`ed cell. Kept so container `args::*` types can forward + /// uniformly. #[inline] - fn unprotect(&mut self) { - if let Self::Buffer(b) = self { - b.buffer.value.unprotect(); - } - } + fn unprotect(&mut self) {} } /// `node.PathOrFileDescriptor`. diff --git a/src/runtime/api/BunObject.rs b/src/runtime/api/BunObject.rs index 25189b61d259..b131ba714273 100644 --- a/src/runtime/api/BunObject.rs +++ b/src/runtime/api/BunObject.rs @@ -1095,7 +1095,7 @@ fn do_resolve(global_this: &JSGlobalObject, arguments: &[JSValue]) -> JsResult JsResult), Blob(webcore::blob::Any), ArrayBuffer(jsc::array_buffer::ArrayBufferStrong), Memfd(Fd), @@ -301,9 +303,7 @@ impl Stdio { Self::Memfd(fd) => SpawnOptionsStdio::Pipe(*fd), #[cfg(windows)] Self::Memfd(_) => panic!("This should never happen"), - Self::Path(pathlike) => { - SpawnOptionsStdio::Path(pathlike.slice().to_vec().into_boxed_slice()) - } + Self::Path(path) => SpawnOptionsStdio::Path(core::mem::take(path)), Self::Inherit => SpawnOptionsStdio::Inherit, Self::Ignore => SpawnOptionsStdio::Ignore, }; @@ -632,7 +632,7 @@ impl Stdio { return Ok(()); } PathOrFileDescriptor::Path(ref path) => { - *self = Stdio::Path(path.clone()); + *self = Stdio::Path(Box::from(path.slice())); return Ok(()); } } diff --git a/src/runtime/node/node_fs.rs b/src/runtime/node/node_fs.rs index dd6773bdc442..50c199e64ef2 100644 --- a/src/runtime/node/node_fs.rs +++ b/src/runtime/node/node_fs.rs @@ -2716,7 +2716,7 @@ pub mod args { let fd = FD::from_js_required(ctx, arguments)?; let buffers = VectorArrayBuffer::from_js( ctx, - arguments.protect_eat_next().ok_or_else(|| { + arguments.next_eat().ok_or_else(|| { ctx.throw_invalid_arguments(format_args!("Expected an ArrayBufferView[]")) })?, // The iovec pointers outlive this call on the async path; root diff --git a/src/runtime/node/node_fs_binding.rs b/src/runtime/node/node_fs_binding.rs index 99415ff3c126..ef399883d0b2 100644 --- a/src/runtime/node/node_fs_binding.rs +++ b/src/runtime/node/node_fs_binding.rs @@ -1,4 +1,3 @@ -use core::mem::ManuallyDrop; use core::ptr::NonNull; use bun_jsc::call_frame::ArgumentsSlice; @@ -35,7 +34,6 @@ where // for the duration of argument parsing on the JS thread. let vm: &VirtualMachine = global.bun_vm(); let mut slice = ArgumentsSlice::init(vm, frame.arguments()); - // `defer slice.deinit()` → `Drop for ArgumentsSlice`. // `defer if (@hasDecl(Arguments, "deinit")) args.deinit()` → `Drop for A` // (every `args::*` field type — `PathLike`, `StringOrBuffer`, `Vec`, … — @@ -73,30 +71,13 @@ fn run_async( ) -> JsResult { // SAFETY: JS-thread borrow of the per-thread VM; outlives `slice`. let vm: &mut VirtualMachine = global.bun_vm().as_mut(); - let mut slice = ManuallyDrop::new(ArgumentsSlice::init(vm, frame.arguments())); + let mut slice = ArgumentsSlice::init(vm, frame.arguments()); slice.will_be_async = true; - // `ManuallyDrop` keeps `slice` alive past return when ownership transfers - // to the Task: dropped only on the early-return - // error/abort branches; on the success path the Task owns `args` (whose - // protected JSValues are released by `Drop for ThreadSafe` when the - // Task completes), and `slice` is intentionally not dropped — its - // `Drop`-unprotect would race that. - - let mut args = match ::from_js(global, &mut slice) { - Ok(a) => a, - Err(err) => { - // SAFETY: not yet dropped; only drop site for this path. - unsafe { ManuallyDrop::drop(&mut slice) }; - return Err(err); - } - }; + let mut args = ::from_js(global, &mut slice)?; if global.has_exception() { args.unprotect(); - drop(args); - // SAFETY: not yet dropped; only drop site for this path. - unsafe { ManuallyDrop::drop(&mut slice) }; return Ok(JSValue::ZERO); } @@ -109,9 +90,6 @@ fn run_async( abort_error, ); args.unprotect(); - drop(args); - // SAFETY: not yet dropped; only drop site for this path. - unsafe { ManuallyDrop::drop(&mut slice) }; return Ok(promise); } } @@ -181,27 +159,16 @@ impl Binding { // ── Hand-written bindings for ops outside `NodeFSFunctionEnum` ──────── - /// `callAsync(.cp)` — `AsyncCpTask::create` copies its paths via - /// `to_thread_safe()`, so the arena is dropped with `slice`. + /// `callAsync(.cp)`. pub(crate) fn cp(this: &Self, global: &JSGlobalObject, frame: &CallFrame) -> JsResult { // SAFETY: JS-thread borrow of the per-thread VM; outlives `slice`. let vm: &mut VirtualMachine = global.bun_vm().as_mut(); - let mut slice = ManuallyDrop::new(ArgumentsSlice::init(vm, frame.arguments())); + let mut slice = ArgumentsSlice::init(vm, frame.arguments()); slice.will_be_async = true; - let cp_args = match args::Cp::from_js(global, &mut slice) { - Ok(a) => a, - Err(err) => { - // SAFETY: not yet dropped; only drop site for this path. - unsafe { ManuallyDrop::drop(&mut slice) }; - return Err(err); - } - }; + let cp_args = args::Cp::from_js(global, &mut slice)?; if global.has_exception() { - drop(cp_args); - // SAFETY: not yet dropped; only drop site for this path. - unsafe { ManuallyDrop::drop(&mut slice) }; return Ok(JSValue::ZERO); } @@ -243,22 +210,12 @@ impl Binding { ) -> JsResult { // SAFETY: JS-thread borrow of the per-thread VM; outlives `slice`. let vm: &mut VirtualMachine = global.bun_vm().as_mut(); - let mut slice = ManuallyDrop::new(ArgumentsSlice::init(vm, frame.arguments())); + let mut slice = ArgumentsSlice::init(vm, frame.arguments()); slice.will_be_async = true; - let rd_args = match args::Readdir::from_js(global, &mut slice) { - Ok(a) => a, - Err(err) => { - // SAFETY: not yet dropped; only drop site for this path. - unsafe { ManuallyDrop::drop(&mut slice) }; - return Err(err); - } - }; + let rd_args = args::Readdir::from_js(global, &mut slice)?; if global.has_exception() { - drop(rd_args); - // SAFETY: not yet dropped; only drop site for this path. - unsafe { ManuallyDrop::drop(&mut slice) }; return Ok(JSValue::ZERO); } diff --git a/src/runtime/node/types.rs b/src/runtime/node/types.rs index 465c7c6f07b1..2d655dbea20d 100644 --- a/src/runtime/node/types.rs +++ b/src/runtime/node/types.rs @@ -1191,7 +1191,7 @@ impl PathLikeExt for PathLike { return Err(err); } - arguments.protect_eat(); + arguments.eat(); Ok(Some(Self::Buffer(buffer))) } @@ -1208,7 +1208,7 @@ impl PathLikeExt for PathLike { return Err(err); } - arguments.protect_eat(); + arguments.eat(); Ok(Some(Self::Buffer(buffer))) } diff --git a/src/runtime/webcore/Blob.rs b/src/runtime/webcore/Blob.rs index 8c2a9d0eef54..4b577728cb22 100644 --- a/src/runtime/webcore/Blob.rs +++ b/src/runtime/webcore/Blob.rs @@ -3649,7 +3649,6 @@ impl BlobExt for Blob { } } - path_or_fd.to_thread_safe(); core::mem::replace( path_or_fd, PathOrFileDescriptor::Path(crate::webcore::node_types::PathLike::String( @@ -6868,13 +6867,24 @@ pub trait FileOpener: Sized { #[cfg(windows)] fn open_callback(&self) -> fn(&mut Self, Fd); + /// The store's path bytes. `get_fd_by_opening` runs on a pool thread, where + /// the store's `PathLike` may be read but not cloned or dropped (a + /// `PinnedBuffer`'s refcount is JS-thread-only), so it copies these into + /// its `PathBuffer` instead of cloning. + fn store_path(&self) -> &[u8] { + match self.pathlike() { + PathOrFileDescriptor::Path(p) => p.slice(), + PathOrFileDescriptor::Fd(_) => unreachable!(), + } + } + fn get_fd_by_opening(&mut self, callback: fn(&mut Self, Fd)) { let mut buf = bun_paths::PathBuffer::uninit(); - let path_string = match self.pathlike() { - PathOrFileDescriptor::Path(p) => p.clone(), + let len = match self.pathlike() { + PathOrFileDescriptor::Path(p) => p.slice_z_with_force_copy::(&mut buf).len(), PathOrFileDescriptor::Fd(_) => unreachable!(), }; - let path = path_string.slice_z(&mut buf); + let path = bun_core::ZStr::from_buf(&buf[..], len); #[cfg(windows)] { @@ -6892,17 +6902,11 @@ pub trait FileOpener: Sized { // SAFETY: req is the live uv_fs_t from the open request. let result = unsafe { (*req).result }; if let Some(err_enum) = result.err_enum_e() { - let path_string_2 = match self_.pathlike() { - PathOrFileDescriptor::Path(p) => p.clone(), - PathOrFileDescriptor::Fd(_) => unreachable!(), - }; + let system_error = bun_sys::Error::from_code(err_enum, bun_sys::Tag::open) + .with_path(self_.store_path()) + .to_system_error(); self_.set_errno(bun_errno::from_errno(err_enum as i32).into()); - self_.set_system_error( - bun_sys::Error::from_code(err_enum, bun_sys::Tag::open) - .with_path(path_string_2.slice()) - .to_system_error() - .into(), - ); + self_.set_system_error(system_error.into()); self_.set_opened_fd(bun_sys::Fd::INVALID); } else { self_.set_opened_fd(Fd::from_uv(result.to_fd())); @@ -6941,13 +6945,11 @@ pub trait FileOpener: Sized { ) }; if let Some(errno) = rc.err_enum_e() { + let system_error = bun_sys::Error::from_code(errno, bun_sys::Tag::open) + .with_path(self.store_path()) + .to_system_error(); self.set_errno(bun_errno::from_errno(errno as i32).into()); - self.set_system_error( - bun_sys::Error::from_code(errno, bun_sys::Tag::open) - .with_path(path_string.slice()) - .to_system_error() - .into(), - ); + self.set_system_error(system_error.into()); self.set_opened_fd(bun_sys::Fd::INVALID); // `callback` may free `self` (see comment above) — must be the // last thing we touch on this path. @@ -6971,7 +6973,7 @@ pub trait FileOpener: Sized { } bun_sys::Result::Err(err) => { if err.get_errno() == bun_sys::E::ENOENT { - match self.try_mkdirp(err.clone(), path, path_string.slice()) { + match self.try_mkdirp(err.clone(), path, path.as_bytes()) { Retry::Continue => continue, Retry::Fail => { // `mkdir_if_not_exists` already populated @@ -6982,10 +6984,9 @@ pub trait FileOpener: Sized { Retry::No => {} } } + let err = err.with_path(self.store_path()); self.set_errno(bun_errno::from_errno(err.errno as i32).into()); - self.set_system_error(jsc::SysErrorJsc::to_system_error( - &err.with_path(path_string.slice()), - )); + self.set_system_error(jsc::SysErrorJsc::to_system_error(&err)); self.set_opened_fd(Fd::INVALID); break; } diff --git a/src/runtime/webcore/blob/Store.rs b/src/runtime/webcore/blob/Store.rs index f34c88640bcb..c8cdbebab764 100644 --- a/src/runtime/webcore/blob/Store.rs +++ b/src/runtime/webcore/blob/Store.rs @@ -128,7 +128,6 @@ impl StoreExt for Store { credentials: S3Credentials, ) -> Result, crate::Error> { let mut path = pathlike; - // this actually protects/refs the pathlike path.to_thread_safe(); // Compute the extension-derived fallback before moving `path` into the @@ -144,9 +143,13 @@ impl StoreExt for Store { } fn init_file( - pathlike: PathOrFileDescriptor, + mut pathlike: PathOrFileDescriptor, mime_type: Option, ) -> Result, crate::Error> { + // A Store is read from other threads and dropped from the Blob cell's + // GC finalizer, so it must not hold a call-scoped `PathLike::Buffer`. + pathlike.to_thread_safe(); + // Compute the extension-derived fallback before moving `pathlike` into // the Store so we don't need to clone the owned PathOrFileDescriptor. let mime_type = mime_type.or_else(|| match &pathlike { diff --git a/src/runtime/webcore/blob/copy_file.rs b/src/runtime/webcore/blob/copy_file.rs index 4c6ba47e57d7..c2fca0e50fe6 100644 --- a/src/runtime/webcore/blob/copy_file.rs +++ b/src/runtime/webcore/blob/copy_file.rs @@ -145,9 +145,10 @@ impl CopyFile { let instance = jsc::SystemError::from(system_error) .to_error_instance_with_async_stack(global_this, promise); - if let Some(store) = self.store.take() { - drop(store); // deref() - } + // Only now: `source_file_store.pathlike` (read above) may borrow from + // the source store. + drop(self.source_store.take()); + drop(self.store.take()); promise.reject(global_this, Ok(instance)) } @@ -156,11 +157,10 @@ impl CopyFile { promise: &mut JSPromise, global_this: &JSGlobalObject, ) -> jsc::JsResult<()> { - drop(self.source_store.take()); // source_store.?.deref() - if self.system_error.is_some() { return self.reject(promise, global_this); } + drop(self.source_store.take()); promise.resolve( global_this, @@ -1015,8 +1015,9 @@ fn read_write_loop_capped( // `source_file_store.pathlike` is a `PathLike` clone that is independently // droppable — `PathLike::clone` dupes owned string buffers (freed by the // clone's own `CowSlice` drop), bumps refs for WTF-backed slices, and only -// shares the backing for borrowed-string/Buffer variants (whose owner is kept -// alive by the `source_store` `StoreRef`). Each clone's field `Drop` frees +// shares the backing for borrowed-string/Buffer/PinnedBuffer variants (whose +// owner is kept alive by the `source_store` `StoreRef`, so that is dropped only +// after the last read of the path — see `reject`). Each clone's field `Drop` frees // exactly what it owns; the `StoreRef`s release just their Store refcounts on // drop. No explicit `Drop` impl is needed. diff --git a/test/js/bun/spawn/spawn.test.ts b/test/js/bun/spawn/spawn.test.ts index 327683631745..7daf008e3034 100644 --- a/test/js/bun/spawn/spawn.test.ts +++ b/test/js/bun/spawn/spawn.test.ts @@ -321,6 +321,24 @@ for (let [gcTick, label] of [ expect(await readableStreamToText(stdout!)).toBe("hello there!"); }); + it("Bun.file(Buffer) wrapped in a Response works as stdin", async () => { + // The Response body's file blob is moved out during option parsing, so + // spawn is the last holder of that store: the stdio option must own the + // path bytes rather than borrow them from the store's pinned Buffer. + const stdinPath = join(tmpdirSync(), "stdin.txt"); + writeFileSync(stdinPath, "hello there!"); + const body = new Response(Bun.file(Buffer.from(stdinPath))); + gcTick(); + Bun.gc(true); + const { stdout } = spawn({ + cmd: [bunExe(), "-e", "process.stdin.pipe(process.stdout)"], + stdout: "pipe", + stdin: body, + }); + gcTick(); + expect(await readableStreamToText(stdout!)).toBe("hello there!"); + }); + it("Bun.file() works as stdin and stdout", async () => { const stdinPath = join(tmpdirSync(), "stdout.txt"); writeFileSync(stdinPath, "hello!"); diff --git a/test/js/bun/util/bun-file.test.ts b/test/js/bun/util/bun-file.test.ts index 6a422f38589f..970eb79cb36f 100644 --- a/test/js/bun/util/bun-file.test.ts +++ b/test/js/bun/util/bun-file.test.ts @@ -110,6 +110,77 @@ test("Bun.file().arrayBuffer() errors include async stack frames", async () => { expect(caught.stack).toContain("at async caller"); }); +test("Bun.file() with a Buffer/Uint8Array path survives GC of the blobs", async () => { + // The file store used to protect() the path's JS buffer and never release + // it, and unpinned it from the Blob's GC destructor (touching a JS cell + // mid-sweep). Now it holds the backing store directly: the blobs and the + // buffers are both collectable, and reads still see the path. + await using dir = tempDir("bun-file-buffer-path-gc", { + "hello.txt": "hello", + "run.js": ` + const { join } = require("path"); + const { heapStats } = require("bun:jsc"); + const existing = join(process.argv[2], "hello.txt"); + // A Buffer over 1000 bytes starts out without an ArrayBuffer behind it. Pad + // with "./" by hand (join() would normalize it away) to just past that, + // staying under macOS's 1024-byte PATH_MAX. Not on Windows: Bun.write hands + // libuv the raw path, and 1000 un-normalized chars is past MAX_PATH there. + const pad = process.platform === "win32" ? "" : Buffer.alloc(Math.ceil((1004 - process.argv[2].length) / 2) * 2, "./").toString(); + const longDir = process.argv[2] + "/" + pad; + const existingLong = longDir + "hello.txt"; + const protectedBefore = heapStats().protectedObjectCount; + const uint8Before = heapStats().objectTypeCounts.Uint8Array ?? 0; + for (let i = 0; i < 2000; i++) { + Bun.file(Buffer.from(join(process.argv[2], "missing-" + i))); + Bun.file(new TextEncoder().encode(join(process.argv[2], "missing-u8-" + i))); + Bun.file(new TextEncoder().encode(join(process.argv[2], "missing-ab-" + i)).buffer); + if (i % 8 === 0) Bun.file(Buffer.from(longDir + "no-" + (i % 10))); + } + const keep = [ + Bun.file(Buffer.from(existing)), + Bun.file(new TextEncoder().encode(existing)), + Bun.file(new TextEncoder().encode(existing).buffer), + Bun.file(Buffer.from(existingLong)), + ]; + Bun.gc(true); + Bun.gc(true); + const stats = heapStats(); + // The threadpool open/copy paths read the store's path off the JS thread. + await Bun.write(Bun.file(Buffer.from(longDir + "sub/out.txt")), Buffer.alloc(300 * 1024, "x").toString()); + await Bun.write(Bun.file(new TextEncoder().encode(longDir + "copy.txt")), keep[3]); + console.log(JSON.stringify({ + longPathBytes: pad === "" || (Buffer.from(existingLong).length > 1000 && Buffer.from(existingLong).length < 1024), + exists: await Promise.all(keep.map(f => f.exists())), + text: await Promise.all(keep.map(f => f.text())), + written: [(await Bun.file(longDir + "sub/out.txt").text()).length, await Bun.file(longDir + "copy.txt").text()], + missing: await Bun.file(Buffer.from(join(process.argv[2], "missing-0"))).exists(), + leakedProtects: stats.protectedObjectCount - protectedBefore > 100, + leakedBuffers: (stats.objectTypeCounts.Uint8Array ?? 0) - uint8Before > 100, + })); + `, + }); + + await using proc = Bun.spawn({ + cmd: [bunExe(), join(dir, "run.js"), dir], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + + expect(stderr).toBe(""); + expect(JSON.parse(stdout)).toEqual({ + longPathBytes: true, + exists: [true, true, true, true], + text: ["hello", "hello", "hello", "hello"], + written: [300 * 1024, "hello"], + missing: false, + leakedProtects: false, + leakedBuffers: false, + }); + expect(exitCode).toBe(0); +}); + test("Bun.file().json() with UTF-8 BOM does not free an interior pointer", async () => { // When a file starts with EF BB BF, the BOM is stripped before parsing and // the temporary read buffer is freed. Previously the *post-strip* slice was diff --git a/test/js/node/fs/fs-leak.test.js b/test/js/node/fs/fs-leak.test.js index 228d6d5a95f3..47998d9420dc 100644 --- a/test/js/node/fs/fs-leak.test.js +++ b/test/js/node/fs/fs-leak.test.js @@ -2,6 +2,7 @@ const { expect, test } = require("bun:test"); const fs = require("fs"); const { tmpdir, devNull } = require("os"); +const { bunExe, bunEnv, tempDir } = require("harness"); function getMaxFd() { const dev_null = fs.openSync(devNull, "r"); @@ -125,3 +126,116 @@ test("createReadStream file handle does not leak file descriptors", async () => expect(getMaxFd()).toBe(start); expect(n_bytes).toBe("hello world\n".repeat(1000).length); }); + +// https://github.com/oven-sh/bun/issues/32191 +test("async fs ops with Buffer path arguments do not leak the path argument", async () => { + const N = 64; + const WARM = 16; + const script = ` + const fs = require("node:fs"); + const util = require("node:util"); + const { heapStats } = require("bun:jsc"); + + const N = ${N}; + const WARM = ${WARM}; + + function liveCounts(types) { + Bun.gc(true); + Bun.gc(true); + const counts = heapStats().objectTypeCounts; + return types.map(t => counts[t] ?? 0); + } + + // Returns the worst delta across the observed heap types. + async function measure(types, op) { + if (!Array.isArray(types)) types = [types]; + for (let i = 0; i < WARM; i++) await op(N + i); + const before = liveCounts(types); + for (let i = 0; i < N; i++) await op(i); + const after = liveCounts(types); + return Math.max(...after.map((v, idx) => v - before[idx])); + } + + // Expect the exact error so a parse-time rejection cannot silently skip + // the async path and make a segment vacuous. + const expectEnoent = e => { + if (e.code !== "ENOENT") throw e; + }; + const expectAborted = e => { + if (e.name !== "AbortError") throw e; + }; + + const deltas = {}; + deltas.accessBufferPath = await measure("Uint8Array", i => + fs.promises.access(Buffer.from("missing-" + i)).catch(expectEnoent), + ); + deltas.accessArrayBufferPath = await measure("ArrayBuffer", i => { + const bytes = Buffer.from("missing-" + i); + const path = new ArrayBuffer(bytes.length); + new Uint8Array(path).set(bytes); + return fs.promises.access(path).catch(expectEnoent); + }); + deltas.writeFileBufferPath = await measure("Uint8Array", i => + fs.promises.writeFile(Buffer.from("out-" + (i % 2) + ".txt"), "x"), + ); + deltas.abortedWriteFileBufferPath = await measure("Uint8Array", i => + fs.promises + .writeFile(Buffer.from("out-aborted.txt"), Buffer.from("data-" + i), { signal: AbortSignal.abort() }) + .catch(expectAborted), + ); + // writev/readv buffers take per-element roots at parse and the array root + // at schedule; both must be released at completion, so watch both the + // element buffers and the array wrapper. + const writev = util.promisify(fs.writev); + const readv = util.promisify(fs.readv); + const vfd = fs.openSync("vec.txt", "w+"); + deltas.writevBuffers = await measure(["Uint8Array", "Array"], i => + writev(vfd, [Buffer.from("vec-a-" + i), Buffer.from("vec-b-" + i)], 0), + ); + deltas.readvBuffers = await measure(["Uint8Array", "Array"], i => + readv(vfd, [Buffer.alloc(8), Buffer.alloc(8)], 0), + ); + fs.closeSync(vfd); + // readdir is a separate hand-written binding (Binding::readdir) from the + // generic run_async path above, so exercise its Buffer-path root too. + // (cp is the other hand-written binding, but its JS wrapper rejects + // non-string paths, so a Buffer path never reaches the native binding.) + deltas.readdirBufferPath = await measure("Uint8Array", i => fs.promises.readdir(Buffer.from("."))); + if (!fs.existsSync("out-0.txt") || !fs.existsSync("out-1.txt")) { + throw new Error("writeFile segment did not write its files"); + } + console.log(JSON.stringify(deltas)); + `; + + using dir = tempDir("fs-buffer-path-leak", {}); + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", script], + env: bunEnv, + cwd: String(dir), + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + + let deltas; + try { + deltas = JSON.parse(stdout.trim()); + } catch { + throw new Error(`fixture did not produce JSON (exit ${exitCode}):\nstdout: ${stdout}\nstderr: ${stderr}`); + } + + // Before the fix, every async call with a Buffer/ArrayBuffer path argument + // left the argument permanently gcProtect'ed, so each delta equaled N. + const verdict = Object.fromEntries( + Object.entries(deltas).map(([op, delta]) => [op, delta < N / 4 ? "ok" : `leaked ${delta} objects over ${N} calls`]), + ); + expect(verdict).toEqual({ + accessBufferPath: "ok", + accessArrayBufferPath: "ok", + writeFileBufferPath: "ok", + abortedWriteFileBufferPath: "ok", + writevBuffers: "ok", + readvBuffers: "ok", + readdirBufferPath: "ok", + }); + expect(exitCode).toBe(0); +}); diff --git a/test/js/web/workers/worker_blob.test.ts b/test/js/web/workers/worker_blob.test.ts index aaf58ae0ea3e..25b1ace02a69 100644 --- a/test/js/web/workers/worker_blob.test.ts +++ b/test/js/web/workers/worker_blob.test.ts @@ -1,4 +1,6 @@ import { expect, test } from "bun:test"; +import { bunEnv, bunExe, tempDir } from "harness"; +import { join } from "path"; test("Worker from a Blob", async () => { const worker = new Worker( @@ -124,3 +126,60 @@ test("Worker on a revoked blob still works", async () => { expect(revoked).toBe("revoked."); }); + +test("object URLs for Bun.file(Buffer) blobs can be resolved and revoked from a Worker", async () => { + // The registry shares one file store across VMs. Its Buffer path is held by a + // ref on the main VM's ArrayBuffer, so the worker letting go of the last + // reference must hand the release back to the main thread rather than touch + // that refcount itself — in both directions (main-owned released by worker, + // worker-owned released by main after the worker is gone). + using dir = tempDir("worker-objecturl-buffer-path", { + "data.txt": "hello", + "worker.js": ` + const mine = []; + for (let i = 0; i < 20; i++) mine.push(URL.createObjectURL(Bun.file(Buffer.from(process.argv[2] + "/data.txt")))); + self.onmessage = async e => { + let ok = 0; + for (const u of e.data) { + if ((await (await fetch(u)).text()) === "hello") ok++; + URL.revokeObjectURL(u); + } + Bun.gc(true); + postMessage({ ok, mine }); + }; + `, + "main.js": ` + const dir = process.argv[2]; + const urls = []; + for (let i = 0; i < 50; i++) { + urls.push(URL.createObjectURL(Bun.file(Buffer.from(dir + "/data.txt")))); + urls.push(URL.createObjectURL(Bun.file(new TextEncoder().encode(dir + "/data.txt").buffer))); + } + Bun.gc(true); + const w = new Worker(dir + "/worker.js", { argv: [dir] }); + w.postMessage(urls); + const { ok, mine } = await new Promise((resolve, reject) => { + w.onmessage = e => resolve(e.data); + w.onerror = e => reject(e.error ?? new Error(e.message)); + }); + const fromWorker = []; + for (const u of mine) fromWorker.push(await (await fetch(u)).text()); + await w.terminate(); + Bun.gc(true); + Bun.gc(true); + const stillResolves = await fetch(urls[0]).then(() => true, () => false); + console.log(JSON.stringify({ ok, fromWorker: fromWorker.every(t => t === "hello") && fromWorker.length, stillResolves })); + `, + }); + + await using proc = Bun.spawn({ + cmd: [bunExe(), join(String(dir), "main.js"), String(dir)], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(JSON.parse(stdout)).toEqual({ ok: 100, fromWorker: 20, stillResolves: false }); + expect(exitCode).toBe(0); +});