[BugFix] Use file:// rendezvous for single-node executors to eliminate startup port races (#50999)
Signed-off-by: Parsa Mazaheri <p_mazaheri@apple.com> Signed-off-by: aoshen02 <aoshen02@users.noreply.github.com> Signed-off-by: aoshen02 <aoshen@inferact.ai> Co-authored-by: Parsa Mazaheri <p_mazaheri@apple.com> Co-authored-by: aoshen02 <aoshen02@users.noreply.github.com> Co-authored-by: Claude <noreply@anthropic.com>
aoshen02 committed
Aug 10, 2026 at 10:00 UTC
11ba93f3646d4c5476c3b3fd56835589701f0fb1
6 files changed
+142
-13
tests/distributed/test_file_store.py
new
+116
@@ -0,0 +1,116 @@
1
+# SPDX-License-Identifier: Apache-2.0
2
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
+
4
+from types import SimpleNamespace
5
+
6
+import torch
7
+
8
+import vllm.config
9
+import vllm.distributed.parallel_state as parallel_state
10
+from vllm.distributed.parallel_state import (
11
+ destroy_distributed_environment,
12
+ init_distributed_environment,
13
+)
14
+
15
+
16
+def _file_store_worker(rank, init_method, result_queue):
17
+ try:
18
+ init_distributed_environment(
19
+ world_size=2,
20
+ rank=rank,
21
+ local_rank=rank,
22
+ backend="gloo",
23
+ distributed_init_method=init_method,
24
+ )
25
+ result = torch.tensor(rank)
26
+ torch.distributed.all_reduce(result)
27
+ result_queue.put(result.item())
28
+ finally:
29
+ destroy_distributed_environment()
30
+
31
+
32
+def test_file_store_preserves_raw_path(tmp_path, monkeypatch):
33
+ monkeypatch.setenv("VLLM_DISTRIBUTED_USE_SPLIT_GROUP", "0")
34
+ store_path = tmp_path / "store#?"
35
+
36
+ try:
37
+ init_distributed_environment(
38
+ world_size=1,
39
+ rank=0,
40
+ local_rank=0,
41
+ backend="gloo",
42
+ distributed_init_method=f"file://{store_path}",
43
+ )
44
+ assert store_path.exists()
45
+ finally:
46
+ destroy_distributed_environment()
47
+
48
+
49
+def test_file_store_rendezvouses_local_processes(tmp_path, monkeypatch):
50
+ monkeypatch.setenv("VLLM_DISTRIBUTED_USE_SPLIT_GROUP", "0")
51
+ context = torch.multiprocessing.get_context("spawn")
52
+ result_queue = context.SimpleQueue()
53
+
54
+ torch.multiprocessing.spawn(
55
+ _file_store_worker,
56
+ args=(f"file://{tmp_path / 'store#?'}", result_queue),
57
+ nprocs=2,
58
+ join=True,
59
+ )
60
+
61
+ assert sorted(result_queue.get() for _ in range(2)) == [1, 1]
62
+
63
+
64
+def test_multinode_overrides_file_store_rendezvous(monkeypatch):
65
+ monkeypatch.setenv("VLLM_DISTRIBUTED_USE_SPLIT_GROUP", "0")
66
+ parallel_config = SimpleNamespace(
67
+ distributed_executor_backend="mp",
68
+ nnodes=2,
69
+ data_parallel_size=1,
70
+ enable_elastic_ep=False,
71
+ data_parallel_rank=0,
72
+ world_size_across_dp=2,
73
+ master_addr="192.0.2.1",
74
+ master_port=29500,
75
+ nnodes_within_dp=2,
76
+ )
77
+ config = SimpleNamespace(parallel_config=parallel_config)
78
+ calls = []
79
+ original_world = parallel_state._WORLD
80
+ original_node_count = parallel_state._NODE_COUNT
81
+
82
+ monkeypatch.setattr(vllm.config, "get_current_vllm_config_or_none", lambda: config)
83
+ monkeypatch.setattr(torch.distributed, "is_initialized", lambda: False)
84
+ monkeypatch.setattr(torch.distributed, "is_backend_available", lambda _: True)
85
+ monkeypatch.setattr(torch.distributed, "get_world_size", lambda: 2)
86
+ monkeypatch.setattr(
87
+ torch.distributed, "init_process_group", lambda **kwargs: calls.append(kwargs)
88
+ )
89
+ monkeypatch.setattr(
90
+ parallel_state,
91
+ "init_world_group",
92
+ lambda *_: SimpleNamespace(world_size=2, cpu_group=None),
93
+ )
94
+
95
+ try:
96
+ init_distributed_environment(
97
+ world_size=2,
98
+ rank=0,
99
+ local_rank=0,
100
+ backend="gloo",
101
+ distributed_init_method="file:///tmp/vllm_dist_test",
102
+ )
103
+ finally:
104
+ parallel_state._WORLD = original_world
105
+ parallel_state._NODE_COUNT = original_node_count
106
+
107
+ assert calls == [
108
+ {
109
+ "backend": "gloo",
110
+ "init_method": "tcp://192.0.2.1:29500",
111
+ "store": None,
112
+ "world_size": 2,
113
+ "rank": 0,
114
+ "timeout": None,
115
+ }
116
+ ]
vllm/distributed/parallel_state.py
+11
-2
@@ -1489,6 +1489,7 @@ def _init_process_group_for_split_group(
1489
*,
1490
backend: str,
1491
distributed_init_method: str,
1492
+ store: Store | None,
1493
world_size: int,
1494
rank: int,
1495
local_rank: int,
@@ -1512,7 +1513,8 @@ def _init_process_group_for_split_group(
1513
device_id = None
1514
torch.distributed.init_process_group(
1515
backend=init_backend,
1515
- init_method=distributed_init_method,
1516
+ init_method=distributed_init_method if store is None else None,
1517
+ store=store,
1518
world_size=world_size,
1519
rank=rank,
1520
timeout=timeout,
@@ -1654,6 +1656,11 @@ def init_distributed_environment(
1656
"Fallback Gloo backend is not available."
1657
)
1658
backend = "gloo"
1659
+ store = None
1660
+ if distributed_init_method.startswith("file://"):
1661
+ store = torch.distributed.FileStore(
1662
+ distributed_init_method.removeprefix("file://"), world_size
1663
+ )
1664
if envs.VLLM_DISTRIBUTED_USE_SPLIT_GROUP:
1665
# split_group needs local_rank early to compute device_id for
1666
# the eager init. local_rank is not available in torch
@@ -1667,6 +1674,7 @@ def init_distributed_environment(
1674
_init_process_group_for_split_group(
1675
backend=backend,
1676
distributed_init_method=distributed_init_method,
1677
+ store=store,
1678
world_size=world_size,
1679
rank=rank,
1680
local_rank=local_rank,
@@ -1676,7 +1684,8 @@ def init_distributed_environment(
1684
# this backend is used for WORLD
1685
torch.distributed.init_process_group(
1686
backend=backend,
1679
- init_method=distributed_init_method,
1687
+ init_method=distributed_init_method if store is None else None,
1688
+ store=store,
1689
world_size=world_size,
1690
rank=rank,
1691
timeout=timeout,
vllm/utils/network_utils.py
+5
@@ -5,6 +5,7 @@ import ipaddress
5
import os
6
import socket
7
import sys
8
+import tempfile
9
import warnings
10
from collections.abc import (
11
Iterator,
@@ -131,6 +132,10 @@ def get_distributed_init_method(ip: str, port: int) -> str:
132
return get_tcp_uri(ip, port)
133
134
135
+def get_file_store_init_method() -> str:
136
+ return f"file://{tempfile.gettempdir()}/vllm_dist_{uuid4().hex}"
137
+
138
+
139
def get_tcp_uri(ip: str, port: int) -> str:
140
if is_valid_ipv6_address(ip):
141
return f"tcp://[{ip}]:{port}"
vllm/v1/executor/multiproc_executor.py
+2
-7
@@ -46,10 +46,8 @@ from vllm.platforms import current_platform
46
from vllm.tracing import instrument, maybe_init_worker_tracer
47
from vllm.utils import numa_utils
48
from vllm.utils.network_utils import (
49
- get_distributed_init_method,
49
+ get_file_store_init_method,
50
get_ip,
51
- get_loopback_ip,
52
- get_open_port,
51
)
52
from vllm.utils.ompmultiprocessing import OMPProcessManager
53
from vllm.utils.system_utils import (
@@ -129,10 +127,7 @@ class MultiprocExecutor(Executor):
127
128
set_multiprocessing_worker_envs(self.local_world_size)
129
132
- # use the loopback address get_loopback_ip() for communication.
133
- distributed_init_method = get_distributed_init_method(
134
- get_loopback_ip(), get_open_port()
135
- )
130
+ distributed_init_method = get_file_store_init_method()
131
self.rpc_broadcast_mq: MessageQueue | None = None
132
scheduler_output_handle: Handle | None = None
133
# Initialize worker and set up message queues for SchedulerOutputs
vllm/v1/executor/uniproc_executor.py
+2
-2
@@ -12,7 +12,7 @@ import torch.distributed as dist
12
import vllm.envs as envs
13
from vllm.logger import init_logger
14
from vllm.platforms import current_platform
15
-from vllm.utils.network_utils import get_distributed_init_method, get_ip, get_open_port
15
+from vllm.utils.network_utils import get_file_store_init_method
16
from vllm.v1.core.sched.output import GrammarOutput, SchedulerOutput
17
from vllm.v1.executor.abstract import Executor
18
from vllm.v1.executor.vllm_net_devices import set_worker_net_device
@@ -70,7 +70,7 @@ class UniProcExecutor(Executor):
70
71
def _distributed_args(self) -> tuple[str, int, int]:
72
"""Return (distributed_init_method, rank, local_rank)."""
73
- distributed_init_method = get_distributed_init_method(get_ip(), get_open_port())
73
+ distributed_init_method = get_file_store_init_method()
74
# set local rank as the device index if specified
75
device_info = self.vllm_config.device_config.device.__str__().split(":")
76
local_rank = int(device_info[1]) if len(device_info) > 1 else 0
vllm/v1/worker/cpu_worker.py
+6
-2
@@ -157,8 +157,12 @@ class CPUWorker(Worker):
157
158
torch.set_num_threads = skip_set_num_threads
159
160
- # Note: unique identifier for creating allreduce shared memory
161
- os.environ["VLLM_DIST_IDENT"] = self.distributed_init_method.split(":")[-1]
160
+ init_method = self.distributed_init_method
161
+ os.environ["VLLM_DIST_IDENT"] = (
162
+ os.path.basename(init_method.removeprefix("file://"))
163
+ if init_method.startswith("file://")
164
+ else init_method.split(":")[-1]
165
+ )
166
# Initialize the distributed environment.
167
init_worker_distributed_environment(
168
self.vllm_config,