[Rust Frontend] Expose gRPC port in Python vllm serve (#59659)
Co-authored-by: Bugen Zhao <i@bugenzhao.com> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Alec Flowers <aflowers@nvidia.com> Signed-off-by: Bugen Zhao <i@bugenzhao.com>
This commit is contained in:
co-authored by
Bugen Zhao
Claude Opus 5.5
parent
b558f160a2
commit
9d2a6f52b2
@@ -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
|
||||
|
||||
+24
-7
@@ -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<i32>,
|
||||
/// TCP port for the gRPC Inference service. When not set, no gRPC server is
|
||||
/// started.
|
||||
#[arg(long)]
|
||||
#[serde(default)]
|
||||
pub grpc_port: Option<u16>,
|
||||
/// 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<i32>,
|
||||
input_address: String,
|
||||
output_address: String,
|
||||
coordinator_address: Option<String>,
|
||||
@@ -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<HttpListenerMode>,
|
||||
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<i32>,
|
||||
/// 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<String>,
|
||||
/// TCP port for the gRPC Inference and Control services. When not set, no
|
||||
/// gRPC server is started.
|
||||
#[arg(long)]
|
||||
pub grpc_port: Option<u16>,
|
||||
|
||||
/// 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,
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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<String>,
|
||||
/// HTTP listener setup.
|
||||
pub listener_mode: HttpListenerMode,
|
||||
/// gRPC listener setup. When `None`, no gRPC server is started.
|
||||
pub grpc_listener_mode: Option<HttpListenerMode>,
|
||||
/// 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<u16>,
|
||||
/// 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
|
||||
|
||||
@@ -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<Arc<AppState>> {
|
||||
@@ -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");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
|
||||
@@ -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],
|
||||
|
||||
@@ -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 <component>` 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")
|
||||
|
||||
+8
-1
@@ -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] = [
|
||||
|
||||
Reference in New Issue
Block a user