109 Advanced

109-arc-threads — Arc<T>: Thread-Safe Shared Ownership

Functional Programming

Tutorial

The Problem

When multiple threads need read access to the same data, you need a mechanism for shared ownership with thread-safe reference counting. Rust's Rc<T> is single-threaded — it uses non-atomic reference counting that is not safe to clone across threads. Arc<T> (Atomically Reference-Counted) uses atomic operations for the count, making it safe to share across threads.

Arc<T> is the Rust equivalent of shared_ptr in C++ (thread-safe variant) and the GC-managed heap references in OCaml, but with explicit reference counting visible to the programmer.

🎯 Learning Outcomes

• Understand when to use Arc<T> versus Rc<T> (thread-safe vs single-threaded)

• Clone Arc<T> to share ownership across thread boundaries

• Combine Arc<Mutex<T>> for shared mutable state across threads

• Understand the overhead: atomic operations on the reference count

• Use Arc in map-reduce patterns for parallel data processing

Code Example

use std::sync::Arc;
use std::thread;

pub fn parallel_sum(data: Arc<Vec<i32>>) -> i32 {
    let mid = data.len() / 2;

    let left = Arc::clone(&data);
    let h1 = thread::spawn(move || left[..mid].iter().sum::<i32>());

    let right = Arc::clone(&data);
    let h2 = thread::spawn(move || right[mid..].iter().sum::<i32>());

    h1.join().unwrap() + h2.join().unwrap()
}

(* OCaml's GC handles shared lifetimes automatically.
   In OCaml 5, Domains share heap values with no extra annotation. *)
let parallel_sum data =
  let mid = Array.length data / 2 in
  let d1 = Domain.spawn (fun () ->
    Array.fold_left (+) 0 (Array.sub data 0 mid)) in
  let d2 = Domain.spawn (fun () ->
    Array.fold_left (+) 0 (Array.sub data mid (Array.length data - mid))) in
  Domain.join d1 + Domain.join d2

Key Differences

Explicit vs implicit: Rust's Arc::clone explicitly increments the reference count; OCaml's GC manages shared references automatically.

Atomic overhead: Rust's Arc uses atomic CAS operations for the count (more expensive than Rc's simple increment); OCaml's GC has its own overhead but it is amortized.

Mutation safety: Rust requires Arc<Mutex<T>> for shared mutation; OCaml 5 uses Mutex or Atomic from the standard library.

Move vs clone: In Rust, you must Arc::clone before moving into a thread; OCaml values can be used in multiple domains without explicit cloning.

OCaml Approach

OCaml's GC is not concurrent by default (the global interpreter lock in pre-5.0 OCaml), but OCaml 5 introduces Domains for true parallelism:

(* OCaml 5 with Domains *)
let parallel_sum data =
  let mid = Array.length data / 2 in
  let d = Domain.spawn (fun () -> Array.fold_left (+) 0 (Array.sub data 0 mid)) in
  let right_sum = Array.fold_left (+) 0 (Array.sub data mid (Array.length data - mid)) in
  Domain.join d + right_sum

OCaml's GC handles shared data automatically — no explicit reference counting needed. Data sharing across domains is safe for immutable values.

Full Source

#![allow(clippy::all)]
// Example 109: Arc<T> — Thread-Safe Shared Ownership
//
// Arc<T> = Atomic Reference Count. Like Rc<T> but thread-safe.
// Clone an Arc to share ownership across threads; the value is
// freed only after every thread drops its clone.

use std::sync::{Arc, Mutex};
use std::thread;

// --- Approach 1: Shared immutable data across threads ----------------------

