Channels and Strands in Rill: A Deep Dive into Concurrency

In the previous post we took a general tour of Rill. This post focuses on one topic, which I think is the most enjoyable side of the language: strands and channels. We will go from exactly what a rendezvous channel guarantees to buffered channels, from broadcasting with close to pipelines, from a concurrent prime sieve to timeouts with select, from graceful shutdown on a signal to SDL-style state machines and the sharing check on several cores.

I compiled and ran every program in this post before writing it; the output blocks are the real output of those runs. (The strings inside the programs are in Turkish, because they are exactly the programs I ran.) The measurements were taken on an Apple M4 with a single worker unless stated otherwise.

What is a strand?

A strand is Rill's lightweight thread. It is the same idea that Go calls a goroutine, Erlang a process and Java a virtual thread; the name comes from the strands of a rope: a program is made of many strands twisted together. (A note for C++ network programmers: strand in Boost.Asio is something else entirely, a serialised execution context. The two are unrelated.)

One keyword is enough to start a strand:

spawn worker(jobs, results)

spawn runs the whole expression on the new strand. That small detail explains a classic trap when writing servers. If you write spawn handle(accept(server)), the accept also runs on the new strand. The main loop never waits; every turn it spawns a new strand parked on accept and spins. The right way is to bind first and then hand over:

fn serve(server) =
  conn = tcp_accept(server)
  spawn handle(conn)
  serve(server)

Here the loop itself is the one that waits, and it hands the connection to the strand. Because serve calls itself in tail position, the stack does not grow either.

Strands are cheap because Rill uses a copy-on-block stack model. A running strand runs on its worker's stack; when it parks, the part of the stack it used is copied and kept. In the measurements in the Rill repository a parked strand takes about 300 bytes.

The rendezvous channel: a handshake

channel() creates a channel with no buffer. On such a channel the sender waits until a receiver arrives, and the receiver waits until a sender arrives. The value only changes hands when the two meet. To see it, let's send three messages between two strands and print every step:

fn ping(ch, n) =
  if n == 0 then ()
  else
    println("ping: " + int_to_str(n) + " gönderiliyor")
    send(ch, n)
    println("ping: " + int_to_str(n) + " teslim edildi")
    ping(ch, n - 1)

fn pong(ch, k) =
  if k == 0 then ()
  else
    v = recv(ch)
    println("pong: " + int_to_str(v) + " alındı")
    pong(ch, k - 1)

fn main() =
  ch = channel()
  spawn ping(ch, 3)
  pong(ch, 3)

ping: 3 gönderiliyor
ping: 3 teslim edildi
ping: 2 gönderiliyor
pong: 3 alındı
pong: 2 alındı
ping: 2 teslim edildi
ping: 1 gönderiliyor
ping: 1 teslim edildi
pong: 1 alındı

The order of the lines may look jumbled at first, but it reflects the rule exactly. No "teslim edildi" (delivered) line can happen before that value has been received on the other side. Since the printing continues independently on both strands after the handshake, which one reaches the screen first is up to the scheduler. What is guaranteed is not order but synchronisation: a send returning is proof that a recv has completed.

Buffered channels and mailboxes

channel(n) holds up to n values in a buffer. The sender only waits when the buffer is full, and the receiver only when it is empty. To watch the producer run ahead, let's put the consumer to sleep for 50 ms:

fn producer(ch, i, n) =
  if i > n then ()
  else
    send(ch, i)
    println("üretici: " + int_to_str(i) + " kuyruğa kondu")
    producer(ch, i + 1, n)

fn main() =
  ch = channel(3)
  spawn producer(ch, 1, 5)
  sleep_ms(50)
  println("tüketici uyandı")
  for _ in 0..5
    println("tüketici: " + int_to_str(recv(ch)))

üretici: 1 kuyruğa kondu
üretici: 2 kuyruğa kondu
üretici: 3 kuyruğa kondu
tüketici uyandı
tüketici: 1
üretici: 4 kuyruğa kondu
tüketici: 2
tüketici: 3
tüketici: 4
tüketici: 5

