Rename data -> cache

Tabrizian · Tabrizian · commit abb41e4b6443 · 2025-08-10T16:53:55.000-07:00
Signed-off-by: Iman Tabrizian &lt;10105175+tabrizian@users.noreply.github.com&gt;
diff --git a/cpp/include/tensorrt_llm/batch_manager/cacheTransceiver.h b/cpp/include/tensorrt_llm/batch_manager/cacheTransceiver.h
@@ -110,9 +110,9 @@ class CacheTransceiver : public BaseCacheTransceiver
 
     void setContextState(LlmRequest* llmRequest);
 
-    std::unique_ptr<DataResponder> mDataResponder;
+    std::unique_ptr<DataResponder> mCacheSender;
     std::unique_ptr<DataRequester> mDataRequester;
-    std::vector<std::pair<LlmRequest*, std::future<void>>> mResponderFutures;
+    std::vector<std::pair<LlmRequest*, std::future<void>>> mSenderFutures;
     std::vector<std::pair<LlmRequest*, std::future<void>>> mRequesterFutures;
     mpi::MpiComm const *mMpiGroupComm{nullptr}, *mMpiWorldComm{nullptr};
     std::shared_ptr<mpi::MpiComm> mMpiGroupTensorParaComm, mMpiGroupPipeParaComm, mMpiGroupDataComm,
diff --git a/cpp/tensorrt_llm/batch_manager/cacheTransceiver.cpp b/cpp/tensorrt_llm/batch_manager/cacheTransceiver.cpp
@@ -195,10 +195,10 @@ CacheTransceiver::CacheTransceiver(kv_cache_manager::BaseKVCacheManager* cacheMa
     auto makeFormatter = [cacheManager, isMLA, this]()
     { return createCacheFormatter(cacheManager, mCacheTransBufferManager.get(), isMLA); };
 
-    mDataResponder = std::make_unique<DataResponder>(
-        std::make_unique<DataSenderImpl>(mManager.get(), *mCacheState, worldConfig.getRank(), makeFormatter()));
+    mCacheSender = std::make_unique<DataResponder>(
+        std::make_unique<CacheSenderImpl>(mManager.get(), *mCacheState, worldConfig.getRank(), makeFormatter()));
     mDataRequester = std::make_unique<DataRequester>(
-        std::make_unique<DataReceiverImpl>(mManager.get(), *mCacheState, worldConfig.getRank(), makeFormatter()));
+        std::make_unique<CacheReceiverImpl>(mManager.get(), *mCacheState, worldConfig.getRank(), makeFormatter()));
 
     initializeCommState();
 }
@@ -214,7 +214,7 @@ CacheTransceiver::~CacheTransceiver()
 
 void CacheTransceiver::initializeCommState()
 {
-    mCommState = std::addressof(mDataResponder->getCommState());
+    mCommState = std::addressof(mCacheSender->getCommState());
 }
 
 void CacheTransceiver::setContextState(LlmRequest* llmRequest)
@@ -250,8 +250,8 @@ void CacheTransceiver::respondAndSendAsync(LlmRequest* llmRequest)
         return;
     }
     setContextState(llmRequest);
-    auto future = mDataResponder->respondAndSendAsync(*llmRequest);
-    mResponderFutures.emplace_back(llmRequest, std::move(future));
+    auto future = mCacheSender->respondAndSendAsync(*llmRequest);
+    mSenderFutures.emplace_back(llmRequest, std::move(future));
 }
 
 void CacheTransceiver::respondAndSendLayerWise(
@@ -266,8 +266,8 @@ void CacheTransceiver::respondAndSendLayerWise(
 
         llmRequest->setState(LlmRequestState::kDISAGG_CONTEXT_INIT_AND_TRANS);
         setContextState(llmRequest.get());
-        auto future = mDataResponder->respondAndSendAsync(*llmRequest);
-        mResponderFutures.emplace_back(llmRequest.get(), std::move(future));
+        auto future = mCacheSender->respondAndSendAsync(*llmRequest);
+        mSenderFutures.emplace_back(llmRequest.get(), std::move(future));
     }
 }
 
