走进 vLLM（四）：从单卡到多机——MultiProcExecutor 与分布式服务

From UniprocExecutor to MultiProcExecutor

With the core techniques in place, we can now talk about scaling up.

Suppose your model weights no longer fit into a single GPU’s VRAM.

The first option is to shard the model across multiple GPUs on the same node using tensor parallelism (e.g., TP=8). If the model still doesn’t fit, the next step is pipeline parallelism across nodes.

Notes:

• Intranode bandwidth is significantly higher than internode, which is why tensor parallelism (TP) is generally preferred over pipeline parallelism (PP). (It is also true that PP communicates less data than TP.)
• I’m not covering expert parallelism (EP) since we’re focusing on standard transformers rather than MoE, nor sequence parallelism, as TP and PP are the most commonly used in practice.

At this stage, we need multiple GPU processes (workers) and an orchestration layer to coordinate them. That’s exactly what MultiProcExecutor provides.

How this works in vLLM:

1. MultiProcExecutor initializes an rpc_broadcast_mq message queue (implemented with shared memory under the hood).
2. The constructor loops over world_size (e.g. TP=8 ⇒ world_size=8) and spawns a daemon process for each rank via WorkerProc.make_worker_process.
3. For each worker, the parent first creates a reader and writer pipe.
4. The new process runs WorkerProc.worker_main, which instantiates a worker (going through the same “init device”, “load model”, etc. as in UniprocExecutor).
5. Each worker determines whether it is the driver (rank 0 in the TP group) or a regular worker. Every worker sets up two queues:

• rpc_broadcast_mq (shared with the parent) for receiving work.
• worker_response_mq for sending responses back.

6. During initialization, each child sends its worker_response_mq handle to the parent via the pipe. Once all are received, the parent unblocks — this completes coordination.
7. Workers then enter a busy loop, blocking on rpc_broadcast_mq.dequeue. When a work item arrives, they execute it (just like in UniprocExecutor, but now with TP/PP-specific partitioned work). Results are sent back through worker_response_mq.enqueue.
8. At runtime, when a request arrives, MultiProcExecutor enqueues it into rpc_broadcast_mq (non-blocking) for all children workers. It then waits on the designated output rank’s worker_response_mq.dequeue to collect the final result.

From the engine’s perspective, nothing has changed — all of this multiprocessing complexity is abstracted away through a call to model executor’s execute_model.

• In the UniProcExecutor case: execute_model directly leads to calling execute_model on the worker
• In the MultiProcExecutor case: execute_model indirectly leads to calling execute_model on each worker through rpc_broadcast_mq

At this point, we can run models that are as large as resources allow using the same engine interface.

The next step is to scale out: enable data parallelism (DP > 1) replicating the model across nodes, add a lightweight DP coordination layer, introduce load balancing across replicas, and place one or more API servers in front to handle incoming traffic.

Distributed system serving vLLM

There are many ways to set up serving infrastructure, but to stay concrete, here’s one example: suppose we have two H100 nodes and want to run four vLLM engines across them.

If the model requires TP=4, we can configure the nodes like this.

On the first node, run the engine in headless mode (no API server) with the following arguments:

and run that same command on the other node with few tweaks:

• no –headless
• modify DP start rank

说明

This assumes networking is configured so all nodes can reach the specified IP and port.

How does this work in VLLM?

On the headless server node

On the headless node, a CoreEngineProcManager launches 2 processes (per –data-parallel-size-local) each running EngineCoreProc.run_engine_core. Each of these functions creates a DPEngineCoreProc (the engine core) and then enters its busy loop.

DPEngineCoreProc initializes its parent EngineCoreProc (child of EngineCore), which:

1. Creates an input_queue and output_queue (queue.Queue).
2. Performs an initial handshake with the frontend on the other node using a DEALER ZMQ socket (async messaging lib), and receives coordination address info.
3. Initializes DP group (e.g. using NCCL backend).
4. Initializes the EngineCore with MultiProcExecutor (TP=4 on 4 GPUs as described earlier).
5. Creates a ready_event (threading.Event).
6. Starts an input deamon thread (threading.Thread) running process_input_sockets(…, ready_event). Similarly starts an output thread.
7. Still in the main thread, waits on ready_event until all input threads across all 4 processes (spanning the 2 nodes) have completed the coordination handshake finally executing ready_event.set().
8. Once unblocked, sends a “ready” message to the frontend with metadata (e.g., num_gpu_blocks available in paged KV cache memory).
9. The main, input, and output threads then enter their respective busy loops.

TL;DR: We end up with 4 child processes (one per DP replica), each running a main, input, and output thread. They complete a coordination handshake with the DP coordinator and frontend, then all three threads per process run in steady-state busy loops.

Current steady state:

