Atomics, Send & Sync

use std::sync::atomic::{AtomicUsize, Ordering};
use std::thread;

fn main() {
    let hits = AtomicUsize::new(0);

    thread::scope(|s| {
        for _ in 0..8 {
            s.spawn(|| {
                for _ in 0..10_000 {
                    hits.fetch_add(1, Ordering::Relaxed);
                }
            });
        }
    });

    println!("hits: {}", hits.load(Ordering::Relaxed));
}

use std::sync::atomic::{AtomicU64, Ordering};
use std::thread;

static REQUESTS: AtomicU64 = AtomicU64::new(0);

fn handle_request(path: &str) -> usize {
    REQUESTS.fetch_add(1, Ordering::Relaxed);
    path.len()
}

fn main() {
    let handles: Vec<_> = (0..4)
        .map(|i| thread::spawn(move || (0..25).map(|_| handle_request(if i % 2 == 0 { "/home" } else { "/api" })).sum::<usize>()))
        .collect();
    let bytes: usize = handles.into_iter().map(|h| h.join().unwrap()).sum();

    println!("requests: {}", REQUESTS.load(Ordering::Relaxed));
    println!("path bytes: {bytes}");
}

use std::collections::HashSet;
use std::sync::atomic::{AtomicU32, Ordering};
use std::thread;

static NEXT_ORDER_ID: AtomicU32 = AtomicU32::new(1000);

fn next_order_id() -> u32 {
    NEXT_ORDER_ID.fetch_add(1, Ordering::Relaxed)
}

fn main() {
    let ids: Vec<u32> = thread::scope(|s| {
        let handles: Vec<_> = (0..4).map(|_| s.spawn(|| (0..50).map(|_| next_order_id()).collect::<Vec<_>>())).collect();
        handles.into_iter().flat_map(|h| h.join().unwrap()).collect()
    });

    let unique: HashSet<u32> = ids.iter().copied().collect();
    println!("{} ids, {} unique", ids.len(), unique.len());
    println!("range: {}..={}", ids.iter().min().unwrap(), ids.iter().max().unwrap());
}

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread;
use std::time::Duration;

fn main() {
    let running = Arc::new(AtomicBool::new(true));

    let r = Arc::clone(&running);
    let worker = thread::spawn(move || {
        let mut polls = 0;
        while r.load(Ordering::Acquire) {
            polls += 1; // poll a queue, check a socket...
            thread::sleep(Duration::from_millis(5));
        }
        polls
    });

    thread::sleep(Duration::from_millis(40));
    running.store(false, Ordering::Release);
    let polls = worker.join().unwrap();
    println!("worker polled and then stopped: {}", polls > 0);
}

use std::sync::atomic::{AtomicU64, Ordering};
use std::thread;

static BYTES_SENT: AtomicU64 = AtomicU64::new(0);

fn main() {
    let mut reports = Vec::new();
    for second in 1..=3 {
        thread::scope(|s| {
            for _ in 0..4 {
                s.spawn(move || BYTES_SENT.fetch_add(100 * second, Ordering::Relaxed));
            }
        });
        // Read this interval's total and reset to zero in one atomic step.
        reports.push(BYTES_SENT.swap(0, Ordering::Relaxed));
    }
    println!("bytes per interval: {reports:?}");
    println!("after reset: {}", BYTES_SENT.load(Ordering::Relaxed));
}

use std::sync::atomic::{AtomicU32, AtomicU8, Ordering};
use std::thread;

const FLAG_DB: u8 = 0b001;
const FLAG_CACHE: u8 = 0b010;
const FLAG_QUEUE: u8 = 0b100;

fn main() {
    let max_latency = AtomicU32::new(0);
    let ready = AtomicU8::new(0);
    let latencies = [[12, 40, 7], [90, 3, 15], [22, 61, 5]];

    thread::scope(|s| {
        for (worker, samples) in latencies.iter().enumerate() {
            let (max_latency, ready) = (&max_latency, &ready);
            s.spawn(move || {
                for &ms in samples {
                    max_latency.fetch_max(ms, Ordering::Relaxed);
                }
                ready.fetch_or([FLAG_DB, FLAG_CACHE, FLAG_QUEUE][worker], Ordering::Relaxed);
            });
        }
    });

    println!("max latency: {} ms", max_latency.load(Ordering::Relaxed));
    println!("ready flags: {:03b}", ready.load(Ordering::Relaxed));
    ready.fetch_and(!FLAG_CACHE, Ordering::Relaxed); // cache went down
    println!("after cache failure: {:03b}", ready.load(Ordering::Relaxed));
}

use std::sync::atomic::{AtomicU32, Ordering};
use std::thread;

const NO_LEADER: u32 = 0;

fn main() {
    let leader = AtomicU32::new(NO_LEADER);

    let winners: Vec<u32> = thread::scope(|s| {
        let handles: Vec<_> = (1..=5)
            .map(|node_id| {
                let leader = &leader;
                s.spawn(move || {
                    match leader.compare_exchange(NO_LEADER, node_id, Ordering::AcqRel, Ordering::Acquire) {
                        Ok(_) => Some(node_id),       // we set it: we're the leader
                        Err(_current) => None,         // someone else got there first
                    }
                })
            })
            .collect();
        handles.into_iter().filter_map(|h| h.join().unwrap()).collect()
    });

    println!("number of winners: {}", winners.len());
    println!("winner matches stored leader: {}", winners[0] == leader.load(Ordering::Acquire));
}

use std::sync::atomic::{AtomicU32, Ordering};
use std::thread;

// Take `n` items from stock if enough are left. Never goes below zero.
fn reserve(stock: &AtomicU32, n: u32) -> Result<u32, u32> {
    stock.fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| current.checked_sub(n))
}

// The same thing, spelled out as a manual CAS loop.
fn reserve_manual(stock: &AtomicU32, n: u32) -> Result<u32, u32> {
    let mut current = stock.load(Ordering::Acquire);
    loop {
        let Some(new) = current.checked_sub(n) else { return Err(current) };
        match stock.compare_exchange_weak(current, new, Ordering::AcqRel, Ordering::Acquire) {
            Ok(prev) => return Ok(prev),
            Err(actual) => current = actual, // someone changed it: retry with the fresh value
        }
    }
}

fn main() {
    let stock = AtomicU32::new(100);

    let sold: u32 = thread::scope(|s| {
        let handles: Vec<_> = (0..8)
            .map(|i| {
                let stock = &stock;
                s.spawn(move || {
                    let mut sold = 0;
                    for _ in 0..10 {
                        let r = if i % 2 == 0 { reserve(stock, 3) } else { reserve_manual(stock, 3) };
                        if r.is_ok() {
                            sold += 3;
                        }
                    }
                    sold
                })
            })
            .collect();
        handles.into_iter().map(|h| h.join().unwrap()).sum()
    });

    println!("sold {sold}, left {}", stock.load(Ordering::Acquire));
    println!("nothing oversold: {}", sold + stock.load(Ordering::Acquire) == 100);
}

use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::thread;

static DATA: AtomicU64 = AtomicU64::new(0);
static READY: AtomicBool = AtomicBool::new(false);

fn main() {
    thread::scope(|s| {
        s.spawn(|| {
            DATA.store(42, Ordering::Relaxed); // 1. write the data
            READY.store(true, Ordering::Release); // 2. publish: "data is ready"
        });

        s.spawn(|| {
            while !READY.load(Ordering::Acquire) {} // 3. wait until published
            // 4. Guaranteed to see 42: Acquire "sees" everything before the Release.
            assert_eq!(DATA.load(Ordering::Relaxed), 42);
        });
    });
    println!("the reader saw the published data");
}

use std::sync::Mutex;
use std::sync::atomic::{AtomicU64, Ordering};
use std::thread;

// Wrong tool: count and total are updated separately, so a reader can see
// a count that doesn't match the total yet.
struct RacyStats {
    count: AtomicU64,
    total_ms: AtomicU64,
}

// Right tool: both fields change together under one lock.
#[derive(Default)]
struct Stats {
    count: u64,
    total_ms: u64,
}

fn main() {
    let racy = RacyStats { count: AtomicU64::new(0), total_ms: AtomicU64::new(0) };
    let safe = Mutex::new(Stats::default());

    thread::scope(|s| {
        for _ in 0..4 {
            s.spawn(|| {
                for _ in 0..1000 {
                    racy.count.fetch_add(1, Ordering::Relaxed);
                    racy.total_ms.fetch_add(10, Ordering::Relaxed);
                    let mut st = safe.lock().unwrap();
                    st.count += 1;
                    st.total_ms += 10;
                }
            });
        }
    });

    // Once all threads are done, both agree. The difference is what a reader
    // sees *during* the run: only the Mutex version is always consistent.
    let st = safe.lock().unwrap();
    println!("atomic: {} / {}", racy.count.load(Ordering::Relaxed), racy.total_ms.load(Ordering::Relaxed));
    println!("mutex:  {} / {}", st.count, st.total_ms);
}

use std::collections::HashMap;
use std::sync::OnceLock;
use std::sync::atomic::{AtomicU32, Ordering};
use std::thread;

static INIT_CALLS: AtomicU32 = AtomicU32::new(0);
static COUNTRY_NAMES: OnceLock<HashMap<&'static str, &'static str>> = OnceLock::new();

fn country_name(code: &str) -> Option<&'static str> {
    let table = COUNTRY_NAMES.get_or_init(|| {
        INIT_CALLS.fetch_add(1, Ordering::Relaxed);
        HashMap::from([("DE", "Germany"), ("IN", "India"), ("JP", "Japan")]) // pretend: loaded from a file
    });
    table.get(code).copied()
}

fn main() {
    let names: Vec<Option<&str>> = thread::scope(|s| {
        let handles: Vec<_> = ["IN", "JP", "XX", "DE"].map(|c| s.spawn(move || country_name(c))).into_iter().collect();
        handles.into_iter().map(|h| h.join().unwrap()).collect()
    });
    println!("{names:?}");
    println!("initializer ran {} time(s)", INIT_CALLS.load(Ordering::Relaxed));
}

use std::collections::HashSet;
use std::sync::LazyLock;

static STOP_WORDS: LazyLock<HashSet<&'static str>> = LazyLock::new(|| {
    "a an and are as at be by for from in is it of on or the to with".split(' ').collect()
});

fn keywords(text: &str) -> Vec<String> {
    text.split_whitespace()
        .map(|w| w.trim_matches(|c: char| !c.is_alphanumeric()).to_lowercase())
        .filter(|w| !w.is_empty() && !STOP_WORDS.contains(w.as_str()))
        .collect()
}

fn main() {
    println!("{:?}", keywords("The quick brown fox is jumping over the lazy dog."));
    println!("{} stop words loaded", STOP_WORDS.len());
}

use std::sync::OnceLock;
use std::thread;

struct Document {
    text: String,
    word_count: OnceLock<usize>,
}

impl Document {
    fn new(text: &str) -> Self {
        Document { text: text.to_string(), word_count: OnceLock::new() }
    }

    fn word_count(&self) -> usize {
        *self.word_count.get_or_init(|| self.text.split_whitespace().count()) // computed at most once
    }
}

fn main() {
    let doc = Document::new("Rust makes concurrent code safe and fast");
    println!("computed yet? {}", doc.word_count.get().is_some());

    let counts: Vec<usize> = thread::scope(|s| {
        let hs: Vec<_> = (0..3).map(|_| s.spawn(|| doc.word_count())).collect();
        hs.into_iter().map(|h| h.join().unwrap()).collect()
    });
    println!("{counts:?}, computed yet? {}", doc.word_count.get().is_some());
}

use std::cell::{Cell, RefCell};
use std::rc::Rc;
use std::sync::{Arc, Mutex, RwLock};

fn is_send<T: Send>() {}
fn is_sync<T: Sync>() {}

fn main() {
    // These all compile, so each type has the trait.
    is_send::<String>();
    is_sync::<String>();
    is_send::<Arc<Vec<i32>>>();
    is_sync::<Arc<Vec<i32>>>();
    is_sync::<Mutex<Vec<i32>>>();
    is_sync::<RwLock<String>>();
    is_send::<Cell<i32>>(); // Send: can be moved to another thread...
    is_send::<RefCell<i32>>();
    // is_sync::<Cell<i32>>();   // ...but NOT Sync: can't be shared (error E0277)
    // is_send::<Rc<i32>>();     // Rc is neither Send nor Sync
    let _ = (Rc::new(1), Cell::new(1), RefCell::new(1));
    println!("all checks compiled");
}

use std::thread::{self, JoinHandle};
use std::time::Instant;

