RayDistributedExecutor:Ray で worker を多ノード多カードに展開する

源码版本v0.25.1

役割

単 GPU シナリオでは UniProcExecutor が 1 プロセスで完結しますが、TP>1 またはノードをまたぐ場合にはプロセス間通信が必須です。RayDistributedExecutor(ray_executor.py:64-68)は Ray バックエンドの実装です。各 worker を 1 つの Ray actor に包み、placement group が指定する GPU に分布させ、Ray Compiled Graph でパイプラインを構築して forward を直列に繋ぎます。uses_ray = True、supports_pp = True(ray_executor.py:67-68)で、Ray でオーケストレーションしつつパイプライン並列 (pipeline parallelism) をもサポートすることを示します。

UniProcExecutor と同様に Executor を継承し、基底クラスの execute_model/sample_tokens/profile/sleep などのシグネチャはそのまま変更せず、_init_executor、collective_rpc、execute_model、check_health、shutdown(ray_executor.py:70-97)だけをオーバーライドします。違いは RPC の経路だけです:同プロセスでは直接メソッドを呼び、Ray では worker.execute_method.remote(...)(ray_executor.py:484-489)します。

設計動機

• Ray actor と worker が 1 対 1 対応:各 GPU に 1 つの RayWorkerWrapper actor(ray_executor.py:200-213)。Ray がスケジューリング、フォールトトレランス、リソンス会計を担当し、executor 自身はプロセスを spawn しません。
• placement group がトポロジーを固定:actor は PlacementGroupSchedulingStrategy で bundle に紐付け(ray_executor.py:192-196)、GPU 親和性は起動前に決まり、実行期には移動しません。
• 同一ノード優先でソート:actor の作成順序は Ray が決める(乱れ得る)ため、すべての actor を作り終えた後に IP で再ソートし、driver と同じノードのものを前に置き(ray_executor.py:234-258)、その後 collective_rpc("adjust_rank", args=(rerank_mapping,)) で rank を修正します。
• Ray Compiled Graph で forward を実行:毎 step RPC で execute_model を呼ぶのではなく、forward_dag(ray_executor.py:527-620)を構築します。PP=2/TP=4 のとき 4 つの worker を InputNode -> TP group -> MultiOutputNode に直列にし、中間テンソルは with_tensor_transport(transport="nccl"|"shm") で NCCL または共有メモリで転送(ray_executor.py:582-591)します。
• 環境変数を自動複製:driver プロセス内の vLLM 関連の環境変数は update_environment_variables で各 worker に一括 push(ray_executor.py:305-327)され、WORKER_SPECIFIC_ENV_VARS を除くことで worker 同士の汚染を回避します。
• execute_model を 2 段に分割:sampler を有効にした場合、execute_model はすぐには走らず self.scheduler_output を保存するだけにし、sample_tokens が来てから _execute_dag(ray_executor.py:390-432)を走らせ、スケジューラと sampler の grammar bitmask 計算を forward と重ねられるようにします。

主要ファイル

• RayDistributedExecutor._init_executor:64-97 — initialize_ray_cluster、placement group 取得、_init_workers_ray 呼び出し、has_connector、uses_sampler、scheduler_output のプレースホルダ初期化。
• shutdown:99-113 — forward_dag.teardown() 後に ray.kill(worker) を順次実行、ログは SIGTERM が想定通りの挙動であることを示す。
• _init_workers_ray:143-213 — bundle_indices ごとに ray.remote(...)(RayWorkerWrapper).remote(rpc_rank=rank)、GPU は num_gpus=num_gpus、他プラットフォームは resources={ray_device_key: num_gpus}。
• rerank:217-258 — ray.get で各 actor の node IP を取得し、「driver 同一ノード優先 → ノードの actor 数が少ない優先 → IP が小さい優先」でソート、その後 collective_rpc("adjust_rank", ...) で rpc_rank を変更。
• init_worker + load_model:343-368 — all_kwargs を組み立てた後、一度に collective_rpc("init_worker") + collective_rpc("init_device") + collective_rpc("load_model")。
• pp_tp_workers:370-378 — PP×TP の 2 次元で self.pp_tp_workers を埋め、後の compiled DAG 構築に使用。
• execute_model + sample_tokens:390-432 — 2 段階:execute_model は scheduler_output を保存するだけで、実際の実行は sample_tokens 内の _execute_dag で行う。
• _execute_dag:434-468 — 初回呼び出し時に self.forward_dag = self._compiled_ray_dag(...)、forward_dag.execute((scheduler_output, grammar_output)) が refs を返し、PP 無効なら refs[0].get() でブロックして結果を取得、非ブロックなら FutureWrapper を返す。
• collective_rpc:470-495 — method をシリアライズ(cloudpickle.dumps)し、worker ごとに execute_method.remote(...)、その後 ray.get(ray_worker_outputs, timeout=timeout)。non_block=True は FutureWrapper を返す。
• _compiled_ray_dag:527-620 — InputNode + worker.execute_model_ray.bind(...) で DAG を組み、PP 間は with_tensor_transport(transport) でテンソルを渡し、最後に experimental_compile(...)。
• check_health:625-628 — 現バージョンは直接 return、将来の本当の Ray actor ヘルスチェック用の TODO を残す。
• RayWorkerWrapper:56-91 — WorkerWrapperBase を継承し、adjust_rank、get_node_ip、get_node_and_physical_gpu_ids、execute_method(例外を「デッドロックの可能性あり」ログで包んでから再送出)を追加。

