diff --git a/jenkins/scripts/perf/submit.py b/jenkins/scripts/perf/submit.py index 72b176df3a95..f1c54fb13690 100755 --- a/jenkins/scripts/perf/submit.py +++ b/jenkins/scripts/perf/submit.py @@ -67,6 +67,17 @@ def _import_precheck_config(llm_src): # Test list parsing # --------------------------------------------------------------------------- # def _read_test_list_lines(test_list_path): + """Read runnable entries from a generated test list. + + Args: + test_list_path: Path to the generated test-list file. + + Returns: + Non-empty, non-comment test-list lines in source order. + + Raises: + ValueError: If the test list has no runnable entries. + """ with open(test_list_path, "r") as f: lines = [] for line in f: @@ -79,6 +90,14 @@ def _read_test_list_lines(test_list_path): def _pytest_command_tokens(script_prefix_lines): + """Parse the exported pytest command from a launch-script prefix. + + Args: + script_prefix_lines: Lines from the launch-script prefix. + + Returns: + Shell-parsed pytest command tokens, or an empty list when the export is absent. + """ pytest_command_line = next( (line for line in script_prefix_lines if "export pytestCommand=" in line), "" ) @@ -91,6 +110,15 @@ def _pytest_command_tokens(script_prefix_lines): def _pytest_option(tokens, option): + """Return a pytest option's value from command tokens. + + Args: + tokens: Shell-parsed pytest command tokens. + option: Option name to find, including its leading dashes. + + Returns: + The option value, or ``None`` when the option or its value is absent. + """ for index, token in enumerate(tokens): if token == option: return tokens[index + 1] if index + 1 < len(tokens) else None @@ -100,7 +128,14 @@ def _pytest_option(tokens, option): def _test_nodeid(test_line): - """Strip test-list markers from a line, matching pytest's selected nodeid.""" + """Strip test-list markers from a line to recover pytest's node ID. + + Args: + test_line: Test-list entry, optionally followed by execution markers. + + Returns: + The pytest node ID from the entry. + """ return re.split( r"\s+(?:XFAIL|SKIP|UNSTABLE|TIMEOUT)(?:\s|$)", test_line, @@ -109,6 +144,19 @@ def _test_nodeid(test_line): def _load_pytest_split_durations(tokens, llm_src): + """Load pytest-split duration data using the launcher's path fallback. + + Args: + tokens: Shell-parsed pytest command tokens. + llm_src: TensorRT-LLM source-tree path used for the repository fallback. + + Returns: + A pair containing the duration mapping and the path it was loaded from. + + Raises: + FileNotFoundError: If neither the configured path nor its repository fallback exists. + ValueError: If the duration data is not a mapping or legacy list of pairs. + """ durations_option = _pytest_option(tokens, "--durations-path") if durations_option: durations_path = durations_option @@ -138,7 +186,20 @@ def _load_pytest_split_durations(tokens, llm_src): def _select_least_duration_group(lines, durations, splits, group): - """Mirror pytest-split's LeastDurationAlgorithm exactly.""" + """Mirror pytest-split's ``LeastDurationAlgorithm`` exactly. + + Args: + lines: Test-list entries to distribute among the split groups. + durations: Mapping from pytest node IDs to recorded durations. + splits: Number of duration-balanced groups to create. + group: One-indexed group to return. + + Returns: + Entries assigned to the requested group, in their original test-list order. + + Raises: + ValueError: If ``splits`` is less than one or ``group`` is outside its range. + """ if splits < 1: raise ValueError(f"pytest --splits must be >= 1, got {splits}") if group < 1 or group > splits: diff --git a/tests/integration/defs/perf/test_perf_sanity.py b/tests/integration/defs/perf/test_perf_sanity.py index 4119913b1fa4..c45aed587803 100644 --- a/tests/integration/defs/perf/test_perf_sanity.py +++ b/tests/integration/defs/perf/test_perf_sanity.py @@ -106,6 +106,10 @@ def ensure_bench_serving_repo() -> str: # once EVERY ctx/gen worker has finished model load + autotune + warmup. AGG_SERVER_READY_TIMEOUT = 1800 DISAGG_SERVER_READY_TIMEOUT = 3600 +# GEN workers normally reap within seconds after benchmark_status is written. +# Keep this well below the whole-test timeout so a stuck multi-node srun cannot +# turn the optional log-flush synchronization into a pytest/Slurm cancellation. +GEN_LOG_SENTINEL_TIMEOUT = 120 def server_ready_timeout(default: int, mode: str) -> int: @@ -340,13 +344,14 @@ def parse_gen_worker_device_step_time( end-of-file are considered for gen_server_{i}.log — used to slice out a single client's iteration segment. - The log is read exactly once. The caller (DisaggTestCmds.run_cmd) blocks - on the gen_server_{i}.done sentinels before calling this, so every gen - srun has already exited and its &> aggregate log is fully flushed — there - is no partially-written tail to poll for. This replaces the earlier - settle-poll heuristic, which could return a mean over a truncated prefix - when it accepted the first repeated row count while the log was still - flushing across NFS (nvbugs 6487036 / 6487040). + The log is read exactly once. The caller (DisaggTestCmds.run_cmd) normally + waits for the gen_server_{i}.done sentinels first, so every gen srun has + exited and its &> aggregate log is fully flushed. If the dedicated + sentinel wait expires, the caller parses the current contents instead of + consuming the whole-test timeout; a missing metric then hard-fails before + upload. This replaces the earlier settle-poll heuristic, which could + accept a truncated prefix while the log was still flushing across NFS + (nvbugs 6487036 / 6487040 / 6487038). """ per_file_scans, _total_count = _scan_gen_worker_device_step_time( output_dir, num_gen_servers, start_offsets @@ -1319,7 +1324,11 @@ def wait_for_benchmark_ready( ) time.sleep(10) - def wait_for_gen_log_sentinels(self, poll_interval: float = 2.0) -> bool: + def wait_for_gen_log_sentinels( + self, + timeout: float = GEN_LOG_SENTINEL_TIMEOUT, + poll_interval: float = 2.0, + ) -> bool: """Block until every gen worker signals that its log is fully written. Each gen worker's srun in slurm_launch_draft.sh redirects all of its @@ -1328,25 +1337,26 @@ def wait_for_gen_log_sentinels(self, poll_interval: float = 2.0) -> bool: flushed). The benchmark writes benchmark_status *before* calling this, which is what lets the gen srun exit — so this is not circular. - Returns True once all sentinels exist, or False if self.timeout is - reached first. On False the caller still parses whatever is on disk: - the sentinel is a correctness optimization against reading a - mid-flush log (nvbugs 6487036 / 6487040), never a hang risk for CI. + Returns True once all sentinels exist, or False if the dedicated + sentinel timeout is reached first. On False the caller falls back to + parsing the current log contents. The bounded wait prevents a stuck + multi-node srun from consuming the whole-test timeout and triggering + Slurm's kill-on-bad-exit cascade (nvbugs 6487036 / 6487040 / 6487038). """ sentinels = [ os.path.join(self.test_output_dir, f"gen_server_{i}.done") for i in range(self.num_gen_servers) ] - start_time = time.time() + start_time = time.monotonic() while True: missing = [p for p in sentinels if not os.path.exists(p)] if not missing: print_info("All gen worker log sentinels present; log flush complete.") return True - elapsed_time = time.time() - start_time - if elapsed_time > self.timeout: + elapsed_time = time.monotonic() - start_time + if elapsed_time > timeout: print_info( - f"Timeout ({self.timeout}s) waiting for gen worker log " + f"Timeout ({timeout}s) waiting for gen worker log " f"sentinels {missing}; parsing current log contents." ) return False @@ -1355,6 +1365,36 @@ def wait_for_gen_log_sentinels(self, poll_interval: float = 2.0) -> bool: ) time.sleep(poll_interval) + def _append_gen_worker_device_step_time( + self, + pending_device_step_time: List[dict], + outputs: List[str], + ) -> None: + """Wait for GEN log flush, then append each pending client metric. + + A sentinel timeout is a bounded teardown fallback, not a reason to + discard metrics that are already present in the GEN logs. If the + fallback parse finds no usable metric, check_test_failure still fails + the gen_only run before results are uploaded. + """ + if not pending_device_step_time: + return + + self.wait_for_gen_log_sentinels() + for record in pending_device_step_time: + device_step_time_mean = parse_gen_worker_device_step_time( + self.test_output_dir, + self.num_gen_servers, + start_offsets=record["start_offsets"], + ) + if device_step_time_mean is None: + continue + summary_line = f"Average Per Iter Device Step Time (ms): {device_step_time_mean:.2f}" + with open(record["benchmark_file_path"], "a") as benchmark_ctx: + benchmark_ctx.write(f"\n{summary_line}\n") + idx = record["output_index"] + outputs[idx] = f"{outputs[idx]}\n{summary_line}\n" + def get_server_logs(self, server_idx: int) -> List[str]: server_logs = [] for i in range(self.num_ctx_servers): @@ -1468,13 +1508,16 @@ def run_cmd(self, server_idx: int) -> List[str]: elif self.disagg_serving_type == "BENCHMARK": # Perf-benchmark clients whose gen-worker device step time must be - # parsed once the gen logs are flushed. The parse is deferred out of + # parsed after the gen-log flush wait. The parse is deferred out of # the client loop because gen_server_*.log keeps being written until # the gen srun exits, and the gen srun only exits after # benchmark_status is written in the finally below. Parsing inside # the loop (as before) could read a truncated / not-yet-flushed log # and report a wrong mean (nvbugs 6487036 / 6487040). pending_device_step_time: List[dict] = [] + collect_device_step_time = ( + configs_for_idx is not None and configs_for_idx[2].benchmark_mode == "gen_only" + ) try: disagg_server_hostname, disagg_server_port = ( self._get_disagg_server_hostname_and_port(server_idx) @@ -1506,11 +1549,15 @@ def run_cmd(self, server_idx: int) -> List[str]: ) print_info(f"Starting benchmark. cmd is {client_cmd_with_port}") - # Snapshot gen_server log sizes so the per-client - # average covers only iterations driven by this client. - gen_log_start_offsets = gen_worker_log_sizes( - self.test_output_dir, self.num_gen_servers - ) + # Snapshot gen_server log sizes so the gen_only + # per-client average covers only iterations driven by + # this client. Other modes do not emit this metric and + # must not wait for the GEN teardown sentinel. + gen_log_start_offsets = None + if collect_device_step_time: + gen_log_start_offsets = gen_worker_log_sizes( + self.test_output_dir, self.num_gen_servers + ) bench_env = copy.deepcopy(os.environ) if client_config: @@ -1525,16 +1572,17 @@ def run_cmd(self, server_idx: int) -> List[str]: benchmark_ctx.write(output) outputs.append(output) - # Defer the gen-worker device-step-time parse until the - # gen logs are flushed (see below); remember where to - # write the summary back. - pending_device_step_time.append( - { - "output_index": len(outputs) - 1, - "benchmark_file_path": benchmark_file_path, - "start_offsets": gen_log_start_offsets, - } - ) + if collect_device_step_time: + # Defer the gen-worker device-step-time parse until + # the gen logs are flushed (see below); remember + # where to write the summary back. + pending_device_step_time.append( + { + "output_index": len(outputs) - 1, + "benchmark_file_path": benchmark_file_path, + "start_offsets": gen_log_start_offsets, + } + ) else: print_info( f"Skipping perf benchmark for client {client_idx}: " @@ -1570,27 +1618,12 @@ def run_cmd(self, server_idx: int) -> List[str]: # benchmark_status is written, so the gen workers can now stop and # their srun will exit and drop gen_server_{i}.done. Wait once for - # those sentinels (bounded by self.timeout), then parse each - # benchmark client's gen-worker device step time a single time: the - # flushed log is complete, so no settle polling is needed. Only - # gen_only runs emit prev_device_step_time; other modes parse to - # None and skip the summary line. - if pending_device_step_time: - self.wait_for_gen_log_sentinels() - for record in pending_device_step_time: - device_step_time_mean = parse_gen_worker_device_step_time( - self.test_output_dir, - self.num_gen_servers, - start_offsets=record["start_offsets"], - ) - if device_step_time_mean is not None: - summary_line = ( - f"Average Per Iter Device Step Time (ms): {device_step_time_mean:.2f}" - ) - with open(record["benchmark_file_path"], "a") as benchmark_ctx: - benchmark_ctx.write(f"\n{summary_line}\n") - idx = record["output_index"] - outputs[idx] = f"{outputs[idx]}\n{summary_line}\n" + # those sentinels (bounded independently of the whole-test timeout), + # then parse each benchmark client's gen-worker device step time a + # single time. A timeout falls back to the current log contents. + # Only gen_only runs populate this queue; other modes skip both the + # sentinel wait and device-step-time parsing. + self._append_gen_worker_device_step_time(pending_device_step_time, outputs) return outputs diff --git a/tests/unittest/scripts/test_perf_sanity_helpers.py b/tests/unittest/scripts/test_perf_sanity_helpers.py new file mode 100644 index 000000000000..f344529c2f35 --- /dev/null +++ b/tests/unittest/scripts/test_perf_sanity_helpers.py @@ -0,0 +1,82 @@ +# 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. + +import sys +from pathlib import Path + +import pytest + +pytest.importorskip("torch._inductor") + +REPO_ROOT = Path(__file__).resolve().parent.parent.parent.parent +sys.path.insert(0, str(REPO_ROOT / "tests" / "integration")) + +from defs.perf import test_perf_sanity as perf_sanity # noqa: E402 + + +def test_sentinel_timeout_falls_back_to_current_gen_logs( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + benchmark_log = tmp_path / "trtllm-benchmark.0.0.log" + benchmark_log.write_text("benchmark output", encoding="utf-8") + outputs = ["benchmark output"] + pending = [ + { + "output_index": 0, + "benchmark_file_path": str(benchmark_log), + "start_offsets": [10, 20], + } + ] + commands = perf_sanity.DisaggTestCmds( + server_cmds=[], + client_cmds={}, + timeout=1, + hostname="localhost", + disagg_serving_type="BENCHMARK", + num_ctx_servers=1, + num_gen_servers=2, + output_dir=str(tmp_path), + test_output_dir=str(tmp_path), + ) + + monkeypatch.setattr( + perf_sanity.DisaggTestCmds, + "wait_for_gen_log_sentinels", + lambda self: False, + ) + parse_calls: list[tuple[str, int, list[int]]] = [] + + def parse_device_step_time( + output_dir: str, + num_gen_servers: int, + start_offsets: list[int], + ) -> float: + parse_calls.append((output_dir, num_gen_servers, start_offsets)) + return 7.25 + + monkeypatch.setattr( + perf_sanity, + "parse_gen_worker_device_step_time", + parse_device_step_time, + ) + + commands._append_gen_worker_device_step_time(pending, outputs) + + assert parse_calls == [(str(tmp_path), 2, [10, 20])] + assert outputs == ["benchmark output\nAverage Per Iter Device Step Time (ms): 7.25\n"] + assert benchmark_log.read_text(encoding="utf-8").endswith( + "\nAverage Per Iter Device Step Time (ms): 7.25\n" + ) diff --git a/tests/unittest/scripts/test_perf_submit.py b/tests/unittest/scripts/test_perf_submit.py index ed4fb8dedea7..3a824627abbd 100644 --- a/tests/unittest/scripts/test_perf_submit.py +++ b/tests/unittest/scripts/test_perf_submit.py @@ -77,6 +77,28 @@ def ci_submit_module(monkeypatch: pytest.MonkeyPatch) -> ModuleType: return _load_module(SUBMIT_PATHS[0], monkeypatch) +def _select_ci_test_case_line( + ci_submit_module: ModuleType, + tmp_path: Path, + pytest_options: str, + split_group: int, +) -> str: + test_list_path = tmp_path / "test_list.txt" + test_list_path.write_text( + "perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-gb300-kimi]\n", + encoding="utf-8", + ) + script_prefix_lines = [ + f'export pytestCommand="pytest --splitting-algorithm least_duration {pytest_options}"' + ] + return ci_submit_module.select_test_case_line( + test_list_path, + tmp_path, + script_prefix_lines, + split_group=split_group, + ) + + @pytest.mark.parametrize( "concurrency", (True, 1.5, [], {}, "0", 0, "-1", -1, "1.5", "not-an-integer", None), @@ -230,20 +252,11 @@ def test_ci_submit_rejects_split_group_disagreement( ci_submit_module: ModuleType, tmp_path: Path, ) -> None: - test_list_path = tmp_path / "test_list.txt" - test_list_path.write_text( - "perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-gb300-kimi]\n", - encoding="utf-8", - ) - script_prefix_lines = [ - 'export pytestCommand="pytest --splitting-algorithm least_duration --splits 1 --group 1"' - ] - with pytest.raises(ValueError, match="disagrees with pytest --group"): - ci_submit_module.select_test_case_line( - test_list_path, + _select_ci_test_case_line( + ci_submit_module, tmp_path, - script_prefix_lines, + pytest_options="--splits 1 --group 1", split_group=2, ) @@ -252,21 +265,11 @@ def test_ci_submit_rejects_missing_pytest_split_durations( ci_submit_module: ModuleType, tmp_path: Path, ) -> None: - test_list_path = tmp_path / "test_list.txt" - test_list_path.write_text( - "perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-gb300-kimi]\n", - encoding="utf-8", - ) - script_prefix_lines = [ - 'export pytestCommand="pytest --splitting-algorithm least_duration ' - '--splits 1 --group 1 --durations-path /remote/.test_durations"' - ] - expected_path = tmp_path / "tests" / "integration" / "defs" / ".test_durations" with pytest.raises(FileNotFoundError, match=f"durations file not found: {expected_path}"): - ci_submit_module.select_test_case_line( - test_list_path, + _select_ci_test_case_line( + ci_submit_module, tmp_path, - script_prefix_lines, + pytest_options="--splits 1 --group 1 --durations-path /remote/.test_durations", split_group=1, )