Fail-fast on PD subprocess exit and scheduler exception (#26298)

python/sglang/srt/environ.py

@@ -294,6 +294,7 @@ class Envs:
     SGLANG_DYNAMIC_CHUNKING_SMOOTH_FACTOR = EnvFloat(0.75)
     SGLANG_SCHEDULER_SKIP_ALL_GATHER = EnvBool(False)
     SGLANG_SCHEDULER_DECREASE_PREFILL_IDLE = EnvBool(False)
+    SGLANG_KILLPG_ON_SCHEDULER_EXCEPTION = EnvBool(False)
     SGLANG_PREFILL_DELAYER_MAX_DELAY_PASSES = EnvInt(None)
     SGLANG_PREFILL_DELAYER_TOKEN_USAGE_LOW_WATERMARK = EnvFloat(None)
     SGLANG_DATA_PARALLEL_BUDGET_INTERVAL = EnvInt(1)

python/sglang/srt/managers/scheduler.py

@@ -3841,6 +3841,13 @@ def run_scheduler_process(
         traceback = get_exception_traceback()
         logger.error(f"Scheduler hit an exception: {traceback}")
         parent_process.send_signal(signal.SIGQUIT)
+        # Opt-in: SIGKILL the pgroup so sibling ranks don't spew thousands
+        # of NCCL/TCPStore tracebacks before they finally die.
+        if envs.SGLANG_KILLPG_ON_SCHEDULER_EXCEPTION.get():
+            try:
+                os.killpg(os.getpgrp(), signal.SIGKILL)
+            except Exception:
+                pass
     finally:
         if scheduler is not None:
             # FPM has a background ZMQ publisher thread that needs explicit

python/sglang/test/server_fixtures/disaggregation_fixture.py

@@ -14,6 +14,7 @@ from sglang.test.test_utils import (
     is_in_ci,
     popen_launch_pd_server,
     popen_with_error_check,
+    start_subprocess_fail_fast_watcher,
 )
 from sglang.utils import wait_for_http_ready
 
@@ -39,6 +40,7 @@ class PDDisaggregationServerBase(CustomTestCase):
             f"{cls.base_host=} {cls.lb_port=} {cls.prefill_port=} {cls.decode_port=} {cls.bootstrap_port=}"
         )
         cls.process_lb, cls.process_decode, cls.process_prefill = None, None, None
+        cls._fail_fast_stop = None
 
         # config transfer backend and rdma devices
         if is_in_ci():
@@ -110,6 +112,13 @@ class PDDisaggregationServerBase(CustomTestCase):
         cls.wait_server_ready(cls.prefill_url + "/health", process=cls.process_prefill)
         cls.wait_server_ready(cls.decode_url + "/health", process=cls.process_decode)
         cls.launch_lb()
+        cls._fail_fast_stop = start_subprocess_fail_fast_watcher(
+            [
+                ("prefill", cls.process_prefill),
+                ("decode", cls.process_decode),
+                ("lb", cls.process_lb),
+            ]
+        )
 
     @classmethod
     def launch_lb(cls):
@@ -141,6 +150,11 @@ class PDDisaggregationServerBase(CustomTestCase):
 
     @classmethod
     def tearDownClass(cls):
+        # Stop the watcher BEFORE killing processes: kill_process_tree
+        # below makes them exit with a negative signal rc, which would
+        # otherwise trip the watcher and os._exit out of pytest mid-teardown.
+        if cls._fail_fast_stop is not None:
+            cls._fail_fast_stop.set()
         os.environ.pop("MC_TCP_ENABLE_CONNECTION_POOL")
         for process in [cls.process_lb, cls.process_decode, cls.process_prefill]:
             if process:

python/sglang/test/test_utils.py

@@ -574,6 +574,41 @@ def popen_with_error_check(command: list[str]):
     return process
 
 
+def start_subprocess_fail_fast_watcher(
+    named_procs: list[tuple[str, subprocess.Popen]],
+) -> threading.Event:
+    """Abort the test runner the moment any watched subprocess exits non-zero.
+
+    Caller must `.set()` the returned Event before intentional teardown."""
+    stop = threading.Event()
+
+    def watcher():
+        while not stop.is_set():
+            for name, proc in named_procs:
+                rc = proc.poll() if proc else None
+                if rc is None or rc == 0:
+                    continue
+                if stop.is_set():
+                    return
+                sys.stderr.write(
+                    f"[FIXTURE FAIL-FAST] {name} (pid={proc.pid}) exited "
+                    f"rc={rc}; aborting.\n"
+                )
+                sys.stderr.flush()
+                for _, sib in named_procs:
+                    if sib and sib is not proc:
+                        try:
+                            kill_process_tree(sib.pid, wait_timeout=10)
+                        except Exception:
+                            pass
+                # POSIX: signal N -> 128+N (os._exit masks negatives via & 0xff).
+                os._exit(rc if rc >= 0 else 128 + (-rc))
+            time.sleep(0.1)
+
+    threading.Thread(target=watcher, daemon=True, name="SubprocFailFastWatcher").start()
+    return stop
+
+
 def _try_enable_offline_mode_if_cache_complete(
     model_name_or_path: str, env: dict, other_args: Optional[list[str]] = None
 ) -> Optional[str]: