lightseekorg--tokenspeed/python/tokenspeed/runtime/layers/attention/kv_cache/base.py

# Copyright (c) 2026 LightSeek Foundation
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

from __future__ import annotations

from typing import TYPE_CHECKING

import torch

from tokenspeed.runtime.configs.paged_cache_spec import PagedCacheGroupSpec
from tokenspeed.runtime.layers.paged_attention import PagedAttention
from tokenspeed.runtime.utils import get_colorful_logger

if TYPE_CHECKING:
    from tokenspeed.runtime.cache.kvstore_controller import LayerDoneCounter

logger = get_colorful_logger(__name__)


class BaseTokenToKVPool:
    """A memory pool that maps a token location to its kv cache data."""

    paged_cache_group_specs: tuple[PagedCacheGroupSpec, ...] = ()
    paged_cache_group_page_counts: dict[str, int] = {}
    supports_hierarchical_kv_cache: bool = True

    def __init__(
        self,
        size: int,
        dtype: torch.dtype,
        device: str,
        max_batch_size: int,
        max_context_len: int,
        page_size: int,
        rank: int,
    ):
        self.dtype = dtype
        self.rank = rank
        self.size = size
        self.page_size = page_size
        if dtype in (torch.float8_e5m2, torch.float8_e4m3fn):
            #  Store as torch.uint8 because Tensor.index_put is not implemented for torch.float8_e5m2
            self.store_dtype = torch.uint8
        else:
            self.store_dtype = dtype
        self.device = device
        self.offload_chunk_page_num = 1024
        self.token_slot_refs = None

        # default state for optional layer-wise transfer control
        self.layer_transfer_counter = None
        logger.info(
            f"Initialized token to kv pool with size {size}, dtype {dtype}, device {device}, page size {page_size}, rank {rank}"
        )

    @classmethod
    def cell_size(self) -> int:
        raise NotImplementedError()

    def register_layer_transfer_counter(self, layer_transfer_counter: LayerDoneCounter):
        self.layer_transfer_counter = layer_transfer_counter

    def set_token_slot_refs(self, token_slot_refs: torch.Tensor):
        self.token_slot_refs = token_slot_refs

    def bind_paged_cache_scheduler(self, scheduler: object) -> None:
        """Optional hook for model-specific paged-cache diagnostics."""
        return None

    @torch.no_grad()
    def clear_kv_buffers(self) -> None:
        """Zero the KV buffers in place.

        Used by sleep/wake: after resume_memory_occupation re-maps the KV region
        its pages hold garbage, so zero them. Subclasses store buffers under
        different attributes (``k_buffer``/``v_buffer`` for MHA, ``kv_buffer`` —
        possibly tuples — for MLA); introspect the known names so every pool is
        covered without per-class overrides. For non-quantized KV this is
        belt-and-suspenders (paging overwrites); for FP8 KV it removes garbage.
        """
        attrs = (
            "k_buffer",
            "v_buffer",
            "kv_buffer",
            # DeepSeek V4 pool buffer names.
            "swa_kv_buffer",
            "compressed_kv_buffer",
            "compressor_state_buffer",
            "indexer_kv_buffer",
            "indexer_state_buffer",
        )
        for attr in attrs:
            for entry in getattr(self, attr, None) or []:
                items = entry if isinstance(entry, (tuple, list)) else (entry,)
                for t in items:
                    if torch.is_tensor(t):
                        t.zero_()

    def maybe_log_paged_cache_group_pages(self) -> None:
        return None

    def get_key_buffer(self, layer_id: int) -> torch.Tensor:
        raise NotImplementedError()

    def get_value_buffer(self, layer_id: int) -> torch.Tensor:
        raise NotImplementedError()

    def get_kv_buffer(self, layer_id: int) -> tuple[torch.Tensor, torch.Tensor]:
        raise NotImplementedError()

    def set_kv_buffer(
        self,
        layer: PagedAttention,
        loc: torch.Tensor,
        cache_k: torch.Tensor,
        cache_v: torch.Tensor,
    ) -> None:
        raise NotImplementedError()

    def get_cpu_copy(self, page_indices: list[int]) -> torch.Tensor:
        raise NotImplementedError()

    def load_cpu_copy(
        self, kv_cache_cpu: torch.Tensor, page_indices: list[int]
    ) -> None:
        raise NotImplementedError()

    @property
    def prefix_cache_required_group_ids(self) -> tuple[str, ...] | None:
        """None means adjunct disabled; subclasses return required group ids."""
        return None

    # Buffer metadata used by prefill/decode disaggregation.
    def get_contiguous_buf_infos(self):
        raise NotImplementedError()

    def get_contiguous_buf_unit_lens(self):
        return [1] * len(self.get_contiguous_buf_infos()[2])

    # Layerwise buffer offsets used by prefill/decode disaggregation.
    def get_layerwise_buf_info_offsets(self, start_idx=0):
        raise NotImplementedError()