Потоки и каналы

К параллельному коду kola подходит так же, как к ресурсам и эффектам: правило, которое иначе пришлось бы держать в голове, проверяет компилятор. Здесь это правило о разделяемой изменяемой памяти, той самой, которую обычно охраняет блокировка. В kola такой памяти нет, поэтому гонку данных невозможно даже написать. Средства для параллельной работы — потоки и каналы.

В чём проблема

Гонка данных возникает, когда два потока обращаются к одной и той же памяти, никак не упорядочив обращения, и хотя бы один из них пишет. Последствия бывают любыми: от совершенно незаметных до испорченного указателя. Большинство языков полагаются на дисциплину. Данные защищают мьютексом (блокировкой, которую в каждый момент может держать только один поток), захватывают его и отпускают, помнят, какая блокировка что охраняет, и захватывают блокировки всегда в одном и том же порядке. Никто не проверяет, что всё это не забыто.

У kola есть и своя причина этим заниматься. Внутри одного региона две ссылки могут указывать на одно и то же значение, и писать можно через обе: правила уникальности для &r mut, как в Rust, нет. Это безопасно, потому что ссылка всегда указывает на живое значение правильного типа. Но стоит наивно добавить потоки, и безопасность пропадает: два потока с одной и той же &r mut T — это уже гонка данных. Поэтому kola нужно правило, по которому каждый регион достижим только из одного потока.

Не делить, а перемещать

Правило (SPEC-CARD §11) такое: ссылки никогда не пересекают границу потока, а владение перемещается через контексты spawn и через каналы. Поток, которому нужны данные другого потока, не заимствует их, а получает, и у отправителя их больше нет. Это передача сообщений, и проверяют её регионы (в тип ссылки входит регион, который открыл один-единственный поток) и линейность (всё, что пересекает границу, расходуется на стороне отправителя).

Поэтому в обычном коде нет ни Mutex, ни блокировок, ни упорядочивания памяти. (Атомарные операции в std есть, но они нужны для работы с железом и для голого железа под эффектом hw, а не для координации потоков kola.) Нет и async/await, нет future-объектов. Логику протокола, которой приходится вести множество незавершённых операций в одном потоке, пишут как явный конечный автомат step(event, now) -> intents, без скрытых точек приостановки.

spawn, Thread[R] и join

sys.spawn(entry, ctx) запускает поток, который выполняет entry(ctx), и возвращает Thread[R], где R — то, что возвращает entry. entry — обычная функция верхнего уровня, а не замыкание (замыканий в kola нет), поэтому её ряд эффектов виден в месте вызова. В ctx не должно быть ссылок: простые данные копируются, а ресурсы и владеющие Vec перемещаются.

use std::log;

fn square(n: u64) -> u64 !{} {
    n *% n
}

fn main(mut sys: Sys) -> u8 !{io, block, alloc(heap), log} {
    let t = sys.spawn(square, 7u64);
    let result = t.join();
    log::info(f"7 squared is {result}");
    0
}

Thread[R] линеен, как файл: его нужно дождаться через join на каждом пути, включая ранние возвраты. Если о нём забыть, компилятор ещё до запуска программы выдаст ту же диагностику, что и для утёкшего файла. Запущенный поток получает стек в 8 МиБ на любой платформе, как и главный.

Каналы: chan, Tx[T] и Rx[T]

sys.chan(cap) создаёт канал ограниченной ёмкости и возвращает (Tx[T], Rx[T]). Оба конца линейны, поэтому каждый из них либо закрывают (tx.close(), rx.close()), либо перемещают в поток, который его закроет. Тип сообщений выводится из того, как используются концы, но его можно написать и явно: let (tx, rx): (Tx[u64], Rx[u64]) = sys.chan(4);.

