burst-test.py
2.15 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
"""压并发：12 条同时投给调度层，看三件事。
1. 派发会不会打偏——调度层是「读 health 挑最闲的」，同时提交时几个请求可能读到同一份快照。
顺序提交测不出这个，必须真并发。
2. queue_position 和 eta_seconds 在排队时说了什么。
3. 队列有没有上限、超了会怎样。
"""
import json, time, urllib.request
from collections import Counter
from concurrent.futures import ThreadPoolExecutor
GW = "http://127.0.0.1:8000"
N = 12
SPEC = {"prompt": "一只橘猫趴在洒满阳光的窗台上，慢慢眨眼", "frames": 22, "steps": 4, "turbo": True}
def call(path, data=None, timeout=60):
req = urllib.request.Request(GW + path, data=data,
headers={"Content-Type": "application/json"} if data else {})
with urllib.request.urlopen(req, timeout=timeout) as r:
return json.load(r)
def submit(i):
t = time.time()
job = call("/v1/videos", json.dumps(dict(SPEC, seed=100 + i)).encode())
return i, job, time.time() - t
t0 = time.time()
with ThreadPoolExecutor(max_workers=N) as pool:
results = sorted(pool.map(submit, range(N)))
print(f"12 条并发提交耗时 {time.time() - t0:.2f}s\n")
for i, job, dt in results:
print(f" #{i:<2} {job['backend'][-4:]} queue_position={job['queue_position']} "
f"eta={job['eta_seconds']}s ({dt:.2f}s)")
spread = Counter(job["backend"][-4:] for _, job, _ in results)
print(f"\n派发分布: {dict(sorted(spread.items()))} 理想是每个 {N // 4} 条")
ids = [job["id"] for _, job, _ in results]
print("\n--- 轮询")
while True:
time.sleep(30)
states = [call(f"/v1/videos/{jid}") for jid in ids]
c = Counter(s["status"] for s in states)
print(f" {dict(c)}", flush=True)
if all(s["status"] in ("done", "failed") for s in states):
break
total = time.time() - t0
print(f"\n12 条全部完成，墙钟 {total:.0f}s")
for s in sorted(states, key=lambda s: s["created_at"]):
took = s["finished_at"] - s["started_at"]
wait = s["started_at"] - s["created_at"]
print(f" {s['id'][:6]} {s['status']} 排队{wait:5.0f}s 执行{took:4.0f}s {s['backend'][-4:]}")