aiohttp swallows asyncio.CancelledError during connection timeout

[ssigut]

Describe the bug

There is a race condition in code that handles connection timeout. If you call cancel on a task that is currently pending in create_connection and connection timeout was already fired then asyncio.CancelledError is not propagated and you get asyncio.TimeoutError instead. The main problem is in how timeouts are handled in async_timeout package. When exitting the context manager after timeout had passed all CancelledError exceptions are swallowed and TimeoutError is raised instead. Unfortunately this is true also if you explicitly cancel the task yourself.

The main problem is that you cannot cancel a task that is using aiohttp because you never know if CancelledError will be raised.

To Reproduce

EDIT: THIS REPRODUCER DOES NOT SHOW THE BEHAVIOUR CORRECTLY - PLEASE REFER TO COMMENTS BELLOW!

import asyncio
from async_timeout import timeout


async def test_task():
    with timeout(1):
        await asyncio.sleep(10)


async def main():
    t = asyncio.create_task(test_task())
    await asyncio.sleep(2)
    t.cancel()
    try:
        await t
    except asyncio.TimeoutError:
        print("Raised TimeoutError")
    except asyncio.CancelledError:
        print("Raised CancelledError")

asyncio.run(main())

Expected behavior

asyncio.CancelledError should never be suppressed when you cancel the task explicitly.

Logs/tracebacks

---

Python Version

Python 3.8.10

aiohttp Version

3.7.4.post0

multidict Version

4.7.6

yarl Version

1.6.0

OS

Linux

Related component

Client

Additional context

No response

Code of Conduct

• I agree to follow the aio-libs Code of Conduct

[Dreamsorcerer]

Firstly, this is not the async-timeout repo (you just linked to it).
Secondly, the timeout happens after 1 second, your cancellation happens after 2 seconds. So, the timeout has already happened and the behaviour looks correct to me. If I change 1 -> 5 I get a CancelledError as expected.

[ssigut]

I noticed this problem when using aiohttp so this is the reason I created the bug here. I understand your point, however how are you supposed to cancel task that is running aiohttp code when it can swallow the CancelledError exception?

[Dreamsorcerer]

I'm not sure what you mean, nothing was swallowed? The task was cancelled by the timeout, your t.cancel() is a noop as the task was already cancelled.

If you can demonstrate the confusing behaviour in aiohttp, or explain where it happens, then maybe we can help..

[ssigut]

Imagine you have a retry logic that retries the request when they fail. Something along these lines:

async def worker(cl: aiohttp.ClientSession):
    while True:
        try:
            r = await cl.get(..., timeout=...)
        except (asyncio.TimeoutError, OSError):
            await asyncio.sleep(10)
        else:
            do_something(r)

If you want to cancel this task, there is a risk that it will never exit.

[Dreamsorcerer]

Right, so this would better show the problem, right?

import asyncio
from async_timeout import timeout


async def test_task():
    while True:
        print("RETRY")
        try:
            with timeout(1):
                await asyncio.sleep(10)
        except asyncio.TimeoutError:
            pass


async def main():
    t = asyncio.create_task(test_task())
    await asyncio.sleep(2)
    t.cancel()
    await t

asyncio.run(main())

This will sometimes run forever without cancelling the task.

Let me think about it....

[ssigut]

The main problem is that these two operations are not atomic:

• task cancellation done by the timeout trigger task spawned in async-timeout
• raising TimeoutError in timeout context manager

You can actually squeeze an explicit Task.cancel() operation between them and thus loose the error. Now I realize my first reproducer was not really showing what I meant.

[ssigut]

Right, so this would better show the problem, right?

import asyncio
from async_timeout import timeout


async def test_task():
    while True:
        print("RETRY")
        try:
            with timeout(1):
                await asyncio.sleep(10)
        except asyncio.TimeoutError:
            pass


async def main():
    t = asyncio.create_task(test_task())
    await asyncio.sleep(2)
    t.cancel()
    await t

asyncio.run(main())

Let me think about it....

Your example would not cause the problem. It is really hard to create good reproducer for it. It is highly dependent on timing and I am getting the issue in about 5 cancellations in 1000.

[Dreamsorcerer]

I get about 50/50 on that example whether the task cancels or sits in an infinite loop.

[ssigut]

Oh ok, I was confused because for me it terminates every time, however when I added one print inside the loop, then it runs in infinite loop. So yeah - it is dependent on the timing, however it pretty much demonstrates what I had on mind.

[ssigut]

As I wrote the problem is that the timeout triggered cancellation and the actual raise TimeoutError are done in two separate operations and not atomically. You can therefore call explicit Task.cancel() on your task and loose the Exception if you time the cancel wrong - (timeout cancellation task ran by the context manager has already triggered, however TimeoutError was not yet raised). The problem here is the asyncio context switch upon await inside the context manager.

It probably creates issues also if you sequence your explicit cancellation right before the timeout firing as well. During high contention I get errors pretty much every run even when not trying to reproduce the issue.

[ssigut]

As I thought it works also the opposite way (sequencing explicit cancellation right before firing the timeout):

import asyncio
from async_timeout import timeout

t = None

async def test_task2():
    try:
        with timeout(1):
            print("Waiting for timeout")
            await asyncio.sleep(10)
    except:
        print("Done waiting for timeout")
        raise

async def test_task1():
    print("Starting explicit cancellation task")
    await asyncio.sleep(1)
    print("Explicitly cancelling the task")
    t.cancel()


async def main():
    asyncio.create_task(test_task1())
    global t
    t = asyncio.create_task(test_task2())
    await asyncio.sleep(3)
    try:
        await t
    except BaseException as e:
        print(f"{type(e).__name__}({e})")

asyncio.run(main())

Produces this output for me:

Starting explicit cancellation task
Waiting for timeout
Explicitly cancelling the task
Done waiting for timeout
TimeoutError()

Clearly explicit cancellation was done before the timeout was fired yet it still raised TimeoutError.

If you try to change the sleep in task1 to 0.9 then you get CancelledError because asyncio gets chance to schedule task2 before the timeout is fired and thus can raise the exception early before overwriting it to TimeoutError

[Dreamsorcerer]

OK, previous examples make it clear why it's a problem. This is a minimal reproducer that can be used as a test. If you add a print to Timeout._on_timeout(), you'll see that the cancel happens before the timeout, but the exit condition will still override this to a cancelled error:

async def test_task(deadline, loop):
    print("TRY")
    with Timeout(deadline, loop):
        await asyncio.sleep(10)


async def main():
    loop = asyncio.get_running_loop()
    deadline = loop.time() + 1
    t = asyncio.create_task(test_task(deadline, loop))

    def cancel():
        print("CANCEL")
        t.cancel()

    loop.call_at(deadline, cancel)
    await asyncio.sleep(2)
    await t

[Dreamsorcerer]

This is rather tricky to resolve as I just don't see any information being available to figure out where the cancellations came from (or how many there were).

Playing with private variables, I can get the previous example to work by changing Timeout._on_timeout() to this:

        if task._fut_waiter and task._fut_waiter.cancelled():
            return
        task.cancel()
        self._state = _State.TIMEOUT

But, if the cancel happens between the timeout and the exit, I can't see any way to tell that the task has been cancelled a second time, which I can reproduce with this example:

async def test_task(deadline, loop):
    print("TRY")
    with Timeout(deadline, loop):
        await asyncio.sleep(10)


async def main():
    loop = asyncio.get_running_loop()
    deadline = loop.time() + 1
    t = asyncio.create_task(test_task(deadline, loop))

    def cancel():
        print("CANCEL")
        t.cancel()

    loop.call_at(deadline+.000001, cancel)
    await asyncio.sleep(2)
    await t