Deadlock in create_subprocess_exec when using Semaphore and asyncio.subprocess.PIPE

[Timmmm]

Bug report

Bug description:

import asyncio

async def _stream_subprocess(id: int, command: "list[str]"):
    proc = await asyncio.create_subprocess_exec(*command, stdout=asyncio.subprocess.PIPE)
    await proc.wait()
    print(f"{id}: Done")


async def run(id: int, command: "list[str]"):
    print(f"{id}: Running")
    await _stream_subprocess(id, command)
    print(f"{id}: Throwing")

    raise RuntimeError("failed")


async def run_with_parallelism_limit(id: int, command: "list[str]", limit: asyncio.Semaphore):
    async with limit:
        print(f"{id}: Calling run")
        await run(id, command)
        print(f"{id}: Run finished")


async def main():
    sem = asyncio.Semaphore(1)
    await asyncio.gather(
        run_with_parallelism_limit(0, ["python", "--version"], sem),
        run_with_parallelism_limit(1, ["python", "--version"], sem),
    )


if __name__ == "__main__":
    asyncio.run(main())

Output:

0: Calling run
0: Running
0: Done
0: Throwing
1: Calling run
1: Running

And then it deadlocks. If you press Ctrl-C it seems to be stuck in Runner._cancel_all_tasks(). Unfortunately Debugpy couldn't even interrupt it at that point.

If you remove stdout=asyncio.subprocess.PIPE then it doesn't deadlock.

CPython versions tested on:

3.9, 3.11

Operating systems tested on:

Linux, Windows

[graingert]

The problem seems to be that asyncio.gather is leaking tasks when one throws and then asyncio.run is running Runner._cancel_all_tasks, if you use a TaskGroup this doesn't have an issue:

import asyncio


async def _stream_subprocess(id: int, command: "list[str]"):
    proc = await asyncio.create_subprocess_exec(
        *command, stdout=asyncio.subprocess.PIPE
    )
    await proc.wait()
    print(f"{id}: Done")


async def run(id: int, command: "list[str]"):
    print(f"{id}: Running")
    await _stream_subprocess(id, command)
    print(f"{id}: Throwing")

    raise RuntimeError("failed")


async def run_with_parallelism_limit(
    id: int, command: "list[str]", limit: asyncio.Semaphore
):
    async with limit:
        print(f"{id}: Calling run")
        await run(id, command)
        print(f"{id}: Run finished")


async def main():
    sem = asyncio.Semaphore(1)
    async with asyncio.TaskGroup() as tg:
        tg.create_task(run_with_parallelism_limit(0, ["python", "--version"], sem))
        tg.create_task(run_with_parallelism_limit(1, ["python", "--version"], sem))


if __name__ == "__main__":
    asyncio.run(main())

This issue is a duplicate of #103847