Threads and channels
kola treats concurrency the way it treats resources and effects: a rule you would otherwise have to remember becomes one the compiler checks. Here the rule is about shared mutable memory, the thing a lock protects. kola removes it, so a data race is not a bug you can write. The tools are threads and channels.
The problem
A data race is two threads touching the same memory with no ordering between them, at least one of them writing. The result can be anything from nothing visible to a corrupted pointer. Most languages handle it with discipline: guard the data with a mutex (a lock that only one thread can hold at a time), take the lock, release it, remember which lock guards what, and acquire them in a consistent order. Nothing checks that you remembered.
kola has a particular reason to care. Inside one region, two references may alias and both write; there is no Rust-style uniqueness rule for &r mut. That is safe because a reference always points at a live value of the right type. Add threads naively and it stops being safe: two threads holding the same &r mut T is a data race. So kola needs a rule that keeps each region reachable from one thread only.
Don't share, move
The rule (SPEC-CARD §11): references never cross a thread boundary; ownership moves, through spawn contexts and channels. A thread that needs data from another does not borrow it. It receives it, and the sender no longer has it. This is message passing, and it is checked by regions (a reference is typed by a region that one thread opened) and by linearity (whatever crosses is consumed on the sending side).
As a result there is no Mutex, no lock and no memory-ordering surface in ordinary code. (std does have atomic operations, but they belong to hardware and bare-metal work under the hw effect, not to coordinating kola threads.) There is also no async/await and no futures. Protocol logic that juggles many pending operations on one thread is written as an explicit state machine, step(event, now) -> intents, with no hidden suspension points.
spawn, Thread[R] and join
sys.spawn(entry, ctx) starts a thread that runs entry(ctx) and returns a Thread[R], where R is what entry returns. entry is an ordinary top-level function, never a closure (kola has none), so its effect row is visible at the call site. ctx must be reference-free: plain data is copied, and resources and owned Vecs are moved.
use std::log;
fn square(n: u64) -> u64 !{} {
n *% n
}
fn main(mut sys: Sys) -> u8 !{io, block, alloc(heap), log} {
let t = sys.spawn(square, 7u64);
let result = t.join();
log::info(f"7 squared is {result}");
0
}Thread[R] is linear like a file: it must be joined on every path, early returns included. Forgetting one gets the same diagnostic as a leaked file, before the program runs. A spawned thread gets an 8 MiB stack on every host, the same as the main thread.
Channels: chan, Tx[T] and Rx[T]
sys.chan(cap) creates a bounded channel and returns (Tx[T], Rx[T]). Both ends are linear, so each is closed (tx.close(), rx.close()) or moved into a thread that will close it. The message type comes from how the ends are used, or you can write it: let (tx, rx): (Tx[u64], Rx[u64]) = sys.chan(4);.
tx.send(v) blocks while the queue is full. That gives you backpressure by default: a fast producer is slowed to the pace of its consumer instead of filling memory. It returns Result[(), SendErr[T]]. If the receiver is gone, the answer is Err(SendErr::Closed(v)), and v is the value you sent, handed back to you. For data that hardly matters. For a resource it matters a great deal: an undelivered Listener or File is yours again, and you have to consume it (Err(SendErr::Closed(l)) => { l.close(); }), because nothing has a destructor to do it for you. For the same reason let _ = tx.send(..) is refused on a channel of resources.
rx.recv(timeout_ms) waits at most timeout_ms milliseconds and returns a Recv[T]: Msg(v) for a message, Timeout if the wait expired with the queue empty, or Closed if every sender is gone. Most languages fold the last two into one ambiguous "nothing". Keeping them apart is what lets a loop distinguish "not yet" from "never again". Recv and SendErr are pub enums in std::threads.
There is no shared box that both threads point into. Everything that crosses the channel is copied or moved at send. An owned Vec crosses by move at any depth inside a message: the buffer changes owner and nothing is copied, so handing a large frame to another thread costs the same as handing over a small one. While a view into that buffer is still in use, the hand-off is refused, because the next owner may write those bytes.
A worker and a pipeline
This program has two shapes in it: a spawned worker that squares a number and reports back through join, and a two-thread bounded pipeline in which one thread produces five squares and the other sums them.
Look closely at the consumer's loop, because it carries the most important rule about channels: a loop that drains a channel ends on Closed and never on Timeout. Closed is a proven end of stream. Every Tx is linear, so the number of live senders is known exactly, and the channel reports Closed when the last one goes. Timeout says only that one wait expired with the queue empty, which means the sender is slow, not finished.
A loop that stops on Timeout abandons every message still on its way. It returns a smaller number, exits with status 0 and traps nothing. Under load, which is exactly when waits expire, that is silent data loss, and the code looks as though it handled every case. In a drain, the timeout is how often the loop wakes up to look again, not a reason to stop.
use std::threads;
use std::log;
fn square(n: u64) -> u64 !{} {
n *% n
}
// Producer: sends five squares, then closes its end.
fn produce(tx: Tx[u64]) -> () !{io, block} {
let mut t = tx;
let mut i: u64 = 0;
while i < 5u64 {
let _r = t.send(i *% i);
i = i +% 1;
}
t.close();
}
// Consumer: drains until the sender closes. A timeout is not an ending.
fn sum_channel(rx: Rx[u64]) -> u64 !{io, block} {
let mut r = rx;
let mut total: u64 = 0;
let mut done = false;
while !done {
match r.recv(100u64) {
Recv::Msg(v) => { total = total +% v; }
Recv::Timeout => {}
Recv::Closed => { done = true; }
}
}
r.close();
total
}
fn main(mut sys: Sys) -> u8 !{io, block, alloc(heap), log} {
let t = sys.spawn(square, 7u64);
let result = t.join();
log::info(f"7 squared is {result}");
// A bounded pipeline: one producer thread, one consumer thread.
let (tx, rx) = sys.chan(4);
let prod = sys.spawn(produce, tx);
let cons = sys.spawn(sum_channel, rx);
prod.join();
let total = cons.join();
log::info(f"sum of squares = {total}");
0
}
test spawn_join_returns_result(mut sim: Sim) {
let t = sim.spawn(square, 7u64);
assert t.join() == 49u64;
}
test pipeline_sums_squares(mut sim: Sim) {
let (tx, rx) = sim.chan(4);
let prod = sim.spawn(produce, tx);
let total = sum_channel(rx);
prod.join();
assert total == 30u64;
}kola run prints [info] 7 squared is 49 and [info] sum of squares = 30. "Give up after N milliseconds" belongs in a function whose result can say it gave up; a drain's cannot. The std helpers threads::drain_into, threads::actor_loop and threads::actor_loop_in follow the same rule.
Shutting down with messages still queued: rx.drain()
Sometimes the receiver has to stop before the senders do: the service is shutting down, or a worker pool is being resized. If the queue holds plain data, rx.close() discards it. If it holds resources (jobs, connections, open files), discarding them would leak them, so rx.close() over a queued resource traps with "rx.close() would drop 1 queued resource message(s); take them back with rx.drain() first".
rx.drain() is the way out. It closes the channel to senders, so any later send answers Err(SendErr::Closed(v)) and hands the value back to its sender. Then it returns one queued message together with the receiver, Some((msg, rx)), or None when the queue is empty, in which case the receiver is consumed. You call it in a loop until it answers None:
resource struct Job { id: u64 }
fn finish(j: Job) -> u64 !{} {
let Job { id } = j;
id
}
// Send a job. If the receiver is gone, the job comes back and is finished here.
fn submit(tx: &mut Tx[Job], j: Job) -> bool !{io, block} {
match tx.send(j) {
Ok(_) => true,
Err(SendErr::Closed(back)) => { let _ = finish(back); false }
}
}
// The owner stops early. Whatever is still queued comes back to it.
fn shut_down(rx: Rx[Job]) -> u64 !{io, block} {
let mut rx = rx;
let mut handed_back: u64 = 0;
loop {
match rx.drain() {
Some((job, next)) => {
handed_back = handed_back +% finish(job);
rx = next;
}
None => { break; }
}
}
handed_back
}
test queued_jobs_are_handed_back(mut sim: Sim) {
let (tx, rx) = sim.chan(4);
let mut tx = tx;
let a = submit(&mut tx, Job { id: 1 });
let b = submit(&mut tx, Job { id: 2 });
let back = shut_down(rx);
let c = submit(&mut tx, Job { id: 3 }); // closed to senders now
assert a && b;
assert back == 3; // both queued jobs came back
assert !c;
tx.close();
}The checker follows this loop: after a loop, the state of each variable is the merge of its state at every break, the same way the two arms of an if are merged. Here the only break is in the None arm, where the receiver has been consumed, so rx is finished when the loop ends. drain runs on the interpreter and the C backend; the own backend refuses it by name for now.
One loop first
Before you reach for anything below, know what a kola service normally looks like: one serve loop. Periodic work such as a heartbeat, a lease renewal or a stats flush does not get its own thread and timer. It is a rate-limited tick inside the same loop: compare the clock with a deadline at the top of each iteration, do the work if it is due, and go on to the next unit of real work otherwise. This is the shape to start from, and the mailbox below is what you escalate to.
One loop touching its own state needs no locks, so lock-ordering bugs cannot be written at all. Its execution order is deterministic, so a bug reproduces the same way each time, and kola report checks reasons about one body rather than an interleaving.
// Periodic work is a rate-limited tick inside the serve loop,
// not a second thread with a timer. The loop owns everything it touches.
struct State { served: u64, ticks: u64, next_tick_ns: u64 }
fn tick(s: State) -> State !{} {
State { served: s.served, ticks: s.ticks +% 1u64, next_tick_ns: s.next_tick_ns }
}
fn serve_loop(clock: &Clock, rx: Rx[u64], period_ns: u64) -> State !{io, block} {
let mut r = rx;
let mut st = State { served: 0u64, ticks: 0u64, next_tick_ns: 0u64 };
let mut running = true;
while running {
// 1. periodic work, due or not, at the top of the iteration
let now = clock.now();
if now >= st.next_tick_ns {
st = tick(st);
st.next_tick_ns = now +% period_ns;
}
// 2. one unit of real work, on a bounded wait
match r.recv(50u64) {
Recv::Msg(v) => { st.served = st.served +% v; }
Recv::Timeout => {}
Recv::Closed => { running = false; }
}
}
r.close();
st
}
fn feed(tx: Tx[u64]) -> () !{io, block} {
let mut t = tx;
let _a = t.send(3u64);
let _b = t.send(4u64);
t.close();
}
test one_loop_serves_and_ticks(mut sim: Sim) {
let (tx, rx) = sim.chan(2);
let feeder = sim.spawn(feed, tx);
let st = serve_loop(&sim.clock, rx, 1000000u64);
feeder.join();
assert st.served == 7u64;
assert st.ticks >= 1u64;
}The cost is that a single loop serializes. One slow unit of work delays everything queued behind it, and the loop must never block on an unbounded wait. Keep every operation in it bounded and it stays fair without a scheduler. Move to a mailbox when a resource must be owned by one thread while several producers reach it, or when one unit of work is long enough to starve the loop.
Many senders, one owner: tx.split and the mailbox
If N workers need one resource and the language refuses shared mutable memory, the resource is owned by one thread and reached by message. That calls for a mailbox: many senders, one receiver. tx.split() -> (Tx[T], Tx[T]) builds it by consuming one sender and returning two.
Every Tx stays linear, so the number of senders is visible in the source, one split per worker, right where the workers are spawned. The channel reports Closed to the owner when the last sender closes. Because the checker proves that every Tx is closed exactly once on every path, the owner can neither stop early (while a sender is still live) nor wait forever (for a Closed that never comes). A cloneable sender has a run-time reference count; a linear one has a proof.
use std::threads;
// The shared resource: a slot counter. Plain data, owned by one thread.
// Nothing here is atomic or locked, because nothing else can reach it.
struct Slots { free: u64, handed: u64 }
// One request folded into the owner's state, passed to actor_loop by name.
fn serve(s: Slots, q: u64) -> Slots !{io, block} {
if q == 1u64 {
if s.free > 0u64 {
return Slots { free: s.free -% 1u64, handed: s.handed +% 1u64 };
}
}
s
}
// The owner's whole body is the std helper: receive until the last
// sender closes, then return the state.
fn pool_owner(rx: Rx[u64]) -> u64 !{io, block, log} {
let end = threads::actor_loop[Slots, u64](rx, Slots { free: 2u64, handed: 0u64 }, serve, 200u64);
end.handed
}
// A worker holds exactly its share of the mailbox and no Sys, so it cannot
// open a channel, a socket or a file. The spawn site is the whole grant.
fn worker(tx: Tx[u64]) -> u64 !{io, block} {
let mut t = tx;
let _sent = t.send(1u64);
t.close();
1u64
}
test two_workers_one_owner(mut sim: Sim) {
let (tx, rx) = sim.chan(4);
let owner = sim.spawn(pool_owner, rx);
// one split per worker: the fan-in count is written out
let (tx, t1) = tx.split();
let w1 = sim.spawn(worker, t1);
let w2 = sim.spawn(worker, tx); // the last share
let a = w1.join();
let b = w2.join();
let handed = owner.join();
assert a == 1u64;
assert b == 1u64;
assert handed == 2u64;
}When a worker needs its own answer back, the request carries its route home: resource struct Ask { q: u64, reply: Tx[u64] }, with a reply channel inside the message. Every answer is then point to point, with no correlation ids and no shared reply queue.
Waiting on a mailbox or a socket
A daemon has to react to incoming connections and to messages from its workers without polling each in turn. A Poller is kola's readiness waiter, the counterpart of epoll or select in C: you register several things with it and then wait until any of them is ready. poller.add_chan(&rx) registers a channel's receiving end in the same Poller as sockets (the socket half is in Networking). One wait then sleeps until a socket is ready, a message is queued or the last sender has closed, and next_ready says which. A ready mailbox is drained without blocking, with rx.recv(0).
That zero timeout is the one legitimate exception to the drain rule above. With a timeout of 0, nothing was waited for, so Timeout cannot mean "the sender is slow". It means "nothing is queued right now", which ends this batch and returns control to wait. The stream itself still ends where it always does, on Closed. Likewise a single, non-looping recv with a real timeout ("no reply within 100 ms, give up") is fine, because there is no rest of the stream to lose.
use std::net;
use std::threads;
fn producer(tx: Tx[u64]) -> () !{io, block} {
let mut t = tx;
let mut i: u64 = 0;
while i < 3u64 {
let _r = t.send(i +% 100u64);
i = i +% 1;
}
t.close();
}
test select_over_mailbox(mut sim: Sim) {
let (tx, rx) = sim.chan(2);
let prod = sim.spawn(producer, tx);
let mut rxm = rx;
let mut poller = Poller::new();
let mid = poller.add_chan(&rxm).unwrap();
let mut total: u64 = 0;
let mut running = true;
while running {
let _n = poller.wait(500u64).unwrap();
match poller.next_ready() {
Some(id) => {
assert id == mid;
// the mailbox is ready: drain it without blocking
match rxm.recv(0u64) {
Recv::Msg(v) => { total = total +% v; }
Recv::Timeout => {}
Recv::Closed => { running = false; }
}
}
None => { running = false; }
}
}
assert total == 303u64;
let _p = poller.close();
rxm.close();
prod.join();
}Sleeping without starving the others
clock.sleep(ms) is the one Clock operation that waits, which is why its row is !{io, block}. Like every blocking capability operation, it hands the scheduler to another thread, so a sleeping thread never stops its siblings from running. Under a simulated clock a sleep advances virtual time without any real waiting, so testing a timeout path costs nothing. A freestanding build refuses sleep, because there is no scheduler there to yield to.
fn nap_then_report(clock: &Clock, ms: u64) -> u64 !{io, block} {
let t0 = clock.now();
clock.sleep(ms);
let t1 = clock.now();
t1 -% t0
}
test the_sim_clock_advances_virtually(mut sim: Sim) {
// 10 ms in nanoseconds, plus 1000 ns for the second reading:
// every read of the simulated clock advances it.
let elapsed = nap_then_report(&sim.clock, 10u64);
assert elapsed == 10001000u64;
}A timeout bounds one wait, not a loop of them. A peer that answers just inside each timeout can keep a loop of recv(100) calls going forever. A real deadline is one clock.now() taken before the loop and compared on every turn, the same pattern as the tick in the serve loop.
When the compiler refuses
Both guarantees are enforced. A reference cannot be passed as a spawn context:
fn work[r](b: &r [u8]) -> usize !{} { b.len }
fn main(mut sys: Sys) -> u8 !{io, block} {
region r {
let b = alloc in r [1u8, 2u8, 3u8];
let t = sys.spawn(work, b);
let _n = t.join();
}
0
}error[card §11]: a reference cannot cross a thread boundary in a spawn context
--> r1.kl:6:27
|
6 | let t = sys.spawn(work, b);
| ^^^^
= help: move owned data or transferable resources; InterruptGuard holds its installing thread and must be restored thereA Pool cannot cross either, for a reason its type does not show: a pool of data is copyable, so moving it into a spawn context would leave both threads holding the same slots. Own it in one thread and reach it by message.
A thread that is spawned and never joined is an unconsumed resource:
fn work(n: u64) -> u64 !{} { n }
fn main(mut sys: Sys) -> u8 !{io, block} {
let t = sys.spawn(work, 1u64);
0
}error[card §2]: resource `t` is not consumed on this path
--> r2.kl:4:5
|
4 | let t = sys.spawn(work, 1u64);
| ^^^
= help: no implicit destructors: move it on, return it, destructure it, or pass it to a consuming fnIt is the same diagnostic as for any unconsumed resource (Resources and linearity). An Option[Thread[R]] holding None owes nothing, so declaring a slot first and filling it later is ordinary code; the Some carries the obligation. The same holds one layer out: a File moved into a worker (Filesystem) or a Conn handed to a connection thread (Networking) travels by move, never by borrow.
Determinism, and what runs in parallel
Because memory is never shared, a schedule is observable only at channel operations, spawns and joins, which is a small space that can be replayed. The interpreter, under both kola run and kola test, runs threads cooperatively, one at a time, switching only at those operations, on a seeded deterministic scheduler: the same --seed N reproduces the same trace every time. Under the simulated clock every bounded wait (recv, wait_readable, clock.sleep) is due at an absolute virtual time. The earliest one fires first and virtual time jumps to it, so a 30-second wait next to a sibling's 700-second sleep measures 30 seconds. Ties go to the lower thread id. A denied wait, such as recv(0) answering Timeout, yields before the caller retries, so a polling loop cannot starve its siblings.
The normative reference is SPEC-CARD §11 (kola help spec-card).