diff options
Diffstat (limited to 'lib/pure/concurrency/threadpool.nim')
| -rw-r--r-- | lib/pure/concurrency/threadpool.nim | 187 |
1 files changed, 92 insertions, 95 deletions
diff --git a/lib/pure/concurrency/threadpool.nim b/lib/pure/concurrency/threadpool.nim index 1f7df7c00..06ed2fe54 100644 --- a/lib/pure/concurrency/threadpool.nim +++ b/lib/pure/concurrency/threadpool.nim @@ -7,20 +7,25 @@ # distribution, for details about the copyright. # -## Implements Nim's `spawn <manual_experimental.html#parallel-amp-spawn>`_. -## -## **See also:** -## * `threads module <threads.html>`_ -## * `channels module <channels.html>`_ -## * `locks module <locks.html>`_ -## * `asyncdispatch module <asyncdispatch.html>`_ +{.deprecated: "use the nimble packages `malebolgia`, `taskpools` or `weave` instead".} + +## Implements Nim's `parallel & spawn statements <manual_experimental.html#parallel-amp-spawn>`_. ## ## Unstable API. +## +## See also +## ======== +## * `threads module <typedthreads.html>`_ for basic thread support +## * `locks module <locks.html>`_ for locks and condition variables +## * `asyncdispatch module <asyncdispatch.html>`_ for asynchronous IO when not compileOption("threads"): {.error: "Threadpool requires --threads:on option.".} -import cpuinfo, cpuload, locks, os +import std/[cpuinfo, cpuload, locks, os] + +when defined(nimPreviewSlimSystem): + import std/[assertions, typedthreads, sysatomics] {.push stackTrace:off.} @@ -51,38 +56,35 @@ proc signal(cv: var Semaphore) = release(cv.L) signal(cv.c) -const CacheLineSize = 32 # true for most archs +const CacheLineSize = 64 # true for most archs type - Barrier {.compilerProc.} = object + Barrier {.compilerproc.} = object entered: int cv: Semaphore # Semaphore takes 3 words at least - when sizeof(int) < 8: - cacheAlign: array[CacheLineSize-4*sizeof(int), byte] - left: int - cacheAlign2: array[CacheLineSize-sizeof(int), byte] - interest: bool # whether the master is interested in the "all done" event + left {.align(CacheLineSize).}: int + interest {.align(CacheLineSize).} : bool # whether the master is interested in the "all done" event -proc barrierEnter(b: ptr Barrier) {.compilerProc, inline.} = +proc barrierEnter(b: ptr Barrier) {.compilerproc, inline.} = # due to the signaling between threads, it is ensured we are the only # one with access to 'entered' so we don't need 'atomicInc' here: inc b.entered # also we need no 'fence' instructions here as soon 'nimArgsPassingDone' # will be called which already will perform a fence for us. -proc barrierLeave(b: ptr Barrier) {.compilerProc, inline.} = +proc barrierLeave(b: ptr Barrier) {.compilerproc, inline.} = atomicInc b.left when not defined(x86): fence() # We may not have seen the final value of b.entered yet, # so we need to check for >= instead of ==. if b.interest and b.left >= b.entered: signal(b.cv) -proc openBarrier(b: ptr Barrier) {.compilerProc, inline.} = +proc openBarrier(b: ptr Barrier) {.compilerproc, inline.} = b.entered = 0 b.left = 0 b.interest = false -proc closeBarrier(b: ptr Barrier) {.compilerProc.} = +proc closeBarrier(b: ptr Barrier) {.compilerproc.} = fence() if b.left != b.entered: b.cv.initSemaphore() @@ -101,8 +103,8 @@ type cv: Semaphore idx: int - FlowVarBase* = ref FlowVarBaseObj ## Untyped base class for ``FlowVar[T]``. - FlowVarBaseObj = object of RootObj + FlowVarBase* = ref FlowVarBaseObj ## Untyped base class for `FlowVar[T] <#FlowVar>`_. + FlowVarBaseObj {.acyclic.} = object of RootObj ready, usesSemaphore, awaited: bool cv: Semaphore # for 'blockUntilAny' support ai: ptr AwaitInfo @@ -111,10 +113,10 @@ type # be RootRef here otherwise the wrong GC keeps track of it! owner: pointer # ptr Worker - FlowVarObj[T] = object of FlowVarBaseObj + FlowVarObj[T] {.acyclic.} = object of FlowVarBaseObj blob: T - FlowVar*{.compilerProc.}[T] = ref FlowVarObj[T] ## A data flow variable. + FlowVar*[T] {.compilerproc.} = ref FlowVarObj[T] ## A data flow variable. ToFreeQueue = object len: int @@ -137,8 +139,8 @@ type const threadpoolWaitMs {.intdefine.}: int = 100 -proc blockUntil*(fv: FlowVarBase) = - ## Waits until the value for the ``fv`` arrives. +proc blockUntil*(fv: var FlowVarBaseObj) = + ## Waits until the value for `fv` arrives. ## ## Usually it is not necessary to call this explicitly. if fv.usesSemaphore and not fv.awaited: @@ -185,7 +187,7 @@ proc attach(fv: FlowVarBase; i: int): bool = result = false release(fv.cv.L) -proc finished(fv: FlowVarBase) = +proc finished(fv: var FlowVarBaseObj) = doAssert fv.ai.isNil, "flowVar is still attached to an 'blockUntilAny'" # we have to protect against the rare cases where the owner of the flowVar # simply disregards the flowVar and yet the "flowVar" has not yet written @@ -208,16 +210,18 @@ proc finished(fv: FlowVarBase) = # the worker thread waits for "data" to be set to nil before shutting down owner.data = nil -proc fvFinalizer[T](fv: FlowVar[T]) = finished(fv) +proc `=destroy`[T](fv: var FlowVarObj[T]) = + finished(fv) + `=destroy`(fv.blob) -proc nimCreateFlowVar[T](): FlowVar[T] {.compilerProc.} = - new(result, fvFinalizer) +proc nimCreateFlowVar[T](): FlowVar[T] {.compilerproc.} = + new(result) -proc nimFlowVarCreateSemaphore(fv: FlowVarBase) {.compilerProc.} = +proc nimFlowVarCreateSemaphore(fv: FlowVarBase) {.compilerproc.} = fv.cv.initSemaphore() fv.usesSemaphore = true -proc nimFlowVarSignal(fv: FlowVarBase) {.compilerProc.} = +proc nimFlowVarSignal(fv: FlowVarBase) {.compilerproc.} = if fv.ai != nil: acquire(fv.ai.cv.L) fv.ai.idx = fv.idx @@ -228,60 +232,51 @@ proc nimFlowVarSignal(fv: FlowVarBase) {.compilerProc.} = signal(fv.cv) proc awaitAndThen*[T](fv: FlowVar[T]; action: proc (x: T) {.closure.}) = - ## Blocks until the ``fv`` is available and then passes its value - ## to ``action``. + ## Blocks until `fv` is available and then passes its value + ## to `action`. ## - ## Note that due to Nim's parameter passing semantics this - ## means that ``T`` doesn't need to be copied so ``awaitAndThen`` can - ## sometimes be more efficient than `^ proc <#^,FlowVar[T]>`_. - blockUntil(fv) - when T is string or T is seq: + ## Note that due to Nim's parameter passing semantics, this + ## means that `T` doesn't need to be copied, so `awaitAndThen` can + ## sometimes be more efficient than the `^ proc <#^,FlowVar[T]>`_. + blockUntil(fv[]) + when defined(nimV2): + action(fv.blob) + elif T is string or T is seq: action(cast[T](fv.data)) elif T is ref: {.error: "'awaitAndThen' not available for FlowVar[ref]".} else: action(fv.blob) - finished(fv) + finished(fv[]) proc unsafeRead*[T](fv: FlowVar[ref T]): ptr T = ## Blocks until the value is available and then returns this value. - blockUntil(fv) - result = cast[ptr T](fv.data) - finished(fv) - -proc `^`*[T](fv: FlowVar[ref T]): ref T = - ## Blocks until the value is available and then returns this value. - blockUntil(fv) - let src = cast[ref T](fv.data) + blockUntil(fv[]) when defined(nimV2): - result = src + result = cast[ptr T](fv.blob) else: - deepCopy result, src - finished(fv) + result = cast[ptr T](fv.data) + finished(fv[]) proc `^`*[T](fv: FlowVar[T]): T = ## Blocks until the value is available and then returns this value. - blockUntil(fv) - when T is string or T is seq: - let src = cast[T](fv.data) - when defined(nimV2): - result = src - else: - deepCopy result, src + blockUntil(fv[]) + when not defined(nimV2) and (T is string or T is seq or T is ref): + deepCopy result, cast[T](fv.data) else: result = fv.blob - finished(fv) + finished(fv[]) proc blockUntilAny*(flowVars: openArray[FlowVarBase]): int = - ## Awaits any of the given ``flowVars``. Returns the index of one ``flowVar`` + ## Awaits any of the given `flowVars`. Returns the index of one `flowVar` ## for which a value arrived. ## - ## A ``flowVar`` only supports one call to ``blockUntilAny`` at the same time. - ## That means if you ``blockUntilAny([a,b])`` and ``blockUntilAny([b,c])`` - ## the second call will only block until ``c``. If there is no ``flowVar`` left + ## A `flowVar` only supports one call to `blockUntilAny` at the same time. + ## That means if you `blockUntilAny([a,b])` and `blockUntilAny([b,c])` + ## the second call will only block until `c`. If there is no `flowVar` left ## to be able to wait on, -1 is returned. ## - ## **Note**: This results in non-deterministic behaviour and should be avoided. + ## **Note:** This results in non-deterministic behaviour and should be avoided. var ai: AwaitInfo ai.cv.initSemaphore() var conflicts = 0 @@ -302,9 +297,9 @@ proc blockUntilAny*(flowVars: openArray[FlowVarBase]): int = destroySemaphore(ai.cv) proc isReady*(fv: FlowVarBase): bool = - ## Determines whether the specified ``FlowVarBase``'s value is available. + ## Determines whether the specified `FlowVarBase`'s value is available. ## - ## If ``true``, awaiting ``fv`` will not block. + ## If `true`, awaiting `fv` will not block. if fv.usesSemaphore and not fv.awaited: acquire(fv.cv.L) result = fv.cv.counter > 0 @@ -312,31 +307,31 @@ proc isReady*(fv: FlowVarBase): bool = else: result = true -proc nimArgsPassingDone(p: pointer) {.compilerProc.} = +proc nimArgsPassingDone(p: pointer) {.compilerproc.} = let w = cast[ptr Worker](p) signal(w.taskStarted) const - MaxThreadPoolSize* = 256 ## Maximum size of the thread pool. 256 threads - ## should be good enough for anybody ;-) - MaxDistinguishedThread* = 32 ## Maximum number of "distinguished" threads. + MaxThreadPoolSize* {.intdefine.} = 256 ## Maximum size of the thread pool. 256 threads + ## should be good enough for anybody ;-) + MaxDistinguishedThread* {.intdefine.} = 32 ## Maximum number of "distinguished" threads. type - ThreadId* = range[0..MaxDistinguishedThread-1] + ThreadId* = range[0..MaxDistinguishedThread-1] ## A thread identifier. var currentPoolSize: int maxPoolSize = MaxThreadPoolSize minPoolSize = 4 - gSomeReady : Semaphore + gSomeReady: Semaphore readyWorker: ptr Worker # A workaround for recursion deadlock issue # https://github.com/nim-lang/Nim/issues/4597 var numSlavesLock: Lock - numSlavesRunning {.guard: numSlavesLock}: int - numSlavesWaiting {.guard: numSlavesLock}: int + numSlavesRunning {.guard: numSlavesLock.}: int + numSlavesWaiting {.guard: numSlavesLock.}: int isSlave {.threadvar.}: bool numSlavesLock.initLock @@ -409,7 +404,7 @@ proc setMinPoolSize*(size: range[1..MaxThreadPoolSize]) = proc setMaxPoolSize*(size: range[1..MaxThreadPoolSize]) = ## Sets the maximum thread pool size. The default value of this - ## is ``MaxThreadPoolSize`` (256). + ## is `MaxThreadPoolSize <#MaxThreadPoolSize>`_. maxPoolSize = size if currentPoolSize > maxPoolSize: for i in maxPoolSize..currentPoolSize-1: @@ -449,43 +444,45 @@ proc setup() = for i in 0..<currentPoolSize: activateWorkerThread(i) proc preferSpawn*(): bool = - ## Use this proc to determine quickly if a ``spawn`` or a direct call is + ## Use this proc to determine quickly if a `spawn` or a direct call is ## preferable. ## - ## If it returns ``true``, a ``spawn`` may make sense. In general - ## it is not necessary to call this directly; use `spawnX template + ## If it returns `true`, a `spawn` may make sense. In general + ## it is not necessary to call this directly; use the `spawnX template ## <#spawnX.t>`_ instead. result = gSomeReady.counter > 0 -proc spawn*(call: typed): void {.magic: "Spawn".} - ## Always spawns a new task, so that the ``call`` is never executed on +proc spawn*(call: sink typed) {.magic: "Spawn".} = + ## Always spawns a new task, so that the `call` is never executed on ## the calling thread. ## - ## ``call`` has to be proc call ``p(...)`` where ``p`` is gcsafe and has a - ## return type that is either ``void`` or compatible with ``FlowVar[T]``. + ## `call` has to be a proc call `p(...)` where `p` is gcsafe and has a + ## return type that is either `void` or compatible with `FlowVar[T]`. + discard "It uses `nimSpawn3` internally" -proc pinnedSpawn*(id: ThreadId; call: typed): void {.magic: "Spawn".} - ## Always spawns a new task on the worker thread with ``id``, so that - ## the ``call`` is **always** executed on the thread. +proc pinnedSpawn*(id: ThreadId; call: sink typed) {.magic: "Spawn".} = + ## Always spawns a new task on the worker thread with `id`, so that + ## the `call` is **always** executed on the thread. ## - ## ``call`` has to be proc call ``p(...)`` where ``p`` is gcsafe and has a - ## return type that is either ``void`` or compatible with ``FlowVar[T]``. + ## `call` has to be a proc call `p(...)` where `p` is gcsafe and has a + ## return type that is either `void` or compatible with `FlowVar[T]`. + discard "It uses `nimSpawn4` internally" -template spawnX*(call): void = +template spawnX*(call) = ## Spawns a new task if a CPU core is ready, otherwise executes the ## call in the calling thread. ## - ## Usually it is advised to use `spawn proc <#spawn,typed>`_ in order to - ## not block the producer for an unknown amount of time. + ## Usually, it is advised to use the `spawn proc <#spawn,sinktyped>`_ + ## in order to not block the producer for an unknown amount of time. ## - ## ``call`` has to be proc call ``p(...)`` where ``p`` is gcsafe and has a - ## return type that is either 'void' or compatible with ``FlowVar[T]``. + ## `call` has to be a proc call `p(...)` where `p` is gcsafe and has a + ## return type that is either 'void' or compatible with `FlowVar[T]`. (if preferSpawn(): spawn call else: call) proc parallel*(body: untyped) {.magic: "Parallel".} ## A parallel section can be used to execute a block in parallel. ## - ## ``body`` has to be in a DSL that is a particular subset of the language. + ## `body` has to be in a DSL that is a particular subset of the language. ## ## Please refer to `the manual <manual_experimental.html#parallel-amp-spawn>`_ ## for further information. @@ -496,7 +493,7 @@ var initLock stateLock -proc nimSpawn3(fn: WorkerProc; data: pointer) {.compilerProc.} = +proc nimSpawn3(fn: WorkerProc; data: pointer) {.compilerproc.} = # implementation of 'spawn' that is used by the code generator. while true: if selectWorker(readyWorker, fn, data): return @@ -581,7 +578,7 @@ var initLock distinguishedLock -proc nimSpawn4(fn: WorkerProc; data: pointer; id: ThreadId) {.compilerProc.} = +proc nimSpawn4(fn: WorkerProc; data: pointer; id: ThreadId) {.compilerproc.} = acquire(distinguishedLock) if not distinguishedData[id].initialized: activateDistinguishedThread(id) @@ -592,7 +589,7 @@ proc nimSpawn4(fn: WorkerProc; data: pointer; id: ThreadId) {.compilerProc.} = proc sync*() = - ## A simple barrier to wait for all ``spawn``'ed tasks. + ## A simple barrier to wait for all `spawn`ed tasks. ## ## If you need more elaborate waiting, you have to use an explicit barrier. while true: |