[https://nvbugs/6448152][perf] diagnose 2x admission without PP consensus

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 <cstddef>
 #include <future>
 #include <memory>
 #include <mutex>
@@ -54,6 +55,7 @@ class BaseKVCacheManager;
 
 class CacheSender;
 class CacheReceiver;
+class ContextTransferVoteMailbox;
 
 class CacheTransceiverComm
 {
@@ -148,6 +150,25 @@ class CacheTransceiverComm
         TLLM_THROW("Input arguments only supported in mpi");
     }
 
+    [[nodiscard]] std::unique_ptr<mpi::MpiRequest> sendAsync(
+        void const* buffer, std::size_t size, mpi::MpiType dtype, int dest, mpi::MpiTag tag) const
+    {
+        TLLM_CHECK_WITH_INFO(isMpi(), "Point-to-point cache-transceiver status messages require MPI.");
+        return mMpiComm->sendAsync(buffer, size, dtype, dest, tag);
+    }
+
+    [[nodiscard]] bool iprobe(int source, mpi::MpiTag tag, MPI_Status* status) const
+    {
+        TLLM_CHECK_WITH_INFO(isMpi(), "Point-to-point cache-transceiver status messages require MPI.");
+        return mMpiComm->iprobe(source, tag, status);
+    }
+
+    void recv(void* buffer, std::size_t size, mpi::MpiType dtype, int source, mpi::MpiTag tag) const
+    {
+        TLLM_CHECK_WITH_INFO(isMpi(), "Point-to-point cache-transceiver status messages require MPI.");
+        static_cast<void>(mMpiComm->recv(buffer, size, dtype, source, tag));
+    }
+
     CacheTransceiverComm split(int color, int key)
     {
         if (isMpi())
@@ -244,6 +265,7 @@ class CacheTransceiver : public BaseCacheTransceiver
         std::optional<executor::CacheTransceiverConfig> cacheTransceiverConfig = std::nullopt,
         std::vector<SizeType32> const& rnnLayerNumPerPP = {});
 
+    // Constructor used by the PyExecutor Nanobind path. The legacy C++ factory uses the ModelConfig overload above.
     CacheTransceiver(kv_cache_manager::BaseKVCacheManager* cacheManager, std::vector<SizeType32> numKvHeadsPerLayer,
         SizeType32 sizePerHead, SizeType32 tokensPerBlock, runtime::WorldConfig const& worldConfig,
         std::vector<SizeType32> const& attentionLayerNumPerPP, nvinfer1::DataType dataType,
@@ -253,7 +275,8 @@ class CacheTransceiver : public BaseCacheTransceiver
         std::vector<SizeType32> const& rnnLayerNumPerPP = {})
         : CacheTransceiver(cacheManager,
             executor::kv_cache::CacheState::ModelConfig{numKvHeadsPerLayer, sizePerHead, tokensPerBlock}, worldConfig,
-            attentionLayerNumPerPP, dataType, attentionType, cacheTransceiverConfig, rnnLayerNumPerPP)
+            attentionLayerNumPerPP, dataType, attentionType, cacheTransceiverConfig, rnnLayerNumPerPP,
+            /*enableWorkerPublishedContextConsensus=*/true)
     {
     }
 
@@ -279,6 +302,13 @@ class CacheTransceiver : public BaseCacheTransceiver
     [[nodiscard]] bool hasPoisonedTransferBuffer() const override;
 
 private:
+    CacheTransceiver(kv_cache_manager::BaseKVCacheManager* cacheManager,
+        executor::kv_cache::CacheState::ModelConfig const& cacheStateModelCfg, runtime::WorldConfig const& worldConfig,
+        std::vector<SizeType32> const& attentionLayerNumPerPP, nvinfer1::DataType dataType,
+        executor::kv_cache::CacheState::AttentionType attentionType,
+        std::optional<executor::CacheTransceiverConfig> cacheTransceiverConfig,
+        std::vector<SizeType32> const& rnnLayerNumPerPP, bool enableWorkerPublishedContextConsensus);
+
     void initializeCommState();
 
     void setContextState(LlmRequest* llmRequest);
