Chapter 20: Concurrency with OS threads

A native Loup binary can run code on several operating-system threads at once. This chapter covers starting threads and collecting their results, the Sendable rules that decide what may cross a thread boundary, the locks and atomics for shared state, thread pools, and which parts of the specified concurrency library are still planned. The normative text is the Language Reference §2-1.8 (threads) and §2-2 (shared memory). Chapter 21 covers async.

The thread API is the same on every delivery path, but what runs in parallel is not:

A program that spawns work and then joins it gives the same results on all three. A program in which one thread waits for another to do something (a flag, a handshake) works natively and under Stage0, but would wait for ever where bodies run at the spawn site.

Thread, Mutex, RwLock, Atomic and ThreadPool are built into the compiler; you use them without an import. The specification places them in a sync.thread module, which does not exist yet.

Mistake: importing the thread types

@mode(host.native)

from sync.thread import (Thread, Mutex);

func main(): void {
    let t = Thread.spawn(func(): void { println("worker"); });
    t.join();
}

main.lphc: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.

The note is misleading (this is a host.native file); the real reason is that no sync.* module ships in 1.25.1. Delete the import line and the program runs.

Starting threads and collecting results

Thread.spawnWith starts a thread that computes a value; join() waits for the thread and returns the value:

@mode(host.native)

func sumRange(lo: int, hi: int): int {
    var total = 0;
    for (i in lo..hi) {
        total = total + i;
    }
    return total;
}

func main(): void {
    let low = Thread.spawnWith(func(): int { return sumRange(0, 500_000); });
    let high = Thread.spawnWith(func(): int { return sumRange(500_000, 1_000_000); });
    let total: int = low.join() + high.join();
    println(total);
}

Listing 20-1: Splitting a sum across two threads

499999500000

Thread.spawn is the form for a body that returns nothing. Its handle is a Thread, whose join() returns void, so using it for a computation fails at the first use of the result:

@mode(host.native)

func main(): void {
    let t = Thread.spawn(func(): int { return 6 * 7; });
    println(t.join());
}

main.lphc:5:13: error[E0001]: argument 1 of 'println': expected 'Any', found 'void'

Use Thread.spawnWith, which returns a ThreadHandle whose join() returns the body's value.

The other members: Thread.sleep(ms), Thread.hardwareConcurrency() (the number of cores), Thread.current(), t.detach(), and the field name: Option[string]. A thread may be joined once; a second join(), or a join() after detach(), is a LifetimeError ("join() on an already-joined thread — a thread may be joined at most once").

Exceptions in a thread

An exception that escapes a thread's body is raised again by join(), so the thread that waits for the result also handles the failure:

func parsePort(text: string): int {
    if (text == "") {
        throw ValueError("empty port");
    }
    return text.length();
}

func main(): void {
    let worker = Thread.spawnWith(func(): int { return parsePort(""); });
    println("worker started");
    try {
        println(worker.join());
    } catch (ValueError as e) {
        println(i"worker failed: {$ e.message $}");
    }
}

Listing 20-2: A thread's exception surfaces at join()

worker started
worker failed: empty port

A failed thread that is never joined loses its exception, so join every thread whose failure matters.

Sendable: what may cross a thread boundary

A closure passed to Thread.spawn, Thread.spawnWith or ThreadPool.run may only capture Sendable values (§2-1.8.1). These are Sendable without any declaration:

• the primitive types, string and bytes;
• list, dict, set, Option and Result of Sendable types;
• structs and enums whose fields and payloads are all Sendable (unless they carry a lifetime, chapter 18);
• Mutex[T], RwLock[T] and Atomic[T].

A class is never Sendable by default. It becomes Sendable by declaring implements Sendable, and then every var field must be a Mutex, RwLock or Atomic; let fields may be of any Sendable type. A synchronized class may also declare it (chapter 15), and @unsafe("sendable: …") overrides the check (chapter 19).

Here is a request-statistics object that several threads update. An ordinary class with plain var fields would be rejected at the capture (E_NON_SENDABLE_CAPTURE, chapter 15); with Atomic and Mutex fields it is accepted:

class Stats implements Sendable {
    pub let requests: Atomic[int] = Atomic.new(0);
    pub let slowest: Mutex[int] = Mutex.new(0);

    pub func record(latencyMs: int): void {
        this.requests.fetchAdd(1);
        this.slowest.withLock(func(g: MutexGuard[int]): void {
            if (latencyMs > g.get()) {
                g.set(latencyMs);
            }
        });
    }
}

