[TRTLLM-14010][feat] report KV cache transfer state on executor hangs

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 <atomic>
 #include <cstddef>
 #include <fstream>
 #include <future>
@@ -309,15 +310,47 @@ class CacheTransceiver : public BaseCacheTransceiver
     [[nodiscard]] std::vector<char> 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);
 
     // 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<LlmRequest*> const& completedRequests);
+    void publishStatusSnapshot() noexcept;
 
     std::unique_ptr<CacheSender> mCacheSender;
     std::unique_ptr<CacheReceiver> mCacheReceiver;
@@ -338,6 +371,12 @@ class CacheTransceiver : public BaseCacheTransceiver
     std::unordered_set<LlmRequest::RequestIdType> mCompletedRequesterRequestIds;
     std::unordered_set<LlmRequest::RequestIdType> mFailedRequesterRequestIds;
     std::unordered_map<LlmRequest::RequestIdType, std::shared_ptr<LlmRequest>> 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<CacheTransceiverComm> mGroupComm;

cpp/tensorrt_llm/batch_manager/cacheTransceiver.cpp

@@ -64,6 +64,7 @@
 #include <numeric>
 #include <random>
 #include <sstream>
+#include <system_error>
 #include <thread>
 #include <unordered_map>
 #include <unordered_set>
@@ -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,73 @@ 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::unique_lock<std::mutex> 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;
+    }
+    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<std::mutex> 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 +862,7 @@ void CacheTransceiver::respondAndSendAsync(std::shared_ptr<LlmRequest> llmReques
     setContextState(llmRequest.get());
     auto future = mCacheSender->sendAsync(llmRequest);
     mSenderFutures.emplace_back(std::move(llmRequest), std::move(future));
+    publishStatusSnapshot();
 }
 
 void CacheTransceiver::respondAndSendLayerWise(
@@ -797,6 +880,7 @@ void CacheTransceiver::respondAndSendLayerWise(
         auto future = mCacheSender->sendAsync(llmRequest);
         mSenderFutures.emplace_back(llmRequest, std::move(future));
     }
+    publishStatusSnapshot();
 }
 
 void CacheTransceiver::requestAndReceiveSync(std::shared_ptr<LlmRequest> llmRequest)
@@ -806,6 +890,7 @@ void CacheTransceiver::requestAndReceiveSync(std::shared_ptr<LlmRequest> 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 +939,7 @@ void CacheTransceiver::requestAndReceiveAsync(std::shared_ptr<LlmRequest> 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<LlmRequest::RequestIdType> gatherRequestIds(
@@ -1024,7 +1110,7 @@ RequestStatuses CacheTransceiver::checkContextTransferStatus(
             contextCompleteRequestIds.push_back(request->mRequestId);
         }
     }
-
+    publishStatusSnapshot();
     std::unordered_map<LlmRequest::RequestIdType, int> frequencyMap;
     if ((syncComm) && syncComm->getSize() > 1)
     {
@@ -1174,6 +1260,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 +1359,7 @@ RequestStatuses CacheTransceiver::checkContextTransferStatus(
             requestId, mCompletedSenderRequestIds, mFailedSenderRequestIds, mSenderRequestsAwaitingConsensus);
     }
 
+    publishStatusSnapshot();
     return requestsStatus;
 }
 
@@ -1313,6 +1402,7 @@ void CacheTransceiver::checkGenTransferStatus(std::optional<int> const& atLeastR
             collectReadyRequestIds();
         }
     }
+    publishStatusSnapshot();
     std::unordered_map<LlmRequest::RequestIdType, int> frequencyMap;
 
     std::vector<LlmRequest::RequestIdType> toBlockRequestIds;
@@ -1466,6 +1556,8 @@ void CacheTransceiver::checkGenTransferStatus(std::optional<int> 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<RequestIdType>{});
@@ -1524,6 +1616,7 @@ void CacheTransceiver::checkGenTransferStatus(std::optional<int> 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.

cpp/tensorrt_llm/nanobind/batch_manager/cacheTransceiver.cpp

@@ -27,6 +27,7 @@
 #include <nanobind/nanobind.h>
 #include <nanobind/stl/optional.h>
 #include <nanobind/stl/shared_ptr.h>
+#include <nanobind/stl/string.h>
 #include <nanobind/stl/unique_ptr.h>
 #include <nanobind/stl/vector.h>
 #include <nanobind/trampoline.h>
@@ -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<SizeType32>{});
+            nb::arg("rnn_layer_num_per_pp") = std::vector<SizeType32>{})
+        .def("get_status_dump", &tb::CacheTransceiver::getStatusDump);
 
     nb::class_<tb::CacheTransceiverComm>(m, "CacheTransceiverComm")
         .def(

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,48 @@ 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:
+            sessions_snapshot = list(sessions.values())
+            status_counts = Counter()
+            receiver_ready = 0
+            for session in sessions_snapshot:
+                status = session.status
+                if isinstance(status, SessionStatus):
+                    status_counts[status] += 1
+                else:
+                    status_counts["unknown"] += 1
+
+                if include_receiver_ready:
+                    receiver_ready += int(bool(session.receiver_ready))
+
+            fields = [
+                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]}",
+                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(sessions_snapshot)}")
+            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

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,13 +114,38 @@ def run_loop():
         self.loop_thread = threading.Thread(target=run_loop, daemon=True, name="hang_detector_loop")
         self.loop_thread.start()
 
-    async def _detect_hang(self):
+    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) -> None:
         await asyncio.sleep(self.timeout)
         with self.lock:
-            self._detected = True
-            logger.error(f"Hang detected after {self.timeout} seconds.")
+            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:
+                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}"
+                )
+        try:
             print_all_stacks()
-            self.on_detected()
+        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()
 
     def detected(self):
         """Return True if hang is detected."""