add dp_proxy for vLLM DP, write kv_cache_analysis.md, fix DP bench port

experiments/dsv4_h200_max_context_length/kv_cache_analysis.md

@@ -0,0 +1,100 @@
+# DeepSeek-V4-Flash 长上下文显存分析
+
+> 针对 `dsv4_h200_max_context_length` 实验结果（8×H200，TP=8，output_len=1，concurrency=1）的显存复盘。
+
+## 1. 关键模型参数
+
+| 参数 | 数值 |
+|---|---|
+| 总参数量 | ~158B（safetensors 汇总） |
+| Checkpoint 大小 | ~149 GiB（fp8 权重 + scale） |
+| 层数 `num_hidden_layers` | 43 |
+| Hidden size | 4096 |
+| Attention heads | 64 |
+| KV heads | 1 |
+| 每层 KV `head_dim` | 512 |
+| KV cache dtype | fp8 |
+| 每层压缩比 | 2 层不压缩，21 层 4:1，20 层 128:1 |
+| `max_position_embeddings` | **1,048,576** |
+| H200 单卡显存 | 143 GiB |
+
+## 2. 理论显存估算
+
+### 2.1 权重
+
+```
+149 GiB / 8 (TP) ≈ 18.6 GiB / GPU
+```
+
+vLLM 实际启动日志：`Model loading took 20.07 GiB memory`，与理论值基本吻合（包含少量加载开销）。
+
+### 2.2 KV cache
+
+按压缩后的平均每层 KV 大小估算：
+
+- 不压缩的 2 层：512 bytes/token
+- 4:1 压缩的 21 层：128 bytes/token
+- 128:1 压缩的 20 层：4 bytes/token
+
+```
+平均每层 = (2×512 + 21×128 + 20×4) / 43 ≈ 88 bytes/token
+43 层 ≈ 3.8 KiB/token
+```
+
+再考虑 indexer、alignment 等辅助缓存，vLLM 实际日志给出的数字是：
+
+```
+Available KV cache memory: 101.39 GiB
+GPU KV cache size:         14,668,151 tokens
+=> 约 7.0 KiB / token / GPU
+```
+
+因此 1M 上下文的 KV cache 约为：
+
+```
+1,048,576 × 7.0 KiB ≈ 7.0 GiB / GPU
+```
+
+### 2.3 1M 上下文单卡总占用
+
+| 组件 | 估算 |
+|---|---|
+| 权重 | ~20 GiB |
+| KV cache（1M tokens） | ~7 GiB |
+| CUDA graph / 启动开销 | ~1 GiB |
+| 激活 / 临时 buffer | 数 GiB（prefill 峰值） |
+| **合计** | **~30 GiB / GPU** |
+
+vLLM 可用显存按 `--gpu-memory-utilization 0.90` 计算：
+
+```
+143 GiB × 0.9 ≈ 128.7 GiB
+```
+
+**显存完全够用。**
+
+## 3. 与实测结果对比
+
+| 实测 target_len | max_model_len | 结果 |
+|---|---|---|
+| 917,504 | 918,528 | ✅ 通过 |
+| 1,048,576 | 1,049,600 | ❌ server failed to start |
+
+vLLM 1M 失败日志的关键信息：
+
+```text
+User-specified max_model_len (1049600) is greater than the derived max_model_len
+(max_position_embeddings=1048576 or model_max_length=None in model's config.json).
+```
+
+这说明：
+
+1. **917,504 通过是合理的**：此时 KV cache 约 `6.1 GiB/GPU`，加上权重和开销远不到显存上限。
+2. **1,048,576 失败是合理的**：实验里 `max_model_len = input_len + 1024 pad = 1,049,600`，超过了模型 config 声明的 `max_position_embeddings=1,048,576`。**这是位置编码/配置上限，不是显存上限**。
+3. 实验报告的最大长度 **917,504 是偏保守的**：理论上在 output_len=1 时，最大 input length 可以接近 `1,048,575`（只要 `max_model_len ≤ 1,048,576`）。实验因为采用了固定步长并加了 1024 的 pad，没有测到这个边界。
+
+## 4. 结论
+
+- **从显存角度看，1M 上下文在 8×H200 上完全可以放下**：单卡约 30 GiB，仅占可用显存的 1/4 左右。
+- **真正限制到不了 1M 的是 `max_position_embeddings=1048576`**，而不是显存。
+- `dsv4_h200_max_context_length` 的结果与理论估算一致；测出的 917,504 是“该实验配置下能跑通的最大值”，但不是“硬件/显存能支持的最大值”。

