diff --git a/evalscope/cli/start_perf.py b/evalscope/cli/start_perf.py index 826bca679..130dabcc6 100644 --- a/evalscope/cli/start_perf.py +++ b/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 diff --git a/evalscope/perf/arguments.py b/evalscope/perf/arguments.py index 58f370163..96f0e806c 100644 --- a/evalscope/perf/arguments.py +++ b/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): diff --git a/evalscope/perf/core/http_client.py b/evalscope/perf/core/http_client.py index 51628713f..d002ebf09 100644 --- a/evalscope/perf/core/http_client.py +++ b/evalscope/perf/core/http_client.py @@ -78,10 +78,16 @@ 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 @@ -89,10 +95,26 @@ async def post(self, body) -> BenchmarkData: 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): diff --git a/evalscope/perf/core/metrics_consumer.py b/evalscope/perf/core/metrics_consumer.py index c2a602992..5378df8fa 100644 --- a/evalscope/perf/core/metrics_consumer.py +++ b/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) diff --git a/evalscope/perf/main.py b/evalscope/perf/main.py index e7df45a95..3a2a520ad 100644 --- a/evalscope/perf/main.py +++ b/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) diff --git a/evalscope/perf/utils/benchmark_util.py b/evalscope/perf/utils/benchmark_util.py index e436a3a81..a13cdba3e 100644 --- a/evalscope/perf/utils/benchmark_util.py +++ b/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: diff --git a/evalscope/perf/utils/db_util.py b/evalscope/perf/utils/db_util.py index c73fe23cf..eaecc45df 100644 --- a/evalscope/perf/utils/db_util.py +++ b/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 diff --git a/evalscope/perf/utils/handler.py b/evalscope/perf/utils/handler.py index 09d6422a6..f8945458e 100644 --- a/evalscope/perf/utils/handler.py +++ b/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 diff --git a/tests/perf/test_arguments_validation.py b/tests/perf/test_arguments_validation.py new file mode 100644 index 000000000..b9b5a9b18 --- /dev/null +++ b/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 diff --git a/tests/perf/test_async_lifecycle.py b/tests/perf/test_async_lifecycle.py index bdea8fd7e..c5cd17f82 100644 --- a/tests/perf/test_async_lifecycle.py +++ b/tests/perf/test_async_lifecycle.py @@ -1,10 +1,15 @@ import asyncio +import subprocess +import sys +import textwrap +from pathlib import Path from types import SimpleNamespace from typing import Any, AsyncIterator, Dict, List, Tuple import pytest from aiohttp import web +import evalscope.perf.main as perf_main from evalscope.perf.arguments import Arguments from evalscope.perf.benchmark import run_benchmark from evalscope.perf.core import pipeline @@ -13,7 +18,7 @@ from evalscope.perf.core.strategies.multi_turn import MultiTurnStrategy from evalscope.perf.core.strategies.open_loop import OpenLoopStrategy from evalscope.perf.utils.db_util import get_result_db_path -from evalscope.perf.utils.handler import exception_handler +from evalscope.perf.utils.handler import exception_handler, signal_handler def _make_args(**kwargs: Any) -> Arguments: @@ -197,6 +202,127 @@ def test_existing_result_database_raises_file_exists_error(tmp_path) -> None: get_result_db_path(SimpleNamespace(outputs_dir=str(tmp_path))) +class TestSignalHandlerGracefulShutdown: + """SIGINT/SIGTERM must cancel pending tasks instead of stopping the loop. + + ``loop.stop()`` used to make ``run_until_complete`` raise a confusing + ``RuntimeError: Event loop stopped before Future completed`` while skipping + every ``finally`` block of the benchmark coroutine. + """ + + def test_running_coroutine_is_cancelled_and_cleanup_runs(self) -> None: + loop = asyncio.new_event_loop() + cleanup: List[str] = [] + + async def benchmark_coroutine() -> None: + # Simulate the loop firing the registered SIGINT callback mid-run. + loop.call_soon(signal_handler, 'SIGINT', loop) + try: + await asyncio.sleep(30) + finally: + cleanup.append('ran') + + try: + with pytest.raises(asyncio.CancelledError): + loop.run_until_complete(benchmark_coroutine()) + assert cleanup == ['ran'] + finally: + loop.close() + + def test_in_flight_requests_are_cancelled_too(self) -> None: + loop = asyncio.new_event_loop() + in_flight_cleanup: List[str] = [] + + async def in_flight_request() -> None: + try: + await asyncio.Event().wait() + finally: + in_flight_cleanup.append('ran') + + async def benchmark_coroutine() -> None: + task = asyncio.create_task(in_flight_request()) + await asyncio.sleep(0) # let the in-flight request start + loop.call_soon(signal_handler, 'SIGTERM', loop) + await asyncio.sleep(30) + + try: + with pytest.raises(asyncio.CancelledError): + loop.run_until_complete(benchmark_coroutine()) + # Drain the cancellation delivered to the in-flight request. + loop.run_until_complete(asyncio.sleep(0)) + assert in_flight_cleanup == ['ran'] + finally: + loop.close() + + def test_sigint_exits_cleanly_after_cleanup(self, tmp_path: Path) -> None: + script = textwrap.dedent( + """ + import asyncio + import os + import signal + import sys + + sys.path.insert(0, sys.argv[2]) + + from evalscope.perf.arguments import Arguments + import evalscope.perf.main as perf_main + + args = Arguments(model='test-model', api='openai', number=1, parallel=1, rate=-1) + args.number = 1 + args.parallel = 1 + args.rate = -1 + + async def interrupted(_: Arguments) -> None: + loop = asyncio.get_running_loop() + loop.call_later(0.05, os.kill, os.getpid(), signal.SIGINT) + try: + await asyncio.Event().wait() + finally: + print('CLEANUP_RAN=True', flush=True) + + output_path = sys.argv[1] + + def run_test_perf(cli_args: object) -> None: + perf_main.run_one_benchmark(args, output_path) + + perf_main.run_benchmark = interrupted + perf_main.run_perf_benchmark = run_test_perf + + from evalscope.cli.cli import run_cmd + + sys.argv = ['evalscope', 'perf', '--model', 'test-model'] + run_cmd() + """ + ) + + result = subprocess.run( + [sys.executable, '-c', script, str(tmp_path), str(Path(__file__).parents[2])], + cwd=tmp_path, + capture_output=True, + text=True, + timeout=30, + check=False, + ) + + assert result.returncode == 130 + assert 'CLEANUP_RAN=True' in result.stdout + assert 'Traceback' not in result.stderr + assert 'CancelledError' not in result.stderr + + def test_internal_cancellation_is_not_treated_as_a_signal( + self, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + ) -> None: + async def cancelled_benchmark(_args: Arguments) -> None: + raise asyncio.CancelledError + + monkeypatch.setattr(perf_main, 'run_benchmark', cancelled_benchmark) + + with pytest.raises(asyncio.CancelledError): + perf_main.run_one_benchmark(_make_args(), str(tmp_path)) + + def test_aiohttp_client_context_returns_self_and_closes() -> None: async def run() -> None: client = AioHttpClient(_make_args(), None) diff --git a/tests/perf/test_metrics_consumer.py b/tests/perf/test_metrics_consumer.py new file mode 100644 index 000000000..8fdf91389 --- /dev/null +++ b/tests/perf/test_metrics_consumer.py @@ -0,0 +1,125 @@ +"""Unit tests for the perf metrics consumer's visualizer fast path. + +The consumer used to build a metrics snapshot +(``accumulator.to_result().create_message(...)``) and jump to a worker thread +(``await asyncio.to_thread(maybe_log_to_visualizer, ...)``) for *every* +request even when ``args.visualizer`` is None (the default), where +``maybe_log_to_visualizer`` is a no-op. The snapshot+thread hop must only +happen when a visualizer is actually configured. +""" +import asyncio +from typing import Any, List + +import pytest + +from evalscope.perf.arguments import Arguments +from evalscope.perf.core import metrics_consumer +from evalscope.perf.core.metrics_consumer import statistic_benchmark_metric +from evalscope.perf.utils.benchmark_util import BenchmarkData, MetricsAccumulator + + +class _DummyPlugin: + """Minimal api plugin: finalize() only needs parse_responses().""" + + def parse_responses(self, responses, request=None, **kwargs): + return 10, 5 + + +def _bench_data(index: int) -> BenchmarkData: + data = BenchmarkData( + success=True, + start_time=float(index), + completed_time=float(index) + 1.0, + query_latency=1.0, + first_chunk_latency=0.2, + prompt_tokens=10, + completion_tokens=5, + is_stream=True, + ) + data.request = '{}' + data.response_messages = [] + return data + + +def _run_consumer(args: Arguments, n_requests: int) -> MetricsAccumulator: + """Drive statistic_benchmark_metric with n_requests successful records.""" + + async def go() -> MetricsAccumulator: + queue: asyncio.Queue = asyncio.Queue() + completed = asyncio.Event() + consumer_task = asyncio.create_task(statistic_benchmark_metric(queue, args, _DummyPlugin(), completed)) + for i in range(n_requests): + await queue.put(_bench_data(i)) + completed.set() + accumulator, trace_summary, timeline, db_path = await consumer_task + return accumulator + + return asyncio.run(go()) + + +def _make_args(tmp_path, **kwargs: Any) -> Arguments: + kwargs.setdefault('log_every_n_query', 100) + args = Arguments(model='test-model', api='openai', **kwargs) + args.number = 2 + args.outputs_dir = str(tmp_path) + return args + + +@pytest.fixture +def counters(monkeypatch: pytest.MonkeyPatch): + """Record to_result() and maybe_log_to_visualizer() invocations.""" + calls = {'to_result': 0} + original_to_result = MetricsAccumulator.to_result + + def counting_to_result(self: MetricsAccumulator): + calls['to_result'] += 1 + return original_to_result(self) + + recorded_visualizer_calls: List[dict] = [] + + def fake_visualizer(args: Arguments, message: dict) -> None: + recorded_visualizer_calls.append(message) + + monkeypatch.setattr(MetricsAccumulator, 'to_result', counting_to_result) + monkeypatch.setattr(metrics_consumer, 'maybe_log_to_visualizer', fake_visualizer) + return calls, recorded_visualizer_calls + + +class TestVisualizerFastPath: + + def test_no_visualizer_skips_per_request_snapshot(self, tmp_path, counters) -> None: + calls, visualizer_calls = counters + args = _make_args(tmp_path) # visualizer=None (default) + + accumulator = _run_consumer(args, n_requests=2) + + assert visualizer_calls == [] + # Only the final result snapshot; none of the former per-request ones. + assert calls['to_result'] == 1 + assert accumulator.succeed_requests == 2 + + def test_visualizer_configured_logs_per_request(self, tmp_path, counters) -> None: + calls, visualizer_calls = counters + args = _make_args(tmp_path, visualizer='swanlab') + + accumulator = _run_consumer(args, n_requests=2) + + assert len(visualizer_calls) == 2 + # 2 per-request snapshots + 1 final. + assert calls['to_result'] == 3 + assert accumulator.succeed_requests == 2 + + def test_periodic_logging_still_emits_without_visualizer(self, tmp_path, counters, monkeypatch) -> None: + logged: List[str] = [] + monkeypatch.setattr(metrics_consumer.logger, 'info', lambda msg, *a, **kw: logged.append(str(msg))) + args = _make_args(tmp_path, log_every_n_query=1) + + accumulator = _run_consumer(args, n_requests=2) + + assert accumulator.succeed_requests == 2 + # Every request crossed the log_every_n_query boundary and logged its + # metrics message even though no visualizer is configured. Other + # logger.info traffic (db path, progress) is filtered out. + metric_messages = [msg for msg in logged if msg.startswith('{')] + assert len(metric_messages) == 2 + assert all('"Success Requests"' in msg for msg in metric_messages) diff --git a/tests/perf/test_percentile_metrics.py b/tests/perf/test_percentile_metrics.py new file mode 100644 index 000000000..6c3a7d358 --- /dev/null +++ b/tests/perf/test_percentile_metrics.py @@ -0,0 +1,55 @@ +"""Unit tests for the nearest-rank percentile calculation in perf metrics. + +``calculate_percentiles`` used ``int(n * p / 100)`` as the index, which biased +every percentile one rank high (n=100 -> p99 = max; n=2 -> p50 = larger value). +These tests pin the nearest-rank semantics: the p-th percentile is the value at +1-based rank ceil(p / 100 * n). +""" +import math + +import pytest + +from evalscope.perf.utils.db_util import calculate_percentiles + + +class TestCalculatePercentiles: + + def test_nearest_rank_on_one_hundred_values(self): + # Previously returned {50: 50, 99: 99} (one rank too high). + assert calculate_percentiles(list(range(100)), [50, 99]) == {50: 49, 99: 98} + + def test_median_of_two_values_is_the_smaller(self): + # int(2 * 50 / 100) == 1 picked the larger value; nearest rank picks rank 1. + assert calculate_percentiles([1.0, 2.0], [50]) == {50: 1.0} + + def test_single_value_all_percentiles(self): + assert calculate_percentiles([7.0], [0, 1, 50, 99, 100]) == {0: 7.0, 1: 7.0, 50: 7.0, 99: 7.0, 100: 7.0} + + def test_percentile_zero_is_min(self): + assert calculate_percentiles([3.0, 1.0, 2.0], [0]) == {0: 1.0} + + def test_percentile_hundred_or_more_is_max(self): + assert calculate_percentiles([3.0, 1.0, 2.0], [100, 150]) == {100: 3.0, 150: 3.0} + + def test_empty_data_returns_nan(self): + result = calculate_percentiles([], [0, 50, 100]) + assert all(math.isnan(v) for v in result.values()) + + def test_missing_value_returns_nan(self): + result = calculate_percentiles([None], [0, 50, 100]) + assert all(math.isnan(v) for v in result.values()) + + def test_input_list_is_sorted_in_place(self): + data = [9.0, 1.0, 5.0] + calculate_percentiles(data, [50]) + assert data == [1.0, 5.0, 9.0] + + def test_small_sample_nearest_rank_boundaries(self): + # n=3: p1 -> ceil(0.03)=1 -> idx 0; p50 -> ceil(1.5)=2 -> idx 1; p99 -> ceil(2.97)=3 -> idx 2. + assert calculate_percentiles([10.0, 20.0, 30.0], [1, 50, 99]) == {1: 10.0, 50: 20.0, 99: 30.0} + + @pytest.mark.parametrize('n', [1, 2, 3, 7, 10, 100]) + def test_percentiles_never_exceed_max_or_fall_below_min(self, n): + data = list(range(n)) + result = calculate_percentiles(data, [1, 5, 25, 50, 75, 95, 99]) + assert all(0 <= v <= n - 1 for v in result.values()) diff --git a/tests/perf/test_wall_time_failures.py b/tests/perf/test_wall_time_failures.py new file mode 100644 index 000000000..ead00f3e8 --- /dev/null +++ b/tests/perf/test_wall_time_failures.py @@ -0,0 +1,179 @@ +"""Unit tests for failure-record timing in perf metrics (wall_time / QPS). + +``AioHttpClient.post`` used to build untimed failure records +(``start_time=0.0``), and ``MetricsAccumulator._update_wall_time`` folded them +into the wall-clock window via ``min(...)``. A single failed request therefore +pinned ``_wall_start`` to 0.0 (the perf_counter epoch), inflating wall_time to +hours and collapsing QPS to ~0. Failure records must carry a real timestamp, +and untimed records must never widen the window. +""" +import asyncio +import time +from typing import Any + +import pytest + +from evalscope.perf.arguments import Arguments +from evalscope.perf.core.http_client import AioHttpClient +from evalscope.perf.utils.benchmark_util import BenchmarkData, MetricsAccumulator + + +class _DummyPlugin: + """Minimal api plugin: finalize() only needs parse_responses().""" + + def parse_responses(self, responses, request=None, **kwargs): + return 10, 5 + + +def _success(start: float, end: float) -> BenchmarkData: + data = BenchmarkData( + success=True, + start_time=start, + completed_time=end, + query_latency=end - start, + first_chunk_latency=0.2, + prompt_tokens=10, + completion_tokens=5, + is_stream=True, + ) + data.request = '{}' + data.response_messages = [] + return data + + +class TestWallTimeWithFailedRequests: + + def test_untimed_failure_does_not_corrupt_wall_time(self): + accumulator = MetricsAccumulator() + accumulator.update(_success(10.0, 11.0), _DummyPlugin()) + # Untimed failure record, as produced by legacy code paths. + accumulator.update(BenchmarkData(success=False, start_time=0.0, completed_time=0.0), _DummyPlugin()) + + result = accumulator.to_result() + + assert accumulator.wall_time == pytest.approx(1.0) + assert result.qps == pytest.approx(1.0) + assert result.failed_requests == 1 + + def test_timestamped_failure_still_expands_window(self): + accumulator = MetricsAccumulator() + accumulator.update(_success(10.0, 11.0), _DummyPlugin()) + # A properly stamped failure widens the window like any real request. + accumulator.update(BenchmarkData(success=False, start_time=9.5, completed_time=11.5), _DummyPlugin()) + + assert accumulator.wall_time == pytest.approx(2.0) + + def test_only_untimed_failures_keeps_guard_wall_time(self): + # Before any timed record arrives the guard value must stay in place. + accumulator = MetricsAccumulator() + accumulator.update(BenchmarkData(success=False, start_time=0.0, completed_time=0.0), _DummyPlugin()) + + assert accumulator.wall_time == 1.0 + + def test_incomplete_failure_does_not_corrupt_wall_time(self) -> None: + accumulator = MetricsAccumulator() + accumulator.update(_success(10.0, 11.0), _DummyPlugin()) + accumulator.update(BenchmarkData(success=False, start_time=12.0, completed_time=0.0), _DummyPlugin()) + + assert accumulator.wall_time == pytest.approx(1.0) + + +class _ExplodingPlugin: + """Api plugin whose process_request() raises, e.g. a buggy custom plugin.""" + + def extract_body_meta(self, body, headers): + return headers, None + + async def process_request(self, client_session, url, headers, body): + raise RuntimeError('plugin exploded') + + def parse_responses(self, responses: Any, request: Any = None, **kwargs: Any) -> tuple[int, int]: + return 10, 5 + + +class _SlowFailureThenSuccessPlugin(_ExplodingPlugin): + + def __init__(self) -> None: + self.calls = 0 + + async def process_request(self, client_session: Any, url: str, headers: dict, body: Any) -> BenchmarkData: + self.calls += 1 + if self.calls == 1: + await asyncio.sleep(0.06) + raise RuntimeError('slow failure') + + start = time.perf_counter() + await asyncio.sleep(0.01) + return _success(start, time.perf_counter()) + + +class _ReturningSlowFailurePlugin(_ExplodingPlugin): + + async def process_request(self, client_session: Any, url: str, headers: dict, body: Any) -> BenchmarkData: + start = time.perf_counter() + await asyncio.sleep(0.06) + return BenchmarkData(success=False, error='HTTP 500', start_time=start) + + +class TestHttpClientFailureRecordsAreTimestamped: + + def test_slow_failure_is_included_in_wall_time_and_qps(self) -> None: + async def run() -> tuple[BenchmarkData, BenchmarkData]: + args = Arguments(model='test-model', api='openai') + args.parallel = 1 + client = AioHttpClient(args, _SlowFailureThenSuccessPlugin()) + async with client: + failure = await client.post({'stream': True}) + success = await client.post({'stream': True}) + return failure, success + + failure, success = asyncio.run(run()) + accumulator = MetricsAccumulator() + accumulator.update(failure, _DummyPlugin()) + accumulator.update(success, _DummyPlugin()) + result = accumulator.to_result() + expected_wall_time = success.completed_time - failure.start_time + + assert failure.completed_time - failure.start_time >= 0.05 + assert result.total_time == pytest.approx(expected_wall_time) + assert result.qps == pytest.approx(1 / expected_wall_time) + + def test_plugin_failure_gets_a_completion_timestamp(self) -> None: + async def run() -> BenchmarkData: + args = Arguments(model='test-model', api='openai') + args.parallel = 1 + client = AioHttpClient(args, _ReturningSlowFailurePlugin()) + async with client: + return await client.post({'stream': True}) + + failure = asyncio.run(run()) + + assert failure.completed_time - failure.start_time >= 0.05 + + @pytest.mark.parametrize('error', [RuntimeError('plugin exploded'), asyncio.TimeoutError()]) + def test_failure_record_carries_perf_counter_timestamp(self, error): + async def run() -> BenchmarkData: + args = Arguments(model='test-model', api='openai') + # run_one_benchmark() collapses sweep lists to scalars before the + # HTTP client is constructed; mirror that here. + args.parallel = 1 + client = AioHttpClient(args, _ExplodingPlugin()) + client.api_plugin.process_request = self._raising(error) + async with client: + return await client.post({'stream': True}) + + before = time.perf_counter() + data = asyncio.run(run()) + after = time.perf_counter() + + assert data.success is False + assert data.is_stream is True + assert before <= data.start_time <= after + assert data.completed_time >= data.start_time + + @staticmethod + def _raising(error: BaseException): + async def process_request(client_session, url, headers, body): + raise error + + return process_request