diff --git a/docs/usage/security.md b/docs/usage/security.md index 50c4149be8..f069de0def 100644 --- a/docs/usage/security.md +++ b/docs/usage/security.md @@ -303,6 +303,25 @@ Only deployments whose API clients are trusted should start the server with `--trust-request-mm-kwargs` to restore per-request overrides. Do not enable this option on an endpoint exposed to untrusted clients. +### P2P KV-Offload Peer Session Limits + +Deployments that enable the P2P secondary tier (`OffloadingConnector` with `TieringOffloadingSpec` and a `p2p` secondary tier) accept peer addresses from `kv_transfer_params` in API requests. Without limits, an attacker with API access could submit many unique non-listening peer addresses, each consuming ZeroMQ sockets until the context quota is exhausted, crashing the EngineCore. + +vLLM enforces several configurable limits to prevent this: + +| Environment Variable | Default | Description | +| --- | --- | --- | +| `VLLM_P2P_MAX_PEERS` | `128` | Maximum concurrent P2P peer sessions. New peers (both outbound and inbound) are rejected when this limit is reached. | +| `VLLM_P2P_HANDSHAKE_TIMEOUT_S` | `30` | Seconds to wait for a newly connected peer to complete the P2P handshake. Sessions that remain connected but not ready after this deadline are torn down. | +| `VLLM_P2P_IDLE_TIMEOUT_S` | `300` | Seconds a P2P session may remain idle (no pending work, no messages) before it is eligible for eviction. Idle eviction runs only when remaining peer slots are within 8 of `VLLM_P2P_MAX_PEERS`. Set to `0` to disable idle eviction. | + +Connections that fail (ZMQ socket errors, handshake failures) are handled gracefully — the request falls back to local prefill instead of crashing the engine. + +**Recommendations:** + +- **Public-facing deployments with P2P enabled:** Lower `VLLM_P2P_MAX_PEERS` to the expected number of legitimate peers (e.g., `16` or `32`). +- **Sensitive deployments:** Consider deploying behind a reverse proxy that strips or validates `kv_transfer_params` from API requests to prevent client-supplied peer addresses entirely. + ## Tool Server and MCP Security vLLM supports connecting to external tool servers via the `--tool-server` argument. This enables models to call tools through the Responses API (`/v1/responses`). Tool server support works with all models — it is not limited to specific model architectures. diff --git a/tests/test_envs.py b/tests/test_envs.py index dfd08a9ec2..7356539bcd 100644 --- a/tests/test_envs.py +++ b/tests/test_envs.py @@ -60,6 +60,24 @@ def test_p2p_side_channel_defaults_and_override(monkeypatch: pytest.MonkeyPatch) assert envs.VLLM_P2P_SIDE_CHANNEL_PORT == 5799 +def test_p2p_resource_limit_defaults_and_override( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.delenv("VLLM_P2P_MAX_PEERS", raising=False) + monkeypatch.delenv("VLLM_P2P_HANDSHAKE_TIMEOUT_S", raising=False) + monkeypatch.delenv("VLLM_P2P_IDLE_TIMEOUT_S", raising=False) + assert envs.VLLM_P2P_MAX_PEERS == 128 + assert envs.VLLM_P2P_HANDSHAKE_TIMEOUT_S == 30 + assert envs.VLLM_P2P_IDLE_TIMEOUT_S == 300 + + monkeypatch.setenv("VLLM_P2P_MAX_PEERS", "64") + monkeypatch.setenv("VLLM_P2P_HANDSHAKE_TIMEOUT_S", "15") + monkeypatch.setenv("VLLM_P2P_IDLE_TIMEOUT_S", "60") + assert envs.VLLM_P2P_MAX_PEERS == 64 + assert envs.VLLM_P2P_HANDSHAKE_TIMEOUT_S == 15 + assert envs.VLLM_P2P_IDLE_TIMEOUT_S == 60 + + def test_getattr_with_cache(monkeypatch: pytest.MonkeyPatch): monkeypatch.setenv("VLLM_HOST_IP", "1.1.1.1") monkeypatch.setenv("VLLM_PORT", "1234") diff --git a/tests/v1/kv_offload/tiering/p2p/test_manager.py b/tests/v1/kv_offload/tiering/p2p/test_manager.py index e35b1d022e..930158c7f6 100644 --- a/tests/v1/kv_offload/tiering/p2p/test_manager.py +++ b/tests/v1/kv_offload/tiering/p2p/test_manager.py @@ -647,6 +647,7 @@ class _FakeSession: close_req_ids: list[str] | None = None, close_stores: list[int] | None = None, close_failed_serves: list[ReqContext] | None = None, + last_activity_at: float | None = None, ) -> None: self.peer_id = peer_id self.alive = alive @@ -663,6 +664,9 @@ class _FakeSession: self.stores_added: list[tuple[str, list, object, int]] = [] self.attached: list[object] = [] self.finishes: list[str] = [] + self.last_activity_at: float = ( + last_activity_at if last_activity_at is not None else time.monotonic() + ) # Mirror P2PSession._server._inflight (transfer_id → handle) and # P2PSession._client.has_active_loads for the shutdown-drain and # drain_jobs paths. Tests populate _server._inflight when needed. @@ -1844,3 +1848,272 @@ class TestBindHostPortDefaults: mgr_b = self._construct(monkeypatch, host="localhost", port=5710) assert mgr_a._local_id == mgr_b._local_id == "localhost:5710" assert mgr_a._nixl_agent_name != mgr_b._nixl_agent_name + + +# --------------------------------------------------------------------------- +# Tests for session capacity limit (VLLM_P2P_MAX_PEERS) +# --------------------------------------------------------------------------- + + +class TestSessionCapLimit: + """_get_or_create_session and _accept_new_peers reject new peers when + the session count reaches VLLM_P2P_MAX_PEERS.""" + + def _make_with_control(self): + mgr = _make_manager() + created = [] + + class FakeControl: + def connect(self, peer_id): + conn = _RecordingConn(peer_id) + created.append(conn) + return conn + + def poll(self): + return [] + + class FakeData: + block_len = 4096 + base_addr = 0x1000 + num_blocks = 16 + config_fingerprint = "" + + def get_agent_metadata(self): + return b"meta" + + def add_remote_peer(self, *a, **k): + pass + + def remove_remote_peer(self, pid): + pass + + mgr._control = FakeControl() # type: ignore[assignment] + mgr._data = FakeData() # type: ignore[assignment] + return mgr, created + + def test_outbound_rejected_at_capacity(self, monkeypatch): + monkeypatch.setenv("VLLM_P2P_MAX_PEERS", "2") + import vllm.envs + + vllm.envs.__dict__.pop("VLLM_P2P_MAX_PEERS", None) + + mgr, created = self._make_with_control() + mgr._sessions["peer-a:1"] = _FakeSession(peer_id="peer-a:1") # type: ignore[assignment] + mgr._sessions["peer-b:2"] = _FakeSession(peer_id="peer-b:2") # type: ignore[assignment] + + result = mgr._get_or_create_session("peer-c:3") + assert result is None + assert "peer-c:3" not in mgr._sessions + assert len(created) == 0 + + def test_on_new_request_records_failure_on_cap(self, monkeypatch): + monkeypatch.setenv("VLLM_P2P_MAX_PEERS", "1") + import vllm.envs + + vllm.envs.__dict__.pop("VLLM_P2P_MAX_PEERS", None) + + mgr, _ = self._make_with_control() + mgr._sessions["existing:1"] = _FakeSession(peer_id="existing:1") # type: ignore[assignment] + + ctx = _req_context( + kv_params=_remote_prefiller_kv_params( + remote_host="10.0.0.2", remote_port=9000, kv_request_id="req-cap" + ) + ) + mgr.on_new_request(ctx) + assert "req-cap" in mgr._failed_req_ids + + def test_inbound_rejected_at_capacity(self, monkeypatch): + monkeypatch.setenv("VLLM_P2P_MAX_PEERS", "1") + import vllm.envs + + vllm.envs.__dict__.pop("VLLM_P2P_MAX_PEERS", None) + + mgr, _ = self._make_with_control() + mgr._sessions["existing:1"] = _FakeSession(peer_id="existing:1") # type: ignore[assignment] + + conn = _RecordingConn("new-peer:2") + mgr._accept_new_peers([conn]) + assert conn.close_calls == 1 + assert "new-peer:2" not in mgr._sessions + + def test_existing_session_reused_below_cap(self, monkeypatch): + monkeypatch.setenv("VLLM_P2P_MAX_PEERS", "1") + import vllm.envs + + vllm.envs.__dict__.pop("VLLM_P2P_MAX_PEERS", None) + + mgr, _ = self._make_with_control() + existing = _FakeSession(peer_id="existing:1") + mgr._sessions["existing:1"] = existing # type: ignore[assignment] + + result = mgr._get_or_create_session("existing:1") + assert result is existing + + +# --------------------------------------------------------------------------- +# Tests for graceful error boundary in _get_or_create_session +# --------------------------------------------------------------------------- + + +class TestGetOrCreateSessionErrorBoundary: + def test_connect_failure_returns_none(self): + mgr = _make_manager() + + class FailingControl: + def connect(self, peer_id): + raise RuntimeError("socket exhausted") + + def poll(self): + return [] + + mgr._control = FailingControl() # type: ignore[assignment] + result = mgr._get_or_create_session("bad:1") + assert result is None + assert "bad:1" not in mgr._sessions + + def test_session_construction_failure_closes_conn(self): + """connect() succeeding but P2PSession construction failing must + close the conn so it is not left registered without a session.""" + mgr = _make_manager() + created: list[_RecordingConn] = [] + + class RaisingConn(_RecordingConn): + def send(self, msg): + raise RuntimeError("send failed during connect") + + class Control: + def connect(self, peer_id): + conn = RaisingConn(peer_id) + created.append(conn) + return conn + + def poll(self): + return [] + + class FakeData: + block_len = 4096 + base_addr = 0x1000 + num_blocks = 16 + config_fingerprint = "" + + def get_agent_metadata(self): + return b"meta" + + mgr._control = Control() # type: ignore[assignment] + mgr._data = FakeData() # type: ignore[assignment] + result = mgr._get_or_create_session("bad:2") + assert result is None + assert "bad:2" not in mgr._sessions + assert len(created) == 1 + assert created[0].close_calls == 1 + + +# --------------------------------------------------------------------------- +# Tests for idle session eviction +# --------------------------------------------------------------------------- + + +class TestIdleSessionEviction: + def _make_with_fake_data(self): + mgr = _make_manager() + + class FakeControl: + def poll(self): + return [] + + class FakeData: + def remove_remote_peer(self, pid): + pass + + mgr._control = FakeControl() # type: ignore[assignment] + mgr._data = FakeData() # type: ignore[assignment] + return mgr + + def test_idle_session_evicted(self, monkeypatch): + monkeypatch.setenv("VLLM_P2P_IDLE_TIMEOUT_S", "10") + monkeypatch.setenv("VLLM_P2P_MAX_PEERS", "1") + import vllm.envs + + vllm.envs.__dict__.pop("VLLM_P2P_IDLE_TIMEOUT_S", None) + vllm.envs.__dict__.pop("VLLM_P2P_MAX_PEERS", None) + + mgr = self._make_with_fake_data() + idle_session = _FakeSession( + peer_id="idle:1", + last_activity_at=time.monotonic() - 20, + ) + mgr._sessions["idle:1"] = idle_session # type: ignore[assignment] + + mgr._reap_idle_sessions() + assert "idle:1" not in mgr._sessions + + def test_idle_session_kept_below_headroom(self, monkeypatch): + monkeypatch.setenv("VLLM_P2P_IDLE_TIMEOUT_S", "10") + import vllm.envs + + vllm.envs.__dict__.pop("VLLM_P2P_IDLE_TIMEOUT_S", None) + vllm.envs.__dict__.pop("VLLM_P2P_MAX_PEERS", None) + + mgr = self._make_with_fake_data() + idle_session = _FakeSession( + peer_id="idle:1", + last_activity_at=time.monotonic() - 20, + ) + mgr._sessions["idle:1"] = idle_session # type: ignore[assignment] + + mgr._reap_idle_sessions() + assert "idle:1" in mgr._sessions + + def test_active_session_kept(self, monkeypatch): + monkeypatch.setenv("VLLM_P2P_IDLE_TIMEOUT_S", "10") + monkeypatch.setenv("VLLM_P2P_MAX_PEERS", "1") + import vllm.envs + + vllm.envs.__dict__.pop("VLLM_P2P_IDLE_TIMEOUT_S", None) + vllm.envs.__dict__.pop("VLLM_P2P_MAX_PEERS", None) + + mgr = self._make_with_fake_data() + active = _FakeSession( + peer_id="active:1", + last_activity_at=time.monotonic(), + ) + mgr._sessions["active:1"] = active # type: ignore[assignment] + + mgr._reap_idle_sessions() + assert "active:1" in mgr._sessions + + def test_session_with_pending_work_not_evicted(self, monkeypatch): + monkeypatch.setenv("VLLM_P2P_IDLE_TIMEOUT_S", "1") + monkeypatch.setenv("VLLM_P2P_MAX_PEERS", "1") + import vllm.envs + + vllm.envs.__dict__.pop("VLLM_P2P_IDLE_TIMEOUT_S", None) + vllm.envs.__dict__.pop("VLLM_P2P_MAX_PEERS", None) + + mgr = self._make_with_fake_data() + busy = _FakeSession( + peer_id="busy:1", + last_activity_at=time.monotonic() - 100, + ) + busy._server._inflight[1] = object() + mgr._sessions["busy:1"] = busy # type: ignore[assignment] + + mgr._reap_idle_sessions() + assert "busy:1" in mgr._sessions + + def test_disabled_when_zero(self, monkeypatch): + monkeypatch.setenv("VLLM_P2P_IDLE_TIMEOUT_S", "0") + import vllm.envs + + vllm.envs.__dict__.pop("VLLM_P2P_IDLE_TIMEOUT_S", None) + + mgr = self._make_with_fake_data() + old = _FakeSession( + peer_id="old:1", + last_activity_at=time.monotonic() - 999999, + ) + mgr._sessions["old:1"] = old # type: ignore[assignment] + + mgr._reap_idle_sessions() + assert "old:1" in mgr._sessions diff --git a/tests/v1/kv_offload/tiering/p2p/test_sessions.py b/tests/v1/kv_offload/tiering/p2p/test_sessions.py index 09739345be..300cbacd66 100644 --- a/tests/v1/kv_offload/tiering/p2p/test_sessions.py +++ b/tests/v1/kv_offload/tiering/p2p/test_sessions.py @@ -2698,3 +2698,141 @@ class TestTransferDoneMsgValidation: } with pytest.raises(ValueError, match="success"): TransferDoneMsg.validate(msg) + + +# --------------------------------------------------------------------------- +# Tests for handshake deadline +# --------------------------------------------------------------------------- + + +class TestHandshakeDeadline: + """A connected-but-not-ready session is marked dead after the + handshake timeout expires.""" + + def test_marks_dead_after_timeout(self, monkeypatch): + monkeypatch.setenv("VLLM_P2P_HANDSHAKE_TIMEOUT_S", "1") + import vllm.envs + + vllm.envs.__dict__.pop("VLLM_P2P_HANDSHAKE_TIMEOUT_S", None) + + transport = FakeDataTransport() + conn = FakeConnection(peer_id="stale:9999") + session = P2PSession( + peer_id="stale:9999", + local_id="local:1", + transport=transport, + local_block_len=4096, + local_hash_seed=_DEFAULT_HASH_SEED, + conn=conn, + ) + + assert session.connected + assert not session.ready + assert session.alive + + session._connected_at = time.monotonic() - 5 + + session.poll() + assert not session.alive + + def test_no_timeout_when_ready(self, monkeypatch): + monkeypatch.setenv("VLLM_P2P_HANDSHAKE_TIMEOUT_S", "1") + import vllm.envs + + vllm.envs.__dict__.pop("VLLM_P2P_HANDSHAKE_TIMEOUT_S", None) + + transport = FakeDataTransport() + conn = FakeConnection(peer_id="good:1234") + session = P2PSession( + peer_id="good:1234", + local_id="local:1", + transport=transport, + local_block_len=4096, + local_hash_seed=_DEFAULT_HASH_SEED, + conn=conn, + ) + + # Complete the handshake by feeding ConnectMsg + ConnectAckMsg. + conn._inbox = [ + { + TYPE_KEY: ConnectMsg.TYPE, + ConnectMsg.PEER_ID: "good:1234", + ConnectMsg.AGENT_METADATA: b"meta", + ConnectMsg.BASE_ADDR: 0x1000, + ConnectMsg.NUM_BLOCKS: 16, + ConnectMsg.BLOCK_LEN: 4096, + ConnectMsg.CONFIG_FINGERPRINT: "", + ConnectMsg.HASH_SEED: _DEFAULT_HASH_SEED, + }, + {TYPE_KEY: ConnectAckMsg.TYPE, ConnectAckMsg.PEER_ID: "good:1234"}, + ] + session.poll() + assert session.ready + + session._connected_at = time.monotonic() - 100 + session.poll() + assert session.alive + + def test_buffered_ack_at_deadline_completes_handshake(self, monkeypatch): + """A ConnectAck already in the inbox at the deadline must complete + the handshake instead of being discarded by mark_dead.""" + monkeypatch.setenv("VLLM_P2P_HANDSHAKE_TIMEOUT_S", "1") + import vllm.envs + + vllm.envs.__dict__.pop("VLLM_P2P_HANDSHAKE_TIMEOUT_S", None) + + transport = FakeDataTransport() + conn = FakeConnection(peer_id="late-ack:1") + session = P2PSession( + peer_id="late-ack:1", + local_id="local:1", + transport=transport, + local_block_len=4096, + local_hash_seed=_DEFAULT_HASH_SEED, + conn=conn, + ) + session._connected_at = time.monotonic() - 5 + conn._inbox = [ + { + TYPE_KEY: ConnectMsg.TYPE, + ConnectMsg.PEER_ID: "late-ack:1", + ConnectMsg.AGENT_METADATA: b"meta", + ConnectMsg.BASE_ADDR: 0x1000, + ConnectMsg.NUM_BLOCKS: 16, + ConnectMsg.BLOCK_LEN: 4096, + ConnectMsg.CONFIG_FINGERPRINT: "", + ConnectMsg.HASH_SEED: _DEFAULT_HASH_SEED, + }, + {TYPE_KEY: ConnectAckMsg.TYPE, ConnectAckMsg.PEER_ID: "late-ack:1"}, + ] + session.poll() + assert session.ready + assert session.alive + + def test_activity_timestamp_updated_on_recv(self): + transport = FakeDataTransport() + conn = FakeConnection(peer_id="peer:1") + session = P2PSession( + peer_id="peer:1", + local_id="local:1", + transport=transport, + local_block_len=4096, + local_hash_seed=_DEFAULT_HASH_SEED, + conn=conn, + ) + before = session._last_activity_at + + conn._inbox = [ + { + TYPE_KEY: ConnectMsg.TYPE, + ConnectMsg.PEER_ID: "peer:1", + ConnectMsg.AGENT_METADATA: b"meta", + ConnectMsg.BASE_ADDR: 0x1000, + ConnectMsg.NUM_BLOCKS: 16, + ConnectMsg.BLOCK_LEN: 4096, + ConnectMsg.CONFIG_FINGERPRINT: "", + ConnectMsg.HASH_SEED: _DEFAULT_HASH_SEED, + }, + ] + session.poll() + assert session._last_activity_at >= before diff --git a/tests/v1/kv_offload/tiering/p2p/test_zmq_transport.py b/tests/v1/kv_offload/tiering/p2p/test_zmq_transport.py index f814a04887..90e87bb407 100644 --- a/tests/v1/kv_offload/tiering/p2p/test_zmq_transport.py +++ b/tests/v1/kv_offload/tiering/p2p/test_zmq_transport.py @@ -336,3 +336,47 @@ class TestZmqReconnect: finally: transport_a.close() transport_b.close() + + +class TestExpandedMonitorEvents: + """The monitor mask includes EVENT_DISCONNECTED and EVENT_CLOSED so + both are treated as fatal.""" + + def test_fatal_events_include_disconnected_and_closed(self): + from vllm.v1.kv_offload.tiering.p2p.control.zmq import ( + _FATAL_MONITOR_EVENTS, + ) + + assert zmq.EVENT_DISCONNECTED in _FATAL_MONITOR_EVENTS + assert zmq.EVENT_CLOSED in _FATAL_MONITOR_EVENTS + + def test_monitor_mask_covers_fatal_events(self): + from vllm.v1.kv_offload.tiering.p2p.control.zmq import ( + _FATAL_MONITOR_EVENTS, + _MONITOR_EVENT_MASK, + ) + + for ev in _FATAL_MONITOR_EVENTS: + assert _MONITOR_EVENT_MASK & ev == ev + + def test_disconnect_detected_after_peer_shutdown(self): + """When a peer's transport closes, the outbound connection is + eventually marked dead by the monitor.""" + transport_a, port_a = _make_transport() + transport_b, _ = _make_transport() + + try: + conn_out = transport_b.connect(f"127.0.0.1:{port_a}") + conn_out.send({"type": "hello"}) + _wait_for_inbound(transport_a) + + transport_a.close() + + end = time.monotonic() + 15 + while conn_out.alive and time.monotonic() < end: + transport_b.poll() + time.sleep(0.05) + + assert not conn_out.alive + finally: + transport_b.close() diff --git a/vllm/envs.py b/vllm/envs.py index 02c282fd8e..14489a3f25 100755 --- a/vllm/envs.py +++ b/vllm/envs.py @@ -233,6 +233,9 @@ if TYPE_CHECKING: VLLM_NIXL_SIDE_CHANNEL_PORT: int = 5600 VLLM_P2P_SIDE_CHANNEL_HOST: str = "localhost" VLLM_P2P_SIDE_CHANNEL_PORT: int = 5710 + VLLM_P2P_MAX_PEERS: int = 128 + VLLM_P2P_HANDSHAKE_TIMEOUT_S: int = 30 + VLLM_P2P_IDLE_TIMEOUT_S: int = 300 VLLM_EC_SIDE_CHANNEL_HOST: str = "localhost" VLLM_EC_SIDE_CHANNEL_PORT: int = 5601 VLLM_MOONCAKE_BOOTSTRAP_PORT: int = 8998 @@ -1739,6 +1742,21 @@ environment_variables: dict[str, Callable[[], Any]] = { "VLLM_P2P_SIDE_CHANNEL_PORT": lambda: int( os.getenv("VLLM_P2P_SIDE_CHANNEL_PORT", "5710") ), + # Maximum number of concurrent P2P peer sessions. Limits ZMQ socket + # consumption to prevent denial-of-service via unbounded peer + # creation from client-supplied kv_transfer_params. Default: 128. + "VLLM_P2P_MAX_PEERS": lambda: int(os.getenv("VLLM_P2P_MAX_PEERS", "128")), + # Seconds to wait for a P2P peer handshake to complete (connected + # but not yet ready). Sessions exceeding this deadline are reaped. + # Prevents never-responding peers from accumulating indefinitely. + "VLLM_P2P_HANDSHAKE_TIMEOUT_S": lambda: int( + os.getenv("VLLM_P2P_HANDSHAKE_TIMEOUT_S", "30") + ), + # Seconds a P2P session may remain idle (no pending work, no + # messages) before it is eligible for eviction. Idle eviction runs + # only when remaining peer slots are within 8 of VLLM_P2P_MAX_PEERS. + # Default: 300. Set to 0 to disable. + "VLLM_P2P_IDLE_TIMEOUT_S": lambda: int(os.getenv("VLLM_P2P_IDLE_TIMEOUT_S", "300")), # IP address used for the EC connector's ZMQ side channel # (producer ROUTER bind, consumer DEALER dial). "VLLM_EC_SIDE_CHANNEL_HOST": lambda: os.getenv( diff --git a/vllm/v1/kv_offload/tiering/p2p/control/zmq.py b/vllm/v1/kv_offload/tiering/p2p/control/zmq.py index b28ed76dac..074c3869f4 100644 --- a/vllm/v1/kv_offload/tiering/p2p/control/zmq.py +++ b/vllm/v1/kv_offload/tiering/p2p/control/zmq.py @@ -27,11 +27,35 @@ _HEARTBEAT_IVL_MS = 2000 _HEARTBEAT_TIMEOUT_MS = 10000 _HEARTBEAT_TTL_MS = 10000 +# Events that indicate a fatal connection failure — the peer will never +# become reachable through this socket. +_FATAL_MONITOR_EVENTS: frozenset[int] = frozenset() # populated after import + # Shared sentinels returned when there is nothing to report. _EMPTY_INBOX: tuple[dict, ...] = () _EMPTY_NEW_CONNECTIONS: tuple[ControlConnection, ...] = () +def _build_fatal_events() -> frozenset[int]: + events = {zmq.EVENT_DISCONNECTED, zmq.EVENT_CLOSED} + for attr in ( + "EVENT_HANDSHAKE_FAILED_AUTH", + "EVENT_HANDSHAKE_FAILED_PROTOCOL", + "EVENT_HANDSHAKE_FAILED_NO_DETAIL", + ): + val = getattr(zmq, attr, None) + if val is not None: + events.add(val) + return frozenset(events) + + +_FATAL_MONITOR_EVENTS = _build_fatal_events() + +_MONITOR_EVENT_MASK: int = 0 +for _ev in _FATAL_MONITOR_EVENTS: + _MONITOR_EVENT_MASK |= _ev + + def _tcp_addr(host: str, port: int | str) -> str: return f"tcp://{host}:{port}" @@ -247,7 +271,7 @@ class ZmqTransport(ControlTransport): safe_id = peer_id.replace(":", "-").replace("/", "-") monitor_addr = f"inproc://p2p-monitor-{safe_id}-{self._monitor_seq}" self._monitor_seq += 1 - dealer.monitor(monitor_addr, zmq.EVENT_DISCONNECTED) + dealer.monitor(monitor_addr, _MONITOR_EVENT_MASK) monitor_sock = self._zmq_ctx.socket(zmq.PAIR) monitor_sock.connect(monitor_addr) @@ -319,29 +343,33 @@ class ZmqTransport(ControlTransport): self._pending_inbound.append((sender_id, msg)) def _check_monitors(self) -> None: - """Non-blocking: check all monitor sockets for disconnection.""" + """Non-blocking: check all monitor sockets for fatal events.""" for conn in self._connections.values(): if not conn.alive: continue - try: - event = zmq.utils.monitor.recv_monitor_message( - conn.monitor_socket, zmq.NOBLOCK - ) - except zmq.Again: - continue - except zmq.ZMQError as exc: - logger.warning( - "ZmqTransport %s: monitor error for peer %s: %s", - self._local_id, - conn.peer_id, - exc, - ) - continue + while True: + try: + event = zmq.utils.monitor.recv_monitor_message( + conn.monitor_socket, zmq.NOBLOCK + ) + except zmq.Again: + break + except zmq.ZMQError as exc: + logger.warning( + "ZmqTransport %s: monitor error for peer %s: %s", + self._local_id, + conn.peer_id, + exc, + ) + break - if event["event"] == zmq.EVENT_DISCONNECTED: - logger.debug( - "ZmqTransport %s: peer %s disconnected", - self._local_id, - conn.peer_id, - ) - conn.mark_dead() + ev = event["event"] + if ev in _FATAL_MONITOR_EVENTS: + logger.debug( + "ZmqTransport %s: peer %s fatal event 0x%x", + self._local_id, + conn.peer_id, + ev, + ) + conn.mark_dead() + break diff --git a/vllm/v1/kv_offload/tiering/p2p/manager.py b/vllm/v1/kv_offload/tiering/p2p/manager.py index f759a4481e..b7268eb420 100644 --- a/vllm/v1/kv_offload/tiering/p2p/manager.py +++ b/vllm/v1/kv_offload/tiering/p2p/manager.py @@ -69,6 +69,11 @@ _SHUTDOWN_DRAIN_TIMEOUT_S = 3.0 # enough to avoid busy-spinning the scheduler thread. _DRAIN_SLEEP_S = 0.001 +# Idle sessions are only reaped when remaining capacity is at most this +# many slots. Session setup is expensive; below this headroom, keep idle +# peers so a later burst does not pay handshake again. +_IDLE_REAP_HEADROOM = 8 + def _remote_prefiller_params(kv_params: dict | None) -> dict | None: """Return the ``remote_prefiller`` sub-dict, or None if absent. @@ -428,7 +433,9 @@ class P2PSecondaryTierManager(SecondaryTierManager): _annotate_req_context(req_context) source = req_context.get_state(P2PSourceInfo) if source is not None: - self._get_or_create_session(source.peer_id) + 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) @@ -685,38 +692,78 @@ class P2PSecondaryTierManager(SecondaryTierManager): self._hash_seed = get_none_hash_seed() return self._hash_seed - def _get_or_create_session(self, peer_id: str) -> P2PSession: + 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 + (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 - 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, - ) + + 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: @@ -827,6 +874,53 @@ class P2PSecondaryTierManager(SecondaryTierManager): 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 # ------------------------------------------------------------------ @@ -893,6 +987,7 @@ class P2PSecondaryTierManager(SecondaryTierManager): self._reap_dead_sessions() self._reap_unbound_stores() + self._reap_idle_sessions() # ------------------------------------------------------------------ # Lifecycle diff --git a/vllm/v1/kv_offload/tiering/p2p/session/session.py b/vllm/v1/kv_offload/tiering/p2p/session/session.py index f43e1a5aab..ef490304c9 100644 --- a/vllm/v1/kv_offload/tiering/p2p/session/session.py +++ b/vllm/v1/kv_offload/tiering/p2p/session/session.py @@ -18,9 +18,11 @@ received our ConnectMsg, after which queued outgoing messages are flushed. from __future__ import annotations import contextlib +import time from collections.abc import Sequence from typing import TYPE_CHECKING, NamedTuple +import vllm.envs as envs from vllm.logger import init_logger from vllm.v1.kv_offload.base import OffloadKey from vllm.v1.kv_offload.tiering.p2p.control.base import ControlConnection @@ -128,6 +130,10 @@ class P2PSession: # Consecutive non-protocol dispatch errors. Reset on success. self._dispatch_error_count: int = 0 + now = time.monotonic() + self._connected_at: float | None = None + self._last_activity_at: float = now + # kv_request_ids whose FetchMsg arrived during the current poll # tick. Drained and returned in the next poll() result so the # manager can bind kv_request_id → session and replay any @@ -168,6 +174,11 @@ class P2PSession: """True while inbound loads or outbound transfers are outstanding.""" return self._client.has_active_loads or self._server.has_inflight_transfers + @property + def last_activity_at(self) -> float: + """Monotonic timestamp of the last meaningful session activity.""" + return self._last_activity_at + # ------------------------------------------------------------------ # Connection lifecycle # ------------------------------------------------------------------ @@ -181,6 +192,9 @@ class P2PSession: if self._conn is not None: raise ValueError(f"P2PSession {self.peer_id}: already connected") self._conn = conn + now = time.monotonic() + self._connected_at = now + self._last_activity_at = now self._send_connect() # ------------------------------------------------------------------ @@ -249,17 +263,39 @@ class P2PSession: def poll(self) -> SessionPollResult: """Process incoming messages, drive transfers, apply timeouts.""" if self._conn is None: - # Pending session — store-job timeouts still apply so buffered - # jobs that never get picked up are surfaced as failures. return SessionPollResult( loads=[], stores=self._server.collect_idle_timeouts(), new_fetch_ids=[], ) - for msg in self._conn.recv(): + # Drain the inbox first so a ConnectAck already buffered at the + # deadline (e.g. after a scheduler stall) can complete the + # handshake instead of being discarded by mark_dead. + messages = self._conn.recv() + if messages: + self._last_activity_at = time.monotonic() + for msg in messages: self._on_message(msg) + # Handshake deadline: a peer that never completes the handshake + # is stuck in connected-but-not-ready state. Mark it dead so the + # manager reaps it instead of accumulating sockets indefinitely. + if ( + not self._send_ready + and self._conn.alive + and self._connected_at is not None + and time.monotonic() - self._connected_at + > envs.VLLM_P2P_HANDSHAKE_TIMEOUT_S + ): + logger.warning( + "P2PSession %s: handshake timed out after %ds " + "— marking connection dead", + self.peer_id, + envs.VLLM_P2P_HANDSHAKE_TIMEOUT_S, + ) + self._conn.mark_dead() + loads = self._client.collect_results() stores = self._server.collect_results() self._server.drain_pending_aborts() @@ -533,6 +569,7 @@ class P2PSession: return try: self._conn.send(msg) + self._last_activity_at = time.monotonic() logger.debug("P2PSession %s: sent %s", self.peer_id, msg.get(TYPE_KEY)) except Exception: # A send failure means the connection is broken. Swallowing it