[npu] Combine NPU test fixes from #35472 and #34516 (#36180)

python/sglang/test/ascend/e2e/test_npu_accuracy_utils.py

@@ -5,7 +5,6 @@ import logging
 import os
 import re
 import shutil
-import signal
 import subprocess
 import sys
 import threading
@@ -17,6 +16,7 @@ from sglang.srt.utils import kill_process_tree
 from sglang.test.ascend.e2e.test_npu_multi_node_utils import (
     SERVICE_PORT,
     check_role,
+    kill_process_group,
     launch_pd_mix_node,
     launch_pd_separation_node,
     launch_router,
@@ -105,21 +105,6 @@ def get_max_retries(datasets):
     return 1
 
 
-def _kill_evalscope_session(process):
-    """Kill the whole evalscope session (process.pid is the leader == pgid)."""
-    if process is None:
-        return
-    try:
-        os.killpg(process.pid, signal.SIGKILL)
-        logger.info(f"run_evalscope killed session pgid={process.pid}")
-    except ProcessLookupError:
-        logger.info(f"run_evalscope session pgid={process.pid} already gone")
-    except PermissionError:
-        logger.warning(
-            f"run_evalscope no permission to kill session pgid={process.pid}"
-        )
-
-
 def run_evalscope(
     host,
     port,
@@ -215,7 +200,7 @@ def run_evalscope(
             f"returncode={process.returncode}"
         )
 
-        _kill_evalscope_session(process)
+        kill_process_group(process)
 
         if process.returncode != 0:
             logger.error(f"Command failed with return code: {process.returncode}")
@@ -272,14 +257,14 @@ def run_evalscope(
             logger.warning("Process did not terminate gracefully, killing it...")
             process.kill()
             logger.info("Process killed")
-        _kill_evalscope_session(process)
+        kill_process_group(process)
         raise
 
     except Exception as e:
         logger.error(f"Error executing command: {e}")
         process.terminate()
         process.wait(timeout=5)
-        _kill_evalscope_session(process)
+        kill_process_group(process)
         raise
 
 

python/sglang/test/ascend/e2e/test_npu_multi_node_utils.py

@@ -1,6 +1,7 @@
 import logging
 import os
 import re
+import signal
 import socket
 import subprocess
 import threading
@@ -1079,3 +1080,22 @@ class TestNpuMultiNodePdSepTestCaseBase(CustomTestCase):
             expect_accuracy,
             f'Accuracy is {str(metrics["accuracy"])}, is lower than {expect_accuracy}',
         )
+
+
+def kill_process_group(process):
+    """SIGKILL the whole process group led by ``process.pid`` (== pgid).
+
+    ``process`` is a ``subprocess.Popen`` launched with ``start_new_session=True``,
+    so its pid equals the process-group id. A ``None`` process is ignored.
+    """
+    if process is None:
+        return
+    try:
+        os.killpg(process.pid, signal.SIGKILL)
+        logger.info(f"killed process group pgid={process.pid}")
+    except ProcessLookupError:
+        logger.info(f"process group pgid={process.pid} already gone")
+    except PermissionError:
+        logger.warning(f"no permission to kill process group pgid={process.pid}")
+    except OSError as e:
+        logger.warning(f"failed to kill process group pgid={process.pid}: {e}")

python/sglang/test/ascend/e2e/test_npu_performance_utils.py

@@ -26,6 +26,7 @@ from sglang.test.ascend.e2e.test_npu_multi_node_utils import (
     NAMESPACE,
     SERVICE_PORT,
     check_role,
+    kill_process_group,
     launch_pd_mix_node,
     launch_pd_separation_node,
     launch_router,
@@ -205,6 +206,9 @@ MAX_SERVER_KEEP_ALIVE_TIME = 3600
 
 # Timeouts and delays
 SERVER_INITIALIZATION_DELAY = 120
+BENCHMARK_STDOUT_DRAIN_GRACE = 10  # Grace seconds to wait for the direct child to exit after it drains stdout / exits
+BENCHMARK_STDOUT_IDLE_TIMEOUT = 300  # Idle timeout: no output for this long means the benchmark is stuck, then force-kill
+BENCHMARK_WATCHDOG_POLL_INTERVAL = 30  # Watchdog polling interval (seconds)
 
 # Test parameters
 PROMPTS_MULTIPLIER = 4
@@ -474,6 +478,9 @@ def run_bench_serving(
     # Run benchmark command and capture output
     metrics = {"mean_ttft": None, "mean_tpot": None, "total_tps": None}
 
+    # Launch the benchmark in its own session/process group so a dead server
+    # cannot wedge the run: on timeout we nuke the whole group (including any
+    # re-parented descendants) instead of waiting forever on its stdout.
     process = subprocess.Popen(
         cmd_args,
         stdout=subprocess.PIPE,
@@ -481,11 +488,59 @@ def run_bench_serving(
         text=True,
         bufsize=1,
         env=env,
+        start_new_session=True,
     )
+
+    reader_done = threading.Event()
+    # Timestamp of the last output line; the watchdog reads it to detect a
+    # silent/stuck benchmark.
+    last_activity = time.time()
+
+    def _kill_on_timeout():
+        while True:
+            # Direct child has exited, but stdout may still be held open by a
+            # grandchild; wait for the reader to drain within the grace period,
+            # otherwise force-kill the whole process group.
+            if process.poll() is not None:
+                if not reader_done.wait(timeout=BENCHMARK_STDOUT_DRAIN_GRACE):
+                    logger.error(
+                        f"Benchmark stdout still open after process exit, "
+                        f"killing process group {process.pid}"
+                    )
+                    kill_process_group(process)
+                return
+            # stdout has drained, but the direct child may still be alive:
+            # give it a bounded time to exit, otherwise nuke the group.
+            if reader_done.is_set():
+                try:
+                    process.wait(timeout=BENCHMARK_STDOUT_DRAIN_GRACE)
+                except subprocess.TimeoutExpired:
+                    logger.error(
+                        f"Benchmark {process.pid} still alive after stdout EOF, "
+                        f"killing process group {process.pid}"
+                    )
+                    kill_process_group(process)
+                return
+            # No output for too long while still alive -> likely hung.
+            idle = time.time() - last_activity
+            if idle > BENCHMARK_STDOUT_IDLE_TIMEOUT:
+                logger.error(
+                    f"Benchmark produced no output for {idle:.0f}s "
+                    f"(> {BENCHMARK_STDOUT_IDLE_TIMEOUT}s), "
+                    f"killing process group {process.pid}"
+                )
+                kill_process_group(process)
+                return
+            time.sleep(BENCHMARK_WATCHDOG_POLL_INTERVAL)
+
+    watchdog = threading.Thread(target=_kill_on_timeout, daemon=True)
+    watchdog.start()
+
     try:
         # Read output line by line
         with open(result_file, "a", encoding="utf-8") as f:
             for line in process.stdout:
+                last_activity = time.time()
                 if line.strip():
                     print(line, end="")
                 f.write(line)
@@ -508,6 +563,7 @@ def run_bench_serving(
                     parts = stripped_line.split()
                     if len(parts) >= 5:
                         metrics["mean_e2e_latency"] = parts[4]
+        reader_done.set()
         process.wait()
         if process.returncode != 0:
             logger.error(
@@ -516,6 +572,7 @@ def run_bench_serving(
     except Exception as e:
         logger.error(f"Error running benchmark: {e}")
     finally:
+        reader_done.set()
         if process.stdout is not None and not process.stdout.closed:
             process.stdout.close()