@@ -373,7 +373,7 @@ void CacheTransceiver::checkContextTransferStatus(std::optional<int> const& atLe
     bool blockAll = !atLeastRequestNum.has_value();
     auto syncComm = mCacheState->getParallelConfig().mEnableAttentionDP ? mMpiGroupTPInDPComm : mMpiGroupTensorParaComm;
     std::vector<LlmRequest::RequestIdType> contextCompleteRequestIds;
-    for (auto&& [request, future] : mResponderFutures)
+    for (auto&& [request, future] : mSenderFutures)
     {
         if (future.wait_for(std::chrono::milliseconds(0)) == std::future_status::ready)
         {
@@ -413,23 +413,22 @@ void CacheTransceiver::checkContextTransferStatus(std::optional<int> const& atLe
 
     // Make sure there are at least atLeastRequestNum requests in toCompleteIdSet.
     // This will preserve the order of insertion for KVCache transfer requests.
-    for (auto it = mResponderFutures.begin();
-         atLeastRequestNum.value_or(0) > static_cast<int>(toCompleteIdSet.size()) && it != mResponderFutures.end();
-         ++it)
+    for (auto it = mSenderFutures.begin();
+         atLeastRequestNum.value_or(0) > static_cast<int>(toCompleteIdSet.size()) && it != mSenderFutures.end(); ++it)
     {
         auto& [request, future] = *it;
         toCompleteIdSet.insert(request->mRequestId);
     }
 
     // Complete all the requests in toCompleteIdSet
-    for (auto it = mResponderFutures.begin(); it != mResponderFutures.end();)
+    for (auto it = mSenderFutures.begin(); it != mSenderFutures.end();)
     {
         auto& [request, future] = *it;
         if (blockAll || (toCompleteIdSet.find(request->mRequestId) != toCompleteIdSet.end()))
         {
             future.get();
             request->setState(LlmRequestState::kDISAGG_CONTEXT_COMPLETE);
-            it = mResponderFutures.erase(it);
+            it = mSenderFutures.erase(it);
         }
         else
         {
diff --git a/cpp/tensorrt_llm/batch_manager/dataTransceiver.cpp b/cpp/tensorrt_llm/batch_manager/dataTransceiver.cpp
@@ -148,14 +148,14 @@ class DataResponder::Impl
         auto future = promise.get_future();
         {
             {
-                std::unique_lock lkResp(mResponderMutex);
+                std::unique_lock lkResp(mSenderMutex);
                 mReadyResponses.emplace(
                     llmRequest.mRequestId, Response{std::addressof(llmRequest), std::move(promise)});
             }
             std::unique_lock lkCond(mCondMutex);
             mAnyReady = true;
         }
-        mResponderCv.notify_all();
+        mSenderCv.notify_all();
         return future;
     }
 
@@ -208,7 +208,7 @@ class DataResponder::Impl
                 if (!mAnyReady)
                 {
                     std::unique_lock lk(mCondMutex);
-                    mResponderCv.wait(lk, [this]() { return (mAnyReady || mTerminate); });
+                    mSenderCv.wait(lk, [this]() { return (mAnyReady || mTerminate); });
                 }
                 if (mTerminate)
                 {
@@ -263,7 +263,7 @@ class DataResponder::Impl
                         "mReadyResponses size is: %zu. mpi rank :%d     ",
                         mCurrentRequest.value(), mReadyResponses.size(), mpi::MpiComm::world().getRank());
                     std::unique_lock lk(mCondMutex);
-                    mResponderCv.wait(lk, [this]() { return (mAnyReady || mTerminate); });
+                    mSenderCv.wait(lk, [this]() { return (mAnyReady || mTerminate); });
                 }
             }
         }
@@ -285,13 +285,13 @@ class DataResponder::Impl
         }
         // We don't have to wait for the future. If another thread is sending data, it won't pay attention
         // to the terminate flag.
-        mResponderCv.notify_all();
+        mSenderCv.notify_all();
     }
 
     void removeResponse(std::map<RequestIdType, Response>::iterator it)
     {
         {
-            std::unique_lock lkResp(mResponderMutex);
+            std::unique_lock lkResp(mSenderMutex);
             mReadyResponses.erase(it);
         }
         if (mReadyResponses.empty())
@@ -313,16 +313,16 @@ class DataResponder::Impl
 
     [[nodiscard]] std::map<RequestIdType, Response>::iterator getCurrentResponse()
     {
-        std::unique_lock lk(mResponderMutex);
+        std::unique_lock lk(mSenderMutex);
         return mReadyResponses.find(getCurrentRequestId());
     }
 
 private:
     std::optional<RequestIdType> mCurrentRequest;
     std::map<RequestIdType, Response> mReadyResponses;
-    std::mutex mResponderMutex, mCondMutex;
+    std::mutex mSenderMutex, mCondMutex;
     std::atomic<bool> mAnyReady{false}, mTerminate{false};
-    std::condition_variable mResponderCv;
+    std::condition_variable mSenderCv;
     std::future<void> mResponseFuture;
     std::unique_ptr<DataSender> mSender;
     std::unordered_map<LlmRequest::RequestIdType, int> mRemainSendCount;
@@ -333,9 +333,9 @@ class DataRequester::Impl
 {
 public:
     Impl(std::unique_ptr<DataReceiver> receiver)
-        : mReceiver{std::move(receiver)}
+        : mCacheReceiver{std::move(receiver)}
     {
-        TLLM_CHECK(mReceiver);
+        TLLM_CHECK(mCacheReceiver);
         TLLM_CUDA_CHECK(cudaGetDevice(&mDeviceId));
     }
 
@@ -400,8 +400,8 @@ class DataRequester::Impl
             llmRequest.getContextPhaseParams().value().getReqId());
         llmRequest.setKvCacheTransferStart(std::chrono::steady_clock::now());
         TLLM_CUDA_CHECK(cudaSetDevice(mDeviceId));
-        auto session = mReceiver->sendRequestInfo(llmRequest);
-        mReceiver->receiveSync(session);
+        auto session = mCacheReceiver->sendRequestInfo(llmRequest);
+        mCacheReceiver->receiveSync(session);
         llmRequest.setKvCacheTransferEnd(std::chrono::steady_clock::now());
 
         TLLM_LOG_DEBUG(mpi::MpiComm::world().getRank(),
@@ -507,7 +507,7 @@ class DataRequester::Impl
         }
     }
 
-    std::unique_ptr<DataReceiver> mReceiver;
+    std::unique_ptr<DataReceiver> mCacheReceiver;
     int mDeviceId{-1};
 
     std::vector<std::future<void>> mRequestFutures;
diff --git a/cpp/tensorrt_llm/batch_manager/dataTransceiverImpl.cpp b/cpp/tensorrt_llm/batch_manager/dataTransceiverImpl.cpp
@@ -47,7 +47,7 @@ static fs::path getTransferOutputPath(char const* tag)
     return {};
 }
 
