From e136eba39fc08564f1e8603f6834dab60d8bed6c Mon Sep 17 00:00:00 2001 From: luohuan19 Date: Tue, 14 Jul 2026 02:33:28 -0700 Subject: [PATCH 1/3] ci: prevent orphaned DeepSeek V4 engines on test interruption The DeepSeek V4 accuracy guard left ~800 GiB of orphaned engines behind whenever the CI runner interrupted it: a bare SIGTERM takes CPython down at SIG_DFL without unwinding the stack, so the `finally` that reaps the server (started in its own session via start_new_session) never ran, and the engine lived on holding its NPU cards and shared memory. Test hardening: - _raise_on_termination(): trap SIGTERM/SIGINT/SIGHUP and re-raise as TerminatedBySignal (a KeyboardInterrupt) so `finally` teardown runs; deafen the signals first so the runner's escalation can't abort cleanup mid-flight. - _set_pdeathsig() preexec hook: PR_SET_PDEATHSIG backstops the one signal we cannot catch (SIGKILL of pytest), closing the fork/prctl race. - _stop_process_group(term_grace): escalate to SIGKILL within ~1s on the interrupted path (the runner SIGKILLs us ~3s after SIGTERM) and stay patient (20s) on the normal path. - Cover the new paths with unit tests. CI workflow: - Switch the orphan-cleanup step from `task-submit --list` (truncates the command column to 77 chars, so the marker grep can silently miss) to `--find`, which matches the untruncated command and prints bare task-ids; report how many tasks were killed. - Raise task-submit --timeout 1200 -> 3600 to tolerate longer device acquisition. --- .github/workflows/ci.yml | 42 +++-- tests/test_deepseek_v4_accuracy.py | 244 +++++++++++++++++++++++++---- 2 files changed, 246 insertions(+), 40 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 949103f3..0162fffa 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -188,7 +188,7 @@ jobs: run_cmd="$run_cmd PTO2_RING_TASK_WINDOW=$PTO2_RING_TASK_WINDOW" run_cmd="$run_cmd PTO2_RING_HEAP=$PTO2_RING_HEAP" run_cmd="$run_cmd python -m pytest tests/test_qwen3_accuracy.py -q -s" - task-submit --device auto --timeout 1200 --max-time 1800 --run "$run_cmd" + task-submit --device auto --timeout 3600 --max-time 1800 --run "$run_cmd" - name: Run Qwen3 serving guard (prefix cache, chunked prefill, multi-batch) env: @@ -210,7 +210,7 @@ jobs: run_cmd="$run_cmd PTO2_RING_TASK_WINDOW=$PTO2_RING_TASK_WINDOW" run_cmd="$run_cmd PTO2_RING_HEAP=$PTO2_RING_HEAP" run_cmd="$run_cmd python -m pytest tests/test_qwen3_serving.py -q -s" - task-submit --device auto --timeout 1200 --max-time 1800 --run "$run_cmd" + task-submit --device auto --timeout 3600 --max-time 1800 --run "$run_cmd" - name: Run DeepSeek V4 HTTP generation accuracy guard env: @@ -242,7 +242,7 @@ jobs: run_cmd="$run_cmd SERVING_WORKER_STEP_TIMEOUT=$SERVING_WORKER_STEP_TIMEOUT" run_cmd="$run_cmd python -m pytest tests/test_deepseek_v4_accuracy.py -q -s" task-submit --device auto --device-num 8 --ignore-whitelist \ - --timeout 1200 --max-time 1800 \ + --timeout 3600 --max-time 1800 \ --run "$run_cmd" - name: Kill orphaned task-submit tasks @@ -250,14 +250,34 @@ jobs: working-directory: . run: | marker="pypto-serving-ci-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}" - task-submit --list 2>/dev/null \ - | grep -F "CI_TASK_MARKER=$marker" \ - | grep -oE 'task_[0-9_]+' \ - | sort -u \ - | while read -r task_id; do - echo "Killing orphaned task $task_id" - task-submit --kill "$task_id" || true - done || true + target="CI_TASK_MARKER=$marker" + echo "cleaning up task-submit tasks matching $target" + + # Match with `--find`, NOT `--list`: --list is a human-facing view that + # truncates the command column to 77 chars, so a grep for the full + # marker can silently miss and leave orphaned tasks holding devices. + # --find matches against the untruncated COMMAND and prints bare + # task-ids. It needs task-submit >= 2026-07-14 on the runner. + if ! ids=$(task-submit --find "$target"); then + echo "::error::task-submit --find failed — is task-submit up to date on this runner?" + exit 1 + fi + + if [ -z "$ids" ]; then + echo " no orphaned tasks" + exit 0 + fi + + # A cleanup step that prints nothing is indistinguishable from one that + # is quietly broken — report what we killed. + n=0 + while read -r task_id; do + [ -n "$task_id" ] || continue + echo " killing $task_id" + task-submit --kill "$task_id" || echo "::warning::failed to kill $task_id" + n=$((n + 1)) + done <<< "$ids" + echo "killed $n orphaned task(s)" platform-build: runs-on: ubuntu-latest diff --git a/tests/test_deepseek_v4_accuracy.py b/tests/test_deepseek_v4_accuracy.py index 439ae55e..45975240 100644 --- a/tests/test_deepseek_v4_accuracy.py +++ b/tests/test_deepseek_v4_accuracy.py @@ -11,6 +11,8 @@ from __future__ import annotations +import contextlib +import ctypes import io import json import os @@ -38,6 +40,103 @@ OVERALL_TIMEOUT_SECONDS = 1650 HEARTBEAT_SECONDS = 30 +# How long to let the server group settle after SIGTERM before escalating to SIGKILL. +# +# On the normal path (test passed or failed on its own) we can afford to be patient. +# +# On the interrupted path we cannot. task-submit's watchdog sends SIGTERM to our +# process group and escalates to SIGKILL 3 seconds later; the --kill path allows +# 5 (KILL_GRACE). Once that SIGKILL lands, pytest is gone and nobody is left to +# reap the server — which is exactly how ~800 GiB of orphaned engines accumulated. +# So when we are being terminated, the whole teardown has to fit inside ~3s. +NORMAL_TERM_GRACE_SECONDS = 20.0 +SIGNAL_TERM_GRACE_SECONDS = 1.0 + +# prctl(2) option number. Not exposed by the signal/os modules, so spell it out. +_PR_SET_PDEATHSIG = 1 + +_TERMINATION_SIGNALS = tuple( + sig + for sig in ( + getattr(signal, "SIGTERM", None), + getattr(signal, "SIGINT", None), + getattr(signal, "SIGHUP", None), + ) + if sig is not None +) + + +class TerminatedBySignal(KeyboardInterrupt): + """Raised in the main thread when the CI runner asks this process to stop. + + Subclasses KeyboardInterrupt so pytest treats it as an interrupt and tears the + session down, rather than recording it as an ordinary test failure and moving + on to the next test. + """ + + def __init__(self, signum: int) -> None: + super().__init__(f"terminated by signal {signum}") + self.signum = signum + + +@contextlib.contextmanager +def _raise_on_termination(): + """Turn SIGTERM/SIGINT/SIGHUP into an exception so that `finally` blocks run. + + CPython leaves SIGTERM at SIG_DFL, so a plain `kill` tears the interpreter down + where it stands: no stack unwind, no `finally`, no `atexit`. The server we + started with start_new_session=True is in its own session and process group, so + it survives the signal that killed us and is inherited by init — still holding + its NPU cards and several hundred GiB of shared memory. Raising from the handler + is what gives the teardown below a chance to run at all. + """ + if threading.current_thread() is not threading.main_thread(): + # signal.signal() is main-thread-only. Under pytest we always are on it; + # degrade to a no-op rather than exploding if that ever stops being true. + yield + return + + def handler(signum: int, _frame) -> None: + # Deafen ourselves before unwinding. The runner sends SIGTERM and then + # escalates, and a second delivery landing inside the teardown would abort + # it halfway — leaving behind exactly the orphan we are trying to prevent. + # SIGKILL still gets through; PR_SET_PDEATHSIG below is the answer to that. + for sig in _TERMINATION_SIGNALS: + with contextlib.suppress(OSError, ValueError): + signal.signal(sig, signal.SIG_IGN) + raise TerminatedBySignal(signum) + + previous: dict[int, object] = {} + for sig in _TERMINATION_SIGNALS: + try: + previous[sig] = signal.signal(sig, handler) + except (OSError, ValueError): + continue + try: + yield + finally: + for sig, old_handler in previous.items(): + with contextlib.suppress(OSError, ValueError): + signal.signal(sig, old_handler) + + +def _set_pdeathsig() -> None: + """Child-side preexec hook: ask the kernel to SIGKILL us if our parent dies. + + Backstop for the one signal we cannot catch. If pytest is SIGKILLed, no handler + runs and no teardown happens — but PR_SET_PDEATHSIG is enforced by the kernel, + so the server still dies. It only covers the process we fork directly, so the + server's own workers can in principle linger; treat this as a safety net under + _raise_on_termination(), not as a replacement for it. + """ + libc = ctypes.CDLL("libc.so.6", use_errno=True) + if libc.prctl(_PR_SET_PDEATHSIG, signal.SIGKILL, 0, 0, 0) != 0: + raise OSError(ctypes.get_errno(), "prctl(PR_SET_PDEATHSIG) failed") + # Close the fork/prctl race: if the parent already died in that window, the + # death signal fired before we asked for it and will never be re-sent. + if os.getppid() == 1: + os._exit(1) + def _task_devices() -> tuple[int, ...]: raw_devices = os.environ.get("TASK_DEVICE", "") @@ -177,7 +276,9 @@ def send_request() -> None: raise TimeoutError("DeepSeek completion exceeded the end-to-end timeout") -def _stop_process_group(process: subprocess.Popen) -> None: +def _stop_process_group( + process: subprocess.Popen, term_grace: float = NORMAL_TERM_GRACE_SECONDS +) -> None: try: os.killpg(process.pid, signal.SIGTERM) except ProcessLookupError: @@ -187,7 +288,7 @@ def _stop_process_group(process: subprocess.Popen) -> None: return try: - process.wait(timeout=20) + process.wait(timeout=term_grace) except subprocess.TimeoutExpired: try: os.killpg(process.pid, signal.SIGKILL) @@ -205,8 +306,9 @@ def _stop_process_group(process: subprocess.Popen) -> None: return # The server parent may exit before a worker child. Give the process group a - # short grace period, then kill any remaining descendants. - shutdown_deadline = time.monotonic() + 2 + # short grace period, then kill any remaining descendants. Bounded by term_grace + # so the interrupted path still fits inside the runner's SIGKILL window. + shutdown_deadline = time.monotonic() + min(2.0, term_grace) while time.monotonic() < shutdown_deadline: try: os.killpg(process.pid, 0) @@ -244,31 +346,47 @@ def test_deepseek_v4_http_completion_matches_expected_text(tmp_path: Path) -> No log_path = tmp_path / "deepseek-v4-server.log" deadline = time.monotonic() + OVERALL_TIMEOUT_SECONDS - try: - with log_path.open("w", encoding="utf-8") as server_log: - process = subprocess.Popen( - _server_command(model_dir, devices, port), - cwd=ROOT, - stdout=server_log, - stderr=subprocess.STDOUT, - start_new_session=True, - text=True, - ) - try: - _wait_for_health(process, port, deadline) - response = _request_completion(process, port, deadline) - print(f"DeepSeek completion response: {response}", flush=True) - - assert response.get("model") == MODEL_ID - choices = response.get("choices") - assert isinstance(choices, list) and len(choices) == 1 - assert choices[0].get("text") == EXPECTED_TEXT - assert choices[0].get("finish_reason") == "length" - finally: - _stop_process_group(process) - except BaseException: - _print_server_log(log_path) - raise + with _raise_on_termination(): + try: + with log_path.open("w", encoding="utf-8") as server_log: + # `process` is bound before the try so that a signal arriving between + # fork and assignment still reaches the teardown. Popen's own internals + # are covered by PR_SET_PDEATHSIG. + process = None + try: + process = subprocess.Popen( + _server_command(model_dir, devices, port), + cwd=ROOT, + stdout=server_log, + stderr=subprocess.STDOUT, + start_new_session=True, + preexec_fn=_set_pdeathsig, + text=True, + ) + _wait_for_health(process, port, deadline) + response = _request_completion(process, port, deadline) + print(f"DeepSeek completion response: {response}", flush=True) + + assert response.get("model") == MODEL_ID + choices = response.get("choices") + assert isinstance(choices, list) and len(choices) == 1 + assert choices[0].get("text") == EXPECTED_TEXT + assert choices[0].get("finish_reason") == "length" + finally: + if process is not None: + # Being torn down by the runner means we have ~3s before SIGKILL, + # so escalate to SIGKILL fast. On the normal path, be patient and + # let the engine shut down cleanly. + interrupted = isinstance(sys.exc_info()[1], TerminatedBySignal) + _stop_process_group( + process, + term_grace=( + SIGNAL_TERM_GRACE_SECONDS if interrupted else NORMAL_TERM_GRACE_SECONDS + ), + ) + except BaseException: + _print_server_log(log_path) + raise def test_completion_http_error_includes_response_body(monkeypatch) -> None: @@ -309,6 +427,74 @@ def wait(timeout): assert "still alive after SIGKILL" in capsys.readouterr().out +def test_stop_process_group_honours_term_grace(monkeypatch) -> None: + """The interrupted path must escalate to SIGKILL well inside the runner's window.""" + waits: list[float] = [] + signals: list[int] = [] + + class StuckProcess: + pid = 123 + + @staticmethod + def wait(timeout): + waits.append(timeout) + raise subprocess.TimeoutExpired("server", timeout) + + monkeypatch.setattr(os, "killpg", lambda _pid, sig: signals.append(sig)) + + _stop_process_group(StuckProcess(), term_grace=SIGNAL_TERM_GRACE_SECONDS) + + assert waits[0] == SIGNAL_TERM_GRACE_SECONDS + assert signals[:2] == [signal.SIGTERM, signal.SIGKILL] + + +def test_sigterm_unwinds_the_stack_so_teardown_runs() -> None: + """The whole point: a bare SIGTERM must not skip `finally`. + + Without _raise_on_termination() this test does not fail — it takes the pytest + process down with it, which is precisely the production bug. + """ + torn_down: list[str] = [] + + with pytest.raises(TerminatedBySignal) as excinfo: + with _raise_on_termination(): + try: + os.kill(os.getpid(), signal.SIGTERM) + # The handler runs at the next bytecode boundary; give it one. + time.sleep(1) + finally: + torn_down.append("server stopped") + + assert excinfo.value.signum == signal.SIGTERM + assert torn_down == ["server stopped"] + + +def test_repeated_sigterm_cannot_abort_teardown() -> None: + """The runner escalates. A second SIGTERM must not interrupt cleanup in progress.""" + torn_down: list[str] = [] + + with pytest.raises(TerminatedBySignal): + with _raise_on_termination(): + try: + os.kill(os.getpid(), signal.SIGTERM) + time.sleep(1) + finally: + # Simulates the escalation landing mid-teardown. It must be ignored, + # not turned into a second exception that abandons the cleanup. + os.kill(os.getpid(), signal.SIGTERM) + time.sleep(0.2) + torn_down.append("server stopped") + + assert torn_down == ["server stopped"] + + +def test_signal_handlers_are_restored_on_exit() -> None: + original = signal.getsignal(signal.SIGTERM) + with _raise_on_termination(): + assert signal.getsignal(signal.SIGTERM) is not original + assert signal.getsignal(signal.SIGTERM) is original + + def test_print_server_log_reads_only_tail(tmp_path, capsys) -> None: log_path = tmp_path / "server.log" log_path.write_bytes(b"excluded-prefix\n" + b"x" * 60000 + b"\nincluded-tail\n") From 760b05cf8e6e91610ecbff035895c8b1cde57c7c Mon Sep 17 00:00:00 2001 From: luohuan19 Date: Tue, 14 Jul 2026 02:52:14 -0700 Subject: [PATCH 2/3] =?UTF-8?q?test:=20address=20review=20=E2=80=94=20load?= =?UTF-8?q?=20libc=20at=20import,=20explicit=20interrupted=20flag?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Load libc.so.6 at module import instead of inside the preexec_fn, so the child does not dlopen after fork (unsafe if another thread held the loader lock across the fork); _set_pdeathsig() now uses the pre-loaded handle. - Replace the fragile sys.exc_info() probe in the finally block with an explicit `except TerminatedBySignal: interrupted = True; raise`. --- tests/test_deepseek_v4_accuracy.py | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/tests/test_deepseek_v4_accuracy.py b/tests/test_deepseek_v4_accuracy.py index 45975240..38c5d9d7 100644 --- a/tests/test_deepseek_v4_accuracy.py +++ b/tests/test_deepseek_v4_accuracy.py @@ -55,6 +55,14 @@ # prctl(2) option number. Not exposed by the signal/os modules, so spell it out. _PR_SET_PDEATHSIG = 1 +# Load libc once, at import time. _set_pdeathsig() runs as a preexec_fn — after +# fork but before exec — where calling ctypes.CDLL() would dlopen in the child and +# is unsafe if any other thread in the parent held the loader lock across the fork. +try: + _libc = ctypes.CDLL("libc.so.6", use_errno=True) +except (OSError, AttributeError): + _libc = None + _TERMINATION_SIGNALS = tuple( sig for sig in ( @@ -129,8 +137,9 @@ def _set_pdeathsig() -> None: server's own workers can in principle linger; treat this as a safety net under _raise_on_termination(), not as a replacement for it. """ - libc = ctypes.CDLL("libc.so.6", use_errno=True) - if libc.prctl(_PR_SET_PDEATHSIG, signal.SIGKILL, 0, 0, 0) != 0: + if _libc is None: + raise OSError("libc.so.6 not loaded") + if _libc.prctl(_PR_SET_PDEATHSIG, signal.SIGKILL, 0, 0, 0) != 0: raise OSError(ctypes.get_errno(), "prctl(PR_SET_PDEATHSIG) failed") # Close the fork/prctl race: if the parent already died in that window, the # death signal fired before we asked for it and will never be re-sent. @@ -353,6 +362,7 @@ def test_deepseek_v4_http_completion_matches_expected_text(tmp_path: Path) -> No # fork and assignment still reaches the teardown. Popen's own internals # are covered by PR_SET_PDEATHSIG. process = None + interrupted = False try: process = subprocess.Popen( _server_command(model_dir, devices, port), @@ -372,12 +382,14 @@ def test_deepseek_v4_http_completion_matches_expected_text(tmp_path: Path) -> No assert isinstance(choices, list) and len(choices) == 1 assert choices[0].get("text") == EXPECTED_TEXT assert choices[0].get("finish_reason") == "length" + except TerminatedBySignal: + interrupted = True + raise finally: if process is not None: # Being torn down by the runner means we have ~3s before SIGKILL, # so escalate to SIGKILL fast. On the normal path, be patient and # let the engine shut down cleanly. - interrupted = isinstance(sys.exc_info()[1], TerminatedBySignal) _stop_process_group( process, term_grace=( From ede9e36500a28b934da32cb1cb6cd11e1e91abf2 Mon Sep 17 00:00:00 2001 From: luohuan19 Date: Tue, 14 Jul 2026 02:55:28 -0700 Subject: [PATCH 3/3] ci: count only successful orphan-task kills Increment the killed counter inside the success branch so the "killed N orphaned task(s)" summary matches the warnings and never overcounts a kill that actually failed. --- .github/workflows/ci.yml | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0162fffa..9e6bc81b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -274,8 +274,11 @@ jobs: while read -r task_id; do [ -n "$task_id" ] || continue echo " killing $task_id" - task-submit --kill "$task_id" || echo "::warning::failed to kill $task_id" - n=$((n + 1)) + if task-submit --kill "$task_id"; then + n=$((n + 1)) + else + echo "::warning::failed to kill $task_id" + fi done <<< "$ids" echo "killed $n orphaned task(s)"