diff --git a/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/AGENTS.md b/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/AGENTS.md index a8a5ab5a6875..72e274d7289f 100644 --- a/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/AGENTS.md +++ b/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/AGENTS.md @@ -65,14 +65,34 @@ must eventually be `close()`d. Its normal flow is: for later requests; `stopCommitting()` finalizes this process. 5. Suspend or close the request, returning pages to holding/eviction ownership. -`LifeCycleRegistry` maps model layers to lifecycle groups. Attention lifecycles +`LifeCycleRegistry` maps layer buffers to lifecycle groups. Attention lifecycles may have sliding-window and sink-token rules; SSM lifecycles represent a recurrent-state checkpoint. Both layer configs expose `reusePolicy`: `REQUIRED` constrains the prefix match, `OPTIONAL` offers state at the required endpoint for explicit selection on first resume, and `PRIVATE` never publishes reusable pages. -All buffers within a layer share that policy. SSM lifecycles are deduplicated by -policy, so a manager can have up to three SSM groups; iterate `ssmLifeCycleIds()` -for working state and use `requiredSsmLifeCycleId()` only for prefix pruning. +A buffer may override its layer's default through `BufferConfig::reusePolicy`. +This applies to attention and SSM buffers, so one model layer may span several +lifecycle groups. Use `getBufferGroupId(layerId, role)` and `bufferGrouping()` +(`get_buffer_group_id` and `buffer_grouping` in Python) for exact membership. +The legacy `getLayerGroupId(layerId)` and `layerGrouping()` are deprecated in +documentation only: uniform layers retain their results, while a split layer +raises an ambiguity error (for the queried layer or the entire partition, +respectively). No deprecation warnings are emitted. + +Page conversion reads `BufferAttr::scratchPagesPerBlock`, counting only the +layer's buffers in that coalesced slot. Scratch capacity uses the maximum exact +layer share across pools within each lifecycle. Layer counts are temporary +layout calculations, not persistent `(layer, lifecycle)` attributes. + +Execution backends that share a page or state index across coupled buffers must +validate that they use the same policy; independent policies require independent +index tables. In particular, the default attention view couples K/V and scales, +and Mamba kernels couple recurrent, convolution, PLE, and replay state. Native +KVCM supports independent policies for callers that provide those indices. + +SSM lifecycles are deduplicated by policy, so a manager can have up to three SSM +groups; iterate `ssmLifeCycleIds()` for working state and use +`requiredSsmLifeCycleId()` only for prefix pruning. Optional SSM checkpoints require an exact endpoint match, including partial blocks. Fresh or unselected recurrent state must be initialized or reconstructed by the caller before inference. A pool-group index is a storage-layout index and is diff --git a/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/config.h b/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/config.h index 7b84dae436b6..dbd993e86c9d 100644 --- a/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/config.h +++ b/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/config.h @@ -130,8 +130,21 @@ inline std::optional cacheTierMaxQuota(CacheTierConfig const& cfg) return host != nullptr ? host->maxQuota : std::nullopt; } +//! Controls whether a lifecycle participates in prefix matching and page reuse. +enum class ReusePolicy +{ + //! Commit reusable pages and require live-window coverage when matching a prefix. + REQUIRED = 0, + //! Keep request-local working pages; never publish or reuse them through the radix tree. + PRIVATE = 1, + //! Commit reusable pages without constraining the REQUIRED prefix match. The caller selects + //! available groups at resume; missing or unselected groups use fresh pages for recomputation. + //! These pages may be evicted independently without removing the REQUIRED prefix. + OPTIONAL = 2, +}; + // --------------------------------------------------------------------------- -// Buffer configuration (one KV buffer inside an attention layer). +// Buffer configuration (one buffer inside an attention or SSM layer). // --------------------------------------------------------------------------- struct BufferConfig @@ -148,6 +161,15 @@ struct BufferConfig //! Buffers may coalesce only when their group and expanded size match. //! This does not change their lifecycle or shared slot allocation. int coalescingGroup = 0; + + //! Reuse policy of this buffer; nullopt inherits the layer's policy. Overrides can + //! place buffers of one attention or SSM layer in different lifecycle groups. + std::optional reusePolicy; + + [[nodiscard]] ReusePolicy effectiveReusePolicy(ReusePolicy layerPolicy) const noexcept + { + return reusePolicy.value_or(layerPolicy); + } }; // --------------------------------------------------------------------------- @@ -178,19 +200,6 @@ inline void validateNoDuplicateBufferRoles(std::vector const& buff } // namespace detail -//! Controls whether a layer lifecycle participates in prefix matching and page reuse. -enum class ReusePolicy -{ - //! Commit reusable pages and require live-window coverage when matching a prefix. - REQUIRED = 0, - //! Keep request-local working pages; never publish or reuse them through the radix tree. - PRIVATE = 1, - //! Commit reusable pages without constraining the REQUIRED prefix match. The caller selects - //! available groups at resume; missing or unselected groups use fresh pages for recomputation. - //! These pages may be evicted independently without removing the REQUIRED prefix. - OPTIONAL = 2, -}; - struct AttentionLayerConfig { static constexpr LayerType type = LayerType::ATTENTION; @@ -205,6 +214,7 @@ struct AttentionLayerConfig std::optional numSinkTokens; // Private working pages never constrain prefix matching or enter the tree. + // Every buffer of an attention layer uses this policy. ReusePolicy reusePolicy = ReusePolicy::REQUIRED; [[nodiscard]] std::optional windowSize() const noexcept @@ -235,6 +245,7 @@ struct SsmLayerConfig LayerId layerId = 0; std::vector buffers; + // Default policy of the layer's buffers; BufferConfig::reusePolicy may override it per buffer. ReusePolicy reusePolicy = ReusePolicy::REQUIRED; void validate() const @@ -256,6 +267,11 @@ struct SsmLayerConfig using LayerConfig = std::variant; +inline std::vector const& layerBuffers(LayerConfig const& layer) +{ + return std::visit([](auto const& cfg) -> std::vector const& { return cfg.buffers; }, layer); +} + // --------------------------------------------------------------------------- // KVCacheDesc — describes one KV cache request's capacity and history length. // Mirrors _config.py::KVCacheDesc. diff --git a/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/kvCacheManager.cpp b/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/kvCacheManager.cpp index 1ec31e1bc0ae..8bce78489ea3 100644 --- a/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/kvCacheManager.cpp +++ b/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/kvCacheManager.cpp @@ -30,6 +30,7 @@ #include #include #include +#include namespace tensorrt_llm::batch_manager::kv_cache_manager_v2 { @@ -72,10 +73,18 @@ std::vector PageIndexConverter::operator()( // Scratch block: slot IDs come from ScratchDesc, not base_indices. int blockPos = ordinal - scratch->range.beg; int totalOffset = blockPos * scratchPages; + bool const crossesSlots = scale % scratchPages != 0; + if (crossesSlots) + { + // Unequal layer shares can put a block across two slots. Pack + // the layer's consecutive roles before resolving the slot ID, + // so a trailing role cannot wrap over an earlier block's data. + totalOffset += appliedLayerOffset % scratchPages; + } int slotIdx = totalOffset / scale; int slotId = scratch->slotIds[static_cast(slotIdx)]; int offset = totalOffset % scale; - index = slotId * scale + (offset + appliedLayerOffset) % scale; + index = slotId * scale + (crossesSlots ? offset : (offset + appliedLayerOffset) % scale); } else if (baseIndices[toSizeT(ordinal)] == kBadPageIndex.value()) { @@ -333,11 +342,9 @@ bool KvCacheManager::isSparse(LayerId layerId, DataRole role) const PageIndexConverter KvCacheManager::getPageIndexConverter(LayerId layerId, DataRole role) const { auto const& attr = mStorage->getBufferAttr(layerId, role); - auto const& layerAttr = mStorage->getLayerAttr(layerId); int scale = mStorage->mSlotToPageIndices.at(attr.lifeCycleId).at(attr.poolIndex); int offset = exactDiv(static_cast(attr.offset), static_cast(attr.size)); - int scratchPages = layerAttr.slotUtil.at(attr.poolIndex); - return PageIndexConverter{scale, attr.expansion, offset, scratchPages}; + return PageIndexConverter{scale, attr.expansion, offset, attr.scratchPagesPerBlock}; } std::optional KvCacheManager::supportsIndexMode(PageIndexMode mode) const @@ -461,16 +468,49 @@ std::vector KvCacheManager::layerIds() const LayerGroupId KvCacheManager::getLayerGroupId(LayerId layerId) const { - return mStorage->layerToLifeCycleIds().at(layerId); + auto const& lifeCycles = mStorage->layerToLifeCycleIds().at(layerId); + if (lifeCycles.size() != 1) + { + throw std::invalid_argument("Layer " + std::to_string(layerId) + " has buffers in " + + std::to_string(lifeCycles.size()) + + " layer groups because its buffers use different reuse policies; look up its layer group per " + "buffer with get_buffer_group_id(layer_id, role) / getBufferGroupId(layerId, role)"); + } + return lifeCycles.front(); +} + +LayerGroupId KvCacheManager::getBufferGroupId(LayerId layerId, DataRole const& role) const +{ + return mStorage->getBufferAttr(layerId, role).lifeCycleId; } TypedVec> KvCacheManager::layerGrouping() const { LifeCycleId numLc = mLifeCycles.size(); TypedVec> result(numLc); - for (auto const& [lid, lc] : mStorage->layerToLifeCycleIds()) + for (auto const& [lid, lifeCycles] : mStorage->layerToLifeCycleIds()) + { + result.at(getLayerGroupId(lid)).push_back(lid); + } + return result; +} + +TypedVec> KvCacheManager::bufferGrouping() const +{ + TypedVec> result(mLifeCycles.size()); + std::unordered_map> buffersByLayer; + for (auto const& buffer : allBufferIds()) { - result.at(lc).push_back(lid); + buffersByLayer[buffer.layerId].push_back(buffer); + } + // Keep layer order consistent with layerGrouping() for consumers that pair + // group membership with a declared list of global model layer IDs. + for (auto const& [layer, lifeCycles] : mStorage->layerToLifeCycleIds()) + { + for (auto const& buffer : buffersByLayer.at(layer)) + { + result.at(getBufferGroupId(buffer.layerId, buffer.role)).push_back(buffer); + } } return result; } diff --git a/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/kvCacheManager.h b/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/kvCacheManager.h index 98cff3e0d1cf..3b0ac0e7c0de 100644 --- a/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/kvCacheManager.h +++ b/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/kvCacheManager.h @@ -229,15 +229,23 @@ class KvCacheManager : public std::enable_shared_from_this int numLayers() const noexcept; std::vector layerIds() const; + + //! Deprecated: use getBufferGroupId(). Returns the shared group of a uniform layer. + //! Throws std::invalid_argument if the queried layer spans several lifecycle groups. LayerGroupId getLayerGroupId(LayerId layerId) const; - // Layer grouping: layers with the same lifecycle share pool allocation. - // NOTE: the iteration order of the layer lists (and of the groups) is NOT - // part of the API contract and may differ across backends/runs. Do not rely - // on it to infer buffer/pool memory order — query poolGroupDescs() - // (PoolGroupDesc::pools[i].baseAddress + coalescedBuffers) for that. + //! Group of one buffer; throws std::out_of_range for an unknown buffer. + LayerGroupId getBufferGroupId(LayerId layerId, DataRole const& role) const; + + //! Deprecated: use bufferGrouping(). Partitions uniform layers by lifecycle. + //! Throws std::invalid_argument if any layer spans several groups. Member order + //! is unspecified; use poolGroupDescs() to inspect physical buffer ordering. TypedVec> layerGrouping() const; + //! Buffer membership indexed by lifecycle group ID. Each buffer appears once. + //! Member order is unspecified and does not describe physical memory ordering. + TypedVec> bufferGrouping() const; + // Iterator over all buffer identifiers. Mirrors Python's all_buffer_ids property. std::vector allBufferIds() const; diff --git a/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/lifeCycleRegistry.cpp b/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/lifeCycleRegistry.cpp index 706e4ae45068..49ce831d8127 100644 --- a/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/lifeCycleRegistry.cpp +++ b/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/lifeCycleRegistry.cpp @@ -27,22 +27,21 @@ namespace tensorrt_llm::batch_manager::kv_cache_manager_v2 // makeLifeCycle — factory dispatching on LayerConfig variant. // --------------------------------------------------------------------------- -LifeCycle makeLifeCycle(LayerConfig const& layer, int tokensPerBlock) +LifeCycle makeLifeCycle(LayerConfig const& layer, BufferConfig const& buffer, int tokensPerBlock) { return std::visit( [&](auto const& cfg) -> LifeCycle { - cfg.validate(); + auto const policy = buffer.effectiveReusePolicy(cfg.reusePolicy); using T = std::decay_t; if constexpr (std::is_same_v) { - return SsmLifeCycle{cfg.reusePolicy}; + return SsmLifeCycle{policy}; } else { - bool const isSparse = !cfg.buffers.empty() && cfg.buffers.front().isSparse; return AttnLifeCycle::make( - cfg.slidingWindowSize, cfg.numSinkTokens, tokensPerBlock, isSparse, cfg.reusePolicy); + cfg.slidingWindowSize, cfg.numSinkTokens, tokensPerBlock, buffer.isSparse, policy); } }, layer); @@ -54,9 +53,8 @@ LifeCycle makeLifeCycle(LayerConfig const& layer, int tokensPerBlock) LifeCycleRegistry::LifeCycleRegistry(KVCacheManagerConfig const& config) { - for (auto const& layer : config.layers) + auto const registerLifeCycle = [this](LifeCycle const& lc) { - LifeCycle lc = makeLifeCycle(layer, config.tokensPerBlock); if (mLifeCycleIdMap.find(lc) == mLifeCycleIdMap.end()) { check(); @@ -66,6 +64,14 @@ LifeCycleRegistry::LifeCycleRegistry(KVCacheManagerConfig const& config) if (std::holds_alternative(lc)) mSsmLifeCycleIds.push_back(id); } + }; + for (auto const& layer : config.layers) + { + std::visit([](auto const& cfg) { cfg.validate(); }, layer); + for (auto const& buffer : layerBuffers(layer)) + { + registerLifeCycle(makeLifeCycle(layer, buffer, config.tokensPerBlock)); + } } check(); } diff --git a/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/lifeCycleRegistry.h b/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/lifeCycleRegistry.h index 6fc4f9faf95b..8397be0178fa 100644 --- a/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/lifeCycleRegistry.h +++ b/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/lifeCycleRegistry.h @@ -108,9 +108,9 @@ struct AttnLifeCycle }; // --------------------------------------------------------------------------- -// SsmLifeCycle — lifecycle for SSM (State Space Model) layers. +// SsmLifeCycle — lifecycle for SSM (State Space Model) buffers. // All blocks before the last full block are stale (recurrent state). -// SSM layers with the same reuse policy share a lifecycle. +// SSM buffers with the same effective reuse policy share a lifecycle. // --------------------------------------------------------------------------- struct SsmLifeCycle { @@ -178,8 +178,8 @@ inline HalfOpenRange computeScratchRange( return intersect(capStale, inputRange); } -// Factory: create a LifeCycle from a LayerConfig variant. -LifeCycle makeLifeCycle(LayerConfig const& layer, int tokensPerBlock); +// Factory: create the LifeCycle that holds one buffer of a layer, using the buffer's effective reuse policy. +LifeCycle makeLifeCycle(LayerConfig const& layer, BufferConfig const& buffer, int tokensPerBlock); // Integer id assigned to each unique LifeCycle. using LifeCycleId = StrongIndex; diff --git a/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/storage/config.cpp b/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/storage/config.cpp index e4131f77b853..5f1e49dbf0ef 100644 --- a/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/storage/config.cpp +++ b/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/storage/config.cpp @@ -69,6 +69,11 @@ std::map StorageConfig::bufferAttributes() const for (PoolIndex poolIndex{0}; poolIndex < variant.coalescedBuffers.size(); ++poolIndex) { auto const& cb = variant.coalescedBuffers[poolIndex]; + std::unordered_map buffersPerLayer; + for (auto const& bufId : cb.bufferIds) + { + ++buffersPerLayer[bufId.layerId]; + } size_t offset = 0; for (auto const& bufId : cb.bufferIds) { @@ -77,7 +82,8 @@ std::map StorageConfig::bufferAttributes() const if (it != expansion.end()) exp = it->second; - ret[bufId] = BufferAttr{lcId, poolIndex, offset, cb.singleBufferSize, exp}; + ret[bufId] = BufferAttr{ + lcId, poolIndex, offset, cb.singleBufferSize, exp, buffersPerLayer.at(bufId.layerId)}; offset += cb.singleBufferSize; } } @@ -86,63 +92,22 @@ std::map StorageConfig::bufferAttributes() const return ret; } -std::unordered_map StorageConfig::layerToLifeCycleIds() const +std::unordered_map> StorageConfig::layerToLifeCycleIds() const { - std::unordered_map map; + std::unordered_map> map; for (auto const& [bufId, attr] : bufferAttributes()) { - auto [it, inserted] = map.emplace(bufId.layerId, attr.lifeCycleId); - if (!inserted) + auto& ids = map[bufId.layerId]; + if (std::find(ids.begin(), ids.end(), attr.lifeCycleId) == ids.end()) { - TLLM_CHECK_DEBUG(it->second == attr.lifeCycleId); + ids.push_back(attr.lifeCycleId); } } - return map; -} - -std::map StorageConfig::layerAttributes() const -{ - std::map ret; - for (auto const& pg : slotDescList) + for (auto& [layerId, ids] : map) { - for (auto const& variant : pg.variants) - { - LifeCycleId lcId = variant.lifeCycleId; - for (PoolIndex poolIndex{0}; poolIndex < variant.coalescedBuffers.size(); ++poolIndex) - { - auto const& cb = variant.coalescedBuffers[poolIndex]; - // Count how many buffers per layer in this coalesced buffer. - std::unordered_map slotUtilPerLayer; - for (auto const& bufId : cb.bufferIds) - { - slotUtilPerLayer[bufId.layerId] += 1; - } - int buffersPerSlot = cb.numBuffers(); - - for (auto const& [layerId, count] : slotUtilPerLayer) - { - auto it = ret.find(layerId); - if (it == ret.end()) - { - LayerAttr attr; - attr.lifeCycleId = lcId; - attr.slotUtil.resize(variant.coalescedBuffers.size(), 0); - attr.slotUtilFracMax = Rational{0, 1}; - it = ret.emplace(layerId, std::move(attr)).first; - } - auto& attr = it->second; - TLLM_CHECK_DEBUG(attr.lifeCycleId == lcId); - attr.slotUtil[poolIndex] = count; - Rational frac{count, buffersPerSlot}; - if (frac > attr.slotUtilFracMax) - { - attr.slotUtilFracMax = frac; - } - } - } - } + std::sort(ids.begin(), ids.end()); } - return ret; + return map; } // --------------------------------------------------------------------------- @@ -165,24 +130,18 @@ StorageConfig createStorageConfig(KVCacheManagerConfig const& config) for (auto const& layer : config.layers) { - LifeCycle lc = makeLifeCycle(layer, tokensPerBlock); - LifeCycleId lcId = registry.getId(lc); - - std::visit( - [&](auto const& cfg) - { - for (auto const& buf : cfg.buffers) - { - int tpbo = buf.tokensPerBlockOverride.value_or(tokensPerBlock); - int exp = exactDiv(tokensPerBlock, tpbo); - size_t expandedSize = buf.size * static_cast(exp); - - BufferId bid{cfg.layerId, buf.role}; - expansionMap[bid] = exp; - bufferGroups[lcId][{buf.coalescingGroup, expandedSize}].push_back(bid); - } - }, - layer); + LayerId const layerId = std::visit([](auto const& cfg) { return cfg.layerId; }, layer); + for (auto const& buf : layerBuffers(layer)) + { + LifeCycleId const lcId = registry.getId(makeLifeCycle(layer, buf, tokensPerBlock)); + int tpbo = buf.tokensPerBlockOverride.value_or(tokensPerBlock); + int exp = exactDiv(tokensPerBlock, tpbo); + size_t expandedSize = buf.size * static_cast(exp); + + BufferId bid{layerId, buf.role}; + expansionMap[bid] = exp; + bufferGroups[lcId][{buf.coalescingGroup, expandedSize}].push_back(bid); + } } // Build one SlotDescVariant per life cycle. diff --git a/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/storage/config.h b/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/storage/config.h index ef7fee8ec1b6..8564d25e6046 100644 --- a/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/storage/config.h +++ b/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/storage/config.h @@ -28,6 +28,7 @@ #include #include #include +#include #include namespace tensorrt_llm::batch_manager::kv_cache_manager_v2 @@ -164,14 +165,14 @@ struct BufferAttr { LifeCycleId lifeCycleId{0}; PoolIndex poolIndex{0}; - size_t offset = 0; // byte offset within the slot - size_t size = 0; // expanded size of the buffer (after expansion) - int expansion = 1; // expansion factor (tokens_per_block / tokens_per_block_override) + size_t offset = 0; // byte offset within the slot + size_t size = 0; // expanded size of the buffer (after expansion) + int expansion = 1; // expansion factor (tokens_per_block / tokens_per_block_override) + int scratchPagesPerBlock = 1; // buffers of this layer in the same coalesced slot }; // --------------------------------------------------------------------------- // Rational — exact fraction for slot utilization computations. -// Mirrors Python's fractions.Fraction used in _storage/_config.py::LayerAttr. // --------------------------------------------------------------------------- struct Rational { @@ -196,22 +197,6 @@ struct Rational } }; -// --------------------------------------------------------------------------- -// LayerAttr — per-layer storage attributes for scratch slot management. -// Mirrors _storage/_config.py::LayerAttr. -// --------------------------------------------------------------------------- -struct LayerAttr -{ - LifeCycleId lifeCycleId{0}; - - // Number of sub-pages within a single coalesced slot that belong to this layer, - // per pool within the pool group. - TypedVec slotUtil; - - // Fraction of slot_util to total number of buffers in the slot, max over all pools. - Rational slotUtilFracMax; -}; - // --------------------------------------------------------------------------- // StorageConfig — complete storage layout for the KV cache. // Mirrors _storage/_config.py::StorageConfig. @@ -230,12 +215,9 @@ struct StorageConfig // Attribute map for each buffer. std::map bufferAttributes() const; - // Map LayerId → LifeCycleId. - std::unordered_map layerToLifeCycleIds() const; - - // Per-layer storage attributes (slot utilization within coalesced slots). - // Mirrors _storage/_config.py::StorageConfig.layer_attributes(). - std::map layerAttributes() const; + // Map LayerId → the lifecycles of its buffers, in ascending order. A layer spans several + // lifecycles when its buffers use different reuse policies. + std::unordered_map> layerToLifeCycleIds() const; LifeCycleId numLifeCycles() const; }; diff --git a/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/storageManager.cpp b/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/storageManager.cpp index 4d2f7cf740f5..a336d836435a 100644 --- a/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/storageManager.cpp +++ b/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/storageManager.cpp @@ -230,15 +230,30 @@ StorageManager::StorageManager(LifeCycleRegistry const& lifeCycles, StorageConfi // their page sizes. Nothing may read a cold level before then. TLLM_CHECK(appendLevelSlotDescList(config.slotDescList) == kHotLevel); - // Compute layer attributes and slot utilization fractions for scratch support. - mLayerAttributes = config.layerAttributes(); + // A scratch slot serves one layer at a time. Reserve the largest layer share + // across the coalesced pools of each lifecycle, retaining exact fractions. mSlotUtilFracMax.resize(lifeCycles.size(), Rational{0, 1}); - for (auto const& [layerId, layerAttr] : mLayerAttributes) + for (auto const& slotDesc : config.slotDescList) { - LifeCycleId const lcIdx = layerAttr.lifeCycleId; - if (layerAttr.slotUtilFracMax > mSlotUtilFracMax[lcIdx]) + for (auto const& variant : slotDesc.variants) { - mSlotUtilFracMax[lcIdx] = layerAttr.slotUtilFracMax; + for (auto const& buffers : variant.coalescedBuffers) + { + std::unordered_map counts; + for (auto const& buffer : buffers.bufferIds) + { + ++counts[buffer.layerId]; + } + for (auto const& [layer, count] : counts) + { + Rational const fraction{count, buffers.numBuffers()}; + auto& maximum = mSlotUtilFracMax[variant.lifeCycleId]; + if (fraction > maximum) + { + maximum = fraction; + } + } + } } } @@ -1764,14 +1779,6 @@ TypedVec StorageManager::poolGroupDescs() const return result; } -LayerAttr const& StorageManager::getLayerAttr(LayerId layerId) const -{ - auto it = mLayerAttributes.find(layerId); - if (it == mLayerAttributes.end()) - throw std::out_of_range("Unknown LayerId for LayerAttr"); - return it->second; -} - SlotCount StorageManager::numSlots(PoolGroupIndex pgIdx, CacheLevel level) const { return mLevels.at(level).storage->numSlots(pgIdx); diff --git a/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/storageManager.h b/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/storageManager.h index 2b423590aeb7..d74ebf2fdd26 100644 --- a/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/storageManager.h +++ b/cpp/tensorrt_llm/batch_manager/kv_cache_manager_v2/storageManager.h @@ -306,9 +306,6 @@ class StorageManager : public std::enable_shared_from_this // Pool group base address without per-layer offset. MemAddress getMemPoolBaseAddress(PoolGroupIndex pgIdx, PoolIndex poolIdx) const; - // Per-layer storage attributes. - LayerAttr const& getLayerAttr(LayerId layerId) const; - // Address of a slot's buffer in a specific pool at a cache level. Address slotAddress(CacheLevel level, PoolGroupIndex pgIdx, SlotId slotId, PoolIndex poolIdx) const; @@ -325,8 +322,8 @@ class StorageManager : public std::enable_shared_from_this // Pool-group slot count (number of pages). SlotCount numSlots(PoolGroupIndex pgIdx, CacheLevel level = kHotLevel) const; - // Layer-to-lifecycle mapping (for KvCacheManager queries). - std::unordered_map const& layerToLifeCycleIds() const noexcept + // Layer-to-lifecycles mapping (for KvCacheManager queries); see StorageConfig::layerToLifeCycleIds(). + std::unordered_map> const& layerToLifeCycleIds() const noexcept { return mLayerToLifeCycleIds; } @@ -433,14 +430,11 @@ class StorageManager : public std::enable_shared_from_this TypedVec mBatchingLayerGroupIds; // Codec-selected PageIndexPair memory location for each lifecycle. TypedVec mPageIndexLocations; - std::unordered_map mLayerToLifeCycleIds; + std::unordered_map> mLayerToLifeCycleIds; // slot-to-page-index scale factors: [lcId][poolIdx] TypedVec> mSlotToPageIndices; - // Per-layer storage attributes for scratch slot management. - std::map mLayerAttributes; - // Max slot utilization fraction per lifecycle (across all layers in that lifecycle). TypedVec mSlotUtilFracMax; diff --git a/cpp/tensorrt_llm/nanobind/batch_manager/kvCacheManagerV2.cpp b/cpp/tensorrt_llm/nanobind/batch_manager/kvCacheManagerV2.cpp index 57e6d5049f29..c7824b955665 100644 --- a/cpp/tensorrt_llm/nanobind/batch_manager/kvCacheManagerV2.cpp +++ b/cpp/tensorrt_llm/nanobind/batch_manager/kvCacheManagerV2.cpp @@ -1617,14 +1617,15 @@ void KvCacheManagerV2Bindings::initBindings(nb::module_& m) .def("validate", &kv::DiskCacheTierConfig::validate) DEF_COPY(kv::DiskCacheTierConfig); nb::class_(m, "BufferConfig") - .def(nb::init, bool, int>(), nb::arg("role"), nb::arg("size"), - nb::arg("tokens_per_block_override") = std::nullopt, nb::arg("is_sparse") = false, - nb::arg("coalescing_group") = 0) + .def(nb::init, bool, int, std::optional>(), + nb::arg("role"), nb::arg("size"), nb::arg("tokens_per_block_override") = std::nullopt, + nb::arg("is_sparse") = false, nb::arg("coalescing_group") = 0, nb::arg("reuse_policy") = std::nullopt) .def_rw("role", &kv::BufferConfig::role) .def_rw("size", &kv::BufferConfig::size) .def_rw("tokens_per_block_override", &kv::BufferConfig::tokensPerBlockOverride) .def_rw("is_sparse", &kv::BufferConfig::isSparse) - .def_rw("coalescing_group", &kv::BufferConfig::coalescingGroup) DEF_COPY(kv::BufferConfig); + .def_rw("coalescing_group", &kv::BufferConfig::coalescingGroup) + .def_rw("reuse_policy", &kv::BufferConfig::reusePolicy) DEF_COPY(kv::BufferConfig); nb::class_(m, "AttentionLayerConfig") .def(nb::init, std::optional, std::optional, @@ -2936,14 +2937,22 @@ void KvCacheManagerV2Bindings::initBindings(nb::module_& m) .def_prop_ro("layer_ids", &kv::KvCacheManager::layerIds) .def_prop_ro( "layer_grouping", [](kv::KvCacheManager const& self) { return self.layerGrouping().raw(); }, - "Layers grouped by shared lifecycle/pool allocation. The iteration order of the " - "layer lists (and of the groups) is NOT an API contract and may differ across " - "backends/runs; do not rely on it for buffer/pool memory order -- use " - "pool_group_descs (PoolGroupDesc.pools[i].base_address + coalesced_buffers) instead.") + "Deprecated: use buffer_grouping. Partitions uniform layers by lifecycle; raises ValueError " + "if any layer spans several groups. Member order does not describe physical memory order.") + .def_prop_ro( + "buffer_grouping", [](kv::KvCacheManager const& self) { return self.bufferGrouping().raw(); }, + "Buffer membership indexed by lifecycle group ID. Each buffer appears once; member order is unspecified.") .def( "get_layer_group_id", [](kv::KvCacheManager const& self, kv::LayerId layerId) { return self.getLayerGroupId(layerId).value(); }, - nb::arg("layer_id")) + nb::arg("layer_id"), + "Deprecated: use get_buffer_group_id. Returns a uniform layer's group; raises ValueError for a split " + "layer.") + .def( + "get_buffer_group_id", + [](kv::KvCacheManager const& self, kv::LayerId layerId, kv::DataRole const& role) + { return self.getBufferGroupId(layerId, role).value(); }, + nb::arg("layer_id"), nb::arg("role"), "Lifecycle group of one buffer of the layer.") .def("get_page_index_converter", &kv::KvCacheManager::getPageIndexConverter, nb::arg("layer_id"), nb::arg("data_role")) .def( diff --git a/cpp/tests/unit_tests/batch_manager/kvCacheManagerV2ConfigTest.cpp b/cpp/tests/unit_tests/batch_manager/kvCacheManagerV2ConfigTest.cpp index 99756c07befc..44f0afc77fcb 100644 --- a/cpp/tests/unit_tests/batch_manager/kvCacheManagerV2ConfigTest.cpp +++ b/cpp/tests/unit_tests/batch_manager/kvCacheManagerV2ConfigTest.cpp @@ -23,7 +23,9 @@ #include #include +#include #include +#include namespace { @@ -162,4 +164,76 @@ TEST(KvCacheManagerV2ConfigTest, SparseBuffersCoalesceByExpandedSize) EXPECT_EQ(second.expansion, 1); } +TEST(KvCacheManagerV2ConfigTest, AttentionBufferPoliciesOverrideLayerDefault) +{ + auto config = makeConfig(); + auto& buffer = std::get(config.layers.front()).buffers.front(); + buffer.reusePolicy = ReusePolicy::REQUIRED; + EXPECT_NO_THROW(config.validate()); + EXPECT_NO_THROW(createStorageConfig(config)); + for (auto const policy : {ReusePolicy::PRIVATE, ReusePolicy::OPTIONAL}) + { + buffer.reusePolicy = policy; + EXPECT_NO_THROW(config.validate()); + auto const storage = createStorageConfig(config); + LifeCycleRegistry const registry(config); + auto const attr = storage.bufferAttributes().at(BufferId{0, buffer.role}); + EXPECT_EQ(std::get(registry[attr.lifeCycleId]).reusePolicy, policy); + } +} + +TEST(KvCacheManagerV2ConfigTest, SsmBufferReusePoliciesSplitLayerLifecycles) +{ + auto config = makeTieredConfig(); + config.commitMinSnapshot = true; + for (LayerId layerId = 1; layerId <= 2; ++layerId) + { + config.layers.push_back(SsmLayerConfig{.layerId = layerId, + .buffers = {BufferConfig{.role = "state", .size = 4096}, + BufferConfig{.role = "replay", .size = 2048, .reusePolicy = ReusePolicy::PRIVATE}, + BufferConfig{.role = "draft", .size = 2048, .reusePolicy = ReusePolicy::OPTIONAL}}}); + } + // A layer-level PRIVATE layer shares the lifecycle of PRIVATE buffer overrides. + config.layers.push_back(SsmLayerConfig{ + .layerId = 3, .buffers = {BufferConfig{.role = "replay", .size = 2048}}, .reusePolicy = ReusePolicy::PRIVATE}); + ASSERT_NO_THROW(config.validate()); + + LifeCycleRegistry const registry(config); + auto const storage = createStorageConfig(config); + auto const attributes = storage.bufferAttributes(); + auto lifeCycle = [&](LayerId layerId, DataRole const& role) { + return attributes.at(BufferId{layerId, role}).lifeCycleId; + }; + auto const required = lifeCycle(1, "state"); + auto const privateLc = lifeCycle(1, "replay"); + auto const optional = lifeCycle(1, "draft"); + + // Lifecycles are registered in buffer order, after the attention layer's. + EXPECT_EQ(registry.size(), LifeCycleId{4}); + EXPECT_EQ(storage.numLifeCycles(), LifeCycleId{4}); + EXPECT_EQ(required, LifeCycleId{1}); + EXPECT_EQ(privateLc, LifeCycleId{2}); + EXPECT_EQ(optional, LifeCycleId{3}); + EXPECT_EQ(registry.ssmLifeCycleIds(), (std::vector{required, privateLc, optional})); + EXPECT_EQ(registry.requiredSsmLifeCycleId(), required); + EXPECT_EQ(std::get(registry[privateLc]).reusePolicy, ReusePolicy::PRIVATE); + EXPECT_EQ(std::get(registry[optional]).reusePolicy, ReusePolicy::OPTIONAL); + EXPECT_EQ(lifeCycle(2, "state"), required); + EXPECT_EQ(lifeCycle(2, "replay"), privateLc); + EXPECT_EQ(lifeCycle(2, "draft"), optional); + EXPECT_EQ(lifeCycle(3, "replay"), privateLc); + + auto const layerLifeCycles = storage.layerToLifeCycleIds(); + EXPECT_EQ(layerLifeCycles.size(), 4U); + EXPECT_EQ(layerLifeCycles.at(0), (std::vector{LifeCycleId{0}})); + EXPECT_EQ(layerLifeCycles.at(1), (std::vector{required, privateLc, optional})); + EXPECT_EQ(layerLifeCycles.at(2), (std::vector{required, privateLc, optional})); + EXPECT_EQ(layerLifeCycles.at(3), (std::vector{privateLc})); + + auto const grouping = storage.lifeCycleGrouping(); + EXPECT_NE(grouping[required], grouping[privateLc]); + EXPECT_NE(grouping[required], grouping[optional]); + EXPECT_NE(grouping[privateLc], grouping[optional]); +} + } // namespace diff --git a/cpp/tests/unit_tests/batch_manager/kvCacheManagerV2StatsTest.cpp b/cpp/tests/unit_tests/batch_manager/kvCacheManagerV2StatsTest.cpp index 952d7e566e84..738d7a5b3a9b 100644 --- a/cpp/tests/unit_tests/batch_manager/kvCacheManagerV2StatsTest.cpp +++ b/cpp/tests/unit_tests/batch_manager/kvCacheManagerV2StatsTest.cpp @@ -30,9 +30,13 @@ #include #include #include +#include #include #include +#include #include +#include +#include #include namespace @@ -410,6 +414,334 @@ TEST_F(ReplayCacheTest, OptionalSsmEvictionPreservesRequiredPrefix) manager->shutdown(); } +// One buffer of an SSM layer whose recurrent state is REQUIRED while its replay buffers override the +// layer's policy with PRIVATE. +struct SplitSsmBuffer +{ + LayerId layer; + DataRole role; + size_t size; + bool isPrivate; + // Distinct fill byte of this buffer. + uint8_t pattern; +}; + +// Layers 1-3 each keep two REQUIRED state roles and two PRIVATE replay roles. Layer 1 also owns a +// PRIVATE role the other layers lack, like a request-wide counter stored once. +std::vector splitSsmBuffers() +{ + std::vector buffers; + for (LayerId layer = 1; layer <= 3; ++layer) + { + auto const pattern = [layer](int role) { return static_cast(layer * 32 + role * 4); }; + buffers.push_back({layer, "state", 4096, false, pattern(0)}); + buffers.push_back({layer, "conv", 4096, false, pattern(1)}); + buffers.push_back({layer, "old_x", 4096, true, pattern(2)}); + buffers.push_back({layer, "old_dt", 1024, true, pattern(3)}); + if (layer == 1) + { + buffers.push_back({layer, "count", 512, true, pattern(4)}); + } + } + return buffers; +} + +KVCacheManagerConfig makeSplitSsmConfig(bool withHostTier) +{ + auto config = makeConfig(); + config.commitMinSnapshot = true; + config.enablePartialCommit = false; + if (withHostTier) + { + config.cacheTiers.emplace_back(HostCacheTierConfig{4 << 20}); + } + std::map layers; + for (auto const& buffer : splitSsmBuffers()) + { + auto& layer = layers[buffer.layer]; + layer.layerId = buffer.layer; + layer.buffers.push_back(BufferConfig{.role = buffer.role, + .size = buffer.size, + .reusePolicy = buffer.isPrivate ? std::optional{ReusePolicy::PRIVATE} : std::nullopt}); + } + for (auto& [layerId, layer] : layers) + { + config.layers.emplace_back(std::move(layer)); + } + return config; +} + +// Address of a buffer within the SSM page that `beam` of `cache` currently holds. +MemAddress ssmBufferAddress(KvCacheManager const& manager, KvCache const& cache, SplitSsmBuffer const& buffer, int beam) +{ + auto const slot = cache.getSsmBlockBaseIndex(manager.getBufferGroupId(buffer.layer, buffer.role), BeamIndex{beam}); + EXPECT_GE(slot, 0); + return manager.getMemPoolBaseAddress(buffer.layer, buffer.role) + + static_cast(slot) * manager.getPageIndexScale(buffer.layer, buffer.role) + * manager.getPageStride(buffer.layer, buffer.role); +} + +// Fill byte of a buffer, offset by `salt` to tell writers apart. +uint8_t splitSsmPattern(SplitSsmBuffer const& buffer, int salt) +{ + return static_cast(buffer.pattern + salt); +} + +TEST_F(ReplayCacheTest, SsmBufferPolicySplitsLayerAcrossGroups) +{ + auto manager = std::make_shared(makeSplitSsmConfig(/*withHostTier=*/true)); + auto const attention = manager->getLayerGroupId(LayerId{0}); + auto const required = manager->getBufferGroupId(1, "state"); + auto const privateGroup = manager->getBufferGroupId(1, "old_x"); + EXPECT_EQ(manager->getBufferGroupId(0, "key"), attention); + EXPECT_EQ(manager->lifeCycles().size(), LifeCycleId{3}); + EXPECT_NE(required, privateGroup); + EXPECT_TRUE(requiresPrefixCoverage(manager->lifeCycles()[required])); + EXPECT_TRUE(isPrivateLifeCycle(manager->lifeCycles()[privateGroup])); + for (auto const& buffer : splitSsmBuffers()) + { + SCOPED_TRACE(std::to_string(buffer.layer) + "/" + buffer.role); + EXPECT_EQ(manager->getBufferGroupId(buffer.layer, buffer.role), buffer.isPrivate ? privateGroup : required); + EXPECT_THROW(manager->getLayerGroupId(buffer.layer), std::invalid_argument); + } + EXPECT_THROW(manager->getBufferGroupId(1, "missing"), std::out_of_range); + + EXPECT_EQ(manager->numLayers(), 4); + auto layerIds = manager->layerIds(); + std::sort(layerIds.begin(), layerIds.end()); + EXPECT_EQ(layerIds, (std::vector{0, 1, 2, 3})); + EXPECT_THROW(manager->layerGrouping(), std::invalid_argument); + auto const grouping = manager->bufferGrouping(); + ASSERT_EQ(grouping.size(), LayerGroupId{3}); + size_t count = 0; + for (LayerGroupId group{0}; group < grouping.size(); ++group) + { + for (auto const& buffer : grouping[group]) + { + EXPECT_EQ(manager->getBufferGroupId(buffer.layerId, buffer.role), group); + ++count; + } + } + EXPECT_EQ(count, manager->allBufferIds().size()); + for (auto const level : {kHotLevel, CacheLevel{1}}) + { + EXPECT_NE(manager->storage().getPoolGroupIndex(level, required), + manager->storage().getPoolGroupIndex(level, privateGroup)); + } + + // Each buffer's page index addresses its own lifecycle's pool: state and conv share one REQUIRED + // pool, while the PRIVATE pools hold old_x, old_dt and count, largest first. + std::map const scales{{"state", 6}, {"conv", 6}, {"old_x", 3}, {"old_dt", 3}, {"count", 1}}; + std::map const scratchPages{{"state", 2}, {"conv", 2}, {"old_x", 1}, {"old_dt", 1}, {"count", 1}}; + auto cache = manager->createKvCache({}, {}, 1); + ASSERT_TRUE(cache->resume(mStream)); + ASSERT_TRUE(cache->resize(8)); + for (auto const& buffer : splitSsmBuffers()) + { + SCOPED_TRACE(std::to_string(buffer.layer) + "/" + buffer.role); + auto const converter = manager->getPageIndexConverter(buffer.layer, buffer.role); + EXPECT_EQ(converter.scale, scales.at(buffer.role)); + EXPECT_EQ(converter.scale, manager->getPageIndexScale(buffer.layer, buffer.role)); + EXPECT_EQ(converter.scratchPagesPerBlock, scratchPages.at(buffer.role)); + EXPECT_EQ(manager->getPageStride(buffer.layer, buffer.role), static_cast(buffer.size)); + int const slot + = cache->getSsmBlockBaseIndex(manager->getBufferGroupId(buffer.layer, buffer.role), kDefaultBeamIndex); + ASSERT_GE(slot, 0); + auto const perLayerIndex = converter(std::vector{slot}, PageIndexMode::PER_LAYER).at(0); + auto const sharedIndex = converter(std::vector{slot}, PageIndexMode::SHARED).at(0); + EXPECT_EQ(manager->getMemPoolBaseAddress(buffer.layer, buffer.role, PageIndexMode::PER_LAYER) + + static_cast(perLayerIndex) * buffer.size, + manager->getMemPoolBaseAddress(buffer.layer, buffer.role, PageIndexMode::SHARED) + + static_cast(sharedIndex) * buffer.size); + EXPECT_EQ(ssmBufferAddress(*manager, *cache, buffer, 0), + manager->getMemPoolBaseAddress(buffer.layer, buffer.role, PageIndexMode::SHARED) + + static_cast(sharedIndex) * buffer.size); + } + cache->close(); + manager->shutdown(); +} + +TEST_F(ReplayCacheTest, PrivateSsmBuffersStayOutOfSnapshotsAndFollowBeams) +{ + for (int length : {7, 8}) + { + SCOPED_TRACE(length); + auto manager = std::make_shared(makeSplitSsmConfig(/*withHostTier=*/false)); + auto const required = manager->getBufferGroupId(1, "state"); + auto const privateGroup = manager->getBufferGroupId(1, "old_x"); + auto const tokens = makeTokens(length); + TokenSpan const span{tokens.data(), length}; + auto fill = [&](KvCache const& cache, SplitSsmBuffer const& buffer, int salt) + { + EXPECT_EQ(cudaMemsetAsync(reinterpret_cast(ssmBufferAddress(*manager, cache, buffer, 0)), + splitSsmPattern(buffer, salt), buffer.size, mStream), + cudaSuccess); + }; + auto check = [&](KvCache const& cache, SplitSsmBuffer const& buffer, int beam, int salt) + { + EXPECT_EQ(cudaStreamSynchronize(mStream), cudaSuccess); + std::vector data(buffer.size); + EXPECT_EQ( + cudaMemcpy(data.data(), reinterpret_cast(ssmBufferAddress(*manager, cache, buffer, beam)), + data.size(), cudaMemcpyDeviceToHost), + cudaSuccess); + auto const pattern = splitSsmPattern(buffer, salt); + EXPECT_TRUE(std::all_of(data.begin(), data.end(), [pattern](uint8_t value) { return value == pattern; })) + << buffer.layer << "/" << buffer.role << " beam " << beam; + }; + + auto source = manager->createKvCache({}, {}, 1); + ASSERT_TRUE(source->resume(mStream)); + ASSERT_TRUE(source->resize(length)); + for (auto const& buffer : splitSsmBuffers()) + { + fill(*source, buffer, 0); + } + source->commit(span); + source->stopCommitting(); + { + auto match = manager->radixTree().match({}, span, false, true); + ASSERT_EQ(match.numTokens, length); + ASSERT_FALSE(match.blocks.empty()); + EXPECT_NE(match.blocks.back()->getPage(required), nullptr); + for (auto const* block : match.blocks) + { + EXPECT_EQ(block->getPage(privateGroup), nullptr); + } + } + + // The reader restores the REQUIRED snapshot but starts from a fresh PRIVATE slot. + auto reader = manager->createKvCache({}, span, 2, {}, length); + ASSERT_EQ(reader->numCommittedTokens(), length); + EXPECT_TRUE(reader->reuseStatus()[required.value()].complete); + EXPECT_FALSE(reader->reuseStatus()[privateGroup.value()].complete); + ASSERT_TRUE(reader->resume(mStream)); + reader->commitPendingStats(); + auto const snapshotStats = manager->getAndResetSsmSnapshotIterationStats(); + ASSERT_EQ(snapshotStats.size(), 1); + ASSERT_EQ(snapshotStats.count(required), 1); + EXPECT_EQ(snapshotStats.at(required).iterSnapshotHits, 1); + for (auto const& buffer : splitSsmBuffers()) + { + if (buffer.isPrivate) + { + EXPECT_NE( + ssmBufferAddress(*manager, *source, buffer, 0), ssmBufferAddress(*manager, *reader, buffer, 0)); + fill(*reader, buffer, 1); + } + else + { + check(*reader, buffer, 0, 0); + } + } + + // Beam expansion copies both lifecycles into every beam's own slot. + ASSERT_TRUE(reader->resize(length + 4)); + reader->setBeamWidth(BeamIndex{3}); + for (int beam = 0; beam < 3; ++beam) + { + for (auto const& buffer : splitSsmBuffers()) + { + check(*reader, buffer, beam, buffer.isPrivate ? 1 : 0); + if (beam > 0) + { + EXPECT_NE(ssmBufferAddress(*manager, *reader, buffer, beam), + ssmBufferAddress(*manager, *reader, buffer, 0)); + } + } + } + reader->setBeamWidth(BeamIndex{1}); + reader->close(); + source->close(); + manager->shutdown(); + } +} + +TEST_F(ReplayCacheTest, SuspendedSsmStateSurvivesOffloadAndSlotChange) +{ + auto manager = std::make_shared(makeSplitSsmConfig(/*withHostTier=*/true)); + auto const required = manager->getBufferGroupId(1, "state"); + auto const privateGroup = manager->getBufferGroupId(1, "old_x"); + auto const tokens = makeTokens(8); + TokenSpan const span{tokens.data(), 8}; + auto cache = manager->createKvCache({}, {}, 1); + ASSERT_TRUE(cache->resume(mStream)); + ASSERT_TRUE(cache->resize(10)); + for (auto const& buffer : splitSsmBuffers()) + { + EXPECT_EQ(cudaMemsetAsync(reinterpret_cast(ssmBufferAddress(*manager, *cache, buffer, 0)), + splitSsmPattern(buffer, 0), buffer.size, mStream), + cudaSuccess); + } + cache->commit(span); + std::map oldSlots; + for (auto const group : {required, privateGroup}) + { + oldSlots[group] = cache->getSsmBlockBaseIndex(group); + ASSERT_GE(oldSlots[group], 0); + } + cache->suspend(); + + // Claim every GPU slot of both SSM pool groups. The suspended cache's held state is evicted to + // host memory, and its former GPU slots are overwritten while this test owns them. + auto& storage = manager->storage(); + std::vector> oldSlotReservations; + for (auto const group : {required, privateGroup}) + { + auto const pool = storage.getPoolGroupIndex(kHotLevel, group); + auto const stats = storage.getStatistics(kHotLevel, pool); + EXPECT_GE(stats.evictable, 1); + auto slots = storage.newSlotsForPoolGroup(kHotLevel, pool, stats.free + stats.evictable); + auto const coldPool = storage.getPoolGroupIndex(CacheLevel{1}, group); + auto const coldStats = storage.getStatistics(CacheLevel{1}, coldPool); + EXPECT_GE(coldStats.total - coldStats.free, 1); + for (auto& slot : slots) + { + if (slot.slotId().value() == oldSlots[group]) + { + oldSlotReservations.emplace_back(group, std::move(slot)); + } + else + { + storage.releaseSlot(group, kHotLevel, std::move(slot)); + } + } + } + ASSERT_EQ(oldSlotReservations.size(), 2); + for (auto const& buffer : splitSsmBuffers()) + { + auto const group = manager->getBufferGroupId(buffer.layer, buffer.role); + auto const address = manager->getMemPoolBaseAddress(buffer.layer, buffer.role) + + static_cast(oldSlots[group]) * manager->getPageIndexScale(buffer.layer, buffer.role) + * manager->getPageStride(buffer.layer, buffer.role); + EXPECT_EQ(cudaMemsetAsync(reinterpret_cast(address), 0xEE, buffer.size, mStream), cudaSuccess); + } + EXPECT_EQ(cudaStreamSynchronize(mStream), cudaSuccess); + + ASSERT_TRUE(cache->resume(mStream)); + for (auto const group : {required, privateGroup}) + { + EXPECT_NE(cache->getSsmBlockBaseIndex(group), oldSlots[group]); + } + EXPECT_EQ(cudaStreamSynchronize(mStream), cudaSuccess); + for (auto const& buffer : splitSsmBuffers()) + { + std::vector data(buffer.size); + EXPECT_EQ(cudaMemcpy(data.data(), reinterpret_cast(ssmBufferAddress(*manager, *cache, buffer, 0)), + data.size(), cudaMemcpyDeviceToHost), + cudaSuccess); + auto const pattern = splitSsmPattern(buffer, 0); + EXPECT_TRUE(std::all_of(data.begin(), data.end(), [pattern](uint8_t value) { return value == pattern; })) + << buffer.layer << "/" << buffer.role; + } + for (auto& [group, slot] : oldSlotReservations) + { + storage.releaseSlot(group, kHotLevel, std::move(slot)); + } + cache->close(); + manager->shutdown(); +} + TEST_F(ReplayCacheTest, ClosingScratchCacheReleasesEverySlotExactlyOnce) { auto config = makeConfig(); diff --git a/docs/source/torch/kv_cache_manager.md b/docs/source/torch/kv_cache_manager.md index a95499e0e389..e599223bee36 100644 --- a/docs/source/torch/kv_cache_manager.md +++ b/docs/source/torch/kv_cache_manager.md @@ -71,3 +71,62 @@ In `KvCacheCreator._create_kv_cache_manager`, the `KVCacheManager` is instantiat For local testing or proof of concept, update these lines to use your implementation. Then, test it to ensure the `PyExecutor` runs with your customized `KVCacheManager`. + + +## Native V2 buffer reuse policies + +Custom cache managers can override a layer's reuse policy per buffer through +`tensorrt_llm.runtime.kv_cache_manager_v2`. Both attention and SSM layers default +to `ReusePolicy.REQUIRED`; a buffer's `reuse_policy=None` inherits its layer's +policy. For example, a custom recurrent consumer with independent state indices +can configure: + +```python +from tensorrt_llm.runtime.kv_cache_manager_v2 import BufferConfig, ReusePolicy, SsmLayerConfig + +layer = SsmLayerConfig( + layer_id=0, + reuse_policy=ReusePolicy.REQUIRED, + buffers=[ + BufferConfig(role="state", size=4096), + BufferConfig(role="checkpoint", size=2048, reuse_policy=ReusePolicy.OPTIONAL), + BufferConfig(role="scratch", size=1024, reuse_policy=ReusePolicy.PRIVATE), + ], +) +``` + +REQUIRED buffers constrain the reusable prefix. OPTIONAL buffers are candidates +at that endpoint; select their group IDs through `resume(optional_reuse_groups=...)` +on the first resume. Attention candidates retain the block-aligned sliding-window +checkpoint contract; an SSM snapshot must match the endpoint exactly. PRIVATE +buffers remain request-local and are never published as reusable checkpoints. +The consumer must initialize or reconstruct private and unselected optional state +before inference. Unselected optional snapshots count as neither hits nor misses. + +Use `manager.get_buffer_group_id(layer_id, role)` to resolve a buffer and +`manager.buffer_grouping[group_id]` to enumerate membership. Buffers with the same +lifecycle properties share a group, including across layers; selecting a group +selects all of its buffers. Policy overrides add at most three policy variants +per attention window/sink/sparsity combination and at most three SSM groups. +Coalescing remains within lifecycle groups and compatible buffer sizes/coalescing +groups. + +`get_layer_group_id(layer_id)` and `layer_grouping` are deprecated in documentation +only and produce no deprecation warnings. The former returns a uniform layer's +group and raises `ValueError` for a split layer. The latter preserves its +partition-of-layers result and raises if any layer is split. `LayerGroupId` retains +its name; it identifies a lifecycle group, not a physical storage pool. + +Existing Python cache-manager consumers continue using the legacy layer APIs +and require one lifecycle per layer. They do not adopt per-buffer policy splits; +the legacy API rejects ambiguous layer mappings. Consumers can migrate to the +buffer APIs when they add support for independent indices. Current Mamba kernels +couple recurrent, convolution, PLE and replay buffers through one state-slot +index; separate replay indices are a follow-up. This native configuration does +not add an LLM API configuration field. + +The native page-index converter also supports unequal layer shares in scratch +slots, including a layer's block crossing two physical slots. Existing attention +scratch-table kernels require whole layer blocks per slot. Custom consumers of +unequal shares that cross slots must use the native converter or supply compatible +index generation; existing Python consumers have not adopted that layout. diff --git a/tensorrt_llm/_torch/pyexecutor/kv_cache/kv_cache_manager_v2.py b/tensorrt_llm/_torch/pyexecutor/kv_cache/kv_cache_manager_v2.py index 1473ad77f053..63c79d1ef238 100644 --- a/tensorrt_llm/_torch/pyexecutor/kv_cache/kv_cache_manager_v2.py +++ b/tensorrt_llm/_torch/pyexecutor/kv_cache/kv_cache_manager_v2.py @@ -3490,7 +3490,11 @@ def _remove_zero_size_buffers(self, config: KVCacheManagerConfigPy) -> KVCacheMa ) ) else: - layers.append(SsmLayerConfig(layer_id=layer_id, buffers=buffers)) + layers.append( + SsmLayerConfig( + layer_id=layer_id, buffers=buffers, reuse_policy=layer.reuse_policy + ) + ) if not layers: raise ValueError("KVCacheManagerV2 requires at least one non-empty local cache layer") diff --git a/tensorrt_llm/runtime/kv_cache_manager_v2/__init__.pyi b/tensorrt_llm/runtime/kv_cache_manager_v2/__init__.pyi index 081da3ba2e29..8b904bee3cf0 100644 --- a/tensorrt_llm/runtime/kv_cache_manager_v2/__init__.pyi +++ b/tensorrt_llm/runtime/kv_cache_manager_v2/__init__.pyi @@ -257,18 +257,27 @@ class DiskCacheTierConfig: def tier(self) -> CacheTier: ... def validate(self) -> None: ... +class ReusePolicy(enum.IntEnum): + REQUIRED = 0 + PRIVATE = 1 + OPTIONAL = 2 + @dataclass(slots=True) class BufferConfig: + """One buffer of a layer. + + ``reuse_policy`` of ``None`` uses the layer's policy. Only SSM layers accept a + different policy; the buffer then belongs to the SSM layer group of that + policy while keeping its layer ID, so one SSM layer can span several layer + groups. + """ + role: DataRole size: int tokens_per_block_override: int | None = None is_sparse: bool = False coalescing_group: int = 0 - -class ReusePolicy(enum.IntEnum): - REQUIRED = 0 - PRIVATE = 1 - OPTIONAL = 2 + reuse_policy: ReusePolicy | None = None @dataclass(slots=True) class AttentionLayerConfig: @@ -282,8 +291,9 @@ class AttentionLayerConfig: @dataclass(slots=True) class SsmLayerConfig: - """Recurrent buffers sharing a layer-level reuse policy. + """Recurrent buffers with a layer-level default reuse policy. + A buffer's ``reuse_policy`` overrides the default for that buffer. OPTIONAL state is restored only when selected on first resume at the exact matched endpoint. The caller initializes or reconstructs all fresh state. """ @@ -808,6 +818,7 @@ class PageIndexConverter: scale: int expansion: int layer_offset: int + scratch_pages_per_block: int def __call__( self, @@ -936,10 +947,15 @@ class KVCacheManager: def num_layers(self) -> int: ... @property def layer_ids(self) -> Iterator[LayerId]: ... + # Deprecated: use get_buffer_group_id. Raises ValueError for a split layer. def get_layer_group_id(self, layer_id: LayerId) -> LayerGroupId: ... + def get_buffer_group_id(self, layer_id: LayerId, role: DataRole) -> LayerGroupId: ... + # Deprecated: use buffer_grouping. Raises ValueError if any layer is split. @property def layer_grouping(self) -> Sequence[Sequence[LayerId]]: ... @property + def buffer_grouping(self) -> Sequence[Sequence[BufferId]]: ... + @property def all_buffer_ids(self) -> Iterator[BufferId]: ... def get_aggregated_pages(self, buffers: Iterable[BufferId]) -> Iterator[AggregatedPageDesc]: ... @property diff --git a/tests/unittest/_torch/executor/kv_cache/test_kvcm2_integration.py b/tests/unittest/_torch/executor/kv_cache/test_kvcm2_integration.py index addb326f0a91..83bd77f0c2f4 100644 --- a/tests/unittest/_torch/executor/kv_cache/test_kvcm2_integration.py +++ b/tests/unittest/_torch/executor/kv_cache/test_kvcm2_integration.py @@ -77,6 +77,7 @@ KVCacheManagerConfig, LayerId, PageIndexMode, + ReusePolicy, ReuseScope, SsmLayerConfig, _introspection, @@ -963,6 +964,62 @@ def test_zero_size_filter_rejects_empty_local_cache() -> None: manager._remove_zero_size_buffers(config) +def test_zero_size_filter_preserves_layer_and_buffer_reuse_policies() -> None: + manager = object.__new__(KVCacheManagerV2) + manager.pp_layers = [0, 1, 2, 3] + manager.num_layers = 4 + manager.num_local_layers = 4 + manager.layer_offsets = {0: 0, 1: 1, 2: 2, 3: 3} + manager.num_kv_heads_per_layer = [2, 0, 0, 0] + manager.head_dim_per_layer = [64, 64, 64, 64] + manager.max_attention_window_vec = [None, None, None, None] + + config = KVCacheManagerConfig( + tokens_per_block=TOKENS_PER_BLOCK, + cache_tiers=[GpuCacheTierConfig(quota=1 << 20)], + layers=[ + AttentionLayerConfig( + layer_id=LayerId(0), + buffers=[BufferConfig(role=Role.KEY, size=1024)], + reuse_policy=ReusePolicy.PRIVATE, + ), + AttentionLayerConfig( + layer_id=LayerId(1), buffers=[BufferConfig(role=Role.KEY, size=0)] + ), + SsmLayerConfig( + layer_id=LayerId(2), + buffers=[ + BufferConfig(role=DataRole("ssm_state"), size=256), + BufferConfig(role=DataRole("conv_state"), size=0), + BufferConfig(role=DataRole("old_x"), size=64, reuse_policy=ReusePolicy.PRIVATE), + ], + reuse_policy=ReusePolicy.OPTIONAL, + ), + SsmLayerConfig( + layer_id=LayerId(3), + buffers=[BufferConfig(role=DataRole("ssm_state"), size=256)], + reuse_policy=ReusePolicy.PRIVATE, + ), + ], + commit_min_snapshot=True, + ) + + filtered = manager._remove_zero_size_buffers(config) + + assert manager.pp_layers == [0, 2, 3] + assert [layer.layer_id for layer in filtered.layers] == [0, 1, 2] + attention_layer, split_layer, private_layer = filtered.layers + assert attention_layer.reuse_policy == ReusePolicy.PRIVATE + assert isinstance(split_layer, SsmLayerConfig) + assert split_layer.reuse_policy == ReusePolicy.OPTIONAL + assert [(buffer.role, buffer.reuse_policy) for buffer in split_layer.buffers] == [ + (DataRole("ssm_state"), None), + (DataRole("old_x"), ReusePolicy.PRIVATE), + ] + assert isinstance(private_layer, SsmLayerConfig) + assert private_layer.reuse_policy == ReusePolicy.PRIVATE + + @pytest.mark.parametrize("sliding_window_size", [None, 8]) @pytest.mark.parametrize("attention_first", [False, True]) def test_event_window_sizes_filter_attention_without_backend_internals( diff --git a/tests/unittest/kv_cache_manager_v2_tests/test_optional_checkpoint.py b/tests/unittest/kv_cache_manager_v2_tests/test_optional_checkpoint.py index 83f7317e5169..fc361e9fcd9f 100644 --- a/tests/unittest/kv_cache_manager_v2_tests/test_optional_checkpoint.py +++ b/tests/unittest/kv_cache_manager_v2_tests/test_optional_checkpoint.py @@ -93,7 +93,8 @@ def test_reuse_status_and_optional_selection_bindings(priority): @pytest.mark.parametrize("length", [7, 8]) @pytest.mark.parametrize("select_optional", [False, True]) -def test_ssm_layer_reuse_policies(length, select_optional): +@pytest.mark.parametrize("split_layer", [False, True]) +def test_ssm_layer_reuse_policies(length, select_optional, split_layer): layers = [ kv.AttentionLayerConfig(layer_id=0, buffers=[kv.BufferConfig(role="key", size=4096)]), *[ @@ -111,6 +112,22 @@ def test_ssm_layer_reuse_policies(length, select_optional): ) ], ] + if split_layer: + layers = [ + layers[0], + kv.SsmLayerConfig( + layer_id=1, + reuse_policy=kv.ReusePolicy.PRIVATE, + buffers=[ + kv.BufferConfig(role=role, size=4096, reuse_policy=policy) + for role, policy in ( + ("state", kv.ReusePolicy.REQUIRED), + ("checkpoint", kv.ReusePolicy.OPTIONAL), + ("replay", None), + ) + ], + ), + ] config = kv.KVCacheManagerConfig( tokens_per_block=4, cache_tiers=[kv.GpuCacheTierConfig(quota=8 << 20)], @@ -128,8 +145,13 @@ def test_ssm_layer_reuse_policies(length, select_optional): assert source.resize(length) source.commit(tokens, is_end=True) source.close() - required, optional, private = [manager.get_layer_group_id(i) for i in (1, 2, 3)] + required, optional, private = ( + [manager.get_buffer_group_id(1, role) for role in ("state", "checkpoint", "replay")] + if split_layer + else [manager.get_layer_group_id(i) for i in (1, 2, 3)] + ) assert len({required, optional, private}) == 3 + manager.get_and_reset_ssm_snapshot_iteration_stats() reader = manager.create_kv_cache(input_tokens=tokens) caches.append(reader) assert reader.num_committed_tokens == length @@ -140,6 +162,14 @@ def test_ssm_layer_reuse_policies(length, select_optional): assert reader.resume( stream.cuda_stream, optional_reuse_groups=[optional] if select_optional else [] ) + reader.commit_pending_stats() + stats = manager.get_and_reset_ssm_snapshot_iteration_stats() + assert stats[required].iter_snapshot_hits == 1 + if select_optional: + assert stats[optional].iter_snapshot_lookups == stats[optional].iter_snapshot_hits == 1 + elif optional in stats: + assert stats[optional].iter_snapshot_lookups == 0 + assert stats[optional].iter_snapshot_hits == stats[optional].iter_snapshot_misses == 0 indices = [ reader.get_ssm_block_base_index(group) for group in (required, optional, private) ] @@ -153,3 +183,197 @@ def test_ssm_layer_reuse_policies(length, select_optional): for cache in reversed(caches): cache.close() manager.shutdown() + + +@pytest.mark.parametrize("length", [7, 8]) +def test_ssm_buffer_reuse_policies(length): + replay = kv.BufferConfig(role="old_x", size=2048, reuse_policy=kv.ReusePolicy.PRIVATE) + assert replay.reuse_policy == kv.ReusePolicy.PRIVATE + assert kv.BufferConfig(role="state", size=4096).reuse_policy is None + layers = [ + kv.AttentionLayerConfig(layer_id=0, buffers=[kv.BufferConfig(role="key", size=4096)]), + *[ + kv.SsmLayerConfig( + layer_id=i, + buffers=[ + kv.BufferConfig(role="state", size=4096), + kv.BufferConfig(role="conv", size=4096), + kv.BufferConfig(role="old_x", size=2048, reuse_policy=kv.ReusePolicy.PRIVATE), + ], + ) + for i in (1, 2) + ], + ] + config = kv.KVCacheManagerConfig( + tokens_per_block=4, + cache_tiers=[kv.GpuCacheTierConfig(quota=8 << 20)], + layers=layers, + commit_min_snapshot=True, + ) + assert config.layers[1].buffers[2].reuse_policy == kv.ReusePolicy.PRIVATE + manager = kv.KVCacheManager(config) + stream = torch.cuda.Stream() + tokens = list(range(length)) + caches = [] + try: + required = manager.get_buffer_group_id(1, "state") + private = manager.get_buffer_group_id(layer_id=1, role="old_x") + assert required != private + assert manager.get_buffer_group_id(2, "conv") == required + assert manager.get_buffer_group_id(2, "old_x") == private + assert manager.get_layer_group_id(0) == manager.get_buffer_group_id(0, "key") + for layer_id in (1, 2): + with pytest.raises(ValueError, match="layer groups"): + manager.get_layer_group_id(layer_id) + assert manager.num_layers == 3 + assert sorted(manager.layer_ids) == [0, 1, 2] + with pytest.raises(ValueError, match="layer groups"): + _ = manager.layer_grouping + for group in (required, private): + assert sorted({buffer.layer_id for buffer in manager.buffer_grouping[group]}) == [1, 2] + grouped = [ + (buffer.layer_id, buffer.role) for group in manager.buffer_grouping for buffer in group + ] + assert len(grouped) == len(set(grouped)) == len(list(manager.all_buffer_ids)) + + source = manager.create_kv_cache(input_tokens=[]) + caches.append(source) + assert source.resume(stream.cuda_stream) + assert source.resize(length) + source.commit(tokens, is_end=True) + source.close() + reader = manager.create_kv_cache(input_tokens=tokens) + caches.append(reader) + assert reader.num_committed_tokens == length + assert reader.reuse_status[required].complete + assert not reader.reuse_status[private].complete + assert reader.resume(stream.cuda_stream) + indices = [reader.get_ssm_block_base_index(group) for group in (required, private)] + assert all(index >= 0 for index in indices) + reader.suspend() + assert reader.resume(stream.cuda_stream) + assert indices == [reader.get_ssm_block_base_index(group) for group in (required, private)] + finally: + for cache in reversed(caches): + cache.close() + manager.shutdown() + + +@pytest.mark.parametrize("length", [7, 8]) +@pytest.mark.parametrize("select_optional", [False, True]) +def test_attention_buffer_reuse_policies(length, select_optional): + config = kv.KVCacheManagerConfig( + tokens_per_block=4, + cache_tiers=[kv.GpuCacheTierConfig(quota=4 << 20)], + layers=[ + kv.AttentionLayerConfig( + layer_id=0, + reuse_policy=kv.ReusePolicy.PRIVATE, + sliding_window_size=4, + buffers=[ + kv.BufferConfig(role="key", size=4096, reuse_policy=kv.ReusePolicy.REQUIRED), + kv.BufferConfig(role="index", size=2048, reuse_policy=kv.ReusePolicy.OPTIONAL), + kv.BufferConfig(role="scratch", size=1024), + ], + ) + ], + ) + manager = kv.KVCacheManager(config) + stream = torch.cuda.Stream() + caches = [] + try: + required, optional, private = [ + manager.get_buffer_group_id(0, role) for role in ("key", "index", "scratch") + ] + assert len({required, optional, private}) == 3 + source = manager.create_kv_cache(input_tokens=[]) + caches.append(source) + assert source.resume(stream.cuda_stream) + assert source.resize(length) + source.commit(list(range(length)), is_end=True) + source.close() + reader = manager.create_kv_cache(input_tokens=list(range(length))) + caches.append(reader) + assert reader.num_committed_tokens == length + assert reader.reuse_status[required].complete + optional_complete = length % config.tokens_per_block == 0 + assert reader.reuse_status[optional].complete == optional_complete + assert not reader.reuse_status[private].complete + if select_optional and not optional_complete: + with pytest.raises(ValueError): + reader.resume(stream.cuda_stream, optional_reuse_groups=[optional]) + select_optional = False + assert reader.resume( + stream.cuda_stream, optional_reuse_groups=[optional] if select_optional else [] + ) + reader.suspend() + assert reader.resume(stream.cuda_stream) + finally: + for cache in reversed(caches): + cache.close() + manager.shutdown() + + +@pytest.mark.parametrize("split", [False, True]) +def test_scratch_addresses_with_unequal_buffer_membership(split): + layers = [] + for layer in range(3): + buffers = [kv.BufferConfig(role="key", size=2048, tokens_per_block_override=2)] + if layer < 2: + buffers.extend( + [ + kv.BufferConfig(role="value", size=4096), + kv.BufferConfig(role="side", size=512, coalescing_group=7), + ] + ) + buffers.append( + kv.BufferConfig( + role="private", size=1024, reuse_policy=kv.ReusePolicy.PRIVATE if split else None + ) + ) + layers.append( + kv.AttentionLayerConfig(layer_id=layer, buffers=buffers, sliding_window_size=8) + ) + manager = kv.KVCacheManager( + kv.KVCacheManagerConfig( + tokens_per_block=4, + layers=layers, + cache_tiers=[kv.GpuCacheTierConfig(quota=4 << 20)], + swa_scratch_reuse=kv.SwaScratchReuseConfig(), + ) + ) + stream = torch.cuda.Stream() + cache = manager.create_kv_cache(input_tokens=list(range(64))) + try: + assert cache.resume(stream.cuda_stream) + assert cache.resize(64) + required = manager.get_buffer_group_id(0, "key") + scratch = cache.get_scratch_desc(required) + assert scratch is not None + blocks = scratch.range.end - scratch.range.beg + # The side pool has one buffer for each of two layers and needs half + # a slot per logical block, more than K/V's two-fifths share. + assert len(scratch.slot_ids) == (blocks + 1) // 2 + for layer in layers: + spans = [] + for buffer in layer.buffers: + group = manager.get_buffer_group_id(layer.layer_id, buffer.role) + converter = manager.get_page_index_converter(layer.layer_id, buffer.role) + indices = converter( + cache.get_base_page_indices(group), + kv.PageIndexMode.PER_LAYER, + cache.get_scratch_desc(group), + ) + base = manager.get_mem_pool_base_address( + layer.layer_id, buffer.role, kv.PageIndexMode.PER_LAYER + ) + stride = manager.get_page_stride(layer.layer_id, buffer.role) + assert len(indices) == 16 * converter.expansion + for index in indices: + assert index >= 0 + spans.append((base + index * stride, base + (index + 1) * stride)) + spans.sort() + assert all(end <= following for (_, end), (following, _) in zip(spans, spans[1:])) + finally: + cache.close() + manager.shutdown()