From 501b7851e49ca30f2fee8aed6f7beaabdd1a768f Mon Sep 17 00:00:00 2001 From: Mick Date: Mon, 21 Sep 2026 08:43:22 +0800 Subject: [PATCH] [diffusion] CI: guard E2E/loading latency with runner-aware baselines (#39206) Co-authored-by: Mick Qian --- .../check-pr-test-health/action.test.cjs | 73 +++++ .../actions/check-pr-test-health/action.yml | 10 +- .github/workflows/lint.yml | 3 + .github/workflows/pr-test-multimodal-gen.yml | 7 +- .../sglang/kernels/ops/diffusion/__init__.py | 1 + .../diffusion/ext/mesh_processor/__init__.py | 6 +- .../runtime/disaggregation/orchestrator.py | 1 + .../multimodal_gen/runtime/launch_server.py | 1 + .../runtime/pipelines_core/stages/base.py | 2 + .../model_specific_stages/hunyuan3d/paint.py | 4 + .../stages/timestep_preparation.py | 11 +- .../runtime/utils/perf_logger.py | 8 +- .../runtime/warmup_request_builder.py | 7 +- .../multimodal_gen/test/runner/PERFORMANCE.md | 196 +++++++++++++ .../test/runner/pytest_runner.py | 51 +++- .../test/scripts/gen_perf_baselines.py | 5 + .../multimodal_gen/test/server/gpu_cases.py | 13 +- .../test/server/perf_baselines/5090.json | 6 + .../test/server/perf_baselines/b200.json | 39 +++ .../test/server/perf_baselines/h100.json | 227 +++++++++++++-- .../test/server/realtime_consistency.py | 17 +- .../test/server/test_server_common.py | 199 +++++++++---- .../test/server/test_server_utils.py | 56 +++- .../test/server/testcase_configs.py | 25 ++ .../sglang/multimodal_gen/test/test_utils.py | 2 +- .../test_realtime_consistency_harness.py | 140 +++++++++ .../test/unit/test_cfg_parallel_warmup.py | 99 +++++++ .../test_hunyuan3d_native_texture_models.py | 19 +- .../test/unit/test_load_inclusive_e2e.py | 194 +++++++++++++ .../unit/test_performance_failure_policy.py | 268 ++++++++++++++++++ .../test/unit/test_performance_metrics.py | 83 ++++++ .../test/unit/test_runner_perf_baselines.py | 78 +++++ .../unit/test_sequential_server_requests.py | 155 +++++++++- .../test/unit/test_suite_partitioning.py | 5 + .../unit/test_timestep_preparation_logging.py | 65 +++++ .../unit/test_video_url_request_extras.py | 36 +++ 36 files changed, 1974 insertions(+), 138 deletions(-) create mode 100644 .github/actions/check-pr-test-health/action.test.cjs create mode 100644 python/sglang/multimodal_gen/test/runner/PERFORMANCE.md create mode 100644 python/sglang/multimodal_gen/test/unit/test_load_inclusive_e2e.py create mode 100644 python/sglang/multimodal_gen/test/unit/test_performance_failure_policy.py create mode 100644 python/sglang/multimodal_gen/test/unit/test_runner_perf_baselines.py create mode 100644 python/sglang/multimodal_gen/test/unit/test_timestep_preparation_logging.py create mode 100644 python/sglang/multimodal_gen/test/unit/test_video_url_request_extras.py diff --git a/.github/actions/check-pr-test-health/action.test.cjs b/.github/actions/check-pr-test-health/action.test.cjs new file mode 100644 index 000000000..715f122d2 --- /dev/null +++ b/.github/actions/check-pr-test-health/action.test.cjs @@ -0,0 +1,73 @@ +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const path = require('node:path'); +const { test } = require('node:test'); + +const yaml = fs.readFileSync(path.join(__dirname, 'action.yml'), 'utf8'); +const script = yaml.split(' script: |\n')[1] + .split('\n').map(line => line.replace(/^ /, '')).join('\n'); +const run = new (Object.getPrototypeOf(async function () {}).constructor)( + 'github', 'context', 'core', 'process', script, +); + +async function check({ snapshot = [], live = [], lint = 'success', event = 'pull_request' } = {}) { + const failures = []; + let labelReads = 0; + let jobReads = 0; + const labels = names => names.map(name => ({ name })); + const github = { + rest: { + checks: { listForRef: async () => ({ data: { check_runs: [ + { app: { slug: 'github-actions' }, status: 'completed', conclusion: lint }, + ] } }) }, + pulls: { get: async args => { + assert.equal(args.pull_number, 42); + labelReads++; + return { data: { labels: labels(live) } }; + } }, + repos: { listPullRequestsAssociatedWithCommit: async () => { + labelReads++; + return { data: [{ labels: labels(live) }] }; + } }, + actions: { listJobsForWorkflowRun: () => {} }, + }, + paginate: async () => { + jobReads++; + return [{ name: 'model-test', status: 'completed', conclusion: 'failure', steps: [] }]; + }, + }; + await run(github, { + eventName: event, repo: { owner: 'owner', repo: 'repo' }, sha: 'head', runId: 1, + payload: event === 'pull_request' + ? { pull_request: { number: 42, head: { sha: 'head' }, labels: labels(snapshot) } } + : {}, + }, { info: () => {}, setFailed: message => failures.push(message) }, { env: {} }); + return { failures, labelReads, jobReads }; +} + +test('a label added after the event bypasses sibling failures on rerun', async () => { + assert.deepEqual(await check({ live: ['bypass-fastfail'] }), + { failures: [], labelReads: 1, jobReads: 0 }); +}); + +test('a removed label does not continue bypassing sibling failures', async () => { + const result = await check({ snapshot: ['bypass-fastfail'] }); + assert.equal(result.labelReads, 1); + assert.equal(result.jobReads, 1); + assert.match(result.failures[0], /root cause job\(s\): model-test/); +}); + +test('bypass never skips a failed lint check', async () => { + assert.deepEqual(await check({ live: ['bypass-fastfail'], lint: 'failure' }), + { failures: ['Fast-fail: lint check failed'], labelReads: 0, jobReads: 0 }); +}); + +test('non-PR events retain associated-PR label lookup', async () => { + assert.deepEqual(await check({ event: 'workflow_dispatch', live: ['bypass-fastfail'] }), + { failures: [], labelReads: 1, jobReads: 0 }); +}); + +test('scheduled runs remain exempt', async () => { + assert.deepEqual(await check({ event: 'schedule' }), + { failures: [], labelReads: 0, jobReads: 0 }); +}); diff --git a/.github/actions/check-pr-test-health/action.yml b/.github/actions/check-pr-test-health/action.yml index 8cafb66e9..84fa7547f 100644 --- a/.github/actions/check-pr-test-health/action.yml +++ b/.github/actions/check-pr-test-health/action.yml @@ -50,8 +50,14 @@ runs: // Skip the jobs-failed check when the PR carries the bypass-fastfail label. // Lint check above still runs. let labels = []; - if (context.payload.pull_request?.labels) { - labels = context.payload.pull_request.labels.map(l => l.name); + if (context.payload.pull_request?.number) { + // reruns retain the original event payload, including stale labels + const { data: pr } = await github.rest.pulls.get({ + owner: context.repo.owner, + repo: context.repo.repo, + pull_number: context.payload.pull_request.number, + }); + labels = pr.labels.map(l => l.name); } else { const { data: prs } = await github.rest.repos.listPullRequestsAssociatedWithCommit({ owner: context.repo.owner, diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index 4a89b953f..2041ea817 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -76,6 +76,9 @@ jobs: - name: Check cookbook authoring contracts run: node docs/scripts/check_cookbook_configs.mjs + - name: Test PR health action + run: node --test .github/actions/check-pr-test-health/action.test.cjs + - name: Cache mint uses: actions/cache@v4 with: diff --git a/.github/workflows/pr-test-multimodal-gen.yml b/.github/workflows/pr-test-multimodal-gen.yml index 1baa2c41d..f15e8a45b 100644 --- a/.github/workflows/pr-test-multimodal-gen.yml +++ b/.github/workflows/pr-test-multimodal-gen.yml @@ -130,7 +130,7 @@ jobs: path: | python/sglang/multimodal_gen/test/execution_report_*.json python/diffusion-results.json - retention-days: 1 + retention-days: 7 - name: Upload diffusion failure artifacts if: always() @@ -280,6 +280,9 @@ jobs: fail-fast: false matrix: ${{ fromJson(needs.compute-diffusion-partitions.outputs.matrix-2gpu) }} steps: + - name: Record retry deadline + run: echo "SGLANG_DIFFUSION_RETRY_DEADLINE=$(( $(date +%s) + 2400 ))" >> "$GITHUB_ENV" + - name: Checkout code uses: actions/checkout@v4 with: @@ -327,7 +330,7 @@ jobs: path: | python/sglang/multimodal_gen/test/execution_report_*.json python/diffusion-results.json - retention-days: 1 + retention-days: 7 - name: Upload diffusion failure artifacts if: always() diff --git a/python/sglang/kernels/ops/diffusion/__init__.py b/python/sglang/kernels/ops/diffusion/__init__.py index 0f77f525c..452950901 100644 --- a/python/sglang/kernels/ops/diffusion/__init__.py +++ b/python/sglang/kernels/ops/diffusion/__init__.py @@ -736,6 +736,7 @@ _EXPORTS: dict[str, str] = { "interpolate": "ext.hunyuan3d_rasterizer", "rasterize": "ext.hunyuan3d_rasterizer", "meshVerticeInpaint": "ext.mesh_processor", + "load_mesh_processor": "ext.mesh_processor", } diff --git a/python/sglang/kernels/ops/diffusion/ext/mesh_processor/__init__.py b/python/sglang/kernels/ops/diffusion/ext/mesh_processor/__init__.py index fa399908a..030f6aef1 100644 --- a/python/sglang/kernels/ops/diffusion/ext/mesh_processor/__init__.py +++ b/python/sglang/kernels/ops/diffusion/ext/mesh_processor/__init__.py @@ -19,7 +19,7 @@ _abs_path = os.path.dirname(os.path.abspath(__file__)) _mesh_processor_kernel = None -def _load_mesh_processor(): +def load_mesh_processor(): """JIT compile and load the mesh processor kernel.""" global _mesh_processor_kernel @@ -47,7 +47,7 @@ def meshVerticeInpaint( method: str = "smooth", ) -> Tuple[np.ndarray, np.ndarray]: """Inpaint texture using mesh vertex connectivity.""" - kernel = _load_mesh_processor() + kernel = load_mesh_processor() texture = np.ascontiguousarray(texture, dtype=np.float32) mask = np.ascontiguousarray(mask, dtype=np.uint8) @@ -61,4 +61,4 @@ def meshVerticeInpaint( ) -__all__ = ["meshVerticeInpaint"] +__all__ = ["load_mesh_processor", "meshVerticeInpaint"] diff --git a/python/sglang/multimodal_gen/runtime/disaggregation/orchestrator.py b/python/sglang/multimodal_gen/runtime/disaggregation/orchestrator.py index 8f3faf68d..fd3f85157 100644 --- a/python/sglang/multimodal_gen/runtime/disaggregation/orchestrator.py +++ b/python/sglang/multimodal_gen/runtime/disaggregation/orchestrator.py @@ -63,6 +63,7 @@ def _deserialize_request_metrics(data: dict | None) -> RequestMetrics | None: metrics = RequestMetrics(request_id=data["request_id"]) metrics.stages = data.get("stages", {}) + metrics.denoising_stages = set(data.get("denoising_stages", ())) metrics.steps = data.get("steps", []) metrics.total_duration_ms = data.get("total_duration_ms", 0.0) for name, snapshot in data.get("memory_snapshots", {}).items(): diff --git a/python/sglang/multimodal_gen/runtime/launch_server.py b/python/sglang/multimodal_gen/runtime/launch_server.py index 5d06929b9..d3fa0a1de 100644 --- a/python/sglang/multimodal_gen/runtime/launch_server.py +++ b/python/sglang/multimodal_gen/runtime/launch_server.py @@ -211,6 +211,7 @@ def launch_server(server_args: ServerArgs, launch_http_server: bool = True): reader.close() logger.debug("All workers are ready") + logger.info("[server-load] workers_ready_monotonic_ns=%d", time.monotonic_ns()) if node_rank != 0: # The TokenizerManager / HTTP surface lives on the node that owns diff --git a/python/sglang/multimodal_gen/runtime/pipelines_core/stages/base.py b/python/sglang/multimodal_gen/runtime/pipelines_core/stages/base.py index 1defacad7..f2b3a5a96 100644 --- a/python/sglang/multimodal_gen/runtime/pipelines_core/stages/base.py +++ b/python/sglang/multimodal_gen/runtime/pipelines_core/stages/base.py @@ -417,6 +417,8 @@ class PipelineStage(StageDedupMixin, ABC): # Execute the actual stage logic with unified profiling. previous_batch_is_warmup = self._current_batch_is_warmup metrics = batch.metrics + if metrics is not None and self.role_affinity == RoleType.DENOISER: + metrics.denoising_stages.add(stage_name) warmup_metrics = metrics if batch.is_warmup else None previous_active_stage = ( warmup_metrics.active_stage_name if warmup_metrics is not None else None diff --git a/python/sglang/multimodal_gen/runtime/pipelines_core/stages/model_specific_stages/hunyuan3d/paint.py b/python/sglang/multimodal_gen/runtime/pipelines_core/stages/model_specific_stages/hunyuan3d/paint.py index ca680fc9a..9cd08e3c8 100644 --- a/python/sglang/multimodal_gen/runtime/pipelines_core/stages/model_specific_stages/hunyuan3d/paint.py +++ b/python/sglang/multimodal_gen/runtime/pipelines_core/stages/model_specific_stages/hunyuan3d/paint.py @@ -19,6 +19,7 @@ from PIL import Image from torch import nn from transformers import PreTrainedTokenizerBase +from sglang.kernels.ops.diffusion import load_mesh_processor from sglang.multimodal_gen.configs.models.encoders import BaseEncoderOutput from sglang.multimodal_gen.configs.pipeline_configs.hunyuan3d import ( Hunyuan3D2PipelineConfig, @@ -790,6 +791,9 @@ class Hunyuan3DPaintPostprocessStage(PipelineStage): def forward(self, batch: Req, server_args: ServerArgs) -> OutputBatch: del server_args + if batch.is_warmup: + # compile without exporting warmup meshes or textures + load_mesh_processor() if batch.is_warmup or batch.extra.get("_mesh_failed"): return OutputBatch(output_file_paths=[], metrics=batch.metrics) diff --git a/python/sglang/multimodal_gen/runtime/pipelines_core/stages/timestep_preparation.py b/python/sglang/multimodal_gen/runtime/pipelines_core/stages/timestep_preparation.py index 21430f51c..1b86a65bc 100644 --- a/python/sglang/multimodal_gen/runtime/pipelines_core/stages/timestep_preparation.py +++ b/python/sglang/multimodal_gen/runtime/pipelines_core/stages/timestep_preparation.py @@ -8,6 +8,7 @@ This module contains implementations of timestep preparation stages for diffusio """ import inspect +import logging from dataclasses import dataclass from typing import Any, Callable, Tuple @@ -161,8 +162,14 @@ class TimestepPreparationStage(PipelineStage): # Update batch with prepared timesteps batch.timesteps = timesteps batch.scheduler = scheduler - if not batch.is_warmup: - self.log_debug("timesteps: %s", timesteps) + if not batch.is_warmup and logger.isEnabledFor(logging.DEBUG): + # format on cpu to avoid first-use cuda kernels in tensor repr + logger.debug( + "[%s] timesteps (%s): %s", + self.__class__.__name__, + timesteps.device, + timesteps.detach().cpu(), + ) return batch def build_dedup_fingerprint( diff --git a/python/sglang/multimodal_gen/runtime/utils/perf_logger.py b/python/sglang/multimodal_gen/runtime/utils/perf_logger.py index b14dda7b1..3739bda96 100644 --- a/python/sglang/multimodal_gen/runtime/utils/perf_logger.py +++ b/python/sglang/multimodal_gen/runtime/utils/perf_logger.py @@ -58,6 +58,7 @@ class RequestMetrics: def __init__(self, request_id: str): self.request_id = request_id self.stages: Dict[str, float] = {} + self.denoising_stages: set[str] = set() self.steps: list[float] = [] self.steps_by_stage: Dict[str, list[float]] = {} self.stage_iterations: Dict[str, tuple[int, int]] = {} @@ -112,6 +113,7 @@ class RequestMetrics: return { "request_id": self.request_id, "stages": self.stages, + "denoising_stages": sorted(self.denoising_stages), "steps": self.steps, "total_duration_ms": self.total_duration_ms, "memory_snapshots": { @@ -461,7 +463,11 @@ class PerformanceLogger: Note that this accords to the time spent internally in server, postprocess is not included """ formatted_stages = [ - {"name": name, "execution_time_ms": duration_ms} + { + "name": name, + "execution_time_ms": duration_ms, + "is_denoising": name in metrics.denoising_stages, + } for name, duration_ms in metrics.stages.items() ] diff --git a/python/sglang/multimodal_gen/runtime/warmup_request_builder.py b/python/sglang/multimodal_gen/runtime/warmup_request_builder.py index b6a2a2263..c060a9493 100644 --- a/python/sglang/multimodal_gen/runtime/warmup_request_builder.py +++ b/python/sglang/multimodal_gen/runtime/warmup_request_builder.py @@ -365,12 +365,11 @@ def _resolve_warmup_num_frames( if num_frames is None: return num_frames - # Breakable CUDA graph replays only exact latent shapes: the warmup - # request must run the full serving frame count so its captured graphs - # match serving signatures (mirrors the uncapped-steps rule in - # _resolve_warmup_steps). + # explicit frame counts and breakable CUDA graphs must keep the requested + # latent shape; only default server warmup applies the bounded frame cap if ( not server_based_warmup + or isinstance(explicit_num_frames, int) or getattr(server_args, "enable_breakable_cuda_graph", False) is True ): warmup_num_frames = num_frames diff --git a/python/sglang/multimodal_gen/test/runner/PERFORMANCE.md b/python/sglang/multimodal_gen/test/runner/PERFORMANCE.md new file mode 100644 index 000000000..8f1ea6157 --- /dev/null +++ b/python/sglang/multimodal_gen/test/runner/PERFORMANCE.md @@ -0,0 +1,196 @@ +# CI performance guards and baselines + +## Metric and failure contracts + +Every generated testcase must report finite, positive E2E, including cases with +`run_perf_check=False`. Missing request records, absent performance logs and +missing/invalid E2E fail CI. `run_perf_check=False` disables stage/step and +memory checks, not the request's E2E threshold guard. Baseline generation still +requires valid E2E but skips baseline comparisons. Explicit GT generation skips +validation. + +A performance failure stops the testcase's remaining repeated requests and +subsequent checks for that attempt. Performance failures use the existing +pytest retry budget (at most six retries), rerunning only failed items. +Exhausting the budget still fails CI; missing metrics and exceeded thresholds +are never treated as passing. Consistency failures remain non-retryable. +Standalone infrastructure failures retain their existing retry policy. +Valid failed measurements are recorded +before threshold validation; realtime chunk and memory guards remain enabled +according to their existing configuration. + +## B200 runner baselines + +`b200.json` keeps the existing Verda references as its defaults. Its +`runner_overrides` map applies metric overrides by the GitHub `RUNNER_NAME` +prefix. DeepInfra runners (`b200-di*`) use separate E2E references for the two +cases below; unknown runners keep the defaults. Loading, stage/step and memory +references, other cases, and the 25% E2E tolerance are unchanged. + +| Case | Default E2E (ms) | DeepInfra E2E (ms) | +| --- | ---: | ---: | +| `flux1_modelopt_nvfp4_t2i` | 836.71 | 1334.16 | +| `qwen_image_2512_modelopt_nvfp4_t2i` | 9650.06 | 16126.87 | + +The pool mismatch was observed in [B200 CI job 103856482654](https://github.com/sgl-project/sglang/actions/runs/34805428031/job/103856482654). +The DeepInfra references are the medians of three unprofiled, warmed requests +using the CI case configuration at commit +`9b7e11f32b88d4c1bfd9cf44550a30a702dd9df8`: +Flux: 1367.47, 1327.48, 1334.16 ms; Qwen: 16126.87, 16806.36, 15240.80 ms. +The matching Verda measurements were 811.26, 771.97, 783.31 ms and +9169.16, 9045.49, 9071.78 ms, respectively. These calibrate the runner pools; +they do not establish a model-level root cause for the difference. + +When refreshing a pool-specific reference, update its `runner_overrides` entry, +not the shared `scenarios` entry. The baseline generation script writes shared +scenarios; use a separate `--out` file when collecting pool-specific candidates. + +### Cirrascale historical CI reference + +`b200-cirrascale1-0123` has separate E2E references of 1574.32 ms for +`flux1_modelopt_nvfp4_t2i` and 17742.04 ms for +`qwen_image_2512_modelopt_nvfp4_t2i`. The measured Cirrascale 3 runners below also +have separate references. Other Cirrascale runners retain the defaults until +calibrated. + +Historical jobs on this runner, all using driver 580.126.20, already recorded +the slower timings before this PR's changes, with unchanged B200 case definitions: + +| PR / CI job | Flux1 E2E (ms) | Qwen2512 E2E (ms) | +| --- | ---: | ---: | +| [#39021](https://github.com/sgl-project/sglang/actions/runs/34594936520/job/103248571918) | 1772.69 | 17836.93 | +| [#38782](https://github.com/sgl-project/sglang/actions/runs/34595934993/job/103251774792) | 2617.81 | 17742.04 | +| [#39022](https://github.com/sgl-project/sglang/actions/runs/34670636885/job/103506632867) | 1574.32 | 38392.34 | +| [#39291](https://github.com/sgl-project/sglang/actions/runs/34750152864/job/103707842024) | 3942.31 | 19846.93 | + +Use the minimum observed E2E for each case, not the noisy maximum or a median +inflated by slow runs. The existing 25% tolerance yields limits of 1967.90 ms +and 22177.55 ms. Large transient slowdowns must still fail and use the bounded +failed-item retry policy. Historical green jobs did not enforce the new E2E +guard; their recorded timings, not their green status, support these references. +This does not identify the underlying host/GPU contention mechanism. Loading, +other metrics, other cases, and other runner references remain unchanged. + +### Cirrascale 3 historical CI references + +Match `b200-cirrascale3-0123` and `b200-cirrascale3-4567` separately, without +extending the override to unmeasured runners. Their two NVFP4 case definitions, +sampling configuration and model implementations are unchanged in the historical +comparisons below. These are warmed request timings, not model download or load +times. The independent PRs did not include this PR's E2E guard changes. + +| PR / CI job | Runner suffix | Flux1 E2E (ms) | Qwen2512 E2E (ms) | +| --- | --- | ---: | ---: | +| [#40265](https://github.com/sgl-project/sglang/actions/runs/35437052621/job/105881489389) | `3-0123` | 1470.24 | 35377.30 | +| [#39206, earlier head](https://github.com/sgl-project/sglang/actions/runs/35433818034/job/105873136034) | `3-0123` | 1471.59 | 17894.49 | +| [#40293](https://github.com/sgl-project/sglang/actions/runs/35433914839/job/105879952157) | `3-4567` | 1547.77 | 17346.65 | +| [#39983](https://github.com/sgl-project/sglang/actions/runs/35448201646/job/105999591501) | `3-4567` | 1471.19 | 18294.78 | +| [#40374](https://github.com/sgl-project/sglang/actions/runs/35465121078/job/105956031846) | `3-4567` | 1500.43 | 18433.93 | + +The earlier #39206 row uses the minimum of its seven attempts, not their noisy +maximum. Apply the same minimum-observed rule per runner across these records: +1470.24 / 17894.49 ms for `3-0123`, and 1471.19 / 17346.65 ms for `3-4567`. +Keep the 25% tolerance and all other metrics unchanged. In particular, the +35-second historical Qwen outlier still fails; it is not a new reference. +These records establish a pre-existing runner-specific mismatch with the Verda +reference, not the underlying cause of contention or a claim that all runs pass. + +## Initial loading references + +The initial H100 loading references cover 28 cases with at least three distinct +CI runs whose maximum/minimum startup-time ratio is at most 1.25. Each reference +is the minimum measured `load_time_ms`, rounded to two decimals; repeated requests +sharing one server do not count as separate startups. These are process-start to +all-workers-ready measurements, excluding warmup, not checkpoint-I/O-only times. + +Sources are PR Test Base runs [34750203901](https://github.com/sgl-project/sglang/actions/runs/34750203901), +[34751664379](https://github.com/sgl-project/sglang/actions/runs/34751664379), +[34753876622](https://github.com/sgl-project/sglang/actions/runs/34753876622), and +[34755864886](https://github.com/sgl-project/sglang/actions/runs/34755864886). +Only saved valid loading measurements are used; this does not claim those runs +passed all other checks. No E2E reference or tolerance is raised. Cross-run +stability is a conservative selection criterion, not proof of an optimal load +time. The first batches left variable cases uncalibrated; the best-observed +references below now give those cases a loading guard without claiming stable +runtime. B200 and 5090 references are not inferred from H100 or development H200 +measurements. + +The six RTX 5090 loading references use the same selection rule, from runs +[34753876622](https://github.com/sgl-project/sglang/actions/runs/34753876622), +[34755864886](https://github.com/sgl-project/sglang/actions/runs/34755864886), and +[34764411082](https://github.com/sgl-project/sglang/actions/runs/34764411082). +Their per-case maximum/minimum ratios range from 1.048 to 1.203. The last run's +six recorded requests passed E2E validation and failed the then-missing loading +baseline check; subsequent checks and MiniMax's second request did not run. +The existing MiniMax wall-clock tolerance override is unchanged. + +The same three runs also provide initial two-H100 loading references for +`ltx_2.3_one_stage_ti2v`, `ltx_2.3_two_stage_t2v_2gpus`, +`wan2_1_t2v_1.3b_cfg_parallel`, and `zimage_image_t2i_2_gpus`. +Seven partition-0 cases also meet this criterion: `flux2_modelopt_fp8_tp2_t2i`, +`flux_image_t2i_2_gpus`, `ideogram4_fp8_tp2_t2i`, `qwen_image_t2i_2_gpus`, +`wan2_2_i2v_a14b_2gpu`, `wan2_2_t2v_a14b_lora_2gpu`, and +`wan2_2_t2v_a14b_teacache_2gpu`. All eleven cases meet the same three-run +stability criterion. Existing loading references are not raised when a later +run exceeds their limits. + +Five more two-H100 references use runs +[34755864886](https://github.com/sgl-project/sglang/actions/runs/34755864886), +[34764411082](https://github.com/sgl-project/sglang/actions/runs/34764411082), and +[34766506722](https://github.com/sgl-project/sglang/actions/runs/34766506722): +`flux_2_image_t2i_2_gpus`, `fsdp-inference`, `mova_360p_tp2`, +`wan2_1_i2v_14b_720P_2gpu`, and `zimage_image_t2i_2_gpus_non_square`. +Their maximum/minimum loading ratios range from 1.093 to 1.205. Each reference +is the minimum observed loading time rounded to two decimal places; the same +existing tolerances apply. These are initial references, not claims that the +full testcases or all later checks passed. + +Seven further H100 references use that same recent three-run window: +`flux_image_t2i`, `flux_2_ti2i`, `joyai_image_edit_ti2i`, +`qwen_image_edit_2509_ti2i`, `qwen_image_layered_i2i`, +`minimax_h3_t2va_2gpu_h100`, and `qwen_image_edit_modelopt_fp8_ti2i`. +Their maximum/minimum ratios in this window range from 1.006 to 1.250 +(the largest unrounded ratio is 1.249565). Earlier historical runs vary more; +these references do not claim stability across the entire history. Each value +is the minimum in the stated window, rounded to two decimals. Existing loading +references, E2E references, and tolerances are unchanged. + +Four additional single-H100 references follow the same minimum-of-three rule. +`flux_2_image_t2i_upscaling_4x` uses runs 34764411082, 34766506722, and +[34770008768](https://github.com/sgl-project/sglang/actions/runs/34770008768). +`flux_2_t2i_customized_vae_path`, `flux_2_ti2i_multi_image_cache_dit`, and +`zimage_image_t2i` use runs 34766506722, 34770008768, and +[34771349145](https://github.com/sgl-project/sglang/actions/runs/34771349145). +Their maximum/minimum ratios range from 1.090 to 1.224. These are initial +loading references only; earlier variable runs remain diagnostic evidence, +and no existing performance reference or tolerance is increased. + +`qwen_image_edit_ti2i` and `lingbot_world_realtime_plastic_beach` use runs +34764411082, 34766506722, and 34771349145. Their minimum loading times are +36793.38 ms and 31600.49 ms, respectively, with maximum/minimum ratios of +1.1745 and 1.1256. The realtime case uses the same process-start-to-ready +loading boundary, excluding warmup; its request E2E reference is unchanged. + +### Best-observed references for variable cases + +The remaining 16 H100 and five B200 cases use the fastest valid startup in +the saved reports from runs 34755864886, 34764411082, 34766506722, +34770008768, and 34771349145, where available. Each value is rounded to two +decimals. H100 cases have three to five distinct startups; B200 cases have +two. These are initial measured references, not claims of stability or proof +that every historical run passed. Requiring all noisy samples to converge +before establishing a guard would leave these cases without a quantified limit. + +No existing reference or tolerance is increased. Samples above the resulting +limit still fail, and their infrastructure/code diagnosis remains separate. +In particular, LTX HQ retains the observed 116.531 s startup as its reference, +not the later 187-208 s startups. Downloads in H200 development measurements +are not used to establish either GPU pool's loading references. + +| Source run | GPU | Cases supplying the minimum | +| --- | --- | --- | +| 34755864886 | H100 | `fast_hunyuan_video`, `joy_echo_t2v_2gpu`, `wan2_1_i2v_14b_480P_2gpu`, `wan2_1_t2v_14b_2gpu`, `wan2_2_t2v_a14b_2gpu` | +| 34764411082 | H100 | `lingbot_video_moe_t2v`, `ltx_2_3_hq_pipeline`, `qwen_image_t2i_2_gpus_extra_high`, `wan22_modelopt_fp8_t2v` | +| 34766506722 | H100 | `ltx_2_3_two_stage_ti2v_2gpus`, `ltx_2_5_diffusion_decoder_2gpus`, `ltx_2_two_stage_t2v`, `minimax_h3_ref2va_video_audio_2gpu_h100`, `sana_wm_ti2v`, `wan2_1_i2v_14b_lora_2gpu`, `wan2_1_t2v_1_3b_cache_dit_sp_only_2gpu` | +| 34755864886 | B200 | `flux1_modelopt_nvfp4_t2i`, `flux2_modelopt_nvfp4_t2i` | +| 34764411082 | B200 | `ideogram4_nvfp4_t2i`, `qwen_image_2512_modelopt_nvfp4_t2i`, `wan22_modelopt_nvfp4_t2v` | diff --git a/python/sglang/multimodal_gen/test/runner/pytest_runner.py b/python/sglang/multimodal_gen/test/runner/pytest_runner.py index 46d12901e..5e5e39b01 100644 --- a/python/sglang/multimodal_gen/test/runner/pytest_runner.py +++ b/python/sglang/multimodal_gen/test/runner/pytest_runner.py @@ -1,7 +1,9 @@ from __future__ import annotations +import os import subprocess import sys +import time import xml.etree.ElementTree as ET from pathlib import Path from typing import Sequence @@ -104,7 +106,6 @@ def _run_pytest_attempt(cmd: list[str]) -> tuple[int, str]: stderr=subprocess.STDOUT, bufsize=0, ) - output_bytes = bytearray() while True: chunk = process.stdout.read(4096) @@ -118,6 +119,18 @@ def _run_pytest_attempt(cmd: list[str]) -> tuple[int, str]: return process.returncode, output_bytes.decode("utf-8", errors="replace") +def _estimate_failed_test_time(xml_path: str | None, attempt_time: float) -> float: + if xml_path is None or not Path(xml_path).exists(): + return attempt_time + + failed_time = sum( + float(testcase.get("time", "0")) + for testcase in ET.parse(xml_path).getroot().iter("testcase") + if testcase.find("failure") is not None or testcase.find("error") is not None + ) + return failed_time if failed_time > 0 else attempt_time + + def _extract_collection_line(full_output: str) -> str | None: for line in full_output.splitlines(): stripped = line.strip() @@ -157,8 +170,7 @@ def _summary_has_retryable_failure(summary_lines: list[str]) -> bool: for line in summary_lines: lowered = line.lower() if ( - "[performance]" in line - or "SafetensorError" in line + "SafetensorError" in line or "FileNotFoundError" in line or "TimeoutError" in line or "out of memory" in lowered @@ -185,11 +197,10 @@ def _is_retryable_failure(full_output: str) -> bool: if _is_consistency_failure(full_output): return False + if "[performance]" in full_output: + return True + summary_lines = _extract_short_test_summary(full_output) - is_perf_assertion = ( - "multimodal_gen/test/server/test_server_utils.py" in full_output - and "AssertionError" in full_output - ) is_aggregated_retryable_failure = _summary_has_retryable_failure(summary_lines) is_flaky_ci_assertion = ( @@ -202,12 +213,7 @@ def _is_retryable_failure(full_output: str) -> bool: "out of memory" in full_output.lower() or "oom killer" in full_output.lower() ) - return ( - is_perf_assertion - or is_aggregated_retryable_failure - or is_flaky_ci_assertion - or is_oom_error - ) + return is_aggregated_retryable_failure or is_flaky_ci_assertion or is_oom_error def _print_attempt_tail_summary( @@ -284,6 +290,8 @@ def run_pytest( base_cmd.extend(["-k", filter_expr]) max_retries = 6 + retry_deadline = os.environ.get("SGLANG_DIFFUSION_RETRY_DEADLINE") + retry_deadline = float(retry_deadline) if retry_deadline else None attempt_reports = [] for i in range(max_retries + 1): is_retry = i > 0 @@ -298,7 +306,9 @@ def run_pytest( f"for {len(files)} assigned item(s)" ) + attempt_start = time.monotonic() returncode, full_output = _run_pytest_attempt(cmd) + attempt_time = time.monotonic() - attempt_start retryable = returncode not in (0, 5) and _is_retryable_failure(full_output) attempt_reports.append( { @@ -343,6 +353,21 @@ def run_pytest( _print_attempt_tail_summary(attempt_reports, len(files)) return (returncode, list(all_executed_cases), all_case_results) + if retry_deadline is not None: + remaining = retry_deadline - time.time() + retry_estimate = _estimate_failed_test_time(junit_xml_path, attempt_time) + # leave headroom for pytest startup and variation in the failed cases + required = retry_estimate * 1.1 + 30 + if remaining < required: + print( + f"Retry budget exhausted: {remaining:.1f}s remaining, " + f"next failed-item retry needs approximately {required:.1f}s. " + "Preserving the failing result instead of starting another attempt.", + flush=True, + ) + _print_attempt_tail_summary(attempt_reports, len(files)) + return (returncode, list(all_executed_cases), all_case_results) + print( f"Retryable failure detected on attempt {i + 1}. " "Retrying only previously failed items." diff --git a/python/sglang/multimodal_gen/test/scripts/gen_perf_baselines.py b/python/sglang/multimodal_gen/test/scripts/gen_perf_baselines.py index ef646f5f7..75e20fe1d 100644 --- a/python/sglang/multimodal_gen/test/scripts/gen_perf_baselines.py +++ b/python/sglang/multimodal_gen/test/scripts/gen_perf_baselines.py @@ -1,6 +1,7 @@ import argparse import inspect import json +import math import os import re import sys @@ -137,6 +138,9 @@ def _run_case(case: DiffusionTestCase) -> dict: perf = PerformanceSummary.from_req_perf_record( rec, BASELINE_CONFIG.step_fractions ) + for name, value in (("load", ctx.load_time_ms), ("E2E", perf.e2e_ms)): + if value is None or not (math.isfinite(value) and value > 0): + raise ValueError(f"{case.id}: {name} duration missing or invalid") if case.server_args.modality == "video" and sp.num_frames and sp.num_frames > 0: if "per_frame_generation" not in perf.stage_metrics: perf.stage_metrics["per_frame_generation"] = perf.e2e_ms / sp.num_frames @@ -147,6 +151,7 @@ def _run_case(case: DiffusionTestCase) -> dict: str(k): round(v, 2) for k, v in perf.all_denoise_steps.items() }, "expected_e2e_ms": round(perf.e2e_ms, 2), + "expected_load_ms": round(ctx.load_time_ms, 2), "expected_avg_denoise_ms": round(perf.avg_denoise_ms, 2), "expected_median_denoise_ms": round(perf.median_denoise_ms, 2), } diff --git a/python/sglang/multimodal_gen/test/server/gpu_cases.py b/python/sglang/multimodal_gen/test/server/gpu_cases.py index 6bbad707b..9a856fc86 100644 --- a/python/sglang/multimodal_gen/test/server/gpu_cases.py +++ b/python/sglang/multimodal_gen/test/server/gpu_cases.py @@ -115,6 +115,7 @@ ONE_GPU_CASES: list[DiffusionTestCase] = [ ), PI05_ACTION_CI_sampling_params, run_perf_check=False, + perf_warmup_requests=1, run_component_accuracy_check=False, run_t2v_input_reference_check=False, ), @@ -181,6 +182,10 @@ ONE_GPU_CASES: list[DiffusionTestCase] = [ DiffusionServerArgs( model_path=DEFAULT_COSMOS3_NANO_MODEL_NAME_FOR_TEST, modality="image", + extras=[ + "--warmup-num-frames 1", + "--component-residency transformer=resident", + ], ), COSMOS3_NANO_CI_sampling_params, run_perf_check=False, @@ -254,6 +259,8 @@ ONE_GPU_CASES: list[DiffusionTestCase] = [ DiffusionServerArgs( model_path=DEFAULT_COSMOS3_NANO_MODEL_NAME_FOR_TEST, modality="video", + # the latency baseline measures the warmed, resident transformer + extras=["--component-residency transformer=resident"], env_vars={"SGLANG_DISABLE_COSMOS3_GUARDRAILS": "1"}, ), DiffusionSamplingParams( @@ -398,6 +405,7 @@ ONE_GPU_CASES: list[DiffusionTestCase] = [ "sana_wm_ti2v", DiffusionServerArgs( model_path=DEFAULT_SANA_WM_STREAMING_MODEL_NAME_FOR_TEST, + extras=["--warmup-resolutions 384x640"], ), SANA_WM_TI2V_CI_sampling_params, run_perf_check=False, @@ -1091,6 +1099,9 @@ TWO_GPU_CASES = [ # decoder headroom on 80 GB GPUs. extras=[ "--load-diffusion-decoder", + "--warmup-resolutions 768x448", + "--warmup-num-frames 49", + """--warmup-sampling-params '{"use_diffusion_decoder":true}'""", "--component-residency " "transformer=component-offload,text_encoder=component-offload", ], @@ -1102,7 +1113,6 @@ TWO_GPU_CASES = [ expect_audio_output=True, extras={"seed": 42, "use_diffusion_decoder": True}, ), - run_perf_check=False, run_component_accuracy_check=False, ), # I2V LoRA test case @@ -1160,7 +1170,6 @@ TWO_GPU_CASES = [ ring_degree=2, ), replace(T2I_sampling_params, extras={"quality": "extra-high"}), - run_perf_check=False, run_component_accuracy_check=False, run_models_api_check=False, run_t2v_input_reference_check=False, diff --git a/python/sglang/multimodal_gen/test/server/perf_baselines/5090.json b/python/sglang/multimodal_gen/test/server/perf_baselines/5090.json index 275c8c558..acae51a60 100644 --- a/python/sglang/multimodal_gen/test/server/perf_baselines/5090.json +++ b/python/sglang/multimodal_gen/test/server/perf_baselines/5090.json @@ -40,6 +40,7 @@ }, "scenarios": { "flux_2_klein_base_image_t2i": { + "expected_load_ms": 30018.87, "stages_ms": { "DecodingStage": 18.19, "DenoisingStage": 17612.9, @@ -67,6 +68,7 @@ "estimated_full_test_time_s": 94.0 }, "wan2_1_t2v_1.3b": { + "expected_load_ms": 46026.44, "stages_ms": { "DecodingStage": 495.01, "DenoisingStage": 21452.59, @@ -93,6 +95,7 @@ "estimated_full_test_time_s": 160.9 }, "turbo_wan2_1_t2v_1.3b": { + "expected_load_ms": 43084.93, "stages_ms": { "InputValidationStage": 0.11, "TextEncodingStage": 634.42, @@ -120,6 +123,7 @@ "estimated_full_test_time_s": 200.3 }, "zimage_image_t2i": { + "expected_load_ms": 43930.51, "stages_ms": { "DecodingStage": 7.11, "DenoisingStage": 2178.8, @@ -146,6 +150,7 @@ "estimated_full_test_time_s": 329.8 }, "flux_image_t2i_layerwise_cpu_offload_5090": { + "expected_load_ms": 47725.91, "stages_ms": { "InputValidationStage": 0.11, "TextEncodingStage": 204.63, @@ -194,6 +199,7 @@ "estimated_full_test_time_s": 90.0 }, "minimax_h3_t2va_consumer_budget_1gpu_5090": { + "expected_load_ms": 47708.93, "stages_ms": { "InputValidationStage": 0.06, "MiniMaxH3PartitionAdmissionStage": 0.03, diff --git a/python/sglang/multimodal_gen/test/server/perf_baselines/b200.json b/python/sglang/multimodal_gen/test/server/perf_baselines/b200.json index aa81eee87..004947744 100644 --- a/python/sglang/multimodal_gen/test/server/perf_baselines/b200.json +++ b/python/sglang/multimodal_gen/test/server/perf_baselines/b200.json @@ -5,6 +5,40 @@ "description": "Reference estimates for B200-only diffusion cases, split out from the shared diffusion baseline file.", "last_updated": "2026-07-01" }, + "runner_overrides": { + "b200-cirrascale1-0123": { + "flux1_modelopt_nvfp4_t2i": { + "expected_e2e_ms": 1574.32 + }, + "qwen_image_2512_modelopt_nvfp4_t2i": { + "expected_e2e_ms": 17742.04 + } + }, + "b200-cirrascale3-0123": { + "flux1_modelopt_nvfp4_t2i": { + "expected_e2e_ms": 1470.24 + }, + "qwen_image_2512_modelopt_nvfp4_t2i": { + "expected_e2e_ms": 17894.49 + } + }, + "b200-cirrascale3-4567": { + "flux1_modelopt_nvfp4_t2i": { + "expected_e2e_ms": 1471.19 + }, + "qwen_image_2512_modelopt_nvfp4_t2i": { + "expected_e2e_ms": 17346.65 + } + }, + "b200-di": { + "flux1_modelopt_nvfp4_t2i": { + "expected_e2e_ms": 1334.16 + }, + "qwen_image_2512_modelopt_nvfp4_t2i": { + "expected_e2e_ms": 16126.87 + } + } + }, "tolerances": { "long_term": { "e2e": 0.15, @@ -40,6 +74,7 @@ }, "scenarios": { "flux1_modelopt_nvfp4_t2i": { + "expected_load_ms": 38841.5, "stages_ms": { "InputValidationStage": 0.04, "TextEncodingStage": 16.55, @@ -68,6 +103,7 @@ "estimated_full_test_time_s": 71.2 }, "flux2_modelopt_nvfp4_t2i": { + "expected_load_ms": 71333.99, "stages_ms": { "InputValidationStage": 0.04, "TextEncodingStage": 330.43, @@ -97,6 +133,7 @@ "estimated_full_test_time_s": 592.3 }, "qwen_image_2512_modelopt_nvfp4_t2i": { + "expected_load_ms": 46619.98, "stages_ms": { "InputValidationStage": 0.04, "TextEncodingStage": 221.4, @@ -163,6 +200,7 @@ "estimated_full_test_time_s": 120.0 }, "wan22_modelopt_nvfp4_t2v": { + "expected_load_ms": 79653.94, "stages_ms": { "InputValidationStage": 0.04, "TextEncodingStage": 316.23, @@ -191,6 +229,7 @@ "estimated_full_test_time_s": 181.8 }, "ideogram4_nvfp4_t2i": { + "expected_load_ms": 33439.28, "stages_ms": { "InputValidationStage": 0.04, "Ideogram4TextEncodingStage": 129.35, 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 f1a6219ed..78708d108 100644 --- a/python/sglang/multimodal_gen/test/server/perf_baselines/h100.json +++ b/python/sglang/multimodal_gen/test/server/perf_baselines/h100.json @@ -39,7 +39,32 @@ ] }, "scenarios": { + "sana_wm_ti2v": { + "expected_load_ms": 32625.62, + "stages_ms": {}, + "denoise_step_ms": {}, + "expected_e2e_ms": 7239.81, + "expected_avg_denoise_ms": 0.0, + "expected_median_denoise_ms": 0.0 + }, + "sana_video_2b_t2v": { + "expected_load_ms": 25382.51, + "stages_ms": {}, + "denoise_step_ms": {}, + "expected_e2e_ms": 2472.55, + "expected_avg_denoise_ms": 0.0, + "expected_median_denoise_ms": 0.0 + }, + "pi05_action_http": { + "expected_load_ms": 25954.41, + "stages_ms": {}, + "denoise_step_ms": {}, + "expected_e2e_ms": 21.37, + "expected_avg_denoise_ms": 0.0, + "expected_median_denoise_ms": 0.0 + }, "qwen_image_t2i": { + "expected_load_ms": 40799.72, "stages_ms": { "TextEncodingStage": 232.03, "DenoisingStage": 12402.12, @@ -110,6 +135,7 @@ "estimated_full_test_time_s": 133.1 }, "qwen_image_t2i_2_gpus": { + "expected_load_ms": 41606.84, "stages_ms": { "InputValidationStage": 0.04, "TextEncodingStage": 457.66, @@ -188,33 +214,101 @@ "estimated_full_test_time_s": 300.0 }, "qwen_image_t2i_2_gpus_extra_high": { - "stages_ms": {}, - "denoise_step_ms": {}, - "expected_e2e_ms": 0.0, - "expected_avg_denoise_ms": 0.0, - "expected_median_denoise_ms": 0.0, + "expected_load_ms": 46120.38, + "notes": "H100 CI run 34695613376, job 103558583416, head 5ae9847fc7092a7c75a275556d672f7507a1f933; request 1 after server warmup. E2E also agrees with runs 34693958419 and 34687251777. Existing tolerances are unchanged.", + "stages_ms": { + "InputValidationStage": 0.04, + "TextEncodingStage": 455.77, + "LatentPreparationStage": 0.16, + "TimestepPreparationStage": 18.03, + "DenoisingStage": 9335.7, + "DecodingStage": 51.15 + }, + "denoise_step_ms": { + "0": 203.41, + "1": 191.47, + "2": 188.81, + "3": 186.65, + "4": 184.78, + "5": 185.17, + "6": 190.42, + "7": 191.36, + "8": 190.9, + "9": 183.73, + "10": 187.01, + "11": 184.09, + "12": 183.56, + "13": 182.9, + "14": 182.7, + "15": 197.53, + "16": 191.23, + "17": 188.21, + "18": 186.65, + "19": 184.42, + "20": 186.86, + "21": 185.31, + "22": 185.24, + "23": 184.98, + "24": 184.09, + "25": 183.76, + "26": 183.12, + "27": 182.28, + "28": 183.34, + "29": 183.79, + "30": 182.83, + "31": 182.17, + "32": 181.26, + "33": 182.59, + "34": 182.78, + "35": 182.32, + "36": 182.42, + "37": 182.26, + "38": 182.04, + "39": 183.17, + "40": 182.72, + "41": 181.68, + "42": 187.57, + "43": 190.77, + "44": 185.43, + "45": 185.76, + "46": 190.02, + "47": 182.43, + "48": 181.89, + "49": 183.58 + }, + "expected_e2e_ms": 9878.54, + "expected_avg_denoise_ms": 185.67, + "expected_median_denoise_ms": 184.09, + "load_peak_vram_mb": 42026.0, + "runtime_peak_vram_mb": 46696.0, + "warmup_peak_vram_mb": 43424.0, + "load_peak_allocated_mb": 41798.14, + "runtime_peak_allocated_mb": 44854.98, "estimated_full_test_time_s": 54.4 }, "flux2_modelopt_fp8_tp2_t2i": { + "expected_load_ms": 50693.24, "stages_ms": {}, "denoise_step_ms": {}, - "expected_e2e_ms": 0.0, + "expected_e2e_ms": 2089.62, "expected_avg_denoise_ms": 0.0, "expected_median_denoise_ms": 0.0, "estimated_full_test_time_s": 75.4 }, "joy_echo_t2v_2gpu": { + "expected_load_ms": 46333.83, "stages_ms": {}, "denoise_step_ms": {}, - "expected_e2e_ms": 0.0, + "expected_e2e_ms": 2469.07, "expected_avg_denoise_ms": 0.0, "expected_median_denoise_ms": 0.0, "estimated_full_test_time_s": 62.8 }, "ideogram4_fp8_tp2_t2i": { + "expected_load_ms": 33515.44, "stages_ms": {}, "denoise_step_ms": {}, - "expected_e2e_ms": 0.0, + "expected_e2e_ms": 8975.57, "expected_avg_denoise_ms": 0.0, "expected_median_denoise_ms": 0.0, "estimated_full_test_time_s": 60.8 @@ -283,6 +377,7 @@ "estimated_full_test_time_s": 120.0 }, "flux_image_t2i": { + "expected_load_ms": 29629.92, "stages_ms": { "TimestepPreparationStage": 32.58, "DenoisingStage": 6447.42, @@ -353,6 +448,7 @@ "estimated_full_test_time_s": 127.4 }, "flux_2_image_t2i": { + "expected_load_ms": 55239.13, "stages_ms": { "TimestepPreparationStage": 15.77, "DenoisingStage": 22276.29, @@ -424,6 +520,7 @@ "estimated_full_test_time_s": 145.2 }, "flux_2_klein_image_t2i": { + "expected_load_ms": 28510.14, "stages_ms": { "DecodingStage": 6.41, "InputValidationStage": 0.03, @@ -449,6 +546,7 @@ "estimated_full_test_time_s": 120.5 }, "flux_2_klein_base_image_t2i": { + "expected_load_ms": 26405.08, "stages_ms": { "InputValidationStage": 0.03, "TextEncodingStage": 53.19, @@ -520,6 +618,7 @@ "estimated_full_test_time_s": 124.4 }, "flux_2_ti2i": { + "expected_load_ms": 54947.9, "stages_ms": { "TextEncodingStage": 364.98, "DenoisingStage": 44465.25, @@ -591,6 +690,7 @@ "estimated_full_test_time_s": 168.9 }, "flux_2_ti2i_multi_image_cache_dit": { + "expected_load_ms": 54569.41, "stages_ms": { "ImageVAEEncodingStage": 156.01, "DenoisingStage": 23101.61, @@ -662,6 +762,7 @@ "estimated_full_test_time_s": 148.6 }, "flux_image_t2i_2_gpus": { + "expected_load_ms": 31766.33, "stages_ms": { "InputValidationStage": 0.03, "TextEncodingStage": 28.24, @@ -732,6 +833,7 @@ "estimated_full_test_time_s": 44.9 }, "zimage_image_t2i": { + "expected_load_ms": 28221.57, "stages_ms": { "DecodingStage": 6.23, "InputValidationStage": 0.03, @@ -761,6 +863,7 @@ "estimated_full_test_time_s": 116.3 }, "zimage_image_t2i_fp8": { + "expected_load_ms": 27623.5, "stages_ms": { "InputValidationStage": 0.04, "TextEncodingStage": 131.15, @@ -790,6 +893,7 @@ "estimated_full_test_time_s": 123.7 }, "zimage_image_t2i_multi_lora": { + "expected_load_ms": 36426.15, "stages_ms": { "InputValidationStage": 0.03, "TextEncodingStage": 129.63, @@ -819,6 +923,7 @@ "estimated_full_test_time_s": 162.1 }, "zimage_image_t2i_2_gpus": { + "expected_load_ms": 31619.48, "stages_ms": { "InputValidationStage": 0.04, "TextEncodingStage": 255.84, @@ -848,6 +953,7 @@ "estimated_full_test_time_s": 39.3 }, "qwen_image_edit_ti2i": { + "expected_load_ms": 36793.38, "stages_ms": { "InputValidationStage": 24.78, "ImageEncodingStage": 714.55, @@ -919,6 +1025,7 @@ "estimated_full_test_time_s": 153.6 }, "joyai_image_edit_ti2i": { + "expected_load_ms": 34589.89, "stages_ms": { "InputValidationStage": 16.43, "ImageEncodingStage": 602.8, @@ -980,6 +1087,7 @@ "estimated_full_test_time_s": 117.6 }, "qwen_image_t2i_cache_dit_enabled": { + "expected_load_ms": 34703.95, "stages_ms": { "InputValidationStage": 0.03, "TextEncodingStage": 231.07, @@ -1050,6 +1158,7 @@ "estimated_full_test_time_s": 124.9 }, "wan2_1_t2v_1.3b_teacache_enabled": { + "expected_load_ms": 32869.61, "stages_ms": { "TextEncodingStage": 534.94, "DenoisingStage": 3853.51, @@ -1120,6 +1229,7 @@ "estimated_full_test_time_s": 126.0 }, "wan2_1_t2v_1.3b": { + "expected_load_ms": 30234.05, "stages_ms": { "DecodingStage": 361.6, "InputValidationStage": 0.03, @@ -1190,6 +1300,7 @@ "estimated_full_test_time_s": 129.3 }, "wan2_1_t2v_1.3b_cfg_parallel": { + "expected_load_ms": 30166.76, "stages_ms": { "LatentPreparationStage": 0.1, "InputValidationStage": 0.05, @@ -1260,6 +1371,7 @@ "estimated_full_test_time_s": 45.5 }, "turbo_wan2_1_t2v_1.3b": { + "expected_load_ms": 27199.67, "stages_ms": { "InputValidationStage": 0.03, "TextEncodingStage": 364.61, @@ -1285,6 +1397,7 @@ "estimated_full_test_time_s": 124.7 }, "ltx_2_two_stage_t2v": { + "expected_load_ms": 47470.86, "stages_ms": { "InputValidationStage": 0.03, "TextEncodingStage": 403.5, @@ -1356,6 +1469,7 @@ "estimated_full_test_time_s": 100.4 }, "wan2_2_ti2v_5b": { + "expected_load_ms": 37605.69, "stages_ms": { "InputValidationStage": 706.07, "TextEncodingStage": 327.65, @@ -1426,6 +1540,7 @@ "estimated_full_test_time_s": 141.7 }, "qwen_image_edit_2509_ti2i": { + "expected_load_ms": 37508.7, "stages_ms": { "ImageEncodingStage": 587.89, "DenoisingStage": 37539.33, @@ -1487,6 +1602,7 @@ "estimated_full_test_time_s": 160.2 }, "qwen_image_layered_i2i": { + "expected_load_ms": 37439.25, "stages_ms": { "QwenImageLayeredBeforeDenoisingStage": 144.26, "TimestepPreparationStage": 0.0, @@ -1555,6 +1671,7 @@ "estimated_full_test_time_s": 161.5 }, "fastwan2_2_ti2v_5b": { + "expected_load_ms": 31379.65, "stages_ms": { "InputValidationStage": 400.0, "TextEncodingStage": 327.82, @@ -1578,6 +1695,7 @@ "estimated_full_test_time_s": 125.2 }, "fast_hunyuan_video": { + "expected_load_ms": 32322.52, "stages_ms": { "InputValidationStage": 0.04, "TextEncodingStage": 252.98, @@ -1606,6 +1724,7 @@ "estimated_full_test_time_s": 77.0 }, "wan2_2_i2v_a14b_2gpu": { + "expected_load_ms": 78667.8, "stages_ms": { "InputValidationStage": 15.59, "ImageVAEEncodingStage": 1484.2, @@ -1667,6 +1786,7 @@ "estimated_full_test_time_s": 168.7 }, "wan2_1_i2v_14b_480P_2gpu": { + "expected_load_ms": 46402.22, "stages_ms": { "InputValidationStage": 7.88, "LatentPreparationStage": 0.1, @@ -1740,6 +1860,7 @@ "estimated_full_test_time_s": 128.8 }, "wan2_1_i2v_14b_720P_2gpu": { + "expected_load_ms": 37549.49, "stages_ms": { "InputValidationStage": 11.58, "TextEncodingStage": 327.33, @@ -1812,6 +1933,7 @@ "estimated_full_test_time_s": 182.2 }, "wan2_2_t2v_a14b_2gpu": { + "expected_load_ms": 61832.44, "stages_ms": { "InputValidationStage": 0.03, "TextEncodingStage": 314.11, @@ -1873,6 +1995,7 @@ "estimated_full_test_time_s": 188.0 }, "wan2_1_t2v_14b_2gpu": { + "expected_load_ms": 45201.26, "stages_ms": { "TextEncodingStage": 325.79, "DecodingStage": 637.43, @@ -1943,6 +2066,7 @@ "estimated_full_test_time_s": 96.3 }, "wan2_2_t2v_a14b_lora_2gpu": { + "expected_load_ms": 61440.19, "stages_ms": { "InputValidationStage": 0.03, "TextEncodingStage": 325.03, @@ -2003,6 +2127,7 @@ "estimated_full_test_time_s": 576.3 }, "wan2_1_t2v_1_3b_lora_1gpu": { + "expected_load_ms": 30139.26, "stages_ms": { "InputValidationStage": 0.05, "TextEncodingStage": 329.6, @@ -2073,6 +2198,7 @@ "estimated_full_test_time_s": 129.6 }, "wan2_1_i2v_14b_lora_2gpu": { + "expected_load_ms": 66554.82, "stages_ms": { "InputValidationStage": 12.06, "TextEncodingStage": 325.79, @@ -2145,6 +2271,7 @@ "estimated_full_test_time_s": 509.3 }, "flux_2_image_t2i_2_gpus": { + "expected_load_ms": 53086.36, "stages_ms": { "InputValidationStage": 0.04, "TextEncodingStage": 372.55, @@ -2216,6 +2343,7 @@ "estimated_full_test_time_s": 75.0 }, "qwen_image_edit_2511_ti2i": { + "expected_load_ms": 38727.15, "stages_ms": { "DecodingStage": 18.69, "InputValidationStage": 48.98, @@ -2277,6 +2405,7 @@ "estimated_full_test_time_s": 143.7 }, "fsdp-inference": { + "expected_load_ms": 31380.54, "stages_ms": { "InputValidationStage": 0.03, "LatentPreparationStage": 0.11, @@ -2306,6 +2435,7 @@ "estimated_full_test_time_s": 61.3 }, "hunyuan3d_shape_gen": { + "expected_load_ms": 31925.92, "stages_ms": { "Hunyuan3DShapeBeforeDenoisingStage": 54.28, "Hunyuan3DShapeDenoisingStage": 1698.03, @@ -2377,6 +2507,7 @@ "estimated_full_test_time_s": 420.1 }, "wan2_1_t2v_1.3b_frame_interp_2x": { + "expected_load_ms": 30176.3, "stages_ms": { "InputValidationStage": 0.05, "TextEncodingStage": 534.56, @@ -2447,6 +2578,7 @@ "estimated_full_test_time_s": 129.3 }, "flux_2_image_t2i_upscaling_4x": { + "expected_load_ms": 52489.37, "stages_ms": { "TextEncodingStage": 373.42, "DenoisingStage": 21888.49, @@ -2518,6 +2650,7 @@ "estimated_full_test_time_s": 145.1 }, "wan2_1_t2v_1.3b_upscaling_4x": { + "expected_load_ms": 31719.84, "stages_ms": { "DecodingStage": 362.62, "InputValidationStage": 0.04, @@ -2588,6 +2721,7 @@ "estimated_full_test_time_s": 129.3 }, "wan2_1_t2v_1.3b_frame_interp_2x_upscaling_4x": { + "expected_load_ms": 30083.06, "stages_ms": { "InputValidationStage": 0.04, "TextEncodingStage": 532.52, @@ -2658,6 +2792,7 @@ "estimated_full_test_time_s": 129.4 }, "ltx_2.3_one_stage_ti2v": { + "expected_load_ms": 39264.11, "stages_ms": { "InputValidationStage": 3.49, "TextEncodingStage": 397.91, @@ -2711,6 +2846,7 @@ "estimated_full_test_time_s": 167.9 }, "ltx_2.3_two_stage_t2v_2gpus": { + "expected_load_ms": 45206.77, "stages_ms": { "InputValidationStage": 0.04, "TextEncodingStage": 399.08, @@ -2772,14 +2908,41 @@ "estimated_full_test_time_s": 186.0 }, "ltx_2_5_diffusion_decoder_2gpus": { - "stages_ms": {}, - "denoise_step_ms": {}, - "expected_e2e_ms": 0.0, - "expected_avg_denoise_ms": 0.0, - "expected_median_denoise_ms": 0.0, + "expected_load_ms": 28133.24, + "stages_ms": { + "InputValidationStage": 0.05, + "TextEncodingStage": 634.13, + "LTX2TextConnectorStage": 317.43, + "LTX2DurationStage": 0.01, + "LTX2SigmaPreparationStage": 0.03, + "TimestepPreparationStage": 246.11, + "LTX2AVLatentPreparationStage": 0.31, + "LTX2ImageEncodingStage": 0.02, + "LTX2AVDenoisingStage": 3750.7, + "LTX2AVDecodingStage": 3382.33 + }, + "denoise_step_ms": { + "0": 292.65, + "1": 259.94, + "2": 256.47, + "3": 256.26, + "4": 259.62, + "5": 255.44, + "6": 258.02, + "7": 255.91 + }, + "expected_e2e_ms": 10141.6, + "expected_avg_denoise_ms": 261.79, + "expected_median_denoise_ms": 257.25, + "load_peak_vram_mb": 1906, + "runtime_peak_vram_mb": 69546, + "warmup_peak_vram_mb": 28396, + "load_peak_allocated_mb": 1876.14, + "runtime_peak_allocated_mb": 48399, "estimated_full_test_time_s": 545.1 }, "ltx_2_3_two_stage_ti2v_2gpus": { + "expected_load_ms": 50003.37, "stages_ms": { "InputValidationStage": 3.21, "TextEncodingStage": 411.15, @@ -2841,6 +3004,7 @@ "estimated_full_test_time_s": 101.4 }, "longlive2_t2v": { + "expected_load_ms": 29328.4, "stages_ms": { "InputValidationStage": 0.04, "LongLive2TextEncodingStage": 328.05, @@ -2861,6 +3025,7 @@ "estimated_full_test_time_s": 153.1 }, "longlive2_i2v": { + "expected_load_ms": 27107.25, "stages_ms": { "InputValidationStage": 23.02, "LongLive2TextEncodingStage": 327.98, @@ -2881,6 +3046,7 @@ "estimated_full_test_time_s": 149.4 }, "lingbot_video_moe_t2v": { + "expected_load_ms": 22654.46, "stages_ms": { "InputValidationStage": 0.06, "LingBotVideoTextEncodingStage": 89.03, @@ -2913,6 +3079,7 @@ "estimated_full_test_time_s": 600.0 }, "lingbot_world_realtime_plastic_beach": { + "expected_load_ms": 31600.49, "stages_ms": { "RealtimeInputValidationStage": 0.09, "RealtimeTextEncodingStage": 0.04, @@ -2924,7 +3091,7 @@ "CausalVaeDecodingStage": 217.75 }, "denoise_step_ms": {}, - "expected_e2e_ms": 1912.29, + "expected_e2e_ms": 18687.61, "expected_avg_denoise_ms": 0.0, "expected_median_denoise_ms": 0.0, "load_peak_vram_mb": 42814.0, @@ -2934,6 +3101,7 @@ "estimated_full_test_time_s": 126.0 }, "ltx_2_3_hq_pipeline": { + "expected_load_ms": 116530.53, "stages_ms": { "InputValidationStage": 4.63, "TextEncodingStage": 401.75, @@ -2982,6 +3150,7 @@ "estimated_full_test_time_s": 363.2 }, "qwen_image_t2i_cache_dit_scm_config_diffusers_1gpu": { + "expected_load_ms": 35585.13, "stages_ms": { "DiffusersExecutionStage": 1075.16 }, @@ -2996,6 +3165,7 @@ "estimated_full_test_time_s": 98.3 }, "cosmos3_nano_t2i": { + "expected_load_ms": 28691.77, "stages_ms": { "Cosmos3ImagePreprocessStage": 0.01, "Cosmos3TokenizationStage": 155.89, @@ -3047,6 +3217,7 @@ "estimated_full_test_time_s": 65.0 }, "cosmos3_nano_t2v": { + "expected_load_ms": 30688.3, "stages_ms": { "Cosmos3ImagePreprocessStage": 0.01, "Cosmos3TokenizationStage": 143.82, @@ -3068,30 +3239,34 @@ "estimated_full_test_time_s": 65.0 }, "flux_2_t2i_customized_vae_path": { + "expected_load_ms": 52103.72, "stages_ms": {}, "denoise_step_ms": {}, - "expected_e2e_ms": 0.0, + "expected_e2e_ms": 23073.55, "expected_avg_denoise_ms": 0.0, "expected_median_denoise_ms": 0.0, "estimated_full_test_time_s": 574.4 }, "wan2_2_t2v_a14b_teacache_2gpu": { + "expected_load_ms": 63938.03, "stages_ms": {}, "denoise_step_ms": {}, - "expected_e2e_ms": 0.0, + "expected_e2e_ms": 82478.23, "expected_avg_denoise_ms": 0.0, "expected_median_denoise_ms": 0.0, "estimated_full_test_time_s": 186.4 }, "wan2_1_t2v_1_3b_cache_dit_sp_only_2gpu": { + "expected_load_ms": 34366.19, "stages_ms": {}, "denoise_step_ms": {}, - "expected_e2e_ms": 0.0, + "expected_e2e_ms": 1077.99, "expected_avg_denoise_ms": 0.0, "expected_median_denoise_ms": 0.0, "estimated_full_test_time_s": 52.2 }, "minimax_h3_ref2va_video_audio_2gpu_h100": { + "expected_load_ms": 68100.42, "stages_ms": { "InputValidationStage": 0.08, "MiniMaxH3PartitionAdmissionStage": 0.04, @@ -3120,6 +3295,7 @@ "estimated_full_test_time_s": 340.0 }, "minimax_h3_t2va_2gpu_h100": { + "expected_load_ms": 79373.57, "stages_ms": { "InputValidationStage": 0.05, "MiniMaxH3PartitionAdmissionStage": 0.03, @@ -3151,14 +3327,16 @@ "estimated_full_test_time_s": 235.0 }, "mova_360p_tp2": { + "expected_load_ms": 60480.85, "stages_ms": {}, "denoise_step_ms": {}, - "expected_e2e_ms": 0.0, + "expected_e2e_ms": 58242.09, "expected_avg_denoise_ms": 0.0, "expected_median_denoise_ms": 0.0, "estimated_full_test_time_s": 141.9 }, "zimage_image_t2i_2_gpus_non_square": { + "expected_load_ms": 30675.16, "stages_ms": { "InputValidationStage": 0.04, "TextEncodingStage": 259.45, @@ -3188,9 +3366,10 @@ "estimated_full_test_time_s": 40.5 }, "flux1_modelopt_fp8_t2i": { + "expected_load_ms": 33073.38, "stages_ms": {}, "denoise_step_ms": {}, - "expected_e2e_ms": 0.0, + "expected_e2e_ms": 1138.33, "expected_avg_denoise_ms": 0.0, "expected_median_denoise_ms": 0.0, "estimated_full_test_time_s": 50.3 @@ -3204,14 +3383,16 @@ "estimated_full_test_time_s": 498.1 }, "wan22_modelopt_fp8_t2v": { + "expected_load_ms": 63380.1, "stages_ms": {}, "denoise_step_ms": {}, - "expected_e2e_ms": 0.0, + "expected_e2e_ms": 7908.23, "expected_avg_denoise_ms": 0.0, "expected_median_denoise_ms": 0.0, "estimated_full_test_time_s": 89.5 }, "hunyuanvideo_modelopt_fp8_t2v": { + "expected_load_ms": 25315.9, "stages_ms": { "InputValidationStage": 0.05, "TextEncodingStage": 33.67, @@ -3244,17 +3425,19 @@ "estimated_full_test_time_s": 64.3 }, "qwen_image_modelopt_fp8_t2i": { + "expected_load_ms": 47967.61, "stages_ms": {}, "denoise_step_ms": {}, - "expected_e2e_ms": 0.0, + "expected_e2e_ms": 3038.04, "expected_avg_denoise_ms": 0.0, "expected_median_denoise_ms": 0.0, "estimated_full_test_time_s": 65.5 }, "qwen_image_edit_modelopt_fp8_ti2i": { + "expected_load_ms": 49282.42, "stages_ms": {}, "denoise_step_ms": {}, - "expected_e2e_ms": 0.0, + "expected_e2e_ms": 2964.8, "expected_avg_denoise_ms": 0.0, "expected_median_denoise_ms": 0.0, "estimated_full_test_time_s": 73.7 diff --git a/python/sglang/multimodal_gen/test/server/realtime_consistency.py b/python/sglang/multimodal_gen/test/server/realtime_consistency.py index c40a36545..5eb3d4bf3 100644 --- a/python/sglang/multimodal_gen/test/server/realtime_consistency.py +++ b/python/sglang/multimodal_gen/test/server/realtime_consistency.py @@ -6,6 +6,7 @@ import asyncio import os import statistics import tempfile +import time from dataclasses import dataclass from pathlib import Path from typing import Any @@ -50,9 +51,11 @@ class RealtimeChunkStats: class RealtimeCollectionResult: frames: list[np.ndarray] chunk_stats: list[RealtimeChunkStats] + e2e_ms: float _REALTIME_CHUNK_STATS_BY_CASE: dict[str, list[RealtimeChunkStats]] = {} +_REALTIME_E2E_MS_BY_CASE: dict[str, float] = {} _REALTIME_KEY_FRAMES_BY_CASE: dict[str, list[np.ndarray]] = {} @@ -217,15 +220,20 @@ def validate_realtime_perf_stats( def record_realtime_perf_stats( - case_id: str, chunk_stats: list[RealtimeChunkStats] + case_id: str, chunk_stats: list[RealtimeChunkStats], e2e_ms: float ) -> None: _REALTIME_CHUNK_STATS_BY_CASE[case_id] = list(chunk_stats) + _REALTIME_E2E_MS_BY_CASE[case_id] = e2e_ms def pop_realtime_perf_stats(case_id: str) -> list[RealtimeChunkStats]: return _REALTIME_CHUNK_STATS_BY_CASE.pop(case_id, []) +def pop_realtime_e2e_ms(case_id: str) -> float | None: + return _REALTIME_E2E_MS_BY_CASE.pop(case_id, None) + + def select_realtime_key_frames(frames: list[np.ndarray]) -> list[np.ndarray]: if not frames: return [] @@ -357,6 +365,8 @@ async def collect_realtime_output( sent_event_indices.add(event_idx) async with websockets.connect(ws_url, max_size=None, ping_interval=None) as ws: + # exclude server startup, warmup and the later mp4 consistency encoding + request_start = time.perf_counter() await ws.send(msgspec.msgpack.encode(init_payload)) await send_events_for_boundary(ws, -1) @@ -399,5 +409,8 @@ async def collect_realtime_output( if header.get("is_final_frame_batch", True): received_chunks.add(chunk_index) await send_events_for_boundary(ws, chunk_index) + e2e_ms = (time.perf_counter() - request_start) * 1000 - return RealtimeCollectionResult(frames=frames, chunk_stats=chunk_stats) + return RealtimeCollectionResult( + frames=frames, chunk_stats=chunk_stats, e2e_ms=e2e_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 a43263865..ed8dc9d36 100644 --- a/python/sglang/multimodal_gen/test/server/test_server_common.py +++ b/python/sglang/multimodal_gen/test/server/test_server_common.py @@ -8,6 +8,7 @@ Each collected request prints a performance log before validation. from __future__ import annotations import json +import math import os import queue import threading @@ -26,6 +27,7 @@ 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.realtime_consistency import ( RealtimeChunkStats, + pop_realtime_e2e_ms, pop_realtime_key_frames, pop_realtime_perf_stats, validate_realtime_perf_stats, @@ -101,6 +103,10 @@ _SERVER_FATAL_LOG_PATTERNS = ( _CASE_LOG_SEPARATOR = "=" * 88 +class PerformanceValidationError(AssertionError): + """A terminal performance failure, including across repeated requests.""" + + def _print_case_log_separator(case_id: str, state: str) -> None: print( f"\n{_CASE_LOG_SEPARATOR}\n" @@ -371,11 +377,14 @@ class DiffusionServerBase: log_path = ctx.perf_log_path log_wait_timeout = 30 - req_perf_record = wait_for_req_perf_record( - rid, - log_path, - timeout=log_wait_timeout, - ) + try: + req_perf_record = wait_for_req_perf_record( + rid, + log_path, + timeout=log_wait_timeout, + ) + except AssertionError as exc: + raise PerformanceValidationError(f"[performance] {case_id}: {exc}") from exc return (req_perf_record, content) @@ -384,8 +393,13 @@ class DiffusionServerBase: case: DiffusionTestCase, perf_record: RequestPerfRecord, request_index: int = 1, + load_time_ms: float | None = None, ) -> None: """Validate metrics and record results.""" + if perf_record is None: + raise PerformanceValidationError( + f"[performance] {case.id}: request performance record is missing" + ) is_baseline_generation_mode = os.environ.get("SGLANG_GEN_BASELINE", "0") == "1" scenario = BASELINE_CONFIG.scenarios.get(case.id) @@ -412,22 +426,25 @@ class DiffusionServerBase: ) summary = validator.collect_metrics(perf_record) - self._print_performance_log(case, summary, scenario) + summary.load_time_ms = load_time_ms self._record_performance_result(case, summary, request_index) + self._print_performance_log(case, summary, scenario) + + if is_baseline_generation_mode: + _PENDING_BASELINE_DUMPS.setdefault(case.id, []).append(summary) + return + + if missing_scenario: + self._dump_baseline_for_testcase(case, summary, missing_scenario) + pytest.fail( + f"Testcase '{case.id}' not found in {get_perf_baseline_update_path()}" + ) + + # disabling stage checks must not disable the request's e2e guard + validator.validate_e2e(summary) + validator.validate_load(summary) if case.run_perf_check: - if is_baseline_generation_mode: - _PENDING_BASELINE_DUMPS.setdefault(case.id, []).append(summary) - return - - if missing_scenario: - self._dump_baseline_for_testcase(case, summary, missing_scenario) - if missing_scenario: - pytest.fail( - f"Testcase '{case.id}' not found in {get_perf_baseline_update_path()}" - ) - return - if current_platform.is_cuda(): expected_load_peak_vram_mb = scenario.load_peak_vram_mb expected_runtime_peak_vram_mb = scenario.runtime_peak_vram_mb @@ -476,6 +493,57 @@ class DiffusionServerBase: chunk_stats: list[RealtimeChunkStats], request_index: int = 1, ) -> None: + e2e_ms = pop_realtime_e2e_ms(case.id) + scenario = BASELINE_CONFIG.scenarios.get(case.id) + summary = PerformanceSummary(e2e_ms, 0, 0, {}, [], {}, {}) + check_memory = case.run_perf_check and current_platform.is_cuda() + if check_memory: + request_id = next( + (stat.request_id for stat in reversed(chunk_stats) if stat.request_id), + None, + ) + if request_id is None: + pytest.fail(f"{case.id}: realtime chunk stats are missing request IDs") + + perf_record = wait_for_req_perf_record( + request_id, ctx.perf_log_path, timeout=30 + ) + if perf_record is None: + pytest.fail( + f"{case.id}: realtime request performance record is missing" + ) + if scenario is None: + pytest.fail( + f"Testcase '{case.id}' not found in {get_perf_baseline_update_path()}" + ) + validator = PerformanceValidator( + scenario=scenario, + tolerances=BASELINE_CONFIG.tolerances, + step_fractions=BASELINE_CONFIG.step_fractions, + ) + summary = validator.collect_metrics(perf_record) + # the last chunk's record supplies memory peaks, not the session's e2e + summary.e2e_ms = e2e_ms + + summary.load_time_ms = ctx.load_time_ms + self._record_performance_result(case, summary, request_index) + self._print_performance_log(case, summary, scenario) + if os.environ.get("SGLANG_GEN_BASELINE", "0") == "1": + _PENDING_BASELINE_DUMPS.setdefault(case.id, []).append(summary) + return + + if scenario is None: + pytest.fail( + f"Testcase '{case.id}' not found in {get_perf_baseline_update_path()}" + ) + if not check_memory: + validator = PerformanceValidator( + scenario=scenario, + tolerances=BASELINE_CONFIG.tolerances, + step_fractions=BASELINE_CONFIG.step_fractions, + ) + validator.validate_e2e(summary) + validator.validate_load(summary) validate_realtime_perf_stats( case.id, chunk_stats, @@ -484,48 +552,7 @@ class DiffusionServerBase: case.sampling_params.realtime_perf_ignore_initial_chunks ), ) - if not case.run_perf_check or not current_platform.is_cuda(): - return - - request_id = next( - (stat.request_id for stat in reversed(chunk_stats) if stat.request_id), - None, - ) - if request_id is None: - pytest.fail(f"{case.id}: realtime chunk stats are missing request IDs") - - perf_record = wait_for_req_perf_record( - request_id, - ctx.perf_log_path, - timeout=30, - ) - if perf_record is None: - pytest.fail(f"{case.id}: realtime request performance record is missing") - - scenario = BASELINE_CONFIG.scenarios.get(case.id) - if scenario is None: - pytest.fail( - f"Testcase '{case.id}' not found in {get_perf_baseline_update_path()}" - ) - - validator = PerformanceValidator( - scenario=scenario, - tolerances=BASELINE_CONFIG.tolerances, - step_fractions=BASELINE_CONFIG.step_fractions, - ) - summary = validator.collect_metrics(perf_record) - self._print_performance_log(case, summary, scenario) - self._record_performance_result(case, summary, request_index) - - if os.environ.get("SGLANG_GEN_BASELINE", "0") == "1": - logger.info( - "%s realtime peak VRAM baseline: load=%.0fMiB, runtime=%.0fMiB, " - "warmup=%.0fMiB", - case.id, - summary.load_peak_vram_mb, - summary.runtime_peak_vram_mb, - summary.warmup_peak_vram_mb, - ) + if not check_memory: return if scenario.load_peak_vram_mb is None or scenario.runtime_peak_vram_mb is None: @@ -555,12 +582,28 @@ class DiffusionServerBase: summary: PerformanceSummary, request_index: int = 1, ) -> None: + if not isinstance(summary.e2e_ms, (int, float)) or not ( + math.isfinite(summary.e2e_ms) and summary.e2e_ms > 0 + ): + raise PerformanceValidationError( + f"[performance] {case.id}: E2E duration missing or invalid: " + f"{summary.e2e_ms!r}" + ) + if summary.load_time_ms is None or not ( + math.isfinite(summary.load_time_ms) and summary.load_time_ms > 0 + ): + raise PerformanceValidationError( + f"[performance] {case.id}: Load duration missing or invalid: " + f"{summary.load_time_ms!r}" + ) result = { "class_name": type(self).__name__, "test_name": case.id, "request_index": request_index, "modality": case.server_args.modality, "e2e_ms": summary.e2e_ms, + "load_time_ms": summary.load_time_ms, + "load_inclusive_e2e_ms": summary.load_time_ms + summary.e2e_ms, "avg_denoise_ms": summary.avg_denoise_ms, "median_denoise_ms": summary.median_denoise_ms, "load_peak_vram_mb": summary.load_peak_vram_mb, @@ -595,6 +638,8 @@ class DiffusionServerBase: f"--- Performance Log: {case.id} ---", ( f" e2e={summary.e2e_ms:.2f}ms, " + f"load={summary.load_time_ms:.2f}ms, " + f"load_inclusive_e2e={summary.load_time_ms + summary.e2e_ms:.2f}ms, " f"avg_denoise={summary.avg_denoise_ms:.2f}ms, " f"median_denoise={summary.median_denoise_ms:.2f}ms, " f"load_peak_vram={summary.load_peak_vram_mb:.0f}MiB, " @@ -662,6 +707,7 @@ class DiffusionServerBase: "stages_ms": stages_formatted, "denoise_step_ms": denoise_steps_formatted, "expected_e2e_ms": round(max(s.e2e_ms for s in summaries), 2), + "expected_load_ms": round(max(s.load_time_ms for s in summaries), 2), "expected_avg_denoise_ms": round( max(s.avg_denoise_ms for s in summaries), 2 ), @@ -1573,6 +1619,28 @@ Pinned revision used by this check: {SGL_TEST_FILES_CI_DATA_REVISION} if case.run_lora_dynamic_load_check: self._test_dynamic_lora_loading(diffusion_server, case) + for warmup_index in range(case.perf_warmup_requests): + label = f"request warmup {warmup_index + 1}/{case.perf_warmup_requests}" + _print_case_log_separator(case.id, f"BEGIN {label}") + generate_fn = get_generate_fn( + model_path=case.server_args.model_path, + modality=case.server_args.modality, + sampling_params=case.sampling_params, + ) + record, _ = self.run_and_collect( + diffusion_server, case.id, generate_fn, collect_perf=True + ) + if record is None or not ( + math.isfinite(record.total_duration_ms) and record.total_duration_ms > 0 + ): + raise PerformanceValidationError( + f"[performance] {case.id}: {label} E2E duration missing or invalid" + ) + print( + f"[server-test] {case.id}: {label} e2e={record.total_duration_ms:.4f}ms" + ) + _print_case_log_separator(case.id, f"END {label}") + failures = [] for request_index in range(1, case.perf_repeat_requests + 1): label = f"request {request_index}/{case.perf_repeat_requests}" @@ -1586,6 +1654,9 @@ Pinned revision used by this check: {SGL_TEST_FILES_CI_DATA_REVISION} str(Path(artifact_dir) / f"request-{request_index}"), ) self._test_diffusion_request(case, diffusion_server, request_index) + except PerformanceValidationError as exc: + _print_case_log_separator(case.id, f"FAILED {label}") + raise PerformanceValidationError(f"[{label}] {exc}") from exc except pytest.skip.Exception as exc: if request_index == 1: raise @@ -1635,6 +1706,10 @@ Pinned revision used by this check: {SGL_TEST_FILES_CI_DATA_REVISION} except BaseException as exc: if isinstance(exc, (KeyboardInterrupt, SystemExit)): raise + if name == "performance" and isinstance( + exc, (AssertionError, pytest.fail.Exception) + ): + raise PerformanceValidationError(f"[performance] {exc}") from exc failures.append((name, str(exc))) if is_realtime_case: @@ -1651,7 +1726,9 @@ Pinned revision used by this check: {SGL_TEST_FILES_CI_DATA_REVISION} else: run_case_check( "performance", - lambda: self._validate_and_record(case, perf_record, request_index), + lambda: self._validate_and_record( + case, perf_record, request_index, diffusion_server.load_time_ms + ), ) if case.server_args.custom_validator == "mesh": diff --git a/python/sglang/multimodal_gen/test/server/test_server_utils.py b/python/sglang/multimodal_gen/test/server/test_server_utils.py index 513edd5e9..9b87e5ada 100644 --- a/python/sglang/multimodal_gen/test/server/test_server_utils.py +++ b/python/sglang/multimodal_gen/test/server/test_server_utils.py @@ -6,7 +6,9 @@ from __future__ import annotations import asyncio import base64 +import math import os +import re import shlex import subprocess import sys @@ -187,6 +189,7 @@ class ServerContext: log_dir: Path _stdout_fh: Any = field(repr=False) _log_thread: threading.Thread | None = field(default=None, repr=False) + load_time_ms: float | None = None def log_tail(self, lines: int = 200) -> str: """Return recent server output for failure diagnostics.""" @@ -422,6 +425,9 @@ class ServerManager: # regardless of log-level configuration. print(f"[server-test] Running command: {cmd_str}", flush=True) + load_started_ns = time.monotonic_ns() + load_finished_ns = None + load_ready = threading.Event() process = subprocess.Popen( command, stdout=subprocess.PIPE, @@ -437,9 +443,17 @@ class ServerManager: def _log_pipe(pipe: Any, file: Any) -> None: """Read from pipe and write to file and stdout.""" + nonlocal load_finished_ns try: with pipe: for line in iter(pipe.readline, ""): + match = re.search( + r"\[server-load\] workers_ready_monotonic_ns=(\d+)", + line, + ) + if match and load_finished_ns is None: + load_finished_ns = int(match.group(1)) + load_ready.set() sys.stdout.write(line) sys.stdout.flush() file.write(line) @@ -474,6 +488,10 @@ class ServerManager: ) try: self._wait_for_ready(process, stdout_path) + # health includes warmup; the worker marker's clock excludes it + load_ready.wait(timeout=5) + if load_finished_ns is not None: + context.load_time_ms = (load_finished_ns - load_started_ns) / 1e6 except BaseException: context.cleanup() raise @@ -713,7 +731,7 @@ class PerformanceValidator: if self.is_baseline_generation_mode: return summary - self._validate_e2e(summary) + self.validate_e2e(summary) self._validate_denoise_agg(summary) self._validate_denoise_steps(summary) self._validate_stages(summary) @@ -738,9 +756,15 @@ class PerformanceValidator: return profile_tolerance return max(profile_tolerance, override) - def _validate_e2e(self, summary: PerformanceSummary) -> None: + def validate_e2e(self, summary: PerformanceSummary) -> None: """Validate end-to-end performance.""" - assert summary.e2e_ms > 0, "E2E duration missing" + assert math.isfinite(summary.e2e_ms) and summary.e2e_ms > 0, ( + "E2E duration missing or invalid" + ) + expected = self.scenario.expected_e2e_ms + assert math.isfinite(expected) and expected > 0, ( + "E2E baseline missing or invalid" + ) self._assert_le( "E2E Latency", summary.e2e_ms, @@ -748,6 +772,24 @@ class PerformanceValidator: self._timing_tol(self.tolerances.e2e), ) + def validate_load(self, summary: PerformanceSummary) -> None: + load_ms = summary.load_time_ms + expected_load_ms = self.scenario.expected_load_ms + assert load_ms is not None and math.isfinite(load_ms) and load_ms > 0, ( + "Load duration missing or invalid" + ) + assert ( + expected_load_ms is not None + and math.isfinite(expected_load_ms) + and expected_load_ms > 0 + ), "Load baseline missing or invalid" + self._assert_le( + "Load Latency (excluding warmup)", + load_ms, + expected_load_ms, + self._timing_tol(self.tolerances.e2e), + ) + def _validate_denoise_agg(self, summary: PerformanceSummary) -> None: """Validate aggregate denoising metrics.""" assert summary.avg_denoise_ms > 0, "Denoising step timings missing" @@ -801,7 +843,7 @@ class PerformanceValidator: assert actual is not None, f"Stage {stage} timing missing" tolerance = self._timing_tol( self.tolerances.denoise_stage - if stage == "DenoisingStage" + if stage in summary.denoising_stages else self.tolerances.non_denoise_stage ) if stage.endswith("DecodingStage"): @@ -1479,9 +1521,9 @@ def get_generate_fn( size=sampling_params.output_size, seconds=video_seconds, extra_body={ - "reference_url": sampling_params.image_path, "fps": sampling_params.fps, "num_frames": sampling_params.num_frames, + **extra_body, }, ) @@ -1541,7 +1583,9 @@ def get_generate_fn( require_chunk_stats=True, ) ) - record_realtime_perf_stats(case_id, realtime_output.chunk_stats) + record_realtime_perf_stats( + case_id, realtime_output.chunk_stats, realtime_output.e2e_ms + ) record_realtime_key_frames(case_id, realtime_output.frames) fps = int(sampling_params.fps or 24) video_bytes = encode_realtime_frames_to_mp4(realtime_output.frames, fps=fps) diff --git a/python/sglang/multimodal_gen/test/server/testcase_configs.py b/python/sglang/multimodal_gen/test/server/testcase_configs.py index c3855d5a4..1840f84be 100644 --- a/python/sglang/multimodal_gen/test/server/testcase_configs.py +++ b/python/sglang/multimodal_gen/test/server/testcase_configs.py @@ -115,6 +115,7 @@ class ScenarioConfig: expected_avg_denoise_ms: float expected_median_denoise_ms: float estimated_full_test_time_s: float | None = None + expected_load_ms: float | None = None load_peak_vram_mb: float | None = None runtime_peak_vram_mb: float | None = None # Peak of the warmup calibration probe (the default workload's full shape @@ -146,6 +147,7 @@ class ScenarioConfig: expected_avg_denoise_ms=float(cfg["expected_avg_denoise_ms"]), expected_median_denoise_ms=float(cfg["expected_median_denoise_ms"]), estimated_full_test_time_s=optional_float("estimated_full_test_time_s"), + expected_load_ms=optional_float("expected_load_ms"), load_peak_vram_mb=optional_float("load_peak_vram_mb"), runtime_peak_vram_mb=optional_float("runtime_peak_vram_mb"), warmup_peak_vram_mb=optional_float("warmup_peak_vram_mb"), @@ -172,6 +174,15 @@ class BaselineConfig: with path.open("r", encoding="utf-8") as fh: data = json.load(fh) + # runner pools with the same gpu can have different host-side latency + runner_name = os.environ.get("RUNNER_NAME", "") + for prefix, overrides in data.get("runner_overrides", {}).items(): + if runner_name.startswith(prefix): + for name, metrics in overrides.items(): + data["scenarios"][name].update(metrics) + print(f"--- Performance Runner Baseline: {prefix} ---") + break + # Get tolerance profile, defaulting to 'pr_test' profile_name = "pr_test" tolerances = ToleranceConfig.load_profile( @@ -320,6 +331,7 @@ class DiffusionTestCase: run_perf_check: bool = True # Validate every repetition against the same baseline and GT. perf_repeat_requests: int = 1 + perf_warmup_requests: int = 0 run_consistency_check: bool = True run_component_accuracy_check: bool = True run_models_api_check: bool = True @@ -333,12 +345,19 @@ class DiffusionTestCase: def __post_init__(self) -> None: if self.perf_repeat_requests < 1: raise ValueError(f"{self.id}: perf_repeat_requests must be positive") + if self.perf_warmup_requests < 0: + raise ValueError(f"{self.id}: perf_warmup_requests must be non-negative") if self.sampling_params is None: object.__setattr__( self, "sampling_params", get_default_sampling_params_for_server_args(self.server_args), ) + if ( + self.perf_warmup_requests + and self.sampling_params.realtime_num_chunks is not None + ): + raise ValueError(f"{self.id}: request warmup requires non-realtime metrics") has_startup_lora = self.server_args.lora_path is not None has_dynamic_lora = self.server_args.dynamic_lora_path is not None @@ -468,6 +487,8 @@ class PerformanceSummary: frames_per_second: float | None = None total_frames: int | None = None avg_frame_time_ms: float | None = None + denoising_stages: set[str] = field(default_factory=set) + load_time_ms: float | None = None @staticmethod def from_req_perf_record( @@ -489,10 +510,13 @@ class PerformanceSummary: # convert from list to dict stage_metrics = {} + denoising_stages = set() for item in record.stages: if isinstance(item, dict) and "name" in item: val = item.get("execution_time_ms", 0.0) stage_metrics[item["name"]] = val + if item.get("is_denoising", item["name"] == "DenoisingStage"): + denoising_stages.add(item["name"]) load_peak_vram_mb = float( record.memory_snapshots.get("load_peak", {}).get("peak_reserved_mb", 0.0) @@ -528,6 +552,7 @@ class PerformanceSummary: step_metrics=step_durations, sampled_steps=sampled_steps, all_denoise_steps=per_step, + denoising_stages=denoising_stages, load_peak_vram_mb=load_peak_vram_mb, runtime_peak_vram_mb=runtime_peak_vram_mb, warmup_peak_vram_mb=warmup_peak_vram_mb, diff --git a/python/sglang/multimodal_gen/test/test_utils.py b/python/sglang/multimodal_gen/test/test_utils.py index 826ae678a..2a2284ef6 100644 --- a/python/sglang/multimodal_gen/test/test_utils.py +++ b/python/sglang/multimodal_gen/test/test_utils.py @@ -40,7 +40,7 @@ logger = init_logger(__name__) # NPU/ascend) is read from sgl-project/ci-data-diffusion, where the GT-gen workflows # publish. SGL_TEST_FILES_CI_DATA_REPO = "sgl-project/ci-data-diffusion" -SGL_TEST_FILES_CI_DATA_REVISION = "90a87cce5cdef73a9cd461f6d611ac66becef835" +SGL_TEST_FILES_CI_DATA_REVISION = "252710158cd4c74b7604808a385e93f7bced6d28" # The NPU pin is kept as a separate branch so ascend GT can be bumped independently # when it's regenerated on its own cadence. diff --git a/python/sglang/multimodal_gen/test/unit/realtime/test_realtime_consistency_harness.py b/python/sglang/multimodal_gen/test/unit/realtime/test_realtime_consistency_harness.py index a17d54e1f..48f6d85f0 100644 --- a/python/sglang/multimodal_gen/test/unit/realtime/test_realtime_consistency_harness.py +++ b/python/sglang/multimodal_gen/test/unit/realtime/test_realtime_consistency_harness.py @@ -14,6 +14,8 @@ from sglang.multimodal_gen.runtime.realtime.video import ( RAW_RGBA_DELTA_GZIP_CONTENT_TYPE, build_delta_gzip_raw_rgb_payload, ) +from sglang.multimodal_gen.runtime.utils.perf_logger import RequestPerfRecord +from sglang.multimodal_gen.test.server import test_server_common from sglang.multimodal_gen.test.server.realtime_consistency import ( build_realtime_event_payload, build_realtime_init_payload, @@ -24,16 +26,24 @@ from sglang.multimodal_gen.test.server.realtime_consistency import ( prepare_realtime_first_frame, realtime_ws_url, record_realtime_key_frames, + record_realtime_perf_stats, select_realtime_key_frames, summarize_realtime_perf_stats, validate_realtime_perf_stats, ) +from sglang.multimodal_gen.test.server.test_server_common import ( + DiffusionServerBase, + PerformanceValidationError, +) from sglang.multimodal_gen.test.server.test_server_utils import get_generate_fn from sglang.multimodal_gen.test.server.testcase_configs import ( DiffusionSamplingParams, + DiffusionServerArgs, + DiffusionTestCase, LONGLIVE2_I2V_CI_sampling_params, LONGLIVE2_T2V_CI_sampling_params, REALTIME_MODEL_sampling_params, + ScenarioConfig, ) # Request construction @@ -340,6 +350,26 @@ def test_collect_realtime_output_skips_and_records_chunk_stats(monkeypatch): np.testing.assert_array_equal(result.frames[1], second) assert [stat.chunk_index for stat in result.chunk_stats] == [0, 1] assert [stat.chunk_total_ms for stat in result.chunk_stats] == [31.0, 32.0] + assert result.e2e_ms > 0 + record_realtime_perf_stats("stream-e2e", result.chunk_stats, result.e2e_ms) + case = DiffusionTestCase( + "stream-e2e", + DiffusionServerArgs(model_path="test", modality="video"), + DiffusionSamplingParams(prompt="test"), + run_perf_check=False, + ) + runner = DiffusionServerBase() + runner._perf_results = [] + monkeypatch.setenv("SGLANG_GEN_BASELINE", "0") + monkeypatch.setitem( + test_server_common.BASELINE_CONFIG.scenarios, + case.id, + ScenarioConfig({}, {}, 1000, 0, 0, expected_load_ms=100), + ) + runner._validate_realtime_performance( + SimpleNamespace(load_time_ms=100), case, result.chunk_stats + ) + assert runner._perf_results[0]["e2e_ms"] == result.e2e_ms assert websocket.sent == [ {"type": "init", "prompt": "test"}, { @@ -350,6 +380,116 @@ def test_collect_realtime_output_skips_and_records_chunk_stats(monkeypatch): ] +@pytest.mark.parametrize("e2e_ms", [None, 0, -1, float("nan"), float("inf")]) +def test_realtime_requires_e2e_without_threshold_checks(e2e_ms): + case = DiffusionTestCase( + "missing-stream-e2e", + DiffusionServerArgs(model_path="test", modality="video"), + DiffusionSamplingParams(prompt="test"), + run_perf_check=False, + ) + runner = DiffusionServerBase() + runner._perf_results = [] + if e2e_ms is not None: + record_realtime_perf_stats(case.id, [], e2e_ms) + with pytest.raises( + PerformanceValidationError, match="E2E duration missing or invalid" + ): + runner._validate_realtime_performance( + SimpleNamespace(load_time_ms=100), case, [] + ) + + +@pytest.mark.parametrize("baseline", [None, 0, float("nan"), 1000]) +def test_realtime_e2e_guard_without_chunk_thresholds(monkeypatch, baseline): + monkeypatch.setenv("SGLANG_GEN_BASELINE", "0") + monkeypatch.setattr(test_server_common.current_platform, "is_hip", lambda: False) + case = DiffusionTestCase( + "stream-e2e-threshold", + DiffusionServerArgs(model_path="test", modality="video"), + DiffusionSamplingParams(prompt="test"), + run_perf_check=False, + ) + if baseline is None: + monkeypatch.delitem( + test_server_common.BASELINE_CONFIG.scenarios, case.id, raising=False + ) + error, message = pytest.fail.Exception, "not found" + else: + monkeypatch.setitem( + test_server_common.BASELINE_CONFIG.scenarios, + case.id, + ScenarioConfig({}, {}, baseline, 0, 0), + ) + error = AssertionError + message = ( + "E2E Latency" if baseline == 1000 else "E2E baseline missing or invalid" + ) + runner = DiffusionServerBase() + runner._perf_results = [] + record_realtime_perf_stats(case.id, [], 2000) + with pytest.raises(error, match=message): + runner._validate_realtime_performance( + SimpleNamespace(load_time_ms=100), case, [] + ) + assert runner._perf_results[0]["e2e_ms"] == 2000 + + +@pytest.mark.parametrize("peak_mb", [1000, 2000]) +def test_realtime_memory_guard_retains_session_e2e(monkeypatch, peak_mb): + case = DiffusionTestCase( + "stream-memory-e2e", + DiffusionServerArgs(model_path="test", modality="video"), + DiffusionSamplingParams(prompt="test"), + run_perf_check=True, + ) + scenario = ScenarioConfig( + {}, + {}, + 2000, + 0, + 0, + load_peak_vram_mb=1000, + runtime_peak_vram_mb=1000, + expected_load_ms=100, + ) + monkeypatch.setitem(test_server_common.BASELINE_CONFIG.scenarios, case.id, scenario) + monkeypatch.setattr(test_server_common.current_platform, "is_cuda", lambda: True) + monkeypatch.setattr(test_server_common.current_platform, "is_hip", lambda: False) + monkeypatch.setenv("SGLANG_GEN_BASELINE", "0") + record = RequestPerfRecord( + request_id="last-chunk", + commit_hash="test", + tag="test", + stages=[], + steps=[], + total_duration_ms=20, + memory_snapshots={ + "load_peak": {"peak_reserved_mb": 1000}, + "runtime_peak": {"peak_reserved_mb": peak_mb}, + }, + ) + monkeypatch.setattr( + test_server_common, "wait_for_req_perf_record", lambda *a, **k: record + ) + stats = [ + parse_realtime_chunk_stats( + msgspec.msgpack.decode(_packed_realtime_chunk_stats(0)) + ) + ] + record_realtime_perf_stats(case.id, stats, 2000) + runner = DiffusionServerBase() + runner._perf_results = [] + ctx = SimpleNamespace(perf_log_path="unused", load_time_ms=100) + if peak_mb > 1000: + with pytest.raises(AssertionError, match="Runtime Peak VRAM"): + runner._validate_realtime_performance(ctx, case, stats) + else: + runner._validate_realtime_performance(ctx, case, stats) + assert runner._perf_results[0]["e2e_ms"] == 2000 + assert runner._perf_results[0]["runtime_peak_vram_mb"] == peak_mb + + def test_collect_realtime_output_accepts_combined_frame_batch(monkeypatch): frame = np.arange(12, dtype=np.uint8).reshape(2, 2, 3) websocket = _FakeRealtimeWebSocket( diff --git a/python/sglang/multimodal_gen/test/unit/test_cfg_parallel_warmup.py b/python/sglang/multimodal_gen/test/unit/test_cfg_parallel_warmup.py index 68cc59db4..717176a90 100644 --- a/python/sglang/multimodal_gen/test/unit/test_cfg_parallel_warmup.py +++ b/python/sglang/multimodal_gen/test/unit/test_cfg_parallel_warmup.py @@ -27,9 +27,13 @@ from sglang.multimodal_gen.configs.pipeline_configs.flux_finetuned import ( from sglang.multimodal_gen.configs.pipeline_configs.longlive2 import ( LongLive2T2VConfig, ) +from sglang.multimodal_gen.configs.pipeline_configs.ltx_2_5 import LTX25PipelineConfig +from sglang.multimodal_gen.configs.pipeline_configs.sana_wm import SanaWMPipelineConfig from sglang.multimodal_gen.configs.sample.longlive2 import LongLive2SamplingParams +from sglang.multimodal_gen.configs.sample.ltx_2_5 import LTX25SamplingParams from sglang.multimodal_gen.configs.sample.minimax_h3 import MiniMaxH3SamplingParams from sglang.multimodal_gen.configs.sample.sampling_params import SamplingParams +from sglang.multimodal_gen.configs.sample.sana_wm import SanaWMSamplingParams from sglang.multimodal_gen.runtime.entrypoints.control_requests import ( SetLoraReq, UnmergeLoraWeightsReq, @@ -60,6 +64,8 @@ from sglang.multimodal_gen.runtime.warmup_request_builder import ( should_include_warmup_image, supports_synthetic_warmup, ) +from sglang.multimodal_gen.test.server.gpu_cases import ONE_GPU_CASES, TWO_GPU_CASES +from sglang.multimodal_gen.test.server.testcase_configs import _get_extra_arg_value def _make_bare_scheduler(enable_cfg_parallel: bool) -> Scheduler: @@ -653,6 +659,99 @@ class TestWarmupReqCfgParallel(unittest.TestCase): self.assertEqual(num_frames, 17) pipeline_config.adjust_num_frames.assert_called_once_with(17) + def test_server_warmup_preserves_explicit_frames_without_cuda_graphs(self): + server_args = SimpleNamespace( + pipeline_config=LTX25PipelineConfig(), + enable_breakable_cuda_graph=False, + pipeline_class_name="LTX2Pipeline", + num_gpus=2, + warmup_num_frames=49, + ) + + num_frames = _resolve_warmup_num_frames( + server_args, LTX25SamplingParams(), server_based_warmup=True + ) + + self.assertEqual(num_frames, 57) + + def test_sana_ci_warmup_matches_formal_shape(self): + case = next(case for case in ONE_GPU_CASES if case.id == "sana_wm_ti2v") + resolution = _get_extra_arg_value( + case.server_args.extras, "--warmup-resolutions" + ) + server_args = SimpleNamespace( + pipeline_config=SanaWMPipelineConfig(), + pipeline_class_name=None, + model_path=case.server_args.model_path, + model_id=None, + backend="sglang", + num_gpus=1, + warmup_steps=1, + warmup_num_frames=None, + warmup_sampling_params=None, + enable_breakable_cuda_graph=False, + enable_torch_compile=False, + enable_cfg_parallel=False, + ) + with patch.object( + SamplingParams, "from_pretrained", return_value=SanaWMSamplingParams() + ): + reqs = build_warmup_reqs( + server_args, + warmup_resolutions=[resolution], + warmup_input_path="synthetic-warmup.png", + server_based_warmup=True, + ) + self.assertEqual(resolution, case.sampling_params.output_size) + self.assertEqual(len(reqs), 1) + self.assertEqual((reqs[0].width, reqs[0].height), (384, 640)) + self.assertEqual(reqs[0].num_frames, case.sampling_params.num_frames) + + def test_ltx25_ci_warmup_matches_formal_decoder_and_shape(self): + case = next( + case + for case in TWO_GPU_CASES + if case.id == "ltx_2_5_diffusion_decoder_2gpus" + ) + extras = case.server_args.extras + server_args = SimpleNamespace( + pipeline_config=LTX25PipelineConfig(), + pipeline_class_name=None, + model_path=case.server_args.model_path, + model_id=None, + backend="sglang", + num_gpus=2, + warmup_steps=1, + warmup_num_frames=int(_get_extra_arg_value(extras, "--warmup-num-frames")), + warmup_sampling_params=_get_extra_arg_value( + extras, "--warmup-sampling-params" + ), + enable_breakable_cuda_graph=False, + enable_torch_compile=False, + enable_cfg_parallel=False, + ) + resolution = _get_extra_arg_value(extras, "--warmup-resolutions") + with patch.object( + SamplingParams, "from_pretrained", return_value=LTX25SamplingParams() + ): + reqs = build_warmup_reqs( + server_args, + warmup_resolutions=[resolution], + warmup_input_path="synthetic-warmup.png", + server_based_warmup=True, + ) + + self.assertEqual(len(reqs), 1) + req = reqs[0] + self.assertEqual(resolution, case.sampling_params.output_size) + self.assertEqual(server_args.warmup_num_frames, case.sampling_params.num_frames) + self.assertEqual((req.width, req.height, req.num_frames), (768, 448, 57)) + self.assertEqual( + req.sampling_params.use_diffusion_decoder, + case.sampling_params.extras["use_diffusion_decoder"], + ) + self.assertEqual(req.num_inference_steps, 2) + def test_server_based_warmup_uses_video_supported_resolution_budget(self): server_args = MagicMock() server_args.warmup_steps = 1 diff --git a/python/sglang/multimodal_gen/test/unit/test_hunyuan3d_native_texture_models.py b/python/sglang/multimodal_gen/test/unit/test_hunyuan3d_native_texture_models.py index cbfec5d64..264ff798d 100644 --- a/python/sglang/multimodal_gen/test/unit/test_hunyuan3d_native_texture_models.py +++ b/python/sglang/multimodal_gen/test/unit/test_hunyuan3d_native_texture_models.py @@ -2,11 +2,14 @@ import unittest from types import SimpleNamespace +from unittest.mock import Mock, patch +import numpy as np import torch from diffusers import AutoencoderKL as DiffusersAutoencoderKL from diffusers import LCMScheduler, UNet2DConditionModel +from sglang.kernels.ops.diffusion.ext import mesh_processor from sglang.multimodal_gen.configs.models.vaes.stable_diffusion import ( StableDiffusionVAEConfig, ) @@ -185,9 +188,19 @@ class TestHunyuan3DWarmupOutput(unittest.TestCase): def test_paint_postprocess_skips_export_during_warmup(self): stage = Hunyuan3DPaintPostprocessStage(Hunyuan3D2PipelineConfig()) - - output = stage.forward(self._batch(), SimpleNamespace()) - + kernel = Mock() + with ( + patch.object(mesh_processor, "_mesh_processor_kernel", None), + patch.object( + mesh_processor, "load_extension_with_recovery", return_value=kernel + ) as build, + ): + output = stage.forward(self._batch(), SimpleNamespace()) + build.assert_called_once() + array = np.zeros((1, 3), dtype=np.float32) + mesh_processor.meshVerticeInpaint(array, array, array, array, array, array) + build.assert_called_once() + kernel.meshVerticeInpaint.assert_called_once() self.assertEqual(output.output_file_paths, []) diff --git a/python/sglang/multimodal_gen/test/unit/test_load_inclusive_e2e.py b/python/sglang/multimodal_gen/test/unit/test_load_inclusive_e2e.py new file mode 100644 index 000000000..3d2d3b00a --- /dev/null +++ b/python/sglang/multimodal_gen/test/unit/test_load_inclusive_e2e.py @@ -0,0 +1,194 @@ +import io +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from sglang.multimodal_gen.configs.pipeline_configs.base import PipelineConfig +from sglang.multimodal_gen.runtime import launch_server as launcher +from sglang.multimodal_gen.runtime.server_args import ServerArgs +from sglang.multimodal_gen.runtime.utils.perf_logger import RequestPerfRecord +from sglang.multimodal_gen.test.scripts import gen_perf_baselines +from sglang.multimodal_gen.test.server import test_server_utils as utils +from sglang.multimodal_gen.test.server.test_server_utils import PerformanceValidator +from sglang.multimodal_gen.test.server.testcase_configs import ( + DiffusionSamplingParams, + DiffusionServerArgs, + DiffusionTestCase, + PerformanceSummary, + ScenarioConfig, + ToleranceConfig, +) + + +@pytest.fixture +def validator(monkeypatch): + monkeypatch.setenv("SGLANG_GEN_BASELINE", "0") + scenario = ScenarioConfig.from_dict( + { + "stages_ms": {}, + "denoise_step_ms": {}, + "expected_e2e_ms": 1000, + "expected_avg_denoise_ms": 0, + "expected_median_denoise_ms": 0, + "expected_load_ms": 4000, + } + ) + return PerformanceValidator(scenario, ToleranceConfig(0.25, 0, 0, 0, 0), []) + + +def test_slow_loading_fails_even_when_inference_passes(validator): + summary = PerformanceSummary(1000, 0, 0, {}, [], {}, {}, load_time_ms=6000) + validator.validate_e2e(summary) + with pytest.raises(AssertionError, match="Load Latency"): + validator.validate_load(summary) + + +def test_fast_loading_cannot_hide_inference_regression(validator): + summary = PerformanceSummary(2000, 0, 0, {}, [], {}, {}, load_time_ms=1000) + validator.validate_load(summary) + with pytest.raises(AssertionError, match="E2E Latency"): + validator.validate_e2e(summary) + + +@pytest.mark.parametrize("load_time_ms", [None, 0, float("nan"), 1234.5]) +def test_baseline_script_preserves_required_load_measurement(monkeypatch, load_time_ms): + case = DiffusionTestCase( + "load-baseline", + DiffusionServerArgs("test", modality="image"), + DiffusionSamplingParams(prompt="test"), + ) + context = SimpleNamespace( + port=1234, load_time_ms=load_time_ms, perf_log_path="unused", cleanup=Mock() + ) + monkeypatch.setattr( + gen_perf_baselines, + "ServerManager", + Mock(return_value=Mock(start=lambda: context)), + ) + monkeypatch.setattr(gen_perf_baselines, "get_dynamic_server_port", lambda: 1234) + monkeypatch.setattr(gen_perf_baselines, "_build_server_extra_args", lambda case: "") + monkeypatch.setattr(gen_perf_baselines, "_openai_client", Mock()) + monkeypatch.setattr( + gen_perf_baselines, + "get_generate_fn", + lambda **kwargs: lambda *args: ("request", b"output"), + ) + monkeypatch.setattr(gen_perf_baselines.current_platform, "is_cuda", lambda: False) + record = RequestPerfRecord( + request_id="request", + commit_hash="test", + tag="test", + stages=[], + steps=[], + total_duration_ms=1000, + ) + monkeypatch.setattr( + gen_perf_baselines, "wait_for_req_perf_record", lambda *args, **kwargs: record + ) + if load_time_ms == 1234.5: + scenario = ScenarioConfig.from_dict(gen_perf_baselines._run_case(case)) + assert scenario.expected_load_ms == 1234.5 + assert scenario.expected_e2e_ms == 1000 + else: + with pytest.raises(ValueError, match="load duration missing or invalid"): + gen_perf_baselines._run_case(case) + context.cleanup.assert_called_once() + + +@pytest.mark.parametrize("duration", [None, 0, -1, float("nan"), float("inf")]) +def test_missing_or_invalid_load_duration_fails(validator, duration): + summary = PerformanceSummary(1000, 0, 0, {}, [], {}, {}, load_time_ms=duration) + with pytest.raises(AssertionError, match="Load duration missing or invalid"): + validator.validate_load(summary) + + +@pytest.mark.parametrize("duration", [None, 0, -1, float("nan"), float("inf")]) +def test_missing_or_invalid_load_baseline_fails(validator, duration): + validator.scenario.expected_load_ms = duration + summary = PerformanceSummary(1000, 0, 0, {}, [], {}, {}, load_time_ms=4000) + with pytest.raises(AssertionError, match="Load baseline missing or invalid"): + validator.validate_load(summary) + + +def test_fast_inference_cannot_hide_loading_regression(validator): + summary = PerformanceSummary(1, 0, 0, {}, [], {}, {}, load_time_ms=5500) + validator.validate_e2e(summary) + with pytest.raises(AssertionError, match="Load Latency"): + validator.validate_load(summary) + + +def test_repeated_requests_use_same_load_measurement(validator): + for inference_ms in (1000, 900, 950): + summary = PerformanceSummary( + inference_ms, 0, 0, {}, [], {}, {}, load_time_ms=4000 + ) + validator.validate_e2e(summary) + validator.validate_load(summary) + + +@pytest.mark.parametrize("workers", [1, 2]) +@pytest.mark.parametrize("warmup_seconds", [0, 120]) +def test_server_load_clock_excludes_warmup( + monkeypatch, tmp_path, workers, warmup_seconds, validator +): + clock = [1_000_000_000] + output = io.StringIO() + monkeypatch.setattr(utils.time, "monotonic_ns", lambda: clock[0]) + monkeypatch.setattr(utils.current_platform, "is_hip", lambda: False) + monkeypatch.setattr(utils.tempfile, "gettempdir", lambda: str(tmp_path)) + monkeypatch.setattr( + utils, "prepare_perf_log", lambda: (tmp_path, tmp_path / "perf.jsonl") + ) + monkeypatch.setattr(launcher, "configure_logger", Mock()) + monkeypatch.setattr(launcher, "logger", Mock()) + launcher.logger.info.side_effect = lambda message, *args: output.write( + (message % args) + "\n" + ) + + def ready(): + clock[0] += 1_000_000_000 + return {"status": "ready"} + + worker_context = Mock() + worker_context.Pipe.side_effect = lambda **kwargs: (Mock(recv=ready), Mock()) + monkeypatch.setattr(launcher.mp, "get_context", Mock(return_value=worker_context)) + monkeypatch.setattr(launcher, "shutdown_scheduler_processes", Mock()) + + def warmup(args): + clock[0] += warmup_seconds * 1_000_000_000 + + monkeypatch.setattr(launcher, "launch_http_server_only", warmup) + monkeypatch.setattr(ServerArgs, "__post_init__", lambda self: None) + args = ServerArgs( + model_path="test", + num_gpus=workers, + nnodes=1, + node_rank=0, + master_port=1234, + webui=False, + pipeline_config=PipelineConfig(), + ) + + def spawn(*unused_args, **unused_kwargs): + launcher.launch_server(args) + return SimpleNamespace(pid=1234, stdout=io.StringIO(output.getvalue())) + + monkeypatch.setattr(utils.subprocess, "Popen", spawn) + manager = utils.ServerManager("test", 1234) + monkeypatch.setattr(manager, "_wait_for_ready", Mock()) + context = manager.start() + launcher.mp.get_context.assert_called_once_with("spawn") + assert worker_context.Process.call_count == workers + for call in worker_context.Process.call_args_list: + restored_args = call.kwargs["args"][0].server_args.materialize() + assert isinstance(restored_args, ServerArgs) + assert restored_args.num_gpus == workers + context._log_thread.join(timeout=5) + assert not context._log_thread.is_alive() + assert context.load_time_ms == workers * 1000 + summary = PerformanceSummary( + 1000, 0, 0, {}, [], {}, {}, load_time_ms=context.load_time_ms + ) + validator.validate_e2e(summary) + validator.validate_load(summary) diff --git a/python/sglang/multimodal_gen/test/unit/test_performance_failure_policy.py b/python/sglang/multimodal_gen/test/unit/test_performance_failure_policy.py new file mode 100644 index 000000000..ea2941a9c --- /dev/null +++ b/python/sglang/multimodal_gen/test/unit/test_performance_failure_policy.py @@ -0,0 +1,268 @@ +import os +import subprocess +import sys +import textwrap +import time + +import pytest + +from sglang.multimodal_gen.runtime.utils.perf_logger import RequestPerfRecord +from sglang.multimodal_gen.test.runner.pytest_runner import ( + _estimate_failed_test_time, + _is_retryable_failure, + run_pytest, +) +from sglang.multimodal_gen.test.server import test_server_common as common +from sglang.multimodal_gen.test.server.testcase_configs import ( + DiffusionSamplingParams, + DiffusionServerArgs, + DiffusionTestCase, + ScenarioConfig, +) + + +@pytest.mark.parametrize("generate_baseline", [False, True]) +def test_e2e_only_does_not_require_stage_metrics(monkeypatch, generate_baseline): + monkeypatch.setenv("SGLANG_GEN_BASELINE", str(int(generate_baseline))) + case = DiffusionTestCase( + "e2e_only", + DiffusionServerArgs(model_path="test", modality="image"), + DiffusionSamplingParams(prompt="test"), + run_perf_check=False, + ) + scenario = ScenarioConfig({}, {}, 1000, 0, 0, expected_load_ms=100) + monkeypatch.setitem(common.BASELINE_CONFIG.scenarios, case.id, scenario) + monkeypatch.setattr(common, "_PENDING_BASELINE_DUMPS", {}) + server = common.DiffusionServerBase() + server._perf_results = [] + record = RequestPerfRecord( + request_id="guard", + commit_hash="test", + tag="guard", + stages=[], + steps=[], + total_duration_ms=2000 if generate_baseline else 1000, + ) + server._validate_and_record(case, record, load_time_ms=100) + assert len(server._perf_results) == 1 + assert bool(common._PENDING_BASELINE_DUMPS) == generate_baseline + + +@pytest.mark.parametrize( + "output", + [ + "multimodal_gen/test/server/test_server_utils.py: AssertionError", + "Consistency check failed for example\nTimeoutError", + "[performance] Validation failed\nConsistency check failed for example", + ], +) +def test_validation_failures_are_not_retryable(output): + assert not _is_retryable_failure(output) + + +@pytest.mark.parametrize( + "output", + [ + "[performance] Validation failed for 'E2E Latency'", + "[performance] Validation failed for 'Load Latency (excluding warmup)'", + "[performance] Validation failed for 'Average Denoise Step'\nTimeoutError", + "[performance] E2E missing or invalid\nCUDA out of memory", + ], +) +def test_performance_failures_are_retryable(output): + assert _is_retryable_failure(output) + + +@pytest.mark.parametrize( + "output", ["TimeoutError", "SafetensorError", "CUDA out of memory"] +) +def test_infrastructure_failure_policy_is_unchanged(output): + assert _is_retryable_failure(output) + + +@pytest.mark.parametrize("with_deadline", [False, True]) +def test_performance_retry_recovers_only_failed_items( + tmp_path, monkeypatch, with_deadline +): + monkeypatch.setenv("PYTEST_DISABLE_PLUGIN_AUTOLOAD", "1") + if with_deadline: + monkeypatch.setenv("SGLANG_DIFFUSION_RETRY_DEADLINE", str(time.time() + 600)) + else: + monkeypatch.delenv("SGLANG_DIFFUSION_RETRY_DEADLINE", raising=False) + test_file = tmp_path / "test_retry.py" + test_file.write_text( + "from pathlib import Path\n" + "def test_slow():\n" + " marker = Path(__file__).with_suffix('.attempt')\n" + " if not marker.exists():\n" + " marker.touch()\n" + " assert False, '[performance] Validation failed for E2E Latency'\n" + "def test_fast():\n" + " marker = Path(__file__).with_suffix('.passed')\n" + " assert not marker.exists(), 'passing case must not rerun'\n" + " marker.touch()\n" + ) + code, _, _ = run_pytest([str(test_file)]) + assert code == 0 + assert test_file.with_suffix(".attempt").exists() + assert test_file.with_suffix(".passed").exists() + + +def test_retry_budget_preserves_failure_and_report(tmp_path, monkeypatch, capfd): + monkeypatch.setenv("PYTEST_DISABLE_PLUGIN_AUTOLOAD", "1") + monkeypatch.setenv("SGLANG_DIFFUSION_RETRY_DEADLINE", str(time.time() - 1)) + test_file = tmp_path / "test_budget.py" + test_file.write_text( + "import pytest\n" + "@pytest.mark.parametrize('case_id', ['slow_case'])\n" + "def test_slow(case_id):\n" + " assert False, '[performance] Validation failed for E2E Latency'\n" + ) + report = tmp_path / "junit.xml" + code, executed, results = run_pytest([str(test_file)], junit_xml_path=str(report)) + output = capfd.readouterr().out + assert code == 1 + assert executed == ["slow_case"] + assert results == {"slow_case": "fail"} + assert output.count("Starting pytest attempt") == 1 + assert "Retry budget exhausted" in output + assert "Pytest Tail Summary" in output + + +def test_retry_estimate_excludes_successful_cases(tmp_path): + report = tmp_path / "junit.xml" + report.write_text( + "" + '' + '' + '' + "" + ) + assert _estimate_failed_test_time(str(report), 130) == 30 + assert _estimate_failed_test_time(None, 130) == 130 + + +@pytest.mark.parametrize( + "problem", + [ + "regression", + "missing_baseline", + "missing_record", + "missing_e2e", + "missing_log", + "e2e_only_regression", + "e2e_only_missing_baseline", + "e2e_only_zero_baseline", + "e2e_only_nan_baseline", + ], +) +def test_performance_failure_survives_real_pytest_runner(tmp_path, problem): + # exercise the validator, request loop, pytest output and retry classifier together + test_file = tmp_path / "test_guard.py" + test_file.write_text( + textwrap.dedent( + """ + import pytest + from types import SimpleNamespace + from sglang.multimodal_gen.runtime.utils.perf_logger import RequestPerfRecord + from sglang.multimodal_gen.test.server import test_server_common as common + from sglang.multimodal_gen.test.test_utils import wait_for_req_perf_record + from sglang.multimodal_gen.test.server.testcase_configs import ( + DiffusionSamplingParams, DiffusionServerArgs, DiffusionTestCase, ScenarioConfig, + ) + + @pytest.mark.parametrize("case_id", ["threshold_guard"]) + def test_guard(case_id, monkeypatch, tmp_path): + server = common.DiffusionServerBase() + server._perf_results = [] + case = DiffusionTestCase( + case_id, + DiffusionServerArgs(model_path="test", modality="image", lora_path="test-lora"), + DiffusionSamplingParams(prompt="test"), + run_lora_basic_api_check=True, perf_repeat_requests=2, + run_consistency_check=False, run_models_api_check=False, + run_perf_check=not PROBLEM.startswith("e2e_only_") and PROBLEM not in ("missing_record", "missing_e2e", "missing_log"), + ) + scenario = ScenarioConfig({}, {}, 1000, 100, 100, expected_load_ms=100) + if PROBLEM == "e2e_only_zero_baseline": + scenario.expected_e2e_ms = 0 + if PROBLEM == "e2e_only_nan_baseline": + scenario.expected_e2e_ms = float("nan") + if PROBLEM in ("missing_baseline", "e2e_only_missing_baseline"): + monkeypatch.delitem(common.BASELINE_CONFIG.scenarios, case_id, raising=False) + else: + monkeypatch.setitem(common.BASELINE_CONFIG.scenarios, case_id, scenario) + monkeypatch.setattr(common.current_platform, "is_cuda", lambda: False) + monkeypatch.setattr(common.current_platform, "is_hip", lambda: False) + monkeypatch.setattr(common, "get_generate_fn", lambda **kwargs: None) + requests = [] + lora_checks = [] + def collect(*args, **kwargs): + requests.append(1) + if PROBLEM == "missing_record": + return None, b"" + return RequestPerfRecord( + request_id="guard", commit_hash="test", tag="guard", + stages=[], steps=[100], + total_duration_ms=None if PROBLEM == "missing_e2e" else 2000, + ), b"" + context = SimpleNamespace(load_time_ms=100) + if PROBLEM == "missing_log": + log_path = tmp_path / "empty-perf.jsonl" + log_path.write_text("") + context = SimpleNamespace(perf_log_path=log_path, load_time_ms=100) + monkeypatch.setattr(server, "_client", lambda ctx: None) + def generate(*args): + requests.append(1) + return "guard", b"" + monkeypatch.setattr(server, "_run_generation_with_server_watchdog", generate) + monkeypatch.setattr( + common, "wait_for_req_perf_record", + lambda rid, path, timeout: wait_for_req_perf_record(rid, path, timeout=0.01), + ) + else: + monkeypatch.setattr(server, "run_and_collect", collect) + monkeypatch.setattr( + server, "_test_lora_api_functionality", + lambda *args: lora_checks.append(1), + ) + try: + server._test_diffusion_generation_impl(case, context) + finally: + print(f"GUARD_REQUESTS={len(requests)} LORA_CHECKS={len(lora_checks)}") + print(f"RETAINED_METRICS={len(server._perf_results)}") + + def test_unrelated_timeout(): + raise TimeoutError("independent infrastructure failure") + """ + ).replace("PROBLEM", repr(problem)) + ) + report = tmp_path / "junit.xml" + env = os.environ.copy() + env.update( + PYTEST_DISABLE_PLUGIN_AUTOLOAD="1", + SGLANG_GEN_BASELINE="0", + SGLANG_GEN_GT="0", + ) + command = ( + "from sglang.multimodal_gen.test.runner.pytest_runner import run_pytest; " + f"result = run_pytest([{str(test_file)!r}], junit_xml_path={str(report)!r}); " + "print('GUARD_RESULT', result); raise SystemExit(result[0])" + ) + result = subprocess.run( + [sys.executable, "-c", command], + cwd=tmp_path, + env=env, + capture_output=True, + text=True, + timeout=600, + ) + output = result.stdout + result.stderr + assert result.returncode == 1, output + assert "[performance]" in output, output + assert "GUARD_REQUESTS=1 LORA_CHECKS=0" in output, output + retained = 0 if problem in {"missing_record", "missing_e2e", "missing_log"} else 1 + assert f"RETAINED_METRICS={retained}" in output, output + assert output.count("Starting pytest attempt") == 7, output + assert "Max retry exceeded (6)" in output, output + assert "'threshold_guard': 'fail'" in output, output diff --git a/python/sglang/multimodal_gen/test/unit/test_performance_metrics.py b/python/sglang/multimodal_gen/test/unit/test_performance_metrics.py index 76b2f6410..299654691 100644 --- a/python/sglang/multimodal_gen/test/unit/test_performance_metrics.py +++ b/python/sglang/multimodal_gen/test/unit/test_performance_metrics.py @@ -7,14 +7,30 @@ import torch import sglang.multimodal_gen.runtime.managers.gpu_worker as gpu_worker_module import sglang.multimodal_gen.runtime.managers.memory_managers.component_manager as component_manager_module +import sglang.multimodal_gen.runtime.utils.perf_logger as perf_logger_module +from sglang.multimodal_gen.runtime.disaggregation.orchestrator import ( + _deserialize_request_metrics, +) from sglang.multimodal_gen.runtime.managers.gpu_worker import GPUWorker from sglang.multimodal_gen.runtime.managers.memory_managers.component_manager import ( WarmupPhasePeak, ) from sglang.multimodal_gen.runtime.pipelines_core.schedule_batch import OutputBatch +from sglang.multimodal_gen.runtime.pipelines_core.stages.base import PipelineStage +from sglang.multimodal_gen.runtime.pipelines_core.stages.denoising import DenoisingStage +from sglang.multimodal_gen.runtime.pipelines_core.stages.model_specific_stages.ltx_2.denoising_av import ( + LTX2RefinementStage, +) +from sglang.multimodal_gen.runtime.pipelines_core.stages.model_specific_stages.minimax_h3.stages.denoising import ( + MiniMaxH3DenoisingStage, +) +from sglang.multimodal_gen.runtime.pipelines_core.stages.text_encoding import ( + TextEncodingStage, +) from sglang.multimodal_gen.runtime.platforms import current_platform from sglang.multimodal_gen.runtime.utils.perf_logger import ( MemorySnapshot, + PerformanceLogger, RequestMetrics, RequestPerfRecord, ) @@ -26,6 +42,7 @@ from sglang.multimodal_gen.test.server.testcase_configs import ( ScenarioConfig, ToleranceConfig, ) +from sglang.multimodal_gen.test.test_utils import read_perf_logs @pytest.fixture(autouse=True) @@ -71,6 +88,72 @@ def test_request_metrics_attributes_steps_and_iterations_to_active_stage(): } +@pytest.mark.parametrize("roundtrip", [False, True]) +@pytest.mark.parametrize( + "stage_class,profile_name,is_denoising", + [ + (DenoisingStage, "DenoisingStage", True), + (MiniMaxH3DenoisingStage, "MiniMaxH3DenoisingStage", True), + (LTX2RefinementStage, "LTX2RefinementStage", True), + (LTX2RefinementStage, "custom_refinement", True), + (DenoisingStage, "BeforeDenoisingStage", True), + (TextEncodingStage, "BeforeDenoisingStage", False), + (TextEncodingStage, "TextEncodingStage", False), + ], +) +def test_stage_role_reaches_performance_guard( + stage_class, profile_name, is_denoising, roundtrip, monkeypatch, tmp_path +): + # skip model construction and kernels, retaining the real stage role, + # call boundary, profiler, log writer/reader and threshold validator + stage = stage_class.__new__(stage_class) + stage.server_args = SimpleNamespace( + enable_layerwise_nvtx_marker=False, comfyui_mode=False + ) + stage.set_profile_stage_name(profile_name) + monkeypatch.setattr(stage, "forward", lambda batch, args: batch) + monkeypatch.setattr( + stage, "verify_input", PipelineStage.verify_input.__get__(stage) + ) + monkeypatch.setattr( + stage, "verify_output", PipelineStage.verify_output.__get__(stage) + ) + monkeypatch.setattr(current_platform, "get_available_gpu_memory", lambda **_: 100) + monkeypatch.setattr(current_platform, "is_hip", lambda: False) + monkeypatch.setenv("SGLANG_PERF_LOG_DIR", str(tmp_path)) + monkeypatch.setattr(perf_logger_module, "get_is_main_process", lambda: True) + monkeypatch.setattr(perf_logger_module, "get_git_commit_hash", lambda: "test") + metrics = RequestMetrics("stage-role") + batch = SimpleNamespace(is_warmup=False, metrics=metrics, perf_dump_path="metrics") + with patch.object(perf_logger_module.time, "perf_counter", side_effect=[10, 11.5]): + assert stage(batch, stage.server_args) is batch + metrics.total_duration_ms = 1500 + if roundtrip: + metrics = _deserialize_request_metrics( + json.loads(json.dumps(metrics.to_dict())) + ) + PerformanceLogger.log_request_summary(metrics) + (record,) = read_perf_logs(tmp_path / "performance.log") + assert record.stages == [ + { + "name": profile_name, + "execution_time_ms": 1500.0, + "is_denoising": is_denoising, + } + ] + validator = PerformanceValidator( + ScenarioConfig({profile_name: 1000}, {}, 1500, 1, 1), + ToleranceConfig(0.25, 0.25, 0.8, 0.3, 0.2), + (), + ) + summary = validator.collect_metrics(record) + if is_denoising: + with pytest.raises(AssertionError, match="Stage '"): + validator._validate_stages(summary) + else: + validator._validate_stages(summary) + + def test_performance_summary_separates_load_and_runtime_peaks(): summary = PerformanceSummary.from_req_perf_record( _perf_record( diff --git a/python/sglang/multimodal_gen/test/unit/test_runner_perf_baselines.py b/python/sglang/multimodal_gen/test/unit/test_runner_perf_baselines.py new file mode 100644 index 000000000..abae60ca5 --- /dev/null +++ b/python/sglang/multimodal_gen/test/unit/test_runner_perf_baselines.py @@ -0,0 +1,78 @@ +from dataclasses import replace + +import pytest + +from sglang.multimodal_gen.test.server.test_server_utils import PerformanceValidator +from sglang.multimodal_gen.test.server.testcase_configs import ( + BaselineConfig, + PerformanceSummary, + get_perf_baseline_path, +) + + +@pytest.mark.parametrize( + "runner", + ["b200-fin03-4-4567", "b200-cirrascale2", "b200-cirrascale4-0123", "unknown", ""], +) +def test_default_runner_baseline(monkeypatch, runner): + monkeypatch.setenv("RUNNER_NAME", runner) + config = BaselineConfig.load(get_perf_baseline_path("b200")) + assert config.scenarios["flux1_modelopt_nvfp4_t2i"].expected_e2e_ms == 836.71 + assert ( + config.scenarios["qwen_image_2512_modelopt_nvfp4_t2i"].expected_e2e_ms + == 9650.06 + ) + + +@pytest.mark.parametrize( + "runner,flux,qwen", + [ + ("b200-di01-4567", 1334.16, 16126.87), + ("b200-cirrascale1-0123", 1574.32, 17742.04), + ("b200-cirrascale3-0123", 1470.24, 17894.49), + ("b200-cirrascale3-4567", 1471.19, 17346.65), + ], +) +def test_runner_override_preserves_other_metrics(monkeypatch, runner, flux, qwen): + monkeypatch.delenv("RUNNER_NAME", raising=False) + default = BaselineConfig.load(get_perf_baseline_path("b200")) + h100_default = BaselineConfig.load(get_perf_baseline_path("h100")) + monkeypatch.setenv("RUNNER_NAME", runner) + pool = BaselineConfig.load(get_perf_baseline_path("b200")) + expected = { + "flux1_modelopt_nvfp4_t2i": flux, + "qwen_image_2512_modelopt_nvfp4_t2i": qwen, + } + for name, scenario in default.scenarios.items(): + assert pool.scenarios[name] == replace( + scenario, expected_e2e_ms=expected.get(name, scenario.expected_e2e_ms) + ) + assert pool.tolerances == default.tolerances + assert pool.step_fractions == default.step_fractions + assert BaselineConfig.load(get_perf_baseline_path("h100")) == h100_default + + +@pytest.mark.parametrize( + "runner", + [ + "b200-di01-4567", + "b200-fin03-4-4567", + "b200-cirrascale1-0123", + "b200-cirrascale3-0123", + "b200-cirrascale3-4567", + ], +) +def test_runner_baseline_enforces_e2e_boundary(monkeypatch, runner): + monkeypatch.setenv("RUNNER_NAME", runner) + monkeypatch.setenv("SGLANG_GEN_BASELINE", "0") + monkeypatch.delenv("SGLANG_E2E_TOLERANCE", raising=False) + config = BaselineConfig.load(get_perf_baseline_path("b200")) + for name in ("flux1_modelopt_nvfp4_t2i", "qwen_image_2512_modelopt_nvfp4_t2i"): + scenario = config.scenarios[name] + validator = PerformanceValidator( + scenario, config.tolerances, config.step_fractions + ) + limit = scenario.expected_e2e_ms * (1 + config.tolerances.e2e) + validator.validate_e2e(PerformanceSummary(limit - 1, 0, 0, {}, [], {}, {})) + with pytest.raises(AssertionError, match="E2E Latency"): + validator.validate_e2e(PerformanceSummary(limit + 1, 0, 0, {}, [], {}, {})) 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 index aa74cabaa..24f54accc 100644 --- a/python/sglang/multimodal_gen/test/unit/test_sequential_server_requests.py +++ b/python/sglang/multimodal_gen/test/unit/test_sequential_server_requests.py @@ -1,6 +1,7 @@ import json import os from dataclasses import replace +from types import SimpleNamespace from unittest.mock import Mock import pytest @@ -20,6 +21,88 @@ from sglang.multimodal_gen.test.server.testcase_configs import ( pytest_plugins = ["pytester"] +@pytest.mark.parametrize("load_time_ms", [None, 0, float("nan"), 1000]) +def test_load_guard_is_terminal_without_stage_checks( + harness, monkeypatch, load_time_ms +): + runner, case = harness + case = replace(case, run_perf_check=False) + monkeypatch.setattr( + runner, "run_and_collect", Mock(return_value=(_perf_record(), b"output")) + ) + with pytest.raises(test_server_common.PerformanceValidationError, match="Load"): + runner.test_diffusion_generation( + case, SimpleNamespace(load_time_ms=load_time_ms) + ) + assert runner.run_and_collect.call_count == 1 + + +@pytest.mark.parametrize("load_time_ms", [None, 0, float("nan")]) +def test_baseline_generation_requires_loading_measurement( + harness, monkeypatch, load_time_ms +): + runner, case = harness + monkeypatch.setenv("SGLANG_GEN_BASELINE", "1") + monkeypatch.setattr( + runner, "run_and_collect", Mock(return_value=(_perf_record(), b"output")) + ) + with pytest.raises(test_server_common.PerformanceValidationError, match="Load"): + runner.test_diffusion_generation( + case, SimpleNamespace(load_time_ms=load_time_ms) + ) + assert test_server_common._PENDING_BASELINE_DUMPS == {} + + +def test_request_warmup_is_separate_from_guarded_requests(harness, monkeypatch, capsys): + runner, case = harness + case = replace(case, perf_warmup_requests=1) + cold = _perf_record() + cold.total_duration_ms = 3000 + generate = Mock( + side_effect=[ + (cold, b"output"), + (_perf_record(), b"output"), + (_perf_record(), b"output"), + ] + ) + monkeypatch.setattr(runner, "run_and_collect", generate) + runner.test_diffusion_generation(case, SimpleNamespace(load_time_ms=100)) + assert generate.call_count == 3 + assert len(runner._perf_results) == 2 + assert runner._validate_consistency.call_count == 2 + assert "request warmup 1/1 e2e=3000.0000ms" in capsys.readouterr().out + + +@pytest.mark.parametrize("duration", [None, 0, float("nan"), float("inf")]) +def test_request_warmup_requires_e2e(harness, monkeypatch, duration): + runner, case = harness + case = replace(case, perf_warmup_requests=1) + record = _perf_record() + if duration is None: + record = None + else: + record.total_duration_ms = duration + generate = Mock(return_value=(record, b"output")) + monkeypatch.setattr(runner, "run_and_collect", generate) + with pytest.raises( + test_server_common.PerformanceValidationError, match="warmup.*E2E" + ): + runner.test_diffusion_generation(case, SimpleNamespace(load_time_ms=100)) + assert generate.call_count == 1 + + +def test_request_after_warmup_still_enforces_e2e(harness, monkeypatch): + runner, case = harness + case = replace(case, perf_warmup_requests=1, run_perf_check=False) + slow = _perf_record() + slow.total_duration_ms = 3000 + generate = Mock(side_effect=[(_perf_record(), b"output"), (slow, b"output")]) + monkeypatch.setattr(runner, "run_and_collect", generate) + with pytest.raises(test_server_common.PerformanceValidationError, match="E2E"): + runner.test_diffusion_generation(case, SimpleNamespace(load_time_ms=100)) + assert generate.call_count == 2 + + def _perf_record(): return RequestPerfRecord( request_id="request", @@ -48,6 +131,7 @@ def harness(monkeypatch): expected_avg_denoise_ms=5, expected_median_denoise_ms=5, estimated_full_test_time_s=1, + expected_load_ms=100, load_peak_vram_mb=1000, runtime_peak_vram_mb=2000, ) @@ -109,18 +193,30 @@ def test_each_request_failure_fails_case(harness, monkeypatch, bad_request, fail outputs[bad_request] = RuntimeError("server request failed") generate = Mock(side_effect=outputs) monkeypatch.setattr(runner, "run_and_collect", generate) - ctx = object() + ctx = SimpleNamespace(load_time_ms=100) - with pytest.raises(pytest.fail.Exception, match=f"request {bad_request + 1}/2"): + terminal = failure in {"performance", "load_peak", "runtime_peak", "missing_memory"} + error_type = ( + test_server_common.PerformanceValidationError + if terminal + else pytest.fail.Exception + ) + with pytest.raises(error_type, match=f"request {bad_request + 1}/2"): runner.test_diffusion_generation(case, ctx) - assert generate.call_count == 2 + expected_requests = bad_request + 1 if terminal else 2 + assert generate.call_count == expected_requests 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 + expected_consistency = ( + bad_request if terminal else (1 if failure == "generation" else 2) ) + assert runner._validate_consistency.call_count == expected_consistency # Even failed performance measurements must survive in the report. - expected = [i + 1 for i in range(2) if failure != "generation" or i != bad_request] + expected = [ + i + 1 + for i in range(expected_requests) + if failure != "generation" or i != bad_request + ] assert [r["request_index"] for r in runner._perf_results] == expected @@ -131,7 +227,7 @@ def test_both_requests_pass(harness, monkeypatch): "run_and_collect", Mock(side_effect=[(_perf_record(), b"first"), (_perf_record(), b"second")]), ) - runner.test_diffusion_generation(case, object()) + runner.test_diffusion_generation(case, SimpleNamespace(load_time_ms=100)) assert runner._validate_consistency.call_count == 2 assert [call.args[1] for call in runner._validate_consistency.call_args_list] == [ b"first", @@ -140,6 +236,37 @@ def test_both_requests_pass(harness, monkeypatch): assert [r["request_index"] for r in runner._perf_results] == [1, 2] +@pytest.mark.parametrize("run_perf_check", [False, True]) +@pytest.mark.parametrize("e2e_ms", [None, 0, -1, float("nan"), float("inf")]) +def test_e2e_is_required_even_without_threshold_checks( + harness, monkeypatch, run_perf_check, e2e_ms +): + runner, case = harness + case = replace(case, run_perf_check=run_perf_check) + record = _perf_record() + record.total_duration_ms = e2e_ms + generate = Mock(return_value=(record, b"output")) + monkeypatch.setattr(runner, "run_and_collect", generate) + + with pytest.raises( + test_server_common.PerformanceValidationError, + match="E2E duration missing or invalid", + ): + runner.test_diffusion_generation(case, SimpleNamespace(load_time_ms=100)) + assert generate.call_count == 1 + assert not runner._perf_results + + +def test_disabled_threshold_checks_still_record_e2e(harness, monkeypatch): + runner, case = harness + case = replace(case, run_perf_check=False) + monkeypatch.setattr( + runner, "run_and_collect", Mock(return_value=(_perf_record(), b"output")) + ) + runner.test_diffusion_generation(case, SimpleNamespace(load_time_ms=100)) + assert [r["e2e_ms"] for r in runner._perf_results] == [100, 100] + + 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)) @@ -150,7 +277,7 @@ def test_request_artifacts_do_not_overwrite_each_other(harness, monkeypatch, tmp return _perf_record(), b"output" monkeypatch.setattr(runner, "run_and_collect", generate) - runner.test_diffusion_generation(case, object()) + runner.test_diffusion_generation(case, SimpleNamespace(load_time_ms=100)) assert artifact_dirs == [str(tmp_path / f"request-{i}") for i in (1, 2)] assert os.environ["SGLANG_DIFFUSION_ARTIFACT_DIR"] == str(tmp_path) @@ -168,7 +295,7 @@ def test_later_skip_cannot_hide_earlier_failure(harness, monkeypatch): ), ) with pytest.raises(pytest.fail.Exception, match="failed first"): - runner.test_diffusion_generation(case, object()) + runner.test_diffusion_generation(case, SimpleNamespace(load_time_ms=100)) def test_second_request_cannot_be_skipped_after_first_passes(harness, monkeypatch): @@ -179,7 +306,7 @@ def test_second_request_cannot_be_skipped_after_first_passes(harness, monkeypatc 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()) + runner.test_diffusion_generation(case, SimpleNamespace(load_time_ms=100)) def test_empty_content_is_not_a_consistency_pass(harness): @@ -206,7 +333,7 @@ def test_audio_checked_even_when_video_consistency_fails( 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()) + runner.test_diffusion_generation(case, SimpleNamespace(load_time_ms=100)) assert runner._validate_consistency.call_count == 2 assert runner._validate_audio_consistency.call_count == 2 @@ -252,7 +379,7 @@ def test_perf_fixture_retains_failed_case_results(pytester, monkeypatch): DiffusionServerArgs("test", modality="image"), DiffusionSamplingParams(prompt="test"), ) - summary = PerformanceSummary(100, 5, 5, {}, [], {}, {}) + summary = PerformanceSummary(100, 5, 5, {}, [], {}, {}, load_time_ms=100) for index in (1, 2): self._record_performance_result(case, summary, index) if case_id == "failed": @@ -281,7 +408,7 @@ def test_gt_generation_runs_both_requests(harness, monkeypatch): monkeypatch.setattr(runner, "run_and_collect", generate) save = Mock() monkeypatch.setattr(runner, "_save_gt_output", save) - runner.test_diffusion_generation(case, object()) + runner.test_diffusion_generation(case, SimpleNamespace(load_time_ms=100)) assert [(call.args[0].id, call.args[1]) for call in save.call_args_list] == [ ("first", b"first"), ("first", b"second"), @@ -303,7 +430,7 @@ def test_baseline_generation_keeps_worst_of_both_requests(harness, monkeypatch): monkeypatch.setattr( runner, "run_and_collect", Mock(side_effect=[(r, b"output") for r in records]) ) - runner.test_diffusion_generation(case, object()) + runner.test_diffusion_generation(case, SimpleNamespace(load_time_ms=100)) summaries = test_server_common._PENDING_BASELINE_DUMPS[case.id] assert len(summaries) == 2 log = Mock() diff --git a/python/sglang/multimodal_gen/test/unit/test_suite_partitioning.py b/python/sglang/multimodal_gen/test/unit/test_suite_partitioning.py index 6c4214177..a372d2db1 100644 --- a/python/sglang/multimodal_gen/test/unit/test_suite_partitioning.py +++ b/python/sglang/multimodal_gen/test/unit/test_suite_partitioning.py @@ -140,3 +140,8 @@ def test_qwen_quality_variants_use_the_same_generation_request(): assert extra_high.prompt == lossless.prompt assert extra_high.output_size == lossless.output_size assert extra_high.extras == {"quality": "extra-high"} + + scenarios = json.loads(_H100_BASELINE_PATH.read_text())["scenarios"] + for case_id in ("qwen_image_t2i_2_gpus", "qwen_image_t2i_2_gpus_extra_high"): + assert cases[case_id].run_perf_check + assert scenarios[case_id]["expected_e2e_ms"] > 0 diff --git a/python/sglang/multimodal_gen/test/unit/test_timestep_preparation_logging.py b/python/sglang/multimodal_gen/test/unit/test_timestep_preparation_logging.py new file mode 100644 index 000000000..ea6d97c02 --- /dev/null +++ b/python/sglang/multimodal_gen/test/unit/test_timestep_preparation_logging.py @@ -0,0 +1,65 @@ +import logging +from types import SimpleNamespace +from unittest.mock import patch + +import pytest +import torch + +from sglang.multimodal_gen.configs.pipeline_configs.flux import ( + Flux2KleinBasePipelineConfig, +) +from sglang.multimodal_gen.configs.sample.sampling_params import SamplingParams +from sglang.multimodal_gen.runtime.models.schedulers.scheduling_flow_match_euler_discrete import ( + FlowMatchEulerDiscreteScheduler, +) +from sglang.multimodal_gen.runtime.pipelines_core.schedule_batch import Req +from sglang.multimodal_gen.runtime.pipelines_core.stages import ( + base, +) +from sglang.multimodal_gen.runtime.pipelines_core.stages import ( + timestep_preparation as module, +) + + +@pytest.mark.parametrize("warmup", [False, True]) +@pytest.mark.parametrize("debug", [False, True]) +def test_timestep_logging_preserves_scheduler_and_skips_unused_copy(warmup, debug): + args = SimpleNamespace(pipeline_config=Flux2KleinBasePipelineConfig()) + scheduler = FlowMatchEulerDiscreteScheduler() + with patch.object(base, "get_global_server_args", return_value=args): + stage = module.TimestepPreparationStage(scheduler) + batch = Req(sampling_params=SamplingParams(num_inference_steps=4)) + batch.is_warmup = warmup + records = [] + + class Capture(logging.Handler): + def emit(self, record): + records.append(record) + self.format(record) + + test_logger = logging.Logger( + "timestep-test", logging.DEBUG if debug else logging.INFO + ) + test_logger.addHandler(Capture()) + with ( + patch.object(module, "logger", test_logger), + patch.object( + module, "get_local_torch_device", return_value=torch.device("cpu") + ), + ): + if debug and not warmup: + result = stage.forward(batch, args) + else: + with patch.object( + torch.Tensor, "detach", side_effect=AssertionError("unused log copy") + ): + result = stage.forward(batch, args) + assert result is batch + assert batch.scheduler is scheduler + assert batch.timesteps is scheduler.timesteps + assert len(records) == int(debug and not warmup) + if records: + value = records[0].args[-1] + assert value.device.type == "cpu" + assert torch.equal(value, batch.timesteps) + assert "TimestepPreparationStage" in records[0].getMessage() diff --git a/python/sglang/multimodal_gen/test/unit/test_video_url_request_extras.py b/python/sglang/multimodal_gen/test/unit/test_video_url_request_extras.py new file mode 100644 index 000000000..75f327f41 --- /dev/null +++ b/python/sglang/multimodal_gen/test/unit/test_video_url_request_extras.py @@ -0,0 +1,36 @@ +from unittest.mock import Mock + +import pytest + +from sglang.multimodal_gen.test.server.test_server_utils import get_generate_fn +from sglang.multimodal_gen.test.server.testcase_configs import DiffusionSamplingParams + + +def test_url_video_request_preserves_sampling_extras(): + extras = { + "profile": True, + "num_profiled_timesteps": 5, + "num_inference_steps": 12, + "seed": 0, + } + original_extras = extras.copy() + params = DiffusionSamplingParams( + prompt="test", + image_path="https://example.com/input.png", + direct_url_test=True, + fps=24, + num_frames=25, + extras=extras, + ) + client = Mock() + client.videos.create.side_effect = ConnectionError("stop at transport boundary") + generate = get_generate_fn("test-model", "video", params) + with pytest.raises(ConnectionError, match="stop at transport boundary"): + generate("url-video", client) + assert client.videos.create.call_args.kwargs["extra_body"] == { + "reference_url": params.image_path, + "fps": 24, + "num_frames": 25, + **original_extras, + } + assert params.extras == original_extras