Concurrency patterns

A worker pool#

Task. Hand a list of jobs to several workers and collect the results.

oxl✓ checked in CI

fn square(n: i64) -> i64  pure => n * n

fn worker(jobs: Receiver<i64>, results: Sender<(i64, i64)>) io, time {
  while let n? = jobs.recv().ok() {  // loops while at least one sender is alive
    _ = results.send((n, square(n)))
  }
}

fn main() io, time {
  let (resTx, resRx) = chan<(i64, i64)>(cap: 8)
  var senders: List<Sender<i64>> = []
  var sums: Map<i64, i64> = {:}
  scope {
    for _ in 0..3 {
      let (tx, rx) = chan<i64>(cap: 4)  // every worker has its own job channel
      senders.push(tx)
      _ = spawn worker(rx, resTx.clone())  // a Sender can be cloned, a Receiver cannot
    }
    var i = 0
    for n in 1..=6 {
      _ = senders[i % 3].send(n)  // deal the jobs round-robin
      i += 1
    }
    drop(senders)  // closed the job channels: the workers will finish
    drop(resTx)
    while let item? = resRx.recv().ok() {
      sums[item.0] = item.1
    }
  }
  var keys = sums.keys()
  keys.sort()
  for k in keys { print("{k}^2 = {sums.get(k) ?? 0}") }
}

Output

1^2 = 1
2^2 = 4
3^2 = 9
4^2 = 16
5^2 = 25
6^2 = 36

Every worker has its own job channel: a Receiver is a single consumer and cannot be cloned, but a Sender can - all workers write results into one channel through resTx.clone().
Closing a channel means destroying all its senders: drop(senders) closes the job channels, the workers leave while let n? = jobs.recv().ok(); drop(resTx) releases the original, and the result-reading loop ends when the last clone is gone.
Results arrive in arbitrary order, so in the example they are put in a Map and printed by sorted keys.
scope waits for all workers: after it there are guaranteed to be no "orphaned" tasks.

A pipeline#

Task. Process a stream of values through consecutive stages, each in its own task.

oxl✓ checked in CI

fn gen(tx: Sender<i64>) io, time {
  for i in 1..=5 { _ = tx.send(i) }
}

fn double(rx: Receiver<i64>, tx: Sender<i64>) io, time {  // a stage: reads, transforms, writes on
  while let n? = rx.recv().ok() { _ = tx.send(n * 2) }
}

fn addTen(rx: Receiver<i64>, tx: Sender<i64>) io, time {
  while let n? = rx.recv().ok() { _ = tx.send(n + 10) }
}

fn main() io, time {
  let (aTx, aRx) = chan<i64>(cap: 2)
  let (bTx, bRx) = chan<i64>(cap: 2)
  let (cTx, cRx) = chan<i64>(cap: 2)
  var out: List<i64> = []
  scope {
    _ = spawn gen(aTx)  // aTx moved into gen: when gen ends the channel closes
    _ = spawn double(aRx, bTx)
    _ = spawn addTen(bRx, cTx)
    while let n? = cRx.recv().ok() { out.push(n) }
  }
  print("{out}")
}

Output

[12, 14, 16, 18, 20]

Every stage is a function "channel in -> channel out". The buffer size (cap: 2) limits how far a stage can run ahead: a fast stage waits for a slow one, memory does not grow.
The sender moves into the stage function and is closed when it ends - this "end of data" signal cascades through the whole pipeline.
The order is kept because every stage has one processing thread; for parallelism inside a stage use a worker pool.

A timeout#

Task. Wait for a result no longer than a given time.

oxl✓ checked in CI

fn slowLookup(ms: i64) -> str  time {
  sleep(Duration.ofMillis(ms))
  "found after {ms}ms"
}

fn attempt(ms: i64) -> str  io, time {
  let (tx, rx) = chan<str>(cap: 1)
  scope {
    spawn {
      _ = tx.send(slowLookup(ms))
    }
    select {
      recv str v from rx => v
      after 50ms => "timed out"  // wait at most 50 ms
    }
  }
}

fn main() io, time {
  print(attempt(10))
  print(attempt(500))
}

Output

found after 10ms
timed out

select waits for the first ready event: a value from the channel or after 50ms. After the timeout the slow task keeps running until the end of the scope; to cancel it use the spawn handle (handle.cancel()).
For call chains there is a shorter form - … within 300ms (tour), which also cancels nested calls.

Shared state: STM#

Task. Transfer money between accounts from several tasks without locks and races.

oxl✓ checked in CI

use std.stm { TVar, check }

static FROM: TVar<i64> = TVar(100)
static TO: TVar<i64> = TVar(0)

fn transfer(amount: i64) {
  atomic {  // a transaction: all or nothing, restarted on conflict
    let have = FROM.get()
    check(have >= amount)  // not enough money: wait until FROM changes
    FROM.set(have - amount)
    TO.update(x => x + amount)
  }
}

fn main() io, time {
  scope {
    for _ in 0..5 { _ = spawn transfer(10) }
  }
  let a = atomic { FROM.get() }
  let b = atomic { TO.get() }
  print("from {a} to {b} total {a + b}")
}

Output

from 50 to 50 total 100

atomic { … } is a transaction: either all changes are visible or none; on a conflict with another transaction it is restarted, so the body of atomic must have no side effects other than operations on TVar.
check(have >= amount) suspends the transaction until one of the variables it read changes - this is how waiting for a condition is implemented without polling.
The invariant "the sum is constant" (50 + 50) holds under any interleaving of five concurrent transfers.

Streams#

Task. Hand values to a receiver as they become ready.

oxl✓ checked in CI

fn countdown(from: i32) -> stream<i32> {  // a stream is an asynchronous sequence
  for i in 0..from {
    yield from - i  // yield hands the next value to the consumer
    sleep(1ms)
  }
}

fn main() io, time {
  var seen: List<i32> = []
  for await n in countdown(4) {  // for await reads the stream as values arrive
    seen.push(n)
  }
  print("{seen}")
}

Output

[4, 3, 2, 1]

A function with the result stream<T> may use yield; the value is handed to the receiver at once and the function continues after yield.
for await x in stream { … } reads the stream to the end; streams support the std.stream operators (map, filter, batch, merge…) and connect with |> - see the streams example.

Back: cookbook.

Concurrency patterns

A worker pool#

A pipeline#

A timeout#

Shared state: STM#

Streams#

See also