データフロー

起動期の _init_executor は _init_workers_ray を通ります。この区間が最も長いです:bundle 選択、placement group で RayWorkerWrapper actor を作成、その後 ray.get で各 actor の node IP を取得してソート、rerank します:

# 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))

実行期に EngineCore.step が execute_model に到達したとき、uses_sampler=True かつこのバッチに token があれば、executor は scheduler_output を一時保存するだけで復帰(ray_executor.py:401-407)します。sample_tokens が呼ばれたときに初めて _execute_dag に進み、初回は compiled DAG を構築します:

# 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])
    ...

制御プレーンの RPC(initialize_from_config、compile_or_warm_up_model、determine_available_memory など)はすべて collective_rpc で、worker ごとに execute_method.remote(...)(ray_executor.py:484-489)してから ray.get で回収します。これはデータプレーン(forward が compiled DAG を走る)とは独立した 2 つの経路です。

境界と失敗

• ノード IP は一意でなければならない:_init_workers_ray は n_nodes == n_ips を明示的にチェックし、満たさなければ RuntimeError を投げ、VLLM_HOST_IP を調べるよう促します(ray_executor.py:295-303)。
• 単一ノードはループバックを使用:同一ノードで複数 actor のとき driver_ip = "127.0.0.1" とし、get_ip() が誤った NIC を取って worker 間通信が失敗するのを防ぎ(ray_executor.py:329-341)します。
• 状態機械の順序が違えばエラー:execute_model は self.scheduler_output is not None を見て直ちに「sample_tokens() must be called after execute_model() returns None」を投げ(ray_executor.py:395-399)ます。
• Ray Compiled Graph のバージョンと依存:_check_ray_cgraph_installation は ray>=2.43.0、ray[cgraph]、オプションの cupy を要求(ray_executor.py:497-525)し、満たさなければ起動期に fail-fast します。
• TPU/XPU は共有メモリを使用:_init_executor が TPU/XPU を検出したとき VLLM_USE_RAY_COMPILED_DAG_CHANNEL_TYPE=shm を強制(ray_executor.py:73-75)し、NVIDIA NCCL を回避します。
• check_health は現在空:RayDistributedExecutor.check_health は直接 return(ray_executor.py:625-628)し、actor が本当に死んだときは ray.get の例外送出に頼るしかなく、現在能動的なハートビートはありません。
• shutdown は強制 kill:forward_dag.teardown() の後に ray.kill(worker)(ray_executor.py:111-112)し、worker 内の未フラッシュ状態は失われます。ドキュメントでは SIGTERM のログが想定通りであることを明記しています。

まとめ

RayDistributedExecutor は v1 が多ノード多カードシナリオで提供する実行器実装です。各 worker を Ray actor に包み、placement group でトポロジーを固定し、Ray Compiled Graph で forward DAG を構築して TP/PP を直列に繋ぎます。制御プレーン RPC は引き続き actor の execute_method 経由です。UniProcExecutor が EngineCore に晒すインターフェースは完全に一致し、下層の RPC 機構が変わっただけです。実行器をどう抽象化しているかは /executor/executor、worker 内部でどう forward を走らせるかは /worker/worker と /worker/gpu-model-runner を参照してください。

公式資料:vLLM 文档 · README