[KV Offload] Split tiering_lookup_delay into sync/async histograms (#47679)
Signed-off-by: srinivas_oo7 <sklinkedin0120@gmail.com> Co-authored-by: srinivas_oo7 <sklinkedin0120@gmail.com>
Srinivas Krovvidi committed
Jul 16, 2026 at 12:03 UTC
ce653856186e80b2a927a611cb184aeac64d2862
4 files changed
+208
-5
tests/v1/kv_offload/tiering/test_tiering_offloading.py
+84
-2
@@ -35,6 +35,7 @@ from vllm.v1.kv_offload.tiering.base import (
35
JobMetadata,
36
JobResult,
37
SecondaryTierManager,
38
+ TieringOffloadingMetrics,
39
)
40
from vllm.v1.kv_offload.tiering.example.manager import ExampleSecondaryTierManager
41
from vllm.v1.kv_offload.tiering.factory import SecondaryTierFactory
@@ -233,9 +234,9 @@ class TestTieringOffloadingManager:
234
secondary_tiers=[self.secondary_tier1, self.secondary_tier2],
235
)
236
236
- def _simulate_on_schedule_end(self):
237
+ def _simulate_on_schedule_end(self, new_req_ids: list[str] | None = None):
238
"""Simulate end of scheduler step: lifecycle flush + drain events."""
238
- ctx = ScheduleEndContext(new_req_ids=[], preempted_req_ids=())
239
+ ctx = ScheduleEndContext(new_req_ids=new_req_ids or [], preempted_req_ids=())
240
self.manager.on_schedule_end(ctx)
241
list(self.manager.take_events())
242
@@ -395,6 +396,87 @@ class TestTieringOffloadingManager:
396
# Next lookup should succeed
397
assert count_hits(self.manager, blocks) == 3
398
399
+ def test_lookup_reports_sync_delay_for_resolved_lookups(self, manager_setup):
400
+ """Resolved lookups report one sync delay sample on allocation."""
401
+ self._start_request()
402
+ blocks = to_keys(range(2))
403
+
404
+ # No tier has these blocks: they resolve immediately as misses.
405
+ for block in blocks:
406
+ assert self.manager.lookup(block, _CTX) is LookupResult.MISS
407
+
408
+ stats = self.manager.get_stats()
409
+ if stats is not None:
410
+ assert f"{TieringOffloadingMetrics.LOOKUP_SYNC_DELAY}_count" not in (
411
+ stats.reduce()
412
+ )
413
+
414
+ self._simulate_on_schedule_end(new_req_ids=[_CTX.req_id])
415
+
416
+ stats = self.manager.get_stats()
417
+ assert stats is not None
418
+ reduced = stats.reduce()
419
+ assert reduced[f"{TieringOffloadingMetrics.LOOKUP_SYNC_DELAY}_count"] == 1
420
+ assert f"{TieringOffloadingMetrics.LOOKUP_ASYNC_DELAY}_count" not in reduced
421
+
422
+ def test_lookup_reports_async_delay_across_promotion(self, manager_setup):
423
+ """A new request reports async delay at schedule end."""
424
+ self._start_request()
425
+ block = to_keys(range(1))[0]
426
+ self.secondary_tier1.blocks[block] = True
427
+
428
+ # First lookup finds the block in a secondary tier and defers.
429
+ assert self.manager.lookup(block, _CTX) is LookupResult.RETRY
430
+
431
+ # The first scheduler step reports async delay for the new request.
432
+ self._simulate_on_schedule_end(new_req_ids=[_CTX.req_id])
433
+ stats = self.manager.get_stats()
434
+ assert stats is not None
435
+ reduced = stats.reduce()
436
+ assert reduced[f"{TieringOffloadingMetrics.LOOKUP_SYNC_DELAY}_count"] == 1
437
+ assert reduced[f"{TieringOffloadingMetrics.LOOKUP_ASYNC_DELAY}_count"] == 1
438
+
439
+ # Promotion completes on the next scheduler step.
440
+ self._simulate_on_schedule_end()
441
+
442
+ # Next lookup resolves via the now-promoted primary-tier block.
443
+ assert self.manager.lookup(block, _CTX) is LookupResult.HIT
444
+
445
+ stats = self.manager.get_stats()
446
+ if stats is not None:
447
+ assert f"{TieringOffloadingMetrics.LOOKUP_ASYNC_DELAY}_count" not in (
448
+ stats.reduce()
449
+ )
450
+
451
+ def test_lookup_reports_async_delay_on_request_finish(self, manager_setup):
452
+ """Never-allocated lookup delays flush at teardown."""
453
+ ctx = ReqContext(req_id="req_lookup_finish")
454
+ self._start_request(ctx)
455
+ block = to_keys(range(1))[0]
456
+ self.secondary_tier1.blocks[block] = True
457
+
458
+ # Lookup finds the block in a secondary tier and defers.
459
+ assert self.manager.lookup(block, ctx) is LookupResult.RETRY
460
+
461
+ self._simulate_on_schedule_end()
462
+ stats = self.manager.get_stats()
463
+ if stats is not None:
464
+ assert f"{TieringOffloadingMetrics.LOOKUP_SYNC_DELAY}_count" not in (
465
+ stats.reduce()
466
+ )
467
+ assert f"{TieringOffloadingMetrics.LOOKUP_ASYNC_DELAY}_count" not in (
468
+ stats.reduce()
469
+ )
470
+
471
+ # Request finishes before the deferred lookup is ever resolved.
472
+ self.manager.on_request_finished(ctx)
473
+
474
+ stats = self.manager.get_stats()
475
+ assert stats is not None
476
+ reduced = stats.reduce()
477
+ assert reduced[f"{TieringOffloadingMetrics.LOOKUP_SYNC_DELAY}_count"] == 1
478
+ assert reduced[f"{TieringOffloadingMetrics.LOOKUP_ASYNC_DELAY}_count"] == 1
479
+
480
def test_partial_lookup(self, manager_setup):
481
"""Test lookup with partial hits."""
482
blocks = to_keys(range(5))
vllm/v1/kv_offload/tiering/base.py
+7
@@ -31,6 +31,13 @@ if TYPE_CHECKING:
31
JobId = int
32
33
34
+class TieringOffloadingMetrics:
35
+ """Metric names for TieringOffloadingManager."""
36
+
37
+ LOOKUP_SYNC_DELAY = "vllm:kv_offload_tiering_lookup_sync_delay_seconds"
38
+ LOOKUP_ASYNC_DELAY = "vllm:kv_offload_tiering_lookup_async_delay_seconds"
39
+
40
+
41
@dataclass
42
class JobMetadata:
43
"""Metadata for an in-flight async transfer job."""
vllm/v1/kv_offload/tiering/manager.py
+71
-3
@@ -20,6 +20,7 @@ Key Design Principles:
20
protecting blocks from eviction until complete_read() is called
21
"""
22
23
+import time
24
from collections.abc import Collection, Iterable, Sequence
25
from dataclasses import dataclass, field
26
@@ -50,6 +51,7 @@ from vllm.v1.kv_offload.tiering.base import (
51
JobMetadata,
52
ParentManager,
53
SecondaryTierManager,
54
+ TieringOffloadingMetrics,
55
)
56
57
logger = init_logger(__name__)
@@ -70,6 +72,10 @@ class RequestState:
72
pending_primary_stores: int = 0
73
is_finished: bool = False
74
request_level_tiers: set[SecondaryTierManager] | None = None
75
+ sync_lookup_delay: float = 0.0
76
+ # time.monotonic() of this request's first deferred secondary-tier lookup;
77
+ # None once consumed (observed) or while no secondary lookup is pending.
78
+ secondary_lookup_start_time: float | None = None
79
80
81
class CPUPrimaryTierOffloadingManager(CPUOffloadingManager):
@@ -212,6 +218,10 @@ class TieringOffloadingManager(OffloadingManager):
218
for tier in self.secondary_tiers
219
}
220
221
+ # Buffers manager-level observations (e.g. lookup delay) between
222
+ # get_stats() calls; merged in and reset each time get_stats() runs.
223
+ self._stats = OffloadingConnectorStats()
224
+
225
def _next_job_id(self) -> JobId:
226
"""Generate a unique job ID for async transfer tracking."""
227
job_id = self._job_id_counter
@@ -301,28 +311,68 @@ class TieringOffloadingManager(OffloadingManager):
311
# in time for a promotion this lookup may initiate.
312
self._maybe_process_finished_jobs()
313
314
+ req_state = self._req_state.get(req_context.req_id)
315
+
316
primary_hit = self.primary_tier.lookup(key, req_context)
317
if primary_hit is LookupResult.HIT:
318
return LookupResult.HIT
319
if primary_hit is LookupResult.HIT_PENDING:
320
return LookupResult.HIT_PENDING
321
322
+ lookup_start = time.monotonic()
323
any_retry = False
324
for tier in self.secondary_tiers:
325
if tier is exclude_tier:
326
continue
327
result = tier.lookup(key, req_context)
328
if result is LookupResult.HIT:
316
- if not self._initiate_promotion(tier, key, req_context):
317
- return LookupResult.MISS
318
- return LookupResult.RETRY
329
+ promoted = self._initiate_promotion(tier, key, req_context)
330
+ self._accumulate_lookup_sync_delay(req_state, lookup_start)
331
+ if (
332
+ req_state is not None
333
+ and promoted
334
+ and req_state.secondary_lookup_start_time is None
335
+ ):
336
+ req_state.secondary_lookup_start_time = lookup_start
337
+ return LookupResult.MISS if not promoted else LookupResult.RETRY
338
if result is LookupResult.RETRY:
339
any_retry = True
340
341
+ self._accumulate_lookup_sync_delay(req_state, lookup_start)
342
if any_retry:
343
+ if req_state is not None and req_state.secondary_lookup_start_time is None:
344
+ req_state.secondary_lookup_start_time = lookup_start
345
return LookupResult.RETRY
346
return LookupResult.MISS
347
348
+ def _accumulate_lookup_sync_delay(
349
+ self, req_state: RequestState | None, start_time: float
350
+ ) -> None:
351
+ """Accumulate secondary-tier lookup time until allocation or finish."""
352
+ if req_state is not None:
353
+ req_state.sync_lookup_delay += time.monotonic() - start_time
354
+
355
+ def _maybe_observe_lookup_sync_delay(self, req_state: RequestState) -> None:
356
+ delay = req_state.sync_lookup_delay
357
+ if delay == 0:
358
+ return
359
+ req_state.sync_lookup_delay = 0.0
360
+ self._stats.observe_histogram(
361
+ TieringOffloadingMetrics.LOOKUP_SYNC_DELAY,
362
+ delay,
363
+ )
364
+
365
+ def _maybe_observe_lookup_async_delay(self, req_state: RequestState) -> None:
366
+ """Flush a pending deferred secondary-tier lookup timer, if any."""
367
+ start_time = req_state.secondary_lookup_start_time
368
+ if start_time is None:
369
+ return
370
+ req_state.secondary_lookup_start_time = None
371
+ self._stats.observe_histogram(
372
+ TieringOffloadingMetrics.LOOKUP_ASYNC_DELAY,
373
+ time.monotonic() - start_time,
374
+ )
375
+
376
def _initiate_promotion(
377
self,
378
tier: SecondaryTierManager,
@@ -666,6 +716,8 @@ class TieringOffloadingManager(OffloadingManager):
716
if tier is exclude_tier:
717
continue
718
tier.on_request_finished(state.req_context)
719
+ self._maybe_observe_lookup_sync_delay(state)
720
+ self._maybe_observe_lookup_async_delay(state)
721
del self._req_state[req_id]
722
723
@override
@@ -692,6 +744,13 @@ class TieringOffloadingManager(OffloadingManager):
744
for tier in self.secondary_tiers:
745
tier.on_schedule_end(context)
746
747
+ for req_id in context.new_req_ids:
748
+ state = self._req_state.get(req_id)
749
+ if state is None:
750
+ continue
751
+ self._maybe_observe_lookup_sync_delay(state)
752
+ self._maybe_observe_lookup_async_delay(state)
753
+
754
@override
755
def has_pending_work(self) -> bool:
756
# In-flight primary<->secondary transfers (pending promotions are
@@ -745,6 +804,8 @@ class TieringOffloadingManager(OffloadingManager):
804
continue
805
for tier in self.secondary_tiers:
806
tier.on_request_finished(state.req_context)
807
+ self._maybe_observe_lookup_sync_delay(state)
808
+ self._maybe_observe_lookup_async_delay(state)
809
finished_req_ids.append(req_id)
810
811
self.primary_tier.reset_cache()
@@ -769,6 +830,13 @@ class TieringOffloadingManager(OffloadingManager):
830
else:
831
stats.aggregate(tier_stats)
832
833
+ if not self._stats.is_empty():
834
+ if stats is None:
835
+ stats = self._stats
836
+ else:
837
+ stats.aggregate(self._stats)
838
+ self._stats = OffloadingConnectorStats()
839
+
840
return stats
841
842
@override
vllm/v1/kv_offload/tiering/spec.py
+46
@@ -39,6 +39,7 @@ from typing_extensions import override
39
from vllm.logger import init_logger
40
from vllm.v1.kv_offload.base import (
41
CanonicalKVCaches,
42
+ OffloadingHistogramMetadata,
43
OffloadingManager,
44
OffloadingMetricMetadata,
45
)
@@ -46,6 +47,7 @@ from vllm.v1.kv_offload.config import OffloadingConfig
47
from vllm.v1.kv_offload.cpu.gpu_worker import CPUOffloadingWorker
48
from vllm.v1.kv_offload.cpu.shared_offload_region import SharedOffloadRegion
49
from vllm.v1.kv_offload.cpu.spec import CPUOffloadingSpec
50
+from vllm.v1.kv_offload.tiering.base import TieringOffloadingMetrics
51
from vllm.v1.kv_offload.tiering.factory import SecondaryTierFactory
52
from vllm.v1.kv_offload.tiering.manager import (
53
CPUPrimaryTierOffloadingManager,
@@ -76,6 +78,50 @@ class TieringOffloadingSpec(CPUOffloadingSpec):
78
cls, extra_config: dict[str, Any]
79
) -> dict[str, OffloadingMetricMetadata]:
80
metrics = super().build_metric_definitions(extra_config)
81
+ metrics[TieringOffloadingMetrics.LOOKUP_SYNC_DELAY] = (
82
+ OffloadingHistogramMetadata(
83
+ documentation=(
84
+ "Histogram of total blocking time spent querying secondary "
85
+ "tiers for a request, accumulated from first lookup until "
86
+ "the request is allocated or finishes, in seconds."
87
+ ),
88
+ buckets=(
89
+ 0.00001,
90
+ 0.00005,
91
+ 0.0001,
92
+ 0.0005,
93
+ 0.001,
94
+ 0.005,
95
+ 0.01,
96
+ 0.05,
97
+ 0.1,
98
+ 0.5,
99
+ 1,
100
+ ),
101
+ )
102
+ )
103
+ metrics[TieringOffloadingMetrics.LOOKUP_ASYNC_DELAY] = (
104
+ OffloadingHistogramMetadata(
105
+ documentation=(
106
+ "Histogram of wall-clock time from a request's first deferred "
107
+ "secondary-tier lookup until the request is allocated or "
108
+ "finishes, in seconds."
109
+ ),
110
+ buckets=(
111
+ 0.0001,
112
+ 0.0005,
113
+ 0.001,
114
+ 0.005,
115
+ 0.01,
116
+ 0.05,
117
+ 0.1,
118
+ 0.5,
119
+ 1,
120
+ 5,
121
+ 10,
122
+ ),
123
+ )
124
+ )
125
secondary_tier_configs = extra_config.get("secondary_tiers", [])
126
if not isinstance(secondary_tier_configs, list):
127
raise ValueError("secondary_tiers must be a list of tier configurations")