From d5431ff89434c80a82cb22a749a141d12d8d295f Mon Sep 17 00:00:00 2001 From: fzyzcjy <5236035+fzyzcjy@users.noreply.github.com> Date: Mon, 22 Dec 2025 17:00:18 +0800 Subject: [PATCH] Tiny add stuck simulation (#15613) --- python/sglang/srt/environ.py | 3 + .../srt/managers/data_parallel_controller.py | 2 + .../srt/managers/detokenizer_manager.py | 2 + .../sglang/srt/managers/tokenizer_manager.py | 1 + python/sglang/srt/utils/watchdog.py | 20 +++++ test/srt/run_suite.py | 1 + test/srt/test_soft_watchdog.py | 74 +++++++++++++++++++ 7 files changed, 103 insertions(+) create mode 100644 test/srt/test_soft_watchdog.py diff --git a/python/sglang/srt/environ.py b/python/sglang/srt/environ.py index 21f8e16f8..e9cb4518d 100644 --- a/python/sglang/srt/environ.py +++ b/python/sglang/srt/environ.py @@ -150,6 +150,9 @@ class Envs: # Test & Debug SGLANG_IS_IN_CI = EnvBool(False) SGLANG_IS_IN_CI_AMD = EnvBool(False) + SGLANG_TEST_STUCK_DETOKENIZER = EnvFloat(0) + SGLANG_TEST_STUCK_DP_CONTROLLER = EnvFloat(0) + SGLANG_TEST_STUCK_TOKENIZER = EnvFloat(0) IS_BLACKWELL = EnvBool(False) SGLANG_SET_CPU_AFFINITY = EnvBool(False) SGLANG_PROFILE_WITH_STACK = EnvBool(True) diff --git a/python/sglang/srt/managers/data_parallel_controller.py b/python/sglang/srt/managers/data_parallel_controller.py index 128dcfe32..ae8c2111a 100644 --- a/python/sglang/srt/managers/data_parallel_controller.py +++ b/python/sglang/srt/managers/data_parallel_controller.py @@ -27,6 +27,7 @@ import psutil import setproctitle import zmq +from sglang.srt.environ import envs from sglang.srt.layers.dp_attention import compute_dp_attention_world_info from sglang.srt.managers.io_struct import ( BlockReqInput, @@ -178,6 +179,7 @@ class DataParallelController: debug_name="DataParallelController", watchdog_timeout=server_args.soft_watchdog_timeout, soft=True, + test_stuck_time=envs.SGLANG_TEST_STUCK_DP_CONTROLLER.get(), ) def send_to_all_workers(self, obj): diff --git a/python/sglang/srt/managers/detokenizer_manager.py b/python/sglang/srt/managers/detokenizer_manager.py index 3ff06fe63..60fb1a9c2 100644 --- a/python/sglang/srt/managers/detokenizer_manager.py +++ b/python/sglang/srt/managers/detokenizer_manager.py @@ -25,6 +25,7 @@ import pybase64 import setproctitle import zmq +from sglang.srt.environ import envs from sglang.srt.managers.io_struct import ( BatchEmbeddingOutput, BatchMultimodalDecodeReq, @@ -116,6 +117,7 @@ class DetokenizerManager(MultiHttpWorkerDetokenizerMixin): debug_name="DetokenizerManager", watchdog_timeout=server_args.soft_watchdog_timeout, soft=True, + test_stuck_time=envs.SGLANG_TEST_STUCK_DETOKENIZER.get(), ) def event_loop(self): diff --git a/python/sglang/srt/managers/tokenizer_manager.py b/python/sglang/srt/managers/tokenizer_manager.py index 41ef9822f..904cdc2ed 100644 --- a/python/sglang/srt/managers/tokenizer_manager.py +++ b/python/sglang/srt/managers/tokenizer_manager.py @@ -409,6 +409,7 @@ class TokenizerManager(TokenizerCommunicatorMixin, TokenizerManagerMultiItemMixi debug_name="TokenizerManager", watchdog_timeout=server_args.soft_watchdog_timeout, soft=True, + test_stuck_time=envs.SGLANG_TEST_STUCK_TOKENIZER.get(), ) async def generate_request( diff --git a/python/sglang/srt/utils/watchdog.py b/python/sglang/srt/utils/watchdog.py index 2bbdcffbe..36ddd7c57 100644 --- a/python/sglang/srt/utils/watchdog.py +++ b/python/sglang/srt/utils/watchdog.py @@ -21,13 +21,18 @@ class Watchdog: debug_name: str, watchdog_timeout: Optional[float], soft: bool = False, + test_stuck_time: float = 0, ) -> Watchdog: if watchdog_timeout is None: + assert ( + test_stuck_time == 0 + ), f"stuck tester can be enabled only if soft watchdog is enabled." return _WatchdogNoop() return _WatchdogReal( debug_name=debug_name, watchdog_timeout=watchdog_timeout, soft=soft, + test_stuck_time=test_stuck_time, ) def feed(self): @@ -44,9 +49,11 @@ class _WatchdogReal(Watchdog): debug_name: str, watchdog_timeout: float, soft: bool = False, + test_stuck_time: float = 0, ): self._counter = 0 self._active = True + self._test_stuck_time = test_stuck_time self._raw = WatchdogRaw( debug_name=debug_name, get_counter=lambda: self._counter, @@ -55,8 +62,21 @@ class _WatchdogReal(Watchdog): soft=soft, ) logger.info(f"Watchdog {self._raw.debug_name} initialized.") + if self._test_stuck_time > 0: + logger.info( + f"Watchdog {self._raw.debug_name} is configured to use {test_stuck_time=}." + ) def feed(self): + if self._test_stuck_time > 0: + logger.info( + f"Watchdog {self._raw.debug_name} start deliberately stuck for {self._test_stuck_time}s" + ) + time.sleep(self._test_stuck_time) + logger.info( + f"Watchdog {self._raw.debug_name} end deliberately stuck for {self._test_stuck_time}s" + ) + self._counter += 1 @contextmanager diff --git a/test/srt/run_suite.py b/test/srt/run_suite.py index 34be3572a..5291f9fea 100644 --- a/test/srt/run_suite.py +++ b/test/srt/run_suite.py @@ -209,6 +209,7 @@ suites = { TestFile("test_profile_v2.py"), TestFile("models/test_ministral3_models.py"), TestFile("test_mistral_large3_basic.py"), + TestFile("test_soft_watchdog.py"), ], } diff --git a/test/srt/test_soft_watchdog.py b/test/srt/test_soft_watchdog.py new file mode 100644 index 000000000..b824de4bf --- /dev/null +++ b/test/srt/test_soft_watchdog.py @@ -0,0 +1,74 @@ +import io +import unittest + +import requests + +from sglang.srt.environ import envs +from sglang.srt.utils import kill_process_tree +from sglang.test.test_utils import ( + DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH, + DEFAULT_URL_FOR_TEST, + CustomTestCase, + popen_launch_server, +) + + +class BaseTestSoftWatchdog: + env_override = None + expected_message = None + + @classmethod + def setUpClass(cls): + cls.stdout = io.StringIO() + cls.stderr = io.StringIO() + + with cls.env_override(): + cls.process = popen_launch_server( + "Qwen/Qwen3-0.6B", + DEFAULT_URL_FOR_TEST, + timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH, + other_args=[ + "--soft-watchdog-timeout", + "20", + "--skip-server-warmup", + ], + return_stdout_stderr=(cls.stdout, cls.stderr), + ) + + @classmethod + def tearDownClass(cls): + kill_process_tree(cls.process.pid) + cls.stdout.close() + cls.stderr.close() + + def test_watchdog_triggers(self): + print("Start call /generate API", flush=True) + try: + requests.post( + DEFAULT_URL_FOR_TEST + "/generate", + json={ + "text": "Hello, please repeat this sentence for 1000 times.", + "sampling_params": {"max_new_tokens": 100, "temperature": 0}, + }, + timeout=30, + ) + except requests.exceptions.ReadTimeout as e: + print(f"requests.post timeout (but expected): {e}") + print("End call /generate API", flush=True) + + combined_output = self.stdout.getvalue() + self.stderr.getvalue() + self.assertIn(self.expected_message, combined_output) + + +class TestSoftWatchdogDetokenizer(BaseTestSoftWatchdog, CustomTestCase): + env_override = lambda: envs.SGLANG_TEST_STUCK_DETOKENIZER.override(30) + expected_message = "DetokenizerManager watchdog timeout" + + +class TestSoftWatchdogTokenizer(BaseTestSoftWatchdog, CustomTestCase): + env_override = lambda: envs.SGLANG_TEST_STUCK_TOKENIZER.override(30) + expected_message = "TokenizerManager watchdog timeout" + + +if __name__ == "__main__": + unittest.main()