[HiSparse] Log steady-state max concurrency and expose host-tier utilization gauges (#58949)

Signed-off-by: Matthew Bonanni <mbonanni@redhat.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: Matthew Bonanni <mbonanni@redhat.com>

docs/design/hisparse.md

@@ -37,6 +37,14 @@ memory consumption is therefore topology- and implementation-dependent. The
 realized capacity may be slightly smaller because the budget is rounded down
 to complete host blocks.
 
+Startup logs report two concurrency bounds at `max_model_len`. The generic
+`Maximum concurrency` line charges each request its full admission footprint,
+including the in-flight window of every resident group. The `HiSparse
+steady-state maximum concurrency` line charges running requests that read from
+host only their active tail pages, plus one request being admitted at its full
+footprint. Host-pool metrics are listed in
+[Metrics](../usage/metrics.md#hisparse-kv-connector-metrics).
+
 ## Ownership
 
 | Thing | Owner | What “owner” means |

docs/mkdocs/gen_files/generate_metrics.py

@@ -23,6 +23,10 @@ METRIC_SOURCE_FILES = [
         "path": "vllm/distributed/kv_transfer/kv_connector/v1/nixl/stats.py",
         "key": "metrics-nixl",
     },
+    {
+        "path": "vllm/distributed/kv_transfer/kv_connector/v1/hisparse/stats.py",
+        "key": "metrics-hisparse",
+    },
     {
         "path": "vllm/v1/simple_kv_offload/metrics.py",
         "key": "metrics-simple-kv-offload",

docs/usage/metrics.md

@@ -45,6 +45,15 @@ The following metrics are exposed:
 
 --8<-- "gen:metrics-nixl"
 
+## HiSparse KV Connector Metrics
+
+These metrics are exposed when the `HiSparseConnector` KV connector is
+configured. The host-pool gauges are reported by the scheduler once per engine
+step, for the host pool shared by all tensor-parallel ranks; evictable cached
+blocks count as free.
+
+--8<-- "gen:metrics-hisparse"
+
 ## Simple CPU Offload Connector Metrics
 
 These metrics are exposed when the `SimpleCPUOffloadConnector` KV connector

tests/v1/core/test_kv_cache_utils.py

@@ -60,6 +60,7 @@ from vllm.v1.core.kv_cache_utils import (
 from vllm.v1.hisparse.layout import (
     create_hisparse_layout,
     get_hisparse_kv_cache_groups,
+    get_hisparse_steady_state_concurrency,
 )
 from vllm.v1.kv_cache_interface import (
     ChunkedLocalAttentionSpec,
@@ -198,6 +199,48 @@ def test_hisparse_hma_uses_resolved_gpu_block_size(
     assert scheduler_block_size == hash_block_size == gpu_block_size
 
 
+def test_hisparse_steady_state_concurrency_excludes_spilled_resident_pages():
+    """Running requests pin only their resident tail; the newest still needs its
+    full in-flight window to be admitted."""
+    block_size = 16
+    config = SimpleNamespace(
+        model_config=SimpleNamespace(max_model_len=10 * block_size),
+        parallel_config=SimpleNamespace(decode_context_parallel_size=1),
+        max_in_flight_tokens=10 * block_size,
+    )
+    attn_spec = FullAttentionSpec(
+        block_size=block_size, num_kv_heads=1, head_size=64, dtype=torch.float16
+    )
+    page_size = attn_spec.page_size_bytes
+    kv_cache_config = KVCacheConfig(
+        num_blocks=64,
+        kv_cache_tensors=[],
+        kv_cache_groups=[
+            KVCacheGroupSpec(["source"], attn_spec, host_resident=True),
+            KVCacheGroupSpec(["indexer"], attn_spec),
+            KVCacheGroupSpec(
+                ["resident"],
+                HiSparseResidentSpec(block_size=block_size, page_size=page_size),
+            ),
+            KVCacheGroupSpec(
+                ["hot"],
+                HiSparseHotSpec(
+                    block_size=block_size, page_size=page_size, blocks_per_request=4
+                ),
+            ),
+        ],
+        hisparse_host_num_blocks=100,
+    )
+
+    # Admission takes indexer 10 + resident 10 + hot 4 = 24 blocks and steady
+    # state 16, so 3 fit (2 x 16 + 24 <= 64) but not 4 (3 x 16 + 24 > 64).
+    assert get_hisparse_steady_state_concurrency(config, kv_cache_config) == 3.5
+    # The worst-case bound charges all 10 resident pages.
+    assert get_max_concurrency_for_kv_cache_config(
+        config, kv_cache_config
+    ) == pytest.approx(64 / 24)
+
+
 @pytest.mark.parametrize("extra_blocks,ok", [(1, False), (2, True)])
 def test_hisparse_host_pool_must_fit_max_model_len(monkeypatch, extra_blocks, ok):
     """Every page needs a host block, so one max_model_len request must fit."""

tests/v1/kv_connector/unit/test_hisparse_connector.py

@@ -64,6 +64,42 @@ def test_hisparse_requires_block_outermost_device_layout():
     assert HiSparseConnector.get_required_kvcache_layout(MagicMock()) == "BLHNC"
 
 
+def test_scheduler_stats_report_host_pool_usage():
+    """The scheduler-side stats sample the coordinator's host pool level."""
+    from tests.v1.core.test_prefix_caching import (
+        HISPARSE_BLOCK_SIZE,
+        make_hisparse_kv_cache_manager,
+        make_request,
+        sha256,
+    )
+    from vllm.v1.core.kv_cache_utils import init_none_hash
+
+    init_none_hash(sha256)
+
+    manager = make_hisparse_kv_cache_manager(32, host_num_blocks=8)
+    scheduler = HiSparseConnectorScheduler(async_speculative=False)
+    scheduler.bind_coordinator(get_hisparse_coordinator(manager))
+    coordinator = scheduler.coordinator
+
+    stats = scheduler.get_kv_connector_stats()
+    assert stats.data["host_cache_usage_perc"] == [0.0]
+    assert stats.data["pending_page_transfers"] == [0]
+
+    request = make_request(
+        "request", list(range(4 * HISPARSE_BLOCK_SIZE)), HISPARSE_BLOCK_SIZE, sha256
+    )
+    assert manager.allocate_slots(request, num_new_tokens=64) is not None
+    host_blocks = coordinator.host_manager.req_to_blocks[request.request_id]
+    num_used = sum(not block.is_null for block in host_blocks)
+
+    # The pool reserves the null block, so 8 configured blocks yield 7 usable.
+    expected = pytest.approx(num_used / 7)
+    assert 0 < num_used <= 7
+    for _ in range(2):
+        stats = scheduler.get_kv_connector_stats()
+        assert stats.data["host_cache_usage_perc"] == [expected]
+
+
 def test_no_forward_enqueues_deferred_hisparse_transfers():
     """A zero-token step must still enqueue deferred post-forward transfers."""
     connector = object.__new__(ActiveKVConnector)

tests/v1/kv_connector/unit/test_hisparse_stats.py

@@ -26,6 +26,28 @@ def test_record_snapshot_and_reduce():
     assert reduced["HiSparse host-to-device bytes"] == 64
 
 
+def test_record_host_usage_reports_latest_level():
+    stats = HiSparseKVConnectorStats()
+    stats.record_host_usage(usage=0.4, pending_page_transfers=2)
+    stats.record_host_usage(usage=0.8, pending_page_transfers=0)
+
+    reduced = stats.reduce()
+    assert reduced["HiSparse host KV cache usage %"] == 80.0
+
+
+def test_aggregate_keeps_latest_host_usage_level():
+    first = HiSparseKVConnectorStats()
+    first.record_host_usage(usage=0.4, pending_page_transfers=2)
+    second = HiSparseKVConnectorStats()
+    second.record_host_usage(usage=0.8, pending_page_transfers=0)
+    second.record_host_usage(usage=0.6, pending_page_transfers=1)
+
+    first.aggregate(second)
+
+    assert first.data["host_cache_usage_perc"] == [0.4, 0.6]
+    assert first.data["pending_page_transfers"] == [2, 1]
+
+
 def test_aggregate_extends_snapshot_deltas():
     first = HiSparseKVConnectorStats()
     first.record_snapshot(hits=7, misses=3, host_to_device_bytes=48)
@@ -34,11 +56,9 @@ def test_aggregate_extends_snapshot_deltas():
 
     first.aggregate(second)
 
-    assert first.data == {
-        "cache_hits": [7, 5],
-        "cache_misses": [3, 1],
-        "host_to_device_bytes": [48, 16],
-    }
+    assert first.data["cache_hits"] == [7, 5]
+    assert first.data["cache_misses"] == [3, 1]
+    assert first.data["host_to_device_bytes"] == [48, 16]
 
 
 def test_aggregate_skips_empty_stats():
@@ -60,3 +80,54 @@ def test_build_kv_connector_stats_round_trip():
     assert rebuilt is not None
     assert isinstance(rebuilt, HiSparseKVConnectorStats)
     assert rebuilt.reduce() == stats.reduce()
+
+
+def test_prom_metrics_observe_host_usage_gauges():
+    from types import SimpleNamespace
+    from typing import Any
+
+    from prometheus_client import Counter, Gauge, Histogram
+
+    from vllm.distributed.kv_transfer.kv_connector.v1.hisparse.stats import (
+        HiSparsePromMetrics,
+    )
+
+    class _FakeMetric:
+        def __init__(self, **kwargs: Any):
+            self.kwargs = kwargs
+            self.increments: list[int | float] = []
+            self.set_values: list[int | float] = []
+
+        def labels(self, *labelvalues: object) -> "_FakeMetric":
+            return self
+
+        def inc(self, value: int | float) -> None:
+            self.increments.append(value)
+
+        def set(self, value: int | float) -> None:
+            self.set_values.append(value)
+
+    created: dict[str, _FakeMetric] = {}
+
+    class _NamedFake(_FakeMetric):
+        def __init__(self, **kwargs: Any):
+            super().__init__(**kwargs)
+            created[kwargs["name"]] = self
+
+    prom = HiSparsePromMetrics(
+        vllm_config=SimpleNamespace(kv_transfer_config=None),
+        metric_types={Gauge: _NamedFake, Counter: _NamedFake, Histogram: _NamedFake},
+        labelnames=["model_name"],
+        per_engine_labelvalues={0: ["model"]},
+    )
+
+    stats = HiSparseKVConnectorStats()
+    stats.record_snapshot(hits=3, misses=2, host_to_device_bytes=32)
+    stats.record_host_usage(usage=0.5, pending_page_transfers=1)
+    stats.record_host_usage(usage=0.75, pending_page_transfers=3)
+    prom.observe(stats.to_dict())
+
+    assert created["vllm:hisparse_cache_hits"].increments == [3]
+    assert created["vllm:hisparse_host_to_device_bytes"].increments == [32]
+    assert created["vllm:hisparse_host_cache_usage_perc"].set_values == [0.75]
+    assert created["vllm:hisparse_pending_page_transfers"].set_values == [3]

tests/v1/worker/test_utils.py

@@ -92,6 +92,8 @@ def test_hisparse_worker_get_kv_connector_stats_reads_completed_snapshot(monkeyp
         "cache_hits": [12],
         "cache_misses": [4],
         "host_to_device_bytes": [64],
+        "host_cache_usage_perc": [],
+        "pending_page_transfers": [],
     }
 
 

vllm/distributed/kv_transfer/kv_connector/v1/hisparse/connector.py

@@ -30,10 +30,12 @@ from vllm.distributed.kv_transfer.kv_connector.v1.metrics import (
     PromMetric,
     PromMetricT,
 )
+from vllm.logger import init_logger
 from vllm.v1.attention.backend import AttentionMetadata
 from vllm.v1.core.kv_cache_utils import KVCacheBlockCopy
 from vllm.v1.core.sched.output import SchedulerOutput
 from vllm.v1.hisparse.coordinator import get_hisparse_coordinator
+from vllm.v1.hisparse.layout import get_hisparse_steady_state_concurrency
 from vllm.v1.hisparse.types import SparseKVOffloadCommand, SparseKVRowMirror
 from vllm.v1.outputs import KVConnectorOutput
 
@@ -44,6 +46,8 @@ if TYPE_CHECKING:
     from vllm.v1.kv_cache_interface import KVCacheConfig
     from vllm.v1.request import Request
 
+logger = init_logger(__name__)
+
 
 @dataclass
 class HiSparseConnectorMetadata(KVConnectorMetadata):
@@ -99,6 +103,17 @@ class HiSparseConnectorScheduler:
         assert self.coordinator is None
         self.coordinator = coordinator
 
+    def get_kv_connector_stats(self) -> HiSparseKVConnectorStats | None:
+        if self.coordinator is None:
+            return None
+        host_pool = self.coordinator.get_host_block_pool()
+        assert host_pool is not None
+        stats = HiSparseKVConnectorStats()
+        stats.record_host_usage(
+            host_pool.get_usage(), len(self.coordinator.pending_spills)
+        )
+        return stats
+
     def build_connector_meta(
         self, scheduler_output: SchedulerOutput
     ) -> HiSparseConnectorMetadata:
@@ -223,6 +238,16 @@ class HiSparseConnector(KVConnectorBase_V1, SupportsHMA):
                 ),
                 draft_kv_lookahead=vllm_config.num_lookahead_tokens,
             )
+            max_model_len = vllm_config.model_config.max_model_len
+            steady_concurrency = get_hisparse_steady_state_concurrency(
+                vllm_config, kv_cache_config
+            )
+            logger.info_once(
+                "HiSparse steady-state maximum concurrency for %s tokens per "
+                "request: %.2fx (running requests reading from host).",
+                f"{max_model_len:,}",
+                steady_concurrency,
+            )
         elif role == KVConnectorRole.WORKER:
             self.connector_worker = HiSparseConnectorWorker(
                 vllm_config, kv_cache_config
@@ -263,9 +288,11 @@ class HiSparseConnector(KVConnectorBase_V1, SupportsHMA):
         self.connector_worker.reset_hot_state()
 
     def get_kv_connector_stats(self) -> KVConnectorStats | None:
-        if self.connector_worker is None:
-            return None
-        return self.connector_worker.get_kv_connector_stats()
+        if self.connector_worker is not None:
+            return self.connector_worker.get_kv_connector_stats()
+        if self.connector_scheduler is not None:
+            return self.connector_scheduler.get_kv_connector_stats()
+        return None
 
     @classmethod
     def build_kv_connector_stats(

vllm/distributed/kv_transfer/kv_connector/v1/hisparse/stats.py

@@ -14,14 +14,7 @@ from vllm.distributed.kv_transfer.kv_connector.v1.metrics import (
 )
 from vllm.v1.metrics.utils import create_metric_per_engine
 
-_HISPARSE_COUNTERS: tuple[tuple[str, str], ...] = (
-    ("cache_hits", "Number of HiSparse device hot-buffer hits."),
-    ("cache_misses", "Number of HiSparse device hot-buffer misses."),
-    (
-        "host_to_device_bytes",
-        "Bytes transferred from host KV storage to HiSparse hot buffers.",
-    ),
-)
+_HISPARSE_LEVEL_KEYS = frozenset({"host_cache_usage_perc", "pending_page_transfers"})
 
 
 @dataclass
@@ -30,6 +23,8 @@ class HiSparseKVConnectorStats(KVConnectorStats):
 
     Each list entry is the delta recorded over one device counter snapshot
     interval, so a list can hold multiple snapshots per logging interval.
+    The host-tier gauges are level values, not deltas: aggregation keeps
+    only the most recent observation.
     """
 
     def __post_init__(self):
@@ -39,10 +34,12 @@ class HiSparseKVConnectorStats(KVConnectorStats):
 
     def reset(self):
         # Must be serializable
-        self.data: dict[str, list[int]] = {
+        self.data: dict[str, list[int | float]] = {
             "cache_hits": [],
             "cache_misses": [],
             "host_to_device_bytes": [],
+            "host_cache_usage_perc": [],
+            "pending_page_transfers": [],
         }
 
     def record_snapshot(self, hits: int, misses: int, host_to_device_bytes: int):
@@ -50,21 +47,32 @@ class HiSparseKVConnectorStats(KVConnectorStats):
         self.data["cache_misses"].append(misses)
         self.data["host_to_device_bytes"].append(host_to_device_bytes)
 
+    def record_host_usage(self, usage: float, pending_page_transfers: int) -> None:
+        self.data["host_cache_usage_perc"].append(usage)
+        self.data["pending_page_transfers"].append(pending_page_transfers)
+
     def aggregate(self, other: KVConnectorStats) -> KVConnectorStats:
         if not other.is_empty():
             for k, v in other.data.items():
                 accumulator = self.data[k]
                 assert isinstance(accumulator, list)
-                accumulator.extend(v)
+                if k in _HISPARSE_LEVEL_KEYS:
+                    accumulator.extend(v[-1:])
+                else:
+                    accumulator.extend(v)
         return self
 
     def reduce(self) -> dict[str, int | float]:
         # Compute compact representative stats suitable for CLI logging.
-        return {
+        reduced: dict[str, int | float] = {
             "HiSparse hot-buffer hits": sum(self.data["cache_hits"]),
             "HiSparse hot-buffer misses": sum(self.data["cache_misses"]),
             "HiSparse host-to-device bytes": sum(self.data["host_to_device_bytes"]),
         }
+        if self.data["host_cache_usage_perc"]:
+            usage = self.data["host_cache_usage_perc"][-1]
+            reduced["HiSparse host KV cache usage %"] = round(100 * usage, 1)
+        return reduced
 
     def is_empty(self) -> bool:
         return all(len(values) == 0 for values in self.data.values())
@@ -80,18 +88,66 @@ class HiSparsePromMetrics(KVConnectorPromMetrics):
     ):
         super().__init__(vllm_config, metric_types, labelnames, per_engine_labelvalues)
 
-        self.hisparse_counters: dict[str, Any] = {}
-        for name, documentation in _HISPARSE_COUNTERS:
-            counter = self._counter_cls(
-                name=f"vllm:hisparse_{name}",
-                documentation=documentation,
-                labelnames=labelnames,
-            )
-            self.hisparse_counters[name] = create_metric_per_engine(
-                counter, self.per_engine_labelvalues
-            )
+        counter_cache_hits = self._counter_cls(
+            name="vllm:hisparse_cache_hits",
+            documentation="Number of HiSparse device hot-buffer hits.",
+            labelnames=labelnames,
+        )
+        counter_cache_misses = self._counter_cls(
+            name="vllm:hisparse_cache_misses",
+            documentation="Number of HiSparse device hot-buffer misses.",
+            labelnames=labelnames,
+        )
+        counter_host_to_device_bytes = self._counter_cls(
+            name="vllm:hisparse_host_to_device_bytes",
+            documentation=(
+                "Bytes transferred from host KV storage to HiSparse hot buffers."
+            ),
+            labelnames=labelnames,
+        )
+        self.hisparse_counters: dict[str, Any] = {
+            "cache_hits": create_metric_per_engine(
+                counter_cache_hits, self.per_engine_labelvalues
+            ),
+            "cache_misses": create_metric_per_engine(
+                counter_cache_misses, self.per_engine_labelvalues
+            ),
+            "host_to_device_bytes": create_metric_per_engine(
+                counter_host_to_device_bytes, self.per_engine_labelvalues
+            ),
+        }
+
+        gauge_host_cache_usage = self._gauge_cls(
+            name="vllm:hisparse_host_cache_usage_perc",
+            documentation=(
+                "HiSparse host KV-cache usage. 1 means 100 percent usage. "
+                "Evictable cached blocks count as free."
+            ),
+            multiprocess_mode="mostrecent",
+            labelnames=labelnames,
+        )
+        gauge_pending_page_transfers = self._gauge_cls(
+            name="vllm:hisparse_pending_page_transfers",
+            documentation=(
+                "HiSparse page transfers (write-backs and restores) not yet completed."
+            ),
+            multiprocess_mode="mostrecent",
+            labelnames=labelnames,
+        )
+        self.hisparse_gauges: dict[str, Any] = {
+            "host_cache_usage_perc": create_metric_per_engine(
+                gauge_host_cache_usage, self.per_engine_labelvalues
+            ),
+            "pending_page_transfers": create_metric_per_engine(
+                gauge_pending_page_transfers, self.per_engine_labelvalues
+            ),
+        }
 
     def observe(self, transfer_stats_data: dict[str, Any], engine_idx: int = 0):
         for name, counter in self.hisparse_counters.items():
             for value in transfer_stats_data.get(name, []):
                 counter[engine_idx].inc(value)
+        for name, gauge in self.hisparse_gauges.items():
+            values = transfer_stats_data.get(name, [])
+            if values:
+                gauge[engine_idx].set(values[-1])

vllm/v1/hisparse/coordinator.py

@@ -19,6 +19,7 @@ from vllm.v1.core.single_type_kv_cache_manager import (
     SingleTypeKVCacheManager,
 )
 from vllm.v1.hisparse.types import (
+    ACTIVE_TAIL_PAGES,
     SparseKVOffloadCommand,
     SparseKVPageTransfer,
     SparseKVRowMirror,
@@ -32,10 +33,6 @@ from vllm.v1.request import Request
 if TYPE_CHECKING:
     from vllm.v1.core.kv_cache_manager import KVCacheManager
 
-# Sealed pages this many positions behind the block-table tail stay pinned so
-# a page written by an in-flight step is never handed out under it.
-_ACTIVE_TAIL_PAGES = 2
-
 
 @dataclass
 class _PendingPublication:
@@ -343,7 +340,7 @@ class HiSparseCoordinator:
             req_blocks = manager.req_to_blocks.get(request_id)
             if (
                 req_blocks is None
-                or page_idx >= len(req_blocks) - _ACTIVE_TAIL_PAGES
+                or page_idx >= len(req_blocks) - ACTIVE_TAIL_PAGES
                 or req_blocks[page_idx].is_null
             ):
                 return None

vllm/v1/hisparse/layout.py

@@ -14,6 +14,7 @@ from vllm.v1.hisparse.runtime import (
     get_hisparse_host_block_stride,
     use_shared_hisparse_host_pool,
 )
+from vllm.v1.hisparse.types import ACTIVE_TAIL_PAGES
 from vllm.v1.kv_cache_interface import (
     HiSparseHotSpec,
     HiSparseResidentSpec,
@@ -373,3 +374,34 @@ def get_hisparse_kv_cache_config(
             vllm_config.cache_config.prefix_cache_retention_interval
         ),
     )
+
+
+def get_hisparse_steady_state_concurrency(
+    vllm_config: VllmConfig, kv_cache_config: KVCacheConfig
+) -> float:
+    """Max concurrency at max_model_len once running requests read from host.
+
+    A request reading from host pins only its active tail of each resident
+    group, but admitting one still takes its full in-flight window, so the
+    bound is all-but-one requests at steady state plus one being admitted.
+    """
+    admission_blocks = 0
+    steady_blocks = 0
+    host_blocks = 0
+    for group in kv_cache_config.kv_cache_groups:
+        spec = group.kv_cache_spec
+        required = cdiv(spec.max_memory_usage_bytes(vllm_config), spec.page_size_bytes)
+        if group.host_resident:
+            host_blocks += required
+            continue
+        admission_blocks += required
+        if isinstance(spec, HiSparseResidentSpec):
+            required = min(required, ACTIVE_TAIL_PAGES)
+        steady_blocks += required
+    assert kv_cache_config.hisparse_host_num_blocks is not None
+    num_blocks = kv_cache_config.num_blocks
+    if num_blocks < admission_blocks:
+        gpu_concurrency = num_blocks / admission_blocks
+    else:
+        gpu_concurrency = 1 + (num_blocks - admission_blocks) / steady_blocks
+    return min(gpu_concurrency, kv_cache_config.hisparse_host_num_blocks / host_blocks)

vllm/v1/hisparse/types.py

@@ -3,6 +3,10 @@
 
 from dataclasses import dataclass
 
+# Sealed pages this many positions behind the block-table tail stay pinned so
+# a page written by an in-flight step is never handed out under it.
+ACTIVE_TAIL_PAGES = 2
+
 
 @dataclass(frozen=True)
 class SparseKVPageTransfer: