Cancellation and Timeouts

The Asynchronous Input/Output lesson left open the question of a task that is written in the source but never completes: a task can pause forever, so what happens if it is never touched again? This course’s closing lesson takes up that question. In Rust’s asynchronous model, there is no separate call to stop a task — a future only advances for as long as it is polled, so never polling it again is itself what cancels that task.

Go’s Concurrency course measured cancellation as one of four ways a goroutine can stop — a stopping mechanism. This lesson measures the same concept differently: a canceled task has no execution flow of its own to stop — cancellation here is a state of not advancing, the fate of a task that sits written in the source and is never polled again. Two separate paths are measured: a task can be canceled by being manually dropped, or it can time out by being dropped automatically once a poll budget runs out. Both fall into the same class.

A Dropped Task Is Never Polled Again

Being dropped does not by itself mean being canceled. A completed task, too, once the value it produced has been read, does not go on sitting in memory — it too gets dropped, but that is a cleanup, not a cancellation. The distinction is in whether the task has finished at the moment it is dropped: a task dropped before finishing will never be polled again, a finished task has nothing left to poll anyway. The following core excerpt holds this distinction with two fields: started marks whether the task was polled at least once, completed marks whether it produced its final value. A task dropped while only the first is true and the second is false is counted as canceled; this is measured by sending a record from the task’s own Drop implementation.

// advance.rs — a slice of the core: cancellation established by dropping, and a timeout
// modeled with a poll budget.

use std::future::Future;
use std::pin::Pin;
use std::sync::mpsc::Sender;
use std::sync::{Arc, Condvar, Mutex};
use std::task::{Context, Poll, Wake, Waker};

pub struct Notifier {
    pub ready: Mutex<bool>,
    pub condvar: Condvar,
}

impl Wake for Notifier {
    fn wake(self: Arc<Self>) {
        *self.ready.lock().unwrap() = true;
        self.condvar.notify_one();
    }
}

pub fn run<F: Future>(task: F) -> (F::Output, u32) {
    let mut task = Box::pin(task);
    let n = Arc::new(Notifier {
        ready: Mutex::new(false),
        condvar: Condvar::new(),
    });
    let waker: Waker = n.clone().into();
    let mut context = Context::from_waker(&waker);
    let mut poll_count = 0;
    loop {
        poll_count += 1;
        match task.as_mut().poll(&mut context) {
            Poll::Ready(x) => return (x, poll_count),
            Poll::Pending => {
                let mut ready = n.ready.lock().unwrap();
                while !*ready {
                    ready = n.condvar.wait(ready).unwrap();
                }
                *ready = false;
            }
        }
    }
}

// a task that can be manually dropped after being polled once: started work is
// measurable, completion is only recorded if a second poll actually happens.
pub struct CancelableJob {
    pub name: &'static str,
    pub started: bool,
    pub completed: bool,
    pub channel: Sender<String>,
}

impl Future for CancelableJob {
    type Output = &'static str;
    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<&'static str> {
        if !self.started {
            self.started = true;
            self.channel.send(format!("{}-started", self.name)).unwrap();
            cx.waker().wake_by_ref();
            Poll::Pending
        } else {
            self.completed = true;
            self.channel.send(format!("{}-completed", self.name)).unwrap();
            Poll::Ready(self.name)
        }
    }
}

impl Drop for CancelableJob {
    fn drop(&mut self) {
        if self.started && !self.completed {
            self.channel.send(format!("{}-dropped", self.name)).unwrap();
        }
    }
}

// a long job that advances with a counter decreasing by one on every poll; if it is
// dropped before completing, the remaining amount stays above zero.
pub struct LongJob {
    pub remaining: u32,
    pub name: &'static str,
    pub channel: Sender<String>,
}

impl Future for LongJob {
    type Output = &'static str;
    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<&'static str> {
        if self.remaining == 0 {
            self.channel.send(format!("{}-completed", self.name)).unwrap();
            Poll::Ready(self.name)
        } else {
            self.remaining -= 1;
            cx.waker().wake_by_ref();
            Poll::Pending
        }
    }
}

