[HiSparse] Report host-pool block residency as HiSparse metrics (#59485)
Signed-off-by: Matthew Bonanni <mbonanni@redhat.com> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
783ac12071
commit
4c43254d52
@@ -50,7 +50,10 @@ The following metrics are exposed:
|
||||
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.
|
||||
blocks count as free. With `--kv-cache-metrics`, the
|
||||
`vllm:hisparse_host_block_*_seconds` histograms sample host-pool block
|
||||
residency, like the device `vllm:kv_block_*_seconds` histograms, which cover
|
||||
only the GPU pool.
|
||||
|
||||
--8<-- "gen:metrics-hisparse"
|
||||
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
import contextlib
|
||||
import ctypes
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
import torch
|
||||
@@ -25,6 +25,7 @@ from vllm.distributed.kv_transfer.kv_connector.v1.hisparse.worker import (
|
||||
HiSparseConnectorWorker,
|
||||
)
|
||||
from vllm.distributed.kv_transfer.kv_connector.v1.multi_connector import MultiConnector
|
||||
from vllm.v1.core.kv_cache_metrics import KVCacheMetricsCollector
|
||||
from vllm.v1.hisparse import runtime as runtime_module
|
||||
from vllm.v1.hisparse.coordinator import get_hisparse_coordinator
|
||||
from vllm.v1.hisparse.runtime import HiSparseCacheHandle
|
||||
@@ -33,6 +34,7 @@ from vllm.v1.hisparse.types import (
|
||||
SparseKVPageTransfer,
|
||||
SparseKVRowMirror,
|
||||
)
|
||||
from vllm.v1.metrics.stats import KVCacheEvictionEvent
|
||||
from vllm.v1.worker.gpu.kv_connector import ActiveKVConnector
|
||||
|
||||
|
||||
@@ -100,6 +102,46 @@ def test_scheduler_stats_report_host_pool_usage():
|
||||
assert stats.data["host_cache_usage_perc"] == [expected]
|
||||
|
||||
|
||||
def test_hisparse_host_residency_reported_as_hisparse_metrics():
|
||||
"""Host blocks sample into HiSparse's own residency metrics."""
|
||||
from tests.v1.core.test_prefix_caching import (
|
||||
make_hisparse_kv_cache_config,
|
||||
make_kv_cache_manager,
|
||||
)
|
||||
from tests.v1.core.utils import create_requests
|
||||
|
||||
collector = KVCacheMetricsCollector(sample_rate=1.0)
|
||||
manager = make_kv_cache_manager(
|
||||
make_hisparse_kv_cache_config(2, 2),
|
||||
max_model_len=128,
|
||||
enable_caching=True,
|
||||
hash_block_size=16,
|
||||
metrics_collector=collector,
|
||||
)
|
||||
coordinator = get_hisparse_coordinator(manager)
|
||||
assert coordinator.host_manager is not None
|
||||
scheduler = HiSparseConnectorScheduler(async_speculative=False)
|
||||
scheduler.bind_coordinator(coordinator)
|
||||
device = manager.block_pool
|
||||
host = coordinator.host_manager.block_pool
|
||||
request = create_requests(1, num_tokens=16)[0]
|
||||
for pool, birth in ((device, 1), (host, 2)):
|
||||
with patch("time.monotonic_ns", return_value=birth * 10**9):
|
||||
block = pool.get_new_blocks(1)[0]
|
||||
assert block.block_id == 1
|
||||
pool.cache_full_blocks(request, [block], 0, 1, 16, 0)
|
||||
with patch("time.monotonic_ns", return_value=4_000_000_000):
|
||||
host.free_blocks([host.blocks[1]])
|
||||
host.evict_blocks({1})
|
||||
with patch("time.monotonic_ns", return_value=6_000_000_000):
|
||||
device.evict_blocks({1})
|
||||
|
||||
assert collector.drain_events() == [KVCacheEvictionEvent(5.0, 5.0, ())]
|
||||
stats = scheduler.get_kv_connector_stats()
|
||||
assert stats.data["host_block_lifetime_seconds"] == [2.0]
|
||||
assert stats.data["host_block_idle_before_evict_seconds"] == [2.0]
|
||||
|
||||
|
||||
def test_no_forward_enqueues_deferred_hisparse_transfers():
|
||||
"""A zero-token step must still enqueue deferred post-forward transfers."""
|
||||
connector = object.__new__(ActiveKVConnector)
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
|
||||
|
||||
import pytest
|
||||
|
||||
from vllm.distributed.kv_transfer.kv_connector.v1.hisparse.connector import (
|
||||
HiSparseConnector,
|
||||
)
|
||||
@@ -82,7 +84,8 @@ def test_build_kv_connector_stats_round_trip():
|
||||
assert rebuilt.reduce() == stats.reduce()
|
||||
|
||||
|
||||
def test_prom_metrics_observe_host_usage_gauges():
|
||||
@pytest.mark.parametrize("kv_cache_metrics", [True, False])
|
||||
def test_prom_metrics_observe_host_metrics(kv_cache_metrics: bool):
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
@@ -91,12 +94,14 @@ def test_prom_metrics_observe_host_usage_gauges():
|
||||
from vllm.distributed.kv_transfer.kv_connector.v1.hisparse.stats import (
|
||||
HiSparsePromMetrics,
|
||||
)
|
||||
from vllm.v1.metrics.stats import KVCacheEvictionEvent
|
||||
|
||||
class _FakeMetric:
|
||||
def __init__(self, **kwargs: Any):
|
||||
self.kwargs = kwargs
|
||||
self.increments: list[int | float] = []
|
||||
self.set_values: list[int | float] = []
|
||||
self.observed: list[float] = []
|
||||
|
||||
def labels(self, *labelvalues: object) -> "_FakeMetric":
|
||||
return self
|
||||
@@ -107,6 +112,9 @@ def test_prom_metrics_observe_host_usage_gauges():
|
||||
def set(self, value: int | float) -> None:
|
||||
self.set_values.append(value)
|
||||
|
||||
def observe(self, value: float) -> None:
|
||||
self.observed.append(value)
|
||||
|
||||
created: dict[str, _FakeMetric] = {}
|
||||
|
||||
class _NamedFake(_FakeMetric):
|
||||
@@ -115,7 +123,12 @@ def test_prom_metrics_observe_host_usage_gauges():
|
||||
created[kwargs["name"]] = self
|
||||
|
||||
prom = HiSparsePromMetrics(
|
||||
vllm_config=SimpleNamespace(kv_transfer_config=None),
|
||||
vllm_config=SimpleNamespace(
|
||||
kv_transfer_config=None,
|
||||
observability_config=SimpleNamespace(
|
||||
kv_cache_metrics=kv_cache_metrics, custom_histogram_buckets=None
|
||||
),
|
||||
),
|
||||
metric_types={Gauge: _NamedFake, Counter: _NamedFake, Histogram: _NamedFake},
|
||||
labelnames=["model_name"],
|
||||
per_engine_labelvalues={0: ["model"]},
|
||||
@@ -125,9 +138,21 @@ def test_prom_metrics_observe_host_usage_gauges():
|
||||
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())
|
||||
stats.record_host_evictions([KVCacheEvictionEvent(5.0, 2.0, (1.0, 2.0))])
|
||||
later = HiSparseKVConnectorStats()
|
||||
later.record_host_evictions([KVCacheEvictionEvent(3.0, 3.0, ())])
|
||||
prom.observe(stats.aggregate(later).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]
|
||||
if not kv_cache_metrics:
|
||||
assert not any("host_block" in name for name in created)
|
||||
return
|
||||
assert created["vllm:hisparse_host_block_lifetime_seconds"].observed == [5.0, 3.0]
|
||||
assert created["vllm:hisparse_host_block_idle_before_evict_seconds"].observed == [
|
||||
2.0,
|
||||
3.0,
|
||||
]
|
||||
assert created["vllm:hisparse_host_block_reuse_gap_seconds"].observed == [1.0, 2.0]
|
||||
|
||||
@@ -94,6 +94,9 @@ def test_hisparse_worker_get_kv_connector_stats_reads_completed_snapshot(monkeyp
|
||||
"host_to_device_bytes": [64],
|
||||
"host_cache_usage_perc": [],
|
||||
"pending_page_transfers": [],
|
||||
"host_block_lifetime_seconds": [],
|
||||
"host_block_idle_before_evict_seconds": [],
|
||||
"host_block_reuse_gap_seconds": [],
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -112,6 +112,8 @@ class HiSparseConnectorScheduler:
|
||||
stats.record_host_usage(
|
||||
host_pool.get_usage(), len(self.coordinator.pending_spills)
|
||||
)
|
||||
if host_pool.metrics_collector is not None:
|
||||
stats.record_host_evictions(host_pool.metrics_collector.drain_events())
|
||||
return stats
|
||||
|
||||
def build_connector_meta(
|
||||
|
||||
@@ -12,6 +12,8 @@ from vllm.distributed.kv_transfer.kv_connector.v1.metrics import (
|
||||
PromMetric,
|
||||
PromMetricT,
|
||||
)
|
||||
from vllm.v1.metrics.buckets import histogram_buckets
|
||||
from vllm.v1.metrics.stats import KVCacheEvictionEvent
|
||||
from vllm.v1.metrics.utils import create_metric_per_engine
|
||||
|
||||
_HISPARSE_LEVEL_KEYS = frozenset({"host_cache_usage_perc", "pending_page_transfers"})
|
||||
@@ -40,6 +42,9 @@ class HiSparseKVConnectorStats(KVConnectorStats):
|
||||
"host_to_device_bytes": [],
|
||||
"host_cache_usage_perc": [],
|
||||
"pending_page_transfers": [],
|
||||
"host_block_lifetime_seconds": [],
|
||||
"host_block_idle_before_evict_seconds": [],
|
||||
"host_block_reuse_gap_seconds": [],
|
||||
}
|
||||
|
||||
def record_snapshot(self, hits: int, misses: int, host_to_device_bytes: int):
|
||||
@@ -51,6 +56,12 @@ class HiSparseKVConnectorStats(KVConnectorStats):
|
||||
self.data["host_cache_usage_perc"].append(usage)
|
||||
self.data["pending_page_transfers"].append(pending_page_transfers)
|
||||
|
||||
def record_host_evictions(self, events: list[KVCacheEvictionEvent]) -> None:
|
||||
for event in events:
|
||||
self.data["host_block_lifetime_seconds"].append(event.lifetime_seconds)
|
||||
self.data["host_block_idle_before_evict_seconds"].append(event.idle_seconds)
|
||||
self.data["host_block_reuse_gap_seconds"].extend(event.reuse_gaps_seconds)
|
||||
|
||||
def aggregate(self, other: KVConnectorStats) -> KVConnectorStats:
|
||||
if not other.is_empty():
|
||||
for k, v in other.data.items():
|
||||
@@ -143,6 +154,53 @@ class HiSparsePromMetrics(KVConnectorPromMetrics):
|
||||
),
|
||||
}
|
||||
|
||||
self.hisparse_histograms: dict[str, Any] = {}
|
||||
if vllm_config.observability_config.kv_cache_metrics:
|
||||
residency_buckets = histogram_buckets(
|
||||
"kv_cache_residency",
|
||||
overrides=vllm_config.observability_config.custom_histogram_buckets,
|
||||
)
|
||||
histogram_host_block_lifetime = self._histogram_cls(
|
||||
name="vllm:hisparse_host_block_lifetime_seconds",
|
||||
documentation=(
|
||||
"Histogram of HiSparse host KV block lifetime from allocation "
|
||||
"to eviction. Sampled metrics (controlled by "
|
||||
"--kv-cache-metrics-sample)."
|
||||
),
|
||||
buckets=residency_buckets,
|
||||
labelnames=labelnames,
|
||||
)
|
||||
histogram_host_block_idle_before_evict = self._histogram_cls(
|
||||
name="vllm:hisparse_host_block_idle_before_evict_seconds",
|
||||
documentation=(
|
||||
"Histogram of HiSparse host KV block idle time before eviction. "
|
||||
"Sampled metrics (controlled by --kv-cache-metrics-sample)."
|
||||
),
|
||||
buckets=residency_buckets,
|
||||
labelnames=labelnames,
|
||||
)
|
||||
histogram_host_block_reuse_gap = self._histogram_cls(
|
||||
name="vllm:hisparse_host_block_reuse_gap_seconds",
|
||||
documentation=(
|
||||
"Histogram of time gaps between consecutive HiSparse host KV "
|
||||
"block accesses. Sampled metrics (controlled by "
|
||||
"--kv-cache-metrics-sample)."
|
||||
),
|
||||
buckets=residency_buckets,
|
||||
labelnames=labelnames,
|
||||
)
|
||||
self.hisparse_histograms = {
|
||||
"host_block_lifetime_seconds": create_metric_per_engine(
|
||||
histogram_host_block_lifetime, self.per_engine_labelvalues
|
||||
),
|
||||
"host_block_idle_before_evict_seconds": create_metric_per_engine(
|
||||
histogram_host_block_idle_before_evict, self.per_engine_labelvalues
|
||||
),
|
||||
"host_block_reuse_gap_seconds": create_metric_per_engine(
|
||||
histogram_host_block_reuse_gap, 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, []):
|
||||
@@ -151,3 +209,6 @@ class HiSparsePromMetrics(KVConnectorPromMetrics):
|
||||
values = transfer_stats_data.get(name, [])
|
||||
if values:
|
||||
gauge[engine_idx].set(values[-1])
|
||||
for name, histogram in self.hisparse_histograms.items():
|
||||
for value in transfer_stats_data.get(name, []):
|
||||
histogram[engine_idx].observe(value)
|
||||
|
||||
@@ -9,6 +9,7 @@ from typing import TYPE_CHECKING, ClassVar
|
||||
from vllm.distributed.kv_events import MEDIUM_CPU
|
||||
from vllm.utils.math_utils import cdiv
|
||||
from vllm.v1.core.block_pool import BlockPool
|
||||
from vllm.v1.core.kv_cache_metrics import KVCacheMetricsCollector
|
||||
from vllm.v1.core.kv_cache_utils import (
|
||||
BlockHashList,
|
||||
BlockHashListWithBlockSize,
|
||||
@@ -2272,7 +2273,11 @@ class HiSparseSourceManager(FullAttentionManager):
|
||||
enable_caching=self.enable_caching and self.kv_cache_spec.prefix_cacheable,
|
||||
hash_block_size=device_pool.hash_block_size,
|
||||
enable_kv_cache_events=device_pool.enable_kv_cache_events,
|
||||
metrics_collector=None,
|
||||
metrics_collector=(
|
||||
KVCacheMetricsCollector(device_pool.metrics_collector.sample_rate)
|
||||
if device_pool.metrics_collector is not None
|
||||
else None
|
||||
),
|
||||
medium=MEDIUM_CPU,
|
||||
event_owner=device_pool,
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user