Параллельная обработка: воркеры и потоки

Упаковка - это вычисления, поэтому ей выделены два собственных потока. Бронирование курьера ждёт сети, поэтому четыре бронирования идут одновременно.

По умолчанию подписка обрабатывает доставки по одной, в том порядке, в каком их передаёт брокер. Воркеры позволяют ей обрабатывать несколько доставок одновременно на рантайме приложения. Они подходят для обработчика, который ждёт ввода-вывода. Потоки выделяют подписке собственные потоки. Они подходят для обработчика, который ведёт вычисления: иначе он задерживал бы рантайм, который обслуживает все остальные подписки. И воркеры, и потоки - настройки подписки, тело обработчика при этом не меняется. Подписка с воркерами или потоками больше не сохраняет порядок доставок. workers(n, by_key) сохраняет порядок для каждого ключа.

Упаковка на потоках

pack разбивает оплаченный заказ на посылки и отвечает планом упаковки. Оплаченный заказ за границами магазина отбрасывается до того, как спланирована хоть одна посылка. threads(2) запускает вместе с подпиской два потока, и доставки распределяются между ними по очереди. Декодирование, тело обработчика, ответ и завершение доставки выполняются в потоке, который получил доставку.

Макросы

shipping.rs

/// Packs a paid order into parcels.
// Packing is computation: two threads of its own keep it off the app's runtime, which serves I/O.
#[subscriber("payments.settled", threads(2), reply)]
async fn pack(paid: &Paid) -> Result<Packed, HandlerOutcome> {
    if !within_bounds(&paid.lines) {
        tracing::warn!(
            order = paid.order,
            "a paid order past the shop's bounds, dropped"
        );
        return Err(HandlerOutcome::drop());
    }
    Ok(Packed {
        order: paid.order,
        customer: paid.customer.clone(),
        parcels: plan(&paid.lines),
    })
}

Вручную

shipping.rs

/// Packs a paid order into parcels.
struct Pack;

impl Handle<Paid, Packed, (), (), Shop> for Pack {
    fn handle(
        &self,
        paid: &Paid,
        _outs: &(),
        _ctx: &mut Context<'_, (), Shop>,
    ) -> impl Future<Output = Result<Packed, HandlerOutcome>> {
        if !within_bounds(&paid.lines) {
            tracing::warn!(
                order = paid.order,
                "a paid order past the shop's bounds, dropped"
            );
            return ready(Err(HandlerOutcome::drop()));
        }
        ready(Ok(Packed {
            order: paid.order,
            customer: paid.customer.clone(),
            parcels: plan(&paid.lines),
        }))
    }
}

Бронирование на воркерах

dispatch бронирует курьера для упакованного заказа и сообщает покупателю код отслеживания. workers(4) допускает до четырёх бронирований одновременно. Следующая доставка ждёт свободного воркера.

Макросы

shipping.rs

/// Books a courier for a packed order and tells the customer.
// Booking waits on the courier's API, so four bookings run at once.
#[subscriber("shipping.packed", workers(4))]
async fn dispatch(packed: &Packed, Out(notices): Out<impl Publisher, Notices>) -> HandlerOutcome {
    let tracking = book_courier(packed).await;
    let notice = Notice {
        customer: packed.customer.clone(),
        text: format!("Order {} is on its way: {tracking}.", packed.order),
    };
    match notices.message(&notice).publish().await {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

Вручную

shipping.rs

/// Books a courier for a packed order and tells the customer.
struct Dispatch;

impl<Entry> Handle<Packed, (), Outs<(Entry,)>, (), Shop> for Dispatch
where
    Entry: OutEntry<Notices, Wire: Publisher>,
{
    async fn handle(
        &self,
        packed: &Packed,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_, (), Shop>,
    ) -> Result<(), HandlerOutcome> {
        let tracking = book_courier(packed).await;
        let notice = Notice {
            customer: packed.customer.clone(),
            text: format!("Order {} is on its way: {tracking}.", packed.order),
        };
        let sent = outs.get(Notices).message(&notice).publish().await;
        sent.map_err(|_| HandlerOutcome::retry())
    }
}

Роутер

Роутер отгрузки читает внутренние темы магазина, поэтому декодирует CBOR. Ответ использует кодек по умолчанию, если его позиция не задаёт другой, поэтому ответ pack задаёт CBOR. Слот использует кодек роутера, если не задаёт свой, поэтому уведомления задают JSON. В ручной форме обе настройки - шаги цепочки монтирования: threads и workers.

Счётная часть работает на своих потоках, поэтому рантайм приложения только ждёт ввода-вывода, и магазин запускает его в потоке, который вызывает main: #[ruststream::app(flavor = "current_thread")], а в ручной форме run_main_on(AppRuntime::CurrentThread, app). Точка входа с HTTP из главы про HTTP делает то же через #[tokio::main(flavor = "current_thread")].

Макросы

shipping.rs

/// The shipping side's subscribers. Both read the shop's own topics, so both decode CBOR.
pub(crate) fn routes() -> impl RouterDef<MemoryBroker, Shop> {
    Router::new()
        .with_codec(INTERNAL)
        .include(pack)
        // A reply encodes with the default codec unless its position names one.
        .out_reply(Publish)
        .codec(INTERNAL)
        .build()
        .include(dispatch)
        // A slot encodes with the router's codec unless it names one; notices stay JSON.
        .out(Notices, Publish)
        .codec(JsonCodec)
        .build()
}

Вручную

shipping.rs

/// The shipping side's subscribers. Both read the shop's own topics, so both decode CBOR.
pub(crate) fn routes() -> impl RouterDef<MemoryBroker, Shop> {
    Router::new()
        .with_codec(INTERNAL)
        .include(
            subscriber(Paid::DESTINATION, Pack)
                .describe("Packs a paid order into parcels.")
                // Packing is computation: two threads of its own keep it off the app's runtime.
                .threads(nonzero!(2))
                .reply()
                .build(),
        )
        // A reply encodes with the default codec unless its position names one.
        .out_reply(Publish)
        .codec(INTERNAL)
        .build()
        .include(
            subscriber(Packed::DESTINATION, Dispatch)
                .describe("Books a courier for a packed order and tells the customer.")
                // Booking waits on the courier's API, so four bookings run at once.
                .workers(nonzero!(4))
                .build(),
        )
        // A slot encodes with the router's codec unless it names one; notices stay JSON.
        .out(Notices, Publish)
        .codec(JsonCodec)
        .build()
}

Потоки окупаются, когда вычисления занимают от десяти микросекунд на доставку. На более лёгком обработчике они обходятся дороже, чем экономят. Диагностика опроса находит обработчики, которые задерживают рантайм.