OS threads and channels

Note: The thread and sync packages, and so the channels and the actors built on them, run on Windows and Linux (x86-64), not yet on macOS.

Actors and parallel each run on a pool of worker threads that the runtime starts for you. Sometimes you want a thread of your own instead: to run a library that blocks, to give one job a thread to itself, or to build a scheduler, a server loop or a pipeline of your own design. This page and the next are the roll-your-own layer. This one covers starting and joining operating-system threads, and passing values between them through channels; the next covers atomic values, the futex and plain locks. Afterwards you can write a program that manages its own threads, safely. Here four threads each count the primes in a quarter of a million, and send their counts back over a channel:

mod quarters;

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

fn isPrime(n u64) bool {
  if n < 2u64 { return false; }
  var d = 2u64;
  while d * d <= n {
    if n % d == 0u64 { return false; }
    d += 1u64;
  }
  true;
}

// What a thread is handed: its stretch of numbers, and where to send the count
struct Job {
  pub from u64;
  pub to u64;
  pub out Sender[u64];
}

fn count(var j Job) void {
  var found = 0u64;
  var n = j.from;
  while n < j.to {
    if isPrime(n) { found += 1u64; }
    n += 1u64;
  }
  j.out <- found;                        // send the count back
}

fn main() {
  var tx Sender[u64];
  var rx Receiver[u64];
  tx, rx = Channel[u64].make();

  // Four OS threads, each counting the primes in a quarter of a million
  let a = thread.start(new Job(0u64, 250000u64, tx.clone()), &count);
  let b = thread.start(new Job(250000u64, 500000u64, tx.clone()), &count);
  let c = thread.start(new Job(500000u64, 750000u64, tx.clone()), &count);
  let d = thread.start(new Job(750000u64, 1000000u64, tx.clone()), &count);

  var total = 0u64;
  each i in 1 .. 4 {
    match rx.recv() {                    // waits for the next count
      case let n Some { total += n; }
      else {}
    }
  }
  print <- "There are ", total, " primes below a million\n";
  a.join(); b.join(); c.join(); d.join();
}

It prints the Guide's answer, found a different way:

There are 78498 primes below a million

Compare it with the Guide's parallel each, which did the same in one line. Here the program decided how to split the work, and the split is uneven: testing a number near a million takes longer than testing one near zero, so the last quarter's thread finishes well after the first's, which then sits idle. A parallel each splits as it runs, and evens that out. That is the trade: full control, and every choice yours.

OS threads: the thread package#

The thread package is the operating-system layer every other model is built on: starting and joining OS threads. It is deliberately small, and it is ordinary library code, bound to the operating system through extern functions; the compiler knows nothing of it. 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. Its source, packages/thread/src/thread.cone, documents the same rules at greater length.

import thread;

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

fn main() {
  let t = thread.start(100i64, &triangle);  // a new thread runs triangle(100)
  ...                                        // while main goes on
  let 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, as Job does above.
• 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 (Saying what is behind a reference).
• 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 (What may cross a thread has the whole rule). These cross:

• A value the thread owns outright: a number, a struct of them, a Uni owner, a uni reference, a channel's end.
• 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 an imm or opaq 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.

Sendable shows the refusal of a Job holding a &mut, with its message.

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. In an actor, sleep with aio's timers instead, which hold no worker.
• 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, Inside an actor and What the body may do). 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.

Channels#

Threads that share nothing still need to talk. A channel is a queue between threads: any number of threads put values in at one end, and one thread takes them out at the other, oldest first. The sync package's Channel[T] has any number of sending ends and one receiving end, and every value it carries must be Sendable.

An actor's mailbox is a queue of the same shape, many senders and one receiver, but it is the actors package's own, built into each actor; a Channel is for plain threads, and for anything else that wants a queue it can hold, such as the kitchen's "done" signal on the I/O page. Today they are separate mechanisms.

There are three kinds:

• Channel[T].make() answers both ends of an unbounded channel, whose sends never wait.
• Channel[T].bounded(n) answers both ends of a bounded one, holding at most n values (rounded up to a power of two), whose sends wait while it is full, so a sender faster than its receiver is held to its pace. The two kinds' ends are the same types, so code taking a Sender[T] takes either.
• Channel[T].spsc(n) makes a bounded channel with one sender. Its sending end is a SoloSender[T], which has no clone(), and whose sends take it as &mut: there is exactly one, and one thread sends at a time. Knowing that from the type, the channel needs no atomic read-modify-write on either side, which makes it the fastest of the three where one thread feeds another: a pipeline's stages, or a logging thread's own buffer. Its receiving end is the same Receiver[T], and its methods are the others'.

Each end is a move type: another sender is made only by clone(), as the four threads' were above, and the receiver has no clone, so exactly one thread takes from it.

import sync use *;

fn main() {
  var tx Sender[Job];
  var rx Receiver[Job];
  tx, rx = Channel[Job].make();   // the two ends; or Channel[Job].bounded(64)
  let tx2 = tx.clone();           // a second sender, said out loud
  tx <- job;                      // send: waits only while a bounded channel is full
  match rx.recv() {               // waits while it is empty
    case let j Some {run(j);}
    case is None[Job] {}          // closed, and every value taken
  }

  var solo SoloSender[Job];
  var rx2 Receiver[Job];
  solo, rx2 = Channel[Job].spsc(1024);  // one sender, no clone
  solo <- job;
}

• tx.send(v), or tx <- v, puts a value in, waiting while a bounded channel is full. It answers None, or, when the receiver has gone or closed the channel, Some holding the value, handed back rather than lost. A list after <- sends each entry in turn, tx <- a, 3 of next() (the send operator's entries), and drops what each send answers. tx.trySend(v) never waits, handing the value back from a full channel too; tx.sendTimeout(v, d) waits at most d, a Duration of the time package, for room.
• rx.recv() waits for the next value, oldest first; rx.tryRecv() never waits; rx.recvTimeout(d) waits at most d, a Duration. Each answers an Option.
• Nothing panics, and no one has to close the channel by hand. The last sender's death closes it: the receiver takes what is left, and then recv answers None. rx.close() refuses later sends while what came before still drains. The receiver's death closes it too, wakes every send waiting for room to hand its value back, and finalizes each value still in it, once.
• A send waiting on a full bounded channel waits for as long as it stays full: a receiver that stops taking without going leaves it waiting for good. Two threads that each send to the other through bounded channels can so wait on each other; where that may happen, trySend, sendTimeout or an unbounded channel is the answer.

A send is one compare-and-swap, into blocks of 32 slots allocated a block at a time (unbounded) or a ring of n slots allocated at the start (bounded); the receiver takes a value with no read-modify-write at all. A single-sender channel's send is a plain store as well, into Lamport's ring, each side keeping a copy of the other's index and reading the real one only when its copy says full or empty. A thread sleeps, on the futex, only when there is nothing to do: the receiver when the channel is empty, a sender when a bounded channel is full.

The one-shot channel#

Often a thread needs exactly one value back: the answer to a request, or the result of a call made on another thread. A Oneshot[T], from the sync package, carries one value, once, from one thread to one. Oneshot[T].make() answers its two ends as a tuple, as Channel[T].make() does: a OneshotSender[T] and a OneshotReceiver[T]. T must be Sendable. Each end is a move type with no clone(), and each crosses threads. The sender typically moves to the thread that will answer, and the requester waits on the receiver:

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

// The answering thread: takes the sending end, sends one reply
fn answer(tx OneshotSender[u64]) u64 {
  var t = tx;
  t.send(42u64);
  0u64;
}

fn main() {
  var tx OneshotSender[u64];
  var rx OneshotReceiver[u64];
  tx, rx = Oneshot[u64].make();
  let t = thread.start(tx, &answer);     // the sending end moves to the answering thread
  match rx.recv() {                      // waits for the one value
    case let v Some {printUInt(v); printStr("\n");}
    case is None[u64] {printStr("no reply\n");}
  }
  t.join();
}

This prints 42.

• tx.send(v) puts the value in, and never waits. It answers None once the value is there, and Some holding the value, handed back rather than lost, when a value was sent already or the receiver has gone. It takes the sender as &mut, so a sender held in a field sends in place.
• rx.recv() waits for the value, and answers None when the sender went without sending, or the value was taken already. rx.recvTimeout(d) waits at most d, a Duration of the time package (it may run over by a tick); rx.tryRecv() never waits.
• rx.isReady() says whether the value is there to be taken; rx.isClosed(), whether nothing more will come; tx.isClosed(), whether a send would hand its value back.
• A sender that dies without sending closes the channel, so a receiver finds out, asleep or not. A value sent and never taken is finalized exactly once, by whichever end goes last.

import stdio use *;
import sync use *;

fn main() {
  // A second send hands its value back, and a receive after the take finds nothing
  {
    var tx OneshotSender[u64];
    var rx OneshotReceiver[u64];
    tx, rx = Oneshot[u64].make();
    tx.send(5u64);
    match tx.send(6u64) {
      case let back Some {printStr("handed back "); printUInt(back); printStr("\n");}
      case is None[u64] {}
    }
    match rx.recv() {
      case let v Some {printStr("received "); printUInt(v); printStr("\n");}
      case is None[u64] {}
    }
    match rx.recv() {
      case let v Some {}
      case is None[u64] {printStr("nothing more\n");}
    }
  }
  // The sender going without sending: the receiver is told at once
  {
    var rx OneshotReceiver[u64];
    {
      var tx OneshotSender[u64];
      tx, rx = Oneshot[u64].make();
    }
    match rx.recv() {
      case let v Some {}
      case is None[u64] {printStr("the sender went\n");}
    }
  }
  // The receiver going: a send would hand its value back
  {
    var tx OneshotSender[u64];
    {
      var rx OneshotReceiver[u64];
      tx, rx = Oneshot[u64].make();
    }
    printStr(if tx.isClosed() {"the receiver went\n";} else {"?\n";});
  }
}

handed back 6
received 5
nothing more
the sender went
the receiver went

A send is one swap and a receive that finds the value is one load; the futex is used only when the receiver really sleeps. The iocore package's Target.Once hands an I/O operation's answer to a thread waiting on one (What the loop does).

You can now start, join and park your own threads, and pass values between them with channels. Under the channels, the parking and the runtime's idle workers lie three smaller pieces: atomic values, the futex, and the plain locks built from them, which are the last page of this part.