Build a request metrics aggregator with fibre mpsc

You'll build a small Tokio program where several request handlers report every request they serve to one aggregator task, which owns the per-route counters and prints a report at the end. Along the way you'll swap between fibre's unbounded, bounded and rendezvous mpsc channels, see what a bounded send does when the channel is full and feed the same aggregator from a plain OS thread.

fibre = "0"

Set up the project#

Start a new binary crate:

cargo new fibre-metrics
cd fibre-metrics

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"] }

Every step uses the current_thread runtime. All the tasks share one thread, so the output comes out the same way on every run and you can compare it line for line with what's shown here.

Send events to one aggregator#

mpsc is fibre's many-to-one channel. Any number of senders can push into it and exactly one receiver takes everything out. That fits state with a single owner: request counters, an audit log or a connection table. Instead of wrapping the state in a lock that every handler contends on, the handlers send events and one task applies them in order.

Start with one handler and the unbounded async flavour. Replace src/main.rs with this:

use fibre::mpsc;

#[derive(Debug)]
struct Hit {
    route: &'static str,
    status: u16,
    millis: u64,
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (mut tx, mut rx) = mpsc::unbounded_async::<Hit>();

    tokio::spawn(async move {
        let requests = [(200, 12), (401, 3), (200, 15)];
        for (status, millis) in requests {
            let hit = Hit { route: "/login", status, millis };
            tx.send(hit).await.expect("aggregator is running");
        }
    });

    while let Ok(hit) = rx.recv().await {
        println!("aggregator: {} {} in {}ms", hit.route, hit.status, hit.millis);
    }
    println!("aggregator: every sender is gone");
}

mpsc::unbounded_async() returns an UnboundedAsyncSender and an UnboundedAsyncReceiver. Both send and recv take &mut self on the unbounded handles, which is why tx and rx are declared mut. An unbounded send never waits for room. The future completes as soon as it's polled and it fails only if the receiver is gone.

recv waits for the next hit. When every sender has been dropped and nothing is left in the channel, it returns Err(RecvError::Disconnected) instead. Here the only sender lives inside the handler task and is dropped when that task finishes, so the aggregator loop ends on its own after the third hit. Run it:

$ cargo run
aggregator: /login 200 in 12ms
aggregator: /login 401 in 3ms
aggregator: /login 200 in 15ms
aggregator: every sender is gone

Clone a sender per handler#

Senders on an mpsc channel are Clone. Each handler gets its own clone and they all feed the same receiver. The receiver is not Clone, so the compiler guarantees there is only ever one aggregator. That's what lets it keep its statistics in a plain BTreeMap with no lock around it.

Replace src/main.rs with a version that spawns three handlers and builds a per-route report:

use fibre::mpsc;
use std::collections::BTreeMap;

#[derive(Debug)]
struct Hit {
    route: &'static str,
    status: u16,
    millis: u64,
}

#[derive(Default)]
struct RouteStats {
    count: u32,
    errors: u32,
    total_millis: u64,
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (tx, mut rx) = mpsc::unbounded_async::<Hit>();

    let handlers = [
        ("/login", [(200, 12), (401, 3), (200, 15)]),
        ("/search", [(200, 40), (200, 35), (500, 90)]),
        ("/health", [(200, 1), (200, 1), (200, 2)]),
    ];

    for (route, requests) in handlers {
        let mut tx = tx.clone();
        tokio::spawn(async move {
            for (status, millis) in requests {
                if tx.send(Hit { route, status, millis }).await.is_err() {
                    break;
                }
            }
        });
    }
    drop(tx);

    let mut stats: BTreeMap<&str, RouteStats> = BTreeMap::new();
    while let Ok(hit) = rx.recv().await {
        let route = stats.entry(hit.route).or_default();
        route.count += 1;
        route.total_millis += hit.millis;
        if hit.status >= 400 {
            route.errors += 1;
        }
    }

    for (route, s) in &stats {
        let avg = s.total_millis / s.count as u64;
        println!("{route:<8} {} requests, {} errors, avg {avg}ms", s.count, s.errors);
    }
}

/health  3 requests, 0 errors, avg 1ms
/login   3 requests, 1 errors, avg 10ms
/search  3 requests, 1 errors, avg 55ms

The drop(tx) after the loop matters. Each handler's clone is dropped when its task finishes, but the original tx belongs to main. If you leave it alive, the channel always has one sender left and recv waits forever for a hit that never comes.

Hits from different handlers can interleave in any order in the channel. Hits from one handler always arrive in the order that handler sent them. The report is sorted by route, so the order hits arrive in doesn't change what gets printed.

Report from a sync thread#

Not every producer lives in the async runtime. A background job running on its own OS thread can report into the same aggregator. Every fibre handle converts between its sync and async forms with to_sync() and to_async(). The two forms work on the same channel at once.

Add a cleanup job on a std::thread right after the handler loop, before drop(tx):

    let mut cron_tx = tx.clone().to_sync();
    let cron = std::thread::spawn(move || {
        for millis in [250, 310] {
            if cron_tx.send(Hit { route: "/cron", status: 200, millis }).is_err() {
                break;
            }
        }
    });

tx.clone().to_sync() turns a fresh clone into an UnboundedSyncSender. Its send is a plain function call that returns a Result straight away. Wait for the thread after the aggregator loop, just before the report:

    cron.join().unwrap();

/cron    2 requests, 0 errors, avg 280ms
/health  3 requests, 0 errors, avg 1ms
/login   3 requests, 1 errors, avg 10ms
/search  3 requests, 1 errors, avg 55ms

The aggregator loop didn't change. It ends once every async handler and the cleanup thread have dropped their senders. The cleanup hits land at a different point among the others on each run, but the report comes out the same.

Bound the channel#

An unbounded channel accepts every hit immediately, which is fine while the aggregator keeps up. If it falls behind, hits pile up in memory without limit. mpsc::bounded_async(n) caps the channel at n queued items. When it's full, send returns a future that waits until the aggregator takes an item out and frees a slot.

To see that, give the channel room for 2 hits, make the aggregator take 100ms per hit and have a single handler send a burst of five. The handler prints a line each time a hit gets in. The helper ms rounds the elapsed time to the nearest 100ms so the numbers stay stable. Here is the whole file:

use fibre::mpsc;
use std::time::Duration;
use tokio::time::{sleep, Instant};

#[derive(Debug)]
struct Hit {
    route: &'static str,
    status: u16,
    millis: u64,
}

fn ms(start: Instant) -> u128 {
    (start.elapsed().as_millis() + 50) / 100 * 100
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (tx, rx) = mpsc::bounded_async::<Hit>(2);
    let start = Instant::now();

    let burst = [(200, 40), (200, 35), (500, 90), (200, 38), (200, 41)];
    tokio::spawn(async move {
        for (status, millis) in burst {
            if tx.send(Hit { route: "/search", status, millis }).await.is_err() {
                break;
            }
            println!("{:>4}ms handler: sent {status} {millis}ms", ms(start));
        }
    });

    while let Ok(hit) = rx.recv().await {
        sleep(Duration::from_millis(100)).await;
        println!("{:>4}ms aggregator: recorded {} {} {}ms", ms(start), hit.route, hit.status, hit.millis);
    }
    println!("{:>4}ms aggregator: every sender is gone", ms(start));
}

Bounded senders and receivers take &self, so neither tx nor rx needs mut any more. Run it:

   0ms handler: sent 200 40ms
   0ms handler: sent 200 35ms
   0ms handler: sent 500 90ms
 100ms aggregator: recorded /search 200 40ms
 100ms handler: sent 200 38ms
 200ms aggregator: recorded /search 200 35ms
 200ms handler: sent 200 41ms
 300ms aggregator: recorded /search 500 90ms
 400ms aggregator: recorded /search 200 38ms
 500ms aggregator: recorded /search 200 41ms
 500ms aggregator: every sender is gone

At time zero three hits get in. Two fill the channel and the aggregator has already taken the first one out to work on, which frees a slot for the third. After that the handler's send waits. Each time the aggregator takes the next hit, one slot opens and one more hit gets in, so from then on the handler moves at the aggregator's pace. Nothing is lost and the channel never holds more than 2 hits.

