🐛 fix(execute): adopt CPython subprocess stream handling (#3715)

Subprocess output reading could deadlock on Windows during parallel test
execution or when handling large amounts of output. The root cause was a
chicken-and-egg problem where reader threads waited for the subprocess
to close its pipes, but the subprocess wouldn't be terminated until
after the readers exited. 🔒 This manifested as timeouts in CI when tox
tried to clean up long-running backend processes.

The fix adopts CPython's approach to subprocess stream handling. On
Unix, we switched from `select.select()` to the `selectors` module with
proper EOF detection and interrupt handling. On Windows, we use
overlapped I/O with non-blocking polling, mirroring CPython's
implementation. Most importantly, we reordered the shutdown sequence in
`pep517_backend.py` to terminate subprocesses before stopping reader
threads, ensuring pipes close and pending I/O operations complete
naturally. ⚡

This eliminates arbitrary timeouts and race conditions in the process
cleanup logic. The implementation now handles EINTR signals gracefully
and reads larger chunks (32KB instead of 1KB) for better performance
with high-volume output.

**References:**
- CPython's subprocess implementation:
https://github.com/python/cpython/blob/main/Lib/subprocess.py
- Windows overlapped I/O:
https://docs.microsoft.com/en-us/windows/win32/fileio/synchronous-and-asynchronous-i-o
- Python `selectors` module:
https://docs.python.org/3/library/selectors.html
- `_overlapped` module:
https://github.com/python/cpython/blob/main/Modules/overlapped.c

Author: gaborbernat

docs/changelog/3715.bugfix.rst

@@ -0,0 +1 @@
+Adopt CPython's subprocess stream handling to fix deadlocks and improve performance when reading subprocess output

src/tox/execute/local_sub_process/__init__.py

@@ -227,10 +227,6 @@ def __enter__(self) -> ExecuteStatus:
         self._read_stderr.__enter__()
         self._read_stdout = ReadViaThread(stdout.send(process), self.out_handler, name=f"out-{pid}", drain=drain)
         self._read_stdout.__enter__()
-
-        if sys.platform == "win32":  # explicit check for mypy:  # pragma: win32 cover
-            process.stderr.read = self._read_stderr._drain_stream  # noqa: SLF001  # ty: ignore[invalid-assignment] # monkey-patching drain onto Popen stream
-            process.stdout.read = self._read_stdout._drain_stream  # noqa: SLF001  # ty: ignore[invalid-assignment] # monkey-patching drain onto Popen stream
         return status
 
     def __exit__(

src/tox/execute/local_sub_process/read_via_thread.py

@@ -1,4 +1,4 @@
-"""A reader that drain a stream via its file no on a background thread."""
+"""A reader that drains a stream via its file descriptor, following CPython's subprocess approach."""
 
 from __future__ import annotations
 
@@ -17,7 +17,7 @@
         from typing_extensions import Self
 
 
-WAIT_GENERAL = 0.05  # stop thread join every so often (give chance to a signal interrupt)
+WAIT_GENERAL = 0.05
 
 
 class ReadViaThread(ABC):
@@ -38,10 +38,11 @@ def __exit__(
         exc_val: BaseException | None,
         exc_tb: TracebackType | None,
     ) -> None:
-        self.stop.set()  # signal thread to stop
-        while self.thread.is_alive():  # wait until it stops
+        self.stop.set()
+        while self.thread.is_alive():
             self.thread.join(WAIT_GENERAL)
-        self._drain_stream()  # read anything left
+        if self._on_exit_drain:
+            self._drain_stream()
 
     @abstractmethod
     def _read_stream(self) -> None:

src/tox/execute/local_sub_process/read_via_thread_unix.py

@@ -1,50 +1,93 @@
-"""On UNIX we use select.select to ensure we drain in a non-blocking fashion."""
+"""On UNIX we use selectors to drain streams efficiently, following CPython's subprocess implementation."""
 
 from __future__ import annotations
 
+import contextlib  # pragma: win32 no cover
 import errno  # pragma: win32 no cover
 import os  # pragma: win32 no cover
-import select  # pragma: win32 no cover
-from typing import TYPE_CHECKING
+import selectors  # pragma: win32 no cover
+from typing import TYPE_CHECKING, Any
 
 from .read_via_thread import ReadViaThread  # pragma: win32 no cover
 
 if TYPE_CHECKING:
     from collections.abc import Callable
 
-STOP_EVENT_CHECK_PERIODICITY_IN_MS = 0.01  # pragma: win32 no cover
+TIMEOUT_FOR_INTERRUPT = 0.05  # pragma: win32 no cover
+READ_CHUNK_SIZE = 32768  # pragma: win32 no cover
 
 
 class ReadViaThreadUnix(ReadViaThread):  # pragma: win32 no cover
     def __init__(self, file_no: int, handler: Callable[[bytes], int], name: str, drain: bool) -> None:  # noqa: FBT001
         super().__init__(file_no, handler, name, drain)
 
     def _read_stream(self) -> None:
-        while not self.stop.is_set():
-            # we need to drain the stream, but periodically give chance for the thread to break if the stop event has
-            # been set (this is so that an interrupt can be handled)
-            if self._read_available() is None:  # pragma: no branch
-                break  # pragma: no cover
+        selector = selectors.DefaultSelector()
+        try:
+            selector.register(self.file_no, selectors.EVENT_READ)
+        except (OSError, ValueError):  # pragma: no cover
+            return
+
+        try:
+            self._read_until_eof(selector)
+        finally:
+            selector.close()
+
+    def _read_until_eof(self, selector: selectors.DefaultSelector) -> None:
+        while selector.get_map() and not self.stop.is_set():
+            try:
+                ready = selector.select(timeout=TIMEOUT_FOR_INTERRUPT)
+            except (InterruptedError, OSError) as exception:
+                if isinstance(exception, OSError) and exception.errno != errno.EINTR:
+                    raise
+                continue
+
+            if not ready:
+                continue
+
+            for key, _ in ready:
+                self._read_chunk(selector, key)
 
     def _drain_stream(self) -> None:
-        # no block just poll
-        while True:
-            if self._read_available(timeout=0) is not True:  # pragma: no branch
-                break  # pragma: no cover
+        selector = selectors.DefaultSelector()
+        try:
+            selector.register(self.file_no, selectors.EVENT_READ)
+        except (OSError, ValueError):  # pragma: no cover
+            return
 
-    def _read_available(self, timeout: float = STOP_EVENT_CHECK_PERIODICITY_IN_MS) -> bool | None:
         try:
-            ready, __, ___ = select.select([self.file_no], [], [], timeout)
-            if ready:
-                data = os.read(self.file_no, 1024)  # read up to 1024 characters
-                # If the end of the file referred to by fd has been reached, an empty bytes object is returned.
-                if data:
-                    self.handler(data)
-                    return True
-        except OSError as exception:  # pragma: no cover
-            # Bad file descriptor or Input/output error
-            if exception.errno in {errno.EBADF, errno.EIO}:
-                return None
-            raise
+            while selector.get_map():
+                try:
+                    ready = selector.select(timeout=0)
+                except (InterruptedError, OSError) as exception:
+                    if isinstance(exception, OSError) and exception.errno != errno.EINTR:  # pragma: no cover
+                        raise  # pragma: no cover
+                    continue
+
+                if not ready:
+                    break
+
+                for key, _ in ready:
+                    self._read_chunk(selector, key)
+        finally:
+            selector.close()
+
+    def _read_chunk(self, selector: selectors.DefaultSelector, key: selectors.SelectorKey) -> None:
+        try:
+            data = os.read(key.fd, READ_CHUNK_SIZE)
+        except OSError as exception:
+            if exception.errno == errno.EINTR:
+                return
+            if exception.errno not in {errno.EBADF, errno.EIO}:  # pragma: no cover
+                raise  # pragma: no cover
+            data = b""
+
+        if data:
+            self.handler(data)
         else:
-            return False
+            self._safe_unregister(selector, key.fileobj)
+
+    @staticmethod
+    def _safe_unregister(selector: selectors.DefaultSelector, fileobj: Any) -> None:
+        with contextlib.suppress(KeyError, ValueError):
+            selector.unregister(fileobj)

src/tox/execute/local_sub_process/read_via_thread_windows.py

@@ -1,69 +1,65 @@
-"""On Windows we use overlapped mechanism, borrowing it from asyncio (but without the event loop)."""
+"""On Windows we use overlapped I/O for efficient real-time stream reading."""
 
 from __future__ import annotations  # pragma: win32 cover
 
-import _overlapped  # pragma: win32 cover # noqa: PLC2701  # ty: ignore[unresolved-import] # no typeshed stubs
-import logging  # pragma: win32 cover
-from asyncio.windows_utils import (
-    BUFSIZE,
-)  # pragma: win32 cover
-from time import sleep  # pragma: win32 cover
+import _overlapped  # type: ignore[import-untyped]  # pragma: win32 cover # noqa: PLC2701
+import time  # pragma: win32 cover
 from typing import TYPE_CHECKING
 
 from .read_via_thread import ReadViaThread  # pragma: win32 cover
 
 if TYPE_CHECKING:
     from collections.abc import Callable
 
+READ_CHUNK_SIZE = 32768  # pragma: win32 cover
+POLL_INTERVAL = 0.05  # pragma: win32 cover
+ERROR_IO_INCOMPLETE = 996  # pragma: win32 cover
+
 
 class ReadViaThreadWindows(ReadViaThread):  # pragma: win32 cover
     def __init__(self, file_no: int, handler: Callable[[bytes], int], name: str, drain: bool) -> None:  # noqa: FBT001
         super().__init__(file_no, handler, name, drain)
-        self.closed = False
-        self._ov: _overlapped.Overlapped | None = None
-        self._waiting_for_read = False
 
     def _read_stream(self) -> None:
-        keep_reading = True
-        while keep_reading:  # try to read at least once
-            wait = self._read_batch()
-            if wait is None:
-                break
-            if wait is True:
-                sleep(0.01)  # sleep for 10ms if there was no data to read and try again
-            keep_reading = not self.stop.is_set()
+        try:
+            while not self.stop.is_set():
+                ov = _overlapped.Overlapped(0)
+                try:
+                    ov.ReadFile(self.file_no, READ_CHUNK_SIZE)
+                except OSError:
+                    break
+
+                while True:
+                    try:
+                        data = ov.getresult(False)  # noqa: FBT003
+                        break
+                    except OSError as exception:
+                        if getattr(exception, "winerror", None) != ERROR_IO_INCOMPLETE:
+                            return
+                        time.sleep(POLL_INTERVAL)
+
+                if not data:
+                    break
+                self.handler(data)
+        except OSError:  # pragma: no cover
+            pass
 
     def _drain_stream(self) -> None:
-        wait: bool | None = self.closed
-        while wait is False:
-            wait = self._read_batch()
+        try:
+            while True:
+                ov = _overlapped.Overlapped(0)
+                try:
+                    ov.ReadFile(self.file_no, READ_CHUNK_SIZE)
+                except OSError:
+                    break
+
+                try:
+                    data = ov.getresult(True)  # noqa: FBT003
+                except OSError:
+                    break
 
-    def _read_batch(self) -> bool | None:
-        """:returns: None means error can no longer read, True wait for result, False try again"""
-        if self._waiting_for_read is False:
-            self._ov = _overlapped.Overlapped(0)  # can use it only once to read a batch
-            try:  # read up to BUFSIZE at a time
-                self._ov.ReadFile(self.file_no, BUFSIZE)
-                self._waiting_for_read = True
-            except OSError:
-                self.closed = True
-                return None
-        try:  # wait=False to not block and give chance for the stop check
-            data = self._ov.getresult(False)  # noqa: FBT003  # ty: ignore[unresolved-attribute] # https://github.com/astral-sh/ty/issues/160
-        except OSError as exception:
-            # 996 (0x3E4) Overlapped I/O event is not in a signaled state.
-            # 995 (0x3E3) The I/O operation has been aborted because of either a thread exit or an application request.
-            win_error = getattr(exception, "winerror", None)
-            if win_error == 996:  # noqa: PLR2004
-                return True
-            if win_error != 995:  # noqa: PLR2004
-                logging.error("failed to read %r", exception)  # noqa: TRY400
-            return None
-        else:
-            self._ov = None
-            self._waiting_for_read = False
-            if data:
+                if not data:
+                    break
                 self.handler(data)
-            else:
-                return None
-        return False
+        except OSError:  # pragma: no cover
+            pass

src/tox/execute/pep517_backend.py

@@ -86,12 +86,12 @@ def _handler(into: bytearray, content: bytes) -> None:
     def close(self) -> None:
         if self._local_execute is not None:  # pragma: no branch
             execute, _status = self._local_execute
-            execute.__exit__(None, None, None)
             if execute.process is not None and execute.process.returncode is None:  # pragma: no cover
                 try:  # pragma: no cover
                     execute.process.wait(timeout=0.1)  # pragma: no cover
                 except TimeoutExpired:  # pragma: no cover
                     execute.process.terminate()  # pragma: no cover  # if does not stop on its own kill it
+            execute.__exit__(None, None, None)
             self._local_execute = None
         self.is_alive = False