[Bugfix] Account for local DP workers in startup thread allocation (#52385)
Signed-off-by: real-cpu <zhaochenrui757@gmail.com>
crZhao committed
Aug 17, 2026 at 12:10 UTC
3fc28939099a3bad1bcd27e8eec20d9afe921879
2 files changed
+51
-1
tests/distributed/test_multiproc_executor.py
+47
@@ -11,15 +11,62 @@ import multiprocessing
11
import os
12
import socket
13
14
+import pytest
15
+
16
from tests.utils import multi_gpu_test
17
from vllm.config import VllmConfig
18
from vllm.engine.arg_utils import EngineArgs
19
from vllm.v1.core.sched.output import SchedulerOutput
20
+from vllm.v1.executor import multiproc_executor
21
from vllm.v1.executor.multiproc_executor import MultiprocExecutor
22
23
MODEL = "facebook/opt-125m"
24
25
26
+@pytest.mark.parametrize(
27
+ ("local_world_size", "data_parallel_size_local", "expected_num_local_procs"),
28
+ [(1, 4, 4), (4, 1, 4), (2, 0, 2)],
29
+)
30
+def test_multiproc_executor_counts_all_local_dp_workers(
31
+ monkeypatch: pytest.MonkeyPatch,
32
+ local_world_size: int,
33
+ data_parallel_size_local: int,
34
+ expected_num_local_procs: int,
35
+):
36
+ """All colocated DP workers share the node's startup CPU budget."""
37
+ executor = object.__new__(MultiprocExecutor)
38
+ executor.world_size = local_world_size
39
+ executor.local_world_size = local_world_size
40
+ executor.parallel_config = type(
41
+ "ParallelConfig",
42
+ (),
43
+ {"data_parallel_size_local": data_parallel_size_local},
44
+ )()
45
+
46
+ monkeypatch.setattr(
47
+ executor,
48
+ "_get_parallel_sizes",
49
+ lambda: (local_world_size, 1, 1),
50
+ )
51
+
52
+ class StopExecutorInit(Exception):
53
+ pass
54
+
55
+ def capture_num_local_procs(num_local_procs: int):
56
+ assert num_local_procs == expected_num_local_procs
57
+ raise StopExecutorInit
58
+
59
+ monkeypatch.setattr(
60
+ multiproc_executor,
61
+ "set_multiprocessing_worker_envs",
62
+ capture_num_local_procs,
63
+ )
64
+
65
+ with pytest.raises(StopExecutorInit):
66
+ executor._init_executor()
67
+ executor._finalizer.detach()
68
+
69
+
70
def create_vllm_config(
71
tensor_parallel_size: int = 1,
72
pipeline_parallel_size: int = 1,
vllm/v1/executor/multiproc_executor.py
+4
-1
@@ -130,7 +130,10 @@ class MultiprocExecutor(Executor):
130
f"_parallel_size ({pcp_size}). "
131
)
132
133
- set_multiprocessing_worker_envs(self.local_world_size)
133
+ num_local_procs = self.local_world_size * max(
134
+ 1, self.parallel_config.data_parallel_size_local
135
+ )
136
+ set_multiprocessing_worker_envs(num_local_procs)
137
138
if aiter_requires_tcp_store():
139
distributed_init_method = get_distributed_init_method(