Конкурентность
Единица исполнения — лёгкая задача. Приостанавливаемые функции (time, net, io…) отдают поток в точках приостановки, а
пул потоков ОС распределяет задачи между ядрами. Нет async и «цвета» функций: любую функцию можно вызвать как обычную,
и она приостановится, если нужно (§8.1).
Структурная конкурентность: scope и spawn#
spawn допустим только внутри scope; выход из scope ждёт всех детей, а ошибка одного отменяет остальных (§8.5).
fn compute(n: i32) -> i32 time {
sleep(10ms)
n * n
}
fn main() io, time {
let (a, b) = scope {
let ha = spawn compute(3)
let hb = spawn compute(4)
(ha.join()!, hb.join()!)
}
print("{a} {b}")
}9 16
all, race, any#
Task.of { … } откладывает вычисление; all запускает задачи параллельно, сохраняет порядок результатов и отменяет остальные
при первой ошибке (§8.3).
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("x", 100ms) }, Task.of { job("y", 5ms) }])
for r in rs { print(r) }
}done y done x x y
Цепочки и дедлайны#
-> соединяет шаги, кортеж даёт параллельное разветвление, !> ловит ошибку, within ограничивает всю цепочку по времени
(§8.4).
const PROFILE_DEADLINE = 300ms
type UserId = distinct i64
@error data NotFound { id: i64 }
data User { id: UserId; name: str }
data Order { total: i64 }
data Page { orders: List<Order>; recs: List<str> }
impl Page { static fn empty() -> Page => Page{ orders: [], recs: [] } }
fn getUser(id: UserId) -> User | NotFound | Timeout time {
sleep(5ms)
if id.raw == 0 { fail NotFound{ id: 0 } }
User{ id, name: "ann" }
}
fn recentFor(id: UserId) -> List<Order> | Timeout time {
sleep(20ms)
[Order{ total: 100 }, Order{ total: 250 }]
}
fn forUser(id: UserId) -> List<str> | Timeout time {
sleep(if id.raw == 2 { 2s } else { 10ms })
["book", "pen"]
}
fn profilePage(id: UserId) -> Page | Timeout time {
await getUser(id)
-> (recentFor(it.id), forUser(it.id)) // два запроса параллельно
-> (orders, recs) => Page{ orders, recs }
!> NotFound => Page.empty() // нет пользователя — пустая страница
within PROFILE_DEADLINE // превышение даёт Timeout: он остаётся в типе результата
}
fn main() io, time {
match profilePage(UserId(1)) {
Page p => print("orders {p.orders.len()} recs {p.recs.len()}")
Timeout => print("timeout")
}
match profilePage(UserId(2)) { // forUser спит 2 с — дольше дедлайна
Page p => print("orders {p.orders.len()} recs {p.recs.len()}")
Timeout => print("timeout")
}
match profilePage(UserId(0)) {
Page p => print("empty page: {p.orders.len()}")
Timeout => print("timeout")
}
}orders 2 recs 2 timeout empty page: 0
Каналы и select#
chan<T>(cap: n) создаёт пару Sender/Receiver. select ждёт первую готовую ветку, after d — ветка по таймауту
(§8.7).
fn main() io, time {
let (tx, rx) = chan<i32>(cap: 1)
let r = select {
recv i32 v from rx => "got {v}"
after 50ms => "timeout"
}
print(r)
_ = tx.send(1)
print("{rx.recv().ok()}")
}timeout 1
STM#
Общее изменяемое состояние без блокировок: TVar и atomic { … } — сериализуемая транзакция, которая перезапускается при
конфликте; check(cond) ждёт изменения прочитанных переменных (§8.9). Тело atomic — pure кроме операций TVar.
use std.stm { TVar, check }
static A: TVar<i32> = TVar(100)
static B: TVar<i32> = TVar(0)
fn transfer(amt: i32) {
atomic {
let a = A.get()
check(a >= amt) // иначе транзакция ждёт, пока A изменится
A.set(a - amt)
B.update(x => x + amt)
}
}
fn main() io {
transfer(30)
let a = atomic { A.get() }
let b = atomic { B.get() }
print("{a} {b}")
}70 30