vllm.v1.kv_offload.tiering.manager

TieringOffloadingManager: Multi-tier KV cache offloading orchestrator.

This manager coordinates between a CPU primary tier (with direct GPU access) and zero or more secondary tiers (Storage, Network, etc.) to provide hierarchical KV cache offloading.

Key Design Principles: 1. Always offload to all tiers — When a chunk is stored to the primary tier, it is cascaded to ALL secondary tiers 2. Primary tier is the gateway — Secondary tiers cannot access GPU memory directly; all data flows through the CPU primary tier 3. Staged promotion — Chunks in secondary tiers must be promoted to the primary tier before GPU can access them 4. Transparent retry mechanism — Return None from lookup() to signal "data is being promoted, try later" 5. ref_cnt as eviction protection — primary.prepare_read() increments ref_cnt, protecting chunks from eviction until complete_read() is called

Classes:

• CPUPrimaryTierOffloadingManager –

  CPUOffloadingManager with a primary/secondary transfer interface.

• PendingPromotion –

  Accumulator for chunks awaiting submit_load() for one (tier, request).

• TieringOffloadingManager –

  Orchestrates multi-tier KV cache offloading.

CPUPrimaryTierOffloadingManager

Bases: CPUOffloadingManager

CPUOffloadingManager with a primary/secondary transfer interface.

The inherited prepare_store/complete_store/prepare_load/complete_load are the GPU-facing OffloadingManager interface. These aliases expose the same operations from the secondary tier perspective, where read/write refers to secondary accessing primary. This avoids confusion when reading TieringOffloadingManager code (e.g. calling prepare_load inside a cascade/store path would be misleading).

Methods:

• get_kv_memoryview –

  Return the memoryview over the primary tier's KV cache buffer.

• prepare_read –

  Pin chunks for a CPU-to-secondary transfer.

Source code in vllm/v1/kv_offload/tiering/manager.py

class CPUPrimaryTierOffloadingManager(CPUOffloadingManager):
    """CPUOffloadingManager with a primary/secondary transfer interface.

    The inherited prepare_store/complete_store/prepare_load/complete_load are the
    GPU-facing OffloadingManager interface. These aliases expose the same operations
    from the secondary tier perspective, where read/write refers to secondary
    accessing primary. This avoids confusion when reading TieringOffloadingManager
    code (e.g. calling prepare_load inside a cascade/store path would be misleading).
    """

    def __init__(
        self,
        num_chunks: int,
        mmap_region: SharedOffloadRegion,
        cache_policy: str = "lru",
        cache_policy_module_path: str | None = None,
        enable_events: bool = False,
    ):
        super().__init__(
            num_chunks=num_chunks,
            cache_policy=cache_policy,
            cache_policy_module_path=cache_policy_module_path,
            enable_events=enable_events,
        )
        self._mmap_region = mmap_region
        # read/write is for CPU<->secondary transfers,
        # load/store is for CPU<->GPU transfers.
        # These aliases avoid calling prepare_load inside a store path.
        self.complete_read = self.complete_load
        self.prepare_write = self.prepare_store
        self.complete_write = self.complete_store

        self._kv_memoryview = mmap_region.create_kv_memoryview()

    def prepare_read(
        self, keys: Collection[OffloadKey], req_context: ReqContext
    ) -> LoadStoreSpec:
        """Pin chunks for a CPU-to-secondary transfer.

        Cascade reads are implementation details of tiering, not additional
        request accesses, so they must not alter request-scoped recency.
        """
        return self._prepare_load(keys, req_context, record_access=False)

    def get_kv_memoryview(self) -> memoryview:
        """Return the memoryview over the primary tier's KV cache buffer.

        The view has shape (num_chunks, row_stride_bytes) and is backed by the
        SharedOffloadRegion mmap.  Secondary tiers address chunk *c* as
        ``view[c]``.
        """
        return self._kv_memoryview

    @override
    def shutdown(self) -> None:
        super().shutdown()
        self._kv_memoryview.release()
        self._mmap_region.cleanup()

get_kv_memoryview()

Return the memoryview over the primary tier's KV cache buffer.

The view has shape (num_chunks, row_stride_bytes) and is backed by the SharedOffloadRegion mmap. Secondary tiers address chunk c as view[c].

Source code in vllm/v1/kv_offload/tiering/manager.py

def get_kv_memoryview(self) -> memoryview:
    """Return the memoryview over the primary tier's KV cache buffer.

    The view has shape (num_chunks, row_stride_bytes) and is backed by the
    SharedOffloadRegion mmap.  Secondary tiers address chunk *c* as
    ``view[c]``.
    """
    return self._kv_memoryview

prepare_read(keys, req_context)

Pin chunks for a CPU-to-secondary transfer.

Cascade reads are implementation details of tiering, not additional request accesses, so they must not alter request-scoped recency.

Source code in vllm/v1/kv_offload/tiering/manager.py

def prepare_read(
    self, keys: Collection[OffloadKey], req_context: ReqContext
) -> LoadStoreSpec:
    """Pin chunks for a CPU-to-secondary transfer.

    Cascade reads are implementation details of tiering, not additional
    request accesses, so they must not alter request-scoped recency.
    """
    return self._prepare_load(keys, req_context, record_access=False)

PendingPromotion dataclass

Accumulator for chunks awaiting submit_load() for one (tier, request).

Source code in vllm/v1/kv_offload/tiering/manager.py

@dataclass
class PendingPromotion:
    """Accumulator for chunks awaiting submit_load() for one (tier, request)."""

    req_context: ReqContext
    keys: list[OffloadKey] = field(default_factory=list)
    chunk_ids: list[int] = field(default_factory=list)

TieringOffloadingManager

Bases: OffloadingManager

Orchestrates multi-tier KV cache offloading.

This manager coordinates between a CPU primary tier (with direct GPU access) and zero or more secondary tiers (Storage, Network, etc.) to provide hierarchical KV cache offloading.

Key internal state

• Minimal state tracking; relies on secondary tiers to report completion via get_finished_jobs()
• Secondary tiers return JobResult objects containing all necessary information
• job_id_counter: monotonically increasing counter for job IDs

Methods:

• __init__ –

  Initialize the TieringOffloadingManager.

• complete_load –

  Mark chunks as done loading from primary tier to GPU.

• complete_store –

  Mark chunks as done storing from GPU to primary tier.

• create_store_job –

  Pin chunks in the primary tier and create a tracked store job.

• lookup –

  Check whether a single chunk is offloaded and ready.

• on_new_request –

  Query each secondary tier for its offload policy preference.

• on_schedule_end –

  End-of-schedule hook: process finished jobs, flush deferred

• poll_tiers –

  Advance the control plane of tiers that cannot wait for a step.

• prepare_load –

  Prepare chunks to be loaded from primary tier to GPU.

• prepare_store –

  Prepare chunks to be stored from GPU to primary tier.

• reset_cache –

  Reset transfer bookkeeping and primary-tier cache.

• shutdown –

  Shut down secondary tiers before releasing primary resources.

• take_events –

  Yield events owned by the primary and secondary tiers.

• touch –

  Mark chunks as recently used in all tiers.

Source code in vllm/v1/kv_offload/tiering/manager.py

