From befa9cca7ea3332dae94878a5b3fecf17c68dee0 Mon Sep 17 00:00:00 2001 From: Josh Taillon Date: Fri, 30 Oct 2026 17:16:27 -0600 Subject: [PATCH 1/3] fix(py): poll a macOS worker's state to kill it when it stops itself CPython 3.14.10 (gh-158287) makes asyncio's child watcher on macOS take a stopped child's wait report with waitid(WSTOPPED | WNOHANG). The stop watcher waited for that same report with WNOWAIT, so when asyncio's thread got there first the watcher never woke, and the call ran out its timeout instead of reporting a crash. On macOS the watcher now polls proc_pidinfo() for the worker's state, which shows SSTOP regardless of who took the report. This also covers macOS before Python 3.13, which has no os.waitid, so the driver test now runs there too. Other platforms keep the waitid wait. --- pkg-py/src/commons/_execution/_backend.py | 116 +++++++++++++++++++--- pkg-py/tests/test_execution_backend.py | 24 ++++- pkg-py/tests/test_execution_driver.py | 5 +- 3 files changed, 129 insertions(+), 16 deletions(-) diff --git a/pkg-py/src/commons/_execution/_backend.py b/pkg-py/src/commons/_execution/_backend.py index f6e1e6d97..770db0856 100644 --- a/pkg-py/src/commons/_execution/_backend.py +++ b/pkg-py/src/commons/_execution/_backend.py @@ -12,11 +12,14 @@ import asyncio import contextlib +import ctypes import os import shutil import signal +import sys import tempfile import threading +import time from collections.abc import Coroutine from pathlib import Path from typing import Any, Literal, Protocol, runtime_checkable @@ -313,24 +316,31 @@ def _kill_when_stopped(pid: int) -> None: 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 + else has a reason to). On macOS, Python 3.14 before 3.14.10 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. + Killing the worker lets that ``waitpid()`` return at once, and the + driver reports a stopped worker as a crash rather than waiting out + the call's timeout. + + On macOS this polls the process table, because from 3.14.10 asyncio + takes the stop report that ``waitid`` would otherwise leave for this + thread. Elsewhere ``WNOWAIT`` leaves the exit status for asyncio to + collect. Without + either (Windows), 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: + if sys.platform == "darwin": + stopped = _darwin_wait_for_stop(pid) + elif hasattr(os, "waitid"): + try: + info = os.waitid(os.P_PID, pid, os.WEXITED | os.WSTOPPED | os.WNOWAIT) + except ChildProcessError: + return + stopped = info is not None and info.si_code == os.CLD_STOPPED + else: return - if info is None or info.si_code != os.CLD_STOPPED: + if not stopped: return with contextlib.suppress(ProcessLookupError, PermissionError): os.killpg(pid, signal.SIGKILL) @@ -339,6 +349,84 @@ def _kill_when_stopped(pid: int) -> None: os.kill(pid, signal.SIGKILL) +# How often the macOS stop watcher reads a worker's state. +STOP_POLL = 0.1 + +# From and . +_PROC_PIDTBSDINFO = 3 +_SSTOP = 4 + + +class _ProcBsdInfo(ctypes.Structure): + """``struct proc_bsdinfo``, which ``proc_pidinfo()`` fills.""" + + _fields_ = [ + *( + (name, ctypes.c_uint32) + for name in ( + "pbi_flags", + "pbi_status", + "pbi_xstatus", + "pbi_pid", + "pbi_ppid", + "pbi_uid", + "pbi_gid", + "pbi_ruid", + "pbi_rgid", + "pbi_svuid", + "pbi_svgid", + "rfu_1", + ) + ), + ("pbi_comm", ctypes.c_char * 16), + ("pbi_name", ctypes.c_char * 32), + *( + (name, ctypes.c_uint32) + for name in ( + "pbi_nfiles", + "pbi_pgid", + "pbi_pjobc", + "e_tdev", + "e_tpgid", + ) + ), + ("pbi_nice", ctypes.c_int32), + ("pbi_start_tvsec", ctypes.c_uint64), + ("pbi_start_tvusec", ctypes.c_uint64), + ] + + +def _darwin_wait_for_stop(pid: int) -> bool: + """Poll the worker's state until it stops (True) or is gone (False). + + ``proc_pidinfo()`` fails once the worker is a zombie. The start time + guards against acting on a later process that reuses the PID. + """ + proc_pidinfo = ctypes.CDLL(None).proc_pidinfo + proc_pidinfo.argtypes = [ + ctypes.c_int, + ctypes.c_int, + ctypes.c_uint64, + ctypes.c_void_p, + ctypes.c_int, + ] + proc_pidinfo.restype = ctypes.c_int + info = _ProcBsdInfo() + size = ctypes.sizeof(info) + started = None + while True: + if proc_pidinfo(pid, _PROC_PIDTBSDINFO, 0, ctypes.byref(info), size) != size: + return False + start = (info.pbi_start_tvsec, info.pbi_start_tvusec) + if started is None: + started = start + elif start != started: + return False + if info.pbi_status == _SSTOP: + return True + time.sleep(STOP_POLL) + + 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_backend.py b/pkg-py/tests/test_execution_backend.py index d6d5cf92b..e9e3cf1ef 100644 --- a/pkg-py/tests/test_execution_backend.py +++ b/pkg-py/tests/test_execution_backend.py @@ -13,6 +13,8 @@ import os import signal import subprocess +import sys +import threading import time from pathlib import Path from typing import Any, cast @@ -27,6 +29,7 @@ LocalBackend, LocalSession, WorkerSession, + _kill_when_stopped, _read_tail, _signal_tree, _terminate, @@ -347,7 +350,6 @@ def test_the_default_grace_is_the_documented_constant() -> None: assert LocalBackend()._terminate_grace == TERMINATE_GRACE - @pytest.mark.skipif( not hasattr(os, "waitid"), reason="os.waitid is POSIX, and on macOS 3.13+" ) @@ -359,3 +361,23 @@ def test_a_group_left_with_only_a_zombie_can_still_be_signalled() -> None: os.waitid(os.P_PID, process.pid, os.WEXITED | os.WNOWAIT) # pyrefly: ignore[missing-attribute] _signal_tree(cast(Any, process), signal.SIGTERM) assert process.wait() == 0 + + +@pytest.mark.skipif(sys.platform != "darwin", reason="reads the macOS process table") +def test_a_stopped_worker_is_killed_after_another_waiter_took_the_stop_report() -> None: + # Python 3.14.10's asyncio on macOS takes the report of a stopped child + # before the stop watcher can see it. + process = subprocess.Popen(["sleep", "30"], start_new_session=True) + try: + os.kill(process.pid, signal.SIGSTOP) + os.waitpid(process.pid, os.WUNTRACED) + watcher = threading.Thread( + target=_kill_when_stopped, args=(process.pid,), daemon=True + ) + watcher.start() + watcher.join(timeout=5) + assert not watcher.is_alive(), "the stopped worker was never killed" + assert process.wait(timeout=5) == -signal.SIGKILL + finally: + process.kill() + process.wait() diff --git a/pkg-py/tests/test_execution_driver.py b/pkg-py/tests/test_execution_driver.py index cbef47d20..a7a31aa92 100644 --- a/pkg-py/tests/test_execution_driver.py +++ b/pkg-py/tests/test_execution_driver.py @@ -10,6 +10,7 @@ import asyncio import os import signal +import sys import time from datetime import date from typing import Annotated @@ -357,7 +358,9 @@ async def test_a_child_ignoring_sigterm_does_not_outlive_the_close(): ) -@pytest.mark.skipif(not hasattr(os, "waitid"), reason="needs os.waitid") +@pytest.mark.skipif( + sys.platform != "darwin" and 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") From f1aad908f4082c4db3bfa0b54bd025c372a010df Mon Sep 17 00:00:00 2001 From: Josh Taillon Date: Fri, 30 Oct 2026 17:22:55 -0600 Subject: [PATCH 2/3] fix(py): read a macOS worker's start time when it is spawned The stop watcher took the worker's start time on its first poll, so a worker reaped before the thread first ran could be confused with a later process that reused its PID. LocalSession now reads it on the event loop's thread right after the spawn, before asyncio can reap the worker. --- pkg-py/src/commons/_execution/_backend.py | 52 +++++++++++++++-------- pkg-py/tests/test_execution_backend.py | 42 +++++++++++++----- 2 files changed, 66 insertions(+), 28 deletions(-) diff --git a/pkg-py/src/commons/_execution/_backend.py b/pkg-py/src/commons/_execution/_backend.py index 770db0856..0a2fc38da 100644 --- a/pkg-py/src/commons/_execution/_backend.py +++ b/pkg-py/src/commons/_execution/_backend.py @@ -240,7 +240,7 @@ def __init__( self._exit.add_done_callback(self._kill_group) threading.Thread( target=_kill_when_stopped, - args=(process.pid,), + args=(process.pid, _process_start(process.pid)), name=f"commons-worker-stop-{process.pid}", daemon=True, ).start() @@ -311,7 +311,7 @@ async def _finish(self) -> None: shutil.rmtree(self.scratch, ignore_errors=True) -def _kill_when_stopped(pid: int) -> None: +def _kill_when_stopped(pid: int, started: tuple[int, int] | None = None) -> None: """Kill the worker's process group if the worker is stopped. Waits until the worker stops or exits, and returns without acting if @@ -325,13 +325,16 @@ def _kill_when_stopped(pid: int) -> None: On macOS this polls the process table, because from 3.14.10 asyncio takes the stop report that ``waitid`` would otherwise leave for this - thread. Elsewhere ``WNOWAIT`` leaves the exit status for asyncio to - collect. Without + thread. ``started`` is the worker's start time from ``_process_start``, + so a later process that reuses the PID is left alone. Elsewhere + ``WNOWAIT`` leaves the exit status for asyncio to collect. Without either (Windows), a stopped worker is not killed here, and the call fails when its timeout runs out. """ if sys.platform == "darwin": - stopped = _darwin_wait_for_stop(pid) + if started is None: + return + stopped = _darwin_wait_for_stop(pid, started) elif hasattr(os, "waitid"): try: info = os.waitid(os.P_PID, pid, os.WEXITED | os.WSTOPPED | os.WNOWAIT) @@ -396,12 +399,8 @@ class _ProcBsdInfo(ctypes.Structure): ] -def _darwin_wait_for_stop(pid: int) -> bool: - """Poll the worker's state until it stops (True) or is gone (False). - - ``proc_pidinfo()`` fails once the worker is a zombie. The start time - guards against acting on a later process that reuses the PID. - """ +def _darwin_proc_info(pid: int) -> _ProcBsdInfo | None: + """Read a process's entry, or None once it is a zombie or gone.""" proc_pidinfo = ctypes.CDLL(None).proc_pidinfo proc_pidinfo.argtypes = [ ctypes.c_int, @@ -413,14 +412,31 @@ def _darwin_wait_for_stop(pid: int) -> bool: proc_pidinfo.restype = ctypes.c_int info = _ProcBsdInfo() size = ctypes.sizeof(info) - started = None + if proc_pidinfo(pid, _PROC_PIDTBSDINFO, 0, ctypes.byref(info), size) != size: + return None + return info + + +def _process_start(pid: int) -> tuple[int, int] | None: + """The start time that identifies the process behind ``pid`` on macOS. + + Read on the event loop's thread right after the spawn, before asyncio + can reap the worker and free its PID for reuse. None off macOS, or if + the worker is already a zombie. + """ + if sys.platform != "darwin": + return None + info = _darwin_proc_info(pid) + if info is None: + return None + return (info.pbi_start_tvsec, info.pbi_start_tvusec) + + +def _darwin_wait_for_stop(pid: int, started: tuple[int, int]) -> bool: + """Poll the worker's state until it stops (True) or is gone (False).""" while True: - if proc_pidinfo(pid, _PROC_PIDTBSDINFO, 0, ctypes.byref(info), size) != size: - return False - start = (info.pbi_start_tvsec, info.pbi_start_tvusec) - if started is None: - started = start - elif start != started: + info = _darwin_proc_info(pid) + if info is None or (info.pbi_start_tvsec, info.pbi_start_tvusec) != started: return False if info.pbi_status == _SSTOP: return True diff --git a/pkg-py/tests/test_execution_backend.py b/pkg-py/tests/test_execution_backend.py index e9e3cf1ef..46cad9df4 100644 --- a/pkg-py/tests/test_execution_backend.py +++ b/pkg-py/tests/test_execution_backend.py @@ -30,6 +30,7 @@ LocalSession, WorkerSession, _kill_when_stopped, + _process_start, _read_tail, _signal_tree, _terminate, @@ -363,21 +364,42 @@ def test_a_group_left_with_only_a_zombie_can_still_be_signalled() -> None: assert process.wait() == 0 -@pytest.mark.skipif(sys.platform != "darwin", reason="reads the macOS process table") -def test_a_stopped_worker_is_killed_after_another_waiter_took_the_stop_report() -> None: +def stopped_after_another_waiter_took_the_report() -> subprocess.Popen[bytes]: # Python 3.14.10's asyncio on macOS takes the report of a stopped child # before the stop watcher can see it. process = subprocess.Popen(["sleep", "30"], start_new_session=True) + os.kill(process.pid, signal.SIGSTOP) + os.waitpid(process.pid, os.WUNTRACED) + return process + + +def watch_for_stop(pid: int, started: tuple[int, int] | None) -> bool: + """Run the stop watcher, and say whether it returned in time.""" + watcher = threading.Thread( + target=_kill_when_stopped, args=(pid, started), daemon=True + ) + watcher.start() + watcher.join(timeout=5) + return not watcher.is_alive() + + +@pytest.mark.skipif(sys.platform != "darwin", reason="reads the macOS process table") +def test_a_stopped_worker_is_killed_after_another_waiter_took_the_stop_report() -> None: + process = stopped_after_another_waiter_took_the_report() try: - os.kill(process.pid, signal.SIGSTOP) - os.waitpid(process.pid, os.WUNTRACED) - watcher = threading.Thread( - target=_kill_when_stopped, args=(process.pid,), daemon=True - ) - watcher.start() - watcher.join(timeout=5) - assert not watcher.is_alive(), "the stopped worker was never killed" + assert watch_for_stop(process.pid, _process_start(process.pid)) assert process.wait(timeout=5) == -signal.SIGKILL finally: process.kill() process.wait() + + +@pytest.mark.skipif(sys.platform != "darwin", reason="reads the macOS process table") +def test_a_stopped_process_that_reused_the_pid_is_left_alone() -> None: + process = stopped_after_another_waiter_took_the_report() + try: + assert watch_for_stop(process.pid, (0, 0)) + assert process.poll() is None + finally: + process.kill() + process.wait() From 36556b8c7780a72829a5fae0da01f4bf31337fff Mon Sep 17 00:00:00 2001 From: Josh Taillon Date: Fri, 30 Oct 2026 17:27:07 -0600 Subject: [PATCH 3/3] fix(py): check the platform where the worker's start time is read pyrefly evaluates sys.platform checks for the platform it runs on, so on Linux the early return left the rest of _process_start typed as unreachable. The caller now decides whether to read it. --- pkg-py/src/commons/_execution/_backend.py | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/pkg-py/src/commons/_execution/_backend.py b/pkg-py/src/commons/_execution/_backend.py index 0a2fc38da..25d2b42fa 100644 --- a/pkg-py/src/commons/_execution/_backend.py +++ b/pkg-py/src/commons/_execution/_backend.py @@ -240,7 +240,10 @@ def __init__( self._exit.add_done_callback(self._kill_group) threading.Thread( target=_kill_when_stopped, - args=(process.pid, _process_start(process.pid)), + args=( + process.pid, + _process_start(process.pid) if sys.platform == "darwin" else None, + ), name=f"commons-worker-stop-{process.pid}", daemon=True, ).start() @@ -421,11 +424,9 @@ def _process_start(pid: int) -> tuple[int, int] | None: """The start time that identifies the process behind ``pid`` on macOS. Read on the event loop's thread right after the spawn, before asyncio - can reap the worker and free its PID for reuse. None off macOS, or if - the worker is already a zombie. + can reap the worker and free its PID for reuse. None if the worker is + already a zombie. """ - if sys.platform != "darwin": - return None info = _darwin_proc_info(pid) if info is None: return None