Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 66 additions & 0 deletions tests/v1/kv_offload/cpu/test_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -529,6 +529,72 @@ def test_prepare_load_preserves_key_order():
manager.complete_load([key_a, key_b, key_c], _EMPTY_REQ_CTX) # order irrelevant


def test_prepare_load_failure_reports_eviction():
"""A key evicted after lookup() is reported with its eviction age."""
manager = make_cpu_manager(num_chunks=1)
key_a, key_b = to_key(1), to_key(2)
manager.prepare_store([key_a], _EMPTY_REQ_CTX)
manager.complete_store([key_a], _EMPTY_REQ_CTX)
assert manager.lookup(key_a, _EMPTY_REQ_CTX) is LookupResult.HIT
manager.prepare_store([key_b], _EMPTY_REQ_CTX)

req_context = make_req_context(req_id="req-1")
req_context.set_offload_key_position(key_a, 16)
with pytest.raises(AssertionError) as exc_info:
manager.prepare_load([key_a], req_context)
message = str(exc_info.value)
assert "req_id='req-1'" in message
assert "per missing key: [(16, " in message
assert "per missing key: [(16, None)]" not in message


def test_prepare_load_failure_reports_never_stored_key():
"""A key that was never stored shows up at the edge of the load range,
with no eviction behind it."""
manager = make_cpu_manager(num_chunks=4)
keys = to_keys([1, 2, 3])
manager.prepare_store(keys[1:], _EMPTY_REQ_CTX)
manager.complete_store(keys[1:], _EMPTY_REQ_CTX)

req_context = make_req_context()
for i, key in enumerate(keys):
req_context.set_offload_key_position(key, (i + 1) * 16)
with pytest.raises(AssertionError) as exc_info:
manager.prepare_load(keys, req_context)
message = str(exc_info.value)
assert "1 of 3 keys in this load are missing" in message
assert "per missing key: [(16, None)]; load end_token range: 16-48" in message


def test_prepare_load_failure_eviction_history_is_bounded(monkeypatch):
monkeypatch.setattr(CPUOffloadingManager, "EVICTION_HISTORY_SIZE", 1)
manager = make_cpu_manager(num_chunks=1)
for key in to_keys([1, 2, 3]):
manager.prepare_store([key], _EMPTY_REQ_CTX)
manager.complete_store([key], _EMPTY_REQ_CTX)

# Key 1 was evicted first and has left the one-entry history.
with pytest.raises(AssertionError) as exc_info:
manager.prepare_load([to_key(1)], _EMPTY_REQ_CTX)
message = str(exc_info.value)
assert "per missing key: [(None, None)]" in message
assert "eviction history: 1 keys" in message


def test_prepare_load_failure_after_reset_has_no_eviction_history():
manager = make_cpu_manager(num_chunks=1)
for key in to_keys([1, 2]):
manager.prepare_store([key], _EMPTY_REQ_CTX)
manager.complete_store([key], _EMPTY_REQ_CTX)
manager.reset_cache()

with pytest.raises(AssertionError) as exc_info:
manager.prepare_load([to_key(1)], _EMPTY_REQ_CTX)
message = str(exc_info.value)
assert "per missing key: [(None, None)]" in message
assert "eviction history: 0 keys" in message


def test_lru_batch_eviction_failure_is_atomic():
manager = make_cpu_manager(num_chunks=4, cache_policy="lru")
policy = manager._policy
Expand Down
48 changes: 46 additions & 2 deletions vllm/v1/kv_offload/cpu/manager.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
from collections import OrderedDict
import time
from collections import OrderedDict, deque
from collections.abc import Collection, Iterable, Sequence
from dataclasses import dataclass, field

