Production

use std::time::Duration;

use nuclei::runtime::{RngSeed, UnhandledPanic};
use nuclei::{Builder, IoBackend};

fn main() -> std::io::Result<()> {
    let backend = if cfg!(target_os = "linux") {
        IoBackend::Epoll
    } else {
        nuclei::Proactor::backend()
    };
    let runtime = Builder::new()
        .worker_threads(4) // default: one per core
        .max_worker_threads(8) // default: four times worker_threads
        .pin_workers(true) // Linux; default true
        .thread_name("edge") // default nuclei-worker, nuclei-blocking
        .thread_stack_size(2 * 1024 * 1024) // default 8 MiB
        .max_nesting(64) // default 256
        .max_blocking_threads(128) // default 512
        .thread_keep_alive(Duration::from_secs(5)) // default 10 seconds
        .local_spawn_margin(16) // default 32
        .handoff_after(Some(Duration::from_millis(20))) // default 10 ms
        .io_buffer_cap(32 * 1024 * 1024) // default 64 MiB per worker
        .write_behind_linger(Duration::from_secs(5)) // default one minute
        .event_interval(31) // default 61
        .global_queue_interval(15) // default 31
        .max_io_events_per_tick(256) // default: all that are ready
        .disable_lifo_slot()
        .rng_seed(RngSeed::from_bytes(b"replay"))
        .unhandled_panic(UnhandledPanic::ShutdownRuntime)
        .backend(backend) // default: the process's choice
        .build()?;

    let used = runtime.block_on(runtime.spawn(async { nuclei::Proactor::backend() }));
    println!("workers run {used:?}");
    runtime.shutdown_timeout(Duration::from_secs(1));
    Ok(())
}

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

use nuclei::Builder;

fn main() -> std::io::Result<()> {
    let spawned = Arc::new(AtomicUsize::new(0));
    let polls = Arc::new(AtomicUsize::new(0));
    let runtime = Builder::new()
        .worker_threads(2)
        .on_thread_start(|| println!("worker up"))
        .on_task_spawn({
            let spawned = Arc::clone(&spawned);
            move |_task| {
                spawned.fetch_add(1, Ordering::Relaxed);
            }
        })
        .on_before_task_poll({
            let polls = Arc::clone(&polls);
            move |_task| {
                polls.fetch_add(1, Ordering::Relaxed);
            }
        })
        .build()?;

    runtime.block_on(async {
        for n in 0..10 {
            nuclei::spawn(async move { n }).await;
        }
    });
    assert_eq!(spawned.load(Ordering::Relaxed), 10);
    assert!(polls.load(Ordering::Relaxed) >= 10);
    Ok(())
}

use std::sync::mpsc;
use std::time::Duration;

use nuclei::{Builder, task};

fn main() -> std::io::Result<()> {
    let (told, heard) = mpsc::channel();
    let runtime = Builder::new()
        .worker_threads(1)
        .handoff_after(Some(Duration::from_millis(10)))
        // Called on the watch's thread for a poll that runs past the handoff.
        .on_long_poll(move |task| {
            let _ = told.send(task.name().map(str::to_owned));
        })
        .build()?;

    let name = runtime.block_on(async move {
        let blocker = task::Builder::new()
            .name("report")
            .worker(0)
            .spawn(async move {
                // A blocking call inside a poll, on purpose: it waits for the report of itself.
                heard.recv_timeout(Duration::from_secs(10))
            })?;
        Ok::<_, std::io::Error>(blocker.await)
    })?;
    assert_eq!(name, Ok(Some("report".to_owned())));
    Ok(())
}

use nuclei::Proactor;
use nuclei::config::{IoUringConfiguration, NucleiConfig};

fn main() {
    // Once, before any IO: the io_uring settings of every shard.
    Proactor::with_config(NucleiConfig {
        iouring: IoUringConfiguration {
            queue_len: 4096,
            recv_buffers: 512,
            recv_buffer_size: 16 * 1024,
            ..IoUringConfiguration::default()
        },
        backend: None, // io_uring where this process may have it, else the poller
    });
    nuclei::block_on(async {
        println!("IO runs on {:?}", Proactor::backend());
    });
}

#[nuclei::main]
async fn main() {
    nuclei::spawn(async {}).await;

    let stats = nuclei::stats();
    for shard in &stats.shards {
        println!(
            "{:?} on {:?}: {} tasks, {} polls, {} IO in flight, {} of {} buffer bytes",
            shard.worker,
            shard.backend,
            shard.tasks,
            shard.polls,
            shard.ops_in_flight,
            shard.buffer_bytes,
            shard.buffer_cap,
        );
        for (kind, count) in shard.errors.iter() {
            println!("  {count} errors of kind {kind:?}");
        }
    }
    println!("{} blocking threads", stats.blocking_threads);
}

use std::time::Duration;

use nuclei::Builder;
use nuclei::io::{AsyncReadExt, AsyncWriteExt};
use nuclei::net::{TcpListener, TcpStream};

