vllm.v1.kv_offload.tiering.p2p.manager

P2PSecondaryTierManager: Secondary tier for P2P KV cache sharing.

Owns transports and a single bidirectional P2PSession per remote peer.

Classes:

• P2PDestInfo –

  Producer side: a remote fetches this request's blocks from us.

• P2PSecondaryTierManager –

  Secondary tier for P2P KV cache sharing.

• P2PSourceInfo –

  Consumer side: this request fetches from a remote (prefiller or peer).

P2PDestInfo dataclass

Producer side: a remote fetches this request's blocks from us.

kv_request_id is None when the remote_decoder block is present but malformed (no id); the block's presence still marks the request as remote-decode, so submit_store must fail rather than store locally.

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

@dataclass(slots=True)
class P2PDestInfo:
    """Producer side: a remote fetches this request's blocks from us.

    ``kv_request_id`` is None when the ``remote_decoder`` block is present
    but malformed (no id); the block's presence still marks the request as
    remote-decode, so submit_store must fail rather than store locally.
    """

    kv_request_id: str | None

P2PSecondaryTierManager

Bases: SecondaryTierManager

Secondary tier for P2P KV cache sharing.

A single P2PSession per remote peer handles both client-role (loading blocks from the peer) and server-role (serving blocks to the peer) over the same control connection.

Single-threaded: every public method runs on the scheduler thread, and the engine drives polling via get_finished_jobs() once per step. has_pending_work() keeps the engine ticking so the control transport and existing sessions are polled even when no requests are scheduled.

Methods:

• __init__ –

  Initialize the P2P secondary tier manager.

• drain_jobs –

  Block until every submitted load/store job has completed or failed.

• on_new_request –

  Parse kv_transfer_params once and open the outbound session.

• on_request_finished –

  Cancels pending loads and prunes session-scoped state.

• serve_external_requests –

  Serve inbound peer lookups against the tiering manager.

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

class P2PSecondaryTierManager(SecondaryTierManager):
    """Secondary tier for P2P KV cache sharing.

    A single P2PSession per remote peer handles both client-role (loading
    blocks from the peer) and server-role (serving blocks to the peer)
    over the same control connection.

    Single-threaded: every public method runs on the scheduler thread, and
    the engine drives polling via ``get_finished_jobs()`` once per step.
    ``has_pending_work()`` keeps the engine ticking so the control transport
    and existing sessions are polled even when no requests are scheduled.
    """

    cache_hit_source: ClassVar[CacheHitSource] = CacheHitSource.P2P

    def __init__(
        self,
        offloading_spec: OffloadingSpec,
        primary_kv_view: memoryview,
        tier_type: str = "p2p",
        host: str | None = None,
        port: int | None = None,
        backends: list[str] | None = None,
        num_threads: int = 4,
        unbound_store_timeout_s: float = _UNBOUND_STORE_TIMEOUT_S,
        **kwargs: Any,
    ) -> None:
        """Initialize the P2P secondary tier manager.

        All keyword arguments after ``primary_kv_view`` come from the
        ``secondary_tiers`` entry in ``kv_connector_extra_config``. See
        ``docs/features/kv_offloading_usage.md`` for the user-facing
        configuration reference.

        Args:
            offloading_spec: Owning ``OffloadingSpec`` (provides normalized
                model, parallel, and cache layout configuration).
            primary_kv_view: Memoryview over the CPU primary tier; the
                NIXL agent registers this region for RDMA transfers.
            tier_type: Tier identifier (defaults to ``"p2p"``).
            host: Address the ZMQ control socket binds to, used verbatim
                as both the bind address and the identity peers dial back
                (mirrors the NIXL connector's ``VLLM_NIXL_SIDE_CHANNEL_HOST``;
                no auto-detection). Defaults to
                ``VLLM_P2P_SIDE_CHANNEL_HOST`` (``localhost``) when not set;
                must be set to the node's routable IP for cross-host P2P so
                remote peers can reach the socket.
            port: Base port for the ZMQ control socket. Must be
                reachable from peers. Defaults to
                ``VLLM_P2P_SIDE_CHANNEL_PORT`` (``5710``) when not set.
                The bound port is ``base + data_parallel_index`` so each
                DP replica gets a distinct port (one socket per replica,
                like NIXL); for DP=1 the offset is 0.
            backends: NIXL transport backends (e.g. ``["UCX"]``,
                ``["MOONCAKE"]``, ``["LIBFABRIC"]``). Defaults to
                ``["UCX"]``. When any non-UCX backend is requested, the
                NIXL agent is initialized with ``backends=...``;
                otherwise it falls back to a UCX-only agent with
                ``num_threads`` threads.
            num_threads: NIXL agent worker threads for the UCX-only
                branch. Ignored when ``backends`` contains a non-UCX
                entry.
            unbound_store_timeout_s: Seconds a producer holds stored blocks
                for a consumer that has not fetched them yet, before
                ``_reap_unbound_stores`` drops them. Raise it for
                deployments whose prefills outlast the default; the blocks
                keep primary-tier slots pinned for the whole window.
            **kwargs: Reserved for future tier-specific options.

        Raises:
            ValueError: If ``unbound_store_timeout_s`` is not a positive
                number, or anything convertible to one.

        """
        backpressure_detector = kwargs.pop("backpressure_detector", None)
        if backpressure_detector is not None:
            # The generic (store-latency) detector is unreliable for P2P: in
            # PD mode dropping a store can leave the decoder waiting until the
            # load timeout instead of failing fast, and rendezvous time makes
            # store latency a poor pressure signal. Reject it until a
            # P2P-specific fail-fast path exists.
            raise ValueError(
                "Backpressure is not supported for the P2P secondary tier. "
                "The generic store-latency detector cannot fail fast in PD "
                "mode and rendezvous time makes store latency an unreliable "
                "pressure signal. Remove the 'backpressure' config from the "
                "p2p tier."
            )
        super().__init__(
            offloading_spec,
            primary_kv_view,
            tier_type,
        )
        try:
            timeout_s = float(unbound_store_timeout_s)
        except (TypeError, ValueError):
            timeout_s = float("nan")
        if not timeout_s > 0:
            raise ValueError(
                f"unbound_store_timeout_s must be a positive number, got "
                f"{unbound_store_timeout_s!r}"
            )
        self._unbound_store_timeout_s = timeout_s

        # Block hashes chain from NONE_HASH (see v1/core/kv_cache_utils.py).
        # Peers whose seeds differ compute different hashes for identical
        # content, so lookups silently miss and no KV crosses the wire. The
        # seed is advertised and verified against each peer during the
        # handshake, so a mismatch is rejected loudly instead of degrading
        # silently. Resolved lazily in _get_hash_seed: this tier is built
        # before init_none_hash runs, and a non-cryptographic hash algorithm
        # seeds NONE_HASH randomly, so the value is only known afterwards.
        self._hash_seed: str | None = None
        if host is None:
            host = envs.VLLM_P2P_SIDE_CHANNEL_HOST
        if port is None:
            port = envs.VLLM_P2P_SIDE_CHANNEL_PORT
        # One control socket per DP replica: offset the base by the global
        # data-parallel index so replicas on a host don't collide (mirrors
        # NIXL). For DP=1 the index is 0, leaving the base port unchanged.
        dp_index = offloading_spec.config.parallel.data_parallel_index
        port = int(port) + dp_index
        # Two decoupled identities:
        #   _local_id (``host:port``): the ZMQ control identity that peers
        #     dial back, used verbatim (the socket binds this host/port and
        #     the address is parsed back into host:port by the remote).
        #   _nixl_agent_name (uuid4): the NIXL agent name. It is never dialed
        #     — it travels opaquely inside the agent metadata blob — so it
        #     only needs to be globally unique. A per-process uuid guarantees
        #     that even for peers sharing a host:port (mirrors the NIXL
        #     connector; avoids the "remote agent name equals local" reject).
        self._local_id = f"{host}:{port}"
        self._nixl_agent_name = str(uuid.uuid4())

        config_fields = FileMapper.from_offloading_spec(
            root_dir="",
            offloading_spec=offloading_spec,
            blocks_per_file=offloading_spec.blocks_per_chunk,
            parallel_agnostic=True,
        ).get_run_config()
        self._data: DataTransport = NixlTransport(
            self._nixl_agent_name,
            primary_kv_view,
            config_fields=config_fields,
            backends=backends,
            num_threads=int(num_threads),
        )
        self._control: ControlTransport = ZmqTransport(self._local_id, host, port)

        self._sessions: dict[str, P2PSession] = {}
        # kv_request_id → session, set when the bound session has received
        # FetchMsg for that id. submit_store after binding routes directly
        # to the session; before binding, batches are parked in
        # _unbound_stores below. Stays in sync with _sessions: entries
        # pointing to a reaped session are purged in _reap_dead_sessions.
        self._kv_to_session: dict[str, P2PSession] = {}
        # kv_request_id → list of batches submit_store'd before any peer
        # asked for that id. Drained into a session by _on_session_fetch
        # when the corresponding FetchMsg arrives, or surfaced as failures
        # by _reap_unbound_stores after _unbound_store_timeout_s.
        self._unbound_stores: dict[str, list[_UnboundStoreBatch]] = {}
        # kv_request_id → time its parked batches were reaped. Nothing under
        # one of these ids can be served any more, so a FetchMsg for it is
        # rejected on the spot by _poll_once instead of parking demand until
        # the consumer's own load timeout, and submit_store fails rather than
        # re-parking under it — a prefill that outran the timeout would
        # otherwise keep pinning fresh primary-tier slots, one timeout at a
        # time. Entries are pruned after _REAPED_ID_RETENTION_S.
        self._reaped_stores: dict[str, float] = {}

        self._finished_jobs: list[JobResult] = []
        # kv_request_ids that hit a transport/session failure; On load lookup()
        # rejects them so the request falls back to local prefill.
        self._failed_req_ids: set[str] = set()
        # Synthetic lookup ctxs from reaped sessions still owing a
        # ``parent.on_request_finished`` (the session's failed_serves). The
        # dead session had no parent handle at teardown; these are flushed
        # at the top of the next ``serve_external_requests`` where the
        # handle is valid.
        self._failed_serve_ctxs: list[ReqContext] = []

    # ------------------------------------------------------------------
    # SecondaryTierManager interface
    # ------------------------------------------------------------------

    @override
    def lookup(self, key: OffloadKey, req_context: ReqContext) -> LookupResult:
        source = req_context.get_state(P2PSourceInfo)
        if source is None:
            return LookupResult.MISS
        if source.kv_request_id in self._failed_req_ids:
            return LookupResult.MISS

        # Symmetric-P2P consumer (``remote_kv_source`` sub-dict): probe the
        # peer asynchronously. First call registers the (kv_request_id,
        # key) entry and returns RETRY; flush_pending_lookups()
        # in on_schedule_end batches the LookupMsg; a later step's
        # lookup() returns HIT/MISS once LookupRespMsg has arrived.
        # PD path (``remote_prefiller`` sub-dict only) keeps the eager HIT.
        if source.do_probe:
            session = self._sessions.get(source.peer_id)
            if session is None:
                return LookupResult.MISS
            result = session.register_lookup(source.kv_request_id, key)
            if result is True:
                return LookupResult.HIT
            if result is False:
                return LookupResult.MISS
            return LookupResult.RETRY

        # PD consumer (we are the decoder): all kv blocks should be on the
        # prefiller side. Return HIT immediately.
        return LookupResult.HIT

    @override
    def on_new_request(self, req_context: ReqContext) -> RequestOffloadingContext:
        """Parse kv_transfer_params once and open the outbound session.

        Parses the P2P routing state onto ``req_context`` (cached for the
        later lookup/submit/finish calls). On the consumer side
        (``remote_prefiller`` for PD or ``remote_kv_source`` for symmetric
        P2P), open a session toward the producer at remote_host:remote_port
        so submit_load can issue FetchMsg as soon as it fires. On the
        prefiller side, sessions are created when the consumer's inbound
        connection arrives in _accept_new_peers — submit_store no longer
        pre-creates anything.

        Producer-leg requests (``remote_decoder`` present) ask for
        REQUEST_LEVEL: the peer needs every block of the request, not just
        the ones this request computed.
        """
        _annotate_req_context(req_context)
        source = req_context.get_state(P2PSourceInfo)
        if source is not None:
            session = self._get_or_create_session(source.peer_id)
            if session is None:
                self._failed_req_ids.add(source.kv_request_id)
        dest = req_context.get_state(P2PDestInfo)
        if dest is not None and dest.kv_request_id:
            return RequestOffloadingContext(policy=OffloadPolicy.REQUEST_LEVEL)
        return RequestOffloadingContext()

    @override
    def on_request_finished(self, req_context: ReqContext) -> None:
        """Cancels pending loads and prunes session-scoped state.

        Consumer side (``remote_prefiller`` for PD or ``remote_kv_source``
        for symmetric-P2P): looks up the session by peer_id because the
        producer's address is what addresses the client-role load to
        cancel; also drops any pending symmetric-P2P lookup state via
        ``session.finish_request``.
        Prefiller side (``remote_decoder`` set): looks up via kv_request_id
        because peer_id is no longer carried on store-time
        kv_transfer_params; if a session has bound the id, finish it. If
        no session has bound the id yet, this is a no-op: parked batches
        in `_unbound_stores` are left in place and cleaned up only by
        `_reap_unbound_stores` after `_unbound_store_timeout_s`.
        """
        source = req_context.get_state(P2PSourceInfo)
        dest = req_context.get_state(P2PDestInfo)
        kv_request_id = source.kv_request_id if source is not None else None
        if kv_request_id is None and dest is not None:
            kv_request_id = dest.kv_request_id
        if not kv_request_id:
            return
        self._failed_req_ids.discard(kv_request_id)

        if source is not None:
            session = self._sessions.get(source.peer_id)
            if session is not None:
                session.finish_request(kv_request_id)
            return

        # Prefiller-side finish: identify the session via kv_request_id.
        session = self._kv_to_session.pop(kv_request_id, None)
        if session is not None:
            session.finish_request(kv_request_id)
            return

    @override
    def submit_store(self, job_metadata: TransferJob) -> None:
        job_id = job_metadata.job_id
        keys = list(job_metadata.keys)
        block_ids = job_metadata.chunk_ids.tolist()

        assert len(keys) == len(block_ids)

        dest = job_metadata.req_context.get_state(P2PDestInfo)
        logger.debug(
            "P2P %s: submit_store ENTRY job_id=%d blocks=%d "
            "remote_decoder=%s kv_request_id=%s",
            self._local_id,
            job_id,
            len(block_ids),
            dest is not None,
            dest.kv_request_id if dest is not None else None,
        )
        # Absent ``remote_decoder`` block => not a remote-decode request:
        # succeed locally without parking. An empty/malformed dict is still
        # a remote-decode signal and must fail the missing-id check below.
        if dest is None:
            self._finished_jobs.append(JobResult(job_id=job_id, success=True))
            return

        kv_request_id = dest.kv_request_id
        if not kv_request_id:
            logger.warning(
                "P2P %s: submit_store missing kv_request_id",
                self._local_id,
            )
            self._finished_jobs.append(JobResult(job_id=job_id, success=False))
            return

        # This id's earlier batches were already reaped, so the consumer has
        # either given up or will be rejected by the fetch path. Parking more
        # would pin primary-tier slots for another full timeout on blocks
        # nobody can fetch, so fail the job now — a prefill that outran the
        # timeout falls back to local recompute on the decoder either way.
        if kv_request_id in self._reaped_stores:
            logger.warning(
                "P2P %s: submit_store for reaped kv_request_id=%s job_id=%d "
                "— failing without parking",
                self._local_id,
                kv_request_id,
                job_id,
            )
            self._finished_jobs.append(JobResult(job_id=job_id, success=False))
            return

        # Fast path: a session has already received FetchMsg for this id,
        # so we can route the batch straight into its ServerRole.
        session = self._kv_to_session.get(kv_request_id)
        if session is not None:
            session.add_stored_blocks(kv_request_id, keys, block_ids, job_id)
            return

        # No session bound yet — park the batch keyed by kv_request_id.
        # _on_session_fetch drains it on the first FetchMsg; if no peer
        # ever asks, _reap_unbound_stores surfaces the job as failed.
        self._unbound_stores.setdefault(kv_request_id, []).append(
            _UnboundStoreBatch(
                job_id=job_id,
                keys=keys,
                block_ids=block_ids,
            )
        )
        logger.debug(
            "P2P %s: parked submit_store kv_request_id=%s job_id=%d blocks=%d",
            self._local_id,
            kv_request_id,
            job_id,
            len(block_ids),
        )

    @override
    def submit_load(self, job_metadata: TransferJob) -> None:
        job_id = job_metadata.job_id
        keys = list(job_metadata.keys)
        block_ids = job_metadata.chunk_ids

        source = job_metadata.req_context.get_state(P2PSourceInfo)
        logger.debug(
            "P2P %s: submit_load ENTRY job_id=%d blocks=%d kv_request_id=%s peer=%s",
            self._local_id,
            job_id,
            len(block_ids),
            source.kv_request_id if source is not None else None,
            source.peer_id if source is not None else None,
        )
        if source is None:
            logger.debug(
                "P2P %s: submit_load job_id=%d FAILED missing consumer params",
                self._local_id,
                job_id,
            )
            self._finished_jobs.append(JobResult(job_id=job_id, success=False))
            return

        kv_request_id = source.kv_request_id
        peer_id = source.peer_id

        if not keys:
            logger.debug(
                "P2P %s: submit_load job_id=%d short-circuit success (no keys)",
                self._local_id,
                job_id,
            )
            self._finished_jobs.append(JobResult(job_id=job_id, success=True))
            return

        session = self._sessions.get(peer_id)
        if session is None:
            logger.warning(
                "P2P %s: submit_load job_id=%d NO SESSION for peer=%s",
                self._local_id,
                job_id,
                peer_id,
            )
            self._finished_jobs.append(JobResult(job_id=job_id, success=False))
            self._failed_req_ids.add(kv_request_id)
            return
        logger.debug(
            "P2P %s: submit_load job_id=%d -> request_blocks peer=%s "
            "kv_request_id=%s blocks=%d session_ready=%s",
            self._local_id,
            job_id,
            peer_id,
            kv_request_id,
            len(block_ids),
            session.ready,
        )
        session.request_blocks(job_id, kv_request_id, keys, block_ids)

    @override
    def get_finished_jobs(self) -> Iterable[JobResult]:
        # Drive one polling sweep on the scheduler thread, then hand off
        # whatever has accumulated. The engine calls this once per step
        # (and keeps stepping while has_pending_work() is True).
        self._poll_once()
        result = self._finished_jobs
        self._finished_jobs = []
        return result

    @override
    def has_pending_work(self) -> bool:
        # The engine tick is the only driver of _control.poll() and
        # session.poll(); without it we miss new peer connects and
        # inbound fetch messages on existing sessions. Keep the engine
        # ticking for the lifetime of this manager.
        return True

    @override
    def drain_jobs(self) -> None:
        """Block until every submitted load/store job has completed or failed.

        Loops calling ``_poll_once()`` until no session has outstanding
        inbound loads or in-flight outbound stores. Mid-flight transfers
        are NOT cancelled — the caller (``TieringOffloadingManager.reset_cache``)
        needs the primary memoryview to be quiescent, not aborted. Results
        accumulate in ``_finished_jobs`` and are surfaced by the next
        ``get_finished_jobs()`` call.
        """
        start = time.monotonic()
        warned = False
        while True:
            self._poll_once()
            pending = any(s.has_pending_work for s in self._sessions.values())
            if not pending:
                return
            if not warned and time.monotonic() - start > 5.0:
                logger.warning(
                    "P2PSecondaryTierManager.drain_jobs: still draining "
                    "after 5s; a stuck transfer will block the engine.",
                )
                warned = True
            time.sleep(_DRAIN_SLEEP_S)

    @override
    def serve_external_requests(self, parent: ParentManager) -> None:
        """Serve inbound peer lookups against the tiering manager.

        Called once per scheduler step (before this tier's
        ``on_schedule_end``) with a ``parent`` handle valid only for the
        duration of the call — the sole window in which the P2P server
        role may query the tiering manager. First release bookkeeping for
        the failed serves left by a reaped session, then let every live
        session resolve its enqueued inbound LookupMsgs.
        """
        if self._failed_serve_ctxs:
            for ctx in self._failed_serve_ctxs:
                parent.on_request_finished(ctx)
            self._failed_serve_ctxs = []
        for session in self._sessions.values():
            session.serve_external_requests(parent)

    @override
    def on_schedule_end(self, context: ScheduleEndContext) -> None:
        # Flush any p2p lookups aggregated during this step.
        # One LookupMsg per (peer, kv_request_id) with unsent entries;
        # send-gating happens inside the session if not yet ready.
        for session in self._sessions.values():
            session.flush_pending_lookups()

    # ------------------------------------------------------------------
    # Internal
    # ------------------------------------------------------------------

    def _get_hash_seed(self) -> str:
        """The seed NONE_HASH was derived from, resolved on first session."""
        if self._hash_seed is None:
            self._hash_seed = get_none_hash_seed()
        return self._hash_seed

    def _get_or_create_session(self, peer_id: str) -> P2PSession | None:
        """Return the existing session for peer_id, or open one outbound.

        Consumer-side helper for on_new_request: when ``remote_prefiller``
        (PD) or ``remote_kv_source`` (symmetric P2P) is set, the consumer
        must reach the producer at peer_id. If we already have a session
        toward that peer (from a prior load or a peer-initiated inbound),
        reuse it; otherwise open an outbound ControlConnection and build a
        connected session.

        Returns None if the session could not be created (capacity
        exceeded or transport error).
        """
        session = self._sessions.get(peer_id)
        if session is not None:
            return session

        max_peers = envs.VLLM_P2P_MAX_PEERS
        if len(self._sessions) >= max_peers:
            logger.warning(
                "P2P %s: rejecting outbound peer %s — session limit "
                "reached (%d/%d). Set VLLM_P2P_MAX_PEERS to increase.",
                self._local_id,
                peer_id,
                len(self._sessions),
                max_peers,
            )
            return None

        conn = None
        try:
            conn = self._control.connect(peer_id)
            session = P2PSession(
                peer_id=peer_id,
                local_id=self._local_id,
                transport=self._data,
                local_block_len=self._data.block_len,
                local_hash_seed=self._get_hash_seed(),
                conn=conn,
            )
        except Exception:
            logger.exception(
                "P2P %s: failed to open connection to %s",
                self._local_id,
                peer_id,
            )
            if conn is not None:
                conn.close()
            return None

        self._sessions[peer_id] = session
        return session

    def _accept_new_peers(self, new_connections: Sequence[ControlConnection]) -> None:
        max_peers = envs.VLLM_P2P_MAX_PEERS
        for conn in new_connections:
            logger.info(
                "P2P %s: accepting incoming connection from %s",
                self._local_id,
                conn.peer_id,
            )
            if len(self._sessions) >= max_peers:
                logger.warning(
                    "P2P %s: rejecting inbound peer %s — session limit "
                    "reached (%d/%d). Set VLLM_P2P_MAX_PEERS to increase.",
                    self._local_id,
                    conn.peer_id,
                    len(self._sessions),
                    max_peers,
                )
                conn.close()
                continue
            try:
                existing = self._sessions.get(conn.peer_id)
                if existing is not None:
                    raise ValueError(f"duplicate connection from {conn.peer_id}")
                self._sessions[conn.peer_id] = P2PSession(
                    peer_id=conn.peer_id,
                    local_id=self._local_id,
                    transport=self._data,
                    local_block_len=self._data.block_len,
                    local_hash_seed=self._get_hash_seed(),
                    conn=conn,
                )
                logger.info(
                    "P2P %s: created connected session for %s",
                    self._local_id,
                    conn.peer_id,
                )
            except (ValueError, KeyError, TypeError, AssertionError) as exc:
                logger.error("P2P %s: rejecting peer: %s", self._local_id, exc)
                conn.close()

    def _reap_dead_sessions(self) -> None:
        # Reap connected sessions whose connection died — peer is gone.
        # Stranded prefiller-side stores are no longer tracked through a
        # session (they live in _unbound_stores keyed by kv_request_id);
        # _reap_unbound_stores handles their timeout independently.
        dead: list[str] | None = None
        for pid, s in self._sessions.items():
            if s.connected and not s.alive:
                if dead is None:
                    dead = []
                dead.append(pid)
        if dead is None:
            return
        for pid in dead:
            session = self._sessions.pop(pid)
            # Purge any kv_request_id → session entries pointing at this
            # session so subsequent submit_stores fall back to the unbound
            # path (which will time out into failure if no peer rebinds).
            stale_kv_ids = [
                kid for kid, s in self._kv_to_session.items() if s is session
            ]
            for kid in stale_kv_ids:
                del self._kv_to_session[kid]
            close_result = session.close()
            for job_id in close_result.failed_jobs:
                self._finished_jobs.append(JobResult(job_id=job_id, success=False))
            for job_id in close_result.failed_stores:
                self._finished_jobs.append(JobResult(job_id=job_id, success=False))
            # Fail every client-side request (in-flight loads plus unresolved
            # symmetric-P2P probes) toward the dead peer so lookup() returns
            # MISS (local prefill) instead of RETRY forever — even if a fresh
            # session to the same peer is later opened by another request.
            self._failed_req_ids.update(close_result.failed_req_ids)
            # Release the TieringManager's per-request bookkeeping for the
            # dead session's synthetic lookups on the next serve_external_requests.
            self._failed_serve_ctxs.extend(close_result.failed_serves)
            self._data.remove_remote_peer(pid)
            logger.warning("P2P %s: peer %s down", self._local_id, pid)

    def _reap_unbound_stores(self) -> None:
        """Time out submit_store batches that no peer has ever fetched.

        Walks `_unbound_stores` for entries whose oldest batch is older
        than `_unbound_store_timeout_s`. Drops the kv_request_id, surfaces
        every batched job as failed, and records the id in `_reaped_stores`
        so a late inbound FetchMsg is rejected in one round trip and a late
        `submit_store` fails instead of re-parking under the reaped id.

        Also expires `_reaped_stores` entries whose consumer can no longer
        be waiting, which is why this runs before the empty check below.
        """
        now = time.monotonic()
        if self._reaped_stores:
            retention_deadline = now - _REAPED_ID_RETENTION_S
            for kid in [
                kid
                for kid, reaped_at in self._reaped_stores.items()
                if reaped_at <= retention_deadline
            ]:
                del self._reaped_stores[kid]

        if not self._unbound_stores:
            return
        deadline = now - self._unbound_store_timeout_s
        expired: list[str] | None = None
        for kid, batches in self._unbound_stores.items():
            # Batches are appended in arrival order, so the head is oldest.
            if batches and batches[0].submitted_at <= deadline:
                if expired is None:
                    expired = []
                expired.append(kid)
        if expired is None:
            return
        for kid in expired:
            batches = self._unbound_stores.pop(kid)
            self._reaped_stores[kid] = now
            for batch in batches:
                self._finished_jobs.append(
                    JobResult(job_id=batch.job_id, success=False)
                )
            logger.warning(
                "P2P %s: unbound store kv_request_id=%s timed out after %.0fs "
                "without a fetch — failing %d job(s)",
                self._local_id,
                kid,
                self._unbound_store_timeout_s,
                len(batches),
            )

    def _reap_idle_sessions(self) -> None:
        """Evict idle sessions when remaining peer slots are scarce.

        A session is idle when it has no pending work and its last activity
        timestamp exceeds ``VLLM_P2P_IDLE_TIMEOUT_S``. Eviction runs only
        when ``len(_sessions) >= VLLM_P2P_MAX_PEERS - _IDLE_REAP_HEADROOM``
        so a quiet but healthy mesh is not torn down. Uses the same
        teardown path as ``_reap_dead_sessions``.
        """
        idle_timeout = envs.VLLM_P2P_IDLE_TIMEOUT_S
        if idle_timeout <= 0:
            return
        max_peers = envs.VLLM_P2P_MAX_PEERS
        headroom = min(_IDLE_REAP_HEADROOM, max(0, max_peers - 1))
        if len(self._sessions) < max_peers - headroom:
            return
        deadline = time.monotonic() - idle_timeout
        idle: list[str] | None = None
        for pid, s in self._sessions.items():
            if not s.has_pending_work and s.last_activity_at <= deadline:
                if idle is None:
                    idle = []
                idle.append(pid)
        if idle is None:
            return
        for pid in idle:
            session = self._sessions.pop(pid)
            stale_kv_ids = [
                kid for kid, s in self._kv_to_session.items() if s is session
            ]
            for kid in stale_kv_ids:
                del self._kv_to_session[kid]
            close_result = session.close()
            for job_id in close_result.failed_jobs:
                self._finished_jobs.append(JobResult(job_id=job_id, success=False))
            for job_id in close_result.failed_stores:
                self._finished_jobs.append(JobResult(job_id=job_id, success=False))
            self._failed_req_ids.update(close_result.failed_req_ids)
            self._failed_serve_ctxs.extend(close_result.failed_serves)
            self._data.remove_remote_peer(pid)
            logger.info(
                "P2P %s: evicted idle peer %s (idle > %ds)",
                self._local_id,
                pid,
                idle_timeout,
            )

    # ------------------------------------------------------------------
    # Polling
    # ------------------------------------------------------------------

    def _poll_once(self) -> None:
        """One sweep of the polling work.

        Drains the control transport, polls every session, accumulates
        their results into ``_finished_jobs``, and reaps any dead sessions.
        Runs on the scheduler thread.
        """
        new_connections = self._control.poll()
        if new_connections:
            logger.info(
                "P2P %s: _poll_once got %d new connection(s): %s",
                self._local_id,
                len(new_connections),
                [c.peer_id for c in new_connections],
            )

        self._accept_new_peers(new_connections)

        for session in self._sessions.values():
            result = session.poll()
            for lr in result.loads:
                self._finished_jobs.append(
                    JobResult(job_id=lr.job_id, success=lr.success)
                )
                if not lr.success:
                    self._failed_req_ids.add(lr.kv_request_id)
            for sr in result.stores:
                self._finished_jobs.append(
                    JobResult(job_id=sr.job_id, success=sr.success)
                )
            # Bind kv_request_id → session for any FetchMsg this tick and
            # replay any submit_store batches parked while no peer was
            # asking. ServerRole.on_fetch already recorded the demand
            # inline in dispatch, so the replayed add_stored_blocks calls
            # match that demand and submit transfers immediately.
            for kv_request_id in result.new_fetch_ids:
                if kv_request_id in self._reaped_stores:
                    # This id's blocks were already reaped, so no
                    # submit_store will ever satisfy the demand on_fetch
                    # just recorded. Finalize the round now: the peer gets
                    # TransferDoneMsg(success=False) on this tick and falls
                    # back to local prefill, instead of parking until its
                    # own load timeout expires. The id is deliberately left
                    # unbound so any later submit_store keeps taking the
                    # reject path. Batches re-parked between the reap and the
                    # retention prune are failed here rather than left to a
                    # second reap, which would pin their slots for another
                    # full timeout.
                    session.finish_request(kv_request_id)
                    for batch in self._unbound_stores.pop(kv_request_id, ()):
                        self._finished_jobs.append(
                            JobResult(job_id=batch.job_id, success=False)
                        )
                    continue
                self._kv_to_session[kv_request_id] = session
                for batch in self._unbound_stores.pop(kv_request_id, ()):
                    session.add_stored_blocks(
                        kv_request_id, batch.keys, batch.block_ids, batch.job_id
                    )

        self._reap_dead_sessions()
        self._reap_unbound_stores()
        self._reap_idle_sessions()

    # ------------------------------------------------------------------
    # Lifecycle
    # ------------------------------------------------------------------

    @override
    def shutdown(self) -> None:
        self._drain_inflight_for_shutdown()
        for session in self._sessions.values():
            # Orphan ctxs from close() are intentionally dropped: the manager
            # is being torn down, so there is no next serve_external_requests
            # to flush them and no TieringManager left to release.
            session.close()
        self._sessions.clear()
        self._kv_to_session.clear()
        # Surface buffered store jobs as failed so the engine doesn't
        # leak them; the manager is going away after this call.
        for batches in self._unbound_stores.values():
            for batch in batches:
                self._finished_jobs.append(
                    JobResult(job_id=batch.job_id, success=False)
                )
        self._unbound_stores.clear()
        self._reaped_stores.clear()
        self._control.close()
        self._data.close()

    def _drain_inflight_for_shutdown(self) -> None:
        """Best-effort drain of inflight transfers before closing _data.

        Mirrors session._drain_abort but as a single bounded loop. Collects
        inflight transfer_ids from each session, repeatedly calls
        _data.cancel(..., mode="wait") and _data.poll() so handles can
        surface as done/failed, and falls back to mode="immediate" once
        _SHUTDOWN_DRAIN_TIMEOUT_S elapses so a wedged peer can't hang us.
        """
        ids = [tid for s in self._sessions.values() for tid in s._server._inflight]
        if not ids:
            return
        deadline = time.monotonic() + _SHUTDOWN_DRAIN_TIMEOUT_S
        still: list[int] = ids
        while still and time.monotonic() < deadline:
            still = list(self._data.cancel(still, mode="wait"))
            if not still:
                break
            # poll() advances NIXL handle state so the next wait-cancel
            # has a chance to release the handles.
            self._data.poll()
            time.sleep(_DRAIN_SLEEP_S)
        if still:
            logger.warning(
                "P2P %s: shutdown drain timed out after %.1fs with %d "
                "transfers still inflight — force-cancelling",
                self._local_id,
                _SHUTDOWN_DRAIN_TIMEOUT_S,
                len(still),
            )
            self._data.cancel(still, mode="immediate")

