From 79c5b75803563f6346996253c0d164fee717fb5e Mon Sep 17 00:00:00 2001 From: Araq Date: Sun, 7 Jun 2026 17:29:39 +0200 Subject: [PATCH 1/3] virtual threads --- lib/std/private/threadtypes.nim | 61 +++++++++++++ lib/std/typedthreads.nim | 142 +++++++++++++++++++++--------- lib/system/threadimpl.nim | 4 +- lib/system/threadlocalstorage.nim | 8 +- 4 files changed, 167 insertions(+), 48 deletions(-) diff --git a/lib/std/private/threadtypes.nim b/lib/std/private/threadtypes.nim index a1cdf21dc0f77..f1305cd91ae7e 100644 --- a/lib/std/private/threadtypes.nim +++ b/lib/std/private/threadtypes.nim @@ -1,5 +1,7 @@ include system/inclrtl +import std/private/syslocks + const hasSharedHeap* = defined(boehmgc) or defined(gogc) # don't share heaps; every thread has its own when defined(windows): @@ -161,10 +163,69 @@ type const hasAllocStack* = defined(zephyr) # maybe freertos too? +# ---------------- Virtual-thread primitives ---------------- +# +# BinSem is a reusable binary semaphore — `wait` blocks until `post` has been +# called (or returns immediately if `post` arrived first); on wake it clears +# the flag so the same instance can be cycled across many dispatches. +# Defined here (rather than in threadpool_impl.nim) because `Thread[TArg]` +# embeds a `ThreadBase` that contains one. + +type + BinSem* = object + L: SysLock + C: SysCond + signaled: bool + + ThreadBase* = object + ## Per-virtual-thread control block: done semaphore + currently-assigned + ## worker. Embedded in `Thread[TArg]`. `worker` is held as an opaque + ## `pointer` because the `Worker` type lives in `threadpool_impl.nim`, + ## which is included downstream of this file. + done*: BinSem + worker*: pointer # ptr Worker (opaque here) + +proc initBinSem*(s: var BinSem) {.inline.} = + initSysLock(s.L) + initSysCond(s.C) + s.signaled = false + +proc deinitBinSem*(s: var BinSem) {.inline.} = + deinitSys(s.L) + deinitSysCond(s.C) + +proc postBinSem*(s: var BinSem) = + acquireSys(s.L) + s.signaled = true + signalSysCond(s.C) + releaseSys(s.L) + +proc waitBinSem*(s: var BinSem) = + acquireSys(s.L) + while not s.signaled: + waitSysCond(s.C, s.L) + s.signaled = false + releaseSys(s.L) + +proc resetBinSem*(s: var BinSem) = + acquireSys(s.L) + s.signaled = false + releaseSys(s.L) + +proc initThreadBase*(t: var ThreadBase) {.inline.} = + initBinSem(t.done) + t.worker = nil + type Thread*[TArg] = object core*: PGcThread sys*: SysThread + base*: ptr ThreadBase # heap-allocated so copies of Thread[TArg] (e.g. + # joinThread's by-value parameter) keep pointing + # at the SAME done-semaphore. Allocated by + # createThread; never freed under the current + # design (V control blocks are small and + # typically program-lifetime). when TArg is void: dataFn*: proc () {.nimcall, gcsafe.} else: diff --git a/lib/std/typedthreads.nim b/lib/std/typedthreads.nim index 998d95c10063c..8e5adaad6604e 100644 --- a/lib/std/typedthreads.nim +++ b/lib/std/typedthreads.nim @@ -76,7 +76,7 @@ deinitLock(l) ]## -import std/private/[threadtypes] +import std/private/[threadtypes, syslocks] export Thread import system/ansi_c @@ -84,6 +84,11 @@ import system/ansi_c when defined(nimPreviewSlimSystem): import std/assertions +# Virtual-thread worker pool primitives. Included as a file (not imported) +# so it inherits the OS thread bindings (`pthread_create`, `createThread` on +# Windows, etc.) already in scope from `threadtypes`. +include private/threadpool_impl + when defined(genode): import genode/env @@ -147,6 +152,31 @@ else: proc threadProcWrapper[TArg](closure: pointer): pointer {.noconv.} = result = nil nimThreadProcWrapperBody(closure) + +# Per-TArg adapter for the virtual-thread pool. The worker calls this through +# the `TaskEntry` function pointer; the closure is `addr(Thread[TArg])`. +# Per-worker GC bootstrap (initGC, threadType, globalsSlot, setStackBottom) +# happens once in `workerMain` before any task runs, so this entry just runs +# the user proc. Note: `deallocOsPages` / `deallocThreadStorage(core)` are +# absent — the worker is immortal and reuses TLS across virtual threads. +proc poolTaskEntry[TArg](closure: pointer) {.nimcall, gcsafe, raises: [].} = + let thrd = cast[ptr Thread[TArg]](closure) + try: + when TArg is void: + thrd.dataFn() + else: + when defined(nimV2): + thrd.dataFn(thrd.data) + else: + var x = default(TArg) + deepCopy(x, thrd.data) + thrd.dataFn(x) + except: + when declared(threadTrouble): + threadTrouble() + finally: + when hasAllocStack: + deallocThreadStorage(thrd.rawStack) {.pop.} proc running*[TArg](t: Thread[TArg]): bool {.inline.} = @@ -162,18 +192,24 @@ when hostOS == "windows": proc joinThread*[TArg](t: Thread[TArg]) {.inline.} = ## Waits for the thread `t` to finish. - discard waitForSingleObject(t.sys, -1'i32) + when defined(noThreadReuse): + discard waitForSingleObject(t.sys, -1'i32) + else: + poolWaitDone(t.base) proc joinThreads*[TArg](t: varargs[Thread[TArg]]) = ## Waits for every thread in `t` to finish. - var a: array[MAXIMUM_WAIT_OBJECTS, SysThread] = default(array[MAXIMUM_WAIT_OBJECTS, SysThread]) - var k = 0 - while k < len(t): - var count = min(len(t) - k, MAXIMUM_WAIT_OBJECTS) - for i in 0..(count - 1): a[i] = t[i + k].sys - discard waitForMultipleObjects(int32(count), - cast[ptr SysThread](addr(a)), 1, -1) - inc(k, MAXIMUM_WAIT_OBJECTS) + when defined(noThreadReuse): + var a: array[MAXIMUM_WAIT_OBJECTS, SysThread] = default(array[MAXIMUM_WAIT_OBJECTS, SysThread]) + var k = 0 + while k < len(t): + var count = min(len(t) - k, MAXIMUM_WAIT_OBJECTS) + for i in 0..(count - 1): a[i] = t[i + k].sys + discard waitForMultipleObjects(int32(count), + cast[ptr SysThread](addr(a)), 1, -1) + inc(k, MAXIMUM_WAIT_OBJECTS) + else: + for i in 0..t.high: joinThread(t[i]) elif defined(genode): proc joinThread*[TArg](t: Thread[TArg]) {.importcpp.} @@ -186,7 +222,10 @@ elif defined(genode): else: proc joinThread*[TArg](t: Thread[TArg]) {.inline.} = ## Waits for the thread `t` to finish. - discard pthread_join(t.sys, nil) + when defined(noThreadReuse): + discard pthread_join(t.sys, nil) + else: + poolWaitDone(t.base) proc joinThreads*[TArg](t: varargs[Thread[TArg]]) = ## Waits for every thread in `t` to finish. @@ -217,16 +256,25 @@ when hostOS == "windows": ## Entry point is the proc `tp`. ## `param` is passed to `tp`. `TArg` can be `void` if you ## don't need to pass any data to the thread. - t.core = cast[PGcThread](allocThreadStorage(sizeof(GcThread))) - - when TArg isnot void: t.data = param - t.dataFn = tp - when hasSharedHeap: t.core.stackSize = ThreadStackSize - var dummyThreadId: int32 = 0'i32 - t.sys = createThread(nil, ThreadStackSize, threadProcWrapper[TArg], - addr(t), 0'i32, dummyThreadId) - if t.sys <= 0: - raise newException(ResourceExhaustedError, "cannot create thread") + when defined(noThreadReuse): + t.core = cast[PGcThread](allocThreadStorage(sizeof(GcThread))) + + when TArg isnot void: t.data = param + t.dataFn = tp + when hasSharedHeap: t.core.stackSize = ThreadStackSize + var dummyThreadId: int32 = 0'i32 + t.sys = createThread(nil, ThreadStackSize, threadProcWrapper[TArg], + addr(t), 0'i32, dummyThreadId) + if t.sys <= 0: + raise newException(ResourceExhaustedError, "cannot create thread") + else: + when TArg isnot void: t.data = param + t.dataFn = tp + initThreadBase(t.base) + poolDispatch(cast[ptr ThreadBase](addr t.base), poolTaskEntry[TArg], addr t) + # Surface the underlying worker's OS handle (matches today's + # post-createThread semantics; goes stale once the V completes). + t.sys = workerOsThread(cast[ptr Worker](t.base.worker)) proc pinToCpu*[Arg](t: var Thread[Arg]; cpu: Natural) = ## Pins a thread to a `CPU`:idx:. @@ -266,28 +314,38 @@ else: ## Entry point is the proc `tp`. `param` is passed to `tp`. ## `TArg` can be `void` if you ## don't need to pass any data to the thread. - t.core = cast[PGcThread](allocThreadStorage(sizeof(GcThread))) - - when TArg isnot void: t.data = param - t.dataFn = tp - when hasSharedHeap: t.core.stackSize = ThreadStackSize - var a {.noinit.}: Pthread_attr - doAssert pthread_attr_init(a) == 0 - when hasAllocStack: - var - rawstk = allocThreadStorage(ThreadStackSize + StackGuardSize) - stk = cast[pointer](cast[uint](rawstk) + StackGuardSize) - let setstacksizeResult = pthread_attr_setstack(addr a, stk, ThreadStackSize) - t.rawStack = rawstk + when defined(noThreadReuse): + t.core = cast[PGcThread](allocThreadStorage(sizeof(GcThread))) + + when TArg isnot void: t.data = param + t.dataFn = tp + when hasSharedHeap: t.core.stackSize = ThreadStackSize + var a {.noinit.}: Pthread_attr + doAssert pthread_attr_init(a) == 0 + when hasAllocStack: + var + rawstk = allocThreadStorage(ThreadStackSize + StackGuardSize) + stk = cast[pointer](cast[uint](rawstk) + StackGuardSize) + let setstacksizeResult = pthread_attr_setstack(addr a, stk, ThreadStackSize) + t.rawStack = rawstk + else: + let setstacksizeResult = pthread_attr_setstacksize(a, ThreadStackSize) + + when not defined(ios): + # This fails on iOS + doAssert(setstacksizeResult == 0) + if pthread_create(t.sys, a, threadProcWrapper[TArg], addr(t)) != 0: + raise newException(ResourceExhaustedError, "cannot create thread") + doAssert pthread_attr_destroy(a) == 0 else: - let setstacksizeResult = pthread_attr_setstacksize(a, ThreadStackSize) - - when not defined(ios): - # This fails on iOS - doAssert(setstacksizeResult == 0) - if pthread_create(t.sys, a, threadProcWrapper[TArg], addr(t)) != 0: - raise newException(ResourceExhaustedError, "cannot create thread") - doAssert pthread_attr_destroy(a) == 0 + when TArg isnot void: t.data = param + t.dataFn = tp + if t.base == nil: + t.base = cast[ptr ThreadBase](c_malloc(csize_t sizeof(ThreadBase))) + zeroMem(t.base, sizeof(ThreadBase)) + initThreadBase(t.base[]) + poolDispatch(t.base, poolTaskEntry[TArg], addr t) + t.sys = workerOsThread(cast[ptr Worker](t.base.worker)) proc pinToCpu*[Arg](t: var Thread[Arg]; cpu: Natural) = ## Pins a thread to a `CPU`:idx:. diff --git a/lib/system/threadimpl.nim b/lib/system/threadimpl.nim index dcd1b267a03b6..5667d89d3a043 100644 --- a/lib/system/threadimpl.nim +++ b/lib/system/threadimpl.nim @@ -10,12 +10,12 @@ when not defined(useNimRtl): when declared(initGC): initGC() when not emulatedThreadVars: - type ThreadType {.pure.} = enum + type ThreadType* {.pure.} = enum None = 0, NimThread = 1, ForeignThread = 2 var - threadType {.rtlThreadVar.}: ThreadType + threadType* {.rtlThreadVar.}: ThreadType threadType = ThreadType.NimThread diff --git a/lib/system/threadlocalstorage.nim b/lib/system/threadlocalstorage.nim index 625e602bf5c32..acff19739a83f 100644 --- a/lib/system/threadlocalstorage.nim +++ b/lib/system/threadlocalstorage.nim @@ -6,7 +6,7 @@ when defined(windows): proc threadVarAlloc(): ThreadVarSlot {. importc: "TlsAlloc", stdcall, header: "".} - proc threadVarSetValue(dwTlsIndex: ThreadVarSlot, lpTlsValue: pointer) {. + proc threadVarSetValue*(dwTlsIndex: ThreadVarSlot, lpTlsValue: pointer) {. importc: "TlsSetValue", stdcall, header: "".} proc tlsGetValue(dwTlsIndex: ThreadVarSlot): pointer {. importc: "TlsGetValue", stdcall, header: "".} @@ -44,7 +44,7 @@ elif defined(genode): var mainTls: pointer - proc threadVarSetValue(s: ThreadVarSlot, value: pointer) {.inline.} = + proc threadVarSetValue*(s: ThreadVarSlot, value: pointer) {.inline.} = if offMainThread(): threadVarSetValue(value); else: @@ -90,7 +90,7 @@ else: proc threadVarAlloc(): ThreadVarSlot {.inline.} = result = default(ThreadVarSlot) discard pthread_key_create(addr(result), nil) - proc threadVarSetValue(s: ThreadVarSlot, value: pointer) {.inline.} = + proc threadVarSetValue*(s: ThreadVarSlot, value: pointer) {.inline.} = discard pthread_setspecific(s, value) proc threadVarGetValue(s: ThreadVarSlot): pointer {.inline.} = result = pthread_getspecific(s) @@ -104,7 +104,7 @@ when emulatedThreadVars: when emulatedThreadVars: - var globalsSlot: ThreadVarSlot + var globalsSlot*: ThreadVarSlot when not defined(useNimRtl): var mainThread: GcThread From fd24e427036f2dc67cb0655a6d5b28a320e24578 Mon Sep 17 00:00:00 2001 From: Araq Date: Mon, 27 Jul 2026 06:13:02 +0200 Subject: [PATCH 2/3] removed pointless comment --- lib/std/typedthreads.nim | 3 --- 1 file changed, 3 deletions(-) diff --git a/lib/std/typedthreads.nim b/lib/std/typedthreads.nim index 8e5adaad6604e..9d7198b4723b4 100644 --- a/lib/std/typedthreads.nim +++ b/lib/std/typedthreads.nim @@ -84,9 +84,6 @@ import system/ansi_c when defined(nimPreviewSlimSystem): import std/assertions -# Virtual-thread worker pool primitives. Included as a file (not imported) -# so it inherits the OS thread bindings (`pthread_create`, `createThread` on -# Windows, etc.) already in scope from `threadtypes`. include private/threadpool_impl when defined(genode): From 4caba86162ade4d37f3d0790df96d300931eb338 Mon Sep 17 00:00:00 2001 From: Araq Date: Mon, 27 Jul 2026 06:16:57 +0200 Subject: [PATCH 3/3] added missing file --- lib/std/private/threadpool_impl.nim | 190 ++++++++++++++++++++++++++++ 1 file changed, 190 insertions(+) create mode 100644 lib/std/private/threadpool_impl.nim diff --git a/lib/std/private/threadpool_impl.nim b/lib/std/private/threadpool_impl.nim new file mode 100644 index 0000000000000..f0841f87afbeb --- /dev/null +++ b/lib/std/private/threadpool_impl.nim @@ -0,0 +1,190 @@ +# +# +# Nim's Runtime Library +# (c) Copyright 2026 Andreas Rumpf +# +# See the file "copying.txt", included in this +# distribution, for details about the copyright. +# + +## Virtual-thread worker pool. +## +## Included from `std/typedthreads`. +## +## Workers never exit. createThread pops one off the idle stack (or +## pthread_create / CreateThread a new one when the stack is empty), hands +## it a task via a per-worker BinSem, and the worker runs it, pushes itself +## back onto the idle stack, and posts the requesting virtual thread's +## `done` flag. joinThread waits on that flag — `pthread_join` is not used. + +{.push stackTrace: off.} + +const + PoolWorkerStackMask = 1024 * 256 * sizeof(int) - 1 + PoolWorkerStackSize* = PoolWorkerStackMask + 1 - 4096 + ## Matches typedthreads' default ThreadStackSize for non-embedded targets. + +type + TaskEntry* = proc (closure: pointer) {.nimcall, gcsafe, raises: [].} + ## Worker-side entry: static function pointer plus opaque closure. + ## typedthreads' adapter builds the closure from `TArg`. + + Worker* = object + osThread: SysThread + wake: BinSem + # Per-worker TLS-emulation storage. With virtual threads on this worker + # all `var x {.threadvar.}` resolve to fields in `gcThread.tls` — so the + # storage persists across V's on the same worker. This is what makes + # `var allocator {.threadvar.}: MemRegion` keep its chunks across + # createThread/joinThread cycles. + gcThread: GcThread + # Task slot. Written by `poolDispatch` while the worker is parked on + # `wake`; read by `workerMain` once it wakes. + entry: TaskEntry + closure: pointer + base: ptr ThreadBase + next: ptr Worker + +# ---------------- idle stack ---------------- + +var + poolLock: SysLock + idleTop: ptr Worker + +initSysLock(poolLock) + +proc pushIdle(w: ptr Worker) = + acquireSys(poolLock) + w.next = idleTop + idleTop = w + releaseSys(poolLock) + +proc popIdle(): ptr Worker = + acquireSys(poolLock) + result = idleTop + if result != nil: + idleTop = result.next + result.next = nil + releaseSys(poolLock) + +# ---------------- worker main ---------------- + +proc workerMain(arg: pointer) {.nimcall, gcsafe, raises: [].} = + let w = cast[ptr Worker](arg) + + # One-shot per-worker setup. The stack-bottom mark is a frame in + # workerMain — every task call frame sits at a higher (deeper) address, + # so the bottom we register here remains valid for the worker's whole + # process-lifetime. Emulated-TLS bootstrap (`globalsSlot`) is done + # earlier, inside the C-callable thunk, since any Nim-convention proc + # call may touch threadvars in its prologue. + when not defined(boehmgc) and not defined(gogc) and not defined(gcRegions) and + not defined(gcDestructors) and not defined(gcHooks): + # Mirrors `usesDestructors` in system.nim (not exported, so duplicated). + var stackMark {.volatile.}: pointer + nimGC_setStackBottom(addr(stackMark)) + when declared(initGC): + initGC() + when declared(threadType): + threadType = ThreadType.NimThread + + while true: + waitBinSem(w.wake) + # Snapshot the task slot locally before pushing back to idle, since the + # slot can be overwritten as soon as another caller pops this worker. + let entry = w.entry + let closure = w.closure + let base = w.base + try: + entry(closure) + except CatchableError: + discard + # Per-V destruction handlers fire here and the seq is cleared so V+1 + # starts with a fresh handler list (decision 1). + when declared(nimThreadDestructionHandlers): + for i in countdown(nimThreadDestructionHandlers.len-1, 0): + try: nimThreadDestructionHandlers[i]() + except CatchableError: discard + nimThreadDestructionHandlers.setLen 0 + # `pushIdle` BEFORE `postBinSem(done)` so a join-then-redispatch can land + # on the same worker. Reversing order is correctness-preserving but + # silently degrades reuse. + pushIdle(w) + postBinSem(base.done) + +# ---------------- platform-specific spawn ---------------- + +when defined(windows): + proc workerThunkWin(arg: pointer): int32 {.stdcall.} = + # Bind the emulated-TLS slot to this worker's GcThread BEFORE any + # nimcall proc runs — `workerMain` (and anything it calls) may touch + # threadvars in its function prologue, which under emulated TLS would + # deref a nil slot if globalsSlot hadn't been set for this OS thread. + when declared(globalsSlot): + let w = cast[ptr Worker](arg) + threadVarSetValue(globalsSlot, addr(w.gcThread)) + workerMain(arg) + result = 0'i32 + + proc spawnOsWorker(w: ptr Worker) = + var dummy: int32 = 0'i32 + let h = createThread(nil, PoolWorkerStackSize.int32, workerThunkWin, + cast[pointer](w), 0'i32, dummy) + if h <= 0: + raise newException(ResourceExhaustedError, "cannot create pool worker") + w.osThread = h + +elif defined(genode): + # Genode keeps the legacy 1:1 model until its C++-side runtime adapts. + proc spawnOsWorker(w: ptr Worker) = + raise newException(ResourceExhaustedError, + "virtual-thread pool not implemented for Genode") + +else: + proc workerThunkPosix(arg: pointer): pointer {.noconv.} = + # Bind the emulated-TLS slot to this worker's GcThread BEFORE any + # nimcall proc runs — see workerThunkWin for the rationale. + when declared(globalsSlot): + let w = cast[ptr Worker](arg) + threadVarSetValue(globalsSlot, addr(w.gcThread)) + workerMain(arg) + result = nil + + proc spawnOsWorker(w: ptr Worker) = + var attr: Pthread_attr + discard pthread_attr_init(attr) + discard pthread_attr_setstacksize(attr, PoolWorkerStackSize) + if pthread_create(w.osThread, attr, workerThunkPosix, cast[pointer](w)) != 0: + raise newException(ResourceExhaustedError, "cannot create pool worker") + discard pthread_attr_destroy(attr) + +proc spawnWorker(): ptr Worker = + result = cast[ptr Worker](c_malloc(csize_t sizeof(Worker))) + zeroMem(result, sizeof(Worker)) + initBinSem(result.wake) + spawnOsWorker(result) + +proc workerOsThread*(w: ptr Worker): SysThread {.inline.} = + ## Read accessor for `Thread[TArg].sys` plumbing. + w.osThread + +# ---------------- public API ---------------- + +proc poolDispatch*(base: ptr ThreadBase; entry: TaskEntry; closure: pointer) = + ## Hand `entry(closure)` to a worker. Demand-grown: pops the idle stack + ## first, only spawns a fresh OS thread when the stack is empty. + resetBinSem(base.done) + var w = popIdle() + if w == nil: + w = spawnWorker() + w.entry = entry + w.closure = closure + w.base = base + base.worker = cast[pointer](w) + postBinSem(w.wake) + +proc poolWaitDone*(base: ptr ThreadBase) = + ## Block until the most recent `poolDispatch` on `base` completes. + waitBinSem(base.done) + +{.pop.}