Thread Communications
On this page
Note: Not built yet: concurrent data structures other than channels and atomic values. Not yet checked: a mut global reached from several threads, and a value's own methods that change it through a &. The sync package runs on Windows and Linux (x86-64), not yet on macOS.
It can be a huge performance benefit that threads share memory and other resources, particularly when the threads need to exchange or share large, complex data structures. However, thread-shared memory also opens up safety risks when communications are not properly synchronized.
To mediate these risks, Cone offers several mechanisms that facilitate fast, convenient, safe communications between threads.
Race condition safety#
Any data passed or shared between threads must be done so safely. Cone ensures this by only allowing "sendable" data to be sent to another thread. The only type of data that is not sendable is a reference that has the "mut" or "ro" permission, or any aggregate type containing such a reference. Data of any other type is sendable, including shared, immutable references, unique references (which are moved from one thread to another), and any other move or copy type.
These restrictions apply to all mechanisms that support communications between threads, such as thread creation, concurrent data structures, and more.
What the compiler checks today. Library code marks what crosses threads with the built-in marker trait Sendable, as a constraint on a generic: thread.start and thread.startWith ask it of the argument and of what the thread returns, and Channel[T], Sender[T], SoloSender[T] and Receiver[T] of T. The compiler grants Sendable to a type unless it holds, directly or in a field, an array element, a tuple element or an enum's variant, one of these:
- a borrowed reference, of any permission, &imm included: a borrow's lifetime is checked in one thread alone, until scoped threads join inside it (but for a borrow that lasts the whole program, below);
- an owner that may be copied whose permission is not race-safe: Rc[mut, T], Arc[mut, T], and a lock permission for one thread, Rc[Rwcell, T]; a lock permission for threads crosses, Arc[Mutex, T] and Arc[Rwlock, T], where T may (Lock Permissions);
- an owner that may be copied in a region that does not declare ThreadSafe, whatever its permission: Rc[imm, T], whose count is not atomic. Arc[imm, T] and Arc[opaq, T] cross;
- a traced reference, Gc or an actor's ActorGc, whatever its permission: the first's collector is single threaded, and the second stays in the actor whose collector moves it (The actor's own garbage-collected region);
- a raw pointer, whose target nothing checks.
An owner that moves takes its value with it: a uni owner of any region, and any owner of So, crosses when what it points at may. So does an Arc[imm, T] owner: an Arc of a value holding a mut reference is refused. A reference to a function crosses. A type holding raw pointers it shares safely declares is Sendable, which the compiler takes on trust, as a channel's ends, ThreadRef, and the collections package's List and Dict do; an instance of a generic type declaring it is Sendable only where its type arguments are.
What is checked is the type of the value, not how a variable was declared: a local declared mut x = 5 holds a number, which is sendable, while &mut x is a borrow, which is not. A refusal names the part of the type that binds it to its thread:
struct Job {pub data &mut Counter;}
thread.start(job, &work);
// Error 1194: start requires A is Sendable, and Job is not Sendable: Job.data is &mut Counter,
// a borrowed reference of permission mut, ...
- an owner that may be copied in a region that does not declare ThreadSafe (an Rc): two threads copying it write its count at once without an atomic operation;
- a traced reference, Gc, whose copies are roots in a collector that is single threaded (a read-only ActorGc[imm, T] is the exception, in the passes of a parallel each in its own actor's behaviour: In an actor's method);
- a reference that writes through a path others share: a &mut (Cone's is shared mutable, so what holds one writes through a & of it), or an Arc[mut, T]. A closure that names a &mut variable is not Shareable, whatever its () takes, nor is a struct with a &mut field;
- a reference to a trait or a callable (&<Shape, &<fn(i32) i32, So[Shape]) that does not say what is behind it: the implementers are not known, so nothing says a borrow of the value may be shared (Rust's &dyn Trait is not Sync either). + Shareable, or + Sendable, below, vouches.
Raw pointers of a struct that is not generic are trusted, as the parallel-body check has always trusted them (nothing is followed behind one), a type holding ones it shares safely declares is Shareable, and an instance of a generic type is Shareable only where its type arguments are. A plain borrow, an Arc[imm, T] and an Arc[Mutex, T] of a Shareable T, an atomic and a number are. A parallel each body may name a value from outside the loop only if it is Shareable (Parallel each); a generic asks it of its argument as it asks Sendable, [F fn() + Shareable].
A reference to a trait or a callable hides the type of the value behind it, so it may say what that value is with +, after the type: So[fn(x f32) f32 + Sendable], &<Shape + Shareable, or both, in either order. A value is made such a reference only if its type is each marker, and the error names what is not. A parallel each body may name a reference to a trait or a callable from outside the loop only if it carries one of the two markers:
imm flock = new So[fn(x f32, z f32) f32 + Sendable](fn(x f32, z f32) [hf] f32 { ... });
imm t = thread.start(flock, &run); // Ok: hf is an Arc[imm, Grid], listed
// With hf an Rc[imm, Grid]: `So[uni, fn(f32, f32) f32 + Sendable]` says the closure behind it
// is Sendable, and this closure is not: it holds 'hf' (Rc[imm, Grid]), ...
A reference with markers is a reference with fewer, never the other way: a So[fn() + Sendable] goes where a So[fn()] is wanted. Only these two markers follow a + in a reference's type today; the intersection of other traits is not built.
A borrow that lasts the whole program ('static, Lifetimes) is checked in no thread's scope, so it may cross threads and actors in either of two cases, and in no other:
- it is imm or opaq, and what it reaches is safe for two threads to read at once, as the value of an Arc[imm, T] must be: no Rc and no Gc in it, no raw pointer, no borrow that is itself refused;
- it is uni, and what it reaches is safe to move. A uni borrow moves, so one holder only has it.
An ro borrow never crosses, even of a global: ro only stops this holder from writing, and another holder (mut) may still change what it reads (Permissions). Nor does a mut one. Nor does a borrow of an invariant lifetime (&'=a T, a key): the lifetime is an arena's brand, and the arena dies when its owner drops it. A uni borrow of a named global is refused where it is made ("cannot obtain this permission"): any code reaches a global by name, so it is not unique. A uni borrow of an owner that nothing else can reach would be unique, and crosses; an owner leaked for the rest of the program, which would make one, is not built yet, but a signature may already take one: fn f(p &'static uni T).
A generic says that a type parameter may be such a borrow by bounding it, [A Sendable + 'static] as thread.start does, and the call is then checked to hand it only borrows of the whole program (Lifetime bounds); without the bound, Sendable refuses every borrow as before. An actor's initializer or behaviour takes one by writing it 'static, and what it reaches is asked the same way. So is a behaviour's reply: it may be &'static imm T, &'static opaq T or &'static uni T, or a struct holding one, under the same rule.
imm table Array[u64, 3] = [1u64, 2u64, 3u64];
fn sum(t &imm Array[u64, 3]) u64 { (*t)[0] + (*t)[1] + (*t)[2]; }
fn peek(t &Array[u64, 3]) u64 { (*t)[0]; }
imm t = thread.start(&imm table, &sum); // Ok: an imm borrow of a global
actor Reader {
total u64;
pub async do addTable(self, t &'static imm Array[u64, 3]) { total = total + (*t)[0]; }
}
thread.start(&table, &peek); // Error 1194: &ro Array is not Sendable: it is a borrowed
// reference of permission ro ...
thread.start(&imm onTheStack, &sum); // Error 1249: must be global
Not checked yet: a mut global reached from several threads; a value's own methods that change it through &. What the compiler grants Sendable it also grants Shareable, so an Arc[imm, T] of one may be read by several threads at once; a type declaring Sendable over raw pointers promises that of them, and declares Shareable as well where its borrows may be shared (above). Actor messages are checked as a channel's values are: every parameter of an actor's behaviour, and of its initializer, must be Sendable, and one that is not is refused where it is declared; an actor's state is not asked, since it never crosses as a value (Actors).
Message-based Communications#
An efficient model for thread-based concurrency is independent actors (threads) that throw work to each other in the form of messages placed on an actor's queue. Each actor pulls off messages from its queue in order, and then performs the work requested by the message. For now, let's assume that actors do not need shared, mutable access to objects nor do they perform any resource-blocking operations.
A straightforward way to implement this actor model involves using library-packaged actors that own and use concurrent message queues (or channels). Such multiple-producer-single-consumer (MPSC) channels are themselves library packaged-delivered concurrent FIFO data structures. Any actor that wants some actor to perform work adds that work to the actor's queue in the form of a message. As part of a continuous loop, the actor owning the queue pops a message off the queue and completely performs the non-blocking work it requests.
Actors, which own such a queue, have a page of their own, Actors, and so has what an actor does while it waits for another's answer, Awaiting a reply.
Channels#
The sync package's Channel[T] is such a queue, usable directly between threads, with any number of sending ends and one receiving end. Channel[T].make() answers both ends of an unbounded channel, whose sends never wait; Channel[T].bounded(n), 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. Each end is a move type: another sender is made only by clone(), and the receiver has no clone, so exactly one thread takes from it.
Channel[T].spsc(n) makes the third kind, bounded as the second is, 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'.
import sync use *;
mut tx Sender[Job];
mut rx Receiver[Job];
tx, rx = Channel[Job].make(); // the two ends; or Channel[Job].bounded(64)
imm 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 imm j Some {run(j);}
case is None[Job] {} // closed, and every value taken
}
mut solo SoloSender[Job];
mut 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 append operator), 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 reply channel#
A Oneshot[T], from the sync package, carries one value, once, from one thread to one: the reply to a request, or the answer to a synchronous call. 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 {
mut t = tx;
t.send(42u64);
0u64;
}
fn main() {
mut tx OneshotSender[u64];
mut rx OneshotReceiver[u64];
tx, rx = Oneshot[u64].make();
imm t = thread.start(tx, &answer); // the sending end moves to the answering thread
match rx.recv() { // waits for the one value
case imm 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
{
mut tx OneshotSender[u64];
mut rx OneshotReceiver[u64];
tx, rx = Oneshot[u64].make();
tx.send(5u64);
match tx.send(6u64) {
case imm back Some {printStr("handed back "); printUInt(back); printStr("\n");}
case is None[u64] {}
}
match rx.recv() {
case imm v Some {printStr("received "); printUInt(v); printStr("\n");}
case is None[u64] {}
}
match rx.recv() {
case imm v Some {}
case is None[u64] {printStr("nothing more\n");}
}
}
// The sender going without sending: the receiver is told at once
{
mut rx OneshotReceiver[u64];
{
mut tx OneshotSender[u64];
tx, rx = Oneshot[u64].make();
}
match rx.recv() {
case imm v Some {}
case is None[u64] {printStr("the sender went\n");}
}
}
// The receiver going: a send would hand its value back
{
mut tx OneshotSender[u64];
{
mut 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 (I/O and Concurrency).
Sharing immutable data: Arc#
Immutable data is shared between threads through the sync package's Arc region (import sync use Arc), whose count is atomic: Arc[imm, T] owners may be copied and dropped on any thread, and the value is freed once, after the last (Region strategies). An Rc owner's count is not atomic, and it must stay on one thread.
Shared, Mutable Objects#
In most cases, the above approach is the most efficient when threads do not require shared access to mutable objects. This is because using thread-shared, mutable objects carries at least these performance penalties: the run-time cost of the atomic synchronization primitives (especially locks), and the way mutation invalidates the memory cache for other threads.
For this reason (and to minimize the risk of deadlocks), it is desirable to send only small-sized messages between threads (typically references) and only use the static permissions on references. Static permissions offer the best performance, because they guarantee at compile-time that shared, mutable access to the same memory location cannot happen between threads. No run-time mechanisms are needed to enforce these guarantees. Although mutable objects cannot be shared between threads, immutable can be. Even better, it is possible to move a uniquely-owned object from one thread to another.
That said, there are definitely situations where we might want threads to share access to mutable objects. This can be supported using lock permissions, shared data structures or atomic primitives.
Lock Permissions#
Cone's locked permissions have already been introduced. Lock permissions use runtime locks to synchronize when multiple threads need (mutable) access to the same object. The lock permissions Mutex and Rwlock are built, on Arc.
The locks provide an atomic envelope around a set of operations, so that no other thread can access the object while another has it locked. If threads request multiple locks at the same time, careful design is needed to avoid the risk of deadlocks.
Concurrent data structures#
The message queue used by actor is one of several kinds of concurrent data structures. Other examples include concurrent arrays and dictionaries.
Concurrent data structures are implemented in a way that makes it always possible to access and mutate these structures across multiple threads. Every operation (method) on these types is essentially atomic. Importantly, these operations never use locks and never block.
Atomic primitives#
Atomic primitives are a small collection of atomic (mutable) operations that are safe to perform on objects shared between threads. For example, it is safe to increment or decrement some shared integer. They are core's Atomic[T].
These atomic operations are the building blocks that make possible many of the synchronized communication mechanisms already described. Underneath them are the raw atomic operations core declares in mem (Intrinsics), which act through a raw pointer and so need trust.