Rename data -> cache

Tabrizian · Tabrizian · commit 762884cea562 · 2025-07-24T19:28:34.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
@@ -194,10 +194,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();
 }
@@ -213,7 +213,7 @@ CacheTransceiver::~CacheTransceiver()
 
 void CacheTransceiver::initializeCommState()
 {
-    mCommState = std::addressof(mDataResponder->getCommState());
+    mCommState = std::addressof(mCacheSender->getCommState());
 }
 
 void CacheTransceiver::setContextState(LlmRequest* llmRequest)
@@ -249,8 +249,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(
@@ -265,8 +265,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));
     }
 }
 
@@ -372,7 +372,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)
         {
@@ -412,23 +412,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
@@ -111,14 +111,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;
     }
 
@@ -171,7 +171,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)
                 {
@@ -226,7 +226,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); });
                 }
             }
         }
@@ -248,13 +248,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())
@@ -276,16 +276,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;
@@ -296,9 +296,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));
     }
 
@@ -363,8 +363,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(),
@@ -470,7 +470,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
@@ -30,7 +30,7 @@ static int32_t tagFromRequestId(LlmRequest::RequestIdType requestId)
     return ((requestId & 0xFFF) << 8) | (kDATA_TAG & 0xFF);
 }
 
-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()}}
@@ -41,7 +41,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);
@@ -93,7 +93,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());
@@ -102,32 +102,32 @@ 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());
     std::unique_lock<std::mutex> lk(mMtxForMap);
     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()}}
@@ -138,7 +138,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();
@@ -204,12 +204,12 @@ TransferSession DataReceiverImpl::sendRequestInfo(LlmRequest const& llmRequest)
         contextState, resource->mBufferManager, &llmRequest);
 }
 
-void DataReceiverImpl::receiveSync(TransferSession& session)
+void CacheReceiverImpl::receiveSync(TransferSession& session)
 {
     mFormatter->unformat(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);
@@ -221,7 +221,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
@@ -40,12 +40,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;
@@ -69,12 +69,12 @@ class DataSenderImpl : public DataSender, public TransceiverTag
     runtime::BufferManager mBufferManager;
 };
 
-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
`@@ -194,10 +194,10 @@ CacheTransceiver::CacheTransceiver(kv_cache_manager::BaseKVCacheManager* cacheMa`
`194`	`194`	`auto makeFormatter = [cacheManager, isMLA, this]()`
`195`	`195`	`{ return createCacheFormatter(cacheManager, mCacheTransBufferManager.get(), isMLA); };`
`196`	`196`
`197`		`- mDataResponder = std::make_unique<DataResponder>(`
`198`		`- std::make_unique<DataSenderImpl>(mManager.get(), *mCacheState, worldConfig.getRank(), makeFormatter()));`
	`197`	`+ mCacheSender = std::make_unique<DataResponder>(`
	`198`	`+ std::make_unique<CacheSenderImpl>(mManager.get(), *mCacheState, worldConfig.getRank(), makeFormatter()));`
`199`	`199`	`mDataRequester = std::make_unique<DataRequester>(`
`200`		`- std::make_unique<DataReceiverImpl>(mManager.get(), *mCacheState, worldConfig.getRank(), makeFormatter()));`
	`200`	`+ std::make_unique<CacheReceiverImpl>(mManager.get(), *mCacheState, worldConfig.getRank(), makeFormatter()));`
`201`	`201`
`202`	`202`	`initializeCommState();`
`203`	`203`	`}`
`@@ -213,7 +213,7 @@ CacheTransceiver::~CacheTransceiver()`
`213`	`213`
`214`	`214`	`void CacheTransceiver::initializeCommState()`
`215`	`215`	`{`
`216`		`- mCommState = std::addressof(mDataResponder->getCommState());`
	`216`	`+ mCommState = std::addressof(mCacheSender->getCommState());`
`217`	`217`	`}`
`218`	`218`
`219`	`219`	`void CacheTransceiver::setContextState(LlmRequest* llmRequest)`
`@@ -249,8 +249,8 @@ void CacheTransceiver::respondAndSendAsync(LlmRequest* llmRequest)`
`249`	`249`	`return;`
`250`	`250`	`}`
`251`	`251`	`setContextState(llmRequest);`
`252`		`- auto future = mDataResponder->respondAndSendAsync(*llmRequest);`
`253`		`- mResponderFutures.emplace_back(llmRequest, std::move(future));`
	`252`	`+ auto future = mCacheSender->respondAndSendAsync(*llmRequest);`
	`253`	`+ mSenderFutures.emplace_back(llmRequest, std::move(future));`
`254`	`254`	`}`
`255`	`255`
`256`	`256`	`void CacheTransceiver::respondAndSendLayerWise(`
`@@ -265,8 +265,8 @@ void CacheTransceiver::respondAndSendLayerWise(`
`265`	`265`
`266`	`266`	`llmRequest->setState(LlmRequestState::kDISAGG_CONTEXT_INIT_AND_TRANS);`
`267`	`267`	`setContextState(llmRequest.get());`
`268`		`- auto future = mDataResponder->respondAndSendAsync(*llmRequest);`
`269`		`- mResponderFutures.emplace_back(llmRequest.get(), std::move(future));`
	`268`	`+ auto future = mCacheSender->respondAndSendAsync(*llmRequest);`
	`269`	`+ mSenderFutures.emplace_back(llmRequest.get(), std::move(future));`
`270`	`270`	`}`
`271`	`271`	`}`
`272`	`272`
`@@ -372,7 +372,7 @@ void CacheTransceiver::checkContextTransferStatus(std::optional<int> const& atLe`
`372`	`372`	`bool blockAll = !atLeastRequestNum.has_value();`
`373`	`373`	`auto syncComm = mCacheState->getParallelConfig().mEnableAttentionDP ? mMpiGroupTPInDPComm : mMpiGroupTensorParaComm;`
`374`	`374`	`std::vector<LlmRequest::RequestIdType> contextCompleteRequestIds;`
`375`		`- for (auto&& [request, future] : mResponderFutures)`
	`375`	`+ for (auto&& [request, future] : mSenderFutures)`
`376`	`376`	`{`
`377`	`377`	`if (future.wait_for(std::chrono::milliseconds(0)) == std::future_status::ready)`
`378`	`378`	`{`
`@@ -412,23 +412,22 @@ void CacheTransceiver::checkContextTransferStatus(std::optional<int> const& atLe`
`412`	`412`
`413`	`413`	`// Make sure there are at least atLeastRequestNum requests in toCompleteIdSet.`
`414`	`414`	`// This will preserve the order of insertion for KVCache transfer requests.`
`415`		`- for (auto it = mResponderFutures.begin();`
`416`		`- atLeastRequestNum.value_or(0) > static_cast<int>(toCompleteIdSet.size()) && it != mResponderFutures.end();`
`417`		`- ++it)`
	`415`	`+ for (auto it = mSenderFutures.begin();`
	`416`	`+ atLeastRequestNum.value_or(0) > static_cast<int>(toCompleteIdSet.size()) && it != mSenderFutures.end(); ++it)`
`418`	`417`	`{`
`419`	`418`	`auto& [request, future] = *it;`
`420`	`419`	`toCompleteIdSet.insert(request->mRequestId);`
`421`	`420`	`}`
`422`	`421`
`423`	`422`	`// Complete all the requests in toCompleteIdSet`
`424`		`- for (auto it = mResponderFutures.begin(); it != mResponderFutures.end();)`
	`423`	`+ for (auto it = mSenderFutures.begin(); it != mSenderFutures.end();)`
`425`	`424`	`{`
`426`	`425`	`auto& [request, future] = *it;`
`427`	`426`	`if (blockAll \|\| (toCompleteIdSet.find(request->mRequestId) != toCompleteIdSet.end()))`
`428`	`427`	`{`
`429`	`428`	`future.get();`
`430`	`429`	`request->setState(LlmRequestState::kDISAGG_CONTEXT_COMPLETE);`
`431`		`- it = mResponderFutures.erase(it);`
	`430`	`+ it = mSenderFutures.erase(it);`
`432`	`431`	`}`
`433`	`432`	`else`
`434`	`433`	`{`
Original file line number	Diff line number	Diff line change
`@@ -111,14 +111,14 @@ class DataResponder::Impl`
`111`	`111`	`auto future = promise.get_future();`
`112`	`112`	`{`
`113`	`113`	`{`
`114`		`- std::unique_lock lkResp(mResponderMutex);`
	`114`	`+ std::unique_lock lkResp(mSenderMutex);`
`115`	`115`	`mReadyResponses.emplace(`
`116`	`116`	`llmRequest.mRequestId, Response{std::addressof(llmRequest), std::move(promise)});`
`117`	`117`	`}`
`118`	`118`	`std::unique_lock lkCond(mCondMutex);`
`119`	`119`	`mAnyReady = true;`
`120`	`120`	`}`
`121`		`- mResponderCv.notify_all();`
	`121`	`+ mSenderCv.notify_all();`
`122`	`122`	`return future;`
`123`	`123`	`}`
`124`	`124`
`@@ -171,7 +171,7 @@ class DataResponder::Impl`
`171`	`171`	`if (!mAnyReady)`
`172`	`172`	`{`
`173`	`173`	`std::unique_lock lk(mCondMutex);`
`174`		`- mResponderCv.wait(lk, [this]() { return (mAnyReady \|\| mTerminate); });`
	`174`	`+ mSenderCv.wait(lk, [this]() { return (mAnyReady \|\| mTerminate); });`
`175`	`175`	`}`
`176`	`176`	`if (mTerminate)`
`177`	`177`	`{`
`@@ -226,7 +226,7 @@ class DataResponder::Impl`
`226`	`226`	`"mReadyResponses size is: %zu. mpi rank :%d ",`
`227`	`227`	`mCurrentRequest.value(), mReadyResponses.size(), mpi::MpiComm::world().getRank());`
`228`	`228`	`std::unique_lock lk(mCondMutex);`
`229`		`- mResponderCv.wait(lk, [this]() { return (mAnyReady \|\| mTerminate); });`
	`229`	`+ mSenderCv.wait(lk, [this]() { return (mAnyReady \|\| mTerminate); });`
`230`	`230`	`}`
`231`	`231`	`}`
`232`	`232`	`}`
`@@ -248,13 +248,13 @@ class DataResponder::Impl`
`248`	`248`	`}`
`249`	`249`	`// We don't have to wait for the future. If another thread is sending data, it won't pay attention`
`250`	`250`	`// to the terminate flag.`
`251`		`- mResponderCv.notify_all();`
	`251`	`+ mSenderCv.notify_all();`
`252`	`252`	`}`
`253`	`253`
`254`	`254`	`void removeResponse(std::map<RequestIdType, Response>::iterator it)`
`255`	`255`	`{`
`256`	`256`	`{`
`257`		`- std::unique_lock lkResp(mResponderMutex);`
	`257`	`+ std::unique_lock lkResp(mSenderMutex);`
`258`	`258`	`mReadyResponses.erase(it);`
`259`	`259`	`}`
`260`	`260`	`if (mReadyResponses.empty())`
`@@ -276,16 +276,16 @@ class DataResponder::Impl`
`276`	`276`
`277`	`277`	`[[nodiscard]] std::map<RequestIdType, Response>::iterator getCurrentResponse()`
`278`	`278`	`{`
`279`		`- std::unique_lock lk(mResponderMutex);`
	`279`	`+ std::unique_lock lk(mSenderMutex);`
`280`	`280`	`return mReadyResponses.find(getCurrentRequestId());`
`281`	`281`	`}`
`282`	`282`
`283`	`283`	`private:`
`284`	`284`	`std::optional<RequestIdType> mCurrentRequest;`
`285`	`285`	`std::map<RequestIdType, Response> mReadyResponses;`
`286`		`- std::mutex mResponderMutex, mCondMutex;`
	`286`	`+ std::mutex mSenderMutex, mCondMutex;`
`287`	`287`	`std::atomic<bool> mAnyReady{false}, mTerminate{false};`
`288`		`- std::condition_variable mResponderCv;`
	`288`	`+ std::condition_variable mSenderCv;`
`289`	`289`	`std::future<void> mResponseFuture;`
`290`	`290`	`std::unique_ptr<DataSender> mSender;`
`291`	`291`	`std::unordered_map<LlmRequest::RequestIdType, int> mRemainSendCount;`
`@@ -296,9 +296,9 @@ class DataRequester::Impl`
`296`	`296`	`{`
`297`	`297`	`public:`
`298`	`298`	`Impl(std::unique_ptr<DataReceiver> receiver)`
`299`		`- : mReceiver{std::move(receiver)}`
	`299`	`+ : mCacheReceiver{std::move(receiver)}`
`300`	`300`	`{`
`301`		`- TLLM_CHECK(mReceiver);`
	`301`	`+ TLLM_CHECK(mCacheReceiver);`
`302`	`302`	`TLLM_CUDA_CHECK(cudaGetDevice(&mDeviceId));`
`303`	`303`	`}`
`304`	`304`
`@@ -363,8 +363,8 @@ class DataRequester::Impl`
`363`	`363`	`llmRequest.getContextPhaseParams().value().getReqId());`
`364`	`364`	`llmRequest.setKvCacheTransferStart(std::chrono::steady_clock::now());`
`365`	`365`	`TLLM_CUDA_CHECK(cudaSetDevice(mDeviceId));`
`366`		`- auto session = mReceiver->sendRequestInfo(llmRequest);`
`367`		`- mReceiver->receiveSync(session);`
	`366`	`+ auto session = mCacheReceiver->sendRequestInfo(llmRequest);`
	`367`	`+ mCacheReceiver->receiveSync(session);`
`368`	`368`	`llmRequest.setKvCacheTransferEnd(std::chrono::steady_clock::now());`
`369`	`369`
`370`	`370`	`TLLM_LOG_DEBUG(mpi::MpiComm::world().getRank(),`
`@@ -470,7 +470,7 @@ class DataRequester::Impl`
`470`	`470`	`}`
`471`	`471`	`}`
`472`	`472`
`473`		`- std::unique_ptr<DataReceiver> mReceiver;`
	`473`	`+ std::unique_ptr<DataReceiver> mCacheReceiver;`
`474`	`474`	`int mDeviceId{-1};`
`475`	`475`
`476`	`476`	`std::vector<std::future<void>> mRequestFutures;`
Original file line number	Diff line number	Diff line change
`@@ -30,7 +30,7 @@ static int32_t tagFromRequestId(LlmRequest::RequestIdType requestId)`
`30`	`30`	`return ((requestId & 0xFFF) << 8) \| (kDATA_TAG & 0xFF);`
`31`	`31`	`}`
`32`	`32`
`33`		`-DataSenderImpl::DataSenderImpl(executor::kv_cache::ConnectionManager* manager,`
	`33`	`+CacheSenderImpl::CacheSenderImpl(executor::kv_cache::ConnectionManager* manager,`
`34`	`34`	`executor::kv_cache::CacheState selfCacheState, SizeType32 selfIndex, std::unique_ptr<BaseCacheFormatter> formatter)`
`35`	`35`	`: mManager{manager}`
`36`	`36`	`, mSelfState{std::move(selfCacheState), executor::kv_cache::CommState{manager->getCommState()}}`
`@@ -41,7 +41,7 @@ DataSenderImpl::DataSenderImpl(executor::kv_cache::ConnectionManager* manager,`
`41`	`41`	`TLLM_CHECK(mManager->getCommState().getSelfIdx() == selfIndex);`
`42`	`42`	`}`
`43`	`43`
`44`		`-[[nodiscard]] RequestInfo DataSenderImpl::recvRequestInfo()`
	`44`	`+[[nodiscard]] RequestInfo CacheSenderImpl::recvRequestInfo()`
`45`	`45`	`{`
`46`	`46`	`using DataContext = tensorrt_llm::executor::kv_cache::DataContext;`
`47`	`47`	`auto* agentConnectionManager = dynamic_cast<executor::kv_cache::AgentConnectionManager*>(mManager);`
`@@ -93,7 +93,7 @@ DataSenderImpl::DataSenderImpl(executor::kv_cache::ConnectionManager* manager,`
`93`	`93`	`return info;`
`94`	`94`	`}`
`95`	`95`
`96`		`-void DataSenderImpl::sendSync(LlmRequest const& llmRequest)`
	`96`	`+void CacheSenderImpl::sendSync(LlmRequest const& llmRequest)`
`97`	`97`	`{`
`98`	`98`	`auto it = mRequestToSession.find(llmRequest.mRequestId);`
`99`	`99`	`TLLM_CHECK(it != mRequestToSession.end());`
`@@ -102,32 +102,32 @@ void DataSenderImpl::sendSync(LlmRequest const& llmRequest)`
`102`	`102`	`mFormatter->format(session);`
`103`	`103`	`}`
`104`	`104`
`105`		`-[[nodiscard]] executor::kv_cache::CommState const& DataSenderImpl::getCommState() const`
	`105`	`+[[nodiscard]] executor::kv_cache::CommState const& CacheSenderImpl::getCommState() const`
`106`	`106`	`{`
`107`	`107`	`return mSelfState.getCommState().value();`
`108`	`108`	`}`
`109`	`109`
`110`		`-void DataSenderImpl::setCommState(executor::kv_cache::CommState commState)`
	`110`	`+void CacheSenderImpl::setCommState(executor::kv_cache::CommState commState)`
