Arc<Mutex>

Guides

Rust walkthroughs

← Back to all guides

How do I work with Barrier for Thread Coordination in Rust?

Walkthrough

Barrier is a synchronization primitive that allows multiple threads to synchronize at a specific point in their execution. When a thread reaches a barrier, it blocks until all threads have reached the same barrier point. Once all threads arrive, they are all released simultaneously.

Key concepts:

  • Barrier point — A synchronization point where threads wait
  • Count — Number of threads that must arrive before release
  • BarrierWaitResult — Returned by wait(), check if this thread was the "leader"
  • Reusable — Barriers can be reused after all threads pass through

When to use Barrier:

  • Parallel computations with phases
  • Bulk-synchronous parallel algorithms
  • Coordinated thread startup or shutdown
  • Multi-stage parallel processing
  • Benchmarking parallel code sections

When NOT to use Barrier:

  • Producer-consumer patterns (use channels)
  • Event signaling (use Condvar or channels)
  • One-time events (use Once or OnceLock)

Code Examples

Basic Barrier Usage

use std::sync::{Arc, Barrier};
use std::thread;
 
fn main() {
    // Create a barrier for 3 threads
    let barrier = Arc::new(Barrier::new(3));
    let mut handles = vec![];
    
    for i in 0..3 {
        let barrier = Arc::clone(&barrier);
        
        let handle = thread::spawn(move || {
            println!("Thread {} before barrier", i);
            
            // Wait at barrier
            barrier.wait();
            
            println!("Thread {} after barrier", i);
        });
        
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
}

Barrier with Leader Detection

use std::sync::{Arc, Barrier};
use std::thread;
 
fn main() {
    let barrier = Arc::new(Barrier::new(4));
    let mut handles = vec![];
    
    for i in 0..4 {
        let barrier = Arc::clone(&barrier);
        
        let handle = thread::spawn(move || {
            println!("Thread {} working...", i);
            thread::sleep(std::time::Duration::from_millis(100 * (i + 1)));
            
            let result = barrier.wait();
            
            // One thread becomes the "leader"
            if result.is_leader() {
                println!("Thread {} is the leader!", i);
            }
            
            println!("Thread {} continues", i);
        });
        
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
}

Multi-Phase Parallel Computation

use std::sync::{Arc, Barrier};
use std::thread;
 
fn main() {
    let num_threads = 4;
    let barrier = Arc::new(Barrier::new(num_threads));
    let mut handles = vec![];
    
    for i in 0..num_threads {
        let barrier = Arc::clone(&barrier);
        
        let handle = thread::spawn(move || {
            // Phase 1: Initialize
            println!("Thread {}: Phase 1 - Initializing", i);
            thread::sleep(std::time::Duration::from_millis(50));
            barrier.wait();
            
            // Phase 2: Process
            println!("Thread {}: Phase 2 - Processing", i);
            thread::sleep(std::time::Duration::from_millis(100));
            barrier.wait();
            
            // Phase 3: Finalize
            println!("Thread {}: Phase 3 - Finalizing", i);
            thread::sleep(std::time::Duration::from_millis(30));
            barrier.wait();
            
            println!("Thread {}: Complete!", i);
        });
        
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
}

Barrier with Shared State

use std::sync::{Arc, Barrier, Mutex};
use std::thread;
 
fn main() {
    let num_threads = 4;
    let barrier = Arc::new(Barrier::new(num_threads));
    let results = Arc::new(Mutex::new(Vec::new()));
    let mut handles = vec![];
    
    for i in 0..num_threads {
        let barrier = Arc::clone(&barrier);
        let results = Arc::clone(&results);
        
        let handle = thread::spawn(move || {
            // Each thread computes a value
            let value = i * i;
            
            // Store partial result
            results.lock().unwrap().push(value);
            
            // Wait for all threads to finish computing
            barrier.wait();
            
            // Leader aggregates results
            let sum: i32 = results.lock().unwrap().iter().sum();
            
            barrier.wait();
            
            println!("Thread {} sees sum: {}", i, sum);
        });
        
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
}

Parallel Matrix Computation

use std::sync::{Arc, Barrier, Mutex};
use std::thread;
 
fn main() {
    let rows = 4;
    let cols = 4;
    let matrix: Vec<Vec<i32>> = (0..rows).map(|r| (0..cols).map(|c| (r * cols + c) as i32).collect()).collect();
    let result = Arc::new(Mutex::new(vec![vec![0i32; cols]; rows]));
    let barrier = Arc::new(Barrier::new(rows));
    
    let mut handles = vec![];
    
    for row in 0..rows {
        let barrier = Arc::clone(&barrier);
        let result = Arc::clone(&result);
        let row_data = matrix[row].clone();
        
        let handle = thread::spawn(move || {
            // Phase 1: Compute row
            let computed: Vec<i32> = row_data.iter().map(|&x| x * 2).collect();
            {
                let mut result = result.lock().unwrap();
                result[row] = computed;
            }
            
            barrier.wait();
            
            // Phase 2: Read all results
            let snapshot = result.lock().unwrap().clone();
            println!("Row {} sees: {:?}", row, snapshot);
        });
        
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
}

Barrier for Benchmarking

use std::sync::{Arc, Barrier};
use std::thread;
use std::time::Instant;
 
fn main() {
    let num_threads = 4;
    let barrier = Arc::new(Barrier::new(num_threads + 1)); // +1 for main
    let start_barrier = Arc::new(Barrier::new(num_threads + 1));
    let mut handles = vec![];
    
    for i in 0..num_threads {
        let barrier = Arc::clone(&barrier);
        let start_barrier = Arc::clone(&start_barrier);
        
        let handle = thread::spawn(move || {
            start_barrier.wait();  // Synchronized start
            
            // Work being benchmarked
            let mut sum = 0u64;
            for n in 0..1_000_000 {
                sum = sum.wrapping_add(n);
            }
            
            barrier.wait();  // Synchronized end
            
            sum
        });
        
        handles.push(handle);
    }
    
    // Synchronized start
    start_barrier.wait();
    let start = Instant::now();
    
    // Wait for all threads to complete
    barrier.wait();
    let duration = start.elapsed();
    
    println!("All threads completed in {:?}", duration);
    
    for handle in handles {
        handle.join().unwrap();
    }
}

Cyclic Barrier Pattern

use std::sync::{Arc, Barrier, Mutex};
use std::thread;
 
fn main() {
    let num_threads = 3;
    let num_iterations = 3;
    let barrier = Arc::new(Barrier::new(num_threads));
    let iteration = Arc::new(Mutex::new(0));
    
    let mut handles = vec![];
    
    for i in 0..num_threads {
        let barrier = Arc::clone(&barrier);
        let iteration = Arc::clone(&iteration);
        
        let handle = thread::spawn(move || {
            for _ in 0..num_iterations {
                let current_iter = *iteration.lock().unwrap();
                println!("Thread {} working on iteration {}", i, current_iter);
                
                thread::sleep(std::time::Duration::from_millis(50));
                
                let result = barrier.wait();
                
                if result.is_leader() {
                    println!("--- Iteration {} complete ---", current_iter);
                    *iteration.lock().unwrap() += 1;
                }
                
                barrier.wait();  // Ensure leader updates iteration before others read
            }
        });
        
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
}

Barrier with Timeout (Using wait_timeout)

Note: Barrier::wait_timeout is available in nightly Rust or you can combine with other primitives.

use std::sync::{Arc, Barrier};
use std::thread;
use std::time::Duration;
 
fn main() {
    let barrier = Arc::new(Barrier::new(3));
    let mut handles = vec![];
    
    for i in 0..3 {
        let barrier = Arc::clone(&barrier);
        
        let handle = thread::spawn(move || {
            // Simulate different arrival times
            thread::sleep(Duration::from_millis(100 * i as u64));
            println!("Thread {} reached barrier", i);
            
            barrier.wait();
            
            println!("Thread {} released", i);
        });
        
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
}

Barrier vs Condvar Comparison

use std::sync::{Arc, Barrier, Condvar, Mutex};
use std::thread;
 
fn main() {
    // Barrier: All threads wait for each other
    let barrier = Arc::new(Barrier::new(3));
    
    let mut handles = vec![];
    for i in 0..3 {
        let barrier = Arc::clone(&barrier);
        handles.push(thread::spawn(move || {
            println!("Barrier Thread {} waiting", i);
            barrier.wait();
            println!("Barrier Thread {} continuing", i);
        }));
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
    
    println!("---");
    
    // Condvar: One thread signals another
    let pair = Arc::new((Mutex::new(false), Condvar::new()));
    let pair_clone = Arc::clone(&pair);
    
    let waiter = thread::spawn(move || {
        let (lock, cvar) = &*pair_clone;
        let mut ready = lock.lock().unwrap();
        while !*ready {
            ready = cvar.wait(ready).unwrap();
        }
        println!("Condvar Thread woke up");
    });
    
    thread::sleep(std::time::Duration::from_millis(100));
    
    {
        let (lock, cvar) = &*pair;
        *lock.lock().unwrap() = true;
        cvar.notify_one();
    }
    
    waiter.join().unwrap();
}

Parallel Data Processing Pipeline

use std::sync::{Arc, Barrier, Mutex};
use std::thread;
 
struct Pipeline {
    stage1_barrier: Arc<Barrier>,
    stage2_barrier: Arc<Barrier>,
    stage3_barrier: Arc<Barrier>,
    data: Arc<Mutex<Vec<i32>>>,
}
 
impl Pipeline {
    fn new(num_workers: usize) -> Self {
        Self {
            stage1_barrier: Arc::new(Barrier::new(num_workers)),
            stage2_barrier: Arc::new(Barrier::new(num_workers)),
            stage3_barrier: Arc::new(Barrier::new(num_workers)),
            data: Arc::new(Mutex::new(vec![0; num_workers])),
        }
    }
}
 
fn main() {
    let num_workers = 4;
    let pipeline = Pipeline::new(num_workers);
    let mut handles = vec![];
    
    for worker_id in 0..num_workers {
        let stage1 = Arc::clone(&pipeline.stage1_barrier);
        let stage2 = Arc::clone(&pipeline.stage2_barrier);
        let stage3 = Arc::clone(&pipeline.stage3_barrier);
        let data = Arc::clone(&pipeline.data);
        
        let handle = thread::spawn(move || {
            // Stage 1: Generate
            {
                let mut data = data.lock().unwrap();
                data[worker_id] = (worker_id + 1) as i32;
            }
            stage1.wait();
            
            // Stage 2: Transform (double values)
            {
                let mut data = data.lock().unwrap();
                data[worker_id] *= 2;
            }
            stage2.wait();
            
            // Stage 3: Aggregate (compute sum)
            let sum: i32 = {
                let data = data.lock().unwrap();
                data.iter().sum()
            };
            stage3.wait();
            
            println!("Worker {} final sum: {}", worker_id, sum);
        });
        
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
}

Thread Startup Synchronization

use std::sync::{Arc, Barrier};
use std::thread;
use std::time::Instant;
 
fn main() {
    let num_threads = 5;
    let start_barrier = Arc::new(Barrier::new(num_threads + 1)); // +1 for main
    
    let mut handles = vec![];
    
    for i in 0..num_threads {
        let barrier = Arc::clone(&start_barrier);
        
        let handle = thread::spawn(move || {
            println!("Thread {} ready", i);
            barrier.wait();
            
            let start = Instant::now();
            // Simulate work
            thread::sleep(std::time::Duration::from_millis(100));
            
            start.elapsed()
        });
        
        handles.push(handle);
    }
    
    println!("Main thread ready");
    start_barrier.wait();
    println!("All threads starting now!");
    
    for (i, handle) in handles.into_iter().enumerate() {
        let elapsed = handle.join().unwrap();
        println!("Thread {} took {:?}", i, elapsed);
    }
}

Barrier for Parallel Sort Phases

use std::sync::{Arc, Barrier, Mutex};
use std::thread;
 
fn main() {
    let num_threads = 4;
    let barrier = Arc::new(Barrier::new(num_threads));
    let data = Arc::new(Mutex::new(vec![64, 25, 12, 22, 11, 90, 45, 33,
                                          1, 78, 56, 89, 23, 67, 34, 99]));
    let sorted = Arc::new(Mutex::new(false));
    
    let mut handles = vec![];
    
    for thread_id in 0..num_threads {
        let barrier = Arc::clone(&barrier);
        let data = Arc::clone(&data);
        let sorted = Arc::clone(&sorted);
        
        let handle = thread::spawn(move || {
            for pass in 0..data.lock().unwrap().len() {
                {
                    let mut arr = data.lock().unwrap();
                    let chunk_size = arr.len() / num_threads;
                    let start = thread_id * chunk_size;
                    let end = if thread_id == num_threads - 1 {
                        arr.len()
                    } else {
                        (thread_id + 1) * chunk_size
                    };
                    
                    // Local sort on chunk
                    arr[start..end].sort();
                }
                
                barrier.wait();
                
                // Leader checks if fully sorted
                if thread_id == 0 {
                    let arr = data.lock().unwrap();
                    let is_sorted = arr.windows(2).all(|w| w[0] <= w[1]);
                    *sorted.lock().unwrap() = is_sorted;
                }
                
                barrier.wait();
                
                if *sorted.lock().unwrap() {
                    break;
                }
            }
            
            if thread_id == 0 {
                println!("Sorted: {:?}", *data.lock().unwrap());
            }
        });
        
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
}

Barrier for Map-Reduce Pattern

use std::sync::{Arc, Barrier, Mutex};
use std::thread;
use std::collections::HashMap;
 
fn main() {
    let data = vec!["apple", "banana", "apple", "cherry", "banana", 
                    "apple", "date", "cherry", "elderberry", "fig"];
    let num_workers = 3;
    
    let map_barrier = Arc::new(Barrier::new(num_workers));
    let reduce_barrier = Arc::new(Barrier::new(num_workers));
    let partial_results = Arc::new(Mutex::new(Vec::new()));
    let final_result = Arc::new(Mutex::new(HashMap::new()));
    
    let mut handles = vec![];
    
    for worker_id in 0..num_workers {
        let map_barrier = Arc::clone(&map_barrier);
        let reduce_barrier = Arc::clone(&reduce_barrier);
        let partial_results = Arc::clone(&partial_results);
        let final_result = Arc::clone(&final_result);
        let chunk = data.chunks((data.len() + num_workers - 1) / num_workers)
                        .nth(worker_id)
                        .unwrap_or(&[])
                        .to_vec();
        
        let handle = thread::spawn(move || {
            // Map phase: count words in chunk
            let mut local_counts = HashMap::new();
            for word in chunk {
                *local_counts.entry(word).or_insert(0) += 1;
            }
            
            partial_results.lock().unwrap().push(local_counts);
            map_barrier.wait();
            
            // Reduce phase: aggregate partial results
            let result = reduce_barrier.wait();
            if result.is_leader() {
                let mut final_counts = HashMap::new();
                for partial in partial_results.lock().unwrap().iter() {
                    for (word, count) in partial {
                        *final_counts.entry(*word).or_insert(0) += count;
                    }
                }
                *final_result.lock().unwrap() = final_counts;
            }
            
            reduce_barrier.wait();
            
            if worker_id == 0 {
                println!("Final counts: {:?}", *final_result.lock().unwrap());
            }
        });
        
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
}

Summary

Barrier Methods:

Method Description
new(n) Create a barrier for n threads
wait() Block until all threads reach barrier
wait().is_leader() Check if this thread is the "leader"

Barrier vs Other Primitives:

Primitive Purpose
Barrier All threads wait for each other (synchronization point)
Condvar Thread waits for signal from another thread
Channel Thread sends/receives data to/from another thread
Barrier Multiple threads synchronize at same point
Once One-time initialization (single use)

Common Patterns:

Pattern Description
Multi-phase computation Parallel algorithms with distinct phases
Bulk-synchronous parallel BSP model implementation
Synchronized startup All threads start at same time
Benchmarking Measure parallel section execution
Map-reduce Parallel map, then reduce phase
Pipeline stages Phased parallel processing

Key Points:

  • Barrier::new(n) creates a barrier for n threads
  • wait() blocks until all n threads have called wait()
  • BarrierWaitResult::is_leader() identifies one thread as leader
  • Barriers are reusable after all threads pass through
  • Use for phased parallel computations
  • Combine with Mutex for shared state access
  • For signaling between threads, use Condvar or channels
  • Thread-safe and efficient for coordination
  • Part of std::sync module