RayDistributedExecutor: scaling workers across multi-node multi-GPU with Ray

源码版本v0.25.1

Responsibilities

In the single-GPU case UniProcExecutor gets the job done in one process, but as soon as TP>1 or cross-node execution is involved, inter-process communication is mandatory. RayDistributedExecutor(ray_executor.py:64-68) is the Ray-backed implementation: it wraps each worker as a Ray actor, distributes them onto the GPUs specified by a placement group, and then uses a Ray Compiled Graph to wire up a pipeline that strings the forward pass together. uses_ray = True、supports_pp = True(ray_executor.py:67-68) signals both that it uses Ray for orchestration and that it supports pipeline parallelism.

It inherits from Executor just like UniProcExecutor; the base class's execute_model/sample_tokens/profile/sleep signatures are unchanged, and only _init_executor, collective_rpc, execute_model, check_health, shutdown are overridden(ray_executor.py:70-97). The only difference is how the RPC travels: in-process is a direct method call, Ray is worker.execute_method.remote(...)(ray_executor.py:484-489).

Design motivation

• Ray actor maps one-to-one to worker: one RayWorkerWrapper actor per GPU(ray_executor.py:200-213); Ray handles scheduling, fault tolerance, and resource accounting, so the executor does not spawn processes itself.
• placement group pins the topology: actors are bound to bundles via PlacementGroupSchedulingStrategy(ray_executor.py:192-196); GPU affinity is fixed before startup and does not drift at runtime.
• Same-node-first ordering: actor creation order is Ray's decision (potentially unordered), so after creating all actors they are re-sorted by IP, with actors co-located with the driver ranked first(ray_executor.py:234-258), then collective_rpc("adjust_rank", args=(rerank_mapping,)) resets the rank.
• Ray Compiled Graph runs the forward: instead of RPC-calling execute_model every step, a forward_dag is built(ray_executor.py:527-620); with PP=2/TP=4 the four workers are strung into InputNode -> TP group -> MultiOutputNode, and intermediate tensors use with_tensor_transport(transport="nccl"|"shm") to go over NCCL or shared memory(ray_executor.py:582-591).
• Environment variables auto-replicated: vLLM-related env vars on the driver process are batch-pushed to every worker via update_environment_variables(ray_executor.py:305-327), except for WORKER_SPECIFIC_ENV_VARS, to avoid cross-worker pollution.
• execute_model split into two stages: with the sampler enabled, execute_model does not run immediately but only stores self.scheduler_output, waiting for sample_tokens to trigger _execute_dag(ray_executor.py:390-432), so the scheduler and the sampler's grammar bitmask computation can overlap with the forward.

Key files

• RayDistributedExecutor._init_executor:64-97 — initialize_ray_cluster, obtain placement group, call _init_workers_ray, initialize has_connector, uses_sampler, and the scheduler_output placeholder.
• shutdown:99-113 — forward_dag.teardown() then ray.kill(worker) one by one; the log notes that SIGTERM is expected behavior.
• _init_workers_ray:143-213 — for each bundle_index, ray.remote(...)(RayWorkerWrapper).remote(rpc_rank=rank); GPUs use num_gpus=num_gpus, other platforms go through resources={ray_device_key: num_gpus}.
• rerank:217-258 — ray.get fetches each actor's node IP, sorts by "same driver node first -> fewer actors on the node first -> smaller IP first", then collective_rpc("adjust_rank", ...) updates rpc_rank.
• init_worker + load_model:343-368 — assembles all_kwargs then issues collective_rpc("init_worker") + collective_rpc("init_device") + collective_rpc("load_model") in one shot.
• pp_tp_workers:370-378 — fills self.pp_tp_workers in a PP×TP 2D layout for the compiled DAG construction.
• execute_model + sample_tokens:390-432 — two stages: execute_model only stores scheduler_output; the real run is in sample_tokens via _execute_dag.
• _execute_dag:434-468 — on first call self.forward_dag = self._compiled_ray_dag(...), forward_dag.execute((scheduler_output, grammar_output)) returns refs; with PP disabled, refs[0].get() blocks for the result, otherwise a FutureWrapper is returned.
• collective_rpc:470-495 — serializes the method (cloudpickle.dumps), calls execute_method.remote(...) per worker, then ray.get(ray_worker_outputs, timeout=timeout); non_block=True returns a FutureWrapper.
• _compiled_ray_dag:527-620 — composes the DAG with InputNode + worker.execute_model_ray.bind(...); PP stages pass tensors via with_tensor_transport(transport), and finally experimental_compile(...).
• check_health:625-628 — in this version it just returns; a TODO is left for a real Ray actor health check later.
• RayWorkerWrapper:56-91 — inherits WorkerWrapperBase, adds adjust_rank, get_node_ip, get_node_and_physical_gpu_ids, execute_method(wraps exceptions with a "possible deadlock" log before re-raising).

