diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 949103f3..9e6bc81b 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,37 @@ 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" + 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)" platform-build: runs-on: ubuntu-latest diff --git a/tests/test_deepseek_v4_accuracy.py b/tests/test_deepseek_v4_accuracy.py index 439ae55e..38c5d9d7 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,112 @@ 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 + +# 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 ( + 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. + """ + 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. + if os.getppid() == 1: + os._exit(1) + def _task_devices() -> tuple[int, ...]: raw_devices = os.environ.get("TASK_DEVICE", "") @@ -177,7 +285,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 +297,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 +315,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 +355,50 @@ 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 + interrupted = False + 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" + 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. + _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 +439,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")