From 79c5b75803563f6346996253c0d164fee717fb5e Mon Sep 17 00:00:00 2001 From: Araq Date: Sun, 7 Jun 2026 17:29:39 +0200 Subject: [PATCH] virtual threads --- lib/std/private/threadtypes.nim | 61 +++++++++++++ lib/std/typedthreads.nim | 138 +++++++++++++++++++++--------- lib/system/threadimpl.nim | 4 +- lib/system/threadlocalstorage.nim | 8 +- 4 files changed, 165 insertions(+), 46 deletions(-) diff --git a/lib/std/private/threadtypes.nim b/lib/std/private/threadtypes.nim index a1cdf21dc0..f1305cd91a 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 998d95c100..8e5adaad66 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 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") + 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 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 + 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 dcd1b267a0..5667d89d3a 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 625e602bf5..acff19739a 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