From 3d2e7cc6019da2aa0d5b1d0b2cf7bf30fe850dc6 Mon Sep 17 00:00:00 2001 From: Alex Nails Date: Wed, 8 Jul 2026 03:27:25 +0530 Subject: [PATCH] [gRPC] Native server: launcher + HTTP + server args wiring (3/4) (#23508) Co-authored-by: Claude Opus 4.7 (1M context) --- python/sglang/launch_server.py | 13 +- python/sglang/srt/entrypoints/grpc_server.py | 6 +- python/sglang/srt/entrypoints/http_server.py | 83 +++++++++++-- python/sglang/srt/environ.py | 11 +- python/sglang/srt/server_args.py | 101 ++++++++++++---- .../unit/server_args/test_server_args.py | 112 ++++++++++++++++++ 6 files changed, 286 insertions(+), 40 deletions(-) diff --git a/python/sglang/launch_server.py b/python/sglang/launch_server.py index 04a5ef919..8091692fd 100644 --- a/python/sglang/launch_server.py +++ b/python/sglang/launch_server.py @@ -13,10 +13,10 @@ suppress_noisy_warnings() def run_server(server_args): - """Run the server based on server_args.grpc_mode and server_args.encoder_only.""" + """Run the server based on the gRPC flags and server_args.encoder_only.""" if server_args.encoder_only: # For encoder disaggregation - if server_args.grpc_mode: + if server_args.smg_grpc_mode or server_args.grpc_mode: from sglang.srt.disaggregation.encode_grpc_server import ( serve_grpc_encoder, ) @@ -26,10 +26,11 @@ def run_server(server_args): from sglang.srt.disaggregation.encode_server import launch_server launch_server(server_args) - elif server_args.grpc_mode: - # TODO: Once the native Rust gRPC server starts alongside HTTP in the - # default path below (controlled by SGLANG_ENABLE_GRPC / SGLANG_GRPC_PORT), - # remove this legacy SMG path and the grpc_mode flag. + elif server_args.smg_grpc_mode: + # Legacy SMG gRPC server (--smg-grpc-mode, or the deprecated --grpc-mode + # which __post_init__ folds into smg_grpc_mode). The native Rust gRPC + # server is a separate path, enabled by --grpc-port, that starts + # alongside the default HTTP server below. from sglang.srt.entrypoints.grpc_server import serve_grpc asyncio.run(serve_grpc(server_args)) diff --git a/python/sglang/srt/entrypoints/grpc_server.py b/python/sglang/srt/entrypoints/grpc_server.py index 320e04a94..c3e22762c 100644 --- a/python/sglang/srt/entrypoints/grpc_server.py +++ b/python/sglang/srt/entrypoints/grpc_server.py @@ -5,7 +5,7 @@ A lightweight HTTP sidecar is started alongside the gRPC server to expose: - /metrics (Prometheus, when --enable-metrics is set) - /start_profile, /stop_profile (profiling control) -The sidecar is started on --grpc-http-sidecar-port (default: --port + 1) +The sidecar is started on --smg-http-sidecar-port (default: --port + 1) once the gRPC request manager is ready, regardless of whether --enable-metrics is set. """ @@ -168,8 +168,8 @@ async def serve_grpc(server_args, model_info=None): sidecar_app = web.Application() sidecar_runner = None sidecar_port = ( - server_args.grpc_http_sidecar_port - if server_args.grpc_http_sidecar_port is not None + server_args.smg_http_sidecar_port + if server_args.smg_http_sidecar_port is not None else server_args.port + 1 ) diff --git a/python/sglang/srt/entrypoints/http_server.py b/python/sglang/srt/entrypoints/http_server.py index bd9d7eafa..3b5b569b3 100644 --- a/python/sglang/srt/entrypoints/http_server.py +++ b/python/sglang/srt/entrypoints/http_server.py @@ -260,6 +260,8 @@ async def init_multi_tokenizer() -> ServerArgs: @asynccontextmanager async def lifespan(fast_api_app: FastAPI): + grpc_handle = None + warmup_thread = None if getattr(fast_api_app, "is_single_tokenizer_mode", False): server_args = fast_api_app.server_args warmup_thread_kwargs = fast_api_app.warmup_thread_kwargs @@ -375,20 +377,38 @@ async def lifespan(fast_api_app: FastAPI): ) logger.info("Warmup ended") - # Execute the general warmup - warmup_thread = threading.Thread( - target=_wait_and_warmup, - kwargs=warmup_thread_kwargs, - ) - warmup_thread.start() - - # Start the HTTP server + # Start the native gRPC server and warmup inside the try so a failure in + # either still runs the finally cleanup below. Native gRPC is enabled via + # --grpc-port / SGLANG_GRPC_PORT; only the single-tokenizer process is + # gRPC-capable (__post_init__ rejects --tokenizer-worker-num > 1). try: + if ( + getattr(fast_api_app, "is_single_tokenizer_mode", False) + and server_args.grpc_port is not None + and not (server_args.smg_grpc_mode or server_args.grpc_mode) + ): + grpc_handle = _start_native_grpc_server_for_runtime( + server_args=server_args, + tokenizer_manager=_global_state.tokenizer_manager, + template_manager=_global_state.template_manager, + scheduler_info=_global_state.scheduler_info, + ) + + # Execute the general warmup + warmup_thread = threading.Thread( + target=_wait_and_warmup, + kwargs=warmup_thread_kwargs, + ) + warmup_thread.start() + + # Start the HTTP server yield finally: + _shutdown_native_grpc_server(grpc_handle) if tool_server is not None and hasattr(tool_server, "aclose"): await tool_server.aclose() - warmup_thread.join() + if warmup_thread is not None: + warmup_thread.join() # Fast API @@ -2476,6 +2496,51 @@ def _setup_and_run_http_server( _global_state.tokenizer_manager.socket_mapping.clear_all_sockets() +def _start_native_grpc_server_for_runtime( + server_args, + tokenizer_manager, + template_manager, + scheduler_info, +): + try: + from sglang.srt.entrypoints.grpc_bridge import RuntimeHandle + from sglang.srt.grpc import _core as grpc_native + except ImportError as e: + raise RuntimeError( + "Native gRPC extension (sglang.srt.grpc._core) not found in this wheel, " + "but --grpc-port was set. The extension is built from " + "rust/sglang-grpc/ via setuptools-rust during wheel build. Either " + "install a wheel that includes the extension or unset --grpc-port." + ) from e + + runtime_handle = RuntimeHandle( + tokenizer_manager=tokenizer_manager, + template_manager=template_manager, + server_args=server_args, + scheduler_info=scheduler_info or {}, + ) + + grpc_handle = grpc_native.start_server( + host=server_args.host, + port=server_args.grpc_port, + runtime_handle=runtime_handle, + worker_threads=server_args.grpc_worker_threads, + ) + logger.info( + f"Native gRPC server started on {server_args.host}:{server_args.grpc_port}" + ) + return grpc_handle + + +def _shutdown_native_grpc_server(grpc_handle) -> None: + if grpc_handle is None: + return + try: + grpc_handle.shutdown() + except Exception as e: + logger.warning(f"Failed to shut down native gRPC server: {e}") + + def launch_server( server_args: ServerArgs, init_tokenizer_manager_func: Callable = init_tokenizer_manager, diff --git a/python/sglang/srt/environ.py b/python/sglang/srt/environ.py index 44c467025..2942f53da 100644 --- a/python/sglang/srt/environ.py +++ b/python/sglang/srt/environ.py @@ -777,9 +777,12 @@ class Envs: # Encoder receiver selection: http|grpc (used by EPD paths). SGLANG_ENCODER_MM_RECEIVER_MODE = EnvStr("http") - # Native gRPC server (internal, not yet user-facing) + # Native gRPC server. SGLANG_GRPC_PORT is the env fallback for the + # --grpc-port CLI flag; setting either enables the native server alongside + # HTTP. The worker-threads knob stays env-only (internal tuning, no CLI + # surface). SGLANG_GRPC_PORT = EnvInt(None) - SGLANG_ENABLE_GRPC = EnvBool(False) + SGLANG_GRPC_WORKER_THREADS = EnvInt(4) # External models SGLANG_EXTERNAL_MODEL_PACKAGE = EnvStr("") @@ -1043,6 +1046,10 @@ def _convert_SGL_to_SGLANG(): _convert_SGL_to_SGLANG() +_warn_deprecated_env_to_cli_flag( + "SGLANG_ENABLE_GRPC", + "Please use '--grpc-port' to enable the native gRPC server.", +) _warn_deprecated_env_to_cli_flag( "SGLANG_SCHEDULER_DECREASE_PREFILL_IDLE", "Please use '--enable-prefill-delayer' instead.", diff --git a/python/sglang/srt/server_args.py b/python/sglang/srt/server_args.py index 63591ed6c..1a73fab3c 100644 --- a/python/sglang/srt/server_args.py +++ b/python/sglang/srt/server_args.py @@ -977,7 +977,22 @@ class ServerArgs: host: A[str, "The host of the HTTP server."] = "127.0.0.1" port: A[int, "The port of the HTTP server."] = 30000 fastapi_root_path: A[str, "App is behind a path based routing proxy."] = "" - grpc_mode: A[bool, "If set, use gRPC server instead of HTTP server."] = False + smg_grpc_mode: A[ + bool, + "Use the legacy SMG gRPC server (smg-grpc-servicer) instead of the HTTP " + "server. Replaces the deprecated --grpc-mode.", + ] = False + grpc_mode: A[ + bool, + "(Deprecated, use --smg-grpc-mode) Legacy SMG gRPC server selector.", + ] = False + grpc_port: A[ + Optional[int], + "Port for the native gRPC server, started alongside HTTP. Setting this " + "(or SGLANG_GRPC_PORT) enables the native gRPC server; it is off by " + "default. In legacy --smg-grpc-mode this is the SMG server port and " + "defaults to --port + 10000.", + ] = None skip_server_warmup: A[bool, "If set, skip warmup."] = False warmups: A[ Optional[str], @@ -1154,9 +1169,12 @@ class ServerArgs: ] = None show_time_cost: A[bool, "Show time cost of custom marks."] = False enable_metrics: A[bool, "Enable log prometheus metrics."] = False - grpc_http_sidecar_port: A[ + smg_http_sidecar_port: A[ Optional[int], - "Port for the HTTP sidecar server in gRPC mode (--grpc-mode). Serves Prometheus metrics and profiling endpoints. Defaults to --port + 1. Not used in HTTP mode.", + Arg( + help="Port for the HTTP sidecar server in legacy SMG gRPC mode (--smg-grpc-mode). Serves Prometheus metrics and profiling endpoints. Defaults to --port + 1. Not used in HTTP mode.", + aliases=["--grpc-http-sidecar-port"], + ), ] = None enable_mfu_metrics: A[bool, "Enable estimated MFU-related prometheus metrics."] = ( False @@ -3011,20 +3029,65 @@ class ServerArgs: ) setattr(self, attr, "dsv4") - # Native gRPC flags — env-only for now, not exposed as CLI args. - # Set as instance attributes (not dataclass fields) to avoid - # argparse namespace lookup in from_cli_args. - self.enable_grpc = envs.SGLANG_ENABLE_GRPC.get() + # --grpc-mode is a deprecated alias for --smg-grpc-mode. + if self.grpc_mode and not self.smg_grpc_mode: + logger.warning( + "--grpc-mode is deprecated and will be removed in a future " + "version. Use --smg-grpc-mode for the legacy SMG gRPC server, " + "or --grpc-port for the native gRPC server." + ) + self.smg_grpc_mode = True + + # Native gRPC tuning knob is env-only; --grpc-port (CLI) enables the + # native server, falling back to SGLANG_GRPC_PORT. + self.grpc_worker_threads = envs.SGLANG_GRPC_WORKER_THREADS.get() grpc_port_env = envs.SGLANG_GRPC_PORT.get() - self.grpc_port = ( - grpc_port_env if grpc_port_env is not None else self.port + 10000 - ) + if self.grpc_port is None and grpc_port_env is not None: + self.grpc_port = grpc_port_env - if not (1 <= self.grpc_port <= 65535): - raise ValueError( - f"SGLANG_GRPC_PORT ({self.grpc_port}) must be between 1 and 65535" - ) + # Legacy SMG defaults its port to --port + 10000. Derive/validate only + # when gRPC is in use, so HTTP-only high ports don't fail validation. + legacy_grpc = self.smg_grpc_mode or self.grpc_mode + if legacy_grpc and self.grpc_port is None: + self.grpc_port = self.port + 10000 + + if self.grpc_port is not None: + if not (1 <= self.grpc_port <= 65535): + raise ValueError( + "--grpc-port / SGLANG_GRPC_PORT " + f"({self.grpc_port}) must be between 1 and 65535" + ) + if self.grpc_worker_threads < 1: + raise ValueError( + "SGLANG_GRPC_WORKER_THREADS " + f"({self.grpc_worker_threads}) must be >= 1" + ) + + # Native gRPC is incompatible with launch paths it doesn't wire into. + # Legacy takes precedence over grpc_port, keeping re-runs idempotent. + native_grpc = self.grpc_port is not None and not legacy_grpc + if native_grpc: + if self.use_ray: + raise ValueError( + "--grpc-port is not supported with --use-ray: the Ray " + "serve launch path does not start the native gRPC server." + ) + if self.encoder_only: + raise ValueError( + "--grpc-port is not supported with --encoder-only: " + "encoder disaggregation uses its own server." + ) + if self.tokenizer_worker_num > 1: + raise ValueError( + "Native gRPC does not yet support --tokenizer-worker-num > 1. " + "Unset --grpc-port or set --tokenizer-worker-num 1." + ) + if self.api_key or self.admin_api_key: + raise ValueError( + "--grpc-port is incompatible with --api-key/--admin-api-key: " + "the native gRPC listener bypasses HTTP auth middleware." + ) def _handle_prefill_delayer_env_compat(self): if envs.SGLANG_SCHEDULER_DECREASE_PREFILL_IDLE.get(): @@ -6810,13 +6873,11 @@ class ServerArgs: "Communications quantization is only supported for NPU device" ) - if ( - self.enable_grpc - and self.grpc_port is not None - and self.grpc_port == self.port - ): + # grpc_port is None for HTTP-only launches, so the == comparison is + # already False there; no explicit None check needed. + if not (self.smg_grpc_mode or self.grpc_mode) and self.grpc_port == self.port: raise ValueError( - f"SGLANG_GRPC_PORT ({self.grpc_port}) must differ from --port ({self.port})" + f"--grpc-port ({self.grpc_port}) must differ from --port ({self.port})" ) # TODO: Also validate grpc_port != metrics_http_port and grpc_port != nccl_port diff --git a/test/registered/unit/server_args/test_server_args.py b/test/registered/unit/server_args/test_server_args.py index 35f252bb1..74605e66c 100644 --- a/test/registered/unit/server_args/test_server_args.py +++ b/test/registered/unit/server_args/test_server_args.py @@ -8,6 +8,7 @@ from unittest.mock import MagicMock, patch import sglang.srt.server_args as server_args_module from sglang.srt.arg_groups.speculative_hook import handle_speculative_decoding +from sglang.srt.environ import envs from sglang.srt.layers.cp.base import is_cp_enabled, is_interleave from sglang.srt.model_executor.cuda_graph_config import ( Backend, @@ -1382,6 +1383,117 @@ class TestSamplingBackendTokenOracleEnvGate(CustomTestCase): self.assertEqual(parsed.sampling_backend, "token_oracle") +class TestGrpcServerArgs(CustomTestCase): + """Native gRPC is enabled by --grpc-port (or SGLANG_GRPC_PORT) and runs + alongside HTTP; --smg-grpc-mode (and the deprecated --grpc-mode) select the + legacy SMG server. Worker-threads / max-prefill-tokens are env-only knobs. + + The gRPC setup lives in ServerArgs._handle_deprecated_args, which + __post_init__ skips for dummy models, so these tests build a dummy + ServerArgs and invoke that handler directly (mirroring the real flow for a + concrete model path). + """ + + @staticmethod + def _args(**kwargs): + return ServerArgs(model_path="dummy", **kwargs) + + def test_defaults_native_grpc_off_legacy_off(self): + sa = self._args() + sa._handle_deprecated_args() + self.assertIsNone(sa.grpc_port) + self.assertFalse(sa.smg_grpc_mode) + + def test_http_only_high_port_does_not_derive_grpc_port(self): + sa = self._args(port=56000) + sa._handle_deprecated_args() + self.assertIsNone(sa.grpc_port) + + def test_grpc_port_enables_native_and_env_knobs(self): + sa = self._args(grpc_port=50051) + with envs.SGLANG_GRPC_WORKER_THREADS.override(8): + sa._handle_deprecated_args() + self.assertEqual(sa.grpc_port, 50051) + self.assertEqual(sa.grpc_worker_threads, 8) + + def test_env_grpc_port_enables_native(self): + sa = self._args(port=30000) + with envs.SGLANG_GRPC_PORT.override(45000): + sa._handle_deprecated_args() + self.assertEqual(sa.grpc_port, 45000) + + def test_legacy_smg_derives_grpc_port_from_http_port(self): + sa = self._args(port=30000, smg_grpc_mode=True) + sa._handle_deprecated_args() + self.assertEqual(sa.grpc_port, 40000) + + def test_grpc_mode_is_deprecated_alias_for_smg_grpc_mode(self): + sa = self._args(grpc_mode=True) + with self.assertLogs(server_args_module.logger, level="WARNING") as cm: + sa._handle_deprecated_args() + self.assertTrue(sa.smg_grpc_mode) + self.assertTrue(any("--grpc-mode is deprecated" in line for line in cm.output)) + + def test_legacy_smg_takes_precedence_over_grpc_port(self): + sa = self._args(grpc_port=50051, smg_grpc_mode=True) + sa._handle_deprecated_args() + self.assertTrue(sa.smg_grpc_mode) + self.assertEqual(sa.grpc_port, 50051) + + def test_native_grpc_rejects_multi_tokenizer(self): + sa = self._args(grpc_port=40000, tokenizer_worker_num=2) + with self.assertRaises(ValueError): + sa._handle_deprecated_args() + + def test_native_grpc_rejects_http_auth(self): + sa = self._args(grpc_port=40000, api_key="secret") + with self.assertRaises(ValueError): + sa._handle_deprecated_args() + + def test_invalid_grpc_worker_threads_rejected(self): + sa = self._args(grpc_port=40000) + with envs.SGLANG_GRPC_WORKER_THREADS.override(0): + with self.assertRaises(ValueError): + sa._handle_deprecated_args() + + def test_start_server_call_site_matches_native_signature(self): + """Regression for the startup blocker: the native start_server binding + only accepts (host, port, runtime_handle, worker_threads, ...). The + arg-parsing tests above never call start_server, so a stray kwarg (e.g. + the removed max_prefill_tokens) would only surface as a TypeError at + launch. This mocks the native extension and locks the kwarg set.""" + import sys + + from sglang.srt.entrypoints import http_server + + fake_core = SimpleNamespace(start_server=MagicMock(return_value="handle")) + fake_bridge = SimpleNamespace(RuntimeHandle=MagicMock(return_value="rt")) + server_args = SimpleNamespace( + host="127.0.0.1", grpc_port=50051, grpc_worker_threads=4 + ) + with patch.dict( + sys.modules, + { + "sglang.srt.grpc": SimpleNamespace(_core=fake_core), + "sglang.srt.grpc._core": fake_core, + "sglang.srt.entrypoints.grpc_bridge": fake_bridge, + }, + ): + handle = http_server._start_native_grpc_server_for_runtime( + server_args=server_args, + tokenizer_manager=MagicMock(), + template_manager=MagicMock(), + scheduler_info={}, + ) + + self.assertEqual(handle, "handle") + _, kwargs = fake_core.start_server.call_args + self.assertEqual( + set(kwargs), {"host", "port", "runtime_handle", "worker_threads"} + ) + self.assertNotIn("max_prefill_tokens", kwargs) + + class TestTwoBatchOverlapBackend(CustomTestCase): """Non-EP DP two-batch-overlap backend requirement.