[CI/Build][The Rock][BugFix] Use fork method in test_multiproc_executor_multi_node for py 3.14 compat and fix test_multiproc_executor_shutdown_cleanup (#48655)
Signed-off-by: Randall Smith <Randall.Smith@amd.com> Co-authored-by: Andreas Karatzas <akaratza@amd.com>
rasmith committed
Jul 21, 2026 at 15:40 UTC
60d443f738bfc13001973d36445f022cba499721
2 files changed
+19
-5
tests/distributed/test_multiproc_executor.py
+8
-5
@@ -254,8 +254,8 @@ def test_multiproc_executor_shutdown_cleanup():
254
for worker in executor.workers:
255
assert not worker.proc.is_alive(), "Worker processes should be terminated"
256
257
- # Verify shutdown event is set
258
- assert executor.shutdown_event.is_set(), "Shutdown event should be set"
257
+ # Verify shutdown flag is set
258
+ assert executor.shutting_down, "Shutdown flag should be set"
259
260
# Multiple shutdowns should be safe (idempotent)
261
executor.shutdown()
@@ -292,7 +292,6 @@ def test_multiproc_executor_pipeline_parallel():
292
"Max concurrent batches should follow the configured PP/async "
293
"scheduling policy"
294
)
295
-
295
finally:
296
# Clean up
297
executor.shutdown()
@@ -338,6 +337,10 @@ def test_multiproc_executor_multi_node():
337
- Node 1 (rank 1): Uses GPUs 2,3 (CUDA_VISIBLE_DEVICES=2,3) with TP=2
338
Total world_size = 4, nnodes = 2
339
"""
340
+ # Python 3.14+ changed default multiprocessing start method to 'forkserver'
341
+ # which cannot pickle nested functions. Use 'fork' for this test.
342
+ mp_ctx = multiprocessing.get_context("fork")
343
+
344
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
345
s.bind(("", 0))
346
port = s.getsockname()[1]
@@ -405,12 +408,12 @@ def test_multiproc_executor_multi_node():
408
executor.shutdown()
409
410
# Create a queue to collect results from both processes
408
- result_queue: multiprocessing.Queue[dict[str, int | bool]] = multiprocessing.Queue()
411
+ result_queue: multiprocessing.Queue[dict[str, int | bool]] = mp_ctx.Queue()
412
413
# Start both node processes
414
processes = []
415
for node_rank in range(2):
413
- p = multiprocessing.Process(
416
+ p = mp_ctx.Process(
417
target=run_node,
418
args=(node_rank, result_queue, port),
419
name=f"Node{node_rank}",
vllm/v1/executor/multiproc_executor.py
+11
@@ -246,6 +246,17 @@ class MultiprocExecutor(Executor):
246
247
self.output_rank = self._get_output_rank()
248
249
+ def get_response_mqs(self, unique_reply_rank: int = -1) -> list[MessageQueue]:
250
+ assert unique_reply_rank >= -1 and unique_reply_rank < self.world_size, (
251
+ f"unique_reply_rank must be -1 or < world_size,"
252
+ f"unique_reply_rank = {unique_reply_rank}, "
253
+ f"world_size={self.world_size}"
254
+ )
255
+ ranks = (
256
+ [unique_reply_rank] if unique_reply_rank != -1 else range(self.world_size)
257
+ )
258
+ return [self.workers[rank].worker_response_mq for rank in ranks]
259
+
260
def _get_parallel_sizes(self) -> tuple[int, int, int]:
261
self.world_size = self.parallel_config.world_size
262
assert self.world_size % self.parallel_config.nnodes_within_dp == 0, (