Concurrency
The unit of execution is a lightweight task. Suspending functions (time, net, io, …) give up the thread at suspension
points and the OS thread pool spreads tasks over the cores. There is no async and no function "colour": any function can be
called like an ordinary one and suspends when it needs to (§8.1).
Structured concurrency: scope and spawn#
spawn is allowed only inside scope; leaving a scope waits for all children, and the error of one cancels the others (§8.5).
fn compute(n: i32) -> i32 time {
sleep(10ms)
n * n
}
fn main() io, time {
let (a, b) = scope {
let ha = spawn compute(3)
let hb = spawn compute(4)
(ha.join()!, hb.join()!)
}
print("{a} {b}")
}9 16
all, race, any#
Task.of { … } defers a computation; all runs the tasks in parallel, keeps the order of the results and cancels the rest on
the first error (§8.3).
fn job(tag: str, d: Duration) -> str io, time {
sleep(d)
print("done {tag}")
tag
}
fn main() io, time {
let rs = all([Task.of { job("x", 100ms) }, Task.of { job("y", 5ms) }])
for r in rs { print(r) }
}done y done x x y
Chains and deadlines#
-> joins steps, a tuple fans out in parallel, !> catches an error, within limits the whole chain in time (§8.4).
const PROFILE_DEADLINE = 300ms
type UserId = distinct i64
@error data NotFound { id: i64 }
data User { id: UserId; name: str }
data Order { total: i64 }
data Page { orders: List<Order>; recs: List<str> }
impl Page { static fn empty() -> Page => Page{ orders: [], recs: [] } }
fn getUser(id: UserId) -> User | NotFound | Timeout time {
sleep(5ms)
if id.raw == 0 { fail NotFound{ id: 0 } }
User{ id, name: "ann" }
}
fn recentFor(id: UserId) -> List<Order> | Timeout time {
sleep(20ms)
[Order{ total: 100 }, Order{ total: 250 }]
}
fn forUser(id: UserId) -> List<str> | Timeout time {
sleep(if id.raw == 2 { 2s } else { 10ms })
["book", "pen"]
}
fn profilePage(id: UserId) -> Page | Timeout time {
await getUser(id)
-> (recentFor(it.id), forUser(it.id))
-> (orders, recs) => Page{ orders, recs }
!> NotFound => Page.empty()
within PROFILE_DEADLINE
}
fn main() io, time {
match profilePage(UserId(1)) {
Page p => print("orders {p.orders.len()} recs {p.recs.len()}")
Timeout => print("timeout")
}
match profilePage(UserId(2)) {
Page p => print("orders {p.orders.len()} recs {p.recs.len()}")
Timeout => print("timeout")
}
match profilePage(UserId(0)) {
Page p => print("empty page: {p.orders.len()}")
Timeout => print("timeout")
}
}orders 2 recs 2 timeout empty page: 0
Channels and select#
chan<T>(cap: n) creates a Sender/Receiver pair. select waits for the first ready branch, after d is the timeout branch
(§8.7).
fn main() io, time {
let (tx, rx) = chan<i32>(cap: 1)
let r = select {
recv i32 v from rx => "got {v}"
after 50ms => "timeout"
}
print(r)
_ = tx.send(1)
print("{rx.recv().ok()}")
}timeout 1
STM#
Shared mutable state without locks: TVar and atomic { … }, a serializable transaction that restarts on conflict;
check(cond) waits until the variables it read change (§8.9). The body of atomic is pure apart from TVar operations.
use std.stm { TVar, check }
static A: TVar<i32> = TVar(100)
static B: TVar<i32> = TVar(0)
fn transfer(amt: i32) {
atomic {
let a = A.get()
check(a >= amt)
A.set(a - amt)
B.update(x => x + amt)
}
}
fn main() io {
transfer(30)
let a = atomic { A.get() }
let b = atomic { B.get() }
print("{a} {b}")
}70 30