Concurrency · Lesson 25.6

Wait group

Source
Wait for a group of detached workers with a WaitGroup — Add before they start, Done as each one's last step, Wait in Main — and collect their answers in numbered slots so the output keeps its order.
You'll need: Move, Thread, Mutex

Join waits for one thread whose handle you kept. Often there is a whole group of them, started in a loop and let go, and all that matters is that every one has finished — before the results are read, before the program moves on to its next phase, before it exits.

A WaitGroup counts unfinished work:

import Sync::{ Mutex, WaitGroup };

Counting the work

MethodCalled byDoes
Add(n)the starting thread, before startingadds n pieces of pending work
Done()each worker, as its last stepmarks one piece done
Wait()whoever needs the resultssleeps until nothing is pending

The order matters. All the work is counted in before any of it starts:

group.Add(Workers as uint32);

Had each worker called Add itself, Main could reach Wait before any worker had started, see a count of zero, and carry on with nothing done.

flowchart LR
    a["`Add(4)`<br/>pending: 4"] --> w1["worker 1 `Done`"]
    a --> w2["worker 2 `Done`"]
    a --> w3["worker 3 `Done`"]
    a --> w4["worker 4 `Done`"]
    w1 --> z["pending: 0"]
    w2 --> z
    w3 --> z
    w4 --> z
    z --> r["`Wait` returns"]

Detached workers

Main never needs the workers' handles, so it starts them in an ordinary loop and detaches each one at once. Everything a worker needs travels in its Job, with its own handles to the shared results and to the wait group:

let job = Job {
    slot: index, first: first, last: first + Size - 1,
    results: results.Clone(), group: group.Clone()
};

A Job holds handles, which cannot be copied, so the job itself cannot be either. It is moved into Spawn with <-:

var worker = Spawn(CountPrimes, <-job) catch { else => return 1 };
worker.Detach();

The worker counts the primes in its range, stores the count, and calls Done last — after its result is stored, so that whoever Wait wakes finds every result in place:

Store(job.results, job.slot, count);
job.group.Done();

Like Mutex<T>, a WaitGroup is a handle, and Clone gives each worker its own.

Results in order

The workers finish in whatever order they happen to, and that order changes from run to run. So none of them prints. Each one writes its count into its own numbered slot — slot 0 for the first range, slot 1 for the second — under the mutex, and once Wait returns, Main prints the slots in order. The answers come out the same way every time, whichever worker was quickest.

Two relatives

Two more counters in the Sync package coordinate threads by counting.

A Semaphore holds a number of permits. Acquire takes one, waiting while none is left, and Release gives it back — so a semaphore of two lets at most two threads into a stretch of code at once, which suits a resource that only so many may use:

let permits = Semaphore(2);
crew.permits.Acquire();
crew.permits.Release();

A Barrier is a meeting point for a fixed number of threads: each Wait blocks until all of them have arrived, and then they all go on together. It suits work done in phases, where nobody may start phase two until everyone has finished phase one. Wait returns true on exactly one thread of each round — the last to arrive — for work to do once per round:

if crew.barrier.Wait() {
    PrintLine("last");
}
TypeWaits untilSuits
WaitGroupthe pending work reaches zerowaiting for a group to finish
Semaphorea permit is freelimiting how many run at once
Barrierevery party of the round has arrivedphases that must start together

The program

The whole lesson is one package in the Examples repository. Its comments explain every step.

Src/Main.rux
// `Join` waits for one thread whose handle you kept. Often there is a whole group of them, started
// in a loop and let go, and what matters is only that all of them are finished.
//
// A `WaitGroup` counts unfinished work. `Add(n)` raises the count before the work starts, each
// worker calls `Done` as its last step, and `Wait` sleeps until the count is back at zero. The
// waiting thread never needs the workers' handles, so they can be detached.
//
// Workers that finish in any order must not print as they go, or the lines come out shuffled.
// Here each one writes its answer into its own numbered slot, and `Main` prints the slots in
// order once `Wait` returns.
import Io::PrintLine;
import Sync::{ Mutex, WaitGroup };
import Thread::Spawn;

const Workers: uint = 4;
const Size: uint64 = 25000;

// One slot per worker, filled in whatever order the workers finish.
struct Results {
    primes: uint64[4];
}

// Everything one worker needs: its slot and its range, and handles to what the group shares.
struct Job {
    slot: uint;
    first: uint64;
    last: uint64;
    results: Mutex<Results>;
    group: WaitGroup;
}

func IsPrime(value: uint64) -> bool {
    if value < 2 { return false; }
    var divisor: uint64 = 2;
    while divisor * divisor <= value {
        if value % divisor == 0 { return false; }
        divisor = divisor + 1;
    }
    return true;
}

func Store(results: &Mutex<Results>, slot: uint, count: uint64) {
    let guard = results.Lock();
    guard.Get().primes[slot] = count;
}

func CountPrimes(job: Job) -> () {
    var count: uint64 = 0;
    for value in job.first..=job.last {
        if IsPrime(value) { count = count + 1; }
    }
    Store(job.results, job.slot, count);
    // The last step: once every worker has called this, `Wait` returns.
    job.group.Done();
}

func Main() -> int {
    let results = Mutex<Results>(Results { primes: [0, 0, 0, 0] });
    let group = WaitGroup();

    // Count all the work in before starting any of it, so `Wait` cannot see zero too early.
    group.Add(Workers as uint32);
    for index in 0..Workers {
        let first = index as uint64 * Size + 1;
        let job = Job {
            slot: index, first: first, last: first + Size - 1,
            results: results.Clone(), group: group.Clone()
        };
        // The job holds handles, so it cannot be copied and is moved in. `Spawn(CountPrimes, job)`
        // is refused:
        //
        //     error: move-only value 'job' requires an explicit '<-' in argument
        //     help: prefix the argument with '<-', as in 'Take(<-job)'
        //
        // Nobody joins these threads, so their handles are let go at once.
        var worker = Spawn(CountPrimes, <-job) catch { else => return 1 };
        worker.Detach();
    }

    // Sleeps until every worker has called `Done`.
    group.Wait();

    let guard = results.Lock();
    var total: uint64 = 0;
    for index in 0..Workers {
        let first = index as uint64 * Size + 1;
        let count = guard.Get().primes[index];
        PrintLine("{:>6} to {:>6}: {} primes", first, first + Size - 1, count);
        total = total + count;
    }
    PrintLine("all {} workers done: {} primes up to {}", Workers, total, Workers as uint64 * Size);
    return 0;
}

Besides Io, its Rux.toml lists Sync and Thread under [Dependencies].

Run it

cd Examples/Concurrency/WaitGroup
rux run
     1 to  25000: 2762 primes
 25001 to  50000: 2371 primes
 50001 to  75000: 2260 primes
 75001 to 100000: 2199 primes
all 4 workers done: 9592 primes up to 100000

Common mistakes

Try it yourself

  1. Use eight workers of 12500 numbers each. You will need eight slots in Results.
  2. Have each worker also record the largest prime in its range, in a second array of slots.
  3. Give each Job a clone of one Barrier(4), and have the workers wait at it between counting and storing. Does the same worker's Wait return true on every run?

Learn more

  • Sync — the package in the API reference, with Semaphore and Barrier
  • Thread — joining one thread, and detaching
  • Mutex — the lock around the slots
  • Once — the next lesson: work done exactly once, whoever asks first