Expand Down Expand Up @@ -57,6 +58,9 @@ class CPUOffloadingManager(OffloadingManager):
to the CachePolicy implementation.
"""

# Recent evictions, reported if prepare_load finds a key missing.
EVICTION_HISTORY_SIZE = 10_000

def __init__(
self,
num_chunks: int,
Expand Down Expand Up @@ -88,6 +92,9 @@ def __init__(
self.stores_skipped_in_current_batch: int = 0
self.allocation_sizes_in_current_batch: list[int] = []
self._cache_generation = 0
self._recent_evictions: deque[tuple[OffloadKey, float]] = deque(
maxlen=self.EVICTION_HISTORY_SIZE
)

# Number of chunk references. It is ordered so can evict the LRU entry in O(1).
self.counts: OrderedDict[OffloadKey, int] | None = (
Expand Down Expand Up @@ -223,7 +230,9 @@ def _prepare_load(
chunks = []
for key in keys:
chunk = self._policy.get(key)
assert chunk is not None, f"Chunk {key!r} not found in cache"
assert chunk is not None, self._missing_chunk_message(

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Some findings from the investigation

After analyzing the lifecycle, it appears that lookup() is intentionally observational: it doesn't pin a chunk because speculative lookups may be discarded. Protection only begins in prepare_load().
However, the current prepare_load() function treats a missing chunk as an error and asserts. If a subsequent promotion can call primary_tier.prepare_write() and evict a lookup hit before prepare_load(), then the system needs to define one of two contracts:

  1. Lookup hits remain valid until the prepare_load() function is called, and must be protected accordingly.
  2. Lookup hits may become stale, so the prepare_load() function needs a mechanism to handle stale/miss situations instead of simply asserting that they shouldn't occur.

key, keys, req_context
)
assert chunk.is_ready, f"Chunk {key!r} is not ready for reading"
if chunk.ref_cnt == 0:
self._policy.mark_non_evictable(key)
Expand All @@ -235,6 +244,38 @@ def _prepare_load(
self._record_request_cache_access(keys, req_context, reused_keys=keys)
return self._get_load_store_spec(keys, chunks)

def _missing_chunk_message(
self,
key: OffloadKey,
keys: Collection[OffloadKey],
req_context: ReqContext,
) -> str:
now = time.monotonic()
evicted_at = dict(self._recent_evictions)
position = req_context.get_offload_key_position
missing = [k for k in keys if self._policy.get(k) is None]
missing_info = [
(position(k), round(now - evicted_at[k], 3) if k in evicted_at else None)
for k in missing[:16]
]
load_positions = [p for k in keys if (p := position(k)) is not None]
load_range = (
f"{min(load_positions)}-{max(load_positions)}" if load_positions else None
)
history_span = (
round(now - self._recent_evictions[0][1], 3)
if self._recent_evictions
else 0.0
)
return (
f"Chunk {key!r} not found in cache (group={get_offload_group_idx(key)}, "
f"req_id={req_context.req_id!r}). {len(missing)} of {len(keys)} keys "
f"in this load are missing. (end_token, seconds since eviction) per "
f"missing key: {missing_info}; load end_token range: {load_range}; "
f"eviction history: {len(self._recent_evictions)} keys over "
f"{history_span}s."
)

@override
def touch(self, keys: Collection[OffloadKey], req_context: ReqContext) -> None:
self._policy.touch(keys, req_context)
Expand Down Expand Up @@ -334,9 +375,11 @@ def prepare_store(
self._num_evictable_cache_chunks -= len(evicted)
assert self._num_evictable_cache_chunks >= 0

now = time.monotonic()
for key, chunk in evicted:
self._free_chunk(chunk)
to_evict.append(key)
self._recent_evictions.append((key, now))

if to_evict and self.events is not None:
self.events.append(
Expand Down Expand Up @@ -444,6 +487,7 @@ def reset_cache(self) -> None:

self._free_list.clear()
self._num_allocated_chunks = 0
self._recent_evictions.clear()

@override
def take_events(self) -> Iterable[OffloadingEvent]:
Expand Down
Loading