class TieringOffloadingManager(OffloadingManager):
    """Orchestrates multi-tier KV cache offloading.

    This manager coordinates between a CPU primary tier (with direct GPU access)
    and zero or more secondary tiers (Storage, Network, etc.) to provide
    hierarchical KV cache offloading.

    Key internal state:
      - Minimal state tracking; relies on secondary tiers to report completion
        via get_finished_jobs()
      - Secondary tiers return JobResult objects containing all necessary
        information
      - job_id_counter: monotonically increasing counter for job IDs
    """

    def __init__(
        self,
        primary_tier: CPUPrimaryTierOffloadingManager,
        secondary_tiers: list[SecondaryTierManager] | None = None,
        tier_poll_interval_s: float = DEFAULT_TIER_POLL_INTERVAL_S,
    ):
        """Initialize the TieringOffloadingManager.

        Args:
            primary_tier: The primary tier manager (CPU-based).
            secondary_tiers: List of secondary tier managers (e.g., Storage,
                            Network). Can be None or empty list.
            tier_poll_interval_s: Pause between control-plane rounds for
                tiers that set serves_external_requests. Zero or negative
                disables the thread, leaving those tiers serviced once per
                engine step as before.

        """
        # Guards this manager and every tier it composes, the primary tier
        # included; the primary's own lock is not used underneath it. A real
        # lock because the control-plane thread enters the manager too.
        self._lock = threading.Lock()
        self.primary_tier: CPUPrimaryTierOffloadingManager = primary_tier
        self.secondary_tiers = secondary_tiers or []

        self._job_id_counter: int = 0
        # Job tracking: maps job_id to metadata for all in-flight transfers.
        # TransferJob.is_promotion distinguishes direction:
        #   True:  secondary → primary (promotion)
        #   False: primary → secondary (cascade)
        self._jobs: dict[JobId, JobMetadata] = {}
        primary_view = self.primary_tier.get_kv_memoryview()
        assert primary_view.strides is not None
        self._metrics = TieringMetricsTracker(
            tier_types=[tier.tier_type for tier in self.secondary_tiers],
            num_primary_chunks=self.primary_tier._num_chunks,
            primary_chunk_size=primary_view.strides[0],
        )

        # Pending promotion requests accumulated during lookup() calls; flushed
        # as one batched submit_load() per (tier, request) in on_schedule_end().
        # Outer key: tier index. Inner key: req_context.req_id — the same ReqContext
        # object is reused for all chunk lookups of a given request per engine step.
        self._pending_load_submissions: dict[int, dict[str, PendingPromotion]] = {}

        # Gate for once-per-step execution of _maybe_process_finished_jobs().
        # Reset at the end of each step in on_schedule_end().
        self._processed_jobs_this_step: bool = False

        # Per-request state for prepared GPU->primary stores and finalization.
        # Secondary tiers are finalized only after pending primary stores reach
        # complete_store(), since complete_store() can still submit cascades.
        self._req_state: dict[str, RequestState] = {}

        # Preserve the original tier for this request's cache-hit metrics,
        # even after its KV blocks are promoted into host memory
        # (otherwise a disk or P2P hit would be mislabeled as a host hit).
        self._request_load_sources: dict[str, dict[OffloadKey, CacheHitSource]] = {}

        # Cached ParentManager wrappers for each secondary tier.
        self._tier_parents: dict[SecondaryTierManager, _SecondaryTierFacingParent] = {
            tier: _SecondaryTierFacingParent(self, tier_idx)
            for tier_idx, tier in enumerate(self.secondary_tiers)
        }

        self._tier_index: dict[SecondaryTierManager, int] = {
            tier: i for i, tier in enumerate(self.secondary_tiers)
        }

        # Tiers that serve external requests, and so must also be serviced
        # between engine steps.
        self._external_serving_tiers: list[tuple[int, SecondaryTierManager]] = [
            (i, tier)
            for i, tier in enumerate(self.secondary_tiers)
            if tier.serves_external_requests
        ]
        self._tier_poll_interval_s = tier_poll_interval_s
        self._control_plane_stop = threading.Event()
        self._control_plane_thread: threading.Thread | None = None
        self._start_control_plane()

    @property
    @override
    def lock(self) -> threading.Lock:
        return self._lock

    # ------------------------------------------------------------------
    # Control plane
    # ------------------------------------------------------------------

    def _start_control_plane(self) -> None:
        """Start the control-plane thread, if any tier asked for one."""
        if not self._external_serving_tiers or self._tier_poll_interval_s <= 0:
            return
        self._control_plane_thread = threading.Thread(
            target=self._control_plane_loop,
            name="vllm_offload_control_plane",
            daemon=True,
        )
        self._control_plane_thread.start()
        logger.info(
            "KV offload control-plane thread started for tier(s) %s, interval %.3fs",
            [tier.tier_type for _, tier in self._external_serving_tiers],
            self._tier_poll_interval_s,
        )

    def _stop_control_plane(self) -> None:
        """Stop and join the control-plane thread.

        Must run before any tier teardown: the thread drives tier transports,
        and closing one underneath it can crash the process rather than raise.
        """
        self._control_plane_stop.set()
        thread = self._control_plane_thread
        if thread is None:
            return
        thread.join(timeout=_CONTROL_PLANE_JOIN_TIMEOUT_S)
        if thread.is_alive():
            logger.error(
                "KV offload control-plane thread did not exit within %.1fs",
                _CONTROL_PLANE_JOIN_TIMEOUT_S,
            )
        else:
            self._control_plane_thread = None

    def _control_plane_loop(self) -> None:
        stalled_since: float | None = None
        errors = 0
        while not self._control_plane_stop.is_set():
            if not self._lock.acquire(timeout=_CONTROL_PLANE_LOCK_TIMEOUT_S):
                now = time.monotonic()
                if stalled_since is None:
                    stalled_since = now
                elif now - stalled_since > _CONTROL_PLANE_STALL_WARN_S:
                    logger.warning(
                        "KV offload control-plane thread has not acquired the "
                        "manager lock for %.0fs; peer requests are only being "
                        "served at engine-step boundaries.",
                        now - stalled_since,
                    )
                    stalled_since = None
                continue
            stalled_since = None
            try:
                self.poll_tiers()
                errors = 0
            except Exception:
                # Never let a round kill the thread silently: on_schedule_end()
                # still serves these tiers, so persistent failure degrades to
                # per-step servicing instead of stopping the control plane.
                errors += 1
                logger.exception("KV offload control-plane round failed")
            finally:
                self._lock.release()
            if errors >= _CONTROL_PLANE_MAX_CONSECUTIVE_ERRORS:
                logger.error(
                    "KV offload control-plane thread stopping after %d "
                    "consecutive failures; falling back to per-step servicing.",
                    errors,
                )
                return
            # Yield unconditionally, and only after release(). Python locks are
            # not fair, so a release-then-reacquire loop could starve the
            # scheduler thread; this pause is what bounds its wait to one round.
            self._control_plane_stop.wait(self._tier_poll_interval_s)

    def poll_tiers(self) -> None:
        """Advance the control plane of tiers that cannot wait for a step.

        Polls those tiers for finished jobs and lets them serve whatever their
        counterpart has asked for -- the same two steps on_schedule_end() runs,
        narrowed to the tiers that opted in.

        Excludes _flush_pending_promotions() and _flush_pending_cascades() by
        design: both are per-step batching points, and running them here would
        fragment batches the scheduler thread accumulated during schedule(). A
        promotion that a lookup() from this path initiates therefore waits for
        the next on_schedule_end() to be submitted.

        The caller must hold self.lock for the whole call. serve_external_requests()
        relies on that: it establishes lookup() HITs and then pins them, and the
        two must not be separated by an eviction.

        Serving can look keys up through the parent, and lookup() runs the
        step's once-per-step poll. From here that poll would reach every tier
        off-thread, re-enter the serving tier's get_finished_jobs() from inside
        its own serve, and leave the gate set so the next step skips its poll.
        The round has just polled its tiers itself, so it holds the gate shut
        for its duration and then puts it back untouched.
        """
        gate = self._processed_jobs_this_step
        self._processed_jobs_this_step = True
        try:
            self._process_finished_jobs(self._external_serving_tiers)
            for _, tier in self._external_serving_tiers:
                tier.serve_external_requests(self._tier_parents[tier])
        finally:
            self._processed_jobs_this_step = gate

    @property
    def _transfer_jobs(self) -> dict[JobId, JobMetadata]:
        return self._jobs

    def _next_job_id(self) -> JobId:
        """Generate a unique job ID for async transfer tracking."""
        job_id = self._job_id_counter
        self._job_id_counter += 1
        return job_id

    def _register_job(self, transfer_job: TransferJob, tier_idx: int) -> None:
        job_metadata = JobMetadata(transfer_job, tier_idx)
        self._jobs[transfer_job.job_id] = job_metadata
        self._metrics.on_job_registered(job_metadata)

    def _pop_job(self, job_id: JobId) -> JobMetadata | None:
        return self._jobs.pop(job_id, None)

    def _maybe_process_finished_jobs(self):
        """Poll secondary tiers for completed jobs (at most once per step).

        Guarded by _processed_jobs_this_step: the first call in an engine step
        does the actual polling; subsequent calls are no-ops. The flag is reset
        in on_schedule_end() at the end of each step.
        """
        if self._processed_jobs_this_step:
            return
        self._processed_jobs_this_step = True
        self._process_finished_jobs()

    def _complete_promotion(
        self, job_metadata: JobMetadata, completed_job: JobResult
    ) -> None:
        transfer_job = job_metadata.transfer_job
        successful_keys = completed_job.successful_keys
        failed_keys: Collection[OffloadKey]
        if completed_job.success:
            successful_keys = transfer_job.keys
            failed_keys = ()
        elif successful_keys:
            failed_keys_set = set(transfer_job.keys)
            assert failed_keys_set.issuperset(successful_keys), (
                f"Finished promotion job_id {completed_job.job_id} "
                "reported unknown successful keys"
            )
            failed_keys_set.difference_update(successful_keys)
            failed_keys = failed_keys_set
        else:
            successful_keys = ()
            failed_keys = transfer_job.keys

        load_sources = self._request_load_sources.get(transfer_job.req_context.req_id)
        if load_sources is not None:
            source = self.secondary_tiers[job_metadata.tier_idx].cache_hit_source
            for key in failed_keys:
                if load_sources.get(key) == source:
                    del load_sources[key]

        if successful_keys:
            self.primary_tier.complete_write(
                successful_keys,
                transfer_job.req_context,
                True,
            )
        if failed_keys:
            self.primary_tier.complete_write(
                failed_keys,
                transfer_job.req_context,
                False,
            )

    def _process_finished_jobs(
        self,
        tiers: Sequence[tuple[int, SecondaryTierManager]] | None = None,
    ):
        """Unconditionally poll secondary tiers for completed jobs.

        This method:
        1. Calls get_finished_jobs() on each secondary tier
        2. For completed stores (primary→secondary): calls primary.complete_read()
           to decrement ref_cnt
        3. For completed loads (secondary→primary): calls primary.complete_write()
           to make chunks available

        Args:
            tiers: (index, tier) pairs to poll. Defaults to every secondary
                tier. Pass a subset to poll only the tiers a caller is
                responsible for, so a tier that expects to be polled on the
                scheduler thread is not dragged elsewhere.

        """
        for i, tier in tiers if tiers is not None else enumerate(self.secondary_tiers):
            for completed_job in tier.get_finished_jobs():
                job_id = completed_job.job_id
                job_metadata = self._pop_job(job_id)
                assert job_metadata is not None, (
                    f"Finished job_id {job_id} from tier #{i}"
                    f" ({tier.tier_type}) not in _jobs"
                )
                assert job_metadata.tier_idx == i, (
                    f"Finished job_id {job_id} reported by tier #{i}"
                    f" but belongs to tier #{job_metadata.tier_idx}"
                )
                transfer_job = job_metadata.transfer_job
                self._metrics.on_job_finished(job_metadata, completed_job)

                if transfer_job.is_promotion:
                    # secondary→primary transfer (promotion) completed.
                    # Make chunks available in primary tier.
                    self._complete_promotion(job_metadata, completed_job)
                else:
                    # primary→secondary transfer completed.
                    # Decrement ref_cnt on primary chunks.
                    self.primary_tier.complete_read(
                        transfer_job.keys, transfer_job.req_context
                    )
                    if completed_job.success:
                        self._update_backpressure(tier, job_metadata, completed_job)

    def _should_store_to_tier(
        self, tier: SecondaryTierManager, num_blocks: int
    ) -> bool:
        detector = tier.bp_detector
        if detector is None:
            return True
        return detector.should_store(num_blocks)

    def _update_backpressure(
        self,
        tier: SecondaryTierManager,
        job_metadata: JobMetadata,
        completed_job: JobResult,
    ) -> None:
        detector = tier.bp_detector
        if detector is None:
            return
        was_under_pressure = detector.is_under_pressure()
        tj = job_metadata.transfer_job
        num_bytes = (
            completed_job.transfer_bytes
            if completed_job.transfer_bytes is not None
            else len(tj.keys) * tier.block_size_bytes
        )
        detector.update(tj.submit_time, num_bytes)
        if detector.is_under_pressure() != was_under_pressure:
            tier_idx = self._tier_index[tier]
            logger.info(
                "Tier #%d (%s) back-pressure %s (stats=%s)",
                tier_idx,
                tier.tier_type,
                "activated" if detector.is_under_pressure() else "cleared",
                detector.stats,
            )

    @override
    def lookup(
        self,
        key: OffloadKey,
        req_context: ReqContext,
        *,
        exclude_tier_idx: int | None = None,
    ) -> LookupResult:
        """Check whether a single chunk is offloaded and ready.

        Algorithm:
            1. Process any completed async jobs first.
            2. Query primary tier — short-circuit on hit or in-flight.
            3. On primary miss, query secondary tiers — stop on first
               hit and initiate promotion.

        Args:
            key: Chunk hash to look up.
            req_context: Per-request context.
            exclude_tier_idx: Skip this tier index during the lookup.

        Returns:
            HIT       — chunk is ready in the primary tier.
            HIT_PENDING — chunk found but not yet readable (write
                        in-flight on the primary tier).
            RETRY     — promotion started or a secondary tier is busy.
            MISS      — chunk not found in any tier, or primary is full
                        and cannot accept a promotion.

        """
        # Poll first so a promotion that finished since the last call is
        # already reflected as HIT (not stale HIT_PENDING/MISS) below, and
        # so chunks freed by cascade or promotion completions are evictable
        # in time for a promotion this lookup may initiate.
        self._maybe_process_finished_jobs()

        start_time = time.monotonic()
        primary_hit = self.primary_tier.lookup(key, req_context)
        lookup_duration = time.monotonic() - start_time
        self._metrics.on_lookup(
            req_context,
            key,
            self._metrics.primary_tier_label,
            primary_hit,
            lookup_duration,
        )
        if primary_hit is LookupResult.HIT:
            return LookupResult.HIT
        if primary_hit is LookupResult.HIT_PENDING:
            return LookupResult.HIT_PENDING

        any_retry = False
        for i, tier in enumerate(self.secondary_tiers):
            if i == exclude_tier_idx:
                continue
            if not req_context.load_tier_filter.allows(tier.medium, tier.locality):
                continue
            labelvalues = self._metrics.tier_label(i)
            start_time = time.monotonic()
            result = tier.lookup(key, req_context)
            lookup_duration = time.monotonic() - start_time
            if result is LookupResult.HIT:
                self._metrics.on_lookup(
                    req_context,
                    key,
                    labelvalues,
                    result,
                    lookup_duration,
                )
                promoted = self._initiate_promotion(i, key, req_context)
                return LookupResult.MISS if not promoted else LookupResult.HIT_PENDING
            if result is LookupResult.RETRY:
                any_retry = True
            self._metrics.on_lookup(
                req_context,
                key,
                labelvalues,
                result,
                lookup_duration,
            )

        if any_retry:
            return LookupResult.RETRY
        return LookupResult.MISS

    @override
    def get_load_source(
        self, key: OffloadKey, req_context: ReqContext
    ) -> CacheHitSource:
        load_sources = self._request_load_sources.get(req_context.req_id)
        if load_sources is not None and key in load_sources:
            return load_sources[key]
        return self.primary_tier.get_load_source(key, req_context)

    def _initiate_promotion(
        self,
        tier_idx: int,
        key: OffloadKey,
        req_context: ReqContext,
    ) -> bool:
        """Queue a chunk for promotion from a secondary tier to the primary tier.

        Allocates space in the primary tier immediately (sets ref_cnt=-1 so
        subsequent lookups within the same step see the slot as in-flight),
        then defers the actual submit_load() call to _flush_pending_promotions()
        so all chunks queued during one engine step are submitted as a single
        batched job.

        Args:
            tier_idx: The secondary tier index to promote from
            key: Chunk to promote
            req_context: Per-request context forwarded to primary.prepare_write().

        Returns:
            True if promotion was initiated, False if primary tier is full.

        """
        # Allocate space in primary tier for promoted chunk.
        # Must happen immediately so primary.lookup() returns None (in-flight)
        # for this key on any subsequent lookup() call within the same step,
        # preventing duplicate promotion attempts.
        primary_write_result = self.primary_tier.prepare_write([key], req_context)

        if primary_write_result is None:
            # Primary tier is full; caller should treat the chunk as unavailable
            # rather than retrying indefinitely.
            self._metrics.on_promotion_allocation_failure()
            return False

        store_spec = primary_write_result.store_spec
        assert isinstance(store_spec, CPULoadStoreSpec)
        load_sources = self._request_load_sources.setdefault(req_context.req_id, {})
        source = self.secondary_tiers[tier_idx].cache_hit_source
        for promoted_key in primary_write_result.keys_to_store:
            load_sources[promoted_key] = source
        # Defer submit_load to on_schedule_end(). Group by (tier, request) so
        # each request's chunks are submitted as one batched job per tier.
        tier_pending = self._pending_load_submissions.setdefault(tier_idx, {})
        ctx_id = req_context.req_id
        if ctx_id not in tier_pending:
            tier_pending[ctx_id] = PendingPromotion(
                keys=[], chunk_ids=[], req_context=req_context
            )
        entry = tier_pending[ctx_id]
        entry.keys.extend(primary_write_result.keys_to_store)
        entry.chunk_ids.extend(store_spec.chunk_ids)
        return True

    def _flush_pending_promotions(self) -> None:
        """Submit one batched submit_load() per (tier, request).

        Called from on_schedule_end() at the end of each scheduler step,
        flushing all promotion requests deferred during lookup().
        """
        if not self._pending_load_submissions:
            return

        for tier_idx, pending_by_ctx in self._pending_load_submissions.items():
            tier = self.secondary_tiers[tier_idx]
            for entry in pending_by_ctx.values():
                job_id = self._next_job_id()
                job_metadata = TransferJob(
                    job_id=job_id,
                    keys=entry.keys,
                    chunk_ids=np.array(entry.chunk_ids, dtype=np.int32),
                    is_promotion=True,
                    req_context=entry.req_context,
                )
                self._register_job(job_metadata, tier_idx)
                tier.submit_load(job_metadata)

        self._pending_load_submissions.clear()

    @override
    def prepare_load(
        self, keys: Collection[OffloadKey], req_context: ReqContext
    ) -> LoadStoreSpec:
        """Prepare chunks to be loaded from primary tier to GPU.

        Callers only pass keys already confirmed HIT by lookup() earlier this
        step.

        This increments ref_cnt on the chunks in the primary tier, protecting
        them from eviction during the transfer.

        Args:
            keys: Chunks to prepare for loading.
            req_context: Per-request context.

        Returns:
            LoadStoreSpec for reading from primary tier.

        """
        return self.primary_tier.prepare_load(keys, req_context)

    @override
    def touch(self, keys: Collection[OffloadKey], req_context: ReqContext):
        """Mark chunks as recently used in all tiers.

        Args:
            keys: Chunks to mark as recently used.
            req_context: Per-request context.

        """
        self.primary_tier.touch(keys, req_context)
        for tier in self.secondary_tiers:
            tier.touch(keys, req_context)

    @override
    def complete_load(self, keys: Collection[OffloadKey], req_context: ReqContext):
        """Mark chunks as done loading from primary tier to GPU.

        This decrements ref_cnt on the chunks in the primary tier, allowing
        them to be evicted again.

        Args:
            keys: Chunks that finished loading.
            req_context: Per-request context.

        """
        self.primary_tier.complete_load(keys, req_context)

    @override
    def prepare_store(
        self, keys: Collection[OffloadKey], req_context: ReqContext
    ) -> PrepareStoreOutput | None:
        """Prepare chunks to be stored from GPU to primary tier.

        CRITICAL: This method polls for finished jobs FIRST to ensure that any
        completed async transfers have their ref_cnt decremented before the
        primary tier makes eviction decisions.

        For request-level tiers, chunks already present in the primary tier
        are immediately cascaded via submit_store().

        Args:
            keys: Chunks to prepare for storing.
            req_context: Per-request context.

        Returns:
            PrepareStoreOutput describing where to store chunks and what was
            evicted, or None if store cannot proceed.

        """
        # Step 1: Poll for completed async jobs FIRST
        # _process_finished_jobs() handles two kinds of completions here:
        #  - Cascade completions (store to a secondary tier, either a local
        #    cascade or a store job created for a remote requester via
        #    create_store_job()): decrements ref_cnt on the primary chunks
        #    that were read, making them evictable again once ref_cnt hits 0.
        #  - Promotion completions (secondary->primary loads): sets a
        #    not-yet-ready chunk's ref_cnt from -1 to 0 via complete_write(),
        #    making it evictable for the first time.
        # Both must be accounted for before the eviction decision below.
        # Unconditional, not _maybe_process_finished_jobs(): eviction must see
        # the freshest completions, and on_schedule_end now runs at the very end
        # of the step, so the once-per-step gate would otherwise still be set by
        # this step's first lookup() and skip the poll.
        self._processed_jobs_this_step = True
        self._process_finished_jobs()

        # Step 2: Store to primary tier (new chunks only).
        # Cascading of these newly-stored chunks to ALL secondary tiers
        # happens later in complete_store(), after the GPU→Primary transfer
        # completes.
        primary_result = self.primary_tier.prepare_store(keys, req_context)

        if primary_result is None:
            return None

        if primary_result.keys_to_store:
            state = self._req_state[req_context.req_id]
            state.pending_primary_stores += 1

        # Step 3: For request-level tiers, cascade chunks already in primary
        request_level_tiers = self._req_state[req_context.req_id].request_level_tiers
        if request_level_tiers:
            keys_to_store_set = set(primary_result.keys_to_store)
            keys_already_in_primary = tuple(
                k for k in keys if k not in keys_to_store_set
            )
            if keys_already_in_primary:
                self._cascade_existing_chunks_to_request_level_tiers(
                    keys_already_in_primary, req_context, request_level_tiers
                )

        return primary_result

    def _cascade_existing_chunks_to_request_level_tiers(
        self,
        keys: Sequence[OffloadKey],
        req_context: ReqContext,
        request_level_tiers: set[int],
    ) -> None:
        """For tiers that requested request-level policy, submit_store() for
        chunks that are already present in the primary tier.

        A key whose primary write is still in flight (HIT_PENDING) cannot be
        dropped: prepare_store already excluded it as present, and the
        scheduler advances past its chunk, so no path offers it again. Park it
        instead. MISS keys are dropped, since nothing is there to read.

        The primary tier resolves every key it holds, so RETRY cannot reach
        here. Parking on it would have no guarantee of ever draining, which is
        what makes parking HIT_PENDING safe, so it is rejected rather than
        guessed at.
        """
        state = self._req_state[req_context.req_id]
        ready_keys = []
        for key in keys:
            result = self.primary_tier.lookup(key, req_context)
            if result is LookupResult.HIT:
                ready_keys.append(key)
            elif result is LookupResult.HIT_PENDING:
                state.pending_cascade_keys.append(key)
            else:
                assert result is LookupResult.MISS, (
                    f"primary tier returned {result} for a cascade key"
                )
        if not ready_keys:
            return

        for tier_idx in request_level_tiers:
            tier = self.secondary_tiers[tier_idx]
            if not self._should_store_to_tier(tier, len(ready_keys)):
                continue
            job_metadata = self.create_store_job(ready_keys, req_context, tier_idx)
            tier.submit_store(job_metadata)

    def _flush_pending_cascades(self) -> None:
        """Retry request-level cascades parked on an in-flight primary write.

        A parked key always resolves, to HIT or to MISS, so the set drains and
        a request cannot be held from finalization forever.
        """
        for req_id, state in list(self._req_state.items()):
            if not state.pending_cascade_keys:
                continue
            assert state.request_level_tiers
            keys, state.pending_cascade_keys = state.pending_cascade_keys, []
            self._cascade_existing_chunks_to_request_level_tiers(
                keys, state.req_context, state.request_level_tiers
            )
            self._maybe_finalize_request(req_id)

    @override
    def complete_store(
        self,
        keys: Collection[OffloadKey],
        req_context: ReqContext,
        success: bool = True,
    ) -> None:
        """Mark chunks as done storing from GPU to primary tier.

        This is where secondary tier cascading happens — after chunks are
        confirmed to be in the primary tier, they are cascaded to ALL
        secondary tiers.

        For each secondary tier:
        1. Call primary.prepare_read() to get LoadStoreSpec AND increment
           ref_cnt (protecting chunks during async transfer)
        2. Call tier.submit_store() to start async transfer: primary→secondary
        3. Track the job in _store_jobs dictionary

        Args:
            keys: Chunks that finished storing.
            success: Whether the GPU→primary transfer succeeded.
            req_context: Per-request context forwarded to primary.prepare_read().

        """
        # Step 1: Complete store in primary tier (makes chunks loadable)
        self.primary_tier.complete_store(keys, req_context, success)

        if success:
            # Step 2: Cascade to ALL secondary tiers
            # For each secondary tier, call primary.prepare_read() to get the
            # LoadStoreSpec AND to increment ref_cnt (protecting chunks from
            # eviction during the async transfer). One prepare_read() call per
            # secondary tier.
            for tier_idx, tier in enumerate(self.secondary_tiers):
                if not self._should_store_to_tier(tier, len(keys)):
                    continue
                job_metadata = self.create_store_job(keys, req_context, tier_idx)
                tier.submit_store(job_metadata)

        # Note: The async transfers are now in flight. Their completion is
        # tracked via get_finished_jobs() / _maybe_process_finished_jobs().
        req_id = req_context.req_id
        state = self._req_state[req_id]
        assert state.pending_primary_stores > 0
        state.pending_primary_stores -= 1
        self._maybe_finalize_request(req_id)

    def create_store_job(
        self,
        keys: Collection[OffloadKey],
        req_context: ReqContext,
        tier_idx: int = 0,
    ) -> TransferJob:
        """Pin chunks in the primary tier and create a tracked store job.

        Calls prepare_read() to increment ref_cnt (protecting chunks
        from eviction during the async transfer), allocates a job ID,
        and registers the job in _jobs.

        The caller is responsible for the actual data transfer and
        reporting completion via get_finished_jobs().
        """
        primary_chunks_spec = self.primary_tier.prepare_read(keys, req_context)
        assert isinstance(primary_chunks_spec, CPULoadStoreSpec)
        job_id = self._next_job_id()
        job_metadata = TransferJob(
            job_id=job_id,
            keys=keys,
            chunk_ids=primary_chunks_spec.chunk_ids,
            is_promotion=False,
            req_context=req_context,
        )
        self._register_job(job_metadata, tier_idx)
        return job_metadata

    @override
    def on_new_request(
        self,
        req_context: ReqContext,
        *,
        exclude_tier_idx: int | None = None,
    ) -> RequestOffloadingContext:
        """Query each secondary tier for its offload policy preference.

        Returns REQUEST_LEVEL if ANY secondary tier wants request-level.
        Only stores REQUEST_LEVEL tier decisions for use in prepare_store.
        """
        state = RequestState(req_context=req_context)
        self._metrics.on_new_request(req_context)
        for tier_idx, tier in enumerate(self.secondary_tiers):
            if tier_idx == exclude_tier_idx:
                continue
            tier_ctx = tier.on_new_request(req_context)
            if tier_ctx.policy == OffloadPolicy.REQUEST_LEVEL:
                if state.request_level_tiers is None:
                    state.request_level_tiers = set()
                state.request_level_tiers.add(tier_idx)
        self._req_state[req_context.req_id] = state

        policy = (
            OffloadPolicy.REQUEST_LEVEL
            if state.request_level_tiers
            else OffloadPolicy.CHUNK_LEVEL
        )
        return RequestOffloadingContext(policy=policy)

    @override
    def on_request_finished(
        self,
        req_context: ReqContext,
        *,
        exclude_tier_idx: int | None = None,
    ) -> None:
        self.primary_tier.on_request_finished(req_context)
        state = self._req_state[req_context.req_id]
        state.is_finished = True
        self._maybe_finalize_request(req_context.req_id, exclude_tier_idx)

    def _maybe_finalize_request(
        self,
        req_id: str,
        exclude_tier_idx: int | None = None,
    ) -> None:
        """Finalize secondary tiers once no more cascades can be submitted.

        Their finalization is delayed until pending GPU->primary stores
        finish, since those callbacks may still submit secondary stores.
        """
        state = self._req_state[req_id]
        if not state.is_finished:
            return
        if state.pending_primary_stores != 0:
            return
        if state.pending_cascade_keys:
            return

        for tier_idx, tier in enumerate(self.secondary_tiers):
            if tier_idx == exclude_tier_idx:
                continue
            tier.on_request_finished(state.req_context)
        self._metrics.on_request_finished(state.req_context)
        self._request_load_sources.pop(req_id, None)
        del self._req_state[req_id]

    @override
    def on_schedule_end(self, context: ScheduleEndContext) -> None:
        """End-of-schedule hook: process finished jobs, flush deferred
        promotions, and reset the per-step gate.

        Called once per scheduler step, as the last manager call of the step,
        from OffloadingConnectorScheduler.build_connector_meta().
        """
        # Catch-all poll: guarantees jobs are processed even on steps where
        # lookup()/prepare_store() were never called (e.g. no requests
        # scheduled but a tier still has_pending_work()).
        self._maybe_process_finished_jobs()

        for tier in self.secondary_tiers:
            tier.serve_external_requests(self._tier_parents[tier])

        # Reset the per-step gate AFTER serve_external_requests so that
        # lookup() calls within it skip redundant _process_finished_jobs().
        self._processed_jobs_this_step = False

        self._flush_pending_promotions()
        # Keys parked by this step's prepare_store cannot have completed yet, so
        # they are simply re-parked here and retried on the next step. That is
        # safe: _maybe_finalize_request() holds off while pending_cascade_keys is
        # non-empty, so nothing finalizes early and the list still drains.
        self._flush_pending_cascades()
        for tier in self.secondary_tiers:
            tier.on_schedule_end(context)

        for req_id in context.new_req_ids:
            state = self._req_state.get(req_id)
            if state is None:
                continue
            self._metrics.on_request_allocated(state.req_context)

    @override
    def has_pending_work(self) -> bool:
        # In-flight primary<->secondary transfers (pending promotions are
        # translated to transfer jobs in on_schedule_end), plus any work the
        # secondary tiers themselves still have outstanding.
        return (
            bool(self._jobs)
            or any(state.pending_cascade_keys for state in self._req_state.values())
            or any(tier.has_pending_work() for tier in self.secondary_tiers)
        )

    @override
    def take_events(self) -> Iterable[OffloadingEvent]:
        """Yield events owned by the primary and secondary tiers.

        Yields:
            New OffloadingEvents collected by each tier since the last call.

        """
        yield from self.primary_tier.take_events()
        for tier in self.secondary_tiers:
            yield from tier.take_events()

    @override
    def reset_cache(self) -> None:
        """Reset transfer bookkeeping and primary-tier cache.

        Called during sleep, weight update, or resume. Each secondary tier
        drains its in-flight transfers via drain_jobs() so no tier I/O is
        touching primary memory before the primary tier is reset. A stuck
        tier will block here visibly — preferable to silent corruption
        from reusing primary slots while a transfer is mid-copy.

        Secondary tiers are intentionally not reset: persistent stores
        (FS, network) keep their data across resets. Active request state is
        retained so those requests can continue after the reset; finished
        requests are finalized and removed.
        """
        for tier in self.secondary_tiers:
            tier.drain_jobs()
        # All tier I/O has stopped; consume their completion notifications
        # so manager bookkeeping is consistent before the primary reset.
        self._process_finished_jobs()
        assert not self._jobs

        # Deferred promotion submissions reserve primary slots that the
        # reset below invalidates; their submit_load() has not yet been
        # called so no tier I/O is touching that memory.
        self._pending_load_submissions.clear()
        self._request_load_sources.clear()
        self._metrics.assert_idle()

        finished_req_ids = []
        for req_id, state in self._req_state.items():
            state.pending_primary_stores = 0
            state.pending_cascade_keys.clear()
            if not state.is_finished:
                continue
            for tier in self.secondary_tiers:
                tier.on_request_finished(state.req_context)
            self._metrics.on_request_finished(state.req_context)
            finished_req_ids.append(req_id)

        self.primary_tier.reset_cache()

        for req_id in finished_req_ids:
            del self._req_state[req_id]
        self._processed_jobs_this_step = False

        for tier in self.secondary_tiers:
            if tier.bp_detector is not None:
                tier.bp_detector.reset()

    @override
    def get_stats(self) -> OffloadingConnectorStats | None:
        stats = self.primary_tier.get_stats()

        if stats is not None and stats.is_empty():
            stats = None

        self._metrics.record_backpressure(self.secondary_tiers)
        metrics_stats = self._metrics.take_stats()
        if metrics_stats is not None:
            if stats is None:
                stats = metrics_stats
            else:
                stats.aggregate(metrics_stats)

        for tier in self.secondary_tiers:
            tier_stats = tier.get_stats()
            if tier_stats is None or tier_stats.is_empty():
                continue
            if stats is None:
                stats = tier_stats
            else:
                stats.aggregate(tier_stats)

        return stats

    @override
    def shutdown(self) -> None:
        """Shut down secondary tiers before releasing primary resources.

        Stops the control-plane thread first: it drives tier transports, and
        tearing one down underneath it (closing a socket, destroying a ZMQ
        context) can take the process down rather than raise. If the thread
        will not exit, hold the manager lock across tier teardown so it cannot
        be mid-round, and skip teardown entirely if even that is unavailable --
        a lingering daemon thread in an exiting process beats a crash.

        Every secondary tier is given a shutdown attempt. If any shutdown
        fails, preserve the primary mmap because a failed tier may still use it.
        """
        self._stop_control_plane()
        if self._control_plane_thread is not None:
            if not self._lock.acquire(timeout=_CONTROL_PLANE_JOIN_TIMEOUT_S):
                logger.error(
                    "KV offload control-plane thread is still running and the "
                    "manager lock is unavailable; skipping tier shutdown."
                )
                return
            try:
                self._shutdown_tiers()
            finally:
                self._lock.release()
            return
        self._shutdown_tiers()

    def _shutdown_tiers(self) -> None:
        shutdown_error: Exception | None = None
        for tier_idx, tier in enumerate(self.secondary_tiers):
            try:
                tier.shutdown()
            except Exception as exc:
                shutdown_error = exc
                logger.exception(
                    "Failed to shut down secondary tier #%d "
                    "(tier_type=%s, impl_class=%s)",
                    tier_idx,
                    tier.tier_type,
                    type(tier).__name__,
                )

        if shutdown_error is not None:
            raise shutdown_error

        self.primary_tier.shutdown()

