ch41.md 5.1 KB
Chapter 41. Threading
There are two ways to introduce parallelism in Nim: A rather low-level API for thread creation where a thread of execution is directly provided by the operating system, and a high-level API for annotating potential parallelism that the runtime is free to exploit, but does not have to. The reason for this is that sending a task to a different processor to execute might be more expensive than executing it on the current processor immediately. A Nim program starts to run in the so-called “main thread”. The main thread runs every top level statement. In order to run things in parallel at least one additional thread needs to be created. 41.1. createThread The low-level API is centered around a Thread[T] type. A Thread[T] runs a proc of the type proc (x: T), in other words: the T is the type of a single parameter than can be passed to the proc at thread creation: proc worker(s: string) {.thread.} = 1
echo s
var background: Thread[string] 2 createThread background, worker, "abc" 3 echo "xyz" 4 joinThread background 5 echo "control flow converged" 6 1 The proc worker runs in parallel with the main thread. So it is annotated
as a thread proc. The thread pragma is an alias for gcsafe.
2 Declares a variable named background of type Thread of string. 3 Creates a single background thread and attaches it to a variable named
background.
257
4 Outputs "xyz" but competes with the output of the background thread. 5 Waits for the background thread to finish execution. 6 At this point the background thread has finished and control flow
continues in the main thread only.
This program can either produce "abc" followed by "xyz" or "xyz" followed by "abc". It is not deterministic because both the background thread and the main thread compete for the stdout stream resource that echo uses and one threads gets to output its message first. 41.2. Single worker, single channel Creating a thread is an expensive operation and so usually one tries to keep a thread running and give it more than one task. This is typically accomplished by having the thread run a loop that waits on a queue or channel for items to work on: var chan: Channel[string] 1 chan.open() 2 proc worker() {.thread.} = 3 while true: 4
let task = chan.recv() 5
if task.len == 0: break 6
echo task 7
proc log(msg: string) = 8 chan.send msg var logger: Thread[void] 9 createThread logger, worker 10 log "a" 11 log "b" log "c" log "" 12 joinThread logger 13 chan.close() 14 1 Declares a channel named chan of type Channel of string. 2 A channel has to be opened before something can be sent. 3 The proc worker runs in parallel with the main thread. So it is annotated
as a thread proc. The thread pragma is an alias for gcsafe.
258 4 The while true loop and break inside the loop is a common idiom when processing channel items. 5 Blocks until an item arrives at chan. 6 In this rather adhoc protocol an empty string indicates that the thread should stop processing. 7 Processes the task. 8 Logging a message is as simple as sending the string to chan. 9 The thread does not need any data at startup so a Thread of void is used. 10 Attaches the worker proc to the logger thread variable. 11 The strings "a" then "b" then "c" should be logged. 12 Sends the empty string to tell the background thread to shut down. 13 Waits for the background to finish. 14 A channel must be closed explicitly to free its resources.
259
41.3. Multiple workers, single channel So far we have only used a single background thread. Together with the main thread we can only make effective use of 2 CPU cores. In order to make use of all available cores an array of worker threads can be used. The following program outlines how this can be done: var chan: Channel[string] chan.open() proc worker(threadIndex: int) {.thread.} = 1
while true:
let task = chan.recv()
if task.len == 0: break
echo "Thread ", threadIndex, ": ", task 2
proc log(msg: string) =
chan.send msg
var loggers: array[8, Thread[int]] 3 for i in 0 ..< loggers.len:
createThread loggers[i], worker, i 4
log "a" 5 log "b" log "c" for i in 0 ..< loggers.len:
log "" 6
joinThreads loggers 7 chan.close() 1 A thread index is passed to every worker thread. 2 The thread index is echoed in addition to task. 3 Declares a thread pool named loggers with 8 entries of type Thread[int]. 4 Starts up every thread in the array loggers. The index i is passed to the
threadIndex parameter.
5 Logs some messages. 6 Sends every thread in the thread pool the "should stop" message. 7 Waits for all threads in the thread pool to stop execution.