asyncio: InvalidStateError in BaseSubprocessTransport._call_connection_lost() due to double set_result on exit waiters

[daandemeyer]

Bug report

Bug description

When SIGINT arrives while subprocess _connect_pipes tasks are in flight, asyncio.run() cancels the main task which cascades into cancelling _connect_pipes. This leaves _pipes_connected=False.

The fix in GH-140805 (backported as GH-141446) added code to _try_finish() that wakes up exit waiters when _pipes_connected is False to prevent hangs. However, _try_finish() then also checks whether all pipes are disconnected — and when _pipes is empty or all pipes are disconnected, it schedules _call_connection_lost(). _call_connection_lost() iterates the same _exit_waiters list and calls set_result() again, raising InvalidStateError.

The flow:

1. _connect_pipes is cancelled → _pipes_connected stays False
2. Process exits → _process_exited() → _try_finish()
3. _try_finish() sees _pipes_connected is False → calls waiter.set_result(self._returncode) for all non-cancelled waiters
4. _try_finish() checks all(p is not None and p.disconnected for p in self._pipes.values()) — this is True (vacuously when _pipes is empty, or because pipes were disconnected)
5. Schedules _call_connection_lost()
6. _call_connection_lost() iterates _exit_waiters → waiter.set_result(self._returncode) → InvalidStateError because the waiter already has a result from step 3

Suggested fix

Change if not waiter.cancelled() to if not waiter.done() in _call_connection_lost():

def _call_connection_lost(self, exc):
    try:
        self._protocol.connection_lost(exc)
    finally:
        for waiter in self._exit_waiters:
            if not waiter.done():          # was: if not waiter.cancelled()
                waiter.set_result(self._returncode)
        self._exit_waiters = None
        ...

Reproducer

import asyncio
import os
import signal
import subprocess


async def main() -> None:
    loop = asyncio.get_running_loop()
    loop.call_later(0.001, os.kill, os.getpid(), signal.SIGINT)

    procs: list[asyncio.subprocess.Process] = []

    tasks = [
        asyncio.create_task(
            asyncio.create_subprocess_exec(
                "sleep", "10",
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            )
        )
        for _ in range(50)
    ]

    try:
        await asyncio.gather(*tasks, return_exceptions=True)
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        pass
    finally:
        for task in tasks:
            if task.done() and not task.cancelled() and task.exception() is None:
                procs.append(task.result())

        for proc in procs:
            try:
                proc.kill()
            except ProcessLookupError:
                pass

        for proc in procs:
            await proc.wait()


try:
    asyncio.run(main())
except KeyboardInterrupt:
    pass

Output:

Exception in callback BaseSubprocessTransport._call_connection_lost()
handle: <Handle BaseSubprocessTransport._call_connection_lost()>
Traceback (most recent call last):
  File "/.../asyncio/events.py", line 94, in _run
    self._context.run(self._callback, *self._args)
  File "/.../asyncio/base_subprocess.py", line 282, in _call_connection_lost
    waiter.set_result(self._returncode)
asyncio.exceptions.InvalidStateError: invalid state

CPython versions tested on:

3.14.2

Operating systems tested on:

Linux

Linked PRs

• gh-103847: fix cancellation safety of asyncio.create_subprocess_exec #140805 / [3.14] gh-103847: fix cancellation safety of asyncio.create_subprocess_exec (GH-140805) #141446 (introduced the regression)

Linked PRs

• gh-145541: Fix InvalidStateError in BaseSubprocessTransport._call_connection_lost() #145554

[daandemeyer]

cc @kumaraditya303

[daandemeyer]

There are actually two variations of this bug. Both _try_finish() and _call_connection_lost() need the fix.

Variation 1 (already described above): _try_finish() wakes waiters, then _call_connection_lost() wakes them again.

Variation 2: _try_finish() wakes waiters (because _pipes_connected=False), then later a pipe transport's _call_connection_lost() fires (e.g. _UnixWritePipeTransport._call_connection_lost()), which calls _pipe_connection_lost() → _try_finish() a second time. Since _finished was not set on the first call (not all pipes were disconnected yet), and _pipes_connected is still False, it tries set_result() on the same already-resolved waiters.

Exception in callback _UnixWritePipeTransport._call_connection_lost()
handle: <Handle _UnixWritePipeTransport._call_connection_lost()>
Traceback (most recent call last):
  File "/usr/lib/python3.14/asyncio/events.py", line 94, in _run
    self._context.run(self._callback, *self._args)
  File "/usr/lib/python3.14/asyncio/unix_events.py", line 825, in _call_connection_lost
    self._protocol.connection_lost(exc)
  File "/usr/lib/python3.14/asyncio/base_subprocess.py", line 305, in connection_lost
    self.proc._pipe_connection_lost(self.fd, exc)
  File "/usr/lib/python3.14/asyncio/base_subprocess.py", line 227, in _pipe_connection_lost
    self._try_finish()
  File "/usr/lib/python3.14/asyncio/base_subprocess.py", line 269, in _try_finish
    waiter.set_result(self._returncode)
asyncio.exceptions.InvalidStateError: invalid state

So the fix needs if not waiter.done() in both places:

# In _try_finish():
if not self._pipes_connected:
    for waiter in self._exit_waiters:
        if not waiter.done():          # was: if not waiter.cancelled()
            waiter.set_result(self._returncode)

# In _call_connection_lost():
for waiter in self._exit_waiters:
    if not waiter.done():              # was: if not waiter.cancelled()
        waiter.set_result(self._returncode)

[kumaraditya303]

Do you want to submit a PR?

[daandemeyer]

I can submit a PR if you think the fix is the right way to go