[https://nvbugs/6336747][fix] Fail fast when executor worker stalls

* Why?

A stuck or disconnected executor worker left the proxy blocked
indefinitely: the request queue uses an unbounded send HWM with no
send timeout, so request_queue.put -> socket.send never returned once
the worker stopped draining, and the error monitor never tripped. In
CI this could surface as a ~1h hang ending in an opaque timeout kill.
The stall itself is non-deterministic and not yet root-caused.

* What?

Make the failure fast and legible instead:

- Bound request submission: poll the socket for send-readiness and
check worker liveness, raising RequestError if the worker has not
accepted the request within a timeout.
- Add a progress watchdog to the error monitor that marks the worker
stalled and aborts in-flight requests when no result arrives while
requests are outstanding.
- Honor the previously-ignored timeout in GenerationResult.result()
and bound the per-request wait in the VideoMME evaluator.
- On a detected stall, signal the worker (SIGUSR1/faulthandler) to
dump all thread stacks so the next occurrence is diagnosable.

This mitigates the hang and captures worker state; it does not fix
the underlying intermittent stall.

Signed-off-by: William Zhang <133824995+2ez4bz@users.noreply.github.com>

Author: 2ez4bz

tensorrt_llm/evaluate/audio_asr.py

@@ -20,7 +20,6 @@
 from typing import Any, Iterable, NamedTuple, Optional
 
 import soundfile
-from tqdm import tqdm
 
 import tensorrt_llm.profiler as profiler
 from tensorrt_llm.inputs import (
@@ -36,7 +35,13 @@
 from tensorrt_llm.logger import logger
 from tensorrt_llm.sampling_params import SamplingParams
 
-from .interface import Evaluator, get_chat_template_kwargs, get_model_context
+from .interface import (
+    RESULT_WAIT_TIMEOUT_SECS,
+    Evaluator,
+    get_chat_template_kwargs,
+    get_model_context,
+)
+from .progress import tqdm_with_time_prefix
 
 
 class MultimodalASRSample(NamedTuple):
@@ -174,15 +179,19 @@ def evaluate(
         input_context = self._make_input_context(llm)
         dataset = _load_local_hf_dataset(self.dataset_path, self.split)
         num_samples = self._get_num_samples(dataset)
-        samples = list(tqdm(self._iter_samples(dataset), desc="Loading samples", total=num_samples))
+        samples = list(
+            tqdm_with_time_prefix(
+                self._iter_samples(dataset), desc="Loading samples", total=num_samples
+            )
+        )
         inputs = [
             self._make_input(llm, sample, input_context)
-            for sample in tqdm(samples, desc="Loading inputs")
+            for sample in tqdm_with_time_prefix(samples, desc="Loading inputs")
         ]
         futures = []
         references = []
         scoring_samples = []
-        for sample, request_input in tqdm(
+        for sample, request_input in tqdm_with_time_prefix(
             zip(samples, inputs, strict=True), desc="Submitting requests", total=len(samples)
         ):
             params = (
@@ -197,7 +206,10 @@ def evaluate(
             )
             references.append(sample.transcript)
             scoring_samples.append(_sample_for_scoring(sample))
-        outputs = [future.result() for future in tqdm(futures, desc="Fetching responses")]
+        outputs = [
+            future.result(timeout=RESULT_WAIT_TIMEOUT_SECS)
+            for future in tqdm_with_time_prefix(futures, desc="Fetching responses")
+        ]
 
         profiler.stop("trtllm exec")
         elapsed_time = profiler.elapsed_time_in_sec("trtllm exec")

tensorrt_llm/evaluate/interface.py

@@ -29,6 +29,15 @@
 from ..logger import logger
 from ..sampling_params import SamplingParams
 
+# Per-request upper bound (seconds) on how long an evaluator waits for a single response before
+# failing fast. A stalled or dead executor worker would otherwise block `future.result()`
+# indefinitely, turning an evaluation into potential hangs.
+# This is a backstop: it is intentionally larger than the executor's stall watchdog
+# (`TLLM_EXECUTOR_STALL_TIMEOUT_SECS`, default 300s) so the watchdog's more-informative
+# `RequestError` normally surfaces first; no healthy single request should come close to it.
+RESULT_WAIT_TIMEOUT_SECS = float(
+    os.environ.get("TLLM_EVAL_RESULT_TIMEOUT_SECS", "600"))
+
 
 def get_chat_template_kwargs(
         template_owner: Any,
@@ -145,7 +154,7 @@ def evaluate(self,
             auxiliaries.append(aux)
         results = []
         for output in tqdm(outputs, desc="Fetching responses"):
-            results.append(output.result())
+            results.append(output.result(timeout=RESULT_WAIT_TIMEOUT_SECS))
 
         if self.output_dir:
             dump_inference_results(self.output_dir, results,

tensorrt_llm/evaluate/lm_eval.py

@@ -44,8 +44,8 @@
 from ..llmapi import RequestOutput
 from ..logger import logger
 from ..sampling_params import SamplingParams
-from .interface import (Evaluator, dump_inference_results,
-                        get_chat_template_kwargs)
+from .interface import (RESULT_WAIT_TIMEOUT_SECS, Evaluator,
+                        dump_inference_results, get_chat_template_kwargs)
 
 # NOTE: lm_eval uses "<image>" as the default image placeholder
 # https://github.com/EleutherAI/lm-evaluation-harness/blob/7f04db12d2f8e7a99a0830d99eb78130e1ba2122/lm_eval/models/hf_vlms.py#L25
@@ -176,7 +176,7 @@ def generate_until(self, requests, disable_tqdm: bool = False) -> List[str]:
         for output in tqdm(results,
                            desc="Fetching responses",
                            disable=disable_tqdm):
-            outputs.append(output.result())
+            outputs.append(output.result(timeout=RESULT_WAIT_TIMEOUT_SECS))
 
         if self.output_dir:
             dump_inference_results(self.output_dir, outputs,
@@ -434,7 +434,7 @@ def generate_until(self, requests, disable_tqdm: bool = False) -> List[str]:
         for output in tqdm(results,
                            desc="Fetching responses",
                            disable=disable_tqdm):
-            outputs.append(output.result())
+            outputs.append(output.result(timeout=RESULT_WAIT_TIMEOUT_SECS))
 
         if self.output_dir:
             dump_inference_results(self.output_dir, outputs,

tensorrt_llm/evaluate/progress.py

@@ -0,0 +1,34 @@
+# 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.
+from datetime import datetime
+from typing import Any
+
+from tqdm import tqdm
+
+_TIME_PREFIX_BAR_FORMAT = "{current_time} {l_bar}{bar}{r_bar}"
+
+
+class _TimePrefixTqdm(tqdm):
+    @property
+    def format_dict(self) -> dict[str, Any]:
+        format_dict = super().format_dict
+        format_dict["current_time"] = datetime.now().strftime("%H:%M:%S")
+        return format_dict
+
+
+def tqdm_with_time_prefix(*args: Any, **kwargs: Any) -> _TimePrefixTqdm:
+    """Return a tqdm progress bar with the current time rendered before the description."""
+    kwargs.setdefault("bar_format", _TIME_PREFIX_BAR_FORMAT)
+    return _TimePrefixTqdm(*args, **kwargs)

tensorrt_llm/executor/ipc.py

@@ -181,6 +181,22 @@ def poll(self, timeout: int) -> bool:
         else:
             return False
 
+    def poll_send(self, timeout: float) -> bool:
+        """Return True if a message can be sent within *timeout* seconds.
+
+        Args:
+            timeout (float): Timeout in seconds.
+
+        Returns:
+            For a PAIR socket whose peer has disconnected, the socket enters mute state, and this
+            returns `False` instead of blocking, letting callers detect a dead/disconnected peer
+            rather than blocking forever in `send()`.
+        """
+        self.setup_lazily()
+        self._check_thread_safety()
+        return bool(
+            self.socket.poll(timeout=int(timeout * 1000), flags=zmq.POLLOUT))
+
     def put(self, obj: Any, routing_id: Optional[bytes] = None):
         self.setup_lazily()
         self._check_thread_safety()