vllm.entrypoints.openai.api_server ¶

def _startup_prefetch_weights(vllm_config: "VllmConfig") -> None:
    """Kick off reading model weight shards into OS page cache from the
    parent APIServer. EngineCore will read the same files a few seconds
    later from the child; by then the kernel already has them ready.

    All work (directory resolution, HF/ModelScope cache lookup, globbing,
    and the reads themselves) runs inside the background thread so we do
    not block the asyncio event loop.

    Best-effort: any failure (unknown model location, permission, etc.) is
    swallowed — vLLM's existing in-child prefetch then runs normally.
    """
    import threading

    # Capture only the small scalar fields the thread needs. Avoid holding
    # a reference to vllm_config (which contains unpicklable objects) for
    # longer than necessary.
    model_ref = vllm_config.model_config.model
    revision = vllm_config.model_config.revision
    download_dir = vllm_config.load_config.download_dir

    def _prefetch_worker() -> None:
        import glob
        import os

        from vllm import envs

        candidate_dir: str | None = None

        # 1. Local path?
        if os.path.isdir(model_ref):
            candidate_dir = model_ref
        else:
            # 2. HF / ModelScope repo id — resolve to the local cache
            # snapshot dir using the same revision / cache_dir the weight
            # loader will use, so we prefetch the right files.
            try:
                if envs.VLLM_USE_MODELSCOPE:
                    from modelscope.hub.snapshot_download import (
                        snapshot_download,
                    )

                    candidate_dir = snapshot_download(
                        model_id=model_ref,
                        revision=revision,
                        cache_dir=download_dir,
                        local_files_only=True,
                    )
                else:
                    from huggingface_hub import snapshot_download

                    candidate_dir = snapshot_download(
                        repo_id=model_ref,
                        revision=revision,
                        cache_dir=download_dir,
                        allow_patterns=[
                            "*.safetensors",
                            "*.bin",
                            "*.json",
                            "*tokenizer*",
                        ],
                        local_files_only=True,
                    )
            except Exception:
                return  # not cached yet or not a known repo id

        if not candidate_dir or not os.path.isdir(candidate_dir):
            return

        # Weight shards: large files, read into page cache.
        shard_paths = sorted(
            glob.glob(os.path.join(candidate_dir, "*.safetensors"))
            + glob.glob(os.path.join(candidate_dir, "*.bin"))
        )
        # Tokenizer/config sidecars: small, but re-opened in the child and
        # add synchronous open+read latency when the disk is cold.
        sidecar_paths = sorted(
            glob.glob(os.path.join(candidate_dir, "*.json"))
            + glob.glob(os.path.join(candidate_dir, "tokenizer.model"))
            + glob.glob(os.path.join(candidate_dir, "*tokenizer*"))
        )
        shard_paths.extend(sidecar_paths)
        if not shard_paths:
            return

        logger.debug(
            "Parent-side weight prefetch starting for %d files in %s",
            len(shard_paths),
            candidate_dir,
        )

        # Match vLLM's in-child prefetch block size + thread count.
        block_size = 16 * 1024 * 1024  # 16 MB
        # Read shards in parallel across 8 worker threads (bounded) to
        # saturate multi-spindle / multi-queue storage without thrashing.
        from concurrent.futures import ThreadPoolExecutor

        def read_one(p: str) -> None:
            try:
                with open(p, "rb") as f:
                    while f.read(block_size):
                        pass
            except Exception:
                pass

        with ThreadPoolExecutor(max_workers=8) as pool:
            list(pool.map(read_one, shard_paths))

    threading.Thread(
        target=_prefetch_worker,
        daemon=True,
        name="vllm-parent-weight-prefetch",
    ).start()

async def build_and_serve(
    engine_client: EngineClient,
    listen_address: str,
    sock: socket.socket,
    args: Namespace,
    **uvicorn_kwargs,
) -> asyncio.Task:
    """Build FastAPI app, initialize state, and start serving.

    Returns the shutdown task for the caller to await.
    """

    # Get uvicorn log config (from file or with endpoint filter)
    log_config = get_uvicorn_log_config(args)
    if log_config is not None:
        uvicorn_kwargs["log_config"] = log_config

    supported_tasks = await engine_client.get_supported_tasks()
    model_config = engine_client.model_config

    logger.info("Supported tasks: %s", supported_tasks)
    app = build_app(args, supported_tasks, model_config)
    await init_app_state(engine_client, app.state, args, supported_tasks)

    logger.info("Starting vLLM server on %s", listen_address)

    return await serve_http(
        app,
        sock=sock,
        enable_ssl_refresh=args.enable_ssl_refresh,
        host=args.host,
        port=args.port,
        log_level=args.uvicorn_log_level,
        # NOTE: When the 'disable_uvicorn_access_log' value is True,
        # no access log will be output.
        access_log=not args.disable_uvicorn_access_log,
        timeout_keep_alive=envs.VLLM_HTTP_TIMEOUT_KEEP_ALIVE,
        ssl_keyfile=args.ssl_keyfile,
        ssl_certfile=args.ssl_certfile,
        ssl_ca_certs=args.ssl_ca_certs,
        ssl_cert_reqs=args.ssl_cert_reqs,
        ssl_ciphers=args.ssl_ciphers,
        h11_max_incomplete_event_size=args.h11_max_incomplete_event_size,
        h11_max_header_count=args.h11_max_header_count,
        **uvicorn_kwargs,
    )

async def build_and_serve_renderer(
    vllm_config: VllmConfig,
    listen_address: str,
    sock: socket.socket,
    args: Namespace,
    **uvicorn_kwargs,
) -> asyncio.Task:
    """Build FastAPI app for a CPU-only render server, initialize state, and
    start serving.

    Returns the shutdown task for the caller to await.
    """

    # Get uvicorn log config (from file or with endpoint filter)
    log_config = get_uvicorn_log_config(args)
    if log_config is not None:
        uvicorn_kwargs["log_config"] = log_config

    app = build_app(args, ("render",))
    await init_render_app_state(vllm_config, app.state, args)

    logger.info("Starting vLLM server on %s", listen_address)

    return await serve_http(
        app,
        sock=sock,
        enable_ssl_refresh=args.enable_ssl_refresh,
        host=args.host,
        port=args.port,
        log_level=args.uvicorn_log_level,
        # NOTE: When the 'disable_uvicorn_access_log' value is True,
        # no access log will be output.
        access_log=not args.disable_uvicorn_access_log,
        timeout_keep_alive=envs.VLLM_HTTP_TIMEOUT_KEEP_ALIVE,
        ssl_keyfile=args.ssl_keyfile,
        ssl_certfile=args.ssl_certfile,
        ssl_ca_certs=args.ssl_ca_certs,
        ssl_cert_reqs=args.ssl_cert_reqs,
        ssl_ciphers=args.ssl_ciphers,
        h11_max_incomplete_event_size=args.h11_max_incomplete_event_size,
        h11_max_header_count=args.h11_max_header_count,
        **uvicorn_kwargs,
    )

@asynccontextmanager
async def build_async_engine_client_from_engine_args(
    engine_args: AsyncEngineArgs,
    *,
    usage_context: UsageContext = UsageContext.OPENAI_API_SERVER,
    client_config: dict[str, Any] | None = None,
) -> AsyncIterator[EngineClient]:
    """
    Create EngineClient, either:
        - in-process using the AsyncLLMEngine Directly
        - multiprocess using AsyncLLMEngine RPC

    Returns the Client or None if the creation failed.
    """

    # Create the EngineConfig (determines if we can use V1).
    vllm_config = engine_args.create_engine_config(usage_context=usage_context)

    # [startup] Start prefetching model weight shards into the OS page cache
    # in a background thread from the PARENT APIServer process. EngineCore
    # will page-fault on these same files ~10-15 s later (after fork + CUDA
    # context + distributed init + model init). For large-weight cases
    # (tens of GB) this parent-side head start meaningfully shrinks the
    # prefetch+load phase that the engine's in-child prefetch otherwise
    # barely overlaps.
    #
    # Skip in API-only workers that connect to an already-running EngineCore
    # (multi-API-server / disaggregated setups): those processes never load
    # weights, and if we prefetched from all of them we'd contend with the
    # engine's own read. Presence of an `input_address` in client_config is
    # the current marker that this worker is headless.
    #
    # Best-effort: if the model is a local path, glob for safetensors; if
    # it's a repo-id, try to resolve via HF hub (or ModelScope) local cache.
    # Any failure silently falls through to the existing in-child prefetch
    # path. All I/O (incl. directory resolution) runs inside the BG thread
    # so the asyncio event loop is never blocked.
    if not (client_config and client_config.get("input_address")):
        _startup_prefetch_weights(vllm_config)

    from vllm.v1.engine.async_llm import AsyncLLM

    async_llm: AsyncLLM | None = None

    # Don't mutate the input client_config
    client_config = dict(client_config) if client_config else {}
    client_count = client_config.pop("client_count", 1)
    client_index = client_config.pop("client_index", 0)

    try:
        async_llm = AsyncLLM.from_vllm_config(
            vllm_config=vllm_config,
            usage_context=usage_context,
            enable_log_requests=engine_args.enable_log_requests,
            aggregate_engine_logging=engine_args.aggregate_engine_logging,
            disable_log_stats=engine_args.disable_log_stats,
            client_addresses=client_config,
            client_count=client_count,
            client_index=client_index,
        )

        # Don't keep the dummy data in memory
        assert async_llm is not None
        await async_llm.reset_mm_cache()

        yield async_llm
    finally:
        if async_llm:
            async_llm.shutdown()

async def init_render_app_state(
    vllm_config: VllmConfig,
    state: State,
    args: Namespace,
) -> None:
    """Initialise FastAPI app state for a CPU-only render server.

    Unlike :func:`init_app_state` this function does not require an
    :class:`~vllm.engine.protocol.EngineClient`; it bootstraps the
    preprocessing pipeline (renderer, input_processor)
    directly from the :class:`~vllm.config.VllmConfig`.
    """
    from vllm.entrypoints.chat_utils import load_chat_template
    from vllm.entrypoints.openai.models.serving import OpenAIModelRegistry
    from vllm.entrypoints.serve.render.serving import OpenAIServingRender
    from vllm.renderers import renderer_from_config

    served_model_names = args.served_model_name or [args.model]
    model_registry = OpenAIModelRegistry(
        model_config=vllm_config.model_config,
        base_model_paths=[
            BaseModelPath(name=name, model_path=args.model)
            for name in served_model_names
        ],
    )

    if args.enable_log_requests:
        request_logger = RequestLogger(max_log_len=args.max_log_len)
    else:
        request_logger = None

    renderer = renderer_from_config(vllm_config)
    resolved_chat_template = load_chat_template(args.chat_template)

    state.openai_serving_render = OpenAIServingRender(
        model_config=vllm_config.model_config,
        renderer=renderer,
        model_registry=model_registry,
        request_logger=request_logger,
        chat_template=resolved_chat_template,
        chat_template_content_format=args.chat_template_content_format,
        trust_request_chat_template=args.trust_request_chat_template,
        enable_auto_tools=args.enable_auto_tool_choice,
        exclude_tools_when_tool_choice_none=args.exclude_tools_when_tool_choice_none,
        tool_parser=args.tool_call_parser,
        reasoning_parser=args.structured_outputs_config.reasoning_parser,
        default_chat_template_kwargs=args.default_chat_template_kwargs,
        log_error_stack=args.log_error_stack,
    )

    state.openai_serving_models = model_registry

    # Expose tokenization via the render handler (no engine required).
    state.openai_serving_tokenization = state.openai_serving_render

    state.vllm_config = vllm_config
    # Disable stats logging — there is no engine to poll.
    state.log_stats = False
    state.engine_client = None
    state.args = args
    state.enable_server_load_tracking = False
    state.server_load_metrics = 0

async def run_server(args, **uvicorn_kwargs) -> None:
    """Run a single-worker API server."""

    # Add process-specific prefix to stdout and stderr.
    decorate_logs("APIServer")

    listen_address, sock = setup_server(args)
    await run_server_worker(listen_address, sock, args, **uvicorn_kwargs)

async def run_server_worker(
    listen_address, sock, args, client_config=None, **uvicorn_kwargs
) -> None:
    """Run a single API server worker."""

    if args.tool_parser_plugin and len(args.tool_parser_plugin) > 3:
        ToolParserManager.import_tool_parser(args.tool_parser_plugin)

    if args.reasoning_parser_plugin and len(args.reasoning_parser_plugin) > 3:
        ReasoningParserManager.import_reasoning_parser(args.reasoning_parser_plugin)

    async with build_async_engine_client(
        args,
        client_config=client_config,
    ) as engine_client:
        shutdown_task = await build_and_serve(
            engine_client, listen_address, sock, args, **uvicorn_kwargs
        )
    # NB: Await server shutdown only after the backend context is exited
    try:
        await shutdown_task
    finally:
        sock.close()

@instrument(span_name="API server setup")
def setup_server(args):
    """Validate API server args, set up signal handler, create socket
    ready to serve."""

    log_version_and_model(logger, VLLM_VERSION, args.model)
    log_non_default_args(args)

    if args.tool_parser_plugin and len(args.tool_parser_plugin) > 3:
        ToolParserManager.import_tool_parser(args.tool_parser_plugin)

    if args.reasoning_parser_plugin and len(args.reasoning_parser_plugin) > 3:
        ReasoningParserManager.import_reasoning_parser(args.reasoning_parser_plugin)

    validate_api_server_args(args)

    # workaround to make sure that we bind the port before the engine is set up.
    # This avoids race conditions with ray.
    # see https://github.com/vllm-project/vllm/issues/8204
    if args.uds:
        sock = create_server_unix_socket(args.uds)
    else:
        sock_addr = (args.host or "", args.port)
        sock = create_server_socket(sock_addr)

    # workaround to avoid footguns where uvicorn drops requests with too
    # many concurrent requests active
    set_ulimit()

    def signal_handler(*_) -> None:
        # Interrupt server on sigterm while initializing
        raise KeyboardInterrupt("terminated")

    signal.signal(signal.SIGTERM, signal_handler)

    if args.uds:
        listen_address = f"unix:{args.uds}"
    else:
        addr, port = sock_addr
        is_ssl = args.ssl_keyfile and args.ssl_certfile
        host_part = f"[{addr}]" if is_valid_ipv6_address(addr) else addr or "0.0.0.0"
        listen_address = f"http{'s' if is_ssl else ''}://{host_part}:{port}"
    return listen_address, sock