tokio::select with async redis client and channel for cancelation

[BratSinot]

Greetings!

I have some problem with task cancellation. So, example:

            let mut redis = ConnectionManager::new(REDIS_CLIENT.clone()).await?;
<...>
let (tx, mut cancel) = mpsc::channel(1);
let task = tokio::task::spawn(async move {
    loop {
        let data = tokio::select! {
            data = redis.xread_options::<_, _, StreamReadReply>(mwi, &now, &opts) => data,
            _ = cancel.recv() => break,
        };
<...>
    }
});

When I send tx.send(()).await;, yeah, loop are breaking. But, future with redis.xread_options (from what I read) just suspends without releasing data. So, as result, I got dangling connection to Redis DB.

P.S. I'm using redis from here: https://github.com/mitsuhiko/redis-rs

[Darksonn]

Well, if xread_options doesn't support cancellation, then there isn't much you can do. One change you might make is the following:

loop {
    if cancel.try_recv() != Err(TryRecvError::Empty) { break; }
    let data = redis.xread_options::<_, _, StreamReadReply>(mwi, &now, &opts);
    if cancel.try_recv() != Err(TryRecvError::Empty) { break; }
    ...
}

Of course this way the cancellation does not take effect immediately, but if Redis doesn't support that, there isn't much you can do about it. One variant would be to do something like this:

loop {
    if cancel.try_recv() != Err(TryRecvError::Empty) { break; }
    let data = tokio::select! {
        data = redis.xread_options::<_, _, StreamReadReply>(mwi, &now, &opts) => data,
        _ = async { (&mut cancel).await; sleep(Duration::from_millis(500)).await; } => break,
    };
    if cancel.try_recv() != Err(TryRecvError::Empty) { break; }
    ...
}

Then we only cancel it if it has stalled for 500 ms. You would just close the connection if it gets cancelled during the operation.

[BratSinot]

I noticed one more thing: if xread get some data after abortion, connection will be closed.

[BratSinot]

In your second variant, the problem will be the same. Break in my code works, but future returned from redis.xread_options is saved somewhere and connection isn't closed.

[Darksonn]

You will probably need to ask the authors of the redis crate if you want additional guidance.