diff --git a/.github/workflows/miri.yml b/.github/workflows/miri.yml index 1ee1447252a2..0a60cf9553ab 100644 --- a/.github/workflows/miri.yml +++ b/.github/workflows/miri.yml @@ -22,6 +22,7 @@ on: - "src/ptr/**" - "src/resolve_builtins/**" - "src/shell_parser/**" + - "src/threading/**" - "src/wyhash/**" - "scripts/rust-miri.ts" - "Cargo.toml" diff --git a/scripts/rust-miri.ts b/scripts/rust-miri.ts index 6d690e4a82cf..094450ff06f5 100644 --- a/scripts/rust-miri.ts +++ b/scripts/rust-miri.ts @@ -44,6 +44,7 @@ const MIRI_CRATES = [ "bun_ptr", "bun_resolve_builtins", "bun_shell_parser", + "bun_threading", "bun_wyhash", ]; diff --git a/src/bundler/ParseTask.rs b/src/bundler/ParseTask.rs index d9020c420c3a..16d7c88ae8fa 100644 --- a/src/bundler/ParseTask.rs +++ b/src/bundler/ParseTask.rs @@ -2865,7 +2865,8 @@ pub mod parse_worker { // SAFETY: `result` is a valid heap pointer with `task` at the given offset; // ownership transfers to the mini event loop which frees it after `on_complete_mini`. unsafe { - mini.enqueue_task_concurrent_with_extra_ctx::>( + bun_event_loop::MiniEventLoop::MiniEventLoop::enqueue_task_concurrent_with_extra_ctx::>( + &raw const **mini, result, on_complete_mini, offset_of!(Result, task), diff --git a/src/bundler/ServerComponentParseTask.rs b/src/bundler/ServerComponentParseTask.rs index b2f10b316297..e29c215d05fc 100644 --- a/src/bundler/ServerComponentParseTask.rs +++ b/src/bundler/ServerComponentParseTask.rs @@ -141,7 +141,8 @@ fn task_callback_wrap(thread_pool_task: *mut ThreadPoolTask) { // SAFETY: `result` is a freshly Box-leaked `parse_task::Result` (above) and // `offset_of!(parse_task::Result, task)` is the intrusive task field within it. unsafe { - mini.enqueue_task_concurrent_with_extra_ctx::>( + bun_event_loop::MiniEventLoop::MiniEventLoop::enqueue_task_concurrent_with_extra_ctx::>( + &raw const **mini, result, on_complete_mini, offset_of!(parse_task::Result, task), diff --git a/src/bundler/bundle_v2.rs b/src/bundler/bundle_v2.rs index a2a31fd5e974..922b8152b7d5 100644 --- a/src/bundler/bundle_v2.rs +++ b/src/bundler/bundle_v2.rs @@ -4343,7 +4343,8 @@ pub mod bv2_impl { // SAFETY: `load` is a valid &mut for the duration of the enqueue; // the mini loop dispatches `on_load_mini` on the bundler thread. unsafe { - mini.enqueue_task_concurrent_with_extra_ctx::>( + bun_event_loop::MiniEventLoop::MiniEventLoop::enqueue_task_concurrent_with_extra_ctx::>( + &raw const **mini, std::ptr::from_mut(load), on_load_mini, core::mem::offset_of!(jsc_api::JSBundler::Load, task), @@ -4377,7 +4378,8 @@ pub mod bv2_impl { // SAFETY: `resolve` is a valid &mut for the duration of the enqueue; // the mini loop dispatches `on_resolve_mini` on the bundler thread. unsafe { - mini.enqueue_task_concurrent_with_extra_ctx::>( + bun_event_loop::MiniEventLoop::MiniEventLoop::enqueue_task_concurrent_with_extra_ctx::>( + &raw const **mini, std::ptr::from_mut(resolve), on_resolve_mini, core::mem::offset_of!(jsc_api::JSBundler::Resolve, task), diff --git a/src/event_loop/MiniEventLoop.rs b/src/event_loop/MiniEventLoop.rs index 8c3bd7dd6a54..39c5475c69fb 100644 --- a/src/event_loop/MiniEventLoop.rs +++ b/src/event_loop/MiniEventLoop.rs @@ -175,7 +175,7 @@ impl MiniEventLoop { /// returning accessor is intentionally **not** provided: `UwsLoop::tick()` /// fires FilePoll callbacks which re-enter this struct via the /// `EventLoopCtx` vtable (`platform_event_loop`) and via - /// `EventLoopHandle::Mini` (e.g. `enqueue_task_concurrent` → `wakeup()`), + /// `EventLoopHandle::Mini` (e.g. `enqueue_task_concurrent`, which wakes it), /// so a held `&mut UwsLoop` across `.tick()` would alias. The loop is also /// a C-owned handle whose internals are mutated by uSockets itself. All /// access goes through the raw pointer instead. @@ -383,23 +383,43 @@ impl MiniEventLoop { } } - /// `task` must outlive the queued work item; ownership of the intrusive - /// node stays with the caller until the callback runs. - pub fn enqueue_task_concurrent(&mut self, task: NonNull) { - self.concurrent_tasks.push(task); - // SAFETY: see `loop_ptr()` invariant. - unsafe { (*self.loop_ptr()).wakeup() }; + /// Post `task` to the thread that owns this loop and wake it. Any thread. + /// + /// Raw pointer, not a receiver: the owner may run the task and free the + /// loop as soon as it is queued (`Bun.build` boxes one loop per pass), i.e. + /// before this returns. A reference argument would assert `*this` for the + /// whole call; here nothing reads `*this` after the push. + /// + /// # Safety + /// `this` must point to a live `MiniEventLoop` when called; its owner may + /// free it once the task is queued. `task` must outlive the queued work + /// item; ownership of the intrusive node stays with the caller until the + /// callback runs. + pub unsafe fn enqueue_task_concurrent( + this: *const Self, + task: NonNull, + ) { + // SAFETY: `*this` is live until `push_raw` publishes `task` (fn contract) + // and is not touched after; `loop_` outlives this struct (see `loop_ptr`) + // and `us_wakeup_loop` is thread-safe. + unsafe { + let loop_ = (*this).loop_; + ConcurrentTaskQueue::push_raw(&raw const (*this).concurrent_tasks, task); + bun_uws::us_wakeup_loop(loop_); + } } - /// The caller supplies `field_offset = core::mem::offset_of!(C, )` of the - /// embedded `AnyTaskWithExtraContext`. + /// Initialize the `AnyTaskWithExtraContext` embedded in `*ctx` at + /// `field_offset = core::mem::offset_of!(C, )`, then post it via + /// [`enqueue_task_concurrent`](Self::enqueue_task_concurrent). /// /// # Safety + /// `this` as for [`enqueue_task_concurrent`](Self::enqueue_task_concurrent). /// `field_offset == offset_of!(C, )` where `: AnyTaskWithExtraContext`, /// and `ctx` is non-null and outlives the queued task (intrusive node; ownership stays /// with caller). pub unsafe fn enqueue_task_concurrent_with_extra_ctx( - &mut self, + this: *const Self, ctx: *mut C, callback: fn(*mut C, *mut P), field_offset: usize, @@ -409,12 +429,9 @@ impl MiniEventLoop { // SAFETY: `task` points at a properly aligned `AnyTaskWithExtraContext` field of `*ctx`. unsafe { task.write(New::::init(ctx, callback)) }; - // SAFETY: `task` was just initialized above and is non-null (derived from `ctx`). - self.concurrent_tasks - .push(unsafe { NonNull::new_unchecked(task) }); - - // SAFETY: see `loop_ptr()` invariant. - unsafe { (*self.loop_ptr()).wakeup() }; + // SAFETY: `task` was just initialized above and is non-null (derived from + // `ctx`); `this` is forwarded under the same contract. + unsafe { Self::enqueue_task_concurrent(this, NonNull::new_unchecked(task)) }; } } diff --git a/src/jsc/VmHandle.rs b/src/jsc/VmHandle.rs index 1a536254a518..7d54cc92b386 100644 --- a/src/jsc/VmHandle.rs +++ b/src/jsc/VmHandle.rs @@ -27,6 +27,7 @@ use bun_threading::{Condvar, Mutex}; use crate::event_loop::EventLoop; use crate::virtual_machine::VirtualMachine; use bun_event_loop::ConcurrentTask::ConcurrentTask as ConcurrentTaskItem; +use bun_event_loop::MiniEventLoop::MiniEventLoop; #[repr(u8)] #[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug)] @@ -599,7 +600,7 @@ pub enum ConcurrentPoster { /// Erased handle of the JS loop's VM (obtained from the `EventLoopHandle` /// itself, so it is correct whichever thread constructs the poster). Js(bun_event_loop::JsPoster), - Mini(bun_ptr::BackRef), + Mini(bun_ptr::BackRef), } impl ConcurrentPoster { @@ -650,11 +651,9 @@ impl ConcurrentPoster { ) { match self { ConcurrentPoster::Mini(mini) => { - let mut mini = *mini; - // SAFETY: per `EventLoopHandle::Mini` invariant — the mini loop is - // alive for as long as work it created runs; its concurrent queue - // push is thread-safe. - unsafe { mini.get_mut() }.enqueue_task_concurrent(task); + // SAFETY: `EventLoopHandle::Mini` invariant — the loop is alive while + // work it created runs, and this post is that work's last step. + unsafe { MiniEventLoop::enqueue_task_concurrent(mini.as_const_ptr(), task) }; } ConcurrentPoster::Js(..) => debug_assert!(false, "post_mini on a Js poster"), } diff --git a/src/runtime/api/JSBundler.rs b/src/runtime/api/JSBundler.rs index ab5c7b353482..9944772ef2ee 100644 --- a/src/runtime/api/JSBundler.rs +++ b/src/runtime/api/JSBundler.rs @@ -1535,7 +1535,8 @@ pub mod js_bundler { bun_event_loop::AnyEventLoop::Mini(mini) => { // `mini.enqueueTaskConcurrentWithExtraCtx( // Load, BundleV2, this, BundleV2.onNotifyDeferMini, .task)` - mini.enqueue_task_concurrent_with_extra_ctx::>( + bun_event_loop::MiniEventLoop::MiniEventLoop::enqueue_task_concurrent_with_extra_ctx::>( + &raw const **mini, std::ptr::from_mut::(self), on_notify_defer_mini_wrap, core::mem::offset_of!(Load, task), diff --git a/src/runtime/shell/builtin/yes.rs b/src/runtime/shell/builtin/yes.rs index 5b97cb46616b..50c488efcafb 100644 --- a/src/runtime/shell/builtin/yes.rs +++ b/src/runtime/shell/builtin/yes.rs @@ -6,6 +6,7 @@ use crate::shell::states::cmd::Exec; use crate::shell::yield_::Yield; use bun_event_loop::ConcurrentTask::AutoDeinit; +use bun_event_loop::MiniEventLoop::MiniEventLoop; use bun_event_loop::{EventLoopTask, TaskTag, Taskable, task_tag}; #[derive(Clone, Copy, PartialEq, Eq, Default)] @@ -226,7 +227,7 @@ impl YesTask { // Same-thread bounce on the loop's own thread: always accepted. let _ = owner.js_poster().post(ct); } - EventLoopHandle::Mini(mut mini) => { + EventLoopHandle::Mini(mini) => { (*mini.loop_).tick(); let at = core::ptr::NonNull::new_unchecked(match &mut (*this).concurrent_task { @@ -235,7 +236,7 @@ impl YesTask { } EventLoopTask::Js(_) => unreachable!(), }); - mini.get_mut().enqueue_task_concurrent(at); + MiniEventLoop::enqueue_task_concurrent(mini.as_const_ptr(), at); } } } diff --git a/src/spawn/process.rs b/src/spawn/process.rs index dadd2fbb871b..d9f4acefa5fb 100644 --- a/src/spawn/process.rs +++ b/src/spawn/process.rs @@ -932,6 +932,7 @@ pub mod waiter_thread_posix { use super::*; use bun_event_loop::AnyTaskWithExtraContext::{AnyTaskWithExtraContext, New as AnyTaskNew}; use bun_event_loop::ConcurrentTask::{ConcurrentTask, Task, TaskTag}; + use bun_event_loop::MiniEventLoop::MiniEventLoop; use bun_event_loop::task_tag; use bun_threading::UnboundedQueue; @@ -1198,19 +1199,23 @@ pub mod waiter_thread_posix { } } } - EventLoopHandle::Mini(mut mini) => { + EventLoopHandle::Mini(mini) => { let out = ResultTaskMini::::new(ResultTaskMini { result, subprocess: process, task: AnyTaskWithExtraContext::default(), }); - // SAFETY: `out` just produced by heap::alloc — non-null. + // SAFETY: `out` just produced by heap::alloc — non-null; + // `mini` is the handle's live loop (`EventLoopHandle::Mini` + // invariant) and this post is the last thing the waiter + // thread does with it. unsafe { (*out).task = AnyTaskNew::, ()>::init( out, ResultTaskMini::::run_from_main_thread_mini, ); - mini.get_mut().enqueue_task_concurrent( + MiniEventLoop::enqueue_task_concurrent( + mini.as_const_ptr(), core::ptr::NonNull::new_unchecked(core::ptr::addr_of_mut!( (*out).task )), diff --git a/src/threading/WaitGroup.rs b/src/threading/WaitGroup.rs index 4f45613b1555..fa8d324db010 100644 --- a/src/threading/WaitGroup.rs +++ b/src/threading/WaitGroup.rs @@ -92,6 +92,12 @@ mod tests { // After `wait()` returns the caller may drop the `WaitGroup`; `finish()` // must therefore not touch `self` once it has published `raw_count == 0`. + // Miri (Tree Borrows) currently reports this as violated: the `&self` of + // `finish` / `Mutex::unlock` still covers the group when the waiter frees it. + #[cfg_attr( + miri, + ignore = "finish(&self) is still on the finisher's stack when wait() returns" + )] #[test] fn wait_returning_means_finish_is_done_with_self() { for _ in 0..10_000 { diff --git a/src/threading/unbounded_queue.rs b/src/threading/unbounded_queue.rs index ef75baa2d433..22c2995371fc 100644 --- a/src/threading/unbounded_queue.rs +++ b/src/threading/unbounded_queue.rs @@ -207,12 +207,36 @@ impl UnboundedQueue { /// accessed outside this queue. The caller transfers logical ownership of /// the node to the queue until a `pop`/`pop_batch` returns it. pub fn push(&self, item: NonNull) { - self.push_batch(item, item); + // SAFETY: `self` is live for the whole call. + unsafe { Self::push_batch(self, item, item) }; + } + + /// [`push`](Self::push) for a producer whose queue the consumer may free as + /// soon as the node is visible (`MiniEventLoop::enqueue_task_concurrent`). + /// A `&self` argument would assert the queue's storage, padding included, + /// until this returns; [`push_batch`](Self::push_batch) asserts nothing + /// after its publishing store. + /// + /// # Safety + /// `this` must point to a live queue when called (the consumer may free it + /// once the node is visible); `item` as for [`push`](Self::push). + pub unsafe fn push_raw(this: *const Self, item: NonNull) { + // SAFETY: fn contract. + unsafe { Self::push_batch(this, item, item) }; } /// `first..=last` must form a valid intrusive chain of live `T` nodes. The /// caller transfers logical ownership of every node in the chain. - pub(crate) fn push_batch(&self, first: NonNull, last: NonNull) { + /// + /// The chain becomes visible to the consumer at the `next` store (queue was + /// non-empty) or the `front` store (queue was empty). A + /// [`push_raw`](Self::push_raw) caller's queue may be freed from that point + /// on, so nothing after either store may touch `*this`. + /// + /// # Safety + /// `this` must point to a live queue when called; the consumer may free it + /// once the chain is visible. + pub(crate) unsafe fn push_batch(this: *const Self, first: NonNull, last: NonNull) { let (first, last) = (first.as_ptr(), last.as_ptr()); // SAFETY: caller guarantees `last` is a live node (NonNull is non-null). unsafe { T::set_next(last, ptr::null_mut()) }; @@ -229,13 +253,16 @@ impl UnboundedQueue { } debug_assert!(item == last, "`last` should be reachable from `first`"); } - let old_back = self.back.0.swap(last, Ordering::AcqRel); + // SAFETY: the chain is not visible yet, so `*this` is live (fn contract). + let old_back = unsafe { (*this).back.0.swap(last, Ordering::AcqRel) }; if !old_back.is_null() { // SAFETY: `old_back` was the previous tail, still live (its `next` // is null and no consumer has popped past it yet — see `pop`). unsafe { T::atomic_store_next(old_back, first, Ordering::Release) }; } else { - self.front.0.store(first, Ordering::Release); + // SAFETY: the chain is not visible until this store completes, so + // `*this` is live for it (fn contract). + unsafe { (*this).front.0.store(first, Ordering::Release) }; } } @@ -340,3 +367,91 @@ impl UnboundedQueue { self.back.0.load(Ordering::Acquire).is_null() } } + +#[cfg(test)] +mod tests { + use super::*; + + struct Node(Link); + + // SAFETY: `0` is the node's only link field. + unsafe impl Linked for Node { + unsafe fn link(item: *mut Self) -> *const Link { + // SAFETY: `item` is a live node (queue contract). + unsafe { ptr::addr_of!((*item).0) } + } + } + + struct SendPtr(*mut T); + // SAFETY: the tests below only dereference the pointer on the thread that + // currently owns the pointee (see each use). + unsafe impl Send for SendPtr {} + + // The consumer of a `push_raw` producer may free the queue as soon as the + // node is visible (`MiniEventLoop::enqueue_task_concurrent`: the node can be + // the last thing the loop's owner was waiting for), so `push_batch` must not + // touch `*this` after its publishing store. Under Miri (`bun run + // rust:miri`) any such access is reported as a race with the consumer's + // free, whatever the interleaving; natively this only checks delivery. The + // two passes cover the publishing store on `front` (empty queue) and on the + // previous tail's link (non-empty queue, where the consumer also writes + // `front` itself while taking the second node). + #[test] + fn consumer_may_free_the_queue_once_the_node_is_visible() { + let iterations = if cfg!(miri) { 64 } else { 10_000 }; + for prefill in [false, true] { + for _ in 0..iterations { + let queue = Box::into_raw(Box::new(UnboundedQueue::::new())); + let mut first = Node(Link::new()); + let mut second = Node(Link::new()); + let expected = if prefill { + // SAFETY: the queue was just allocated and nothing else + // refers to it yet. + unsafe { (*queue).push(NonNull::from(&mut first)) }; + 2 + } else { + 1 + }; + + let consumer = { + let queue = SendPtr(queue); + std::thread::spawn(move || { + let queue = queue; + // Addresses, so the result can cross back to the producer. + let mut popped: Vec = Vec::with_capacity(expected); + while popped.len() < expected { + // SAFETY: the queue is live until this thread frees + // it below. + let node = unsafe { (*queue.0).pop() }; + if node.is_null() { + std::thread::yield_now(); + } else { + popped.push(node as usize); + } + } + // Every node the producer will ever push has arrived: + // the property under test is that freeing here is + // fine whatever the producer is still doing. + // SAFETY: allocated with `Box::new` above; the producer + // holds no reference to it, only the pointer it posts + // through, and it never dereferences that again. + drop(unsafe { Box::from_raw(queue.0) }); + popped + }) + }; + + // SAFETY: the queue is live at least until this node is + // visible, which is all `push_raw` requires. + unsafe { UnboundedQueue::push_raw(queue, NonNull::from(&mut second)) }; + + let popped = consumer.join().unwrap(); + let mut want = Vec::new(); + if prefill { + want.push(&raw mut first as usize); + } + want.push(&raw mut second as usize); + assert_eq!(popped, want); + } + } + } +}