@tduncan / Nvidia-Optimization-Stack-VLLM Mirror · read-only / commits / 8619e7158c

Fork 0

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 15, 04:45 UTC

[BugFix] Fix multi-node offline data parallel (#19937)

Signed-off-by: Nick Hill <nhill@redhat.com>

Nick Hill committed Jun 24, 2025 at 12:45 UTC 8619e7158cc6d95534d3985257999323445c36e0

5 files changed +31 -4

.buildkite/test-pipeline.yaml

@@ -615,13 +615,16 @@ steps:

615 - vllm/executor/

616 - vllm/model_executor/models/

617 - tests/distributed/

618 + - tests/examples/offline_inference/data_parallel.py

619 commands:

620 - # the following commands are for the first node, with ip 192.168.10.10 (ray environment already set up)

621 - 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'

622 + - python3 ../examples/offline_inference/data_parallel.py --dp-size=2 --tp-size=1 --node-size=2 --node-rank=0 --master-addr=192.168.10.10 --master-port=12345 --enforce-eager --trust-remote-code

623 - VLLM_MULTI_NODE=1 pytest -v -s distributed/test_multi_node_assignment.py

624 - VLLM_MULTI_NODE=1 pytest -v -s distributed/test_pipeline_parallel.py

625 - # the following commands are for the second node, with ip 192.168.10.11 (ray environment already set up)

626 - 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'

627 + - python3 ../examples/offline_inference/data_parallel.py --dp-size=2 --tp-size=1 --node-size=2 --node-rank=1 --master-addr=192.168.10.10 --master-port=12345 --enforce-eager --trust-remote-code

628

629 - label: Distributed Tests (2 GPUs) # 40min

630 mirror_hardwares: [amdexperimental]

vllm/entrypoints/llm.py

@@ -1568,6 +1568,8 @@ class LLM:

1568 pbar.update(n)

1569 else:

1570 pbar.update(1)

1571 + if pbar.n == num_requests:

1572 + pbar.refresh()

1573

1574 if use_tqdm:

1575 pbar.close()

vllm/v1/engine/core.py

+6 -2

@@ -877,12 +877,16 @@ class DPEngineCoreProc(EngineCoreProc):

877 local_unfinished_reqs)

878

879 if not self.engines_running:

880 - if self.dp_rank == 0:

880 + if self.dp_rank == 0 or not self.has_coordinator:

881 # Notify client that we are pausing the loop.

882 logger.debug("Wave %d finished, pausing engine loop.",

883 self.current_wave)

884 + # In the coordinator case, dp rank 0 sends updates to the

885 + # coordinator. Otherwise (offline spmd case), each rank

886 + # sends the update to its colocated front-end process.

887 + client_index = -1 if self.has_coordinator else 0

888 self.output_queue.put_nowait(

885 - (-1,

889 + (client_index,

890 EngineCoreOutputs(wave_complete=self.current_wave)))

891 self.current_wave += 1

892

vllm/v1/engine/core_client.py

+19 -1

@@ -155,6 +155,11 @@ class EngineCoreClient(ABC):

155 kwargs: Optional[dict[str, Any]] = None) -> list[_R]:

156 raise NotImplementedError

157

158 + def dp_engines_running(self) -> bool:

159 + """Returns True id data parallel engines are collectively in a

160 + running state."""

161 + raise NotImplementedError

162 +

163 async def get_output_async(self) -> EngineCoreOutputs:

164 raise NotImplementedError

165

@@ -282,6 +287,9 @@ class InprocClient(EngineCoreClient):

287 kwargs: Optional[dict[str, Any]] = None) -> list[_R]:

288 return self.engine_core.collective_rpc(method, timeout, args, kwargs)

289

290 + def dp_engines_running(self) -> bool:

291 + return False

292 +

293

294 @dataclass

295 class BackgroundResources:

@@ -384,6 +392,9 @@ class MPClient(EngineCoreClient):

392 dp_size = parallel_config.data_parallel_size

393 dp_rank = parallel_config.data_parallel_rank

394

395 + # State used for data parallel.

396 + self.engines_running = False

397 +

398 # SPMD mode is where there is an LLM instance per DP rank and

399 # one core engine per LLM, see

400 # examples/offline_inference/data_parallel.py.

@@ -539,6 +550,9 @@ class MPClient(EngineCoreClient):

550 while self.pending_messages and self.pending_messages[-1][0].done:

551 self.pending_messages.pop()

552

553 + def dp_engines_running(self) -> bool:

554 + return self.engines_running

555 +

556

557 def _process_utility_output(output: UtilityOutput,

558 utility_results: dict[int, AnyFuture]):

@@ -562,6 +576,7 @@ class SyncMPClient(MPClient):

576 log_stats=log_stats,

577 )

578

579 + self.is_dp = self.vllm_config.parallel_config.data_parallel_size > 1

580 self.outputs_queue = queue.Queue[Union[EngineCoreOutputs, Exception]]()

581

582 # Ensure that the outputs socket processing thread does not have

@@ -623,6 +638,8 @@ class SyncMPClient(MPClient):

638 outputs = self.outputs_queue.get()

639 if isinstance(outputs, Exception):

640 raise self._format_exception(outputs) from None

641 + if outputs.wave_complete is not None:

642 + self.engines_running = False

643 return outputs

644

645 def _send_input(self, request_type: EngineCoreRequestType, request: Any):

@@ -650,6 +667,8 @@ class SyncMPClient(MPClient):

667 return future.result()

668

669 def add_request(self, request: EngineCoreRequest) -> None:

670 + if self.is_dp:

671 + self.engines_running = True

672 self._send_input(EngineCoreRequestType.ADD, request)

673

674 def abort_requests(self, request_ids: list[str]) -> None:

@@ -911,7 +930,6 @@ class DPAsyncMPClient(AsyncMPClient):

930 client_addresses: Optional[dict[str, str]] = None,

931 client_index: int = 0):

932 self.current_wave = 0

914 - self.engines_running = False

933 # To route aborts to the correct engine.

934 self.reqs_in_flight: dict[str, CoreEngine] = {}

935

vllm/v1/engine/llm_engine.py

+1 -1

@@ -160,7 +160,7 @@ class LLMEngine:

160 def has_unfinished_requests(self) -> bool:

161 has_unfinished = self.output_processor.has_unfinished_requests()

162 if self.dp_group is None:

163 - return has_unfinished

163 + return has_unfinished or self.engine_core.dp_engines_running()

164 return self.has_unfinished_requests_dp(has_unfinished)

165

166 def has_unfinished_requests_dp(self, has_unfinished: bool) -> bool: