qqtang/sskj
1
2
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add custom benchmark setup for vLLM TP=2 on dsv4_h200

Browse Source 
- Create configuration file for dsv4_h200_vllm_tp2_custom_bench.
- Implement result parsing script to handle JSONL outputs from bench_client.py.
- Develop run_bench.sh script to orchestrate multi-service vLLM benchmarking.
- Add start_server.sh script to launch independent vLLM services on specified GPU pairs.
This commit is contained in:
Quantong Qiu 2026-07-10 09:06:10 +00:00
parent e62ab7c970
commit db00c1b1ac
6 changed files with 1555 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

271
experiments/dsv4_h200_vllm_tp2_custom_bench/README.md Normal file
View File

@ -0,0 +1,271 @@
# dsv4_h200_vllm_tp2_custom_bench
## 目的
当 vLLM 使用 **TP=2**（Tensor Parallel = 2）部署时，单张 H200 上可以同时运行 **4 个独立服务**（8 卡 / 2 = 4 服务），每个服务独占 2 张 GPU：
| 服务 | 端口 | GPU 对 |
|------|------|--------|
| 1 | 30005 | 0, 1 |
| 2 | 30006 | 2, 3 |
| 3 | 30007 | 4, 5 |
| 4 | 30008 | 6, 7 |
此时 `sglang.bench_serving` 存在以下问题：
1. **低并发下无法测出真实吞吐**：并发数不大时，请求可能只落在一个服务上，其他 3 个服务空闲，bench 报告的吞吐只是单服务的 1/4。
2. **TTFT 测量不够精确**：`sglang.bench_serving` 的计时粒度较粗，对于低延迟场景（TP=2 下 TTFT 可能 < 100ms）误差较大。
3. **不支持多服务负载均衡**：只能测一个端口，无法反映整个 8 卡集群性能。
因此本实验使用 **自定义压测客户端 `bench_client.py`**，直接调用 OpenAI `/v1/chat/completions` API，通过 `aiohttp` 并发发送请求，并记录每个请求的精确时间戳。同时通过 **负载均衡** 将请求均匀分发到 4 个服务上。
## 与 sglang.bench_serving 的区别
| 特性 | sglang.bench_serving | bench_client.py (本实验) |
|------|----------------------|--------------------------|
| 协议 | 内部协议 | OpenAI API (`/v1/chat/completions`) |
| 并发控制 | 进程级 | `asyncio.Semaphore` |
| TTFT 测量 | 粗略 | 精确到 `time.perf_counter()` |
| TPOT 测量 | 基于总时间估算 | 基于首 token 后每个 token 的间隔 |
| 多服务支持 | ❌ 单端口 | ✅ 轮询/随机分发到多个端口 |
| 低并发精度 | 一般 | 高（适合 TP=2 场景） |
| 集群吞吐 | 只能测单服务 | 可测 4 服务总吞吐 |
## 文件说明
| 文件 | 说明 |
|------|------|
| `bench_client.py` | 自定义异步压测客户端，支持多服务负载均衡 |
| `config.env` | 实验配置（场景、模型路径、端口列表、LB 策略等） |
| `run_bench.sh` | 编排脚本：启动 4 个服务 → 运行压测 → 解析结果 |
| `start_server.sh` | 同时启动 4 个 vLLM TP=2 服务（每服务 2 GPU） |
| `parse_results.py` | 解析 bench_client.py 输出的 JSONL，生成 `results.json` + `report.md` |
## 快速运行
```bash
# 完整流程（自动启动 4 个服务、压测、生成报告）
bash experiments/dsv4_h200_vllm_tp2_custom_bench/run_bench.sh
# 跳过服务管理（已有 4 个服务在运行）
SKIP_MANAGE_SERVER=1 bash experiments/dsv4_h200_vllm_tp2_custom_bench/run_bench.sh
# 单独运行压测客户端（多服务，带健康检查）
python3 experiments/dsv4_h200_vllm_tp2_custom_bench/bench_client.py \
--host 127.0.0.1 \
--ports 30005,30006,30007,30008 \
--model deepseek-v4-flash \
--concurrency 32 --input-len 512 --output-len 256 --num-prompts 640 \
--lb-strategy round_robin \
--health-check-interval 5 \
--health-max-failures 2 \
--health-recovery-interval 10 \
--request-max-retries 2 \
--output-file /tmp/test.jsonl
# 单独运行压测客户端（单服务，向后兼容）
python3 experiments/dsv4_h200_vllm_tp2_custom_bench/bench_client.py \
--host 127.0.0.1 --port 30005 \
--model deepseek-v4-flash \
--concurrency 8 --input-len 512 --output-len 256 --num-prompts 160 \
--output-file /tmp/test_single.jsonl
# 解析已有结果
python3 experiments/dsv4_h200_vllm_tp2_custom_bench/parse_results.py \
experiments/dsv4_h200_vllm_tp2_custom_bench/results/<RUN_ID>
```
## 场景设计
本实验覆盖了从 **单并发** 到 **128 并发** 的梯度，以及不同输入/输出长度组合：
| 并发 | 输入长度 | 输出长度 | 请求数 | 说明 |
|------|----------|----------|--------|------|
| 1 | 512 | 256 | 20 | 单并发基线，测纯延迟 |
| 4 | 512 | 256 | 80 | 低并发，4 服务各 1 req |
| 8 | 512 | 256 | 160 | 中低并发 |
| 16 | 512 | 256 | 320 | 中等并发 |
| 32 | 512 | 256 | 640 | 中高并发 |
| 64 | 512 | 256 | 1280 | 高并发 |
| 128 | 512 | 256 | 2560 | 极限并发，测集群吞吐上限 |
| 1 | 2048 | 512 | 20 | 长输入基线 |
| 8 | 2048 | 512 | 160 | 长输入中并发 |
| 32 | 2048 | 512 | 640 | 长输入高并发 |
| 128 | 2048 | 512 | 2560 | 长输入极限并发 |
| 1 | 4096 | 1024 | 20 | 超长输入基线 |
| 8 | 4096 | 1024 | 160 | 超长输入中并发 |
| 32 | 4096 | 1024 | 640 | 超长输入高并发 |
| 128 | 4096 | 1024 | 2560 | 超长输入极限并发 |
## 健康检查与故障自动跳过
`bench_client.py` 内置了 **后台健康检查 + 请求级重试** 机制，确保某个服务挂了不会导致整个 benchmark 失败。
### 健康检查机制
| 参数 | 默认值 | 说明 |
|------|--------|------|
| `--health-check-interval` | 5.0 | 每 N 秒检查一次所有服务 |
| `--health-max-failures` | 2 | 连续失败 N 次才标记为不健康 |
| `--health-recovery-interval` | 10.0 | 不健康服务 N 秒后自动重试恢复 |
| `--request-max-retries` | 2 | 单个请求失败最多重试 N 次 |
### 故障处理流程
```
请求 → 负载均衡选择端口 → 发送请求
↓
成功 → 记录结果
↓
失败 → 换另一个端口重试（最多 --request-max-retries 次）
↓
全部失败 → 记录失败，继续下一个请求
```
后台同时运行健康检查：
- 每 5 秒对所有端口执行 `/health` 探测
- 连续 2 次失败的服务被标记为 **unhealthy**，不再接收新请求
- 10 秒后自动尝试恢复，如果恢复成功则重新加入负载均衡池
### 输出中的故障信息
汇总 JSON 包含以下字段：
```json
{
"completed": 638,
"failed": 2,
"retried": 5,
"healthy_ports_at_end": [30005, 30006, 30007],
"per_port_stats": {
"30005": {"count": 160, "failed": 0, ...},
"30006": {"count": 159, "failed": 1, ...},
"30007": {"count": 160, "failed": 0, ...},
"30008": {"count": 159, "failed": 1, ...}
}
}
```
- `failed`: 最终失败的请求数（重试后仍失败）
- `retried`: 至少重试过一次的请求数
- `healthy_ports_at_end`: 压测结束时仍健康的服务端口
- `per_port_stats[].failed`: 每个端口的失败数
### 场景间健康检查
`run_bench.sh` 在每个场景结束后会打印一次服务健康状态：
```
[2026-07-10T08:45:00+00:00] service health check: healthy=[30005 30006 30007 30008] unhealthy=[]
[2026-07-10T08:45:30+00:00] service health check: healthy=[30005 30006 30007] unhealthy=[30008]
[2026-07-10T08:45:30+00:00] WARNING: 1 service(s) unhealthy: 30008
```
这样即使某个服务在压测中途 OOM 崩溃，其他服务仍能继续工作，benchmark 不会中断。
### 自定义故障参数
```bash
python3 bench_client.py \
--ports 30005,30006,30007,30008 \
--health-check-interval 3 \
--health-max-failures 1 \
--health-recovery-interval 5 \
--request-max-retries 3 \
...
```
- 更敏感：间隔 3 秒、1 次失败即标记不健康、5 秒恢复
- 更容错：单个请求最多重试 3 次
## 负载均衡策略
`bench_client.py` 支持三种负载均衡策略：
| 策略 | 说明 | 适用场景 |
|------|------|----------|
| `round_robin` | 请求按顺序轮询分配到各端口 | **默认**，最公平，推荐 |
| `random` | 随机选择一个端口 | 简单，分布可能不均 |
| `least_conn` | 选择当前连接数最少的端口 | 需要追踪 in-flight，当前为 random 占位 |
通过 `--lb-strategy` 参数选择：
```bash
python3 bench_client.py --lb-strategy round_robin --ports 30005,30006,30007,30008 ...
```
## 输出格式
### JSONL 原始数据 (`raw_outputs/vllm_*.jsonl`)
每行一个 JSON 对象，包含请求分发到的目标端口：
```json
{
"request_id": 0,
"input_tokens": 512,
"output_tokens": 256,
"first_token_time_ms": 45.23,
"e2e_latency_ms": 5234.56,
"inter_token_latencies": [18.5, 19.2, ...],
"tpot_ms": 19.8,
"mean_itl_ms": 19.5,
"success": true,
"error": "",
"target_port": 30005
}
```
最后一行是汇总数据，包含 **各端口统计**：
```json
{
"completed": 640,
"failed": 0,
"duration": 12.34,
"request_throughput": 51.88,
"mean_ttft_ms": 48.5,
"p95_ttft_ms": 62.3,
"per_port_stats": {
"30005": {"count": 160, "mean_ttft_ms": 47.2, "p95_ttft_ms": 58.1, "mean_e2e_ms": 5200.0},
"30006": {"count": 160, "mean_ttft_ms": 49.1, "p95_ttft_ms": 64.3, "mean_e2e_ms": 5250.0},
"30007": {"count": 160, "mean_ttft_ms": 48.8, "p95_ttft_ms": 61.5, "mean_e2e_ms": 5230.0},
"30008": {"count": 160, "mean_ttft_ms": 48.9, "p95_ttft_ms": 65.2, "mean_e2e_ms": 5270.0}
},
"lb_strategy": "round_robin",
"num_services": 4
}
```
### 结构化结果 (`results.json`)
遵循仓库统一的 [JSON Schema](../../BENCHMARK_WORKFLOW.md#final-json-schema)。
### 人类可读报告 (`report.md`)
Markdown 表格，包含所有场景的 TTFT/TPOT/E2E 均值与分位数，以及 SLO 达标状态。
## 单独启动/停止 4 个服务
```bash
# 启动 4 个服务
bash experiments/dsv4_h200_vllm_tp2_custom_bench/start_server.sh
# 停止（run_bench.sh 会自动停止，但也可以手动）
for port in 30005 30006 30007 30008; do
pid_file="experiments/dsv4_h200_vllm_tp2_custom_bench/server_port${port}.pid"
[[ -f "$pid_file" ]] && kill "$(cat "$pid_file")" 2>/dev/null || true
rm -f "$pid_file"
done
pkill -9 -f "vllm serve.*DeepSeek-V4-Flash" 2>/dev/null || true
```
## 注意事项
- `bench_client.py` 优先使用 `aiohttp`，如果未安装则自动回退到 `urllib`（但会丢失 streaming 和精确 TTFT/ITL 测量）。
- 默认 `temperature=0.0` 以保证输出长度稳定。
- 输入 token 数通过空格分隔的单词数估算，实际 tokenizer 可能略有偏差。
- 4 个服务共享同一模型文件，确保 `/data/models/DeepSeek-V4-Flash` 存在且可访问。
- 每个服务占用约 2 张 GPU 的 90% 显存，确保无其他进程占用 GPU。
- 健康检查依赖 `/health` 端点，确保 vLLM 服务已启用该端点（vLLM 默认启用）。
- 如果某个服务频繁崩溃，检查 GPU 显存是否充足（OOM 是最常见原因）。

617
experiments/dsv4_h200_vllm_tp2_custom_bench/bench_client.py Executable file
View File

@ -0,0 +1,617 @@
#!/usr/bin/env python3
"""Custom benchmark client for vLLM / OpenAI-compatible API with multi-service load balancing and health-check failover.
Sends concurrent requests to multiple inference servers and records per-request
latency metrics (TTFT, TPOT, ITL, E2E) with fine-grained timing.
Designed for TP=2 on 8-GPU setups where 4 independent services run on
(0,1), (2,3), (4,5), (6,7). Requests are distributed across services via
round-robin or random load balancing. Unhealthy services are automatically
skipped with periodic retry.
Usage:
# Single service (backward compatible)
python3 bench_client.py \
--host 127.0.0.1 --port 30005 \
--model deepseek-v4-flash \
--concurrency 64 --input-len 512 --output-len 256 --num-prompts 320 \
--output-file results.jsonl
# Multi-service (4 services on ports 30005-30008)
python3 bench_client.py \
--host 127.0.0.1 \
--ports 30005,30006,30007,30008 \
--model deepseek-v4-flash \
--concurrency 64 --input-len 512 --output-len 256 --num-prompts 320 \
--lb-strategy round_robin \
--output-file results.jsonl
The output JSONL contains one object per request with raw timing data.
A final summary line ("completed" field) aggregates all requests.
"""
import argparse
import asyncio
import json
import random
import sys
import time
import urllib.request
from dataclasses import asdict, dataclass, field
from typing import Any
# Try aiohttp first; fall back to urllib for stdlib-only environments.
try:
import aiohttp
HAS_AIOHTTP = True
except ImportError:
HAS_AIOHTTP = False
# ---------------------------------------------------------------------------
# Data structures
# ---------------------------------------------------------------------------
@dataclass
class RequestResult:
request_id: int
input_tokens: int = 0
output_tokens: int = 0
first_token_time_ms: float = 0.0 # TTFT
e2e_latency_ms: float = 0.0 # total wall time
inter_token_latencies: list[float] = field(default_factory=list)
success: bool = False
error: str = ""
target_port: int = 0 # which service handled this request
retry_count: int = 0 # how many times this request was retried
@property
def tpot_ms(self) -> float:
"""Mean TPOT = (e2e - ttft) / (output_tokens - 1) if >1 token."""
if self.output_tokens <= 1:
return 0.0
return (self.e2e_latency_ms - self.first_token_time_ms) / (self.output_tokens - 1)
@property
def mean_itl_ms(self) -> float:
if not self.inter_token_latencies:
return 0.0
return sum(self.inter_token_latencies) / len(self.inter_token_latencies)
def to_dict(self) -> dict[str, Any]:
d = asdict(self)
d["tpot_ms"] = self.tpot_ms
d["mean_itl_ms"] = self.mean_itl_ms
return d
# ---------------------------------------------------------------------------
# Health checker
# ---------------------------------------------------------------------------
class HealthChecker:
"""Periodically check service health and maintain a healthy-port list."""
def __init__(
self,
host: str,
ports: list[int],
check_interval: float = 5.0,
max_failures: int = 2,
recovery_interval: float = 10.0,
):
self.host = host
self.all_ports = ports
self.check_interval = check_interval
self.max_failures = max_failures
self.recovery_interval = recovery_interval
# port -> consecutive failure count
self._failures: dict[int, int] = {p: 0 for p in ports}
# port -> last healthy timestamp
self._last_healthy: dict[int, float] = {p: time.monotonic() for p in ports}
self._lock = asyncio.Lock()
self._task: asyncio.Task | None = None
async def _check_one(self, port: int, session: aiohttp.ClientSession | None) -> bool:
url = f"http://{self.host}:{port}/health"
try:
if HAS_AIOHTTP and session is not None:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
return resp.status == 200
else:
req = urllib.request.Request(url, method="GET")
with urllib.request.urlopen(req, timeout=5) as resp:
return resp.status == 200
except Exception:
return False
async def _check_all(self, session: aiohttp.ClientSession | None) -> None:
checks = [self._check_one(p, session) for p in self.all_ports]
results = await asyncio.gather(*checks, return_exceptions=True)
now = time.monotonic()
async with self._lock:
for port, ok in zip(self.all_ports, results):
if isinstance(ok, Exception):
ok = False
if ok:
self._failures[port] = 0
self._last_healthy[port] = now
else:
self._failures[port] += 1
async def start(self, session: aiohttp.ClientSession | None) -> None:
async def _loop() -> None:
while True:
await asyncio.sleep(self.check_interval)
await self._check_all(session)
self._task = asyncio.create_task(_loop())
async def stop(self) -> None:
if self._task is not None:
self._task.cancel()
try:
await self._task
except asyncio.CancelledError:
pass
@property
async def healthy_ports(self) -> list[int]:
now = time.monotonic()
async with self._lock:
healthy = []
for p in self.all_ports:
if self._failures[p] < self.max_failures:
healthy.append(p)
elif now - self._last_healthy[p] > self.recovery_interval:
# Give it another chance after recovery_interval.
self._failures[p] = max(0, self._failures[p] - 1)
healthy.append(p)
return healthy
async def is_healthy(self, port: int) -> bool:
ports = await self.healthy_ports
return port in ports
# ---------------------------------------------------------------------------
# Load balancer with failover
# ---------------------------------------------------------------------------
class LoadBalancer:
"""Distribute requests across healthy backend ports with auto-failover."""
def __init__(
self,
ports: list[int],
health_checker: HealthChecker,
strategy: str = "round_robin",
):
self.all_ports = ports
self.health_checker = health_checker
self.strategy = strategy
self._idx = 0
self._lock = asyncio.Lock()
def _next_round_robin(self, healthy: list[int]) -> int:
if not healthy:
return self.all_ports[self._idx % len(self.all_ports)]
for _ in range(len(self.all_ports)):
port = self.all_ports[self._idx % len(self.all_ports)]
self._idx += 1
if port in healthy:
return port
# Fallback: all unhealthy, pick first healthy anyway.
return healthy[0] if healthy else self.all_ports[0]
def _next_random(self, healthy: list[int]) -> int:
if healthy:
return random.choice(healthy)
return random.choice(self.all_ports)
async def next_port(self) -> int:
healthy = await self.health_checker.healthy_ports
async with self._lock:
if self.strategy == "round_robin":
return self._next_round_robin(healthy)
elif self.strategy in ("random", "least_conn"):
return self._next_random(healthy)
else:
return self._next_round_robin(healthy)
@property
def port_count(self) -> int:
return len(self.all_ports)
# ---------------------------------------------------------------------------
# Token helpers
# ---------------------------------------------------------------------------
def make_prompt(token_len: int, vocab_size: int = 32000) -> str:
"""Generate a random-ish prompt of approximately `token_len` tokens."""
words = ["the", "quick", "brown", "fox", "jumps", "over", "lazy",
"dog", "hello", "world", "deep", "seek", "model", "test",
"benchmark", "performance", "latency", "throughput", "token"]
needed = max(token_len, 1)
tokens: list[str] = []
while len(tokens) < needed:
tokens.extend(words)
return " ".join(tokens[:needed])
# ---------------------------------------------------------------------------
# aiohttp-based async request (preferred)
# ---------------------------------------------------------------------------
async def send_request_aiohttp(
session: aiohttp.ClientSession,
host: str,
port: int,
model: str,
request_id: int,
prompt: str,
max_tokens: int,
temperature: float,
) -> RequestResult:
url = f"http://{host}:{port}/v1/chat/completions"
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": max_tokens,
"temperature": temperature,
"stream": True,
}
headers = {"Content-Type": "application/json"}
result = RequestResult(request_id=request_id, target_port=port)
start_time = time.perf_counter()
first_token_received = False
last_token_time = 0.0
try:
async with session.post(url, json=payload, headers=headers) as resp:
if resp.status != 200:
text = await resp.text()
result.error = f"HTTP {resp.status}: {text[:200]}"
return result
async for line in resp.content:
line = line.decode("utf-8").strip()
if not line or not line.startswith("data: "):
continue
data_str = line[len("data: "):]
if data_str == "[DONE]":
break
try:
chunk = json.loads(data_str)
except json.JSONDecodeError:
continue
choices = chunk.get("choices", [])
if not choices:
continue
delta = choices[0].get("delta", {})
content = delta.get("content", "")
if content:
now = time.perf_counter()
if not first_token_received:
result.first_token_time_ms = (now - start_time) * 1000
first_token_received = True
else:
result.inter_token_latencies.append((now - last_token_time) * 1000)
last_token_time = now
result.output_tokens += 1
result.e2e_latency_ms = (time.perf_counter() - start_time) * 1000
result.input_tokens = len(prompt.split()) # approximate
result.success = True
except asyncio.TimeoutError:
result.error = "timeout"
except Exception as e:
result.error = str(e)[:200]
return result
# ---------------------------------------------------------------------------
# urllib-based synchronous request (fallback, no aiohttp)
# ---------------------------------------------------------------------------
def send_request_urllib(
host: str,
port: int,
model: str,
request_id: int,
prompt: str,
max_tokens: int,
temperature: float,
) -> RequestResult:
"""Blocking fallback using stdlib urllib. Used when aiohttp is absent."""
url = f"http://{host}:{port}/v1/chat/completions"
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": max_tokens,
"temperature": temperature,
"stream": False,
}
data = json.dumps(payload).encode("utf-8")
headers = {"Content-Type": "application/json", "Content-Length": str(len(data))}
result = RequestResult(request_id=request_id, target_port=port)
start_time = time.perf_counter()
try:
req = urllib.request.Request(url, data=data, headers=headers, method="POST")
with urllib.request.urlopen(req, timeout=600) as resp:
body = json.loads(resp.read().decode("utf-8"))
choices = body.get("choices", [])
if not choices:
result.error = "no choices in response"
return result
content = choices[0].get("message", {}).get("content", "")
result.output_tokens = len(content.split()) if content else 0
result.e2e_latency_ms = (time.perf_counter() - start_time) * 1000
result.first_token_time_ms = result.e2e_latency_ms
result.input_tokens = len(prompt.split())
result.success = True
except Exception as e:
result.error = str(e)[:200]
return result
# ---------------------------------------------------------------------------
# Unified send_request wrapper with retry & failover
# ---------------------------------------------------------------------------
async def send_request_with_retry(
session: aiohttp.ClientSession | None,
host: str,
lb: LoadBalancer,
model: str,
request_id: int,
prompt: str,
max_tokens: int,
temperature: float,
max_retries: int = 2,
) -> RequestResult:
"""Send a request, retrying on failure with a different port each time."""
result = RequestResult(request_id=request_id)
for attempt in range(max_retries + 1):
port = await lb.next_port()
if HAS_AIOHTTP and session is not None:
result = await send_request_aiohttp(
session, host, port, model, request_id, prompt, max_tokens, temperature
)
else:
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(
None, send_request_urllib,
host, port, model, request_id, prompt, max_tokens, temperature
)
result.retry_count = attempt
if result.success:
return result
# If failed, try another port on next iteration (unless last attempt).
if attempt < max_retries:
await asyncio.sleep(0.5 * (attempt + 1)) # exponential backoff-ish
return result
# ---------------------------------------------------------------------------
# Benchmark orchestration
# ---------------------------------------------------------------------------
async def run_benchmark(
host: str,
ports: list[int],
lb: LoadBalancer,
health_checker: HealthChecker,
model: str,
concurrency: int,
input_len: int,
output_len: int,
num_prompts: int,
temperature: float,
output_file: str,
) -> dict[str, Any]:
"""Run the benchmark and write a JSONL file."""
prompts = [make_prompt(input_len) for _ in range(num_prompts)]
results: list[RequestResult] = []
semaphore = asyncio.Semaphore(concurrency)
start = time.perf_counter()
if HAS_AIOHTTP:
connector = aiohttp.TCPConnector(limit=concurrency * 2)
timeout = aiohttp.ClientTimeout(total=600, sock_read=300)
async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
await health_checker.start(session)
async def bounded_send(idx: int, prompt: str) -> RequestResult:
async with semaphore:
return await send_request_with_retry(
session, host, lb, model, idx, prompt, output_len, temperature
)
tasks = [bounded_send(i, p) for i, p in enumerate(prompts)]
results = await asyncio.gather(*tasks)
await health_checker.stop()
else:
await health_checker.start(None)
from concurrent.futures import ThreadPoolExecutor
async def bounded_send(idx: int, prompt: str) -> RequestResult:
async with semaphore:
return await send_request_with_retry(
None, host, lb, model, idx, prompt, output_len, temperature
)
tasks = [bounded_send(i, p) for i, p in enumerate(prompts)]
results = await asyncio.gather(*tasks)
await health_checker.stop()
duration = time.perf_counter() - start
# Write JSONL
with open(output_file, "w", encoding="utf-8") as f:
for r in results:
f.write(json.dumps(r.to_dict(), ensure_ascii=False) + "\n")
# Compute summary
successful = [r for r in results if r.success]
failed = len(results) - len(successful)
retried = [r for r in results if r.retry_count > 0]
def percentile(values: list[float], p: float) -> float:
if not values:
return 0.0
s = sorted(values)
k = (len(s) - 1) * p / 100.0
f = int(k)
c = min(f + 1, len(s) - 1)
return s[f] + (k - f) * (s[c] - s[f])
ttfts = [r.first_token_time_ms for r in successful]
tpots = [r.tpot_ms for r in successful if r.output_tokens > 1]
e2es = [r.e2e_latency_ms for r in successful]
itls = [r.mean_itl_ms for r in successful if r.inter_token_latencies]
total_input_tokens = sum(r.input_tokens for r in successful)
total_output_tokens = sum(r.output_tokens for r in successful)
# Per-port breakdown
per_port_stats: dict[int, dict[str, Any]] = {}
for port in ports:
port_results = [r for r in successful if r.target_port == port]
port_failed = [r for r in results if r.target_port == port and not r.success]
if port_results or port_failed:
port_ttfts = [r.first_token_time_ms for r in port_results]
port_e2es = [r.e2e_latency_ms for r in port_results]
per_port_stats[port] = {
"count": len(port_results),
"failed": len(port_failed),
"mean_ttft_ms": sum(port_ttfts) / len(port_ttfts) if port_ttfts else 0.0,
"p95_ttft_ms": percentile(port_ttfts, 95) if port_ttfts else 0.0,
"mean_e2e_ms": sum(port_e2es) / len(port_e2es) if port_e2es else 0.0,
}
# Health-check stats
healthy_at_end = await health_checker.healthy_ports
summary = {
"completed": len(successful),
"failed": failed,
"retried": len(retried),
"duration": duration,
"request_throughput": len(successful) / duration if duration > 0 else 0.0,
"input_throughput": total_input_tokens / duration if duration > 0 else 0.0,
"output_throughput": total_output_tokens / duration if duration > 0 else 0.0,
"total_throughput": (total_input_tokens + total_output_tokens) / duration if duration > 0 else 0.0,
"total_input_tokens": total_input_tokens,
"total_output_tokens": total_output_tokens,
"mean_ttft_ms": sum(ttfts) / len(ttfts) if ttfts else 0.0,
"median_ttft_ms": percentile(ttfts, 50),
"p90_ttft_ms": percentile(ttfts, 90),
"p95_ttft_ms": percentile(ttfts, 95),
"p99_ttft_ms": percentile(ttfts, 99),
"mean_tpot_ms": sum(tpots) / len(tpots) if tpots else 0.0,
"median_tpot_ms": percentile(tpots, 50),
"p90_tpot_ms": percentile(tpots, 90),
"p95_tpot_ms": percentile(tpots, 95),
"p99_tpot_ms": percentile(tpots, 99),
"mean_e2e_latency_ms": sum(e2es) / len(e2es) if e2es else 0.0,
"median_e2e_latency_ms": percentile(e2es, 50),
"p90_e2e_latency_ms": percentile(e2es, 90),
"p95_e2e_latency_ms": percentile(e2es, 95),
"p99_e2e_latency_ms": percentile(e2es, 99),
"mean_itl_ms": sum(itls) / len(itls) if itls else 0.0,
"median_itl_ms": percentile(itls, 50),
"p90_itl_ms": percentile(itls, 90),
"p95_itl_ms": percentile(itls, 95),
"p99_itl_ms": percentile(itls, 99),
"per_port_stats": per_port_stats,
"healthy_ports_at_end": healthy_at_end,
"lb_strategy": lb.strategy,
"num_services": lb.port_count,
}
# Append summary as the last line
with open(output_file, "a", encoding="utf-8") as f:
f.write(json.dumps(summary, ensure_ascii=False) + "\n")
return summary
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def main() -> None:
parser = argparse.ArgumentParser(description="Custom vLLM benchmark client with multi-service LB and health-check failover")
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=30005,
help="Single service port (backward compatible)")
parser.add_argument("--ports", default="",
help="Comma-separated list of ports for multi-service, e.g. 30005,30006,30007,30008")
parser.add_argument("--model", default="deepseek-v4-flash")
parser.add_argument("--concurrency", type=int, default=1)
parser.add_argument("--input-len", type=int, default=512)
parser.add_argument("--output-len", type=int, default=256)
parser.add_argument("--num-prompts", type=int, default=10)
parser.add_argument("--temperature", type=float, default=0.0)
parser.add_argument("--lb-strategy", default="round_robin",
choices=["round_robin", "random", "least_conn"],
help="Load balancing strategy across services")
parser.add_argument("--health-check-interval", type=float, default=5.0,
help="Seconds between health checks (default: 5)")
parser.add_argument("--health-max-failures", type=int, default=2,
help="Consecutive failures before marking a port unhealthy (default: 2)")
parser.add_argument("--health-recovery-interval", type=float, default=10.0,
help="Seconds before retrying an unhealthy port (default: 10)")
parser.add_argument("--request-max-retries", type=int, default=2,
help="Max retries per request on failure (default: 2)")
parser.add_argument("--output-file", required=True)
args = parser.parse_args()
# Determine ports: --ports takes precedence, else fall back to --port.
if args.ports:
ports = [int(p.strip()) for p in args.ports.split(",")]
else:
ports = [args.port]
health_checker = HealthChecker(
host=args.host,
ports=ports,
check_interval=args.health_check_interval,
max_failures=args.health_max_failures,
recovery_interval=args.health_recovery_interval,
)
lb = LoadBalancer(ports, health_checker=health_checker, strategy=args.lb_strategy)
summary = asyncio.run(run_benchmark(
host=args.host,
ports=ports,
lb=lb,
health_checker=health_checker,
model=args.model,
concurrency=args.concurrency,
input_len=args.input_len,
output_len=args.output_len,
num_prompts=args.num_prompts,
temperature=args.temperature,
output_file=args.output_file,
))
print(json.dumps(summary, indent=2, ensure_ascii=False))
if __name__ == "__main__":
main()

75
experiments/dsv4_h200_vllm_tp2_custom_bench/config.env Normal file
View File

@ -0,0 +1,75 @@
# Experiment configuration for dsv4_h200_vllm_tp2_custom_bench
# Custom benchmark client for multi-service vLLM TP=2 on 8x H200.
#
# This experiment runs 4 independent vLLM services (TP=2 each) on GPU pairs:
# Service 1: port 30005, GPUs 0,1
# Service 2: port 30006, GPUs 2,3
# Service 3: port 30007, GPUs 4,5
# Service 4: port 30008, GPUs 6,7
#
# bench_client.py distributes requests across all 4 services via round-robin
# load balancing, giving accurate cluster-wide throughput even at low
# per-service concurrency.
EXPERIMENT="dsv4_h200_vllm_tp2_custom_bench"
MODEL_NAME="DeepSeek-V4-Flash"
MODEL_PATH="/data/models/DeepSeek-V4-Flash"
SERVED_MODEL_NAME="deepseek-v4-flash"
# Primary port (backward compatible); multi-service ports are in PORTS.
PORT="30005"
PORTS="30005,30006,30007,30008"
BACKEND="vllm"
ENGINE="vllm"
# Native virtual environments on the host.
VENV_SERVER="/data/user1/yy/envs/vllm"
VENV_CLIENT="/data/user1/yy/envs/vllm"
# Load balancing strategy: round_robin | random | least_conn
LB_STRATEGY="round_robin"
# Benchmark scenarios: "concurrency input_len output_len num_prompts".
# For multi-service we include low-concurrency scenarios to show how
# requests spread across 4 services.
SCENARIOS=(
"1 512 256 20"
"4 512 256 80"
"8 512 256 160"
"16 512 256 320"
"32 512 256 640"
"64 512 256 1280"
"128 512 256 2560"
"1 2048 512 20"
"8 2048 512 160"
"32 2048 512 640"
"128 2048 512 2560"
"1 4096 1024 20"
"8 4096 1024 160"
"32 4096 1024 640"
"128 4096 1024 2560"
"1 8192 1024 20"
"8 8192 1024 160"
"32 8192 1024 640"
"128 8192 1024 2560"
"1 16384 1024 20"
"8 16384 1024 160"
"32 16384 1024 640"
"128 16384 1024 2560"
"1 32768 1024 20"
"8 32768 1024 160"
"32 32768 1024 640"
"128 32768 1024 2560"
"1 65536 1024 20"
"8 65536 1024 160"
"32 65536 1024 640"
"128 65536 1024 2560"
"1 131072 1024 20"
"8 131072 1024 160"
"32 131072 1024 640"
"128 131072 1024 2560"
)
# Server start script bundled with this experiment.
SERVER_START_SCRIPT="${SCRIPT_DIR}/start_server.sh"

