[None][fix] Bound GEN log sentinel wait

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:

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