Crate stop_token

Cooperative cancellation for async Rust.

Status

Experimental. The library works as is, breaking changes will bump major version, but there are no guarantees of long-term support.

Motivation

Rust futures come with a build-in cancellation mechanism: dropping a future prevents any further progress of the future. This is a hard cancellation mechanism, meaning that the future can potentially be cancelled at any .await expression.

Sometimes, you need are more fine-grained cancellation mechanism. Imagine a chat server that relays messages to peers. Sending a single message potentially needs several writes on the socket object. That means that, if we use hard-cancellation for client connections, a connection can be abruptly terminated mid-message (even mid-emoji, if we are especially unlucky). What we need here is cooperative cancellation: client connection should be gracefully shutdown between the messages.

More generally, if you have an event processing loop like

while let Some(event) = work.next().await {
    process_event(event).await
}

you usually want to maintain an invariant that each event is either fully processed or not processed at all. If you need to terminate this loop early, you want to do this between iteration.

Usage

You can use this crate to create a deadline received through a StopToken. You can think of a StopSource + StopToken as a single-producer, multi-consumer channel that receives a single message to “stop” when the producer is dropped:

use async_std::prelude::*;
use async_std::{stream, task};

use stop_token::prelude::*;
use stop_token::StopSource;

use std::time::Duration;

#[async_std::main]
async fn main() {
    // Create a stop source and generate a token.
    let src = StopSource::new();
    let stop = src.token();

    // When stop source is dropped, the loop will stop.
    // Move the source to a task, and drop it after 100 millis.
    task::spawn(async move {
        task::sleep(Duration::from_millis(100)).await;
        drop(src);
    });

    // Create a stream that generates numbers until
    // it receives a signal it needs to stop.
    let mut work = stream::repeat(12u8).until(stop);

    // Loop over each item in the stream.
    while let Some(Ok(ev)) = work.next().await {
        println!("{}", ev);
    }
}

Or Duration or Instant to create a time-based deadline:

use async_std::prelude::*;
use async_std::stream;

use stop_token::prelude::*;

use std::time::Duration;

#[async_std::main]
async fn main() {
    // Create a stream that generates numbers for 100 millis.
    let stop = Duration::from_millis(100);
    let mut work = stream::repeat(12u8).until(stop);

    // Loop over each item in the stream.
    while let Some(Ok(ev)) = work.next().await {
        println!("{}", ev);
    }
}

Features

The time submodule is empty when no features are enabled. To implement IntoDeadline for Instant and Duration you can enable one of the following features:

• async-io: for use with the async-std or smol runtimes.
• tokio: for use with the tokio runtime.

Lineage

The cancellation system is a subset of C# CancellationToken / CancellationTokenSource. The StopToken / StopTokenSource terminology is borrowed from C++ paper P0660.

Modules

future

Extension methods and types for the Future trait.

prelude

A prelude for stop-token.

stream

Extension methods and types for the Stream trait.

time

Create deadlines from Duration and Instant types.

Structs

StopSource

StopSource produces StopToken and cancels all of its tokens on drop.

StopToken

StopToken is a future which completes when the associated StopSource is dropped.

TimedOutError

An error returned when a future times out.

Traits

IntoDeadline

Conversion into a deadline.