A high-throughput and memory-efficient inference and serving engine for LLMs
Mirrored from https://github.com/tduncan-lum/Nvidia-Optimization-Stack---VLLM.git
Synced Nov 2, 13:30 UTC
[Bugfix] Exclude Ray DP from #42585's deferred port allocation (#43864)
Signed-off-by: Vadim Gimpelson <vadim.gimpelson@gmail.com>
Vadim Gimpelson committed
May 28, 2026 at 19:55 UTC
5d126dd155c0c0c9d346d87a6f7ea070fe0c99b7
2 files changed
+238
-10
tests/v1/engine/test_core_engine_actor_manager.py
+222
-1
@@ -2,6 +2,8 @@
2
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
4
import os
5
+import socket
6
+import time
7
import uuid
8
from pathlib import Path
9
from types import SimpleNamespace
@@ -9,9 +11,17 @@ from typing import Any
11
12
import pytest
13
import ray
14
+import zmq
15
16
+from vllm.utils.network_utils import make_zmq_socket, split_zmq_path
17
from vllm.v1.engine.core import EngineCoreActorMixin
14
-from vllm.v1.engine.utils import CoreEngineActorManager, EngineZmqAddresses
18
+from vllm.v1.engine.utils import (
19
+ CoreEngineActorManager,
20
+ EngineZmqAddresses,
21
+ get_engine_zmq_addresses,
22
+ launch_core_engines,
23
+)
24
+from vllm.v1.utils import APIServerProcessManager
25
26
27
class _StubEngineCoreActor(EngineCoreActorMixin):
@@ -42,6 +52,48 @@ class _StubEngineCoreActor(EngineCoreActorMixin):
52
def get_nixl_side_channel_host(self) -> str | None:
53
return os.environ.get("VLLM_NIXL_SIDE_CHANNEL_HOST")
54
55
+ def get_addresses(self) -> tuple[list[str], list[str]]:
56
+ """Return the addresses snapshot the actor was constructed with.
57
+
58
+ Used by the Ray-DP regression test to assert that no ``tcp://host:0``
59
+ placeholders were pickled into the actor at ``.remote()`` time.
60
+ """
61
+ return list(self.addresses.inputs), list(self.addresses.outputs)
62
+
63
+
64
+# Module-level stub worker for the Ray-DP regression test. Must be importable
65
+# by ``multiprocessing.spawn`` (no closures, no nesting). Mirrors the worker
66
+# in ``tests/entrypoints/test_api_server_process_manager.py``.
67
+def _bind_and_report_worker(listen_address, sock, args, client_config):
68
+ """Bind ROUTER/PULL with a kernel-assigned port, report the actual
69
+ endpoints back via ``actual_address_pipe``, then exit."""
70
+ ctx = zmq.Context()
71
+ try:
72
+ in_sock = make_zmq_socket(
73
+ ctx, client_config["input_address"], zmq.ROUTER, bind=True
74
+ )
75
+ out_sock = make_zmq_socket(
76
+ ctx, client_config["output_address"], zmq.PULL, bind=True
77
+ )
78
+ try:
79
+ pipe = client_config["actual_address_pipe"]
80
+ try:
81
+ pipe.send(
82
+ {
83
+ "input_address": in_sock.getsockopt(zmq.LAST_ENDPOINT).decode(),
84
+ "output_address": out_sock.getsockopt(
85
+ zmq.LAST_ENDPOINT
86
+ ).decode(),
87
+ }
88
+ )
89
+ finally:
90
+ pipe.close()
91
+ finally:
92
+ in_sock.close(linger=0)
93
+ out_sock.close(linger=0)
94
+ finally:
95
+ ctx.term()
96
+
97
98
class _DummyExecutor:
99
pass
@@ -134,3 +186,172 @@ def test_driver_nixl_side_channel_host_does_not_leak_to_engine_core_actor(
186
else:
187
for pg in created_placement_groups:
188
ray.util.remove_placement_group(pg)
189
+
190
+
191
+@pytest.fixture
192
+def ray_context_dp2():
193
+ """Ray context sized for two stub actors (each PG needs ~1 CPU)."""
194
+ started_ray = False
195
+ if not ray.is_initialized():
196
+ project_root = str(Path(__file__).resolve().parents[3])
197
+ ray.init(
198
+ num_cpus=4,
199
+ runtime_env={"env_vars": {"PYTHONPATH": project_root}},
200
+ log_to_driver=False,
201
+ )
202
+ started_ray = True
203
+
204
+ yield
205
+
206
+ if started_ray:
207
+ ray.shutdown()
208
+
209
+
210
+def _make_vllm_config_ray_dp_multinode() -> SimpleNamespace:
211
+ """Minimal vllm_config that drives the Ray-DP multi-API-server path:
212
+ ``data_parallel_size != data_parallel_size_local`` forces TCP placeholders
213
+ (multi-node fan-out), and ``data_parallel_backend="ray"`` routes
214
+ ``launch_core_engines`` through the Ray branch.
215
+ """
216
+ return SimpleNamespace(
217
+ parallel_config=SimpleNamespace(
218
+ data_parallel_size=2,
219
+ data_parallel_size_local=1,
220
+ data_parallel_rank=0,
221
+ data_parallel_rank_local=None,
222
+ data_parallel_master_ip="127.0.0.1",
223
+ data_parallel_backend="ray",
224
+ data_parallel_rpc_port=29550,
225
+ local_engines_only=False,
226
+ enable_elastic_ep=False,
227
+ world_size=1,
228
+ ),
229
+ model_config=SimpleNamespace(multimodal_config=None, is_moe=False),
230
+ cache_config=SimpleNamespace(),
231
+ needs_dp_coordinator=False,
232
+ kv_transfer_config=None,
233
+ # ``_apply_dp_identity_suffix`` reads and rewrites this.
234
+ instance_id="vllm-ray-dp-regression-test",
235
+ )
236
+
237
+
238
+@pytest.mark.timeout(120)
239
+@pytest.mark.usefixtures("ray_context_dp2")
240
+def test_ray_dp_addresses_resolved_before_actor_creation(
241
+ monkeypatch: pytest.MonkeyPatch,
242
+) -> None:
243
+ """Regression guard for the Ray-DP + multi-API-server hang from PR #42585.
244
+
245
+ ``launch_core_engines`` Ray branch pickles ``addresses`` into each engine
246
+ actor at ``.remote()`` time, and ``EngineCoreActorMixin._perform_handshakes``
247
+ is a no-op, so the actor uses that pickled snapshot for the rest of its
248
+ life. If ``run_multi_api_server`` allocates ``addresses`` as
249
+ ``tcp://host:0`` placeholders (its default), the actors hold placeholders
250
+ forever and DEALER-connect to port 0 — ZMQ ``connect`` is async and does
251
+ not raise, so the failure mode is a deterministic hang.
252
+
253
+ The Ray-DP carve-out in ``run_multi_api_server`` forces
254
+ ``defer_api_server_ports=False`` when ``data_parallel_backend == "ray"``
255
+ so addresses are pre-allocated in the driver and Ray pickles real ports
256
+ into each actor. This test mirrors that call-site logic and asserts the
257
+ actors hold real (non-placeholder) endpoints. If the carve-out is
258
+ removed without an alternative fix, the test fails.
259
+ """
260
+ created_placement_groups: list[Any] = []
261
+
262
+ def create_dp_placement_groups(vllm_config: Any):
263
+ pg1 = _make_cpu_placement_group()
264
+ pg2 = _make_cpu_placement_group()
265
+ created_placement_groups.extend([pg1, pg2])
266
+ return [pg1, pg2], [0, 0]
267
+
268
+ monkeypatch.setattr("vllm.v1.engine.core.EngineCoreActor", _StubEngineCoreActor)
269
+ monkeypatch.setattr(
270
+ CoreEngineActorManager,
271
+ "create_dp_placement_groups",
272
+ staticmethod(create_dp_placement_groups),
273
+ )
274
+
275
+ vllm_config = _make_vllm_config_ray_dp_multinode()
276
+
277
+ # Mirror run_multi_api_server's address-allocation logic. The Ray DP
278
+ # carve-out forces pre-allocation so the addresses pickled into engine
279
+ # actors at .remote() time are real, not ``tcp://host:0``.
280
+ is_ray_dp = vllm_config.parallel_config.data_parallel_backend == "ray"
281
+ addresses = get_engine_zmq_addresses(
282
+ vllm_config,
283
+ num_api_servers=2,
284
+ defer_api_server_ports=not is_ray_dp,
285
+ )
286
+
287
+ sock = socket.socket()
288
+ engine_manager: CoreEngineActorManager | None = None
289
+ actor_snapshots: list[tuple[list[str], list[str]]] = []
290
+ api_server_manager: APIServerProcessManager | None = None
291
+ try:
292
+ # Ray actors are spawned here, pickling ``addresses`` into each one.
293
+ with launch_core_engines(
294
+ vllm_config,
295
+ executor_class=_DummyExecutor,
296
+ log_stats=False,
297
+ addresses=addresses,
298
+ num_api_servers=2,
299
+ ) as (
300
+ engine_manager,
301
+ _coordinator,
302
+ _addresses_out,
303
+ _tensor_queue,
304
+ ):
305
+ assert isinstance(engine_manager, CoreEngineActorManager)
306
+
307
+ # API-server children bind to the pre-allocated ports.
308
+ api_server_manager = APIServerProcessManager(
309
+ listen_address="tcp://127.0.0.1:0",
310
+ sock=sock,
311
+ args="test_args",
312
+ num_servers=2,
313
+ input_addresses=addresses.inputs,
314
+ output_addresses=addresses.outputs,
315
+ target_server_fn=_bind_and_report_worker,
316
+ )
317
+
318
+ # run_multi_api_server skips ``gather_actual_addresses`` for
319
+ # Ray DP (addresses are already real). Mirror that.
320
+ if not is_ray_dp:
321
+ actual_inputs, actual_outputs = (
322
+ api_server_manager.gather_actual_addresses(timeout=15.0)
323
+ )
324
+ addresses.inputs = actual_inputs
325
+ addresses.outputs = actual_outputs
326
+
327
+ # Snapshot what each Ray actor actually holds.
328
+ actors = (
329
+ engine_manager.local_engine_actors + engine_manager.remote_engine_actors
330
+ )
331
+ actor_snapshots = ray.get(
332
+ [actor.get_addresses.remote() for actor in actors]
333
+ )
334
+ finally:
335
+ if api_server_manager is not None:
336
+ api_server_manager.shutdown()
337
+ time.sleep(0.2)
338
+ sock.close()
339
+ if engine_manager is not None:
340
+ engine_manager.shutdown()
341
+ else:
342
+ for pg in created_placement_groups:
343
+ ray.util.remove_placement_group(pg)
344
+
345
+ # Every Ray actor must hold real, non-placeholder addresses.
346
+ assert actor_snapshots, "expected at least one Ray actor to be created"
347
+ for actor_inputs, actor_outputs in actor_snapshots:
348
+ for url in actor_inputs + actor_outputs:
349
+ scheme, _host, port = split_zmq_path(url)
350
+ assert scheme == "tcp", url
351
+ assert port and int(port) > 0, (
352
+ f"Ray actor was pickled with placeholder address {url!r}; "
353
+ "``run_multi_api_server`` must pre-allocate ports for the "
354
+ "Ray DP backend so the actors hold real endpoints by the "
355
+ "time they DEALER-connect. See PR #42585 / Ray-DP "
356
+ "multi-API-server regression."
357
+ )
vllm/entrypoints/cli/serve.py
+16
-9
@@ -308,13 +308,16 @@ def run_multi_api_server(args: argparse.Namespace):
308
309
from vllm.v1.engine.utils import get_engine_zmq_addresses
310
311
- # Per-API-server ports are picked by the kernel at each child's bind()
312
- # to avoid parent-probe vs child-bind TOCTOU; Rust front-end opts out
313
- # because it has no port-report-back channel.
311
+ # Defer port allocation to the child's bind() to avoid TOCTOU, except
312
+ # for Rust front-end and Ray DP, which can't see the post-bind rebind
313
+ # (CLI-arg subprocess / pickled-into-actor snapshot respectively) and
314
+ # so pre-allocate driver-side -- reintroducing the original race only
315
+ # there.
316
+ is_ray_dp = parallel_config.data_parallel_backend == "ray"
317
addresses = get_engine_zmq_addresses(
318
vllm_config,
319
num_api_servers,
317
- defer_api_server_ports=not rust_frontend_path,
320
+ defer_api_server_ports=not (rust_frontend_path or is_ray_dp),
321
)
322
323
with launch_core_engines(
@@ -348,11 +351,15 @@ def run_multi_api_server(args: argparse.Namespace):
351
tensor_queue=tensor_queue,
352
)
353
351
- # Forward each child's bound endpoints to the engine handshake
352
- # (runs on ``with`` exit).
353
- actual_inputs, actual_outputs = api_server_manager.gather_actual_addresses()
354
- addresses.inputs = actual_inputs
355
- addresses.outputs = actual_outputs
354
+ if not is_ray_dp:
355
+ # Forward each child's bound endpoints to the engine handshake
356
+ # (runs on ``with`` exit). Skipped for Ray DP, where addresses
357
+ # are pre-allocated above and Ray actors already hold them.
358
+ actual_inputs, actual_outputs = (
359
+ api_server_manager.gather_actual_addresses()
360
+ )
361
+ addresses.inputs = actual_inputs
362
+ addresses.outputs = actual_outputs
363
364
# Wait for API servers.
365
try: