diff --git a/.github/workflows/diffusion-ci-gt-gen.yml b/.github/workflows/diffusion-ci-gt-gen.yml index 4e335655a..92844245b 100644 --- a/.github/workflows/diffusion-ci-gt-gen.yml +++ b/.github/workflows/diffusion-ci-gt-gen.yml @@ -112,4 +112,4 @@ jobs: - name: Publish GT images to sglang-bot/sglang-ci-data env: GITHUB_TOKEN: ${{ secrets.GH_PAT_FOR_NIGHTLY_CI_DATA }} - run: python scripts/ci/utils/publish_diffusion_gt.py --source-dir gt_images + run: python scripts/ci/utils/diffusion/publish_diffusion_gt.py --source-dir gt_images diff --git a/.github/workflows/nightly-test-nvidia.yml b/.github/workflows/nightly-test-nvidia.yml index f4d4561c9..cffd45834 100644 --- a/.github/workflows/nightly-test-nvidia.yml +++ b/.github/workflows/nightly-test-nvidia.yml @@ -26,6 +26,7 @@ on: - 'nightly-test-perf-4-gpu-b200' - 'nightly-test-perf-8-gpu-b200' - 'nightly-test-kernel-1-gpu-h100' + - 'nightly-test-diffusion-comparison' workflow_call: inputs: ref: @@ -511,7 +512,7 @@ jobs: - name: Collect diffusion performance metrics if: always() run: | - python3 scripts/ci/utils/save_diffusion_metrics.py \ + python3 scripts/ci/utils/diffusion/save_diffusion_metrics.py \ --gpu-config 1-gpu-h100 \ --run-id ${{ github.run_id }} \ --output python/diffusion-metrics-1gpu-partition-${{ matrix.part }}.json \ @@ -571,7 +572,7 @@ jobs: - name: Collect diffusion performance metrics if: always() run: | - python3 scripts/ci/utils/save_diffusion_metrics.py \ + python3 scripts/ci/utils/diffusion/save_diffusion_metrics.py \ --gpu-config 2-gpu-h100 \ --run-id ${{ github.run_id }} \ --output python/diffusion-metrics-2gpu-partition-${{ matrix.part }}.json \ @@ -649,6 +650,69 @@ jobs: - uses: ./.github/actions/upload-cuda-coredumps if: always() + # Diffusion cross-framework comparison + nightly-test-diffusion-comparison: + if: github.repository == 'sgl-project/sglang' && (inputs.job_filter == '' || inputs.job_filter == 'all' || inputs.job_filter == 'nightly-test-diffusion-comparison') + runs-on: 4-gpu-h100 + timeout-minutes: 240 + steps: + - name: Checkout code + uses: actions/checkout@v4 + with: + ref: ${{ inputs.ref || github.ref }} + + - name: Install dependencies + run: | + bash scripts/ci/cuda/ci_install_dependency.sh diffusion + + - name: Run cross-framework comparison + env: + GITHUB_SHA: ${{ github.sha }} + GITHUB_RUN_ID: ${{ github.run_id }} + PYTHONUNBUFFERED: "1" + timeout-minutes: 210 + run: | + python3 -u scripts/ci/utils/diffusion/run_comparison.py \ + --output comparison-results.json + + - name: Generate dashboard + if: always() + env: + GH_PAT_FOR_NIGHTLY_CI_DATA: ${{ secrets.GH_PAT_FOR_NIGHTLY_CI_DATA }} + run: | + python3 scripts/ci/utils/diffusion/generate_diffusion_dashboard.py \ + --results comparison-results.json \ + --output dashboard.md \ + --charts-dir comparison-charts \ + --fetch-history \ + --step-summary + + - name: Publish to sglang-ci-data + if: always() + env: + GH_PAT_FOR_NIGHTLY_CI_DATA: ${{ secrets.GH_PAT_FOR_NIGHTLY_CI_DATA }} + run: | + python3 scripts/ci/utils/diffusion/publish_comparison_results.py \ + --results comparison-results.json \ + --dashboard dashboard.md \ + --charts-dir comparison-charts + + - name: Upload comparison artifacts + if: always() + uses: actions/upload-artifact@v4 + with: + name: diffusion-comparison-${{ github.run_id }} + path: | + comparison-results.json + dashboard.md + comparison-charts/ + comparison-logs/ + retention-days: 90 + if-no-files-found: ignore + + - uses: ./.github/actions/upload-cuda-coredumps + if: always() + # Consolidate performance metrics from all jobs consolidate-metrics: if: github.repository == 'sgl-project/sglang' && always() @@ -710,6 +774,7 @@ jobs: - nightly-test-multimodal-server-2-gpu - nightly-test-perf-4-gpu-b200 - nightly-test-specialized-8-gpu-b200 + - nightly-test-diffusion-comparison - consolidate-metrics runs-on: ubuntu-latest steps: diff --git a/python/sglang/multimodal_gen/runtime/entrypoints/openai/image_api.py b/python/sglang/multimodal_gen/runtime/entrypoints/openai/image_api.py index 116208451..18d5ed983 100644 --- a/python/sglang/multimodal_gen/runtime/entrypoints/openai/image_api.py +++ b/python/sglang/multimodal_gen/runtime/entrypoints/openai/image_api.py @@ -136,6 +136,7 @@ async def generations( enable_upscaling=request.enable_upscaling, upscaling_model_path=request.upscaling_model_path, upscaling_scale=request.upscaling_scale, + perf_dump_path=request.perf_dump_path, ) batch = prepare_request( server_args=server_args, diff --git a/python/sglang/multimodal_gen/runtime/entrypoints/openai/protocol.py b/python/sglang/multimodal_gen/runtime/entrypoints/openai/protocol.py index 425a17f7b..9831d368a 100644 --- a/python/sglang/multimodal_gen/runtime/entrypoints/openai/protocol.py +++ b/python/sglang/multimodal_gen/runtime/entrypoints/openai/protocol.py @@ -53,6 +53,8 @@ class ImageGenerationsRequest(BaseModel): upscaling_model_path: Optional[str] = None upscaling_scale: Optional[int] = 4 diffusers_kwargs: Optional[Dict[str, Any]] = None # kwargs for diffusers backend + # Performance profiling + perf_dump_path: Optional[str] = None # Video API protocol models @@ -109,6 +111,8 @@ class VideoGenerationsRequest(BaseModel): output_compression: Optional[int] = None output_path: Optional[str] = None diffusers_kwargs: Optional[Dict[str, Any]] = None # kwargs for diffusers backend + # Performance profiling + perf_dump_path: Optional[str] = None class VideoListResponse(BaseModel): diff --git a/python/sglang/multimodal_gen/runtime/entrypoints/openai/video_api.py b/python/sglang/multimodal_gen/runtime/entrypoints/openai/video_api.py index d85ce7d40..2a444111d 100644 --- a/python/sglang/multimodal_gen/runtime/entrypoints/openai/video_api.py +++ b/python/sglang/multimodal_gen/runtime/entrypoints/openai/video_api.py @@ -81,6 +81,7 @@ def _build_video_sampling_params(request_id: str, request: VideoGenerationsReque output_path=request.output_path, output_compression=request.output_compression, output_quality=request.output_quality, + perf_dump_path=request.perf_dump_path, ) diff --git a/python/sglang/multimodal_gen/runtime/layers/lora/linear.py b/python/sglang/multimodal_gen/runtime/layers/lora/linear.py index e83f75a49..6b426ac64 100644 --- a/python/sglang/multimodal_gen/runtime/layers/lora/linear.py +++ b/python/sglang/multimodal_gen/runtime/layers/lora/linear.py @@ -33,7 +33,7 @@ from sglang.multimodal_gen.runtime.layers.vocab_parallel_embedding import ( ) from sglang.multimodal_gen.utils import get_mixed_precision_state -torch._dynamo.config.recompile_limit = 16 +torch._dynamo.config.recompile_limit = 64 class BaseLayerWithLoRA(nn.Module): diff --git a/python/sglang/multimodal_gen/runtime/managers/gpu_worker.py b/python/sglang/multimodal_gen/runtime/managers/gpu_worker.py index 8757da74e..df9f7c455 100644 --- a/python/sglang/multimodal_gen/runtime/managers/gpu_worker.py +++ b/python/sglang/multimodal_gen/runtime/managers/gpu_worker.py @@ -297,6 +297,19 @@ class GPUWorker: # Avoid logging warmup perf records that share the same request_id. if not req.is_warmup: PerformanceLogger.log_request_summary(metrics=output_batch.metrics) + + # dump per-request perf report to specified file (server mode) + if ( + req.perf_dump_path is not None + and not req.is_warmup + and output_batch.metrics is not None + ): + PerformanceLogger.dump_benchmark_report( + file_path=req.perf_dump_path, + metrics=output_batch.metrics, + meta={"model": self.server_args.model_path}, + tag="server_perf_dump", + ) except Exception as e: logger.error( f"Error executing request {req.request_id}: {e}", exc_info=True diff --git a/scripts/ci/utils/diffusion/__init__.py b/scripts/ci/utils/diffusion/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/scripts/ci/utils/diffusion/comparison_configs.json b/scripts/ci/utils/diffusion/comparison_configs.json new file mode 100644 index 000000000..7f655a1ab --- /dev/null +++ b/scripts/ci/utils/diffusion/comparison_configs.json @@ -0,0 +1,156 @@ +{ + "_comment": "Per-model comparison config. Only frameworks listed under each case are tested. vLLM-Omni disabled until dep install issues resolved.", + "test_image_url": "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/diffusers/cat.png", + "cases": [ + { + "id": "flux1_dev_t2i_1024", + "model": "black-forest-labs/FLUX.1-dev", + "task": "text-to-image", + "prompt": "A futuristic cyberpunk city at night, neon lights reflecting on wet streets", + "width": 1024, + "height": 1024, + "num_inference_steps": 50, + "guidance_scale": 4.0, + "seed": 42, + "num_gpus": 1, + "frameworks": { + "sglang": { + "serve_args": "--enable-torch-compile --warmup --dit-layerwise-offload false", + "extra_env": {} + } + } + }, + { + "id": "flux2_dev_t2i_1024", + "model": "black-forest-labs/FLUX.2-dev", + "task": "text-to-image", + "prompt": "A futuristic cyberpunk city at night, neon lights reflecting on wet streets", + "width": 1024, + "height": 1024, + "num_inference_steps": 50, + "guidance_scale": 4.0, + "seed": 42, + "num_gpus": 1, + "frameworks": { + "sglang": { + "serve_args": "--enable-torch-compile --warmup --dit-layerwise-offload false", + "extra_env": {} + } + } + }, + { + "id": "qwen_image_2512_t2i_1024", + "model": "Qwen/Qwen-Image-2512", + "task": "text-to-image", + "prompt": "A futuristic cyberpunk city at night, neon lights reflecting on wet streets", + "width": 1024, + "height": 1024, + "num_inference_steps": 50, + "guidance_scale": 4.0, + "seed": 42, + "num_gpus": 1, + "frameworks": { + "sglang": { + "serve_args": "--enable-torch-compile --warmup", + "extra_env": {} + } + } + }, + { + "id": "qwen_image_edit_2511", + "model": "Qwen/Qwen-Image-Edit-2511", + "task": "image-edit", + "prompt": "Make the cat wear a red hat", + "reference_image": true, + "width": 1024, + "height": 1024, + "num_inference_steps": 50, + "guidance_scale": 4.0, + "seed": 42, + "num_gpus": 1, + "frameworks": { + "sglang": { + "serve_args": "--enable-torch-compile --warmup", + "extra_env": {} + } + } + }, + { + "id": "zimage_turbo_t2i_1024", + "model": "Tongyi-MAI/Z-Image-Turbo", + "task": "text-to-image", + "prompt": "A futuristic cyberpunk city at night, neon lights reflecting on wet streets", + "width": 1024, + "height": 1024, + "num_inference_steps": 9, + "guidance_scale": 4.0, + "seed": 42, + "num_gpus": 1, + "frameworks": { + "sglang": { + "serve_args": "--enable-torch-compile --warmup", + "extra_env": {} + } + } + }, + { + "id": "wan22_t2v_a14b_720p", + "model": "Wan-AI/Wan2.2-T2V-A14B-Diffusers", + "task": "text-to-video", + "prompt": "A cat and a dog baking a cake together in a kitchen.", + "width": 1280, + "height": 720, + "num_frames": 81, + "num_inference_steps": 2, + "guidance_scale": 5.0, + "seed": 42, + "num_gpus": 4, + "frameworks": { + "sglang": { + "serve_args": "--enable-torch-compile --warmup --enable-cfg-parallel --ulysses-degree 2 --text-encoder-cpu-offload --pin-cpu-memory", + "extra_env": {} + } + } + }, + { + "id": "wan22_ti2v_5b_720p", + "model": "Wan-AI/Wan2.2-TI2V-5B-Diffusers", + "task": "text-image-to-video", + "prompt": "The cat starts walking slowly towards the camera.", + "reference_image": true, + "width": 1280, + "height": 720, + "num_frames": 81, + "num_inference_steps": 50, + "guidance_scale": 5.0, + "seed": 42, + "num_gpus": 1, + "frameworks": { + "sglang": { + "serve_args": "--enable-torch-compile --warmup", + "extra_env": {} + } + } + }, + { + "id": "wan22_i2v_a14b_720p", + "model": "Wan-AI/Wan2.2-I2V-A14B-Diffusers", + "task": "image-to-video", + "prompt": "The cat starts walking slowly towards the camera.", + "reference_image": true, + "width": 1280, + "height": 720, + "num_frames": 81, + "num_inference_steps": 2, + "guidance_scale": 5.0, + "seed": 42, + "num_gpus": 4, + "frameworks": { + "sglang": { + "serve_args": "--enable-torch-compile --warmup --enable-cfg-parallel --ulysses-degree 2 --text-encoder-cpu-offload --pin-cpu-memory", + "extra_env": {} + } + } + } + ] +} diff --git a/scripts/ci/utils/diffusion/generate_diffusion_dashboard.py b/scripts/ci/utils/diffusion/generate_diffusion_dashboard.py new file mode 100644 index 000000000..183611ef0 --- /dev/null +++ b/scripts/ci/utils/diffusion/generate_diffusion_dashboard.py @@ -0,0 +1,665 @@ +"""Generate a Markdown dashboard for diffusion cross-framework comparisons. + +Reads current comparison results + historical data from sglang-ci-data repo +and produces a Markdown report with tables and trend charts saved as PNG files. + +Usage: + python3 scripts/ci/utils/diffusion/generate_diffusion_dashboard.py \ + --results comparison-results.json \ + --output dashboard.md \ + --charts-dir comparison-charts/ \ + --history-dir history/ # optional, local history JSONs + --fetch-history # fetch from GitHub API instead +""" + +import argparse +import json +import os +import sys +from datetime import datetime, timezone + +# --------------------------------------------------------------------------- +# History fetching (from sglang-ci-data repo via GitHub API) +# --------------------------------------------------------------------------- + +CI_DATA_REPO_OWNER = "sglang-bot" +CI_DATA_REPO_NAME = "sglang-ci-data" +CI_DATA_BRANCH = "main" +HISTORY_PREFIX = "diffusion-comparisons" +MAX_HISTORY_RUNS = 7 + +# Base URL for chart images pushed to sglang-ci-data +CHARTS_RAW_BASE_URL = ( + f"https://raw.githubusercontent.com/{CI_DATA_REPO_OWNER}/{CI_DATA_REPO_NAME}" + f"/{CI_DATA_BRANCH}/{HISTORY_PREFIX}/charts" +) + + +def _github_get(url: str, token: str) -> dict | list | None: + """Simple GET to GitHub API.""" + from urllib.error import HTTPError + from urllib.request import Request, urlopen + + headers = { + "Accept": "application/vnd.github+json", + "Authorization": f"Bearer {token}", + "X-GitHub-Api-Version": "2022-11-28", + } + req = Request(url, headers=headers) + try: + with urlopen(req) as resp: + return json.loads(resp.read().decode("utf-8")) + except HTTPError as e: + print(f" Warning: GitHub API request failed ({e.code}): {url}") + return None + except Exception as e: + print(f" Warning: GitHub API request error: {e}") + return None + + +def fetch_history_from_github(token: str) -> list[dict]: + """Fetch recent comparison result JSONs from sglang-ci-data repo.""" + print("Fetching historical comparison data from GitHub...") + url = ( + f"https://api.github.com/repos/{CI_DATA_REPO_OWNER}/{CI_DATA_REPO_NAME}" + f"/contents/{HISTORY_PREFIX}?ref={CI_DATA_BRANCH}" + ) + listing = _github_get(url, token) + if not listing or not isinstance(listing, list): + print(" No historical data found.") + return [] + + # Filter JSON files and sort by name (date prefix) descending + json_files = sorted( + [f for f in listing if f["name"].endswith(".json")], + key=lambda f: f["name"], + reverse=True, + )[:MAX_HISTORY_RUNS] + + history = [] + for entry in json_files: + raw_url = entry.get("download_url") + if not raw_url: + continue + data = _github_get(raw_url, token) + if data and isinstance(data, dict): + history.append(data) + print(f" Loaded {len(history)} historical run(s).") + return history + + +def load_history_from_dir(history_dir: str) -> list[dict]: + """Load historical JSONs from a local directory.""" + if not os.path.isdir(history_dir): + return [] + files = sorted( + [f for f in os.listdir(history_dir) if f.endswith(".json")], + reverse=True, + )[:MAX_HISTORY_RUNS] + history = [] + for fname in files: + try: + with open(os.path.join(history_dir, fname)) as f: + history.append(json.load(f)) + except Exception: + pass + return history + + +# --------------------------------------------------------------------------- +# Dashboard generation +# --------------------------------------------------------------------------- + + +def _fmt_latency(val: float | None) -> str: + if val is None: + return "N/A" + return f"{val:.2f}" + + +def _fmt_speedup(sglang_lat: float | None, other_lat: float | None) -> str: + if sglang_lat is None or other_lat is None or sglang_lat <= 0: + return "N/A" + ratio = other_lat / sglang_lat + return f"{ratio:.2f}x" + + +def _short_date(ts: str) -> str: + """Extract short date from ISO timestamp.""" + try: + dt = datetime.fromisoformat(ts.replace("Z", "+00:00")) + return dt.strftime("%b %d") + except Exception: + return ts[:10] + + +def _short_sha(sha: str) -> str: + return sha[:7] if sha and sha != "unknown" else "?" + + +def _assess_risk( + cid: str, + current_cases: dict[str, dict[str, float | None]], + history: list[dict], + other_frameworks: list[str], +) -> tuple[str, str]: + """Assess risk for a given case, returning (emoji, reason). + + Rules (checked in order): + - N/A latency → ❌ broken + - History exists: SGLang latency >5% vs avg of last 3 runs → ⚠️ regression + - Competitor exists & SGLang slower → 🔴 competitive risk + - SGLang faster than all competitors by >20% → 🟢 strong advantage + - SGLang faster than all competitors by ≤20% → 🟡 moderate advantage + - Default → ✅ stable + """ + sg_lat = current_cases.get(cid, {}).get("sglang") + + # Broken: sglang latency is N/A + if sg_lat is None: + return "❌", f"{cid}: SGLang latency is N/A (broken)" + + # Check regression against 3-run historical average + if history: + hist_lats: list[float] = [] + for run in history[:3]: + run_cases = _extract_case_results(run) + h_lat = run_cases.get(cid, {}).get("sglang") + if h_lat is not None: + hist_lats.append(h_lat) + if hist_lats: + avg_3 = sum(hist_lats) / len(hist_lats) + if avg_3 > 0 and (sg_lat - avg_3) / avg_3 > 0.05: + pct = (sg_lat - avg_3) / avg_3 * 100 + return ( + "⚠️", + f"{cid}: SGLang regression +{pct:.1f}% vs 3-run avg " + f"({sg_lat:.2f}s vs {avg_3:.2f}s)", + ) + + # Check competitive risk + if other_frameworks: + competitor_lats: dict[str, float] = {} + for ofw in other_frameworks: + olat = current_cases.get(cid, {}).get(ofw) + if olat is not None: + competitor_lats[ofw] = olat + + if competitor_lats: + # SGLang slower than any competitor? + for ofw, olat in competitor_lats.items(): + if sg_lat > olat: + return ( + "🔴", + f"{cid}: SGLang slower than {ofw} " + f"({sg_lat:.2f}s vs {olat:.2f}s)", + ) + + # SGLang faster — check margin + min_competitor = min(competitor_lats.values()) + advantage = (min_competitor - sg_lat) / min_competitor + if advantage > 0.20: + return "🟢", "" + else: + return "🟡", "" + + # Default: stable + return "✅", "" + + +def _trend_emoji(current: float | None, previous: float | None) -> str: + if current is None or previous is None: + return "" + diff_pct = (current - previous) / previous * 100 + if diff_pct < -2: + return " :arrow_down:" # faster (good) + elif diff_pct > 2: + return " :arrow_up:" # slower (bad) + return " :left_right_arrow:" + + +def _extract_case_results(run_data: dict) -> dict[str, dict[str, float | None]]: + """Extract {case_id: {framework: latency}} from a run.""" + mapping: dict[str, dict[str, float | None]] = {} + for r in run_data.get("results", []): + cid = r["case_id"] + fw = r["framework"] + if cid not in mapping: + mapping[cid] = {} + mapping[cid][fw] = r.get("latency_s") + return mapping + + +def _sanitize_filename(name: str) -> str: + """Sanitize a case ID to be a safe filename.""" + return name.replace("/", "_").replace(" ", "_").replace(":", "_") + + +def generate_dashboard( + current: dict, + history: list[dict], + charts_dir: str | None = None, +) -> str: + """Generate full markdown dashboard. + + If charts_dir is provided, saves chart PNGs as files to that directory + and references them via raw.githubusercontent URLs. Otherwise, charts + are omitted. + + Returns the markdown string. + """ + lines: list[str] = [] + lines.append("# Diffusion Cross-Framework Performance Dashboard\n") + ts = current.get("timestamp", datetime.now(timezone.utc).isoformat()) + sha = current.get("commit_sha", "unknown") + lines.append(f"*Generated: {_short_date(ts)} | Commit: `{_short_sha(sha)}`*\n") + + current_cases = _extract_case_results(current) + case_ids = list(current_cases.keys()) + + # ---- Regression detection ---- + REGRESSION_THRESHOLD = 0.05 # 5% + regressions: list[str] = [] + if history: + prev_cases = _extract_case_results(history[0]) + for cid in case_ids: + for fw in ("sglang", "vllm-omni"): + cur = current_cases.get(cid, {}).get(fw) + prev = prev_cases.get(cid, {}).get(fw) + if cur and prev and prev > 0: + pct = (cur - prev) / prev + if pct > REGRESSION_THRESHOLD: + regressions.append( + f"**{cid}** ({fw}): {prev:.2f}s -> {cur:.2f}s " + f"(+{pct*100:.1f}%)" + ) + + if regressions: + lines.append("> [!WARNING]\n> **Performance Regression Detected**\n>") + for reg in regressions: + lines.append(f"> - {reg}") + lines.append("\n") + + # Discover all frameworks present in results + all_frameworks = [] + seen_fw = set() + for r in current.get("results", []): + fw = r["framework"] + if fw not in seen_fw: + all_frameworks.append(fw) + seen_fw.add(fw) + # Ensure sglang is first + if "sglang" in all_frameworks: + all_frameworks.remove("sglang") + all_frameworks.insert(0, "sglang") + other_frameworks = [fw for fw in all_frameworks if fw != "sglang"] + + # ---- Section 1: Cross-Framework Comparison (current run) ---- + lines.append("## Cross-Framework Performance Comparison\n") + + # Compute risk assessments for all cases + risk_map: dict[str, tuple[str, str]] = {} + for cid in case_ids: + risk_map[cid] = _assess_risk(cid, current_cases, history, other_frameworks) + + # Dynamic header + header = "| Model | Risk |" + sep = "|-------|------|" + for fw in all_frameworks: + header += f" {fw} (s) |" + sep += "---------|" + for ofw in other_frameworks: + header += f" vs {ofw} |" + sep += "---------|" + lines.append(header) + lines.append(sep) + + # One row per case (deduplicated by case_id) + seen_cases = set() + for r in current.get("results", []): + cid = r["case_id"] + if cid in seen_cases: + continue + seen_cases.add(cid) + + case_fws = current_cases.get(cid, {}) + sg_lat = case_fws.get("sglang") + + risk_emoji, _ = risk_map.get(cid, ("✅", "")) + row = f"| {r['model'].split('/')[-1]} | {risk_emoji} |" + # Latency columns -- bold the fastest + lats = {fw: case_fws.get(fw) for fw in all_frameworks} + valid_lats = [v for v in lats.values() if v is not None] + min_lat = min(valid_lats) if valid_lats else None + for fw in all_frameworks: + lat = lats[fw] + if lat is not None and min_lat is not None and lat == min_lat: + row += f" **{_fmt_latency(lat)}** |" + else: + row += f" {_fmt_latency(lat)} |" + # Speedup columns + for ofw in other_frameworks: + row += f" {_fmt_speedup(sg_lat, case_fws.get(ofw))} |" + lines.append(row) + + # ---- Section 2: SGLang Performance Trend ---- + if history: + lines.append("\n## SGLang Performance Trend (Last 7 Runs)\n") + + # Build header + header = "| Date | Commit |" + sep = "|------|--------|" + for cid in case_ids: + header += f" {cid} (s) |" + sep += "---------|" + header += " Trend |" + sep += "-------|" + lines.append(header) + lines.append(sep) + + # Current run first + all_runs = [current] + history + for i, run in enumerate(all_runs): + run_cases = _extract_case_results(run) + date = _short_date(run.get("timestamp", "")) + sha_s = _short_sha(run.get("commit_sha", "")) + row = f"| {date} | `{sha_s}` |" + for cid in case_ids: + lat = run_cases.get(cid, {}).get("sglang") + row += f" {_fmt_latency(lat)} |" + # Trend vs next (older) run + if i + 1 < len(all_runs): + prev_cases = _extract_case_results(all_runs[i + 1]) + emojis = [] + for cid in case_ids: + cur = run_cases.get(cid, {}).get("sglang") + prev = prev_cases.get(cid, {}).get("sglang") + emojis.append(_trend_emoji(cur, prev)) + row += " ".join(emojis) + " |" + else: + row += " -- |" + lines.append(row) + + # ---- Section 3: Cross-Framework Speedup Trend (only if multiple frameworks) ---- + if history and other_frameworks: + lines.append("\n## SGLang vs vLLM-Omni Speedup Over Time\n") + + header = "| Date |" + sep = "|------|" + for cid in case_ids: + header += f" {cid} |" + sep += "---------|" + lines.append(header) + lines.append(sep) + + all_runs = [current] + history + for run in all_runs: + run_cases = _extract_case_results(run) + date = _short_date(run.get("timestamp", "")) + row = f"| {date} |" + for cid in case_ids: + sg = run_cases.get(cid, {}).get("sglang") + vl = run_cases.get(cid, {}).get("vllm-omni") + row += f" {_fmt_speedup(sg, vl)} |" + lines.append(row) + + # ---- Section 4: Matplotlib Trend Charts (saved as PNG files) ---- + if history and charts_dir: + all_runs = list(reversed([current] + history)) # chronological order + + def _chart_label(run: dict) -> str: + d = _short_date(run.get("timestamp", "")) + s = _short_sha(run.get("commit_sha", "")) + return f"{d}\n({s})" + + try: + import matplotlib + + matplotlib.use("Agg") + import matplotlib.pyplot as plt + + os.makedirs(charts_dir, exist_ok=True) + + # Per-case latency trend charts + for cid in case_ids: + labels = [] + sg_vals = [] + vl_vals = [] + for run in all_runs: + run_cases = _extract_case_results(run) + sg = run_cases.get(cid, {}).get("sglang") + vl = run_cases.get(cid, {}).get("vllm-omni") + if sg is None: + continue + labels.append(_chart_label(run)) + sg_vals.append(sg) + vl_vals.append(vl) + + if not sg_vals: + continue + + has_vl = any(v is not None for v in vl_vals) + fig, ax = plt.subplots(figsize=(max(6, len(labels) * 1.2), 4)) + + # SGLang line + ax.plot( + range(len(sg_vals)), + sg_vals, + "o-", + color="#2563eb", + linewidth=2, + markersize=6, + label="SGLang", + ) + for i, v in enumerate(sg_vals): + ax.annotate( + f"{v:.2f}s", + (i, v), + textcoords="offset points", + xytext=(0, 10), + ha="center", + fontsize=8, + fontweight="bold", + color="#2563eb", + ) + + # vLLM-Omni line (if data exists) + if has_vl: + vl_clean = [v if v is not None else float("nan") for v in vl_vals] + ax.plot( + range(len(vl_clean)), + vl_clean, + "s--", + color="#dc2626", + linewidth=2, + markersize=5, + label="vLLM-Omni", + ) + for i, v in enumerate(vl_vals): + if v is not None: + ax.annotate( + f"{v:.2f}s", + (i, v), + textcoords="offset points", + xytext=(0, -14), + ha="center", + fontsize=8, + color="#dc2626", + ) + + ax.set_xticks(range(len(labels))) + ax.set_xticklabels(labels, fontsize=7) + ax.set_ylabel("Latency (s)") + ax.set_title(f"Latency Trend -- {cid}", fontsize=11, fontweight="bold") + ax.legend(loc="upper right", fontsize=8) + ax.grid(True, alpha=0.3) + ax.set_ylim(bottom=0) + + filename = f"latency_{_sanitize_filename(cid)}.png" + chart_path = os.path.join(charts_dir, filename) + fig.savefig(chart_path, format="png", dpi=120, bbox_inches="tight") + plt.close(fig) + print(f" Saved chart: {chart_path}") + + chart_url = f"{CHARTS_RAW_BASE_URL}/{filename}" + lines.append(f"\n### Latency Trend: {cid}\n") + lines.append(f"![Latency Trend {cid}]({chart_url})\n") + + # Speedup trend chart (only if multiple frameworks) + if other_frameworks: + fig, ax = plt.subplots(figsize=(max(6, len(all_runs) * 1.2), 4)) + colors = ["#2563eb", "#dc2626", "#16a34a", "#ea580c"] + for ci_idx, cid in enumerate(case_ids): + speedups = [] + run_labels = [] + for run in all_runs: + run_cases = _extract_case_results(run) + sg = run_cases.get(cid, {}).get("sglang") + vl = run_cases.get(cid, {}).get("vllm-omni") + if sg and vl and sg > 0: + speedups.append(vl / sg) + else: + speedups.append(None) + run_labels.append(_chart_label(run)) + clean = [v if v is not None else float("nan") for v in speedups] + ax.plot( + range(len(clean)), + clean, + "o-", + color=colors[ci_idx % len(colors)], + linewidth=2, + markersize=5, + label=cid, + ) + + ax.set_xticks(range(len(run_labels))) + ax.set_xticklabels(run_labels, fontsize=7) + ax.set_ylabel("Speedup (x)") + ax.set_title( + "SGLang Speedup Over vLLM-Omni", fontsize=11, fontweight="bold" + ) + ax.axhline(y=1.0, color="gray", linestyle=":", alpha=0.5) + ax.legend(loc="upper left", fontsize=7) + ax.grid(True, alpha=0.3) + + filename = "speedup_trend.png" + chart_path = os.path.join(charts_dir, filename) + fig.savefig(chart_path, format="png", dpi=120, bbox_inches="tight") + plt.close(fig) + print(f" Saved chart: {chart_path}") + + chart_url = f"{CHARTS_RAW_BASE_URL}/{filename}" + lines.append("\n### Speedup Trend (SGLang vs vLLM-Omni)\n") + lines.append(f"![Speedup Trend]({chart_url})\n") + + except ImportError: + lines.append("\n*Charts unavailable (matplotlib not installed)*\n") + + # ---- Risk Notification ---- + alert_cases = [ + (cid, emoji, reason) + for cid, (emoji, reason) in risk_map.items() + if emoji in ("⚠️", "🔴", "❌") + ] + if alert_cases: + lines.append("\n> [!CAUTION]") + lines.append("> **Action Required — Performance Alert**") + lines.append(">") + lines.append("> The following cases need attention:") + for _cid, _emoji, reason in alert_cases: + lines.append(f"> - {reason}") + lines.append(">") + lines.append("> cc @mickqian @bbuf @yhyang201\n") + + # Footer + lines.append("\n---") + lines.append( + "*Generated by `generate_diffusion_dashboard.py` in SGLang nightly CI.*" + ) + + return "\n".join(lines) + "\n" + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def main(): + parser = argparse.ArgumentParser( + description="Generate diffusion cross-framework comparison dashboard" + ) + parser.add_argument( + "--results", + required=True, + help="Path to comparison-results.json from current run", + ) + parser.add_argument( + "--output", + default="dashboard.md", + help="Output markdown file path", + ) + parser.add_argument( + "--charts-dir", + default="comparison-charts", + help="Directory to save chart PNG files (default: comparison-charts/)", + ) + parser.add_argument( + "--history-dir", + default=None, + help="Local directory containing historical comparison JSONs", + ) + parser.add_argument( + "--fetch-history", + action="store_true", + help="Fetch history from sglang-ci-data GitHub repo", + ) + parser.add_argument( + "--step-summary", + action="store_true", + help="Also write to $GITHUB_STEP_SUMMARY", + ) + + args = parser.parse_args() + + # Load current results + with open(args.results) as f: + current = json.load(f) + print(f"Loaded current results: {len(current.get('results', []))} entries") + + # Load history + history: list[dict] = [] + if args.fetch_history: + token = os.environ.get("GH_PAT_FOR_NIGHTLY_CI_DATA") or os.environ.get( + "GITHUB_TOKEN" + ) + if token: + history = fetch_history_from_github(token) + else: + print("Warning: No GitHub token available, skipping history fetch") + elif args.history_dir: + history = load_history_from_dir(args.history_dir) + print(f"Loaded {len(history)} historical run(s) from {args.history_dir}") + + # Generate dashboard + markdown = generate_dashboard(current, history, charts_dir=args.charts_dir) + + # Write output + os.makedirs(os.path.dirname(args.output) or ".", exist_ok=True) + with open(args.output, "w") as f: + f.write(markdown) + print(f"Dashboard written to {args.output}") + + # Write to GitHub Step Summary + if args.step_summary: + summary_file = os.environ.get("GITHUB_STEP_SUMMARY") + if summary_file: + with open(summary_file, "a") as f: + f.write(markdown) + print("Dashboard appended to $GITHUB_STEP_SUMMARY") + else: + print("Warning: $GITHUB_STEP_SUMMARY not set, skipping") + + +if __name__ == "__main__": + main() diff --git a/scripts/ci/utils/diffusion/publish_comparison_results.py b/scripts/ci/utils/diffusion/publish_comparison_results.py new file mode 100644 index 000000000..e7cfbb8c1 --- /dev/null +++ b/scripts/ci/utils/diffusion/publish_comparison_results.py @@ -0,0 +1,217 @@ +"""Publish diffusion comparison results to sglang-bot/sglang-ci-data repo. + +Pushes comparison-results.json, dashboard.md, and chart PNG files to the +ci-data repository for historical tracking. Chart PNGs are stored under +diffusion-comparisons/charts/ so they can be referenced via +raw.githubusercontent URLs in the dashboard markdown (GitHub Step Summary +blocks data: URIs). + +Usage: + python3 scripts/ci/utils/diffusion/publish_comparison_results.py \ + --results comparison-results.json \ + --dashboard dashboard.md \ + --charts-dir comparison-charts/ +""" + +import argparse +import os +import sys +import time +from datetime import datetime, timezone +from pathlib import Path + +# Reuse GitHub API helpers from publish_traces +sys.path.insert(0, str(Path(__file__).parent)) +from ..publish_traces import ( + create_blobs, + create_commit, + create_tree, + get_branch_sha, + get_tree_sha, + is_permission_error, + is_rate_limit_error, + update_branch_ref, + verify_token_permissions, +) + +# Repository configuration +REPO_OWNER = "sglang-bot" +REPO_NAME = "sglang-ci-data" +BRANCH = "main" +STORAGE_PREFIX = "diffusion-comparisons" + + +def _collect_chart_files(charts_dir: str) -> list[tuple[str, bytes]]: + """Collect PNG chart files from directory for upload.""" + files: list[tuple[str, bytes]] = [] + if not charts_dir or not os.path.isdir(charts_dir): + return files + + for entry in sorted(os.listdir(charts_dir)): + if not entry.lower().endswith(".png"): + continue + full_path = os.path.join(charts_dir, entry) + if not os.path.isfile(full_path): + continue + with open(full_path, "rb") as f: + content = f.read() + # Store charts under diffusion-comparisons/charts/ + repo_path = f"{STORAGE_PREFIX}/charts/{entry}" + files.append((repo_path, content)) + + return files + + +def publish_comparison( + results_path: str, + dashboard_path: str | None = None, + charts_dir: str | None = None, +) -> None: + """Publish comparison results, dashboard, and charts to ci-data repo.""" + token = os.environ.get("GH_PAT_FOR_NIGHTLY_CI_DATA") or os.environ.get( + "GITHUB_TOKEN" + ) + if not token: + print("Error: GH_PAT_FOR_NIGHTLY_CI_DATA or GITHUB_TOKEN not set") + sys.exit(1) + + run_id = os.environ.get("GITHUB_RUN_ID", "local") + run_number = os.environ.get("GITHUB_RUN_NUMBER", "0") + + # Verify permissions + perm = verify_token_permissions(REPO_OWNER, REPO_NAME, token) + if perm == "rate_limited": + print("Warning: Rate limited, skipping publish") + return + elif not perm: + print("Error: Token permission verification failed") + sys.exit(1) + + # Prepare files to upload + files_to_upload: list[tuple[str, bytes]] = [] + + # Results JSON: stored with date prefix for chronological ordering + date_prefix = datetime.now(timezone.utc).strftime("%Y-%m-%d") + results_target = f"{STORAGE_PREFIX}/{date_prefix}_{run_id}.json" + with open(results_path, "rb") as f: + files_to_upload.append((results_target, f.read())) + + # Dashboard markdown: always overwrite latest + if dashboard_path and os.path.exists(dashboard_path): + dashboard_target = f"{STORAGE_PREFIX}/dashboard.md" + with open(dashboard_path, "rb") as f: + files_to_upload.append((dashboard_target, f.read())) + + # Chart PNG files + chart_files = _collect_chart_files(charts_dir) + if chart_files: + print(f"Found {len(chart_files)} chart PNG(s) to upload") + files_to_upload.extend(chart_files) + + print(f"Publishing {len(files_to_upload)} file(s) to {REPO_OWNER}/{REPO_NAME}") + + # Create blobs + try: + tree_items = create_blobs(REPO_OWNER, REPO_NAME, files_to_upload, token) + except Exception as e: + if is_rate_limit_error(e): + print("Warning: Rate limited during blob creation, skipping") + return + if is_permission_error(e): + print(f"Error: No write permission to {REPO_OWNER}/{REPO_NAME}") + sys.exit(1) + raise + + # Commit with retry (handle concurrent writes) + max_retries = 5 + retry_delay = 5 + + for attempt in range(max_retries): + try: + branch_sha = get_branch_sha(REPO_OWNER, REPO_NAME, BRANCH, token) + tree_sha = get_tree_sha(REPO_OWNER, REPO_NAME, branch_sha, token) + + new_tree_sha = create_tree( + REPO_OWNER, REPO_NAME, tree_sha, tree_items, token + ) + + commit_msg = ( + f"Diffusion comparison results for run {run_id} (#{run_number})" + ) + commit_sha = create_commit( + REPO_OWNER, REPO_NAME, new_tree_sha, branch_sha, commit_msg, token + ) + + update_branch_ref(REPO_OWNER, REPO_NAME, BRANCH, commit_sha, token) + print( + f"Successfully published comparison results (commit {commit_sha[:7]})" + ) + return + + except Exception as e: + is_retryable = False + if hasattr(e, "error_body"): + body = getattr(e, "error_body", "") + if "Update is not a fast forward" in body: + is_retryable = True + elif "Object does not exist" in body: + is_retryable = True + + from urllib.error import HTTPError + + if isinstance(e, HTTPError) and e.code in [422, 500, 502, 503, 504]: + is_retryable = True + + if is_rate_limit_error(e): + print("Warning: Rate limited, skipping publish") + return + + if is_permission_error(e): + print(f"Error: No write permission to {REPO_OWNER}/{REPO_NAME}") + sys.exit(1) + + if is_retryable and attempt < max_retries - 1: + print( + f"Attempt {attempt + 1}/{max_retries} failed, retrying in {retry_delay}s..." + ) + time.sleep(retry_delay) + else: + print(f"Failed to publish after {attempt + 1} attempts: {e}") + raise + + +def main(): + parser = argparse.ArgumentParser( + description="Publish diffusion comparison results to sglang-ci-data" + ) + parser.add_argument( + "--results", + required=True, + help="Path to comparison-results.json", + ) + parser.add_argument( + "--dashboard", + default=None, + help="Path to dashboard.md (optional)", + ) + parser.add_argument( + "--charts-dir", + default=None, + help="Directory containing chart PNG files to upload (optional)", + ) + + args = parser.parse_args() + + if not os.path.exists(args.results): + print(f"Error: Results file not found: {args.results}") + sys.exit(1) + + publish_comparison( + results_path=args.results, + dashboard_path=args.dashboard, + charts_dir=args.charts_dir, + ) + + +if __name__ == "__main__": + main() diff --git a/scripts/ci/utils/publish_diffusion_gt.py b/scripts/ci/utils/diffusion/publish_diffusion_gt.py similarity index 99% rename from scripts/ci/utils/publish_diffusion_gt.py rename to scripts/ci/utils/diffusion/publish_diffusion_gt.py index c77dbcd93..eda5de09f 100644 --- a/scripts/ci/utils/publish_diffusion_gt.py +++ b/scripts/ci/utils/diffusion/publish_diffusion_gt.py @@ -10,7 +10,7 @@ import sys # Allow importing from the same directory (scripts/ci/utils/) sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) -from publish_traces import ( +from ..publish_traces import ( create_blobs, create_commit, create_tree, diff --git a/scripts/ci/utils/diffusion/run_comparison.py b/scripts/ci/utils/diffusion/run_comparison.py new file mode 100644 index 000000000..d24b64b52 --- /dev/null +++ b/scripts/ci/utils/diffusion/run_comparison.py @@ -0,0 +1,951 @@ +"""Cross-framework comparison benchmark for diffusion serving. + +Launches servers (SGLang, vLLM-Omni, LightX2V) for each test case, sends a +single request, measures end-to-end latency, and writes comparison-results.json. + +Usage: + # Full run (requires GPU) + python3 scripts/ci/utils/diffusion/run_comparison.py + + # Dry-run (config parsing + command preview only) + python3 scripts/ci/utils/diffusion/run_comparison.py --dry-run + + # Run only specific case(s) + python3 scripts/ci/utils/diffusion/run_comparison.py --case-ids flux1_dev_t2i_1024 + + # Run only specific framework(s) + python3 scripts/ci/utils/diffusion/run_comparison.py --frameworks sglang +""" + +import argparse +import base64 +import io +import json +import os +import signal +import subprocess +import sys +import tempfile +import threading +import time +from datetime import datetime, timezone +from pathlib import Path + +import requests + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- +CONFIGS_PATH = Path(__file__).parent / "comparison_configs.json" +INSTALL_SCRIPT = Path(__file__).parents[1] / "install_comparison_frameworks.sh" +DEFAULT_HOST = "127.0.0.1" +DEFAULT_PORT = 30000 +HEALTH_TIMEOUT = ( + 2400 # seconds (40 min — FLUX.2-dev needs ~10 min download + torch.compile) +) +REQUEST_TIMEOUT = 1200 # seconds +GPU_CLEAR_WAIT = 15 # seconds between framework runs + +# Frameworks that need separate installation (conflict with sglang's deps) +INSTALLABLE_FRAMEWORKS = {"vllm-omni", "lightx2v"} + +# Cached reference image (downloaded once) +_cached_ref_image: bytes | None = None +_cached_ref_image_path: str | None = None + + +# --------------------------------------------------------------------------- +# Server lifecycle — command builders +# --------------------------------------------------------------------------- + + +def _build_sglang_cmd(case: dict, fw_cfg: dict, port: int) -> list[str]: + cmd = [ + "sglang", + "serve", + "--model-path", + case["model"], + "--port", + str(port), + "--host", + DEFAULT_HOST, + ] + if case["num_gpus"] > 1: + cmd += ["--num-gpus", str(case["num_gpus"])] + if fw_cfg.get("serve_args", "").strip(): + cmd += fw_cfg["serve_args"].strip().split() + return cmd + + +def _build_vllm_cmd(case: dict, fw_cfg: dict, port: int) -> list[str]: + cmd = [ + "vllm", + "serve", + case["model"], + "--omni", + "--port", + str(port), + "--host", + DEFAULT_HOST, + ] + if fw_cfg.get("serve_args", "").strip(): + cmd += fw_cfg["serve_args"].strip().split() + return cmd + + +def _resolve_hf_model_path(model_id: str) -> str: + """Resolve a HuggingFace model ID to a local cache path, or return as-is.""" + if os.path.isdir(model_id): + return model_id + try: + from huggingface_hub import snapshot_download + + path = snapshot_download(model_id) + print(f" Resolved {model_id} -> {path}") + return path + except Exception: + return model_id + + +def _write_lightx2v_config(case: dict) -> str: + """Write a minimal LightX2V config JSON and return its path.""" + cfg = { + "infer_steps": case.get("num_inference_steps", 50), + "guidance_scale": case.get("guidance_scale", 4.0), + "seed": case.get("seed", 42), + } + if "num_frames" in case: + cfg["target_video_length"] = case["num_frames"] + if "height" in case: + cfg["height"] = case["height"] + if "width" in case: + cfg["width"] = case["width"] + + config_path = os.path.join( + tempfile.gettempdir(), f"lightx2v_config_{case['id']}.json" + ) + with open(config_path, "w") as f: + json.dump(cfg, f) + return config_path + + +def _build_lightx2v_cmd(case: dict, fw_cfg: dict, port: int) -> list[str]: + """Build LightX2V server launch command. + + Single GPU: python -m lightx2v.server --model_path ... --model_cls ... --task ... --port ... + Multi GPU: torchrun --nproc_per_node=N -m lightx2v.server ... + + LightX2V requires a local model path and a config JSON with infer params. + """ + model_cls = fw_cfg["model_cls"] + task = fw_cfg["lightx2v_task"] + num_gpus = case["num_gpus"] + model_path = _resolve_hf_model_path(case["model"]) + config_path = _write_lightx2v_config(case) + + server_args = [ + "--model_path", + model_path, + "--model_cls", + model_cls, + "--task", + task, + "--config_json", + config_path, + "--host", + DEFAULT_HOST, + "--port", + str(port), + ] + if fw_cfg.get("serve_args", "").strip(): + server_args += fw_cfg["serve_args"].strip().split() + + if num_gpus > 1: + cmd = [ + "torchrun", + f"--nproc_per_node={num_gpus}", + "-m", + "lightx2v.server", + ] + server_args + else: + cmd = ["python3", "-m", "lightx2v.server"] + server_args + + return cmd + + +def build_server_cmd(framework: str, case: dict, fw_cfg: dict, port: int) -> list[str]: + builders = { + "sglang": _build_sglang_cmd, + "vllm-omni": _build_vllm_cmd, + "lightx2v": _build_lightx2v_cmd, + } + builder = builders.get(framework) + if builder is None: + raise ValueError(f"Unknown framework: {framework}") + return builder(case, fw_cfg, port) + + +# --------------------------------------------------------------------------- +# Server lifecycle — health check & cleanup +# --------------------------------------------------------------------------- + +# Health check endpoints per framework +HEALTH_ENDPOINTS = { + "sglang": "/health", + "vllm-omni": "/health", + "lightx2v": "/v1/service/status", +} + + +def wait_for_health( + base_url: str, framework: str = "sglang", timeout: int = HEALTH_TIMEOUT +) -> None: + """Poll health endpoint until 200, then verify model is loaded.""" + endpoint = HEALTH_ENDPOINTS.get(framework, "/health") + health_url = f"{base_url}{endpoint}" + print(f" Waiting for server at {health_url} ...") + start = time.time() + while True: + try: + resp = requests.get(health_url, timeout=2) + if resp.status_code == 200: + break + except requests.exceptions.RequestException: + pass + if time.time() - start > timeout: + raise TimeoutError( + f"Server at {health_url} did not start within {timeout}s" + ) + time.sleep(2) + + # For SGLang, /health can return 200 before model routes are registered. + # Poll /v1/models to confirm the model is fully loaded. + if framework == "sglang": + models_url = f"{base_url}/v1/models" + while True: + try: + resp = requests.get(models_url, timeout=5) + if resp.status_code == 200: + break + except requests.exceptions.RequestException: + pass + if time.time() - start > timeout: + raise TimeoutError(f"Model at {models_url} not ready within {timeout}s") + time.sleep(2) + + elapsed = time.time() - start + print(f" Server ready in {elapsed:.1f}s") + + +KILLALL_SCRIPT = Path(__file__).parents[3] / "killall_sglang.sh" + + +def kill_server(proc: subprocess.Popen) -> None: + """Kill server process tree and clean up GPU processes.""" + if proc.poll() is not None: + return + try: + os.killpg(os.getpgid(proc.pid), signal.SIGTERM) + except (ProcessLookupError, PermissionError): + pass + try: + proc.wait(timeout=30) + except subprocess.TimeoutExpired: + try: + os.killpg(os.getpgid(proc.pid), signal.SIGKILL) + except (ProcessLookupError, PermissionError): + pass + proc.wait(timeout=10) + # Use killall_sglang.sh for thorough cleanup (esp. multi-GPU workers) + if KILLALL_SCRIPT.exists(): + subprocess.run( + ["bash", str(KILLALL_SCRIPT)], + timeout=30, + capture_output=True, + ) + + +# --------------------------------------------------------------------------- +# Reference image helpers +# --------------------------------------------------------------------------- + + +def _get_ref_image_bytes(config: dict) -> bytes: + """Download and cache the shared test reference image.""" + global _cached_ref_image + if _cached_ref_image is not None: + return _cached_ref_image + url = config.get("test_image_url", "") + if not url: + raise RuntimeError("No test_image_url in config for image-conditioned case") + print(f" Downloading reference image from {url} ...") + resp = requests.get(url, timeout=60) + resp.raise_for_status() + _cached_ref_image = resp.content + return _cached_ref_image + + +def _get_ref_image_b64(config: dict) -> str: + """Get reference image as base64 string.""" + return base64.b64encode(_get_ref_image_bytes(config)).decode("utf-8") + + +def _get_ref_image_path(config: dict) -> str: + """Save reference image to a temp file and return path.""" + global _cached_ref_image_path + if _cached_ref_image_path and os.path.exists(_cached_ref_image_path): + return _cached_ref_image_path + data = _get_ref_image_bytes(config) + fd, path = tempfile.mkstemp(suffix=".png") + with os.fdopen(fd, "wb") as f: + f.write(data) + _cached_ref_image_path = path + return path + + +# --------------------------------------------------------------------------- +# Request helpers — SGLang (OpenAI-compatible) +# --------------------------------------------------------------------------- + + +def _build_sglang_payload(case: dict) -> dict: + """Build common SGLang request payload.""" + payload = { + "model": case["model"], + "prompt": case["prompt"], + "size": f"{case['width']}x{case['height']}", + "n": 1, + "response_format": "b64_json", + } + for key in ("num_inference_steps", "guidance_scale", "seed", "num_frames"): + if key in case: + payload[key] = case[key] + return payload + + +def _read_perf_dump(perf_dump_path: str, timeout: float = 10.0) -> float | None: + """Read total_duration_ms from a perf dump JSON written by the server. + + The server writes the file asynchronously after the HTTP response, + so we poll briefly. + """ + deadline = time.time() + timeout + while time.time() < deadline: + try: + with open(perf_dump_path) as f: + data = json.load(f) + total_ms = data.get("total_duration_ms") + if total_ms is not None: + return total_ms / 1000.0 + except (FileNotFoundError, json.JSONDecodeError): + pass + time.sleep(0.5) + return None + + +def send_image_request_sglang( + base_url: str, case: dict, perf_dump_path: str | None = None +) -> float: + """Send a single T2I request via SGLang's /v1/images/generations.""" + payload = _build_sglang_payload(case) + if perf_dump_path: + payload["perf_dump_path"] = perf_dump_path + + start = time.time() + resp = requests.post( + f"{base_url}/v1/images/generations", + json=payload, + timeout=REQUEST_TIMEOUT, + ) + client_latency = time.time() - start + resp.raise_for_status() + data = resp.json() + if "data" not in data or len(data["data"]) == 0: + raise RuntimeError(f"Image request returned no data: {data}") + + if perf_dump_path: + server_latency = _read_perf_dump(perf_dump_path) + if server_latency is not None: + print( + f" Image generated in {server_latency:.2f}s (server-side), " + f"client={client_latency:.2f}s" + ) + return server_latency + print(f" Image generated in {client_latency:.2f}s") + return client_latency + + +def send_video_request_sglang( + base_url: str, case: dict, perf_dump_path: str | None = None +) -> float: + """Send a single T2V request via SGLang's /v1/videos (async).""" + payload = _build_sglang_payload(case) + if perf_dump_path: + payload["perf_dump_path"] = perf_dump_path + + start = time.time() + + # Submit job + resp = requests.post( + f"{base_url}/v1/videos", + json=payload, + timeout=REQUEST_TIMEOUT, + ) + resp.raise_for_status() + job = resp.json() + job_id = job.get("id") + if not job_id: + raise RuntimeError(f"Video submit returned no job id: {job}") + + # Poll for completion + poll_url = f"{base_url}/v1/videos/{job_id}" + while True: + time.sleep(1) + poll_resp = requests.get(poll_url, timeout=30) + poll_resp.raise_for_status() + poll_data = poll_resp.json() + status = poll_data.get("status") + if status == "completed": + break + elif status == "failed": + raise RuntimeError(f"Video generation failed: {poll_data}") + if time.time() - start > REQUEST_TIMEOUT: + raise TimeoutError(f"Video generation timed out after {REQUEST_TIMEOUT}s") + + client_latency = time.time() - start + + if perf_dump_path: + server_latency = _read_perf_dump(perf_dump_path) + if server_latency is not None: + print( + f" Video generated in {server_latency:.2f}s (server-side), " + f"client={client_latency:.2f}s" + ) + return server_latency + print(f" Video generated in {client_latency:.2f}s") + return client_latency + + +def send_image_conditioned_request_sglang( + base_url: str, case: dict, config: dict, perf_dump_path: str | None = None +) -> float: + """Send an image-conditioned request (edit/I2V/TI2V) via SGLang multipart API.""" + task = case["task"] + ref_bytes = _get_ref_image_bytes(config) + + # Build multipart form — field name depends on endpoint: + # image edits use "image", video (I2V/TI2V) uses "input_reference" + if task in ("image-to-video", "text-image-to-video"): + file_field = "input_reference" + else: + file_field = "image" + files = {file_field: ("ref.png", io.BytesIO(ref_bytes), "image/png")} + data = { + "model": case["model"], + "prompt": case["prompt"], + "size": f"{case['width']}x{case['height']}", + "n": "1", + "response_format": "b64_json", + } + for key in ("num_inference_steps", "guidance_scale", "seed", "num_frames"): + if key in case: + data[key] = str(case[key]) + if perf_dump_path: + data["perf_dump_path"] = perf_dump_path + # Choose endpoint based on task + if task in ("image-edit", "image-to-image"): + endpoint = "/v1/images/edits" + elif task in ("image-to-video", "text-image-to-video"): + endpoint = "/v1/videos" + else: + endpoint = "/v1/images/generations" + + start = time.time() + resp = requests.post( + f"{base_url}{endpoint}", + files=files, + data=data, + timeout=REQUEST_TIMEOUT, + ) + + # For video endpoints, need to poll + if task in ("image-to-video", "text-image-to-video"): + resp.raise_for_status() + job = resp.json() + job_id = job.get("id") + if not job_id: + raise RuntimeError(f"Video submit returned no job id: {job}") + poll_url = f"{base_url}/v1/videos/{job_id}" + while True: + time.sleep(1) + poll_resp = requests.get(poll_url, timeout=30) + poll_resp.raise_for_status() + poll_data = poll_resp.json() + status = poll_data.get("status") + if status == "completed": + break + elif status == "failed": + raise RuntimeError(f"Video generation failed: {poll_data}") + if time.time() - start > REQUEST_TIMEOUT: + raise TimeoutError(f"Timed out after {REQUEST_TIMEOUT}s") + else: + resp.raise_for_status() + + client_latency = time.time() - start + + if perf_dump_path: + server_latency = _read_perf_dump(perf_dump_path) + if server_latency is not None: + print( + f" Generated in {server_latency:.2f}s (server-side), " + f"client={client_latency:.2f}s" + ) + return server_latency + print(f" Generated in {client_latency:.2f}s (sglang, image-conditioned)") + return client_latency + + +# --------------------------------------------------------------------------- +# Request helpers — vLLM-Omni +# --------------------------------------------------------------------------- + + +def send_request_vllm_omni(base_url: str, case: dict, config: dict) -> float: + """Send request via vLLM-Omni's /v1/chat/completions endpoint.""" + extra_body = { + "height": case["height"], + "width": case["width"], + "num_inference_steps": case.get("num_inference_steps", 50), + "guidance_scale": case.get("guidance_scale", 4.0), + "seed": case.get("seed", 42), + } + if "num_frames" in case: + extra_body["num_frames"] = case["num_frames"] + + # Build message content (text or text+image) + content: list[dict] | str = case["prompt"] + if case.get("reference_image"): + ref_b64 = _get_ref_image_b64(config) + content = [ + { + "type": "image_url", + "image_url": {"url": f"data:image/png;base64,{ref_b64}"}, + }, + {"type": "text", "text": case["prompt"]}, + ] + + payload = { + "model": case["model"], + "messages": [{"role": "user", "content": content}], + "extra_body": extra_body, + } + + start = time.time() + resp = requests.post( + f"{base_url}/v1/chat/completions", + json=payload, + timeout=REQUEST_TIMEOUT, + ) + latency = time.time() - start + resp.raise_for_status() + data = resp.json() + choices = data.get("choices", []) + if not choices: + raise RuntimeError(f"vLLM-Omni request returned no choices: {data}") + print(f" Generated in {latency:.2f}s (vllm-omni)") + return latency + + +# --------------------------------------------------------------------------- +# Request helpers — LightX2V +# --------------------------------------------------------------------------- + + +def send_request_lightx2v(base_url: str, case: dict, config: dict) -> float: + """Send request via LightX2V's async task API.""" + task = case["task"] + if task in ("text-to-image", "image-edit"): + endpoint = "/v1/tasks/image" + else: + endpoint = "/v1/tasks/video" + + payload = { + "prompt": case["prompt"], + "seed": case.get("seed", 42), + "infer_steps": case.get("num_inference_steps", 50), + } + # LightX2V uses target_video_length for frames, height/width directly + if "num_frames" in case: + payload["target_video_length"] = case["num_frames"] + if "height" in case: + payload["height"] = case["height"] + if "width" in case: + payload["width"] = case["width"] + if "guidance_scale" in case: + payload["guidance_scale"] = case["guidance_scale"] + # Image-conditioned: LightX2V accepts image_path (URL or local path) + if case.get("reference_image"): + payload["image_path"] = config.get("test_image_url", "") + + start = time.time() + + # Submit task + resp = requests.post( + f"{base_url}{endpoint}", + json=payload, + timeout=REQUEST_TIMEOUT, + ) + resp.raise_for_status() + task_data = resp.json() + task_id = task_data.get("task_id") + if not task_id: + raise RuntimeError(f"LightX2V submit returned no task_id: {task_data}") + + # Poll for completion + poll_url = f"{base_url}/v1/tasks/{task_id}/status" + while True: + time.sleep(1) + poll_resp = requests.get(poll_url, timeout=30) + poll_resp.raise_for_status() + poll_data = poll_resp.json() + status = poll_data.get("task_status", "").upper() + if status == "COMPLETED": + break + elif status in ("FAILED", "CANCELLED"): + raise RuntimeError(f"LightX2V task {status}: {poll_data}") + if time.time() - start > REQUEST_TIMEOUT: + raise TimeoutError(f"LightX2V task timed out after {REQUEST_TIMEOUT}s") + + latency = time.time() - start + print(f" Generated in {latency:.2f}s (lightx2v)") + return latency + + +# --------------------------------------------------------------------------- +# Unified request dispatcher +# --------------------------------------------------------------------------- + + +def send_request( + base_url: str, + case: dict, + framework: str = "sglang", + config: dict | None = None, + perf_dump_path: str | None = None, +) -> float: + config = config or {} + if framework == "vllm-omni": + return send_request_vllm_omni(base_url, case, config) + elif framework == "lightx2v": + return send_request_lightx2v(base_url, case, config) + # SGLang — use OpenAI-compatible endpoints with optional perf log + task = case["task"] + if case.get("reference_image"): + return send_image_conditioned_request_sglang( + base_url, case, config, perf_dump_path + ) + elif task == "text-to-image": + return send_image_request_sglang(base_url, case, perf_dump_path) + elif task == "text-to-video": + return send_video_request_sglang(base_url, case, perf_dump_path) + else: + raise ValueError(f"Unknown task type: {task}") + + +# --------------------------------------------------------------------------- +# Main orchestrator +# --------------------------------------------------------------------------- + + +def run_single( + case: dict, + framework: str, + fw_cfg: dict, + port: int, + log_dir: Path, + config: dict | None = None, +) -> dict: + """Run a single (case, framework) combination. Returns result dict.""" + result = { + "case_id": case["id"], + "framework": framework, + "model": case["model"], + "task": case["task"], + "latency_s": None, + "error": None, + } + + cmd = build_server_cmd(framework, case, fw_cfg, port) + print(f"\n Command: {' '.join(cmd)}") + + env = os.environ.copy() + env.update(fw_cfg.get("extra_env", {})) + + # perf_dump_path for SGLang server-side timing (passed in request, zero overhead when None) + perf_dump_path = None + if framework == "sglang": + perf_dump_path = os.path.join(str(log_dir), f"perf_{case['id']}_measured.json") + + log_file = log_dir / f"{case['id']}_{framework}.log" + log_fh = open(log_file, "w", encoding="utf-8", buffering=1) + log_thread = None + + proc = None + try: + proc = subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + env=env, + preexec_fn=os.setsid, + text=True, + bufsize=1, + ) + + # Tee server output to both log file and stdout (like test_server_utils) + def _log_pipe(pipe, fh): + try: + for line in iter(pipe.readline, ""): + sys.stdout.write(f" [server] {line}") + sys.stdout.flush() + fh.write(line) + except ValueError: + pass # pipe closed + + log_thread = threading.Thread(target=_log_pipe, args=(proc.stdout, log_fh)) + log_thread.daemon = True + log_thread.start() + + base_url = f"http://{DEFAULT_HOST}:{port}" + wait_for_health(base_url, framework) + + # Warmup requests (not measured, no perf dump) + # Use few steps to be fast — server's own warmup (warmup_steps=3) handles + # torch.compile compilation; these external warmups just stabilize triton + # kernel specializations across requests. + WARMUP_STEPS = 3 + warmup_case = {**case, "num_inference_steps": WARMUP_STEPS} + for wi in range(1, 3): + print(f" Sending warmup request ({wi}/2, {WARMUP_STEPS} steps)...") + try: + send_request(base_url, warmup_case, framework, config) + except Exception as e: + print(f" Warmup request {wi} failed (non-fatal): {e}") + + # Measured request — pass perf_dump_path for SGLang server-side timing + if perf_dump_path and os.path.exists(perf_dump_path): + os.remove(perf_dump_path) + print(" Sending measured request...") + latency = send_request( + base_url, case, framework, config, perf_dump_path=perf_dump_path + ) + result["latency_s"] = round(latency, 3) + + except Exception as e: + result["error"] = str(e) + print(f" ERROR: {e}") + finally: + if proc: + kill_server(proc) + if log_thread: + log_thread.join(timeout=5) + log_fh.close() + + return result + + +def _install_framework(fw_name: str, dry_run: bool = False) -> bool: + """Install a comparison framework via the install script. Returns True on success.""" + if fw_name not in INSTALLABLE_FRAMEWORKS: + return True + if not INSTALL_SCRIPT.exists(): + print(f" WARNING: Install script not found at {INSTALL_SCRIPT}") + return False + if dry_run: + print(f" [DRY-RUN] Would install: bash {INSTALL_SCRIPT} {fw_name}") + return True + print(f"\n{'='*60}") + print(f"Installing framework: {fw_name}") + print(f"{'='*60}") + ret = subprocess.run( + ["bash", str(INSTALL_SCRIPT), fw_name], + timeout=600, + ) + if ret.returncode != 0: + print(f" WARNING: {fw_name} installation failed (exit {ret.returncode})") + return False + return True + + +def run_comparison( + config: dict, + case_ids: list[str] | None = None, + frameworks: list[str] | None = None, + port: int = DEFAULT_PORT, + output: str = "comparison-results.json", + dry_run: bool = False, +) -> dict: + """Run all comparison cases, grouped by framework to minimize installs. + + Order: sglang first (already installed), then vllm-omni, then lightx2v. + Each non-sglang framework is installed right before its cases run. + """ + timestamp = datetime.now(timezone.utc).isoformat() + commit_sha = os.environ.get("GITHUB_SHA", "unknown") + run_id = os.environ.get("GITHUB_RUN_ID", "local") + + log_dir = Path("comparison-logs") + log_dir.mkdir(exist_ok=True) + + # Collect all (case, framework) pairs, grouped by framework + fw_order = ["sglang", "vllm-omni", "lightx2v"] + fw_cases: dict[str, list[tuple[dict, dict]]] = {fw: [] for fw in fw_order} + + for case in config["cases"]: + if case_ids and case["id"] not in case_ids: + continue + for fw_name, fw_cfg in case["frameworks"].items(): + if frameworks and fw_name not in frameworks: + continue + if fw_name not in fw_cases: + fw_cases[fw_name] = [] + fw_cases[fw_name].append((case, fw_cfg)) + + results = [] + installed_fws: set[str] = set() + + for fw_name in fw_order: + pairs = fw_cases.get(fw_name, []) + if not pairs: + continue + + # Install framework if needed (once per framework) + if fw_name not in installed_fws and fw_name in INSTALLABLE_FRAMEWORKS: + if not _install_framework(fw_name, dry_run): + # Skip all cases for this framework + for case, _ in pairs: + results.append( + { + "case_id": case["id"], + "framework": fw_name, + "model": case["model"], + "task": case["task"], + "latency_s": None, + "error": f"{fw_name} installation failed", + } + ) + continue + installed_fws.add(fw_name) + + for case, fw_cfg in pairs: + print(f"\n{'='*60}") + print(f"Case: {case['id']} | Model: {case['model']} | Framework: {fw_name}") + print(f"{'='*60}") + + if dry_run: + cmd = build_server_cmd(fw_name, case, fw_cfg, port) + print(f" [DRY-RUN] Would run: {' '.join(cmd)}") + results.append( + { + "case_id": case["id"], + "framework": fw_name, + "model": case["model"], + "task": case["task"], + "latency_s": None, + "error": "dry-run", + } + ) + continue + + result = run_single(case, fw_name, fw_cfg, port, log_dir, config) + results.append(result) + + # Wait for GPU memory to clear + print(f" Waiting {GPU_CLEAR_WAIT}s for GPU memory to clear...") + time.sleep(GPU_CLEAR_WAIT) + + output_data = { + "timestamp": timestamp, + "commit_sha": commit_sha, + "run_id": run_id, + "results": results, + } + + os.makedirs(os.path.dirname(output) or ".", exist_ok=True) + with open(output, "w") as f: + json.dump(output_data, f, indent=2) + print(f"\nResults written to {output}") + + # Print summary table + print(f"\n{'='*60}") + print("SUMMARY") + print(f"{'='*60}") + for r in results: + lat = f"{r['latency_s']:.2f}s" if r["latency_s"] else r.get("error", "N/A") + print(f" {r['case_id']:30s} | {r['framework']:12s} | {lat}") + + return output_data + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def main(): + parser = argparse.ArgumentParser( + description="Cross-framework diffusion serving comparison benchmark" + ) + parser.add_argument( + "--config", + default=str(CONFIGS_PATH), + help="Path to comparison_configs.json", + ) + parser.add_argument( + "--case-ids", + nargs="+", + default=None, + help="Only run specific case IDs", + ) + parser.add_argument( + "--frameworks", + nargs="+", + default=None, + help="Only run specific frameworks (sglang, vllm-omni, lightx2v)", + ) + parser.add_argument( + "--port", + type=int, + default=DEFAULT_PORT, + help="Server port", + ) + parser.add_argument( + "--output", + default="comparison-results.json", + help="Output JSON path", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Parse config and print commands without launching servers", + ) + + args = parser.parse_args() + + with open(args.config) as f: + config = json.load(f) + + print(f"Loaded {len(config['cases'])} comparison case(s) from {args.config}") + + run_comparison( + config=config, + case_ids=args.case_ids, + frameworks=args.frameworks, + port=args.port, + output=args.output, + dry_run=args.dry_run, + ) + + +if __name__ == "__main__": + main() diff --git a/scripts/ci/utils/save_diffusion_metrics.py b/scripts/ci/utils/diffusion/save_diffusion_metrics.py similarity index 98% rename from scripts/ci/utils/save_diffusion_metrics.py rename to scripts/ci/utils/diffusion/save_diffusion_metrics.py index 6df70c732..60bdc0825 100755 --- a/scripts/ci/utils/save_diffusion_metrics.py +++ b/scripts/ci/utils/diffusion/save_diffusion_metrics.py @@ -5,7 +5,7 @@ This script reads diffusion test results from the pytest stash and saves them with metadata for the performance dashboard. Usage: - python3 scripts/ci/utils/save_diffusion_metrics.py \ + python3 scripts/ci/utils/diffusion/save_diffusion_metrics.py \ --gpu-config 1-gpu-h100 \ --run-id 12345678 \ --output test/diffusion-metrics-1gpu.json \ diff --git a/scripts/ci/utils/merge_metrics.py b/scripts/ci/utils/merge_metrics.py index 5612fbc9e..128dc5961 100755 --- a/scripts/ci/utils/merge_metrics.py +++ b/scripts/ci/utils/merge_metrics.py @@ -25,6 +25,7 @@ def find_partition_files(input_dir: str) -> list[str]: patterns = [ os.path.join(input_dir, "**/metrics-*.json"), os.path.join(input_dir, "**/diffusion-metrics-*.json"), + os.path.join(input_dir, "**/comparison-metrics-*.json"), ] files = set() for pattern in patterns: