diff --git a/tests/integration/defs/perf/test_gen_worker_device_step_time_parser.py b/tests/integration/defs/perf/test_gen_worker_device_step_time_parser.py new file mode 100644 index 000000000000..ce649b18a8f5 --- /dev/null +++ b/tests/integration/defs/perf/test_gen_worker_device_step_time_parser.py @@ -0,0 +1,239 @@ +# 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. +"""Unit tests for the gen-worker per-iter device-step-time parser. + +These are pure-CPU tests over synthetic ``gen_server_{i}.log`` content — no GPU, +no model. They pin the behavior of ``parse_gen_worker_device_step_time`` and its +helpers ``_scan_gen_worker_device_step_time`` / ``_mean_at_mode_ngen``. + +Regression coverage for nvbugs 6487036 / 6487040: a gen_only disagg run whose +gen worker logged perfectly good ``prev_device_step_time`` values still parsed +the ``mean_gen_worker_per_iter_device_step_time`` metric as ``None`` — because +PR #16298 began requiring a parseable ``num_generation_tokens`` on every line +and dropped a whole worker when none matched. A ``None`` metric then raised a +spurious ``check_test_failure``. The fix falls back to an un-bucketed all-iter +mean when ``num_generation_tokens`` is unparsable on every row of a worker, so +a present metric is never lost. + +``test_perf_sanity`` pulls heavy integration imports (conftest -> +``tensorrt_llm.bindings``, the OpenSearch DB utils, ...) at module load, so this +test reads a fixed source slice out of that module and ``exec``-s just the +self-contained parser block with a stubbed ``print_info`` rather than importing +the whole module. The block is purely regex + file IO + arithmetic, so exec-ing +it in isolation is faithful. +""" + +import os +import re +import time +from typing import Dict, List, Optional, Tuple + +import pytest + +_THIS_DIR = os.path.dirname(os.path.abspath(__file__)) +_TPS_PATH = os.path.join(_THIS_DIR, "test_perf_sanity.py") + + +def _load_parser_namespace() -> dict: + """Exec the parser block out of test_perf_sanity.py into an isolated ns. + + We slice from the ``_DEVICE_STEP_TIME_RE`` module constant through the end + of ``parse_gen_worker_device_step_time`` (the marker is the next top-level + ``def add_perf_metric_value``). This avoids importing test_perf_sanity's + GPU/DB dependency chain while running the exact shipped source. + """ + src = open(_TPS_PATH).read() + start = src.index("_DEVICE_STEP_TIME_RE = re.compile") + end = src.index("def add_perf_metric_value(") + block = src[start:end] + ns = { + "re": re, + "os": os, + "time": time, + "Optional": Optional, + "List": List, + "Dict": Dict, + "Tuple": Tuple, + # print_info is imported at module scope in test_perf_sanity; stub it. + "print_info": lambda *a, **k: None, + } + # _TPS_PATH is a fixed, repository-local source file (the sibling + # test_perf_sanity.py), not external/untrusted input; exec-ing a slice of it + # is how we run the shipped parser without its GPU/DB import chain. + exec(compile(block, _TPS_PATH, "exec"), ns) # noqa: S102 + return ns + + +_NS = _load_parser_namespace() +parse_gen_worker_device_step_time = _NS["parse_gen_worker_device_step_time"] +_scan_gen_worker_device_step_time = _NS["_scan_gen_worker_device_step_time"] +_mean_at_mode_ngen = _NS["_mean_at_mode_ngen"] + + +def _iter_line(iter_n: int, dt_ms, ngen_render: Optional[str]) -> str: + """Build a gen-worker iter log line matching the real emitter format. + + ``ngen_render`` is the verbatim ``'num_generation_tokens': <...>`` fragment + to embed in the states dict, or ``None`` to omit the key entirely. + """ + states = "'num_ctx_requests': 0, 'num_ctx_tokens': 0" + if ngen_render is not None: + states += f", {ngen_render}" + states += ", 'cached_kv_tokens': 0" + return ( + f"[07/22/2026-00:00:00] [TRT-LLM] [I] [_torch][RANK 0] " + f"iter = {iter_n}, global_rank = 0, rank = 0, " + f"num_scheduled_requests = 1, kv_cache_util = 0.100, " + f"currank_total_requests = 1/1, host_step_time = 5.0ms, " + f"prev_device_step_time = {dt_ms}ms, timestamp = 2026-07-22 00:00:00, " + f"states = {{{states}}}" + ) + + +def _write_gen_log(output_dir: str, idx: int, lines: List[str]) -> None: + with open(os.path.join(output_dir, f"gen_server_{idx}.log"), "w") as f: + f.write("\n".join(lines) + "\n") + + +# A short settle window so the polling loop returns promptly in unit tests +# (two scans of static files agree immediately). +_FAST = dict(settle_timeout=0.2, poll_interval=0.01) + + +def test_healthy_bare_int_uses_mode_ngen_bucket(tmp_path): + """Bare-int num_generation_tokens: the steady-state mode-ngen mean is used. + + Two ngen buckets; the larger-count bucket (256, 20 rows @ 15.0) is the mode + and wins over a 3-row 512 bucket @ 99.0. + """ + lines = [_iter_line(i, 15.0, "'num_generation_tokens': 256") for i in range(5, 25)] + lines += [_iter_line(i, 99.0, "'num_generation_tokens': 512") for i in range(25, 28)] + _write_gen_log(str(tmp_path), 0, lines) + assert parse_gen_worker_device_step_time(str(tmp_path), 1, **_FAST) == pytest.approx(15.0) + + +def test_iters_below_five_are_skipped(tmp_path): + """Iters 0-4 (KV-transfer + warmup) never contribute to the mean.""" + lines = [_iter_line(i, 1000.0, "'num_generation_tokens': 4") for i in range(0, 5)] + lines += [_iter_line(i, 12.0, "'num_generation_tokens': 4") for i in range(5, 15)] + _write_gen_log(str(tmp_path), 0, lines) + assert parse_gen_worker_device_step_time(str(tmp_path), 1, **_FAST) == pytest.approx(12.0) + + +def test_na_device_step_time_is_ignored(tmp_path): + """A non-numeric prev_device_step_time = N/A line does not match / crash.""" + lines = [_iter_line(5, "N/A", "'num_generation_tokens': 4")] # unmatched + lines += [_iter_line(i, 14.0, "'num_generation_tokens': 4") for i in range(6, 16)] + _write_gen_log(str(tmp_path), 0, lines) + assert parse_gen_worker_device_step_time(str(tmp_path), 1, **_FAST) == pytest.approx(14.0) + + +def test_missing_num_generation_tokens_falls_back_to_all_iter_mean(tmp_path): + """A worker whose lines omit num_generation_tokens must not parse to None. + + Regression for nvbugs 6487036 / 6487040: it falls back to the un-bucketed + all-iter mean of prev_device_step_time instead of dropping the worker. + """ + lines = [_iter_line(i, 13.5, ngen_render=None) for i in range(5, 25)] + _write_gen_log(str(tmp_path), 0, lines) + result = parse_gen_worker_device_step_time(str(tmp_path), 1, **_FAST) + assert result is not None + assert result == pytest.approx(13.5) + + +@pytest.mark.parametrize( + "ngen_render", + [ + "'num_generation_tokens': tensor(256)", + "'num_generation_tokens': np.int64(256)", + ], +) +def test_unparsable_num_generation_tokens_falls_back(tmp_path, ngen_render): + """An unparsable num_generation_tokens render still yields a value. + + When num_generation_tokens renders in a form the bucket regex can't read + (e.g. a tensor / numpy scalar repr) on every row, fall back to the all-iter + mean rather than dropping the worker to None. + """ + lines = [_iter_line(i, 12.9, ngen_render) for i in range(5, 25)] + _write_gen_log(str(tmp_path), 0, lines) + result = parse_gen_worker_device_step_time(str(tmp_path), 1, **_FAST) + assert result is not None + assert result == pytest.approx(12.9) + + +def test_mixed_workers_bucket_and_fallback_average(tmp_path): + """Multi-worker average combines a bucketed worker with a fallback worker. + + worker0 (bucketed) -> 15.0, worker1 (ngen absent) -> 20.0, mean across + workers -> 17.5. + """ + _write_gen_log( + str(tmp_path), + 0, + [_iter_line(i, 15.0, "'num_generation_tokens': 256") for i in range(5, 25)], + ) + _write_gen_log( + str(tmp_path), + 1, + [_iter_line(i, 20.0, ngen_render=None) for i in range(5, 25)], + ) + assert parse_gen_worker_device_step_time(str(tmp_path), 2, **_FAST) == pytest.approx(17.5) + + +def test_no_usable_lines_returns_none(tmp_path): + """No iter>=5 numeric prev_device_step_time line anywhere yields None. + + This is the only case that should still return None. + """ + _write_gen_log( + str(tmp_path), + 0, + [ + "some banner line without a device step time", + _iter_line(2, 99.0, "'num_generation_tokens': 4"), # iter < 5 + _iter_line(3, 99.0, ngen_render=None), # iter < 5 + ], + ) + assert parse_gen_worker_device_step_time(str(tmp_path), 1, **_FAST) is None + + +def test_missing_log_file_returns_none(tmp_path): + """No gen_server_*.log at all -> None (nothing to parse).""" + assert parse_gen_worker_device_step_time(str(tmp_path), 1, **_FAST) is None + + +def test_scan_counts_all_usable_rows_regardless_of_ngen(tmp_path): + """_scan total_count counts every iter>=5 numeric row (the settle signal). + + Rows whose num_generation_tokens did not parse are still counted, and the + per-file record still carries the all-iter fallback aggregate for them. + """ + # Distinct step times for the bucketed vs fallback-only rows so the asserts + # can tell the mode-ngen bucket mean apart from the all-iter mean: a healthy + # bucket must win over the fallback when any bucket exists. + lines = [_iter_line(i, 10.0, "'num_generation_tokens': 8") for i in range(5, 15)] + lines += [_iter_line(i, 99.0, ngen_render=None) for i in range(15, 20)] + _write_gen_log(str(tmp_path), 0, lines) + per_file_scans, total_count = _scan_gen_worker_device_step_time(str(tmp_path), 1) + assert total_count == 15 # 10 bucketed + 5 fallback-only, all iter>=5 + assert len(per_file_scans) == 1 + by_ngen, all_count, all_mean = per_file_scans[0] + assert by_ngen == {8: (10, pytest.approx(10.0))} + assert all_count == 15 + assert all_mean == pytest.approx((10 * 10.0 + 5 * 99.0) / 15) + # With a non-empty bucket present, selection uses the bucket mean (10.0), + # NOT the all-iter mean — the fallback only fires when buckets are empty. + assert _mean_at_mode_ngen(per_file_scans) == pytest.approx(10.0) diff --git a/tests/integration/defs/perf/test_perf_sanity.py b/tests/integration/defs/perf/test_perf_sanity.py index 0234e06dba7c..a36c6992bc0b 100644 --- a/tests/integration/defs/perf/test_perf_sanity.py +++ b/tests/integration/defs/perf/test_perf_sanity.py @@ -151,6 +151,15 @@ def ensure_bench_serving_repo() -> str: # is not) so the scanner can bucket rows by ngen without silently dropping # any line whose states dict is printed before prev_device_step_time. See # _scan_gen_worker_device_step_time. +# +# num_generation_tokens can legitimately fail to match on a given line: the +# steady-state `states` dict is a raw repr and, while every current writer +# stores a bare int, the field can be absent on some prepare paths or on a +# partially-flushed tail line. When it fails to match on EVERY usable row of a +# gen worker, the scanner falls back to that worker's un-bucketed all-iter mean +# rather than dropping the worker (which would collapse a present metric to +# None and raise a spurious check_test_failure — nvbugs 6487036 / 6487040). See +# _scan_gen_worker_device_step_time / _mean_at_mode_ngen. _DEVICE_STEP_TIME_RE = re.compile(r"iter\s*=\s*(\d+),.*?prev_device_step_time\s*=\s*([\d.]+)\s*ms") _NUM_GEN_TOKENS_RE = re.compile(r"'num_generation_tokens':\s*(\d+)") @@ -169,19 +178,35 @@ def gen_worker_log_sizes(output_dir: str, num_gen_servers: int) -> List[int]: return sizes +# Per-file scan record: (by_ngen, all_iter_count, all_iter_mean). +# by_ngen -> num_generation_tokens -> (count, Welford mean) for rows +# whose num_generation_tokens parsed; the steady-state +# mode-ngen mean is taken from this when it is non-empty. +# all_iter_count -> number of iter >= 5 numeric prev_device_step_time rows in +# this file, regardless of whether num_generation_tokens +# parsed. +# all_iter_mean -> Welford mean of prev_device_step_time over those rows. +# all_iter_* is the fallback used when by_ngen is empty (num_generation_tokens +# was absent or rendered in an unparsable form on every row) so a file with a +# perfectly good prev_device_step_time stream is never dropped to None. +_GenFileScan = Tuple[Dict[int, Tuple[int, float]], int, float] + + def _scan_gen_worker_device_step_time( output_dir: str, num_gen_servers: int, start_offsets: Optional[List[int]] = None, -) -> Tuple[List[Dict[int, Tuple[int, float]]], int]: +) -> Tuple[List[_GenFileScan], int]: """Single-pass scan of the gen logs. - Returns (per_file_by_ngen, total_count): - - per_file_by_ngen: one dict per file that produced >=1 usable line, - mapping num_generation_tokens -> (count, Welford mean of - prev_device_step_time) over rows with iter >= 5 and a numeric - prev_device_step_time. Rows lacking num_generation_tokens on the same - line are skipped for the mean but still counted for settle detection. + Returns (per_file_scans, total_count): + - per_file_scans: one (by_ngen, all_iter_count, all_iter_mean) tuple per + file that produced >=1 usable line (iter >= 5 with a numeric + prev_device_step_time). ``by_ngen`` maps num_generation_tokens -> + (count, Welford mean of prev_device_step_time); rows whose + num_generation_tokens did not parse are omitted from ``by_ngen`` but + still fold into ``all_iter_count`` / ``all_iter_mean`` (and + ``total_count``). See _mean_at_mode_ngen for how the two are combined. - total_count: the number of iter >= 5 rows with a numeric prev_device_step_time across all files. This is monotonic as new lines flush across NFS (rows only get appended) so the caller can use @@ -195,7 +220,7 @@ def _scan_gen_worker_device_step_time( (model load) write partial multibyte sequences that would otherwise raise UnicodeDecodeError mid-scan. """ - per_file_by_ngen: List[Dict[int, Tuple[int, float]]] = [] + per_file_scans: List[_GenFileScan] = [] total_count = 0 for i in range(num_gen_servers): log_path = os.path.join(output_dir, f"gen_server_{i}.log") @@ -209,6 +234,8 @@ def _scan_gen_worker_device_step_time( ) by_ngen: Dict[int, Tuple[int, float]] = {} + all_count = 0 + all_mean = 0.0 with open(log_path, errors="replace") as f: if seek_to: f.seek(seek_to) @@ -219,41 +246,64 @@ def _scan_gen_worker_device_step_time( if int(m.group(1)) < 5: continue total_count += 1 + dt = float(m.group(2)) + # Fallback aggregate over every usable row, independent of + # whether num_generation_tokens parsed. + all_count += 1 + all_mean += (dt - all_mean) / all_count ngen_m = _NUM_GEN_TOKENS_RE.search(line) if ngen_m is None: continue ngen = int(ngen_m.group(1)) - dt = float(m.group(2)) count, mean = by_ngen.get(ngen, (0, 0.0)) count += 1 mean += (dt - mean) / count by_ngen[ngen] = (count, mean) - if by_ngen: - per_file_by_ngen.append(by_ngen) - return per_file_by_ngen, total_count + if all_count: + per_file_scans.append((by_ngen, all_count, all_mean)) + return per_file_scans, total_count def _mean_at_mode_ngen( - per_file_by_ngen: List[Dict[int, Tuple[int, float]]], + per_file_scans: List[_GenFileScan], ) -> Optional[float]: - """Aggregate per-file per-ngen buckets into a single mean. - - Within each file pick the num_generation_tokens value with the most - iterations (the mode) and take its Welford mean; ties break to the - largest ngen because the steady-state plateau is the upper of any tied - clusters. Mode is more robust than strict == max — a one-off spike where - a single iter's ngen briefly exceeds the sustained batch would otherwise - collapse the mean to 1-2 samples. Then average the per-file means across - workers. Returns None if no file had a usable row. + """Aggregate per-file scans into a single mean. + + Within each file, prefer the steady-state bucket: pick the + num_generation_tokens value with the most iterations (the mode) and take + its Welford mean; ties break to the largest ngen because the steady-state + plateau is the upper of any tied clusters. Mode is more robust than strict + == max — a one-off spike where a single iter's ngen briefly exceeds the + sustained batch would otherwise collapse the mean to 1-2 samples. + + When a file produced usable prev_device_step_time rows but none carried a + parseable num_generation_tokens (empty ``by_ngen``), fall back to that + file's un-bucketed all-iter mean instead of dropping the file. This keeps + a present metric from collapsing to None (and a spurious + check_test_failure) when the states dict omits num_generation_tokens or + renders it in an unexpected form (nvbugs 6487036 / 6487040); it degrades + to the pre-steady-state-restriction behavior only for that file. + + Then average the per-file means across workers. Returns None only if no + file had a usable row at all. """ means: List[float] = [] - for by_ngen in per_file_by_ngen: - if not by_ngen: - continue - _mode_ngen, (_count, mean) = max(by_ngen.items(), key=lambda kv: (kv[1][0], kv[0])) - means.append(mean) + fell_back = False + for by_ngen, all_count, all_mean in per_file_scans: + if by_ngen: + _mode_ngen, (_count, mean) = max(by_ngen.items(), key=lambda kv: (kv[1][0], kv[0])) + means.append(mean) + elif all_count: + means.append(all_mean) + fell_back = True if not means: return None + if fell_back: + print_info( + "parse_gen_worker_device_step_time: num_generation_tokens not " + "parseable on any row of at least one gen worker; using the " + "un-bucketed all-iter prev_device_step_time mean for that worker." + ) return sum(means) / len(means) @@ -275,7 +325,10 @@ def parse_gen_worker_device_step_time( below the steady-state cost. Using the mode (rather than strict == max) is robust against a single iter whose ngen briefly spikes above the sustained batch, which would otherwise collapse the mean to 1-2 samples. - Returns None if no usable line is found in any file. + If a worker logged usable prev_device_step_time rows but none carried a + parseable num_generation_tokens, that worker falls back to its un-bucketed + all-iter mean (see _mean_at_mode_ngen) so a present metric is never dropped + to None. Returns None only if no usable line is found in any file. When start_offsets is provided, only the bytes from start_offsets[i] to end-of-file are considered for gen_server_{i}.log — used to slice out a @@ -296,20 +349,20 @@ def parse_gen_worker_device_step_time( deadline = time.time() + settle_timeout prev_count = -1 while True: - per_file_by_ngen, total_count = _scan_gen_worker_device_step_time( + per_file_scans, total_count = _scan_gen_worker_device_step_time( output_dir, num_gen_servers, start_offsets ) # Non-empty and unchanged since the last poll → the flush has settled. if total_count > 0 and total_count == prev_count: - return _mean_at_mode_ngen(per_file_by_ngen) + return _mean_at_mode_ngen(per_file_scans) if time.time() >= deadline: - if per_file_by_ngen: + if per_file_scans: print_info( f"parse_gen_worker_device_step_time: settle_timeout " f"({settle_timeout}s) reached with {total_count} line(s); " "returning current mean." ) - return _mean_at_mode_ngen(per_file_by_ngen) + return _mean_at_mode_ngen(per_file_scans) return None prev_count = total_count time.sleep(poll_interval) diff --git a/tests/integration/test_lists/waives.txt b/tests/integration/test_lists/waives.txt index 6e48b1ebc11d..31dc744bd620 100644 --- a/tests/integration/test_lists/waives.txt +++ b/tests/integration/test_lists/waives.txt @@ -318,14 +318,9 @@ perf/test_perf_sanity.py::test_e2e[disagg_upload-e2e-gb300_kimi-k25-thinking-fp4 perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-b200_deepseek-r1-fp4_8k1k_con1536_ctx1_dep4_gen1_dep8_eplb0_mtp1_ccb-NIXL] SKIP (https://nvbugs/6478615) perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-b200_deepseek-r1-fp4_8k1k_con1_ctx1_dep4_gen1_tep8_eplb0_mtp3_ccb-NIXL] SKIP (https://nvbugs/6490049) perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-b200_deepseek-r1-fp4_8k1k_con256_ctx1_dep4_gen1_dep8_eplb0_mtp1_ccb-NIXL] SKIP (https://nvbugs/6490049) -perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-gb200_deepseek-r1-fp4_8k1k_con1024_ctx1_dep4_gen1_dep32_eplb0_mtp3_ccb-NIXL] SKIP (https://nvbugs/6490049) -perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-gb200_deepseek-r1-fp4_8k1k_con1_ctx1_dep4_gen1_tep8_eplb0_mtp3_ccb-NIXL] SKIP (https://nvbugs/6490049) perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-gb200_deepseek-v32-fp4_32k4k_con2048_ctx1_dep4_gen1_dep32_eplb288_mtp1_ccb-NIXL] SKIP (https://nvbugs/6374872) perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-gb200_deepseek-v32-fp4_32k4k_con256_ctx1_dep4_gen1_dep32_eplb0_mtp3_ccb-NIXL] SKIP (https://nvbugs/6490049) perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-gb200_deepseek-v32-fp4_32k4k_con256_ctx1_dep8_gen1_dep8_eplb0_mtp0_ccb-NIXL] SKIP (https://nvbugs/6374893) -perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-gb200_deepseek-v32-fp4_8k1k_con1_ctx1_dep4_gen1_tep8_eplb0_mtp3_ccb-NIXL] SKIP (https://nvbugs/6490049) -perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-gb200_deepseek-v32-fp4_8k1k_con4096_ctx1_dep4_gen1_dep32_eplb256_mtp0_ccb-NIXL] SKIP (https://nvbugs/6490049) -perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-gb200_gpt-oss-120b-fp4_8k1k_con128_ctx1_tp1_gen1_tp4_eplb0_mtp0_ccb-NIXL] SKIP (https://nvbugs/6490049) perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-gb200_gpt-oss-120b-fp4_8k1k_con4_ctx1_tp1_gen1_tp4_eplb0_mtp0_ccb-NIXL] SKIP (https://nvbugs/6490049) perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-gb200_kimi-k25-thinking-fp4_8k1k_con1024_ctx1_dep4_gen1_dep32_eplb416_mtp3_ccb-NIXL] SKIP (https://nvbugs/6379406) perf/test_perf_sanity.py::test_e2e[disagg_upload-gen_only-gb200_qwen3-235b-fp4_8k1k_con1_ctx1_tp1_gen1_tep4_eplb0_mtp0_ccb-NIXL] SKIP (https://nvbugs/6490049)