Graceful shutdown with SIGTERM for child processes

python/sglang/srt/entrypoints/http_server.py

@@ -155,7 +155,11 @@
     read_from_shared_memory,
     write_data_for_multi_tokenizer,
 )
-from sglang.srt.managers.tokenizer_manager import ServerStatus, TokenizerManager
+from sglang.srt.managers.tokenizer_manager import (
+    ServerStatus,
+    TokenizerManager,
+    shutdown_scheduler_and_child_processes,
+)
 from sglang.srt.observability.func_timer import enable_func_timer
 from sglang.srt.observability.trace import (
     process_tracing_init,
@@ -2664,6 +2668,11 @@ async def _run_with_ssl_refresh():
                 )
     finally:
         if server_args.tokenizer_worker_num > 1:
+            if subprocess_watchdog is not None:
+                subprocess_watchdog.stop()
+            shutdown_scheduler_and_child_processes(
+                tokenizer_manager.dispatch_scheduler_shutdown
+            )
             if multi_tokenizer_args_shm is not None:
                 multi_tokenizer_args_shm.unlink()
             if _global_state is not None:

python/sglang/srt/environ.py

@@ -364,6 +364,8 @@ class Envs:
     # else /tmp); see debug_utils/cuda_coredump.py.
     SGLANG_CUDA_COREDUMP_DIR = EnvStr(None)
     SGLANG_FORCE_SHUTDOWN = EnvBool(False)
+    SGLANG_CHILD_PROCESS_SHUTDOWN_TIMEOUT = EnvFloat(10.0)
+    SGLANG_SCHEDULER_SHUTDOWN_TIMEOUT = EnvFloat(15.0)
     SGLANG_PYSPY_DUMP_BEFORE_CRASH = EnvBool(True)
     SGLANG_CUDA_COREDUMP_BEFORE_CRASH = EnvBool(True)
     SGLANG_CUDA_COREDUMP_BEFORE_CRASH_WAIT_SECS = EnvFloat(60.0)

python/sglang/srt/managers/data_parallel_controller.py

@@ -58,6 +58,8 @@
 from sglang.srt.utils import numa_utils
 from sglang.srt.utils.common import (
     configure_logger,
+    graceful_kill_process_tree,
+    install_graceful_sigterm_handler,
     kill_itself_when_parent_died,
     maybe_reindex_device_id,
 )
@@ -820,6 +822,16 @@ def run_data_parallel_controller_process(
     kill_itself_when_parent_died()
     parent_process = psutil.Process().parent()
 
+    # SIGTERM → propagate graceful shutdown to scheduler children, which would
+    # otherwise be SIGKILLed via PR_SET_PDEATHSIG before they finish cleanup.
+    install_graceful_sigterm_handler(
+        logger,
+        "data_parallel_controller",
+        on_shutdown=lambda: graceful_kill_process_tree(
+            timeout=envs.SGLANG_CHILD_PROCESS_SHUTDOWN_TIMEOUT.get()
+        ),
+    )
+
     # This process reads the config namespaces before spawning schedulers.
     publish(server_args, role="dp_controller")
     configure_logger(server_args)

python/sglang/srt/managers/multi_tokenizer_mixin.py

@@ -37,6 +37,7 @@
 import zmq.asyncio
 
 from sglang.srt.disaggregation.utils import TransferBackend
+from sglang.srt.environ import envs
 from sglang.srt.managers.disagg_service import start_disagg_service
 from sglang.srt.managers.io_struct import (
     BaseBatchReq,
@@ -48,6 +49,7 @@
     FreezeGCReq,
     PauseContinueBroadcastReq,
     PauseGenerationReqInput,
+    ShutdownReq,
     TokenizerWorkerRegistrationReq,
     async_sock_recv,
     async_sock_send,
@@ -480,6 +482,13 @@ def __init__(
         # Shared socket mapping (both coroutines run on self._loop, so safe)
         self.socket_mapping = SocketMapping()
 
+    def dispatch_scheduler_shutdown(self) -> None:
+        future = asyncio.run_coroutine_threadsafe(
+            async_sock_send(self.send_to_scheduler, ShutdownReq()),
+            self._loop,
+        )
+        future.result(timeout=envs.SGLANG_SCHEDULER_SHUTDOWN_TIMEOUT.get())
+
     def set_startup_time(self, startup_time: Dict[str, Any]) -> None:
         self.startup_time = startup_time
 

python/sglang/srt/managers/scheduler.py

@@ -310,7 +310,7 @@
     suppress_other_loggers,
     triton_load_watch,
 )
-from sglang.srt.utils.common import is_npu
+from sglang.srt.utils.common import install_graceful_sigterm_handler, is_npu
 from sglang.srt.utils.hf_transformers_utils import (
     get_processor,
     get_tokenizer,
@@ -5021,6 +5021,18 @@ def run_scheduler_process(
     # worker's own publish.
     publish(server_args, role="scheduler")
     parent_process = psutil.Process().parent()
+    scheduler = None
+
+    install_graceful_sigterm_handler(
+        logger,
+        f"scheduler process (TP{tp_rank} PP{pp_rank})",
+        on_shutdown=lambda: (
+            setattr(scheduler, "gracefully_exit", True)
+            if scheduler is not None
+            else None
+        ),
+        is_shutting_down=lambda: (scheduler is not None and scheduler.gracefully_exit),
+    )
 
     # Set up tracing
     if server_args.enable_trace:
@@ -5037,7 +5049,6 @@ def run_scheduler_process(
         trace_set_thread_info(thread_label, tp_rank, dp_rank, pp_rank)
 
     # Create a scheduler and run the event loop
-    scheduler = None
     try:
         scheduler = Scheduler(
             server_args,

python/sglang/srt/managers/tokenizer_manager.py

@@ -34,7 +34,17 @@
 from enum import Enum
 from functools import lru_cache
 from http import HTTPStatus
-from typing import Any, Awaitable, Dict, Iterable, List, Optional, Tuple, Union
+from typing import (
+    Any,
+    Awaitable,
+    Callable,
+    Dict,
+    Iterable,
+    List,
+    Optional,
+    Tuple,
+    Union,
+)
 
 import fastapi
 import numpy as np
@@ -132,6 +142,7 @@
     freeze_gc,
     get_bool_env_var,
     get_or_create_event_loop,
+    graceful_kill_process_tree,
     kill_process_tree,
 )
 from sglang.srt.utils.aio_rwlock import RWLock
@@ -159,6 +170,25 @@
 logger = logging.getLogger(__name__)
 
 
+def shutdown_scheduler_and_child_processes(
+    dispatch_shutdown: Callable[[], None],
+) -> None:
+    try:
+        dispatch_shutdown()
+    except Exception:
+        logger.exception("Failed to dispatch scheduler shutdown request")
+    else:
+        deadline = time.monotonic() + envs.SGLANG_SCHEDULER_SHUTDOWN_TIMEOUT.get()
+        while time.monotonic() < deadline and collect_scheduler_processes():
+            time.sleep(0.1)
+
+    graceful_kill_process_tree(
+        os.getpid(),
+        include_parent=False,
+        timeout=envs.SGLANG_CHILD_PROCESS_SHUTDOWN_TIMEOUT.get(),
+    )
+
+
 def _reject_missing_dispatched_encoder_embedding(server_args, request_obj, mm_inputs):
     """Do not silently turn a failed EPD request into local vision work."""
     if (
@@ -3128,15 +3158,17 @@ async def sigterm_watchdog(self):
                 )
                 self.dump_requests_before_crash()
                 self.force_exit_handler()
-                break
+                kill_process_tree(os.getpid(), include_parent=True)
+                sys.exit(0)
 
             elif get_bool_env_var("SGL_FORCE_SHUTDOWN"):
                 # if force shutdown flag set, exit immediately
                 logger.error(
                     "Signal SIGTERM received while force shutdown flag set. Force exiting."
                 )
                 self.force_exit_handler()
-                break
+                kill_process_tree(os.getpid(), include_parent=True)
+                sys.exit(0)
 
             logger.info(
                 f"Gracefully exiting... Remaining number of requests {remain_num_req}. Remaining requests {remaining_rids=}."
@@ -3146,17 +3178,18 @@ async def sigterm_watchdog(self):
             else:
                 break
 
-        # Stop the watchdog: child exits are expected during shutdown, not crashes.
+        # Child exits are expected after this point.
         if self._subprocess_watchdog is not None:
             self._subprocess_watchdog.stop()
-        # Ask schedulers to release resources in userspace and exit (see
-        # ShutdownReq), then wait for them before hard-killing the rest.
-        self._dispatch_to_scheduler(ShutdownReq())
-        deadline = time.monotonic() + 15
-        while time.monotonic() < deadline and collect_scheduler_processes():
-            time.sleep(0.1)
-        kill_process_tree(os.getpid(), include_parent=True)
-        sys.exit(0)
+        if self.server_args.tokenizer_worker_num == 1:
+            # Do not disable the soft watchdog here: handle_loop already holds
+            # that context while parked on recv, and disable() is not reentrant.
+            shutdown_scheduler_and_child_processes(
+                lambda: self._dispatch_to_scheduler(ShutdownReq())
+            )
+
+        # SystemExit is absorbed by the task wrapper.
+        os._exit(0)
 
     def force_exit_handler(self):
         """Put some custom force exit logic here."""

python/sglang/srt/utils/common.py

@@ -2127,6 +2127,108 @@ def kill_process_tree(
         _wait_for_reap_or_raise(killed, wait_timeout)
 
 
+def graceful_kill_process_tree(
+    parent_pid=None,
+    include_parent: bool = False,
+    skip_pid: int = None,
+    timeout: float = 10.0,
+):
+    """Gracefully terminate a process tree: SIGTERM first, wait, then SIGKILL stragglers.
+
+    Args:
+        parent_pid: Target PID, defaults to current process.
+        include_parent: Also kill the parent after children are done.
+        skip_pid: A child PID to leave untouched.
+        timeout: Seconds to wait before escalating to SIGKILL.
+    """
+    if parent_pid is None:
+        parent_pid = os.getpid()
+        include_parent = False
+
+    try:
+        itself = psutil.Process(parent_pid)
+    except psutil.NoSuchProcess:
+        return
+
+    children = itself.children(recursive=True)
+
+    signaled = []
+    for child in children:
+        if child.pid == skip_pid:
+            continue
+        try:
+            logger.info(
+                f"Sending SIGTERM to child process {child.pid} ({child.name()})"
+            )
+            child.terminate()
+            signaled.append(child)
+        except psutil.NoSuchProcess:
+            pass
+
+    if signaled:
+        logger.info(
+            f"Waiting up to {timeout}s for {len(signaled)} child process(es) "
+            "to terminate gracefully..."
+        )
+        deadline = time.monotonic() + timeout
+        alive = signaled
+        while True:
+            alive = _still_holding_resources(alive)
+            if not alive or time.monotonic() >= deadline:
+                break
+            time.sleep(0.1)
+
+        if alive:
+            logger.warning(
+                f"{len(alive)} child process(es) did not terminate within "
+                f"{timeout}s, sending SIGKILL: pids={[p.pid for p in alive]}"
+            )
+            for child in alive:
+                try:
+                    child.kill()
+                except psutil.NoSuchProcess:
+                    pass
+            kill_deadline = time.monotonic() + 3
+            while _still_holding_resources(alive) and time.monotonic() < kill_deadline:
+                time.sleep(0.1)
+        else:
+            logger.info("All child processes terminated gracefully.")
+
+    if include_parent:
+        try:
+            itself.kill()
+        except psutil.NoSuchProcess:
+            pass
+
+
+def install_graceful_sigterm_handler(
+    logger: logging.Logger,
+    label: str,
+    on_shutdown: Optional[Callable[[], None]] = None,
+    is_shutting_down: Optional[Callable[[], bool]] = None,
+):
+    """Install a SIGTERM handler that unwinds through process cleanup.
+
+    An already-running cleanup is left uninterrupted. Other exits remain nonzero
+    so an active SubprocessWatchdog still treats them as unexpected.
+    """
+
+    def _handler(signum, frame):
+        signal.signal(signal.SIGTERM, signal.SIG_IGN)
+        if is_shutting_down is not None and is_shutting_down():
+            logger.info("SIGTERM received in %s while cleanup is in progress.", label)
+            return
+        logger.info("SIGTERM received in %s; starting cleanup before exit.", label)
+        if on_shutdown is not None:
+            try:
+                on_shutdown()
+            except Exception:
+                logger.exception("Error in SIGTERM cleanup hook")
+        sys.exit(128 + signum)
+
+    signal.signal(signal.SIGTERM, _handler)
+
+
 def monkey_patch_p2p_access_check():
     """
     Monkey patch the slow p2p access check.

test/registered/unit/managers/test_multi_tokenizer_mixin.py

@@ -1,12 +1,17 @@
-import unittest
+import sys
+from unittest.mock import Mock, sentinel
+
+import pytest
 
 from sglang.test.ci.ci_register import register_cpu_ci
-from sglang.test.test_utils import maybe_stub_sgl_kernel
+from sglang.test.test_utils import CustomTestCase, maybe_stub_sgl_kernel
 
 maybe_stub_sgl_kernel()
 
+from sglang.srt.environ import envs
 from sglang.srt.managers.io_struct import BatchStrOutput
 from sglang.srt.managers.multi_tokenizer_mixin import (
+    MultiTokenizerRouter,
     TokenizerWorker,
     _handle_output_by_index,
     get_tokenizer_worker_class,
@@ -79,7 +84,7 @@ def _make_batch_str_output() -> BatchStrOutput:
     )
 
 
-class TestMultiTokenizerMixin(unittest.TestCase):
+class TestMultiTokenizerMixin(CustomTestCase):
     def test_batch_str_output_preserves_cached_tokens_details(self):
         output = _make_batch_str_output()
 
@@ -105,6 +110,31 @@ def test_get_tokenizer_worker_class_rejects_non_worker(self):
         with self.assertRaisesRegex(TypeError, "TokenizerWorker"):
             get_tokenizer_worker_class(InvalidServerArgs())
 
+    def test_router_dispatches_shutdown_on_its_event_loop(self):
+        router = MultiTokenizerRouter.__new__(MultiTokenizerRouter)
+        router.send_to_scheduler = sentinel.socket
+        router._loop = sentinel.loop
+        send = Mock(return_value=sentinel.coroutine)
+        completed = Mock()
+        submit = Mock(return_value=completed)
+
+        with (
+            envs.SGLANG_SCHEDULER_SHUTDOWN_TIMEOUT.override(2.0),
+            pytest.MonkeyPatch.context() as monkeypatch,
+        ):
+            monkeypatch.setattr(
+                "sglang.srt.managers.multi_tokenizer_mixin.async_sock_send", send
+            )
+            monkeypatch.setattr(
+                "sglang.srt.managers.multi_tokenizer_mixin.asyncio.run_coroutine_threadsafe",
+                submit,
+            )
+            router.dispatch_scheduler_shutdown()
+
+        send.assert_called_once()
+        submit.assert_called_once_with(sentinel.coroutine, sentinel.loop)
+        completed.result.assert_called_once_with(timeout=2.0)
+
 
 if __name__ == "__main__":
-    unittest.main()
+    sys.exit(pytest.main([__file__]))