Worker-Thread Patterns

Channels enable structured worker-thread patterns—including Request-Reply (RPC over channels), Pipeline stages, and Fan-Out / Fan-In scatter-gather—without shared mutexes on application state.

Pattern 1: Request-Reply via Embedded One-Shot Reply Channel

Because mpsc channels are unidirectional (Sender $\rightarrow$ Receiver), a caller that sends a command to an actor or worker thread and wants to wait for that specific command's result embeds a reply_tx: mpsc::SyncSender<Response> (or Sender<Response>) directly inside the request struct!

Caller Thread                                  Dedicated Actor / Worker Thread
+-----------------------------------------+    +-------------------------------------+
| 1. Creates `(reply_tx, reply_rx)`       |    |                                     |
| 2. Sends `Request { arg, reply_tx }` ---+--> | 3. Receives `Request` from `cmd_rx` |
| 4. Waits on `reply_rx.recv()`       <---+--- | 5. Computes result & sends on       |
+-----------------------------------------+    |    `req.reply_tx.send(result)`!     |
                                               +-------------------------------------+

use std::collections::HashMap;
use std::sync::mpsc::{self, SyncSender};
use std::thread;

enum KvQuery {
    Set {
        key: String,
        value: i64,
    },
    Get {
        key: String,
        reply_tx: SyncSender<Option<i64>>,
    },
}

fn main() {
    let (cmd_tx, cmd_rx) = mpsc::channel::<KvQuery>();

    // Dedicated state-owning actor thread (owns `HashMap` with ZERO Mutex!):
    let actor = thread::spawn(move || {
        let mut store = HashMap::new();
        for query in cmd_rx {
            match query {
                KvQuery::Set { key, value } => {
                    store.insert(key, value);
                }
                KvQuery::Get { key, reply_tx } => {
                    let _ = reply_tx.send(store.get(&key).copied());
                }
            }
        }
    });

    cmd_tx
        .send(KvQuery::Set {
            key: String::from("balance"),
            value: 500,
        })
        .unwrap();

    // Send a `Get` request with a dedicated reply channel:
    let (reply_tx, reply_rx) = mpsc::sync_channel(1);
    cmd_tx
        .send(KvQuery::Get {
            key: String::from("balance"),
            reply_tx,
        })
        .unwrap();

    println!("actor replied with balance: {:?}", reply_rx.recv().unwrap());

    drop(cmd_tx);
    actor.join().unwrap();
}

actor replied with balance: Some(500)

Pattern 2: Multi-Stage Processing Pipeline

In a pipeline topology, Stage 1 generates or reads items and sends them over chan_1 to Stage 2, which transforms them and sends them over chan_2 to Stage 3. Because each stage owns its local state and drops its outgoing Sender when its incoming Receiver closes, the entire pipeline shuts down in cascade automatically from front to back:

[ Stage 1: Generator ] --(chan_1)--> [ Stage 2: Filter/Map ] --(chan_2)--> [ Stage 3: Sink ]
(Drops `tx1` on exit)                (Exits when `rx1` closes,              (Exits when `rx2` closes)
                                      dropping `tx2`)

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx_raw, rx_raw) = mpsc::sync_channel::<u32>(4);
    let (tx_processed, rx_processed) = mpsc::sync_channel::<String>(4);

    // Stage 1: Emit raw numbers 1..=4
    let stage_1 = thread::spawn(move || {
        for n in 1..=4 {
            tx_raw.send(n).unwrap();
        }
        // `tx_raw` drops here, signaling EOF to Stage 2.
    });

    // Stage 2: Filter even numbers and format them
    let stage_2 = thread::spawn(move || {
        for n in rx_raw {
            if n % 2 == 0 {
                tx_processed.send(format!("item #{n} (squared = {})", n * n)).unwrap();
            }
        }
        // `tx_processed` drops here, signaling EOF to main sink.
    });

    // Stage 3 (Sink): Collect final processed strings
    for output in rx_processed {
        println!("pipeline sink received: {}", output);
    }

    stage_1.join().unwrap();
    stage_2.join().unwrap();
}

pipeline sink received: item #2 (squared = 4)
pipeline sink received: item #4 (squared = 16)