[Bugfix][EC Connector] Allocate producer control ports at startup

docs/features/ec_cpu_connector.md

@@ -71,7 +71,7 @@ vllm serve <model> --ec-transfer-config '{
 
 ### Protocol
 
-- **Control plane**: ZMQ. The producer binds a `ROUTER` socket on `VLLM_EC_SIDE_CHANNEL_HOST:VLLM_EC_SIDE_CHANNEL_PORT`; each consumer opens one `DEALER` connection per producer peer, with ZMQ heartbeating (2s interval, 4s timeout, 8s TTL) to detect a dead peer. `XferReq`/`XferAck` are `msgspec` msgpack structs, versioned by `EC_CONNECTOR_VERSION` (currently `1`) — a version mismatch is NACKed.
+- **Control plane**: ZMQ. By default, each producer binds a `ROUTER` socket on the local IP returned by vLLM's `get_ip()` with an OS-assigned port and advertises the bound port through `ec_transfer_params.peer_port`. Optional host and port overrides are listed below. Each consumer opens one `DEALER` connection per producer peer, with ZMQ heartbeating (2s interval, 4s timeout, 8s TTL) to detect a dead peer. `XferReq`/`XferAck` are `msgspec` msgpack structs, versioned by `EC_CONNECTOR_VERSION` (currently `1`) — a version mismatch is NACKed.
 - **Compatibility check**: every `XferReq` carries a SHA-256 hash over `(vllm_version, model, dtype, block_size_bytes)`; the producer NACKs (`NACK_INCOMPAT`) any peer whose hash differs.
 - **Ack statuses**: `OK`, `NACK_MISSING` (the producer no longer holds the encoding), `NACK_NOT_READY` (held, but its save has not landed), `NACK_INCOMPAT`, `NACK_VERSION`, `NACK_INTERNAL`. Only `NACK_NOT_READY` is retryable — `RETRYABLE_NACKS` in `cpu/protocol.py` is what both ends consult, so the classification lives with the wire vocabulary rather than at each call site.
 - **Data plane**: NIXL, `UCX` backend, consumer-initiated `READ` — the consumer pulls bytes directly out of the producer's registered mmap region; the producer never pushes.
@@ -98,14 +98,14 @@ EC transfer is configured via `--ec-transfer-config` (CLI) or the `ec_transfer_c
 | `consumer_ack_timeout_s` | `float` | No (default `2.0`) | How long a consumer waits for an `XferAck` before giving up on a read. The producer answers `XferReq`s from its own scheduler step, so its reply latency scales with the encoder's `--max-num-batched-tokens`: a loaded encoder whose steps run longer than this makes consumers abandon reads the producer is about to grant. Raise it for large encoder batches. |
 | `ec_cpu_bytes` | `int` | Yes | Total size, in bytes, of the shared CPU mmap region. `ECCPUConnector` raises `ValueError` if unset. Block count = `ec_cpu_bytes // block_size_bytes`, where each block holds one LLM input-embedding vector. Entries allocate enough blocks for their modality-specific output shape, including visual DeepStack features. |
 
-### Environment Variables
+The control address uses vLLM's standard IP discovery. Set `VLLM_HOST_IP` to select a specific network interface when automatic discovery is unsuitable.
 
-| Variable | Default | Description |
+| Environment variable | Default | Description |
 | --- | --- | --- |
-| `VLLM_EC_SIDE_CHANNEL_HOST` | `localhost` | Host the producer's ZMQ `ROUTER` socket binds to. Set to a routable address (e.g. the pod IP) for multi-instance/multi-node P2P — the default only works when producer and consumer share a host. |
-| `VLLM_EC_SIDE_CHANNEL_PORT` | `5601` | Port for the same ZMQ `ROUTER` socket. |
+| `VLLM_EC_SIDE_CHANNEL_HOST` | Unset | Override the EC control host instead of using `get_ip()`. Must be reachable by consumers; wildcard addresses are rejected. |
+| `VLLM_EC_SIDE_CHANNEL_PORT` | `0` | `0` selects an OS-assigned port per producer. A nonzero value binds `base_port + data_parallel_index`; an occupied port fails startup. |
 
-Both are read only when `ec_enable_nixl=True` on a producer (`ec_role="ec_producer"` or `"ec_both"`).
+For fixed ports, assign non-overlapping ranges to independent instances sharing a host. Consumers use the advertised address and do not need matching port configuration.
 
 ## Limitations
 

tests/v1/ec_connector/integration/test_nixl_failure.py

@@ -30,7 +30,6 @@
 )
 from vllm.multimodal.utils import encode_image_url
 from vllm.platforms import current_platform
-from vllm.utils.network_utils import get_open_port
 
 MODEL = os.environ.get("MODEL", "Qwen/Qwen2.5-VL-3B-Instruct")
 
@@ -77,11 +76,7 @@ def config(role):
                 "--ec-transfer-config",
                 config("ec_producer"),
             ],
-            env_dict={
-                "CUDA_VISIBLE_DEVICES": devices[0],
-                "VLLM_EC_SIDE_CHANNEL_HOST": "127.0.0.1",
-                "VLLM_EC_SIDE_CHANNEL_PORT": str(get_open_port()),
-            },
+            env_dict={"CUDA_VISIBLE_DEVICES": devices[0]},
         ) as encoder,
         RemoteOpenAIServer(
             MODEL,

tests/v1/ec_connector/unit/test_control.py

@@ -70,6 +70,7 @@ def test_close_sets_alive_false():
 from vllm.distributed.ec_transfer.ec_connector.cpu.control.zmq import (  # noqa: E402
     ZmqClientConnection,
     ZmqClientTransport,
+    ZmqServerTransport,
 )
 from vllm.distributed.ec_transfer.ec_connector.cpu.utils import PeerAddr  # noqa: E402
 
@@ -283,3 +284,38 @@ def test_zmq_client_transport_close_closes_all_connections(
     dealer_a.close.assert_called()
     dealer_b.close.assert_called()
     assert transport._connections == {}
+
+
+def test_zmq_servers_allocate_distinct_reachable_ports():
+    """Simultaneous producers must publish usable ports without configuration."""
+    servers = []
+    client = ZmqClientTransport()
+    try:
+        for _ in range(2):
+            servers.append(ZmqServerTransport(host="127.0.0.1"))
+        assert len({server.port for server in servers}) == 2
+        for server in servers:
+            assert 0 < server.port <= 65535
+            conn = client.connect(("127.0.0.1", server.port))
+            conn.send(b"request")
+            messages = server.poll(timeout_ms=5000)
+            assert len(messages) == 1
+            identity, payload = messages[0]
+            assert payload == b"request"
+            server.send(identity, b"reply")
+            assert conn.dealer.poll(timeout=5000)
+            assert conn.recv() == [b"reply"]
+    finally:
+        client.close()
+        for server in servers:
+            server.close()
+
+
+def test_zmq_server_rejects_an_occupied_explicit_port():
+    server = ZmqServerTransport(host="127.0.0.1")
+    try:
+        with pytest.raises(zmq.ZMQError) as exc:
+            ZmqServerTransport(host="127.0.0.1", port=server.port)
+        assert exc.value.errno == zmq.EADDRINUSE
+    finally:
+        server.close()

tests/v1/ec_connector/unit/test_scheduler_nixl_ctor.py

@@ -14,6 +14,7 @@
 )
 from vllm.distributed.ec_transfer.ec_connector.cpu.scheduler import ECCPUScheduler
 from vllm.distributed.nixl_utils import NixlWrapper
+from vllm.utils.network_utils import get_open_port
 
 _N = 16
 _BS = 64
@@ -126,21 +127,73 @@ def fail_control_transport():
 
 
 @pytest.mark.skipif(NixlWrapper is None, reason="Requires NIXL package")
-def test_gate_on_wires_data_transport_and_producer_session(monkeypatch):
-    # Port 0 lets the OS pick an ephemeral port so the real ZMQ ROUTER bind
-    # in ProducerSession.start() cannot collide with another test/process.
-    monkeypatch.setenv("VLLM_EC_SIDE_CHANNEL_HOST", "127.0.0.1")
-    monkeypatch.setenv("VLLM_EC_SIDE_CHANNEL_PORT", "0")
+@pytest.mark.parametrize("dp_index", [0, 1])
+@pytest.mark.parametrize("port_mode", ["unset", "zero", "fixed"])
+def test_gate_on_wires_data_transport_and_producer_session(
+    monkeypatch, dp_index, port_mode
+):
+    monkeypatch.setenv("VLLM_HOST_IP", "127.0.0.1")
+    monkeypatch.delenv("VLLM_EC_SIDE_CHANNEL_HOST", raising=False)
+    monkeypatch.delenv("VLLM_EC_SIDE_CHANNEL_PORT", raising=False)
+    port = 0
+    if port_mode == "fixed":
+        port = get_open_port()
+        monkeypatch.setenv("VLLM_EC_SIDE_CHANNEL_PORT", str(port - dp_index))
+        monkeypatch.setenv("VLLM_EC_SIDE_CHANNEL_HOST", "localhost")
+    elif port_mode == "zero":
+        monkeypatch.setenv("VLLM_EC_SIDE_CHANNEL_PORT", "0")
     monkeypatch.setattr(sched_mod, "create_ec_shared_region", lambda cfg: _region())
     cfg = create_ec_vllm_config(ec_role="ec_both")
+    cfg.parallel_config.data_parallel_index = dp_index
     cfg.ec_transfer_config.ec_connector_extra_config["ec_enable_nixl"] = True
     s = ECCPUScheduler(cfg)
     try:
         assert s._nixl_enabled is True
         assert s._data is not None
         assert s._compat_hash is not None
         assert s._producer_session is not None
-        assert s._peer_host == "127.0.0.1"
+        assert s._peer_host == ("localhost" if port else "127.0.0.1")
+        assert s._peer_port is not None and 0 < s._peer_port <= 65535
+        if port:
+            assert s._peer_port == port
+        else:
+            assert s._peer_port != dp_index  # Port 0 must not get the DP offset.
         assert s._transport is not None  # is_ec_consumer is also True
     finally:
         s.shutdown()
+
+
+@pytest.mark.skipif(NixlWrapper is None, reason="Requires NIXL package")
+def test_producers_allocate_distinct_ports_and_agent_names(monkeypatch):
+    """DP replicas retain their configured identity without peer collisions."""
+    monkeypatch.setenv("VLLM_HOST_IP", "127.0.0.1")
+    monkeypatch.delenv("VLLM_EC_SIDE_CHANNEL_HOST", raising=False)
+    monkeypatch.delenv("VLLM_EC_SIDE_CHANNEL_PORT", raising=False)
+    monkeypatch.setattr(sched_mod, "create_ec_shared_region", lambda cfg: _region())
+    engine_id = f"ec-test-{uuid.uuid4()}"
+    schedulers = []
+    expected_names = []
+    try:
+        for engine, dp_index in [(engine_id, 0), (engine_id, 1), (None, 0)]:
+            cfg = create_ec_vllm_config(ec_role="ec_producer")
+            if engine is not None:
+                cfg.ec_transfer_config.engine_id = engine
+            cfg.parallel_config.data_parallel_index = dp_index
+            cfg.ec_transfer_config.ec_connector_extra_config["ec_enable_nixl"] = True
+            expected_names.append(f"{cfg.ec_transfer_config.engine_id}_dp{dp_index}")
+            schedulers.append(ECCPUScheduler(cfg))
+        ports = [s._peer_port for s in schedulers]
+        assert all(port is not None and 0 < port <= 65535 for port in ports)
+        assert len(set(ports)) == len(schedulers)
+        for index, scheduler in enumerate(schedulers):
+            peer = schedulers[(index + 1) % len(schedulers)]._data
+            name = peer.add_remote_peer(
+                scheduler._data.get_agent_metadata(),
+                scheduler._data.get_mem_descriptor(),
+            )
+            if isinstance(name, bytes):
+                name = name.decode()
+            assert name == expected_names[index]
+    finally:
+        for scheduler in reversed(schedulers):
+            scheduler.shutdown()

vllm/distributed/ec_transfer/ec_connector/cpu/control/zmq.py

@@ -174,17 +174,27 @@ class ZmqServerTransport:
     routes a reply back to a specific peer by ZMQ identity.
     """
 
-    def __init__(self, host: str, port: int) -> None:
+    def __init__(self, host: str, port: int = 0) -> None:
         self._ctx = zmq.Context()
-        self._router = make_zmq_socket(
-            ctx=self._ctx,
-            path=make_zmq_path(scheme="tcp", host=host, port=port),
-            socket_type=zmq.ROUTER,
-            bind=True,
-        )
+        try:
+            self._router = make_zmq_socket(
+                ctx=self._ctx,
+                path=make_zmq_path(scheme="tcp", host=host, port=port),
+                socket_type=zmq.ROUTER,
+                bind=True,
+            )
+        except Exception:
+            self._ctx.destroy(linger=0)
+            raise
         self._poller = zmq.Poller()
         self._poller.register(self._router, zmq.POLLIN)
 
+    @property
+    def port(self) -> int:
+        """Return the actual bound port, including when port 0 was requested."""
+        endpoint = self._router.getsockopt_string(zmq.LAST_ENDPOINT)
+        return int(endpoint.rsplit(":", 1)[1])
+
     def poll(self, timeout_ms: int = 0) -> list[tuple[bytes, bytes]]:
         """Wait up to timeout_ms for incoming messages, then drain all pending.
 

vllm/distributed/ec_transfer/ec_connector/cpu/scheduler/__init__.py

@@ -140,6 +140,7 @@ def _setup_nixl(self, vllm_config: "VllmConfig") -> None:
             ProducerSession,
         )
         from vllm.distributed.nixl_utils import NixlWrapper, nixl_agent_config
