★ Pinned backend #Rust #Async #Tokio #Performance

Rust async stream processing: walking a 4 GB event log without blocking the runtime

Rayhan Kabir @rayhan

Systems engineer · Rust · 1 week ago

↑ 186 👍 62 ❤️ 19 💡 28

We ingest a rolling 4 GB event log across hundreds of thousands of lines, and the first cut blocked a worker on every 8 MB page fetch. Here is the async stream version that keeps the runtime hot end to end — patch, review comments and all.

The streaming ingest, before and after

The current shape, then the reviewed patch. Lines carrying review comments expand — click one to read the discussion where it was written.

As it ships today

src/ingest.rs read-only preview

use futures::stream::{self, StreamExt};
use tokio::io::{BufReader, AsyncBufReadExt};

pub async fn ingest(path: &str) -> Result<(), Error> {
    let file = File::open(path).await?;
    let lines = BufReader::new(file).lines();

    // Cap in-flight pages; the decoder runs as its own task.
    let events = stream::iter(lines)
        .filter_map(|line| async move { line.ok() })
        .buffer_unordered(8);

    events.for_each(handle).await;

    Ok(())
}

A 4 GB rolling log, hundreds of thousands of lines, one worker parked on every 8 MB page fetch.

Reviewed patch

src/ingest.rs

Replace the blocking page loop with an ordered async stream. Concurrency is capped at 8 to keep backpressure on our side of the socket.

                                
                                use tokio::fs::File;
                            
                                use tokio::io::{BufReader, AsyncBufReadExt};
                            
                                @@ -8,6 +8,11 @@ pub async fn ingest(path: &str) {
                            
                                    -
                                    let mut cursor = 0;
                                    💬 2
                                    — has review comments, activate to read
                                RKRayhan Kabir review
This loop held the worker hostage: every `blocking_fetch` parked the thread for ~200ms while the socket ran cold between pages.
PMPriya Menon 4h ago
That cold-socket gap is exactly the 140ms → 18ms shape we saw on the dashboard. Same disease, different runtime.

                                -
                                loop {
                            
                                -
                                let bytes = blocking_fetch(cursor, PAGE_SIZE)?;
                            
                                -
                                for line in bytes.lines() { handle(line)?; }
                            
                                -
                                cursor += PAGE_SIZE;
                            
                                -
                                }
                            
                                    +
                                    let file = File::open(path).await?;
                                    💬 1
                                    — has review comments, activate to read
                                TATanzil Ahmed review
Opening through tokio keeps the open() itself non-blocking — worth a module-doc note, because `std::fs::File` compiles fine here and quietly reintroduces the stall.

                                +
                                let lines = BufReader::new(file).lines();
                            
                                +
                                let events = stream::iter(lines)
                            
                                +
                                .filter_map(|line| async move { line.ok() })
                            
                                    +
                                    .buffer_unordered(8);
                                    💬 3
                                    — has review comments, activate to read
                                RKRayhan Kabir review
Eight is not magic: it is core count minus headroom for the decoder task. Sweep 4 → 32 in the bench harness; throughput plateaus at 8 on our volume.
DODaniel Okafor 2h ago
Plateau-at-8 matches our queue worker findings. The cap also protects you when someone runs this on a 4-core CI box.
TATanzil Ahmed 1h ago
Then the constant deserves a name and this comment — sending a patch.

                                +
                                events.for_each(handle).await;

The conversation

Priya Menon 5h ago

The before/after trace is worth pinning on its own: worker busy-time went from 41% to 94% once the fetch pages stopped stalling the scheduler.
Kenji Sato 3h ago

If you wrap this in join with the decoder stage, the two overlap end to end. Benchmarked 1.4× on a 2-core box.
Aisha Rahman 1h ago

Bookmarking this as the review example for our onboarding doc: real patch, real line comments, real reasoning about backpressure.

Back to the feed