Tour: One step across GPUs

1Ten processes, four GPUs, two separate worlds#

We start after start-up; the tour Tour: Start-up of vllm serve covers it. Two things decide the process layout. First, --api-server-count defaults to the DP size in internal mode, so there are two API servers and the CLI process stays as their supervisor (vllm/entrypoints/cli/serve.py:124-125, vllm/entrypoints/cli/serve.py:151-152). Second, each engine core of a dense model becomes an independent EngineCoreProc with DP size 1:

            if data_parallel and vllm_config.model_config.is_moe:
                # Set data parallel rank for this engine process.
                parallel_config.data_parallel_rank = dp_rank
                engine_core = DPEngineCoreProc(*args, **kwargs)
            else:
                # Non-MoE DP ranks are completely independent, so treat like DP=1.
                # Note that parallel_config.data_parallel_index will still reflect
                # the original DP rank.
                parallel_config.reconfigure_for_independent_dp_rank()
                engine_core = EngineCoreProc(*args, engine_index=dp_rank, **kwargs)

Engine 0 gets GPUs 0 and 1, engine 1 gets GPUs 2 and 3 ("[2, 3] for local_dp_rank=1", vllm/v1/engine/utils.py:441-445). Each engine core starts a MultiprocExecutor with two worker processes. The DP size is 1 inside each engine. So each pair of workers builds its own world of two ranks. It has TP group [0, 1], and PP and DP groups of one rank (Process groups: how vLLM slices its ranks). The workers rename themselves Worker_TP0 and Worker_TP1; the name has a DP part only when the worker's DP group is larger than 1 (vllm/v1/executor/multiproc_executor.py:1101-1109).

So there are 10 processes: CLI supervisor, 2 API servers, 1 DP coordinator, 2 engine cores and 4 workers (derived; The process and thread architecture of vllm serve gives the rules).

2Each API server picks an engine with the DPLB score#

We are in API server 0, in AsyncLLM.add_request, which calls the engine core client. With DP > 1 and internal load balancing, that is a DPLBAsyncMPClient. Before it sends, it stamps the request with the current wave and its own client_index:

    async def add_request_async(self, request: EngineCoreRequest) -> None:
        self._ensure_stats_update_task()

        request.current_wave = self.current_wave
        request.client_index = self.client_index

        # The engine that receives it wakes the other ranks if they are idle.
        chosen_engine = self.get_core_engine_for_request(request)
        await self._send_input(EngineCoreRequestType.ADD, request, chosen_engine)

        self._ensure_output_queue_task()

The engine choice is the score of Data parallel serving: several engine cores behind one endpoint: max(client_count × in-flight, waiting + running) plus a penalty for waiting requests on a full KV cache (vllm/v1/engine/core_client.py:1777-1786). Both engines are idle, so every score is 0. Ties go to the first engine of the scan, because only a strictly smaller score replaces the minimum (line 1787). The scan starts at a different engine per API server:

        self.eng_start_index = (
            len(self.core_engines) * self.client_index
        ) // client_count

• API server 0: (2 × 0) // 2 = 0, so request A goes to engine 0.
• API server 1: (2 × 1) // 2 = 1, so request B goes to engine 1.

Each server then adds client_count = 2 to the chosen engine's waiting count, adds 1 to its own in-flight count, and rotates its start index (vllm/v1/engine/core_client.py:1792-1803). (Derived from the cited lines.)

3The request crosses ZMQ to its engine core#

The engine identity is the DP rank as two little-endian bytes (vllm/v1/engine/core_client.py:748-751), so A's frames start with b"\x00\x00" and B's with b"\x01\x00". The ROUTER socket routes on that frame:

    def _send_input_message(
        self, message: tuple[bytestr, ...], engine: EngineIdentity
    ) -> Awaitable[Any]:
        self.ensure_alive()
        # Any zero-copy tensor/ndarray frames are kept alive by zmq itself
        # until it's finished sending them (there is a ref chain from the underlying
        # memoryview back to the original owning tensor/ndarray).
        return self.input_socket.send_multipart((engine,) + message, copy=False)

Each engine core connects one DEALER socket per API server. Its input thread receives the frames, decodes the EngineCoreRequest, builds the Request, and queues it for the busy loop (vllm/v1/engine/core.py:1977-1993). The request keeps its client_index, so the outputs can find their way back in stop 11. The page Processes, ZMQ and shared memory explains the frames.

4Each engine core schedules on its own#

Engine core 0 now has request A, engine core 1 has request B. Each runs the normal busy loop of The engine core loop: threads, queues and the busy loop: step_with_batch_queue schedules the 300 (or 200) prompt tokens, as in Tour: One engine step. Nothing ties the two engines together:

• They are plain EngineCoreProc objects, not DPEngineCoreProc, so there are no waves, no dummy batches and no DP all-reduce (stop 1).
• When an engine has no work, it blocks on its input queue: has_work() is false when nothing is running, waiting or in the batch queue (vllm/v1/engine/core.py:1596-1602).
• Their steps are not aligned in time. Engine 1 may run three steps while engine 0 runs two.

5One shared-memory write reaches both workers#

The engine core calls execute_model on its multiproc executor. That is one collective_rpc with a reply only from the output rank:

    def execute_model(  # type: ignore[override]
        self, scheduler_output: SchedulerOutput, non_block: bool = False
    ) -> ModelRunnerOutput | None | Future[ModelRunnerOutput | None]:
        return self.collective_rpc(
            "execute_model",
            args=(scheduler_output,),
            unique_reply_rank=self.output_rank,
            non_block=non_block,
            timeout=envs.VLLM_EXECUTE_MODEL_TIMEOUT_SECONDS,
            kv_output_aggregator=self.kv_output_aggregator,
            ec_output_aggregator=self.ec_output_aggregator,
        )

        if isinstance(method, str):
            send_method = method
        else:
            send_method = cloudpickle.dumps(method, protocol=pickle.HIGHEST_PROTOCOL)
        self.rpc_broadcast_mq.enqueue((send_method, args, kwargs, output_rank))

        response_mqs: Sequence[MessageQueue] = self.response_mqs
        if output_rank is not None:
            response_mqs = (response_mqs[output_rank],)

Line 428 is the only write. The MessageQueue pickles ("execute_model", (scheduler_output,), {}, 0) once into a 16 MiB chunk of a ring in /dev/shm. Both workers read the same bytes (distributed/device_communicators/shm_broadcast.py: MessageQueue). Lines 431-432 already decide that the executor will read only the reply queue of rank 0. The step deadline is VLLM_EXECUTE_MODEL_TIMEOUT_SECONDS, 300 s by default (vllm/envs.py:1861-1863).

6Each worker dequeues and runs execute_model#

We are now in one of the worker processes, in its main thread. It was blocked in the queue and wakes up with the message:

    def worker_busy_loop(self):
        """Main busy loop for Multiprocessing Workers."""
        assert self.rpc_broadcast_mq is not None
        while True:
            with get_watch_dog().disable():
                rpc_request = self.rpc_broadcast_mq.dequeue(indefinite=True)
            self._execute_worker_rpc(rpc_request)
            del rpc_request

    def _execute_worker_rpc(
        self,
        rpc_request: tuple[str | bytes, tuple[Any, ...], dict[str, Any], int | None],
    ) -> None:
        """Execute one RPC in a separate frame from the dequeue loop."""
        method, args, kwargs, output_rank = rpc_request
        try:
            if isinstance(method, str):
                func = getattr(self.worker, method)
            elif isinstance(method, bytes):
                func = partial(cloudpickle.loads(method), self.worker)

getattr(self.worker, "execute_model") leads to Worker.execute_model. With PP = 1 this worker is both the first and the last stage, so it receives no intermediate tensors and calls the model runner directly (vllm/v1/worker/gpu_worker.py:1336-1353). From here, both workers run the same Python code on the same batch, each with half of every weight matrix (The tensor-parallel layers (linear.py and vocab_parallel_embedding.py)). Model Runner V2 also checks for DP padding, but with a DP size of 1 there is nothing to agree on (vllm/v1/worker/gpu/dp_utils.py:278-279).

7The first collective: the embedding all-reduce#

Each rank holds half of the vocabulary rows of the embedding table. A rank looks up only the token ids in its half and writes zeros for the others; then the two partial results are summed:

        if not self.use_fused_embedding:
            output_parallel.masked_fill_(input_mask.unsqueeze(-1), 0)
        if self.reduce_results:
            output_parallel = (
                self.parallel_group.all_reduce(output_parallel)
                if self.parallel_group is not None
                else tensor_model_parallel_all_reduce(output_parallel)
            )
        return output_parallel.view(output_dtype)

For A's 300 tokens, the tensor is [300, 4096] BF16, 2.34 MiB, and it is summed over the TP group [GPU 0, GPU 1]. Engine 1 does the same for B on GPUs 2 and 3, at its own time. The tour Tour: One forward pass gives the shapes of every layer (its deep dive "What changes with TP = 2").

8Two all-reduces per decoder layer, on the TP group#

Inside each of the 32 decoder layers, o_proj and down_proj are row-parallel. Each rank computes a partial sum, and the layer adds them:

        if self.reduce_results and self.tp_size > 1:
            output = tensor_model_parallel_all_reduce(output_parallel)
        else:
            output = output_parallel

tensor_model_parallel_all_reduce goes to the TP group (vllm/distributed/communication_op.py:12-14). On CUDA, the group calls the registered op with only its name, so the compiled graph keeps one opaque node:

        # Bypass the function if we are using only 1 GPU.
        if self.world_size == 1:
            return input_

        if self.use_custom_op_call:
            return torch.ops.vllm.all_reduce(input_, group_name=self.unique_name)
        else:
            return self._all_reduce_out_place(input_)

The op ends in CudaCommunicator.all_reduce, which tries FlashInfer, the custom all-reduce, symmetric memory and pynccl in a fixed order (vllm/distributed/device_communicators/cuda_communicator.py:362-446). Which one takes our 2.34 MiB tensor depends on the GPU (derived from the size rules on Process groups: how vLLM slices its ranks):

• Compute capability 8.x (for example A100), TP = 2 on one node with P2P access: FlashInfer all-reduce has no size entry and stays off. The custom all-reduce takes the tensor, because 2.34 MiB is below its 8 MiB limit.
• Compute capability 9.0 or 10.x with FlashInfer installed: at O2 the compile pass fuse_allreduce_rms can fuse an all-reduce with the RMSNorm after it. A fused all-reduce does not reach this method (vllm/config/vllm.py:178-202, Tour: One forward pass). An all-reduce that is not fused goes to FlashInfer first; its limit at TP = 2 is 64 MiB (vllm/compilation/passes/fusion/allreduce_rms_fusion.py:104-115).

Either way, the step has 1 + 2 × 32 = 65 all-reduces in total, all on the TP group. The DP and PP groups have one rank and run nothing.

9sample_tokens: one all-gather, then both ranks sample#

The engine core next calls sample_tokens, the second broadcast of the step (same path as stop 5). Each rank computes logits for its half of the vocabulary; the LM head then gathers them:

    def _gather_logits(self, logits: torch.Tensor) -> torch.Tensor:
        """gather/all-gather the logits tensor across model parallel group."""
        if self.use_all_gather:
            # Gather is not supported for some devices such as TPUs.
            # Use all-gather instead.
            # NOTE(woosuk): Here, the outputs of every device should not be None
            # because XLA requires strict SPMD among all devices. Every device
            # should execute the same operations after gathering the logits.
            logits = tensor_model_parallel_all_gather(logits)
        else:
            # None may be returned for rank > 0
            logits = tensor_model_parallel_gather(logits)
        return logits

