Mojo struct
PagedKVCache
@register_passable(trivial)
struct PagedKVCache[dtype_: DType, kv_params_: KVCacheStaticParams, page_size: Int]
The PagedKVCache is a wrapper around the KVCache blocks for a given layer. It is used to access the KVCache blocks for PagedAttention.
Parameters
- dtype_ (
DType): The dtype of the kv-cache. - kv_params_ (
KVCacheStaticParams): The kv-cache static parameters. - page_size (
Int): The size of the page.
Fields
- blocks (
LayoutTensor[PagedKVCache[dtype_, kv_params_, page_size].dtype, PagedKVCache[dtype_, kv_params_, page_size].blocks_layout, MutAnyOrigin]): - cache_lengths (
LayoutTensor[DType.uint32, Layout(IntTuple(-1)), ImmutAnyOrigin]): - lookup_table (
LayoutTensor[DType.uint32, Layout.row_major[2](), ImmutAnyOrigin]): - max_seq_length (
UInt32): - max_cache_length (
UInt32):
Implemented traits
AnyType,
Copyable,
DevicePassable,
ImplicitlyCopyable,
KVCacheT,
Movable,
UnknownDestructibility
Aliases
__copyinit__is_trivial
alias __copyinit__is_trivial = True
__del__is_trivial
alias __del__is_trivial = True
__moveinit__is_trivial
alias __moveinit__is_trivial = True
blocks_layout
alias blocks_layout = Layout.row_major(PagedKVCache[dtype_, kv_params_, page_size].blocks_shape)
blocks_shape
alias blocks_shape = IntTuple(-1, page_size, Int(PagedKVCache[dtype_, kv_params_, page_size].kv_params), Int(PagedKVCache[dtype_, kv_params_, page_size].kv_params))
blocks_type
alias blocks_type = LayoutTensor[PagedKVCache[dtype_, kv_params_, page_size].dtype, PagedKVCache[dtype_, kv_params_, page_size].blocks_layout, MutAnyOrigin]
device_type
alias device_type = PagedKVCache[dtype_, kv_params_, page_size]
dtype
alias dtype = dtype_
kv_params
alias kv_params = kv_params_
page_size_
alias page_size_ = page_size
Methods
__init__
__init__(blocks: LayoutTensor[PagedKVCache[dtype_, kv_params_, page_size].dtype, PagedKVCache[dtype_, kv_params_, page_size].blocks_layout, MutAnyOrigin], cache_lengths: LayoutTensor[DType.uint32, Layout(IntTuple(-1)), ImmutAnyOrigin], lookup_table: LayoutTensor[DType.uint32, Layout.row_major[2](), ImmutAnyOrigin], max_seq_length: UInt32, max_cache_length: UInt32) -> Self
get_type_name
static get_type_name() -> String
Returns:
String
get_device_type_name
static get_device_type_name() -> String
Returns:
String
max_tile_size
cache_lengths_nd
cache_lengths_nd(self) -> LayoutTensor[DType.uint32, Layout(IntTuple(-1)), ImmutAnyOrigin]
Returns:
cache_length
cache_length(self, batch_idx: Int) -> Int
Returns the length of the cache for a given batch index.
Returns:
row_idx
row_idx(self, batch_idx: UInt32, tok_idx: UInt32) -> UInt32
Returns the row idx when viewing the memory as a matrix.
Returns:
col_idx
col_idx(self, head_idx: UInt32) -> UInt32
Returns the col idx when viewing the memory as a matrix.
Returns:
create_tma_tile
create_tma_tile[tile_m: Int, tile_n: Int, swizzle_mode: TensorMapSwizzle, *, is_k_major: Bool](self, ctx: DeviceContext) -> TMATensorTile[PagedKVCache[dtype_, kv_params_, page_size].dtype, tile_layout_k_major[PagedKVCache[dtype_, kv_params_, page_size].dtype, tile_m, tile_n, swizzle_mode]() if is_k_major else tile_layout_mn_major[PagedKVCache[dtype_, kv_params_, page_size].dtype, tile_n, tile_m, swizzle_mode](), _tma_desc_tile_layout[PagedKVCache[dtype_, kv_params_, page_size].dtype, 2, IndexList[2, DType.int64](tile_m, tile_n, Tuple[]()), is_k_major, swizzle_mode](), is_k_major]
Creates a TMA tile for this KV cache.
Returns:
TMATensorTile
load
load[width: Int](self, bs: Int, head_idx: Int, tok_idx: Int, head_dim_idx: Int) -> SIMD[PagedKVCache[dtype_, kv_params_, page_size].dtype, width]
Loads an element from the given index.
Returns:
store
store(self, bs: Int, head_idx: Int, tok_idx: Int, head_dim_idx: Int, val: SIMD[PagedKVCache[dtype_, kv_params_, page_size].dtype, size])
Stores an element at the given index.
empty_cache
empty_cache(self) -> Bool
Returns true if the cache_lengths for all requests is 0, false otherwise.
Returns:
max_prompt_length
max_prompt_length(self) -> UInt32
Returns the maximum sequence length across all batches of the current request.
Returns:
max_context_length
max_context_length(self) -> UInt32
Returns the maximum cache length used across all batches of the current request.
Returns:
block_paged_ptr
block_paged_ptr[tile_size: Int](self, batch_idx: Int, start_tok_idx: Int, head_idx: Int, head_dim_idx: Int = 0) -> LegacyUnsafePointer[Scalar[PagedKVCache[dtype_, kv_params_, page_size].dtype]]
Returns:
LegacyUnsafePointer
Was this page helpful?
Thank you! We'll create more content like this.
Thank you for helping us improve!