fn run_timed<F, T>(name: &'static str, work: F) -> JoinHandle<(T, bool)>
where
    F: FnOnce() -> T + Send + 'static,
    T: Send + 'static,
{
    thread::Builder::new()
        .name(name.to_string())
        .spawn(move || {
            let start = Instant::now();
            let result = work();
            (result, start.elapsed().as_secs() < 5)
        })
        .unwrap()
}

fn main() {
    let words = vec![String::from("send"), String::from("sync")];
    let h1 = run_timed("sum", || (1..=100u32).sum::<u32>());
    let h2 = run_timed("words", move || words.join(" + "));

    println!("{:?}", h1.join().unwrap());
    println!("{:?}", h2.join().unwrap());
}

use std::thread;

type Task = Box<dyn FnOnce() -> String + Send>;

fn main() {
    let customer = String::from("ACME");
    let tasks: Vec<Task> = vec![
        Box::new(|| String::from("compress logs")),
        Box::new(move || format!("invoice {customer}")),
        Box::new(|| (1..=5).map(|n| n.to_string()).collect::<Vec<_>>().join(",")),
    ];

    let worker = thread::spawn(move || tasks.into_iter().map(|t| t()).collect::<Vec<_>>());
    println!("{:?}", worker.join().unwrap());
}

use std::error::Error;
use std::thread;

type BoxError = Box<dyn Error + Send + Sync>;

fn load_port(raw: &str) -> Result<u16, BoxError> {
    let port: u16 = raw.parse()?;
    if port < 1024 {
        return Err(format!("port {port} is privileged").into());
    }
    Ok(port)
}

fn main() {
    let handles: Vec<_> = ["8080", "80", "http"].map(|raw| thread::spawn(move || load_port(raw))).into_iter().collect();
    for h in handles {
        match h.join().unwrap() {
            Ok(port) => println!("port {port}"),
            Err(e) => println!("error: {e}"),
        }
    }
}

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

fn main() {
    let shared = Arc::new(String::from("config"));

    thread::scope(|s| {
        for _ in 0..4 {
            s.spawn(|| {
                let mut clones = Vec::new();
                for _ in 0..1000 {
                    clones.push(Arc::clone(&shared)); // atomic increment
                }
                // all 1000 clones dropped here: atomic decrements
            });
        }
    });

    // With a non-atomic count, lost updates would leave this wrong.
    println!("strong count after 4000 clones and drops: {}", Arc::strong_count(&shared));
}

struct MyBuffer {
    ptr: *mut u8, // raw pointer: makes the struct !Send and !Sync by default
    len: usize,
}

// SAFETY: MyBuffer owns its allocation exclusively and never shares `ptr`,
// so moving it to another thread is safe.
unsafe impl Send for MyBuffer {}

use std::sync::atomic::{AtomicU64, Ordering::Relaxed};
use std::thread;

#[derive(Default)]
struct ServerStats {
    requests: AtomicU64,
    errors: AtomicU64,
    bytes_out: AtomicU64,
    max_latency_us: AtomicU64,
}

impl ServerStats {
    fn record(&self, status: u16, bytes: u64, latency_us: u64) {
        self.requests.fetch_add(1, Relaxed);
        if status >= 500 {
            self.errors.fetch_add(1, Relaxed);
        }
        self.bytes_out.fetch_add(bytes, Relaxed);
        self.max_latency_us.fetch_max(latency_us, Relaxed);
    }

    fn summary(&self) -> String {
        let req = self.requests.load(Relaxed);
        format!(
            "{req} requests, {:.1}% errors, {} KiB out, max {} us",
            100.0 * self.errors.load(Relaxed) as f64 / req as f64,
            self.bytes_out.load(Relaxed) / 1024,
            self.max_latency_us.load(Relaxed)
        )
    }
}

fn main() {
    let stats = ServerStats::default();
    thread::scope(|s| {
        for worker in 0..4u64 {
            let stats = &stats;
            s.spawn(move || {
                for i in 0..250u64 {
                    let status = if (worker * 250 + i) % 50 == 0 { 503 } else { 200 };
                    stats.record(status, 2048, 100 + (i * 7 + worker * 13) % 900);
                }
            });
        }
    });
    println!("{}", stats.summary());
}

use std::sync::LazyLock;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::thread;
use std::time::Duration;

struct Config {
    workers: usize,
    poll_ms: u64,
}

static CONFIG: LazyLock<Config> = LazyLock::new(|| Config { workers: 3, poll_ms: 2 }); // pretend: read env vars
static SHUTDOWN: AtomicBool = AtomicBool::new(false);
static SERVED: AtomicU64 = AtomicU64::new(0);

fn worker_loop(id: usize) -> (usize, u64) {
    let mut served = 0;
    while !SHUTDOWN.load(Ordering::Acquire) {
        served += 1; // "handle a request"
        SERVED.fetch_add(1, Ordering::Relaxed);
        thread::sleep(Duration::from_millis(CONFIG.poll_ms));
    }
    (id, served)
}

fn main() {
    let handles: Vec<_> = (0..CONFIG.workers).map(|id| thread::spawn(move || worker_loop(id))).collect();

    thread::sleep(Duration::from_millis(50));
    SHUTDOWN.store(true, Ordering::Release); // e.g. on Ctrl-C
    let per_worker: Vec<(usize, u64)> = handles.into_iter().map(|h| h.join().unwrap()).collect();

    let sum: u64 = per_worker.iter().map(|(_, n)| n).sum();
    println!("{} workers stopped", per_worker.len());
    println!("every worker served something: {}", per_worker.iter().all(|(_, n)| *n > 0));
    println!("totals agree: {}", sum == SERVED.load(Ordering::Relaxed));
}

use std::collections::HashSet;
use std::sync::atomic::{AtomicU64, Ordering};
use std::thread;

struct Sequence {
    next_block_start: AtomicU64,
    block_size: u64,
}

struct LocalIds<'a> {
    seq: &'a Sequence,
    next: u64,
    end: u64,
}

impl Sequence {
    fn local(&self) -> LocalIds<'_> {
        LocalIds { seq: self, next: 0, end: 0 }
    }
}

impl LocalIds<'_> {
    fn next_id(&mut self) -> u64 {
        if self.next == self.end {
            let start = self.seq.next_block_start.fetch_add(self.seq.block_size, Ordering::Relaxed);
            self.next = start;
            self.end = start + self.seq.block_size;
        }
        let id = self.next;
        self.next += 1;
        id
    }
}

fn main() {
    let seq = Sequence { next_block_start: AtomicU64::new(1), block_size: 100 };

    let all: Vec<u64> = thread::scope(|s| {
        let hs: Vec<_> = (0..4)
            .map(|_| {
                s.spawn(|| {
                    let mut ids = seq.local();
                    (0..250).map(|_| ids.next_id()).collect::<Vec<_>>()
                })
            })
            .collect();
        hs.into_iter().flat_map(|h| h.join().unwrap()).collect()
    });

    let unique: HashSet<_> = all.iter().collect();
    println!("{} ids, all unique: {}", all.len(), unique.len() == all.len());
    println!("blocks reserved: {}", (seq.next_block_start.load(Ordering::Relaxed) - 1) / seq.block_size);
}

use std::collections::HashMap;
use std::sync::OnceLock;
use std::sync::atomic::{AtomicU32, Ordering};
use std::thread;
use std::time::Duration;

struct Recommender {
    model: OnceLock<HashMap<&'static str, Vec<&'static str>>>,
    warmups: AtomicU32,
}

impl Recommender {
    fn model(&self) -> &HashMap<&'static str, Vec<&'static str>> {
        self.model.get_or_init(|| {
            self.warmups.fetch_add(1, Ordering::Relaxed);
            thread::sleep(Duration::from_millis(50)); // expensive load
            HashMap::from([("book", vec!["bookmark", "lamp"]), ("tent", vec!["sleeping bag", "stove"])])
        })
    }

    fn recommend(&self, item: &str) -> Vec<&'static str> {
        self.model().get(item).cloned().unwrap_or_default()
    }
}

fn main() {
    let rec = Recommender { model: OnceLock::new(), warmups: AtomicU32::new(0) };

    let rec = &rec;
    let answers: Vec<Vec<&str>> = thread::scope(|s| {
        let hs: Vec<_> = ["book", "tent", "book", "sofa"].map(|item| s.spawn(move || rec.recommend(item))).into_iter().collect();
        hs.into_iter().map(|h| h.join().unwrap()).collect()
    });

    println!("{answers:?}");
    println!("warm-ups performed: {}", rec.warmups.load(Ordering::Relaxed));
}

use std::sync::atomic::{AtomicUsize, Ordering};
use std::thread;
use std::time::Duration;

struct Latch {
    remaining: AtomicUsize,
}

impl Latch {
    fn count_down(&self) {
        self.remaining.fetch_sub(1, Ordering::Release);
    }

    fn wait(&self) {
        while self.remaining.load(Ordering::Acquire) > 0 {
            thread::yield_now();
        }
    }
}

fn main() {
    let latch = Latch { remaining: AtomicUsize::new(3) };
    let services = ["db", "cache", "search"];

    let started: Vec<&str> = thread::scope(|s| {
        let hs: Vec<_> = services
            .iter()
            .enumerate()
            .map(|(i, name)| {
                let latch = &latch;
                s.spawn(move || {
                    thread::sleep(Duration::from_millis(10 * (i as u64 + 1))); // start up
                    latch.count_down();
                    *name
                })
            })
            .collect();
        latch.wait();
        println!("all services up, accepting traffic");
        hs.into_iter().map(|h| h.join().unwrap()).collect()
    });
    println!("{started:?}");
}