| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | |
| 4 | """Integration tests for MultiprocExecutor at the executor level. |
| 5 | This test directly tests the executor without going through the LLM interface, |
| 6 | focusing on executor initialization, RPC calls, and distributed execution. |
| 7 | """ |
| 8 | |
| 9 | import multiprocessing |
| 10 | import os |
| 11 | import socket |
| 12 | |
| 13 | import pytest |
| 14 | |
| 15 | from tests.utils import multi_gpu_test |
| 16 | from vllm.config import VllmConfig |
| 17 | from vllm.engine.arg_utils import EngineArgs |
| 18 | from vllm.platforms import current_platform |
| 19 | from vllm.v1.core.sched.output import SchedulerOutput |
| 20 | from vllm.v1.executor import multiproc_executor |
| 21 | from vllm.v1.executor.multiproc_executor import MultiprocExecutor |
| 22 | |
| 23 | MODEL = "facebook/opt-125m" |
| 24 | |
| 25 | |
| 26 | @pytest.mark.parametrize( |
| 27 | ("local_world_size", "data_parallel_size_local", "expected_num_local_procs"), |
| 28 | [(1, 4, 4), (4, 1, 4), (2, 0, 2)], |
| 29 | ) |
| 30 | def test_multiproc_executor_counts_all_local_dp_workers( |
| 31 | monkeypatch: pytest.MonkeyPatch, |
| 32 | local_world_size: int, |
| 33 | data_parallel_size_local: int, |
| 34 | expected_num_local_procs: int, |
| 35 | ): |
| 36 | """All colocated DP workers share the node's startup CPU budget.""" |
| 37 | executor = object.__new__(MultiprocExecutor) |
| 38 | executor.world_size = local_world_size |
| 39 | executor.local_world_size = local_world_size |
| 40 | executor.parallel_config = type( |
| 41 | "ParallelConfig", |
| 42 | (), |
| 43 | {"data_parallel_size_local": data_parallel_size_local}, |
| 44 | )() |
| 45 | |
| 46 | monkeypatch.setattr( |
| 47 | executor, |
| 48 | "_get_parallel_sizes", |
| 49 | lambda: (local_world_size, 1, 1), |
| 50 | ) |
| 51 | |
| 52 | class StopExecutorInit(Exception): |
| 53 | pass |
| 54 | |
| 55 | def capture_num_local_procs(num_local_procs: int): |
| 56 | assert num_local_procs == expected_num_local_procs |
| 57 | raise StopExecutorInit |
| 58 | |
| 59 | monkeypatch.setattr( |
| 60 | multiproc_executor, |
| 61 | "set_multiprocessing_worker_envs", |
| 62 | capture_num_local_procs, |
| 63 | ) |
| 64 | |
| 65 | with pytest.raises(StopExecutorInit): |
| 66 | executor._init_executor() |
| 67 | executor._finalizer.detach() |
| 68 | |
| 69 | |
| 70 | def create_vllm_config( |
| 71 | tensor_parallel_size: int = 1, |
| 72 | pipeline_parallel_size: int = 1, |
| 73 | max_model_len: int = 256, |
| 74 | gpu_memory_utilization: float = 0.3, |
| 75 | distributed_executor_backend: str = "mp", |
| 76 | nnodes: int = 1, |
| 77 | node_rank: int = 0, |
| 78 | master_port: int = 0, |
| 79 | ) -> VllmConfig: |
| 80 | """Create a VllmConfig for testing using EngineArgs.""" |
| 81 | engine_args = EngineArgs( |
| 82 | model=MODEL, |
| 83 | tensor_parallel_size=tensor_parallel_size, |
| 84 | pipeline_parallel_size=pipeline_parallel_size, |
| 85 | max_model_len=max_model_len, |
| 86 | gpu_memory_utilization=gpu_memory_utilization, |
| 87 | distributed_executor_backend=distributed_executor_backend, |
| 88 | enforce_eager=True, |
| 89 | ) |
| 90 | vllm_config = engine_args.create_engine_config() |
| 91 | |
| 92 | # Override distributed node settings if needed |
| 93 | if nnodes > 1 or node_rank > 0: |
| 94 | vllm_config.parallel_config.nnodes = nnodes |
| 95 | vllm_config.parallel_config.node_rank = node_rank |
| 96 | vllm_config.parallel_config.master_port = master_port |
| 97 | if nnodes > 1: |
| 98 | vllm_config.parallel_config.disable_custom_all_reduce = True |
| 99 | |
| 100 | return vllm_config |
| 101 | |
| 102 | |
| 103 | def create_test_scheduler_output(num_requests: int = 1) -> SchedulerOutput: |
| 104 | """Create a minimal SchedulerOutput for testing.""" |
| 105 | # This is a simplified version - in practice you'd need proper |
| 106 | # SchedulerOutput construction based on the actual vLLM v1 API |
| 107 | return SchedulerOutput( |
| 108 | scheduled_new_reqs=[], |
| 109 | scheduled_resumed_reqs=[], |
| 110 | scheduled_running_reqs=[], |
| 111 | num_scheduled_tokens={}, |
| 112 | total_num_scheduled_tokens=0, |
| 113 | ) |
| 114 | |
| 115 | |
| 116 | def test_multiproc_executor_initialization(): |
| 117 | """Test that MultiprocExecutor can be initialized with proper config.""" |
| 118 | vllm_config = create_vllm_config( |
| 119 | tensor_parallel_size=1, |
| 120 | pipeline_parallel_size=1, |
| 121 | ) |
| 122 | |
| 123 | # Create executor - this should initialize workers |
| 124 | executor = MultiprocExecutor(vllm_config=vllm_config) |
| 125 | |
| 126 | # Verify executor properties |
| 127 | assert executor.world_size == 1, "World size should be 1 for single GPU" |
| 128 | assert executor.local_world_size == 1, "Local world size should be 1" |
| 129 | assert hasattr(executor, "workers"), "Executor should have workers" |
| 130 | assert len(executor.workers) == 1, "Should have 1 worker for single GPU" |
| 131 | |
| 132 | # Clean up |
| 133 | executor.shutdown() |
| 134 | |
| 135 | |
| 136 | @multi_gpu_test(num_gpus=2) |
| 137 | def test_multiproc_executor_initialization_tensor_parallel(): |
| 138 | """Test MultiprocExecutor initialization with tensor parallelism.""" |
| 139 | vllm_config = create_vllm_config( |
| 140 | tensor_parallel_size=2, |
| 141 | pipeline_parallel_size=1, |
| 142 | ) |
| 143 | |
| 144 | # Create executor |
| 145 | executor = MultiprocExecutor(vllm_config=vllm_config) |
| 146 | |
| 147 | # Verify executor properties |
| 148 | assert executor.world_size == 2, "World size should be 2 for TP=2" |
| 149 | assert executor.local_world_size == 2, "Local world size should be 2" |
| 150 | assert len(executor.workers) == 2, "Should have 2 workers for TP=2" |
| 151 | |
| 152 | # Verify output rank calculation |
| 153 | output_rank = executor._get_output_rank() |
| 154 | assert output_rank == 0, "Output rank should be 0 for TP=2, PP=1" |
| 155 | |
| 156 | # Clean up |
| 157 | executor.shutdown() |
| 158 | |
| 159 | |
| 160 | @multi_gpu_test(num_gpus=2) |
| 161 | def test_multiproc_executor_collective_rpc(): |
| 162 | """Test collective RPC calls to all workers.""" |
| 163 | vllm_config = create_vllm_config( |
| 164 | tensor_parallel_size=2, |
| 165 | pipeline_parallel_size=1, |
| 166 | ) |
| 167 | |
| 168 | # Create executor |
| 169 | executor = MultiprocExecutor(vllm_config=vllm_config) |
| 170 | |
| 171 | try: |
| 172 | # Test check_health RPC - should work without errors |
| 173 | executor.check_health() |
| 174 | |
| 175 | # Test that RPC works correctly |
| 176 | # Note: We're just testing that the RPC mechanism works, |
| 177 | # not testing actual model execution here |
| 178 | assert not executor.is_failed, "Executor should not be in failed state" |
| 179 | |
| 180 | finally: |
| 181 | # Clean up |
| 182 | executor.shutdown() |
| 183 | |
| 184 | |
| 185 | def test_multiproc_executor_failure_callback(): |
| 186 | """Test failure callback registration and invocation.""" |
| 187 | vllm_config = create_vllm_config( |
| 188 | tensor_parallel_size=1, |
| 189 | pipeline_parallel_size=1, |
| 190 | ) |
| 191 | |
| 192 | executor = MultiprocExecutor(vllm_config=vllm_config) |
| 193 | |
| 194 | try: |
| 195 | # Test callback registration |
| 196 | callback_invoked = [] |
| 197 | |
| 198 | def test_callback(): |
| 199 | callback_invoked.append(True) |
| 200 | |
| 201 | # Register callback |
| 202 | executor.register_failure_callback(test_callback) |
| 203 | |
| 204 | # Callback should not be invoked yet |
| 205 | assert len(callback_invoked) == 0, "Callback should not be invoked immediately" |
| 206 | |
| 207 | # Simulate failure |
| 208 | executor.is_failed = True |
| 209 | |
| 210 | # Register another callback - should be invoked immediately |
| 211 | executor.register_failure_callback(test_callback) |
| 212 | assert len(callback_invoked) == 1, ( |
| 213 | "Callback should be invoked when executor is failed" |
| 214 | ) |
| 215 | |
| 216 | finally: |
| 217 | # Clean up |
| 218 | executor.shutdown() |
| 219 | |
| 220 | |
| 221 | @multi_gpu_test(num_gpus=2) |
| 222 | def test_multiproc_executor_worker_monitor(): |
| 223 | """Test that worker monitor is set up correctly.""" |
| 224 | vllm_config = create_vllm_config( |
| 225 | tensor_parallel_size=2, |
| 226 | pipeline_parallel_size=1, |
| 227 | ) |
| 228 | |
| 229 | executor = MultiprocExecutor(vllm_config=vllm_config) |
| 230 | |
| 231 | try: |
| 232 | # Verify all worker processes are alive |
| 233 | for worker in executor.workers: |
| 234 | assert worker.proc.is_alive(), f"Worker rank {worker.rank} should be alive" |
| 235 | |
| 236 | # Verify executor is not in failed state |
| 237 | assert not executor.is_failed, "Executor should not be in failed state" |
| 238 | |
| 239 | finally: |
| 240 | # Clean up |
| 241 | executor.shutdown() |
| 242 | |
| 243 | # After shutdown, workers should be terminated |
| 244 | import time |
| 245 | |
| 246 | time.sleep(0.5) # Give processes time to terminate |
| 247 | for worker in executor.workers: |
| 248 | assert not worker.proc.is_alive(), ( |
| 249 | f"Worker rank {worker.rank} should terminate after shutdown" |
| 250 | ) |
| 251 | |
| 252 | |
| 253 | @multi_gpu_test(num_gpus=2) |
| 254 | def test_multiproc_executor_get_response_message_queues(): |
| 255 | """Test message queue retrieval for different ranks.""" |
| 256 | vllm_config = create_vllm_config( |
| 257 | tensor_parallel_size=2, |
| 258 | pipeline_parallel_size=1, |
| 259 | ) |
| 260 | |
| 261 | executor = MultiprocExecutor(vllm_config=vllm_config) |
| 262 | |
| 263 | try: |
| 264 | # Get all message queues |
| 265 | all_queues = executor.get_response_mqs() |
| 266 | assert len(all_queues) == 2, "Should have 2 message queues for 2 workers" |
| 267 | |
| 268 | # Get message queue for specific rank |
| 269 | rank0_queue = executor.get_response_mqs(unique_reply_rank=0) |
| 270 | assert len(rank0_queue) == 1, "Should have 1 message queue for rank 0" |
| 271 | |
| 272 | rank1_queue = executor.get_response_mqs(unique_reply_rank=1) |
| 273 | assert len(rank1_queue) == 1, "Should have 1 message queue for rank 1" |
| 274 | |
| 275 | finally: |
| 276 | # Clean up |
| 277 | executor.shutdown() |
| 278 | |
| 279 | |
| 280 | def test_multiproc_executor_shutdown_cleanup(): |
| 281 | """Test that shutdown properly cleans up resources.""" |
| 282 | vllm_config = create_vllm_config( |
| 283 | tensor_parallel_size=1, |
| 284 | pipeline_parallel_size=1, |
| 285 | ) |
| 286 | |
| 287 | executor = MultiprocExecutor(vllm_config=vllm_config) |
| 288 | |
| 289 | # Verify executor is set up |
| 290 | assert hasattr(executor, "workers"), "Executor should have workers" |
| 291 | assert len(executor.workers) > 0, "Should have at least one worker" |
| 292 | |
| 293 | # Shutdown |
| 294 | executor.shutdown() |
| 295 | |
| 296 | # Verify cleanup |
| 297 | import time |
| 298 | |
| 299 | time.sleep(0.5) # Give processes time to terminate |
| 300 | |
| 301 | for worker in executor.workers: |
| 302 | assert not worker.proc.is_alive(), "Worker processes should be terminated" |
| 303 | |
| 304 | # Verify shutdown flag is set |
| 305 | assert executor.shutting_down, "Shutdown flag should be set" |
| 306 | |
| 307 | # Multiple shutdowns should be safe (idempotent) |
| 308 | executor.shutdown() |
| 309 | executor.shutdown() |
| 310 | |
| 311 | |
| 312 | @multi_gpu_test(num_gpus=4) |
| 313 | def test_multiproc_executor_pipeline_parallel(): |
| 314 | """Test MultiprocExecutor with pipeline parallelism.""" |
| 315 | vllm_config = create_vllm_config( |
| 316 | tensor_parallel_size=2, |
| 317 | pipeline_parallel_size=2, |
| 318 | ) |
| 319 | |
| 320 | executor = MultiprocExecutor(vllm_config=vllm_config) |
| 321 | |
| 322 | try: |
| 323 | # Verify executor properties |
| 324 | assert executor.world_size == 4, "World size should be 4 for TP=2, PP=2" |
| 325 | assert len(executor.workers) == 4, "Should have 4 workers" |
| 326 | |
| 327 | # Verify output rank calculation |
| 328 | # For TP=2, PP=2: output should be from the last PP stage (ranks 2-3) |
| 329 | # Specifically rank 2 (first rank of last PP stage) |
| 330 | output_rank = executor._get_output_rank() |
| 331 | assert output_rank == 2, "Output rank should be 2 (first rank of last PP stage)" |
| 332 | |
| 333 | # V2 model runner uses one extra batch to overlap async scheduling. |
| 334 | expected_concurrent_batches = 2 + int( |
| 335 | vllm_config.scheduler_config.async_scheduling |
| 336 | and vllm_config.use_v2_model_runner |
| 337 | ) |
| 338 | assert vllm_config.max_concurrent_batches == expected_concurrent_batches, ( |
| 339 | "Max concurrent batches should follow the configured PP/async " |
| 340 | "scheduling policy" |
| 341 | ) |
| 342 | finally: |
| 343 | # Clean up |
| 344 | executor.shutdown() |
| 345 | |
| 346 | |
| 347 | def test_multiproc_executor_properties(): |
| 348 | """Test various executor properties and configurations.""" |
| 349 | vllm_config = create_vllm_config( |
| 350 | tensor_parallel_size=1, |
| 351 | pipeline_parallel_size=1, |
| 352 | ) |
| 353 | |
| 354 | executor = MultiprocExecutor(vllm_config=vllm_config) |
| 355 | |
| 356 | try: |
| 357 | # Test supports_pp property |
| 358 | assert MultiprocExecutor.supports_pp is True, ( |
| 359 | "MultiprocExecutor should support pipeline parallelism" |
| 360 | ) |
| 361 | |
| 362 | # Test world_size calculation |
| 363 | assert executor.world_size == ( |
| 364 | executor.parallel_config.tensor_parallel_size |
| 365 | * executor.parallel_config.pipeline_parallel_size |
| 366 | ), "World size should equal TP * PP" |
| 367 | |
| 368 | # Test local_world_size calculation |
| 369 | assert executor.local_world_size == ( |
| 370 | executor.parallel_config.world_size // executor.parallel_config.nnodes |
| 371 | ), "Local world size should be world_size / nnodes" |
| 372 | |
| 373 | finally: |
| 374 | # Clean up |
| 375 | executor.shutdown() |
| 376 | |
| 377 | |
| 378 | @multi_gpu_test(num_gpus=4) |
| 379 | def test_multiproc_executor_multi_node(): |
| 380 | """Test MultiprocExecutor with multi-node configuration. |
| 381 | This simulates 2 nodes with TP=4: |
| 382 | - Node 0 (rank 0): Uses GPUs 0,1 (device_control_env_var=0,1) with TP=2 |
| 383 | - Node 1 (rank 1): Uses GPUs 2,3 (device_control_env_var=2,3) with TP=2 |
| 384 | Total world_size = 4, nnodes = 2 |
| 385 | """ |
| 386 | # Python 3.14+ changed default multiprocessing start method to 'forkserver' |
| 387 | # which cannot pickle nested functions. Use 'fork' for this test. |
| 388 | mp_ctx = multiprocessing.get_context("fork") |
| 389 | |
| 390 | with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: |
| 391 | s.bind(("", 0)) |
| 392 | port = s.getsockname()[1] |
| 393 | # symm_mem does not work for simulating multi instance in single node |
| 394 | os.environ["VLLM_ALLREDUCE_USE_SYMM_MEM"] = "0" |
| 395 | |
| 396 | # Use the platform-specific device control env var (e.g. |
| 397 | # CUDA_VISIBLE_DEVICES on CUDA, ZE_AFFINITY_MASK on XPU) so that each |
| 398 | # node only sees its own devices on different platform. |
| 399 | device_control_env_var = current_platform.device_control_env_var |
| 400 | |
| 401 | def run_node(node_rank: int, result_queue: multiprocessing.Queue, port: int): |
| 402 | """Run a single node's executor.""" |
| 403 | executor = None |
| 404 | try: |
| 405 | # Restrict visible devices for this node |
| 406 | if node_rank == 0: |
| 407 | os.environ[device_control_env_var] = "0,1" |
| 408 | else: |
| 409 | os.environ[device_control_env_var] = "2,3" |
| 410 | |
| 411 | # Create config for this node |
| 412 | vllm_config = create_vllm_config( |
| 413 | tensor_parallel_size=4, # Total TP across all nodes |
| 414 | pipeline_parallel_size=1, |
| 415 | nnodes=2, # 2 nodes |
| 416 | node_rank=node_rank, |
| 417 | master_port=port, # same port |
| 418 | ) |
| 419 | |
| 420 | # Create executor for this node |
| 421 | executor = MultiprocExecutor(vllm_config=vllm_config) |
| 422 | |
| 423 | # Verify node-specific properties |
| 424 | assert executor.world_size == 4, ( |
| 425 | f"World size should be 4 on node {node_rank}" |
| 426 | ) |
| 427 | assert executor.local_world_size == 2, ( |
| 428 | f"Local world size should be 2 on node {node_rank}" |
| 429 | ) |
| 430 | assert len(executor.workers) == 2, ( |
| 431 | f"Should have 2 local workers on node {node_rank}" |
| 432 | ) |
| 433 | |
| 434 | # Verify worker ranks are correct for this node |
| 435 | expected_ranks = [node_rank * 2, node_rank * 2 + 1] |
| 436 | actual_ranks = sorted([w.rank for w in executor.workers]) |
| 437 | assert actual_ranks == expected_ranks, ( |
| 438 | f"Node {node_rank} should have workers " |
| 439 | f"with ranks {expected_ranks}, got {actual_ranks}" |
| 440 | ) |
| 441 | # Verify all workers are alive |
| 442 | for worker in executor.workers: |
| 443 | assert worker.proc.is_alive(), ( |
| 444 | f"Worker rank {worker.rank} should be alive on node {node_rank}" |
| 445 | ) |
| 446 | # executor.gen |
| 447 | # Put success result in queue BEFORE shutdown to avoid hanging |
| 448 | result_queue.put({"node": node_rank, "success": True}) |
| 449 | import time |
| 450 | |
| 451 | time.sleep(2) |
| 452 | executor.shutdown() |
| 453 | except Exception as e: |
| 454 | # Put failure result in queue |
| 455 | result_queue.put({"node": node_rank, "success": False, "error": str(e)}) |
| 456 | raise e |
| 457 | finally: |
| 458 | if executor is not None: |
| 459 | executor.shutdown() |
| 460 | |
| 461 | # Create a queue to collect results from both processes |
| 462 | result_queue: multiprocessing.Queue[dict[str, int | bool]] = mp_ctx.Queue() |
| 463 | |
| 464 | # Start both node processes |
| 465 | processes = [] |
| 466 | for node_rank in range(2): |
| 467 | p = mp_ctx.Process( |
| 468 | target=run_node, |
| 469 | args=(node_rank, result_queue, port), |
| 470 | name=f"Node{node_rank}", |
| 471 | ) |
| 472 | p.start() |
| 473 | processes.append(p) |
| 474 | |
| 475 | # Wait for both processes to complete |
| 476 | all_completed = True |
| 477 | for p in processes: |
| 478 | p.join(timeout=60) |
| 479 | if p.is_alive(): |
| 480 | p.terminate() |
| 481 | p.join(timeout=20) |
| 482 | if p.is_alive(): |
| 483 | p.kill() |
| 484 | p.join() |
| 485 | all_completed = False |
| 486 | |
| 487 | # Check results from both nodes |
| 488 | results: list[dict[str, int | bool]] = [] |
| 489 | while len(results) < 2: |
| 490 | try: |
| 491 | result = result_queue.get(timeout=1) |
| 492 | results.append(result) |
| 493 | except Exception: |
| 494 | pass |
| 495 | assert all_completed, "Not all processes completed successfully" |
| 496 | assert len(results) == 2, f"Expected 2 results, got {len(results)}" |
| 497 | assert results[0]["success"], f"Node 0 failed: {results[0]}" |
| 498 | assert results[1]["success"], f"Node 1 failed: {results[1]}" |