__init__(offloading_spec, primary_kv_view, tier_type='p2p', host=None, port=None, backends=None, num_threads=4, unbound_store_timeout_s=_UNBOUND_STORE_TIMEOUT_S, **kwargs)

Initialize the P2P secondary tier manager.

All keyword arguments after primary_kv_view come from the secondary_tiers entry in kv_connector_extra_config. See docs/features/kv_offloading_usage.md for the user-facing configuration reference.

Parameters:

• offloading_spec

  (OffloadingSpec) –

  Owning OffloadingSpec (provides normalized model, parallel, and cache layout configuration).

• primary_kv_view

  (memoryview) –

  Memoryview over the CPU primary tier; the NIXL agent registers this region for RDMA transfers.

• tier_type

  (str, default: 'p2p' ) –

  Tier identifier (defaults to "p2p").

• host

  (str | None, default: None ) –

  Address the ZMQ control socket binds to, used verbatim as both the bind address and the identity peers dial back (mirrors the NIXL connector's VLLM_NIXL_SIDE_CHANNEL_HOST; no auto-detection). Defaults to VLLM_P2P_SIDE_CHANNEL_HOST (localhost) when not set; must be set to the node's routable IP for cross-host P2P so remote peers can reach the socket.

• port

  (int | None, default: None ) –

  Base port for the ZMQ control socket. Must be reachable from peers. Defaults to VLLM_P2P_SIDE_CHANNEL_PORT (5710) when not set. The bound port is base + data_parallel_index so each DP replica gets a distinct port (one socket per replica, like NIXL); for DP=1 the offset is 0.

• backends

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

  NIXL transport backends (e.g. ["UCX"], ["MOONCAKE"], ["LIBFABRIC"]). Defaults to ["UCX"]. When any non-UCX backend is requested, the NIXL agent is initialized with backends=...; otherwise it falls back to a UCX-only agent with num_threads threads.

• num_threads

  (int, default: 4 ) –

  NIXL agent worker threads for the UCX-only branch. Ignored when backends contains a non-UCX entry.

• unbound_store_timeout_s

  (float, default: _UNBOUND_STORE_TIMEOUT_S ) –

  Seconds a producer holds stored blocks for a consumer that has not fetched them yet, before _reap_unbound_stores drops them. Raise it for deployments whose prefills outlast the default; the blocks keep primary-tier slots pinned for the whole window.

• **kwargs

  (Any, default: {} ) –

  Reserved for future tier-specific options.

Raises:

• ValueError –

  If unbound_store_timeout_s is not a positive number, or anything convertible to one.

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

def __init__(
    self,
    offloading_spec: OffloadingSpec,
    primary_kv_view: memoryview,
    tier_type: str = "p2p",
    host: str | None = None,
    port: int | None = None,
    backends: list[str] | None = None,
    num_threads: int = 4,
    unbound_store_timeout_s: float = _UNBOUND_STORE_TIMEOUT_S,
    **kwargs: Any,
) -> None:
    """Initialize the P2P secondary tier manager.

    All keyword arguments after ``primary_kv_view`` come from the
    ``secondary_tiers`` entry in ``kv_connector_extra_config``. See
    ``docs/features/kv_offloading_usage.md`` for the user-facing
    configuration reference.

    Args:
        offloading_spec: Owning ``OffloadingSpec`` (provides normalized
            model, parallel, and cache layout configuration).
        primary_kv_view: Memoryview over the CPU primary tier; the
            NIXL agent registers this region for RDMA transfers.
        tier_type: Tier identifier (defaults to ``"p2p"``).
        host: Address the ZMQ control socket binds to, used verbatim
            as both the bind address and the identity peers dial back
            (mirrors the NIXL connector's ``VLLM_NIXL_SIDE_CHANNEL_HOST``;
            no auto-detection). Defaults to
            ``VLLM_P2P_SIDE_CHANNEL_HOST`` (``localhost``) when not set;
            must be set to the node's routable IP for cross-host P2P so
            remote peers can reach the socket.
        port: Base port for the ZMQ control socket. Must be
            reachable from peers. Defaults to
            ``VLLM_P2P_SIDE_CHANNEL_PORT`` (``5710``) when not set.
            The bound port is ``base + data_parallel_index`` so each
            DP replica gets a distinct port (one socket per replica,
            like NIXL); for DP=1 the offset is 0.
        backends: NIXL transport backends (e.g. ``["UCX"]``,
            ``["MOONCAKE"]``, ``["LIBFABRIC"]``). Defaults to
            ``["UCX"]``. When any non-UCX backend is requested, the
            NIXL agent is initialized with ``backends=...``;
            otherwise it falls back to a UCX-only agent with
            ``num_threads`` threads.
        num_threads: NIXL agent worker threads for the UCX-only
            branch. Ignored when ``backends`` contains a non-UCX
            entry.
        unbound_store_timeout_s: Seconds a producer holds stored blocks
            for a consumer that has not fetched them yet, before
            ``_reap_unbound_stores`` drops them. Raise it for
            deployments whose prefills outlast the default; the blocks
            keep primary-tier slots pinned for the whole window.
        **kwargs: Reserved for future tier-specific options.

    Raises:
        ValueError: If ``unbound_store_timeout_s`` is not a positive
            number, or anything convertible to one.

    """
    backpressure_detector = kwargs.pop("backpressure_detector", None)
    if backpressure_detector is not None:
        # The generic (store-latency) detector is unreliable for P2P: in
        # PD mode dropping a store can leave the decoder waiting until the
        # load timeout instead of failing fast, and rendezvous time makes
        # store latency a poor pressure signal. Reject it until a
        # P2P-specific fail-fast path exists.
        raise ValueError(
            "Backpressure is not supported for the P2P secondary tier. "
            "The generic store-latency detector cannot fail fast in PD "
            "mode and rendezvous time makes store latency an unreliable "
            "pressure signal. Remove the 'backpressure' config from the "
            "p2p tier."
        )
    super().__init__(
        offloading_spec,
        primary_kv_view,
        tier_type,
    )
    try:
        timeout_s = float(unbound_store_timeout_s)
    except (TypeError, ValueError):
        timeout_s = float("nan")
    if not timeout_s > 0:
        raise ValueError(
            f"unbound_store_timeout_s must be a positive number, got "
            f"{unbound_store_timeout_s!r}"
        )
    self._unbound_store_timeout_s = timeout_s

    # Block hashes chain from NONE_HASH (see v1/core/kv_cache_utils.py).
    # Peers whose seeds differ compute different hashes for identical
    # content, so lookups silently miss and no KV crosses the wire. The
    # seed is advertised and verified against each peer during the
    # handshake, so a mismatch is rejected loudly instead of degrading
    # silently. Resolved lazily in _get_hash_seed: this tier is built
    # before init_none_hash runs, and a non-cryptographic hash algorithm
    # seeds NONE_HASH randomly, so the value is only known afterwards.
    self._hash_seed: str | None = None
    if host is None:
        host = envs.VLLM_P2P_SIDE_CHANNEL_HOST
    if port is None:
        port = envs.VLLM_P2P_SIDE_CHANNEL_PORT
    # One control socket per DP replica: offset the base by the global
    # data-parallel index so replicas on a host don't collide (mirrors
    # NIXL). For DP=1 the index is 0, leaving the base port unchanged.
    dp_index = offloading_spec.config.parallel.data_parallel_index
    port = int(port) + dp_index
    # Two decoupled identities:
    #   _local_id (``host:port``): the ZMQ control identity that peers
    #     dial back, used verbatim (the socket binds this host/port and
    #     the address is parsed back into host:port by the remote).
    #   _nixl_agent_name (uuid4): the NIXL agent name. It is never dialed
    #     — it travels opaquely inside the agent metadata blob — so it
    #     only needs to be globally unique. A per-process uuid guarantees
    #     that even for peers sharing a host:port (mirrors the NIXL
    #     connector; avoids the "remote agent name equals local" reject).
    self._local_id = f"{host}:{port}"
    self._nixl_agent_name = str(uuid.uuid4())

    config_fields = FileMapper.from_offloading_spec(
        root_dir="",
        offloading_spec=offloading_spec,
        blocks_per_file=offloading_spec.blocks_per_chunk,
        parallel_agnostic=True,
    ).get_run_config()
    self._data: DataTransport = NixlTransport(
        self._nixl_agent_name,
        primary_kv_view,
        config_fields=config_fields,
        backends=backends,
        num_threads=int(num_threads),
    )
    self._control: ControlTransport = ZmqTransport(self._local_id, host, port)

    self._sessions: dict[str, P2PSession] = {}
    # kv_request_id → session, set when the bound session has received
    # FetchMsg for that id. submit_store after binding routes directly
    # to the session; before binding, batches are parked in
    # _unbound_stores below. Stays in sync with _sessions: entries
    # pointing to a reaped session are purged in _reap_dead_sessions.
    self._kv_to_session: dict[str, P2PSession] = {}
    # kv_request_id → list of batches submit_store'd before any peer
    # asked for that id. Drained into a session by _on_session_fetch
    # when the corresponding FetchMsg arrives, or surfaced as failures
    # by _reap_unbound_stores after _unbound_store_timeout_s.
    self._unbound_stores: dict[str, list[_UnboundStoreBatch]] = {}
    # kv_request_id → time its parked batches were reaped. Nothing under
    # one of these ids can be served any more, so a FetchMsg for it is
    # rejected on the spot by _poll_once instead of parking demand until
    # the consumer's own load timeout, and submit_store fails rather than
    # re-parking under it — a prefill that outran the timeout would
    # otherwise keep pinning fresh primary-tier slots, one timeout at a
    # time. Entries are pruned after _REAPED_ID_RETENTION_S.
    self._reaped_stores: dict[str, float] = {}

    self._finished_jobs: list[JobResult] = []
    # kv_request_ids that hit a transport/session failure; On load lookup()
    # rejects them so the request falls back to local prefill.
    self._failed_req_ids: set[str] = set()
    # Synthetic lookup ctxs from reaped sessions still owing a
    # ``parent.on_request_finished`` (the session's failed_serves). The
    # dead session had no parent handle at teardown; these are flushed
    # at the top of the next ``serve_external_requests`` where the
    # handle is valid.
    self._failed_serve_ctxs: list[ReqContext] = []

_drain_inflight_for_shutdown()

Best-effort drain of inflight transfers before closing _data.

Mirrors session._drain_abort but as a single bounded loop. Collects inflight transfer_ids from each session, repeatedly calls _data.cancel(..., mode="wait") and _data.poll() so handles can surface as done/failed, and falls back to mode="immediate" once _SHUTDOWN_DRAIN_TIMEOUT_S elapses so a wedged peer can't hang us.

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

def _drain_inflight_for_shutdown(self) -> None:
    """Best-effort drain of inflight transfers before closing _data.

    Mirrors session._drain_abort but as a single bounded loop. Collects
    inflight transfer_ids from each session, repeatedly calls
    _data.cancel(..., mode="wait") and _data.poll() so handles can
    surface as done/failed, and falls back to mode="immediate" once
    _SHUTDOWN_DRAIN_TIMEOUT_S elapses so a wedged peer can't hang us.
    """
    ids = [tid for s in self._sessions.values() for tid in s._server._inflight]
    if not ids:
        return
    deadline = time.monotonic() + _SHUTDOWN_DRAIN_TIMEOUT_S
    still: list[int] = ids
    while still and time.monotonic() < deadline:
        still = list(self._data.cancel(still, mode="wait"))
        if not still:
            break
        # poll() advances NIXL handle state so the next wait-cancel
        # has a chance to release the handles.
        self._data.poll()
        time.sleep(_DRAIN_SLEEP_S)
    if still:
        logger.warning(
            "P2P %s: shutdown drain timed out after %.1fs with %d "
            "transfers still inflight — force-cancelling",
            self._local_id,
            _SHUTDOWN_DRAIN_TIMEOUT_S,
            len(still),
        )
        self._data.cancel(still, mode="immediate")

_get_hash_seed()

The seed NONE_HASH was derived from, resolved on first session.

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

def _get_hash_seed(self) -> str:
    """The seed NONE_HASH was derived from, resolved on first session."""
    if self._hash_seed is None:
        self._hash_seed = get_none_hash_seed()
    return self._hash_seed

_get_or_create_session(peer_id)

Return the existing session for peer_id, or open one outbound.

Consumer-side helper for on_new_request: when remote_prefiller (PD) or remote_kv_source (symmetric P2P) is set, the consumer must reach the producer at peer_id. If we already have a session toward that peer (from a prior load or a peer-initiated inbound), reuse it; otherwise open an outbound ControlConnection and build a connected session.

Returns None if the session could not be created (capacity exceeded or transport error).

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

def _get_or_create_session(self, peer_id: str) -> P2PSession | None:
    """Return the existing session for peer_id, or open one outbound.

    Consumer-side helper for on_new_request: when ``remote_prefiller``
    (PD) or ``remote_kv_source`` (symmetric P2P) is set, the consumer
    must reach the producer at peer_id. If we already have a session
    toward that peer (from a prior load or a peer-initiated inbound),
    reuse it; otherwise open an outbound ControlConnection and build a
    connected session.

    Returns None if the session could not be created (capacity
    exceeded or transport error).
    """
    session = self._sessions.get(peer_id)
    if session is not None:
        return session

    max_peers = envs.VLLM_P2P_MAX_PEERS
    if len(self._sessions) >= max_peers:
        logger.warning(
            "P2P %s: rejecting outbound peer %s — session limit "
            "reached (%d/%d). Set VLLM_P2P_MAX_PEERS to increase.",
            self._local_id,
            peer_id,
            len(self._sessions),
            max_peers,
        )
        return None

    conn = None
    try:
        conn = self._control.connect(peer_id)
        session = P2PSession(
            peer_id=peer_id,
            local_id=self._local_id,
            transport=self._data,
            local_block_len=self._data.block_len,
            local_hash_seed=self._get_hash_seed(),
            conn=conn,
        )
    except Exception:
        logger.exception(
            "P2P %s: failed to open connection to %s",
            self._local_id,
            peer_id,
        )
        if conn is not None:
            conn.close()
        return None

    self._sessions[peer_id] = session
    return session

_poll_once()

One sweep of the polling work.

Drains the control transport, polls every session, accumulates their results into _finished_jobs, and reaps any dead sessions. Runs on the scheduler thread.

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

def _poll_once(self) -> None:
    """One sweep of the polling work.

    Drains the control transport, polls every session, accumulates
    their results into ``_finished_jobs``, and reaps any dead sessions.
    Runs on the scheduler thread.
    """
    new_connections = self._control.poll()
    if new_connections:
        logger.info(
            "P2P %s: _poll_once got %d new connection(s): %s",
            self._local_id,
            len(new_connections),
            [c.peer_id for c in new_connections],
        )

    self._accept_new_peers(new_connections)

    for session in self._sessions.values():
        result = session.poll()
        for lr in result.loads:
            self._finished_jobs.append(
                JobResult(job_id=lr.job_id, success=lr.success)
            )
            if not lr.success:
                self._failed_req_ids.add(lr.kv_request_id)
        for sr in result.stores:
            self._finished_jobs.append(
                JobResult(job_id=sr.job_id, success=sr.success)
            )
        # Bind kv_request_id → session for any FetchMsg this tick and
        # replay any submit_store batches parked while no peer was
        # asking. ServerRole.on_fetch already recorded the demand
        # inline in dispatch, so the replayed add_stored_blocks calls
        # match that demand and submit transfers immediately.
        for kv_request_id in result.new_fetch_ids:
            if kv_request_id in self._reaped_stores:
                # This id's blocks were already reaped, so no
                # submit_store will ever satisfy the demand on_fetch
                # just recorded. Finalize the round now: the peer gets
                # TransferDoneMsg(success=False) on this tick and falls
                # back to local prefill, instead of parking until its
                # own load timeout expires. The id is deliberately left
                # unbound so any later submit_store keeps taking the
                # reject path. Batches re-parked between the reap and the
                # retention prune are failed here rather than left to a
                # second reap, which would pin their slots for another
                # full timeout.
                session.finish_request(kv_request_id)
                for batch in self._unbound_stores.pop(kv_request_id, ()):
                    self._finished_jobs.append(
                        JobResult(job_id=batch.job_id, success=False)
                    )
                continue
            self._kv_to_session[kv_request_id] = session
            for batch in self._unbound_stores.pop(kv_request_id, ()):
                session.add_stored_blocks(
                    kv_request_id, batch.keys, batch.block_ids, batch.job_id
                )

    self._reap_dead_sessions()
    self._reap_unbound_stores()
    self._reap_idle_sessions()

_reap_idle_sessions()

Evict idle sessions when remaining peer slots are scarce.

A session is idle when it has no pending work and its last activity timestamp exceeds VLLM_P2P_IDLE_TIMEOUT_S. Eviction runs only when len(_sessions) >= VLLM_P2P_MAX_PEERS - _IDLE_REAP_HEADROOM so a quiet but healthy mesh is not torn down. Uses the same teardown path as _reap_dead_sessions.

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

def _reap_idle_sessions(self) -> None:
    """Evict idle sessions when remaining peer slots are scarce.

    A session is idle when it has no pending work and its last activity
    timestamp exceeds ``VLLM_P2P_IDLE_TIMEOUT_S``. Eviction runs only
    when ``len(_sessions) >= VLLM_P2P_MAX_PEERS - _IDLE_REAP_HEADROOM``
    so a quiet but healthy mesh is not torn down. Uses the same
    teardown path as ``_reap_dead_sessions``.
    """
    idle_timeout = envs.VLLM_P2P_IDLE_TIMEOUT_S
    if idle_timeout <= 0:
        return
    max_peers = envs.VLLM_P2P_MAX_PEERS
    headroom = min(_IDLE_REAP_HEADROOM, max(0, max_peers - 1))
    if len(self._sessions) < max_peers - headroom:
        return
    deadline = time.monotonic() - idle_timeout
    idle: list[str] | None = None
    for pid, s in self._sessions.items():
        if not s.has_pending_work and s.last_activity_at <= deadline:
            if idle is None:
                idle = []
            idle.append(pid)
    if idle is None:
        return
    for pid in idle:
        session = self._sessions.pop(pid)
        stale_kv_ids = [
            kid for kid, s in self._kv_to_session.items() if s is session
        ]
        for kid in stale_kv_ids:
            del self._kv_to_session[kid]
        close_result = session.close()
        for job_id in close_result.failed_jobs:
            self._finished_jobs.append(JobResult(job_id=job_id, success=False))
        for job_id in close_result.failed_stores:
            self._finished_jobs.append(JobResult(job_id=job_id, success=False))
        self._failed_req_ids.update(close_result.failed_req_ids)
        self._failed_serve_ctxs.extend(close_result.failed_serves)
        self._data.remove_remote_peer(pid)
        logger.info(
            "P2P %s: evicted idle peer %s (idle > %ds)",
            self._local_id,
            pid,
            idle_timeout,
        )

_reap_unbound_stores()

Time out submit_store batches that no peer has ever fetched.

Walks _unbound_stores for entries whose oldest batch is older than _unbound_store_timeout_s. Drops the kv_request_id, surfaces every batched job as failed, and records the id in _reaped_stores so a late inbound FetchMsg is rejected in one round trip and a late submit_store fails instead of re-parking under the reaped id.

Also expires _reaped_stores entries whose consumer can no longer be waiting, which is why this runs before the empty check below.

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

def _reap_unbound_stores(self) -> None:
    """Time out submit_store batches that no peer has ever fetched.

    Walks `_unbound_stores` for entries whose oldest batch is older
    than `_unbound_store_timeout_s`. Drops the kv_request_id, surfaces
    every batched job as failed, and records the id in `_reaped_stores`
    so a late inbound FetchMsg is rejected in one round trip and a late
    `submit_store` fails instead of re-parking under the reaped id.

    Also expires `_reaped_stores` entries whose consumer can no longer
    be waiting, which is why this runs before the empty check below.
    """
    now = time.monotonic()
    if self._reaped_stores:
        retention_deadline = now - _REAPED_ID_RETENTION_S
        for kid in [
            kid
            for kid, reaped_at in self._reaped_stores.items()
            if reaped_at <= retention_deadline
        ]:
            del self._reaped_stores[kid]

    if not self._unbound_stores:
        return
    deadline = now - self._unbound_store_timeout_s
    expired: list[str] | None = None
    for kid, batches in self._unbound_stores.items():
        # Batches are appended in arrival order, so the head is oldest.
        if batches and batches[0].submitted_at <= deadline:
            if expired is None:
                expired = []
            expired.append(kid)
    if expired is None:
        return
    for kid in expired:
        batches = self._unbound_stores.pop(kid)
        self._reaped_stores[kid] = now
        for batch in batches:
            self._finished_jobs.append(
                JobResult(job_id=batch.job_id, success=False)
            )
        logger.warning(
            "P2P %s: unbound store kv_request_id=%s timed out after %.0fs "
            "without a fetch — failing %d job(s)",
            self._local_id,
            kid,
            self._unbound_store_timeout_s,
            len(batches),
        )

drain_jobs()

Block until every submitted load/store job has completed or failed.

Loops calling _poll_once() until no session has outstanding inbound loads or in-flight outbound stores. Mid-flight transfers are NOT cancelled — the caller (TieringOffloadingManager.reset_cache) needs the primary memoryview to be quiescent, not aborted. Results accumulate in _finished_jobs and are surfaced by the next get_finished_jobs() call.

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

@override
def drain_jobs(self) -> None:
    """Block until every submitted load/store job has completed or failed.

    Loops calling ``_poll_once()`` until no session has outstanding
    inbound loads or in-flight outbound stores. Mid-flight transfers
    are NOT cancelled — the caller (``TieringOffloadingManager.reset_cache``)
    needs the primary memoryview to be quiescent, not aborted. Results
    accumulate in ``_finished_jobs`` and are surfaced by the next
    ``get_finished_jobs()`` call.
    """
    start = time.monotonic()
    warned = False
    while True:
        self._poll_once()
        pending = any(s.has_pending_work for s in self._sessions.values())
        if not pending:
            return
        if not warned and time.monotonic() - start > 5.0:
            logger.warning(
                "P2PSecondaryTierManager.drain_jobs: still draining "
                "after 5s; a stuck transfer will block the engine.",
            )
            warned = True
        time.sleep(_DRAIN_SLEEP_S)

on_new_request(req_context)

Parse kv_transfer_params once and open the outbound session.

Parses the P2P routing state onto req_context (cached for the later lookup/submit/finish calls). On the consumer side (remote_prefiller for PD or remote_kv_source for symmetric P2P), open a session toward the producer at remote_host:remote_port so submit_load can issue FetchMsg as soon as it fires. On the prefiller side, sessions are created when the consumer's inbound connection arrives in _accept_new_peers — submit_store no longer pre-creates anything.

Producer-leg requests (remote_decoder present) ask for REQUEST_LEVEL: the peer needs every block of the request, not just the ones this request computed.

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

@override
def on_new_request(self, req_context: ReqContext) -> RequestOffloadingContext:
    """Parse kv_transfer_params once and open the outbound session.

    Parses the P2P routing state onto ``req_context`` (cached for the
    later lookup/submit/finish calls). On the consumer side
    (``remote_prefiller`` for PD or ``remote_kv_source`` for symmetric
    P2P), open a session toward the producer at remote_host:remote_port
    so submit_load can issue FetchMsg as soon as it fires. On the
    prefiller side, sessions are created when the consumer's inbound
    connection arrives in _accept_new_peers — submit_store no longer
    pre-creates anything.

    Producer-leg requests (``remote_decoder`` present) ask for
    REQUEST_LEVEL: the peer needs every block of the request, not just
    the ones this request computed.
    """
    _annotate_req_context(req_context)
    source = req_context.get_state(P2PSourceInfo)
    if source is not None:
        session = self._get_or_create_session(source.peer_id)
        if session is None:
            self._failed_req_ids.add(source.kv_request_id)
    dest = req_context.get_state(P2PDestInfo)
    if dest is not None and dest.kv_request_id:
        return RequestOffloadingContext(policy=OffloadPolicy.REQUEST_LEVEL)
    return RequestOffloadingContext()

on_request_finished(req_context)

Cancels pending loads and prunes session-scoped state.

Consumer side (remote_prefiller for PD or remote_kv_source for symmetric-P2P): looks up the session by peer_id because the producer's address is what addresses the client-role load to cancel; also drops any pending symmetric-P2P lookup state via session.finish_request. Prefiller side (remote_decoder set): looks up via kv_request_id because peer_id is no longer carried on store-time kv_transfer_params; if a session has bound the id, finish it. If no session has bound the id yet, this is a no-op: parked batches in _unbound_stores are left in place and cleaned up only by _reap_unbound_stores after _unbound_store_timeout_s.

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

@override
def on_request_finished(self, req_context: ReqContext) -> None:
    """Cancels pending loads and prunes session-scoped state.

    Consumer side (``remote_prefiller`` for PD or ``remote_kv_source``
    for symmetric-P2P): looks up the session by peer_id because the
    producer's address is what addresses the client-role load to
    cancel; also drops any pending symmetric-P2P lookup state via
    ``session.finish_request``.
    Prefiller side (``remote_decoder`` set): looks up via kv_request_id
    because peer_id is no longer carried on store-time
    kv_transfer_params; if a session has bound the id, finish it. If
    no session has bound the id yet, this is a no-op: parked batches
    in `_unbound_stores` are left in place and cleaned up only by
    `_reap_unbound_stores` after `_unbound_store_timeout_s`.
    """
    source = req_context.get_state(P2PSourceInfo)
    dest = req_context.get_state(P2PDestInfo)
    kv_request_id = source.kv_request_id if source is not None else None
    if kv_request_id is None and dest is not None:
        kv_request_id = dest.kv_request_id
    if not kv_request_id:
        return
    self._failed_req_ids.discard(kv_request_id)

    if source is not None:
        session = self._sessions.get(source.peer_id)
        if session is not None:
            session.finish_request(kv_request_id)
        return

    # Prefiller-side finish: identify the session via kv_request_id.
    session = self._kv_to_session.pop(kv_request_id, None)
    if session is not None:
        session.finish_request(kv_request_id)
        return

serve_external_requests(parent)

Serve inbound peer lookups against the tiering manager.

Called once per scheduler step (before this tier's on_schedule_end) with a parent handle valid only for the duration of the call — the sole window in which the P2P server role may query the tiering manager. First release bookkeeping for the failed serves left by a reaped session, then let every live session resolve its enqueued inbound LookupMsgs.

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

@override
def serve_external_requests(self, parent: ParentManager) -> None:
    """Serve inbound peer lookups against the tiering manager.

    Called once per scheduler step (before this tier's
    ``on_schedule_end``) with a ``parent`` handle valid only for the
    duration of the call — the sole window in which the P2P server
    role may query the tiering manager. First release bookkeeping for
    the failed serves left by a reaped session, then let every live
    session resolve its enqueued inbound LookupMsgs.
    """
    if self._failed_serve_ctxs:
        for ctx in self._failed_serve_ctxs:
            parent.on_request_finished(ctx)
        self._failed_serve_ctxs = []
    for session in self._sessions.values():
        session.serve_external_requests(parent)

P2PSourceInfo dataclass

Consumer side: this request fetches from a remote (prefiller or peer).

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

@dataclass(slots=True)
class P2PSourceInfo:
    """Consumer side: this request fetches from a remote (prefiller or peer)."""

    kv_request_id: str
    peer_id: str
    do_probe: bool  # False for remote_prefiller (PD), True for remote_kv_source

_UnboundStoreBatch dataclass

A submit_store batch parked at the manager before any peer has fetched.

Indexed by kv_request_id only — the prefiller no longer learns the peer identity at store time. When a FetchMsg(kv_request_id) arrives on some session, the manager binds the kv_request_id to that session and replays every parked batch into ServerRole via session.add_stored_blocks.

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

@dataclass
class _UnboundStoreBatch:
    """A submit_store batch parked at the manager before any peer has fetched.

    Indexed by kv_request_id only — the prefiller no longer learns the peer
    identity at store time. When a FetchMsg(kv_request_id) arrives on some
    session, the manager binds the kv_request_id to that session and replays
    every parked batch into ServerRole via session.add_stored_blocks.
    """

    job_id: int
    keys: list[OffloadKey]
    block_ids: Sequence[int]
    submitted_at: float = field(default_factory=time.monotonic)

_annotate_req_context(req_context)

Parse kv_transfer_params once and cache the P2P routing state.

Called from on_new_request; later calls for the same request read the cached P2PSourceInfo/P2PDestInfo via get_state instead of re-parsing.

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

def _annotate_req_context(req_context: ReqContext) -> None:
    """Parse kv_transfer_params once and cache the P2P routing state.

    Called from ``on_new_request``; later calls for the same request read
    the cached ``P2PSourceInfo``/``P2PDestInfo`` via ``get_state`` instead
    of re-parsing.
    """
    source = _parse_source(req_context.kv_transfer_params)
    if source is not None:
        req_context.set_state(source)
    dest = _parse_dest(req_context.kv_transfer_params)
    if dest is not None:
        req_context.set_state(dest)

_parse_dest(kv_params)

Parse the producer remote_decoder sub-dict into a P2PDestInfo, or None if the block is absent (not a remote-decode request).

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

def _parse_dest(kv_params: dict | None) -> P2PDestInfo | None:
    """Parse the producer ``remote_decoder`` sub-dict into a ``P2PDestInfo``,
    or None if the block is absent (not a remote-decode request)."""
    role = _remote_decoder_params(kv_params)
    if role is None:
        return None
    return P2PDestInfo(kv_request_id=role.get("kv_request_id") or None)

_parse_source(kv_params)

Parse the consumer sub-dict (PD remote_prefiller or symmetric remote_kv_source) into a P2PSourceInfo, or None if absent/incomplete.

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

def _parse_source(kv_params: dict | None) -> P2PSourceInfo | None:
    """Parse the consumer sub-dict (PD ``remote_prefiller`` or symmetric
    ``remote_kv_source``) into a ``P2PSourceInfo``, or None if absent/incomplete."""
    role = _remote_prefiller_params(kv_params)
    do_probe = False
    if role is None:
        role = _remote_kv_source_params(kv_params)
        do_probe = True
    if not role:
        return None
    peer_id = _peer_id_from_params(role)
    kv_request_id = role.get("kv_request_id")
    if peer_id is None or not kv_request_id:
        return None
    return P2PSourceInfo(
        kv_request_id=kv_request_id,
        peer_id=peer_id,
        do_probe=do_probe,
    )

_peer_id_from_params(role_params)

Build host:port peer_id from a role-scoped sub-dict, or None.

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

def _peer_id_from_params(role_params: dict) -> str | None:
    """Build ``host:port`` peer_id from a role-scoped sub-dict, or None."""
    host = role_params.get("remote_host")
    port = role_params.get("remote_port")
    if host and port:
        return f"{host}:{port}"
    return None

_remote_decoder_params(kv_params)

Return the remote_decoder sub-dict, or None if absent.

Set on prefiller requests to name the remote decoder they serve; carries kv_request_id.

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

def _remote_decoder_params(kv_params: dict | None) -> dict | None:
    """Return the ``remote_decoder`` sub-dict, or None if absent.

    Set on prefiller requests to name the remote decoder they serve;
    carries kv_request_id.
    """
    if not kv_params:
        return None
    return kv_params.get("remote_decoder")

_remote_kv_source_params(kv_params)

Return the remote_kv_source sub-dict, or None if absent.

Set on symmetric-P2P consumer requests to name the remote source they pull from; carries kv_request_id, remote_host, remote_port.

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

def _remote_kv_source_params(kv_params: dict | None) -> dict | None:
    """Return the ``remote_kv_source`` sub-dict, or None if absent.

    Set on symmetric-P2P consumer requests to name the remote source they
    pull from; carries kv_request_id, remote_host, remote_port.
    """
    if not kv_params:
        return None
    return kv_params.get("remote_kv_source")

_remote_prefiller_params(kv_params)

Return the remote_prefiller sub-dict, or None if absent.

Set on decoder requests to name the remote prefiller they pull from; carries kv_request_id, remote_host, remote_port.

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

def _remote_prefiller_params(kv_params: dict | None) -> dict | None:
    """Return the ``remote_prefiller`` sub-dict, or None if absent.

    Set on decoder requests to name the remote prefiller they pull from;
    carries kv_request_id, remote_host, remote_port.
    """
    if not kv_params:
        return None
    return kv_params.get("remote_prefiller")