diff --git a/.gitignore b/.gitignore index 194bb09..99e68f0 100755 --- a/.gitignore +++ b/.gitignore @@ -111,6 +111,8 @@ uv.lock .mypy_cache/ # custom +/artifacts/ +/benchmarks/aiperf/ /models /data .vscode @@ -142,6 +144,11 @@ docs/**/modelzoo.md *.MP3 *.mp3 +# Source-controlled AIPerf benchmark profiles use JSON despite the global rule. +!benchmarks/telefuser_aiperf/configs/*.json +!benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json +!benchmarks/baseline/sglang_lingbot_stream/configs/*.json + project output work_dirs diff --git a/CLAUDE.md b/CLAUDE.md index 51b087d..750a2a6 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -60,7 +60,7 @@ telefuser/ ├── feature_cache/ # Feature caching: AdaTaylorCache ├── cache/ # General cache management ├── offload/ # CPU offload strategies -├── metrics/ # Metrics collection and monitoring +├── metrics/ # Metrics collection, monitoring, and raw runtime facts ├── orchestrator/ # Request orchestration and actor-based streaming scheduler ├── worker/ # Distributed worker management ├── entrypoints/ # CLI entry points @@ -124,6 +124,13 @@ TeleFuser's model follows a strict layered architecture for operations: - May use `torch.library.custom_op` for torch.compile compatibility - Only used by ops/ layer, never directly by models/ +### Benchmark Metrics Boundary + +- `telefuser/metrics/runtime.py` measures synchronized target-side phase duration and allocator peaks. +- Target services emit only raw, bounded phase, chunk, and runtime facts. AIPerf owns warmup exclusion, + aggregation, semantic mapping, artifacts, and visualization. +- Client delivery and target compute metrics remain distinct (`stream_fps` versus `chunk_compute_fps`). + ## Code Style - PEP8 with ruff (line length: 120) @@ -135,6 +142,7 @@ TeleFuser's model follows a strict layered architecture for operations: | Topic | English | Chinese | |-------|---------|---------| +| AIPerf Benchmark | [docs/en/benchmark_aiperf.md](docs/en/benchmark_aiperf.md) | [docs/zh/benchmark_aiperf.md](docs/zh/benchmark_aiperf.md) | | Adding New Example | [docs/en/adding_new_example.md](docs/en/adding_new_example.md) | [docs/zh/adding_new_example.md](docs/zh/adding_new_example.md) | | Adding New Model | [docs/en/adding_new_model.md](docs/en/adding_new_model.md) | [docs/zh/adding_new_model.md](docs/zh/adding_new_model.md) | | Adding New Stage | [docs/en/adding_new_stage.md](docs/en/adding_new_stage.md) | [docs/zh/adding_new_stage.md](docs/zh/adding_new_stage.md) | diff --git a/benchmarks/baseline/sglang_lingbot_stream/README.md b/benchmarks/baseline/sglang_lingbot_stream/README.md new file mode 100644 index 0000000..633d3d9 --- /dev/null +++ b/benchmarks/baseline/sglang_lingbot_stream/README.md @@ -0,0 +1,53 @@ +# SGLang-Diffusion LingBot Stream Baseline + +This target compares TeleFuser LingBot streaming with the diffusion runtime in +`sgl-project/sglang` (`sglang.multimodal_gen`). It uses WebSocket + MessagePack while +TeleFuser uses WebRTC + DataChannel; AIPerf normalizes both into the same session and +control timeline. + +## Requirements + +- a version-pinned SGLang checkout that provides `LingBotWorldCausalDMDPipeline`; +- `robbyant/lingbot-world-fast-diffusers` or an equivalent local model path; +- the AIPerf checkout prepared by `scripts/setup_aiperf_repo.sh`. + +The launcher does not monkeypatch SGLang internals. Missing dependencies or incompatible +CUDA kernels must be fixed in the SGLang environment or recorded as a failed +qualification, not hidden behind an unversioned shim. + +## Start the target + +```bash +bash benchmarks/baseline/sglang_lingbot_stream/scripts/run_service.sh +``` + +Common overrides: + +```bash +SGLANG_PYTHON=/path/to/venv/bin/python \ +SGLANG_LINGBOT_MODEL_PATH=/path/to/model \ +SGLANG_LINGBOT_NUM_GPUS=1 \ +SGLANG_LINGBOT_ULYSSES_DEGREE=1 \ + bash benchmarks/baseline/sglang_lingbot_stream/scripts/run_service.sh +``` + +The default address is `http://127.0.0.1:30000`; readiness is checked at `/health`. + +## Run AIPerf + +```bash +bash benchmarks/baseline/sglang_lingbot_stream/scripts/run_stream_bench.sh + +bash benchmarks/baseline/sglang_lingbot_stream/scripts/run_stream_bench.sh \ + benchmarks/baseline/sglang_lingbot_stream/configs/stream_lingbot_world_fast_compare.json +``` + +The baseline reuses +`benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json`. AIPerf maps the shared +directional controls to SGLang camera actions and keeps implementation-specific fields +as raw evidence. + +For a valid performance comparison, use the same accelerator count, prompt, first +frame, FPS, session window, control trace, dtype, attention/cache geometry, and offload +policy. Record the exact SGLang and model revisions. Mock, native fallback, CPU offload, +and layerwise offload runs require separate qualifications. diff --git a/benchmarks/baseline/sglang_lingbot_stream/benchmark_contract.yaml b/benchmarks/baseline/sglang_lingbot_stream/benchmark_contract.yaml new file mode 100644 index 0000000..488c60b --- /dev/null +++ b/benchmarks/baseline/sglang_lingbot_stream/benchmark_contract.yaml @@ -0,0 +1,104 @@ +# Benchmark contract example for a WebSocket stream-world baseline. +# Enum comments use "one-of". List comments use "one-or-more". +contract_version: v1 # one-of: v1. Bump only when the contract schema changes. +name: sglang_lingbot_world_stream # Stable benchmark target id used in reports and automation. +mode: stream_world # one-of: batch_video, stream_world. +implementation: sglang_diffusion # Example values: telefuser, diffusers, sglang_diffusion. +model_family: lingbot_world_fast # Example values: wan_video, lingbot_world_fast, hunyuan_video, ltx_video. +model: robbyant/lingbot-world-fast-diffusers # Concrete model or service profile under test. +supported_tasks: # one-or-more: t2v, i2v, ti2v, bidirectional. + - bidirectional +transport: websocket # one-of: http, http_polling, websocket, sse, webrtc. +adapter: sglang_websocket # Built-in AIPerf adapter; target repository owns no transport implementation. +endpoint: + health_path: /health # Service readiness path. + metadata_path: /v1/models # Optional SGLang model and pipeline identity snapshot. + websocket_path: /v1/realtime_video/generate # WebSocket path for realtime video sessions. + models_path: /v1/models # Optional model metadata path. +request_encoding: + message_format: msgpack # one-of for this baseline: msgpack. + init_required_fields: # one-or-more. Required fields in the first WebSocket message. + - type + - prompt + - first_frame + - size + - fps + - num_frames + init_parameters: # Mapping from benchmark parameter names to SGLang init payload fields. + model: model + prompt: prompt + image_path: first_frame + size: size + fps: fps + num_frames: num_frames + control_channel: + transport: websocket_message # one-of for WebSocket: websocket_message. + message_type: event # Runtime control messages use type=event. + kind: camera_actions # one-of for LingBot realtime controls: camera_actions, prompt. + payload_mode: state # one-of for camera_actions: script, state. + action_tokens: # one-or-more. SGLang LingBot camera action tokens. + - w + - a + - s + - d + - i + - j + - k + - l + key_mapping: + ArrowUp: w + ArrowDown: s + ArrowLeft: a + ArrowRight: d +result_delivery: + media: websocket_frame_batch # one-of for this baseline: websocket_frame_batch. + metadata: websocket_chunk_stats # one-of for this baseline: websocket_chunk_stats. + session_log: sessions.jsonl # Per-session result records. + event_log: events/{phase}_{logical_session_index}_{session_id}.jsonl # Per-session event trace template. +workload: + mode: bidirectional # one-of: server_push, bidirectional. + task: bidirectional # one-of for this service: bidirectional. + size: 832x480 + fps: 16 + session_count: 1 + warmup_sessions: 1 + session_duration_s: 90.0 + control_trace: benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json # Timed control-event workload. + request_extra: # Service-specific request config passed through to SGLang. + num_frames: 9 + num_inference_steps: 4 + guidance_scale: 1.0 + realtime_causal_sink_size: 6 + realtime_causal_kv_cache_num_frames: 9 + realtime_output_format: webp # one-of: webp, jpeg, raw. + output_compression: 95 + max_chunks: 8 +metrics: # one-or-more. Choose all metrics emitted by this benchmark mode. + - connected_latency_ms + - first_frame_latency_ms + - first_metadata_latency_ms + - stream_fps + - session_runtime_s + - frames_received + - control_ack_latency_ms + - control_to_next_frame_latency_ms + - chunk_request_prepare_seconds + - chunk_compute_seconds + - chunk_encode_seconds + - chunk_output_pacing_seconds + - chunk_output_header_write_seconds + - chunk_output_payload_write_seconds + - chunk_output_write_seconds + - chunk_total_seconds + - chunk_compute_fps + - chunk_raw_output_bytes + - chunk_wire_output_bytes + - chunk_output_batches + - chunk_peak_reserved_bytes + - success_rate +limits: + active_sessions: 1 # Current SGLang realtime endpoint accepts one active session for this target. +artifacts: # Paths consumed by automation and documentation. + config: benchmarks/baseline/sglang_lingbot_stream/configs/stream_lingbot_world_fast_compare.json + control_trace: benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json + runner: aiperf profile --stream-config diff --git a/benchmarks/baseline/sglang_lingbot_stream/configs/stream_lingbot_world_fast_compare.json b/benchmarks/baseline/sglang_lingbot_stream/configs/stream_lingbot_world_fast_compare.json new file mode 100644 index 0000000..76597d4 --- /dev/null +++ b/benchmarks/baseline/sglang_lingbot_stream/configs/stream_lingbot_world_fast_compare.json @@ -0,0 +1,34 @@ +{ + "contract": "benchmarks/baseline/sglang_lingbot_stream/benchmark_contract.yaml", + "server_url": "http://127.0.0.1:30000", + "mode": "bidirectional", + "task": "bidirectional", + "prompt": "walk forward through the scene", + "image_path": "examples/data/1.png", + "fps": 16, + "session_count": 1, + "warmup_sessions": 1, + "warmup_chunks": 1, + "session_duration_s": 90.0, + "stagger_s": 0.0, + "control_trace_path": "benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json", + "request_extra": { + "num_frames": 9, + "num_inference_steps": 4, + "guidance_scale": 1.0, + "realtime_causal_sink_size": 6, + "realtime_causal_kv_cache_num_frames": 9, + "realtime_output_format": "webp", + "output_compression": 95, + "max_chunks": 8, + "realtime_output_pacing": false + }, + "transport": { + "connect_timeout_s": 60.0, + "message_timeout_s": 180.0 + }, + "server_metrics": { + "enabled": false + }, + "artifacts_dir": "artifacts/sglang_lingbot_stream/stream_lingbot_compare" +} diff --git a/benchmarks/baseline/sglang_lingbot_stream/configs/stream_lingbot_world_fast_quick.json b/benchmarks/baseline/sglang_lingbot_stream/configs/stream_lingbot_world_fast_quick.json new file mode 100644 index 0000000..6ad373c --- /dev/null +++ b/benchmarks/baseline/sglang_lingbot_stream/configs/stream_lingbot_world_fast_quick.json @@ -0,0 +1,33 @@ +{ + "contract": "benchmarks/baseline/sglang_lingbot_stream/benchmark_contract.yaml", + "server_url": "http://127.0.0.1:30000", + "mode": "bidirectional", + "task": "bidirectional", + "prompt": "walk forward through the scene", + "image_path": "examples/data/1.png", + "fps": 16, + "session_count": 1, + "warmup_sessions": 0, + "session_duration_s": 12.0, + "stagger_s": 0.0, + "control_trace_path": "benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json", + "request_extra": { + "num_frames": 9, + "num_inference_steps": 4, + "guidance_scale": 1.0, + "realtime_causal_sink_size": 6, + "realtime_causal_kv_cache_num_frames": 9, + "realtime_output_format": "webp", + "output_compression": 95, + "max_chunks": 8, + "realtime_output_pacing": false + }, + "transport": { + "connect_timeout_s": 30.0, + "message_timeout_s": 120.0 + }, + "server_metrics": { + "enabled": false + }, + "artifacts_dir": "artifacts/sglang_lingbot_stream/stream_lingbot_quick" +} diff --git a/benchmarks/baseline/sglang_lingbot_stream/scripts/run_service.sh b/benchmarks/baseline/sglang_lingbot_stream/scripts/run_service.sh new file mode 100755 index 0000000..fe34921 --- /dev/null +++ b/benchmarks/baseline/sglang_lingbot_stream/scripts/run_service.sh @@ -0,0 +1,47 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../../.." && pwd)" +cd "${ROOT_DIR}" + +if [[ -z "${CUDA_HOME:-}" && -d /usr/local/cuda ]]; then + export CUDA_HOME=/usr/local/cuda +fi + +if [[ -n "${SGLANG_EXTRA_PYTHONPATH:-}" ]]; then + export PYTHONPATH="${SGLANG_EXTRA_PYTHONPATH}${PYTHONPATH:+:${PYTHONPATH}}" +fi + +SGLANG_BIN="${SGLANG_BIN:-sglang}" +SGLANG_PYTHON="${SGLANG_PYTHON:-}" +SERVICE_PORT="${SGLANG_LINGBOT_PORT:-30000}" +MODEL_PATH="${SGLANG_LINGBOT_MODEL_PATH:-robbyant/lingbot-world-fast-diffusers}" +MODEL_ID="${SGLANG_LINGBOT_MODEL_ID:-lingbot-world-fast-diffusers}" +MODEL_TYPE="${SGLANG_LINGBOT_MODEL_TYPE:-diffusion}" +PIPELINE_CLASS="${SGLANG_LINGBOT_PIPELINE_CLASS:-LingBotWorldCausalDMDPipeline}" +PERFORMANCE_MODE="${SGLANG_LINGBOT_PERFORMANCE_MODE:-speed}" +ATTENTION_BACKEND_CONFIG="${SGLANG_LINGBOT_ATTENTION_BACKEND_CONFIG:-VSA_sparsity=0.0}" +NUM_GPUS="${SGLANG_LINGBOT_NUM_GPUS:-1}" +ULYSSES_DEGREE="${SGLANG_LINGBOT_ULYSSES_DEGREE:-1}" +DIT_CPU_OFFLOAD="${SGLANG_LINGBOT_DIT_CPU_OFFLOAD:-false}" +TEXT_ENCODER_CPU_OFFLOAD="${SGLANG_LINGBOT_TEXT_ENCODER_CPU_OFFLOAD:-false}" + +if [[ -n "${SGLANG_PYTHON}" ]]; then + SGLANG_CMD=("${SGLANG_PYTHON}" -c "from sglang.cli.main import main; main()") +else + read -r -a SGLANG_CMD <<< "${SGLANG_BIN}" +fi + +exec "${SGLANG_CMD[@]}" serve \ + --model-type "${MODEL_TYPE}" \ + --model-path "${MODEL_PATH}" \ + --model-id "${MODEL_ID}" \ + --pipeline-class-name "${PIPELINE_CLASS}" \ + --performance-mode "${PERFORMANCE_MODE}" \ + --attention-backend-config "${ATTENTION_BACKEND_CONFIG}" \ + --port "${SERVICE_PORT}" \ + --num-gpus "${NUM_GPUS}" \ + --ulysses-degree "${ULYSSES_DEGREE}" \ + --dit-cpu-offload "${DIT_CPU_OFFLOAD}" \ + --text-encoder-cpu-offload "${TEXT_ENCODER_CPU_OFFLOAD}" \ + "$@" diff --git a/benchmarks/baseline/sglang_lingbot_stream/scripts/run_stream_bench.sh b/benchmarks/baseline/sglang_lingbot_stream/scripts/run_stream_bench.sh new file mode 100755 index 0000000..1385d59 --- /dev/null +++ b/benchmarks/baseline/sglang_lingbot_stream/scripts/run_stream_bench.sh @@ -0,0 +1,49 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../../.." && pwd)" +cd "${ROOT_DIR}" + +AIPERF_DIR="${ROOT_DIR}/benchmarks/aiperf" +UV_BIN="${AIPERF_UV_BIN:-uv}" +CONFIG_PATH="${1:-benchmarks/baseline/sglang_lingbot_stream/configs/stream_lingbot_world_fast_quick.json}" +if [[ $# -gt 0 ]]; then + shift +fi + +if [[ ! -f "${AIPERF_DIR}/pyproject.toml" ]]; then + echo "AIPerf checkout not found. Run: bash scripts/setup_aiperf_repo.sh" >&2 + exit 1 +fi +if ! command -v "${UV_BIN}" >/dev/null 2>&1; then + echo "uv is required: https://docs.astral.sh/uv/getting-started/installation/" >&2 + exit 1 +fi + +SERVER_URL="${SGLANG_STREAM_BENCH_URL:-http://127.0.0.1:30000}" +SERVER_ARGS=(--stream-server-url "${SERVER_URL}") +for argument in "$@"; do + if [[ "${argument}" == "--stream-server-url" || "${argument}" == --stream-server-url=* ]]; then + SERVER_ARGS=() + break + fi +done + +RESOURCE_ARGS=() +RESOURCE_HISTORY_URL="${AIPERF_HISTORY_URL:-}" +RESOURCE_TARGET_PID="${AIPERF_RESOURCE_TARGET_PID:-${SGLANG_STREAM_BENCH_PID:-}}" +if [[ -n "${RESOURCE_HISTORY_URL}" || -n "${RESOURCE_TARGET_PID}" ]]; then + if [[ -z "${RESOURCE_HISTORY_URL}" || -z "${RESOURCE_TARGET_PID}" ]]; then + echo "AIPERF_HISTORY_URL and AIPERF_RESOURCE_TARGET_PID must be set together" >&2 + exit 2 + fi + RESOURCE_ARGS+=(--stream-resource-history-url "${RESOURCE_HISTORY_URL}") + RESOURCE_ARGS+=(--stream-resource-target-pid "${RESOURCE_TARGET_PID}") +fi + +exec "${UV_BIN}" run --frozen --no-dev --project "${AIPERF_DIR}" \ + aiperf profile \ + --stream-config "${CONFIG_PATH}" \ + "${SERVER_ARGS[@]}" \ + "${RESOURCE_ARGS[@]}" \ + "$@" diff --git a/benchmarks/telefuser_aiperf/benchmark_contract.yaml b/benchmarks/telefuser_aiperf/benchmark_contract.yaml new file mode 100644 index 0000000..3f980eb --- /dev/null +++ b/benchmarks/telefuser_aiperf/benchmark_contract.yaml @@ -0,0 +1,42 @@ +# Benchmark contract example for a batch-video target. +# Enum comments use "one-of". List comments use "one-or-more". +contract_version: v1 # one-of: v1. Bump only when the contract schema changes. +name: telefuser_aiperf_video # Stable benchmark target id used in reports and automation. +mode: batch_video # one-of: batch_video, stream_world. +implementation: telefuser # Example values: telefuser, diffusers, sglang_diffusion. +model_family: wan_video # Example values: wan_video, lingbot_world_fast, hunyuan_video, ltx_video. +model: Wan2.1-I2V-14B-480P # Concrete model or service profile under test. +supported_tasks: # one-or-more: t2v, i2v, ti2v, bidirectional. + - i2v +transport: http # one-of: http, http_polling, websocket, sse, webrtc. +endpoint: + path: /v1/videos # HTTP path used by the benchmark runner. + protocol: openai_video_generation # one-of for batch video: openai_video_generation. +request_encoding: + content_type: multipart/form-data # one-of: application/json, multipart/form-data. + media_inputs: # Mapping from dataset field to request field. + first_image_path: input_reference # Example: local image path becomes multipart input_reference. + parameters: # Mapping from benchmark parameter names to wire payload fields. + prompt: prompt + negative_prompt: negative_prompt + seed: seed + size: size + seconds: seconds +result_delivery: + polling_path: /v1/videos/{id} # Async status polling path. + content_path: /v1/videos/{id}/content # Generated media download path. +workload: + resolution: 832x480 # Example WxH string. Keep aligned across compared targets. + frames: 81 + inference_steps: 40 + fps: 16 + guidance_scale: 5.0 +metrics: # one-or-more. Choose all metrics emitted by this benchmark mode. + - request_latency + - completion_latency + - download_latency + - request_throughput + - success_rate +artifacts: # Paths consumed by automation and documentation. + 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/stream_lingbot_world_fast_compare.json b/benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_compare.json new file mode 100644 index 0000000..e922f1f --- /dev/null +++ b/benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_compare.json @@ -0,0 +1,42 @@ +{ + "contract": "benchmarks/telefuser_aiperf/stream_benchmark_contract.yaml", + "server_url": "http://127.0.0.1:8088", + "mode": "bidirectional", + "task": "bidirectional", + "prompt": "walk forward through the scene", + "image_path": "examples/data/1.png", + "fps": 16, + "session_count": 1, + "warmup_sessions": 1, + "warmup_chunks": 1, + "session_duration_s": 90.0, + "stagger_s": 0.0, + "control_trace_path": "benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json", + "request_extra": { + "chunk_size": 3, + "frame_num": 81, + "sample_shift": 5.0, + "control_mode": "cam", + "show_control_hud": false, + "benchmark_metrics": true + }, + "transport": { + "connect_timeout_s": 60.0, + "frame_timeout_s": 180.0, + "ice_gather_timeout_s": 5.0, + "shutdown_timeout_s": 5.0, + "receive_audio": false + }, + "server_metrics": { + "enabled": true, + "urls": [ + "http://127.0.0.1:8088/v1/service/metrics" + ], + "collection_interval_s": 1.0, + "export_raw_jsonl": true + }, + "observability": { + "mapping": "builtin:telefuser" + }, + "artifacts_dir": "artifacts/telefuser_aiperf/stream_lingbot_compare" +} diff --git a/benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_quick.json b/benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_quick.json new file mode 100644 index 0000000..31ff159 --- /dev/null +++ b/benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_quick.json @@ -0,0 +1,42 @@ +{ + "contract": "benchmarks/telefuser_aiperf/stream_benchmark_contract.yaml", + "server_url": "http://127.0.0.1:8088", + "mode": "bidirectional", + "task": "bidirectional", + "prompt": "walk forward through the scene", + "image_path": "examples/data/1.png", + "fps": 16, + "session_count": 1, + "warmup_sessions": 0, + "warmup_chunks": 1, + "session_duration_s": 30.0, + "stagger_s": 0.0, + "control_trace_path": "benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json", + "request_extra": { + "chunk_size": 3, + "frame_num": 81, + "sample_shift": 5.0, + "control_mode": "cam", + "show_control_hud": false, + "benchmark_metrics": true + }, + "transport": { + "connect_timeout_s": 30.0, + "frame_timeout_s": 60.0, + "ice_gather_timeout_s": 5.0, + "shutdown_timeout_s": 5.0, + "receive_audio": false + }, + "server_metrics": { + "enabled": true, + "urls": [ + "http://127.0.0.1:8088/v1/service/metrics" + ], + "collection_interval_s": 1.0, + "export_raw_jsonl": true + }, + "observability": { + "mapping": "builtin:telefuser" + }, + "artifacts_dir": "artifacts/telefuser_aiperf/stream_lingbot_quick" +} 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..a61e103 --- /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/v1/service/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..3d851f7 --- /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:832x480} + 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/stream_lingbot_controls.json b/benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json new file mode 100644 index 0000000..d70b5dd --- /dev/null +++ b/benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json @@ -0,0 +1,60 @@ +{ + "events": [ + { + "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" + } + }, + { + "delay_s": 6.4, + "message": { + "type": "control", + "key": "ArrowUp", + "action": "press" + } + } + ] +} diff --git a/benchmarks/telefuser_aiperf/data/video_prompts.jsonl b/benchmarks/telefuser_aiperf/data/video_prompts.jsonl new file mode 100644 index 0000000..79ff677 --- /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": "examples/data/101235-video-720_0.png"} +{"text": "The subject begins walking forward with subtle parallax in the background, natural lighting, smooth motion", "image": "examples/data/101235-video-720_0.png"} +{"text": "A gentle push-in camera move with soft environmental motion, realistic dynamics, no sudden cuts", "image": "examples/data/101235-video-720_0.png"} +{"text": "The scene gains light wind and small secondary motion while preserving composition and subject details", "image": "examples/data/101235-video-720_0.png"} +{"text": "A short cinematic continuation with stable structure, coherent motion, and clean temporal consistency", "image": "examples/data/101235-video-720_0.png"} diff --git a/benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh b/benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh new file mode 100755 index 0000000..92e9947 --- /dev/null +++ b/benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh @@ -0,0 +1,66 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)" +cd "${ROOT_DIR}" + +AIPERF_DIR="${ROOT_DIR}/benchmarks/aiperf" +UV_BIN="${AIPERF_UV_BIN:-uv}" +CONFIG_PATH="${1:-benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_quick.json}" +if [[ $# -gt 0 ]]; then + shift +fi + +if [[ ! -f "${AIPERF_DIR}/pyproject.toml" ]]; then + echo "AIPerf checkout not found. Run: bash scripts/setup_aiperf_repo.sh" >&2 + exit 1 +fi +if ! command -v "${UV_BIN}" >/dev/null 2>&1; then + echo "uv is required: https://docs.astral.sh/uv/getting-started/installation/" >&2 + exit 1 +fi + +SERVER_URL="${TELEFUSER_STREAM_BENCH_URL:-http://127.0.0.1:8088}" +SERVER_ARGS=(--stream-server-url "${SERVER_URL}") +for argument in "$@"; do + if [[ "${argument}" == "--stream-server-url" || "${argument}" == --stream-server-url=* ]]; then + SERVER_ARGS=() + break + fi +done +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+=(--stream-ice-host-ip "${ice_host_ip}") + fi + done +fi + +METRICS_ARGS=() +if [[ -n "${TELEFUSER_STREAM_BENCH_METRICS_URL:-}" ]]; then + METRICS_ARGS+=(--stream-server-metrics-url "${TELEFUSER_STREAM_BENCH_METRICS_URL}") +fi + +RESOURCE_ARGS=() +RESOURCE_HISTORY_URL="${AIPERF_HISTORY_URL:-}" +RESOURCE_TARGET_PID="${AIPERF_RESOURCE_TARGET_PID:-${TELEFUSER_STREAM_BENCH_PID:-}}" +if [[ -n "${RESOURCE_HISTORY_URL}" || -n "${RESOURCE_TARGET_PID}" ]]; then + if [[ -z "${RESOURCE_HISTORY_URL}" || -z "${RESOURCE_TARGET_PID}" ]]; then + echo "AIPERF_HISTORY_URL and AIPERF_RESOURCE_TARGET_PID must be set together" >&2 + exit 2 + fi + RESOURCE_ARGS+=(--stream-resource-history-url "${RESOURCE_HISTORY_URL}") + RESOURCE_ARGS+=(--stream-resource-target-pid "${RESOURCE_TARGET_PID}") +fi + +exec "${UV_BIN}" run --frozen --no-dev --project "${AIPERF_DIR}" --extra streaming-webrtc \ + aiperf profile \ + --stream-config "${CONFIG_PATH}" \ + "${SERVER_ARGS[@]}" \ + "${ICE_HOST_ARGS[@]}" \ + "${METRICS_ARGS[@]}" \ + "${RESOURCE_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..2f3869b --- /dev/null +++ b/benchmarks/telefuser_aiperf/scripts/run_video_bench.sh @@ -0,0 +1,33 @@ +#!/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_DIR="${ROOT_DIR}/benchmarks/aiperf" +UV_BIN="${AIPERF_UV_BIN:-uv}" +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 [[ ! -f "${AIPERF_DIR}/pyproject.toml" ]]; then + echo "AIPerf checkout not found. Run: bash scripts/setup_aiperf_repo.sh" >&2 + exit 1 +fi +if ! command -v "${UV_BIN}" >/dev/null 2>&1; then + echo "uv is required: https://docs.astral.sh/uv/getting-started/installation/" >&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}" +exec "${UV_BIN}" run --frozen --no-dev --project "${AIPERF_DIR}" aiperf profile --config "${CONFIG_PATH}" diff --git a/benchmarks/telefuser_aiperf/stream_benchmark_contract.yaml b/benchmarks/telefuser_aiperf/stream_benchmark_contract.yaml new file mode 100644 index 0000000..be088d2 --- /dev/null +++ b/benchmarks/telefuser_aiperf/stream_benchmark_contract.yaml @@ -0,0 +1,79 @@ +# Benchmark contract example for a WebRTC stream-world target. +# Enum comments use "one-of". List comments use "one-or-more". +contract_version: v1 # one-of: v1. Bump only when the contract schema changes. +name: telefuser_lingbot_world_fast_stream # Stable benchmark target id used in reports and automation. +mode: stream_world # one-of: batch_video, stream_world. +implementation: telefuser # Example values: telefuser, diffusers, sglang_diffusion. +model_family: lingbot_world_fast # Example values: wan_video, lingbot_world_fast, hunyuan_video, ltx_video. +model: LingBot-World-Fast # Concrete model or service profile under test. +supported_tasks: # one-or-more: t2v, i2v, ti2v, bidirectional. + - bidirectional +transport: webrtc # one-of: http, http_polling, websocket, sse, webrtc. +adapter: telefuser_webrtc # Built-in AIPerf adapter; target repository owns no transport implementation. +endpoint: + health_path: /v1/service/health # Service readiness path. + metadata_path: /v1/service/metadata # Optional target environment and startup phase facts. + offer_path: /v1/stream/webrtc/offer # WebRTC SDP offer/answer path. + delete_path_template: /v1/stream/webrtc/{session_id} # Session cleanup path template. +request_encoding: + offer_content_type: application/json # one-of: application/json. + offer_required_fields: # one-or-more. Required JSON fields in the offer payload. + - sdp + - type + - task + offer_parameters: # Mapping from benchmark parameter names to wire payload fields. + session_id: session_id + prompt: prompt + fps: fps + image_path: image_path + config: config + control_channel: + transport: datachannel # one-of for WebRTC: datachannel. + label: telefuser # DataChannel label expected by the service. + message_types: # one-or-more. Message types understood by the stream benchmark semantics. + - control + - status + - chunk + - done + - error +result_delivery: + media: rtp_video_track # one-of for WebRTC: rtp_video_track. + metadata: datachannel # one-of for WebRTC: datachannel. + session_log: sessions.jsonl # Per-session result records. + event_log: events/{phase}_{logical_session_index}_{session_id}.jsonl # Per-session event trace template. +workload: + mode: bidirectional # one-of: server_push, bidirectional. + task: bidirectional # one-of for this service: bidirectional. + fps: 16 + session_count: 1 + warmup_sessions: 1 + session_duration_s: 90.0 + control_trace: benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json # Timed control-event workload. + request_extra: # Service-specific request config passed through to the stream service. + chunk_size: 3 + frame_num: 81 + sample_shift: 5.0 + control_mode: cam # one-of for LingBotWorldFast: cam. + show_control_hud: false + benchmark_metrics: true # Emit synchronized runtime/chunk facts for AIPerf aggregation. +metrics: # one-or-more. Choose all metrics emitted by this benchmark mode. + - offer_rtt_ms + - connected_latency_ms + - first_frame_latency_ms + - first_metadata_latency_ms + - stream_fps + - session_runtime_s + - frames_received + - control_ack_latency_ms + - control_to_next_frame_latency_ms + - pipeline_init_seconds + - runtime_creation_seconds + - chunk_compute_seconds + - chunk_compute_fps + - success_rate +limits: + active_sessions: 1 # Service limit for this target. The harness still exposes session_count. +artifacts: # Paths consumed by automation and documentation. + config: benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_compare.json + control_trace: benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json + runner: aiperf profile --stream-config diff --git a/docs/en/benchmark_aiperf.md b/docs/en/benchmark_aiperf.md new file mode 100644 index 0000000..9c9f1c9 --- /dev/null +++ b/docs/en/benchmark_aiperf.md @@ -0,0 +1,226 @@ +# TeleFuser and AIPerf + +TeleFuser exposes raw target-side facts; AIPerf owns workload execution, aggregation, +resource collection, artifacts, GreptimeDB history, and visualization. This separation +keeps the same benchmark and dashboard reusable across TeleFuser, SGLang-Diffusion, and +future targets. + +The included assets cover: + +- Wan2.1 image-to-video through the OpenAI-compatible `/v1/videos` API; +- LingBot-World-Fast sessions through WebRTC and DataChannel; +- a LingBot SGLang-Diffusion baseline through WebSocket and MessagePack. + +## Repository layout + +```text +benchmarks/ +├── telefuser_aiperf/ # TeleFuser contracts, configs, data, launchers +├── baseline/sglang_lingbot_stream/ # Stream baseline +└── aiperf/ # Ignored external AIPerf checkout +``` + +The AIPerf implementation is not vendored into TeleFuser. The setup script always uses +`/benchmarks/aiperf`; neither the setup script nor the launchers accept a +checkout-path override. Install +[uv](https://docs.astral.sh/uv/getting-started/installation/), then run this once from +the TeleFuser repository root: + +```bash +bash scripts/setup_aiperf_repo.sh +``` + +The script clones AIPerf, creates its isolated runtime environment with WebRTC support, +and creates `/artifacts` for benchmark output and History imports. The +dashboard is bundled, so runtime users do not need Node.js or a separate frontend +process. Pin a commit for reproducible runs: + +```bash +AIPERF_REF= bash scripts/setup_aiperf_repo.sh +``` + +`AIPERF_REPO_URL`, `AIPERF_BRANCH`, and `AIPERF_REF` may select the source and revision, +but never change the checkout location. + +## Batch video + +Start the fixed Wan2.1 I2V target: + +```bash +telefuser serve \ + examples/wan_video/wan21_14b_image_to_video_480p_service.py \ + --port 8000 \ + --task i2v +``` + +Run a smoke profile or the fixed comparison workload: + +```bash +bash benchmarks/telefuser_aiperf/scripts/run_video_bench.sh + +bash benchmarks/telefuser_aiperf/scripts/run_video_bench.sh \ + benchmarks/telefuser_aiperf/configs/video_generation_wan21_i2v_480p_compare.yaml +``` + +The launcher checks `/v1/service/health` before profiling. Common overrides include +`TELEFUSER_AIPERF_URL`, `TELEFUSER_AIPERF_CONCURRENCY`, +`TELEFUSER_AIPERF_REQUESTS`, `TELEFUSER_AIPERF_SIZE`, and +`TELEFUSER_AIPERF_SECONDS`. + +## LingBot stream + +Start TeleFuser: + +```bash +telefuser stream-serve \ + examples/lingbot/lingbot_world_fast_image_to_video_h100.py \ + -p 8088 \ + --skip-validation +``` + +Then run: + +```bash +bash benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh + +bash benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh \ + benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_compare.json +``` + +The stream config enables `benchmark_metrics`. TeleFuser then reports synchronized raw +facts for runtime creation, actor-graph chunk compute, cache geometry, and environment +identity. Allocator peaks are omitted because generation runs in child actors and the +service-process allocator cannot represent the complete graph; active AIPerf resource +telemetry supplies process-tree GPU-memory curves instead. The native WebRTC path does +not report a separate payload encoding duration because encoding happens after the target +chunk fact. AIPerf computes warmup-aware summaries and keeps client delivery separate +from target compute. + +The shared timed control trace is stored at +`benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json`. + +## SGLang-Diffusion baseline + +Use a compatible, version-pinned `sgl-project/sglang` environment. The TeleFuser tree +does not patch SGLang modules at import time. + +```bash +bash benchmarks/baseline/sglang_lingbot_stream/scripts/run_service.sh +bash benchmarks/baseline/sglang_lingbot_stream/scripts/run_stream_bench.sh +``` + +The baseline uses the same prompt, first frame, FPS target, session window, and control +trace. The adapter translates only transport semantics. For a performance comparison, +record the exact SGLang commit and model revision, use GPU-resident speed mode, and keep +offload and fallback settings identical. An OOM is a result for that configuration; do +not replace it with a mock or offloaded result under the same label. + +Both documented launch commands default to one GPU. Override both targets explicitly +when comparing another accelerator count. + +## Configs + +| Config | Purpose | +|---|---| +| `video_generation_quick.yaml` | Batch connectivity and latency smoke test | +| `video_generation_e2e.yaml` | Batch warmup, trace, records, and server metrics | +| `video_generation_rate.yaml` | Poisson-arrival Batch load | +| `video_generation_wan21_i2v_480p_compare.yaml` | Fixed Wan I2V comparison | +| `stream_lingbot_world_fast_quick.json` | Bounded Stream smoke test | +| `stream_lingbot_world_fast_compare.json` | Fixed LingBot Stream comparison | + +SGLang equivalents are under `benchmarks/baseline/sglang_lingbot_stream/configs`. + +## Metric interpretation + +The most important distinction is scope: + +| Metric | Meaning | +|---|---| +| `stream_fps` | Frames received by the client divided by client session time | +| `chunk_compute_fps` | Frames divided by compute time for one target chunk | +| `chunk_compute_fps_weighted` | `sum(frames) / sum(compute_seconds)` after warmup exclusion | + +AIPerf presents metrics under five stable dimensions: delivery, latency, throughput, +target execution, and resources. Implementation-specific fields remain raw evidence and +map into these canonical leaves; they do not become separate top-level metrics. + +## Active resource history + +Docker provides the shortest persistent GreptimeDB setup: + +```bash +docker volume create aiperf-greptime-data +docker run -d --name aiperf-greptime --restart unless-stopped \ + -p 127.0.0.1:4000:4000 \ + -v aiperf-greptime-data:/greptimedb_data \ + greptime/greptimedb:latest \ + standalone start \ + --http-addr 0.0.0.0:4000 \ + --data-home /greptimedb_data +``` + +Pin the image tag or digest for production. The named volume keeps history across +container restarts. Then start the bundled AIPerf API and frontend from the TeleFuser +repository root; the `artifacts` root matches the benchmark launchers: + +```bash +uv run --frozen --no-dev --project benchmarks/aiperf aiperf history serve \ + --greptime-url http://127.0.0.1:4000 \ + --greptime-database public \ + --artifact-root artifacts \ + --host 127.0.0.1 \ + --port 8095 +``` + +Verify the stack: + +```bash +curl --fail http://127.0.0.1:8095/api/v1/history/health +curl --fail -X POST 'http://127.0.0.1:4000/v1/sql?db=public' \ + --data-urlencode 'sql=SELECT 1 AS ready' +``` + +Enable active collection on the target host: + +```bash +export AIPERF_HISTORY_URL=http://:8095 +export AIPERF_RESOURCE_TARGET_PID= + +bash benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh +``` + +The agent recursively observes the target process tree. It samples every second, +uploads every 15 seconds, timestamps samples at the source, and flushes at termination. +It reports process, container-when-detectable, and machine facts for CPU, memory, GPU, +VRAM, Ethernet, and RDMA. Capacity is kept separate from usage. + +GreptimeDB is mandatory for History and active reporting. Startup, query, or final-flush +failure is surfaced; there is no SQLite, in-memory, or direct-file query fallback. + +Open `http://127.0.0.1:8095/` for the Chinese desktop dashboard. It supports two run +groups, canonical metric-tree selection, aggregate curves, resource timelines, and +cross-run comparison. + +For a remote benchmark host, keep the default loopback bind and forward it securely: + +```bash +ssh -L 8095:127.0.0.1:8095 user@benchmark-host +``` + +## Artifacts and reproducibility + +Batch and stream launchers write timestamped artifacts below `artifacts/`. Stream +artifacts include summaries, session and event JSONL, target metadata, normalized +metrics, and a standalone HTML report. + +Every performance result should retain: + +- TeleFuser or SGLang commit and model revision; +- accelerator model/count, driver, CUDA, PyTorch, and dtype; +- workload config and control trace; +- warmup policy and successful/failed session counts; +- offload, cache, attention, and fallback settings. + +See the Chinese [benchmark design](/TeleFuser/zh/benchmark_aiperf_design/) for protocol and +ownership details. diff --git a/docs/en/index.md b/docs/en/index.md index 7fe16f1..c96457e 100644 --- a/docs/en/index.md +++ b/docs/en/index.md @@ -96,6 +96,7 @@ telefuser stream-serve examples/lingbot/lingbot_world_fast_image_to_video_h100.p Service GuideBatch serving, task APIs, and SDK. Stream ServerWebRTC streaming and bidirectional control. Stream SchedulerActor ownership, bounded dataflow, lifecycle, metrics, and GPU placement. +AIPerf BenchmarkBatch, stream, baseline, and history workflows. ConfigurationRuntime, attention, quantization, and offload settings. Parallel InferenceDistributed processing strategies. Adding New ModelIntegrate new model architectures and stages. diff --git a/docs/zh/benchmark_aiperf.md b/docs/zh/benchmark_aiperf.md new file mode 100644 index 0000000..6fb8d04 --- /dev/null +++ b/docs/zh/benchmark_aiperf.md @@ -0,0 +1,208 @@ +# TeleFuser 与 AIPerf + +TeleFuser 只暴露目标侧原始事实；AIPerf 统一负责 workload 执行、指标聚合、资源采集、产物、GreptimeDB +历史服务和前端展示。这样同一套 benchmark 与界面可以复用于 TeleFuser、SGLang-Diffusion 和后续实现。 + +当前资产覆盖： + +- 通过 OpenAI 兼容 `/v1/videos` API 测试 Wan2.1 图生视频； +- 通过 WebRTC 与 DataChannel 测试 LingBot-World-Fast； +- 通过 WebSocket 与 MessagePack 测试 SGLang-Diffusion LingBot baseline。 + +## 仓库边界 + +```text +benchmarks/ +├── telefuser_aiperf/ # TeleFuser contract、配置、数据和启动器 +├── baseline/sglang_lingbot_stream/ # Stream baseline +└── aiperf/ # 被 Git 忽略的外部 AIPerf checkout +``` + +TeleFuser 不 vendoring AIPerf 实现。安装脚本与所有 launcher 都固定使用 +`/benchmarks/aiperf`，不提供 checkout 路径覆盖。先安装 +[uv](https://docs.astral.sh/uv/getting-started/installation/)，然后在 TeleFuser 仓库根目录执行一次： + +```bash +bash scripts/setup_aiperf_repo.sh +``` + +脚本会 clone AIPerf、创建包含 WebRTC 支持的隔离运行环境，并创建用于 benchmark 输出与 History +导入的 `/artifacts`。前端产物已经内置，普通用户不需要安装 Node.js，也不需要单独启动 +前端进程。正式实验应固定 AIPerf commit： + +```bash +AIPERF_REF= bash scripts/setup_aiperf_repo.sh +``` + +`AIPERF_REPO_URL`、`AIPERF_BRANCH` 和 `AIPERF_REF` 只控制来源与 revision，不改变 checkout 位置。 + +## Batch 视频测试 + +启动固定 Wan2.1 I2V target： + +```bash +telefuser serve \ + examples/wan_video/wan21_14b_image_to_video_480p_service.py \ + --port 8000 \ + --task i2v +``` + +执行快速测试或固定对比 workload： + +```bash +bash benchmarks/telefuser_aiperf/scripts/run_video_bench.sh + +bash benchmarks/telefuser_aiperf/scripts/run_video_bench.sh \ + benchmarks/telefuser_aiperf/configs/video_generation_wan21_i2v_480p_compare.yaml +``` + +启动器会先检查 `/v1/service/health`。常用覆盖变量包括 +`TELEFUSER_AIPERF_URL`、`TELEFUSER_AIPERF_CONCURRENCY`、 +`TELEFUSER_AIPERF_REQUESTS`、`TELEFUSER_AIPERF_SIZE` 和 +`TELEFUSER_AIPERF_SECONDS`。 + +## LingBot Stream 测试 + +启动 TeleFuser： + +```bash +telefuser stream-serve \ + examples/lingbot/lingbot_world_fast_image_to_video_h100.py \ + -p 8088 \ + --skip-validation +``` + +执行测试： + +```bash +bash benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh + +bash benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh \ + benchmarks/telefuser_aiperf/configs/stream_lingbot_world_fast_compare.json +``` + +Stream 配置通过 `benchmark_metrics: true` 开启目标侧原始事实。TeleFuser 同步记录 runtime 创建、actor graph +chunk 计算、cache 几何和运行环境。生成工作位于子 actor 中，服务进程的 allocator 无法代表完整 actor graph， +因此不对 LingBot 上报不完整的 allocator 峰值；完整进程树显存曲线由 AIPerf 主动资源采集提供。原生 WebRTC +的编码位于 chunk fact 之后，因此当前不伪造独立编码耗时。AIPerf 负责跳过 warmup 并生成聚合结果。 + +TeleFuser 与 SGLang 共用 +`benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json` 中的定时控制 trace。 + +## SGLang-Diffusion baseline + +使用兼容且固定版本的 `sgl-project/sglang` 环境。TeleFuser 仓库不会在 import 时 monkeypatch SGLang +内部模块。 + +```bash +bash benchmarks/baseline/sglang_lingbot_stream/scripts/run_service.sh +bash benchmarks/baseline/sglang_lingbot_stream/scripts/run_stream_bench.sh +``` + +Baseline 固定 prompt、首帧、FPS、session 时长和控制 trace，只由 adapter 转换 transport 语义。正式性能 +对比必须记录 SGLang commit 与模型 revision，使用 GPU-resident speed mode，并保持 offload、fallback、cache +和 attention 设置一致。某个配置 OOM 就应记录为该配置失败，不能用 mock 或 offload 结果替代。 + +本文两条启动命令默认都使用 1 张 GPU；比较其他卡数时必须同时显式覆盖两个 target。 + +## 配置清单 + +| 配置 | 用途 | +|---|---| +| `video_generation_quick.yaml` | Batch 连通性与延迟 smoke test | +| `video_generation_e2e.yaml` | Batch warmup、trace、records 和服务指标 | +| `video_generation_rate.yaml` | Poisson 到达负载 | +| `video_generation_wan21_i2v_480p_compare.yaml` | 固定 Wan I2V 对比 workload | +| `stream_lingbot_world_fast_quick.json` | 有界 Stream smoke test | +| `stream_lingbot_world_fast_compare.json` | 固定 LingBot Stream 对比 workload | + +SGLang 对应配置位于 `benchmarks/baseline/sglang_lingbot_stream/configs`。 + +## 指标解释 + +必须区分指标 scope： + +| 指标 | 含义 | +|---|---| +| `stream_fps` | 客户端收到帧数除以客户端 session 时间 | +| `chunk_compute_fps` | 单个目标 chunk 的帧数除以计算时间 | +| `chunk_compute_fps_weighted` | 排除 warmup 后的 `sum(frames) / sum(compute_seconds)` | + +AIPerf 按交付、时延、吞吐、目标执行和资源五个稳定维度展示指标。不同实现的细分上报先保留为原始证据， +再映射到 canonical leaf，不扩张成新的顶层指标。 + +## 主动资源上报与历史曲线 + +使用 Docker 可以直接启动带持久化卷的 GreptimeDB： + +```bash +docker volume create aiperf-greptime-data +docker run -d --name aiperf-greptime --restart unless-stopped \ + -p 127.0.0.1:4000:4000 \ + -v aiperf-greptime-data:/greptimedb_data \ + greptime/greptimedb:latest \ + standalone start \ + --http-addr 0.0.0.0:4000 \ + --data-home /greptimedb_data +``` + +生产环境应固定镜像 tag 或 digest。命名卷会在容器重启后保留历史数据。随后在 TeleFuser 仓库根目录 +启动内置的 AIPerf 后端和中文前端；`artifacts` 与 benchmark launcher 的输出目录一致： + +```bash +uv run --frozen --no-dev --project benchmarks/aiperf aiperf history serve \ + --greptime-url http://127.0.0.1:4000 \ + --greptime-database public \ + --artifact-root artifacts \ + --host 127.0.0.1 \ + --port 8095 +``` + +检查前后端和数据库： + +```bash +curl --fail http://127.0.0.1:8095/api/v1/history/health +curl --fail -X POST 'http://127.0.0.1:4000/v1/sql?db=public' \ + --data-urlencode 'sql=SELECT 1 AS ready' +``` + +在 target 所在机器开启主动采集： + +```bash +export AIPERF_HISTORY_URL=http://:8095 +export AIPERF_RESOURCE_TARGET_PID= + +bash benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh +``` + +Agent 递归观测目标进程树，默认每 1 秒采样、每 15 秒上报，并在任务结束时 flush。它采集 CPU、内存、 +GPU、显存、Ethernet 和 RDMA 的进程、可探测容器与整机事实。用量曲线与整机容量始终分开；CPU 以一个 +逻辑核为 100%，多核和多卡允许超过 100%。 + +GreptimeDB 是 History 与主动上报的强依赖。启动、查询或最终 flush 失败会直接暴露，不会切换到 SQLite、 +内存索引或文件直查。 + +打开 `http://127.0.0.1:8095/` 查看中文桌面界面。页面支持左右两组 Run、按 canonical 指标树选择图表、 +同时展示 avg/P95/P99、资源时间折线和跨实验对比。 + +远端实验机建议保持默认 loopback 监听，并通过 SSH 安全转发： + +```bash +ssh -L 8095:127.0.0.1:8095 user@benchmark-host +``` + +## 产物与复现要求 + +Batch 和 Stream 启动器默认将带时间戳的结果写入 `artifacts/`。Stream 产物包含 summary、session/event +JSONL、目标 metadata、normalized metrics 和独立 HTML 报告。 + +正式性能结果至少应保留： + +- TeleFuser 或 SGLang commit 与模型 revision； +- GPU 型号/数量、driver、CUDA、PyTorch 和 dtype； +- workload 配置与 control trace； +- warmup 规则及成功/失败 session 数； +- offload、cache、attention 和 fallback 设置。 + +协议和职责边界见 [TeleFuser 与 AIPerf Benchmark 设计](benchmark_aiperf_design.md)。动态实验数值保存在 +GreptimeDB 和可重放产物中，不写入稳定用户文档。 diff --git a/docs/zh/benchmark_aiperf_design.md b/docs/zh/benchmark_aiperf_design.md new file mode 100644 index 0000000..983196e --- /dev/null +++ b/docs/zh/benchmark_aiperf_design.md @@ -0,0 +1,214 @@ +# TeleFuser 与 AIPerf Benchmark 设计 + +本文定义稳定的职责、协议和指标语义。具体实验数值、机器地址和运行状态不属于设计文档，应保存在 +GreptimeDB 与可重放产物中。 + +## 1. 目标与非目标 + +设计目标： + +- 同一 workload 可以比较 TeleFuser、SGLang-Diffusion 和后续实现； +- 客户端交付性能、目标侧计算性能和资源使用互不混淆； +- Batch 与 Stream 共用产物、历史查询和展示维度； +- target 只上报原始、有限、有时间戳的事实； +- AIPerf 统一负责采集生命周期、warmup、聚合、映射、存储和界面。 + +非目标： + +- 不在 TeleFuser 内复制 AIPerf、GreptimeDB client 或历史前端； +- 不用 mock、offload 或 fallback 结果替代正式 GPU-resident 结果； +- 不把实现私有字段全部提升为用户可选的顶层指标。 + +## 2. 仓库与依赖边界 + +```mermaid +flowchart LR + TF[TeleFuser target] -->|raw phase/chunk/runtime facts| AP[AIPerf] + SG[SGLang target] -->|native stream facts| AP + AP -->|canonical artifacts| GT[GreptimeDB] + AP --> UI[Vue history dashboard] + AG[AIPerf resource agent] -->|timestamped batches| AP + TF -. no database dependency .-> GT + SG -. no AIPerf package dependency .-> AP +``` + +| 组件 | 归属 | 职责 | +|---|---|---| +| TeleFuser runtime | TeleFuser | 同步测量 target phase/chunk，暴露环境与 cache 原始事实 | +| Target adapter | AIPerf | 将 HTTP、WebRTC、WebSocket 等 wire event 转成统一 session timeline | +| 聚合与语义映射 | AIPerf | warmup、percentile、weighted FPS、canonical metric | +| Resource agent | AIPerf | 采样目标进程树、cgroup、机器和设备资源并主动上报 | +| History API/UI | AIPerf | GreptimeDB schema、查询、左右 Run 对比和图表 | +| Contract/config/data | Target 仓库 | 固定 target 能力、workload 和可复现入口 | + +`benchmarks/aiperf/` 是被 Git 忽略的固定外部 checkout。`scripts/setup_aiperf_repo.sh` 与 benchmark launcher +不接受路径覆盖，AIPerf 必须位于 `/benchmarks/aiperf`。repo URL、branch 和 ref 可以调整，正式运行 +仍必须固定 commit，不能只记录可移动分支名。 + +## 3. 场景与实现 + +| 场景 | TeleFuser | Baseline | 公平性边界 | +|---|---|---|---| +| Batch Video | OpenAI 兼容 `/v1/videos` | 兼容相同 contract 的外部 target | prompt、输入图、尺寸、帧数、steps、seed 一致 | +| Stream World | WebRTC media + DataChannel | SGLang WebSocket + MessagePack | prompt、首帧、FPS、session、control trace、GPU/offload 策略一致 | +| Transport Mock | WebRTC mock | WebSocket mock | 只比较 transport/harness，不解释模型性能 | + +Transport 可以不同，但 adapter 输出的逻辑事件必须一致：连接、首帧、控制发送、控制确认、下一帧、chunk +事实和 session 结束。 + +## 4. Target 原始事实协议 + +### 4.1 通用规则 + +- duration 使用单调时钟；跨进程/跨机器样本同时携带源端 UTC 时间戳； +- CUDA phase 在开始和结束边界同步目标设备； +- 数值必须有限且非负，不可用字段省略或为 `null`，不能伪造为零； +- memory 使用 bytes 作为线协议单位；显示层再转换为 MB/GB； +- target 不排除 warmup、不计算 percentile、不生成跨 Run 结论。 + +### 4.2 Phase fact + +```json +{ + "name": "pipeline_init", + "seconds": 12.3, + "memory": [ + { + "device": "cuda:0", + "peak_allocated_bytes": 123, + "peak_reserved_bytes": 456 + } + ] +} +``` + +TeleFuser Stream metadata 可以提供 `pipeline_init`；首次 LingBot chunk 之前提供 `runtime_creation`。 + +### 4.3 Chunk fact + +```json +{ + "index": 3, + "frames": 3, + "compute_seconds": 0.45, + "memory": [] +} +``` + +TeleFuser 的 `compute_seconds` 从 chunk 提交 actor graph 前开始，覆盖 encode、denoise、decode 与目标内调度， +结束于原始帧返回。`encode_seconds` 是 AIPerf 支持的可选事实，仅在 target 能给出有界 payload 编码阶段时 +上报；当前 TeleFuser 原生 WebRTC 编码发生在 chunk fact 之后，因此省略该字段。客户端网络接收、播放 pacing +和 UI 渲染不进入 target compute。当前 LingBot 生成位于子 actor 中，服务进程的 CUDA allocator 统计不能覆盖 +完整 actor graph，因此 `memory` 保持为空；进程树显存时序由 AIPerf resource telemetry 采集，不能拿它替代 +reset-scoped allocator peak。 + +### 4.4 Runtime fact + +LingBot runtime 只上报稳定几何信息： + +- width、height、latent frames 和 frame tokens； +- chunk size 与 max attention size； +- local attention、sink 和 KV cache capacity。 + +软件环境至少包含 TeleFuser commit、Python、PyTorch、CUDA 以及可见 GPU 型号、compute capability 和显存容量。 + +## 5. AIPerf 聚合语义 + +### 5.1 Scope + +| Scope | 示例 | 聚合规则 | +|---|---|---| +| Event | control ack、frame arrival | 保留单事件时间线 | +| Chunk | `chunk_compute_fps` | `frames / compute_seconds` | +| Session | `stream_fps`、first-frame latency | 每个 session 独立计算 | +| Run | `chunk_compute_fps_weighted` | warmup 后 `sum(frames) / sum(compute_seconds)` | + +`stream_fps`、`chunk_compute_fps` 和 `chunk_compute_fps_weighted` 不得合并。avg、P95、P99 是同一个 canonical +指标的统计曲线，不是三项独立指标。 + +### 5.2 五个核心维度 + +| 维度 | Canonical leaf 示例 | +|---|---| +| 交付 | success rate、frames received、stream FPS | +| 时延 | request、first frame、control ack、control-to-frame | +| 吞吐 | request throughput、weighted compute FPS | +| 目标执行 | pipeline/runtime phase、chunk compute、可选 encode、allocator peak | +| 资源 | CPU、内存、GPU、显存、网络 | + +TeleFuser 与 SGLang 的私有字段先保存为 raw point，再由版本化 mapping 映射到这些 leaf。无法等价的字段保持 +私有或 unavailable，不通过改名制造可比性。 + +## 6. 主动资源上报 + +Resource agent 与 target PID 同机运行： + +1. 注册 run 与 source identity； +2. 每 1 秒采样； +3. 每 15 秒有界批量上报； +4. 任务完成、失败或取消时立即 final flush； +5. 注册、批次或 final flush 未获确认时使启用资源采集的 benchmark 失败。 + +每个点包含 `run_id`、metric、subject、source timestamp、value、unit 和区分设备/网卡/cgroup 的 labels。 + +资源 subject： + +- `process_used`：目标 PID 及其后代； +- `container_used`：能可靠解析的 cgroup charged usage； +- `machine_used`：整机或物理设备使用； +- `machine_total`：整机/设备容量； +- `container_total`：有限容器上限，仅作为容量事实。 + +采集规则： + +- CPU 以一个逻辑核为 100%，多核进程和整机允许超过 100%； +- GPU/显存按物理设备保留 labels，同一 Run、同一 subject 内才允许堆叠； +- Ethernet 使用网卡 byte counter；RDMA 使用 active-port counter； +- 机器网络 counter 不能可靠归因到进程，因此不伪造 `process_used`； +- cgroup v1/v2 无法解析的上限保持 unavailable，不拿整机容量替代； +- 通用 cgroup 没有可移植网络带宽上限，因此不生成容器网络容量。 + +## 7. GreptimeDB 与前端 + +GreptimeDB 是 History 的唯一在线存储。服务启动或建表失败直接失败，查询失败返回 503；不存在 SQLite、 +内存索引或文件直查 fallback。JSON/JSONL 是可重放导入源，不是在线查询后端。 + +部署时由 TeleFuser setup 在固定 `benchmarks/aiperf` checkout 内创建独立的无 dev 依赖运行环境，并创建 +固定 `artifacts` 导入根目录；Vue 构建产物随 AIPerf 提供，API 与前端由同一 History 进程服务，不引入 +第二个 Node.js 运行进程。GreptimeDB 使用独立持久化卷，开发机默认只监听 loopback；远端查看通过 SSH +tunnel 或带认证的反向代理完成。 + +前端使用与指标树相同的五维顺序： + +- 左侧固定树只列 canonical leaf； +- TeleFuser/SGLang 和 avg/P95/P99 作为图中曲线，不拆成树节点； +- 右侧支持左右两组 Run，并按固定维度和 leaf 顺序排列卡片； +- 勾选只控制整项指标的显示/隐藏，不改变卡片顺序； +- 百分比用量允许堆叠，容量使用独立单线小图； +- 连续折线节点只在悬浮时显示；tooltip 位于浏览器 top layer，不能被卡片裁剪； +- 内存、显存和带宽按量级显示 KB/MB/GB/TB 与 KB/s/MB/s/GB/s/TB/s。 + +## 8. 产物与失败语义 + +每个 Run 至少保留： + +- resolved config、contract 和 AIPerf commit； +- summary、session/event JSONL 和 normalized points； +- target metadata 与资源 source identity； +- 成功、失败、超时、OOM 和取消数量； +- 独立 HTML 报告。 + +正式对比只使用 workload 与资源策略一致的成功 Run。OOM、超时和连接失败本身也是结果，不从分母删除； +mock、native fallback 或 offload Run 必须使用不同 qualification，不能伪装成正式配置。 + +## 9. 验证要求 + +提交前至少覆盖： + +- runtime timer、设备去重、finite value 与 allocator fact 单元测试； +- TeleFuser phase/chunk event 及 API 指标透传； +- TeleFuser WebRTC 与 SGLang WebSocket mock contract； +- warmup 排除、weighted FPS 和 canonical mapping； +- resource schema、cgroup、Ethernet/RDMA 与 final flush； +- GreptimeDB 强依赖、幂等导入、API 查询和 Vue production build； +- 一次固定配置的真实 TeleFuser Run；SGLang 公平配置无法完成时保留明确失败证据。 diff --git a/docs/zh/index.md b/docs/zh/index.md index 7177e20..2690f9d 100644 --- a/docs/zh/index.md +++ b/docs/zh/index.md @@ -95,6 +95,7 @@ telefuser stream-serve examples/lingbot/lingbot_world_fast_image_to_video_h100.p 服务指南批量服务、任务 API 和 SDK。 流式服务WebRTC 流式传输和双向控制。 流式调度器Actor 所有权、有界数据流、生命周期、指标和 GPU 卡位。 +AIPerf 基准测试Batch、stream、baseline 与历史指标工作流。 配置运行时、注意力、量化和卸载配置。 并行推理分布式处理策略。 新增模型集成新的模型架构和阶段。 diff --git a/examples/stream_server/webrtc_bidirectional_demo.py b/examples/stream_server/webrtc_bidirectional_demo.py index ee96aa1..bfdd712 100644 --- a/examples/stream_server/webrtc_bidirectional_demo.py +++ b/examples/stream_server/webrtc_bidirectional_demo.py @@ -8,7 +8,7 @@ Usage: # 1. Start the LingBot stream server: - telefuser stream-serve examples/stream_server/stream_lingbot_world_fast.py -p 8088 --skip-validation + telefuser stream-serve examples/lingbot/lingbot_world_fast_image_to_video_h100.py -p 8088 --skip-validation # 2. Start this client (opens browser): python examples/stream_server/webrtc_bidirectional_demo.py --server-url http://localhost:8088 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..a19968e --- /dev/null +++ b/examples/wan_video/wan21_14b_image_to_video_480p_service.py @@ -0,0 +1,188 @@ +from __future__ import annotations + +import os +from typing import TYPE_CHECKING, Any + +import torch +from PIL import Image + +from telefuser.core.config import AttentionConfig, AttnImplType, WeightOffloadType +from telefuser.core.module_manager import ModuleManager +from telefuser.service.core.contract_templates import build_pipeline_manifest, build_task_contract_template +from telefuser.utils.video import save_video + +if TYPE_CHECKING: + from telefuser.pipelines.wan_video.wan21_video import Wan21VideoPipeline + +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"], +) -> Wan21VideoPipeline: + """Build Wan2.1 14B 480P I2V pipeline for TeleFuser service.""" + from telefuser.pipelines.wan_video.wan21_video import ( + Wan21VideoPipeline, + Wan21VideoPipelineConfig, + ) + + 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: Wan21VideoPipeline, + image: Image.Image, + prompt: str, + negative_prompt: str = "", + seed: int = PPL_CONFIG["seed"], + resolution: str = PPL_CONFIG["resolution"], +) -> list[Image.Image]: + """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: Wan21VideoPipeline, + first_image_path: str, + prompt: str, + negative_prompt: str, + seed: int, + output_path: str, + resolution: str = PPL_CONFIG["resolution"], + **kwargs: Any, +) -> dict[str, str]: + """Run the fixed benchmark workload and save the result to file.""" + if not first_image_path: + raise ValueError("run_with_file requires first_image_path") + image = Image.open(first_image_path).convert("RGB") + 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 48c5856..36ebf43 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -120,6 +120,7 @@ plugins: nav_translations: Home: 首页 User Guides: 用户指南 + Benchmarks: 基准测试 Parallel Inference: 并行推理 Developer Guides: 开发者指南 Configuration: 配置 @@ -127,6 +128,7 @@ plugins: Service: 服务指南 Stream Server: 流式服务 Stream Scheduler: 流式调度器 + TeleFuser and AIPerf: TeleFuser 与 AIPerf Service Metadata: 服务元数据 Adding New Example: 新增示例 Model Loading: 模型加载 @@ -162,6 +164,8 @@ nav: - adding_new_example.md - model_loading.md - offload.md + - Benchmarks: + - TeleFuser and AIPerf: benchmark_aiperf.md - Configuration: - configuration.md - Parallel Inference: diff --git a/scripts/setup_aiperf_repo.sh b/scripts/setup_aiperf_repo.sh new file mode 100755 index 0000000..a7a5f4c --- /dev/null +++ b/scripts/setup_aiperf_repo.sh @@ -0,0 +1,96 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +AIPERF_REPO_URL="${AIPERF_REPO_URL:-https://github.com/ActivePeter/aiperf.git}" +AIPERF_DIR="${ROOT_DIR}/benchmarks/aiperf" +AIPERF_BRANCH="${AIPERF_BRANCH:-teleai}" +AIPERF_REF="${AIPERF_REF:-}" +UV_BIN="${AIPERF_UV_BIN:-uv}" +UPDATE_REPO=0 + +usage() { + cat <<'EOF' +Usage: scripts/setup_aiperf_repo.sh [options] + +Clone or update the external AIPerf checkout and install its runtime environment. + +Options: + --repo-url URL Git URL (default: https://github.com/ActivePeter/aiperf.git) + --branch NAME Branch used for a new clone (default: teleai) + --ref REF Branch, tag, or commit to check out after fetching + --update Fast-forward the current branch; refuses a dirty checkout + -h, --help Show this help +EOF +} + +while [[ $# -gt 0 ]]; do + case "$1" in + --repo-url) + AIPERF_REPO_URL="$2" + shift 2 + ;; + --branch) + AIPERF_BRANCH="$2" + shift 2 + ;; + --ref) + AIPERF_REF="$2" + shift 2 + ;; + --update) + UPDATE_REPO=1 + shift + ;; + -h|--help) + usage + exit 0 + ;; + *) + echo "Unknown option: $1" >&2 + usage >&2 + exit 2 + ;; + esac +done + +if ! command -v "${UV_BIN}" >/dev/null 2>&1; then + echo "uv is required to install AIPerf: https://docs.astral.sh/uv/getting-started/installation/" >&2 + exit 1 +fi + +if [[ ! -d "${AIPERF_DIR}/.git" ]]; then + if [[ -e "${AIPERF_DIR}" ]]; then + echo "Path exists and is not an AIPerf checkout: ${AIPERF_DIR}" >&2 + exit 1 + fi + mkdir -p "$(dirname "${AIPERF_DIR}")" + git clone --branch "${AIPERF_BRANCH}" "${AIPERF_REPO_URL}" "${AIPERF_DIR}" +fi + +if [[ "${UPDATE_REPO}" -eq 1 || -n "${AIPERF_REF}" ]]; then + if [[ -n "$(git -C "${AIPERF_DIR}" status --porcelain)" ]]; then + echo "AIPerf checkout has local changes: ${AIPERF_DIR}" >&2 + exit 1 + fi + git -C "${AIPERF_DIR}" fetch origin +fi + +if [[ "${UPDATE_REPO}" -eq 1 ]]; then + current_branch="$(git -C "${AIPERF_DIR}" symbolic-ref --quiet --short HEAD || true)" + if [[ -z "${current_branch}" ]]; then + echo "Cannot fast-forward a detached AIPerf checkout" >&2 + exit 1 + fi + git -C "${AIPERF_DIR}" pull --ff-only origin "${current_branch}" +fi + +if [[ -n "${AIPERF_REF}" ]]; then + git -C "${AIPERF_DIR}" checkout "${AIPERF_REF}" +fi + +"${UV_BIN}" sync --no-dev --project "${AIPERF_DIR}" --extra streaming-webrtc +mkdir -p "${ROOT_DIR}/artifacts" + +echo "AIPerf ready: ${AIPERF_DIR}" +git -C "${AIPERF_DIR}" rev-parse HEAD diff --git a/telefuser/metrics/__init__.py b/telefuser/metrics/__init__.py index 77f4232..5f8940f 100644 --- a/telefuser/metrics/__init__.py +++ b/telefuser/metrics/__init__.py @@ -50,6 +50,14 @@ get_metrics_registry, reset_global_registry, ) +from .runtime import ( + RuntimeMeasurement, + canonical_cuda_devices, + collect_runtime_environment, + finish_runtime_measurement, + start_runtime_measurement, + visible_cuda_devices, +) from .service_metrics import ( ServiceMetrics, get_service_metrics, @@ -90,6 +98,13 @@ "ServiceMetrics", "get_service_metrics", "reset_service_metrics", + # Runtime benchmark facts + "RuntimeMeasurement", + "canonical_cuda_devices", + "visible_cuda_devices", + "start_runtime_measurement", + "finish_runtime_measurement", + "collect_runtime_environment", # Configuration "MetricsConfig", "default_metrics_config", diff --git a/telefuser/metrics/runtime.py b/telefuser/metrics/runtime.py new file mode 100644 index 0000000..8c4d9a9 --- /dev/null +++ b/telefuser/metrics/runtime.py @@ -0,0 +1,139 @@ +"""Lightweight target-side measurements for external benchmark collectors.""" + +from __future__ import annotations + +import platform +import subprocess +import time +from collections.abc import Callable, Sequence +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import torch + + +@dataclass(frozen=True) +class RuntimeMeasurement: + """State needed to finish one synchronized runtime measurement.""" + + started_at_s: float + devices: tuple[str, ...] + capture_peak_memory: bool + clock: Callable[[], float] = field(repr=False, compare=False) + + +def canonical_cuda_devices( + values: Sequence[str | torch.device], +) -> tuple[str, ...]: + """Return unique canonical CUDA device names, ignoring non-CUDA values.""" + + devices: list[str] = [] + for value in values: + device = torch.device(value) + if device.type != "cuda": + continue + index = device.index if device.index is not None else 0 + canonical = f"cuda:{index}" + if canonical not in devices: + devices.append(canonical) + return tuple(devices) + + +def visible_cuda_devices() -> tuple[str, ...]: + """Return every CUDA device visible to the current process.""" + + if not torch.cuda.is_available(): + return () + return tuple(f"cuda:{index}" for index in range(torch.cuda.device_count())) + + +def _synchronize(devices: Sequence[str]) -> None: + if not torch.cuda.is_available(): + return + for device in devices: + torch.cuda.synchronize(device) + + +def start_runtime_measurement( + devices: Sequence[str | torch.device], + *, + capture_peak_memory: bool = False, + clock: Callable[[], float] = time.perf_counter, +) -> RuntimeMeasurement: + """Synchronize devices, optionally reset peaks, and start a phase timer.""" + + canonical = canonical_cuda_devices(devices) + _synchronize(canonical) + if capture_peak_memory and torch.cuda.is_available(): + for device in canonical: + torch.cuda.reset_peak_memory_stats(device) + return RuntimeMeasurement( + started_at_s=clock(), + devices=canonical, + capture_peak_memory=capture_peak_memory, + clock=clock, + ) + + +def finish_runtime_measurement(measurement: RuntimeMeasurement) -> dict[str, Any]: + """Finish a measurement and return AIPerf-compatible phase facts.""" + + _synchronize(measurement.devices) + seconds = max(measurement.clock() - measurement.started_at_s, 0.0) + memory: list[dict[str, int | str]] = [] + if measurement.capture_peak_memory and torch.cuda.is_available(): + memory = [ + { + "device": device, + "peak_allocated_bytes": int(torch.cuda.max_memory_allocated(device)), + "peak_reserved_bytes": int(torch.cuda.max_memory_reserved(device)), + } + for device in measurement.devices + ] + return {"seconds": seconds, "memory": memory} + + +def _git_commit(repo_root: Path | None) -> str | None: + if repo_root is None: + return None + try: + return subprocess.run( + ["git", "-C", str(repo_root), "rev-parse", "HEAD"], + check=True, + capture_output=True, + text=True, + timeout=5, + ).stdout.strip() + except (OSError, subprocess.CalledProcessError, subprocess.TimeoutExpired): + return None + + +def collect_runtime_environment( + devices: Sequence[str | torch.device], + *, + repo_root: Path | None = None, +) -> dict[str, Any]: + """Collect stable software and accelerator identity for benchmark artifacts.""" + + canonical = canonical_cuda_devices(devices) + gpu_info: list[dict[str, Any]] = [] + if torch.cuda.is_available(): + for device in canonical: + properties = torch.cuda.get_device_properties(device) + gpu_info.append( + { + "device": device, + "name": properties.name, + "compute_capability": (f"{properties.major}.{properties.minor}"), + "total_memory_bytes": int(properties.total_memory), + } + ) + return { + "python_version": platform.python_version(), + "platform": platform.platform(), + "torch_version": torch.__version__, + "cuda_version": torch.version.cuda, + "telefuser_git_commit": _git_commit(repo_root), + "gpus": gpu_info, + } diff --git a/telefuser/pipelines/lingbot_world_fast/pipeline.py b/telefuser/pipelines/lingbot_world_fast/pipeline.py index 990f807..388151c 100644 --- a/telefuser/pipelines/lingbot_world_fast/pipeline.py +++ b/telefuser/pipelines/lingbot_world_fast/pipeline.py @@ -513,6 +513,7 @@ def _create_initialized_session( frame_tokens=frame_tokens, chunk_size=session_config.chunk_size, max_attention_size=max_attention_size, + kv_cache_capacity_tokens=kv_size, cache_handle=cache_handle, ) try: diff --git a/telefuser/pipelines/lingbot_world_fast/service.py b/telefuser/pipelines/lingbot_world_fast/service.py index a7860e8..95447d9 100644 --- a/telefuser/pipelines/lingbot_world_fast/service.py +++ b/telefuser/pipelines/lingbot_world_fast/service.py @@ -18,6 +18,7 @@ import torch from PIL import Image, ImageDraw, ImageFont +from telefuser.metrics.runtime import RuntimeMeasurement, finish_runtime_measurement, start_runtime_measurement from telefuser.utils.logging import logger from telefuser.utils.profiler import ProfilingContext4Debug @@ -273,6 +274,7 @@ def create_session(self, config: dict) -> str: ) ), show_control_hud=bool(config.get("show_control_hud", defaults.get("show_control_hud", True))), + benchmark_metrics=bool(config.get("benchmark_metrics", defaults.get("benchmark_metrics", False))), ) control_context = self.pipeline.control_context(session_config) state = LingBotWorldFastSessionState( @@ -845,6 +847,46 @@ def _next_realtime_control( return control_builder.defer(directional_chunk), applied_controls return None + def _benchmark_devices(self) -> tuple[str | torch.device, ...]: + """Return every CUDA-capable pipeline device for benchmark synchronization.""" + devices: list[str | torch.device] = [] + for name in ("device", "text_device", "vae_encode_device", "vae_decode_device", "vae_device"): + device = getattr(self.pipeline, name, None) + if device is not None and device not in devices: + devices.append(device) + parallel_config = getattr(getattr(self.pipeline, "config", None), "parallel_config", None) + for device_id in getattr(parallel_config, "device_ids", None) or (): + device = f"cuda:{int(device_id)}" + if device not in devices: + devices.append(device) + return tuple(devices) + + def _start_benchmark_measurement( + self, + state: LingBotWorldFastSessionState, + ) -> RuntimeMeasurement | None: + if not state.config.benchmark_metrics: + return None + try: + # Generation runs in child actors, while this timer lives in the service + # process. Process-local allocator peaks would therefore be incomplete. + return start_runtime_measurement(self._benchmark_devices(), capture_peak_memory=False) + except Exception as exc: + logger.warning(f"Could not start LingBot benchmark measurement: {exc}") + return None + + @staticmethod + def _finish_benchmark_measurement( + measurement: RuntimeMeasurement | None, + ) -> dict[str, object] | None: + if measurement is None: + return None + try: + return finish_runtime_measurement(measurement) + except Exception as exc: + logger.warning(f"Could not finish LingBot benchmark measurement: {exc}") + return None + def _run_actor_worker_loop( self, state: LingBotWorldFastSessionState, @@ -856,7 +898,11 @@ def _run_actor_worker_loop( first_item = self._next_realtime_control(state, control_context, control_builder, 0, emit_status, block=True) if first_item is None: return - runtime = self.pipeline._create_initialized_session(state.config, progress_callback=emit_status) + runtime_measurement = self._start_benchmark_measurement(state) + try: + runtime = self.pipeline._create_initialized_session(state.config, progress_callback=emit_status) + finally: + runtime_facts = self._finish_benchmark_measurement(runtime_measurement) state.generation_session = runtime if not state.active: return @@ -872,10 +918,13 @@ def _run_actor_worker_loop( latent_frames=runtime.latent_f, total_chunks=runtime.chunk_count, stream_progress=self._stream_progress(state, runtime), + runtime=self._runtime_metadata(runtime), + **({"measurement": {"name": "runtime_creation", **runtime_facts}} if runtime_facts is not None else {}), ) submitted = 0 controls_by_chunk: dict[int, list[str] | None] = {} + measurements_by_chunk: dict[int, RuntimeMeasurement] = {} def raise_scheduler_error() -> None: error = streaming_runtime.error(streaming_session) @@ -887,9 +936,13 @@ def submit_chunk(item: tuple[object, list[str] | None]) -> None: deferred_control, applied_controls = item control = self.pipeline._resolve_control(deferred_control) self.pipeline._validate_control(runtime, control) + chunk_measurement = self._start_benchmark_measurement(state) if not streaming_runtime.try_submit_chunk(streaming_session, submitted, control): + self._finish_benchmark_measurement(chunk_measurement) raise RuntimeError("LingBot streaming ingress became unavailable after capacity check") controls_by_chunk[submitted] = applied_controls + if chunk_measurement is not None: + measurements_by_chunk[submitted] = chunk_measurement emit_status( "generating_chunk", index=submitted, @@ -912,6 +965,7 @@ def submit_chunk(item: tuple[object, list[str] | None]) -> None: if not frames: raise RuntimeError(f"LingBot actor emitted no frames for chunk {result_index}") applied_controls = controls_by_chunk.pop(result_index, None) + chunk_facts = self._finish_benchmark_measurement(measurements_by_chunk.pop(result_index, None)) if state.config.show_control_hud: frames = self._overlay_control_hud(frames, applied_controls) self._put_output( @@ -946,6 +1000,18 @@ def submit_chunk(item: tuple[object, list[str] | None]) -> None: ), runtime_metrics=self._runtime_metrics(state), stream_progress=self._stream_progress(state, runtime), + **( + { + "measurement": { + "index": result_index, + "frames": len(frames), + "compute_seconds": chunk_facts["seconds"], + "memory": chunk_facts["memory"], + } + } + if chunk_facts is not None + else {} + ), ) if runtime.current_chunk_index >= runtime.chunk_count or not state.active: @@ -1038,6 +1104,19 @@ def _run_worker_loop( if self._sessions.get(session_id) is state: self._sessions.pop(session_id, None) + def _runtime_metadata(self, runtime: LingBotWorldFastGenerationSession) -> dict[str, int]: + return { + "height": int(runtime.height), + "width": int(runtime.width), + "latent_frames": int(runtime.latent_f), + "frame_tokens": int(runtime.frame_tokens), + "chunk_size": int(runtime.chunk_size), + "max_attention_size": int(runtime.max_attention_size), + "kv_local_attn_size": int(self.pipeline.dit.local_attn_size), + "kv_sink_size": int(self.pipeline.dit.sink_size), + "kv_cache_capacity_tokens": int(runtime.kv_cache_capacity_tokens), + } + def push_chunk(self, session_id: str, chunk: dict) -> None: state = self._sessions.get(session_id) if state is None or not state.active: diff --git a/telefuser/pipelines/lingbot_world_fast/session.py b/telefuser/pipelines/lingbot_world_fast/session.py index 0060e40..6a3604e 100644 --- a/telefuser/pipelines/lingbot_world_fast/session.py +++ b/telefuser/pipelines/lingbot_world_fast/session.py @@ -63,6 +63,7 @@ class LingBotWorldFastSessionConfig: control_pitch_step_degrees: float = 2.0 control_pitch_limit_degrees: float = 85.0 show_control_hud: bool = True + benchmark_metrics: bool = False class LingBotWorldFastSessionStatus(str, Enum): @@ -88,6 +89,7 @@ class LingBotWorldFastGenerationSession: frame_tokens: int = 0 chunk_size: int = 0 max_attention_size: int = 0 + kv_cache_capacity_tokens: int = 0 cache_handle: int | None = None current_chunk_index: int = 0 emitted_frames: int = 0 diff --git a/telefuser/service/api/openai/adapter.py b/telefuser/service/api/openai/adapter.py index 4b4fc4e..8f73652 100644 --- a/telefuser/service/api/openai/adapter.py +++ b/telefuser/service/api/openai/adapter.py @@ -20,6 +20,7 @@ validate_image_size, ) from telefuser.service.api.schema import TaskRequest +from telefuser.service_types import AspectRatio from telefuser.utils.logging import logger VIDEO_REFERENCE_EXTENSIONS = {".mp4", ".avi", ".mov", ".mkv", ".webm"} @@ -153,6 +154,19 @@ 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 + inferred_aspect_ratio = infer_aspect_ratio(width, height) + if inferred_aspect_ratio in AspectRatio.values(): + aspect_ratio = inferred_aspect_ratio base_fields = { "task": task, @@ -170,8 +184,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/api/schema.py b/telefuser/service/api/schema.py index 58c8dee..fa23bbd 100644 --- a/telefuser/service/api/schema.py +++ b/telefuser/service/api/schema.py @@ -71,6 +71,8 @@ class TaskResponse(BaseModel): task_id: str task_status: TaskStatus output_path: str + peak_memory_mb: float | None = Field(default=None, ge=0, description="Peak accelerator memory used by the task.") + inference_time_s: float | None = Field(default=None, ge=0, description="Task inference duration in seconds.") class StopTaskResponse(BaseModel): diff --git a/telefuser/service/core/pipeline_service.py b/telefuser/service/core/pipeline_service.py index ca7e80f..196f164 100644 --- a/telefuser/service/core/pipeline_service.py +++ b/telefuser/service/core/pipeline_service.py @@ -4,12 +4,15 @@ import asyncio import json +import math import threading +import time from types import ModuleType from typing import Any import torch.multiprocessing as mp +from telefuser.platforms import current_platform from telefuser.service_types import TaskType from telefuser.utils.logging import logger @@ -30,6 +33,74 @@ mp.set_start_method("spawn", force=True) +_BYTES_PER_MIB = 1024 * 1024 + + +def _nonnegative_finite_float(value: Any) -> float | None: + if isinstance(value, bool): + return None + try: + number = float(value) + except (TypeError, ValueError): + return None + return number if math.isfinite(number) and number >= 0 else None + + +def _accelerator_available() -> bool: + try: + return bool(current_platform.is_accelerator_available()) + except Exception: + return False + + +def _synchronize_accelerator() -> None: + if not _accelerator_available(): + return + try: + current_platform.synchronize() + except Exception: + pass + + +def _reset_peak_memory_stats() -> None: + if not _accelerator_available(): + return + try: + _synchronize_accelerator() + current_platform.reset_peak_memory_stats() + except Exception: + pass + + +def _peak_memory_mb() -> float | None: + if not _accelerator_available(): + return None + try: + peak_bytes = current_platform.max_memory_allocated() + except Exception: + return None + peak_memory_bytes = _nonnegative_finite_float(peak_bytes) + return peak_memory_bytes / _BYTES_PER_MIB if peak_memory_bytes is not None else None + + +def _raw_runtime_metrics(raw: Any) -> tuple[float | None, float | None]: + if not isinstance(raw, dict): + return None, None + metrics = raw.get("metrics") if isinstance(raw.get("metrics"), dict) else {} + inference_time_s = _nonnegative_finite_float(raw.get("inference_time_s")) + if inference_time_s is None: + inference_time_s = _nonnegative_finite_float(metrics.get("inference_time_s")) + peak_memory_mb = _nonnegative_finite_float(raw.get("peak_memory_mb")) + if peak_memory_mb is None: + peak_memory_mb = _nonnegative_finite_float(metrics.get("peak_memory_mb")) + if peak_memory_mb is None: + peak_memory_gb = _nonnegative_finite_float(raw.get("peak_memory_gb")) + if peak_memory_gb is None: + peak_memory_gb = _nonnegative_finite_float(metrics.get("peak_memory_gb")) + if peak_memory_gb is not None: + peak_memory_mb = peak_memory_gb * 1024.0 + return inference_time_s, peak_memory_mb + class PipelineService: """Pipeline service with security validation for pipeline configuration files. @@ -195,12 +266,17 @@ async def run_task_with_stop_event( timeout_s = timeout_s if timeout_s is not None else float(self.server_config.task_timeout) + _reset_peak_memory_stats() + inference_started_at = time.perf_counter() result = await self._runner.run( task_data=task_data, stop_event=stop_event, timeout_s=timeout_s, output_root=output_root, ) + _synchronize_accelerator() + measured_inference_time_s = time.perf_counter() - inference_started_at + raw_inference_time_s, raw_peak_memory_mb = _raw_runtime_metrics(result.raw) return { "task_id": task_data.get("task_id", ""), @@ -208,6 +284,10 @@ async def run_task_with_stop_event( "output_path": result.output_path or "", "message": result.message, "raw": result.raw, + "inference_time_s": ( + raw_inference_time_s if raw_inference_time_s is not None else measured_inference_time_s + ), + "peak_memory_mb": raw_peak_memory_mb if raw_peak_memory_mb is not None else _peak_memory_mb(), } def server_metadata(self) -> dict: diff --git a/telefuser/service/core/stream_pipeline_service.py b/telefuser/service/core/stream_pipeline_service.py index 7b6f0b5..0ad318a 100644 --- a/telefuser/service/core/stream_pipeline_service.py +++ b/telefuser/service/core/stream_pipeline_service.py @@ -19,10 +19,19 @@ import asyncio import inspect import threading +import time from collections.abc import AsyncGenerator +from pathlib import Path from types import ModuleType from typing import Any, Protocol, runtime_checkable +from telefuser.metrics.runtime import ( + RuntimeMeasurement, + collect_runtime_environment, + finish_runtime_measurement, + start_runtime_measurement, + visible_cuda_devices, +) from telefuser.utils.logging import logger from ..security.security_validator import ( @@ -44,6 +53,7 @@ STREAM_MODE_SERVER_PUSH = "server_push" STREAM_MODE_BIDIRECTIONAL = "bidirectional" +_REPO_ROOT = Path(__file__).resolve().parents[3] # --------------------------------------------------------------------------- # Service protocols – pipeline authors implement one of these @@ -98,6 +108,8 @@ def __init__( self.ppl_file: str | None = None self._module: ModuleType | None = None self._module_name: str | None = None + self._startup_measurement: dict[str, Any] | None = None + self._runtime_environment: dict[str, Any] = {} self.security_level = security_level or self.server_config.security_level self.security_validator = PipelineSecurityValidator( @@ -118,6 +130,16 @@ def start_service(self, ppl_file: str, skip_validation: bool = False, gpu_num: i logger.warning("Stream service is already running") return True + self._startup_measurement = None + self._runtime_environment = {} + devices = visible_cuda_devices() + started_at = time.perf_counter() + measurement: RuntimeMeasurement | None = None + try: + measurement = start_runtime_measurement(devices, capture_peak_memory=True) + except Exception as exc: + logger.warning(f"Could not initialize stream startup metrics: {exc}") + try: if not skip_validation: validate_pipeline_file( @@ -147,6 +169,20 @@ def start_service(self, ppl_file: str, skip_validation: bool = False, gpu_num: i self.stream_mode = self._detect_mode(self.service) self.service.start() self.is_running = True + try: + facts = ( + finish_runtime_measurement(measurement) + if measurement is not None + else {"seconds": max(time.perf_counter() - started_at, 0.0), "memory": []} + ) + except Exception as exc: + logger.warning(f"Could not finish stream startup metrics: {exc}") + facts = {"seconds": max(time.perf_counter() - started_at, 0.0), "memory": []} + self._startup_measurement = {"name": "pipeline_init", **facts} + try: + self._runtime_environment = collect_runtime_environment(devices, repo_root=_REPO_ROOT) + except Exception as exc: + logger.warning(f"Could not collect stream runtime environment: {exc}") logger.info(f"Stream service started: mode={self.stream_mode}, file={ppl_file}") return True @@ -266,13 +302,20 @@ def has_session(self, session_id: str) -> bool: # -- metadata ------------------------------------------------------------ def server_metadata(self) -> dict[str, Any]: - return { + metadata = { "service_type": "stream", "stream_mode": self.stream_mode, "pipeline_file": self.ppl_file, "security_level": self.security_level.name if self.security_level else "NONE", "runner": "StreamPipelineService", } + startup_measurement = getattr(self, "_startup_measurement", None) + runtime_environment = getattr(self, "_runtime_environment", {}) + if startup_measurement is not None: + metadata["performance"] = {"phases": [startup_measurement]} + if runtime_environment: + metadata["environment"] = runtime_environment + return metadata # -- internals ----------------------------------------------------------- diff --git a/telefuser/service/core/task_manager.py b/telefuser/service/core/task_manager.py index 4163e5b..c04c133 100644 --- a/telefuser/service/core/task_manager.py +++ b/telefuser/service/core/task_manager.py @@ -28,6 +28,8 @@ class TaskInfo: end_time: datetime | None = None error: str | None = None output_path: str | None = None + peak_memory_mb: float | None = None + inference_time_s: float | None = None stop_event: threading.Event = field(default_factory=threading.Event) thread: threading.Thread | None = None @@ -118,7 +120,14 @@ def _transition_to(self, task_id: str, target: TaskStatus, from_statuses: tuple[ self._tasks.move_to_end(task_id) return task - def complete_task(self, task_id: str, output_path: str | None = None) -> None: + def complete_task( + self, + task_id: str, + output_path: str | None = None, + *, + peak_memory_mb: float | None = None, + inference_time_s: float | None = None, + ) -> None: """Mark task as completed with metrics.""" with self._lock: if task_id not in self._tasks: @@ -139,7 +148,9 @@ def complete_task(self, task_id: str, output_path: str | None = None) -> None: if task.start_time: duration = (task.end_time - task.start_time).total_seconds() + task.inference_time_s = inference_time_s if inference_time_s is not None else duration get_service_metrics().record_task_completed(duration) + task.peak_memory_mb = peak_memory_mb def fail_task(self, task_id: str, error: str) -> None: """Mark task as failed with metrics.""" @@ -213,6 +224,8 @@ def get_task_status(self, task_id: str) -> dict[str, Any] | None: "end_time": task.end_time, "error": task.error, "output_path": task.output_path, + "peak_memory_mb": task.peak_memory_mb, + "inference_time_s": task.inference_time_s, } status.update(self._serialize_task_message(task.message)) return status diff --git a/telefuser/service/core/task_processor.py b/telefuser/service/core/task_processor.py index 36798ef..16fe8ec 100644 --- a/telefuser/service/core/task_processor.py +++ b/telefuser/service/core/task_processor.py @@ -142,7 +142,12 @@ async def _process_task(self, task_id: str) -> None: ) if result: - self.task_manager.complete_task(task_id, result.output_path) + self.task_manager.complete_task( + task_id, + result.output_path, + peak_memory_mb=result.peak_memory_mb, + inference_time_s=result.inference_time_s, + ) logger.info(f"Task {task_id} completed successfully") else: if task_info.stop_event.is_set(): diff --git a/telefuser/service/core/task_service.py b/telefuser/service/core/task_service.py index 08c7066..927f4ca 100644 --- a/telefuser/service/core/task_service.py +++ b/telefuser/service/core/task_service.py @@ -182,6 +182,8 @@ async def update_audio_path(audio_name: str, message: TaskRequest, task_data: di task_id=message.task_id, task_status=TaskStatus.COMPLETED, output_path=str(output_path), + peak_memory_mb=result.get("peak_memory_mb"), + inference_time_s=result.get("inference_time_s"), ) else: raise RuntimeError(result.get("message") or "Inference failed") diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 40ba7af..cfe77be 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -117,6 +117,8 @@ def make_stream_svc(service, stream_mode: str): svc.ppl_file = "mock.py" svc._module = None svc._module_name = None + svc._startup_measurement = None + svc._runtime_environment = {} svc.security_level = None svc.security_validator = None return svc diff --git a/tests/unit/openai/test_adapter.py b/tests/unit/openai/test_adapter.py index f0268b8..cbaf4b2 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 dimensions while using a TaskRequest-compatible aspect ratio.""" + 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 == "16:9" + def test_video_to_task_image_to_video(self): """Convert I2V request (auto-detected from input).""" openai_req = VideoGenerationsRequest( diff --git a/tests/unit/openai/test_video_routes.py b/tests/unit/openai/test_video_routes.py index a68add5..cfb2c4d 100644 --- a/tests/unit/openai/test_video_routes.py +++ b/tests/unit/openai/test_video_routes.py @@ -34,6 +34,8 @@ def client(tmp_path): "task_id": "vid_123", "status": TaskStatus.COMPLETED.value, "output_path": str(tmp_path / "video.mp4"), + "peak_memory_mb": 2048.0, + "inference_time_s": 12.5, } ) task_manager.get_task = MagicMock( @@ -143,6 +145,8 @@ def test_get_video_success(self, client): assert response.status_code == 200 data = response.json() assert data["id"] == "vid_123" + assert data["peak_memory_mb"] == 2048.0 + assert data["inference_time_s"] == 12.5 def test_delete_completed_video_reports_completed_status(self, client): """Delete on a terminal task reports the real terminal status.""" diff --git a/tests/unit/pipelines/lingbot_world_fast/test_service_metrics.py b/tests/unit/pipelines/lingbot_world_fast/test_service_metrics.py new file mode 100644 index 0000000..e2c3110 --- /dev/null +++ b/tests/unit/pipelines/lingbot_world_fast/test_service_metrics.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import torch +from PIL import Image + +from telefuser.metrics.runtime import start_runtime_measurement +from telefuser.pipelines.lingbot_world_fast.service import LingBotWorldFastService +from telefuser.pipelines.lingbot_world_fast.session import ( + LingBotWorldFastGenerationSession, + LingBotWorldFastSessionConfig, + LingBotWorldFastSessionState, +) + + +def test_lingbot_actor_service_emits_runtime_and_chunk_measurements() -> None: + config = LingBotWorldFastSessionConfig( + prompt="test", + image=Image.new("RGB", (16, 16)), + chunk_size=1, + frame_num=1, + benchmark_metrics=True, + show_control_hud=False, + ) + runtime = LingBotWorldFastGenerationSession( + config=config, + height=480, + width=832, + latent_f=1, + frame_tokens=1560, + chunk_size=1, + max_attention_size=32760, + kv_cache_capacity_tokens=32760, + cache_handle=7, + ) + pipeline = MagicMock() + pipeline.device = "cpu" + pipeline.text_device = torch.device("cpu") + pipeline.vae_encode_device = torch.device("cpu") + pipeline.vae_decode_device = torch.device("cpu") + pipeline.vae_device = torch.device("cpu") + pipeline.config = SimpleNamespace(parallel_config=SimpleNamespace(device_ids=None)) + pipeline.dit = SimpleNamespace(local_attn_size=18, sink_size=6) + pipeline._create_initialized_session.return_value = runtime + pipeline._resolve_control.return_value = torch.zeros(1) + + streaming_runtime = MagicMock() + streaming_session = SimpleNamespace(session_id="actor-session") + streaming_runtime.create_session.return_value = streaming_session + streaming_runtime.error.return_value = None + streaming_runtime.try_submit_chunk.return_value = True + streaming_runtime.poll_frames.return_value = [(0, [Image.new("RGB", (16, 16)) for _ in range(3)])] + pipeline._get_streaming_runtime.return_value = streaming_runtime + + service = LingBotWorldFastService(pipeline) + state = LingBotWorldFastSessionState(config=config, control_context=SimpleNamespace()) + statuses: list[dict[str, object]] = [] + + def emit_status(stage: str, **data: object) -> None: + statuses.append({"stage": stage, **data}) + + with ( + patch.object(service, "_next_realtime_control", return_value=(object(), None)), + patch.object(service, "_put_output"), + patch( + "telefuser.pipelines.lingbot_world_fast.service.start_runtime_measurement", + wraps=start_runtime_measurement, + ) as start_measurement, + ): + service._run_actor_worker_loop(state, state.control_context, MagicMock(), emit_status) + + runtime_ready = next(item for item in statuses if item.get("stage") == "runtime_ready") + chunk_sent = next(item for item in statuses if item.get("stage") == "chunk_sent") + + assert runtime_ready["measurement"]["name"] == "runtime_creation" + assert runtime_ready["runtime"]["kv_cache_capacity_tokens"] == 32760 + assert runtime_ready["runtime"]["kv_local_attn_size"] == 18 + assert chunk_sent["measurement"]["index"] == 0 + assert chunk_sent["measurement"]["frames"] == 3 + assert chunk_sent["measurement"]["compute_seconds"] >= 0 + assert chunk_sent["measurement"]["memory"] == [] + assert "encode_seconds" not in chunk_sent["measurement"] + assert start_measurement.call_count == 2 + assert all(call.kwargs["capture_peak_memory"] is False for call in start_measurement.call_args_list) diff --git a/tests/unit/pipelines/wan_video/test_service_examples.py b/tests/unit/pipelines/wan_video/test_service_examples.py index 382206b..f8def51 100644 --- a/tests/unit/pipelines/wan_video/test_service_examples.py +++ b/tests/unit/pipelines/wan_video/test_service_examples.py @@ -2,6 +2,9 @@ import importlib.util from pathlib import Path +from unittest.mock import MagicMock, patch + +from PIL import Image def _load_module(path: Path): @@ -25,3 +28,41 @@ def test_wan22_cache_service_imports_without_cacheseek() -> None: module = _load_module(root / "examples/wan_video/wan22_14b_text_to_video_service.py") assert module.CACHE_CONFIG["enable_latent_cache"] is True + + +def test_wan21_i2v_benchmark_service_uses_fixed_workload() -> None: + root = Path(__file__).resolve().parents[4] + module = _load_module(root / "examples/wan_video/wan21_14b_image_to_video_480p_service.py") + pipeline = MagicMock(return_value=[Image.new("RGB", (8, 8))]) + + module.run(pipeline, Image.new("RGB", (8, 8)), "test prompt", seed=7) + + kwargs = pipeline.call_args.kwargs + assert (kwargs["width"], kwargs["height"]) == (832, 480) + assert kwargs["num_frames"] == 81 + assert kwargs["num_inference_steps"] == 40 + assert kwargs["seed"] == 7 + + +def test_wan21_i2v_benchmark_service_writes_requested_output(tmp_path) -> None: + root = Path(__file__).resolve().parents[4] + module = _load_module(root / "examples/wan_video/wan21_14b_image_to_video_480p_service.py") + image_path = tmp_path / "input.png" + output_path = tmp_path / "output.mp4" + Image.new("RGB", (8, 8)).save(image_path) + + with ( + patch.object(module, "run", return_value=[Image.new("RGB", (8, 8))]), + patch.object(module, "save_video") as save_video, + ): + result = module.run_with_file( + MagicMock(), + first_image_path=str(image_path), + prompt="test", + negative_prompt="", + seed=42, + output_path=str(output_path), + ) + + assert result == {"output_path": str(output_path)} + save_video.assert_called_once() diff --git a/tests/unit/service/test_pipeline_service_metrics.py b/tests/unit/service/test_pipeline_service_metrics.py new file mode 100644 index 0000000..6ec41d6 --- /dev/null +++ b/tests/unit/service/test_pipeline_service_metrics.py @@ -0,0 +1,120 @@ +from __future__ import annotations + +import threading +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +import pytest + +from telefuser.service.core.pipeline_runner import PipelineRunResult +from telefuser.service.core.pipeline_service import PipelineService +from telefuser.service_types import PipelineRunStatus + + +@pytest.mark.asyncio +async def test_pipeline_service_exports_measured_runtime_and_peak_memory() -> None: + service = PipelineService.__new__(PipelineService) + service.is_running = True + service.pipeline = object() + service._runner = SimpleNamespace( + run=AsyncMock( + return_value=PipelineRunResult( + status=PipelineRunStatus.SUCCESS, + output_path="result.mp4", + ) + ) + ) + + with ( + patch( + "telefuser.service.core.pipeline_service.current_platform.is_accelerator_available", + return_value=True, + ), + patch("telefuser.service.core.pipeline_service.current_platform.synchronize") as synchronize, + patch("telefuser.service.core.pipeline_service.current_platform.reset_peak_memory_stats") as reset_peak, + patch( + "telefuser.service.core.pipeline_service.current_platform.max_memory_allocated", + return_value=2 * 1024 * 1024 * 1024, + ), + ): + result = await service.run_task_with_stop_event( + {"task_id": "task-1"}, + threading.Event(), + timeout_s=10.0, + ) + + reset_peak.assert_called_once_with() + assert synchronize.call_count == 2 + assert result["inference_time_s"] >= 0 + assert result["peak_memory_mb"] == 2048.0 + + +@pytest.mark.asyncio +async def test_pipeline_service_prefers_pipeline_reported_runtime_metrics() -> None: + service = PipelineService.__new__(PipelineService) + service.is_running = True + service.pipeline = object() + service._runner = SimpleNamespace( + run=AsyncMock( + return_value=PipelineRunResult( + status=PipelineRunStatus.SUCCESS, + output_path="result.mp4", + raw={ + "metrics": { + "inference_time_s": 12.5, + "peak_memory_gb": 3.0, + } + }, + ) + ) + ) + + with ( + patch( + "telefuser.service.core.pipeline_service.current_platform.is_accelerator_available", + return_value=True, + ), + patch("telefuser.service.core.pipeline_service.current_platform.synchronize"), + patch("telefuser.service.core.pipeline_service.current_platform.reset_peak_memory_stats"), + ): + result = await service.run_task_with_stop_event( + {"task_id": "task-1"}, + threading.Event(), + timeout_s=10.0, + ) + + assert result["inference_time_s"] == 12.5 + assert result["peak_memory_mb"] == 3072.0 + + +@pytest.mark.asyncio +async def test_pipeline_service_ignores_invalid_top_level_metrics() -> None: + service = PipelineService.__new__(PipelineService) + service.is_running = True + service.pipeline = object() + service._runner = SimpleNamespace( + run=AsyncMock( + return_value=PipelineRunResult( + status=PipelineRunStatus.SUCCESS, + output_path="result.mp4", + raw={ + "inference_time_s": float("nan"), + "peak_memory_mb": True, + "metrics": {"inference_time_s": 2.5, "peak_memory_mb": 1024.0}, + }, + ) + ) + ) + + with patch( + "telefuser.service.core.pipeline_service.current_platform.is_accelerator_available", + return_value=False, + ): + result = await service.run_task_with_stop_event( + {"task_id": "task-1"}, + threading.Event(), + timeout_s=10.0, + ) + + assert result["inference_time_s"] == 2.5 + assert result["peak_memory_mb"] == 1024.0 diff --git a/tests/unit/service/test_stream_pipeline_metrics.py b/tests/unit/service/test_stream_pipeline_metrics.py new file mode 100644 index 0000000..a7f2b67 --- /dev/null +++ b/tests/unit/service/test_stream_pipeline_metrics.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +from types import SimpleNamespace + +from telefuser.service.core import stream_pipeline_service as stream_module +from telefuser.service.core.stream_pipeline_service import StreamPipelineService + + +class _FakeServerPushService: + def start(self) -> None: + return None + + def stop(self) -> None: + return None + + async def serve(self, request): + if False: + yield request + + +def test_stream_service_metadata_exposes_startup_performance( + monkeypatch, +) -> None: + service = _FakeServerPushService() + module = SimpleNamespace(get_service=lambda: service) + monkeypatch.setattr( + stream_module, + "load_pipeline_module", + lambda *args, **kwargs: (module, "fake_module"), + ) + monkeypatch.setattr(stream_module, "visible_cuda_devices", lambda: ()) + monkeypatch.setattr( + stream_module, + "finish_runtime_measurement", + lambda measurement: {"seconds": 4.0, "memory": []}, + ) + monkeypatch.setattr( + stream_module, + "collect_runtime_environment", + lambda *args, **kwargs: {"torch_version": "test"}, + ) + pipeline_service = StreamPipelineService() + + assert pipeline_service.start_service("fake.py", skip_validation=True) is True + metadata = pipeline_service.server_metadata() + + assert metadata["performance"]["phases"] == [{"name": "pipeline_init", "seconds": 4.0, "memory": []}] + assert metadata["environment"] == {"torch_version": "test"} diff --git a/tests/unit/service/test_task_runtime.py b/tests/unit/service/test_task_runtime.py index 283291f..dd0bda2 100644 --- a/tests/unit/service/test_task_runtime.py +++ b/tests/unit/service/test_task_runtime.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +from datetime import timedelta from types import SimpleNamespace from telefuser.service.api.schema import TaskRequest, TaskResponse @@ -26,6 +27,25 @@ def test_complete_task_does_not_override_cancelled_status() -> None: assert status["output_path"] == message.output_path +def test_complete_task_exports_runtime_metrics() -> None: + task_manager = TaskManager(max_queue_size=10) + task_id = task_manager.create_task(TaskRequest(task="t2v")) + task = task_manager.start_task(task_id) + task.start_time -= timedelta(seconds=2) + + task_manager.complete_task( + task_id, + output_path="result.mp4", + peak_memory_mb=2048.0, + inference_time_s=1.5, + ) + + status = task_manager.get_task_status(task_id) + assert status is not None + assert status["peak_memory_mb"] == 2048.0 + assert status["inference_time_s"] == 1.5 + + def test_artifact_cleanup_snapshot_splits_active_and_terminal_tasks() -> None: task_manager = TaskManager() active_id = task_manager.create_task(SimpleNamespace(output_path="active.mp4")) diff --git a/tests/unit/test_runtime_metrics.py b/tests/unit/test_runtime_metrics.py new file mode 100644 index 0000000..3eb0d74 --- /dev/null +++ b/tests/unit/test_runtime_metrics.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +from pathlib import Path + +import torch + +from telefuser.metrics.runtime import ( + canonical_cuda_devices, + collect_runtime_environment, + finish_runtime_measurement, + start_runtime_measurement, +) + + +def test_runtime_measurement_uses_injected_monotonic_clock() -> None: + ticks = iter([10.0, 12.5]) + measurement = start_runtime_measurement([], clock=lambda: next(ticks)) + + result = finish_runtime_measurement(measurement) + + assert result == {"seconds": 2.5, "memory": []} + + +def test_canonical_cuda_devices_deduplicates_and_ignores_cpu() -> None: + assert canonical_cuda_devices([torch.device("cpu"), "cuda", "cuda:1", "cuda:1"]) == ("cuda:0", "cuda:1") + + +def test_runtime_environment_has_stable_software_identity() -> None: + environment = collect_runtime_environment([], repo_root=Path("/missing")) + + assert environment["python_version"] + assert environment["platform"] + assert environment["torch_version"] + assert environment["gpus"] == []