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
diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index 961cadb79..f78f9b9f7 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -5,6 +5,26 @@ 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
+ (`#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
+ 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
- Changed the default name for a task spawned with ``TaskGroup.create_task(func())`` to
match the default task name for the analogous task spawned with
``TaskGroup.start_soon(func)`` or ``TaskGroup.start(func)`` in more situations.
diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py
index c8022792e..b07f60022 100644
--- a/src/anyio/_backends/_asyncio.py
+++ b/src/anyio/_backends/_asyncio.py
@@ -7,8 +7,8 @@
import math
import os
import socket
+import subprocess
import sys
-import threading
import weakref
from asyncio import (
AbstractEventLoop,
@@ -34,7 +34,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 +41,6 @@
getcoroutinestate,
)
from io import IOBase
-from os import PathLike
from queue import Queue
from signal import Signals
from socket import AddressFamily, SocketKind
@@ -111,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)
@@ -1069,150 +886,184 @@ 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.
+
+ 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).
+ """
+
+ read_queue: deque[bytes]
+ read_event: asyncio.Event
+ write_event: asyncio.Event
+ exception: Exception | None = None
+ is_at_eof: bool = False
+
+ 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
+
+ self.read_event.set()
+ self.write_event.set()
+
+ def data_received(self, data: bytes) -> None:
+ # ProactorEventloop sometimes sends bytearray instead of bytes
+ self.read_queue.append(bytes(data))
+ self.read_event.set()
+
+ def eof_received(self) -> bool | None:
+ self.is_at_eof = True
+ self.read_event.set()
+ return True
+
+ def pause_writing(self) -> None:
+ self.write_event = asyncio.Event()
+
+ def resume_writing(self) -> None:
+ self.write_event.set()
+
+
+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")
- data = await self._stream.read(max_bytes)
- if data:
- return data
- else:
- raise EndOfStream
-
- async def aclose(self) -> None:
- self._stream.set_exception(ClosedResourceError())
- await AsyncIOBackend.checkpoint()
+ 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()
+ try:
+ await self._protocol.read_event.wait()
+ finally:
+ self._transport.pause_reading()
+ else:
+ await AsyncIOBackend.checkpoint()
+ try:
+ 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()
-@dataclass(eq=False)
-class StreamWriterWrapper(abc.ByteSendStream):
- _stream: asyncio.StreamWriter
- _closed: bool = field(init=False, default=False)
+ raise EndOfStream from None
- 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
+ if len(chunk) > max_bytes:
+ # Split the oversized chunk
+ chunk, leftover = chunk[:max_bytes], chunk[max_bytes:]
+ self._protocol.read_queue.appendleft(leftover)
- raise
+ # 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()
- if not stream_paused:
- await AsyncIOBackend.cancel_shielded_checkpoint()
+ return chunk
async def aclose(self) -> None:
self._closed = True
- self._stream.close()
- await AsyncIOBackend.checkpoint()
-
-
-@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
+ if not self._transport.is_closing():
+ self._transport.close()
- 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
+ await AsyncIOBackend.checkpoint()
- async def wait(self) -> int:
- await self._exited.wait()
- assert self._process.returncode is not None
- return self._process.returncode
+ def _abort(self) -> None:
+ self._closed = True
+ if not self._transport.is_closing():
+ self._transport.close()
- def terminate(self) -> None:
- self._process.terminate()
- def kill(self) -> None:
- self._process.kill()
+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
- def send_signal(self, signal: int) -> None:
- self._process.send_signal(signal)
+ 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 pid(self) -> int:
- return self._process.pid
+ try:
+ self._transport.write(item)
+ except RuntimeError as exc:
+ if self._transport.is_closing():
+ raise BrokenResourceError from exc
+ else:
+ raise
- @property
- def returncode(self) -> int | None:
- return self._process.returncode
+ await self._protocol.write_event.wait()
- @property
- def stdin(self) -> abc.ByteSendStream | None:
- return self._stdin
+ async def aclose(self) -> None:
+ self._closed = True
+ if not self._transport.is_closing():
+ self._transport.close()
- @property
- def stdout(self) -> abc.ByteReceiveStream | None:
- return self._stdout
+ await AsyncIOBackend.checkpoint()
- @property
- def stderr(self) -> abc.ByteReceiveStream | None:
- return self._stderr
+ 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[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._abort() # 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:
@@ -2429,24 +2280,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(
@@ -2725,46 +2558,134 @@ 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" 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
+ from asyncio.windows_utils import pipe as windows_pipe
+
+ 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:
- 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(write_fd, "wb", 0)
+ child_fd = read_fd
+
+ try:
+ transport, protocol = await loop.connect_write_pipe(
+ _ProcessPipeProtocol, pipe_obj
)
+ except BaseException:
+ try:
+ pipe_obj.close()
+ finally:
+ 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 _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" 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
+ 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))
+ 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()
+ 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:
+ try:
+ pipe_obj.close()
+ finally:
+ os.close(child_fd)
+ raise
+
+ return _ProcessReceivePipeStream(transport, protocol), child_fd
+
+ @classmethod
+ async def wait_for_child_exit(cls, process: subprocess.Popen[bytes]) -> None:
+ if sys.platform == "win32":
+ 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)
+ 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:
@@ -2773,7 +2694,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 602b35b5c..046b479c2 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
@@ -415,6 +416,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()
@@ -1209,7 +1294,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):
@@ -1241,6 +1333,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/_asyncio_proactor_thread.py b/src/anyio/_core/_asyncio_proactor_thread.py
new file mode 100644
index 000000000..db6770996
--- /dev/null
+++ b/src/anyio/_core/_asyncio_proactor_thread.py
@@ -0,0 +1,123 @@
+"""
+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
+from ._asyncio_runner import Runner
+
+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: 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.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()
+ 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._stop_event.set)
+ 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())
+
+ 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."""
+
+ 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 _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
+
+ with _proactor_thread_lock:
+ if _proactor_thread is None:
+ _proactor_thread = ProactorThread()
+ _proactor_thread.start()
+
+ return _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()
diff --git a/src/anyio/_core/_asyncio_windows_process.py b/src/anyio/_core/_asyncio_windows_process.py
new file mode 100644
index 000000000..457e51e18
--- /dev/null
+++ b/src/anyio/_core/_asyncio_windows_process.py
@@ -0,0 +1,103 @@
+"""
+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
+# 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)
+
+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.UnregisterWaitEx.argtypes = [wintypes.HANDLE, wintypes.HANDLE]
+kernel32.UnregisterWaitEx.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:
+ # 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.UnregisterWaitEx(waiter._wait_handle, INVALID_HANDLE_VALUE)
+
+ kernel32.CloseHandle(process_handle)
diff --git a/src/anyio/_core/_subprocesses.py b/src/anyio/_core/_subprocesses.py
index a6590ca62..8b94fbdd1 100644
--- a/src/anyio/_core/_subprocesses.py
+++ b/src/anyio/_core/_subprocesses.py
@@ -1,17 +1,271 @@
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_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[bytes]) -> None:
+ """
+ Backend-agnostic POSIX implementation of
+ :meth:`~anyio.abc.AsyncBackend.wait_for_child_exit`.
+
+ 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" and hasattr(os, "pidfd_open"):
+ 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_for_exit,
+ (process,),
+ 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:
+ if self._popen.poll() is not None:
+ return
+ try:
+ self._popen.terminate()
+ except ProcessLookupError:
+ pass
+
+ def kill(self) -> None:
+ if self._popen.poll() is not None:
+ return
+ try:
+ self._popen.kill()
+ except ProcessLookupError:
+ pass
+
+ def send_signal(self, signal: Signals) -> None:
+ if self._popen.poll() is not None:
+ return
+ try:
+ self._popen.send_signal(signal)
+ except ProcessLookupError:
+ pass
+
+ @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..dea56ea45 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[bytes]) -> None:
+ """
+ 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 implementation may reap the process
+ itself, as long as it does so via ``process``).
+ """
+
@classmethod
@abstractmethod
def setup_process_pool_exit_at_shutdown(cls, workers: set[Process]) -> None:
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():
diff --git a/tests/test_subprocesses.py b/tests/test_subprocesses.py
index 132a78833..6c7e60c7b 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,125 @@ 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)
+
+
+@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
+
+
+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