[https://nvbugs/6609977][test] Scope gen_only per-iter device step time to the measured request

tests/integration/defs/perf/test_perf_sanity.py

@@ -204,6 +204,23 @@ def server_ready_timeout(default: int, mode: str) -> int:
 _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+)")
 
+# currank_total_requests = <current-rank-fetched>/<total-fetched> on every
+# iter line (py_executor.py: num_fetch_requests_cur_rank / num_fetch_requests).
+# The denominator increments when the worker fetches a new request, so it
+# marks request boundaries inside one client's log segment: with warmup
+# enabled the segment holds warmup (total = 1) then the measured request
+# (total = 2).
+_CURRANK_TOTAL_REQUESTS_RE = re.compile(r"currank_total_requests\s*=\s*\d+/(\d+)")
+
+# Usable rows to drop once the request count crosses skip_leading_requests,
+# mirroring the iter < 5 startup skip. prev_device_step_time on row N is the
+# device time of iter N-1, so post-boundary rows still carry earlier time:
+# row +0 is the previous request's last step, row +1 is the inter-request
+# gap (the stall this skip exists to exclude), and measured runs show +2
+# still settling with steady state from +3 on. 5 = 3 measured + 2 margin
+# for longer gaps, and costs <2% of the shortest (~256-row) request.
+_REQUEST_BOUNDARY_SETTLE_ROWS = 5
+
 
 def gen_worker_log_sizes(output_dir: str, num_gen_servers: int) -> List[int]:
     """Current byte size of each gen_server_{i}.log (0 if missing).
@@ -223,9 +240,19 @@ def _scan_gen_worker_device_step_time(
     output_dir: str,
     num_gen_servers: int,
     start_offsets: Optional[List[int]] = None,
+    skip_leading_requests: int = 0,
 ) -> Tuple[List[Tuple[Dict[int, Tuple[int, float]], int, float]], int]:
     """Single-pass scan of the gen logs.
 
+    When skip_leading_requests > 0 (warmup lanes), each file's stats cover
+    only rows after currank_total_requests' denominator exceeds that many
+    requests, plus a _REQUEST_BOUNDARY_SETTLE_ROWS settle skip — so the
+    warmup request and the inter-request stall stay out of the mean
+    (nvbugs 6609977 / TRTLLM-15394). If a file never crosses the boundary
+    (e.g. the warmup request was routed elsewhere, or the log format drops
+    currank_total_requests), that file falls back to the full-window stats,
+    which is the pre-warmup-aware behavior.
+
     Returns (per_file_scans, total_count):
       - per_file_scans: one entry per file that produced >=1 usable row, each
         a tuple (by_ngen, all_count, all_mean):
