Multiple-Producer Channels

A multiple-producer channel fans messages from $N$ concurrent producer threads into a single consumer thread (M:1 fan-in topology). In std::sync::mpsc, each producer thread is given a cloned Sender<T> (or SyncSender<T>), and all messages sent by any producer are interleaved safely into the receiver's queue.

FIFO Ordering Guarantees Across Multiple Producers

• Per-producer FIFO order: Messages sent sequentially by the same producer thread (tx1.send(A) followed by tx1.send(B)) are guaranteed to arrive at the Receiver<T> in the exact order A then B.
• Cross-producer interleaving: Messages sent concurrently by different producer threads (tx1 vs. tx2) arrive interleaved in the order their atomic enqueue operations complete at runtime.

Fan-In Topology (3 Producers -> 1 Aggregator Consumer):

  Worker 0 (tx.clone()) --- [ (0, "step A"), (0, "step B") ] ---\
  Worker 1 (tx.clone()) --- [ (1, "step A"), (1, "step B") ] ----+--> [ Receiver ]
  Worker 2 (tx.clone()) --- [ (2, "step A"), (2, "step B") ] ---/
  (Per-worker order `(i, "step A")` before `(i, "step B")` is strictly preserved!)

Typed Enum Messages for Heterogeneous Producers

When different subsystem threads (such as a network listener, a timer thread, and a signal watcher) feed events into a single central event loop, define an enum Event covering all producer message types and share Sender<Event> across all producers:

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

#[derive(Debug)]
enum SystemEvent {
    HttpRequest { client_ip: &'static str, path: &'static str },
    HealthTick { uptime_secs: u64 },
    Shutdown,
}

fn main() {
    let (tx, rx) = mpsc::channel::<SystemEvent>();

    let http_tx = tx.clone();
    let http_worker = thread::spawn(move || {
        http_tx
            .send(SystemEvent::HttpRequest {
                client_ip: "10.0.0.5",
                path: "/api/orders",
            })
            .unwrap();
    });

    let timer_tx = tx.clone();
    let timer_worker = thread::spawn(move || {
        timer_tx
            .send(SystemEvent::HealthTick { uptime_secs: 60 })
            .unwrap();
    });

    http_worker.join().unwrap();
    timer_worker.join().unwrap();
    tx.send(SystemEvent::Shutdown).unwrap();

    for event in rx {
        match event {
            SystemEvent::HttpRequest { client_ip, path } => {
                println!("handled request from {} to {}", client_ip, path);
            }
            SystemEvent::HealthTick { uptime_secs } => {
                println!("health check at uptime {}s", uptime_secs);
            }
            SystemEvent::Shutdown => {
                println!("received Shutdown event; exiting loop");
                break;
            }
        }
    }
}

handled request from 10.0.0.5 to /api/orders
health check at uptime 60s
received Shutdown event; exiting loop