/// Parallel sum: split a Vec into two halves, sum each half in its own thread.
/// The Vec is wrapped in Arc so both threads can read it without copying.
pub fn parallel_sum(data: Arc<Vec<i32>>) -> i32 {
    let mid = data.len() / 2;

    // Clone the Arc (bumps atomic counter) — no heap allocation of data.
    let left = Arc::clone(&data);
    let handle_left = thread::spawn(move || left[..mid].iter().sum::<i32>());

    let right = Arc::clone(&data);
    let handle_right = thread::spawn(move || right[mid..].iter().sum::<i32>());

    handle_left.join().unwrap() + handle_right.join().unwrap()
}

// --- Approach 2: Map-reduce with Arc-shared configuration ------------------

/// A processing configuration shared read-only across worker threads.
#[derive(Debug)]
pub struct Config {
    pub multiplier: i32,
    pub offset: i32,
}

/// Apply `config` to every element across `n_threads` worker threads.
/// Each thread owns a clone of the Arc — the Config is never copied.
pub fn parallel_map(data: Vec<i32>, config: Arc<Config>, n_threads: usize) -> Vec<i32> {
    let data = Arc::new(data);
    let len = data.len();
    let chunk = len.div_ceil(n_threads);

    let handles: Vec<_> = (0..n_threads)
        .map(|i| {
            let data = Arc::clone(&data);
            let cfg = Arc::clone(&config);
            thread::spawn(move || {
                let start = i * chunk;
                let end = (start + chunk).min(len);
                data[start..end]
                    .iter()
                    .map(|&x| x * cfg.multiplier + cfg.offset)
                    .collect::<Vec<i32>>()
            })
        })
        .collect();

    handles
        .into_iter()
        .flat_map(|h| h.join().unwrap())
        .collect()
}

// --- Approach 3: Arc<Mutex<T>> for shared mutable state --------------------

/// Accumulate results from multiple threads into a shared counter.
/// Arc owns the Mutex; Mutex guards the i32 inside.
pub fn concurrent_count(items: Vec<i32>) -> i32 {
    let total: Arc<Mutex<i32>> = Arc::new(Mutex::new(0));

    let handles: Vec<_> = items
        .into_iter()
        .map(|x| {
            let total = Arc::clone(&total);
            thread::spawn(move || {
                let mut guard = total.lock().unwrap();
                *guard += x;
            })
        })
        .collect();

    for h in handles {
        h.join().unwrap();
    }

    let result = *total.lock().unwrap();
    result
}

/// Demonstrate Arc reference counting: count rises with clones, falls with drops.
pub fn arc_ref_count_demo() -> (usize, usize) {
    let a: Arc<Vec<i32>> = Arc::new(vec![1, 2, 3]);
    let b = Arc::clone(&a);
    let count_two = Arc::strong_count(&a); // 2: `a` + `b`
    drop(b);
    let count_one = Arc::strong_count(&a); // 1: only `a`
    (count_two, count_one)
}

// --------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_parallel_sum_correctness() {
        let data = Arc::new((1..=100).collect::<Vec<i32>>());
        assert_eq!(parallel_sum(data), 5050);
    }

    #[test]
    fn test_parallel_sum_empty() {
        let data = Arc::new(vec![]);
        assert_eq!(parallel_sum(data), 0);
    }

    #[test]
    fn test_parallel_sum_single() {
        let data = Arc::new(vec![42]);
        assert_eq!(parallel_sum(data), 42);
    }

    #[test]
    fn test_parallel_map_applies_config() {
        let cfg = Arc::new(Config {
            multiplier: 2,
            offset: 1,
        });
        let result = parallel_map(vec![1, 2, 3, 4], cfg, 2);
        // Each x → x*2 + 1
        assert_eq!(result, vec![3, 5, 7, 9]);
    }

    #[test]
    fn test_parallel_map_single_thread() {
        let cfg = Arc::new(Config {
            multiplier: 3,
            offset: 0,
        });
        let result = parallel_map(vec![1, 2, 3], cfg, 1);
        assert_eq!(result, vec![3, 6, 9]);
    }

    #[test]
    fn test_concurrent_count() {
        let items: Vec<i32> = (1..=10).collect();
        assert_eq!(concurrent_count(items), 55);
    }

    #[test]
    fn test_concurrent_count_empty() {
        assert_eq!(concurrent_count(vec![]), 0);
    }

    #[test]
    fn test_arc_ref_count() {
        let (two, one) = arc_ref_count_demo();
        assert_eq!(two, 2);
        assert_eq!(one, 1);
    }
}

(* Example 109: Arc<T> — Thread-Safe Sharing *)

(* OCaml's GC is thread-safe by default. Sharing data between
   threads (via Domain in OCaml 5) just works. *)

(* Approach 1: Shared immutable data across "workers" *)
let process_chunk data start len =
  let sum = ref 0 in
  for i = start to start + len - 1 do
    sum := !sum + data.(i)
  done;
  !sum

let approach1 () =
  let data = Array.init 100 (fun i -> i + 1) in
  let sum1 = process_chunk data 0 50 in
  let sum2 = process_chunk data 50 50 in
  let total = sum1 + sum2 in
  assert (total = 5050);
  Printf.printf "Total: %d\n" total

(* Approach 2: Map-reduce pattern *)
let map_reduce mapper reducer init data =
  let mapped = List.map mapper data in
  List.fold_left reducer init mapped

let approach2 () =
  let data = [1; 2; 3; 4; 5] in
  let result = map_reduce (fun x -> x * x) ( + ) 0 data in
  assert (result = 55);
  Printf.printf "Sum of squares: %d\n" result

(* Approach 3: Parallel word count simulation *)
let count_words text =
  let words = String.split_on_char ' ' text in
  List.length (List.filter (fun w -> String.length w > 0) words)

let approach3 () =
  let texts = ["hello world"; "foo bar baz"; "one"] in
  let counts = List.map count_words texts in
  let total = List.fold_left ( + ) 0 counts in
  assert (total = 6);
  Printf.printf "Total words: %d\n" total

let () =
  approach1 ();
  approach2 ();
  approach3 ();
  Printf.printf "✓ All tests passed\n"

✓ Tests Rust test suite

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_parallel_sum_correctness() {
        let data = Arc::new((1..=100).collect::<Vec<i32>>());
        assert_eq!(parallel_sum(data), 5050);
    }

    #[test]
    fn test_parallel_sum_empty() {
        let data = Arc::new(vec![]);
        assert_eq!(parallel_sum(data), 0);
    }

    #[test]
    fn test_parallel_sum_single() {
        let data = Arc::new(vec![42]);
        assert_eq!(parallel_sum(data), 42);
    }

    #[test]
    fn test_parallel_map_applies_config() {
        let cfg = Arc::new(Config {
            multiplier: 2,
            offset: 1,
        });
        let result = parallel_map(vec![1, 2, 3, 4], cfg, 2);
        // Each x → x*2 + 1
        assert_eq!(result, vec![3, 5, 7, 9]);
    }

    #[test]
    fn test_parallel_map_single_thread() {
        let cfg = Arc::new(Config {
            multiplier: 3,
            offset: 0,
        });
        let result = parallel_map(vec![1, 2, 3], cfg, 1);
        assert_eq!(result, vec![3, 6, 9]);
    }

    #[test]
    fn test_concurrent_count() {
        let items: Vec<i32> = (1..=10).collect();
        assert_eq!(concurrent_count(items), 55);
    }

    #[test]
    fn test_concurrent_count_empty() {
        assert_eq!(concurrent_count(vec![]), 0);
    }

    #[test]
    fn test_arc_ref_count() {
        let (two, one) = arc_ref_count_demo();
        assert_eq!(two, 2);
        assert_eq!(one, 1);
    }
}

Deep Comparison

OCaml vs Rust: Arc<T> — Thread-Safe Shared Ownership

Side-by-Side Code

OCaml

(* OCaml's GC handles shared lifetimes automatically.
   In OCaml 5, Domains share heap values with no extra annotation. *)
let parallel_sum data =
  let mid = Array.length data / 2 in
  let d1 = Domain.spawn (fun () ->
    Array.fold_left (+) 0 (Array.sub data 0 mid)) in
  let d2 = Domain.spawn (fun () ->
    Array.fold_left (+) 0 (Array.sub data mid (Array.length data - mid))) in
  Domain.join d1 + Domain.join d2

Rust (idiomatic — Arc for shared ownership)

use std::sync::Arc;
use std::thread;

pub fn parallel_sum(data: Arc<Vec<i32>>) -> i32 {
    let mid = data.len() / 2;

    let left = Arc::clone(&data);
    let h1 = thread::spawn(move || left[..mid].iter().sum::<i32>());

    let right = Arc::clone(&data);
    let h2 = thread::spawn(move || right[mid..].iter().sum::<i32>());

    h1.join().unwrap() + h2.join().unwrap()
}

Rust (Arc<Mutex<T>> for shared mutable state)

use std::sync::{Arc, Mutex};
use std::thread;

pub fn concurrent_count(items: Vec<i32>) -> i32 {
    let total = Arc::new(Mutex::new(0_i32));
    let handles: Vec<_> = items.into_iter().map(|x| {
        let t = Arc::clone(&total);
        thread::spawn(move || { *t.lock().unwrap() += x; })
    }).collect();
    for h in handles { h.join().unwrap(); }
    *total.lock().unwrap()
}

Type Signatures

Concept	OCaml	Rust
Shared reference	GC-managed `'a`	`Arc<T>` (atomic ref count)
Thread handle	`'a Domain.t`	`JoinHandle<T>`
Shared mutable	`Mutex.t` + GC ref	`Arc<Mutex<T>>`
Clone cost	pointer copy (GC)	atomic increment (cheap)
Lifetime proof	runtime (GC traces)	compile-time (ownership rules)

Key Insights

Explicit vs implicit sharing: OCaml's GC tracks all references invisibly; Rust requires you to opt in to shared ownership with Arc::clone, making the sharing explicit and auditable at the call site.

**Rc<T> vs Arc<T>:** Rc<T> uses non-atomic (single-threaded) reference counting and is not Send; the compiler rejects sending it across thread boundaries. Arc<T> uses atomic operations and is both Send + Sync, so the compiler allows it.

Immutability is the default: Arc<T> gives you &T — shared immutable access. For mutation you must add Mutex<T> (or RwLock<T>), spelling out both "this is shared" and "this is mutable" separately. This prevents data races at compile time.

No GC overhead: Arc<T> pays only for the two atomic integers (strong count + weak count) stored next to the value on the heap. There is no stop-the-world pause and no scanning of the entire heap; the value is freed the instant the last Arc drops.

**move closures capture the clone:** Rust's move keyword transfers the cloned Arc into the thread closure, proving to the borrow checker that the thread owns its own reference and cannot outlive the data.

When to Use Each Style

**Use Arc<T> (immutable):* when multiple threads only need to read* shared data — config, lookup tables, parsed input. Zero-cost reads after the initial atomic clone.

**Use Arc<Mutex<T>>:* when multiple threads need to write* shared state — counters, accumulators, work queues. The Mutex ensures exclusive access; the Arc ensures the Mutex itself lives long enough.

Exercises

Implement a thread pool using Arc<Mutex<VecDeque<Box<dyn FnOnce() + Send>>>> where worker threads pull and execute tasks.

Build a concurrent frequency counter using Arc<Mutex<HashMap<String, usize>>> and verify correct counting from multiple threads.

Demonstrate that using Arc<T> without Mutex for shared immutable data is both safe and faster than Arc<Mutex<T>> for read-heavy workloads.

Open Source Repos

functional-rust

View the source for this example on GitHub — OCaml and Rust side by side in the repo.

Rust