diff --git a/tests/integration/defs/perf/test_perf_sanity.py b/tests/integration/defs/perf/test_perf_sanity.py index a7aef4aee2fb..d6211edc8049 100644 --- a/tests/integration/defs/perf/test_perf_sanity.py +++ b/tests/integration/defs/perf/test_perf_sanity.py @@ -24,11 +24,12 @@ import socket import subprocess import time +from collections import deque from typing import Dict, List, NamedTuple, Optional, Tuple import pytest import yaml -from test_common.error_utils import report_error +from test_common.error_utils import check_error, report_error from test_common.http_utils import fail_if_proc_died, wait_for_endpoint_ready from test_common.perf_sanity_matching import get_client_match_keys, get_server_match_keys @@ -98,6 +99,142 @@ def ensure_bench_serving_repo() -> str: DEFAULT_TIMEOUT = 10800 + +# Bound the benchmark client run. +# +# The client subprocess had no timeout at all, and every other harness wait is +# bounded by DEFAULT_TIMEOUT (10800s) -- which sits *above* the pytest per-test +# marker, so none of them can expire first. A stall anywhere below the HTTP +# layer therefore surfaced only as "the client is still running", and the stage +# burned its whole Slurm allocation before something external killed it with no +# diagnostic. Measured: stages running 1.7-2.3h against 18-41 minute budgets, +# producing no results XML at all. +# +# One hour is deliberately generous against the largest per-test budget in the +# perf-sanity lists, so this bounds the pathological case without touching +# healthy long runs. Set to 0 to disable. +BENCHMARK_CLIENT_TIMEOUT_ENV_VAR_NAME = "TRTLLM_PERF_SANITY_CLIENT_TIMEOUT_SEC" +DEFAULT_BENCHMARK_CLIENT_TIMEOUT = 3600 + +# Grace period between SIGTERM and SIGKILL when stopping a server. A worker +# wedged in a non-interruptible native call never runs the Python signal +# handler, and the bare wait() this replaces would block teardown indefinitely. +SERVER_TERMINATE_GRACE_SEC = 60 + +# How much of each server log to attach when the client bound fires. +SERVER_LOG_TAIL_LINES = 60 + + +def _benchmark_client_timeout() -> Optional[int]: + """Effective client bound in seconds, or None when explicitly disabled.""" + raw = os.environ.get(BENCHMARK_CLIENT_TIMEOUT_ENV_VAR_NAME) + if raw is None: + return DEFAULT_BENCHMARK_CLIENT_TIMEOUT + try: + value = int(raw) + except ValueError: + print_info( + f"{BENCHMARK_CLIENT_TIMEOUT_ENV_VAR_NAME}={raw!r} is not an integer; " + f"falling back to {DEFAULT_BENCHMARK_CLIENT_TIMEOUT}s" + ) + return DEFAULT_BENCHMARK_CLIENT_TIMEOUT + return None if value <= 0 else value + + +def stop_process(proc, name: str, grace: int = SERVER_TERMINATE_GRACE_SEC) -> None: + """SIGTERM a server, then SIGKILL it if it has not exited within `grace`. + + Replaces a bare ``terminate(); wait()``. Teardown must not be able to hang: + a rank blocked in native code never reaches the Python signal handler, and + the unbounded wait would hold the whole allocation until Slurm intervenes. + """ + if proc.poll() is not None: + return + proc.terminate() + try: + proc.wait(timeout=grace) + return + except subprocess.TimeoutExpired: + print_info(f"{name} did not exit within {grace}s of SIGTERM; escalating to SIGKILL") + proc.kill() + try: + proc.wait(timeout=grace) + except subprocess.TimeoutExpired: + print_info(f"{name} is still alive after SIGKILL; leaving it to the harness") + + +def run_benchmark_client(cmd, env, server_logs) -> str: + """Run a benchmark client under a deadline. + + Preserves ``check_output`` semantics -- combined stdout/stderr returned on + success, ``CalledProcessError`` on a nonzero exit -- and adds the bound that + was missing: if the client neither finishes nor fails within the budget, + kill it and raise naming what it was waiting on, with the partial client + output and the server-side context attached. + + Without this the stage cannot fail on its own; only Slurm or Jenkins stops + it, hours later, with no results XML. + """ + timeout_s = _benchmark_client_timeout() + started = time.monotonic() + proc = subprocess.Popen(cmd, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + try: + raw, _ = proc.communicate(timeout=timeout_s) + except subprocess.TimeoutExpired: + elapsed = time.monotonic() - started + proc.kill() + # Drain whatever the client wrote before it was killed -- that partial + # output is usually the only record of how far the run got. + try: + raw, _ = proc.communicate(timeout=SERVER_TERMINATE_GRACE_SEC) + except subprocess.TimeoutExpired: + raw = b"" + partial = (raw or b"").decode(errors="replace") + + # Two sources, because neither alone is sufficient. check_error() + # matches ERROR_KEYWORDS, which are Python exception names -- it does + # NOT match "[TRT-LLM] [E]" lines, so a server that died the TRT-LLM way + # surfaces nothing. Deliberately not widening ERROR_KEYWORDS here: it + # also drives wait_for_endpoint_ready()'s fast-fail, where a benign [E] + # during startup would start failing healthy runs. Keyword hits are + # best-effort; a raw tail is always attached. + keyword_hits = [] + tails = [] + for log_path in server_logs or []: + base = os.path.basename(log_path) + for line_idx, line in check_error(log_path): + keyword_hits.append(f"{base}:{line_idx}: {line}") + try: + with open(log_path, "r", errors="replace") as handle: + # deque(maxlen=) streams the file and keeps only the tail. + # readlines() would materialise the whole log first, and + # measured perf-sanity gen logs reach 77-232 MB -- risking + # an OOM on the rank that is already failing, which would + # swallow the very timeout report we are assembling. + tail = list(deque(handle, maxlen=SERVER_LOG_TAIL_LINES)) + except OSError: + continue + if tail: + tails.append(f"--- tail of {base} ---\n" + "".join(tail)) + detail = "\n".join(keyword_hits[-20:]) if keyword_hits else "" + tail_blob = "\n".join(tails) if tails else "" + + raise RuntimeError( + f"Benchmark client made no progress for {elapsed:.0f}s " + f"(bound {timeout_s}s, set {BENCHMARK_CLIENT_TIMEOUT_ENV_VAR_NAME} to change " + f"it, 0 disables). The client was killed so the stage fails here instead of " + f"running to the harness timeout.\n" + f"--- server-side error keywords ---\n{detail}\n" + f"{tail_blob}\n" + f"--- last client output ---\n{partial[-4000:]}" + ) from None + + output = (raw or b"").decode(errors="replace") + if proc.returncode != 0: + raise subprocess.CalledProcessError(proc.returncode, cmd, output=output.encode()) + return output + + # Defaults for the server *ready* wait, separate from the whole-test timeout: # a server that is not healthy after this long is not going to be, and failing # here (with server-log tails, see wait_for_endpoint_ready) instead of at the @@ -1146,11 +1283,9 @@ def run_cmd(self, server_idx: int) -> List[str]: client_env = copy.deepcopy(os.environ) if client_config: client_env.update(client_config.to_env()) - output = subprocess.check_output( - client_cmd_with_port, - stderr=subprocess.STDOUT, - env=client_env, - ).decode() + output = run_benchmark_client( + client_cmd_with_port, client_env, [server_file_path] + ) with open(client_file_path, "w") as client_ctx: client_ctx.write(output) @@ -1176,8 +1311,7 @@ def run_cmd(self, server_idx: int) -> List[str]: finally: if server_proc: - server_proc.terminate() - server_proc.wait() + stop_process(server_proc, "server") return outputs @@ -1488,8 +1622,7 @@ def run_cmd(self, server_idx: int) -> List[str]: ) finally: print_info(f"Server {self.disagg_serving_type} stopped") - server_proc.terminate() - server_proc.wait() + stop_process(server_proc, "server") elif self.disagg_serving_type == "DISAGG_SERVER": try: @@ -1517,8 +1650,7 @@ def run_cmd(self, server_idx: int) -> List[str]: ) finally: print_info(f"Disagg server {self.disagg_serving_type} stopped") - disagg_server_proc.terminate() - disagg_server_proc.wait() + stop_process(disagg_server_proc, "disagg server") elif self.disagg_serving_type == "BENCHMARK": # Perf-benchmark clients whose gen-worker device step time must be @@ -1576,11 +1708,11 @@ def run_cmd(self, server_idx: int) -> List[str]: bench_env = copy.deepcopy(os.environ) if client_config: bench_env.update(client_config.to_env()) - output = subprocess.check_output( + output = run_benchmark_client( client_cmd_with_port, - env=bench_env, - stderr=subprocess.STDOUT, - ).decode() + bench_env, + self.get_server_logs(server_idx), + ) with open(benchmark_file_path, "w") as benchmark_ctx: benchmark_ctx.write(output) diff --git a/tests/unittest/others/test_perf_sanity_bounds.py b/tests/unittest/others/test_perf_sanity_bounds.py new file mode 100644 index 000000000000..5e754b3aaa33 --- /dev/null +++ b/tests/unittest/others/test_perf_sanity_bounds.py @@ -0,0 +1,194 @@ +# SPDX-FileCopyrightText: Copyright (c) 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. +"""Bounds on the perf-sanity harness waits. + +The perf-sanity stages could not fail on their own: the benchmark client +subprocess had no timeout, and every other harness wait was bounded at +DEFAULT_TIMEOUT (10800s), which sits above the pytest per-test marker. A stall +therefore surfaced only as "the client is still running" until Slurm or Jenkins +killed the stage hours later, with no results XML and no diagnostic. + +These tests pin the two bounds that close that: a deadline on the client, and +SIGTERM->SIGKILL escalation on server teardown. +""" + +import os +import subprocess +import sys +import time + +import pytest + +# Required to run in the CPU-Generic stage: tests/unittest/conftest.py's +# pytest_ignore_collect drops any file whose source lacks this literal when +# pytest runs with -m cpu_only. Nothing here needs a GPU. +pytestmark = pytest.mark.cpu_only + +_INTEGRATION = os.path.join(os.path.dirname(__file__), "..", "..", "integration") +if _INTEGRATION not in sys.path: + sys.path.insert(0, os.path.abspath(_INTEGRATION)) + +perf_sanity = pytest.importorskip("defs.perf.test_perf_sanity") + + +# --------------------------------------------------------------------------- +# The client deadline +# --------------------------------------------------------------------------- + + +def test_client_timeout_defaults_to_one_hour(monkeypatch): + """The default must be finite -- an unset env var previously meant 'forever'.""" + # delenv, not os.environ.pop: pop would drop the variable for the rest of + # the pytest process and silently change later tests in the same worker. + monkeypatch.delenv(perf_sanity.BENCHMARK_CLIENT_TIMEOUT_ENV_VAR_NAME, raising=False) + assert perf_sanity._benchmark_client_timeout() == 3600 + + +def test_client_timeout_honours_the_env_override(monkeypatch): + monkeypatch.setenv(perf_sanity.BENCHMARK_CLIENT_TIMEOUT_ENV_VAR_NAME, "120") + assert perf_sanity._benchmark_client_timeout() == 120 + + +def test_client_timeout_zero_disables_the_bound(monkeypatch): + """0 must mean 'no deadline' (None), not 'expire immediately'.""" + monkeypatch.setenv(perf_sanity.BENCHMARK_CLIENT_TIMEOUT_ENV_VAR_NAME, "0") + assert perf_sanity._benchmark_client_timeout() is None + + +def test_client_timeout_falls_back_when_the_env_var_is_garbage(monkeypatch): + """A typo in CI config must not silently restore the unbounded behaviour.""" + monkeypatch.setenv(perf_sanity.BENCHMARK_CLIENT_TIMEOUT_ENV_VAR_NAME, "not-a-number") + assert perf_sanity._benchmark_client_timeout() == 3600 + + +# --------------------------------------------------------------------------- +# Teardown escalation +# --------------------------------------------------------------------------- + + +def test_stop_process_reaps_a_cooperative_process(): + proc = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"]) + perf_sanity.stop_process(proc, "cooperative", grace=10) + assert proc.poll() is not None, "process should be reaped after SIGTERM" + + +def test_stop_process_escalates_to_sigkill_when_sigterm_is_ignored(): + """A rank wedged in native code never runs the SIGTERM handler. + + The bare terminate(); wait() this replaces would block teardown forever. + """ + ignores_sigterm = ( + "import signal, time\nsignal.signal(signal.SIGTERM, signal.SIG_IGN)\ntime.sleep(60)\n" + ) + proc = subprocess.Popen([sys.executable, "-c", ignores_sigterm]) + # Let the child install its handler before we signal it. + time.sleep(1.0) + + started = time.monotonic() + perf_sanity.stop_process(proc, "stubborn", grace=3) + elapsed = time.monotonic() - started + + assert proc.poll() is not None, "SIGKILL should have reaped the process" + assert elapsed < 30, f"stop_process took {elapsed:.1f}s; it must not wait out the full sleep" + + +def test_stop_process_is_a_noop_for_an_already_dead_process(): + proc = subprocess.Popen([sys.executable, "-c", "pass"]) + proc.wait() + perf_sanity.stop_process(proc, "already-dead", grace=5) + assert proc.poll() is not None + + +# --------------------------------------------------------------------------- +# The client runner itself: the stage must be able to fail on its own. +# --------------------------------------------------------------------------- + + +def test_client_output_is_returned_on_success(): + out = perf_sanity.run_benchmark_client( + [sys.executable, "-c", "print('benchmark done')"], dict(os.environ), [] + ) + assert "benchmark done" in out + + +def test_client_nonzero_exit_still_raises(): + """check_output semantics must survive: a failing client fails the test.""" + with pytest.raises(subprocess.CalledProcessError): + perf_sanity.run_benchmark_client( + [sys.executable, "-c", "import sys; sys.exit(3)"], dict(os.environ), [] + ) + + +def test_client_hang_is_bounded_and_names_the_knob(monkeypatch, tmp_path): + """The regression that mattered: a client that never returns. + + Before, this ran until Slurm killed the stage hours later. It must now + raise promptly, name the env var, and carry the server-side errors. + """ + monkeypatch.setenv(perf_sanity.BENCHMARK_CLIENT_TIMEOUT_ENV_VAR_NAME, "2") + server_log = tmp_path / "trtllm-serve.CTX_0.0.log" + server_log.write_text("some line\n[TRT-LLM] [E] Error in event loop: boom\n") + + started = time.monotonic() + with pytest.raises(RuntimeError) as excinfo: + perf_sanity.run_benchmark_client( + [sys.executable, "-c", "import time; time.sleep(120)"], + dict(os.environ), + [str(server_log)], + ) + elapsed = time.monotonic() - started + + msg = str(excinfo.value) + assert perf_sanity.BENCHMARK_CLIENT_TIMEOUT_ENV_VAR_NAME in msg + assert "made no progress" in msg + assert elapsed < 60, f"bound did not fire promptly ({elapsed:.1f}s)" + + +def test_client_hang_surfaces_server_side_errors(monkeypatch, tmp_path): + monkeypatch.setenv(perf_sanity.BENCHMARK_CLIENT_TIMEOUT_ENV_VAR_NAME, "2") + server_log = tmp_path / "trtllm-serve.GEN_0.0.log" + server_log.write_text("[TRT-LLM] [E] Error in event loop: kaboom\n") + + with pytest.raises(RuntimeError) as excinfo: + perf_sanity.run_benchmark_client( + [sys.executable, "-c", "import time; time.sleep(120)"], + dict(os.environ), + [str(server_log)], + ) + msg = str(excinfo.value) + # A "[TRT-LLM] [E]" line matches no ERROR_KEYWORDS entry, so the keyword + # scan finds nothing -- it must still reach the reader via the log tail. + assert "kaboom" in msg, ( + "the server-side error is the whole point of failing here rather than " + "letting the harness time out blind" + ) + assert "tail of" in msg + + +def test_client_hang_surfaces_keyword_errors_too(monkeypatch, tmp_path): + """The keyword scan still contributes when the log does match.""" + monkeypatch.setenv(perf_sanity.BENCHMARK_CLIENT_TIMEOUT_ENV_VAR_NAME, "2") + server_log = tmp_path / "trtllm-serve.CTX_0.0.log" + server_log.write_text("RuntimeError: engine exploded\n") + + with pytest.raises(RuntimeError) as excinfo: + perf_sanity.run_benchmark_client( + [sys.executable, "-c", "import time; time.sleep(120)"], + dict(os.environ), + [str(server_log)], + ) + msg = str(excinfo.value) + assert "engine exploded" in msg + assert "trtllm-serve.CTX_0.0.log:1:" in msg