How am I supposed to handle asyncio.CancelledError on a Future?

[Rémy Hubscher]

Hello,

I am using aioredis on a BLPOP which is a blocking call on a redis connection.

redis.blpop is a Future created here: https://github.com/aio-libs/aioredis/blob/master/aioredis/connection.py#L154-L157

When using asyncio.wait_for(redis.blpop("channel"), timeout=5) if the timeout is raised before the blpop happens a asyncio.CancelledError is raised and the Future is marked as cancelled.

Later in the code I am asking for another redis.blpop("channel") and the next message is not catched because the cancelled task consume it nevertheless.

I have created a peace of code to reproduce the problem: https://github.com/aio-libs/aioredis/issues/59#issue-66158818

Do you have any idea of how I can handle this CancelledError in aioredis in order to really cancel the Future?

Thank you for your help.

Kind regards,

Rémy

[Gustavo Carneiro]

On 8 April 2015 at 11:14, Rémy Hubscher <hubsch...@gmail.com> wrote:

Hello,

I am using aioredis on a BLPOP which is a blocking call on a redis connection.

redis.blpop is a Future created here: https://github.com/aio-libs/aioredis/blob/master/aioredis/connection.py#L154-L157

When using asyncio.wait_for(redis.blpop("channel"), timeout=5) if the timeout is raised before the blpop happens a asyncio.CancelledError is raised and the Future is marked as cancelled.

Later in the code I am asking for another redis.blpop("channel") and the next message is not catched because the cancelled task consume it nevertheless.

This sounds like this bug: http://bugs.python.org/issue23812

 

I have created a peace of code to reproduce the problem: https://github.com/aio-libs/aioredis/issues/59#issue-66158818

Do you have any idea of how I can handle this CancelledError in aioredis in order to really cancel the Future?

Thank you for your help.

Kind regards,

Rémy

--

Gustavo J. A. M. Carneiro

Gambit Research
"The universe is always one step beyond logic." -- Frank Herbert

[Rémy Hubscher]

It is also what Victor Stinner said at first, but I am quite sure it is.

Also I proposed to inherit from Future to improve the cancel() call and made this patch: https://github.com/aio-libs/aioredis/pull/59/files

Could you tell me if it is the right way to do it?

[Rémy Hubscher]

Ok so continuing to investigate, my question is summarize by the following:

#!/usr/bin/env python
import asyncio
import hiredis
from aioredis.util import encode_command
from aioredis.errors import ProtocolError, ReplyError


@asyncio.coroutine
def run_me():
    reader, writer = yield from asyncio.open_connection("localhost", 6379)
    writer.write(encode_command("BLPOP", "channel", 0))
    try:
        task = asyncio.Task(reader.read(65536))
        data = yield from asyncio.wait_for(task, 1)
    except asyncio.TimeoutError:
        task.cancel()
        data = yield from reader.read(65536)
    finally:
        parser = hiredis.Reader(protocolError=ProtocolError,
                                replyError=ReplyError)
        parser.feed(data)
        return parser.gets()

data = asyncio.get_event_loop().run_until_complete(run_me())
print(data)


How can I stop the reader in order to read it again later?

[Rémy Hubscher]

I tried to use reader.feed_data() for that but the next call to yield from reader.read() return the data feeded.

[Rémy Hubscher]

Ok so to end this thread, @popravich told me that the problem is on the server side, since cancelling reader.read() doesn't actually stop the server from waiting for the request.

For the BLPOP timeout case, we should probably let the redis server handle the timeout (as the BPOP command let us do so).

To handle the CancelledSignal, @popravich is going to rework the RedisPool and RedisConnection so that RedisConnection uses a pool to make sure that each command will be executed in a connection ready pool and that we can safely close a connection in case of CancelledError when waiting for the server answer.

Thank you all for your help on this.

Best regards,

Rémy