Chapter 13: Concurrency

What you'll learn: async functions and await, structured concurrency with await group, async generators and for await, atomic {} blocks, offloading work with spawnBlocking, worker scripts with spawnWorker, and threads — and, for each, what the Loup 1.25 interpreter actually does today.

The Language Reference (§2-1) describes a cooperative event loop: an async call starts a task, tasks interleave at every await, and several waits overlap. The syntax and the type rules of that design are implemented, and programs written against them run correctly. The scheduling is not there yet:

How concurrency runs in Loup 1.25. loup run executes an async function's body immediately, to completion, when it is called; await then hands back the finished result. Tasks therefore run one after another, in call order, and two waits never overlap. spawnBlocking runs its closure at the call site. Threads exist, but take turns under a global lock, so they do not run CPU work in parallel. Write your code against the specified model — it will keep working when the scheduler lands — but do not expect a speedup from it today.

Async Functions and await

An async func is declared with a Task[T] return type, and its return statements produce a plain T. await takes a Task[T] and gives back the T:

@mode(host.scripted)

async func loadConfig(path: string): Task[string] {
    return i"port = 8080  # from {$ path $}";
}

async func describe(path: string): Task[string] {
    let text = await loadConfig(path);
    return text.toUpperCase();
}

let summary = await describe("app.toml");
println(summary);

Listing 13-1: An async function awaited by another, and by the script

PORT = 8080  # FROM APP.TOML

await is allowed inside an async func and at the top level of the script you run. The bare form async func loadConfig(path: string): string is accepted too and means the same thing, but Task[string] is the canonical spelling: it says what a caller receives.

Mistake: await in an ordinary function

@mode(host.scripted)

async func loadConfig(path: string): Task[string] {
    return i"port = 8080  # from {$ path $}";
}

func describe(path: string): string {
    let text = await loadConfig(path);
    return text.toUpperCase();
}

println(describe("app.toml"));

main.lpscr:8:16: error[E_AWAIT_NOT_ASYNC]: E_AWAIT_NOT_ASYNC: `await` is only valid inside an `async func` body, an entry script's top level, or the REPL (§2-1.1)

A function that waits must itself be async, so that its caller knows it gets a task back. Declare describe as async func describe(path: string): Task[string] and call it with await describe("app.toml"), as in Listing 13-1.

Mistake: forgetting await

Calling an async func gives you a Task[T], not a T:

@mode(host.scripted)

async func loadConfig(path: string): Task[string] {
    return i"port = 8080  # from {$ path $}";
}

let text = loadConfig("app.toml");
println(text.toUpperCase());

main.lpscr:8:9: error[E0001]: type 'Task[string]' has no method 'toUpperCase'

The fix is let text = await loadConfig("app.toml");. If you annotate the binding — let text: string = loadConfig("app.toml"); — the message names both types instead: cannot assign 'Task[string]' to binding 'text' of type 'string'.

What Runs When

In the specified model, calling an async func starts a task and returns at once, so the usual way to overlap two waits is to call both functions first and await them afterwards. This program prints when each body runs:

@mode(host.scripted)

async func fetchUser(id: int): Task[string] {
    println(i"  fetching user {$ id $}");
    return i"user-{$ id $}";
}

async func fetchPostCount(id: int): Task[int] {
    println(i"  counting posts for {$ id $}");
    return id * 3;
}

println("start");
let userTask = fetchUser(1);
println("fetchUser called");
let postsTask = fetchPostCount(1);
println("fetchPostCount called");

let user = await userTask;
let posts = await postsTask;
println(i"{$ user $} has {$ posts $} posts");

Listing 13-2: Starting two tasks before awaiting them

start
  fetching user 1
fetchUser called
  counting posts for 1
fetchPostCount called
user-1 has 3 posts

Each body runs completely at the moment it is called, before the next line of the script. The result is correct, and the program is written the right way for the day tasks really overlap; today it takes as long as running the two calls in sequence.

Errors Cross await

An exception thrown inside a task comes out of the await that receives its result, so try/catch works exactly as in synchronous code:

@mode(host.scripted)

class FetchError : Exception {}

async func download(url: string): Task[string] {
    if (url.startsWith("bad:")) {
        throw FetchError(i"cannot reach {$ url $}");
    }
    return i"<page from {$ url $}>";
}

