std/threadpool

Implements Nim's parallel & spawn statements.

Unstable API.

See also

• threads module for basic thread support
• locks module for locks and condition variables
• asyncdispatch module for asynchronous IO

Imports

cpuinfo, cpuload, locks, os

Types

FlowVar[T] {.compilerproc.} = ref FlowVarObj[T]

A data flow variable. Source Edit

FlowVarBase = ref FlowVarBaseObj

Untyped base class for FlowVar[T]. Source Edit

ThreadId = range[0 .. MaxDistinguishedThread - 1]

A thread identifier. Source Edit

Consts

MaxDistinguishedThread {.intdefine.} = 32

Maximum number of "distinguished" threads. Source Edit

MaxThreadPoolSize {.intdefine.} = 256

Maximum size of the thread pool. 256 threads should be good enough for anybody ;-) Source Edit

Procs

proc `^`[T](fv: FlowVar[T]): T

Blocks until the value is available and then returns this value. Source Edit

proc awaitAndThen[T](fv: FlowVar[T]; action: proc (x: T) {.closure.})

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 the ^ proc.

Source Edit

proc blockUntil(fv: var FlowVarBaseObj) {....raises: [], tags: [], forbids: [].}

Waits until the value for fv arrives.

Usually it is not necessary to call this explicitly.

Source Edit

proc blockUntilAny(flowVars: openArray[FlowVarBase]): int {....raises: [],
    tags: [], forbids: [].}

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 to be able to wait on, -1 is returned.

Note: This results in non-deterministic behaviour and should be avoided.

Source Edit

proc isReady(fv: FlowVarBase): bool {....raises: [], tags: [], forbids: [].}

Determines whether the specified FlowVarBase's value is available.

If true, awaiting fv will not block.

Source Edit

proc parallel(body: untyped) {.magic: "Parallel", ...raises: [], tags: [],
                               forbids: [].}

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.

Please refer to the manual for further information.

Source Edit

proc pinnedSpawn(id: ThreadId; call: sink typed) {.magic: "Spawn", ...raises: [],
    tags: [], forbids: [].}

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 a proc call p(...) where p is gcsafe and has a return type that is either void or compatible with FlowVar[T].

Source Edit

proc preferSpawn(): bool {....raises: [], tags: [], forbids: [].}

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 the spawnX template instead.

Source Edit

proc setMaxPoolSize(size: range[1 .. MaxThreadPoolSize]) {....raises: [], tags: [],
    forbids: [].}

Sets the maximum thread pool size. The default value of this is MaxThreadPoolSize. Source Edit

proc setMinPoolSize(size: range[1 .. MaxThreadPoolSize]) {....raises: [], tags: [],
    forbids: [].}

Sets the minimum thread pool size. The default value of this is 4. Source Edit

proc spawn(call: sink typed) {.magic: "Spawn", ...raises: [], tags: [], forbids: [].}

Always spawns a new task, so that the call is never executed on the calling thread.

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].

Source Edit

proc sync() {....raises: [], tags: [TimeEffect], forbids: [].}

A simple barrier to wait for all spawned tasks.

If you need more elaborate waiting, you have to use an explicit barrier.

Source Edit

proc unsafeRead[T](fv: FlowVar[ref T]): ptr T

Blocks until the value is available and then returns this value. Source Edit

Templates

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 the spawn proc in order to not block the producer for an unknown amount of time.

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].

Source Edit