Build a request and reply actor with fibre oneshot

You'll build a small bank actor: one Tokio task owns the account balances and every other part of the program talks to it by sending a request with a oneshot reply channel inside. Along the way you'll see what a caller gets when the actor never answers, how to put a timeout on a reply, how a plain thread waits for one and how to reuse reply channels from a pool when requests come in fast.

fibre = "0"

Set up the project#

Start a new binary crate:

cargo new fibre-bank
cd fibre-bank

Add fibre and Tokio to Cargo.toml. The program needs Tokio's single-threaded runtime, its macros and its timers.

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

Everything runs on the current_thread runtime, so the output comes out in the same order every time and you can compare it line for line with what's shown here.

Ask the actor for a value#

An actor is a task that owns some state and changes it only in response to messages. Callers can't touch the state directly, so they need a way to get an answer back. The usual shape is a request queue into the actor plus a fresh one-value reply channel per request. The request queue here is an mpsc channel, since many callers send to one actor. Each reply is a oneshot.

fibre's oneshot module has a few constructors. oneshot::exclusive() is the one for plain request and reply: it gives you an ExclusiveSender and an ExclusiveReceiver, neither of which can be cloned. Sending on it is a single slot write. oneshot::oneshot() returns a sender you can clone, for the case where several tasks race to fill one reply and only the first send wins. This tutorial uses exclusive().

Replace src/main.rs with this:

use fibre::mpsc;
use fibre::oneshot::{self, ExclusiveSender};
use std::collections::HashMap;