experiments/dsv4_h200_vllm_tp_dp_matrix/config.env

@@ -12,6 +12,9 @@ SERVED_MODEL_NAME="deepseek-v4-flash"
 VLLM_PORT="${VLLM_PORT:-30030}"
 # vLLM 0.24.0 multi-port DP supervisor listens on this hard-coded port.
 VLLM_DP_SUPERVISOR_PORT="${VLLM_DP_SUPERVISOR_PORT:-9256}"
+# The bench client does NOT talk to the supervisor directly; we put a
+# round-robin proxy in front of the DP child servers.
+VLLM_DP_PROXY_PORT="${VLLM_DP_PROXY_PORT:-30040}"
 
 VENV_VLLM="${VENV_VLLM:-/data/user1/yy/envs/vllm}"
 VENV_SGLANG="${VENV_SGLANG:-/data/user1/yy/envs/sglang}"

experiments/dsv4_h200_vllm_tp_dp_matrix/run_bench.sh

@@ -32,7 +32,8 @@ log "matrix_mode=${MATRIX_MODE} matrix_file=${MATRIX_FILE}"
 # Helpers
 # ---------------------------------------------------------------------------
 
-health_port_for() {
+# Port used by run_bench to check the vLLM server (supervisor for DP>1).
+server_health_port() {
   local dp="$1"
   if [[ "$dp" -gt 1 ]]; then
     echo "$VLLM_DP_SUPERVISOR_PORT"
@@ -41,16 +42,35 @@ health_port_for() {
   fi
 }
 
+# Port used by the benchmark client (proxy for DP>1).
+bench_port() {
+  local dp="$1"
+  if [[ "$dp" -gt 1 ]]; then
+    echo "$VLLM_DP_PROXY_PORT"
+  else
+    echo "$VLLM_PORT"
+  fi
+}
+
 is_server_healthy() {
   local port="$1"
   curl --fail --silent --show-error --max-time 5 "http://127.0.0.1:${port}/health" >/dev/null 2>&1
 }
 
+is_proxy_healthy() {
+  local port="$1"
+  # The bench client uses /v1/models as the ready signal, so the proxy must
+  # forward it successfully.
+  curl --fail --silent --show-error --max-time 5 "http://127.0.0.1:${port}/v1/models" >/dev/null 2>&1
+}
+
 stop_server() {
   local tp="$1"
   local dp="$2"
-  local pid_file="/data/user1/yy/${EXPERIMENT}_vllm_tp${tp}_dp${dp}.pid"
 
+  stop_proxy "$tp" "$dp"
+
+  local pid_file="/data/user1/yy/${EXPERIMENT}_vllm_tp${tp}_dp${dp}.pid"
   if [[ -f "$pid_file" ]]; then
     local pid
     pid="$(cat "$pid_file")"
@@ -69,6 +89,74 @@ stop_server() {
   sleep 2
 }
 
+start_proxy() {
+  local tp="$1"
+  local dp="$2"
+  if [[ "$dp" -le 1 ]]; then
+    return 0
+  fi
+
+  local proxy_port="$VLLM_DP_PROXY_PORT"
+  local base_port="$VLLM_PORT"
+  local backends=""
+  for ((r = 0; r < dp; r++)); do
+    backends="${backends:+,}$((base_port + r))"
+  done
+
+  local pid_file="/data/user1/yy/${EXPERIMENT}_proxy_tp${tp}_dp${dp}.pid"
+  rm -f "$pid_file"
+
+  log "starting DP proxy on port ${proxy_port} -> backends ${backends}"
+  nohup "${VENV_VLLM}/bin/python" "${SCRIPT_DIR}/../../scripts/common/dp_proxy.py" \
+    --port "$proxy_port" \
+    --backends "$backends" \
+    --timeout 300 \
+    > "${log_dir_global}/vllm_tp${tp}_dp${dp}.proxy.log" 2>&1 &
+
+  local pid=$!
+  echo $pid > "$pid_file"
+
+  for i in $(seq 1 60); do
+    if is_proxy_healthy "$proxy_port"; then
+      log "DP proxy is ready on port ${proxy_port}"
+      return 0
+    fi
+    if ! kill -0 "$pid" 2>/dev/null; then
+      log "ERROR: DP proxy exited early"
+      tail -100 "${log_dir_global}/vllm_tp${tp}_dp${dp}.proxy.log"
+      return 1
+    fi
+    sleep 1
+  done
+
+  log "ERROR: DP proxy not healthy after 60s"
+  tail -100 "${log_dir_global}/vllm_tp${tp}_dp${dp}.proxy.log"
+  return 1
+}
+
+stop_proxy() {
+  local tp="$1"
+  local dp="$2"
+  if [[ "$dp" -le 1 ]]; then
+    return 0
+  fi
+
+  local pid_file="/data/user1/yy/${EXPERIMENT}_proxy_tp${tp}_dp${dp}.pid"
+  if [[ -f "$pid_file" ]]; then
+    local pid
+    pid="$(cat "$pid_file")"
+    if kill -0 "$pid" 2>/dev/null; then
+      log "stopping DP proxy pid=${pid}"
+      kill "$pid" 2>/dev/null || true
+      sleep 2
+      kill -9 "$pid" 2>/dev/null || true
+    fi
+    rm -f "$pid_file"
+  fi
+  pkill -9 -f "dp_proxy.py.*--port ${VLLM_DP_PROXY_PORT}" 2>/dev/null || true
+  sleep 1
+}
+
 build_server_args() {
   local tp="$1"
   local dp="$2"
@@ -101,12 +189,19 @@ start_server() {
     >> "${log_dir_global}/vllm_tp${tp}_dp${dp}.server.outer.log" 2>&1
 
   local hport
-  hport="$(health_port_for "$dp")"
+  hport="$(server_health_port "$dp")"
   if ! is_server_healthy "$hport"; then
     log "error: vllm server tp=${tp} dp=${dp} failed health check on port ${hport}"
     return 1
   fi
   log "vllm server tp=${tp} dp=${dp} is healthy on port ${hport}"
+
+  if [[ "$dp" -gt 1 ]]; then
+    if ! start_proxy "$tp" "$dp"; then
+      log "error: failed to start DP proxy for tp=${tp} dp=${dp}"
+      return 1
+    fi
+  fi
 }
 
 run_warmup() {
@@ -114,14 +209,14 @@ run_warmup() {
   local dp="$2"
   local input_len="$3"
   local output_len="$4"
-  local hport
+  local bport
-  hport="$(health_port_for "$dp")"
+  bport="$(bench_port "$dp")"
 
   log "warming up tp=${tp} dp=${dp} (input=${input_len}, output=${output_len}, num=1)"
   "$PYTHON" "${SCRIPT_DIR}/../../scripts/common/warmup.py" \
     --backend vllm \
     --host 127.0.0.1 \
-    --port "$hport" \
+    --port "$bport" \
     --input-len "$input_len" \
     --output-len "$output_len" \
     --num 1 \
@@ -298,7 +393,7 @@ run_parallel_config() {
       "$PYTHON" -m sglang.bench_serving \
         --backend vllm \
         --host 127.0.0.1 \
-        --port "$(health_port_for "$dp")" \
+        --port "$(bench_port "$dp")" \
         --dataset-name random \
         --random-input-len "$isl" \
         --random-output-len "$dsl" \

scripts/common/dp_proxy.py

@@ -0,0 +1,173 @@
+#!/usr/bin/env python3
+"""Lightweight round-robin reverse proxy for vLLM data-parallel child servers.
+
+vLLM's --data-parallel-multi-port-external-lb only exposes aggregated /health on
+a supervisor port; it does not forward /v1/models, /v1/completions, etc.  This
+proxy sits in front of the child API servers and round-robins requests across
+them.
+
+Usage:
+    python3 dp_proxy.py --port 30040 --backends 30030,30031,30032,30033
+"""
+import argparse
+import asyncio
+import itertools
+from contextlib import asynccontextmanager
+
+import httpx
+from fastapi import FastAPI, Request, Response
+from fastapi.responses import StreamingResponse
+from uvicorn import Config, Server
+
+
+HOP_BY_HOP_HEADERS = {
+    "connection",
+    "keep-alive",
+    "proxy-authenticate",
+    "proxy-authorization",
+    "te",
+    "trailers",
+    "transfer-encoding",
+    "upgrade",
+    "content-encoding",
+    "content-length",
+}
+
+
+class DPProxy:
+    def __init__(self, backends: list[int], timeout: float = 60.0):
+        self.backend_urls = [f"http://127.0.0.1:{port}" for port in backends]
+        self.cycle = itertools.cycle(self.backend_urls)
+        self.lock = asyncio.Lock()
+        self.timeout = timeout
+        self.client = httpx.AsyncClient(
+            timeout=httpx.Timeout(timeout),
+            limits=httpx.Limits(max_connections=100, max_keepalive_connections=20),
+        )
+
+    async def close(self):
+        await self.client.aclose()
+
+    async def _next_backend(self) -> str:
+        async with self.lock:
+            return next(self.cycle)
+
+    async def _is_healthy(self, base_url: str) -> bool:
+        try:
+            r = await self.client.get(f"{base_url}/health", timeout=2.0)
+            return r.status_code == 200
+        except Exception:
+            return False
+
+    async def health(self) -> Response:
+        results = await asyncio.gather(
+            *(self._is_healthy(url) for url in self.backend_urls)
+        )
+        if all(results):
+            return Response(status_code=200)
+        return Response(status_code=503)
+
+    async def ready(self) -> Response:
+        # Bench clients use /v1/models as the ready signal.  We forward that
+        # to a healthy backend; this endpoint just checks /health.
+        return await self.health()
+
+    async def proxy(self, request: Request) -> Response:
+        path = request.url.path
+        if request.url.query:
+            path = f"{path}?{request.url.query}"
+
+        body = await request.body()
+        headers = dict(request.headers)
+        headers.pop("host", None)
+
+        tried = set()
+        while len(tried) < len(self.backend_urls):
+            base_url = await self._next_backend()
+            if base_url in tried:
+                # Avoid infinite loop when only one backend remains.
+                break
+            tried.add(base_url)
+
+            target = f"{base_url}{path}"
+            try:
+                backend_req = self.client.build_request(
+                    method=request.method,
+                    url=target,
+                    headers=headers,
+                    content=body,
+                )
+                backend_resp = await self.client.send(
+                    backend_req, follow_redirects=False, stream=True
+                )
+            except Exception:
+                continue
+
+            if backend_resp.status_code >= 500:
+                await backend_resp.aclose()
+                continue
+
+            response_headers = {
+                k: v
+                for k, v in backend_resp.headers.items()
+                if k.lower() not in HOP_BY_HOP_HEADERS
+            }
+            return StreamingResponse(
+                backend_resp.aiter_bytes(),
+                status_code=backend_resp.status_code,
+                headers=response_headers,
+                background=None,
+            )
+
+        return Response(
+            content=b"All DP backends are unavailable",
+            status_code=503,
+        )
+
+
+def make_app(proxy: DPProxy) -> FastAPI:
+    app = FastAPI()
+
+    @app.get("/health", include_in_schema=False)
+    async def health():
+        return await proxy.health()
+
+    @app.get("/ready", include_in_schema=False)
+    @app.get("/readyz", include_in_schema=False)
+    async def ready():
+        return await proxy.ready()
+
+    @app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"])
+    async def catch_all(request: Request):
+        return await proxy.proxy(request)
+
+    return app
+
+
+async def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--port", type=int, required=True)
+    parser.add_argument(
+        "--backends",
+        type=lambda s: [int(x.strip()) for x in s.split(",")],
+        required=True,
+        help="Comma-separated list of child vLLM server ports",
+    )
+    parser.add_argument("--timeout", type=float, default=300.0)
+    parser.add_argument("--log-level", default="warning")
+    args = parser.parse_args()
+
+    proxy = DPProxy(args.backends, timeout=args.timeout)
+    app = make_app(proxy)
+
+    config = Config(app, host="0.0.0.0", port=args.port, log_level=args.log_level)
+    server = Server(config)
+
+    try:
+        await server.serve()
+    finally:
+        await proxy.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())