diff --git a/tensorrt_llm/_torch/pyexecutor/py_executor.py b/tensorrt_llm/_torch/pyexecutor/py_executor.py index 2fa5a5d0607b..f8577ae03e3e 100644 --- a/tensorrt_llm/_torch/pyexecutor/py_executor.py +++ b/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): diff --git a/tensorrt_llm/executor/worker.py b/tensorrt_llm/executor/worker.py index f173aabc224a..b88fc4f40ef7 100644 --- a/tensorrt_llm/executor/worker.py +++ b/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() diff --git a/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-r1-fp4_128k8k_con256_ctx1_pp4_gen1_dep8_eplb0_mtp1_ccb-NIXL.yaml b/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-r1-fp4_128k8k_con256_ctx1_pp4_gen1_dep8_eplb0_mtp1_ccb-NIXL.yaml index 8b76516d6a3f..b50bafb61462 100644 --- a/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-r1-fp4_128k8k_con256_ctx1_pp4_gen1_dep8_eplb0_mtp1_ccb-NIXL.yaml +++ b/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 diff --git a/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-r1-fp4_8k1k_con4096_ctx1_dep4_gen1_dep16_eplb0_mtp1_ccb-NIXL.yaml b/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-r1-fp4_8k1k_con4096_ctx1_dep4_gen1_dep16_eplb0_mtp1_ccb-NIXL.yaml index f5d27a87811b..4b52b8cb8d52 100644 --- a/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-r1-fp4_8k1k_con4096_ctx1_dep4_gen1_dep16_eplb0_mtp1_ccb-NIXL.yaml +++ b/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: - 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 diff --git a/tests/scripts/perf-sanity/disaggregated/gb300_kimi-k25-thinking-fp4_8k1k_con4096_ctx1_dep4_gen1_dep16_eplb0_mtp0_ccb-NIXL.yaml b/tests/scripts/perf-sanity/disaggregated/gb300_kimi-k25-thinking-fp4_8k1k_con4096_ctx1_dep4_gen1_dep16_eplb0_mtp0_ccb-NIXL.yaml index 7bbe295f8371..2b2bc8a82f48 100644 --- a/tests/scripts/perf-sanity/disaggregated/gb300_kimi-k25-thinking-fp4_8k1k_con4096_ctx1_dep4_gen1_dep16_eplb0_mtp0_ccb-NIXL.yaml +++ b/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: - 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 diff --git a/tests/unittest/_torch/executor/test_py_executor.py b/tests/unittest/_torch/executor/test_py_executor.py index c9c05b39137f..8515f72a07ce 100644 --- a/tests/unittest/_torch/executor/test_py_executor.py +++ b/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) diff --git a/tests/unittest/executor/test_event_loop_error_broadcast.py b/tests/unittest/executor/test_event_loop_error_broadcast.py index f5b72d22e3f6..17d916db1915 100644 --- a/tests/unittest/executor/test_event_loop_error_broadcast.py +++ b/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."""