[Bugfix][KV Offload][P2P] Fail a dead peer's sibling ports on the same host at once (opt-in)

docs/features/kv_offloading_usage.md

@@ -313,6 +313,7 @@ Block content hashes must match across instances for peers to exchange blocks (s
 | `backends` | no | `["UCX"]` | NIXL transport backends. See [NixlConnector Usage Guide](nixl_connector_usage.md#selecting-a-nixl-transport-backend-plugin) for available backends and selection guidance. |
 | `num_threads` | no | `4` | NIXL agent worker threads. Only used when `backends` is UCX-only; ignored when any non-UCX backend is requested. |
 | `unbound_store_timeout_s` | no | `60` | Seconds a producer holds stored blocks for a consumer that has not fetched them yet. Raise it for deployments whose prefills outlast the default; the blocks keep primary-tier CPU slots pinned for the whole window. Once it expires a late fetch is rejected in a single round trip, so the consumer falls back to local prefill immediately. |
+| `group_peers_by_host` | no | `false` | Treat every control port on a peer's host as one failure domain, as for the DP ranks of one pod. When a session to one port dies, the sessions to the host's other ports are torn down too (their in-flight loads fail and those requests prefill locally), and the host gets no new outbound sessions for 10 s. Leave it off when independent instances share a host. |
 
 The `backends` and `num_threads` options mirror the conditional logic used by [`NixlConnector`](nixl_connector_usage.md#selecting-a-nixl-transport-backend-plugin): when any non-UCX backend is configured, NIXL is initialised with `backends=...`; otherwise it falls back to a UCX-only agent with the configured `num_threads`. This lets the P2P tier use a different transport (e.g. `MOONCAKE`, `GDS_MT`, `LIBFABRIC`) than the main `NixlConnector` running in the same process.
 

tests/v1/kv_offload/tiering/p2p/test_manager.py

@@ -147,6 +147,8 @@ def _make_manager() -> P2PSecondaryTierManager:
     mgr._failed_serve_ctxs = []
     mgr._retained = {}
     mgr._retained_force_released = 0
+    mgr._group_peers_by_host = False
+    mgr._quarantined_hosts = {}
     return mgr
 
 
@@ -2910,6 +2912,83 @@ def send(self, msg: dict) -> None:
         assert self.PEER not in mgr._sessions
 
 
+class TestGroupPeersByHost:
+    """With group_peers_by_host, a dead peer takes down the sessions to the
+    other ports on its host (the DP ranks of one pod): their loads fail now
+    instead of at the load timeout, and new requests naming the host prefill
+    locally for _HOST_QUARANTINE_S (#56828)."""
+
+    def _reap(self, monkeypatch, group: bool = True):
+        warnings: list[str] = []
+        monkeypatch.setattr(
+            manager_module.logger,
+            "warning",
+            lambda msg, *args, **_: warnings.append(msg % args),
+        )
+        monkeypatch.setattr(
+            manager_module,
+            "P2PSession",
+            lambda *, peer_id, **_: _FakeSession(peer_id=peer_id),
+        )
+        mgr = _make_manager()
+        mgr._group_peers_by_host = group
+        connected: list[str] = []
+
+        def connect(pid: str) -> _RecordingConn:
+            connected.append(pid)
+            return _RecordingConn(pid)
+
+        mgr._control = SimpleNamespace(  # type: ignore[assignment]
+            connect=connect
+        )
+        removed: list[str] = []
+        mgr._data = SimpleNamespace(  # type: ignore[assignment]
+            remove_remote_peer=removed.append, block_len=1
+        )
+        for i, (pid, alive) in enumerate(
+            [("10.0.0.1:7777", False), ("10.0.0.1:7778", True), ("10.0.0.2:7777", True)]
+        ):
+            mgr._sessions[pid] = _FakeSession(  # type: ignore[assignment]
+                peer_id=pid, alive=alive, close_jobs=[i], close_req_ids=[f"r{i}"]
+            )
+        mgr._reap_dead_sessions()
+        return mgr, connected, removed, warnings
+
+    def test_dead_peer_fails_loads_to_other_ports_on_its_host(self, monkeypatch):
+        mgr, _, removed, warnings = self._reap(monkeypatch)
+
+        assert {r.job_id for r in mgr._finished_jobs if not r.success} == {0, 1}
+        assert mgr._failed_req_ids == {"r0", "r1"}
+        assert set(mgr._sessions) == {"10.0.0.2:7777"}
+        assert sorted(removed) == ["10.0.0.1:7777", "10.0.0.1:7778"]
+        assert (
+            "10.0.0.1:7777 on the same host is down; failed 1 load(s)" in (warnings[-1])
+        )
+
+    def test_new_request_to_quarantined_host_prefills_locally(self, monkeypatch):
+        mgr, connected, _, _ = self._reap(monkeypatch)
+        ctx = _req_context(_remote_prefiller_kv_params("10.0.0.1", 7779, "r3"))
+
+        mgr.on_new_request(ctx)
+        assert connected == []
+        assert mgr.lookup(OffloadKey(b"k"), ctx) == LookupResult.MISS
+
+        later = time.monotonic() + manager_module._HOST_QUARANTINE_S
+        monkeypatch.setattr(manager_module.time, "monotonic", lambda: later)
+        mgr.on_new_request(
+            _req_context(_remote_prefiller_kv_params("10.0.0.1", 7779, "r4"))
+        )
+        assert connected == ["10.0.0.1:7779"]
+
+    def test_off_by_default(self, monkeypatch):
+        mgr, connected, _, _ = self._reap(monkeypatch, group=False)
+
+        assert {r.job_id for r in mgr._finished_jobs} == {0}
+        assert set(mgr._sessions) == {"10.0.0.1:7778", "10.0.0.2:7777"}
+        mgr.on_new_request(_req_context(_remote_prefiller_kv_params("10.0.0.1", 7779)))
+        assert connected == ["10.0.0.1:7779"]
+
+
 # ---------------------------------------------------------------------------
 # Serving a peer while a cache reset holds promotions
 # ---------------------------------------------------------------------------

vllm/v1/kv_offload/tiering/p2p/manager.py

@@ -91,6 +91,16 @@
 _RECONNECT_MIN_BACKOFF_S = 1.0
 _RECONNECT_MAX_BACKOFF_S = 60.0
 
+# With group_peers_by_host, how long a host whose peer went down gets no
+# new outbound sessions. Matches the ZMQ heartbeat timeout, the time a dead
+# port takes to be detected on its own.
+_HOST_QUARANTINE_S = 10.0
+
+
+def _host_of(peer_id: str) -> str:
+    """The host part of a ``host:port`` peer_id."""
+    return peer_id.rpartition(":")[0]
+
 
 def _remote_prefiller_params(kv_params: dict | None) -> dict | None:
     """Return the ``remote_prefiller`` sub-dict, or None if absent.
@@ -249,6 +259,7 @@ def __init__(
         backends: list[str] | None = None,
         num_threads: int = 4,
         unbound_store_timeout_s: float = _UNBOUND_STORE_TIMEOUT_S,
+        group_peers_by_host: bool = False,
         **kwargs: Any,
     ) -> None:
         """Initialize the P2P secondary tier manager.
@@ -291,6 +302,12 @@ def __init__(
                 ``_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.
+            group_peers_by_host: Treat every port on a peer's host as one
+                failure domain, as for the DP ranks of one pod. When a
+                session dies, the sessions to the host's other ports are
+                torn down too and the host gets no new outbound sessions
+                for ``_HOST_QUARANTINE_S``. Leave off when independent
+                instances share a host.
             **kwargs: Reserved for future tier-specific options.
 
         Raises:
@@ -414,6 +431,9 @@ def __init__(
         # of reaped sessions that may still be writing; see _poll_retained.
         self._retained: dict[int, tuple[str, frozenset[int], float]] = {}
         self._retained_force_released = 0
+        self._group_peers_by_host = bool(group_peers_by_host)
+        # host → monotonic time until which it gets no new outbound session.
+        self._quarantined_hosts: dict[str, float] = {}
 
     # ------------------------------------------------------------------
     # SecondaryTierManager interface
@@ -759,6 +779,8 @@ def _get_or_create_session(self, peer_id: str) -> P2PSession | None:
             return session
         if self._reconnect_wait(peer_id) > 0 or self._retained_peer(peer_id):
             return None
+        if self._quarantined_hosts.get(_host_of(peer_id), 0.0) > time.monotonic():
+            return None
 
         max_peers = envs.VLLM_P2P_MAX_PEERS
         if len(self._sessions) >= max_peers:
@@ -875,7 +897,24 @@ def _reap_dead_sessions(self) -> None:
                 dead.append(pid)
         if dead is None:
             return
-        for pid in dead:
+        # With group_peers_by_host: every other session to a dead peer's host,
+        # mapped to that peer. Its loads fail now instead of at their load
+        # timeout, and the host gets no new outbound sessions for a while.
+        siblings: dict[str, str] = {}
+        if self._group_peers_by_host:
+            now = time.monotonic()
+            self._quarantined_hosts = {
+                h: t for h, t in self._quarantined_hosts.items() if t > now
+            }
+            dead_hosts = {_host_of(pid): pid for pid in dead}
+            for host in dead_hosts:
+                self._quarantined_hosts[host] = now + _HOST_QUARANTINE_S
+            siblings = {
+                pid: dead_hosts[_host_of(pid)]
+                for pid in self._sessions
+                if pid not in dead and _host_of(pid) in dead_hosts
+            }
+        for pid in dead + list(siblings):
             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
@@ -908,11 +947,17 @@ def _reap_dead_sessions(self) -> None:
                 backoff = self._schedule_reconnect_backoff(pid)
                 if backoff:
                     fault_info += f"; reconnect backoff {backoff:.0f}s"
+            reason = session.death_reason
+            if pid in siblings:
+                reason = (
+                    f"{siblings[pid]} on the same host is down; "
+                    f"failed {len(close_result.failed_jobs)} load(s)"
+                )
             logger.warning(
                 "P2P %s: peer %s down (%s)%s",
                 self._local_id,
                 pid,
-                session.death_reason,
+                reason,
                 fault_info,
             )
 

vllm/v1/kv_offload/tiering/p2p/session/client.py

@@ -524,10 +524,13 @@ def collect_results(self) -> list[LoadResult]:
                 if load.aborted_at is None:
                     if now - load.submitted_at >= load.timeout_s:
                         logger.warning(
-                            "P2PSession %s: %s round=%s timed out, sending abort",
+                            "P2PSession %s: %s round=%s job_id=%d timed out "
+                            "%.1fs after submit, sending abort",
                             self._peer_id,
                             req_id,
                             round_seq,
+                            load.job_id,
+                            now - load.submitted_at,
                         )
                         self._abort_load(req_id, round_seq, load)
                 elif (