[Feature][KV Pool] Introduce AscendStoreCoordinator for hybrid cache-hit and mask coordination (vllm-project#10393)

## Reference Design

This PR ports the external-store cache-hit coordination model from vLLM
MooncakeStore HybridKVCacheCoordinator and MooncakeStoreCoordinator into
AscendStore KV Pool. The implementation keeps AscendStore key and
address generation, while using coordinator-style convergence and
per-group reachable masks for hybrid cache groups.
Ref:vllm-project/vllm#44774

## Main Changes

- Add AscendStoreCoordinator to mirror MooncakeStore external-cache hit
coordination over AscendStore keys.
- Wire the coordinator into KVPoolWorker and ChunkedTokenDatabase so
lookup, save, and load share the same per-group mask semantics.
- Preserve DeepSeek V4 compressed cache-family granularity: c4 and c128
keys are generated over raw-token spans, while transfer addresses remain
in cache-domain blocks.
- Carry num_prompt_tokens through request metadata so retention-aware
reachable masks can match the scheduler/cache-manager behavior.
- Keep the path disabled for non-hybrid or single-group KV Pool users,
preserving the legacy behavior.
- Add unit coverage for coordinator hit convergence, compressed
effective granularity, and transfer mask application.

## Expected Behavior

- Hybrid AscendStore lookup returns the common external prefix only when
required KV cache groups satisfy the coordinator policy.
- Sliding-window, Mamba align, and compressed DSV4 groups use
manager-compatible reachable masks instead of storing or loading every
physical chunk.
- DSV4 c128 participates at the 16K effective boundary when block size
is 128, rather than as 128-token chunks.
- Store skips chunks that cannot serve a future hit, reducing
external-store footprint for long-context hybrid models.
- Load fetches only chunks required by the per-group mask and does not
regress single-group KV Pool behavior.

## Validation

vLLM version: v0.20.1
vLLM main: vllm-project/vllm@7021be66e


- vLLM version: v0.23.0
- vLLM main:
vllm-project/vllm@a30addc

---------

Signed-off-by: Pz1116 <zpbzpb123123@gmail.com>
Co-authored-by: zqs164 <1640410765@qq.com>
Co-authored-by: Abstrey <846134527@qq.com>
Co-authored-by: zmc1997 <742133686@qq.com>
Co-authored-by: DreamerLeader <2270923832@qq.com>
Co-authored-by: bowgneo <gavin.gq.bao@gmail.com>

Author: 6 people

tests/ut/distributed/ascend_store/_mock_deps.py

@@ -23,20 +23,23 @@
     import tests.ut.distributed.ascend_store._mock_deps  # noqa: F401, E402
 """
 
+import importlib.util
 import logging
 import os
 import sys
 import types
+from typing import Any
 from unittest.mock import MagicMock
 
 # ---------------------------------------------------------------------------
 # Mock torch / torch_npu
 # ---------------------------------------------------------------------------
-if "torch" not in sys.modules:
+if "torch" not in sys.modules and importlib.util.find_spec("torch") is None:
     _torch = types.ModuleType("torch")
     _torch.Tensor = MagicMock  # type: ignore[attr-defined]
     _torch.bool = "bool"  # type: ignore[attr-defined]
     _torch.float16 = "float16"  # type: ignore[attr-defined]
+    _torch.float32 = "float32"  # type: ignore[attr-defined]
     _torch.zeros = MagicMock(return_value=MagicMock())  # type: ignore[attr-defined]
     _torch.sum = MagicMock(return_value=0)  # type: ignore[attr-defined]
     _torch.device = MagicMock()  # type: ignore[attr-defined]
@@ -56,6 +59,7 @@
 # ---------------------------------------------------------------------------
 # Mock vllm modules
 # ---------------------------------------------------------------------------
+_MOCK_VLLM_DEPS = importlib.util.find_spec("vllm") is None
 _vllm_mock_modules = [
     "vllm",
     "vllm.config",
@@ -83,31 +87,38 @@
     "vllm.v1.attention",
     "vllm.v1.attention.backend",
     "vllm.v1.core",
+    "vllm.v1.core.block_pool",
     "vllm.v1.core.kv_cache_manager",
     "vllm.v1.core.kv_cache_utils",
     "vllm.v1.core.sched",
     "vllm.v1.core.sched.output",
+    "vllm.v1.core.single_type_kv_cache_manager",
     "vllm.v1.kv_cache_interface",
+    "vllm.v1.kv_cache_spec_registry",
     "vllm.v1.outputs",
     "vllm.v1.request",
     "vllm.v1.serial_utils",
 ]
-for _mod_name in _vllm_mock_modules:
-    if _mod_name not in sys.modules:
-        sys.modules[_mod_name] = MagicMock()
+if _MOCK_VLLM_DEPS:
+    for _mod_name in _vllm_mock_modules:
+        if _mod_name not in sys.modules:
+            sys.modules[_mod_name] = MagicMock()
 
-sys.modules["vllm.utils.math_utils"].cdiv = lambda a, b: -(-a // b)  # type: ignore[attr-defined]
-sys.modules["vllm.logger"].logger = logging.getLogger("vllm")  # type: ignore[attr-defined]
+if _MOCK_VLLM_DEPS:
+    sys.modules["vllm.utils.math_utils"].cdiv = lambda a, b: -(-a // b)  # type: ignore[attr-defined]
+    sys.modules["vllm.logger"].logger = logging.getLogger("vllm")  # type: ignore[attr-defined]
 
-_base_mod = sys.modules["vllm.distributed.kv_transfer.kv_connector.v1.base"]
+_base_mod: Any = (
+    sys.modules["vllm.distributed.kv_transfer.kv_connector.v1.base"] if _MOCK_VLLM_DEPS else types.SimpleNamespace()
+)
 _base_mod.KVConnectorBase_V1 = type("KVConnectorBase_V1", (), {"__init__": lambda self, **kw: None})  # type: ignore[attr-defined]
 _base_mod.KVConnectorMetadata = type("KVConnectorMetadata", (), {})  # type: ignore[attr-defined]
 _base_mod.KVConnectorWorkerMetadata = type("KVConnectorWorkerMetadata", (), {})  # type: ignore[attr-defined]
 _base_mod.KVConnectorRole = MagicMock()  # type: ignore[attr-defined]
 _base_mod.KVConnectorRole.SCHEDULER = "SCHEDULER"
 _base_mod.KVConnectorRole.WORKER = "WORKER"
 
-_events_mod = sys.modules["vllm.distributed.kv_events"]
+_events_mod: Any = sys.modules["vllm.distributed.kv_events"] if _MOCK_VLLM_DEPS else types.SimpleNamespace()
 _events_mod.KVCacheEvent = type("KVCacheEvent", (), {})  # type: ignore[attr-defined]
 _events_mod.KVConnectorKVEvents = type("KVConnectorKVEvents", (), {})  # type: ignore[attr-defined]
 
@@ -127,14 +138,212 @@ def __getattr__(self, name):
     {"__init__": lambda self, **kwargs: self.__dict__.update(kwargs)},
 )
 
-_kv_cache_utils_mod = sys.modules["vllm.v1.core.kv_cache_utils"]
+_kv_cache_utils_mod: Any = sys.modules["vllm.v1.core.kv_cache_utils"] if _MOCK_VLLM_DEPS else types.SimpleNamespace()
 _kv_cache_utils_mod.BlockHash = bytes  # type: ignore[attr-defined]
 _kv_cache_utils_mod.maybe_convert_block_hash = lambda x: x  # type: ignore[attr-defined]
 
-_sched_output_mod = sys.modules["vllm.v1.core.sched.output"]
+
+class _FakeKVCacheBlock:
+    def __init__(self, block_id=0, **kwargs):
+        self.block_id = block_id
+        self.__dict__.update(kwargs)
+
+
+class _FakeKVCacheSpec:
+    def __init__(self, block_size=16, **kwargs):
+        self.block_size = block_size
+        for key, value in kwargs.items():
+            setattr(self, key, value)
+
+    def __eq__(self, other):
+        return type(self) is type(other) and self.__dict__ == getattr(other, "__dict__", {})
+
+    def copy_with_new_block_size(self, block_size):
+        kwargs = self.__dict__.copy()
+        kwargs["block_size"] = block_size
+        return type(self)(**kwargs)
+
+    @property
+    def page_size_bytes(self):
+        num_kv_heads = getattr(self, "num_kv_heads", 1)
+        head_size = getattr(self, "head_size", 1)
+        dtype = getattr(self, "dtype", None)
+        dtype_size = getattr(dtype, "itemsize", None)
+        if dtype_size is None and dtype is not None and hasattr(dtype, "element_size"):
+            dtype_size = dtype.element_size()
+        return self.block_size * num_kv_heads * head_size * int(dtype_size or 1) * 2
+
+
+class _FakeFullAttentionSpec(_FakeKVCacheSpec):
+    pass
+
+
+class _FakeSlidingWindowSpec(_FakeKVCacheSpec):
+    def __init__(self, block_size=16, sliding_window=32, **kwargs):
+        super().__init__(block_size=block_size, sliding_window=sliding_window, **kwargs)
+
+
+class _FakeMambaSpec(_FakeKVCacheSpec):
+    def __init__(self, block_size=16, **kwargs):
+        super().__init__(block_size=block_size, **kwargs)
+        self.num_speculative_blocks = getattr(self, "num_speculative_blocks", 0)
+
+
+class _FakeUniformTypeKVCacheSpecs(_FakeKVCacheSpec):
+    def __init__(self, block_size=16, kv_cache_specs=None, **kwargs):
+        super().__init__(block_size=block_size, **kwargs)
+        self.kv_cache_specs = kv_cache_specs or {}
+
+    @classmethod
+    def from_specs(cls, kv_cache_specs):
+        if not kv_cache_specs:
+            return None
+        first_spec = next(iter(kv_cache_specs.values()))
+        return cls(
+            block_size=getattr(first_spec, "block_size", 16),
+            kv_cache_specs=kv_cache_specs,
+        )
+
+
+class _FakeKVCacheGroupSpec:
+    def __init__(self, layer_names=None, kv_cache_spec=None, is_eagle_group=False):
+        self.layer_names = layer_names or []
+        self.kv_cache_spec = kv_cache_spec or _FakeFullAttentionSpec()
+        self.is_eagle_group = is_eagle_group
+
+
+class _FakeKVCacheConfig:
+    def __init__(self, num_blocks=1, kv_cache_tensors=None, kv_cache_groups=None):
+        self.num_blocks = num_blocks
+        self.kv_cache_tensors = kv_cache_tensors or []
+        self.kv_cache_groups = kv_cache_groups or []
+
+
+_kv_cache_utils_mod.KVCacheBlock = _FakeKVCacheBlock  # type: ignore[attr-defined]
+_kv_cache_utils_mod.BlockHashList = list  # type: ignore[attr-defined]
+
+
+class _FakeBlockPool:
+    def __init__(self, *args, **kwargs):
+        self.null_block = _FakeKVCacheBlock(block_id=0)
+        self._next_block_id = 1
+
+    def get_new_blocks(self, num_blocks):
+        blocks = []
+        for _ in range(num_blocks):
+            blocks.append(_FakeKVCacheBlock(block_id=self._next_block_id))
+            self._next_block_id += 1
+        return blocks
+
+
+if _MOCK_VLLM_DEPS:
+    sys.modules["vllm.v1.core.block_pool"].BlockPool = _FakeBlockPool  # type: ignore[attr-defined]
+
+
+class _FakeSingleTypeKVCacheManager:
+    def __init__(self, *args, **kwargs):
+        self._mock = MagicMock()
+
+    def __getattr__(self, name):
+        return getattr(self._mock, name)
+
+    @classmethod
+    def reachable_block_mask(
+        cls,
+        start_block,
+        end_block,
+        alignment_tokens,
+        kv_cache_spec,
+        use_eagle,
+        retention_interval=None,
+        num_prompt_tokens=None,
+    ):
+        return None
+
+    @classmethod
+    def find_longest_cache_hit(
+        cls,
+        block_hashes,
+        max_length,
+        kv_cache_group_ids,
+        block_pool,
+        kv_cache_spec,
+        drop_eagle_block=False,
+        alignment_tokens=16,
+        dcp_world_size=1,
+        pcp_world_size=1,
+    ):
+        computed: tuple[list[object], ...] = tuple([] for _ in kv_cache_group_ids)
+        max_blocks = max_length // kv_cache_spec.block_size
+        for block_hash in list(block_hashes)[:max_blocks]:
+            cached = block_pool.get_cached_block(block_hash, kv_cache_group_ids)
+            if not cached:
+                break
+            for blocks, block in zip(computed, cached):
+                blocks.append(block)
+        if drop_eagle_block and computed and computed[0]:
+            for blocks in computed:
+                blocks.pop()
+        return computed
+
+
+class _FakeSlidingWindowManager(_FakeSingleTypeKVCacheManager):
+    @classmethod
+    def reachable_block_mask(
+        cls,
+        start_block,
+        end_block,
+        alignment_tokens,
+        kv_cache_spec,
+        use_eagle,
+        retention_interval=None,
+        num_prompt_tokens=None,
+    ):
+        if alignment_tokens is None:
+            return None
+        per_segment = max(alignment_tokens // kv_cache_spec.block_size, 1)
+        return [(idx + 1) % per_segment == 0 for idx in range(start_block, end_block)]
+
+
+_single_type_mod: Any = (
+    sys.modules["vllm.v1.core.single_type_kv_cache_manager"] if _MOCK_VLLM_DEPS else types.SimpleNamespace()
+)
+_single_type_mod.SingleTypeKVCacheManager = _FakeSingleTypeKVCacheManager  # type: ignore[attr-defined]
+_single_type_mod.FullAttentionManager = _FakeSingleTypeKVCacheManager  # type: ignore[attr-defined]
+_single_type_mod.SlidingWindowManager = _FakeSlidingWindowManager  # type: ignore[attr-defined]
+_single_type_mod.MambaManager = _FakeSingleTypeKVCacheManager  # type: ignore[attr-defined]
+_single_type_mod.spec_manager_map = {  # type: ignore[attr-defined]
+    _FakeFullAttentionSpec: _FakeSingleTypeKVCacheManager,
+    _FakeSlidingWindowSpec: _FakeSlidingWindowManager,
+    _FakeMambaSpec: _FakeSingleTypeKVCacheManager,
+}
+
+_kv_interface_mod: Any = sys.modules["vllm.v1.kv_cache_interface"] if _MOCK_VLLM_DEPS else types.SimpleNamespace()
+_kv_interface_mod.KVCacheSpec = _FakeKVCacheSpec  # type: ignore[attr-defined]
+_kv_interface_mod.FullAttentionSpec = _FakeFullAttentionSpec  # type: ignore[attr-defined]
+_kv_interface_mod.SlidingWindowSpec = _FakeSlidingWindowSpec  # type: ignore[attr-defined]
+_kv_interface_mod.MambaSpec = _FakeMambaSpec  # type: ignore[attr-defined]
+_kv_interface_mod.UniformTypeKVCacheSpecs = _FakeUniformTypeKVCacheSpecs  # type: ignore[attr-defined]
+_kv_interface_mod.KVCacheGroupSpec = _FakeKVCacheGroupSpec  # type: ignore[attr-defined]
+_kv_interface_mod.KVCacheConfig = _FakeKVCacheConfig  # type: ignore[attr-defined]
+
+
+class _FakeKVCacheSpecRegistry:
+    @classmethod
+    def get_manager_class(cls, kv_cache_spec):
+        if isinstance(kv_cache_spec, _FakeSlidingWindowSpec):
+            return _FakeSlidingWindowManager
+        return _FakeSingleTypeKVCacheManager
+
+
+if _MOCK_VLLM_DEPS:
+    sys.modules["vllm.v1.kv_cache_spec_registry"].KVCacheSpecRegistry = _FakeKVCacheSpecRegistry  # type: ignore[attr-defined]
+
+_sched_output_mod: Any = sys.modules["vllm.v1.core.sched.output"] if _MOCK_VLLM_DEPS else types.SimpleNamespace()
 _sched_output_mod.NewRequestData = MagicMock  # type: ignore[attr-defined]
 
-sys.modules["vllm.envs"].VLLM_RPC_BASE_PATH = "/tmp/vllm_rpc"  # type: ignore[attr-defined]
+if _MOCK_VLLM_DEPS:
+    sys.modules["vllm.envs"].VLLM_RPC_BASE_PATH = "/tmp/vllm_rpc"  # type: ignore[attr-defined]
 
 # ---------------------------------------------------------------------------
 # Mock external backends

tests/ut/distributed/ascend_store/test_ascend_store_connector.py

@@ -18,18 +18,26 @@
 import unittest
 from unittest.mock import MagicMock, patch
 
-from vllm.distributed.kv_events import KVCacheEvent
-
+# isort: off
 import tests.ut.distributed.ascend_store._mock_deps  # noqa: F401, E402
+from vllm.distributed.kv_events import KVCacheEvent
 from vllm_ascend.distributed.kv_transfer.kv_pool.ascend_store.ascend_store_connector import (
     AscendStoreConnector,
     AscendStoreKVEvents,
 )
 
+# isort: on
+
+
+def _mock_events(num_workers=1):
+    events = AscendStoreKVEvents(num_workers=num_workers)
+    events._aggregator = MagicMock()
+    return events
+
 
 class TestAscendStoreKVEvents(unittest.TestCase):
     def _make_events(self, num_workers=1):
-        return AscendStoreKVEvents(num_workers=num_workers)
+        return _mock_events(num_workers=num_workers)
 
     def test_add_and_get_events(self):
         ev = self._make_events()
@@ -162,7 +170,7 @@ def test_update_connector_output_with_events(self, mock_scheduler_cls):
             role=KVConnectorRole.SCHEDULER,
             kv_cache_config=None,
         )
-        events = AscendStoreKVEvents(num_workers=1)
+        events = _mock_events(num_workers=1)
         mock_kv_events = [MagicMock()]
         events._aggregator.get_all_events.return_value = mock_kv_events
         events._aggregator.get_number_of_workers.return_value = 1
@@ -183,15 +191,15 @@ def test_update_connector_output_accumulate(self, mock_scheduler_cls):
             kv_cache_config=None,
         )
         # First update
-        events1 = AscendStoreKVEvents(num_workers=1)
+        events1 = _mock_events(num_workers=1)
         events1._aggregator.get_all_events.return_value = [MagicMock()]
         events1._aggregator.get_number_of_workers.return_value = 1
         output1 = MagicMock()
         output1.kv_cache_events = events1
         connector.update_connector_output(output1)
 
         # Second update
-        events2 = AscendStoreKVEvents(num_workers=1)
+        events2 = _mock_events(num_workers=1)
         events2._aggregator.get_all_events.return_value = [MagicMock()]
         events2._aggregator.get_number_of_workers.return_value = 1
         output2 = MagicMock()
@@ -214,7 +222,7 @@ def test_take_events(self, mock_scheduler_cls):
         self.assertEqual(result, [])
 
         # With events
-        events = AscendStoreKVEvents(num_workers=1)
+        events = _mock_events(num_workers=1)
         mock_event = MagicMock()
         events._aggregator.get_common_events.return_value = [mock_event]
         events._aggregator.get_all_events.return_value = [mock_event]