Feed a Tokio task from a parser thread

You'll build a small program where an ordinary OS thread parses temperature readings and hands them to an async Tokio task over one spsc channel. The thread uses the channel's blocking API and the task uses its async API. Along the way you'll see what a full buffer does to the producer, how the channel closes from either end and how to move readings in batches.

fibre = "0"

Set up the project#

Start a new binary crate:

cargo new fibre-readings
cd fibre-readings

Add fibre and Tokio to Cargo.toml. The consumer runs on Tokio's single-threaded runtime and uses its timers to play a slow sink later on.

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

spsc is fibre's single-producer, single-consumer channel. It's always bounded and it's the fastest channel in the crate, because with exactly one sender and one receiver it can run as a lock-free ring buffer. It fits any one-to-one pipe: a reader thread feeding a writer, a decoder feeding a renderer or, here, a parser feeding an async task.

Send readings from a thread to a task#

The input is a few lines of text, one room and one temperature per line. One line is broken on purpose. The parser thread turns each good line into a Reading and sends it. The async consumer prints each one as it arrives. Replace src/main.rs with this:

use fibre::spsc;
use std::thread;

const INPUT: &str = "\
kitchen 21.5
attic 31.0
hallway 19.0
garage n/a
office 22.5
bedroom 20.5
cellar 14.5
";

struct Reading {
    room: String,
    celsius: f32,
}

fn parse(line: &str) -> Option<Reading> {
    let (room, value) = line.split_once(' ')?;
    let celsius = value.parse().ok()?;
    Some(Reading { room: room.to_string(), celsius })
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (tx, rx) = spsc::bounded_sync::<Reading>(4);
    let mut rx = rx.to_async();

    let producer = thread::spawn(move || {
        let mut skipped = 0;
        for line in INPUT.lines() {
            match parse(line) {
                Some(reading) => {
                    if tx.send(reading).is_err() {
                        break;
                    }
                }
                None => skipped += 1,
            }
        }
        skipped
    });

    let mut count = 0;
    while let Ok(reading) = rx.recv().await {
        println!("consumer: {} is {:.1}C", reading.room, reading.celsius);
        count += 1;
    }
    let skipped = producer.join().unwrap();
    println!("consumer: channel closed after {count} readings, {skipped} skipped");
}

spsc::bounded_sync(4) returns a BoundedSyncSender and a BoundedSyncReceiver for a buffer that holds 4 readings. rx.to_async() turns the receiver into a BoundedAsyncReceiver on the same channel. The conversion costs nothing. It's how one channel ends up with a blocking end and an async end. Going the other way, every async handle has a to_sync().

Neither handle is Clone, so the compiler holds you to one producer and one consumer. The async receiver's methods take &mut self for the same reason, which is why rx is declared mut.

The sender moves into the thread and is dropped when the thread's closure returns. That closes the channel. The consumer's recv keeps returning readings until the buffer is empty, then resolves with Err(RecvError::Disconnected) and the while let loop ends. The thread returns its skip count, so join hands it back to the consumer without a second channel. Run it:

$ cargo run
consumer: kitchen is 21.5C
consumer: attic is 31.0C
consumer: hallway is 19.0C
consumer: office is 22.5C
consumer: bedroom is 20.5C
consumer: cellar is 14.5C
consumer: channel closed after 6 readings, 1 skipped

Calling join from inside async code blocks the runtime while it waits. Here that's fine: the loop only ends once the sender is dropped, so the thread has already finished or is just returning.

Watch a slow consumer#

A bounded channel makes the producer wait when the consumer falls behind. On a plain thread that wait is an ordinary blocking call, which is why the parser lives on its own thread: a blocked send there parks the thread and leaves the Tokio runtime free.

To see it, shrink the buffer to 2, make the consumer take 100ms per reading and have both sides print a timestamp. The helper ms rounds the elapsed time to the nearest 100ms so the numbers stay stable from run to run. The producer now counts what it sent and prints what happens to each line. Here is the whole file:

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

