Build a pooled order pricer with fibre OneshotHostPool

You'll build an order pricer where every order lives in a record taken from a fixed pool, with its reply channel embedded in the same record. A caller fills the record, hands the reply sender to a pricing task and reads the order back from the record when the price arrives. Along the way you'll see which side can read the record, what happens when every record is in use and how to take several records at once.

fibre = "0"

Set up the project#

Start a new binary crate:

cargo new fibre-pricer
cd fibre-pricer

Add fibre and Tokio to Cargo.toml. The pricing task runs on Tokio's single-threaded runtime.

[dependencies]
fibre = "0"
tokio = { version = "1", features = ["macros", "rt"] }

The program uses the current_thread runtime, so the output comes out in the same order every run and you can compare it line for line with what's shown here.

Put an order in a pooled record#

A request and reply usually needs two things per request: somewhere to keep the request (a struct, often boxed or in an Arc) and a reply channel. fibre's OneshotPairPool recycles the reply channel, but the request itself is still yours to allocate. OneshotHostPool pools both. Each cell in the pool is a whole record of a type you define. The record has a PoolSlot<T> field that becomes the reply channel. It is the zero-allocation path for the full request: a request and reply through the host pool performs no allocation.

You build the pool with OneshotHostPool::new(capacity, make, project). make builds a blank record. project is a plain function that points at the PoolSlot field inside a record; it must return the same field every time. The record type has to be Send + Sync.

Replace src/main.rs with this:

use fibre::oneshot::{OneshotHostPool, PoolSlot};

struct Order {
    id: u32,
    sku: &'static str,
    quantity: u32,
    price: PoolSlot<Option<u64>>,
}

fn blank_order() -> Order {
    Order { id: 0, sku: "", quantity: 0, price: PoolSlot::new() }
}

fn price_slot(order: &Order) -> &PoolSlot<Option<u64>> {
    &order.price
}

fn main() {
    let orders = OneshotHostPool::new(2, blank_order, price_slot);

    let (reply, mut answer) = orders
        .pair_init(|order| {
            order.id = 1;
            order.sku = "green-tea";
            order.quantity = 3;
        })
        .expect("a free record");

    let order = answer.host();
    println!("order {}: {} x {}", order.id, order.quantity, order.sku);

    reply.send(Some(450)).expect("the receiver is still here");
    println!("reply: {:?}", answer.recv_blocking());
}

pair_init takes a free record from the pool, runs your closure on it and hands back a HostSender and a HostReceiver for the reply channel inside that record. The closure gets &mut Order and nothing else can touch the record while it runs. That's where you write the request. Set every field the request uses in this closure, since a record comes back from the pool as the last request left it.

answer.host() gives you &Order, the record the channel rides in. It stays readable until the receiver finishes. The sender and receiver otherwise work like the pooled pair from OneshotPairPool: send takes the sender by value and never waits, recv().await is for async code and recv_blocking() is for plain threads. The reply here is already in the slot when recv_blocking() runs, so it returns straight away. Run it:

$ cargo run
order 1: 3 x green-tea
reply: Ok(Some(450))

The value is Option<u64> because the pricer will answer None for a product it doesn't sell. PoolSlot itself is opaque. You only create it with PoolSlot::new() and it only turns into a channel through the pool's projection.

Hand the reply sender to a pricing task#

Now move the pricing into its own task. Callers send it a PriceRequest over an mpsc queue. Each request carries the HostSender for the order's reply.

The pricing task can't read the order record. host() is a method on HostReceiver only; HostSender has no view of the record. The record stays with the caller, who holds the receiver. Anything the pricing task needs goes in the request message. Here that's the SKU.

Replace src/main.rs with this:

use fibre::mpsc;
use fibre::oneshot::{HostReceiver, HostSender, OneshotHostPool, PoolSlot};

struct Order {
    id: u32,
    sku: &'static str,
    quantity: u32,
    price: PoolSlot<Option<u64>>,
}

fn blank_order() -> Order {
    Order { id: 0, sku: "", quantity: 0, price: PoolSlot::new() }
}

fn price_slot(order: &Order) -> &PoolSlot<Option<u64>> {
    &order.price
}

struct PriceRequest {
    sku: &'static str,
    reply: HostSender<Order, Option<u64>>,
}

async fn run_pricing(inbox: mpsc::BoundedAsyncReceiver<PriceRequest>) {
    while let Ok(request) = inbox.recv().await {
        let unit_price = match request.sku {
            "green-tea" => Some(450),
            "oolong" => Some(700),
            _ => None,
        };
        let _ = request.reply.send(unit_price);
    }
    println!("pricing: no more requests, exiting");
}

async fn finish(mut answer: HostReceiver<Order, Option<u64>>) {
    let order = answer.host();
    let (id, sku, quantity) = (order.id, order.sku, order.quantity);
    match answer.recv().await {
        Ok(Some(unit)) => println!("order {id}: {quantity} x {sku} at {unit} = {}", unit * u64::from(quantity)),
        Ok(None) => println!("order {id}: {sku} is not for sale"),
        Err(err) => println!("order {id}: no reply ({err:?})"),
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (requests, inbox) = mpsc::bounded_async::<PriceRequest>(8);
    let pricing = tokio::spawn(run_pricing(inbox));
    let orders = OneshotHostPool::new(2, blank_order, price_slot);

    let basket = [("green-tea", 3), ("oolong", 1), ("matcha", 2), ("green-tea", 1)];
    for (id, (sku, quantity)) in (1..).zip(basket) {
        let (reply, answer) = orders
            .pair_init(|order| {
                order.id = id;
                order.sku = sku;
                order.quantity = quantity;
            })
            .expect("a free record");
        requests.send(PriceRequest { sku, reply }).await.expect("pricing is running");
        finish(answer).await;
    }

    drop(requests);
    pricing.await.unwrap();
}

finish takes only the HostReceiver and gets everything else from the record: the order id, the SKU and the quantity it needs to work out the total. It copies those fields out before awaiting the reply, because recv() borrows the receiver mutably and host() borrows it shared, so the two can't overlap. When finish returns, the receiver is dropped. The sender was already used up by send, so both handles are done and the record goes back to the pool.

order 1: 3 x green-tea at 450 = 1350
order 2: 1 x oolong at 700 = 700
order 3: matcha is not for sale
order 4: 1 x green-tea at 450 = 450
pricing: no more requests, exiting

Four orders ran through a pool of two records. Each order waits for its price before the next one starts, so only one record is ever out at a time. pair_init pops a record that already exists in the pool rather than building a new one. The mpsc queue into the pricing task is a separate channel. The host pool covers the record and its reply, not that queue.

The pricing task ignores the result of send. It fails only when the caller has dropped its receiver. At that point there's nobody to tell. Either way the record returns to the pool once both handles are finished, whichever finishes last.

Run out of records#

Waiting for each price before sending the next order wastes the pool. A real caller sends several orders and collects the prices as they come back. The pool's capacity is then a hard cap on orders in flight: when every record is out, pair_init returns None.

Change the caller to keep its pending receivers in a queue. When the pool is empty, it finishes the oldest order to put that order's record back and then tries again. Add the import:

use std::collections::VecDeque;

Then replace main:

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (requests, inbox) = mpsc::bounded_async::<PriceRequest>(8);
    let pricing = tokio::spawn(run_pricing(inbox));
    let orders = OneshotHostPool::new(2, blank_order, price_slot);
    let mut in_flight = VecDeque::new();

    let basket = [("green-tea", 3), ("oolong", 1), ("matcha", 2), ("green-tea", 1)];
    for (id, (sku, quantity)) in (1..).zip(basket) {
        let fill = move |order: &mut Order| {
            order.id = id;
            order.sku = sku;
            order.quantity = quantity;
        };
        let (reply, answer) = match orders.pair_init(fill) {
            Some(pair) => pair,
            None => {
                println!("order {id}: no free record, waiting for the oldest reply");
                finish(in_flight.pop_front().expect("an order in flight")).await;
                orders.pair_init(fill).expect("a record was just returned")
            }
        };
        requests.send(PriceRequest { sku, reply }).await.expect("pricing is running");
        println!("order {id}: sent");
        in_flight.push_back(answer);
    }
    while let Some(answer) = in_flight.pop_front() {
        finish(answer).await;
    }

    drop(requests);
    pricing.await.unwrap();
}

