Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

Threads and shared data

The async module starts and manages operating-system threads. Pepper keeps threads safe by controlling what they can share: only thread-safe values can be reached from more than one thread, and the compiler checks the types that claim to be.

On WebAssembly (--target wasm32-wasip1) a program has one thread: Thread.spawn panics. Mutexes, channels and atomics still work, as long as nothing waits for another thread (see WebAssembly).

Starting and joining threads

import async;

const finished = async.Atomic<int>(0);

# A thread body is a function that takes nothing and returns nothing.
count_to_a_million() {
    var me = async.Thread.current();
    assert(me.name() == "counter", "a spawned thread sees its own Thread");
    var total: long = 0;
    var i = 0;
    while (i < 1000000) {
        total = total + 1;
        i = i + 1;
    }
    finished.fetch_add(1);
}

main(): int {
    var main_thread = async.Thread.current();
    assert(main_thread.name() == "main", "the main thread is named main");

    var first = async.Thread.spawn(count_to_a_million, "counter");
    var second = async.Thread.spawn(count_to_a_million, "counter");
    assert(first.id() != second.id(), "every thread has its own id");

    first.join();
    second.join();
    assert(first.is_finished(), "join waits for the body to return");
    assert(finished.load() == 2, "both bodies ran");

    # How many threads can run at once, and on how many cores.
    assert(async.logical_cpus() >= async.physical_cpus(), "hyperthreads count once in physical_cpus");
    return 0;
}
  • Thread.spawn(body, name) starts a new thread that runs body, a function of type ThreadBody (an alias for shared () => void), and returns its Thread. The name is optional.
  • join() waits for the body to return. After join(), the joining thread sees everything the other thread did. A thread can be joined once.
  • Thread.current() returns the calling thread’s Thread. For a spawned thread it is the same object that spawn returned. The main thread’s is named main.
  • id() returns the thread’s ThreadId, a number that no other thread in the program has. Ids come from the runtime, starting at 1, and are not the operating system’s thread ids. A ThreadId can be compared and used as a HashMap key.
  • is_finished() reports whether the body has returned. is_joinable() reports whether join() may be called.
  • Thread.sleep(milliseconds) pauses the calling thread (in a task, only the task: see Tasks).
  • Dropping every reference to a Thread without joining it detaches the thread. It keeps running, and its resources are freed when it finishes.
  • A panic or failed assert on any thread ends the whole program.

The body can be a named function or a closure. A closure passes data to the thread through its captures:

var total = async.Atomic<long>(0);
var step: long = 5;
var worker = async.Thread.spawn(() => { total.fetch_add(step); });
worker.join();

The body is a shared function value, so the closure may capture only thread-safe values (numbers, Atomic<T>, [Shared] structs, shared function values). The compiler checks each capture and reports one that is not thread-safe where the closure uses it:

main.pep:5:46: error: closure passed as 'shared () => void' captures 'name' of type 'String', which is not thread-safe
    5 |     var t = async.Thread.spawn(() => { print(name); });
      |                                              ^~~~
main.pep:4:9: note: capture a thread-safe value instead, such as an 'async.Atomic<T>' or a '[Shared]' struct
    4 |     var name = "worker";
      |         ^~~~

Every module-scope variable is const and thread-safe, so any thread may use one (see below).

The Thread reference documents every method.

Processor counts

  • async.logical_cpus() returns how many threads the machine can run at the same moment. A core with hyperthreading (SMT) counts two or more times. This is a good default size for a pool of CPU-bound worker threads. In a Linux container with a CPU quota (docker run --cpus), it is the quota rounded up, when that is fewer (see Containers).
  • async.physical_cpus() counts each physical core once. On Linux it reads the kernel’s CPU topology, on macOS hw.physicalcpu, and on Windows GetLogicalProcessorInformationEx. Inside a virtual machine it counts the cores the hypervisor presents, because the guest cannot see the host’s hardware.

Thread-local variables

A module-scope var marked [ThreadLocal] has a separate copy on every thread, like C++ thread_local or a Rust #[thread_local] static:

import async;

import [Thread] from async;

# A separate `depth` on every thread, each starting at 0.
[ThreadLocal]
var depth: int = 0;

const deepest = async.Atomic<int>(0);

descend(levels: int) {
    depth += 1;
    if (depth > deepest.load()) {
        deepest.store(depth);
    }
    if (levels > 1) {
        descend(levels - 1);
    }
    depth -= 1;
}

work() {
    descend(5);
}

main(): int {
    depth = 100;   # changes only the main thread's copy
    var a = Thread.spawn(work);
    var b = Thread.spawn(work);
    a.join();
    b.join();
    assert(deepest.load() == 5, "each thread counted from its own 0");
    assert(depth == 100, "the main thread's copy is its own");
    return 0;
}
  • Every thread’s copy starts from the initializer, which must be a constant: a number, bool or char literal (negated, or cast), an enum case, or nothing, for the type’s default (zero, or null for a struct). To start from anything else, leave the initializer out and set the value on the thread’s first use.
  • Reading or assigning one touches only the calling thread’s copy, so it is a plain var: it need not be const like other module-scope variables, and needs no atomics or locks. A read is a single load from the thread’s storage (a %fs-relative load on x86-64 Linux), not a call.
  • Its type must be thread-safe (see What threads can share): a number, a raw pointer, a [Shared] struct, a shared function type, or a tuple or [Value] struct of them. A value of any other type, such as a String, goes in a ThreadLocal<T>.
  • When a thread started with Thread.spawn exits (the async runtime’s workers and blocking threads included), each of its thread-locals is set to its default and the struct it held is released, last declared first. The main thread’s are never released, like every other module-scope variable, and a thread started by C code does not release its own.
  • A task sees the thread-locals of the worker thread it runs on, as with a ThreadLocal<T>.

[ThreadLocal] applies only to a module-scope variable. A variable whose type is not thread-safe, or whose initializer is not a constant, is reported at the type or the initializer:

main.pep:3:14: error: thread-local variable 'scratch' has type 'Plain', which is not thread-safe
    3 | var scratch: Plain;
      |              ^~~~~
note: a per-thread value of any type is an 'async.ThreadLocal<T>' (after 'import async')
main.pep:1:8: note: 'Plain' is not declared [Shared]
    1 | struct Plain { var x: int; }
      |        ^~~~~

main.pep:3:18: error: thread-local variable 'depth' must start from a constant
    3 | var depth: int = start();
      |                  ^~~~~
note: every thread starts from this value; leave it out and assign the value on the thread's first use instead

Per-thread values with ThreadLocal

ThreadLocal<T> is a variable with a separate value on every thread, of any type:

import async;

import [Thread, ThreadLocal] from async;

# One variable, a separate value on every thread.
const calls = ThreadLocal<int>();
const totals = async.Atomic<long>(0);

record_call() {
    calls.set(calls.get() + 1);
}

work() {
    record_call();
    record_call();
    record_call();
    totals.fetch_add(calls.get());   # 3 on every thread
}

main(): int {
    record_call();
    var a = Thread.spawn(work);
    var b = Thread.spawn(work);
    a.join();
    b.join();
    assert(totals.load() == 6, "each thread counted its own calls");
    assert(calls.get() == 1, "the main thread's value is its own");
    return 0;
}
  • get() returns the calling thread’s value, or the default of T (zero, or null for a struct) until that thread calls set.
  • set(value) replaces only the calling thread’s value.
  • is_set() and clear() test for and remove the calling thread’s value.
  • When a thread started with Thread.spawn exits, its value is released, even if the ThreadLocal itself was destroyed first. As with [ThreadLocal] variables, the main thread’s value is released only by clear(), and a thread started by C code leaks the values it did not clear.
  • Each thread keeps its values in a table of its own, held in [ThreadLocal] variables, so get() costs a few loads and no call, and a program can have any number of ThreadLocals.

A ThreadLocal is usually a module-scope variable. Its T does not need to be thread-safe, because each value stays on the thread that stored it. For a T that is not thread-safe, such as a String, the value is reached like a Mutex’s (see What a lock body may do):

  • with (var value = local.enter()) { ... } binds this thread’s value; the else block runs when the thread has none;
  • get() is a compile error;
  • set(value) takes a new value: a constructor call, a string literal, copy(x), or move(x).

Otherwise a function called while a Mutex is locked could keep part of the locked value in a thread-local, and a later get() would hand it out after the lock was released. The ThreadLocal reference has the details.

Tasks that take turns on one worker thread share its ThreadLocal values; a TaskLocal<T> gives each task its own.

What threads can share

Every struct value is a counted reference (see Memory and reference counting). Updating a count from two threads at once can free an object too early or never, so by default a struct’s counts are plain, fast increments, and its objects must stay on one thread.

A struct declared [Shared] has atomic reference counts, so references to one object can be copied and dropped on any thread:

import async;

# A struct that threads can share: declared [Shared], every field const and
# thread-safe.
[Shared]
struct Settings {
    const port: int;
    const retries: int;
    construct(this.port = 0, this.retries = 0) {}
}

# async.Atomic<T> of a [Shared] struct: threads read the current settings, and one
# of them can replace them.
const current = async.Atomic<Settings>(Settings(8080, 3));
const seen = async.Atomic<long>(0);

read_settings() {
    var settings = current.load();
    seen.fetch_add(settings.port);
}

main(): int {
    var readers = Array<async.Thread>();
    var i = 0;
    while (i < 4) {
        readers.append(async.Thread.spawn(read_settings));
        i = i + 1;
    }
    for (var reader in readers) {
        reader.join();
    }
    assert(seen.load() == 4 * 8080, "every reader saw the settings");

    current.store(Settings(9090, 5));
    assert(current.load().port == 9090, "replaced");
    return 0;
}

The compiler checks every [Shared] struct:

  • Every field must be const. It is set once, by the constructor, and never assigned again, so threads cannot race on it. To change shared state, give the struct a thread-safe member such as an Atomic<T>.

  • Every field must have a thread-safe type. These are all the thread-safe types, the same ones the constraint T: Shared accepts:

    • every number type (int, long, byte, char, short, float, double and the rest), bool, and void;
    • enums whose cases carry no values, which are stored as integers;
    • [Shared] structs, and enums whose cases carry values that are declared [Shared];
    • weak references to thread-safe types;
    • raw pointers (__ptr<T>, and void*) to thread-safe types;
    • shared function types, such as shared () => void (see Function types), and C function pointer types (extern (int) => int);
    • tuples whose elements are all thread-safe;
    • inside a generic, a type parameter constrained T: Shared.

    Nothing else is. A struct is never thread-safe on its own, even when all its fields are numbers: a struct value is a reference to a reference-counted object, and only a [Shared] struct counts those references atomically. So String, Array and other structs that are not [Shared] are not thread-safe, and neither is a plain function type such as () => void:

    main.pep:3:14: error: field 'F' of shared struct 'H' has type '() => void', which is not thread-safe
    3 |     const F: () => void;
      |              ^~~~~~~~~~
note: declare the function type 'shared () => void' to share it between threads

A generic [Shared] struct whose parameter has no constraint is checked for each set of type arguments, and the error points at the code that asked for the specialization:

[Shared]
struct Cell<T> {
    const value: T;
}

make<T>(): Cell<T> { return Cell<T>(); }

main(): int {
    var c = make<String>();
    return 0;
}

main.pep:9:13: error: 'make<String>' requires thread-safe type arguments; 'String' is not thread-safe
    9 |     var c = make<String>();
      |             ^~~~~~~~~~~~
main.pep:6:29: note: 'Cell<String>' is required here, in 'make<String>'
    6 | make<T>(): Cell<T> { return Cell<T>(); }
      |                             ^~~~~~~
main.pep:3:18: note: required by field 'value' declared here
    3 |     const value: T;
      |                  ^
runtime/string.pep:54:15: note: 'String' is not declared [Shared]

A generic can state the rule up front with the built-in constraint T: Shared: [Shared] struct Cell<T: Shared> accepts exactly the type arguments above, and a wrong one is reported where it is written ('String' does not conform to 'Shared', required by 'T' in 'Cell'). The standard library’s Atomic<T>, spawn, run_blocking and Lazy<T> do this, so Atomic<String> is reported at String.

A field that is not const, or that has a type that is not thread-safe, is reported at the field:

main.pep:3:9: error: field 'Count' of shared struct 'Counter' must be 'const'
    3 |     int Count;
      |         ^~~~~
note: threads change a shared struct's state through a thread-safe member such as 'async.Atomic<T>' (after 'import async')

Thread, ThreadLocal<T> and Atomic<T> are [Shared] themselves, and so are the synchronization types below: Channel<T>, Mutex<T>, Condition<T>, Once and Lazy<T>.

Module-scope variables

Every module-scope variable is implicitly shared, exported or not: any function can use it, and any function can run on any thread. So it follows the same two rules as a field of a [Shared] struct:

  • It must be const. To change the value, give it a thread-safe type such as Atomic<T>, or a ThreadLocal<T> for a value per thread. The one exception is a [ThreadLocal] variable, which every thread has its own copy of.
  • It must have a thread-safe type from the list above. A String, an Array, a struct that is not [Shared], a trait value, or a plain function value cannot be a module-scope variable; keep it in a local, or in a ThreadLocal<T>.
const MAX_WORKERS = 8;                  # a number
const jobs_done = async.Atomic<long>(0);  # changed with fetch_add
const settings = Settings(8080, 3);     # a [Shared] struct
const scratch = ThreadLocal<String>();  # a String per thread
const double: shared (int) => int = (x: int) => x * 2;  # a shared function
[ThreadLocal]
var depth: int = 0;                     # a copy per thread

A variable that breaks a rule is reported at its name or type:

main.pep:1:5: error: module-scope variable 'hits' must be 'const'
    1 | var hits = 0;
      |     ^~~~
note: every thread can reach a module-scope variable; change its state through a thread-safe type such as 'async.Atomic<T>' (after 'import async'), or declare it '[ThreadLocal]' to give each thread its own copy

main.pep:2:15: error: module-scope variable 'origin' has type 'Plain', which is not thread-safe
    2 | const origin: Plain = Plain();
      |               ^~~~~
main.pep:1:8: note: 'Plain' is not declared [Shared]
    1 | struct Plain { var x: int; }
      |        ^~~~~

Module-scope variables are initialized before main runs, in the order they are declared, and each file’s after those of the files it imports and of the runtime. So an initializer may use the variables of the files it imports.

Channels

A Channel<T> carries values from threads that send to threads that receive, with any number of each, like a Go channel. (In a task, a wait parks only the task, and so do the waits of Mutex and Condition below.)

import async;

# The main thread sends numbers to a pool of workers, and the workers send
# their squares back on a second channel.
const jobs = async.Channel<long>(16);    # bounded: send waits while 16 are queued
const results = async.Channel<long>();   # unbounded: send never waits

square_numbers() {
    # The loop ends once `jobs` is closed and empty.
    for (var n in jobs) {
        results.send(n * n);
    }
}

# A `shared` function value is thread-safe, so a channel can carry work.
const tasks = async.Channel<async.ThreadBody>();
const ran = async.Atomic<long>(0);

run_tasks() {
    for (var task in tasks) {
        task();
    }
}

main(): int {
    var workers = Array<async.Thread>();
    var i = 0;
    while (i < 4) {
        workers.append(async.Thread.spawn(square_numbers));
        i = i + 1;
    }
    var n: long = 1;
    while (n <= 100) {
        jobs.send(n);
        n = n + 1;
    }
    jobs.close();   # no more work: each worker's loop ends
    for (var worker in workers) {
        worker.join();
    }
    results.close();

    var total: long = 0;
    for (var square in results) {
        total = total + square;
    }
    assert(total == 338350, "the sum of the first 100 squares");

    # receive() returns a Result: an error once the channel is closed and
    # empty. A closed channel refuses new values.
    assert(results.receive().is_err(), "nothing left");
    assert(!results.send(1), "closed");

    var runner = async.Thread.spawn(run_tasks);
    tasks.send(() => { ran.fetch_add(1); });
    tasks.send(() => { ran.fetch_add(10); });
    tasks.close();
    runner.join();
    assert(ran.load() == 11, "the runner ran both tasks");
    return 0;
}
  • Channel<T>(capacity) makes a channel that holds at most capacity values; send waits while it is full. Channel<T>() (capacity 0) is unbounded, and send never waits.
  • send(value) queues a value and returns true, or returns false if the channel is closed. try_send(value) does not wait: it also returns false when a bounded channel is full.
  • receive() waits for the next value and returns a Result<T>: the value, or an error once the channel is closed and empty. try_receive() returns an error right away when nothing is waiting, and receive_for(milliseconds) when the time runs out.
  • close() ends the stream and wakes every waiting sender and receiver. Values already queued are still received. Only close() closes a channel: dropping it, or ending the threads that use it, does not.
  • for (var v in channel) receives until the channel is closed and empty. Several threads can loop over one channel; each value goes to one of them.
  • Values from one sender arrive in the order they were sent.
  • The count property and is_closed() report the channel’s state, which other threads may change right after.

send and try_send take a new value, which nothing on the sending thread references, and hand it to the receiver without copying it: a constructor call, a literal, a call that returns a fresh value, move(x) (a variable that holds the only reference) or copy(x) (when the sender keeps using x). A String, an Array or a struct of them can be sent, and changing the original after sending a copy does not change what was sent:

import async;

# A plain struct: no destructor and no raw pointer, so `copy` copies it field
# by field. Its Strings and Array are not thread-safe, but a Channel takes
# only new values, and a copy is one.
struct Order {
    String customer;
    Array<String> items;
    Order next;

    construct(this.customer = "") {
        items = Array<String>();
    }
}

const orders = async.Channel<Order>();
const letters = async.Atomic<long>(0);

count_letters() {
    for (var order in orders) {
        for (var item in order.items) {
            letters.fetch_add(item.length as long);
        }
    }
}

main(): int {
    # copy() makes a new object for everything that is not thread-safe.
    var tea = Order("ada");
    tea.items.append("tea");
    var copied = copy(tea);
    copied.items[0].append(" and cake");
    assert(tea.items[0] == "tea", "the original is unchanged");

    # Objects referenced twice stay shared in the copy, and cycles are kept.
    var a = Order("a");
    var b = Order("b");
    a.next = b;
    b.next = a;
    var pair = copy(a);
    assert(pair.next.next === pair, "the copy has the same cycle");
    a.next = __null;
    pair.next = __null;

    # send() takes a new value: send a copy to keep changing the order.
    var worker = async.Thread.spawn(count_letters);
    var order = Order("grace");
    order.items.append("coffee");
    orders.send(copy(order));
    order.items.append("biscuits");
    # Or hand the value itself over with `move`; `last` is null afterwards.
    var last = Order("hopper");
    last.items.append("tea");
    orders.send(move(last));
    orders.close();
    worker.join();
    assert(letters.load() == 9, "the worker saw only what was sent");
    return 0;
}

Sending a variable that something else may still reference is an error, and the editor offers both fixes:

main.pep:7:16: error: 'line' is not a new value, but parameter 'value' takes a value that nothing else references
    7 |     lines.send(line);
      |                ^~~~
      |                move(line)
runtime/async/channel.pep:173:17: note: 'value' is declared [Move] here
  173 |     send([Move] value: T): bool {
      |                 ^~~~~
note: hand it over with 'move(line)', or pass a new value, such as 'copy(line)'

A thread-safe value (a number, a [Shared] struct, a shared function value such as a ThreadBody) passes freely, which makes a channel of ThreadBody values a work queue. Since nothing is copied, a channel can carry any type, trait values and values that cannot be copied (an io.File) included.

The Channel reference documents every method.

Waiting on several channels with Select

A Select<E> waits on several channels at once and takes the first value any of them receives (or, with a send arm, sends to the first with room), like Go’s select. Each channel is added with a function that turns its values into an E, usually a case of an enum of your own that carries the value, so the channels may carry different types and a match handles the result:

import async;

# Every value the select can return: one case per channel.
enum Event {
    Line(text: String),
    Reading(value: long),
    Stop,
}

const lines = async.Channel<String>();
const readings = async.Channel<long>();
const stop = async.Channel<bool>();

read_lines() {
    lines.send("first");
    lines.send("second");
    lines.close();
}

read_sensor() {
    var i: long = 1;
    while (i <= 3) {
        readings.send(i * 10);
        i = i + 1;
    }
    readings.close();
}

main(): int {
    var select = async.Select<Event>();
    select.receive(lines, (text: String): Event => Event.Line(text));
    select.receive(readings, (value: long): Event => Event.Reading(value));
    select.receive(stop, (unused: bool): Event => Event.Stop);

    var threads = [async.Thread.spawn(read_lines), async.Thread.spawn(read_sensor)];
    var text = 0;
    var total: long = 0;
    var running = true;
    while (running) {
        match (select.wait()) {
            Ok(event) => {
                match (event) {
                    Line(line) => {
                        text = text + 1;
                    }
                    Reading(value) => {
                        total = total + value;
                    }
                    Stop => {
                        running = false;
                    }
                }
            }
            Err(error) => {
                # Every channel is closed and empty (`stop` included, below).
                running = false;
            }
        }
        if (text == 2 && total == 60) {
            stop.close();
        }
    }
    for (var thread in threads) {
        thread.join();
    }
    assert(text == 2 && total == 60, "every value arrived once");
    return 0;
}
  • receive(channel, wrap) adds a channel. wrap runs on the waiting thread, with the value received.
  • wait() waits until a channel has a value and returns it as an E, in a Result<E>. wait_for(milliseconds) gives up after that long, and try_wait() does not wait at all.
  • The value is taken under that channel’s own lock, so no other receiver can take it between “ready” and “take”.
  • When several channels have values, each wait starts just after the channel that fired last, so one busy channel cannot starve the others.
  • A channel that is closed and empty is skipped. Once every channel is, wait() returns an error (every channel is closed). A timeout’s error is the time ran out.
  • A Select belongs to the thread (or task) that made it; it is not [Shared]. In a task, waiting parks the task, and a cancel or the task’s timeout ends the wait.

A Select can also wait to send, like a case ch <- v in Go’s select. send(channel, value, sent) adds a send arm: wait() fires it once channel has room (a bounded channel that is full waits), and returns what sent returns. Here a loop keeps receiving jobs while each result waits for room in a bounded channel:

import async;

# What one wait did: a job arrived, or a result went out.
enum Step {
    Job(value: long),
    Sent,
}

const jobs = async.Channel<long>();
# Holds one result at a time, so sending can wait.
const results = async.Channel<long>(1);

feed() {
    var i: long = 1;
    while (i <= 5) {
        jobs.send(i);
        i = i + 1;
    }
    jobs.close();
}

main(): int {
    var select = async.Select<Step>();
    select.receive(jobs, (value: long): Step => Step.Job(value));
    var feeder = async.Thread.spawn(feed);
    var received = 0;
    var waiting = 0;
    var total: long = 0;
    while (received < 5 || waiting > 0) {
        match (select.wait()) {
            Ok(step) => {
                match (step) {
                    Job(value) => {
                        received = received + 1;
                        # Goes into `results` once it has room; until then
                        # the select keeps receiving jobs.
                        select.send(results, value * value, (): Step => Step.Sent);
                        waiting = waiting + 1;
                    }
                    Sent => {
                        waiting = waiting - 1;
                        # Take the result back out, so the next send has room.
                        total = total + results.receive().value();
                    }
                }
            }
            Err(error) => {
                panic(error.message);
            }
        }
    }
    feeder.join();
    assert(total == 55, "every square was sent once");
    return 0;
}
  • value must be a new value, as for Channel.send (move(x), copy(x), a constructor call, …). The select holds it and moves it into the channel only when that arm fires, under the channel’s lock, so it is never sent when another arm fires first.
  • A send arm fires once: after its value is sent it leaves the select, so the next wait() does not send it again. Add another with send for the next value.
  • A send arm whose channel is closed is skipped like a closed receive channel, and its value is dropped with the select.

The Select reference documents every method.

Mutexes and conditions

A Mutex<T> owns a value that one thread at a time can use. The way to reach the value is a with block:

import async;

# Each balance is an Atomic, but moving money changes both, and no thread
# may see it in neither account. A Mutex makes the pair of updates one step.
[Shared]
struct Accounts {
    const checking: async.Atomic<long>;
    const savings: async.Atomic<long>;
    construct(long checking = 0, long savings = 0) {
        this.checking = async.Atomic<long>(checking);
        this.savings = async.Atomic<long>(savings);
    }
}

const accounts = async.Mutex<Accounts>(Accounts(1000, 0));
const bad_totals = async.Atomic<long>(0);

move_to_savings() {
    var i = 0;
    while (i < 1000) {
        with (var a = accounts.lock()) {
            a.checking.fetch_sub(1);
            a.savings.fetch_add(1);
        }   # unlocked here, on every path out of the block
        i = i + 1;
    }
}

audit() {
    var i = 0;
    while (i < 1000) {
        with (var a = accounts.lock()) {
            if (a.checking.load() + a.savings.load() != 1000) {
                bad_totals.fetch_add(1);
            }
        }
        i = i + 1;
    }
}

main(): int {
    var mover = async.Thread.spawn(move_to_savings);
    var auditor = async.Thread.spawn(audit);
    mover.join();
    auditor.join();
    assert(bad_totals.load() == 0, "the auditor never saw money missing");

    # try_lock does not wait: the body runs if the lock was free, and the
    # else block runs if another thread held it.
    with (var a = accounts.try_lock()) {
        assert(a.savings.load() == 1000, "everything was moved");
    } else {
        assert(false, "no other thread holds the lock now");
    }
    return 0;
}
  • with (var v = mutex.lock()) { ... } waits until no other thread holds the lock, binds the value, and unlocks when the block ends by any path: the end of the body, return, break or continue. Everything a thread did while it held the lock is visible to the next thread that takes it.
  • try_lock() does not wait. With with (var v = mutex.try_lock()) { ... } else { ... }, the body runs if the lock was free and the else block runs if it was not.
  • A Mutex is not re-entrant: calling lock() again on the thread that holds it panics instead of deadlocking. is_held() reports whether the calling thread holds it.
  • A panic while a Mutex is held ends the program, so a Mutex is never left “poisoned”.
  • Taking a free lock is one atomic instruction, and a thread that finds it held spins briefly before it sleeps. Locks sleep on the operating system’s wait-on-address call (futex on Linux, __ulock_wait on macOS, WaitOnAddress on Windows), as Rust’s do.

T is a struct type: a struct of your own, an Array<String>, or a [Shared] struct whose Atomic fields must change as a unit. Mutex<int> is an error (use async.Atomic<int>). A Mutex of values that are not thread-safe is kept apart from the rest of the program by the compiler; see What a lock body may do.

  • Mutex<T>(value) and replace(value) take a new value: a constructor call, a string literal, a call that returns a fresh value, copy(x), or move(x), never a variable something else still references. The caller hands its reference over, so nothing is copied.
  • replace(value) swaps the whole value under the lock and returns the old one, which the caller then owns outright.

A value built in a local variable goes in with move, which hands over the variable’s reference instead of copying the value:

import async;

main(): int {
    # Build the list here, then hand it to the Mutex: `move` gives up
    # `lines`' reference, so nothing is copied and `lines` is empty after.
    var lines = Array<String>();
    lines.append("first");
    var log = async.Mutex<Array<String>>(move(lines));

    var line = "second";
    with (var list = log.lock()) {
        # `line` holds the only reference to its String, so it can move in.
        list.append(move(line));
    }

    # Swap in a new list; the old one comes back without a copy either.
    var next = Array<String>();
    next.append("third");
    var old = log.replace(move(next));
    print($"{old.count}");     # 2
    print(old[1]);               # second
    return 0;
}

A Condition<T> lets a thread that holds a Mutex sleep until another thread changes the value. Get one with mutex.condition():

import async;

# A baker makes pies and eaters take them. The eaters sleep on a Condition
# until there is a pie, or until the bakery closes.
[Shared]
struct Bakery {
    const pies: async.Atomic<long>;
    const closed: async.Atomic<int>;
    construct() {
        pies = async.Atomic<long>(0);
        closed = async.Atomic<int>(0);
    }
}

const bakery = async.Mutex<Bakery>(Bakery());
const changed = bakery.condition();
const eaten = async.Atomic<long>(0);

eat() {
    while (true) {
        with (var b = bakery.lock()) {
            # Wait in a loop: a wakeup can be spurious, and another eater
            # may take the pie first.
            while (b.pies.load() == 0 && b.closed.load() == 0) {
                changed.wait();   # unlocks, sleeps, locks again
            }
            if (b.pies.load() == 0) {
                return;   # closed and nothing left; the block unlocks
            }
            b.pies.fetch_sub(1);
        }
        eaten.fetch_add(1);
    }
}

main(): int {
    var eaters = Array<async.Thread>();
    var i = 0;
    while (i < 3) {
        eaters.append(async.Thread.spawn(eat));
        i = i + 1;
    }
    i = 0;
    while (i < 50) {
        with (var b = bakery.lock()) {
            b.pies.fetch_add(1);
            changed.notify_one();
        }
        i = i + 1;
    }
    with (var b = bakery.lock()) {
        b.closed.store(1);
        changed.notify_all();
    }
    for (var eater in eaters) {
        eater.join();
    }
    assert(eaten.load() == 50, "every pie was eaten once");
    return 0;
}
  • wait() must run while the caller holds the Mutex, normally inside its with block. It unlocks, sleeps until notified, and locks again before returning. It panics if the caller does not hold the Mutex.
  • A thread can wake without a notification (a spurious wakeup), and another thread may get to the value first, so always wait in a loop that re-checks what it waits for.
  • wait_for(milliseconds) gives up after that long and returns false.
  • notify_one() wakes one waiting thread, and notify_all() wakes them all.

wait_while(predicate) writes that loop for you: it waits while predicate returns true for the locked value, asking again after every wakeup, and returns with the Mutex still locked. wait_while_for(milliseconds, predicate) gives up after that long and returns false:

import async;

# A worker takes jobs from a queue, sleeping while it is empty.
const jobs = async.Mutex<Array<long>>(Array<long>());
const changed = jobs.condition();
const done = async.Atomic<long>(0);

work() {
    var taken = 0;
    while (taken < 5) {
        with (var queue = jobs.lock()) {
            # Waits while the predicate holds; spurious wakeups are handled.
            # The predicate is given the locked value: it cannot capture it.
            changed.wait_while((waiting: Array<long>): bool => waiting.count == 0);
            done.fetch_add(queue.remove_at(0));
            taken = taken + 1;
        }
    }
}

main(): int {
    var worker = async.Thread.spawn(work);
    var i: long = 1;
    while (i <= 5) {
        with (var queue = jobs.lock()) {
            queue.append(i);
            changed.notify_all();
        }
        i = i + 1;
    }
    worker.join();
    assert(done.load() == 15, "every job was taken once");

    with (var queue = jobs.lock()) {
        # Gives up after 10 milliseconds: nothing is queued.
        var ready = changed.wait_while_for(10, (waiting: Array<long>): bool => waiting.count == 0);
        assert(!ready, "the time ran out");
    }
    return 0;
}

The predicate is given the locked value as its parameter, since a closure cannot capture part of a locked value. It must be a closure written at the call, and the compiler checks its body as it checks a lock body (see What a lock body may do): it cannot keep any part of the value, or bring in a value from outside.

main.pep:10:26: error: 'queue[0]' is part of the value locked by 'queue' and cannot outlive the lock
   10 |             saved.append(queue[0]);
      |                          ^~~~~~~~
main.pep:9:27: note: locked here
    9 |         ready.wait_while((queue: Array<String>): bool => {
      |                           ^~~~~
note: 'String' is not thread-safe, so no part of the locked value may be kept outside the block; take 'copy(queue[0])'

The Mutex reference documents Mutex, MutexGuard and Condition.

What a lock body may do

A value that is not thread-safe counts its references plainly, so no thread may hold a reference into a Mutex’s value while the Mutex is unlocked. The compiler checks each with block over a lock so that none can:

import async;

# A Mutex of values that are not thread-safe: a list of Strings that
# several threads add to.
const names = async.Mutex<Array<String>>(Array<String>());

add_names(prefix: String) {
    var i = 0;
    while (i < 3) {
        with (var list = names.lock()) {
            list.append("worker");            # a new String: fine
            list.append(prefix + "!");        # `+` returns a new String: fine
            list.append(copy(prefix));        # a copy of an outside value: fine
            var n = list.count;             # a number leaves freely
            if (list.contains(prefix)) {      # comparing keeps nothing
                n = n + 1;
            }
        }
        i = i + 1;
    }
}

# Part of the locked value leaves the block only as a copy.
first_name(): String {
    with (var list = names.lock()) {
        if (list.count > 0) {
            return list[0].copy();
        }
    }
    return "";
}

main(): int {
    var a = async.Thread.spawn(() => { add_names("a"); });
    var b = async.Thread.spawn(() => { add_names("b"); });
    a.join();
    b.join();

    var first = first_name();
    first.append(" (mine)");   # changing the copy leaves the list alone
    with (var list = names.lock()) {
        assert(list.count == 18, "every name was added");
        assert(list[0] == "worker", "the locked String is unchanged");
    }

    # replace swaps in a new value and hands back the old one, which the
    # caller then owns outright.
    var old = names.replace(Array<String>());
    assert(old.count == 18, "the old list");

    # A thread's own value that is not thread-safe is reached through `with`.
    var scratch = async.ThreadLocal<String>();
    scratch.set("draft");
    with (var text = scratch.enter()) {
        text.append(" 2");
    }
    with (var text = scratch.enter()) {
        assert(text == "draft 2", "this thread's value");
    }
    return 0;
}

Inside the block, every value that is not thread-safe is either part of the locked value (the binding, its fields and elements, and what calls on them return), from outside (a parameter, this, or a variable declared before the block), or new. A new value may join either side; the other two must never meet:

  • Moving in. A call, constructor or assignment that would link an outside value with the locked one is an error. Pass a new value instead: copy(line), a literal, or the result of a call that returns a fresh value. A variable that holds the only reference to its value can hand it over with move(line).

    main.pep:8:21: error: 'line' cannot move into the value locked by 'names'
    8 |         list.append(line);
      |                     ^~~~
main.pep:7:28: note: locked here
    7 |     with (var list = names.lock()) {
      |                            ^~~~
note: 'String' is not thread-safe, so the locked value cannot share it with the rest of the program; pass 'copy(line)'

  • Moving out. Storing part of the locked value outside the block, or returning it from the block, is an error; take a copy (copy(list[0]), or list[0].copy(), which returns a fresh value).

    main.pep:8:17: error: 'list[0]' is part of the value locked by 'names' and cannot outlive the lock
    8 |         first = list[0];
      |                 ^~~~~~~

  • Closures. A closure written in the block cannot capture part of the locked value, since a function value could carry it out.

  • Two locks. Values locked by two nested blocks stay apart as well.

  • Numbers pass freely, as do every thread-safe value and every new one.

Which calls link which values comes from what the called function does: the compiler works out, for every function, whether it keeps an argument (as append keeps its element) and whether its result shares anything with its arguments. A result that shares nothing, such as that of +, substr, copy() or a constructor, is fresh, and may cross in either direction. The language server shows “Returns a fresh value” when hovering over such a function. A call through a function value, a value an extern function hands back, and a value read through raw memory that a [Shared] struct holds are never fresh: something else may reference them. The runtime takes the few values it hands out that nothing else references out of raw memory with move(p[i]), a promise the compiler cannot check and allows only in the runtime: a channel’s receive() (the value send handed over) and a Mutex’s replace().

A function that returns a tuple has one such fact per element: after var (fresh, same) = split(line);, fresh may move into a locked value while same, which split returned from its argument, may not.

The guard a lock() returns can only be the scope of a with header: storing it in a variable or field is an error, so no lock outlives its block.

Two attributes carry these rules, and code outside the runtime may use them too:

  • [Isolated] on a struct makes it a guard: it can only be the scope of a with header, and the block is checked as above. MutexGuard<T> and ThreadLocalGuard<T> are [Isolated].
  • [Move] on a parameter (construct([Move] T value)) requires a new value for it at every call: a constructor call, a literal, a fresh result, copy(x), move(x), or any thread-safe value. The caller hands its reference over, so the function receives it without a count step.

Checking confinement at run time

Raw memory (__ptr<T>) and extern code are promises the compiler cannot check, so a program that uses them can still keep a reference into a locked value. Set the environment variable PEP_CHECK_SYNC when running a program to check at every unlock that nothing outside the locked value refers into it:

$ PEP_CHECK_SYNC=1 ./main
PEP_CHECK_SYNC: an object in the value locked by a 'Mutex<Box>' is still referenced from outside it

The check copies the locked value to count the references inside it, so it costs time and memory in proportion to the value’s size at each unlock; it is off unless the variable is set. It checks a Mutex<T> whose T is not thread-safe and can be copied.

Reader/writer locks

An RwLock<T> lets many threads read a value at once, or one thread write it:

import async;

# A price list that many threads read and one thread now and then updates.
# A writer changes both prices together, so a reader never sees one updated
# and the other not.
[Shared]
struct Prices {
    const coffee: async.Atomic<long>;
    const tea: async.Atomic<long>;
    construct() {
        coffee = async.Atomic<long>(300);
        tea = async.Atomic<long>(250);
    }
}

const prices = async.RwLock<Prices>(Prices());
const mismatches = async.Atomic<long>(0);

check_prices() {
    var i = 0;
    while (i < 1000) {
        with (var p = prices.read()) {   # many readers at once
            if (p.coffee.load() - p.tea.load() != 50) {
                mismatches.fetch_add(1);
            }
        }
        i = i + 1;
    }
}

raise_prices() {
    var i = 0;
    while (i < 100) {
        with (var p = prices.write()) {   # no reader sees the middle
            p.coffee.fetch_add(10);
            p.tea.fetch_add(10);
        }
        i = i + 1;
    }
}

main(): int {
    var threads = Array<async.Thread>();
    var i = 0;
    while (i < 3) {
        threads.append(async.Thread.spawn(check_prices));
        i = i + 1;
    }
    threads.append(async.Thread.spawn(raise_prices));
    for (var thread in threads) {
        thread.join();
    }
    assert(mismatches.load() == 0, "no reader saw a half-done update");
    with (var p = prices.read()) {
        assert(p.coffee.load() == 1300, "every raise landed");
    }
    return 0;
}
  • with (var v = lock.read()) { ... } waits until no writer holds or waits for the lock; any number of threads may read at the same time.
  • with (var v = lock.write()) { ... } waits until no other thread holds the lock at all.
  • try_read() and try_write() do not wait: the else block runs when the lock is not free.
  • replace(value) swaps the whole value under the write lock and returns the old one.
  • Writers come first: while a writer waits, new readers wait behind it, so readers cannot keep it out forever. A thread that reads must therefore not ask to read again before its block ends, and asking to write while holding the lock panics instead of deadlocking.

T must be thread-safe, as for Atomic<T>: typically a [Shared] struct whose Atomic fields a writer changes together. Readers that hold the value at once each copy a reference to it, so a value that is not thread-safe, such as an Array<String>, would have its reference counts changed by several threads at a time; it is a compile error at the type argument. Put such a value in a Mutex<T>. When readers only need the latest value and writers replace it whole, an Atomic<T> holding a [Shared] struct does the same job without a lock.

Semaphores, barriers and latches

Three types coordinate threads by counting. Each is [Shared], so keep it in a module-scope variable or pass it to threads:

  • A Semaphore(permits) lets at most permits threads hold a permit at once. acquire() waits for a free permit and takes it, try_acquire() takes one only if one is free, and release() gives one back. with (semaphore.permit()) { ... } holds a permit for the block and gives it back on every path out.
  • A Barrier(count) makes groups of count threads meet: wait() sleeps until count threads have called it, then all go on together. It returns true on exactly one of them, the leader, and resets for the next round.
  • A Latch(count) opens once count_down() has been called count times. wait() sleeps until it is open, and is_ready() asks without waiting. It is one-shot: counting down an open latch panics.
import async;

const WORKERS = 4;

# At most two workers use the printer at once.
const printer = async.Semaphore(2);
const printing = async.Atomic<long>(0);
const most_printing = async.Atomic<long>(0);

# Every worker finishes a step before any starts the next.
const step_done = async.Barrier(WORKERS);
const steps_led = async.Atomic<long>(0);

# The main thread waits until every worker has set up.
const ready = async.Latch(WORKERS);

work() {
    ready.count_down();
    var step = 0;
    while (step < 3) {
        with (printer.permit()) {
            var now = printing.fetch_add(1) + 1;
            if (now > most_printing.load()) {
                most_printing.store(now);
            }
            printing.fetch_sub(1);
        }   # the permit is given back here
        if (step_done.wait()) {
            steps_led.fetch_add(1);   # one leader per step
        }
        step = step + 1;
    }
}

main(): int {
    var workers = Array<async.Thread>();
    var i = 0;
    while (i < WORKERS) {
        workers.append(async.Thread.spawn(work));
        i = i + 1;
    }
    ready.wait();
    assert(ready.is_ready(), "every worker counted down");
    for (var worker in workers) {
        worker.join();
    }
    assert(most_printing.load() <= 2, "the semaphore let two in at most");
    assert(steps_led.load() == 3, "each step had exactly one leader");
    return 0;
}

Each is a Mutex over its counts with a Condition to sleep on, so in a task their waits park the task, not its worker.

Once and Lazy

Once runs a function exactly once, however many threads ask, and Lazy<T> builds a value the first time a thread asks for it:

import async;

[Shared]
struct Settings {
    const workers: long;
    construct(this.workers = 1) {}
}

const builds = async.Atomic<long>(0);

load_settings(): Settings {
    builds.fetch_add(1);
    return Settings(async.logical_cpus() as long);
}

# Built by the first thread that calls get(), not before main runs.
const settings = async.Lazy<Settings>(load_settings);

# Runs initialize() once, however many threads call it.
const setup = async.Once();
const setups = async.Atomic<long>(0);

initialize() {
    setups.fetch_add(1);
}

work() {
    setup.call(initialize);
    assert(settings.get().workers >= 1, "every thread sees the settings");
}

main(): int {
    assert(!settings.is_built(), "nothing is built before first use");
    var workers = Array<async.Thread>();
    var i = 0;
    while (i < 4) {
        workers.append(async.Thread.spawn(work));
        i = i + 1;
    }
    for (var worker in workers) {
        worker.join();
    }
    assert(builds.load() == 1, "the settings were built once");
    assert(setups.load() == 1, "initialize ran once");
    assert(setup.is_done(), "done");
    return 0;
}
  • once.call(body) runs body if no call has run one yet. Threads that call it while the body runs wait for it, and every caller sees what the body did. is_done() reports whether it has run.
  • Lazy<T>(initializer) takes a shared () => T function. The first get() runs it; later calls, from any thread, return the same value. is_built() reports whether it has run.
  • A Lazy<T> is the usual way to give a module-scope variable a value that is expensive to build, or that cannot be built before main runs. Its T must be thread-safe, since every thread gets the same value.

The Once reference has the details.

Limitations

  • Lazy<T> holds only thread-safe values.
  • A value received from a channel moves into a locked value directly (list.append(jobs.receive().value())), but not through a variable declared outside the block: such a variable is an outside value, whatever it holds, so it needs copy, or move when it holds the only reference (list.append(move(job))).
  • There is no reader-writer lock, and no barrier or semaphore. A Mutex is not re-entrant.
  • The value a Select’s wait() returns is not a new value for a lock body, since wrap is a function value the compiler cannot follow; copy it to move it into a locked value.
  • Deadlocks are not detected, except locking a Mutex twice on one thread, which panics.