From 7a865f3f0d80c804d2d2c4067b7afa52bfddbccf Mon Sep 17 00:00:00 2001 From: Chien-Chun Hung <2679986+chienchunhung@users.noreply.github.com> Date: Fri, 17 Jul 2026 15:10:48 -0700 Subject: [PATCH 1/5] [NVBUG-6448152][test] establish post-#15139 merge treatment Signed-off-by: Chien-Chun Hung <2679986+chienchunhung@users.noreply.github.com> From e83aa182f277b4a4a462baef43396efede4f5a27 Mon Sep 17 00:00:00 2001 From: Yanchao Lu Date: Mon, 29 Jun 2026 22:48:48 +0800 Subject: [PATCH 2/5] [None][ci] Disable home mount for sbatch L0 tests (#15713) Signed-off-by: Yanchao Lu (cherry picked from commit 833ddd2a59037a7dd961ded5a9ce4a6d010d1bcb) Signed-off-by: Chien-Chun Hung <2679986+chienchunhung@users.noreply.github.com> --- jenkins/L0_Test.groovy | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/jenkins/L0_Test.groovy b/jenkins/L0_Test.groovy index 306b2bc3084f..f8296852fdd6 100644 --- a/jenkins/L0_Test.groovy +++ b/jenkins/L0_Test.groovy @@ -1,3 +1,19 @@ +/* + * Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved. + * + * 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. + */ + @Library(['bloom-jenkins-shared-lib@main', 'trtllm-jenkins-shared-lib@main']) _ import java.lang.InterruptedException @@ -980,6 +996,8 @@ def getMountListForSlurmTest(SlurmCluster cluster, boolean useSbatch = false) throw new Exception("Unsupported container runtime: ${cluster.containerRuntime}") } + // TODO: Add mounts for different cache directories like pip, triton, etc. + return mounts } @@ -1208,6 +1226,7 @@ def runLLMTestlistWithSbatch(pipeline, platform, testList, config=VANILLA_CONFIG "--container-image=$containerImageArg", "--container-workdir=$jobWorkspace", "--container-mounts=$mounts", + "--no-container-mount-home", "--container-env=NVIDIA_IMEX_CHANNELS" ] envVarsToExport.each { varName, varValue -> From 4edb4d797f54192159c1fba93f8968cf90d3c6f4 Mon Sep 17 00:00:00 2001 From: Chien-Chun Hung <2679986+chienchunhung@users.noreply.github.com> Date: Sat, 18 Jul 2026 10:45:44 -0700 Subject: [PATCH 3/5] [NVBUG-6448152][test] backport sender wakeup fix Apply only the CacheSender readiness synchronization portion of #15737 to the exact #15139 boundary experiment. This keeps the historical control and treatment matched while removing the known lost-wakeup confound. Signed-off-by: Chien-Chun Hung <2679986+chienchunhung@users.noreply.github.com> --- .../batch_manager/dataTransceiver.cpp | 273 +++++++++--------- 1 file changed, 140 insertions(+), 133 deletions(-) diff --git a/cpp/tensorrt_llm/batch_manager/dataTransceiver.cpp b/cpp/tensorrt_llm/batch_manager/dataTransceiver.cpp index 853866687d53..b3eb1c0f9843 100644 --- a/cpp/tensorrt_llm/batch_manager/dataTransceiver.cpp +++ b/cpp/tensorrt_llm/batch_manager/dataTransceiver.cpp @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 * * Licensed under the Apache License, Version 2.0 (the "License"); @@ -33,6 +33,8 @@ #include #include #include +#include +#include #include namespace tensorrt_llm::batch_manager @@ -286,7 +288,6 @@ class CacheSender::Impl TLLM_CHECK(mManager); TLLM_CHECK(mManager->getCommState().getSelfIdx() == selfIndex); TLLM_CUDA_CHECK(cudaGetDevice(&mDeviceId)); - mCurrentRequest = std::nullopt; mResponseFuture = std::async(std::launch::async, &Impl::response, this); int asyncSendThreadNum = common::getEnvKVCacheSendMaxConcurrenceNum(); for (int i = 0; i < asyncSendThreadNum; i++) @@ -303,12 +304,13 @@ class CacheSender::Impl auto future = promise.get_future(); llmRequest->setKvCacheTransferStart(LlmRequest::getSteadyClockNow()); { - { - std::scoped_lock lkResp(mSenderMutex); - mReadyResponses.emplace(llmRequest->mRequestId, Response{llmRequest, std::move(promise)}); - } - std::unique_lock lkCond(mCondMutex); - mAnyReady = true; + std::scoped_lock lock(mSenderMutex); + TLLM_CHECK_WITH_INFO( + !mTerminate, "Cannot enqueue request %zu after CacheSender termination", llmRequest->mRequestId); + auto const result + = mReadyResponses.emplace(llmRequest->mRequestId, Response{llmRequest, std::move(promise)}); + TLLM_CHECK_WITH_INFO( + result.second, "Request %zu is already queued for KV cache transfer", llmRequest->mRequestId); } mSenderCv.notify_all(); return future; @@ -425,11 +427,11 @@ class CacheSender::Impl bool cancelRequest(LlmRequest const& llmRequest) { bool isCancelled = false; - std::scoped_lock lkResp(mSenderMutex); + std::scoped_lock lock(mSenderMutex); auto it = mReadyResponses.find(llmRequest.mRequestId); // If the request is not the current request and already in the ready queue, we can cancel it. if (it != mReadyResponses.end() - && (!mCurrentRequest.has_value() || getCurrentRequestId() != llmRequest.mRequestId)) + && (!mCurrentRequest.has_value() || mCurrentRequest.value() != llmRequest.mRequestId)) { mCancelledRequests.insert(llmRequest.mRequestId); isCancelled = true; @@ -546,191 +548,196 @@ class CacheSender::Impl void asyncSendAndRemoveResponse(RequestIdType id, Response resp) noexcept { - std::unique_lock lk(mAsyncSendResource.mMtxForQueue); - mAsyncSendResource.mSendQueue.emplace_back(std::move(resp)); - mAsyncSendResource.mCVforQueue.notify_one(); + try + { + std::unique_lock lock(mAsyncSendResource.mMtxForQueue); + mAsyncSendResource.mSendQueue.emplace_back(std::move(resp)); + mAsyncSendResource.mCVforQueue.notify_one(); + } + catch (std::exception const& err) + { + TLLM_LOG_ERROR("Failed to queue asynchronous KV cache send for request %zu: %s", id, err.what()); + failResponse(resp, std::current_exception()); + } } - void sendResponse(std::map::iterator it) + void sendResponse(RequestIdType reqId) { - auto reqId = mCurrentRequest.value(); - auto count = --mRemainSendCount[reqId]; - TLLM_CHECK(count >= 0); - if (count == 0) + bool isReady = true; { - mRemainSendCount.erase(reqId); - - // Check if the request is cancelled - bool isReady = true; + std::scoped_lock lock(mSenderMutex); + TLLM_CHECK(mCurrentRequest.has_value() && mCurrentRequest.value() == reqId); + TLLM_CHECK(mReadyResponses.find(reqId) != mReadyResponses.end()); + auto countIt = mRemainSendCount.find(reqId); + TLLM_CHECK(countIt != mRemainSendCount.end()); + auto const count = --countIt->second; + TLLM_CHECK(count >= 0); + if (count > 0) { - std::scoped_lock lk(mSenderMutex); - if (mCancelledRequests.find(reqId) != mCancelledRequests.end()) - { - isReady = false; - } + mCurrentRequest = std::nullopt; + return; } - sendReadySignal(reqId, isReady); + mRemainSendCount.erase(countIt); + isReady = mCancelledRequests.find(reqId) == mCancelledRequests.end(); + } - if (isReady) + // Keep mCurrentRequest set while notifying the peer so cancellation cannot change the decision after it has + // been made. The network operation must not run under mSenderMutex. + sendReadySignal(reqId, isReady); + + Response response; + { + std::scoped_lock lock(mSenderMutex); + auto it = mReadyResponses.find(reqId); + TLLM_CHECK(it != mReadyResponses.end()); + response = std::move(it->second); + mReadyResponses.erase(it); + mCancelledRequests.erase(reqId); + mCurrentRequest = std::nullopt; + } + + if (isReady) + { + if (dynamic_cast(mManager) != nullptr) { - if (dynamic_cast(mManager) != nullptr) - { - // our nixl impl seems only support recv and send in the same thread - // if we use zmq as control path, we may avoid this issue - sendAndRemoveResponse(it->first, std::move(it->second)); - } - else - { - // if we send data in another thread, multiple rank may send data for different requests at the same - // time with gen DP case. - asyncSendAndRemoveResponse(it->first, std::move(it->second)); - } - removeResponse(it); + // Our NIXL implementation only supports recv and send in the same thread. Using ZMQ for the control + // path may avoid this limitation. + sendAndRemoveResponse(reqId, std::move(response)); } else { - // TODO: if the generation does not require the kv cache, the request will - // not be removed from mCancelledRequests. This should be handled by timeout. - auto const cancelledReqId = mCurrentRequest.value(); - Response cancelledResponse; - { - std::scoped_lock lkResp(mSenderMutex); - auto it = mReadyResponses.find(cancelledReqId); - TLLM_CHECK(it != mReadyResponses.end()); - // Move out before erasing so the promise survives the - // map cleanup and can be resolved (vs. destroyed unfulfilled, - // which would surface as std::future_error: Broken promise). - cancelledResponse = std::move(it->second); - mReadyResponses.erase(it); - mCancelledRequests.erase(cancelledReqId); - mRemainSendCount.erase(cancelledReqId); - } - cancelledResponse.mPromise.set_exception(std::make_exception_ptr( - TLLM_REQUEST_EXCEPTION(cancelledReqId, common::RequestErrorCode::kNETWORK_ERROR, - "KV cache transfer for request %zu was cancelled", cancelledReqId))); - mCurrentRequest = std::nullopt; - - if (mReadyResponses.empty()) - { - std::unique_lock lk(mCondMutex); - mAnyReady = false; - } + // If we send data in another thread, multiple ranks may send data for different requests at the same + // time with generation attention DP. + asyncSendAndRemoveResponse(reqId, std::move(response)); } } - mCurrentRequest = std::nullopt; + else + { + response.mPromise.set_exception(std::make_exception_ptr(TLLM_REQUEST_EXCEPTION(reqId, + common::RequestErrorCode::kNETWORK_ERROR, "KV cache transfer for request %zu was cancelled", reqId))); + } } void response() noexcept { + std::exception_ptr responseException; try { tensorrt_llm::common::setThreadName("dataTransResp"); TLLM_CUDA_CHECK(cudaSetDevice(mDeviceId)); - while (!mTerminate || !mAnyReady) + while (true) { - if (!mAnyReady) { - std::unique_lock lk(mCondMutex); - mSenderCv.wait(lk, [this]() { return (mAnyReady || mTerminate); }); + std::unique_lock lock(mSenderMutex); + mSenderCv.wait(lock, [this]() { return mTerminate || !mReadyResponses.empty(); }); + if (mTerminate) + { + break; + } } - if (mTerminate) + + auto const requestInfo = recvRequestInfo(); + if (mTerminate || !mManager->isRunning()) { break; } - if (!mReadyResponses.empty()) - { - auto const& requestInfo = recvRequestInfo(); - if (mTerminate || !mManager->isRunning()) - { - return; - } - auto reqId = requestInfo.getRequestId(); + auto const reqId = requestInfo.getRequestId(); - { - std::scoped_lock lk(mSenderMutex); - mCurrentRequest = reqId; - } - - if (mRemainSendCount.find(reqId) == mRemainSendCount.end()) - { - mRemainSendCount[reqId] = getCounterpartsCount(reqId); - } - } - auto it = getCurrentResponse(); - if (it != mReadyResponses.end()) + if (mRemainSendCount.find(reqId) == mRemainSendCount.end()) { - sendResponse(it); + mRemainSendCount[reqId] = getCounterpartsCount(reqId); } - else + { - auto it = getCurrentResponse(); - while (it == mReadyResponses.end()) + std::unique_lock lock(mSenderMutex); + mCurrentRequest = reqId; + mSenderCv.wait(lock, + [this, reqId]() { return mTerminate || mReadyResponses.find(reqId) != mReadyResponses.end(); }); + if (mTerminate) { - std::unique_lock lk(mCondMutex); - mSenderCv.wait(lk, [this]() { return (mAnyReady || mTerminate); }); - if (mTerminate) - { - break; - } - it = getCurrentResponse(); + mCurrentRequest = std::nullopt; + break; } - sendResponse(it); } + sendResponse(reqId); } } catch (std::exception const& err) { TLLM_LOG_ERROR("Exception in CacheSender response: %s", err.what()); - for (auto& it : mReadyResponses) - { - it.second.mPromise.set_exception(std::current_exception()); - } + responseException = std::current_exception(); } + + if (!responseException) + { + responseException + = std::make_exception_ptr(std::runtime_error("CacheSender terminated before response completed")); + } + { + std::scoped_lock lock(mSenderMutex); + mTerminate = true; + } + mSenderCv.notify_all(); + failPendingResponses(responseException); } void terminate() { { - std::unique_lock lk(mCondMutex); + std::scoped_lock lock(mSenderMutex); mTerminate = true; } - // We don't have to wait for the future. If another thread is sending data, it won't pay attention - // to the terminate flag. mSenderCv.notify_all(); - mAsyncSendResource.mTerminate = true; + if (mResponseFuture.valid()) + { + mResponseFuture.get(); + } + + std::deque pendingAsyncResponses; + { + std::scoped_lock lock(mAsyncSendResource.mMtxForQueue); + mAsyncSendResource.mTerminate = true; + pendingAsyncResponses.swap(mAsyncSendResource.mSendQueue); + } mAsyncSendResource.mCVforQueue.notify_all(); for (auto& future : mAsyncSendFutures) { future.get(); } - if (mResponseFuture.valid()) + auto const exception + = std::make_exception_ptr(std::runtime_error("CacheSender terminated before asynchronous send completed")); + for (auto& response : pendingAsyncResponses) { - mResponseFuture.get(); + failResponse(response, exception); } } - void removeResponse(std::map::iterator it) + void failResponse(Response& response, std::exception_ptr const& exception) noexcept { + try { - std::scoped_lock lkResp(mSenderMutex); - mReadyResponses.erase(it); + response.mPromise.set_exception(exception); } - if (mReadyResponses.empty()) + catch (std::future_error const& err) { - std::unique_lock lkCond(mCondMutex); - mAnyReady = false; + TLLM_LOG_ERROR("Failed to set CacheSender response exception: %s", err.what()); } } - [[nodiscard]] RequestIdType getCurrentRequestId() const + void failPendingResponses(std::exception_ptr const& exception) noexcept { - return mCurrentRequest.value(); - } - - [[nodiscard]] std::map::iterator getCurrentResponse() - { - std::scoped_lock lk(mSenderMutex); - return mReadyResponses.find(getCurrentRequestId()); + std::map pendingResponses; + { + std::scoped_lock lock(mSenderMutex); + pendingResponses.swap(mReadyResponses); + mCurrentRequest = std::nullopt; + mCancelledRequests.clear(); + mRemainSendCount.clear(); + } + for (auto& entry : pendingResponses) + { + failResponse(entry.second, exception); + } } public: @@ -746,9 +753,9 @@ class CacheSender::Impl std::optional mCurrentRequest; std::set mCancelledRequests; std::map mReadyResponses; - std::mutex mSenderMutex, mCondMutex; - std::atomic mAnyReady{false}, mTerminate{false}; - std::condition_variable mSenderCv, mResponderCv; + std::mutex mSenderMutex; + std::atomic mTerminate{false}; + std::condition_variable mSenderCv; std::future mResponseFuture; std::unordered_map mRemainSendCount; AsyncSendResource mAsyncSendResource; From ad79abd04e9ec675d820e2ff7a39cfa540a8cba4 Mon Sep 17 00:00:00 2001 From: Chien-Chun Hung <2679986+chienchunhung@users.noreply.github.com> Date: Sat, 18 Jul 2026 14:20:15 -0700 Subject: [PATCH 4/5] [NVBUG-6448152][test] trace disagg transfer lifecycle Signed-off-by: Chien-Chun Hung <2679986+chienchunhung@users.noreply.github.com> --- .../batch_manager/cacheTransceiver.h | 11 + .../batch_manager/cacheTransceiver.cpp | 294 +++++++++++++++++- tensorrt_llm/_torch/pyexecutor/py_executor.py | 138 ++++++++ ...tx1_pp4_gen1_dep8_eplb0_mtp1_ccb-NIXL.yaml | 2 +- 4 files changed, 439 insertions(+), 6 deletions(-) diff --git a/cpp/include/tensorrt_llm/batch_manager/cacheTransceiver.h b/cpp/include/tensorrt_llm/batch_manager/cacheTransceiver.h index 80a06ea5ddf7..f79d51ae59ae 100644 --- a/cpp/include/tensorrt_llm/batch_manager/cacheTransceiver.h +++ b/cpp/include/tensorrt_llm/batch_manager/cacheTransceiver.h @@ -26,6 +26,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 @@ -278,6 +279,16 @@ class CacheTransceiver : public BaseCacheTransceiver void setContextState(LlmRequest* llmRequest); + // Default-off, transition-only lifecycle tracing for NVBUG#6448152. + int mWorldRank; + bool mNvbug6448152TraceEnabled; + std::uint64_t mNvbug6448152ContextCheckSequence{0}; + std::uint64_t mNvbug6448152ContextTpConsensusSequence{0}; + std::uint64_t mNvbug6448152ContextPpConsensusSequence{0}; + std::uint64_t mNvbug6448152GenerationCheckSequence{0}; + std::uint64_t mNvbug6448152ContextEnqueuedTransitionCount{0}; + std::unordered_set mNvbug6448152SenderWaitTimeoutIds; + std::unique_ptr mCacheSender; std::unique_ptr mCacheReceiver; // shared_ptr (not raw LlmRequest*) so the futures hold a strong reference for diff --git a/cpp/tensorrt_llm/batch_manager/cacheTransceiver.cpp b/cpp/tensorrt_llm/batch_manager/cacheTransceiver.cpp index 18c2b19e9eeb..6d3bfa658fdf 100644 --- a/cpp/tensorrt_llm/batch_manager/cacheTransceiver.cpp +++ b/cpp/tensorrt_llm/batch_manager/cacheTransceiver.cpp @@ -53,6 +53,7 @@ #include "tensorrt_llm/runtime/utils/mpiUtils.h" #include "tensorrt_llm/runtime/utils/pgUtils.h" #include +#include #include #include #include @@ -68,6 +69,13 @@ namespace using RequestIdType = LlmRequest::RequestIdType; +constexpr long long kNvbug6448152SlowConsensusThresholdUs = 1'000'000; + +bool isNvbug6448152TraceEnabled() +{ + return common::getBoolEnv("TRTLLM_NVBUG_6448152_TRACE"); +} + enum class TransferConsensusState : std::uint64_t { kCompleted = 1, @@ -276,7 +284,9 @@ CacheTransceiver::CacheTransceiver(kv_cache_manager::BaseKVCacheManager* cacheMa executor::kv_cache::CacheState::AttentionType attentionType, std::optional cacheTransceiverConfig, rnn_state_manager::RnnStateManager* rnnStateManager, std::vector const& rnnLayerNumPerPP) - : mCacheTransceiverConfig{cacheTransceiverConfig} + : mWorldRank{worldConfig.getRank()} + , mNvbug6448152TraceEnabled{isNvbug6448152TraceEnabled()} + , mCacheTransceiverConfig{cacheTransceiverConfig} , mRnnStateManager{rnnStateManager} { using tensorrt_llm::batch_manager::kv_cache_manager::CacheFormatter; @@ -555,6 +565,17 @@ void CacheTransceiver::respondAndSendAsync(std::shared_ptr llmReques setContextState(llmRequest.get()); auto future = mCacheSender->sendAsync(llmRequest); mSenderFutures.emplace_back(std::move(llmRequest), std::move(future)); + if (mNvbug6448152TraceEnabled) + { + mNvbug6448152ContextEnqueuedTransitionCount++; + auto const requestId = mSenderFutures.back().first->mRequestId; + TLLM_LOG_INFO( + "NVBUG6448152_DIAG event=transfer_start side=ctx rank=%d request_id=%llu check_seq=%llu " + "future_count=%zu awaiting_count=%zu timeout_count=%zu", + mWorldRank, static_cast(requestId), + static_cast(mNvbug6448152ContextCheckSequence), mSenderFutures.size(), + mSenderRequestsAwaitingConsensus.size(), mNvbug6448152SenderWaitTimeoutIds.size()); + } } void CacheTransceiver::respondAndSendLayerWise( @@ -571,6 +592,16 @@ void CacheTransceiver::respondAndSendLayerWise( setContextState(llmRequest.get()); auto future = mCacheSender->sendAsync(llmRequest); mSenderFutures.emplace_back(llmRequest, std::move(future)); + if (mNvbug6448152TraceEnabled) + { + mNvbug6448152ContextEnqueuedTransitionCount++; + TLLM_LOG_INFO( + "NVBUG6448152_DIAG event=transfer_start side=ctx_layerwise rank=%d request_id=%llu " + "check_seq=%llu future_count=%zu awaiting_count=%zu timeout_count=%zu", + mWorldRank, static_cast(llmRequest->mRequestId), + static_cast(mNvbug6448152ContextCheckSequence), mSenderFutures.size(), + mSenderRequestsAwaitingConsensus.size(), mNvbug6448152SenderWaitTimeoutIds.size()); + } } } @@ -601,6 +632,15 @@ 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); + if (mNvbug6448152TraceEnabled) + { + TLLM_LOG_INFO( + "NVBUG6448152_DIAG event=transfer_start side=gen rank=%d request_id=%llu check_seq=%llu " + "future_count=%zu awaiting_count=%zu timeout_count=%zu", + mWorldRank, static_cast(requestPtr->mRequestId), + static_cast(mNvbug6448152GenerationCheckSequence), mRequesterFutures.size(), + mRequesterRequestsAwaitingConsensus.size(), mTimedOutRequesterIds.size()); + } } std::vector gatherRequestIds( @@ -700,6 +740,22 @@ void updateKVCacheTransferBW(std::shared_ptr const& mComm, RequestStatuses CacheTransceiver::checkContextTransferStatus( std::optional const& atLeastRequestNum, bool markComplete) { + std::uint64_t contextCheckSequence = 0; + bool traceConsensusTransition = false; + bool periodicConsensusCheckpoint = false; + std::uint64_t enqueuedTransitionCount = 0; + size_t localTerminalTransitionCount = 0; + size_t firstWaitTimeoutTransitionCount = 0; + if (mNvbug6448152TraceEnabled) + { + contextCheckSequence = ++mNvbug6448152ContextCheckSequence; + enqueuedTransitionCount = mNvbug6448152ContextEnqueuedTransitionCount; + mNvbug6448152ContextEnqueuedTransitionCount = 0; + periodicConsensusCheckpoint + = contextCheckSequence % 512 == 0 && (!mSenderFutures.empty() || !mSenderRequestsAwaitingConsensus.empty()); + traceConsensusTransition = enqueuedTransitionCount > 0 || periodicConsensusCheckpoint; + } + bool blockAll = !atLeastRequestNum.has_value(); std::optional senderFutureTimeoutMs = std::nullopt; // If blockAll is true, we want to block and not use a timeout @@ -794,14 +850,47 @@ RequestStatuses CacheTransceiver::checkContextTransferStatus( { future.get(); bool const failed = request->getState() == LlmRequestState::kDISAGG_TRANS_ERROR; + long elapsedMs = 0; + if (mNvbug6448152TraceEnabled) + { + auto const elapsed = std::chrono::duration_cast( + LlmRequest::getSteadyClockNow() - request->getKvCacheTransferStart()); + elapsedMs = static_cast(elapsed.count()); + } recordLocalTransferOutcome(requestId, request, failed, mCompletedSenderRequestIds, mFailedSenderRequestIds, mSenderRequestsAwaitingConsensus); it = mSenderFutures.erase(it); + if (mNvbug6448152TraceEnabled) + { + traceConsensusTransition = true; + localTerminalTransitionCount++; + TLLM_LOG_INFO( + "NVBUG6448152_DIAG event=local_future_terminal side=ctx rank=%d request_id=%llu " + "outcome=%s elapsed_ms=%ld check_seq=%llu future_count=%zu awaiting_count=%zu " + "timeout_count=%zu", + mWorldRank, static_cast(requestId), failed ? "failed" : "completed", + elapsedMs, static_cast(contextCheckSequence), mSenderFutures.size(), + mSenderRequestsAwaitingConsensus.size(), mNvbug6448152SenderWaitTimeoutIds.size()); + } } else if (status == std::future_status::timeout) { - TLLM_LOG_WARNING("Timed out waiting for context KV cache transfer after %d milliseconds.", - senderFutureTimeoutMs.value()); + if (!mNvbug6448152TraceEnabled) + { + TLLM_LOG_WARNING("Timed out waiting for context KV cache transfer after %d milliseconds.", + senderFutureTimeoutMs.value()); + } + else if (mNvbug6448152SenderWaitTimeoutIds.insert(requestId).second) + { + traceConsensusTransition = true; + firstWaitTimeoutTransitionCount++; + TLLM_LOG_WARNING( + "NVBUG6448152_DIAG event=first_future_wait_timeout side=ctx rank=%d request_id=%llu " + "wait_ms=%d check_seq=%llu future_count=%zu awaiting_count=%zu timeout_count=%zu", + mWorldRank, static_cast(requestId), senderFutureTimeoutMs.value(), + static_cast(contextCheckSequence), mSenderFutures.size(), + mSenderRequestsAwaitingConsensus.size(), mNvbug6448152SenderWaitTimeoutIds.size()); + } ++it; } else @@ -812,6 +901,18 @@ RequestStatuses CacheTransceiver::checkContextTransferStatus( recordLocalTransferOutcome(requestId, request, /*failed=*/true, mCompletedSenderRequestIds, mFailedSenderRequestIds, mSenderRequestsAwaitingConsensus); it = mSenderFutures.erase(it); + if (mNvbug6448152TraceEnabled) + { + traceConsensusTransition = true; + localTerminalTransitionCount++; + TLLM_LOG_INFO( + "NVBUG6448152_DIAG event=local_future_terminal side=ctx rank=%d request_id=%llu " + "outcome=unexpected_future_status check_seq=%llu future_count=%zu awaiting_count=%zu " + "timeout_count=%zu", + mWorldRank, static_cast(requestId), + static_cast(contextCheckSequence), mSenderFutures.size(), + mSenderRequestsAwaitingConsensus.size(), mNvbug6448152SenderWaitTimeoutIds.size()); + } } } catch (std::exception const& e) @@ -820,6 +921,17 @@ RequestStatuses CacheTransceiver::checkContextTransferStatus( recordLocalTransferOutcome(requestId, request, /*failed=*/true, mCompletedSenderRequestIds, mFailedSenderRequestIds, mSenderRequestsAwaitingConsensus); it = mSenderFutures.erase(it); + if (mNvbug6448152TraceEnabled) + { + traceConsensusTransition = true; + localTerminalTransitionCount++; + TLLM_LOG_INFO( + "NVBUG6448152_DIAG event=local_future_terminal side=ctx rank=%d request_id=%llu " + "outcome=exception check_seq=%llu future_count=%zu awaiting_count=%zu timeout_count=%zu", + mWorldRank, static_cast(requestId), + static_cast(contextCheckSequence), mSenderFutures.size(), + mSenderRequestsAwaitingConsensus.size(), mNvbug6448152SenderWaitTimeoutIds.size()); + } } } else @@ -829,8 +941,117 @@ RequestStatuses CacheTransceiver::checkContextTransferStatus( } RequestStatuses requestsStatus{}; - auto const consensusOutcome - = reduceTransferStates(syncComm, mGroupPipeParaComm, mCompletedSenderRequestIds, mFailedSenderRequestIds); + TransferConsensusOutcome consensusOutcome; + std::uint64_t tpConsensusSequence = 0; + std::uint64_t ppConsensusSequence = 0; + long long tpConsensusDurationUs = 0; + long long ppConsensusDurationUs = 0; + if (mNvbug6448152TraceEnabled) + { + tpConsensusSequence = ++mNvbug6448152ContextTpConsensusSequence; + ppConsensusSequence = ++mNvbug6448152ContextPpConsensusSequence; + + int const tpRank = syncComm != nullptr ? syncComm->getRank() : 0; + int const tpSize = syncComm != nullptr ? syncComm->getSize() : 1; + if (traceConsensusTransition) + { + TLLM_LOG_INFO( + "NVBUG6448152_DIAG event=consensus_enter side=ctx stage=tp rank=%d comm_rank=%d comm_size=%d " + "check_seq=%llu consensus_seq=%llu local_terminal_transitions=%zu " + "first_wait_timeout_transitions=%zu enqueued_transitions=%llu periodic_checkpoint=%d " + "completed_count=%zu failed_count=%zu future_count=%zu awaiting_count=%zu timeout_count=%zu", + mWorldRank, tpRank, tpSize, static_cast(contextCheckSequence), + static_cast(tpConsensusSequence), localTerminalTransitionCount, + firstWaitTimeoutTransitionCount, static_cast(enqueuedTransitionCount), + static_cast(periodicConsensusCheckpoint), mCompletedSenderRequestIds.size(), + mFailedSenderRequestIds.size(), mSenderFutures.size(), mSenderRequestsAwaitingConsensus.size(), + mNvbug6448152SenderWaitTimeoutIds.size()); + } + auto const tpConsensusStart = std::chrono::steady_clock::now(); + auto const tpOutcome = reduceTransferStates(syncComm, mCompletedSenderRequestIds, mFailedSenderRequestIds); + tpConsensusDurationUs + = std::chrono::duration_cast(std::chrono::steady_clock::now() - tpConsensusStart) + .count(); + if (traceConsensusTransition) + { + TLLM_LOG_INFO( + "NVBUG6448152_DIAG event=consensus_exit side=ctx stage=tp rank=%d comm_rank=%d comm_size=%d " + "check_seq=%llu consensus_seq=%llu duration_us=%lld local_terminal_transitions=%zu " + "first_wait_timeout_transitions=%zu enqueued_transitions=%llu completed_count=%zu failed_count=%zu " + "future_count=%zu awaiting_count=%zu timeout_count=%zu", + mWorldRank, tpRank, tpSize, static_cast(contextCheckSequence), + static_cast(tpConsensusSequence), tpConsensusDurationUs, + localTerminalTransitionCount, firstWaitTimeoutTransitionCount, + static_cast(enqueuedTransitionCount), tpOutcome.completedRequestIds.size(), + tpOutcome.failedRequestIds.size(), mSenderFutures.size(), mSenderRequestsAwaitingConsensus.size(), + mNvbug6448152SenderWaitTimeoutIds.size()); + } + else if (tpConsensusDurationUs >= kNvbug6448152SlowConsensusThresholdUs) + { + TLLM_LOG_WARNING( + "NVBUG6448152_DIAG event=consensus_slow side=ctx stage=tp rank=%d comm_rank=%d comm_size=%d " + "check_seq=%llu consensus_seq=%llu duration_us=%lld completed_count=%zu failed_count=%zu " + "future_count=%zu awaiting_count=%zu timeout_count=%zu", + mWorldRank, tpRank, tpSize, static_cast(contextCheckSequence), + static_cast(tpConsensusSequence), tpConsensusDurationUs, + tpOutcome.completedRequestIds.size(), tpOutcome.failedRequestIds.size(), mSenderFutures.size(), + mSenderRequestsAwaitingConsensus.size(), mNvbug6448152SenderWaitTimeoutIds.size()); + } + + int const ppRank = mGroupPipeParaComm != nullptr ? mGroupPipeParaComm->getRank() : 0; + int const ppSize = mGroupPipeParaComm != nullptr ? mGroupPipeParaComm->getSize() : 1; + if (traceConsensusTransition) + { + TLLM_LOG_INFO( + "NVBUG6448152_DIAG event=consensus_enter side=ctx stage=pp rank=%d comm_rank=%d comm_size=%d " + "check_seq=%llu consensus_seq=%llu local_terminal_transitions=%zu " + "first_wait_timeout_transitions=%zu enqueued_transitions=%llu periodic_checkpoint=%d " + "completed_count=%zu failed_count=%zu future_count=%zu awaiting_count=%zu timeout_count=%zu", + mWorldRank, ppRank, ppSize, static_cast(contextCheckSequence), + static_cast(ppConsensusSequence), localTerminalTransitionCount, + firstWaitTimeoutTransitionCount, static_cast(enqueuedTransitionCount), + static_cast(periodicConsensusCheckpoint), tpOutcome.completedRequestIds.size(), + tpOutcome.failedRequestIds.size(), mSenderFutures.size(), mSenderRequestsAwaitingConsensus.size(), + mNvbug6448152SenderWaitTimeoutIds.size()); + } + auto const ppConsensusStart = std::chrono::steady_clock::now(); + consensusOutcome + = reduceTransferStates(mGroupPipeParaComm, tpOutcome.completedRequestIds, tpOutcome.failedRequestIds); + ppConsensusDurationUs + = std::chrono::duration_cast(std::chrono::steady_clock::now() - ppConsensusStart) + .count(); + if (traceConsensusTransition) + { + TLLM_LOG_INFO( + "NVBUG6448152_DIAG event=consensus_exit side=ctx stage=pp rank=%d comm_rank=%d comm_size=%d " + "check_seq=%llu consensus_seq=%llu duration_us=%lld local_terminal_transitions=%zu " + "first_wait_timeout_transitions=%zu enqueued_transitions=%llu completed_count=%zu failed_count=%zu " + "future_count=%zu awaiting_count=%zu timeout_count=%zu", + mWorldRank, ppRank, ppSize, static_cast(contextCheckSequence), + static_cast(ppConsensusSequence), ppConsensusDurationUs, + localTerminalTransitionCount, firstWaitTimeoutTransitionCount, + static_cast(enqueuedTransitionCount), consensusOutcome.completedRequestIds.size(), + consensusOutcome.failedRequestIds.size(), mSenderFutures.size(), + mSenderRequestsAwaitingConsensus.size(), mNvbug6448152SenderWaitTimeoutIds.size()); + } + else if (ppConsensusDurationUs >= kNvbug6448152SlowConsensusThresholdUs) + { + TLLM_LOG_WARNING( + "NVBUG6448152_DIAG event=consensus_slow side=ctx stage=pp rank=%d comm_rank=%d comm_size=%d " + "check_seq=%llu consensus_seq=%llu duration_us=%lld completed_count=%zu failed_count=%zu " + "future_count=%zu awaiting_count=%zu timeout_count=%zu", + mWorldRank, ppRank, ppSize, static_cast(contextCheckSequence), + static_cast(ppConsensusSequence), ppConsensusDurationUs, + consensusOutcome.completedRequestIds.size(), consensusOutcome.failedRequestIds.size(), + mSenderFutures.size(), mSenderRequestsAwaitingConsensus.size(), + mNvbug6448152SenderWaitTimeoutIds.size()); + } + } + else + { + consensusOutcome + = reduceTransferStates(syncComm, mGroupPipeParaComm, mCompletedSenderRequestIds, mFailedSenderRequestIds); + } for (auto const requestId : consensusOutcome.failedRequestIds) { auto const requestIt = mSenderRequestsAwaitingConsensus.find(requestId); @@ -843,6 +1064,20 @@ RequestStatuses CacheTransceiver::checkContextTransferStatus( mTimedOutSenderIds.erase(requestId); eraseLocalTransferOutcome( requestId, mCompletedSenderRequestIds, mFailedSenderRequestIds, mSenderRequestsAwaitingConsensus); + if (mNvbug6448152TraceEnabled) + { + bool const hadWaitTimeout = mNvbug6448152SenderWaitTimeoutIds.erase(requestId) > 0; + TLLM_LOG_INFO( + "NVBUG6448152_DIAG event=global_commit side=ctx rank=%d request_id=%llu outcome=failed " + "check_seq=%llu tp_consensus_seq=%llu pp_consensus_seq=%llu tp_duration_us=%lld " + "pp_duration_us=%lld had_wait_timeout=%d future_count=%zu awaiting_count=%zu timeout_count=%zu", + mWorldRank, static_cast(requestId), + static_cast(contextCheckSequence), + static_cast(tpConsensusSequence), + static_cast(ppConsensusSequence), tpConsensusDurationUs, ppConsensusDurationUs, + static_cast(hadWaitTimeout), mSenderFutures.size(), mSenderRequestsAwaitingConsensus.size(), + mNvbug6448152SenderWaitTimeoutIds.size()); + } } for (auto const requestId : consensusOutcome.completedRequestIds) { @@ -859,6 +1094,20 @@ RequestStatuses CacheTransceiver::checkContextTransferStatus( mTimedOutSenderIds.erase(requestId); eraseLocalTransferOutcome( requestId, mCompletedSenderRequestIds, mFailedSenderRequestIds, mSenderRequestsAwaitingConsensus); + if (mNvbug6448152TraceEnabled) + { + bool const hadWaitTimeout = mNvbug6448152SenderWaitTimeoutIds.erase(requestId) > 0; + TLLM_LOG_INFO( + "NVBUG6448152_DIAG event=global_commit side=ctx rank=%d request_id=%llu outcome=completed " + "mark_complete=%d check_seq=%llu tp_consensus_seq=%llu pp_consensus_seq=%llu tp_duration_us=%lld " + "pp_duration_us=%lld had_wait_timeout=%d future_count=%zu awaiting_count=%zu timeout_count=%zu", + mWorldRank, static_cast(requestId), static_cast(markComplete), + static_cast(contextCheckSequence), + static_cast(tpConsensusSequence), + static_cast(ppConsensusSequence), tpConsensusDurationUs, ppConsensusDurationUs, + static_cast(hadWaitTimeout), mSenderFutures.size(), mSenderRequestsAwaitingConsensus.size(), + mNvbug6448152SenderWaitTimeoutIds.size()); + } } return requestsStatus; @@ -866,6 +1115,12 @@ RequestStatuses CacheTransceiver::checkContextTransferStatus( void CacheTransceiver::checkGenTransferStatus(std::optional const& atLeastRequestNum) { + std::uint64_t generationCheckSequence = 0; + if (mNvbug6448152TraceEnabled) + { + generationCheckSequence = ++mNvbug6448152GenerationCheckSequence; + } + bool blockAll = !atLeastRequestNum.has_value(); std::vector genTransferReadyRequestIds; for (auto&& [request, future] : mRequesterFutures) @@ -1002,10 +1257,12 @@ void CacheTransceiver::checkGenTransferStatus(std::optional const& atLeastR } if (blockAll || toCompleteIdSet.find(requestId) != toCompleteIdSet.end()) { + bool localFailed = true; try { it->second.get(); bool const failed = request->getState() == LlmRequestState::kDISAGG_TRANS_ERROR; + localFailed = failed; if (failed) { // The receiver uses the error state as a local transfer-failed signal. @@ -1034,6 +1291,15 @@ void CacheTransceiver::checkGenTransferStatus(std::optional const& atLeastR request->getContextPhaseParams().value().getReqId()); } it = mRequesterFutures.erase(it); + if (mNvbug6448152TraceEnabled) + { + TLLM_LOG_INFO( + "NVBUG6448152_DIAG event=local_future_terminal side=gen rank=%d request_id=%llu outcome=%s " + "check_seq=%llu future_count=%zu awaiting_count=%zu timeout_count=%zu", + mWorldRank, static_cast(requestId), localFailed ? "failed" : "completed", + static_cast(generationCheckSequence), mRequesterFutures.size(), + mRequesterRequestsAwaitingConsensus.size(), mTimedOutRequesterIds.size()); + } } else { @@ -1054,6 +1320,15 @@ void CacheTransceiver::checkGenTransferStatus(std::optional const& atLeastR mTimedOutRequesterIds.erase(requestId); eraseLocalTransferOutcome( requestId, mCompletedRequesterRequestIds, mFailedRequesterRequestIds, mRequesterRequestsAwaitingConsensus); + if (mNvbug6448152TraceEnabled) + { + TLLM_LOG_INFO( + "NVBUG6448152_DIAG event=global_commit side=gen rank=%d request_id=%llu outcome=failed " + "check_seq=%llu future_count=%zu awaiting_count=%zu timeout_count=%zu", + mWorldRank, static_cast(requestId), + static_cast(generationCheckSequence), mRequesterFutures.size(), + mRequesterRequestsAwaitingConsensus.size(), mTimedOutRequesterIds.size()); + } } for (auto const requestId : consensusOutcome.completedRequestIds) { @@ -1072,6 +1347,15 @@ void CacheTransceiver::checkGenTransferStatus(std::optional const& atLeastR mTimedOutRequesterIds.erase(requestId); eraseLocalTransferOutcome( requestId, mCompletedRequesterRequestIds, mFailedRequesterRequestIds, mRequesterRequestsAwaitingConsensus); + if (mNvbug6448152TraceEnabled) + { + TLLM_LOG_INFO( + "NVBUG6448152_DIAG event=global_commit side=gen rank=%d request_id=%llu outcome=completed " + "check_seq=%llu future_count=%zu awaiting_count=%zu timeout_count=%zu", + mWorldRank, static_cast(requestId), + static_cast(generationCheckSequence), mRequesterFutures.size(), + mRequesterRequestsAwaitingConsensus.size(), mTimedOutRequesterIds.size()); + } } } diff --git a/tensorrt_llm/_torch/pyexecutor/py_executor.py b/tensorrt_llm/_torch/pyexecutor/py_executor.py index 1b624f72d88b..7488fe59568c 100644 --- a/tensorrt_llm/_torch/pyexecutor/py_executor.py +++ b/tensorrt_llm/_torch/pyexecutor/py_executor.py @@ -2,6 +2,7 @@ import datetime import functools import os +import sys import threading import time import traceback @@ -95,6 +96,33 @@ # Default: "0" (only rank 0 prints, matching existing behavior). PROFILE_LOG_RANKS_ENV_VAR_NAME = "TLLM_PROFILE_LOG_RANKS" +# Diagnostic-only lifecycle tracing for NVBUG 6448152. The targeted CI +# configuration enables this explicitly; normal execution remains unchanged. +NVBUG_6448152_TRACE_ENV_VAR_NAME = "TRTLLM_NVBUG_6448152_TRACE" +NVBUG_6448152_CTX_STATUS_CALL_SOURCES = frozenset({ + "_pp_retry_until_can_schedule", + "_executor_loop", + "_executor_loop_pp", + "_handle_executed_batch", + "_prepare_and_schedule_batch", + "_send_kv_async", +}) + + +def _nvbug_6448152_trace_enabled() -> bool: + return os.environ.get(NVBUG_6448152_TRACE_ENV_VAR_NAME, "0") == "1" + + +def _nvbug_6448152_ctx_status_call_source() -> str: + """Find the executor call site above the NVTX decorator wrapper.""" + frame = sys._getframe(2) + while frame is not None: + name = frame.f_code.co_name + if name in NVBUG_6448152_CTX_STATUS_CALL_SOURCES: + return name + frame = frame.f_back + return "unknown" + class PPCommTag(IntEnum): """ @@ -645,6 +673,9 @@ def on_detected(): self.gather_all_responses = False self.kv_cache_transceiver = kv_cache_transceiver + self._nvbug_6448152_trace_enabled = _nvbug_6448152_trace_enabled() + self._nvbug_6448152_ctx_status_call_sequence = 0 + self._nvbug_6448152_ctx_status_signatures = set() self.is_benchmark_disagg = (self.benchmark_req_queues_size > 0 and self.kv_cache_transceiver is not None) # True while the benchmark disagg fill phase is in progress (waiting @@ -799,10 +830,29 @@ def _end_transfer_and_maybe_terminate(self, request: LlmRequest): self._pending_transfer_responses.append( (request.py_request_id, response)) if self.async_transfer_manager.end_transfer(request): + if self._nvbug_6448152_trace_enabled: + logger.warning( + "NVBUG6448152_DIAG event=python_global_commit_observed " + "phase=ctx source=_end_transfer_and_maybe_terminate " + f"rank={self.global_rank} pp_rank={self.dist.pp_rank} " + f"iter={self.iter_counter} req={request.py_request_id} " + "inflight=" + f"{len(self.async_transfer_manager.requests_in_transfer())}" + ) self.active_requests.remove(request) self._terminate_request(request) return if self.async_transfer_manager.end_transfer(request): + if (self._nvbug_6448152_trace_enabled + and request.is_context_only_request): + logger.warning( + "NVBUG6448152_DIAG event=python_global_commit_observed " + "phase=ctx source=_end_transfer_and_maybe_terminate " + f"rank={self.global_rank} pp_rank={self.dist.pp_rank} " + f"iter={self.iter_counter} req={request.py_request_id} " + "inflight=" + f"{len(self.async_transfer_manager.requests_in_transfer())}" + ) # When should_store_blocks is True, _handle_responses already # terminated this request via the early-termination path # (enable_partial_reuse_for_disagg branch). Skip the redundant @@ -4064,6 +4114,16 @@ def flag_if_kv_transfer_timed_out(req: LlmRequest, type: str) -> None: f"Terminating {type} request {req.py_request_id} due to KV cache transfer timeout" ) req.py_kv_transfer_timed_out = True + if self._nvbug_6448152_trace_enabled: + logger.warning( + "NVBUG6448152_DIAG event=python_timeout_flag " + f"phase={type} source=_check_kv_transfer_timeout " + f"rank={self.global_rank} pp_rank={self.dist.pp_rank} " + f"iter={self.iter_counter} req={req.py_request_id} " + f"active={len(self.active_requests)} " + "inflight=" + f"{len(self.async_transfer_manager.requests_in_transfer())}" + ) for req in self.async_transfer_manager.requests_in_transfer().values(): flag_if_kv_transfer_timed_out(req, "context") @@ -4419,6 +4479,15 @@ def kv_connector_request_finished(req: LlmRequest): # Order is important here: we need to start the transfer before responding # to make sure the blocks are stored for reuse before they are sent. self.async_transfer_manager.start_transfer(req) + if self._nvbug_6448152_trace_enabled: + logger.warning( + "NVBUG6448152_DIAG event=python_transfer_start " + "phase=ctx source=_send_kv_async " + f"rank={self.global_rank} pp_rank={self.dist.pp_rank} " + f"iter={self.iter_counter} req={req.py_request_id} " + "inflight=" + f"{len(self.async_transfer_manager.requests_in_transfer())}" + ) self.kv_cache_transceiver.respond_and_send_async(req) if self.kv_cache_transceiver.kv_transfer_timeout_ms is not None: @@ -4454,11 +4523,47 @@ def _check_cache_transfer_errors(self, error_msg_prefix: str): @nvtx_range("_check_disagg_ctx_cache_transfer_status") def _check_disagg_ctx_cache_transfer_status(self, atLeastNum: int = 0): + trace_call = self._nvbug_6448152_trace_enabled + call_started_ns = 0 + call_source = "disabled" + call_sequence = 0 + if trace_call: + self._nvbug_6448152_ctx_status_call_sequence += 1 + call_sequence = self._nvbug_6448152_ctx_status_call_sequence + call_source = _nvbug_6448152_ctx_status_call_source() + inflight = len(self.async_transfer_manager.requests_in_transfer()) + call_signature = (call_source, atLeastNum, inflight) + periodic_checkpoint = (inflight > 0 and call_sequence % 512 == 0) + if (call_signature not in self._nvbug_6448152_ctx_status_signatures + or periodic_checkpoint): + logger.warning( + "NVBUG6448152_DIAG event=python_ctx_status_enter " + "phase=ctx " + f"source={call_source} rank={self.global_rank} " + f"pp_rank={self.dist.pp_rank} iter={self.iter_counter} " + f"call_seq={call_sequence} at_least={atLeastNum} " + f"inflight={inflight} " + f"periodic_checkpoint={int(periodic_checkpoint)}") + self._nvbug_6448152_ctx_status_signatures.add(call_signature) + call_started_ns = time.monotonic_ns() + finished_requests, error_requests = self.kv_cache_transceiver.check_context_transfer_status( atLeastNum) completed_req_ids = set(finished_requests + error_requests) + if trace_call: + duration_us = (time.monotonic_ns() - call_started_ns) // 1000 + if completed_req_ids or duration_us >= 1_000_000: + logger.warning( + "NVBUG6448152_DIAG event=python_ctx_status_exit " + "phase=ctx " + f"source={call_source} rank={self.global_rank} " + f"pp_rank={self.dist.pp_rank} iter={self.iter_counter} " + f"call_seq={call_sequence} duration_us={duration_us} " + f"completed={len(finished_requests)} " + f"failed={len(error_requests)}") + requests_in_transfer = self.async_transfer_manager.requests_in_transfer( ) @@ -4808,6 +4913,16 @@ def _terminate_request(self, request: LlmRequest): # so they must bypass the PP termination handler to avoid stale # sequences in the KV cache manager (the handler delays removal, # but the dummy ID is reused every iteration). + if self._nvbug_6448152_trace_enabled and request.is_context_only_request: + termination_path = ( + "pp_ring" if self._disagg_pp_termination_handler is not None + and not request.is_dummy_request else "immediate") + logger.warning( + "NVBUG6448152_DIAG event=python_terminate_dispatch " + f"phase=ctx source=_terminate_request rank={self.global_rank} " + f"pp_rank={self.dist.pp_rank} iter={self.iter_counter} " + f"req={request.py_request_id} path={termination_path}") + if (self._disagg_pp_termination_handler is not None and not request.is_dummy_request): self._disagg_pp_termination_handler.terminate(request) @@ -4815,7 +4930,30 @@ def _terminate_request(self, request: LlmRequest): self._do_terminate_request(request) def _do_terminate_request(self, request: LlmRequest): + trace_context_free = (self._nvbug_6448152_trace_enabled + and request.is_context_only_request) + if trace_context_free: + logger.warning( + "NVBUG6448152_DIAG event=python_resource_free_enter phase=ctx " + f"source=_do_terminate_request rank={self.global_rank} " + f"pp_rank={self.dist.pp_rank} iter={self.iter_counter} " + f"req={request.py_request_id} " + f"timed_out={int(request.py_kv_transfer_timed_out)} " + "kv_release_path=resource_manager " + "should_store_blocks=" + f"{int(self.async_transfer_manager.should_store_blocks)}") self.resource_manager.free_resources(request) + if trace_context_free: + logger.warning( + "NVBUG6448152_DIAG event=python_resource_free_exit phase=ctx " + f"source=_do_terminate_request rank={self.global_rank} " + f"pp_rank={self.dist.pp_rank} iter={self.iter_counter} " + f"req={request.py_request_id} " + "inflight=" + f"{len(self.async_transfer_manager.requests_in_transfer())} " + "kv_release_path=resource_manager " + "should_store_blocks=" + f"{int(self.async_transfer_manager.should_store_blocks)}") if self.gather_all_responses or self.dist.rank == 0: self.result_wait_queues.pop(request.py_request_id, None) 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 795a6192bf41..86222d552b3c 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 @@ -36,7 +36,7 @@ environment: build_wheel: false work_dir: worker_env_var: TLLM_LOG_LEVEL=INFO TRTLLM_SERVER_DISABLE_GC=1 TRTLLM_WORKER_DISABLE_GC=1 TLLM_SPEC_DECODE_FORCE_NUM_ACCEPTED_TOKENS=1 - TRTLLM_ENABLE_PDL=1 ENROOT_ALLOW_DEV=yes + TRTLLM_ENABLE_PDL=1 ENROOT_ALLOW_DEV=yes TRTLLM_NVBUG_6448152_TRACE=1 server_env_var: TRTLLM_SERVER_DISABLE_GC=1 profiling: nsys_on: false From 231948dc9e094c4517bf3cb77e0048ff646ae171 Mon Sep 17 00:00:00 2001 From: Chien-Chun Hung <2679986+chienchunhung@users.noreply.github.com> Date: Sat, 18 Jul 2026 17:48:42 -0700 Subject: [PATCH 5/5] [NVBUG-6448152][test] isolate CTX consensus regression Signed-off-by: Chien-Chun Hung <2679986+chienchunhung@users.noreply.github.com> --- .../batch_manager/cacheTransceiver.h | 2 + .../batch_manager/cacheTransceiver.cpp | 78 ++++++++++++++++--- ...tx1_pp4_gen1_dep8_eplb0_mtp1_ccb-NIXL.yaml | 2 +- 3 files changed, 69 insertions(+), 13 deletions(-) diff --git a/cpp/include/tensorrt_llm/batch_manager/cacheTransceiver.h b/cpp/include/tensorrt_llm/batch_manager/cacheTransceiver.h index f79d51ae59ae..a6bd15f80347 100644 --- a/cpp/include/tensorrt_llm/batch_manager/cacheTransceiver.h +++ b/cpp/include/tensorrt_llm/batch_manager/cacheTransceiver.h @@ -282,6 +282,8 @@ class CacheTransceiver : public BaseCacheTransceiver // Default-off, transition-only lifecycle tracing for NVBUG#6448152. int mWorldRank; bool mNvbug6448152TraceEnabled; + // Diagnostic-only CTX status behavior for NVBUG#6448152. + bool mNvbug6448152CtxLegacyStatusEnabled; std::uint64_t mNvbug6448152ContextCheckSequence{0}; std::uint64_t mNvbug6448152ContextTpConsensusSequence{0}; std::uint64_t mNvbug6448152ContextPpConsensusSequence{0}; diff --git a/cpp/tensorrt_llm/batch_manager/cacheTransceiver.cpp b/cpp/tensorrt_llm/batch_manager/cacheTransceiver.cpp index 6d3bfa658fdf..687373223764 100644 --- a/cpp/tensorrt_llm/batch_manager/cacheTransceiver.cpp +++ b/cpp/tensorrt_llm/batch_manager/cacheTransceiver.cpp @@ -76,6 +76,11 @@ bool isNvbug6448152TraceEnabled() return common::getBoolEnv("TRTLLM_NVBUG_6448152_TRACE"); } +bool isNvbug6448152CtxLegacyStatusEnabled() +{ + return common::getBoolEnv("TRTLLM_NVBUG_6448152_CTX_LEGACY_STATUS"); +} + enum class TransferConsensusState : std::uint64_t { kCompleted = 1, @@ -286,6 +291,7 @@ CacheTransceiver::CacheTransceiver(kv_cache_manager::BaseKVCacheManager* cacheMa rnn_state_manager::RnnStateManager* rnnStateManager, std::vector const& rnnLayerNumPerPP) : mWorldRank{worldConfig.getRank()} , mNvbug6448152TraceEnabled{isNvbug6448152TraceEnabled()} + , mNvbug6448152CtxLegacyStatusEnabled{isNvbug6448152CtxLegacyStatusEnabled()} , mCacheTransceiverConfig{cacheTransceiverConfig} , mRnnStateManager{rnnStateManager} { @@ -749,6 +755,13 @@ RequestStatuses CacheTransceiver::checkContextTransferStatus( if (mNvbug6448152TraceEnabled) { contextCheckSequence = ++mNvbug6448152ContextCheckSequence; + if (mNvbug6448152CtxLegacyStatusEnabled && contextCheckSequence == 1) + { + TLLM_LOG_WARNING( + "NVBUG6448152_DIAG event=ctx_legacy_status_active side=ctx rank=%d " + "outcome_consensus=disabled gen_status=unchanged", + mWorldRank); + } enqueuedTransitionCount = mNvbug6448152ContextEnqueuedTransitionCount; mNvbug6448152ContextEnqueuedTransitionCount = 0; periodicConsensusCheckpoint @@ -820,9 +833,11 @@ RequestStatuses CacheTransceiver::checkContextTransferStatus( toCompleteIdSet.insert(request->mRequestId); } + RequestStatuses requestsStatus{}; + // Record local terminal outcomes for requests selected this round. The // request is reported only after all ranks in the sync group agree that the - // request reached a terminal state. + // request reached a terminal state, unless the diagnostic legacy mode is enabled. for (auto it = mSenderFutures.begin(); it != mSenderFutures.end();) { auto& [request, future] = *it; @@ -849,7 +864,6 @@ RequestStatuses CacheTransceiver::checkContextTransferStatus( if (status == std::future_status::ready || !senderFutureTimeoutMs.has_value()) { future.get(); - bool const failed = request->getState() == LlmRequestState::kDISAGG_TRANS_ERROR; long elapsedMs = 0; if (mNvbug6448152TraceEnabled) { @@ -857,8 +871,23 @@ RequestStatuses CacheTransceiver::checkContextTransferStatus( LlmRequest::getSteadyClockNow() - request->getKvCacheTransferStart()); elapsedMs = static_cast(elapsed.count()); } - recordLocalTransferOutcome(requestId, request, failed, mCompletedSenderRequestIds, - mFailedSenderRequestIds, mSenderRequestsAwaitingConsensus); + bool const failed = !mNvbug6448152CtxLegacyStatusEnabled + && request->getState() == LlmRequestState::kDISAGG_TRANS_ERROR; + if (mNvbug6448152CtxLegacyStatusEnabled) + { + requestsStatus.completedRequestIds.insert(requestId); + if (markComplete) + { + request->setState(LlmRequestState::kDISAGG_CONTEXT_COMPLETE); + } + mTimedOutSenderIds.erase(requestId); + mNvbug6448152SenderWaitTimeoutIds.erase(requestId); + } + else + { + recordLocalTransferOutcome(requestId, request, failed, mCompletedSenderRequestIds, + mFailedSenderRequestIds, mSenderRequestsAwaitingConsensus); + } it = mSenderFutures.erase(it); if (mNvbug6448152TraceEnabled) { @@ -895,11 +924,22 @@ RequestStatuses CacheTransceiver::checkContextTransferStatus( } else { - TLLM_LOG_ERROR( - "Future returned unexpected status for request %ld. Recording as failed.", requestId); - - recordLocalTransferOutcome(requestId, request, /*failed=*/true, mCompletedSenderRequestIds, - mFailedSenderRequestIds, mSenderRequestsAwaitingConsensus); + if (mNvbug6448152CtxLegacyStatusEnabled) + { + TLLM_LOG_ERROR( + "Future returned unexpected status for request %ld. Marking as error.", requestId); + request->setState(LlmRequestState::kDISAGG_TRANS_ERROR); + requestsStatus.errorRequestIds.insert(requestId); + mTimedOutSenderIds.erase(requestId); + mNvbug6448152SenderWaitTimeoutIds.erase(requestId); + } + else + { + TLLM_LOG_ERROR( + "Future returned unexpected status for request %ld. Recording as failed.", requestId); + recordLocalTransferOutcome(requestId, request, /*failed=*/true, mCompletedSenderRequestIds, + mFailedSenderRequestIds, mSenderRequestsAwaitingConsensus); + } it = mSenderFutures.erase(it); if (mNvbug6448152TraceEnabled) { @@ -918,8 +958,18 @@ RequestStatuses CacheTransceiver::checkContextTransferStatus( catch (std::exception const& e) { TLLM_LOG_ERROR("Error occurred during context transfer for request %ld: %s", requestId, e.what()); - recordLocalTransferOutcome(requestId, request, /*failed=*/true, mCompletedSenderRequestIds, - mFailedSenderRequestIds, mSenderRequestsAwaitingConsensus); + if (mNvbug6448152CtxLegacyStatusEnabled) + { + request->setState(LlmRequestState::kDISAGG_TRANS_ERROR); + requestsStatus.errorRequestIds.insert(requestId); + mTimedOutSenderIds.erase(requestId); + mNvbug6448152SenderWaitTimeoutIds.erase(requestId); + } + else + { + recordLocalTransferOutcome(requestId, request, /*failed=*/true, mCompletedSenderRequestIds, + mFailedSenderRequestIds, mSenderRequestsAwaitingConsensus); + } it = mSenderFutures.erase(it); if (mNvbug6448152TraceEnabled) { @@ -940,7 +990,11 @@ RequestStatuses CacheTransceiver::checkContextTransferStatus( } } - RequestStatuses requestsStatus{}; + if (mNvbug6448152CtxLegacyStatusEnabled) + { + return requestsStatus; + } + TransferConsensusOutcome consensusOutcome; std::uint64_t tpConsensusSequence = 0; std::uint64_t ppConsensusSequence = 0; 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 86222d552b3c..8e281893910e 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 @@ -36,7 +36,7 @@ environment: build_wheel: false work_dir: worker_env_var: TLLM_LOG_LEVEL=INFO TRTLLM_SERVER_DISABLE_GC=1 TRTLLM_WORKER_DISABLE_GC=1 TLLM_SPEC_DECODE_FORCE_NUM_ACCEPTED_TOKENS=1 - TRTLLM_ENABLE_PDL=1 ENROOT_ALLOW_DEV=yes TRTLLM_NVBUG_6448152_TRACE=1 + TRTLLM_ENABLE_PDL=1 ENROOT_ALLOW_DEV=yes TRTLLM_NVBUG_6448152_TRACE=1 TRTLLM_NVBUG_6448152_CTX_LEGACY_STATUS=1 server_env_var: TRTLLM_SERVER_DISABLE_GC=1 profiling: nsys_on: false