[Bugfix][KV Offload][P2P] Support IPv6 literals in the ZMQ control channel

docs/features/kv_offloading_usage.md

@@ -342,6 +342,11 @@ Rather than embedding `host`/`port` in each `secondary_tiers` entry, set them on
 - `VLLM_P2P_SIDE_CHANNEL_HOST` (default `localhost`): address the P2P control socket binds to. It is used **verbatim** as both the bind address and the identity peers dial back — there is no auto-detection (this mirrors `VLLM_NIXL_SIDE_CHANNEL_HOST`). The default binds the loopback interface only, so peers on another host cannot reach it. **For any cross-host P2P deployment you must set this explicitly to the node's routable IP** (e.g. the pod IP) before launching `vllm serve` — otherwise remote peers will fail to connect. The NIXL agent name is a separate per-process identifier, so peers sharing a `host:port` never collide.
 - `VLLM_P2P_SIDE_CHANNEL_PORT` (default `5710`): base port for the P2P control socket. The port actually bound is `VLLM_P2P_SIDE_CHANNEL_PORT + data_parallel_index` — one socket per DP replica, matching NIXL (for DP=1 the offset is 0). The peer's port is passed as `remote_port` in `kv_transfer_params`; the router/EPP that selects the DP rank (e.g. via the `X-data-parallel-rank` header) computes `remote_port = base + rank`. The DP-index offset separates replicas *within* one deployment; two co-located *deployments* (a prefiller and a decoder on the same host) still need distinct base ports (e.g. decoder base `5711`) to avoid a bind collision.
 
+IPv6 literals are also supported for the control channel, in `host`,
+`VLLM_P2P_SIDE_CHANNEL_HOST`, and the peer's `remote_host`. Write them without
+brackets (for example, `::1` for local testing); a bracketed literal such as
+`[::1]` is accepted too and normalized. Supply the port separately.
+
 #### Orchestration-Layer Protocol
 
 The P2P tier does not decide *which* peer to pull from — that is the orchestration layer's job (the router/EPP and its scheduler). The orchestrator drives every transfer through a request's `kv_transfer_params` dict: it picks the request's role, allocates a unique transaction ID, and supplies the remote peer's address. All block lookup, hash matching, and NIXL transfer happen at the tier level below; the orchestrator only sets the correct role keys and enforces the allowed combinations.

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

@@ -24,10 +24,12 @@
 _WEDGED_TRANSPORTS: list[ZmqTransport] = []
 
 
-def _free_port() -> int:
+def _free_port(host: str = "127.0.0.1") -> int:
     """Find a free TCP port."""
-    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-        s.bind(("127.0.0.1", 0))
+    host = host.removeprefix("[").removesuffix("]")
+    family = socket.AF_INET6 if ":" in host else socket.AF_INET
+    with socket.socket(family, socket.SOCK_STREAM) as s:
+        s.bind((host, 0))
         return s.getsockname()[1]
 
 
@@ -40,7 +42,7 @@ def _make_transport(host: str = "127.0.0.1", attempts: int = 8):
     """
     last_err: Exception | None = None
     for _ in range(attempts):
-        port = _free_port()
+        port = _free_port(host)
         try:
             return ZmqTransport(f"{host}:{port}", host, port), port
         except (zmq.ZMQError, OSError) as e:
@@ -147,18 +149,24 @@ def test_connect_and_send_message(self):
             transport_a.close()
             transport_b.close()
 
-    def test_bidirectional_messaging(self):
+    @pytest.mark.parametrize("host_a", ["127.0.0.1", "::1"])
+    @pytest.mark.parametrize("host_b", ["127.0.0.1", "::1"])
+    def test_bidirectional_messaging(self, host_a, host_b):
         """Both sides can send and receive after connection."""
-        transport_a, port_a = _make_transport()
-        transport_b, _ = _make_transport()
+        transport_a, port_a = _make_transport(host_a)
+        transport_b, port_b = _make_transport(host_b)
 
         try:
-            conn_b = transport_b.connect(f"127.0.0.1:{port_a}")
+            conn_b = transport_b.connect(f"{host_a}:{port_a}")
             conn_b.send({"type": "connect", "from": "b"})
 
             new_conns = _wait_for_inbound(transport_a)
             assert len(new_conns) == 1
             conn_a = new_conns[0]
+            assert conn_a.peer_id == f"{host_b}:{port_b}"
+            assert _wait_for_messages(transport_a, conn_a, 1) == [
+                {"type": "connect", "from": "b"}
+            ]
 
             conn_a.send({"type": "reply", "from": "a"})
 
@@ -168,6 +176,31 @@ def test_bidirectional_messaging(self):
             transport_a.close()
             transport_b.close()
 
+    def test_bracketed_ipv6_host_is_accepted(self):
+        """A bracketed IPv6 literal binds and dials like the bare form."""
+        transport_a, port_a = _make_transport("::1")
+        transport_b, port_b = _make_transport("[::1]")
+
+        try:
+            conn_b = transport_b.connect(f"::1:{port_a}")
+            conn_b.send({"type": "connect", "from": "b"})
+
+            new_conns = _wait_for_inbound(transport_a)
+            assert len(new_conns) == 1
+            conn_a = new_conns[0]
+            assert conn_a.peer_id == f"[::1]:{port_b}"
+            assert _wait_for_messages(transport_a, conn_a, 1) == [
+                {"type": "connect", "from": "b"}
+            ]
+
+            conn_a.send({"type": "reply", "from": "a"})
+            assert _wait_for_messages(transport_b, conn_b, 1) == [
+                {"type": "reply", "from": "a"}
+            ]
+        finally:
+            transport_a.close()
+            transport_b.close()
+
     def test_poll_returns_empty_when_no_connections(self):
         transport, _ = _make_transport()
         try:

vllm/v1/kv_offload/tiering/p2p/control/zmq.py

@@ -16,6 +16,7 @@
 import zmq.utils.monitor
 
 from vllm.logger import init_logger
+from vllm.utils.network_utils import is_valid_ipv6_address, make_zmq_path
 from vllm.v1.kv_offload.tiering.p2p.control.base import (
     ControlConnection,
     ControlTransport,
@@ -66,8 +67,15 @@ def _build_fatal_events() -> frozenset[int]:
     _MONITOR_EVENT_MASK |= _ev
 
 
+def _normalize_host(host: str) -> str:
+    """Strip the brackets from a bracketed IPv6 literal, if present."""
+    if host.startswith("[") and host.endswith("]"):
+        return host[1:-1]
+    return host
+
+
 def _tcp_addr(host: str, port: int | str) -> str:
-    return f"tcp://{host}:{port}"
+    return make_zmq_path("tcp", host, int(port))
 
 
 def _apply_heartbeat(sock: zmq.Socket) -> None:
@@ -167,6 +175,7 @@ class ZmqTransport(ControlTransport):
     def __init__(self, local_id: str, host: str, port: int) -> None:
         self._local_id = local_id
         self._closed = False
+        host = _normalize_host(host)
 
         self._connections: dict[str, ZmqConnection] = {}
         self._pending_inbound: list[tuple[str, dict]] = []
@@ -177,6 +186,8 @@ def __init__(self, local_id: str, host: str, port: int) -> None:
         self._zmq_ctx = zmq.Context()
         self._router: zmq.Socket = self._zmq_ctx.socket(zmq.ROUTER)
         _apply_heartbeat(self._router)
+        if is_valid_ipv6_address(host):
+            self._router.setsockopt(zmq.IPV6, 1)
         bind_addr = _tcp_addr(host, port)
         self._router.bind(bind_addr)
         logger.info("ZmqTransport %s: ROUTER bound on %s", self._local_id, bind_addr)
@@ -274,6 +285,7 @@ def _open_connection(
     ) -> ZmqConnection:
         """Create a DEALER socket + monitor and register the connection."""
         host, port_str = peer_id.rsplit(":", 1)
+        host = _normalize_host(host)
         dealer_addr = _tcp_addr(host, port_str)
 
         logger.debug(
@@ -286,6 +298,8 @@ def _open_connection(
 
         dealer = self._zmq_ctx.socket(zmq.DEALER)
         _apply_heartbeat(dealer)
+        if is_valid_ipv6_address(host):
+            dealer.setsockopt(zmq.IPV6, 1)
         dealer.identity = self._local_id.encode()
         # ZMQ_IMMEDIATE stays 0: messages queue before the TCP connect
         # completes, so the ConnectMsg sent right after connect() fits.