`111`	`111`	`{`
`112`	`112`	`mSelfState.setCommState(std::move(commState));`
`113`	`113`	`}`
`114`	`114`
`115`		`-[[nodiscard]] size_t DataSenderImpl::getCounterpartsCount(LlmRequest::RequestIdType requestId) const`
	`115`	`+[[nodiscard]] size_t CacheSenderImpl::getCounterpartsCount(LlmRequest::RequestIdType requestId) const`
`116`	`116`	`{`
`117`	`117`	`auto it = mRequestToSession.find(requestId);`
`118`	`118`	`TLLM_CHECK(it != mRequestToSession.end());`
`119`	`119`	`return it->second.getConnections().size();`
`120`	`120`	`}`
`121`	`121`
`122`		`-void DataSenderImpl::release(LlmRequest::RequestIdType requestId)`
	`122`	`+void CacheSenderImpl::release(LlmRequest::RequestIdType requestId)`
`123`	`123`	`{`
`124`	`124`	`auto it = mRequestToSession.find(requestId);`
`125`	`125`	`TLLM_CHECK(it != mRequestToSession.end());`
`126`	`126`	`std::unique_lock<std::mutex> lk(mMtxForMap);`
`127`	`127`	`mRequestToSession.erase(it);`
`128`	`128`	`}`
`129`	`129`
`130`		`-DataReceiverImpl::DataReceiverImpl(executor::kv_cache::ConnectionManager* manager,`
	`130`	`+CacheReceiverImpl::CacheReceiverImpl(executor::kv_cache::ConnectionManager* manager,`
`131`	`131`	`executor::kv_cache::CacheState selfCacheState, SizeType32 selfIndex, std::unique_ptr<BaseCacheFormatter> formatter)`
`132`	`132`	`: mManager{manager}`
`133`	`133`	`, mSelfState{std::move(selfCacheState), executor::kv_cache::CommState{manager->getCommState()}}`
`@@ -138,7 +138,7 @@ DataReceiverImpl::DataReceiverImpl(executor::kv_cache::ConnectionManager* manage`
`138`	`138`	`TLLM_CHECK(mFormatter);`
`139`	`139`	`}`
`140`	`140`
`141`		`-TransferSession DataReceiverImpl::sendRequestInfo(LlmRequest const& llmRequest)`
	`141`	`+TransferSession CacheReceiverImpl::sendRequestInfo(LlmRequest const& llmRequest)`
`142`	`142`	`{`
`143`	`143`	`uint64_t requestId = llmRequest.getContextPhaseParams().value().getReqId();`
`144`	`144`	`auto const& contextState = llmRequest.getDataTransceiverState();`
`@@ -204,12 +204,12 @@ TransferSession DataReceiverImpl::sendRequestInfo(LlmRequest const& llmRequest)`
`204`	`204`	`contextState, resource->mBufferManager, &llmRequest);`
`205`	`205`	`}`
`206`	`206`
`207`		`-void DataReceiverImpl::receiveSync(TransferSession& session)`
	`207`	`+void CacheReceiverImpl::receiveSync(TransferSession& session)`
`208`	`208`	`{`
`209`	`209`	`mFormatter->unformat(session);`
`210`	`210`	`}`
`211`	`211`
`212`		`-void DataReceiverImpl::sendRequestInfo(executor::kv_cache::Connection const* connection, RequestInfo const& info)`
	`212`	`+void CacheReceiverImpl::sendRequestInfo(executor::kv_cache::Connection const* connection, RequestInfo const& info)`
`213`	`213`	`{`
`214`	`214`	`std::ostringstream oss;`
`215`	`215`	`RequestInfo::serialize(info, oss);`
`@@ -221,7 +221,7 @@ void DataReceiverImpl::sendRequestInfo(executor::kv_cache::Connection const* con`
`221`	`221`	`connection->send(executor::kv_cache::DataContext{kINFO_TAG}, serializedInfo.data(), infoSize);`
`222`	`222`	`}`
`223`	`223`
`224`		`-std::unique_ptr<DataReceiverImpl::ReceiveCacheResource> const& DataReceiverImpl::getReceiveCacheResource(`
	`224`	`+std::unique_ptr<CacheReceiverImpl::ReceiveCacheResource> const& CacheReceiverImpl::getReceiveCacheResource(`
`225`	`225`	`LlmRequest const& llmRequest)`
`226`	`226`	`{`
`227`	`227`	`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`