__init__(primary_tier, secondary_tiers=None, tier_poll_interval_s=DEFAULT_TIER_POLL_INTERVAL_S)

Initialize the TieringOffloadingManager.

Parameters:

• primary_tier

  (CPUPrimaryTierOffloadingManager) –

  The primary tier manager (CPU-based).

• secondary_tiers

  (list[SecondaryTierManager] | None, default: None ) –

  List of secondary tier managers (e.g., Storage, Network). Can be None or empty list.

• tier_poll_interval_s

  (float, default: DEFAULT_TIER_POLL_INTERVAL_S ) –

  Pause between control-plane rounds for tiers that set serves_external_requests. Zero or negative disables the thread, leaving those tiers serviced once per engine step as before.

Source code in vllm/v1/kv_offload/tiering/manager.py

def __init__(
    self,
    primary_tier: CPUPrimaryTierOffloadingManager,
    secondary_tiers: list[SecondaryTierManager] | None = None,
    tier_poll_interval_s: float = DEFAULT_TIER_POLL_INTERVAL_S,
):
    """Initialize the TieringOffloadingManager.

    Args:
        primary_tier: The primary tier manager (CPU-based).
        secondary_tiers: List of secondary tier managers (e.g., Storage,
                        Network). Can be None or empty list.
        tier_poll_interval_s: Pause between control-plane rounds for
            tiers that set serves_external_requests. Zero or negative
            disables the thread, leaving those tiers serviced once per
            engine step as before.

    """
    # Guards this manager and every tier it composes, the primary tier
    # included; the primary's own lock is not used underneath it. A real
    # lock because the control-plane thread enters the manager too.
    self._lock = threading.Lock()
    self.primary_tier: CPUPrimaryTierOffloadingManager = primary_tier
    self.secondary_tiers = secondary_tiers or []

    self._job_id_counter: int = 0
    # Job tracking: maps job_id to metadata for all in-flight transfers.
    # TransferJob.is_promotion distinguishes direction:
    #   True:  secondary → primary (promotion)
    #   False: primary → secondary (cascade)
    self._jobs: dict[JobId, JobMetadata] = {}
    primary_view = self.primary_tier.get_kv_memoryview()
    assert primary_view.strides is not None
    self._metrics = TieringMetricsTracker(
        tier_types=[tier.tier_type for tier in self.secondary_tiers],
        num_primary_chunks=self.primary_tier._num_chunks,
        primary_chunk_size=primary_view.strides[0],
    )

    # Pending promotion requests accumulated during lookup() calls; flushed
    # as one batched submit_load() per (tier, request) in on_schedule_end().
    # Outer key: tier index. Inner key: req_context.req_id — the same ReqContext
    # object is reused for all chunk lookups of a given request per engine step.
    self._pending_load_submissions: dict[int, dict[str, PendingPromotion]] = {}

    # Gate for once-per-step execution of _maybe_process_finished_jobs().
    # Reset at the end of each step in on_schedule_end().
    self._processed_jobs_this_step: bool = False

    # Per-request state for prepared GPU->primary stores and finalization.
    # Secondary tiers are finalized only after pending primary stores reach
    # complete_store(), since complete_store() can still submit cascades.
    self._req_state: dict[str, RequestState] = {}

    # Preserve the original tier for this request's cache-hit metrics,
    # even after its KV blocks are promoted into host memory
    # (otherwise a disk or P2P hit would be mislabeled as a host hit).
    self._request_load_sources: dict[str, dict[OffloadKey, CacheHitSource]] = {}

    # Cached ParentManager wrappers for each secondary tier.
    self._tier_parents: dict[SecondaryTierManager, _SecondaryTierFacingParent] = {
        tier: _SecondaryTierFacingParent(self, tier_idx)
        for tier_idx, tier in enumerate(self.secondary_tiers)
    }

    self._tier_index: dict[SecondaryTierManager, int] = {
        tier: i for i, tier in enumerate(self.secondary_tiers)
    }

    # Tiers that serve external requests, and so must also be serviced
    # between engine steps.
    self._external_serving_tiers: list[tuple[int, SecondaryTierManager]] = [
        (i, tier)
        for i, tier in enumerate(self.secondary_tiers)
        if tier.serves_external_requests
    ]
    self._tier_poll_interval_s = tier_poll_interval_s
    self._control_plane_stop = threading.Event()
    self._control_plane_thread: threading.Thread | None = None
    self._start_control_plane()

