Nuclei

Chapter 3 of 17

Runtime

The ways into the runtime: the attribute, block_on, and runtimes of your own.

The attribute

#[nuclei::main] turns an async main into a program: its body runs in block_on on the default runtime. It may return anything main may, such as std::io::Result<()>. Its options, comma separated:

  • threads = N: a runtime of its own with N workers; 0 runs every task on the thread inside block_on.
  • start_paused = true: a runtime of its own whose clock starts paused (see Time).
  • builder = "path": a runtime of its own, built from the Builder the function at path returns; threads and start_paused, when given, apply on top of it.
  • crate = "path": where the crate is, for code that re-exports it.
src/main.rs
#[nuclei::main(threads = 2)]
async fn main() {
    let stats = nuclei::stats();
    let workers = stats.shards.iter().filter(|s| s.worker.is_some()).count();
    println!("{workers} worker shards");
}
output
2 worker shards

With builder, every setting comes from a function of your own:

src/main.rs
// The runtime's settings, from a function of the program's own.
fn runtime() -> nuclei::Builder {
    nuclei::Builder::new()
        .worker_threads(2)
        .thread_name("server")
}

#[nuclei::main(builder = "runtime")]
async fn main() {
    let thread = nuclei::spawn(async { std::thread::current().name().map(str::to_owned) });
    assert_eq!(thread.await.as_deref(), Some("server"));
}

block_on

nuclei::block_on runs a future to its end on the calling thread. The first call, or the first spawn, starts the default runtime, so a plain fn main needs nothing else. (nuclei::drive is the same call under the name earlier versions used.)

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

fn main() {
    // The first call starts the default runtime: one worker shard per core.
    let sum = nuclei::block_on(async {
        nuclei::time::sleep(Duration::from_millis(10)).await;
        (1..=10).sum::<u32>()
    });
    assert_eq!(sum, 55);
}

block_on nests

block_on works anywhere, inside a task too. There it nests: the shard keeps running its other tasks, timers and IO until the future completes, so synchronous code deep in a call stack can wait on a future without holding the shard.

src/main.rs
/// A synchronous helper, deep in a call stack, that needs a future's output.
fn lookup(key: u32) -> u32 {
    nuclei::block_on(async move { key * 2 })
}

#[nuclei::main]
async fn main() {
    let task = nuclei::spawn(async {
        // block_on nests: the shard runs its other tasks while it waits.
        lookup(21)
    });
    assert_eq!(task.await, 42);
}

Two rules come with it. The task that called block_on is not polled again until the call returns, so a future that waits for that very task never completes. And each level runs on the thread's stack, so the depth is bounded by max_nesting; one level more panics with a message that says so.

A runtime of your own

Runtime::new() starts one with every setting at its default; Runtime::builder() sets them (every setting). A runtime spawns, runs block_on and runs blocking work, as the free functions do on the default runtime.

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

use nuclei::runtime::Runtime;

fn main() -> std::io::Result<()> {
    let runtime = Runtime::builder()
        .worker_threads(4)
        .thread_name("api-worker")
        .max_blocking_threads(64)
        .build()?;

    let task = runtime.spawn(async { "served" });
    assert_eq!(runtime.block_on(task), "served");

    // Stop the workers, waiting for them one second at most.
    runtime.shutdown_timeout(Duration::from_secs(1));
    Ok(())
}

Dropping a Runtime stops its workers, drops the tasks they still hold, and waits for the threads to exit. Bytes that dropped handles wrote behind still go out, each for at most write_behind_linger; one still under way when the drop is done carries on on a thread of its own. shutdown_timeout bounds the whole wait; shutdown_background does not wait at all.

build checks every setting first: a value no runtime can work with, or two that do not fit together, fails with InvalidInput naming the setting, its value and what it must be (Errors).

Handles and entering

runtime.handle() gives a RuntimeHandle: cheap to clone, usable from any thread to spawn on that runtime or to block_on. A handle does not keep its runtime running.

src/main.rs
use nuclei::runtime::{Runtime, RuntimeHandle};

fn main() -> std::io::Result<()> {
    let runtime = Runtime::new()?;
    let handle = runtime.handle();

    // A plain thread hands work to the runtime through its handle.
    let thread = std::thread::spawn(move || handle.block_on(handle.spawn(async { 2 + 2 })));
    assert_eq!(thread.join().unwrap(), 4);

    // Outside block_on this thread is inside no runtime...
    assert!(RuntimeHandle::try_current().is_err());
    {
        // ...until enter sets one: the free functions use it while the guard lives.
        let _guard = runtime.enter();
        assert_eq!(RuntimeHandle::try_current().unwrap().id(), runtime.id());
        let task = nuclei::spawn(async { "on the entered runtime" });
        assert_eq!(runtime.block_on(task), "on the entered runtime");
    }
    Ok(())
}
  • RuntimeHandle::current(): the runtime this thread is inside, or the default one.
  • RuntimeHandle::try_current(): the runtime this thread is inside, or an error; it never starts the default one.
  • runtime.enter(): makes the free functions (spawn, spawn_blocking, ...) use this runtime on this thread until the guard drops.
  • id(): a RuntimeId, different for every runtime of the process.

Choosing the worker

spawn picks the shard for you. To pick it yourself, name the worker: indices run from 0 to workers(), and worker_index() says which one the calling code runs on.

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

use nuclei::runtime::{Runtime, worker_index};

fn main() -> std::io::Result<()> {
    let runtime = Runtime::builder().worker_threads(2).build()?;
    let handle = runtime.handle();
    assert_eq!(handle.workers(), 2);

    // A task placed on worker 1 runs there for its whole life.
    let task = handle.spawn_on_worker(1, async { worker_index() })?;
    assert_eq!(runtime.block_on(task), Some(1));

    // One task per worker, its future made there: it need not be Send.
    let each = handle.on_each_worker(|| async {
        let here = Rc::new(worker_index());
        nuclei::task::yield_now().await;
        *here
    });
    let mut seen = runtime.block_on(each.join_all());
    seen.sort();
    assert_eq!(seen, [Some(0), Some(1)]);

    // A worker the runtime does not have is refused.
    assert!(handle.spawn_on_worker(2, async {}).is_err());
    Ok(())
}
  • spawn_on_worker(k, future) places a task on worker k, where it stays for its whole life: a handoff moves the worker's shard to another thread with its tasks, and worker_index() still answers k.
  • spawn_local_on_worker(k, make) runs make on worker k and spawns the future it makes there, so the future need not be Send.
  • on_each_worker(make) does that once on every worker that takes tasks and gives a JoinSet with one task per worker: a listener per core, a cache per core, a mesh of channels between them.
  • task::Builder::worker(k) places a task spawned with options the same way.

A worker the runtime does not have, or one that stopped, refuses the task with InvalidInput; a task it took is never dropped unrun.

A value per worker

shard_local! declares a static, a ShardLocal, that holds one value per worker, made on its first use there: a counter, a cache or a buffer that the worker's tasks share without a lock. The value lives in the worker's shard, not in its thread, so it moves with the shard on a handoff and the worker's tasks always see the same one.

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

use nuclei::runtime::Runtime;

nuclei::shard_local! {
    // One counter per worker, made on its first use there.
    static SERVED: Cell<u64> = Cell::new(0);
}

fn main() -> std::io::Result<()> {
    let runtime = Runtime::builder().worker_threads(2).build()?;
    let handle = runtime.handle();
    for worker in [0, 0, 0, 1] {
        let served = handle.spawn_on_worker(worker, async {
            SERVED.with(|n| n.set(n.get() + 1));
        })?;
        runtime.block_on(served);
    }

    // Each worker reads its own counter.
    let counts = handle.on_each_worker(|| async { SERVED.with(Cell::get) });
    let mut counts = runtime.block_on(counts.join_all());
    counts.sort();
    assert_eq!(counts, [1, 3]);

    // Code that runs on no worker has no value.
    assert!(SERVED.try_with(Cell::get).is_err());
    Ok(())
}

with runs a closure on this worker's value, and the shard stays on the thread while it runs: keep the closure short. Only code on a worker reaches one: off the workers (in block_on's own thread, on the blocking pool, on another thread) try_with answers a ShardAccessError and with panics.

The default runtime, configured

The free functions use the runtime the calling thread is inside, or else the default runtime. To configure the default one, build it with build_global before anything starts it.

src/main.rs
fn main() -> std::io::Result<()> {
    // Configure the default runtime before anything starts it.
    nuclei::Builder::new().worker_threads(2).build_global()?;

    // From here on the free functions use it.
    let task = nuclei::spawn(async { "on the default runtime" });
    assert_eq!(nuclei::block_on(task), "on the default runtime");
    Ok(())
}

build_global fails with the default runtime is already running once a free function, the attribute or an earlier build_global started it: call it first thing in main.