Actor Pattern - Cancel a task

[georgemp]

I'm trying to write a set of chained actors to handle a HTTP request using Tokio and Reqwest. Each actor would modify or act on the request in some way, before forwarding it to the next actor in the chain. In this fashion, it would reach the terminal actor, which wraps a reqwest client and, will actually do the network fetch.

My basic actor design is as follows - This one just prints out the Http Request received.

use {crate::HttpLoaderAction, tokio::sync::mpsc};

pub enum HttpLoaderAction {
    Load(HttpRequest, oneshot::Sender<Result<HttpResponse, Error>>),
    Cancel(HttpRequestId, oneshot::Sender<Result<(), Error>>),
    Reset(oneshot::Sender<Result<(), Error>>),
}

struct PrintLoaderInternal {
    next_loader_tx: mpsc::Sender<HttpLoaderAction>,
}

pub struct PrintLoader {
    tx: mpsc::Sender<HttpLoaderAction>,
}

impl PrintLoader {
    pub fn new(next_loader_tx: mpsc::Sender<HttpLoaderAction>) -> Self {
        PrintLoader {
            tx: new_print_loader(next_loader_tx),
        }
    }

    pub async fn send(
        &self,
        msg: HttpLoaderAction,
    ) -> Result<(), mpsc::error::SendError<HttpLoaderAction>> {
        self.tx.send(msg).await
    }
}

fn new_print_loader(
    next_loader_tx: mpsc::Sender<HttpLoaderAction>,
) -> mpsc::Sender<HttpLoaderAction> {
    let (action_tx, mut action_rx) = mpsc::channel::<HttpLoaderAction>(16);
    let print_loader_internal = PrintLoaderInternal { next_loader_tx };
    tokio::spawn(async move {
        while let Some(msg) = action_rx.recv().await {
            match msg {
                HttpLoaderAction::Load(request, _) => {
                    println!("{:?}", request);
                }
                _ => {}
            };
            print_loader_internal.next_loader_tx.send(msg);
        }
    });

    action_tx
}

Each HttpRequest has a id. I would like to be able to cancel a request (by id) if needed, after it has been initiated.

I could do a cancel by setting up a cancel channel and using a tokio_select! as

tokio_select! {
    _ = my_load_task() => {},
    _ = cancel_channel.recv() => {}
}

A couple of issues that come to mind -

1. With the current design, the cancelmessage is received on the same channel as the load request. So, it would necessarily process only after the load. Ideally, I would like the chain to be broken as soon as I get a cancel message. But, at the moment I think it would propogate until it gets to a loader that actually awaits something in it's load action.

2. I could setup a second cancellation channel, spawn another task and wrap this up in tokio_select!. But, I'm not sure how I could set up the patterns to match against the Id. If I were to have two request A and B running, and, I send a message on the cancellation channel to cancel A, I imagine the message would be received by the tokio_select! wrapping B as well - even though the Id doesn't match, it would process the message, return, and effectively cancel the load of B too.

3. To avoid this I could setup a separate cancellation channel for each request, store this somewhere, and discard it once the load either completes or is cancelled.

Would there be a simpler way to do this? Thanks

[Darksonn]

Ultimately you are going to need some direct communication to the task you are cancelling. For this, there are many methods, e.g. an oneshot channel or AbortRegistration from the futures crate.

If you want cancellation messages, I would probably go for two separate channels. One for new jobs, and one for cancellations. Not only does this allow cancellations to catch up, it also prevents cancellations from being caught behind a large number of requests waiting in line. It does pose another problem though: cancellation before the task arrives. There are some tombstone methods where you remember the cancelled tasks, but this seems complicated.

To me, the simplest method appears having an oneshot receiver travel around together with the task. Then you can have a hashmap of ids somewhere when you want to send cancellations, and the final task could send messages to the owner of the hash map when tasks have finished, or you could periodically call is_closed on them to detect closed channels.