diff --git a/tests/v1/kv_offload/cpu/test_manager.py b/tests/v1/kv_offload/cpu/test_manager.py index 6e4cbb1c6b8..aa4fb829597 100644 --- a/tests/v1/kv_offload/cpu/test_manager.py +++ b/tests/v1/kv_offload/cpu/test_manager.py @@ -14,12 +14,13 @@ from vllm.v1.kv_offload.base import ( ReqContext, make_offload_key, ) -from vllm.v1.kv_offload.cpu.common import CPULoadStoreSpec +from vllm.v1.kv_offload.cpu.common import ( + CPULoadStoreSpec, + CPUOffloadingMetrics, +) from vllm.v1.kv_offload.cpu.manager import CPUOffloadingManager from vllm.v1.kv_offload.cpu.policies.arc import ARCCachePolicy -STORES_SKIPPED = "vllm:kv_offload_stores_skipped" - def make_req_context( req_id: str = "", kv_transfer_params: dict | None = None @@ -181,10 +182,45 @@ def test_filter_reused_manager_reports_stores_skipped_counter(): ) stats = manager.get_stats() assert stats is not None - assert stats.reduce()[STORES_SKIPPED] == 3 + assert stats.reduce()[CPUOffloadingMetrics.STORES_SKIPPED] == 3 stats = manager.get_stats() assert stats is not None - assert stats.reduce()[STORES_SKIPPED] == 0 + assert stats.reduce()[CPUOffloadingMetrics.STORES_SKIPPED] == 0 + + +def test_cpu_manager_reports_cache_usage_gauge(): + def check_usage_stats(manager: CPUOffloadingManager, value: float): + stats = manager.get_stats() + assert stats is not None + assert stats.reduce()[ + CPUOffloadingMetrics.CPU_CACHE_USAGE_PERC + ] == pytest.approx(value) + + # Zero-capacity manager always reports 0.0 + manager = make_cpu_manager(num_blocks=0) + check_usage_stats(manager, 0.0) + + # Empty manager (4 blocks, none allocated): usage = 0.0 + manager = make_cpu_manager(num_blocks=4) + check_usage_stats(manager, 0.0) + + # After allocating 2 of 4 blocks: usage = 0.5 + manager.prepare_store(to_keys([1, 2]), _EMPTY_REQ_CTX) + check_usage_stats(manager, 0.5) + + # After filling all 4 blocks: usage = 1.0 + manager.prepare_store(to_keys([3, 4]), _EMPTY_REQ_CTX) + check_usage_stats(manager, 1.0) + + # After completing store, the blocks becomes evictable as it is not actively used + # and usage drops. + manager.complete_store(to_keys([1, 2]), _EMPTY_REQ_CTX) + check_usage_stats(manager, 0.5) + + # After completing store, the blocks becomes evictable as it is not actively used + # and usage drops. + manager.complete_store(to_keys([3, 4]), _EMPTY_REQ_CTX) + check_usage_stats(manager, 0.0) def test_cpu_manager(): diff --git a/vllm/v1/kv_offload/cpu/common.py b/vllm/v1/kv_offload/cpu/common.py index 46bca1b9065..14c96680fd0 100644 --- a/vllm/v1/kv_offload/cpu/common.py +++ b/vllm/v1/kv_offload/cpu/common.py @@ -4,7 +4,10 @@ from typing_extensions import override from vllm.v1.kv_offload.base import BlockIDsLoadStoreSpec -METRIC_STORES_SKIPPED = "vllm:kv_offload_stores_skipped" + +class CPUOffloadingMetrics: + STORES_SKIPPED = "vllm:kv_offload_stores_skipped" + CPU_CACHE_USAGE_PERC = "vllm:kv_offload_cpu_cache_usage_perc" class CPULoadStoreSpec(BlockIDsLoadStoreSpec): diff --git a/vllm/v1/kv_offload/cpu/manager.py b/vllm/v1/kv_offload/cpu/manager.py index 7835d35309a..7d92844d1f4 100644 --- a/vllm/v1/kv_offload/cpu/manager.py +++ b/vllm/v1/kv_offload/cpu/manager.py @@ -18,7 +18,10 @@ from vllm.v1.kv_offload.base import ( ReqContext, RequestOffloadingContext, ) -from vllm.v1.kv_offload.cpu.common import METRIC_STORES_SKIPPED, CPULoadStoreSpec +from vllm.v1.kv_offload.cpu.common import ( + CPULoadStoreSpec, + CPUOffloadingMetrics, +) from vllm.v1.kv_offload.cpu.policies.arc import ARCCachePolicy from vllm.v1.kv_offload.cpu.policies.base import BlockStatus, CachePolicy from vllm.v1.kv_offload.cpu.policies.lru import LRUCachePolicy @@ -282,13 +285,21 @@ class CPUOffloadingManager(OffloadingManager): self.events.clear() def get_stats(self) -> OffloadingConnectorStats | None: - if self.store_threshold < 2: - return None - stats = OffloadingConnectorStats() - stats.increase_counter( - METRIC_STORES_SKIPPED, - self.stores_skipped_in_current_batch, + + # Compute cache usage. + num_used = ( + self._num_allocated_blocks + - len(self._free_list) + - self._num_evictable_cache_blocks ) - self.stores_skipped_in_current_batch = 0 + usage = num_used / self._num_blocks if self._num_blocks > 0 else 0.0 + stats.set_gauge(CPUOffloadingMetrics.CPU_CACHE_USAGE_PERC, usage) + + if self.store_threshold >= 2: + stats.increase_counter( + CPUOffloadingMetrics.STORES_SKIPPED, + self.stores_skipped_in_current_batch, + ) + self.stores_skipped_in_current_batch = 0 return stats diff --git a/vllm/v1/kv_offload/cpu/spec.py b/vllm/v1/kv_offload/cpu/spec.py index b8fb893f14d..9b1dff24a87 100644 --- a/vllm/v1/kv_offload/cpu/spec.py +++ b/vllm/v1/kv_offload/cpu/spec.py @@ -14,11 +14,15 @@ from vllm.v1.kv_offload.base import ( GPULoadStoreSpec, LoadStoreSpec, OffloadingCounterMetadata, + OffloadingGaugeMetadata, OffloadingManager, OffloadingMetricMetadata, OffloadingSpec, ) -from vllm.v1.kv_offload.cpu.common import METRIC_STORES_SKIPPED, CPULoadStoreSpec +from vllm.v1.kv_offload.cpu.common import ( + CPULoadStoreSpec, + CPUOffloadingMetrics, +) from vllm.v1.kv_offload.cpu.gpu_worker import CpuGpuOffloadingHandlers from vllm.v1.kv_offload.cpu.manager import CPUOffloadingManager from vllm.v1.kv_offload.worker.worker import OffloadingHandler @@ -31,17 +35,27 @@ class CPUOffloadingSpec(OffloadingSpec): def build_metric_definitions( cls, extra_config: dict[str, Any] ) -> dict[str, OffloadingMetricMetadata]: - store_threshold = int(extra_config.get("store_threshold", 0)) - if store_threshold < 2: - return {} - return { - METRIC_STORES_SKIPPED: OffloadingCounterMetadata( + definitions: dict[str, OffloadingMetricMetadata] = { + CPUOffloadingMetrics.CPU_CACHE_USAGE_PERC: OffloadingGaugeMetadata( documentation=( - "Number of KV offload stores skipped because the reuse " - "threshold was not reached." + "Fraction of CPU KV-cache space currently pinned by active " + "transfers (0.0 = idle, 1.0 = saturated). Sustained high " + "values indicate transfers (stores or promotions) may be " + "dropped due to insufficient capacity." ), ) } + store_threshold = int(extra_config.get("store_threshold", 0)) + if store_threshold >= 2: + definitions[CPUOffloadingMetrics.STORES_SKIPPED] = ( + OffloadingCounterMetadata( + documentation=( + "Number of KV offload stores skipped because the reuse " + "threshold was not reached." + ), + ) + ) + return definitions def __init__(self, vllm_config: VllmConfig, kv_cache_config: KVCacheConfig): super().__init__(vllm_config, kv_cache_config)