Highest level          tokio::sync::mpsc / broadcast / watch
                       (message passing — default for new code)

                       Arc<Mutex<T>> / Arc<RwLock<T>>
                       (shared mutable state — common, easy to get right)

                       parking_lot::{Mutex, RwLock, Condvar}
                       (faster sync locks, no poisoning)

                       Atomics (AtomicUsize, AtomicBool, AtomicPtr)
                       (single-word lock-free state)

Lowest level           UnsafeCell + unsafe + loom + miri
                       (custom lock-free / wait-free primitives)

Need to share state between tasks?
├── State is configuration (read-only after start)
│   └── Arc<Config>  (no lock needed)
├── State is a queue of work
│   └── tokio::sync::mpsc::channel(cap)
├── State is "latest value" published to many readers
│   └── tokio::sync::watch::channel(initial)
├── State is broadcast (every consumer sees every value)
│   └── tokio::sync::broadcast::channel(cap)
├── State is request-response within one task tree
│   └── tokio::sync::oneshot::channel()
├── State is a counter
│   └── AtomicU64 (or AtomicUsize)
├── State is a flag / set-once
│   └── AtomicBool / OnceLock<T> / OnceCell<T>
├── State needs mutation across many tasks/threads, cheap critical sections
│   ├── async context           → tokio::sync::Mutex<T>
│   └── sync context (no .await held) → parking_lot::Mutex<T>
├── State needs mutation, many readers, few writers
│   ├── async context           → tokio::sync::RwLock<T>
│   └── sync context            → parking_lot::RwLock<T>
└── State is a custom lock-free primitive (channels, hazard pointers)
    └── UnsafeCell + atomics + loom-tested + miri-tested + a co-author

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

let c = AtomicUsize::new(0);

// Just need a count, no synchronization with other data
c.fetch_add(1, Ordering::Relaxed);

// Reading a counter that was incremented from elsewhere
let n = c.load(Ordering::Relaxed);

static READY: AtomicBool = AtomicBool::new(false);
static mut DATA: Option<Config> = None;

// Producer thread:
unsafe { DATA = Some(load_config()); }
READY.store(true, Ordering::Release);

// Consumer thread:
if READY.load(Ordering::Acquire) {
    // SAFETY: producer's Release pairs with our Acquire; if we see READY=true,
    // we are guaranteed to also see the DATA write that happened-before it.
    let cfg = unsafe { DATA.as_ref().unwrap() };
}

let m = std::sync::Mutex::new(0u64);
let guard = m.lock().unwrap();
something_async().await;  // ❌ guard is held across await
*guard += 1;

{
    let mut guard = m.lock().unwrap();
    *guard += 1;
}  // guard released
something_async().await;

let (tx, mut rx) = tokio::sync::mpsc::channel::<Job>(256);

tokio::spawn(async move {
    while let Some(job) = rx.recv().await {
        process(job).await;
    }
});

tx.send(Job::new()).await?;  // backpressure: awaits if full

let (tx, mut rx) = tokio::sync::watch::channel(Config::default());

// Producer:
tx.send(new_config)?;

// Consumer:
loop {
    rx.changed().await?;
    let cfg = rx.borrow();
    apply(&cfg);
}

let (tx, _) = tokio::sync::broadcast::channel::<Event>(1024);
let mut rx1 = tx.subscribe();
let mut rx2 = tx.subscribe();

while let Ok(event) = rx1.recv().await {
    // ...
}

let (tx, rx) = tokio::sync::oneshot::channel::<Response>();
worker.send(Request { reply: tx }).await?;
let response = rx.await?;

let sem = Arc::new(tokio::sync::Semaphore::new(10));

for task in tasks {
    let permit = sem.clone().acquire_owned().await?;
    tokio::spawn(async move {
        let _hold = permit;       // released when task exits
        process(task).await
    });
}

let shared = Arc::new(BigData::new());
for _ in 0..workers {
    let s = shared.clone();
    tokio::spawn(async move { use_data(&s).await });
}

use std::sync::{OnceLock, LazyLock};

// Lazy initialization, computed on first read
static CONFIG: LazyLock<Config> = LazyLock::new(|| Config::load_from_env().unwrap());

fn get_config() -> &'static Config {
    &CONFIG
}

// One-shot publication, set explicitly
static DB: OnceLock<sqlx::PgPool> = OnceLock::new();

#[tokio::main]
async fn main() {
    let pool = sqlx::PgPool::connect(&env_url()).await.unwrap();
    DB.set(pool).expect("only set once");
    // Now everywhere: DB.get().unwrap()
}

[target.'cfg(loom)'.dev-dependencies]
loom = "0.7"

#[cfg(loom)]
use loom::sync::atomic::{AtomicUsize, Ordering};
#[cfg(not(loom))]
use std::sync::atomic::{AtomicUsize, Ordering};

#[cfg(loom)]
use loom::sync::Arc;
#[cfg(not(loom))]
use std::sync::Arc;

#[cfg(loom)]
mod loom_tests {
    use super::*;
    use loom::thread;

    #[test]
    fn concurrent_push_pop_preserves_order() {
        loom::model(|| {
            let queue = Arc::new(MyQueue::new());
            let q1 = queue.clone();
            let q2 = queue.clone();
            let h1 = thread::spawn(move || q1.push(1));
            let h2 = thread::spawn(move || q2.pop());
            h1.join().unwrap();
            h2.join().unwrap();
            // Assert the invariant: queue is in a coherent state.
        });
    }
}

RUSTFLAGS="--cfg loom" cargo test --release -- --test-threads 1

struct MyHandle { raw: *mut FfiObject }

// SAFETY: FfiObject's documented contract states that move-between-threads
// is safe as long as concurrent use is externally synchronized. We do not
// implement Sync because the FFI object is single-threaded once obtained.
unsafe impl Send for MyHandle {}
// Do NOT impl Sync — the FFI is not thread-safe.