diff --git a/pkg-py/src/commons/_execution/_backend.py b/pkg-py/src/commons/_execution/_backend.py index d373a17a..f6e1e6d9 100644 --- a/pkg-py/src/commons/_execution/_backend.py +++ b/pkg-py/src/commons/_execution/_backend.py @@ -16,6 +16,7 @@ import shutil import signal import tempfile +import threading from collections.abc import Coroutine from pathlib import Path from typing import Any, Literal, Protocol, runtime_checkable @@ -234,6 +235,12 @@ def __init__( self._closing: asyncio.Task[None] | None = None self._exit = asyncio.ensure_future(process.wait()) self._exit.add_done_callback(self._kill_group) + threading.Thread( + target=_kill_when_stopped, + args=(process.pid,), + name=f"commons-worker-stop-{process.pid}", + daemon=True, + ).start() @property def stdin(self) -> asyncio.StreamWriter: @@ -301,6 +308,37 @@ async def _finish(self) -> None: shutil.rmtree(self.scratch, ignore_errors=True) +def _kill_when_stopped(pid: int) -> None: + """Kill the worker's process group if the worker is stopped. + + Waits until the worker stops or exits, and returns without acting if + it exits. Model code can stop its own process with SIGSTOP (nothing + else has a reason to). On macOS, Python 3.14's asyncio mistakes a + stopped child for an exited one, then blocks the event loop in + ``waitpid()`` until the child really exits, which freezes the host. + Killing the worker lets that ``waitpid()`` return at once. This runs + wherever ``os.waitid`` exists, so the driver reports a stopped worker + as a crash on every such platform. + + ``WNOWAIT`` leaves the exit status for asyncio to collect. Without + ``os.waitid`` (macOS before Python 3.13), a stopped worker is not + killed here, and the call fails when its timeout runs out. + """ + if not hasattr(os, "waitid"): + return + try: + info = os.waitid(os.P_PID, pid, os.WEXITED | os.WSTOPPED | os.WNOWAIT) + except ChildProcessError: + return + if info is None or info.si_code != os.CLD_STOPPED: + return + with contextlib.suppress(ProcessLookupError, PermissionError): + os.killpg(pid, signal.SIGKILL) + return + with contextlib.suppress(ProcessLookupError): + os.kill(pid, signal.SIGKILL) + + def _signal_tree(process: asyncio.subprocess.Process, sig: signal.Signals) -> None: """Signal the child's whole process group, falling back to the child alone. diff --git a/pkg-py/tests/test_execution_driver.py b/pkg-py/tests/test_execution_driver.py index 3e7795cf..ef38fcb8 100644 --- a/pkg-py/tests/test_execution_driver.py +++ b/pkg-py/tests/test_execution_driver.py @@ -10,7 +10,6 @@ import asyncio import os import signal -import sys import time from datetime import date from typing import Annotated @@ -22,7 +21,7 @@ from commons import measure, semantic_layer from commons._execution._backend import WORKER_SCRIPT, LocalBackend, LocalSession from commons._execution._driver import Failure, Worker -from commons._execution._protocol import Error, Result, Text +from commons._execution._protocol import Error, Ready, Result, Text, encode_message from commons._execution._sandbox import protection_mode from commons._handles import HandleStore from commons._measures import Injected @@ -355,18 +354,37 @@ async def test_a_child_ignoring_sigterm_does_not_outlive_the_close(): ) -@pytest.mark.skipif( - sys.platform == "darwin" and sys.version_info >= (3, 14), - reason="asyncio on 3.14 reads macOS's waitid() report of a stopped child " - "as an exit and blocks the event loop in waitpid() until it really exits", -) -async def test_a_worker_that_stops_reading_fails_the_call_instead_of_hanging(): - async with make_worker(call_timeout=0.5) as worker: +@pytest.mark.skipif(not hasattr(os, "waitid"), reason="needs os.waitid") +async def test_a_worker_that_stops_itself_is_killed_and_reported_as_a_crash(): + async with make_worker(call_timeout=30) as worker: await worker.run("1") process = process_of(worker) - # A stopped worker cannot drain its stdin; a call bigger than the - # pipe buffer would block the write forever without a bound on it. - process.send_signal(signal.SIGSTOP) + started = time.monotonic() + reply = await worker.run( + "import os, signal\nos.kill(os.getpid(), signal.SIGSTOP)" + ) + assert isinstance(reply, Failure) + assert "crashed" in reply.message + # Killed when it stopped, rather than when the call timed out. + assert time.monotonic() - started < 10 + assert process.returncode == -signal.SIGKILL + reply = await worker.run("1 + 1") + assert isinstance(reply, Result) + assert reply.value == 2 + + +async def test_a_worker_that_stops_reading_fails_the_call_instead_of_hanging(tmp_path): + # A worker that says it is ready and then never reads stdin. Writing it a + # call bigger than the pipe buffer blocks until the write's time limit. + script = tmp_path / "worker.py" + script.write_text( + "import sys, time\n" + f"sys.stdout.buffer.write({encode_message(Ready())!r})\n" + "sys.stdout.flush()\n" + "time.sleep(60)\n" + ) + backend = LocalBackend(worker_script=script, terminate_grace=0.1) + async with make_worker(backend=backend, call_timeout=0.5) as worker: reply = await worker.run("x = " + "1" * (1024 * 1024)) assert isinstance(reply, Failure) assert "stopped reading" in reply.message