feat: add native gRPC sidecar module launcher (#31076)
Signed-off-by: Ishan Dhanani <ishandhanani@gmail.com> Signed-off-by: Connor Carpenter <connorc@nvidia.com> Co-authored-by: Connor Carpenter <connorc@nvidia.com>
This commit is contained in:
co-authored by
Connor Carpenter
parent
74338e94f1
commit
21065bc862
@@ -265,6 +265,7 @@ async def init_multi_tokenizer() -> ServerArgs:
|
|||||||
@asynccontextmanager
|
@asynccontextmanager
|
||||||
async def lifespan(fast_api_app: FastAPI):
|
async def lifespan(fast_api_app: FastAPI):
|
||||||
grpc_handle = None
|
grpc_handle = None
|
||||||
|
sidecar = None
|
||||||
warmup_thread = None
|
warmup_thread = None
|
||||||
if getattr(fast_api_app, "is_single_tokenizer_mode", False):
|
if getattr(fast_api_app, "is_single_tokenizer_mode", False):
|
||||||
server_args = fast_api_app.server_args
|
server_args = fast_api_app.server_args
|
||||||
@@ -397,6 +398,10 @@ async def lifespan(fast_api_app: FastAPI):
|
|||||||
template_manager=_global_state.template_manager,
|
template_manager=_global_state.template_manager,
|
||||||
scheduler_info=_global_state.scheduler_info,
|
scheduler_info=_global_state.scheduler_info,
|
||||||
)
|
)
|
||||||
|
if server_args.sidecar is not None:
|
||||||
|
from sglang.srt.entrypoints.sidecar import start_sidecar
|
||||||
|
|
||||||
|
sidecar = start_sidecar(server_args)
|
||||||
|
|
||||||
# Execute the general warmup
|
# Execute the general warmup
|
||||||
warmup_thread = threading.Thread(
|
warmup_thread = threading.Thread(
|
||||||
@@ -408,6 +413,11 @@ async def lifespan(fast_api_app: FastAPI):
|
|||||||
# Start the HTTP server
|
# Start the HTTP server
|
||||||
yield
|
yield
|
||||||
finally:
|
finally:
|
||||||
|
if sidecar is not None:
|
||||||
|
try:
|
||||||
|
sidecar.stop()
|
||||||
|
except Exception:
|
||||||
|
logger.exception("Failed to stop sidecar")
|
||||||
_shutdown_native_grpc_server(grpc_handle)
|
_shutdown_native_grpc_server(grpc_handle)
|
||||||
if tool_server is not None and hasattr(tool_server, "aclose"):
|
if tool_server is not None and hasattr(tool_server, "aclose"):
|
||||||
await tool_server.aclose()
|
await tool_server.aclose()
|
||||||
|
|||||||
@@ -0,0 +1,130 @@
|
|||||||
|
# Copyright 2023-2026 SGLang Team
|
||||||
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
# you may not use this file except in compliance with the License.
|
||||||
|
# You may obtain a copy of the License at
|
||||||
|
#
|
||||||
|
# http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
#
|
||||||
|
# Unless required by applicable law or agreed to in writing, software
|
||||||
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
# See the License for the specific language governing permissions and
|
||||||
|
# limitations under the License.
|
||||||
|
"""Lifecycle management for an optional local native gRPC sidecar."""
|
||||||
|
|
||||||
|
import argparse
|
||||||
|
import importlib
|
||||||
|
import logging
|
||||||
|
import multiprocessing as mp
|
||||||
|
import os
|
||||||
|
|
||||||
|
from sglang.srt.utils.common import kill_itself_when_parent_died, kill_process_tree
|
||||||
|
from sglang.srt.utils.network import NetworkAddress
|
||||||
|
from sglang.srt.utils.watchdog import SubprocessWatchdog
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
SGLANG_GRPC_ENDPOINT_ENV = "SGLANG_GRPC_ENDPOINT"
|
||||||
|
_DEFAULT_SIDECAR_SHUTDOWN_TIMEOUT = 45.0
|
||||||
|
|
||||||
|
|
||||||
|
def _loopback_host(host: str) -> str:
|
||||||
|
if not host or host == "0.0.0.0":
|
||||||
|
return "127.0.0.1"
|
||||||
|
if host in ("::", "[::]"):
|
||||||
|
return "::1"
|
||||||
|
return host
|
||||||
|
|
||||||
|
|
||||||
|
def build_sidecar_endpoint(server_args) -> str:
|
||||||
|
return NetworkAddress(
|
||||||
|
_loopback_host(server_args.host), server_args.grpc_port
|
||||||
|
).to_url()
|
||||||
|
|
||||||
|
|
||||||
|
def _parse_sidecar_args(args: list[str] | None) -> tuple[list[str], float]:
|
||||||
|
parser = argparse.ArgumentParser(add_help=False, allow_abbrev=False)
|
||||||
|
parser.add_argument(
|
||||||
|
"--sidecar-shutdown-timeout",
|
||||||
|
type=float,
|
||||||
|
default=_DEFAULT_SIDECAR_SHUTDOWN_TIMEOUT,
|
||||||
|
)
|
||||||
|
parsed, provider_args = parser.parse_known_args(args or [])
|
||||||
|
if parsed.sidecar_shutdown_timeout <= 0:
|
||||||
|
raise ValueError("--sidecar-shutdown-timeout must be greater than 0.")
|
||||||
|
return provider_args, parsed.sidecar_shutdown_timeout
|
||||||
|
|
||||||
|
|
||||||
|
def _run_sidecar(module_name: str, args: list[str], endpoint: str) -> None:
|
||||||
|
kill_itself_when_parent_died()
|
||||||
|
os.environ[SGLANG_GRPC_ENDPOINT_ENV] = endpoint
|
||||||
|
try:
|
||||||
|
main = getattr(importlib.import_module(module_name), "main")
|
||||||
|
except (AttributeError, ImportError) as e:
|
||||||
|
raise RuntimeError(
|
||||||
|
f"--sidecar requires importable module {module_name!r} "
|
||||||
|
"with a main(argv) function."
|
||||||
|
) from e
|
||||||
|
|
||||||
|
if not callable(main):
|
||||||
|
raise RuntimeError(
|
||||||
|
f"--sidecar requires module {module_name!r} to expose "
|
||||||
|
"a callable main(argv)."
|
||||||
|
)
|
||||||
|
|
||||||
|
main(args)
|
||||||
|
|
||||||
|
|
||||||
|
class Sidecar:
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
proc,
|
||||||
|
module_name: str,
|
||||||
|
shutdown_timeout: float,
|
||||||
|
):
|
||||||
|
self.proc = proc
|
||||||
|
self.module_name = module_name
|
||||||
|
self.shutdown_timeout = shutdown_timeout
|
||||||
|
self._watchdog = SubprocessWatchdog(
|
||||||
|
processes=[proc], process_names=[module_name]
|
||||||
|
)
|
||||||
|
|
||||||
|
def start(self) -> None:
|
||||||
|
self.proc.start()
|
||||||
|
self._watchdog.start()
|
||||||
|
logger.info(
|
||||||
|
"Sidecar module %s started pid=%s",
|
||||||
|
self.module_name,
|
||||||
|
self.proc.pid,
|
||||||
|
)
|
||||||
|
|
||||||
|
def stop(self) -> None:
|
||||||
|
self._watchdog.stop()
|
||||||
|
if self.proc.is_alive():
|
||||||
|
self.proc.terminate()
|
||||||
|
self.proc.join(timeout=self.shutdown_timeout)
|
||||||
|
else:
|
||||||
|
self.proc.join(timeout=0)
|
||||||
|
|
||||||
|
if self.proc.is_alive():
|
||||||
|
logger.warning("Sidecar module did not terminate; killing process tree")
|
||||||
|
kill_process_tree(self.proc.pid, wait_timeout=self.shutdown_timeout)
|
||||||
|
|
||||||
|
|
||||||
|
def start_sidecar(server_args) -> Sidecar:
|
||||||
|
module_name = server_args.sidecar
|
||||||
|
assert module_name is not None
|
||||||
|
sidecar_args, shutdown_timeout = _parse_sidecar_args(server_args.sidecar_args)
|
||||||
|
endpoint = build_sidecar_endpoint(server_args)
|
||||||
|
proc = mp.get_context("spawn").Process(
|
||||||
|
name=f"sglang_sidecar_{module_name}",
|
||||||
|
target=_run_sidecar,
|
||||||
|
args=(module_name, sidecar_args, endpoint),
|
||||||
|
)
|
||||||
|
sidecar = Sidecar(
|
||||||
|
proc,
|
||||||
|
module_name,
|
||||||
|
shutdown_timeout=shutdown_timeout,
|
||||||
|
)
|
||||||
|
sidecar.start()
|
||||||
|
return sidecar
|
||||||
@@ -1181,6 +1181,24 @@ class ServerArgs:
|
|||||||
"defaults to --port + 10000.",
|
"defaults to --port + 10000.",
|
||||||
NS("serving"),
|
NS("serving"),
|
||||||
] = None
|
] = None
|
||||||
|
sidecar: A[
|
||||||
|
Optional[str],
|
||||||
|
"Start a locally managed sidecar against the native gRPC server. "
|
||||||
|
"The selected module must expose main(argv) and read the resolved "
|
||||||
|
"native gRPC endpoint from SGLANG_GRPC_ENDPOINT. Requires --grpc-port "
|
||||||
|
"or SGLANG_GRPC_PORT.",
|
||||||
|
NS("serving"),
|
||||||
|
] = None
|
||||||
|
sidecar_args: A[
|
||||||
|
Optional[List[str]],
|
||||||
|
Arg(
|
||||||
|
help="JSON array passed to the selected sidecar module's "
|
||||||
|
"main(argv) function. --sidecar-shutdown-timeout SECONDS is "
|
||||||
|
"consumed by SGLang.",
|
||||||
|
type_parser=json_list_type,
|
||||||
|
),
|
||||||
|
NS("serving"),
|
||||||
|
] = None
|
||||||
skip_server_warmup: A[bool, "If set, skip warmup.", NS("serving")] = False
|
skip_server_warmup: A[bool, "If set, skip warmup.", NS("serving")] = False
|
||||||
warmups: A[
|
warmups: A[
|
||||||
Optional[str],
|
Optional[str],
|
||||||
@@ -3730,6 +3748,23 @@ class ServerArgs:
|
|||||||
# Native gRPC is incompatible with launch paths it doesn't wire into.
|
# Native gRPC is incompatible with launch paths it doesn't wire into.
|
||||||
# Legacy takes precedence over grpc_port, keeping re-runs idempotent.
|
# Legacy takes precedence over grpc_port, keeping re-runs idempotent.
|
||||||
native_grpc = self.grpc_port is not None and not legacy_grpc
|
native_grpc = self.grpc_port is not None and not legacy_grpc
|
||||||
|
if self.sidecar_args is not None:
|
||||||
|
if self.sidecar is None:
|
||||||
|
raise ValueError("--sidecar-args requires --sidecar.")
|
||||||
|
if not isinstance(self.sidecar_args, list) or not all(
|
||||||
|
isinstance(arg, str) for arg in self.sidecar_args
|
||||||
|
):
|
||||||
|
raise ValueError("--sidecar-args must be a JSON array of strings.")
|
||||||
|
if self.sidecar is not None:
|
||||||
|
if not self.sidecar.strip():
|
||||||
|
raise ValueError("--sidecar must not be empty.")
|
||||||
|
if legacy_grpc:
|
||||||
|
raise ValueError(
|
||||||
|
"--sidecar requires SGLang's native gRPC server; "
|
||||||
|
"it cannot be combined with --smg-grpc-mode/--grpc-mode."
|
||||||
|
)
|
||||||
|
if self.grpc_port is None:
|
||||||
|
raise ValueError("--sidecar requires --grpc-port or SGLANG_GRPC_PORT.")
|
||||||
if native_grpc:
|
if native_grpc:
|
||||||
if self.use_ray:
|
if self.use_ray:
|
||||||
raise ValueError(
|
raise ValueError(
|
||||||
|
|||||||
@@ -9,6 +9,13 @@ from unittest.mock import MagicMock, patch
|
|||||||
|
|
||||||
import sglang.srt.server_args as server_args_module
|
import sglang.srt.server_args as server_args_module
|
||||||
from sglang.srt.arg_groups.speculative_hook import handle_speculative_decoding
|
from sglang.srt.arg_groups.speculative_hook import handle_speculative_decoding
|
||||||
|
from sglang.srt.entrypoints.sidecar import (
|
||||||
|
SGLANG_GRPC_ENDPOINT_ENV,
|
||||||
|
Sidecar,
|
||||||
|
_run_sidecar,
|
||||||
|
build_sidecar_endpoint,
|
||||||
|
start_sidecar,
|
||||||
|
)
|
||||||
from sglang.srt.environ import envs
|
from sglang.srt.environ import envs
|
||||||
from sglang.srt.layers.cp.base import is_cp_enabled, is_interleave
|
from sglang.srt.layers.cp.base import is_cp_enabled, is_interleave
|
||||||
from sglang.srt.model_executor.cuda_graph_config import (
|
from sglang.srt.model_executor.cuda_graph_config import (
|
||||||
@@ -1536,6 +1543,133 @@ class TestGrpcServerArgs(CustomTestCase):
|
|||||||
sa._handle_deprecated_args()
|
sa._handle_deprecated_args()
|
||||||
self.assertEqual(sa.grpc_port, 45000)
|
self.assertEqual(sa.grpc_port, 45000)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _sidecar_parser():
|
||||||
|
parser = server_args_module.argparse.ArgumentParser()
|
||||||
|
ServerArgs.add_cli_args(parser)
|
||||||
|
return parser
|
||||||
|
|
||||||
|
def test_sidecar_builds_loopback_grpc_endpoints(self):
|
||||||
|
self.assertEqual(
|
||||||
|
build_sidecar_endpoint(SimpleNamespace(host="0.0.0.0", grpc_port=50051)),
|
||||||
|
"http://127.0.0.1:50051",
|
||||||
|
)
|
||||||
|
self.assertEqual(
|
||||||
|
build_sidecar_endpoint(SimpleNamespace(host="::", grpc_port=50051)),
|
||||||
|
"http://[::1]:50051",
|
||||||
|
)
|
||||||
|
self.assertEqual(
|
||||||
|
build_sidecar_endpoint(SimpleNamespace(host="[::]", grpc_port=50051)),
|
||||||
|
"http://[::1]:50051",
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_sidecar_args_parse_as_exact_json_argv(self):
|
||||||
|
argv = ["--flag", "value"]
|
||||||
|
parsed = self._sidecar_parser().parse_args(
|
||||||
|
["--model-path", "dummy", "--sidecar-args", json.dumps(argv)]
|
||||||
|
)
|
||||||
|
self.assertEqual(parsed.sidecar_args, argv)
|
||||||
|
|
||||||
|
def test_start_sidecar_passes_endpoint_and_provider_argv_separately(self):
|
||||||
|
server_args = SimpleNamespace(
|
||||||
|
sidecar="example.sidecar",
|
||||||
|
sidecar_args=[
|
||||||
|
"--sidecar-shutdown-timeout",
|
||||||
|
"42",
|
||||||
|
"--grpc-connections",
|
||||||
|
"2",
|
||||||
|
],
|
||||||
|
host="127.0.0.1",
|
||||||
|
grpc_port=50051,
|
||||||
|
)
|
||||||
|
with (
|
||||||
|
patch("sglang.srt.entrypoints.sidecar.mp.get_context") as get_context,
|
||||||
|
patch("sglang.srt.entrypoints.sidecar.Sidecar") as sidecar_class,
|
||||||
|
):
|
||||||
|
start_sidecar(server_args)
|
||||||
|
|
||||||
|
process_kwargs = get_context.return_value.Process.call_args.kwargs
|
||||||
|
self.assertEqual(process_kwargs["name"], "sglang_sidecar_example.sidecar")
|
||||||
|
self.assertEqual(process_kwargs["target"], _run_sidecar)
|
||||||
|
self.assertEqual(
|
||||||
|
process_kwargs["args"],
|
||||||
|
(
|
||||||
|
"example.sidecar",
|
||||||
|
["--grpc-connections", "2"],
|
||||||
|
"http://127.0.0.1:50051",
|
||||||
|
),
|
||||||
|
)
|
||||||
|
sidecar_class.assert_called_once_with(
|
||||||
|
get_context.return_value.Process.return_value,
|
||||||
|
"example.sidecar",
|
||||||
|
shutdown_timeout=42.0,
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_sidecar_requires_native_grpc(self):
|
||||||
|
sa = self._args(sidecar="example.sidecar")
|
||||||
|
with self.assertRaisesRegex(ValueError, "requires --grpc-port"):
|
||||||
|
sa._handle_deprecated_args()
|
||||||
|
|
||||||
|
def test_sidecar_rejects_legacy_grpc(self):
|
||||||
|
sa = self._args(sidecar="example.sidecar", smg_grpc_mode=True)
|
||||||
|
with self.assertRaisesRegex(ValueError, "native gRPC server"):
|
||||||
|
sa._handle_deprecated_args()
|
||||||
|
|
||||||
|
def test_sidecar_rejects_empty_value(self):
|
||||||
|
sa = self._args(sidecar="", grpc_port=50051)
|
||||||
|
with self.assertRaisesRegex(ValueError, "must not be empty"):
|
||||||
|
sa._handle_deprecated_args()
|
||||||
|
|
||||||
|
def test_sidecar_sets_endpoint_env_before_import_and_calls_main(self):
|
||||||
|
main = MagicMock()
|
||||||
|
|
||||||
|
def import_module(module_name):
|
||||||
|
self.assertEqual(module_name, "example.sidecar")
|
||||||
|
self.assertEqual(
|
||||||
|
os.environ[SGLANG_GRPC_ENDPOINT_ENV],
|
||||||
|
"http://127.0.0.1:50051",
|
||||||
|
)
|
||||||
|
self.assertEqual(os.environ["DYN_NAMESPACE"], "pluh")
|
||||||
|
return SimpleNamespace(main=main)
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch.dict(
|
||||||
|
os.environ,
|
||||||
|
{
|
||||||
|
SGLANG_GRPC_ENDPOINT_ENV: "http://stale.example:1",
|
||||||
|
"DYN_NAMESPACE": "pluh",
|
||||||
|
},
|
||||||
|
),
|
||||||
|
patch("sglang.srt.entrypoints.sidecar.kill_itself_when_parent_died"),
|
||||||
|
patch(
|
||||||
|
"sglang.srt.entrypoints.sidecar.importlib.import_module",
|
||||||
|
side_effect=import_module,
|
||||||
|
),
|
||||||
|
):
|
||||||
|
_run_sidecar(
|
||||||
|
"example.sidecar",
|
||||||
|
["--provider-flag", "value"],
|
||||||
|
"http://127.0.0.1:50051",
|
||||||
|
)
|
||||||
|
|
||||||
|
main.assert_called_once_with(["--provider-flag", "value"])
|
||||||
|
|
||||||
|
def test_sidecar_stop_uses_configured_shutdown_timeout(self):
|
||||||
|
proc = MagicMock(pid=1234)
|
||||||
|
proc.is_alive.side_effect = [True, True]
|
||||||
|
sidecar = Sidecar(
|
||||||
|
proc,
|
||||||
|
"example.sidecar",
|
||||||
|
shutdown_timeout=42.0,
|
||||||
|
)
|
||||||
|
|
||||||
|
with patch("sglang.srt.entrypoints.sidecar.kill_process_tree") as kill_tree:
|
||||||
|
sidecar.stop()
|
||||||
|
|
||||||
|
proc.terminate.assert_called_once_with()
|
||||||
|
proc.join.assert_called_once_with(timeout=42.0)
|
||||||
|
kill_tree.assert_called_once_with(1234, wait_timeout=42.0)
|
||||||
|
|
||||||
def test_legacy_smg_derives_grpc_port_from_http_port(self):
|
def test_legacy_smg_derives_grpc_port_from_http_port(self):
|
||||||
sa = self._args(port=30000, smg_grpc_mode=True)
|
sa = self._args(port=30000, smg_grpc_mode=True)
|
||||||
sa._handle_deprecated_args()
|
sa._handle_deprecated_args()
|
||||||
|
|||||||
Reference in New Issue
Block a user