The fill closure captures only copyable values, so it's Copy and can be passed to pair_init twice. Each HostReceiver in in_flight is a complete pending order: the order data and the place its price will land, in one value, with no side table keyed by order id.

order 1: sent
order 2: sent
order 3: no free record, waiting for the oldest reply
order 1: 3 x green-tea at 450 = 1350
order 3: sent
order 4: no free record, waiting for the oldest reply
order 2: 1 x oolong at 700 = 700
order 4: sent
order 3: matcha is not for sale
order 4: 1 x green-tea at 450 = 450
pricing: no more requests, exiting

Orders 1 and 2 take both records. Order 3 finds the pool empty, so the caller waits for order 1's price. Dropping order 1's receiver in finish returns its record. The retry for order 3 gets it. Order 4 goes the same way with order 2's record. The retry can't fail: only this caller uses the pool. A record is available again as soon as both its handles are done.

Size the pool for the number of requests you want in flight at once. A caller that gives up and drops its receiver doesn't lose the record: it goes back when the pricing task's send finishes.

Take several records at once#

When a caller has a batch of requests ready, pair_init_batch(n, init) takes n records in one go. init is FnMut and runs once per record, so it can pull the next line from an iterator. It returns Option<Vec<(HostSender, HostReceiver)>> and gives you either all n pairs or None.

Add this to main, just before drop(requests):

    let restock = [("oolong", 2), ("green-tea", 5)];
    let mut lines = (5..).zip(restock);
    let mut batch = orders
        .pair_init_batch(restock.len(), |order| {
            let (id, (sku, quantity)) = lines.next().expect("one line per record");
            order.id = id;
            order.sku = sku;
            order.quantity = quantity;
        })
        .expect("both records are free");
    batch.sort_by_key(|(_, answer)| answer.host().id);

    let extra = orders.pair_init_batch(1, |order| order.id = 7);
    println!("batch of {} taken, another record free: {}", batch.len(), extra.is_some());

    let mut answers = Vec::new();
    for (reply, answer) in batch {
        let sku = answer.host().sku;
        requests.send(PriceRequest { sku, reply }).await.expect("pricing is running");
        answers.push(answer);
    }
    for answer in answers {
        finish(answer).await;
    }

The program sorts the pairs by the id in each record and reads the SKU for each request from the record itself. Reading the record is the only way to tell which order a pair belongs to.

order 1: sent
order 2: sent
order 3: no free record, waiting for the oldest reply
order 1: 3 x green-tea at 450 = 1350
order 3: sent
order 4: no free record, waiting for the oldest reply
order 2: 1 x oolong at 700 = 700
order 4: sent
order 3: matcha is not for sale
order 4: 1 x green-tea at 450 = 450
batch of 2 taken, another record free: false
order 5: 2 x oolong at 700 = 1400
order 6: 5 x green-tea at 450 = 2250
pricing: no more requests, exiting

The batch takes both records, so the second request for even one record gets None. The two restock orders are sent together and their prices collected afterwards. When main drops requests, the pricing task's recv() returns an error, its loop ends and the program exits.

Where to go next#

The channels guide at https://github.com/excsn/fibre/blob/main/channels/README.GUIDE.md compares the oneshot constructors: oneshot(), exclusive(), pair_pool() and OneshotHostPool, with benchmark figures for the two pools on record-carrying workloads. It also covers the mpsc, mpmc, spsc and spmc channels you'd use for the request queue.

The full API, including PoolSlot, HostSender, HostReceiver and every error type, is on docs.rs at https://docs.rs/fibre.