If a handler would rather drop metrics than slow down a request, use try_send. It never waits. When the channel is full it returns TrySendError::Full with the hit inside, so you can count it, log it or throw it away. Add use fibre::error::TrySendError; at the top and swap the body of the handler loop for this:

            match tx.try_send(Hit { route: "/search", status, millis }) {
                Ok(()) => println!("{:>4}ms handler: sent {status} {millis}ms", ms(start)),
                Err(TrySendError::Full(hit)) => println!("{:>4}ms handler: dropped {} {}ms, channel full", ms(start), hit.status, hit.millis),
                Err(_) => break,
            }

   0ms handler: sent 200 40ms
   0ms handler: sent 200 35ms
   0ms handler: dropped 500 90ms, channel full
   0ms handler: dropped 200 38ms, channel full
   0ms handler: dropped 200 41ms, channel full
 100ms aggregator: recorded /search 200 40ms
 200ms aggregator: recorded /search 200 35ms
 200ms aggregator: every sender is gone

The handler never yields between its try_send calls, so the aggregator doesn't get a chance to take anything out before the burst is over. The first two hits fill the channel and the other three are dropped. The catch-all arm covers TrySendError::Closed, which means the aggregator is gone.

Hand off audit entries with rendezvous#

Some events shouldn't sit in a buffer at all. An audit entry for a sensitive action is one: the handler should only carry on once the audit writer actually has the entry in hand. The mpsc::rendezvous module builds a channel with no capacity. A send completes only when the receiver takes that exact item, so each send is a direct handoff from one sender to the receiver. Senders are still Clone and there is still one receiver.

This step also shows the other way an mpsc channel ends. The audit writer only has room for three entries. When it's full it drops its receiver while a handler is still sending. Replace src/main.rs with this:

use fibre::mpsc::rendezvous;
use std::time::Duration;
use tokio::time::{sleep, Instant};

struct AuditEntry {
    user: &'static str,
    action: &'static str,
}

fn ms(start: Instant) -> u128 {
    (start.elapsed().as_millis() + 50) / 100 * 100
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (tx, rx) = rendezvous::rendezvous_async::<AuditEntry>();
    let start = Instant::now();

    let sessions = [("ana", 0, ["login", "export"]), ("ben", 150, ["login", "delete"])];

    let mut handles = Vec::new();
    for (user, delay, actions) in sessions {
        let tx = tx.clone();
        handles.push(tokio::spawn(async move {
            sleep(Duration::from_millis(delay)).await;
            for action in actions {
                match tx.send(AuditEntry { user, action }).await {
                    Ok(()) => println!("{:>4}ms {user}: {action} handed to the audit log", ms(start)),
                    Err(_) => {
                        println!("{:>4}ms {user}: audit log closed, {action} refused", ms(start));
                        break;
                    }
                }
            }
        }));
    }
    drop(tx);

    let mut room = 3;
    while let Ok(entry) = rx.recv().await {
        sleep(Duration::from_millis(100)).await;
        println!("{:>4}ms audit: wrote {} {}", ms(start), entry.user, entry.action);
        room -= 1;
        if room == 0 {
            println!("{:>4}ms audit: log full, closing", ms(start));
            break;
        }
    }
    drop(rx);

    for handle in handles {
        handle.await.unwrap();
    }
}

   0ms ana: login handed to the audit log
 100ms audit: wrote ana login
 100ms ana: export handed to the audit log
 200ms audit: wrote ana export
 200ms ben: login handed to the audit log
 300ms audit: wrote ben login
 300ms audit: log full, closing
 300ms ben: audit log closed, delete refused

Ana's first send completes at once, because the audit writer is already waiting in recv. Her second one waits until the writer finishes the first entry and calls recv again at 100ms. Ben's session starts at 150ms and his send waits until the writer is ready at 200ms. Nothing ever sits in the channel between a send and a receive.

After the third entry the writer breaks out of its loop and drops rx. Ben's pending send fails with SendError::Closed and he stops. Closing from the receiving side is immediate: the receiver doesn't drain anything and every later send fails. That's why the handlers in every step here break out of their loops on a send error instead of unwrapping it. For a clean shutdown where nothing is lost, close from the sending side as the earlier steps do and let the aggregator drain what's left.

For sync code, rendezvous::rendezvous() returns the blocking handles. The rendezvous flavour has no batch API.

Where to go next#

The channels guide at https://github.com/excsn/fibre/blob/main/channels/README.GUIDE.md covers the rest of mpsc, including recv_timeout for a sync aggregator that also needs to flush on a timer and the batch API, where recv_batch lets an aggregator take up to max events in one call. The async receivers also implement futures::Stream, so an aggregator can be written with StreamExt combinators. The same guide covers spsc, spmc, mpmc and oneshot for the other shapes of producer and consumer.

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