Broadcast price ticks to every consumer with fibre spmc

You'll build a small Tokio program where one price feed sends ticks to several consumers over a bounded spmc channel and every consumer sees every tick. Along the way you'll add a consumer while the feed is running, watch the slowest consumer set the pace for the feed and let consumers leave early without breaking the others. The program shuts down on its own when the feed ends or when nobody is listening any more.

fibre = "0"

Set up the project#

Start a new binary crate:

cargo new fibre-ticks
cd fibre-ticks

Add fibre, Tokio and futures-util to Cargo.toml. The program only needs Tokio's single-threaded runtime, its macros and its timers. futures-util provides StreamExt, which you'll use on a receiver in the last step.

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

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

Send every tick to two consumers#

spmc is fibre's broadcast channel: one sender and any number of receivers. Unlike mpmc, where each message goes to exactly one receiver, an spmc message is cloned and delivered to every receiver, so T has to be Clone. The channel is always bounded. Replace src/main.rs with this:

use fibre::spmc;

#[derive(Debug, Clone)]
struct Tick {
    symbol: &'static str,
    price: f64,
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (tx, rx) = spmc::bounded_async::<Tick>(4);

    let mut consumers = Vec::new();
    for name in ["chart", "audit"] {
        let rx = rx.clone();
        consumers.push(tokio::spawn(async move {
            while let Ok(tick) = rx.recv().await {
                println!("{name}: {} {:.2}", tick.symbol, tick.price);
            }
            println!("{name}: feed closed");
        }));
    }
    drop(rx);

    let ticks = [("ACME", 101.25), ("ACME", 101.50), ("ACME", 100.75)];
    for (symbol, price) in ticks {
        tx.send(Tick { symbol, price }).await.expect("a consumer is listening");
    }
    drop(tx);

    for consumer in consumers {
        consumer.await.unwrap();
    }
}

spmc::bounded_async(4) returns a BoundedAsyncSender and a BoundedAsyncReceiver. The sender is not Clone, which is what makes the channel single-producer. The receiver is Clone and each clone is an independent consumer with its own place in the stream. Here chart and audit each get a clone and main drops the original with drop(rx).

That drop matters more on spmc than on other channels. A receiver still counts as a consumer while it's alive, even if nobody ever calls recv on it. The feed can't get more than 4 ticks ahead of any consumer, so once 4 ticks had piled up behind the forgotten original, every later send would wait forever.

Run it:

$ cargo run
chart: ACME 101.25
chart: ACME 101.50
chart: ACME 100.75
chart: feed closed
audit: ACME 101.25
audit: ACME 101.50
audit: ACME 100.75
audit: feed closed

Both consumers got all three ticks in order. main sends all three before it awaits anything, so the consumers only start running once it waits on the first handle. chart then reads through its three ticks and audit reads the same three again. Reading a tick in one consumer doesn't take it away from the other.

The end of the stream works like the other fibre channels. drop(tx) disconnects the channel. Each receiver still drains the ticks buffered for it and only then gets Err(RecvError::Disconnected), which ends its while let loop.

Let the slowest consumer set the pace#

A bounded broadcast channel has to keep each tick until every consumer has read it. So the feed can only run as far ahead as the slowest consumer allows. To see it, shrink the channel to 2, make audit take 100ms per tick and have the feed print a line each time a tick gets in. The helper ms rounds the elapsed time to the nearest 100ms so the numbers stay stable from run to run. Here's the whole file:

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

#[derive(Debug, Clone)]
struct Tick {
    symbol: &'static str,
    price: f64,
}

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

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

    let mut consumers = Vec::new();
    for (name, delay) in [("chart", 0), ("audit", 100)] {
        let rx = rx.clone();
        consumers.push(tokio::spawn(async move {
            while let Ok(tick) = rx.recv().await {
                sleep(Duration::from_millis(delay)).await;
                println!("{:>4}ms {name}: {} {:.2}", ms(start), tick.symbol, tick.price);
            }
            println!("{:>4}ms {name}: feed closed", ms(start));
        }));
    }
    drop(rx);

    let prices = [101.25, 101.50, 100.75, 100.50, 101.00];
    for (n, price) in prices.into_iter().enumerate() {
        tx.send(Tick { symbol: "ACME", price }).await.expect("a consumer is listening");
        println!("{:>4}ms feed: sent #{n}", ms(start));
    }
    drop(tx);

    for consumer in consumers {
        consumer.await.unwrap();
    }
}

   0ms feed: sent #0
   0ms feed: sent #1
   0ms feed: sent #2
   0ms chart: ACME 101.25
   0ms chart: ACME 101.50
   0ms chart: ACME 100.75
 100ms audit: ACME 101.25
 100ms feed: sent #3
 100ms chart: ACME 100.50
 200ms audit: ACME 101.50
 200ms feed: sent #4
 200ms chart: ACME 101.00
 200ms chart: feed closed
 300ms audit: ACME 100.75
 400ms audit: ACME 100.50
 500ms audit: ACME 101.00
 500ms audit: feed closed

At time zero the feed gets three ticks in. The first two fill the buffer. Then audit takes 101.25 out and starts its 100ms of work, which frees one slot for tick #2. From there the feed's send waits for audit: #3 gets in at 100ms when audit picks up its next tick and #4 at 200ms.

chart does no work at all, yet it prints 100.50 at 100ms and 101.00 at 200ms. It has nothing to read in between because the feed is stuck waiting on audit. On an spmc channel every consumer runs at the speed of the slowest one, give or take the buffer. A bigger capacity absorbs short stalls. It won't help a consumer that is slower on average.

If the feed would rather drop ticks than wait, use try_send. It never waits. When any consumer's view of the buffer is full, it returns TrySendError::Full with the tick inside. Swap the send line and its println! in the feed loop for this and add use fibre::error::TrySendError; at the top:

match tx.try_send(Tick { symbol: "ACME", price }) {
    Ok(()) => println!("{:>4}ms feed: sent #{n}", ms(start)),
    Err(TrySendError::Full(tick)) => println!("{:>4}ms feed: dropped #{n} at {:.2}, a consumer is behind", ms(start), tick.price),
    Err(_) => break,
}

   0ms feed: sent #0
   0ms feed: sent #1
   0ms feed: dropped #2 at 100.75, a consumer is behind
   0ms feed: dropped #3 at 100.50, a consumer is behind
   0ms feed: dropped #4 at 101.00, a consumer is behind
   0ms chart: ACME 101.25
   0ms chart: ACME 101.50
   0ms chart: feed closed
 100ms audit: ACME 101.25
 200ms audit: ACME 101.50
 200ms audit: feed closed

The feed now finishes at time zero. Ticks #2 to #4 are gone for everyone, chart included, because the channel has one buffer shared by all consumers. The catch-all arm covers TrySendError::Closed, which means every receiver is gone. Put the blocking send back before moving on.

Add a consumer while the feed runs#

Consumers don't have to exist up front. Any receiver can be cloned at any time and the clone joins as a new consumer. This time the feed runs in its own task and sends a tick every 100ms, like a real feed would. main reads as the chart. After its second tick it clones its receiver and hands the clone to a new recorder task. The consumer loop moves into a watch function so both can share it:

use fibre::spmc;
use std::time::Duration;
use tokio::time::sleep;

#[derive(Debug, Clone)]
struct Tick {
    symbol: &'static str,
    price: f64,
}

async fn watch(name: &'static str, rx: spmc::BoundedAsyncReceiver<Tick>) {
    while let Ok(tick) = rx.recv().await {
        println!("{name}: {} {:.2}", tick.symbol, tick.price);
    }
    println!("{name}: feed closed");
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (tx, rx) = spmc::bounded_async::<Tick>(4);

    let feed = tokio::spawn(async move {
        let prices = [101.25, 101.50, 100.75, 100.50, 101.00];
        for price in prices {
            sleep(Duration::from_millis(100)).await;
            tx.send(Tick { symbol: "ACME", price }).await.expect("a consumer is listening");
        }
    });

    let mut recorder = None;
    let mut seen = 0;
    while let Ok(tick) = rx.recv().await {
        println!("chart: {} {:.2}", tick.symbol, tick.price);
        seen += 1;
        if seen == 2 {
            recorder = Some(tokio::spawn(watch("recorder", rx.clone())));
        }
    }
    println!("chart: feed closed");

    feed.await.unwrap();
    if let Some(recorder) = recorder {
        recorder.await.unwrap();
    }
}

chart: ACME 101.25
chart: ACME 101.50
recorder: ACME 100.75
chart: ACME 100.75
recorder: ACME 100.50
chart: ACME 100.50
recorder: ACME 101.00
recorder: feed closed
chart: ACME 101.00
chart: feed closed

recorder starts with the third tick. A clone starts at the same point in the stream as the receiver it was cloned from. rx had already read 101.25 and 101.50 when it was cloned, so those two are not repeated for recorder. Everything sent after that reaches both.

