@tduncan / Nvidia-Optimization-Stack-VLLM / commits / 426d48bfa1

[KV Offload] Add optional tier locality to FS/OBJ KV events (#48281)

Signed-off-by: Change72 <changg@nvidia.com> Co-authored-by: Codex <codex@openai.com> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Chang Guo committed Jul 17, 2026 at 00:31 UTC 426d48bfa149582664d48f89df21ec9beae5c37b
11 files changed +326 -10
docs/features/kv_offloading_usage.md
+5 -1
@@ -83,9 +83,11 @@ Each entry in `secondary_tiers` is a dict with a required `type` field plus tier
83
84 The filesystem and object-store tiers can publish hash-only `BlockStored` KV events for blocks they successfully store, tagged with a stable per-tier `medium` (`FS` for the filesystem tier, `OBJ` for the object-store tier). Set `enable_kv_events: true` in the tier's entry to opt in; events are published only when KV cache events are also enabled globally via `--kv-events-config`.
85
86 +Set the optional `locality` tier field to `LOCAL` or `REMOTE` to describe the tier's storage location relative to the publishing vLLM instance. `LOCAL` marks storage local to that instance, while `REMOTE` marks storage that is not local to it. When the setting is omitted, locality is unspecified. vLLM does not infer it from the tier type, so an OBJ tier is not implicitly `REMOTE`. A KV event includes `locality` only when the tier explicitly configures it. This metadata describes the tier property without implying that a consumer can already route requests to its blocks.
87 +
88 ### Filesystem (FS)
89
88 -The filesystem tier (`type: "fs"`) writes blocks to a directory on local storage.
90 +The filesystem tier (`type: "fs"`) writes blocks to a filesystem directory.
91
92 | Key | Required | Default | Notes |
93 | --- | --- | --- | --- |
@@ -94,6 +96,7 @@ The filesystem tier (`type: "fs"`) writes blocks to a directory on local storage
96 | `n_read_threads` | no | `16` | Read-priority I/O threads (load path). |
97 | `n_write_threads` | no | `16` | Write-priority I/O threads (store path). |
98 | `enable_kv_events` | no | `false` | Publish `BlockStored` KV events (medium `FS`) for successfully stored blocks. Requires KV cache events to be enabled globally. |
99 +| `locality` | no | unspecified | `LOCAL` or `REMOTE` relative to the publishing vLLM instance. Included in the tier's KV events only when explicitly configured. |
100
101 Each thread group prefers its own queue but pulls from the other when its primary queue is empty, so a write-heavy or read-heavy burst won't leave the off-priority queue waiting. Size the totals to your storage's effective concurrency.
102
@@ -134,6 +137,7 @@ The object-store tier (`type: "obj"`) offloads blocks to an S3-compatible object
137 | `prefix` | no | `""` | Key prefix prepended to all object keys. |
138 | `io_threads` | no | `4` | Number of NIXL OBJ backend I/O threads. |
139 | `enable_kv_events` | no | `false` | Publish `BlockStored` KV events (medium `OBJ`) for successfully stored blocks. Requires KV cache events to be enabled globally. |
140 +| `locality` | no | unspecified | `LOCAL` or `REMOTE` relative to the publishing vLLM instance. Included in the tier's KV events only when explicitly configured; OBJ does not imply `REMOTE`. |
141
142 `store_config` fields:
143
examples/features/kv_events/kv_events_subscriber.py
+4
@@ -42,12 +42,16 @@ class BlockStored(KVCacheEvent):
42 """
43
44 group_idx: int | None = None
45 + kv_cache_spec_kind: str | None = None
46 + kv_cache_spec_sliding_window: int | None = None
47 + locality: str | None = None
48
49
50 class BlockRemoved(KVCacheEvent):
51 block_hashes: list[ExternalBlockHash]
52 medium: str | None
53 group_idx: int | None = None
54 + locality: str | None = None
55
56
57 class AllBlocksCleared(KVCacheEvent):
tests/distributed/test_kv_cache_events.py
+99
@@ -1,6 +1,9 @@
1 # SPDX-License-Identifier: Apache-2.0
2 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
4 +from typing import Any
5 +
6 +import msgspec
7 import pytest
8
9 from vllm.distributed.kv_events import BlockRemoved, BlockStored
@@ -9,9 +12,44 @@ from vllm.distributed.kv_events import BlockRemoved, BlockStored
12 _FAKE_HASH: bytes = b"\xab" * 32
13
14
15 +class _LegacyBlockStored(
16 + msgspec.Struct,
17 + omit_defaults=True, # type: ignore[call-arg]
18 + gc=False, # type: ignore[call-arg]
19 + tag="BlockStored", # type: ignore[call-arg]
20 +):
21 + """BlockStored wire schema before locality was added."""
22 +
23 + block_hashes: list[bytes]
24 + parent_block_hash: bytes | None
25 + token_ids: list[int]
26 + block_size: int
27 + lora_id: int | None
28 + medium: str | None
29 + lora_name: str | None
30 + extra_keys: list[tuple[Any, ...] | None] | None = None
31 + group_idx: int | None = None
32 + kv_cache_spec_kind: str | None = None
33 + kv_cache_spec_sliding_window: int | None = None
34 +
35 +
36 +class _LegacyBlockRemoved(
37 + msgspec.Struct,
38 + omit_defaults=True, # type: ignore[call-arg]
39 + gc=False, # type: ignore[call-arg]
40 + tag="BlockRemoved", # type: ignore[call-arg]
41 +):
42 + """BlockRemoved wire schema before locality was added."""
43 +
44 + block_hashes: list[bytes]
45 + medium: str | None
46 + group_idx: int | None = None
47 +
48 +
49 def _make_block_stored(
50 group_idx: int | None = None,
51 kv_cache_spec_sliding_window: int | None = None,
52 + locality: str | None = None,
53 ) -> BlockStored:
54 return BlockStored(
55 block_hashes=[_FAKE_HASH],
@@ -23,16 +61,19 @@ def _make_block_stored(
61 lora_name=None,
62 group_idx=group_idx,
63 kv_cache_spec_sliding_window=kv_cache_spec_sliding_window,
64 + locality=locality,
65 )
66
67
68 def _make_block_removed(
69 group_idx: int | None = None,
70 + locality: str | None = None,
71 ) -> BlockRemoved:
72 return BlockRemoved(
73 block_hashes=[_FAKE_HASH],
74 medium="GPU",
75 group_idx=group_idx,
76 + locality=locality,
77 )
78
79
@@ -84,3 +125,61 @@ def test_block_stored_hash_differs_by_sliding_window():
125 event_a = _make_block_stored(group_idx=1, kv_cache_spec_sliding_window=128)
126 event_b = _make_block_stored(group_idx=1, kv_cache_spec_sliding_window=256)
127 assert hash(event_a) != hash(event_b)
128 +
129 +
130 +@pytest.mark.parametrize(
131 + ("event_a", "event_b"),
132 + [
133 + (
134 + _make_block_stored(locality="LOCAL"),
135 + _make_block_stored(locality="REMOTE"),
136 + ),
137 + (
138 + _make_block_removed(locality="LOCAL"),
139 + _make_block_removed(locality="REMOTE"),
140 + ),
141 + ],
142 +)
143 +def test_event_hash_differs_by_locality(
144 + event_a: BlockStored | BlockRemoved,
145 + event_b: BlockStored | BlockRemoved,
146 +):
147 + assert hash(event_a) != hash(event_b)
148 +
149 +
150 +def test_block_stored_locality_is_wire_compatible():
151 + legacy = _LegacyBlockStored(
152 + block_hashes=[_FAKE_HASH],
153 + parent_block_hash=None,
154 + token_ids=[1, 2, 3, 4],
155 + block_size=4,
156 + lora_id=None,
157 + medium="GPU",
158 + lora_name=None,
159 + group_idx=2,
160 + kv_cache_spec_sliding_window=128,
161 + )
162 + legacy_payload = msgspec.msgpack.encode(legacy)
163 + assert (
164 + msgspec.msgpack.encode(
165 + _make_block_stored(
166 + group_idx=2,
167 + kv_cache_spec_sliding_window=128,
168 + )
169 + )
170 + == legacy_payload
171 + )
172 + assert msgspec.msgpack.decode(legacy_payload, type=BlockStored).locality is None
173 + new_payload = msgspec.msgpack.encode(_make_block_stored(locality="LOCAL"))
174 + assert msgspec.msgpack.decode(new_payload)["locality"] == "LOCAL"
175 + assert msgspec.msgpack.decode(new_payload, type=_LegacyBlockStored).medium == "GPU"
176 +
177 +
178 +def test_block_removed_locality_is_wire_compatible():
179 + legacy = _LegacyBlockRemoved(block_hashes=[_FAKE_HASH], medium="GPU")
180 + legacy_payload = msgspec.msgpack.encode(legacy)
181 + assert msgspec.msgpack.encode(_make_block_removed()) == legacy_payload
182 + assert msgspec.msgpack.decode(legacy_payload, type=BlockRemoved).locality is None
183 + new_payload = msgspec.msgpack.encode(_make_block_removed(locality="REMOTE"))
184 + assert msgspec.msgpack.decode(new_payload)["locality"] == "REMOTE"
185 + assert msgspec.msgpack.decode(new_payload, type=_LegacyBlockRemoved).medium == "GPU"
tests/v1/kv_connector/unit/offloading_connector/test_events.py
+75 -5
@@ -7,7 +7,7 @@ import torch
7
8 from tests.v1.kv_connector.unit.utils import create_vllm_config
9 from vllm.config import KVEventsConfig, KVTransferConfig
10 -from vllm.distributed.kv_events import MEDIUM_CPU, BlockRemoved, BlockStored
10 +from vllm.distributed.kv_events import MEDIUM_CPU, MEDIUM_FS, BlockRemoved, BlockStored
11 from vllm.distributed.kv_transfer.kv_connector.v1.offloading.config import (
12 build_offloading_config,
13 )
@@ -26,6 +26,7 @@ from vllm.v1.kv_cache_interface import (
26 KVCacheSpecKind,
27 )
28 from vllm.v1.kv_offload.base import (
29 + Locality,
30 OffloadingEvent,
31 OffloadingKVEventsConfig,
32 OffloadKey,
@@ -103,12 +104,81 @@ def _record_chunks(
104 return keys
105
106
106 -def _stored_event(keys: list[OffloadKey]) -> OffloadingEvent:
107 - return OffloadingEvent(keys=keys, medium=_CPU_MEDIUM, removed=False)
107 +def _stored_event(
108 + keys: list[OffloadKey],
109 + locality: Locality | None = None,
110 + medium: str = _CPU_MEDIUM,
111 +) -> OffloadingEvent:
112 + return OffloadingEvent(
113 + keys=keys,
114 + medium=medium,
115 + removed=False,
116 + locality=locality,
117 + )
118 +
119 +
120 +def _removed_event(
121 + keys: list[OffloadKey],
122 + locality: Locality | None = None,
123 + medium: str = _CPU_MEDIUM,
124 +) -> OffloadingEvent:
125 + return OffloadingEvent(
126 + keys=keys,
127 + medium=medium,
128 + removed=True,
129 + locality=locality,
130 + )
131 +
132 +
133 +def test_take_events_forwards_locality_to_rich_store():
134 + tracker = _tracker()
135 + req = _request(block_hashes=[_hash(0)], token_count=4)
136 + key = _record_chunks(tracker, req, _group_config(), num_chunks=1)[0]
137 +
138 + events = list(
139 + tracker.take_events(
140 + [_stored_event([key], locality=Locality.LOCAL, medium=MEDIUM_FS)]
141 + )
142 + )
143
144 + assert len(events) == 1
145 + assert isinstance(events[0], BlockStored)
146 + assert events[0].token_ids == [1, 2, 3, 4]
147 + assert events[0].block_size == 4
148 + assert events[0].locality == "LOCAL"
149
110 -def _removed_event(keys: list[OffloadKey]) -> OffloadingEvent:
111 - return OffloadingEvent(keys=keys, medium=_CPU_MEDIUM, removed=True)
150 +
151 +def test_take_events_forwards_locality_to_placeholder_store():
152 + tracker = _tracker(self_describing_kv_events=False)
153 + req = _request(block_hashes=[_hash(0)], token_count=4)
154 + key = _record_chunks(tracker, req, _group_config(), num_chunks=1)[0]
155 +
156 + events = list(
157 + tracker.take_events(
158 + [_stored_event([key], locality=Locality.REMOTE, medium=MEDIUM_FS)]
159 + )
160 + )
161 +
162 + assert len(events) == 1
163 + assert isinstance(events[0], BlockStored)
164 + assert events[0].block_size == 0
165 + assert events[0].locality == "REMOTE"
166 +
167 +
168 +def test_take_events_forwards_locality_to_remove():
169 + tracker = _tracker()
170 + req = _request(block_hashes=[_hash(0)], token_count=4)
171 + key = _record_chunks(tracker, req, _group_config(), num_chunks=1)[0]
172 +
173 + events = list(
174 + tracker.take_events(
175 + [_removed_event([key], locality=Locality.LOCAL, medium=MEDIUM_FS)]
176 + )
177 + )
178 +
179 + assert len(events) == 1
180 + assert isinstance(events[0], BlockRemoved)
181 + assert events[0].locality == "LOCAL"
182
183
184 def test_take_events_publishes_routable_block_stored():
tests/v1/kv_offload/tiering/test_fs_tier.py
+64
@@ -20,6 +20,7 @@ import torch
20
21 from vllm.distributed.kv_events import MEDIUM_FS
22 from vllm.v1.kv_offload.base import (
23 + Locality,
24 LookupResult,
25 OffloadingEvent,
26 OffloadingKVEventsConfig,
@@ -35,6 +36,7 @@ from vllm.v1.kv_offload.config import (
36 OffloadingParallelConfig,
37 )
38 from vllm.v1.kv_offload.tiering.base import JobMetadata
39 +from vllm.v1.kv_offload.tiering.factory import SecondaryTierFactory
40 from vllm.v1.kv_offload.tiering.fs.manager import (
41 FileSystemTierManager,
42 )
@@ -186,6 +188,7 @@ def fs_tier_with_events(tmp_path):
188 n_read_threads=4,
189 n_write_threads=4,
190 enable_kv_events=True,
191 + locality="LOCAL",
192 )
193 yield tier
194 tier.shutdown()
@@ -251,6 +254,40 @@ def test_invalid_path_raises_at_construction():
254 )
255
256
257 +@pytest.mark.parametrize("locality", ["local", ""])
258 +def test_invalid_locality_raises_at_construction(tmp_path, locality):
259 + tensor = _page_aligned_zero_tensor(4, _BLOCK_ELEMENTS)
260 +
261 + with pytest.raises(ValueError, match="Locality"):
262 + FileSystemTierManager(
263 + offloading_spec=_MOCK_OFFLOADING_SPEC,
264 + primary_kv_view=memoryview(tensor.numpy()),
265 + tier_type="fs",
266 + root_dir=str(tmp_path),
267 + locality=locality,
268 + )
269 +
270 +
271 +def test_factory_forwards_locality_to_fs_tier(tmp_path):
272 + tensor = _page_aligned_zero_tensor(4, _BLOCK_ELEMENTS)
273 + tier = SecondaryTierFactory.create_secondary_tier(
274 + {
275 + "type": "fs",
276 + "root_dir": str(tmp_path),
277 + "n_read_threads": 1,
278 + "n_write_threads": 1,
279 + "locality": "LOCAL",
280 + },
281 + memoryview(tensor.numpy()),
282 + _MOCK_OFFLOADING_SPEC,
283 + )
284 + try:
285 + assert isinstance(tier, FileSystemTierManager)
286 + assert tier.locality is Locality.LOCAL
287 + finally:
288 + tier.shutdown()
289 +
290 +
291 def test_failed_load_missing_file(fs_tier):
292 """Test that loading a block whose file does not exist results in a failed job."""
293 tier, _ = fs_tier
@@ -442,11 +479,38 @@ def test_successful_store_emits_stored_event(fs_tier_with_events):
479 assert events[0].keys == keys
480 # Literal medium pins the wire contract, not just the constant choice.
481 assert events[0].medium == "FS"
482 + assert events[0].locality is Locality.LOCAL
483 assert not events[0].removed
484 # take_events drains the buffer.
485 assert list(tier.take_events()) == []
486
487
488 +@pytest.mark.parametrize(
489 + ("locality", "expected"),
490 + [(None, None), ("REMOTE", Locality.REMOTE)],
491 +)
492 +def test_store_event_uses_configured_locality(tmp_path, locality, expected):
493 + tensor = _page_aligned_zero_tensor(4, _BLOCK_ELEMENTS)
494 + locality_config = {} if locality is None else {"locality": locality}
495 + tier = FileSystemTierManager(
496 + offloading_spec=_make_offloading_spec(enable_kv_cache_events=True),
497 + primary_kv_view=memoryview(tensor.numpy()),
498 + tier_type="fs",
499 + root_dir=str(tmp_path),
500 + enable_kv_events=True,
501 + **locality_config,
502 + )
503 + try:
504 + tier.submit_store(make_job(1, [key(1)], [0]))
505 + assert all(r.success for r in drain(tier))
506 +
507 + events = list(tier.take_events())
508 + assert len(events) == 1
509 + assert events[0].locality is expected
510 + finally:
511 + tier.shutdown()
512 +
513 +
514 def test_load_job_emits_no_event(fs_tier_with_events):
515 tier = fs_tier_with_events
516 tier.submit_store(make_job(1, [key(1)], [0]))
tests/v1/kv_offload/tiering/test_obj_tier.py
+31
@@ -15,9 +15,11 @@ from types import SimpleNamespace
15 from unittest.mock import MagicMock, patch
16
17 import numpy as np
18 +import pytest
19 import torch
20
21 from vllm.v1.kv_offload.base import (
22 + Locality,
23 LookupResult,
24 OffloadingKVEventsConfig,
25 OffloadKey,
@@ -265,6 +267,12 @@ def lookup_and_wait(
267 # ---------------------------------------------------------------------------
268
269
270 +@pytest.mark.parametrize("locality", ["local", ""])
271 +def test_invalid_locality_raises_at_construction(locality):
272 + with pytest.raises(ValueError, match="Locality"):
273 + _make_tier(locality=locality)
274 +
275 +
276 class TestMockObjTierBasic:
277 def setup_method(self):
278 self.tier, self.agent = _make_tier(num_blocks=4)
@@ -453,6 +461,7 @@ class TestObjTierKVEvents:
461 self.tier, self.agent = _make_tier(
462 offloading_spec=_make_events_spec(enable_kv_cache_events=True),
463 enable_kv_events=True,
464 + locality="REMOTE",
465 )
466
467 def test_successful_store_emits_stored_event(self):
@@ -466,10 +475,32 @@ class TestObjTierKVEvents:
475 assert events[0].keys == keys
476 # Literal medium pins the wire contract, not just the constant choice.
477 assert events[0].medium == "OBJ"
478 + assert events[0].locality is Locality.REMOTE
479 assert not events[0].removed
480 # take_events drains the buffer.
481 assert list(self.tier.take_events()) == []
482
483 + @pytest.mark.parametrize(
484 + ("locality", "expected"),
485 + [(None, None), ("LOCAL", Locality.LOCAL)],
486 + )
487 + def test_store_event_uses_configured_locality(self, locality, expected):
488 + locality_config = {} if locality is None else {"locality": locality}
489 + tier, _ = _make_tier(
490 + offloading_spec=_make_events_spec(enable_kv_cache_events=True),
491 + enable_kv_events=True,
492 + **locality_config,
493 + )
494 + try:
495 + tier.submit_store(make_job(1, [key(1)], [0]))
496 + assert all(r.success for r in drain(tier))
497 +
498 + events = list(tier.take_events())
499 + assert len(events) == 1
500 + assert events[0].locality is expected
501 + finally:
502 + tier.shutdown()
503 +
504 def test_mixed_job_results_emit_event_only_for_successful_job(self):
505 """With a failed and a successful store job resolving in the same
506 poll, exactly one event is emitted and its keys belong to the
vllm/distributed/kv_events.py
+6
@@ -74,6 +74,8 @@ class BlockStored(KVCacheEvent):
74 # filter groups as they are learned. Remove events only need group_idx+hash.
75 kv_cache_spec_kind: str | None = None
76 kv_cache_spec_sliding_window: int | None = None
77 + locality: str | None = None
78 + """LOCAL or REMOTE relative to the publisher; None means unspecified."""
79
80 def __hash__(self) -> int:
81 return hash(
@@ -88,6 +90,7 @@ class BlockStored(KVCacheEvent):
90 self.group_idx,
91 self.kv_cache_spec_kind,
92 self.kv_cache_spec_sliding_window,
93 + self.locality,
94 )
95 )
96
@@ -96,6 +99,8 @@ class BlockRemoved(KVCacheEvent):
99 block_hashes: list[ExternalBlockHash]
100 medium: str | None
101 group_idx: int | None = None
102 + locality: str | None = None
103 + """LOCAL or REMOTE relative to the publisher; None means unspecified."""
104
105 def __hash__(self) -> int:
106 return hash(
@@ -103,6 +108,7 @@ class BlockRemoved(KVCacheEvent):
108 tuple(self.block_hashes),
109 self.medium,
110 self.group_idx,
111 + self.locality,
112 )
113 )
114
vllm/distributed/kv_transfer/kv_connector/v1/offloading/events.py
+12 -2
@@ -198,7 +198,12 @@ class OffloadingEventsTracker:
198 kv_cache_spec=group_config.kv_event_group_spec,
199 )
200
201 - def _placeholder_stored(self, key: OffloadKey, medium: str) -> BlockStored:
201 + def _placeholder_stored(
202 + self,
203 + key: OffloadKey,
204 + medium: str,
205 + locality: str | None,
206 + ) -> BlockStored:
207 return BlockStored(
208 block_hashes=[
209 maybe_convert_block_hash(BlockHash(get_offload_block_hash(key)))
@@ -210,12 +215,14 @@ class OffloadingEventsTracker:
215 medium=medium,
216 lora_name=None,
217 group_idx=get_offload_group_idx(key),
218 + locality=locality,
219 )
220
221 def _take_stored_event(self, event: OffloadingEvent) -> Iterable[KVCacheEvent]:
222 # Metadata is read, NOT popped: the entry must survive until the
223 # eviction event so BlockRemoved can fan out to the same hashes.
224 # Events are self-contained (own parent), so key order is free.
225 + locality = event.locality.value if event.locality is not None else None
226 for key in event.keys:
227 meta = self._pending_event_metadata.get(key)
228 if meta is None:
@@ -227,7 +234,7 @@ class OffloadingEventsTracker:
234 "placeholder payload. Expected for non-full-attention "
235 "groups; otherwise indicates a missing populate path."
236 )
230 - yield self._placeholder_stored(key, event.medium)
237 + yield self._placeholder_stored(key, event.medium, locality)
238 continue
239
240 yield BlockStored(
@@ -252,10 +259,12 @@ class OffloadingEventsTracker:
259 kv_cache_spec_sliding_window=(
260 meta.kv_cache_spec.kv_cache_spec_sliding_window
261 ),
262 + locality=locality,
263 )
264
265 def _take_removed_event(self, event: OffloadingEvent) -> Iterable[KVCacheEvent]:
266 # Keep group_idx unambiguous if a manager batch spans groups.
267 + locality = event.locality.value if event.locality is not None else None
268 by_group: dict[int, list] = {}
269 for key in event.keys:
270 meta = self._pending_event_metadata.pop(key, None)
@@ -283,4 +292,5 @@ class OffloadingEventsTracker:
292 block_hashes=hashes,
293 medium=event.medium,
294 group_idx=group_idx,
295 + locality=locality,
296 )
vllm/v1/kv_offload/base.py
+8
@@ -97,12 +97,20 @@ class PrepareStoreOutput:
97 evicted_keys: list[OffloadKey]
98
99
100 +class Locality(Enum):
101 + """Locality of a tier's storage relative to the publishing instance."""
102 +
103 + LOCAL = "LOCAL"
104 + REMOTE = "REMOTE"
105 +
106 +
107 @dataclass
108 class OffloadingEvent:
109 keys: list[OffloadKey]
110 medium: str
111 # True if blocks are removed, False if stored
112 removed: bool
113 + locality: Locality | None = None
114
115
116 """
vllm/v1/kv_offload/tiering/fs/manager.py
+11 -1
@@ -33,6 +33,7 @@ from typing_extensions import override
33 from vllm.distributed.kv_events import MEDIUM_FS
34 from vllm.logger import init_logger
35 from vllm.v1.kv_offload.base import (
36 + Locality,
37 LookupResult,
38 OffloadingEvent,
39 OffloadKey,
@@ -110,6 +111,7 @@ class FileSystemTierManager(SecondaryTierManager):
111 n_read_threads: int = 16,
112 n_write_threads: int = 16,
113 enable_kv_events: bool = False,
114 + locality: str | None = None,
115 ):
116 """
117 Args:
@@ -123,8 +125,11 @@ class FileSystemTierManager(SecondaryTierManager):
125 enable_kv_events: Emit BlockStored KV events for blocks
126 successfully stored to this tier. Effective only when KV
127 cache events are enabled globally (kv_events_config).
128 + locality: Whether this tier's storage is LOCAL or REMOTE relative
129 + to the publishing vLLM instance.
130 """
131 super().__init__(offloading_spec, primary_kv_view, tier_type)
132 + self.locality = Locality(locality) if locality is not None else None
133
134 self.events: list[OffloadingEvent] | None = None
135 if enable_kv_events:
@@ -223,7 +228,12 @@ class FileSystemTierManager(SecondaryTierManager):
228 keys = self._store_job_keys.pop(job_id, None)
229 if success and keys:
230 self.events.append(
226 - OffloadingEvent(keys=keys, medium=self.medium, removed=False)
231 + OffloadingEvent(
232 + keys=keys,
233 + medium=self.medium,
234 + removed=False,
235 + locality=self.locality,
236 + )
237 )
238 results.append(JobResult(job_id=job_id, success=success))
239 return results
vllm/v1/kv_offload/tiering/obj/manager.py
+11 -1
@@ -12,6 +12,7 @@ from vllm.distributed.nixl_utils import NixlWrapper as nixl_agent
12 from vllm.distributed.nixl_utils import nixl_agent_config
13 from vllm.logger import init_logger
14 from vllm.v1.kv_offload.base import (
15 + Locality,
16 LookupResult,
17 OffloadingEvent,
18 OffloadKey,
@@ -108,6 +109,7 @@ class ObjectStoreSecondaryTierManager(SecondaryTierManager):
109 prefix: str = "",
110 io_threads: int = 4,
111 enable_kv_events: bool = False,
112 + locality: str | None = None,
113 ):
114 """
115 Args:
@@ -120,8 +122,11 @@ class ObjectStoreSecondaryTierManager(SecondaryTierManager):
122 enable_kv_events: Emit BlockStored KV events for blocks
123 successfully stored to this tier. Effective only when KV
124 cache events are enabled globally (kv_events_config).
125 + locality: Whether this tier's storage is LOCAL or REMOTE relative
126 + to the publishing vLLM instance.
127 """
128 super().__init__(offloading_spec, primary_kv_view, tier_type)
129 + self.locality = Locality(locality) if locality is not None else None
130
131 self.events: list[OffloadingEvent] | None = None
132 if enable_kv_events:
@@ -323,7 +328,12 @@ class ObjectStoreSecondaryTierManager(SecondaryTierManager):
328 keys = self._store_job_keys.pop(result.job_id, None)
329 if result.success and keys:
330 self.events.append(
326 - OffloadingEvent(keys=keys, medium=self.medium, removed=False)
331 + OffloadingEvent(
332 + keys=keys,
333 + medium=self.medium,
334 + removed=False,
335 + locality=self.locality,
336 + )
337 )
338 return results
339