Step 11. Concurrency

Why#

Concurrent programs break where a task gets "lost": it went into the background, failed and nobody noticed, or keeps running when its result is no longer needed. In OxL tasks are structured: a task started in a scope block cannot outlive that block, and an error of one task cancels the others. There are no async/await keywords and no "coloring" of functions: any suspendable function can be called in the usual way, and it yields the thread to the scheduler when it has to wait.

Three tools for different cases:

scope and spawn#

fn compute(n: i32) -> i32  time {
  sleep(10ms)
  n * n
}

fn main() io, time {
  let (a, b) = scope {  // scope waits for all children
    let ha = spawn compute(3)  // spawn only works inside a scope
    let hb = spawn compute(4)
    (ha.join()!, hb.join()!)
  }
  print("{a} {b}")
}

9 16

• scope { … } is a block that waits for everything started inside it with spawn. The value of the block is its last expression.
• spawn compute(3) starts the call as a separate task and returns a handle; ha.join() waits for the result.
• Both tasks sleep for 10 ms at the same time, so the whole block takes about 10 ms, not 20.

all: in parallel, keeping the order#

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("slow", 60ms) }, Task.of { job("fast", 5ms) }])  // all: in parallel, the order of results is kept
  for r in rs { print(r) }
}

done fast
done slow
slow
fast

Task.of { … } defers a call, and all starts all tasks at once. The results come back in the order in which the tasks are listed (not in the order of completion: fast printed "done" earlier, but it is second in the list). If one of the tasks fails, the others are cancelled. The related race and any return the first result or the first success.

Channels and select#

fn producer(tx: Sender<i32>) time {
  for i in 1..=3 {
    _ = tx.send(i * 10)
    sleep(5ms)
  }
}

fn main() io, time {
  let (tx, rx) = chan<i32>(cap: 2)  // a channel with a buffer of 2 values
  scope {
    spawn producer(tx)
    var got = 0
    while got < 3 {
      let r = select {  // select takes the first ready branch
        recv i32 v from rx => "got {v}"
        after 500ms => "timeout"  // the timeout branch
      }
      print(r)
      got += 1
    }
  }
}

got 10
got 20
got 30

• chan<i32>(cap: 2) creates a (Sender, Receiver) pair with a buffer for 2 values; send waits when the buffer is full - this gives natural back pressure on a fast producer.
• select { … } waits for the first ready branch: recv … from rx, send, or after 500ms - a timeout. Add an after branch so that select never waits forever.
• Values passed between tasks must be "transferable": no Rc and no borrows (§7.3). The compiler checks this rule, so there are no data races in safe code.

Actors#

actor Counter {
  var n: i32 = 0  // the actor's state is reachable only from its msgs
  msg inc(by: i32) { n += by }
  msg get() -> i32 { n }
}

fn main() io {
  let c = spawn Counter()
  c.inc(2) -> c.inc(3) -> c.get() -> print("{it}")  // messages are processed one at a time, in order
}

5

An actor owns its var state, and only its msg methods can change it. Messages are processed one at a time in the order of arrival, so no locks are needed. spawn Counter() creates an actor, and c.inc(2) -> c.inc(3) -> c.get() -> print("{it}") is a chain where each following message is sent after the answer to the previous one. Sharded actors (actor Account sharded by id) work the same way, but one instance per key and spread over the nodes of a cluster - see the next step.

Common mistake: spawn outside a scope#

fn work() {}

fn main() {
  let h = spawn work()
}

spawn-outside-scope

spawn-outside-scope: "lost" tasks are forbidden. Wrap the start in scope { … }. If background work really has to live long (for example for the whole life of the node), there is the node's background scope node.background - an explicit choice that is visible in the code.

Remember#

• scope + spawn for parallel computations; all for a list of tasks; channels and select for streams of values; actor for state.
• An error of one task cancels the others; tasks do not outlive their scope.
• You need neither async nor mutexes: the effects time, io, net and actor messages themselves serve as suspension points.

Next: a small service.