[Core] Drop the Ray driver-node GPU pin left over from the legacy executor (#61713)
Signed-off-by: Woosuk Kwon <woosuk@inferact.ai> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
3f1e57225a
commit
4a89316efa
@@ -640,7 +640,7 @@ steps:
|
||||
- tests/distributed/
|
||||
- examples/features/data_parallel/data_parallel_offline.py
|
||||
commands:
|
||||
- ./.buildkite/scripts/run-multi-node-test.sh /vllm-workspace/tests 2 2 $IMAGE_TAG "VLLM_TEST_SAME_HOST=0 torchrun --nnodes 2 --nproc-per-node=2 --rdzv_backend=c10d --rdzv_endpoint=192.168.10.10 distributed/test_same_node.py | grep 'Same node test passed' && NUM_NODES=2 torchrun --nnodes 2 --nproc-per-node=2 --rdzv_backend=c10d --rdzv_endpoint=192.168.10.10 distributed/test_node_count.py | grep 'Node count test passed' && python3 ../examples/features/data_parallel/data_parallel_offline.py -dp=2 -tp=1 --dp-num-nodes=2 --dp-node-rank=0 --dp-master-addr=192.168.10.10 --dp-master-port=12345 --enforce-eager --trust-remote-code && VLLM_MULTI_NODE=1 pytest -v -s distributed/test_multi_node_assignment.py && VLLM_MULTI_NODE=1 pytest -v -s distributed/test_pipeline_parallel.py" "VLLM_TEST_SAME_HOST=0 torchrun --nnodes 2 --nproc-per-node=2 --rdzv_backend=c10d --rdzv_endpoint=192.168.10.10 distributed/test_same_node.py | grep 'Same node test passed' && NUM_NODES=2 torchrun --nnodes 2 --nproc-per-node=2 --rdzv_backend=c10d --rdzv_endpoint=192.168.10.10 distributed/test_node_count.py | grep 'Node count test passed' && python3 ../examples/features/data_parallel/data_parallel_offline.py -dp=2 -tp=1 --dp-num-nodes=2 --dp-node-rank=1 --dp-master-addr=192.168.10.10 --dp-master-port=12345 --enforce-eager --trust-remote-code"
|
||||
- ./.buildkite/scripts/run-multi-node-test.sh /vllm-workspace/tests 2 2 $IMAGE_TAG "VLLM_TEST_SAME_HOST=0 torchrun --nnodes 2 --nproc-per-node=2 --rdzv_backend=c10d --rdzv_endpoint=192.168.10.10 distributed/test_same_node.py | grep 'Same node test passed' && NUM_NODES=2 torchrun --nnodes 2 --nproc-per-node=2 --rdzv_backend=c10d --rdzv_endpoint=192.168.10.10 distributed/test_node_count.py | grep 'Node count test passed' && python3 ../examples/features/data_parallel/data_parallel_offline.py -dp=2 -tp=1 --dp-num-nodes=2 --dp-node-rank=0 --dp-master-addr=192.168.10.10 --dp-master-port=12345 --enforce-eager --trust-remote-code && VLLM_MULTI_NODE=1 pytest -v -s distributed/test_pipeline_parallel.py" "VLLM_TEST_SAME_HOST=0 torchrun --nnodes 2 --nproc-per-node=2 --rdzv_backend=c10d --rdzv_endpoint=192.168.10.10 distributed/test_same_node.py | grep 'Same node test passed' && NUM_NODES=2 torchrun --nnodes 2 --nproc-per-node=2 --rdzv_backend=c10d --rdzv_endpoint=192.168.10.10 distributed/test_node_count.py | grep 'Node count test passed' && python3 ../examples/features/data_parallel/data_parallel_offline.py -dp=2 -tp=1 --dp-num-nodes=2 --dp-node-rank=1 --dp-master-addr=192.168.10.10 --dp-master-port=12345 --enforce-eager --trust-remote-code"
|
||||
|
||||
- label: ":nvidia: (L4) Pipeline + Context Parallelism"
|
||||
key: pipeline-context-parallelism-4-gpus
|
||||
|
||||
@@ -157,8 +157,6 @@ cryptography==50.0.1
|
||||
# azure-storage-blob
|
||||
# msal
|
||||
# pyjwt
|
||||
cupy-cuda12x==13.6.0
|
||||
# via ray
|
||||
cycler==0.12.1
|
||||
# via matplotlib
|
||||
datamodel-code-generator==0.79.0
|
||||
@@ -239,8 +237,6 @@ fastavro==1.12.2
|
||||
# via cohere
|
||||
fastparquet==2024.11.0
|
||||
# via genai-perf
|
||||
fastrlock==0.8.2
|
||||
# via cupy-cuda12x
|
||||
fastsafetensors==0.3.3
|
||||
# via -r requirements/test/cuda.in
|
||||
filelock==4.0.7
|
||||
@@ -569,7 +565,6 @@ numpy==2.2.6
|
||||
# albumentations
|
||||
# bm25s
|
||||
# contourpy
|
||||
# cupy-cuda12x
|
||||
# datasets
|
||||
# decord
|
||||
# einx
|
||||
|
||||
@@ -22,7 +22,7 @@ vector_quantize_pytorch # required for minicpmo_26 test
|
||||
vocos # required for minicpmo_26 test
|
||||
peft>=0.20.0 # required for phi-4-mm test
|
||||
pqdm
|
||||
ray[cgraph,default]>=2.56.0,<2.57 # Ray Compiled Graph, required by pipeline parallelism tests; sharded-RDT weight transfer needs Ray Direct Transport (>=2.56)
|
||||
ray[default]>=2.56.0,<2.57 # sharded-RDT weight transfer needs Ray Direct Transport (>=2.56)
|
||||
sentence-transformers>=5.2.0 # required for embedding tests
|
||||
soundfile # required for audio tests
|
||||
jiwer # required for audio tests
|
||||
|
||||
@@ -170,8 +170,6 @@ cuda-pathfinder==1.3.3
|
||||
# via cuda-bindings
|
||||
cuda-toolkit==13.0.3.0
|
||||
# via torch
|
||||
cupy-cuda12x==13.6.0
|
||||
# via ray
|
||||
cycler==0.12.1
|
||||
# via matplotlib
|
||||
datamodel-code-generator==0.79.0
|
||||
@@ -255,8 +253,6 @@ fastavro==1.12.2
|
||||
# via cohere
|
||||
fastparquet==2024.11.0
|
||||
# via genai-perf
|
||||
fastrlock==0.8.2
|
||||
# via cupy-cuda12x
|
||||
fastsafetensors==0.3.3
|
||||
# via
|
||||
# -c requirements/cuda.txt
|
||||
@@ -604,7 +600,6 @@ numpy==2.2.6
|
||||
# albumentations
|
||||
# bm25s
|
||||
# contourpy
|
||||
# cupy-cuda12x
|
||||
# datasets
|
||||
# decord
|
||||
# einx
|
||||
|
||||
@@ -16,7 +16,7 @@ librosa # required for audio tests
|
||||
vocos # required for minicpmo_26 test
|
||||
peft
|
||||
pqdm
|
||||
ray[cgraph,default]>=2.56.0,<2.57 # Ray Compiled Graph, required by pipeline parallelism tests; sharded-RDT weight transfer needs Ray Direct Transport (>=2.56)
|
||||
ray[default]>=2.56.0,<2.57 # sharded-RDT weight transfer needs Ray Direct Transport (>=2.56)
|
||||
sentence-transformers>=5.2.0 # required for embedding tests
|
||||
soundfile # required for audio tests
|
||||
jiwer # required for audio tests
|
||||
|
||||
@@ -21,7 +21,7 @@ vector_quantize_pytorch # required for minicpmo_26 test
|
||||
vocos # required for minicpmo_26 test
|
||||
peft>=0.20.0 # required for phi-4-mm test
|
||||
pqdm
|
||||
ray[cgraph,default]==2.55.1 # Includes worker startup getenv/setenv race fix
|
||||
ray[default]==2.55.1 # Includes worker startup getenv/setenv race fix
|
||||
sentence-transformers>=5.2.0 # required for embedding tests
|
||||
soundfile # required for audio tests
|
||||
jiwer # required for audio tests
|
||||
|
||||
@@ -606,7 +606,6 @@ numpy==2.2.6
|
||||
# bitsandbytes
|
||||
# bm25s
|
||||
# contourpy
|
||||
# cupy-cuda12x
|
||||
# datasets
|
||||
# decord
|
||||
# einx
|
||||
@@ -1439,7 +1438,6 @@ zipp==3.23.0
|
||||
# cuda-bindings
|
||||
# cuda-pathfinder
|
||||
# cuda-toolkit
|
||||
# cupy-cuda12x
|
||||
# nvidia-cublas
|
||||
# nvidia-cuda-cupti
|
||||
# nvidia-cuda-nvrtc
|
||||
|
||||
@@ -216,15 +216,6 @@ def test_models_distributed(
|
||||
pytest.skip(f"Skip test for {TARGET_TEST_SUITE}")
|
||||
|
||||
with monkeypatch.context() as monkeypatch_context:
|
||||
if (
|
||||
model == "meta-llama/Llama-3.2-1B-Instruct"
|
||||
and distributed_executor_backend == "ray"
|
||||
and attention_backend == ""
|
||||
and TARGET_TEST_SUITE == "L4"
|
||||
and enable_prompt_embeds
|
||||
): # noqa
|
||||
pytest.skip("enable_prompt_embeds does not work with ray compiled dag.")
|
||||
|
||||
for k, v in extra_env.items():
|
||||
monkeypatch_context.setenv(k, v)
|
||||
|
||||
|
||||
@@ -1,79 +0,0 @@
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
|
||||
"""Make sure ray assigns GPU workers to the correct node.
|
||||
|
||||
Run:
|
||||
```sh
|
||||
cd $VLLM_PATH/tests
|
||||
|
||||
pytest distributed/test_multi_node_assignment.py
|
||||
```
|
||||
"""
|
||||
|
||||
import os
|
||||
import time
|
||||
|
||||
import pytest
|
||||
import ray
|
||||
from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
|
||||
|
||||
from vllm import initialize_ray_cluster
|
||||
from vllm.config import ParallelConfig
|
||||
from vllm.utils.network_utils import get_ip
|
||||
from vllm.v1.executor.ray_utils import available_resources_per_node
|
||||
|
||||
VLLM_MULTI_NODE = os.getenv("VLLM_MULTI_NODE", "0") == "1"
|
||||
|
||||
|
||||
@pytest.mark.skipif(
|
||||
not VLLM_MULTI_NODE, reason="Need at least 2 nodes to run the test."
|
||||
)
|
||||
def test_multi_node_assignment() -> None:
|
||||
# NOTE: important to keep this class definition here
|
||||
# to let ray use cloudpickle to serialize it.
|
||||
class Actor:
|
||||
def get_ip(self):
|
||||
return get_ip()
|
||||
|
||||
for _ in range(10):
|
||||
config = ParallelConfig(1, 2)
|
||||
initialize_ray_cluster(config)
|
||||
|
||||
current_ip = get_ip()
|
||||
workers = []
|
||||
for bundle_id, bundle in enumerate(config.placement_group.bundle_specs):
|
||||
if not bundle.get("GPU", 0):
|
||||
continue
|
||||
scheduling_strategy = PlacementGroupSchedulingStrategy(
|
||||
placement_group=config.placement_group,
|
||||
placement_group_capture_child_tasks=True,
|
||||
placement_group_bundle_index=bundle_id,
|
||||
)
|
||||
|
||||
worker = ray.remote(
|
||||
num_cpus=0,
|
||||
num_gpus=1,
|
||||
scheduling_strategy=scheduling_strategy,
|
||||
)(Actor).remote()
|
||||
worker_ip = ray.get(worker.get_ip.remote())
|
||||
assert worker_ip == current_ip
|
||||
workers.append(worker)
|
||||
|
||||
for worker in workers:
|
||||
ray.kill(worker)
|
||||
|
||||
ray.util.remove_placement_group(config.placement_group)
|
||||
|
||||
gpus_needed = len(workers)
|
||||
deadline = time.monotonic() + 60
|
||||
node_id = ray.get_runtime_context().get_node_id()
|
||||
while True:
|
||||
free_gpus = available_resources_per_node().get(node_id, {}).get("GPU", 0)
|
||||
if free_gpus >= gpus_needed:
|
||||
break
|
||||
|
||||
remaining = deadline - time.monotonic()
|
||||
if remaining <= 0:
|
||||
raise TimeoutError("placement group resources were not released")
|
||||
|
||||
time.sleep(min(1, remaining))
|
||||
@@ -292,7 +292,7 @@ class RayExecutorV2(MultiprocExecutor):
|
||||
# Step 1: Initialize Ray cluster and retrieve placement group
|
||||
if ray is None:
|
||||
raise ImportError("Using Ray backend requires installation of ray.")
|
||||
initialize_ray_cluster(self.parallel_config, require_gpu_on_driver=False)
|
||||
initialize_ray_cluster(self.parallel_config)
|
||||
placement_group = self.parallel_config.placement_group
|
||||
|
||||
tp_size, pp_size, pcp_size = self._get_parallel_sizes()
|
||||
|
||||
@@ -8,7 +8,6 @@ from collections import defaultdict
|
||||
from vllm.config import ParallelConfig
|
||||
from vllm.logger import init_logger
|
||||
from vllm.platforms import current_platform
|
||||
from vllm.utils.network_utils import get_ip
|
||||
|
||||
logger = init_logger(__name__)
|
||||
PG_WAIT_TIMEOUT = 1800
|
||||
@@ -30,14 +29,6 @@ try:
|
||||
from ray.util import placement_group_table
|
||||
from ray.util.placement_group import PlacementGroup
|
||||
|
||||
try:
|
||||
from ray._private.state import available_resources_per_node
|
||||
except ImportError:
|
||||
# Ray 2.9.x doesn't expose `available_resources_per_node`
|
||||
from ray._private.state import state as _state
|
||||
|
||||
available_resources_per_node = _state._available_resources_per_node
|
||||
|
||||
ray_import_err = None
|
||||
|
||||
except ImportError as e:
|
||||
@@ -65,15 +56,8 @@ def _verify_bundles(
|
||||
placement_group: "PlacementGroup",
|
||||
parallel_config: ParallelConfig,
|
||||
device_str: str,
|
||||
require_gpu_on_driver: bool = True,
|
||||
):
|
||||
"""Verify a given placement group has bundles located in the right place.
|
||||
|
||||
There are 2 rules.
|
||||
- Warn if all tensor parallel workers cannot fit in a single node.
|
||||
- Fail if driver node is not included in a placement group
|
||||
(only when require_gpu_on_driver is True).
|
||||
"""
|
||||
"""Warn if all tensor parallel workers cannot fit in a single node."""
|
||||
assert ray.is_initialized(), (
|
||||
"Ray is not initialized although distributed-executor-backend is ray."
|
||||
)
|
||||
@@ -87,17 +71,6 @@ def _verify_bundles(
|
||||
|
||||
for bundle_idx, node_id in bundle_to_node_ids.items():
|
||||
node_id_to_bundle[node_id].append(bundles[bundle_idx])
|
||||
driver_node_id = ray.get_runtime_context().get_node_id()
|
||||
|
||||
if require_gpu_on_driver and driver_node_id not in node_id_to_bundle:
|
||||
raise RuntimeError(
|
||||
f"driver node id {driver_node_id} is not included in a placement "
|
||||
f"group {placement_group.id}. Node id -> bundles "
|
||||
f"{node_id_to_bundle}. "
|
||||
"You don't have enough GPUs available in a current node. Check "
|
||||
"`ray status` and `ray list nodes` to see if you have available "
|
||||
"GPUs in a node `{driver_node_id}` before starting an vLLM engine."
|
||||
)
|
||||
|
||||
for node_id, bundles in node_id_to_bundle.items():
|
||||
if len(bundles) < parallel_config.tensor_parallel_size:
|
||||
@@ -293,7 +266,6 @@ def _wait_until_pg_ready(current_placement_group: "PlacementGroup"):
|
||||
def initialize_ray_cluster(
|
||||
parallel_config: ParallelConfig,
|
||||
ray_address: str | None = None,
|
||||
require_gpu_on_driver: bool = True,
|
||||
):
|
||||
"""Initialize the distributed cluster with Ray.
|
||||
|
||||
@@ -305,10 +277,6 @@ def initialize_ray_cluster(
|
||||
parallel_config: The configurations for parallel execution.
|
||||
ray_address: The address of the Ray cluster. If None, uses
|
||||
the default Ray cluster address.
|
||||
require_gpu_on_driver: If True (default), require at least one GPU
|
||||
on the current (driver) node and pin the first PG bundle to it.
|
||||
Set to False for executors like RayExecutorV2 where all GPU work
|
||||
is delegated to remote Ray actors.
|
||||
|
||||
"""
|
||||
assert_ray_available()
|
||||
@@ -404,25 +372,6 @@ def initialize_ray_cluster(
|
||||
{device_str: 1.0} for _ in range(parallel_config.world_size)
|
||||
]
|
||||
|
||||
# vLLM engine is also a worker to execute model with an accelerator,
|
||||
# so it requires to have the device in a current node. Check if
|
||||
# the current node has at least one device.
|
||||
current_ip = get_ip()
|
||||
current_node_id = ray.get_runtime_context().get_node_id()
|
||||
current_node_resource = available_resources_per_node()[current_node_id]
|
||||
if require_gpu_on_driver:
|
||||
if current_node_resource.get(device_str, 0) < 1:
|
||||
raise ValueError(
|
||||
f"Current node has no {device_str} available. "
|
||||
f"{current_node_resource=}. vLLM engine cannot start "
|
||||
f"without {device_str}. Make sure you have at least 1 "
|
||||
f"{device_str} available in a node "
|
||||
f"{current_node_id=} {current_ip=}."
|
||||
)
|
||||
# This way, at least bundle is required to be created in a
|
||||
# current node.
|
||||
placement_group_specs[0][f"node:{current_ip}"] = 0.001
|
||||
|
||||
# By default, Ray packs resources as much as possible.
|
||||
current_placement_group = ray.util.placement_group(
|
||||
placement_group_specs, strategy="PACK"
|
||||
@@ -430,8 +379,6 @@ def initialize_ray_cluster(
|
||||
_wait_until_pg_ready(current_placement_group)
|
||||
|
||||
assert current_placement_group is not None
|
||||
_verify_bundles(
|
||||
current_placement_group, parallel_config, device_str, require_gpu_on_driver
|
||||
)
|
||||
_verify_bundles(current_placement_group, parallel_config, device_str)
|
||||
# Set the placement group in the parallel config
|
||||
parallel_config.placement_group = current_placement_group
|
||||
|
||||
Reference in New Issue
Block a user