The producer queued the first three values without waiting for the consumer at all, and parked on the fourth because the buffer was full. It woke up once the consumer took a value. The buffer absorbs the difference in speed between producer and consumer, and at the same time gives natural backpressure: the producer cannot get arbitrarily far ahead of the consumer.

There is a third option: mailbox(). A mailbox grows without bound, so the sender never waits. It is ideal for two parties sending to each other at the same moment, because two buffered channels can each wait for the other to drain and deadlock. Its cost is clear too: nothing slows the sender down. In a moment we will see why mailboxes are the default for state machines.

Finishing: close and recv_opt

A strand parked on recv stays there until a value arrives. Tearing down whatever was going to send it leaves the strand where it is, holding its stack. In a server that opens a strand per connection, that is a leak as large as the number of connections it has ever had. close exists for exactly this:

fn produce(ch, n) =
  if n <= 0 then close(ch)
  else
    send(ch, n)
    produce(ch, n - 1)

fn drain(ch, acc) = match recv_opt(ch)
  None -> acc
  Some(v) -> drain(ch, acc + v)

fn main() =
  ch = channel(4)
  spawn produce(ch, 100)
  println(drain(ch, 0))
  println(closed(ch))

5050
true

Several important design decisions show in this example:

• close is a broadcast. Sending a value reaches one receiver; closing reaches everyone parked on the channel at once. A program shutting down does not need to know how many strands are waiting.
• Buffered values are not lost. The producer can close the moment it has nothing more to say, and the consumer still sees everything in flight. The total coming out at exactly 5050 is the proof.
• Closing twice is not an error. A shutdown that runs from two directions does not need to arrange who gets there first.
• recv_opt is for channels that end. It returns Some(v) while there is a value, and None once the channel is closed and empty.

So why doesn't every recv return an Option? Because most channels never end. Making every one of them an Option would put a match in front of every receive in the language to pay for the few that do. Instead, a plain recv from a closed channel stops the program and says what to do:

fn main() =
  ch = channel(1)
  close(ch)
  println(recv(ch))

rill: recv on a closed channel - use recv_opt

That is the same answer an index outside a buffer gets, and for the same reason: there is no value to hand back that would not be a lie. Sending into a closed channel stops the program too. The rule that avoids it is the one Go arrived at: whoever sends, closes.

Pipelines

Put channels and close together and building a pipeline of connected stages is very natural. Each stage closes its own output when its input closes, so the "done" information flows down the chain by itself. Let's generate the numbers from 1 to 1,000,000, keep the multiples of 3, square them and add them up:

fn numbers(out, i, n) =
  if i > n then close(out)
  else
    send(out, i)
    numbers(out, i + 1, n)

fn keep(inp, out, p) = match recv_opt(inp)
  None -> close(out)
  Some(v) ->
    if p(v) then send(out, v) else ()
    keep(inp, out, p)

fn square(inp, out) = match recv_opt(inp)
  None -> close(out)
  Some(v) ->
    send(out, v * v)
    square(inp, out)

fn total(inp, acc) = match recv_opt(inp)
  None -> acc
  Some(v) -> total(inp, acc + v)

fn main() =
  a = channel(64)
  b = channel(64)
  c = channel(64)
  spawn numbers(a, 1, 1000000)
  spawn keep(a, b, \x -> x % 3 == 0)
  spawn square(b, c)
  println(total(c, 0))

111111277777611111

I checked the result independently: the sum of the squares of the multiples of 3 is exactly this number by the formula 9 · n(n+1)(2n+1)/6 with n = 333,333. It also matters that the keep stage takes a lambda: stages are ordinary functions that can be reused and combined however you like. The buffers of 64 between the stages keep the strands from waiting on each other for every single value.

A concurrent prime sieve

There is a beautiful example everyone learning Go runs into at some point: the concurrent sieve of Eratosthenes. One strand generates numbers starting from 2. The main flow takes the first number at the end of the chain as a prime and adds a new strand to the chain that says "filter out the multiples of this prime". In Rill it looks like this:

