# Build a pooled order pricer with fibre OneshotHostPool

**For:** Rust developers who know the basics of Tokio and have used a oneshot reply channel.

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.

Before you start: you need Rust with Cargo. fibre comes from crates.io as `fibre = "0"`.

## Set up the project

Start a new binary crate:

```sh
cargo new fibre-pricer
cd fibre-pricer
```

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

```toml title="Cargo.toml"
[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:

```rust title="src/main.rs"
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:

```shell-session
$ 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:

```rust title="src/main.rs"
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.

```text
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:

```rust title="src/main.rs"
use std::collections::VecDeque;
```

Then replace `main`:

```rust title="src/main.rs"
#[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.

```text
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)`:

```rust title="src/main.rs"
    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.

```text
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](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](https://docs.rs/fibre).