async func main(): Task[void] {
    try {
        let page = await download("bad://example");
        println(page);
    } catch (FetchError as e) {
        println(i"download failed: {$ e.message $}");
    }
}

await main();

Listing 13-3: Catching an exception from an awaited task

download failed: cannot reach bad://example

A script can call its own main() like any other function. Wrapping the work in an async func main is a habit worth keeping: await group and for await (below) are only allowed inside an async func.

await group: Structured Concurrency

await group { … } starts several child tasks with spawn and does not finish until all of them have finished. binding = spawn expr stores each child's result in a variable declared before the group:

@mode(host.scripted)

async func wordCount(text: string): Task[int] {
    return text.split(" ").length();
}

async func compare(): Task[void] {
    var left = 0;
    var right = 0;
    await group {
        left = spawn wordCount("the quick brown fox");
        right = spawn wordCount("jumps over the lazy dog");
    }
    println(i"left: {$ left $}, right: {$ right $}");
}

await compare();

Listing 13-4: Two child tasks in an await group

left: 4, right: 5

If a child throws, the group stops and the exception comes out of await group:

@mode(host.scripted)

async func checkAge(age: int): Task[int] {
    if (age < 0) {
        throw ValueError(i"negative age: {$ age $}");
    }
    return age;
}

async func total(): Task[int] {
    var a = 0;
    var b = 0;
    await group {
        a = spawn checkAge(30);
        b = spawn checkAge(-2);
    }
    return a + b;
}

try {
    println(await total());
} catch (ValueError as e) {
    println(i"group failed: {$ e.message $}");
}

Listing 13-5: A failing child task ends the group

group failed: negative age: -2

The specification adds that the group cancels the other children and waits for them to settle before rethrowing. With eager execution there is nothing left running to cancel: a child that comes after the failing one simply never starts.

spawn is only accepted directly inside an await group block; anywhere else it is 'spawn' can only appear inside an 'await group' block. (The Language Reference allows spawn in any async context.)

Mistake: declaring the targets with let

The Language Reference's own example declares the result bindings with let and no initializer, and assigns them in the group. The 1.25 compiler rejects that:

@mode(host.scripted)

async func wordCount(text: string): Task[int] {
    return text.split(" ").length();
}

async func compare(): Task[void] {
    let left: int;
    let right: int;
    await group {
        left = spawn wordCount("the quick brown fox");
        right = spawn wordCount("jumps over the lazy dog");
    }
    println(i"left: {$ left $}, right: {$ right $}");
}

await compare();

main.lpscr:11:9: error[E0001]: cannot assign to immutable binding 'left'
main.lpscr:12:9: error[E0001]: cannot assign to immutable binding 'right'

Declare each target with var and a starting value, as Listing 13-4 does.

⚠ Planned. TaskGroup — TaskGroup.collect(items, body), race, withTimeout, withCancellation and collectBounded (§2-1.3) — is not defined in Loup 1.25 (undefined identifier `TaskGroup` ). For a list of inputs, loop and await each call, or spawn a fixed number of children in an await group.

Async Generators and for await

An async gen func produces values over time; for await consumes them. Unlike ordinary async calls, an async generator is lazy: each value is produced when the loop asks for it.

@mode(host.scripted)

async gen func pages(count: int): AsyncIterable[string] {
    var n = 1;
    while (n <= count) {
        println(i"  loading page {$ n $}");
        yield i"page {$ n $}";
        n = n + 1;
    }
}

async func main(): Task[void] {
    for await (page in pages(3)) {
        println(i"read {$ page $}");
    }
}

await main();

Listing 13-6: An async generator consumed with for await

  loading page 1
read page 1
  loading page 2
read page 2
  loading page 3
read page 3

Mistake: a reserved word as a parameter name

A natural name for a starting point is from:

@mode(host.scripted)

async gen func countdown(from: int): AsyncIterable[int] {
    var n = from;
    while (n > 0) {
        yield n;
        n = n - 1;
    }
}

main.lpscr:3:26: error[E0001]: `from` is a reserved word (§1-3.4) and cannot name a parameter
  note: rename the parameter — e.g. `from_value`

from begins an import (from io.fs import (…)), so it cannot name anything. Rename the parameter — start reads well:

@mode(host.scripted)

async gen func countdown(start: int): AsyncIterable[int] {
    var n = start;
    while (n > 0) {
        yield n;
        n = n - 1;
    }
}

async func main(): Task[void] {
    for await (n in countdown(3)) {
        println(n);
    }
}

await main();

3
2
1

Mistake: for await over a list

for await needs an AsyncIterable[T]. A list is an ordinary iterable:

@mode(host.scripted)

async func main(): Task[void] {
    for await (n in [1, 2, 3]) {
        println(n);
    }
}

await main();

main.lpscr:4:5: error[E_NOT_ASYNC_ITERABLE]: E_NOT_ASYNC_ITERABLE: `for await` needs an `AsyncIterable[T]`, found 'list[int]' (§2-1.4)

Use a plain for (n in [1, 2, 3]) for a list.

⚠ Planned. The async-iterator combinators map, filter, take, merge and zip (§2-1.5) are not implemented (type 'AsyncIterable[int]' has no method 'filter'). Filter inside the for await body instead.

atomic {}: No Suspension Here

In the specified model, another task can run at every await. A check followed by an action with an await between them — "is the item in stock? … then take it" — can therefore act on stale information. An atomic {} block (§2-1.7) is a region in which the compiler forbids every suspension point, so the check and the action happen together:

@mode(host.scripted)

var stock: dict[string, int] = {"apple": 3};

async func audit(item: string): Task[void] {
    println(i"  audit: one {$ item $} taken");
}

async func take(item: string): Task[bool] {
    var taken = false;
    atomic {
        let have = stock.get(item) ?? 0;
        if (have > 0) {
            stock[item] = have - 1;
            taken = true;
        }
    }
    if (taken) {
        await audit(item);
    }
    return taken;
}

println(await take("apple"));
println(stock);

Listing 13-7: A check-then-act sequence inside atomic {}

  audit: one apple taken
true
{"apple": 2}

Mistake: awaiting inside atomic {}

@mode(host.scripted)

var stock: dict[string, int] = {"apple": 3};

async func audit(item: string): Task[void] {
    println(i"  audit: one {$ item $} taken");
}

async func take(item: string): Task[bool] {
    var taken = false;
    atomic {
        let have = stock.get(item) ?? 0;
        if (have > 0) {
            await audit(item);
            stock[item] = have - 1;
            taken = true;
        }
    }
    return taken;
}

println(await take("apple"));

main.lpscr:14:13: error[E_ATOMIC_AWAIT]: E_ATOMIC_AWAIT: await expression is not allowed inside atomic {} block

The point of the block is that nothing can interleave inside it, and an await is exactly where interleaving happens. Calling an async func without awaiting it is refused as well (call to async function 'audit' is not allowed inside atomic {} block). Record what you need inside the block and do the waiting after it, as Listing 13-7 does.

atomic {} is a compile-time guarantee about one task's suspension points, not a lock: it does nothing about threads.

Offloading Work with spawnBlocking

spawnBlocking(body) hands a synchronous closure to a pool of worker threads and returns a Task[T] for its result. It is meant for CPU-heavy or blocking work that would otherwise hold up every other task. It is a built-in of every script; no import is needed:

@mode(host.scripted)

func checksum(data: list[int]): int {
    var total = 0;
    for (b in data) {
        total = (total * 31 + b) % 1000003;
    }
    return total;
}

let data = [3, 1, 4, 1, 5, 9, 2, 6];
let sum = await spawnBlocking(func() => checksum(data));
println(i"checksum: {$ sum $}");

Listing 13-8: Offloading a computation with spawnBlocking

checksum: 693191

In Loup 1.25 the closure runs at the call site, before spawnBlocking returns, so two spawnBlocking calls take as long as running the two closures in a row. The rules below are enforced already.

Mistake: importing it from sync.thread

The Language Reference declares spawnBlocking in the sync.thread module, and its examples import it from there:

@mode(host.scripted)

from sync.thread import (spawnBlocking);

let sum = await spawnBlocking(func() => 6 * 7);
println(sum);

