[https://nvbugs/6487038][fix] Stop single-rank disagg errors from crashing all gen ranks

tensorrt_llm/_torch/pyexecutor/py_executor.py

@@ -5870,7 +5870,18 @@ def _pad_attention_dp_dummy_request(self):
         if not self.enable_attention_dp:
             return
 
-        assert self.expected_num_active_requests >= len(self.active_requests)
+        # Disagg KV-transfer-error requests can transiently linger in
+        # active_requests (len > the ADP-consensus expected count) before cleanup
+        # drains them. It self-corrects and the padding below keys on the
+        # schedulable count, so warn rather than assert -- a hard assert here
+        # would crash the gen loop on every ADP rank.
+        if self.expected_num_active_requests < len(self.active_requests):
+            logger.warning_once(
+                "expected_num_active_requests "
+                f"({self.expected_num_active_requests}) < active_requests "
+                f"({len(self.active_requests)}); transient disagg-error "
+                "overshoot, continuing",
+                key="adp_dummy_active_overshoot")
         num_active_request = self._count_schedulable_active_requests()
 
         if self._should_skip_dummy_for_benchmark_disagg(num_active_request):

tensorrt_llm/executor/worker.py

@@ -80,8 +80,19 @@ def __init__(
             name="await_response_thread")
 
     def start_thread(self, thread: ManagedThread):
-        if self.engine.can_enqueue_requests() and not thread.is_alive():
-            thread.start()
+        if not self.engine.can_enqueue_requests():
+            return
+        if thread.is_alive():
+            return
+        if thread.ident is not None:
+            # Already exited: either stop() at shutdown (nothing to surface) or
+            # an engine event-loop crash, where restarting masks it into a peer
+            # MPI-collective hang. Wrap, since start() runs on every submit().
+            err = getattr(self.engine, "_event_loop_error", None)
+            if err is not None:
+                raise RequestError(str(err)) from err
+            return
+        thread.start()
 
     def await_response_task(self) -> bool:
         return self._await_response_helper()

tests/scripts/perf-sanity/disaggregated/gb300_deepseek-r1-fp4_128k8k_con256_ctx1_pp4_gen1_dep8_eplb0_mtp1_ccb-NIXL.yaml

@@ -64,7 +64,10 @@ worker_config:
     cache_transceiver_config:
       max_tokens_in_buffer: 131104
       backend: NIXL
-      transceiver_runtime: CPP
+      # Deliberately no kv_cache_bounce_size_mb / TRTLLM_KV_TRANSFER_NUM_THREADS
+      # unlike the 8k1k cases: measured on this shape they cost ~8% end-to-end
+      # (13691 -> 12604 tok/s), and this case already runs near its CI timeout.
+      transceiver_runtime: PYTHON
       kv_transfer_timeout_ms: 600000
     disable_overlap_scheduler: true
     speculative_config: &id001
@@ -92,7 +95,7 @@ worker_config:
     cache_transceiver_config:
       max_tokens_in_buffer: 131104
       backend: NIXL
-      transceiver_runtime: CPP
+      transceiver_runtime: PYTHON
       kv_transfer_timeout_ms: 600000
     disable_overlap_scheduler: true
     speculative_config: *id001

tests/scripts/perf-sanity/disaggregated/gb300_deepseek-r1-fp4_8k1k_con4096_ctx1_dep4_gen1_dep16_eplb0_mtp1_ccb-NIXL.yaml

@@ -35,7 +35,7 @@ environment:
   trtllm_repo: ''
   build_wheel: false
   work_dir: <full_path_to_work_dir>
-  worker_env_var: TLLM_LOG_LEVEL=INFO TRTLLM_SERVER_DISABLE_GC=1 TRTLLM_WORKER_DISABLE_GC=1 TRTLLM_ENABLE_PDL=1 ENROOT_ALLOW_DEV=yes TLLM_SPEC_DECODE_FORCE_NUM_ACCEPTED_TOKENS=1
+  worker_env_var: TLLM_LOG_LEVEL=INFO TRTLLM_SERVER_DISABLE_GC=1 TRTLLM_WORKER_DISABLE_GC=1 TRTLLM_ENABLE_PDL=1 ENROOT_ALLOW_DEV=yes TLLM_SPEC_DECODE_FORCE_NUM_ACCEPTED_TOKENS=1 TRTLLM_KV_TRANSFER_NUM_THREADS=4
   server_env_var: TRTLLM_SERVER_DISABLE_GC=1
 profiling:
   nsys_on: false
@@ -63,7 +63,9 @@ worker_config:
     cache_transceiver_config:
       max_tokens_in_buffer: 16384
       backend: NIXL
-      transceiver_runtime: CPP
+      transceiver_runtime: PYTHON
+      kv_cache_bounce_size_mb: 2048
+      kv_transfer_timeout_ms: 600000
     disable_overlap_scheduler: true
     speculative_config: &id001
       decoding_type: MTP
@@ -91,6 +93,8 @@ worker_config:
     cache_transceiver_config:
       max_tokens_in_buffer: 16384
       backend: NIXL
-      transceiver_runtime: CPP
+      transceiver_runtime: PYTHON
+      kv_cache_bounce_size_mb: 2048
+      kv_transfer_timeout_ms: 600000
     disable_overlap_scheduler: true
     speculative_config: *id001

tests/scripts/perf-sanity/disaggregated/gb300_kimi-k25-thinking-fp4_8k1k_con4096_ctx1_dep4_gen1_dep16_eplb0_mtp0_ccb-NIXL.yaml

@@ -35,7 +35,7 @@ environment:
   trtllm_repo: ''
   build_wheel: false
   work_dir: <full_path_to_work_dir>
-  worker_env_var: TLLM_LOG_LEVEL=INFO TRTLLM_SERVER_DISABLE_GC=1 TRTLLM_WORKER_DISABLE_GC=1 TRTLLM_ENABLE_PDL=1 ENROOT_ALLOW_DEV=yes
+  worker_env_var: TLLM_LOG_LEVEL=INFO TRTLLM_SERVER_DISABLE_GC=1 TRTLLM_WORKER_DISABLE_GC=1 TRTLLM_ENABLE_PDL=1 ENROOT_ALLOW_DEV=yes TRTLLM_KV_TRANSFER_NUM_THREADS=4
   server_env_var: TRTLLM_SERVER_DISABLE_GC=1
 profiling:
   nsys_on: false
@@ -64,6 +64,8 @@ worker_config:
       max_tokens_in_buffer: 16384
       backend: NIXL
       transceiver_runtime: PYTHON
+      kv_cache_bounce_size_mb: 2048
+      kv_transfer_timeout_ms: 600000
     disable_overlap_scheduler: true
     trust_remote_code: true
     num_postprocess_workers: 4
@@ -90,5 +92,7 @@ worker_config:
       max_tokens_in_buffer: 16384
       backend: NIXL
       transceiver_runtime: PYTHON
+      kv_cache_bounce_size_mb: 2048
+      kv_transfer_timeout_ms: 600000
     disable_overlap_scheduler: true
     trust_remote_code: true

tests/unittest/_torch/executor/test_py_executor.py

@@ -1473,6 +1473,44 @@ def test_pad_dummy_added_when_only_to_complete_requests_disagg():
     assert len(stub.active_requests) == 2
 
 
+def test_pad_dummy_tolerates_active_request_overshoot():
+    # A transient overshoot (len(active_requests) > expected_num_active_requests,
+    # when disagg transfer-error requests linger a tick before cleanup) used to
+    # trip a hard assert that crashed the gen loop on every ADP rank. It must now
+    # warn and continue instead of raising.
+    stub = _StubADPExecutor()
+    stub.active_requests = [
+        _make_adp_request(_STATE_GENERATION_IN_PROGRESS),
+        _make_adp_request(_STATE_GENERATION_IN_PROGRESS),
+    ]
+    stub.expected_num_active_requests = 1  # < len(active_requests) == 2
+
+    # Must not raise AssertionError (the pre-fix behavior on overshoot).
+    _run_pad(stub)
+
+    # Both requests are schedulable, so no dummy is added; pin it so the test
+    # cannot pass on an early return or a stray pad.
+    assert stub.add_dummy_calls == []
+    assert len(stub.active_requests) == 2
+
+
+def test_pad_dummy_added_when_overshoot_has_no_schedulable_requests():
+    # The branch that matters: overshoot AND nothing schedulable (all at
+    # GENERATION_TO_COMPLETE) must still pad, or can_queue goes False
+    # fleet-wide.
+    stub = _StubADPExecutor()
+    stub.active_requests = [
+        _make_adp_request(_STATE_GENERATION_TO_COMPLETE, request_id=1),
+        _make_adp_request(_STATE_GENERATION_TO_COMPLETE, request_id=2),
+    ]
+    stub.expected_num_active_requests = 1  # < len(active_requests) == 2
+
+    _run_pad(stub)
+
+    assert len(stub.add_dummy_calls) == 1
+    assert len(stub.active_requests) == 3
+
+
 def test_pad_dummy_added_when_only_wait_scheduler_requests_disagg():
     # Gen-first mode on the context server: DISAGG_CONTEXT_WAIT_SCHEDULER
     # sits BELOW the scheduler's window [CONTEXT_INIT, GENERATION_TO_COMPLETE)

tests/unittest/executor/test_event_loop_error_broadcast.py

@@ -16,7 +16,8 @@
 import pytest
 
 from tensorrt_llm.executor.base_worker import AwaitResponseHelper
-from tensorrt_llm.executor.utils import ErrorResponse
+from tensorrt_llm.executor.utils import ErrorResponse, RequestError
+from tensorrt_llm.executor.worker import GenerationExecutorWorker
 
 pytestmark = pytest.mark.cpu_only
 
@@ -87,6 +88,96 @@ def _make_helper(engine, num_pending: int = 1):
     return helper
 
 
+class _ThreadStub:
+    """ManagedThread stand-in: ident set + not alive means "already exited"."""
+
+    def __init__(self, *, alive=False, ident=1):
+        self._alive = alive
+        self.ident = ident
+        self.starts = 0
+
+    def is_alive(self):
+        return self._alive
+
+    def start(self):
+        self.starts += 1
+
+
+class _StartThreadWorkerStub:
+    """Minimal stand-in for the worker start_thread() binds to.
+
+    It only reads self.engine, so a plain object avoids an uninitialized
+    GenerationExecutorWorker whose destructor would raise at collection time.
+    """
+
+    def __init__(self, event_loop_error=None, can_enqueue=True):
+        self.engine = _EngineStub(event_loop_error=event_loop_error)
+        self.engine.can_enqueue_requests = lambda: can_enqueue
+
+
+class TestStartThreadAfterExit:
+    """start_thread must surface an engine crash instead of restarting."""
+
+    def test_surfaces_engine_error_as_request_error(self):
+        original = RuntimeError("kv cache OOM")
+        worker = _StartThreadWorkerStub(event_loop_error=original)
+        thread = _ThreadStub()
+
+        with pytest.raises(RequestError) as excinfo:
+            GenerationExecutorWorker.start_thread(worker, thread)
+
+        # Chained, not re-raised: the caller still sees the real cause.
+        assert excinfo.value.__cause__ is original
+        assert "kv cache OOM" in str(excinfo.value)
+        assert thread.starts == 0
+
+    def test_repeated_calls_do_not_accumulate_traceback(self):
+        # start() runs on every submit(); re-raising the same object would grow
+        # its __traceback__ one frame per call.
+        original = RuntimeError("kv cache OOM")
+        worker = _StartThreadWorkerStub(event_loop_error=original)
+        thread = _ThreadStub()
+
+        raised = []
+        for _ in range(3):
+            with pytest.raises(RequestError) as excinfo:
+                GenerationExecutorWorker.start_thread(worker, thread)
+            raised.append(excinfo.value)
+
+        assert len({id(e) for e in raised}) == 3
+        assert all(e.__cause__ is original for e in raised)
+
+    def test_post_shutdown_exit_returns_quietly(self):
+        # The other exit path: shutdown() called ManagedThread.stop(), so
+        # stop_event ended run() and there is no error to report.
+        worker = _StartThreadWorkerStub(event_loop_error=None)
+        thread = _ThreadStub()
+
+        GenerationExecutorWorker.start_thread(worker, thread)
+
+        assert thread.starts == 0
+
+    def test_fresh_thread_is_started(self):
+        worker = _StartThreadWorkerStub(event_loop_error=None)
+        thread = _ThreadStub(ident=None)
+
+        GenerationExecutorWorker.start_thread(worker, thread)
+
+        assert thread.starts == 1
+
+    def test_does_not_start_when_enqueueing_is_disabled(self):
+        # The can_enqueue_requests() guard returns before the error check, so a
+        # stashed error must not surface either.
+        worker = _StartThreadWorkerStub(
+            event_loop_error=RuntimeError("should not surface"), can_enqueue=False
+        )
+        thread = _ThreadStub(ident=None)
+
+        GenerationExecutorWorker.start_thread(worker, thread)
+
+        assert thread.starts == 0
+
+
 class TestAwaitResponseHelperEventLoopError:
     def test_normal_path_returns_true(self):
         """No engine error and no responses: ManagedThread should keep going."""