From 59dbadb1106354b076c61fb4b6787a015fe2dcc2 Mon Sep 17 00:00:00 2001 From: yy-fighting Date: Thu, 9 Jul 2026 10:19:30 +0000 Subject: [PATCH] add dp_proxy for vLLM DP, write kv_cache_analysis.md, fix DP bench port --- .../kv_cache_analysis.md | 100 ++++++++++ .../dsv4_h200_vllm_tp_dp_matrix/config.env | 3 + .../dsv4_h200_vllm_tp_dp_matrix/run_bench.sh | 109 ++++++++++- scripts/common/dp_proxy.py | 173 ++++++++++++++++++ 4 files changed, 378 insertions(+), 7 deletions(-) create mode 100644 experiments/dsv4_h200_max_context_length/kv_cache_analysis.md create mode 100644 scripts/common/dp_proxy.py diff --git a/experiments/dsv4_h200_max_context_length/kv_cache_analysis.md b/experiments/dsv4_h200_max_context_length/kv_cache_analysis.md new file mode 100644 index 0000000..8b31fcb --- /dev/null +++ b/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 是“该实验配置下能跑通的最大值”，但不是“硬件/显存能支持的最大值”。 diff --git a/experiments/dsv4_h200_vllm_tp_dp_matrix/config.env b/experiments/dsv4_h200_vllm_tp_dp_matrix/config.env index 51401bc..a02e497 100644 --- a/experiments/dsv4_h200_vllm_tp_dp_matrix/config.env +++ b/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}" diff --git a/experiments/dsv4_h200_vllm_tp_dp_matrix/run_bench.sh b/experiments/dsv4_h200_vllm_tp_dp_matrix/run_bench.sh index 6e3ed6e..99aa94b 100755 --- a/experiments/dsv4_h200_vllm_tp_dp_matrix/run_bench.sh +++ b/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 - hport="$(health_port_for "$dp")" + local bport + 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" \ diff --git a/scripts/common/dp_proxy.py b/scripts/common/dp_proxy.py new file mode 100644 index 0000000..ff69bb1 --- /dev/null +++ b/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())