diff --git a/tests/dfx/perf/scripts/run_diffusion_benchmark.py b/tests/dfx/perf/scripts/run_diffusion_benchmark.py index 22e844427..d4ae73fa6 100644 --- a/tests/dfx/perf/scripts/run_diffusion_benchmark.py +++ b/tests/dfx/perf/scripts/run_diffusion_benchmark.py @@ -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, + ) diff --git a/tests/dfx/perf/tests/test_qwen_image_vllm_omni.json b/tests/dfx/perf/tests/test_qwen_image_vllm_omni.json index e05f1b945..a6eacf88e 100644 --- a/tests/dfx/perf/tests/test_qwen_image_vllm_omni.json +++ b/tests/dfx/perf/tests/test_qwen_image_vllm_omni.json @@ -181,5 +181,39 @@ } } ] + }, + { + "test_name": "test_qwen_image_single_device_step_execution_high_concurrency", + "description": "Single-device Qwen-Image step-level execution benchmark at 512x512, 20 steps, concurrency sweep", + "server_type": "vllm-omni", + "server_params": { + "model": "Qwen/Qwen-Image", + "serve_args": { + "enable-diffusion-pipeline-profiler": true, + "step-execution": true, + "max-num-seqs": 8 + } + }, + "benchmark_params": [ + { + "name": "512x512_steps20_high_concurrency", + "dataset": "random", + "task": "t2i", + "width": 512, + "height": 512, + "num-inference-steps": 20, + "num-prompts": [20, 40, 80, 160], + "max-concurrency": [1, 2, 4, 8], + "warmup-requests": 8, + "warmup-concurrency": 8, + "warmup-num-inference-steps": 20, + "enable-negative-prompt": true, + "baseline": { + "throughput_qps": [0.30, 0.50, 0.60, 0.70], + "latency_mean": [3.50, 4.00, 7.00, 12.00], + "peak_memory_mb_mean": [67000, 67000, 67000, 67000] + } + } + ] } ]