fn generate(ch, i) =
  send(ch, i)
  generate(ch, i + 1)

fn filter_out(inp, out, p) =
  v = recv(inp)
  if v % p != 0 then send(out, v) else ()
  filter_out(inp, out, p)

fn sieve(ch, k, acc) =
  if k == 0 then reverse(acc)
  else
    p = recv(ch)
    next = channel()
    spawn filter_out(ch, next, p)
    sieve(next, k - 1, Cons(p, acc))

fn main() =
  ch = channel()
  spawn generate(ch, 2)
  println(sieve(ch, 20, Nil))

Cons(2, Cons(3, Cons(5, Cons(7, Cons(11, Cons(13, Cons(17, Cons(19, Cons(23, Cons(29, Cons(31, Cons(37, Cons(41, Cons(43, Cons(47, Cons(53, Cons(59, Cons(61, Cons(67, Cons(71, Nil))))))))))))))))))))

For the first 20 primes, 21 strands and 21 rendezvous channels are running. There is no shared memory and no lock anywhere; each strand only knows its own input and output channel. generate and filter_out are infinite loops, but both are tail calls, so the stack does not grow. When main returns the program ends, and the strands left behind are abandoned, just as in Go.

select: waiting on several channels at once

Real programs rarely wait on just one channel. select waits on several channels at once and runs the arm of whichever is ready first. Arms are tried in order, so on a tie the earlier arm wins. A send arm is ready when the send can go through. There can be at most 64 arms.

Polling without blocking: default

A default arm turns select into a non-blocking poll: if no channel is ready, that arm runs immediately.

fn poll(ch) =
  v = select
    x = recv(ch) -> "hazır: " + int_to_str(x)
    default -> "henüz bir şey yok"
  v

fn main() =
  ch = channel(1)
  println(poll(ch))
  send(ch, 42)
  println(poll(ch))

henüz bir şey yok
hazır: 42

Watching a clock and a job together

Rill has no Ticker type, because a ticker is three lines and a channel. Let's write a watcher that prints a progress message at a steady rate until a long job finishes, by watching a ticker and the job in the same select:

fn ticker(ch, ms) =
  sleep_ms(ms)
  send(ch, 1)
  ticker(ch, ms)

fn job(done) =
  sleep_ms(230)
  send(done, "iş bitti")

fn watch(tick, done, n) =
  r = select
    msg = recv(done) -> msg + " (" + int_to_str(n) + " tik)"
    t = recv(tick) ->
      println("  ... hâlâ çalışıyor, tik " + int_to_str(n + t))
      watch(tick, done, n + t)
  r

fn main() =
  tick = channel(1)
  done = channel()
  spawn ticker(tick, 50)
  spawn job(done)
  println(watch(tick, done, 0))

  ... hâlâ çalışıyor, tik 1
  ... hâlâ çalışıyor, tik 2
  ... hâlâ çalışıyor, tik 3
  ... hâlâ çalışıyor, tik 4
iş bitti (4 tik)

Giving the ticker a channel(1) is a design choice: a tick can wait while the watcher is busy, but ticks do not pile up. A sleeping strand uses no processor at all; a thousand strands can be asleep at once and the program is still idle. If you are writing a fixed-rate loop, sleep for the distance to the next deadline rather than a constant; otherwise the slack in the operating system's timers adds up.

Deadlines: timeout(ms)

A timeout(ms) arm becomes ready when nothing else has been ready for that long. We saw it in the previous post; the important engineering detail here is that this arm spends no strand and no channel. It is parked inside the poller next to the channels. A server that puts a deadline on every request in flight does not pay a strand per request for it. A few rules:

• The shortest deadline wins, so one wait can carry both a warning and a limit.
• A deadline already in the past is ready at once. So timeout(0) is a poll, another spelling of default.
• default and timeout cannot be written in the same select, because default means "do not wait" and the deadline could never be reached.

Graceful shutdown: signals and close

