use std::time::Duration;
use tokio_util::sync::CancellationToken;
pub struct Worker {
id: usize,
token: CancellationToken,
}
impl Worker {
pub fn new(id: usize, token: CancellationToken) -> Self {
Worker { id, token }
}
pub async fn run(self) {
loop {
tokio::select! {
biased;
_ = self.token.cancelled() => {
tracing::info!(worker = self.id, "cancellation observed, stopping");
break;
}
result = self.process() => {
if let Err(err) = result {
tracing::warn!(worker = self.id, %err, "work item failed");
}
}
}
}
self.drain().await;
}
async fn process(&self) -> Result<(), std::io::Error> {
tokio::time::sleep(Duration::from_millis(250)).await;
Ok(())
}
async fn drain(&self) {
tracing::debug!(worker = self.id, "flushing in-flight state before exit");
tokio::time::sleep(Duration::from_millis(20)).await;
}
}
use std::time::Duration;
use futures::future::join_all;
use tokio::signal;
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
use crate::worker::Worker;
pub struct Supervisor {
root: CancellationToken,
handles: Vec<JoinHandle<()>>,
}
impl Supervisor {
pub fn new() -> Self {
Supervisor { root: CancellationToken::new(), handles: Vec::new() }
}
pub fn spawn_workers(&mut self, count: usize) {
for id in 0..count {
let token = self.root.child_token();
let worker = Worker::new(id, token);
self.handles.push(tokio::spawn(worker.run()));
}
}
pub async fn run_until_shutdown(mut self) {
let token = self.root.clone();
tokio::spawn(async move {
if signal::ctrl_c().await.is_ok() {
tracing::info!("shutdown signal received");
token.cancel();
}
});
self.root.cancelled().await;
self.await_workers().await;
}
async fn await_workers(self) {
let join = join_all(self.handles);
match tokio::time::timeout(Duration::from_secs(10), join).await {
Ok(_) => tracing::info!("all workers stopped cleanly"),
Err(_) => tracing::error!("timed out waiting for workers, forcing exit"),
}
}
}
mod supervisor;
mod worker;
use supervisor::Supervisor;
#[tokio::main]
async fn main() {
tracing_subscriber::fmt()
.with_max_level(tracing::Level::DEBUG)
.init();
let mut supervisor = Supervisor::new();
supervisor.spawn_workers(4);
tracing::info!("service started; press Ctrl-C to shut down");
supervisor.run_until_shutdown().await;
tracing::info!("service exited");
}
This snippet demonstrates cooperative task cancellation in an async Rust service built on Tokio, using tokio_util::sync::CancellationToken to fan out a single shutdown signal to many long-running tasks. The core problem is that dropping a JoinHandle does not cancel the future it drives cleanly — an in-flight database write or a partially sent frame may be aborted at an arbitrary await point. A CancellationToken instead makes cancellation an explicit, observable event that each task selects on, so tasks decide where it is safe to stop.
In worker.rs, the Worker holds a token and its run loop uses tokio::select! with the biased branch calling token.cancelled().await. Biasing means the cancellation branch is polled first on every loop iteration, guaranteeing the loop exits promptly rather than starting yet another unit of work once shutdown has begun. The process step is itself awaited inside the same select!, so a shutdown that arrives mid-task interrupts the future at its next await point without leaving the loop in an inconsistent state.
The key structural idea appears in supervisor.rs. A single parent token is created, and each worker receives parent.child_token(). Child tokens are cancelled when the parent is cancelled, but can also be cancelled independently, which is how the supervisor kills one misbehaving worker without touching the others. spawn_workers collects every JoinHandle so the shutdown path can join_all them and observe clean completion rather than firing cancellation and walking away.
The run_until_shutdown method wires an OS signal to the token: when Ctrl-C arrives it calls token.cancel(), which wakes every cancelled() future across all workers simultaneously. It then awaits the join set with a timeout so a stuck task cannot hang the process forever — a common pitfall where cooperative cancellation degrades into an unbounded wait.
The main.rs tab shows the assembly: build a Supervisor, register work, and hand control to the runtime. The trade-off of this pattern is that cancellation is cooperative — a task that never reaches an await point or ignores its token will not stop — but in exchange it yields predictable, leak-free shutdown and fine-grained control that abrupt task aborts cannot provide.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
Share this code
Here's the card — post it anywhere.