_cascade_existing_chunks_to_request_level_tiers(keys, req_context, request_level_tiers)

For tiers that requested request-level policy, submit_store() for chunks that are already present in the primary tier.

A key whose primary write is still in flight (HIT_PENDING) cannot be dropped: prepare_store already excluded it as present, and the scheduler advances past its chunk, so no path offers it again. Park it instead. MISS keys are dropped, since nothing is there to read.

The primary tier resolves every key it holds, so RETRY cannot reach here. Parking on it would have no guarantee of ever draining, which is what makes parking HIT_PENDING safe, so it is rejected rather than guessed at.

Source code in vllm/v1/kv_offload/tiering/manager.py

def _cascade_existing_chunks_to_request_level_tiers(
    self,
    keys: Sequence[OffloadKey],
    req_context: ReqContext,
    request_level_tiers: set[int],
) -> None:
    """For tiers that requested request-level policy, submit_store() for
    chunks that are already present in the primary tier.

    A key whose primary write is still in flight (HIT_PENDING) cannot be
    dropped: prepare_store already excluded it as present, and the
    scheduler advances past its chunk, so no path offers it again. Park it
    instead. MISS keys are dropped, since nothing is there to read.

    The primary tier resolves every key it holds, so RETRY cannot reach
    here. Parking on it would have no guarantee of ever draining, which is
    what makes parking HIT_PENDING safe, so it is rejected rather than
    guessed at.
    """
    state = self._req_state[req_context.req_id]
    ready_keys = []
    for key in keys:
        result = self.primary_tier.lookup(key, req_context)
        if result is LookupResult.HIT:
            ready_keys.append(key)
        elif result is LookupResult.HIT_PENDING:
            state.pending_cascade_keys.append(key)
        else:
            assert result is LookupResult.MISS, (
                f"primary tier returned {result} for a cascade key"
            )
    if not ready_keys:
        return

    for tier_idx in request_level_tiers:
        tier = self.secondary_tiers[tier_idx]
        if not self._should_store_to_tier(tier, len(ready_keys)):
            continue
        job_metadata = self.create_store_job(ready_keys, req_context, tier_idx)
        tier.submit_store(job_metadata)