We already have the two pieces a server needs to shut down cleanly on Ctrl-C: wait_signal() parks until the program receives a signal and says which one, and close broadcasts that to everyone. One strand waits and tells the others; the waiter does not need to know how many others there are.

fn worker(id, stop, n) =
  r = select
    recv(stop) -> "işçi " + int_to_str(id) + " durdu, " + int_to_str(n) + " iş yapmıştı"
    timeout(20) -> worker(id, stop, n + 1)
  r

fn run(id, stop, report) = send(report, worker(id, stop, 0))

fn main() =
  stop = channel()
  report = channel(4)
  for id in 0..3
    spawn run(id, stop, report)
  println("hazır, Ctrl-C bekleniyor")
  s = wait_signal()
  println("sinyal: " + int_to_str(s))
  close(stop)
  for _ in 0..3
    println(recv(report))

I started the program and sent kill -INT about half a second later:

hazır, Ctrl-C bekleniyor
sinyal: 2
işçi 0 durdu, 25 iş yapmıştı
işçi 1 durdu, 25 iş yapmıştı
işçi 2 durdu, 25 iş yapmıştı

The subtle point is that the recv(stop) -> arm binds nothing. A closed channel makes a receive arm ready, but it has no value to produce. An arm written as x = recv(stop) -> would have to produce a value, and the program would stop saying so. That way close(stop) reaches every worker, whatever the other arm is doing.

On the signal side a few details have been thought through as well. A signal can arrive on any thread the operating system picks, between any two instructions. So nothing the program wrote runs there: the handler writes a single byte to a pipe and everything else happens on the strand. SIGUSR1 and SIGUSR2 are deliberately missing, because their numbers are 30/31 on macOS and 10/12 on Linux. A number that means one thing where it was written and another where it runs is worse than not having it at all. For re-reading configuration there is the traditional SIGHUP. A program that is only waiting for a signal is not considered deadlocked, because it is waiting on something that answers.

A deadlock is not silent

A program in which everyone waits for everyone else says so instead of hanging. If two strands each wait to receive a value before sending:

fn main() =
  a = channel()
  b = channel()
  spawn send(a, recv(b))
  send(b, recv(a))

rill: deadlock - all strands are blocked

The runtime does not count sleeping strands, or strands waiting on a socket or a signal, as deadlocked, because they are waiting on something that answers. It reports the situation and stops the program only when everyone is waiting on each other on channels.

State machines: machine

A strand that waits for a signal, does what the signal asks according to the state it is in, and waits again: that is what telephone exchanges called a process and wrote in SDL. In Rill it is a tail-recursive function over a channel and a state. machine is the way to write that function as the machine it is. Let's write a turnstile that opens when paid and locks again when someone passes, or when nobody passes within 3 seconds:

type Signal
  Coin(kurus: Int)
  Push
  Report
  Status(text: Str)

type Gate
  Locked(credit: Int)
  Open

machine gate(me: Chan(Mail(Signal)), log: Chan(Str))
  state Locked(credit)
    Coin(k) if credit + k >= 500 ->
      send(log, "jeton tamam, kapı açıldı")
      Open
    Coin(k) ->
      send(log, "bakiye: " + int_to_str(credit + k))
      Locked(credit + k)
    Push ->
      send(log, "kilitli, önce ödeme")
      stay
  state Open
    Push ->
      send(log, "geçildi, kilitlendi")
      Locked(0)
    Coin(_) ->
      send(log, "zaten açık, para iade")
      stay
    after 3000 ->
      send(log, "kimse geçmedi, yeniden kilitlendi")
      Locked(0)
  state *
    Report ->
      reply(Status("rapor istendi"))
      stay

fn show_log(log, n) =
  if n == 0 then ()
  else
    println(recv(log))
    show_log(log, n - 1)

fn main() =
  env = mailbox()
  g = mailbox()
  log = channel(16)
  spawn gate(g, log, Locked(0))
  post(g, env, Push)
  post(g, env, Coin(250))
  post(g, env, Coin(250))
  post(g, env, Coin(100))
  post(g, env, Push)
  post(g, env, Coin(500))
  show_log(log, 6)
  println(recv(log))
  post(g, env, Report)
  println(recv(env).signal)

