Паттерны конкурентности

Пул воркеров#

Задача. Раздать список заданий нескольким воркерам и собрать результаты.

oxl✓ проверяется в 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() {  // цикл идёт, пока открыт хотя бы один отправитель
    _ = 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)  // у каждого воркера свой канал заданий
      senders.push(tx)
      _ = spawn worker(rx, resTx.clone())  // Sender можно клонировать, Receiver — нет
    }
    var i = 0
    for n in 1..=6 {
      _ = senders[i % 3].send(n)  // раздаём задания по кругу
      i += 1
    }
    drop(senders)  // закрыли каналы заданий — воркеры завершатся
    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}") }
}

Вывод

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

У каждого воркера свой канал заданий: Receiver — единственный потребитель, его нельзя клонировать, зато Sender клонируется — все воркеры пишут результаты в один канал через resTx.clone().
Закрытие канала — это уничтожение всех его отправителей: drop(senders) закрывает каналы заданий, воркеры выходят из while let n? = jobs.recv().ok(); drop(resTx) отпускает оригинал, и цикл чтения результатов завершается, когда ушёл последний клон.
Результаты приходят в произвольном порядке, поэтому в примере они складываются в Map и выводятся по отсортированным ключам.
scope ждёт всех воркеров: после него гарантированно нет «осиротевших» задач.

Конвейер#

Задача. Обработать поток значений последовательными стадиями, каждая — в своей задаче.

oxl✓ проверяется в 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 {  // стадия: читает, преобразует, пишет дальше
  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 переехал в gen: когда gen кончит, канал закроется
    _ = spawn double(aRx, bTx)
    _ = spawn addTen(bRx, cTx)
    while let n? = cRx.recv().ok() { out.push(n) }
  }
  print("{out}")
}

Вывод

[12, 14, 16, 18, 20]

Каждая стадия — функция «канал на входе → канал на выходе». Размер буфера (cap: 2) ограничивает опережение: быстрая стадия ждёт медленную, память не растёт.
Отправитель переезжает в функцию-стадию и закрывается, когда она завершается, — этот сигнал «конец данных» каскадом проходит по всему конвейеру.
Порядок сохраняется, потому что у каждой стадии один поток обработки; для параллелизма внутри стадии используйте пул воркеров.

Таймаут#

Задача. Подождать результат не дольше заданного времени.

oxl✓ проверяется в 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"  // ждём не дольше 50 мс
    }
  }
}

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

Вывод

found after 10ms
timed out

select ждёт первое готовое событие: значение из канала или after 50ms. Медленная задача после таймаута продолжит работу до конца scope; для отмены используйте дескриптор spawn (handle.cancel()).
Для цепочек вызовов есть более короткая форма — … within 300ms (тур), она отменяет и вложенные вызовы.

Общее состояние: STM#

Задача. Переводить деньги между счётами из нескольких задач без блокировок и гонок.

oxl✓ проверяется в CI

use std.stm { TVar, check }

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

fn transfer(amount: i64) {
  atomic {  // транзакция: всё или ничего, перезапуск при конфликте
    let have = FROM.get()
    check(have >= amount)  // нет денег — ждём, пока FROM изменится
    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}")
}

Вывод

from 50 to 50 total 100

atomic { … } — транзакция: либо видны все изменения, либо ни одного; при конфликте с другой транзакцией она перезапускается, поэтому тело atomic не должно иметь побочных эффектов, кроме операций над TVar.
check(have >= amount) приостанавливает транзакцию, пока одна из прочитанных переменных не изменится, — так реализуется ожидание условия без опроса.
Инвариант «сумма постоянна» (50 + 50) выполняется при любом чередовании пяти параллельных переводов.

Потоки#

Задача. Отдавать значения получателю по мере готовности.

oxl✓ проверяется в CI

fn countdown(from: i32) -> stream<i32> {  // stream — асинхронная последовательность
  for i in 0..from {
    yield from - i  // yield отдаёт очередное значение подписчику
    sleep(1ms)
  }
}

fn main() io, time {
  var seen: List<i32> = []
  for await n in countdown(4) {  // for await читает поток по мере поступления
    seen.push(n)
  }
  print("{seen}")
}

Вывод

[4, 3, 2, 1]

Функция с результатом stream<T> может использовать yield; значение отдаётся получателю сразу, а функция продолжает работу после yield.
for await x in stream { … } читает поток до конца; потоки поддерживают операторы std.stream (map, filter, batch, merge…) и соединяются через |> — см. пример streams.

Вернуться: кулинарная книга.

Паттерны конкурентности

Пул воркеров#

Конвейер#

Таймаут#

Общее состояние: STM#

Потоки#

См. также