• Input thread — blocks on the input socket until a request is routed from the API server; upon receipt, it decodes the payload, enqueues a work item via input_queue.put_nowait(…), and returns to blocking on the socket.
• Main thread — wakes on input_queue.get(…), feeds the request to the engine; MultiProcExecutor runs the forward pass and enqueues results to output_queue.
• Output thread — wakes on output_queue.get(…), sends the result back to the API server, then resumes blocking.

Additional mechanics:

• DP wave counter — the system tracks “waves”; when all engines become idle they quiesce, and the counter increments when new work arrives (useful for coordination/metrics).
• Control messages — the API server can send more than just inference requests (e.g., aborts and utility/control RPCs).
• Dummy steps for lockstep — if any DP replica has work, all replicas execute a forward step; replicas without requests perform a dummy step to participate in required synchronization points (avoids blocking the active replica).

说明

Lockstep clarification: this is actually only required for MoE models where the expert layers form an EP or TP group while attention layers are still DP. It’s currently always done with DP - this is just because there’s limited use for “built-in” non-MoE DP since you could just run multiple independent vLLMs and load-balance between them in a normal way.

Now for the second part, what happens on the API server node?

On the API server node

We instantiate an AsyncLLM object (an asyncio wrapper around the LLM engine). Internally this creates a DPLBAsyncMPClient (data-parallel, load-balancing, asynchronous, multiprocessing client).

Inside the parent class of MPClient, the launch_core_engines function runs and:

1. Creates the ZMQ addresses used for the startup handshake (as seen on the headless node).
2. Spawns a DPCoordinator process.
3. Creates a CoreEngineProcManager (same as on the headless node).

Inside AsyncMPClient (child of MPClient), we:

1. Create an outputs_queue (asyncio.Queue).
2. We create an asyncio task process_outputs_socket which communicates (through the output socket) with output threads of all 4 DPEngineCoreProc and writes into outputs_queue.
3. Subsequently one more asyncio task output_handler from AsyncLLM reads from this queue and finally sends out information to the create_completion function.

Inside DPAsyncMPClient we create an asyncio task run_engine_stats_update_task which communicates with DP coordinator.

The DP coordinator mediates between the frontend (API server) and backend (engine cores). It:

• Periodically sends load-balancing info (queue sizes, waiting/running requests) to the frontend’s run_engine_stats_update_task.
• Handles SCALE_ELASTIC_EP commands from the frontend by dynamically changing the number of engines (only works with Ray backend).
• Sends START_DP_WAVE events to the backend (when triggered by frontend) and reports wave-state updates back.

To recap, the frontend (AsyncLLM) runs several asyncio tasks (remember: concurrent, not parallel):

• A class of tasks handles input requests through the generate path (each new client request spawns a new asyncio task).
• Two tasks (process_outputs_socket, output_handler) process output messages from the underlying engines.
• One task (run_engine_stats_update_task) maintains communication with the DP coordinator: sending wave triggers, polling LB state, and handling dynamic scaling requests.

Finally, the main server process creates a FastAPI app and mounts endpoints such as OpenAIServingCompletion and OpenAIServingChat, which expose /completion, /chat/completion, and others. The stack is then served via Uvicorn.

So, putting it all together, here’s the full request lifecycle!

You send from your terminal:

What happens next:

1. The request hits OpenAIServingCompletion’s create_completion route on the API server.
2. The function tokenizes the prompt asynchronously, and prepares metadata (request ID, sampling params, timestamp, etc.).
3. It then calls AsyncLLM.generate, which follows the same flow as the synchronous engine, eventually invoking DPAsyncMPClient.add_request_async.
4. This in turn calls get_core_engine_for_request, which does load balancing across engines based on the DP coordinator’s state (picking the one that has minimal score / lowest load: score = len(waiting) * 4 + len(running)).
5. The ADD request is sent to the chosen engine’s input_socket.
6. At that engine:

• Input thread — unblocks, decodes data from the input socket, and places a work item on the input_queue for the main thread.
• Main thread — unblocks on input_queue, adds the request to the engine, and repeatedly calls engine_core.step(), enqueueing intermediate results to output_queue until a stop condition is met.

说明

Reminder: step() calls the scheduler, model executor (which in turn can be MultiProcExecutor!), etc. We have already seen this!

• Output thread — unblocks on output_queue and sends results back through the output socket.

7. Those results trigger the AsyncLLM output asyncio tasks (process_outputs_socket and output_handler), which propagate tokens back to FastAPI’s create_completion route.
8. FastAPI attaches metadata (finish reason, logprobs, usage info, etc.) and returns a JSONResponse via Uvicorn to your terminal!

And just like that, your completion came back — the whole distributed machinery hidden behind a simple curl command! :) So much fun!!!

Additional notes:

• When adding more API servers, load balancing is handled at the OS/socket level. From the application’s perspective, nothing significant changes — the complexity is hidden.
• With Ray as a DP backend, you can expose a URL endpoint (/scale_elastic_ep) that enables automatic scaling of the number of engine replicas up or down.