kilitli, önce ödeme
bakiye: 250
jeton tamam, kapı açıldı
zaten açık, para iade
geçildi, kilitlendi
jeton tamam, kapı açıldı
kimse geçmedi, yeniden kilitlendi
Status(rapor istendi)

The machine reads almost like a specification:

• The states are the constructors of an ordinary type (Gate). The types of their fields are written once.
• Under each state, the signals it answers sit like the arms of a match. An arm's value is the next state. stay means remaining in the current state, and stop means ending the strand.
• after 3000 -> is the arm taken when no signal arrives for that long.
• state * holds the arms every state answers; here every state can report.
• Everything that arrives is a Mail: the signal and the mailbox that sent it. reply(sig) answers the sender; the signal does not need to carry a reply channel. When main talks from outside, post(to, from, sig) names the mailbox the answer comes back to (env).
• A signal that no arm names is consumed and the state kept; that is what SDL does with an unexpected signal. Every state of the type has to appear in the machine, or the compiler names the missing one.

The compiler turns this machine into exactly the function you would write by hand: a match on the state, a recv in each state (or a select with a timeout where there is an after), and tail calls that call itself with the next state. It costs nothing beyond the wait.

Better still, the machine is a diagram before it is code. rill diagram turnstile.rill --svg draws the SDL process diagram below straight from the declaration. The diagram is read off the code and not kept beside it, so it can never fall behind:

Virtual time: testing the model

The last line of the turnstile really waits 3 seconds because of the 3-second after. With RILL_VIRTUAL_TIME=1 the program runs on a clock of its own: time does not pass while anything runs, and when every strand is waiting and the nearest thing waited for is a timer, the clock jumps straight to it.

$ time ./turnstile
real 3.18
$ RILL_VIRTUAL_TIME=1 time ./turnstile
real 0.00

The output is identical, and the time drops from 3.18 seconds to zero. On top of that, with virtual time the trace of a run is the same every time, which makes a model's behaviour testable. RILL_TRACE=1 writes every machine event as a line, and rill msc turns those lines into the message sequence chart SDL tools used to draw.

Machines can do much more: timers that send the machine a signal later (set(Retry, 200)), keeping a signal for a later state (save), priority (priority), nested states, and system block diagrams whose routes between machines are checked at compile time. Those are the subject of another post.

Scale: a million strands

I wanted to measure how cheap strands are myself. The program below starts N strands, each parked on a "stop" channel. Then a single close wakes them all, and each one reports to a mailbox:

fn parked(stop, done) =
  r = select
    recv(stop) -> 1
  send(done, r)

fn count(done, k, acc) = if k == 0 then acc else count(done, k - 1, acc + recv(done))

fn main() =
  n = str_to_int(unwrap_or(nth(args(), 1), "1000"))
  stop = channel()
  done = mailbox()
  t0 = now_nanos()
  for _ in 0..n
    spawn parked(stop, done)
  yield()
  sleep_ms(1)
  t1 = now_nanos()
  close(stop)
  woke = count(done, n, 0)
  t2 = now_nanos()
  println(int_to_str(n) + " strand: başlat+park " + int_to_str((t1 - t0) / 1000000) + " ms, close ile uyandır " + int_to_str((t2 - t1) / 1000000) + " ms, uyanan " + int_to_str(woke))

1000 strand: başlat+park 1 ms, close ile uyandır 0 ms, uyanan 1000
10000 strand: başlat+park 5 ms, close ile uyandır 1 ms, uyanan 10000
100000 strand: başlat+park 22 ms, close ile uyandır 11 ms, uyanan 100000
1000000 strand: başlat+park 162 ms, close ile uyandır 124 ms, uyanan 1000000

