Index tagging: the pattern that makes concurrency testable

Build a worker pool. Compute the Collatz stopping time of every job, and return the results in input order.

pub fn parallel_map_collatz(jobs: Vec<u64>, workers: usize) -> Vec<u32>

Collatz stopping time: how many steps to reach 1, halving evens and doing 3n + 1 on odds. 1 takes 0 steps, 2 takes 1, 3 takes 7, 27 takes 111. All jobs are >= 1. Empty input gives []; treat workers == 0 as 1.

The shape is a real pool, not a static split: one job queue, workers threads pulling from it, results coming back on a second channel.

The tests run the same jobs at workers = 1, 4, 16 and demand identical output. That is the assertion the whole problem exists to satisfy.

The problem this pattern solves

Jobs take wildly different amounts of time — 77031 takes 350 steps, 1 takes zero. So the results arrive in an order nobody chose, and it changes from run to run. Sorting cannot help you: the answer is a sequence, and [0, 1, 7, 111] sorted is still [0, 1, 7, 111] only by luck.

The answer is to stop hoping the order survives and carry it in the data:

1. tag each job with its input index before it goes on the queue;
2. send (index, result) back;
3. write into out[index] — a pre-sized vec![0; n] — instead of pushing.

Completion order becomes irrelevant, because nothing depends on it any more.

This is the reusable idea of the whole track, and the honest answer to “how do I test concurrent code”: design the observable result to be order-independent rather than hoping the scheduler cooperates. It is also a real engineering habit — the same trick makes retries, partial failures and out-of-order network replies tractable.

Sharing a Receiver

The starter’s error:

error[E0382]: use of moved value: `job_rx`
   = note: move occurs because `job_rx` has type `Receiver<(usize, u64)>`,
           which does not implement the `Copy` trait

mpsc is multi-producer, single-consumer. Sender is Clone; Receiver deliberately is not. So the std way to build a pool is to make the single consumer shared:

let queue = Arc::new(Mutex::new(job_rx));

Each worker locks, takes one job, unlocks. That works because a Mutex makes the single receiver safely reachable from many threads — sharing ownership (Arc) and serialising access (Mutex), the two orthogonal concerns again.

A true multi-consumer channel would be neater, and std has one — but std::sync::mpmc is still nightly-only (tracking issue #126840), which is why crossbeam-channel exists and why every std worker-pool example you will read looks like this one.

The block scope that matters most

let job = {
    let guard = queue.lock().expect("poisoned");
    guard.recv()
};                       // <- guard dropped HERE
match job { ... }        // work happens with the lock released

Get this wrong and everything still compiles, all the tests still pass, and your pool is completely serial — because a worker holding the queue lock while it computes prevents every other worker from taking a job. This is the most common worker-pool bug in any language, it produces no diagnostic, and the only symptom is that adding threads does not help.

Write the block. Then look at it and say why it is there.

Shutdown

Two shutdowns to get right, both by ownership:

• drop(job_tx) after queueing, so recv() eventually returns Err(RecvError) and workers can break instead of blocking forever;
• drop(res_tx) in the main thread after the spawn loop, so for (i, steps) in res_rx terminates once every worker’s clone is gone.

Forget either and the program hangs with no error message at all. That failure gets a problem of its own, next.

type_complexity, and the right fix

Arc<Mutex<mpsc::Receiver<(usize, u64)>>> is close to the limit clippy will tolerate in a signature, and pool types get worse from here. The fix is a type alias, not an #[allow]:

type JobQueue = Arc<Mutex<Receiver<(usize, u64)>>>;

Naming the thing improves the code independently of the lint, which is the test of whether a lint was worth listening to.