@tduncan / Nvidia-Optimization-Stack-VLLM / commits / 23e26e0588

[KV Offloading] Back-pressure detection and remediation (#50045)

Signed-off-by: Bill Nell <bnell@redhat.com> Signed-off-by: bnellnm <49004751+bnellnm@users.noreply.github.com> Co-authored-by: Itay Etelis <92247226+Etelis@users.noreply.github.com>

bnellnm committed Sep 18, 2026 at 14:47 UTC 23e26e058839fb2a3e77c78fbf2184f563593227
13 files changed +1464 -5
tests/v1/kv_connector/unit/offloading_connector/test_events.py
+89
@@ -682,3 +682,92 @@ def test_tiering_accepts_self_describing_kv_events():
682 assert spec.kv_events_config.enable_kv_cache_events
683 assert spec.kv_events_config.self_describing_kv_events
684 assert tracker.self_describing_enabled
685 +
686 +
687 +def _build_tiering_spec(secondary_tiers, *, top_level_backpressure=None):
688 + vllm_config = create_vllm_config(
689 + block_size=4,
690 + max_num_batched_tokens=16,
691 + disable_hybrid_kv_cache_manager=False,
692 + )
693 + extra_config = {
694 + "spec_name": "TieringOffloadingSpec",
695 + "cpu_bytes_to_use": 1 << 20,
696 + "secondary_tiers": secondary_tiers,
697 + }
698 + if top_level_backpressure is not None:
699 + extra_config["backpressure"] = top_level_backpressure
700 + vllm_config.kv_transfer_config = KVTransferConfig(
701 + kv_connector="OffloadingConnector",
702 + kv_role="kv_both",
703 + kv_connector_extra_config=extra_config,
704 + )
705 + kv_cache_config = KVCacheConfig(
706 + num_blocks=0,
707 + kv_cache_tensors=[],
708 + kv_cache_groups=[
709 + KVCacheGroupSpec(
710 + ["layer"],
711 + FullAttentionSpec(
712 + block_size=4,
713 + num_kv_heads=1,
714 + head_size=1,
715 + dtype=torch.float32,
716 + ),
717 + )
718 + ],
719 + )
720 + return TieringOffloadingSpec(build_offloading_config(vllm_config, kv_cache_config))
721 +
722 +
723 +def test_partial_tier_backpressure_inherits_top_level_defaults():
724 + """A partial tier override still inherits missing fields.
725 +
726 + Field-by-field merge means an override that sets only ``high_water_s``
727 + keeps ``backpressure_cls`` (and any other field) from the top-level
728 + default, so the resolved dict reaching the factory is complete.
729 + """
730 + spec = _build_tiering_spec(
731 + [{"type": "example", "backpressure": {"high_water_s": 0.1}}],
732 + top_level_backpressure={
733 + "backpressure_cls": "EMABackpressureDetector",
734 + "high_water_s": 0.5,
735 + "low_water_s": 0.05,
736 + },
737 + )
738 +
739 + resolved = spec.secondary_tier_configs[0]["backpressure"]
740 + # Tier override wins for the field it sets.
741 + assert resolved["high_water_s"] == 0.1
742 + # Missing fields fall back to the top-level default.
743 + assert resolved["backpressure_cls"] == "EMABackpressureDetector"
744 + assert resolved["low_water_s"] == 0.05
745 +
746 +
747 +def test_tier_override_wins_over_top_level():
748 + """Precedence is tier override > top-level default, field-by-field."""
749 + spec = _build_tiering_spec(
750 + [
751 + {
752 + "type": "example",
753 + "backpressure": {"high_water_s": 0.1, "low_water_s": 0.02},
754 + }
755 + ],
756 + top_level_backpressure={
757 + "backpressure_cls": "EMABackpressureDetector",
758 + "high_water_s": 0.5,
759 + "low_water_s": 0.05,
760 + },
761 + )
762 +
763 + resolved = spec.secondary_tier_configs[0]["backpressure"]
764 + # Tier override wins for the fields it sets.
765 + assert resolved["high_water_s"] == 0.1
766 + assert resolved["low_water_s"] == 0.02
767 + # Fields only in top-level still fill in.
768 + assert resolved["backpressure_cls"] == "EMABackpressureDetector"
769 +
770 +
771 +def test_tier_without_backpressure_stays_unconfigured():
772 + spec = _build_tiering_spec([{"type": "example"}])
773 + assert "backpressure" not in spec.secondary_tier_configs[0]
tests/v1/kv_offload/tiering/p2p/test_manager.py
+34
@@ -180,6 +180,40 @@ class TestInitHashSeed:
180 assert mgr._get_hash_seed() == "random-seed-abc"
181
182
183 +class TestBackpressureRejected:
184 + def _patch_transports(self, monkeypatch) -> None:
185 + monkeypatch.setattr(manager_module, "NixlTransport", lambda *a, **k: object())
186 + monkeypatch.setattr(manager_module, "ZmqTransport", lambda *a, **k: object())
187 + monkeypatch.setattr(
188 + manager_module.FileMapper,
189 + "from_offloading_spec",
190 + lambda **k: SimpleNamespace(get_run_config=lambda: {}),
191 + )
192 +
193 + def test_explicit_backpressure_detector_rejected(self, monkeypatch):
194 + """An explicitly configured detector must fail fast for P2P.
195 +
196 + The generic store-latency detector cannot fail fast in PD mode and
197 + rendezvous time makes store latency an unreliable pressure signal,
198 + so P2P rejects it rather than degrading silently.
199 + """
200 + self._patch_transports(monkeypatch)
201 + with pytest.raises(ValueError, match="not supported for the P2P"):
202 + P2PSecondaryTierManager(
203 + offloading_spec=_init_offloading_spec(),
204 + primary_kv_view=memoryview(bytearray(16)),
205 + backpressure_detector=object(),
206 + )
207 +
208 + def test_no_detector_constructs_normally(self, monkeypatch):
209 + self._patch_transports(monkeypatch)
210 + mgr = P2PSecondaryTierManager(
211 + offloading_spec=_init_offloading_spec(),
212 + primary_kv_view=memoryview(bytearray(16)),
213 + )
214 + assert mgr.bp_detector is None
215 +
216 +
217 # ---------------------------------------------------------------------------
218 # Tests for _peer_id_from_params
219 # ---------------------------------------------------------------------------
tests/v1/kv_offload/tiering/test_backpressure.py new
+706
@@ -0,0 +1,706 @@
1 +# SPDX-License-Identifier: Apache-2.0
2 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 +"""Unit tests for back-pressure detection in TieringOffloadingManager.
4 +
5 +These tests use a DelayedSecondaryTierManager that holds completed jobs
6 +until explicitly released, allowing precise control over when the manager
7 +observes store completions and their apparent latency.
8 +"""
9 +
10 +import time
11 +from collections.abc import Iterable
12 +from unittest.mock import MagicMock, patch
13 +
14 +import pytest
15 +import torch
16 +
17 +from vllm.v1.kv_offload.base import (
18 + LookupResult,
19 + OffloadKey,
20 + OffloadPolicy,
21 + ReqContext,
22 + RequestOffloadingContext,
23 + ScheduleEndContext,
24 + make_offload_key,
25 +)
26 +from vllm.v1.kv_offload.tiering.backpressure import (
27 + DropAccountingPolicy,
28 + DropStorePolicy,
29 + EMABackpressureDetector,
30 + ThrottledDropPolicy,
31 +)
32 +from vllm.v1.kv_offload.tiering.base import (
33 + JobResult,
34 + SecondaryTierManager,
35 + TieringOffloadingMetrics,
36 + TransferJob,
37 +)
38 +from vllm.v1.kv_offload.tiering.manager import (
39 + CPUPrimaryTierOffloadingManager,
40 + TieringOffloadingManager,
41 +)
42 +
43 +_BP_EMA_ALPHA = EMABackpressureDetector.DEFAULT_ALPHA
44 +_BP_WARMUP = EMABackpressureDetector.DEFAULT_WARMUP_COMPLETIONS
45 +# Water marks in s/MiB, scaled for the test's tiny 16-byte blocks.
46 +_BP_HIGH_WATER_S = 1000.0
47 +_BP_LOW_WATER_S = 500.0
48 +
49 +_CTX = ReqContext(req_id="test")
50 +_MOCK_OFFLOADING_SPEC = MagicMock()
51 +
52 +
53 +def _mock_mmap_region(num_blocks: int, row_bytes: int = 16):
54 + mock = MagicMock()
55 + view = memoryview(torch.zeros((num_blocks, row_bytes), dtype=torch.int8).numpy())
56 + mock.create_kv_memoryview.return_value = view
57 + return mock
58 +
59 +
60 +def to_keys(int_ids: Iterable[int]) -> list[OffloadKey]:
61 + return [make_offload_key(str(i).encode(), 0) for i in int_ids]
62 +
63 +
64 +class DelayedSecondaryTierManager(SecondaryTierManager):
65 + """Secondary tier that holds completed store jobs until released.
66 +
67 + Jobs are stored immediately but their completion is not reported
68 + by get_finished_jobs() until release_jobs() is called. This lets
69 + tests control the apparent latency of store operations by adjusting
70 + submit_time on the JobMetadata before triggering the completion poll.
71 + """
72 +
73 + def __init__(
74 + self, offloading_spec, primary_kv_view, tier_type, backpressure_detector=None
75 + ):
76 + super().__init__(
77 + offloading_spec,
78 + primary_kv_view,
79 + tier_type,
80 + backpressure_detector=backpressure_detector,
81 + )
82 + self.blocks: dict[OffloadKey, bool] = {}
83 + self._held_jobs: list[JobResult] = []
84 + self._released_jobs: list[JobResult] = []
85 + self._request_policy: OffloadPolicy | None = None
86 +
87 + def lookup(self, key, req_context):
88 + return LookupResult.HIT if key in self.blocks else LookupResult.MISS
89 +
90 + def submit_store(self, job_metadata: TransferJob) -> None:
91 + for key in job_metadata.keys:
92 + self.blocks[key] = True
93 + self._held_jobs.append(JobResult(job_id=job_metadata.job_id, success=True))
94 +
95 + def submit_load(self, job_metadata: TransferJob) -> None:
96 + self._released_jobs.append(JobResult(job_id=job_metadata.job_id, success=True))
97 +
98 + def get_finished_jobs(self) -> Iterable[JobResult]:
99 + result = self._released_jobs
100 + self._released_jobs = []
101 + return result
102 +
103 + def release_jobs(self):
104 + self._released_jobs.extend(self._held_jobs)
105 + self._held_jobs.clear()
106 +
107 + def on_new_request(self, req_context):
108 + if self._request_policy is not None:
109 + return RequestOffloadingContext(policy=self._request_policy)
110 + return RequestOffloadingContext()
111 +
112 + def drain_jobs(self):
113 + self.release_jobs()
114 +
115 + def has_pending_work(self):
116 + return bool(self._held_jobs)
117 +
118 + def get_num_blocks(self):
119 + return len(self.blocks)
120 +
121 +
122 +class TestDefaultConfig:
123 + def test_fs_gets_local_watermarks(self):
124 + cfg = EMABackpressureDetector.default_config("fs")
125 + assert cfg["high_water_s"] == EMABackpressureDetector.LOCAL_HIGH_WATER_S
126 + assert cfg["low_water_s"] == EMABackpressureDetector.LOCAL_LOW_WATER_S
127 +
128 + def test_obj_gets_network_watermarks(self):
129 + cfg = EMABackpressureDetector.default_config("obj")
130 + assert cfg["high_water_s"] == EMABackpressureDetector.NETWORK_HIGH_WATER_S
131 + assert cfg["low_water_s"] == EMABackpressureDetector.NETWORK_LOW_WATER_S
132 +
133 + def test_p2p_gets_local_watermarks(self):
134 + cfg = EMABackpressureDetector.default_config("p2p")
135 + assert cfg["high_water_s"] == EMABackpressureDetector.LOCAL_HIGH_WATER_S
136 + assert cfg["low_water_s"] == EMABackpressureDetector.LOCAL_LOW_WATER_S
137 +
138 + def test_fs_with_remote_locality_gets_network_watermarks(self):
139 + cfg = EMABackpressureDetector.default_config("fs", locality="REMOTE")
140 + assert cfg["high_water_s"] == EMABackpressureDetector.NETWORK_HIGH_WATER_S
141 + assert cfg["low_water_s"] == EMABackpressureDetector.NETWORK_LOW_WATER_S
142 +
143 + def test_fs_with_local_locality_gets_local_watermarks(self):
144 + cfg = EMABackpressureDetector.default_config("fs", locality="LOCAL")
145 + assert cfg["high_water_s"] == EMABackpressureDetector.LOCAL_HIGH_WATER_S
146 + assert cfg["low_water_s"] == EMABackpressureDetector.LOCAL_LOW_WATER_S
147 +
148 +
149 +class TestIdleDecay:
150 + """Tests for EMA idle decay — recovery from pressure without completions."""
151 +
152 + def _make_detector(self, **kwargs):
153 + defaults = dict(
154 + high_water_s=_BP_HIGH_WATER_S,
155 + low_water_s=_BP_LOW_WATER_S,
156 + decay_half_life_s=2.0,
157 + )
158 + defaults.update(kwargs)
159 + return EMABackpressureDetector(**defaults)
160 +
161 + def test_ema_decays_while_under_pressure(self):
162 + bp = self._make_detector()
163 + bp._completions = _BP_WARMUP
164 + bp._ema = _BP_HIGH_WATER_S * 2
165 + bp._under_pressure = True
166 + bp._last_update = time.monotonic()
167 +
168 + with patch("time.monotonic", return_value=bp._last_update + 4.1):
169 + # ~2.05 half-lives → EMA * 0.24, well below low water (500)
170 + assert bp.is_under_pressure() is False
171 + assert bp.store_latency_ema < _BP_LOW_WATER_S
172 +
173 + def test_ema_stays_pressured_if_not_enough_decay(self):
174 + bp = self._make_detector()
175 + bp._completions = _BP_WARMUP
176 + bp._ema = _BP_HIGH_WATER_S * 4
177 + bp._under_pressure = True
178 + bp._last_update = time.monotonic()
179 +
180 + with patch("time.monotonic", return_value=bp._last_update + 2.0):
181 + # 2s = 1 half-life → EMA * 0.5 = HIGH * 2, still above LOW
182 + assert bp.is_under_pressure() is True
183 + assert bp.store_latency_ema == pytest.approx(_BP_HIGH_WATER_S * 2, rel=0.01)
184 +
185 + def test_no_decay_before_warmup(self):
186 + bp = self._make_detector()
187 + bp._ema = _BP_HIGH_WATER_S * 2
188 + bp._under_pressure = True
189 + bp._last_update = time.monotonic()
190 +
191 + with patch("time.monotonic", return_value=bp._last_update + 10.0):
192 + # Still in warmup (_completions=0 < _warmup_completions),
193 + # so decay should not be applied in is_under_pressure.
194 + assert bp.is_under_pressure() is True
195 + assert bp.store_latency_ema == _BP_HIGH_WATER_S * 2
196 +
197 + def test_no_decay_applied_during_new_sample(self):
198 + """on_store_completed applies the new sample directly without
199 + idle decay — decay only runs in is_under_pressure()."""
200 + bp = self._make_detector()
201 + bp._completions = _BP_WARMUP
202 + bp._ema = _BP_HIGH_WATER_S * 2
203 + bp._under_pressure = True
204 + t0 = time.monotonic()
205 + bp._last_update = t0
206 +
207 + block_bytes = 16
208 + with patch("time.monotonic", return_value=t0 + 4.0):
209 + bp.on_store_completed(0.0, block_bytes)
210 + # No decay in on_store_completed: EMA = alpha*0 + (1-alpha)*2000
211 + expected = (1 - _BP_EMA_ALPHA) * _BP_HIGH_WATER_S * 2
212 + assert bp.store_latency_ema == pytest.approx(expected, rel=0.01)
213 +
214 + def test_reset_clears_last_update(self):
215 + bp = self._make_detector()
216 + bp._last_update = time.monotonic()
217 + bp.reset()
218 + assert bp._last_update == 0.0
219 +
220 +
221 +class TestBackpressure:
222 + @pytest.fixture
223 + def setup(self):
224 + mock_region = _mock_mmap_region(20)
225 + self.primary = CPUPrimaryTierOffloadingManager(
226 + num_chunks=20, mmap_region=mock_region
227 + )
228 + mock_view = mock_region.create_kv_memoryview()
229 + self.tier = DelayedSecondaryTierManager(
230 + offloading_spec=_MOCK_OFFLOADING_SPEC,
231 + primary_kv_view=mock_view,
232 + tier_type="delayed",
233 + backpressure_detector=EMABackpressureDetector(
234 + high_water_s=_BP_HIGH_WATER_S,
235 + low_water_s=_BP_LOW_WATER_S,
236 + ),
237 + )
238 + self.manager = TieringOffloadingManager(
239 + primary_tier=self.primary,
240 + secondary_tiers=[self.tier],
241 + )
242 +
243 + def _start_request(self, ctx=_CTX):
244 + if ctx.req_id not in self.manager._req_state:
245 + self.manager.on_new_request(ctx)
246 +
247 + def _store_blocks(self, keys, ctx=_CTX):
248 + self._start_request(ctx)
249 + result = self.manager.prepare_store(keys, ctx)
250 + assert result is not None
251 + self.manager.complete_store(keys, ctx, success=True)
252 +
253 + def _simulate_on_schedule_end(self):
254 + ctx = ScheduleEndContext(new_req_ids=[], preempted_req_ids=())
255 + self.manager.on_schedule_end(ctx)
256 +
257 + def _backdate_held_jobs(self, age_s: float):
258 + """Set submit_time on all in-flight store jobs so they appear
259 + to have taken ``age_s`` seconds when completed."""
260 + self.manager._processed_jobs_this_step = False
261 + now = time.monotonic()
262 + for job_id, meta in self.manager._jobs.items():
263 + tj = meta.transfer_job
264 + if not tj.is_promotion and tj.submit_time > 0:
265 + tj.submit_time = now - age_s
266 +
267 + def test_ema_updates_on_store_completion(self, setup):
268 + bp = self.tier.bp_detector
269 + fast_latency = 0.01
270 + # EMA is in s/MiB; with 16-byte blocks, scale = MiB / block_bytes.
271 + scale = EMABackpressureDetector._MIB / self.tier.block_size_bytes
272 +
273 + # During warmup, EMA stays at 0.
274 + for i in range(_BP_WARMUP - 1):
275 + keys = to_keys([300 + i])
276 + self._store_blocks(keys)
277 + self._backdate_held_jobs(fast_latency)
278 + self.tier.release_jobs()
279 + self._simulate_on_schedule_end()
280 + assert bp.store_latency_ema == 0.0
281 +
282 + # Final warmup sample seeds EMA with the mean.
283 + keys = to_keys([400])
284 + self._store_blocks(keys)
285 + self._backdate_held_jobs(fast_latency)
286 + self.tier.release_jobs()
287 + self._simulate_on_schedule_end()
288 + fast_s_per_mib = fast_latency * scale
289 + assert bp.store_latency_ema == pytest.approx(fast_s_per_mib, rel=0.1)
290 +
291 + # After warmup, EMA updates normally.
292 + new_latency = 0.05
293 + keys = to_keys([401])
294 + self._store_blocks(keys)
295 + self._backdate_held_jobs(new_latency)
296 + self.tier.release_jobs()
297 + self._simulate_on_schedule_end()
298 + new_s_per_mib = new_latency * scale
299 + expected = _BP_EMA_ALPHA * new_s_per_mib + (1 - _BP_EMA_ALPHA) * fast_s_per_mib
300 + assert bp.store_latency_ema == pytest.approx(expected, rel=0.1)
301 +
302 + def test_pressure_activates_above_high_water(self, setup):
303 + bp = self.tier.bp_detector
304 +
305 + # Warm up the detector with fast completions first.
306 + for i in range(_BP_WARMUP):
307 + keys = to_keys([200 + i])
308 + self._store_blocks(keys)
309 + self._backdate_held_jobs(0.001)
310 + self.tier.release_jobs()
311 + self._simulate_on_schedule_end()
312 +
313 + keys = to_keys([100])
314 + self._store_blocks(keys)
315 + self._simulate_on_schedule_end()
316 +
317 + # 0.1s for a single 16-byte block → ~6553 s/MiB, well above high water.
318 + self._backdate_held_jobs(0.1)
319 + self.tier.release_jobs()
320 + self._simulate_on_schedule_end()
321 +
322 + assert bp.store_latency_ema > _BP_HIGH_WATER_S
323 + assert bp.is_under_pressure() is True
324 +
325 + def test_pressure_clears_below_low_water(self, setup):
326 + bp = self.tier.bp_detector
327 + bp.store_latency_ema = _BP_HIGH_WATER_S * 2
328 + bp._under_pressure = True
329 + bp._completions = _BP_WARMUP
330 +
331 + # Drive several fast completions to bring EMA below low water.
332 + block_id = 100
333 + for _ in range(50):
334 + if bp.store_latency_ema < _BP_LOW_WATER_S:
335 + break
336 + keys = to_keys([block_id])
337 + block_id += 1
338 + self._store_blocks(keys)
339 + # Pressure is on, so the cascade was skipped. Force-submit
340 + # a store job manually to get a completion to feed the EMA.
341 + job_meta = self.manager.create_store_job(keys, _CTX)
342 + job_meta.submit_time = time.monotonic() - 0.001
343 + self.tier.submit_store(job_meta)
344 + self.tier.release_jobs()
345 + self._simulate_on_schedule_end()
346 + else:
347 + pytest.fail(f"EMA did not recover below low water: {bp.store_latency_ema}")
348 +
349 + assert bp.is_under_pressure() is False
350 +
351 + def test_hysteresis_prevents_oscillation(self, setup):
352 + bp = self.tier.bp_detector
353 + bp._completions = _BP_WARMUP
354 + block_bytes = self.tier.block_size_bytes
355 +
356 + # Set EMA between low and high water marks (in s/MiB).
357 + mid = (_BP_LOW_WATER_S + _BP_HIGH_WATER_S) / 2
358 + bp.store_latency_ema = mid
359 + bp._last_update = time.monotonic()
360 +
361 + # Raw seconds that produce `mid` s/MiB for a given number of blocks.
362 + def mid_latency_s(num_blocks: int) -> float:
363 + return mid * (num_blocks * block_bytes / EMABackpressureDetector._MIB)
364 +
365 + # When not under pressure, mid-range EMA should not activate.
366 + bp._under_pressure = False
367 + keys = to_keys(range(2))
368 + self._store_blocks(keys)
369 + self._backdate_held_jobs(mid_latency_s(len(keys)))
370 + self.tier.release_jobs()
371 + self._simulate_on_schedule_end()
372 + assert bp.is_under_pressure() is False
373 +
374 + # When under pressure, mid-range EMA should not deactivate.
375 + bp._under_pressure = True
376 + bp._last_update = time.monotonic()
377 + keys2 = to_keys(range(10, 12))
378 + # Force-submit since pressure is on (cascade would skip).
379 + for k in keys2:
380 + self.primary.prepare_store([k], _CTX)
381 + self.primary.complete_store([k], _CTX)
382 + job_meta = self.manager.create_store_job(keys2, _CTX)
383 + job_meta.submit_time = time.monotonic() - mid_latency_s(len(keys2))
384 + self.tier.submit_store(job_meta)
385 + self.tier.release_jobs()
386 + self._simulate_on_schedule_end()
387 + assert bp.is_under_pressure() is True
388 +
389 + def test_stores_skipped_under_pressure(self, setup):
390 + bp = self.tier.bp_detector
391 + bp._under_pressure = True
392 + bp.store_latency_ema = _BP_HIGH_WATER_S * 4
393 + initial_blocks = self.tier.get_num_blocks()
394 +
395 + keys = to_keys(range(50, 53))
396 + self._store_blocks(keys)
397 +
398 + # Blocks should be in primary but NOT in secondary.
399 + for k in keys:
400 + assert self.primary.lookup(k, _CTX) is LookupResult.HIT
401 + assert self.tier.get_num_blocks() == initial_blocks
402 +
403 + def test_stores_resume_after_pressure_clears(self, setup):
404 + bp = self.tier.bp_detector
405 +
406 + # Start under pressure — stores skipped.
407 + bp._under_pressure = True
408 + bp.store_latency_ema = _BP_HIGH_WATER_S * 4
409 + keys1 = to_keys(range(60, 62))
410 + self._store_blocks(keys1)
411 + assert all(k not in self.tier.blocks for k in keys1)
412 +
413 + # Clear pressure — next store should cascade.
414 + bp._under_pressure = False
415 + keys2 = to_keys(range(70, 72))
416 + self._store_blocks(keys2)
417 + assert all(k in self.tier.blocks for k in keys2)
418 +
419 + def test_dropped_store_count_tracked(self, setup):
420 + bp = self.tier.bp_detector
421 + bp._under_pressure = True
422 + bp.store_latency_ema = _BP_HIGH_WATER_S * 4
423 + policy = bp.policy
424 + assert policy.pop_stores_dropped() == (0, 0)
425 +
426 + self._store_blocks(to_keys([80]))
427 + assert policy._stores_dropped == 1
428 + assert policy._blocks_dropped == 1
429 +
430 + self._store_blocks(to_keys([81, 82]))
431 + assert policy._stores_dropped == 2
432 + assert policy._blocks_dropped == 3
433 +
434 + def test_metrics_reported_via_get_stats(self, setup):
435 + bp = self.tier.bp_detector
436 + bp._under_pressure = True
437 + bp.store_latency_ema = 2.5
438 + policy = bp.policy
439 + policy._stores_dropped = 5
440 + policy._blocks_dropped = 12
441 +
442 + stats = self.manager.get_stats()
443 + assert stats is not None
444 + reduced = stats.reduce()
445 +
446 + ema_key = TieringOffloadingMetrics.BACKPRESSURE_STORE_LATENCY_EMA
447 + stores_key = TieringOffloadingMetrics.BACKPRESSURE_STORES_DROPPED
448 + blocks_key = TieringOffloadingMetrics.BACKPRESSURE_BLOCKS_DROPPED
449 + assert reduced[f"{ema_key}:('1:delayed',)"] == pytest.approx(2.5)
450 + assert reduced[f"{stores_key}:('1:delayed',)"] == 5
451 + assert reduced[f"{blocks_key}:('1:delayed',)"] == 12
452 +
453 + # Dropped counts reset after get_stats.
454 + assert policy._stores_dropped == 0
455 + assert policy._blocks_dropped == 0
456 +
457 + def test_reset_cache_clears_backpressure(self, setup):
458 + bp = self.tier.bp_detector
459 + bp.store_latency_ema = 5.0
460 + bp._under_pressure = True
461 +
462 + self.manager.reset_cache()
463 +
464 + assert bp.store_latency_ema == 0.0
465 + assert bp.is_under_pressure() is False
466 +
467 + def test_request_level_tier_respects_backpressure(self, setup):
468 + """Request-level cascade also skips pressured tiers."""
469 + # Store blocks in primary first.
470 + existing = to_keys(range(3))
471 + self._store_blocks(existing)
472 + self.tier.release_jobs()
473 + self._simulate_on_schedule_end()
474 + self._simulate_on_schedule_end()
475 +
476 + # Make tier request-level for a new request.
477 + self.tier._request_policy = OffloadPolicy.REQUEST_LEVEL
478 + ctx = ReqContext(req_id="req_rl_bp")
479 + self.manager.on_new_request(ctx)
480 +
481 + # Activate pressure.
482 + bp = self.tier.bp_detector
483 + bp._under_pressure = True
484 + bp.store_latency_ema = _BP_HIGH_WATER_S * 4
485 + initial_blocks = self.tier.get_num_blocks()
486 +
487 + # prepare_store with existing + new blocks.
488 + new = to_keys(range(10, 12))
489 + all_keys = existing + new
490 + result = self.manager.prepare_store(all_keys, ctx)
491 + assert result is not None
492 +
493 + # Existing blocks would normally cascade to request-level tier,
494 + # but should be skipped under pressure.
495 + assert self.tier.get_num_blocks() == initial_blocks
496 +
497 + def test_loads_continue_under_pressure(self, setup):
498 + """Loads from a pressured tier still work."""
499 + blocks = to_keys(range(3))
500 + for b in blocks:
501 + self.tier.blocks[b] = True
502 +
503 + bp = self.tier.bp_detector
504 + bp._under_pressure = True
505 + bp.store_latency_ema = _BP_HIGH_WATER_S * 4
506 +
507 + # Lookups should still initiate promotion.
508 + for b in blocks:
509 + result = self.manager.lookup(b, _CTX)
510 + assert result is LookupResult.HIT_PENDING
511 +
512 +
513 +class TestThrottledDropPolicy:
514 + """Tests for proportional throttling under pressure."""
515 +
516 + def _make_detector(self, **kwargs):
517 + defaults = dict(
518 + high_water_s=_BP_HIGH_WATER_S,
519 + low_water_s=_BP_LOW_WATER_S,
520 + cooldown_s=0.0,
521 + )
522 + defaults.update(kwargs)
523 + return EMABackpressureDetector(**defaults)
524 +
525 + def test_no_drop_when_not_under_pressure(self):
526 + bp = self._make_detector()
527 + for _ in range(20):
528 + assert bp.should_store(1) is True
529 +
530 + def test_no_drop_at_exactly_high_water(self):
531 + bp = self._make_detector()
532 + bp._under_pressure = True
533 + bp.store_latency_ema = _BP_HIGH_WATER_S
534 + for _ in range(20):
535 + assert bp.should_store(1) is True
536 +
537 + def test_partial_drop_at_moderate_pressure(self):
538 + bp = self._make_detector()
539 + bp._under_pressure = True
540 + bp.store_latency_ema = _BP_HIGH_WATER_S * 2
541 + results = [bp.should_store(1) for _ in range(20)]
542 + assert any(results), "some stores should be allowed"
543 + assert not all(results), "some stores should be dropped"
544 +
545 + def test_full_drop_at_severe_pressure(self):
546 + bp = self._make_detector()
547 + bp._under_pressure = True
548 + bp.store_latency_ema = _BP_HIGH_WATER_S * 4
549 + for _ in range(20):
550 + assert bp.should_store(1) is False
551 +
552 + def test_drop_rate_increases_with_severity(self):
553 + """Higher EMA → higher drop rate."""
554 + drop_counts = []
555 + for multiplier in [1.5, 2.0, 2.5, 3.5]:
556 + bp = self._make_detector()
557 + bp._under_pressure = True
558 + bp.store_latency_ema = _BP_HIGH_WATER_S * multiplier
559 + drops = sum(1 for _ in range(100) if not bp.should_store(1))
560 + drop_counts.append(drops)
561 + for i in range(len(drop_counts) - 1):
562 + assert drop_counts[i] <= drop_counts[i + 1]
563 +
564 + def test_explicit_drop_store_policy_still_drops_all(self):
565 + """DropStorePolicy (explicit) still drops everything under pressure."""
566 + bp = EMABackpressureDetector(
567 + high_water_s=_BP_HIGH_WATER_S,
568 + low_water_s=_BP_LOW_WATER_S,
569 + policy=DropStorePolicy(),
570 + cooldown_s=0.0,
571 + )
572 + bp._under_pressure = True
573 + bp.store_latency_ema = _BP_HIGH_WATER_S * 1.5
574 + for _ in range(20):
575 + assert bp.should_store(1) is False
576 +
577 +
578 +class TestSharedDropAccounting:
579 + """Both policies share drop accounting via DropAccountingPolicy."""
580 +
581 + @pytest.mark.parametrize("policy_cls", [DropStorePolicy, ThrottledDropPolicy])
582 + def test_shared_base(self, policy_cls):
583 + assert issubclass(policy_cls, DropAccountingPolicy)
584 +
585 + @pytest.mark.parametrize("policy_cls", [DropStorePolicy, ThrottledDropPolicy])
586 + def test_pop_returns_and_resets(self, policy_cls):
587 + policy = policy_cls()
588 + policy.on_store_skipped(3)
589 + policy.on_store_skipped(2)
590 + assert policy.pop_stores_dropped() == (2, 5)
591 + assert policy.pop_stores_dropped() == (0, 0)
592 +
593 + @pytest.mark.parametrize("policy_cls", [DropStorePolicy, ThrottledDropPolicy])
594 + def test_reset_clears_counters(self, policy_cls):
595 + policy = policy_cls()
596 + policy.on_store_skipped(4)
597 + policy.reset()
598 + assert policy.pop_stores_dropped() == (0, 0)
599 +
600 + def test_throttled_reset_also_clears_call_count(self):
601 + policy = ThrottledDropPolicy()
602 + policy._call_count = 7
603 + policy.reset()
604 + assert policy._call_count == 0
605 +
606 +
607 +class TestHealthyBypass:
608 + """Tests for the fast-path bypass when a tier is consistently healthy."""
609 +
610 + def _make_detector(self, **kwargs):
611 + defaults = dict(
612 + high_water_s=_BP_HIGH_WATER_S,
613 + low_water_s=_BP_LOW_WATER_S,
614 + cooldown_s=0.0,
615 + )
616 + defaults.update(kwargs)
617 + return EMABackpressureDetector(**defaults)
618 +
619 + def test_not_healthy_initially(self):
620 + bp = self._make_detector()
621 + assert bp.is_healthy() is False
622 +
623 + def test_healthy_after_streak(self):
624 + bp = self._make_detector()
625 + bp._completions = _BP_WARMUP
626 + bp._last_update = time.monotonic()
627 + block_bytes = 16
628 + fast_s_per_mib = _BP_LOW_WATER_S * 0.5
629 + fast_elapsed = fast_s_per_mib * (block_bytes / EMABackpressureDetector._MIB)
630 + for _ in range(EMABackpressureDetector._HEALTHY_THRESHOLD + _BP_WARMUP):
631 + bp.on_store_completed(fast_elapsed, block_bytes)
632 + assert bp.is_healthy() is True
633 +
634 + def test_healthy_resets_on_high_latency(self):
635 + bp = self._make_detector()
636 + bp._completions = _BP_WARMUP
637 + bp._healthy_streak = EMABackpressureDetector._HEALTHY_THRESHOLD
638 + bp._ema = _BP_LOW_WATER_S
639 + bp._last_update = time.monotonic()
640 + block_bytes = 16
641 + slow_s_per_mib = _BP_HIGH_WATER_S * 2
642 + slow_elapsed = slow_s_per_mib * (block_bytes / EMABackpressureDetector._MIB)
643 + bp.on_store_completed(slow_elapsed, block_bytes)
644 + assert bp.is_healthy() is False
645 +
646 + def test_healthy_bypasses_policy(self):
647 + bp = self._make_detector()
648 + bp._completions = _BP_WARMUP
649 + bp._healthy_streak = EMABackpressureDetector._HEALTHY_THRESHOLD
650 + assert bp.should_store(1) is True
651 +
652 + def test_reset_clears_healthy_streak(self):
653 + bp = self._make_detector()
654 + bp._healthy_streak = EMABackpressureDetector._HEALTHY_THRESHOLD
655 + bp.reset()
656 + assert bp._healthy_streak == 0
657 + assert bp.is_healthy() is False
658 +
659 +
660 +class TestCooldown:
661 + """Tests for the post-pressure cooldown period."""
662 +
663 + def _make_detector(self, **kwargs):
664 + defaults = dict(
665 + high_water_s=_BP_HIGH_WATER_S,
666 + low_water_s=_BP_LOW_WATER_S,
667 + cooldown_s=1.0,
668 + )
669 + defaults.update(kwargs)
670 + return EMABackpressureDetector(**defaults)
671 +
672 + def test_cooldown_throttles_stores(self):
673 + bp = self._make_detector()
674 + bp._completions = _BP_WARMUP
675 + bp._under_pressure = True
676 + bp._ema = _BP_HIGH_WATER_S * 2
677 + bp._last_update = time.monotonic()
678 +
679 + with patch("time.monotonic") as mock_time:
680 + t0 = bp._last_update + 10.0
681 + mock_time.return_value = t0
682 + assert bp.is_under_pressure() is False
683 + assert bp._pressure_cleared_at > 0
684 +
685 + mock_time.return_value = t0 + 0.1
686 + results = [bp.should_store(1) for _ in range(10)]
687 + assert sum(results) == 5, "cooldown should allow ~50%"
688 +
689 + def test_no_cooldown_when_never_pressured(self):
690 + bp = self._make_detector()
691 + for _ in range(20):
692 + assert bp.should_store(1) is True
693 +
694 + def test_no_cooldown_after_window_expires(self):
695 + bp = self._make_detector()
696 + bp._pressure_cleared_at = time.monotonic() - 2.0
697 + for _ in range(20):
698 + assert bp.should_store(1) is True
699 +
700 + def test_reset_clears_cooldown(self):
701 + bp = self._make_detector()
702 + bp._pressure_cleared_at = time.monotonic()
703 + bp._cooldown_count = 5
704 + bp.reset()
705 + assert bp._pressure_cleared_at == 0.0
706 + assert bp._cooldown_count == 0
vllm/v1/kv_offload/tiering/backpressure.py new
+384
@@ -0,0 +1,384 @@
1 +# SPDX-License-Identifier: Apache-2.0
2 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 +from __future__ import annotations
4 +
5 +import time
6 +from abc import ABC, abstractmethod
7 +
8 +
9 +class BackpressurePolicy(ABC):
10 + """Decides what action to take when a detector signals pressure."""
11 +
12 + @abstractmethod
13 + def should_store(self, detector: BackpressureDetector) -> bool: ...
14 +
15 + @abstractmethod
16 + def on_store_skipped(self, num_blocks: int) -> None: ...
17 +
18 + @abstractmethod
19 + def pop_stores_dropped(self) -> tuple[int, int]:
20 + """Return and reset (stores_dropped, blocks_dropped)."""
21 + ...
22 +
23 + @abstractmethod
24 + def reset(self) -> None: ...
25 +
26 +
27 +class DropAccountingPolicy(BackpressurePolicy):
28 + """Base policy that tracks dropped store/block counters.
29 +
30 + Subclasses implement ``should_store``; the shared drop accounting
31 + (``on_store_skipped``/``pop_stores_dropped``/``reset``) lives here so
32 + every policy reports these counters identically. Subclasses that add
33 + their own state override ``reset`` and call ``super().reset()``.
34 + """
35 +
36 + def __init__(self):
37 + self._stores_dropped: int = 0
38 + self._blocks_dropped: int = 0
39 +
40 + def on_store_skipped(self, num_blocks: int) -> None:
41 + self._stores_dropped += 1
42 + self._blocks_dropped += num_blocks
43 +
44 + def pop_stores_dropped(self) -> tuple[int, int]:
45 + stores, blocks = self._stores_dropped, self._blocks_dropped
46 + self._stores_dropped = 0
47 + self._blocks_dropped = 0
48 + return stores, blocks
49 +
50 + def reset(self) -> None:
51 + self._stores_dropped = 0
52 + self._blocks_dropped = 0
53 +
54 +
55 +class DropStorePolicy(DropAccountingPolicy):
56 + """Silently drop stores to pressured tiers."""
57 +
58 + def should_store(self, detector) -> bool:
59 + return not detector.is_under_pressure()
60 +
61 +
62 +class ThrottledDropPolicy(DropAccountingPolicy):
63 + """Drop stores proportionally to pressure severity.
64 +
65 + Instead of dropping all stores when pressure is detected, the drop
66 + rate scales with how far the EMA exceeds the high watermark::
67 +
68 + overshoot = (ema - high_water) / high_water
69 + drop_rate = min(1.0, overshoot / ramp_factor)
70 +
71 + With the default ``ramp_factor=2.0``, stores ramp from 0% drop at
72 + the high watermark to 100% drop at 3x the high watermark. This
73 + keeps secondary tiers populated under moderate pressure, reducing
74 + cache misses (and therefore recomputation / ITL regression) while
75 + still protecting TTFT under severe congestion.
76 + """
77 +
78 + def __init__(self, ramp_factor: float = 2.0):
79 + super().__init__()
80 + self._call_count: int = 0
81 + self._ramp_factor = ramp_factor
82 +
83 + def should_store(self, detector: BackpressureDetector) -> bool:
84 + # Proportional throttling needs the EMA and high watermark, which
85 + # only the EMA detector exposes.
86 + assert isinstance(detector, EMABackpressureDetector), (
87 + "ThrottledDropPolicy requires an EMABackpressureDetector"
88 + )
89 + if not detector.is_under_pressure():
90 + return True
91 + ema = detector.store_latency_ema
92 + high = detector._high
93 + if high <= 0:
94 + return False
95 + overshoot = max(0.0, ema - high) / high
96 + drop_rate = min(1.0, overshoot / self._ramp_factor)
97 + if drop_rate <= 0:
98 + return True
99 + if drop_rate >= 1.0:
100 + return False
101 + self._call_count += 1
102 + period = max(2, round(1.0 / drop_rate))
103 + return (self._call_count % period) != 0
104 +
105 + def reset(self) -> None:
106 + super().reset()
107 + self._call_count = 0
108 +
109 +
110 +class BackpressureDetector(ABC):
111 + """Observes store completion signals and determines pressure state."""
112 +
113 + def __init__(
114 + self,
115 + policy: BackpressurePolicy | None = None,
116 + ):
117 + self._policy = policy or ThrottledDropPolicy()
118 +
119 + @property
120 + def policy(self) -> BackpressurePolicy:
121 + return self._policy
122 +
123 + @abstractmethod
124 + def on_store_completed(self, elapsed_s: float, num_bytes: int) -> None: ...
125 +
126 + @abstractmethod
127 + def is_under_pressure(self) -> bool: ...
128 +
129 + @abstractmethod
130 + def reset(self) -> None: ...
131 +
132 + def update(self, submit_time: float, num_bytes: int) -> None:
133 + """Update pressure state from a completed store job.
134 +
135 + Args:
136 + submit_time: ``time.monotonic()`` when the job was submitted.
137 + num_bytes: Total bytes written (num_blocks * block_size_bytes).
138 +
139 + """
140 + if num_bytes <= 0:
141 + return
142 + elapsed = time.monotonic() - submit_time
143 + self.on_store_completed(elapsed, num_bytes)
144 +
145 + def should_store(self, num_blocks: int) -> bool:
146 + """Check policy and record skip if rejected."""
147 + if not self._policy.should_store(self):
148 + self._policy.on_store_skipped(num_blocks)
149 + return False
150 + return True
151 +
152 + @classmethod
153 + def default_config(
154 + cls, tier_type: str, *, locality: str | None = None
155 + ) -> dict | None:
156 + """Return default constructor kwargs for ``tier_type``, or None.
157 +
158 + Args:
159 + tier_type: The tier type string (e.g. ``"fs"``, ``"obj"``).
160 + locality: Optional locality hint from the tier config
161 + (``"LOCAL"`` or ``"REMOTE"``). When ``"REMOTE"``, tiers
162 + that would normally get local-storage watermarks (e.g.
163 + ``"fs"``) receive network watermarks instead.
164 +
165 + """
166 + return None
167 +
168 + @property
169 + def stats(self) -> dict[str, float]:
170 + return {}
171 +
172 +
173 +class EMABackpressureDetector(BackpressureDetector):
174 + """EMA of store latency normalized by transfer size.
175 +
176 + The EMA tracks seconds per megabyte (s/MiB) so that the metric is
177 + comparable regardless of how many blocks are in a job or how large
178 + each block is. Water marks are in the same unit.
179 +
180 + Default water marks are derived from fio benchmarks on the WDC H100
181 + cluster. Two presets are provided:
182 +
183 + LOCAL (NVMe/SSD): NVMe sustains ~5 GB/s writes.
184 + high=0.005 s/MiB (~200 MB/s) catches severe congestion;
185 + low=0.001 s/MiB (~1 GB/s) requires meaningful recovery.
186 +
187 + NETWORK (CephFS, object store, ``obj``tiers, or any tier
188 + with ``"locality": "REMOTE"``): CephFS sustains ~1.5 GB/s.
189 + high=0.020 s/MiB (~50 MB/s); low=0.005 s/MiB (~200 MB/s).
190 +
191 + ``obj`` tiers get NETWORK defaults automatically. An
192 + ``fs`` tier defaults to LOCAL (appropriate for NVMe/SSD); set
193 + ``"locality": "REMOTE"`` in the tier config for network-backed
194 + filesystems like CephFS.
195 + """
196 +
197 + _MIB = 1 << 20
198 +
199 + # EMA smoothing factor: higher values (→1) react faster to latency
200 + # spikes but are noisier; lower values (→0) smooth more but lag.
201 + DEFAULT_ALPHA = 0.15
202 +
203 + # Number of store completions to collect before seeding the EMA.
204 + # During warmup, pressure is never signalled; the EMA is initialized
205 + # to the mean of the warmup samples to avoid cold-start false positives.
206 + DEFAULT_WARMUP_COMPLETIONS = 10
207 +
208 + # When under pressure and no completions arrive, the EMA decays
209 + # toward zero so the detector can eventually recover. The half-life
210 + # controls how quickly: after this many seconds of silence the EMA
211 + # halves. With a 2 s half-life and high_water=0.005, an EMA frozen
212 + # at 0.010 reaches low_water=0.001 in ~6.6 s.
213 + DEFAULT_DECAY_HALF_LIFE_S = 2.0
214 +
215 + # After pressure clears, stores are throttled to 50% for this many
216 + # seconds to prevent a burst of stores from re-overwhelming the tier.
217 + DEFAULT_COOLDOWN_S = 1.0
218 +
219 + # Number of consecutive sub-low-watermark completions before the
220 + # detector enters a "healthy" fast-path that bypasses policy checks.
221 + _HEALTHY_THRESHOLD = 10
222 +
223 + LOCAL_HIGH_WATER_S = 0.010
224 + LOCAL_LOW_WATER_S = 0.002
225 +
226 + NETWORK_HIGH_WATER_S = 0.050
227 + NETWORK_LOW_WATER_S = 0.010
228 +
229 + _NETWORK_TIER_TYPES = frozenset({"obj"})
230 +
231 + @classmethod
232 + def default_config(
233 + cls, tier_type: str, *, locality: str | None = None
234 + ) -> dict | None:
235 + is_remote = locality is not None and locality.upper() == "REMOTE"
236 + if is_remote or tier_type in cls._NETWORK_TIER_TYPES:
237 + return {
238 + "high_water_s": cls.NETWORK_HIGH_WATER_S,
239 + "low_water_s": cls.NETWORK_LOW_WATER_S,
240 + }
241 + return {
242 + "high_water_s": cls.LOCAL_HIGH_WATER_S,
243 + "low_water_s": cls.LOCAL_LOW_WATER_S,
244 + }
245 +
246 + def __init__(
247 + self,
248 + high_water_s: float,
249 + low_water_s: float,
250 + alpha: float = DEFAULT_ALPHA,
251 + warmup_completions: int = DEFAULT_WARMUP_COMPLETIONS,
252 + decay_half_life_s: float = DEFAULT_DECAY_HALF_LIFE_S,
253 + cooldown_s: float = DEFAULT_COOLDOWN_S,
254 + policy: BackpressurePolicy | None = None,
255 + ):
256 + super().__init__(policy=policy)
257 + if not (0 < alpha <= 1):
258 + raise ValueError(f"alpha must be in (0, 1], got {alpha}")
259 + if low_water_s > high_water_s:
260 + raise ValueError(
261 + f"low_water_s ({low_water_s}) must be <= high_water_s ({high_water_s})"
262 + )
263 +
264 + self._alpha = alpha
265 + self._high = high_water_s
266 + self._low = low_water_s
267 + self._warmup_completions = warmup_completions
268 + self._decay_half_life_s = decay_half_life_s
269 + self._cooldown_s = cooldown_s
270 + # Exponential moving average of store latency in seconds.
271 + self._ema: float = 0.0
272 + # Whether the tier is currently considered under pressure.
273 + self._under_pressure: bool = False
274 + # Number of store completions observed (for warmup gating).
275 + self._completions: int = 0
276 + # Samples collected during warmup; used to seed the EMA.
277 + self._warmup_samples: list[float] = []
278 + # Monotonic timestamp of the last EMA update; used for idle decay.
279 + self._last_update: float = 0.0
280 + # Consecutive completions with EMA below low watermark.
281 + self._healthy_streak: int = 0
282 + # Monotonic timestamp when pressure was last cleared (for cooldown).
283 + self._pressure_cleared_at: float = 0.0
284 + self._cooldown_count: int = 0
285 +
286 + def _apply_idle_decay(self) -> None:
287 + """Decay the EMA toward zero based on elapsed idle time.
288 +
289 + Uses the formula ``ema * 0.5^(dt / half_life)`` so the EMA
290 + halves every ``decay_half_life_s`` seconds of silence. This
291 + lets the detector recover from pressure when no stores are
292 + attempted (and therefore no completions arrive to update the
293 + EMA organically).
294 + """
295 + now = time.monotonic()
296 + dt = now - self._last_update
297 + if dt <= 0 or self._decay_half_life_s <= 0:
298 + return
299 + self._ema *= 0.5 ** (dt / self._decay_half_life_s)
300 + self._last_update = now
301 +
302 + def on_store_completed(self, elapsed_s: float, num_bytes: int) -> None:
303 + now = time.monotonic()
304 + s_per_mib = elapsed_s / (num_bytes / self._MIB)
305 + self._completions += 1
306 + if self._completions <= self._warmup_completions:
307 + self._warmup_samples.append(s_per_mib)
308 + if self._completions == self._warmup_completions:
309 + self._ema = sum(self._warmup_samples) / len(self._warmup_samples)
310 + self._last_update = now
311 + return
312 + self._ema = self._alpha * s_per_mib + (1 - self._alpha) * self._ema
313 + self._last_update = now
314 + if self._ema > self._high:
315 + self._under_pressure = True
316 + self._pressure_cleared_at = 0.0
317 + self._healthy_streak = 0
318 + elif self._ema < self._low:
319 + if self._under_pressure:
320 + self._pressure_cleared_at = now
321 + self._cooldown_count = 0
322 + self._under_pressure = False
323 + self._healthy_streak += 1
324 + else:
325 + self._healthy_streak = 0
326 +
327 + def is_under_pressure(self) -> bool:
328 + if self._under_pressure and self._completions >= self._warmup_completions:
329 + self._apply_idle_decay()
330 + if self._ema < self._low:
331 + self._under_pressure = False
332 + self._pressure_cleared_at = time.monotonic()
333 + self._cooldown_count = 0
334 + return self._under_pressure
335 +
336 + def is_healthy(self) -> bool:
337 + """True when the tier has been consistently fast.
338 +
339 + After ``_HEALTHY_THRESHOLD`` consecutive store completions with
340 + EMA below the low watermark (and no active pressure), the
341 + detector enters a healthy fast-path: ``should_store()`` returns
342 + ``True`` immediately, skipping the policy check entirely.
343 + """
344 + return (
345 + not self._under_pressure and self._healthy_streak >= self._HEALTHY_THRESHOLD
346 + )
347 +
348 + def should_store(self, num_blocks: int) -> bool:
349 + under_pressure = self.is_under_pressure()
350 + if self.is_healthy():
351 + return True
352 + if (
353 + not under_pressure
354 + and self._pressure_cleared_at > 0
355 + and time.monotonic() - self._pressure_cleared_at < self._cooldown_s
356 + ):
357 + self._cooldown_count += 1
358 + if self._cooldown_count % 2 == 0:
359 + self._policy.on_store_skipped(num_blocks)
360 + return False
361 + return super().should_store(num_blocks)
362 +
363 + def reset(self) -> None:
364 + self._ema = 0.0
365 + self._under_pressure = False
366 + self._completions = 0
367 + self._warmup_samples.clear()
368 + self._last_update = 0.0
369 + self._healthy_streak = 0
370 + self._pressure_cleared_at = 0.0
371 + self._cooldown_count = 0
372 + self._policy.reset()
373 +
374 + @property
375 + def store_latency_ema(self) -> float:
376 + return self._ema
377 +
378 + @store_latency_ema.setter
379 + def store_latency_ema(self, value: float) -> None:
380 + self._ema = value
381 +
382 + @property
383 + def stats(self) -> dict[str, float]:
384 + return {"store_latency_ema": self._ema}
vllm/v1/kv_offload/tiering/base.py
+27 -1
@@ -2,9 +2,10 @@
2 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 """Abstract interfaces and data types for the secondary tiering layer."""
4
5 +import time
6 from abc import ABC, abstractmethod
7 from collections.abc import Collection, Iterable
7 -from dataclasses import dataclass
8 +from dataclasses import dataclass, field
9 from typing import TYPE_CHECKING, Any, ClassVar
10
11 import numpy as np
@@ -20,6 +21,9 @@ from vllm.v1.kv_offload.base import (
21 RequestOffloadingContext,
22 ScheduleEndContext,
23 )
24 +from vllm.v1.kv_offload.tiering.backpressure import (
25 + BackpressureDetector,
26 +)
27
28 if TYPE_CHECKING:
29 from vllm.distributed.kv_transfer.kv_connector.v1.offloading.metrics import (
@@ -52,6 +56,11 @@ class TieringOffloadingMetrics:
56 )
57 ACTIVE_PROMOTION_JOBS = "vllm:kv_offload_tiering_active_promotion_jobs"
58 ACTIVE_CASCADE_JOBS = "vllm:kv_offload_tiering_active_cascade_jobs"
59 + BACKPRESSURE_STORE_LATENCY_EMA = (
60 + "vllm:kv_offload_tiering_backpressure_store_latency_ema"
61 + )
62 + BACKPRESSURE_STORES_DROPPED = "vllm:kv_offload_tiering_backpressure_stores_dropped"
63 + BACKPRESSURE_BLOCKS_DROPPED = "vllm:kv_offload_tiering_backpressure_blocks_dropped"
64
65
66 @dataclass
@@ -63,6 +72,7 @@ class TransferJob:
72 chunk_ids: np.ndarray
73 is_promotion: bool
74 req_context: ReqContext
75 + submit_time: float = field(default_factory=time.monotonic)
76
77
78 @dataclass
@@ -77,6 +87,7 @@ class JobResult:
87 # indicated by `success`. Must be a subset of the job's original keys.
88 successful_keys: Collection[OffloadKey] | None = None
89 transfer_time: float | None = None
90 + transfer_bytes: int | None = None
91
92
93 class ParentManager(ABC):
@@ -136,18 +147,33 @@ class SecondaryTierManager(ABC):
147 offloading_spec: "OffloadingSpec",
148 primary_kv_view: memoryview,
149 tier_type: str,
150 + backpressure_detector: BackpressureDetector | None = None,
151 ) -> None:
152 """Args:
153 offloading_spec: Offloading configuration.
154 primary_kv_view: Memoryview of the primary tier's CPU KV cache.
155 tier_type: Tier type identifier, set by SecondaryTierFactory
156 from the registered tier type.
157 + backpressure_detector: Optional `BackpressureDetector`.
158
159 """
160 self._offloading_spec = offloading_spec
161 self._primary_kv_view: memoryview = primary_kv_view
162 + assert primary_kv_view.strides is not None, (
163 + "primary_kv_view.strides cannot be None"
164 + )
165 + self._block_size_bytes: int = primary_kv_view.strides[0]
166 self.tier_type = tier_type
167 self.locality: Locality | None = None
168 + self._bp_detector = backpressure_detector
169 +
170 + @property
171 + def block_size_bytes(self) -> int:
172 + return self._block_size_bytes
173 +
174 + @property
175 + def bp_detector(self) -> BackpressureDetector | None:
176 + return self._bp_detector
177
178 @abstractmethod
179 def lookup(self, key: OffloadKey, req_context: ReqContext) -> LookupResult:
vllm/v1/kv_offload/tiering/example/manager.py
+4
@@ -31,6 +31,7 @@ logger = logging.getLogger(__name__)
31
32 if TYPE_CHECKING:
33 from vllm.v1.kv_offload.base import OffloadingSpec
34 + from vllm.v1.kv_offload.tiering.backpressure import BackpressureDetector
35
36
37 class ExampleSecondaryTierManager(SecondaryTierManager):
@@ -49,6 +50,7 @@ class ExampleSecondaryTierManager(SecondaryTierManager):
50 primary_kv_view: memoryview,
51 tier_type: str,
52 custom_param: int = 0,
53 + backpressure_detector: "BackpressureDetector | None" = None,
54 ):
55 """Initialize the example secondary tier.
56
@@ -57,12 +59,14 @@ class ExampleSecondaryTierManager(SecondaryTierManager):
59 primary_kv_view: Memoryview over the primary tier's KV buffer.
60 tier_type: Name identifying this tier type.
61 custom_param: Dummy parameter demonstrating custom args.
62 + backpressure_detector: Optional backpressure detector.
63
64 """
65 super().__init__(
66 offloading_spec=offloading_spec,
67 primary_kv_view=primary_kv_view,
68 tier_type=tier_type,
69 + backpressure_detector=backpressure_detector,
70 )
71
72 logger.info(
vllm/v1/kv_offload/tiering/factory.py
+42
@@ -8,11 +8,33 @@ from vllm.logger import init_logger
8 from vllm.v1.kv_offload.tiering.base import SecondaryTierManager
9
10 if TYPE_CHECKING:
11 + from vllm.v1.kv_offload.backpressure import BackpressureDetector
12 from vllm.v1.kv_offload.base import OffloadingSpec
13
14 logger = init_logger(__name__)
15
16
17 +_DEFAULT_PACKAGE = "vllm.v1.kv_offload.tiering.backpressure"
18 +
19 +
20 +def _import_class(name: str) -> type:
21 + """Import a class by name.
22 +
23 + If ``name`` contains a dot it is treated as a fully qualified path
24 + (e.g. ``'my_package.MyDetector'``). Otherwise it is looked up in
25 + the default backpressure module
26 + (``vllm.v1.kv_offload.tiering.backpressure``), so short names like
27 + ``'DropStorePolicy'`` or ``'EMABackpressureDetector'`` work out of
28 + the box.
29 + """
30 + if "." in name:
31 + module_path, _, class_name = name.rpartition(".")
32 + else:
33 + module_path, class_name = _DEFAULT_PACKAGE, name
34 + module = importlib.import_module(module_path)
35 + return getattr(module, class_name)
36 +
37 +
38 class SecondaryTierFactory:
39 """Registry for SecondaryTierManager implementations, resolved by type.
40
@@ -47,10 +69,30 @@ class SecondaryTierFactory:
69 config = tier_config.copy()
70 tier_type = config.pop("type")
71 config.pop("module_path", None)
72 + locality = config.get("locality")
73 + bp_config = config.pop("backpressure", None)
74 + bp_detector = None
75 + if bp_config is not None:
76 + bp_config = bp_config.copy()
77 + bp_cls_name = bp_config.pop("backpressure_cls", None)
78 + if bp_cls_name is None:
79 + raise ValueError(
80 + "backpressure config must include 'backpressure_cls' "
81 + "(e.g. 'EMABackpressureDetector')"
82 + )
83 + policy_cls_name = bp_config.pop("policy_cls", None)
84 + policy = _import_class(policy_cls_name)() if policy_cls_name else None
85 + detector_cls: type[BackpressureDetector] = _import_class(bp_cls_name)
86 + defaults = detector_cls.default_config(tier_type, locality=locality)
87 + if defaults:
88 + for k, v in defaults.items():
89 + bp_config.setdefault(k, v)
90 + bp_detector = detector_cls(policy=policy, **bp_config)
91 return tier_cls(
92 offloading_spec=offloading_spec,
93 primary_kv_view=primary_kv_view,
94 tier_type=tier_type,
95 + backpressure_detector=bp_detector,
96 **config,
97 )
98
vllm/v1/kv_offload/tiering/fs/manager.py
+14 -1
@@ -40,6 +40,7 @@ from vllm.v1.kv_offload.base import (
40 )
41 from vllm.v1.kv_offload.file_mapper import FileMapper
42 from vllm.v1.kv_offload.tiering.async_lookup import AsyncLookupManager
43 +from vllm.v1.kv_offload.tiering.backpressure import BackpressureDetector
44 from vllm.v1.kv_offload.tiering.base import (
45 JobId,
46 JobResult,
@@ -115,6 +116,7 @@ class FileSystemTierManager(SecondaryTierManager):
116 n_write_threads: int = 16,
117 enable_kv_events: bool = False,
118 locality: str | None = None,
119 + backpressure_detector: BackpressureDetector | None = None,
120 ):
121 """Args:
122 offloading_spec: Contains normalized offloading configuration and
@@ -129,9 +131,12 @@ class FileSystemTierManager(SecondaryTierManager):
131 cache events are enabled globally (kv_events_config).
132 locality: Whether this tier's storage is LOCAL or REMOTE relative
133 to the publishing vLLM instance.
134 + backpressure_detector: Optional backpressure detector.
135
136 """
134 - super().__init__(offloading_spec, primary_kv_view, tier_type)
137 + super().__init__(
138 + offloading_spec, primary_kv_view, tier_type, backpressure_detector
139 + )
140 self.locality = Locality(locality) if locality is not None else None
141
142 self.events: list[OffloadingEvent] | None = None
@@ -150,6 +155,8 @@ class FileSystemTierManager(SecondaryTierManager):
155 # Keys of in-flight load (promotion) jobs, so a failed load can mark
156 # its own cached lookup verdicts False (see get_finished_jobs).
157 self._load_job_keys: dict[JobId, list[OffloadKey]] = {}
158 + # Block count per in-flight job, used to report transfer_bytes.
159 + self._job_block_counts: dict[JobId, int] = {}
160 # Per load job: how many blocks loaded before a failure (partial keep).
161 # Written by the pool worker inside the load task before it raises (so
162 # before task_done publishes the job); read on the scheduler thread in
@@ -225,6 +232,7 @@ class FileSystemTierManager(SecondaryTierManager):
232 self._block_size,
233 self._use_o_direct,
234 )
235 + self._job_block_counts[job_metadata.job_id] = len(keys)
236 self._pool.enqueue_store(job_metadata.job_id, 1, [task])
237
238 @override
@@ -234,6 +242,7 @@ class FileSystemTierManager(SecondaryTierManager):
242 # keys as a miss (see get_finished_jobs).
243 keys = list(job_metadata.keys)
244 self._load_job_keys[job_id] = keys
245 + self._job_block_counts[job_id] = len(keys)
246 paths = [self.file_mapper.get_file_name(key) for key in keys]
247 offsets = [int(cid) * self._block_size for cid in job_metadata.chunk_ids]
248
@@ -272,6 +281,8 @@ class FileSystemTierManager(SecondaryTierManager):
281 as a miss here (scheduler thread)."""
282 results = []
283 for job_id, success, transfer_time in self._pool.get_finished():
284 + block_count = self._job_block_counts.pop(job_id, 0)
285 + transfer_bytes = block_count * self._block_size if block_count else None
286 if self.events is not None:
287 keys = self._store_job_keys.pop(job_id, None)
288 if success and keys:
@@ -299,6 +310,7 @@ class FileSystemTierManager(SecondaryTierManager):
310 success=False,
311 successful_keys=tuple(successful) if successful else None,
312 transfer_time=transfer_time,
313 + transfer_bytes=transfer_bytes,
314 )
315 )
316 continue
@@ -307,6 +319,7 @@ class FileSystemTierManager(SecondaryTierManager):
319 job_id=job_id,
320 success=success,
321 transfer_time=transfer_time,
322 + transfer_bytes=transfer_bytes,
323 )
324 )
325 return results
vllm/v1/kv_offload/tiering/manager.py
+51 -1
@@ -238,6 +238,10 @@ class TieringOffloadingManager(OffloadingManager):
238 for tier_idx, tier in enumerate(self.secondary_tiers)
239 }
240
241 + self._tier_index: dict[SecondaryTierManager, int] = {
242 + tier: i for i, tier in enumerate(self.secondary_tiers)
243 + }
244 +
245 @property
246 def _transfer_jobs(self) -> dict[JobId, JobMetadata]:
247 return self._jobs
@@ -337,6 +341,43 @@ class TieringOffloadingManager(OffloadingManager):
341 self.primary_tier.complete_read(
342 transfer_job.keys, transfer_job.req_context
343 )
344 + if completed_job.success:
345 + self._update_backpressure(tier, job_metadata, completed_job)
346 +
347 + def _should_store_to_tier(
348 + self, tier: SecondaryTierManager, num_blocks: int
349 + ) -> bool:
350 + detector = tier.bp_detector
351 + if detector is None:
352 + return True
353 + return detector.should_store(num_blocks)
354 +
355 + def _update_backpressure(
356 + self,
357 + tier: SecondaryTierManager,
358 + job_metadata: JobMetadata,
359 + completed_job: JobResult,
360 + ) -> None:
361 + detector = tier.bp_detector
362 + if detector is None:
363 + return
364 + was_under_pressure = detector.is_under_pressure()
365 + tj = job_metadata.transfer_job
366 + num_bytes = (
367 + completed_job.transfer_bytes
368 + if completed_job.transfer_bytes is not None
369 + else len(tj.keys) * tier.block_size_bytes
370 + )
371 + detector.update(tj.submit_time, num_bytes)
372 + if detector.is_under_pressure() != was_under_pressure:
373 + tier_idx = self._tier_index[tier]
374 + logger.info(
375 + "Tier #%d (%s) back-pressure %s (stats=%s)",
376 + tier_idx,
377 + tier.tier_type,
378 + "activated" if detector.is_under_pressure() else "cleared",
379 + detector.stats,
380 + )
381
382 @override
383 def lookup(
@@ -643,8 +684,10 @@ class TieringOffloadingManager(OffloadingManager):
684 return
685
686 for tier_idx in request_level_tiers:
646 - job_metadata = self.create_store_job(ready_keys, req_context, tier_idx)
687 tier = self.secondary_tiers[tier_idx]
688 + if not self._should_store_to_tier(tier, len(ready_keys)):
689 + continue
690 + job_metadata = self.create_store_job(ready_keys, req_context, tier_idx)
691 tier.submit_store(job_metadata)
692
693 def _flush_pending_cascades(self) -> None:
@@ -698,6 +741,8 @@ class TieringOffloadingManager(OffloadingManager):
741 # eviction during the async transfer). One prepare_read() call per
742 # secondary tier.
743 for tier_idx, tier in enumerate(self.secondary_tiers):
744 + if not self._should_store_to_tier(tier, len(keys)):
745 + continue
746 job_metadata = self.create_store_job(keys, req_context, tier_idx)
747 tier.submit_store(job_metadata)
748
@@ -904,6 +949,10 @@ class TieringOffloadingManager(OffloadingManager):
949 del self._req_state[req_id]
950 self._processed_jobs_this_step = False
951
952 + for tier in self.secondary_tiers:
953 + if tier.bp_detector is not None:
954 + tier.bp_detector.reset()
955 +
956 @override
957 def get_stats(self) -> OffloadingConnectorStats | None:
958 stats = self.primary_tier.get_stats()
@@ -911,6 +960,7 @@ class TieringOffloadingManager(OffloadingManager):
960 if stats is not None and stats.is_empty():
961 stats = None
962
963 + self._metrics.record_backpressure(self.secondary_tiers)
964 metrics_stats = self._metrics.take_stats()
965 if metrics_stats is not None:
966 if stats is None:
vllm/v1/kv_offload/tiering/metrics.py
+31
@@ -12,6 +12,7 @@ from vllm.distributed.kv_transfer.kv_connector.v1.offloading.metrics import (
12 from vllm.v1.kv_offload.base import LookupResult, OffloadKey, ReqContext
13 from vllm.v1.kv_offload.tiering.base import (
14 JobResult,
15 + SecondaryTierManager,
16 TieringOffloadingMetrics,
17 TransferJob,
18 )
@@ -130,6 +131,36 @@ class TieringMetricsTracker:
131 TieringOffloadingMetrics.PROMOTION_ALLOCATION_FAILURES
132 )
133
134 + def record_backpressure(
135 + self,
136 + tiers: list[SecondaryTierManager],
137 + ) -> None:
138 + for i, tier in enumerate(tiers):
139 + detector = tier.bp_detector
140 + if detector is None:
141 + continue
142 + label = self.tier_label(i)
143 + ema = detector.stats.get("store_latency_ema")
144 + if ema is not None:
145 + self._stats.set_gauge(
146 + TieringOffloadingMetrics.BACKPRESSURE_STORE_LATENCY_EMA,
147 + ema,
148 + labelvalues=label,
149 + )
150 + stores_dropped, blocks_dropped = detector.policy.pop_stores_dropped()
151 + if stores_dropped > 0:
152 + self._stats.increase_counter(
153 + TieringOffloadingMetrics.BACKPRESSURE_STORES_DROPPED,
154 + stores_dropped,
155 + labelvalues=label,
156 + )
157 + if blocks_dropped > 0:
158 + self._stats.increase_counter(
159 + TieringOffloadingMetrics.BACKPRESSURE_BLOCKS_DROPPED,
160 + blocks_dropped,
161 + labelvalues=label,
162 + )
163 +
164 def take_stats(self) -> OffloadingConnectorStats | None:
165 active_transfer_stats = OffloadingConnectorStats()
166 self._observe_active_transfer_stats(active_transfer_stats)
vllm/v1/kv_offload/tiering/obj/manager.py
+9 -1
@@ -22,6 +22,7 @@ from vllm.v1.kv_offload.base import (
22 )
23 from vllm.v1.kv_offload.file_mapper import FileMapper
24 from vllm.v1.kv_offload.tiering.async_lookup import AsyncLookupManager
25 +from vllm.v1.kv_offload.tiering.backpressure import BackpressureDetector
26 from vllm.v1.kv_offload.tiering.base import (
27 JobId,
28 JobResult,
@@ -112,6 +113,7 @@ class ObjectStoreSecondaryTierManager(SecondaryTierManager):
113 io_threads: int = 4,
114 enable_kv_events: bool = False,
115 locality: str | None = None,
116 + backpressure_detector: BackpressureDetector | None = None,
117 ):
118 """Args:
119 offloading_spec: Offloading configuration.
@@ -125,9 +127,15 @@ class ObjectStoreSecondaryTierManager(SecondaryTierManager):
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 + backpressure_detector: Optional backpressure detector.
131
132 """
130 - super().__init__(offloading_spec, primary_kv_view, tier_type)
133 + super().__init__(
134 + offloading_spec,
135 + primary_kv_view,
136 + tier_type,
137 + backpressure_detector,
138 + )
139 self.locality = Locality(locality) if locality is not None else None
140
141 self.events: list[OffloadingEvent] | None = None
vllm/v1/kv_offload/tiering/p2p/manager.py
+19 -1
@@ -265,7 +265,25 @@ class P2PSecondaryTierManager(SecondaryTierManager):
265 number, or anything convertible to one.
266
267 """
268 - super().__init__(offloading_spec, primary_kv_view, tier_type)
268 + backpressure_detector = kwargs.pop("backpressure_detector", None)
269 + if backpressure_detector is not None:
270 + # The generic (store-latency) detector is unreliable for P2P: in
271 + # PD mode dropping a store can leave the decoder waiting until the
272 + # load timeout instead of failing fast, and rendezvous time makes
273 + # store latency a poor pressure signal. Reject it until a
274 + # P2P-specific fail-fast path exists.
275 + raise ValueError(
276 + "Backpressure is not supported for the P2P secondary tier. "
277 + "The generic store-latency detector cannot fail fast in PD "
278 + "mode and rendezvous time makes store latency an unreliable "
279 + "pressure signal. Remove the 'backpressure' config from the "
280 + "p2p tier."
281 + )
282 + super().__init__(
283 + offloading_spec,
284 + primary_kv_view,
285 + tier_type,
286 + )
287 try:
288 timeout_s = float(unbound_store_timeout_s)
289 except (TypeError, ValueError):
vllm/v1/kv_offload/tiering/spec.py
+54
@@ -247,6 +247,34 @@ class TieringOffloadingSpec(CPUOffloadingSpec):
247 assert isinstance(tier_config, dict)
248 tier_cls = SecondaryTierFactory.get_tier_class(tier_config)
249 metrics.update(tier_cls.build_metric_definitions(tier_config))
250 +
251 + metrics[TieringOffloadingMetrics.BACKPRESSURE_STORE_LATENCY_EMA] = (
252 + OffloadingGaugeMetadata(
253 + documentation=(
254 + "Exponential moving average of store latency "
255 + "for back-pressure detection, in s/MiB."
256 + ),
257 + labelnames=("tier",),
258 + )
259 + )
260 + metrics[TieringOffloadingMetrics.BACKPRESSURE_STORES_DROPPED] = (
261 + OffloadingCounterMetadata(
262 + documentation=(
263 + "Number of store operations dropped due to "
264 + "back-pressure on a secondary tier."
265 + ),
266 + labelnames=("tier",),
267 + )
268 + )
269 + metrics[TieringOffloadingMetrics.BACKPRESSURE_BLOCKS_DROPPED] = (
270 + OffloadingCounterMetadata(
271 + documentation=(
272 + "Number of blocks dropped due to back-pressure on a secondary tier."
273 + ),
274 + labelnames=("tier",),
275 + )
276 + )
277 +
278 return metrics
279
280 def __init__(self, config: OffloadingConfig):
@@ -259,6 +287,32 @@ class TieringOffloadingSpec(CPUOffloadingSpec):
287 if not isinstance(self.secondary_tier_configs, list):
288 raise ValueError("secondary_tiers must be a list of tier configurations")
289
290 + # Backpressure config is merged field-by-field in priority order
291 + # (highest first):
292 + # 1. Per-tier ``backpressure`` dict in the tier config
293 + # 2. Top-level ``backpressure`` in kv_connector_extra_config
294 + # Merging per field (rather than per whole dict) means a partial
295 + # tier override still inherits missing fields from the top-level
296 + # default, so the resolved dict reaching the factory is complete.
297 + # Within each tier's resolved dict, tier-type-aware water marks
298 + # are filled in last so a bare ``"backpressure": {}`` picks up
299 + # sensible thresholds for the storage medium.
300 + bp_defaults = self.extra_config.get("backpressure")
301 +
302 + for tier_cfg in self.secondary_tier_configs:
303 + tier_override = tier_cfg.get("backpressure")
304 + # Overlay from lowest to highest precedence so higher-precedence
305 + # fields win while lower-precedence ones fill in the gaps.
306 + merged: dict[str, Any] = {}
307 + for source in (bp_defaults, tier_override):
308 + if source:
309 + merged.update(source)
310 + # Only set a resolved dict when at least one source contributed
311 + # (or an explicit ``backpressure`` key was present, e.g. ``{}``),
312 + # so tiers without any backpressure config stay unconfigured.
313 + if merged or "backpressure" in tier_cfg:
314 + tier_cfg["backpressure"] = merged
315 +
316 # Scheduler-side mmap (rank=None); kept for cleanup
317 self._scheduler_mmap: SharedOffloadRegion | None = None
318