[Bug]: Multi vllm engine and tp > 1 will cause device error

[zhuo97]

Your current environment

Atlas 800T A2

🐛 Describe the bug

Below is my test script test_vllm_ray.py

import ray
from vllm import LLM, SamplingParams

ray.init()

@ray.remote
class VLLMEngine:
    def __init__(self, model_name, tensor_parallel_size):
        self.llm = LLM(
            model=model_name,
            tensor_parallel_size=tensor_parallel_size,
        )

    def generate(self, prompts):
        sampling_params = SamplingParams(temperature=0.7, max_tokens=100)
        outputs = self.llm.generate(prompts, sampling_params)
        return outputs

model_name= "xxx"
tensor_parallel_size = 2
num_engines = 2

engines = [VLLMEngine.options(resources={"NPU": tensor_parallel_size}).remote(model_name, tensor_parallel_size) for _ in range(num_engines)]

prompts = [
    "Explain the concept of quantum computing in simple terms.",
    "What are the benefits of using Ray for distributed computing?",
]

for i in range(len(prompts)):
    results = ray.get([engines[i % num_engines].generate.remote([prompts[i]])])

for i, output in enumerate(results):
    print(f"Prompt: {prompts[i]}")
    for generated_text in output:
        print(f"Generated: {generated_text.outputs[0].text}")
    print("-" * 50)

If RAY_EXPERIMENTAL_NOSET_ASCEND_RT_VISIBLE_DEVICES=0, which default value is zero, then num_engines=2, tensor_parallel_size=2 will cause error.

python test_vllm_ray.py

[zhuo97]

If we use bundles in vllm-ascend, it will cause device error. Below is my test script test_vllm_ray_with_bundles.py.

import ray
import os
from vllm import LLM, SamplingParams
from ray.util.placement_group import placement_group
from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy

ray.init()

@ray.remote
class VLLMEngine:
    def __init__(self, model_name, tensor_parallel_size, distributed_executor_backend, bundle_indices, num_gpus):
        
        if bundle_indices is not None:
            os.environ["VLLM_RAY_PER_WORKER_GPUS"] = str(num_gpus) 
            os.environ["VLLM_RAY_BUNDLE_INDICES"] = ",".join(map(str, bundle_indices))
            print(f"creating LLM with bundle_indices={bundle_indices}")

        self.llm = LLM(
            model=model_name,
            tensor_parallel_size=tensor_parallel_size,
            distributed_executor_backend=distributed_executor_backend,
        )

    def generate(self, prompts):
        sampling_params = SamplingParams(temperature=0.7, max_tokens=100)
        outputs = self.llm.generate(prompts, sampling_params)
        return outputs

model_name= "xxx"
tensor_parallel_size = 2
num_engines = 1
distributed_executor_backend = "ray"

bundles = [{"NPU": 1, "CPU": 1} for _ in range(num_engines * tensor_parallel_size)]
shared_pg = placement_group(bundles, strategy="PACK")
ray.get(shared_pg.ready())

engines = []
for i in range(num_engines):
    bundle_indices = list(range(i * tensor_parallel_size, (i+1) * tensor_parallel_size))
    
    scheduling_strategy = PlacementGroupSchedulingStrategy(
        placement_group=shared_pg,
        placement_group_capture_child_tasks=True,
        placement_group_bundle_index=i * tensor_parallel_size,
    )
    
    engines.append(
        VLLMEngine.options(
            num_cpus=0,
            num_gpus=0,
            resources={"NPU": 0},
            scheduling_strategy=scheduling_strategy,
        ).remote(
            model_name=model_name,
            tensor_parallel_size=tensor_parallel_size,
            distributed_executor_backend=distributed_executor_backend,
            bundle_indices=bundle_indices,
            num_gpus=1,
        )
    )    

prompts = [
    "Explain the concept of quantum computing in simple terms.",
    "What are the benefits of using Ray for distributed computing?",
]

for i in range(len(prompts)):
    results = ray.get([engines[i % num_engines].generate.remote([prompts[i]])])

for i, output in enumerate(results):
    print(f"Prompt: {prompts[i]}")
    for generated_text in output:
        print(f"Generated: {generated_text.outputs[0].text}")
    print("-" * 50)

[311dada]

Following the advice here, I added import vllm during the initialization of LLMRayActor. After setting export RAY_EXPERIMENTAL_NOSET_ASCEND_RT_VISIBLE_DEVICES=1, the setup works when num_engines=1 and tp=2. However, when I try num_engines=2 and tp=2, it throws the device error. What’s even stranger is that only the workers launched by the second engine fail, and it spawns 3 workers (instead of the expected 2). The first engine runs without any issues. I‘m new to the implementation details of vLLM. Do you have any suggestions for resolving this issue? Thanks!

