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

[Test] Qwen-Image Perf Test with High Concurrency (#2822)

Browse Source 
Signed-off-by: Didan Deng <33117903+wtomin@users.noreply.github.com>
Signed-off-by: Gao Han <hgaoaf@connect.ust.hk>
Co-authored-by: Gao Han <hgaoaf@connect.ust.hk>
This commit is contained in:
Didan DengandGao Han authored and GitHub committed 2026-05-26 02:56:01 +00:00
1 parent 52020c7af7
commit 410a6a20b0
2 files changed
+227 -33
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/dfx/perf/scripts/run_diffusion_benchmark.py
+193 -33
View File
@@ -654,9 +654,47 @@ def run_benchmark(
# ---------------------------------------------------------------------------
def assert_result(result: dict[str, Any], params: dict[str, Any], *, assert_baseline: bool) -> None:
"""Assert request completion; optionally assert metrics against ``baseline`` in *params*."""
num_prompts = params.get("num-prompts", 10)
def _resolve_baseline_value(
baseline_raw: Any,
*,
sweep_index: int | None,
max_concurrency: Any = None,
request_rate: Any = None,
) -> Any:
"""Pick the baseline threshold for this sweep step."""
if baseline_raw is None:
return 100000
if isinstance(baseline_raw, dict):
if max_concurrency is not None:
for key in (max_concurrency, str(max_concurrency)):
if key in baseline_raw:
return baseline_raw[key]
if request_rate is not None:
for key in (request_rate, str(request_rate)):
if key in baseline_raw:
return baseline_raw[key]
raise KeyError(
f"baseline dict has no key for max_concurrency={max_concurrency!r} "
f"or request_rate={request_rate!r}; keys={list(baseline_raw.keys())!r}"
)
if isinstance(baseline_raw, (list, tuple)):
if sweep_index is None:
raise ValueError("sweep_index is required when baseline is a list or tuple")
return baseline_raw[sweep_index]
return baseline_raw
def assert_result(
result: dict[str, Any],
params: dict[str, Any],
num_prompts: int,
*,
sweep_index: int | None = None,
max_concurrency: Any = None,
request_rate: Any = None,
assert_baseline: bool = True,
) -> None:
"""Assert that benchmark metrics satisfy the configured baselines."""
completed = result.get("completed_requests", result.get("completed", 0))
assert completed == num_prompts, f"Expected {num_prompts} completed requests, got {completed}"
@@ -666,9 +704,15 @@ def assert_result(result: dict[str, Any], params: dict[str, Any], *, assert_base
print("Skipping performance assertions.")
return
for metric, threshold in params.get("baseline", {}).items():
for metric, baseline_raw in params.get("baseline", {}).items():
current = result.get(metric)
assert current is not None, f"Metric '{metric}' not found in result: {list(result.keys())}"
threshold = _resolve_baseline_value(
baseline_raw,
sweep_index=sweep_index,
max_concurrency=max_concurrency,
request_rate=request_rate,
)
if "throughput" in metric:
assert current >= threshold, f"{metric}: {current:.4f} < baseline {threshold}"
else:
@@ -692,6 +736,111 @@ def _resolve_benchmark_endpoint(server_cfg: dict[str, Any], params: dict[str, An
return _default_benchmark_endpoint_for_task(cast(str, params.get("task", "t2i")))
def _to_list(value: Any) -> list[Any]:
if value is None:
return []
return [value] if not isinstance(value, (list, tuple)) else list(value)
def _build_run_params(
params: dict[str, Any],
*,
num_prompts: int,
sweep_index: int | None = None,
request_rate: Any | None = None,
max_concurrency: Any | None = None,
) -> dict[str, Any]:
run_params = {
key: value for key, value in params.items() if key not in {"request-rate", "max-concurrency", "num-prompts"}
}
run_params["num-prompts"] = num_prompts
if request_rate is not None:
run_params["request-rate"] = request_rate
if max_concurrency is not None:
run_params["max-concurrency"] = max_concurrency
if "baseline" in params:
run_params["baseline"] = {
metric: _resolve_baseline_value(
baseline_raw,
sweep_index=sweep_index,
max_concurrency=max_concurrency,
request_rate=request_rate,
)
for metric, baseline_raw in params["baseline"].items()
}
return run_params
def _iter_sweep_runs(params: dict[str, Any]) -> list[dict[str, Any]]:
request_rate_list = _to_list(params.get("request-rate"))
num_prompt_list = _to_list(params.get("num-prompts", 10))
max_concurrency_list = _to_list(params.get("max-concurrency"))
max_len = max(len(request_rate_list), len(max_concurrency_list))
if len(num_prompt_list) == 1 and max_len > 1:
num_prompt_list = num_prompt_list * max_len
elif max_len == 1 and len(num_prompt_list) > 1:
if len(request_rate_list) == 1:
request_rate_list = request_rate_list * len(num_prompt_list)
if len(max_concurrency_list) == 1:
max_concurrency_list = max_concurrency_list * len(num_prompt_list)
max_len = max(len(request_rate_list), len(max_concurrency_list))
elif len(num_prompt_list) != max_len and max_len > 0:
raise ValueError("The number of prompts does not match the request-rate or max-concurrency")
sweep_runs: list[dict[str, Any]] = []
for i, (request_rate, num_prompts) in enumerate(zip(request_rate_list, num_prompt_list)):
sweep_runs.append(
{
"params": _build_run_params(
params,
request_rate=request_rate,
num_prompts=num_prompts,
sweep_index=i,
),
"num_prompts": num_prompts,
"sweep_index": i,
"request_rate": request_rate,
"max_concurrency": None,
}
)
for i, (max_concurrency, num_prompts) in enumerate(zip(max_concurrency_list, num_prompt_list)):
sweep_runs.append(
{
"params": _build_run_params(
params,
max_concurrency=max_concurrency,
num_prompts=num_prompts,
sweep_index=i,
request_rate="inf",
),
"num_prompts": num_prompts,
"sweep_index": i,
"request_rate": None,
"max_concurrency": max_concurrency,
}
)
if not sweep_runs:
default_num_prompts = num_prompt_list[0]
sweep_runs.append(
{
"params": _build_run_params(
params,
num_prompts=default_num_prompts,
),
"num_prompts": default_num_prompts,
"sweep_index": None,
"request_rate": None,
"max_concurrency": None,
}
)
return sweep_runs
# ---------------------------------------------------------------------------
# Test entry point
# ---------------------------------------------------------------------------
@@ -715,36 +864,47 @@ def test_diffusion_performance_benchmark(diffusion_server, benchmark_params, req
test_name = benchmark_params["test_name"]
params = benchmark_params["params"]
server_cfg = getattr(diffusion_server, "server_cfg", {})
endpoint = _resolve_benchmark_endpoint(server_cfg, params)
sweep_runs = _iter_sweep_runs(params)
result = run_benchmark(
host=diffusion_server.host,
port=diffusion_server.port,
model=diffusion_server.model,
params=params,
test_name=test_name,
endpoint=endpoint,
server_cfg=server_cfg,
source_file=cast(str, CONFIG_FILE_PATH),
)
for sweep_run in sweep_runs:
endpoint = _resolve_benchmark_endpoint(server_cfg, sweep_run["params"])
result = run_benchmark(
host=diffusion_server.host,
port=diffusion_server.port,
model=diffusion_server.model,
params=sweep_run["params"],
test_name=test_name,
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}, endpoint={endpoint}):")
for key in (
"throughput_qps",
"latency_mean",
"latency_median",
"latency_p50",
"latency_p99",
"peak_memory_mb_max",
"peak_memory_mb_mean",
"peak_memory_mb_median",
):
if key in result:
print(f" {key}: {result[key]:.4f}")
print(f"\n{'=' * 60}")
print(f"Results for {test_name} (server={diffusion_server.server_type}, endpoint={endpoint}):")
for key in (
"throughput_qps",
"latency_mean",
"latency_median",
"latency_p50",
"latency_p99",
"peak_memory_mb_max",
"peak_memory_mb_mean",
"peak_memory_mb_median",
):
if key in result:
print(f" {key}: {result[key]:.4f}")
print(f"\n Aggregated results: {AGGREGATED_RESULT_FILE}")
print("=" * 60)
print(f"\n Aggregated results: {AGGREGATED_RESULT_FILE}")
print("=" * 60)
assert_baseline = request.config.getoption("--assert-baseline", default=False)
assert_result(result, params, assert_baseline=assert_baseline)
assert_baseline = request.config.getoption("--assert-baseline", default=False)
assert_result(
result,
params,
sweep_run["num_prompts"],
sweep_index=sweep_run["sweep_index"],
max_concurrency=sweep_run["max_concurrency"],
request_rate=sweep_run["request_rate"],
assert_baseline=assert_baseline,
)