diff --git a/tests/benchmarks/patch/test_patch.py b/tests/benchmarks/patch/test_patch.py index 3eb77e8001e..6ee5e66cbff 100644 --- a/tests/benchmarks/patch/test_patch.py +++ b/tests/benchmarks/patch/test_patch.py @@ -2045,3 +2045,79 @@ def fake_get_samples_old(args, tokenizer, **kwargs): # No upstream kwargs: unchanged legacy delegate call. assert patch.get_samples(args, sentinel) == ["delegated"] assert calls[-1] == (args, sentinel, {}) + + +@pytest.mark.asyncio +async def test_benchmark_preserves_stage_metrics_request_order_and_missing_snapshots(monkeypatch): + """Persist compact formal-request snapshots in input order, excluding warmups and retaining gaps.""" + second_finished = asyncio.Event() + completion_order = [] + # Request 1 carries the empty dict that openai-chat-omni initializes before any SSE merge. + snapshots = [ + {"1": {"num_tokens_out": 1536, "finish_reason": "length", "vllm_itls_ms": [8.0, 9.0]}}, + {}, + {"1": {"num_tokens_out": 486, "finish_reason": "stop"}, "2": {"audio_frames": 24000, "audio_duration_s": 1.0}}, + ] + expected = [ + {"1": {"num_tokens_out": 1536, "finish_reason": "length"}}, + None, + {"1": {"num_tokens_out": 486, "finish_reason": "stop"}, "2": {"audio_frames": 24000, "audio_duration_s": 1.0}}, + ] + + async def request_func(request_func_input, session, pbar=None): + request_id = request_func_input.request_id + if not request_id: + return MixRequestFuncOutput(success=True, stage_metrics={"1": {"num_tokens_out": 999}}) + index = int(request_id) + if index == 0: + await second_finished.wait() + elif index == 1: + second_finished.set() + completion_order.append(index) + return MixRequestFuncOutput( + success=index != 1, + stage_metrics=snapshots[index], + prompt_len=1, + output_tokens=900, + ttft=0.01, + text_latency=0.1, + latency=0.1, + ) + + monkeypatch.setitem(patch.ASYNC_REQUEST_FUNCS, "test-stage-metrics", request_func) + result = await patch.benchmark( + task_type=patch.TaskType.GENERATION, + endpoint_type="test-stage-metrics", + api_url="http://unused/v1/chat/completions", + base_url="http://unused", + model_id="test-model", + model_name="test-model", + tokenizer=None, + input_requests=[ + patch.SampleRequest(prompt="hello", prompt_len=1, expected_output_len=900, request_id=str(i)) + for i in range(3) + ], + logprobs=None, + request_rate=float("inf"), + burstiness=1.0, + disable_tqdm=True, + num_warmups=2, + profile=False, + selected_percentile_metrics=[], + selected_percentiles=[], + ignore_eos=True, + goodput_config_dict={}, + max_concurrency=3, + lora_modules=None, + extra_headers=None, + extra_body={"return_stage_metrics": True}, + ready_check_timeout_sec=0, + ) + assert completion_order.index(1) < completion_order.index(0) + assert result["request_stage_metrics"] == expected + + +@pytest.mark.parametrize("snapshot", [None, {}, "not-a-dict"]) +def test_compact_request_stage_metrics_drops_empty_snapshots(snapshot): + """Empty chat-omni snapshots must not make every result persist request_stage_metrics.""" + assert patch._compact_request_stage_metrics(snapshot) is None diff --git a/tests/dfx/perf/scripts/run_benchmark.py b/tests/dfx/perf/scripts/run_benchmark.py index e026aa8878b..67689e13653 100644 --- a/tests/dfx/perf/scripts/run_benchmark.py +++ b/tests/dfx/perf/scripts/run_benchmark.py @@ -332,6 +332,36 @@ def _resolve_num_warmups(params: dict[str, Any], *, default: int) -> int: return value +def _assert_fixed_stage_workloads(result: dict[str, Any], params: dict[str, Any], num_prompts: int) -> None: + """Reject successful responses that did less work than a fixed-length benchmark requested.""" + extra_body = params.get("extra_body") or params.get("extra-body") or {} + fixed_lengths = { + str(stage): sampling["max_tokens"] + for stage, sampling in enumerate(extra_body.get("sampling_params_list", [])) + if isinstance(sampling, dict) + and isinstance(sampling.get("max_tokens"), int) + and sampling["max_tokens"] > 0 + and sampling.get("min_tokens") == sampling["max_tokens"] + } + if not fixed_lengths: + return + + snapshots = result.get("request_stage_metrics") + assert isinstance(snapshots, list) and len(snapshots) == num_prompts, ( + "Fixed stage workload: missing per-request stage metrics" + ) + for request_index, snapshot in enumerate(snapshots): + for stage, expected in fixed_lengths.items(): + metrics = snapshot.get(stage) if isinstance(snapshot, dict) else None + actual = metrics.get("num_tokens_out") if isinstance(metrics, dict) else None + assert actual == expected, ( + f"Fixed stage workload: request {request_index}, stage {stage}: " + f"expected {expected} output tokens, got {actual!r}" + ) + stages = ", ".join(f"stage {stage}={expected}" for stage, expected in fixed_lengths.items()) + print(f"Fixed stage workload OK: {num_prompts} requests, num_tokens_out {stages}") + + def _is_finite_number(value: object) -> bool: return isinstance(value, int | float) and not isinstance(value, bool) and math.isfinite(value) @@ -448,6 +478,7 @@ def _maybe_assert_aggregate_ia_qtf1(params: dict[str, object], acc_summary: dict def assert_result(result, params, num_prompt) -> None: assert result["completed"] == num_prompt, "Request failures exist" + _assert_fixed_stage_workloads(result, params, num_prompt) if params.get("dataset_name") == "omniinteract": summary = result.get("omniinteract") assert isinstance(summary, dict), "OmniInteract summary is missing" diff --git a/tests/dfx/perf/tests/test_qwen3_omni_no_async_chunk.json b/tests/dfx/perf/tests/test_qwen3_omni_no_async_chunk.json index 186bac4521b..9fb31e273dc 100644 --- a/tests/dfx/perf/tests/test_qwen3_omni_no_async_chunk.json +++ b/tests/dfx/perf/tests/test_qwen3_omni_no_async_chunk.json @@ -1,6 +1,7 @@ [ { "test_name": "test_qwen3_omni", + "description": "random row: fixed workload (Thinker 900 / Talker 1536 tokens, min_tokens == max_tokens) for comparable audio RTF; H100 baseline pending recalibration, see #8100", "mark": [ { "hardware_marks": { @@ -43,17 +44,28 @@ "random_output_len": 900, "ignore_eos": true, "percentile-metrics": "ttft,tpot,itl,e2el,audio_rtf,audio_ttfp,audio_duration", - "baseline": { - "H100": { - "mean_audio_rtf": [ - 0.169, - 0.2041, - 0.2382, - 0.3177, - 0.4595 - ] - } - } + "seed": 0, + "random_range_ratio": 0.0, + "extra_body": { + "return_stage_metrics": true, + "sampling_params_list": [ + { + "temperature": 0.0, + "min_tokens": 900, + "max_tokens": 900, + "ignore_eos": true + }, + { + "temperature": 0.9, + "top_k": 50, + "repetition_penalty": 1.05, + "seed": 0, + "min_tokens": 1536, + "max_tokens": 1536 + } + ] + }, + "baseline": {} }, { "dataset_name": "random-mm", diff --git a/tests/dfx/perf/tests/test_runner_metadata.py b/tests/dfx/perf/tests/test_runner_metadata.py index 044e105c22e..0eb5b9f75b1 100644 --- a/tests/dfx/perf/tests/test_runner_metadata.py +++ b/tests/dfx/perf/tests/test_runner_metadata.py @@ -932,3 +932,58 @@ def test_omni_tpot_baseline_rejects_missing_or_nonfinite_sample(num_tpot_samples {"baseline": {"H100": {"mean_tpot_ms": 20.0}}}, 1, ) + + +# Mirrors the Thinker/Talker lengths in test_qwen3_omni_no_async_chunk.json. +_FIXED_SAMPLING = { + "sampling_params_list": [{"min_tokens": 900, "max_tokens": 900}, {"min_tokens": 1536, "max_tokens": 1536}] +} +_FIXED_OK = {"0": {"num_tokens_out": 900}, "1": {"num_tokens_out": 1536}} + + +@pytest.mark.parametrize("extra_body_key", ["extra_body", "extra-body"]) +def test_fixed_stage_workload_accepts_exact_lengths(extra_body_key, capsys): + from tests.dfx.perf.scripts.run_benchmark import assert_result + + assert_result( + {"completed": 2, "failed": 0, "request_stage_metrics": [_FIXED_OK, _FIXED_OK]}, + {extra_body_key: _FIXED_SAMPLING}, + 2, + ) + assert "Fixed stage workload OK: 2 requests, num_tokens_out stage 0=900, stage 1=1536" in capsys.readouterr().out + + +@pytest.mark.parametrize( + "snapshots", + [ + None, + [], + [_FIXED_OK], + [_FIXED_OK, None], + [_FIXED_OK, {}], + [_FIXED_OK, {"0": {"num_tokens_out": 900}, "1": {}}], + [_FIXED_OK, {"0": {"num_tokens_out": 900}, "1": {"num_tokens_out": 486}}], + [_FIXED_OK, {"0": {"num_tokens_out": 900}, "1": {"num_tokens_out": 1537}}], + [_FIXED_OK, {"1": {"num_tokens_out": 1536}}], + [_FIXED_OK, {"0": {"num_tokens_out": 512}, "1": {"num_tokens_out": 1536}}], + ], +) +def test_fixed_stage_workload_rejects_missing_or_wrong_lengths(snapshots): + from tests.dfx.perf.scripts.run_benchmark import assert_result + + with pytest.raises(AssertionError, match="Fixed stage workload"): + assert_result({"completed": 2, "request_stage_metrics": snapshots}, {"extra_body": _FIXED_SAMPLING}, 2) + + +def test_fixed_stage_workload_rejects_request_failures(): + from tests.dfx.perf.scripts.run_benchmark import assert_result + + with pytest.raises(AssertionError, match="Request failures"): + assert_result({"completed": 1, "failed": 1}, {}, 2) + + +@pytest.mark.parametrize("extra_body", [{}, {"sampling_params_list": [{"max_tokens": 900}]}]) +def test_variable_stage_workload_does_not_require_stage_metrics(extra_body): + from tests.dfx.perf.scripts.run_benchmark import assert_result + + assert_result({"completed": 2}, {"extra_body": extra_body}, 2) diff --git a/vllm_omni/benchmarks/patch/patch.py b/vllm_omni/benchmarks/patch/patch.py index f03fe6c9b03..b0ca826df09 100644 --- a/vllm_omni/benchmarks/patch/patch.py +++ b/vllm_omni/benchmarks/patch/patch.py @@ -1203,6 +1203,27 @@ def _update_output_stage_metrics_from_payload( output.stage_metrics.update(stage_snapshot) +# Per-request stage fields persisted in benchmark results. Full snapshots carry +# per-token latency lists, which would inflate every chat-omni result file. +_REQUEST_STAGE_METRIC_FIELDS = ( + defs.NUM_TOKENS_OUT, + "finish_reason", + defs.AUDIO_FRAMES, + f"{defs.AUDIO_DURATION}_s", +) + + +def _compact_request_stage_metrics(snapshot: object) -> dict[str, dict] | None: + """Keep the stage fields used for workload checks; empty snapshots become None.""" + if not isinstance(snapshot, dict) or not snapshot: + return None + return { + stage: {field: metrics[field] for field in _REQUEST_STAGE_METRIC_FIELDS if field in metrics} + for stage, metrics in snapshot.items() + if isinstance(metrics, dict) + } + + def _apply_chat_stage0_token_timings(output: MixRequestFuncOutput) -> bool: """Apply native Stage 0 timings from a chat response snapshot.""" stage_metrics = output.stage_metrics @@ -3635,6 +3656,14 @@ def measured_ttft(output: RequestFuncOutput) -> float | None: "input_lens": [output.prompt_len for output in outputs], "errors": [output.error for output in outputs], } + # Preserve request order, including missing snapshots, so CI can verify + # fixed stage workloads without parsing logs or storing audio payloads. + request_stage_metrics = [ + _compact_request_stage_metrics(getattr(output, "stage_metrics", None)) for output in outputs + ] + if any(request_stage_metrics): + result["request_stage_metrics"] = request_stage_metrics + # Plain-vLLM backends (e.g. the vLLM-text perf config) return upstream # RequestFuncOutput objects without the Mix duplex fields; read them # tolerantly or the whole benchmark result is discarded ("fallback to