姊妹篇

 vllm 分析(十) ——Hisparse上篇
 vllm 分析（十二）——Hisparse下篇

 上篇主要介绍了 HiSparse 的基本概念，并基于 PR 42326 的代码实现展开分析。在 PR 42326 的基础上，PR51323 做了许多改进，扩展了对更多模型的支持，例如 DeepSeek V4。

概述

 针对 DeepSeek V4，vLLM 定义了多种 KV Cache Group，并基于 KVCacheSpec 计算 num block。DeepSeek V4 的 KV Cache 布局可参考：vllm 分析(十一)——deepseek v4 kv cache layout。

新增的KVCacheSpec

HiSparseHotSpec

@dataclass(frozen=True, kw_only=True)
class HiSparseHotSpec(KVCacheSpec):
    """Ephemeral per-request HiSparse hot-cache allocation."""

    page_size: int
    blocks_per_request: int

    @property
    def page_size_bytes(self) -> int:
        return self.page_size

    def max_memory_usage_bytes(self, vllm_config: VllmConfig) -> int:
        return self.blocks_per_request * self.page_size

    def max_num_blocks_per_req(self, vllm_config: VllmConfig, max_len: int) -> int:
        return self.blocks_per_request


@dataclass(frozen=True, kw_only=True)
class HiSparseResidentSpec(KVCacheSpec):
    """Reclaimable GPU-resident pages for host-backed HiSparse KV."""

    page_size: int

    @property
    def page_size_bytes(self) -> int:
        return self.page_size

    def max_memory_usage_bytes(self, vllm_config: VllmConfig) -> int:
        return cdiv(vllm_config.model_config.max_model_len, self.block_size) * (
            self.page_size
        )

    def max_num_blocks_per_req(self, vllm_config: VllmConfig, max_len: int) -> int:
        return cdiv(max_len, self.block_size)

 HiSparseHotSpec 描述的内存参数用于申请 Hot Cache（位于 HBM）。HiSparseResidentSpec 描述的内存参数用于申请 Resident Cache（位于 HBM）。对于 DeepSeek V4 模型，即存储 C4A 的 KV。

 PR 51323 实现了一套较复杂的 KV 存储机制。当推理系统的 HBM Block 充足时，C4A 的 KV 可直接存储于 HBM 上，Sparse Attention 的计算无需从主机 DRAM 中 Swap In KV Cache。当 HBM 空间不足时，则会从 Resident Cache 中回收空间，具体参见：HiSparseCoordinator.reclaim_resident_blocks

 fully_resident 标志决定是否走 Swap In 路径，相关逻辑见：HiSparseCacheHandle.resolve_topk。

class HiSparseCacheHandle:
    def resolve_topk(
        self,
        req_id_per_token: torch.Tensor,
        block_table: torch.Tensor,
        topk_indices: torch.Tensor,
        *,
        block_size: int,
        return_valid_counts: bool = False,
        plan_row_offset: int = 0,
        prefetch_followers: bool = True,
    ) -> HiSparseTopKResult:
        num_tokens = topk_indices.shape[0]
        if self.fully_resident:
            assert self.block_table is not None and self.view is not None
            converted = triton_convert_req_index_to_global_index(
                req_id_per_token[:num_tokens],
                self.block_table,
                topk_indices,
                BLOCK_SIZE=self.view.block_size,
                PHYSICAL_BLOCK_STRIDE=self.view.attention_block_stride,
                NUM_TOPK_TOKENS=topk_indices.shape[1],
                BLOCK_N=gcd(topk_indices.shape[1], 128),
                return_valid_counts=return_valid_counts,
            )
            if return_valid_counts:
                indices, valid_counts = converted
                return self.view.attention_cache, indices, valid_counts
            return self.view.attention_cache, converted
        if self.runtime.leader is not None:
            return self.runtime.apply_plan(
                block_size=block_size,
                num_tokens=num_tokens,
                return_valid_counts=return_valid_counts,
                plan_row_offset=plan_row_offset,
            )
        return self.runtime.swap_in(
            resident=self,
            req_id_per_token=req_id_per_token[:num_tokens],
            block_table=block_table,
            topk_indices=topk_indices,
            block_size=block_size,
            return_valid_counts=return_valid_counts,
            produce_plan=bool(self.runtime.followers),
            plan_row_offset=plan_row_offset,
            prefetch_followers=prefetch_followers,
        )

