[Refactor]Rename diffusion benchmark backend to endpoint (#3137)
Signed-off-by: bjf-frz <frz123db@gmail.com> Signed-off-by: bjfwhite <baijingfan1@huawei.com> Co-authored-by: bjfwhite <baijingfan1@huawei.com>
This commit is contained in:
1 parent
2c6b1bb0c0
commit
1fa734419e
19 files changed
+352
-87
No files matched your search
@@ -5,16 +5,15 @@ This runner separates two concepts:
|
||||
|
||||
1. ``server_type``: how the serving process is started.
|
||||
Currently only ``vllm-omni`` is supported here.
|
||||
2. ``benchmark_backend``: which serving API the benchmark client calls.
|
||||
Examples: ``vllm-omni`` for ``/v1/chat/completions`` and ``v1/videos``
|
||||
for async video jobs.
|
||||
2. ``benchmark_endpoint``: which serving API the benchmark client calls.
|
||||
Examples: ``/v1/chat/completions`` and ``/v1/videos``.
|
||||
|
||||
A config JSON file is REQUIRED via --test-config-file:
|
||||
pytest run_diffusion_benchmark.py --test-config-file tests/dfx/perf/tests/test_qwen_image_vllm_omni.json
|
||||
|
||||
All benchmark results for a session are consolidated into a single JSON file under
|
||||
BENCHMARK_RESULT_DIR (override via the DIFFUSION_BENCHMARK_DIR environment variable).
|
||||
Each entry in the file contains the test metadata (test_name, backend, benchmark_params,
|
||||
Each entry in the file contains the test metadata (test_name, endpoint, benchmark_params,
|
||||
timestamp) together with the raw metrics returned by the benchmark script.
|
||||
"""
|
||||
|
||||
@@ -33,6 +32,8 @@ from typing import Any, cast
|
||||
import psutil
|
||||
import pytest
|
||||
|
||||
from benchmarks.diffusion.backends import endpoint_filename_token, normalize_endpoint
|
||||
|
||||
pytestmark = [pytest.mark.diffusion, pytest.mark.full_model]
|
||||
|
||||
os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"
|
||||
@@ -400,7 +401,7 @@ def _unique_server_params(configs: list[dict[str, Any]]) -> list[dict[str, Any]]
|
||||
"model": cfg["server_params"]["model"],
|
||||
"serve_args_dict": serve_args_dict,
|
||||
"serve_args": _build_serve_args(serve_args_dict),
|
||||
"benchmark_backend": cfg.get("benchmark_backend"),
|
||||
"benchmark_endpoint": cfg.get("benchmark_endpoint", cfg.get("benchmark_backend")),
|
||||
"server_params": cfg["server_params"],
|
||||
}
|
||||
)
|
||||
@@ -482,7 +483,7 @@ def run_benchmark(
|
||||
model: str,
|
||||
params: dict[str, Any],
|
||||
test_name: str,
|
||||
backend: str = "vllm-omni",
|
||||
endpoint: str = "/v1/chat/completions",
|
||||
server_cfg: dict[str, Any] | None = None,
|
||||
source_file: str = "",
|
||||
) -> dict[str, Any]:
|
||||
@@ -490,18 +491,19 @@ def run_benchmark(
|
||||
|
||||
The raw metrics are written to a temporary file by the subprocess. After
|
||||
the run completes the metrics are merged with full metadata (test_name,
|
||||
backend, benchmark_params, timestamp, flat reporting fields) and appended
|
||||
endpoint, benchmark_params, timestamp, flat reporting fields) and appended
|
||||
to the session-wide aggregated JSON file (AGGREGATED_RESULT_FILE). The
|
||||
temporary file is removed afterwards. Subprocess stdout/stderr are tee'd
|
||||
to a .log file under BENCHMARK_RESULT_DIR/logs/; its path is stored in
|
||||
the record.
|
||||
"""
|
||||
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
|
||||
endpoint = normalize_endpoint(endpoint)
|
||||
|
||||
log_dir = BENCHMARK_RESULT_DIR / "logs"
|
||||
log_dir.mkdir(parents=True, exist_ok=True)
|
||||
backend_label = backend.replace("/", "_")
|
||||
log_file = log_dir / f"{test_name}_{backend_label}_{timestamp}.log"
|
||||
endpoint_label = endpoint_filename_token(endpoint)
|
||||
log_file = log_dir / f"{test_name}_{endpoint_label}_{timestamp}.log"
|
||||
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", prefix="diffusion_bench_tmp_", delete=False) as tmp:
|
||||
tmp_result_file = Path(tmp.name)
|
||||
@@ -517,8 +519,8 @@ def run_benchmark(
|
||||
str(port),
|
||||
"--model",
|
||||
model,
|
||||
"--backend",
|
||||
backend,
|
||||
"--endpoint",
|
||||
endpoint,
|
||||
"--dataset",
|
||||
params.get("dataset", "random"),
|
||||
"--task",
|
||||
@@ -545,7 +547,7 @@ def run_benchmark(
|
||||
# cause truncated or out-of-order log output when stdout is piped).
|
||||
cmd = [cmd[0], "-u"] + cmd[1:]
|
||||
|
||||
print(f"\nRunning benchmark (backend={backend}): {' '.join(cmd)}")
|
||||
print(f"\nRunning benchmark (endpoint={endpoint}): {' '.join(cmd)}")
|
||||
print(f" Log file: {log_file}")
|
||||
|
||||
# Redirect stdout + stderr directly to the log file at the OS level
|
||||
@@ -597,7 +599,7 @@ def run_benchmark(
|
||||
|
||||
record: dict[str, Any] = {
|
||||
"test_name": test_name,
|
||||
"backend": backend,
|
||||
"endpoint": endpoint,
|
||||
"timestamp": timestamp,
|
||||
"server_params": server_cfg.get("server_params"),
|
||||
"benchmark_params": params,
|
||||
@@ -605,7 +607,7 @@ def run_benchmark(
|
||||
"log_file": str(log_file),
|
||||
"Model": model,
|
||||
"Framework": server_type,
|
||||
"API Backend": backend,
|
||||
"API Endpoint": endpoint,
|
||||
"Hardware": "",
|
||||
"Deployment": "",
|
||||
"Task": params.get("task", "t2i"),
|
||||
@@ -669,21 +671,21 @@ def assert_result(result: dict[str, Any], params: dict[str, Any]) -> None:
|
||||
assert current <= threshold, f"{metric}: {current:.4f} > baseline {threshold}"
|
||||
|
||||
|
||||
def _default_benchmark_backend_for_task(task: str) -> str:
|
||||
"""Return the default client-side benchmark backend for a diffusion task."""
|
||||
def _default_benchmark_endpoint_for_task(task: str) -> str:
|
||||
"""Return the default client-side benchmark endpoint for a diffusion task."""
|
||||
if task in {"t2v", "i2v", "ti2v"}:
|
||||
return "v1/videos"
|
||||
return "/v1/videos"
|
||||
if task in {"t2i", "i2i", "ti2i"}:
|
||||
return "vllm-omni"
|
||||
raise ValueError(f"Unsupported task for benchmark backend resolution: {task}")
|
||||
return "/v1/chat/completions"
|
||||
raise ValueError(f"Unsupported task for benchmark endpoint resolution: {task}")
|
||||
|
||||
|
||||
def _resolve_benchmark_backend(server_cfg: dict[str, Any], params: dict[str, Any]) -> str:
|
||||
def _resolve_benchmark_endpoint(server_cfg: dict[str, Any], params: dict[str, Any]) -> str:
|
||||
"""Resolve which serving API the benchmark client should call."""
|
||||
configured = server_cfg.get("benchmark_backend")
|
||||
configured = server_cfg.get("benchmark_endpoint")
|
||||
if configured:
|
||||
return cast(str, configured)
|
||||
return _default_benchmark_backend_for_task(cast(str, params.get("task", "t2i")))
|
||||
return normalize_endpoint(cast(str, configured))
|
||||
return _default_benchmark_endpoint_for_task(cast(str, params.get("task", "t2i")))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -711,7 +713,7 @@ def test_diffusion_performance_benchmark(diffusion_server, benchmark_params):
|
||||
test_name = benchmark_params["test_name"]
|
||||
params = benchmark_params["params"]
|
||||
server_cfg = getattr(diffusion_server, "server_cfg", {})
|
||||
backend = _resolve_benchmark_backend(server_cfg, params)
|
||||
endpoint = _resolve_benchmark_endpoint(server_cfg, params)
|
||||
|
||||
result = run_benchmark(
|
||||
host=diffusion_server.host,
|
||||
@@ -719,13 +721,13 @@ def test_diffusion_performance_benchmark(diffusion_server, benchmark_params):
|
||||
model=diffusion_server.model,
|
||||
params=params,
|
||||
test_name=test_name,
|
||||
backend=backend,
|
||||
endpoint=endpoint,
|
||||
server_cfg=server_cfg,
|
||||
source_file=cast(str, CONFIG_FILE_PATH),
|
||||
)
|
||||
|
||||
print(f"\n{'=' * 60}")
|
||||
print(f"Results for {test_name} (server={diffusion_server.server_type}, backend={backend}):")
|
||||
print(f"Results for {test_name} (server={diffusion_server.server_type}, endpoint={endpoint}):")
|
||||
for key in (
|
||||
"throughput_qps",
|
||||
"latency_mean",
|
||||
|
||||
Reference in new issue
Block a user