Tiny add stuck simulation (#15613)

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)

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):

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):

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(

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

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"),
     ],
 }
 

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()