main.lpscr:3:6: error[E0001]: `sync.thread` is not available in the scripted runtime
  note: `Mutex[T]`/`RwLock[T]` (`sync.thread`) require OS-thread primitives under `@mode(host.dual)` or `@mode(host.native)` (Runtime §2-1.8.2); they are not available in the scripted runtime.

No sync.* module exists in Loup 1.25 — the import fails under loup run in host.dual too, whatever the note suggests. Delete the import: spawnBlocking (and spawnWorker, below) are prelude built-ins.

Mistake: capturing a class instance

The closure is meant to run on another thread, so everything it captures must be Sendable: numbers, strings, collections of them, structs and enums are; class instances are not unless they say so.

@mode(host.scripted)

class Tally {
    pub var count: int = 0;
}

let tally = Tally();
let next = await spawnBlocking(func() => tally.count + 1);
println(next);

main.lpscr:8:18: error[E_BLOCKING_CAPTURE]: E_BLOCKING_CAPTURE: captured variable 'tally' of type 'Tally' does not implement Sendable — every value captured by a closure passed to spawnBlocking must be Sendable
  note: v1.20 §2-1.8.1: classes are never implicitly Sendable. Declare `class Tally implements Sendable` (every `var` field must then be Mutex[T], RwLock[T] or Atomic[T]), or pass the data as a Sendable value.

A class instance is a shared, mutable reference; two threads holding it could race. Copy out the value the closure needs and capture that:

@mode(host.scripted)

class Tally {
    pub var count: int = 0;
}

let tally = Tally();
let current = tally.count;
let next = await spawnBlocking(func() => current + 1);
println(next);

1

The note's other suggestion, a Sendable class whose var fields are Mutex[T] or Atomic[T], is not available in scripts yet (see "What Is Planned" below).

Mistake: await inside the closure

The closure runs synchronously on a pool thread, so it cannot suspend:

@mode(host.scripted)

async func fetchSize(): Task[int] {
    return 1024;
}

let size = await spawnBlocking(func() => await fetchSize());
println(size);

main.lpscr:7:42: error[E_AWAIT_IN_BLOCKING]: E_AWAIT_IN_BLOCKING: `await` inside a `spawnBlocking` body — the body runs synchronously on the blocking pool and cannot suspend; await the result OUTSIDE (`await spawnBlocking(...)`) or use `spawn` for async work
  note: v1.20 §2-1.8.5
main.lpscr:7:42: error[E_AWAIT_NOT_ASYNC]: E_AWAIT_NOT_ASYNC: `await` is only valid inside an `async func` body, an entry script's top level, or the REPL (§2-1.1)

Await asynchronous work directly (let size = await fetchSize();); keep spawnBlocking for synchronous code.

Worker Scripts with spawnWorker

A worker is a separate script with its own state that exchanges messages with the script that started it (§2-1.10.2). The worker exports workerMain, which receives a WorkerChannel; channel.receive() returns the next message, or None when there are no more, and channel.reply(msg) sends a message back:

@mode(host.scripted)

pub async func workerMain(channel: WorkerChannel): Task[void] {
    while (true) {
        let msg = await channel.receive();
        if (msg == None) {
            break;
        }
        let text = msg as string;
        channel.reply(text.toUpperCase());
    }
}

Listing 13-9: upper_worker.lpscr, a worker that upper-cases each message

The main script starts it with spawnWorker, sends it messages, and receives the replies:

@mode(host.scripted)

async func main(): Task[void] {
    let worker = await spawnWorker("./upper_worker.lpscr");
    for (word in ["alpha", "beta", "gamma"]) {
        worker.send(word);
    }

    var results: list[string] = [];
    while (true) {
        let reply = await worker.receive();
        if (reply == None) {
            break;
        }
        results.push(reply as string);
    }
    await worker.join();
    println(results);
}

await main();

Listing 13-10: main.lpscr starts the worker and collects its replies (a two-file example)

["ALPHA", "BETA", "GAMMA"]

Messages are typed Any in Loup 1.25, so the worker and the host cast what they receive (msg as string).

Workers run in batch mode today: the first receive() or join() runs workerMain to completion over the messages sent so far, and the worker has then finished. So send everything first, then receive. A send after the first receive fails with Uncaught WorkerTerminatedError: send on a terminated worker.

