diff --git a/.github/workflows/pr-test.yml b/.github/workflows/pr-test.yml index 49924a444..b67269e22 100644 --- a/.github/workflows/pr-test.yml +++ b/.github/workflows/pr-test.yml @@ -308,14 +308,34 @@ jobs: multimodal-gen-test: needs: [check-changes] if: needs.check-changes.outputs.multimodal_gen == 'true' - runs-on: ubuntu-latest + runs-on: 1-gpu-runner steps: - name: Checkout code uses: actions/checkout@v4 - - name: Run placeholder test + + - name: Download CUDA 12.9 kernel artifacts + if: needs.check-changes.outputs.sgl_kernel == 'true' + uses: actions/download-artifact@v4 + with: + path: sgl-kernel/dist/ + merge-multiple: true + pattern: wheel-python3.10-cuda12.9 + + - name: Install dependencies run: | - echo "Running multimodal_gen tests..." - echo "This is a placeholder for future tests." + CUSTOM_BUILD_SGL_KERNEL=${{needs.check-changes.outputs.sgl_kernel}} bash scripts/ci/ci_install_dependency.sh diffusion + + - name: Clean Corrupted Hugging Face Model Cache + run: | + echo "Temp: Deleting potentially corrupted Qwen/Qwen-Image and Qwen/Qwen-Image-Edit cache to ensure a fresh download." + rm -rf /hf_home/hub/models--Qwen--Qwen-Image + rm -rf /hf_home/hub/models--Qwen--Qwen-Image-Edit + + - name: Run diffusion server tests + timeout-minutes: 60 + run: | + cd python + pytest -s -v --log-cli-level=INFO sglang/multimodal_gen/test/server/test_server_performance.py # Adding a single CUDA13 smoke test to verify that the kernel builds and runs # TODO: Add back this test when it can pass on CI diff --git a/python/sglang/multimodal_gen/runtime/managers/gpu_worker.py b/python/sglang/multimodal_gen/runtime/managers/gpu_worker.py index 3338482ff..7d2152846 100644 --- a/python/sglang/multimodal_gen/runtime/managers/gpu_worker.py +++ b/python/sglang/multimodal_gen/runtime/managers/gpu_worker.py @@ -28,7 +28,6 @@ from sglang.multimodal_gen.runtime.utils.logging_utils import ( logger = init_logger(__name__) -# ANSI color codes CYAN = "\033[1;36m" RESET = "\033[0;0m" @@ -96,6 +95,16 @@ class GPUWorker: req = batch[0] output_batch = self.pipeline.forward(req, server_args) if req.perf_logger: + logging_info = getattr(output_batch, "logging_info", None) or getattr( + req, "logging_info", None + ) + if logging_info: + try: + req.perf_logger.log_stage_metrics(logging_info) + except Exception: + logger.exception( + "Failed to log stage metrics for request %s", req.request_id + ) req.perf_logger.log_total_duration("total_inference_time") return output_batch diff --git a/python/sglang/multimodal_gen/runtime/models/encoders/qwen2_5vl.py b/python/sglang/multimodal_gen/runtime/models/encoders/qwen2_5vl.py index 08184cccb..ebac284a7 100644 --- a/python/sglang/multimodal_gen/runtime/models/encoders/qwen2_5vl.py +++ b/python/sglang/multimodal_gen/runtime/models/encoders/qwen2_5vl.py @@ -54,7 +54,13 @@ from sglang.multimodal_gen.runtime.utils.common import add_prefix # limitations under the License. """Inference-only Qwen2-VL model compatible with HuggingFace weights.""" import logging -from typing import Callable, Iterable, Optional, Tuple, Union, Unpack +from typing import Callable, Iterable, Optional, Tuple, Union + +try: + from typing import Unpack # type: ignore[attr-defined] +except ImportError: + # Python 3.10 and below + from typing_extensions import Unpack import torch import torch.nn as nn diff --git a/python/sglang/multimodal_gen/runtime/pipelines/schedule_batch.py b/python/sglang/multimodal_gen/runtime/pipelines/schedule_batch.py index 8d6ba6dcc..06dded31b 100644 --- a/python/sglang/multimodal_gen/runtime/pipelines/schedule_batch.py +++ b/python/sglang/multimodal_gen/runtime/pipelines/schedule_batch.py @@ -193,6 +193,9 @@ class Req: VSA_sparsity: float = 0.0 perf_logger: PerformanceLogger | None = None + # stage logging + logging_info: PipelineLoggingInfo = field(default_factory=PipelineLoggingInfo) + # profile profile: bool = False num_profiled_timesteps: int = 8 diff --git a/python/sglang/multimodal_gen/runtime/pipelines/stages/base.py b/python/sglang/multimodal_gen/runtime/pipelines/stages/base.py index 6eeed5f78..1e6671270 100644 --- a/python/sglang/multimodal_gen/runtime/pipelines/stages/base.py +++ b/python/sglang/multimodal_gen/runtime/pipelines/stages/base.py @@ -185,6 +185,8 @@ class PipelineStage(ABC): raise # Execute the actual stage logic + logging_info = getattr(batch, "logging_info", None) + if envs.SGL_DIFFUSION_STAGE_LOGGING: logger.info("[%s] Starting execution", stage_name) start_time = time.perf_counter() @@ -197,7 +199,27 @@ class PipelineStage(ABC): stage_name, execution_time * 1000, ) - batch.logging_info.add_stage_execution_time(stage_name, execution_time) + if logging_info is not None: + try: + logging_info.add_stage_execution_time( + stage_name, execution_time + ) + except Exception: + logger.warning( + "[%s] Failed to record stage timing on batch.logging_info", + stage_name, + exc_info=True, + ) + perf_logger = getattr(batch, "perf_logger", None) + if perf_logger is not None: + try: + perf_logger.log_stage_metric(stage_name, execution_time * 1000) + except Exception: + logger.warning( + "[%s] Failed to log stage metric to performance logger", + stage_name, + exc_info=True, + ) except Exception as e: execution_time = time.perf_counter() - start_time logger.error( diff --git a/python/sglang/multimodal_gen/runtime/utils/performance_logger.py b/python/sglang/multimodal_gen/runtime/utils/performance_logger.py index fb4f4e399..28480fe0f 100644 --- a/python/sglang/multimodal_gen/runtime/utils/performance_logger.py +++ b/python/sglang/multimodal_gen/runtime/utils/performance_logger.py @@ -6,11 +6,16 @@ import os import subprocess import time from datetime import datetime +from typing import Any from dateutil.tz import UTC -project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../")) -LOG_DIR = os.path.join(project_root, "logs") +LOG_DIR = os.environ.get("SGLANG_PERF_LOG_DIR") +if LOG_DIR: + LOG_DIR = os.path.abspath(LOG_DIR) +else: + project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../")) + LOG_DIR = os.path.join(project_root, "logs") # Configure a specific logger for performance metrics perf_logger = logging.getLogger("performance") @@ -74,3 +79,59 @@ class PerformanceLogger: "steps": self.step_timings, } perf_logger.info(json.dumps(log_entry)) + + def log_stage_metric(self, stage_name: str, duration_ms: float): + """Logs a single pipeline stage timing entry.""" + log_entry = { + "timestamp": datetime.now(UTC).isoformat(), + "request_id": self.request_id, + "commit_hash": self.commit_hash, + "tag": "pipeline_stage_metric", + "stage": stage_name, + "duration_ms": duration_ms, + } + perf_logger.info(json.dumps(log_entry)) + + def log_stage_metrics(self, stages: Any): + """ + Persist per-stage execution stats to performance.log. + + Args: + stages: Either a PipelineLoggingInfo instance or any object exposing + a mapping of stage metadata via a `stages` attribute/dict. + """ + if stages is None: + return + + if hasattr(stages, "stages"): + stage_items = getattr(stages, "stages", {}).items() + elif isinstance(stages, dict): + stage_items = stages.items() + else: + return + + formatted_stages: list[dict[str, Any]] = [] + for name, info in stage_items: + if not info: + continue + entry = {"name": name} + execution_time = info.get("execution_time") + if execution_time is not None: + entry["execution_time_ms"] = execution_time * 1000 + for key, value in info.items(): + if key == "execution_time": + continue + entry[key] = value + formatted_stages.append(entry) + + if not formatted_stages: + return + + log_entry = { + "timestamp": datetime.now(UTC).isoformat(), + "request_id": self.request_id, + "commit_hash": self.commit_hash, + "tag": "pipeline_stage_metrics", + "stages": formatted_stages, + } + perf_logger.info(json.dumps(log_entry)) diff --git a/python/sglang/multimodal_gen/test/server/conftest.py b/python/sglang/multimodal_gen/test/server/conftest.py new file mode 100644 index 000000000..96b49591b --- /dev/null +++ b/python/sglang/multimodal_gen/test/server/conftest.py @@ -0,0 +1,53 @@ +_GLOBAL_PERF_RESULTS = [] + + +def pytest_sessionfinish(session): + """ + This hook is called by pytest at the end of the entire test session. + It prints a consolidated summary of all performance results. + """ + if not _GLOBAL_PERF_RESULTS: + return + + print("\n\n" + "=" * 35 + " Performance Summary " + "=" * 35) + print( + f"{'Test Suite':<30} | {'Test Name':<20} | {'E2E (ms)':>12} | {'Avg Denoise (ms)':>18} | {'Median Denoise (ms)':>20}" + ) + print( + "-" * 30 + + "-+-" + + "-" * 20 + + "-+-" + + "-" * 12 + + "-+-" + + "-" * 18 + + "-+-" + + "-" * 20 + ) + + for entry in sorted(_GLOBAL_PERF_RESULTS, key=lambda x: x["class_name"]): + print( + f"{entry['class_name']:<30} | {entry['test_name']:<20} | {entry['e2e_ms']:>12.2f} | " + f"{entry['avg_denoise_ms']:>18.2f} | {entry['median_denoise_ms']:>20.2f}" + ) + + print("=" * 91) + + print("\n\n" + "=" * 36 + " Detailed Reports " + "=" * 37) + for entry in sorted(_GLOBAL_PERF_RESULTS, key=lambda x: x["class_name"]): + print(f"\n--- Details for {entry['class_name']} / {entry['test_name']} ---") + stage_report = ", ".join( + f"{name}:{duration:.2f}ms" + for name, duration in entry.get("stage_metrics", {}).items() + ) + if stage_report: + print(f" Stages: {stage_report}") + + sampled_steps = entry.get("sampled_steps") or {} + if sampled_steps: + step_report = ", ".join( + f"{idx}:{duration:.2f}ms" + for idx, duration in sorted(sampled_steps.items()) + ) + print(f" Sampled Steps: {step_report}") + print("=" * 91) diff --git a/python/sglang/multimodal_gen/test/server/perf_baselines.json b/python/sglang/multimodal_gen/test/server/perf_baselines.json new file mode 100644 index 000000000..4b301a5ad --- /dev/null +++ b/python/sglang/multimodal_gen/test/server/perf_baselines.json @@ -0,0 +1,82 @@ +{ + "metadata": { + "model": "Qwen/Qwen-Image", + "hardware": "CI H100 80GB pool", + "description": "Reference numbers captured from the CI diffusion server baseline run" + }, + "tolerances": { + "e2e": 0.25, + "stage": 0.3, + "denoise_step": 0.1, + "denoise_agg": 0.1 + }, + "sampling": { + "step_fractions": [ + 0.0, + 0.2, + 0.4, + 0.6, + 0.8, + 1.0 + ], + "warmup_requests": { + "text": 1, + "image_edit": 0 + } + }, + "scenarios": { + "text_to_image": { + "notes": "Single-image generation using the default prompt", + "expected_e2e_ms": 74500.0, + "expected_avg_denoise_ms": 422.42, + "expected_median_denoise_ms": 410.62, + "stages_ms": { + "InputValidationStage": 0.1, + "TextEncodingStage": 834.2, + "ConditioningStage": 0.1, + "TimestepPreparationStage": 10.6, + "LatentPreparationStage": 5.2, + "DenoisingStage": 21202.6, + "DecodingStage": 476.12 + }, + "denoise_step_ms": { + "0": 1077.77, "1": 345.13, "2": 413.8, "3": 405.49, "4": 408.14, "5": 409.06, + "6": 408.85, "7": 410.53, "8": 407.51, "9": 409.44, "10": 408.65, "11": 410.14, + "12": 411.74, "13": 409.59, "14": 409.17, "15": 410.78, "16": 410.66, "17": 410.58, + "18": 411.27, "19": 410.51, "20": 409.03, "21": 410.16, "22": 409.42, "23": 411.03, + "24": 410.18, "25": 409.72, "26": 410.26, "27": 410.21, "28": 410.71, "29": 410.76, + "30": 411.06, "31": 410.1, "32": 410.55, "33": 410.77, "34": 410.74, "35": 411.75, + "36": 410.78, "37": 411.56, "38": 410.85, "39": 411.08, "40": 411.12, "41": 411.1, + "42": 411.09, "43": 410.87, "44": 411.37, "45": 411.68, "46": 411.0, "47": 410.09, + "48": 412.72, "49": 410.42 + } + }, + "image_edit": { + "notes": "single uploaded reference image, Qwen/Qwen-Image-Edit", + "expected_e2e_ms": 138500.0, + "expected_avg_denoise_ms": 720.0, + "expected_median_denoise_ms": 718.0, + "stages_ms": { + "InputValidationStage": 14, + "ImageEncodingStage": 990.0, + "ImageVAEEncodingStage": 252.76, + "ConditioningStage": 0.13, + "TimestepPreparationStage": 13.78, + "LatentPreparationStage": 9.18, + "DenoisingStage": 36000.0, + "DecodingStage": 645 + }, + "denoise_step_ms": { + "0": 720.0, "1": 720.0, "2": 720.0, "3": 720.0, "4": 720.0, "5": 720.0, + "6": 720.0, "7": 720.0, "8": 720.0, "9": 720.0, "10": 720.0, "11": 720.0, + "12": 720.0, "13": 720.0, "14": 720.0, "15": 720.0, "16": 720.0, "17": 720.0, + "18": 720.0, "19": 720.0, "20": 720.0, "21": 720.0, "22": 720.0, "23": 720.0, + "24": 720.0, "25": 720.0, "26": 720.0, "27": 720.0, "28": 720.0, "29": 720.0, + "30": 720.0, "31": 720.0, "32": 720.0, "33": 720.0, "34": 720.0, "35": 720.0, + "36": 720.0, "37": 720.0, "38": 720.0, "39": 720.0, "40": 720.0, "41": 720.0, + "42": 720.0, "43": 720.0, "44": 720.0, "45": 720.0, "46": 720.0, "47": 720.0, + "48": 720.0, "49": 720.0 + } + } + } +} diff --git a/python/sglang/multimodal_gen/test/server/test_server_performance.py b/python/sglang/multimodal_gen/test/server/test_server_performance.py new file mode 100644 index 000000000..61df5550a --- /dev/null +++ b/python/sglang/multimodal_gen/test/server/test_server_performance.py @@ -0,0 +1,490 @@ +# Server-based diffusion performance test: +# - Launches an sglang diffusion server via the CLI. +# - Issues an OpenAI-compatible Images API request. +# - Extracts all performance metrics from performance.log (no stdout parsing). +# - Verifies E2E, stage-level, and denoising-step latencies with configurable buffers. + +from __future__ import annotations + +import base64 +import json +import os +import statistics +import subprocess +import tempfile +import time +from pathlib import Path +from typing import Any, Sequence + +import pytest +from openai import OpenAI + +from sglang.multimodal_gen.runtime.utils.common import kill_process_tree +from sglang.multimodal_gen.runtime.utils.logging_utils import init_logger +from sglang.multimodal_gen.test.server.conftest import _GLOBAL_PERF_RESULTS +from sglang.multimodal_gen.test.test_utils import ( + get_dynamic_server_port, + is_jpeg, + is_png, + prepare_perf_log, + read_perf_records, + sample_step_indices, + wait_for_perf_record, + wait_for_stage_metrics, +) + +logger = init_logger(__name__) + + +_BASELINE_PATH = Path(__file__).with_name("perf_baselines.json") +with _BASELINE_PATH.open("r", encoding="utf-8") as _fh: + _BASELINE_CONFIG = json.load(_fh) + +_SCENARIOS = _BASELINE_CONFIG["scenarios"] +_TEXT_SCENARIO = _SCENARIOS["text_to_image"] +_IMAGE_EDIT_SCENARIO = _SCENARIOS["image_edit"] + +STEP_SAMPLE_FRACTIONS: Sequence[float] = tuple( + _BASELINE_CONFIG["sampling"]["step_fractions"] +) + +_WARMUP_DEFAULTS = _BASELINE_CONFIG["sampling"].get("warmup_requests", {}) +_DEFAULT_WARMUP_TEXT = int(_WARMUP_DEFAULTS.get("text", 1)) +_DEFAULT_WARMUP_EDIT = int(_WARMUP_DEFAULTS.get("image_edit", 0)) + +_TOLERANCES = _BASELINE_CONFIG["tolerances"] + + +def _tolerance_from_env(var_name: str, default: float) -> float: + override = os.environ.get(var_name) + if override is not None: + return float(override) + return float(default) + + +E2E_TOLERANCE_RATIO = _tolerance_from_env("SGLANG_E2E_TOLERANCE", _TOLERANCES["e2e"]) +STAGE_TOLERANCE_RATIO = _tolerance_from_env( + "SGLANG_STAGE_TIME_TOLERANCE", _TOLERANCES["stage"] +) +DENOISE_STEP_TOLERANCE_RATIO = _tolerance_from_env( + "SGLANG_DENOISE_STEP_TOLERANCE", _TOLERANCES["denoise_step"] +) +DENOISE_AGG_TOLERANCE_RATIO = _tolerance_from_env( + "SGLANG_DENOISE_AGG_TOLERANCE", _TOLERANCES["denoise_agg"] +) + + +def _decode_and_validate_image(b64_json: str) -> None: + image_bytes = base64.b64decode(b64_json) + assert is_png(image_bytes) or is_jpeg( + image_bytes + ), "Warm-up image must be PNG or JPEG" + + +def _run_warmup_requests(cls, port: int) -> None: + warmup_text_requests = int(getattr(cls, "WARMUP_TEXT_REQUESTS", 1)) + warmup_edit_requests = int(getattr(cls, "WARMUP_IMAGE_EDIT_REQUESTS", 0)) + if warmup_text_requests <= 0 and warmup_edit_requests <= 0: + return + + client = OpenAI( + api_key="sglang-anything", + base_url=f"http://localhost:{port}/v1", + ) + prompt = getattr(cls, "PROMPT", "A colorful raccoon icon") + output_size = getattr(cls, "OUTPUT_SIZE", "1024x1024") + + logger.info( + "[server-test] Running %s text warm-up(s) and %s edit warm-up(s)", + warmup_text_requests, + warmup_edit_requests, + ) + + for _ in range(warmup_text_requests): + result = client.images.generate( + model=getattr(cls, "MODEL_PATH"), + prompt=prompt, + n=1, + size=output_size, + response_format="b64_json", + ) + _decode_and_validate_image(result.data[0].b64_json) + + if warmup_edit_requests > 0: + edit_prompt = getattr(cls, "IMAGE_EDIT_PROMPT", None) + edit_path: Path | None = getattr(cls, "IMAGE_EDIT_PATH", None) + if not edit_prompt or not edit_path or not edit_path.exists(): + logger.warning( + "[server-test] Skipping image-edit warm-up: prompt=%s path=%s exists=%s", + bool(edit_prompt), + edit_path, + edit_path.exists() if edit_path else False, + ) + return + for _ in range(warmup_edit_requests): + with edit_path.open("rb") as fh: + result = client.images.edit( + model=getattr(cls, "MODEL_PATH"), + image=fh, + prompt=edit_prompt, + n=1, + size=output_size, + response_format="b64_json", + ) + _decode_and_validate_image(result.data[0].b64_json) + + +@pytest.fixture(scope="class") +def diffusion_server(request): + cls = request.cls + + log_dir, perf_log_path = prepare_perf_log(Path(__file__)) + + default_port = get_dynamic_server_port() + port = int(os.environ.get("SGLANG_TEST_SERVER_PORT", default_port)) + port = getattr(cls, "SERVER_PORT", port) + + model = getattr(cls, "MODEL_PATH") + wait_deadline = float(os.environ.get("SGLANG_TEST_WAIT_SECS", "1200")) + serve_extra_args = os.environ.get("SGLANG_TEST_SERVE_ARGS", "") + + safe_model_name = model.replace("/", "_") + stdout_path = ( + Path(tempfile.gettempdir()) / f"sgl_server_{port}_{safe_model_name}.log" + ) + stdout_path.unlink(missing_ok=True) + + base_command = [ + "sglang", + "serve", + "--model-path", + model, + "--port", + str(port), + "--log-level=debug", + ] + if serve_extra_args.strip(): + base_command += serve_extra_args.strip().split() + + env = os.environ.copy() + env["SGL_DIFFUSION_STAGE_LOGGING"] = "1" + env["SGLANG_PERF_LOG_DIR"] = log_dir.as_posix() + + stdout_fh = stdout_path.open("w", encoding="utf-8", buffering=1) + process = subprocess.Popen( + base_command, + stdout=stdout_fh, + stderr=subprocess.STDOUT, + text=True, + bufsize=1, + env=env, + ) + logger.info( + "[server-test] Starting diffusion server pid=%s, model=%s, log=%s", + process.pid, + model, + stdout_path.as_posix(), + ) + + start = time.time() + server_ready_message = "Application startup complete." + server_ready = False + + while time.time() - start < wait_deadline: + if process.poll() is not None: + tail = "" + try: + tail = "\n".join( + stdout_path.read_text( + encoding="utf-8", errors="ignore" + ).splitlines()[-200:] + ) + except Exception: + pass + raise RuntimeError( + f"Server exited early (code {process.returncode}). Last logs:\n{tail}" + ) + + if stdout_path.exists(): + try: + log_content = stdout_path.read_text(encoding="utf-8", errors="ignore") + if server_ready_message in log_content: + logger.info("[server-test] Server is fully loaded and ready.") + server_ready = True + break + except Exception as e: + logger.debug("Could not read server log file yet: %s", e) + + logger.info( + "[server-test] Waiting for server to initialize... elapsed=%ss", + int(time.time() - start), + ) + time.sleep(5) + + if not server_ready: + tail = "" + try: + tail = "\n".join( + stdout_path.read_text(encoding="utf-8", errors="ignore").splitlines()[ + -200: + ] + ) + except Exception: + pass + raise TimeoutError( + f"Server did not become ready within {wait_deadline}s. Last logs:\n{tail}" + ) + + ctx = { + "port": port, + "stdout_file": stdout_path, + "process": process, + "model": model, + "fh": stdout_fh, + "perf_log_path": perf_log_path, + "log_dir": log_dir, + } + request.cls.server_ctx = ctx + request.cls.perf_log_path = perf_log_path + + grace = float(getattr(cls, "STARTUP_GRACE_SECONDS", 0.0) or 0.0) + if grace > 0: + logger.info( + "[server-test] Waiting %.1fs before warm-ups to let model settle", grace + ) + time.sleep(grace) + + try: + _run_warmup_requests(cls, port) + except Exception as exc: + logger.error("Warm-up requests failed: %s", exc) + kill_process_tree(process.pid) + raise + + yield ctx + + try: + kill_process_tree(process.pid) + except Exception: + pass + try: + stdout_fh.flush() + stdout_fh.close() + except Exception: + pass + + +@pytest.mark.usefixtures("diffusion_server") +class DiffusionPerfTestBase: + MODEL_PATH: str + # SERVER_PORT = int(os.environ.get("SGLANG_TEST_SERVER_PORT", "30100")) + PROMPT = "A Logo With Bold Large Text: SGL Diffusion" + IMAGE_EDIT_PROMPT: str | None = None + IMAGE_EDIT_PATH = Path(__file__).resolve().parents[1] / "test_files" / "girl.jpg" + OUTPUT_SIZE = "1024x1024" + WARMUP_TEXT_REQUESTS = _DEFAULT_WARMUP_TEXT + WARMUP_IMAGE_EDIT_REQUESTS = _DEFAULT_WARMUP_EDIT + STARTUP_GRACE_SECONDS = 0.0 + + STAGE_EXPECTATIONS: dict + STEP_EXPECTATIONS: dict + EXPECTED_E2E_MS: float + EXPECTED_AVG_DENOISE_MS: float + EXPECTED_MEDIAN_DENOISE_MS: float + + _perf_results: list[dict[str, Any]] = [] + + @classmethod + def setup_class(cls): + cls._perf_results = [] + + @classmethod + def teardown_class(cls): + for result in cls._perf_results: + result["class_name"] = cls.__name__ + _GLOBAL_PERF_RESULTS.append(result) + + def _client(self) -> OpenAI: + return OpenAI( + api_key="sglang-anything", + base_url=f"http://localhost:{self.server_ctx['port']}/v1", + ) + + def _perf_log_path(self) -> Path: + return self.server_ctx["perf_log_path"] + + def _record_result(self, test_name: str, summary: dict[str, Any]) -> None: + if not summary: + return + entry = {"test_name": test_name, **summary} + self.__class__._perf_results.append(entry) + + def _run_and_collect_records(self, generate_fn) -> tuple[dict, dict]: + log_path = self._perf_log_path() + prev_len = len(read_perf_records(log_path)) + generate_fn() + perf_record, _ = wait_for_perf_record( + "total_inference_time", + prev_len, + log_path, + ) + stage_metrics, _ = wait_for_stage_metrics( + perf_record.get("request_id", ""), + prev_len, + len(self.STAGE_EXPECTATIONS), + log_path, + ) + return perf_record, stage_metrics + + def _generate_image(self): + client = self._client() + result = client.images.generate( + model=self.MODEL_PATH, + prompt=self.PROMPT, + n=1, + size=self.OUTPUT_SIZE, + response_format="b64_json", + ) + image_bytes = base64.b64decode(result.data[0].b64_json) + assert is_png(image_bytes) or is_jpeg( + image_bytes + ), "Generated image must be PNG or JPEG" + + def _generate_image_edit(self): + if not self.IMAGE_EDIT_PROMPT: + pytest.skip("Image edit prompt not configured") + if not self.IMAGE_EDIT_PATH.exists(): + pytest.skip(f"Image edit file missing: {self.IMAGE_EDIT_PATH}") + client = self._client() + with self.IMAGE_EDIT_PATH.open("rb") as fh: + result = client.images.edit( + model=self.MODEL_PATH, + image=fh, + prompt=self.IMAGE_EDIT_PROMPT, + n=1, + size=self.OUTPUT_SIZE, + response_format="b64_json", + ) + image_bytes = base64.b64decode(result.data[0].b64_json) + assert is_png(image_bytes) or is_jpeg( + image_bytes + ), "Edited image must be PNG or JPEG" + + def _assert_metrics(self, perf_record: dict, stage_metrics: dict): + e2e_ms = float(perf_record.get("total_duration_ms", 0.0)) + assert e2e_ms > 0, "E2E duration missing from perf log" + e2e_upper = self.EXPECTED_E2E_MS * (1 + E2E_TOLERANCE_RATIO) + assert ( + e2e_ms <= e2e_upper + ), f"E2E time {e2e_ms:.2f}ms exceeds allowed {e2e_upper:.2f}ms" + + steps = [ + step + for step in perf_record.get("steps", []) or [] + if step.get("name") == "denoising_step_guided" and "duration_ms" in step + ] + assert steps, "Denoising step timings missing from perf log" + + durations = [float(step["duration_ms"]) for step in steps] + avg_duration = sum(durations) / len(durations) + median_duration = statistics.median(durations) + + avg_upper = self.EXPECTED_AVG_DENOISE_MS * (1 + DENOISE_AGG_TOLERANCE_RATIO) + med_upper = self.EXPECTED_MEDIAN_DENOISE_MS * (1 + DENOISE_AGG_TOLERANCE_RATIO) + assert ( + avg_duration <= avg_upper + ), f"Avg denoise {avg_duration:.2f}ms exceeds {avg_upper:.2f}ms" + assert ( + median_duration <= med_upper + ), f"Median denoise {median_duration:.2f}ms exceeds {med_upper:.2f}ms" + + avg_per_step = { + int(step.get("index")): float(step["duration_ms"]) + for step in steps + if step.get("index") is not None + } + sample_indices = sample_step_indices(avg_per_step, STEP_SAMPLE_FRACTIONS) + sampled_steps = {idx: avg_per_step[idx] for idx in sample_indices} + for idx in sample_indices: + expected = self.STEP_EXPECTATIONS.get(idx) + if expected is None: + continue + actual = avg_per_step[idx] + upper_bound = expected * (1 + DENOISE_STEP_TOLERANCE_RATIO) + assert ( + actual <= upper_bound + ), f"Denoise step {idx} took {actual:.2f}ms > allowed {upper_bound:.2f}ms" + + assert stage_metrics, "Stage metrics missing from performance log" + for stage, expected in self.STAGE_EXPECTATIONS.items(): + actual = stage_metrics.get(stage) + assert actual is not None, f"Stage {stage} timing missing" + upper_bound = expected * (1 + STAGE_TOLERANCE_RATIO) + assert ( + actual <= upper_bound + ), f"Stage {stage} took {actual:.2f}ms > allowed {upper_bound:.2f}ms" + + # Log to pytest console during the run for immediate feedback + logger.info( + "[Perf] %s/%s: E2E %.2f ms; Avg denoise %.2f ms; Median %.2f ms", + self.__class__.__name__, + perf_record.get("test_name", "test"), + e2e_ms, + avg_duration, + median_duration, + ) + + return { + "e2e_ms": e2e_ms, + "avg_denoise_ms": avg_duration, + "median_denoise_ms": median_duration, + "stage_metrics": stage_metrics, + "sampled_steps": sampled_steps, + } + + +class TestQwenImageGeneration(DiffusionPerfTestBase): + """Performance tests for the Qwen/Qwen-image model.""" + + MODEL_PATH = "Qwen/Qwen-Image" + STARTUP_GRACE_SECONDS = 30.0 + WARMUP_IMAGE_EDIT_REQUESTS = 0 + STAGE_EXPECTATIONS = _TEXT_SCENARIO["stages_ms"] + STEP_EXPECTATIONS = { + int(k): v for k, v in _TEXT_SCENARIO["denoise_step_ms"].items() + } + EXPECTED_E2E_MS = float(_TEXT_SCENARIO["expected_e2e_ms"]) + EXPECTED_AVG_DENOISE_MS = float(_TEXT_SCENARIO["expected_avg_denoise_ms"]) + EXPECTED_MEDIAN_DENOISE_MS = float(_TEXT_SCENARIO["expected_median_denoise_ms"]) + + def test_text_to_image_performance(self): + perf_record, stage_metrics = self._run_and_collect_records(self._generate_image) + summary = self._assert_metrics(perf_record, stage_metrics) + self._record_result("text_to_image", summary) + + +class TestQwenImageEdit(DiffusionPerfTestBase): + """Performance tests for the Qwen/Qwen-Image-Edit model.""" + + MODEL_PATH = "Qwen/Qwen-Image-Edit" + IMAGE_EDIT_PROMPT = "Convert 2D style to 3D style" + OUTPUT_SIZE = "1024x1536" + STARTUP_GRACE_SECONDS = 30.0 + WARMUP_TEXT_REQUESTS = 0 + WARMUP_IMAGE_EDIT_REQUESTS = 1 + STAGE_EXPECTATIONS = _IMAGE_EDIT_SCENARIO["stages_ms"] + STEP_EXPECTATIONS = { + int(k): v for k, v in _IMAGE_EDIT_SCENARIO["denoise_step_ms"].items() + } + EXPECTED_E2E_MS = float(_IMAGE_EDIT_SCENARIO["expected_e2e_ms"]) + EXPECTED_AVG_DENOISE_MS = float(_IMAGE_EDIT_SCENARIO["expected_avg_denoise_ms"]) + EXPECTED_MEDIAN_DENOISE_MS = float( + _IMAGE_EDIT_SCENARIO["expected_median_denoise_ms"] + ) + + def test_image_edit_performance(self): + perf_record, stage_metrics = self._run_and_collect_records( + self._generate_image_edit + ) + summary = self._assert_metrics(perf_record, stage_metrics) + self._record_result("image_edit", summary) diff --git a/python/sglang/multimodal_gen/test/test_files/girl.jpg b/python/sglang/multimodal_gen/test/test_files/girl.jpg new file mode 100644 index 000000000..63ba28d39 Binary files /dev/null and b/python/sglang/multimodal_gen/test/test_files/girl.jpg differ diff --git a/python/sglang/multimodal_gen/test/test_files/rabbit.jpg b/python/sglang/multimodal_gen/test/test_files/rabbit.jpg deleted file mode 100644 index 56747c94a..000000000 Binary files a/python/sglang/multimodal_gen/test/test_files/rabbit.jpg and /dev/null differ diff --git a/python/sglang/multimodal_gen/test/test_utils.py b/python/sglang/multimodal_gen/test/test_utils.py index 86847797a..0e947dae0 100644 --- a/python/sglang/multimodal_gen/test/test_utils.py +++ b/python/sglang/multimodal_gen/test/test_utils.py @@ -1,5 +1,6 @@ # Copied and adapted from: https://github.com/hao-ai-lab/FastVideo import dataclasses +import json import os import shlex import socket @@ -7,11 +8,13 @@ import subprocess import sys import time import unittest -from typing import Optional +from pathlib import Path +from typing import Optional, Sequence from PIL import Image from sglang.multimodal_gen.configs.sample.base import DataType +from sglang.multimodal_gen.runtime.utils.common import get_bool_env_var from sglang.multimodal_gen.runtime.utils.logging_utils import init_logger logger = init_logger(__name__) @@ -52,11 +55,37 @@ def probe_port(host="127.0.0.1", port=30010, timeout=2.0) -> bool: return False +def is_in_ci() -> bool: + return get_bool_env_var("SGLANG_IS_IN_CI") + + +def get_dynamic_server_port() -> int: + cuda_devices = os.environ.get("CUDA_VISIBLE_DEVICES", "0") + if not cuda_devices: + cuda_devices = "0" + try: + first_device_id = int(cuda_devices.split(",")[0].strip()[0]) + except (ValueError, IndexError): + first_device_id = 0 + + if is_in_ci(): + base_port = 10000 + first_device_id * 2000 + else: + base_port = 20000 + first_device_id * 1000 + + return base_port + 1000 + + def is_mp4(data): idx = data.find(b"ftyp") return 0 <= idx <= 32 +def is_jpeg(data: bytes) -> bool: + # JPEG files start with: FF D8 FF + return data.startswith(b"\xff\xd8\xff") + + def is_png(data): # PNG files start with: 89 50 4E 47 0D 0A 1A 0A return data.startswith(b"\x89PNG\r\n\x1a\n") @@ -77,6 +106,113 @@ def check_image_size(ut, image, width, height): ut.assertEqual(image.size, (width, height)) +def get_perf_log_dir(start_file: Path) -> Path: + """Mirror runtime/utils/performance_logger.py behaviour for locating logs.""" + this_file = start_file.resolve() + root_logs = this_file.parents[3] / "logs" + fallback = this_file.parents[2] / "logs" + return root_logs if root_logs.exists() or not fallback.exists() else fallback + + +def _ensure_log_path(log_dir: Path) -> Path: + log_dir.mkdir(parents=True, exist_ok=True) + return log_dir / "performance.log" + + +def clear_perf_log(log_dir: Path) -> Path: + """Delete the perf log file so tests can watch for fresh entries.""" + log_path = _ensure_log_path(log_dir) + if log_path.exists(): + log_path.unlink() + logger.info("[server-test] Monitoring perf log at %s", log_path.as_posix()) + return log_path + + +def prepare_perf_log(start_file: Path) -> tuple[Path, Path]: + """Convenience helper to resolve and clear the perf log in one call.""" + log_dir = get_perf_log_dir(start_file) + log_path = clear_perf_log(log_dir) + return log_dir, log_path + + +def read_perf_records(log_path: Path) -> list[dict]: + if not log_path.exists(): + return [] + records: list[dict] = [] + with log_path.open("r", encoding="utf-8") as fh: + for line in fh: + line = line.strip() + if not line: + continue + try: + records.append(json.loads(line)) + except json.JSONDecodeError: + continue + return records + + +def wait_for_perf_record( + tag: str, + prev_len: int, + log_path: Path, + timeout: float = 120.0, +) -> tuple[dict, int]: + deadline = time.time() + timeout + while time.time() < deadline: + records = read_perf_records(log_path) + if len(records) > prev_len: + for rec in records[prev_len:]: + if rec.get("tag") == tag: + return rec, len(records) + time.sleep(0.5) + raise AssertionError( + f"Timeout waiting for perf log entry '{tag}' (start_len={prev_len})" + ) + + +def wait_for_stage_metrics( + request_id: str, + prev_len: int, + expected_count: int, + log_path: Path, + timeout: float = 120.0, +) -> tuple[dict[str, float], int]: + deadline = time.time() + timeout + metrics: dict[str, float] = {} + while time.time() < deadline: + records = read_perf_records(log_path) + for rec in records[prev_len:]: + if ( + rec.get("tag") == "pipeline_stage_metric" + and rec.get("request_id") == request_id + ): + stage = rec.get("stage") + duration = rec.get("duration_ms") + if stage is not None and duration is not None: + metrics[str(stage)] = float(duration) + if len(metrics) >= expected_count: + return metrics, len(records) + time.sleep(0.5) + raise AssertionError( + f"Timeout waiting for stage metrics for request {request_id} " + f"(collected={len(metrics)} expected={expected_count})" + ) + + +def sample_step_indices( + step_map: dict[int, float], fractions: Sequence[float] +) -> list[int]: + if not step_map: + return [] + max_idx = max(step_map.keys()) + indices = set() + for fraction in fractions: + idx = min(max_idx, max(0, int(round(fraction * max_idx)))) + if idx in step_map: + indices.add(idx) + return sorted(indices) + + @dataclasses.dataclass class TestResult: name: str diff --git a/scripts/ci/ci_install_dependency.sh b/scripts/ci/ci_install_dependency.sh index e456fd0f8..36cc4928a 100755 --- a/scripts/ci/ci_install_dependency.sh +++ b/scripts/ci/ci_install_dependency.sh @@ -4,6 +4,7 @@ set -euxo pipefail IS_BLACKWELL=${IS_BLACKWELL:-0} CU_VERSION="cu129" +OPTIONAL_DEPS="${1:-}" # Detect system architecture ARCH=$(uname -m) @@ -93,8 +94,14 @@ else $PIP_CMD uninstall flashinfer-python flashinfer-cubin flashinfer-jit-cache || true fi +EXTRAS="dev" +if [ -n "$OPTIONAL_DEPS" ]; then + EXTRAS="dev,${OPTIONAL_DEPS}" +fi +echo "Installing python extras: [${EXTRAS}]" + # Install the main package -$PIP_CMD install -e "python[dev]" --extra-index-url https://download.pytorch.org/whl/${CU_VERSION} $PIP_INSTALL_SUFFIX +$PIP_CMD install -e "python[${EXTRAS}]" --extra-index-url https://download.pytorch.org/whl/${CU_VERSION} $PIP_INSTALL_SUFFIX # Install router for pd-disagg test $PIP_CMD install sglang-router $PIP_INSTALL_SUFFIX @@ -126,11 +133,8 @@ if [ "$IS_BLACKWELL" != "1" ]; then # For lmms_evals evaluating MMMU git clone --branch v0.5 --depth 1 https://github.com/EvolvingLMMs-Lab/lmms-eval.git $PIP_CMD install -e lmms-eval/ $PIP_INSTALL_SUFFIX - - # Install xformers - $PIP_CMD install xformers --index-url https://download.pytorch.org/whl/${CU_VERSION} --no-deps $PIP_INSTALL_SUFFIX fi - +$PIP_CMD uninstall xformers || true # Show current packages $PIP_CMD list python3 -c "import torch; print(torch.version.cuda)"