_get_hisparse_hma_config

_get_hisparse_hma_config

def _get_hisparse_hma_config(
    vllm_config: VllmConfig,
    groups: KVCacheGroupSpec | list[KVCacheGroupSpec],
    available_memory: int,
    host_budget: int,
    *,
    log_layout: bool = True,
) -> KVCacheConfig:

    resident_groups: list[KVCacheGroupSpec] = []
    hot_groups: list[KVCacheGroupSpec] = []
    current: list[tuple[str, KVCacheSpec]] = []
    current_page = 0

    def append_hot_group(layers: list[tuple[str, KVCacheSpec]]) -> None:
        page_sizes = {spec.page_size_bytes for _, spec in layers}
        if len(page_sizes) != 1:
            raise ValueError(
                "HiSparse hot-cache groups require one page size, got "
                f"{sorted(page_sizes)}."
            )
        page_size = page_sizes.pop()
        resident_groups.append(
            KVCacheGroupSpec(
                [
                    name[: -len(HISPARSE_HOT_SUFFIX)] + HISPARSE_RESIDENT_SUFFIX
                    for name, _ in layers
                ],
                HiSparseResidentSpec(
                    block_size=gpu_block_size,
                    page_size=page_size,
                ),
                block_pool_id=0,
                enable_prefix_caching=False,
                enable_kv_transfer=True,
            )
        )
        hot_groups.append(
            KVCacheGroupSpec(
                [name for name, _ in layers],
                HiSparseHotSpec(
                    block_size=gpu_block_size,
                    page_size=page_size,
                    blocks_per_request=hot_blocks_per_request,
                ),
                block_pool_id=0,
                enable_prefix_caching=False,
                enable_kv_transfer=False,
            )
        )

    for unit in hot_units:
        unit_page = sum(spec.page_size_bytes for _, spec in unit)
        if current and current_page + unit_page > indexer_page:
            append_hot_group(current)
            current = []
            current_page = 0
        current.extend(unit)
        current_page += unit_page
    if current:
        append_hot_group(current)


    gpu_groups = [
        indexer_group,
        *resident_groups,
        *hot_groups,
        *gpu_regular_groups,
        *gpu_other_regular_groups,
    ]
    gpu_stride, gpu_layers_by_offset = _get_packed_kv_cache_layout(gpu_groups)
    hot_page_alignment = math.lcm(
        *(group.kv_cache_spec.page_size_bytes for group in hot_groups)
    )
    gpu_stride = round_up(gpu_stride, hot_page_alignment)
    host_page = sum(spec.page_size_bytes for spec in source_specs.values())
    host_num_blocks = host_budget // host_page
    gpu_num_blocks = available_memory // gpu_stride
    override = vllm_config.cache_config.num_gpu_blocks_override
    if override is not None:
        host_num_blocks = gpu_num_blocks = override
    if host_num_blocks <= 0 or gpu_num_blocks <= 0:
        raise ValueError(
            "HiSparse HMA has no allocatable blocks: "
            f"host={host_num_blocks}, gpu={gpu_num_blocks}."
        )

	  # 1 cpu 侧KVCacheTensor配置
    tensors = [
        KVCacheTensor(
            size=spec.page_size_bytes * host_num_blocks,
            shared_by=[name],
            host_resident=True,
            block_pool_id=None,
        )
        for name, spec in source_specs.items()
    ]
    gpu_size = gpu_stride * gpu_num_blocks
    # 2 gpu 侧KVCacheTensor配置
    tensors.extend(
        KVCacheTensor(
            size=gpu_size,
            shared_by=names,
            offset=offset,
            block_stride=gpu_stride,
            block_pool_id=0,
        )
        for offset, names in sorted(gpu_layers_by_offset.items())
    )

    return KVCacheConfig(
        num_blocks=gpu_num_blocks,
        num_blocks_by_pool=[gpu_num_blocks],
        kv_cache_tensors=tensors,
        kv_cache_groups=[
            source_group,
            indexer_group,
            *resident_groups,
            *hot_groups,
            *gpu_regular_groups,
            *gpu_other_regular_groups,
        ],
        hisparse_host_num_blocks=host_num_blocks,
    )

