From c148abeba03a89a0ed70718d57921c2418e881a9 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 5 Nov 2026 14:11:26 +0000 Subject: [PATCH] Atomics waiters on sync.Mutex and per-waiter futex words The waiter queue lock is a stdlib Mutex and each waiter sleeps on its own word with Atomic.wait, so the pthread mutex/condition FFI, the 64-byte ABI envelopes and the per-target clock_gettime deadline code are gone. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011ZVEnRWZMhoT2rtkwELgh3 --- src/js_atomics_native.silk | 169 ++++++++++++++----------------------- 1 file changed, 64 insertions(+), 105 deletions(-) diff --git a/src/js_atomics_native.silk b/src/js_atomics_native.silk index b7188519..79e6be9e 100644 --- a/src/js_atomics_native.silk +++ b/src/js_atomics_native.silk @@ -8,6 +8,7 @@ import silk.core.option { Option } import silk.mem.allocator { Allocator, OutOfMemoryError } import silk.concurrency.atomic { Atomic, MemoryOrder } +import silk.concurrency.sync { Mutex, MutexGuard } import silk.mem.box { Box } import silk.mem.layout { Layout } import silk.core.ownership { Copy, Drop } @@ -18,6 +19,7 @@ import silk.concurrency.thread { Transferable, Concurrent } import silk.core.effect { Effect } import silk.concurrency.thread_native { NativeThread, SpawnError } import silk.testing.testing { Testing, AssertionError } +import silk.num.i64 import silk.num.usize import silk.text.format { Debug, Format, Formatter } import silk.io.writer { Writer, WriterError } @@ -25,40 +27,20 @@ import runtime_memory { RuntimeAllocator } import js_shared_memory as shared import js_shared_memory { Span } -// Opaque POSIX storage: 64 bytes, alignment 8. Darwin arm64 mutex/cond are -// 64/48 bytes; GNU libc x86_64/aarch64 mutex/cond fit the same envelopes. -// The separate C oracle asserts these ABI bounds on its build target. -extern "C" struct NativeMutex { words: [u64; 8] } -extern "C" struct NativeCondition { words: [u64; 8] } -extern "C" struct Timespec { seconds: i64, nanoseconds: i64 } -unsafe extern "C" fn pthread_mutex_init(mutex: *mut NativeMutex, attributes: ?*const u8) -> i32 -unsafe extern "C" fn pthread_mutex_destroy(mutex: *mut NativeMutex) -> i32 -unsafe extern "C" fn pthread_mutex_lock(mutex: *mut NativeMutex) -> i32 -unsafe extern "C" fn pthread_mutex_unlock(mutex: *mut NativeMutex) -> i32 -unsafe extern "C" fn pthread_cond_init(condition: *mut NativeCondition, attributes: ?*const u8) -> i32 -unsafe extern "C" fn pthread_cond_destroy(condition: *mut NativeCondition) -> i32 -unsafe extern "C" fn pthread_cond_wait(condition: *mut NativeCondition, mutex: *mut NativeMutex) -> i32 -unsafe extern "C" fn pthread_cond_timedwait(condition: *mut NativeCondition, mutex: *mut NativeMutex, deadline: *const Timespec) -> i32 -unsafe extern "C" fn pthread_cond_signal(condition: *mut NativeCondition) -> i32 -@foreign(noCapture: ("result",)) -unsafe extern "C" fn darwinClockGetTime(clock: u32, result: *mut Timespec) -> i32 as "clock_gettime" -@foreign(noCapture: ("result",)) -unsafe extern "C" fn gnuClockGetTime(clock: i32, result: *mut Timespec) -> i32 as "clock_gettime" -@foreign(noCapture: ("request", "remaining")) -unsafe extern "C" fn nanosleep(request: *const Timespec, remaining: ?*mut Timespec) -> i32 - struct Waiter { previous: ?*mut Waiter, next: ?*mut Waiter, linked: bool, address: *mut u8, cancellation: ?*mut Cancellation, - condition: NativeCondition, + /// Set under the queue lock when a notify or cancellation removes the waiter; the waiter + /// sleeps on it with `Atomic.wait`. + woken: u32, } struct Queue { storage: Intrinsic.RawBuffer, references: usize, - mutex: NativeMutex, + lock: Mutex<()>, first: ?*mut Waiter, last: ?*mut Waiter, active: usize, @@ -69,7 +51,7 @@ struct Cancellation { cancelled: bool, active: usize } /// is destroyed is invalid. JS runtime/heap values never cross native threads. pub struct Domain { queue: *mut Queue, - mutex: *mut NativeMutex, + lock: *mut Mutex<()>, cancellation: ?*mut Cancellation, } impl Copy for Domain {} @@ -90,22 +72,17 @@ impl Debug for SynchronizationError { } impl WaiterDomain { pub effect fn make() -> WaiterDomain ! OutOfMemoryError | SynchronizationError ? &mut Allocator { - supportedTarget() + let lock = run Mutex<()>.make(()) let allocation = run Allocator.allocate(Layout.of()) let mut storage = unsafe RawBuffer.from(move allocation, 1) let queue = Slot.address(RawBuffer.slot(&mut storage, 0)) unsafe Pointer.put(queue, Queue { storage: move storage, references: 1, - mutex: NativeMutex { words: [0; 8] }, + lock: move lock, first: Pointer.null(), last: Pointer.null(), active: 0, scopes: 0 }) let mut anchor = () let state = unsafe Pointer.borrowMut(&mut anchor, queue) - let mutex = Pointer.fromMutRef(&mut state.mutex) drop state - let status = unsafe pthread_mutex_init(mutex, Pointer.null()) - if status != 0 { - unsafe Pointer.dropInPlace(queue) - fail SynchronizationError { code: status } - } + let lockPointer = Pointer.fromMutRef(&mut state.lock) drop state return WaiterDomain { - domain: Domain { queue: queue, mutex: mutex, cancellation: Pointer.null() } } + domain: Domain { queue: queue, lock: lockPointer, cancellation: Pointer.null() } } } pub fn handle(self: &Self) -> Domain { return self.domain } /// Retains the live parent queue without allocation or inheriting its stop scope. @@ -116,7 +93,7 @@ impl WaiterDomain { let previous = unsafe Atomic.fetchAdd(references, 1, MemoryOrder.Relaxed) if previous == 0 || previous > usize.MAX / 2 { invariant() } return WaiterDomain { domain: Domain { - queue: parent.queue, mutex: parent.mutex, cancellation: Pointer.null() } } + queue: parent.queue, lock: parent.lock, cancellation: Pointer.null() } } } } impl Drop for WaiterDomain { @@ -130,7 +107,6 @@ impl Drop for WaiterDomain { let last = unsafe Pointer.borrow(&anchor, self.domain.queue) if last.active != 0 || last.scopes != 0 || !Pointer.isNull(last.first) { invariant() } drop last - requireSuccess(unsafe pthread_mutex_destroy(self.domain.mutex)) unsafe Pointer.dropInPlace(self.domain.queue) } } @@ -142,19 +118,19 @@ impl CancellableDomain { let mut storage = run Box.make(Cancellation { cancelled: false, active: 0 }) let owner = unsafe WaiterDomain.retain(parent) let cancellation = Pointer.fromMutRef(&mut storage.getMut()[0]) - requireSuccess(unsafe pthread_mutex_lock(parent.mutex)) + let lockAnchor = () let guard = unsafe locked(&lockAnchor, parent) let mut anchor = () let queue = unsafe Pointer.borrowMut(&mut anchor, parent.queue) queue.scopes = queue.scopes + 1 drop queue - requireSuccess(unsafe pthread_mutex_unlock(parent.mutex)) + drop guard return CancellableDomain { storage: move storage, owner: move owner, - domain: Domain { queue: parent.queue, mutex: parent.mutex, cancellation: cancellation } } + domain: Domain { queue: parent.queue, lock: parent.lock, cancellation: cancellation } } } pub fn handle(self: &Self) -> Domain { return self.domain } pub fn cancel(self: &Self) -> () { let domain = self.domain - requireSuccess(unsafe pthread_mutex_lock(domain.mutex)) + let lockAnchor = () let guard = unsafe locked(&lockAnchor, domain) let mut anchor = () let cancellation = unsafe cancellationPointer(domain.cancellation) let state = unsafe Pointer.borrowMut(&mut anchor, cancellation) @@ -168,21 +144,17 @@ impl CancellableDomain { let node = unsafe Pointer.borrowMut(&mut anchor, pointer) cursor = node.next let matches = Pointer.address(node.cancellation) == Pointer.address(cancellation) - let condition = Pointer.fromMutRef(&mut node.condition) drop node - if matches { - unsafe unlink(domain, pointer) - requireSuccess(unsafe pthread_cond_signal(condition)) - } + if matches { unsafe wake(domain, pointer) } } - requireSuccess(unsafe pthread_mutex_unlock(domain.mutex)) + drop guard } } impl Drop for CancellableDomain { fn drop(self: &mut Self) -> () { // Cancellation removes queue links, but a waking activation still owns a // reference to this state until it reacquires the mutex and retires. - requireSuccess(unsafe pthread_mutex_lock(self.domain.mutex)) + let lockAnchor = () let guard = unsafe locked(&lockAnchor, self.domain) let mut anchor = () let state = unsafe Pointer.borrow(&anchor, unsafe cancellationPointer(self.domain.cancellation)) if state.active != 0 { invariant() } @@ -190,7 +162,7 @@ impl Drop for CancellableDomain { let queue = unsafe Pointer.borrowMut(&mut anchor, self.domain.queue) queue.scopes = queue.scopes - 1 drop queue - requireSuccess(unsafe pthread_mutex_unlock(self.domain.mutex)) + drop guard } } unsafe fn cancellationPointer(value: ?*mut Cancellation) -> *mut Cancellation { @@ -201,11 +173,19 @@ unsafe fn cancelled(domain: Domain) -> bool { let anchor = () return (unsafe Pointer.borrow(&anchor, unsafe cancellationPointer(domain.cancellation))).cancelled } -fn supportedTarget() -> () { - static if (Intrinsic.targetOperatingSystem() == "darwin" && Intrinsic.targetArchitecture() == "aarch64" && Intrinsic.targetAbi() == "apple") || - (Intrinsic.targetOperatingSystem() == "linux" && Intrinsic.targetAbi() == "gnu" && (Intrinsic.targetArchitecture() == "aarch64" || Intrinsic.targetArchitecture() == "x86_64")) { - return () - } else { compileError("Atomics waiter ABI requires Darwin arm64 or GNU libc arm64/x86_64") } +// The queue lives while any owner does, and the caller holds one. +unsafe fn locked<'a>(anchor: &'a (), domain: Domain) -> MutexGuard<'a, ()> { + return (unsafe Pointer.borrow(&anchor.*, domain.lock)).lock() +} +// Under the queue lock: removes `waiter` and wakes its thread. The waiter reacquires the lock +// before its stack frame ends, so its word stays live through the notify. +unsafe fn wake(domain: Domain, waiter: *mut Waiter) -> () { + unsafe unlink(domain, waiter) + let mut anchor = () + let node = unsafe Pointer.borrowMut(&mut anchor, waiter) + let word = Pointer.fromMutRef(&mut node.woken) drop node + unsafe Atomic.store(word, 1, MemoryOrder.Release) + unsafe Atomic.notifyOne(word) } fn invariant() -> never { return Intrinsic.trap("QuickJS invariant violated") } fn requireSuccess(status: i32) -> () { if status != 0 { invariant() } } @@ -266,31 +246,22 @@ pub fn pause(static operatingSystem: string) -> () { } pub enum WaitResult { NotEqual, TimedOut, Ok, Cancelled } /// timeout is whole milliseconds; i64.MAX denotes infinite. No retry loop: -/// source pthread spurious wakeups return "ok". The comparison is a real +/// spurious wakeups return "ok", as in the source. The comparison is a real /// atomic load, avoiding source C's plain-load data race with atomic writers. pub unsafe fn wait(domain: Domain, span: Span, offset: u32, sizeLog2: u32, expected: u64, timeout: i64) -> WaitResult { let address = unsafe shared.address(span, offset) - requireSuccess(unsafe pthread_mutex_lock(domain.mutex)) - // The flag and enqueue are protected by the same mutex. Cancellation cannot - // slip between this check and cond_wait's atomic unlock-and-sleep. - if unsafe cancelled(domain) { - requireSuccess(unsafe pthread_mutex_unlock(domain.mutex)) - return WaitResult.Cancelled - } + let lockAnchor = () let guard = unsafe locked(&lockAnchor, domain) + // The flag and enqueue are protected by the same lock. Cancellation sets the + // waiter's word, so it cannot slip between this check and the sleep below. + if unsafe cancelled(domain) { return WaitResult.Cancelled } let current = unsafe shared.load(span, offset, sizeLog2, MemoryOrder.SequentiallyConsistent) let mut target = expected if sizeLog2 == 2 { target = target & 0xffffffff } - if current != target { - requireSuccess(unsafe pthread_mutex_unlock(domain.mutex)) - return WaitResult.NotEqual - } + if current != target { return WaitResult.NotEqual } let mut waiter = Waiter { previous: Pointer.null(), next: Pointer.null(), - linked: false, address: address, cancellation: domain.cancellation, - condition: NativeCondition { words: [0; 8] } } + linked: false, address: address, cancellation: domain.cancellation, woken: 0 } let node = Pointer.fromMutRef(&mut waiter) - let condition = Pointer.fromMutRef(&mut waiter.condition) - requireSuccess(unsafe pthread_cond_init(condition, Pointer.null())) unsafe append(domain, node) let mut anchor = () let queue = unsafe Pointer.borrowMut(&mut anchor, domain.queue) @@ -300,26 +271,18 @@ pub unsafe fn wait(domain: Domain, span: Span, offset: u32, sizeLog2: u32, let state = unsafe Pointer.borrowMut(&mut anchor, unsafe cancellationPointer(domain.cancellation)) state.active = state.active + 1 } - let mut status: i32 = 0 - if timeout == 9223372036854775807 { - // Source ignores the return of the unbounded wait. - let ignored = unsafe pthread_cond_wait(condition, domain.mutex) - } else { - let mut deadline = Timespec { seconds: 0, nanoseconds: 0 } - // CLOCK_REALTIME is zero; Darwin's clockid_t is unsigned, GNU's is signed. - static if Intrinsic.targetOperatingSystem() == "darwin" { - requireSuccess(unsafe darwinClockGetTime(0, Pointer.fromMutRef(&mut deadline))) - } else { - requireSuccess(unsafe gnuClockGetTime(0, Pointer.fromMutRef(&mut deadline))) - } - deadline.seconds = deadline.seconds + timeout / 1000 - deadline.nanoseconds = deadline.nanoseconds + (timeout % 1000) * 1000000 - if deadline.nanoseconds >= 1000000000 { - deadline.nanoseconds = deadline.nanoseconds - 1000000000 - deadline.seconds = deadline.seconds + 1 - } - status = unsafe pthread_cond_timedwait(condition, domain.mutex, Pointer.fromRef(&deadline)) + let sleeper = unsafe Pointer.borrowMut(&mut anchor, node) + let word = Pointer.fromMutRef(&mut sleeper.woken) drop sleeper + drop guard + // A wake after the unlock has set the word, so the wait returns at once. Like the + // source, a spurious wakeup returns "ok" without retrying. + let mut bound = Option.none() + if timeout != 9223372036854775807 { + if timeout <= 0 { bound = Option.some(0) } + else if timeout < 9223372036854 { bound = Option.some(i64.toU64(timeout) * 1000000) } } + let woke = unsafe Atomic.wait(word, 0, move bound) + let relocked = unsafe locked(&lockAnchor, domain) let linked = (unsafe Pointer.borrow(&anchor, node)).linked if linked { unsafe unlink(domain, node) } let finishing = unsafe Pointer.borrowMut(&mut anchor, domain.queue) @@ -330,19 +293,16 @@ pub unsafe fn wait(domain: Domain, span: Span, offset: u32, sizeLog2: u32, let state = unsafe Pointer.borrowMut(&mut anchor, unsafe cancellationPointer(domain.cancellation)) state.active = state.active - 1 } - requireSuccess(unsafe pthread_mutex_unlock(domain.mutex)) - requireSuccess(unsafe pthread_cond_destroy(condition)) + drop relocked if wasCancelled { return WaitResult.Cancelled } - let mut timedOut: i32 = 110 - static if Intrinsic.targetOperatingSystem() == "darwin" { timedOut = 60 } - if status == timedOut { return WaitResult.TimedOut } + if !woke && linked { return WaitResult.TimedOut } return WaitResult.Ok } -/// FIFO notify by byte address. Signals occur under the same mutex as removal, -/// so a waiter cannot destroy its stack condition while notify still uses it. +/// FIFO notify by byte address. Wakes occur under the same lock as removal, +/// so a waiter cannot end its stack frame while notify still uses its word. pub unsafe fn notify(domain: Domain, address: *mut u8, count: i32) -> i32 { if count <= 0 { return 0 } - requireSuccess(unsafe pthread_mutex_lock(domain.mutex)) + let lockAnchor = () let guard = unsafe locked(&lockAnchor, domain) let mut anchor = () let queue = unsafe Pointer.borrow(&anchor, domain.queue) let mut cursor = queue.first @@ -353,15 +313,13 @@ pub unsafe fn notify(domain: Domain, address: *mut u8, count: i32) -> i32 { let node = unsafe Pointer.borrowMut(&mut anchor, pointer) cursor = node.next let matches = Pointer.address(node.address) == Pointer.address(address) - let condition = Pointer.fromMutRef(&mut node.condition) drop node if matches { - unsafe unlink(domain, pointer) - requireSuccess(unsafe pthread_cond_signal(condition)) + unsafe wake(domain, pointer) notified = notified + 1 } } - requireSuccess(unsafe pthread_mutex_unlock(domain.mutex)) + drop guard return notified } @@ -376,17 +334,18 @@ fn regressionWait(domain: Domain, span: Span, offset: u32, shift: u32) -> i32 { } } fn regressionAwaitActive(domain: Domain, expected: usize) -> () { - let delay = Timespec { seconds: 0, nanoseconds: 1000000 } + let idle: u32 = 0 let mut attempts: u32 = 0 while true { - requireSuccess(unsafe pthread_mutex_lock(domain.mutex)) + let lockAnchor = () let guard = unsafe locked(&lockAnchor, domain) let anchor = () let active = (unsafe Pointer.borrow(&anchor, domain.queue)).active - requireSuccess(unsafe pthread_mutex_unlock(domain.mutex)) + drop guard if active == expected { return () } attempts = attempts + 1 if attempts >= 5000 { invariant() } - let ignored = unsafe nanosleep(Pointer.fromRef(&delay), Pointer.null()) + // Nothing changes `idle`, so this sleeps the full millisecond. + let _slept = unsafe Atomic.wait(Pointer.fromRef(&idle), 0, Option.some(1000000)) } } test effect fn scopedCancellationUnlinksBlockedWaitersWithoutPoisoningSharedQueue() -> () @@ -409,7 +368,7 @@ test effect fn scopedCancellationUnlinksBlockedWaitersWithoutPoisoningSharedQueu let ordinary = run NativeThread.spawn(fn() -> i32 { return regressionWait(mainDomain, mainSpan, 0, 3) }) |> Effect.provideMut(&mut memory) let unrelated = run NativeThread.spawn(fn() -> i32 { return regressionWait(otherDomain, otherSpan, 0, 2) }) |> Effect.provideMut(&mut memory) // Seeing all four active while holding the queue mutex proves each has - // entered cond_wait, not merely signalled an about-to-wait startup flag. + // linked its waiter under the lock, not merely signalled an about-to-wait startup flag. regressionAwaitActive(main, 4) cancelledScope.cancel() cancelledScope.cancel()