Repository navigation
[TRTLLM-11513][fix] eliminate dangling reference and TOCTOU races in KVCacheManager #12251
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 15 commits
5a29d81
c3a9589
177bfc2
13aa0e1
f604c45
11acc93
e4773f4
11dfe67
4b1f83d
350b6e1
606a745
d39d595
782ddc3
5b1b93f
5b3ce02
6529a63
5753193
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -2270,7 +2270,9 @@ SizeType32 KVCacheManager::getNeededBlocksOneStep( | |
| return 0; | ||
| } | ||
|
|
||
| auto const numCurrTokens = getSequence(req.mRequestId).getNumTokens(); | ||
| auto const seqPtr_ = getSequence(req.mRequestId); | ||
| TLLM_CHECK_WITH_INFO(seqPtr_, "No sequence found for request %lu", req.mRequestId); | ||
| auto const numCurrTokens = seqPtr_->getNumTokens(); | ||
| auto const generatedTokens = numCurrTokens - req.getPromptLen(); | ||
| auto const maxTokensToAddToKVCache = req.mMaxNewTokens - generatedTokens; | ||
| auto const tokensPerStep = req.getNumDraftTokens() + 1; | ||
|
|
@@ -2322,7 +2324,7 @@ SizeType32 KVCacheManager::getRemainingBlocksToCompletion(LlmRequest const& req, | |
| if (seqIt != mSequences.end()) | ||
| { | ||
| auto const& seq = seqIt->second; | ||
| numAllocBlocksPerBeam = seq.getCacheBlockIds(windowSize).at(0).size(); | ||
| numAllocBlocksPerBeam = seq->getCacheBlockIds(windowSize).at(0).size(); | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -2426,7 +2428,17 @@ void BlockManager::updateCacheBlockOffsetsAtIdx(GenerationRequest& sequence, Siz | |
| void KVCacheManager::addToken(RequestIdType requestId) | ||
| { | ||
| // TODO: add streamLLM support | ||
| auto& sequence = getSequence(requestId); | ||
| auto seqPtr = getSequence(requestId); | ||
| if (!seqPtr) | ||
| { | ||
| return; | ||
| } | ||
| auto seqLock = seqPtr->getLock(); | ||
| if (seqPtr->isRemoved()) | ||
| { | ||
| return; | ||
| } | ||
| auto& sequence = *seqPtr; | ||
|
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I think we shouldn't return silently here for the conditions above. If the conditions happen, it should be a logic bug in the higher level and we should not fail silently.
Collaborator
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The issue here is that we have a separate thread that calls removeSequence, which can potentially remove the sequence before we can lock it. The silent return is necessary because the rest of the code will crash if the sequence has been removed. The higher level code should also perform the same check and take the sequence out of the active batch if it has been removed. Perhaps I should add a warning message saying the sequence has been removed?
Collaborator
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I am really not a fan of the design that has a separate thread decide if sequence should be removed or not. It near guarantees that you will have one thread preparing the sequence for another step while the other thread is removing it. The silent returns are there to prevent the server from crashing when this happens.
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I think we should have a guard at the higher level caller then. Failing silently here is likely not a good convention. |
||
| sequence.addNewTokens(1); | ||
| mBlockManager.adjustBlocksIfNeeded(sequence); | ||
| } | ||
|
|
@@ -2488,14 +2500,18 @@ void KVCacheManager::addSequence( | |
| ? llmRequest->getKvCacheRetentionConfig().value_or(executor::KvCacheRetentionConfig()) | ||
| : executor::KvCacheRetentionConfig(); | ||
|
|
||
| auto const [seqIt, emplaceDone] = [&] | ||
| auto seq = std::make_shared<GenerationRequest>( | ||
| requestId, inputLength, beamWidth, mBlockManager.getWindowSizesMetadata(), std::move(kvCacheRetentionConfig)); | ||
| { | ||
| auto lck = std::scoped_lock(mSequencesMtx); | ||
| return mSequences.try_emplace(requestId, requestId, inputLength, beamWidth, | ||
| mBlockManager.getWindowSizesMetadata(), kvCacheRetentionConfig); | ||
| }(); | ||
| TLLM_CHECK(emplaceDone); | ||
| auto& sequence = seqIt->second; | ||
| auto const [_, emplaceDone] = mSequences.emplace(requestId, seq); | ||
| if (!emplaceDone) | ||
| { | ||
|
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The assertion
Collaborator
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Calling addSequence more than once for the same sequence should not be an error. There are some cases in the python executor were addSequence would be called more than once, if not for some book-keeping that keeps track of whether addSequence has already been called. KVCacheManager already knows if addSequence has been called for a given sequence, doing a silent return when emplaceDone is false makes it safe to call addSequence more than once for the same sequence. We can remove this book-keeping logic in executor with this change. |
||
| return; | ||
| } | ||
| } | ||
|
|
||
| auto& sequence = *seq; | ||
|
|
||
| // Get statistics for block allocations/reuse pre request. | ||
| SizeType32 const numAllocTotalBlocksPreRequest = mBlockManager.getNumAllocTotalBlocks(); | ||
|
|
@@ -2574,18 +2590,12 @@ void KVCacheManager::addSequence( | |
| void KVCacheManager::storeContextBlocks(LlmRequest const& llmRequest) | ||
| { | ||
| auto const requestId = llmRequest.mRequestId; | ||
| bool found = false; | ||
| { | ||
| // protect the mSequences | ||
| std::scoped_lock lock(mSequencesMtx); | ||
| found = mSequences.find(requestId) != mSequences.end(); | ||
| } | ||
| if (found) | ||
| auto seqPtr = getSequence(requestId); | ||
| if (seqPtr) | ||
| { | ||
| auto& sequence = getSequence(requestId); | ||
| if (mEnableBlockReuse && !llmRequest.isDummyRequest()) | ||
| { | ||
| mBlockManager.storeContextBlocks(sequence, llmRequest); | ||
| mBlockManager.storeContextBlocks(*seqPtr, llmRequest); | ||
| } | ||
| } | ||
| else | ||
|
|
@@ -2600,7 +2610,14 @@ void KVCacheManager::storeNewBlock(LlmRequest const& llmRequest) | |
| // - Beam search is NOT enabled | ||
| // - Block reuse is enabled. | ||
| auto const requestId = llmRequest.mRequestId; | ||
| auto& sequence = getSequence(requestId); | ||
| auto seqPtr = getSequence(requestId); | ||
| TLLM_CHECK_WITH_INFO(seqPtr, "No sequence found for request %lu", requestId); | ||
| auto seqLock = seqPtr->getLock(); | ||
| if (seqPtr->isRemoved()) | ||
|
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Should this also be a
Collaborator
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. No. If the sequence was removed before we could lock it, we can't store the new block for reuse, which is a minor inconvenience, not a fatal error. The silent return is necessary to prevent a crash. |
||
| { | ||
| return; | ||
| } | ||
| auto& sequence = *seqPtr; | ||
| if (sequence.getBeamWidth() > 1 || !mEnableBlockReuse) | ||
| { | ||
| return; | ||
|
|
@@ -2612,21 +2629,30 @@ std::optional<KVCacheBlock::IdType> KVCacheManager::removeSequence( | |
| RequestIdType requestId, OptionalRef<LlmRequest const> llmRequest, bool pinBlocks) | ||
| { | ||
| TLLM_LOG_TRACE("[%s]::%s start", isCrossKv() ? "CROSS" : "SELF", __PRETTY_FUNCTION__); | ||
| auto sequenceNode = [this, requestId] | ||
| auto seqPtr = [this, requestId]() -> std::shared_ptr<GenerationRequest> | ||
| { | ||
| std::scoped_lock lock(mSequencesMtx); | ||
| return mSequences.extract(requestId); | ||
| auto it = mSequences.find(requestId); | ||
| if (it == mSequences.end()) | ||
| { | ||
| return nullptr; | ||
| } | ||
| auto ptr = it->second; | ||
| mSequences.erase(it); | ||
| return ptr; | ||
| }(); | ||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
| std::optional<KVCacheBlock::IdType> lastStoredId = std::nullopt; | ||
| if (!sequenceNode.empty()) | ||
| if (seqPtr) | ||
| { | ||
| auto seqLock = seqPtr->getLock(); | ||
| seqPtr->markRemoved(); | ||
| if (mEnableBlockReuse) | ||
| { | ||
| lastStoredId = mBlockManager.releaseBlocks(sequenceNode.mapped(), llmRequest, pinBlocks); | ||
| lastStoredId = mBlockManager.releaseBlocks(*seqPtr, llmRequest, pinBlocks); | ||
| } | ||
| else | ||
| { | ||
| lastStoredId = mBlockManager.releaseBlocks(sequenceNode.mapped(), std::nullopt, pinBlocks); | ||
| lastStoredId = mBlockManager.releaseBlocks(*seqPtr, std::nullopt, pinBlocks); | ||
| } | ||
| } | ||
| if (mBlockManager.isSequenceHeld(requestId)) | ||
|
|
@@ -2643,7 +2669,14 @@ std::vector<KVCacheBlock::IdType> KVCacheManager::storeBlocksForReuse( | |
| RequestIdType requestId, OptionalRef<LlmRequest const> llmRequest, bool pinBlocks) | ||
| { | ||
| TLLM_LOG_TRACE("[%s]::%s start", isCrossKv() ? "CROSS" : "SELF", __PRETTY_FUNCTION__); | ||
| auto& sequence = getSequence(requestId); | ||
| auto seqPtr = getSequence(requestId); | ||
| TLLM_CHECK_WITH_INFO(seqPtr, "No sequence found for request %lu", requestId); | ||
| auto seqLock = seqPtr->getLock(); | ||
| if (seqPtr->isRemoved()) | ||
| { | ||
| return {}; | ||
| } | ||
| auto& sequence = *seqPtr; | ||
| auto pinnedBlockIds = mBlockManager.storeBlocksForReuse(sequence, llmRequest, pinBlocks); | ||
| TLLM_LOG_TRACE("[%s]::%s stop", isCrossKv() ? "CROSS" : "SELF", __PRETTY_FUNCTION__); | ||
| return pinnedBlockIds; | ||
|
|
@@ -2657,8 +2690,14 @@ void KVCacheManager::schedulingRemoveSequence(RequestIdType requestId) | |
|
|
||
| void KVCacheManager::pinBlocks(RequestIdType requestId) | ||
| { | ||
| auto& sequence = getSequence(requestId); | ||
| mBlockManager.pinBlocks(sequence); | ||
| auto seqPtr = getSequence(requestId); | ||
| TLLM_CHECK_WITH_INFO(seqPtr, "No sequence found for request %lu", requestId); | ||
| auto seqLock = seqPtr->getLock(); | ||
| if (seqPtr->isRemoved()) | ||
| { | ||
| return; | ||
| } | ||
| mBlockManager.pinBlocks(*seqPtr); | ||
| } | ||
|
|
||
| void KVCacheManager::unpinBlocksById(std::vector<KVCacheBlock::IdType> const& blockIds) | ||
|
|
@@ -2688,7 +2727,9 @@ tle::RetentionPriority KVCacheManager::getPriorityByBlockId(KVCacheBlock::IdType | |
|
|
||
| SizeType32 KVCacheManager::copyBlockOffsets(ITensor& output, SizeType32 outputSlotOffset, RequestIdType requestId) const | ||
| { | ||
| auto const& sequence = getSequence(requestId); | ||
| auto const seqPtr = getSequence(requestId); | ||
| TLLM_CHECK_WITH_INFO(seqPtr, "No sequence found for request %lu", requestId); | ||
| auto const& sequence = *seqPtr; | ||
| auto const beamWidth = sequence.getBeamWidth(); | ||
|
|
||
| auto* dstPtr = bufferCast<tk::KVCacheIndex>(output); | ||
|
|
@@ -2998,10 +3039,8 @@ BlocksPerWindow BaseKVCacheManager::calculateMaxNumBlocks(executor::KvCacheConfi | |
| return windowSizeToBlocks; | ||
| } | ||
|
|
||
| void KVCacheManager::removeToken(RequestIdType requestId) | ||
| void KVCacheManager::removeTokenLocked(GenerationRequest& sequence) | ||
| { | ||
| // TODO: add streamLLM support | ||
| auto& sequence = getSequence(requestId); | ||
| if (sequence.getNumTokens() == 0) | ||
|
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Add TLLM_CHECK to ensure lock acquired. Or encapsulate the lock acquisition and token removal loop in the same function to fool-proof this function to be mistakenly called.
Collaborator
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I think this would defeat the purpose of this method. The method declaration states that it should not be called unless caller already owns the lock. It is a private method that is only used in some loops where the caller already owns the lock, calling std::mutex::lock again would cause a deadlock. There is no way to check lock status, calling try_lock when thread already owns the lock would cause undefined behavior.
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I think you mistaken a bit. |
||
| { | ||
| return; | ||
|
|
@@ -3018,36 +3057,48 @@ void KVCacheManager::removeToken(RequestIdType requestId) | |
| } | ||
| } | ||
|
|
||
| void KVCacheManager::rewindKVCache(RequestIdType requestId, SizeType32 rewindLengths) | ||
| void KVCacheManager::removeToken(RequestIdType requestId) | ||
| { | ||
| // Check if the sequence still exists before rewinding | ||
| // In overlap mode with MTP, the request may have been terminated and removed | ||
| // from mSequences before rewindKVCache is called | ||
| // TODO: add streamLLM support | ||
| auto seqPtr = getSequence(requestId); | ||
| if (!seqPtr) | ||
| { | ||
| std::scoped_lock lck(mSequencesMtx); | ||
| if (mSequences.find(requestId) == mSequences.end()) | ||
| { | ||
| TLLM_LOG_DEBUG("Request %lu has already been removed from KV cache manager, skipping rewind", requestId); | ||
| return; | ||
| } | ||
| return; | ||
| } | ||
|
|
||
| for (SizeType32 si = 0; si < rewindLengths; ++si) | ||
| auto seqLock = seqPtr->getLock(); | ||
| if (seqPtr->isRemoved()) | ||
| { | ||
| removeToken(requestId); | ||
| return; | ||
| } | ||
| removeTokenLocked(*seqPtr); | ||
| } | ||
|
|
||
| GenerationRequest const& KVCacheManager::getSequence(RequestIdType requestId) const | ||
| void KVCacheManager::rewindKVCache(RequestIdType requestId, SizeType32 rewindLengths) | ||
| { | ||
| auto lck = std::scoped_lock(mSequencesMtx); | ||
| return mSequences.at(requestId); | ||
| // Hold the per-sequence lock for the entire loop so removeSequence cannot | ||
| // slip in between iterations and cause a partial rewind. | ||
| auto seqPtr = getSequence(requestId); | ||
| if (!seqPtr) | ||
| { | ||
| TLLM_LOG_DEBUG("Request %lu has already been removed from KV cache manager, skipping rewind", requestId); | ||
| return; | ||
| } | ||
| auto seqLock = seqPtr->getLock(); | ||
| if (seqPtr->isRemoved()) | ||
| { | ||
| return; | ||
| } | ||
| for (SizeType32 si = 0; si < rewindLengths; ++si) | ||
| { | ||
| removeTokenLocked(*seqPtr); | ||
| } | ||
| } | ||
|
|
||
| GenerationRequest& KVCacheManager::getSequence(RequestIdType requestId) | ||
| std::shared_ptr<GenerationRequest> KVCacheManager::getSequence(RequestIdType requestId) const | ||
| { | ||
| auto lck = std::scoped_lock(mSequencesMtx); | ||
| return mSequences.at(requestId); | ||
| auto it = mSequences.find(requestId); | ||
| return it != mSequences.end() ? it->second : nullptr; | ||
| } | ||
|
|
||
| SizeType32 BaseKVCacheManager::getSinkBubbleLength(SizeType32 sinkTokenLen, SizeType32 tokensPerBlock) | ||
|
|
@@ -3060,7 +3111,9 @@ SizeType32 BaseKVCacheManager::getSinkBubbleLength(SizeType32 sinkTokenLen, Size | |
| std::vector<std::vector<SizeType32>> const& KVCacheManager::getCacheBlockIds( | ||
| RequestIdType requestId, SizeType32 windowSize) const | ||
| { | ||
| return getSequence(requestId).getCacheBlockIds(windowSize); | ||
| auto seqPtr = getSequence(requestId); | ||
| TLLM_CHECK_WITH_INFO(seqPtr, "No sequence found for request %lu", requestId); | ||
| return seqPtr->getCacheBlockIds(windowSize); | ||
| } | ||
|
Comment on lines
+3130
to
3133
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Line 3072 returns 🤖 Prompt for AI Agents
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The concern looks legit.
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🧩 Analysis chain
🏁 Script executed: Length of output: 55400 🏁 Script executed: Length of output: 75259
1. In - [[nodiscard]] virtual std::vector<std::vector<SizeType32>> const& getCacheBlockIds(
+ [[nodiscard]] virtual std::vector<std::vector<SizeType32>> getCacheBlockIds(
LlmRequest::RequestIdType requestId, SizeType32 windowSize) const
= 0;In - std::vector<std::vector<SizeType32>> const& getCacheBlockIds(
+ std::vector<std::vector<SizeType32>> getCacheBlockIds(
LlmRequest::RequestIdType requestId, SizeType32 windowSize) const override;2. -std::vector<std::vector<SizeType32>> const& KVCacheManager::getCacheBlockIds(
+std::vector<std::vector<SizeType32>> KVCacheManager::getCacheBlockIds(
RequestIdType requestId, SizeType32 windowSize) const
{
auto seqPtr = getSequence(requestId);
TLLM_CHECK_WITH_INFO(seqPtr, "No sequence found for request %lu", requestId);
- return seqPtr->getCacheBlockIds(windowSize);
+ return seqPtr->getCacheBlockIds(windowSize); // copy made while seqPtr is alive
}3. - std::vector<std::vector<SizeType32>> const& getCacheBlockIds(
+ std::vector<std::vector<SizeType32>> getCacheBlockIds(
tb::LlmRequest::RequestIdType requestId, SizeType32 windowSize) const override
{
NB_OVERRIDE_PURE(getCacheBlockIds, requestId, windowSize);
}4. This one needs special attention. After the fix, - auto const& allBlockIds = kvCacheManager.getCacheBlockIds(requestId, maxAttentionWindow)[0];
+ auto allBlockIds = kvCacheManager.getCacheBlockIds(requestId, maxAttentionWindow)[0];Failed to handle agent chat message. Please try again.
Collaborator
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. It is a legit concern.
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
|
|
||
| std::vector<std::vector<std::vector<SizeType32>>> KVCacheManager::getBatchCacheBlockIds( | ||
|
|
@@ -3070,22 +3123,27 @@ std::vector<std::vector<std::vector<SizeType32>>> KVCacheManager::getBatchCacheB | |
| result.reserve(requestIds.size()); | ||
| for (auto const& requestId : requestIds) | ||
| { | ||
| auto const& sequence = getSequence(requestId); | ||
| result.emplace_back(sequence.getCacheBlockIds(windowSize)); | ||
| auto const seqPtr = getSequence(requestId); | ||
| TLLM_CHECK_WITH_INFO(seqPtr, "No sequence found for request %lu", requestId); | ||
| result.emplace_back(seqPtr->getCacheBlockIds(windowSize)); | ||
| } | ||
| return result; | ||
| } | ||
|
|
||
| std::optional<KVCacheBlock::IdType> KVCacheManager::getLastBlockId(LlmRequest::RequestIdType requestId) const | ||
| { | ||
| auto const& seq = getSequence(requestId); | ||
| auto const seq = getSequence(requestId); | ||
| if (!seq) | ||
| { | ||
| return std::nullopt; | ||
| } | ||
| // Use the first window size | ||
| auto firstWindowSize = mBlockManager.getFirstWindowSize(); | ||
| if (firstWindowSize == 0) | ||
| { | ||
| return std::nullopt; | ||
| } | ||
| auto const& perBeam = seq.getCacheBlockIds(firstWindowSize); | ||
| auto const& perBeam = seq->getCacheBlockIds(firstWindowSize); | ||
| if (perBeam.empty() || perBeam[0].empty()) | ||
| { | ||
| return std::nullopt; | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit: Why the extra
_in the variable name?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It was generated like that by Claude.