tx.send(v) блокируется, пока очередь заполнена. Так обратное давление (backpressure) получается по умолчанию: быстрый производитель замедляется до темпа своего потребителя, а не забивает память. Результат send имеет тип Result[(), SendErr[T]]. Если получателя уже нет, ответ будет Err(SendErr::Closed(v)), где v — отправленное вами значение, которое вернулось к вам. Для данных это почти ничего не значит. Для ресурса это очень важно: недоставленный Listener или File снова ваш, и израсходовать его придётся вам (Err(SendErr::Closed(l)) => { l.close(); }), потому что деструктора, который сделал бы это за вас, нет. По той же причине let _ = tx.send(..) на канале ресурсов отвергается.

rx.recv(timeout_ms) ждёт не дольше timeout_ms миллисекунд и возвращает Recv[T]: Msg(v), если пришло сообщение, Timeout, если время ожидания истекло, а очередь пуста, или Closed, если все отправители ушли. Большинство языков сливают два последних случая в одно двусмысленное «ничего». Именно потому, что они разделены, цикл может отличить «пока нет» от «больше никогда». Recv и SendErr — это pub enum из std::threads.

Общей коробки, в которую смотрели бы оба потока, нет. Всё, что проходит через канал, копируется или перемещается в момент send. Владеющий Vec проходит перемещением на любой глубине внутри сообщения: у буфера меняется владелец, а сами байты не копируются, так что передать другому потоку большой кадр стоит столько же, сколько маленький. Пока представление этого буфера ещё используется, передачу компилятор отвергнет: следующий владелец может переписать эти байты.

Рабочий поток и конвейер

В этой программе две типичные схемы: рабочий поток, который возводит число в квадрат и сообщает результат через join, и конвейер из двух потоков на ограниченном канале, где один поток производит пять квадратов, а другой их суммирует.

Присмотритесь к циклу потребителя: в нём заключено самое важное правило о каналах. Цикл, который дочитывает канал, заканчивается на Closed и никогда не заканчивается на Timeout. Closed — доказанный конец передачи. Каждый Tx линеен, поэтому число живых отправителей известно точно, и канал сообщает Closed, когда уходит последний из них. Timeout говорит лишь о том, что одно ожидание истекло при пустой очереди: отправитель медлит, но ещё не закончил.

Цикл, который останавливается на Timeout, бросает все сообщения, которые ещё в пути. Он возвращает меньшее число, программа завершается с кодом 0, и никакой трап не срабатывает. Под нагрузкой, а ожидания истекают именно тогда, это тихая потеря данных, причём код выглядит так, будто обработал все случаи. При дочитывании тайм-аут задаёт, как часто цикл просыпается и смотрит снова, а не служит поводом остановиться.

use std::threads;
use std::log;

fn square(n: u64) -> u64 !{} {
    n *% n
}

// Производитель: отправляет пять квадратов и закрывает свой конец.
fn produce(tx: Tx[u64]) -> () !{io, block} {
    let mut t = tx;
    let mut i: u64 = 0;
    while i < 5u64 {
        let _r = t.send(i *% i);
        i = i +% 1;
    }
    t.close();
}

// Потребитель: читает, пока отправитель не закроет канал. Тайм-аут — не конец.
fn sum_channel(rx: Rx[u64]) -> u64 !{io, block} {
    let mut r = rx;
    let mut total: u64 = 0;
    let mut done = false;
    while !done {
        match r.recv(100u64) {
            Recv::Msg(v) => { total = total +% v; }
            Recv::Timeout => {}
            Recv::Closed => { done = true; }
        }
    }
    r.close();
    total
}

fn main(mut sys: Sys) -> u8 !{io, block, alloc(heap), log} {
    let t = sys.spawn(square, 7u64);
    let result = t.join();
    log::info(f"7 squared is {result}");

    // Ограниченный конвейер: один поток-производитель, один поток-потребитель.
    let (tx, rx) = sys.chan(4);
    let prod = sys.spawn(produce, tx);
    let cons = sys.spawn(sum_channel, rx);
    prod.join();
    let total = cons.join();
    log::info(f"sum of squares = {total}");
    0
}

test spawn_join_returns_result(mut sim: Sim) {
    let t = sim.spawn(square, 7u64);
    assert t.join() == 49u64;
}

