Skip to main content

Concurrency: workers and threads

Packing computes, so it gets two threads of its own; booking a courier waits on the network, so four bookings run at once.

A subscription handles one delivery at a time by default, in the order the broker hands them over. Workers let it handle several at once on the app's runtime: the right tool for a handler that waits on I/O. Threads give it threads of its own: the right tool for a handler that computes, since it would otherwise hold up the runtime that serves every other subscription. Both are settings of the subscription, and the handler's body does not change. A subscription with workers or threads no longer keeps its deliveries in order; workers(n, by_key) keeps the order of each key.

Packing on threads​

pack splits a paid order into parcels and answers with the plan. A paid order past the shop's bounds is dropped before any parcel is planned. threads(2) starts two threads with the subscription; the deliveries go to them in turn. Decoding, the body, the reply and the settlement all run on the thread that got the delivery.

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),
    })
}

Booking on workers​

dispatch books a courier for a packed order and tells the customer the tracking code. workers(4) keeps up to four bookings in flight; the next delivery waits for a free worker.

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(),
    }
}

The router​

The shipping router reads the shop's own topics, so it decodes CBOR. A reply takes the default codec unless its position names one, so pack's reply names CBOR. A slot takes the router's codec unless it names one, so the notices name JSON. In the manual form the two settings are steps of the mount chain: threads and workers.

With the computing step on threads of its own, the app's runtime only waits on I/O, so the shop runs it on the thread that calls main: #[ruststream::app(flavor = "current_thread")], or run_main_on(AppRuntime::CurrentThread, app) in the manual form. The HTTP entry point of a later chapter does the same with #[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()
}

Threads pay off from about ten microseconds of computation per delivery; on a lighter handler they cost more than they save. Poll diagnostics find the handlers that hold up the runtime.