+        from vllm.utils.network_utils import get_ip
         from vllm.version import __version__ as VLLM_VERSION
 
         if NixlWrapper is None or nixl_agent_config is None:
@@ -149,6 +150,7 @@ def _setup_nixl(self, vllm_config: "VllmConfig") -> None:
             )
         engine_id = self._ec_config.engine_id
         assert engine_id is not None
+        dp_index = vllm_config.parallel_config.data_parallel_index
         # How long a consumer waits for an XferAck. The producer answers from
         # its scheduler step, so its reply latency scales with the encoder's
         # batch size: a deployment whose steps run longer must raise this.
@@ -163,24 +165,31 @@ def _setup_nixl(self, vllm_config: "VllmConfig") -> None:
             dtype=str(self._dtype),
             block_size_bytes=self._region.block_size_bytes,
         )
-        if self._is_producer:
-            self._peer_host = envs.VLLM_EC_SIDE_CHANNEL_HOST
-            self._peer_port = envs.VLLM_EC_SIDE_CHANNEL_PORT
-
         self._data = NixlDataTransport(
-            agent_name=engine_id,
+            agent_name=f"{engine_id}_dp{dp_index}",
             base_ptr=self._region.blocks.data_ptr(),
             num_blocks=self._region.num_blocks,
             block_size_bytes=self._region.block_size_bytes,
             total_size_bytes=self._region.num_blocks * self._region.block_size_bytes,
         )
         if self._is_producer:
-            assert self._peer_host is not None
-            assert self._peer_port is not None
+            self._peer_host = envs.VLLM_EC_SIDE_CHANNEL_HOST or get_ip()
+            if self._peer_host in ("0.0.0.0", "::", "*"):
+                raise ValueError(
+                    "Set VLLM_EC_SIDE_CHANNEL_HOST or VLLM_HOST_IP to a "
+                    "reachable address, not a wildcard."
+                )
+            base_port = envs.VLLM_EC_SIDE_CHANNEL_PORT
+            port = base_port + dp_index if base_port else 0
+            if base_port < 0 or not 0 <= port <= 65535:
+                raise ValueError(
+                    "VLLM_EC_SIDE_CHANNEL_PORT + data_parallel_index "
+                    "must be between 1 and 65535, or use port 0 for auto allocation."
+                )
+            transport = ZmqServerTransport(host=self._peer_host, port=port)
+            self._peer_port = transport.port
             self._producer_session = ProducerSession(
-                transport=ZmqServerTransport(
-                    host=self._peer_host, port=self._peer_port
-                ),
+                transport=transport,
                 data=self._data,
                 cache=self._cache,
                 compat_hash=self._compat_hash,

vllm/envs.py

@@ -243,8 +243,8 @@
     VLLM_P2P_HANDSHAKE_TIMEOUT_S: int = 30
     VLLM_P2P_IDLE_TIMEOUT_S: int = 300
     VLLM_P2P_TRANSFER_STALL_TIMEOUT_S: float = 20.0
-    VLLM_EC_SIDE_CHANNEL_HOST: str = "localhost"
-    VLLM_EC_SIDE_CHANNEL_PORT: int = 5601
+    VLLM_EC_SIDE_CHANNEL_HOST: str = ""
+    VLLM_EC_SIDE_CHANNEL_PORT: int = 0
     VLLM_MOONCAKE_BOOTSTRAP_PORT: int = 8998
     VLLM_MOONCAKE_STORE_TIER_LOG: bool = False
     VLLM_MOONCAKE_LOAD_RECV_THREADS: int = 1
@@ -1816,15 +1816,11 @@ def _resolve_rust_cli_path() -> str | None:
     "VLLM_P2P_TRANSFER_STALL_TIMEOUT_S": lambda: float(
         os.getenv("VLLM_P2P_TRANSFER_STALL_TIMEOUT_S", "20")
     ),
-    # 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(
-        "VLLM_EC_SIDE_CHANNEL_HOST", "localhost"
-    ),
-    # Port for the EC connector's ZMQ side channel; advertised to peers
-    # via `ec_transfer_params.peer_port` on the producer's response.
+    # Optional EC control address override; otherwise use get_ip().
+    "VLLM_EC_SIDE_CHANNEL_HOST": lambda: os.getenv("VLLM_EC_SIDE_CHANNEL_HOST", ""),
+    # EC control base port plus DP index. Zero requests an OS-assigned port.
     "VLLM_EC_SIDE_CHANNEL_PORT": lambda: int(
-        os.getenv("VLLM_EC_SIDE_CHANNEL_PORT", "5601")
+        os.getenv("VLLM_EC_SIDE_CHANNEL_PORT", "0")
     ),
     # Per-request timeout (seconds) when a prefiller worker registers with
     # the Mooncake bootstrap server.