From 9adfbb264d879862f609cafc7d99eb382e66a043 Mon Sep 17 00:00:00 2001 From: nitely Date: Wed, 22 Jul 2026 02:15:05 -0300 Subject: [PATCH 1/3] Fix futexes_linux wait/wake syscall number --- constantine.nimble | 2 + .../threadpool/primitives/futexes_linux.nim | 4 +- tests/threadpool/t_backoff.nim | 157 ++++++++++++++++++ tests/threadpool/t_futexes.nim | 97 +++++++++++ 4 files changed, 258 insertions(+), 2 deletions(-) create mode 100644 tests/threadpool/t_backoff.nim create mode 100644 tests/threadpool/t_futexes.nim diff --git a/constantine.nimble b/constantine.nimble index fe51baca5..99774eb90 100644 --- a/constantine.nimble +++ b/constantine.nimble @@ -701,6 +701,8 @@ const testDescThreadpool: seq[string] = @[ "benchmarks-threadpool/histogram_2D/threadpool_histogram.nim", "benchmarks-threadpool/logsumexp/threadpool_logsumexp.nim", "tests/threadpool/t_257_threads.nim", + "tests/threadpool/t_backoff.nim", + "tests/threadpool/t_futexes.nim", ] const testDescMultithreadedCrypto: seq[string] = @[ diff --git a/constantine/threadpool/primitives/futexes_linux.nim b/constantine/threadpool/primitives/futexes_linux.nim index 1e362bc79..5e68416ef 100644 --- a/constantine/threadpool/primitives/futexes_linux.nim +++ b/constantine/threadpool/primitives/futexes_linux.nim @@ -16,11 +16,11 @@ export MemoryOrder # ------------------------------------------------------------------------ const - NR_Futex = 202 - FUTEX_WAIT_PRIVATE = 128 FUTEX_WAKE_PRIVATE = 129 +var NR_Futex {.importc: "SYS_futex", header: "".}: clong + proc syscall(sysno: clong): cint {.importc, header:"", varargs.} proc sysFutex( diff --git a/tests/threadpool/t_backoff.nim b/tests/threadpool/t_backoff.nim new file mode 100644 index 000000000..a379dfcac --- /dev/null +++ b/tests/threadpool/t_backoff.nim @@ -0,0 +1,157 @@ +# 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, unittest], + constantine/threadpool/crossthread/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 diff --git a/tests/threadpool/t_futexes.nim b/tests/threadpool/t_futexes.nim new file mode 100644 index 000000000..60429d0c9 --- /dev/null +++ b/tests/threadpool/t_futexes.nim @@ -0,0 +1,97 @@ +# 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, unittest], + constantine/threadpool/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 s: WaitState + s.futex.initialize() + s.futex.store(1, moRelease) + + var thr: Thread[ptr WaitState] + createThread(thr, waiter, addr s) + joinThread(thr) + + check s.woke.load(moAcquire) + s.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() From 0dfd10d5afca6b5260f43cf0637bd4c56f9f69db Mon Sep 17 00:00:00 2001 From: Esteban C Borsani Date: Wed, 22 Jul 2026 02:24:31 -0300 Subject: [PATCH 2/3] Update constantine/threadpool/primitives/futexes_linux.nim Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com> --- constantine/threadpool/primitives/futexes_linux.nim | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/constantine/threadpool/primitives/futexes_linux.nim b/constantine/threadpool/primitives/futexes_linux.nim index 5e68416ef..c07e927c4 100644 --- a/constantine/threadpool/primitives/futexes_linux.nim +++ b/constantine/threadpool/primitives/futexes_linux.nim @@ -19,7 +19,7 @@ const FUTEX_WAIT_PRIVATE = 128 FUTEX_WAKE_PRIVATE = 129 -var NR_Futex {.importc: "SYS_futex", header: "".}: clong +let NR_Futex {.importc: "SYS_futex", header: "".}: clong proc syscall(sysno: clong): cint {.importc, header:"", varargs.} From e5dc4e2a2f5ef4796acb6d0986b170cd1c787e95 Mon Sep 17 00:00:00 2001 From: nitely Date: Wed, 22 Jul 2026 03:16:27 -0300 Subject: [PATCH 3/3] wip --- tests/threadpool/t_futexes.nim | 15 +++++---------- 1 file changed, 5 insertions(+), 10 deletions(-) diff --git a/tests/threadpool/t_futexes.nim b/tests/threadpool/t_futexes.nim index 60429d0c9..384f3c2ed 100644 --- a/tests/threadpool/t_futexes.nim +++ b/tests/threadpool/t_futexes.nim @@ -68,16 +68,11 @@ suite "Futex": s.futex.teardown() test "wait() returns immediately when value != expected": - var s: WaitState - s.futex.initialize() - s.futex.store(1, moRelease) - - var thr: Thread[ptr WaitState] - createThread(thr, waiter, addr s) - joinThread(thr) - - check s.woke.load(moAcquire) - s.futex.teardown() + 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