@@ -267,6 +294,14 @@ def _scan_gen_worker_device_step_time(
         by_ngen: Dict[int, Tuple[int, float]] = {}
         all_count = 0
         all_mean = 0.0
+        # Post-boundary accumulators mirror the full-window ones; used only
+        # when skip_leading_requests > 0 and the boundary is actually seen.
+        post_by_ngen: Dict[int, Tuple[int, float]] = {}
+        post_all_count = 0
+        post_all_mean = 0.0
+        seen_requests = 0
+        post_rows_seen = 0
+        skipped_post_boundary_dts: List[float] = []
         with open(log_path, errors="replace") as f:
             if seek_to:
                 f.seek(seek_to)
@@ -278,19 +313,48 @@ def _scan_gen_worker_device_step_time(
                     continue
                 total_count += 1
                 dt = float(m.group(2))
-                # All-iter fallback aggregate (every usable row).
+                # Per-ngen bucket key (rows without a parseable ngen only
+                # feed the all-iter fallback aggregates).
+                ngen_m = _NUM_GEN_TOKENS_RE.search(line)
+                ngen = int(ngen_m.group(1)) if ngen_m is not None else None
+                # Full-window aggregates (every usable row).
                 all_count += 1
                 all_mean += (dt - all_mean) / all_count
-                # Per-ngen bucket (only rows with a parseable ngen).
-                ngen_m = _NUM_GEN_TOKENS_RE.search(line)
-                if ngen_m is None:
-                    continue
-                ngen = int(ngen_m.group(1))
-                count, mean = by_ngen.get(ngen, (0, 0.0))
-                count += 1
-                mean += (dt - mean) / count
-                by_ngen[ngen] = (count, mean)
-        if all_count:
+                if ngen is not None:
+                    count, mean = by_ngen.get(ngen, (0, 0.0))
+                    count += 1
+                    mean += (dt - mean) / count
+                    by_ngen[ngen] = (count, mean)
+                if skip_leading_requests:
+                    req_m = _CURRANK_TOTAL_REQUESTS_RE.search(line)
+                    if req_m is not None:
+                        seen_requests = max(seen_requests, int(req_m.group(1)))
+                    if seen_requests > skip_leading_requests:
+                        post_rows_seen += 1
+                        if post_rows_seen <= _REQUEST_BOUNDARY_SETTLE_ROWS:
+                            skipped_post_boundary_dts.append(dt)
+                        else:
+                            post_all_count += 1
+                            post_all_mean += (dt - post_all_mean) / post_all_count
+                            if ngen is not None:
+                                count, mean = post_by_ngen.get(ngen, (0, 0.0))
+                                count += 1
+                                mean += (dt - mean) / count
+                                post_by_ngen[ngen] = (count, mean)
+        boundary_seen = seen_requests > skip_leading_requests
+        if boundary_seen:
+            print_info(
+                f"Dropped {len(skipped_post_boundary_dts)} post-boundary "
+                f"device-step rows from {log_path}: {skipped_post_boundary_dts}"
+            )
+            if post_all_count:
+                per_file_scans.append((post_by_ngen, post_all_count, post_all_mean))
+            else:
+                print_info(
+                    f"No device-step rows remain after request-boundary settling for "
+                    f"{log_path}: all_count={all_count}, post_rows_seen={post_rows_seen}"
+                )
+        elif all_count:
             per_file_scans.append((by_ngen, all_count, all_mean))
     return per_file_scans, total_count
 
@@ -328,6 +392,7 @@ def parse_gen_worker_device_step_time(
     output_dir: str,
     num_gen_servers: int,
     start_offsets: Optional[List[int]] = None,
+    skip_leading_requests: int = 0,
 ) -> Optional[float]:
     """Mean per-iter prev_device_step_time (ms) across all gen workers.
 
@@ -348,6 +413,11 @@ 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.
 
+    skip_leading_requests (warmup lanes pass 1) further narrows each file's
+    window to rows after that many requests have been admitted, so the
+    warmup request and the boundary stall stay out of the mean — see
+    _scan_gen_worker_device_step_time (nvbugs 6609977 / TRTLLM-15394).
+
     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
@@ -358,7 +428,7 @@ def parse_gen_worker_device_step_time(
     (nvbugs 6487036 / 6487040 / 6487038).
     """
     per_file_scans, _total_count = _scan_gen_worker_device_step_time(
-        output_dir, num_gen_servers, start_offsets
+        output_dir, num_gen_servers, start_offsets, skip_leading_requests
     )
     return _mean_at_mode_ngen(per_file_scans)
 
@@ -1436,6 +1506,7 @@ def _append_gen_worker_device_step_time(
                 self.test_output_dir,
                 self.num_gen_servers,
                 start_offsets=record["start_offsets"],
+                skip_leading_requests=record.get("skip_leading_requests", 0),
             )
             if device_step_time_mean is None:
                 continue
@@ -1652,6 +1723,13 @@ def run_cmd(self, server_idx: int) -> List[str]:
                                     "output_index": len(outputs) - 1,
                                     "benchmark_file_path": benchmark_file_path,
                                     "start_offsets": gen_log_start_offsets,
+                                    # With warmup on, the offset window spans
+                                    # warmup + measured request; tell the
+                                    # parser to drop the leading request
+                                    # (nvbugs 6609977 / TRTLLM-15394).
+                                    "skip_leading_requests": (
+                                        1 if client_config and client_config.warmup else 0
+                                    ),
                                 }
                             )
                     else:

tests/unittest/scripts/test_perf_sanity_helpers.py

@@ -91,13 +91,23 @@ def test_sentinel_timeout_falls_back_to_current_gen_logs(
 ) -> None:
     benchmark_log = tmp_path / "trtllm-benchmark.0.0.log"
     benchmark_log.write_text("benchmark output", encoding="utf-8")
