[CI] feat: add early exit to wait_for_server when process dies (#18602)

python/sglang/test/server_fixtures/disaggregation_fixture.py

@@ -5,8 +5,6 @@ import time
 import warnings
 from urllib.parse import urlparse
 
-import requests
-
 from sglang.srt.environ import envs
 from sglang.srt.utils import kill_process_tree
 from sglang.test.test_utils import (
@@ -16,6 +14,7 @@ from sglang.test.test_utils import (
     is_in_ci,
     popen_with_error_check,
 )
+from sglang.utils import wait_for_http_ready
 
 logger = logging.getLogger(__name__)
 
@@ -72,23 +71,14 @@ class PDDisaggregationServerBase(CustomTestCase):
         ]
         print("Starting load balancer:", shlex.join(lb_command))
         cls.process_lb = popen_with_error_check(lb_command)
-        cls.wait_server_ready(cls.lb_url + "/health")
+        cls.wait_server_ready(cls.lb_url + "/health", process=cls.process_lb)
 
     @classmethod
-    def wait_server_ready(cls, url, timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH):
-        start_time = time.perf_counter()
-        while True:
-            try:
-                response = requests.get(url)
-                if response.status_code == 200:
-                    print(f"Server {url} is ready")
-                    return
-            except Exception:
-                pass
-
-            if time.perf_counter() - start_time > timeout:
-                raise RuntimeError(f"Server {url} failed to start in {timeout}s")
-            time.sleep(1)
+    def wait_server_ready(
+        cls, url, timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH, process=None
+    ):
+        wait_for_http_ready(url=url, timeout=timeout, process=process)
+        print(f"Server {url} is ready")
 
     @classmethod
     def tearDownClass(cls):

python/sglang/utils.py

@@ -464,37 +464,82 @@ def terminate_process(process):
         release_port(lock_socket)
 
 
-def wait_for_server(base_url: str, timeout: int = None) -> None:
+def _raise_if_process_exited(process: Optional[Any]) -> None:
+    if process is None:
+        return
+
+    if hasattr(process, "poll"):
+        return_code = process.poll()
+        if return_code is not None:
+            raise RuntimeError(f"Server process exited with code {return_code}")
+        return
+
+    if hasattr(process, "is_alive") and not process.is_alive():
+        return_code = getattr(process, "exitcode", None)
+        if return_code is None:
+            raise RuntimeError("Server process exited")
+        raise RuntimeError(f"Server process exited with code {return_code}")
+
+
+def _is_wait_timeout(start_time: float, timeout: Optional[int]) -> bool:
+    if timeout is None:
+        return False
+    return time.perf_counter() - start_time > timeout
+
+
+def wait_for_http_ready(
+    url: str,
+    timeout: Optional[int] = None,
+    process: Optional[Any] = None,
+    headers: Optional[dict] = None,
+    request_timeout: int = 5,
+) -> None:
+    """Wait for an HTTP endpoint to return status 200."""
+    start_time = time.perf_counter()
+    while True:
+        _raise_if_process_exited(process)
+        try:
+            response = requests.get(url, headers=headers, timeout=request_timeout)
+            if response.status_code == 200:
+                return
+        except requests.exceptions.RequestException:
+            _raise_if_process_exited(process)
+
+        if _is_wait_timeout(start_time, timeout):
+            raise TimeoutError(
+                f"Endpoint {url} did not become ready within timeout period"
+            )
+        time.sleep(1)
+
+
+def wait_for_server(
+    base_url: str,
+    timeout: int = None,
+    process: Optional[subprocess.Popen] = None,
+) -> None:
     """Wait for the server to be ready by polling the /v1/models endpoint.
 
     Args:
-        base_url: The base URL of the server
+        base_url: The base URL of the server.
         timeout: Maximum time to wait in seconds. None means wait forever.
+        process: Optional server process used for early-exit checks.
     """
-    start_time = time.perf_counter()
-    while True:
-        try:
-            response = requests.get(
-                f"{base_url}/v1/models",
-                headers={"Authorization": "Bearer None"},
-            )
-            if response.status_code == 200:
-                time.sleep(5)
-                print_highlight(
-                    """\n
-                    NOTE: Typically, the server runs in a separate terminal.
-                    In this notebook, we run the server and notebook code together, so their outputs are combined.
-                    To improve clarity, the server logs are displayed in the original black color, while the notebook outputs are highlighted in blue.
-                    To reduce the log length, we set the log level to warning for the server, the default log level is info.
-                    We are running those notebooks in a CI environment, so the throughput is not representative of the actual performance.
-                    """
-                )
-                break
-
-            if timeout and time.perf_counter() - start_time > timeout:
-                raise TimeoutError("Server did not become ready within timeout period")
-        except requests.exceptions.RequestException:
-            time.sleep(1)
+    wait_for_http_ready(
+        url=f"{base_url}/v1/models",
+        timeout=timeout,
+        process=process,
+        headers={"Authorization": "Bearer None"},
+    )
+    time.sleep(5)
+    print_highlight(
+        """\n
+        NOTE: Typically, the server runs in a separate terminal.
+        In this notebook, we run the server and notebook code together, so their outputs are combined.
+        To improve clarity, the server logs are displayed in the original black color, while the notebook outputs are highlighted in blue.
+        To reduce the log length, we set the log level to warning for the server, the default log level is info.
+        We are running those notebooks in a CI environment, so the throughput is not representative of the actual performance.
+        """
+    )
 
 
 class TypeBasedDispatcher: