Tokio / Chapter 5 of 6

Signals, processes, and concurrency

Use Tokio signals, child processes, UDP, semaphores, watch, and broadcast. Handle blocking work and non-Send tasks with Rust examples.

9 topicsUpdated 22 Oct 2026

16 / 26

watch and broadcast#

watch keeps the latest state; broadcast fans out each message to subscribers.

Primitive	Who receives?	Typical use
`watch`	Each receiver sees the current value and can observe changes	Configuration, health, current state
`broadcast`	Every active receiver gets each new message, subject to buffer lag	Notifications, event fan-out
`mpsc`	Exactly one receiver handles each message	Work queue
`oneshot`	One receiver gets one value	A single request reply

Rustwatch: last value wins

use tokio::sync::watch;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = watch::channel(0usize);
    let listener = tokio::spawn(async move {
        rx.changed().await.unwrap();
        println!("latest: {}", *rx.borrow_and_update());
    });
    tx.send_replace(42);
    listener.await.unwrap();
}

Rustbroadcast: two subscribers

use tokio::sync::broadcast;

#[tokio::main]
async fn main() {
    let (tx, mut a) = broadcast::channel::<String>(16);
    let mut b = tx.subscribe();
    tx.send("deployed".to_owned()).unwrap();
    println!("a: {}", a.recv().await.unwrap());
    println!("b: {}", b.recv().await.unwrap());
}

Lag is observable

A slow broadcast receiver may get RecvError::Lagged(n) when older messages are overwritten. Handle it if every event matters. For watch, use borrow_and_update() after changed() to mark the value seen.

17 / 26

Semaphore and Notify#

Semaphore limits concurrent work; Notify wakes a task when a condition changes.

A permit represents one available slot. Acquire before work, and dropping the permit releases it. Use acquire_owned() when the permit must move into a spawned task. Notify is a wake-up signal, not a queue of arbitrary messages.

RustLimit active jobs to four

use std::{sync::Arc, time::Duration};
use tokio::{sync::Semaphore, task::JoinSet};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let limit = Arc::new(Semaphore::new(4));
    let mut tasks = JoinSet::new();
    for job in 0..40 {
        let permit = Arc::clone(&limit).acquire_owned().await?;
        tasks.spawn(async move {
            let _permit = permit; // drop releases it
            tokio::time::sleep(Duration::from_millis(20)).await;
            job
        });
    }
    while let Some(result) = tasks.join_next().await {
        println!("finished {}", result?);
    }
    Ok(())
}

RustOne signal wakes a waiter

use std::sync::Arc;
use tokio::sync::Notify;

#[tokio::main]
async fn main() {
    let notify = Arc::new(Notify::new());
    let waiting = Arc::clone(&notify);
    let worker = tokio::spawn(async move {
        waiting.notified().await;
        println!("work is available");
    });
    notify.notify_one();
    worker.await.unwrap();
}

Choosing the primitive

Use a bounded mpsc channel when each notification has a payload or multiple queued items. Notify is useful with separate protected state; notify_one() can store one permit if no waiter is ready.

18 / 26

Line protocols and framed messages#

TCP is a byte stream: a single read is not necessarily a complete message.

Decide how messages end: newline, a length prefix, or another frame format. BufReader::lines() handles newline-delimited UTF-8; it strips the line ending. Set size limits before accepting untrusted, unbounded input.

RustLine-based TCP handler

use tokio::{
    io::{self, AsyncBufReadExt, AsyncWriteExt, BufReader},
    net::TcpStream,
};

async fn echo_lines(stream: TcpStream) -> io::Result<()> {
    let (reader, mut writer) = stream.into_split();
    let mut lines = BufReader::new(reader).lines();
    while let Some(line) = lines.next_line().await? {
        if line.len() > 4096 {
            return Err(io::Error::new(io::ErrorKind::InvalidData, "too long"));
        }
        writer.write_all(line.as_bytes()).await?;
        writer.write_all(b"\n").await?;
    }
    Ok(())
}

#[tokio::main]
async fn main() -> io::Result<()> {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:7001").await?;
    loop {
        let (stream, _) = listener.accept().await?;
        tokio::spawn(async move {
            if let Err(error) = echo_lines(stream).await {
                eprintln!("client: {error}");
            }
        });
    }
}

Size limit detail

The example checks length after a full line arrives, so it does not cap buffering while a client sends a line without a newline. For strict memory limits, implement incremental bounded framing or use a framed codec with a configured maximum.

19 / 26

Blocking code, CPU work, and yielding#

Never call long blocking functions directly on a Tokio worker thread.

