Nuclei

Chapter 11 of 17

Sync

Locks, semaphores, notifications, once-cells and lock-free structures for async code, and twins of them for one shard.

Futures and wakers, nothing else

Every wait in nuclei::sync is a future built from futures and wakers alone, so it works under any executor, not only Nuclei's. A contended wait queues and returns Pending; it never blocks a thread.

  • Familiar names. Mutex, RwLock, Semaphore, Notify, Barrier, OnceCell and SetOnce keep the names, signatures and errors of the established async sync API: code written against it compiles after a change of import.
  • Every wait has a try_ form that never waits, and a blocking_ form for threads that may block. On a Nuclei shard a blocking_ call nests a block_on, so the shard keeps running its other tasks.
  • Dropping a wait loses nothing. A dropped future withdraws its place in line and passes on what was handed to it, so no wakeup is lost. A Barrier wait is the one exception: an arrival that was counted stays counted.
  • Fair to the shard. On a Nuclei shard, a lock, an acquire or a receive that completes spends the task's budget, so a loop on one that is always ready still lets other tasks run. Under another executor there is no budget.

Mutex

src/main.rs
use std::sync::Arc;

use nuclei::sync::Mutex;

#[nuclei::main]
async fn main() {
    let counter = Arc::new(Mutex::new(0u32));
    let tasks: Vec<_> = (0..8)
        .map(|_| {
            let counter = Arc::clone(&counter);
            nuclei::spawn(async move {
                // lock waits as a future: the shard runs other tasks meanwhile.
                *counter.lock().await += 1;
            })
        })
        .collect();
    for task in tasks {
        task.await;
    }
    assert_eq!(*counter.lock().await, 8);

    // An owned guard holds the lock by an Arc, so it can be moved into a task.
    let guard = Arc::clone(&counter).lock_owned().await;
    nuclei::spawn(async move { assert_eq!(*guard, 8) }).await;

    // try_lock never waits.
    assert!(counter.try_lock().is_ok());
}

Uncontended, taking and releasing the lock is one atomic operation each. It does not poison: a panic in a holder unlocks it. lock_owned holds the lock through an Arc, so the guard can move into a task; MutexGuard::map narrows a guard to a part of the value.

RwLock

Readers share the lock, a writer has it alone. An upgradable read sits beside plain readers (one at a time) and turns into a write without letting another writer in between: read, decide, then write.

src/main.rs
use nuclei::sync::{RwLock, RwLockUpgradableReadGuard};

#[nuclei::main]
async fn main() {
    let config = RwLock::new(String::from("v1"));

    // Readers share the lock.
    let a = config.read().await;
    let b = config.read().await;
    assert_eq!((a.as_str(), b.as_str()), ("v1", "v1"));
    drop((a, b));

    // An upgradable read sits beside plain readers, and becomes a write
    // without letting another writer in between.
    let check = config.upgradable_read().await;
    if check.as_str() == "v1" {
        let mut write = RwLockUpgradableReadGuard::upgrade(check).await;
        write.push_str("+patch");
    }
    assert_eq!(*config.read().await, "v1+patch");
}

A write guard downgrades to a read guard, and with_max_readers bounds how many readers share it.

Reentrant locks

ReentrantMutex and ReentrantRwLock may be taken again by the holder that has them. The holder is an Owner token, not a thread: a task can move between threads, and many tasks can share one.

src/main.rs
use nuclei::sync::{Owner, ReentrantMutex};

#[nuclei::main]
async fn main() {
    let state = ReentrantMutex::new(Vec::<&str>::new());
    // The lock is held by an owner, not a thread: tasks move between threads.
    let me = Owner::new();

    let outer = state.lock(&me).await;
    // The same owner takes it again without waiting for itself.
    let inner = state.lock(&me).await;
    assert!(inner.is_empty() && outer.is_empty());
    drop(inner);
    drop(outer);

    // Another owner waits until every hold of the first is gone.
    let other = Owner::new();
    assert!(state.try_lock(&other).is_some());
}

Semaphore

A semaphore hands out permits; a permit goes back when it drops. It bounds how many tasks do something at once.

src/main.rs
use std::sync::Arc;
use std::time::Duration;

use nuclei::sync::{Mutex, Semaphore};
use nuclei::time::sleep;

