Race Condition with Cancelled Asyncio Tasks

[TheTechromancer]

First off, I want to say thanks for making httpx and sorry in advance, as I know this kind of bug can be really annoying to troubleshoot. If it weren't impacting our production systems, I probably wouldn't give it much consideration.

I've run into a race condition which occurs when a task is canceled, but only about 1 out of 1000 times. This has been a difficult one to reproduce in isolation, but I finally was able to do it with the following code:

import httpx
import asyncio
import traceback

client = httpx.AsyncClient(timeout=5)

### 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.get("http://127.0.0.1:1234")) 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 httpx.HTTPError:
            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.

On further inspection it seems that if a task is cancelled at precisely the wrong time, the client is put into a degraded state, unable to issue requests. You can see this in the speed of the iterations, which over the course of this test become more and more erratic until finally the client fails to respond and generates the PoolTimeout.

The bug will not manifest if a new client is instantiated at the beginning of every iteration (e.g. the first line of return_first_response()).

Please let me know if there's anything I can do to help with squashing this.
Thanks!

[karpetrosyan]

Hello, at the moment, httpcore, the networking component of httpx, does not support cancellations; there have been several issues and discussions about this.

refs: #2742
httpcore PR which resolves this issue: encode/httpcore#726
httpcore issue: encode/httpcore#642

This issue has already been resolved in the httpcore master branch, so it should be resolved in httpx with the next httpcore release.
Next httpcore release PR: encode/httpcore#768

[TheTechromancer]

Thanks for your response, I'm glad to see there's already discussion around this.

I took some time to look over the issues/PRs you posted, and I agree those issues seem very similar. However in my tests with the latest master branch of httpcore, I'm still experiencing the bug.

It's possible this could be a separate issue, since it does seem to be a very narrow race condition. Or maybe I should be testing from a different branch?

[karpetrosyan]

Thank you very much for your efforts. I can try to assist you in locating the source of the problem, but first let's move this issue into the httpcore because httpcore handles all network staff.

This is the relevant httpcore example

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())

[karpetrosyan]

I recommend that you open a discussion in httpcore so that we can work through this in the proper repository.

[TheTechromancer]

Okay will do, thanks for your help.

[TheTechromancer]

New discussion opened here: encode/httpcore#783