const INPUT: &str = "\
kitchen 21.5
attic 31.0
hallway 19.0
garage n/a
office 22.5
bedroom 20.5
cellar 14.5
";

struct Reading {
    room: String,
    celsius: f32,
}

fn parse(line: &str) -> Option<Reading> {
    let (room, value) = line.split_once(' ')?;
    let celsius = value.parse().ok()?;
    Some(Reading { room: room.to_string(), celsius })
}

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

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (tx, rx) = spsc::bounded_sync::<Reading>(2);
    let mut rx = rx.to_async();
    let start = Instant::now();

    let producer = thread::spawn(move || {
        let mut sent = 0;
        for line in INPUT.lines() {
            let Some(reading) = parse(line) else {
                println!("{:>4}ms producer: skipped {line:?}", ms(start));
                continue;
            };
            let room = reading.room.clone();
            if tx.send(reading).is_err() {
                println!("{:>4}ms producer: consumer gone, stopping at {room}", ms(start));
                break;
            }
            println!("{:>4}ms producer: queued {room}", ms(start));
            sent += 1;
        }
        println!("{:>4}ms producer: done, {sent} sent", ms(start));
        sent
    });

    let mut handled = 0;
    while let Ok(reading) = rx.recv().await {
        sleep(Duration::from_millis(100)).await;
        println!("{:>4}ms consumer: {} is {:.1}C", ms(start), reading.room, reading.celsius);
        handled += 1;
    }
    let sent = producer.join().unwrap();
    println!("{:>4}ms consumer: channel closed, {sent} sent, {handled} handled", ms(start));
}

   0ms producer: queued kitchen
   0ms producer: queued attic
   0ms producer: queued hallway
   0ms producer: skipped "garage n/a"
 100ms consumer: kitchen is 21.5C
 100ms producer: queued office
 200ms consumer: attic is 31.0C
 200ms producer: queued bedroom
 300ms consumer: hallway is 19.0C
 300ms producer: queued cellar
 300ms producer: done, 6 sent
 400ms consumer: office is 22.5C
 500ms consumer: bedroom is 20.5C
 600ms consumer: cellar is 14.5C
 600ms consumer: channel closed, 6 sent, 6 handled

At time zero three readings get in: the consumer has already taken kitchen out while attic and hallway fill the two slots. The producer skips the broken line and then blocks on office. From then on it gets one reading in every time the consumer takes one out, so each queued line comes right after a consumer line. The producer finishes at 300ms with two readings still buffered. Dropping its sender doesn't throw them away: the consumer handles both and only then sees the channel close.

If the parser should never wait, use try_send instead. It returns TrySendError::Full with the reading inside when the buffer is full. The thread can then decide whether to retry, drop or log it.

Stop from the consumer side#

So far the producer has closed the channel. The consumer can close it too, by dropping its receiver. Say the consumer should stop at the first room above 30C. Replace the consumer loop and the lines after it with this:

    let mut handled = 0;
    while let Ok(reading) = rx.recv().await {
        sleep(Duration::from_millis(100)).await;
        println!("{:>4}ms consumer: {} is {:.1}C", ms(start), reading.room, reading.celsius);
        handled += 1;
        if reading.celsius > 30.0 {
            println!("{:>4}ms consumer: too hot, stopping", ms(start));
            break;
        }
    }
    drop(rx);
    let sent = producer.join().unwrap();
    println!("{:>4}ms consumer: producer stopped, {sent} sent, {handled} handled", ms(start));

   0ms producer: queued kitchen
   0ms producer: queued attic
   0ms producer: queued hallway
   0ms producer: skipped "garage n/a"
 100ms consumer: kitchen is 21.5C
 100ms producer: queued office
 200ms consumer: attic is 31.0C
 200ms consumer: too hot, stopping
 200ms producer: consumer gone, stopping at bedroom
 200ms producer: done, 4 sent
 200ms consumer: producer stopped, 4 sent, 2 handled

