diff --git a/docs/changelog/+probe-descendant-pipes.bugfix.rst b/docs/changelog/+probe-descendant-pipes.bugfix.rst new file mode 100644 index 0000000..7b93990 --- /dev/null +++ b/docs/changelog/+probe-descendant-pipes.bugfix.rst @@ -0,0 +1 @@ +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 bcade0d..aae2d85 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,29 +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() - for reader in readers: - reader.join() + # 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: @@ -328,30 +350,34 @@ 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 - if not chunk: - break - chunks.append(chunk[: _MAX_QUERY_OUTPUT + 1 - size]) - size += len(chunk) - if size > _MAX_QUERY_OUTPUT: - overflow.set() - failed = True + chunk = decoder.decode(data, final=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 + 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..ae40d59 100644 --- a/tests/test_cached_py_info.py +++ b/tests/test_cached_py_info.py @@ -6,8 +6,11 @@ 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 +from time import monotonic from typing import TYPE_CHECKING, Final from unittest.mock import MagicMock, patch @@ -28,6 +31,8 @@ ) if TYPE_CHECKING: + from typing import IO + from pytest_mock import MockerFixture @@ -97,7 +102,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 +172,15 @@ 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) + + +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", [ @@ -173,25 +193,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( 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