[CI/Build] Use file rendezvous for local distributed tests (#51652)
Signed-off-by: yu-xin-c <2182712990@qq.com>
Stellar鱼 committed
Aug 12, 2026 at 10:48 UTC
3fa89225bc5f36ad3ada1edf0ff8c99425d7ff24
3 files changed
+31
-12
tests/compile/passes/distributed/test_async_tp.py
+10
-7
@@ -29,7 +29,7 @@ from vllm.distributed.parallel_state import (
29
initialize_model_parallel,
30
)
31
from vllm.platforms import current_platform
32
-from vllm.utils.network_utils import get_open_port
32
+from vllm.utils.network_utils import get_file_store_init_method
33
from vllm.utils.system_utils import update_environment_variables
34
from vllm.utils.torch_utils import set_random_seed
35
@@ -266,7 +266,7 @@ def test_async_tp_pass_replace(
266
dynamic: bool,
267
):
268
num_processes = 2
269
- master_port = str(get_open_port())
269
+ distributed_init_method = get_file_store_init_method()
270
271
def run_torch_spawn(fn, nprocs):
272
# need to use torch.mp.spawn otherwise will have problems with
@@ -281,7 +281,7 @@ def test_async_tp_pass_replace(
281
hidden_size,
282
dtype,
283
dynamic,
284
- master_port,
284
+ distributed_init_method,
285
),
286
nprocs=nprocs,
287
)
@@ -314,7 +314,7 @@ def async_tp_pass_on_test_model(
314
hidden_size: int,
315
dtype: torch.dtype,
316
dynamic: bool,
317
- master_port: str = "0",
317
+ distributed_init_method: str,
318
):
319
set_random_seed(0)
320
@@ -328,13 +328,16 @@ def async_tp_pass_on_test_model(
328
"RANK": str(local_rank),
329
"LOCAL_RANK": str(local_rank),
330
"WORLD_SIZE": str(world_size),
331
- "MASTER_ADDR": "localhost",
332
- "MASTER_PORT": master_port,
331
}
332
)
333
334
# initialize distributed
337
- init_distributed_environment()
335
+ init_distributed_environment(
336
+ world_size=world_size,
337
+ rank=local_rank,
338
+ distributed_init_method=distributed_init_method,
339
+ local_rank=local_rank,
340
+ )
341
342
# configure vllm config for SequenceParallelismPass
343
vllm_config = VllmConfig()
tests/distributed/test_dcp_a2a.py
+13
-5
@@ -17,7 +17,7 @@ import torch.distributed as dist
17
18
import vllm.envs as envs
19
from vllm.config.parallel import ParallelConfig
20
-from vllm.utils.network_utils import get_open_port
20
+from vllm.utils.network_utils import get_file_store_init_method
21
from vllm.utils.system_utils import update_environment_variables
22
23
mp.set_start_method("spawn", force=True)
@@ -76,7 +76,7 @@ def _assert_packed_a2a_close(
76
77
78
def _distributed_run(fn, world_size: int, extra_env: dict[str, str]) -> None:
79
- port = str(get_open_port())
79
+ distributed_init_method = get_file_store_init_method()
80
processes: list[mp.Process] = []
81
for rank in range(world_size):
82
env = {
@@ -84,8 +84,7 @@ def _distributed_run(fn, world_size: int, extra_env: dict[str, str]) -> None:
84
"LOCAL_RANK": str(rank),
85
"WORLD_SIZE": str(world_size),
86
"LOCAL_WORLD_SIZE": str(world_size),
87
- "MASTER_ADDR": "localhost",
88
- "MASTER_PORT": port,
87
+ "DISTRIBUTED_INIT_METHOD": distributed_init_method,
88
**extra_env,
89
}
90
process = mp.Process(target=fn, args=(env,))
@@ -483,14 +482,23 @@ class TestPackedA2AKernels:
482
def _distributed_packed_a2a_worker(env: dict[str, str]) -> None:
483
update_environment_variables(env)
484
local_rank = int(env["LOCAL_RANK"])
485
+ world_size = int(env["WORLD_SIZE"])
486
torch.accelerator.set_device_index(local_rank)
487
if envs.VLLM_DISTRIBUTED_USE_SPLIT_GROUP:
488
dist.init_process_group(
489
backend="cpu:gloo,cuda:nccl",
490
device_id=torch.device(f"cuda:{local_rank}"),
491
+ init_method=env["DISTRIBUTED_INIT_METHOD"],
492
+ rank=local_rank,
493
+ world_size=world_size,
494
)
495
else:
493
- dist.init_process_group(backend="nccl")
496
+ dist.init_process_group(
497
+ backend="nccl",
498
+ init_method=env["DISTRIBUTED_INIT_METHOD"],
499
+ rank=local_rank,
500
+ world_size=world_size,
501
+ )
502
use_workspace = env.get("USE_WORKSPACE") == "1"
503
if use_workspace:
504
from vllm.v1.worker.workspace import init_workspace_manager
tests/utils_/test_network_utils.py
+8
@@ -8,6 +8,7 @@ import zmq
8
9
from vllm.utils import network_utils
10
from vllm.utils.network_utils import (
11
+ get_file_store_init_method,
12
get_open_port,
13
get_open_ports_list,
14
get_tcp_uri,
@@ -19,6 +20,13 @@ from vllm.utils.network_utils import (
20
)
21
22
23
+def test_get_file_store_init_method_is_unique():
24
+ init_methods = {get_file_store_init_method() for _ in range(2)}
25
+
26
+ assert len(init_methods) == 2
27
+ assert all(method.startswith("file://") for method in init_methods)
28
+
29
+
30
def _call_with_timeout(func, timeout: float = 10.0):
31
"""Run func in a daemon thread so a livelock regression fails the test
32
quickly instead of hanging the suite."""