-DataSenderImpl::DataSenderImpl(executor::kv_cache::ConnectionManager* manager,
+CacheSenderImpl::CacheSenderImpl(executor::kv_cache::ConnectionManager* manager,
     executor::kv_cache::CacheState selfCacheState, SizeType32 selfIndex, std::unique_ptr<BaseCacheFormatter> formatter)
     : mManager{manager}
     , mSelfState{std::move(selfCacheState), executor::kv_cache::CommState{manager->getCommState()}}
@@ -58,7 +58,7 @@ DataSenderImpl::DataSenderImpl(executor::kv_cache::ConnectionManager* manager,
     TLLM_CHECK(mManager->getCommState().getSelfIdx() == selfIndex);
 }
 
-[[nodiscard]] RequestInfo DataSenderImpl::recvRequestInfo()
+[[nodiscard]] RequestInfo CacheSenderImpl::recvRequestInfo()
 {
     using DataContext = tensorrt_llm::executor::kv_cache::DataContext;
     auto* agentConnectionManager = dynamic_cast<executor::kv_cache::AgentConnectionManager*>(mManager);
@@ -111,7 +111,7 @@ DataSenderImpl::DataSenderImpl(executor::kv_cache::ConnectionManager* manager,
     return info;
 }
 
-void DataSenderImpl::sendSync(LlmRequest const& llmRequest)
+void CacheSenderImpl::sendSync(LlmRequest const& llmRequest)
 {
     auto it = mRequestToSession.find(llmRequest.mRequestId);
     TLLM_CHECK(it != mRequestToSession.end());
@@ -120,24 +120,24 @@ void DataSenderImpl::sendSync(LlmRequest const& llmRequest)
     mFormatter->format(session);
 }
 
-[[nodiscard]] executor::kv_cache::CommState const& DataSenderImpl::getCommState() const
+[[nodiscard]] executor::kv_cache::CommState const& CacheSenderImpl::getCommState() const
 {
     return mSelfState.getCommState().value();
 }
 
-void DataSenderImpl::setCommState(executor::kv_cache::CommState commState)
+void CacheSenderImpl::setCommState(executor::kv_cache::CommState commState)
 {
     mSelfState.setCommState(std::move(commState));
 }
 
-[[nodiscard]] size_t DataSenderImpl::getCounterpartsCount(LlmRequest::RequestIdType requestId) const
+[[nodiscard]] size_t CacheSenderImpl::getCounterpartsCount(LlmRequest::RequestIdType requestId) const
 {
     auto it = mRequestToSession.find(requestId);
     TLLM_CHECK(it != mRequestToSession.end());
     return it->second.getConnections().size();
 }
 
-void DataSenderImpl::release(LlmRequest::RequestIdType requestId)
+void CacheSenderImpl::release(LlmRequest::RequestIdType requestId)
 {
     auto it = mRequestToSession.find(requestId);
     TLLM_CHECK(it != mRequestToSession.end());
@@ -156,7 +156,7 @@ void DataSenderImpl::release(LlmRequest::RequestIdType requestId)
     mRequestToSession.erase(it);
 }
 
-DataReceiverImpl::DataReceiverImpl(executor::kv_cache::ConnectionManager* manager,
+CacheReceiverImpl::CacheReceiverImpl(executor::kv_cache::ConnectionManager* manager,
     executor::kv_cache::CacheState selfCacheState, SizeType32 selfIndex, std::unique_ptr<BaseCacheFormatter> formatter)
     : mManager{manager}
     , mSelfState{std::move(selfCacheState), executor::kv_cache::CommState{manager->getCommState()}}
@@ -167,7 +167,7 @@ DataReceiverImpl::DataReceiverImpl(executor::kv_cache::ConnectionManager* manage
     TLLM_CHECK(mFormatter);
 }
 
-TransferSession DataReceiverImpl::sendRequestInfo(LlmRequest const& llmRequest)
+TransferSession CacheReceiverImpl::sendRequestInfo(LlmRequest const& llmRequest)
 {
     uint64_t requestId = llmRequest.getContextPhaseParams().value().getReqId();
     auto const& contextState = llmRequest.getDataTransceiverState();
@@ -233,7 +233,7 @@ TransferSession DataReceiverImpl::sendRequestInfo(LlmRequest const& llmRequest)
         contextState, resource->mBufferManager, &llmRequest, !common::getEnvKVCacheTransferOutputPath().empty());
 }
 
-void DataReceiverImpl::receiveSync(TransferSession& session)
+void CacheReceiverImpl::receiveSync(TransferSession& session)
 {
     mFormatter->unformat(session);
     if (!common::getEnvKVCacheTransferOutputPath().empty())
@@ -250,7 +250,7 @@ void DataReceiverImpl::receiveSync(TransferSession& session)
     }
 }
 
-void DataReceiverImpl::sendRequestInfo(executor::kv_cache::Connection const* connection, RequestInfo const& info)
+void CacheReceiverImpl::sendRequestInfo(executor::kv_cache::Connection const* connection, RequestInfo const& info)
 {
     std::ostringstream oss;
     RequestInfo::serialize(info, oss);
@@ -262,7 +262,7 @@ void DataReceiverImpl::sendRequestInfo(executor::kv_cache::Connection const* con
     connection->send(executor::kv_cache::DataContext{kINFO_TAG}, serializedInfo.data(), infoSize);
 }
 
-std::unique_ptr<DataReceiverImpl::ReceiveCacheResource> const& DataReceiverImpl::getReceiveCacheResource(
+std::unique_ptr<CacheReceiverImpl::ReceiveCacheResource> const& CacheReceiverImpl::getReceiveCacheResource(
     LlmRequest const& llmRequest)
 {
     std::scoped_lock<std::mutex> lock(mProcessIoResouceMutex);
diff --git a/cpp/tensorrt_llm/batch_manager/dataTransceiverImpl.h b/cpp/tensorrt_llm/batch_manager/dataTransceiverImpl.h
@@ -42,12 +42,12 @@ struct TransceiverTag
 
 using BaseCacheFormatter = kv_cache_manager::BaseCacheFormatter;
 
-class DataSenderImpl : public DataSender, public TransceiverTag
+class CacheSenderImpl : public DataSender, public TransceiverTag
 {
 public:
     using SizeType32 = tensorrt_llm::runtime::SizeType32;
 
-    DataSenderImpl(executor::kv_cache::ConnectionManager* manager, executor::kv_cache::CacheState selfCacheState,
+    CacheSenderImpl(executor::kv_cache::ConnectionManager* manager, executor::kv_cache::CacheState selfCacheState,
         SizeType32 selfIndex, std::unique_ptr<BaseCacheFormatter> formatter);
 
     [[nodiscard]] RequestInfo recvRequestInfo() override;
@@ -72,12 +72,12 @@ class DataSenderImpl : public DataSender, public TransceiverTag
     std::ofstream mMeasuresFile;
 };
 
-class DataReceiverImpl : public DataReceiver, public TransceiverTag
+class CacheReceiverImpl : public DataReceiver, public TransceiverTag
 {
 public:
     using SizeType32 = tensorrt_llm::runtime::SizeType32;
 
-    DataReceiverImpl(executor::kv_cache::ConnectionManager* manager, executor::kv_cache::CacheState selfCacheState,
+    CacheReceiverImpl(executor::kv_cache::ConnectionManager* manager, executor::kv_cache::CacheState selfCacheState,
         SizeType32 selfIndex, std::unique_ptr<BaseCacheFormatter> formatter);
 
     TransferSession sendRequestInfo(LlmRequest const& llmRequest) override;
diff --git a/cpp/tensorrt_llm/executor/cache_transmission/agent_utils/connection.cpp b/cpp/tensorrt_llm/executor/cache_transmission/agent_utils/connection.cpp
@@ -81,7 +81,7 @@ void AgentConnection::send(DataContext const& ctx, void const* data, size_t size
     MemoryDesc srcDesc{
         reinterpret_cast<uintptr_t>(data), size, static_cast<uint32_t>(mAgentConnectionManager->getDeviceId())};
     MemoryDescs srcDescs{MemoryType::kVRAM, {srcDesc}};
-    auto dstBaseDesc = mSenderState.mReceiverBufferDesc;
+    auto dstBaseDesc = mSenderState.mCacheReceiverBufferDesc;
     MemoryDesc dstDesc{dstBaseDesc.getAddr() + (mSenderState.validSegmentIdx * size), size, dstBaseDesc.getDeviceId()};
     TLLM_LOG_DEBUG(
         "send dstDesc: %p, size: %ld ,validSegmentIdx: %ld", dstDesc.getAddr(), size, mSenderState.validSegmentIdx);
@@ -137,9 +137,9 @@ void AgentConnection::sendRequestAndBufferInfo(
     mAgentConnectionManager->getAgent()->notifySyncMessage(mRemoteAgentName, ss.str());
 }
 
-void AgentConnection::setSenderState(MemoryDesc mReceiverBufferDesc, int validSegmentIdx)
+void AgentConnection::setSenderState(MemoryDesc mCacheReceiverBufferDesc, int validSegmentIdx)
 {
-    mSenderState.mReceiverBufferDesc = mReceiverBufferDesc;
+    mSenderState.mCacheReceiverBufferDesc = mCacheReceiverBufferDesc;
     mSenderState.validSegmentIdx = validSegmentIdx;
 }
 
diff --git a/cpp/tensorrt_llm/executor/cache_transmission/agent_utils/connection.h b/cpp/tensorrt_llm/executor/cache_transmission/agent_utils/connection.h
@@ -175,7 +175,7 @@ class AgentConnection : public Connection
     void recv(DataContext const& ctx, void* data, size_t size) const override;
     void sendRequestAndBufferInfo(
         batch_manager::RequestInfo& requestInfo, std::optional<size_t> cacheBufferId, int validConnectionIdx);
-    void setSenderState(MemoryDesc mReceiverBufferDesc, int valideSegmentIdx);
+    void setSenderState(MemoryDesc mCacheReceiverBufferDesc, int valideSegmentIdx);
     [[nodiscard]] std::optional<size_t> getCacheBufferId() const;
     void setHasLoadRemoteAgent(bool hasLoadRemoteAgent);
     [[nodiscard]] bool hasLoadRemoteAgent() const;
@@ -186,7 +186,7 @@ class AgentConnection : public Connection
 
     struct SenderState
     {
-        MemoryDesc mReceiverBufferDesc{nullptr, 0, 0};
+        MemoryDesc mCacheReceiverBufferDesc{nullptr, 0, 0};
         int validSegmentIdx{0};
         SenderState() = default;
     };
diff --git a/cpp/tests/batch_manager/cacheTransceiverTest.cpp b/cpp/tests/batch_manager/cacheTransceiverTest.cpp

Original file line number	Diff line number	Diff line change
`@@ -195,10 +195,10 @@ CacheTransceiver::CacheTransceiver(kv_cache_manager::BaseKVCacheManager* cacheMa`
`195`	`195`	`auto makeFormatter = [cacheManager, isMLA, this]()`
`196`	`196`	`{ return createCacheFormatter(cacheManager, mCacheTransBufferManager.get(), isMLA); };`
`197`	`197`
`198`		`- mDataResponder = std::make_unique<DataResponder>(`
`199`		`- std::make_unique<DataSenderImpl>(mManager.get(), *mCacheState, worldConfig.getRank(), makeFormatter()));`
	`198`	`+ mCacheSender = std::make_unique<DataResponder>(`
	`199`	`+ std::make_unique<CacheSenderImpl>(mManager.get(), *mCacheState, worldConfig.getRank(), makeFormatter()));`
`200`	`200`	`mDataRequester = std::make_unique<DataRequester>(`
`201`		`- std::make_unique<DataReceiverImpl>(mManager.get(), *mCacheState, worldConfig.getRank(), makeFormatter()));`
	`201`	`+ std::make_unique<CacheReceiverImpl>(mManager.get(), *mCacheState, worldConfig.getRank(), makeFormatter()));`
`202`	`202`
`203`	`203`	`initializeCommState();`
`204`	`204`	`}`
`@@ -214,7 +214,7 @@ CacheTransceiver::~CacheTransceiver()`
`214`	`214`
`215`	`215`	`void CacheTransceiver::initializeCommState()`
`216`	`216`	`{`
`217`		`- mCommState = std::addressof(mDataResponder->getCommState());`
	`217`	`+ mCommState = std::addressof(mCacheSender->getCommState());`
`218`	`218`	`}`
`219`	`219`
`220`	`220`	`void CacheTransceiver::setContextState(LlmRequest* llmRequest)`
`@@ -250,8 +250,8 @@ void CacheTransceiver::respondAndSendAsync(LlmRequest* llmRequest)`
`250`	`250`	`return;`
`251`	`251`	`}`
`252`	`252`	`setContextState(llmRequest);`
`253`		`- auto future = mDataResponder->respondAndSendAsync(*llmRequest);`
`254`		`- mResponderFutures.emplace_back(llmRequest, std::move(future));`
	`253`	`+ auto future = mCacheSender->respondAndSendAsync(*llmRequest);`
	`254`	`+ mSenderFutures.emplace_back(llmRequest, std::move(future));`
`255`	`255`	`}`
`256`	`256`
`257`	`257`	`void CacheTransceiver::respondAndSendLayerWise(`
`@@ -266,8 +266,8 @@ void CacheTransceiver::respondAndSendLayerWise(`
`266`	`266`
`267`	`267`	`llmRequest->setState(LlmRequestState::kDISAGG_CONTEXT_INIT_AND_TRANS);`
`268`	`268`	`setContextState(llmRequest.get());`
`269`		`- auto future = mDataResponder->respondAndSendAsync(*llmRequest);`
`270`		`- mResponderFutures.emplace_back(llmRequest.get(), std::move(future));`
	`269`	`+ auto future = mCacheSender->respondAndSendAsync(*llmRequest);`
	`270`	`+ mSenderFutures.emplace_back(llmRequest.get(), std::move(future));`
`271`	`271`	`}`
`272`	`272`	`}`
`273`	`273`
`@@ -373,7 +373,7 @@ void CacheTransceiver::checkContextTransferStatus(std::optional<int> const& atLe`
`373`	`373`	`bool blockAll = !atLeastRequestNum.has_value();`
`374`	`374`	`auto syncComm = mCacheState->getParallelConfig().mEnableAttentionDP ? mMpiGroupTPInDPComm : mMpiGroupTensorParaComm;`
`375`	`375`	`std::vector<LlmRequest::RequestIdType> contextCompleteRequestIds;`
`376`		`- for (auto&& [request, future] : mResponderFutures)`
	`376`	`+ for (auto&& [request, future] : mSenderFutures)`
`377`	`377`	`{`
`378`	`378`	`if (future.wait_for(std::chrono::milliseconds(0)) == std::future_status::ready)`
`379`	`379`	`{`
`@@ -413,23 +413,22 @@ void CacheTransceiver::checkContextTransferStatus(std::optional<int> const& atLe`
`413`	`413`
`414`	`414`	`// Make sure there are at least atLeastRequestNum requests in toCompleteIdSet.`
`415`	`415`	`// This will preserve the order of insertion for KVCache transfer requests.`
`416`		`- for (auto it = mResponderFutures.begin();`
`417`		`- atLeastRequestNum.value_or(0) > static_cast<int>(toCompleteIdSet.size()) && it != mResponderFutures.end();`
`418`		`- ++it)`
	`416`	`+ for (auto it = mSenderFutures.begin();`
	`417`	`+ atLeastRequestNum.value_or(0) > static_cast<int>(toCompleteIdSet.size()) && it != mSenderFutures.end(); ++it)`
`419`	`418`	`{`
`420`	`419`	`auto& [request, future] = *it;`
`421`	`420`	`toCompleteIdSet.insert(request->mRequestId);`
`422`	`421`	`}`
`423`	`422`
`424`	`423`	`// Complete all the requests in toCompleteIdSet`
`425`		`- for (auto it = mResponderFutures.begin(); it != mResponderFutures.end();)`
	`424`	`+ for (auto it = mSenderFutures.begin(); it != mSenderFutures.end();)`
`426`	`425`	`{`
`427`	`426`	`auto& [request, future] = *it;`
`428`	`427`	`if (blockAll \|\| (toCompleteIdSet.find(request->mRequestId) != toCompleteIdSet.end()))`
`429`	`428`	`{`
`430`	`429`	`future.get();`
`431`	`430`	`request->setState(LlmRequestState::kDISAGG_CONTEXT_COMPLETE);`
`432`		`- it = mResponderFutures.erase(it);`
	`431`	`+ it = mSenderFutures.erase(it);`
`433`	`432`	`}`
`434`	`433`	`else`
`435`	`434`	`{`
Original file line number	Diff line number	Diff line change
`@@ -148,14 +148,14 @@ class DataResponder::Impl`
`148`	`148`	`auto future = promise.get_future();`
`149`	`149`	`{`
`150`	`150`	`{`
`151`		`- std::unique_lock lkResp(mResponderMutex);`
	`151`	`+ std::unique_lock lkResp(mSenderMutex);`
`152`	`152`	`mReadyResponses.emplace(`
`153`	`153`	`llmRequest.mRequestId, Response{std::addressof(llmRequest), std::move(promise)});`
`154`	`154`	`}`
`155`	`155`	`std::unique_lock lkCond(mCondMutex);`
`156`	`156`	`mAnyReady = true;`
`157`	`157`	`}`
`158`		`- mResponderCv.notify_all();`
	`158`	`+ mSenderCv.notify_all();`
`159`	`159`	`return future;`
`160`	`160`	`}`
`161`	`161`
`@@ -208,7 +208,7 @@ class DataResponder::Impl`
`208`	`208`	`if (!mAnyReady)`
`209`	`209`	`{`
`210`	`210`	`std::unique_lock lk(mCondMutex);`
`211`		`- mResponderCv.wait(lk, [this]() { return (mAnyReady \|\| mTerminate); });`
	`211`	`+ mSenderCv.wait(lk, [this]() { return (mAnyReady \|\| mTerminate); });`
`212`	`212`	`}`
`213`	`213`	`if (mTerminate)`
`214`	`214`	`{`
`@@ -263,7 +263,7 @@ class DataResponder::Impl`
`263`	`263`	`"mReadyResponses size is: %zu. mpi rank :%d ",`
`264`	`264`	`mCurrentRequest.value(), mReadyResponses.size(), mpi::MpiComm::world().getRank());`
`265`	`265`	`std::unique_lock lk(mCondMutex);`
`266`		`- mResponderCv.wait(lk, [this]() { return (mAnyReady \|\| mTerminate); });`
	`266`	`+ mSenderCv.wait(lk, [this]() { return (mAnyReady \|\| mTerminate); });`
`267`	`267`	`}`
`268`	`268`	`}`
`269`	`269`	`}`
`@@ -285,13 +285,13 @@ class DataResponder::Impl`
`285`	`285`	`}`
`286`	`286`	`// We don't have to wait for the future. If another thread is sending data, it won't pay attention`
`287`	`287`	`// to the terminate flag.`
`288`		`- mResponderCv.notify_all();`
	`288`	`+ mSenderCv.notify_all();`
`289`	`289`	`}`
`290`	`290`
`291`	`291`	`void removeResponse(std::map<RequestIdType, Response>::iterator it)`
`292`	`292`	`{`
`293`	`293`	`{`
`294`		`- std::unique_lock lkResp(mResponderMutex);`
	`294`	`+ std::unique_lock lkResp(mSenderMutex);`
`295`	`295`	`mReadyResponses.erase(it);`
`296`	`296`	`}`
`297`	`297`	`if (mReadyResponses.empty())`
`@@ -313,16 +313,16 @@ class DataResponder::Impl`
`313`	`313`
`314`	`314`	`[[nodiscard]] std::map<RequestIdType, Response>::iterator getCurrentResponse()`
`315`	`315`	`{`
`316`		`- std::unique_lock lk(mResponderMutex);`
	`316`	`+ std::unique_lock lk(mSenderMutex);`
`317`	`317`	`return mReadyResponses.find(getCurrentRequestId());`
`318`	`318`	`}`
`319`	`319`
`320`	`320`	`private:`
`321`	`321`	`std::optional<RequestIdType> mCurrentRequest;`
`322`	`322`	`std::map<RequestIdType, Response> mReadyResponses;`
`323`		`- std::mutex mResponderMutex, mCondMutex;`
	`323`	`+ std::mutex mSenderMutex, mCondMutex;`
`324`	`324`	`std::atomic<bool> mAnyReady{false}, mTerminate{false};`
`325`		`- std::condition_variable mResponderCv;`
	`325`	`+ std::condition_variable mSenderCv;`
`326`	`326`	`std::future<void> mResponseFuture;`
`327`	`327`	`std::unique_ptr<DataSender> mSender;`
`328`	`328`	`std::unordered_map<LlmRequest::RequestIdType, int> mRemainSendCount;`
`@@ -333,9 +333,9 @@ class DataRequester::Impl`
`333`	`333`	`{`
`334`	`334`	`public:`
`335`	`335`	`Impl(std::unique_ptr<DataReceiver> receiver)`
`336`		`- : mReceiver{std::move(receiver)}`
	`336`	`+ : mCacheReceiver{std::move(receiver)}`
`337`	`337`	`{`
`338`		`- TLLM_CHECK(mReceiver);`
	`338`	`+ TLLM_CHECK(mCacheReceiver);`
`339`	`339`	`TLLM_CUDA_CHECK(cudaGetDevice(&mDeviceId));`
`340`	`340`	`}`
`341`	`341`
`@@ -400,8 +400,8 @@ class DataRequester::Impl`
`400`	`400`	`llmRequest.getContextPhaseParams().value().getReqId());`
`401`	`401`	`llmRequest.setKvCacheTransferStart(std::chrono::steady_clock::now());`
`402`	`402`	`TLLM_CUDA_CHECK(cudaSetDevice(mDeviceId));`
`403`		`- auto session = mReceiver->sendRequestInfo(llmRequest);`
`404`		`- mReceiver->receiveSync(session);`
	`403`	`+ auto session = mCacheReceiver->sendRequestInfo(llmRequest);`
	`404`	`+ mCacheReceiver->receiveSync(session);`
`405`	`405`	`llmRequest.setKvCacheTransferEnd(std::chrono::steady_clock::now());`
`406`	`406`
`407`	`407`	`TLLM_LOG_DEBUG(mpi::MpiComm::world().getRank(),`
`@@ -507,7 +507,7 @@ class DataRequester::Impl`
`507`	`507`	`}`
`508`	`508`	`}`
`509`	`509`
`510`		`- std::unique_ptr<DataReceiver> mReceiver;`
	`510`	`+ std::unique_ptr<DataReceiver> mCacheReceiver;`
`511`	`511`	`int mDeviceId{-1};`
`512`	`512`
`513`	`513`	`std::vector<std::future<void>> mRequestFutures;`
Original file line number	Diff line number	Diff line change
`@@ -47,7 +47,7 @@ static fs::path getTransferOutputPath(char const* tag)`
`47`	`47`	`return {};`
`48`	`48`	`}`
`49`	`49`
`50`		`-DataSenderImpl::DataSenderImpl(executor::kv_cache::ConnectionManager* manager,`
	`50`	`+CacheSenderImpl::CacheSenderImpl(executor::kv_cache::ConnectionManager* manager,`
`51`	`51`	`executor::kv_cache::CacheState selfCacheState, SizeType32 selfIndex, std::unique_ptr<BaseCacheFormatter> formatter)`
`52`	`52`	`: mManager{manager}`
`53`	`53`	`, mSelfState{std::move(selfCacheState), executor::kv_cache::CommState{manager->getCommState()}}`
`@@ -58,7 +58,7 @@ DataSenderImpl::DataSenderImpl(executor::kv_cache::ConnectionManager* manager,`
`58`	`58`	`TLLM_CHECK(mManager->getCommState().getSelfIdx() == selfIndex);`
`59`	`59`	`}`
`60`	`60`
`61`		`-[[nodiscard]] RequestInfo DataSenderImpl::recvRequestInfo()`
	`61`	`+[[nodiscard]] RequestInfo CacheSenderImpl::recvRequestInfo()`
`62`	`62`	`{`
`63`	`63`	`using DataContext = tensorrt_llm::executor::kv_cache::DataContext;`
`64`	`64`	`auto* agentConnectionManager = dynamic_cast<executor::kv_cache::AgentConnectionManager*>(mManager);`
`@@ -111,7 +111,7 @@ DataSenderImpl::DataSenderImpl(executor::kv_cache::ConnectionManager* manager,`
`111`	`111`	`return info;`
`112`	`112`	`}`
`113`	`113`
`114`		`-void DataSenderImpl::sendSync(LlmRequest const& llmRequest)`
	`114`	`+void CacheSenderImpl::sendSync(LlmRequest const& llmRequest)`
`115`	`115`	`{`
`116`	`116`	`auto it = mRequestToSession.find(llmRequest.mRequestId);`
`117`	`117`	`TLLM_CHECK(it != mRequestToSession.end());`
`@@ -120,24 +120,24 @@ void DataSenderImpl::sendSync(LlmRequest const& llmRequest)`
`120`	`120`	`mFormatter->format(session);`
`121`	`121`	`}`
`122`	`122`
`123`		`-[[nodiscard]] executor::kv_cache::CommState const& DataSenderImpl::getCommState() const`
	`123`	`+[[nodiscard]] executor::kv_cache::CommState const& CacheSenderImpl::getCommState() const`
`124`	`124`	`{`
`125`	`125`	`return mSelfState.getCommState().value();`
`126`	`126`	`}`
`127`	`127`
`128`		`-void DataSenderImpl::setCommState(executor::kv_cache::CommState commState)`
	`128`	`+void CacheSenderImpl::setCommState(executor::kv_cache::CommState commState)`
`129`	`129`	`{`
`130`	`130`	`mSelfState.setCommState(std::move(commState));`
`131`	`131`	`}`
`132`	`132`
`133`		`-[[nodiscard]] size_t DataSenderImpl::getCounterpartsCount(LlmRequest::RequestIdType requestId) const`
	`133`	`+[[nodiscard]] size_t CacheSenderImpl::getCounterpartsCount(LlmRequest::RequestIdType requestId) const`
`134`	`134`	`{`
`135`	`135`	`auto it = mRequestToSession.find(requestId);`
`136`	`136`	`TLLM_CHECK(it != mRequestToSession.end());`
`137`	`137`	`return it->second.getConnections().size();`
`138`	`138`	`}`
`139`	`139`
`140`		`-void DataSenderImpl::release(LlmRequest::RequestIdType requestId)`
	`140`	`+void CacheSenderImpl::release(LlmRequest::RequestIdType requestId)`
`141`	`141`	`{`
`142`	`142`	`auto it = mRequestToSession.find(requestId);`
`143`	`143`	`TLLM_CHECK(it != mRequestToSession.end());`
`@@ -156,7 +156,7 @@ void DataSenderImpl::release(LlmRequest::RequestIdType requestId)`
`156`	`156`	`mRequestToSession.erase(it);`
`157`	`157`	`}`
`158`	`158`
`159`		`-DataReceiverImpl::DataReceiverImpl(executor::kv_cache::ConnectionManager* manager,`
	`159`	`+CacheReceiverImpl::CacheReceiverImpl(executor::kv_cache::ConnectionManager* manager,`
`160`	`160`	`executor::kv_cache::CacheState selfCacheState, SizeType32 selfIndex, std::unique_ptr<BaseCacheFormatter> formatter)`
`161`	`161`	`: mManager{manager}`
`162`	`162`	`, mSelfState{std::move(selfCacheState), executor::kv_cache::CommState{manager->getCommState()}}`
`@@ -167,7 +167,7 @@ DataReceiverImpl::DataReceiverImpl(executor::kv_cache::ConnectionManager* manage`
`167`	`167`	`TLLM_CHECK(mFormatter);`
`168`	`168`	`}`
`169`	`169`
`170`		`-TransferSession DataReceiverImpl::sendRequestInfo(LlmRequest const& llmRequest)`
	`170`	`+TransferSession CacheReceiverImpl::sendRequestInfo(LlmRequest const& llmRequest)`
`171`	`171`	`{`
`172`	`172`	`uint64_t requestId = llmRequest.getContextPhaseParams().value().getReqId();`
`173`	`173`	`auto const& contextState = llmRequest.getDataTransceiverState();`
`@@ -233,7 +233,7 @@ TransferSession DataReceiverImpl::sendRequestInfo(LlmRequest const& llmRequest)`
`233`	`233`	`contextState, resource->mBufferManager, &llmRequest, !common::getEnvKVCacheTransferOutputPath().empty());`
`234`	`234`	`}`
`235`	`235`
`236`		`-void DataReceiverImpl::receiveSync(TransferSession& session)`
	`236`	`+void CacheReceiverImpl::receiveSync(TransferSession& session)`
`237`	`237`	`{`
`238`	`238`	`mFormatter->unformat(session);`
`239`	`239`	`if (!common::getEnvKVCacheTransferOutputPath().empty())`
`@@ -250,7 +250,7 @@ void DataReceiverImpl::receiveSync(TransferSession& session)`
`250`	`250`	`}`
`251`	`251`	`}`
`252`	`252`
`253`		`-void DataReceiverImpl::sendRequestInfo(executor::kv_cache::Connection const* connection, RequestInfo const& info)`
	`253`	`+void CacheReceiverImpl::sendRequestInfo(executor::kv_cache::Connection const* connection, RequestInfo const& info)`
`254`	`254`	`{`
`255`	`255`	`std::ostringstream oss;`
`256`	`256`	`RequestInfo::serialize(info, oss);`
`@@ -262,7 +262,7 @@ void DataReceiverImpl::sendRequestInfo(executor::kv_cache::Connection const* con`
`262`	`262`	`connection->send(executor::kv_cache::DataContext{kINFO_TAG}, serializedInfo.data(), infoSize);`
`263`	`263`	`}`
`264`	`264`
`265`		`-std::unique_ptr<DataReceiverImpl::ReceiveCacheResource> const& DataReceiverImpl::getReceiveCacheResource(`
	`265`	`+std::unique_ptr<CacheReceiverImpl::ReceiveCacheResource> const& CacheReceiverImpl::getReceiveCacheResource(`
`266`	`266`	`LlmRequest const& llmRequest)`
`267`	`267`	`{`
`268`	`268`	`std::scoped_lock<std::mutex> lock(mProcessIoResouceMutex);`
Original file line number	Diff line number	Diff line change
`@@ -81,7 +81,7 @@ void AgentConnection::send(DataContext const& ctx, void const* data, size_t size`
`81`	`81`	`MemoryDesc srcDesc{`
`82`	`82`	`reinterpret_cast<uintptr_t>(data), size, static_cast<uint32_t>(mAgentConnectionManager->getDeviceId())};`
`83`	`83`	`MemoryDescs srcDescs{MemoryType::kVRAM, {srcDesc}};`
`84`		`- auto dstBaseDesc = mSenderState.mReceiverBufferDesc;`
	`84`	`+ auto dstBaseDesc = mSenderState.mCacheReceiverBufferDesc;`
`85`	`85`	`MemoryDesc dstDesc{dstBaseDesc.getAddr() + (mSenderState.validSegmentIdx * size), size, dstBaseDesc.getDeviceId()};`
`86`	`86`	`TLLM_LOG_DEBUG(`
`87`	`87`	`"send dstDesc: %p, size: %ld ,validSegmentIdx: %ld", dstDesc.getAddr(), size, mSenderState.validSegmentIdx);`
`@@ -137,9 +137,9 @@ void AgentConnection::sendRequestAndBufferInfo(`
`137`	`137`	`mAgentConnectionManager->getAgent()->notifySyncMessage(mRemoteAgentName, ss.str());`
`138`	`138`	`}`
`139`	`139`
`140`		`-void AgentConnection::setSenderState(MemoryDesc mReceiverBufferDesc, int validSegmentIdx)`
	`140`	`+void AgentConnection::setSenderState(MemoryDesc mCacheReceiverBufferDesc, int validSegmentIdx)`
`141`	`141`	`{`
`142`		`- mSenderState.mReceiverBufferDesc = mReceiverBufferDesc;`
	`142`	`+ mSenderState.mCacheReceiverBufferDesc = mCacheReceiverBufferDesc;`
`143`	`143`	`mSenderState.validSegmentIdx = validSegmentIdx;`
`144`	`144`	`}`
`145`	`145`