Fixes #5; Event notifier is a bottleneck for very small tasks (#54)

Fix #5 ; Ported from [Constantine
threadpool](https://github.com/mratsim/constantine/tree/master/constantine/threadpool)

Changes:

- Implement event count backoff from Constantine
- Fallback to generic futexes. [Constantine errors out
instead](https://github.com/mratsim/constantine/blob/ea8c268603a5c5f5be479dda9ba27dcbdc51dade/constantine/threadpool/primitives/futexes.nim#L18).
- Removed `foreignThreadsParked` redundant logic to match constantine.
- Additional fixes:
  - https://github.com/mratsim/constantine/pull/623
  - https://github.com/mratsim/constantine/pull/624
  - https://github.com/mratsim/constantine/pull/625

In the fib bench, this is ~10x faster. In the SPC it's ~5x faster when
setting task granularity to 1. It's 2x faster in nqueens and heat.

It also does not run into the event notifier race conditions reproduced
by `tests/stress/test_shutdown.nim` that cause a hang.

.github/workflows/ci.yml

@@ -16,4 +16,5 @@ jobs:
         nim -v
         nimble list --installed --version
         env NIMLANG=c nimble test
+        env NIMLANG=c nimble test_generic_futex
         env NIMLANG=c nimble test_bench

.gitignore

@@ -4,3 +4,4 @@ nimcache/
 build/
 nimble.develop
 nimble.paths
+constantine

taskpools.nimble

@@ -53,8 +53,13 @@ task test, "Run tests":
   for mode in ["", "-d:release", "-d:danger"]:
     runTests(mode)
 
+task test_generic_futex, "Run tests with generic futex":
+  for mode in ["", "-d:release", "-d:danger"]:
+    run mode & " -d:taskpoolsGenericFutex", "tests/test_all.nim"
+
 proc runBenchs(args: string) =
   run args, "benchmarks/dfs/taskpool_dfs.nim"
+  # run args, "benchmarks/fibonacci/taskpool_fib.nim"
   run args, "benchmarks/heat/taskpool_heat.nim"
   run args, "benchmarks/nqueens/taskpool_nqueens.nim"
   run args, "benchmarks/iqs_latency/taskpool_iqs_latency.nim"

taskpools/backoff.nim

@@ -0,0 +1,122 @@
+# taskpools
+# Copyright (c) 2018-2019    Status Research & Development GmbH
+# Copyright (c) 2020-Present Mamy André-Ratsimbazafy
+# Licensed and distributed under either of
+#   * MIT license (license terms in the root directory or at http://opensource.org/licenses/MIT).
+#   * Apache v2 license (license terms in the root directory or at http://www.apache.org/licenses/LICENSE-2.0).
+# at your option. This file may not be copied, modified, or distributed except according to those terms.
+
+import
+  std/atomics,
+  ./primitives/futexes
+
+{.push raises:[], checks:off.}
+
+# ############################################################
+#
+#                      Eventcount
+#
+# ############################################################
+
+type
+  EventCount* = object
+    ## The lock-free equivalent of a condition variable.
+    ##
+    ## Usage, if a thread needs to be parked until a condition is true
+    ##       and signaled by another thread:
+    ## ```Nim
+    ## if condition:
+    ##   return
+    ##
+    ## while true:
+    ##   ticket = ec.sleepy()
+    ##   if condition:
+    ##     ec.cancelSleep()
+    ##     break
+    ##   else:
+    ##     ec.sleep()
+    ## ```
+    waitset: Atomic[uint32]
+    # type waitset = object
+    #   committedSleep {.bitsize: 16.}: uint32
+    #   preSleep       {.bitsize: 16.}: uint32
+    #
+    # We need a precise committed sleep count for the `syncAll` barrier because
+    # a `preSleep` waiter may steal a task and create more work.
+    events: Futex
+
+  ParkingTicket* = object
+    epoch: uint32
+
+const # bitfield setup
+  #   Low 16 bits are waiters, up to 2¹⁶ = 65536 threads are supported
+  #   Next 16 bits are pre-waiters, planning to wait but not committed.
+  #
+  # OS limitations:
+  # - Windows 10 supports up to 256 cores (https://www.microsoft.com/en-us/microsoft-365/blog/2017/12/15/windows-10-pro-workstations-power-advanced-workloads/)
+  # - Linux CPUSET supports up to 1024 threads (https://man7.org/linux/man-pages/man3/CPU_SET.3.html)
+  #
+  # Hardware limitations:
+  # - Xeon Platinum 8490H, 60C/120T per socket
+  #   - up to 8 sockets: 960 threads
+
+  kPreWaitShift = 16'u32
+  kPreWait      = 1'u32 shl kPreWaitShift
+  kWait         = 1'u32
+  kCommitToWait = kWait - kPreWait
+  kWaitMask     = kPreWait-1
+  kPreWaitMask  = not kWaitMask
+
+func initialize*(ec: var EventCount) {.inline.} =
+  ec.waitset.store(0, moRelaxed)
+  ec.events.initialize()
+
+func `=destroy`*(ec: var EventCount) {.inline.} =
+  ec.events.teardown()
+
+proc sleepy*(ec: var EventCount): ParkingTicket {.noinit, inline.} =
+  ## To be called before checking if the condition to not sleep is met.
+  ## Returns a ticket to be used when committing to sleep
+  discard ec.waitset.fetchAdd(kPreWait, moRelease)
+  result.epoch = ec.events.load(moAcquire)
+
+proc sleep*(ec: var EventCount, ticket: ParkingTicket) {.inline.} =
+  ## Put a thread to sleep until notified.
+  ## If the ticket becomes invalid (a notification has been received)
+  ## by the time sleep is called, the thread won't enter sleep
+  discard ec.waitset.fetchAdd(kCommitToWait, moRelease)
+
+  while ec.events.load(moAcquire) == ticket.epoch:
+    ec.events.wait(ticket.epoch)
+
+  discard ec.waitset.fetchSub(kWait, moRelease)
+
+proc cancelSleep*(ec: var EventCount) {.inline.} =
+  ## Cancel a sleep that was scheduled.
+  discard ec.waitset.fetchSub(kPreWait, moRelease)
+
+proc wake*(ec: var EventCount) {.inline.} =
+  ## Prevent an idle thread from sleeping
+  ## or wake a sleeping one if there wasn't any idle
+  discard ec.events.increment(1, moRelease)
+  let waiters = ec.waitset.load(moAcquire)
+  if (waiters and kPreWaitMask) != 0:
+    # Some threads are in prewait and will see the event count change
+    # no need to do an expensive syscall
+    return
+  if waiters != 0:
+    ec.events.wake()
+
+proc wakeAll*(ec: var EventCount) {.inline.} =
+  ## Wake all threads if at least 1 is parked
+  discard ec.events.increment(1, moRelease)
+  let waiters = ec.waitset.load(moAcquire)
+  if (waiters and kWaitMask) != 0:
+    ec.events.wakeAll()
+
+proc getNumWaiters*(ec: var EventCount): tuple[preSleep, committedSleep: int32] {.noinit, inline.} =
+  ## Get the number of idle threads:
+  ## (preSleep, committedSleep)
+  let waiters = ec.waitset.load(moAcquire)
+  result.preSleep = cast[int32]((waiters and kPreWaitMask) shr kPreWaitShift)
+  result.committedSleep = cast[int32](waiters and kWaitMask)

taskpools/chase_lev_deques.nim

@@ -98,6 +98,27 @@ proc `[]=`[T](buf: var Buf[T], index: int, item: T) {.inline.} =
 proc `[]`[T](buf: var Buf[T], index: int): T {.inline.} =
   result = buf.rawBuffer[index and buf.mask].load(moRelaxed)
 
+proc peek*[T](deque: var ChaseLevDeque[T]): int =
+  ## Estimates the number of items pending in the deque.
+  ## In a single-producer multi-consumer setting:
+  ## - If called by the producer (owner) the true number might be less
+  ##   due to consumers stealing items concurrently.
+  ## - If called by a consumer the true number is undefined
+  ##   as other consumers also steal items concurrently and
+  ##   the producer pushes/pops them concurrently.
+  ##
+  ## If the producer peeks and this returns 0, the queue is empty.
+  ##
+  ## This is a non-locking operation.
+  let # Handle race conditions
+    b = deque.bottom.load(moRelaxed)  # Only the producer peeks in the taskpool so moRelaxed is enough
+    t = deque.top.load(moAcquire)
+
+  if b >= t:
+    return b-t
+  else:
+    return 0
+
 proc grow[T](deque: var ChaseLevDeque[T], buf: var ptr Buf[T], top, bottom: int) {.inline.} =
   ## Double the buffer size
   ## bottom is the last item index

taskpools/event_notifiers.nim

@@ -1,95 +0,0 @@
-# taskpools
-# Copyright (c) 2021-2023 Status Research & Development GmbH
-# Licensed and distributed under either of
-#   * MIT license (license terms in the root directory or at http://opensource.org/licenses/MIT).
-#   * Apache v2 license (license terms in the root directory or at http://www.apache.org/licenses/LICENSE-2.0).
-# at your option. This file may not be copied, modified, or distributed except according to those terms.
-
-# event_notifier.nim
-# ------------------
-# This file implements an event notifier.
-# It allows putting idle threads to sleep or waking them up.
-
-# Design
-# Currently it is a shared lock + condition variable (a.k.a. a semaphore)
-#
-# In the future an eventcount might be considered, an event count significantly
-# reduces scheduler overhead by removing lock acquisition from critical path.
-# See overview and implementations at
-# https://gist.github.com/mratsim/04a29bdd98d6295acda4d0677c4d0041
-#
-# Weave "one event-notifier per thread" further reduces overhead
-# but requires the threadpool to be message-passing based.
-# https://github.com/mratsim/weave/blob/a230cce98a8524b2680011e496ec17de3c1039f2/weave/cross_thread_com/event_notifiers.nim
-
-{.push raises: [].} # Ensure no exceptions can happen
-
-import
-  std/locks,
-  ./instrumentation/contracts
-
-type
-  EventNotifier* = object
-    ## This data structure allows threads to be parked when no events are pending
-    ## and woken up when a new event is.
-    # Lock must be aligned to a cache-line to avoid false-sharing.
-    lock{.align: 64.}: Lock
-    cond: Cond
-    parked: int
-    signals: int
-
-{.push overflowChecks: off.}       # We don't want exceptions (for Defect) in a multithreaded context
-                                   # but we don't to deal with underflow of unsigned int either
-                                   # say "if a < b - c" with c > b
-
-func initialize*(en: var EventNotifier) {.inline.} =
-  ## Initialize the event notifier
-  en.lock.initLock()
-  en.cond.initCond()
-  en.parked = 0
-  en.signals = 0
-
-func `=destroy`*(en: var EventNotifier) {.inline.} =
-  en.cond.deinitCond()
-  en.lock.deinitLock()
-
-func `=copy`*(dst: var EventNotifier, src: EventNotifier) {.error: "An event notifier cannot be copied".}
-func `=sink`*(dst: var EventNotifier, src: EventNotifier) {.error: "An event notifier cannot be moved".}
-
-proc park*(en: var EventNotifier) {.inline.} =
-  ## Wait until we are signaled of an event
-  ## Thread is parked and does not consume CPU resources
-  en.lock.acquire()
-
-  if en.signals > 0:
-    en.signals -= 1
-    en.lock.release()
-    return
-
-  en.parked += 1
-  while en.signals == 0: # handle spurious wakeups
-    en.cond.wait(en.lock)
-  en.parked -= 1
-  en.signals -= 1
-
-  postCondition: en.signals >= 0
-  en.lock.release()
-
-proc notify*(en: var EventNotifier) {.inline.} =
-  ## Unpark a thread if any is available
-  en.lock.acquire()
-
-  if en.parked > 0:
-    en.signals += 1
-    en.cond.signal()
-
-  en.lock.release()
-
-proc getParked*(en: var EventNotifier): int {.inline.} =
-  ## Get the number of parked thread
-  en.lock.acquire()
-  result = en.parked
-  en.lock.release()
-
-{.pop.} # overflowChecks
-{.pop.} # raises: [AssertionDefect]

taskpools/injection_queues.nim

@@ -24,14 +24,14 @@
 
 import std/atomics
 
-const TasksBetweenInjectionDrains* {.intdefine: "taskpoolsIqDrainTick".} = 61
+const tasksBetweenInjectionDrains* {.intdefine: "taskpoolsIqDrainTick".} = 61
   ## Drain the injection queue after processing at most this many local tasks,
   ## so externally submitted tasks are not starved while a worker churns through
   ## a local deque that internal spawns keep refilling. Prime to avoid resonance
   ## with regular workload sizes. Override with `-d:taskpoolsIqDrainTick:N`.
 
 static:
-  doAssert TasksBetweenInjectionDrains > 0,
+  doAssert tasksBetweenInjectionDrains > 0,
     "taskpoolsIqDrainTick must be a positive integer"
 
 type
@@ -49,8 +49,9 @@ proc init*[T](q: var InjectionQueue[T]) {.inline.} =
   ## Reset the queue to empty. Must be called before any push/drain.
   q.head.store(default(T), moRelaxed)
 
-proc push*[T](q: var InjectionQueue[T], node: T) {.inline.} =
+proc push*[T](q: var InjectionQueue[T], node: T, wasEmpty: var bool) {.inline.} =
   ## Push a node onto the queue from any thread (lock-free MPMC).
+  ## `wasEmpty` is set to `true` if the queue was empty before the push.
   ##
   ## The release CAS on the head makes the plain write to the intrusive link
   ## visible to the draining worker after its acquire exchange.
@@ -58,6 +59,7 @@ proc push*[T](q: var InjectionQueue[T], node: T) {.inline.} =
   while true:
     node.injectionNext = headOld  # plain write; ordered by the release CAS below
     if q.head.compareExchange(headOld, node, moRelease, moRelaxed):
+      wasEmpty = headOld.isNil
       break
 
 iterator drain*[T](q: var InjectionQueue[T]): T {.inline.} =

taskpools/primitives/futexes.nim

@@ -0,0 +1,23 @@
+# taskpools
+# Copyright (c) 2019 Mamy André-Ratsimbazafy
+# Copyright (c) 2021-2025 Status Research & Development GmbH
+# Licensed and distributed under either of
+#   * MIT license (license terms in the root directory or at http://opensource.org/licenses/MIT).
+#   * Apache v2 license (license terms in the root directory or at http://www.apache.org/licenses/LICENSE-2.0).
+# at your option. This file may not be copied, modified, or distributed except according to those terms.
+
+when defined(taskpoolsGenericFutex):
+  import ./futexes_generic
+  export futexes_generic
+elif defined(linux):
+  import ./futexes_linux
+  export futexes_linux
+elif defined(windows):
+  import ./futexes_windows
+  export futexes_windows
+elif defined(osx):
+  import ./futexes_macos
+  export futexes_macos
+else:
+  import ./futexes_generic
+  export futexes_generic

taskpools/primitives/futexes_generic.nim

@@ -0,0 +1,47 @@
+# taskpools
+# Copyright (c) 2021-2026 Status Research & Development GmbH
+# Licensed and distributed under either of
+#   * MIT license (license terms in the root directory or at http://opensource.org/licenses/MIT).
+#   * Apache v2 license (license terms in the root directory or at http://www.apache.org/licenses/LICENSE-2.0).
+# at your option. This file may not be copied, modified, or distributed except according to those terms.
+
+import std/[atomics, locks]
+export MemoryOrder
+
+type
+  Futex* = object
+    value: Atomic[uint32]
+    lock: Lock
+    cond: Cond
+
+proc initialize*(futex: var Futex) {.inline.} =
+  futex.value.store(0, moRelaxed)
+  initLock(futex.lock)
+  initCond(futex.cond)
+
+proc teardown*(futex: var Futex) {.inline.} =
+  futex.value.store(0, moRelaxed)
+  deinitLock(futex.lock)
+  deinitCond(futex.cond)
+
+proc load*(futex: var Futex, order: MemoryOrder): uint32 {.inline.} =
+  futex.value.load(order)
+
+proc store*(futex: var Futex, value: uint32, order: MemoryOrder) {.inline.} =
+  futex.value.store(value, order)
+
+proc increment*(futex: var Futex, value: uint32, order: MemoryOrder): uint32 {.inline.} =
+  futex.value.fetchAdd(value, order)
+
+proc wait*(futex: var Futex, expected: uint32) {.inline.} =
+  withLock(futex.lock):
+    if futex.value.load(moAcquire) == expected:
+      wait(futex.cond, futex.lock)
+
+proc wake*(futex: var Futex) {.inline.} =
+  withLock(futex.lock):
+    signal(futex.cond)
+
+proc wakeAll*(futex: var Futex) {.inline.} =
+  withLock(futex.lock):
+    broadcast(futex.cond)

taskpools/primitives/futexes_linux.nim

@@ -0,0 +1,79 @@
+# taskpools
+# Copyright (c) 2019 Mamy André-Ratsimbazafy
+# Copyright (c) 2021-2025 Status Research & Development GmbH
+# Licensed and distributed under either of
+#   * MIT license (license terms in the root directory or at http://opensource.org/licenses/MIT).
+#   * Apache v2 license (license terms in the root directory or at http://www.apache.org/licenses/LICENSE-2.0).
+# at your option. This file may not be copied, modified, or distributed except according to those terms.
+
+# A wrapper for linux futex.
+# Condition variables do not always wake on signal which can deadlock the runtime
+# so we need to roll up our sleeves and use the low-level futex API.
+
+import std/atomics
+export MemoryOrder
+
+# OS primitives
+# ------------------------------------------------------------------------
+
+const
+  FUTEX_WAIT_PRIVATE = 128
+  FUTEX_WAKE_PRIVATE = 129
+
+let NR_Futex {.importc: "SYS_futex", header: "<sys/syscall.h>".}: clong
+
+proc syscall(sysno: clong): cint {.importc, header:"<unistd.h>", varargs.}
+
+proc sysFutex(
+       futexAddr: pointer, operation: uint32, expected: uint32 or int32,
+       timeout: pointer = nil, val2: pointer = nil, val3: cint = 0): cint {.inline.} =
+  ## See https://web.archive.org/web/20230208151430/http://locklessinc.com/articles/futex_cheat_sheet/
+  ## and https://www.akkadia.org/drepper/futex.pdf
+  syscall(NR_Futex, futexAddr, operation, expected, timeout, val2, val3)
+
+# Futex API
+# ------------------------------------------------------------------------
+
+type
+  Futex* = object
+    value: Atomic[uint32]
+
+proc initialize*(futex: var Futex) {.inline.} =
+  futex.value.store(0, moRelaxed)
+
+proc teardown*(futex: var Futex) {.inline.} =
+  futex.value.store(0, moRelaxed)
+
+proc load*(futex: var Futex, order: MemoryOrder): uint32 {.inline.} =
+  futex.value.load(order)
+
+proc store*(futex: var Futex, value: uint32, order: MemoryOrder) {.inline.} =
+  futex.value.store(value, order)
+
+proc increment*(futex: var Futex, value: uint32, order: MemoryOrder): uint32 {.inline.} =
+  ## Increment a futex value, returns the previous one.
+  futex.value.fetchAdd(value, order)
+
+proc wait*(futex: var Futex, expected: uint32) {.inline.} =
+  ## Suspend a thread if the value of the futex is the same as expected.
+
+  # Returns 0 in case of a successful suspend
+  # If value are different, it returns EWOULDBLOCK
+  # We discard as this is not needed and simplifies compat with Windows futex
+  discard sysFutex(futex.value.addr, FUTEX_WAIT_PRIVATE, expected)
+
+proc wake*(futex: var Futex) {.inline.} =
+  ## Wake one thread (from the same process)
+
+  # Returns the number of actually woken threads
+  # or a Posix error code (if negative)
+  # We discard as this is not needed and simplifies compat with Windows futex
+  discard sysFutex(futex.value.addr, FUTEX_WAKE_PRIVATE, 1)
+
+proc wakeAll*(futex: var Futex) {.inline.} =
+  ## Wake all threads (from the same process)
+
+  # Returns the number of actually woken threads
+  # or a Posix error code (if negative)
+  # We discard as this is not needed and simplifies compat with Windows futex
+  discard sysFutex(futex.value.addr, FUTEX_WAKE_PRIVATE, high(int32))

taskpools/primitives/futexes_macos.nim

@@ -0,0 +1,114 @@
+# taskpools
+# Copyright (c) 2019 Mamy André-Ratsimbazafy
+# Copyright (c) 2021-2025 Status Research & Development GmbH
+# Licensed and distributed under either of
+#   * MIT license (license terms in the root directory or at http://opensource.org/licenses/MIT).
+#   * Apache v2 license (license terms in the root directory or at http://www.apache.org/licenses/LICENSE-2.0).
+# at your option. This file may not be copied, modified, or distributed except according to those terms.
+
+import std/atomics
+export MemoryOrder
+
+# OS primitives
+# ------------------------------------------------------------------------
+
+# Darwin futexes.
+# https://github.com/odin-lang/Odin/blob/6983813b4ece0e48539dc1d3e7c7437569db1dc2/core/sync/futex_darwin.odin
+
+{.push hint[XDeclaredButNotUsed]: off.}
+
+const UL_COMPARE_AND_WAIT            = 1
+const UL_UNFAIR_LOCK                 = 2
+const UL_COMPARE_AND_WAIT_SHARED     = 3
+const UL_UNFAIR_LOCK64_SHARED        = 4
+const UL_COMPARE_AND_WAIT64          = 5
+const UL_COMPARE_AND_WAIT64_SHARED   = 6
+# obsolete names
+const UL_OSSPINLOCK                  = UL_COMPARE_AND_WAIT
+const UL_HANDOFFLOCK                 = UL_UNFAIR_LOCK
+#  These operation code are only implemented in (DEVELOPMENT || DEBUG) kernels
+const UL_DEBUG_SIMULATE_COPYIN_FAULT = 253
+const UL_DEBUG_HASH_DUMP_ALL         = 254
+const UL_DEBUG_HASH_DUMP_PID         = 255
+
+# operation bits [15, 8] contain the flags for __ulock_wake
+#
+const ULF_WAKE_ALL                   =  0x00000100
+const ULF_WAKE_THREAD                =  0x00000200
+const ULF_WAKE_ALLOW_NON_OWNER       =  0x00000400
+
+# operation bits [23, 16] contain the flags for __ulock_wait
+#
+# @const ULF_WAIT_WORKQ_DATA_CONTENTION
+# The waiter is contending on this lock for synchronization around global data.
+# This causes the workqueue subsystem to not create new threads to offset for
+# waiters on this lock.
+#
+# @const ULF_WAIT_CANCEL_POINT
+# This wait is a cancelation point
+#
+# @const ULF_WAIT_ADAPTIVE_SPIN
+# Use adaptive spinning when the thread that currently holds the unfair lock
+# is on core.
+const ULF_WAIT_WORKQ_DATA_CONTENTION = 0x00010000
+const ULF_WAIT_CANCEL_POINT          = 0x00020000
+const ULF_WAIT_ADAPTIVE_SPIN         = 0x00040000
+
+# operation bits [31, 24] contain the generic flags
+const ULF_NO_ERRNO                   = 0x01000000
+
+# masks
+const UL_OPCODE_MASK                 = 0x000000FF
+const UL_FLAGS_MASK                  = 0xFFFFFF00
+const ULF_GENERIC_MASK               = 0xFFFF0000
+
+const ULF_WAIT_MASK     = ULF_NO_ERRNO or
+                          ULF_WAIT_WORKQ_DATA_CONTENTION or
+                          ULF_WAIT_CANCEL_POINT or
+                          ULF_WAIT_ADAPTIVE_SPIN
+
+const ULF_WAKE_MASK     = ULF_NO_ERRNO or
+                          ULF_WAKE_ALL or
+                          ULF_WAKE_THREAD or
+                          ULF_WAKE_ALLOW_NON_OWNER
+
+proc ulock_wait(operation: uint32, address: pointer, expected: uint64, timeout: uint32): cint {.importc:"__ulock_wait", noconv.}
+proc ulock_wait2(operation: uint32, address: pointer, expected: uint64, timeout, value2: uint64): cint {.importc:"__ulock_wait2", noconv.}
+proc ulock_wake(operation: uint32, address: pointer, wake_value: uint64): cint {.importc:"__ulock_wake", noconv.}
+
+# Futex API
+# ------------------------------------------------------------------------
+
+type
+  Futex* = object
+    value: Atomic[uint32]
+
+proc initialize*(futex: var Futex) {.inline.} =
+  futex.value.store(0, moRelaxed)
+
+proc teardown*(futex: var Futex) {.inline.} =
+  futex.value.store(0, moRelaxed)
+
+proc load*(futex: var Futex, order: MemoryOrder): uint32 {.inline.} =
+  futex.value.load(order)
+
+proc store*(futex: var Futex, value: uint32, order: MemoryOrder) {.inline.} =
+  futex.value.store(value, order)
+
+proc increment*(futex: var Futex, value: uint32, order: MemoryOrder): uint32 {.inline.} =
+  ## Increment a futex value, returns the previous one.
+  futex.value.fetchAdd(value, order)
+
+proc wait*(futex: var Futex, expected: uint32) {.inline.} =
+  ## Suspend a thread if the value of the futex is the same as expected.
+  discard ulock_wait(UL_COMPARE_AND_WAIT or ULF_NO_ERRNO, futex.value.addr, uint64 expected, 0)
+
+proc wake*(futex: var Futex) {.inline.} =
+  ## Wake one thread (from the same process)
+  discard ulock_wake(UL_COMPARE_AND_WAIT or ULF_NO_ERRNO, futex.value.addr, 0)
+
+proc wakeAll*(futex: var Futex) {.inline.} =
+  ## Wake all threads (from the same process)
+  discard ulock_wake(UL_COMPARE_AND_WAIT or ULF_WAKE_ALL or ULF_NO_ERRNO, futex.value.addr, 0)
+
+{.pop.}

taskpools/primitives/futexes_windows.nim

@@ -0,0 +1,75 @@
+# taskpools
+# Copyright (c) 2019 Mamy André-Ratsimbazafy
+# Copyright (c) 2021-2025 Status Research & Development GmbH
+# Licensed and distributed under either of
+#   * MIT license (license terms in the root directory or at http://opensource.org/licenses/MIT).
+#   * Apache v2 license (license terms in the root directory or at http://www.apache.org/licenses/LICENSE-2.0).
+# at your option. This file may not be copied, modified, or distributed except according to those terms.
+
+# An implementation of futex using Windows primitives
+
+# We don't import winlean directly because it pollutes the library with
+# a global variable inet_ntop that stores a proc from "Ws2_32.dll"
+
+import std/atomics
+export MemoryOrder
+
+# OS primitives
+# ------------------------------------------------------------------------
+
+type
+  WinBool* = int32
+    ## WinBool uses opposite convention as posix, != 0 meaning success.
+
+const INFINITE = -1'i32
+
+# Contrary to the documentation, the futex related primitives are NOT in kernel32.dll
+# but in API-MS-Win-Core-Synch-l1-2-0.dll ¯\_(ツ)_/¯
+proc WaitOnAddress(
+        Address: pointer, CompareAddress: pointer,
+        AddressSize: csize_t, dwMilliseconds: int32
+       ): WinBool {.importc, stdcall, dynlib: "API-MS-Win-Core-Synch-l1-2-0.dll".}
+  # The Address should be volatile
+
+proc WakeByAddressSingle(Address: pointer) {.importc, stdcall, dynlib: "API-MS-Win-Core-Synch-l1-2-0.dll".}
+proc WakeByAddressAll(Address: pointer) {.importc, stdcall, dynlib: "API-MS-Win-Core-Synch-l1-2-0.dll".}
+
+# Futex API
+# ------------------------------------------------------------------------
+
+type
+  Futex* = object
+    value: Atomic[uint32]
+
+proc initialize*(futex: var Futex) {.inline.} =
+  futex.value.store(0, moRelaxed)
+
+proc teardown*(futex: var Futex) {.inline.} =
+  futex.value.store(0, moRelaxed)
+
+proc load*(futex: var Futex, order: MemoryOrder): uint32 {.inline.} =
+  futex.value.load(order)
+
+proc store*(futex: var Futex, value: uint32, order: MemoryOrder) {.inline.} =
+  futex.value.store(value, order)
+
+proc increment*(futex: var Futex, value: uint32, order: MemoryOrder): uint32 {.inline.} =
+  ## Increment a futex value, returns the previous one.
+  futex.value.fetchAdd(value, order)
+
+proc wait*(futex: var Futex, expected: uint32) {.inline.} =
+  ## Suspend a thread if the value of the futex is the same as expected.
+
+  # Returns TRUE if the wait succeeds or FALSE if not.
+  # getLastError() will contain the error information, for example
+  # if it failed due to a timeout.
+  # We discard as this is not needed and simplifies compat with Linux futex
+  discard WaitOnAddress(futex.value.addr, expected.addr, csize_t sizeof(expected), INFINITE)
+
+proc wake*(futex: var Futex) {.inline.} =
+  ## Wake one thread (from the same process)
+  WakeByAddressSingle(futex.value.addr)
+
+proc wakeAll*(futex: var Futex) {.inline.} =
+  ## Wake all threads (from the same process)
+  WakeByAddressAll(futex.value.addr)

taskpools/taskpools.nim

@@ -41,7 +41,7 @@ import
   system/ansi_c,
   std/[atomics, cpuinfo, isolation, macros, random, typetraits],
   ./[
-    ast_utils, channels_spsc_single, chase_lev_deques, event_notifiers, flowvars,
+    ast_utils, backoff, chase_lev_deques, flowvars,
     injection_queues, sparsesets,
   ],
   ./primitives/[barriers, allocs],
@@ -80,7 +80,6 @@ type
     currentTask: TaskNode
 
     # Synchronization
-    eventNotifier: ptr EventNotifier # shared event notifier
     signal: ptr Signal               # owned signal
 
     # Thefts
@@ -90,13 +89,13 @@ type
 
   Taskpool* = ptr object
     ## A taskpool schedules procedures to be executed in parallel
-    barrier: SyncBarrier
+    barrier {.align: 64.}: SyncBarrier
       ## Barrier for initialization and teardown
     # --- Align: 64
-    eventNotifier: EventNotifier
-      ## Puts thread to sleep
-
-    numThreads*{.align: 64.}: int
+    globalBackoff {.align: 64.}: EventCount
+      ## Multi-producer multi-consumer backoff
+    # --- Align: 64
+    numThreads* {.align: 64.}: int
     workerDeques: ptr UncheckedArray[ChaseLevDeque[TaskNode]]
       ## Direct access for task stealing
     workers: ptr UncheckedArray[Thread[(Taskpool, WorkerID)]]
@@ -148,7 +147,6 @@ proc setupWorker() =
   ctx.victims.allocate(ctx.taskpool.numThreads)
 
   # Synchronization
-  ctx.eventNotifier = addr ctx.taskpool.eventNotifier
   ctx.signal = addr ctx.taskpool.workerSignals[ctx.id]
   ctx.signal.terminate.store(false, moRelaxed)
 
@@ -210,30 +208,36 @@ proc runTask(tn: var TaskNode) {.inline.} =
   tn.callback(tn.args)
   tn.tp_free()
 
-proc schedule(ctx: WorkerContext, tn: sink TaskNode) {.inline.} =
-  ## Schedule a task in the taskpool
+proc schedule(ctx: WorkerContext, tn: sink TaskNode, forceWake = false) {.inline.} =
+  ## Schedule a task in the taskpool.
+  ## This wakes another worker if our local queue is empty
+  ## or forceWake is true.
   debug: log("Worker %2d: schedule task 0x%.08x (parent 0x%.08x, current 0x%.08x)\n", ctx.id, tn, tn.parent, ctx.currentTask)
+
+  # Instead of notifying every time a task is scheduled, we notify
+  # only when the worker queue is empty. This is a good approximation
+  # of starvation in work-stealing.
+  let wasEmpty = ctx.taskDeque[].peek() == 0
   ctx.taskDeque[].push(tn)
-  ctx.taskpool.eventNotifier.notify()
+  if forceWake or wasEmpty:
+    ctx.taskpool.globalBackoff.wake()
 
 proc submitTask(tp: Taskpool, tn: TaskNode) {.inline.} =
   ## Push a task onto the injection queue from any thread.
   ## Workers will drain the queue into their Chase-Lev deques, making tasks stealable.
-  tp.injectionQueue.push(tn)
-  tp.eventNotifier.notify()
+  var wasEmpty = false
+  tp.injectionQueue.push(tn, wasEmpty)
+  if wasEmpty:
+    # only one wake is needed; the worker will wake one after draining the queue,
+    # the next worker will wake one after steal, and so on.
+    tp.globalBackoff.wake()
 
 proc drainInjectionQueue(ctx: var WorkerContext) {.inline.} =
   ## Atomically claim the entire injection queue and push all tasks into
   ## the calling worker's Chase-Lev deque, where they become stealable.
   ## Only one worker wins the exchange; the others drain nothing.
-  var count = 0
   for node in ctx.taskpool.injectionQueue.drain():
     ctx.taskDeque[].push(node)
-    inc count
-  # Wake workers so the newly stealable tasks get parallel attention.
-  let toWake = min(count, ctx.taskpool.eventNotifier.getParked())
-  for _ in 0 ..< toWake:
-    ctx.taskpool.eventNotifier.notify()
 
 # Scheduler
 # ---------------------------------------------
@@ -257,7 +261,7 @@ proc trySteal(ctx: var WorkerContext): TaskNode =
 
 proc eventLoop(ctx: var WorkerContext) =
   ## Each worker thread executes this loop over and over.
-  while not ctx.signal.terminate.load(moRelaxed):
+  while true:
     # 1. Pick from local deque
     debug: log("Worker %2d: eventLoop 1 - searching task from local deque\n", ctx.id)
     var processed = 0'u32
@@ -265,34 +269,42 @@ proc eventLoop(ctx: var WorkerContext) =
       debug: log("Worker %2d: eventLoop 1 - running task 0x%.08x (parent 0x%.08x, current 0x%.08x)\n", ctx.id, taskNode, taskNode.parent, ctx.currentTask)
       taskNode.runTask()
       inc processed
-      if processed >= TasksBetweenInjectionDrains:
+      if processed >= tasksBetweenInjectionDrains:
         processed = 0
         ctx.drainInjectionQueue()
 
-    # 2. Drain the injection queue into our Chase-Lev deque so externally submitted
-    #    tasks become local work (and stealable by other workers).
+    let ticket = ctx.taskpool.globalBackoff.sleepy()
+
+    # Drain the injection queue into our Chase-Lev deque so externally submitted
+    # tasks become local work (and stealable by other workers).
     ctx.drainInjectionQueue()
 
-    # 3. Re-check local deque; it may now contain injected tasks.
-    var taskNode = ctx.taskDeque[].pop()
-    if not taskNode.isNil:
-      debug: log("Worker %2d: eventLoop 3 - running injected task 0x%.08x\n", ctx.id, taskNode)
+    if (var taskNode = ctx.taskDeque[].pop(); not taskNode.isNil):
+      # 2. Local queue contains injected tasks.
+      debug: log("Worker %2d: eventLoop 2 - running injected task 0x%.08x\n", ctx.id, taskNode)
+      ctx.taskpool.globalBackoff.cancelSleep()
+      ctx.taskpool.globalBackoff.wake()
       taskNode.runTask()
-      continue # back to step 1
-
-    # 4. Run out of tasks, become a thief
-    debug: log("Worker %2d: eventLoop 4 - becoming a thief\n", ctx.id)
-    var stolenTask = ctx.trySteal()
-    if not stolenTask.isNil:
-      # 4.a Run stolen task
-      debug: log("Worker %2d: eventLoop 4.a - stole task 0x%.08x (parent 0x%.08x, current 0x%.08x)\n", ctx.id, stolenTask, stolenTask.parent, ctx.currentTask)
+    elif (var stolenTask = ctx.trySteal(); not stolenTask.isNil):
+      # 3. Run stolen task
+      debug: log("Worker %2d: eventLoop 3 - stole task 0x%.08x (parent 0x%.08x, current 0x%.08x)\n", ctx.id, stolenTask, stolenTask.parent, ctx.currentTask)
+      # We managed to steal a task, cancel sleep
+      ctx.taskpool.globalBackoff.cancelSleep()
+      # Theft successful, there might be more work for idle threads, wake one
+      # cancelSleep must be done before as wake has an optimization
+      # to not notify when a thread is sleepy
+      ctx.taskpool.globalBackoff.wake()
       stolenTask.runTask()
+    elif ctx.signal.terminate.load(moAcquire):
+      # 4. Taskpool has no more tasks and we were signaled to terminate
+      ctx.taskpool.globalBackoff.cancelSleep()
+      debug: log("Worker %2d: eventLoop 4 - terminated\n", ctx.id)
+      break
     else:
-      # 4.b Park the thread until a new task enters the taskpool.
-      #     submitTask calls notify() so parked workers wake when work arrives.
-      debug: log("Worker %2d: eventLoop 4.b - sleeping\n", ctx.id)
-      ctx.eventNotifier[].park()
-      debug: log("Worker %2d: eventLoop 4.b - waking\n", ctx.id)
+      # 5. Park the thread until a new task enters the taskpool
+      debug: log("Worker %2d: eventLoop 5.a - sleeping\n", ctx.id)
+      ctx.taskpool.globalBackoff.sleep(ticket)
+      debug: log("Worker %2d: eventLoop 5.b - waking\n", ctx.id)
 
 # Tasking
 # ---------------------------------------------
@@ -309,7 +321,7 @@ proc forceFuture*[T](fv: Flowvar[T], parentResult: var T) =
   template ctx: untyped = workerContext
 
   template isFutReady(): untyped =
-    fv.chan[].tryRecv(parentResult)
+    fv.tryComplete(parentResult)
 
   if isFutReady():
     return
@@ -327,7 +339,7 @@ proc forceFuture*[T](fv: Flowvar[T], parentResult: var T) =
   while (var taskNode = ctx.taskDeque[].pop(); not taskNode.isNil):
     if taskNode.parent != ctx.currentTask:
       debug: log("Worker %2d: sync 1 - skipping non-direct descendant task 0x%.08x (parent 0x%.08x, current 0x%.08x)\n", ctx.id, taskNode, taskNode.parent, ctx.currentTask)
-      ctx.schedule(taskNode)
+      ctx.schedule(taskNode, forceWake = true) # reschedule task and wake a sibling to take it over.
       break
     debug: log("Worker %2d: sync 1 - running task 0x%.08x (parent 0x%.08x, current 0x%.08x)\n", ctx.id, taskNode, taskNode.parent, ctx.currentTask)
     taskNode.runTask()
@@ -344,6 +356,8 @@ proc forceFuture*[T](fv: Flowvar[T], parentResult: var T) =
     var taskNode = ctx.trySteal()
 
     if not taskNode.isNil:
+      # Theft successful, there might be more work for idle threads, wake one
+      ctx.taskpool.globalBackoff.wake()
       # We stole some task, we hope we advance our awaited task
       debug: log("Worker %2d: sync 2.1 - stole task 0x%.08x (parent 0x%.08x, current 0x%.08x)\n", ctx.id, taskNode, taskNode.parent, ctx.currentTask)
       taskNode.runTask()
@@ -372,45 +386,37 @@ proc syncAll*(tp: Taskpool) =
   preCondition: ctx.currentTask.isRootTask()
 
   # Empty all tasks
-  var foreignThreadsParked = false
-  while not foreignThreadsParked:
+  while true:
     # 1. Empty local tasks
     debug: log("Worker %2d: syncAll 1 - searching task from local deque\n", ctx.id)
     while (var taskNode = ctx.taskDeque[].pop(); not taskNode.isNil):
       debug: log("Worker %2d: syncAll 1 - running task 0x%.08x (parent 0x%.08x, current 0x%.08x)\n", ctx.id, taskNode, taskNode.parent, ctx.currentTask)
       taskNode.runTask()
 
-    # 2. Drain injection queue into local deque so externally submitted tasks
-    #    are not left stranded while we wait for the pool to go idle.
+    # Drain injection queue into local deque so externally submitted tasks
+    # are not left stranded while we wait for the pool to go idle.
     ctx.drainInjectionQueue()
 
-    # 3. Re-check local deque; it may now contain injected tasks.
     if (var taskNode = ctx.taskDeque[].pop(); not taskNode.isNil):
-      debug: log("Worker %2d: syncAll 3 - running injected task 0x%.08x\n", ctx.id, taskNode)
+      # 2. Local queue contains injected tasks.
+      debug: log("Worker %2d: syncAll 2 - running injected task 0x%.08x\n", ctx.id, taskNode)
+      ctx.taskpool.globalBackoff.wake()
       taskNode.runTask()
-      continue # back to step 1
-
-    if tp.numThreads == 1 or foreignThreadsParked:
+    elif (var taskNode = ctx.trySteal(); not taskNode.isNil):
+      # 3. We stole some task
+      debug: log("Worker %2d: syncAll 3 - stole task 0x%.08x (parent 0x%.08x, current 0x%.08x)\n", ctx.id, taskNode, taskNode.parent, ctx.currentTask)
+      # Theft successful, there might be more work for idle threads, wake one
+      ctx.taskpool.globalBackoff.wake()
+      taskNode.runTask()
+    elif tp.globalBackoff.getNumWaiters() == (0'i32, int32(tp.numThreads - 1)):
+      # 4. all threads besides the current are parked (and none are
+      #    in pre-sleep, so none can still grab a task and create work)
+      debugTermination:
+        log("Worker %2d: syncAll 4 - termination, all other threads sleeping\n", ctx.id)
       break
-
-    # 4. Help other threads
-    debug: log("Worker %2d: syncAll 4 - becoming a thief\n", ctx.id)
-    var taskNode = ctx.trySteal()
-
-    if not taskNode.isNil:
-      # 4.1 We stole some task
-      debug: log("Worker %2d: syncAll 4.1 - stole task 0x%.08x (parent 0x%.08x, current 0x%.08x)\n", ctx.id, taskNode, taskNode.parent, ctx.currentTask)
-      taskNode.runTask()
     else:
-      # 4.2 No task to steal
-      if tp.eventNotifier.getParked() == tp.numThreads - 1:
-        # 4.2.1 all threads besides the current are parked
-        debugTermination:
-          log("Worker %2d: syncAll 4.2.1 - termination, all other threads sleeping\n", ctx.id)
-        foreignThreadsParked = true
-      else:
-        # 4.2.2 We don't park as there is no notif for task completion
-        cpuRelax()
+      # 5. We don't park as there is no notif for task completion
+      cpuRelax()
 
   debugTermination:
     log(">>> Worker %2d leaves barrier <<<\n", ctx.id)
@@ -427,7 +433,7 @@ proc new*(T: type Taskpool, numThreads = countProcessors()): T {.raises: [Catcha
   var tp = tp_allocAligned(TpObj, sizeof(TpObj) + 64, 64)
 
   tp.barrier.init(numThreads.int32)
-  tp.eventNotifier.initialize()
+  tp.globalBackoff.initialize()
   tp.numThreads = numThreads
   tp.injectionQueue.init()
   tp.workerDeques = tp_allocArrayAligned(ChaseLevDeque[TaskNode], numThreads, alignment = 64)
@@ -463,7 +469,7 @@ proc cleanup(tp: var Taskpool) =
   tp.workerSignals.tp_freeAligned()
   tp.workers.tp_freeAligned()
   tp.workerDeques.tp_freeAligned()
-  `=destroy`(tp.eventNotifier)
+  `=destroy`(tp.globalBackoff)
   tp.barrier.delete()
 
   tp.tp_freeAligned()
@@ -475,11 +481,9 @@ proc shutdown*(tp: var Taskpool) =
 
   # Signal termination to all threads
   for i in 0 ..< tp.numThreads:
-    tp.workerSignals[i].terminate.store(true, moRelaxed)
+    tp.workerSignals[i].terminate.store(true, moRelease)
 
-  let parked = tp.eventNotifier.getParked()
-  for i in 0 ..< parked:
-    tp.eventNotifier.notify()
+  tp.globalBackoff.wakeAll()
 
   # 1 matching barrier in worker_entry_fn
   discard tp.barrier.wait()

tests/stress/test_shutdown.nim

@@ -0,0 +1,41 @@
+# taskpools
+# Copyright (c) 2021-2026 Status Research & Development GmbH
+# Licensed and distributed under either of
+#   * MIT license (license terms in the root directory or at http://opensource.org/licenses/MIT).
+#   * Apache v2 license (license terms in the root directory or at http://www.apache.org/licenses/LICENSE-2.0).
+# at your option. This file may not be copied, modified, or distributed except according to those terms.
+
+{.push raises: [], gcsafe.}
+
+import
+  unittest2,
+  ../../taskpools
+
+proc nothing() =
+  discard
+
+proc retint(): int =
+  42
+
+# Just test this won't hang; note this is a pathological case
+# threadpools are usually created once per program life-time;
+# it's also very slow, maybe just run it in linux.
+
+# TODO: leaks thread handlers in windows and raises ResourceExhaustedError
+#       https://github.com/nim-lang/Nim/issues/23350
+
+when not defined(windows):
+  suite "Shutdown stress test":
+    test "no arguments, no result":
+      for _ in 0 .. 1_000_000:
+        var tp = Taskpool.new()
+        tp.spawn(nothing())
+        tp.syncAll()
+        tp.shutdown()
+
+    test "result, no arguments":
+      for _ in 0 .. 1_000_000:
+        var tp = Taskpool.new()
+        check sync(tp.spawn(retint())) == 42
+        tp.syncAll()
+        tp.shutdown()

tests/test_all.nim

@@ -8,11 +8,13 @@
 {.warning[UnusedImport]:off .}
 
 import ./[
+  test_backoff,
   test_bpc,
   test_calltypes,
   test_dfs,
   test_external_queue,
   test_fib,
+  test_futexes,
   test_heat,
   test_nqueens,
   test_single_thread,

tests/test_backoff.nim

@@ -0,0 +1,158 @@
+# taskpools
+# Copyright (c) 2021-2026 Status Research & Development GmbH
+# Licensed and distributed under either of
+#   * MIT license (license terms in the root directory or at http://opensource.org/licenses/MIT).
+#   * Apache v2 license (license terms in the root directory or at http://www.apache.org/licenses/LICENSE-2.0).
+# at your option. This file may not be copied, modified, or distributed except according to those terms.
+
+{.push raises: [], gcsafe.}
+
+import
+  std/[atomics, os],
+  unittest2,
+  ../taskpools/backoff
+
+# Tests for the EventCount (the lock-free condition-variable equivalent in
+# backoff.nim). The parking protocol is:
+#
+#   while not condition:
+#     ticket = ec.sleepy()      # announce intent to sleep (pre-wait)
+#     if condition:
+#       ec.cancelSleep()        # bail out, undo the pre-wait
+#       break
+#     ec.sleep(ticket)          # commit and park until wake() bumps the epoch
+#
+# A wake() between sleepy() and sleep() bumps the epoch, invalidating the
+# ticket so sleep() returns without parking. That is how lost wakeups are
+# avoided, and we test it directly (single-threaded, deterministic).
+
+const
+  observeWindowMs = 100
+  maxParkedSpins = 10_000
+
+type
+  ParkState = object
+    ec: EventCount
+    reached: Atomic[bool]
+    condition: Atomic[bool]
+    spins: Atomic[int]         # outer-loop iterations while unsignalled
+    woke: Atomic[bool]
+
+  WakeAllState = object
+    ec: EventCount
+    condition: Atomic[bool]
+    woke: Atomic[int]
+
+proc spinUntilBool(a: var Atomic[bool], expected: bool) =
+  while a.load(moAcquire) != expected:
+    discard
+
+proc parker(s: ptr ParkState) {.thread.} =
+  s.reached.store(true, moRelease)
+  while not s.condition.load(moAcquire):
+    let ticket = s.ec.sleepy()
+    if s.condition.load(moAcquire):
+      s.ec.cancelSleep()
+      break
+    s.ec.sleep(ticket)
+    discard s.spins.fetchAdd(1, moRelaxed)
+  s.woke.store(true, moRelease)
+
+proc multiParker(s: ptr WakeAllState) {.thread.} =
+  while not s.condition.load(moAcquire):
+    let ticket = s.ec.sleepy()
+    if s.condition.load(moAcquire):
+      s.ec.cancelSleep()
+      break
+    s.ec.sleep(ticket)
+  discard s.woke.fetchAdd(1, moRelease)
+
+suite "EventCount":
+  test "sleepy() then cancelSleep() leaves no waiters":
+    var ec: EventCount
+    ec.initialize()
+
+    check ec.getNumWaiters().preSleep == 0
+    check ec.getNumWaiters().committedSleep == 0
+    discard ec.sleepy()
+    check ec.getNumWaiters().preSleep == 1
+    check ec.getNumWaiters().committedSleep == 0
+    ec.cancelSleep()
+    check ec.getNumWaiters().preSleep == 0
+    check ec.getNumWaiters().committedSleep == 0
+
+  test "sleep() does not park when the ticket is stale":
+    # wake() between sleepy() and sleep() bumps the epoch; sleep() must observe
+    # the change and return immediately instead of blocking forever.
+    var ec: EventCount
+    ec.initialize()
+
+    let ticket = ec.sleepy()
+    ec.wake()                       # invalidates the ticket's epoch
+    ec.sleep(ticket)                # would hang if it parked on the stale epoch
+
+    check ec.getNumWaiters().preSleep == 0
+    check ec.getNumWaiters().committedSleep == 0
+
+  test "sleep() parks the thread until wake()":
+    var s: ParkState
+    s.ec.initialize()
+
+    var thr: Thread[ptr ParkState]
+    createThread(thr, parker, addr s)
+
+    # This is racy but if the futex does not
+    # wait and return immediately, it should register
+    # more than maxParkedSpins in observeWindowMs.
+    spinUntilBool(s.reached, true)
+    sleep(observeWindowMs)
+    check s.spins.load(moAcquire) < maxParkedSpins
+    check not s.woke.load(moAcquire)
+
+    s.condition.store(true, moRelease)
+    s.ec.wake()
+    joinThread(thr)
+
+    check s.woke.load(moAcquire)
+    check s.ec.getNumWaiters().preSleep == 0
+    check s.ec.getNumWaiters().committedSleep == 0
+
+  test "wakeAll() releases every parked waiter":
+    const numWaiters = 4
+    var s: WakeAllState
+    s.ec.initialize()
+
+    var threads: array[numWaiters, Thread[ptr WakeAllState]]
+    for t in mitems(threads):
+      createThread(t, multiParker, addr s)
+
+    while s.ec.getNumWaiters().committedSleep != numWaiters:
+      discard
+
+    s.condition.store(true, moRelease)
+    s.ec.wakeAll()
+    joinThreads(threads)
+
+    check s.woke.load(moAcquire) == numWaiters
+    check s.ec.getNumWaiters().preSleep == 0
+    check s.ec.getNumWaiters().committedSleep == 0
+
+  test "supports more than 256 committed waiters":
+    const numWaiters = 257
+    var s: WakeAllState
+    s.ec.initialize()
+
+    var threads = newSeq[Thread[ptr WakeAllState]](numWaiters)
+    for t in mitems(threads):
+      createThread(t, multiParker, addr s)
+
+    while s.ec.getNumWaiters().committedSleep != numWaiters:
+      discard
+
+    s.condition.store(true, moRelease)
+    s.ec.wakeAll()
+    joinThreads(threads)
+
+    check s.woke.load(moAcquire) == numWaiters
+    check s.ec.getNumWaiters().preSleep == 0
+    check s.ec.getNumWaiters().committedSleep == 0

tests/test_external_queue.nim

@@ -30,13 +30,13 @@ proc submitter(ctx: Context) {.thread.} =
   for i in 0 ..< ctx.numTasks:
     ctx.tp.spawn work(ctx.executed)
 
-proc work2(): int =
+proc workInt(): int =
   123
 
 proc submitterFv(ctx: Context) {.thread.} =
   var futs = newSeq[Flowvar[int]]()
   for i in 0 ..< ctx.numTasks:
-    futs.add ctx.tp.spawn work2()
+    futs.add ctx.tp.spawn workInt()
   for fut in futs:
     doAssert sync(fut) == 123
     discard ctx.executed[].fetchAdd(1, moRelaxed)
@@ -108,3 +108,73 @@ suite "External threads task queue":
     joinThreads(threads)
     tp.syncAll()
     check executed.load(moAcquire) == externalThreads * tasksPerThread
+
+# submitTask only wakes a worker on the injection queue's empty->non-empty
+# transition. Unlike an internal spawn (whose task's owner is live and always
+# drains its own deque before parking), an injected task has no owning thread:
+# its only consumers are the pool workers, which may all be parked. So the
+# empty->non-empty wake is the sole guarantee that a consumer shows up. A lost
+# wake there is a liveness bug (a hang), invisible to ThreadSanitizer, so we
+# exercise it directly. A small pool is used so workers park quickly between
+# submissions, maximizing the empty->non-empty edges.
+
+proc pingPong(ctx: Context) {.thread.} =
+  ## Submit one task and block until it completes before submitting the next.
+  ## Each round takes the injection queue empty -> non-empty -> empty, so a
+  ## worker must be woken from park every round; a lost wake hangs here.
+  for i in 0 ..< ctx.numTasks:
+    let fv = ctx.tp.spawn workInt()
+    doAssert sync(fv) == 123
+    discard ctx.executed[].fetchAdd(1, moRelaxed)
+
+suite "External threads injection wake edge":
+  setup:
+    # Small pool so workers park quickly between injections, driving the
+    # empty->non-empty transition submitTask's wake optimization hinges on.
+    var tp = Taskpool.new(2)
+
+  teardown:
+    tp.syncAll()
+    tp.shutdown()
+
+  test "ping-pong; externalThreads=1; rounds=50_000":
+    # Deterministic edge: the queue is provably empty between rounds, so every
+    # submission must wake a parked worker.
+    const
+      externalThreads = 1
+      rounds = 50_000
+    var executed: Atomic[int]
+    var threads = newSeq[Thread[Context]](externalThreads)
+    for t in mitems(threads):
+      createThread(t, pingPong, (tp, rounds, addr executed))
+    joinThreads(threads)
+    tp.syncAll()
+    check executed.load(moAcquire) == externalThreads * rounds
+
+  test "ping-pong; externalThreads=8; rounds=20_000":
+    # Concurrent submitters contend on the injection queue head while it churns
+    # empty <-> non-empty under a 2-worker pool.
+    const
+      externalThreads = 8
+      rounds = 20_000
+    var executed: Atomic[int]
+    var threads = newSeq[Thread[Context]](externalThreads)
+    for t in mitems(threads):
+      createThread(t, pingPong, (tp, rounds, addr executed))
+    joinThreads(threads)
+    tp.syncAll()
+    check executed.load(moAcquire) == externalThreads * rounds
+
+  test "trickle; externalThreads=8; tasksPerThread=100_000":
+    # High-throughput fire-and-forget submissions onto a small pool: the queue
+    # repeatedly drains to empty and is re-armed under heavy contention.
+    const
+      externalThreads = 8
+      tasksPerThread = 100_000
+    var executed: Atomic[int]
+    var threads = newSeq[Thread[Context]](externalThreads)
+    for t in mitems(threads):
+      createThread(t, submitter, (tp, tasksPerThread, addr executed))
+    joinThreads(threads)
+    tp.syncAll()
+    check executed.load(moAcquire) == externalThreads * tasksPerThread

tests/test_futexes.nim

@@ -0,0 +1,93 @@
+# taskpools
+# Copyright (c) 2021-2026 Status Research & Development GmbH
+# Licensed and distributed under either of
+#   * MIT license (license terms in the root directory or at http://opensource.org/licenses/MIT).
+#   * Apache v2 license (license terms in the root directory or at http://www.apache.org/licenses/LICENSE-2.0).
+# at your option. This file may not be copied, modified, or distributed except according to those terms.
+
+{.push raises: [], gcsafe.}
+
+import
+  std/[atomics, os],
+  unittest2,
+  ../taskpools/primitives/futexes
+
+const
+  observeWindowMs = 100
+  maxParkedSpins = 10_000
+
+type
+  WaitState = object
+    futex: Futex
+    reachedWait: Atomic[bool]  # waiter has entered its wait loop
+    spins: Atomic[int]         # times wait() returned while still unsignalled
+    woke: Atomic[bool]         # waiter observed the signal and left the loop
+
+  WakeAllState = object
+    futex: Futex
+    ready: Atomic[int]         # count of waiters that entered their wait loop
+    woke: Atomic[int]          # count of waiters released after the signal
+
+proc spinUntil[T](a: var Atomic[T], expected: T) =
+  while a.load(moAcquire) != expected:
+    discard
+
+proc waiter(s: ptr WaitState) {.thread.} =
+  s.reachedWait.store(true, moRelease)
+  while s.futex.load(moAcquire) == 0:
+    s.futex.wait(0)
+    discard s.spins.fetchAdd(1, moRelaxed)
+  s.woke.store(true, moRelease)
+
+proc wakeAllWaiter(s: ptr WakeAllState) {.thread.} =
+  discard s.ready.fetchAdd(1, moRelease)
+  while s.futex.load(moAcquire) == 0:
+    s.futex.wait(0)
+  discard s.woke.fetchAdd(1, moRelease)
+
+suite "Futex":
+  test "wait() parks the thread until wake()":
+    var s: WaitState
+    s.futex.initialize()
+
+    var thr: Thread[ptr WaitState]
+    createThread(thr, waiter, addr s)
+
+    # This is racy but if the futex does not
+    # wait and return immediately, it should register
+    # more than maxParkedSpins in observeWindowMs.
+    spinUntil(s.reachedWait, true)
+    sleep(observeWindowMs)
+    check s.spins.load(moAcquire) < maxParkedSpins
+    check not s.woke.load(moAcquire)
+
+    s.futex.store(1, moRelease)
+    s.futex.wake()
+    joinThread(thr)
+    check s.woke.load(moAcquire)
+
+    s.futex.teardown()
+
+  test "wait() returns immediately when value != expected":
+    var futex: Futex
+    futex.initialize()
+    futex.store(1, moRelease)
+    futex.wait(0)  # won't hang because value != expected
+    futex.teardown()
+
+  test "wakeAll() releases every parked waiter":
+    const numWaiters = 4
+    var s: WakeAllState
+    s.futex.initialize()
+
+    var threads: array[numWaiters, Thread[ptr WakeAllState]]
+    for t in mitems(threads):
+      createThread(t, wakeAllWaiter, addr s)
+
+    spinUntil(s.ready, numWaiters)
+    s.futex.store(1, moRelease)
+    s.futex.wakeAll()
+    joinThreads(threads)
+
+    check s.woke.load(moAcquire) == numWaiters
+    s.futex.teardown()