Threads and parallel loops

Four tools, from the one that needs no thinking to the one that needs the most: for parallel for data that splits into independent pieces, threads with a result, channels between threads, and a mutex around shared state. Plus using arena, which is about memory rather than threads but keeps them company in the fast paths.

import std.thread

// ----- for parallel: data parallelism without threads in your code -----------

/// Is `n` prime? Slow on purpose: trial division.
fn is_prime(n: u64) -> bool {
    if n < 2 { return false }
    var d: u64 = 2
    while d * d <= n {
        if n % d == 0 { return false }
        d += 1
    }
    return true
}

fn count_primes_parallel(limit: u64) -> u64 {
    // one slot per chunk; each iteration writes its own slot, so nothing is shared
    var counts: [16]u64 = undefined
    let out = counts[..]
    let chunk = limit / 16
    for parallel _slot, i in counts[..] {
        let from = i as u64 * chunk
        let to = if i == 15 { limit } else { from + chunk }
        var found: u64 = 0
        var n = from
        while n < to { if is_prime(n) { found += 1 }; n += 1 }
        out[i] = found
    }
    var total: u64 = 0
    for c in counts { total += c }
    return total
}

// ----- threads with results ---------------------------------------------------

struct Range { from: u64, to: u64 }

fn sum_range(r: *mut Range) -> u64 {
    var s: u64 = 0
    var i = r.from
    while i < r.to { s += i; i += 1 }
    return s
}

// ----- a channel between a producer and the main thread ---------------------------

struct Producer { out: *mut thread.Channel(String), n: u32 }

fn produce(p: *mut Producer) {
    for i in 0..p.n { p.out.send(format("message {}", .{i})) }
    p.out.close()
}

// ----- a mutex around shared state -------------------------------------------------

struct Bumper { counter: *mut thread.Mutex(i64), times: i64 }

fn bump(b: *mut Bumper) {
    for _ in 0..b.times {
        let n = b.counter.lock()
        n.*+= 1
        b.counter.unlock()
    }
}

fn main() {
    println("{} primes below 20000", .{count_primes_parallel(20000)})

    var a = thread.spawn(Range, u64, sum_range, Range{ .from = 0, .to = 500000 })
    var b = thread.spawn(Range, u64, sum_range, Range{ .from = 500000, .to = 1000000 })
    println("sum {}", .{a.join() + b.join()})

    var ch = thread.channel(String)
    var producer = thread.run(Producer, produce, Producer{ .out = &mut ch, .n = 5 })
    while true {
        let msg = ch.recv() orelse break
        println("got {}", .{msg})
    }
    producer.join()
    ch.free()

    var counter = thread.mutex(i64, 0)
    var workers = List(thread.Worker(Bumper)).new()
    for _ in 0..4 { workers.append(thread.run(Bumper, bump, Bumper{ .counter = &mut counter, .times = 1000 })) }
    for i in 0..workers.len { workers[i].join() }
    let n = counter.lock()
    println("counter {}", .{n.*})
    counter.unlock()
    counter.free()

    // an arena: allocate freely inside, everything is freed at once at the end
    var lengths = List(usize).new()
    using arena {
        for i in 0..1000 {
            let s = format("item number {}", .{i})
            lengths.append(s.len)
        }
    }
    println("{} strings measured, first {} bytes", .{lengths.len, lengths[0]})
    println("{} hardware threads", .{if thread.count() >= 1 { "some" } else { "no" }})
}

2262 primes below 20000
sum 499999500000
got message 0
got message 1
got message 2
got message 3
got message 4
counter 4000
1000 strings measured, first 13 bytes
some hardware threads

for parallel#

    for parallel _slot, i in counts[..] {
        ...
        out[i] = found
    }

A for parallel loop runs its body across a pool of worker threads, one chunk of the index range each. The rules that make this safe are checked by the compiler: the body may not have the shared_mutable effect (no globals, no locks), may not return or break, and writes its results through a mutable slice indexed by i, so that every iteration touches its own slot and nothing else. A panic in a worker is re-raised in the caller once all workers finish. Counting primes in sixteen ranges is the shape of most of these loops: split, compute, write to slot i, sum at the end.

The loop itself carries the blocks effect (it joins its workers), and that is all the caller sees.

Threads with results#

    var a = thread.spawn(Range, u64, sum_range, Range{ .from = 0, .to = 500000 })
    println("sum {}", .{a.join() + b.join()})

std.thread.spawn(T, R, f, arg) runs f(&mut arg) on a new thread; the Thread(T, R) owns the argument until join returns the result. Types come first because the function is generic; the argument is a value you give away (own), so the thread's data is its own and nothing is shared by accident. A panic inside the thread surfaces from join.

thread.run is the same for functions without a result (a Worker(T)), and t.arg() after join reads what the function left in its argument.

Channels#

    var ch = thread.channel(String)
    var producer = thread.run(Producer, produce, Producer{ .out = &mut ch, .n = 5 })
    while true {
        let msg = ch.recv() orelse break

A Channel(T) is a queue with a lock inside. send moves a value in; recv waits for one and returns null once the channel is closed and drained, which is the loop's exit; try_recv does not wait. Channels are values that threads share by pointer, so the owner joins every thread that uses one before it goes out of scope, and calls free when done. The producer here sends five Strings and closes; the ownership of each message moves from the producer to the consumer through the channel, so neither side clones anything.

Mutexes#

    let n = counter.lock()
    n.* += 1
    counter.unlock()

Mutex(T) wraps a value; lock returns a *mut T that is valid until unlock. Locking has the blocks and shared_mutable effects, which is why a for parallel body cannot do it, and why the thread pool and the mutex are two different tools: the loop for when the work splits cleanly, the mutex for when it genuinely does not.

Arenas#

    using arena {
        for i in 0..1000 {
            let s = format("item number {}", .{i})
            lengths.append(s.len)
        }
    }

using arena { } swaps a bump allocator in for the block: every value created inside comes from one growing region, individual releases are no-ops, and the whole region is freed at once when the block ends. A thousand temporary strings cost one allocation. Containers created outside the block (lengths) keep using the heap when they grow inside it, so collecting results into an outer list is safe; values created inside must not be kept past the end. It is the allocation strategy for a parse, a frame, a request: work that has a clear end.

Threads and the rest of the language#

Data races are not memory-safety violations in Nexium's promise (the specification says so, in section 13), and the type system does not prevent them: it makes them visible. A function that touches shared state has shared_mutable; one that starts a thread has nondeterministic; a for parallel body cannot have the first at all. The roadmap's 1.4 adds scoped threads (joined when their block ends, so a thread cannot outlive the data it was given) and select over channels.

Next: a network service.