fix(perf): improve metric accuracy, validation, and shutdown

evalscope/cli/start_perf.py

@@ -27,11 +27,14 @@ def define_args(parsers: ArgumentParser):
 
     def execute(self):
         try:
-            from evalscope.perf.main import run_perf_benchmark
+            from evalscope.perf.main import PerfBenchmarkInterrupted, run_perf_benchmark
         except ImportError as e:
             raise ImportError(
                 f'Failed to import run_perf_benchmark from evalscope.perf.main, due to {e}. '
                 "Please run `pip install 'evalscope[perf]'`."
             )
 
-        run_perf_benchmark(self.args)
+        try:
+            run_perf_benchmark(self.args)
+        except PerfBenchmarkInterrupted as e:
+            raise SystemExit(e.exit_code) from None

evalscope/perf/arguments.py

@@ -480,6 +480,11 @@ def _validate_queue_size_multiplier(cls, v: int) -> int:
     def _validate_in_flight_task_multiplier(cls, v: int) -> int:
         return _at_least_one(v)
 
+    @field_validator('log_every_n_query', mode='after')
+    @classmethod
+    def _validate_log_every_n_query(cls, v: int) -> int:
+        return _at_least_one(v)
+
     @field_validator('num_workers', mode='after')
     @classmethod
     def _validate_num_workers(cls, v: int) -> int:
@@ -633,6 +638,12 @@ def _resolve_tokenize_prompt_url(self) -> None:
 
     def _validate_sweep_params(self) -> None:
         """Validate number/parallel/rate consistency after normalization."""
+        if self.multi_turn and self.open_loop:
+            raise ValueError(
+                '--multi-turn is not supported in open-loop mode: turn N cannot be dispatched before the '
+                'response of turn N-1 has been appended to the conversation context, which contradicts '
+                'open-loop scheduling (dispatch independent of in-flight requests).'
+            )
         if self.open_loop:
             self._validate_open_loop_sweep_params()
             return
@@ -664,6 +675,8 @@ def _validate_closed_loop_sweep_params(self) -> None:
                 f'The length of number and parallel should be the same, '
                 f'but got number: {self.number} and parallel: {self.parallel}'
             )
+        if any(p <= 0 for p in self.parallel):
+            raise ValueError(f'--parallel values must be > 0, but got: {self.parallel}')
 
     @contextmanager
     def output_context(self, path: str):

evalscope/perf/core/http_client.py

@@ -78,21 +78,43 @@ async def post(self, body) -> BenchmarkData:
         Returns:
             BenchmarkData: The benchmark data object containing request and response information.
         """
+        start_time = time.perf_counter()
         try:
             headers, request_id = self.api_plugin.extract_body_meta(body, self.headers)
             # Delegate the request processing to the API plugin
             output = await self.api_plugin.process_request(self.client, self.url, headers, body)
+            if not output.success:
+                if output.start_time <= 0:
+                    output.start_time = start_time
+                if output.completed_time < output.start_time:
+                    output.completed_time = time.perf_counter()
             if request_id:
                 output.request_id = request_id
             return output
         except asyncio.TimeoutError as e:
             logger.error(
                 f'TimeoutError: total_timeout: {self.total_timeout}, connect_timeout: {self.connect_timeout}, read_timeout: {self.read_timeout}. Please set longer timeout.'  # noqa: E501
             )
-            return BenchmarkData(success=False, error=str(e), is_stream=is_stream_body(body))
+            return self._failure_record(body, str(e), start_time)
         except (aiohttp.ClientConnectorError, Exception) as e:
             logger.error(e)
-            return BenchmarkData(success=False, error=str(e), is_stream=is_stream_body(body))
+            return self._failure_record(body, str(e), start_time)
+
+    @staticmethod
+    def _failure_record(body, error: str, start_time: float) -> BenchmarkData:
+        """Build a BenchmarkData for a request that never produced a response.
+
+        The start time is captured before request preparation so failures cover
+        the same lifecycle and use the same ``perf_counter()`` clock as successful
+        requests.
+        """
+        return BenchmarkData(
+            success=False,
+            error=error,
+            is_stream=is_stream_body(body),
+            start_time=start_time,
+            completed_time=time.perf_counter(),
+        )
 
     @staticmethod
     async def on_request_start(session, context, params: aiohttp.TraceRequestStartParams):

evalscope/perf/core/metrics_consumer.py

@@ -118,11 +118,16 @@ async def statistic_benchmark_metric(
                     await asyncio.to_thread(con.commit)
                     processed_since_commit = 0
 
-                message = accumulator.to_result().create_message(api_type=args.api)
-
-                await asyncio.to_thread(maybe_log_to_visualizer, args, message)
+                # Snapshot metrics and ship them to the visualizer off the event
+                # loop.  Skipped entirely when no visualizer is configured (the
+                # default): building the message for every request would be
+                # pure per-request overhead.
+                if args.visualizer:
+                    message = accumulator.to_result().create_message(api_type=args.api)
+                    await asyncio.to_thread(maybe_log_to_visualizer, args, message)
 
                 if int(accumulator.n_total) % args.log_every_n_query == 0:
+                    message = accumulator.to_result().create_message(api_type=args.api)
                     msg = json.dumps(message, ensure_ascii=False, indent=2)
                     logger.info(msg)
 

evalscope/perf/main.py

@@ -5,6 +5,7 @@
 import threading
 import time
 from argparse import Namespace
+from typing import Optional
 
 from evalscope.constants import HEARTBEAT_INTERVAL_SEC
 from evalscope.utils.asyncio_runtime import shutdown_event_loop
@@ -18,7 +19,12 @@
 from .multi_turn_benchmark import run_multi_turn_benchmark
 from .sla.sla_run import run_sla_auto_tune
 from .utils.db_util import get_output_path
-from .utils.handler import add_signal_handlers, install_uvloop_if_available
+from .utils.handler import (
+    PerfBenchmarkInterrupted,
+    ShutdownSignalState,
+    add_signal_handlers,
+    install_uvloop_if_available,
+)
 from .utils.local_server import start_app
 from .utils.log_utils import init_visualizer
 from .utils.report.generate_report import gen_perf_html_report
@@ -47,10 +53,11 @@ def run_one_benchmark(args: Arguments, output_path: str = None):
         install_uvloop_if_available()
 
     loop = asyncio.new_event_loop()
+    signal_state: Optional[ShutdownSignalState] = None
     # Only add signal handlers in main thread
     if platform.system() != 'Windows' and threading.current_thread() is threading.main_thread():
         try:
-            add_signal_handlers(loop)
+            signal_state = add_signal_handlers(loop)
         except ValueError as e:
             logger.warning(f'Cannot add signal handlers (running in non-main thread): {e}')
 
@@ -64,6 +71,10 @@ def run_one_benchmark(args: Arguments, output_path: str = None):
                 metrics_result, percentile_result, trace_summary, workload_throughput = loop.run_until_complete(
                     run_benchmark(args)
                 )
+    except asyncio.CancelledError:
+        if signal_state is None or signal_state.signal_name is None:
+            raise
+        raise PerfBenchmarkInterrupted(signal_state) from None
     finally:
         shutdown_event_loop(loop)
 

evalscope/perf/utils/benchmark_util.py

@@ -285,7 +285,14 @@ def update(self, data: BenchmarkData, api_plugin) -> None:
         self._update_wall_time(data)
 
     def _update_wall_time(self, data: BenchmarkData) -> None:
-        """Expand the wall-clock window to cover *data*'s lifecycle."""
+        """Expand the wall-clock window to cover *data*'s lifecycle.
+
+        Records without a valid timing interval are skipped: folding an absent
+        start or a completion before its start into the window would corrupt
+        QPS and throughput.
+        """
+        if data.start_time <= 0 or data.completed_time < data.start_time:
+            return
         if self._wall_start is None:
             self._wall_start = data.start_time
         else:

evalscope/perf/utils/db_util.py

@@ -1,5 +1,6 @@
 import base64
 import json
+import math
 import os
 import pickle
 import re