test pipeline_sums_squares(mut sim: Sim) {
    let (tx, rx) = sim.chan(4);
    let prod = sim.spawn(produce, tx);
    let total = sum_channel(rx);
    prod.join();
    assert total == 30u64;
}

kola run печатает [info] 7 squared is 49 и [info] sum of squares = 30. Логике «сдаться через N миллисекунд» место в функции, результат которой может сообщить, что она сдалась; у цикла дочитывания такого результата нет. Тому же правилу следуют помощники из std: threads::drain_into, threads::actor_loop и threads::actor_loop_in.

Завершение с непустой очередью: rx.drain()

Иногда получателю нужно остановиться раньше отправителей: сервис завершает работу или пул рабочих потоков меняет размер. Если в очереди лежат простые данные, rx.close() их отбрасывает. Если там ресурсы (задания, соединения, открытые файлы), отбросить их значило бы допустить утечку, поэтому, когда в очереди остался ресурс, на rx.close() срабатывает трап с сообщением «rx.close() would drop 1 queued resource message(s); take them back with rx.drain() first».

Выход из положения — rx.drain(). Этот вызов закрывает канал для отправителей, так что любой последующий send отвечает Err(SendErr::Closed(v)) и возвращает значение своему отправителю. Затем он отдаёт одно сообщение из очереди вместе с получателем, Some((msg, rx)), или None, если очередь пуста, и тогда получатель израсходован. Вызывают его в цикле, пока он не ответит None:

resource struct Job { id: u64 }

fn finish(j: Job) -> u64 !{} {
    let Job { id } = j;
    id
}

// Отправить задание. Если получателя нет, задание возвращается и завершается здесь.
fn submit(tx: &mut Tx[Job], j: Job) -> bool !{io, block} {
    match tx.send(j) {
        Ok(_) => true,
        Err(SendErr::Closed(back)) => { let _ = finish(back); false }
    }
}

// Владелец останавливается раньше. Всё, что ещё в очереди, возвращается к нему.
fn shut_down(rx: Rx[Job]) -> u64 !{io, block} {
    let mut rx = rx;
    let mut handed_back: u64 = 0;
    loop {
        match rx.drain() {
            Some((job, next)) => {
                handed_back = handed_back +% finish(job);
                rx = next;
            }
            None => { break; }
        }
    }
    handed_back
}

test queued_jobs_are_handed_back(mut sim: Sim) {
    let (tx, rx) = sim.chan(4);
    let mut tx = tx;
    let a = submit(&mut tx, Job { id: 1 });
    let b = submit(&mut tx, Job { id: 2 });
    let back = shut_down(rx);
    let c = submit(&mut tx, Job { id: 3 });  // для отправителей канал уже закрыт
    assert a && b;
    assert back == 3;                        // оба задания из очереди вернулись
    assert !c;
    tx.close();
}

Чекер прослеживает этот цикл: после loop состояние каждой переменной получается слиянием её состояний во всех break, так же как сливаются две ветки if. Здесь единственный break стоит в ветке None, где получатель уже израсходован, поэтому к концу цикла с rx покончено. drain работает в интерпретаторе и в C-бэкенде; собственный бэкенд его пока не поддерживает и прямо об этом сообщает.

Сначала один цикл

Прежде чем браться за что-то из описанного ниже, полезно знать, как обычно выглядит сервис на kola: это один цикл обслуживания. Периодическая работа вроде пульса (heartbeat), продления аренды или сброса статистики не получает собственного потока и таймера. Это тик с ограничением частоты внутри того же цикла: в начале каждой итерации показание часов сравнивается с дедлайном, и если срок подошёл, работа выполняется, а если нет, цикл переходит к следующей единице настоящей работы. С этой формы и стоит начинать, а к почтовому ящику, описанному ниже, переходят, когда её перестаёт хватать.

Одному циклу, который работает только со своим состоянием, блокировки не нужны, так что ошибок порядка захвата блокировок в нём не написать вовсе. Порядок выполнения детерминирован: ошибка каждый раз воспроизводится одинаково, и kola report checks рассуждает об одном теле функции, а не о чередовании потоков.

