| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | |
| 4 | import asyncio |
| 5 | import contextlib |
| 6 | import signal |
| 7 | import socket |
| 8 | from collections.abc import Generator |
| 9 | from functools import partial |
| 10 | from typing import Any |
| 11 | |
| 12 | import uvicorn |
| 13 | from fastapi import FastAPI |
| 14 | |
| 15 | from vllm import envs |
| 16 | from vllm.engine.protocol import EngineClient |
| 17 | from vllm.entrypoints.launchers.utils.ssl import SSLCertRefresher |
| 18 | from vllm.entrypoints.serve.utils.api_utils import ( |
| 19 | log_non_default_args, |
| 20 | log_version_and_model, |
| 21 | ) |
| 22 | from vllm.logger import init_logger |
| 23 | from vllm.reasoning import ReasoningParserManager |
| 24 | from vllm.tool_parsers import ToolParserManager |
| 25 | from vllm.tracing import instrument |
| 26 | from vllm.utils.network_utils import find_process_using_port, is_valid_ipv6_address |
| 27 | from vllm.utils.system_utils import set_ulimit |
| 28 | from vllm.version import __version__ as VLLM_VERSION |
| 29 | |
| 30 | from .utils.constants import ( |
| 31 | H11_MAX_HEADER_COUNT_DEFAULT, |
| 32 | H11_MAX_INCOMPLETE_EVENT_SIZE_DEFAULT, |
| 33 | ) |
| 34 | |
| 35 | logger = init_logger(__name__) |
| 36 | |
| 37 | |
| 38 | class NoSignalServer(uvicorn.Server): |
| 39 | """Uvicorn server that never installs its own SIGINT/SIGTERM handlers. |
| 40 | |
| 41 | Callers register their own handlers on the event loop for graceful |
| 42 | shutdown; uvicorn's would race with and override them (see #49668). |
| 43 | """ |
| 44 | |
| 45 | @contextlib.contextmanager |
| 46 | def capture_signals(self) -> Generator[None, None, None]: |
| 47 | yield |
| 48 | |
| 49 | |
| 50 | async def serve_http( |
| 51 | app: FastAPI, |
| 52 | sock: socket.socket | None, |
| 53 | enable_ssl_refresh: bool = False, |
| 54 | **uvicorn_kwargs: Any, |
| 55 | ): |
| 56 | """Start a FastAPI app using Uvicorn, with support for custom Uvicorn config |
| 57 | options. Supports http header limits via h11_max_incomplete_event_size and |
| 58 | h11_max_header_count. |
| 59 | """ |
| 60 | logger.info("Available routes are:") |
| 61 | # post endpoints |
| 62 | for route in app.routes: |
| 63 | methods = getattr(route, "methods", None) |
| 64 | path = getattr(route, "path", None) |
| 65 | |
| 66 | if methods is None or path is None: |
| 67 | continue |
| 68 | |
| 69 | logger.info("Route: %s, Methods: %s", path, ", ".join(methods)) |
| 70 | |
| 71 | # other endpoints |
| 72 | for route in app.routes: |
| 73 | endpoint = getattr(route, "endpoint", None) |
| 74 | methods = getattr(route, "methods", None) |
| 75 | path = getattr(route, "path", None) |
| 76 | |
| 77 | if endpoint is None or path is None or methods is not None: |
| 78 | continue |
| 79 | |
| 80 | logger.info("Route: %s, Endpoint: %s", path, endpoint.__name__) |
| 81 | |
| 82 | # Extract header limit options if present |
| 83 | h11_max_incomplete_event_size = uvicorn_kwargs.pop( |
| 84 | "h11_max_incomplete_event_size", None |
| 85 | ) |
| 86 | h11_max_header_count = uvicorn_kwargs.pop("h11_max_header_count", None) |
| 87 | |
| 88 | # Set safe defaults if not provided |
| 89 | if h11_max_incomplete_event_size is None: |
| 90 | h11_max_incomplete_event_size = H11_MAX_INCOMPLETE_EVENT_SIZE_DEFAULT |
| 91 | if h11_max_header_count is None: |
| 92 | h11_max_header_count = H11_MAX_HEADER_COUNT_DEFAULT |
| 93 | |
| 94 | config = uvicorn.Config(app, **uvicorn_kwargs) |
| 95 | # Set header limits |
| 96 | config.h11_max_incomplete_event_size = h11_max_incomplete_event_size |
| 97 | config.h11_max_header_count = h11_max_header_count |
| 98 | config.load() |
| 99 | server = NoSignalServer(config) |
| 100 | app.state.server = server |
| 101 | |
| 102 | loop = asyncio.get_running_loop() |
| 103 | |
| 104 | engine_client = app.state.engine_client |
| 105 | watchdog_task = ( |
| 106 | loop.create_task(watchdog_loop(server, engine_client)) |
| 107 | if engine_client is not None |
| 108 | else None |
| 109 | ) |
| 110 | server_task = loop.create_task(server.serve(sockets=[sock] if sock else None)) |
| 111 | |
| 112 | ssl_cert_refresher = ( |
| 113 | None |
| 114 | if not enable_ssl_refresh |
| 115 | else SSLCertRefresher( |
| 116 | ssl_context=config.ssl, |
| 117 | key_path=config.ssl_keyfile, |
| 118 | cert_path=config.ssl_certfile, |
| 119 | ca_path=config.ssl_ca_certs, |
| 120 | ) |
| 121 | ) |
| 122 | |
| 123 | shutdown_event = asyncio.Event() |
| 124 | |
| 125 | def signal_handler() -> None: |
| 126 | if shutdown_event.is_set(): |
| 127 | return |
| 128 | logger.info_once("[shutdown] API server: shutdown triggered") |
| 129 | shutdown_event.set() |
| 130 | |
| 131 | async def dummy_shutdown() -> None: |
| 132 | pass |
| 133 | |
| 134 | loop.add_signal_handler(signal.SIGINT, signal_handler) |
| 135 | loop.add_signal_handler(signal.SIGTERM, signal_handler) |
| 136 | |
| 137 | async def handle_shutdown() -> None: |
| 138 | await shutdown_event.wait() |
| 139 | |
| 140 | if engine_client is not None: |
| 141 | timeout = engine_client.vllm_config.shutdown_timeout |
| 142 | mode = "abort" if timeout == 0 else "drain" |
| 143 | |
| 144 | logger.info( |
| 145 | "[shutdown] API server: stopping engine client mode=%s timeout=%ss", |
| 146 | mode, |
| 147 | timeout, |
| 148 | ) |
| 149 | |
| 150 | await loop.run_in_executor( |
| 151 | None, partial(engine_client.shutdown, timeout=timeout) |
| 152 | ) |
| 153 | logger.info_once("[shutdown] API server: engine client stopped") |
| 154 | |
| 155 | server.should_exit = True |
| 156 | logger.info_once("[shutdown] API server: signalling HTTP server shutdown") |
| 157 | server_task.cancel() |
| 158 | if watchdog_task is not None: |
| 159 | watchdog_task.cancel() |
| 160 | if ssl_cert_refresher: |
| 161 | ssl_cert_refresher.stop() |
| 162 | |
| 163 | shutdown_task = loop.create_task(handle_shutdown()) |
| 164 | |
| 165 | try: |
| 166 | await server_task |
| 167 | return dummy_shutdown() |
| 168 | except asyncio.CancelledError: |
| 169 | port = uvicorn_kwargs["port"] |
| 170 | process = find_process_using_port(port) |
| 171 | if process is not None: |
| 172 | logger.warning( |
| 173 | "port %s is used by process %s launched with command:\n%s", |
| 174 | port, |
| 175 | process, |
| 176 | " ".join(process.cmdline()), |
| 177 | ) |
| 178 | logger.info_once("[shutdown] API server: shutting down FastAPI HTTP server") |
| 179 | return server.shutdown() |
| 180 | finally: |
| 181 | shutdown_task.cancel() |
| 182 | if watchdog_task is not None: |
| 183 | watchdog_task.cancel() |
| 184 | |
| 185 | |
| 186 | async def watchdog_loop(server: uvicorn.Server, engine: EngineClient): |
| 187 | """# Watchdog task that runs in the background, checking |
| 188 | # for error state in the engine. Needed to trigger shutdown |
| 189 | # if an exception arises is StreamingResponse() generator. |
| 190 | """ |
| 191 | VLLM_WATCHDOG_TIME_S = 5.0 |
| 192 | while True: |
| 193 | await asyncio.sleep(VLLM_WATCHDOG_TIME_S) |
| 194 | terminate_if_errored(server, engine) |
| 195 | |
| 196 | |
| 197 | def terminate_if_errored(server: uvicorn.Server, engine: EngineClient): |
| 198 | """See discussions here on shutting down a uvicorn server |
| 199 | https://github.com/encode/uvicorn/discussions/1103 |
| 200 | In this case we cannot await the server shutdown here |
| 201 | because handler must first return to close the connection |
| 202 | for this request. |
| 203 | """ |
| 204 | engine_errored = engine.errored and not engine.is_running |
| 205 | if not envs.VLLM_KEEP_ALIVE_ON_ENGINE_DEATH and engine_errored: |
| 206 | server.should_exit = True |
| 207 | |
| 208 | |
| 209 | def create_server_socket( |
| 210 | addr: tuple[str, int], |
| 211 | *, |
| 212 | reuse_port: bool, |
| 213 | ) -> socket.socket: |
| 214 | family = socket.AF_INET |
| 215 | if is_valid_ipv6_address(addr[0]): |
| 216 | family = socket.AF_INET6 |
| 217 | |
| 218 | sock = socket.socket(family=family, type=socket.SOCK_STREAM) |
| 219 | sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| 220 | if reuse_port: |
| 221 | sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) |
| 222 | sock.bind(addr) |
| 223 | |
| 224 | return sock |
| 225 | |
| 226 | |
| 227 | def create_server_unix_socket(path: str) -> socket.socket: |
| 228 | sock = socket.socket(family=socket.AF_UNIX, type=socket.SOCK_STREAM) |
| 229 | sock.bind(path) |
| 230 | return sock |
| 231 | |
| 232 | |
| 233 | def validate_api_server_args(args): |
| 234 | valid_tool_parses = ToolParserManager.list_registered() |
| 235 | if args.enable_auto_tool_choice and args.tool_call_parser not in valid_tool_parses: |
| 236 | raise KeyError( |
| 237 | f"invalid tool call parser: {args.tool_call_parser} " |
| 238 | f"(chose from {{ {','.join(valid_tool_parses)} }})" |
| 239 | ) |
| 240 | |
| 241 | valid_reasoning_parsers = ReasoningParserManager.list_registered() |
| 242 | if ( |
| 243 | reasoning_parser := args.structured_outputs_config.reasoning_parser |
| 244 | ) and reasoning_parser not in valid_reasoning_parsers: |
| 245 | raise KeyError( |
| 246 | f"invalid reasoning parser: {reasoning_parser} " |
| 247 | f"(chose from {{ {','.join(valid_reasoning_parsers)} }})" |
| 248 | ) |
| 249 | |
| 250 | |
| 251 | @instrument(span_name="API server setup") |
| 252 | def setup_server(args, *, reuse_port: bool): |
| 253 | """Validate API server args and create the server socket.""" |
| 254 | log_version_and_model(logger, VLLM_VERSION, args.model) |
| 255 | log_non_default_args(args) |
| 256 | |
| 257 | if args.tool_parser_plugin and len(args.tool_parser_plugin) > 3: |
| 258 | ToolParserManager.import_tool_parser(args.tool_parser_plugin) |
| 259 | |
| 260 | if args.reasoning_parser_plugin and len(args.reasoning_parser_plugin) > 3: |
| 261 | ReasoningParserManager.import_reasoning_parser(args.reasoning_parser_plugin) |
| 262 | |
| 263 | validate_api_server_args(args) |
| 264 | |
| 265 | # workaround to make sure that we bind the port before the engine is set up. |
| 266 | # This avoids race conditions with ray. |
| 267 | # see https://github.com/vllm-project/vllm/issues/8204 |
| 268 | if args.uds: |
| 269 | sock = create_server_unix_socket(args.uds) |
| 270 | else: |
| 271 | sock_addr = (args.host or "", args.port) |
| 272 | sock = create_server_socket(sock_addr, reuse_port=reuse_port) |
| 273 | |
| 274 | # workaround to avoid footguns where uvicorn drops requests with too |
| 275 | # many concurrent requests active |
| 276 | set_ulimit() |
| 277 | |
| 278 | if args.uds: |
| 279 | listen_address = f"unix:{args.uds}" |
| 280 | else: |
| 281 | addr, port = sock_addr |
| 282 | is_ssl = args.ssl_keyfile and args.ssl_certfile |
| 283 | host_part = f"[{addr}]" if is_valid_ipv6_address(addr) else addr or "0.0.0.0" |
| 284 | listen_address = f"http{'s' if is_ssl else ''}://{host_part}:{port}" |
| 285 | return listen_address, sock |