Select and join: Learn Nuclei

join! and try_join!

nuclei::join! polls every future given, all at once, in the task that awaits it, and gives their outputs as a tuple in the order written. nuclei::try_join! does the same with futures that give a Result: it gives Ok with every output, or the first error at once, dropping the rest.

src/main.rs

use std::time::Duration;

use nuclei::time::sleep;

async fn fetch(name: &'static str, ms: u64) -> &'static str {
    sleep(Duration::from_millis(ms)).await;
    name
}

async fn parse(text: &str) -> Result<u32, std::num::ParseIntError> {
    text.parse()
}

#[nuclei::main]
async fn main() {
    // join! runs both at once in this task: about 30 ms, not 50.
    let (user, orders) = nuclei::join!(fetch("user", 30), fetch("orders", 20));
    assert_eq!((user, orders), ("user", "orders"));

    // try_join! gives every output, or the first error.
    let both = nuclei::try_join!(parse("4"), parse("2"));
    assert_eq!(both, Ok((4, 2)));
    assert!(nuclei::try_join!(parse("4"), parse("x")).is_err());
}

The futures run concurrently, not in parallel: they share the task. To use more cores, spawn them. Each poll starts one future later than the last, so a future that is always ready cannot starve the others; biased; as the first argument polls in the order written.

select!

nuclei::select! waits on several branches and runs the handler of the first to finish, dropping the other futures. Each branch is pattern = future => handler.

src/main.rs

use std::time::Duration;

use nuclei::sync::mpsc;
use nuclei::time::sleep;

#[nuclei::main]
async fn main() {
    let (tx, mut jobs) = mpsc::channel::<u32>(8);
    nuclei::spawn(async move {
        for job in 1..=3 {
            tx.send(job).await.unwrap();
        }
    })
    .detach();

    let mut done = Vec::new();
    loop {
        // The first branch to finish runs; the other futures are dropped.
        nuclei::select! {
            Some(job) = jobs.recv() => done.push(job),
            _ = sleep(Duration::from_secs(1)) => break,
            // Every branch disabled (the channel closed and gave None): else.
            else => break,
        }
    }
    assert_eq!(done, [1, 2, 3]);
}

Patterns filter. A future whose output does not match its pattern disables its branch, and the others go on: Some(job) = jobs.recv() stops matching once the channel closes.
else ends it. When every branch is disabled, the else branch runs; without one, select! panics.
Preconditions. pattern = future, if condition => handler leaves a branch out when the condition is false.
Fair by default. Polling starts at a random branch; biased; first polls in the order written.
Handlers are plain code. They run after the futures are dropped, so return, ?, break and continue act on the code around the select!.

What a losing branch loses

A branch that loses is dropped where it stood. Nuclei's IO loses no byte when dropped (see How it works), and a sync wait or a channel receive withdraws without losing a value or a wakeup; a Barrier wait is the one exception, its arrival stays counted. State a future of your own holds halfway is dropped with it.