From then on each tick shows up in both consumers. Both are waiting when a tick arrives, so which of the two prints it first depends on the order they are woken in, which fibre doesn't promise. What stays fixed is the order of ticks within each consumer and the point where recorder joins. When the feed task finishes, its sender is dropped and both loops end after the last tick.

Let consumers leave early#

A consumer can stop whenever it likes. Dropping its receiver removes it from the channel. The feed stops waiting on it and the other consumers carry on. When the last receiver is gone, send returns Err(SendError::Closed) and the feed knows there is no point in continuing.

The finished program below has two consumers that each leave early. alerts waits for the first price under 101.00, reports it and exits. BoundedAsyncReceiver implements futures::Stream, so it uses StreamExt::filter and next instead of a recv loop. chart reads four ticks and then drops its receiver. Replace all of src/main.rs with it:

use fibre::spmc;
use futures_util::StreamExt;
use std::future::ready;
use std::time::Duration;
use tokio::time::sleep;

#[derive(Debug, Clone)]
struct Tick {
    symbol: &'static str,
    price: f64,
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (tx, rx) = spmc::bounded_async::<Tick>(4);

    let feed = tokio::spawn(async move {
        let prices = [101.25, 101.50, 100.75, 100.50, 101.00, 101.75];
        for (n, price) in prices.into_iter().enumerate() {
            sleep(Duration::from_millis(100)).await;
            if tx.send(Tick { symbol: "ACME", price }).await.is_err() {
                println!("feed: no consumers left, stopping at #{n}");
                return;
            }
        }
        println!("feed: all ticks sent");
    });

    let alerts_rx = rx.clone();
    let alerts = tokio::spawn(async move {
        let mut low = alerts_rx.filter(|tick| ready(tick.price < 101.00));
        if let Some(tick) = low.next().await {
            println!("alerts: {} fell to {:.2}, alert sent", tick.symbol, tick.price);
        }
    });

    let mut seen = 0;
    while let Ok(tick) = rx.recv().await {
        println!("chart: {} {:.2}", tick.symbol, tick.price);
        seen += 1;
        if seen == 4 {
            break;
        }
    }
    drop(rx);
    println!("chart: closed after {seen} ticks");

    alerts.await.unwrap();
    feed.await.unwrap();
}

chart: ACME 101.25
chart: ACME 101.50
alerts: ACME fell to 100.75, alert sent
chart: ACME 100.75
chart: ACME 100.50
chart: closed after 4 ticks
feed: no consumers left, stopping at #4

As in the last step, which consumer prints first for the same tick can vary. alerts sees 101.25 and 101.50, skips them, returns on 100.75 and drops low along with the receiver inside it. The feed keeps going without it and chart still gets its fourth tick. After that chart breaks out of its loop and drop(rx) removes the last receiver. The feed's next send at 500ms fails, so it stops at #4 instead of sending into a channel nobody reads.

Shut down in order#

An spmc channel has the same two shutdown paths as the other fibre channels. The difference is how many consumers each one reaches:

feed drops tx        -> every receiver drains its buffered ticks, then recv() returns Err
a consumer drops rx  -> that consumer stops at once, the feed and other consumers carry on
last rx dropped      -> the feed's next send() returns Err(SendError::Closed)

Closing from the sending side is the clean stop. Each consumer finishes every tick that was sent before the close, however far behind it was. Dropping a receiver (or calling close() on it) doesn't drain anything. That consumer just stops reading, which is fine for a consumer that has got what it came for, like alerts.

The sender has an explicit close() too. It takes &mut self, so the binding has to be mutable. In the first step's program, change the let line and replace drop(tx) with the close() call:

let (mut tx, rx) = spmc::bounded_async::<Tick>(4);

tx.close().expect("the feed closes once");

The output is the same as before. Consumers see the same thing as with a dropped sender: they drain what's buffered and then get Err(RecvError::Disconnected). Keep the rule from the first step in mind as well. Every receiver you keep around, even one you only hold to clone later, is a consumer the feed has to wait for.

Where to go next#

The channels guide at https://github.com/excsn/fibre/blob/main/channels/README.GUIDE.md covers the rest of fibre. For broadcast it also has spmc::topic, a publish-subscribe variant where consumers subscribe to named topics and a slow consumer loses messages instead of holding up the sender. It also covers the batch API (send_batch and recv_batch work on spmc too) and mixing sync threads with async tasks on one channel using to_sync() and to_async(), for example to hand a receiver clone to a blocking thread that writes ticks to disk.

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