Шаг 11. Конкурентность

Зачем#

Конкурентные программы ломаются там, где задача «потерялась»: ушла в фон, упала, а никто не заметил, или продолжает работать, когда её результат уже не нужен. В OxL задачи структурны: задача, запущенная в блоке scope, не может пережить этот блок, а ошибка одной задачи отменяет остальные. Нет ключевых слов async/await и «окрашивания» функций: любую приостанавливаемую функцию можно вызвать обычным образом, и она уступит поток планировщику, когда ей придётся ждать.

Три инструмента на разные случаи:

scope и spawn#

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

fn main() io, time {
  let (a, b) = scope {  // scope ждёт всех детей
    let ha = spawn compute(3)  // spawn — только внутри scope
    let hb = spawn compute(4)
    (ha.join()!, hb.join()!)
  }
  print("{a} {b}")
}

9 16

• scope { … } — блок, который ждёт всех запущенных внутри spawn. Значение блока — его последнее выражение.
• spawn compute(3) запускает вызов как отдельную задачу и возвращает дескриптор; ha.join() ждёт результат.
• Обе задачи спят по 10 мс одновременно, поэтому весь блок занимает примерно 10 мс, а не 20.

all: параллельно с сохранением порядка#

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: параллельно, порядок результатов сохраняется
  for r in rs { print(r) }
}

done fast
done slow
slow
fast

Task.of { … } откладывает вызов, а all запускает все задачи сразу. Результаты возвращаются в том порядке, в котором задачи перечислены (а не в порядке завершения: fast напечатал «done» раньше, но в списке стоит вторым). Если одна из задач завершилась ошибкой, остальные отменяются. Связанные race и any возвращают первый результат или первый успех.

Каналы и 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)  // канал с буфером на 2 значения
  scope {
    spawn producer(tx)
    var got = 0
    while got < 3 {
      let r = select {  // select берёт первую готовую ветку
        recv i32 v from rx => "got {v}"
        after 500ms => "timeout"  // ветка таймаута
      }
      print(r)
      got += 1
    }
  }
}

got 10
got 20
got 30

• chan<i32>(cap: 2) создаёт пару (Sender, Receiver) с буфером на 2 значения; send ждёт, если буфер полон — так получается естественное «давление» назад на быстрого производителя.
• select { … } ждёт первую готовую ветку: recv … from rx, send, или after 500ms — таймаут. Без ветки after и при закрытом канале select не зависнет вечно.
• Значения, передаваемые между задачами, должны быть «передаваемыми»: без Rc и без заимствований (§7.3). Это правило проверяет компилятор, поэтому гонок данных в безопасном коде нет.

Акторы#

actor Counter {
  var n: i32 = 0  // состояние актора доступно только из его msg
  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}")  // сообщения обрабатываются по одному, по порядку
}

5

actor владеет своим состоянием var, и менять его могут только его msg-методы. Сообщения обрабатываются по одному в порядке поступления, поэтому блокировки не нужны. spawn Counter() создаёт актор, c.inc(2) -> c.inc(3) -> c.get() -> print("{it}") — цепочка, где каждое следующее сообщение отправляется после ответа на предыдущее. Шардированные акторы (actor Account sharded by id) работают так же, но по экземпляру на ключ и распределены по узлам кластера — см. следующий шаг.

Типичная ошибка: spawn вне scope#

fn work() {}

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

spawn-outside-scope

spawn-outside-scope: «потерянные» задачи запрещены. Оберните запуск в scope { … }. Если фоновая работа действительно должна жить долго (например, на всё время жизни узла), для этого есть фоновый scope узла node.background — явный и заметный в коде выбор.

Что запомнить#

• scope + spawn — для параллельных вычислений; all — для списка задач; каналы и select — для потоков значений; actor — для состояния.
• Ошибка одной задачи отменяет остальные; задачи не переживают свой scope.
• Не нужно ни async, ни мьютексов: эффекты time, io, net и сообщения акторов сами служат точками приостановки.

Дальше: маленький сервис.