Rust Channels vs Shared State for Concurrent Workflows

Spawning one thread per job ties the thread count directly to the workload. Here, two fixed workers pull four endpoint checks from a bounded queue instead. That small change forces three useful design decisions into the open: who owns submitted work, how results come back, and what makes every worker stop.

The example targets Rust 2024 with rustc and Cargo 1.98.1. It uses OS threads and synchronous channels from the standard library. Tokio and async Rust remain the subject of Article 23.

1. Build a worker pool with a bounded channel

mpsc::sync_channel creates a bounded multi-producer, single-consumer channel. send can place a message immediately while the buffer has room. Once the buffer is full, it blocks the calling thread until the consumer frees a slot. A capacity of 0 creates a rendezvous channel, where sender and receiver must meet. That blocking point is the backpressure boundary: it limits how far production can outrun consumption.

The example accepts the worker count and queue capacity separately.

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct PoolConfig {
    pub workers: usize,
    pub queue_capacity: usize,
}

1.1. Share one receiver among several workers

A standard mpsc channel has one Receiver. The example shares receive access through Arc<Mutex<Receiver<Job>>>. While one worker holds the lock and waits in recv, no other worker can receive from that same receiver. The guard is released at the end of the inner block, so workers can still run check concurrently after taking a job.

let (work_sender, work_receiver) = mpsc::sync_channel::<Job>(config.queue_capacity);
let work_receiver = Arc::new(Mutex::new(work_receiver));
let (result_sender, result_receiver) = mpsc::channel::<(usize, CheckOutcome)>();

for _ in 0..config.workers {
    let work_receiver = Arc::clone(&work_receiver);
    let result_sender = result_sender.clone();
    let checker = Arc::clone(&checker);
    handles.push(thread::spawn(move || {
        loop {
            let job = {
                let receiver = work_receiver
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner());
                receiver.recv()
            };
            let Ok(job) = job else {
                break WorkerExit::WorkChannelDisconnected;
            };
            // Run the check and send the result after releasing the receiver lock.
        }
    }));
}

This is compact and explicit, which suits a teaching example. It also serializes job acquisition through one mutex-protected receiver and makes no fairness guarantee. Queue capacity is not a throughput dial, either. Choose it from the number of waiting jobs the process can afford and the point where producers are allowed to block.

1.2. Mark every place that can block

The example has three blocking boundaries. A producer can wait in send when the bounded work queue is full. Workers wait in recv when it is empty. The coordinating thread can wait on result recv and later on join. The result channel is unbounded, so workers do not block while reporting results. Production code may need a separate capacity policy when results are large or the consumer is slow.

The example submits every job before collecting results. This makes progress because only the work channel is bounded. If both channels were bounded and the producer kept submitting before reading any result, workers could fill the result queue while the producer filled the work queue. Both sides could then wait forever.

2. Use ownership and disconnection for shutdown

Channel shutdown can be represented by ownership rather than a separate stop flag. After every Sender or SyncSender is dropped, the receiver drains buffered messages and then reports disconnection. In the other direction, dropping the Receiver makes later sends return an error.

2.1. Drop the last work sender deliberately

The coordinating thread moves each Endpoint, together with its input position, into a Job. It must release the original sender after submission finishes.

for (index, endpoint) in endpoints.into_iter().enumerate() {
    if work_sender.send(Job { index, endpoint }).is_err() {
        channel_error = Some(MonitorError::WorkChannelDisconnected);
        break;
    }
}
drop(work_sender);

Workers do not hold cloned work senders. Once the coordinator calls drop(work_sender), no work sender remains alive. Each worker drains the queue, observes disconnection from recv, and exits its loop. Unlike sending one sentinel per worker, this scheme does not require shutdown-message counts to match worker counts.

2.2. Drop result senders, then join every worker

Each worker receives a clone of the result sender. The coordinator immediately drops its original copy. The result receiver can therefore observe disconnection after all worker copies disappear. On the success path, the coordinator receives the expected number of results and then joins every handle.

Disconnection signals the end of a channel; it does not prove that thread cleanup has finished. Calling JoinHandle::join establishes that the worker has ended and lets code after the join observe its memory effects. If one join reports a panic, the example still joins every remaining handle before returning WorkerPanicked.

3. Separate completion order from output order

Worker completion order depends on scheduling and check duration. Printing results as they arrive would make output order vary between runs. The example instead sends the original input index with each job and returns that index with the result.

let mut indexed = Vec::with_capacity(expected);
for _ in 0..expected {
    match result_receiver.recv() {
        Ok(result) => indexed.push(result),
        Err(_) => {
            channel_error = Some(MonitorError::ResultChannelDisconnected);
            break;
        }
    }
}

indexed.sort_by_key(|(index, _)| *index);
let outcomes = indexed.into_iter().map(|(_, outcome)| outcome).collect();

