Skip to content
22 changes: 9 additions & 13 deletions src/bundler/bundle_v2.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1458,21 +1458,22 @@ pub mod bv2_impl {
/// installed after the last fallible step in `BundleV2::init`, so the
/// box is never dropped while the watcher holds a pointer to it.
fn __bun_jsc_enable_hot_module_reloading_for_bundler(
bv2: core::ptr::NonNull<super::BundleV2<'static>>,
bv2: &'static mut super::BundleV2<'static>,
);
}

/// `Watcher.enableHotModuleReloading(this, null)` for `bun build --watch`.
#[inline]
pub(crate) fn enable_hot_module_reloading_for_bundler(bv2: *mut super::BundleV2<'_>) {
let bv2 = core::ptr::NonNull::new(bv2.cast::<super::BundleV2<'static>>())
.expect("BundleV2 watcher: bv2 is non-null");
// SAFETY: link-time-resolved Rust-ABI fn in `bun_jsc::hot_reloader`.
// Not `safe fn`: the callee dereferences `bv2`, so it must point to a
// live `BundleV2` whose backing allocation outlives the watcher (sole
// caller is `BundleV2::init`; the box is leaked on the success path —
// see the watch-mode caveat above).
unsafe { __bun_jsc_enable_hot_module_reloading_for_bundler(bv2) }
// `bv2` is the live boxed `BundleV2` from `BundleV2::init`, leaked on
// the success path (see the watch-mode caveat above), so widening it
// to `'static` for the process-lifetime watcher holds.
unsafe {
__bun_jsc_enable_hot_module_reloading_for_bundler(
&mut *bv2.cast::<super::BundleV2<'static>>(),
)
}
}

/// Bytecode generation entry point for the linker: marks the calling
Expand Down Expand Up @@ -7632,11 +7633,6 @@ pub mod bv2_impl {
this.on_after_decrement_scan_counter();
}
}

/// To satisfy the interface from NewHotReloader()
pub fn bust_dir_cache(&mut self, path: &[u8]) -> bool {
self.transpiler.resolver.bust_dir_cache(path)
}
}