Data flow

At startup, _init_executor goes through _init_workers_ray, the longest section: select bundles, create RayWorkerWrapper actors per placement group, then ray.get each actor's node IP for sorting and reranking:

# vllm/v1/executor/ray_executor.py L191-L213
for rank, bundle_id in enumerate(bundle_indices):
    scheduling_strategy = PlacementGroupSchedulingStrategy(
        placement_group=placement_group,
        placement_group_capture_child_tasks=True,
        placement_group_bundle_index=bundle_id,
    )

    if current_platform.ray_device_key == "GPU":
        worker = ray.remote(
            num_cpus=0,
            num_gpus=num_gpus,
            scheduling_strategy=scheduling_strategy,
            **ray_remote_kwargs,
        )(RayWorkerWrapper).remote(rpc_rank=rank)
    else:
        worker = ray.remote(
            num_cpus=0,
            num_gpus=0,
            resources={current_platform.ray_device_key: num_gpus},
            scheduling_strategy=scheduling_strategy,
            **ray_remote_kwargs,
        )(RayWorkerWrapper).remote(rpc_rank=rank)

    worker_metadata.append(RayWorkerMetaData(worker=worker, created_rank=rank))

At runtime, when EngineCore.step reaches execute_model, if uses_sampler=True and the batch has tokens, the executor only stashes scheduler_output and returns(ray_executor.py:401-407); sample_tokens is the one that actually runs _execute_dag, building the compiled DAG on first call:

# vllm/v1/executor/ray_executor.py L434-L452
def _execute_dag(self, scheduler_output, grammar_output, non_block=False):
    if self.forward_dag is None:
        self.forward_dag = self._compiled_ray_dag(enable_asyncio=False)

    refs = self.forward_dag.execute((scheduler_output, grammar_output))

    if not self.has_connector:
        if not non_block:
            output = refs[0].get()
            detach_zero_copy_from_model_runner_output(output)
            return output
        return FutureWrapper(refs[0])
    ...

Control-plane RPCs (initialize_from_config, compile_or_warm_up_model, determine_available_memory, etc.) all go through collective_rpc, i.e. per-worker execute_method.remote(...)(ray_executor.py:484-489), then ray.get to gather results. This is a separate channel from the data plane (forward via the compiled DAG).

Boundaries and failure modes

• Node IPs must be unique: _init_workers_ray explicitly checks n_nodes == n_ips, otherwise raises RuntimeError and points you to VLLM_HOST_IP(ray_executor.py:295-303).
• Loopback address on a single node: when multiple actors co-locate on the same node, driver_ip = "127.0.0.1" avoids get_ip() picking the wrong NIC and breaking worker-to-worker communication(ray_executor.py:329-341).
• State machine out of order raises: execute_model raises "sample_tokens() must be called after execute_model() returns None" if it sees self.scheduler_output is not None(ray_executor.py:395-399).
• Ray Compiled Graph version and dependencies: _check_ray_cgraph_installation requires ray>=2.43.0, ray[cgraph], optional cupy(ray_executor.py:497-525), otherwise startup fails fast.
• TPU/XPU uses shared memory: when _init_executor detects TPU/XPU it forces VLLM_USE_RAY_COMPILED_DAG_CHANNEL_TYPE=shm(ray_executor.py:73-75) to bypass NVIDIA NCCL.
• check_health is currently empty: RayDistributedExecutor.check_health just returns(ray_executor.py:625-628); a dead actor is only surfaced via an exception thrown by ray.get — there is no proactive heartbeat yet.
• shutdown force-kills: forward_dag.teardown() followed by ray.kill(worker)(ray_executor.py:111-112); any un-flushed state inside the worker is lost, and the docs explicitly note that the SIGTERM log is expected.

Summary

RayDistributedExecutor is v1's executor implementation for the multi-node multi-GPU case. It wraps each worker as a Ray actor, pins the topology with a placement group, and uses a Ray Compiled Graph to build a forward DAG that strings TP/PP together; control-plane RPCs still go through the actor's execute_method. It exposes the exact same interface to EngineCore as UniProcExecutor, only swapping the underlying RPC mechanism. For how the executor is abstracted, see /executor/executor; for how the worker runs the forward internally, see /worker/worker and /worker/gpu-model-runner.

See official docs: vLLM docs · README