[DP][V1] Fix rank set in DP scenario

Signed-off-by: MengqingCao <cmq0113@163.com>

Author: MengqingCao

tests/multicard/test_data_parallel.py

@@ -0,0 +1,68 @@
+#
+# Copyright (c) 2025 Huawei Technologies Co., Ltd. All Rights Reserved.
+# Copyright 2023 The vLLM team.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""
+Compare the outputs of vLLM with and without aclgraph.
+Run `pytest tests/multicard/test_data_parallel.py`.
+"""
+
+import os
+
+import pytest
+
+from tests.conftest import VllmRunner
+from tests.model_utils import check_outputs_equal
+from vllm_ascend.utils import vllm_version_is
+
+MODELS = ["Qwen/Qwen2.5-0.5B-Instruct"]
+
+
+@pytest.mark.skipif(vllm_version_is("0.9.0"),
+                    reason="Data parallel only support on >= 0.9.1")
+@pytest.mark.skipif(os.getenv("VLLM_USE_V1") == "0",
+                    reason="Data parallel only support on v1")
+@pytest.mark.parametrize("model", MODELS)
+@pytest.mark.parametrize("max_tokens", [32])
+def test_data_parallel_correctness(
+    model: str,
+    max_tokens: int,
+) -> None:
+    example_prompts = [
+        "Hello, my name is", "The president of the United States is",
+        "The capital of France is", "The future of AI is"
+    ]
+
+    with VllmRunner(model_name=model,
+                    max_model_len=1024,
+                    max_num_seqs=16,
+                    data_parallel_size=2,
+                    distributed_executor_backend="mp") as vllm_model:
+        vllm_dp_outputs = vllm_model.generate_greedy(example_prompts,
+                                                     max_tokens)
+
+    with VllmRunner(
+            model_name=model,
+            max_model_len=1024,
+            max_num_seqs=16,
+    ) as vllm_model:
+        vllm_outputs = vllm_model.generate_greedy(example_prompts, max_tokens)
+
+    check_outputs_equal(
+        outputs_0_lst=vllm_outputs,
+        outputs_1_lst=vllm_dp_outputs,
+        name_0="vllm_outputs",
+        name_1="vllm_dp_outputs",
+    )

vllm_ascend/patch/platform/patch_0_9_0/patch_distributed.py

@@ -1,10 +1,10 @@
 import torch
+import vllm
 from torch.distributed import ProcessGroup
 from torch.distributed.distributed_c10d import (Backend, PrefixStore,
                                                 _get_default_timeout,
                                                 is_nccl_available)
 from torch.distributed.rendezvous import rendezvous
-from vllm.distributed import utils
 
 
 def stateless_init_torch_distributed_process_group(
@@ -113,4 +113,4 @@ def stateless_init_torch_distributed_process_group(
     return pg
 
 
-utils.stateless_init_torch_distributed_process_group = stateless_init_torch_distributed_process_group
+vllm.distributed.utils.stateless_init_torch_distributed_process_group = stateless_init_torch_distributed_process_group

vllm_ascend/patch/platform/patch_common/patch_distributed.py

@@ -21,10 +21,9 @@
 import vllm.distributed
 import vllm.envs as envs
 from torch.distributed import ProcessGroup
-from vllm.config import ParallelConfig, VllmConfig
+from vllm.config import ParallelConfig
 from vllm.distributed.utils import \
     stateless_init_torch_distributed_process_group
-from vllm.v1.engine.core import DPEngineCoreProc
 
 
 def ascend_destroy_model_parallel():
@@ -79,21 +78,6 @@ def stateless_init_dp_group(self) -> "ProcessGroup":
     return dp_group
 
 
-def _init_data_parallel(self, vllm_config: VllmConfig):
-    # Configure NPUs and stateless process group for data parallel.
-    dp_rank = vllm_config.parallel_config.data_parallel_rank
-    dp_size = vllm_config.parallel_config.data_parallel_size
-    local_dp_rank = vllm_config.parallel_config.data_parallel_rank_local
-
-    assert dp_size > 1
-    assert 0 <= local_dp_rank <= dp_rank < dp_size
-
-    self.local_dp_rank = local_dp_rank
-    self.dp_group = vllm_config.parallel_config.stateless_init_dp_group()
-    self.current_wave = 0
-
-
 vllm.distributed.parallel_state.destroy_model_parallel = ascend_destroy_model_parallel
-DPEngineCoreProc._init_data_parallel = _init_data_parallel
 ParallelConfig.get_next_dp_init_port = parallel_config_get_dp_port
 ParallelConfig.stateless_init_dp_group = stateless_init_dp_group

vllm_ascend/worker/worker_v1.py

@@ -75,12 +75,6 @@ def __init__(
                          distributed_init_method=distributed_init_method,
                          is_driver_worker=is_driver_worker)
 
-        # NOTE(Yizhou): Since we do not set ASCEND_RT_VISIBLE_DEVICES in
-        # vllm_ascend, we need to set the device id manually.
-        local_dp_rank = self.vllm_config.parallel_config.data_parallel_rank_local
-        world_size = self.vllm_config.parallel_config.world_size
-        self.local_rank_across_dp = local_dp_rank * world_size + self.local_rank
-
         # Try to import mindie_turbo to accelerate vLLM inference.
         try_register_lib(
             "mindie_turbo",
@@ -119,7 +113,7 @@ def wake_up(self, tags: Optional[list[str]] = None) -> None:
 
     def init_device(self):
         if self.device_config.device.type == "npu":
-            self.device = torch.device(f"npu:{self.local_rank_across_dp}")
+            self.device = torch.device(f"npu:{self.local_rank}")
             NPUPlatform.set_device(self.device)
             NPUPlatform.empty_cache()
             self.init_npu_memory = NPUPlatform.mem_get_info()[0]