[Bugfix] Seed RayExecutorV2 TCPStore port by DP rank to avoid collisions (#45960)
Signed-off-by: Seiji Eicher <seiji@anyscale.com> Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
Seiji Eicher committed
Jun 30, 2026 at 07:37 UTC
ded66764586d547851a7f53c79f5770b1b01c9df
2 files changed
+64
-1
tests/distributed/test_ray_v2_executor.py
+38
@@ -17,6 +17,7 @@ import ray
17
from vllm import LLM
18
from vllm.config import VllmConfig
19
from vllm.engine.arg_utils import EngineArgs
20
+from vllm.v1.executor import ray_executor_v2
21
from vllm.v1.executor.ray_executor_v2 import RayExecutorV2
22
23
pytestmark = pytest.mark.usefixtures("enable_ray_v2_backend")
@@ -95,6 +96,43 @@ def assert_executor(executor, tp_size, pp_size):
96
assert handle.node_id is not None
97
98
99
+def test_select_tcpstore_port_seeds_disjoint_windows(monkeypatch):
100
+ """Co-located DP engines scan distinct, adjacent port windows, so two
101
+ engines on a node cannot pick the same TCPStore port."""
102
+ requested = []
103
+
104
+ def fake_get_open_port(start_port, max_attempts):
105
+ requested.append((start_port, max_attempts))
106
+ return start_port
107
+
108
+ monkeypatch.setattr(ray_executor_v2, "_get_open_port", fake_get_open_port)
109
+
110
+ ports = [
111
+ RayExecutorV2._select_tcpstore_port(rank, master_port=29500)
112
+ for rank in range(4)
113
+ ]
114
+
115
+ assert requested == [(29600, 32), (29632, 32), (29664, 32), (29696, 32)]
116
+ assert len(set(ports)) == 4
117
+
118
+
119
+def test_select_tcpstore_port_non_dp_uses_random(monkeypatch):
120
+ """A non-DP engine has no local rank and uses a random port."""
121
+ monkeypatch.setattr(ray_executor_v2, "get_open_port", lambda: 54321)
122
+ assert RayExecutorV2._select_tcpstore_port(None, master_port=29500) == 54321
123
+
124
+
125
+def test_select_tcpstore_port_full_window_uses_random(monkeypatch):
126
+ """A fully occupied window falls back to a random port."""
127
+
128
+ def raise_full(start_port, max_attempts):
129
+ raise RuntimeError("no open port")
130
+
131
+ monkeypatch.setattr(ray_executor_v2, "_get_open_port", raise_full)
132
+ monkeypatch.setattr(ray_executor_v2, "get_open_port", lambda: 54321)
133
+ assert RayExecutorV2._select_tcpstore_port(0, master_port=29500) == 54321
134
+
135
+
136
@pytest.mark.parametrize("tp_size, pp_size", [(1, 1), (2, 1), (4, 1), (2, 2)])
137
def test_ray_v2_executor(tp_size, pp_size):
138
"""Validate RayExecutorV2 with various TP/PP configs."""
vllm/v1/executor/ray_executor_v2.py
+26
-1
@@ -17,6 +17,7 @@ from vllm.distributed.device_communicators.shm_broadcast import (
17
from vllm.logger import init_logger
18
from vllm.platforms import current_platform
19
from vllm.utils.network_utils import (
20
+ _get_open_port,
21
get_distributed_init_method,
22
get_open_port,
23
)
@@ -259,6 +260,25 @@ class RayExecutorV2(MultiprocExecutor):
260
return {"num_gpus": num_devices}
261
return {"num_gpus": 0, "resources": {device_key: num_devices}}
262
263
+ @staticmethod
264
+ def _select_tcpstore_port(local_dp_rank: int | None, master_port: int) -> int:
265
+ """Pick the torch.distributed TCPStore port for this engine.
266
+
267
+ Co-located DP engines choosing this port with a shared random search
268
+ collide intermittently. Seeding by node-local DP rank gives each a
269
+ disjoint window. Non-DP engines and full windows fall back to a
270
+ random port.
271
+ """
272
+ if local_dp_rank is None:
273
+ return get_open_port()
274
+ # Offset past the DP master port reserved range, one window per rank.
275
+ window = 32
276
+ start_port = master_port + 100 + local_dp_rank * window
277
+ try:
278
+ return _get_open_port(start_port=start_port, max_attempts=window)
279
+ except RuntimeError:
280
+ return get_open_port()
281
+
282
def _init_executor(self) -> None:
283
"""Initialize the RayExecutorV2 executor."""
284
self._finalizer = weakref.finalize(self, self.shutdown)
@@ -308,7 +328,12 @@ class RayExecutorV2(MultiprocExecutor):
328
# The TCPStore server runs on rank 0's node, so all workers
329
# must be able to reach this address.
330
dist_ip = bundle_assignments[0]["node_ip"]
311
- distributed_init_method = get_distributed_init_method(dist_ip, get_open_port())
331
+ parallel_config = self.vllm_config.parallel_config
332
+ port = self._select_tcpstore_port(
333
+ parallel_config.data_parallel_rank_local,
334
+ parallel_config.data_parallel_master_port,
335
+ )
336
+ distributed_init_method = get_distributed_init_method(dist_ip, port)
337
338
# Step 4: Create broadcast MessageQueue.
339
# Workers on the driver node use shared memory; the rest use TCP.