Race Condition with Asyncio Cancelled Tasks

[TheTechromancer]

Hello, I was directed to open a discussion here based on this one over at httpx.

The gist of the race condition is that when a web request is cancelled within a specific time window, it will degrade the state of the connection pool, leaving it unable to issue further requests.

It can be reproduced with the following code:

import httpcore
import asyncio
import traceback

client = httpcore.AsyncConnectionPool()

### STEP 1: python -m http.server ###

async def cancel_tasks(tasks):
    for task in tasks:
        task.cancel()
    for task in tasks:
        try:
            await task
        except asyncio.CancelledError:
            pass


async def return_first_response():
    num_tasks = 100
    tasks = [
        asyncio.create_task(
            client.request(
                "GET",
                "http://127.0.0.1:8000",
                extensions={
                    "timeout": {"read": 5, "write": 5, "connect": 5, "pool": 5}
                },
            )
        )
        for _ in range(num_tasks)
    ]
    done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
    done = next(iter(done))
    print("+", end="", flush=True)
    await cancel_tasks([t for t in tasks if t != done])
    return await done


async def main():
    while 1:
        try:
            await return_first_response()
            print(".", end="", flush=True)
        except Exception:
            print(f"Got error: {traceback.format_exc()}")
            continue

asyncio.run(main())

In essence, this code will create a batch of web requests, get the first successful result, then cancel the rest, rinse and repeat.

On my machine it runs for about ten to twenty iterations, hangs for the specified timeout (5 seconds), then throws an httpx.PoolTimeout error and fails to recover.

Thanks!

[karpetrosyan]

It's strange because at the end of the program, the connection pool is full of closed connections; these connections were supposed to be removed from the pool, but that doesn't happen.

[TheTechromancer]

I suppose that means whichever task is responsible for removing the closed connections is most likely the one being inadvertently cancelled.

[TheTechromancer]

Did a little more digging, and from the traceback it's clear the task being cancelled is from anyio (httpcore does not itself create tasks):

asyncio.exceptions.CancelledError: Cancelled by cancel scope 7ff0f7ff1d10

At least some of the strange behavior seems to originate inside anyio, since instead of propagating the exception immediately, it waits for timeout seconds before raising a TimeoutError. As we all know async cancellations are a pain for everyone everywhere, so maybe that's intended, but in any case the behavior is slightly interesting.

Full traceback for reference:

Traceback (most recent call last):
  File "/usr/lib/python3.11/site-packages/anyio/_core/_tasks.py", line 115, in fail_after
    yield cancel_scope
  File "/usr/lib/python3.11/site-packages/httpcore/_synchronization.py", line 125, in wait
    await self._anyio_event.wait()
  File "/usr/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 1616, in wait
    await self._event.wait()
  File "/usr/lib/python3.11/asyncio/locks.py", line 213, in wait
    await fut
asyncio.exceptions.CancelledError: Cancelled by cancel scope 7ff0f7ff1d10

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/lib/python3.11/site-packages/httpcore/_exceptions.py", line 10, in map_exceptions
    yield
  File "/usr/lib/python3.11/site-packages/httpcore/_synchronization.py", line 124, in wait
    with anyio.fail_after(timeout):
  File "/usr/lib/python3.11/contextlib.py", line 155, in __exit__
    self.gen.throw(typ, value, traceback)
  File "/usr/lib/python3.11/site-packages/anyio/_core/_tasks.py", line 118, in fail_after
    raise TimeoutError
TimeoutError

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/usr/lib/python3.11/site-packages/httpcore/_async/interfaces.py", line 43, in request
    response = await self.handle_async_request(request)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.11/site-packages/httpcore/_async/connection_pool.py", line 242, in handle_async_request
    raise exc
  File "/usr/lib/python3.11/site-packages/httpcore/_async/connection_pool.py", line 233, in handle_async_request
    connection = await status.wait_for_connection(timeout=timeout)
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.11/site-packages/httpcore/_async/connection_pool.py", line 35, in wait_for_connection
    await self._connection_acquired.wait(timeout=timeout)
  File "/usr/lib/python3.11/site-packages/httpcore/_synchronization.py", line 123, in wait
    with map_exceptions(anyio_exc_map):
  File "/usr/lib/python3.11/contextlib.py", line 155, in __exit__
    self.gen.throw(typ, value, traceback)
  File "/usr/lib/python3.11/site-packages/httpcore/_exceptions.py", line 14, in map_exceptions
    raise to_exc(exc) from exc
httpcore.PoolTimeout

[karpetrosyan]

Oh, that's difficult to track. This part of httpcore is very complex and not always intuitive. What are your thoughts on this, @tomchristie ?

[tomchristie]

So taking a bit of a look into it, and my first response here is... just use trio.

(And when you can't use trio, just use anyio.)

asyncio exposes a who bunch of API primitives that you really don't want to have access too, and from the example above I can't immediately tell if the problem is something to do with httpx cancellation semantics, or something to do with asyncio gnarliness.

My first pass to something like this is "how does this look if I re-write it with a trio/anyio design.

Here's what I came up with...

import trio
import httpcore


TIMEOUT = {"read": 5, "write": 5, "connect": 5, "pool": 5}


async def send_request(client, seen_response):
    response = await client.request("GET", "http://127.0.0.1:8000", extensions={"timeout": TIMEOUT})

    # Print the first response that we see and set the "seen_response" flag.
    if not seen_response.is_set():
        print(response)
        seen_response.set()


async def main():
    async with httpcore.AsyncConnectionPool() as client:
        async with trio.open_nursery() as nursery:
            # This event is used to signal the first response we recieve.
            seen_response = trio.Event()

            # Kick off one hundred HTTP requests.
            for idx in range(100):
                nursery.start_soon(send_request, client, seen_response)
            
            # As soon as we see a response we can cancel out of the nursery,
            # rather than allowing all tasks to complete.
            await seen_response.wait()
            nursery.cancel_scope.cancel()


trio.run(main)

Which... works exactly as expected.

I then tried this using anyio...

import anyio
import httpcore


TIMEOUT = {"read": 5, "write": 5, "connect": 5, "pool": 5}


async def send_request(client, seen_response):
    response = await client.request("GET", "http://127.0.0.1:8000", extensions={"timeout": TIMEOUT})

    # Print the first response that we see and set the "seen_response" flag.
    if not seen_response.is_set():
        print(response)
        seen_response.set()


async def main():
    async with httpcore.AsyncConnectionPool() as client:
        async with anyio.create_task_group() as task_group:
            # This event is used to signal the first response we recieve.
            seen_response = anyio.Event()

            # Kick off one hundred HTTP requests.
            for idx in range(100):
                task_group.start_soon(send_request, client, seen_response)
            
            # As soon as we see a response we can cancel out of the nursery,
            # rather than allowing all tasks to complete.
            await seen_response.wait()
            task_group.cancel_scope.cancel()


anyio.run(main)

Okay, also good.

Using Python 3.11 we have TaskGroup available. We're still potentially exposed to a bunch of API we don't want, but it's much better...

import asyncio
import httpcore


TIMEOUT = {"read": 5, "write": 5, "connect": 5, "pool": 5}


async def send_request(client, seen_response):
    response = await client.request("GET", "http://127.0.0.1:8000", extensions={"timeout": TIMEOUT})

    # Print the first response that we see and set the "seen_response" flag.
    if not seen_response.is_set():
        print(response)
        seen_response.set()


async def main():
    async with httpcore.AsyncConnectionPool() as client:
        async with asyncio.TaskGroup() as task_group:
            # This event is used to signal the first response we recieve.
            seen_response = asyncio.Event()

            # Kick off one hundred HTTP requests.
            tasks = []
            for idx in range(100):
                task = task_group.create_task(send_request(client, seen_response))
                tasks.append(task)
            
            # As soon as we see a response we can cancel out of the nursery,
            # rather than allowing all tasks to complete.
            await seen_response.wait()
            for task in tasks:
                task.cancel()


asyncio.run(main())

That also works okay. Anyways, at least two things to say here...

• I'm not clear from what you've raised if there's an issue in httpcore that needs addressing or not. If someone's able to replicate the issue using structured concurrency semantics then yes we have an issue. I wasn't able to do that above.
• Trio is a more throughly designed alternative to asyncio. We're in a good position to make this clear to folks, and help lead them towards happy directions.

[TheTechromancer]

@tomchristie thanks for taking the time to look into this.

I'll admit this isn't the first cancellation-related race condition we've run into with asyncio, so I'm not opposed to a trio migration, although I just finished converting our synchronous codebase to asyncio, so that particular idea causes me some amount of weariness. 🫠 (if you felt the same way about this very issue, I wouldn't blame you).

The examples you provided indeed do not reveal the bug. But I was able to reproduce it in trio by iterating again with the same httpcore client, like so:

import trio
import httpcore


TIMEOUT = {"read": 5, "write": 5, "connect": 5, "pool": 5}


async def send_request(client, seen_response):
    response = await client.request("GET", "http://127.0.0.1:8000", extensions={"timeout": TIMEOUT})

    # Print the first response that we see and set the "seen_response" flag.
    if not seen_response.is_set():
        print(response)
        seen_response.set()


async def main():
    async with httpcore.AsyncConnectionPool() as client:
        while 1: # <-----------------
            async with trio.open_nursery() as nursery:
                # This event is used to signal the first response we recieve.
                seen_response = trio.Event()

                # Kick off one hundred HTTP requests.
                for idx in range(100):
                    nursery.start_soon(send_request, client, seen_response)
                
                # As soon as we see a response we can cancel out of the nursery,
                # rather than allowing all tasks to complete.
                await seen_response.wait()
                nursery.cancel_scope.cancel()

trio.run(main)

For me, this reproduces the httpcore.PoolTimeout error on the second iteration, seeming to support the notion that the pool is entering into a degraded state. It is perhaps not completely degraded on the first iteration, but by the second or third it reaches the point where requests are no longer possible.

Let me know what you think.

[TheTechromancer]

Is this okay to graduate to an issue?

[karpetrosyan]

Yes! I think it is an issue

[TheTechromancer]

Issue opened here: #785