diff --git a/benchmarks/baseline/diffusers_wan_i2v/README.md b/benchmarks/baseline/diffusers_wan_i2v/README.md new file mode 100644 index 0000000..def692a --- /dev/null +++ b/benchmarks/baseline/diffusers_wan_i2v/README.md @@ -0,0 +1,74 @@ +# Diffusers Wan2.1 I2V Baseline + +This directory contains a standalone benchmark baseline for the official +`Wan-AI/Wan2.1-I2V-14B-480P-Diffusers` pipeline. + +The goal is to compare: + +- the same model +- the same I2V workload +- the same async `/v1/videos` HTTP semantics +- while changing only the inference framework + +Current fixed workload: + +- model: `Wan-AI/Wan2.1-I2V-14B-480P-Diffusers` +- resolution: `480p` +- effective size: `832x480` +- frames: `81` +- denoising steps: `40` +- guidance scale: `5.0` +- output fps: `16` + +## Layout + +- `service.py`: standalone FastAPI server +- `configs/`: AIPerf configs for the baseline +- `scripts/run_service.sh`: launch helper +- `scripts/run_video_bench.sh`: benchmark helper + +## Start The Service + +```bash +python3 benchmarks/baseline/diffusers_wan_i2v/service.py +``` + +Or: + +```bash +bash benchmarks/baseline/diffusers_wan_i2v/scripts/run_service.sh +``` + +Default address: + +- `http://127.0.0.1:8010` + +Health check: + +```bash +curl http://127.0.0.1:8010/v1/service/health +``` + +## Run AIPerf + +Install the vendored AIPerf first: + +```bash +pip install -e ./benchmarks/aiperf +``` + +Then run: + +```bash +bash benchmarks/baseline/diffusers_wan_i2v/scripts/run_video_bench.sh +``` + +This reuses the same prompt dataset as the TeleFuser benchmark: + +- `benchmarks/telefuser_aiperf/data/video_prompts.jsonl` + +## Notes + +- The service supports both `multipart/form-data` and JSON requests. +- `reference_url` is treated as a local filesystem path in the current benchmark dataset. +- Remote HTTP image URLs are intentionally rejected in this first baseline to keep the benchmark focused on inference rather than network fetch behavior. diff --git a/benchmarks/baseline/diffusers_wan_i2v/benchmark_contract.yaml b/benchmarks/baseline/diffusers_wan_i2v/benchmark_contract.yaml new file mode 100644 index 0000000..b206817 --- /dev/null +++ b/benchmarks/baseline/diffusers_wan_i2v/benchmark_contract.yaml @@ -0,0 +1,40 @@ +contract_version: v1 +name: diffusers_wan_i2v +mode: batch_video +implementation: diffusers +model_family: wan_video +model: Wan2.1-I2V-14B-480P-Diffusers +supported_tasks: + - i2v +transport: http +endpoint: + path: /v1/videos + protocol: openai_video_generation +request_encoding: + content_type: multipart/form-data + media_inputs: + first_image_path: input_reference + parameters: + prompt: prompt + negative_prompt: negative_prompt + seed: seed + size: size + seconds: seconds +result_delivery: + polling_path: /v1/videos/{id} + content_path: /v1/videos/{id}/content +workload: + resolution: 832x480 + frames: 81 + inference_steps: 40 + fps: 16 + guidance_scale: 5.0 +metrics: + - request_latency + - completion_latency + - download_latency + - request_throughput + - success_rate +artifacts: + config: benchmarks/baseline/diffusers_wan_i2v/configs/video_generation_compare.yaml + dataset: benchmarks/telefuser_aiperf/data/video_prompts.jsonl diff --git a/benchmarks/baseline/diffusers_wan_i2v/configs/video_generation_compare.yaml b/benchmarks/baseline/diffusers_wan_i2v/configs/video_generation_compare.yaml new file mode 100644 index 0000000..bba583b --- /dev/null +++ b/benchmarks/baseline/diffusers_wan_i2v/configs/video_generation_compare.yaml @@ -0,0 +1,54 @@ +# yaml-language-server: $schema=../../../aiperf/src/aiperf/config/schema/aiperf-config.schema.json + +schemaVersion: "2.0" + +randomSeed: 42 + +benchmark: + model: diffusers-wan21-i2v-480p + + endpoint: + url: ${DIFFUSERS_WAN_AIPERF_URL:http://127.0.0.1:8010} + type: video_generation + timeout: 7200.0 + downloadVideoContent: true + extra: + size: ${DIFFUSERS_WAN_AIPERF_SIZE:480p} + seconds: ${DIFFUSERS_WAN_AIPERF_SECONDS:5} + seed: ${DIFFUSERS_WAN_AIPERF_SEED:42} + model: ${DIFFUSERS_WAN_AIPERF_MODEL:Wan-AI/Wan2.1-I2V-14B-480P-Diffusers} + + tokenizer: + name: builtin + + dataset: + type: file + path: ./benchmarks/telefuser_aiperf/data/video_prompts.jsonl + format: single_turn + sampling: sequential + + warmup: + type: concurrency + concurrency: 1 + requests: 1 + excludeFromResults: true + + profiling: + type: concurrency + concurrency: ${DIFFUSERS_WAN_AIPERF_CONCURRENCY:1} + requests: ${DIFFUSERS_WAN_AIPERF_REQUESTS:2} + duration: ${DIFFUSERS_WAN_AIPERF_DURATION:7200} + gracePeriod: ${DIFFUSERS_WAN_AIPERF_GRACE_PERIOD:900} + + artifacts: + dir: ./artifacts/diffusers_wan_i2v/wan21_i2v_480p_compare + summary: [json] + records: [jsonl] + showTraceTiming: true + trace: true + + serverMetrics: + enabled: ${DIFFUSERS_WAN_AIPERF_SERVER_METRICS:false} + urls: + - ${DIFFUSERS_WAN_AIPERF_METRICS_URL:http://127.0.0.1:8010/metrics} + formats: [json, csv] diff --git a/benchmarks/baseline/diffusers_wan_i2v/configs/video_generation_e2e.yaml b/benchmarks/baseline/diffusers_wan_i2v/configs/video_generation_e2e.yaml new file mode 100644 index 0000000..1bc971d --- /dev/null +++ b/benchmarks/baseline/diffusers_wan_i2v/configs/video_generation_e2e.yaml @@ -0,0 +1,54 @@ +# yaml-language-server: $schema=../../aiperf/src/aiperf/config/schema/aiperf-config.schema.json + +schemaVersion: "2.0" + +randomSeed: 42 + +benchmark: + model: diffusers-wan21-i2v-480p + + endpoint: + url: ${DIFFUSERS_WAN_AIPERF_URL:http://127.0.0.1:8010} + type: video_generation + timeout: 7200.0 + downloadVideoContent: true + extra: + size: ${DIFFUSERS_WAN_AIPERF_SIZE:480p} + seconds: ${DIFFUSERS_WAN_AIPERF_SECONDS:5} + seed: ${DIFFUSERS_WAN_AIPERF_SEED:42} + model: ${DIFFUSERS_WAN_AIPERF_MODEL:Wan-AI/Wan2.1-I2V-14B-480P-Diffusers} + + tokenizer: + name: builtin + + dataset: + type: file + path: ./benchmarks/telefuser_aiperf/data/video_prompts.jsonl + format: single_turn + sampling: sequential + + warmup: + type: concurrency + concurrency: 1 + requests: 1 + excludeFromResults: true + + profiling: + type: concurrency + concurrency: ${DIFFUSERS_WAN_AIPERF_CONCURRENCY:2} + requests: ${DIFFUSERS_WAN_AIPERF_REQUESTS:6} + duration: ${DIFFUSERS_WAN_AIPERF_DURATION:7200} + gracePeriod: ${DIFFUSERS_WAN_AIPERF_GRACE_PERIOD:900} + + artifacts: + dir: ./artifacts/diffusers_wan_i2v/video_e2e + summary: [json] + records: [jsonl] + showTraceTiming: true + trace: true + + serverMetrics: + enabled: ${DIFFUSERS_WAN_AIPERF_SERVER_METRICS:true} + urls: + - ${DIFFUSERS_WAN_AIPERF_METRICS_URL:http://127.0.0.1:8010/metrics} + formats: [json, csv] diff --git a/benchmarks/baseline/diffusers_wan_i2v/scripts/run_service.sh b/benchmarks/baseline/diffusers_wan_i2v/scripts/run_service.sh new file mode 100644 index 0000000..e760bc3 --- /dev/null +++ b/benchmarks/baseline/diffusers_wan_i2v/scripts/run_service.sh @@ -0,0 +1,11 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)" +cd "${ROOT_DIR}" + +PYTHON_BIN="${PYTHON_BIN:-python3}" +SERVICE_HOST="${DIFFUSERS_WAN_I2V_HOST:-127.0.0.1}" +SERVICE_PORT="${DIFFUSERS_WAN_I2V_PORT:-8010}" + +exec "${PYTHON_BIN}" benchmarks/baseline/diffusers_wan_i2v/service.py diff --git a/benchmarks/baseline/diffusers_wan_i2v/scripts/run_video_bench.sh b/benchmarks/baseline/diffusers_wan_i2v/scripts/run_video_bench.sh new file mode 100644 index 0000000..16a19f2 --- /dev/null +++ b/benchmarks/baseline/diffusers_wan_i2v/scripts/run_video_bench.sh @@ -0,0 +1,23 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)" +cd "${ROOT_DIR}" + +CONFIG_PATH="${1:-benchmarks/baseline/diffusers_wan_i2v/configs/video_generation_e2e.yaml}" +SERVER_URL="${DIFFUSERS_WAN_AIPERF_URL:-http://127.0.0.1:8010}" +HEALTH_URL="${DIFFUSERS_WAN_AIPERF_HEALTH_URL:-${SERVER_URL}/v1/service/health}" +AIPERF_BIN="${AIPERF_BIN:-aiperf}" + +if ! command -v "${AIPERF_BIN}" >/dev/null 2>&1; then + echo "aiperf is not installed. Set AIPERF_BIN or run: pip install -e ./benchmarks/aiperf" >&2 + exit 1 +fi + +if command -v curl >/dev/null 2>&1; then + echo "Checking baseline service health: ${HEALTH_URL}" + curl --fail --silent --show-error "${HEALTH_URL}" >/dev/null +fi + +echo "Running AIPerf with config: ${CONFIG_PATH}" +"${AIPERF_BIN}" profile --config "${CONFIG_PATH}" diff --git a/benchmarks/baseline/diffusers_wan_i2v/service.py b/benchmarks/baseline/diffusers_wan_i2v/service.py new file mode 100644 index 0000000..f8a3421 --- /dev/null +++ b/benchmarks/baseline/diffusers_wan_i2v/service.py @@ -0,0 +1,473 @@ +from __future__ import annotations + +import base64 +import binascii +import io +import os +import threading +import time +import uuid +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import torch +import uvicorn +from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile +from fastapi.responses import FileResponse, JSONResponse, Response +from PIL import Image + +from diffusers import AutoencoderKLWan, WanImageToVideoPipeline +from diffusers.utils import export_to_video +from transformers import CLIPVisionModel + +DEFAULT_MODEL_ID = "Wan-AI/Wan2.1-I2V-14B-480P-Diffusers" +DEFAULT_OUTPUT_DIR = "artifacts/diffusers_wan_i2v" +DEFAULT_HOST = "127.0.0.1" +DEFAULT_PORT = 8010 +DEFAULT_GUIDANCE_SCALE = 5.0 +DEFAULT_NUM_FRAMES = 81 +DEFAULT_NUM_INFERENCE_STEPS = 40 +DEFAULT_TARGET_FPS = 16 +DEFAULT_HEIGHT = 480 +DEFAULT_WIDTH = 832 +DEFAULT_TIMEOUT_SECONDS = 7200.0 +VIDEO_STATUS_QUEUED = "queued" +VIDEO_STATUS_GENERATING = "generating" +VIDEO_STATUS_COMPLETED = "completed" +VIDEO_STATUS_FAILED = "failed" +VIDEO_STATUS_CANCELLED = "cancelled" + + +def _coerce_env_int(name: str, default: int) -> int: + value = os.environ.get(name) + if value is None or value == "": + return default + return int(value) + + +def _coerce_env_float(name: str, default: float) -> float: + value = os.environ.get(name) + if value is None or value == "": + return default + return float(value) + + +@dataclass +class ServiceConfig: + model_id: str = os.environ.get("DIFFUSERS_WAN_I2V_MODEL_ID", DEFAULT_MODEL_ID) + output_dir: Path = Path(os.environ.get("DIFFUSERS_WAN_I2V_OUTPUT_DIR", DEFAULT_OUTPUT_DIR)) + host: str = os.environ.get("DIFFUSERS_WAN_I2V_HOST", DEFAULT_HOST) + port: int = _coerce_env_int("DIFFUSERS_WAN_I2V_PORT", DEFAULT_PORT) + torch_dtype: torch.dtype = torch.bfloat16 + device: str = os.environ.get("DIFFUSERS_WAN_I2V_DEVICE", "cuda") + guidance_scale: float = _coerce_env_float("DIFFUSERS_WAN_I2V_GUIDANCE_SCALE", DEFAULT_GUIDANCE_SCALE) + num_frames: int = _coerce_env_int("DIFFUSERS_WAN_I2V_NUM_FRAMES", DEFAULT_NUM_FRAMES) + num_inference_steps: int = _coerce_env_int( + "DIFFUSERS_WAN_I2V_NUM_INFERENCE_STEPS", DEFAULT_NUM_INFERENCE_STEPS + ) + target_fps: int = _coerce_env_int("DIFFUSERS_WAN_I2V_TARGET_FPS", DEFAULT_TARGET_FPS) + width: int = _coerce_env_int("DIFFUSERS_WAN_I2V_WIDTH", DEFAULT_WIDTH) + height: int = _coerce_env_int("DIFFUSERS_WAN_I2V_HEIGHT", DEFAULT_HEIGHT) + timeout_seconds: float = _coerce_env_float("DIFFUSERS_WAN_I2V_TIMEOUT_SECONDS", DEFAULT_TIMEOUT_SECONDS) + + +@dataclass +class JobRecord: + job_id: str + prompt: str + model: str + size: str + seconds: int + status: str = VIDEO_STATUS_QUEUED + progress: int = 0 + created_at: int = field(default_factory=lambda: int(time.time())) + completed_at: int | None = None + output_path: str | None = None + error: dict[str, Any] | None = None + inference_time_s: float | None = None + peak_memory_mb: float | None = None + + def to_response(self, base_url: str) -> dict[str, Any]: + url = None + if self.status == VIDEO_STATUS_COMPLETED: + url = f"{base_url}/v1/videos/{self.job_id}/content" + return { + "id": self.job_id, + "object": "video", + "model": self.model, + "status": self.status, + "progress": self.progress, + "created_at": self.created_at, + "size": self.size, + "seconds": str(self.seconds), + "quality": "standard", + "url": url, + "completed_at": self.completed_at, + "error": self.error, + "file_path": self.output_path, + "inference_time_s": self.inference_time_s, + "peak_memory_mb": self.peak_memory_mb, + } + + +class DiffusersWanService: + """Standalone async HTTP service for the official Diffusers Wan2.1 I2V 480P pipeline.""" + + def __init__(self, config: ServiceConfig) -> None: + self.config = config + self.output_dir = config.output_dir + self.input_dir = self.output_dir / "inputs" + self.video_dir = self.output_dir / "videos" + self.output_dir.mkdir(parents=True, exist_ok=True) + self.input_dir.mkdir(parents=True, exist_ok=True) + self.video_dir.mkdir(parents=True, exist_ok=True) + self._jobs: dict[str, JobRecord] = {} + self._lock = threading.RLock() + self._pipeline = self._load_pipeline() + + def _load_pipeline(self) -> WanImageToVideoPipeline: + image_encoder = CLIPVisionModel.from_pretrained( + self.config.model_id, + subfolder="image_encoder", + torch_dtype=torch.float32, + ) + vae = AutoencoderKLWan.from_pretrained( + self.config.model_id, + subfolder="vae", + torch_dtype=torch.float32, + ) + pipe = WanImageToVideoPipeline.from_pretrained( + self.config.model_id, + vae=vae, + image_encoder=image_encoder, + torch_dtype=self.config.torch_dtype, + ) + pipe.to(self.config.device) + return pipe + + def create_job(self, *, prompt: str, model: str, size: str, seconds: int) -> JobRecord: + job_id = str(uuid.uuid4()) + job = JobRecord( + job_id=job_id, + prompt=prompt, + model=model or self.config.model_id, + size=size, + seconds=seconds, + ) + with self._lock: + self._jobs[job_id] = job + return job + + def get_job(self, job_id: str) -> JobRecord | None: + with self._lock: + return self._jobs.get(job_id) + + def list_jobs(self) -> list[JobRecord]: + with self._lock: + return list(self._jobs.values()) + + def cancel_job(self, job_id: str) -> JobRecord | None: + with self._lock: + job = self._jobs.get(job_id) + if job is None: + return None + if job.status in {VIDEO_STATUS_COMPLETED, VIDEO_STATUS_FAILED, VIDEO_STATUS_CANCELLED}: + return job + job.status = VIDEO_STATUS_CANCELLED + job.progress = 0 + job.completed_at = int(time.time()) + job.error = {"message": "Job cancelled by user"} + return job + + def _resolve_image_from_reference( + self, + *, + input_reference_file: UploadFile | None, + input_reference_value: str | None, + reference_url: str | None, + job_id: str, + ) -> Image.Image: + if input_reference_file is not None and input_reference_file.filename: + content = input_reference_file.file.read() + if not content: + raise HTTPException(status_code=400, detail="Uploaded input_reference is empty") + image = Image.open(io.BytesIO(content)).convert("RGB") + suffix = Path(input_reference_file.filename).suffix or ".png" + (self.input_dir / f"{job_id}{suffix}").write_bytes(content) + return image + + ref = input_reference_value or reference_url + if not ref: + raise HTTPException(status_code=400, detail="I2V benchmark service requires input_reference or reference_url") + + if ref.startswith("http://") or ref.startswith("https://"): + raise HTTPException( + status_code=400, + detail="Remote HTTP reference_url is not supported by this baseline service; use a local path or upload", + ) + + if ref.startswith("data:"): + try: + _, b64_data = ref.split(",", 1) + image_bytes = base64.b64decode(b64_data, validate=True) + except (ValueError, binascii.Error) as exc: + raise HTTPException(status_code=400, detail=f"Malformed data URL for input reference: {exc}") from exc + image = Image.open(io.BytesIO(image_bytes)).convert("RGB") + (self.input_dir / f"{job_id}.png").write_bytes(image_bytes) + return image + + path = Path(ref) + if not path.exists() or not path.is_file(): + raise HTTPException(status_code=400, detail=f"Reference image not found: {ref}") + return Image.open(path).convert("RGB") + + def _run_job(self, job_id: str, image: Image.Image, negative_prompt: str | None, seed: int | None) -> None: + job = self.get_job(job_id) + if job is None: + return + + with self._lock: + if job.status == VIDEO_STATUS_CANCELLED: + return + job.status = VIDEO_STATUS_GENERATING + job.progress = 5 + + output_path = self.video_dir / f"{job_id}.mp4" + start = time.perf_counter() + peak_memory_mb: float | None = None + + try: + if torch.cuda.is_available(): + torch.cuda.reset_peak_memory_stats() + + generator = None + if seed is not None: + generator = torch.Generator(device=self.config.device).manual_seed(int(seed)) + + result = self._pipeline( + image=image, + prompt=job.prompt, + negative_prompt=negative_prompt, + height=self.config.height, + width=self.config.width, + num_frames=self.config.num_frames, + num_inference_steps=self.config.num_inference_steps, + guidance_scale=self.config.guidance_scale, + generator=generator, + ) + frames = result.frames[0] + export_to_video(frames, str(output_path), fps=self.config.target_fps) + + elapsed = time.perf_counter() - start + if torch.cuda.is_available(): + peak_memory_mb = torch.cuda.max_memory_allocated() / (1024 * 1024) + + with self._lock: + if job.status == VIDEO_STATUS_CANCELLED: + output_path.unlink(missing_ok=True) + return + job.status = VIDEO_STATUS_COMPLETED + job.progress = 100 + job.completed_at = int(time.time()) + job.output_path = str(output_path) + job.inference_time_s = elapsed + job.peak_memory_mb = peak_memory_mb + except Exception as exc: + with self._lock: + job.status = VIDEO_STATUS_FAILED + job.progress = 0 + job.completed_at = int(time.time()) + job.error = {"message": str(exc)} + finally: + if torch.cuda.is_available(): + torch.cuda.empty_cache() + + def submit_job( + self, + *, + prompt: str, + model: str | None, + size: str | None, + seconds: int | None, + seed: int | None, + negative_prompt: str | None, + input_reference_file: UploadFile | None, + input_reference_value: str | None, + reference_url: str | None, + ) -> JobRecord: + job = self.create_job( + prompt=prompt, + model=model or self.config.model_id, + size=size or f"{self.config.width}x{self.config.height}", + seconds=seconds or max(1, self.config.num_frames // self.config.target_fps), + ) + image = self._resolve_image_from_reference( + input_reference_file=input_reference_file, + input_reference_value=input_reference_value, + reference_url=reference_url, + job_id=job.job_id, + ) + worker = threading.Thread( + target=self._run_job, + args=(job.job_id, image, negative_prompt, seed), + daemon=True, + name=f"diffusers-wan-job-{job.job_id}", + ) + worker.start() + return job + + +service_config = ServiceConfig() +service = DiffusersWanService(service_config) +app = FastAPI( + title="Diffusers Wan2.1 I2V Benchmark Service", + description="Standalone async HTTP baseline for Wan2.1-I2V-14B-480P using official Diffusers.", + version="0.1.0", +) + + +@app.get("/v1/service/health") +async def health() -> dict[str, Any]: + return { + "status": "healthy", + "service": "diffusers-wan-i2v", + "model_id": service_config.model_id, + "device": service_config.device, + "workload": { + "width": service_config.width, + "height": service_config.height, + "num_frames": service_config.num_frames, + "num_inference_steps": service_config.num_inference_steps, + "guidance_scale": service_config.guidance_scale, + "fps": service_config.target_fps, + }, + } + + +@app.get("/metrics") +async def metrics() -> Response: + lines = [ + "# HELP diffusers_wan_jobs_total Number of tracked jobs", + "# TYPE diffusers_wan_jobs_total gauge", + f"diffusers_wan_jobs_total {len(service.list_jobs())}", + ] + completed = sum(1 for job in service.list_jobs() if job.status == VIDEO_STATUS_COMPLETED) + failed = sum(1 for job in service.list_jobs() if job.status == VIDEO_STATUS_FAILED) + lines.extend( + [ + "# HELP diffusers_wan_jobs_completed_total Number of completed jobs", + "# TYPE diffusers_wan_jobs_completed_total counter", + f"diffusers_wan_jobs_completed_total {completed}", + "# HELP diffusers_wan_jobs_failed_total Number of failed jobs", + "# TYPE diffusers_wan_jobs_failed_total counter", + f"diffusers_wan_jobs_failed_total {failed}", + ] + ) + return Response("\n".join(lines) + "\n", media_type="text/plain; version=0.0.4") + + +@app.post("/v1/videos") +async def create_video( + request: Request, + prompt: str | None = Form(None), + input_reference: UploadFile | None = File(None), + reference_url: str | None = Form(None), + model: str | None = Form(None), + seconds: int | None = Form(None), + size: str | None = Form(None), + seed: int | None = Form(None), + negative_prompt: str | None = Form(None), +) -> JSONResponse: + content_type = request.headers.get("content-type", "").lower() + + if "application/json" in content_type: + body = await request.json() + prompt_value = body.get("prompt") + input_reference_value = body.get("input_reference") + reference_url_value = body.get("reference_url") + model_value = body.get("model") + seconds_value = body.get("seconds") + size_value = body.get("size") + seed_value = body.get("seed") + negative_prompt_value = body.get("negative_prompt") + upload = None + else: + prompt_value = prompt + input_reference_value = None + reference_url_value = reference_url + model_value = model + seconds_value = seconds + size_value = size + seed_value = seed + negative_prompt_value = negative_prompt + upload = input_reference + + if not prompt_value: + raise HTTPException(status_code=422, detail="prompt is required") + + job = service.submit_job( + prompt=prompt_value, + model=model_value, + size=size_value, + seconds=seconds_value, + seed=seed_value, + negative_prompt=negative_prompt_value, + input_reference_file=upload, + input_reference_value=input_reference_value, + reference_url=reference_url_value, + ) + base_url = str(request.base_url).rstrip("/") + return JSONResponse(job.to_response(base_url)) + + +@app.get("/v1/videos") +async def list_videos(request: Request) -> dict[str, Any]: + base_url = str(request.base_url).rstrip("/") + jobs = sorted(service.list_jobs(), key=lambda item: item.created_at, reverse=True) + return { + "object": "list", + "data": [job.to_response(base_url) for job in jobs], + "has_more": False, + "first_id": jobs[0].job_id if jobs else None, + "last_id": jobs[-1].job_id if jobs else None, + } + + +@app.get("/v1/videos/{video_id}") +async def retrieve_video(video_id: str, request: Request) -> dict[str, Any]: + job = service.get_job(video_id) + if job is None: + raise HTTPException(status_code=404, detail=f"Video {video_id} not found") + base_url = str(request.base_url).rstrip("/") + return job.to_response(base_url) + + +@app.delete("/v1/videos/{video_id}") +async def delete_video(video_id: str, request: Request) -> dict[str, Any]: + job = service.cancel_job(video_id) + if job is None: + raise HTTPException(status_code=404, detail=f"Video {video_id} not found") + base_url = str(request.base_url).rstrip("/") + return job.to_response(base_url) + + +@app.get("/v1/videos/{video_id}/content") +async def get_video_content(video_id: str) -> FileResponse: + job = service.get_job(video_id) + if job is None: + raise HTTPException(status_code=404, detail=f"Video {video_id} not found") + if job.status != VIDEO_STATUS_COMPLETED or not job.output_path: + raise HTTPException(status_code=400, detail=f"Video {video_id} is not ready") + path = Path(job.output_path) + if not path.exists(): + raise HTTPException(status_code=404, detail=f"Video content not found: {job.output_path}") + return FileResponse(path, media_type="video/mp4", filename=path.name) + + +def main() -> None: + uvicorn.run(app, host=service_config.host, port=service_config.port, log_level="warning") + + +if __name__ == "__main__": + main() diff --git a/benchmarks/telefuser_aiperf/README.md b/benchmarks/telefuser_aiperf/README.md new file mode 100644 index 0000000..795ecd7 --- /dev/null +++ b/benchmarks/telefuser_aiperf/README.md @@ -0,0 +1,110 @@ +# TeleFuser AIPerf Benchmarks + +This directory contains TeleFuser-specific benchmark assets for running end-to-end video serving tests with AIPerf. + +Current scope: +- `telefuser serve` batch video generation over the OpenAI-compatible `/v1/videos` API +- `telefuser stream-serve` WebRTC end-to-end session benchmarking +- End-to-end latency, throughput, HTTP trace, frame-level latency, and optional server-metrics scraping + +## Layout + +- `configs/`: AIPerf YAML configs for batch video serving and JSON configs for stream benchmarking +- `data/`: Prompt files and control traces +- `scripts/`: Helper scripts for launching batch and stream benchmarks + +## Prerequisites + +1. Install TeleFuser and start a batch video server. +2. Install AIPerf from the vendored clone under `benchmarks/aiperf`. + +Example TeleFuser startup: + +```bash +telefuser serve \ + examples/wan_video/wan21_14b_image_to_video_h100.py \ + --port 8000 \ + --task i2v +``` + +Example AIPerf install: + +```bash +pip install -e ./benchmarks/aiperf +``` + +## Quick Start + +### Batch Video + +Run a small benchmark against TeleFuser `/v1/videos`: + +```bash +bash benchmarks/telefuser_aiperf/scripts/run_video_bench.sh +``` + +This uses: +- `configs/video_generation_quick.yaml` +- file-backed prompts from `data/video_prompts.jsonl` +- TeleFuser server at `http://127.0.0.1:8000` + +### Stream Serve + +Start a TeleFuser stream server first: + +```bash +telefuser stream-serve \ + examples/lingbot/stream_lingbot_world_fast.py \ + -p 8088 \ + --skip-validation +``` + +Then run the WebRTC benchmark: + +```bash +bash benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh +``` + +This uses: +- `configs/stream_lingbot_world_fast_quick.json` +- control trace from `data/stream_lingbot_controls.json` +- TeleFuser stream server at `http://127.0.0.1:8088` + +If no explicit ICE allowlist is provided, `run_stream_bench.py` will auto-detect a non-loopback local IP for WebRTC host candidate gathering. Only pin `--ice-host-ip` or `TELEFUSER_WEBRTC_ICE_HOST_IPS` when you need a specific address. + +The script records: +- SDP offer / answer RTT +- WebRTC connect latency +- first-frame latency +- steady-state stream FPS +- control acknowledge latency +- control-to-next-frame latency +- the helper raises the open-file limit to `8192` by default; override with `TELEFUSER_BENCH_NOFILE_LIMIT` if needed + +## Config Notes + +TeleFuser and AIPerf line up well for batch video serving: +- TeleFuser exposes `/v1/videos`, `/v1/videos/{id}`, and `/v1/videos/{id}/content` +- AIPerf `video_generation` already speaks the same async submit/poll/download flow + +The provided configs intentionally: +- use `endpoint.type: video_generation` +- enable `downloadVideoContent` only in the full E2E variant +- point `serverMetrics.urls` at `/v1/service/metrics` only when metrics scraping is desired +- do not enable AIPerf's built-in readiness probe, because TeleFuser batch video serving does not expose a stable `/v1/models` or chat-style readiness path for `video_generation` + +Instead, the helper script checks: + +```bash +curl http://127.0.0.1:8000/v1/service/health +``` + +## Stream Scope Notes + +The current stream benchmark is a TeleFuser-specific harness layered next to AIPerf, not inside AIPerf core yet. + +Reasons: +- `telefuser stream-serve` is WebRTC session based, not plain HTTP request / response +- world-model streaming needs frame-level and control-loop metrics +- current `LingBotWorldFastService` only allows one active session at a time, so the harness supports concurrency parameters but this pipeline is effectively single-session today +- `python3 scripts/remote_bench_sync.py stream-bootstrap` will auto-detect the usable model zoo root and create a repo-local `model_zoo` symlink on the remote host, so the default relative path works without extra manual env setup diff --git a/benchmarks/telefuser_aiperf/benchmark_contract.yaml b/benchmarks/telefuser_aiperf/benchmark_contract.yaml new file mode 100644 index 0000000..ef0110f --- /dev/null +++ b/benchmarks/telefuser_aiperf/benchmark_contract.yaml @@ -0,0 +1,40 @@ +contract_version: v1 +name: telefuser_aiperf_video +mode: batch_video +implementation: telefuser +model_family: wan_video +model: Wan2.1-I2V-14B-480P +supported_tasks: + - i2v +transport: http +endpoint: + path: /v1/videos + protocol: openai_video_generation +request_encoding: + content_type: multipart/form-data + media_inputs: + first_image_path: input_reference + parameters: + prompt: prompt + negative_prompt: negative_prompt + seed: seed + size: size + seconds: seconds +result_delivery: + polling_path: /v1/videos/{id} + content_path: /v1/videos/{id}/content +workload: + resolution: 832x480 + frames: 81 + inference_steps: 40 + fps: 16 + guidance_scale: 5.0 +metrics: + - request_latency + - completion_latency + - download_latency + - request_throughput + - success_rate +artifacts: + config: benchmarks/telefuser_aiperf/configs/video_generation_wan21_i2v_480p_compare.yaml + dataset: benchmarks/telefuser_aiperf/data/video_prompts.jsonl diff --git a/benchmarks/telefuser_aiperf/configs/video_generation_e2e.yaml b/benchmarks/telefuser_aiperf/configs/video_generation_e2e.yaml new file mode 100644 index 0000000..be4d057 --- /dev/null +++ b/benchmarks/telefuser_aiperf/configs/video_generation_e2e.yaml @@ -0,0 +1,53 @@ +# yaml-language-server: $schema=../../aiperf/src/aiperf/config/schema/aiperf-config.schema.json + +schemaVersion: "2.0" + +randomSeed: 42 + +benchmark: + model: telefuser-video + + endpoint: + url: ${TELEFUSER_AIPERF_URL:http://127.0.0.1:8000} + type: video_generation + timeout: 3600.0 + downloadVideoContent: false + extra: + size: ${TELEFUSER_AIPERF_SIZE:1280x720} + seconds: ${TELEFUSER_AIPERF_SECONDS:4} + seed: ${TELEFUSER_AIPERF_SEED:1024} + + tokenizer: + name: builtin + + dataset: + type: file + path: ./benchmarks/telefuser_aiperf/data/video_prompts.jsonl + format: single_turn + sampling: sequential + + warmup: + type: concurrency + concurrency: 1 + requests: 1 + excludeFromResults: true + + profiling: + type: concurrency + concurrency: ${TELEFUSER_AIPERF_CONCURRENCY:2} + requests: ${TELEFUSER_AIPERF_REQUESTS:6} + duration: ${TELEFUSER_AIPERF_DURATION:7200} + gracePeriod: ${TELEFUSER_AIPERF_GRACE_PERIOD:900} + + artifacts: + dir: ./artifacts/telefuser_aiperf/video_e2e + summary: [json] + records: [jsonl] + showTraceTiming: true + trace: true + + serverMetrics: + enabled: ${TELEFUSER_AIPERF_SERVER_METRICS:true} + urls: + - ${TELEFUSER_AIPERF_METRICS_URL:http://127.0.0.1:8000/metrics} + formats: [json, csv] diff --git a/benchmarks/telefuser_aiperf/configs/video_generation_quick.yaml b/benchmarks/telefuser_aiperf/configs/video_generation_quick.yaml new file mode 100644 index 0000000..1d6057c --- /dev/null +++ b/benchmarks/telefuser_aiperf/configs/video_generation_quick.yaml @@ -0,0 +1,32 @@ +# yaml-language-server: $schema=../../aiperf/src/aiperf/config/schema/aiperf-config.schema.json + +schemaVersion: "2.0" + +randomSeed: 42 + +benchmark: + model: telefuser-video + + endpoint: + url: ${TELEFUSER_AIPERF_URL:http://127.0.0.1:8000} + type: video_generation + timeout: 1800.0 + + tokenizer: + name: builtin + + dataset: + type: file + path: ./benchmarks/telefuser_aiperf/data/video_prompts.jsonl + format: single_turn + sampling: sequential + + profiling: + type: concurrency + concurrency: 1 + requests: 1 + + artifacts: + dir: ./artifacts/telefuser_aiperf/video_quick + summary: [json] + records: [jsonl] diff --git a/benchmarks/telefuser_aiperf/configs/video_generation_rate.yaml b/benchmarks/telefuser_aiperf/configs/video_generation_rate.yaml new file mode 100644 index 0000000..d6623f1 --- /dev/null +++ b/benchmarks/telefuser_aiperf/configs/video_generation_rate.yaml @@ -0,0 +1,38 @@ +# yaml-language-server: $schema=../../aiperf/src/aiperf/config/schema/aiperf-config.schema.json + +schemaVersion: "2.0" + +randomSeed: 42 + +benchmark: + model: telefuser-video + + endpoint: + url: ${TELEFUSER_AIPERF_URL:http://127.0.0.1:8000} + type: video_generation + timeout: 3600.0 + extra: + size: ${TELEFUSER_AIPERF_SIZE:1024x576} + seconds: ${TELEFUSER_AIPERF_SECONDS:4} + + tokenizer: + name: builtin + + dataset: + type: file + path: ./benchmarks/telefuser_aiperf/data/video_prompts.jsonl + format: single_turn + sampling: sequential + + profiling: + type: poisson + rate: ${TELEFUSER_AIPERF_RATE:0.2} + concurrency: ${TELEFUSER_AIPERF_MAX_CONCURRENCY:4} + requests: ${TELEFUSER_AIPERF_REQUESTS:10} + gracePeriod: 300 + + artifacts: + dir: ./artifacts/telefuser_aiperf/video_rate + summary: [json] + records: [jsonl] + showTraceTiming: true diff --git a/benchmarks/telefuser_aiperf/configs/video_generation_wan21_i2v_480p_compare.yaml b/benchmarks/telefuser_aiperf/configs/video_generation_wan21_i2v_480p_compare.yaml new file mode 100644 index 0000000..d0b5f47 --- /dev/null +++ b/benchmarks/telefuser_aiperf/configs/video_generation_wan21_i2v_480p_compare.yaml @@ -0,0 +1,53 @@ +# yaml-language-server: $schema=../../aiperf/src/aiperf/config/schema/aiperf-config.schema.json + +schemaVersion: "2.0" + +randomSeed: 42 + +benchmark: + model: telefuser-wan21-i2v-480p + + endpoint: + url: ${TELEFUSER_AIPERF_URL:http://127.0.0.1:8000} + type: video_generation + timeout: 7200.0 + downloadVideoContent: true + extra: + size: ${TELEFUSER_AIPERF_SIZE:480p} + seconds: ${TELEFUSER_AIPERF_SECONDS:5} + seed: ${TELEFUSER_AIPERF_SEED:42} + + tokenizer: + name: builtin + + dataset: + type: file + path: ./benchmarks/telefuser_aiperf/data/video_prompts.jsonl + format: single_turn + sampling: sequential + + warmup: + type: concurrency + concurrency: 1 + requests: 1 + excludeFromResults: true + + profiling: + type: concurrency + concurrency: ${TELEFUSER_AIPERF_CONCURRENCY:1} + requests: ${TELEFUSER_AIPERF_REQUESTS:2} + duration: ${TELEFUSER_AIPERF_DURATION:7200} + gracePeriod: ${TELEFUSER_AIPERF_GRACE_PERIOD:900} + + artifacts: + dir: ./artifacts/telefuser_aiperf/wan21_i2v_480p_compare + summary: [json] + records: [jsonl] + showTraceTiming: true + trace: true + + serverMetrics: + enabled: ${TELEFUSER_AIPERF_SERVER_METRICS:false} + urls: + - ${TELEFUSER_AIPERF_METRICS_URL:http://127.0.0.1:8000/v1/service/metrics} + formats: [json, csv] diff --git a/benchmarks/telefuser_aiperf/data/video_prompts.jsonl b/benchmarks/telefuser_aiperf/data/video_prompts.jsonl new file mode 100644 index 0000000..c080bf1 --- /dev/null +++ b/benchmarks/telefuser_aiperf/data/video_prompts.jsonl @@ -0,0 +1,5 @@ +{"text": "The character turns slightly toward the camera and breathes naturally, cinematic motion, consistent identity", "image": "/workspace/TeleFuser/examples/data/101235-video-720_0.png"} +{"text": "The subject begins walking forward with subtle parallax in the background, natural lighting, smooth motion", "image": "/workspace/TeleFuser/examples/data/101235-video-720_0.png"} +{"text": "A gentle push-in camera move with soft environmental motion, realistic dynamics, no sudden cuts", "image": "/workspace/TeleFuser/examples/data/101235-video-720_0.png"} +{"text": "The scene gains light wind and small secondary motion while preserving composition and subject details", "image": "/workspace/TeleFuser/examples/data/101235-video-720_0.png"} +{"text": "A short cinematic continuation with stable structure, coherent motion, and clean temporal consistency", "image": "/workspace/TeleFuser/examples/data/101235-video-720_0.png"} diff --git a/benchmarks/telefuser_aiperf/scripts/run_stream_bench.py b/benchmarks/telefuser_aiperf/scripts/run_stream_bench.py new file mode 100644 index 0000000..6b3bda8 --- /dev/null +++ b/benchmarks/telefuser_aiperf/scripts/run_stream_bench.py @@ -0,0 +1,872 @@ +#!/usr/bin/env python3 +"""Run an end-to-end WebRTC stream benchmark against ``telefuser stream-serve``. + +This benchmark is intentionally separate from the vendored AIPerf core because +TeleFuser world-model streaming uses WebRTC session lifecycles instead of plain +HTTP request/response flows. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import os +import subprocess +import sys +import time +import urllib.error +import urllib.request +import uuid +from collections import deque +from dataclasses import asdict, dataclass, field +from pathlib import Path +from typing import Any + +try: + from aiortc import RTCConfiguration, RTCIceServer, RTCPeerConnection, RTCSessionDescription +except ImportError as exc: # pragma: no cover - runtime dependency check + RTCConfiguration = None + RTCIceServer = None + RTCPeerConnection = None + RTCSessionDescription = None + AIORTC_IMPORT_ERROR = exc +else: # pragma: no cover - exercised only when dependency exists + AIORTC_IMPORT_ERROR = None + +from telefuser.webrtc_ice import configure_ice_host_addresses + + +DEFAULT_CONTROL_TRACE = [ + {"delay_s": 1.0, "message": {"type": "control", "key": "ArrowUp", "action": "press"}}, + {"delay_s": 1.8, "message": {"type": "control", "key": "ArrowUp", "action": "release"}}, + {"delay_s": 2.8, "message": {"type": "control", "key": "ArrowLeft", "action": "press"}}, + {"delay_s": 3.6, "message": {"type": "control", "key": "ArrowLeft", "action": "release"}}, + {"delay_s": 4.6, "message": {"type": "control", "key": "ArrowRight", "action": "press"}}, + {"delay_s": 5.4, "message": {"type": "control", "key": "ArrowRight", "action": "release"}}, +] + + +@dataclass +class BenchConfig: + server_url: str = "http://127.0.0.1:8088" + offer_path: str = "/v1/stream/webrtc/offer" + delete_path_template: str = "/v1/stream/webrtc/{session_id}" + mode: str = "bidirectional" + task: str = "bidirectional" + prompt: str = "walk forward through the scene" + image_path: str | None = None + fps: int = 16 + session_count: int = 1 + warmup_sessions: int = 0 + session_duration_s: float = 12.0 + stagger_s: float = 0.0 + connect_timeout_s: float = 30.0 + frame_timeout_s: float = 60.0 + ice_gather_timeout_s: float = 5.0 + shutdown_timeout_s: float = 5.0 + receive_audio: bool = False + control_trace_path: str | None = None + ice_host_ips: list[str] | None = None + request_extra: dict[str, Any] = field(default_factory=dict) + artifacts_dir: str = "artifacts/telefuser_aiperf/stream_bench" + turn_url: str | None = None + turn_username: str | None = None + turn_credential: str | None = None + force_turn_relay: bool = False + print_events: bool = False + + +@dataclass +class ControlEventResult: + index: int + scheduled_delay_s: float + message: dict[str, Any] + sent_offset_s: float + ack_latency_ms: float | None = None + next_frame_latency_ms: float | None = None + + +@dataclass +class SessionResult: + logical_session_index: int + phase: str + mode: str + planned_session_id: str + session_id: str + success: bool = False + error: str | None = None + offer_rtt_ms: float | None = None + connected_latency_ms: float | None = None + first_frame_latency_ms: float | None = None + first_metadata_latency_ms: float | None = None + session_runtime_s: float | None = None + frames_received: int = 0 + metadata_messages: int = 0 + status_messages: int = 0 + done_received: bool = False + stream_fps: float | None = None + last_status_stage: str | None = None + control_events: list[ControlEventResult] = field(default_factory=list) + artifacts_event_file: str | None = None + + +def _parse_json_value(raw: str) -> Any: + try: + return json.loads(raw) + except json.JSONDecodeError: + return raw + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--config", help="Optional JSON config file") + parser.add_argument("--server-url", default=argparse.SUPPRESS, help="TeleFuser stream server base URL") + parser.add_argument("--mode", choices=("server_push", "bidirectional"), default=argparse.SUPPRESS) + parser.add_argument("--task", default=argparse.SUPPRESS, help="Request task field sent in the WebRTC offer body") + parser.add_argument("--prompt", default=argparse.SUPPRESS, help="Prompt sent in the initial offer body") + parser.add_argument("--image-path", default=argparse.SUPPRESS, help="Image path visible to the server process") + parser.add_argument("--fps", type=int, default=argparse.SUPPRESS, help="Requested stream FPS") + parser.add_argument("--session-count", type=int, default=argparse.SUPPRESS, help="Number of profiled sessions") + parser.add_argument("--warmup-sessions", type=int, default=argparse.SUPPRESS, help="Warmup sessions before profile") + parser.add_argument("--session-duration-s", type=float, default=argparse.SUPPRESS) + parser.add_argument("--stagger-s", type=float, default=argparse.SUPPRESS) + parser.add_argument("--connect-timeout-s", type=float, default=argparse.SUPPRESS) + parser.add_argument("--frame-timeout-s", type=float, default=argparse.SUPPRESS) + parser.add_argument("--ice-gather-timeout-s", type=float, default=argparse.SUPPRESS) + parser.add_argument("--shutdown-timeout-s", type=float, default=argparse.SUPPRESS) + parser.add_argument("--control-trace-path", default=argparse.SUPPRESS, help="JSON file with timed control events") + parser.add_argument( + "--ice-host-ip", + action="append", + default=argparse.SUPPRESS, + help="Allowlisted ICE host IP for candidate gathering. Repeat to allow multiple addresses.", + ) + parser.add_argument("--request-extra-json", default=argparse.SUPPRESS, help="Inline JSON merged into request body") + parser.add_argument("--artifacts-dir", default=argparse.SUPPRESS, help="Artifact root directory") + parser.add_argument("--turn-url", default=argparse.SUPPRESS, help="TURN server URL for WebRTC ICE") + parser.add_argument("--turn-username", default=argparse.SUPPRESS) + parser.add_argument("--turn-credential", default=argparse.SUPPRESS) + parser.add_argument("--receive-audio", action=argparse.BooleanOptionalAction, default=argparse.SUPPRESS) + parser.add_argument("--force-turn-relay", action=argparse.BooleanOptionalAction, default=argparse.SUPPRESS) + parser.add_argument("--print-events", action=argparse.BooleanOptionalAction, default=argparse.SUPPRESS) + return parser + + +def _load_config(args: argparse.Namespace) -> BenchConfig: + merged: dict[str, Any] = asdict(BenchConfig()) + if args.config: + file_data = json.loads(Path(args.config).read_text(encoding="utf-8")) + if not isinstance(file_data, dict): + raise ValueError(f"Config file must contain a JSON object: {args.config}") + merged.update(file_data) + + cli_values = vars(args).copy() + cli_values.pop("config", None) + request_extra_json = cli_values.pop("request_extra_json", None) + if request_extra_json is not None: + cli_values["request_extra"] = _parse_json_value(request_extra_json) + if "ice_host_ip" in cli_values: + cli_values["ice_host_ips"] = cli_values.pop("ice_host_ip") + + merged.update(cli_values) + request_extra = merged.get("request_extra") or {} + if not isinstance(request_extra, dict): + raise ValueError("request_extra must be a JSON object") + merged["request_extra"] = request_extra + return BenchConfig(**merged) + + +def _detect_default_ice_host_ips() -> list[str] | None: + """Return a small list of non-loopback host IPs for local ICE gathering.""" + + try: + result = subprocess.run( + ["hostname", "-I"], + check=True, + capture_output=True, + text=True, + ) + except Exception: + return None + + ips: list[str] = [] + for raw_ip in result.stdout.split(): + ip = raw_ip.strip() + if not ip or ip.startswith("127.") or ip == "::1": + continue + if ip not in ips: + ips.append(ip) + + return ips or None + + +def _build_rtc_configuration(config: BenchConfig): + if RTCConfiguration is None or RTCIceServer is None: + raise RuntimeError( + "aiortc is required for stream benchmarking. Install TeleFuser with `pip install -e \".[webrtc]\"`." + ) from AIORTC_IMPORT_ERROR + + ice_servers: list[Any] = [] + if config.turn_url: + ice_server_kwargs = {"urls": config.turn_url} + if config.turn_username: + ice_server_kwargs["username"] = config.turn_username + if config.turn_credential: + ice_server_kwargs["credential"] = config.turn_credential + ice_servers.append(RTCIceServer(**ice_server_kwargs)) + # aiortc 1.14.0 stalls on this benchmark's recvonly video + DataChannel + # flow when MAX_BUNDLE is forced. Use the default bundle policy instead. + return RTCConfiguration(iceServers=ice_servers or None) + + +async def _http_json_request(url: str, *, method: str, timeout_s: float, payload: dict | None = None) -> dict: + headers = {"Content-Type": "application/json"} + data = None if payload is None else json.dumps(payload).encode("utf-8") + request = urllib.request.Request(url=url, data=data, headers=headers, method=method) + + def _run() -> dict: + try: + with urllib.request.urlopen(request, timeout=timeout_s) as response: + body = response.read() + return json.loads(body.decode("utf-8")) if body else {} + except urllib.error.HTTPError as exc: + body = exc.read().decode("utf-8", errors="replace") + raise RuntimeError(f"HTTP {exc.code} {exc.reason}: {body}") from exc + except urllib.error.URLError as exc: + raise RuntimeError(f"Failed to reach {url}: {exc.reason}") from exc + + return await asyncio.to_thread(_run) + + +async def _wait_for_ice_complete(peer_connection, timeout_s: float) -> bool: + deadline = time.perf_counter() + timeout_s + while time.perf_counter() < deadline: + if peer_connection.iceGatheringState == "complete": + return True + await asyncio.sleep(0.05) + return peer_connection.iceGatheringState == "complete" + + +def _load_control_trace(path: str | None, mode: str) -> list[dict[str, Any]]: + if mode != "bidirectional": + return [] + if path is None: + return DEFAULT_CONTROL_TRACE + + payload = json.loads(Path(path).read_text(encoding="utf-8")) + events = payload.get("events", payload) if isinstance(payload, dict) else payload + if not isinstance(events, list): + raise ValueError(f"Control trace must be a list or {{\"events\": [...]}}: {path}") + normalised: list[dict[str, Any]] = [] + for idx, event in enumerate(events): + if not isinstance(event, dict): + raise ValueError(f"Control trace event #{idx} must be a JSON object") + delay_s = float(event.get("delay_s", 0.0)) + message = event.get("message") + if not isinstance(message, dict): + raise ValueError(f"Control trace event #{idx} is missing object field `message`") + normalised.append({"delay_s": delay_s, "message": message}) + return sorted(normalised, key=lambda item: item["delay_s"]) + + +def _percentile(values: list[float], percentile: float) -> float: + if not values: + raise ValueError("percentile() requires at least one value") + ordered = sorted(values) + if len(ordered) == 1: + return ordered[0] + position = (len(ordered) - 1) * percentile + lower = int(position) + upper = min(lower + 1, len(ordered) - 1) + fraction = position - lower + return ordered[lower] + (ordered[upper] - ordered[lower]) * fraction + + +def _summarise(values: list[float]) -> dict[str, float] | None: + if not values: + return None + return { + "count": float(len(values)), + "min": min(values), + "mean": sum(values) / len(values), + "p50": _percentile(values, 0.50), + "p90": _percentile(values, 0.90), + "p99": _percentile(values, 0.99), + "max": max(values), + } + + +class StreamBenchSession: + """Single WebRTC session benchmark runner.""" + + def __init__( + self, + *, + config: BenchConfig, + logical_session_index: int, + phase: str, + artifacts_dir: Path, + rtc_configuration, + control_trace: list[dict[str, Any]], + ) -> None: + self.config = config + self.logical_session_index = logical_session_index + self.phase = phase + self.artifacts_dir = artifacts_dir + self.rtc_configuration = rtc_configuration + self.control_trace = control_trace + self.planned_session_id = f"{phase}-{logical_session_index:03d}-{uuid.uuid4().hex[:10]}" + self.session_id = self.planned_session_id + self.result = SessionResult( + logical_session_index=logical_session_index, + phase=phase, + mode=config.mode, + planned_session_id=self.planned_session_id, + session_id=self.session_id, + ) + self._events: list[dict[str, Any]] = [] + self._pc = None + self._data_channel = None + self._session_started_at = 0.0 + self._offer_sent_at = 0.0 + self._first_frame_at: float | None = None + self._last_frame_at: float | None = None + self._first_metadata_at: float | None = None + self._connected_at: float | None = None + self._connected_event = asyncio.Event() + self._done_event = asyncio.Event() + self._state_monitor_task: asyncio.Task | None = None + self._track_tasks: list[asyncio.Task] = [] + self._control_sender_task: asyncio.Task | None = None + self._pending_control_ack_indices: deque[int] = deque() + self._pending_control_frame_indices: deque[int] = deque() + + def _record_event(self, event_type: str, **payload: Any) -> None: + event = { + "event": event_type, + "session_id": self.session_id, + "logical_session_index": self.logical_session_index, + "phase": self.phase, + "timestamp": time.time(), + } + event.update(payload) + self._events.append(event) + if self.config.print_events: + print(json.dumps(event, ensure_ascii=False)) + + def _build_request_body(self) -> dict[str, Any]: + request_options = dict(self.config.request_extra) + request_fps = int(request_options.get("fps", self.config.fps)) + body = { + "session_id": self.planned_session_id, + "sdp": self._pc.localDescription.sdp, + "type": self._pc.localDescription.type, + "task": self.config.task, + "prompt": self.config.prompt, + "fps": request_fps, + "config": dict(request_options), + } + body.update(request_options) + if self.config.image_path: + body["image_path"] = self.config.image_path + return body + + def _current_state_snapshot(self) -> dict[str, str]: + if self._pc is None: + return {} + return { + "signaling_state": self._pc.signalingState, + "ice_gathering_state": self._pc.iceGatheringState, + "ice_connection_state": self._pc.iceConnectionState, + "connection_state": self._pc.connectionState, + } + + async def _monitor_peer_states(self) -> None: + """Record state transitions even when aiortc callbacks stay quiet.""" + + last_snapshot: dict[str, str] | None = None + while not self._done_event.is_set(): + if self._pc is None: + return + snapshot = self._current_state_snapshot() + if snapshot and snapshot != last_snapshot: + self._record_event("peer_state_snapshot", **snapshot) + last_snapshot = snapshot + if snapshot.get("connection_state") == "connected": + return + await asyncio.sleep(0.5) + + async def _consume_video_track(self, track) -> None: + try: + while True: + await track.recv() + now = time.perf_counter() + self.result.frames_received += 1 + if self._first_frame_at is None: + self._first_frame_at = now + self.result.first_frame_latency_ms = (now - self._session_started_at) * 1000.0 + self._record_event("first_frame") + self._last_frame_at = now + if self._pending_control_frame_indices: + control_index = self._pending_control_frame_indices.popleft() + control_event = self.result.control_events[control_index] + if control_event.next_frame_latency_ms is None: + control_event.next_frame_latency_ms = (now - self._session_started_at) * 1000.0 - ( + control_event.sent_offset_s * 1000.0 + ) + except Exception as exc: + self._record_event("video_track_ended", detail=str(exc)) + + async def _consume_audio_track(self, track) -> None: + try: + while True: + await track.recv() + except Exception as exc: + self._record_event("audio_track_ended", detail=str(exc)) + + def _handle_datachannel_message(self, raw_message: str) -> None: + now = time.perf_counter() + self.result.metadata_messages += 1 + if self._first_metadata_at is None: + self._first_metadata_at = now + self.result.first_metadata_latency_ms = (now - self._session_started_at) * 1000.0 + + try: + payload = json.loads(raw_message) + except json.JSONDecodeError: + self._record_event("datachannel_message_invalid", raw_message=raw_message) + return + + message_type = payload.get("type") + if message_type == "done": + self.result.done_received = True + self._done_event.set() + self._record_event("done_message", payload=payload) + return + + data = payload.get("data") if isinstance(payload.get("data"), dict) else payload + stage = data.get("stage") + if stage is not None: + self.result.status_messages += 1 + self.result.last_status_stage = str(stage) + if self._pending_control_ack_indices and stage in {"control_state", "applying_direction_control"}: + control_index = self._pending_control_ack_indices.popleft() + control_event = self.result.control_events[control_index] + if control_event.ack_latency_ms is None: + control_event.ack_latency_ms = (now - self._session_started_at) * 1000.0 - ( + control_event.sent_offset_s * 1000.0 + ) + self._record_event("datachannel_message", payload=payload) + + async def _send_control_trace(self) -> None: + if self._data_channel is None: + return + + trace_started_at = time.perf_counter() + for event_index, entry in enumerate(self.control_trace): + deadline = trace_started_at + float(entry["delay_s"]) + delay = deadline - time.perf_counter() + if delay > 0: + await asyncio.sleep(delay) + if self._data_channel.readyState != "open": + self._record_event("control_trace_aborted", reason=f"datachannel_state={self._data_channel.readyState}") + return + message = dict(entry["message"]) + payload = json.dumps(message) + sent_at = time.perf_counter() + control_event = ControlEventResult( + index=event_index, + scheduled_delay_s=float(entry["delay_s"]), + message=message, + sent_offset_s=sent_at - self._session_started_at, + ) + self.result.control_events.append(control_event) + self._pending_control_ack_indices.append(event_index) + self._pending_control_frame_indices.append(event_index) + self._data_channel.send(payload) + self._record_event("control_sent", payload=message) + + def _register_peer_callbacks(self) -> None: + @self._pc.on("signalingstatechange") + async def _on_signalingstatechange() -> None: + self._record_event("signaling_state", state=self._pc.signalingState) + + @self._pc.on("icegatheringstatechange") + async def _on_icegatheringstatechange() -> None: + self._record_event("ice_gathering_state", state=self._pc.iceGatheringState) + + @self._pc.on("iceconnectionstatechange") + async def _on_iceconnectionstatechange() -> None: + state = self._pc.iceConnectionState + self._record_event("ice_connection_state", state=state) + if state in {"failed", "closed", "disconnected"}: + self._done_event.set() + + @self._pc.on("connectionstatechange") + async def _on_connectionstatechange() -> None: + state = self._pc.connectionState + self._record_event("connection_state", state=state) + if state == "connected" and self._connected_at is None: + self._connected_at = time.perf_counter() + self.result.connected_latency_ms = (self._connected_at - self._session_started_at) * 1000.0 + self._connected_event.set() + if state in {"failed", "closed", "disconnected"}: + self._done_event.set() + + @self._pc.on("track") + def _on_track(track) -> None: + self._record_event("remote_track", kind=track.kind) + if track.kind == "video": + self._track_tasks.append(asyncio.create_task(self._consume_video_track(track))) + elif track.kind == "audio": + self._track_tasks.append(asyncio.create_task(self._consume_audio_track(track))) + + def _create_bidirectional_channel(self) -> None: + self._data_channel = self._pc.createDataChannel("telefuser") + + @self._data_channel.on("open") + def _on_open() -> None: + self._record_event("datachannel_open") + if self._control_sender_task is None and self.control_trace: + self._control_sender_task = asyncio.create_task(self._send_control_trace()) + + @self._data_channel.on("message") + def _on_message(message) -> None: + if isinstance(message, bytes): + message = message.decode("utf-8", errors="replace") + self._handle_datachannel_message(str(message)) + + @self._data_channel.on("close") + def _on_close() -> None: + self._record_event("datachannel_close") + self._done_event.set() + + async def run(self) -> SessionResult: + self._session_started_at = time.perf_counter() + self._record_event("session_start", mode=self.config.mode) + + self._pc = RTCPeerConnection(configuration=self.rtc_configuration or RTCConfiguration()) + self._register_peer_callbacks() + self._state_monitor_task = asyncio.create_task(self._monitor_peer_states()) + + if self.config.mode == "bidirectional": + self._create_bidirectional_channel() + + self._pc.addTransceiver("video", direction="recvonly") + if self.config.receive_audio: + self._pc.addTransceiver("audio", direction="recvonly") + + try: + offer = await self._pc.createOffer() + await self._pc.setLocalDescription(offer) + ice_complete = await _wait_for_ice_complete(self._pc, self.config.ice_gather_timeout_s) + self._record_event("ice_gathering_complete", complete=ice_complete) + + request_body = self._build_request_body() + self._record_event("offer_request", body=request_body) + self._offer_sent_at = time.perf_counter() + answer = await _http_json_request( + f"{self.config.server_url.rstrip('/')}{self.config.offer_path}", + method="POST", + timeout_s=self.config.connect_timeout_s, + payload=request_body, + ) + answered_at = time.perf_counter() + self.result.offer_rtt_ms = (answered_at - self._offer_sent_at) * 1000.0 + self.session_id = str(answer.get("session_id", self.session_id)) + self.result.session_id = self.session_id + self._record_event("offer_answer", answer=answer) + await self._pc.setRemoteDescription(RTCSessionDescription(sdp=answer["sdp"], type=answer["type"])) + self._record_event("post_answer_state", **self._current_state_snapshot()) + + try: + await asyncio.wait_for(self._connected_event.wait(), timeout=self.config.connect_timeout_s) + except asyncio.TimeoutError: + raise TimeoutError( + f"WebRTC connection did not reach connected state within {self.config.connect_timeout_s:.1f}s" + ) + + try: + await asyncio.wait_for(self._done_event.wait(), timeout=self.config.session_duration_s) + except asyncio.TimeoutError: + self._record_event("session_duration_elapsed", duration_s=self.config.session_duration_s) + + if self._first_frame_at is None: + wait_remaining = self.config.frame_timeout_s + while self._first_frame_at is None and wait_remaining > 0: + started_wait = time.perf_counter() + try: + await asyncio.wait_for(self._done_event.wait(), timeout=min(wait_remaining, 0.25)) + except asyncio.TimeoutError: + pass + wait_remaining -= time.perf_counter() - started_wait + if self._first_frame_at is not None: + break + + if self._first_frame_at is not None: + self.result.success = True + if self._last_frame_at is not None and self._last_frame_at > self._first_frame_at: + self.result.stream_fps = (self.result.frames_received - 1) / (self._last_frame_at - self._first_frame_at) + elif self.result.error is None: + self.result.error = "No video frame received before session shutdown" + except Exception as exc: + self.result.error = str(exc) + self._record_event("session_error", error=str(exc)) + finally: + await self._shutdown() + finished_at = time.perf_counter() + self.result.session_runtime_s = finished_at - self._session_started_at + self._write_event_artifact() + return self.result + + async def _shutdown(self) -> None: + if self._data_channel is not None and self._data_channel.readyState == "open": + try: + self._data_channel.send(json.dumps({"type": "stop"})) + self._record_event("stop_sent") + except Exception as exc: + self._record_event("stop_send_failed", error=str(exc)) + + delete_url = f"{self.config.server_url.rstrip('/')}{self.config.delete_path_template.format(session_id=self.session_id)}" + try: + await _http_json_request(delete_url, method="DELETE", timeout_s=self.config.shutdown_timeout_s) + self._record_event("session_delete") + except Exception as exc: + self._record_event("session_delete_failed", error=str(exc)) + + if self._control_sender_task is not None and not self._control_sender_task.done(): + self._control_sender_task.cancel() + try: + await self._control_sender_task + except asyncio.CancelledError: + pass + except Exception: + pass + + if self._state_monitor_task is not None and not self._state_monitor_task.done(): + self._state_monitor_task.cancel() + try: + await self._state_monitor_task + except asyncio.CancelledError: + pass + except Exception: + pass + + for task in self._track_tasks: + if not task.done(): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + except Exception: + pass + + if self._pc is not None: + try: + await self._pc.close() + except Exception as exc: + self._record_event("peer_close_failed", error=str(exc)) + + def _write_event_artifact(self) -> None: + events_dir = self.artifacts_dir / "events" + events_dir.mkdir(parents=True, exist_ok=True) + event_file = events_dir / f"{self.phase}_{self.logical_session_index:03d}_{self.session_id}.jsonl" + with event_file.open("w", encoding="utf-8") as handle: + for event in self._events: + handle.write(json.dumps(event, ensure_ascii=False) + "\n") + self.result.artifacts_event_file = str(event_file) + + +async def _run_phase( + *, + config: BenchConfig, + phase: str, + session_count: int, + artifacts_dir: Path, + rtc_configuration, + control_trace: list[dict[str, Any]], +) -> list[SessionResult]: + sessions: list[StreamBenchSession] = [] + tasks: list[asyncio.Task] = [] + for index in range(session_count): + session = StreamBenchSession( + config=config, + logical_session_index=index, + phase=phase, + artifacts_dir=artifacts_dir, + rtc_configuration=rtc_configuration, + control_trace=control_trace, + ) + sessions.append(session) + + async def _start_session(current_session: StreamBenchSession, delay_s: float) -> SessionResult: + if delay_s > 0: + await asyncio.sleep(delay_s) + return await current_session.run() + + tasks.append(asyncio.create_task(_start_session(session, config.stagger_s * index))) + return list(await asyncio.gather(*tasks)) + + +def _result_to_dict(result: SessionResult) -> dict[str, Any]: + payload = asdict(result) + payload["control_events"] = [asdict(event) for event in result.control_events] + return payload + + +def _build_summary( + *, + config: BenchConfig, + warmup_results: list[SessionResult], + profile_results: list[SessionResult], + started_at_iso: str, +) -> dict[str, Any]: + successful_profile_results = [result for result in profile_results if result.success] + all_control_events = [ + control_event + for result in profile_results + for control_event in result.control_events + if control_event.ack_latency_ms is not None or control_event.next_frame_latency_ms is not None + ] + summary = { + "started_at_utc": started_at_iso, + "config": asdict(config), + "warmup": { + "attempted_sessions": len(warmup_results), + "successful_sessions": sum(1 for result in warmup_results if result.success), + "failed_sessions": sum(1 for result in warmup_results if not result.success), + }, + "profile": { + "attempted_sessions": len(profile_results), + "successful_sessions": len(successful_profile_results), + "failed_sessions": len(profile_results) - len(successful_profile_results), + "success_rate": (len(successful_profile_results) / len(profile_results)) if profile_results else 0.0, + "metrics": { + "offer_rtt_ms": _summarise([result.offer_rtt_ms for result in profile_results if result.offer_rtt_ms]), + "connected_latency_ms": _summarise( + [result.connected_latency_ms for result in profile_results if result.connected_latency_ms] + ), + "first_frame_latency_ms": _summarise( + [result.first_frame_latency_ms for result in profile_results if result.first_frame_latency_ms] + ), + "first_metadata_latency_ms": _summarise( + [ + result.first_metadata_latency_ms + for result in profile_results + if result.first_metadata_latency_ms is not None + ] + ), + "stream_fps": _summarise([result.stream_fps for result in profile_results if result.stream_fps]), + "session_runtime_s": _summarise( + [result.session_runtime_s for result in profile_results if result.session_runtime_s] + ), + "frames_received": _summarise([float(result.frames_received) for result in profile_results]), + "control_ack_latency_ms": _summarise( + [event.ack_latency_ms for event in all_control_events if event.ack_latency_ms is not None] + ), + "control_to_next_frame_latency_ms": _summarise( + [ + event.next_frame_latency_ms + for event in all_control_events + if event.next_frame_latency_ms is not None + ] + ), + }, + }, + } + return summary + + +def _print_console_summary(summary: dict[str, Any]) -> None: + profile = summary["profile"] + metrics = profile["metrics"] + print( + f"Profile sessions: {profile['successful_sessions']}/{profile['attempted_sessions']} succeeded " + f"(success_rate={profile['success_rate']:.2%})" + ) + for metric_name in ( + "offer_rtt_ms", + "connected_latency_ms", + "first_frame_latency_ms", + "stream_fps", + "control_ack_latency_ms", + "control_to_next_frame_latency_ms", + ): + metric = metrics.get(metric_name) + if metric is None: + print(f"{metric_name}: n/a") + continue + print( + f"{metric_name}: count={int(metric['count'])} " + f"mean={metric['mean']:.2f} p50={metric['p50']:.2f} p90={metric['p90']:.2f} max={metric['max']:.2f}" + ) + + +async def _run_benchmark(config: BenchConfig) -> dict[str, Any]: + if RTCPeerConnection is None: + raise RuntimeError( + "aiortc is required for stream benchmarking. Install TeleFuser with `pip install -e \".[webrtc]\"`." + ) from AIORTC_IMPORT_ERROR + + resolved_ice_host_ips = config.ice_host_ips + if not resolved_ice_host_ips: + resolved_ice_host_ips = _detect_default_ice_host_ips() + config.ice_host_ips = resolved_ice_host_ips + configure_ice_host_addresses(resolved_ice_host_ips) + started_at_iso = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + control_trace = _load_control_trace(config.control_trace_path, config.mode) + artifacts_root = Path(config.artifacts_dir) / time.strftime("%Y%m%d_%H%M%S", time.gmtime()) + artifacts_root.mkdir(parents=True, exist_ok=True) + rtc_configuration = _build_rtc_configuration(config) + + warmup_results = await _run_phase( + config=config, + phase="warmup", + session_count=config.warmup_sessions, + artifacts_dir=artifacts_root, + rtc_configuration=rtc_configuration, + control_trace=control_trace, + ) + profile_results = await _run_phase( + config=config, + phase="profile", + session_count=config.session_count, + artifacts_dir=artifacts_root, + rtc_configuration=rtc_configuration, + control_trace=control_trace, + ) + + summary = _build_summary( + config=config, + warmup_results=warmup_results, + profile_results=profile_results, + started_at_iso=started_at_iso, + ) + summary_path = artifacts_root / "summary.json" + sessions_path = artifacts_root / "sessions.jsonl" + summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + with sessions_path.open("w", encoding="utf-8") as handle: + for result in [*warmup_results, *profile_results]: + handle.write(json.dumps(_result_to_dict(result), ensure_ascii=False) + "\n") + + summary["artifacts_dir"] = str(artifacts_root) + summary["summary_path"] = str(summary_path) + summary["sessions_path"] = str(sessions_path) + return summary + + +def main() -> None: + parser = _build_parser() + args = parser.parse_args() + config = _load_config(args) + summary = asyncio.run(_run_benchmark(config)) + _print_console_summary(summary) + print(f"Artifacts: {summary['artifacts_dir']}") + print(f"Summary JSON: {summary['summary_path']}") + print(f"Session records: {summary['sessions_path']}") + + +if __name__ == "__main__": + main() diff --git a/benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh b/benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh new file mode 100644 index 0000000..3dea4f0 --- /dev/null +++ b/benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh @@ -0,0 +1,42 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)" +cd "${ROOT_DIR}" +export PYTHONPATH="${ROOT_DIR}${PYTHONPATH:+:${PYTHONPATH}}" + +CONFIG_PATH="${1:-benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_quick.json}" +if [[ $# -gt 0 ]]; then + shift +fi + +NOFILE_LIMIT="${TELEFUSER_BENCH_NOFILE_LIMIT:-8192}" +if ! ulimit -n "${NOFILE_LIMIT}" >/dev/null 2>&1; then + echo "Warning: failed to raise open-file limit to ${NOFILE_LIMIT}" >&2 +fi + +SERVER_URL="${TELEFUSER_STREAM_BENCH_URL:-http://127.0.0.1:8088}" +HEALTH_URL="${TELEFUSER_STREAM_BENCH_HEALTH_URL:-${SERVER_URL}/v1/service/health}" +PYTHON_BIN="${TELEFUSER_STREAM_BENCH_PYTHON:-python3}" +ICE_HOST_IPS="${TELEFUSER_STREAM_BENCH_ICE_HOST_IPS:-}" +ICE_HOST_ARGS=() +if [[ -n "${ICE_HOST_IPS}" ]]; then + IFS=',' read -r -a _ICE_HOST_IP_ARRAY <<< "${ICE_HOST_IPS}" + for ice_host_ip in "${_ICE_HOST_IP_ARRAY[@]}"; do + if [[ -n "${ice_host_ip}" ]]; then + ICE_HOST_ARGS+=(--ice-host-ip "${ice_host_ip}") + fi + done +fi + +if command -v curl >/dev/null 2>&1; then + echo "Checking TeleFuser stream health: ${HEALTH_URL}" + curl --fail --silent --show-error "${HEALTH_URL}" >/dev/null +fi + +echo "Running stream benchmark with config: ${CONFIG_PATH}" +"${PYTHON_BIN}" benchmarks/telefuser_aiperf/scripts/run_stream_bench.py \ + --config "${CONFIG_PATH}" \ + --server-url "${SERVER_URL}" \ + "${ICE_HOST_ARGS[@]}" \ + "$@" diff --git a/benchmarks/telefuser_aiperf/scripts/run_video_bench.sh b/benchmarks/telefuser_aiperf/scripts/run_video_bench.sh new file mode 100755 index 0000000..02163f2 --- /dev/null +++ b/benchmarks/telefuser_aiperf/scripts/run_video_bench.sh @@ -0,0 +1,28 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)" +cd "${ROOT_DIR}" + +CONFIG_PATH="${1:-benchmarks/telefuser_aiperf/configs/video_generation_quick.yaml}" +SERVER_URL="${TELEFUSER_AIPERF_URL:-http://127.0.0.1:8000}" +HEALTH_URL="${TELEFUSER_AIPERF_HEALTH_URL:-${SERVER_URL}/v1/service/health}" +AIPERF_BIN="${AIPERF_BIN:-aiperf}" +NOFILE_LIMIT="${TELEFUSER_BENCH_NOFILE_LIMIT:-8192}" + +if ! ulimit -n "${NOFILE_LIMIT}" >/dev/null 2>&1; then + echo "Warning: failed to raise open-file limit to ${NOFILE_LIMIT}" >&2 +fi + +if ! command -v "${AIPERF_BIN}" >/dev/null 2>&1; then + echo "aiperf is not installed. Set AIPERF_BIN or run: pip install -e ./benchmarks/aiperf" >&2 + exit 1 +fi + +if command -v curl >/dev/null 2>&1; then + echo "Checking TeleFuser health: ${HEALTH_URL}" + curl --fail --silent --show-error "${HEALTH_URL}" >/dev/null +fi + +echo "Running AIPerf with config: ${CONFIG_PATH}" +"${AIPERF_BIN}" profile --config "${CONFIG_PATH}" diff --git a/docs/en/benchmark_aiperf.md b/docs/en/benchmark_aiperf.md new file mode 100644 index 0000000..fdebe3b --- /dev/null +++ b/docs/en/benchmark_aiperf.md @@ -0,0 +1,105 @@ +# TeleFuser and AIPerf Benchmark + +This page describes how to benchmark TeleFuser batch video serving with AIPerf. + +Current scope: + +- `telefuser serve` +- OpenAI-compatible `/v1/videos` API +- end-to-end request latency and throughput +- optional HTTP trace and TeleFuser metrics scraping + +Not covered yet: + +- `telefuser stream-serve` +- WebRTC / DataChannel world-model sessions +- frame-level and control-latency metrics + +For remote stream setup, use: + +```bash +python3 scripts/remote_bench_sync.py stream-bootstrap +``` + +That helper syncs the stream assets, installs WebRTC dependencies, auto-detects a usable `model_zoo` root, and writes a non-destructive `model_zoo` symlink inside the remote repo for later `telefuser stream-serve` startup. + +Both `run_stream_bench.sh` and `run_video_bench.sh` raise the open-file limit to `8192` by default to avoid AIPerf/WebRTC startup failures; override with `TELEFUSER_BENCH_NOFILE_LIMIT` if your environment needs a different value. + +For stream runs, `run_stream_bench.py` also auto-detects non-loopback local IPs for ICE candidate gathering when no explicit allowlist is provided. Use `--ice-host-ip` or `TELEFUSER_WEBRTC_ICE_HOST_IPS` only when you need to pin a specific address. + +## Assets + +The benchmark assets live in: + +```text +benchmarks/telefuser_aiperf/ +├── README.md +├── configs/ +├── data/ +└── scripts/ +``` + +## Prerequisites + +1. Start a TeleFuser batch video server. +2. Install AIPerf from the vendored clone. + +Example: + +```bash +telefuser serve \ + examples/wan_video/wan21_14b_image_to_video_h100.py \ + --port 8000 \ + --task i2v + +pip install -e ./benchmarks/aiperf +``` + +## Quick Start + +Run the default benchmark: + +```bash +bash benchmarks/telefuser_aiperf/scripts/run_video_bench.sh +``` + +The script checks: + +```bash +curl http://127.0.0.1:8000/v1/service/health +``` + +and then runs AIPerf with: + +```bash +aiperf profile --config benchmarks/telefuser_aiperf/configs/video_generation_quick.yaml +``` + +## Configs + +- `video_generation_quick.yaml`: minimal connectivity and latency check +- `video_generation_e2e.yaml`: fuller E2E run with warmup, download, trace, and optional server metrics +- `video_generation_rate.yaml`: Poisson arrival benchmark for rate-based load + +## Why Built-In Readiness Probe Is Disabled + +These configs intentionally do not use AIPerf's built-in `waitForModel*` readiness probe. + +Reason: + +- TeleFuser is being benchmarked through `video_generation` +- AIPerf readiness probing is better aligned with chat/completions/embeddings-style endpoints +- TeleFuser does not currently expose a stable `/v1/models` path for this use case + +So the helper script uses TeleFuser's `/v1/service/health` instead. + +## Limitation + +This is a **batch video serving benchmark**, not yet a **real-time world-model streaming benchmark**. + +For `stream-serve`, the next layer should add: + +1. a WebRTC transport +2. a world-model session endpoint +3. an interactive trace workload +4. frame-level latency metrics diff --git a/docs/en/index.md b/docs/en/index.md index 3278364..d13aaad 100644 --- a/docs/en/index.md +++ b/docs/en/index.md @@ -56,6 +56,7 @@ telefuser stream-serve examples/stream_server/stream_lingbot_world_fast.py -p 80 - **[Service Guide](service.md)** — Batch serving, task APIs, and SDK - **[Stream Server](stream_server.md)** — WebRTC streaming and bidirectional control +- **[TeleFuser and AIPerf Benchmark](benchmark_aiperf.md)** — Batch video serving benchmark with AIPerf - **[Configuration](configuration.md)** — Runtime and model configuration - **[Parallel Inference](parallel.md)** — Distributed processing strategies - **[Adding New Model](adding_new_model.md)** — Integrate new models diff --git a/docs/zh/benchmark_aiperf.md b/docs/zh/benchmark_aiperf.md new file mode 100644 index 0000000..a6f218c --- /dev/null +++ b/docs/zh/benchmark_aiperf.md @@ -0,0 +1,605 @@ +# TeleFuser 与 AIPerf Benchmark + +本文档是 TeleFuser 的 benchmark 执行入口，主要维护： + +- 批处理视频服务的压测运行方式 +- AIPerf 配置、参数和结果文件 +- 远程同步、部署和复现实验脚本 + +更细的 benchmark 设计、分层、对标 shortlist 和 `stream-serve` 规划放在： + +- [TeleFuser 与 AIPerf Benchmark 设计](benchmark_aiperf_design.md) + +## 目录结构 + +仓库中新增的 benchmark 资产位于： + +```text +benchmarks/telefuser_aiperf/ +├── README.md +├── configs/ +│ ├── video_generation_quick.yaml +│ ├── video_generation_e2e.yaml +│ ├── video_generation_rate.yaml +│ └── stream_lingbot_world_fast_quick.json +├── data/ +│ ├── video_prompts.jsonl +│ └── stream_lingbot_controls.json +└── scripts/ + ├── run_video_bench.sh + └── run_stream_bench.sh +``` + +## 前置条件 + +### 1. 启动 TeleFuser 批处理服务 + +例如启动一个视频生成服务： + +```bash +telefuser serve \ + examples/wan_video/wan21_14b_image_to_video_h100.py \ + --port 8000 \ + --task i2v +``` + +确认服务健康检查可访问： + +```bash +curl http://127.0.0.1:8000/v1/service/health +``` + +### 2. 安装 AIPerf + +当前仓库已经 vendored 一份 AIPerf clone 到 `benchmarks/aiperf`，可直接本地安装： + +```bash +pip install -e ./benchmarks/aiperf +``` + +安装后确认 `aiperf` 命令可用： + +```bash +aiperf --help +``` + +## 快速开始 + +### 批处理视频 + +运行一个最小视频 benchmark： + +```bash +bash benchmarks/telefuser_aiperf/scripts/run_video_bench.sh +``` + +默认使用： + +- 配置文件：`benchmarks/telefuser_aiperf/configs/video_generation_quick.yaml` +- 服务地址：`http://127.0.0.1:8000` +- prompt 文件：`benchmarks/telefuser_aiperf/data/video_prompts.jsonl` + +脚本会先检查： + +```bash +curl http://127.0.0.1:8000/v1/service/health +``` + +然后执行： + +```bash +aiperf profile --config benchmarks/telefuser_aiperf/configs/video_generation_quick.yaml +``` + +### 流式世界模型 + +先启动 TeleFuser 流式服务： + +```bash +telefuser stream-serve \ + examples/lingbot/stream_lingbot_world_fast.py \ + -p 8088 \ + --skip-validation +``` + +确认流式服务健康检查可访问： + +```bash +curl http://127.0.0.1:8088/v1/service/health +``` + +然后运行最小流式 benchmark： + +```bash +bash benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh +``` + +默认使用： + +- 配置文件：`benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_quick.json` +- 控制 trace：`benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json` +- 服务地址：`http://127.0.0.1:8088` + +如果远端机器有很多网卡或容器 veth，`run_stream_bench.py` 现在会默认自动探测本机的非 loopback 地址并用于 ICE host candidate 收敛。 + +只有在你明确知道要锁定某个地址时，才手动传 `--ice-host-ip` 或设置 `TELEFUSER_WEBRTC_ICE_HOST_IPS`。不要默认写 `127.0.0.1`，否则在远程主机 / 容器网络里很容易把 WebRTC 连接锁死在本机回环地址。 + +这条路径当前不是 AIPerf 原生 endpoint，而是 TeleFuser 侧独立的 WebRTC benchmark harness。 + +如果在远程环境执行，建议直接用： + +```bash +python3 scripts/remote_bench_sync.py stream-bootstrap +``` + +它会同步代码、安装 WebRTC 依赖、自动探测可用的 `model_zoo` 根目录，并在远程仓库里写回一个非破坏性的 `model_zoo` 软链，方便后续直接起 `telefuser stream-serve`。 + +另外，`run_stream_bench.sh` 和 `run_video_bench.sh` 会默认把 open-file limit 提升到 `8192`，避免 AIPerf/WebRTC 在启动时撞到文件描述符上限；如果你的环境需要不同值，可以用 `TELEFUSER_BENCH_NOFILE_LIMIT` 覆盖。 + +## 可用配置 + +### `video_generation_quick.yaml` + +适合快速联通性验证。 + +特点： + +- 5 个请求 +- 并发 1 +- 不下载视频内容 +- 只看最基本的端到端结果 + +使用： + +```bash +bash benchmarks/telefuser_aiperf/scripts/run_video_bench.sh \ + benchmarks/telefuser_aiperf/configs/video_generation_quick.yaml +``` + +### `video_generation_e2e.yaml` + +适合更完整的 E2E benchmark。 + +特点： + +- 1 个 warmup 请求 +- 默认 profiling 并发 2 +- 默认 profiling 请求数 6 +- 开启 `downloadVideoContent: true` +- 开启 HTTP trace 导出 +- 可抓取 TeleFuser `/v1/service/metrics` + +使用： + +```bash +bash benchmarks/telefuser_aiperf/scripts/run_video_bench.sh \ + benchmarks/telefuser_aiperf/configs/video_generation_e2e.yaml +``` + +### `video_generation_rate.yaml` + +适合看低频到中频请求率下的表现。 + +特点： + +- `poisson` 到达模式 +- 默认速率 `0.2 req/s` +- 默认最大并发 4 +- 适合观察排队与时延波动 + +使用： + +```bash +bash benchmarks/telefuser_aiperf/scripts/run_video_bench.sh \ + benchmarks/telefuser_aiperf/configs/video_generation_rate.yaml +``` + +### `stream_lingbot_world_fast_quick.json` + +适合流式世界模型最小联通和端到端时延验证。 + +特点： + +- 1 个 session +- `WebRTC + DataChannel` +- 默认运行 `12s` +- 按固定 control trace 注入方向控制 +- 输出 `offer RTT`、`首帧时延`、`稳态 FPS`、`control ack latency`、`control-to-next-frame latency` + +使用： + +```bash +bash benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh \ + benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_quick.json +``` + +## 常用环境变量 + +这些配置支持通过环境变量覆盖： + +```bash +export TELEFUSER_AIPERF_URL=http://127.0.0.1:8000 +export TELEFUSER_AIPERF_METRICS_URL=http://127.0.0.1:8000/v1/service/metrics +export TELEFUSER_AIPERF_CONCURRENCY=4 +export TELEFUSER_AIPERF_REQUESTS=12 +export TELEFUSER_AIPERF_SIZE=1280x720 +export TELEFUSER_AIPERF_SECONDS=4 +export TELEFUSER_AIPERF_SERVER_METRICS=true +``` + +例如跑一个更高并发的完整 E2E： + +```bash +export TELEFUSER_AIPERF_CONCURRENCY=4 +export TELEFUSER_AIPERF_REQUESTS=12 + +bash benchmarks/telefuser_aiperf/scripts/run_video_bench.sh \ + benchmarks/telefuser_aiperf/configs/video_generation_e2e.yaml +``` + +流式 benchmark 也支持通过 CLI 覆盖，例如： + +```bash +bash benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh \ + benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_quick.json \ + --session-count 1 \ + --session-duration-s 20 \ + --print-events +``` + +## 远程一键同步与校验 + +为了方便把这套 benchmark 资产同步到远程测试机，仓库中新增了： + +```bash +python3 scripts/remote_bench_sync.py --help +``` + +这个脚本默认面向当前测试机： + +- `root@116.238.240.2` +- SSH 端口 `30724` +- 远程仓库路径 `/workspace/TeleFuser` +- TeleFuser 服务 Python `/root/venv/cu128/bin/python` +- 独立 benchmark Python `/root/venv/telefuser-bench/bin/python` + +推荐模式是**分离部署**： + +- 已经部署好的 TeleFuser 服务环境继续保留 +- 额外创建一个独立的 benchmark venv，只安装 AIPerf 和压测依赖 + +这样可以避免 benchmark 依赖覆盖 TeleFuser 现有运行环境里的： + +- `transformers` +- `protobuf` +- `pillow` +- `huggingface_hub` + +如果你要跑 `telefuser stream-serve` 的世界模型流式 benchmark，推荐直接走一键自动化预备： + +```bash +python3 scripts/remote_bench_sync.py stream-bootstrap +``` + +这条命令会自动完成： + +- 同步流式 benchmark 资产 +- 安装 AIPerf 和 `aiortc` / `opencv-python-headless` +- 安装 TeleFuser 到独立服务环境 +- 校验 `examples/lingbot/stream_lingbot_world_fast.py` +- 校验 `model_zoo` 路径 +- 打印远程 GPU 状态 + +如果你只想更新安装而不重新同步： + +```bash +python3 scripts/remote_bench_sync.py install --install-webrtc --install-telefuser +``` + +如果目标机器变化，可以通过参数覆盖： + +```bash +python3 scripts/remote_bench_sync.py \ + --host \ + --port \ + --user \ + --remote-repo \ + --remote-python \ + --remote-bench-python \ + gpu-status +``` + +### 1. 查看远程 GPU 状态 + +```bash +python3 scripts/remote_bench_sync.py gpu-status +``` + +这个命令会通过 SSH 执行 `nvidia-smi`，输出： + +- 每张卡的利用率 +- 显存占用 +- 当前活跃计算进程 + +### 2. 首次全量同步 + +```bash +python3 scripts/remote_bench_sync.py bootstrap +``` + +这个命令会把当前管理的 benchmark 资产全量同步到远程，包括： + +- `benchmarks/aiperf` +- `benchmarks/telefuser_aiperf` +- `docs/en/benchmark_aiperf.md` +- `docs/zh/benchmark_aiperf.md` +- `docs/en/index.md` +- `docs/zh/index.md` +- `mkdocs.yml` + +如果你还希望同步完成后直接安装并校验： + +```bash +python3 scripts/remote_bench_sync.py bootstrap --install --verify +``` + +默认行为是： + +- 创建独立 benchmark venv +- 在独立环境中安装 vendored `benchmarks/aiperf` +- 复用同一个 benchmark venv 运行流式 benchmark Python 脚本 +- 不修改当前 TeleFuser 服务环境 + +如果你明确希望同时刷新 TeleFuser 服务环境： + +```bash +python3 scripts/remote_bench_sync.py bootstrap --install --verify --install-telefuser +``` + +如果你想强制重建独立 benchmark venv： + +```bash +python3 scripts/remote_bench_sync.py bootstrap --install --verify --recreate-bench-venv +``` + +如果当前只想演练命令而不真正执行： + +```bash +python3 scripts/remote_bench_sync.py --dry-run bootstrap --install --verify +``` + +### 3. 后续增量同步 + +首次同步后，脚本会在本地写一个状态文件，用于记录上次已同步的文件哈希： + +```text +~/.cache/telefuser/remote_bench_sync/.json +``` + +后续直接运行： + +```bash +python3 scripts/remote_bench_sync.py sync +``` + +脚本只会上传发生变化的受管文件。 + +如果你想看本次计划同步哪些文件： + +```bash +python3 scripts/remote_bench_sync.py sync --print-files +``` + +如果你想忽略状态文件并强制重新全量上传： + +```bash +python3 scripts/remote_bench_sync.py sync --full +``` + +### 4. 远程安装 + +```bash +python3 scripts/remote_bench_sync.py install +``` + +默认会在远程执行： + +- 创建独立 venv：`/root/venv/telefuser-bench` +- `pip install -e /workspace/TeleFuser/benchmarks/aiperf` + +也就是说，默认**不会**修改现有 TeleFuser 服务环境。 + +如果你确实要刷新 TeleFuser 服务环境： + +```bash +python3 scripts/remote_bench_sync.py install --install-telefuser +``` + +如果你想强制重建独立 benchmark venv： + +```bash +python3 scripts/remote_bench_sync.py install --recreate-bench-venv +``` + +如果你只想更新源码，不想拉依赖，可以使用： + +```bash +python3 scripts/remote_bench_sync.py install --no-deps +``` + +如果你当前不想安装 AIPerf，只想做其它步骤： + +```bash +python3 scripts/remote_bench_sync.py install --skip-aiperf +``` + +### 5. 远程校验 + +```bash +python3 scripts/remote_bench_sync.py verify +``` + +校验内容包括： + +- benchmark 配置文件是否存在 +- TeleFuser 服务环境中的 `import telefuser` +- TeleFuser 服务环境中的 `telefuser --help` +- 独立 benchmark 环境中的 `import aiperf` +- 独立 benchmark 环境中的 `aiperf --help` +- 独立 benchmark 环境中的 `python benchmarks/telefuser_aiperf/scripts/run_stream_bench.py --help` + +注意： + +- 如果你之前使用的是 `install --no-deps`，那么对应环境的 CLI 校验可能失败 +- 如果你只想检查 benchmark 环境，可以使用： + +```bash +python3 scripts/remote_bench_sync.py verify --skip-telefuser +``` + +### 6. 运行 benchmark + +#### 6.1 一键跑 `Wan2.1-I2V-14B-480P` 对比 + +如果远程机器上的 benchmark 资产和隔离环境已经准备好，可以直接用一条命令顺序跑： + +- `TeleFuser` +- `Diffusers` baseline + +```bash +python3 scripts/remote_bench_sync.py batch-compare \ + --framework both \ + --gpu 0 +``` + +这条命令会自动完成： + +- 校验远程 `telefuser` / `aiperf` 环境 +- 解析 TeleFuser batch 模型目录 +- 对指定 GPU 执行 `gpu_burner.sh stop ` +- 检查该 GPU 是否真的空闲 +- 拉起 TeleFuser 480P 服务 +- 跑对齐后的 TeleFuser AIPerf config +- 停掉 TeleFuser 服务 +- 拉起 Diffusers baseline 服务 +- 跑对齐后的 Diffusers AIPerf config +- 输出两边最新 `summary.json` 路径和关键指标 + +默认行为是**非破坏性的**： + +- 只会停止指定 GPU 上的 burner 占位 +- 不会主动杀掉远程机器上已有的其它计算任务 +- 如果 GPU 在停止 burner 之后仍然被真实任务占用，会直接失败 + +如果你希望它等待 GPU 空闲，而不是立即失败，可以显式给等待窗口： + +```bash +python3 scripts/remote_bench_sync.py batch-compare \ + --framework both \ + --gpu 0 \ + --gpu-idle-timeout 600 +``` + +如果你只想单独跑一边： + +```bash +python3 scripts/remote_bench_sync.py batch-compare --framework telefuser --gpu 0 +python3 scripts/remote_bench_sync.py batch-compare --framework diffusers --gpu 0 +``` + +这条自动化入口当前固定使用： + +- TeleFuser config: + `benchmarks/telefuser_aiperf/configs/video_generation_wan21_i2v_480p_compare.yaml` +- Diffusers config: + `benchmarks/baseline/diffusers_wan_i2v/configs/video_generation_compare.yaml` + +如果远程环境还没准备好，先执行： + +```bash +python3 scripts/remote_bench_sync.py sync +python3 scripts/remote_bench_sync.py install --install-telefuser +python3 scripts/remote_bench_sync.py verify +``` + +#### 6.2 手工跑 TeleFuser batch benchmark + +当 TeleFuser 服务已经在远程机器上启动后，可以直接指定独立 benchmark 环境中的 `aiperf` 可执行文件： + +```bash +export AIPERF_BIN=/root/venv/telefuser-bench/bin/aiperf + +bash benchmarks/telefuser_aiperf/scripts/run_video_bench.sh \ + benchmarks/telefuser_aiperf/configs/video_generation_e2e.yaml +``` + +这样 benchmark 走独立环境，TeleFuser 服务继续走原来的运行环境，两边互不覆盖。 + +#### 6.3 手工跑流式 benchmark + +流式 benchmark 则直接走独立环境里的 Python： + +```bash +export TELEFUSER_STREAM_BENCH_PYTHON=/root/venv/telefuser-bench/bin/python + +bash benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh \ + benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_quick.json +``` + +## 为什么没有启用 AIPerf 自带 readiness probe + +当前这套配置没有使用 AIPerf 的 `waitForModelTimeout` / `waitForModelMode`。 + +原因是： + +- TeleFuser 这里对接的是 `video_generation` endpoint +- AIPerf 的 readiness probe 更偏向 `chat` / `completions` / `embeddings` +- TeleFuser 当前没有稳定的 `/v1/models` 路径可用于这类 probe + +因此这里改为更直接的做法：在 benchmark 脚本里先检查 TeleFuser 的 `/v1/service/health`。 + +## 指标解释 + +这套 benchmark 里最值得关注的通常是： + +- `Request Latency` + 表示一次完整视频请求从提交到完成的端到端耗时。 + +- `Request Throughput` + 表示单位时间内完成的请求数。 + +- HTTP Trace 相关指标 + 用于区分时间主要耗在： + - 连接建立 + - 请求发送 + - 服务端等待 + - 响应接收 + +- TeleFuser 服务指标 + 如果启用了 `serverMetrics`，可以同时对照： + - 队列长度 + - 已创建 / 已完成 / 已失败任务数 + - GPU 指标 + +## 结果文件 + +结果默认写到： + +```text +artifacts/telefuser_aiperf/ +``` + +不同配置写入不同子目录，例如： + +- `artifacts/telefuser_aiperf/video_quick` +- `artifacts/telefuser_aiperf/video_e2e` +- `artifacts/telefuser_aiperf/video_rate` +- `artifacts/telefuser_aiperf/stream_lingbot_quick/` + +通常可以关注： + +- summary JSON +- raw records JSONL +- server metrics JSON / CSV +- stream benchmark 的逐 session JSONL +- stream benchmark 的逐事件 JSONL diff --git a/docs/zh/benchmark_aiperf_design.md b/docs/zh/benchmark_aiperf_design.md new file mode 100644 index 0000000..ec5dc8a --- /dev/null +++ b/docs/zh/benchmark_aiperf_design.md @@ -0,0 +1,684 @@ +# TeleFuser 与 AIPerf Benchmark 设计 + +本文档记录 benchmark 的设计决策、对标边界和后续扩展方向。 + +具体跑法、脚本、参数和结果文件说明请看： + +- [TeleFuser 与 AIPerf Benchmark](benchmark_aiperf.md) + +## 1. Benchmark 分层 + +### 1.1 批处理视频服务 benchmark + +这类场景指的是： + +- 用户提交一次文本或图像条件 +- 服务端异步生成一个完整视频 +- 客户端轮询任务状态 +- 最终拿到视频文件 + +这一类当前最适合： + +- `telefuser serve` +- OpenAI 兼容 `/v1/videos` +- `AIPerf` 端到端 HTTP benchmark + +### 1.2 世界模型流式系统 benchmark + +这类场景指的是： + +- 长生命周期 session +- 持续输入控制信号 +- 持续输出视频帧或媒体流 +- 更关注控制反馈闭环，而不是单次完整视频生成 + +这一类更接近： + +- `telefuser stream-serve` +- WebRTC + DataChannel +- 世界模型在线交互推理 + +这两类负载的评价指标不同，不能混在同一套 benchmark 结论里。 + +### 1.3 世界模型实时流式系统框架 shortlist + +如果 TeleFuser 后面要做 `stream-serve`，更合适的对标对象应该是系统框架 / serving runtime，而不是模型本体。 + +当前更接近的开源系统层候选是： + +- `SGLang-Omni` +- `vLLM-Omni` +- `vLLM` +- `SGLang` +- `Ray Serve` +- `Triton Inference Server` + +如果需要实时媒体传输层，还可以单独观察： + +- `LiveKit` +- `Janus` +- `mediasoup` +- `aiortc` +- `Pion` + +这些是 transport 组件，不是推理框架本身。 + +需要明确区分的是： + +- `Matrix-Game`、`MineWorld`、`LongLive`、`CausVid`、`Helios` 更偏“模型 / 研究系统 / reference system” +- 它们适合用来定义负载和行为模式 +- 但不应混进“系统框架 shortlist” + +真正的世界模型流式端到端 benchmark 仍需要我们自己定义请求协议、会话状态和指标。 + +### 1.4 `stream-serve` 第一批系统框架对比顺序 + +如果后面要真的落地 `stream-serve` 的端到端 benchmark，第一批建议先比： + +1. `TeleFuser stream-serve` +2. `SGLang-Omni` +3. `vLLM-Omni` + +优先级原因很简单： + +- `SGLang-Omni` 更贴近我们现在想做的“长会话、多阶段、低延迟在线生成”系统形态 +- `vLLM-Omni` 可以作为更通用的第二基线 +- 两者都比通用 `Ray Serve` / `FastAPI` 更接近推理系统本体 + +如果需要把 transport 也纳入对比，再单独固定 WebRTC / SSE / WebSocket 实现，不要和推理框架混算。 + +### 1.5 批处理视频模型对标顺序 + +如果目标是评估 **TeleFuser 与其它视频推理框架** 的差异，那么第一轮 benchmark 必须满足下面这些约束： + +1. **模型相同** + - 例如都跑 `Wan2.1-I2V-14B-480P` + - 或都跑 `HunyuanVideo-1.5` + - 或都跑 `LTX-2` + +2. **任务相同** + - `t2v` 只和 `t2v` 比 + - `i2v` 只和 `i2v` 比 + - 不把 `t2v`、`i2v`、`ti2v` 混在一起 + +3. **输入相同** + - 同 prompt + - 同输入图 + - 同输出分辨率 + - 同帧数或等价生成长度 + - 同 diffusion steps + +4. **硬件相同** + - 同 GPU 型号 + - 同卡数 + - 同显存约束 + - 尽量控制一致的软件栈 + +5. **只变推理框架** + - 可以比较 `TeleFuser`、`Diffusers`、`xDiT`、`LightX2V`、`FastVideo`、`SGLang Diffusion` + - 但不能把“更换模型蒸馏版权重”误当成“只变框架” + +只要不满足这些条件，就不应被视为第一轮主 benchmark。 + +推荐优先级： + +1. `Wan2.1-I2V-14B-480P` +2. `HunyuanVideo-1.5` +3. `LTX-2` + +### 1.6 当前已落地的第一版可执行基线 + +仓库里已经补齐了第一版“同模型、只变框架”的可执行闭环，当前不是只停留在规划层面了： + +- TeleFuser 侧固定 480P 服务示例：[examples/wan_video/wan21_14b_image_to_video_480p_service.py](../../examples/wan_video/wan21_14b_image_to_video_480p_service.py) +- Diffusers 侧独立基线服务：[benchmarks/baseline/diffusers_wan_i2v/service.py](../../benchmarks/baseline/diffusers_wan_i2v/service.py) + +这版首个可直接开跑的对照组合固定为： + +- 模型：`Wan2.1-I2V-14B-480P` +- 分辨率：`832x480` +- 帧数：`81` +- 推理步数：`40` +- `guidance_scale=5.0` +- `fps=16` + +这里刻意把 TeleFuser 侧的 480P 服务从通用面积启发式改成了固定 `832x480`，目的是避免与官方 Diffusers 基线出现宽高不一致，保证第一轮对照可比。 + +### 1.7 AIPerf 对批处理视频负载的支持情况 + +AIPerf 对 `telefuser serve` 这类批处理视频服务已经有明确支持，当前覆盖的核心能力是： + +- OpenAI 兼容 `/v1/videos` 的异步提交 / 轮询 / 下载流程 +- `video_generation` endpoint +- `--download-video-content` 把下载时间纳入端到端时延 +- `multipart/form-data` 请求编码 +- `--extra-inputs` 传递视频生成参数 +- `single_turn` / `multi_turn` 这类输入组织方式 + +对 TeleFuser 这条线来说，AIPerf 已经能支撑两类最重要的批处理视频负载： + +1. **T2V** + - 直接用 `video_generation` endpoint + - 用文本 prompt 驱动 `/v1/videos` + +2. **I2V** + - 同样走 `video_generation` endpoint + - 通过 `reference_url` 或 `input_reference` 传入参考图 + - 我们当前的 `telefuser_aiperf` prompt 文件已经在用 `reference_url` 这条路 + +也就是说： + +- AIPerf 不是只能测纯文本视频生成 +- 它已经可以覆盖我们当前 TeleFuser 批处理视频 benchmark 里的 `T2V / I2V` 主线 +- 但它的上游文档和教程仍然更偏通用视频生成，不是专门为 TeleFuser 的 `I2V` 场景定制 + +当前对我们最有用的落点是： + +- `telefuser serve` + `AIPerf` 已足够做 batch video E2E benchmark +- `Wan2.1-I2V-14B-480P` 这条线已经可以直接跑 +- 后续若要扩展到更复杂的 `I2V` 变体，只需要继续补 prompt/schema 适配，而不是重做 benchmark 框架 + +### 1.8 Baseline 接口协议标准 + +随着后面要逐步接入更多对照实现，例如： + +- `Diffusers` +- `xDiT` +- `FastVideo` +- `LightX2V` +- `SGLang Diffusion` +- `vLLM-Omni` + +仓库里需要一套统一的 **baseline benchmark contract**，否则每接一个新 baseline，就会重复发明一套： + +- 目录布局 +- 启动脚本 +- 健康检查方式 +- HTTP / WebRTC 适配层 +- 指标采集字段 +- 文档说明格式 + +这里要标准化的不是 baseline 的内部实现，而是 **benchmark 视角下的最小可测协议**。 + +#### 1.8.1 标准化目标 + +第一版建议定义成 **versioned contract**： + +- `contract_version: v1` +- `mode: batch_video | stream_world` + +第一版只覆盖两类主负载： + +1. `batch_video` + - 对应 `telefuser serve` + - 对应 AIPerf 当前已经能覆盖的 `/v1/videos` 异步完整视频生成 + +2. `stream_world` + - 对应 `telefuser stream-serve` + - 对应 WebRTC 长会话、持续控制、持续输出的视频流 benchmark + +后续如果再扩展： + +- `audio_video_realtime` +- `multi_camera_world` +- `tool_augmented_agent_stream` + +应在 `v1` 之外继续演进，而不是一开始把所有可能性都揉进同一套协议。 + +#### 1.8.2 目录协议 + +第三方或对照实现统一放在： + +- `benchmarks/baseline//` + +推荐最小目录结构： + +- `README.md` +- `service.py` 或 `launcher.py` +- `configs/` +- `scripts/` +- `benchmark_contract.yaml` + +其中： + +- `telefuser_aiperf/` 不是 baseline，它是 TeleFuser 自己的 benchmark harness +- `baseline/` 下面放“只变推理框架”的对照实现 +- `benchmark_contract.yaml` 用来描述 baseline 能力，而不是替代 README +- 当前 TeleFuser benchmark harness 和 Diffusers baseline 已经分别提供了第一版薄 contract： + - [benchmarks/telefuser_aiperf/benchmark_contract.yaml](../../benchmarks/telefuser_aiperf/benchmark_contract.yaml) + - [benchmarks/baseline/diffusers_wan_i2v/benchmark_contract.yaml](../../benchmarks/baseline/diffusers_wan_i2v/benchmark_contract.yaml) + +#### 1.8.3 能力声明协议 + +每个 baseline 建议暴露一个能力声明文件，例如 `benchmark_contract.yaml`，最少包含： + +- `contract_version` +- `baseline_name` +- `mode` +- `model_family` +- `supported_tasks` +- `transport` +- `request_encoding` +- `result_delivery` +- `metrics` + +建议第一版字段示例： + +```yaml +contract_version: v1 +baseline_name: diffusers_wan_i2v +mode: batch_video +model_family: wan21_i2v_14b_480p +supported_tasks: + - i2v +transport: http +request_encoding: + - multipart_form + - json +result_delivery: + - poll_status + - download_content +capabilities: + supports_reference_url: true + supports_input_reference_upload: true + supports_cancel: true +metrics: + - request_latency + - completion_latency + - download_latency + - peak_memory_mb +``` + +这一层的价值是： + +- benchmark 脚本能先检查 baseline 支持什么 +- 文档和自动化部署脚本可以直接消费这份声明 +- 后续接新 baseline 时，先补 contract，再补适配脚本 + +#### 1.8.4 `batch_video` 最小接口协议 + +对 `batch_video` 类型 baseline，建议统一最小 HTTP 面： + +- `GET /v1/service/health` +- `POST /v1/videos` +- `GET /v1/videos/{id}` +- `GET /v1/videos/{id}/content` +- `DELETE /v1/videos/{id}` 可选 + +最小请求语义建议统一为： + +- `prompt` +- `model` +- `size` +- `seconds` +- `seed` +- `negative_prompt` +- `reference_url` 或 `input_reference` + +允许 baseline 自己额外扩展字段，但 benchmark 第一轮只依赖最小公共子集。 + +`GET /v1/videos/{id}` 的最小返回字段建议统一为： + +- `id` +- `status` +- `progress` +- `created_at` +- `completed_at` +- `error` +- `url` 或 `file_path` + +其中状态集合建议至少覆盖： + +- `queued` +- `generating` +- `completed` +- `failed` +- `cancelled` + +这样 AIPerf 和后续自定义 harness 都可以复用同一套 submit / poll / download 流程。 + +#### 1.8.5 `stream_world` 最小接口协议 + +对 `stream_world` 类型 baseline，建议统一最小会话面： + +- `GET /v1/service/health` +- `POST /v1/stream/webrtc/offer` +- `DELETE /v1/stream/webrtc/{session_id}` + +如果使用 DataChannel，建议至少约定这些消息类别： + +- `control` +- `status` +- `chunk` +- `done` +- `error` + +其中： + +- `control` 由客户端发送，代表方向控制、动作控制或结构化控制输入 +- `status` 由服务端发送，代表会话阶段和状态机推进 +- `chunk` 由服务端发送，代表一批可消费的输出 +- `done` 代表本轮或整个 session 正常结束 +- `error` 代表 session 失败或不可恢复异常 + +不要求所有 baseline 都与 TeleFuser 的 DataChannel payload 完全一致，但必须有一层 adapter 能映射到这套 benchmark 语义。 + +#### 1.8.6 指标协议 + +接口统一之后，还需要统一结果侧最小指标集合。 + +`batch_video` 至少建议采： + +- `request_latency` +- `queue_latency` +- `completion_latency` +- `download_latency` +- `success_rate` +- `throughput` + +`stream_world` 至少建议采： + +- `offer_rtt` +- `connected_latency` +- `first_frame_latency` +- `first_metadata_latency` +- `steady_state_fps` +- `control_ack_latency` +- `control_to_next_frame_latency` +- `session_success_rate` + +如果 baseline 能提供服务端附加指标，也建议通过统一字段上报，例如： + +- `inference_time_s` +- `peak_memory_mb` +- `server_metrics_url` + +#### 1.8.7 非目标 + +这套标准当前 **不** 试图做下面这些事情： + +- 不统一 baseline 的内部代码结构 +- 不强制所有 baseline 使用同一种 Web 框架 +- 不强制所有 baseline 使用同一种任务队列实现 +- 不把 transport 协议细节过度收紧到无法接第三方系统 + +也就是说： + +- 我们标准化的是“怎么测” +- 不是“怎么实现” + +#### 1.8.8 第一批落地顺序 + +这套 baseline contract 第一批建议先落到： + +1. `benchmarks/baseline/diffusers_wan_i2v/` +2. 后续第二个 batch baseline +3. 再考虑 `stream_world` 类型 baseline + +原因是： + +- `diffusers_wan_i2v` 已经是现成的第一版对照实现 +- 最容易先把目录协议、能力声明和 batch_video 最小接口约束跑通 +- `stream_world` 的协议复杂度更高，适合在 batch contract 稳定后再继续抽象 + +这部分设计一旦落地，后续新增 baseline 时应优先遵守 contract，再补实现，而不是先写一套散落脚本再回头整理。 + +#### 1.8.9 Batch Compare 自动化约束 + +对于第一版 `Wan2.1-I2V-14B-480P` 对照，不应该继续靠手工 SSH 一条条拼： + +- 停 burner +- 起 TeleFuser +- 跑 AIPerf +- 停服务 +- 起 baseline +- 再跑一次 + +这条链路现在应该固化成一个总入口： + +```bash +python3 scripts/remote_bench_sync.py batch-compare +``` + +这条自动化入口的职责边界应该明确成下面这样： + +1. 先校验远程环境 + - `telefuser` CLI 可用 + - `aiperf` CLI 可用 + - compare config 存在 + +2. 只管理“我们自己的 benchmark 资源” + - 可以对指定 GPU 执行 burner `stop` + - 可以拉起和停止本次 benchmark 启动的服务进程 + - 不应默认杀掉远程机器上其它已在运行的真实计算任务 + +3. GPU 资源策略默认非破坏性 + - `stop burner` 之后检查目标 GPU 是否真的空闲 + - 如果还有其它 compute process，占用信息直接报出来 + - 默认立即失败，或按显式参数进入等待模式 + +4. 对比执行顺序固定 + - 同一张 GPU + - 先 TeleFuser + - 再 Diffusers baseline + - 使用同一组对齐后的 compare config + +5. 结果输出标准化 + - 返回各自最新 `summary.json` 路径 + - 汇总 `request_latency` + - 汇总 `request_throughput` + - 后续如果补 `completion_latency` / `download_latency` 也沿同一路径扩展 + +这样做的目的不是把所有逻辑塞进一个巨型脚本，而是把： + +- 资源预留 +- 服务启停 +- benchmark 执行 +- 结果采集 + +变成一条可重复、可审计、默认安全的 batch compare 链路。 + +## 2. 设计边界 + +### 2.1 不把通用 serving 框架混进第一层主对标 + +`FastAPI`、`Ray Serve`、`BentoML`、`Xinference` 这类项目当然有价值，但它们解决的问题主要是： + +- HTTP 服务封装 +- 副本管理 +- 路由与部署 +- 多租户和线上治理 + +它们不是视频推理优化本身。 + +### 2.2 世界模型流式 benchmark 仍需要独立 harness + +当前开源里仍然没有一个像 AIPerf 这样成熟、通用的 **世界模型流式系统 benchmark**。 + +`AIPerf` 更适合当前的完整视频生成 HTTP 服务 benchmark，`telefuser stream-serve` 最终仍然需要单独建设一套 benchmark harness。 + +这类 benchmark 更应该关注： + +- session 建立时延 +- 首帧时延 +- steady-state FPS +- frame jitter +- control-to-frame latency +- 长会话稳定性 +- 会话并发数 +- 断线重连与恢复 + +当前仓库里已经补了一版最小可跑 harness： + +- `benchmarks/telefuser_aiperf/scripts/run_stream_bench.py` +- `benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh` +- `benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_quick.json` + +这一版的边界是： + +- 先对齐 `telefuser stream-serve` 的真实 WebRTC offer / answer 协议 +- 先对齐 `LingBotWorldFastService` 的 DataChannel 控制消息格式 +- 先把核心时延指标打通，不强行塞进 AIPerf core + +当前可直接采的指标有： + +- `offer RTT` +- `connected latency` +- `first frame latency` +- `first metadata latency` +- `stream FPS` +- `control ack latency` +- `control-to-next-frame latency` + +当前已知限制也要写清楚： + +- benchmark harness 本身支持多 session 参数 +- 但 `LingBotWorldFastService` 当前代码只允许 `1 active session` +- 所以现在这条 pipeline 还不能拿来做真正的多会话并发对比 + +### 2.2.1 自动化部署与运行链路 + +这条流式 benchmark 不应该依赖手工 SSH 拼命令。建议把准备流程固定成一条自动化链路： + +1. `sync` + - 同步 `benchmarks/telefuser_aiperf` + - 同步 `examples/lingbot/stream_lingbot_world_fast.py` + - 同步文档和 burner 脚本 + +2. `install` + - 在独立 benchmark venv 安装 AIPerf + - 安装 `aiortc` / `opencv-python-headless` + - 可选安装 TeleFuser 到独立服务 venv + +3. `verify` + - 检查脚本、配置、模型入口是否存在 + - 检查 `telefuser` / `aiperf` CLI + - 检查 WebRTC 依赖是否已装 + - 检查流式模型 zoo 路径是否可见 + +4. `start` + - 由远程服务环境启动 `telefuser stream-serve` + - 由 benchmark 环境启动 `run_stream_bench.sh` + +补充说明： + +- `stream-bootstrap` 会优先探测 `TF_MODEL_ZOO_PATH`，再回退到仓库内 `model_zoo` 和几个常见远程挂载点 +- 一旦解析到可用根目录，会在远程仓库下创建非破坏性的 `model_zoo -> 实际路径` 软链，后续启动命令可以直接用默认相对路径 +- 这样能避免每次手工记住 `TF_MODEL_ZOO_PATH`，也能保证 TeleFuser 服务和 benchmark 使用同一份权重根目录 + +建议的总入口是： + +```bash +python3 scripts/remote_bench_sync.py stream-bootstrap +``` + +这会把流式 benchmark 的前置条件一次性准备好，减少手工出错。 + +## 3. 后续扩展方向 + +如果后续要 benchmark `telefuser stream-serve`，建议在同一份总文档下继续扩展为： + +1. `SGLang-Omni` / `vLLM-Omni` 的系统框架接入 +2. WebRTC / SSE / WebSocket transport +3. world-model session endpoint +4. interactive trace workload +5. frame-level metrics +6. control-to-frame latency metrics + +这样才能覆盖 TeleFuser 最核心的实时世界模型场景，并保持和批处理视频 benchmark 在同一入口下管理。 + +## 4. AIPerf 需要做哪些改造 + +AIPerf 现在已经足够支撑 `telefuser serve` 这类批处理视频压测，但如果要把它扩展到 `stream-serve` 和更完整的世界模型流式场景，还需要补下面这些能力。 + +### 4.1 保留的部分 + +不需要重写的部分： + +- benchmark runner 和 config 解析 +- 并发控制、warmup、请求数、速率模型 +- 结果汇总、summary JSON、原始记录导出 +- HTTP trace 采集 +- 远程执行、结果落盘、artifact 管理 + +这些都可以继续复用。 + +### 4.2 需要新增的抽象 + +AIPerf **已经支持** chat 场景里的 multi-turn conversation benchmark，包括： + +- `multi_turn` 数据集 +- `--conversation-turn-mean` / `--session-turns-mean` +- `--conversation-turn-delay-mean` / `--session-turn-delay-mean` +- `sticky-user-sessions` + +所以这里不是“补一个多轮对话能力”这么简单，而是要把现有多轮对话能力进一步抽象成适合 `stream-serve` 的会话模型： + +- `request` 之外增加更明确的 `session` 生命周期 +- 支持 `session start / session step / session end` +- 支持同一 session 内多轮控制消息 +- 支持流式响应里的分帧事件 +- 支持把一个 session 的生命周期完整打包成一条 trace + +这样它才能覆盖 `stream-serve`，而不只是覆盖 chat multi-turn。 + +### 4.3 需要新增的 transport adapter + +现在 AIPerf 主要是 HTTP 请求模型。后续要对齐 `stream-serve`，建议加 adapter 层： + +- HTTP polling adapter +- WebSocket adapter +- SSE adapter +- WebRTC adapter + +每种 adapter 只负责 transport，不负责 benchmark 逻辑。 + +### 4.4 需要新增的 workload model + +除了现在这种 batch video workload，还要再加： + +- 长会话交互 workload +- 多轮控制 workload +- 连续帧输出 workload +- 动态 prompt / control message workload +- session 断开与恢复 workload + +如果只保留单次请求模型，就测不到 world model 的真实交互成本。 + +### 4.5 需要新增的指标 + +批处理视频继续看： + +- request latency +- throughput +- success rate +- peak memory +- tail latency + +流式世界模型还要看： + +- first frame latency +- steady-state FPS +- frame jitter +- control-to-frame latency +- session duration stability +- reconnect latency + +### 4.6 推荐的改造顺序 + +建议按这个顺序做： + +1. 保持现有 batch video benchmark 不动 +2. 把 `session` 抽象加到 AIPerf +3. 加 WebSocket / SSE adapter +4. 再补 WebRTC adapter +5. 最后接 `stream-serve` workload 和 frame-level 指标 + +这样可以先不破坏现有 `telefuser serve` 压测，同时逐步覆盖实时世界模型场景。 + +## 5. 参考链接 + +- [SGLang-Omni](https://sgl-project.github.io/sglang-omni/) +- [vLLM-Omni](https://docs.vllm.ai/projects/vllm-omni/en/latest/) +- [TeleFuser 480P 服务示例](../../examples/wan_video/wan21_14b_image_to_video_480p_service.py) +- [Diffusers baseline 服务](../../benchmarks/baseline/diffusers_wan_i2v/service.py) diff --git a/docs/zh/index.md b/docs/zh/index.md index c5f2a40..3d01156 100644 --- a/docs/zh/index.md +++ b/docs/zh/index.md @@ -56,6 +56,8 @@ telefuser stream-serve examples/stream_server/stream_lingbot_world_fast.py -p 80 - **[服务指南](service.md)** — 批量服务、任务 API 和 SDK - **[流式服务](stream_server.md)** — WebRTC 流式传输和双向控制 +- **[TeleFuser 与 AIPerf Benchmark](benchmark_aiperf.md)** — 批处理视频压测与运行入口 +- **[TeleFuser 与 AIPerf Benchmark 设计](benchmark_aiperf_design.md)** — 对标分层、stream-serve 规划和固定 workload 设计 - **[配置](configuration.md)** — 运行时和模型配置 - **[并行推理](parallel.md)** — 分布式处理策略 - **[新增模型](adding_new_model.md)** — 集成新模型 diff --git a/examples/lingbot/stream_lingbot_world_fast.py b/examples/lingbot/stream_lingbot_world_fast.py index f982bb8..381064a 100644 --- a/examples/lingbot/stream_lingbot_world_fast.py +++ b/examples/lingbot/stream_lingbot_world_fast.py @@ -1,6 +1,7 @@ from __future__ import annotations import os +from pathlib import Path import torch @@ -11,15 +12,15 @@ ) from telefuser.pipelines.lingbot_world_fast.service import LingBotWorldFastService -TF_MODEL_ZOO_PATH = os.environ.get("TF_MODEL_ZOO_PATH", "model_zoo") +TF_MODEL_ZOO_PATH = Path(os.environ.get("TF_MODEL_ZOO_PATH", "model_zoo")).expanduser().resolve() PPL_CONFIG = dict( name="lingbot_world_fast_stream", # LingBot-World-Fast reuses the Wan2.2 base weights (VAE + T5 text encoder + ``google/umt5-xxl`` # tokenizer), which ship in the shared Wan2.2-I2V-A14B directory. The DiT fast weights live in their - # own ``lingbot-world-fast`` directory, given as an absolute path so the pipeline keeps it standalone + # own ``lingbot-world-fast`` directory, passed as an absolute path so the pipeline keeps it standalone # rather than nesting it under ``checkpoint_dir``. - checkpoint_dir=TF_MODEL_ZOO_PATH + "/Wan2.2-I2V-A14B", - fast_checkpoint_subdir=TF_MODEL_ZOO_PATH + "/lingbot-world-fast", + checkpoint_dir=str(TF_MODEL_ZOO_PATH / "Wan2.2-I2V-A14B"), + fast_checkpoint_subdir=str(TF_MODEL_ZOO_PATH / "lingbot-world-fast"), control_type="cam", vae_device="cuda", vae_device_id=0, diff --git a/examples/wan_video/wan21_14b_image_to_video_480p_service.py b/examples/wan_video/wan21_14b_image_to_video_480p_service.py new file mode 100644 index 0000000..d06787d --- /dev/null +++ b/examples/wan_video/wan21_14b_image_to_video_480p_service.py @@ -0,0 +1,177 @@ +from __future__ import annotations + +import os + +import torch +from PIL import Image + +from telefuser.core.config import AttentionConfig, AttnImplType, WeightOffloadType +from telefuser.core.module_manager import ModuleManager +from telefuser.pipelines.wan_video.wan21_video import ( + Wan21VideoPipeline, + Wan21VideoPipelineConfig, +) +from telefuser.service.core.contract_templates import build_pipeline_manifest, build_task_contract_template +from telefuser.utils.video import save_video + +TF_MODEL_ZOO_PATH = os.environ.get("TF_MODEL_ZOO_PATH", "model_zoo") +PPL_CONFIG = dict( + name="wan21_14B_i2v_480p_service", + model_root=TF_MODEL_ZOO_PATH + "/Wan2.1-I2V-14B-480P", + negative_prompt="Camera shake, overly saturated colors, overexposed, static, blurry details, subtitles, style, artwork, painting, frame, still, overall grayish, worst quality, low quality, JPEG compression artifacts, ugly, incomplete, extra fingers, poorly drawn hands, poorly drawn face, deformed, disfigured, malformed limbs, fused fingers, static frames, cluttered background, three legs, crowded background, walking backwards", + num_inference_steps=40, + num_frames=81, + resolution="480p", + cfg_scale=5.0, + cfg_skip_ratio=0.0, + seed=42, + tiled=False, + sample_solver="unipc", + attn_impl=AttnImplType.TORCH_SDPA, + sigma_shift=5.0, + target_fps=16, +) + +PIPELINE_MANIFEST = build_pipeline_manifest( + pipeline_name=PPL_CONFIG["name"], + supported_tasks=["i2v"], + task_contracts={ + "i2v": build_task_contract_template( + "i2v", + parameter_overrides={ + "prompt": { + "description": "Positive guidance text prompt for Wan2.1 I2V 480P inference.", + "required": True, + }, + "negative_prompt": { + "default": "", + "description": "Optional negative prompt appended before the built-in Wan2.1 negative prompt.", + }, + "seed": { + "default": PPL_CONFIG["seed"], + "description": "Random seed used for the denoising trajectory.", + }, + "resolution": { + "default": PPL_CONFIG["resolution"], + "description": "Output resolution for this service example.", + "enum": ["480p"], + }, + "target_video_length": { + "default": 5, + "description": "Accepted by the OpenAI-compatible API but ignored by this fixed-workload benchmark service.", + "exposed": False, + }, + }, + excluded_parameters=("aspect_ratio",), + ) + }, +) + + +def get_dit_path_list(model_root: str) -> list[str]: + """Generate DiT model shard paths based on model_root.""" + return [ + f"{model_root}/diffusion_pytorch_model-00001-of-00007.safetensors", + f"{model_root}/diffusion_pytorch_model-00002-of-00007.safetensors", + f"{model_root}/diffusion_pytorch_model-00003-of-00007.safetensors", + f"{model_root}/diffusion_pytorch_model-00004-of-00007.safetensors", + f"{model_root}/diffusion_pytorch_model-00005-of-00007.safetensors", + f"{model_root}/diffusion_pytorch_model-00006-of-00007.safetensors", + f"{model_root}/diffusion_pytorch_model-00007-of-00007.safetensors", + ] + + +def get_pipeline(parallelism: int = 1, model_root: str = PPL_CONFIG["model_root"]): + """Build Wan2.1 14B 480P I2V pipeline for TeleFuser service.""" + dit_path_list = get_dit_path_list(model_root) + module_manager = ModuleManager(device="cpu") + module_manager.load_models( + [f"{model_root}/models_clip_open-clip-xlm-roberta-large-vit-huge-14.pth"], + torch_dtype=torch.float16, + ) + module_manager.load_models([f"{model_root}/Wan2.1_VAE.pth"], torch_dtype=torch.bfloat16) + module_manager.load_models([dit_path_list], torch_dtype=torch.bfloat16) + module_manager.load_models([f"{model_root}/models_t5_umt5-xxl-enc-bf16.pth"], torch_dtype=torch.bfloat16) + + pipe = Wan21VideoPipeline(device="cuda", torch_dtype=torch.bfloat16) + pipe_config = Wan21VideoPipelineConfig() + pipe_config.dit_config.attention_config = AttentionConfig.dense_attention(PPL_CONFIG["attn_impl"]) + pipe_config.dit_config.offload_config.offload_type = WeightOffloadType.MODEL_CPU_OFFLOAD + pipe_config.clip_config.offload_config.offload_type = WeightOffloadType.MODEL_CPU_OFFLOAD + pipe_config.vae_config.offload_config.offload_type = WeightOffloadType.MODEL_CPU_OFFLOAD + pipe_config.text_encoding_config.offload_config.offload_type = WeightOffloadType.MODEL_CPU_OFFLOAD + pipe_config.sample_solver = PPL_CONFIG["sample_solver"] + pipe_config.enable_clip_stage = True + + if parallelism > 1: + cfg_scale = PPL_CONFIG["cfg_scale"] + if cfg_scale > 1: + pipe_config.dit_config.parallel_config.cfg_degree = 2 + pipe_config.dit_config.parallel_config.sp_ulysses_degree = parallelism // 2 + else: + pipe_config.dit_config.parallel_config.sp_ulysses_degree = parallelism + + pipe_config.dit_config.parallel_config.device_ids = list(range(parallelism)) + pipe_config.enable_denoising_parallel = True + + pipe.init(module_manager, pipe_config) + return pipe + + +def run( + pipeline, + image: Image.Image, + prompt: str, + negative_prompt: str = "", + seed: int = PPL_CONFIG["seed"], + resolution: str = PPL_CONFIG["resolution"], +): + """Convert a static image to a video sequence using the fixed 480P Wan2.1 benchmark workload.""" + if resolution != "480p": + raise ValueError(f"Unsupported benchmark resolution: {resolution}") + + # Keep TeleFuser aligned with the official Diffusers Wan2.1-I2V-14B-480P + # benchmark shape instead of the repo-wide 480p area heuristic. + width, height = 832, 480 + video = pipeline( + prompt=prompt, + input_image=image, + negative_prompt=f"{negative_prompt} {PPL_CONFIG['negative_prompt']}", + num_inference_steps=PPL_CONFIG["num_inference_steps"], + num_frames=PPL_CONFIG["num_frames"], + cfg_scale=PPL_CONFIG["cfg_scale"], + seed=seed, + tiled=PPL_CONFIG["tiled"], + height=height, + width=width, + sigma_shift=PPL_CONFIG["sigma_shift"], + ) + return video + + +def run_with_file( + pipeline, + image: Image.Image, + prompt: str, + negative_prompt: str, + seed: int, + output_path: str, + resolution: str = PPL_CONFIG["resolution"], + **kwargs, +) -> dict: + """Run the fixed benchmark workload and save the result to file.""" + video = run( + pipeline, + image, + prompt, + negative_prompt, + seed, + resolution, + ) + save_video( + video, + output_path, + fps=PPL_CONFIG["target_fps"], + quality=6, + ) + return {"output_path": str(output_path)} diff --git a/mkdocs.yml b/mkdocs.yml index 424456d..782d516 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -122,12 +122,14 @@ plugins: nav_translations: Home: 首页 User Guides: 用户指南 + Benchmarks: 基准测试 Parallel Inference: 并行推理 Developer Guides: 开发者指南 Configuration: 配置 Tools: 工具 Service: 服务指南 Stream Server: 流式服务 + TeleFuser and AIPerf Benchmark: TeleFuser 与 AIPerf Benchmark Service Metadata: 服务元数据 Adding New Example: 新增示例 Model Loading: 模型加载 @@ -163,6 +165,8 @@ nav: - adding_new_example.md - model_loading.md - offload.md + - Benchmarks: + - TeleFuser and AIPerf Benchmark: benchmark_aiperf.md - Configuration: - configuration.md - Parallel Inference: diff --git a/pyproject.toml b/pyproject.toml index 695d1ed..1b1dc79 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,6 +39,12 @@ dependencies = [ "loguru", "openai", "fastapi", + # ASGI server runtime used to actually host the FastAPI app for + # `telefuser serve` / `stream-serve` in telefuser/service/main.py. + "uvicorn[standard]>=0.34.0,<1", + # Multipart/form-data parser required for FastAPI file upload and HTML form + # endpoints in service/api/routers/tasks.py and service/api/openai/*.py. + "python-multipart>=0.0.20", "click", "ftfy", "ray", diff --git a/telefuser/models/__init__.py b/telefuser/models/__init__.py index cd38b5d..c6a0920 100644 --- a/telefuser/models/__init__.py +++ b/telefuser/models/__init__.py @@ -9,16 +9,20 @@ # Cross-file registration: LTX 2.3 shared checkpoint contains 3 models # that live in separate files with circular import dependencies, so it -# cannot be registered inside any one of them. -from telefuser.core.model_registry import register_model_config -from telefuser.models.ltx_dit import LTXVideoTransformer -from telefuser.models.ltx_gemma_text_encoder import LTXEmbeddingsProcessor -from telefuser.models.ltx_video_vae import LTXVideoVAE - -register_model_config( - None, - "f3a83ecf3995dcc4fae2d27e08ad5767", - ["ltx_embeddings_processor", "ltx_video_vae", "ltx_dit"], - [LTXEmbeddingsProcessor, LTXVideoVAE, LTXVideoTransformer], - "official", -) +# cannot be registered inside any one of them. Keep this optional so +# stream-serving pipelines do not fail to import when LTX extras are absent. +try: + from telefuser.core.model_registry import register_model_config + from telefuser.models.ltx_dit import LTXVideoTransformer + from telefuser.models.ltx_gemma_text_encoder import LTXEmbeddingsProcessor + from telefuser.models.ltx_video_vae import LTXVideoVAE +except Exception: + pass +else: + register_model_config( + None, + "f3a83ecf3995dcc4fae2d27e08ad5767", + ["ltx_embeddings_processor", "ltx_video_vae", "ltx_dit"], + [LTXEmbeddingsProcessor, LTXVideoVAE, LTXVideoTransformer], + "official", + ) diff --git a/telefuser/pipelines/lingbot_world_fast/pipeline.py b/telefuser/pipelines/lingbot_world_fast/pipeline.py index 4e22735..d08ce5a 100644 --- a/telefuser/pipelines/lingbot_world_fast/pipeline.py +++ b/telefuser/pipelines/lingbot_world_fast/pipeline.py @@ -98,7 +98,11 @@ def init(self, module_manager, config: LingBotWorldFastPipelineConfig) -> None: self.vae.load_state_dict(vae_state_dict, strict=False) self.vae = self.vae.to(device=self.vae_device, dtype=self.torch_dtype).eval() - fast_path = checkpoint_root / config.fast_checkpoint_subdir + fast_path = Path(config.fast_checkpoint_subdir).expanduser() + if not fast_path.is_absolute(): + fast_path = (checkpoint_root / fast_path).resolve() + else: + fast_path = fast_path.resolve() self.dit = LingBotWorldFastDiT.from_pretrained( str(fast_path), torch_dtype=config.dit_torch_dtype, diff --git a/telefuser/service/api/openai/adapter.py b/telefuser/service/api/openai/adapter.py index 9289c8a..91b72aa 100644 --- a/telefuser/service/api/openai/adapter.py +++ b/telefuser/service/api/openai/adapter.py @@ -153,6 +153,17 @@ def _video_to_task(req: VideoGenerationsRequest, task_type: str | None = None) - target_video_length = req.seconds or 4 aspect_ratio = ASPECT_RATIOS.get(req.size, "16:9") ref_path = req.input_reference or req.reference_url or "" + extra_fields: dict[str, Any] = {"model": req.model} + if req.size: + extra_fields["openai_video_size"] = req.size + try: + width, height = validate_image_size(req.size) + except ValueError: + pass + else: + extra_fields["width"] = width + extra_fields["height"] = height + aspect_ratio = infer_aspect_ratio(width, height) base_fields = { "task": task, @@ -170,8 +181,6 @@ def _video_to_task(req: VideoGenerationsRequest, task_type: str | None = None) - else: base_fields["first_image_path"] = ref_path - extra_fields = {"model": req.model} - return create_extended_task_request(base_fields, extra_fields) @staticmethod diff --git a/telefuser/service/core/pipeline_runner.py b/telefuser/service/core/pipeline_runner.py index dbc5dcd..2b4bbcb 100644 --- a/telefuser/service/core/pipeline_runner.py +++ b/telefuser/service/core/pipeline_runner.py @@ -47,6 +47,32 @@ def _coerce_output_path(value: Any) -> str | None: return str(value) +def _first_image_path(task_data: dict[str, Any]) -> Any | None: + return task_data.get("first_image_path") or task_data.get("image_path") + + +def _load_rgb_image(path: Any) -> Any: + from PIL import Image + + return Image.open(path).convert("RGB") + + +def _inject_media_aliases(kwargs: dict[str, Any], task_data: dict[str, Any]) -> None: + first_path = _first_image_path(task_data) + if first_path: + if "image" not in kwargs: + kwargs["image"] = _load_rgb_image(first_path) + kwargs.setdefault("image_path", first_path) + + last_path = task_data.get("last_image_path") + if last_path: + kwargs.setdefault("last_image_path", last_path) + + ref_video_path = task_data.get("ref_video_path") + if ref_video_path: + kwargs.setdefault("video_path", ref_video_path) + + def _select_kwargs( fn: Any, *, @@ -58,17 +84,22 @@ def _select_kwargs( Prefer passing full task_data when **kwargs is supported; otherwise pass a compatible subset and apply a small set of common alias conversions for orchestrator-based scripts. """ + kwargs: dict[str, Any] = {} try: sig = inspect.signature(fn) except (TypeError, ValueError): - return dict(task_data) - - params = list(sig.parameters.values()) - accepts_var_kw = any(p.kind == inspect.Parameter.VAR_KEYWORD for p in params) - if accepts_var_kw: - return dict(task_data) + sig = None + + if sig is None: + kwargs.update(task_data) + params = [] + accepts_var_kw = True + else: + params = list(sig.parameters.values()) + accepts_var_kw = any(p.kind == inspect.Parameter.VAR_KEYWORD for p in params) + if accepts_var_kw: + kwargs.update(task_data) - kwargs: dict[str, Any] = {} for p in params: if p.kind in (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.VAR_POSITIONAL): continue @@ -85,17 +116,18 @@ def _select_kwargs( continue if name in ("image", "input_image"): - first_path = task_data.get("first_image_path") or task_data.get("image_path") + first_path = _first_image_path(task_data) if first_path: - from PIL import Image - - kwargs[name] = Image.open(first_path).convert("RGB") + kwargs[name] = _load_rgb_image(first_path) continue if name == "ppl_config" and module is not None and hasattr(module, "PPL_CONFIG"): kwargs[name] = getattr(module, "PPL_CONFIG") continue + if accepts_var_kw: + _inject_media_aliases(kwargs, task_data) + return kwargs diff --git a/telefuser/service/main.py b/telefuser/service/main.py index 35f4afe..b0c297f 100644 --- a/telefuser/service/main.py +++ b/telefuser/service/main.py @@ -8,6 +8,7 @@ from telefuser.core.config import TELEFUSER_LOGO from telefuser.service_types import TaskType from telefuser.utils.logging import logger +from telefuser.webrtc_ice import configure_ice_host_addresses from .core.config import server_config from .core.container import ServiceContainer @@ -88,6 +89,7 @@ def run_stream_server( Unlike run_server (request-response), this loads a stream pipeline that exposes get_service() and serves via WebRTC or WebSocket. """ + configure_ice_host_addresses() server_config.host = host server_config.port = port diff --git a/telefuser/webrtc_ice.py b/telefuser/webrtc_ice.py new file mode 100644 index 0000000..288670e --- /dev/null +++ b/telefuser/webrtc_ice.py @@ -0,0 +1,67 @@ +"""Helpers for constraining WebRTC ICE host candidate gathering.""" + +from __future__ import annotations + +import ipaddress +import os +from collections.abc import Iterable + + +def _normalize_host_ips(host_ips: Iterable[str]) -> list[str]: + normalized: list[str] = [] + seen: set[str] = set() + for raw_host_ip in host_ips: + host_ip = raw_host_ip.strip() + if not host_ip: + continue + canonical_host_ip = str(ipaddress.ip_address(host_ip)) + if canonical_host_ip in seen: + continue + seen.add(canonical_host_ip) + normalized.append(canonical_host_ip) + return normalized + + +def _matches_ip_version(host_ip: str, *, use_ipv4: bool, use_ipv6: bool) -> bool: + version = ipaddress.ip_address(host_ip).version + if version == 4: + return use_ipv4 + if version == 6: + return use_ipv6 + return False + + +def configure_ice_host_addresses(host_ips: Iterable[str] | None = None) -> None: + """Limit aiortc host candidate gathering to a small, explicit IP allowlist.""" + + if host_ips is None: + raw_host_ips = os.environ.get("TELEFUSER_WEBRTC_ICE_HOST_IPS", "") + if not raw_host_ips.strip(): + return + host_ips = raw_host_ips.split(",") + + allowed_host_ips = _normalize_host_ips(host_ips) + if not allowed_host_ips: + return + + try: + from aioice import ice as aioice_ice + except Exception: + return + + current_get_host_addresses = aioice_ice.get_host_addresses + if getattr(current_get_host_addresses, "_telefuser_host_filter", False): + return + + def _get_host_addresses(use_ipv4: bool, use_ipv6: bool) -> list[str]: + filtered = [ + host_ip + for host_ip in allowed_host_ips + if _matches_ip_version(host_ip, use_ipv4=use_ipv4, use_ipv6=use_ipv6) + ] + if filtered: + return filtered + return current_get_host_addresses(use_ipv4=use_ipv4, use_ipv6=use_ipv6) + + _get_host_addresses._telefuser_host_filter = True # type: ignore[attr-defined] + aioice_ice.get_host_addresses = _get_host_addresses diff --git a/tests/unit/openai/test_adapter.py b/tests/unit/openai/test_adapter.py index f0268b8..4a99543 100644 --- a/tests/unit/openai/test_adapter.py +++ b/tests/unit/openai/test_adapter.py @@ -65,6 +65,21 @@ def test_video_to_task_text_to_video(self): assert task_req.target_video_length == 5 assert task_req.aspect_ratio == "16:9" + def test_video_to_task_preserves_exact_openai_size(self): + """Keep exact WxH size available for fixed-workload benchmark services.""" + openai_req = VideoGenerationsRequest( + prompt="a cat playing", + seconds=5, + size="832x480", + ) + task_req = OpenAIRequestAdapter.to_task_request(openai_req) + + assert task_req.resolution == "480p" + assert task_req.openai_video_size == "832x480" + assert task_req.width == 832 + assert task_req.height == 480 + assert task_req.aspect_ratio == "26:15" + def test_video_to_task_image_to_video(self): """Convert I2V request (auto-detected from input).""" openai_req = VideoGenerationsRequest( diff --git a/tests/unit/service/test_pipeline_runner.py b/tests/unit/service/test_pipeline_runner.py new file mode 100644 index 0000000..727c8d8 --- /dev/null +++ b/tests/unit/service/test_pipeline_runner.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +from pathlib import Path + +from PIL import Image + +from telefuser.service.core.pipeline_runner import _select_kwargs + + +def test_select_kwargs_injects_image_for_var_kwargs_pipeline(tmp_path: Path) -> None: + image_path = tmp_path / "input.png" + Image.new("RGB", (16, 16), color="white").save(image_path) + + def run_with_file(pipeline, prompt, output_path, **kwargs): + return None + + kwargs = _select_kwargs( + run_with_file, + task_data={ + "prompt": "test prompt", + "output_path": "out.mp4", + "first_image_path": str(image_path), + }, + module=None, + ) + + assert kwargs["prompt"] == "test prompt" + assert kwargs["output_path"] == "out.mp4" + assert kwargs["image_path"] == str(image_path) + assert isinstance(kwargs["image"], Image.Image) + assert kwargs["image"].size == (16, 16)