[KV Offloading] Per-request tier filtering with TierFilter/TierMatcher

[ronensc]

Purpose

Adds per-request tier filtering to KV cache offloading.
Requests can specify which secondary tiers to load from via kv_transfer_params["kv_load_tiers"], filtering by medium and locality.

See detailed design discussion: #48123 (comment)

Key changes:

• Medium enum: CPU/STORAGE (coarse granularity, both FS and OBJ → STORAGE)
• TierMatcher(NamedTuple) with optional medium and locality fields
• TierFilter with allows() — applied to secondary tiers only (primary always participates)
• _parse_tier_filter() validates request params with warning logging
• MEDIUM_STORAGE wire constant in kv_events.py (replaces MEDIUM_FS/MEDIUM_OBJ)
• Locality enum on SecondaryTierManager base class

Design decisions (confirmed by reviewers):

• Filter gates secondary tiers only; primary always participates (it's the promotion target)
• None on tier side = "unconstrained" (passes any filter) — supports P2P tiers with dynamic locality
• None on matcher side = "match anything"

Test Plan

pytest -v tests/v1/kv_offload/tiering/ \
          tests/v1/kv_offload/cpu/test_manager.py \
          tests/v1/kv_connector/unit/offloading_connector/test_events.py \
          tests/v1/kv_connector/unit/offloading_connector/test_scheduler.py

Test Result

437 passed, 17 warnings in 183.26s (0:03:03) 

---

Essential Elements of an Effective PR Description Checklist

• The purpose of the PR, such as "Fix some issue (link existing issues this PR will resolve)".
• The test plan, such as providing test command.
• The test results, such as pasting the results comparison before and after, or e2e results
• (Optional) The necessary documentation update, such as updating supported_models.md and examples for a new model.

[claude]

Claude Code Review

This pull request is from a fork — automated review is disabled. A repository maintainer can comment @claude review to run a one-time review.

[ronensc]

Open question: how should unknown lookup_scope values be handled? Currently falls through to "all" silently.
Options:
(1) log a warning — but this is per-block hot path, would spam on typos;
(2) reject the request upstream at the API/connector layer;
(3) keep silent for forward-compat with newer clients sending values an older server doesn't know.

[ronensc]

Open question: should we support a server-level config to change the default lookup_scope from "all" to "primary"? Use case: keep store cascade (writes to all tiers for durability) but avoid promotion on reads unless a request explicitly opts in with lookup_scope="all".

[orozery]

How about this instead?

@dataclass
class ReqContext:
    req_id: str
    kv_transfer_params: dict[str, Any] | None = None
    kv_lookup_scope: LookupScope = LookupScope.ALL

[ronensc]

Wouldn't we still need to parse "lookup_scope" from kv_transfer_params regardless?
IIUC this would require _create_req_context in the offloading connector scheduler to extract and set the field:

def _create_req_context(req: Request) -> ReqContext:
    return ReqContext(
        req_id=req.request_id,
        kv_transfer_params=req.kv_transfer_params,
        kv_lookup_scope=LookupScope(req.kv_transfer_params.get("lookup_scope", "all")),
    )

This makes the offloading connector aware of a tiering-specific concept.
Am I missing something?

[orozery]

This is problematic since "primary" is a relative scope.
I think it should be something concrete (e.g. GPU/CPU etc.).

[ronensc]

Done in e85757b

[orozery]

@ronensc Did some thinking with Claude and came up with a longer-term design.
Would be happy to get your thoughts, and see if we can align this PR to it.
We can start with just supporting the medium filtering for load.

Per-request KV cache tier filtering

Overview

A per-request mechanism to control which offloading tiers participate in load (lookup + promotion) and store (offload + cascade) paths.

API (kv_transfer_params)

Two optional keys:

Key	Default	Controls
kv_load_tiers	all tiers	lookup + promotion
kv_store_tiers	all tiers	offload + cascade

Each is a list of tier matchers. A tier participates if it matches any entry (OR). Each entry is an AND of its specified fields; omitted fields are unconstrained.

{
  "kv_load_tiers": [
    {"medium": "cpu", "locality": "global"},
    {"medium": "storage", "locality": "local"}
  ]
}

Filter axes

Medium — storage type:

Value	Examples
cpu	Host DRAM (primary tier, P2P)
storage	Persistent (local FS, object store)

Future: accelerator — handled above the offloading manager; the tiering layer never sees it.

Locality — placement relative to the vLLM instance:

Value	Meaning
local	Co-located with this instance (no IPC/network)
global	Anywhere

Future values slot in between: host (same machine), rack, dc, region. A tier declares one value; the request lists accepted values.

Entry syntax

Each entry is {field: value, ...}. Omitted fields are unconstrained (match anything). Multiple entries are OR'd — a tier matches if it satisfies any single entry fully.

[{"medium": "cpu"}, {"medium": "storage", "locality": "local"}]

Means: any CPU tier, OR a storage tier that is local.

Examples

Intent	Param value
Primary tier only	[{"medium": "cpu", "locality": "local"}]
CPU tiers everywhere	[{"medium": "cpu"}]
Everything local	[{"locality": "local"}]
Remote CPU + local storage	[{"medium": "cpu", "locality": "global"}, {"medium": "storage", "locality": "local"}]
GDS store bypass CPU	kv_store_tiers: [{"medium": "storage"}]
Everything (default)	omit key entirely

Implementation

Parse once in _create_req_context (offloading connector). This consolidates all kv_transfer_params parsing — including existing P2P params (prefill, decode) which today are re-parsed on every P2P tier call.

@dataclass
class ReqContext:
    req_id: str
    kv_transfer_params: dict[str, Any] | None = None
    load_tier_filter: TierFilter = TierFilter.ALL
    store_tier_filter: TierFilter = TierFilter.ALL
    prefill_params: dict[str, Any] | None = None
    decode_params: dict[str, Any] | None = None

TierFilter wraps the parsed list of matchers:

class TierMatcher(NamedTuple):
    medium: Medium | None = None    # None = unconstrained
    locality: Locality | None = None

@dataclass(frozen=True)
class TierFilter:
    matchers: tuple[TierMatcher, ...] = ()  # empty = match nothing

    ALL: ClassVar["TierFilter"]  # matches everything

    def allows(self, medium: Medium, locality: Locality) -> bool:
        return any(
            (m.medium is None or m.medium == medium)
            and (m.locality is None or m.locality == locality)
            for m in self.matchers
        )

TierFilter.ALL = TierFilter(matchers=(TierMatcher(),))  # single unconstrained entry

Enforce at the tier (not the tiering manager). Each SecondaryTierManager checks against the filter:

class FileSystemTierManager(SecondaryTierManager):
    medium = Medium.STORAGE
    locality = Locality.LOCAL  # set from config

    def lookup(self, key, req_context):
        if not req_context.load_tier_filter.allows(self.medium, self.locality):
            return LookupResult.MISS
        ...

The tiering manager stays unaware of filtering semantics.

[ronensc]

@orozery Thanks for the detailed design, this is a good direction.

Two thoughts:

Parsing location — _create_req_context lives in the offloading connector scheduler. Parsing tier filter semantics there makes it aware of tiering-specific concepts. Should we keep the parsing local to the tiering layer (e.g., in TieringOffloadingManager.on_new_request or the lookup loop itself)?

Enforcement location — I'd prefer centralizing the filter check in the tiering manager's loop rather than in each tier. The manager already decides who to call (e.g. exclude_tier), so this is the same pattern generalized. Benefits: a new tier can't forget to implement the check, and we skip the function call entirely for filtered-out tiers. The tier just declares its properties (medium, locality); the manager decides whether to query it.

[orozery]

Parsing location — _create_req_context lives in the offloading connector scheduler. Parsing tier filter semantics there makes it aware of tiering-specific concepts. Should we keep the parsing local to the tiering layer (e.g., in TieringOffloadingManager.on_new_request or the lookup loop itself)?

My thinking that the existing kv_transfer_params fields used by the p2p tier, as well as the new proposed fields here should be "officially supported" by the offloading connector.
We still include kv_transfer_params in req_context to allow secondary tiers to define whatever custom params they want.

Enforcement location — I'd prefer centralizing the filter check in the tiering manager's loop rather than in each tier. The manager already decides who to call (e.g. exclude_tier), so this is the same pattern generalized. Benefits: a new tier can't forget to implement the check, and we skip the function call entirely for filtered-out tiers. The tier just declares its properties (medium, locality); the manager decides whether to query it.

The problem is that you can have complex tiers that don't have a static medium/locality.
One example is the P2P tier itself.

[mergify]

This pull request has merge conflicts that must be resolved before it can be
merged. Please rebase the PR, @ronensc.

https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/working-with-forks/syncing-a-fork

[orozery]

Suggested change

	event_medium: ClassVar[str] = MEDIUM_OBJ
	filter_medium: ClassVar[Medium | None] = Medium.STORAGE
	medium: ClassVar[Medium | None] = Medium.STORAGE

Let's reconcile to a single classvar, and switch OffloadingEvent to use the Medium enum instead of str.
Then, in OffloadingEventsTracker (events.py) let's create a util function to convert Medium to str (e.g. MEDIUM_OBJ).

Same change on the fs tier.

[ronensc]

done in 435d1fb

[orozery]

Re-thinking this, since we already introduce Medium as a first-class citizen in kv_offload/base.py, I think your previous approach actually fits.
i.e., add a Medium | None classvar on SecondaryTierManager, and applying the filter on the tiering manager.

[ronensc]

done in c74e55c

[mergify]

This pull request has merge conflicts that must be resolved before it can be
merged. Please rebase the PR, @ronensc.

https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/working-with-forks/syncing-a-fork

[ronensc]

I removed MEDIUM_CPU/MEDIUM_FS/MEDIUM_OBJ (now covered by Medium enum) but kept MEDIUM_GPU — it's used in block_pool.py and I wasn't sure if it's OK for v1/core/block_pool.py to import from v1/kv_offload/base.py.

[orozery]

Actually, we don't want to touch this file.
It defines an API which is distinct from the one in kv_offload.
Instead, in the offloading connector we want translate kv_offload Medium to one of the strs defined here in kv_events.py.

[ronensc]

Makes sense. Done in 886bb89

Note that since we consolidated filter_medium and event_medium into a single enum, we no longer have Medium.STORAGE as articulated in the original design (instead we have FS and OBJ).

[orozery]

@Change72 Didn't we agree that medium will be CPU/STORAGE (and not FS/OBJ)?

[Change72]

Yes — that was my understanding as well. We agreed that Medium should remain coarse (CPU / STORAGE) at the end of #48281. FS and OBJ events should report as STORAGE.

[ronensc]

Thanks @orozery @Change72. I've updated Medium to CPU/STORAGE as agreed in #48281. Both FS and OBJ tiers now declare medium = Medium.STORAGE.

Two questions:

1. MEDIUM_FS and MEDIUM_OBJ in kv_events.py are now unused by the offloading path. Should we remove them?
2. _MEDIUM_TO_EVENT_STR maps Medium.CPU → MEDIUM_CPU (from kv_events.py), but Medium.STORAGE → bare "STORAGE" string (no constant exists). I hesitated to add MEDIUM_STORAGE = "STORAGE" to kv_events.py given the earlier guidance not to touch that file. Should we add it?

[orozery]

Yes to both.

[ronensc]

done in 4a39181

[mergify]

This pull request has merge conflicts that must be resolved before it can be
merged. Please rebase the PR, @ronensc.

https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/working-with-forks/syncing-a-fork

[orozery]

Thanks @ronensc @Change72 !