What are the trade-offs between `futures::stream::StreamExt::ready_chunks` and `chunks` for batch processing?

chunks unconditionally buffers exactly N items before emitting a batch, introducing artificial latency and backpressure dependency, while ready_chunks emits immediately available items up to N without waiting, trading batch size consistency for lower latency and reduced blocking. The fundamental difference is waiting behavior: chunks blocks until the batch is full, ready_chunks takes what's ready now.

The Batching Problem

use futures::stream::{self, StreamExt};
 
async fn batching_problem() {
    // Scenario: You're processing a stream of items and want to batch them
    // for efficiency (e.g., database inserts, API calls)
    
    let stream = stream::iter(vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
    
    // The question: How do you batch these items?
    // 
    // Option 1: chunks(N) - Wait until we have N items
    // Option 2: ready_chunks(N) - Take up to N items that are ready now
    
    // The choice affects:
    // - Latency: How long until we process the first item?
    // - Throughput: How efficiently do we process?
    // - Backpressure: How does slow downstream affect upstream?
}

Batching trades latency for efficiency; the method choice determines the trade-off balance.

chunks: Complete Batching

use futures::stream::{self, StreamExt};
 
async fn chunks_example() {
    let stream = stream::iter(vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
    
    // chunks(3) waits until exactly 3 items are available
    // Then emits them as a Vec
    let batches: Vec<Vec<i32>> = stream.chunks(3).collect().await;
    
    println!("Batches: {:?}", batches);
    // Output: [[1, 2, 3], [4, 5, 6], [7, 8, 9], [10]]
    //          ^^^^^^^^^^  ^^^^^^^^^^  ^^^^^^^^^^  ^^^^
    //          3 items     3 items     3 items     Remaining (less than 3)
    
    // Key behaviors:
    // 1. Each batch has exactly 3 items (except possibly the last)
    // 2. Waits for 3 items before emitting
    // 3. Last batch contains remaining items when stream ends
}

chunks(N) collects exactly N items before emitting, guaranteeing batch size (except last).

ready_chunks: Immediate Availability

use futures::stream::{self, StreamExt};
use std::time::Duration;
use tokio::time::sleep;
 
async fn ready_chunks_example() {
    // ready_chunks emits items that are immediately available
    // It doesn't wait to fill the batch
    
    let stream = stream::iter(vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
    
    // ready_chunks(3) takes UP TO 3 ready items
    // If only 1 is ready, it emits just 1
    let batches: Vec<Vec<i32>> = stream.ready_chunks(3).collect().await;
    
    println!("Batches: {:?}", batches);
    // Output depends on what's immediately available
    // Often: [[1, 2, 3], [4, 5, 6], [7, 8, 9], [10]]
    // But could be different based on async scheduling
    
    // Key behaviors:
    // 1. Each batch has AT MOST 3 items
    // 2. Doesn't wait for more items if some are ready
    // 3. Emits immediately when items are available
}

ready_chunks(N) emits up to N immediately available items, never waiting to fill the batch.

The Key Difference: Waiting Behavior

use futures::stream::{self, StreamExt};
use tokio::time::{sleep, Duration, interval};
 
async fn waiting_behavior() {
    // Stream that produces items slowly
    let slow_stream = stream::unfold(0, |mut count| async move {
        sleep(Duration::from_millis(100)).await;
        count += 1;
        Some((count, count))
    });
    
    // Take 10 items
    let mut stream = slow_stream.take(10);
    
    // With chunks(3):
    // - Waits for 3 items (300ms minimum per batch)
    // - Higher latency for each batch
    // - But guarantees 3 items per batch
    
    // With ready_chunks(3):
    // - Emits as soon as items are ready
    // - Lower latency (could emit 1 item immediately)
    // - Smaller batches possible
    
    // Simulate async consumer with different speeds
    tokio::spawn(async move {
        let start = tokio::time::Instant::now();
        
        // Using chunks
        let mut chunks_stream = stream::iter(0..10).chunks(3);
        while let Some(batch) = chunks_stream.next().await {
            let elapsed = start.elapsed();
            println!("chunks batch at {:?}: {:?}", elapsed, batch);
        }
    });
    
    tokio::spawn(async move {
        let start = tokio::time::Instant::now();
        
        // Using ready_chunks
        let mut ready_stream = stream::iter(0..10).ready_chunks(3);
        while let Some(batch) = ready_stream.next().await {
            let elapsed = start.elapsed();
            println!("ready_chunks batch at {:?}: {:?}", elapsed, batch);
        }
    });
}

chunks waits; ready_chunks takes what's available now.

Backpressure Implications

use futures::stream::{self, StreamExt};
use tokio::sync::mpsc;
 
async fn backpressure_demo() {
    let (tx, rx) = mpsc::channel::<i32>(100);
    
    // Producer: Fast
    tokio::spawn(async move {
        for i in 0..100 {
            tx.send(i).await.unwrap();
        }
    });
    
    // Consumer: Slow
    // chunks creates backpressure to producer
    // Because chunks(10) won't emit until 10 items arrive
    // The channel fills up, slowing the producer
    
    let mut chunks_stream = rx.map(|x| async move {
        tokio::time::sleep(Duration::from_millis(50)).await;
        x
    }).buffer_unordered(10).chunks(10);
    
    // With ready_chunks(10):
    // Consumer takes what's ready immediately
    // Less backpressure - empties channel faster
    // But potentially smaller batches
    
    let mut ready_stream = rx.map(|x| async move {
        tokio::time::sleep(Duration::from_millis(50)).await;
        x
    }).buffer_unordered(10).ready_chunks(10);
}

chunks creates stronger backpressure because it holds items waiting for a full batch.

Latency Comparison

use futures::stream::{self, StreamExt};
use tokio::time::{Duration, Instant};
 
async fn latency_comparison() {
    // Stream with varying availability
    let stream = stream::iter(0..20);
    
    // chunks(5): Latency = time to receive 5 items
    // First batch emitted only after 5 items accumulated
    let start = Instant::now();
    let _chunks_batches: Vec<Vec<i32>> = stream.clone().chunks(5).collect().await;
    let chunks_time = start.elapsed();
    println!("chunks time: {:?}", chunks_time);
    
    // ready_chunks(5): Latency = time to receive first available item
    // First batch could be smaller than 5 if items arrive slowly
    let start = Instant::now();
    let _ready_batches: Vec<Vec<i32>> = stream.ready_chunks(5).collect().await;
    let ready_time = start.elapsed();
    println!("ready_chunks time: {:?}", ready_time);
    
    // ready_chunks typically has lower latency because it doesn't wait
    // But chunks provides more consistent batch sizes
}

ready_chunks has lower latency; chunks has more consistent batch sizes.

When to Use chunks

use futures::stream::{self, StreamExt};
 
async fn when_chunks() {
    // Use chunks when:
    
    // 1. Batch size consistency is critical
    //    (e.g., database bulk inserts benefit from fixed sizes)
    async fn bulk_insert(items: Vec<Record>) -> Result<(), Error> {
        // Database is optimized for fixed-size batches
        // chunks(100) ensures exactly 100 items per insert (usually)
        // This can be more efficient than variable-sized inserts
        Ok(())
    }
    
    // 2. Downstream processing expects complete batches
    //    (e.g., API calls with rate limits)
    async fn rate_limited_api(items: Vec<Request>) -> Result<(), Error> {
        // Each API call has overhead
        // Waiting for full batches amortizes overhead
        Ok(())
    }
    
    // 3. You want predictable memory usage
    //    chunks(N) guarantees at most N items buffered
    let stream = stream::iter(0..1000);
    let batched = stream.chunks(50);
    // Exactly 50 items buffered per batch (max)
    // Predictable memory footprint
    
    // 4. Backpressure is desired to slow upstream
    //    If downstream is slow, chunks will buffer
    //    Eventually slowing the producer
    
    // Example: Database bulk insert
    let records: Vec<Record> = fetch_records().await;
    let mut batches = stream::iter(records).chunks(100);
    
    while let Some(batch) = batches.next().await {
        database_insert_batch(&batch).await.unwrap();
        // Guaranteed ~100 items per insert (efficient)
    }
}
 
struct Record;
async fn fetch_records() -> Vec<Record> { vec![] }
async fn database_insert_batch(_: &[Record]) -> Result<(), Error> { Ok(()) }
struct Error;

Use chunks when batch size consistency and efficiency matter more than latency.

When to Use ready_chunks

use futures::stream::{self, StreamExt};
use tokio::time::{interval, Duration};
 
async fn when_ready_chunks() {
    // Use ready_chunks when:
    
    // 1. Low latency is critical
    //    (e.g., user-facing requests, real-time processing)
    async fn process_user_request(items: Vec<Request>) {
        // User is waiting - don't delay for more items
        // Process what's ready immediately
    }
    
    // 2. Stream has unpredictable timing
    //    (e.g., network events, user input)
    let user_events = stream_user_events(); // Variable timing
    let mut batched = user_events.ready_chunks(10);
    // Emits when events arrive, doesn't wait for 10
    
    // 3. You want to reduce backpressure
    //    ready_chunks empties buffers faster
    //    Less memory pressure, faster flow-through
    
    // 4. Processing can handle variable batch sizes
    //    (e.g., best-effort batching, opportunistic aggregation)
    
    // Example: Real-time event processing
    let mut events = stream_events().ready_chunks(50);
    
    while let Some(batch) = events.next().await {
        // Process available events immediately
        // Don't wait for 50 if only 10 are ready
        // User sees results faster
        
        let results = process_batch(&batch).await;
        send_results(results).await;
    }
}
 
struct Request;
async fn stream_user_events() -> impl futures::Stream<Item = i32> { stream::iter(0..10) }
async fn stream_events() -> impl futures::Stream<Item = i32> { stream::iter(0..10) }
async fn process_batch(_: &[i32]) -> Vec<i32> { vec![] }
async fn send_results(_: Vec<i32>) {}

Use ready_chunks when latency matters more than batch size consistency.

Combining with Timeouts

use futures::stream::{self, StreamExt};
use tokio::time::{timeout, Duration};
 
async fn combined_with_timeout() {
    let stream = stream::iter(0..20);
    
    // Pattern: chunks with timeout
    // Wait for N items OR timeout, whichever comes first
    // This isn't built-in, but can be composed
    
    // ready_chunks is particularly useful with async timing
    // because it respects async readiness
    
    // Example: Process items, but don't wait forever
    let mut stream = stream::iter(0..20);
    
    loop {
        let batch = timeout(
            Duration::from_millis(100),
            stream_ready_batch(&mut stream, 5)
        ).await;
        
        match batch {
            Ok(Some(items)) => {
                println!("Got batch: {:?}", items);
                // Process batch
            }
            Ok(None) => break, // Stream ended
            Err(_) => {
                println!("Timeout - no items available");
                // Handle timeout
            }
        }
    }
}
 
async fn stream_ready_batch<S>(stream: &mut S, max_size: usize) -> Option<Vec<S::Item>>
where
    S: futures::Stream + Unpin,
{
    // Helper to get ready items
    // ready_chunks essentially does this internally
    Some(stream.ready_chunks(max_size).next().await?)
}

Combine batching strategies with timeouts for responsive systems.

Batch Size Guarantees

use futures::stream::{self, StreamExt};
 
async fn batch_size_guarantees() {
    let stream = stream::iter(0..11);
    
    // chunks(5): Guarantees
    // - Each batch (except last) has exactly 5 items
    // - Last batch has remaining (1 item)
    // - Never emits empty batches
    let chunks_batches: Vec<Vec<i32>> = stream.clone().chunks(5).collect().await;
    assert_eq!(chunks_batches.len(), 3); // 5, 5, 1
    assert_eq!(chunks_batches[0].len(), 5);
    assert_eq!(chunks_batches[1].len(), 5);
    assert_eq!(chunks_batches[2].len(), 1);
    
    // ready_chunks(5): Guarantees
    // - Each batch has AT MOST 5 items
    // - Could be 1, 2, 3, 4, or 5 items
    // - Never emits empty batches
    // - Depends on async scheduling
    let ready_batches: Vec<Vec<i32>> = stream.ready_chunks(5).collect().await;
    // ready_batches could be [[0,1,2,3,4], [5,6,7,8,9], [10]]
    // Or [[0], [1,2,3], [4,5,6,7], [8,9,10]]
    // Depends on what's "ready" at poll time
    
    // For synchronous streams (iter), ready_chunks often behaves like chunks
    // The difference appears with truly async sources
}

chunks guarantees batch size; ready_chunks guarantees maximum size.

Memory Implications

use futures::stream::{self, StreamExt};
 
async fn memory_implications() {
    // chunks(N): Buffers up to N items
    // Memory = N items (bounded)
    let stream = stream::iter(0..1000000);
    let mut chunks_stream = stream.chunks(100);
    // At most 100 items in memory per batch
    
    // ready_chunks(N): Also buffers up to N items
    // But might release smaller batches sooner
    let stream = stream::iter(0..1000000);
    let mut ready_stream = stream.ready_chunks(100);
    // Similar memory bounds
    
    // Key difference: When the buffer is released
    // chunks: Releases after N accumulated
    // ready_chunks: Releases when items available
    
    // For backpressure-sensitive systems:
    // chunks can cause upstream to slow down more
    // Because it holds items longer
}

Both have similar memory bounds, but different release timing.

Practical Example: Database Batching

use futures::stream::{self, StreamExt};
use tokio::time::{sleep, Duration};
 
async fn database_batching_example() {
    // Scenario: Insert records into a database
    // Trade-off: Batch size vs latency
    
    let records = fetch_records_stream().await;
    
    // Approach 1: chunks (batch-oriented)
    // Better throughput, higher latency per record
    let mut batched = records.clone().chunks(100);
    
    while let Some(batch) = batched.next().await {
        // Each insert has 100 records (usually)
        // Fewer API calls, more efficient for database
        insert_batch(&batch).await;
        println!("Inserted {} records", batch.len());
    }
    
    // Approach 2: ready_chunks (latency-oriented)
    // Lower latency, variable batch size
    let mut ready_batched = records.ready_chunks(100);
    
    while let Some(batch) = ready_batched.next().await {
        // Insert available records immediately
        // May have fewer than 100, but records processed sooner
        insert_batch(&batch).await;
        println!("Inserted {} records (variable batch)", batch.len());
    }
    
    // Choose based on requirements:
    // - chunks: Background jobs, analytics, non-time-sensitive
    // - ready_chunks: User-facing, real-time, latency-sensitive
}
 
async fn fetch_records_stream() -> impl futures::Stream<Item = Record> {
    stream::iter((0..500).map(|i| Record { id: i }))
}
async fn insert_batch(_: &[Record]) {}
#[derive(Clone)]
struct Record { id: i32 }

Choose based on whether throughput or latency is the priority.

Practical Example: Event Processing

use futures::stream::{self, StreamExt};
use tokio::sync::mpsc;
use tokio::time::{interval, Duration};
 
async fn event_processing_example() {
    let (tx, mut rx) = mpsc::channel::<Event>(1000);
    
    // Simulate events arriving at variable rates
    tokio::spawn(async move {
        let mut count = 0;
        for burst in 0..5 {
            // Burst of events
            for _ in 0..20 {
                tx.send(Event { id: count }).await.unwrap();
                count += 1;
            }
            // Pause between bursts
            sleep(Duration::from_millis(100)).await;
        }
    });
    
    // With chunks(10):
    // - First burst: ~2 batches of 10
    // - But waits for 10 items
    // - 100ms gap might cause waiting
    
    // With ready_chunks(10):
    // - Takes available items immediately
    // - Might get batch of 20 from first burst (limited to 10)
    // - Then immediately available
    
    let mut events = rx.ready_chunks(10);
    
    while let Some(batch) = events.next().await {
        println!("Processing {} events", batch.len());
        // Process immediately - lower latency
    }
}
 
struct Event { id: i32 }

Event processing benefits from ready_chunks for immediate responsiveness.

Combining with Other Operators

use futures::stream::{self, StreamExt};
use tokio::time::{timeout, Duration};
 
async fn combined_operators() {
    // ready_chunks after buffer: Accumulate then emit ready
    let stream = stream::iter(0..100);
    let batched = stream
        .buffer(10)  // Buffer up to 10
        .ready_chunks(5);  // Then emit up to 5 ready
    
    // ready_chunks after map: Batch transformed results
    let stream = stream::iter(0..100);
    let batched = stream
        .then(|x| async move { x * 2 })
        .ready_chunks(10);
    
    // chunks after filter: Batch only matching items
    let stream = stream::iter(0..100);
    let batched = stream
        .filter(|x| async move { x % 2 == 0 })
        .chunks(5);  // Batches of 5 even numbers
    
    // ready_chunks with timeout for hybrid behavior
    let stream = stream::iter(0..100);
    let batched = stream
        .ready_chunks(10)
        .map(|batch| async move {
            // Process batch
            batch
        });
    
    // The key: ready_chunks works naturally with async streams
    // Because it respects async poll semantics
}

ready_chunks integrates with other stream operators naturally.

Performance Characteristics

use futures::stream::{self, StreamExt};
 
async fn performance_comparison() {
    // chunks(N):
    // - Accumulates exactly N items in buffer
    // - Single allocation per batch
    // - Predictable memory: N items buffered
    // - Latency: O(N * item_arrival_time)
    
    // ready_chunks(N):
    // - Checks what's immediately available
    // - Allocates for up to N items
    // - Memory: Up to N items buffered
    // - Latency: O(1) for available items
    
    // Both are O(N) space complexity
    // chunks has predictable batch sizes
    // ready_chunks has unpredictable batch sizes but lower latency
    
    // CPU overhead:
    // Both have minimal overhead - just buffer management
    // The real difference is in latency vs throughput trade-off
}

Overhead is similar; the difference is latency vs throughput trade-off.

Summary Table

fn summary() {
    // | Characteristic       | chunks(N)          | ready_chunks(N)    |
    // |----------------------|--------------------|--------------------|
    // | Batch size           | Exactly N (mostly) | Up to N            |
    // | Latency              | Higher (waits)     | Lower (immediate)  |
    // | Backpressure         | Stronger           | Weaker             |
    // | Memory usage         | N items            | Up to N items      |
    // | Throughput           | Higher (full batch)| Variable           |
    // | Empty batches        | Never              | Never              |
    
    // | Use chunks when...         | Use ready_chunks when...     |
    // |----------------------------|------------------------------|
    // | Batch size matters          | Latency matters              |
    // | Downstream needs full batch | Processing can handle small  |
    // | Backpressure desired       | Fast flow-through needed     |
    // | Database bulk operations   | Real-time event processing   |
    // | Rate-limited APIs          | User-facing responses        |
}

Synthesis

Quick reference:

use futures::stream::{self, StreamExt};
 
let stream = stream::iter(0..100);
 
// chunks(N): Wait for exactly N items
let batches = stream.clone().chunks(10);
// Guarantees: batch.len() == 10 (except last)
// Latency: Higher (waits for accumulation)
// Use for: Database inserts, API calls, batch jobs
 
// ready_chunks(N): Take up to N ready items
let batches = stream.ready_chunks(10);
// Guarantees: batch.len() <= 10
// Latency: Lower (emits immediately)
// Use for: Event processing, user requests, real-time
 
// Performance note: Both buffer up to N items
// The difference is WHEN they emit, not HOW MUCH memory they use

Key insight: chunks and ready_chunks represent two different batching philosophies. chunks(N) is the "accumulator" approach: collect exactly N items before emitting, prioritizing batch size consistency and throughput. This is ideal when downstream operations benefit from predictable batch sizes—database bulk inserts have optimal sizes, API rate limits work better with consistent payloads, and memory allocation is predictable. ready_chunks(N) is the "opportunistic" approach: take up to N items that are immediately available at poll time, prioritizing latency over batch consistency. This is ideal when responsiveness matters more than efficiency—user-facing requests shouldn't wait for batch accumulation, real-time event processing needs immediate action, and backpressure should be minimized. The memory footprint is similar (both buffer up to N items), but the timing differs: chunks introduces artificial latency proportional to N and item arrival rate, while ready_chunks has essentially zero additional latency. Use chunks for batch-oriented processing where efficiency dominates; use ready_chunks for latency-sensitive processing where responsiveness dominates. Neither is universally better—choose based on whether your system optimizes for throughput or latency.

What are the trade-offs between futures::stream::StreamExt::ready_chunks and chunks for batch processing?