From dfdcbbee10a2d6e42f35403880e656b9efa63b28 Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sun, 19 Jul 2026 13:44:17 +0100 Subject: [PATCH 01/15] Add a shared, backend-agnostic subprocess implementation (#783) Move the subprocess lifecycle logic (Process, spawning and reaping) into a single implementation in anyio._core._subprocesses, built on subprocess.Popen() plus a small set of backend primitives: * create_subprocess_stdin_pipe / create_subprocess_output_pipe * wait_for_child_exit The asyncio backend implements the pipes with loop.connect_read_pipe / connect_write_pipe (IOCP on Windows via the ProactorEventLoop) and waits on the process handle via the proactor on Windows; on POSIX it reaps via pidfd/waitid. The Trio backend uses raw fd pipe streams on POSIX and keeps its native implementation on Windows. This removes the reliance on undocumented backend internals (e.g. StreamReader.set_exception()). This also resolves the asyncio/Trio inconsistencies from discussion #828: * Process.wait() no longer waits for the standard streams to close, so it returns promptly once the process exits (even with a pipe inherited by a grandchild) * Process.returncode now polls on both backends, so it no longer returns a stale None after exit * signalling an already-exited process is consistently a no-op Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01LZMpdx7hNR89fwdmiMFmVJ --- docs/versionhistory.rst | 19 ++ src/anyio/_backends/_asyncio.py | 368 +++++++++++++++++-------------- src/anyio/_backends/_trio.py | 120 +++++++++- src/anyio/_core/_subprocesses.py | 234 +++++++++++++++++++- src/anyio/abc/_eventloop.py | 37 ++++ tests/test_subprocesses.py | 38 +++- 6 files changed, 643 insertions(+), 173 deletions(-) diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst index a2303bf77..345d4ece5 100644 --- a/docs/versionhistory.rst +++ b/docs/versionhistory.rst @@ -3,6 +3,25 @@ Version history This library adheres to `Semantic Versioning 2.0 `_. +**UNRELEASED** + +- Changed subprocess handling to use a shared, backend-agnostic implementation built on + ``subprocess.Popen()`` plus a small set of backend primitives (asynchronous pipes and + child-process reaping). This is used by the asyncio backend on all platforms and by the + Trio backend on POSIX (Trio keeps its native implementation on Windows), and removes the + reliance on undocumented backend internals + (`#783 `_; PR by @graingert) +- Fixed several inconsistencies between the asyncio and Trio backends when working with + subprocesses (`#828 `_; PR by + @graingert): + + * ``Process.wait()`` no longer waits for the standard streams to close, so it returns + promptly once the process exits, even when a pipe is inherited by a grandchild process + * ``Process.returncode`` now polls the process on both backends, so it no longer returns + a stale ``None`` after the process has exited + * Signalling an already-exited process via ``terminate()``, ``kill()`` or + ``send_signal()`` is now consistently a no-op instead of raising on asyncio + **4.14.2** - Changed ``ByteReceiveStream.receive()`` implementations to raise a ``ValueError`` when diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py index c00c2cd9b..59635ca12 100644 --- a/src/anyio/_backends/_asyncio.py +++ b/src/anyio/_backends/_asyncio.py @@ -7,6 +7,7 @@ import math import os import socket +import subprocess import sys import threading import weakref @@ -34,7 +35,6 @@ from concurrent.futures import Future from contextlib import AbstractContextManager from contextvars import Context, copy_context -from dataclasses import dataclass, field from functools import partial, wraps from inspect import ( CORO_RUNNING, @@ -42,7 +42,6 @@ getcoroutinestate, ) from io import IOBase -from os import PathLike from queue import Queue from signal import Signals from socket import AddressFamily, SocketKind @@ -1067,150 +1066,172 @@ def stop(self, f: asyncio.Task | None = None) -> None: # -@dataclass(eq=False) -class StreamReaderWrapper(abc.ByteReceiveStream): - _stream: asyncio.StreamReader +class _ProcessPipeProtocol(asyncio.Protocol): + """ + Protocol used for the pipes connected to a subprocess' standard streams. - async def receive(self, max_bytes: int = 65536) -> bytes: - if max_bytes < 1: - raise ValueError("max_bytes must be a positive integer") + It works for both read pipes (:meth:`~asyncio.loop.connect_read_pipe`) and write + pipes (:meth:`~asyncio.loop.connect_write_pipe`); the same code path is therefore + used on POSIX (selector loop) and on Windows (``ProactorEventLoop``, i.e. IOCP). + """ - data = await self._stream.read(max_bytes) - if data: - return data - else: - raise EndOfStream + read_queue: deque[bytes] + read_event: asyncio.Event + write_event: asyncio.Event + exception: Exception | None = None + is_at_eof: bool = False - async def aclose(self) -> None: - self._stream.set_exception(ClosedResourceError()) - await AsyncIOBackend.checkpoint() + def connection_made(self, transport: asyncio.BaseTransport) -> None: + self.read_queue = deque() + self.read_event = asyncio.Event() + self.write_event = asyncio.Event() + self.write_event.set() + # Only write transports support this (and only meaningfully so); read + # transports either lack the method or raise NotImplementedError (uvloop) + if hasattr(transport, "set_write_buffer_limits"): + try: + transport.set_write_buffer_limits(0) + except NotImplementedError: + pass + def connection_lost(self, exc: Exception | None) -> None: + if exc: + self.exception = exc -@dataclass(eq=False) -class StreamWriterWrapper(abc.ByteSendStream): - _stream: asyncio.StreamWriter - _closed: bool = field(init=False, default=False) + self.read_event.set() + self.write_event.set() - async def send(self, item: bytes) -> None: - await AsyncIOBackend.checkpoint_if_cancelled() - stream_paused = self._stream._protocol._paused # type: ignore[attr-defined] - try: - self._stream.write(item) - await self._stream.drain() - except (ConnectionResetError, BrokenPipeError, RuntimeError) as exc: - # If closed by us and/or the peer: - # * on stdlib, drain() raises ConnectionResetError or BrokenPipeError - # * on uvloop and Winloop, write() eventually starts raising RuntimeError - if self._closed: - raise ClosedResourceError from exc - elif self._stream.is_closing(): - raise BrokenResourceError from exc + def data_received(self, data: bytes) -> None: + # ProactorEventloop sometimes sends bytearray instead of bytes + self.read_queue.append(bytes(data)) + self.read_event.set() - raise + def eof_received(self) -> bool | None: + self.is_at_eof = True + self.read_event.set() + return True - if not stream_paused: - await AsyncIOBackend.cancel_shielded_checkpoint() + def pause_writing(self) -> None: + self.write_event = asyncio.Event() - async def aclose(self) -> None: - self._closed = True - self._stream.close() - await AsyncIOBackend.checkpoint() + def resume_writing(self) -> None: + self.write_event.set() -@dataclass(eq=False) -class Process(abc.Process): - _process: asyncio.subprocess.Process - _stdin: StreamWriterWrapper | None - _stdout: StreamReaderWrapper | None - _stderr: StreamReaderWrapper | None - _exited: asyncio.Event - _transport: asyncio.SubprocessTransport +class _ProcessReceivePipeStream(abc.ByteReceiveStream): + def __init__( + self, transport: asyncio.ReadTransport, protocol: _ProcessPipeProtocol + ) -> None: + self._transport = transport + self._protocol = protocol + self._receive_guard = ResourceGuard("reading from") + self._closed = False + transport.pause_reading() + + async def receive(self, max_bytes: int = 65536) -> bytes: + if max_bytes < 1: + raise ValueError("max_bytes must be a positive integer") + + with self._receive_guard: + if ( + not self._protocol.read_event.is_set() + and not self._transport.is_closing() + and not self._protocol.is_at_eof + ): + self._transport.resume_reading() + await self._protocol.read_event.wait() + self._transport.pause_reading() + else: + await AsyncIOBackend.checkpoint() - async def aclose(self) -> None: - with CancelScope(shield=True) as scope: - # We need to close the underlying pipe_transports as well to allow a - # process blocking on full buffers to receive SIGPIPE and exit. - if self._stdin: - await self._stdin.aclose() - if pipe := self._transport.get_pipe_transport(0): - pipe.close() - if self._stdout: - await self._stdout.aclose() - if pipe := self._transport.get_pipe_transport(1): - pipe.close() - if self._stderr: - await self._stderr.aclose() - if pipe := self._transport.get_pipe_transport(2): - pipe.close() - - scope.shield = False try: - await self.wait() - except BaseException: - scope.shield = True - # Closing the transport on asyncio also handles sending kill - self._transport.close() - await self.wait() - raise + chunk = self._protocol.read_queue.popleft() + except IndexError: + if self._closed: + raise ClosedResourceError from None + elif self._protocol.exception: + raise BrokenResourceError from self._protocol.exception + else: + # EOF reached and drained; close the transport to release the + # underlying pipe (some loops, e.g. uvloop, don't do this at + # teardown, leading to ResourceWarnings) + if not self._transport.is_closing(): + self._transport.close() + + raise EndOfStream from None + + if len(chunk) > max_bytes: + # Split the oversized chunk + chunk, leftover = chunk[:max_bytes], chunk[max_bytes:] + self._protocol.read_queue.appendleft(leftover) - async def wait(self) -> int: - await self._exited.wait() - assert self._process.returncode is not None - return self._process.returncode + # If the read queue is empty, clear the flag so that the next call will + # block until data is available + if not self._protocol.read_queue: + self._protocol.read_event.clear() - def terminate(self) -> None: - self._process.terminate() + return chunk - def kill(self) -> None: - self._process.kill() + async def aclose(self) -> None: + self._closed = True + if not self._transport.is_closing(): + self._transport.close() - def send_signal(self, signal: int) -> None: - self._process.send_signal(signal) + await AsyncIOBackend.checkpoint() - @property - def pid(self) -> int: - return self._process.pid - @property - def returncode(self) -> int | None: - return self._process.returncode +class _ProcessSendPipeStream(abc.ByteSendStream): + def __init__( + self, transport: asyncio.WriteTransport, protocol: _ProcessPipeProtocol + ) -> None: + self._transport = transport + self._protocol = protocol + self._send_guard = ResourceGuard("writing to") + self._closed = False - @property - def stdin(self) -> abc.ByteSendStream | None: - return self._stdin + async def send(self, item: bytes) -> None: + with self._send_guard: + await AsyncIOBackend.checkpoint() + if self._closed: + raise ClosedResourceError + elif self._protocol.exception is not None: + raise BrokenResourceError from self._protocol.exception + elif self._transport.is_closing(): + # The child closed its end of the pipe + raise BrokenResourceError - @property - def stdout(self) -> abc.ByteReceiveStream | None: - return self._stdout + try: + self._transport.write(item) + except RuntimeError as exc: + if self._transport.is_closing(): + raise BrokenResourceError from exc + else: + raise - @property - def stderr(self) -> abc.ByteReceiveStream | None: - return self._stderr + await self._protocol.write_event.wait() + + async def aclose(self) -> None: + self._closed = True + if not self._transport.is_closing(): + self._transport.close() + + await AsyncIOBackend.checkpoint() def _forcibly_shutdown_process_pool_on_exit( - workers: set[Process], _task: object + workers: set[abc.Process], _task: object ) -> None: - """ - Forcibly shuts down worker processes belonging to this event loop.""" - child_watcher: asyncio.AbstractChildWatcher | None = None # type: ignore[name-defined] - if sys.version_info < (3, 12): - try: - child_watcher = asyncio.get_event_loop_policy().get_child_watcher() - except NotImplementedError: - pass - + """Forcibly shuts down worker processes belonging to this event loop.""" # Close as much as possible (w/o async/await) to avoid warnings for process in workers.copy(): if process.returncode is not None: continue - process._stdin._stream._transport.close() # type: ignore[union-attr] - process._stdout._stream._transport.close() # type: ignore[union-attr] - process._stderr._stream._transport.close() # type: ignore[union-attr] + for stream in (process.stdin, process.stdout, process.stderr): + if stream is not None: + stream._transport.close() # type: ignore[union-attr] + process.kill() - if child_watcher: - child_watcher.remove_child_handler(process.pid) async def _shutdown_process_pool_on_exit(workers: set[abc.Process]) -> None: @@ -2427,24 +2448,6 @@ def run_test( self._raise_async_exceptions() -class _ProcessStreamProtocol(asyncio.subprocess.SubprocessStreamProtocol): - """ - A subprocess protocol that allows us to be notified of ``process_exited`` - - asyncio's own ``Process.wait()`` only resolves once every pipe transport has - disconnected so to get same semantics as on trio and uvloop we need this. - """ - - def __init__(self) -> None: - # Match the standard factory for asyncio.create_process - super().__init__(limit=2**16, loop=asyncio.get_running_loop()) - self.exited = asyncio.Event() - - def process_exited(self) -> None: - super().process_exited() - self.exited.set() - - class AsyncIOBackend(AsyncBackend): @classmethod def run( @@ -2723,46 +2726,77 @@ async def open_process( stdout: int | IO[Any] | None, stderr: int | IO[Any] | None, **kwargs: Any, - ) -> Process: - await cls.checkpoint() - if isinstance(command, PathLike): - command = os.fspath(command) + ) -> abc.Process: + from .._core._subprocesses import _spawn_process - # Use loop.subprocess_shell()/subprocess_exec() rather than their - # asyncio.create_subprocess_*() counterparts to get access to - # transport/protocol. + return await _spawn_process( + command, stdin=stdin, stdout=stdout, stderr=stderr, **kwargs + ) + + @classmethod + async def create_subprocess_stdin_pipe(cls) -> tuple[abc.ByteSendStream, int]: loop = asyncio.get_running_loop() - if isinstance(command, (str, bytes)): - transport, protocol = await loop.subprocess_shell( - _ProcessStreamProtocol, - command, - stdin=stdin, - stdout=stdout, - stderr=stderr, - **kwargs, + if sys.platform == "win32": + import msvcrt + from asyncio.windows_utils import PipeHandle + from asyncio.windows_utils import pipe as windows_pipe + + # The write end (our end) uses overlapped (IOCP) I/O + read_handle, write_handle = windows_pipe(overlapped=(False, True)) + pipe_obj: Any = PipeHandle(write_handle) + child_fd = msvcrt.open_osfhandle(read_handle, os.O_RDONLY) + else: + read_fd, write_fd = os.pipe() + pipe_obj = os.fdopen(write_fd, "wb", 0) + child_fd = read_fd + + try: + transport, protocol = await loop.connect_write_pipe( + _ProcessPipeProtocol, pipe_obj ) + except BaseException: + os.close(child_fd) + raise + + return _ProcessSendPipeStream(transport, protocol), child_fd + + @classmethod + async def create_subprocess_output_pipe(cls) -> tuple[abc.ByteReceiveStream, int]: + loop = asyncio.get_running_loop() + if sys.platform == "win32": + import msvcrt + from asyncio.windows_utils import PipeHandle + from asyncio.windows_utils import pipe as windows_pipe + + # The read end (our end) uses overlapped (IOCP) I/O + read_handle, write_handle = windows_pipe(overlapped=(True, False)) + pipe_obj: Any = PipeHandle(read_handle) + child_fd = msvcrt.open_osfhandle(write_handle, 0) else: - transport, protocol = await loop.subprocess_exec( - _ProcessStreamProtocol, - *command, - stdin=stdin, - stdout=stdout, - stderr=stderr, - **kwargs, + read_fd, write_fd = os.pipe() + pipe_obj = os.fdopen(read_fd, "rb", 0) + child_fd = write_fd + + try: + transport, protocol = await loop.connect_read_pipe( + _ProcessPipeProtocol, pipe_obj ) + except BaseException: + os.close(child_fd) + raise - process = asyncio.subprocess.Process(transport, protocol, loop) - stdin_stream = StreamWriterWrapper(process.stdin) if process.stdin else None - stdout_stream = StreamReaderWrapper(process.stdout) if process.stdout else None - stderr_stream = StreamReaderWrapper(process.stderr) if process.stderr else None - return Process( - process, - stdin_stream, - stdout_stream, - stderr_stream, - protocol.exited, - transport, - ) + return _ProcessReceivePipeStream(transport, protocol), child_fd + + @classmethod + async def wait_for_child_exit(cls, process: subprocess.Popen) -> None: + if sys.platform == "win32": + loop = asyncio.get_running_loop() + # Wait on the process handle via IOCP; no worker thread involved + await loop._proactor.wait_for_handle(int(process._handle)) # type: ignore[attr-defined] + else: + from .._core._subprocesses import wait_for_child_exit + + await wait_for_child_exit(process) @classmethod def setup_process_pool_exit_at_shutdown(cls, workers: set[abc.Process]) -> None: @@ -2771,7 +2805,7 @@ def setup_process_pool_exit_at_shutdown(cls, workers: set[abc.Process]) -> None: name="AnyIO process pool shutdown task", ) find_root_task().add_done_callback( - partial(_forcibly_shutdown_process_pool_on_exit, workers) # type:ignore[arg-type] + partial(_forcibly_shutdown_process_pool_on_exit, workers) ) @classmethod diff --git a/src/anyio/_backends/_trio.py b/src/anyio/_backends/_trio.py index 43d24d233..140351ec3 100644 --- a/src/anyio/_backends/_trio.py +++ b/src/anyio/_backends/_trio.py @@ -4,6 +4,7 @@ import math import os import socket +import subprocess import sys import types import weakref @@ -414,6 +415,90 @@ def stderr(self) -> abc.ByteReceiveStream | None: return self._stderr +class _ProcessPipeReceiveStream(abc.ByteReceiveStream): + def __init__(self, fd: int) -> None: + self._fd = fd + self._closed = False + self._receive_guard = ResourceGuard("reading from") + os.set_blocking(fd, False) + + async def receive(self, max_bytes: int = 65536) -> bytes: + if max_bytes < 1: + raise ValueError("max_bytes must be a positive integer") + + with self._receive_guard: + while True: + if self._closed: + raise ClosedResourceError + + try: + data = os.read(self._fd, max_bytes) + except BlockingIOError: + try: + await wait_readable(self._fd) + except trio.ClosedResourceError: + raise ClosedResourceError from None + except OSError as exc: + if self._closed: + raise ClosedResourceError from None + + raise BrokenResourceError from exc + else: + if data: + return data + + raise EndOfStream + + async def aclose(self) -> None: + if not self._closed: + self._closed = True + notify_closing(self._fd) + os.close(self._fd) + + await trio.lowlevel.checkpoint() + + +class _ProcessPipeSendStream(abc.ByteSendStream): + def __init__(self, fd: int) -> None: + self._fd = fd + self._closed = False + self._send_guard = ResourceGuard("writing to") + os.set_blocking(fd, False) + + async def send(self, item: bytes) -> None: + with self._send_guard: + await trio.lowlevel.checkpoint() + view = memoryview(item) + while view: + if self._closed: + raise ClosedResourceError + + try: + bytes_sent = os.write(self._fd, view) + except BlockingIOError: + try: + await wait_writable(self._fd) + except trio.ClosedResourceError: + raise ClosedResourceError from None + except BrokenPipeError as exc: + raise BrokenResourceError from exc + except OSError as exc: + if self._closed: + raise ClosedResourceError from None + + raise BrokenResourceError from exc + else: + view = view[bytes_sent:] + + async def aclose(self) -> None: + if not self._closed: + self._closed = True + notify_closing(self._fd) + os.close(self._fd) + + await trio.lowlevel.checkpoint() + + class _ProcessPoolShutdownInstrument(trio.abc.Instrument): def after_run(self) -> None: super().after_run() @@ -1208,7 +1293,14 @@ async def open_process( stdout: int | IO[Any] | None, stderr: int | IO[Any] | None, **kwargs: Any, - ) -> Process: + ) -> abc.Process: + if sys.platform != "win32": + from .._core._subprocesses import _spawn_process + + return await _spawn_process( + command, stdin=stdin, stdout=stdout, stderr=stderr, **kwargs + ) + def convert_item(item: StrOrBytesPath) -> str: str_or_bytes = os.fspath(item) if isinstance(str_or_bytes, str): @@ -1240,6 +1332,32 @@ def convert_item(item: StrOrBytesPath) -> str: stderr_stream = ReceiveStreamWrapper(process.stderr) if process.stderr else None return Process(process, stdin_stream, stdout_stream, stderr_stream) + @classmethod + async def create_subprocess_stdin_pipe(cls) -> tuple[abc.ByteSendStream, int]: + read_fd, write_fd = os.pipe() + try: + return _ProcessPipeSendStream(write_fd), read_fd + except BaseException: + os.close(write_fd) + os.close(read_fd) + raise + + @classmethod + async def create_subprocess_output_pipe(cls) -> tuple[abc.ByteReceiveStream, int]: + read_fd, write_fd = os.pipe() + try: + return _ProcessPipeReceiveStream(read_fd), write_fd + except BaseException: + os.close(read_fd) + os.close(write_fd) + raise + + @classmethod + async def wait_for_child_exit(cls, process: subprocess.Popen) -> None: + from .._core._subprocesses import wait_for_child_exit + + await wait_for_child_exit(process) + @classmethod def setup_process_pool_exit_at_shutdown(cls, workers: set[abc.Process]) -> None: trio.lowlevel.spawn_system_task(_shutdown_process_pool, workers) diff --git a/src/anyio/_core/_subprocesses.py b/src/anyio/_core/_subprocesses.py index a6590ca62..ec9473a2c 100644 --- a/src/anyio/_core/_subprocesses.py +++ b/src/anyio/_core/_subprocesses.py @@ -1,17 +1,247 @@ from __future__ import annotations +import math +import os +import subprocess +import sys from collections.abc import AsyncIterable, Iterable, Mapping, Sequence +from functools import partial from io import BytesIO from os import PathLike +from signal import Signals from subprocess import PIPE, CalledProcessError, CompletedProcess from typing import IO, Any, TypeAlias, cast -from ..abc import Process +from ..abc import ByteReceiveStream, ByteSendStream, Process +from ..lowlevel import RunVar from ._eventloop import get_async_backend -from ._tasks import create_task_group +from ._synchronization import CapacityLimiter, Lock +from ._tasks import CancelScope, create_task_group StrOrBytesPath: TypeAlias = str | bytes | PathLike[str] | PathLike[bytes] +# A dedicated, unbounded limiter for the (potentially long-lived) child-reaping worker +# threads, so that they don't starve the default thread limiter. +_child_reaper_limiter: RunVar[CapacityLimiter] = RunVar("_child_reaper_limiter") + + +def _get_child_reaper_limiter() -> CapacityLimiter: + try: + return _child_reaper_limiter.get() + except LookupError: + limiter = CapacityLimiter(math.inf) + _child_reaper_limiter.set(limiter) + return limiter + + +def _sync_wait_reapable(pid: int) -> None: + """Block until the process ``pid`` is reapable, without consuming its exit status.""" + while True: + try: + os.waitid(os.P_PID, pid, os.WEXITED | os.WNOWAIT) + except ChildProcessError: + # Already reaped elsewhere + return + except InterruptedError: + continue + else: + return + + +async def wait_for_child_exit(process: subprocess.Popen) -> None: + """ + Backend-agnostic POSIX implementation of + :meth:`~anyio.abc.AsyncBackend.wait_for_child_exit`. + + On Linux this waits on a pidfd (no worker thread required); on other POSIX systems it + falls back to a ``waitid()`` call in a worker thread. Either way the child's exit + status is left intact so that :meth:`subprocess.Popen.wait` can consume it. + """ + backend = get_async_backend() + if sys.platform == "linux": + try: + pidfd = os.pidfd_open(process.pid) + except OSError: + pass + else: + try: + await backend.wait_readable(pidfd) + finally: + os.close(pidfd) + + return + + await backend.run_sync_in_worker_thread( + _sync_wait_reapable, + (process.pid,), + abandon_on_cancel=True, + limiter=_get_child_reaper_limiter(), + ) + + +class _Process(Process): + """ + A backend-agnostic :class:`~anyio.abc.Process` implementation. + + The process itself is spawned via :class:`subprocess.Popen`; its standard streams and + the waiting for its exit are provided by small, backend-specific primitives + (:meth:`~anyio.abc.AsyncBackend.create_subprocess_stdin_pipe`, + :meth:`~anyio.abc.AsyncBackend.create_subprocess_output_pipe` and + :meth:`~anyio.abc.AsyncBackend.wait_for_child_exit`). This lifecycle logic is + therefore shared between all backends. + """ + + def __init__( + self, + popen: subprocess.Popen, + stdin: ByteSendStream | None, + stdout: ByteReceiveStream | None, + stderr: ByteReceiveStream | None, + ) -> None: + self._popen = popen + self._stdin = stdin + self._stdout = stdout + self._stderr = stderr + self._wait_lock = Lock() + + async def aclose(self) -> None: + with CancelScope(shield=True) as scope: + if self._stdin: + await self._stdin.aclose() + if self._stdout: + await self._stdout.aclose() + if self._stderr: + await self._stderr.aclose() + + scope.shield = False + try: + await self.wait() + except BaseException: + scope.shield = True + self.kill() + await self.wait() + raise + + async def wait(self) -> int: + async with self._wait_lock: + if self._popen.poll() is None: + await get_async_backend().wait_for_child_exit(self._popen) + # The exit status hasn't been consumed yet, so this returns immediately + self._popen.wait() + + return cast(int, self._popen.returncode) + + def terminate(self) -> None: + self._popen.terminate() + + def kill(self) -> None: + self._popen.kill() + + def send_signal(self, signal: Signals) -> None: + self._popen.send_signal(signal) + + @property + def pid(self) -> int: + return self._popen.pid + + @property + def returncode(self) -> int | None: + # Poll so that the return code is up to date even if wait() hasn't been called + # yet (this matches the Trio backend's historical behavior, see discussion #828) + return self._popen.poll() + + @property + def stdin(self) -> ByteSendStream | None: + return self._stdin + + @property + def stdout(self) -> ByteReceiveStream | None: + return self._stdout + + @property + def stderr(self) -> ByteReceiveStream | None: + return self._stderr + + +async def _spawn_process( + command: StrOrBytesPath | Sequence[StrOrBytesPath], + *, + stdin: int | IO[Any] | None, + stdout: int | IO[Any] | None, + stderr: int | IO[Any] | None, + **kwargs: Any, +) -> _Process: + """ + Shared, backend-agnostic implementation of + :meth:`~anyio.abc.AsyncBackend.open_process`. + + Standard streams requested as :data:`subprocess.PIPE` are connected to pipes created + by the active backend; every other value is passed through to + :class:`subprocess.Popen` unchanged. + """ + backend = get_async_backend() + await backend.checkpoint() + if isinstance(command, PathLike): + command = os.fspath(command) + + shell = isinstance(command, (str, bytes)) + stdin_stream: ByteSendStream | None = None + stdout_stream: ByteReceiveStream | None = None + stderr_stream: ByteReceiveStream | None = None + streams: list[ByteSendStream | ByteReceiveStream] = [] + child_fds: list[int] = [] + try: + if stdin == PIPE: + stdin_stream, child_fd = await backend.create_subprocess_stdin_pipe() + streams.append(stdin_stream) + child_fds.append(child_fd) + popen_stdin: Any = child_fd + else: + popen_stdin = stdin + + if stdout == PIPE: + stdout_stream, child_fd = await backend.create_subprocess_output_pipe() + streams.append(stdout_stream) + child_fds.append(child_fd) + popen_stdout: Any = child_fd + else: + popen_stdout = stdout + + if stderr == PIPE: + stderr_stream, child_fd = await backend.create_subprocess_output_pipe() + streams.append(stderr_stream) + child_fds.append(child_fd) + popen_stderr: Any = child_fd + else: + popen_stderr = stderr + + # Popen performs blocking reads on the exec-status pipe during startup, so it + # must not run in the event loop thread. + popen = await backend.run_sync_in_worker_thread( + partial( + subprocess.Popen, + command, + shell=shell, + stdin=popen_stdin, + stdout=popen_stdout, + stderr=popen_stderr, + **kwargs, + ), + (), + ) + except BaseException: + for stream in streams: + await stream.aclose() + + raise + finally: + # The child now holds its own copies of these descriptors; close ours so that + # EOF is delivered correctly once the child exits. + for child_fd in child_fds: + os.close(child_fd) + + return _Process(popen, stdin_stream, stdout_stream, stderr_stream) + async def run_process( command: StrOrBytesPath | Sequence[StrOrBytesPath], diff --git a/src/anyio/abc/_eventloop.py b/src/anyio/abc/_eventloop.py index cad3fa763..f913ccaff 100644 --- a/src/anyio/abc/_eventloop.py +++ b/src/anyio/abc/_eventloop.py @@ -1,6 +1,7 @@ from __future__ import annotations import math +import subprocess import sys from abc import ABCMeta, abstractmethod from collections.abc import AsyncIterator, Awaitable, Callable, Coroutine, Sequence @@ -38,6 +39,7 @@ UNIXDatagramSocket, UNIXSocketStream, ) + from ._streams import ByteReceiveStream, ByteSendStream from ._subprocesses import Process from ._tasks import TaskGroup from ._testing import TestRunner @@ -239,6 +241,41 @@ async def open_process( ) -> Process: pass + @classmethod + @abstractmethod + async def create_subprocess_stdin_pipe(cls) -> tuple[ByteSendStream, int]: + """ + Create a pipe for feeding data to the standard input of a subprocess. + + :return: a tuple of ``(send_stream, child_fd)`` where ``send_stream`` is the + writable end of the pipe and ``child_fd`` is a file descriptor to be passed + as the ``stdin`` argument of :class:`subprocess.Popen` (and closed by the + caller afterwards) + """ + + @classmethod + @abstractmethod + async def create_subprocess_output_pipe(cls) -> tuple[ByteReceiveStream, int]: + """ + Create a pipe for reading the standard output or error of a subprocess. + + :return: a tuple of ``(receive_stream, child_fd)`` where ``receive_stream`` is + the readable end of the pipe and ``child_fd`` is a file descriptor to be + passed as the ``stdout``/``stderr`` argument of :class:`subprocess.Popen` + (and closed by the caller afterwards) + """ + + @classmethod + @abstractmethod + async def wait_for_child_exit(cls, process: subprocess.Popen) -> None: + """ + Wait until the given child process is ready to be reaped. + + When this returns, a call to :meth:`subprocess.Popen.wait` is guaranteed to + return the exit status immediately. The exit status itself must **not** be + consumed by this call. + """ + @classmethod @abstractmethod def setup_process_pool_exit_at_shutdown(cls, workers: set[Process]) -> None: diff --git a/tests/test_subprocesses.py b/tests/test_subprocesses.py index 132a78833..fabd6ac14 100644 --- a/tests/test_subprocesses.py +++ b/tests/test_subprocesses.py @@ -2,6 +2,7 @@ import os import platform +import signal import sys from collections.abc import Callable from pathlib import Path @@ -21,6 +22,7 @@ fail_after, open_process, run_process, + sleep, ) from anyio._core._eventloop import get_async_backend from anyio.streams.buffered import BufferedByteReceiveStream @@ -483,7 +485,37 @@ async def test_close_with_stdout_blocked_subprocess(anyio_backend_name: str) -> with fail_after(5): await process.aclose() except TimeoutError: - if anyio_backend_name == "asyncio": - process._process._transport.close() # type: ignore[attr-defined] - + # Force the process down so it doesn't leak, then fail the test + process.kill() pytest.fail("Process.aclose() deadlocked") + + +async def test_returncode_polls_after_exit() -> None: + """ + ``Process.returncode`` should reflect the real state once the process exits, even + if ``wait()`` was never called, consistently across backends (see discussion #828). + """ + process = await open_process([sys.executable, "-c", ""]) + try: + with fail_after(5): + # Deliberately poll returncode (that's what's under test here) + while process.returncode is None: # noqa: ASYNC110 + await sleep(0.01) + finally: + await process.aclose() + + assert process.returncode == 0 + + +async def test_signal_already_exited_process() -> None: + """ + Signalling an already-exited process must be a no-op rather than raising, on every + backend (see discussion #828). + """ + process = await open_process([sys.executable, "-c", ""]) + async with process: + await process.wait() + # None of these should raise ProcessLookupError or similar + process.terminate() + process.kill() + process.send_signal(signal.SIGTERM) From de574f80ed24d9dd8b64bf0db02ab15ef28569e3 Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sun, 19 Jul 2026 14:05:11 +0100 Subject: [PATCH 02/15] Fix subprocess CI failures on macOS, Windows and pyright - macOS: os.waitid is unavailable before Python 3.13, so reap via a portable helper that uses waitid(WNOWAIT) where available and falls back to Popen.wait() otherwise (pidfd is still used on Linux) - pyright: annotate wait_for_child_exit's parameter as subprocess.Popen[bytes] so AsyncBackend (and everything referencing it, e.g. EventLoopToken) stays type-complete - Windows: use a duplex pipe for the child's stdin, since asyncio's write-pipe transport reads our end to detect closure and therefore needs read access Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01LZMpdx7hNR89fwdmiMFmVJ --- src/anyio/_backends/_asyncio.py | 8 ++++-- src/anyio/_core/_subprocesses.py | 45 +++++++++++++++++++------------- src/anyio/abc/_eventloop.py | 8 +++--- 3 files changed, 37 insertions(+), 24 deletions(-) diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py index 59635ca12..f4d053790 100644 --- a/src/anyio/_backends/_asyncio.py +++ b/src/anyio/_backends/_asyncio.py @@ -2741,8 +2741,12 @@ async def create_subprocess_stdin_pipe(cls) -> tuple[abc.ByteSendStream, int]: from asyncio.windows_utils import PipeHandle from asyncio.windows_utils import pipe as windows_pipe - # The write end (our end) uses overlapped (IOCP) I/O - read_handle, write_handle = windows_pipe(overlapped=(False, True)) + # A duplex pipe is required: asyncio's write-pipe transport issues a read on + # our end to detect when the child closes its side, so our (write) end needs + # read access too. The write end also uses overlapped (IOCP) I/O. + read_handle, write_handle = windows_pipe( + duplex=True, overlapped=(False, True) + ) pipe_obj: Any = PipeHandle(write_handle) child_fd = msvcrt.open_osfhandle(read_handle, os.O_RDONLY) else: diff --git a/src/anyio/_core/_subprocesses.py b/src/anyio/_core/_subprocesses.py index ec9473a2c..89477bea0 100644 --- a/src/anyio/_core/_subprocesses.py +++ b/src/anyio/_core/_subprocesses.py @@ -34,28 +34,37 @@ def _get_child_reaper_limiter() -> CapacityLimiter: return limiter -def _sync_wait_reapable(pid: int) -> None: - """Block until the process ``pid`` is reapable, without consuming its exit status.""" - while True: - try: - os.waitid(os.P_PID, pid, os.WEXITED | os.WNOWAIT) - except ChildProcessError: - # Already reaped elsewhere - return - except InterruptedError: - continue - else: - return +def _sync_wait_for_exit(process: subprocess.Popen[bytes]) -> None: + """ + Block (in a worker thread) until ``process`` has exited. + + Where ``os.waitid()`` is available (Linux, and macOS on Python 3.13+), this uses + ``WNOWAIT`` so the exit status is left intact for :meth:`subprocess.Popen.wait`. + Otherwise it falls back to :meth:`subprocess.Popen.wait`, which reaps the process + directly (that's fine, since the shared ``Process.wait()`` calls it again). + """ + if hasattr(os, "waitid"): + while True: + try: + os.waitid(os.P_PID, process.pid, os.WEXITED | os.WNOWAIT) + except InterruptedError: + continue + except ChildProcessError: + # Already reaped elsewhere + return + else: + return + + process.wait() -async def wait_for_child_exit(process: subprocess.Popen) -> None: +async def wait_for_child_exit(process: subprocess.Popen[bytes]) -> None: """ Backend-agnostic POSIX implementation of :meth:`~anyio.abc.AsyncBackend.wait_for_child_exit`. - On Linux this waits on a pidfd (no worker thread required); on other POSIX systems it - falls back to a ``waitid()`` call in a worker thread. Either way the child's exit - status is left intact so that :meth:`subprocess.Popen.wait` can consume it. + On Linux this waits on a pidfd, without tying up a worker thread. On other POSIX + systems it waits in a worker thread (see :func:`_sync_wait_for_exit`). """ backend = get_async_backend() if sys.platform == "linux": @@ -72,8 +81,8 @@ async def wait_for_child_exit(process: subprocess.Popen) -> None: return await backend.run_sync_in_worker_thread( - _sync_wait_reapable, - (process.pid,), + _sync_wait_for_exit, + (process,), abandon_on_cancel=True, limiter=_get_child_reaper_limiter(), ) diff --git a/src/anyio/abc/_eventloop.py b/src/anyio/abc/_eventloop.py index f913ccaff..dea56ea45 100644 --- a/src/anyio/abc/_eventloop.py +++ b/src/anyio/abc/_eventloop.py @@ -267,13 +267,13 @@ async def create_subprocess_output_pipe(cls) -> tuple[ByteReceiveStream, int]: @classmethod @abstractmethod - async def wait_for_child_exit(cls, process: subprocess.Popen) -> None: + async def wait_for_child_exit(cls, process: subprocess.Popen[bytes]) -> None: """ - Wait until the given child process is ready to be reaped. + Wait until the given child process has exited. When this returns, a call to :meth:`subprocess.Popen.wait` is guaranteed to - return the exit status immediately. The exit status itself must **not** be - consumed by this call. + return the exit status immediately (the implementation may reap the process + itself, as long as it does so via ``process``). """ @classmethod From 0b1f034f4b224a7f6e2a0759e9c4b0c3f95138cf Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sun, 19 Jul 2026 14:19:29 +0100 Subject: [PATCH 03/15] Make Windows subprocess waiting loop-agnostic; support winloop pipes - Wait for child exit on Windows via RegisterWaitForSingleObject (a Windows thread-pool wait delivered through call_soon_threadsafe), which works on the stdlib ProactorEventLoop, winloop and SelectorEventLoop alike and doesn't tie up a Python thread. Replaces the ProactorEventLoop-only _proactor.wait_for_handle. - Create the subprocess pipes appropriately for the running loop: the stdlib ProactorEventLoop takes a PipeHandle, while winloop (libuv) wants a file object backed by a real overlapped file descriptor, like uvloop on POSIX. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01LZMpdx7hNR89fwdmiMFmVJ --- src/anyio/_backends/_asyncio.py | 41 ++++++--- src/anyio/_core/_asyncio_windows_process.py | 97 +++++++++++++++++++++ 2 files changed, 126 insertions(+), 12 deletions(-) create mode 100644 src/anyio/_core/_asyncio_windows_process.py diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py index f4d053790..e0cae31eb 100644 --- a/src/anyio/_backends/_asyncio.py +++ b/src/anyio/_backends/_asyncio.py @@ -2741,13 +2741,20 @@ async def create_subprocess_stdin_pipe(cls) -> tuple[abc.ByteSendStream, int]: from asyncio.windows_utils import PipeHandle from asyncio.windows_utils import pipe as windows_pipe - # A duplex pipe is required: asyncio's write-pipe transport issues a read on - # our end to detect when the child closes its side, so our (write) end needs - # read access too. The write end also uses overlapped (IOCP) I/O. - read_handle, write_handle = windows_pipe( - duplex=True, overlapped=(False, True) - ) - pipe_obj: Any = PipeHandle(write_handle) + if hasattr(loop, "_proactor"): + # stdlib ProactorEventLoop: its write-pipe transport issues a read on + # our end to detect when the child closes its side, so a duplex pipe is + # required. Our (write) end uses overlapped (IOCP) I/O. + read_handle, write_handle = windows_pipe( + duplex=True, overlapped=(False, True) + ) + pipe_obj: Any = PipeHandle(write_handle) + else: + # winloop (libuv) works like uvloop on POSIX: it wants a file object + # backed by a real (overlapped) file descriptor + read_handle, write_handle = windows_pipe(overlapped=(False, True)) + pipe_obj = os.fdopen(msvcrt.open_osfhandle(write_handle, 0), "wb", 0) + child_fd = msvcrt.open_osfhandle(read_handle, os.O_RDONLY) else: read_fd, write_fd = os.pipe() @@ -2774,7 +2781,15 @@ async def create_subprocess_output_pipe(cls) -> tuple[abc.ByteReceiveStream, int # The read end (our end) uses overlapped (IOCP) I/O read_handle, write_handle = windows_pipe(overlapped=(True, False)) - pipe_obj: Any = PipeHandle(read_handle) + if hasattr(loop, "_proactor"): + # stdlib ProactorEventLoop wants the raw pipe handle + pipe_obj: Any = PipeHandle(read_handle) + else: + # winloop (libuv) wants a file object backed by a real fd + pipe_obj = os.fdopen( + msvcrt.open_osfhandle(read_handle, os.O_RDONLY), "rb", 0 + ) + child_fd = msvcrt.open_osfhandle(write_handle, 0) else: read_fd, write_fd = os.pipe() @@ -2792,11 +2807,13 @@ async def create_subprocess_output_pipe(cls) -> tuple[abc.ByteReceiveStream, int return _ProcessReceivePipeStream(transport, protocol), child_fd @classmethod - async def wait_for_child_exit(cls, process: subprocess.Popen) -> None: + async def wait_for_child_exit(cls, process: subprocess.Popen[bytes]) -> None: if sys.platform == "win32": - loop = asyncio.get_running_loop() - # Wait on the process handle via IOCP; no worker thread involved - await loop._proactor.wait_for_handle(int(process._handle)) # type: ignore[attr-defined] + # Works regardless of the event loop implementation (ProactorEventLoop, + # winloop, SelectorEventLoop) and doesn't tie up a Python thread + from .._core._asyncio_windows_process import wait_for_pid + + await wait_for_pid(process.pid) else: from .._core._subprocesses import wait_for_child_exit diff --git a/src/anyio/_core/_asyncio_windows_process.py b/src/anyio/_core/_asyncio_windows_process.py new file mode 100644 index 000000000..6872fe848 --- /dev/null +++ b/src/anyio/_core/_asyncio_windows_process.py @@ -0,0 +1,97 @@ +""" +Waiting for a child process to exit on Windows, independently of the event loop +implementation in use (stdlib ``ProactorEventLoop``, winloop, ``SelectorEventLoop`` …). + +This uses ``RegisterWaitForSingleObject``, which signals completion from a Windows thread +pool thread (i.e. without tying up a Python thread), and wakes the running event loop via +:meth:`~asyncio.loop.call_soon_threadsafe`. +""" + +from __future__ import annotations + +import asyncio +import ctypes +import sys +from ctypes import wintypes +from typing import TYPE_CHECKING + +# This module is only ever imported on Windows; the assertion lets type checkers on other +# platforms skip it (mirrors the pattern used by Trio's Windows-only modules). +assert sys.platform == "win32" or not TYPE_CHECKING + +kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + +SYNCHRONIZE = 0x00100000 +PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 +INFINITE = 0xFFFFFFFF +WT_EXECUTEONLYONCE = 0x00000008 + +_WAIT_CALLBACK = ctypes.WINFUNCTYPE(None, wintypes.LPVOID, wintypes.BOOLEAN) + +kernel32.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD] +kernel32.OpenProcess.restype = wintypes.HANDLE + +kernel32.RegisterWaitForSingleObject.argtypes = [ + ctypes.POINTER(wintypes.HANDLE), + wintypes.HANDLE, + _WAIT_CALLBACK, + wintypes.LPVOID, + wintypes.DWORD, + wintypes.DWORD, +] +kernel32.RegisterWaitForSingleObject.restype = wintypes.BOOL + +kernel32.UnregisterWait.argtypes = [wintypes.HANDLE] +kernel32.UnregisterWait.restype = wintypes.BOOL + +kernel32.CloseHandle.argtypes = [wintypes.HANDLE] +kernel32.CloseHandle.restype = wintypes.BOOL + + +class _ProcessWaiter: + def __init__(self, loop: asyncio.AbstractEventLoop, process_handle: int) -> None: + self._loop = loop + self._process_handle = process_handle + self._wait_handle = wintypes.HANDLE() + self._future: asyncio.Future[None] = loop.create_future() + # Keep the ctypes callback alive for as long as Windows may call it + self._callback = _WAIT_CALLBACK(self._on_signalled) + + def start(self) -> asyncio.Future[None]: + if not kernel32.RegisterWaitForSingleObject( + ctypes.byref(self._wait_handle), + self._process_handle, + self._callback, + None, + INFINITE, + WT_EXECUTEONLYONCE, + ): + raise ctypes.WinError(ctypes.get_last_error()) + + return self._future + + def _on_signalled(self, context: object, timed_out: bool) -> None: + # This runs on a Windows thread pool thread + self._loop.call_soon_threadsafe(self._complete) + + def _complete(self) -> None: + if not self._future.done(): + self._future.set_result(None) + + +async def wait_for_pid(pid: int) -> None: + """Wait until the process with the given PID has exited.""" + process_handle = kernel32.OpenProcess( + SYNCHRONIZE | PROCESS_QUERY_LIMITED_INFORMATION, False, pid + ) + if not process_handle: + raise ctypes.WinError(ctypes.get_last_error()) + + waiter = _ProcessWaiter(asyncio.get_running_loop(), process_handle) + try: + await waiter.start() + finally: + if waiter._wait_handle: + kernel32.UnregisterWait(waiter._wait_handle) + + kernel32.CloseHandle(process_handle) From 56da9676c89691b7b14a264ce4fd46143dbd3efd Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sun, 19 Jul 2026 14:21:45 +0100 Subject: [PATCH 04/15] Guard os.pidfd_open with hasattr for PyPy PyPy (and other interpreters/older kernels) may not provide os.pidfd_open; accessing it raised AttributeError rather than OSError, so it wasn't caught. Fall back to the worker-thread reaping path when it's unavailable. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01LZMpdx7hNR89fwdmiMFmVJ --- src/anyio/_core/_subprocesses.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/anyio/_core/_subprocesses.py b/src/anyio/_core/_subprocesses.py index 89477bea0..157d66f13 100644 --- a/src/anyio/_core/_subprocesses.py +++ b/src/anyio/_core/_subprocesses.py @@ -67,7 +67,7 @@ async def wait_for_child_exit(process: subprocess.Popen[bytes]) -> None: systems it waits in a worker thread (see :func:`_sync_wait_for_exit`). """ backend = get_async_backend() - if sys.platform == "linux": + if sys.platform == "linux" and hasattr(os, "pidfd_open"): try: pidfd = os.pidfd_open(process.pid) except OSError: From 504ae2978d1f31d61dad57f9b827a12efdc51e6c Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sun, 19 Jul 2026 14:26:48 +0100 Subject: [PATCH 05/15] Prefer proactor.wait_for_handle for child wait on ProactorEventLoop Use the native IOCP handle wait on the stdlib ProactorEventLoop, falling back to the RegisterWaitForSingleObject helper only on loops without a proactor (winloop, SelectorEventLoop). wait_for_pid only needs call_soon_threadsafe, so it works on any of them. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01LZMpdx7hNR89fwdmiMFmVJ --- src/anyio/_backends/_asyncio.py | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py index e0cae31eb..7d0471458 100644 --- a/src/anyio/_backends/_asyncio.py +++ b/src/anyio/_backends/_asyncio.py @@ -2809,11 +2809,17 @@ async def create_subprocess_output_pipe(cls) -> tuple[abc.ByteReceiveStream, int @classmethod async def wait_for_child_exit(cls, process: subprocess.Popen[bytes]) -> None: if sys.platform == "win32": - # Works regardless of the event loop implementation (ProactorEventLoop, - # winloop, SelectorEventLoop) and doesn't tie up a Python thread - from .._core._asyncio_windows_process import wait_for_pid + loop = asyncio.get_running_loop() + proactor = getattr(loop, "_proactor", None) + if proactor is not None: + # Native IOCP handle wait on the stdlib ProactorEventLoop + await proactor.wait_for_handle(int(process._handle)) # type: ignore[attr-defined] + else: + # winloop / SelectorEventLoop: use a Windows thread-pool wait, delivered + # through call_soon_threadsafe (no Python thread tied up) + from .._core._asyncio_windows_process import wait_for_pid - await wait_for_pid(process.pid) + await wait_for_pid(process.pid) else: from .._core._subprocesses import wait_for_child_exit From 7c43044e6667d004f41778c9c214275e375aeef9 Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sun, 19 Jul 2026 14:29:20 +0100 Subject: [PATCH 06/15] Use UnregisterWaitEx to avoid a callback-after-cleanup race UnregisterWait() may return while a RegisterWaitForSingleObject callback is still in flight, so on cancellation the callback could fire after the process handle was closed and the ctypes trampoline collected. UnregisterWaitEx() with INVALID_HANDLE_VALUE blocks until in-flight callbacks finish, making cleanup safe. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01LZMpdx7hNR89fwdmiMFmVJ --- src/anyio/_core/_asyncio_windows_process.py | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/src/anyio/_core/_asyncio_windows_process.py b/src/anyio/_core/_asyncio_windows_process.py index 6872fe848..457e51e18 100644 --- a/src/anyio/_core/_asyncio_windows_process.py +++ b/src/anyio/_core/_asyncio_windows_process.py @@ -25,6 +25,8 @@ PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 INFINITE = 0xFFFFFFFF WT_EXECUTEONLYONCE = 0x00000008 +# Passing this to UnregisterWaitEx() blocks until any in-flight callback has finished +INVALID_HANDLE_VALUE = wintypes.HANDLE(-1) _WAIT_CALLBACK = ctypes.WINFUNCTYPE(None, wintypes.LPVOID, wintypes.BOOLEAN) @@ -41,8 +43,8 @@ ] kernel32.RegisterWaitForSingleObject.restype = wintypes.BOOL -kernel32.UnregisterWait.argtypes = [wintypes.HANDLE] -kernel32.UnregisterWait.restype = wintypes.BOOL +kernel32.UnregisterWaitEx.argtypes = [wintypes.HANDLE, wintypes.HANDLE] +kernel32.UnregisterWaitEx.restype = wintypes.BOOL kernel32.CloseHandle.argtypes = [wintypes.HANDLE] kernel32.CloseHandle.restype = wintypes.BOOL @@ -91,7 +93,11 @@ async def wait_for_pid(pid: int) -> None: try: await waiter.start() finally: + # UnregisterWaitEx() with INVALID_HANDLE_VALUE blocks until any callback already + # in flight has finished, so it's safe to release the process handle and let the + # ctypes callback be collected afterwards (avoids a callback-after-cleanup race on + # cancellation). if waiter._wait_handle: - kernel32.UnregisterWait(waiter._wait_handle) + kernel32.UnregisterWaitEx(waiter._wait_handle, INVALID_HANDLE_VALUE) kernel32.CloseHandle(process_handle) From 2cbdf5dd20ed76cbc4be92e42356ad367cfca2a5 Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sun, 19 Jul 2026 14:36:52 +0100 Subject: [PATCH 07/15] Support subprocesses on the Windows SelectorEventLoop The SelectorEventLoop can't do overlapped pipe I/O, so run the subprocess pipes on a background ProactorEventLoop (or winloop) thread, mirroring the existing selector thread helper (started lazily, shut down via threading._register_atexit). Pipe operations are marshalled onto that loop with run_coroutine_threadsafe and awaited on the caller's loop via wrap_future. Waiting for the child still uses the loop-agnostic RegisterWaitForSingleObject helper. A Windows SelectorEventLoop entry is added to the test matrix. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01LZMpdx7hNR89fwdmiMFmVJ --- src/anyio/_backends/_asyncio.py | 24 +++++ src/anyio/_core/_asyncio_proactor_thread.py | 108 ++++++++++++++++++++ tests/conftest.py | 10 ++ 3 files changed, 142 insertions(+) create mode 100644 src/anyio/_core/_asyncio_proactor_thread.py diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py index 7d0471458..7d4c17e70 100644 --- a/src/anyio/_backends/_asyncio.py +++ b/src/anyio/_backends/_asyncio.py @@ -2736,6 +2736,18 @@ async def open_process( @classmethod async def create_subprocess_stdin_pipe(cls) -> tuple[abc.ByteSendStream, int]: loop = asyncio.get_running_loop() + if sys.platform == "win32" and isinstance(loop, asyncio.SelectorEventLoop): + # The SelectorEventLoop can't do overlapped pipe I/O, so run it on a + # background proactor loop and proxy the operations to it + from .._core._asyncio_proactor_thread import ( + ProxySendStream, + get_proactor_thread, + ) + + thread = get_proactor_thread() + inner, child_fd = await thread.run(cls.create_subprocess_stdin_pipe()) + return ProxySendStream(thread, inner), child_fd + if sys.platform == "win32": import msvcrt from asyncio.windows_utils import PipeHandle @@ -2774,6 +2786,18 @@ async def create_subprocess_stdin_pipe(cls) -> tuple[abc.ByteSendStream, int]: @classmethod async def create_subprocess_output_pipe(cls) -> tuple[abc.ByteReceiveStream, int]: loop = asyncio.get_running_loop() + if sys.platform == "win32" and isinstance(loop, asyncio.SelectorEventLoop): + # The SelectorEventLoop can't do overlapped pipe I/O, so run it on a + # background proactor loop and proxy the operations to it + from .._core._asyncio_proactor_thread import ( + ProxyReceiveStream, + get_proactor_thread, + ) + + thread = get_proactor_thread() + inner, child_fd = await thread.run(cls.create_subprocess_output_pipe()) + return ProxyReceiveStream(thread, inner), child_fd + if sys.platform == "win32": import msvcrt from asyncio.windows_utils import PipeHandle diff --git a/src/anyio/_core/_asyncio_proactor_thread.py b/src/anyio/_core/_asyncio_proactor_thread.py new file mode 100644 index 000000000..e187645e6 --- /dev/null +++ b/src/anyio/_core/_asyncio_proactor_thread.py @@ -0,0 +1,108 @@ +""" +A background event loop capable of overlapped (IOCP) pipe I/O, for use when the main +event loop can't do it itself (the Windows ``SelectorEventLoop``). + +This mirrors :mod:`anyio._core._asyncio_selector_thread`: a single background thread runs +a ``ProactorEventLoop`` (or winloop) for the lifetime of the interpreter (shut down via +``threading._register_atexit``). Subprocess pipe operations are marshalled onto it with +:func:`asyncio.run_coroutine_threadsafe` and awaited on the caller's loop via +:func:`asyncio.wrap_future`. +""" + +from __future__ import annotations + +import asyncio +import sys +import threading +from collections.abc import Coroutine +from typing import TYPE_CHECKING, Any, TypeVar + +from ..abc import ByteReceiveStream, ByteSendStream + +assert sys.platform == "win32" or not TYPE_CHECKING + +T = TypeVar("T") + +_proactor_thread_lock = threading.Lock() +_proactor_thread: ProactorThread | None = None + + +def _new_proactor_loop() -> asyncio.AbstractEventLoop: + try: + import winloop + except ImportError: + return asyncio.ProactorEventLoop() + else: + return winloop.new_event_loop() + + +class ProactorThread: + def __init__(self) -> None: + self._loop = _new_proactor_loop() + self._thread = threading.Thread( + target=self._run, name="AnyIO proactor", daemon=True + ) + self._started = threading.Event() + + def _run(self) -> None: + asyncio.set_event_loop(self._loop) + self._loop.call_soon(self._started.set) + try: + self._loop.run_forever() + finally: + self._loop.close() + + def start(self) -> None: + self._thread.start() + self._started.wait() + threading._register_atexit(self._stop) # type: ignore[attr-defined] + + def _stop(self) -> None: + global _proactor_thread + self._loop.call_soon_threadsafe(self._loop.stop) + self._thread.join() + _proactor_thread = None + + async def run(self, coro: Coroutine[Any, Any, T]) -> T: + """Run ``coro`` on the proactor loop, awaiting the result on the caller's loop.""" + future = asyncio.run_coroutine_threadsafe(coro, self._loop) + return await asyncio.wrap_future(future, loop=asyncio.get_running_loop()) + + +class ProxyReceiveStream(ByteReceiveStream): + """Forwards receive/aclose to a stream living on the proactor thread's loop.""" + + def __init__(self, thread: ProactorThread, inner: ByteReceiveStream) -> None: + self._thread = thread + self._inner = inner + + async def receive(self, max_bytes: int = 65536) -> bytes: + return await self._thread.run(self._inner.receive(max_bytes)) + + async def aclose(self) -> None: + await self._thread.run(self._inner.aclose()) + + +class ProxySendStream(ByteSendStream): + """Forwards send/aclose to a stream living on the proactor thread's loop.""" + + def __init__(self, thread: ProactorThread, inner: ByteSendStream) -> None: + self._thread = thread + self._inner = inner + + async def send(self, item: bytes) -> None: + await self._thread.run(self._inner.send(item)) + + async def aclose(self) -> None: + await self._thread.run(self._inner.aclose()) + + +def get_proactor_thread() -> ProactorThread: + global _proactor_thread + + with _proactor_thread_lock: + if _proactor_thread is None: + _proactor_thread = ProactorThread() + _proactor_thread.start() + + return _proactor_thread diff --git a/tests/conftest.py b/tests/conftest.py index 5732caf2c..f17dba9c4 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -72,6 +72,16 @@ def eager_task_loop_factory() -> asyncio.AbstractEventLoop: ), ) +if platform.system() == "Windows": + # The SelectorEventLoop can't do overlapped pipe I/O itself; exercise the fallback + # that runs subprocess pipes on a background proactor loop + asyncio_params.append( + pytest.param( + ("asyncio", {"debug": True, "loop_factory": asyncio.SelectorEventLoop}), + id="asyncio+selector", + ), + ) + backend_params = asyncio_params.copy() available_backends = set(get_available_backends()) for backend_name in get_all_backends(): From 91f7bf8ee2b8f1383686cec20dbdfd8867cf0895 Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sun, 19 Jul 2026 15:06:10 +0100 Subject: [PATCH 08/15] Extract asyncio.Runner backport; use it for the proactor thread - Move the Python 3.10 asyncio.Runner backport out of _backends/_asyncio.py into a standalone _core/_asyncio_runner.py (re-exporting the stdlib Runner on 3.11+). - Drive the proactor thread's loop with asyncio.Runner (loop_factory) and a stop-event serve coroutine, so leftover tasks, async generators and the default executor are cleaned up and the loop is closed properly on shutdown. - Give the subprocess pipe streams (and their SelectorEventLoop proxies) a synchronous _abort() so the process pool's forced shutdown can tear down worker transports at loop completion; the proxy schedules the close on the proactor loop. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01LZMpdx7hNR89fwdmiMFmVJ --- src/anyio/_backends/_asyncio.py | 196 ++----------------- src/anyio/_core/_asyncio_proactor_thread.py | 31 ++- src/anyio/_core/_asyncio_runner.py | 200 ++++++++++++++++++++ 3 files changed, 235 insertions(+), 192 deletions(-) create mode 100644 src/anyio/_core/_asyncio_runner.py diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py index 7d4c17e70..5f0fe3c76 100644 --- a/src/anyio/_backends/_asyncio.py +++ b/src/anyio/_backends/_asyncio.py @@ -9,7 +9,6 @@ import socket import subprocess import sys -import threading import weakref from asyncio import ( AbstractEventLoop, @@ -110,193 +109,12 @@ FileDescriptorLike = object if sys.version_info >= (3, 11): - from asyncio import Runner from typing import TypeVarTuple, Unpack else: - import contextvars - import enum - import signal - from asyncio import coroutines, events, exceptions, tasks - from exceptiongroup import BaseExceptionGroup from typing_extensions import TypeVarTuple, Unpack - class _State(enum.Enum): - CREATED = "created" - INITIALIZED = "initialized" - CLOSED = "closed" - - class Runner: - # Copied from CPython 3.11 - def __init__( - self, - *, - debug: bool | None = None, - loop_factory: Callable[[], AbstractEventLoop] | None = None, - ): - self._state = _State.CREATED - self._debug = debug - self._loop_factory = loop_factory - self._loop: AbstractEventLoop | None = None - self._context = None - self._interrupt_count = 0 - self._set_event_loop = False - - def __enter__(self) -> Runner: - self._lazy_init() - return self - - def __exit__( - self, - exc_type: type[BaseException] | None, - exc_val: BaseException | None, - exc_tb: TracebackType | None, - ) -> None: - self.close() - - def close(self) -> None: - """Shutdown and close event loop.""" - loop = self._loop - if self._state is not _State.INITIALIZED or loop is None: - return - try: - _cancel_all_tasks(loop) - loop.run_until_complete(loop.shutdown_asyncgens()) - if hasattr(loop, "shutdown_default_executor"): - loop.run_until_complete(loop.shutdown_default_executor()) - else: - loop.run_until_complete(_shutdown_default_executor(loop)) - finally: - if self._set_event_loop: - events.set_event_loop(None) - loop.close() - self._loop = None - self._state = _State.CLOSED - - def get_loop(self) -> AbstractEventLoop: - """Return embedded event loop.""" - self._lazy_init() - return self._loop - - def run(self, coro: Coroutine[T_Retval], *, context=None) -> T_Retval: - """Run a coroutine inside the embedded event loop.""" - if not coroutines.iscoroutine(coro): - raise ValueError(f"a coroutine was expected, got {coro!r}") - - if events._get_running_loop() is not None: - # fail fast with short traceback - raise RuntimeError( - "Runner.run() cannot be called from a running event loop" - ) - - self._lazy_init() - - if context is None: - context = self._context - task = context.run(self._loop.create_task, coro) - - if ( - threading.current_thread() is threading.main_thread() - and signal.getsignal(signal.SIGINT) is signal.default_int_handler - ): - sigint_handler = partial(self._on_sigint, main_task=task) - try: - signal.signal(signal.SIGINT, sigint_handler) - except ValueError: - # `signal.signal` may throw if `threading.main_thread` does - # not support signals (e.g. embedded interpreter with signals - # not registered - see gh-91880) - sigint_handler = None - else: - sigint_handler = None - - self._interrupt_count = 0 - try: - return self._loop.run_until_complete(task) - except exceptions.CancelledError: - if self._interrupt_count > 0: - uncancel = getattr(task, "uncancel", None) - if uncancel is not None and uncancel() == 0: - raise KeyboardInterrupt # noqa: B904 - raise # CancelledError - finally: - if ( - sigint_handler is not None - and signal.getsignal(signal.SIGINT) is sigint_handler - ): - signal.signal(signal.SIGINT, signal.default_int_handler) - - def _lazy_init(self) -> None: - if self._state is _State.CLOSED: - raise RuntimeError("Runner is closed") - if self._state is _State.INITIALIZED: - return - if self._loop_factory is None: - self._loop = events.new_event_loop() - if not self._set_event_loop: - # Call set_event_loop only once to avoid calling - # attach_loop multiple times on child watchers - events.set_event_loop(self._loop) - self._set_event_loop = True - else: - self._loop = self._loop_factory() - if self._debug is not None: - self._loop.set_debug(self._debug) - self._context = contextvars.copy_context() - self._state = _State.INITIALIZED - - def _on_sigint(self, signum, frame, main_task: asyncio.Task) -> None: - self._interrupt_count += 1 - if self._interrupt_count == 1 and not main_task.done(): - main_task.cancel() - # wakeup loop if it is blocked by select() with long timeout - self._loop.call_soon_threadsafe(lambda: None) - return - raise KeyboardInterrupt() - - def _cancel_all_tasks(loop: AbstractEventLoop) -> None: - to_cancel = tasks.all_tasks(loop) - if not to_cancel: - return - - for task in to_cancel: - task.cancel() - - loop.run_until_complete(tasks.gather(*to_cancel, return_exceptions=True)) - - for task in to_cancel: - if task.cancelled(): - continue - if task.exception() is not None: - loop.call_exception_handler( - { - "message": "unhandled exception during asyncio.run() shutdown", - "exception": task.exception(), - "task": task, - } - ) - - async def _shutdown_default_executor(loop: AbstractEventLoop) -> None: - """Schedule the shutdown of the default executor.""" - - def _do_shutdown(future: asyncio.futures.Future) -> None: - try: - loop._default_executor.shutdown(wait=True) # type: ignore[attr-defined] - loop.call_soon_threadsafe(future.set_result, None) - except Exception as ex: - loop.call_soon_threadsafe(future.set_exception, ex) - - loop._executor_shutdown_called = True - if loop._default_executor is None: - return - future = loop.create_future() - thread = threading.Thread(target=_do_shutdown, args=(future,)) - thread.start() - try: - await future - finally: - thread.join() - +from .._core._asyncio_runner import Runner T_Retval = TypeVar("T_Retval") T_co = TypeVar("T_co", covariant=True) @@ -1179,6 +997,11 @@ async def aclose(self) -> None: await AsyncIOBackend.checkpoint() + def _abort(self) -> None: + self._closed = True + if not self._transport.is_closing(): + self._transport.close() + class _ProcessSendPipeStream(abc.ByteSendStream): def __init__( @@ -1217,6 +1040,11 @@ async def aclose(self) -> None: await AsyncIOBackend.checkpoint() + def _abort(self) -> None: + self._closed = True + if not self._transport.is_closing(): + self._transport.close() + def _forcibly_shutdown_process_pool_on_exit( workers: set[abc.Process], _task: object @@ -1229,7 +1057,7 @@ def _forcibly_shutdown_process_pool_on_exit( for stream in (process.stdin, process.stdout, process.stderr): if stream is not None: - stream._transport.close() # type: ignore[union-attr] + stream._abort() # type: ignore[union-attr] process.kill() diff --git a/src/anyio/_core/_asyncio_proactor_thread.py b/src/anyio/_core/_asyncio_proactor_thread.py index e187645e6..db6770996 100644 --- a/src/anyio/_core/_asyncio_proactor_thread.py +++ b/src/anyio/_core/_asyncio_proactor_thread.py @@ -18,6 +18,7 @@ from typing import TYPE_CHECKING, Any, TypeVar from ..abc import ByteReceiveStream, ByteSendStream +from ._asyncio_runner import Runner assert sys.platform == "win32" or not TYPE_CHECKING @@ -38,19 +39,25 @@ def _new_proactor_loop() -> asyncio.AbstractEventLoop: class ProactorThread: def __init__(self) -> None: - self._loop = _new_proactor_loop() + self._loop: asyncio.AbstractEventLoop + self._stop_event: asyncio.Event self._thread = threading.Thread( target=self._run, name="AnyIO proactor", daemon=True ) self._started = threading.Event() + async def _serve(self) -> None: + # Runs on the proactor loop; keeps it alive until stop is requested + self._stop_event = asyncio.Event() + self._started.set() + await self._stop_event.wait() + def _run(self) -> None: - asyncio.set_event_loop(self._loop) - self._loop.call_soon(self._started.set) - try: - self._loop.run_forever() - finally: - self._loop.close() + # asyncio.Runner takes care of cancelling leftover tasks, shutting down async + # generators and the default executor, and closing the loop + with Runner(loop_factory=_new_proactor_loop) as runner: + self._loop = runner.get_loop() + runner.run(self._serve()) def start(self) -> None: self._thread.start() @@ -59,7 +66,7 @@ def start(self) -> None: def _stop(self) -> None: global _proactor_thread - self._loop.call_soon_threadsafe(self._loop.stop) + self._loop.call_soon_threadsafe(self._stop_event.set) self._thread.join() _proactor_thread = None @@ -82,6 +89,10 @@ async def receive(self, max_bytes: int = 65536) -> bytes: async def aclose(self) -> None: await self._thread.run(self._inner.aclose()) + def _abort(self) -> None: + # Synchronously (best-effort) close the underlying transport on the proactor loop + self._thread._loop.call_soon_threadsafe(self._inner._abort) # type: ignore[attr-defined] + class ProxySendStream(ByteSendStream): """Forwards send/aclose to a stream living on the proactor thread's loop.""" @@ -96,6 +107,10 @@ async def send(self, item: bytes) -> None: async def aclose(self) -> None: await self._thread.run(self._inner.aclose()) + def _abort(self) -> None: + # Synchronously (best-effort) close the underlying transport on the proactor loop + self._thread._loop.call_soon_threadsafe(self._inner._abort) # type: ignore[attr-defined] + def get_proactor_thread() -> ProactorThread: global _proactor_thread diff --git a/src/anyio/_core/_asyncio_runner.py b/src/anyio/_core/_asyncio_runner.py new file mode 100644 index 000000000..6c1019010 --- /dev/null +++ b/src/anyio/_core/_asyncio_runner.py @@ -0,0 +1,200 @@ +""" +A standalone copy of :class:`asyncio.Runner` (added in Python 3.11), backported for +Python 3.10. On 3.11+ this simply re-exports the stdlib implementation. +""" + +from __future__ import annotations + +import sys + +if sys.version_info >= (3, 11): + from asyncio import Runner as Runner +else: + import asyncio + import contextvars + import enum + import signal + import threading + from asyncio import AbstractEventLoop, coroutines, events, exceptions, tasks + from collections.abc import Callable, Coroutine + from functools import partial + from types import TracebackType + from typing import TypeVar + + T_Retval = TypeVar("T_Retval") + + class _State(enum.Enum): + CREATED = "created" + INITIALIZED = "initialized" + CLOSED = "closed" + + class Runner: + # Copied from CPython 3.11 + def __init__( + self, + *, + debug: bool | None = None, + loop_factory: Callable[[], AbstractEventLoop] | None = None, + ): + self._state = _State.CREATED + self._debug = debug + self._loop_factory = loop_factory + self._loop: AbstractEventLoop | None = None + self._context = None + self._interrupt_count = 0 + self._set_event_loop = False + + def __enter__(self) -> Runner: + self._lazy_init() + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: TracebackType | None, + ) -> None: + self.close() + + def close(self) -> None: + """Shutdown and close event loop.""" + loop = self._loop + if self._state is not _State.INITIALIZED or loop is None: + return + try: + _cancel_all_tasks(loop) + loop.run_until_complete(loop.shutdown_asyncgens()) + if hasattr(loop, "shutdown_default_executor"): + loop.run_until_complete(loop.shutdown_default_executor()) + else: + loop.run_until_complete(_shutdown_default_executor(loop)) + finally: + if self._set_event_loop: + events.set_event_loop(None) + loop.close() + self._loop = None + self._state = _State.CLOSED + + def get_loop(self) -> AbstractEventLoop: + """Return embedded event loop.""" + self._lazy_init() + return self._loop + + def run(self, coro: Coroutine[T_Retval], *, context=None) -> T_Retval: + """Run a coroutine inside the embedded event loop.""" + if not coroutines.iscoroutine(coro): + raise ValueError(f"a coroutine was expected, got {coro!r}") + + if events._get_running_loop() is not None: + # fail fast with short traceback + raise RuntimeError( + "Runner.run() cannot be called from a running event loop" + ) + + self._lazy_init() + + if context is None: + context = self._context + task = context.run(self._loop.create_task, coro) + + if ( + threading.current_thread() is threading.main_thread() + and signal.getsignal(signal.SIGINT) is signal.default_int_handler + ): + sigint_handler = partial(self._on_sigint, main_task=task) + try: + signal.signal(signal.SIGINT, sigint_handler) + except ValueError: + # `signal.signal` may throw if `threading.main_thread` does + # not support signals (e.g. embedded interpreter with signals + # not registered - see gh-91880) + sigint_handler = None + else: + sigint_handler = None + + self._interrupt_count = 0 + try: + return self._loop.run_until_complete(task) + except exceptions.CancelledError: + if self._interrupt_count > 0: + uncancel = getattr(task, "uncancel", None) + if uncancel is not None and uncancel() == 0: + raise KeyboardInterrupt # noqa: B904 + raise # CancelledError + finally: + if ( + sigint_handler is not None + and signal.getsignal(signal.SIGINT) is sigint_handler + ): + signal.signal(signal.SIGINT, signal.default_int_handler) + + def _lazy_init(self) -> None: + if self._state is _State.CLOSED: + raise RuntimeError("Runner is closed") + if self._state is _State.INITIALIZED: + return + if self._loop_factory is None: + self._loop = events.new_event_loop() + if not self._set_event_loop: + # Call set_event_loop only once to avoid calling + # attach_loop multiple times on child watchers + events.set_event_loop(self._loop) + self._set_event_loop = True + else: + self._loop = self._loop_factory() + if self._debug is not None: + self._loop.set_debug(self._debug) + self._context = contextvars.copy_context() + self._state = _State.INITIALIZED + + def _on_sigint(self, signum, frame, main_task: asyncio.Task) -> None: + self._interrupt_count += 1 + if self._interrupt_count == 1 and not main_task.done(): + main_task.cancel() + # wakeup loop if it is blocked by select() with long timeout + self._loop.call_soon_threadsafe(lambda: None) + return + raise KeyboardInterrupt() + + def _cancel_all_tasks(loop: AbstractEventLoop) -> None: + to_cancel = tasks.all_tasks(loop) + if not to_cancel: + return + + for task in to_cancel: + task.cancel() + + loop.run_until_complete(tasks.gather(*to_cancel, return_exceptions=True)) + + for task in to_cancel: + if task.cancelled(): + continue + if task.exception() is not None: + loop.call_exception_handler( + { + "message": "unhandled exception during asyncio.run() shutdown", + "exception": task.exception(), + "task": task, + } + ) + + async def _shutdown_default_executor(loop: AbstractEventLoop) -> None: + """Schedule the shutdown of the default executor.""" + + def _do_shutdown(future: asyncio.futures.Future) -> None: + try: + loop._default_executor.shutdown(wait=True) # type: ignore[attr-defined] + loop.call_soon_threadsafe(future.set_result, None) + except Exception as ex: + loop.call_soon_threadsafe(future.set_exception, ex) + + loop._executor_shutdown_called = True + if loop._default_executor is None: + return + future = loop.create_future() + thread = threading.Thread(target=_do_shutdown, args=(future,)) + thread.start() + try: + await future + finally: + thread.join() From 6f82b383c72d050c422ad427de76a1b0d625e43f Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sun, 19 Jul 2026 15:18:34 +0100 Subject: [PATCH 09/15] Add changelog entry for Windows SelectorEventLoop subprocess support Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01LZMpdx7hNR89fwdmiMFmVJ --- docs/versionhistory.rst | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst index 345d4ece5..477b6ddd6 100644 --- a/docs/versionhistory.rst +++ b/docs/versionhistory.rst @@ -5,6 +5,10 @@ This library adheres to `Semantic Versioning 2.0 `_. **UNRELEASED** +- Added support for running subprocesses on the asyncio ``SelectorEventLoop`` on Windows, + which does not support subprocesses natively; the pipe I/O is handled on a background + ``ProactorEventLoop`` (or winloop) thread + (`#783 `_; PR by @graingert) - Changed subprocess handling to use a shared, backend-agnostic implementation built on ``subprocess.Popen()`` plus a small set of backend primitives (asynchronous pipes and child-process reaping). This is used by the asyncio backend on all platforms and by the From 260440d39a8a930b92e9a05291b2b5962dd187d8 Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sun, 19 Jul 2026 15:24:48 +0100 Subject: [PATCH 10/15] Add tests for reaping fallback, spawn failure and PathLike commands Covers the worker-thread reaping fallback (no os.pidfd_open, and no os.waitid), the spawn-failure cleanup path, and a single PathLike command, which were previously uncovered by the test suite. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01LZMpdx7hNR89fwdmiMFmVJ --- tests/test_subprocesses.py | 45 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/tests/test_subprocesses.py b/tests/test_subprocesses.py index fabd6ac14..596e0c09a 100644 --- a/tests/test_subprocesses.py +++ b/tests/test_subprocesses.py @@ -519,3 +519,48 @@ async def test_signal_already_exited_process() -> None: process.terminate() process.kill() process.send_signal(signal.SIGTERM) + + +@pytest.mark.skipif( + platform.system() == "Windows", reason="POSIX-only reaping fallback" +) +@pytest.mark.parametrize("have_waitid", [True, False], ids=["waitid", "popen-wait"]) +async def test_wait_without_pidfd( + monkeypatch: pytest.MonkeyPatch, have_waitid: bool +) -> None: + """ + Exercise the worker-thread reaping fallback used when ``os.pidfd_open`` is unavailable + (e.g. PyPy or kernels older than 5.3) and, in turn, when ``os.waitid`` is unavailable + too (e.g. macOS before Python 3.13). + """ + monkeypatch.delattr(os, "pidfd_open", raising=False) + if not have_waitid: + monkeypatch.delattr(os, "waitid", raising=False) + + async with await open_process([sys.executable, "-c", "print('hi')"]) as process: + assert process.pid > 0 + assert process.stdout is not None + output = await BufferedByteReceiveStream(process.stdout).receive_exactly(3) + assert output == b"hi\n" + with fail_after(5): + assert await process.wait() == 0 + + +async def test_open_process_nonexistent_executable() -> None: + """ + A failure to spawn the process should propagate and clean up the pipes that were + already created. + """ + with pytest.raises(FileNotFoundError): + await open_process( + [os.path.join(os.getcwd(), "nonexistent-anyio-test-executable")] + ) + + +@pytest.mark.skipif( + platform.system() == "Windows", reason="uses a POSIX executable path" +) +async def test_run_process_pathlike_command() -> None: + """A single ``PathLike`` command is accepted (and run via the shell).""" + result = await run_process(Path("/bin/echo")) + assert result.returncode == 0 From bfeed5fa5642055d5c37234bee207b1b026da27e Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sun, 19 Jul 2026 15:30:46 +0100 Subject: [PATCH 11/15] more coverage --- tests/test_subprocesses.py | 43 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/tests/test_subprocesses.py b/tests/test_subprocesses.py index 596e0c09a..6c7e60c7b 100644 --- a/tests/test_subprocesses.py +++ b/tests/test_subprocesses.py @@ -564,3 +564,46 @@ async def test_run_process_pathlike_command() -> None: """A single ``PathLike`` command is accepted (and run via the shell).""" result = await run_process(Path("/bin/echo")) assert result.returncode == 0 + + +async def test_receive_smaller_than_chunk() -> None: + """ + Receiving with a ``max_bytes`` smaller than a buffered chunk returns just that many + bytes and keeps the rest for the next call. + """ + code = dedent("""\ + import sys + sys.stdout.buffer.write(b"hello") + sys.stdout.flush() + sys.stdin.read() + """) + async with await open_process([sys.executable, "-c", code]) as process: + assert process.stdin is not None + assert process.stdout is not None + data = b"" + with fail_after(5): + while len(data) < 5: + chunk = await process.stdout.receive(1) + assert len(chunk) == 1 + data += chunk + + assert data == b"hello" + await process.stdin.aclose() + + +async def test_send_backpressure() -> None: + """ + Sending more than the pipe buffer to a subprocess that isn't reading yet must apply + backpressure rather than failing or buffering without bound. + """ + code = dedent("""\ + import sys, time + time.sleep(0.2) + sys.stdin.buffer.read() + """) + async with await open_process([sys.executable, "-c", code]) as process: + assert process.stdin is not None + with fail_after(5): + await process.stdin.send(b"x" * 1024 * 1024) + await process.stdin.aclose() + assert await process.wait() == 0 From 2c39c1cdbff91e1858284cd9265336b11db1d624 Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sun, 19 Jul 2026 15:39:55 +0100 Subject: [PATCH 12/15] Apply suggestion from @graingert --- docs/versionhistory.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst index 16ea610be..89902ddc0 100644 --- a/docs/versionhistory.rst +++ b/docs/versionhistory.rst @@ -8,7 +8,7 @@ This library adheres to `Semantic Versioning 2.0 `_. - Added support for running subprocesses on the asyncio ``SelectorEventLoop`` on Windows, which does not support subprocesses natively; the pipe I/O is handled on a background ``ProactorEventLoop`` (or winloop) thread - (`#783 `_; PR by @graingert) + (`#1235 `_; PR by @graingert) - Changed subprocess handling to use a shared, backend-agnostic implementation built on ``subprocess.Popen()`` plus a small set of backend primitives (asynchronous pipes and child-process reaping). This is used by the asyncio backend on all platforms and by the From 48ce8a221c8fc0d11b03e19ef9c6ae4339e6c3e7 Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sun, 19 Jul 2026 15:40:17 +0100 Subject: [PATCH 13/15] Apply suggestion from @graingert --- docs/versionhistory.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst index 89902ddc0..f78f9b9f7 100644 --- a/docs/versionhistory.rst +++ b/docs/versionhistory.rst @@ -8,7 +8,7 @@ This library adheres to `Semantic Versioning 2.0 `_. - Added support for running subprocesses on the asyncio ``SelectorEventLoop`` on Windows, which does not support subprocesses natively; the pipe I/O is handled on a background ``ProactorEventLoop`` (or winloop) thread - (`#1235 `_; PR by @graingert) + (`#1235 `_; PR by @graingert) - Changed subprocess handling to use a shared, backend-agnostic implementation built on ``subprocess.Popen()`` plus a small set of backend primitives (asynchronous pipes and child-process reaping). This is used by the asyncio backend on all platforms and by the From f667063683401094e2d617ac39332c3c04330292 Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sun, 19 Jul 2026 15:44:36 +0100 Subject: [PATCH 14/15] Apply suggestions from code review Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/anyio/_backends/_asyncio.py | 16 ++++++++++++---- src/anyio/_core/_subprocesses.py | 21 ++++++++++++++++++--- 2 files changed, 30 insertions(+), 7 deletions(-) diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py index ad5267df6..b07f60022 100644 --- a/src/anyio/_backends/_asyncio.py +++ b/src/anyio/_backends/_asyncio.py @@ -959,8 +959,10 @@ async def receive(self, max_bytes: int = 65536) -> bytes: and not self._protocol.is_at_eof ): self._transport.resume_reading() - await self._protocol.read_event.wait() - self._transport.pause_reading() + try: + await self._protocol.read_event.wait() + finally: + self._transport.pause_reading() else: await AsyncIOBackend.checkpoint() @@ -2608,7 +2610,10 @@ async def create_subprocess_stdin_pipe(cls) -> tuple[abc.ByteSendStream, int]: _ProcessPipeProtocol, pipe_obj ) except BaseException: - os.close(child_fd) + try: + pipe_obj.close() + finally: + os.close(child_fd) raise return _ProcessSendPipeStream(transport, protocol), child_fd @@ -2655,7 +2660,10 @@ async def create_subprocess_output_pipe(cls) -> tuple[abc.ByteReceiveStream, int _ProcessPipeProtocol, pipe_obj ) except BaseException: - os.close(child_fd) + try: + pipe_obj.close() + finally: + os.close(child_fd) raise return _ProcessReceivePipeStream(transport, protocol), child_fd diff --git a/src/anyio/_core/_subprocesses.py b/src/anyio/_core/_subprocesses.py index 157d66f13..8b94fbdd1 100644 --- a/src/anyio/_core/_subprocesses.py +++ b/src/anyio/_core/_subprocesses.py @@ -141,13 +141,28 @@ async def wait(self) -> int: return cast(int, self._popen.returncode) def terminate(self) -> None: - self._popen.terminate() + if self._popen.poll() is not None: + return + try: + self._popen.terminate() + except ProcessLookupError: + pass def kill(self) -> None: - self._popen.kill() + if self._popen.poll() is not None: + return + try: + self._popen.kill() + except ProcessLookupError: + pass def send_signal(self, signal: Signals) -> None: - self._popen.send_signal(signal) + if self._popen.poll() is not None: + return + try: + self._popen.send_signal(signal) + except ProcessLookupError: + pass @property def pid(self) -> int: From b45c11191812c819dc2576a0a510a7b5e6370ecd Mon Sep 17 00:00:00 2001 From: Thomas Grainger Date: Sun, 19 Jul 2026 15:50:00 +0100 Subject: [PATCH 15/15] add docs --- docs/subprocesses.rst | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/docs/subprocesses.rst b/docs/subprocesses.rst index 5245eae40..e30953817 100644 --- a/docs/subprocesses.rst +++ b/docs/subprocesses.rst @@ -6,6 +6,11 @@ Using subprocesses AnyIO allows you to run arbitrary executables in subprocesses, either as a one-shot call or by opening a process handle for you that gives you more control over the subprocess. +On the asyncio backend, subprocess support now also works on Windows when the event loop +is a ``SelectorEventLoop``. Since that event loop cannot perform pipe I/O for +subprocesses directly, AnyIO handles the subprocess pipes on a background +``ProactorEventLoop`` (or winloop) thread instead. + You can either give the command as a string, in which case it is passed to your default shell (equivalent to ``shell=True`` in :func:`subprocess.run`), or as a sequence of strings (``shell=False``) in which case the executable is the first item in the sequence @@ -55,6 +60,15 @@ launch one with :func:`~open_process`:: See the API documentation of :class:`~.abc.Process` for more information. +There are a few important process semantics to be aware of: + +* :meth:`~.abc.Process.wait` returns as soon as the subprocess exits; it does not wait + for ``stdout`` or ``stderr`` to close +* :attr:`~.abc.Process.returncode` reflects the real exit status once the subprocess has + exited, even if :meth:`~.abc.Process.wait` was never called +* calling :meth:`~.abc.Process.terminate`, :meth:`~.abc.Process.kill` or + :meth:`~.abc.Process.send_signal` on an already-exited process is a no-op + .. _RunInProcess: Running functions in worker processes