// Периодическая работа — тик с ограничением частоты внутри цикла
// обслуживания, а не второй поток с таймером. Цикл владеет всем, что трогает.
struct State { served: u64, ticks: u64, next_tick_ns: u64 }

fn tick(s: State) -> State !{} {
    State { served: s.served, ticks: s.ticks +% 1u64, next_tick_ns: s.next_tick_ns }
}

fn serve_loop(clock: &Clock, rx: Rx[u64], period_ns: u64) -> State !{io, block} {
    let mut r = rx;
    let mut st = State { served: 0u64, ticks: 0u64, next_tick_ns: 0u64 };
    let mut running = true;
    while running {
        // 1. периодическая работа в начале итерации, если подошёл её срок
        let now = clock.now();
        if now >= st.next_tick_ns {
            st = tick(st);
            st.next_tick_ns = now +% period_ns;
        }
        // 2. одна единица настоящей работы с ограниченным ожиданием
        match r.recv(50u64) {
            Recv::Msg(v) => { st.served = st.served +% v; }
            Recv::Timeout => {}
            Recv::Closed => { running = false; }
        }
    }
    r.close();
    st
}

fn feed(tx: Tx[u64]) -> () !{io, block} {
    let mut t = tx;
    let _a = t.send(3u64);
    let _b = t.send(4u64);
    t.close();
}

test one_loop_serves_and_ticks(mut sim: Sim) {
    let (tx, rx) = sim.chan(2);
    let feeder = sim.spawn(feed, tx);
    let st = serve_loop(&sim.clock, rx, 1000000u64);
    feeder.join();
    assert st.served == 7u64;
    assert st.ticks >= 1u64;
}

Цена в том, что один цикл выполняет всё последовательно. Одна медленная единица работы задерживает всё, что стоит за ней в очереди, и блокироваться на неограниченном ожидании цикл не должен никогда. Если каждая операция в нём ограничена по времени, он остаётся справедливым без всякого планировщика. Переходить к почтовому ящику стоит, когда ресурсом должен владеть один поток, а обращаются к нему несколько производителей, или когда одна единица работы так длинна, что надолго отнимает цикл у всех остальных.

Много отправителей, один владелец: tx.split и почтовый ящик

Если N рабочим потокам нужен один ресурс, а разделяемую изменяемую память язык не допускает, ресурсом владеет один поток, а остальные обращаются к нему сообщениями. Для этого нужен почтовый ящик: много отправителей и один получатель. Его строит tx.split() -> (Tx[T], Tx[T]): вызов расходует одного отправителя и возвращает двух.

Каждый Tx остаётся линейным, поэтому число отправителей видно в исходном тексте: по одному split на рабочий поток, прямо там, где рабочие потоки запускаются. Когда закрывается последний отправитель, канал сообщает владельцу Closed. Чекер доказывает, что каждый Tx закрывается ровно один раз на каждом пути, поэтому владелец не может ни остановиться слишком рано (пока какой-то отправитель ещё жив), ни ждать вечно (Closed, который никогда не придёт). У клонируемого отправителя есть счётчик ссылок во время выполнения, у линейного — доказательство.

use std::threads;

// Общий ресурс: счётчик слотов. Простые данные, которыми владеет один поток.
// Ничего атомарного и никаких блокировок: больше никто до него не дотянется.
struct Slots { free: u64, handed: u64 }

// Один запрос, свёрнутый в состояние владельца; передаётся в actor_loop по имени.
fn serve(s: Slots, q: u64) -> Slots !{io, block} {
    if q == 1u64 {
        if s.free > 0u64 {
            return Slots { free: s.free -% 1u64, handed: s.handed +% 1u64 };
        }
    }
    s
}

// Всё тело владельца — помощник из std: принимать, пока не закроется
// последний отправитель, затем вернуть состояние.
fn pool_owner(rx: Rx[u64]) -> u64 !{io, block, log} {
    let end = threads::actor_loop[Slots, u64](rx, Slots { free: 2u64, handed: 0u64 }, serve, 200u64);
    end.handed
}