fn main() -> std::io::Result<()> {
    let runtime = Builder::new()
        .worker_threads(1)
        .count_io(true) // the io-stats feature
        .record_latencies(true) // poll times; with io-stats, IO latencies too
        .build()?;
    runtime.block_on(runtime.spawn(async {
        let listener = TcpListener::bind("127.0.0.1:0").await?;
        let mut client = TcpStream::connect(listener.local_addr()?).await?;
        let (mut server, _) = listener.accept().await?;
        client.write_all(b"ping").await?;
        client.flush().await?;
        let mut buf = [0u8; 4];
        server.read_exact(&mut buf).await?;
        Ok::<_, std::io::Error>(())
    }))?;
    // A worker publishes its counts each time it looks at its IO and timers: a sleep makes it look.
    runtime.block_on(runtime.spawn(nuclei::time::sleep(Duration::from_millis(1))));

    let stats = runtime.stats();
    let worker = &stats.shards[0];
    let io = worker.io.unwrap();
    assert_eq!((io.accepts, io.bytes_read, io.bytes_written), (1, 4, 4));
    let polls = worker.poll_times.as_ref().unwrap();
    for (from, until, samples) in polls.buckets() {
        println!("{samples} polls of {from:?} to {until:?}");
    }
    // Readiness backends (epoll, kqueue) have no wait in the kernel to time.
    if let Some(io) = &worker.io_latencies {
        let timed = [&io.reads, &io.writes, &io.accepts, &io.connects, &io.files];
        println!(
            "{} IO operations timed",
            timed.iter().map(|h| h.count()).sum::<u64>()
        );
    }
    Ok(())
}

use std::time::Duration;

use nuclei::net::{TcpListener, TcpStream};
use nuclei::sync::CancellationToken;
use nuclei::task::JoinSet;
use nuclei::time::timeout;

async fn serve(listener: TcpListener, shutdown: CancellationToken) -> std::io::Result<()> {
    let mut connections = JoinSet::new();
    // Take connections until the token is cancelled.
    while let Some(accepted) = shutdown.run_until_cancelled(listener.accept()).await {
        let (stream, _) = accepted?;
        let stop = shutdown.child_token();
        connections.spawn(async move {
            let mut buf = [0u8; 1024];
            while let Some(Ok(n @ 1..)) = stop.run_until_cancelled(stream.recv(&mut buf)).await {
                let _ = stream.send(&buf[..n]).await;
            }
        });
    }
    // Let the open connections finish, for five seconds at most.
    let drained = timeout(Duration::from_secs(5), connections.join_all()).await;
    if drained.is_err() {
        eprintln!("some connections did not finish in time");
    }
    Ok(())
}

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let addr = listener.local_addr()?;
    let shutdown = CancellationToken::new();

    // Ctrl-C cancels the token.
    let on_ctrl_c = shutdown.clone();
    nuclei::spawn(async move {
        if nuclei::signal::ctrl_c().await.is_ok() {
            on_ctrl_c.cancel();
        }
    })
    .detach();

    let server = nuclei::spawn(serve(listener, shutdown.clone()));

    // Here a client comes and goes, then the program asks for the shutdown itself.
    let client = TcpStream::connect(addr).await?;
    client.send(b"hi").await?;
    let mut buf = [0u8; 2];
    client.recv(&mut buf).await?;
    shutdown.cancel();
    server.await
}

use std::time::Duration;

use nuclei::net::{TcpListener, TcpStream};
use nuclei::time::{self, Instant};

#[nuclei::test]
async fn echoes_a_message() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let addr = listener.local_addr()?;
    nuclei::spawn(async move {
        let (stream, _) = listener.accept().await?;
        let mut buf = [0u8; 64];
        let n = stream.recv(&mut buf).await?;
        stream.send(&buf[..n]).await?;
        std::io::Result::Ok(())
    })
    .detach();

    let client = TcpStream::connect(addr).await?;
    client.send(b"ping").await?;
    let mut buf = [0u8; 4];
    client.recv(&mut buf).await?;
    assert_eq!(&buf, b"ping");
    Ok(())
}

// A runtime of its own with one worker, for a test that needs a known shape.
#[nuclei::test(threads = 1)]
async fn times_out() {
    let never = std::future::pending::<()>();
    assert!(
        time::timeout(Duration::from_millis(10), never)
            .await
            .is_err()
    );
}

// A paused clock: the day-long timeout passes at once.
#[nuclei::test(start_paused = true)]
async fn a_day_passes_at_once() {
    let start = Instant::now();
    let never = std::future::pending::<()>();
    assert!(
        time::timeout(Duration::from_secs(86_400), never)
            .await
            .is_err()
    );
    assert_eq!(start.elapsed(), Duration::from_secs(86_400));
}

use nuclei::GlobalExecutorConfig;

fn main() {
    // The default runtime, set up the way earlier versions of nuclei did it.
    nuclei::init_with_config(
        GlobalExecutorConfig::default()
            .with_env_var("APP_WORKERS") // read the worker count from here first
            .with_min_threads(2)
            .with_max_threads(4),
    );
    nuclei::block_on(async {
        let started = nuclei::spawn_more_threads(2).await.expect("workers start");
        println!("started {started} more workers");
        let stopped = nuclei::stop_thread().await;
        println!("stopped one: {stopped}");
    });
}