| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | |
| 4 | import os |
| 5 | import socket |
| 6 | from collections.abc import Callable |
| 7 | from datetime import timedelta |
| 8 | from typing import TYPE_CHECKING, Any, Literal, overload |
| 9 | |
| 10 | import regex as re |
| 11 | import torch |
| 12 | from pydantic import Field, field_validator, model_validator |
| 13 | from torch.distributed import ProcessGroup, ReduceOp, Store |
| 14 | from typing_extensions import Self |
| 15 | |
| 16 | import vllm.envs as envs |
| 17 | from vllm.config.fault_tolerance import FaultToleranceConfig |
| 18 | from vllm.config.utils import config |
| 19 | from vllm.logger import init_logger |
| 20 | from vllm.platforms import current_platform |
| 21 | from vllm.utils.network_utils import get_open_ports_list |
| 22 | |
| 23 | if TYPE_CHECKING: |
| 24 | from ray.runtime_env import RuntimeEnv |
| 25 | from ray.util.placement_group import PlacementGroup |
| 26 | |
| 27 | from vllm.config.fault_tolerance import FaultToleranceConfig |
| 28 | from vllm.v1.executor import Executor |
| 29 | else: |
| 30 | RuntimeEnv = Any |
| 31 | PlacementGroup = Any |
| 32 | Executor = Any |
| 33 | |
| 34 | logger = init_logger(__name__) |
| 35 | _NUMACTL_CPUSET_PATTERN = re.compile(r"^\d+(?:-\d+)?(?:,\d+(?:-\d+)?)*$") |
| 36 | |
| 37 | ExpertPlacementStrategy = Literal["linear", "round_robin"] |
| 38 | DistributedExecutorBackend = Literal["ray", "mp", "uni", "external_launcher"] |
| 39 | DataParallelBackend = Literal["ray", "mp"] |
| 40 | EPLBPolicyOption = Literal["default"] |
| 41 | DCPCommBackend = Literal["ag_rs", "a2a"] |
| 42 | EPLBCommunicatorBackend = Literal[ |
| 43 | "torch_nccl", "torch_gloo", "torch_xccl", "nixl", "pynccl" |
| 44 | ] |
| 45 | All2AllBackend = Literal[ |
| 46 | "naive", |
| 47 | "pplx", |
| 48 | "deepep_high_throughput", |
| 49 | "deepep_low_latency", |
| 50 | "deepep_v2", |
| 51 | "mori_high_throughput", |
| 52 | "mori_low_latency", |
| 53 | "moonep", |
| 54 | "nixl_ep", |
| 55 | "allgather_reducescatter", |
| 56 | "flashinfer_all2allv", # temporary alias for flashinfer_nvlink_two_sided |
| 57 | "flashinfer_nvlink_two_sided", |
| 58 | "flashinfer_nvlink_one_sided", |
| 59 | ] |
| 60 | |
| 61 | |
| 62 | @config |
| 63 | class EPLBConfig: |
| 64 | """Configuration for Expert Parallel Load Balancing (EP).""" |
| 65 | |
| 66 | window_size: int = Field(default=1000, gt=0) |
| 67 | """Window size for expert load recording.""" |
| 68 | step_interval: int = Field(default=3000, gt=0) |
| 69 | """ |
| 70 | Interval for rearranging experts in expert parallelism. |
| 71 | |
| 72 | Note that if this is greater than the EPLB window size, only the metrics |
| 73 | of the last `lb_window_size` steps will be used for rearranging experts. |
| 74 | """ |
| 75 | |
| 76 | num_redundant_experts: int = Field(default=0, ge=0) |
| 77 | """Number of redundant experts to use for expert parallelism.""" |
| 78 | |
| 79 | log_balancedness: bool = False |
| 80 | """ |
| 81 | Log the balancedness each step of expert parallelism. |
| 82 | This is turned off by default since it will cause communication overhead. |
| 83 | """ |
| 84 | log_balancedness_interval: int = Field(default=1, gt=0) |
| 85 | """ |
| 86 | Interval for logging the balancedness. |
| 87 | """ |
| 88 | use_async: bool = True |
| 89 | """ |
| 90 | Whether to use non-blocking EPLB. |
| 91 | """ |
| 92 | |
| 93 | policy: EPLBPolicyOption = "default" |
| 94 | """The policy type for expert parallel load balancing (EPLB).""" |
| 95 | |
| 96 | communicator: EPLBCommunicatorBackend | None = None |
| 97 | """ |
| 98 | Backend for EPLB expert weight communication: |
| 99 | - "torch_nccl": Use torch.distributed on the device process group |
| 100 | - "torch_gloo": Use torch.distributed gloo with CPU staging |
| 101 | - "torch_xccl": Use torch.distributed XCCL device P2P on XPU |
| 102 | - "nixl": Use NIXL with staged send/recv buffers |
| 103 | - "pynccl": Use PyNccl send/recv |
| 104 | - None: Auto-select backend ("torch_xccl" on XPU, prefers "nixl" |
| 105 | on CUDA, falls back to "torch_gloo") |
| 106 | """ |
| 107 | |
| 108 | @model_validator(mode="after") |
| 109 | def _validate_eplb_config(self) -> Self: |
| 110 | if self.use_async and self.policy != "default": |
| 111 | raise ValueError("Async EPLB is only supported with the default policy.") |
| 112 | if self.use_async and self.communicator in ("torch_nccl", "pynccl"): |
| 113 | raise ValueError( |
| 114 | f"{self.communicator} communicator is incompatible with " |
| 115 | "async EPLB due to NCCL multi-stream conflicts. Use " |
| 116 | "'torch_gloo' or 'nixl' instead, or leave communicator " |
| 117 | "unset for automatic selection." |
| 118 | ) |
| 119 | if self.log_balancedness and self.log_balancedness_interval <= 0: |
| 120 | raise ValueError("log_balancedness_interval must be greater than 0.") |
| 121 | return self |
| 122 | |
| 123 | |
| 124 | @config |
| 125 | class ParallelConfig: |
| 126 | """Configuration for the distributed execution.""" |
| 127 | |
| 128 | pipeline_parallel_size: int = Field(default=1, ge=1) |
| 129 | """Number of pipeline parallel groups.""" |
| 130 | tensor_parallel_size: int = Field(default=1, ge=1) |
| 131 | """Number of tensor parallel groups.""" |
| 132 | prefill_context_parallel_size: int = Field(default=1, ge=1) |
| 133 | """Number of ranks that split prefill sequence computation. PCP expands |
| 134 | the process world size but does not increase the KV-cache shard count.""" |
| 135 | data_parallel_size: int = Field(default=1, ge=1) |
| 136 | """Number of data parallel groups. MoE layers will be sharded according to |
| 137 | the product of the tensor, prefill-context, and data parallel sizes.""" |
| 138 | data_parallel_size_local: int = Field(default=1, ge=0) |
| 139 | """Number of local data parallel groups. A value of 0 is a sentinel used by |
| 140 | the engine-args layer to signal that data parallelism was specified |
| 141 | externally (see `ParallelConfig.__post_init__`).""" |
| 142 | data_parallel_rank: int = Field(default=0, ge=0) |
| 143 | """Rank of the data parallel group. The runtime check at |
| 144 | ``__post_init__`` further bounds this by ``data_parallel_size``.""" |
| 145 | data_parallel_rank_local: int | None = None |
| 146 | """Local rank of the data parallel group, set only in SPMD mode.""" |
| 147 | data_parallel_master_ip: str = "127.0.0.1" |
| 148 | """IP of the data parallel master.""" |
| 149 | data_parallel_rpc_port: int = Field(default=29550, ge=1, le=65535) |
| 150 | """Fixed port for data parallel messaging, shared by all nodes.""" |
| 151 | dp_sync_interval: int = Field(default=16, ge=1) |
| 152 | """Steps between DP finish-sync all-reduces; must match across DP ranks.""" |
| 153 | data_parallel_master_port: int = 29500 |
| 154 | """Port of the data parallel master.""" |
| 155 | data_parallel_backend: DataParallelBackend = "mp" |
| 156 | """Backend to use for data parallel, either "mp" or "ray".""" |
| 157 | data_parallel_external_lb: bool = False |
| 158 | """Whether to use "external" DP LB mode. Applies only to online serving |
| 159 | and when data_parallel_size > 0. This is useful for a "one-pod-per-rank" |
| 160 | wide-EP setup in Kubernetes. Supported only for MoE deployments; non-MoE |
| 161 | models should use independent vLLM instances without --data-parallel-* |
| 162 | arguments. Set implicitly when --data-parallel-rank is provided explicitly |
| 163 | to vllm serve.""" |
| 164 | data_parallel_hybrid_lb: bool = False |
| 165 | """Whether to use "hybrid" DP LB mode. Applies only to online serving |
| 166 | and when data_parallel_size > 0. Enables running an AsyncLLM |
| 167 | and API server on a "per-node" basis where vLLM load balances |
| 168 | between local data parallel ranks, but an external LB balances |
| 169 | between vLLM nodes/replicas. Set explicitly in conjunction with |
| 170 | --data-parallel-start-rank.""" |
| 171 | is_moe_model: bool | None = None |
| 172 | """Whether the deployed model is MoE (if known).""" |
| 173 | enable_expert_parallel: bool = False |
| 174 | """Use expert parallelism instead of tensor parallelism for MoE layers.""" |
| 175 | enable_batch_sharded_sampling: bool | None = None |
| 176 | """Use sharded sampling across tensor parallel ranks. Each rank samples |
| 177 | a slice of the batch instead of every rank sampling all of it. Currently |
| 178 | defaults to False if not set. Enabling it explicitly raises when the config |
| 179 | cannot support it (`tensor_parallel_size` must be > 1, `max_num_seqs` at |
| 180 | least `tensor_parallel_size`, and `max_logprobs` non-negative). Models opt in |
| 181 | by implementing `compute_logits_local`.""" |
| 182 | enable_ep_weight_filter: bool = False |
| 183 | """Skip non-local expert weights during model loading when expert |
| 184 | parallelism is active. Each rank only reads its own expert shard from |
| 185 | disk, which can drastically reduce storage I/O for MoE models with |
| 186 | per-expert weight tensors (e.g. DeepSeek, Mixtral, Kimi-K2.5). Has no |
| 187 | effect on 3D fused-expert checkpoints (e.g. GPT-OSS) or non-MoE |
| 188 | models.""" |
| 189 | enable_eplb: bool = False |
| 190 | """Enable expert parallelism load balancing for MoE layers.""" |
| 191 | eplb_config: EPLBConfig = Field(default_factory=EPLBConfig) |
| 192 | """Expert parallelism configuration.""" |
| 193 | expert_placement_strategy: ExpertPlacementStrategy = "linear" |
| 194 | """The expert placement strategy for MoE layers: |
| 195 | |
| 196 | - "linear": Experts are placed in a contiguous manner. For example, with 4 |
| 197 | experts and 2 ranks, rank 0 will have experts [0, 1] and rank 1 will have |
| 198 | experts [2, 3]. |
| 199 | - "round_robin": Experts are placed in a round-robin manner. For example, |
| 200 | with 4 experts and 2 ranks, rank 0 will have experts [0, 2] and rank 1 |
| 201 | will have experts [1, 3]. This strategy can help improve load balancing |
| 202 | for grouped expert models with no redundant experts.""" |
| 203 | all2all_backend: All2AllBackend = "allgather_reducescatter" |
| 204 | """All2All backend for MoE expert parallel communication. Available options: |
| 205 | |
| 206 | - "allgather_reducescatter": All2all based on allgather and reducescatter |
| 207 | - "deepep_high_throughput": Use deepep high-throughput kernels |
| 208 | - "deepep_low_latency": Use deepep low-latency kernels |
| 209 | - "mori_high_throughput": MoRI EP with InterNodeV1 for multi-node |
| 210 | - "mori_low_latency": MoRI EP with InterNodeV1LL for multi-node |
| 211 | - "moonep": MoonEP balanced EP with dynamic redundant experts (NVLink) |
| 212 | - "nixl_ep": Use nixl-ep kernels |
| 213 | - "flashinfer_nvlink_one_sided": Use flashinfer high-throughput a2a kernels |
| 214 | - "flashinfer_nvlink_two_sided": Use flashinfer two-sided kernels for mnnvl""" |
| 215 | |
| 216 | max_parallel_loading_workers: int | None = Field(default=None, ge=1) |
| 217 | """Maximum number of parallel loading workers when loading model |
| 218 | sequentially in multiple batches. To avoid RAM OOM when using tensor |
| 219 | parallel and large models.""" |
| 220 | |
| 221 | disable_custom_all_reduce: bool = False |
| 222 | """Disable the custom all-reduce kernel and fall back to NCCL.""" |
| 223 | |
| 224 | enable_elastic_ep: bool = False |
| 225 | """Enable elastic expert parallelism with stateless NCCL groups for DP/EP.""" |
| 226 | elastic_ep_max_dp_size: int = Field(default=None, ge=1) # type: ignore[assignment] |
| 227 | """Maximum data parallel size supported by elastic expert parallelism.""" |
| 228 | |
| 229 | enable_dbo: bool = False |
| 230 | """Enable dual batch overlap for the model executor.""" |
| 231 | ubatch_size: int = Field(default=0, ge=0) |
| 232 | """Number of ubatch size.""" |
| 233 | |
| 234 | dbo_decode_token_threshold: int = Field(default=32, ge=0) |
| 235 | """The threshold for dual batch overlap for batches only containing decodes. |
| 236 | If the number of tokens in the request is greater than this threshold, |
| 237 | microbatching will be used. Otherwise, the request will be processed in a |
| 238 | single batch.""" |
| 239 | dbo_prefill_token_threshold: int = Field(default=512, ge=0) # TODO(lucas): tune |
| 240 | """The threshold for dual batch overlap for batches that contain one or more |
| 241 | prefills. If the number of tokens in the request is greater than this |
| 242 | threshold, microbatching will be used. Otherwise, the request will be |
| 243 | processed in a single batch.""" |
| 244 | |
| 245 | disable_nccl_for_dp_synchronization: bool | None = None |
| 246 | """Forces the dp synchronization logic in vllm/v1/worker/dp_utils.py |
| 247 | to use Gloo instead of NCCL for its all reduce. |
| 248 | |
| 249 | Defaults to True when async scheduling is enabled, False otherwise. |
| 250 | """ |
| 251 | |
| 252 | ray_workers_use_nsight: bool = False |
| 253 | """Whether to profile Ray workers with nsight, see https://docs.ray.io/en/latest/ray-observability/user-guides/profiling.html#profiling-nsight-profiler.""" |
| 254 | |
| 255 | ray_runtime_env: RuntimeEnv | None = None |
| 256 | """Ray runtime environment to pass to distributed workers.""" |
| 257 | |
| 258 | placement_group: PlacementGroup | None = None |
| 259 | """ray distributed model workers placement group.""" |
| 260 | |
| 261 | distributed_executor_backend: ( |
| 262 | str | DistributedExecutorBackend | type[Executor] | None |
| 263 | ) = None |
| 264 | """ |
| 265 | Backend to use for distributed model workers, either "ray" or "mp" |
| 266 | (multiprocessing). If the product of pipeline_parallel_size and tensor_parallel_size |
| 267 | is less than or equal to the number of GPUs available, "mp" will be used to |
| 268 | keep processing on a single host. Otherwise, an error will be raised. To use "mp" |
| 269 | you must also set nnodes, and to use "ray" you must manually set |
| 270 | distributed_executor_backend to "ray". |
| 271 | |
| 272 | Note: |
| 273 | [TPU](https://docs.vllm.ai/projects/tpu/en/latest/) platform only supports Ray |
| 274 | for distributed inference. |
| 275 | """ |
| 276 | |
| 277 | worker_cls: str = "auto" |
| 278 | """The full name of the worker class to use. If "auto", the worker class |
| 279 | will be determined based on the platform.""" |
| 280 | sd_worker_cls: str = "auto" |
| 281 | """The full name of the worker class to use for speculative decoding. |
| 282 | If "auto", the worker class will be determined based on the platform.""" |
| 283 | worker_extension_cls: str = "" |
| 284 | """The full name of the worker extension class to use. The worker extension |
| 285 | class is dynamically inherited by the worker class. This is used to inject |
| 286 | new attributes and methods to the worker class for use in collective_rpc |
| 287 | calls.""" |
| 288 | master_addr: str = "127.0.0.1" |
| 289 | """distributed master address for multi-node distributed |
| 290 | inference when distributed_executor_backend is mp.""" |
| 291 | master_port: int = 29501 |
| 292 | """distributed master port for multi-node distributed |
| 293 | inference when distributed_executor_backend is mp.""" |
| 294 | node_rank: int = Field(default=0, ge=0) |
| 295 | """distributed node rank for multi-node distributed |
| 296 | inference when distributed_executor_backend is mp.""" |
| 297 | nnodes: int = Field(default=1, ge=1) |
| 298 | """num of nodes for multi-node distributed |
| 299 | inference when distributed_executor_backend is mp.""" |
| 300 | numa_bind: bool = False |
| 301 | """Enable NUMA binding for GPU worker subprocesses. |
| 302 | |
| 303 | By default, workers are pinned to their GPU's NUMA-local CPUs and |
| 304 | memory; on PCT-capable Xeons they also auto-bind to the SKU's |
| 305 | PCT priority cores. |
| 306 | """ |
| 307 | numa_bind_nodes: list[int] | None = None |
| 308 | """NUMA node to bind each GPU worker to. |
| 309 | |
| 310 | Specify one NUMA node per visible GPU, for example `[0, 0, 1, 1]` |
| 311 | for a 4-GPU system with GPUs 0-1 on NUMA node 0 and GPUs 2-3 on |
| 312 | NUMA node 1. If unset and `numa_bind=True`, vLLM auto-detects the |
| 313 | GPU-to-NUMA topology. The values are passed to `numactl --membind` |
| 314 | and `--cpunodebind`, so they must be valid `numactl` NUMA node indices. |
| 315 | """ |
| 316 | numa_bind_cpus: list[str] | None = None |
| 317 | """Optional CPU lists to bind each GPU worker to. |
| 318 | |
| 319 | Specify one CPU list per visible GPU, for example |
| 320 | `["0-3", "4-7", "8-11", "12-15"]`. When set, vLLM uses |
| 321 | `numactl --physcpubind` instead of `--cpunodebind`. This is useful |
| 322 | for custom policies such as binding to PCT or other high-frequency cores. |
| 323 | Each entry must use `numactl --physcpubind` CPU-list syntax, for example |
| 324 | `"0-3"` or `"0,2,4-7"`. |
| 325 | """ |
| 326 | assigned_physical_gpu_ids: list[int] | None = None |
| 327 | """Mapping from vLLM-local logical GPU IDs to physical GPU IDs. |
| 328 | |
| 329 | For example, ``[2, 3]`` means logical GPU 0 maps to physical GPU 2, |
| 330 | and logical GPU 1 maps to physical GPU 3. Physical IDs are used only |
| 331 | at platform/topology boundaries such as NVML, NIC affinity, P2P |
| 332 | checks, and final CUDA device selection when needed. When None, |
| 333 | logical IDs map to visible device IDs in order.""" |
| 334 | |
| 335 | distributed_timeout_seconds: int | None = None |
| 336 | """Timeout in seconds for distributed operations (e.g., init_process_group). |
| 337 | If set, this value is passed to torch.distributed.init_process_group as the |
| 338 | timeout parameter. If None, PyTorch's default timeout is used (600s for NCCL). |
| 339 | Increase this for multi-node setups where model downloads may be slow.""" |
| 340 | |
| 341 | cpu_distributed_timeout_seconds: int | None = None |
| 342 | """Timeout (in seconds) for cpu communication groups. If None, PyTorch's |
| 343 | default timeout is used (1800s for gloo).""" |
| 344 | |
| 345 | world_size: int = Field(init=False) |
| 346 | """world_size is TPxPP, it affects the number of workers we create.""" |
| 347 | |
| 348 | rank: int = 0 |
| 349 | """Global rank in distributed setup.""" |
| 350 | |
| 351 | _data_parallel_master_port_list: list[int] = Field(default_factory=list) |
| 352 | """List of open port auto-queried for data parallel messaging. |
| 353 | Set to be private as it's not intended to be configured by users. |
| 354 | """ |
| 355 | |
| 356 | _coord_store_port: int = 0 |
| 357 | """Port of the coordination TCPStore. Can be set by the API server; workers |
| 358 | connect as clients to exchange self-picked group ports at runtime.""" |
| 359 | |
| 360 | decode_context_parallel_size: int = Field(default=1, ge=1) |
| 361 | """Number of ranks that shard the decode KV cache. DCP does not expand |
| 362 | the process world size. Without PCP, DCP reuses TP ranks. With PCP, DCP |
| 363 | either spans the PCP axis or the full TP x PCP block.""" |
| 364 | |
| 365 | dcp_kv_cache_interleave_size: int = 1 |
| 366 | """ |
| 367 | Interleave size of kv_cache storage while using DCP. |
| 368 | dcp_kv_cache_interleave_size has been replaced by cp_kv_cache_interleave_size, |
| 369 | and will be deprecated when PCP is fully supported. |
| 370 | |
| 371 | """ |
| 372 | dcp_comm_backend: DCPCommBackend | None = None |
| 373 | """Communication backend for Decode Context Parallel (DCP). |
| 374 | - "ag_rs": AllGather + ReduceScatter (existing behavior) |
| 375 | - "a2a": All-to-All exchange of partial outputs + LSE, then |
| 376 | combine with Triton kernel. Reduces NCCL calls from 3 to 2 |
| 377 | per layer for MLA models. |
| 378 | |
| 379 | `None` selects the model default, which is "ag_rs" unless the model |
| 380 | overrides it via [`set_dcp_defaults`][vllm.config.ParallelConfig.set_dcp_defaults]. |
| 381 | """ |
| 382 | |
| 383 | dcp_q_replicate: bool | None = None |
| 384 | """Replicate the MLA query projection within each DCP group so decode can skip the |
| 385 | query all-gather. |
| 386 | |
| 387 | With DCP the KV cache is sharded across the group, so the standard MLA decode path |
| 388 | all-gathers the query every step. Replicating the (small) query projection at load |
| 389 | time lets each rank materialize the full group-local head set and skip that |
| 390 | collective, at the cost of computing the projection redundantly on every rank |
| 391 | in the group. |
| 392 | """ |
| 393 | |
| 394 | cp_kv_cache_interleave_size: int = 1 |
| 395 | """Interleave size of kv_cache storage while using DCP. |
| 396 | Store interleave_size tokens on dcp_rank i, then store next |
| 397 | interleave_size tokens on dcp_rank i+1. |
| 398 | Interleave_size=1: token-level alignment, where token `i` is stored on |
| 399 | dcp_rank `i % dcp_world_size`. |
| 400 | Interleave_size=block_size: block-level alignment, where tokens are |
| 401 | first populated to the preceding ranks. Tokens are then stored |
| 402 | in (rank i+1, block j) only after (rank i, block j) is fully occupied. |
| 403 | Block_size should be greater than or equal to cp_kv_cache_interleave_size. |
| 404 | Block_size should be divisible by cp_kv_cache_interleave_size. |
| 405 | |
| 406 | When --cp-kv-cache-interleave-size is omitted (None), the interleave size |
| 407 | is resolved automatically based on NIXL transfer requirements. |
| 408 | Explicit settings take priority. |
| 409 | """ |
| 410 | |
| 411 | _allow_auto_resolve_cp_interleave_size: bool = True |
| 412 | """Whether NIXL may select the interleave size automatically.""" |
| 413 | |
| 414 | data_parallel_index: int = Field(init=False) |
| 415 | """Equal to the data parallel rank but not used for torch process groups |
| 416 | and not overridden for dense models.""" |
| 417 | |
| 418 | _api_process_count: int = Field(default=1, gt=0) |
| 419 | """ |
| 420 | The number of API processes initialized. |
| 421 | |
| 422 | Note: |
| 423 | This is an internal config that is only valid for and |
| 424 | should only be set by API server scale-out. |
| 425 | """ |
| 426 | |
| 427 | _api_process_rank: int = Field(default=0, ge=-1) |
| 428 | """ |
| 429 | The rank of this API process, or `-1` for engine core processes |
| 430 | under API server scale-out. |
| 431 | |
| 432 | Note: |
| 433 | This is an internal config that is only valid for and |
| 434 | should only be set by API server scale-out. |
| 435 | """ |
| 436 | |
| 437 | enable_fault_tolerance: bool = False |
| 438 | """Enable fault tolerance for detailed error recovery, |
| 439 | such as scaling down fault DPEngineCore. |
| 440 | """ |
| 441 | |
| 442 | fault_tolerance_config: FaultToleranceConfig = Field( |
| 443 | default_factory=FaultToleranceConfig |
| 444 | ) |
| 445 | """The configurations for fault tolerance.""" |
| 446 | |
| 447 | @field_validator( |
| 448 | "disable_nccl_for_dp_synchronization", "elastic_ep_max_dp_size", mode="wrap" |
| 449 | ) |
| 450 | @classmethod |
| 451 | def _skip_none_validation(cls, value: Any, handler: Callable) -> Any: |
| 452 | """Skip validation if the value is `None` when initialisation is delayed.""" |
| 453 | return None if value is None else handler(value) |
| 454 | |
| 455 | @field_validator("numa_bind_nodes") |
| 456 | @classmethod |
| 457 | def _validate_numa_bind_nodes(cls, value: list[int] | None) -> list[int] | None: |
| 458 | if value is None: |
| 459 | return None |
| 460 | if not value: |
| 461 | raise ValueError("numa_bind_nodes must not be empty.") |
| 462 | if any(node < 0 for node in value): |
| 463 | raise ValueError("numa_bind_nodes must contain non-negative integers.") |
| 464 | return value |
| 465 | |
| 466 | @field_validator("numa_bind_cpus") |
| 467 | @classmethod |
| 468 | def _validate_numa_bind_cpus(cls, value: list[str] | None) -> list[str] | None: |
| 469 | if value is None: |
| 470 | return None |
| 471 | if not value: |
| 472 | raise ValueError("numa_bind_cpus must not be empty.") |
| 473 | |
| 474 | for cpuset in value: |
| 475 | if not cpuset: |
| 476 | raise ValueError("numa_bind_cpus entries must not be empty.") |
| 477 | if not _NUMACTL_CPUSET_PATTERN.fullmatch(cpuset): |
| 478 | raise ValueError( |
| 479 | "numa_bind_cpus entries must use numactl CPU list syntax, " |
| 480 | "for example '0-3' or '0,2,4-7'." |
| 481 | ) |
| 482 | for part in cpuset.split(","): |
| 483 | if "-" not in part: |
| 484 | continue |
| 485 | start_str, end_str = part.split("-", 1) |
| 486 | if int(start_str) > int(end_str): |
| 487 | raise ValueError( |
| 488 | f"numa_bind_cpus ranges must be ascending, but got '{cpuset}'." |
| 489 | ) |
| 490 | return value |
| 491 | |
| 492 | @model_validator(mode="after") |
| 493 | def _validate_parallel_config(self) -> Self: |
| 494 | if self._api_process_rank >= self._api_process_count: |
| 495 | raise ValueError( |
| 496 | "Invalid value of `_api_process_rank`. " |
| 497 | f"Expected to be `-1` or `[0, {self._api_process_count})`, " |
| 498 | f"but found: {self._api_process_rank}" |
| 499 | ) |
| 500 | |
| 501 | if self.enable_fault_tolerance and self._api_process_count > 1: |
| 502 | raise ValueError( |
| 503 | "Fault tolerance requires a single API server process " |
| 504 | f"(--api-server-count=1), but got {self._api_process_count}. " |
| 505 | "The FT system assumes one AsyncMPClient manages all engines." |
| 506 | ) |
| 507 | |
| 508 | if self.all2all_backend in ["pplx", "naive"]: |
| 509 | logger.warning( |
| 510 | "The '%s' all2all backend has been removed. " |
| 511 | "Falling back to 'allgather_reducescatter'.", |
| 512 | self.all2all_backend, |
| 513 | ) |
| 514 | self.all2all_backend = "allgather_reducescatter" |
| 515 | |
| 516 | if self.data_parallel_size_local > self.data_parallel_size: |
| 517 | raise ValueError( |
| 518 | f"data_parallel_size_local ({self.data_parallel_size_local}) " |
| 519 | f"must be <= data_parallel_size ({self.data_parallel_size})" |
| 520 | ) |
| 521 | |
| 522 | if self.data_parallel_size <= 1 and self.data_parallel_external_lb: |
| 523 | raise ValueError( |
| 524 | "data_parallel_external_lb can only be set when data_parallel_size > 1" |
| 525 | ) |
| 526 | |
| 527 | if not self.numa_bind and ( |
| 528 | self.numa_bind_nodes is not None or self.numa_bind_cpus is not None |
| 529 | ): |
| 530 | raise ValueError( |
| 531 | "numa_bind_nodes and numa_bind_cpus require numa_bind=True." |
| 532 | ) |
| 533 | |
| 534 | if self.enable_eplb: |
| 535 | if not current_platform.is_cuda_alike() and not current_platform.is_xpu(): |
| 536 | raise ValueError( |
| 537 | "Expert parallelism load balancing is only supported on " |
| 538 | "CUDA devices or ROCm devices or XPU devices now." |
| 539 | ) |
| 540 | if not self.enable_expert_parallel: |
| 541 | raise ValueError("enable_expert_parallel must be True to use EPLB.") |
| 542 | # The EP group spans the TP x PCP x DP ranks. EPLB therefore needs |
| 543 | # TP, PCP, or DP > 1. |
| 544 | if ( |
| 545 | self.tensor_parallel_size |
| 546 | * self.prefill_context_parallel_size |
| 547 | * self.data_parallel_size |
| 548 | <= 1 |
| 549 | ): |
| 550 | raise ValueError( |
| 551 | "EPLB requires tensor, prefill-context, or data parallelism, " |
| 552 | f"but got TP={self.tensor_parallel_size}, " |
| 553 | f"PCP={self.prefill_context_parallel_size}, " |
| 554 | f"DP={self.data_parallel_size}." |
| 555 | ) |
| 556 | else: |
| 557 | if self.eplb_config.num_redundant_experts != 0: |
| 558 | raise ValueError( |
| 559 | "num_redundant_experts is set to " |
| 560 | f"{self.eplb_config.num_redundant_experts} but EPLB is not " |
| 561 | "enabled. Either enable EPLB or unset " |
| 562 | "num_redundant_experts." |
| 563 | ) |
| 564 | |
| 565 | tp = self.tensor_parallel_size |
| 566 | pcp = self.prefill_context_parallel_size |
| 567 | dcp = self.decode_context_parallel_size |
| 568 | if pcp == 1: |
| 569 | # DCP reuses the TP ranks when PCP is disabled. |
| 570 | if tp % dcp != 0: |
| 571 | raise ValueError(f"tp_size={tp} must be divisible by dcp_size={dcp}.") |
| 572 | elif dcp not in (1, pcp, tp * pcp): |
| 573 | raise ValueError( |
| 574 | "When PCP is enabled, DCP must be disabled, span the PCP " |
| 575 | "axis, or span the full TP x PCP axis. " |
| 576 | f"Got TP={tp}, PCP={pcp}, DCP={dcp}; valid DCP sizes are " |
| 577 | f"{sorted({1, pcp, tp * pcp})}." |
| 578 | ) |
| 579 | |
| 580 | return self |
| 581 | |
| 582 | def set_dcp_defaults( |
| 583 | self, |
| 584 | comm_backend: DCPCommBackend = "ag_rs", |
| 585 | q_replicate: bool = False, |
| 586 | ) -> None: |
| 587 | """Fill in the DCP options the user left unset. |
| 588 | |
| 589 | Models can set their preferred DCP settings by calling this from their |
| 590 | `verify_and_update_config` hook. |
| 591 | """ |
| 592 | if self.dcp_comm_backend is None: |
| 593 | self.dcp_comm_backend = comm_backend |
| 594 | if self.dcp_q_replicate is None: |
| 595 | self.dcp_q_replicate = q_replicate |
| 596 | |
| 597 | @property |
| 598 | def world_size_across_dp(self) -> int: |
| 599 | """Process world size across TP, PCP, PP, and DP.""" |
| 600 | return self.world_size * self.data_parallel_size |
| 601 | |
| 602 | @property |
| 603 | def pcp_shard_decode_requests(self) -> bool: |
| 604 | """Whether PCP can shard decode requests across its ranks. |
| 605 | |
| 606 | PCP-only execution replicates the KV cache, so each decode request can |
| 607 | have a single PCP owner. DCP shards the KV cache and therefore requires |
| 608 | every decode request to run on every participating DCP rank. |
| 609 | """ |
| 610 | return ( |
| 611 | self.prefill_context_parallel_size > 1 |
| 612 | and self.decode_context_parallel_size == 1 |
| 613 | ) |
| 614 | |
| 615 | @property |
| 616 | def use_ubatching(self) -> bool: |
| 617 | return self.enable_dbo or self.ubatch_size > 1 |
| 618 | |
| 619 | @property |
| 620 | def num_ubatches(self) -> int: |
| 621 | return 2 if self.enable_dbo else self.ubatch_size |
| 622 | |
| 623 | @property |
| 624 | def local_engines_only(self) -> bool: |
| 625 | """Client manages local+remote EngineCores in pure internal LB case. |
| 626 | Client manages local EngineCores in hybrid and external LB case. |
| 627 | """ |
| 628 | return self.data_parallel_external_lb or self.data_parallel_hybrid_lb |
| 629 | |
| 630 | @property |
| 631 | def cpu_distributed_timeout(self) -> timedelta | None: |
| 632 | seconds = self.cpu_distributed_timeout_seconds |
| 633 | return timedelta(seconds=seconds) if seconds is not None else None |
| 634 | |
| 635 | def get_next_dp_init_port(self) -> int: |
| 636 | """We might need to initialize process groups in multiple |
| 637 | processes that is related to data parallelism, |
| 638 | e.g. both in the worker and in the engine, which |
| 639 | can live in different processes. To avoid port conflicts, we |
| 640 | pop a new port from the prepared port list each time we need to |
| 641 | initialize a new process group related to data parallelism. |
| 642 | """ |
| 643 | if self._data_parallel_master_port_list: |
| 644 | answer = self._data_parallel_master_port_list.pop() |
| 645 | else: |
| 646 | answer = self.data_parallel_master_port |
| 647 | self.data_parallel_master_port += 1 |
| 648 | |
| 649 | return answer |
| 650 | |
| 651 | def _pick_stateless_dp_port(self) -> tuple[int, socket.socket | None]: |
| 652 | """Return ``(port, listen_socket)`` for DP group init. |
| 653 | |
| 654 | With a coord store, rank 0 binds a socket and publishes the port; |
| 655 | others read it. Without one, pops a pre-allocated port and |
| 656 | returns ``listen_socket=None``. |
| 657 | """ |
| 658 | if not self._coord_store_port: |
| 659 | return self.get_next_dp_init_port(), None |
| 660 | |
| 661 | from vllm.distributed.utils import get_cached_tcp_store_client |
| 662 | |
| 663 | store = get_cached_tcp_store_client( |
| 664 | self.data_parallel_master_ip, self._coord_store_port |
| 665 | ) |
| 666 | |
| 667 | key = "dp_master_port" |
| 668 | if self.data_parallel_rank == 0: |
| 669 | s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| 670 | s.bind((self.data_parallel_master_ip, 0)) |
| 671 | s.listen() |
| 672 | port = s.getsockname()[1] |
| 673 | store.set(key, str(port).encode()) |
| 674 | return port, s |
| 675 | else: |
| 676 | return int(store.get(key).decode()), None |
| 677 | |
| 678 | @overload |
| 679 | def stateless_init_dp_group( |
| 680 | self, return_store: Literal[False] = ... |
| 681 | ) -> ProcessGroup: ... |
| 682 | @overload |
| 683 | def stateless_init_dp_group( |
| 684 | self, return_store: Literal[True] = ... |
| 685 | ) -> tuple[ProcessGroup, Store]: ... |
| 686 | def stateless_init_dp_group( |
| 687 | self, return_store: bool = False |
| 688 | ) -> ProcessGroup | tuple[ProcessGroup, Store]: |
| 689 | # NOTE: In high-concurrency scenarios multiple processes |
| 690 | # can pick the same (currently free) port through a race |
| 691 | # condition when calling `get_open_port()`. When the first |
| 692 | # process binds the port the others will subsequently fail |
| 693 | # with `torch.distributed.DistNetworkError: EADDRINUSE`. |
| 694 | # To make the initialization more robust we retry a few times |
| 695 | # with a fresh port whenever this specific error is observed. |
| 696 | from torch.distributed import DistNetworkError |
| 697 | |
| 698 | from vllm.distributed.utils import ( |
| 699 | stateless_init_torch_distributed_process_group, |
| 700 | ) |
| 701 | |
| 702 | max_retries = 5 |
| 703 | last_exc: Exception | None = None |
| 704 | for _ in range(max_retries): |
| 705 | try: |
| 706 | port, listen_socket = self._pick_stateless_dp_port() |
| 707 | # use gloo since the engine process might not have cuda device |
| 708 | return stateless_init_torch_distributed_process_group( |
| 709 | self.data_parallel_master_ip, |
| 710 | port, |
| 711 | self.data_parallel_rank, |
| 712 | self.data_parallel_size, |
| 713 | backend="gloo", |
| 714 | return_store=return_store, |
| 715 | listen_socket=listen_socket, |
| 716 | timeout=self.cpu_distributed_timeout, |
| 717 | ) |
| 718 | except DistNetworkError as e: |
| 719 | # We only want to retry when the root cause is EADDRINUSE. |
| 720 | if "EADDRINUSE" in str(e): |
| 721 | logger.warning("Address already in use. Retrying with a new port.") |
| 722 | last_exc = e |
| 723 | continue # try again with a new port |
| 724 | raise e |
| 725 | |
| 726 | # If we get here all retries have failed. |
| 727 | assert last_exc is not None |
| 728 | raise last_exc |
| 729 | |
| 730 | # The all_reduce at the end of attention (during o_proj) means that |
| 731 | # inputs are replicated across each rank of the tensor parallel group. |
| 732 | # If using expert-parallelism, replicated tokens results in useless |
| 733 | # duplicate computation and communication. |
| 734 | # |
| 735 | # In this case, ensure the input to the experts is sequence parallel |
| 736 | # to avoid the excess work. |
| 737 | # |
| 738 | @property |
| 739 | def use_sequence_parallel_moe(self) -> bool: |
| 740 | return ( |
| 741 | self.all2all_backend |
| 742 | in ( |
| 743 | "allgather_reducescatter", |
| 744 | "deepep_high_throughput", |
| 745 | "deepep_low_latency", |
| 746 | "deepep_v2", |
| 747 | "flashinfer_nvlink_one_sided", |
| 748 | "mori_high_throughput", |
| 749 | "mori_low_latency", |
| 750 | "nixl_ep", |
| 751 | ) |
| 752 | and self.enable_expert_parallel |
| 753 | and self.tensor_parallel_size > 1 |
| 754 | and self.data_parallel_size > 1 |
| 755 | ) |
| 756 | |
| 757 | @property |
| 758 | def use_all2all(self) -> bool: |
| 759 | return ( |
| 760 | self.data_parallel_size > 1 |
| 761 | or self.use_sequence_parallel_moe |
| 762 | or (self.enable_expert_parallel and self.prefill_context_parallel_size > 1) |
| 763 | ) |
| 764 | |
| 765 | @property |
| 766 | def use_batched_dp_moe(self) -> bool: |
| 767 | return ( |
| 768 | self.all2all_backend |
| 769 | in ( |
| 770 | "deepep_low_latency", |
| 771 | "nixl_ep", |
| 772 | ) |
| 773 | and self.enable_expert_parallel |
| 774 | and self.data_parallel_size > 1 |
| 775 | ) |
| 776 | |
| 777 | @property |
| 778 | def node_rank_within_dp(self) -> int: |
| 779 | return self.node_rank % self.nnodes_within_dp |
| 780 | |
| 781 | @property |
| 782 | def nnodes_within_dp(self) -> int: |
| 783 | """Number of nodes one DP replica spans. |
| 784 | |
| 785 | External LB pins ``data_parallel_size_local`` to 1, so the ratio |
| 786 | rounds down to 0 once DP replicas outnumber nodes. A replica that |
| 787 | does not span nodes still occupies exactly one. |
| 788 | """ |
| 789 | if self.nnodes == 1: |
| 790 | return 1 |
| 791 | data_parallel_node_size = ( |
| 792 | self.data_parallel_size // self.data_parallel_size_local |
| 793 | ) |
| 794 | nnodes_within_dp = self.nnodes // data_parallel_node_size |
| 795 | if self.data_parallel_external_lb: |
| 796 | return max(nnodes_within_dp, 1) |
| 797 | if self.nnodes % data_parallel_node_size != 0: |
| 798 | raise ValueError( |
| 799 | "Invalid data parallel configuration: " |
| 800 | f"nnodes ({self.nnodes}) must be divisible by the number of " |
| 801 | "data parallel node groups " |
| 802 | f"({data_parallel_node_size} = data_parallel_size " |
| 803 | f"{self.data_parallel_size} / data_parallel_size_local " |
| 804 | f"{self.data_parallel_size_local})" |
| 805 | ) |
| 806 | return nnodes_within_dp |
| 807 | |
| 808 | @property |
| 809 | def local_world_size(self) -> int: |
| 810 | return self.world_size // self.nnodes_within_dp |
| 811 | |
| 812 | @staticmethod |
| 813 | def has_unfinished_dp(dp_group: ProcessGroup, has_unfinished: bool) -> bool: |
| 814 | tensor = torch.tensor([has_unfinished], dtype=torch.int32, device="cpu") |
| 815 | # dp rank 0: has_unfinished_seqs=True |
| 816 | # dp rank 1: has_unfinished_seqs=False |
| 817 | # aggregated: has_unfinished_seqs=True |
| 818 | # so this is an OR operation, i.e. MAX in integers |
| 819 | torch.distributed.all_reduce(tensor, op=ReduceOp.MAX, group=dp_group) |
| 820 | aggregated_has_unfinished = bool(tensor.item()) |
| 821 | return aggregated_has_unfinished |
| 822 | |
| 823 | @staticmethod |
| 824 | def sync_dp_state( |
| 825 | dp_group: ProcessGroup, has_unfinished: bool, pending_pause: bool |
| 826 | ) -> tuple[bool, bool]: |
| 827 | """Combined all-reduce for DP state synchronization. |
| 828 | |
| 829 | Uses a single SUM all-reduce on a 2-element tensor: |
| 830 | [0] = 1 if this rank has unfinished work, else 0. |
| 831 | SUM > 0 ≡ logical OR across ranks → any rank has work. |
| 832 | [1] = 1 if this rank has a pending pause request, else 0. |
| 833 | SUM == dp_size ≡ all ranks reached pause consensus. |
| 834 | |
| 835 | has_unfinished_global is true if any rank has unfinished work, |
| 836 | or if some ranks are waiting for a pause consensus. |
| 837 | |
| 838 | Returns: |
| 839 | (has_unfinished_global, pause_consensus) |
| 840 | |
| 841 | """ |
| 842 | tensor = torch.tensor( |
| 843 | [int(has_unfinished), int(pending_pause)], dtype=torch.int32, device="cpu" |
| 844 | ) |
| 845 | torch.distributed.all_reduce(tensor, op=ReduceOp.SUM, group=dp_group) |
| 846 | dp_size = dp_group.size() |
| 847 | pause_count = tensor[1].item() |
| 848 | has_unfinished_global = tensor[0].item() > 0 or pause_count % dp_size != 0 |
| 849 | return has_unfinished_global, pause_count == dp_size |
| 850 | |
| 851 | @staticmethod |
| 852 | def sync_kv_cache_memory_size(dp_group: ProcessGroup, kv_cache_memory: int) -> int: |
| 853 | if kv_cache_memory == -1: |
| 854 | kv_cache_memory = torch.iinfo(torch.int64).max |
| 855 | tensor = torch.tensor([kv_cache_memory], dtype=torch.int64, device="cpu") |
| 856 | # we cannot use broadcast for stateless dp group since it depends |
| 857 | # on global rank |
| 858 | torch.distributed.all_reduce(tensor, op=ReduceOp.MIN, group=dp_group) |
| 859 | return tensor.item() |
| 860 | |
| 861 | def compute_hash(self): |
| 862 | """Provide a hash that uniquely identifies all the configs |
| 863 | that affect the structure of the computation |
| 864 | graph from input ids/embeddings to the final hidden states, |
| 865 | excluding anything before input ids/embeddings and after |
| 866 | the final hidden states. |
| 867 | |
| 868 | This hash is also used for DP worker configuration validation |
| 869 | to prevent hangs from mismatched collective communication patterns. |
| 870 | """ |
| 871 | ignored_factors = { |
| 872 | # Derived/runtime topology, networking, or launch details |
| 873 | "data_parallel_rank", |
| 874 | "data_parallel_rank_local", |
| 875 | "data_parallel_size_local", |
| 876 | "data_parallel_index", |
| 877 | "data_parallel_backend", |
| 878 | "data_parallel_external_lb", |
| 879 | "data_parallel_hybrid_lb", |
| 880 | "data_parallel_master_ip", |
| 881 | "data_parallel_master_port", |
| 882 | "_data_parallel_master_port_list", |
| 883 | "_coord_store_port", |
| 884 | "data_parallel_rpc_port", |
| 885 | "rank", |
| 886 | "master_addr", |
| 887 | "master_port", |
| 888 | "node_rank", |
| 889 | "nnodes", |
| 890 | "max_parallel_loading_workers", |
| 891 | "disable_custom_all_reduce", |
| 892 | "ray_workers_use_nsight", |
| 893 | "ray_runtime_env", |
| 894 | "placement_group", |
| 895 | "distributed_executor_backend", |
| 896 | "worker_cls", |
| 897 | "sd_worker_cls", |
| 898 | "worker_extension_cls", |
| 899 | "_api_process_count", |
| 900 | "_api_process_rank", |
| 901 | # NUMA binding is per-rank host-side memory locality; it does |
| 902 | # not affect collective-communication semantics. When numa_bind |
| 903 | # is enabled with auto-detection, each DP rank stores its own |
| 904 | # NUMA node in numa_bind_nodes (see vllm/utils/numa_utils.py |
| 905 | # `_get_numa_node`), which would otherwise diverge the DP hash. |
| 906 | "numa_bind", |
| 907 | "numa_bind_nodes", |
| 908 | "numa_bind_cpus", |
| 909 | "assigned_physical_gpu_ids", |
| 910 | } |
| 911 | |
| 912 | from vllm.config.utils import get_hash_factors, hash_factors |
| 913 | |
| 914 | factors = get_hash_factors(self, ignored_factors) |
| 915 | return hash_factors(factors) |
| 916 | |
| 917 | def __post_init__(self) -> None: |
| 918 | # Continue with the rest of the initialization |
| 919 | self.world_size = ( |
| 920 | self.pipeline_parallel_size |
| 921 | * self.tensor_parallel_size |
| 922 | * self.prefill_context_parallel_size |
| 923 | ) |
| 924 | |
| 925 | if self.distributed_executor_backend == "external_launcher": |
| 926 | logger.info("Using external launcher for distributed inference.") |
| 927 | self.world_size *= self.data_parallel_size |
| 928 | |
| 929 | if self.enable_elastic_ep: |
| 930 | if not self.enable_eplb: |
| 931 | raise ValueError("Elastic EP is only supported with enable_eplb=True.") |
| 932 | if self.pipeline_parallel_size > 1: |
| 933 | raise ValueError( |
| 934 | "Elastic EP is not supported with pipeline parallelism " |
| 935 | f"(pipeline_parallel_size={self.pipeline_parallel_size})." |
| 936 | ) |
| 937 | if self.data_parallel_external_lb or self.data_parallel_hybrid_lb: |
| 938 | raise NotImplementedError( |
| 939 | "Elastic EP is not compatible with data_parallel_external_lb " |
| 940 | "or data_parallel_hybrid_lb. Elastic EP relies on a single API " |
| 941 | "server and core client to coordinate scale up/down." |
| 942 | ) |
| 943 | if self.eplb_config.use_async: |
| 944 | from vllm.distributed.nixl_utils import is_nixl_available |
| 945 | |
| 946 | if not is_nixl_available(): |
| 947 | raise ValueError( |
| 948 | "Elastic EP with async EPLB requires the NIXL " |
| 949 | "package. Either install NIXL or set " |
| 950 | "--eplb-config.use_async=false." |
| 951 | ) |
| 952 | |
| 953 | if self.data_parallel_size > 1 or self.data_parallel_size_local == 0: |
| 954 | # Data parallel was specified in the engine args. |
| 955 | if self.distributed_executor_backend == "external_launcher": |
| 956 | # For external launcher, |
| 957 | # we need to set the data parallel rank automatically |
| 958 | self.data_parallel_rank = int(os.environ["RANK"]) // ( |
| 959 | self.world_size // self.data_parallel_size |
| 960 | ) |
| 961 | logger.info( |
| 962 | "Set data_parallel_rank to %d automatically.", |
| 963 | self.data_parallel_rank, |
| 964 | ) |
| 965 | if not self.enable_elastic_ep: |
| 966 | if not self._data_parallel_master_port_list: |
| 967 | self._data_parallel_master_port_list = get_open_ports_list(5) |
| 968 | self.data_parallel_master_port = ( |
| 969 | self._data_parallel_master_port_list.pop() |
| 970 | ) |
| 971 | |
| 972 | if not (0 <= self.data_parallel_rank < self.data_parallel_size): |
| 973 | raise ValueError( |
| 974 | f"data_parallel_rank ({self.data_parallel_rank})" |
| 975 | f" must be in the range [0, {self.data_parallel_size})" |
| 976 | ) |
| 977 | else: |
| 978 | # Otherwise fall back to env vars (e.g. for offline SPMD case). |
| 979 | self.data_parallel_size = envs.VLLM_DP_SIZE |
| 980 | self.data_parallel_rank = envs.VLLM_DP_RANK |
| 981 | self.data_parallel_rank_local = envs.VLLM_DP_RANK_LOCAL |
| 982 | self.data_parallel_master_ip = envs.VLLM_DP_MASTER_IP |
| 983 | self.data_parallel_master_port = envs.VLLM_DP_MASTER_PORT |
| 984 | |
| 985 | if self.data_parallel_size > 1 and self.is_moe_model is False: |
| 986 | raise ValueError( |
| 987 | "Offline data parallel mode is not supported/useful" |
| 988 | " for dense models." |
| 989 | ) |
| 990 | |
| 991 | max_dp_size = self.elastic_ep_max_dp_size |
| 992 | self.elastic_ep_max_dp_size = ( |
| 993 | max_dp_size |
| 994 | if self.enable_elastic_ep and max_dp_size is not None |
| 995 | else self.data_parallel_size |
| 996 | ) |
| 997 | if self.elastic_ep_max_dp_size < self.data_parallel_size: |
| 998 | raise ValueError( |
| 999 | "--elastic-ep-max-dp-size must be greater than or equal to " |
| 1000 | f"the initial data_parallel_size ({self.data_parallel_size})." |
| 1001 | ) |
| 1002 | |
| 1003 | self.data_parallel_index = self.data_parallel_rank |
| 1004 | |
| 1005 | if self.distributed_executor_backend == "external_launcher": |
| 1006 | os.environ["VLLM_ENABLE_V1_MULTIPROCESSING"] = "0" |
| 1007 | logger.info("Disabling V1 multiprocessing for external launcher.") |
| 1008 | |
| 1009 | if self.distributed_executor_backend is None and self.world_size_across_dp > 1: |
| 1010 | # We use multiprocessing by default if world_size fits on the |
| 1011 | # current node and we aren't in a ray placement group. |
| 1012 | |
| 1013 | from vllm.v1.executor import ray_utils |
| 1014 | |
| 1015 | backend: DistributedExecutorBackend = "mp" |
| 1016 | ray_found = ray_utils.ray_is_available() |
| 1017 | if current_platform.is_tpu() and envs.VLLM_XLA_USE_SPMD: |
| 1018 | backend = "uni" |
| 1019 | elif current_platform.is_cuda() and self.nnodes > 1: |
| 1020 | backend = "mp" |
| 1021 | elif ( |
| 1022 | current_platform.is_cuda() |
| 1023 | and current_platform.device_count() < self.world_size |
| 1024 | ): |
| 1025 | gpu_count = current_platform.device_count() |
| 1026 | raise ValueError( |
| 1027 | f"World size ({self.world_size}) is larger than the number of " |
| 1028 | f"available GPUs ({gpu_count}) in this node. If this is " |
| 1029 | "intentional and you are using:\n" |
| 1030 | "- ray, set '--distributed-executor-backend ray'.\n" |
| 1031 | "- multiprocessing, set '--nnodes' appropriately." |
| 1032 | ) |
| 1033 | elif self.data_parallel_backend == "ray": |
| 1034 | logger.info( |
| 1035 | "Using ray distributed inference because " |
| 1036 | "data_parallel_backend is ray" |
| 1037 | ) |
| 1038 | backend = "ray" |
| 1039 | elif ray_found: |
| 1040 | if self.placement_group: |
| 1041 | backend = "ray" |
| 1042 | else: |
| 1043 | from ray import is_initialized as ray_is_initialized |
| 1044 | |
| 1045 | if ray_is_initialized(): |
| 1046 | from ray.util import get_current_placement_group |
| 1047 | |
| 1048 | if get_current_placement_group(): |
| 1049 | backend = "ray" |
| 1050 | self.distributed_executor_backend = backend |
| 1051 | logger.debug("Defaulting to use %s for distributed inference", backend) |
| 1052 | |
| 1053 | if self.distributed_executor_backend is None and self.world_size == 1: |
| 1054 | self.distributed_executor_backend = "uni" |
| 1055 | |
| 1056 | if self.max_parallel_loading_workers is not None: |
| 1057 | logger.warning( |
| 1058 | "max_parallel_loading_workers is currently " |
| 1059 | "not supported and will be ignored." |
| 1060 | ) |
| 1061 | allowed_backends = ("mp", "uni", "external_launcher") |
| 1062 | if ( |
| 1063 | self.distributed_executor_backend not in allowed_backends |
| 1064 | and self.nnodes > 1 |
| 1065 | ): |
| 1066 | raise ValueError( |
| 1067 | "nnodes > 1 can only be set when distributed executor " |
| 1068 | "backend is mp, uni or external_launcher." |
| 1069 | ) |
| 1070 | |
| 1071 | if self.enable_eplb and self.eplb_config.communicator is None: |
| 1072 | # Prefer NIXL when available: zero-copy RDMA reads, compatible |
| 1073 | # with both async EPLB and elastic EP. |
| 1074 | # Fallbacks: pynccl for elastic EP (stateless groups need it), |
| 1075 | # torch_gloo for static EP. torch_nccl is avoided because NCCL |
| 1076 | # is incompatible with async EPLB (multi-stream conflicts) and |
| 1077 | # batched isend/irecv hangs under high load. |
| 1078 | # See https://github.com/pytorch/pytorch/issues/174288 |
| 1079 | from vllm.distributed.nixl_utils import is_nixl_available |
| 1080 | |
| 1081 | if current_platform.is_xpu(): |
| 1082 | # On XPU, use the device-native XCCL P2P backend. |
| 1083 | self.eplb_config.communicator = "torch_xccl" |
| 1084 | elif is_nixl_available(): |
| 1085 | self.eplb_config.communicator = "nixl" |
| 1086 | elif self.enable_elastic_ep: |
| 1087 | self.eplb_config.communicator = "pynccl" |
| 1088 | else: |
| 1089 | self.eplb_config.communicator = "torch_gloo" |
| 1090 | |
| 1091 | @property |
| 1092 | def use_ray(self) -> bool: |
| 1093 | return self.distributed_executor_backend == "ray" or ( |
| 1094 | isinstance(self.distributed_executor_backend, type) |
| 1095 | and getattr(self.distributed_executor_backend, "uses_ray", False) |
| 1096 | ) |
| 1097 | |
| 1098 | @model_validator(mode="after") |
| 1099 | def _verify_args(self) -> Self: |
| 1100 | # Lazy import to avoid circular import |
| 1101 | from vllm.v1.executor import Executor |
| 1102 | |
| 1103 | if ( |
| 1104 | self.distributed_executor_backend is not None |
| 1105 | and not isinstance(self.distributed_executor_backend, str) |
| 1106 | and not ( |
| 1107 | isinstance(self.distributed_executor_backend, type) |
| 1108 | and issubclass(self.distributed_executor_backend, Executor) |
| 1109 | ) |
| 1110 | ): |
| 1111 | raise ValueError( |
| 1112 | "Unrecognized distributed executor backend " |
| 1113 | f"{self.distributed_executor_backend}. Supported " |
| 1114 | "values are 'ray', 'mp' 'uni', 'external_launcher', " |
| 1115 | " custom Executor subclass or its import path." |
| 1116 | ) |
| 1117 | if self.use_ray: |
| 1118 | from vllm.v1.executor import ray_utils |
| 1119 | |
| 1120 | ray_utils.assert_ray_available() |
| 1121 | |
| 1122 | if not current_platform.use_custom_allreduce(): |
| 1123 | self.disable_custom_all_reduce = True |
| 1124 | logger.debug( |
| 1125 | "Disabled the custom all-reduce kernel because it is not " |
| 1126 | "supported on current platform." |
| 1127 | ) |
| 1128 | if self.ray_workers_use_nsight and not self.use_ray: |
| 1129 | raise ValueError( |
| 1130 | "Unable to use nsight profiling unless workers run with Ray." |
| 1131 | ) |
| 1132 | |
| 1133 | # A batch below one token per microbatch cannot be split, so the |
| 1134 | # thresholds have to keep it out rather than the split having to cope. |
| 1135 | if self.use_ubatching and ( |
| 1136 | min(self.dbo_decode_token_threshold, self.dbo_prefill_token_threshold) |
| 1137 | < self.num_ubatches |
| 1138 | ): |
| 1139 | raise ValueError( |
| 1140 | "dbo_decode_token_threshold and dbo_prefill_token_threshold must " |
| 1141 | f"be at least the number of microbatches ({self.num_ubatches})." |
| 1142 | ) |
| 1143 | |
| 1144 | return self |
| 1145 | |
| 1146 | def reconfigure_for_independent_dp_rank(self) -> None: |
| 1147 | """Reconfigure for a single independent non-MoE DP rank.""" |
| 1148 | # Capture these before changing DP fields. |
| 1149 | nnodes = self.nnodes_within_dp |
| 1150 | node_rank = self.node_rank_within_dp |
| 1151 | self.data_parallel_size = 1 |
| 1152 | self.data_parallel_size_local = 1 |
| 1153 | self.data_parallel_rank = 0 |
| 1154 | self.nnodes = nnodes |
| 1155 | self.node_rank = node_rank |