[Bugfix][DP] add_dp_placement_groups does not require ray[default] (#57648)
Signed-off-by: beenpow <57864041+beenpow@users.noreply.github.com>
Chanbin Lim committed
Sep 29, 2026 at 11:15 UTC
91de237333e5f72b12e4de9f535e5774f9bfb1de
2 files changed
+149
-21
tests/v1/engine/test_core_engine_actor_manager.py
+109
@@ -20,6 +20,8 @@ from vllm.v1.engine.core_client import BackgroundResources
20
from vllm.v1.engine.utils import (
21
CoreEngineActorManager,
22
EngineZmqAddresses,
23
+ _dp_nodes_master_first,
24
+ _node_ip_from_resources,
25
get_engine_zmq_addresses,
26
launch_core_engines,
27
)
@@ -363,3 +365,110 @@ def test_ray_dp_addresses_resolved_before_actor_creation(
365
"time they DEALER-connect. See PR #42585 / Ray-DP "
366
"multi-API-server regression."
367
)
368
+
369
+
370
+def test_dp_nodes_master_first_orders_and_validates():
371
+ master = {"GPU": 2.0, "node:10.0.0.1": 1.0, "node:__internal_head__": 1.0}
372
+ worker = {"GPU": 2.0, "node:10.0.0.2": 1.0}
373
+
374
+ nodes = _dp_nodes_master_first({"w": worker, "m": master}, "10.0.0.1")
375
+ assert [node_id for node_id, _ in nodes] == ["m", "w"]
376
+
377
+ with pytest.raises(AssertionError, match="DP master node"):
378
+ _dp_nodes_master_first({"w": worker}, "10.0.0.1")
379
+ with pytest.raises(AssertionError, match="one head node"):
380
+ _dp_nodes_master_first({"a": master, "b": dict(master)}, "10.0.0.1")
381
+
382
+
383
+@pytest.fixture
384
+def ray_2gpu_node(monkeypatch: pytest.MonkeyPatch):
385
+ """Real single-node Ray with 2 virtual GPUs and no dashboard.
386
+
387
+ Resource maps and placement groups are real. Without the dashboard any
388
+ ``ray.util.state.list_nodes()`` call fails, the production condition this
389
+ path must survive. Only the platform device key is patched ("" on CPU).
390
+ """
391
+ from ray._private.state import available_resources_per_node
392
+
393
+ if ray.is_initialized():
394
+ ray.shutdown()
395
+ ray.init(num_cpus=4, num_gpus=2, include_dashboard=False, log_to_driver=False)
396
+ monkeypatch.setattr("vllm.v1.engine.utils.current_platform.ray_device_key", "GPU")
397
+
398
+ (node_resources,) = available_resources_per_node().values()
399
+ master_ip = _node_ip_from_resources(node_resources)
400
+ assert master_ip is not None
401
+ created = []
402
+
403
+ def hold_gpus(num_gpus: int):
404
+ """Occupy GPUs so the node looks like it already runs engines."""
405
+ pg = ray.util.placement_group([{"GPU": 1.0}] * num_gpus)
406
+ ray.get(pg.ready(), timeout=60)
407
+ created.append(pg)
408
+
409
+ yield SimpleNamespace(master_ip=master_ip, hold_gpus=hold_gpus, created=created)
410
+
411
+ for pg in created:
412
+ ray.util.remove_placement_group(pg)
413
+ ray.shutdown()
414
+
415
+
416
+def _elastic_ep_config(dp_size: int, world_size: int, master_ip: str):
417
+ return SimpleNamespace(
418
+ parallel_config=SimpleNamespace(
419
+ data_parallel_size=dp_size,
420
+ data_parallel_master_ip=master_ip,
421
+ world_size=world_size,
422
+ )
423
+ )
424
+
425
+
426
+@pytest.mark.timeout(120)
427
+def test_add_dp_placement_groups_does_not_require_ray_default(ray_2gpu_node):
428
+ """DP 1 -> 2 on an idle node: one schedulable group pinned to the master,
429
+ found without the dashboard (``list_nodes`` would fail here)."""
430
+ ip = ray_2gpu_node.master_ip
431
+
432
+ pgs, local_ranks = CoreEngineActorManager.add_dp_placement_groups(
433
+ _elastic_ep_config(dp_size=1, world_size=1, master_ip=ip), 2
434
+ )
435
+ ray_2gpu_node.created.extend(pgs)
436
+
437
+ assert [pg.bundle_specs for pg in pgs] == [
438
+ [{"GPU": 1.0, f"node:{ip}": 0.001}, {"CPU": 1.0}]
439
+ ]
440
+ assert local_ranks == [0]
441
+ ray.get(pgs[0].ready(), timeout=60)
442
+
443
+
444
+@pytest.mark.timeout(120)
445
+def test_add_dp_placement_groups_counts_engines_already_on_node(ray_2gpu_node):
446
+ """One GPU already held -> the new engine gets local rank 1."""
447
+ ray_2gpu_node.hold_gpus(1)
448
+
449
+ pgs, local_ranks = CoreEngineActorManager.add_dp_placement_groups(
450
+ _elastic_ep_config(dp_size=1, world_size=1, master_ip=ray_2gpu_node.master_ip),
451
+ 2,
452
+ )
453
+ ray_2gpu_node.created.extend(pgs)
454
+
455
+ assert len(pgs) == 1
456
+ assert local_ranks == [1]
457
+ ray.get(pgs[0].ready(), timeout=60)
458
+
459
+
460
+@pytest.mark.timeout(120)
461
+def test_add_dp_placement_groups_respects_world_size(ray_2gpu_node):
462
+ """With TP=2 a node with one idle GPU cannot host a new engine."""
463
+ ray_2gpu_node.hold_gpus(1)
464
+
465
+ assert CoreEngineActorManager.add_dp_placement_groups(
466
+ _elastic_ep_config(dp_size=1, world_size=2, master_ip=ray_2gpu_node.master_ip),
467
+ 2,
468
+ ) == ([], [])
469
+
470
+
471
+def test_add_dp_placement_groups_noop_without_growth():
472
+ assert CoreEngineActorManager.add_dp_placement_groups(
473
+ _elastic_ep_config(dp_size=2, world_size=1, master_ip="10.0.0.1"), 2
474
+ ) == ([], [])
vllm/v1/engine/utils.py
+40
-21
@@ -124,6 +124,31 @@ def _get_bundle_node_ip(bundle: dict[str, float]) -> str:
124
raise ValueError(f"Missing node affinity in placement bundle: {bundle}")
125
126
127
+def _dp_nodes_master_first(
128
+ available_resources: dict[str, dict], dp_master_ip: str
129
+) -> list[tuple[str, dict]]:
130
+ """Live Ray nodes as ``(node_id, resources)`` with the DP master first.
131
+
132
+ The master is found through the ``node:<ip>`` resource Ray sets on every
133
+ node, so this works with the resource maps from ``ray._private.state``.
134
+ Do not replace it with ``ray.util.state.list_nodes()``: that API goes
135
+ through the dashboard HTTP server and only exists with ``ray[default]``
136
+ (see #23822).
137
+ """
138
+ master_key = f"node:{dp_master_ip}"
139
+ nodes = sorted(
140
+ available_resources.items(), key=lambda item: master_key not in item[1]
141
+ )
142
+ assert len(nodes) > 0, "No nodes with resources found in Ray cluster."
143
+ assert master_key in nodes[0][1], (
144
+ f"The DP master node (ip: {dp_master_ip}) is missing or dead"
145
+ )
146
+ assert len(nodes) == 1 or master_key not in nodes[1][1], (
147
+ "There can only be one head node"
148
+ )
149
+ return nodes
150
+
151
+
152
def _node_ip_from_resources(node_resources: dict) -> str | None:
153
"""Return the node IP encoded in a Ray per-node resource dict, or None.
154
@@ -561,14 +586,12 @@ class CoreEngineActorManager:
586
placement_groups: list[PlacementGroup] = []
587
local_dp_ranks: list[int] = []
588
564
- dp_master_ip_key = f"node:{dp_master_ip}"
565
- nodes = sorted(
566
- available_resources.values(), key=lambda x: dp_master_ip_key not in x
567
- )
568
- assert len(nodes) > 0, "No nodes with resources found in Ray cluster."
569
- assert dp_master_ip_key in nodes[0], (
570
- f"The DP master node (ip: {dp_master_ip}) is missing or dead"
571
- )
589
+ nodes = [
590
+ node_resources
591
+ for _, node_resources in _dp_nodes_master_first(
592
+ available_resources, dp_master_ip
593
+ )
594
+ ]
595
596
# optionally restrict DP placement to a caller-provided node set.
597
requested_node_ips = {
@@ -767,7 +790,6 @@ class CoreEngineActorManager:
790
available_resources_per_node,
791
total_resources_per_node,
792
)
770
- from ray.util.state import list_nodes
793
794
old_dp_size = old_vllm_config.parallel_config.data_parallel_size
795
num_pg_to_create = new_data_parallel_size - old_dp_size
@@ -778,30 +800,27 @@ class CoreEngineActorManager:
800
dp_master_ip = old_vllm_config.parallel_config.data_parallel_master_ip
801
world_size = old_vllm_config.parallel_config.world_size
802
781
- nodes = list_nodes()
782
- nodes = sorted(nodes, key=lambda node: node.node_ip != dp_master_ip)
783
- assert nodes[0].node_ip == dp_master_ip, "The first node must be the head node"
784
- assert len(nodes) == 1 or nodes[1].node_ip != dp_master_ip, (
785
- "There can only be one head node"
786
- )
787
-
803
+ # Both maps are keyed by node id and only contain live nodes.
804
available_resources = available_resources_per_node()
805
total_resources = total_resources_per_node()
806
+ nodes = _dp_nodes_master_first(available_resources, dp_master_ip)
807
808
placement_groups = []
809
local_dp_ranks = []
810
num_pg_created = 0
811
812
device_str = current_platform.ray_device_key
796
- for node in nodes:
813
+ for node_id, node_resources in nodes:
814
if num_pg_created >= num_pg_to_create:
815
break
816
800
- node_ip = node.node_ip
801
- node_id = node.node_id
802
- if device_str not in available_resources[node_id]:
817
+ node_ip = _node_ip_from_resources(node_resources)
818
+ assert node_ip is not None, (
819
+ f"No node IP key found in node resources: {node_resources}"
820
+ )
821
+ if device_str not in node_resources:
822
continue
804
- available_gpus = int(available_resources[node_id][device_str])
823
+ available_gpus = int(node_resources[device_str])
824
825
# Get total GPUs on this node from the node's resources
826
# Ray stores node resources with node ID as key