threading: publish into a Channel without holding a reference into it past the release

src/runtime/cli/run_command.rs

@@ -3096,7 +3096,10 @@ impl RemoteImageDownload {
         // to the channel.
         // SAFETY: `this` was passed as the callback ctx in `prefetch_remote_images`;
         // `async_http` is the worker-thread temporary whose `.real` points back at
-        // `this.async_http`.
+        // `this.async_http`. `prefetch_remote_images` reads one tick per download
+        // before it returns (freeing `this` and ending `done`'s storage), and this
+        // download's tick is the last statement below, so everything here runs
+        // while both are live, and the channel is live until that publish's unlock.
         unsafe {
             let this = &mut *this;
             let async_http = &mut *async_http;
@@ -3110,7 +3113,10 @@ impl RemoteImageDownload {
             result.body_into(&mut this.response_buffer.list);
             // Channel payload is a placeholder tick — the main thread
             // walks `downloads[]` to read per-task state after N wakeups.
-            let _ = (*this.done).write_item(0);
+            // If this is the last tick, the main thread may free the channel
+            // as soon as it has taken it, which `(*this.done).write_item(0)`
+            // would still hold a reference into (see `Channel::write_item_raw`).
+            let _ = DoneChannel::write_item_raw(this.done, 0);
         }
     }
 }

src/threading/Futex.rs

@@ -55,6 +55,19 @@ pub fn wait_forever(ptr: &AtomicU32, expect: u32) {
 /// Unblocks at most `max_waiters` callers blocked in a `wait()` call on `ptr`.
 #[cold]
 pub fn wake(ptr: &AtomicU32, max_waiters: u32) {
+    wake_raw(core::ptr::from_ref(ptr), max_waiters);
+}
+
+/// [`wake`] for a word that may be freed before this runs: a mutex's unlock
+/// tail (`Mutex::unlock_raw`), where the store that released the lock has to
+/// be the primitive's last access to its own memory because the thread it
+/// released may free the primitive at once. A `&AtomicU32` would assert the
+/// word live until this returns. Every backend's wake side uses the address
+/// only as a key and never reads the word, so this needs no `unsafe`: the
+/// worst a freed or reused address yields is a spurious wakeup, which every
+/// `wait()` loop tolerates.
+#[cold]
+pub(crate) fn wake_raw(ptr: *const AtomicU32, max_waiters: u32) {
     // Avoid calling into the OS if there's nothing to wake up.
     if max_waiters == 0 {
         return;
@@ -106,7 +119,7 @@ mod unsupported_impl {
         unsupported()
     }
 
-    pub(super) fn wake(_ptr: &AtomicU32, _max_waiters: u32) {
+    pub(super) fn wake(_ptr: *const AtomicU32, _max_waiters: u32) {
         unsupported()
     }
 
@@ -162,11 +175,12 @@ mod windows_impl {
         }
     }
 
-    pub(super) fn wake(ptr: &AtomicU32, max_waiters: u32) {
-        let address: *const c_void = ptr.as_ptr().cast();
+    pub(super) fn wake(ptr: *const AtomicU32, max_waiters: u32) {
+        let address: *const c_void = ptr.cast();
         debug_assert!(max_waiters != 0);
 
-        // SAFETY: address points at a live AtomicU32.
+        // SAFETY: RtlWakeAddress* only look `address` up in the waiter table;
+        // they never access the memory, so it need not be live (see `super::wake_raw`).
         unsafe {
             match max_waiters {
                 1 => windows::ntdll::RtlWakeAddressSingle(address),
@@ -261,7 +275,7 @@ mod darwin_impl {
         }
     }
 
-    pub(super) fn wake(ptr: &AtomicU32, max_waiters: u32) {
+    pub(super) fn wake(ptr: *const AtomicU32, max_waiters: u32) {
         let flags = c::UL {
             op: c::ULOp::COMPARE_AND_WAIT,
             no_errno: true,
@@ -270,8 +284,10 @@ mod darwin_impl {
         };
 
         loop {
-            let addr: *const c_void = ptr.as_ptr().cast();
-            // SAFETY: addr points at a live AtomicU32.
+            let addr: *const c_void = ptr.cast();
+            // SAFETY: __ulock_wake only keys the waiter lookup on `addr`; it never
+            // accesses the memory (hence no EFAULT below), so it need not be live
+            // (see `super::wake_raw`).
             let status = unsafe { c::__ulock_wake(flags, addr, 0) };
 
             if status >= 0 {
@@ -346,16 +362,18 @@ mod linux_impl {
         }
     }
 
-    pub(super) fn wake(ptr: &AtomicU32, max_waiters: u32) {
+    pub(super) fn wake(ptr: *const AtomicU32, max_waiters: u32) {
         use bun_sys::linux;
         let val: u32 = match i32::try_from(max_waiters) {
             Ok(v) => v as u32,
             Err(_) => i32::MAX as u32,
         };
-        // SAFETY: ptr.as_ptr() is a valid *const u32 for the duration of the call.
+        // SAFETY: a private FUTEX_WAKE keys the waiter lookup on the address
+        // alone (`get_futex_key` does not touch the memory), so `ptr` need not
+        // point to live memory (see `super::wake_raw`).
         let rc = unsafe {
             linux::futex_3arg(
-                ptr.as_ptr().cast(),
+                ptr.cast(),
                 linux::FutexOp {
                     cmd: linux::FutexCmd::WAKE,
                     private: true,
@@ -367,7 +385,13 @@ mod linux_impl {
         match linux::E::init(rc) {
             linux::E::SUCCESS => {} // successful wake up
             linux::E::INVAL => {}   // invalid futex_wait() on ptr done elsewhere
-            linux::E::FAULT => panic!("futex_wake() returned EFAULT unexpectedly"), // pointer became invalid while doing the wake
+            // The kernel only reports this for an address outside user space.
+            #[cfg(not(miri))]
+            linux::E::FAULT => panic!("futex_wake() returned EFAULT unexpectedly"),
+            // Miri reports it for a word that has already been freed, which
+            // `super::wake_raw` allows.
+            #[cfg(miri)]
+            linux::E::FAULT => {}
             _ => panic!("Unexpected futex_wake() return code"),
         }
     }
@@ -427,15 +451,16 @@ mod freebsd_impl {
         }
     }
 
-    pub(super) fn wake(ptr: &AtomicU32, max_waiters: u32) {
+    pub(super) fn wake(ptr: *const AtomicU32, max_waiters: u32) {
         // The kernel reads n_wake as `int`; passing maxInt(u32) truncates to
         // -1 and umtxq_signal_queue's `++ret >= n_wake` returns after one
         // wakeup. _umtx_op(2): "Specify INT_MAX to wake up all waiters."
         let n: c_ulong = max_waiters.min(c_int::MAX as u32) as c_ulong;
-        // SAFETY: ptr.as_ptr() is valid for the duration of the call.
+        // SAFETY: a private WAKE only keys the waiter lookup on the address; it
+        // never accesses the memory, so it need not be live (see `super::wake_raw`).
         let rc = unsafe {
             libc::_umtx_op(
-                ptr.as_ptr().cast::<c_void>(),
+                ptr.cast::<c_void>().cast_mut(),
                 libc::UMTX_OP_WAKE_PRIVATE,
                 n,
                 core::ptr::null_mut(), // there is no timeout struct
@@ -480,14 +505,16 @@ mod wasm_impl {
         }
     }
 
-    pub fn wake(ptr: &AtomicU32, max_waiters: u32) {
+    pub fn wake(ptr: *const AtomicU32, max_waiters: u32) {
         #[cfg(not(target_feature = "atomics"))]
         compile_error!("WASI target missing cpu feature 'atomics'");
 
         debug_assert!(max_waiters != 0);
-        // SAFETY: ptr.as_ptr() is a valid aligned *mut i32 (AtomicU32 has the same layout).
+        // SAFETY: memory.atomic.notify only keys on the (aligned, in-bounds)
+        // address; linear memory is never unmapped, so a freed word is still a
+        // valid key (see `super::wake_raw`). AtomicU32 has the layout of i32.
         let woken_count = unsafe {
-            core::arch::wasm32::memory_atomic_notify(ptr.as_ptr().cast::<i32>(), max_waiters)
+            core::arch::wasm32::memory_atomic_notify(ptr.cast::<i32>().cast_mut(), max_waiters)
         };
         let _ = woken_count; // can be 0 when linker flag 'shared-memory' is not enabled
     }

src/threading/Mutex.rs

@@ -64,7 +64,24 @@ impl Mutex {
     /// Releases the mutex which was previously acquired with `lock()` or `try_lock()`.
     /// It is undefined behavior if the mutex is unlocked from a different thread that it was locked from.
     pub fn unlock(&self) {
-        self.impl_.unlock()
+        // SAFETY: `self` is held by this thread (fn contract) and live for the
+        // whole call.
+        unsafe { Self::unlock_raw(self) }
+    }
+
+    /// [`unlock`](Self::unlock) for a critical section whose exit is what lets
+    /// another thread free the mutex's owner (`WaitGroup::finish_raw`). A `&self`
+    /// argument would assert the mutex's storage, padding included, until this
+    /// returns; here the store that releases the lock is the last access to
+    /// `*this`, and the futex wake that may follow it goes by address only
+    /// ([`Futex::wake_raw`](crate::futex::wake_raw)).
+    ///
+    /// # Safety
+    /// `this` must point to a mutex this thread holds. It stays valid until the
+    /// lock is released; from then on another thread may free it.
+    pub(crate) unsafe fn unlock_raw(this: *const Self) {
+        // SAFETY: the lock is still held, so `*this` is live (fn contract).
+        unsafe { Impl::unlock_raw(&raw const (*this).impl_) }
     }
 
     /// Debug-only check that the calling thread already holds this mutex.
@@ -185,11 +202,15 @@ impl DebugImpl {
         self.locking_thread.store(current_id, Ordering::Relaxed);
     }
 
+    /// See [`Mutex::unlock_raw`] for the contract.
     #[inline]
-    fn unlock(&self) {
-        debug_assert!(self.locking_thread.load(Ordering::Relaxed) == current_thread_id());
-        self.locking_thread.store(0, Ordering::Relaxed);
-        self.impl_.unlock();
+    unsafe fn unlock_raw(this: *const Self) {
+        // SAFETY: the lock is still held, so `*this` is live (fn contract).
+        unsafe {
+            debug_assert!((*this).locking_thread.load(Ordering::Relaxed) == current_thread_id());
+            (*this).locking_thread.store(0, Ordering::Relaxed);
+            ReleaseImpl::unlock_raw(&raw const (*this).impl_);
+        }
     }
 }
 
@@ -242,10 +263,15 @@ impl WindowsImpl {
         AcquireSRWLockExclusive(&self.srwlock)
     }
 
-    fn unlock(&self) {
-        // SAFETY: caller acquired the lock on this thread (`Mutex::unlock`
-        // contract); releasing without ownership is documented UB on Windows.
-        unsafe { bun_sys::windows::kernel32::ReleaseSRWLockExclusive(self.srwlock.get()) }
+    /// See [`Mutex::unlock_raw`] for the contract.
+    unsafe fn unlock_raw(this: *const Self) {
+        // SAFETY: this thread holds the lock (fn contract), so `*this` is live
+        // up to the release inside the call; releasing without ownership is
+        // documented UB on Windows.
+        unsafe {
+            let srwlock = core::cell::UnsafeCell::raw_get(&raw const (*this).srwlock);
+            bun_sys::windows::kernel32::ReleaseSRWLockExclusive(srwlock)
+        }
     }
 }
 
@@ -277,12 +303,15 @@ pub(crate) struct OsUnfairLock {
 // The type encodes the only pointer-validity precondition, and Apple's runtime
 // detects misuse (recursive lock / unowned unlock) by aborting — which is safe
 // — so `safe fn` discharges the link-time proof and callers need no `unsafe`.
+// `os_unfair_lock_unlock` is the exception: a reference would assert the word
+// live until the call returns, but once it releases the lock another thread
+// may free the word (`Mutex::unlock_raw`), so it takes the address.
 #[cfg(target_vendor = "apple")]
 unsafe extern "C" {
     #[cfg(debug_assertions)]
     safe fn os_unfair_lock_trylock(lock: &core::cell::UnsafeCell<OsUnfairLock>) -> bool;
     safe fn os_unfair_lock_lock(lock: &core::cell::UnsafeCell<OsUnfairLock>);
-    safe fn os_unfair_lock_unlock(lock: &core::cell::UnsafeCell<OsUnfairLock>);
+    fn os_unfair_lock_unlock(lock: *mut OsUnfairLock);
 }
 
 #[cfg(target_vendor = "apple")]
@@ -302,8 +331,11 @@ impl DarwinImpl {
         os_unfair_lock_lock(&self.oul)
     }
 
-    fn unlock(&self) {
-        os_unfair_lock_unlock(&self.oul)
+    /// See [`Mutex::unlock_raw`] for the contract.
+    unsafe fn unlock_raw(this: *const Self) {
+        // SAFETY: this thread holds the lock (fn contract), so `*this` is live
+        // up to the release inside the call, which is its last access.
+        unsafe { os_unfair_lock_unlock(core::cell::UnsafeCell::raw_get(&raw const (*this).oul)) }
     }
 }
 
@@ -382,19 +414,26 @@ impl FutexImpl {
         }
     }
 
-    fn unlock(&self) {
+    /// See [`Mutex::unlock_raw`] for the contract.
+    unsafe fn unlock_raw(this: *const Self) {
         // Unlock the mutex and wake up a waiting thread if any.
         //
         // A waiting thread will acquire with `contended` instead of `locked`
         // which ensures that it wakes up another thread on the next unlock().
         //
         // Release barrier ensures the critical section happens before we let go of the lock
         // and that our critical section happens before the next lock holder grabs the lock.
-        let state = self.state.swap(Self::UNLOCKED, Ordering::Release);
+        //
+        // SAFETY: the lock is still held, so `*this` is live (fn contract).
+        let state_ptr = unsafe { &raw const (*this).state };
+        // SAFETY: as above; the swap is what releases the lock, and the last
+        // access to `*this`. The wake below goes by address because the thread
+        // the swap releases may have freed the mutex by the time it runs.
+        let state = unsafe { (*state_ptr).swap(Self::UNLOCKED, Ordering::Release) };
         debug_assert!(state != Self::UNLOCKED);
 
         if state == Self::CONTENDED {
-            Futex::wake(&self.state, 1);
+            Futex::wake_raw(state_ptr, 1);
         }
     }
 }
@@ -410,7 +449,7 @@ unsafe extern "C" fn Bun__lock(ptr: *mut ReleaseImpl) {
 #[unsafe(no_mangle)]
 unsafe extern "C" fn Bun__unlock(ptr: *mut ReleaseImpl) {
     // SAFETY: C caller passes a valid, initialized ReleaseImpl pointer that this thread locked.
-    unsafe { (*ptr).unlock() }
+    unsafe { ReleaseImpl::unlock_raw(ptr) }
 }
 
 #[unsafe(no_mangle)]