Skip to content

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

Stats and Prometheus metrics for the HiSparse connector.

Classes:

HiSparseKVConnectorStats dataclass

Bases: KVConnectorStats

Container for HiSparse hot-buffer residency metrics.

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.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/hisparse/stats.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
@dataclass
class HiSparseKVConnectorStats(KVConnectorStats):
    """Container for HiSparse hot-buffer residency metrics.

    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):
        if not self.data:
            # Empty container init, no data is passed in.
            self.reset()

    def reset(self):
        # Must be serializable
        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):
        self.data["cache_hits"].append(hits)
        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)
                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.
        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())