| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | |
| 4 | import argparse |
| 5 | import signal |
| 6 | import time |
| 7 | |
| 8 | import uvloop |
| 9 | |
| 10 | import vllm |
| 11 | import vllm.envs as envs |
| 12 | from vllm.entrypoints.cli.types import CLISubcommand |
| 13 | from vllm.entrypoints.launchers.api_server.entry import run_server, setup_server |
| 14 | from vllm.entrypoints.launchers.cli_args import ( |
| 15 | make_arg_parser, |
| 16 | propagate_flash_late_interaction, |
| 17 | validate_parsed_serve_args, |
| 18 | ) |
| 19 | from vllm.entrypoints.launchers.dp_supervisor import run_dp_supervisor |
| 20 | from vllm.entrypoints.launchers.launcher import create_server_socket |
| 21 | from vllm.entrypoints.serve.utils.api_utils import VLLM_SUBCMD_PARSER_EPILOG |
| 22 | from vllm.logger import init_logger |
| 23 | from vllm.reasoning import ReasoningParserManager |
| 24 | from vllm.usage.usage_lib import UsageContext |
| 25 | from vllm.utils.argparse_utils import FlexibleArgumentParser |
| 26 | from vllm.utils.network_utils import get_tcp_uri |
| 27 | from vllm.v1.engine.utils import CoreEngineProcManager, launch_core_engines |
| 28 | from vllm.v1.executor import Executor |
| 29 | from vllm.v1.executor.multiproc_executor import MultiprocExecutor |
| 30 | from vllm.v1.metrics.prometheus import setup_multiprocess_prometheus |
| 31 | from vllm.v1.utils import ( |
| 32 | APIServerProcessManager, |
| 33 | RustFrontendProcessManager, |
| 34 | wait_for_completion_or_failure, |
| 35 | ) |
| 36 | |
| 37 | logger = init_logger(__name__) |
| 38 | |
| 39 | DESCRIPTION = """Launch a local OpenAI-compatible API server to serve LLM |
| 40 | completions via HTTP. Defaults to Qwen/Qwen3-0.6B if no model is specified. |
| 41 | |
| 42 | Search by using: `--help=<ConfigGroup>` to explore options by section (e.g., |
| 43 | --help=ModelConfig, --help=Frontend) |
| 44 | Use `--help=all` to show all available flags at once. |
| 45 | """ |
| 46 | |
| 47 | |
| 48 | class ServeSubcommand(CLISubcommand): |
| 49 | """The `serve` subcommand for the vLLM CLI.""" |
| 50 | |
| 51 | name = "serve" |
| 52 | |
| 53 | @staticmethod |
| 54 | def cmd(args: argparse.Namespace) -> None: |
| 55 | # If model is specified in CLI (as positional arg), it takes precedence |
| 56 | if hasattr(args, "model_tag") and args.model_tag is not None: |
| 57 | args.model = args.model_tag |
| 58 | |
| 59 | if getattr(args, "grpc", False): |
| 60 | from vllm.entrypoints.launchers.grpc_server import serve_grpc |
| 61 | |
| 62 | uvloop.run(serve_grpc(args)) |
| 63 | return |
| 64 | |
| 65 | rust_frontend_path = ( |
| 66 | envs.VLLM_RUST_FRONTEND_PATH if envs.VLLM_USE_RUST_FRONTEND else None |
| 67 | ) |
| 68 | |
| 69 | if args.headless: |
| 70 | if args.api_server_count is not None and args.api_server_count > 0: |
| 71 | raise ValueError( |
| 72 | f"--api-server-count={args.api_server_count} cannot be " |
| 73 | "used with --headless (no API servers are started in " |
| 74 | "headless mode)." |
| 75 | ) |
| 76 | # Default to 0 in headless mode (no API servers) |
| 77 | args.api_server_count = 0 |
| 78 | |
| 79 | # Detect LB mode for defaulting api_server_count. |
| 80 | # Multi-port: --data-parallel-multi-port-external-lb |
| 81 | # External LB: --data-parallel-external-lb or --data-parallel-rank |
| 82 | # Hybrid LB: --data-parallel-hybrid-lb or --data-parallel-start-rank |
| 83 | is_external_lb = ( |
| 84 | args.data_parallel_external_lb or args.data_parallel_rank is not None |
| 85 | ) |
| 86 | |
| 87 | # If --data_parallel_multi_port_external_lb and --data_parallel_hybrid_lb |
| 88 | # are unset, default to hybrid if --data-parallel-start-rank is set |
| 89 | is_hybrid_lb = is_multi_port = False |
| 90 | if ( |
| 91 | not args.data_parallel_hybrid_lb |
| 92 | and not args.data_parallel_multi_port_external_lb |
| 93 | ): |
| 94 | is_hybrid_lb = args.data_parallel_start_rank is not None |
| 95 | else: |
| 96 | is_hybrid_lb = args.data_parallel_hybrid_lb |
| 97 | is_multi_port = args.data_parallel_multi_port_external_lb |
| 98 | |
| 99 | if sum([is_multi_port, is_external_lb, is_hybrid_lb]) > 1: |
| 100 | raise ValueError( |
| 101 | "Cannot use more than one data parallel load balancing mode. " |
| 102 | "Choose one of: --data-parallel-multi-port-external-lb, " |
| 103 | "--data-parallel-external-lb (or --data-parallel-rank), " |
| 104 | "--data-parallel-hybrid-lb (or --data-parallel-start-rank)." |
| 105 | ) |
| 106 | |
| 107 | # Default api_server_count if not explicitly set. |
| 108 | # - Multi-port: 1 (supervisor spawns one server per local DP rank) |
| 109 | # - Rust frontend: 1 (not applicable as it's multithreaded) |
| 110 | # - External LB: 1 (external LB handles distribution) |
| 111 | # - Hybrid LB: Use local DP size (internal LB for local ranks only) |
| 112 | # - Internal LB: Use full DP size |
| 113 | if args.api_server_count is None: |
| 114 | if is_multi_port or is_external_lb or rust_frontend_path: |
| 115 | args.api_server_count = 1 |
| 116 | elif is_hybrid_lb: |
| 117 | args.api_server_count = args.data_parallel_size_local or 1 |
| 118 | if args.api_server_count > 1: |
| 119 | logger.info( |
| 120 | "Defaulting api_server_count to data_parallel_size_local " |
| 121 | "(%d) for hybrid LB mode.", |
| 122 | args.api_server_count, |
| 123 | ) |
| 124 | else: |
| 125 | args.api_server_count = args.data_parallel_size |
| 126 | if args.api_server_count > 1: |
| 127 | logger.info( |
| 128 | "Defaulting api_server_count to data_parallel_size (%d).", |
| 129 | args.api_server_count, |
| 130 | ) |
| 131 | elif rust_frontend_path and args.api_server_count > 1: |
| 132 | logger.warning( |
| 133 | "Ignoring --api-server-count=%d when using rust front-end process", |
| 134 | args.api_server_count, |
| 135 | ) |
| 136 | args.api_server_count = 1 |
| 137 | |
| 138 | # Elastic EP currently only supports running with at most one API server. |
| 139 | if getattr(args, "enable_elastic_ep", False) and args.api_server_count > 1: |
| 140 | logger.warning( |
| 141 | "Elastic EP only supports running with with at most one API server. " |
| 142 | "Capping api_server_count from %d to 1.", |
| 143 | args.api_server_count, |
| 144 | ) |
| 145 | args.api_server_count = 1 |
| 146 | |
| 147 | if is_multi_port: |
| 148 | run_dp_supervisor(args) |
| 149 | elif args.api_server_count < 1: |
| 150 | run_headless(args) |
| 151 | elif args.api_server_count > 1 or rust_frontend_path: |
| 152 | run_multi_api_server(args) |
| 153 | else: |
| 154 | # Single API server (this process). |
| 155 | args.api_server_count = None |
| 156 | uvloop.run(run_server(args)) |
| 157 | |
| 158 | def validate(self, args: argparse.Namespace) -> None: |
| 159 | validate_parsed_serve_args(args) |
| 160 | |
| 161 | def subparser_init( |
| 162 | self, subparsers: argparse._SubParsersAction |
| 163 | ) -> FlexibleArgumentParser: |
| 164 | serve_parser = subparsers.add_parser( |
| 165 | self.name, |
| 166 | help="Launch a local OpenAI-compatible API server to serve LLM " |
| 167 | "completions via HTTP.", |
| 168 | description=DESCRIPTION, |
| 169 | usage="vllm serve [model_tag] [options]", |
| 170 | ) |
| 171 | |
| 172 | serve_parser = make_arg_parser(serve_parser) |
| 173 | serve_parser.epilog = VLLM_SUBCMD_PARSER_EPILOG.format(subcmd=self.name) |
| 174 | return serve_parser |
| 175 | |
| 176 | |
| 177 | def cmd_init() -> list[CLISubcommand]: |
| 178 | return [ServeSubcommand()] |
| 179 | |
| 180 | |
| 181 | def run_headless(args: argparse.Namespace): |
| 182 | if args.api_server_count > 1: |
| 183 | raise ValueError("api_server_count can't be set in headless mode") |
| 184 | |
| 185 | if args.reasoning_parser_plugin and len(args.reasoning_parser_plugin) > 3: |
| 186 | ReasoningParserManager.import_reasoning_parser(args.reasoning_parser_plugin) |
| 187 | |
| 188 | # Create the EngineConfig. |
| 189 | engine_args = vllm.AsyncEngineArgs.from_cli_args(args) |
| 190 | propagate_flash_late_interaction(args, engine_args) |
| 191 | usage_context = UsageContext.OPENAI_API_SERVER |
| 192 | vllm_config = engine_args.create_engine_config( |
| 193 | usage_context=usage_context, headless=True |
| 194 | ) |
| 195 | |
| 196 | if engine_args.data_parallel_hybrid_lb: |
| 197 | raise ValueError("data_parallel_hybrid_lb is not applicable in headless mode") |
| 198 | |
| 199 | parallel_config = vllm_config.parallel_config |
| 200 | local_engine_count = parallel_config.data_parallel_size_local |
| 201 | |
| 202 | if local_engine_count <= 0: |
| 203 | raise ValueError("data_parallel_size_local must be > 0 in headless mode") |
| 204 | |
| 205 | shutdown_requested = False |
| 206 | |
| 207 | # Catch SIGTERM and SIGINT to allow graceful shutdown. |
| 208 | def signal_handler(signum, frame): |
| 209 | nonlocal shutdown_requested |
| 210 | logger.debug("Received %d signal.", signum) |
| 211 | if not shutdown_requested: |
| 212 | shutdown_requested = True |
| 213 | raise SystemExit |
| 214 | |
| 215 | signal.signal(signal.SIGTERM, signal_handler) |
| 216 | signal.signal(signal.SIGINT, signal_handler) |
| 217 | |
| 218 | if parallel_config.node_rank_within_dp > 0: |
| 219 | from vllm.version import __version__ as VLLM_VERSION |
| 220 | |
| 221 | # Run headless workers (for multi-node PP/TP). |
| 222 | host = parallel_config.master_addr |
| 223 | head_node_address = f"{host}:{parallel_config.master_port}" |
| 224 | logger.info( |
| 225 | "Launching vLLM (v%s) headless multiproc executor, " |
| 226 | "with head node address %s for torch.distributed process group.", |
| 227 | VLLM_VERSION, |
| 228 | head_node_address, |
| 229 | ) |
| 230 | |
| 231 | executor = MultiprocExecutor(vllm_config, monitor_workers=False) |
| 232 | executor.start_worker_monitor(inline=True) |
| 233 | return |
| 234 | |
| 235 | host = parallel_config.data_parallel_master_ip |
| 236 | port = parallel_config.data_parallel_rpc_port |
| 237 | handshake_address = get_tcp_uri(host, port) |
| 238 | |
| 239 | logger.info( |
| 240 | "Launching %d data parallel engine(s) in headless mode, " |
| 241 | "with head node address %s.", |
| 242 | local_engine_count, |
| 243 | handshake_address, |
| 244 | ) |
| 245 | |
| 246 | # Create the engines. |
| 247 | engine_manager = CoreEngineProcManager( |
| 248 | local_engine_count=local_engine_count, |
| 249 | start_index=vllm_config.parallel_config.data_parallel_rank, |
| 250 | local_start_index=0, |
| 251 | vllm_config=vllm_config, |
| 252 | local_client=False, |
| 253 | handshake_address=handshake_address, |
| 254 | executor_class=Executor.get_class(vllm_config), |
| 255 | log_stats=not engine_args.disable_log_stats, |
| 256 | ) |
| 257 | |
| 258 | try: |
| 259 | engine_manager.monitor_engine_liveness() |
| 260 | finally: |
| 261 | timeout = None |
| 262 | if shutdown_requested: |
| 263 | timeout = vllm_config.shutdown_timeout |
| 264 | logger.info("Waiting up to %d seconds for processes to exit", timeout) |
| 265 | engine_manager.shutdown(timeout=timeout) |
| 266 | logger.info("Shutting down.") |
| 267 | |
| 268 | |
| 269 | def run_multi_api_server(args: argparse.Namespace): |
| 270 | assert not args.headless |
| 271 | rust_frontend_path = ( |
| 272 | envs.VLLM_RUST_FRONTEND_PATH if envs.VLLM_USE_RUST_FRONTEND else None |
| 273 | ) |
| 274 | num_api_servers: int = args.api_server_count |
| 275 | assert num_api_servers > 0 |
| 276 | |
| 277 | if rust_frontend_path and num_api_servers > 1: |
| 278 | raise ValueError( |
| 279 | "VLLM_RUST_FRONTEND_PATH does not support api_server_count > 1" |
| 280 | ) |
| 281 | |
| 282 | if num_api_servers > 1: |
| 283 | setup_multiprocess_prometheus() |
| 284 | |
| 285 | shutdown_requested = False |
| 286 | |
| 287 | # Catch SIGTERM and SIGINT to allow graceful shutdown. |
| 288 | def signal_handler(signum, frame): |
| 289 | nonlocal shutdown_requested |
| 290 | logger.debug("Received %d signal.", signum) |
| 291 | if not shutdown_requested: |
| 292 | shutdown_requested = True |
| 293 | raise SystemExit |
| 294 | |
| 295 | signal.signal(signal.SIGTERM, signal_handler) |
| 296 | signal.signal(signal.SIGINT, signal_handler) |
| 297 | |
| 298 | listen_address, sock = setup_server(args, reuse_port=num_api_servers > 1) |
| 299 | # `--grpc-port` is only accepted for the Rust frontend, which inherits this |
| 300 | # listener like the HTTP one. gRPC follows the HTTP TCP host, or IPv4 |
| 301 | # loopback when HTTP uses a Unix socket. |
| 302 | grpc_sock = None |
| 303 | if args.grpc_port is not None: |
| 304 | grpc_host = "127.0.0.1" if args.uds else (args.host or "") |
| 305 | grpc_sock = create_server_socket((grpc_host, args.grpc_port), reuse_port=False) |
| 306 | |
| 307 | engine_args = vllm.AsyncEngineArgs.from_cli_args(args) |
| 308 | propagate_flash_late_interaction(args, engine_args) |
| 309 | engine_args._api_process_count = num_api_servers |
| 310 | engine_args._api_process_rank = -1 |
| 311 | |
| 312 | usage_context = UsageContext.OPENAI_API_SERVER |
| 313 | vllm_config = engine_args.create_engine_config(usage_context=usage_context) |
| 314 | |
| 315 | if num_api_servers > 1 and envs.VLLM_ALLOW_RUNTIME_LORA_UPDATING: |
| 316 | raise ValueError( |
| 317 | "VLLM_ALLOW_RUNTIME_LORA_UPDATING cannot be used with api_server_count > 1" |
| 318 | ) |
| 319 | |
| 320 | executor_class = Executor.get_class(vllm_config) |
| 321 | log_stats = not engine_args.disable_log_stats |
| 322 | |
| 323 | parallel_config = vllm_config.parallel_config |
| 324 | dp_rank = parallel_config.data_parallel_rank |
| 325 | assert parallel_config.local_engines_only or dp_rank == 0 |
| 326 | |
| 327 | api_server_manager: APIServerProcessManager | RustFrontendProcessManager | None = ( |
| 328 | None |
| 329 | ) |
| 330 | |
| 331 | from vllm.v1.engine.utils import get_engine_zmq_addresses |
| 332 | |
| 333 | # Defer port allocation to the child's bind() to avoid TOCTOU, except |
| 334 | # for Rust front-end and Ray DP, which can't see the post-bind rebind |
| 335 | # (CLI-arg subprocess / pickled-into-actor snapshot respectively) and |
| 336 | # so pre-allocate driver-side -- reintroducing the original race only |
| 337 | # there. |
| 338 | is_ray_dp = parallel_config.data_parallel_backend == "ray" |
| 339 | addresses = get_engine_zmq_addresses( |
| 340 | vllm_config, |
| 341 | num_api_servers, |
| 342 | defer_api_server_ports=not (rust_frontend_path or is_ray_dp), |
| 343 | ) |
| 344 | |
| 345 | with launch_core_engines( |
| 346 | vllm_config, executor_class, log_stats, addresses |
| 347 | ) as engine_launch: |
| 348 | local_engine_manager = engine_launch.engine_manager |
| 349 | coordinator = engine_launch.coordinator |
| 350 | addresses = engine_launch.addresses |
| 351 | stats_update_address = ( |
| 352 | coordinator.get_stats_publish_address() if coordinator else None |
| 353 | ) |
| 354 | |
| 355 | if rust_frontend_path: |
| 356 | if parallel_config.local_engines_only: |
| 357 | expected_engine_start_index = parallel_config.data_parallel_rank |
| 358 | expected_engine_count = parallel_config.data_parallel_size_local |
| 359 | else: |
| 360 | expected_engine_start_index = 0 |
| 361 | expected_engine_count = parallel_config.data_parallel_size |
| 362 | # Start rust front-end process. |
| 363 | api_server_manager = RustFrontendProcessManager( |
| 364 | binary_path=rust_frontend_path, |
| 365 | sock=sock, |
| 366 | grpc_sock=grpc_sock, |
| 367 | args=args, |
| 368 | input_address=addresses.inputs[0], |
| 369 | output_address=addresses.outputs[0], |
| 370 | engine_start_index=expected_engine_start_index, |
| 371 | engine_count=expected_engine_count, |
| 372 | data_parallel_size=parallel_config.data_parallel_size, |
| 373 | stats_update_address=stats_update_address, |
| 374 | ) |
| 375 | else: |
| 376 | # Start API server(s). |
| 377 | api_server_manager = APIServerProcessManager( |
| 378 | listen_address=listen_address, |
| 379 | sock=sock, |
| 380 | args=args, |
| 381 | num_servers=num_api_servers, |
| 382 | input_addresses=addresses.inputs, |
| 383 | output_addresses=addresses.outputs, |
| 384 | stats_update_address=stats_update_address, |
| 385 | tensor_queue=engine_launch.tensor_queue, |
| 386 | ) |
| 387 | |
| 388 | if not is_ray_dp: |
| 389 | # Forward each child's bound endpoints to the engine handshake |
| 390 | # (runs on ``with`` exit). Skipped for Ray DP, where addresses |
| 391 | # are pre-allocated above and Ray actors already hold them. |
| 392 | actual_inputs, actual_outputs = ( |
| 393 | api_server_manager.gather_actual_addresses() |
| 394 | ) |
| 395 | addresses.inputs = actual_inputs |
| 396 | addresses.outputs = actual_outputs |
| 397 | |
| 398 | # Set frontend processes to watch during engine startup. |
| 399 | # If any of these processes exit before the engines are up, the engine startup |
| 400 | # will be aborted with an error. |
| 401 | engine_launch.watched_frontend_processes = api_server_manager.processes |
| 402 | |
| 403 | # Wait for API servers. |
| 404 | try: |
| 405 | wait_for_completion_or_failure( |
| 406 | api_server_manager=api_server_manager, |
| 407 | engine_manager=local_engine_manager, |
| 408 | coordinator=coordinator, |
| 409 | ) |
| 410 | finally: |
| 411 | timeout = shutdown_by = None |
| 412 | if shutdown_requested: |
| 413 | timeout = vllm_config.shutdown_timeout |
| 414 | shutdown_by = time.monotonic() + timeout |
| 415 | logger.info("Waiting up to %d seconds for processes to exit", timeout) |
| 416 | |
| 417 | def to_timeout(deadline: float | None) -> float | None: |
| 418 | return ( |
| 419 | deadline if deadline is None else max(deadline - time.monotonic(), 0.0) |
| 420 | ) |
| 421 | |
| 422 | api_server_manager.shutdown(timeout=timeout) |
| 423 | if local_engine_manager: |
| 424 | local_engine_manager.shutdown(timeout=to_timeout(shutdown_by)) |
| 425 | if coordinator: |
| 426 | coordinator.shutdown(timeout=to_timeout(shutdown_by)) |