Build a Multithreaded Task Processor
Process tasks across worker threads using Mutex, atomics, and channels in Rust.
Introduction
30 Second Summary
Think about a team of workers sharing a single pile of tasks. Everyone grabs from the same stack, but two people cannot grab the same task at once. This coordination problem is exactly what multithreaded programs solve every day.
In this project, you will build a multithreaded task processor in Rust where 4 worker threads compete to claim tasks from a shared pool. You'll progress from lock-based synchronization to lock-free atomic operations, learning when each approach fits.
What You'll Build
A command-line Rust program that processes 20 simulated tasks across 4 worker threads, all interacting with the same shared data pools to create real contention.
By the end of this project, you'll have:
- A working multithreaded task processor using Arc and Mutex for shared mutable state across threads.
- A lock-free task claiming mechanism using compare_exchange (CAS) with correct memory ordering.
- An atomic progress counter and shutdown flag demonstrating Relaxed and Acquire/Release ordering.
- Secret Mission: Replace shared state with message-passing channels and compare the two concurrency models.
Are there any prerequisites?
You should be comfortable with basic Rust concepts like ownership, borrowing, structs, and closures. You also need the Rust toolchain (rustc and cargo) installed on your machine.
Before We Start
Before We Start...
Set Up the Cargo Project
Every Rust project starts with Cargo, Rust's built-in package manager and build tool.
Cargo handles compiling your code, managing dependencies, and running your program. In this step, you'll create a fresh Cargo project and confirm everything compiles correctly before adding any threading code.
In this step, get ready to:
- Create a new Cargo project.
- Verify the project compiles and runs.
- Open src/main.rs in your editor.
Create a new Cargo project
The cargo new command scaffolds a complete Rust project for you, including the source directory, a manifest file, and a default main.rs.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Terminal and press Enter to open it.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your Cargo project and move into it by running these commands:
cargo new task_processor
cd task_processor- Confirm the project files exist by running:
ls srcYou should see main.rs listed in the output.
What did cargo new create?
Cargo created a task_processor directory containing:
- Cargo.toml is your project manifest. It lists your project name, version, and dependencies.
- src/main.rs is the entry point for your program. It contains a default "Hello, world!" function.
Seeing 'command not found'?
If cargo is not recognized, your Rust toolchain is not installed or not in your PATH.
- Close and reopen your terminal, then try again. Some installations require a fresh shell session.
- If the problem persists, reinstall Rust from the official site at rustup.rs.
help me troubleshoot why cargo is not found in my terminal
Verify the project compiles and runs
Before writing any threading code, confirm that Cargo can compile and execute the default program successfully.
- Build and run the project by running:
cargo runYou should see Cargo compile your project, then print Hello, world! at the end of the output.
What does cargo run do?
This single command does two things. First, it compiles your Rust source code into an executable binary. Then it immediately runs that binary.
The first run takes a moment because Cargo compiles from scratch. Subsequent runs are faster because Cargo only recompiles files that have changed.
Compilation errors?
- Make sure you are inside the task_processor directory. Cargo looks for Cargo.toml in the current directory.
- Run pwd to check your current location. The path should end with task_processor.
help me fix my cargo run error
Open src/main.rs in your editor
Now that the project compiles, open the main source file so you're ready to start writing multithreaded code in the next step.
- Open the task_processor folder in your code editor.
- In the file explorer sidebar, expand the src folder and click main.rs to open it.
You should see the default code that Cargo generated for you:
fn main() {
println!("Hello, world!");
}Why start with this file?
Every Rust program starts execution in the main() function inside src/main.rs. This is the only file you'll edit throughout the entire project.
Your project is ready to go. Next up, you'll replace this default code with a multithreaded task processor that spawns worker threads and shares data between them.
Spawn Threads and Share a Task Pool
Your task processor needs multiple worker threads pulling from the same pool of tasks. In most languages, sharing data between threads is a runtime footgun.
Rust takes a different approach. The compiler itself prevents data races at compile time using its ownership system. In this step, you will see that protection in action, then solve it with Arc and Mutex.
In this step, get ready to:
- Attempt to share a Vec across threads and observe the compiler error.
- Fix the ownership problem with Arc and Mutex.
- Run the program and see all 4 workers pulling from the same task pool.
Attempt to share data across threads
When you spawn a thread in Rust, the closure you pass must own its data. If you try to move the same variable into multiple threads, Rust's ownership model stops you.
- Replace all the code in src/main.rs with this naive attempt:
use std::thread;
fn main() {
// Create 20 task strings
let tasks: Vec<String> = (0..20)
.map(|i| format!("task-{}", i))
.collect();
let mut handles = vec![];
// Try to share `tasks` with 4 worker threads
for worker_id in 0..4 {
let handle = thread::spawn(move || {
println!("[worker-{}] sees {} tasks", worker_id, tasks.len());
});
handles.push(handle);
}
for handle in handles {
handle.join().unwrap();
}
}- Try to compile this by running:
cargo runThe compiler refuses to build your code. You will see error E0382 with the message "use of moved value" pointing at the tasks variable.
Why does this fail?
The first loop iteration moves tasks into the closure because of the move keyword. On the second iteration, tasks no longer exists in the main thread's scope.
Rust enforces single ownership. A Vec<String> does not implement Copy, so moving it into one thread permanently transfers ownership. The other 3 threads can never have it.
Fix the ownership problem with Arc and Mutex
The fix requires two tools. Arc (Atomic Reference Counting) lets multiple threads share ownership of the same data. Mutex (Mutual Exclusion) ensures only one thread can access the data at a time.
- Replace all the code in src/main.rs with the imports and shared data structures:
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;
fn main() {
let num_workers = 4;
let total_tasks = 20;
// Task pool wrapped in Arc<Mutex<Vec>> for shared mutable access
let tasks: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(
(0..total_tasks)
.map(|i| format!("task-{}", i))
.collect(),
));
// Shared results pool
let results: Arc<Mutex<HashMap<usize, String>>> = Arc::new(Mutex::new(HashMap::new()));
let mut handles = vec![];
}What does this code do?
Arc::new(Mutex::new(...)) wraps your data in two layers. The Mutex guards the inner data with a lock. The Arc lets multiple threads hold their own pointer to that lock.
The tasks pool is a Vec<String> that workers will pop from. The results pool is a HashMap<usize, String> that workers write their output to.
- Add the worker spawning loop immediately after let mut handles = vec![]; by pasting this code:
for worker_id in 0..num_workers {
let tasks = Arc::clone(&tasks);
let results = Arc::clone(&results);
let handle = thread::spawn(move || {
loop {
let task_opt = {
let mut pool = tasks.lock().unwrap();
pool.pop()
};
match task_opt {
Some(task) => {
thread::sleep(Duration::from_millis(50));
let result = format!("worker-{}: processed '{}'", worker_id, task);
println!("[worker-{}] {}", worker_id, result);
let mut res = results.lock().unwrap();
res.insert(worker_id, result);
}
None => break,
}
}
});
handles.push(handle);
}What does this code do?
Arc::clone(&tasks) creates a new reference-counted pointer to the same Mutex-guarded data. It does not clone the Vec itself. Each thread gets its own Arc handle that it can move into its closure.
Inside the loop, tasks.lock().unwrap() acquires the lock and returns a MutexGuard. The worker pops a task, then the guard is dropped (releasing the lock) before doing any work. When no tasks remain, pop() returns None and the worker breaks out of the loop.
- Add the join loop and final report after the worker spawning loop by pasting this code:
for handle in handles {
handle.join().unwrap();
}
let final_results = results.lock().unwrap();
println!("\nResults collected: {}", final_results.len());What does this code do?
handle.join().unwrap() blocks the main thread until each worker finishes. Once all workers are done, there is no more contention, so locking results to print the summary is safe and instantaneous.
- Save src/main.rs.
Seeing a syntax error?
Make sure the join loop and final report are inside the fn main() block, after the worker for loop, and that you have a closing } for main at the very end.
help me debug a syntax error in my Rust main.rs file with Arc, Mutex, and thread::spawn
Run the program and observe the output
- Run the program by entering this command:
cargo runYou should see output showing each worker claiming and processing tasks. The order will be non-deterministic because threads run in parallel.
What should I see?
Each line shows a worker ID and the task it processed. You will see all 20 tasks completed, spread across 4 workers. The final line reports how many results were collected in the shared HashMap.
Because the HashMap key is worker_id, only the last result per worker is stored (4 total). You will fix this in a later step when you switch to lock-free task claiming with atomic indices.
Program hangs or no output?
If the program compiles but seems stuck, check that you have the None => break arm in the match statement. Without it, workers loop forever after the pool is empty.
help me figure out why my Rust thread pool program is hanging
✔️ Awesome, I've got everything!
Great. Double check you have saved src/main.rs and that cargo run produces output showing workers processing tasks.
ⓧ I'd like to double check the full code
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;
fn main() {
let num_workers = 4;
let total_tasks = 20;
// Task pool wrapped in Arc<Mutex<Vec>> for shared mutable access
let tasks: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(
(0..total_tasks)
.map(|i| format!("task-{}", i))
.collect(),
));
// Shared results pool
let results: Arc<Mutex<HashMap<usize, String>>> = Arc::new(Mutex::new(HashMap::new()));
let mut handles = vec![];
for worker_id in 0..num_workers {
let tasks = Arc::clone(&tasks);
let results = Arc::clone(&results);
let handle = thread::spawn(move || {
loop {
let task_opt = {
let mut pool = tasks.lock().unwrap();
pool.pop()
};
match task_opt {
Some(task) => {
thread::sleep(Duration::from_millis(50));
let result = format!("worker-{}: processed '{}'", worker_id, task);
println!("[worker-{}] {}", worker_id, result);
let mut res = results.lock().unwrap();
res.insert(worker_id, result);
}
None => break,
}
}
});
handles.push(handle);
}
for handle in handles {
handle.join().unwrap();
}
let final_results = results.lock().unwrap();
println!("\nResults collected: {}", final_results.len());
}Your workers are now pulling from a shared pool and writing results. Next up, you will minimize lock time to dramatically reduce contention and speed up the program.
Minimize Lock Time and Understand Contention
Your workers are now sharing a task pool and results collection. But there is a subtle performance problem lurking in how long each thread holds its lock.
When one thread holds a Mutex lock, every other thread that needs that lock is forced to wait. The longer the lock is held, the more threads pile up idle. In this step, you will make the lock scoping explicit, measure the impact with timing, and see the damage that holding a lock too long causes.
In this step, get ready to:
- Explicitly scope your locks to minimize critical sections.
- Add timing to measure total execution speed.
- Observe the serialization effect of holding locks too long.
Scope your locks to minimize critical sections
A critical section is the region of code that runs while a lock is held. The smaller the critical section, the less time other threads spend waiting.
- In src/main.rs, find the thread::spawn closure (the loop { ... } block inside each worker).
- Replace the entire closure body with the following code:
let handle = thread::spawn(move || {
loop {
// Grab task with minimal lock scope
let task_opt = {
let mut pool = tasks.lock().unwrap();
pool.pop()
}; // Lock released immediately
match task_opt {
Some(task) => {
// Work happens OUTSIDE the lock
thread::sleep(Duration::from_millis(50));
let result = format!("worker-{}: processed '{}'", worker_id, task);
println!("[worker-{}] {}", worker_id, result);
// Brief lock only to store result
{
let mut res = results.lock().unwrap();
res.insert(worker_id, result);
} // Lock dropped immediately
}
None => break,
}
}
});What does this code do?
This refactored loop makes the critical section pattern explicit.
- The tasks lock is held for exactly one operation: pool.pop(). Then the lock drops at the closing brace.
- The expensive work (thread::sleep simulating processing) happens with NO lock held. All 4 workers can process their tasks in parallel.
- The results lock is wrapped in its own { } scope block. This makes the lock drop immediately after the insert, rather than living until the end of the Some(task) arm.
Why use explicit scope blocks?
In Rust, a MutexGuard drops (and releases the lock) when it goes out of scope. By wrapping the lock in its own block, you control exactly when it drops. Without the block, the guard lives until the end of the enclosing scope, potentially holding the lock longer than necessary.
- Also remove the comments above the tasks and results declarations (the lines that say // Task pool wrapped in Arc<Mutex<Vec>>... and // Shared results pool). The new inline comments inside the loop explain the pattern more clearly.
- Save src/main.rs.
Seeing a compile error about moved values?
Make sure result is created before the results lock scope block. If you create the format string inside the { } block, the borrow checker may complain about moved values.
Also check that the println! is before the results lock block, not inside it.
Help me fix my lock scope refactoring in Rust
Add timing to measure contention
To prove that minimizing lock scope actually matters, you need to measure how long the program takes to run. Rust's Instant type captures a precise timestamp.
- In src/main.rs, find the import line at the top of the file:
use std::time::Duration;- Replace it with this to also import Instant:
use std::time::{Duration, Instant};- Find the line let mut handles = vec![]; and add the following line directly above it:
let start = Instant::now();- Now find the let final_results = results.lock().unwrap(); line near the bottom. Replace the remaining lines from there to the end of main with this:
let elapsed = start.elapsed();
let final_results = results.lock().unwrap();
println!("\nResults collected: {}", final_results.len());
println!("Time elapsed: {:?}", elapsed);What does this code do?
- Instant::now() captures the current time before threads start.
- start.elapsed() returns a Duration representing how much time has passed since start was captured.
- The :? format specifier uses the Debug trait to print the duration with full precision (e.g. 283.5ms).
- Save src/main.rs and run the program:
cargo run- You should see each worker claiming tasks, followed by a summary showing the time elapsed. With 4 workers processing 20 tasks of 50ms each, expect a total time around 250-350ms.
Seeing much longer times?
If total time is close to 1000ms (1 second), your work simulation is probably still inside a lock scope. Check that thread::sleep is outside both the tasks lock and the results lock.
Help me figure out why my multithreaded program is running slowly
Observe the serialization effect of holding locks too long
To truly understand why critical section size matters, you will intentionally break the pattern. Moving the sleep inside the tasks lock forces all threads to wait while one thread "works."
- In src/main.rs, find the task-grabbing block inside the loop:
// Grab task with minimal lock scope
let task_opt = {
let mut pool = tasks.lock().unwrap();
pool.pop()
}; // Lock released immediately- Temporarily replace it with this version that holds the lock during the sleep:
// BAD: holding the lock during work simulation!
let task_opt = {
let mut pool = tasks.lock().unwrap();
let task = pool.pop();
thread::sleep(Duration::from_millis(50));
task
}; // Lock held for 50ms - other threads blocked!- Save and run the program again:
cargo runWhat should I see?
The total time jumps to around 1000ms or more. That is roughly 4x slower than before. Even though you have 4 threads, they are forced to wait in line for the lock. This is called serialization. The threads still run, but they cannot make progress in parallel because one thread monopolizes the lock.
You may also notice that one worker processes most or all of the tasks. The other workers spend their time blocked on the lock.
Why didn't this deadlock?
A deadlock requires two or more locks acquired in different orders. Here there is only one lock involved (the tasks lock), so threads simply wait their turn. The program still finishes. It just finishes slowly because the parallelism is destroyed.
- Revert the change. Replace the bad version with the original minimal-lock version:
// Grab task with minimal lock scope
let task_opt = {
let mut pool = tasks.lock().unwrap();
pool.pop()
}; // Lock released immediately- Save src/main.rs and run cargo run one more time to confirm you are back to the fast version (around 250-350ms).
Still seeing slow times after reverting?
Double check that thread::sleep is inside the Some(task) arm (outside both locks), not inside the task-grabbing block.
Help me revert my lock scope experiment
✔️ Awesome, I've got everything!
Great. Double check you have saved src/main.rs with the fast (reverted) version.
ⓧ I'd like to double check the full code
Here is the complete src/main.rs at this point:
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};
fn main() {
let num_workers = 4;
let total_tasks = 20;
let tasks: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(
(0..total_tasks)
.map(|i| format!("task-{}", i))
.collect(),
));
let results: Arc<Mutex<HashMap<usize, String>>> = Arc::new(Mutex::new(HashMap::new()));
let start = Instant::now();
let mut handles = vec![];
for worker_id in 0..num_workers {
let tasks = Arc::clone(&tasks);
let results = Arc::clone(&results);
let handle = thread::spawn(move || {
loop {
// Grab task with minimal lock scope
let task_opt = {
let mut pool = tasks.lock().unwrap();
pool.pop()
}; // Lock released immediately
match task_opt {
Some(task) => {
// Work happens OUTSIDE the lock
thread::sleep(Duration::from_millis(50));
let result = format!("worker-{}: processed '{}'", worker_id, task);
println!("[worker-{}] {}", worker_id, result);
// Brief lock only to store result
{
let mut res = results.lock().unwrap();
res.insert(worker_id, result);
} // Lock dropped immediately
}
None => break,
}
}
});
handles.push(handle);
}
for handle in handles {
handle.join().unwrap();
}
let elapsed = start.elapsed();
let final_results = results.lock().unwrap();
println!("\nResults collected: {}", final_results.len());
println!("Time elapsed: {:?}", elapsed);
}You have proven that critical section size directly controls parallelism. Next up, you will eliminate the Mutex entirely for simple counters and flags by switching to atomic operations.
Add Lock-Free Progress Tracking with Atomics
Your workers now hold locks for the shortest time possible. But the progress tracking itself still relies on indirectly counting through the Mutex-protected results collection.
In this step, you will replace that pattern with atomic operations. Atomics let multiple threads read and write a single value without any lock at all. The hardware guarantees each operation completes as one indivisible unit.
In this step, get ready to:
- Add an AtomicUsize counter for lock-free progress tracking.
- Add an AtomicBool shutdown flag with Acquire/Release memory ordering.
- Print per-task progress without ever touching a Mutex.
Add an atomic progress counter
An AtomicUsize is an integer that multiple threads can increment simultaneously without a lock. The CPU handles the synchronization at the hardware level, making it far cheaper than locking a Mutex.
- In src/main.rs, find the import lines at the top of the file:
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};- Add a new import line for atomic types. Insert use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; between the HashMap import and the Arc, Mutex import, so the top of your file looks like this:
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};- Now find the line where you declare results. Add the atomic counter declaration directly below it:
// Atomic progress counter (lock-free)
let completed = Arc::new(AtomicUsize::new(0));What does this code do?
AtomicUsize::new(0) creates an atomic unsigned integer starting at zero. Wrapping it in Arc lets you share it across threads, just like you did with the Mutex-wrapped collections.
The difference is that reading or writing this value never blocks other threads. There is no lock to acquire or release.
- Inside the for worker_id in 0..num_workers loop, find the existing clone lines. Add a clone for completed after the results clone:
let tasks = Arc::clone(&tasks);
let results = Arc::clone(&results);
let completed = Arc::clone(&completed);- Inside the worker's Some(task) arm, find the old print line:
println!("[worker-{}] {}", worker_id, result);- Delete that line. After the results lock block (the res.insert section), add the atomic increment and the new progress print:
// Atomic increment (Relaxed: only atomicity needed)
let done = completed.fetch_add(1, Ordering::Relaxed) + 1;
println!(
"[worker-{}] finished task ({}/{})",
worker_id, done, total_tasks
);Why is Relaxed ordering sufficient here?
fetch_add atomically increments the counter and returns the previous value. The Ordering::Relaxed argument tells the CPU you only need atomicity, not any ordering guarantee relative to other memory operations.
This is safe because no other data depends on seeing the counter at a specific value. You just need the increment itself to be indivisible. If thread A increments from 5 to 6, thread B will never see a torn or partial value.
- Save src/main.rs.
Seeing a type error on fetch_add?
Make sure you imported Ordering in the atomic import line at the top of the file. Without it, Rust cannot resolve Ordering::Relaxed.
Also confirm that completed is cloned inside the for loop before the thread::spawn call. Without the clone, the move closure cannot capture it.
Help me fix an atomic type error in my Rust worker thread.
Add a shutdown flag with Acquire/Release ordering
A shutdown flag lets you signal all workers to stop from the main thread. An AtomicBool is perfect for this. But unlike the counter, the shutdown flag needs stronger memory ordering to work correctly.
- In src/main.rs, find the completed declaration you just added. Add the shutdown flag directly below it:
// Shutdown flag with Acquire/Release ordering
let shutdown = Arc::new(AtomicBool::new(false));- In the for loop, add a clone for shutdown after the completed clone:
let completed = Arc::clone(&completed);
let shutdown = Arc::clone(&shutdown);- Inside the worker's loop block, add a shutdown check as the very first thing before the task lock. Replace the existing comment at the top of the loop with:
// Check shutdown flag (Acquire pairs with Release on setter)
if shutdown.load(Ordering::Acquire) {
break;
}What is the Acquire/Release pair?
When the main thread stores true into the shutdown flag with Ordering::Release, it "publishes" all memory writes that happened before that store. Think of it as flushing a buffer.
When a worker loads the flag with Ordering::Acquire, it "subscribes" to those published writes. The worker is guaranteed to see everything the setter wrote before flipping the flag.
In this program, the shutdown flag is not set during normal operation. It is wired up so you could trigger it from the main thread if needed (for example, after a timeout). The pattern is ready to use without any Mutex.
- Save src/main.rs.
Compiler complaining about unused variables?
If you see a warning about shutdown being unused, that is expected for now. The shutdown flag is never set to true in this version. Workers will exit naturally when the task pool is empty. The flag is infrastructure for future use.
Help me understand a Rust unused variable warning for my AtomicBool.
Update the summary and run the program
The workers now track their own progress atomically. The last piece is reading that atomic counter in the summary report after all threads finish.
- At the end of the worker closure (after the loop block but still inside the thread::spawn closure), add a completion message:
println!("[worker-{}] done.", worker_id);- Now find the summary section at the bottom of main (after the join loop). Replace it with:
let elapsed = start.elapsed();
let final_results = results.lock().unwrap();
let total_completed = completed.load(Ordering::SeqCst);
println!("\n--- Summary ---");
println!("Total tasks completed: {}/{}", total_completed, total_tasks);
println!("Results in collection: {}", final_results.len());
println!("Time elapsed: {:?}", elapsed);Why SeqCst for the final read?
After all threads have joined, there is no contention. Using Ordering::SeqCst (sequentially consistent) for the final read guarantees you see the absolute latest value. It is the strongest ordering and costs nothing here because no threads are racing.
- Also remove the old comments inside the worker loop (// Grab task with minimal lock scope, // Lock released immediately, // Work happens OUTSIDE the lock, and // Brief lock only to store result). The new atomic comments replace them.
- Save src/main.rs.
- Run the program by running:
cargo runYou should see each worker printing progress like [worker-2] finished task (7/20) as tasks complete. The counter increments atomically across all four threads without any lock.
Why is no Mutex needed for the counter or the flag?
A Mutex protects complex data structures (like a HashMap) where multiple fields must change together. An atomic handles a single value where the CPU can guarantee indivisibility in one instruction.
Your counter only needs "add 1" and your flag only needs "flip to true". Both are single-value operations. Using a Mutex for them would add unnecessary blocking.
Seeing numbers out of order in the output?
That is expected. Thread scheduling is non-deterministic. Worker 0 might print (3/20) after worker 1 prints (5/20). What matters is that the final summary reports exactly 20/20 completed. No tasks are lost or double-counted.
Help me understand why my atomic counter output is out of order.
✔️ Awesome, I've got everything!
Great. Double check you have saved your file before moving on.
ⓧ I'd like to double check the full code
Here is the complete src/main.rs at this point:
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};
fn main() {
let num_workers = 4;
let total_tasks = 20;
let tasks: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(
(0..total_tasks)
.map(|i| format!("task-{}", i))
.collect(),
));
let results: Arc<Mutex<HashMap<usize, String>>> = Arc::new(Mutex::new(HashMap::new()));
// Atomic progress counter (lock-free)
let completed = Arc::new(AtomicUsize::new(0));
// Shutdown flag with Acquire/Release ordering
let shutdown = Arc::new(AtomicBool::new(false));
let start = Instant::now();
let mut handles = vec![];
for worker_id in 0..num_workers {
let tasks = Arc::clone(&tasks);
let results = Arc::clone(&results);
let completed = Arc::clone(&completed);
let shutdown = Arc::clone(&shutdown);
let handle = thread::spawn(move || {
loop {
// Check shutdown flag (Acquire pairs with Release on setter)
if shutdown.load(Ordering::Acquire) {
break;
}
let task_opt = {
let mut pool = tasks.lock().unwrap();
pool.pop()
};
match task_opt {
Some(task) => {
thread::sleep(Duration::from_millis(50));
let result = format!("worker-{}: processed '{}'", worker_id, task);
{
let mut res = results.lock().unwrap();
res.insert(worker_id, result);
}
// Atomic increment (Relaxed: only atomicity needed)
let done = completed.fetch_add(1, Ordering::Relaxed) + 1;
println!(
"[worker-{}] finished task ({}/{})",
worker_id, done, total_tasks
);
}
None => break,
}
}
println!("[worker-{}] done.", worker_id);
});
handles.push(handle);
}
for handle in handles {
handle.join().unwrap();
}
let elapsed = start.elapsed();
let final_results = results.lock().unwrap();
let total_completed = completed.load(Ordering::SeqCst);
println!("\n--- Summary ---");
println!("Total tasks completed: {}/{}", total_completed, total_tasks);
println!("Results in collection: {}", final_results.len());
println!("Time elapsed: {:?}", elapsed);
}Your program now uses two different atomic patterns: a lock-free counter with Relaxed ordering and a shutdown flag with Acquire/Release. Next up, you will replace the Mutex-based task claiming entirely with a lock-free compare-and-swap operation.
Implement Lock-Free Task Claiming with compare_exchange
Your atomics are handling the counter and shutdown flag without any locks. But the biggest Mutex bottleneck remains: every time a worker grabs a task, it locks the entire pool.
In this step, you'll eliminate that lock entirely. Instead of popping from a shared Vec, workers will compete to claim the next available index using compare_exchange. This is a compare-and-swap (CAS) operation: a worker reads the current index, then atomically tries to increment it. If another thread got there first, the swap fails and the worker retries.
In this step, get ready to:
- Refactor the task pool into an immutable Arc<Vec<String>> that needs no Mutex.
- Implement lock-free task claiming with AtomicUsize and compare_exchange.
- Run the program and observe CAS contention in action.
Replace the task pool Mutex with an immutable Vec and atomic index
The current design locks the entire Vec to pop a task. But if workers never modify the Vec, they only need to agree on which index to read next. An atomic counter can track that index, and the Vec itself becomes read-only.
- In src/main.rs, find the current task pool declaration:
let tasks: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(
(0..total_tasks)
.map(|i| format!("task-{}", i))
.collect(),
));
let results: Arc<Mutex<HashMap<usize, String>>> = Arc::new(Mutex::new(HashMap::new()));
// Atomic progress counter (lock-free)
let completed = Arc::new(AtomicUsize::new(0));
// Shutdown flag with Acquire/Release ordering
let shutdown = Arc::new(AtomicBool::new(false));- Replace that entire block (from let tasks down through let shutdown) with the following:
// Shared task data (read-only after creation, so no Mutex needed)
let tasks: Arc<Vec<String>> = Arc::new(
(0..total_tasks)
.map(|i| format!("task-{}", i))
.collect(),
);
// Atomic index for lock-free task claiming
let next_task = Arc::new(AtomicUsize::new(0));
// Shared results pool - threads push completed work here (needs Mutex)
let results: Arc<Mutex<HashMap<usize, String>>> = Arc::new(Mutex::new(HashMap::new()));
// Atomic progress counter (lock-free, no Mutex needed)
let completed = Arc::new(AtomicUsize::new(0));
// Shutdown flag with Acquire/Release ordering
let shutdown = Arc::new(AtomicBool::new(false));What changed here?
The task Vec lost its Mutex. It is now Arc<Vec<String>>. Once created, no thread ever modifies it. Multiple threads can read from it simultaneously without any lock.
The new next_task AtomicUsize starts at 0 and acts as a cursor. Workers will atomically increment it to "claim" a task index. No Mutex needed for this coordination.
Rewrite the worker loop with compare_exchange
The CAS (compare-and-swap) pattern works like this: a worker loads the current index, then tries to atomically swap it to index + 1. If the value hasn't changed since the load, the swap succeeds and the worker "owns" that index. If another thread incremented it first, the swap fails and the worker retries.
- Find the entire for worker_id in 0..num_workers loop (including its body and the handles.push(handle) at the end).
- Replace the full loop with this new version:
for worker_id in 0..num_workers {
let tasks = Arc::clone(&tasks);
let next_task = Arc::clone(&next_task);
let results = Arc::clone(&results);
let completed = Arc::clone(&completed);
let shutdown = Arc::clone(&shutdown);
let handle = thread::spawn(move || {
let mut local_count = 0;
loop {
// Check shutdown flag (Acquire pairs with Release on the setter)
if shutdown.load(Ordering::Acquire) {
break;
}
// Lock-free task claiming with compare_exchange
let index = next_task.load(Ordering::Acquire);
if index >= tasks.len() {
break; // No more tasks
}
// Try to atomically claim this task index
match next_task.compare_exchange(
index,
index + 1,
Ordering::AcqRel,
Ordering::Acquire,
) {
Ok(_) => {
// Successfully claimed task at this index
// Read directly from the shared Vec (no lock needed, it is immutable)
let task = &tasks[index];
// Simulate processing OUTSIDE any lock
thread::sleep(Duration::from_millis(50));
let result = format!("worker-{}: processed '{}'", worker_id, task);
// Brief lock only to store the result
{
let mut res = results.lock().unwrap();
res.insert(index, result);
} // Lock dropped immediately
// Atomic progress update (Relaxed: we only need atomicity here)
let done = completed.fetch_add(1, Ordering::Relaxed) + 1;
println!(
"[worker-{}] finished task {} ({}/{})",
worker_id, index, done, total_tasks
);
local_count += 1;
}
Err(_) => {
// Another thread claimed this index, retry immediately
continue;
}
}
}
println!("[worker-{}] done. Processed {} tasks.", worker_id, local_count);
});
handles.push(handle);
}How does compare_exchange work here?
The worker loads the current value of next_task into index. Then compare_exchange(index, index + 1, ...) says: "If the value is still index, atomically set it to index + 1." If it succeeds, this worker owns that index. If it fails, another worker already incremented it.
The orderings matter. Ordering::AcqRel on success means the operation acts as both an Acquire (so subsequent reads of tasks[index] see consistent data) and a Release (so other threads see the incremented value). Ordering::Acquire on failure ensures the worker sees the latest value when retrying.
Why does the results HashMap still use a Mutex?
HashMap is not an atomic data structure. Multiple threads inserting keys simultaneously would corrupt it. The Mutex lock here is brief (just one insert call) and only held after all the heavy processing is done. This is the critical section minimization pattern from earlier.
Update the summary output and run the program
The final section of main needs a small update to print sample results. This confirms that each task index maps to the correct worker output.
- Find the section at the bottom of main() that starts with for handle in handles:
for handle in handles {
handle.join().unwrap();
}
let elapsed = start.elapsed();
let final_results = results.lock().unwrap();
let total_completed = completed.load(Ordering::SeqCst);
println!("\n--- Summary ---");
println!("Total tasks completed: {}/{}", total_completed, total_tasks);
println!("Results in collection: {}", final_results.len());
println!("Time elapsed: {:?}", elapsed);- Replace that entire section with the following:
// Wait for all workers to finish
for handle in handles {
handle.join().unwrap();
}
let elapsed = start.elapsed();
// Final report - safe to read without contention now
let final_results = results.lock().unwrap();
let total_completed = completed.load(Ordering::SeqCst);
println!("\n--- Summary ---");
println!("Total tasks completed: {}/{}", total_completed, total_tasks);
println!("Results in collection: {}", final_results.len());
println!("Time elapsed: {:?}", elapsed);
println!("\nSample results:");
for i in 0..3usize {
if let Some(result) = final_results.get(&i) {
println!(" {}", result);
}
}What does the sample results section do?
It prints the first three results by index (0, 1, 2). Since tasks are now claimed by index rather than popped from a stack, the HashMap keys correspond to task numbers. This confirms that the CAS-based claiming produces correct, duplicate-free results.
- Save src/main.rs.
- Run the program by executing this command:
cargo run- You should see output showing each worker claiming specific task indices, a progress counter incrementing to 20/20, and a summary with sample results showing which worker processed tasks 0, 1, and 2.
What should I see?
Each line now includes the task index (e.g. [worker-2] finished task 7 (8/20)). Notice that task indices are unique across all workers. No two workers claim the same index. This proves the CAS loop is working correctly.
At the end, each worker reports how many tasks it processed. The distribution should be roughly even (around 5 per worker), though slight imbalances happen due to thread scheduling.
Seeing a compilation error?
If you see an error about tasks.lock() or pool.pop(), you still have remnants of the old Mutex-based loop. Make sure you replaced the entire for worker_id loop body.
If you see a type mismatch on tasks, double check that the declaration uses Arc<Vec<String>> (no Mutex wrapper) and that the Mutex::new() wrapping was removed from the Vec creation.
help me debug a compile error in my compare_exchange task processor
✔️ Awesome, I've got everything!
Great. Double check you've saved your file.
ⓧ I'd like to double check the full code
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};
fn main() {
let num_workers = 4;
let total_tasks = 20;
// Shared task data (read-only after creation, so no Mutex needed)
let tasks: Arc<Vec<String>> = Arc::new(
(0..total_tasks)
.map(|i| format!("task-{}", i))
.collect(),
);
// Atomic index for lock-free task claiming
let next_task = Arc::new(AtomicUsize::new(0));
// Shared results pool - threads push completed work here (needs Mutex)
let results: Arc<Mutex<HashMap<usize, String>>> = Arc::new(Mutex::new(HashMap::new()));
// Atomic progress counter (lock-free, no Mutex needed)
let completed = Arc::new(AtomicUsize::new(0));
// Shutdown flag with Acquire/Release ordering
let shutdown = Arc::new(AtomicBool::new(false));
let start = Instant::now();
let mut handles = vec![];
for worker_id in 0..num_workers {
let tasks = Arc::clone(&tasks);
let next_task = Arc::clone(&next_task);
let results = Arc::clone(&results);
let completed = Arc::clone(&completed);
let shutdown = Arc::clone(&shutdown);
let handle = thread::spawn(move || {
let mut local_count = 0;
loop {
// Check shutdown flag (Acquire pairs with Release on the setter)
if shutdown.load(Ordering::Acquire) {
break;
}
// Lock-free task claiming with compare_exchange
let index = next_task.load(Ordering::Acquire);
if index >= tasks.len() {
break; // No more tasks
}
// Try to atomically claim this task index
match next_task.compare_exchange(
index,
index + 1,
Ordering::AcqRel,
Ordering::Acquire,
) {
Ok(_) => {
// Successfully claimed task at this index
// Read directly from the shared Vec (no lock needed, it is immutable)
let task = &tasks[index];
// Simulate processing OUTSIDE any lock
thread::sleep(Duration::from_millis(50));
let result = format!("worker-{}: processed '{}'", worker_id, task);
// Brief lock only to store the result
{
let mut res = results.lock().unwrap();
res.insert(index, result);
} // Lock dropped immediately
// Atomic progress update (Relaxed: we only need atomicity here)
let done = completed.fetch_add(1, Ordering::Relaxed) + 1;
println!(
"[worker-{}] finished task {} ({}/{})",
worker_id, index, done, total_tasks
);
local_count += 1;
}
Err(_) => {
// Another thread claimed this index, retry immediately
continue;
}
}
}
println!("[worker-{}] done. Processed {} tasks.", worker_id, local_count);
});
handles.push(handle);
}
// Wait for all workers to finish
for handle in handles {
handle.join().unwrap();
}
let elapsed = start.elapsed();
// Final report - safe to read without contention now
let final_results = results.lock().unwrap();
let total_completed = completed.load(Ordering::SeqCst);
println!("\n--- Summary ---");
println!("Total tasks completed: {}/{}", total_completed, total_tasks);
println!("Results in collection: {}", final_results.len());
println!("Time elapsed: {:?}", elapsed);
println!("\nSample results:");
for i in 0..3usize {
if let Some(result) = final_results.get(&i) {
println!(" {}", result);
}
}
}Secret mission
Replace Shared Results with Channels
Your task processor works, but every worker still locks a Mutex to store results. What if workers never needed to lock anything at all? Replace the shared HashMap with message passing and see how channels change the concurrency model.
Clean Up Your Resources
Clean Up Your Resources
Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally, so there are no ongoing costs.
Resources you used:
- The task_processor Cargo project folder on your Desktop.
Keep everything running
No action needed. Choose this if you want to keep experimenting with the multithreaded task processor or extend it further.
- Your task_processor folder stays on your Desktop and you can run cargo run anytime.
Pause - I'll come back to this later
Since this is a local Rust project, there is nothing to pause. Your files sit on disk and consume no resources when you are not running the program.
- Close your code editor and terminal. Your project will be waiting when you return.
Delete - I don't want to use this again
Remove the project folder entirely to free up disk space.
- Delete the task_processor folder by running this command from your Desktop:
macOS
cd ~/Desktop
rm -rf task_processorWindows
cd ~\Desktop
rmdir /s /q task_processor- Confirm the folder is gone by running ls (macOS) or dir (Windows) and verifying task_processor no longer appears.
Nice Work!
Nice Work!
You just built a fully functional multithreaded task processor in Rust. Four worker threads competed for tasks from a shared pool, and you controlled every layer of synchronization from coarse-grained locks down to lock-free atomic operations.
You've learned how to:
- Built a multithreaded task processor using Arc and Mutex for shared mutable state, then minimized critical sections to reduce lock contention.
- Used AtomicUsize with Relaxed ordering for a lock-free counter and AtomicBool with Acquire/Release ordering for a shutdown flag.
- Implemented lock-free task claiming with compare_exchange (CAS), letting workers atomically claim tasks without ever locking the task pool.
- Secret Mission: Replaced shared state with message passing via mpsc channels, eliminating all Mutex contention in worker threads.
Ready to quiz yourself?