From f659ffe71e20e6787a764575dc7323da268f2edd Mon Sep 17 00:00:00 2001 From: Tianyu Guo Date: Thu, 15 Oct 2026 06:48:49 +0000 Subject: [PATCH 1/3] fix: allocate EC control ports at producer startup Bind each EC producer ROUTER socket to an OS-assigned port and advertise the actual endpoint. Keep NIXL peer identities local to each instance and DP rank. Signed-off-by: Tianyu Guo --- docs/features/ec_cpu_connector.md | 3 +- .../integration/test_nixl_failure.py | 2 -- tests/v1/ec_connector/unit/test_control.py | 26 ++++++++++++++++ .../unit/test_scheduler_nixl_ctor.py | 31 ++++++++++++++++--- .../ec_connector/cpu/control/zmq.py | 10 ++++-- .../ec_connector/cpu/scheduler/__init__.py | 14 ++++----- vllm/envs.py | 6 ---- 7 files changed, 69 insertions(+), 23 deletions(-) diff --git a/docs/features/ec_cpu_connector.md b/docs/features/ec_cpu_connector.md index 18274b6dd970..96ead53d743d 100644 --- a/docs/features/ec_cpu_connector.md +++ b/docs/features/ec_cpu_connector.md @@ -71,7 +71,7 @@ vllm serve --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. Each producer binds a `ROUTER` socket on `VLLM_EC_SIDE_CHANNEL_HOST` with an OS-assigned port and advertises the bound port through `ec_transfer_params.peer_port`; no port configuration is required. 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. @@ -103,7 +103,6 @@ EC transfer is configured via `--ec-transfer-config` (CLI) or the `ec_transfer_c | 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. | Both are read only when `ec_enable_nixl=True` on a producer (`ec_role="ec_producer"` or `"ec_both"`). diff --git a/tests/v1/ec_connector/integration/test_nixl_failure.py b/tests/v1/ec_connector/integration/test_nixl_failure.py index ec5cc8518e11..5a9ab22993ed 100644 --- a/tests/v1/ec_connector/integration/test_nixl_failure.py +++ b/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") @@ -80,7 +79,6 @@ def config(role): 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()), }, ) as encoder, RemoteOpenAIServer( diff --git a/tests/v1/ec_connector/unit/test_control.py b/tests/v1/ec_connector/unit/test_control.py index 93701738eb52..e1acceaadf38 100644 --- a/tests/v1/ec_connector/unit/test_control.py +++ b/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,28 @@ 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() diff --git a/tests/v1/ec_connector/unit/test_scheduler_nixl_ctor.py b/tests/v1/ec_connector/unit/test_scheduler_nixl_ctor.py index 8cad9ec65c4d..97a3dc5d4406 100644 --- a/tests/v1/ec_connector/unit/test_scheduler_nixl_ctor.py +++ b/tests/v1/ec_connector/unit/test_scheduler_nixl_ctor.py @@ -126,13 +126,12 @@ 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. +@pytest.mark.parametrize("dp_index", [0, 1]) +def test_gate_on_wires_data_transport_and_producer_session(monkeypatch, dp_index): monkeypatch.setenv("VLLM_EC_SIDE_CHANNEL_HOST", "127.0.0.1") - 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: @@ -141,6 +140,30 @@ def test_gate_on_wires_data_transport_and_producer_session(monkeypatch): 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_port is not None and 0 < s._peer_port <= 65535 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_across_dp_and_instances(monkeypatch): + """Both DP replicas and independent instances can start on one host.""" + monkeypatch.setenv("VLLM_EC_SIDE_CHANNEL_HOST", "127.0.0.1") + monkeypatch.setattr(sched_mod, "create_ec_shared_region", lambda cfg: _region()) + instance_id = f"ec-test-{uuid.uuid4()}" + schedulers = [] + try: + for instance, dp_index in [(instance_id, 0), (instance_id, 1), (None, 0)]: + cfg = create_ec_vllm_config(ec_role="ec_producer") + if instance is not None: + cfg.instance_id = instance + cfg.parallel_config.data_parallel_index = dp_index + cfg.ec_transfer_config.ec_connector_extra_config["ec_enable_nixl"] = True + 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) + finally: + for scheduler in reversed(schedulers): + scheduler.shutdown() diff --git a/vllm/distributed/ec_transfer/ec_connector/cpu/control/zmq.py b/vllm/distributed/ec_transfer/ec_connector/cpu/control/zmq.py index d87bdc792057..3a2870409101 100644 --- a/vllm/distributed/ec_transfer/ec_connector/cpu/control/zmq.py +++ b/vllm/distributed/ec_transfer/ec_connector/cpu/control/zmq.py @@ -174,17 +174,23 @@ 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) -> None: self._ctx = zmq.Context() self._router = make_zmq_socket( ctx=self._ctx, - path=make_zmq_path(scheme="tcp", host=host, port=port), + path=make_zmq_path(scheme="tcp", host=host, port=0), socket_type=zmq.ROUTER, bind=True, ) self._poller = zmq.Poller() self._poller.register(self._router, zmq.POLLIN) + @property + def port(self) -> int: + """Return the port allocated by the OS to the bound socket.""" + 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. diff --git a/vllm/distributed/ec_transfer/ec_connector/cpu/scheduler/__init__.py b/vllm/distributed/ec_transfer/ec_connector/cpu/scheduler/__init__.py index 567f2d59c9f3..950632a42917 100644 --- a/vllm/distributed/ec_transfer/ec_connector/cpu/scheduler/__init__.py +++ b/vllm/distributed/ec_transfer/ec_connector/cpu/scheduler/__init__.py @@ -147,8 +147,10 @@ def _setup_nixl(self, vllm_config: "VllmConfig") -> None: "ec_enable_nixl requires NIXL; install the `nixl` package or " "remove ec_enable_nixl from ec_connector_extra_config." ) - engine_id = self._ec_config.engine_id - assert engine_id is not None + engine_id = ( + f"{vllm_config.instance_id}_dp" + f"{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. @@ -165,7 +167,6 @@ def _setup_nixl(self, vllm_config: "VllmConfig") -> None: ) 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, @@ -176,11 +177,10 @@ def _setup_nixl(self, vllm_config: "VllmConfig") -> None: ) if self._is_producer: assert self._peer_host is not None - assert self._peer_port is not None + transport = ZmqServerTransport(host=self._peer_host) + 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, diff --git a/vllm/envs.py b/vllm/envs.py index 13922923ed57..1590f13a93da 100755 --- a/vllm/envs.py +++ b/vllm/envs.py @@ -236,7 +236,6 @@ 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 VLLM_MOONCAKE_STORE_TIER_LOG: bool = False VLLM_MOONCAKE_LOAD_RECV_THREADS: int = 1 @@ -1751,11 +1750,6 @@ def _resolve_rust_cli_path() -> str | None: "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. - "VLLM_EC_SIDE_CHANNEL_PORT": lambda: int( - os.getenv("VLLM_EC_SIDE_CHANNEL_PORT", "5601") - ), # Per-request timeout (seconds) when a prefiller worker registers with # the Mooncake bootstrap server. "VLLM_MOONCAKE_CONNECTOR_TIMEOUT": lambda: float( From a14af0cf7dd7d14a31647f47f5f402d63e184565 Mon Sep 17 00:00:00 2001 From: Tianyu Guo Date: Thu, 15 Oct 2026 07:19:05 +0000 Subject: [PATCH 2/3] fix: reuse vLLM host discovery for EC control transport Remove the EC-specific host environment variable and use get_ip() with the existing VLLM_HOST_IP override. Signed-off-by: Tianyu Guo --- docs/features/ec_cpu_connector.md | 10 ++-------- tests/v1/ec_connector/integration/test_nixl_failure.py | 5 +---- tests/v1/ec_connector/unit/test_scheduler_nixl_ctor.py | 4 ++-- .../ec_transfer/ec_connector/cpu/scheduler/__init__.py | 7 ++----- vllm/envs.py | 6 ------ 5 files changed, 7 insertions(+), 25 deletions(-) diff --git a/docs/features/ec_cpu_connector.md b/docs/features/ec_cpu_connector.md index 96ead53d743d..9f50b6940e14 100644 --- a/docs/features/ec_cpu_connector.md +++ b/docs/features/ec_cpu_connector.md @@ -71,7 +71,7 @@ vllm serve --ec-transfer-config '{ ### Protocol -- **Control plane**: ZMQ. Each producer binds a `ROUTER` socket on `VLLM_EC_SIDE_CHANNEL_HOST` with an OS-assigned port and advertises the bound port through `ec_transfer_params.peer_port`; no port configuration is required. 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. 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`; no EC-specific host or port configuration is required. 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,13 +98,7 @@ 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 - -| 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. | - -Both are read only when `ec_enable_nixl=True` on a producer (`ec_role="ec_producer"` or `"ec_both"`). +The control address uses vLLM's standard IP discovery. Set `VLLM_HOST_IP` to select a specific network interface when automatic discovery is unsuitable. ## Limitations diff --git a/tests/v1/ec_connector/integration/test_nixl_failure.py b/tests/v1/ec_connector/integration/test_nixl_failure.py index 5a9ab22993ed..7b5a37fc98bd 100644 --- a/tests/v1/ec_connector/integration/test_nixl_failure.py +++ b/tests/v1/ec_connector/integration/test_nixl_failure.py @@ -76,10 +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", - }, + env_dict={"CUDA_VISIBLE_DEVICES": devices[0]}, ) as encoder, RemoteOpenAIServer( MODEL, diff --git a/tests/v1/ec_connector/unit/test_scheduler_nixl_ctor.py b/tests/v1/ec_connector/unit/test_scheduler_nixl_ctor.py index 97a3dc5d4406..9c518e8a330c 100644 --- a/tests/v1/ec_connector/unit/test_scheduler_nixl_ctor.py +++ b/tests/v1/ec_connector/unit/test_scheduler_nixl_ctor.py @@ -128,7 +128,7 @@ def fail_control_transport(): @pytest.mark.skipif(NixlWrapper is None, reason="Requires NIXL package") @pytest.mark.parametrize("dp_index", [0, 1]) def test_gate_on_wires_data_transport_and_producer_session(monkeypatch, dp_index): - monkeypatch.setenv("VLLM_EC_SIDE_CHANNEL_HOST", "127.0.0.1") + monkeypatch.setenv("VLLM_HOST_IP", "127.0.0.1") 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 @@ -149,7 +149,7 @@ def test_gate_on_wires_data_transport_and_producer_session(monkeypatch, dp_index @pytest.mark.skipif(NixlWrapper is None, reason="Requires NIXL package") def test_producers_allocate_distinct_ports_across_dp_and_instances(monkeypatch): """Both DP replicas and independent instances can start on one host.""" - monkeypatch.setenv("VLLM_EC_SIDE_CHANNEL_HOST", "127.0.0.1") + monkeypatch.setenv("VLLM_HOST_IP", "127.0.0.1") monkeypatch.setattr(sched_mod, "create_ec_shared_region", lambda cfg: _region()) instance_id = f"ec-test-{uuid.uuid4()}" schedulers = [] diff --git a/vllm/distributed/ec_transfer/ec_connector/cpu/scheduler/__init__.py b/vllm/distributed/ec_transfer/ec_connector/cpu/scheduler/__init__.py index 950632a42917..8729c31c8707 100644 --- a/vllm/distributed/ec_transfer/ec_connector/cpu/scheduler/__init__.py +++ b/vllm/distributed/ec_transfer/ec_connector/cpu/scheduler/__init__.py @@ -124,7 +124,6 @@ def __init__(self, vllm_config: "VllmConfig") -> None: def _setup_nixl(self, vllm_config: "VllmConfig") -> None: # Lazy imports keep nixl/zmq off the gate-off path. - from vllm import envs from vllm.distributed.ec_transfer.ec_connector.cpu.control.zmq import ( ZmqClientTransport, ZmqServerTransport, @@ -140,6 +139,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: @@ -165,9 +165,6 @@ 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._data = NixlDataTransport( agent_name=engine_id, base_ptr=self._region.blocks.data_ptr(), @@ -176,7 +173,7 @@ def _setup_nixl(self, vllm_config: "VllmConfig") -> None: total_size_bytes=self._region.num_blocks * self._region.block_size_bytes, ) if self._is_producer: - assert self._peer_host is not None + self._peer_host = get_ip() transport = ZmqServerTransport(host=self._peer_host) self._peer_port = transport.port self._producer_session = ProducerSession( diff --git a/vllm/envs.py b/vllm/envs.py index 1590f13a93da..4d8f2453c897 100755 --- a/vllm/envs.py +++ b/vllm/envs.py @@ -235,7 +235,6 @@ 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_MOONCAKE_BOOTSTRAP_PORT: int = 8998 VLLM_MOONCAKE_STORE_TIER_LOG: bool = False VLLM_MOONCAKE_LOAD_RECV_THREADS: int = 1 @@ -1745,11 +1744,6 @@ def _resolve_rust_cli_path() -> str | None: # 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( - "VLLM_EC_SIDE_CHANNEL_HOST", "localhost" - ), # Per-request timeout (seconds) when a prefiller worker registers with # the Mooncake bootstrap server. "VLLM_MOONCAKE_CONNECTOR_TIMEOUT": lambda: float( From 3796b0849b6d3a66103579f044abf1f1a5ae1d8e Mon Sep 17 00:00:00 2001 From: Tianyu Guo Date: Sat, 24 Oct 2026 01:04:42 +0000 Subject: [PATCH 3/3] fix: preserve EC control endpoint overrides with DP offsets Assisted-by: OpenAI Codex Signed-off-by: Tianyu Guo --- docs/features/ec_cpu_connector.md | 9 +++- tests/v1/ec_connector/unit/test_control.py | 10 ++++ .../unit/test_scheduler_nixl_ctor.py | 46 +++++++++++++++---- .../ec_connector/cpu/control/zmq.py | 20 ++++---- .../ec_connector/cpu/scheduler/__init__.py | 26 ++++++++--- vllm/envs.py | 8 ++++ 6 files changed, 95 insertions(+), 24 deletions(-) diff --git a/docs/features/ec_cpu_connector.md b/docs/features/ec_cpu_connector.md index 9f50b6940e14..b895d8dcbb0c 100644 --- a/docs/features/ec_cpu_connector.md +++ b/docs/features/ec_cpu_connector.md @@ -71,7 +71,7 @@ vllm serve --ec-transfer-config '{ ### Protocol -- **Control plane**: ZMQ. 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`; no EC-specific host or port configuration is required. 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. @@ -100,6 +100,13 @@ EC transfer is configured via `--ec-transfer-config` (CLI) or the `ec_transfer_c The control address uses vLLM's standard IP discovery. Set `VLLM_HOST_IP` to select a specific network interface when automatic discovery is unsuitable. +| Environment variable | Default | Description | +| --- | --- | --- | +| `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. | + +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 - No mechanism to notify an orchestrator or peer instance when an encoding is evicted from the CPU tier before it's consumed. A consumer only discovers a miss when its `XferReq` is NACKed (`NACK_MISSING`) and falls back to local recompute. diff --git a/tests/v1/ec_connector/unit/test_control.py b/tests/v1/ec_connector/unit/test_control.py index e1acceaadf38..64bb077c9067 100644 --- a/tests/v1/ec_connector/unit/test_control.py +++ b/tests/v1/ec_connector/unit/test_control.py @@ -309,3 +309,13 @@ def test_zmq_servers_allocate_distinct_reachable_ports(): 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() diff --git a/tests/v1/ec_connector/unit/test_scheduler_nixl_ctor.py b/tests/v1/ec_connector/unit/test_scheduler_nixl_ctor.py index 9c518e8a330c..2f3ea893174d 100644 --- a/tests/v1/ec_connector/unit/test_scheduler_nixl_ctor.py +++ b/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 @@ -127,8 +128,20 @@ def fail_control_transport(): @pytest.mark.skipif(NixlWrapper is None, reason="Requires NIXL package") @pytest.mark.parametrize("dp_index", [0, 1]) -def test_gate_on_wires_data_transport_and_producer_session(monkeypatch, dp_index): +@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 @@ -139,31 +152,48 @@ def test_gate_on_wires_data_transport_and_producer_session(monkeypatch, dp_index 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_across_dp_and_instances(monkeypatch): - """Both DP replicas and independent instances can start on one host.""" +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()) - instance_id = f"ec-test-{uuid.uuid4()}" + engine_id = f"ec-test-{uuid.uuid4()}" schedulers = [] + expected_names = [] try: - for instance, dp_index in [(instance_id, 0), (instance_id, 1), (None, 0)]: + for engine, dp_index in [(engine_id, 0), (engine_id, 1), (None, 0)]: cfg = create_ec_vllm_config(ec_role="ec_producer") - if instance is not None: - cfg.instance_id = instance + 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() diff --git a/vllm/distributed/ec_transfer/ec_connector/cpu/control/zmq.py b/vllm/distributed/ec_transfer/ec_connector/cpu/control/zmq.py index 3a2870409101..7d17d069a374 100644 --- a/vllm/distributed/ec_transfer/ec_connector/cpu/control/zmq.py +++ b/vllm/distributed/ec_transfer/ec_connector/cpu/control/zmq.py @@ -174,20 +174,24 @@ class ZmqServerTransport: routes a reply back to a specific peer by ZMQ identity. """ - def __init__(self, host: str) -> 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=0), - 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 port allocated by the OS to the bound socket.""" + """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]) diff --git a/vllm/distributed/ec_transfer/ec_connector/cpu/scheduler/__init__.py b/vllm/distributed/ec_transfer/ec_connector/cpu/scheduler/__init__.py index 8729c31c8707..b487fa7dee40 100644 --- a/vllm/distributed/ec_transfer/ec_connector/cpu/scheduler/__init__.py +++ b/vllm/distributed/ec_transfer/ec_connector/cpu/scheduler/__init__.py @@ -124,6 +124,7 @@ def __init__(self, vllm_config: "VllmConfig") -> None: def _setup_nixl(self, vllm_config: "VllmConfig") -> None: # Lazy imports keep nixl/zmq off the gate-off path. + from vllm import envs from vllm.distributed.ec_transfer.ec_connector.cpu.control.zmq import ( ZmqClientTransport, ZmqServerTransport, @@ -147,10 +148,9 @@ def _setup_nixl(self, vllm_config: "VllmConfig") -> None: "ec_enable_nixl requires NIXL; install the `nixl` package or " "remove ec_enable_nixl from ec_connector_extra_config." ) - engine_id = ( - f"{vllm_config.instance_id}_dp" - f"{vllm_config.parallel_config.data_parallel_index}" - ) + 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. @@ -166,15 +166,27 @@ def _setup_nixl(self, vllm_config: "VllmConfig") -> None: block_size_bytes=self._region.block_size_bytes, ) 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: - self._peer_host = get_ip() - transport = ZmqServerTransport(host=self._peer_host) + 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=transport, diff --git a/vllm/envs.py b/vllm/envs.py index 4d8f2453c897..69522660e038 100755 --- a/vllm/envs.py +++ b/vllm/envs.py @@ -262,6 +262,8 @@ VLLM_ROCM_QUICK_REDUCE_MIN_SIZE_BYTES_MB: int | None = None VLLM_ROCM_QUICK_REDUCE_QUANTIZATION_MIN_SIZE_KB: int | None = None VLLM_MOONCAKE_ABORT_REQUEST_TIMEOUT: int = 480 + VLLM_EC_SIDE_CHANNEL_HOST: str = "" + VLLM_EC_SIDE_CHANNEL_PORT: int = 0 VLLM_MOONCAKE_CONNECTOR_TIMEOUT: float = 30.0 VLLM_ENABLE_CUDAGRAPH_GC: bool = False VLLM_LOOPBACK_IP: str = "" @@ -1744,6 +1746,12 @@ def _resolve_rust_cli_path() -> str | None: # 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")), + # 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", "0") + ), # Per-request timeout (seconds) when a prefiller worker registers with # the Mooncake bootstrap server. "VLLM_MOONCAKE_CONNECTOR_TIMEOUT": lambda: float(