HiSparseHotSpec 和 HiSparseResidentSpec 使用了相同的 page_size，有相同的byte_offset。
_get_packed_kv_cache_layout
_get_kv_cache_config_packed

block poll的更改

 KVCacheTensor增加block_pool_id元素。
 v0.27.1中，所有的kv cache 都位于HBM，KVCacheCoordinator只有一个block_pool，负责分配block id。
 在 HiSparse PR 的代码中，既有位于 HBM 的 KV Cache，也有位于 CPU DRAM 的 KV Cache。DRAM 上的 KV Cache 块通过 host_block_pool 分配 Block ID。
 KVCacheCoordinator 增加了 block_pools。

class KVCacheCoordinator(ABC):
    """
    Coordinate the KV cache of different KV cache groups.
    """

    enable_partial_hash_hits = False

    def __init__(
        self,
        kv_cache_config: KVCacheConfig,
        max_model_len: int,
        max_in_flight_tokens: int,
        use_eagle: bool,
        enable_caching: bool,
        enable_kv_cache_events: bool,
        dcp_world_size: int,
        pcp_world_size: int,
        scheduler_block_size: int,
        hash_block_size: int,
        metrics_collector: KVCacheMetricsCollector | None = None,
    ):
    
       self.block_pools = tuple(
            BlockPool(
                num_gpu_blocks=num_blocks,
                enable_caching=pool_enable_caching[pool_id],
                hash_block_size=hash_block_size,
                enable_kv_cache_events=enable_kv_cache_events,
                metrics_collector=metrics_collector,
                block_pool_id=pool_id,
            )
            for pool_id, num_blocks in enumerate(kv_cache_config.num_blocks_by_pool)
        )
        # Compatibility alias for callers that only support the traditional
        # single-domain layout.
        self.block_pool = self.block_pools[0]

        source_groups = [
            group
            for group in kv_cache_config.kv_cache_groups
            if group.role is KVCacheGroupRole.HISPARSE_SOURCE
        ]
        host_block_pool = HiSparseCoordinator.create_host_block_pool(
            kv_cache_config,
            enable_caching=(
                enable_caching
                and bool(source_groups)
                and source_groups[0].enable_prefix_caching
            ),
            hash_block_size=hash_block_size,
            enable_kv_cache_events=enable_kv_cache_events,
            metrics_collector=metrics_collector,
        )
        group_block_pools: list[BlockPool] = []
        for group in kv_cache_config.kv_cache_groups:
            if group.role is KVCacheGroupRole.HISPARSE_SOURCE:
                assert host_block_pool is not None
                group_block_pools.append(host_block_pool)
            else:
                assert group.block_pool_id is not None
                group_block_pools.append(self.block_pools[group.block_pool_id])


        self.single_type_managers = tuple(
            get_manager_for_kv_cache_spec(
                kv_cache_spec=kv_cache_group.kv_cache_spec,
                max_in_flight_tokens=max_in_flight_tokens,
                max_model_len=max_model_len,
                block_pool=group_block_pools[i],
                enable_caching=(
                    enable_caching and kv_cache_group.enable_prefix_caching
                ),
                kv_cache_group_id=i,
                dcp_world_size=dcp_world_size,
                pcp_world_size=pcp_world_size,
                scheduler_block_size=self.scheduler_block_size,
                needs_kv_cache_zeroing=(
                    kv_cache_group.block_pool_id
                    in self.kv_cache_config.zeroing_block_pool_ids
                ),
            )
            for i, kv_cache_group in enumerate(self.kv_cache_config.kv_cache_groups)
        )
        self.hisparse_coordinator = HiSparseCoordinator(
            kv_cache_config, self.single_type_managers, max_model_len
        )

init_kv_cache

kv cache的处理细节，参考：vllm 分析(十一)——deepseek v4 kv cache layout。
init_kv_cache

def init_kv_cache(
    runner_kv_caches: list[torch.Tensor | list[torch.Tensor]],
    forward_context: dict[str, Any],
    kv_cache_config: KVCacheConfig,
    attn_groups: list[list[AttentionGroup]],
    device: torch.device,
    cache_dtype: str,
    kernel_block_sizes: list[int],
    vllm_config: VllmConfig,
    block_tables: "BlockTables",
) -> tuple[dict[str, Any], "HiSparseWorker | None"]:
    shared_kv_cache_layers = get_shared_kv_cache_layers(vllm_config)
    kv_cache_raw_tensors = _allocate_kv_cache(
        kv_cache_config, shared_kv_cache_layers, device
    )
    flattened_attn_groups = list(group for groups in attn_groups for group in groups)
    kv_caches = _reshape_kv_cache(
        attn_groups=flattened_attn_groups,
        kv_cache_raw_tensors=kv_cache_raw_tensors,
        kernel_block_sizes=kernel_block_sizes,
        cache_dtype=cache_dtype,
        shared_kv_cache_layers=shared_kv_cache_layers,
        kv_cache_config=kv_cache_config,
    )
    hisparse_worker = None
    if vllm_config.attention_config.hisparse_config is not None:
        hisparse_worker = init_hisparse_worker(
            forward_context=forward_context,
            kv_cache_config=kv_cache_config,
            raw_tensors=kv_cache_raw_tensors,
            kv_caches=kv_caches,
            block_tables=block_tables,
            max_num_reqs=vllm_config.scheduler_config.max_num_seqs,
            max_model_len=vllm_config.model_config.max_model_len,
            max_concurrent_batches=vllm_config.max_concurrent_batches,
            device=device,
        )
    # Dual-attention models (e.g. LongCat-Flash) put two Attention modules per
    # decoder layer, so a layer name carries two integers (layer + module index).
    num_attn_module = (
        2
        if vllm_config.model_config.hf_config.model_type
        in ("longcat_flash", "longcat_flash_ngram")
        else 1
    )
    bindable_caches = {
        name: cache for name, cache in kv_caches.items() if name in forward_context
    }
    bind_kv_cache(bindable_caches, forward_context, runner_kv_caches, num_attn_module)
    runner_kv_caches.extend(
        cache for name, cache in kv_caches.items() if name not in forward_context
    )
    return kv_caches, hisparse_worker

_allocate_kv_cache

_allocate_kv_cache

def _allocate_kv_cache(
    kv_cache_config: KVCacheConfig,
    shared_layers: dict[str, str],
    device: torch.device,
):
    host_bytes = sum(
        tensor.size
        for tensor in kv_cache_config.kv_cache_tensors
        if tensor.host_resident
    )
    if host_bytes:
        check_hisparse_host_memory(host_bytes)

    kv_cache_raw_tensors: dict[str, torch.Tensor] = {}
    packed_backings: dict[int, torch.Tensor] = {}
    for kv_cache_tensor in kv_cache_config.kv_cache_tensors:
        if kv_cache_tensor.host_resident:
            tensor = allocate_pinned_host_pool(kv_cache_tensor.size)
        elif kv_cache_tensor.block_stride > 0:
            assert kv_cache_tensor.block_pool_id is not None
            # Allocate once; all packed tensors alias the same backing.
            packed_backing = packed_backings.get(kv_cache_tensor.block_pool_id)
            if packed_backing is None:
                packed_backing = torch.zeros(
                    kv_cache_tensor.size, dtype=torch.int8, device=device
                )
                packed_backings[kv_cache_tensor.block_pool_id] = packed_backing
            tensor = packed_backing
        else:
            tensor = torch.zeros(kv_cache_tensor.size, dtype=torch.int8, device=device)
        for layer_name in kv_cache_tensor.shared_by:
            kv_cache_raw_tensors[layer_name] = tensor

    layer_names = set()
    for group in kv_cache_config.kv_cache_groups:
        for layer_name in group.layer_names:
            layer_names.add(layer_name)
    assert layer_names == (kv_cache_raw_tensors.keys() | shared_layers.keys()), (
        "Some layers are not correctly initialized"
    )
    return kv_cache_raw_tensors

 allocate_pinned_host_pool用于申请 host cache，位于 cpu dram。

init_hisparse_worker

