Roll your own: atomics, the futex and locks

Note: The sync package, and so the locks, the channels and the actors built on it, runs on Windows and Linux (x86-64), not yet on macOS. Not built yet: the atomic permission, and the rule that a whole atomic value may be replaced plainly only through a unique path.

Under the actors, the channels, the lock permissions and the parallel loop lie three small pieces: atomic values, which several threads change with no lock; the futex, the one way a thread sleeps until another wakes it; and plain locks built from those two. This page is for the program that uses them directly, to write a scheduler, a channel or a data structure of its own, or to count something shared at the lowest cost. Most programs never need to. Afterwards you can use each correctly, and know which of them the compiler checks and which it leaves to you. Here four threads count together, one number with no lock and two with one:

mod hits;

import stdio use *;
import thread;
import sync use *;

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

let stats Stats;               // let: never changed, but for its atomics
let lock Mutex;                // a plain lock, for the pair below
let sum Atomic[u64];
let sumSquares Atomic[u64];

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

fn work(id u64) u64 {
  var i = 0u64;
  while i < 100000u64 {
    stats.hits.add(1u64);      // one indivisible add, no lock
    i += 1u64;
  }
  // Two numbers that must change together: a lock makes them one step
  lock.lock();
  sum.store(sum.load() + id);
  sumSquares.store(sumSquares.load() + id * id);
  lock.unlock();
  id;
}

fn main() {
  let a = thread.start(1u64, &work);
  let b = thread.start(2u64, &work);
  let c = thread.start(3u64, &work);
  let d = thread.start(4u64, &work);
  a.join(); b.join(); c.join(); d.join();
  print <- "hits: ", stats.hits.load(), "\n";
  print <- "sum ", sum.load(), ", sum of squares ", sumSquares.load(), "\n";
}

It prints, on every run:

hits: 400000
sum 10, sum of squares 30

Four hundred thousand additions from four threads at once, and not one lost. Were the count a plain number, two threads could read the same count and each write back one more, and one addition would be lost.

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; 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:

Error 1186: Atomic requires T is Integer or T is bool or T is Pointer, and it is false for T = f64.

These operations are the building blocks of every synchronized mechanism in this part, the channels and lock permissions among them. Underneath them are the raw atomic operations core declares in mem (Intrinsics), which act through a raw pointer and so need trust.

In the program above, stats is declared let, and yet its hits changes: a let variable can never be changed, 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.

The sum and the sum of squares above show the limit of atomics: each change is indivisible on its own, but two changes are two steps, and another thread may come between them. Values that must change together need a lock, or a lock permission.

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:

Error 1187: No method add for Atomic[bool]: it exists only where T is Integer, and bool is not Integer.

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

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

fn push(node *Node) {           // called from any number of threads at once
  var seen = head.load();
  var 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:

  Error 1184: Bad declares AtomicValue, so its field v must be an integer of 8 to 64 bits, a bool or a raw pointer,
  and f32 is none of them.
  

  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 (Moves): 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 let, 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 built 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 thread that waits for another must sleep, not spin, or it burns a core doing nothing. 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 is built on it: the locks below, the thread package's parking, the channels, and the actors package's idle workers. 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.

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

// The waker, on another thread
word.store(1u32);                  // the new value first
sync.futexWake(&word);             // then the wake

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. 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.

A global is shared by every thread, so its type must be Shareable (a rule not yet checked: What may cross a thread). An atomic word is, and a lock makes a read and a write of it one step, as lock does for the pair of sums above. Data too big for an atomic, such as a List, is kept behind a lock permission instead (below).

• 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:

panic at sync.cone:640: sync: Mutex.unlock of a Mutex that is not locked

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. So nothing reaches the value unlocked, and nothing is left locked by forgetting. Reach for a plain lock only where a lock permission cannot go. 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.

That is the bottom of Cone's concurrency: you can now count with atomics, sleep and wake on a futex, and guard data with a plain lock, and you know which of them the compiler checks. With actors, parallel loops, I/O and the GPU above them, you can choose the right tool at every level. Next comes Modularity: how to break a large program into parts that each keep their own counsel.