_flush_pending_cascades()

Retry request-level cascades parked on an in-flight primary write.

A parked key always resolves, to HIT or to MISS, so the set drains and a request cannot be held from finalization forever.

Source code in vllm/v1/kv_offload/tiering/manager.py

def _flush_pending_cascades(self) -> None:
    """Retry request-level cascades parked on an in-flight primary write.

    A parked key always resolves, to HIT or to MISS, so the set drains and
    a request cannot be held from finalization forever.
    """
    for req_id, state in list(self._req_state.items()):
        if not state.pending_cascade_keys:
            continue
        assert state.request_level_tiers
        keys, state.pending_cascade_keys = state.pending_cascade_keys, []
        self._cascade_existing_chunks_to_request_level_tiers(
            keys, state.req_context, state.request_level_tiers
        )
        self._maybe_finalize_request(req_id)

_flush_pending_promotions()

Submit one batched submit_load() per (tier, request).

Called from on_schedule_end() at the end of each scheduler step, flushing all promotion requests deferred during lookup().

Source code in vllm/v1/kv_offload/tiering/manager.py

def _flush_pending_promotions(self) -> None:
    """Submit one batched submit_load() per (tier, request).

    Called from on_schedule_end() at the end of each scheduler step,
    flushing all promotion requests deferred during lookup().
    """
    if not self._pending_load_submissions:
        return

    for tier_idx, pending_by_ctx in self._pending_load_submissions.items():
        tier = self.secondary_tiers[tier_idx]
        for entry in pending_by_ctx.values():
            job_id = self._next_job_id()
            job_metadata = TransferJob(
                job_id=job_id,
                keys=entry.keys,
                chunk_ids=np.array(entry.chunk_ids, dtype=np.int32),
                is_promotion=True,
                req_context=entry.req_context,
            )
            self._register_job(job_metadata, tier_idx)
            tier.submit_load(job_metadata)

    self._pending_load_submissions.clear()

_initiate_promotion(tier_idx, key, req_context)

Queue a chunk for promotion from a secondary tier to the primary tier.

Allocates space in the primary tier immediately (sets ref_cnt=-1 so subsequent lookups within the same step see the slot as in-flight), then defers the actual submit_load() call to _flush_pending_promotions() so all chunks queued during one engine step are submitted as a single batched job.

Parameters:

• tier_idx

  (int) –

  The secondary tier index to promote from

• key

  (OffloadKey) –

  Chunk to promote

• req_context

  (ReqContext) –

  Per-request context forwarded to primary.prepare_write().

Returns:

• bool –

  True if promotion was initiated, False if primary tier is full.

Source code in vllm/v1/kv_offload/tiering/manager.py

def _initiate_promotion(
    self,
    tier_idx: int,
    key: OffloadKey,
    req_context: ReqContext,
) -> bool:
    """Queue a chunk for promotion from a secondary tier to the primary tier.

    Allocates space in the primary tier immediately (sets ref_cnt=-1 so
    subsequent lookups within the same step see the slot as in-flight),
    then defers the actual submit_load() call to _flush_pending_promotions()
    so all chunks queued during one engine step are submitted as a single
    batched job.

    Args:
        tier_idx: The secondary tier index to promote from
        key: Chunk to promote
        req_context: Per-request context forwarded to primary.prepare_write().

    Returns:
        True if promotion was initiated, False if primary tier is full.

    """
    # Allocate space in primary tier for promoted chunk.
    # Must happen immediately so primary.lookup() returns None (in-flight)
    # for this key on any subsequent lookup() call within the same step,
    # preventing duplicate promotion attempts.
    primary_write_result = self.primary_tier.prepare_write([key], req_context)

    if primary_write_result is None:
        # Primary tier is full; caller should treat the chunk as unavailable
        # rather than retrying indefinitely.
        self._metrics.on_promotion_allocation_failure()
        return False

    store_spec = primary_write_result.store_spec
    assert isinstance(store_spec, CPULoadStoreSpec)
    load_sources = self._request_load_sources.setdefault(req_context.req_id, {})
    source = self.secondary_tiers[tier_idx].cache_hit_source
    for promoted_key in primary_write_result.keys_to_store:
        load_sources[promoted_key] = source
    # Defer submit_load to on_schedule_end(). Group by (tier, request) so
    # each request's chunks are submitted as one batched job per tier.
    tier_pending = self._pending_load_submissions.setdefault(tier_idx, {})
    ctx_id = req_context.req_id
    if ctx_id not in tier_pending:
        tier_pending[ctx_id] = PendingPromotion(
            keys=[], chunk_ids=[], req_context=req_context
        )
    entry = tier_pending[ctx_id]
    entry.keys.extend(primary_write_result.keys_to_store)
    entry.chunk_ids.extend(store_spec.chunk_ids)
    return True

_maybe_finalize_request(req_id, exclude_tier_idx=None)

Finalize secondary tiers once no more cascades can be submitted.

Their finalization is delayed until pending GPU->primary stores finish, since those callbacks may still submit secondary stores.

Source code in vllm/v1/kv_offload/tiering/manager.py

def _maybe_finalize_request(
    self,
    req_id: str,
    exclude_tier_idx: int | None = None,
) -> None:
    """Finalize secondary tiers once no more cascades can be submitted.

    Their finalization is delayed until pending GPU->primary stores
    finish, since those callbacks may still submit secondary stores.
    """
    state = self._req_state[req_id]
    if not state.is_finished:
        return
    if state.pending_primary_stores != 0:
        return
    if state.pending_cascade_keys:
        return

    for tier_idx, tier in enumerate(self.secondary_tiers):
        if tier_idx == exclude_tier_idx:
            continue
        tier.on_request_finished(state.req_context)
    self._metrics.on_request_finished(state.req_context)
    self._request_load_sources.pop(req_id, None)
    del self._req_state[req_id]

_maybe_process_finished_jobs()

Poll secondary tiers for completed jobs (at most once per step).

Guarded by _processed_jobs_this_step: the first call in an engine step does the actual polling; subsequent calls are no-ops. The flag is reset in on_schedule_end() at the end of each step.

Source code in vllm/v1/kv_offload/tiering/manager.py

def _maybe_process_finished_jobs(self):
    """Poll secondary tiers for completed jobs (at most once per step).

    Guarded by _processed_jobs_this_step: the first call in an engine step
    does the actual polling; subsequent calls are no-ops. The flag is reset
    in on_schedule_end() at the end of each step.
    """
    if self._processed_jobs_this_step:
        return
    self._processed_jobs_this_step = True
    self._process_finished_jobs()

_next_job_id()

Generate a unique job ID for async transfer tracking.

Source code in vllm/v1/kv_offload/tiering/manager.py

def _next_job_id(self) -> JobId:
    """Generate a unique job ID for async transfer tracking."""
    job_id = self._job_id_counter
    self._job_id_counter += 1
    return job_id

_process_finished_jobs(tiers=None)

Unconditionally poll secondary tiers for completed jobs.

This method: 1. Calls get_finished_jobs() on each secondary tier 2. For completed stores (primary→secondary): calls primary.complete_read() to decrement ref_cnt 3. For completed loads (secondary→primary): calls primary.complete_write() to make chunks available

Parameters:

• tiers

  (Sequence[tuple[int, SecondaryTierManager]] | None, default: None ) –

  (index, tier) pairs to poll. Defaults to every secondary tier. Pass a subset to poll only the tiers a caller is responsible for, so a tier that expects to be polled on the scheduler thread is not dragged elsewhere.

Source code in vllm/v1/kv_offload/tiering/manager.py

def _process_finished_jobs(
    self,
    tiers: Sequence[tuple[int, SecondaryTierManager]] | None = None,
):
    """Unconditionally poll secondary tiers for completed jobs.

    This method:
    1. Calls get_finished_jobs() on each secondary tier
    2. For completed stores (primary→secondary): calls primary.complete_read()
       to decrement ref_cnt
    3. For completed loads (secondary→primary): calls primary.complete_write()
       to make chunks available

    Args:
        tiers: (index, tier) pairs to poll. Defaults to every secondary
            tier. Pass a subset to poll only the tiers a caller is
            responsible for, so a tier that expects to be polled on the
            scheduler thread is not dragged elsewhere.

    """
    for i, tier in tiers if tiers is not None else enumerate(self.secondary_tiers):
        for completed_job in tier.get_finished_jobs():
            job_id = completed_job.job_id
            job_metadata = self._pop_job(job_id)
            assert job_metadata is not None, (
                f"Finished job_id {job_id} from tier #{i}"
                f" ({tier.tier_type}) not in _jobs"
            )
            assert job_metadata.tier_idx == i, (
                f"Finished job_id {job_id} reported by tier #{i}"
                f" but belongs to tier #{job_metadata.tier_idx}"
            )
            transfer_job = job_metadata.transfer_job
            self._metrics.on_job_finished(job_metadata, completed_job)

            if transfer_job.is_promotion:
                # secondary→primary transfer (promotion) completed.
                # Make chunks available in primary tier.
                self._complete_promotion(job_metadata, completed_job)
            else:
                # primary→secondary transfer completed.
                # Decrement ref_cnt on primary chunks.
                self.primary_tier.complete_read(
                    transfer_job.keys, transfer_job.req_context
                )
                if completed_job.success:
                    self._update_backpressure(tier, job_metadata, completed_job)

_start_control_plane()

Start the control-plane thread, if any tier asked for one.

Source code in vllm/v1/kv_offload/tiering/manager.py

def _start_control_plane(self) -> None:
    """Start the control-plane thread, if any tier asked for one."""
    if not self._external_serving_tiers or self._tier_poll_interval_s <= 0:
        return
    self._control_plane_thread = threading.Thread(
        target=self._control_plane_loop,
        name="vllm_offload_control_plane",
        daemon=True,
    )
    self._control_plane_thread.start()
    logger.info(
        "KV offload control-plane thread started for tier(s) %s, interval %.3fs",
        [tier.tier_type for _, tier in self._external_serving_tiers],
        self._tier_poll_interval_s,
    )

_stop_control_plane()

Stop and join the control-plane thread.

Must run before any tier teardown: the thread drives tier transports, and closing one underneath it can crash the process rather than raise.

Source code in vllm/v1/kv_offload/tiering/manager.py

def _stop_control_plane(self) -> None:
    """Stop and join the control-plane thread.

    Must run before any tier teardown: the thread drives tier transports,
    and closing one underneath it can crash the process rather than raise.
    """
    self._control_plane_stop.set()
    thread = self._control_plane_thread
    if thread is None:
        return
    thread.join(timeout=_CONTROL_PLANE_JOIN_TIMEOUT_S)
    if thread.is_alive():
        logger.error(
            "KV offload control-plane thread did not exit within %.1fs",
            _CONTROL_PLANE_JOIN_TIMEOUT_S,
        )
    else:
        self._control_plane_thread = None

complete_load(keys, req_context)

Mark chunks as done loading from primary tier to GPU.

This decrements ref_cnt on the chunks in the primary tier, allowing them to be evicted again.

Parameters:

• keys

  (Collection[OffloadKey]) –

  Chunks that finished loading.

• req_context

  (ReqContext) –

  Per-request context.

Source code in vllm/v1/kv_offload/tiering/manager.py

@override
def complete_load(self, keys: Collection[OffloadKey], req_context: ReqContext):
    """Mark chunks as done loading from primary tier to GPU.

    This decrements ref_cnt on the chunks in the primary tier, allowing
    them to be evicted again.

    Args:
        keys: Chunks that finished loading.
        req_context: Per-request context.

    """
    self.primary_tier.complete_load(keys, req_context)

complete_store(keys, req_context, success=True)

Mark chunks as done storing from GPU to primary tier.

This is where secondary tier cascading happens — after chunks are confirmed to be in the primary tier, they are cascaded to ALL secondary tiers.

For each secondary tier: 1. Call primary.prepare_read() to get LoadStoreSpec AND increment ref_cnt (protecting chunks during async transfer) 2. Call tier.submit_store() to start async transfer: primary→secondary 3. Track the job in _store_jobs dictionary

Parameters:

• keys

  (Collection[OffloadKey]) –

  Chunks that finished storing.

• success

  (bool, default: True ) –

  Whether the GPU→primary transfer succeeded.

• req_context

  (ReqContext) –

  Per-request context forwarded to primary.prepare_read().

Source code in vllm/v1/kv_offload/tiering/manager.py

@override
def complete_store(
    self,
    keys: Collection[OffloadKey],
    req_context: ReqContext,
    success: bool = True,
) -> None:
    """Mark chunks as done storing from GPU to primary tier.

    This is where secondary tier cascading happens — after chunks are
    confirmed to be in the primary tier, they are cascaded to ALL
    secondary tiers.

    For each secondary tier:
    1. Call primary.prepare_read() to get LoadStoreSpec AND increment
       ref_cnt (protecting chunks during async transfer)
    2. Call tier.submit_store() to start async transfer: primary→secondary
    3. Track the job in _store_jobs dictionary

    Args:
        keys: Chunks that finished storing.
        success: Whether the GPU→primary transfer succeeded.
        req_context: Per-request context forwarded to primary.prepare_read().

    """
    # Step 1: Complete store in primary tier (makes chunks loadable)
    self.primary_tier.complete_store(keys, req_context, success)

    if success:
        # Step 2: Cascade to ALL secondary tiers
        # For each secondary tier, call primary.prepare_read() to get the
        # LoadStoreSpec AND to increment ref_cnt (protecting chunks from
        # eviction during the async transfer). One prepare_read() call per
        # secondary tier.
        for tier_idx, tier in enumerate(self.secondary_tiers):
            if not self._should_store_to_tier(tier, len(keys)):
                continue
            job_metadata = self.create_store_job(keys, req_context, tier_idx)
            tier.submit_store(job_metadata)

    # Note: The async transfers are now in flight. Their completion is
    # tracked via get_finished_jobs() / _maybe_process_finished_jobs().
    req_id = req_context.req_id
    state = self._req_state[req_id]
    assert state.pending_primary_stores > 0
    state.pending_primary_stores -= 1
    self._maybe_finalize_request(req_id)

