From a31002219ff1775427ce5c614e3cf0b255344370 Mon Sep 17 00:00:00 2001 From: Bo Deng Date: Fri, 24 Jul 2026 09:15:03 +0000 Subject: [PATCH 1/5] [TRTLLM-14010][feat] report KV cache transfer state on executor hangs Signed-off-by: Bo Deng --- .../batch_manager/cacheTransceiver.h | 39 +++++++++ .../batch_manager/cacheTransceiver.cpp | 87 ++++++++++++++++++- .../batch_manager/cacheTransceiver.cpp | 4 +- .../_torch/disaggregation/transceiver.py | 65 +++++++++++++- .../_torch/pyexecutor/hang_detector.py | 12 +++ .../_torch/pyexecutor/kv_cache_transceiver.py | 7 ++ tensorrt_llm/_torch/pyexecutor/py_executor.py | 3 + 7 files changed, 214 insertions(+), 3 deletions(-) diff --git a/cpp/include/tensorrt_llm/batch_manager/cacheTransceiver.h b/cpp/include/tensorrt_llm/batch_manager/cacheTransceiver.h index 0c7724c657fc..1b0fc4d23200 100644 --- a/cpp/include/tensorrt_llm/batch_manager/cacheTransceiver.h +++ b/cpp/include/tensorrt_llm/batch_manager/cacheTransceiver.h @@ -27,6 +27,7 @@ #include "tensorrt_llm/executor/dataTransceiverState.h" #include "tensorrt_llm/runtime/utils/mpiUtils.h" #include "tensorrt_llm/runtime/utils/pgUtils.h" +#include #include #include #include @@ -309,8 +310,39 @@ class CacheTransceiver : public BaseCacheTransceiver [[nodiscard]] std::vector getSerializedDataTransceiverState() const override; [[nodiscard]] bool hasPoisonedTransferBuffer() const override; + /// Return a human-readable dump of transceiver state for debugging hangs. + std::string getStatusDump() const; private: + struct StatusSnapshot + { + size_t senderAsyncActive{0}; + size_t requesterAsyncActive{0}; + size_t timedOutSenders{0}; + size_t timedOutRequesters{0}; + size_t cancelingSenders{0}; + size_t cancelingRequesters{0}; + size_t completedSenders{0}; + size_t completedRequesters{0}; + size_t failedSenders{0}; + size_t failedRequesters{0}; + size_t sendersAwaitingConsensus{0}; + size_t requestersAwaitingConsensus{0}; + }; + + class SyncRequesterStatusGuard + { + public: + explicit SyncRequesterStatusGuard(CacheTransceiver& transceiver); + ~SyncRequesterStatusGuard() noexcept; + + SyncRequesterStatusGuard(SyncRequesterStatusGuard const&) = delete; + SyncRequesterStatusGuard& operator=(SyncRequesterStatusGuard const&) = delete; + + private: + CacheTransceiver& mTransceiver; + }; + void initializeCommState(); void setContextState(LlmRequest* llmRequest); @@ -318,6 +350,7 @@ class CacheTransceiver : public BaseCacheTransceiver // Append one row per completed request to the gen-side transfer summary CSV. Opens the file // lazily on first use; expects timing to already be synced across ranks by the caller. void writeGenTransferSummary(std::vector const& completedRequests); + void publishStatusSnapshot() noexcept; std::unique_ptr mCacheSender; std::unique_ptr mCacheReceiver; @@ -338,6 +371,12 @@ class CacheTransceiver : public BaseCacheTransceiver std::unordered_set mCompletedRequesterRequestIds; std::unordered_set mFailedRequesterRequestIds; std::unordered_map> mRequesterRequestsAwaitingConsensus; + std::atomic_size_t mSyncRequesterActive{0}; + // Live transfer containers are owned by the executor worker thread. Synchronous receive threads update only the + // atomic count above. The executor publishes snapshots after state transitions, while the hang-detector thread only + // copies the snapshot under this short lock. + mutable std::mutex mStatusSnapshotMutex; + StatusSnapshot mStatusSnapshot; mpi::MpiComm const* mMpiWorldComm{nullptr}; std::shared_ptr mGroupComm; diff --git a/cpp/tensorrt_llm/batch_manager/cacheTransceiver.cpp b/cpp/tensorrt_llm/batch_manager/cacheTransceiver.cpp index c4acce47593f..d3654d00c8bb 100644 --- a/cpp/tensorrt_llm/batch_manager/cacheTransceiver.cpp +++ b/cpp/tensorrt_llm/batch_manager/cacheTransceiver.cpp @@ -64,6 +64,7 @@ #include #include #include +#include #include #include #include @@ -102,6 +103,20 @@ using RequestIdType = LlmRequest::RequestIdType; constexpr int kTransferFuturePollIntervalMs = 10; +char const* cacheTransceiverBackendName(executor::CacheTransceiverConfig::BackendType backendType) +{ + using BackendType = executor::CacheTransceiverConfig::BackendType; + switch (backendType) + { + case BackendType::DEFAULT: return "DEFAULT"; + case BackendType::MPI: return "MPI"; + case BackendType::UCX: return "UCX"; + case BackendType::NIXL: return "NIXL"; + case BackendType::MOONCAKE: return "MOONCAKE"; + } + return "UNKNOWN"; +} + // Finite status checks are scheduler polls, not terminal deadlines. Pure polls // use short slices; calls that ask for at least one completion keep bounded // backpressure by waiting up to the configured future timeout. @@ -729,6 +744,65 @@ CacheTransceiver::~CacheTransceiver() } } +std::string CacheTransceiver::getStatusDump() const +{ + auto const backendType = mCacheTransceiverConfig->getBackendType().value(); + StatusSnapshot snapshot; + { + std::lock_guard lock(mStatusSnapshotMutex); + snapshot = mStatusSnapshot; + } + auto const requesterSyncActive = mSyncRequesterActive.load(std::memory_order_relaxed); + std::ostringstream oss; + oss << "KV cache transceiver | backend=" << cacheTransceiverBackendName(backendType) + << " | TX(async_active=" << snapshot.senderAsyncActive << ", timed_out=" << snapshot.timedOutSenders + << ", cancel_requested=" << snapshot.cancelingSenders << ", local_completed=" << snapshot.completedSenders + << ", local_failed=" << snapshot.failedSenders << ", awaiting_consensus=" << snapshot.sendersAwaitingConsensus + << ") | RX(async_active=" << snapshot.requesterAsyncActive << ", sync_active=" << requesterSyncActive + << ", timed_out=" << snapshot.timedOutRequesters << ", cancel_requested=" << snapshot.cancelingRequesters + << ", local_completed=" << snapshot.completedRequesters << ", local_failed=" << snapshot.failedRequesters + << ", awaiting_consensus=" << snapshot.requestersAwaitingConsensus + << ") | poisoned=" << (hasPoisonedTransferBuffer() ? "yes" : "no"); + return oss.str(); +} + +void CacheTransceiver::publishStatusSnapshot() noexcept +{ + StatusSnapshot snapshot; + snapshot.senderAsyncActive = mSenderFutures.size(); + snapshot.requesterAsyncActive = mRequesterFutures.size(); + snapshot.timedOutSenders = mTimedOutSenderIds.size(); + snapshot.timedOutRequesters = mTimedOutRequesterIds.size(); + snapshot.cancelingSenders = mCancelRequestedSenderIds.size(); + snapshot.cancelingRequesters = mCancelRequestedRequesterIds.size(); + snapshot.completedSenders = mCompletedSenderRequestIds.size(); + snapshot.completedRequesters = mCompletedRequesterRequestIds.size(); + snapshot.failedSenders = mFailedSenderRequestIds.size(); + snapshot.failedRequesters = mFailedRequesterRequestIds.size(); + snapshot.sendersAwaitingConsensus = mSenderRequestsAwaitingConsensus.size(); + snapshot.requestersAwaitingConsensus = mRequesterRequestsAwaitingConsensus.size(); + try + { + std::lock_guard lock(mStatusSnapshotMutex); + mStatusSnapshot = snapshot; + } + catch (std::system_error const&) + { + // Status publication is best-effort and must never fail a transfer path. + } +} + +CacheTransceiver::SyncRequesterStatusGuard::SyncRequesterStatusGuard(CacheTransceiver& transceiver) + : mTransceiver{transceiver} +{ + mTransceiver.mSyncRequesterActive.fetch_add(1, std::memory_order_relaxed); +} + +CacheTransceiver::SyncRequesterStatusGuard::~SyncRequesterStatusGuard() noexcept +{ + mTransceiver.mSyncRequesterActive.fetch_sub(1, std::memory_order_relaxed); +} + void CacheTransceiver::initializeCommState() { mCommState = std::addressof(mCacheSender->getCommState()); @@ -780,6 +854,7 @@ void CacheTransceiver::respondAndSendAsync(std::shared_ptr llmReques setContextState(llmRequest.get()); auto future = mCacheSender->sendAsync(llmRequest); mSenderFutures.emplace_back(std::move(llmRequest), std::move(future)); + publishStatusSnapshot(); } void CacheTransceiver::respondAndSendLayerWise( @@ -797,6 +872,7 @@ void CacheTransceiver::respondAndSendLayerWise( auto future = mCacheSender->sendAsync(llmRequest); mSenderFutures.emplace_back(llmRequest, std::move(future)); } + publishStatusSnapshot(); } void CacheTransceiver::requestAndReceiveSync(std::shared_ptr llmRequest) @@ -806,6 +882,7 @@ void CacheTransceiver::requestAndReceiveSync(std::shared_ptr llmRequ auto const contextRequestId = llmRequest->getContextPhaseParams().value().getReqId(); TLLM_LOG_DEBUG("Synchronous KV cache receive request %zu, context request %zu waiting for native completion.", requestId, contextRequestId); + SyncRequesterStatusGuard statusGuard{*this}; try { auto future = mCacheReceiver->receiveAsync(llmRequest); @@ -854,6 +931,7 @@ void CacheTransceiver::requestAndReceiveAsync(std::shared_ptr llmReq auto* requestPtr = llmRequest.get(); mRequesterFutures.emplace_back(std::move(llmRequest), std::move(future)); requestPtr->setState(LlmRequestState::kDISAGG_GENERATION_TRANS_IN_PROGRESS); + publishStatusSnapshot(); } std::vector gatherRequestIds( @@ -1024,7 +1102,7 @@ RequestStatuses CacheTransceiver::checkContextTransferStatus( contextCompleteRequestIds.push_back(request->mRequestId); } } - + publishStatusSnapshot(); std::unordered_map frequencyMap; if ((syncComm) && syncComm->getSize() > 1) { @@ -1174,6 +1252,8 @@ RequestStatuses CacheTransceiver::checkContextTransferStatus( } } + // Publish after local polling and before consensus, which may be the point at which a rank hangs. + publishStatusSnapshot(); RequestStatuses requestsStatus{}; TransferConsensusOutcome consensusOutcome; if (mContextTransferCoordinator) @@ -1271,6 +1351,7 @@ RequestStatuses CacheTransceiver::checkContextTransferStatus( requestId, mCompletedSenderRequestIds, mFailedSenderRequestIds, mSenderRequestsAwaitingConsensus); } + publishStatusSnapshot(); return requestsStatus; } @@ -1313,6 +1394,7 @@ void CacheTransceiver::checkGenTransferStatus(std::optional const& atLeastR collectReadyRequestIds(); } } + publishStatusSnapshot(); std::unordered_map frequencyMap; std::vector toBlockRequestIds; @@ -1466,6 +1548,8 @@ void CacheTransceiver::checkGenTransferStatus(std::optional const& atLeastR } } + // Publish after local polling and before collectives, which may be the point at which a rank hangs. + publishStatusSnapshot(); auto const consensusOutcome = reduceTransferStates(syncComm, mCompletedRequesterRequestIds, mFailedRequesterRequestIds, inflightCancelEnabled ? mTimedOutRequesterIds : std::unordered_set{}); @@ -1524,6 +1608,7 @@ void CacheTransceiver::checkGenTransferStatus(std::optional const& atLeastR eraseLocalTransferOutcome( requestId, mCompletedRequesterRequestIds, mFailedRequesterRequestIds, mRequesterRequestsAwaitingConsensus); } + publishStatusSnapshot(); // Batch-sync timing across ranks in one allgather (instead of per-request), then write // the gen-side transfer summary CSV. diff --git a/cpp/tensorrt_llm/nanobind/batch_manager/cacheTransceiver.cpp b/cpp/tensorrt_llm/nanobind/batch_manager/cacheTransceiver.cpp index 2268e6f8ed03..cc00ea0259c5 100644 --- a/cpp/tensorrt_llm/nanobind/batch_manager/cacheTransceiver.cpp +++ b/cpp/tensorrt_llm/nanobind/batch_manager/cacheTransceiver.cpp @@ -27,6 +27,7 @@ #include #include #include +#include #include #include #include @@ -133,7 +134,8 @@ void tb::CacheTransceiverBindings::initBindings(nb::module_& m) nb::arg("cache_manager"), nb::arg("num_kv_heads_per_layer"), nb::arg("size_per_head"), nb::arg("tokens_per_block"), nb::arg("world_config"), nb::arg("attention_layer_num_per_pp"), nb::arg("dtype"), nb::arg("attention_type"), nb::arg("cache_transceiver_config") = std::nullopt, - nb::arg("rnn_layer_num_per_pp") = std::vector{}); + nb::arg("rnn_layer_num_per_pp") = std::vector{}) + .def("get_status_dump", &tb::CacheTransceiver::getStatusDump); nb::class_(m, "CacheTransceiverComm") .def( diff --git a/tensorrt_llm/_torch/disaggregation/transceiver.py b/tensorrt_llm/_torch/disaggregation/transceiver.py index 2c257a39f731..5ca73e5914f1 100644 --- a/tensorrt_llm/_torch/disaggregation/transceiver.py +++ b/tensorrt_llm/_torch/disaggregation/transceiver.py @@ -1,7 +1,21 @@ +# 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. import os import time import uuid -from collections import defaultdict +from collections import Counter, defaultdict from itertools import chain from typing import Any, Callable, Dict, List, Optional, cast @@ -150,6 +164,55 @@ def _exchange_rank_info(self): logger.info(f"layer_num_per_pp: {layer_num_per_pp}") logger.info(f"self._context_info_endpoint: {self._context_info_endpoint}") + def get_status_dump(self) -> str: + """Return a one-line summary of transceiver state for debugging hangs.""" + + def summarize( + sessions: Dict[int, Any], + include_receiver_ready: bool, + ) -> str: + session_items = list(sessions.items()) + status_counts = Counter() + receiver_ready = 0 + for _, session in session_items: + try: + status = session.status + except Exception: + status_counts["unknown"] += 1 + else: + if isinstance(status, SessionStatus): + status_counts[status] += 1 + else: + status_counts["unknown"] += 1 + + if include_receiver_ready: + try: + receiver_ready += int(bool(session.receiver_ready)) + except Exception: + pass + + fields = [ + f"sessions={len(session_items)}", + f"init={status_counts[SessionStatus.INIT]}", + f"ready_to_transfer={status_counts[SessionStatus.READY]}", + f"transferring={status_counts[SessionStatus.TRANSFERRING]}", + f"kv_transferred={status_counts[SessionStatus.KV_TRANSFERRED]}", + f"fully_transferred={status_counts[SessionStatus.FULLY_TRANSFERRED]}", + f"error={status_counts[SessionStatus.ERROR]}", + f"cancelled={status_counts[SessionStatus.CANCELLED]}", + f"unknown={status_counts['unknown']}", + ] + if include_receiver_ready: + fields.append(f"peer_ready={receiver_ready}/{len(session_items)}") + return ", ".join(fields) + + tx_status = summarize(self._send_sessions, include_receiver_ready=True) + rx_status = summarize(self._recv_sessions, include_receiver_ready=False) + return ( + f"KV cache transceiver | backend=NIXL | TX({tx_status}) | RX({rx_status}) | " + f"waiting_for_peer_info={len(self._wait_reqs)}" + ) + def shutdown(self): if getattr(self, "_shutdown", False): return diff --git a/tensorrt_llm/_torch/pyexecutor/hang_detector.py b/tensorrt_llm/_torch/pyexecutor/hang_detector.py index 2ae692ed5902..c0318249023a 100644 --- a/tensorrt_llm/_torch/pyexecutor/hang_detector.py +++ b/tensorrt_llm/_torch/pyexecutor/hang_detector.py @@ -100,6 +100,7 @@ def __init__( self.lock = threading.Lock() self.active = False self._detected = False + self._status_providers: list[Callable[[], str]] = [] def start(self): """Enable hang detection.""" @@ -113,11 +114,22 @@ def run_loop(): self.loop_thread = threading.Thread(target=run_loop, daemon=True, name="hang_detector_loop") self.loop_thread.start() + def register_status_provider(self, provider: Callable[[], str]): + """Register a callable that returns a status string to dump on hang detection.""" + self._status_providers.append(provider) + async def _detect_hang(self): await asyncio.sleep(self.timeout) with self.lock: self._detected = True logger.error(f"Hang detected after {self.timeout} seconds.") + for provider in self._status_providers: + try: + status = provider() + if status: + logger.error(status) + except Exception: + pass print_all_stacks() self.on_detected() diff --git a/tensorrt_llm/_torch/pyexecutor/kv_cache_transceiver.py b/tensorrt_llm/_torch/pyexecutor/kv_cache_transceiver.py index a686adc97528..c998b5d5cc2a 100644 --- a/tensorrt_llm/_torch/pyexecutor/kv_cache_transceiver.py +++ b/tensorrt_llm/_torch/pyexecutor/kv_cache_transceiver.py @@ -241,6 +241,10 @@ def get_data_transceiver_state(self) -> bytes: """Get the serialized DataTransceiverState (CacheState + CommState).""" return b"" + def get_status_dump(self) -> str: + """Return a human-readable dump of transceiver state for debugging hangs.""" + return "" + def shutdown(self): """Shut down the transceiver and release registered resources.""" @@ -335,6 +339,9 @@ def has_poisoned_transfer_buffer(self) -> bool: return False return self.impl.has_poisoned_transfer_buffer() + def get_status_dump(self) -> str: + return self.impl.get_status_dump() + def prepare_context_requests(self, requests: List[LlmRequest]): # not implemented, an empty placeholder to allow being invoked unconditionally ... diff --git a/tensorrt_llm/_torch/pyexecutor/py_executor.py b/tensorrt_llm/_torch/pyexecutor/py_executor.py index feb02319a3e3..748a47357389 100644 --- a/tensorrt_llm/_torch/pyexecutor/py_executor.py +++ b/tensorrt_llm/_torch/pyexecutor/py_executor.py @@ -918,6 +918,9 @@ def on_detected(): self.gather_all_responses = False self.kv_cache_transceiver = kv_cache_transceiver + if kv_cache_transceiver is not None: + self.hang_detector.register_status_provider( + kv_cache_transceiver.get_status_dump) cache_transceiver_config = getattr(self.llm_args, "cache_transceiver_config", None) max_tokens_in_buffer = getattr(cache_transceiver_config, From b6feb3ffd049954f3202fdbf9a74f3c4330d7df1 Mon Sep 17 00:00:00 2001 From: Bo Deng Date: Fri, 24 Jul 2026 10:09:32 +0000 Subject: [PATCH 2/5] fix by agent Signed-off-by: Bo Deng --- .../batch_manager/cacheTransceiver.cpp | 12 +++++-- .../_torch/disaggregation/transceiver.py | 19 ++++------ .../_torch/pyexecutor/hang_detector.py | 25 ++++++++----- .../executor/test_hang_detector_kill.py | 35 +++++++++++++++++++ 4 files changed, 67 insertions(+), 24 deletions(-) diff --git a/cpp/tensorrt_llm/batch_manager/cacheTransceiver.cpp b/cpp/tensorrt_llm/batch_manager/cacheTransceiver.cpp index d3654d00c8bb..48b8f0cdcb6d 100644 --- a/cpp/tensorrt_llm/batch_manager/cacheTransceiver.cpp +++ b/cpp/tensorrt_llm/batch_manager/cacheTransceiver.cpp @@ -747,12 +747,20 @@ CacheTransceiver::~CacheTransceiver() std::string CacheTransceiver::getStatusDump() const { auto const backendType = mCacheTransceiverConfig->getBackendType().value(); + auto const requesterSyncActive = mSyncRequesterActive.load(std::memory_order_relaxed); StatusSnapshot snapshot; { - std::lock_guard lock(mStatusSnapshotMutex); + std::unique_lock lock(mStatusSnapshotMutex, std::try_to_lock); + if (!lock.owns_lock()) + { + std::ostringstream oss; + oss << "KV cache transceiver | backend=" << cacheTransceiverBackendName(backendType) + << " | snapshot=unavailable | RX(sync_active=" << requesterSyncActive + << ") | poisoned=" << (hasPoisonedTransferBuffer() ? "yes" : "no"); + return oss.str(); + } snapshot = mStatusSnapshot; } - auto const requesterSyncActive = mSyncRequesterActive.load(std::memory_order_relaxed); std::ostringstream oss; oss << "KV cache transceiver | backend=" << cacheTransceiverBackendName(backendType) << " | TX(async_active=" << snapshot.senderAsyncActive << ", timed_out=" << snapshot.timedOutSenders diff --git a/tensorrt_llm/_torch/disaggregation/transceiver.py b/tensorrt_llm/_torch/disaggregation/transceiver.py index 5ca73e5914f1..c3e4237edc68 100644 --- a/tensorrt_llm/_torch/disaggregation/transceiver.py +++ b/tensorrt_llm/_torch/disaggregation/transceiver.py @@ -171,25 +171,18 @@ def summarize( sessions: Dict[int, Any], include_receiver_ready: bool, ) -> str: - session_items = list(sessions.items()) + session_items = list(sessions.copy().items()) status_counts = Counter() receiver_ready = 0 for _, session in session_items: - try: - status = session.status - except Exception: - status_counts["unknown"] += 1 + status = session.status + if isinstance(status, SessionStatus): + status_counts[status] += 1 else: - if isinstance(status, SessionStatus): - status_counts[status] += 1 - else: - status_counts["unknown"] += 1 + status_counts["unknown"] += 1 if include_receiver_ready: - try: - receiver_ready += int(bool(session.receiver_ready)) - except Exception: - pass + receiver_ready += int(bool(session.receiver_ready)) fields = [ f"sessions={len(session_items)}", diff --git a/tensorrt_llm/_torch/pyexecutor/hang_detector.py b/tensorrt_llm/_torch/pyexecutor/hang_detector.py index c0318249023a..bb9a9a62a729 100644 --- a/tensorrt_llm/_torch/pyexecutor/hang_detector.py +++ b/tensorrt_llm/_torch/pyexecutor/hang_detector.py @@ -114,23 +114,30 @@ def run_loop(): self.loop_thread = threading.Thread(target=run_loop, daemon=True, name="hang_detector_loop") self.loop_thread.start() - def register_status_provider(self, provider: Callable[[], str]): - """Register a callable that returns a status string to dump on hang detection.""" - self._status_providers.append(provider) + def register_status_provider(self, provider: Callable[[], str]) -> None: + """Register a nonblocking callable that returns status to dump on hang detection.""" + with self.lock: + self._status_providers.append(provider) - async def _detect_hang(self): + async def _detect_hang(self) -> None: await asyncio.sleep(self.timeout) with self.lock: self._detected = True - logger.error(f"Hang detected after {self.timeout} seconds.") - for provider in self._status_providers: + status_providers = tuple(self._status_providers) + + _best_effort_log_error(f"Hang detected after {self.timeout} seconds.") + try: + for provider in status_providers: try: status = provider() if status: - logger.error(status) - except Exception: - pass + _best_effort_log_error(status) + except Exception as error: # noqa: BLE001 - isolate diagnostic providers + _best_effort_log_error( + f"HangDetector: status provider failed with {type(error).__name__}: {error}" + ) print_all_stacks() + finally: self.on_detected() def detected(self): diff --git a/tests/unittest/_torch/executor/test_hang_detector_kill.py b/tests/unittest/_torch/executor/test_hang_detector_kill.py index b4f122bb8ad3..0962df441cc6 100644 --- a/tests/unittest/_torch/executor/test_hang_detector_kill.py +++ b/tests/unittest/_torch/executor/test_hang_detector_kill.py @@ -14,12 +14,14 @@ # limitations under the License. """HangDetector timer behavior and the hard-kill propagation mechanism (no GPU).""" +import asyncio import os import signal import subprocess import sys import time +from tensorrt_llm._torch.pyexecutor import hang_detector as hang_detector_module from tensorrt_llm._torch.pyexecutor.hang_detector import HangDetector @@ -64,6 +66,39 @@ def test_pause_suppresses_detection(): assert hd.detected() is False +def test_status_provider_errors_are_logged(monkeypatch): + events = [] + + async def no_sleep(_timeout): + pass + + def failing_provider(): + raise RuntimeError("provider failed") + + monkeypatch.setattr(hang_detector_module.asyncio, "sleep", no_sleep) + monkeypatch.setattr( + hang_detector_module, + "_best_effort_log_error", + lambda message: events.append(("log", message)), + ) + monkeypatch.setattr( + hang_detector_module, + "print_all_stacks", + lambda: events.append(("stacks", None)), + ) + + detector = HangDetector(timeout=1, on_detected=lambda: events.append(("detected", None))) + detector.register_status_provider(failing_provider) + detector.register_status_provider(lambda: "transceiver status") + + asyncio.run(detector._detect_hang()) + + messages = "\n".join(message for kind, message in events if kind == "log") + assert "provider failed" in messages + assert "transceiver status" in messages + assert events[-2:] == [("stacks", None), ("detected", None)] + + def test_propagate_hard_kill_self_sigkills_without_mpi(): """With MPI disabled, propagate_hard_kill self-SIGKILLs the process. From 06c5e53268583f2c4df5ab3b8fc07843b8f6f840 Mon Sep 17 00:00:00 2001 From: Bo Deng Date: Mon, 27 Jul 2026 06:16:23 +0000 Subject: [PATCH 3/5] fix Signed-off-by: Bo Deng --- .../_torch/pyexecutor/hang_detector.py | 27 +++++++++---------- 1 file changed, 13 insertions(+), 14 deletions(-) diff --git a/tensorrt_llm/_torch/pyexecutor/hang_detector.py b/tensorrt_llm/_torch/pyexecutor/hang_detector.py index bb9a9a62a729..9a99c7a9f896 100644 --- a/tensorrt_llm/_torch/pyexecutor/hang_detector.py +++ b/tensorrt_llm/_torch/pyexecutor/hang_detector.py @@ -122,23 +122,22 @@ def register_status_provider(self, provider: Callable[[], str]) -> None: async def _detect_hang(self) -> None: await asyncio.sleep(self.timeout) with self.lock: - self._detected = True status_providers = tuple(self._status_providers) _best_effort_log_error(f"Hang detected after {self.timeout} seconds.") - try: - for provider in status_providers: - try: - status = provider() - if status: - _best_effort_log_error(status) - except Exception as error: # noqa: BLE001 - isolate diagnostic providers - _best_effort_log_error( - f"HangDetector: status provider failed with {type(error).__name__}: {error}" - ) - print_all_stacks() - finally: - self.on_detected() + for provider in status_providers: + try: + status = provider() + if status: + _best_effort_log_error(status) + except Exception as error: # noqa: BLE001 - isolate diagnostic providers + _best_effort_log_error( + f"HangDetector: status provider failed with {type(error).__name__}: {error}" + ) + print_all_stacks() + with self.lock: + self._detected = True + self.on_detected() def detected(self): """Return True if hang is detected.""" From cfec146bc16158ffab86dc32032a7f38fce5639d Mon Sep 17 00:00:00 2001 From: Bo Deng Date: Mon, 27 Jul 2026 06:31:35 +0000 Subject: [PATCH 4/5] fix Signed-off-by: Bo Deng --- tensorrt_llm/_torch/pyexecutor/hang_detector.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/tensorrt_llm/_torch/pyexecutor/hang_detector.py b/tensorrt_llm/_torch/pyexecutor/hang_detector.py index 9a99c7a9f896..f6dde7c58959 100644 --- a/tensorrt_llm/_torch/pyexecutor/hang_detector.py +++ b/tensorrt_llm/_torch/pyexecutor/hang_detector.py @@ -124,6 +124,8 @@ async def _detect_hang(self) -> None: with self.lock: status_providers = tuple(self._status_providers) + # All diagnostics are best-effort: nothing may prevent on_detected() + # (hard-kill propagation) from firing. _best_effort_log_error(f"Hang detected after {self.timeout} seconds.") for provider in status_providers: try: @@ -134,7 +136,13 @@ async def _detect_hang(self) -> None: _best_effort_log_error( f"HangDetector: status provider failed with {type(error).__name__}: {error}" ) - print_all_stacks() + try: + print_all_stacks() + except Exception: # noqa: BLE001 - stack dump must not block hard kill + pass + + # Set _detected last so observers (and tests) see it only once + # diagnostics are done and on_detected is about to fire. with self.lock: self._detected = True self.on_detected() From 9a78660925bdc9812fb30f71aea5606951d914b7 Mon Sep 17 00:00:00 2001 From: Bo Deng Date: Wed, 29 Jul 2026 02:22:21 +0000 Subject: [PATCH 5/5] fix Signed-off-by: Bo Deng --- tensorrt_llm/_torch/disaggregation/transceiver.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tensorrt_llm/_torch/disaggregation/transceiver.py b/tensorrt_llm/_torch/disaggregation/transceiver.py index c3e4237edc68..82dd56223969 100644 --- a/tensorrt_llm/_torch/disaggregation/transceiver.py +++ b/tensorrt_llm/_torch/disaggregation/transceiver.py @@ -171,10 +171,10 @@ def summarize( sessions: Dict[int, Any], include_receiver_ready: bool, ) -> str: - session_items = list(sessions.copy().items()) + sessions_snapshot = list(sessions.values()) status_counts = Counter() receiver_ready = 0 - for _, session in session_items: + for session in sessions_snapshot: status = session.status if isinstance(status, SessionStatus): status_counts[status] += 1 @@ -185,7 +185,7 @@ def summarize( receiver_ready += int(bool(session.receiver_ready)) fields = [ - f"sessions={len(session_items)}", + f"sessions={len(sessions_snapshot)}", f"init={status_counts[SessionStatus.INIT]}", f"ready_to_transfer={status_counts[SessionStatus.READY]}", f"transferring={status_counts[SessionStatus.TRANSFERRING]}", @@ -196,7 +196,7 @@ def summarize( f"unknown={status_counts['unknown']}", ] if include_receiver_ready: - fields.append(f"peer_ready={receiver_ready}/{len(session_items)}") + fields.append(f"peer_ready={receiver_ready}/{len(sessions_snapshot)}") return ", ".join(fields) tx_status = summarize(self._send_sessions, include_receiver_ready=True)