From 546e32bd290ea95990aaf26750359eb8129d0ccc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bern=C3=A1t=20G=C3=A1bor?= Date: Mon, 2 Nov 2026 18:09:18 -0800 Subject: [PATCH 1/3] =?UTF-8?q?=F0=9F=90=9B=20fix(query):=20respect=20time?= =?UTF-8?q?out=20with=20child=20pipes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../+probe-descendant-pipes.bugfix.rst | 1 + src/python_discovery/_cached_py_info.py | 3 +-- tests/test_query_timeout.py | 22 +++++++++++++++++++ 3 files changed, 24 insertions(+), 2 deletions(-) create mode 100644 docs/changelog/+probe-descendant-pipes.bugfix.rst diff --git a/docs/changelog/+probe-descendant-pipes.bugfix.rst b/docs/changelog/+probe-descendant-pipes.bugfix.rst new file mode 100644 index 0000000..3a62d26 --- /dev/null +++ b/docs/changelog/+probe-descendant-pipes.bugfix.rst @@ -0,0 +1 @@ +Return when an interpreter query times out even if its child keeps the output pipes open. diff --git a/src/python_discovery/_cached_py_info.py b/src/python_discovery/_cached_py_info.py index bcade0d..9029c13 100644 --- a/src/python_discovery/_cached_py_info.py +++ b/src/python_discovery/_cached_py_info.py @@ -304,8 +304,7 @@ def _collect_query_output(process: Popen[str], timeout: float | None) -> tuple[s if not _query_completed(process, readers, timeout): with suppress(OSError): process.kill() - for reader in readers: - reader.join() + # A descendant may keep the pipes open after the candidate exits. process.wait() return "", "timed out", -1 if read_errors: diff --git a/tests/test_query_timeout.py b/tests/test_query_timeout.py index 6db946f..03b1255 100644 --- a/tests/test_query_timeout.py +++ b/tests/test_query_timeout.py @@ -3,6 +3,7 @@ import logging import os import sys +from time import monotonic from typing import TYPE_CHECKING import pytest @@ -11,6 +12,8 @@ from python_discovery._cached_py_info import _collect_query_output if TYPE_CHECKING: + from pathlib import Path + from pytest_mock import MockerFixture @@ -86,3 +89,22 @@ def test_from_exe_timeout_warning_cache_is_bounded(caplog: pytest.LogCaptureFixt assert [record.getMessage() for record in caplog.records] == [ "ignoring PY_DISCOVERY_TIMEOUT='invalid-timeout-0', not a positive number of seconds; using 15.0" ] + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX shell launcher") +def test_from_exe_timeout_with_inherited_pipes(tmp_path: Path) -> None: + executable = tmp_path / "python3.11" + executable.write_text("#!/bin/sh\n/bin/sleep 3 &\nexit 0\n") + executable.chmod(0o755) + + started = monotonic() + result = PythonInfo.from_exe( + str(executable), + env={**os.environ, "PY_DISCOVERY_TIMEOUT": "0.1"}, + ignore_cache=True, + raise_on_error=False, + resolve_to_host=False, + ) + + assert result is None + assert monotonic() - started < 2.5 From 178415117bdfb83e0b8de749a81f813caf6405f2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bern=C3=A1t=20G=C3=A1bor?= Date: Mon, 2 Nov 2026 18:44:56 -0800 Subject: [PATCH 2/3] =?UTF-8?q?=F0=9F=90=9B=20fix(query):=20stop=20timed-o?= =?UTF-8?q?ut=20pipe=20readers?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A child can keep query pipes open after the candidate exits, leaving daemon readers blocked beyond the timeout. Poll POSIX pipes for readiness and cancel pending Windows reads before joining both readers. --- .../+probe-descendant-pipes.bugfix.rst | 2 +- src/python_discovery/_cached_py_info.py | 66 ++++++++++++++----- src/python_discovery/_windows/_cancel_io.py | 51 ++++++++++++++ tests/test_cached_py_info.py | 50 +++++++++++--- 4 files changed, 140 insertions(+), 29 deletions(-) create mode 100644 src/python_discovery/_windows/_cancel_io.py diff --git a/docs/changelog/+probe-descendant-pipes.bugfix.rst b/docs/changelog/+probe-descendant-pipes.bugfix.rst index 3a62d26..7b93990 100644 --- a/docs/changelog/+probe-descendant-pipes.bugfix.rst +++ b/docs/changelog/+probe-descendant-pipes.bugfix.rst @@ -1 +1 @@ -Return when an interpreter query times out even if its child keeps the output pipes open. +Return after an interpreter query times out without leaving pipe readers blocked by a child process. diff --git a/src/python_discovery/_cached_py_info.py b/src/python_discovery/_cached_py_info.py index 9029c13..b5324d5 100644 --- a/src/python_discovery/_cached_py_info.py +++ b/src/python_discovery/_cached_py_info.py @@ -10,11 +10,14 @@ import secrets import sys import tempfile +from codecs import getincrementaldecoder from collections import OrderedDict from contextlib import contextmanager, suppress +from dataclasses import dataclass from functools import lru_cache -from io import DEFAULT_BUFFER_SIZE +from io import DEFAULT_BUFFER_SIZE, IncrementalNewlineDecoder from pathlib import Path +from select import select from shlex import quote from subprocess import DEVNULL, PIPE, Popen, TimeoutExpired, list2cmdline # ruff:ignore[suspicious-subprocess-import] from threading import Event, Thread @@ -25,6 +28,9 @@ from ._py_info import PythonInfo from ._py_info_collect import MIN_INTERROGATE_VERSION, UNSUPPORTED_EXIT_CODE, UNSUPPORTED_MARKER +if sys.platform == "win32": # pragma: win32 cover + from ._windows._cancel_io import stop_query_reader + if TYPE_CHECKING: from collections.abc import Generator, Mapping from typing import IO, TypedDict @@ -292,28 +298,45 @@ def _collect_query_output(process: Popen[str], timeout: float | None) -> tuple[s raise ValueError(msg) out_chunks: list[str] = [] err_chunks: list[str] = [] - overflow = Event() - read_errors: list[OSError] = [] + state = _QueryReadState(process=process, overflow=Event(), stop=Event(), read_errors=[]) readers = ( - Thread(target=_drain_query_stream, args=(process, stdout, out_chunks, overflow, read_errors), daemon=True), - Thread(target=_drain_query_stream, args=(process, stderr, err_chunks, overflow, read_errors), daemon=True), + Thread(target=_drain_query_stream, args=(state, stdout, out_chunks), daemon=True), + Thread(target=_drain_query_stream, args=(state, stderr, err_chunks), daemon=True), ) for reader in readers: reader.start() if not _query_completed(process, readers, timeout): + state.stop.set() with suppress(OSError): process.kill() # A descendant may keep the pipes open after the candidate exits. process.wait() + _stop_query_readers(readers) return "", "timed out", -1 - if read_errors: - raise read_errors[0] - if overflow.is_set(): + if state.read_errors: + raise state.read_errors[0] + if state.overflow.is_set(): return "", f"produced more than {_MAX_QUERY_OUTPUT} characters of output", -1 return "".join(out_chunks), "".join(err_chunks), process.wait() +@dataclass(frozen=True) +class _QueryReadState: + process: Popen[str] + overflow: Event + stop: Event + read_errors: list[OSError] + + +def _stop_query_readers(readers: tuple[Thread, Thread]) -> None: + for reader in readers: + if sys.platform == "win32": # pragma: win32 cover + stop_query_reader(reader) + else: + reader.join() + + def _query_completed(process: Popen[str], readers: tuple[Thread, Thread], timeout: float | None) -> bool: deadline = None if timeout is None else monotonic() + timeout for reader in readers: @@ -327,30 +350,37 @@ def _query_completed(process: Popen[str], readers: tuple[Thread, Thread], timeou return True -def _drain_query_stream( - process: Popen[str], stream: IO[str], chunks: list[str], overflow: Event, read_errors: list[OSError] -) -> None: +def _drain_query_stream(state: _QueryReadState, stream: IO[str], chunks: list[str]) -> None: size = 0 failed = False + decoder = IncrementalNewlineDecoder(getincrementaldecoder("utf-8")("backslashreplace"), translate=True) with stream: - while True: + while not state.stop.is_set(): try: - chunk = stream.read(DEFAULT_BUFFER_SIZE) + if sys.platform != "win32" and not select([stream], [], [], 0.05)[0]: # pragma: win32 cover + continue + data = os.read(stream.fileno(), DEFAULT_BUFFER_SIZE) except OSError as error: - read_errors.append(error) - failed = True + if not state.stop.is_set(): # pragma: win32 cover # canceled ReadFile raises on Windows + state.read_errors.append(error) + failed = True break + chunk = decoder.decode(data, final=not data) if not chunk: - break + if not data: + break + continue chunks.append(chunk[: _MAX_QUERY_OUTPUT + 1 - size]) size += len(chunk) if size > _MAX_QUERY_OUTPUT: - overflow.set() + state.overflow.set() failed = True break + if not data: + break if failed: with suppress(OSError): - process.kill() + state.process.kill() @lru_cache(maxsize=128) diff --git a/src/python_discovery/_windows/_cancel_io.py b/src/python_discovery/_windows/_cancel_io.py new file mode 100644 index 0000000..7816bc7 --- /dev/null +++ b/src/python_discovery/_windows/_cancel_io.py @@ -0,0 +1,51 @@ +"""Cancel a blocked query pipe read on Windows.""" + +from __future__ import annotations + +from ctypes import WinDLL, WinError, get_last_error, wintypes # ty: ignore[unresolved-import] # Windows-only APIs +from time import monotonic +from typing import TYPE_CHECKING, Final + +if TYPE_CHECKING: + from threading import Thread + + +def stop_query_reader(reader: Thread) -> None: + deadline = monotonic() + 1 + while reader.is_alive() and monotonic() < deadline: + if (thread_id := reader.native_id) is not None: + try: + _cancel_synchronous_io(thread_id) + except OSError: + if reader.is_alive(): + raise + reader.join(0.05) + if reader.is_alive(): + msg = "query output reader did not stop" + raise TimeoutError(msg) + + +def _cancel_synchronous_io(thread_id: int) -> None: + error_not_found: Final[int] = 1168 + kernel = WinDLL("kernel32", use_last_error=True) + open_thread = kernel.OpenThread + open_thread.argtypes = (wintypes.DWORD, wintypes.BOOL, wintypes.DWORD) + open_thread.restype = wintypes.HANDLE + cancel = kernel.CancelSynchronousIo + cancel.argtypes = (wintypes.HANDLE,) + cancel.restype = wintypes.BOOL + close = kernel.CloseHandle + close.argtypes = (wintypes.HANDLE,) + close.restype = wintypes.BOOL + + if not (handle := open_thread(0x0001, wintypes.BOOL(0), thread_id)): + raise WinError(get_last_error()) + try: + if not cancel(handle) and (error := get_last_error()) != error_not_found: + raise WinError(error) + finally: + if not close(handle): + raise WinError(get_last_error()) + + +__all__ = ["stop_query_reader"] diff --git a/tests/test_cached_py_info.py b/tests/test_cached_py_info.py index 93dba45..9487f9a 100644 --- a/tests/test_cached_py_info.py +++ b/tests/test_cached_py_info.py @@ -8,6 +8,8 @@ from io import StringIO from pathlib import Path from subprocess import DEVNULL, PIPE, Popen, TimeoutExpired, list2cmdline +from threading import enumerate as active_threads +from time import monotonic from typing import TYPE_CHECKING, Final from unittest.mock import MagicMock, patch @@ -28,6 +30,8 @@ ) if TYPE_CHECKING: + from typing import IO + from pytest_mock import MockerFixture @@ -97,7 +101,13 @@ def test_run_subprocess_invalid_json(mocker: MockerFixture) -> None: def _mock_process(out: str, err: str, code: int) -> MagicMock: - return MagicMock(stdout=StringIO(out), stderr=StringIO(err), returncode=code, **{"wait.return_value": code}) + streams: list[IO[str]] = [] + for value in (out, err): + read_fd, write_fd = os.pipe() + with os.fdopen(write_fd, "wb") as writer: + writer.write(value.encode()) + streams.append(os.fdopen(read_fd, "r", encoding="utf-8")) + return MagicMock(stdout=streams[0], stderr=streams[1], returncode=code, **{"wait.return_value": code}) def test_run_subprocess_with_cookies(mocker: MockerFixture) -> None: @@ -161,6 +171,10 @@ def test_collect_query_output_limit(stream: str, timeout: float | None) -> None: ) +def test_collect_query_output_normalizes_trailing_newline() -> None: + assert _collect_query_output(_mock_process("\r", "", 0), 1) == ("\n", "", 0) + + @pytest.mark.parametrize( "code", [ @@ -173,25 +187,41 @@ def test_collect_query_output_timeout(code: str) -> None: assert _collect_query_output(process, 0.1) == ("", "timed out", -1) -def test_collect_query_output_timeout_after_eof(mocker: MockerFixture) -> None: - process = mocker.MagicMock( - stdout=StringIO(""), - stderr=StringIO(""), - **{"wait.side_effect": [TimeoutExpired("python", 0), 0]}, +def test_collect_query_output_timeout_reaps_readers() -> None: + code = ( + "import subprocess, sys\n" + "subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(10)'], " + "stdin=subprocess.DEVNULL, stdout=sys.stdout, stderr=sys.stderr)\n" ) + process = Popen([sys.executable, "-c", code], stdin=DEVNULL, stdout=PIPE, stderr=PIPE, text=True) + process.wait(timeout=5) + before = set(active_threads()) + started = monotonic() + + assert _collect_query_output(process, 0.1) == ("", "timed out", -1) + assert monotonic() - started < 2.5 + assert not set(active_threads()) - before + + +def test_collect_query_output_timeout_after_eof(mocker: MockerFixture) -> None: + process = Popen([sys.executable, "-c", ""], stdin=DEVNULL, stdout=PIPE, stderr=PIPE, text=True) + process.wait(timeout=5) + mocker.patch.object(process, "wait", side_effect=[TimeoutExpired("python", 0), 0]) + mocker.patch.object(process, "kill") assert _collect_query_output(process, 1) == ("", "timed out", -1) process.kill.assert_called_once() def test_collect_query_output_read_error(mocker: MockerFixture) -> None: - stream = mocker.MagicMock() - stream.read.side_effect = OSError("broken pipe") - process = mocker.MagicMock(stdout=stream, stderr=StringIO(""), **{"wait.return_value": -9}) + process = Popen([sys.executable, "-c", ""], stdin=DEVNULL, stdout=PIPE, stderr=PIPE, text=True) + process.wait(timeout=5) + mocker.patch("python_discovery._cached_py_info.os.read", side_effect=OSError("broken pipe")) + mocker.patch.object(process, "kill") with pytest.raises(OSError, match="broken pipe"): _collect_query_output(process, 1) - process.kill.assert_called_once() + process.kill.assert_called() @pytest.mark.parametrize( From a04c74e2ca663df2fd874bc197b623f48c35e53d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bern=C3=A1t=20G=C3=A1bor?= Date: Mon, 2 Nov 2026 19:09:01 -0800 Subject: [PATCH 3/3] =?UTF-8?q?=F0=9F=A7=AA=20test(query):=20cover=20split?= =?UTF-8?q?=20UTF-8=20reads?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Force one-byte pipe reads so the test exercises a decoder that yields no character until the remaining bytes arrive. Drop the explicit continue to keep Python 3.9 coverage at 100%. --- src/python_discovery/_cached_py_info.py | 15 ++++++--------- tests/test_cached_py_info.py | 6 ++++++ 2 files changed, 12 insertions(+), 9 deletions(-) diff --git a/src/python_discovery/_cached_py_info.py b/src/python_discovery/_cached_py_info.py index b5324d5..aae2d85 100644 --- a/src/python_discovery/_cached_py_info.py +++ b/src/python_discovery/_cached_py_info.py @@ -366,16 +366,13 @@ def _drain_query_stream(state: _QueryReadState, stream: IO[str], chunks: list[st failed = True break chunk = decoder.decode(data, final=not data) - if not chunk: - if not data: + if chunk: + chunks.append(chunk[: _MAX_QUERY_OUTPUT + 1 - size]) + size += len(chunk) + if size > _MAX_QUERY_OUTPUT: + state.overflow.set() + failed = True break - continue - chunks.append(chunk[: _MAX_QUERY_OUTPUT + 1 - size]) - size += len(chunk) - if size > _MAX_QUERY_OUTPUT: - state.overflow.set() - failed = True - break if not data: break if failed: diff --git a/tests/test_cached_py_info.py b/tests/test_cached_py_info.py index 9487f9a..ae40d59 100644 --- a/tests/test_cached_py_info.py +++ b/tests/test_cached_py_info.py @@ -6,6 +6,7 @@ import sys import tempfile from io import StringIO +from os import read from pathlib import Path from subprocess import DEVNULL, PIPE, Popen, TimeoutExpired, list2cmdline from threading import enumerate as active_threads @@ -175,6 +176,11 @@ def test_collect_query_output_normalizes_trailing_newline() -> None: assert _collect_query_output(_mock_process("\r", "", 0), 1) == ("\n", "", 0) +def test_collect_query_output_decodes_split_utf8() -> None: + with patch("python_discovery._cached_py_info.os.read", side_effect=lambda fd, _: read(fd, 1)): + assert _collect_query_output(_mock_process("é", "", 0), 1) == ("é", "", 0) + + @pytest.mark.parametrize( "code", [