std::thread::sleep, a synchronous database driver, or heavy CPU work inside async code can stall other tasks. spawn_blocking moves bounded blocking work to a separate pool. For a large volume of CPU jobs, bound concurrency yourself or use a dedicated compute pool.

RustMove a synchronous calculation off workers

#[tokio::main]
async fn main() -> Result<(), tokio::task::JoinError> {
    let answer = tokio::task::spawn_blocking(|| {
        std::thread::sleep(std::time::Duration::from_millis(100));
        (1..=10_000u64).sum::<u64>()
    }).await?;
    println!("{answer}");
    Ok(())
}

RustCooperate inside a long async loop

async fn process_items(items: &[u64]) -> u64 {
    let mut total = 0;
    for (index, item) in items.iter().enumerate() {
        total += *item;
        if index % 1000 == 0 {
            tokio::task::yield_now().await;
        }
    }
    total
}

Limits of yield_now

Yielding gives other tasks a chance to run but does not make CPU work parallel or fix slow synchronous I/O. Use a bounded blocking pool or dedicated workers for substantial compute.

20 / 26

UDP datagrams#

UDP preserves message boundaries but offers no delivery or ordering guarantee.

RustReceive one datagram and reply

use tokio::{io, net::UdpSocket};

#[tokio::main]
async fn main() -> io::Result<()> {
    let socket = UdpSocket::bind("127.0.0.1:7002").await?;
    let mut buffer = [0u8; 2048];
    loop {
        let (size, peer) = socket.recv_from(&mut buffer).await?;
        socket.send_to(&buffer[..size], peer).await?;
    }
}

Datagram size

The buffer caps the bytes received from each datagram. A larger datagram can be truncated, so choose a suitable maximum and protocol validation.

21 / 26

Signals and shutdown triggers#

Listen for Ctrl+C, then tell tasks to stop and wait for them to finish.

RustWait for Ctrl+C

use tokio::{io, signal};

#[tokio::main]
async fn main() -> io::Result<()> {
    println!("Press Ctrl+C");
    signal::ctrl_c().await?;
    println!("shutting down");
    Ok(())
}

Portability

ctrl_c() works across supported platforms. Other OS signals need platform-specific handling, such as tokio::signal::unix on Unix.

22 / 26

Run child processes#

Tokio can start a subprocess without blocking the task while it waits.

RustCollect output and inspect the exit status

use tokio::{io, process::Command};

#[tokio::main]
async fn main() -> io::Result<()> {
    let output = Command::new("rustc")
        .arg("--version")
        .output()
        .await?;
    if !output.status.success() {
        eprintln!("command failed: {}", output.status);
    } else {
        println!("{}", String::from_utf8_lossy(&output.stdout));
    }
    Ok(())
}

Child lifetime

Dropping a Child does not necessarily terminate its process. When using a Child handle, arrange to wait or kill it on cancellation; kill_on_drop(true) is available where that policy fits.

23 / 26

Pin a future across select! loops#

A future recreated each loop restarts its clock or state. Pin it once when you need one overall deadline.

RustOne deadline for the whole receive loop

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

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<u32>(8);
    tx.send(1).await.unwrap();
    drop(tx);

    let deadline = tokio::time::sleep(Duration::from_secs(2));
    tokio::pin!(deadline); // stable address for polling again
    loop {
        tokio::select! {
            _ = &mut deadline => { println!("time is up"); break; }
            maybe_value = rx.recv() => match maybe_value {
                Some(value) => println!("got {value}"),
                None => break, // sender closed and queue is empty
            },
        }
    }
}

Pin in plain terms

Some futures keep references to their own state after being polled and cannot safely move in memory. tokio::pin! pins a local future, so &mut future can be reused by successive select! calls.

24 / 26

Local tasks and non-Send values#

Most spawned tasks need Send because the multi-thread runtime may move them between workers.

An Rc<RefCell<T>> is confined to one thread. LocalSet and spawn_local allow it inside a local task, run on a current-thread runtime. Prefer Arc and message passing for ordinary multi-threaded work.

RustLocalSet with Rc

use std::{cell::RefCell, rc::Rc};
use tokio::task::LocalSet;

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let local = LocalSet::new();
    local.run_until(async {
        let value = Rc::new(RefCell::new(0));
        let shared = Rc::clone(&value);
        let task = tokio::task::spawn_local(async move {
            *shared.borrow_mut() += 1;
        });
        task.await.unwrap();
        println!("{}", *value.borrow());
    }).await;
}

Hold RefCell borrows briefly

Do not keep RefCell borrows across .await. Dynamic borrow conflicts panic at runtime even when the task stays on one thread.