Feature request: Unless combinator

[pdolezal]

I miss a compact way to interrupt a future completion with completing another future. There are means for parallel polling, but they do not fit the intended use cases very well – see the discussion about the alternatives below. I'm still a newbie, so perhaps I missed something like a pattern to use for my case, although I did my best to find an existing solution.

A case for Unless

Let's have a long-running task – I think that the term actor would fit here – that receives and handles messages arriving through a channel or a connection. The actor should be stopped, for instance when the application is going to terminate, in a graceful way, so that it can stop receiving new messages and complete all messages in flight before it terminates.

The actor may check the shutdown request at convenient exit points, e.g., when awaiting a new message, and act according to the exit point context. Conceptually, it boils down to parallel polling of two futures: the regular action and the exceptional condition. The first future that completes should be chosen and when both are completed, the regular action should be preferred.

The bias for choosing the regular action favors the graceful termination scenario when everything started is finished completely as well. However, I admit, when the regular action completes always immediately on the first poll, it might lead to ignoring the exceptional condition forever. At least, the behavior is predictable and it allows for a separate check on another dedicated exit point when the regular action is known to be such a quick shooter. Thinking over these details is definitely needed.

Example

At first, let me show a crude and simplistic code snippet that hopefully gives the feeling of the use. In this example let's assume that Signal is a simplified facility for passing a value (I tried a prototype based on tokio::sync::watch). Signal::raised returns a future that completes when the signal is raised and returns the signal value.

async fn processing_task(
    uri: String,
    shutdown: Signal<ShutdownKind>,
) -> Result<(), ProcessingError> {
    let connection = connect_to(uri).await?;

    loop {
        // connection::recv returns a Result specific for the Connection implementation
        let message = match connection.recv().unless(signal.raised()).await {
            Ok(incoming) => incoming.map_err(|_| ProcessingError::ConnectionFailure)?,

            // Note that it is possible to pass even some instructions for the shutdown.
            // It might be a timeout instead… or whatever else.
            Err(ShutdownKind::Graceful) => break,
            Err(ShutdownKind::Immediate) => {
                connection.close().await;
                return Err(ProcessingError::Aborted);
            }
        };

        // Off-load it to a pool that handles messages concurrently in background
        schedule_processing(message).await;
    }

    finish_pending_messages().await;
    connection.close().await;
    Ok(())
}

Current alternatives

Here are the alternatives that I considered and the reasons why I think that they do not fit the use case well enough.

futures::select_biased!

The disadvantage of this macro is that it is quite mouthful for such simple use cases. It works with fused futures and needs pinning the futures first. I'd argue that the example feels more natural than using select_biased!.

tokio::select!

Although tokio::select! does not require fused futures and pinning the futures first, which makes it better usable (I second to #1989), there is no alternative for futures::select_biased!. The bias is an important feature here.

futures::abortable

The use case for abortable seems similar, but it does not support the graceful exit strategy for the aborted future – it is somewhat violent –, while the proposed Unless combinator naturally supports cooperative cancellation protocols. Using the AbortHandle makes it more specialized as well, while Unless seems to cover more use cases than the example suggests (see below).

Something else?

Let me know… I have just seen #543, which resembles this proposal, but it is focused on streams and I'm not quite sure if I understand it correctly.

Other use cases

Using some signal to indicate a cancellation/shutdown/abort request, as the example showed, is not the only use case. What about timeouts?

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

let outcome = do_something().unless(delay_for(Duration::from_secs(5))).await?;

The API sketch

It seems natural to extend FutureExt with the combinator with something like:

fn unless<Fut>(&mut self, condition: Fut) -> Unless<&mut Self, Fut>
where
    Fut: Future,
    Self: Sized,
{
    assert_future::<Result<Self::Output, Fut::Output>, _>(Unless::new(self, condition))
}

Well, I said I'm a newbie, so anybody more experienced will definitely find some rough edges and I don't doubt that are problems that I can't see yet or don't know how to cope with properly, so that the implementation is both safe, performing well and ergonomic.

One of the issues, which I'm thinking of, is how to allow easily using the combinator in the ad-hoc way: simply said, it makes sense to me to await for an action multiple times, which means that unless can't consume self. A silly example: the signal indicates that something should happen, but not at the current point and therefore I'll have to poll the future again – this time perhaps without the unless condition. (But maybe it is not so silly: select! supports conditional guards, which seems similar to this particular case.) Here, I'm doubting the &mut self is the correct thing to specify.