Blocking Receives

A blocking receive (rx.recv(), rx.recv_timeout(dur), or iterating over rx) parks the calling consumer thread in the operating system scheduler whenever the channel queue is empty, waking the thread automatically as soon as a producer sends a message or the channel disconnects.

Comparing Blocking Receive Idioms on Receiver<T>

Consumer Thread State During `for msg in &rx`:

  [ Queue Non-Empty ] ---> Pops front message immediately in user space (`0` syscalls!)
  [ Queue Empty, Senders Alive ] ---> Parks thread via OS futex (`0%` CPU usage while waiting)
  [ Queue Empty, 0 Senders Alive ] -> `Iterator::next` returns `None`, exiting `for` loop!

Periodic Heartbeat Loops with recv_timeout

When a worker thread needs to block waiting for incoming commands and run periodic maintenance (such as flushing a batch buffer, checking a health timer, or logging status) whenever no command arrives within a window, rx.recv_timeout(Duration) handles both without spinning:

use std::sync::mpsc::{self, RecvTimeoutError};
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::channel::<&'static str>();

    let producer = thread::spawn(move || {
        tx.send("job-1").unwrap();
        thread::sleep(Duration::from_millis(15));
        tx.send("job-2").unwrap();
        // `tx` drops here, disconnecting the channel.
    });

    let mut timeouts_observed = 0;
    loop {
        match rx.recv_timeout(Duration::from_millis(5)) {
            Ok(job) => println!("processed {:?}", job),
            Err(RecvTimeoutError::Timeout) => {
                timeouts_observed += 1;
            }
            Err(RecvTimeoutError::Disconnected) => {
                println!(
                    "channel disconnected cleanly (observed {} idle timeouts)",
                    timeouts_observed > 0
                );
                break;
            }
        }
    }

    producer.join().unwrap();
}

processed "job-1"
processed "job-2"
channel disconnected cleanly (observed true idle timeouts)

Batching Optimization: recv() + try_iter()

In high-throughput I/O consumers (such as a database batch writer or log flusher), waking up and issuing a separate disk write per message is inefficient. A classic optimization is calling blocking rx.recv() to sleep until the first item arrives, and then immediately chaining non-blocking rx.try_iter().take(MAX_BATCH - 1) to scoop up any additional items already sitting in the queue before flushing the entire batch in one operation:

use std::sync::mpsc;

fn receive_next_batch<T>(rx: &mpsc::Receiver<T>, max_batch: usize) -> Option<Vec<T>> {
    // 1. Block until at least ONE item arrives (or return `None` if disconnected):
    let first = rx.recv().ok()?;

    // 2. Non-blockingly drain up to `max_batch - 1` additional queued items:
    let mut batch = Vec::with_capacity(max_batch);
    batch.push(first);
    batch.extend(rx.try_iter().take(max_batch - 1));
    Some(batch)
}

fn main() {
    let (tx, rx) = mpsc::channel::<u32>();
    for n in 1..=5 {
        tx.send(n).unwrap();
    }
    drop(tx);

    while let Some(batch) = receive_next_batch(&rx, 3) {
        println!("flushing batch of {} items: {:?}", batch.len(), batch);
    }
}

flushing batch of 3 items: [1, 2, 3]
flushing batch of 2 items: [4, 5]