Struct cancel_safe_futures::sync::ActionPermit

pub struct ActionPermit<'a, T: ?Sized> { /* private fields */ }

Available on crate feature std only.

A token that grants the ability to run one closure against the data guarded by a RobustMutex.

This is produced by the lock family of operations on RobustMutex and is intended to provide robust cancel safety.

For more information, see the documentation for RobustMutex.

Why is this its own type?

A question some users might have is: why not combine lock and perform? Why have this type that sits in the middle?

The answer is that this structure is necessary to provide cancel safety. Consider what happens with a hypothetical lock_and_perform function. Let’s say we use it in a select! statement thus:

use std::sync::LockResult;
use std::time::Duration;
use tokio::time::sleep;

struct MyMutex<T> { /* ... */ }

impl<T> MyMutex<T> {
    fn new(data: T) -> Self {
        /* ... */
        todo!();
    }
    async fn lock_and_perform<U>(self, action: impl FnOnce(&mut T) -> U) -> LockResult<U> {
        /* ... */
    }
}

// Represents some kind of type that is unique and can't be cloned.
struct NonCloneableType(u32);

#[tokio::main]
async fn main() {
    let mutex = MyMutex::new(1);
    let data = NonCloneableType(2);
    let sleep = sleep(Duration::from_secs(1));

    let fut = mutex.lock_and_perform(|n| {
        *n = data.0;
    });

    tokio::select! {
        _ = fut => {
            /* ... */
        }
        _ = sleep => {
            /* ... */
        }
    }
}

Then, if sleep fires before fut, the non-cloneable type is dropped without being used. This leads to cancel unsafety.

This is very similar to the cancel unsafety that futures::SinkExt::send has, and that this crate’s SinkExt::reserve solves.

Implementations

impl<'a, T: ?Sized> ActionPermit<'a, T>

pub fn perform<R, F>(self, action: F) -> R

where F: FnOnce(&mut T) -> R,

Runs a closure with access to the guarded data, consuming the permit in the process and unlocking the mutex once the closure completes.

This is a synchronous closure, which means that it cannot have await points within it. This guarantees cancel safety for this mutex.

Notes

action is not run inside a synchronous context. This means that operations like tokio::sync::mpsc::Sender::blocking_send will panic inside action.

If action panics, the mutex is marked poisoned.

Examples

use cancel_safe_futures::sync::RobustMutex;

#[tokio::main]
async fn main() {
    let mutex = RobustMutex::new(1);

    let permit = mutex.lock().await.unwrap();
    permit.perform(|n| *n = 2);
}

pub async fn perform_async_boxed<R, F>(self, action: F) -> R

where F: for<'lock> FnOnce(&'lock mut T) -> BoxFuture<'lock, R>,

Runs an asynchronous block in the context of the guarded data, consuming the permit in the process and unlocking the mutex once the block completes.

In general, holding asynchronous locks across await points can lead to surprising performance issues. It is strongly recommended that perform is used, or that the code is rewritten to use message passing.

Notes

The mutex is marked poisoned if any of the following occur:

• The future returned by action panics.
• The future returned by this async function is cancelled before being driven to completion.

Due to limitations in stable Rust, this accepts a dynamic BoxFuture rather than a generic future. Once async closures are stabilized, this will switch to them.

Examples

use cancel_safe_futures::sync::RobustMutex;
use futures::FutureExt;  // for FutureExt::boxed()
use std::time::Duration;

#[tokio::main]
async fn main() {
    let mutex = RobustMutex::new(1);

    let permit = mutex.lock().await.unwrap();
    permit.perform_async_boxed(|n| {
        async move {
            tokio::time::sleep(
                std::time::Duration::from_millis(100),
            ).await;
            *n = 2;
        }
        .boxed()
    }).await;

    // Check that the new value of the mutex is 2.
    let permit = mutex.lock().await.unwrap();
    permit.perform(|n| assert_eq!(*n, 2));
}

pub async fn perform_async_boxed_local<R, F>(self, action: F) -> R

where F: for<'lock> FnOnce(&'lock mut T) -> LocalBoxFuture<'lock, R>,

Runs a non-Send asynchronous block in the context of the guarded data, consuming the permit in the process and unlocking the mutex once the block completes.

This is a variant of perform_async_boxed that allows the future to be non-Send.

Trait Implementations

impl<'a, T: Debug + ?Sized> Debug for ActionPermit<'a, T>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.