anodyine/vllm-omni
1
0 Fork 0
Code Issues 4 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

[CI/Build] Resolve hardware-nested perf baselines per concurrency sweep (#5858)

Browse Source 
Signed-off-by: wangyu <410167048@qq.com>
Co-authored-by: amy-why-3459 <wuhaiyan17@huawei.com>
This commit is contained in:
wangyuandamy-why-3459 authored and GitHub committed 2026-08-07 16:54:38 +08:00
1 parent 1be47550d0
commit 0a12ac52ee
33 files changed
+1207 -978
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/dfx/perf/scripts/run_benchmark.py
+75 -83
View File
@@ -1,6 +1,8 @@
import json
import os
import threading
from collections.abc import Callable
from contextlib import AbstractContextManager, ExitStack, contextmanager
from pathlib import Path
from typing import Any
@@ -13,15 +15,10 @@ from tests.dfx.conftest import (
get_runtime_resource_label,
is_diffusion_perf_config,
load_benchmark_configs,
resolve_baseline_value,
run_benchmark,
)
from tests.helpers.runtime import OmniServer
# Compare metrics to each test JSON ``baseline`` block only when pytest is run with ``--assert-baseline``
# (registered in ``tests/dfx/conftest.py``; default: off). ``run_benchmark`` and ``_resolve_baseline_value`` are
# defined in the same module.
#
# Optional JSON field ``mark`` is applied as pytest marks via
# ``create_paired_omni_benchmark_pytest_params`` (e.g. ``"mark": [{"hardware_marks":
# {"res": {"cuda": "H100"}, "num_cards": 2}}, "full_model", "omni"]``).
@@ -63,35 +60,78 @@ paired_benchmark_params = create_paired_omni_benchmark_pytest_params(BENCHMARK_C
_omni_server_lock = threading.Lock()
class _SingleActiveContext:
"""Reuse one active context while its configuration key is unchanged."""
def __init__(self) -> None:
self._key: Any = None
self._stack: ExitStack | None = None
self._value: Any = None
def acquire(self, key: Any, factory: Callable[[], AbstractContextManager[Any]]) -> Any:
if self._stack is not None and key == self._key:
return self._value
self.close()
stack = ExitStack()
value = stack.enter_context(factory())
self._key = key
self._stack = stack
self._value = value
return value
def close(self) -> None:
stack = self._stack
self._key = None
self._stack = None
self._value = None
if stack is not None:
stack.close()
@contextmanager
def _start_omni_server(server_param):
test_name, model, stage_config_path, stage_overrides, extra_cli_args, use_omni = server_param
print(f"Starting OmniServer with test: {test_name}, model: {model}")
server_args: list[str] = []
if use_omni:
server_args += ["--stage-init-timeout", "600", "--init-timeout", "900"]
# --deploy-config and --stage-overrides compose at the CLI (see vllm_omni/entrypoints/utils.py):
# deploy-config sets the base; stage-overrides are applied on top. Both can be set.
if stage_config_path:
server_args = ["--deploy-config", stage_config_path] + server_args
if stage_overrides:
server_args = ["--stage-overrides", stage_overrides] + server_args
if extra_cli_args:
server_args = list(extra_cli_args) + server_args
with OmniServer(model, server_args, use_omni=use_omni) as server:
server.test_name = test_name
print("OmniServer started successfully")
yield server
print("OmniServer stopping...")
print("OmniServer stopped")
@pytest.fixture(scope="module")
def omni_server(request):
def omni_server_context():
"""Start vLLM-Omni server as a subprocess with actual model weights.
Uses session scope so the server starts only once for the entire test session.
Reuse it for adjacent benchmark cases with the same server configuration.
Multi-stage initialization can take 10-20+ minutes.
"""
with _omni_server_lock:
test_name, model, stage_config_path, stage_overrides, extra_cli_args, use_omni = request.param
active_context = _SingleActiveContext()
try:
yield active_context
finally:
active_context.close()
print(f"Starting OmniServer with test: {test_name}, model: {model}")
server_args: list[str] = []
if use_omni:
server_args += ["--stage-init-timeout", "600", "--init-timeout", "900"]
# --deploy-config and --stage-overrides compose at the CLI (see vllm_omni/entrypoints/utils.py):
# deploy-config sets the base; stage-overrides are applied on top. Both can be set.
if stage_config_path:
server_args = ["--deploy-config", stage_config_path] + server_args
if stage_overrides:
server_args = ["--stage-overrides", stage_overrides] + server_args
if extra_cli_args:
server_args = list(extra_cli_args) + server_args
with OmniServer(model, server_args, use_omni=use_omni) as server:
server.test_name = test_name
print("OmniServer started successfully")
yield server
print("OmniServer stopping...")
print("OmniServer stopped")
@pytest.fixture
def omni_server(request, omni_server_context):
return omni_server_context.acquire(request.param, lambda: _start_omni_server(request.param))
@pytest.fixture
@@ -117,16 +157,7 @@ def benchmark_params(request):
}
def assert_result(
result,
params,
num_prompt,
*,
assert_baseline: bool,
sweep_index: int | None = None,
max_concurrency: Any | None = None,
request_rate: Any | None = None,
) -> None:
def assert_result(result, params, num_prompt) -> None:
assert result["completed"] == num_prompt, "Request failures exist"
expected_audio_turns = params.get("expected_duplex_audio_turns_per_session")
if expected_audio_turns is not None:
@@ -139,25 +170,6 @@ def assert_result(
isinstance(metric, dict) and metric.get("audio_turn_count") == expected_audio_turns
for metric in session_metrics
), f"Not every duplex session emitted {expected_audio_turns} audio turns"
if not assert_baseline:
return
baseline_data = params.get("baseline", {})
for metric_name, baseline_raw in baseline_data.items():
current_value = result[metric_name]
baseline_value = resolve_baseline_value(
baseline_raw,
sweep_index=sweep_index,
max_concurrency=max_concurrency,
request_rate=request_rate,
)
if "throughput" in metric_name:
assert current_value >= baseline_value, (
f"Throughput test result was below baseline: {metric_name}: {current_value} < {baseline_value}"
)
else:
assert current_value <= baseline_value, (
f"Test result exceeded baseline: {metric_name}: {current_value} > {baseline_value}"
)
@pytest.mark.benchmark
@@ -166,7 +178,7 @@ def assert_result(
paired_benchmark_params,
indirect=["omni_server", "benchmark_params"],
)
def test_performance_benchmark(omni_server, benchmark_params, request):
def test_performance_benchmark(omni_server, benchmark_params):
test_name = benchmark_params["test_name"]
params = benchmark_params["params"]
dataset_name = params.get("dataset_name", "")
@@ -178,7 +190,6 @@ def test_performance_benchmark(omni_server, benchmark_params, request):
print(f"Running benchmark for model: {model}")
print(f"Benchmark parameters: {benchmark_params}")
assert_baseline = request.config.getoption("--assert-baseline", default=False)
resource_label = get_runtime_resource_label()
def to_list(value, default=None):
@@ -213,7 +224,6 @@ def test_performance_benchmark(omni_server, benchmark_params, request):
"eval_phase",
"trust_remote_code",
"expected_duplex_audio_turns_per_session",
"baseline_hardware",
}
for key, value in params.items():
@@ -239,7 +249,7 @@ def test_performance_benchmark(omni_server, benchmark_params, request):
break
# QPS / request-rate sweep
for i, (qps, num_prompt) in enumerate(zip(qps_list, num_prompt_list)):
for sweep_index, (qps, num_prompt) in enumerate(zip(qps_list, num_prompt_list)):
args = args + ["--request-rate", str(qps), "--num-prompts", str(num_prompt)]
result = run_benchmark(
args=args,
@@ -248,24 +258,15 @@ def test_performance_benchmark(omni_server, benchmark_params, request):
dataset_name=dataset_name,
num_prompt=num_prompt,
baseline_config=params.get("baseline"),
sweep_index=i,
request_rate=qps,
max_concurrency=None,
sweep_index=sweep_index,
random_input_len=params.get("random_input_len"),
random_output_len=params.get("random_output_len"),
resource_label=resource_label,
)
assert_result(
result,
params,
num_prompt,
assert_baseline=assert_baseline,
sweep_index=i,
request_rate=qps,
)
assert_result(result, params, num_prompt)
# concurrency test
for i, (concurrency, num_prompt) in enumerate(zip(max_concurrency_list, num_prompt_list)):
for sweep_index, (concurrency, num_prompt) in enumerate(zip(max_concurrency_list, num_prompt_list)):
args = args + ["--max-concurrency", str(concurrency), "--num-prompts", str(num_prompt), "--request-rate", "inf"]
result = run_benchmark(
args=args,
@@ -274,18 +275,9 @@ def test_performance_benchmark(omni_server, benchmark_params, request):
dataset_name=dataset_name,
num_prompt=num_prompt,
baseline_config=params.get("baseline"),
sweep_index=i,
request_rate=None,
max_concurrency=concurrency,
sweep_index=sweep_index,
random_input_len=params.get("random_input_len"),
random_output_len=params.get("random_output_len"),
resource_label=resource_label,
)
assert_result(
result,
params,
num_prompt,
assert_baseline=assert_baseline,
sweep_index=i,
max_concurrency=concurrency,
)
assert_result(result, params, num_prompt)