create_store_job(keys, req_context, tier_idx=0)

Pin chunks in the primary tier and create a tracked store job.

Calls prepare_read() to increment ref_cnt (protecting chunks from eviction during the async transfer), allocates a job ID, and registers the job in _jobs.

The caller is responsible for the actual data transfer and reporting completion via get_finished_jobs().

Source code in vllm/v1/kv_offload/tiering/manager.py

def create_store_job(
    self,
    keys: Collection[OffloadKey],
    req_context: ReqContext,
    tier_idx: int = 0,
) -> TransferJob:
    """Pin chunks in the primary tier and create a tracked store job.

    Calls prepare_read() to increment ref_cnt (protecting chunks
    from eviction during the async transfer), allocates a job ID,
    and registers the job in _jobs.

    The caller is responsible for the actual data transfer and
    reporting completion via get_finished_jobs().
    """
    primary_chunks_spec = self.primary_tier.prepare_read(keys, req_context)
    assert isinstance(primary_chunks_spec, CPULoadStoreSpec)
    job_id = self._next_job_id()
    job_metadata = TransferJob(
        job_id=job_id,
        keys=keys,
        chunk_ids=primary_chunks_spec.chunk_ids,
        is_promotion=False,
        req_context=req_context,
    )
    self._register_job(job_metadata, tier_idx)
    return job_metadata

lookup(key, req_context, *, exclude_tier_idx=None)

Check whether a single chunk is offloaded and ready.

Algorithm

1. Process any completed async jobs first.
2. Query primary tier — short-circuit on hit or in-flight.
3. On primary miss, query secondary tiers — stop on first hit and initiate promotion.

Parameters:

• key

  (OffloadKey) –

  Chunk hash to look up.

• req_context

  (ReqContext) –

  Per-request context.

• exclude_tier_idx

  (int | None, default: None ) –

  Skip this tier index during the lookup.

Returns:

• LookupResult –

  HIT — chunk is ready in the primary tier.

• LookupResult –

  HIT_PENDING — chunk found but not yet readable (write in-flight on the primary tier).

• LookupResult –

  RETRY — promotion started or a secondary tier is busy.

• LookupResult –

  MISS — chunk not found in any tier, or primary is full and cannot accept a promotion.

Source code in vllm/v1/kv_offload/tiering/manager.py

@override
def lookup(
    self,
    key: OffloadKey,
    req_context: ReqContext,
    *,
    exclude_tier_idx: int | None = None,
) -> LookupResult:
    """Check whether a single chunk is offloaded and ready.

    Algorithm:
        1. Process any completed async jobs first.
        2. Query primary tier — short-circuit on hit or in-flight.
        3. On primary miss, query secondary tiers — stop on first
           hit and initiate promotion.

    Args:
        key: Chunk hash to look up.
        req_context: Per-request context.
        exclude_tier_idx: Skip this tier index during the lookup.

    Returns:
        HIT       — chunk is ready in the primary tier.
        HIT_PENDING — chunk found but not yet readable (write
                    in-flight on the primary tier).
        RETRY     — promotion started or a secondary tier is busy.
        MISS      — chunk not found in any tier, or primary is full
                    and cannot accept a promotion.

    """
    # Poll first so a promotion that finished since the last call is
    # already reflected as HIT (not stale HIT_PENDING/MISS) below, and
    # so chunks freed by cascade or promotion completions are evictable
    # in time for a promotion this lookup may initiate.
    self._maybe_process_finished_jobs()

    start_time = time.monotonic()
    primary_hit = self.primary_tier.lookup(key, req_context)
    lookup_duration = time.monotonic() - start_time
    self._metrics.on_lookup(
        req_context,
        key,
        self._metrics.primary_tier_label,
        primary_hit,
        lookup_duration,
    )
    if primary_hit is LookupResult.HIT:
        return LookupResult.HIT
    if primary_hit is LookupResult.HIT_PENDING:
        return LookupResult.HIT_PENDING

    any_retry = False
    for i, tier in enumerate(self.secondary_tiers):
        if i == exclude_tier_idx:
            continue
        if not req_context.load_tier_filter.allows(tier.medium, tier.locality):
            continue
        labelvalues = self._metrics.tier_label(i)
        start_time = time.monotonic()
        result = tier.lookup(key, req_context)
        lookup_duration = time.monotonic() - start_time
        if result is LookupResult.HIT:
            self._metrics.on_lookup(
                req_context,
                key,
                labelvalues,
                result,
                lookup_duration,
            )
            promoted = self._initiate_promotion(i, key, req_context)
            return LookupResult.MISS if not promoted else LookupResult.HIT_PENDING
        if result is LookupResult.RETRY:
            any_retry = True
        self._metrics.on_lookup(
            req_context,
            key,
            labelvalues,
            result,
            lookup_duration,
        )

    if any_retry:
        return LookupResult.RETRY
    return LookupResult.MISS

on_new_request(req_context, *, exclude_tier_idx=None)

Query each secondary tier for its offload policy preference.

Returns REQUEST_LEVEL if ANY secondary tier wants request-level. Only stores REQUEST_LEVEL tier decisions for use in prepare_store.

Source code in vllm/v1/kv_offload/tiering/manager.py

@override
def on_new_request(
    self,
    req_context: ReqContext,
    *,
    exclude_tier_idx: int | None = None,
) -> RequestOffloadingContext:
    """Query each secondary tier for its offload policy preference.

    Returns REQUEST_LEVEL if ANY secondary tier wants request-level.
    Only stores REQUEST_LEVEL tier decisions for use in prepare_store.
    """
    state = RequestState(req_context=req_context)
    self._metrics.on_new_request(req_context)
    for tier_idx, tier in enumerate(self.secondary_tiers):
        if tier_idx == exclude_tier_idx:
            continue
        tier_ctx = tier.on_new_request(req_context)
        if tier_ctx.policy == OffloadPolicy.REQUEST_LEVEL:
            if state.request_level_tiers is None:
                state.request_level_tiers = set()
            state.request_level_tiers.add(tier_idx)
    self._req_state[req_context.req_id] = state

    policy = (
        OffloadPolicy.REQUEST_LEVEL
        if state.request_level_tiers
        else OffloadPolicy.CHUNK_LEVEL
    )
    return RequestOffloadingContext(policy=policy)

on_schedule_end(context)

End-of-schedule hook: process finished jobs, flush deferred promotions, and reset the per-step gate.

Called once per scheduler step, as the last manager call of the step, from OffloadingConnectorScheduler.build_connector_meta().

Source code in vllm/v1/kv_offload/tiering/manager.py

@override
def on_schedule_end(self, context: ScheduleEndContext) -> None:
    """End-of-schedule hook: process finished jobs, flush deferred
    promotions, and reset the per-step gate.

    Called once per scheduler step, as the last manager call of the step,
    from OffloadingConnectorScheduler.build_connector_meta().
    """
    # Catch-all poll: guarantees jobs are processed even on steps where
    # lookup()/prepare_store() were never called (e.g. no requests
    # scheduled but a tier still has_pending_work()).
    self._maybe_process_finished_jobs()

    for tier in self.secondary_tiers:
        tier.serve_external_requests(self._tier_parents[tier])

    # Reset the per-step gate AFTER serve_external_requests so that
    # lookup() calls within it skip redundant _process_finished_jobs().
    self._processed_jobs_this_step = False

    self._flush_pending_promotions()
    # Keys parked by this step's prepare_store cannot have completed yet, so
    # they are simply re-parked here and retried on the next step. That is
    # safe: _maybe_finalize_request() holds off while pending_cascade_keys is
    # non-empty, so nothing finalizes early and the list still drains.
    self._flush_pending_cascades()
    for tier in self.secondary_tiers:
        tier.on_schedule_end(context)

    for req_id in context.new_req_ids:
        state = self._req_state.get(req_id)
        if state is None:
            continue
        self._metrics.on_request_allocated(state.req_context)

poll_tiers()

Advance the control plane of tiers that cannot wait for a step.

Polls those tiers for finished jobs and lets them serve whatever their counterpart has asked for -- the same two steps on_schedule_end() runs, narrowed to the tiers that opted in.

Excludes _flush_pending_promotions() and _flush_pending_cascades() by design: both are per-step batching points, and running them here would fragment batches the scheduler thread accumulated during schedule(). A promotion that a lookup() from this path initiates therefore waits for the next on_schedule_end() to be submitted.

The caller must hold self.lock for the whole call. serve_external_requests() relies on that: it establishes lookup() HITs and then pins them, and the two must not be separated by an eviction.

Serving can look keys up through the parent, and lookup() runs the step's once-per-step poll. From here that poll would reach every tier off-thread, re-enter the serving tier's get_finished_jobs() from inside its own serve, and leave the gate set so the next step skips its poll. The round has just polled its tiers itself, so it holds the gate shut for its duration and then puts it back untouched.

Source code in vllm/v1/kv_offload/tiering/manager.py

def poll_tiers(self) -> None:
    """Advance the control plane of tiers that cannot wait for a step.

    Polls those tiers for finished jobs and lets them serve whatever their
    counterpart has asked for -- the same two steps on_schedule_end() runs,
    narrowed to the tiers that opted in.

    Excludes _flush_pending_promotions() and _flush_pending_cascades() by
    design: both are per-step batching points, and running them here would
    fragment batches the scheduler thread accumulated during schedule(). A
    promotion that a lookup() from this path initiates therefore waits for
    the next on_schedule_end() to be submitted.

    The caller must hold self.lock for the whole call. serve_external_requests()
    relies on that: it establishes lookup() HITs and then pins them, and the
    two must not be separated by an eviction.

    Serving can look keys up through the parent, and lookup() runs the
    step's once-per-step poll. From here that poll would reach every tier
    off-thread, re-enter the serving tier's get_finished_jobs() from inside
    its own serve, and leave the gate set so the next step skips its poll.
    The round has just polled its tiers itself, so it holds the gate shut
    for its duration and then puts it back untouched.
    """
    gate = self._processed_jobs_this_step
    self._processed_jobs_this_step = True
    try:
        self._process_finished_jobs(self._external_serving_tiers)
        for _, tier in self._external_serving_tiers:
            tier.serve_external_requests(self._tier_parents[tier])
    finally:
        self._processed_jobs_this_step = gate

prepare_load(keys, req_context)

Prepare chunks to be loaded from primary tier to GPU.

Callers only pass keys already confirmed HIT by lookup() earlier this step.

This increments ref_cnt on the chunks in the primary tier, protecting them from eviction during the transfer.

Parameters:

• keys

  (Collection[OffloadKey]) –

  Chunks to prepare for loading.

• req_context

  (ReqContext) –

  Per-request context.

Returns:

• LoadStoreSpec –

  LoadStoreSpec for reading from primary tier.

Source code in vllm/v1/kv_offload/tiering/manager.py

@override
def prepare_load(
    self, keys: Collection[OffloadKey], req_context: ReqContext
) -> LoadStoreSpec:
    """Prepare chunks to be loaded from primary tier to GPU.

    Callers only pass keys already confirmed HIT by lookup() earlier this
    step.

    This increments ref_cnt on the chunks in the primary tier, protecting
    them from eviction during the transfer.

    Args:
        keys: Chunks to prepare for loading.
        req_context: Per-request context.

    Returns:
        LoadStoreSpec for reading from primary tier.

    """
    return self.primary_tier.prepare_load(keys, req_context)

prepare_store(keys, req_context)

Prepare chunks to be stored from GPU to primary tier.

CRITICAL: This method polls for finished jobs FIRST to ensure that any completed async transfers have their ref_cnt decremented before the primary tier makes eviction decisions.

For request-level tiers, chunks already present in the primary tier are immediately cascaded via submit_store().

Parameters:

• keys

  (Collection[OffloadKey]) –

  Chunks to prepare for storing.

• req_context

  (ReqContext) –

  Per-request context.

Returns:

• PrepareStoreOutput | None –

  PrepareStoreOutput describing where to store chunks and what was

• PrepareStoreOutput | None –

  evicted, or None if store cannot proceed.

Source code in vllm/v1/kv_offload/tiering/manager.py

@override
def prepare_store(
    self, keys: Collection[OffloadKey], req_context: ReqContext
) -> PrepareStoreOutput | None:
    """Prepare chunks to be stored from GPU to primary tier.

    CRITICAL: This method polls for finished jobs FIRST to ensure that any
    completed async transfers have their ref_cnt decremented before the
    primary tier makes eviction decisions.

    For request-level tiers, chunks already present in the primary tier
    are immediately cascaded via submit_store().

    Args:
        keys: Chunks to prepare for storing.
        req_context: Per-request context.

    Returns:
        PrepareStoreOutput describing where to store chunks and what was
        evicted, or None if store cannot proceed.

    """
    # Step 1: Poll for completed async jobs FIRST
    # _process_finished_jobs() handles two kinds of completions here:
    #  - Cascade completions (store to a secondary tier, either a local
    #    cascade or a store job created for a remote requester via
    #    create_store_job()): decrements ref_cnt on the primary chunks
    #    that were read, making them evictable again once ref_cnt hits 0.
    #  - Promotion completions (secondary->primary loads): sets a
    #    not-yet-ready chunk's ref_cnt from -1 to 0 via complete_write(),
    #    making it evictable for the first time.
    # Both must be accounted for before the eviction decision below.
    # Unconditional, not _maybe_process_finished_jobs(): eviction must see
    # the freshest completions, and on_schedule_end now runs at the very end
    # of the step, so the once-per-step gate would otherwise still be set by
    # this step's first lookup() and skip the poll.
    self._processed_jobs_this_step = True
    self._process_finished_jobs()

    # Step 2: Store to primary tier (new chunks only).
    # Cascading of these newly-stored chunks to ALL secondary tiers
    # happens later in complete_store(), after the GPU→Primary transfer
    # completes.
    primary_result = self.primary_tier.prepare_store(keys, req_context)

    if primary_result is None:
        return None

    if primary_result.keys_to_store:
        state = self._req_state[req_context.req_id]
        state.pending_primary_stores += 1

    # Step 3: For request-level tiers, cascade chunks already in primary
    request_level_tiers = self._req_state[req_context.req_id].request_level_tiers
    if request_level_tiers:
        keys_to_store_set = set(primary_result.keys_to_store)
        keys_already_in_primary = tuple(
            k for k in keys if k not in keys_to_store_set
        )
        if keys_already_in_primary:
            self._cascade_existing_chunks_to_request_level_tiers(
                keys_already_in_primary, req_context, request_level_tiers
            )

    return primary_result

reset_cache()

Reset transfer bookkeeping and primary-tier cache.

Called during sleep, weight update, or resume. Each secondary tier drains its in-flight transfers via drain_jobs() so no tier I/O is touching primary memory before the primary tier is reset. A stuck tier will block here visibly — preferable to silent corruption from reusing primary slots while a transfer is mid-copy.

Secondary tiers are intentionally not reset: persistent stores (FS, network) keep their data across resets. Active request state is retained so those requests can continue after the reset; finished requests are finalized and removed.

Source code in vllm/v1/kv_offload/tiering/manager.py

@override
def reset_cache(self) -> None:
    """Reset transfer bookkeeping and primary-tier cache.

    Called during sleep, weight update, or resume. Each secondary tier
    drains its in-flight transfers via drain_jobs() so no tier I/O is
    touching primary memory before the primary tier is reset. A stuck
    tier will block here visibly — preferable to silent corruption
    from reusing primary slots while a transfer is mid-copy.

    Secondary tiers are intentionally not reset: persistent stores
    (FS, network) keep their data across resets. Active request state is
    retained so those requests can continue after the reset; finished
    requests are finalized and removed.
    """
    for tier in self.secondary_tiers:
        tier.drain_jobs()
    # All tier I/O has stopped; consume their completion notifications
    # so manager bookkeeping is consistent before the primary reset.
    self._process_finished_jobs()
    assert not self._jobs

    # Deferred promotion submissions reserve primary slots that the
    # reset below invalidates; their submit_load() has not yet been
    # called so no tier I/O is touching that memory.
    self._pending_load_submissions.clear()
    self._request_load_sources.clear()
    self._metrics.assert_idle()

    finished_req_ids = []
    for req_id, state in self._req_state.items():
        state.pending_primary_stores = 0
        state.pending_cascade_keys.clear()
        if not state.is_finished:
            continue
        for tier in self.secondary_tiers:
            tier.on_request_finished(state.req_context)
        self._metrics.on_request_finished(state.req_context)
        finished_req_ids.append(req_id)

    self.primary_tier.reset_cache()

    for req_id in finished_req_ids:
        del self._req_state[req_id]
    self._processed_jobs_this_step = False

    for tier in self.secondary_tiers:
        if tier.bp_detector is not None:
            tier.bp_detector.reset()

shutdown()

Shut down secondary tiers before releasing primary resources.

Stops the control-plane thread first: it drives tier transports, and tearing one down underneath it (closing a socket, destroying a ZMQ context) can take the process down rather than raise. If the thread will not exit, hold the manager lock across tier teardown so it cannot be mid-round, and skip teardown entirely if even that is unavailable -- a lingering daemon thread in an exiting process beats a crash.

Every secondary tier is given a shutdown attempt. If any shutdown fails, preserve the primary mmap because a failed tier may still use it.

Source code in vllm/v1/kv_offload/tiering/manager.py

@override
def shutdown(self) -> None:
    """Shut down secondary tiers before releasing primary resources.

    Stops the control-plane thread first: it drives tier transports, and
    tearing one down underneath it (closing a socket, destroying a ZMQ
    context) can take the process down rather than raise. If the thread
    will not exit, hold the manager lock across tier teardown so it cannot
    be mid-round, and skip teardown entirely if even that is unavailable --
    a lingering daemon thread in an exiting process beats a crash.

    Every secondary tier is given a shutdown attempt. If any shutdown
    fails, preserve the primary mmap because a failed tier may still use it.
    """
    self._stop_control_plane()
    if self._control_plane_thread is not None:
        if not self._lock.acquire(timeout=_CONTROL_PLANE_JOIN_TIMEOUT_S):
            logger.error(
                "KV offload control-plane thread is still running and the "
                "manager lock is unavailable; skipping tier shutdown."
            )
            return
        try:
            self._shutdown_tiers()
        finally:
            self._lock.release()
        return
    self._shutdown_tiers()

take_events()

Yield events owned by the primary and secondary tiers.

Yields:

• Iterable[OffloadingEvent] –

  New OffloadingEvents collected by each tier since the last call.

Source code in vllm/v1/kv_offload/tiering/manager.py

@override
def take_events(self) -> Iterable[OffloadingEvent]:
    """Yield events owned by the primary and secondary tiers.

    Yields:
        New OffloadingEvents collected by each tier since the last call.

    """
    yield from self.primary_tier.take_events()
    for tier in self.secondary_tiers:
        yield from tier.take_events()

touch(keys, req_context)

Mark chunks as recently used in all tiers.

Parameters:

• keys

  (Collection[OffloadKey]) –

  Chunks to mark as recently used.

• req_context

  (ReqContext) –

  Per-request context.

Source code in vllm/v1/kv_offload/tiering/manager.py

@override
def touch(self, keys: Collection[OffloadKey], req_context: ReqContext):
    """Mark chunks as recently used in all tiers.

    Args:
        keys: Chunks to mark as recently used.
        req_context: Per-request context.

    """
    self.primary_tier.touch(keys, req_context)
    for tier in self.secondary_tiers:
        tier.touch(keys, req_context)

_SecondaryTierFacingParent

Bases: ParentManager

Wrapper that implements ParentManager by delegating to the TieringOffloadingManager with exclude_tier_idx set to the origin tier.

Source code in vllm/v1/kv_offload/tiering/manager.py

class _SecondaryTierFacingParent(ParentManager):
    """Wrapper that implements ParentManager by delegating to the
    TieringOffloadingManager with exclude_tier_idx set to the origin tier."""

    __slots__ = ("_m", "_origin_idx")

    def __init__(
        self,
        manager: "TieringOffloadingManager",
        tier_idx: int,
    ):
        self._m = manager
        self._origin_idx = tier_idx

    def on_new_request(self, req_context: ReqContext) -> RequestOffloadingContext:
        return self._m.on_new_request(req_context, exclude_tier_idx=self._origin_idx)

    def lookup(self, key: OffloadKey, req_context: ReqContext) -> LookupResult:
        return self._m.lookup(key, req_context, exclude_tier_idx=self._origin_idx)

    def create_store_job(
        self, keys: Collection[OffloadKey], req_context: ReqContext
    ) -> TransferJob:
        return self._m.create_store_job(keys, req_context, self._origin_idx)

    def on_request_finished(self, req_context: ReqContext) -> None:
        return self._m.on_request_finished(
            req_context, exclude_tier_idx=self._origin_idx
        )