diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 46043bf..c578bb2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -18,6 +18,7 @@ jobs: - run: uv run ruff check . - run: uv run ruff format --check . - run: uv run mypy + - run: uv run mypy --platform win32 test: strategy: @@ -50,13 +51,12 @@ jobs: if: runner.os == 'Linux' run: QSV_BIN="$RUNNER_TEMP/qsv/qsvdp" uv run pytest -q tests/test_integration.py - # Integration tests only: the unit tests' fake qsv relies on a shebang and skips on Windows, - # so process-group cleanup is not exercised here. + # Separate from `test` because Windows steps run in PowerShell and use the MSVC build. test-windows: strategy: fail-fast: false matrix: - python: ["3.10", "3.14"] + python: ["3.10", "3.11", "3.12", "3.13", "3.14"] runs-on: windows-latest defaults: run: @@ -76,10 +76,10 @@ jobs: gh release download --repo dathere/qsv --pattern "qsv-*-x86_64-pc-windows-msvc.zip" --dir $dir Get-ChildItem $dir -Filter *.zip | ForEach-Object { Expand-Archive $_.FullName -DestinationPath $dir -Force } Get-ChildItem $dir - - name: Integration tests (qsv) + - name: Unit tests and integration tests (qsv) run: | $env:QSV_BIN = Join-Path $env:RUNNER_TEMP "qsv\qsv.exe" - uv run pytest -v tests/test_integration.py + uv run pytest -v - name: Integration tests (qsvlite) run: | $env:QSV_BIN = Join-Path $env:RUNNER_TEMP "qsv\qsvlite.exe" diff --git a/CHANGELOG.md b/CHANGELOG.md index bba0f8f..e6ce86b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,14 @@ All notable changes to this project are documented here. The format follows ## [Unreleased] +### Added + +- Windows support. qsv is started suspended and placed in its own Job Object before it runs, + so a timeout or cancellation stops qsv *and* everything it started: `CTRL_BREAK_EVENT` + first, then `TerminateJobObject` after `kill_grace`. Previously only `qsv.exe` itself was + killed. With `inherit_env=False`, `SYSTEMROOT` is still passed on Windows, since many + programs fail to start without it. CI now runs the full test suite on Windows. + ### Changed - The sdist now contains only the package, tests, README, CHANGELOG and LICENSE. 0.1.0's also diff --git a/CLAUDE.md b/CLAUDE.md index f1aa58d..2830947 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -15,7 +15,9 @@ uv run ruff format . && uv run ruff check . && uv run mypy # must be CI runs lint, format check, and mypy, plus tests on ubuntu/macos x Python 3.10–3.14 against qsv, qsvlite, and qsvdp (Linux only). -On Windows (Python 3.10 and 3.14) it runs only the integration tests, against qsv and qsvlite. +On Windows (Python 3.10–3.14) it runs the full suite against qsv, plus the integration tests +against qsvlite. Windows teardown uses Job Objects via ctypes (`_ProcessTree` in `_process.py`); +type-check it with `uv run mypy --platform win32` (the lint job runs both). ## Layout (`src/qsv_client/`) @@ -32,8 +34,9 @@ On Windows (Python 3.10 and 3.14) it runs only the integration tests, against qs - `_Base._build_env` always sets `QSV_ERROR_FORMAT=json`. Older qsv versions ignore it, so their errors are classified by exit code. - Exit codes 0 and 255 (qsv's "warning") count as success. -- On timeout, cancellation, or KeyboardInterrupt, the whole process group is killed: SIGTERM, - then SIGKILL after `kill_grace`. A timeout raises `QsvTimeout` (exit code 124) even when +- On timeout, cancellation, or KeyboardInterrupt, the run's whole process tree is killed after + `kill_grace`: SIGTERM then SIGKILL to the process group on POSIX, `CTRL_BREAK_EVENT` then + `TerminateJobObject` on Windows. A timeout raises `QsvTimeout` (exit code 124) even when `check=False`. - LLM settings for describegpt are passed as env vars (`QSV_LLM_*`), never on argv. - `min_version` is enforced inside `_probe`, so it holds however the capabilities cache is diff --git a/README.md b/README.md index 0c23bac..f1dd20d 100644 --- a/README.md +++ b/README.md @@ -118,8 +118,15 @@ runs. ## Platform notes -Process-group cleanup uses POSIX sessions (Linux, macOS). On Windows, qsv is started in a new -process group and killed directly; its own children are not reaped. +On a timeout or cancellation, the client stops qsv and every process qsv started: + +- **Linux, macOS:** each run gets its own session. The client sends SIGTERM to the process + group, then SIGKILL after `kill_grace`. A process that leaves the group (by calling `setsid`) + survives. +- **Windows:** qsv is started suspended, put in its own Job Object, and only then resumed, so + every process it starts is in the job and can't leave it. The client sends + `CTRL_BREAK_EVENT` to the run's process group, then calls `TerminateJobObject` after + `kill_grace`. ## Development diff --git a/pyproject.toml b/pyproject.toml index 501a654..0f3b676 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,6 +17,7 @@ classifiers = [ "Intended Audience :: Developers", "Operating System :: POSIX", "Operating System :: MacOS", + "Operating System :: Microsoft :: Windows", "Programming Language :: Python :: 3", "Programming Language :: Python :: 3 :: Only", "Topic :: Scientific/Engineering :: Information Analysis", diff --git a/src/qsv_client/_process.py b/src/qsv_client/_process.py index 8cec986..3998be9 100644 --- a/src/qsv_client/_process.py +++ b/src/qsv_client/_process.py @@ -1,14 +1,17 @@ """Spawn qsv and reap it, including any children it started, on timeout or cancellation. qsv commands can start their own children (``validate`` runs pyshacl, ``viz`` a webdriver, -``describegpt`` may shell out). Killing only the qsv pid would orphan them, so on POSIX each -run gets its own session (process group) and the whole group is signalled: SIGTERM first, -then SIGKILL after a grace period. On Windows the process is started in a new process group -and killed directly; grandchildren are not reaped there. - -A descendant that leaves the group (e.g. by calling ``setsid``) survives the kill and may keep -the output pipes open; after a timeout, output is collected for at most ``kill_grace`` seconds -and anything that descendant still holds is dropped. +``describegpt`` may shell out). Killing only the qsv pid would orphan them, so each run's +process tree is tracked and stopped as a whole: first politely, then forcibly after a grace +period. + +- POSIX: each run gets its own session (process group). SIGTERM, then SIGKILL, to the group. + A descendant that leaves the group (e.g. by calling ``setsid``) survives the kill and may + keep the output pipes open; after a timeout, output is collected for at most ``kill_grace`` + seconds and anything that descendant still holds is dropped. +- Windows: qsv is created suspended, put in its own Job Object, then resumed, so every + process it starts is in the job and cannot leave it. ``CTRL_BREAK_EVENT`` to its process + group, then ``TerminateJobObject``. """ from __future__ import annotations @@ -19,6 +22,7 @@ import signal import subprocess import sys +import threading from collections.abc import Coroutine, Mapping, Sequence from dataclasses import dataclass from typing import IO, Any, TypeVar @@ -36,27 +40,192 @@ class RawResult: timed_out: bool -def _spawn_kwargs() -> dict[str, Any]: - if IS_WINDOWS: - return {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP} # type: ignore[attr-defined,unused-ignore] - return {"start_new_session": True} - - -def _signal_group(pid: int, sig: int) -> None: - with contextlib.suppress(ProcessLookupError, PermissionError): - os.killpg(pid, sig) # type: ignore[attr-defined,unused-ignore] +if sys.platform == "win32": + import ctypes + from ctypes import wintypes + + _kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + _kernel32.CreateJobObjectW.argtypes = (wintypes.LPVOID, wintypes.LPCWSTR) + _kernel32.CreateJobObjectW.restype = wintypes.HANDLE + _kernel32.OpenProcess.argtypes = (wintypes.DWORD, wintypes.BOOL, wintypes.DWORD) + _kernel32.OpenProcess.restype = wintypes.HANDLE + _kernel32.AssignProcessToJobObject.argtypes = (wintypes.HANDLE, wintypes.HANDLE) + _kernel32.AssignProcessToJobObject.restype = wintypes.BOOL + _kernel32.TerminateJobObject.argtypes = (wintypes.HANDLE, wintypes.UINT) + _kernel32.TerminateJobObject.restype = wintypes.BOOL + _kernel32.CloseHandle.argtypes = (wintypes.HANDLE,) + _kernel32.CloseHandle.restype = wintypes.BOOL + + class _THREADENTRY32(ctypes.Structure): + _fields_ = ( + ("dwSize", wintypes.DWORD), + ("cntUsage", wintypes.DWORD), + ("th32ThreadID", wintypes.DWORD), + ("th32OwnerProcessID", wintypes.DWORD), + ("tpBasePri", wintypes.LONG), + ("tpDeltaPri", wintypes.LONG), + ("dwFlags", wintypes.DWORD), + ) + + _kernel32.CreateToolhelp32Snapshot.argtypes = (wintypes.DWORD, wintypes.DWORD) + _kernel32.CreateToolhelp32Snapshot.restype = wintypes.HANDLE + _kernel32.Thread32First.argtypes = (wintypes.HANDLE, ctypes.POINTER(_THREADENTRY32)) + _kernel32.Thread32First.restype = wintypes.BOOL + _kernel32.Thread32Next.argtypes = (wintypes.HANDLE, ctypes.POINTER(_THREADENTRY32)) + _kernel32.Thread32Next.restype = wintypes.BOOL + _kernel32.OpenThread.argtypes = (wintypes.DWORD, wintypes.BOOL, wintypes.DWORD) + _kernel32.OpenThread.restype = wintypes.HANDLE + _kernel32.ResumeThread.argtypes = (wintypes.HANDLE,) + _kernel32.ResumeThread.restype = wintypes.DWORD + + _CREATE_SUSPENDED = 0x00000004 + _PROCESS_TERMINATE = 0x0001 + _PROCESS_SET_QUOTA = 0x0100 + _TH32CS_SNAPTHREAD = 0x00000004 + _THREAD_SUSPEND_RESUME = 0x0002 + _INVALID_HANDLE_VALUE = ctypes.c_void_p(-1).value + _RESUME_FAILED = 0xFFFFFFFF + + def _spawn_kwargs() -> dict[str, Any]: + # its own console process group, so CTRL_BREAK_EVENT reaches qsv and its children + # only; suspended, so it can start nothing before it is in its Job Object + return {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP | _CREATE_SUSPENDED} + + def _resume(pid: int) -> bool: + """Resume the threads of a process created with CREATE_SUSPENDED. + + Popen closes the main thread's handle, so find it through a Toolhelp snapshot. + """ + snapshot = _kernel32.CreateToolhelp32Snapshot(_TH32CS_SNAPTHREAD, 0) + if not snapshot or snapshot == _INVALID_HANDLE_VALUE: + return False + resumed = False + try: + entry = _THREADENTRY32() + entry.dwSize = ctypes.sizeof(entry) + more = _kernel32.Thread32First(snapshot, ctypes.byref(entry)) + while more: + if entry.th32OwnerProcessID == pid: + thread = _kernel32.OpenThread(_THREAD_SUSPEND_RESUME, False, entry.th32ThreadID) + if thread: + try: + if _kernel32.ResumeThread(thread) != _RESUME_FAILED: + resumed = True + finally: + _kernel32.CloseHandle(thread) + more = _kernel32.Thread32Next(snapshot, ctypes.byref(entry)) + finally: + _kernel32.CloseHandle(snapshot) + return resumed + + def _create_job(pid: int) -> int | None: + """A Job Object holding ``pid`` (and, from now on, everything it starts), or None.""" + job = _kernel32.CreateJobObjectW(None, None) + if not job: + return None + handle = _kernel32.OpenProcess(_PROCESS_SET_QUOTA | _PROCESS_TERMINATE, False, pid) + try: + if handle and _kernel32.AssignProcessToJobObject(job, handle): + return int(job) + finally: + if handle: + _kernel32.CloseHandle(handle) + _kernel32.CloseHandle(job) + return None + + class _ProcessTree: + def __init__(self, pid: int) -> None: + """Put the (suspended) process in a new Job Object, then let it run.""" + self.pid = pid + self._job: int | None = None + try: + self._job = _create_job(pid) + if not _resume(pid): + raise OSError(f"could not start qsv (pid {pid}): resuming its thread failed") + except BaseException: + # failed or interrupted (Ctrl-C): never leave it suspended, or running untracked + self.kill() + self.close() + raise + + def interrupt(self) -> None: + with contextlib.suppress(OSError): + os.kill(self.pid, signal.CTRL_BREAK_EVENT) + + def kill(self) -> None: + if self._job is not None and _kernel32.TerminateJobObject(self._job, 1): + return + # no job (creation failed): the leader is all we can reach. os.kill on Windows + # is TerminateProcess for anything but the console events. + with contextlib.suppress(OSError): + os.kill(self.pid, signal.SIGTERM) + + def close(self) -> None: + if self._job is not None: + _kernel32.CloseHandle(self._job) + self._job = None + +else: + + def _spawn_kwargs() -> dict[str, Any]: + return {"start_new_session": True} + + def _signal_group(pid: int, sig: int) -> None: + with contextlib.suppress(ProcessLookupError, PermissionError): + os.killpg(pid, sig) + + class _ProcessTree: + def __init__(self, pid: int) -> None: + self.pid = pid + + def interrupt(self) -> None: + _signal_group(self.pid, signal.SIGTERM) + + def kill(self) -> None: + # the whole group, even if the leader already exited: its children may not have + _signal_group(self.pid, signal.SIGKILL) + + def close(self) -> None: + pass + + +def _terminate_sync(proc: subprocess.Popen[bytes], tree: _ProcessTree, grace: float) -> None: + try: + tree.interrupt() + with contextlib.suppress(subprocess.TimeoutExpired): + proc.wait(timeout=grace) + finally: + # even when a second Ctrl-C interrupts the grace wait; then reap the leader, which the + # forced kill ends at once, so it is not left behind as a zombie + tree.kill() + with contextlib.suppress(subprocess.TimeoutExpired): + proc.wait(timeout=grace) + + +def _feed_stdin_in_thread(proc: subprocess.Popen[bytes], data: bytes) -> threading.Thread: + """Write ``data`` to the child's stdin from a thread, then close it. + + Before Python 3.13, ``communicate()`` on Windows writes stdin synchronously before it + checks its timeout, so a child that stops reading would block it forever. Detaching stdin + leaves ``communicate()`` only the reads, which do honor the timeout; killing the tree + breaks the pipe and ends this thread. + """ + pipe = proc.stdin + def feed() -> None: + assert pipe is not None + try: + with contextlib.suppress(OSError): + pipe.write(data) + finally: + with contextlib.suppress(OSError): + pipe.close() -def _terminate_sync(proc: subprocess.Popen[bytes], grace: float) -> None: - if IS_WINDOWS: - with contextlib.suppress(OSError): - proc.kill() - return - _signal_group(proc.pid, signal.SIGTERM) - with contextlib.suppress(subprocess.TimeoutExpired): - proc.wait(timeout=grace) - # SIGKILL the group even if the leader already exited: its children may not have - _signal_group(proc.pid, signal.SIGKILL) # type: ignore[attr-defined,unused-ignore] + thread = threading.Thread(target=feed, name="qsv-stdin", daemon=True) + thread.start() + # only now: if start() failed, the caller still owns stdin and closes it + proc.stdin = None + return thread def run_sync( @@ -78,12 +247,24 @@ def run_sync( cwd=cwd, **_spawn_kwargs(), ) + try: + tree = _ProcessTree(proc.pid) + except BaseException: + # failed or interrupted (Ctrl-C) before there is a tree to tear down: _ProcessTree + # killed what it could; make sure of the leader, reap it and close its pipes + with contextlib.suppress(OSError): + proc.kill() + proc.communicate() + raise + feeder = None timed_out = False try: + if IS_WINDOWS and stdin is not None: + feeder, stdin = _feed_stdin_in_thread(proc, stdin), None out, err = proc.communicate(input=stdin, timeout=timeout) except subprocess.TimeoutExpired: timed_out = True - _terminate_sync(proc, kill_grace) + _terminate_sync(proc, tree, kill_grace) try: out, err = proc.communicate(timeout=kill_grace) except subprocess.TimeoutExpired: @@ -94,10 +275,19 @@ def run_sync( proc.wait() out, err = b"", b"" except BaseException: - # KeyboardInterrupt etc.: never leave a qsv process group running behind us - _terminate_sync(proc, kill_grace) + # KeyboardInterrupt etc.: never leave a qsv process tree running behind us + _terminate_sync(proc, tree, kill_grace) proc.wait() raise + finally: + tree.close() + if feeder is not None: + feeder.join(kill_grace) + # communicate() closes them when it finishes; an exception escaping first would not + for pipe in (proc.stdin, proc.stdout, proc.stderr): + if pipe is not None and not pipe.closed: + with contextlib.suppress(OSError): + pipe.close() return RawResult( exit_code=proc.returncode, stdout=out or b"", @@ -106,19 +296,15 @@ def run_sync( ) -async def _terminate_async(proc: asyncio.subprocess.Process, grace: float) -> None: - if IS_WINDOWS: - with contextlib.suppress(ProcessLookupError): - proc.kill() - with contextlib.suppress(asyncio.TimeoutError): - await asyncio.wait_for(proc.wait(), grace) - return +async def _terminate_async( + proc: asyncio.subprocess.Process, tree: _ProcessTree, grace: float +) -> None: # proc.wait() also waits for the output pipes to close, which a descendant that left the # group can hold open indefinitely, so every wait here is bounded - _signal_group(proc.pid, signal.SIGTERM) + tree.interrupt() with contextlib.suppress(asyncio.TimeoutError): await asyncio.wait_for(proc.wait(), grace) - _signal_group(proc.pid, signal.SIGKILL) # type: ignore[attr-defined,unused-ignore] + tree.kill() with contextlib.suppress(asyncio.TimeoutError): await asyncio.wait_for(proc.wait(), grace) @@ -139,37 +325,35 @@ def _close_pipes(proc: asyncio.subprocess.Process) -> None: transport.close() -def _kill_now(proc: asyncio.subprocess.Process) -> None: +def _kill_now(proc: asyncio.subprocess.Process, tree: _ProcessTree) -> None: """Teardown was itself cancelled (e.g. by ``asyncio.run()`` shutting down): skip the remaining grace and do the parts that matter synchronously.""" - if IS_WINDOWS: - with contextlib.suppress(ProcessLookupError): - proc.kill() - else: - _signal_group(proc.pid, signal.SIGKILL) # type: ignore[attr-defined,unused-ignore] + tree.kill() _close_pipes(proc) -async def _reap_async(proc: asyncio.subprocess.Process, grace: float) -> None: +async def _reap_async(proc: asyncio.subprocess.Process, tree: _ProcessTree, grace: float) -> None: try: - await _terminate_async(proc, grace) + await _terminate_async(proc, tree, grace) await _release_async(proc) except asyncio.CancelledError: - _kill_now(proc) + _kill_now(proc, tree) raise -async def _timeout_cleanup(proc: asyncio.subprocess.Process, grace: float) -> bytes: +async def _timeout_cleanup( + proc: asyncio.subprocess.Process, tree: _ProcessTree, grace: float +) -> bytes: """Kill and release a timed-out run; return whatever stderr arrives within ``grace``.""" try: - await _terminate_async(proc, grace) + await _terminate_async(proc, tree, grace) err = b"" if proc.stderr is not None: with contextlib.suppress(Exception): err = await asyncio.wait_for(proc.stderr.read(), grace) await _release_async(proc) except asyncio.CancelledError: - _kill_now(proc) + _kill_now(proc, tree) raise return err @@ -179,7 +363,7 @@ async def _finish_despite_cancel(coro: Coroutine[Any, Any, _T]) -> _T: ``asyncio.shield`` alone would propagate the cancellation at once, leaving the cleanup in a background task that an ``asyncio.run()`` returning right after would cancel, skipping - the SIGKILL. + the forced kill. """ task = asyncio.ensure_future(coro) cancel: asyncio.CancelledError | None = None @@ -212,17 +396,27 @@ async def run_async( cwd=cwd, **_spawn_kwargs(), ) + try: + tree = _ProcessTree(proc.pid) + except BaseException: + # failed or interrupted before there is a tree to tear down (see run_sync) + with contextlib.suppress(ProcessLookupError): + proc.kill() + await proc.communicate() + raise timed_out = False try: out, err = await asyncio.wait_for(proc.communicate(input=stdin), timeout) except asyncio.TimeoutError: timed_out = True - # a cancellation arriving mid-cleanup must not skip the SIGKILL or the release - out, err = b"", await _finish_despite_cancel(_timeout_cleanup(proc, kill_grace)) + # a cancellation arriving mid-cleanup must not skip the forced kill or the release + out, err = b"", await _finish_despite_cancel(_timeout_cleanup(proc, tree, kill_grace)) except BaseException: - # task cancelled (or the loop is shutting down): reap the group, then propagate - await _finish_despite_cancel(_reap_async(proc, kill_grace)) + # task cancelled (or the loop is shutting down): reap the tree, then propagate + await _finish_despite_cancel(_reap_async(proc, tree, kill_grace)) raise + finally: + tree.close() assert proc.returncode is not None return RawResult( exit_code=proc.returncode, diff --git a/src/qsv_client/client.py b/src/qsv_client/client.py index 9a8d64d..dd5515d 100644 --- a/src/qsv_client/client.py +++ b/src/qsv_client/client.py @@ -118,13 +118,15 @@ def __init__( whole process group is sent SIGTERM, then SIGKILL after ``kill_grace`` seconds, and :class:`QsvTimeout` is raised. env: extra environment variables for every run. - inherit_env: start from ``os.environ`` (True) or from an empty environment. + inherit_env: start from ``os.environ`` (True) or from an empty environment + (on Windows, one holding only ``SYSTEMROOT``). cwd: working directory for every run. llm_api_key / llm_base_url / llm_model: passed to ``describegpt`` as ``QSV_LLM_APIKEY`` / ``QSV_LLM_BASE_URL`` / ``QSV_LLM_MODEL`` environment variables, so the key never appears in the process table. min_version: raise :class:`QsvVersionError` on first use if the binary is older. - kill_grace: seconds to wait after SIGTERM before SIGKILL on timeout or + kill_grace: seconds to wait between the polite stop (SIGTERM, or + ``CTRL_BREAK_EVENT`` on Windows) and the forced kill on timeout or cancellation, and the most spent collecting output after the kill. """ self.binary = find_qsv(binary) @@ -148,6 +150,9 @@ def __repr__(self) -> str: def _build_env(self, extra: Mapping[str, str] | None) -> dict[str, str]: env = dict(os.environ) if self.inherit_env else {} + if not self.inherit_env and _process.IS_WINDOWS and "SYSTEMROOT" in os.environ: + # Windows processes (the C runtime, Winsock) can fail to start without it + env["SYSTEMROOT"] = os.environ["SYSTEMROOT"] env.update(self._env) if extra: env.update(extra) diff --git a/tests/conftest.py b/tests/conftest.py index 10c9454..b9b4e12 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -40,13 +40,23 @@ def csv_files(tmp_path: Path) -> Path: @pytest.fixture def fake_qsv(tmp_path: Path) -> Callable[[str], str]: - """Write an executable stand-in for qsv running the given Python body; return its path.""" - if sys.platform == "win32": - pytest.skip("fake executables use a shebang") + """Write an executable stand-in for qsv running the given Python body; return its path. + + POSIX: a script with a shebang. Windows: the script plus a ``fake-qsv.cmd`` wrapper that + runs it (the extra ``cmd.exe`` is inside the run's Job Object, so it is killed with it). + """ + # qsv writes "\n" on every platform; without this, print() writes "\r\n" on Windows + prelude = 'import sys; sys.stdout.reconfigure(newline="\\n")\n' def make(body: str) -> str: + script = prelude + textwrap.dedent(body) + if sys.platform == "win32": + (tmp_path / "fake-qsv.py").write_text(script) + wrapper = tmp_path / "fake-qsv.cmd" + wrapper.write_text(f'@"{sys.executable}" "%~dp0fake-qsv.py" %*\n') + return str(wrapper) path = tmp_path / "fake-qsv" - path.write_text(f"#!{sys.executable}\n" + textwrap.dedent(body)) + path.write_text(f"#!{sys.executable}\n" + script) path.chmod(path.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) return str(path) diff --git a/tests/test_process.py b/tests/test_process.py index 9f1532d..5f350e0 100644 --- a/tests/test_process.py +++ b/tests/test_process.py @@ -6,6 +6,8 @@ import contextlib import os import signal +import sys +import threading import time from collections.abc import Callable from pathlib import Path @@ -13,7 +15,7 @@ import pytest -from qsv_client import AsyncQsv, Qsv, QsvTimeout, QsvUsageError +from qsv_client import AsyncQsv, Qsv, QsvTimeout, QsvUsageError, _process # The fake starts a grandchild that inherits stdout and records its pid, then hangs. If only # the fake were killed, the grandchild would keep the stdout pipe open and the run would hang @@ -37,14 +39,46 @@ def _is_zombie(pid: int) -> bool: return stat.rsplit(")", 1)[1].split()[0] == "Z" -def _wait_dead(pid: int, within: float = 5.0) -> bool: - deadline = time.monotonic() + within - while time.monotonic() < deadline: +if sys.platform == "win32": + import ctypes + from ctypes import wintypes + + _k32 = ctypes.WinDLL("kernel32", use_last_error=True) + _k32.OpenProcess.argtypes = (wintypes.DWORD, wintypes.BOOL, wintypes.DWORD) + _k32.OpenProcess.restype = wintypes.HANDLE + _k32.GetExitCodeProcess.argtypes = (wintypes.HANDLE, ctypes.POINTER(wintypes.DWORD)) + _k32.GetExitCodeProcess.restype = wintypes.BOOL + _k32.CloseHandle.argtypes = (wintypes.HANDLE,) + _k32.CloseHandle.restype = wintypes.BOOL + _PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 + _STILL_ACTIVE = 259 + + def _alive(pid: int) -> bool: + # not os.kill(pid, 0): on Windows that is TerminateProcess + handle = _k32.OpenProcess(_PROCESS_QUERY_LIMITED_INFORMATION, False, pid) + if not handle: + return False + try: + code = wintypes.DWORD() + ok = _k32.GetExitCodeProcess(handle, ctypes.byref(code)) + return bool(ok) and code.value == _STILL_ACTIVE + finally: + _k32.CloseHandle(handle) + +else: + + def _alive(pid: int) -> bool: try: os.kill(pid, 0) except ProcessLookupError: - return True - if _is_zombie(pid): + return False + return not _is_zombie(pid) + + +def _wait_dead(pid: int, within: float = 5.0) -> bool: + deadline = time.monotonic() + within + while time.monotonic() < deadline: + if not _alive(pid): return True time.sleep(0.05) return False @@ -162,18 +196,36 @@ def test_warning_exit_code_is_success(fake_qsv: Callable[[str], str]) -> None: # Like HANG_WITH_GRANDCHILD, but the grandchild leaves the process group, so the group kill # misses it and it keeps the output pipes open. +# ignore the polite stop: SIGTERM on POSIX, CTRL_BREAK_EVENT (SIGBREAK) on Windows +IGNORE_STOP = """ +import signal +signal.signal(signal.SIGTERM, signal.SIG_IGN) +if hasattr(signal, "SIGBREAK"): + signal.signal(signal.SIGBREAK, signal.SIG_IGN) +""" + HANG_WITH_ESCAPED_GRANDCHILD = """ import subprocess, sys, time -child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"], - start_new_session=True) +# POSIX: its own session, out of the group kill's reach. Windows: its own console process +# group, out of CTRL_BREAK_EVENT's reach (but still inside the run's Job Object). +escape = ({"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP} if sys.platform == "win32" + else {"start_new_session": True}) +child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"], **escape) open(sys.argv[-1], "w").write(str(child.pid)) time.sleep(60) """ def _kill(pid: int) -> None: - with contextlib.suppress(ProcessLookupError): - os.kill(pid, signal.SIGKILL) + # on Windows, os.kill with anything but the console events is TerminateProcess + with contextlib.suppress(OSError): + os.kill(pid, getattr(signal, "SIGKILL", signal.SIGTERM)) + + +def _check_job_killed(pid: int) -> None: + """On Windows nothing escapes the Job Object, so the 'escaped' grandchild must be dead.""" + if sys.platform == "win32": + assert _wait_dead(pid), "the Job Object missed a grandchild in its own process group" def test_timeout_bounded_when_grandchild_escapes_group( @@ -186,6 +238,7 @@ def test_timeout_bounded_when_grandchild_escapes_group( with pytest.raises(QsvTimeout): qsv.run("stats", pidfile, timeout=1) assert time.monotonic() - started < 10 + _check_job_killed(_read_pid(pidfile)) finally: _kill(_read_pid(pidfile)) @@ -200,6 +253,7 @@ async def test_async_timeout_bounded_when_grandchild_escapes_group( with pytest.raises(QsvTimeout): await qsv.run("stats", pidfile, timeout=1) assert time.monotonic() - started < 10 + _check_job_killed(_read_pid(pidfile)) finally: _kill(_read_pid(pidfile)) @@ -246,30 +300,31 @@ def test_relative_binary_with_cwd( assert Qsv(f"./{fake.name}", cwd=other).run("x").stdout == "ok\n" +@pytest.mark.skipif(sys.platform == "win32", reason="Win32 collapses `..` before symlinks") def test_relative_binary_through_symlinked_dir( fake_qsv: Callable[[str], str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: # work/link -> bin, so work/link/../fake-qsv is tmp_path/fake-qsv, not work/fake-qsv - fake_qsv("print('ok')") + fake = Path(fake_qsv("print('ok')")) (tmp_path / "bin").mkdir() work = tmp_path / "work" work.mkdir() (work / "link").symlink_to(tmp_path / "bin") monkeypatch.chdir(work) - assert Qsv("link/../fake-qsv").run("x").stdout == "ok\n" + assert Qsv(f"link/../{fake.name}").run("x").stdout == "ok\n" async def test_async_timeout_bounded_when_leader_ignores_sigterm_and_grandchild_escapes( fake_qsv: Callable[[str], str], tmp_path: Path ) -> None: pidfile = tmp_path / "child.pid" - body = "import signal; signal.signal(signal.SIGTERM, signal.SIG_IGN)\n" - qsv = AsyncQsv(fake_qsv(body + HANG_WITH_ESCAPED_GRANDCHILD), kill_grace=0.5) + qsv = AsyncQsv(fake_qsv(IGNORE_STOP + HANG_WITH_ESCAPED_GRANDCHILD), kill_grace=0.5) started = time.monotonic() try: with pytest.raises(QsvTimeout): await qsv.run("stats", pidfile, timeout=1) assert time.monotonic() - started < 10 + _check_job_killed(_read_pid(pidfile)) finally: _kill(_read_pid(pidfile)) @@ -277,6 +332,8 @@ async def test_async_timeout_bounded_when_leader_ignores_sigterm_and_grandchild_ async def test_async_timeout_releases_pipes_held_by_escaped_grandchild( fake_qsv: Callable[[str], str], tmp_path: Path ) -> None: + if not os.path.isdir("/dev/fd"): + pytest.skip("no /dev/fd to count open descriptors") qsv = AsyncQsv(fake_qsv(HANG_WITH_ESCAPED_GRANDCHILD), kill_grace=0.3) pidfiles = [tmp_path / f"child{i}.pid" for i in range(3)] open_fds = len(os.listdir("/dev/fd")) @@ -291,12 +348,14 @@ async def test_async_timeout_releases_pipes_held_by_escaped_grandchild( _kill(_read_pid(pidfile)) -IGNORE_SIGTERM_AND_HANG = """ -import os, signal, sys, time -signal.signal(signal.SIGTERM, signal.SIG_IGN) +IGNORE_STOP_AND_HANG = ( + IGNORE_STOP + + """ +import os, sys, time open(sys.argv[-1], "w").write(str(os.getpid())) time.sleep(60) """ +) async def test_async_cancel_during_timeout_cleanup_still_kills( @@ -305,7 +364,7 @@ async def test_async_cancel_during_timeout_cleanup_still_kills( # the run times out at 0.5s and starts its 2s SIGTERM grace wait; the outer deadline at # 1s cancels it mid-cleanup, which must not skip the SIGKILL pidfile = tmp_path / "leader.pid" - qsv = AsyncQsv(fake_qsv(IGNORE_SIGTERM_AND_HANG), kill_grace=2) + qsv = AsyncQsv(fake_qsv(IGNORE_STOP_AND_HANG), kill_grace=2) try: with pytest.raises(asyncio.TimeoutError): await asyncio.wait_for(qsv.run("stats", pidfile, timeout=0.5), 1) @@ -321,7 +380,7 @@ def test_asyncio_run_returning_mid_cleanup_still_kills( # as above, but asyncio.run() returns right after the outer deadline, so any cleanup still # running in a background task would be cancelled by the runner's shutdown pidfile = tmp_path / "leader.pid" - qsv = AsyncQsv(fake_qsv(IGNORE_SIGTERM_AND_HANG), kill_grace=2) + qsv = AsyncQsv(fake_qsv(IGNORE_STOP_AND_HANG), kill_grace=2) async def main() -> None: await asyncio.wait_for(qsv.run("stats", pidfile, timeout=0.5), 1) @@ -339,7 +398,7 @@ def test_asyncio_run_cancelling_background_run_mid_cleanup_still_kills( ) -> None: # the runner's shutdown cancels the run *and* its cleanup task directly, mid-grace-wait pidfile = tmp_path / "leader.pid" - qsv = AsyncQsv(fake_qsv(IGNORE_SIGTERM_AND_HANG), kill_grace=2) + qsv = AsyncQsv(fake_qsv(IGNORE_STOP_AND_HANG), kill_grace=2) async def main() -> None: background = asyncio.create_task(qsv.run("stats", pidfile, timeout=0.5)) @@ -351,3 +410,134 @@ async def main() -> None: assert _wait_dead(_read_pid(pidfile), 6), "leader survived the runner's shutdown" finally: _kill(_read_pid(pidfile)) + + +def _wait_reaped(pid: int, within: float = 5.0) -> bool: + """Our own child that nothing else will reap: reap it here, so a SIGKILLed one counts.""" + if sys.platform == "win32": # no zombies (or waitpid) there + return _wait_dead(pid, within) + deadline = time.monotonic() + within + while time.monotonic() < deadline: + try: + done, _ = os.waitpid(pid, os.WNOHANG) # type: ignore[attr-defined,unused-ignore] + except ChildProcessError: + return True + if done: + return True + time.sleep(0.05) + return False + + +@pytest.mark.skipif(not hasattr(signal, "setitimer"), reason="needs SIGALRM") +def test_second_interrupt_during_grace_wait_still_kills( + fake_qsv: Callable[[str], str], tmp_path: Path +) -> None: + # times out at 0.5s and starts its 3s grace wait; a "second Ctrl-C" at 1s interrupts it + pidfile = tmp_path / "leader.pid" + qsv = Qsv(fake_qsv(IGNORE_STOP_AND_HANG), kill_grace=3) + + def interrupt(signum: int, frame: object) -> None: + raise KeyboardInterrupt + + previous = signal.signal(signal.SIGALRM, interrupt) # type: ignore[attr-defined,unused-ignore] + signal.setitimer(signal.ITIMER_REAL, 1.0) # type: ignore[attr-defined,unused-ignore] + try: + with pytest.raises(KeyboardInterrupt): + qsv.run("stats", pidfile, timeout=0.5) + assert _wait_reaped(_read_pid(pidfile)), "leader survived the second interrupt" + finally: + signal.setitimer(signal.ITIMER_REAL, 0) # type: ignore[attr-defined,unused-ignore] + signal.signal(signal.SIGALRM, previous) # type: ignore[attr-defined,unused-ignore] + _kill(_read_pid(pidfile)) + + +# 4 MiB, far more than any OS pipe buffer, to a fake that never reads stdin +HANG_WITHOUT_READING = """ +import time +time.sleep(60) +""" +BIG_STDIN = b"x" * (4 << 20) + + +def test_timeout_bounded_when_stdin_is_not_read(fake_qsv: Callable[[str], str]) -> None: + qsv = Qsv(fake_qsv(HANG_WITHOUT_READING), kill_grace=0.5) + started = time.monotonic() + with pytest.raises(QsvTimeout): + qsv.run("stats", stdin=BIG_STDIN, timeout=1) + assert time.monotonic() - started < 10 + + +async def test_async_timeout_bounded_when_stdin_is_not_read( + fake_qsv: Callable[[str], str], +) -> None: + qsv = AsyncQsv(fake_qsv(HANG_WITHOUT_READING), kill_grace=0.5) + started = time.monotonic() + with pytest.raises(QsvTimeout): + await qsv.run("stats", stdin=BIG_STDIN, timeout=1) + assert time.monotonic() - started < 10 + + +class _Interrupted(BaseException): + """Stands in for Ctrl-C (or any BaseException) landing during setup after the spawn.""" + + +def _interrupting_tree(spawned: list[int]) -> type: + """A setup that is interrupted at once. On Windows qsv is then still suspended (it never + runs), so record the pid we were handed rather than wait for the fake to write one.""" + + class Tree(_process._ProcessTree): + def __init__(self, pid: int) -> None: + spawned.append(pid) + raise _Interrupted + + return Tree + + +def test_interrupt_during_setup_still_kills( + fake_qsv: Callable[[str], str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + spawned: list[int] = [] + monkeypatch.setattr(_process, "_ProcessTree", _interrupting_tree(spawned)) + try: + with pytest.raises(_Interrupted): + Qsv(fake_qsv(IGNORE_STOP_AND_HANG)).run("stats", tmp_path / "leader.pid") + assert _wait_dead(spawned[0]), "leader survived an interrupted setup" + finally: + for pid in spawned: + _kill(pid) + + +async def test_async_interrupt_during_setup_still_kills( + fake_qsv: Callable[[str], str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + spawned: list[int] = [] + monkeypatch.setattr(_process, "_ProcessTree", _interrupting_tree(spawned)) + try: + with pytest.raises(_Interrupted): + await AsyncQsv(fake_qsv(IGNORE_STOP_AND_HANG)).run("stats", tmp_path / "leader.pid") + assert await asyncio.to_thread(_wait_dead, spawned[0]), "leader survived" + finally: + for pid in spawned: + _kill(pid) + + +def test_stdin_feeder_failing_to_start_still_kills( + fake_qsv: Callable[[str], str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # the Windows stdin feeder, forced on here, cannot get a thread + pidfile = tmp_path / "leader.pid" + + def no_thread(self: threading.Thread) -> None: + _read_pid(pidfile) + raise RuntimeError("can't start new thread") + + monkeypatch.setattr(_process, "IS_WINDOWS", True) + monkeypatch.setattr(threading.Thread, "start", no_thread) + try: + with pytest.raises(RuntimeError): + Qsv(fake_qsv(IGNORE_STOP_AND_HANG), kill_grace=0.5).run( + "stats", pidfile, stdin=b"a,b\n" + ) + assert _wait_reaped(_read_pid(pidfile)), "leader survived a failed feeder start" + finally: + _kill(_read_pid(pidfile))