Concurrency

Note: Co-routines beyond generators (a two-way yield), green threads that are paused and resumed, processes working together on one job, and distributed computing are not implemented. The thread and sync packages, and so the locks, the channels and the actors built on them, run on Windows and Linux (x86-64), not yet on macOS. The atomic permission is not implemented, nor is the rule that a whole atomic value may be replaced plainly only through a unique path. actor trait is not implemented.

Concurrency and parallelism can sometimes improve program throughput, by:

• Distributing work across multiple CPUs (many hands make light work).
• Keeping the CPU(s) busy with other work, rather than pausing all work while waiting on i/o to complete.

Most of the heavy lifting for Cone's concurrency and parallelism capability is baked into library packages. Depending on the library packages chosen, many work distribution and scheduling approaches can be taken (ordered by context switching cost):

• Cooperative concurrency (generators or co-routines), where context switching is explicitly baked into the logic. Generators are the form that exists.
• Green (mxn) threads, where context-switching is cooperatively handled by the runtime's work-stealing scheduler, rather than the OS. The actors package (Actors) is the nearest thing: actors are multiplexed on a pool of worker threads by a work-stealing scheduler, and each message is handled to completion on a worker's own stack, or up to an await, where the compiler cuts the behaviour in two. An actor is not a green thread that is paused and resumed.
• OS threads, where context-switching is handled by the OS scheduler. Threads share memory and resources. The thread package provides them, below, with atomic values, the futex and locks to share memory safely, and channels to pass messages.
• OS processes, where context-switching is handled by the OS. Processes are isolated, largely communicating via pipes.
• Distributed computing, where work is distributed across different machines, communicating via network sockets.

As always there are trade-offs: Sometimes the loss in context-switching cost (latency) can be justified by improvements in scalability, fault tolerance, i/o handling, etc.

All the above models support structured concurrency. All new processes or threads are effectively owned resources. Like all owned resources, there is a programmatic point-in-time when owned threads are joined, terminated and freed according to the logic of the finalizer. Libraries also support the ability to split m pieces of work across n independent processes/logic. (A parallel each loop splits the pieces of a loop across the actors package's worker threads; splitting across processes is not implemented.)

Concurrency (and parallelism) introduce safety risks, when implemented poorly:

• Race conditions, where a system's behavior varies inconsistently, depending on the sequence or timing of independent operations.
• Deadlocks, where forward progress has stalled out because independent operations have unresolved contention over the temporary ownership of multiple resources.
• Livelocks, where forward progress has stalled out because of resource starvation that is not caused by deadlocks.

Subsequent pages highlight Cone language features (e.g., permissions and queued methods) that can be used to make synchronized, inter-thread communications easier and safer.

OS threads: the thread package#

The thread package is the operating-system layer every other model is to be built on: starting and joining OS threads. The one blocking primitive the locks, the channels and the schedulers are written over, the futex, is the sync package's, which the thread package imports. It is deliberately small, and it is ordinary library code, bound to the operating system through extern functions; the compiler knows nothing of it. Its source, packages/thread/src/thread.cone, documents the same rules at greater length.

import thread;

fn triangle(n i64) i64 { ... }

fn main() {
  imm t = thread.start(100i64, &triangle);  // a new thread runs triangle(100)
  ...                                        // while main goes on
  imm total = t.join();                      // wait for it: 5050
}

Starting a thread#

thread.start(arg, &f) starts a new OS thread running f(arg) and returns its handle at once, a Thread[R]. The thread is given a function and one value:

• The value is moved in. The thread owns it, and it dies there when f is done with it. A struct carries more than one value.
• f is a plain function taking an A and returning an R, which may be void. A function reference holds no state, which is why a thread is given a value beside its function; a closure travels as that value, in an owner marked + Sendable (Closure References).
• The type arguments are inferred from the value and from the function's signature. A number literal given as the value keeps its own type, i32 unless suffixed, and must agree with f's parameter: 100i64 for a function taking an i64.
• The value and the result must be Sendable, which the compiler checks where start is called (below).

thread.startWith(new thread.Options(name: n, stackSize: s), arg, &f) does the same with options, each of which has a default and may be left out:

• name, empty by default, is copied as the thread starts. It is Cone's own, seen through name(); the operating system's debuggers do not see it.
• stackSize, 0 by default, is how many bytes of address space to reserve for the thread's stack; 0 is the system's default, the executable's on Windows and the C library's on Linux (the process's stack limit, usually 8 MB). A size is rounded up to 64 KB at least, and Windows rounds it up to its allocation granularity and takes one of 4 GB or more as 4 GB less one byte.

A thread that cannot be started, or memory that cannot be allocated for it, is a panic. A panic or a trap on any thread ends the whole process: nothing unwinds, so no thread fails alone while the others go on.

What a thread may be handed#

The value handed to a thread, and the value it returns, must be Sendable: a value the thread may own, or share with the others, without a race. The compiler checks it where thread.start is called, and the same holds for a value sent over a channel or to an actor (Race condition safety). These cross:

• A value the thread owns outright: a number, a struct of them, a So owner, a uni reference.
• An imm or opaq reference, which the permission rules already mark as safe to share, where its region's count, if it keeps one, is atomic: an Arc[imm, T] owner (the sync package's Arc, folded in by import sync use Arc), which thread.start copies, so the value is freed once, after the last thread lets it go (regions); and a borrow of a global, thread.start(&imm table, &sum), since a global lives for the whole program.
• A struct that carries raw pointers the program shares safely, with atomic values or locks, and declares is Sendable. That is a promise the compiler takes on trust.

These are refused, naming what is at fault (Error 1194):

• A mut or ro borrow, and any borrow into the starting thread's stack (an imm borrow of a local is refused as not global, Error 1249).
• An Rc owner that is not uni, since its count is not atomic.
• A reference the collector package's Gc traces, since that collector is single threaded (below).
• A raw pointer, since nothing checks what it points at, unless a struct holding it declares is Sendable.

struct Counter { pub n i64; }
struct Job { pub data &mut Counter; }

fn work(j Job) i64 { j.data.n; }

fn main() {
  mut c = new Counter(1i64);
  imm t = thread.start(new Job(&mut c), &work);
  // Error 1194: start requires A is Sendable, and Job is not Sendable:
  // Job.data is &mut Counter, a borrowed reference of permission mut, ...
}

The handle: join, detach, and join on drop#

A Thread[R] is its thread's one owner, a move type.

• t.join() waits for the thread to finish and returns what f returned. It consumes the handle, so a thread is joined once.
• t.detach() consumes the handle and lets the thread run on alone. What f returns is finalized on the thread, since nobody will read it.
• A handle that dies unjoined joins: its finalizer waits for the thread and finalizes the result nobody took. This is tidiness, not safety. A handle can be kept where it never dies, and a process ends without finalizing what its threads hold, so nothing a program depends on may rest on that join. A thread whose work must be finished is joined.
• t.id(), t.name() and t.unpark() are the thread's own, as below, and t.threadRef() is a ThreadRef to it.

A handle kept in a field or a collection cannot yet be moved out of it to be joined (Cone does not move a value out of a struct's field), so such a handle is joined by its container's death. When main returns, the process ends, and every thread still running ends with it; handles in main's locals have been joined by then.

The running thread#

• thread.current() is a ThreadRef to the calling thread: its identity, kept alive while the ThreadRef lives, which may be sent to other threads. It is a move type; r.copy() makes another. r.id() is the operating system's thread id, which it may reuse once a thread has ended and its handle is gone. r.name() is the name startWith gave, and empty for any other thread, the main thread included. A thread the package did not start takes up an identity the first time it asks for one, and keeps it, since nothing runs when such a thread ends.
• thread.parallelism() is how many logical processors the machine has (on Linux, the ones online), at least 1. It does not count what an affinity mask takes away.
• thread.sleep(d) sleeps at least d, a Duration of the time package (thread.sleep(time.Duration.fromMillis(5))), on Windows' high-resolution timer, so a 1 ms sleep takes 1 to 2 ms rather than the 15.6 ms of the system's timer tick. Where that timer is missing (before Windows 10 version 1803) it falls back to whole milliseconds, rounded up. On Linux it is POSIX's nanosleep, accurate to a few tens of microseconds. A sleep of zero, or of a negative duration, gives up the rest of the time slice to another ready thread of the same priority.
• thread.yieldNow() lets another thread ready to run on the same processor have it, and returns at once if there is none.

Parking#

Each thread has one parking token, and starts without it. thread.park() takes the token, sleeping until there is one; unpark(), on the thread's handle or a ThreadRef to it, makes it available. An unpark sent before the park is therefore not lost, and the park returns at once. Unparks do not add up: two before a park are one token. park never returns without a token, but a parked thread still checks what it was waiting for, since a token may have been meant for an earlier park. Parking is built on the futex.

What a started thread may not use yet#

The collector package's Gc collector is single threaded: its roots are one chain and its heap is not synchronized. A traced reference made, held, reached or dropped on a thread the package started corrupts it. One is not Sendable, so it cannot be handed to a thread or returned from one, but nothing stops a started thread from making one of its own (an actor and a parallel each are refused one for now, Actors and Each). A thread-local global gives each thread a copy starting from its initial value, and nothing finalizes a thread's copies when it ends.

Deliberately not provided: killing a thread, priorities, pinning a thread to a processor, a flag asking a thread to stop, and scoped threads, which would join before a function returns so that their threads could borrow. The package runs on Windows and on Linux.

Atomic values#

An atomic value is a number, a flag or a pointer that several threads may read and change at once, with no lock: each change is indivisible, so no thread ever sees one half done, and when several threads change it together, each change happens whole, one after another. The core package provides it as Atomic[T], a name of every module with no import (it needs nothing of the operating system, so it is core's, as Rust keeps its atomics in core; the sync package holds what does), for a T that is an integer type of 8 to 64 bits (usize and isize among them), bool, or a raw pointer, *T. Its declaration says so: struct Atomic[T] ... where T is Integer or T is bool or T is Pointer (Generic constraints), and any other T is refused where it is written.

struct Stats {
  pub name u32;
  pub hits Atomic[u64];        // atomic because of its type
}

imm stats Stats;

fn @initpure init() {
  stats = new Stats(7u32, Atomic[u64].of(0u64));
}

fn count() {                   // called from any number of threads at once
  stats.hits.add(1u64);        // one indivisible add
  imm n = stats.hits.load();   // the number itself
}

stats is imm, and its hits changes: imm means that nothing changes, except inside the atomic values it holds. Its other fields read as they always did. An atomic value changes only through its own operations:

• Atomic[T].of(v) makes one holding v.
• load() answers the value, and store(v) puts v there.
• swap(v), add(v), sub(v), min(v), max(v), bitAnd(v), bitOr(v) and bitXor(v) change it and answer the value it held before. Add and sub wrap; min and max keep the lesser or the greater, compared signed for a signed T. (and and or are keywords, hence the bit.)
• compareSwap(expected, desired) stores desired only if the value is expected, and answers the value it saw and whether it stored: seen, ok = a.compareSwap(e, d);. A loop around it makes any change indivisible: read, work out the new value, and try again if another thread changed it in between.
• inner() takes the value out of an atomic value held by value, which nobody else can be reaching.

Which of these an atomic value has depends on its T, each method saying with its own where clause when it exists: add, sub, min and max only for an integer, the bit operations for an integer or a bool, and the rest for all three. So an Atomic[bool] is a flag, set by store or bitOr and tested by load, and an Atomic[*Node] is the head of a lock-free list, moved on with compareSwap; flag.add(true) is refused where it is written, since add exists only where T is Integer.

struct Node {
  pub next *Node;
  pub val u64;
}

imm head Atomic[*Node];         // set by init: Atomic[*Node].of(null)

fn push(node *Node) {           // called from any number of threads at once
  mut seen = head.load();
  mut ok = false;
  while !ok {
    mem.writeRaw(node, new Node(seen, (node as &Node).val));   // link to what was seen
    seen, ok = head.compareSwap(seen, node);                  // swing the head, or see anew
  }
}

Each of them is sequentially consistent, the strongest ordering: every thread sees all such operations in one order, and nothing the thread does before or after one moves across it. That is right for nearly every use. For an expert, each has an Ordered form taking a MemOrder after its other arguments: a.addOrdered(1u64, MemOrder.Relaxed), a.loadOrdered(MemOrder.Acquire), a.compareSwapOrdered(e, d, success, failure). An ordering the operation cannot have (a load's Release, for one; the rules are the intrinsics') is known only as the program runs, so it ends the program with a message. Core's mem.atomicAdd and the rest, which take the ordering as a constant written at the call, refuse it where it is written instead.

What the compiler knows: AtomicValue#

Atomic[T] is ordinary library code. What the compiler knows is the built-in marker trait it declares, AtomicValue, and a type may declare it for itself: struct Flag is AtomicValue { pub v u32; }. The compiler holds such a type to four rules:

• It is one number: a struct of exactly one field, of a type some atomic operation acts on — an integer of 8 to 64 bits, a bool or a raw pointer. A float, a struct, a reference or anything else is refused, as are a trait, an enum or a variant declaring it. Atomic[T]'s own where clause admits exactly these, so Atomic[f64] is refused where it is written, once, as a requirement it does not meet.
• It moves, and is never copied (Move): a copy would be a second value that no atomic operation on the first could reach. Declaring Copy beside it is refused.
• A value holding one is never read-only. A global holding an atomic value anywhere inside it — in a field, a field's field, an array — is placed in writable memory even when it is imm, and its reads are never assumed unchanged.
• A const may not hold one, however deep: each use of a const is a fresh copy, which no other use would see change. Use a global.

Not implemented yet: the atomic permission, the reference that allows only atomic operations on what it points at, which every reference to an atomic value is to become; and the rule that a whole atomic value may be replaced plainly only through a unique path. Until they are, Atomic[T]'s operations take any reference (as &opaq), and a plain assignment over an atomic value — *r = s through a &mut Stats — compiles, and is the program's to get right.

The futex#

A futex is a 32-bit atomic word that threads sleep on while it holds a value, and are woken from. It is the one way a thread waits for another, and everything that blocks — the locks below, the thread package's parking, the channels, and the actors package's idle workers — is built on it. It is the sync package's, and its word is an Atomic[u32], taken by any reference, as that type's own operations are.

• sync.futexWait(&word, expected): if the word still holds expected, checked atomically with going to sleep, sleep until woken. It returns when woken, when the word already differed, or spuriously, which the operating system permits. So a waiter loops, re-reading the word.
• sync.futexWaitTimeout(&word, expected, d) is the same, sleeping at most d, a Duration of the time package. It returns false only when the timeout passed. The timeout on Windows is rounded up to whole milliseconds, and its resolution is the system's timer tick, up to 15.6 ms; a timeout of 2³² - 1 ms or more is that long, not forever. On Linux it is to the nanosecond, as the kernel's timers go. A timeout of zero, or a negative one, returns false at once if the word still holds expected.
• sync.futexWake(&word) wakes one waiting thread, and sync.futexWakeAll(&word) every one. A wake with no thread waiting does nothing and is not remembered, so a thread stores the new value first and wakes after.
• The caller keeps the word alive while any thread may wait on it or wake it: nothing checks that. Waits and wakes reach only the threads of one process.

while word.load() == 0u32 {
  sync.futexWait(&word, 0u32);     // sleeps only while the word is still 0
}

Locks: Mutex, Rwlock, Condvar and Once#

The sync package's locks are built on the futex alone, not on the operating system's own locks, as Rust's standard library has built its own since 2022. Each is one Atomic[u32] (an Rwlock, two), holds no operating-system handle, needs no finalizer, and may be moved while nobody holds it or waits on it. Each is made by make(), and each operation takes it by any reference, imm included, since a lock is shared by every thread that uses it.

import sync use *;

imm lock Mutex;             // both set by the module's init:
mut names List[u64];        //   Mutex.make(), List[u64].empty()

fn add(n u64) {             // called from any number of threads
  lock.lock();
  names.push(n);            // touched only while lock is held
  lock.unlock();
}

• Mutex: one thread at a time. m.lock() takes it, sleeping while another thread holds it (after a short spin); m.unlock() gives it up, waking a waiting thread; m.tryLock() takes it only if nobody holds it, and says whether it did.
• Rwlock: any number of readers at once, or one writer alone: readLock(), readUnlock(), tryReadLock(), writeLock(), writeUnlock() and tryWriteLock(). While a writer waits, new readers wait too, so readers cannot starve it.
• Condvar: a thread sleeps, holding no lock, until another says that what it waits for may have changed. c.condWait(&m) unlocks the Mutex the thread holds, sleeps until notified, and locks it again; c.condWaitTimeout(&m, d) sleeps at most d, a time.Duration, and returns false only if that passed; c.notifyOne() and c.notifyAll() wake one waiting thread or all of them. A wait may return spuriously, so a waiter loops, rechecking what it waits for under the Mutex. (wait is a reserved word.)
• Once: a function run exactly once, however many threads ask at once. o.callOnce(f) runs f() if no call on this Once has run one; every other caller sleeps until it has returned, and then sees all it did. o.callOnceWith(arg, f) runs f(arg) likewise, arg moved in, and dropped unused by a call that runs nothing. o.isCompleted() says whether the function has run. What is run is any callable of that signature: a function reference (o.callOnce(&f)), a closure written at the call (o.callOnce(fn() { ready = true; }), which may change what it borrows), or a struct with a () (&mut s). The parameter is a &<mut fn() (Closure References), compiled once and called through the reference, which a start-up that runs once can afford.

These locks, held as values, are unchecked. A lock is a plain lock with explicit lock() and unlock(); it guards nothing the compiler can see, and protecting the data is the program's job. Nothing stops a thread reaching the data without the lock, keeping a reference to it past the unlock, forgetting to unlock, or unlocking a lock another thread holds. A lock is not reentrant: a thread that locks a Mutex it holds sleeps forever. Unlocking a lock that nothing holds is caught, and ends the program with a message. A thread that ends while holding a lock leaves it locked; there is no poisoning, since nothing unwinds.

The checked way is a lock permission. Mutex and Rwlock are also lock permissions: written in the permission slot of an Arc, Arc[Mutex, T], the lock sits in each allocation's header, the reference reaches its value only by a borrow through it, the borrow takes the lock, and the borrow's last use gives it back. A Condvar waits on a Mutex held as a value only: a lock permission's lock is in a header no program names, so no wait is offered while a borrow holds one.