[diffusion] CI: validate every repeated server request (#38185)
This commit is contained in:
Mick
@@ -2,6 +2,37 @@
|
||||
title: "CI Performance Baselines"
|
||||
description: "Generate and update diffusion performance baselines used in CI."
|
||||
---
|
||||
## Sequential Request Checks
|
||||
|
||||
Server cases can exercise request-to-request state without restarting the server:
|
||||
|
||||
```python
|
||||
DiffusionTestCase(
|
||||
id="image_t2i",
|
||||
server_args=server_args,
|
||||
sampling_params=sampling_params,
|
||||
perf_repeat_requests=2,
|
||||
)
|
||||
```
|
||||
|
||||
Every request is checked, including the first one. Performance checks include
|
||||
load/runtime peak memory; consistency checks include audio when
|
||||
`expect_audio_output=True`. A failure in either request fails the case, even if
|
||||
the other succeeds. Reports retain the request index instead of overwriting the
|
||||
first result with the second.
|
||||
|
||||
Keep workloads small, but regenerate baselines and GT whenever sampling
|
||||
parameters change. MiniMax-H3 T2VA and ref2va use different checkpoint
|
||||
partitions, so they remain separate cases, each issuing two requests to its
|
||||
own server.
|
||||
The T2VA case includes VAE layerwise offload to exercise weight restoration
|
||||
between requests; ref2va retains the resident VAE path.
|
||||
|
||||
Use `SGLANG_GEN_BASELINE=1 pytest -s ... -k <case_id>` to collect baselines in
|
||||
the same sequential order, retaining the worst observed value of each metric.
|
||||
`SGLANG_GEN_GT=1` with `SGLANG_GT_OUTPUT_DIR` saves
|
||||
GT; repeated requests share one GT.
|
||||
|
||||
## Perf Baseline Generation Script
|
||||
|
||||
`python/sglang/multimodal_gen/test/scripts/gen_perf_baselines.py` starts a local diffusion server, issues requests for selected test cases, aggregates stage/denoise-step/E2E timings from the perf log, and writes the results back to the `scenarios` section of `perf_baselines.json`.
|
||||
|
||||
@@ -3,33 +3,13 @@ import os
|
||||
|
||||
import pytest
|
||||
|
||||
print("[CONFTEST] Loading conftest.py at import time")
|
||||
|
||||
|
||||
def pytest_configure(config):
|
||||
"""
|
||||
Create the perf results StashKey once and store it in config.
|
||||
This hook runs once per test session, before module double-import issues.
|
||||
"""
|
||||
if not hasattr(config, "_diffusion_perf_key"):
|
||||
config._diffusion_perf_key = pytest.StashKey[list]()
|
||||
print(f"[CONFTEST] Created perf_results_key: {config._diffusion_perf_key}")
|
||||
|
||||
|
||||
def add_perf_results(config, results: list):
|
||||
"""Add performance results to the shared stash."""
|
||||
# Get the shared key from config (created once in pytest_configure)
|
||||
key = config._diffusion_perf_key
|
||||
existing = config.stash.get(key, [])
|
||||
existing.extend(results)
|
||||
config.stash[key] = existing
|
||||
print(f"[CONFTEST] Added {len(results)} results, total now: {len(existing)}")
|
||||
_PERF_RESULTS = pytest.StashKey[list]()
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def perf_config(request):
|
||||
"""Provide access to pytest config for storing perf results."""
|
||||
return request.config
|
||||
def perf_results(request):
|
||||
"""Share results through pytest rather than importing this conftest module."""
|
||||
return request.config.stash.setdefault(_PERF_RESULTS, [])
|
||||
|
||||
|
||||
def _write_github_step_summary(content: str):
|
||||
@@ -54,15 +34,13 @@ def _write_results_json(results: list, output_path: str = "diffusion-results.jso
|
||||
pass
|
||||
|
||||
merged = {
|
||||
(entry.get("class_name"), entry.get("test_name")): entry
|
||||
for entry in existing
|
||||
(
|
||||
entry.get("class_name"),
|
||||
entry.get("test_name"),
|
||||
entry.get("request_index", 1),
|
||||
): entry
|
||||
for entry in existing + results
|
||||
}
|
||||
merged.update(
|
||||
{
|
||||
(entry.get("class_name"), entry.get("test_name")): entry
|
||||
for entry in results
|
||||
}
|
||||
)
|
||||
with open(output_path, "w") as f:
|
||||
json.dump(list(merged.values()), f, indent=2)
|
||||
print(f"[CONFTEST] Wrote results to {output_path}")
|
||||
@@ -83,13 +61,13 @@ def _generate_diffusion_markdown_report(results: list) -> str:
|
||||
|
||||
# Main performance table
|
||||
markdown = header
|
||||
markdown += "| Test Suite | Test Name | Modality | E2E (ms) | Avg Denoise (ms) | Median Denoise (ms) | Load Peak VRAM (MiB) | Runtime Peak VRAM (MiB) | Load Peak Alloc (MiB) | Runtime Peak Alloc (MiB) |\n"
|
||||
markdown += "| ---------- | --------- | -------- | -------- | ---------------- | ------------------- | -------------------- | ----------------------- | --------------------- | ------------------------ |\n"
|
||||
markdown += "| Test Suite | Test Name | Request | Modality | E2E (ms) | Avg Denoise (ms) | Median Denoise (ms) | Load Peak VRAM (MiB) | Runtime Peak VRAM (MiB) | Load Peak Alloc (MiB) | Runtime Peak Alloc (MiB) |\n"
|
||||
markdown += "| ---------- | --------- | ------- | -------- | -------- | ---------------- | ------------------- | -------------------- | ----------------------- | --------------------- | ------------------------ |\n"
|
||||
|
||||
for entry in sorted(results, key=lambda x: (x["class_name"], x["test_name"])):
|
||||
modality = entry.get("modality", "image")
|
||||
markdown += (
|
||||
f"| {entry['class_name']} | {entry['test_name']} | {modality} | "
|
||||
f"| {entry['class_name']} | {entry['test_name']} | {entry.get('request_index', 1)} | {modality} | "
|
||||
f"{entry['e2e_ms']:.2f} | {entry['avg_denoise_ms']:.2f} | "
|
||||
f"{entry['median_denoise_ms']:.2f} | "
|
||||
f"{entry.get('load_peak_vram_mb', 0):.0f} | "
|
||||
@@ -102,8 +80,12 @@ def _generate_diffusion_markdown_report(results: list) -> str:
|
||||
video_results = [r for r in results if r.get("modality") == "video"]
|
||||
if video_results:
|
||||
markdown += "\n### Video Generation Metrics\n\n"
|
||||
markdown += "| Test Name | FPS | Total Frames | Avg Frame Time (ms) |\n"
|
||||
markdown += "| --------- | --- | ------------ | ------------------- |\n"
|
||||
markdown += (
|
||||
"| Test Name | Request | FPS | Total Frames | Avg Frame Time (ms) |\n"
|
||||
)
|
||||
markdown += (
|
||||
"| --------- | ------- | --- | ------------ | ------------------- |\n"
|
||||
)
|
||||
for entry in video_results:
|
||||
fps = entry.get("frames_per_second", "N/A")
|
||||
frames = entry.get("total_frames", "N/A")
|
||||
@@ -112,7 +94,7 @@ def _generate_diffusion_markdown_report(results: list) -> str:
|
||||
fps = f"{fps:.2f}"
|
||||
if isinstance(avg_frame, float):
|
||||
avg_frame = f"{avg_frame:.2f}"
|
||||
markdown += f"| {entry['test_name']} | {fps} | {frames} | {avg_frame} |\n"
|
||||
markdown += f"| {entry['test_name']} | {entry.get('request_index', 1)} | {fps} | {frames} | {avg_frame} |\n"
|
||||
|
||||
return markdown
|
||||
|
||||
@@ -122,26 +104,27 @@ def pytest_sessionfinish(session):
|
||||
This hook is called by pytest at the end of the entire test session.
|
||||
It prints a consolidated summary of all performance results.
|
||||
"""
|
||||
# Get results from stash using the shared key from config
|
||||
key = session.config._diffusion_perf_key
|
||||
results = session.config.stash.get(key, [])
|
||||
print(f"\n[DEBUG] pytest_sessionfinish called, has {len(results)} entries")
|
||||
results = session.config.stash.get(_PERF_RESULTS, [])
|
||||
if not results:
|
||||
print("[DEBUG] No results collected, skipping summary output")
|
||||
return
|
||||
|
||||
sorted_results = sorted(results, key=lambda x: (x["class_name"], x["test_name"]))
|
||||
sorted_results = sorted(
|
||||
results,
|
||||
key=lambda x: (x["class_name"], x["test_name"], x.get("request_index", 1)),
|
||||
)
|
||||
|
||||
# Print to stdout (existing behavior)
|
||||
print("\n\n" + "=" * 35 + " Performance Summary " + "=" * 35)
|
||||
print(
|
||||
f"{'Test Suite':<30} | {'Test Name':<20} | {'E2E (ms)':>12} | {'Avg Denoise (ms)':>18} | {'Median Denoise (ms)':>20} | {'Load Peak (MiB)':>15} | {'Runtime Peak (MiB)':>18} | {'Load Alloc (MiB)':>16} | {'Runtime Alloc (MiB)':>19}"
|
||||
f"{'Test Suite':<30} | {'Test Name':<20} | {'Request':>7} | {'E2E (ms)':>12} | {'Avg Denoise (ms)':>18} | {'Median Denoise (ms)':>20} | {'Load Peak (MiB)':>15} | {'Runtime Peak (MiB)':>18} | {'Load Alloc (MiB)':>16} | {'Runtime Alloc (MiB)':>19}"
|
||||
)
|
||||
print(
|
||||
"-" * 30
|
||||
+ "-+-"
|
||||
+ "-" * 20
|
||||
+ "-+-"
|
||||
+ "-" * 7
|
||||
+ "-+-"
|
||||
+ "-" * 12
|
||||
+ "-+-"
|
||||
+ "-" * 18
|
||||
@@ -159,7 +142,7 @@ def pytest_sessionfinish(session):
|
||||
|
||||
for entry in sorted_results:
|
||||
print(
|
||||
f"{entry['class_name']:<30} | {entry['test_name']:<20} | {entry['e2e_ms']:>12.2f} | "
|
||||
f"{entry['class_name']:<30} | {entry['test_name']:<20} | {entry.get('request_index', 1):>7} | {entry['e2e_ms']:>12.2f} | "
|
||||
f"{entry['avg_denoise_ms']:>18.2f} | {entry['median_denoise_ms']:>20.2f} | "
|
||||
f"{entry.get('load_peak_vram_mb', 0):>15.0f} | "
|
||||
f"{entry.get('runtime_peak_vram_mb', 0):>18.0f} | "
|
||||
@@ -171,7 +154,10 @@ def pytest_sessionfinish(session):
|
||||
|
||||
print("\n\n" + "=" * 36 + " Detailed Reports " + "=" * 37)
|
||||
for entry in sorted_results:
|
||||
print(f"\n--- Details for {entry['class_name']} / {entry['test_name']} ---")
|
||||
print(
|
||||
f"\n--- Details for {entry['class_name']} / {entry['test_name']} "
|
||||
f"/ request {entry.get('request_index', 1)} ---"
|
||||
)
|
||||
stage_report = ", ".join(
|
||||
f"{name}:{duration:.2f}ms"
|
||||
for name, duration in entry.get("stage_metrics", {}).items()
|
||||
|
||||
@@ -747,9 +747,9 @@ TWO_GPU_CASES = [
|
||||
"--performance-mode",
|
||||
"memory",
|
||||
"--layerwise-offload-components",
|
||||
"dit,text_encoder",
|
||||
"--component-residency",
|
||||
"vae=resident",
|
||||
"dit,text_encoder,vae",
|
||||
"--layerwise-resident-layers",
|
||||
"video_vae=36",
|
||||
"--dit-offload-prefetch-size",
|
||||
"1",
|
||||
"--dit-layerwise-resident-layers",
|
||||
@@ -787,6 +787,7 @@ TWO_GPU_CASES = [
|
||||
},
|
||||
),
|
||||
run_perf_check=True,
|
||||
perf_repeat_requests=2,
|
||||
run_consistency_check=True,
|
||||
run_component_accuracy_check=False,
|
||||
run_models_api_check=False,
|
||||
@@ -853,7 +854,7 @@ TWO_GPU_CASES = [
|
||||
"seed": 42,
|
||||
},
|
||||
),
|
||||
run_perf_check=False,
|
||||
perf_repeat_requests=2,
|
||||
run_consistency_check=True,
|
||||
run_component_accuracy_check=False,
|
||||
run_models_api_check=False,
|
||||
|
||||
@@ -3037,12 +3037,32 @@
|
||||
"estimated_full_test_time_s": 52.2
|
||||
},
|
||||
"minimax_h3_ref2va_video_audio_2gpu_h100": {
|
||||
"stages_ms": {},
|
||||
"denoise_step_ms": {},
|
||||
"expected_e2e_ms": 0.0,
|
||||
"expected_avg_denoise_ms": 0.0,
|
||||
"expected_median_denoise_ms": 0.0,
|
||||
"estimated_full_test_time_s": 170.9
|
||||
"stages_ms": {
|
||||
"InputValidationStage": 0.08,
|
||||
"MiniMaxH3PartitionAdmissionStage": 0.04,
|
||||
"MiniMaxH3TextEncodingStage": 1429.48,
|
||||
"MiniMaxH3VisualEncodingStage": 7065.71,
|
||||
"MiniMaxH3AudioEncodingStage": 647.92,
|
||||
"MiniMaxH3LatentPreparationStage": 24.37,
|
||||
"MiniMaxH3TimestepPreparationStage": 0.28,
|
||||
"MiniMaxH3DenoisingStage": 55168.59,
|
||||
"MiniMaxH3DecodingStage": 3183.0
|
||||
},
|
||||
"denoise_step_ms": {
|
||||
"0": 1064.67,
|
||||
"1": 7783.29,
|
||||
"2": 7809.31,
|
||||
"3": 7796.83,
|
||||
"4": 7802.67,
|
||||
"5": 7813.31,
|
||||
"6": 7807.76
|
||||
},
|
||||
"expected_e2e_ms": 67793.18,
|
||||
"expected_avg_denoise_ms": 6839.69,
|
||||
"expected_median_denoise_ms": 7802.67,
|
||||
"load_peak_vram_mb": 15242.0,
|
||||
"runtime_peak_vram_mb": 34560.0,
|
||||
"estimated_full_test_time_s": 340.0
|
||||
},
|
||||
"minimax_h3_t2va_2gpu_h100": {
|
||||
"stages_ms": {
|
||||
@@ -3073,7 +3093,7 @@
|
||||
"runtime_peak_vram_mb": 63312.0,
|
||||
"load_peak_allocated_mb": 15139.0,
|
||||
"runtime_peak_allocated_mb": 22887.0,
|
||||
"estimated_full_test_time_s": 103.9
|
||||
"estimated_full_test_time_s": 235.0
|
||||
},
|
||||
"mova_360p_tp2": {
|
||||
"stages_ms": {},
|
||||
|
||||
@@ -24,7 +24,6 @@ from openai import OpenAI
|
||||
from sglang.multimodal_gen.runtime.platforms import current_platform
|
||||
from sglang.multimodal_gen.runtime.utils.logging_utils import init_logger
|
||||
from sglang.multimodal_gen.runtime.utils.perf_logger import RequestPerfRecord
|
||||
from sglang.multimodal_gen.test.server import conftest
|
||||
from sglang.multimodal_gen.test.server.realtime_consistency import (
|
||||
RealtimeChunkStats,
|
||||
pop_realtime_key_frames,
|
||||
@@ -83,7 +82,7 @@ logger = init_logger(__name__)
|
||||
|
||||
# Track test cases missing estimated_full_test_time_s for time measurement output
|
||||
_MISSING_ESTIMATED_TIME_CASES: set[str] = set()
|
||||
_PENDING_BASELINE_DUMPS: dict[str, tuple[PerformanceSummary, bool]] = {}
|
||||
_PENDING_BASELINE_DUMPS: dict[str, list[PerformanceSummary]] = {}
|
||||
_OPENAI_REQUEST_TIMEOUT_SECS = float(
|
||||
os.environ.get("SGLANG_TEST_OPENAI_REQUEST_TIMEOUT_SECS", "600")
|
||||
)
|
||||
@@ -228,12 +227,12 @@ def diffusion_server(case: DiffusionTestCase) -> ServerContext:
|
||||
|
||||
pending_dump = _PENDING_BASELINE_DUMPS.pop(case.id, None)
|
||||
if pending_dump is not None:
|
||||
summary, missing_scenario = pending_dump
|
||||
DiffusionServerBase()._dump_baseline_for_testcase(
|
||||
case,
|
||||
summary,
|
||||
missing_scenario=missing_scenario,
|
||||
pending_dump[-1],
|
||||
missing_scenario=case.id not in BASELINE_CONFIG.scenarios,
|
||||
measured_full_time=_measured_full_time,
|
||||
repeated_summaries=pending_dump,
|
||||
)
|
||||
|
||||
scenario = BASELINE_CONFIG.scenarios.get(case.id)
|
||||
@@ -263,35 +262,14 @@ class DiffusionServerBase:
|
||||
Each case gets its own server instance via the parametrized fixture.
|
||||
"""
|
||||
|
||||
_perf_results: list[dict[str, Any]] = []
|
||||
_pytest_config = None # Store pytest config for stash access
|
||||
|
||||
@classmethod
|
||||
def setup_class(cls):
|
||||
cls._perf_results = []
|
||||
|
||||
@classmethod
|
||||
def teardown_class(cls):
|
||||
print(
|
||||
f"\n[DEBUG teardown_class] Called for {cls.__name__}, _perf_results has {len(cls._perf_results)} entries"
|
||||
)
|
||||
if cls._pytest_config:
|
||||
# Add results to pytest stash (shared across all import contexts)
|
||||
for result in cls._perf_results:
|
||||
result["class_name"] = cls.__name__
|
||||
conftest.add_perf_results(cls._pytest_config, cls._perf_results)
|
||||
print(
|
||||
f"[DEBUG teardown_class] Added {len(cls._perf_results)} results to stash"
|
||||
)
|
||||
else:
|
||||
print(
|
||||
"[DEBUG teardown_class] No pytest_config available, skipping stash update"
|
||||
)
|
||||
_perf_results: list[dict[str, Any]]
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _capture_pytest_config(self, request):
|
||||
"""Capture pytest config for use in teardown_class."""
|
||||
self.__class__._pytest_config = request.config
|
||||
def _collect_perf_results(self, perf_results):
|
||||
"""Keep case results isolated and retain them even when validation fails."""
|
||||
self._perf_results = []
|
||||
yield
|
||||
perf_results.extend(self._perf_results)
|
||||
|
||||
def _client(self, ctx: ServerContext) -> OpenAI:
|
||||
"""Get OpenAI client for the server."""
|
||||
@@ -405,6 +383,7 @@ class DiffusionServerBase:
|
||||
self,
|
||||
case: DiffusionTestCase,
|
||||
perf_record: RequestPerfRecord,
|
||||
request_index: int = 1,
|
||||
) -> None:
|
||||
"""Validate metrics and record results."""
|
||||
is_baseline_generation_mode = os.environ.get("SGLANG_GEN_BASELINE", "0") == "1"
|
||||
@@ -434,10 +413,11 @@ class DiffusionServerBase:
|
||||
|
||||
summary = validator.collect_metrics(perf_record)
|
||||
self._print_performance_log(case, summary, scenario)
|
||||
self._record_performance_result(case, summary, request_index)
|
||||
|
||||
if case.run_perf_check:
|
||||
if is_baseline_generation_mode:
|
||||
_PENDING_BASELINE_DUMPS[case.id] = (summary, missing_scenario)
|
||||
_PENDING_BASELINE_DUMPS.setdefault(case.id, []).append(summary)
|
||||
return
|
||||
|
||||
if missing_scenario:
|
||||
@@ -489,13 +469,12 @@ class DiffusionServerBase:
|
||||
self._dump_baseline_for_testcase(case, summary, missing_scenario)
|
||||
raise
|
||||
|
||||
self._record_performance_result(case, summary)
|
||||
|
||||
def _validate_realtime_performance(
|
||||
self,
|
||||
ctx: ServerContext,
|
||||
case: DiffusionTestCase,
|
||||
chunk_stats: list[RealtimeChunkStats],
|
||||
request_index: int = 1,
|
||||
) -> None:
|
||||
validate_realtime_perf_stats(
|
||||
case.id,
|
||||
@@ -536,7 +515,7 @@ class DiffusionServerBase:
|
||||
)
|
||||
summary = validator.collect_metrics(perf_record)
|
||||
self._print_performance_log(case, summary, scenario)
|
||||
self._record_performance_result(case, summary)
|
||||
self._record_performance_result(case, summary, request_index)
|
||||
|
||||
if os.environ.get("SGLANG_GEN_BASELINE", "0") == "1":
|
||||
logger.info(
|
||||
@@ -574,9 +553,12 @@ class DiffusionServerBase:
|
||||
self,
|
||||
case: DiffusionTestCase,
|
||||
summary: PerformanceSummary,
|
||||
request_index: int = 1,
|
||||
) -> None:
|
||||
result = {
|
||||
"class_name": type(self).__name__,
|
||||
"test_name": case.id,
|
||||
"request_index": request_index,
|
||||
"modality": case.server_args.modality,
|
||||
"e2e_ms": summary.e2e_ms,
|
||||
"avg_denoise_ms": summary.avg_denoise_ms,
|
||||
@@ -600,10 +582,7 @@ class DiffusionServerBase:
|
||||
}
|
||||
)
|
||||
|
||||
self.__class__._perf_results.append(result)
|
||||
print(
|
||||
f"[DEBUG _validate_and_record] Appended result for {case.id}, class {self.__class__.__name__} now has {len(self.__class__._perf_results)} results"
|
||||
)
|
||||
self._perf_results.append(result)
|
||||
|
||||
def _print_performance_log(
|
||||
self,
|
||||
@@ -663,36 +642,57 @@ class DiffusionServerBase:
|
||||
summary: PerformanceSummary,
|
||||
missing_scenario: bool = False,
|
||||
measured_full_time: float | None = None,
|
||||
repeated_summaries: list[PerformanceSummary] | None = None,
|
||||
) -> None:
|
||||
"""Dump performance metrics as a JSON scenario for baselines."""
|
||||
import json
|
||||
|
||||
# One shared baseline must cover both the first and subsequent requests.
|
||||
summaries = repeated_summaries or [summary]
|
||||
denoise_steps_formatted = {
|
||||
str(k): round(v, 2) for k, v in summary.all_denoise_steps.items()
|
||||
str(k): round(max(s.all_denoise_steps[k] for s in summaries), 2)
|
||||
for k in summary.all_denoise_steps
|
||||
}
|
||||
stages_formatted = {
|
||||
k: round(max(s.stage_metrics[k] for s in summaries), 2)
|
||||
for k in summary.stage_metrics
|
||||
}
|
||||
stages_formatted = {k: round(v, 2) for k, v in summary.stage_metrics.items()}
|
||||
|
||||
baseline = {
|
||||
"stages_ms": stages_formatted,
|
||||
"denoise_step_ms": denoise_steps_formatted,
|
||||
"expected_e2e_ms": round(summary.e2e_ms, 2),
|
||||
"expected_avg_denoise_ms": round(summary.avg_denoise_ms, 2),
|
||||
"expected_median_denoise_ms": round(summary.median_denoise_ms, 2),
|
||||
"expected_e2e_ms": round(max(s.e2e_ms for s in summaries), 2),
|
||||
"expected_avg_denoise_ms": round(
|
||||
max(s.avg_denoise_ms for s in summaries), 2
|
||||
),
|
||||
"expected_median_denoise_ms": round(
|
||||
max(s.median_denoise_ms for s in summaries), 2
|
||||
),
|
||||
}
|
||||
|
||||
if current_platform.is_cuda():
|
||||
baseline.update(
|
||||
{
|
||||
"load_peak_vram_mb": round(summary.load_peak_vram_mb, 2),
|
||||
"runtime_peak_vram_mb": round(summary.runtime_peak_vram_mb, 2),
|
||||
"warmup_peak_vram_mb": round(summary.warmup_peak_vram_mb, 2),
|
||||
"load_peak_allocated_mb": round(summary.load_peak_allocated_mb, 2),
|
||||
"load_peak_vram_mb": round(
|
||||
max(s.load_peak_vram_mb for s in summaries), 2
|
||||
),
|
||||
"runtime_peak_vram_mb": round(
|
||||
max(s.runtime_peak_vram_mb for s in summaries), 2
|
||||
),
|
||||
"warmup_peak_vram_mb": round(
|
||||
max(s.warmup_peak_vram_mb for s in summaries), 2
|
||||
),
|
||||
"load_peak_allocated_mb": round(
|
||||
max(s.load_peak_allocated_mb for s in summaries), 2
|
||||
),
|
||||
"runtime_peak_allocated_mb": round(
|
||||
summary.runtime_peak_allocated_mb, 2
|
||||
max(s.runtime_peak_allocated_mb for s in summaries), 2
|
||||
),
|
||||
"load_peak_host_anon_mb": round(
|
||||
max(s.load_peak_host_anon_mb for s in summaries), 2
|
||||
),
|
||||
"load_peak_host_anon_mb": round(summary.load_peak_host_anon_mb, 2),
|
||||
"runtime_peak_host_anon_mb": round(
|
||||
summary.runtime_peak_host_anon_mb, 2
|
||||
max(s.runtime_peak_host_anon_mb for s in summaries), 2
|
||||
),
|
||||
}
|
||||
)
|
||||
@@ -730,11 +730,10 @@ class DiffusionServerBase:
|
||||
return
|
||||
|
||||
if not content:
|
||||
logger.warning(
|
||||
f"[Consistency] Skipping consistency check for {case.id}: "
|
||||
"content is empty (generation may have timed out)"
|
||||
pytest.fail(
|
||||
f"[Consistency] Empty output for {case.id} "
|
||||
"(generation may have timed out)"
|
||||
)
|
||||
return
|
||||
|
||||
if case.server_args.modality == "action":
|
||||
self._validate_action_consistency(case, content)
|
||||
@@ -901,9 +900,6 @@ Pinned revision used by this check: {SGL_TEST_FILES_CI_DATA_REVISION}
|
||||
f"max_mean_abs_diff={result.max_mean_abs_diff:.4f})"
|
||||
)
|
||||
|
||||
if case.sampling_params.expect_audio_output:
|
||||
self._validate_audio_consistency(case, content)
|
||||
|
||||
def _validate_audio_consistency(
|
||||
self,
|
||||
case: DiffusionTestCase,
|
||||
@@ -1565,31 +1561,58 @@ Pinned revision used by this check: {SGL_TEST_FILES_CI_DATA_REVISION}
|
||||
case: DiffusionTestCase,
|
||||
diffusion_server: ServerContext,
|
||||
):
|
||||
# Check if we're in GT generation mode
|
||||
is_gt_gen_mode = os.environ.get("SGLANG_GEN_GT", "0") == "1"
|
||||
|
||||
# GT generation also needs the dynamic set_lora step before generation.
|
||||
if case.run_lora_dynamic_load_check:
|
||||
self._test_dynamic_lora_loading(diffusion_server, case)
|
||||
|
||||
failures = []
|
||||
for request_index in range(1, case.perf_repeat_requests + 1):
|
||||
label = f"request {request_index}/{case.perf_repeat_requests}"
|
||||
_print_case_log_separator(case.id, f"BEGIN {label}")
|
||||
try:
|
||||
with pytest.MonkeyPatch.context() as request_env:
|
||||
artifact_dir = os.environ.get("SGLANG_DIFFUSION_ARTIFACT_DIR")
|
||||
if artifact_dir and case.perf_repeat_requests > 1:
|
||||
request_env.setenv(
|
||||
"SGLANG_DIFFUSION_ARTIFACT_DIR",
|
||||
str(Path(artifact_dir) / f"request-{request_index}"),
|
||||
)
|
||||
self._test_diffusion_request(case, diffusion_server, request_index)
|
||||
except pytest.skip.Exception as exc:
|
||||
if request_index == 1:
|
||||
raise
|
||||
failures.append(f"[{label}] Required request skipped: {exc}")
|
||||
_print_case_log_separator(case.id, f"FAILED {label}")
|
||||
break
|
||||
except (Exception, pytest.fail.Exception) as exc:
|
||||
failures.append(f"[{label}] {exc}")
|
||||
_print_case_log_separator(case.id, f"FAILED {label}")
|
||||
else:
|
||||
_print_case_log_separator(case.id, f"PASSED {label}")
|
||||
|
||||
if failures:
|
||||
pytest.fail("\n\n".join(failures), pytrace=False)
|
||||
|
||||
def _test_diffusion_request(
|
||||
self,
|
||||
case: DiffusionTestCase,
|
||||
diffusion_server: ServerContext,
|
||||
request_index: int,
|
||||
):
|
||||
is_gt_gen_mode = os.environ.get("SGLANG_GEN_GT", "0") == "1"
|
||||
generate_fn = get_generate_fn(
|
||||
model_path=case.server_args.model_path,
|
||||
modality=case.server_args.modality,
|
||||
sampling_params=case.sampling_params,
|
||||
)
|
||||
|
||||
# Generation - output of the last request is used for both validations.
|
||||
# perf_repeat_requests > 1 asserts a warm second request meets the same
|
||||
# baselines as the first: residency or courier state leaking between
|
||||
# requests shows up here as degradation or an OOM.
|
||||
is_realtime_case = case.sampling_params.realtime_num_chunks is not None
|
||||
for _ in range(max(1, case.perf_repeat_requests)):
|
||||
perf_record, content = self.run_and_collect(
|
||||
diffusion_server,
|
||||
case.id,
|
||||
generate_fn,
|
||||
collect_perf=not is_gt_gen_mode and not is_realtime_case,
|
||||
)
|
||||
perf_record, content = self.run_and_collect(
|
||||
diffusion_server,
|
||||
case.id,
|
||||
generate_fn,
|
||||
collect_perf=not is_gt_gen_mode and not is_realtime_case,
|
||||
)
|
||||
|
||||
if is_gt_gen_mode:
|
||||
# GT generation mode: save output and skip all validations/tests
|
||||
@@ -1614,12 +1637,13 @@ Pinned revision used by this check: {SGL_TEST_FILES_CI_DATA_REVISION}
|
||||
diffusion_server,
|
||||
case,
|
||||
chunk_stats,
|
||||
request_index,
|
||||
),
|
||||
)
|
||||
else:
|
||||
run_case_check(
|
||||
"performance",
|
||||
lambda: self._validate_and_record(case, perf_record),
|
||||
lambda: self._validate_and_record(case, perf_record, request_index),
|
||||
)
|
||||
|
||||
if case.server_args.custom_validator == "mesh":
|
||||
@@ -1646,11 +1670,19 @@ Pinned revision used by this check: {SGL_TEST_FILES_CI_DATA_REVISION}
|
||||
lambda: self._test_t2v_rejects_input_reference(diffusion_server, case),
|
||||
)
|
||||
|
||||
if case.run_consistency_check:
|
||||
if (
|
||||
case.run_consistency_check
|
||||
and os.environ.get("SGLANG_SKIP_CONSISTENCY", "0") != "1"
|
||||
):
|
||||
run_case_check(
|
||||
"consistency",
|
||||
lambda: self._validate_consistency(case, content),
|
||||
)
|
||||
if case.sampling_params.expect_audio_output:
|
||||
run_case_check(
|
||||
"audio consistency",
|
||||
lambda: self._validate_audio_consistency(case, content),
|
||||
)
|
||||
|
||||
if case.run_lora_basic_api_check:
|
||||
run_case_check(
|
||||
|
||||
@@ -312,11 +312,7 @@ class DiffusionTestCase:
|
||||
server_args: DiffusionServerArgs
|
||||
sampling_params: DiffusionSamplingParams | None = None
|
||||
run_perf_check: bool = True
|
||||
# Send the request this many times in one server session; performance and
|
||||
# consistency are validated on the last one. >1 asserts a warm second
|
||||
# request meets the same baselines -- a leak in residency arming, courier
|
||||
# in-flight tracking, or host copies shows up as the second request
|
||||
# degrading or dying.
|
||||
# Validate every repetition against the same baseline and GT.
|
||||
perf_repeat_requests: int = 1
|
||||
run_consistency_check: bool = True
|
||||
run_component_accuracy_check: bool = True
|
||||
@@ -328,6 +324,8 @@ class DiffusionTestCase:
|
||||
run_multi_lora_api_check: bool = False
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if self.perf_repeat_requests < 1:
|
||||
raise ValueError(f"{self.id}: perf_repeat_requests must be positive")
|
||||
if self.sampling_params is None:
|
||||
object.__setattr__(
|
||||
self,
|
||||
|
||||
@@ -0,0 +1,330 @@
|
||||
import json
|
||||
import os
|
||||
from dataclasses import replace
|
||||
from unittest.mock import Mock
|
||||
|
||||
import pytest
|
||||
|
||||
from sglang.multimodal_gen.runtime.utils.perf_logger import RequestPerfRecord
|
||||
from sglang.multimodal_gen.test.server import conftest, test_server_common
|
||||
from sglang.multimodal_gen.test.server.gpu_cases import TWO_GPU_CASES
|
||||
from sglang.multimodal_gen.test.server.testcase_configs import (
|
||||
BaselineConfig,
|
||||
DiffusionSamplingParams,
|
||||
DiffusionServerArgs,
|
||||
DiffusionTestCase,
|
||||
ScenarioConfig,
|
||||
ToleranceConfig,
|
||||
)
|
||||
|
||||
pytest_plugins = ["pytester"]
|
||||
|
||||
|
||||
def _perf_record():
|
||||
return RequestPerfRecord(
|
||||
request_id="request",
|
||||
commit_hash="test",
|
||||
tag="test",
|
||||
stages=[{"name": "DenoisingStage", "execution_time_ms": 10}],
|
||||
steps=[5, 5],
|
||||
total_duration_ms=100,
|
||||
memory_snapshots={
|
||||
"load_peak": {"peak_reserved_mb": 1000, "peak_allocated_mb": 800},
|
||||
"runtime_peak": {"peak_reserved_mb": 2000, "peak_allocated_mb": 1600},
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def harness(monkeypatch):
|
||||
monkeypatch.setenv("SGLANG_GEN_GT", "0")
|
||||
monkeypatch.setenv("SGLANG_GEN_BASELINE", "0")
|
||||
monkeypatch.setenv("SGLANG_SKIP_CONSISTENCY", "0")
|
||||
monkeypatch.setattr(test_server_common.current_platform, "is_cuda", lambda: True)
|
||||
scenario = ScenarioConfig(
|
||||
stages_ms={"DenoisingStage": 10},
|
||||
denoise_step_ms={0: 5, 1: 5},
|
||||
expected_e2e_ms=100,
|
||||
expected_avg_denoise_ms=5,
|
||||
expected_median_denoise_ms=5,
|
||||
estimated_full_test_time_s=1,
|
||||
load_peak_vram_mb=1000,
|
||||
runtime_peak_vram_mb=2000,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
test_server_common,
|
||||
"BASELINE_CONFIG",
|
||||
BaselineConfig(
|
||||
scenarios={"first": scenario},
|
||||
step_fractions=[0, 1],
|
||||
tolerances=ToleranceConfig(0, 0, 0, 0, 0),
|
||||
improvement_threshold=0,
|
||||
),
|
||||
)
|
||||
runner = test_server_common.DiffusionServerBase()
|
||||
runner._perf_results = []
|
||||
monkeypatch.setattr(test_server_common, "_PENDING_BASELINE_DUMPS", {})
|
||||
monkeypatch.setattr(test_server_common, "get_generate_fn", Mock())
|
||||
monkeypatch.setattr(runner, "_validate_consistency", Mock())
|
||||
monkeypatch.setattr(runner, "_validate_audio_consistency", Mock())
|
||||
monkeypatch.setattr(runner, "_dump_baseline_for_testcase", Mock())
|
||||
case = DiffusionTestCase(
|
||||
"first",
|
||||
DiffusionServerArgs(model_path="test", modality="image"),
|
||||
DiffusionSamplingParams(prompt="first"),
|
||||
perf_repeat_requests=2,
|
||||
run_models_api_check=False,
|
||||
run_t2v_input_reference_check=False,
|
||||
)
|
||||
return runner, case
|
||||
|
||||
|
||||
@pytest.mark.parametrize("bad_request", [0, 1])
|
||||
@pytest.mark.parametrize(
|
||||
"failure",
|
||||
[
|
||||
"performance",
|
||||
"load_peak",
|
||||
"runtime_peak",
|
||||
"missing_memory",
|
||||
"consistency",
|
||||
"generation",
|
||||
],
|
||||
)
|
||||
def test_each_request_failure_fails_case(harness, monkeypatch, bad_request, failure):
|
||||
runner, case = harness
|
||||
records = [_perf_record(), _perf_record()]
|
||||
outputs = [(record, b"output") for record in records]
|
||||
if failure == "performance":
|
||||
records[bad_request].total_duration_ms = 10000
|
||||
elif failure in {"load_peak", "runtime_peak"}:
|
||||
records[bad_request].memory_snapshots[failure]["peak_reserved_mb"] = 10000
|
||||
elif failure == "missing_memory":
|
||||
records[bad_request].memory_snapshots.clear()
|
||||
elif failure == "consistency":
|
||||
checks = [None, None]
|
||||
checks[bad_request] = AssertionError("wrong pixels or audio")
|
||||
runner._validate_consistency.side_effect = checks
|
||||
else:
|
||||
outputs[bad_request] = RuntimeError("server request failed")
|
||||
generate = Mock(side_effect=outputs)
|
||||
monkeypatch.setattr(runner, "run_and_collect", generate)
|
||||
ctx = object()
|
||||
|
||||
with pytest.raises(pytest.fail.Exception, match=f"request {bad_request + 1}/2"):
|
||||
runner.test_diffusion_generation(case, ctx)
|
||||
|
||||
assert generate.call_count == 2
|
||||
assert all(call.args[0] is ctx for call in generate.call_args_list)
|
||||
assert runner._validate_consistency.call_count == (
|
||||
1 if failure == "generation" else 2
|
||||
)
|
||||
# Even failed performance measurements must survive in the report.
|
||||
expected = [i + 1 for i in range(2) if failure != "generation" or i != bad_request]
|
||||
assert [r["request_index"] for r in runner._perf_results] == expected
|
||||
|
||||
|
||||
def test_both_requests_pass(harness, monkeypatch):
|
||||
runner, case = harness
|
||||
monkeypatch.setattr(
|
||||
runner,
|
||||
"run_and_collect",
|
||||
Mock(side_effect=[(_perf_record(), b"first"), (_perf_record(), b"second")]),
|
||||
)
|
||||
runner.test_diffusion_generation(case, object())
|
||||
assert runner._validate_consistency.call_count == 2
|
||||
assert [call.args[1] for call in runner._validate_consistency.call_args_list] == [
|
||||
b"first",
|
||||
b"second",
|
||||
]
|
||||
assert [r["request_index"] for r in runner._perf_results] == [1, 2]
|
||||
|
||||
|
||||
def test_request_artifacts_do_not_overwrite_each_other(harness, monkeypatch, tmp_path):
|
||||
runner, case = harness
|
||||
monkeypatch.setenv("SGLANG_DIFFUSION_ARTIFACT_DIR", str(tmp_path))
|
||||
artifact_dirs = []
|
||||
|
||||
def generate(*args, **kwargs):
|
||||
artifact_dirs.append(os.environ["SGLANG_DIFFUSION_ARTIFACT_DIR"])
|
||||
return _perf_record(), b"output"
|
||||
|
||||
monkeypatch.setattr(runner, "run_and_collect", generate)
|
||||
runner.test_diffusion_generation(case, object())
|
||||
assert artifact_dirs == [str(tmp_path / f"request-{i}") for i in (1, 2)]
|
||||
assert os.environ["SGLANG_DIFFUSION_ARTIFACT_DIR"] == str(tmp_path)
|
||||
|
||||
|
||||
def test_later_skip_cannot_hide_earlier_failure(harness, monkeypatch):
|
||||
runner, case = harness
|
||||
monkeypatch.setattr(
|
||||
runner,
|
||||
"run_and_collect",
|
||||
Mock(
|
||||
side_effect=[
|
||||
RuntimeError("failed first"),
|
||||
pytest.skip.Exception("skip second"),
|
||||
]
|
||||
),
|
||||
)
|
||||
with pytest.raises(pytest.fail.Exception, match="failed first"):
|
||||
runner.test_diffusion_generation(case, object())
|
||||
|
||||
|
||||
def test_second_request_cannot_be_skipped_after_first_passes(harness, monkeypatch):
|
||||
runner, case = harness
|
||||
monkeypatch.setattr(
|
||||
runner,
|
||||
"run_and_collect",
|
||||
Mock(side_effect=[(_perf_record(), b"output"), pytest.skip.Exception("skip")]),
|
||||
)
|
||||
with pytest.raises(pytest.fail.Exception, match="Required request skipped"):
|
||||
runner.test_diffusion_generation(case, object())
|
||||
|
||||
|
||||
def test_empty_content_is_not_a_consistency_pass(harness):
|
||||
runner, case = harness
|
||||
with pytest.raises(pytest.fail.Exception, match="Empty output"):
|
||||
test_server_common.DiffusionServerBase._validate_consistency(runner, case, b"")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("bad_request", [0, 1])
|
||||
def test_audio_checked_even_when_video_consistency_fails(
|
||||
harness, monkeypatch, bad_request
|
||||
):
|
||||
runner, case = harness
|
||||
case = replace(
|
||||
case,
|
||||
server_args=replace(case.server_args, modality="video"),
|
||||
sampling_params=replace(case.sampling_params, expect_audio_output=True),
|
||||
)
|
||||
runner._validate_consistency.side_effect = AssertionError("wrong pixels")
|
||||
audio_checks = [None, None]
|
||||
audio_checks[bad_request] = AssertionError("wrong audio")
|
||||
runner._validate_audio_consistency.side_effect = audio_checks
|
||||
monkeypatch.setattr(
|
||||
runner, "run_and_collect", Mock(return_value=(_perf_record(), b"output"))
|
||||
)
|
||||
with pytest.raises(pytest.fail.Exception, match="audio consistency.*wrong audio"):
|
||||
runner.test_diffusion_generation(case, object())
|
||||
assert runner._validate_consistency.call_count == 2
|
||||
assert runner._validate_audio_consistency.call_count == 2
|
||||
|
||||
|
||||
@pytest.mark.parametrize("repeat", [0, -1])
|
||||
def test_invalid_repeat_count(harness, repeat):
|
||||
_, case = harness
|
||||
with pytest.raises(ValueError, match="must be positive"):
|
||||
replace(case, perf_repeat_requests=repeat)
|
||||
|
||||
|
||||
def test_report_keeps_both_requests_and_replaces_retry(tmp_path):
|
||||
path = tmp_path / "results.json"
|
||||
records = [
|
||||
{"class_name": "suite", "test_name": "case", "request_index": i, "e2e_ms": i}
|
||||
for i in (1, 2)
|
||||
]
|
||||
conftest._write_results_json(records, str(path))
|
||||
conftest._write_results_json([{**records[0], "e2e_ms": 3}], str(path))
|
||||
assert json.loads(path.read_text()) == [{**records[0], "e2e_ms": 3}, records[1]]
|
||||
|
||||
|
||||
def test_perf_fixture_retains_failed_case_results(pytester, monkeypatch):
|
||||
monkeypatch.delenv("GITHUB_STEP_SUMMARY", raising=False)
|
||||
pytester.makeconftest(
|
||||
'pytest_plugins = ["sglang.multimodal_gen.test.server.conftest"]'
|
||||
)
|
||||
pytester.makepyfile(
|
||||
"""
|
||||
import pytest
|
||||
|
||||
from sglang.multimodal_gen.test.server.test_server_common import DiffusionServerBase
|
||||
from sglang.multimodal_gen.test.server.testcase_configs import (
|
||||
DiffusionSamplingParams, DiffusionServerArgs, DiffusionTestCase, PerformanceSummary,
|
||||
)
|
||||
|
||||
class TestRequests(DiffusionServerBase):
|
||||
@pytest.mark.parametrize("case_id", ["failed", "passed"])
|
||||
def test_diffusion_generation(self, case_id):
|
||||
assert self._perf_results == []
|
||||
case = DiffusionTestCase(
|
||||
case_id,
|
||||
DiffusionServerArgs("test", modality="image"),
|
||||
DiffusionSamplingParams(prompt="test"),
|
||||
)
|
||||
summary = PerformanceSummary(100, 5, 5, {}, [], {}, {})
|
||||
for index in (1, 2):
|
||||
self._record_performance_result(case, summary, index)
|
||||
if case_id == "failed":
|
||||
pytest.fail("recorded failure")
|
||||
|
||||
class TestOtherRequests(TestRequests):
|
||||
pass
|
||||
"""
|
||||
)
|
||||
result = pytester.runpytest_subprocess("-q")
|
||||
result.assert_outcomes(passed=2, failed=2)
|
||||
records = json.loads((pytester.path / "diffusion-results.json").read_text())
|
||||
assert len(records) == 8
|
||||
assert {(r["class_name"], r["test_name"], r["request_index"]) for r in records} == {
|
||||
(suite, case_id, index)
|
||||
for suite in ("TestRequests", "TestOtherRequests")
|
||||
for case_id in ("failed", "passed")
|
||||
for index in (1, 2)
|
||||
}
|
||||
|
||||
|
||||
def test_gt_generation_runs_both_requests(harness, monkeypatch):
|
||||
runner, case = harness
|
||||
monkeypatch.setenv("SGLANG_GEN_GT", "1")
|
||||
generate = Mock(side_effect=[(None, b"first"), (None, b"second")])
|
||||
monkeypatch.setattr(runner, "run_and_collect", generate)
|
||||
save = Mock()
|
||||
monkeypatch.setattr(runner, "_save_gt_output", save)
|
||||
runner.test_diffusion_generation(case, object())
|
||||
assert [(call.args[0].id, call.args[1]) for call in save.call_args_list] == [
|
||||
("first", b"first"),
|
||||
("first", b"second"),
|
||||
]
|
||||
assert all(not call.kwargs["collect_perf"] for call in generate.call_args_list)
|
||||
runner._validate_consistency.assert_not_called()
|
||||
|
||||
|
||||
def test_baseline_generation_keeps_worst_of_both_requests(harness, monkeypatch):
|
||||
runner, case = harness
|
||||
monkeypatch.setenv("SGLANG_GEN_BASELINE", "1")
|
||||
records = [_perf_record(), _perf_record()]
|
||||
records[0].total_duration_ms = 200
|
||||
records[0].memory_snapshots["load_peak"]["peak_allocated_mb"] = 900
|
||||
records[0].memory_snapshots["warmup_peak"] = {"peak_reserved_mb": 4000}
|
||||
records[1].memory_snapshots["warmup_peak"] = {"peak_reserved_mb": 2000}
|
||||
records[1].memory_snapshots["runtime_peak"]["peak_reserved_mb"] = 3000
|
||||
records[1].memory_snapshots["runtime_peak"]["peak_allocated_mb"] = 2500
|
||||
monkeypatch.setattr(
|
||||
runner, "run_and_collect", Mock(side_effect=[(r, b"output") for r in records])
|
||||
)
|
||||
runner.test_diffusion_generation(case, object())
|
||||
summaries = test_server_common._PENDING_BASELINE_DUMPS[case.id]
|
||||
assert len(summaries) == 2
|
||||
log = Mock()
|
||||
monkeypatch.setattr(test_server_common.logger, "error", log)
|
||||
test_server_common.DiffusionServerBase._dump_baseline_for_testcase(
|
||||
runner, case, summaries[-1], repeated_summaries=summaries
|
||||
)
|
||||
baseline = json.loads(log.call_args.args[0].split(f'"{case.id}": ', 1)[1])
|
||||
assert baseline["expected_e2e_ms"] == 200
|
||||
assert baseline["runtime_peak_vram_mb"] == 3000
|
||||
assert baseline["warmup_peak_vram_mb"] == 4000
|
||||
assert baseline["load_peak_allocated_mb"] == 900
|
||||
assert baseline["runtime_peak_allocated_mb"] == 2500
|
||||
|
||||
|
||||
def test_h3_cases_check_two_short_requests_and_audio():
|
||||
cases = [case for case in TWO_GPU_CASES if case.id.startswith("minimax_h3_")]
|
||||
assert {case.sampling_params.extras["task"] for case in cases} == {"t2va", "ref2va"}
|
||||
for case in cases:
|
||||
assert case.perf_repeat_requests == 2
|
||||
assert case.run_perf_check and case.run_consistency_check
|
||||
assert case.sampling_params.expect_audio_output
|
||||
assert case.sampling_params.extras["num_inference_steps"] <= 8
|
||||
assert case.sampling_params.extras["target"]["duration_seconds"] == 4.0
|
||||
Reference in New Issue
Block a user