A million strands start and park in 162 ms, and a single close broadcast wakes them all in 124 ms. Time and memory scale linearly. To be honest about memory: in this program it comes to about 1.5 KB per strand. That is more than the repository's "parked strand ≈ 300 bytes", because here every strand waits inside a select and, once woken, writes into an unbounded mailbox; that mailbox of a million messages is part of the memory too. These are single-run measurements; the point is to show the order of magnitude rather than an exact number.

More than one core

By default all strands run on a single operating system thread. That is what lets reference counting use non-atomic, and therefore fast, operations. Real parallelism is asked for explicitly, twice: when building and when running.

rill build main.rill -o main --parallel   # reference counts become safe across threads
RILL_THREADS=8 ./main                     # run on eight workers

Workers steal strands that have not started yet from each other. A strand that has already started stays on its worker, because its stack frames live at that worker's addresses. Parallelism helps when strands compute independently. Rendezvous chains where one strand hands a value to the next (like the sieve or the ring above) have nothing to overlap; they run fastest on a single worker.

Sharing is checked at compile time

The one kind of value two strands can disagree about is storage: a Buf, a Map, a StrBuf, or anything that holds one. One writing while the other reads or writes is a race. A --parallel build checks what strands share before the program runs:

fn fill(b: Buf(Int), v) = for i in 0..#b
  b[i] = v

fn main() =
  b = buf_int(8)
  spawn fill(b, 1)
  spawn fill(b, 2)
  println(int_to_str(b[0]))

$ rill check share.rill --parallel
share.rill: type error at line 6: `b` is written by this strand and still used here after
the `spawn`: two strands writing one piece of storage is a race. Hand it over (do not use
it after the spawn), or send each strand its own, or, where the writes are kept apart on
purpose, say so: `apart(b)`

The rules are simple:

1. A value handed to a strand (captured by spawn or sent down a channel) that the giving side does not use afterwards now belongs to that strand. Anything goes.
2. A value used afterwards is shared. Sharing to read is fine. Sharing and writing is refused, and the error names the value and the line.
3. A function value is opaque: what it holds and writes is not in its type. Sharing one between strands is refused unless the program says what it knows.

What the program knows is said with apart(x): "this value is shared on purpose and the writes are kept apart". Each strand might be writing its own range of a buffer. The compiler cannot check the claim, but it makes sure there is a claim to find. In a single-worker build, strands only interleave where they wait, so the same program is not checked there.

A real example: this blog's server

This page is served with exactly these tools. Rill's HTTP server opens a strand for every connection; the strand of a quiet connection is parked and holds nothing. The database connections are a pool sitting inside a buffered channel: a request takes a connection, uses it and puts it back. A request that finds no connection waits, which gives the server free backpressure:

fn with_conn(pool, ask) =
  fd = recv(pool)
  res = ask(fd)
  send(pool, fd)
  res

When a post is read, the view record is written to the database on a separate strand; the page does not wait for the database to write to disk:

fn spawn_view(app: App, slug, referrer) =
  spawn with_conn(app.db, \fd -> record_view(fd, slug, referrer))
  ()

Here spawn running the whole expression on the new strand is exactly what we want: taking a connection from the pool, writing, and putting the connection back all happen outside the strand that answers the request.

Summary

• channel() is a handshake: a send returning is proof that a recv completed. channel(n) absorbs differences in speed and gives backpressure. mailbox() never waits.
• close is a broadcast, loses nothing that is buffered, and can be done twice. recv_opt is for channels that end. Remember the rule: whoever sends, closes.
• select waits on several channels. default is a non-blocking poll, timeout(ms) a deadline that spends no strand.
• wait_signal + close are the two halves of a graceful shutdown. A real deadlock is not silent.
• machine writes a strand as an SDL-style state machine; rill diagram draws it and virtual time makes it testable.
• --parallel gives real parallelism and refuses writes to shared storage at compile time; deliberate exceptions are stated with apart.

Strands are so cheap that the design question stops being "how many threads can I open?" and becomes "what is the natural shape of this work?". A connection is a strand, a stage is a strand, a prime is a strand.

In the next post I will write about OxiDB, the database behind this blog. I would be glad to hear your questions and experiments on X or LinkedIn.