impl Drop for LongJob {
    fn drop(&mut self) {
        if self.remaining > 0 {
            self.channel.send(format!("{}-dropped-midway", self.name)).unwrap();
        }
    }
}

// a task wrapping an inner task with a limited number of polling rights: once the
// budget runs out the inner task is dropped immediately and never polled again.
pub struct TimedTask<F: Future> {
    pub inner: Option<Pin<Box<F>>>,
    pub budget: u32,
    pub name: &'static str,
    pub channel: Sender<String>,
}

impl<F: Future> Future for TimedTask<F> {
    type Output = Result<F::Output, ()>;
    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<F::Output, ()>> {
        if self.budget == 0 {
            self.inner = None;
            self.channel.send(format!("{}-timed-out", self.name)).unwrap();
            return Poll::Ready(Err(()));
        }
        self.budget -= 1;
        match self.inner.as_mut().unwrap().as_mut().poll(cx) {
            Poll::Ready(x) => Poll::Ready(Ok(x)),
            Poll::Pending => Poll::Pending,
        }
    }
}

// cancellation.rs — cancellation established by dropping and a timeout modeled with a
// poll budget, in a single run.
mod advance;
use advance::{run, CancelableJob, LongJob, TimedTask, Notifier};
use std::future::Future;
use std::sync::mpsc::channel;
use std::sync::{Arc, Condvar, Mutex};
use std::task::{Context, Waker};

fn main() {
    let (sender, receiver) = channel::<String>();

    // --- completed task: run polls it until it finishes ---
    let (result, poll_count) = run(CancelableJob {
        name: "task-completed",
        started: false,
        completed: false,
        channel: sender.clone(),
    });

    // --- canceled task: polled once by hand, then dropped ---
    let mut canceled = Box::pin(CancelableJob {
        name: "task-canceled",
        started: false,
        completed: false,
        channel: sender.clone(),
    });
    let n = Arc::new(Notifier { ready: Mutex::new(false), condvar: Condvar::new() });
    let waker: Waker = n.clone().into();
    let mut context = Context::from_waker(&waker);
    let _ = canceled.as_mut().poll(&mut context);
    drop(canceled);

    // --- timeout with a poll budget: one finishes within budget, the other exceeds it ---
    let (result_a, budget_poll_count_a) = run(TimedTask {
        inner: Some(Box::pin(LongJob { remaining: 1, name: "job-a", channel: sender.clone() })),
        budget: 3,
        name: "job-a",
        channel: sender.clone(),
    });
    let (result_b, budget_poll_count_b) = run(TimedTask {
        inner: Some(Box::pin(LongJob { remaining: 5, name: "job-b", channel: sender.clone() })),
        budget: 2,
        name: "job-b",
        channel: sender.clone(),
    });

    drop(sender);
    let mut records: Vec<String> = receiver.try_iter().collect();
    records.sort();

    println!("task-completed finished in {} polls, result: {}", poll_count, result);
    println!("task-canceled manually dropped after 1 poll");
    println!();
    println!("job-a (budget=3, remaining=1) -> {} polls, result: {:?}", budget_poll_count_a, result_a);
    println!("job-b (budget=2, remaining=5) -> {} polls, result: {:?}", budget_poll_count_b, result_b);
    println!();
    println!("records ({}):", records.len());
    for record in &records {
        println!("  {}", record);
    }
}

task-completed finished in 2 polls, result: task-completed
task-canceled manually dropped after 1 poll

job-a (budget=3, remaining=1) -> 2 polls, result: Ok("job-a")
job-b (budget=2, remaining=5) -> 3 polls, result: Err(())

records (7):
  job-a-completed
  job-b-dropped-midway
  job-b-timed-out
  task-canceled-dropped
  task-canceled-started
  task-completed-completed
  task-completed-started