import os
import queue
from collections import defaultdict
from typing import Any, List

import ray
from ray.util.placement_group import placement_group, placement_group_table
from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy

ray.init()

ACCELERATOR_TYPE = "NPU"

def ray_noset_visible_devices(env_vars=os.environ):
    # Refer to
    # https://github.com/ray-project/ray/blob/161849364a784442cc659fb9780f1a6adee85fce/python/ray/_private/accelerators/nvidia_gpu.py#L95-L96
    # https://github.com/ray-project/ray/blob/161849364a784442cc659fb9780f1a6adee85fce/python/ray/_private/accelerators/amd_gpu.py#L102-L103
    # https://github.com/ray-project/ray/blob/161849364a784442cc659fb9780f1a6adee85fce/python/ray/_private/accelerators/npu.py#L94-L95
    # https://github.com/ray-project/ray/blob/161849364a784442cc659fb9780f1a6adee85fce/python/ray/_private/accelerators/hpu.py#L116-L117
    # https://github.com/ray-project/ray/blob/161849364a784442cc659fb9780f1a6adee85fce/python/ray/_private/accelerators/neuron.py#L108-L109
    # https://github.com/ray-project/ray/blob/161849364a784442cc659fb9780f1a6adee85fce/python/ray/_private/accelerators/tpu.py#L171-L172
    # https://github.com/ray-project/ray/blob/161849364a784442cc659fb9780f1a6adee85fce/python/ray/_private/accelerators/intel_gpu.py#L97-L98
    NOSET_VISIBLE_DEVICES_ENV_VARS_LIST = [
        "RAY_EXPERIMENTAL_NOSET_CUDA_VISIBLE_DEVICES",
        "RAY_EXPERIMENTAL_NOSET_ROCR_VISIBLE_DEVICES",
        "RAY_EXPERIMENTAL_NOSET_ASCEND_RT_VISIBLE_DEVICES",
        "RAY_EXPERIMENTAL_NOSET_HABANA_VISIBLE_MODULES",
        "RAY_EXPERIMENTAL_NOSET_NEURON_RT_VISIBLE_CORES",
        "RAY_EXPERIMENTAL_NOSET_TPU_VISIBLE_CHIPS",
        "RAY_EXPERIMENTAL_NOSET_ONEAPI_DEVICE_SELECTOR",
    ]
    return any(env_vars.get(env_var) for env_var in NOSET_VISIBLE_DEVICES_ENV_VARS_LIST)



@ray.remote
def get_all_env_variables():
    import os

    return os.environ


@ray.remote
class LLMRayActor:

    def __init__(self, *args, bundle_indices: list = None, **kwargs):

        noset_visible_devices = kwargs.pop("noset_visible_devices")
        if kwargs.get("distributed_executor_backend") == "ray":
            # a hack to make the script work.
            # stop ray from manipulating CUDA_VISIBLE_DEVICES
            # at the top-level when the distributed_executor_backend is ray.
            os.environ.pop("CUDA_VISIBLE_DEVICES", None)
            os.environ.pop("ASCEND_RT_VISIBLE_DEVICES", None)
        elif noset_visible_devices:
            # We need to set CUDA_VISIBLE_DEVICES to the ray assigned GPU
            # when the distributed_executor_backend is not ray and
            # RAY_EXPERIMENTAL_NOSET_*_VISIBLE_DEVICES is set.
            environ_name = "CUDA_VISIBLE_DEVICES" if ACCELERATOR_TYPE == "GPU" else "ASCEND_RT_VISIBLE_DEVICES"
            os.environ[environ_name] = str(ray.get_runtime_context().get_accelerator_ids()["NPU"][0])
            print("setting ", environ_name, "to ", str(ray.get_runtime_context().get_accelerator_ids()["NPU"][0]))

        # every worker will use 0.2 GPU, so that we can schedule
        # 2 instances on the same GPUs.
        if bundle_indices is not None:
            os.environ["VLLM_RAY_PER_WORKER_GPUS"] = "0.2"
            os.environ["VLLM_RAY_BUNDLE_INDICES"] = ",".join(map(str, bundle_indices))
            print(f"creating LLM with bundle_indices={bundle_indices}")
        
        from vllm import LLM

        self.llm = LLM(*args, **kwargs)
    
    def test(self):
        print("finished!")

    def init_process_group(self, master_address, master_port, rank_offset, world_size, group_name, backend, use_ray):
        return self.llm.collective_rpc(
            "init_process_group",
            args=(master_address, master_port, rank_offset, world_size, group_name, backend, use_ray),
        )

    def update_weight(self, name, dtype, shape, empty_cache=False):
        return self.llm.collective_rpc("update_weight", args=(name, dtype, shape, empty_cache))

    def update_weight_cuda_ipc(self, name, dtype, shape, ipc_handles, empty_cache=False):
        return self.llm.collective_rpc("update_weight_cuda_ipc", args=(name, dtype, shape, ipc_handles, empty_cache))

    def reset_prefix_cache(self):
        self.llm.llm_engine.reset_prefix_cache()

    def sleep(self, level=1):
        self.llm.sleep(level=level)

    def wake_up(self):
        self.llm.wake_up()

    def add_requests(self, actor_rank, *, sampling_params, prompt_token_ids):
        """
        Save the requests from actors and generate responses when all actors have sent their requests
        """
        self.requests[actor_rank] = prompt_token_ids
        self.actor_counter += 1
        if self.actor_counter == self.num_actors:
            assert len(self.requests) == self.num_actors
            num_requests = []
            requests = []
            for actor_rank, request in self.requests.items():
                num_requests.append((actor_rank, len(request)))
                requests.extend(request)

            if len(requests) > 0:
                # For now we assume that all requests have the same sampling params
                responses = self.llm.generate(sampling_params=sampling_params, prompt_token_ids=requests)
            else:
                responses = []

            offset = 0
            self.responses = {}
            for actor_rank, num in num_requests:
                self.responses[actor_rank] = responses[offset : offset + num]
                offset += num

            self.actor_counter = 0
            self.requests = {}

    def get_responses(self, actor_rank):
        """
        Return the responses for the actor with the given rank
        """
        return self.responses.pop(actor_rank)


def create_vllm_engines(
    num_engines: int,
    tensor_parallel_size: int,
    pretrain: str,
):
    vllm_engines = []
    num_gpus = int(tensor_parallel_size == 1)
    distributed_executor_backend = "uni" if tensor_parallel_size == 1 else "ray"
    for i in range(num_engines):
        bundle_indices = None
        scheduling_strategy = None

        # Distributed RLHF
        if tensor_parallel_size > 1:
            bundles = [{"CPU": 1}] + [{ACCELERATOR_TYPE: 1, "CPU": 1}] * tensor_parallel_size
            pg = placement_group(bundles)
            ray.get(pg.ready())

            scheduling_strategy = PlacementGroupSchedulingStrategy(
                placement_group=pg, placement_group_capture_child_tasks=True, placement_group_bundle_index=0
            )

        vllm_engines.append(
            LLMRayActor.options(
                num_cpus=0,
                num_gpus=num_gpus if ACCELERATOR_TYPE == "GPU" else 0,
                resources=None if ACCELERATOR_TYPE == "GPU" else {ACCELERATOR_TYPE: num_gpus},
                scheduling_strategy=scheduling_strategy,
            ).remote(
                model=pretrain,
                tensor_parallel_size=tensor_parallel_size,
                distributed_executor_backend=distributed_executor_backend,
                bundle_indices=None,
                noset_visible_devices=ray_noset_visible_devices(),
            )
        )
    
    return vllm_engines


num_engines = 2
tensor_parallel_size = 2
model = ...

vllm_engines = create_vllm_engines(num_engines, tensor_parallel_size, model)

ray.get(vllm_engines[0].test.remote())

import time
time.sleep(30)

[HollowMan6]

Check OpenRLHF/OpenRLHF#741 (comment) for the solution

[wangxiyuan]

feel free to create a new issue if you have any question. Thanks.

[janelu9]

Following the advice here, I added import vllm during the initialization of LLMRayActor. After setting export RAY_EXPERIMENTAL_NOSET_ASCEND_RT_VISIBLE_DEVICES=1, the setup works when num_engines=1 and tp=2. However, when I try num_engines=2 and tp=2, it throws the device error. What’s even stranger is that only the workers launched by the second engine fail, and it spawns 3 workers (instead of the expected 2). The first engine runs without any issues. I‘m new to the implementation details of vLLM. Do you have any suggestions for resolving this issue? Thanks!

import os
import queue
from collections import defaultdict
from typing import Any, List

import ray
from ray.util.placement_group import placement_group, placement_group_table
from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy

ray.init()

ACCELERATOR_TYPE = "NPU"