@@ -159,9 +160,11 @@ def calculate_percentiles(data: List[float], percentiles: List[int]) -> Dict[int
             if percentile >= 100:
                 value = data[-1] if data else float('nan')
             else:
-                idx = int(n_success_queries * percentile / 100)
-                value = data[idx] if data[idx] is not None else float('nan')
-            results[percentile] = round(value, 2)
+                # Nearest-rank method: the p-th percentile is the value at rank
+                # ceil(p/100 * n) (1-based), i.e. index ceil(p/100 * n) - 1.
+                idx = max(0, math.ceil(n_success_queries * percentile / 100) - 1)
+                value = data[idx]
+            results[percentile] = round(value, 2) if value is not None else float('nan')
         except IndexError:
             results[percentile] = float('nan')
     return results

evalscope/perf/utils/handler.py

@@ -4,6 +4,8 @@
 import os
 import platform
 import signal
+from dataclasses import dataclass
+from typing import Optional
 
 from evalscope.utils.logger import get_logger
 
@@ -18,6 +20,29 @@
 _UVLOOP_INSTALL_ATTEMPTED = False
 
 
+@dataclass
+class ShutdownSignalState:
+    """Signal received by a benchmark loop, if any."""
+
+    signal_name: Optional[str] = None
+
+    @property
+    def exit_code(self) -> int:
+        """Return the conventional shell exit code for the received signal."""
+        if self.signal_name is None:
+            raise RuntimeError('No shutdown signal has been received.')
+        return 128 + getattr(signal, self.signal_name)
+
+
+class PerfBenchmarkInterrupted(Exception):
+    """Raised after a signal-triggered benchmark cancellation finishes cleanup."""
+
+    def __init__(self, signal_state: ShutdownSignalState) -> None:
+        self.signal_name = signal_state.signal_name
+        self.exit_code = signal_state.exit_code
+        super().__init__(f'Benchmark interrupted by {self.signal_name}')
+
+
 def install_uvloop_if_available() -> None:
     """Best-effort enable uvloop as the asyncio event loop policy.
 
@@ -113,14 +138,32 @@ def sync_wrapper(*args, **kwargs):
         return sync_wrapper
 
 
-def signal_handler(signal_name, loop):
-    logger.info('Got signal %s: exit' % signal_name)
-    loop.stop()
+def signal_handler(
+    signal_name: str,
+    loop: asyncio.AbstractEventLoop,
+    signal_state: Optional[ShutdownSignalState] = None,
+) -> None:
+    """Gracefully interrupt a running benchmark loop.
+
+    ``loop.stop()`` aborts the loop mid-flight, which surfaces from
+    ``run_until_complete`` as a confusing ``RuntimeError: Event loop stopped
+    before Future completed`` and skips every ``finally`` block of the running
+    coroutine (request teardown, DB cleanup).  Cancelling the pending tasks
+    instead delivers ``CancelledError`` to the benchmark coroutine, so cleanup
+    runs and ``run_until_complete`` unwinds normally.
+    """
+    if signal_state is not None:
+        signal_state.signal_name = signal_name
+    logger.info(f'Got signal {signal_name}: cancelling pending tasks')
+    for task in asyncio.all_tasks(loop):
+        task.cancel()
 
 
-def add_signal_handlers(loop):
+def add_signal_handlers(loop: asyncio.AbstractEventLoop) -> ShutdownSignalState:
+    signal_state = ShutdownSignalState()
     for signal_name in {'SIGINT', 'SIGTERM'}:
         loop.add_signal_handler(
             getattr(signal, signal_name),
-            functools.partial(signal_handler, signal_name, loop),
+            functools.partial(signal_handler, signal_name, loop, signal_state),
         )
+    return signal_state

tests/perf/test_arguments_validation.py

@@ -0,0 +1,72 @@
+"""Unit tests for perf ``Arguments`` validation guards.
+
+Covers three previously missing validations:
+- ``--open-loop --multi-turn``: used to be accepted silently; open-loop then
+  forced ``parallel=[-1]`` so the multi-turn strategy spawned zero workers and
+  the run produced no requests at all.
+- ``--parallel <= 0`` (closed loop): used to reach the strategy and crash
+  opaquely ('Set of Tasks/Futures is empty' / 'Semaphore initial value must be
+  >= 0').
+- ``--log-every-n-query 0``: used to cause a ZeroDivisionError in the metrics
+  consumer's modulo; now coerced to 1 like the other count-type knobs.
+"""
+import pytest
+
+from evalscope.perf.arguments import Arguments
+
+
+def _args(**kwargs) -> Arguments:
+    return Arguments(model='test-model', url='http://localhost:8080/v1/chat/completions', **kwargs)
+
+
+class TestOpenLoopMultiTurnRejected:
+
+    def test_open_loop_with_multi_turn_raises(self):
+        with pytest.raises(ValueError, match='not supported in open-loop'):
+            _args(open_loop=True, multi_turn=True, rate=1.0, number=2)
+
+    def test_multi_turn_closed_loop_still_accepted(self):
+        args = _args(multi_turn=True, parallel=2, number=2)
+        assert args.multi_turn is True
+
+    def test_open_loop_single_turn_still_accepted(self):
+        args = _args(open_loop=True, rate=1.0, number=2)
+        assert args.parallel == [-1]  # unbounded concurrency marker stays intact
+
+
+class TestParallelPositivity:
+
+    @pytest.mark.parametrize('parallel', [0, -1, -5])
+    def test_non_positive_parallel_rejected(self, parallel):
+        with pytest.raises(ValueError, match='--parallel values must be > 0'):
+            _args(parallel=parallel)
+
+    @pytest.mark.parametrize('parallel', [1, 4, [8]])
+    def test_positive_parallel_accepted(self, parallel):
+        assert _args(parallel=parallel).parallel == ([parallel] if isinstance(parallel, int) else parallel)
+
+    def test_positive_parallel_sweep_accepted(self):
+        assert _args(parallel=[1, 2], number=[10, 10]).parallel == [1, 2]
+
+    def test_zero_inside_sweep_rejected(self):
+        with pytest.raises(ValueError, match='--parallel values must be > 0'):
+            _args(parallel=[1, 0, 2], number=[1, 1, 1])
+
+    def test_open_loop_parallel_marker_not_affected(self):
+        # Open-loop mode force-sets parallel=[-1] internally; the positivity
+        # rule applies only to closed-loop sweeps.
+        args = _args(open_loop=True, rate=1.0, number=2)
+        assert args.parallel == [-1]
+
+
+class TestLogEveryNQuery:
+
+    @pytest.mark.parametrize('value', [0, -3])
+    def test_non_positive_is_coerced_to_one(self, value):
+        assert _args(log_every_n_query=value).log_every_n_query == 1
+
+    def test_positive_value_preserved(self):
+        assert _args(log_every_n_query=50).log_every_n_query == 50
+
+    def test_default_preserved(self):
+        assert _args().log_every_n_query == 100