Cancelling Tasks in Tokio With Abort and Cancellation Tokens

Stopping a task in Tokio has three distinct meanings, and mixing them up is how services end up with half-written files and background work nobody can stop. You can drop a future (cancelling it at its current await), abort a task by handle (the runtime drops it for you), or ask it to stop and let it clean up (a cancellation token). The first two are abrupt: nothing after the current await runs except destructors. The third is cooperative, and it is the one to use whenever stopping mid-way would leave a mess. This post shows each, what runs and what does not, and a worker loop that stops cleanly in the middle of a batch. Everything compiled on Rust 1.99, edition 2024, with tokio 1 and tokio-util 0.7.

Way 1: drop the future

A future that is not being polled is not running. Dropping it, which select!, timeout and a plain drop(fut) all do, cancels it at whatever await it had reached. Locals are dropped in order; the rest of the function never executes:

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

struct Cleanup(&'static str);

impl Drop for Cleanup {
    fn drop(&mut self) {
        println!("drop: {}", self.0);
    }
}

async fn work() {
    let _guard = Cleanup("work guard");
    println!("work: started");
    sleep(Duration::from_secs(5)).await;
    println!("work: finished (never printed if cancelled)");
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let r = timeout(Duration::from_secs(1), work()).await;
    println!("timed out: {}", r.is_err());
}

work: started
drop: work guard
timed out: true

The guard's destructor ran, which is what makes RAII a good fit for cancellation: locks are released, files are closed, connections go back to a pool. What did not run is any async cleanup, because Drop cannot await. That limitation is the reason for the third way, and the data-loss hazards of dropping futures mid-operation are covered in cancellation safety.

Way 2: abort a spawned task

JoinHandle::abort tells the runtime to drop the task's future at its next opportunity, which is the same as way 1 for a task you cannot reach directly. The result is observable through the handle:

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

struct Cleanup(&'static str);

impl Drop for Cleanup {
    fn drop(&mut self) {
        println!("drop: {}", self.0);
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let handle = tokio::spawn(async {
        let _g = Cleanup("task guard");
        sleep(Duration::from_secs(60)).await;
        "unreachable"
    });

    sleep(Duration::from_millis(10)).await;
    handle.abort();

    match handle.await {
        Ok(v) => println!("completed: {v}"),
        Err(e) if e.is_cancelled() => println!("task was cancelled"),
        Err(e) => println!("task failed: {e}"),
    }
}

drop: task guard
task was cancelled

Two properties to know. An abort takes effect at an await point; a task in the middle of a long CPU loop, or one running a spawn_blocking closure, finishes that stretch first (and a blocking closure is never interrupted). And aborting a task that holds a std::sync::Mutex guard is safe because the guard drops, but a task holding a tokio::sync::Mutex across an await that you abort releases it too; the danger is not a stuck lock but the half-done work in between.

Way 3: ask the task to stop

When the task has work to finish, flush, or roll back, abort is the wrong tool, and the task should be asked to stop and given the chance to do it in its own time. The standard device is a CancellationToken from tokio-util: a cloneable signal that any number of tasks can wait on.

use tokio::time::{sleep, Duration};
use tokio_util::sync::CancellationToken;

async fn worker(id: u32, token: CancellationToken) -> u32 {
    let mut done = 0;
    loop {
        tokio::select! {
            _ = token.cancelled() => {
                println!("worker {id}: stop requested after {done} items, flushing");
                sleep(Duration::from_millis(20)).await;
                println!("worker {id}: flushed");
                return done;
            }
            _ = sleep(Duration::from_millis(100)) => {
                done += 1;
            }
        }
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let token = CancellationToken::new();
    let a = tokio::spawn(worker(1, token.clone()));
    let b = tokio::spawn(worker(2, token.child_token()));

    sleep(Duration::from_millis(350)).await;
    token.cancel();

    println!("totals: {} and {}", a.await.unwrap(), b.await.unwrap());
}

worker 2: stop requested after 3 items, flushing
worker 1: stop requested after 3 items, flushing
worker 2: flushed
worker 1: flushed
totals: 3 and 3

The worker saw the token fire, did its async flush (a sleep standing in for a network write that Drop could not do), and returned a value. The owner then collected the results through the handles, rather than losing them. child_token() creates a token that is cancelled when its parent is but can also be cancelled on its own, which gives you a tree: cancel one subsystem without touching its siblings, or cancel everything from the root.

The pattern that makes it correct: select only on safe branches

The structure above is deliberate. The cancelled() branch is cancel safe, and so is sleep. The unit of work happens in the branch body, which select never cancels once a branch has been chosen. If the work itself is a long await, put it in the select as well and make sure it is cancel safe, or the cancellation will interrupt it halfway. A worker that processes a message in several steps should finish the message before checking the token again:

use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};
use tokio_util::sync::CancellationToken;

async fn process(job: u32, log: &mut Vec<String>) {
    log.push(format!("job {job}: step 1"));
    sleep(Duration::from_millis(30)).await;
    log.push(format!("job {job}: step 2"));
}

async fn run(mut rx: mpsc::Receiver<u32>, token: CancellationToken) -> Vec<String> {
    let mut log = Vec::new();
    loop {
        tokio::select! {
            biased;
            _ = token.cancelled() => {
                log.push("stop requested".into());
                break;
            }
            maybe = rx.recv() => match maybe {
                Some(job) => process(job, &mut log).await,
                None => break,
            },
        }
    }
    log
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let (tx, rx) = mpsc::channel(4);
    let token = CancellationToken::new();
    let task = tokio::spawn(run(rx, token.clone()));

    tx.send(1).await.unwrap();
    tx.send(2).await.unwrap();
    sleep(Duration::from_millis(40)).await;
    token.cancel();

    for line in task.await.unwrap() {
        println!("{line}");
    }
}

job 1: step 1
job 1: step 2
job 2: step 1
job 2: step 2
stop requested

Job 2 was in the middle of its two steps when cancellation arrived, and it completed both before the loop looked at the token. That is cooperative cancellation: the stop request is honoured at the next safe point, not instantly. biased; makes select! check branches in order, so the token is examined first on every iteration; without it the branch is chosen at random when several are ready, which is fine but makes tests less repeatable.

Waiting for the group to actually stop

Cancelling a token only asks. To know that every worker has finished its cleanup, you have to wait for the tasks, and a loop over handles works when you hold all of them. When tasks are spawned from many places, TaskTracker from tokio-util counts them for you: spawn through the tracker, close() it when no more will be added, then wait() until the count reaches zero.

use tokio::time::{sleep, Duration};
use tokio_util::sync::CancellationToken;
use tokio_util::task::TaskTracker;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let tracker = TaskTracker::new();
    let token = CancellationToken::new();

    for id in 1..=3u64 {
        let token = token.clone();
        tracker.spawn(async move {
            token.cancelled().await;
            sleep(Duration::from_millis(id * 50)).await;
            println!("task {id}: cleanup done");
        });
    }

    sleep(Duration::from_millis(10)).await;
    println!("requesting stop");
    token.cancel();
    tracker.close();
    tracker.wait().await;
    println!("all {} tasks finished", 3);
}

requesting stop
task 1: cleanup done
task 2: cleanup done
task 3: cleanup done
all 3 tasks finished

The tasks finish in order of their cleanup time, and wait() returns only after the slowest. Without the tracker, the program would have reached the end of main and exited while task 3 was still flushing, which is exactly the half-written-file bug. Always give shutdown a deadline (wrap the wait in timeout) so one stuck cleanup cannot hold the process hostage; the policy question of what to do when the deadline passes is in graceful shutdown.

Choosing

Child tokens deserve one more sentence: a child cancelled on its own does not affect its parent, which lets a connection handler stop itself (a client hung up) without stopping the server. Wire the tree from the top: the process owns the root token, each subsystem takes a child, each connection takes a child of that. A single cancel() at the root then reaches everything, and no subsystem needs to know how many siblings it has.

What cancellation cannot do

It cannot interrupt a thread blocked in a synchronous call, so a task stuck in std::thread::sleep or a blocking read ignores both abort and the token until the call returns. It cannot undo effects that already left the process: a request sent to another service stays sent. And it makes the question of idempotency unavoidable, because the code that cancels cannot know whether the other side did the work. Design operations that are safe to repeat, as in idempotency keys, and treat cancellation as a normal event your code has to survive, not an exceptional one.