Chapter 14 — Threads

Tasks (chapter 13) take turns on one thread, which is ideal for waiting but does nothing for heavy computation. Threads run truly in parallel on several CPU cores. Each thread has its own memory, and values are copied when they move between threads, so two threads can never change the same object at the same time.

Everything here comes from the built-in thread module. Its functions return Result<…, ThreadError>.

spawn and join#

spawn(f) runs the function f on another thread, and spawn(f, arg) passes it one argument. join waits for the thread and returns what f returned:

use io::sync::{println};
use string::{format};
use thread::{spawn, join};

fn count_primes(int limit) -> int {
    let count = 0;
    for n in 2..limit {
        let prime = true;
        let d = 2;
        while d * d <= n {
            if n % d == 0 {
                prime = false;
                break;
            }
            d++;
        }
        if prime {
            count++;
        }
    }
    return count;
}

fn both() -> Result<(int, int), ThreadError> {
    let small = spawn(count_primes, 50000)?;
    let large = spawn(count_primes, 60000)?;
    return Ok((join(small)?, join(large)?));
}

fn main() {
    match both() {
        Ok(counts) => {
            println(format("%i primes below 50000, %i below 60000", counts[0], counts[1]))?;
        },
        Err(e) => {
            println("thread failed: " + show(e))?;
        },
    }
}

Output:

5133 primes below 50000, 6057 below 60000

The two counts run at the same time. both returns its own Result<…, ThreadError>, because, as chapter 6 explained, one function uses ? with one error type; main only prints.

The argument, and anything a lambda captures with use (…), is copied to the new thread. Objects are copied too, so changes a thread makes to its copy are not seen by the caller. Return results instead, or use a channel or a mutex.

Channels#

channel() makes a pair: index 0 is the sending end and 1 the receiving end. send puts a value in, recv waits for the next one, and close tells the receiver no more values are coming. To send work to a thread and get answers back, use two channels:

use io::sync::{println};
use string::{format};
use thread::{spawn, join, channel, send, recv, close, Sender, Receiver};

fn squarer((Receiver, Sender) ends) {
    while let Ok(job) = recv(ends[0]) {
        let n: int = job;
        send(ends[1], n * n)?;
    }
}

fn sum_of_squares() -> Result<int, ThreadError> {
    let jobs = channel()?;
    let results = channel()?;
    let t = spawn(squarer, (jobs[1], results[0]))?;
    for n in 1..=3 {
        send(jobs[0], n)?;
    }
    close(jobs[0])?;
    let total = 0;
    for i in 0..3 {
        total += recv(results[1])?;
    }
    join(t)?;
    return Ok(total);
}

fn main() {
    println(format("%v", sum_of_squares()))?;
}

Output:

Ok(14)

Channels carry any value that can be copied between threads; the annotation let n: int = job; tells the compiler what this one carries.

Mutex#

When several threads must update one value, put it in a mutex. with_lock(m, f) locks it, calls f with the current value, stores the first element of what f returns, and gives you the second:

use io::sync::{println};
use string::{format};
use thread::{spawn, join, mutex, with_lock, Mutex};

fn add_thousand(Mutex counter) {
    for i in 0..1000 {
        with_lock(counter, fn (int n) => (n + 1, 0))?;
    }
}

fn count() -> Result<int, ThreadError> {
    let counter = mutex(0)?;
    let a = spawn(add_thousand, counter)?;
    let b = spawn(add_thousand, counter)?;
    join(a)?;
    join(b)?;
    return Ok(with_lock(counter, fn (int n) => (n, n))?);
}

fn main() {
    println(format("%v", count()))?;
}

Output:

Ok(2000)

The lock is held only while f runs, so no update is lost and no lock is ever left held. For data that is read often and written rarely, rwlock with with_read and with_write lets many readers in at once.

Tasks or threads?#

The compiler can also parallelise some pure recursive functions on its own; coil dissect --effects shows what it decided and why. See effects.

Exercises#

1. Split count_primes(200000) into four threads that each count a quarter of the range, and add up their results.
2. Write a worker pool: three threads reading jobs from one channel and writing results to another.
3. Use a mutex holding a Vec<string> to collect log lines from several threads, then print them.

Next: Chapter 15 — Macros.