284 lines
11 KiB
Python
Executable File
284 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Run and summarize a replicated MiniMax-H3 T2VA throughput benchmark."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
import statistics
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
|
|
DEFAULT_PROMPT = "A cinematic landscape with natural motion and realistic lighting."
|
|
|
|
|
|
def percentile(values: list[float], q: float) -> float:
|
|
if not values:
|
|
return 0.0
|
|
values = sorted(values)
|
|
pos = (len(values) - 1) * q
|
|
lo, hi = math.floor(pos), math.ceil(pos)
|
|
if lo == hi:
|
|
return values[lo]
|
|
return values[lo] * (hi - pos) + values[hi] * (pos - lo)
|
|
|
|
|
|
def load_prompts(path: Path, count: int) -> list[str]:
|
|
prompts = []
|
|
if path.is_file():
|
|
prompts = [line.strip() for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
|
|
if not prompts:
|
|
prompts = [DEFAULT_PROMPT]
|
|
repeats = (count + len(prompts) - 1) // len(prompts)
|
|
return (prompts * repeats)[:count]
|
|
|
|
|
|
def payload(
|
|
args: argparse.Namespace,
|
|
prompt: str,
|
|
seed: int,
|
|
steps: int,
|
|
short_edge: int,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"model": args.model,
|
|
"prompt": prompt,
|
|
"num_outputs_per_prompt": 1,
|
|
"num_inference_steps": steps,
|
|
"flow_shift": args.flow_shift,
|
|
"audio_flow_shift": args.audio_flow_shift,
|
|
"seed": seed,
|
|
"task": "t2va",
|
|
"conditions": [],
|
|
"target": {
|
|
"short_edge": short_edge,
|
|
"aspect_ratio": args.aspect_ratio,
|
|
"duration_seconds": args.duration_seconds,
|
|
},
|
|
}
|
|
|
|
|
|
def run_one(
|
|
session: requests.Session,
|
|
args: argparse.Namespace,
|
|
request_id: str,
|
|
prompt_index: int,
|
|
prompt: str,
|
|
seed: int,
|
|
steps: int,
|
|
short_edge: int,
|
|
) -> dict[str, Any]:
|
|
started_epoch = time.time()
|
|
started = time.monotonic()
|
|
result: dict[str, Any] = {
|
|
"request_id": request_id,
|
|
"task": "t2va",
|
|
"replica_index": args.replica_index,
|
|
"port": args.port,
|
|
"prompt_index": prompt_index,
|
|
"prompt": prompt,
|
|
"seed": seed,
|
|
"num_inference_steps": steps,
|
|
"short_edge": short_edge,
|
|
"aspect_ratio": args.aspect_ratio,
|
|
"duration_seconds": args.duration_seconds,
|
|
"conditions": [],
|
|
"started_at_epoch": started_epoch,
|
|
"success": False,
|
|
"error": None,
|
|
}
|
|
try:
|
|
response = session.post(
|
|
f"http://{args.host}:{args.port}/v1/videos",
|
|
json=payload(args, prompt, seed, steps, short_edge),
|
|
timeout=args.submit_timeout,
|
|
)
|
|
if response.status_code != 200:
|
|
raise RuntimeError(f"submit HTTP {response.status_code}: {response.text[:2000]}")
|
|
status = response.json()
|
|
video_id = status.get("id")
|
|
if not video_id:
|
|
raise RuntimeError(f"submit response has no id: {status}")
|
|
result["video_id"] = video_id
|
|
deadline = time.monotonic() + args.request_timeout
|
|
while status.get("status") not in {"completed", "failed"}:
|
|
if time.monotonic() >= deadline:
|
|
raise TimeoutError(f"video job {video_id} exceeded {args.request_timeout}s")
|
|
time.sleep(args.poll_interval)
|
|
poll = session.get(
|
|
f"http://{args.host}:{args.port}/v1/videos/{video_id}",
|
|
timeout=args.poll_timeout,
|
|
)
|
|
if poll.status_code != 200:
|
|
raise RuntimeError(f"poll HTTP {poll.status_code}: {poll.text[:2000]}")
|
|
status = poll.json()
|
|
if status.get("status") != "completed":
|
|
raise RuntimeError(f"job failed: {status.get('error') or status}")
|
|
result["success"] = True
|
|
result["inference_time_s"] = status.get("inference_time_s")
|
|
result["peak_memory_mb"] = status.get("peak_memory_mb")
|
|
result["file_path"] = status.get("file_path")
|
|
result["status"] = status
|
|
except Exception as exc:
|
|
result["error"] = f"{type(exc).__name__}: {exc}"
|
|
result["latency_s"] = time.monotonic() - started
|
|
result["finished_at_epoch"] = time.time()
|
|
return result
|
|
|
|
|
|
def run(args: argparse.Namespace) -> int:
|
|
prompts = load_prompts(args.prompt_file, args.requests_per_resolution)
|
|
resolutions = [int(item) for item in args.resolutions.split(",") if item.strip()]
|
|
full_plan = [
|
|
(prompt_index, prompt, short_edge)
|
|
for prompt_index, prompt in enumerate(prompts)
|
|
for short_edge in resolutions
|
|
]
|
|
shard = [item for item in full_plan if item[0] % args.num_replicas == args.replica_index]
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
failures = 0
|
|
with requests.Session() as session, args.output.open("w", encoding="utf-8", buffering=1) as output:
|
|
warmup = run_one(
|
|
session,
|
|
args,
|
|
f"warmup-r{args.replica_index}",
|
|
0,
|
|
shard[0][1],
|
|
args.seed,
|
|
args.warmup_inference_steps,
|
|
shard[0][2],
|
|
)
|
|
print(
|
|
f"warmup replica={args.replica_index} success={warmup['success']} "
|
|
f"latency={warmup['latency_s']:.2f}s error={warmup['error']}",
|
|
flush=True,
|
|
)
|
|
if not warmup["success"]:
|
|
return 1
|
|
for request_index, (prompt_index, prompt, short_edge) in enumerate(shard):
|
|
seed = args.seed + prompt_index
|
|
request_id = f"t2va-r{short_edge}-p{prompt_index:02d}"
|
|
result = run_one(
|
|
session,
|
|
args,
|
|
request_id,
|
|
prompt_index,
|
|
prompt,
|
|
seed,
|
|
args.num_inference_steps,
|
|
short_edge,
|
|
)
|
|
output.write(json.dumps(result, ensure_ascii=False) + "\n")
|
|
failures += int(not result["success"])
|
|
print(
|
|
f"request {request_index + 1}/{len(shard)} id={request_id} "
|
|
f"success={result['success']} latency={result['latency_s']:.2f}s "
|
|
f"error={result['error']}",
|
|
flush=True,
|
|
)
|
|
return int(failures > 0)
|
|
|
|
|
|
def summarize(args: argparse.Namespace) -> int:
|
|
rows: list[dict[str, Any]] = []
|
|
for path in sorted(args.input_dir.glob("client_*/results.jsonl")):
|
|
rows.extend(json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip())
|
|
successful = [row for row in rows if row.get("success")]
|
|
latencies = [float(row["latency_s"]) for row in successful]
|
|
starts = [float(row["started_at_epoch"]) for row in rows]
|
|
finishes = [float(row["finished_at_epoch"]) for row in rows]
|
|
wall_s = max(finishes) - min(starts) if starts and finishes else 0.0
|
|
by_replica = {}
|
|
for replica in sorted({int(row["replica_index"]) for row in rows}):
|
|
subset = [row for row in rows if int(row["replica_index"]) == replica]
|
|
ok = [row for row in subset if row.get("success")]
|
|
vals = [float(row["latency_s"]) for row in ok]
|
|
by_replica[str(replica)] = {
|
|
"requests": len(subset),
|
|
"completed": len(ok),
|
|
"latency_mean_s": statistics.fmean(vals) if vals else 0.0,
|
|
"latency_p95_s": percentile(vals, 0.95),
|
|
}
|
|
by_short_edge = {}
|
|
for short_edge in sorted({int(row["short_edge"]) for row in rows}):
|
|
subset = [row for row in rows if int(row["short_edge"]) == short_edge]
|
|
ok = [row for row in subset if row.get("success")]
|
|
vals = [float(row["latency_s"]) for row in ok]
|
|
bucket_starts = [float(row["started_at_epoch"]) for row in subset]
|
|
bucket_finishes = [float(row["finished_at_epoch"]) for row in subset]
|
|
bucket_wall = max(bucket_finishes) - min(bucket_starts) if bucket_starts and bucket_finishes else 0.0
|
|
by_short_edge[str(short_edge)] = {
|
|
"requests": len(subset),
|
|
"completed": len(ok),
|
|
"latency_mean_s": statistics.fmean(vals) if vals else 0.0,
|
|
"latency_p95_s": percentile(vals, 0.95),
|
|
"observed_span_s": bucket_wall,
|
|
"equivalent_machine_qps": 4.0 / statistics.fmean(vals) if vals else 0.0,
|
|
}
|
|
summary = {
|
|
"method": "base",
|
|
"task": "t2va",
|
|
"tp": 2,
|
|
"replicas": 4,
|
|
"expected_requests": args.expected_requests,
|
|
"requests_recorded": len(rows),
|
|
"completed": len(successful),
|
|
"failed": len(rows) - len(successful),
|
|
"machine_wall_s": wall_s,
|
|
"machine_qps": len(successful) / wall_s if wall_s else 0.0,
|
|
"latency_mean_s": statistics.fmean(latencies) if latencies else 0.0,
|
|
"latency_p50_s": percentile(latencies, 0.50),
|
|
"latency_p95_s": percentile(latencies, 0.95),
|
|
"by_replica": by_replica,
|
|
"by_short_edge": by_short_edge,
|
|
"rows": rows,
|
|
}
|
|
args.output.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(json.dumps({key: value for key, value in summary.items() if key not in {"rows"}}, ensure_ascii=False, indent=2))
|
|
return int(len(rows) != args.expected_requests or len(successful) != len(rows))
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser()
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
p = sub.add_parser("run")
|
|
p.add_argument("--host", default="127.0.0.1")
|
|
p.add_argument("--port", type=int, required=True)
|
|
p.add_argument("--model", default="/data/hf_models/MiniMax-H3")
|
|
p.add_argument("--replica-index", type=int, required=True)
|
|
p.add_argument("--num-replicas", type=int, default=4)
|
|
p.add_argument("--prompt-file", type=Path, required=True)
|
|
p.add_argument("--resolutions", default="480,720,768,1080")
|
|
p.add_argument("--requests-per-resolution", type=int, default=4)
|
|
p.add_argument("--num-inference-steps", type=int, default=20)
|
|
p.add_argument("--warmup-inference-steps", type=int, default=5)
|
|
p.add_argument("--duration-seconds", type=float, default=5.0)
|
|
p.add_argument("--aspect-ratio", default="16:9")
|
|
p.add_argument("--flow-shift", type=float, default=12.0)
|
|
p.add_argument("--audio-flow-shift", type=float, default=3.0)
|
|
p.add_argument("--seed", type=int, default=1101)
|
|
p.add_argument("--submit-timeout", type=float, default=120.0)
|
|
p.add_argument("--poll-timeout", type=float, default=30.0)
|
|
p.add_argument("--poll-interval", type=float, default=1.0)
|
|
p.add_argument("--request-timeout", type=float, default=3600.0)
|
|
p.add_argument("--output", type=Path, required=True)
|
|
p = sub.add_parser("summarize")
|
|
p.add_argument("--input-dir", type=Path, required=True)
|
|
p.add_argument("--expected-requests", type=int, required=True)
|
|
p.add_argument("--output", type=Path, required=True)
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
return run(args) if args.command == "run" else summarize(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|