diff --git a/docs/docs/sglang-diffusion/ci_perf.mdx b/docs/docs/sglang-diffusion/ci_perf.mdx index 2f87e6358..c1f333aa1 100644 --- a/docs/docs/sglang-diffusion/ci_perf.mdx +++ b/docs/docs/sglang-diffusion/ci_perf.mdx @@ -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 ` 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`. diff --git a/python/sglang/multimodal_gen/test/server/conftest.py b/python/sglang/multimodal_gen/test/server/conftest.py index 205ac4aa7..47d6f1819 100644 --- a/python/sglang/multimodal_gen/test/server/conftest.py +++ b/python/sglang/multimodal_gen/test/server/conftest.py @@ -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() diff --git a/python/sglang/multimodal_gen/test/server/gpu_cases.py b/python/sglang/multimodal_gen/test/server/gpu_cases.py index c56343e4d..e32b4e93e 100644 --- a/python/sglang/multimodal_gen/test/server/gpu_cases.py +++ b/python/sglang/multimodal_gen/test/server/gpu_cases.py @@ -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, diff --git a/python/sglang/multimodal_gen/test/server/perf_baselines/h100.json b/python/sglang/multimodal_gen/test/server/perf_baselines/h100.json index 1f0d0b489..528e11b99 100644 --- a/python/sglang/multimodal_gen/test/server/perf_baselines/h100.json +++ b/python/sglang/multimodal_gen/test/server/perf_baselines/h100.json @@ -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": {}, diff --git a/python/sglang/multimodal_gen/test/server/test_server_common.py b/python/sglang/multimodal_gen/test/server/test_server_common.py index 58011b481..729400741 100644 --- a/python/sglang/multimodal_gen/test/server/test_server_common.py +++ b/python/sglang/multimodal_gen/test/server/test_server_common.py @@ -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( diff --git a/python/sglang/multimodal_gen/test/server/testcase_configs.py b/python/sglang/multimodal_gen/test/server/testcase_configs.py index 5c8cd4f3e..19067291e 100644 --- a/python/sglang/multimodal_gen/test/server/testcase_configs.py +++ b/python/sglang/multimodal_gen/test/server/testcase_configs.py @@ -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, diff --git a/python/sglang/multimodal_gen/test/unit/test_sequential_server_requests.py b/python/sglang/multimodal_gen/test/unit/test_sequential_server_requests.py new file mode 100644 index 000000000..aa74cabaa --- /dev/null +++ b/python/sglang/multimodal_gen/test/unit/test_sequential_server_requests.py @@ -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