Atomics waiters on sync.Mutex and per-waiter futex words

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,47 +19,28 @@ 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 }
 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<Queue>,
   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<Queue>())
     let mut storage = unsafe RawBuffer.from<Queue>(move allocation, 1)
     let queue = Slot.address(RawBuffer.slot<Queue>(&mut storage, 0))
     unsafe Pointer.put<Queue>(queue, Queue { storage: move storage, references: 1,
-      mutex: NativeMutex { words: [0; 8] },
+      lock: move lock,
       first: Pointer.null<Waiter>(), last: Pointer.null<Waiter>(), 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<u8>())
-    if status != 0 {
-      unsafe Pointer.dropInPlace<Queue>(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<Cancellation>() } }
+      domain: Domain { queue: queue, lock: lockPointer, cancellation: Pointer.null<Cancellation>() } }
   }
   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<usize>(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<Cancellation>() } }
+      queue: parent.queue, lock: parent.lock, cancellation: Pointer.null<Cancellation>() } }
   }
 }
 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<Waiter>(last.first) { invariant() }
     drop last
-    requireSuccess(unsafe pthread_mutex_destroy(self.domain.mutex))
     unsafe Pointer.dropInPlace<Queue>(self.domain.queue)
   }
 }
@@ -142,19 +118,19 @@ impl CancellableDomain {
     let mut storage = run Box<Cancellation>.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,29 +144,25 @@ impl CancellableDomain {
       let node = unsafe Pointer.borrowMut(&mut anchor, pointer)
       cursor = node.next
       let matches = Pointer.address<Cancellation>(node.cancellation) == Pointer.address<Cancellation>(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() }
     drop state
     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<u32>(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<Waiter>(), next: Pointer.null<Waiter>(),
-    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<u8>()))
   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<u64>.none()
+  if timeout != 9223372036854775807 {
+    if timeout <= 0 { bound = Option<u64>.some(0) }
+    else if timeout < 9223372036854 { bound = Option<u64>.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<u8>(node.address) == Pointer.address<u8>(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<Timespec>())
+    // Nothing changes `idle`, so this sleeps the full millisecond.
+    let _slept = unsafe Atomic.wait(Pointer.fromRef(&idle), 0, Option<u64>.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<Allocator>(&mut memory)
   let unrelated = run NativeThread.spawn(fn() -> i32 { return regressionWait(otherDomain, otherSpan, 0, 2) }) |> Effect.provideMut<Allocator>(&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()