-    outputs = ["benchmark output"]
+    warmup_benchmark_log = tmp_path / "trtllm-benchmark.0.1.log"
+    warmup_benchmark_log.write_text("warmup benchmark output", encoding="utf-8")
+    outputs = ["benchmark output", "warmup benchmark output"]
     pending = [
+        # Legacy record without skip_leading_requests: parses the full window.
         {
             "output_index": 0,
             "benchmark_file_path": str(benchmark_log),
             "start_offsets": [10, 20],
-        }
+        },
+        # Warmup-lane record: the parser must skip the leading request.
+        {
+            "output_index": 1,
+            "benchmark_file_path": str(warmup_benchmark_log),
+            "start_offsets": [30, 40],
+            "skip_leading_requests": 1,
+        },
     ]
     commands = perf_sanity.DisaggTestCmds(
         server_cmds=[],
@@ -116,14 +126,15 @@ def test_sentinel_timeout_falls_back_to_current_gen_logs(
         "wait_for_gen_log_sentinels",
         lambda self: False,
     )
-    parse_calls: list[tuple[str, int, list[int]]] = []
+    parse_calls: list[tuple[str, int, list[int], int]] = []
 
     def parse_device_step_time(
         output_dir: str,
         num_gen_servers: int,
         start_offsets: list[int],
+        skip_leading_requests: int = 0,
     ) -> float:
-        parse_calls.append((output_dir, num_gen_servers, start_offsets))
+        parse_calls.append((output_dir, num_gen_servers, start_offsets, skip_leading_requests))
         return 7.25
 
     monkeypatch.setattr(
@@ -134,8 +145,108 @@ def 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 parse_calls == [
+        (str(tmp_path), 2, [10, 20], 0),
+        (str(tmp_path), 2, [30, 40], 1),
+    ]
+    assert outputs == [
+        "benchmark output\nAverage Per Iter Device Step Time (ms): 7.25\n",
+        "warmup 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"
     )
+
+
+def _write_gen_worker_log(path: Path, rows: list[tuple[int, float, int]]) -> None:
+    """Write gen_server_{i}.log iter lines from (iter, prev_device_ms, total_requests)."""
+    lines = []
+    for iter_num, device_ms, total_requests in rows:
+        lines.append(
+            f"[TRT-LLM] [I] [_torch][RANK 0] iter = {iter_num}, "
+            f"num_scheduled_requests = 1, "
+            f"currank_total_requests = 0/{total_requests}, "
+            f"host_step_time = 5.0ms, prev_device_step_time = {device_ms}ms, "
+            "states = {'num_ctx_requests': 0, 'num_ctx_tokens': 0, "
+            "'num_generation_tokens': 4}"
+        )
+    path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+def test_skip_leading_requests_excludes_warmup_and_boundary_stall(
+    tmp_path: Path,
+    capfd: pytest.CaptureFixture[str],
+) -> None:
+    """Only measured steady rows enter the mean on a warmup lane.
+
+    The warmup request, the inter-request stall (row +1 after the request
+    count flips, because prev_device_step_time lags one row), and the
+    settle rows must all stay out.
+    """
+    rows = [(i, 10.0, 1) for i in range(5, 21)]  # warmup request
+    rows += [
+        (21, 10.0, 2),  # +0: previous request's last step
+        (22, 500.0, 2),  # +1: inter-request stall
+        (23, 50.0, 2),  # +2: settling
+        (24, 30.0, 2),  # +3, +4: margin rows
+        (25, 30.0, 2),
+    ]
+    rows += [(i, 20.0, 2) for i in range(26, 56)]  # measured steady state
+    _write_gen_worker_log(tmp_path / "gen_server_0.log", rows)
+
+    blended = perf_sanity.parse_gen_worker_device_step_time(str(tmp_path), 1)
+    measured_only = perf_sanity.parse_gen_worker_device_step_time(
+        str(tmp_path), 1, skip_leading_requests=1
+    )
+
+    assert measured_only == pytest.approx(20.0)
+    output = capfd.readouterr().out
+    assert "Dropped 5 post-boundary device-step rows" in output
+    assert "[10.0, 500.0, 50.0, 30.0, 30.0]" in output
+    # The full window folds in the warmup rows and the 500ms stall.
+    assert blended != pytest.approx(20.0)
+
+
+def test_skip_leading_requests_without_boundary_falls_back(tmp_path: Path) -> None:
+    """A log without a request boundary falls back to the full window.
+
+    This covers e.g. a warmup request that never reached this worker; the
+    metric must degrade to the pre-warmup-aware value instead of None.
+    """
+    rows = [(i, 10.0, 1) for i in range(5, 25)]
+    _write_gen_worker_log(tmp_path / "gen_server_0.log", rows)
+
+    assert perf_sanity.parse_gen_worker_device_step_time(
+        str(tmp_path), 1, skip_leading_requests=1
+    ) == pytest.approx(10.0)
+
+
+def test_skip_leading_requests_with_empty_measured_window_returns_none(
+    tmp_path: Path,
+    capfd: pytest.CaptureFixture[str],
+) -> None:
+    """A detected boundary must not fall back when no measured row survives."""
+    rows = [(i, 10.0, 1) for i in range(5, 15)]
+    rows += [(i, 500.0, 2) for i in range(15, 20)]
+    _write_gen_worker_log(tmp_path / "gen_server_0.log", rows)
+
+    assert (
+        perf_sanity.parse_gen_worker_device_step_time(str(tmp_path), 1, skip_leading_requests=1)
+        is None
+    )
+    output = capfd.readouterr().out
+    assert "all_count=15" in output
+    assert "post_rows_seen=5" in output
+
+
+def test_skip_leading_requests_zero_keeps_full_window(tmp_path: Path) -> None:
+    """skip_leading_requests=0 (every non-warmup lane) parses the whole window.
+
+    Behavior must match the pre-warmup-aware parser even when the log
+    contains a request boundary.
+    """
+    rows = [(i, 10.0, 1) for i in range(5, 15)]
+    rows += [(i, 30.0, 2) for i in range(15, 25)]
+    _write_gen_worker_log(tmp_path / "gen_server_0.log", rows)
+
+    assert perf_sanity.parse_gen_worker_device_step_time(str(tmp_path), 1) == pytest.approx(20.0)