Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion src/event_loop/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,6 @@ bun_dispatch::link_interface! {
fn exit();
fn enqueue_task(task: Task);
fn enqueue_task_concurrent(task: core::ptr::NonNull<ConcurrentTask::ConcurrentTask>);
fn concurrent_poster_end();
fn env() -> *mut bun_dotenv::Loader;
fn top_level_dir() -> *const [u8];
fn create_null_delimited_env_map() -> Result<bun_dotenv::NullDelimitedEnvMap, bun_core::AllocError>;
Expand Down
14 changes: 10 additions & 4 deletions src/jsc/VirtualMachine.rs
Original file line number Diff line number Diff line change
Expand Up @@ -790,6 +790,13 @@ impl VirtualMachine {
unsafe { &*self.event_loop }
}

/// Refuse all future work-pool completion posts (see `ConcurrentPosterGate`). Both loops:
/// macro mode can have pointed tasks at either.
Comment thread
robobun marked this conversation as resolved.
pub fn close_concurrent_posters(&mut self) {
self.regular_event_loop.close_concurrent_posters();
self.macro_event_loop.close_concurrent_posters();
}

/// Alias for [`Self::event_loop_mut`]. Kept for callers migrated on the
/// `runtime-hostfn-safe` branch; both names funnel into the single audited
/// `unsafe` deref above.
Expand Down Expand Up @@ -1638,10 +1645,9 @@ impl VirtualMachine {
bun_http::shutdown_for_exit();

// Release tasks the HTTP daemon posted before observing `is_shutting_down` (else the
// tasklet ⇄ `Box<AsyncHTTP>` cycle leaks); must precede `destructOnExit`. Wait for
// work-pool fs completions first — they post without a shutdown check, so a post
// landing after the drain leaks.
self.event_loop_mut().wait_for_concurrent_posters();
// tasklet ⇄ `Box<AsyncHTTP>` cycle leaks); must precede `destructOnExit`. Close the
// poster gates first so no work-pool fs completion can post after the drain.
Comment thread
robobun marked this conversation as resolved.
self.close_concurrent_posters();
self.event_loop_mut().release_queued_tasks_for_shutdown();

if let Some(rare) = self.rare_data.as_deref_mut() {
Expand Down
98 changes: 74 additions & 24 deletions src/jsc/event_loop.rs
Original file line number Diff line number Diff line change
Expand Up @@ -88,10 +88,10 @@ pub struct EventLoop {

pub entered_event_loop_count: isize,
pub concurrent_ref: AtomicI32,
/// Work-pool tasks that will post to `concurrent_tasks` when they finish (counted on the JS
/// thread before hand-off, decremented on the pool thread after the post). Shutdown waits for
/// zero before the final queue drain so a completion can't land after it and leak.
pub concurrent_posters: AtomicU32,
/// Shared with every scheduled work-pool task whose completion posts back through
/// [`Self::enqueue_task_concurrent`]; see [`ConcurrentPosterGate`]. `None` only after
/// [`Self::deinit`].
Comment thread
robobun marked this conversation as resolved.
pub concurrent_poster_gate: Option<std::sync::Arc<ConcurrentPosterGate>>,
/// Atomic nullable pointer to the next-due `WTFTimer`.
///
/// Note (§Dispatch): payload is `*mut ()` — the real
Expand Down Expand Up @@ -132,7 +132,7 @@ impl Default for EventLoop {
uws_loop: (),
entered_event_loop_count: 0,
concurrent_ref: AtomicI32::new(0),
concurrent_posters: AtomicU32::new(0),
concurrent_poster_gate: Some(std::sync::Arc::new(ConcurrentPosterGate::default())),
imminent_gc_timer: AtomicPtr::new(core::ptr::null_mut()),
#[cfg(unix)]
signal_handler: None,
Expand All @@ -142,6 +142,59 @@ impl Default for EventLoop {
}
}

/// Gate between work-pool completion posts and event-loop teardown. Posters bracket only the
/// enqueue itself in [`Self::begin_post`]/[`Self::end_post`] — never the fs syscall, which can
/// block forever (a `read()` from a FIFO with no writer) — so shutdown's [`Self::close`] waits
/// only for posts already mid-enqueue and refuses everything later. `Arc`-shared with every
/// in-flight task so a poster whose syscall outlives the VM still reads the closed state from
/// live memory; a refused poster frees its own task without touching the VM or the JS heap
/// (`dispose_without_post` in `node_fs.rs`).
Comment thread
robobun marked this conversation as resolved.
#[derive(Default)]
pub struct ConcurrentPosterGate {
/// Bit 31: closed. Bits 0..31: posters currently between `begin_post` and `end_post`.
state: AtomicU32,
}

impl ConcurrentPosterGate {
const CLOSED: u32 = 1 << 31;

/// Work-pool thread: begin a completion post. On `true`, the loop is guaranteed live until
/// the matching [`Self::end_post`] ([`Self::close`] cannot return in between).
Comment thread
robobun marked this conversation as resolved.
#[must_use]
pub fn begin_post(&self) -> bool {
let mut state = self.state.load(Ordering::Relaxed);
loop {
if state & Self::CLOSED != 0 {
return false;
}
match self.state.compare_exchange_weak(
state,
state + 1,
Ordering::Acquire,
Ordering::Relaxed,
) {
Ok(_) => return true,
Err(actual) => state = actual,
}
}
}

/// Work-pool thread: the post finished. Last touch of the event loop.
pub fn end_post(&self) {
let prev = self.state.fetch_sub(1, Ordering::Release);
debug_assert!(prev & !Self::CLOSED > 0);
}

/// JS thread, shutdown: after this returns no further post can land. Bounded by an enqueue,
/// not by the underlying fs operation.
Comment thread
robobun marked this conversation as resolved.
pub fn close(&self) {
self.state.fetch_or(Self::CLOSED, Ordering::AcqRel);
while self.state.load(Ordering::Acquire) & !Self::CLOSED != 0 {
std::thread::yield_now();
}
}
}

mod drain_result {
pub(super) const SUCCESS: u8 = 0;
pub(super) const JS_TERMINATED: u8 = 1;
Expand Down Expand Up @@ -770,6 +823,9 @@ impl EventLoop {
}

pub fn deinit(&mut self) {
// The VM box is raw-dealloc'd without field `Drop`s, so release the gate `Arc` here;
// stranded posters hold their own clones.
Comment thread
robobun marked this conversation as resolved.
drop(self.concurrent_poster_gate.take());
// Free (don't run — running could re-enter the dying VM) queued
// ManagedTask boxes. Other tags are left in place: they were re-queued
// by `release_queued_tasks_for_shutdown` because their callback can't
Expand Down Expand Up @@ -1010,26 +1066,21 @@ impl EventLoop {
self.wakeup();
}

/// See `concurrent_posters`. Call on the JS thread before scheduling a
/// work-pool task whose completion posts via `enqueue_task_concurrent`.
pub fn concurrent_poster_begin(&self) {
let _ = self.concurrent_posters.fetch_add(1, Ordering::SeqCst);
}

/// Pool-thread pair of [`Self::concurrent_poster_begin`]. Must be the
/// poster's last touch of this event loop: once the count hits zero the
/// shutdown thread may drain the queue and tear the VM down.
pub fn concurrent_poster_end(&self) {
let prev = self.concurrent_posters.fetch_sub(1, Ordering::Release);
debug_assert!(prev > 0);
/// Clone of [`Self::concurrent_poster_gate`]. Panics after [`Self::deinit`] — tasks are
/// only created while the VM is live.
Comment thread
robobun marked this conversation as resolved.
pub fn poster_gate(&self) -> std::sync::Arc<ConcurrentPosterGate> {
std::sync::Arc::clone(
self.concurrent_poster_gate
.as_ref()
.expect("poster_gate() after EventLoop::deinit()"),
)
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.

/// Spin until every counted poster has finished its post. Called on the JS thread during
/// shutdown, after the last JS has run and before the final queue drain. Each pending fs
/// operation is finite, so the wait is bounded by syscall latency.
pub fn wait_for_concurrent_posters(&self) {
while self.concurrent_posters.load(Ordering::Acquire) > 0 {
std::thread::yield_now();
/// Close [`Self::concurrent_poster_gate`]. Shutdown calls this before the final queue drain
/// so the drain sees every post that will ever land.
Comment thread
robobun marked this conversation as resolved.
pub fn close_concurrent_posters(&self) {
if let Some(gate) = self.concurrent_poster_gate.as_ref() {
gate.close();
}
}

Expand Down Expand Up @@ -1343,7 +1394,6 @@ bun_event_loop::link_impl_JsEventLoop! {
exit() => (*this).exit(),
enqueue_task(task) => (*this).enqueue_task(task),
enqueue_task_concurrent(task) => (*this).enqueue_task_concurrent(task),
concurrent_poster_end() => (*this).concurrent_poster_end(),
env() => (*this).vm_ref().transpiler.env,
top_level_dir() => core::ptr::from_ref::<[u8]>((*this).vm_ref().top_level_dir()),
create_null_delimited_env_map() =>
Expand Down
9 changes: 5 additions & 4 deletions src/jsc/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1359,10 +1359,11 @@ pub use self::event_loop as EventLoop;
pub mod any_task_job;
pub use self::any_task_job::{AnyTaskJob, AnyTaskJobCtx};
pub use self::event_loop::{
AnyEventLoop, AnyTaskWithExtraContext, ConcurrentCppTask, ConcurrentPromiseTask,
ConcurrentTask, CppTask, DeferredTaskQueue, EventLoopHandle, EventLoopTask, EventLoopTaskPtr,
GarbageCollectionController, JsTerminated, JsTerminatedResult, ManagedTask, MiniEventLoop,
PosixSignalHandle, PosixSignalTask, Task, WorkPool, WorkPoolTask, WorkTask, WorkTaskContext,
AnyEventLoop, AnyTaskWithExtraContext, ConcurrentCppTask, ConcurrentPosterGate,
ConcurrentPromiseTask, ConcurrentTask, CppTask, DeferredTaskQueue, EventLoopHandle,
EventLoopTask, EventLoopTaskPtr, GarbageCollectionController, JsTerminated, JsTerminatedResult,
ManagedTask, MiniEventLoop, PosixSignalHandle, PosixSignalTask, Task, WorkPool, WorkPoolTask,
WorkTask, WorkTaskContext,
};
#[cfg(unix)]
pub type PlatformEventLoop = bun_uws::Loop;
Expand Down
31 changes: 31 additions & 0 deletions src/jsc/node_path.rs
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,12 @@ use crate::array_buffer::MarkedArrayBuffer;
/// [`ThreadSafe<T>`].
pub trait Unprotect {
fn unprotect(&mut self);

/// The VM died before the JS thread could run [`Self::unprotect`] or `Drop`: clear every
/// cleanup that would touch the JS heap or VM-owned state (buffer pins, AbortSignal refs)
/// so the plain `Drop` is safe off the JS thread. Protect counts died with the heap and
/// need no balancing. Required (no default) so every impl accounts for its fields.
Comment thread
robobun marked this conversation as resolved.
fn disarm_for_dead_vm(&mut self);
}

/// RAII guard returned by `into_thread_safe()`: a `T` whose JS-backed buffers
Expand All @@ -51,6 +57,18 @@ impl<T: Unprotect> ThreadSafe<T> {
pub fn adopt(value: T) -> Self {
Self(value)
}

/// Dispose on a work-pool thread after the VM died: skip [`Unprotect::unprotect`], clear
/// the JS-heap-touching `Drop` behavior via [`Unprotect::disarm_for_dead_vm`], then drop
/// the inner `T` to free its owned Rust payloads.
Comment thread
robobun marked this conversation as resolved.
pub fn dispose_for_dead_vm(self) {
let mut this = core::mem::ManuallyDrop::new(self);
this.0.disarm_for_dead_vm();
// SAFETY: `this` is `ManuallyDrop`, so `ThreadSafe::drop` (the
// unprotect) never runs and the inner value is read out exactly once;
// its disarmed `Drop` frees the owned payloads.
drop(unsafe { core::ptr::read(&raw const this.0) });
}
}

impl<T: Unprotect> core::ops::Deref for ThreadSafe<T> {
Expand Down Expand Up @@ -225,6 +243,13 @@ impl Unprotect for PathLike {
b.buffer.value.unprotect();
}
}

fn disarm_for_dead_vm(&mut self) {
// `Drop` unpins a pinned Buffer path — a JS-cell deref the dead heap can't take.
if let Self::Buffer(b) = self {
b.pinned = false;
}
}
}

/// `node.PathOrFileDescriptor`.
Expand Down Expand Up @@ -293,6 +318,12 @@ impl Unprotect for PathOrFileDescriptor {
p.unprotect();
}
}

fn disarm_for_dead_vm(&mut self) {
if let Self::Path(p) = self {
p.disarm_for_dead_vm();
}
}
}

impl core::fmt::Display for PathOrFileDescriptor {
Expand Down
7 changes: 3 additions & 4 deletions src/jsc/web_worker.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1308,10 +1308,9 @@ impl WebWorker {
// teardownJSCVM sets it again.
Bun__JSCTaskScheduler__markShuttingDown(vm.global());
// Reclaim queued CppTasks while JSC is still live (after teardownJSCVM the worker VM is
// dealloc'd-without-Drop so anything still in self.tasks leaks). Work-pool fs completions
// post without a shutdown check; wait for in-flight ones so the drain below sees every
// post.
vm.event_loop_mut().wait_for_concurrent_posters();
// dealloc'd-without-Drop so anything still in self.tasks leaks). Close the poster gates
// first so the drain below sees every work-pool fs completion that will ever land.
Comment thread
robobun marked this conversation as resolved.
vm.close_concurrent_posters();
vm.event_loop_mut().release_queued_tasks_for_shutdown();
if let Some(rare) = vm.rare_data.as_deref_mut() {
rare.release_js_handles();
Expand Down
5 changes: 5 additions & 0 deletions src/runtime/crypto/PBKDF2.rs
Original file line number Diff line number Diff line change
Expand Up @@ -255,6 +255,11 @@ impl bun_jsc::Unprotect for PBKDF2 {
self.password.unprotect();
self.salt.unprotect();
}

fn disarm_for_dead_vm(&mut self) {
self.password.disarm_for_dead_vm();
self.salt.disarm_for_dead_vm();
}
}

pub(crate) struct Pbkdf2Ctx {
Expand Down
5 changes: 5 additions & 0 deletions src/runtime/node/node_crypto_binding.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1126,6 +1126,11 @@ mod _impl {
bun_jsc::Unprotect::unprotect(&mut self.password);
bun_jsc::Unprotect::unprotect(&mut self.salt);
}

fn disarm_for_dead_vm(&mut self) {
bun_jsc::Unprotect::disarm_for_dead_vm(&mut self.password);
bun_jsc::Unprotect::disarm_for_dead_vm(&mut self.salt);
}
}

impl CryptoJobCtx for Scrypt {
Expand Down
Loading
Loading