diff --git a/rust/README.md b/rust/README.md index 140b027b02..42d6c3e16a 100644 --- a/rust/README.md +++ b/rust/README.md @@ -43,6 +43,27 @@ For example: VLLM_USE_RUST_FRONTEND=1 vllm serve Qwen/Qwen3-0.6B ``` +To also expose the Rust frontend's gRPC inference and control services, add +`--grpc-port`. The HTTP listener remains on `--port`: + +```bash +VLLM_USE_RUST_FRONTEND=1 vllm serve Qwen/Qwen3-0.6B \ + --host 127.0.0.1 --port 8000 --grpc-port 50051 +``` + +The gRPC listener binds on `--host` like HTTP, or on `127.0.0.1` when HTTP uses +`--uds`. When HTTP uses TCP, its port and the gRPC port must differ unless +requesting an automatically assigned port with `0`. + +This also works with Python-supervised hybrid data parallelism. It requires a +frontend process and cannot be combined with `--headless` or +`--data-parallel-multi-port-external-lb`. + +The separate `vllm serve MODEL --grpc --port 50051` command launches the Python +SMG `VllmEngine` gRPC server in place of HTTP. The Rust listener exposes the +`vllm.Inference` and `vllm.Control` services alongside HTTP. These are different +gRPC APIs; `--grpc` and `--grpc-port` are mutually exclusive. + ### RL weight synchronization With `VLLM_SERVER_DEV_MODE=1`, the Rust frontend supports the HTTP weight-transfer diff --git a/rust/src/cmd/src/cli.rs b/rust/src/cmd/src/cli.rs index 1fab1c93d1..1269e0b8af 100644 --- a/rust/src/cmd/src/cli.rs +++ b/rust/src/cmd/src/cli.rs @@ -268,11 +268,6 @@ pub struct SharedRuntimeArgs { #[arg(long, value_parser = clap::value_parser!(i32).range(-1..), allow_negative_numbers = true)] #[serde(default)] pub max_logprobs: Option, - /// TCP port for the gRPC Inference service. When not set, no gRPC server is - /// started. - #[arg(long)] - #[serde(default)] - pub grpc_port: Option, /// Maximum time to wait for active requests to drain during shutdown. #[arg(long, default_value_t = 0)] #[serde(default)] @@ -505,6 +500,7 @@ impl SharedRuntimeArgs { fn into_bootstrapped_config( self, listen_fd: i32, + grpc_listen_fd: Option, input_address: String, output_address: String, coordinator_address: Option, @@ -539,6 +535,7 @@ impl SharedRuntimeArgs { generation_config: self.generation_config, served_model_name: self.served_model_name, listener_mode: HttpListenerMode::InheritedFd { fd: listen_fd }, + grpc_listener_mode: grpc_listen_fd.map(|fd| HttpListenerMode::InheritedFd { fd }), tool_call_parser: self.tool_call_parser, reasoning_parser: self.reasoning_parser, tool_strict_level: self.tool_strict_level, @@ -555,7 +552,6 @@ impl SharedRuntimeArgs { tls, api_keys: self.api_key, disable_log_stats: self.disable_log_stats, - grpc_port: self.grpc_port, shutdown_timeout, // The engine is launched and supervised by another process. manages_engine: false, @@ -569,6 +565,7 @@ impl SharedRuntimeArgs { fn into_managed_config( self, listener_mode: HttpListenerMode, + grpc_listener_mode: Option, handshake_address: String, advertised_host: String, engine_count: usize, @@ -600,6 +597,7 @@ impl SharedRuntimeArgs { generation_config: self.generation_config, served_model_name: self.served_model_name, listener_mode, + grpc_listener_mode, tool_call_parser: self.tool_call_parser, reasoning_parser: self.reasoning_parser, tool_strict_level: self.tool_strict_level, @@ -616,7 +614,6 @@ impl SharedRuntimeArgs { tls, api_keys: self.api_key, disable_log_stats: self.disable_log_stats, - grpc_port: self.grpc_port, shutdown_timeout, manages_engine, keep_alive_timeout, @@ -696,6 +693,10 @@ pub struct FrontendArgs { /// supervisor. #[arg(long)] pub listen_fd: i32, + /// Inherited gRPC listening socket file descriptor passed by the Python + /// supervisor. When not set, no gRPC server is started. + #[arg(long)] + pub grpc_listen_fd: Option, /// Frontend input ROUTER socket address that the Python engines will /// connect to. #[arg(long)] @@ -731,6 +732,7 @@ impl FrontendArgs { let data_parallel_size = self.data_parallel_size.unwrap_or(self.engine_count); self.runtime.into_bootstrapped_config( self.listen_fd, + self.grpc_listen_fd, self.input_address, self.output_address, self.coordinator_address, @@ -760,6 +762,10 @@ pub struct ServeArgs { /// Unix domain socket path. If set, host and port arguments are ignored. #[arg(long)] pub uds: Option, + /// TCP port for the gRPC Inference and Control services. When not set, no + /// gRPC server is started. + #[arg(long)] + pub grpc_port: Option, /// Flag to print debug information about CLI argument parsing and exit. #[educe(Debug(ignore))] @@ -789,9 +795,20 @@ impl ServeArgs { port: self.port, }, }; + // gRPC follows the HTTP TCP host. With a Unix socket it defaults to IPv4 + // loopback rather than all interfaces, so the side-car is never + // accidentally network-exposed. + let grpc_listener_mode = self.grpc_port.map(|port| HttpListenerMode::BindTcp { + host: match self.uds { + Some(_) => "127.0.0.1".to_string(), + None => self.host.clone(), + }, + port, + }); self.runtime.clone().into_managed_config( listener_mode, + grpc_listener_mode, handshake_address, self.managed_engine.handshake_host.clone(), self.managed_engine.data_parallel_size, diff --git a/rust/src/cmd/src/cli/tests.rs b/rust/src/cmd/src/cli/tests.rs index 0802d2062a..944febc36c 100644 --- a/rust/src/cmd/src/cli/tests.rs +++ b/rust/src/cmd/src/cli/tests.rs @@ -174,6 +174,7 @@ fn serve_args_forward_python_flags_with_separator() { host: "127.0.0.1", port: 8000, uds: None, + grpc_port: None, runtime: SharedRuntimeArgs { model: "Qwen/Qwen3-0.6B", revision: None, @@ -188,7 +189,6 @@ fn serve_args_forward_python_flags_with_separator() { renderer: Auto, language_model_only: false, max_logprobs: None, - grpc_port: None, shutdown_timeout: 0, http_timeout_keep_alive: None, chat_template: None, @@ -1007,6 +1007,7 @@ fn frontend_args_accept_json() { command: Frontend( FrontendArgs { listen_fd: 3, + grpc_listen_fd: None, input_address: "ipc:///tmp/input.sock", output_address: "ipc:///tmp/output.sock", coordinator_address: Some( @@ -1029,7 +1030,6 @@ fn frontend_args_accept_json() { renderer: Auto, language_model_only: false, max_logprobs: None, - grpc_port: None, shutdown_timeout: 0, http_timeout_keep_alive: None, chat_template: None, @@ -1687,6 +1687,7 @@ fn serve_args_accept_handshake_aliases() { host: "127.0.0.1", port: 8000, uds: None, + grpc_port: None, runtime: SharedRuntimeArgs { model: "Qwen/Qwen3-0.6B", revision: None, @@ -1701,7 +1702,6 @@ fn serve_args_accept_handshake_aliases() { renderer: Auto, language_model_only: false, max_logprobs: None, - grpc_port: None, shutdown_timeout: 0, http_timeout_keep_alive: None, chat_template: None, @@ -1854,6 +1854,7 @@ fn serve_frontend_config_uses_dp_address_as_advertised_host() { host: "127.0.0.1", port: 8000, }, + grpc_listener_mode: None, tool_call_parser: Auto, reasoning_parser: Auto, tool_strict_level: Auto, @@ -1887,7 +1888,6 @@ fn serve_frontend_config_uses_dp_address_as_advertised_host() { tls: None, api_keys: [], disable_log_stats: false, - grpc_port: None, shutdown_timeout: 0ns, manages_engine: true, keep_alive_timeout: 5s, @@ -1949,6 +1949,7 @@ fn serve_frontend_config_keeps_tcp_transport_for_non_local_only_topology() { host: "127.0.0.1", port: 8000, }, + grpc_listener_mode: None, tool_call_parser: Auto, reasoning_parser: Auto, tool_strict_level: Auto, @@ -1982,7 +1983,6 @@ fn serve_frontend_config_keeps_tcp_transport_for_non_local_only_topology() { tls: None, api_keys: [], disable_log_stats: false, - grpc_port: None, shutdown_timeout: 0ns, manages_engine: true, keep_alive_timeout: 5s, @@ -2088,6 +2088,7 @@ fn frontend_config_uses_external_coordinator_when_coordinator_address_is_present listener_mode: InheritedFd { fd: 3, }, + grpc_listener_mode: None, tool_call_parser: None, reasoning_parser: None, tool_strict_level: Auto, @@ -2121,7 +2122,6 @@ fn frontend_config_uses_external_coordinator_when_coordinator_address_is_present tls: None, api_keys: [], disable_log_stats: false, - grpc_port: None, shutdown_timeout: 0ns, manages_engine: false, keep_alive_timeout: 5s, @@ -2139,6 +2139,8 @@ fn serve_frontend_config_uses_unix_listener_when_uds_is_present() { "Qwen/Qwen3-0.6B", "--uds", "/tmp/vllm.sock", + "--grpc-port", + "50051", ]) .unwrap(); @@ -2153,6 +2155,69 @@ fn serve_frontend_config_uses_unix_listener_when_uds_is_present() { path: "/tmp/vllm.sock".to_string(), } ); + assert_eq!( + config.grpc_listener_mode, + Some(HttpListenerMode::BindTcp { + host: "127.0.0.1".to_string(), + port: 50051, + }) + ); +} + +#[test] +fn serve_frontend_config_binds_grpc_on_http_host() { + let cli = Cli::try_parse_from([ + "vllm-rs", + "serve", + "Qwen/Qwen3-0.6B", + "--host", + "0.0.0.0", + "--grpc-port", + "50051", + ]) + .unwrap(); + + let Command::Serve(args) = cli.command else { + panic!("expected serve args"); + }; + let config = args.to_frontend_config("tcp://127.0.0.1:29550".to_string()); + + assert_eq!( + config.grpc_listener_mode, + Some(HttpListenerMode::BindTcp { + host: "0.0.0.0".to_string(), + port: 50051, + }) + ); +} + +#[test] +fn frontend_config_adopts_inherited_grpc_listener() { + let cli = Cli::try_parse_from([ + "vllm-rs", + "frontend", + "--listen-fd", + "3", + "--grpc-listen-fd", + "4", + "--input-address", + "ipc:///tmp/input.sock", + "--output-address", + "ipc:///tmp/output.sock", + "--args-json", + r#"{"model_tag":"Qwen/Qwen3-0.6B","grpc_port":50051}"#, + ]) + .unwrap(); + + let Command::Frontend(args) = cli.command else { + panic!("expected frontend args"); + }; + let config = args.into_config(); + + assert_eq!( + config.grpc_listener_mode, + Some(HttpListenerMode::InheritedFd { fd: 4 }) + ); } #[test] diff --git a/rust/src/server/examples/external_engine_openai_qwen.rs b/rust/src/server/examples/external_engine_openai_qwen.rs index cb5fed4437..43192ab78f 100644 --- a/rust/src/server/examples/external_engine_openai_qwen.rs +++ b/rust/src/server/examples/external_engine_openai_qwen.rs @@ -69,6 +69,7 @@ async fn main() -> Result<()> { host: "127.0.0.1".to_string(), port, }, + grpc_listener_mode: None, tool_call_parser: ParserSelection::Auto, reasoning_parser: ParserSelection::Auto, tool_strict_level: ToolStrictLevel::Auto, @@ -85,7 +86,6 @@ async fn main() -> Result<()> { tls: None, api_keys: Vec::new(), disable_log_stats: false, - grpc_port: None, shutdown_timeout: Duration::ZERO, manages_engine: false, keep_alive_timeout: Duration::from_secs(5), diff --git a/rust/src/server/src/config.rs b/rust/src/server/src/config.rs index b307f8b3ac..8e51f1440b 100644 --- a/rust/src/server/src/config.rs +++ b/rust/src/server/src/config.rs @@ -23,7 +23,7 @@ use vllm_text::backend::hf::HfOverrides; /// when keep-alive is disabled (`0`). pub const DEFAULT_KEEP_ALIVE_TIMEOUT: Duration = Duration::from_secs(5); -/// How the HTTP server obtains its listening socket. +/// How the HTTP or gRPC server obtains its listening socket. #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub enum HttpListenerMode { /// Bind a fresh TCP listener on the given host/port. @@ -229,6 +229,8 @@ pub struct Config { pub served_model_name: Vec, /// HTTP listener setup. pub listener_mode: HttpListenerMode, + /// gRPC listener setup. When `None`, no gRPC server is started. + pub grpc_listener_mode: Option, /// Tool-call parser selection. pub tool_call_parser: ParserSelection, /// Reasoning parser selection. @@ -272,9 +274,6 @@ pub struct Config { /// engine-reported scheduler stats and request lifecycle events are not /// exported. pub disable_log_stats: bool, - /// TCP port for the gRPC Inference service. When `None`, no gRPC server is - /// started. - pub grpc_port: Option, /// Maximum time to wait for active HTTP/gRPC requests to drain on shutdown. pub shutdown_timeout: Duration, /// Whether the caller manages the engine process and shuts it down when diff --git a/rust/src/server/src/lib.rs b/rust/src/server/src/lib.rs index 6a36491fd9..f54f8ce3e4 100644 --- a/rust/src/server/src/lib.rs +++ b/rust/src/server/src/lib.rs @@ -37,7 +37,6 @@ use hyper_util::rt::{TokioIo, TokioTimer}; use hyper_util::server::graceful::GracefulShutdown; use hyper_util::service::TowerToHyperService; pub use render::{RenderConfig, serve_render}; -use tokio::net::TcpListener; use tokio::time::{Instant, sleep_until}; use tokio_util::sync::CancellationToken; use tonic::transport::Server as TonicServer; @@ -74,17 +73,6 @@ fn effective_served_model_names(model: &str, served_model_name: &[String]) -> Ve } } -/// Choose the gRPC listener host. It follows the HTTP TCP host when there is -/// one; otherwise (unix socket or inherited fd) it defaults to IPv4 loopback -/// rather than all interfaces, so the side-car is never accidentally -/// network-exposed. -fn grpc_bind_host(listener_mode: &HttpListenerMode) -> &str { - match listener_mode { - HttpListenerMode::BindTcp { host, .. } => host.as_str(), - HttpListenerMode::BindUnix { .. } | HttpListenerMode::InheritedFd { .. } => "127.0.0.1", - } -} - /// Build the shared application state for one configured model and one engine /// client, including any static LoRA adapters from `--lora-modules`. async fn build_state(config: &Config) -> Result> { @@ -229,13 +217,11 @@ where // Optionally bind the gRPC Inference server on a separate port. Bind // synchronously here so bind errors (port in use, permission denied, ...) // surface before serving rather than being deferred until shutdown. - let grpc_setup = if let Some(grpc_port) = config.grpc_port { - let grpc_host = grpc_bind_host(&config.listener_mode); - let grpc_listener = TcpListener::bind((grpc_host, grpc_port)) + let grpc_setup = if let Some(grpc_listener_mode) = &config.grpc_listener_mode { + let grpc_listener = Listener::bind(grpc_listener_mode) .await - .with_context(|| format!("failed to bind gRPC listener on {grpc_host}:{grpc_port}"))?; - let addr = grpc_listener.local_addr()?; - let grpc_listener = Listener::Tcp(grpc_listener); + .with_context(|| format!("failed to bind gRPC listener for {grpc_listener_mode:?}"))?; + let addr = grpc_listener.local_addr_display()?; // gRPC reuses the HTTP TLS config (same SslContext) plus ALPN h2. let grpc_tls = config .tls @@ -479,23 +465,4 @@ mod tests { served_names ); } - - #[test] - fn grpc_bind_host_follows_http_tcp_host() { - let mode = HttpListenerMode::BindTcp { - host: "0.0.0.0".to_string(), - port: 8000, - }; - assert_eq!(grpc_bind_host(&mode), "0.0.0.0"); - } - - #[test] - fn grpc_bind_host_defaults_to_loopback_without_tcp_host() { - let unix = HttpListenerMode::BindUnix { - path: "/tmp/vllm.sock".to_string(), - }; - let inherited = HttpListenerMode::InheritedFd { fd: 3 }; - assert_eq!(grpc_bind_host(&unix), "127.0.0.1"); - assert_eq!(grpc_bind_host(&inherited), "127.0.0.1"); - } } diff --git a/tests/entrypoints/launchers/api_server/test_api_server_process_manager.py b/tests/entrypoints/launchers/api_server/test_api_server_process_manager.py index 10858b6da4..6c579df838 100644 --- a/tests/entrypoints/launchers/api_server/test_api_server_process_manager.py +++ b/tests/entrypoints/launchers/api_server/test_api_server_process_manager.py @@ -536,3 +536,56 @@ def test_rust_frontend_uses_config_model_as_model_tag(monkeypatch, caplog, tmp_p sock.close() assert '"model_tag": "org/model"' in caplog.text + + +def test_rust_frontend_inherits_grpc_listener(monkeypatch): + """The Python-bound gRPC socket is passed to Rust like the HTTP one.""" + import subprocess as subprocess_mod + + from vllm.entrypoints.launchers.cli_args import make_arg_parser + from vllm.utils.argparse_utils import FlexibleArgumentParser + from vllm.v1.utils import RustFrontendProcessManager + + args = make_arg_parser(FlexibleArgumentParser()).parse_args( + ["--model", "org/model", "--grpc-port", "50051"] + ) + + class _FakeProc: + pid = 4321 + returncode = 0 + + def poll(self): + return 0 + + def wait(self, timeout=None): + return 0 + + popen_calls = [] + + def fake_popen(cmd, **kwargs): + popen_calls.append((cmd, kwargs)) + return _FakeProc() + + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + grpc_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + monkeypatch.setattr(subprocess_mod, "Popen", fake_popen) + RustFrontendProcessManager( + binary_path="/nonexistent/vllm-rs", + sock=sock, + args=args, + input_address="ipc:///tmp/in", + output_address="ipc:///tmp/out", + engine_start_index=0, + engine_count=1, + data_parallel_size=1, + grpc_sock=grpc_sock, + ) + [(cmd, kwargs)] = popen_calls + grpc_fd = grpc_sock.fileno() + flag_index = cmd.index("--grpc-listen-fd") + assert cmd[flag_index + 1] == str(grpc_fd) + assert list(kwargs["pass_fds"]) == [sock.fileno(), grpc_fd] + finally: + sock.close() + grpc_sock.close() diff --git a/tests/entrypoints/launchers/test_cli_args.py b/tests/entrypoints/launchers/test_cli_args.py index 97c1aa5fc7..6b9d496822 100644 --- a/tests/entrypoints/launchers/test_cli_args.py +++ b/tests/entrypoints/launchers/test_cli_args.py @@ -28,7 +28,7 @@ assert CHATML_JINJA_PATH.exists() def _build_vllm_parsers(): vllm_parser = FlexibleArgumentParser() - subparsers = vllm_parser.add_subparsers() + subparsers = vllm_parser.add_subparsers(dest="subparser") serve_parser = subparsers.add_parser("serve") make_arg_parser(serve_parser) return {"vllm": vllm_parser, "vllm serve": serve_parser} @@ -173,6 +173,75 @@ def test_multiple_valid_inputs(serve_parser): ### Tests for serve argument validation that run prior to loading +@pytest.mark.parametrize( + "port, extra_args", + [ + (50051, []), + (0, ["--port", "0"]), + (50051, ["--port", "50051", "--uds", "/tmp/vllm.sock"]), + ], +) +def test_rust_grpc_port_accepts_frontend_launch( + vllm_parser, monkeypatch, port, extra_args +): + monkeypatch.setenv("VLLM_USE_RUST_FRONTEND", "1") + args = vllm_parser.parse_args( + [ + "serve", + "--grpc-port", + str(port), + "--data-parallel-size", + "8", + "--data-parallel-size-local", + "4", + "--data-parallel-start-rank", + "4", + "--data-parallel-hybrid-lb", + *extra_args, + ] + ) + validate_parsed_serve_args(args) + assert args.grpc_port == port + + +@pytest.mark.parametrize( + "rust_enabled, extra_args, port", + [ + (False, [], 50051), + (False, ["--grpc"], 50051), + (True, ["--grpc"], 50051), + (True, ["--headless"], 50051), + (True, ["--api-server-count", "0"], 50051), + (True, ["--api-server-count", "-1"], 50051), + (True, ["--data-parallel-multi-port-external-lb"], 50051), + (True, ["--port", "50051"], 50051), + (True, [], -1), + (True, [], 65536), + ], +) +def test_rust_grpc_port_rejects_incompatible_launch( + vllm_parser, monkeypatch, rust_enabled, extra_args, port +): + monkeypatch.setenv("VLLM_USE_RUST_FRONTEND", "1" if rust_enabled else "0") + args = vllm_parser.parse_args(["serve", "--grpc-port", str(port), *extra_args]) + error = ( + "--grpc and --grpc-port are mutually exclusive" + if "--grpc" in extra_args + else "--grpc-port" + ) + with pytest.raises(ValueError, match=error): + validate_parsed_serve_args(args) + + +def test_rust_grpc_port_requires_serve_subcommand(monkeypatch): + """Direct Python render/API entrypoints must reject a Rust-only listener.""" + monkeypatch.setenv("VLLM_USE_RUST_FRONTEND", "1") + parser = make_arg_parser(FlexibleArgumentParser()) + args = parser.parse_args(["--grpc-port", "50051"]) + with pytest.raises(ValueError, match="--grpc-port requires"): + validate_parsed_serve_args(args) + + def test_enable_auto_choice_passes_without_tool_call_parser(serve_parser): """Ensure validation fails if tool choice is enabled with no call parser.""" # If we enable-auto-tool-choice, explode with no tool-call-parser @@ -265,13 +334,21 @@ def launch_render_parser(): return _build_launch_render_parser() -def test_launch_render_validates_serve_args(launch_render_parser): +@pytest.mark.parametrize( + "cli_args, error", + [ + (["--enable-auto-tool-choice"], TypeError), + (["--grpc-port", "50051"], ValueError), + ], +) +def test_launch_render_validates_serve_args( + launch_render_parser, monkeypatch, cli_args, error +): """`vllm launch render` reuses the serve parser, so it gets the serve checks""" - args = launch_render_parser.parse_args( - args=["launch", "render", "--enable-auto-tool-choice"] - ) + monkeypatch.setenv("VLLM_USE_RUST_FRONTEND", "1") + args = launch_render_parser.parse_args(args=["launch", "render", *cli_args]) assert args.subparser == "launch" - with pytest.raises(TypeError): + with pytest.raises(error): validate_parsed_serve_args(args) diff --git a/vllm/entrypoints/cli/serve.py b/vllm/entrypoints/cli/serve.py index 1f6ca2b843..4c9410488b 100644 --- a/vllm/entrypoints/cli/serve.py +++ b/vllm/entrypoints/cli/serve.py @@ -17,6 +17,7 @@ from vllm.entrypoints.launchers.cli_args import ( validate_parsed_serve_args, ) from vllm.entrypoints.launchers.dp_supervisor import run_dp_supervisor +from vllm.entrypoints.launchers.launcher import create_server_socket from vllm.entrypoints.serve.utils.api_utils import VLLM_SUBCMD_PARSER_EPILOG from vllm.logger import init_logger from vllm.reasoning import ReasoningParserManager @@ -295,6 +296,13 @@ def run_multi_api_server(args: argparse.Namespace): signal.signal(signal.SIGINT, signal_handler) listen_address, sock = setup_server(args, reuse_port=num_api_servers > 1) + # `--grpc-port` is only accepted for the Rust frontend, which inherits this + # listener like the HTTP one. gRPC follows the HTTP TCP host, or IPv4 + # loopback when HTTP uses a Unix socket. + grpc_sock = None + if args.grpc_port is not None: + grpc_host = "127.0.0.1" if args.uds else (args.host or "") + grpc_sock = create_server_socket((grpc_host, args.grpc_port), reuse_port=False) engine_args = vllm.AsyncEngineArgs.from_cli_args(args) propagate_flash_late_interaction(args, engine_args) @@ -355,6 +363,7 @@ def run_multi_api_server(args: argparse.Namespace): api_server_manager = RustFrontendProcessManager( binary_path=rust_frontend_path, sock=sock, + grpc_sock=grpc_sock, args=args, input_address=addresses.inputs[0], output_address=addresses.outputs[0], diff --git a/vllm/entrypoints/launchers/cli_args.py b/vllm/entrypoints/launchers/cli_args.py index a800e1270d..4c1f121372 100644 --- a/vllm/entrypoints/launchers/cli_args.py +++ b/vllm/entrypoints/launchers/cli_args.py @@ -284,6 +284,11 @@ class FrontendArgs(BaseFrontendArgs): """Host name.""" port: int = 8000 """Port number.""" + grpc_port: int | None = None + """Enable the Rust frontend's additional gRPC Inference and Control services + on this port. Requires `VLLM_USE_RUST_FRONTEND=1 vllm serve`; HTTP remains on + `--port`. Binds on `--host`, or 127.0.0.1 with `--uds`. Cannot be combined + with the Python gRPC server's `--grpc` flag.""" data_parallel_supervisor_port: int = 9256 """HTTP port for aggregated health endpoints in multi-port external LB mode.""" @@ -451,8 +456,10 @@ def make_arg_parser(parser: FlexibleArgumentParser) -> FlexibleArgumentParser: "--grpc", action="store_true", default=False, - help="Launch a gRPC server instead of the HTTP OpenAI-compatible " - "server. Requires: pip install vllm[grpc].", + help="Launch the Python gRPC (SMG VllmEngine) server on --port instead " + "of the HTTP OpenAI-compatible server. Requires: pip install vllm[grpc]. " + "Cannot be combined with --grpc-port, which enables the Rust " + "frontend's gRPC services.", ) parser = FrontendArgs.add_cli_args(parser) parser = AsyncEngineArgs.add_cli_args(parser) @@ -460,6 +467,38 @@ def make_arg_parser(parser: FlexibleArgumentParser) -> FlexibleArgumentParser: return parser +def validate_grpc_port_arg(args: argparse.Namespace) -> None: + """Validate the Rust frontend's optional gRPC listener.""" + if getattr(args, "grpc_port", None) is None: + return + + if args.grpc: + raise ValueError( + "--grpc and --grpc-port are mutually exclusive. Use --grpc " + "--port for the Python gRPC server, or VLLM_USE_RUST_FRONTEND=1 " + "with --grpc-port for the Rust frontend." + ) + if not envs.VLLM_USE_RUST_FRONTEND or getattr(args, "subparser", None) != "serve": + raise ValueError("--grpc-port requires VLLM_USE_RUST_FRONTEND=1 vllm serve") + if args.headless or ( + args.api_server_count is not None and args.api_server_count <= 0 + ): + raise ValueError( + "--grpc-port requires a Rust frontend; remove --headless " + "and non-positive --api-server-count" + ) + if args.data_parallel_multi_port_external_lb: + raise ValueError( + "--grpc-port is incompatible with " + "--data-parallel-multi-port-external-lb: " + "its frontends would share the same gRPC port" + ) + if not 0 <= args.grpc_port <= 65535: + raise ValueError("--grpc-port must be between 0 and 65535") + if args.grpc_port != 0 and args.grpc_port == args.port and not args.uds: + raise ValueError("--grpc-port must differ from --port when HTTP uses TCP") + + def validate_parsed_serve_args(args: argparse.Namespace): """Quick checks for model serve args that raise prior to loading.""" # `vllm launch ` builds its parser with make_arg_parser too (see @@ -478,6 +517,8 @@ def validate_parsed_serve_args(args: argparse.Namespace): "the checkpoint's output format." ) + validate_grpc_port_arg(args) + # Enable auto tool needs a tool call parser to be valid if args.enable_auto_tool_choice and not args.tool_call_parser: raise TypeError("Error: --enable-auto-tool-choice requires --tool-call-parser") diff --git a/vllm/v1/utils.py b/vllm/v1/utils.py index f3bd97784e..ac877263a5 100644 --- a/vllm/v1/utils.py +++ b/vllm/v1/utils.py @@ -354,6 +354,7 @@ class RustFrontendProcessManager: engine_count: int, data_parallel_size: int, stats_update_address: str | None = None, + grpc_sock: Any | None = None, ): import os import subprocess @@ -377,6 +378,12 @@ class RustFrontendProcessManager: "--data-parallel-size", str(data_parallel_size), ] + pass_fds = [fd] + if grpc_sock is not None: + grpc_fd = grpc_sock.fileno() + os.set_inheritable(grpc_fd, True) + cmd.extend(["--grpc-listen-fd", str(grpc_fd)]) + pass_fds.append(grpc_fd) if stats_update_address is not None: cmd.extend(["--coordinator-address", stats_update_address]) from vllm.entrypoints.serve.utils.api_utils import jsonify_non_default_args @@ -418,7 +425,7 @@ class RustFrontendProcessManager: redacted_json = json.dumps(redact_sensitive_args(args_dict), sort_keys=True) logger.info("Launching Rust frontend: %s", " ".join(cmd[:-1] + [redacted_json])) - self._proc = subprocess.Popen(cmd, pass_fds=(fd,)) + self._proc = subprocess.Popen(cmd, pass_fds=pass_fds) # Create a process wrapper with a sentinel fd for monitoring self.processes: list[_SubprocessWrapper] = [