⚠ Planned. The specified worker API — ScriptWorker[M].spawn(path) imported from concurrent, a Sendable message type M checked at compile time (E_TRANSFER_NONSENDABLE), workers that run in parallel with the host and exchange messages back and forth — is not in Loup 1.25. spawnWorker and the untyped ScriptWorker/WorkerChannel are the shipped subset.

Threads

Thread.spawn(body) starts a thread; Thread.spawnWith(body) starts one whose closure returns a value, collected with join(). They are built in; no import is needed:

@mode(host.scripted)

func countVowels(text: string): int {
    var n = 0;
    for (ch in text.chars()) {
        if ("aeiou".contains(ch)) {
            n = n + 1;
        }
    }
    return n;
}

let lines = ["call me ishmael", "it was the best of times", "happy families are all alike"];
var handles: list[ThreadHandle[int]] = [];
for (line in lines) {
    handles.push(Thread.spawnWith(func() => countVowels(line)));
}

var total = 0;
for (h in handles) {
    total = total + h.join();
}
println(i"vowels: {$ total $}");

Listing 13-11: Fan out to threads, then join and combine

vowels: 23

The interpreter runs scripts under a global lock, so only one thread executes Loup code at a time. Fan-out-then-join is correct, but CPU-bound threads take as long as running the work in sequence. Genuine parallel threads are a host.native feature; see The Loup Native Book.

Mistake: sharing a class instance with a thread

@mode(host.scripted)

class Tally {
    pub var vowels: int = 0;

    pub modifier func add(n: int): void {
        this.vowels = this.vowels + n;
    }
}

func countVowels(text: string): int {
    var n = 0;
    for (ch in text.chars()) {
        if ("aeiou".contains(ch)) {
            n = n + 1;
        }
    }
    return n;
}

let tally = Tally();
let worker = Thread.spawn(func() {
    safe { tally.add(countVowels("call me ishmael")); }
});
worker.join();
println(tally.vowels);

main.lpscr:22:14: error[E_NON_SENDABLE_CAPTURE]: E_NON_SENDABLE_CAPTURE: captured variable 'tally' of type 'Tally' does not implement Sendable — every value captured by a closure passed to Thread.spawn must be Sendable
  note: v1.20 §2-1.8.1: classes are never implicitly Sendable. Declare `class Tally implements Sendable` (every `var` field must then be Mutex[T], RwLock[T] or Atomic[T]), or pass the data as a Sendable value.

This is the data race the type checker exists to prevent: two threads updating one object with no lock. Do as Listing 13-11 does — each thread returns its result, and the main thread combines the results after join().

What Is Planned

These parts of §2-1 and §2-2 of the Language Reference are not in Loup 1.25. They have no runnable listings in this book.

Summary

• async func f(): Task[T] returns a task; await turns a Task[T] into a T. await is allowed in an async func and at a script's top level; await group and for await need an async func.
• In Loup 1.25 an async body runs completely when it is called. Programs written for the specified model run correctly, but waits do not overlap.
• Exceptions thrown in a task come out of the await, so try/catch works unchanged.
• await group { x = spawn f(); } runs child tasks and joins them; declare the targets with var. A failing child's exception comes out of the group.
• async gen func with yield produces values lazily for for await.
• atomic {} forbids suspension points (E_ATOMIC_AWAIT), closing check-then-act races between tasks.
• spawnBlocking and spawnWorker are prelude built-ins; do not import them from sync.thread. Their closures and captures must be Sendable (E_BLOCKING_CAPTURE, E_NON_SENDABLE_CAPTURE).
• Threads take turns under a global lock in scripts. For real parallelism, move the code to host.native.
• TaskGroup, actors, combinators, sync.*, typed parallel workers, for concurrent and shared memory are planned.

Try It Yourself

1. Write an async func checkLink(url: string): Task[bool] that returns true unless the URL starts with "http://", and check three URLs inside one await group. Then make one of them throw and catch the error around the group.
2. Write an async gen func lines(text: string): AsyncIterable[string] that yields one line of a multi-line string at a time, and number the lines with for await.
3. Move the await in Listing 13-7 into the atomic {} block and read the diagnostic. Then try calling audit(item) there without await.
4. Write a worker script that receives numbers and replies with their squares, and a host that sends 1 to 5 and prints the replies.