Concurrency

Concurrent programs are hard for three reasons: tasks that outlive the code that started them, data that two tasks change at once, and results that depend on timing. Tessera's design takes each of these away. Tasks live in a group block and are finished when it ends. Tasks share values by copying them, like everything else, so there are no data races. And what tasks print comes out in an order that doesn't depend on scheduling.

Groups and tasks

fn count_primes(lo: Int, hi: Int) -> Int {
    let mut n = 0
    for x in lo..<hi {
        if is_prime(x) {
            n = n + 1
        }
    }
    n
}

fn is_prime(x: Int) -> Bool {
    if x < 2 {
        return false
    }
    let mut d = 2
    while d * d <= x {
        if x % d == 0 {
            return false
        }
        d = d + 1
    }
    true
}

fn main(root: Root) {
    let ranges = [(0, 25_000), (25_000, 50_000), (50_000, 75_000), (75_000, 100_000)]
    let counts = group g {
        for (lo, hi) in ranges {
            g.go(|| count_primes(lo, hi))
        }
    }
    root.stdout.print("primes per range: {counts}")
    root.stdout.print("primes below 100000: {counts.sum()}")
}

$ tessera run main.tsr
primes per range: [2762, 2371, 2260, 2199]
primes below 100000: 9592
$ tessera run main.tsr --native
primes per range: [2762, 2371, 2260, 2199]
primes below 100000: 9592

group g { ... } runs its body, in which g.go(f) starts the function f as a task. When the block ends, every task it started has finished; nothing outlives the group. The group's value is the list of the tasks' results, in the order the tasks were started, whichever finished first.

Tasks are cheap. In native code each one is a small fiber, and the runtime runs them on one thread per core, so the four counts above run in parallel. The interpreter runs the same program with the same result.

Each closure passed to g.go captures lo and hi by value, as closures always do. Tasks can't change variables outside them, which is why there are no data races:

fn main(root: Root) {
    let mut total = 0
    let _ = group g {
        for n in [1, 2, 3] {
            g.go(|| {
                total = total + n
            })
        }
    }
    root.stdout.print("{total}")
}

$ tessera check main.tsr
error[E0205]: cannot modify `total` inside a closure
  --> main.tsr:6:17
   |
 6 |                 total = total + n
   |                 ^^^^^
  = note: closures capture variables by value; return the new value from the closure instead

1 error

Return values from the tasks instead, and combine them after the group, as listing 13-1 does with counts.sum().

A few more rules keep groups honest. g can only be used directly in the group's body, never stored or passed on, and the body can't return or use ? to leave early. A task that returns () runs for its effect and adds nothing to the result list. Starting a task needs no capability, but a task can only reach the world through capabilities its closure captured.

Output order

When several tasks print, the output of most concurrent programs depends on which task got there first. In Tessera, it doesn't:

fn main(root: Root) {
    let out = root.stdout
    let clock = root.clock
    let _ = group g {
        for (name, ms) in [("slow", 300), ("medium", 100), ("fast", 0)] {
            g.go(|| {
                clock.sleep(ms)
                out.print("{name} done after {ms} ms")
            })
        }
        out.print("all started")
    }
    out.print("all finished")
}

$ tessera run main.tsr
all started
slow done after 300 ms
medium done after 100 ms
fast done after 0 ms
all finished

The fast task finishes first, but its line comes last. A group's output reads as if its body ran first, then each task in the order it was started. A task's output is held until the tasks before it have finished, then written; the first unfinished task prints directly. The same program prints the same thing every time, in the interpreter and in native code, which makes concurrent programs as easy to test as sequential ones.

Only standard output is held this way. A line written to root.stderr comes out at once, after whatever standard output has been written so far: a warning or an error should show when it happens, not when the task's turn comes.

For tasks that run until stopped, like a server's connections, holding output back would be wrong. group g interleaved { ... } turns ordering off for one group, and its tasks print as things happen. The http module's servers use it.

Channels

Tasks that need to talk while they run use a channel:

fn main(root: Root) {
    let lines: Chan[Str] = channel(4)
    let text = "one fish\ntwo fish\nred fish\nblue fish"
    let results = group g {
        // the producer sends each line, then closes the channel
        g.go(|| {
            for line in text.lines() {
                lines.send(line)
            }
            lines.close()
        })
        // the consumer counts words until the channel is closed and empty
        g.go(|| {
            let mut words = 0
            let mut more = true
            while more {
                match lines.recv() {
                    Some(line) => {
                        words = words + line.split(" ").len()
                    },
                    None => {
                        more = false
                    },
                }
            }
            words
        })
    }
    root.stdout.print("{results}")
}

$ tessera run main.tsr
[8]

channel(4) makes a Chan[Str] that holds up to four values. send waits while it's full, recv waits while it's empty and returns Some(value), and after close(), once the channel is drained, recv returns None. channel(0) holds nothing, so a sender waits until a receiver takes the value.

Channels are the one thing tasks share rather than copy: two tasks that capture the same channel talk through it.

If every task is waiting and nothing can ever wake one, the program stops with all tasks are blocked (deadlock) instead of hanging.

select and timeouts

select waits for whichever of several channel operations can happen first. With clock.after(ms) as one of the arms, that gives a timeout:

// Asks a slow service, but waits at most `limit` ms for the answer.
fn ask(clock: Clock, delay: Int, limit: Int) -> Str {
    let answers: Chan[Str] = channel(1)
    let got = group g {
        g.go(|| {
            clock.sleep(delay)
            answers.send("answer after {delay} ms")
        })
        g.go(|| {
            select {
                a = answers.recv() => a ?? "closed",
                clock.after(limit) => "gave up after {limit} ms",
            }
        })
    }
    got[0]
}

fn main(root: Root) {
    root.stdout.print(ask(root.clock, 50, 500))
    root.stdout.print(ask(root.clock, 500, 50))
}

$ tessera run main.tsr
answer after 50 ms
gave up after 50 ms

An arm is name = ch.recv(), ch.send(value), clock.after(ms), or _, which runs if nothing else is ready right away. When several arms are ready, the first one written wins. A guard, if cond after the operation, turns an arm off, which is how you stop receiving from a closed channel.

Waiting needs the Clock, like reading the time does. A pure function can't depend on timing.

Notice that the second call still takes the full 500 ms: the group waits for the slow task even after the select gave up. To stop tasks that are taking too long, give the group a deadline.

Deadlines and cancellation

fn fetch(clock: Clock, name: Str, ms: Int) -> Str {
    clock.sleep(ms)
    "{name} ({ms} ms)"
}

fn main(root: Root) {
    let clock = root.clock
    let jobs = [("a", 10), ("b", 900), ("c", 50)]
    let pages = group g until clock.after(300) {
        for (name, ms) in jobs {
            g.go(|| fetch(clock, name, ms))
        }
    }
    for p in pages {
        root.stdout.print("{p}")
    }
}

$ tessera run main.tsr
Some("a (10 ms)")
None
Some("c (50 ms)")

group g until clock.after(300) { ... } cancels every task still running 300 ms after the group started. Its value is then a list of Options: Some(result) for tasks that finished, None for tasks that were cancelled.

A cancelled task stops at its next wait: a sleep, a channel, a socket, a select. Its defers run and its sockets are closed, as if it had panicked, but the group passes nothing on. Code that never waits can't be cancelled midway; it finishes, and keeps its result.

For a single operation, the time module wraps this up: time.within(clock, ms, || net.get(url)) gives the result, or an error after ms milliseconds.

Panics in tasks

A panic in a task ends only that task. The group waits for the others, then passes the first panic on to the code that ran the group, where a catch can handle it. Output from every task is still written in its turn, so nothing printed is lost.

Summary

group blocks contain tasks, which are finished when the block ends; results come back in start order. Tasks share nothing but channels, so there are no data races, and output is ordered as if tasks ran one after another. Channels and select connect tasks, clock.after adds timeouts, and group ... until cancels tasks at a deadline. All of it runs the same way in both backends, and chapter 11 showed how to test it on virtual time.

Next we'll put these to work talking to the world: files, HTTP, databases and time.