#[nuclei::main]
async fn main() {
    // At most two jobs at once, however many are spawned.
    let slots = Arc::new(Semaphore::new(2));
    let running = Arc::new(Mutex::new((0u32, 0u32))); // (now, most at once)
    let jobs: Vec<_> = (0..6)
        .map(|_| {
            let (slots, running) = (Arc::clone(&slots), Arc::clone(&running));
            nuclei::spawn(async move {
                let _permit = slots.acquire_owned().await.unwrap();
                {
                    let mut r = running.lock().await;
                    r.0 += 1;
                    r.1 = r.1.max(r.0);
                }
                sleep(Duration::from_millis(5)).await;
                running.lock().await.0 -= 1;
                // The permit goes back when it drops.
            })
        })
        .collect();
    for job in jobs {
        job.await;
    }
    assert_eq!(running.lock().await.1, 2);
    assert_eq!(slots.available_permits(), 2);
}

acquire_many takes several at once, forget keeps a permit from going back, add_permits adds more, and close makes every wait fail.

Notify, Event, Latch, Barrier

TypeWaits end when
Notifynotify_one wakes the oldest waiter, or keeps one permit for the next when nobody waits; notify_waiters wakes every waiter there is.
Eventit is set; it stays set, so later waits end at once.
Latchits count, counted down by count_down, reaches zero; it stays open.
Barriern tasks wait at once; one of them is told it is the leader.
src/main.rs
use std::sync::Arc;

use nuclei::sync::{Barrier, Event, Latch, Notify};

#[nuclei::main]
async fn main() {
    // Notify: wake one waiter; a notify with nobody waiting is kept for the next.
    let notify = Arc::new(Notify::new());
    let waiter = {
        let notify = Arc::clone(&notify);
        nuclei::spawn(async move { notify.notified().await })
    };
    notify.notify_one();
    waiter.await;

    // Event: set once, and every wait, now or later, ends.
    let ready = Arc::new(Event::new());
    let watcher = {
        let ready = Arc::clone(&ready);
        nuclei::spawn(async move { ready.wait().await })
    };
    ready.set();
    watcher.await;
    ready.wait().await;

    // Latch: waits end once the count reaches zero.
    let started = Arc::new(Latch::new(3));
    for _ in 0..3 {
        let started = Arc::clone(&started);
        nuclei::spawn(async move { started.count_down() }).detach();
    }
    started.wait().await;

    // Barrier: each wait ends once n tasks are waiting; one is the leader.
    let barrier = Arc::new(Barrier::new(2));
    let other = {
        let barrier = Arc::clone(&barrier);
        nuclei::spawn(async move { barrier.wait().await.is_leader() })
    };
    let mine = barrier.wait().await.is_leader();
    assert!(mine != other.await);
}

Once-cells and lazy values

OnceCell holds a value set once; get_or_init runs an async initializer, and callers that come meanwhile wait for it. SetOnce is set from anywhere, and wait waits until someone has. Lazy computes its value on first use.

src/main.rs
use std::time::Duration;

use nuclei::sync::{Lazy, OnceCell, SetOnce};

static TABLE: Lazy<Vec<u32>> = Lazy::new(|| (0..10).map(|n| n * n).collect());
static SETTINGS: OnceCell<String> = OnceCell::const_new();

async fn load_settings() -> String {
    nuclei::time::sleep(Duration::from_millis(5)).await;
    "port=7000".to_string()
}

#[nuclei::main]
async fn main() {
    // Lazy: computed on first use, the same value after.
    assert_eq!(TABLE[3], 9);

    // OnceCell: the first caller runs the async initializer; others wait for it.
    let (a, b) = nuclei::join!(
        SETTINGS.get_or_init(load_settings),
        SETTINGS.get_or_init(load_settings),
    );
    assert!(std::ptr::eq(a, b));

    // SetOnce: set once from anywhere; wait until someone has.
    let shutdown_reason = SetOnce::new();
    shutdown_reason.set("maintenance").unwrap();
    assert_eq!(*shutdown_reason.wait().await, "maintenance");
    assert!(shutdown_reason.set("again").is_err());
}

Fast, and fair by count

The locks and the semaphore let a caller take what is free even while others wait, so a release never waits for a sleeping task to run. A waiter that loses MAX_BYPASS times is owed a handoff: nobody gets in ahead of it, and the next release goes to it directly. Fairness is counted, never timed.

The task's budget

On a Nuclei shard, a lock or a semaphore acquire that completes spends one operation of the task's budget, as IO does, and so do a channel receive that completes and a bounded send that completes at its first try. Once the budget is spent the call completes nothing: it wakes its task and answers Pending, so the task goes behind the shard's other tasks.

src/main.rs
use std::cell::Cell;
use std::rc::Rc;

use nuclei::sync::mpsc;

