[Feature] Support Qwen3-TTS prefill-decode (PD) disaggregation

examples/online_serving/qwen3_tts_pd/benchmark/bench_pd.sh

@@ -0,0 +1,166 @@
+#!/usr/bin/env bash
+# Usage:
+#   bash bench_pd.sh                                  # 默认: 1p1d, c=8, 50 prompts
+#   bash bench_pd.sh 1p1d                             # 指定 topology tag (用于结果目录命名)
+#   bash bench_pd.sh 1p1d 8                           # topology, concurrency
+#   bash bench_pd.sh 1p1d 8 50                        # topology, concurrency, num_prompts
+#   bash bench_pd.sh 1p1d 8 50 zh                     # +locale (zh/en)
+#   bash bench_pd.sh 1p1d 8 2 zh smoke                # smoke 模式：不算 warmup 不存盘
+#
+# Special:
+#   bash bench_pd.sh 1p1d sweep                       # 自动扫 c=1,8,16,32,64,128，每档 50 prompts
+#
+# 每档压测跑完后，会自动生成一条“内容自描述”的验证语音，用来听感校验正确性，
+# 例如 3p1d + c=8 会合成朗读 “这是3个编码1个解码并行数为8的测试语音”，
+# 存到对应结果目录下的 check_audio_c<并发>.wav。设 CHECK_AUDIO=0 可关闭。
+#
+# Env override:
+#   MODEL=/root/models/Qwen3-TTS-12Hz-0.6B-Base
+#   HOST=127.0.0.1
+#   PORT=8091
+#   DATASET=/root/datasets/seed-tts-eval
+#   RESULT_ROOT=<this_dir>/results
+#   CHECK_AUDIO=1                                     # 1=压测后生成校验语音(默认), 0=关闭
+
+set -e
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+
+# ---------- 默认参数 ----------
+TAG="${1:-1p1d}"
+ARG2="${2:-8}"
+NUM_PROMPTS="${3:-50}"
+LOCALE="${4:-zh}"
+MODE="${5:-}"
+
+MODEL="${MODEL:-/root/models/Qwen3-TTS-12Hz-0.6B-Base}"
+HOST="${HOST:-127.0.0.1}"
+PORT="${PORT:-8091}"
+DATASET="${DATASET:-/root/datasets/seed-tts-eval}"
+RESULT_ROOT="${RESULT_ROOT:-$SCRIPT_DIR/results}"
+CHECK_AUDIO="${CHECK_AUDIO:-1}"
+
+# 从 topology tag 里解析 “编码(prefill)/解码(decode)” 个数，用于生成自描述文本。
+# 支持形如 1p3d / 3p1d / 1p1d / ar_3p1d / custom_xxx_1p3d 等；解析不到则回退 1/1。
+parse_pd_counts() {
+    local tag="$1"
+    if [[ "$tag" =~ ([0-9]+)p([0-9]+)d ]]; then
+        P_CNT="${BASH_REMATCH[1]}"
+        D_CNT="${BASH_REMATCH[2]}"
+    else
+        P_CNT=1
+        D_CNT=1
+    fi
+}
+
+# 生成一条“内容自描述”的验证语音：朗读 “这是X个编码Y个解码并行数为Z的测试语音”。
+# 便于人工听感校验：一听内容就知道对应哪套拓扑 + 并发。
+gen_check_audio() {
+    local concurrency="$1"
+    local result_dir="$2"
+    [[ "$CHECK_AUDIO" == "1" ]] || return 0
+
+    parse_pd_counts "$TAG"
+    local text="这是${P_CNT}个编码${D_CNT}个解码并行数为${concurrency}的测试语音"
+    local out_wav="$result_dir/check_audio_c${concurrency}.wav"
+
+    echo "==> [check] 生成校验语音: \"$text\" -> $out_wav"
+    if HOST="$HOST" PORT="$PORT" MODEL="$MODEL" SEED_TTS_ROOT="$DATASET" \
+        INPUT_TEXT="$text" \
+        bash "$SCRIPT_DIR/gen_audio.sh" "$LOCALE" 0 "$out_wav"; then
+        echo "==> [check] ✅ 校验语音已保存: $out_wav"
+    else
+        echo "==> [check] !! 校验语音生成失败 (不影响压测结果)"
+    fi
+}
+
+# ---------- 清代理 ----------
+unset http_proxy https_proxy HTTP_PROXY HTTPS_PROXY all_proxy ALL_PROXY
+export no_proxy="localhost,127.0.0.1,::1"
+export NO_PROXY="localhost,127.0.0.1,::1"
+
+# ---------- 单次 bench 的封装 ----------
+run_one() {
+    local concurrency="$1"
+    local n_prompts="$2"
+    local mode="$3"   # "" or "smoke"
+    local result_dir="$RESULT_ROOT/${TAG}_${LOCALE}_c${concurrency}"
+    mkdir -p "$result_dir"
+
+    local warmups=2
+    local extra_save=(--save-result --result-dir "$result_dir")
+    if [[ "$mode" == "smoke" ]]; then
+        warmups=0
+        extra_save=()   # smoke 不存盘
+    fi
+
+    echo "==> bench: tag=$TAG locale=$LOCALE concurrency=$concurrency num_prompts=$n_prompts warmups=$warmups mode=${mode:-normal}"
+    echo "==> result_dir=$result_dir"
+
+    vllm bench serve --omni \
+        --host "$HOST" --port "$PORT" \
+        --model "$MODEL" \
+        --backend openai-audio-speech \
+        --endpoint /v1/audio/speech \
+        --dataset-name seed-tts \
+        --dataset-path "$DATASET" \
+        --seed-tts-locale "$LOCALE" \
+        --num-prompts "$n_prompts" --num-warmups "$warmups" \
+        --extra-body '{"task_type":"Base"}' \
+        --max-concurrency "$concurrency" --request-rate inf \
+        --percentile-metrics ttft,e2el,audio_rtf,audio_ttfp,audio_duration,audio_underrun \
+        --trust-remote-code \
+        "${extra_save[@]}"
+
+    # 压测后做语音正确性校验（smoke 模式跳过）
+    if [[ "$mode" != "smoke" ]]; then
+        gen_check_audio "$concurrency" "$result_dir"
+    fi
+}
+
+# ---------- 健康检查 ----------
+check_server_alive() {
+    # /health
+    if ! curl -sS -m 3 --noproxy '*' "http://${HOST}:${PORT}/health" >/dev/null 2>&1; then
+        return 1
+    fi
+    # /v1/models（更全面，能检测 stage 是否真的就绪）
+    if ! curl -sS -m 5 --noproxy '*' "http://${HOST}:${PORT}/v1/models" >/dev/null 2>&1; then
+        return 1
+    fi
+    # 检查 pid 文件对应进程是否存活
+    for pid_file in "$RESULT_ROOT"/server_*.pid /tmp/server_*.pid; do
+        if [[ -f "$pid_file" ]]; then
+            local pid
+            pid=$(cat "$pid_file" 2>/dev/null)
+            if [[ -n "$pid" ]] && kill -0 "$pid" 2>/dev/null; then
+                return 0
+            fi
+        fi
+    done
+    # 没 pid 文件但 /health 通也算 OK（可能用户手动启动）
+    return 0
+}
+
+if ! check_server_alive; then
+    echo "!! server at ${HOST}:${PORT} not ready or already crashed. Start it first:"
+    echo "   bash $SCRIPT_DIR/start_pd.sh $TAG"
+    echo "!! 看 server 日志定位原因:"
+    echo "   tail -100 $RESULT_ROOT/server_${TAG}.log"
+    exit 1
+fi
+
+# ---------- 分发：sweep / 单次 ----------
+if [[ "$ARG2" == "sweep" ]]; then
+    echo "==> SWEEP mode: c in {1, 8, 16, 32, 64, 128}, num_prompts=$NUM_PROMPTS each"
+    for c in 1 8 16 32 64 128; do
+        run_one "$c" "$NUM_PROMPTS" ""
+        echo
+    done
+    echo "==> ✅ sweep done. Results under: $RESULT_ROOT/${TAG}_${LOCALE}_c*/"
+    ls -d "$RESULT_ROOT"/${TAG}_${LOCALE}_c*/ 2>/dev/null || true
+else
+    run_one "$ARG2" "$NUM_PROMPTS" "$MODE"
+    echo
+    echo "==> ✅ done"
+fi

examples/online_serving/qwen3_tts_pd/benchmark/catch_overgen.py

@@ -0,0 +1,113 @@
+#!/usr/bin/env python3
+"""Catcher for the PD never-stop / over-generation bug.
+
+Sends seed-tts prompts concurrently to /v1/audio/speech, parses each returned
+WAV's duration, and flags requests whose audio is wildly disproportionate to
+the input text (the codec-EOS-miss signature: model keeps babbling past where
+it should stop). Loops rounds until an outlier is caught. Prints the offending
+utt/text and a wall-clock timestamp so the server log can be grepped.
+"""
+import asyncio, base64, io, json, os, sys, time, wave, datetime
+
+HOST = os.environ.get("HOST", "127.0.0.1")
+PORT = int(os.environ.get("PORT", "8091"))
+MODEL = os.environ.get("MODEL", "/root/models/Qwen3-TTS-12Hz-0.6B-Base")
+ROOT = os.environ.get("SEED_TTS_ROOT", "/root/datasets/seed-tts-eval")
+LOCALE = os.environ.get("LOCALE", "zh")
+CONC = int(os.environ.get("CONC", "64"))
+ROUNDS = int(os.environ.get("ROUNDS", "20"))
+# Over-generation flag: absolute seconds, and seconds-per-input-char ratio.
+# Normal zh seed-tts ≈ 4-8s for 10-30 chars → ~0.3s/char. Flag >>that.
+ABS_S = float(os.environ.get("ABS_S", "20"))
+PER_CHAR_S = float(os.environ.get("PER_CHAR_S", "1.2"))
+
+import aiohttp
+
+def load_rows():
+    meta = os.path.join(ROOT, LOCALE, "meta.lst")
+    rows = []
+    with open(meta) as f:
+        for line in f:
+            line = line.strip()
+            if not line or line.startswith("#"):
+                continue
+            parts = line.split("|")
+            if len(parts) < 4:
+                continue
+            utt, ref_text, wav_rel, target = parts[0], parts[1], parts[2], parts[3]
+            wav = os.path.join(ROOT, LOCALE, wav_rel)
+            if os.path.isfile(wav):
+                rows.append((utt, ref_text, wav, target))
+    return rows
+
+def wav_duration(b: bytes):
+    try:
+        with wave.open(io.BytesIO(b), "rb") as w:
+            return w.getnframes() / float(w.getframerate() or 1)
+    except Exception:
+        return None
+
+async def one(session, sem, row):
+    utt, ref_text, wav, target = row
+    with open(wav, "rb") as f:
+        b64 = base64.b64encode(f.read()).decode("ascii")
+    payload = {
+        "model": MODEL, "input": target, "task_type": "Base",
+        "language": "Chinese" if LOCALE == "zh" else "English",
+        "ref_audio": f"data:audio/wav;base64,{b64}", "ref_text": ref_text,
+    }
+    async with sem:
+        t0 = time.time()
+        try:
+            async with session.post(f"http://{HOST}:{PORT}/v1/audio/speech",
+                                    json=payload) as r:
+                data = await r.read()
+        except Exception as e:
+            return {"utt": utt, "err": str(e), "target": target}
+        dt = time.time() - t0
+    dur = wav_duration(data)
+    return {"utt": utt, "target": target, "chars": len(target),
+            "dur": dur, "lat": dt, "bytes": len(data),
+            "is_wav": data[:4] == b"RIFF", "t0": t0}
+
+async def run_round(rows):
+    sem = asyncio.Semaphore(CONC)
+    timeout = aiohttp.ClientTimeout(total=3600)
+    async with aiohttp.ClientSession(timeout=timeout) as session:
+        results = await asyncio.gather(*[one(session, sem, r) for r in rows])
+    return results
+
+def analyze(rn, results):
+    flagged = []
+    durs = [x["dur"] for x in results if x.get("dur")]
+    maxd = max(durs) if durs else 0
+    med = sorted(durs)[len(durs)//2] if durs else 0
+    nbad = sum(1 for x in results if not x.get("is_wav"))
+    for x in results:
+        d = x.get("dur")
+        if d is None:
+            continue
+        if d >= ABS_S or (x["chars"] > 0 and d / x["chars"] >= PER_CHAR_S and d > 8):
+            flagged.append(x)
+    ts = datetime.datetime.now().strftime("%H:%M:%S")
+    print(f"[{ts}] round {rn}: n={len(results)} non_wav={nbad} "
+          f"dur median={med:.1f}s max={maxd:.1f}s flagged={len(flagged)}", flush=True)
+    for x in flagged:
+        tt = datetime.datetime.fromtimestamp(x["t0"]).strftime("%H:%M:%S")
+        print(f"    !!! CAUGHT utt={x['utt']} chars={x['chars']} dur={x['dur']:.1f}s "
+              f"lat={x['lat']:.1f}s sent_at={tt} text={x['target'][:40]!r}", flush=True)
+    return flagged
+
+async def main():
+    rows = load_rows()
+    print(f"loaded {len(rows)} {LOCALE} rows; conc={CONC} rounds={ROUNDS} "
+          f"flag: abs>={ABS_S}s or >={PER_CHAR_S}s/char", flush=True)
+    any_flag = []
+    for rn in range(1, ROUNDS + 1):
+        results = await run_round(rows)
+        f = analyze(rn, results)
+        any_flag += f
+    print(f"DONE total_flagged={len(any_flag)}", flush=True)
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/online_serving/qwen3_tts_pd/benchmark/catch_with_monitor.sh

@@ -0,0 +1,60 @@
+#!/usr/bin/env bash
+# Runs the over-generation catcher and an external freeze monitor. When the
+# pipeline wedges (server log stops advancing AND both GPUs idle while the
+# catcher is still running), dump the freeze scene: last PD_TRACE lines, the
+# resume-vs-submit gap, and per-req submit/decode status for the stuck reqs.
+set -u
+cd /root/vllm-omni/examples/online_serving/qwen3_tts_pd/benchmark
+LOG=results/server_pd_1p1d.log
+DUMP=results/catch/freeze_dump.txt
+mkdir -p results/catch
+: > "$DUMP"
+
+unset http_proxy https_proxy HTTP_PROXY HTTPS_PROXY all_proxy ALL_PROXY
+export no_proxy=localhost,127.0.0.1,::1 NO_PROXY=localhost,127.0.0.1,::1
+
+# launch catcher (many rounds, c=64, zh)
+LOCALE=zh CONC=64 ROUNDS=40 ABS_S=20 python3 catch_overgen.py > results/catch/catch2.log 2>&1 &
+CATCH_PID=$!
+echo "catcher pid=$CATCH_PID"
+
+strip() { sed -E 's/\x1b\[[0-9;]*m//g'; }
+
+last_log_mtime() { stat -c %Y "$LOG" 2>/dev/null || echo 0; }
+
+FROZEN=0
+while kill -0 "$CATCH_PID" 2>/dev/null; do
+    sleep 15
+    now=$(date +%s)
+    lm=$(last_log_mtime)
+    age=$(( now - lm ))
+    util0=$(nvidia-smi --query-gpu=utilization.gpu --format=csv,noheader,nounits -i 0 2>/dev/null | tr -d ' ')
+    util1=$(nvidia-smi --query-gpu=utilization.gpu --format=csv,noheader,nounits -i 1 2>/dev/null | tr -d ' ')
+    echo "[mon $(date +%T)] log_age=${age}s gpu0=${util0}% gpu1=${util1}%"
+    # freeze = log stale >45s and both GPUs idle, catcher still running
+    if [ "$age" -ge 45 ] && [ "${util0:-0}" -le 2 ] && [ "${util1:-0}" -le 2 ]; then
+        FROZEN=1
+        echo "===== FREEZE DETECTED $(date +%T) log_age=${age}s =====" | tee -a "$DUMP"
+        echo "--- tail 40 server log ---" >> "$DUMP"
+        tail -40 "$LOG" | strip | cut -c1-200 >> "$DUMP"
+        echo "--- global counts ---" >> "$DUMP"
+        echo "prefill_submit_ready=$(grep -c qwen3_tts_prefill_submit_ready $LOG)" >> "$DUMP"
+        echo "prefill_to_decode_submitted=$(grep -c prefill_to_decode_submitted $LOG)" >> "$DUMP"
+        echo "decode_resume=$(grep -c qwen3_tts_decode_resume $LOG)" >> "$DUMP"
+        echo "--- LAST prefill_submit_ready ---" >> "$DUMP"
+        grep qwen3_tts_prefill_submit_ready $LOG | strip | tail -3 | cut -c1-200 >> "$DUMP"
+        echo "--- LAST prefill_to_decode_submitted ---" >> "$DUMP"
+        grep prefill_to_decode_submitted $LOG | strip | tail -3 | cut -c1-200 >> "$DUMP"
+        echo "--- LAST decode_resume ---" >> "$DUMP"
+        grep qwen3_tts_decode_resume $LOG | strip | tail -3 | cut -c1-200 >> "$DUMP"
+        echo "--- any recent orchestrator/forward/error lines (last 60 non-route) ---" >> "$DUMP"
+        tail -400 "$LOG" | strip | grep -iE "orchestrator|forward|pick|submit|error|exception|timed out|abort|prewarm" | tail -40 | cut -c1-200 >> "$DUMP"
+        echo "wrote $DUMP"
+        # let it sit 30s more to confirm permanent, then stop catcher
+        sleep 30
+        kill "$CATCH_PID" 2>/dev/null
+        break
+    fi
+done
+echo "===== monitor done frozen=$FROZEN ====="
+echo "--- catcher tail ---"; tail -15 results/catch/catch2.log