Tokio

# Cargo.toml
[dependencies]
tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "io-util", "time", "sync", "signal"] }
tokio-util = { version = "0.7", features = ["rt"] }   # CancellationToken; the "rt" feature adds TaskTracker

use tokio::time::{sleep, timeout, Duration};

fn checksum_all(files: Vec<String>) -> u64 {
    files.iter().map(|f| f.len() as u64).sum() // stands in for CPU-heavy work
}

#[tokio::main]
async fn main() {
    let a = tokio::spawn(async {
        sleep(Duration::from_secs(1)).await;
        "invoice sync done"
    });
    let b = tokio::spawn(async {
        sleep(Duration::from_millis(500)).await;
        "cache warm done"
    });
    // join! waits for both; each JoinHandle yields Result<T, JoinError>
    let (ra, rb) = tokio::join!(a, b);
    println!("{}, {}", ra.unwrap(), rb.unwrap());

    // select! runs both branches until one finishes, then drops the other
    tokio::select! {
        _ = sleep(Duration::from_secs(1)) => println!("slow path"),
        _ = sleep(Duration::from_millis(100)) => println!("fast path"),
    }

    // timeout wraps any future; Err(Elapsed) if it does not finish in time
    match timeout(Duration::from_millis(200), sleep(Duration::from_secs(5))).await {
        Ok(_) => println!("finished"),
        Err(_) => println!("timed out after 200ms"),
    }

    // CPU-heavy or blocking std code goes to the blocking pool
    let total = tokio::task::spawn_blocking(|| checksum_all(vec!["a.log".into(), "b.log".into()]))
        .await
        .unwrap();
    println!("{total}");
}

use tokio::sync::{broadcast, mpsc, oneshot, watch};

#[tokio::main]
async fn main() {
    // mpsc: many producers, one consumer; bounded, so send().await applies backpressure
    let (tx, mut rx) = mpsc::channel::<String>(100);
    for id in 0..3 {
        let tx = tx.clone();
        tokio::spawn(async move { tx.send(format!("worker {id} ready")).await.unwrap(); });
    }
    drop(tx); // recv() returns None only after every sender is dropped
    while let Some(msg) = rx.recv().await {
        println!("{msg}");
    }

    // oneshot: one value, one reply (request/response)
    let (reply_tx, reply_rx) = oneshot::channel::<u32>();
    tokio::spawn(async move { let _ = reply_tx.send(42); });
    println!("answer {}", reply_rx.await.unwrap());

    // broadcast: every receiver sees every message; slow receivers get RecvError::Lagged
    let (btx, mut brx1) = broadcast::channel::<String>(16);
    let mut brx2 = btx.subscribe();
    btx.send("deploy started".into()).unwrap();
    println!("{} / {}", brx1.recv().await.unwrap(), brx2.recv().await.unwrap());

    // watch: latest value only, for config or state that changes
    let (wtx, mut wrx) = watch::channel(String::from("v1"));
    wtx.send("v2".into()).unwrap();
    wrx.changed().await.unwrap();
    println!("config now {}", *wrx.borrow());
}

use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
use tokio_util::sync::CancellationToken;
use tokio_util::task::TaskTracker;

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:8080").await?;
    let token = CancellationToken::new();
    let tracker = TaskTracker::new();

    let shutdown = token.clone();
    tokio::spawn(async move {
        tokio::signal::ctrl_c().await.expect("failed to listen for ctrl-c");
        shutdown.cancel();
    });

    loop {
        tokio::select! {
            _ = token.cancelled() => break,
            accepted = listener.accept() => {
                let (mut socket, addr) = accepted?;
                println!("connection from {addr}");
                let conn_token = token.clone();
                tracker.spawn(async move {
                    let mut buf = [0u8; 1024];
                    loop {
                        tokio::select! {
                            _ = conn_token.cancelled() => return,
                            read = socket.read(&mut buf) => match read {
                                Ok(0) | Err(_) => return,
                                Ok(n) => if socket.write_all(&buf[..n]).await.is_err() { return },
                            },
                        }
                    }
                });
            }
        }
    }

    tracker.close();
    tracker.wait().await; // let in-flight connections finish
    Ok(())
}

use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{Mutex, RwLock, Semaphore};

#[tokio::main]
async fn main() {
    let queue = Arc::new(Mutex::new(vec![1, 2, 3]));
    let q = queue.clone();
    tokio::spawn(async move { q.lock().await.push(4); }).await.unwrap();

    let cache: Arc<RwLock<HashMap<String, String>>> = Arc::new(RwLock::new(HashMap::new()));
    cache.write().await.insert("region".into(), "eu-west-1".into());
    println!("{:?}", cache.read().await.get("region"));

    // At most 10 requests in flight
    let limit = Arc::new(Semaphore::new(10));
    let mut jobs = Vec::new();
    for page in 0..50 {
        let permit = limit.clone().acquire_owned().await.unwrap();
        jobs.push(tokio::spawn(async move {
            let _permit = permit; // released when the task ends
            page * 2             // stands in for an HTTP request
        }));
    }
    for j in jobs { j.await.unwrap(); }
}