206
experiments/dsv4_h200_vllm_tp2_custom_bench/parse_results.py Executable file
View File

@ -0,0 +1,206 @@
#!/usr/bin/env python3
"""Parse custom bench_client.py JSONL outputs for vLLM TP=2 benchmark.
Reads JSONL files produced by bench_client.py (one request per line,
last line is the summary) and generates:
- results.json (appended scenarios, following the standard schema)
- report.md (human-readable summary)
Usage:
python3 parse_results.py <result_root>
"""
import json
import sys
from pathlib import Path
def parse_jsonl(path: Path) -> tuple[dict | None, list[dict]]:
"""Read the summary JSON (last line) and all per-request lines."""
requests: list[dict] = []
summary: dict | None = None
with open(path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
obj = json.loads(line)
except json.JSONDecodeError:
continue
# The summary line has "completed" and "duration" keys.
if "completed" in obj and "duration" in obj:
summary = obj
else:
requests.append(obj)
return summary, requests
def compute_metrics(summary: dict, requests: list[dict]) -> dict:
completed = summary.get("completed", 0)
total = len(requests)
failed = total - completed if total > 0 else 0
duration_s = summary.get("duration", 0.0)
# Extract per-request metrics for percentiles.
ttfts = [r["first_token_time_ms"] for r in requests if r.get("success")]
tpots = [r["tpot_ms"] for r in requests if r.get("success") and r.get("output_tokens", 0) > 1]
e2es = [r["e2e_latency_ms"] for r in requests if r.get("success")]
itls = [r["mean_itl_ms"] for r in requests if r.get("success") and r.get("inter_token_latencies")]
def percentile(values: list[float], p: float) -> float:
if not values:
return 0.0
s = sorted(values)
k = (len(s) - 1) * p / 100.0
f = int(k)
c = min(f + 1, len(s) - 1)
return s[f] + (k - f) * (s[c] - s[f])
return {
"success": completed,
"failed": failed,
"duration_s": duration_s,
"request_throughput": summary.get("request_throughput", 0.0),
"input_token_throughput": summary.get("input_throughput", 0.0),
"output_token_throughput": summary.get("output_throughput", 0.0),
"total_token_throughput": summary.get("total_throughput", 0.0),
"total_input_tokens": summary.get("total_input_tokens", 0),
"total_output_tokens": summary.get("total_output_tokens", 0),
"e2e_ms": {
"mean": summary.get("mean_e2e_latency_ms", 0.0),
"p50": percentile(e2es, 50),
"p90": percentile(e2es, 90),
"p95": percentile(e2es, 95),
"p99": percentile(e2es, 99),
},
"ttft_ms": {
"mean": summary.get("mean_ttft_ms", 0.0),
"p50": percentile(ttfts, 50),
"p90": percentile(ttfts, 90),
"p95": percentile(ttfts, 95),
"p99": percentile(ttfts, 99),
},
"tpot_ms": {
"mean": summary.get("mean_tpot_ms", 0.0),
"p50": percentile(tpots, 50),
"p90": percentile(tpots, 90),
"p95": percentile(tpots, 95),
"p99": percentile(tpots, 99),
},
"itl_ms": {
"mean": summary.get("mean_itl_ms", 0.0),
"p50": percentile(itls, 50),
"p90": percentile(itls, 90),
"p95": percentile(itls, 95),
"p99": percentile(itls, 99),
},
}
def scenario_name(concurrency: int, input_len: int, output_len: int) -> str:
return f"c{concurrency}_i{input_len}_o{output_len}"
def slo_status(metrics: dict, ttft_limit_ms: float = 3000.0, tpot_limit_ms: float = 50.0) -> dict:
ttft_ok = metrics["ttft_ms"]["p95"] < ttft_limit_ms
tpot_ok = metrics["tpot_ms"]["mean"] < tpot_limit_ms
if ttft_ok and tpot_ok:
mark = "✅"
elif ttft_ok or tpot_ok:
mark = "⚠️"
else:
mark = "❌"
return {
"ttft_p95_ok": ttft_ok,
"tpot_mean_ok": tpot_ok,
"overall": mark,
}
def append_scenario(results_json: Path, scenario: dict) -> None:
with open(results_json, "r", encoding="utf-8") as f:
data = json.load(f)
data["scenarios"].append(scenario)
with open(results_json, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
def generate_report(result_root: Path, scenarios: list[dict]) -> None:
report_path = result_root / "report.md"
with open(report_path, "w", encoding="utf-8") as f:
f.write("# H200 vLLM TP=2 Custom Benchmark Report\n\n")
f.write("- **Client**: `bench_client.py` (async OpenAI API, per-request timing)\n")
f.write("- **Backend**: vLLM (TP=2, FP8 KV cache, no speculative decoding)\n")
f.write(f"- **Result root**: `{result_root}`\n\n")
f.write("## Results\n\n")
f.write("| Scenario | Concurrency | Input | Output | Duration(s) | Success | Failed | Req/s | In tok/s | Out tok/s | Total tok/s | Mean TTFT(ms) | P95 TTFT(ms) | P99 TTFT(ms) | Mean TPOT(ms) | P95 TPOT(ms) | P99 TPOT(ms) | Mean E2E(ms) | P95 E2E(ms) | P99 E2E(ms) | SLO |\n")
f.write("|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|\n")
for s in scenarios:
cfg = s["config"]
m = s["metrics"]
slo = s.get("slo_status", {}).get("overall", "")
f.write(
f"| {s['name']} | {cfg['concurrency']} | {cfg['input_len']} | {cfg['output_len']} | "
f"{m['duration_s']:.2f} | {m['success']} | {m['failed']} | {m['request_throughput']:.2f} | "
f"{m['input_token_throughput']:.2f} | {m['output_token_throughput']:.2f} | "
f"{m['total_token_throughput']:.2f} | "
f"{m['ttft_ms']['mean']:.2f} | {m['ttft_ms']['p95']:.2f} | {m['ttft_ms']['p99']:.2f} | "
f"{m['tpot_ms']['mean']:.2f} | {m['tpot_ms']['p95']:.2f} | {m['tpot_ms']['p99']:.2f} | "
f"{m['e2e_ms']['mean']:.2f} | {m['e2e_ms']['p95']:.2f} | {m['e2e_ms']['p99']:.2f} | {slo} |\n"
)
f.write("\n")
f.write("SLO: S2 tier — TTFT P95 < 3000ms, TPOT mean < 50ms. ✅ pass, ⚠️ partial, ❌ fail.\n\n")
def main() -> None:
result_root = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("results")
raw_dir = result_root / "raw_outputs"
results_json = result_root / "results.json"
if not raw_dir.exists():
raise SystemExit(f"raw_outputs directory not found: {raw_dir}")
scenarios = []
for jsonl_path in sorted(raw_dir.glob("vllm_*.jsonl")):
parts = jsonl_path.stem.split("_")
if len(parts) < 5:
continue
concurrency, input_len, output_len = int(parts[2]), int(parts[3]), int(parts[4])
summary, requests = parse_jsonl(jsonl_path)
if summary is None:
continue
metrics = compute_metrics(summary, requests)
scenario = {
"name": scenario_name(concurrency, input_len, output_len),
"config": {
"concurrency": concurrency,
"input_len": input_len,
"output_len": output_len,
"dataset": "random",
"num_prompts": metrics["success"] + metrics["failed"],
},
"metrics": metrics,
"slo_status": slo_status(metrics),
"raw_file": str(jsonl_path),
}
scenarios.append(scenario)
if not scenarios:
print("No benchmark outputs found to parse")
return
if results_json.exists():
for s in scenarios:
append_scenario(results_json, s)
generate_report(result_root, scenarios)
print(f"Parsed {len(scenarios)} scenarios into {result_root}/report.md")
if __name__ == "__main__":
main()

277
experiments/dsv4_h200_vllm_tp2_custom_bench/run_bench.sh Executable file
View File

@ -0,0 +1,277 @@
#!/usr/bin/env bash
# Custom benchmark orchestrator for multi-service vLLM TP=2 on 8x H200.
#
# This runs 4 independent vLLM services (TP=2 each) on GPU pairs (0,1),(2,3),(4,5),(6,7).
# bench_client.py distributes requests across all 4 services via round-robin LB,
# giving accurate cluster-wide throughput even at low per-service concurrency.
#
# Usage:
# bash run_bench.sh
# SKIP_MANAGE_SERVER=1 bash run_bench.sh # skip server start/stop
set -Eeuo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
EXPERIMENT_NAME="$(basename "$SCRIPT_DIR")"
# shellcheck source=/dev/null
source "${SCRIPT_DIR}/../../scripts/common/lib.sh"
# shellcheck source=/dev/null
source "${SCRIPT_DIR}/../../scripts/common/platform.sh"
# shellcheck source=/dev/null
source "${SCRIPT_DIR}/config.env"
RUN_ID="${RUN_ID:-$(date '+%Y%m%d-%H%M%S')}"
RESULT_ROOT="${RESULT_ROOT:-${SCRIPT_DIR}/results/${RUN_ID}}"
RAW_DIR="${RESULT_ROOT}/raw_outputs"
LOG_DIR="${RESULT_ROOT}/logs"
ensure_result_root "$RESULT_ROOT"
log_init "${LOG_DIR}/orchestrator.log"
log "experiment=${EXPERIMENT_NAME}"
log "run_id=${RUN_ID}"
log "result_root=${RESULT_ROOT}"
log "platform=${PLATFORM}"
log "chip=${CHIP}"
log "accelerator=${ACCELERATOR}"
log "engine=${ENGINE}"
log "hardware=${HARDWARE}"
log "model=${MODEL_PATH}"
log "services=${PORTS}"
log "lb_strategy=${LB_STRATEGY}"
log "server_start_script=${SERVER_START_SCRIPT}"
# Write initial metadata for this run.
METADATA_JSON="${RESULT_ROOT}/results.json"
write_metadata_json \
"$METADATA_JSON" \
"$EXPERIMENT_NAME" \
"$RUN_ID" \
"$MODEL_PATH" \
"$BACKEND" \
"$ENGINE" \
"$HARDWARE" \
"$ACCELERATOR" \
"$CHIP" \
"experiments/${EXPERIMENT_NAME}/run_bench.sh" \
"$VENV_SERVER" \
"H200 4x vLLM TP=2 multi-service benchmark using bench_client.py"
# Update metadata with config.
"${VENV_CLIENT}/bin/python" - "$METADATA_JSON" <<'PY'
import json, sys
path = sys.argv[1]
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
data["config"] = {
"tp": 2,
"num_services": 4,
"ports": "30005,30006,30007,30008",
"gpu_pairs": "0,1|2,3|4,5|6,7",
"kv_cache_dtype": "fp8",
"block_size": 256,
"max_num_seqs": 256,
"client": "bench_client.py",
"lb_strategy": "round_robin",
"scenarios": [
"1 512 256 20", "4 512 256 80", "8 512 256 160",
"16 512 256 320", "32 512 256 640", "64 512 256 1280", "128 512 256 2560",
"1 2048 512 20", "8 2048 512 160", "32 2048 512 640", "128 2048 512 2560",
"1 4096 1024 20", "8 4096 1024 160", "32 4096 1024 640", "128 4096 1024 2560"
]
}
with open(path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
PY
# ---------------------------------------------------------------------------
# Server helpers (multi-service)
# ---------------------------------------------------------------------------
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
}
are_all_services_healthy() {
local ports="$1"
IFS=',' read -ra PORT_LIST <<< "$ports"
for p in "${PORT_LIST[@]}"; do
if ! is_server_healthy "$p"; then
return 1
fi
done
return 0
}
stop_all_servers() {
log "stopping all vllm services"
# Kill by PID files first.
for pid_file in "${SCRIPT_DIR}"/server_port*.pid; do
[[ -f "$pid_file" ]] || continue
local pid
pid="$(cat "$pid_file")"
if kill -0 "$pid" 2>/dev/null; then
log "killing pid=${pid}"
kill "$pid" 2>/dev/null || true
sleep 2
kill -9 "$pid" 2>/dev/null || true
fi
rm -f "$pid_file"
done
# Fallback: kill any remaining vllm processes for this model.
pkill -9 -f "vllm serve.*${MODEL_NAME}" 2>/dev/null || true
sleep 2
}
start_all_servers() {
log "starting all vllm services with ${SERVER_START_SCRIPT}"
if [[ ! -x "${SERVER_START_SCRIPT}" ]]; then
log "error: start script not found or not executable: ${SERVER_START_SCRIPT}"
exit 1
fi
bash "${SERVER_START_SCRIPT}" >> "${LOG_DIR}/server.outer.log" 2>&1
if ! are_all_services_healthy "$PORTS"; then
log "error: not all vllm services became healthy"
exit 1
fi
log "all vllm services are healthy"
}
# ---------------------------------------------------------------------------
# Pre-benchmark health check with detailed reporting
# ---------------------------------------------------------------------------
check_services_detailed() {
local ports="$1"
IFS=',' read -ra PORT_LIST <<< "$ports"
local healthy_count=0
local unhealthy_ports=""
for p in "${PORT_LIST[@]}"; do
if is_server_healthy "$p"; then
((healthy_count++))
else
unhealthy_ports="${unhealthy_ports}${p} "
fi
done
echo "$healthy_count"
if [[ -n "$unhealthy_ports" ]]; then
echo "unhealthy: ${unhealthy_ports}" >&2
fi
}
wait_for_all_services() {
local ports="$1"
local max_wait="${2:-240}"
local interval="${3:-5}"
local elapsed=0
while (( elapsed < max_wait )); do
local healthy_count
healthy_count="$(check_services_detailed "$ports" | head -1)"
IFS=',' read -ra PORT_LIST <<< "$ports"
if (( healthy_count == ${#PORT_LIST[@]} )); then
return 0
fi
log "waiting for services... ${healthy_count}/${#PORT_LIST[@]} healthy after ${elapsed}s"
sleep "$interval"
((elapsed += interval))
done
return 1
}
# ---------------------------------------------------------------------------
# Mid-benchmark health check (called between scenarios)
# ---------------------------------------------------------------------------
check_and_report_services() {
local ports="$1"
IFS=',' read -ra PORT_LIST <<< "$ports"
local healthy=()
local unhealthy=()
for p in "${PORT_LIST[@]}"; do
if is_server_healthy "$p"; then
healthy+=("$p")
else
unhealthy+=("$p")
fi
done
log "service health check: healthy=[${healthy[*]}] unhealthy=[${unhealthy[*]}]"
if [[ ${#unhealthy[@]} -gt 0 ]]; then
log "WARNING: ${#unhealthy[@]} service(s) unhealthy: ${unhealthy[*]}"
fi
return ${#unhealthy[@]}
}
on_exit() {
local code=$?
log "orchestrator exiting with code=${code}"
if [[ -z "${SKIP_MANAGE_SERVER:-}" ]]; then
stop_all_servers
fi
exit "$code"
}
trap on_exit EXIT
if [[ -n "${SKIP_MANAGE_SERVER:-}" ]]; then
log "SKIP_MANAGE_SERVER is set, assuming services are already running"
if ! wait_for_all_services "$PORTS"; then
log "error: not all healthy services found on ports ${PORTS}"
exit 1
fi
else
stop_all_servers
start_all_servers
fi
log "===== BENCHMARK START ====="
for scenario in "${SCENARIOS[@]}"; do
read -r concurrency input_len output_len num_prompts <<< "$scenario"
# Fallback for legacy 3-field scenarios.
if [[ -z "${num_prompts:-}" ]]; then
num_prompts="$NUM_PROMPTS"
fi
output_file="${RAW_DIR}/vllm_$(date '+%m%d')_${concurrency}_${input_len}_${output_len}.jsonl"
detail_log="${LOG_DIR}/vllm_c${concurrency}_i${input_len}_o${output_len}.log"
log "running scenario: concurrency=${concurrency} input=${input_len} output=${output_len} num_prompts=${num_prompts}"
"${VENV_CLIENT}/bin/python" "${SCRIPT_DIR}/bench_client.py" \
--host 127.0.0.1 \
--ports "$PORTS" \
--model "$SERVED_MODEL_NAME" \
--concurrency "$concurrency" \
--input-len "$input_len" \
--output-len "$output_len" \
--num-prompts "$num_prompts" \
--lb-strategy "$LB_STRATEGY" \
--health-check-interval 5 \
--health-max-failures 2 \
--health-recovery-interval 10 \
--request-max-retries 2 \
--output-file "$output_file" \
> "$detail_log" 2>&1 || {
log "ERROR: scenario c=${concurrency} i=${input_len} o=${output_len} failed; see ${detail_log}"
# After failure, check service health before continuing.
check_and_report_services "$PORTS"
continue
}
log "finished scenario: output=${output_file}"
# Between scenarios, report service health so we can spot degradation early.
check_and_report_services "$PORTS" || true
done
log "===== BENCHMARK DONE ====="
log "parsing results"
"${VENV_CLIENT}/bin/python" "${SCRIPT_DIR}/parse_results.py" "$RESULT_ROOT" >> "${LOG_DIR}/parse.log" 2>&1 || {
log "WARNING: parser failed; see ${LOG_DIR}/parse.log"
}
log "all results saved to ${RESULT_ROOT}"

109
experiments/dsv4_h200_vllm_tp2_custom_bench/start_server.sh Executable file
View File

@ -0,0 +1,109 @@
#!/bin/bash
# Start 4 independent vLLM TP=2 services on 8x H200.
# Each service runs on a pair of GPUs: (0,1), (2,3), (4,5), (6,7).
# This maximizes GPU utilization for low-concurrency workloads.
set -e
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# shellcheck source=/dev/null
source "${SCRIPT_DIR}/config.env"
VENV="${VENV_SERVER}"
export PATH="$VENV/bin:$PATH"
VLLM="$VENV/bin/vllm"
TP=2
LOG_DIR="${SCRIPT_DIR}/logs"
mkdir -p "$LOG_DIR"
# 4 services, each on 2 GPUs.
# Port and GPU pair mapping.
SERVICES=(
"30005:0,1"
"30006:2,3"
"30007:4,5"
"30008:6,7"
)
# Kill any existing vLLM processes for this model.
pkill -9 -f "vllm serve.*${MODEL_NAME}" 2>/dev/null || true
sleep 2
# Remove old PID files.
rm -f "${SCRIPT_DIR}"/*.pid
for svc in "${SERVICES[@]}"; do
PORT="${svc%%:*}"
GPUS="${svc##*:}"
LOG="${LOG_DIR}/server_port${PORT}_$(date +%Y%m%d_%H%M%S).log"
PID_FILE="${SCRIPT_DIR}/server_port${PORT}.pid"
echo "=== Starting service on port ${PORT}, GPUs ${GPUS} ==="
echo "Log: ${LOG}"
CUDA_VISIBLE_DEVICES="${GPUS}" \
nohup "$VLLM" serve "$MODEL_PATH" \
--trust-remote-code \
--tensor-parallel-size "$TP" \
--kv-cache-dtype fp8 \
--block-size 256 \
--port "$PORT" \
> "$LOG" 2>&1 &
PID=$!
echo $PID > "$PID_FILE"
echo "PID: $PID"
done
echo ""
echo "Waiting for all 4 services to become healthy..."
MAX_WAIT=240
all_healthy=0
for i in $(seq 1 $MAX_WAIT); do
all_healthy=1
for svc in "${SERVICES[@]}"; do
PORT="${svc%%:*}"
if ! curl -s "http://127.0.0.1:${PORT}/health" > /dev/null 2>&1; then
all_healthy=0
break
fi
done
if [[ "$all_healthy" -eq 1 ]]; then
echo "All 4 services are healthy!"
for svc in "${SERVICES[@]}"; do
PORT="${svc%%:*}"
echo " - http://127.0.0.1:${PORT}"
done
exit 0
fi
# Check if any process died early.
for svc in "${SERVICES[@]}"; do
PORT="${svc%%:*}"
PID_FILE="${SCRIPT_DIR}/server_port${PORT}.pid"
if [[ -f "$PID_FILE" ]]; then
PID="$(cat "$PID_FILE")"
if ! kill -0 "$PID" 2>/dev/null; then
echo "ERROR: Service on port ${PORT} (PID ${PID}) exited early"
LOG="${LOG_DIR}/server_port${PORT}_*.log"
tail -200 $LOG 2>/dev/null || true
exit 1
fi
fi
done
if (( i % 10 == 0 )); then
echo "Waiting... (${i}/${MAX_WAIT})"
fi
sleep 5
done
echo "ERROR: Not all services became healthy after ${MAX_WAIT} retries"
for svc in "${SERVICES[@]}"; do
PORT="${svc%%:*}"
LOG="${LOG_DIR}/server_port${PORT}_*.log"
echo "--- Last 50 lines of port ${PORT} ---"
tail -50 $LOG 2>/dev/null || true
done
exit 1