#[nuclei::main]
async fn main() {
    let (tx, mut rx) = mpsc::unbounded_channel();
    for value in 0..1000u32 {
        tx.send(value).unwrap();
    }
    drop(tx);

    let ran = Rc::new(Cell::new(false));
    let other = nuclei::spawn_local({
        let ran = Rc::clone(&ran);
        async move { ran.set(true) }
    });

    // Every receive is ready at once, yet the loop yields when its budget is
    // spent, so the other task runs while values are still waiting.
    let (mut received, mut ran_after) = (0, None);
    while rx.recv().await.is_some() {
        received += 1;
        if ran.get() && ran_after.is_none() {
            ran_after = Some(received);
        }
    }
    assert!(ran_after.is_some_and(|n| n < 1000));
    other.await;
}
  • A wait that queued and was then granted spends nothing, so it keeps what it waited for.
  • The try_ forms never spend. Notify, Event, Latch, Barrier and the once-cells keep no budget.
  • Off a Nuclei shard, under another executor, there is no budget, and these calls never yield for it.
  • coop::unconstrained runs a future with no budget, and budget_yields counts how often a shard's tasks yielded for it.

For the tasks of one shard

nuclei::sync::local has twins of mpsc, oneshot, Semaphore and Mutex for tasks that share one thread: spawn_local tasks, and the future block_on runs. Their state sits in an Rc and plain cells, so no path makes an atomic instruction or a fence, and a message between two tasks takes fewer instructions than on the shared channel (CI counts both, in the mpsc and local_mpsc rows of ci/op_budget.txt).

src/main.rs
use std::rc::Rc;

use nuclei::sync::local::{Mutex, mpsc, oneshot};

#[nuclei::main]
async fn main() {
    // Rc, not Arc: these tasks all stay on this thread's shard.
    let log = Rc::new(Mutex::new(Vec::new()));
    let (tx, mut rx) = mpsc::channel(2);
    for n in 1..=3u32 {
        let (tx, log) = (tx.clone(), Rc::clone(&log));
        nuclei::spawn_local(async move {
            log.lock().await.push(n);
            tx.send(n).await.unwrap();
        })
        .detach();
    }
    drop(tx);

    // A oneshot carries the sum back once every sender is gone.
    let (reply, sum) = oneshot::channel();
    nuclei::spawn_local(async move {
        let mut total = 0;
        while let Some(n) = rx.recv().await {
            total += n;
        }
        reply.send(total).unwrap();
    })
    .detach();

    assert_eq!(sum.await, Ok(6));
    assert_eq!(log.lock().await.len(), 3);
}
  • Not Send. The types are neither Send nor Sync, so the compiler keeps them on the thread that made them. Share one between tasks with an Rc.
  • The same names and errors. Code switches by changing the path. Left out are the blocking_ forms, which on one thread could only wait for themselves, and the weak sender.
  • First come, first served. Waiting sends, acquires and locks are served strictly in the order they came, and a try_ form fails while anyone waits. The shared locks let a caller in ahead of a waiter (fair by count); these never do.
  • Dropping a wait loses nothing, and the budget applies as it does to the shared ones.

Lock-free maps and queues

For state shared across threads without a lock at all:

  • HopscotchMap: a concurrent hash map; values are cloned out.
  • SegQueue: an unbounded queue. ArrayQueue: a bounded one whose push refuses when full.
  • Atom: a value read without a lock and replaced whole.
src/main.rs
use std::sync::Arc;

use nuclei::sync::{ArrayQueue, Atom, HopscotchMap, SegQueue};

#[nuclei::main]
async fn main() {
    // A concurrent map: every operation is lock-free, values are cloned out.
    let sessions = Arc::new(HopscotchMap::new());
    let writers: Vec<_> = (0..4u32)
        .map(|n| {
            let sessions = Arc::clone(&sessions);
            nuclei::spawn(async move { sessions.insert(n, format!("user-{n}")) })
        })
        .collect();
    for writer in writers {
        writer.await;
    }
    assert_eq!(sessions.get(&2).as_deref(), Some("user-2"));
    assert_eq!(sessions.len(), 4);

    // Queues: unbounded, and bounded with a push that refuses when full.
    let work = SegQueue::new();
    work.push("job");
    assert_eq!(work.pop(), Some("job"));
    let ring = ArrayQueue::new(1);
    assert!(ring.push(1).is_ok());
    assert_eq!(ring.push(2), Err(2));

    // Atom: a value read without a lock and replaced whole.
    let config = Atom::new(String::from("v1"));
    config.store(String::from("v2"));
    assert_eq!(config.load_clone(), "v2");
}