[PERF] Unify distributed communication for real-time LingBot-World v2

.gitignore

@@ -162,3 +162,4 @@ telefuser/_version.py
 !examples/data/lingbot_world_fast/image.jpg
 !examples/data/lingbot_world_fast/poses.npy
 !examples/data/lingbot_world_fast/intrinsics.npy
+.models

README.md

@@ -17,9 +17,12 @@ TeleFuser is a high-performance runtime for world model inference and multimodal
 
 ## News 📰
 
+- ✨ **2026-08-03**: Validated LingBot-World v2 target-side real-time generation on **4 x H100 80 GB** at
+  832x480 and 16 FPS. The current 77-frame gate reached **17.14 steady compute FPS**; see the
+  [reproducible benchmark](docs/en/benchmark_aiperf.md#current-four-h100-real-time-gate).
 - ✨ **2026-07-27**: Unified streaming on LiveKit with room sessions, retained multi-session admission, LingBot
   chunk-boundary time slicing, reconnect-friendly browser transport, and server-push/bidirectional contracts.
-- ✨ **2026-07-22**: **NEW** Added [**LingBot-Video**](examples/lingbot_video/README.md) support for Dense and MoE T2I/T2V/TI2V generation, native four-GPU CFG/SP execution, and in-memory MoE refinement.
+- ✨ **2026-07-22**: Added [**LingBot-Video**](examples/lingbot_video/README.md) support for Dense and MoE T2I/T2V/TI2V generation, native four-GPU CFG/SP execution, and in-memory MoE refinement.
 - ✨ **2026-07-15**: Added [**LingBot-World v2**](https://github.com/Robbyant/lingbot-world-v2) support for offline generation, interactive WebRTC streaming, and multi-GPU inference.
 
 - ✨ **2026-07-06**: Added external **CacheSeek** latent cache integration for service-mode cross-request reuse. Cache hits can skip the first N denoising steps; the Wan2.2 cache-enabled service example snapshots `[5, 10, 15, 20, 25]` by default. See [docs/en/latent_cache.md](docs/en/latent_cache.md).
@@ -95,6 +98,12 @@ video = pipe(
 TeleFuser streams `LingBot-World v2` through LiveKit. LingBot-World v2 uses camera control and its v2 PPL defaults;
 its streaming example caps a session at two minutes.
 
+The validated four-H100 configuration sustains 17.14 target-side compute FPS for the default 77-frame, 832x480
+request, above its 16 FPS playback target. This is a synchronized pipeline-compute metric; model loading, LiveKit
+encoding, network delivery, and client rendering are measured separately. See the
+[LingBot example guide](examples/lingbot/README.md#validated-four-h100-real-time-gate) for the exact command and
+chunk timings.
+
 LingBot streaming uses the actor-based scheduler for both offline and service execution. Encode, DiT, and decode may
 overlap even on the same GPU; move stages only when memory placement requires it. See the
 [streaming scheduler guide](docs/en/stream_scheduler.md).
@@ -240,6 +249,7 @@ See [examples/README.md](examples/README.md) for the example runner and baseline
 - [docs/en/stream_server.md](docs/en/stream_server.md): LiveKit streaming, session APIs, data topics, and deployment
 - [docs/en/stream_scheduler.md](docs/en/stream_scheduler.md): actor-based stage scheduling, backpressure, lifecycle, metrics, and LingBot placement
 - [docs/en/parallel.md](docs/en/parallel.md): distributed inference architecture
+- [docs/en/communication.md](docs/en/communication.md): collectives, CUDA IPC, synchronization, and transport lifecycle
 - [docs/en/latent_cache.md](docs/en/latent_cache.md): CacheSeek latent cache integration
 - [docs/en/feature_cache.md](docs/en/feature_cache.md): `AdaTaylorCache`
 - [docs/en/model_loading.md](docs/en/model_loading.md): model loading patterns

README_zh.md

@@ -17,9 +17,12 @@ TeleFuser 是一个面向世界模型推理与多模态生成的高性能运行
 
 ## News 📰
 
+- ✨ **2026-08-03**：LingBot-World v2 已在 **4 张 H100 80 GB** 上通过 832x480、16 FPS 的目标侧实时生成
+  验证。当前 77 帧门禁达到 **17.14 steady compute FPS**，复现方法见
+  [基准文档](docs/zh/benchmark_aiperf.md#当前-77-帧实时计算门禁)。
 - ✨ **2026-07-27**：统一使用 LiveKit 流式后端，支持 room 会话、worker 准入、浏览器自动重连，以及
   server-push 和 bidirectional 两种 pipeline contract。
-- ✨ **2026-07-22**：**NEW** 新增 [**LingBot-Video**](examples/lingbot_video/README.md) 支持，覆盖 Dense/MoE T2I、T2V、TI2V、原生四卡 CFG/SP 推理与内存直传 MoE refiner。
+- ✨ **2026-07-22**：新增 [**LingBot-Video**](examples/lingbot_video/README.md) 支持，覆盖 Dense/MoE T2I、T2V、TI2V、原生四卡 CFG/SP 推理与内存直传 MoE refiner。
 - ✨ **2026-07-15**：新增 [**LingBot-World v2**](https://github.com/Robbyant/lingbot-world-v2) 支持，支持离线生成、交互式 WebRTC 流和多卡推理。
 
 - ✨ **2026-07-06**：新增外部 **CacheSeek** latent cache 集成，支持服务模式下跨请求复用；命中后可跳过前 N 步去噪。Wan2.2 服务示例默认快照 `[5, 10, 15, 20, 25]`。配置和安装方式见 [docs/zh/latent_cache.md](docs/zh/latent_cache.md)。
@@ -94,6 +97,10 @@ video = pipe(
 TeleFuser 通过 LiveKit 传输 `LingBot-World v2`。LingBot-World v2 使用相机控制和 v2 PPL 默认值；其流式
 示例将单个会话上限设为两分钟。
 
+已验证的四卡 H100 配置在默认 77 帧、832x480 请求上达到 17.14 target-side compute FPS，高于 16 FPS
+播放目标。该数值是设备同步后的 pipeline 计算指标；模型加载、LiveKit 编码、网络交付和客户端渲染需单独
+衡量。精确命令和逐 chunk 结果见 [LingBot 示例文档](examples/lingbot/README.md#validated-four-h100-real-time-gate)。
+
 LingBot 的离线与服务执行共用 actor scheduler。即使位于同一张 GPU，encode、DiT 和 decode 也可以重叠；
 仅在显存放置需要时移动 Stage。详见[流式调度器指南](docs/zh/stream_scheduler.md)。
 
@@ -232,6 +239,7 @@ telefuser/
 - [docs/zh/stream_server.md](docs/zh/stream_server.md)：LiveKit 流服务、session API、data topic 和部署
 - [docs/zh/stream_scheduler.md](docs/zh/stream_scheduler.md)：基于 actor 的 Stage 调度、backpressure、生命周期、指标和 LingBot 卡位
 - [docs/zh/parallel.md](docs/zh/parallel.md)：分布式推理架构
+- [docs/zh/communication.md](docs/zh/communication.md)：collective、CUDA IPC、同步与传输生命周期
 - [docs/zh/latent_cache.md](docs/zh/latent_cache.md)：CacheSeek latent cache 集成
 - [docs/zh/feature_cache.md](docs/zh/feature_cache.md)：`AdaTaylorCache`
 - [docs/zh/model_loading.md](docs/zh/model_loading.md)：模型加载方式

benchmarks/telefuser_aiperf/README.md

@@ -143,9 +143,9 @@ In terminal 3, run the one-minute LingBot-World v2 workload:
 bash benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh
 ```
 
-The v2 workload is the launcher default and the only checked-in stream workload. It requests 59.75 seconds of media
-using the model's fixed attention window. Its AIPerf active window is 240 seconds, so a successful command normally
-takes about four minutes rather than one minute. Do not terminate it after media generation becomes quiet.
+The v2 workload is the default TeleFuser stream workload. It requests 59.75 seconds of media using the model's fixed
+attention window. The 240-second AIPerf active window is a timeout ceiling; a successful run exits after the target
+emits its completion status and normally takes about 66 seconds after admission, excluding model loading.
 
 A successful run prints `Stream profile sessions: 1/1 succeeded`, an artifact directory, and an HTML report path.
 Results are written below:
@@ -162,9 +162,53 @@ session, joins its room, receives native video tracks, sends reliable controls o
 `tf.status` and bounded `tf.metrics` messages. It produces AIPerf's standard session results without requiring any
 LiveKit-specific changes in AIPerf.
 
+## SGLang LingBot-World v2, Four GPUs
+
+This target uses SGLang's native MessagePack WebSocket endpoint and does not start LiveKit. It requires an SGLang
+installation containing the `/v1/realtime_video/generate` endpoint and
+`LingBotWorldCausalDMDPipeline`. In terminal 1, launch the four-GPU server:
+
+```bash
+bash benchmarks/telefuser_aiperf/scripts/run_sglang_lingbot_world_v2_4gpu.sh
+```
+
+The launcher defaults to GPUs `0,1,2,3`, port `30000`, and model
+`robbyant/lingbot-world-v2-14b-causal-fast-diffusers`. Explicit `SGLANG_SOURCE_DIR` and `SGLANG_PYTHON` values take
+precedence over an installed `sglang` command. Override the defaults when needed:
+
+```bash
+SGLANG_BIN=/path/to/sglang \
+SGLANG_PYTHON=/path/to/sglang-env/bin/python \
+SGLANG_SOURCE_DIR=/path/to/sglang-source \
+SGLANG_MODEL_PATH=/path/to/lingbot-world-v2-14b-causal-fast-diffusers \
+SGLANG_CUDA_VISIBLE_DEVICES=4,5,6,7 \
+  bash benchmarks/telefuser_aiperf/scripts/run_sglang_lingbot_world_v2_4gpu.sh
+```
+
+Wait until `http://127.0.0.1:30000/health` succeeds, then run AIPerf in terminal 2:
+
+```bash
+bash benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh \
+  benchmarks/telefuser_aiperf/configs/stream_sglang_lingbot_world_v2_4gpu_1min.json
+```
+
+Artifacts are written below
+`artifacts/telefuser_aiperf/stream_sglang_lingbot_v2_4gpu_1min/`. The adapter counts combined and split SGLang frame
+batches, converts the shared keyboard trace to `camera_actions` state events, and maps scheduler, WebP encoding,
+pacing, and WebSocket write timings into AIPerf's standard stream result.
+
+The launcher intentionally passes `--flow-shift 10`, matching the official LingBot-World v2 implementation and the
+TeleFuser workload. The SGLang source default is `5`; use `SGLANG_FLOW_SHIFT=5` only to benchmark SGLang's default
+behavior, and do not compare that run as a numerically equivalent model configuration. The workload also fixes four
+DMD steps, 16 FPS, 60 chunks, a KV window of 18 latent frames plus a six-frame sink, WebP quality 95, and no output
+pacing. SGLang permits one active realtime generation session, so the contract fixes concurrency to one.
+
 ## Troubleshooting
 
 - `The pinned streaming-capable AIPerf or LiveKit is not installed`: rerun `bash scripts/setup_aiperf.sh`.
+- `SGLang is not installed or SGLANG_BIN is invalid`: activate the SGLang environment or set `SGLANG_BIN` to its
+  executable.
+- SGLang connection refused on port 30000: wait for model loading to finish and check `/health`.
 - Connection refused on port 8088: the TeleFuser process is still warming up or has exited; inspect terminal 2.
 - `0/1 succeeded` with zero received frames: confirm LiveKit is still running, restart the TeleFuser service, wait for
   one idle worker, and rerun the benchmark.
@@ -204,7 +248,7 @@ tree. History failures do not silently fall back to an in-memory or file-only da
 configs/             Reproducible batch and streaming workloads
 data/                Prompt and control inputs
 scripts/             Batch and streaming launchers
-telefuser_aiperf/    Source-loaded LiveKit adapter
+telefuser_aiperf/    Source-loaded LiveKit and SGLang realtime adapters
 tests/               Adapter tests
 *_contract.yaml      Target and transport capability contracts
 ```
@@ -219,9 +263,11 @@ AIPerf environment first, then run the checks from the repository root:
 
 PYTHONPATH=benchmarks/telefuser_aiperf \
   .venv-aiperf/bin/python -m pytest \
-  benchmarks/telefuser_aiperf/tests/test_livekit_adapter.py
+  benchmarks/telefuser_aiperf/tests/test_livekit_adapter.py \
+  benchmarks/telefuser_aiperf/tests/test_sglang_adapter.py
 
 bash -n \
   scripts/setup_aiperf.sh \
-  benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh
+  benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh \
+  benchmarks/telefuser_aiperf/scripts/run_sglang_lingbot_world_v2_4gpu.sh
 ```

benchmarks/telefuser_aiperf/configs/stream_sglang_lingbot_world_v2_4gpu_1min.json

@@ -0,0 +1,39 @@
+{
+  "contract": "benchmarks/telefuser_aiperf/sglang_lingbot_world_v2_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/lingbot_world_fast/image.jpg",
+  "fps": 16,
+  "session_count": 1,
+  "warmup_sessions": 0,
+  "warmup_chunks": 1,
+  "session_duration_s": 240.0,
+  "stagger_s": 0.0,
+  "control_trace_path": "benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json",
+  "request_extra": {
+    "size": "832x480",
+    "num_frames": 957,
+    "max_chunks": 60,
+    "num_inference_steps": 4,
+    "guidance_scale": 1.0,
+    "seed": 42,
+    "realtime_output_format": "webp",
+    "output_compression": 95,
+    "realtime_output_pacing": false,
+    "realtime_causal_sink_size": 6,
+    "realtime_causal_kv_cache_num_frames": 18
+  },
+  "transport": {
+    "connect_timeout_s": 60.0,
+    "message_timeout_s": 300.0,
+    "frame_timeout_s": 300.0,
+    "shutdown_timeout_s": 10.0,
+    "receive_audio": false
+  },
+  "server_metrics": {
+    "enabled": false
+  },
+  "artifacts_dir": "artifacts/telefuser_aiperf/stream_sglang_lingbot_v2_4gpu_1min"
+}

benchmarks/telefuser_aiperf/scripts/run_sglang_lingbot_world_v2_4gpu.sh

@@ -0,0 +1,56 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)"
+cd "${ROOT_DIR}"
+
+SGLANG_BIN="${SGLANG_BIN:-}"
+SGLANG_SOURCE_DIR="${SGLANG_SOURCE_DIR:-${ROOT_DIR}/work_dirs/sglang}"
+SGLANG_PYTHON="${SGLANG_PYTHON:-}"
+SGLANG_MODEL_PATH="${SGLANG_MODEL_PATH:-robbyant/lingbot-world-v2-14b-causal-fast-diffusers}"
+SGLANG_HOST="${SGLANG_HOST:-127.0.0.1}"
+SGLANG_PORT="${SGLANG_PORT:-30000}"
+SGLANG_CUDA_VISIBLE_DEVICES="${SGLANG_CUDA_VISIBLE_DEVICES:-0,1,2,3}"
+SGLANG_FLOW_SHIFT="${SGLANG_FLOW_SHIFT:-10}"
+
+if [[ -n "${SGLANG_BIN}" ]]; then
+    if ! command -v "${SGLANG_BIN}" >/dev/null 2>&1; then
+        echo "SGLANG_BIN is not executable: ${SGLANG_BIN}" >&2
+        exit 1
+    fi
+    sglang_command=("${SGLANG_BIN}")
+elif [[ -n "${SGLANG_PYTHON}" && -x "${SGLANG_PYTHON}" \
+    && -f "${SGLANG_SOURCE_DIR}/python/sglang/cli/main.py" ]]; then
+    export PYTHONPATH="${SGLANG_SOURCE_DIR}/python${PYTHONPATH:+:${PYTHONPATH}}"
+    sglang_command=("${SGLANG_PYTHON}" "-c" "from sglang.cli.main import main; main()")
+elif command -v sglang >/dev/null 2>&1; then
+    sglang_command=("$(command -v sglang)")
+else
+    echo "SGLang is unavailable. Set SGLANG_SOURCE_DIR and SGLANG_PYTHON, or set SGLANG_BIN." >&2
+    exit 1
+fi
+
+IFS=',' read -r -a gpu_ids <<< "${SGLANG_CUDA_VISIBLE_DEVICES}"
+if [[ ${#gpu_ids[@]} -ne 4 ]]; then
+    echo "SGLANG_CUDA_VISIBLE_DEVICES must contain exactly four comma-separated GPU IDs." >&2
+    exit 2
+fi
+
+export CUDA_VISIBLE_DEVICES="${SGLANG_CUDA_VISIBLE_DEVICES}"
+export SGLANG_LINGBOT_LAZY_VAE_ENCODE_BLACK_FRAMES="${SGLANG_LINGBOT_LAZY_VAE_ENCODE_BLACK_FRAMES:-60}"
+
+exec "${sglang_command[@]}" serve \
+    --model-type diffusion \
+    --model-path "${SGLANG_MODEL_PATH}" \
+    --pipeline-class-name LingBotWorldCausalDMDPipeline \
+    --host "${SGLANG_HOST}" \
+    --port "${SGLANG_PORT}" \
+    --num-gpus 4 \
+    --ulysses-degree 4 \
+    --flow-shift "${SGLANG_FLOW_SHIFT}" \
+    --dit-cpu-offload false \
+    --text-encoder-cpu-offload false \
+    --vae-config.use-parallel-decode true \
+    --vae-config.parallel-decode-mode spatial \
+    --enable-torch-compile false \
+    "$@"

benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh

@@ -20,8 +20,8 @@ else
 fi
 if [[ -z "${AIPERF_PYTHON}" ]] || ! command -v "${AIPERF_PYTHON}" >/dev/null 2>&1 \
     || ! PYTHONPATH="${ADAPTER_ROOT}${PYTHONPATH:+:${PYTHONPATH}}" \
-        "${AIPERF_PYTHON}" -c 'import livekit, telefuser_aiperf' >/dev/null 2>&1; then
-    echo "The pinned streaming-capable AIPerf or LiveKit is not installed. Run: bash scripts/setup_aiperf.sh" >&2
+        "${AIPERF_PYTHON}" -c 'import livekit, msgspec, websockets, telefuser_aiperf' >/dev/null 2>&1; then
+    echo "The pinned streaming-capable AIPerf dependencies are not installed. Run: bash scripts/setup_aiperf.sh" >&2
     exit 1
 fi
 

benchmarks/telefuser_aiperf/sglang_lingbot_world_v2_benchmark_contract.yaml

@@ -0,0 +1,66 @@
+# Benchmark contract for SGLang's LingBot-World v2 realtime WebSocket target.
+contract_version: v1
+name: sglang_lingbot_world_v2_realtime
+mode: stream_world
+implementation: sglang
+model_family: lingbot_world_v2
+model: LingBot-World-v2-14B-Causal-Fast
+supported_tasks:
+  - bidirectional
+transport: websocket
+adapter: sglang_realtime
+transport_provider: sglang
+endpoint:
+  health_path: /health
+  offer_path: /v1/realtime_video/generate
+request_encoding:
+  content_type: application/msgpack
+  init_type: init
+  control_type: event
+  control_kind: camera_actions
+result_delivery:
+  media: websocket_frame_batch
+  metrics: websocket_chunk_stats
+  event_log: events/{phase}_{logical_session_index}_{session_id}.jsonl
+workload:
+  mode: bidirectional
+  task: bidirectional
+  fps: 16
+  session_count: 1
+  warmup_sessions: 0
+  session_duration_s: 240.0
+  control_trace: benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json
+  request_extra:
+    size: 832x480
+    num_frames: 957
+    max_chunks: 60
+    num_inference_steps: 4
+    guidance_scale: 1.0
+    seed: 42
+    realtime_output_format: webp
+    output_compression: 95
+    realtime_output_pacing: false
+    realtime_causal_sink_size: 6
+    realtime_causal_kv_cache_num_frames: 18
+metrics:
+  - 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
+  - chunk_compute_seconds
+  - chunk_compute_fps
+  - chunk_encode_seconds
+  - chunk_output_write_seconds
+  - success_rate
+limits:
+  active_sessions: 1
+artifacts:
+  config: benchmarks/telefuser_aiperf/configs/stream_sglang_lingbot_world_v2_4gpu_1min.json
+  control_trace: benchmarks/telefuser_aiperf/data/stream_lingbot_controls.json
+  server_runner: benchmarks/telefuser_aiperf/scripts/run_sglang_lingbot_world_v2_4gpu.sh
+  benchmark_runner: benchmarks/telefuser_aiperf/scripts/run_stream_bench.sh

benchmarks/telefuser_aiperf/telefuser_aiperf/__init__.py

@@ -5,6 +5,7 @@
 from aiperf.streaming.adapters import register_stream_adapter
 
 from telefuser_aiperf.adapter import TeleFuserLiveKitAdapter
+from telefuser_aiperf.sglang_adapter import SGLangRealtimeAdapter
 
 
 def register_adapters(*, replace: bool = False) -> None:
@@ -15,6 +16,11 @@ def register_adapters(*, replace: bool = False) -> None:
         TeleFuserLiveKitAdapter,
         replace=replace,
     )
+    register_stream_adapter(
+        "sglang_realtime",
+        SGLangRealtimeAdapter,
+        replace=replace,
+    )
 
 
-__all__ = ["TeleFuserLiveKitAdapter", "register_adapters"]
+__all__ = ["SGLangRealtimeAdapter", "TeleFuserLiveKitAdapter", "register_adapters"]