@@ -299,13 +329,20 @@ class CacheTransceiver : public BaseCacheTransceiver
     std::unordered_set<LlmRequest::RequestIdType> mCompletedSenderRequestIds;
     std::unordered_set<LlmRequest::RequestIdType> mFailedSenderRequestIds;
     std::unordered_map<LlmRequest::RequestIdType, std::shared_ptr<LlmRequest>> mSenderRequestsAwaitingConsensus;
+    // Diagnostic-only tombstones for requests reported locally before PP consensus commits.
+    std::unordered_set<LlmRequest::RequestIdType> mDiagnosticLocallyReportedSenderRequestIds;
     std::unordered_set<LlmRequest::RequestIdType> mCompletedRequesterRequestIds;
     std::unordered_set<LlmRequest::RequestIdType> mFailedRequesterRequestIds;
     std::unordered_map<LlmRequest::RequestIdType, std::shared_ptr<LlmRequest>> mRequesterRequestsAwaitingConsensus;
     mpi::MpiComm const* mMpiWorldComm{nullptr};
 
     std::shared_ptr<CacheTransceiverComm> mGroupComm;
     std::shared_ptr<CacheTransceiverComm> mGroupTensorParaComm, mGroupPipeParaComm, mGroupDataComm, mGroupTPInDPComm;
+    std::unique_ptr<ContextTransferVoteMailbox> mContextTransferVoteMailbox;
+    bool mDiagnosticEarlyLocalContextCompletion{false};
+    bool mDiagnosticNoContextConsensusTraffic{false};
+    bool mDiagnosticObservedConsensusGap{false};
+    bool mDiagnosticObservedNoConsensusCompletion{false};
 
     executor::kv_cache::CommState const* mCommState;
     std::unique_ptr<executor::kv_cache::CacheState> mCacheState;

cpp/include/tensorrt_llm/batch_manager/transferStatusConsensus.h

@@ -0,0 +1,120 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <unordered_map>
+#include <unordered_set>
+#include <vector>
+
+namespace tensorrt_llm::batch_manager
+{
+
+class CacheTransceiverComm;
+
+struct WorkerPublishedConsensusConfig
+{
+    bool enabledForCppBinding{false};
+    bool mpiControlPlane{false};
+    bool nixlUcxBackend{false};
+    int tensorParallelism{1};
+    int contextParallelism{1};
+    int pipelineParallelism{1};
+    bool attentionDp{false};
+    bool inflightCancellation{false};
+    bool transferOverlap{true};
+    bool layerwiseTransfer{false};
+    bool mpiThreadMultiple{false};
+};
+
+//! Return whether the worker-published protocol is supported by one rank's effective configuration.
+[[nodiscard]] bool supportsWorkerPublishedConsensus(WorkerPublishedConsensusConfig const& config);
+
+//! Return whether every rank advertised the same nonzero protocol version.
+[[nodiscard]] bool selectWorkerPublishedConsensus(std::vector<std::uint64_t> const& protocolVersions);
+
+enum class TransferStatusVote : std::uint64_t
+{
+    kCompleted = 1,
+    kFailed = 2,
+};
+
+struct TransferStatusConsensusResult
+{
+    std::unordered_set<std::uint64_t> completedRequestIds;
+    std::unordered_set<std::uint64_t> failedRequestIds;
+};
+
+//! Reduces one immutable terminal vote per participant and request.
+class TransferStatusConsensus
+{
+public:
+    explicit TransferStatusConsensus(int participantCount);
+
+    //! Record a participant's terminal vote. Identical duplicates are ignored; conflicting votes are rejected.
+    void recordVote(int participantRank, std::uint64_t requestId, TransferStatusVote vote);
+
+    //! Return and remove every request for which all participants have cast a terminal vote.
+    [[nodiscard]] TransferStatusConsensusResult takeCompleted();
+
+private:
+    struct RequestVotes
+    {
+        explicit RequestVotes(int participantCount)
+            : votes(static_cast<std::size_t>(participantCount), 0)
+        {
+        }
+
+        std::vector<std::uint64_t> votes;
+        int terminalCount{0};
+        bool failed{false};
+    };
+
+    int mParticipantCount;
+    std::unordered_map<std::uint64_t, RequestVotes> mRequestVotes;
+};
+
+//! Exchanges immutable worker-published terminal votes without requiring peer scheduler participation.
+class ContextTransferVoteMailbox
+{
+public:
+    explicit ContextTransferVoteMailbox(std::shared_ptr<CacheTransceiverComm> comm);
+    ~ContextTransferVoteMailbox();
+
+    ContextTransferVoteMailbox(ContextTransferVoteMailbox const&) = delete;
+    ContextTransferVoteMailbox& operator=(ContextTransferVoteMailbox const&) = delete;
+
+    //! Publish one terminal outcome to every peer. The first internal error is reported by a later scheduler poll.
+    void publishOutcomeToPeers(std::uint64_t requestId, bool failed) noexcept;
+
+    //! Record the already-published local worker outcome in this rank's reducer.
+    [[nodiscard]] bool recordLocalOutcome(std::uint64_t requestId);
+
+    //! Drain peer votes and return newly committed global outcomes.
+    [[nodiscard]] TransferStatusConsensusResult poll();
+
+    //! Stop accepting publications and exchange an ordered close marker with every peer.
+    void shutdown() noexcept;
+
+private:
+    class Impl;
+    std::unique_ptr<Impl> mImpl;
+};
+
+} // namespace tensorrt_llm::batch_manager

cpp/include/tensorrt_llm/runtime/utils/mpiTags.h

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2021-2023, NVIDIA CORPORATION.  All rights reserved.
+ * Copyright (c) 2021-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.
@@ -71,7 +71,10 @@ enum class MpiTag : int
 
     // KvCacheEventManager
     kKvCacheEventSize = 1026,
-    kKvCacheEvent = 1027
+    kKvCacheEvent = 1027,
+
+    // Context-transfer terminal votes.
+    kContextTransferStatus = 1028
 };
 
 } // namespace tensorrt_llm::mpi

cpp/include/tensorrt_llm/runtime/utils/mpiUtils.h

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2021-2023, NVIDIA CORPORATION.  All rights reserved.
+ * Copyright (c) 2021-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.
@@ -235,6 +235,17 @@ class MpiRequest
 #endif
     }
 
+    [[nodiscard]] bool isCompleted()
+    {
+#if ENABLE_MULTI_DEVICE
+        int completed = 0;
+        TLLM_MPI_CHECK(MPI_Test(&mRequest, &completed, MPI_STATUS_IGNORE));
+        return completed != 0;
+#else
+        TLLM_THROW("Multi device support is disabled.");
+#endif
+    }
+
     void cancel()
     {
 #if ENABLE_MULTI_DEVICE
@@ -467,6 +478,9 @@ int getNumNodes();
 
 void initialize(MpiThreadSupport threadMode = MpiThreadSupport::THREAD_MULTIPLE, bool forwardAbortToParent = false);
 
+//! \brief Return the MPI thread support level provided by the active runtime.
+[[nodiscard]] MpiThreadSupport getThreadSupport();
+
 //! \brief Returns true iff the WideEP fault-tolerance MPI mode is enabled.
 //!
 //! Reads the `TLLM_FAULT_TOLERANCE_MODE` environment variable and returns

cpp/tensorrt_llm/batch_manager/CMakeLists.txt

@@ -1,4 +1,4 @@
-# SPDX-FileCopyrightText: Copyright (c) 2023-2025 NVIDIA CORPORATION &
+# SPDX-FileCopyrightText: Copyright (c) 2023-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
@@ -56,6 +56,7 @@ set(SRCS
     rnnCacheTransBuffer.cpp
     runtimeBuffers.cpp
     sequenceSlotManager.cpp
+    transferStatusConsensus.cpp
     transformerBuffers.cpp
     trtEncoderModel.cpp
     trtGptModelInflightBatching.cpp