Concurrency

spawn and join

spawn(function) starts the function on a new thread and gives back its handle, a number. join(handle) waits for that thread to finish. Until you join it, the thread and main run side by side, and main should not read what the thread is still writing.

mut message := ""

func worker():
    message = "the work was done on another thread"
end

func main:
    t := spawn(worker)
    join(t)
    println message
end

Every thread you start should be joined exactly once. If main returns while a thread is still running, that thread is stopped in the middle of its work.

Giving a thread its own data

A thread function may take no parameters, or exactly one pointer: spawn(function, data). The pointer is how each thread gets its own piece of the work, and it is also where the thread leaves its result. The data has to stay alive while the thread runs, so it lives on the heap: allocate it with mem.Allocate and free it after join.

import mem

struct Job:
    pub from: int = 0
    pub to: int = 0
    pub sum: i64 = 0
end

func work(j: Job*):
    mut total := 0 as i64
    repeat j[0].from to j[0].to in i:
        total += i as i64
    end
    j[0].sum = total
end

func new job(first: int, last: int): Job*:
    j := mem.Allocate(sizeof(Job)) as Job*
    j[0] = Job { from = first, to = last }
    ret j
end

func main:
    a := new job(1, 500)
    b := new job(501, 1000)
    t1 := spawn(work, a)
    t2 := spawn(work, b)
    join(t1)
    join(t2)
    println a[0].sum, b[0].sum
    println "total:", a[0].sum + b[0].sum
    mem.Free(a as void*)
    mem.Free(b as void*)
end

The two threads each sum half of the numbers from 1 to 1000. They never touch the same memory, so they need no coordination at all. That is the easiest kind of concurrent program to get right.

Shared data and the mutex

When two threads change the same variable, their updates can collide. count += 1 is really three steps (read, add, write), and if two threads read the same old value, one of the additions is lost. Take the lock out of the next example and the total comes out below 200000, and different on every run.

A mutex from the sync package fixes this: between sync.Lock and sync.Unlock only one thread at a time can be inside. Keep that stretch short, and release the mutex with sync.Destroy when the program no longer needs it.

import sync

mut count := 0
mut lock := sync.NewMutex()

func add():
    repeat 100000:
        sync.Lock(lock)
        count += 1
        sync.Unlock(lock)
    end
end

func main:
    t1 := spawn(add)
    t2 := spawn(add)
    join(t1)
    join(t2)
    println count
    sync.Destroy(lock)
end

Atomic counters

For a single number that many threads bump, the atomic package is lighter than a mutex. atomic.Add does the read, the add and the write as one step that nothing can interrupt. atomic.Load reads the value, and Store, Swap and CompareAndSwap cover the other usual needs.

import atomic

mut hits := atomic.NewI64(0)

func visit():
    repeat 100000:
        atomic.Add(hits, 1)
    end
end

func main:
    t1 := spawn(visit)
    t2 := spawn(visit)
    t3 := spawn(visit)
    join(t1)
    join(t2)
    join(t3)
    println atomic.Load(hits)
    atomic.DestroyI64(hits)
end

Waiting for a group

With many threads, one variable per handle gets awkward. Put the handles in a Vector<i64> and join them in a loop. A sync.WaitGroup answers a related question, "have all of them finished their part?": sync.Add sets how many to wait for, each worker calls sync.Done when it is finished, and sync.Wait blocks until the count reaches zero.

import sync

mut done := 0
mut lock := sync.NewMutex()
mut group := sync.NewWaitGroup()

func job():
    sync.Lock(lock)
    done += 1
    sync.Unlock(lock)
    sync.Done(group)
end

func main:
    mut threads := Vector {} as Vector<i64>
    defer threads.free()
    sync.Add(group, 5)
    repeat 5:
        threads.push(spawn(job))
    end
    sync.Wait(group)
    println "finished:", done
    each t in threads:
        join(t)
    end
    sync.DestroyWaitGroup(group)
    sync.Destroy(lock)
end

Channels

A channel from the chan package carries values from one thread to another, so the threads share nothing but the channel itself. init(n) gives it room for n values; send waits while it is full and recv waits while it is empty. After close() nothing more can be sent, and once the remaining values are taken, recv sets its ok argument to false. That is the signal for a worker to stop.

import chan

mut jobs := chan.Chan {} as chan.Chan<i64>
mut results := chan.Chan {} as chan.Chan<i64>

func worker():
    mut ok := true
    until true:
        n := jobs.recv(ok)
        if not ok:
            break
        end
        results.send(n * n)
    end
end

func main:
    jobs.init(4)
    results.init(16)
    a := spawn(worker)
    b := spawn(worker)
    repeat 1 to 10 in i:
        jobs.send(i as i64)
    end
    jobs.close()
    join(a)
    join(b)
    results.close()
    mut sum := 0 as i64
    mut ok := true
    until true:
        v := results.recv(ok)
        if not ok:
            break
        end
        sum += v
    end
    println "sum of squares:", sum
    jobs.free()
    results.free()
end

The two workers take jobs in whatever order they happen to get them, but the sum does not depend on that order, so the output is the same on every run. With init(0) the channel holds nothing: send waits until a receiver takes the value, so the two threads meet. recv_timeout and send_timeout give up after a number of milliseconds.

Parallel loops

Often all you want is to run the same loop body over many indexes on all the cores. pool.ParallelFor(n, workers, body) calls body(i) for every i from 0 to n - 1 on workers threads and returns when all of them are done. pool.ParallelForAuto uses one worker per core, and pool.CPUCount() tells you how many cores there are.

import pool
import mem

mut squares := null as i64*

func main:
    n := 8
    squares = mem.AllocateZeroed(sizeof(i64) * (n as u64)) as i64*
    pool.ParallelFor(n, 4, (i: i64): squares[i] = i * i end)
    mut total := 0 as i64
    repeat n in i:
        total += squares[i]
    end
    println "last square:", squares[n - 1]
    println "total:", total
    mem.Free(squares as void*)
end

A lambda captures the variables around it by value, so the body cannot add to a local total. Instead each call writes its own slot of shared memory, and main adds the slots up after the loop.

More tools

Rules of thumb

• Join every thread you spawn, and read its results only after the join.
• Data that two threads both change needs a mutex, an atomic or a channel. Data that only one thread touches needs nothing.
• Threads reach shared state through globals or through the pointer given to spawn. A lambda's captured variables are copies.
• Free what you create: sync.Destroy, sync.DestroyWaitGroup, atomic.DestroyI64 and a channel's free().
• If the output must be the same on every run, print from main after the joins, not from inside the workers.