def ray_noset_visible_devices(env_vars=os.environ):
    # Refer to
    # https://github.com/ray-project/ray/blob/161849364a784442cc659fb9780f1a6adee85fce/python/ray/_private/accelerators/nvidia_gpu.py#L95-L96
    # https://github.com/ray-project/ray/blob/161849364a784442cc659fb9780f1a6adee85fce/python/ray/_private/accelerators/amd_gpu.py#L102-L103
    # https://github.com/ray-project/ray/blob/161849364a784442cc659fb9780f1a6adee85fce/python/ray/_private/accelerators/npu.py#L94-L95
    # https://github.com/ray-project/ray/blob/161849364a784442cc659fb9780f1a6adee85fce/python/ray/_private/accelerators/hpu.py#L116-L117
    # https://github.com/ray-project/ray/blob/161849364a784442cc659fb9780f1a6adee85fce/python/ray/_private/accelerators/neuron.py#L108-L109
    # https://github.com/ray-project/ray/blob/161849364a784442cc659fb9780f1a6adee85fce/python/ray/_private/accelerators/tpu.py#L171-L172
    # https://github.com/ray-project/ray/blob/161849364a784442cc659fb9780f1a6adee85fce/python/ray/_private/accelerators/intel_gpu.py#L97-L98
    NOSET_VISIBLE_DEVICES_ENV_VARS_LIST = [
        "RAY_EXPERIMENTAL_NOSET_CUDA_VISIBLE_DEVICES",
        "RAY_EXPERIMENTAL_NOSET_ROCR_VISIBLE_DEVICES",
        "RAY_EXPERIMENTAL_NOSET_ASCEND_RT_VISIBLE_DEVICES",
        "RAY_EXPERIMENTAL_NOSET_HABANA_VISIBLE_MODULES",
        "RAY_EXPERIMENTAL_NOSET_NEURON_RT_VISIBLE_CORES",
        "RAY_EXPERIMENTAL_NOSET_TPU_VISIBLE_CHIPS",
        "RAY_EXPERIMENTAL_NOSET_ONEAPI_DEVICE_SELECTOR",
    ]
    return any(env_vars.get(env_var) for env_var in NOSET_VISIBLE_DEVICES_ENV_VARS_LIST)



@ray.remote
def get_all_env_variables():
    import os

    return os.environ


@ray.remote
class LLMRayActor:

    def __init__(self, *args, bundle_indices: list = None, **kwargs):

        noset_visible_devices = kwargs.pop("noset_visible_devices")
        if kwargs.get("distributed_executor_backend") == "ray":
            # a hack to make the script work.
            # stop ray from manipulating CUDA_VISIBLE_DEVICES
            # at the top-level when the distributed_executor_backend is ray.
            os.environ.pop("CUDA_VISIBLE_DEVICES", None)
            os.environ.pop("ASCEND_RT_VISIBLE_DEVICES", None)
        elif noset_visible_devices:
            # We need to set CUDA_VISIBLE_DEVICES to the ray assigned GPU
            # when the distributed_executor_backend is not ray and
            # RAY_EXPERIMENTAL_NOSET_*_VISIBLE_DEVICES is set.
            environ_name = "CUDA_VISIBLE_DEVICES" if ACCELERATOR_TYPE == "GPU" else "ASCEND_RT_VISIBLE_DEVICES"
            os.environ[environ_name] = str(ray.get_runtime_context().get_accelerator_ids()["NPU"][0])
            print("setting ", environ_name, "to ", str(ray.get_runtime_context().get_accelerator_ids()["NPU"][0]))

        # every worker will use 0.2 GPU, so that we can schedule
        # 2 instances on the same GPUs.
        if bundle_indices is not None:
            os.environ["VLLM_RAY_PER_WORKER_GPUS"] = "0.2"
            os.environ["VLLM_RAY_BUNDLE_INDICES"] = ",".join(map(str, bundle_indices))
            print(f"creating LLM with bundle_indices={bundle_indices}")
        
        from vllm import LLM

        self.llm = LLM(*args, **kwargs)
    
    def test(self):
        print("finished!")

    def init_process_group(self, master_address, master_port, rank_offset, world_size, group_name, backend, use_ray):
        return self.llm.collective_rpc(
            "init_process_group",
            args=(master_address, master_port, rank_offset, world_size, group_name, backend, use_ray),
        )

    def update_weight(self, name, dtype, shape, empty_cache=False):
        return self.llm.collective_rpc("update_weight", args=(name, dtype, shape, empty_cache))

    def update_weight_cuda_ipc(self, name, dtype, shape, ipc_handles, empty_cache=False):
        return self.llm.collective_rpc("update_weight_cuda_ipc", args=(name, dtype, shape, ipc_handles, empty_cache))

    def reset_prefix_cache(self):
        self.llm.llm_engine.reset_prefix_cache()

    def sleep(self, level=1):
        self.llm.sleep(level=level)

    def wake_up(self):
        self.llm.wake_up()

    def add_requests(self, actor_rank, *, sampling_params, prompt_token_ids):
        """
        Save the requests from actors and generate responses when all actors have sent their requests
        """
        self.requests[actor_rank] = prompt_token_ids
        self.actor_counter += 1
        if self.actor_counter == self.num_actors:
            assert len(self.requests) == self.num_actors
            num_requests = []
            requests = []
            for actor_rank, request in self.requests.items():
                num_requests.append((actor_rank, len(request)))
                requests.extend(request)

            if len(requests) > 0:
                # For now we assume that all requests have the same sampling params
                responses = self.llm.generate(sampling_params=sampling_params, prompt_token_ids=requests)
            else:
                responses = []

            offset = 0
            self.responses = {}
            for actor_rank, num in num_requests:
                self.responses[actor_rank] = responses[offset : offset + num]
                offset += num

            self.actor_counter = 0
            self.requests = {}

    def get_responses(self, actor_rank):
        """
        Return the responses for the actor with the given rank
        """
        return self.responses.pop(actor_rank)


def create_vllm_engines(
    num_engines: int,
    tensor_parallel_size: int,
    pretrain: str,
):
    vllm_engines = []
    num_gpus = int(tensor_parallel_size == 1)
    distributed_executor_backend = "uni" if tensor_parallel_size == 1 else "ray"
    for i in range(num_engines):
        bundle_indices = None
        scheduling_strategy = None

        # Distributed RLHF
        if tensor_parallel_size > 1:
            bundles = [{"CPU": 1}] + [{ACCELERATOR_TYPE: 1, "CPU": 1}] * tensor_parallel_size
            pg = placement_group(bundles)
            ray.get(pg.ready())

            scheduling_strategy = PlacementGroupSchedulingStrategy(
                placement_group=pg, placement_group_capture_child_tasks=True, placement_group_bundle_index=0
            )

        vllm_engines.append(
            LLMRayActor.options(
                num_cpus=0,
                num_gpus=num_gpus if ACCELERATOR_TYPE == "GPU" else 0,
                resources=None if ACCELERATOR_TYPE == "GPU" else {ACCELERATOR_TYPE: num_gpus},
                scheduling_strategy=scheduling_strategy,
            ).remote(
                model=pretrain,
                tensor_parallel_size=tensor_parallel_size,
                distributed_executor_backend=distributed_executor_backend,
                bundle_indices=None,
                noset_visible_devices=ray_noset_visible_devices(),
            )
        )
    
    return vllm_engines


num_engines = 2
tensor_parallel_size = 2
model = ...

vllm_engines = create_vllm_engines(num_engines, tensor_parallel_size, model)

ray.get(vllm_engines[0].test.remote())

import time
time.sleep(30)

num_engines =1 didn't work too, Invalid device ID, why?

[zhuo97]

num_engines =1 didn't work too, Invalid device ID, why?

• Use "mp" as vllm-ascend backend could avoid device error. Reference: https://github.com/zhuo97/OpenRLHF/blob/main/openrlhf/trainer/ray/vllm_engine.py
• If you want to use "ray" as vllm-ascend backend, please follow the progress of this PR. This PR needs to wait for torch_npu to be updated before it can be merged.

[janelu9]

• Use "mp" as vllm-ascend backend could avoid device error. Reference: https://github.com/zhuo97/OpenRLHF/blob/main/openrlhf/trainer/ray/vllm_engine.py
• If you want to use "ray" as vllm-ascend backend, please follow the progress of this PR. This PR needs to wait for torch_npu to be updated before it can be merged.

Succeed!

[HemiFate]

请问这个问题目前在vllm-ascend 已经修复了吗？我又遇到了同样的问题。使用ray集群来管理时，多卡的node，只有第一个服务是可以启动成功的，其他的会报RuntimeError: init:build/CMakeFiles/torch_npu.dir/compiler_depend.ts:285 NPU function error: aclrtGetDeviceCount(&device_count), error code is 207004 与上述问题同样的错误。
ray 2.45.0
vllm 0.9.0
vllm-ascend 0.9.0rc2
torch 2.5.1
torch-npu 2.5.1
torchaudio 2.7.0
torchvision 0.20.1