diff --git a/tests/integration/defs/stress_test/disagg_cancel/README.md b/tests/integration/defs/stress_test/disagg_cancel/README.md index 8251840a2d61..8be4c893a469 100644 --- a/tests/integration/defs/stress_test/disagg_cancel/README.md +++ b/tests/integration/defs/stress_test/disagg_cancel/README.md @@ -13,19 +13,20 @@ transceiver under heavy mid-flight cancellation). ## Status -This is the **skeleton** stage. The harness class structure is in -place; thread bodies are intentionally stubs that exit immediately. -The pytest test exercises the lifecycle (`setup → start → -wait_until_done → stop`) only. +The harness class structure and lifecycle are in place. Thread bodies +land incrementally: -Thread bodies are implemented incrementally in subsequent commits, -in roughly this order (read-only first, side-effecting later): +| Thread | Status | +|--------|--------| +| `log_scanner_thread` | Implemented — hard-zero log fail-fast | +| `metrics_thread` | Stub (Step 2) | +| `injector_thread` | Implemented — SIGSTOP/SIGCONT/SIGKILL + respawn | +| `canary_thread` | Stub | +| `load_thread` | Stub | -1. `log_scanner_thread` (read-only — easiest) -2. `metrics_thread` (read-only — almost as easy) -3. `injector_thread` (subprocess control) -4. `canary_thread` (HTTP client + token-equivalence) -5. `load_thread` (wraps existing `run_cancel_stress_test`) +Component-level coverage: `test_log_scanner.py`, `test_injector.py`. +The parametrized marathon pytest still runs a lifecycle smoke until +`setup()` launches a real cluster and the remaining threads are wired. ## File layout @@ -35,6 +36,8 @@ tests/integration/defs/stress_test/disagg_cancel/ ├── __init__.py ├── harness.py (DisaggCancellationStressHarness) ├── test_disagg_cancel_stress.py (pytest entry point) +├── test_log_scanner.py (log_scanner unit tests) +├── test_injector.py (injector unit tests) └── configs/ ├── README.md (YAML schema + how to add a config) ├── marathon_cpp_v1_deepseek.yaml @@ -57,24 +60,70 @@ nightly / weekly via `tests/integration/test_lists/qa/llm_function_stress.txt` (wiring lands together with the load-thread implementation). -### Local smoke (skeleton stage) +### Unit tests (no GPU, no cluster) + +Component tests for individual harness threads run in isolation. They +do **not** need `LLM_MODELS_ROOT`, GPUs, or a TRT-LLM venv with +`transformers` — use `--confcutdir` so pytest skips the parent +`tests/integration/defs/conftest.py`. + +From the repository root: ```bash cd /path/to/TensorRT-LLM -LLM_MODELS_ROOT=/path/to/models \ - pytest -sv tests/integration/defs/stress_test/disagg_cancel/ + +export PYTHONPATH=tests/integration/defs:tests/integration/defs/disaggregated + +# Step 3 — injector thread (SIGSTOP / SIGCONT / SIGKILL + respawn) +python3 -m pytest -c /dev/null -o addopts= \ + --confcutdir=tests/integration/defs/stress_test \ + tests/integration/defs/stress_test/disagg_cancel/test_injector.py -v + +# Step 1 — log scanner (optional sanity alongside injector PR) +python3 -m pytest -c /dev/null -o addopts= \ + --confcutdir=tests/integration/defs/stress_test \ + tests/integration/defs/stress_test/disagg_cancel/test_log_scanner.py -v + +# Marathon YAML parse/validate (includes stress_config.injections schedule) +python3 -m pytest -c /dev/null -o addopts= \ + --confcutdir=tests/integration/defs/stress_test \ + tests/integration/defs/stress_test/disagg_cancel/test_disagg_cancel_stress.py::test_all_marathon_yamls_parse_and_validate -v +``` + +All three together: + +```bash +python3 -m pytest -c /dev/null -o addopts= \ + --confcutdir=tests/integration/defs/stress_test \ + tests/integration/defs/stress_test/disagg_cancel/test_injector.py \ + tests/integration/defs/stress_test/disagg_cancel/test_log_scanner.py \ + tests/integration/defs/stress_test/disagg_cancel/test_disagg_cancel_stress.py::test_all_marathon_yamls_parse_and_validate -q +``` + +In a full TRT-LLM dev container/venv (with `transformers` installed), +the same tests also run under the normal integration pytest path: + +```bash +pytest -sv tests/integration/defs/stress_test/disagg_cancel/test_injector.py +``` + +### Lifecycle smoke (injector not exercised on real workers) + +```bash +pytest -sv tests/integration/defs/stress_test/disagg_cancel/test_disagg_cancel_stress.py::test_disagg_cancellation_marathon ``` -In the skeleton stage this should complete in seconds because all -threads are no-ops; it only verifies the harness lifecycle compiles -and the YAMLs parse. +`setup()` is still a stub, so this only checks harness lifecycle +(`setup` → `start` → `wait` → `stop`). The injector thread exits +immediately because no workers are registered via +`bind_tracked_workers()`. -### Local marathon (once thread bodies are wired) +### Local marathon (after `setup()` + load/canary land) -Once thread bodies are implemented, the same command will run the -full 2-hour marathon against the C++ marathon config. To run a shorter smoke -during development, set `duration_min: 10` and trim -`injections:` in the YAML. +Once `setup()` launches a real 3P3D cluster and registers workers, +the full 2-hour marathon runs via the same pytest entry point. For +development, set `duration_min: 10` and trim `injections:` in the +YAML. ## Pass criteria diff --git a/tests/integration/defs/stress_test/disagg_cancel/harness.py b/tests/integration/defs/stress_test/disagg_cancel/harness.py index 68fbab35c95a..f98be691a415 100644 --- a/tests/integration/defs/stress_test/disagg_cancel/harness.py +++ b/tests/integration/defs/stress_test/disagg_cancel/harness.py @@ -29,7 +29,11 @@ from __future__ import annotations import logging +import os +import random import re +import signal +import subprocess import threading import time import urllib.error @@ -148,6 +152,28 @@ def validate(self) -> None: raise ValueError(f"transceiver must be 'cpp' or 'python', got {self.transceiver!r}") +_INJECTION_TARGET_RE = re.compile(r"^(ctx|gen)_worker_(\d+)$") + + +@dataclass +class _TrackedWorker: + """Runtime binding between a shadow ``WorkerLaunchSpec`` and its subprocess.""" + + spec: WorkerLaunchSpec + wrapper: Any # disagg_test_utils.ProcessWrapper + + +@dataclass(frozen=True) +class _InjectionSpec: + """One entry from ``stress_config.injections`` after validation.""" + + at_min: float + type: str # sigstop | sigkill + target: str + duration_s: Optional[float] = None + respawn_within_s: Optional[float] = None + + @dataclass class WorkerLaunchSpec: """Shadow-tracked launch context for a single worker. @@ -310,6 +336,183 @@ def _compile_patterns(raw_patterns: list[Any]) -> list[tuple[str, re.Pattern[str return compiled +def _parse_injection_schedule(raw_injections: list[Any]) -> list[_InjectionSpec]: + """Normalize and validate the YAML ``injections:`` schedule. + + Malformed entries are logged at ERROR and skipped so a typo in one + slot does not abort the whole marathon. + + Args: + raw_injections: Value of ``stress_config.injections`` from YAML. + + Returns: + Sorted list of validated injection specs (by ``at_min``). + """ + specs: list[_InjectionSpec] = [] + for idx, entry in enumerate(raw_injections): + if not isinstance(entry, dict): + logger.error("[injector] injections[%d] is not a mapping; skipping", idx) + continue + try: + at_min = float(entry["at_min"]) + inj_type = str(entry["type"]).lower() + target = str(entry["target"]) + except (KeyError, TypeError, ValueError) as exc: + logger.error("[injector] injections[%d] missing required fields: %s", idx, exc) + continue + if inj_type not in ("sigstop", "sigkill"): + logger.error("[injector] injections[%d] unsupported type %r; skipping", idx, inj_type) + continue + duration_s = entry.get("duration_s") + respawn_within_s = entry.get("respawn_within_s") + if inj_type == "sigstop": + if duration_s is None: + logger.error("[injector] injections[%d] sigstop missing duration_s; skipping", idx) + continue + try: + duration_s = float(duration_s) + except (TypeError, ValueError): + logger.error( + "[injector] injections[%d] invalid duration_s %r; skipping", + idx, + duration_s, + ) + continue + if inj_type == "sigkill" and respawn_within_s is not None: + try: + respawn_within_s = float(respawn_within_s) + except (TypeError, ValueError): + logger.error( + "[injector] injections[%d] invalid respawn_within_s %r; skipping", + idx, + respawn_within_s, + ) + continue + specs.append( + _InjectionSpec( + at_min=at_min, + type=inj_type, + target=target, + duration_s=duration_s, + respawn_within_s=respawn_within_s, + ) + ) + specs.sort(key=lambda s: s.at_min) + return specs + + +def _resolve_injection_target(target: str, tracked: list[_TrackedWorker]) -> _TrackedWorker: + """Map a YAML target string to a tracked worker. + + Args: + target: One of ``gen_worker_random``, ``ctx_worker_random``, + or ``{ctx|gen}_worker_``. + tracked: Workers registered at cluster setup. + + Returns: + The selected ``_TrackedWorker``. + + Raises: + ValueError: If ``target`` is unknown or no worker matches. + """ + if target == "gen_worker_random": + pool = [t for t in tracked if t.spec.role == "gen"] + elif target == "ctx_worker_random": + pool = [t for t in tracked if t.spec.role == "ctx"] + else: + match = _INJECTION_TARGET_RE.match(target) + if match is None: + raise ValueError(f"unsupported injection target {target!r}") + role, index = match.group(1), int(match.group(2)) + pool = [t for t in tracked if t.spec.role == role and t.spec.index == index] + if not pool: + raise ValueError(f"no worker matches injection target {target!r}") + if target.endswith("_random"): + return random.choice(pool) + return pool[0] + + +def _worker_process_pid(wrapper: Any) -> Optional[int]: + """Return the worker subprocess PID, or ``None`` if unavailable.""" + if wrapper is None or wrapper.process is None: + return None + return wrapper.process.pid + + +def _signal_process(pid: int, sig: signal.Signals, label: str) -> bool: + """Send ``sig`` to ``pid``; log and return False on failure.""" + try: + os.kill(pid, sig) + return True + except ProcessLookupError: + logger.warning("[injector] %s: process %d already gone", label, pid) + return False + except OSError as exc: + logger.warning("[injector] %s: os.kill(%d, %s) failed: %s", label, pid, sig.name, exc) + return False + + +def _execute_sigstop_pause( + wrapper: Any, + duration_s: float, + should_stop: Optional[Callable[[], bool]] = None, +) -> dict[str, Any]: + """SIGSTOP a worker for ``duration_s``, then SIGCONT. + + Args: + wrapper: ``ProcessWrapper`` for the victim worker. + duration_s: Pause length in seconds. + should_stop: Optional predicate polled during the pause; when + it returns True the pause is cut short (and ``interrupted`` + is recorded), but SIGCONT is still sent so the worker is + never left stopped. + + Returns: + Event metadata dict for ``_injection_events``. + """ + pid = _worker_process_pid(wrapper) + outcome: dict[str, Any] = {"type": "sigstop", "duration_s": duration_s, "pid": pid} + if pid is None: + outcome["skipped"] = "no_process" + return outcome + stopped = _signal_process(pid, signal.SIGSTOP, "SIGSTOP") + outcome["sigstop_sent"] = stopped + cont_sent = False + try: + if stopped and duration_s > 0: + deadline = time.monotonic() + duration_s + while time.monotonic() < deadline: + if should_stop is not None and should_stop(): + outcome["interrupted"] = True + break + time.sleep(min(0.1, max(0.0, deadline - time.monotonic()))) + finally: + if stopped: + cont_sent = _signal_process(pid, signal.SIGCONT, "SIGCONT") + outcome["sigcont_sent"] = cont_sent + return outcome + + +def _execute_sigkill(wrapper: Any) -> dict[str, Any]: + """SIGKILL the worker subprocess. + + Returns: + Event metadata dict for ``_injection_events``. + """ + pid = _worker_process_pid(wrapper) + outcome: dict[str, Any] = {"type": "sigkill", "pid": pid} + if pid is None: + outcome["skipped"] = "no_process" + return outcome + outcome["sigkill_sent"] = _signal_process(pid, signal.SIGKILL, "SIGKILL") + if wrapper.process is not None: + try: + wrapper.process.wait(timeout=10.0) + except subprocess.TimeoutExpired: + logger.warning("[injector] SIGKILL: pid %d did not exit within 10s", pid) + return outcome + + # --------------------------------------------------------------------------- # Metrics-thread helpers # --------------------------------------------------------------------------- @@ -417,6 +620,7 @@ def __init__( log_scanner_poll_interval_s: float = 0.5, metrics_scrape_interval_s: float = 30.0, metrics_scrape_timeout_s: float = 5.0, + injector_poll_interval_s: float = 1.0, ) -> None: """Construct a marathon harness. @@ -437,6 +641,10 @@ def __init__( the metrics scrape. Short by design — a slow scrape is recorded as a miss rather than blocking the metrics thread past its next scheduled scrape. + injector_poll_interval_s: Poll cadence (seconds) for the + injector thread while waiting for the next scheduled + event. Tests pass a smaller value to keep wall-clock + latency bounded. Raises: ValueError: If the YAML is malformed or its @@ -455,10 +663,13 @@ def __init__( self._log_scanner_poll_interval_s: float = log_scanner_poll_interval_s self._metrics_scrape_interval_s: float = metrics_scrape_interval_s self._metrics_scrape_timeout_s: float = metrics_scrape_timeout_s + self._injector_poll_interval_s: float = injector_poll_interval_s # Cluster + worker tracking (populated by setup()). self._cluster: Any = None # tuple returned by setup_disagg_cluster self._worker_specs: list[WorkerLaunchSpec] = [] + self._tracked_workers: list[_TrackedWorker] = [] + self._marathon_start_monotonic: float = 0.0 # Thread handles (populated by start()). self._load_thread: Optional[threading.Thread] = None @@ -487,6 +698,30 @@ def setup(self) -> None: """ logger.info("[harness] setup() — stub: cluster not actually launched") + def bind_tracked_workers( + self, + ctx_workers: list[Any], + gen_workers: list[Any], + ctx_specs: list[WorkerLaunchSpec], + gen_specs: list[WorkerLaunchSpec], + ) -> None: + """Register live ``ProcessWrapper`` handles for the injector thread. + + Called by ``setup()`` once ``setup_disagg_cluster`` returns. + Shadow ``WorkerLaunchSpec`` entries must align 1:1 with the + wrapper lists (same ordering as ``setup_disagg_cluster``); + ``zip(..., strict=True)`` enforces this and raises ``ValueError`` + on a length mismatch. + """ + self._tracked_workers = [ + _TrackedWorker(spec=spec, wrapper=wrapper) + for spec, wrapper in zip(ctx_specs, ctx_workers, strict=True) + ] + [ + _TrackedWorker(spec=spec, wrapper=wrapper) + for spec, wrapper in zip(gen_specs, gen_workers, strict=True) + ] + self._worker_specs = list(ctx_specs) + list(gen_specs) + def start(self) -> None: """Spawn the five worker threads. Returns immediately. @@ -496,7 +731,8 @@ def start(self) -> None: stop()`` completes cleanly without waiting out the ``wait_until_done`` timeout. """ - logger.info("[harness] start() — spawning 5 stub threads") + self._marathon_start_monotonic = time.monotonic() + logger.info("[harness] start() — spawning worker threads") self._load_thread = threading.Thread( target=self._load_thread_body, name="stress-load", daemon=True ) @@ -671,11 +907,183 @@ def _canary_thread_body(self) -> None: def _injector_thread_body(self) -> None: """Fire SIGSTOP / SIGCONT / SIGKILL+respawn on the configured schedule. - Stub: no-op. Real implementation reads ``injections:`` schedule - from config and uses ``os.kill`` + ``_run_worker`` to act on - the shadow-tracked ``self._worker_specs``. + Reads ``injections:`` from ``stress_config``, waits until each + ``at_min`` offset from marathon start, then acts on + ``self._tracked_workers`` via ``os.kill``. SIGKILL entries + optionally relaunch the worker via ``disagg_test_utils._run_worker`` + and fail-fast if ``/health`` does not return 200 within + ``respawn_within_s``. + """ + schedule = _parse_injection_schedule(self.config.raw.get("injections") or []) + if not schedule: + logger.warning("[injector] no injections in stress_config; exiting immediately") + return + if not self._tracked_workers: + logger.warning( + "[injector] no tracked workers (setup() did not register any); " + "exiting without running the injection schedule" + ) + return + + pending = list(schedule) + logger.info( + "[injector] armed with %d injection(s) across %d tracked worker(s)", + len(pending), + len(self._tracked_workers), + ) + + try: + while pending and not self.stop_event.is_set() and not self.failed_event.is_set(): + elapsed_s = time.monotonic() - self._marathon_start_monotonic + next_inj = pending[0] + fire_at_s = next_inj.at_min * 60.0 + if elapsed_s < fire_at_s: + wait_s = min(self._injector_poll_interval_s, fire_at_s - elapsed_s) + self.stop_event.wait(timeout=wait_s) + continue + + pending.pop(0) + try: + tracked = _resolve_injection_target(next_inj.target, self._tracked_workers) + except ValueError as exc: + logger.error( + "[injector] skipping injection at T+%.1f min: %s", + next_inj.at_min, + exc, + ) + self._record_injection_event( + { + "at_min": next_inj.at_min, + "target": next_inj.target, + "skipped": str(exc), + } + ) + continue + + role_idx = f"{tracked.spec.role}_{tracked.spec.index}" + logger.info( + "[injector] T+%.1f min: %s on %s (pid=%s)", + next_inj.at_min, + next_inj.type, + role_idx, + _worker_process_pid(tracked.wrapper), + ) + event: dict[str, Any] = { + "at_min": next_inj.at_min, + "elapsed_s": elapsed_s, + "target": next_inj.target, + "role": tracked.spec.role, + "index": tracked.spec.index, + } + if next_inj.type == "sigstop": + event.update( + _execute_sigstop_pause( + tracked.wrapper, + next_inj.duration_s or 0.0, + should_stop=lambda: self.stop_event.is_set() + or self.failed_event.is_set(), + ) + ) + elif next_inj.type == "sigkill": + event.update(_execute_sigkill(tracked.wrapper)) + if next_inj.respawn_within_s is not None: + respawned = self._respawn_tracked_worker( + tracked, timeout_s=next_inj.respawn_within_s + ) + event["respawned"] = respawned + event["respawn_within_s"] = next_inj.respawn_within_s + if not respawned: + self.mark_failed( + f"injector: {role_idx} did not become healthy within " + f"{next_inj.respawn_within_s}s after SIGKILL" + ) + self._record_injection_event(event) + finally: + logger.debug("[injector] exiting; %d injection(s) remaining", len(pending)) + + def _record_injection_event(self, event: dict[str, Any]) -> None: + """Append one injector observation (thread-safe for single writer).""" + self._injection_events.append(event) + + def _respawn_tracked_worker(self, tracked: _TrackedWorker, *, timeout_s: float) -> bool: + """Relaunch a SIGKILLed worker and wait for ``/health``. + + Uses the shadow ``WorkerLaunchSpec`` recorded at setup time so + shared ``ProcessWrapper`` infrastructure stays unchanged. + + Args: + tracked: Worker to relaunch. + timeout_s: Maximum seconds to wait for HTTP 200 on ``/health``. + + Returns: + True if the respawned worker reports healthy within the + deadline; False otherwise. """ - logger.debug("[injector_thread] stub — exiting immediately") + from disagg_test_utils import _run_worker, get_free_port + + spec = tracked.spec + old = tracked.wrapper + + if old is not None and old.log_file is not None: + try: + old.log_file.close() + except OSError: + logger.debug("[injector] closing old log_file raised; ignoring") + + role_key = "ctx" if spec.role == "ctx" else "gen" + save_log = spec.log_path is not None + new_port = get_free_port() + try: + new_wrapper = _run_worker( + spec.model_name, + spec.worker_config, + role_key, + port=new_port, + work_dir=spec.work_dir, + device=spec.device, + save_log=save_log, + env=spec.env, + ) + except Exception: + logger.exception( + "[injector] failed to respawn %s_%d on port %d", + spec.role, + spec.index, + new_port, + ) + return False + tracked.wrapper = new_wrapper + spec.port = new_wrapper.port + if new_wrapper.log_path is not None: + spec.log_path = new_wrapper.log_path + + logger.info( + "[injector] respawned %s_%d on port %s; waiting up to %.0fs for /health", + spec.role, + spec.index, + new_wrapper.port, + timeout_s, + ) + + return self._wait_for_worker_health(new_wrapper.port, timeout_s=timeout_s) + + def _wait_for_worker_health(self, port: int, *, timeout_s: float) -> bool: + """Poll a worker's ``/health`` endpoint until healthy or timed out.""" + import requests + + deadline = time.monotonic() + timeout_s + while time.monotonic() < deadline: + if self.stop_event.is_set() or self.failed_event.is_set(): + return False + try: + request_timeout = min(5.0, max(0.1, deadline - time.monotonic())) + response = requests.get(f"http://localhost:{port}/health", timeout=request_timeout) + if response.status_code == 200: + return True + except requests.RequestException: + logger.debug("[injector] worker %d /health not ready yet", port) + self.stop_event.wait(timeout=min(1.0, max(0.0, deadline - time.monotonic()))) + return False def _log_scanner_thread_body(self) -> None: """Tail all worker logs; fail-fast on any hard-zero pattern hit. diff --git a/tests/integration/defs/stress_test/disagg_cancel/test_injector.py b/tests/integration/defs/stress_test/disagg_cancel/test_injector.py new file mode 100644 index 000000000000..daad46aacc0a --- /dev/null +++ b/tests/integration/defs/stress_test/disagg_cancel/test_injector.py @@ -0,0 +1,506 @@ +# SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Unit tests for ``DisaggCancellationStressHarness._injector_thread_body``. + +The injector thread has no dependency on a running disagg cluster: +feed it ``_TrackedWorker`` entries whose ``wrapper`` points at a +test-owned subprocess, drive a short ``injections:`` schedule, and +assert on ``_injection_events`` / ``failure_reason``. +""" + +from __future__ import annotations + +import subprocess +import sys +import textwrap +import threading +import time +from pathlib import Path +from types import ModuleType, SimpleNamespace +from typing import Any + +import pytest + +from .harness import ( + DisaggCancellationStressHarness, + WorkerLaunchSpec, + _execute_sigkill, + _execute_sigstop_pause, + _parse_injection_schedule, + _resolve_injection_target, + _TrackedWorker, +) + +# --------------------------------------------------------------------------- +# Fixtures / helpers +# --------------------------------------------------------------------------- + + +def _make_spec(role: str, index: int) -> WorkerLaunchSpec: + return WorkerLaunchSpec( + role=role, + index=index, + model_name="dummy", + worker_config={}, + work_dir="/tmp", + port=19000 + index, + device="0", + env={}, + log_path=None, + ) + + +def _make_wrapper(proc: subprocess.Popen[Any]) -> SimpleNamespace: + return SimpleNamespace(process=proc, port=0, log_file=None, log_path=None) + + +@pytest.fixture +def sleeping_subprocess() -> subprocess.Popen[Any]: + """Long-lived child for SIGSTOP/SIGKILL exercises.""" + proc = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(3600)"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + yield proc + if proc.poll() is None: + proc.kill() + proc.wait(timeout=5.0) + + +def _dummy_yaml_with_injections(injections_yaml: str) -> str: + """Build a minimal marathon YAML with a custom ``injections:`` block.""" + # Append injections after dedenting the static prefix only — embedding + # the dynamic block inside dedent() would strip its indentation. + prefix = textwrap.dedent( + """\ + hostname: localhost + model: dummy + backend: pytorch + context_servers: {} + generation_servers: {} + stress_config: + duration_min: 1 + kv_cache_manager: v1 + transceiver: cpp + injections: + """ + ) + injections_block = textwrap.indent(injections_yaml.strip(), " ") + return prefix + injections_block + "\n" + + +def _run_injector_until_done( + h: DisaggCancellationStressHarness, + timeout_s: float = 5.0, +) -> None: + h._marathon_start_monotonic = time.monotonic() + t = threading.Thread(target=h._injector_thread_body, name="test-injector", daemon=True) + t.start() + t.join(timeout=timeout_s) + assert not t.is_alive(), "injector thread did not exit within timeout" + + +# --------------------------------------------------------------------------- +# Parser / resolver unit tests +# --------------------------------------------------------------------------- + + +def test_parse_injection_schedule_sorts_and_validates() -> None: + raw = [ + {"at_min": 30, "type": "sigstop", "target": "gen_worker_0", "duration_s": 10}, + {"at_min": 15, "type": "sigstop", "target": "ctx_worker_0", "duration_s": 5}, + "not-a-dict", + {"at_min": 60, "type": "nope", "target": "gen_worker_0"}, + ] + specs = _parse_injection_schedule(raw) + assert [s.at_min for s in specs] == [15.0, 30.0] + assert specs[0].duration_s == 5.0 + + +def test_resolve_injection_target_fixed_index() -> None: + tracked = [ + _TrackedWorker(spec=_make_spec("gen", 0), wrapper=SimpleNamespace()), + _TrackedWorker(spec=_make_spec("gen", 1), wrapper=SimpleNamespace()), + ] + picked = _resolve_injection_target("gen_worker_0", tracked) + assert picked.spec.index == 0 + + +def test_resolve_injection_target_random_draws_from_role_pool() -> None: + tracked = [ + _TrackedWorker(spec=_make_spec("gen", 0), wrapper=SimpleNamespace()), + _TrackedWorker(spec=_make_spec("gen", 1), wrapper=SimpleNamespace()), + _TrackedWorker(spec=_make_spec("ctx", 0), wrapper=SimpleNamespace()), + ] + picked = _resolve_injection_target("gen_worker_random", tracked) + assert picked.spec.role == "gen" + + +def test_resolve_injection_target_unknown_raises() -> None: + with pytest.raises(ValueError, match="unsupported"): + _resolve_injection_target("bogus_target", []) + + +# --------------------------------------------------------------------------- +# Signal helpers (real subprocess) +# --------------------------------------------------------------------------- + + +def test_sigstop_pause_stops_then_resumes_process( + sleeping_subprocess: subprocess.Popen[Any], +) -> None: + wrapper = _make_wrapper(sleeping_subprocess) + assert sleeping_subprocess.poll() is None + + outcome = _execute_sigstop_pause(wrapper, duration_s=0.3) + + assert outcome["sigstop_sent"] is True + assert outcome["sigcont_sent"] is True + assert sleeping_subprocess.poll() is None + + +def test_sigkill_terminates_process(sleeping_subprocess: subprocess.Popen[Any]) -> None: + wrapper = _make_wrapper(sleeping_subprocess) + outcome = _execute_sigkill(wrapper) + assert outcome["sigkill_sent"] is True + assert sleeping_subprocess.wait(timeout=5.0) is not None + + +def test_sigstop_on_dead_process_is_skipped_gracefully() -> None: + proc = subprocess.Popen( + [sys.executable, "-c", "pass"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + proc.wait(timeout=5.0) + wrapper = _make_wrapper(proc) + outcome = _execute_sigstop_pause(wrapper, duration_s=0.1) + assert outcome["sigstop_sent"] is False + + +# --------------------------------------------------------------------------- +# Injector thread integration (harness wiring) +# --------------------------------------------------------------------------- + + +def test_injector_fires_immediate_sigstop( + tmp_path: Path, sleeping_subprocess: subprocess.Popen[Any] +) -> None: + yaml_path = tmp_path / "stress.yaml" + yaml_path.write_text( + _dummy_yaml_with_injections( + textwrap.dedent( + """\ + - at_min: 0 + type: sigstop + target: gen_worker_0 + duration_s: 0.2 + """ + ) + ) + ) + h = DisaggCancellationStressHarness(yaml_path, injector_poll_interval_s=0.02) + h._tracked_workers = [ + _TrackedWorker(spec=_make_spec("gen", 0), wrapper=_make_wrapper(sleeping_subprocess)) + ] + + _run_injector_until_done(h) + + assert len(h._injection_events) == 1 + ev = h._injection_events[0] + assert ev["type"] == "sigstop" + assert ev["sigstop_sent"] is True + assert sleeping_subprocess.poll() is None + + +def test_injector_runs_two_scheduled_events( + tmp_path: Path, sleeping_subprocess: subprocess.Popen[Any] +) -> None: + yaml_path = tmp_path / "stress.yaml" + yaml_path.write_text( + _dummy_yaml_with_injections( + textwrap.dedent( + """\ + - at_min: 0 + type: sigstop + target: gen_worker_0 + duration_s: 0.1 + - at_min: 0.05 + type: sigstop + target: gen_worker_0 + duration_s: 0.1 + """ + ) + ) + ) + h = DisaggCancellationStressHarness(yaml_path, injector_poll_interval_s=0.02) + h._tracked_workers = [ + _TrackedWorker(spec=_make_spec("gen", 0), wrapper=_make_wrapper(sleeping_subprocess)) + ] + + _run_injector_until_done(h, timeout_s=8.0) + + assert len(h._injection_events) == 2 + + +def test_injector_sigkill_with_respawn_timeout_trips_fail_fast( + tmp_path: Path, sleeping_subprocess: subprocess.Popen[Any], monkeypatch: pytest.MonkeyPatch +) -> None: + yaml_path = tmp_path / "stress.yaml" + yaml_path.write_text( + _dummy_yaml_with_injections( + textwrap.dedent( + """\ + - at_min: 0 + type: sigkill + target: gen_worker_0 + respawn_within_s: 1 + """ + ) + ) + ) + h = DisaggCancellationStressHarness(yaml_path, injector_poll_interval_s=0.02) + h._tracked_workers = [ + _TrackedWorker(spec=_make_spec("gen", 0), wrapper=_make_wrapper(sleeping_subprocess)) + ] + + monkeypatch.setattr(h, "_respawn_tracked_worker", lambda *a, **k: False) + + _run_injector_until_done(h) + + assert h.failed_event.is_set() + assert h.failure_reason is not None + assert "healthy within" in h.failure_reason + assert h._injection_events[0]["respawned"] is False + + +def test_injector_sigkill_respawn_success_records_event( + tmp_path: Path, sleeping_subprocess: subprocess.Popen[Any], monkeypatch: pytest.MonkeyPatch +) -> None: + yaml_path = tmp_path / "stress.yaml" + yaml_path.write_text( + _dummy_yaml_with_injections( + textwrap.dedent( + """\ + - at_min: 0 + type: sigkill + target: gen_worker_0 + respawn_within_s: 30 + """ + ) + ) + ) + h = DisaggCancellationStressHarness(yaml_path, injector_poll_interval_s=0.02) + tracked = _TrackedWorker(spec=_make_spec("gen", 0), wrapper=_make_wrapper(sleeping_subprocess)) + h._tracked_workers = [tracked] + respawned_proc: subprocess.Popen[Any] | None = None + + def _fake_respawn(t: _TrackedWorker, *, timeout_s: float) -> bool: + nonlocal respawned_proc + respawned_proc = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(3600)"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + tracked.wrapper = _make_wrapper(respawned_proc) + return True + + monkeypatch.setattr(h, "_respawn_tracked_worker", _fake_respawn) + + try: + _run_injector_until_done(h) + + assert not h.failed_event.is_set() + assert h._injection_events[0]["respawned"] is True + finally: + if respawned_proc is not None and respawned_proc.poll() is None: + respawned_proc.kill() + respawned_proc.wait(timeout=5.0) + + +def test_respawn_uses_allocated_port_and_bounded_health_wait( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + yaml_path = tmp_path / "stress.yaml" + yaml_path.write_text( + textwrap.dedent( + """\ + hostname: localhost + model: dummy + backend: pytorch + context_servers: {} + generation_servers: {} + stress_config: + duration_min: 1 + kv_cache_manager: v1 + transceiver: cpp + injections: [] + """ + ) + ) + h = DisaggCancellationStressHarness(yaml_path) + spec = _make_spec("gen", 0) + tracked = _TrackedWorker( + spec=spec, + wrapper=SimpleNamespace(process=None, port=19000, log_file=None, log_path=None), + ) + + def _fake_run_worker( + model_name: str, + worker_config: dict[str, Any], + role: str, + port: int, + work_dir: str, + device: str, + save_log: bool, + env: dict[str, str], + ) -> SimpleNamespace: + assert model_name == spec.model_name + assert worker_config == spec.worker_config + assert role == "gen" + assert port == 23456 + assert work_dir == spec.work_dir + assert device == spec.device + assert save_log is False + assert env == spec.env + return SimpleNamespace(process=None, port=port, log_file=None, log_path=None) + + health_calls: list[tuple[int, float]] = [] + + def _fake_health(port: int, *, timeout_s: float) -> bool: + health_calls.append((port, timeout_s)) + return True + + fake_disagg_test_utils = ModuleType("disagg_test_utils") + fake_disagg_test_utils.get_free_port = lambda: 23456 + fake_disagg_test_utils._run_worker = _fake_run_worker + monkeypatch.setitem(sys.modules, "disagg_test_utils", fake_disagg_test_utils) + monkeypatch.setattr(h, "_wait_for_worker_health", _fake_health) + + assert h._respawn_tracked_worker(tracked, timeout_s=7.0) is True + assert tracked.wrapper.port == 23456 + assert spec.port == 23456 + assert health_calls == [(23456, 7.0)] + + +def test_injector_exits_on_stop_event_before_distant_injection(tmp_path: Path) -> None: + yaml_path = tmp_path / "stress.yaml" + yaml_path.write_text( + _dummy_yaml_with_injections( + textwrap.dedent( + """\ + - at_min: 999 + type: sigstop + target: gen_worker_0 + duration_s: 1 + """ + ) + ) + ) + h = DisaggCancellationStressHarness(yaml_path, injector_poll_interval_s=0.05) + proc = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(3600)"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + try: + h._tracked_workers = [ + _TrackedWorker(spec=_make_spec("gen", 0), wrapper=_make_wrapper(proc)) + ] + h._marathon_start_monotonic = time.monotonic() + t = threading.Thread(target=h._injector_thread_body, daemon=True) + t.start() + time.sleep(0.15) + h.stop_event.set() + t.join(timeout=2.0) + assert not t.is_alive() + assert h._injection_events == [] + # Distant injection never fired — child should still be running. + time.sleep(0.1) + assert proc.poll() is None + finally: + proc.kill() + proc.wait(timeout=5.0) + + +def test_injector_no_tracked_workers_exits_without_events(tmp_path: Path) -> None: + yaml_path = tmp_path / "stress.yaml" + yaml_path.write_text( + _dummy_yaml_with_injections( + textwrap.dedent( + """\ + - at_min: 0 + type: sigstop + target: gen_worker_0 + duration_s: 1 + """ + ) + ) + ) + h = DisaggCancellationStressHarness(yaml_path) + _run_injector_until_done(h) + assert h._injection_events == [] + + +def test_injector_empty_schedule_exits_immediately(tmp_path: Path) -> None: + yaml_path = tmp_path / "stress.yaml" + yaml_path.write_text( + textwrap.dedent( + """\ + hostname: localhost + model: dummy + backend: pytorch + context_servers: {} + generation_servers: {} + stress_config: + duration_min: 1 + kv_cache_manager: v1 + transceiver: cpp + injections: [] + """ + ) + ) + h = DisaggCancellationStressHarness(yaml_path) + _run_injector_until_done(h) + assert h._injection_events == [] + + +def test_injector_skips_unknown_target_without_fail_fast( + tmp_path: Path, sleeping_subprocess: subprocess.Popen[Any] +) -> None: + yaml_path = tmp_path / "stress.yaml" + yaml_path.write_text( + _dummy_yaml_with_injections( + textwrap.dedent( + """\ + - at_min: 0 + type: sigstop + target: gen_worker_99 + duration_s: 0.1 + """ + ) + ) + ) + h = DisaggCancellationStressHarness(yaml_path, injector_poll_interval_s=0.02) + h._tracked_workers = [ + _TrackedWorker(spec=_make_spec("gen", 0), wrapper=_make_wrapper(sleeping_subprocess)) + ] + + _run_injector_until_done(h) + + assert not h.failed_event.is_set() + assert len(h._injection_events) == 1 + assert "skipped" in h._injection_events[0] diff --git a/tests/integration/defs/stress_test/disagg_cancel/test_log_scanner.py b/tests/integration/defs/stress_test/disagg_cancel/test_log_scanner.py index a0454b6328f3..747d58f86261 100644 --- a/tests/integration/defs/stress_test/disagg_cancel/test_log_scanner.py +++ b/tests/integration/defs/stress_test/disagg_cancel/test_log_scanner.py @@ -71,8 +71,8 @@ def harness_with_two_workers(tmp_path: Path): gen_log.touch() h._worker_specs = [ - make_spec("ctx", 0, ctx_log), - make_spec("gen", 0, gen_log), + make_spec("ctx", 0, log_path=ctx_log), + make_spec("gen", 0, log_path=gen_log), ] return h, ctx_log, gen_log @@ -326,8 +326,8 @@ def test_specs_with_none_log_path_skip_gracefully(tmp_path: Path): yaml_path.write_text(DUMMY_YAML) h = DisaggCancellationStressHarness(yaml_path) h._worker_specs = [ - make_spec("ctx", 0, None), - make_spec("gen", 0, None), + make_spec("ctx", 0, log_path=None), + make_spec("gen", 0, log_path=None), ] t = threading.Thread(target=h._log_scanner_thread_body, daemon=True) @@ -360,7 +360,7 @@ def test_no_hard_zero_patterns_exits_immediately(tmp_path: Path): h = DisaggCancellationStressHarness(yaml_path) ctx_log = tmp_path / "worker_ctx_18000.log" ctx_log.write_text("Broken promise: A\n") # would match in normal config - h._worker_specs = [make_spec("ctx", 0, ctx_log)] + h._worker_specs = [make_spec("ctx", 0, log_path=ctx_log)] t = threading.Thread(target=h._log_scanner_thread_body, daemon=True) t.start() @@ -401,7 +401,7 @@ def test_invalid_regex_is_skipped_with_warning(tmp_path: Path, caplog): h = DisaggCancellationStressHarness(yaml_path, log_scanner_poll_interval_s=0.02) ctx_log = tmp_path / "worker_ctx_18000.log" ctx_log.write_text("Broken promise: A\n") - h._worker_specs = [make_spec("ctx", 0, ctx_log)] + h._worker_specs = [make_spec("ctx", 0, log_path=ctx_log)] with caplog.at_level("ERROR"): fired = _run_scanner_until_failure_or_timeout(h) @@ -443,7 +443,7 @@ def _mark(reason: str) -> None: def test_log_source_poll_returns_false_when_file_absent(tmp_path: Path): - spec = make_spec("ctx", 0, tmp_path / "missing.log") + spec = make_spec("ctx", 0, log_path=tmp_path / "missing.log") src = _LogSource(spec=spec, path=Path(spec.log_path)) # type: ignore[arg-type] seen: list[str] = [] assert src.poll([("X", re.compile("X"))], _consume_marks(seen)) is False @@ -453,7 +453,7 @@ def test_log_source_poll_returns_false_when_file_absent(tmp_path: Path): def test_log_source_poll_returns_false_on_empty_read(tmp_path: Path): log = tmp_path / "empty.log" log.touch() - spec = make_spec("gen", 0, log) + spec = make_spec("gen", 0, log_path=log) src = _LogSource(spec=spec, path=log) seen: list[str] = [] assert src.poll([("X", re.compile("X"))], _consume_marks(seen)) is False @@ -464,7 +464,7 @@ def test_log_source_poll_returns_false_on_empty_read(tmp_path: Path): def test_log_source_close_is_idempotent(tmp_path: Path): log = tmp_path / "x.log" log.write_text("hello\n") - spec = make_spec("ctx", 0, log) + spec = make_spec("ctx", 0, log_path=log) src = _LogSource(spec=spec, path=log) src.poll([("X", re.compile("X"))], lambda r: None) src.close()