Skip to content

[https://nvbugs/6435642][fix] Revert killed-worker detection (#16338, #16444) #16522

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
sunnyqgg wants to merge 2 commits into from
Closed

[https://nvbugs/6435642][fix] Revert killed-worker detection (#16338, #16444) #16522

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7dbdc5f
Revert "[https://nvbugs/6435642][fix] handle session reuse worker reg…
sunnyqgg Jul 17, 2026
67465e5
Revert "[https://nvbugs/6435642][fix] detect killed MPI executor work…
sunnyqgg Jul 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 6 additions & 40 deletions tensorrt_llm/executor/proxy.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,6 @@
get_spawn_proxy_process_env, is_llm_response,
print_alive_threads)
from .worker import GenerationExecutorWorker, worker_main
from .worker_process_monitor import WorkerProcessIdentity, WorkerProcessMonitor

__all__ = [
"GenerationExecutorProxy",
Expand Down Expand Up @@ -177,7 +176,6 @@ def __init__(

self.dispatch_result_thread: Optional[ManagedThread] = None
self.rpc_client: Optional[RPCClient] = None
self._worker_process_monitor = WorkerProcessMonitor()
self._start_executor_workers(worker_kwargs)

# Create RPC client after workers are started (worker starts RPC server)
Expand Down Expand Up @@ -226,19 +224,6 @@ def _check_mpi_futures(self) -> bool:
return True
return False

def _check_mpi_workers(self) -> bool:
"""Check OS process handles and MPI futures for worker death."""
dead_worker = self._worker_process_monitor.find_dead_worker()
if dead_worker is not None:
self._set_fatal_error(
RuntimeError("MPI worker rank "
f"{dead_worker.rank} (pid {dead_worker.pid}) "
"exited unexpectedly"))
if not self.doing_shutdown:
self.pre_shutdown()
return True
return self._check_mpi_futures()

def _drain_error_queue(self) -> bool:
"""Drain all queued errors, skipping per-request errors.

Expand Down Expand Up @@ -281,7 +266,7 @@ def check_health(self) -> bool:
if self._drain_error_queue():
return self._fatal_error is None and not self.doing_shutdown

if self._check_mpi_workers():
if self._check_mpi_futures():

@coderabbitai coderabbitai Bot Jul 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

Synchronously detect remote worker death. Remote sessions can have an empty mpi_futures list, so the current health check remains healthy until the monitor loop polls later.

  • tensorrt_llm/executor/proxy.py#L269-L269: call _check_remote_worker_death() after the future check and return unhealthy when it reports a death.
  • tests/unittest/executor/test_fatal_error_health_check.py#L200-L214: mirror that check in ConcreteProxyExecutor and add a fake remote-session test asserting immediate unhealthy status and shutdown.

As per path instructions, coverage is insufficient and needs the concrete test above.

📍 Affects 2 files
  • tensorrt_llm/executor/proxy.py#L269-L269 (this comment)
  • tests/unittest/executor/test_fatal_error_health_check.py#L200-L214
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tensorrt_llm/executor/proxy.py` at line 269, Update the health-check flow in
tensorrt_llm/executor/proxy.py at lines 269-269 to call
_check_remote_worker_death() after _check_mpi_futures() and return unhealthy
when a remote worker death is detected. In
tests/unittest/executor/test_fatal_error_health_check.py at lines 200-214,
mirror this behavior in ConcreteProxyExecutor and add a fake remote-session test
verifying immediate unhealthy status and shutdown.

Source: Path instructions

return False

return True
Expand Down Expand Up @@ -358,10 +343,9 @@ def _check_remote_worker_death(self) -> bool:
def _error_monitor_loop(self) -> None:
"""Background thread that reaps a dead engine and drives pre_shutdown.

Checks local MPI worker process handles and futures, remote-session
worker-death notifications, and the error queue using the shared
``_check_mpi_workers()``, ``_check_remote_worker_death()`` and
``_drain_error_queue()`` helpers.
Checks MPI worker futures, remote-session worker-death notifications,
and the error queue using the shared ``_check_mpi_futures()``,
``_check_remote_worker_death()`` and ``_drain_error_queue()`` helpers.

Propagation to pending requests is event-driven via
``_handle_worker_death`` (the MPI future done-callback) where futures
Expand All @@ -371,7 +355,7 @@ def _error_monitor_loop(self) -> None:
"""
while not self.doing_shutdown and self._fatal_error is None:
try:
if self._check_mpi_workers():
if self._check_mpi_futures():
logger.error("Error monitor: MPI worker crash detected, "
"shutting down")
return
Expand Down Expand Up @@ -553,7 +537,7 @@ def mpi_done_callback(future: concurrent.futures.Future):

while True:
if self.worker_init_status_queue.poll(1):
status = self.worker_init_status_queue.get()
ready_signal, error_trace = self.worker_init_status_queue.get()
# Send ACK to the worker
self.worker_init_status_queue.put("ACK")
logger.info("get signal from executor worker")
Expand All @@ -563,7 +547,6 @@ def mpi_done_callback(future: concurrent.futures.Future):
raise RuntimeError("Executor worker died during initialization")
self._handle_background_error()

ready_signal, error_trace = status[:2]
if ready_signal != GenerationExecutorProxy.READY_SIGNAL:
logger.error(f"Executor worker initialization error: {error_trace}")
# Only abort a session this proxy created; an externally owned
Expand All @@ -573,21 +556,6 @@ def mpi_done_callback(future: concurrent.futures.Future):
raise RuntimeError(
"Executor worker returned error") from ready_signal

self._register_worker_processes(status)

def _register_worker_processes(self, status: tuple) -> None:
"""Register identities returned by locally spawned MPI workers.

Test session reuse replaces this module's ``MpiPoolSession`` class
reference with a factory, so identify pool-backed sessions by excluding
the external communication session types.
"""
if not isinstance(
self.mpi_session,
(MpiCommSession, RemoteMpiCommSessionClient)) and len(status) == 3:
worker_process_identities: List[WorkerProcessIdentity] = status[2]
self._worker_process_monitor.register(worker_process_identities)

def _abort_all_requests(self):
# The results can be finished during this loop, so self._results may be changed.
for result in list(self._results.values()):
Expand All @@ -603,8 +571,6 @@ def pre_shutdown(self):
else:
self.doing_shutdown = True

self._worker_process_monitor.close()

# Wake the error monitor thread immediately so it exits cleanly
if hasattr(self, '_shutdown_event'):
self._shutdown_event.set()
Expand Down
5 changes: 1 addition & 4 deletions tensorrt_llm/executor/worker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,6 @@
from .rpc_worker_mixin import RpcWorkerMixin
from .utils import (ErrorResponse, IntraProcessQueue, RequestError,
WorkerCommIpcAddrs)
from .worker_process_monitor import capture_worker_process_identity

__all__ = [
"GenerationExecutorWorker",
Expand Down Expand Up @@ -311,8 +310,6 @@ def notify_proxy_threads_to_quit():
# error to the error_queue in the main thread.

mpi_comm().barrier()
worker_process_identities = mpi_comm().allgather(
capture_worker_process_identity(mpi_rank()))
logger_debug(f"Worker {mpi_rank()} ready to setup backend...\n", "green")

try:
Expand Down Expand Up @@ -353,7 +350,7 @@ def notify_proxy_threads_to_quit():
worker.set_result_queue(result_queue)

# Send ready signal with confirmation
ready_msg = (ready_signal, None, worker_process_identities)
ready_msg = (ready_signal, None)
if not worker_init_status_queue.notify_with_retry(ready_msg):
logger.warning(
"Failed to deliver ready signal to proxy, continuing anyway"
Expand Down
200 changes: 0 additions & 200 deletions tensorrt_llm/executor/worker_process_monitor.py
View file
Open in desktop

This file was deleted.

2 changes: 2 additions & 0 deletions tests/integration/test_lists/waives.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -435,7 +435,9 @@ test_e2e.py::test_multi_nodes_eval[MiniMax-M2-tp16-mmlu] SKIP (https://nvbugs/63
test_e2e.py::test_multi_nodes_eval[MiniMax-M3-tp16-mmlu] SKIP (https://nvbugs/6373561)
test_e2e.py::test_ptp_quickstart_advanced_deepseek_r1_w4afp8_8gpus[DeepSeek-R1-W4AFP8-DeepSeek-R1/DeepSeek-R1-W4AFP8] SKIP (https://nvbugs/5836830)
unittest/_torch/misc/test_share_tensor.py::TestShareTensor::test_share_tensor_different_dtypes SKIP (https://nvbugs/6418021)
unittest/_torch/modeling -k "modeling_out_of_tree" SKIP (https://nvbugs/6426847)
unittest/_torch/modeling -k "modeling_qwen" SKIP (https://nvbugs/6433376)
unittest/_torch/modeling/test_modeling_out_of_tree.py::TestOutOfTree::test_llm_api[True] SKIP (https://nvbugs/6426847)
unittest/_torch/modeling/test_modeling_qwen3_5_vl.py::test_qwen35_dense_vl_resolves_mamba_ssm_cache_dtype SKIP (https://nvbugs/6433376)
unittest/_torch/modules/moe/test_moe_backend.py::test_moe_backend[act=Relu2-e60_k4_h2048_i1408-seq=8-dtype=torch.bfloat16-backend=TRTLLM-quant=NVFP4-routing=Renormalize] SKIP (https://nvbugs/5989912)
unittest/_torch/modules/moe/test_moe_module.py::test_configurable_moe_single_gpu -k "TRTLLM" SKIP (https://nvbugs/6464169)
Expand Down
Loading
Loading