// Рабочий поток держит ровно свою долю почтового ящика и никакого Sys, поэтому
// не может открыть ни канал, ни сокет, ни файл. Всё, что ему выдано, видно в месте spawn.
fn worker(tx: Tx[u64]) -> u64 !{io, block} {
    let mut t = tx;
    let _sent = t.send(1u64);
    t.close();
    1u64
}

test two_workers_one_owner(mut sim: Sim) {
    let (tx, rx) = sim.chan(4);
    let owner = sim.spawn(pool_owner, rx);

    // по одному split на рабочий поток: число отправителей записано явно
    let (tx, t1) = tx.split();
    let w1 = sim.spawn(worker, t1);
    let w2 = sim.spawn(worker, tx);   // последняя доля

    let a = w1.join();
    let b = w2.join();
    let handed = owner.join();

    assert a == 1u64;
    assert b == 1u64;
    assert handed == 2u64;
}

Если рабочему потоку нужен собственный ответ, запрос несёт с собой обратный адрес: resource struct Ask { q: u64, reply: Tx[u64] }, то есть канал для ответа лежит прямо в сообщении. Тогда каждый ответ идёт от точки к точке, без идентификаторов корреляции и без общей очереди ответов.

Ожидание почтового ящика или сокета

Демону нужно реагировать и на входящие соединения, и на сообщения от рабочих потоков, не опрашивая их по очереди. Poller — механизм ожидания готовности в kola, аналог epoll или select в C: в нём регистрируют несколько источников, а потом ждут, пока какой-нибудь из них не будет готов. poller.add_chan(&rx) регистрирует принимающий конец канала в том же Poller, что и сокеты (сокетная половина описана в главе Сеть). Один вызов wait спит, пока не будет готов сокет, не появится сообщение в очереди или не закроется последний отправитель, а next_ready сообщает, что именно произошло. Готовый почтовый ящик опустошают без блокировки, вызовом rx.recv(0).

Нулевой тайм-аут — единственное законное исключение из правила дочитывания, о котором шла речь выше. При тайм-ауте 0 никакого ожидания не было, поэтому Timeout не может означать «отправитель медлит». Он означает «прямо сейчас в очереди ничего нет»: эта порция закончена, и управление возвращается к wait. Передача по-прежнему заканчивается там же, где всегда, на Closed. Точно так же нормален одиночный recv вне цикла с настоящим тайм-аутом («нет ответа за 100 мс, сдаёмся»): остатка передачи, который можно было бы потерять, у него нет.

use std::net;
use std::threads;

fn producer(tx: Tx[u64]) -> () !{io, block} {
    let mut t = tx;
    let mut i: u64 = 0;
    while i < 3u64 {
        let _r = t.send(i +% 100u64);
        i = i +% 1;
    }
    t.close();
}

test select_over_mailbox(mut sim: Sim) {
    let (tx, rx) = sim.chan(2);
    let prod = sim.spawn(producer, tx);
    let mut rxm = rx;
    let mut poller = Poller::new();
    let mid = poller.add_chan(&rxm).unwrap();

    let mut total: u64 = 0;
    let mut running = true;
    while running {
        let _n = poller.wait(500u64).unwrap();
        match poller.next_ready() {
            Some(id) => {
                assert id == mid;
                // почтовый ящик готов: опустошаем его без блокировки
                match rxm.recv(0u64) {
                    Recv::Msg(v) => { total = total +% v; }
                    Recv::Timeout => {}
                    Recv::Closed => { running = false; }
                }
            }
            None => { running = false; }
        }
    }
    assert total == 303u64;

    let _p = poller.close();
    rxm.close();
    prod.join();
}

Как спать, не мешая остальным

clock.sleep(ms) — единственная операция Clock, которая ждёт, поэтому её ряд — !{io, block}. Как и любая блокирующая операция капабилити, она уступает управление другому потоку, так что спящий поток никогда не мешает работать соседям. С симулированными часами сон продвигает виртуальное время без настоящего ожидания, и проверка пути с тайм-аутом ничего не стоит. Freestanding-сборка отвергает sleep: планировщика, которому можно было бы уступить управление, там нет.