pub use bun_core::cheap_prefix_normalizer;
Expand Down
10 changes: 5 additions & 5 deletions src/codegen/generate-host-exports.ts
Original file line number Diff line number Diff line change
Expand Up @@ -467,13 +467,13 @@ ${emitNoMangle(e.abi, e.symbol, "g: &JSGlobalObject", "JSValue", body)}`;
case "rust": {
// `extern "Rust"` link-time hook: forward the safe signature verbatim
// (no pointer rewriting; `extern "Rust"` ABI == native Rust ABI).
// Reference params/returns are rejected — `extern "Rust" fn` items need
// explicit lifetimes for borrowed types and the generator does not
// synthesise them. Use raw pointers in the impl signature instead.
// Reference params need an explicit lifetime (`extern "Rust" fn` items
// cannot elide them and the generator does not synthesise one);
// reference returns are rejected.
for (const p of e.params)
if (p.ty.startsWith("&"))
if (p.ty.startsWith("&") && !/^&'/.test(p.ty))
throw new Error(
`${loc}: HOST_EXPORT(${e.symbol}, rust) param \`${p.raw}\` is a reference; use a raw pointer`,
`${loc}: HOST_EXPORT(${e.symbol}, rust) param \`${p.raw}\` is a reference without an explicit lifetime`,
);
if (e.ret.startsWith("&"))
throw new Error(
Expand Down
219 changes: 193 additions & 26 deletions src/event_loop/ConcurrentTask.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,8 @@
//!
//! The task is run on a thread pool and then the result is returned to the main JavaScript thread.
//!
//! If `auto_delete` is true, the task is automatically deallocated when it's finished.
//! Otherwise, it's expected that the containing struct will deallocate the task.
//! A heap carrier (`create*`) is deallocated by the event loop once its task is
//! dispatched; an intrusive one is part of the struct that owns it.

use crate::ManagedTask;
use bun_threading::UnboundedQueue;
Expand Down Expand Up @@ -164,6 +164,58 @@ pub trait Taskable {
unsafe fn release_unrun(this: *mut Self);
}

/// A task whose queued pointer is a [`ThisPtr`](bun_ptr::ThisPtr) to
/// `Target`, which keeps itself alive for the task; the dispatcher runs it or
/// releases it through here. A zero-sized hop type per tag, declared next to
/// `Target` with its own `unsafe impl` stating the ref protocol.
///
/// # Safety
/// `TAG` is dispatched (in `bun_runtime::dispatch`) to this impl and no other
/// task type uses it; and `Target`'s protocol keeps the pointee of every
/// [`task`](Self::task) it queues alive until `run` / `release_unrun` (a
/// `RefPtr` slot held for the queued task, or an owner that outlives the queue).
pub unsafe trait TaskHop {
type Target;
/// The tag constant from [`task_tag`]; the `bun_runtime::dispatch` match
/// arms MUST agree.
const TAG: TaskTag;
fn run(this: bun_ptr::ThisPtr<Self::Target>) -> crate::JsResult<()>;
/// As [`Taskable::release_unrun`].
fn release_unrun(this: bun_ptr::ThisPtr<Self::Target>);

#[inline]
fn task(this: bun_ptr::ThisPtr<Self::Target>) -> Task {
Task::new(Self::TAG, this.as_ptr().cast::<()>())
}
}

/// A task that owns its payload: queued as the `Box<Self>` leaked into
/// [`Task::ptr`], handed back as that box when the dispatcher runs or
/// releases it.
///
/// # Safety
/// `TAG` is dispatched (in `bun_runtime::dispatch`) to this impl and no other
/// task type uses it, so the dispatcher's `Box::<Self>::from_raw` gets back
/// what [`into_task`](Self::into_task) leaked.
pub unsafe trait BoxedTask: Sized {
/// The tag constant from [`task_tag`]; the `bun_runtime::dispatch` match
/// arms MUST agree.
const TAG: TaskTag;
fn run(self: Box<Self>) -> crate::JsResult<()>;
/// As [`Taskable::release_unrun`].
fn release_unrun(self: Box<Self>);
/// A weak post was refused: the VM this was for has closed (any thread;
/// its heap may be gone). Free the task without touching that VM.
fn refused(self: Box<Self>);

/// The JS thread's own enqueue (`EventLoop::enqueue_task`), which cannot
/// be refused; off-thread posts go through `ConcurrentTask::create_boxed`.
#[inline]
fn into_task(self: Box<Self>) -> Task {
Task::new(Self::TAG, Box::into_raw(self).cast::<()>())
}
}

impl TaskTag {
/// The tag's identifier, for diagnostics.
pub fn name(self) -> &'static str {
Expand Down Expand Up @@ -205,17 +257,24 @@ impl Taskable for crate::ManagedTask::ManagedTask {
}
// ────────────────────────────────────────────────────────────────────────────

/// How a heap carrier (`ConcurrentTask::create*`) frees itself, and the
/// payload if the carrier owns it, when a weak post is refused.
type ReleaseRefused = unsafe fn(core::ptr::NonNull<ConcurrentTask>);

#[repr(C)]
pub struct ConcurrentTask {
pub task: Task,
/// Intrusive MPSC link for [`Queue`]. Plain `AtomicPtr` so the enqueue hot
/// path (`atomic_store_next`, called once per completed work-pool task via
/// `enqueue_task_concurrent`) is a single release-store — no read-modify-write.
pub next: Link<ConcurrentTask>,
/// If `true`, the task is heap-owned and freed by the event loop after
/// dispatch. Immutable after construction; read only on the consumer thread,
/// so it does not need to share a word with the contended `next` link.
pub auto_delete: bool,
/// `Some` on a heap carrier (`create*`): the event loop frees the carrier
/// after dispatching its task, and a refused weak post frees carrier and
/// owned payload through it ([`ConcurrentTask::release_refused`]). `None`
/// on an intrusive carrier, which belongs to its container. Immutable after
/// construction; read only on the consumer thread, so it does not need to
/// share a word with the contended `next` link.
release_refused: Option<ReleaseRefused>,
}

impl Default for ConcurrentTask {
Expand All @@ -225,12 +284,12 @@ impl Default for ConcurrentTask {
// byte + raw pointer); caller must set a real task before use.
task: unsafe { bun_core::ffi::zeroed_unchecked() },
next: Link::new(),
auto_delete: false,
release_refused: None,
}
}
}

// `auto_delete` is deliberately its own field rather than packed into bit 0
// `release_refused` is deliberately its own field rather than packed into bit 0
// of `next`: `Task` is already two words here (tag is not packed into the
// pointer), so the struct was never 16B, and profiling (build/create-next
// benches) showed
Expand All @@ -241,7 +300,7 @@ impl Default for ConcurrentTask {
const _: () = assert!(
core::mem::size_of::<ConcurrentTask>()
== core::mem::size_of::<Task>() + 2 * core::mem::size_of::<usize>(),
"ConcurrentTask = Task + next ptr + auto_delete (padded)"
"ConcurrentTask = Task + next ptr + release_refused"
);

// SAFETY: `link()` always projects to the same embedded `next: Link<Self>`
Expand Down Expand Up @@ -272,11 +331,55 @@ impl ConcurrentTask {
bun_core::heap::into_raw(Box::new(init))
}

/// A heap carrier for `task`, whose payload the carrier does not own
/// (intrusive, or freed by whoever posts it) — except a `ManagedTask`,
/// which it does.
pub fn create(task: Task) -> core::ptr::NonNull<ConcurrentTask> {
/// # Safety
/// `this` is a refused heap carrier.
unsafe fn carrier_only(this: core::ptr::NonNull<ConcurrentTask>) {
// SAFETY: fn contract; `create*` boxed it.
drop(unsafe { bun_core::heap::take(this.as_ptr()) });
}
/// # Safety
/// `this` is a refused heap carrier whose payload is a `ManagedTask`.
unsafe fn managed(this: core::ptr::NonNull<ConcurrentTask>) {
// SAFETY: fn contract; refused ⇒ both boxes are ours.
unsafe {
let task = bun_core::heap::take(this.as_ptr()).task;
crate::ManagedTask::ManagedTask::release(task.ptr.cast());
}
}
let release: ReleaseRefused = if task.tag == task_tag::ManagedTask {
managed
} else {
carrier_only
};
Self::create_with(task, release)
}

/// A heap carrier owning the boxed `task`: a refused weak post hands it to
/// [`BoxedTask::refused`].
pub fn create_boxed<T: BoxedTask>(task: Box<T>) -> core::ptr::NonNull<ConcurrentTask> {
/// # Safety
/// `this` is a refused heap carrier built by `create_boxed::<T>`.
unsafe fn boxed<T: BoxedTask>(this: core::ptr::NonNull<ConcurrentTask>) {
// SAFETY: fn contract; refused ⇒ both boxes are ours, and the
// payload is the `Box<T>` `into_task` leaked.
let task = unsafe {
let task = bun_core::heap::take(this.as_ptr()).task;
Box::from_raw(task.ptr.cast::<T>())
};
T::refused(task);
}
Self::create_with(task.into_task(), boxed::<T>)
}

fn create_with(task: Task, release: ReleaseRefused) -> core::ptr::NonNull<ConcurrentTask> {
let raw = ConcurrentTask::new(ConcurrentTask {
task,
next: Link::new(),
auto_delete: true,
release_refused: Some(release),
});
// SAFETY: `new` heap-allocates via `heap::into_raw` — never null.
unsafe { core::ptr::NonNull::new_unchecked(raw) }
Expand All @@ -302,12 +405,23 @@ impl ConcurrentTask {
of: *mut T,
auto_deinit: AutoDeinit,
) -> &mut ConcurrentTask {
bun_core::mark_binding!();
*self = ConcurrentTask {
task: Task::init(of),
debug_assert!(auto_deinit == AutoDeinit::ManualDeinit);
self.from_task(Task::init(of))
}

/// An intrusive carrier loaded with `task`, ready to post.
pub const fn intrusive(task: Task) -> ConcurrentTask {
ConcurrentTask {
task,
next: Link::new(),
auto_delete: auto_deinit == AutoDeinit::AutoDeinit,
};
release_refused: None,
}
}

/// Load this intrusive carrier with `task`, ready to post.
pub fn from_task(&mut self, task: Task) -> &mut ConcurrentTask {
bun_core::mark_binding!();
*self = Self::intrusive(task);
self
}

Expand All @@ -328,25 +442,78 @@ impl ConcurrentTask {
}

/// A weak poster got `task` back because the target VM has closed: free
/// it if it is a heap task (`create*`); an intrusive one belongs to its
/// container.
/// it if it is a heap carrier (`create*`), with the payload if the carrier
/// owns it; an intrusive one belongs to its container.
///
/// # Safety
/// `task` was just refused and is not queued anywhere.
pub unsafe fn release_refused(task: core::ptr::NonNull<ConcurrentTask>) {
// SAFETY: fn contract.
let inner = unsafe { Self::into_task(task) };
// A callback task (`from_callback`, `ManagedTask::new*`) owns a heap
// `ManagedTask` behind `task.ptr` as well.
if inner.tag == crate::task_tag::ManagedTask {
// SAFETY: as above; refused ⇒ ours.
unsafe { crate::ManagedTask::ManagedTask::release(inner.ptr.cast()) };
// SAFETY: fn contract; `release_refused` was installed by the
// constructor that knows the carrier's and payload's ownership.
unsafe {
if let Some(release) = task.as_ref().release_refused {
release(task);
}
}
}

/// Returns whether this task should be automatically deallocated after execution.
/// Whether this is a heap carrier (`create*`) the event loop frees after
/// dispatching its task.
#[inline]
pub fn auto_delete(&self) -> bool {
self.auto_delete
self.release_refused.is_some()
}
}

#[cfg(test)]
mod tests {
use super::*;
use core::sync::atomic::{AtomicUsize, Ordering};

static REFUSED: AtomicUsize = AtomicUsize::new(0);
static DROPPED: AtomicUsize = AtomicUsize::new(0);

struct Payload {
_buf: Box<[u8; 64]>,
}
impl Drop for Payload {
fn drop(&mut self) {
DROPPED.fetch_add(1, Ordering::SeqCst);
}
}
// SAFETY: test-only; never dispatched, so the borrowed tag cannot collide.
unsafe impl BoxedTask for Payload {
const TAG: TaskTag = task_tag::NapiFinalizerTask;
fn run(self: Box<Self>) -> crate::JsResult<()> {
Ok(())
}
fn release_unrun(self: Box<Self>) {}
fn refused(self: Box<Self>) {
REFUSED.fetch_add(1, Ordering::SeqCst);
}
}

#[test]
fn refused_boxed_carrier_releases_its_payload() {
let carrier = ConcurrentTask::create_boxed(Box::new(Payload {
_buf: Box::new([7; 64]),
}));
// SAFETY: never queued; ours.
unsafe { ConcurrentTask::release_refused(carrier) };
assert_eq!(REFUSED.load(Ordering::SeqCst), 1);
assert_eq!(DROPPED.load(Ordering::SeqCst), 1);

let mut intrusive = ConcurrentTask::intrusive(
Box::new(Payload {
_buf: Box::new([7; 64]),
})
.into_task(),
);
// SAFETY: never queued; an intrusive carrier is left alone.
unsafe { ConcurrentTask::release_refused(core::ptr::NonNull::from(&mut intrusive)) };
assert_eq!(DROPPED.load(Ordering::SeqCst), 1);
// SAFETY: the payload leaked into the intrusive carrier above.
drop(unsafe { Box::from_raw(intrusive.task.ptr.cast::<Payload>()) });
assert_eq!(DROPPED.load(Ordering::SeqCst), 2);
}
}
2 changes: 1 addition & 1 deletion src/event_loop/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ pub mod any_event_loop;
// ─── public surface ─────────────────────────────────────────────────────────

pub type JsResult<T> = core::result::Result<T, bun_core::JsError>;
pub use ConcurrentTask::{Task, TaskTag, Taskable, task_tag};
pub use ConcurrentTask::{BoxedTask, Task, TaskHop, TaskTag, Taskable, task_tag};

// snake_case alias for the file-level-struct module so higher tiers avoid
// the type/module namespace collision on the PascalCase form.
Expand Down
4 changes: 3 additions & 1 deletion src/io/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -478,7 +478,9 @@ pub mod write;
// Byte-level `Write` trait + helpers. Downstream
// crates name these as `bun_io::Write` / `bun_io::FmtAdapter` /
// `bun_io::Result`.
pub use write::{AsFmt, DiscardingWriter, FixedBufferStream, FmtAdapter, IntLe, Result, Write};
pub use write::{
AsFmt, DiscardingWriter, FixedBufferStream, FmtAdapter, IntLe, IoWriterAdapter, Result, Write,
};

pub use max_buf as MaxBuf;
pub use pipes::{Chunk, FileType, ReadState};
Expand Down
Loading
Loading