init_hisparse_worker
cache_handle.bind_cache: 绑定resident cache。
cache_handle.runtime.bind_hot_cache：绑定hot cache。
cache_handle.runtime.bind_source_cache：绑定 host cache。

数据 back up

拷贝kv cache 到 host cache。HiSparseWorker._enqueue_transfers

class HiSparseWorker:
    def _enqueue_transfers(self, transfers: list[SparseKVPageTransfer]) -> None:
        src_staging = self.spill_src_cpu[staging_idx]
        dst_staging = self.spill_dst_cpu[staging_idx]
        src = src_staging.numpy()
        dst = dst_staging.numpy()
        offsets = np.arange(self.kernel_block_size, dtype=np.int64)
        for transfer_idx, transfer in enumerate(transfers):
            start = transfer_idx * self.kernel_block_size
            end = start + self.kernel_block_size
            for cache_idx, cache in enumerate(self.cache_handles):
                block_id = transfer.source_block_ids[
                    cache.runtime.resident_source_index
                ]
                src[cache_idx, start:end] = block_id * self.kernel_block_size + offsets
            host_page = (
                transfer.destination_block_id * self.blocks_per_kv_block
                + transfer.destination_page_offset
            )
            dst[start:end] = host_page * self.kernel_block_size + offsets
        self.spill_src_gpu[:, :num_rows].copy_(
            src_staging[:, :num_rows], non_blocking=True
        )
        self.spill_dst_gpu[:num_rows].copy_(dst_staging[:num_rows], non_blocking=True)
        current_stream = torch.accelerator.current_stream(self.hot_backing.device)
        staging_event.record(current_stream)
        self._spill_staging_index = (staging_idx + 1) % len(self._spill_staging_events)
        torch.ops._C_cache_ops.hisparse_backup_layers(
            self.hot_backing,
            self.backup_layer_offsets,
            self.spill_src_indices_ptrs,
            self.backup_host_anchor,
            self.backup_host_cache_ptrs,
            self.spill_dst_gpu,
            num_rows,
            self.backup_src_block_stride,
            self.backup_src_block_size,
            self.backup_src_rows,
            self.backup_row_value_bytes,
        )
        self.host_write_event.record(current_stream)

hot_backing 就是 hot cache，和 resident cache有相同的offset。spill_src_indices_ptrs中的slot计算基于resident cache 的 block id。

for transfer_idx, transfer in enumerate(transfers):
    start = transfer_idx * self.kernel_block_size
    end = start + self.kernel_block_size
    for cache_idx, cache in enumerate(self.cache_handles):
        block_id = transfer.source_block_ids[
            cache.runtime.resident_source_index
        ]
        src[cache_idx, start:end] = block_id * self.kernel_block_size + offsets
    host_page = (
        transfer.destination_block_id * self.blocks_per_kv_block
        + transfer.destination_page_offset
    )
    dst[start:end] = host_page * self.kernel_block_size + offsets

SparseKVPageTransfer信息的构建, HiSparseCoordinator._plan_spill

class HiSparseCoordinator:
    def _plan_spill(
        self,
        request_id: str,
        page_idx: int,
        *,
        release_after: bool,
        after_forward: bool,
    ) -> bool:
        assert self.host_manager is not None
        state = self._get_request_state(request_id)
        host_block_idx = page_idx // self.pages_per_host_block
        host_blocks = self.host_manager.req_to_blocks.get(request_id)
        if host_blocks is None or host_block_idx >= len(host_blocks):
            return False
        host_block = host_blocks[host_block_idx]
        if host_block.is_null:
            return False

        blocks: list[KVCacheBlock] = []
        for manager in self.resident_managers:
            block = manager.get_resident_page(request_id, page_idx)
            if block is None:
                return False
            blocks.append(block)

        self.host_manager.block_pool.touch([host_block])
        for manager, block in zip(self.resident_managers, blocks):
            manager.block_pool.touch([block])
        spill_id = self.next_spill_id
        self.next_spill_id += 1
        plan = SparseKVPageTransfer(
            transfer_id=spill_id,
            destination_block_id=host_block.block_id,
            destination_page_offset=page_idx % self.pages_per_host_block,
            source_block_ids=tuple(block.block_id for block in blocks),
            after_forward=after_forward,
        )

kv cache manager

get_manager_for_kv_cache_spec

def get_manager_for_kv_cache_spec(
    kv_cache_spec: KVCacheSpec,
    max_in_flight_tokens: int,
    max_model_len: int,
    **kwargs,
) -> SingleTypeKVCacheManager:
    """
    Get the appropriate manager for a given KVCacheSpec.

    Uses the KVCacheSpecRegistry to look up the manager class, supporting
    both built-in and custom specs registered via @register_kv_cache_spec
    and KVCacheSpecRegistry.register.

    Args:
        kv_cache_spec: The KVCacheSpec instance
        max_in_flight_tokens: The max tokens scheduled but not yet settled
            (one batch per concurrent step); see `VllmConfig.max_in_flight_tokens`
        max_model_len: The maximum context length the model could serve
    Returns:
        An instance of the appropriate SingleTypeKVCacheManager subclass
    """
    manager_class = KVCacheSpecRegistry.get_manager_class(kv_cache_spec)
    manager = manager_class(kv_cache_spec, **kwargs)
    return manager


def register_all_kvcache_specs(vllm_config):
    """Built-in spec registration"""

    KVCacheSpecRegistry.register(
        HiSparseHotSpec,
        HiSparseHotManager,
        uniform_type_base_spec=HiSparseHotSpec,
    )
    KVCacheSpecRegistry.register(
        HiSparseResidentSpec,
        HiSparseResidentManager,
        uniform_type_base_spec=HiSparseResidentSpec,
    )

    KVCacheSpecRegistry.register(
        SlidingWindowSpec,
        SlidingWindowManager,
        uniform_type_base_spec=SlidingWindowSpec,
    )
    KVCacheSpecRegistry.register(
        SlidingWindowMLASpec,
        SlidingWindowManager,
        uniform_type_base_spec=SlidingWindowMLASpec,
    )

    KVCacheSpecRegistry.register(
        MLAAttentionSpec, FullAttentionManager, uniform_type_base_spec=FullAttentionSpec
    )

 HiSparseHotSpec注册的是HiSparseHotManager，其block_pool为KVCacheCoordinator.group_block_pools[0]。
 HiSparseResidentSpec注册的是HiSparseResidentManager，其block_pool为KVCacheCoordinator.__init__中的group_block_pools[0]。
 host cache 对应MLAAttentionSpec，代码证据。host cache的使用FullAttentionManager，其block_pool为KVCacheCoordinator__init__中的host_block_pool。

def _get_hisparse_hma_config(
    vllm_config: VllmConfig,
    groups: KVCacheGroupSpec | list[KVCacheGroupSpec],
    available_memory: int,
    host_budget: int,
    *,
    log_layout: bool = True,
) -> KVCacheConfig:
    if is_deepseek_v4:
        specs = {
            name: spec
            for name, spec in all_full_specs.items()
            if isinstance(spec, MLAAttentionSpec) and spec.compress_ratio == 4
        }
        if not specs:
            raise ValueError(
                "HiSparse requires DeepSeek V4 to expose C4 MLA cache layers."
            )
    else:
        specs = all_full_specs

    source_specs: dict[str, KVCacheSpec] = {}
    for name, spec in specs.items():
        if name not in indexer_specs:
            source_specs[name] = spec

    host_page = sum(spec.page_size_bytes for spec in source_specs.values())
    host_num_blocks = host_budget // host_page

reclaim_resident_blocks

 在KV缓存空间不足时，KVCacheManager可以回收“常驻（Resident）”物理块来腾出空间。

KVCacheManager.allocate_slots分配槽位失败（缺块）
    ↓
HiSparseCoordinator.reclaim_resident_blocks(缺口数)
    ↓
HiSparseResidentManager.reclaimable_pages()
    → 扫描所有请求，返回 [旧块]（排除最后两块）
    ↓
协调器根据优先级/热页/预算筛选出待驱逐的 (request_id, block_idx)
    ↓
HiSparseResidentManager.release_resident_page(request_id, block_idx)
    → 槽位置为 Null → 物理块归还 BlockPool
    ↓
