diff --git a/docs/design/hisparse.md b/docs/design/hisparse.md index 8651b3a06d..f5fab3c0f5 100644 --- a/docs/design/hisparse.md +++ b/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 | diff --git a/docs/mkdocs/gen_files/generate_metrics.py b/docs/mkdocs/gen_files/generate_metrics.py index 4d4fa93cc5..32bbefa88c 100644 --- a/docs/mkdocs/gen_files/generate_metrics.py +++ b/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", diff --git a/docs/usage/metrics.md b/docs/usage/metrics.md index 5cb5eebc55..b87ff1c28f 100644 --- a/docs/usage/metrics.md +++ b/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 diff --git a/tests/v1/core/test_kv_cache_utils.py b/tests/v1/core/test_kv_cache_utils.py index 08db9a286b..7a34d07c07 100644 --- a/tests/v1/core/test_kv_cache_utils.py +++ b/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.""" diff --git a/tests/v1/kv_connector/unit/test_hisparse_connector.py b/tests/v1/kv_connector/unit/test_hisparse_connector.py index a2eeb8cd63..f08982a5e9 100644 --- a/tests/v1/kv_connector/unit/test_hisparse_connector.py +++ b/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) diff --git a/tests/v1/kv_connector/unit/test_hisparse_stats.py b/tests/v1/kv_connector/unit/test_hisparse_stats.py index 09904d0a3c..91d00b1e22 100644 --- a/tests/v1/kv_connector/unit/test_hisparse_stats.py +++ b/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] diff --git a/tests/v1/worker/test_utils.py b/tests/v1/worker/test_utils.py index abcbdb24f2..f6f7227954 100644 --- a/tests/v1/worker/test_utils.py +++ b/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": [], } diff --git a/vllm/distributed/kv_transfer/kv_connector/v1/hisparse/connector.py b/vllm/distributed/kv_transfer/kv_connector/v1/hisparse/connector.py index dabf9095db..7e8960f644 100644 --- a/vllm/distributed/kv_transfer/kv_connector/v1/hisparse/connector.py +++ b/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( diff --git a/vllm/distributed/kv_transfer/kv_connector/v1/hisparse/stats.py b/vllm/distributed/kv_transfer/kv_connector/v1/hisparse/stats.py index f2abf51fc4..6f6cfced9e 100644 --- a/vllm/distributed/kv_transfer/kv_connector/v1/hisparse/stats.py +++ b/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]) diff --git a/vllm/v1/hisparse/coordinator.py b/vllm/v1/hisparse/coordinator.py index bc9bff1a8b..766c926b12 100644 --- a/vllm/v1/hisparse/coordinator.py +++ b/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 diff --git a/vllm/v1/hisparse/layout.py b/vllm/v1/hisparse/layout.py index 8ebc63c0ed..98636866f8 100644 --- a/vllm/v1/hisparse/layout.py +++ b/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) diff --git a/vllm/v1/hisparse/types.py b/vllm/v1/hisparse/types.py index 269793ea85..db0ee06e93 100644 --- a/vllm/v1/hisparse/types.py +++ b/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: