Chapter 42. spawn

The Channel and Thread types are available since version 1 of Nim and are widely used. It is common to declare these as global variables. In fact it is highly recommended to do so! While global variables are usually discouraged, in this case they are justified: There is no aliasing possible between global variables so it is easy to see how many threads are used and which channels they use and how the flow of communication looks. The topology of the program is clear. A more elegant mechanism that abstracts away the nitty-gritty details of manual thread pool creation and task delivery is available via spawn: spawn f(args) runs f(args) in parallel or concurrently or potentially in parallel, depending on the implementation. spawn should not be confused with createThread — it wraps the f(args) call in a task and passes this task to some already existing thread pool. So the f should not be a long-running function that receives data from a channel, it should simply be an operation that is expensive enough to be worthwhile to run in parallel. But depending on the current load of the machine it might not actually run in parallel. Variations of spawn are implemented in different third-party libraries. We focus here on the spawn implementation that is provided by the “Malebolgia” library. Malebolgia is the successor of std / threadpool and is particularly simple and effective. (Run the command nimble install malebolgia to install it.) Malebolgia only supports “structured concurrency” which means there is a clear point in the source code where all parallel flows of control converged. In Malebolgia this is at the end of an awaitAll environment:

import malebolgia 1 proc f(i: int) = echo i 2 var m = createMaster() 3 m.awaitAll: 4

for i in 0 ..< 10:
  m.spawn f(i)         5

# all tasks are complete 6 1 Imports the malebolgia dependency that this example requires. 2 f is what we run in parallel. 3 Declares a variable m of type Master which is used for synchronization. 4 Waits until all tasks are completed. 5 Creates a task that runs f(i) on the hidden thread pool. 6 By construction we know that all tasks have been completed at this point. If a spawned task raises an exception, the master object notices and rethrows the exception after awaitAll. If multiple tasks raise an exception only the first exception is kept and rethrown. Please be aware that a spawn in Malebolgia is a hint. The library is allowed to ignore the request for parallelism. In other words the above program might execute as if spawn and awaitAll would not exist: proc f(i: int) = echo i for i in 0 ..< 10:

f(i)

42.1. Return values In some implementations of spawn, if the spawned function has a return value of type T the type of spawn f is a FlowVar of type T. FlowVar is short for “data flow variable”. It has the interesting property that data races are prevented by construction: A data flow variable can be written to only once and a read operation blocks until it was written to. In other words, a read operation synchronizes. But the awaitAll operation also synchronizes! Malebolgia takes this insight to 262 do away with a FlowVar wrapping type. This is particularly useful when every spawn should write to a distinct array location, because we can keep working with seq[T] as opposed to seq[FlowVar[T]]. The following example is a standard benchmark for parallel programming and demonstrates this benefit: import malebolgia proc dfs(depth, breadth: int): int {.gcsafe.} = 1

if depth == 0: return 1
var sums = newSeq[int](breadth) 2
var m = createMaster() 3
m.awaitAll: 4
  for i in 0 ..< breadth:
    m.spawn dfs(depth - 1, breadth) -> sums[i] 5
result = 0
for i in 0 ..< breadth:
  result += sums[i] 6

echo dfs(8, 8) 7 1 The dfs proc needs to be annotated with gcsafe manually because it is

recursive. Reminder: gcsafe means that it does not access global variables
that use managed memory.

2 We collect the results of the subtasks in the seq sums. 3 Creates a Master object m for task coordination. 4 Synchronizes all spawned tasks using an awaitAll block. 5 Spawns subtasks recursively and stores the result in sums[i]. For an

explanation of the arrow →, see the text below.

6 After the awaitAll operation we can read from sums without any locking

or synchronization.

7 Shows how to invoke dfs and output its result. The most important aspect of this example is the → notation: Symbols that denote the target location of the spawn (sums in this case) are treated specially in Malebolgia. The awaitAll macro ensures that these symbols are only written to and are not used in any context that could imply a read operation. The macro detects simple “read/write” and “write/write” conflicts.

42.2. Sharing memory A spawned task might run on a different thread than the calling thread or it might not. The reason is that the task creation step can be more expensive than running the code directly. This implies that an operation that waits for an event to occur and is scheduled before the operation that triggers this event can lead to a program making no progress. A special case of this scenario can happen with channels: If the recv operation is scheduled before a send operation and both operations are scheduled to run on the same thread. In fact, channels are much more low level than people realize: every send must be paired with a corresponding recv operation and yet across most (if not all) programming languages and libraries there is no static check to ensure this! Instead, memory can be shared and locks should then be used to ensure that no data races can happen. Malebolgia offers a type Locker[T] that wraps a container of type T and enforces proper locking operations. The wrapped value can be accessed as lock x as y where x is the Locker object and y is a fresh identifier that denotes the wrapped object, or it can be accessed as unprotected x as y when the wrapped object should be accessed without a locking operation. Inside a concurrently running operation one needs to use lock but after the awaitAll we can use unprotected: import std / [strutils, tables] import malebolgia import malebolgia / lockers proc countWords(filename: string; results: Locker[CountTable[string]]) = 1

for w in splitWhitespace(readFile(filename)): 2
  lock results as r: 3
    r.inc w 4

proc main() =

var m = createMaster()
var results = initLocker initCountTable[string]() 5
m.awaitAll:
  m.spawn countWords("fileA.txt", results) 6
  m.spawn countWords("fileB.txt", results)
unprotected results as r: 7
  r.sort() 8
  echo r

264 main() 1 countWords takes a Locker[CountTable[string]] which is comparable to a

var CountTable[string].

2 Iterates over every word of the input file. A “word” is a substring

separated by whitespace.

3 Acquires results's attached lock and accesses the CountTable[string]

under the name of r. The lock is released after the block of code that is
passed to the lock macro.

4 Under the protection of the lock, tell the CountTable to count the word w. 5 Creates an object of type Locker[CountTable[string]] and binds it to the

name results.

6 Runs countWords for two different files in parallel. 7 Accesses the underlying CountTable as r without any locking. This is safe

because we know that after the awaitAll operation all parallel processing
is complete.

8 Sorts the CountTable so that the most commonly used words come first in

the output.

There are few restrictions on what values can be shared between threads and thus what can be wrapped in a Locker but as usual pointers and the nonrestrictive aliasing they allow for cause trouble. For example, the following program easily subverts the protection of the lock: proc example(results: Locker[ptr int]) =

var x: ptr int
lock results as r:
  x = r # create an alias
# store outside of the lock:
x[] = 13

The situation is worse for ref. A ref T pointer is implemented with reference counting and based on the lifetime-tracking hooks. For performance reasons however, the reference counting does not use atomic CPU instructions. Refs cannot be shared in Nim, but they can be moved across threads! For such a move to be successful, a whole subgraph must be moved and no external references to the data must remain. Even read accesses can be harmful as

they keep local variables alive to a point where their destructors introduce hidden write accesses that can cause data races: proc use(x: ref int) = discard "nothing to do" proc example(results: Locker[seq[ref int]]) = var x = new(int) lock results as r:

 r.add x

use x 1 # scope of x ends 2 1 The use of x outside of the lock keeps x alive and so it is not moved to r but copied. 2 The destructor for x runs here and produces a potential data race! The program is invalid should a different thread run code like lock results as r: r.setLen 0 at the same time that the destructor runs. The following code avoids this problem: proc use(x: ref int) = discard "nothing to do" proc example(results: Locker[seq[ref int]]) = var x = new(int) use x 1 lock results as r:

 r.add ensureMove x 2

1 The unprotected access should happen before x is added to results. 2 The compiler has to ensure that x is moved into results.

ch42.md 9.5 KB Historik Rå

Chapter 42. spawn

ch42.md 9.5 KB

Historik Rå