[TRTLLM-13409][feat] Make a stalled executor worker init self-report

tensorrt_llm/_utils.py

@@ -700,11 +700,19 @@ def is_flashinfer_gdn_supported_arch(sm_version=None):
     return sm_version in (90, 100, 103)
 
 
-def print_all_stacks():
-    """Print stack traces for all threads"""
+def print_all_stacks(log: Optional[Callable[[str], None]] = None) -> None:
+    """Print stack traces for all threads
+
+    Args:
+        log: logging callable used to emit the traces; defaults to
+            ``logger.error``.  Callers dumping stacks for a condition that is
+            not (yet) a fault -- e.g. a slow but healthy startup -- should pass
+            ``logger.warning`` instead.
+    """
+    log = logger.error if log is None else log
     for thread_id, frame in sys._current_frames().items():
-        logger.error(f"Thread {thread_id} stack trace:\n" +
-                     "".join(traceback.format_stack(frame)))
+        log(f"Thread {thread_id} stack trace:\n" +
+            "".join(traceback.format_stack(frame)))
 
 
 def is_trace_enabled(env_var: str):

tensorrt_llm/executor/proxy.py

@@ -19,6 +19,7 @@
 import shutil
 import tempfile
 import threading
+import time
 import weakref
 from queue import Empty
 from typing import Dict, List, Optional, Union
@@ -43,11 +44,12 @@
 from .result import GenerationResult, IterationResult
 from .rpc import RPCClient
 from .rpc.rpc_common import RPCError, get_unique_ipc_addr
-from .utils import (EngineDeadError, ErrorResponse, RequestError,
-                    WorkerCommIpcAddrs, create_mpi_comm_session,
+from .utils import (WORKER_INIT_STALL_WARN_ENV, EngineDeadError, ErrorResponse,
+                    RequestError, WorkerCommIpcAddrs, create_mpi_comm_session,
                     get_spawn_proxy_process_env, is_llm_response,
                     multi_frontend_request_addr, multi_frontend_result_addr,
-                    namespace_client_id, print_alive_threads)
+                    namespace_client_id, print_alive_threads,
+                    worker_init_stall_warn_sec)
 from .worker import GenerationExecutorWorker, worker_main
 from .worker_process_monitor import WorkerProcessIdentity, WorkerProcessMonitor
 
@@ -605,6 +607,10 @@ def _start_dispatch_threads(self):
 
     def _start_executor_workers(self, worker_kwargs):
 
+        # Read the knob before anything is spawned: a bad value raises, and it
+        # must do so while there are still no ranks to leave behind.
+        stall_warn_sec = worker_init_stall_warn_sec()
+
         self_ref = weakref.ref(self)
 
         def mpi_done_callback(future: concurrent.futures.Future):
@@ -642,6 +648,17 @@ def mpi_done_callback(future: concurrent.futures.Future):
 
         self.workers_started = True
 
+        # This loop still exits only on the leader's status (ready or init
+        # error) or on a rank dying.  A third case exists -- every rank alive
+        # but nothing happening -- which neither exit covers; it is not given
+        # a deadline here (a slow-but-healthy startup must not be killed), but
+        # it is no longer silent: the stall report below, together with the
+        # per-rank stack dumps every worker emits on the same schedule, says
+        # which rank is stuck and where.
+        start_time = time.monotonic()
+        next_warn_time = (start_time +
+                          stall_warn_sec) if stall_warn_sec > 0 else None
+
         while True:
             if self.worker_init_status_queue.poll(1):
                 status = self.worker_init_status_queue.get()
@@ -654,6 +671,11 @@ def mpi_done_callback(future: concurrent.futures.Future):
                 raise RuntimeError("Executor worker died during initialization")
             self._handle_background_error()
 
+            now = time.monotonic()
+            if next_warn_time is not None and now >= next_warn_time:
+                next_warn_time = now + stall_warn_sec
+                logger.warning(self._worker_init_stall_report(now - start_time))
+
         ready_signal, error_trace = status[:2]
         if ready_signal != GenerationExecutorProxy.READY_SIGNAL:
             logger.error(f"Executor worker initialization error: {error_trace}")
@@ -666,6 +688,48 @@ def mpi_done_callback(future: concurrent.futures.Future):
 
         self._register_worker_processes(status)
 
+    def _worker_init_stall_report(self, elapsed: float) -> str:
+        """Describe a startup that has gone quiet, and where to attribute it.
+
+        The proxy has no IPC channel to ranks other than the leader before the
+        ready signal arrives, so it cannot collect worker stacks itself.  What
+        it can state is (a) whether any rank has exited -- which distinguishes
+        a stalled initialization from a crash, the two being indistinguishable
+        from the client side today -- and (b) where the per-rank stacks that
+        each worker dumps on the same schedule can be found.
+
+        (a) is only answerable when the session actually hands back worker
+        futures.  ``RemoteMpiCommSessionClient.submit()`` returns ``[]`` (the
+        ranks live in a separate ``mgmn_leader_node`` process under
+        ``trtllm-llmapi-launch``), so on that session type the proxy has *no*
+        liveness signal here and must say so rather than read an empty list as
+        "everyone is fine" -- the same empty-list trap ``pre_shutdown()``
+        documents below.  ``check_worker_error()`` on the session is the
+        authoritative channel there; it is not consulted from this report
+        because reading it consumes the death notice that
+        ``_check_remote_worker_death()`` acts on.
+        """
+        total = len(self.mpi_futures)
+        if total:
+            running = sum(1 for fut in self.mpi_futures if not fut.done())
+            liveness = (
+                f"{running}/{total} worker task(s) are still running, so no "
+                f"rank has exited (a stalled initialization, not a worker "
+                f"crash)")
+        else:
+            liveness = (
+                "whether a rank has exited cannot be told from here: this MPI "
+                "session hands back no worker futures, so the proxy has no "
+                "liveness signal during startup and this report can neither "
+                "confirm nor rule out a crashed rank. The session's "
+                "check_worker_error() channel is authoritative for that")
+        return (f"Executor worker initialization has not completed after "
+                f"{elapsed:.0f}s; {liveness}. Every rank dumps its own thread "
+                f"stacks on the same schedule: search the worker logs for 'has "
+                f"not finished initialization' to see which rank is stuck and "
+                f"where. Tune or silence this report with "
+                f"{WORKER_INIT_STALL_WARN_ENV}.")
+
     def _register_worker_processes(self, status: tuple) -> None:
         """Register identities returned by locally spawned MPI workers.
 

tensorrt_llm/executor/utils.py

@@ -16,6 +16,7 @@
 import asyncio
 import concurrent.futures
 import ctypes
+import math
 import os
 import re
 import sys
@@ -300,6 +301,65 @@ def is_llm_response(instance):
     return hasattr(instance, "has_error")
 
 
+# --- Executor worker startup (initialization) stall reporting ---------------
+#
+# Both the proxy and every worker rank read this.  It is an environment
+# variable rather than an LlmArgs field on purpose: the ``TRTLLM``-prefixed
+# environment is forwarded verbatim to spawned MPI ranks
+# (``MpiPoolSession._start_mpi_pool``), so a single ``export`` sets the
+# proxy-side report and the per-rank stack dumps together, and CI can set it
+# without touching the protected LLM API surface.  It sits next to the existing
+# ``TRTLLM_WORKER_PRINT_STACKS_PERIOD`` / ``TRTLLM_WORKER_DISABLE_GC`` worker
+# knobs.
+
+#: Seconds of silence before the proxy logs a startup-stall report and every
+#: worker rank dumps its own thread stacks.  Diagnostics only -- nothing is
+#: killed and no deadline is imposed -- so this is safe to enable by default
+#: even for a startup that is slow but healthy (e.g. a 671B checkpoint loading
+#: from a cold NFS mount).  Set <= 0 to silence; anything that is not a finite
+#: number is a hard error rather than a silent fallback (see
+#: :func:`float_from_env`).
+WORKER_INIT_STALL_WARN_ENV = "TRTLLM_WORKER_INIT_STALL_WARN_SEC"
+WORKER_INIT_STALL_WARN_DEFAULT_SEC = 600.0
+
+
+def float_from_env(name: str, default: float) -> float:
+    """Read a float-valued environment variable, rejecting anything unusable.
+
+    Unset or blank takes ``default``.  Anything else that is not a finite
+    number raises ``ValueError`` rather than falling back, because this knob
+    configures the reporting that a wedged startup depends on: silently
+    substituting the default would hide the misconfiguration behind exactly
+    the silence this reporting exists to remove, and the variable is only ever
+    set deliberately, so there is no legitimate value to preserve.
+
+    ``nan`` in particular must not survive.  Every comparison against it is
+    False, so it slips past the ``period <= 0`` disable check, and
+    ``threading.Event.wait(nan)`` then returns False immediately (measured:
+    ~10us, no exception) instead of sleeping -- turning the watchdog into a
+    hot loop that dumps every thread's stack on every rank.  ``inf`` is no
+    better: ``Event.wait(inf)`` raises ``OverflowError`` from inside the
+    watchdog thread, killing the reporting silently.
+    """
+    raw = os.getenv(name)
+    if raw is None or not raw.strip():
+        return default
+    try:
+        value = float(raw)
+    except ValueError as e:
+        raise ValueError(f"Invalid {name}={raw!r}: expected a number of "
+                         f"seconds (default {default}).") from e
+    if not math.isfinite(value):
+        raise ValueError(f"Invalid {name}={raw!r}: expected a *finite* number "
+                         f"of seconds (default {default}).")
+    return value
+
+
+def worker_init_stall_warn_sec() -> float:
+    return float_from_env(WORKER_INIT_STALL_WARN_ENV,
+                          WORKER_INIT_STALL_WARN_DEFAULT_SEC)
+
+
 def print_alive_threads():
     assert enable_llm_debug(
     ), "print_alive_threads must be called with enable_llm_debug() enabled"

tensorrt_llm/executor/worker.py

@@ -26,14 +26,57 @@
 from .request import CancellingRequest, GenerationRequest
 from .rpc_worker_mixin import RpcWorkerMixin
 from .utils import (ErrorResponse, IntraProcessQueue, RequestError,
-                    WorkerCommIpcAddrs)
+                    WorkerCommIpcAddrs, worker_init_stall_warn_sec)
 from .worker_process_monitor import capture_worker_process_identity
 
 __all__ = [
     "GenerationExecutorWorker",
 ]
 
 
+def _worker_init_stall_watchdog(init_done: threading.Event,
+                                period: float) -> None:
+    """Report where *this* rank is stuck while its initialization is pending.
+
+    Runs in a daemon thread in every rank.  During initialization the proxy
+    only has an IPC channel to the rank-0 leader, so a rank that is alive but
+    wedged -- typically inside a collective -- is invisible to it: the only
+    process that can say where rank N is stuck is rank N.  Dumping stacks is
+    read-only and never interferes with a slow-but-healthy startup, which is
+    why this is armed by default.
+
+    The dump relies on the wedged call having released the GIL (true for the
+    torch/NCCL collectives this targets).  A rank stuck while holding the GIL
+    cannot report, and simply stays silent.
+    """
+    rank = mpi_rank()
+    pid = os.getpid()
+    waited = 0.0
+    while not init_done.wait(period):
+        waited += period
+        # WARNING, not ERROR: a very large checkpoint loading from a cold mount
+        # is slow, not faulty, and this fires on that run too.
+        logger.warning(
+            f"Executor worker rank {rank} (pid {pid}) has not finished "
+            f"initialization after {waited:.0f}s; dumping the stacks of all "
+            f"threads on this rank.")
+        print_all_stacks(log=logger.warning)
+
+
+def _arm_worker_init_stall_watchdog(
+        init_done: threading.Event) -> Optional[threading.Thread]:
+    """Start the init stall watchdog unless it is disabled by configuration."""
+    period = worker_init_stall_warn_sec()
+    if period <= 0:
+        return None
+    watchdog = threading.Thread(target=_worker_init_stall_watchdog,
+                                args=(init_done, period),
+                                name="worker_init_stall_watchdog",
+                                daemon=True)
+    watchdog.start()
+    return watchdog
+
+
 class GenerationExecutorWorker(RpcWorkerMixin, BaseWorker):
 
     def __init__(
@@ -194,6 +237,11 @@ def _print_stacks():
                                                daemon=True)
         print_stacks_thread.start()
 
+    # Arm before the first collective: everything from here to the end of
+    # engine construction is init-phase work that can wedge a rank.
+    worker_init_done = threading.Event()
+    _arm_worker_init_stall_watchdog(worker_init_done)
+
     mpi_comm().barrier()
 
     if llm_args is not None and llm_args.env_overrides:
@@ -352,8 +400,18 @@ def notify_proxy_threads_to_quit():
             error_msg = (e, traceback.format_exc())
             if not worker_init_status_queue.notify_with_retry(error_msg):
                 logger.error("Failed to deliver error message to proxy")
+        worker_init_done.set()
         return
 
+    if not is_leader:
+        # A subordinate's startup ends with construction: it blocks in
+        # block_subordinates() below and has nothing further to report.  The
+        # leader stays armed until it has actually delivered the ready signal,
+        # because that -- not the end of construction -- is when the proxy's
+        # wait ends, and a leader wedged in between is exactly the case the
+        # proxy cannot see for itself.
+        worker_init_done.set()
+
     # Optionally disable GC (default: not disabled)
     if os.getenv("TRTLLM_WORKER_DISABLE_GC", "0") == "1":
         gc.disable()
@@ -372,10 +430,29 @@ def notify_proxy_threads_to_quit():
 
                 # Send ready signal with confirmation
                 ready_msg = (ready_signal, None, worker_process_identities)
-                if not worker_init_status_queue.notify_with_retry(ready_msg):
-                    logger.warning(
-                        "Failed to deliver ready signal to proxy, continuing anyway"
-                    )
+                ready_delivered = worker_init_status_queue.notify_with_retry(
+                    ready_msg)
+                if ready_delivered:
+                    # The proxy's startup wait has ended, so this rank has
+                    # nothing more to contribute to it.
+                    worker_init_done.set()
+                else:
+                    # Do NOT disarm here. Delivery failing is the one case
+                    # where the proxy is still blocked in its startup wait
+                    # with no idea why -- it is looking for a ready signal
+                    # that will never arrive. Disarming would remove the last
+                    # remaining source of information about that, leaving a
+                    # silent hang: this rank healthy and serving, the proxy
+                    # waiting forever, and nothing reporting either fact.
+                    # Leaving the watchdog armed keeps the per-rank stall
+                    # reports coming, which is exactly the diagnostic this PR
+                    # exists to provide.
+                    logger.error(
+                        "Failed to deliver the ready signal to the proxy. This "
+                        "rank is initialized and will continue serving, but the "
+                        "proxy may still be waiting for a signal it will never "
+                        "receive; leaving the init stall watchdog armed so the "
+                        "condition keeps being reported.")
                 if resource_governor_queue is not None:
                     # Swap rank 0 to the proxy IPC queue after construction.
                     # The resource-governor flag is already enabled on all