minicpm47woek_EPQ / sdq /experiments.py
HelloSun's picture
EPM: expert paging CPU inference for MiniCPM-V-4.7-35B-A3B Q4_K_M
12eb3a5 verified
Raw History Blame Contribute Delete
17.4 kB
#!/usr/bin/env python3
"""SDQ 最佳化實驗 harness:跑一組設定矩陣,產出可重現、可比較的結果。
設計原則
--------
1. **每次量測前丟掉模型檔的 page cache**(POSIX_FADV_DONTNEED),否則上一輪留在
核心記憶體的頁會讓 SSD 讀取量虛高、命中率虛高。
2. **每個設定跑 N 次取中位數**:單次結果抖動 ±10% 很常見,中位數才有意義。
3. **同時回報 KPI**:decode tok/s、cache 命中率、SSD 讀取量/token、峰值 RSS 與
本行程 VmSwap。VmSwap=0 僅表示該行程取樣為 0,不代表主機停用了 swap。
4. **自動檢查正確性指紋**:把每個實驗產生的第一個 token 記下來,
如果某個最佳化讓第一個 token 變了,代表它改變了模型行為 → 必須標記為失敗。
(速度不能拿正確性換,這是本專案的硬規則。)
5. 結果寫成 JSON + Markdown 表格,方便直接貼進 docs/。
用法
----
python -m sdq.experiments --model <gguf> --bin <epq-chat> --suite quick
python -m sdq.experiments --model <gguf> --bin <epq-chat> --suite full --repeat 3
python -m sdq.experiments ... --only thread,direct # 只跑某些實驗
"""
from __future__ import annotations
import argparse
import json
import os
import re
import resource
import statistics
import subprocess
import sys
import time
from pathlib import Path
REPO = Path(__file__).resolve().parent.parent
# 固定的量測條件:改這些會讓不同實驗之間無法比較
FIXED = {
"ctx": "4096",
# S5 起引擎的 n_ubatch 預設是 128(512 在 8 GB 預算下會讓長提示詞 bad_alloc,
# 見 docs/05 §5.1)。這裡必須跟著預設走,否則量到的不是我們要交付的那條路徑。
"ubatch": "128",
"temp": "0",
"n_predict": "48",
"ram_mb": "8192",
}
PROMPTS = [
"請用三句話說明什麼是 SSD、RAM、CPU 之間的資料分層。",
"Write a short paragraph about how mixture-of-experts models route tokens.",
"請列出五種日常會用到電腦的工作,並各用一句話說明。",
]
# ------------------------------------------------------------------ 實驗定義
# 每個實驗 = 一組環境變數 / CLI 參數的覆寫。None 代表沿用預設。
def _suite_quick() -> list[dict]:
"""先摸底:threads、I/O 模式、預取、arena、ubatch。成本低、風險低。"""
out = []
for t in (4, 6, 8, 10, 12, 16):
out.append({"name": f"threads={t}", "args": {"threads": str(t)}})
out += [
{"name": "prefetch=off", "args": {"no_prefetch": True}},
{"name": "fadvise=off", "env": {"SDQ_FADV_DONTNEED": "0"}},
{"name": "arena=2048MB", "args": {"arena_mb": "2048"}},
{"name": "arena=3072MB", "args": {"arena_mb": "3072"}},
{"name": "arena=4608MB", "args": {"arena_mb": "4608"}},
{"name": "ubatch=128", "args": {"ubatch": "128"}},
# OPT-1:Strata 式熱門 expert 靜態釘選(實測對本模型無效,保留作為對照)
# Needs a profile produced by tools/route_overlap.py; point SDQ_PROFILE_IN
# at your own file. Kept as a control even though it measured as no-op.
{"name": "hot-pin=2048", "args": {},
"env": {"SDQ_PROFILE_IN": str(REPO / "bench" / "expert_profile.txt"),
"SDQ_HOT_PIN": "2048"}},
# 排程模式對照
{"name": "modeB(partial=0)", "args": {}, "env": {"SDQ_PARTIAL_CAP_MB": "0"}},
]
return out
def _suite_io() -> list[dict]:
"""I/O 排程:一個 expert 的 gate/up/down 三段要不要同時讀。
這是 docs/05 的主線實驗。單一 pread 在本機約 7 ms(1.9 MB / 275 MB/s),
序列讀三段就是 ~21 ms 的延遲鏈,而且這條鏈的長度與執行緒數無關。
"""
out = [{"name": "baseline", "args": {}}]
for n in (4, 8, 12, 16, 24, 32, 48):
out.append({"name": f"io-threads={n}", "args": {},
"env": {"SDQ_IO_THREADS": str(n)}})
out.append({"name": "io-parallel=off", "args": {},
"env": {"SDQ_IO_PARALLEL": "0"}})
return out
def _suite_cache() -> list[dict]:
"""快取策略:這是命中率與 I/O 量的主戰場。"""
return [
{"name": "baseline", "args": {}},
{"name": "hot-pin=1024", "args": {}, "env": {"SDQ_HOT_PIN": "1024"}},
{"name": "hot-pin=3072", "args": {}, "env": {"SDQ_HOT_PIN": "3072"}},
{"name": "admit=2nd", "args": {}, "env": {"SDQ_ADMIT_MIN_HITS": "2"}},
{"name": "modeA(cap=256)", "args": {}, "env": {"SDQ_PARTIAL_CAP_MB": "256"}},
]
def _suite_slot() -> list[dict]:
"""槽位配置:大小兩種槽位 vs 全部照最大的配。
背景:這個模型 37 層的 expert 是 1900544 B,層 34/38/39 是 2039808 B(down 用 Q6_K)。
舊做法所有槽位都照最大的配 → arena 有 6.8% 是浪費掉的,而 arena 在 8 GB 預算裡
是最貴的資源(實測 +5% 槽位 ≈ +8~9% tok/s)。
SDQ_FORCE_UNIFORM_SLOT=1 回到舊行為當對照。
"""
return [
{"name": "baseline(大小兩種槽位)", "args": {}},
{"name": "uniform(全部照大槽位)", "args": {}, "env": {"SDQ_FORCE_UNIFORM_SLOT": "1"}},
]
def _suite_arena() -> list[dict]:
"""arena 大小掃描:驗證「槽位數 → tok/s」的斜率,並確認 8 GB 預算內的上限。"""
out = [{"name": "baseline", "args": {}}]
for mb in (3072, 3584, 4096):
out.append({"name": f"arena={mb}MB", "args": {"arena_mb": str(mb)}})
return out
def _suite_pf() -> list[dict]:
"""跨層預取:layer L 開頭就替 layer L+1 排預取。
背景:end_token() 是在一個 token 的 40 層「全部算完」之後才排預取的,
那時沒有任何計算可以重疊 → 實測 prefetch 使用率只有 0~8 %,等於整條路徑是關的。
這一組量的是「把預取搬到每層開頭」有沒有價值。
"""
return [
{"name": "baseline(層開頭預取)", "args": {}, "env": {"SDQ_PREFETCH_LAYER_AHEAD": "1"}},
{"name": "layer-ahead=off", "args": {}, "env": {"SDQ_PREFETCH_LAYER_AHEAD": "0"}},
{"name": "prefetch=off", "args": {}, "env": {"SDQ_PREFETCH": "0"}},
{"name": "pf(queue=256,stale=1)", "args": {},
"env": {"SDQ_PREFETCH_LAYER_AHEAD": "1", "SDQ_PREFETCH_QUEUE": "256",
"SDQ_PREFETCH_STALE_TOKENS": "1"}},
]
# The former "spec" suite drove speculative decoding through SDQ_NGRAM_DRAFT, an
# env var that no longer exists anywhere in cpp/. Prompt-lookup n-gram drafting was
# measured at zero accepted tokens on this model (see docs/NEGATIVE_RESULTS.md §4)
# and MTP has its own entry point (cpp/sdq_mtp.cpp -> epq-mtp, see docs/MTP.md),
# so the suite was dead code that would silently no-op rather than measure anything.
SUITES = {"quick": _suite_quick, "io": _suite_io, "cache": _suite_cache,
"slot": _suite_slot, "arena": _suite_arena, "pf": _suite_pf}
# ------------------------------------------------------------------ 執行
def drop_cache(model: str) -> None:
subprocess.run([sys.executable, str(REPO / "tools" / "drop_model_cache.py"), model],
check=False, capture_output=True)
def _limit(ram_mb: str):
def f():
n = int(ram_mb) * 1024 * 1024
resource.setrlimit(resource.RLIMIT_DATA, (n, n))
return f
def first_generated_token(stdout: str) -> str:
"""取出「第一個生成 token」的**空白正規化**指紋,用來偵測輸出有沒有改變。
為什麼要把所有空白都拿掉再取前 24 個字元:
模型的輸出可能以換行開頭,而 stdout 是被分塊寫出的,所以同一次生成在不同
執行裡「assistant: 後面第一個非空行」的位置不一定相同(例如一次是
`assistant: <think>`、另一次是 `assistant:` 換行後才出現 `<think>`)。
如果指紋取「換行前的原始片段」,**完全相同的輸出**會被判成不一致。
量測(2026-10-04,本機):同一提示詞連跑 4 次(含 SDQ_FADV_DONTNEED=1
並丟掉 20.6 GiB page cache),stdout 長度都是 294 bytes、指紋都是 `<think>`;
`tools/determinism_check.py` 另外證明 48 token × 248320 vocab 的 logits
**逐位元組一致** → 分頁路徑是決定性的,過去那些 ⚠️ 都是假警報。
抓不到內容時回空字串,呼叫端不去比對空指紋(寧可少檢查也不要假警報)。
"""
lines = stdout.splitlines()
started = False
chunks: list[str] = []
for line in lines:
if not started:
if line.startswith("assistant:"):
started = True
chunks.append(line[len("assistant:"):])
continue
if line.startswith("[sdq]"):
continue
chunks.append(line)
text = re.sub(r"\s+", "", "".join(chunks))
return text[:24]
def run_once(binary: str, model: str, exp: dict, prompt: str, idx: int) -> dict:
stats_file = Path("/tmp") / f"sdqexp_{os.getpid()}_{idx}.jsonl"
if stats_file.exists():
stats_file.unlink()
cmd = [binary, "-m", model, "-p", prompt, "-n", FIXED["n_predict"],
"-c", FIXED["ctx"], "-t", str(exp.get("args", {}).get("threads", 8)),
"-b", str(exp.get("args", {}).get("ubatch", FIXED["ubatch"])), "--temp", FIXED["temp"],
"--no-chat", "--no-interactive", "--stats-file", str(stats_file),
"--ram-budget-mb", FIXED["ram_mb"]]
if exp.get("args", {}).get("arena_mb"):
cmd += ["--arena-mb", str(exp["args"]["arena_mb"])]
if exp.get("args", {}).get("no_prefetch"):
cmd += ["--no-prefetch"]
env = dict(os.environ)
env.update(exp.get("env", {}))
env.setdefault("SDQ_FADV_DONTNEED", "1")
drop_cache(model)
t0 = time.time()
try:
proc = subprocess.run(cmd, capture_output=True, text=True, env=env,
preexec_fn=_limit(FIXED["ram_mb"]), timeout=900)
except subprocess.TimeoutExpired:
return {"error": "timeout"}
wall = time.time() - t0
if proc.returncode != 0:
return {"error": f"rc={proc.returncode}", "stderr": proc.stderr[-800:]}
stats = {}
if stats_file.exists():
for line in stats_file.read_text().splitlines():
rec = json.loads(line)
stats[rec["phase"]] = rec
stats_file.unlink(missing_ok=True)
dec = stats.get("decode") or {}
tok_s = 0.0
m = re.search(r"decode: (\d+) tok / ([\d.]+)s = ([\d.]+) tok/s", proc.stderr)
if m:
tok_s = float(m.group(3))
peak_rss = 0
vmswap = 0
m = re.search(r"峰值 (\d+) MB / (?:swap|本行程 VmSwap) (\d+) MB", proc.stderr)
if m:
peak_rss, vmswap = int(m.group(1)), int(m.group(2))
# 正確性指紋:第一個生成 token(用來偵測「最佳化偷偷改變了模型行為」)
first_tok = first_generated_token(proc.stdout)
return {
"tok_s": tok_s,
"wall_s": round(wall, 2),
"hit_rate": dec.get("hit_rate"),
"ssd_mb_per_tok": (dec.get("ssd_bytes_per_token") or 0) / 2 ** 20,
"peak_rss_mb": peak_rss,
"vmswap_mb": vmswap,
"n_slots": dec.get("n_slots"),
"evictions": dec.get("evictions"),
"prefetch_used_rate": dec.get("prefetch_used_rate"),
"first_tok": first_tok,
"text": proc.stdout.strip()[-260:],
}
def run_exp(binary: str, model: str, exp: dict, repeat: int, prompts: list[str]) -> dict:
runs = []
for r in range(repeat):
res = run_once(binary, model, exp, prompts[r % len(prompts)], r)
if res.get("error"):
return {"name": exp["name"], "error": res["error"], "stderr": res.get("stderr", "")}
runs.append(res)
ok = [r for r in runs if r["tok_s"] > 0]
if not ok:
return {"name": exp["name"], "error": "沒有任何一次量到 tok/s"}
med = lambda k: statistics.median([r[k] for r in ok if r.get(k) is not None]) if any(r.get(k) is not None for r in ok) else None
# 只比較「同一個提示詞」的重複執行:實驗會輪替提示詞,跨提示詞比字串沒有意義。
# **空指紋不參與比較**:抓不到第一個 token 時回 "",那是 stdout 分塊造成的,
# 不是輸出不一致(見 first_generated_token 的說明)。寧可少檢查也不要假警報。
by_prompt: dict[str, set] = {}
for i, r in enumerate(ok):
if not r["first_tok"]:
continue
by_prompt.setdefault(PROMPTS[i % len(PROMPTS)], set()).add(r["first_tok"])
unstable = [p[:20] for p, v in by_prompt.items() if len(v) > 1]
# 有沒有真的比到東西(沒比到就不能宣稱「輸出穩定」)
checked = sum(len(v) for v in by_prompt.values()) - len(by_prompt)
return {
"name": exp["name"],
"args": exp.get("args", {}), "env": exp.get("env", {}),
"tok_s": round(med("tok_s"), 3),
"tok_s_all": [r["tok_s"] for r in ok],
"hit_rate": round(med("hit_rate"), 2) if med("hit_rate") is not None else None,
"ssd_mb_per_tok": round(med("ssd_mb_per_tok"), 1) if med("ssd_mb_per_tok") is not None else None,
"peak_rss_mb": med("peak_rss_mb"),
"vmswap_mb": med("vmswap_mb"),
"n_slots": med("n_slots"),
"evictions": med("evictions"),
"prefetch_used_rate": med("prefetch_used_rate"),
"stable_output": not unstable,
"stable_checked": checked, # 實際有幾組「同提示詞重複」可比
"unstable_prompts": unstable,
"first_tok": ok[0]["first_tok"],
"runs": len(ok),
}
# ------------------------------------------------------------------ 報告
def to_markdown(results: list[dict]) -> str:
head = ("| 實驗 | decode tok/s | 命中率 | SSD MB/token | 峰值 RSS | swap | 槽數 | 淘汰 | 預取使用率 | 輸出穩定 |")
sep = "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | :---: |"
lines = [head, sep]
base = next((r for r in results if r["name"] in ("baseline", "threads=8")), None)
bt = base["tok_s"] if base else None
for r in results:
if r.get("error"):
lines.append(f"| {r['name']} | ❌ {r['error']} | | | | | | | | |")
continue
delta = ""
if bt:
d = (r["tok_s"] - bt) / bt * 100
delta = f" ({d:+.1f}%)"
lines.append(
f"| {r['name']} | {r['tok_s']}{delta} | {r['hit_rate']}% | {r['ssd_mb_per_tok']} | "
f"{r['peak_rss_mb']} MB | {r['vmswap_mb']} MB | {r['n_slots']} | {r['evictions']} | "
f"{r['prefetch_used_rate']}% | "
f"{'✅' if r['stable_output'] else ('⚠️不一致' if r.get('stable_checked') else '—未比到')} |")
return "\n".join(lines)
def main() -> int:
ap = argparse.ArgumentParser(description="SDQ 最佳化實驗")
ap.add_argument("--model", required=True)
ap.add_argument("--bin", default=str(REPO.parent / "llama.cpp" / "build" / "bin" / "epq-chat"),
help="epq-chat 路徑(預設 <repo>/../llama.cpp/build/bin/epq-chat)")
ap.add_argument("--suite", default="quick", choices=sorted(SUITES) + ["all"])
ap.add_argument("--only", default="", help="只跑名稱包含這些關鍵字的實驗(逗號分隔)")
ap.add_argument("--repeat", type=int, default=2)
ap.add_argument("--out", default=str(REPO / "bench" / "results"))
ap.add_argument("--label", default="")
args = ap.parse_args()
exps: list[dict] = []
names = sorted(SUITES) if args.suite == "all" else [args.suite]
for n in names:
exps += SUITES[n]()
# 去掉重複名稱(不同 suite 可能有 baseline)
seen, uniq = set(), []
for e in exps:
if e["name"] in seen:
continue
seen.add(e["name"])
uniq.append(e)
if args.only:
keys = [k.strip() for k in args.only.split(",") if k.strip()]
uniq = [e for e in uniq if any(k in e["name"] for k in keys)]
print(f"[exp] 模型:{args.model}")
print(f"[exp] 執行檔:{args.bin}")
print(f"[exp] 實驗數:{len(uniq)},每個跑 {args.repeat} 次取中位數")
print(f"[exp] 固定條件:{FIXED}")
print()
results = []
for e in uniq:
print(f"[exp] → {e['name']} ...", end=" ", flush=True)
r = run_exp(args.bin, args.model, e, args.repeat, PROMPTS)
results.append(r)
if r.get("error"):
print(f"❌ {r['error']}")
else:
print(f"{r['tok_s']} tok/s 命中率 {r['hit_rate']}% SSD {r['ssd_mb_per_tok']} MB/tok"
+ ("" if r["stable_output"] or not r.get("stable_checked")
else " ⚠️ 輸出不一致"))
Path(args.out).mkdir(parents=True, exist_ok=True)
label = args.label or args.suite
jf = Path(args.out) / f"exp_{label}.json"
jf.write_text(json.dumps({"fixed": FIXED, "repeat": args.repeat,
"results": results}, ensure_ascii=False, indent=1), encoding="utf-8")
md = to_markdown(results)
mf = Path(args.out) / f"exp_{label}.md"
mf.write_text(md + "\n", encoding="utf-8")
print()
print(md)
print()
print(f"[exp] JSON → {jf}")
print(f"[exp] Markdown → {mf}")
return 0
if __name__ == "__main__":
sys.exit(main())