vllm.entrypoints.launchers.grpc_server

vLLM gRPC Server

Starts a gRPC server backed by AsyncLLM, using the VllmEngineServicer from the smg-grpc-servicer package.

Usage

python -m vllm.entrypoints.launchers.grpc_server --model

Example

python -m vllm.entrypoints.launchers.grpc_server --model meta-llama/Llama-2-7b-hf --host 0.0.0.0 --port 50051

Functions:

• main –

  Main entry point for python -m vllm.entrypoints.launchers.grpc_server.

• serve_grpc –

  Main gRPC serving function.

_InflightRpcs

Bases: ServerInterceptor

Counts the in-flight RPCs of one service, except subscriptions.

An RPC counts from the moment the server dispatches it, before its request message is received, so one whose request is still arriving or being preprocessed (tokenization, media fetch and processing), and is not yet known to AsyncLLM, counts too. It stops counting when the RPC's task finishes, whether or not its handler ran.

Source code in vllm/entrypoints/launchers/grpc_server.py

class _InflightRpcs(grpc.aio.ServerInterceptor):
    """Counts the in-flight RPCs of one service, except subscriptions.

    An RPC counts from the moment the server dispatches it, before its request
    message is received, so one whose request is still arriving or being
    preprocessed (tokenization, media fetch and processing), and is not yet
    known to AsyncLLM, counts too. It stops counting when the RPC's task
    finishes, whether or not its handler ran.
    """

    def __init__(self, service: str):
        self.count = 0
        self._prefix = f"/{service}/"

    async def intercept_service(self, continuation, handler_call_details):
        handler = await continuation(handler_call_details)
        method = handler_call_details.method
        if (
            handler is not None
            and method.startswith(self._prefix)
            and method[len(self._prefix) :] not in _SUBSCRIPTION_METHODS
        ):
            # grpc.aio runs interceptors in the task that serves the RPC.
            self.count += 1
            asyncio.current_task().add_done_callback(self._done)
        return handler

    def _done(self, task: asyncio.Task) -> None:
        self.count -= 1

_drain_and_stop(server, async_llm, shutdown_timeout, inflight) async

Stop the gRPC server and AsyncLLM within --shutdown-timeout.

0 keeps the abort behavior: in-flight RPCs get _RPC_GRACE_S, then EngineCore gets its default join. Otherwise the server refuses new RPCs at once and EngineCore keeps serving until the in-flight RPCs (inflight) and their requests are done; RPCs still running at the deadline (--shutdown-timeout, but no less than _RPC_GRACE_S) are cancelled and abort their requests. EngineCore then drains work that outlives an RPC, such as NIXL prefill blocks the decode side has not read yet, until the same deadline plus _ENGINE_TEARDOWN_S for teardown.

Source code in vllm/entrypoints/launchers/grpc_server.py

async def _drain_and_stop(
    server, async_llm: AsyncLLM, shutdown_timeout: int, inflight: _InflightRpcs
) -> None:
    """Stop the gRPC server and AsyncLLM within ``--shutdown-timeout``.

    0 keeps the abort behavior: in-flight RPCs get ``_RPC_GRACE_S``, then
    EngineCore gets its default join. Otherwise the server refuses new RPCs at
    once and EngineCore keeps serving until the in-flight RPCs (``inflight``)
    and their requests are done; RPCs still running at the deadline
    (``--shutdown-timeout``, but no less than ``_RPC_GRACE_S``) are cancelled
    and abort their requests. EngineCore then drains work that outlives an
    RPC, such as NIXL prefill blocks the decode side has not read yet, until
    the same deadline plus ``_ENGINE_TEARDOWN_S`` for teardown.
    """
    loop = asyncio.get_running_loop()
    if shutdown_timeout == 0:
        await server.stop(grace=_RPC_GRACE_S)
        logger.info("gRPC server stopped")
        await loop.run_in_executor(None, async_llm.shutdown)
        logger.info("AsyncLLM engine stopped")
        return

    grace = max(shutdown_timeout, _RPC_GRACE_S)
    deadline = loop.time() + grace
    logger.info("Draining in-flight requests (shutdown_timeout=%ds)", shutdown_timeout)
    rpcs = asyncio.ensure_future(server.stop(grace=grace))
    await _wait_for_requests(rpcs, async_llm, inflight)
    remaining = max(deadline - loop.time(), 0.0)
    await loop.run_in_executor(
        None, partial(async_llm.shutdown, timeout=remaining + _ENGINE_TEARDOWN_S)
    )
    logger.info("AsyncLLM engine stopped")
    await asyncio.wait([rpcs], timeout=_RPC_FLUSH_S)
    await server.stop(grace=None)
    await rpcs
    logger.info("gRPC server stopped")

_wait_for_requests(rpcs, async_llm, inflight) async

Return once rpcs is done or nothing can still send EngineCore a request.

EngineCore rejects every request it reads once its shutdown has started, even one an RPC forwarded before SIGTERM, so it must keep running while an RPC other than a subscription is in flight or AsyncLLM still has a request. Idle has to hold on two polls in a row, which covers an RPC that gRPC accepted before SIGTERM but has not yet handed to its task.

Source code in vllm/entrypoints/launchers/grpc_server.py

async def _wait_for_requests(
    rpcs: asyncio.Future, async_llm: AsyncLLM, inflight: _InflightRpcs
) -> None:
    """Return once ``rpcs`` is done or nothing can still send EngineCore a request.

    EngineCore rejects every request it reads once its shutdown has started,
    even one an RPC forwarded before SIGTERM, so it must keep running while
    an RPC other than a subscription is in flight or AsyncLLM still has a
    request. Idle has to hold on two polls in a row, which covers an RPC that
    gRPC accepted before SIGTERM but has not yet handed to its task.
    """
    idle = False
    while not rpcs.done():
        was_idle = idle
        idle = inflight.count == 0 and async_llm.get_num_unfinished_requests() == 0
        if was_idle and idle:
            return
        await asyncio.wait([rpcs], timeout=_IDLE_POLL_S)

main()

Main entry point for python -m vllm.entrypoints.launchers.grpc_server.

Source code in vllm/entrypoints/launchers/grpc_server.py

def main():
    """Main entry point for python -m vllm.entrypoints.launchers.grpc_server."""
    parser = FlexibleArgumentParser(
        description="vLLM gRPC Server",
    )

    # Server args
    parser.add_argument(
        "--host",
        type=str,
        default="0.0.0.0",
        help="Host to bind gRPC server to",
    )
    parser.add_argument(
        "--port",
        type=int,
        default=50051,
        help="Port to bind gRPC server to",
    )
    parser = AsyncEngineArgs.add_cli_args(parser)

    args = parser.parse_args()
    configure_logging_from_args(args)

    # Run server
    try:
        uvloop.run(serve_grpc(args))
    except Exception as e:
        logger.exception("Server failed: %s", e)
        sys.exit(1)

serve_grpc(args) async

Main gRPC serving function.

Parameters:

• args

  (Namespace) –

  Parsed command line arguments

Source code in vllm/entrypoints/launchers/grpc_server.py

async def serve_grpc(args: argparse.Namespace):
    """Main gRPC serving function.

    Args:
        args: Parsed command line arguments

    """
    log_version_and_model(logger, VLLM_VERSION, args.model)
    logger.info("vLLM gRPC server args: %s", args)

    start_time = time.time()

    # Create engine args
    engine_args = AsyncEngineArgs.from_cli_args(args)

    # Build vLLM config
    vllm_config = engine_args.create_engine_config(
        usage_context=UsageContext.OPENAI_API_SERVER,
    )

    # Create AsyncLLM
    async_llm = AsyncLLM.from_vllm_config(
        vllm_config=vllm_config,
        usage_context=UsageContext.OPENAI_API_SERVER,
        enable_log_requests=args.enable_log_requests,
        disable_log_stats=args.disable_log_stats,
    )

    # Create servicer
    servicer = VllmEngineServicer(async_llm, start_time)

    # Create gRPC server
    inflight = _InflightRpcs(
        vllm_engine_pb2.DESCRIPTOR.services_by_name["VllmEngine"].full_name
    )
    server = grpc.aio.server(
        interceptors=[inflight],
        options=[
            ("grpc.max_send_message_length", -1),
            ("grpc.max_receive_message_length", -1),
            # A non-streaming Generate writes nothing while the engine
            # decodes, so client keepalive and BDP-probe PINGs arriving in
            # that window are strikes and eventually earn a
            # GOAWAY(ENHANCE_YOUR_CALM, "too_many_pings") that fails every
            # in-flight RPC. Disable the strike count to tolerate them.
            ("grpc.http2.max_ping_strikes", 0),
            # GRPC_ARG_HTTP2_MIN_RECV_PING_INTERVAL_WITHOUT_DATA_MS expands to
            # a string with no "recv", and gRPC Core silently ignores unknown
            # keys, so the misspelt form left the 300s default in force.
            ("grpc.http2.min_ping_interval_without_data_ms", 10000),
            # Without this, gRPC Core widens the floor to a fixed two hours
            # once no streams are open, striking keepalives between requests.
            ("grpc.keepalive_permit_without_calls", True),
        ],
    )

    # Add servicer to server
    vllm_engine_pb2_grpc.add_VllmEngineServicer_to_server(servicer, server)

    # Add standard gRPC health service for Kubernetes probes
    health_servicer = VllmHealthServicer(async_llm)
    health_pb2_grpc.add_HealthServicer_to_server(health_servicer, server)

    # Enable reflection for grpcurl and other tools
    service_names = (
        vllm_engine_pb2.DESCRIPTOR.services_by_name["VllmEngine"].full_name,
        "grpc.health.v1.Health",
        reflection.SERVICE_NAME,
    )
    reflection.enable_server_reflection(service_names, server)

    # Bind to address
    host = args.host or "0.0.0.0"
    address = f"{host}:{args.port}"
    server.add_insecure_port(address)

    try:
        # Start server
        await server.start()
        logger.info("vLLM gRPC server started on %s", address)
        logger.info("Server is ready to accept requests")

        # Start periodic stats logging (mirrors the HTTP server's lifespan task)
        if not args.disable_log_stats:

            async def _force_log():
                while True:
                    await asyncio.sleep(envs.VLLM_LOG_STATS_INTERVAL)
                    await async_llm.do_log_stats()

            stats_task = asyncio.create_task(_force_log())
        else:
            stats_task = None

        # Handle shutdown signals
        loop = asyncio.get_running_loop()
        stop_event = asyncio.Event()
        force_stops = []

        def signal_handler():
            if stop_event.is_set():
                # A second signal skips the drain: cancelling the RPCs aborts
                # their requests, so EngineCore runs out of work and exits.
                logger.info("Received another shutdown signal; cancelling RPCs")
                force_stops.append(loop.create_task(server.stop(grace=None)))
                return
            logger.info("Received shutdown signal")
            stop_event.set()

        for sig in (signal.SIGTERM, signal.SIGINT):
            loop.add_signal_handler(sig, signal_handler)

        try:
            await stop_event.wait()
        except KeyboardInterrupt:
            logger.info("Interrupted by user")
    finally:
        logger.info("Shutting down vLLM gRPC server...")
        if stats_task is not None:
            stats_task.cancel()
        try:
            health_servicer.set_not_serving()
        except Exception:  # broad: must not prevent server.stop() / shutdown()
            logger.warning("Failed to set health status to NOT_SERVING", exc_info=True)
        await _drain_and_stop(server, async_llm, vllm_config.shutdown_timeout, inflight)
        logger.info("Shutdown complete")