Sorting does not change completion order. It restores input order only for presentation. Tests likewise avoid asserting which worker finishes first; they check the final outcomes identifiers and values. This distinction gives deterministic tests without sleeps or timing thresholds.

4. Treat errors and panics as different policies

A failed check, a disconnected channel, and a panic are different events. The example converts a CheckError returned by Checker::check into OutcomeKind::CheckError for that endpoint, then continues with later work. Work-channel or result-channel disconnection ends the whole run with a MonitorError.

4.1. Catch only the panic boundary you intend to recover from

If the checker implementation unwinds, the example wraps only that call in catch_unwind and turns the event into a Panicked outcome.

let kind = match catch_unwind(AssertUnwindSafe(|| checker.check(&job.endpoint))) {
    Ok(Ok(status)) if (200..400).contains(&status) => OutcomeKind::Healthy { status },
    Ok(Ok(status)) => OutcomeKind::Unhealthy { status },
    Ok(Err(error)) => OutcomeKind::CheckError {
        message: error.message,
    },
    Err(_) => OutcomeKind::Panicked,
};

Catching the unwind does not suppress the default panic hook. The hook runs before catch_unwind returns, so a caught panic can still write a message to stderr. Replacing the process-global hook inside each worker merely to quiet output can race with other threads. Hook policy belongs at the application boundary.

The API also avoids exposing the panic payload as a contract. An unwind cannot be caught in a panic=abort build. Nor does converting a panic into one result undo external state changes that happened first. Expected domain failures should still use Result.

5. Choose channels, a mutex, or atomics by the invariant

Message passing is not universally better than shared state. The right primitive depends on who should own changing data, which fields form one invariant, and when observers need a consistent view.

5.1. Use channels to transfer ownership of work

A channel fits when a producer creates a job and one worker consumes that value once. Queue capacity limits waiting work, while sender ownership describes shutdown. The fit is weaker when many workers repeatedly read the same current configuration or update one object in place. Copying every update into messages, or introducing a dedicated owner loop, may add complexity instead of removing it.

5.2. Use `Arc<Mutex<_>>` for multi-field invariants

MutexMetrics updates completed, healthy, unhealthy, check_errors, and panics inside one Summary. The lock guard keeps related field changes in one critical section. When readers need a consistent snapshot of the relationship among those values, one mutex is easier to reason about than several atomic counters.

Do not extend that critical section across a slow check or a blocking channel operation. Doing so raises contention and can create deadlock paths. The example holds the receiver lock for one recv only, and the metrics lock only while changing counters. Poisoning records that another thread panicked while holding the lock; it is not automatic recovery. Code that calls into_inner remains responsible for checking the protected invariant.

5.3. Use atomics for independent telemetry events

Every field in AtomicMetrics is an independent telemetry count. Its fetch_add operations and final load operations use Ordering::Relaxed. Relaxed ordering makes each counter update atomic, but it does not synchronize other data. It cannot publish a result object or create a transactional snapshot across several counters.

The example calls snapshot_after_join only after joining every recording thread, so the final test value is deterministic. The joins provide a separate completion boundary. If code must always observe a live relationship such as completed == healthy + unhealthy + errors + panics, independent atomic counters are the wrong representation. Put the fields behind one lock or send events to a dedicated aggregation owner.

6. Verify the boundaries before choosing the structure

Run these commands from the project directory.

cd examples/article-22-channels-shared-state
cargo fmt --all -- --check
cargo check --all-targets --all-features
cargo clippy --all-targets --all-features -- -D warnings
cargo test --all-features
cargo run --quiet

On Rust 1.98.1 and Cargo 1.98.1, all five commands should exit with code 0. The suite contains 2 library unit tests and 7 integration tests, for 9 tests total. Program output is fixed as follows.

endpoint monitor: workers=2 queue_capacity=2
home: healthy (200)
health: unhealthy (503)
metrics: check-error (fixture timeout)
admin: healthy (204)
summary: completed=4 healthy=2 unhealthy=1 errors=1 panics=0
shutdown: disconnected_workers=2 joined_workers=2

6.1. A conditional selection rule

Start with a bounded channel when work moves from one owner to another and the backlog needs an upper bound. Arc<Mutex<_>> is more direct when several fields change together and readers require one consistent snapshot. A Relaxed atomic can fit an independent counter or flag only when it is not being used to publish other state.

For any of these choices, write down the blocking points and shutdown responsibilities first. The design should reveal who drops the final sender, who drains results, and who joins every handle. Tune worker count and queue capacity from measured load and resource limits. The example's value of 2 is not a production default.

Full source code

The complete runnable source for this article is available in the Chapter 22 project on GitHub.

[Rust Zero to Production 22] Rust Channels vs Shared State for Concurrent Workflows