fn nap_then_report(clock: &Clock, ms: u64) -> u64 !{io, block} {
    let t0 = clock.now();
    clock.sleep(ms);
    let t1 = clock.now();
    t1 -% t0
}

test the_sim_clock_advances_virtually(mut sim: Sim) {
    // 10 мс в наносекундах плюс 1000 нс за второе чтение:
    // каждое чтение симулированных часов сдвигает их вперёд.
    let elapsed = nap_then_report(&sim.clock, 10u64);
    assert elapsed == 10001000u64;
}

Тайм-аут ограничивает одно ожидание, а не цикл из них. Собеседник, который каждый раз отвечает чуть раньше истечения тайм-аута, может держать цикл вызовов recv(100) вечно. Настоящий дедлайн — это одно показание clock.now(), снятое до цикла и сравниваемое на каждом шаге, тот же приём, что и тик в цикле обслуживания.

Когда компилятор отказывает

Компилятор следит за обеими гарантиями. Ссылку нельзя передать в качестве контекста spawn:

fn work[r](b: &r [u8]) -> usize !{} { b.len }

fn main(mut sys: Sys) -> u8 !{io, block} {
    region r {
        let b = alloc in r [1u8, 2u8, 3u8];
        let t = sys.spawn(work, b);
        let _n = t.join();
    }
    0
}

error[card §11]: a reference cannot cross a thread boundary in a spawn context
  --> r1.kl:6:27
   |
 6 |         let t = sys.spawn(work, b);
   |                           ^^^^
   = help: move owned data or transferable resources; InterruptGuard holds its installing thread and must be restored there

Pool тоже не может пересечь эту границу, причём по причине, которой его тип не показывает: пул простых данных копируем, так что, перемещённый в контекст spawn, он оставил бы обоим потокам одни и те же слоты. Пулом владеет один поток, а остальные обращаются к нему сообщениями.

Поток, который запустили и не дождались, — неизрасходованный ресурс:

fn work(n: u64) -> u64 !{} { n }

fn main(mut sys: Sys) -> u8 !{io, block} {
    let t = sys.spawn(work, 1u64);
    0
}

error[card §2]: resource `t` is not consumed on this path
  --> r2.kl:4:5
   |
 4 |     let t = sys.spawn(work, 1u64);
   |     ^^^
   = help: no implicit destructors: move it on, return it, destructure it, or pass it to a consuming fn

Это та же диагностика, что и для любого неизрасходованного ресурса (Ресурсы и линейность). Option[Thread[R]] со значением None ничего не должен, так что объявить слот заранее и заполнить его позже — обычный код; обязательство несёт Some. То же верно и на уровень выше: File, перемещённый в рабочий поток (Файловая система), или Conn, отданный потоку соединения (Сеть), передаются перемещением, а не заимствованием.

Детерминизм и что выполняется параллельно

Память никогда не разделяется, поэтому расписание потоков проявляется только в операциях с каналами, в spawn и в join, а это небольшое пространство, которое можно воспроизвести. Интерпретатор (и в kola run, и в kola test) выполняет потоки кооперативно, по одному, переключаясь только на этих операциях. Управляет ими детерминированный планировщик с зерном, и один и тот же --seed N каждый раз воспроизводит одну и ту же трассу. С симулированными часами у каждого ограниченного ожидания (recv, wait_readable, clock.sleep) есть абсолютный срок в виртуальном времени. Первым срабатывает самый ранний, и виртуальное время прыгает прямо к нему, так что 30-секундное ожидание рядом с 700-секундным сном соседнего потока длится 30 секунд. При равных сроках первым идёт поток с меньшим идентификатором. Безуспешное ожидание, например recv(0), ответивший Timeout, уступает управление, прежде чем вызывающий повторит попытку, поэтому опрашивающий цикл не может лишить соседей процессорного времени.

Нормативное описание — SPEC-CARD §11 (kola help spec-card).