func main(): void {
    let stats = Stats();
    var workers: list[Thread] = [];
    for (batch in [[12, 40, 7], [33, 5], [90, 1, 2, 3]]) {
        workers.push(Thread.spawn(func(): void {
            for (ms in batch) {
                stats.record(ms);
            }
        }));
    }
    for (w in workers) {
        w.join();
    }
    let slowest = stats.slowest.withLock(func(g: MutexGuard[int]) => g.get());
    println(i"{$ stats.requests.load() $} requests, slowest {$ slowest $} ms");
}

Listing 20-3: A Sendable class whose mutable state is atomic or locked

9 requests, slowest 90 ms

The fields are let: the references to the atomic and the mutex never change; what changes is the state inside them, through operations that are safe to call from any thread.

Mistake: a plain var field on a Sendable class

@mode(host.native)

class Stats implements Sendable {
    var requests: int = 0;
}

func main(): void {
}

main.lphc:4:5: error[E_SENDABLE_UNPROTECTED_VAR]: var field 'requests' on Sendable class 'Stats' must be of type Mutex[T], RwLock[T], or Atomic[T], not 'int'

Even an int field is refused: two threads doing requests = requests + 1 at once lose updates. Declare it let requests: Atomic[int] = Atomic.new(0), as Listing 20-3 does.

Known gap. When Thread.spawn is called inside a for loop over a collection (for (x in items) { … Thread.spawn(…) … }), 1.25.1 does not check the closure's captures, so a non-Sendable class slips through. Chapter 25 shows the case. Listings 20-3 and 20-4 are correct because everything they capture is Sendable; keep it that way yourself.

Note. A captured list or dict is Sendable, and the specification says the thread receives a copy, so changes made in the thread are not visible to the spawner. Under loup run the thread changes the spawner's list instead. Do not rely on either behaviour: return results through join(), an Atomic, or a Mutex.

Locks and atomics

Mutex[T]

A Mutex protects a value. withLock takes the lock, passes a guard to the closure, and releases the lock when the closure returns or throws. The guard's get() and set(v) read and write the protected value:

func main(): void {
    let total = Mutex.new(0);
    var workers: list[Thread] = [];
    for (chunk in [[1, 2, 3], [4, 5], [6, 7, 8, 9]]) {
        workers.push(Thread.spawn(func(): void {
            var partial = 0;
            for (x in chunk) {
                partial = partial + x;
            }
            total.withLock(func(g: MutexGuard[int]) => g.set(g.get() + partial));
        }));
    }
    for (w in workers) {
        w.join();
    }
    println(total.withLock(func(g: MutexGuard[int]) => g.get()));
}

Listing 20-4: Threads adding their partial sums under a mutex

45

Annotate the guard parameter (g: MutexGuard[int]); without the annotation the compiler cannot tell what g.get() refers to. tryWithLock runs the body only if the lock is free and returns an Option; isLocked() reports the state.

Mistake: a non-integer Mutex

The specification makes Mutex[T] generic over any type, but Stage0 1.25.1 binds the value as an int:

@mode(host.native)

func main(): void {
    let lastError: Mutex[string] = Mutex.new("");
}

main.lphc:4:46: error[E0001]: argument 1 of 'new': expected 'int', found 'string'

The same holds for RwLock[T] and Atomic[T]. To protect other kinds of state, keep it in a synchronized class (chapter 15), whose modifier methods run under the instance's lock.

Guards and their lifetime

lock() returns the guard directly, for code that cannot fit in a closure. The guard must be released with release(), and forgetting to do so is caught the next time the same thread asks for the lock:

@mode(host.native)

func main(): void {
    let balance = Mutex.new(100);
    let g = balance.lock();
    g.set(g.get() - 30);
    let again = balance.lock();
    println(again.get());
}

Uncaught LifetimeError: Mutex.lock: the lock is already held by THIS thread — re-entrant locking (§2-1.8.2), which deadlocks a real mutex, or a guard that was never released.
  at main (main.lphc:7:17)

A real mutex would simply deadlock here; Loup reports the mistake instead, in the interpreter and in native binaries alike. Call g.release() before locking again, or use withLock, which cannot forget.

RwLock[T]

An RwLock lets many readers in at once, or one writer:

func main(): void {
    let limit = RwLock.new(10);
    let r = limit.readLock();
    println(i"limit {$ r.get() $}, readers {$ limit.readerCount() $}");
    r.release();
    limit.withWrite(func(g: RwLockWriteGuard[int]) => g.set(20));
    println(limit.withRead(func(g: RwLockReadGuard[int]) => g.get()));
}

Listing 20-5: Reading and writing through an RwLock

limit 10, readers 1
20

A read guard has no set; the checker refuses one.

Atomic[T]

For a single counter or flag, an Atomic is cheaper than a lock. Every operation is sequentially consistent:

func main(): void {
    let hits = Atomic.new(0);
    var workers: list[Thread] = [];
    for (i in 0..4) {
        workers.push(Thread.spawn(func(): void {
            for (j in 0..250) {
                hits.fetchAdd(1);
            }
        }));
    }
    for (w in workers) {
        w.join();
    }
    println(hits.load());
    let (swapped, seen) = hits.compareExchange(1000, 0);
    println(i"swapped={$ swapped $} seen={$ seen $} now={$ hits.load() $}");
}

Listing 20-6: An atomic counter and a compare-and-swap

1000
swapped=true seen=1000 now=0

The read-modify-write operations (fetchAdd, fetchSub, fetchAnd, fetchOr, fetchXor, exchange) return the old value. compareExchange(expected, desired) returns whether it swapped and the value it found.

Thread pools

ThreadPool runs closures on a fixed set of workers. run submits a closure and gives back its result; shutdown() stops the pool:

func main(): void {
    let pool = ThreadPool.new(4);
    var squares: list[int] = [];
    for (n in [10, 20, 30]) {
        let r: int = pool.run(func(): int { return n * n; });
        squares.push(r);
    }
    pool.shutdown();
    println(i"{$ squares $} on a pool of {$ pool.size() $}");
}

Listing 20-7: Submitting work to a ThreadPool

[100, 400, 900] on a pool of 4

The specification types run as returning Task[T] for use with await; in 1.25.1 it returns the value directly. Two more rules are checked at run time: ThreadPool.new(0) throws RuntimeError: ThreadPool.new: size must be positive, and run after shutdown() throws LifetimeError: ThreadPool.run after shutdown() — the work would never be performed.

Known gap. In 1.25.1 ThreadPool.run executes its closure immediately on the calling thread, on every delivery path including native binaries; worker queues are not implemented yet. The results are correct, but a pool gives no parallelism. Use Thread.spawnWith for parallel work.

Borrows and threads

A thread can outlive the frame that started it, so the lifetime rules forbid two kinds of capture. A labelled borrow[~a] parameter may not be captured (E_BORROW_THREAD_CAPTURE, chapter 18), and neither may an excl borrow parameter:

@mode(host.native)

struct Frame {
    var pixels: int;
}

func render(excl borrow f: Frame): void {
    let t = Thread.spawn(func(): void { println(f.pixels); });
    t.join();
}

func main(): void {
    var f = Frame(pixels = 4);
    render(f);
}

main.lphc:8:13: error[E_EXCL_BORROW_THREAD_CAPTURE]: `excl borrow` binding 'f' cannot be captured by the closure passed to Thread.spawn — the exclusivity guarantee cannot be upheld across a thread boundary
  note: v1.20 §1-23.4.1.1: an `excl borrow` binding is exclusive to one frame; move the value into the closure instead.

Copy what the thread needs into a local first (let pixels = f.pixels;) and capture that. A unique value can be moved into a thread if its type is Sendable; the spawning code then loses it (chapter 17).

What is planned

The specification's concurrency chapter is larger than what 1.25.1 ships. These parts are ⚠ planned:

• the sync.* modules: channels, semaphores, barriers, condition variables and the shared collections (§2-2.8);
• ThreadLocal[T] and TaskLocal[T] (§2-1.9.7, §2-1.9.8): undefined identifier 'ThreadLocal';
• the multi-threaded async executor and its configuration, runtime.configure(RuntimeConfig(executor = ExecutorConfig(…))) (§2-1.9);
• actors, Actor[M] with a mailbox (§2-1.6); an actor class declaration, used in earlier editions, is not Loup syntax;
• for concurrent parallel loops, which native builds refuse (§2-1.11);
• shared class instances built with shared C(…) (§2-2.6): the checker enforces that every field is Shareable (E_SHARED_FIELD_NOT_SHAREABLE for a string field, for example), but loup run cannot construct one (internal interpreter limitation: unsupported instruction: NewSharedObject).

@concurrent is accepted only on async functions ('@concurrent' can only be applied to async functions), and native builds refuse async; see chapter 21.

Summary

• Thread.spawn runs a void body; Thread.spawnWith returns a handle whose join() gives the body's value or rethrows its exception. No import is needed; sync.thread does not exist yet.
• Native binaries run threads in parallel; loup run runs them one at a time.
• Captures must be Sendable. A class is Sendable only by declaration, with var fields of type Mutex, RwLock or Atomic (E_SENDABLE_UNPROTECTED_VAR), or as a synchronized class.
• Mutex, RwLock and Atomic hold int values in 1.25.1; withLock/withRead/withWrite release automatically, and a forgotten release() is a LifetimeError, not a deadlock.
• ThreadPool.run returns the closure's value but runs it on the caller's thread for now.
• Lifetime-labelled and excl borrow bindings cannot be captured by thread closures.
• Channels, thread-local storage, actors, the multi-threaded executor and for concurrent are planned.