Every built-in platform uses the all-gather, so both ranks hold the full [num_rows, 128256] logits (vllm/platforms/interface.py:1275-1278). Both ranks run the sampler. They draw the same tokens. After warm-up, every worker resets its random seed to a value that depends only on the model seed and the DP index (vllm/v1/worker/gpu_worker.py:1083-1087, vllm/utils/torch_utils.py:533-544). A different DP index gives a different seed, so engines 0 and 1 do not sample unseeded requests alike (the comment at line 536). Each worker then draws the per-request seed of an unseeded request from this same NumPy state, in the same order (vllm/v1/worker/gpu/sample/states.py:54-58). By default, the code has no other synchronization of the sampled tokens between TP ranks (derived). The opt-in --enable-batch-sharded-sampling changes this: each rank samples a slice of the batch, and the ranks all-gather the results (vllm/config/parallel.py:192-198).

10Only TP rank 0 replies#

The output rank is world_size − TP × PCP, the first TP rank of the last PP stage (vllm/v1/executor/multiproc_executor.py:549-563); here that is 2 − 2 = 0. In each worker, the reply is conditional:

            if output_rank is None or self.rank == output_rank:
                self.handle_output(output)

Worker TP0 hands its AsyncOutput to its WorkerAsyncOutputCopy thread. That thread waits for the copy to the CPU and writes (SUCCESS, ModelRunnerOutput) into its own MessageQueue(1, 1) (vllm/v1/executor/multiproc_executor.py:1014-1039, vllm/v1/executor/multiproc_executor.py:1000-1012). Worker TP1 writes nothing; it goes back to its dequeue. The engine core reads the reply when it pops the step from its batch queue, exactly as in stop 10 of the engine-step tour.

11Outputs go back to the API server that sent the request#

After update_from_output, the engine core groups its outputs by client_index, and its output thread sends each group on the PUSH socket of that API server:

                buffer = reuse_buffers.pop() if reuse_buffers else bytearray()
                buffers = encoder.encode_into(outputs, buffer)
                tracker = self._send_msg_tracking_payload(
                    sockets[client_index], buffers
                )

Request A was sent by API server 0, so A's first token goes to API server 0's PULL socket; B's goes to API server 1. Each API server detokenizes only its own requests, and its DPLBAsyncMPClient lowers its in-flight count when a request finishes (vllm/v1/engine/core_client.py:1810-1817).

12Each engine core reports its new counts#

Each engine core compares its waiting and running counts with the last ones it sent. When they changed (here: one new running request), it sends them, with its KV cache usage, to client index −1, the coordinator:

    def _maybe_publish_request_counts(self):
        if not self.publish_dp_lb_stats:
            return

        # Publish our request counts (if they've changed).
        counts = self.scheduler.get_request_counts()
        if counts != self.last_counts:
            self.last_counts = counts
            stats = SchedulerStats(
                *counts, kv_cache_usage=self.scheduler.get_kv_cache_usage()
            )
            self.output_queue.put_nowait((-1, EngineCoreOutputs(scheduler_stats=stats)))

Separately, the scheduler stats of each step go to only one front end, inside the normal outputs (vllm/v1/core/sched/scheduler.py:2516-2521). Prometheus labels them with engine="0" or engine="1" (vllm/v1/metrics/loggers.py:499-504).

13The coordinator publishes the load to both API servers#

The coordinator stores the counts per engine and publishes all of them, at most every 100 ms while they change:

                events = poller.poll(timeout=max(min_timeout, wait_for - elapsed))
                if not events:
                    # Poller timeout - publish current stats to front-ends.
                    if last_step_counts is not None:
                        engine_req_counts_list = last_step_counts
                        last_step_counts = None
                    else:
                        engine_req_counts_list = self._get_engine_counts()
                        stats_changed = False

                    to_publish = (engine_req_counts_list, current_wave, engines_running)
                    publish_front.send(msgspec.msgpack.encode(to_publish))
                    last_publish_time = int(time.time() * 1000)
                    continue

Each API server's stats task replaces its lb_engines with the new counts (vllm/v1/engine/core_client.py:1578-1590). Real numbers now replace the local guesses of stop 2 (+2 waiting on the chosen engine). Engine 0 has 1 running request, and engine 1 has 1. The next request on either API server sees both, plus its own in-flight counts. Meanwhile, each engine core is already in its next step.