# Feed a Tokio task from a parser thread

**For:** Rust developers who mix plain threads with Tokio.

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.

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-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.

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

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

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

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

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

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

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

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

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