allocate_slots 重新检查容量 → 若充足则继续分配

 KVCacheManager.allocate_slots
 HiSparseResidentManager.reclaimable_pages
 HiSparseResidentManager.release_resident_page

 一旦request位于resident kv cache被回收，fully_resident就被置为false。
HiSparseCoordinator.build_offload_command

class HiSparseCoordinator:
    def build_offload_command(
        self,
        request_ids: Sequence[str],
    ) -> SparseKVOffloadCommand | None:
        command = SparseKVOffloadCommand(
            block_table_updates=block_table_updates,
            page_transfers=self.spills_to_send,
            fully_resident=self.are_requests_fully_resident(request_ids),
        )

    def are_requests_fully_resident(self, request_ids: Sequence[str]) -> bool:
        return (
            bool(request_ids)
            and bool(self.resident_managers)
            and all(
                manager.is_fully_resident(request_id)
                for request_id in request_ids
                for manager in self.resident_managers
            )
        )

c4a kv cache 写入

 DeepseekCompressor将压缩后的kv entry，写入到 DeepseekV4Attention中的kv cache。针对Hisparse场景，压缩后的kv entry，写入resident cache。
Compressor.forward

class DeepseekCompressor(nn.Module):
    def forward(
        self,
        # [num_tokens, 2 * self.coff * self.head_dim]
        kv_score: torch.Tensor,
        # [num_tokens]
        positions: torch.Tensor,
        rotary_emb,
    ) -> None:
        cos_sin_cache = rotary_emb.cos_sin_cache
        k_cache_metadata = cast(Any, attn_metadata[self.k_cache_prefix])
        source_k_cache_metadata = k_cache_metadata
        k_cache_layer = self._static_forward_context[self.k_cache_prefix]
        kv_cache = k_cache_layer.kv_cache
        hisparse_cache = k_cache_layer.hisparse_cache
        if hisparse_cache is not None:
            assert hisparse_cache.view is not None
            kv_cache = hisparse_cache.view.cache
            num_kv_slots = source_k_cache_metadata.slot_mapping.numel()
            k_cache_metadata = SimpleNamespace(
                slot_mapping=hisparse_cache.get_compressed_slot_mapping(
                    positions[:num_kv_slots], self.compress_ratio
                )
            )

        compress_norm_rope_store_fn(
            state_cache=state_cache,
            num_actual=num_actual,
            token_to_req_indices=token_to_req_indices,
            positions=positions,
            slot_mapping=slot_mapping,
            block_table=block_table,
            block_size=block_size,
            state_width=state_width,
            cos_sin_cache=cos_sin_cache,
            kv_cache=kv_cache,
            k_cache_metadata=k_cache_metadata,
            pdl_kwargs=pdl_kwargs,
            head_dim=self.head_dim,
            rope_head_dim=self.rope_head_dim,
            compress_ratio=self.compress_ratio,
            overlap=self.overlap,
            use_fp4_cache=self.use_fp4_cache,
            rms_norm_weight=self.norm.weight,
            rms_norm_eps=self.rms_norm_eps,
            quant_block=self._quant_block,
            token_stride=self._token_stride,
            scale_dim=self._scale_dim,
            **extra_kwargs,
        )

        if hisparse_cache is not None and not hisparse_cache.decode_batch:
            hisparse_cache.runtime.backup_rows(
                kv_cache,
                k_cache_metadata.slot_mapping,
                source_k_cache_metadata.slot_mapping,
            )

        if self.head_dim == 128:
            source = k_cache_layer.hisparse_indexer_source
            if source is not None:
                host_cache, source_slot_mapping = source
                src_slots = source_k_cache_metadata.slot_mapping
                num_slots = src_slots.numel()
                dst_slots = compress_hisparse_slot_mapping(
                    source_slot_mapping[:num_slots],
                    positions[:num_slots],
                    logical_block_size=(host_cache.shape[1] * self.compress_ratio),
                    storage_block_size=host_cache.shape[1],
                    compress_ratio=self.compress_ratio,
                )
                torch.ops._C_cache_ops.hisparse_backup_indexer(
                    kv_cache,
                    src_slots,
                    host_cache,
                    dst_slots,
                    self._token_stride,
                )