AP7. task-completed finishes in two polls — one pause, one completion — carrying forward the familiar $n+1$ rule from the async/await lesson. task-canceled is only polled once by hand, sends its -started record, and is then dropped directly; this drop triggers the Drop implementation and sends the -dropped record. These two records make the task’s started work measurable: one poll happened, one cleanup ran. But task-canceled-completed is nowhere among the seven records — because the second poll never happened. The canceled task falls into the not-at-all class: written in the source, touched once, but dropped before completing.

Timeout When the Poll Budget Runs Out

In this course, a timeout is built not with a duration but with a number: the excerpt above’s TimedTask allows the task it wraps to be polled at most budget times. If the wrapped task finishes on its own before the budget runs out, the result comes back as an Ok value; once the budget runs out, TimedTask immediately drops the task inside it — assigning None to self.inner — and never touches it again. This is the automated form of the manual drop mechanism from the previous section: the decision is made not by the task itself, but by the budget wrapping it.

AP8. job-a was constructed with a budget of three polls and used only two of them, finishing with Ok("job-a"). job-b was constructed with a budget of only two polls but needed five; on the third call the budget dropped to zero, so the wrapper drops the inner task and returns Err(()). The inner task’s own Drop implementation sends the -dropped-midway record, because at the moment it was dropped it still had three units of work left — of five units of work, two had completed, three were started but not finished. The job-b-completed record, too, is nowhere among the seven records. job-a falls into the when-polled class; job-b falls into the not-at-all class once its budget runs out — the same fate as a manually dropped task, triggered by a different cause.

The Limiting Measurement

Dropping does not reveal a task’s internal state. The Drop implementation only records what was hand-coded for it to write — CancelableJob tracks the started and completed flags, LongJob tracks the remaining counter value. This means the language does not expose any general “how much work got done” information on its own: unless the author of a task type makes this information measurable, what state a dropped value was in is unknowable outside the source code.

The poll budget is not a duration either. The budget field is an upper bound on how many times something gets polled, but it has no relationship at all to how long a task actually takes in the real world — the same budget runs out after the same number of polls whether every poll of a task is heavy or every poll is instant. A real timeout mechanism pairs this number with a clock; because this course uses no clock, a timeout here is only a measure of how many times it was tried, not of how long it waited.

Closing Table

Summary

• Dropping a task is the cancellation call in Rust’s asynchronous model; a dropped task is never polled again and falls into the not-at-all class.
• A drop is a cancellation only if it happens before the task completes; dropping a completed task is ordinary cleanup — the two states were kept apart by separate records in this lesson.
• A timeout was modeled not with a duration but with a poll budget: a task whose budget ran out was dropped immediately and never polled again — it, too, fell into the not-at-all class.
• A canceled task’s started work stayed measurable — records existed showing that it had started and had been dropped — but a record showing that it had completed never arrived.
• Go’s Concurrency course measured cancellation as a stopping mechanism; this lesson measured the same concept as a state of not advancing.
• The closing table brought the course’s ten lessons and its four classes (on its own, when polled, when something else finishes, not at all) together in a single view.

Course Wrap-Up

This course carried a single question through ten lessons: what does a unit need in order to advance. The answer was always one of the same four names, and the closing table showed how these four names were distributed across ten different units — threads advanced on their own or by waiting on each other, locks and channels carried that waiting, futures advanced only when polled, and this closing lesson showed that a task can also never advance at all — sitting written in the source and never touched again. This was the repository’s fifth, and for now last, concurrency course; five courses together toured the same territory with five different questions.

The next course, The Rust Ecosystem and Advanced Topics, looks at the rest of the language beyond concurrency: macros, the sections of code voluntarily stepped outside the compiler’s safety guarantees, call interfaces built with other languages, and a Rust project’s life cycle from dependency management to publication. The progress question measured in this lesson — what a unit needs in order to advance — is not asked again there; the next course will build its own measurement axis from its own subject.