enum Request {
    Deposit { account: &'static str, amount: u64, reply: ExclusiveSender<u64> },
    Balance { account: &'static str, reply: ExclusiveSender<Option<u64>> },
}

async fn run_bank(inbox: mpsc::BoundedAsyncReceiver<Request>) {
    let mut balances: HashMap<&'static str, u64> = HashMap::new();
    while let Ok(request) = inbox.recv().await {
        match request {
            Request::Deposit { account, amount, reply } => {
                let balance = balances.entry(account).or_insert(0);
                *balance += amount;
                let _ = reply.send(*balance);
            }
            Request::Balance { account, reply } => {
                let _ = reply.send(balances.get(account).copied());
            }
        }
    }
    println!("bank: all callers gone, exiting");
}

#[derive(Clone)]
struct Bank {
    requests: mpsc::BoundedAsyncSender<Request>,
}

impl Bank {
    async fn deposit(&self, account: &'static str, amount: u64) -> u64 {
        let (reply, mut answer) = oneshot::exclusive();
        self.requests.send(Request::Deposit { account, amount, reply }).await.expect("bank is running");
        answer.recv().await.expect("bank replied")
    }

    async fn balance(&self, account: &'static str) -> Option<u64> {
        let (reply, mut answer) = oneshot::exclusive();
        self.requests.send(Request::Balance { account, reply }).await.expect("bank is running");
        answer.recv().await.expect("bank replied")
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (requests, inbox) = mpsc::bounded_async::<Request>(32);
    let actor = tokio::spawn(run_bank(inbox));
    let bank = Bank { requests };

    println!("alice: deposited 50, balance {}", bank.deposit("alice", 50).await);
    println!("alice: deposited 25, balance {}", bank.deposit("alice", 25).await);
    println!("bob: balance {:?}", bank.balance("bob").await);

    drop(bank);
    actor.await.unwrap();
}

Each Request variant carries the sending half of its reply channel. The caller keeps the receiving half, sends the request and then awaits recv() on its receiver. The actor answers with reply.send(value), which takes the sender by value, so a reply can only ever be sent once. A oneshot send never waits: the slot is always free, so there is no async version of it and no backpressure to think about.

Bank is the handle callers use. It wraps the request sender and hides the channel plumbing behind ordinary async methods. It's Clone because mpsc senders are. When the last Bank is dropped, the request queue disconnects, inbox.recv() returns an error and the actor loop ends. Run it:

$ cargo run
alice: deposited 50, balance 50
alice: deposited 25, balance 75
bob: balance None
bank: all callers gone, exiting

The actor ignores the result of reply.send with let _ =. A send fails only when the caller's receiver is already gone. At that point there's nobody left to tell. You'll use that result later.

Handle an actor that never replies#

A reply sender can be dropped without being used. That happens when the actor panics while handling a request or when a handler takes an early return and forgets to reply. The caller doesn't hang in any of these cases: recv() resolves with Err(RecvError::Disconnected) as soon as the sender is gone.

To see it, add a withdraw request whose handler has a gap. An unknown account hits continue and the whole request, reply sender included, is dropped. Add the variant to Request:

    Withdraw { account: &'static str, amount: u64, reply: ExclusiveSender<Option<u64>> },

Add its arm to the match in run_bank:

            Request::Withdraw { account, amount, reply } => {
                let Some(balance) = balances.get_mut(account) else {
                    println!("bank: no account {account}, skipping");
                    continue;
                };
                let left = balance.checked_sub(amount);
                if let Some(left) = left {
                    *balance = left;
                }
                let _ = reply.send(left);
            }

This caller method returns the recv() result as it is, so you can see the error. Add use fibre::error::RecvError; at the top and this method to impl Bank:

    async fn withdraw(&self, account: &'static str, amount: u64) -> Result<Option<u64>, RecvError> {
        let (reply, mut answer) = oneshot::exclusive();
        self.requests.send(Request::Withdraw { account, amount, reply }).await.expect("bank is running");
        answer.recv().await
    }

Then change the calls in main:

    println!("alice: deposited 50, balance {}", bank.deposit("alice", 50).await);
    println!("alice: withdraw 30 -> {:?}", bank.withdraw("alice", 30).await);
    println!("alice: withdraw 30 -> {:?}", bank.withdraw("alice", 30).await);
    println!("carol: withdraw 10 -> {:?}", bank.withdraw("carol", 10).await);
    println!("alice: balance {:?}", bank.balance("alice").await);

alice: deposited 50, balance 50
alice: withdraw 30 -> Ok(Some(20))
alice: withdraw 30 -> Ok(None)
bank: no account carol, skipping
carol: withdraw 10 -> Err(Disconnected)
alice: balance Some(20)
bank: all callers gone, exiting

The second withdrawal is refused with a proper answer, Ok(None). Carol's request gets no answer at all and the caller finds out from Disconnected. The actor keeps running and serves the next request normally. In a real program you'd fix the handler to reply for every request and treat Disconnected as "the actor is broken", but it's good to know the caller is never left waiting.

Put a timeout on the reply#

Sometimes the actor is alive but slow. Add a statement request that takes 200ms to build. The actor also checks the result of send this time: if it gets TrySendError::Closed back, the caller stopped waiting and the value comes back inside the error.

Replace the imports at the top with these:

use fibre::error::{RecvError, TrySendError};
use fibre::mpsc;
use fibre::oneshot::{self, ExclusiveSender};
use std::collections::HashMap;
use std::time::Duration;
use tokio::time::{sleep, timeout};

Add the variant:

    Statement { account: &'static str, reply: ExclusiveSender<String> },

Add its arm to run_bank:

            Request::Statement { account, reply } => {
                sleep(Duration::from_millis(200)).await;
                let text = format!("statement for {account}: {:?}", balances.get(account));
                if let Err(TrySendError::Closed(_)) = reply.send(text) {
                    println!("bank: {account}'s caller gave up, statement discarded");
                }
            }

Add the caller method to impl Bank:

    async fn statement(&self, account: &'static str) -> String {
        let (reply, mut answer) = oneshot::exclusive();
        self.requests.send(Request::Statement { account, reply }).await.expect("bank is running");
        answer.recv().await.expect("bank replied")
    }

In main, wrap the call in tokio::time::timeout. Replace the calls from the last step with these:

    println!("alice: deposited 50, balance {}", bank.deposit("alice", 50).await);

    match timeout(Duration::from_millis(100), bank.statement("alice")).await {
        Ok(text) => println!("alice: {text}"),
        Err(_) => println!("alice: no statement after 100ms, giving up"),
    }
    println!("alice: balance {:?}", bank.balance("alice").await);

    match timeout(Duration::from_millis(300), bank.statement("alice")).await {
        Ok(text) => println!("alice: {text}"),
        Err(_) => println!("alice: no statement after 300ms, giving up"),
    }
    println!("carol: withdraw 10 -> {:?}", bank.withdraw("carol", 10).await);

alice: deposited 50, balance 50
alice: no statement after 100ms, giving up
bank: alice's caller gave up, statement discarded
alice: balance Some(50)
alice: statement for alice: Some(50)
bank: no account carol, skipping
carol: withdraw 10 -> Err(Disconnected)
bank: all callers gone, exiting

When the 100ms timeout fires, Tokio drops the statement future and the ExclusiveReceiver inside it goes with it. Dropping a receiver closes the channel. The actor doesn't know yet: it finishes the statement at 200ms and only learns the caller left when send returns Closed. Meanwhile the balance request is queued behind the statement, so it's answered right after. The second statement has 300ms, which is enough.

A timeout on the reply doesn't cancel the work. If a request is expensive, the actor can check reply.is_closed() before starting and skip requests whose callers have already given up.

Wait for a reply from a plain thread#

Not every caller is async. Every oneshot receiver has recv_blocking() for synchronous code. The sending side needs no sync version, since a oneshot send never waits. The request queue does need a blocking sender. to_sync() converts an async mpsc sender into one.

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

    let requests = bank.requests.clone().to_sync();
    let report = tokio::task::spawn_blocking(move || {
        let (reply, mut answer) = oneshot::exclusive();
        requests.send(Request::Balance { account: "alice", reply }).expect("bank is running");
        answer.recv_blocking().expect("bank replied")
    });
    println!("report thread: alice balance {:?}", report.await.unwrap());

alice: deposited 50, balance 50
alice: no statement after 100ms, giving up
bank: alice's caller gave up, statement discarded
alice: balance Some(50)
alice: statement for alice: Some(50)
bank: no account carol, skipping
carol: withdraw 10 -> Err(Disconnected)
report thread: alice balance Some(50)
bank: all callers gone, exiting

spawn_blocking runs the closure on a separate thread, so recv_blocking() parks that thread and leaves the runtime free to run the actor. The actor sends the reply from async code and the thread wakes up with it. The converted sender is a clone of the request queue sender. It's dropped when the closure returns, so the actor still shuts down once bank is dropped.

Reuse reply channels from a pool#

Each exclusive() call allocates a small channel and frees it once both ends are done. At high request rates that allocation shows up. For that case fibre has OneshotPairPool: a fixed set of reply channels that get recycled. pair() hands out a PooledSender and PooledReceiver that behave like the exclusive pair. When both are finished, the slot goes back into the pool. When every slot is in use, pair() returns None, which gives you a hard cap on requests in flight.

Switch deposits to a pool. Update the oneshot import and add Arc:

use fibre::oneshot::{self, ExclusiveSender, OneshotPairPool, PooledSender};
use std::collections::HashMap;
use std::sync::Arc;

Change the reply type of Deposit:

    Deposit { account: &'static str, amount: u64, reply: PooledSender<u64> },

The actor's Deposit arm doesn't change, because a PooledSender has the same send as an ExclusiveSender. Give Bank the pool and an error for when it's empty:

#[derive(Debug)]
struct Busy;

#[derive(Clone)]
struct Bank {
    requests: mpsc::BoundedAsyncSender<Request>,
    deposit_replies: Arc<OneshotPairPool<u64>>,
}

Replace deposit so it takes its reply channel from the pool:

    async fn deposit(&self, account: &'static str, amount: u64) -> Result<u64, Busy> {
        let (reply, mut answer) = self.deposit_replies.pair().ok_or(Busy)?;
        self.requests.send(Request::Deposit { account, amount, reply }).await.expect("bank is running");
        Ok(answer.recv().await.expect("bank replied"))
    }

In main, build the bank with a pool of 2:

    let bank = Bank { requests, deposit_replies: Arc::new(oneshot::pair_pool(2)) };

Then replace the first deposit line with three concurrent callers and a retry:

    let mut callers = Vec::new();
    for (account, amount) in [("alice", 50), ("bob", 20), ("dave", 30)] {
        let bank = bank.clone();
        callers.push(tokio::spawn(async move {
            match bank.deposit(account, amount).await {
                Ok(balance) => println!("{account}: deposited {amount}, balance {balance}"),
                Err(Busy) => println!("{account}: no reply slot free, try again later"),
            }
        }));
    }
    for caller in callers {
        caller.await.unwrap();
    }
    println!("dave: retry -> {:?}", bank.deposit("dave", 30).await);

dave: no reply slot free, try again later
alice: deposited 50, balance 50
bob: deposited 20, balance 20
dave: retry -> Ok(30)
alice: no statement after 100ms, giving up
bank: alice's caller gave up, statement discarded
alice: balance Some(50)
alice: statement for alice: Some(50)
bank: no account carol, skipping
carol: withdraw 10 -> Err(Disconnected)
report thread: alice balance Some(50)
bank: all callers gone, exiting

The three callers run in the order they were spawned. Alice and Bob each take a slot and wait for their replies. Dave finds both slots in use and gets Busy straight away. Once Alice and Bob have their answers, their slots are back in the pool and Dave's retry goes through.

The pool is Send + Sync, so one Arc shared by every Bank clone is enough. Size it for the number of requests you expect in flight at once. If a caller times out holding a pooled receiver, the slot is returned when the actor's send finishes, so a slow actor holds slots for as long as it's slow. The other request types keep using exclusive(). Mixing the two in one actor is fine.

Where to go next#

The channels guide at https://github.com/excsn/fibre/blob/main/channels/README.GUIDE.md covers the rest of the oneshot module, including OneshotHostPool, which pools a whole request record together with its reply channel so a request and reply allocates nothing. The same guide covers mpsc, mpmc, spsc and spmc and how each mixes sync threads with async tasks.

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