When the consumer breaks out, the producer is blocked sending bedroom. Dropping the receiver wakes it and that send returns Err(SendError::Closed). The producer stops parsing and its thread returns. hallway and office were already in the buffer and are dropped with the receiver, which is why 4 were sent but only 2 handled. A receiver that goes away doesn't drain what's left.

The drop(rx) before join matters. Without it the receiver stays alive until the end of main, the producer stays blocked on a full buffer and join waits forever.

Move readings in batches#

Sending one reading at a time costs one wakeup per reading. When the producer has a pile of items ready at once, the batch API moves them with one synchronization step and one wakeup per batch. send_batch takes a Vec and blocks until every item is in. recv_batch(max) waits for at least one item and then takes whatever is buffered, up to max, without waiting any longer.

This version also builds the channel from the async side and converts the sender with to_sync(), the mirror image of the first step. The consumer starts by pretending to open a connection to wherever the readings go. That takes 100ms and each batch it writes takes another 100ms. Here is the whole file:

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

const INPUT: &str = "\
kitchen 21.5
attic 31.0
hallway 19.0
garage n/a
office 22.5
bedroom 20.5
cellar 14.5
";

struct Reading {
    room: String,
    celsius: f32,
}

fn parse(line: &str) -> Option<Reading> {
    let (room, value) = line.split_once(' ')?;
    let celsius = value.parse().ok()?;
    Some(Reading { room: room.to_string(), celsius })
}

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

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (tx, mut rx) = spsc::bounded_async::<Reading>(4);
    let tx = tx.to_sync();
    let start = Instant::now();

    let producer = thread::spawn(move || {
        let readings: Vec<Reading> = INPUT.lines().filter_map(parse).collect();
        match tx.send_batch(readings) {
            Ok(n) => println!("{:>4}ms producer: all {n} sent", ms(start)),
            Err(e) => println!("{:>4}ms producer: consumer gone, {} unsent", ms(start), e.unsent.len()),
        }
    });

    sleep(Duration::from_millis(100)).await;
    println!("{:>4}ms consumer: connected", ms(start));

    while let Ok(batch) = rx.recv_batch(8).await {
        sleep(Duration::from_millis(100)).await;
        let rows: Vec<String> = batch.iter().map(|r| format!("{}={:.1}", r.room, r.celsius)).collect();
        println!("{:>4}ms consumer: wrote {} readings: {}", ms(start), batch.len(), rows.join(" "));
    }
    println!("{:>4}ms consumer: channel closed", ms(start));
    producer.join().unwrap();
}

 100ms consumer: connected
 100ms producer: all 6 sent
 200ms consumer: wrote 4 readings: kitchen=21.5 attic=31.0 hallway=19.0 office=22.5
 300ms consumer: wrote 2 readings: bedroom=20.5 cellar=14.5
 300ms consumer: channel closed

While the consumer connects, send_batch puts 4 readings into the buffer and blocks with 2 left over. The consumer's first recv_batch(8) takes all 4 in one call. The buffer never holds more than its capacity, so a batch is never bigger than that either, however large max is. Taking them frees the whole buffer, so the producer's last 2 readings go straight in, send_batch returns Ok(6) and the thread ends. The next recv_batch gets those 2. The one after that finds the buffer empty and the sender gone. It returns Err(RecvError::Disconnected).

If the consumer had dropped its receiver partway, send_batch would have returned a SendBatchError whose sent and unsent fields say how far it got, with the unsent readings still inside. The Err arm prints how many were left.

Where to go next#

The channels guide at https://github.com/excsn/fibre/blob/main/channels/README.GUIDE.md covers the rest of the batch API, including the _mut variants that reuse your own Vec and the non-blocking try_send_batch. It also covers spsc::rendezvous for a zero-capacity handoff, the futures::Stream impl on the async receiver and the other channel types for when one producer or one consumer isn't enough.

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