DCU_ECAT_Master/cadence_diag.py

#!/usr/bin/env python3
"""归因"某一拍为什么超时"：把 PDO 时间线与 GC / HTTP 事件时间线做交集。

背景：250us 和 200us 档的达成率都是 96~97%，几乎不随档位变化 —— 丢拍不像
"总线差几十 us"这种比例性原因，更像固定频率的事件(GC 停止世界、HTTP 线程
持 GIL 序列化快照、系统杂活)一撞就是一整个时隙。本脚本从 /api/motor/dbg
拉三条原始时间线，对每一个超时拍给出责任归属。

用法(目标服务必须已在跑，本脚本自己不会拉服务)：
  python cadence_diag.py --period 0.25 --sample 6
  python cadence_diag.py --period 0.2 --gc off        # A/B：关掉 GC 看超时拍是否消失
  python cadence_diag.py --period 0.2 --gc on --hammer 2   # 再叠 2 路 /motor/state 轮询

判读：
  超时拍 = loop_end-t0(work) 超过设定周期的拍；它们正是网格跳掉的拍。
  归因窗口 = 上一拍 loop_end → 本拍 loop_end；GC/HTTP 事件落在窗口内即认领。
  "裸超时"占大头才说明是内核/中断层面的抢占，而不是进程内流程问题。
"""
import argparse
import json
import statistics
import threading
import time
import urllib.request

BASE_DEFAULT = "http://127.0.0.1:8090"


def api(base, path, req):
    r = urllib.request.Request(base + "/api/" + path,
                               data=json.dumps(req).encode(),
                               headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(r, timeout=30) as resp:
        return json.loads(resp.read().decode())


def hammer_worker(base, stop, path="motor/state"):
    while not stop.is_set():
        try:
            api(base, path, {})
        except Exception:
            pass


def classify(beats, gc_ev, http_ev, period_us):
    """给每个超时拍贴标签。返回 (late列表, 计数统计)。beats: 按 t0 排序的样本。"""
    counts = {"GC": 0, "HTTP": 0, "GC+HTTP": 0, "裸超时": 0}
    late = []
    for i in range(1, len(beats)):
        t0, ts, tm, tr, te, k = beats[i]
        work = (te - t0) * 1e6
        if work <= period_us:
            continue
        w0, w1 = beats[i - 1][4], te          # 上一拍结束 → 本拍结束
        g = [e for e in gc_ev if w0 <= e[0] <= w1 and e[1] > 20.0]
        h = [x for x in http_ev if x[0] < w1 and x[1] > w0 and (x[1] - x[0]) * 1e6 > 20.0]
        if g and h:
            tag = "GC+HTTP"
        elif g:
            tag = "GC(gen%d %.0fus)" % (g[-1][2], g[-1][1])
            counts["GC"] += 1
        elif h:
            tag = "HTTP(%s %.0fus)" % (h[-1][2].replace("/api/", ""), (h[-1][1] - h[-1][0]) * 1e6)
            counts["HTTP"] += 1
        else:
            tag = "裸超时"
            counts["裸超时"] += 1
        if g and h:
            counts["GC+HTTP"] += 1
        late.append((work, k, tag, (t0 - beats[i - 1][4]) * 1e6))
    return late, counts


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--base", default=BASE_DEFAULT)
    ap.add_argument("--period", type=float, required=True, help="周期 ms(0.25=250us)")
    ap.add_argument("--sample", type=float, default=6.0, help="设周期后预热秒数，之后才取窗口")
    ap.add_argument("--gc", choices=["on", "off", "keep"], default="keep")
    ap.add_argument("--hammer", type=int, default=0, help="叠加 N 路 /motor/state 轮询线程")
    args = ap.parse_args()
    per_us = args.period * 1000.0

    if args.gc != "keep":
        r = api(args.base, "motor/dbg", {"action": "gc", "enable": args.gc == "on"})
        print(f"GC → {r.get('gc_enabled')}")

    r = api(args.base, "motor/period", {"period_ms": args.period})
    print("period:", r.get("log"))
    if r.get("spin") is False and args.period < 1:
        print("!! 自旋没生效，这组数据没有意义")
        return 1

    stop = threading.Event()
    threads = [threading.Thread(target=hammer_worker, args=(args.base, stop), daemon=True)
               for _ in range(args.hammer)]
    for t in threads:
        t.start()
    try:
        time.sleep(args.sample)
        d = api(args.base, "motor/dbg", {})
    finally:
        stop.set()
        for t in threads:
            t.join(timeout=1)

    beats = sorted(d["cyc"])[:-50]        # 尾部几十拍被 dump 自身的 GIL 占用污染，丢弃
    if len(beats) < 100:
        print("样本不足", len(beats))
        return 1
    iv = [(beats[i][0] - beats[i - 1][0]) * 1e6 for i in range(1, len(beats))]
    hz = 1e6 / statistics.fmean(iv)
    work = sorted((e - t) * 1e6 for t, _, _, _, e, _ in beats)
    print(f"\n窗口 {len(beats)} 拍 ≈ {sum(iv) / 1e6:.2f}s | 周期 {per_us:.0f}us | "
          f"实测 {hz:.0f}Hz (达成 {hz / (1e6 / per_us) * 100:.1f}%) | "
          f"跳拍合计 {sum(k for *_, k in beats)}")
    gc_in_win = [e[2] for e in d["gc_events"] if beats[0][0] <= e[0] <= beats[-1][4]]
    print(f"work p50/p90/max = {work[len(work) // 2]:.0f}/{work[int(len(work) * .9)]:.0f}/"
          f"{work[-1]:.0f}us | 窗口内 GC g0/g1/g2 = "
          f"{gc_in_win.count(0)}/{gc_in_win.count(1)}/{gc_in_win.count(2)}")

    # 事件按窗口时间过滤，再交集
    w0, w1 = beats[0][0], beats[-1][4]
    gc_ev = [e for e in d["gc_events"] if w0 <= e[0] <= w1]
    http_ev = [x for x in d["http"] if x[1] >= w0 and x[0] <= w1]
    late, counts = classify(beats, gc_ev, http_ev, per_us)
    n_late = len(late)
    print(f"超时拍 {n_late}/{len(beats)} = {n_late / len(beats) * 100:.2f}% | "
          + " | ".join(f"{k}: {v}" for k, v in counts.items()))
    if late:
        lateness = sorted(x[0] for x in late)
        print(f"超时拍的 work: 中位 {statistics.median(lateness):.0f}us, "
              f"p90 {lateness[int(len(lateness) * .9)]:.0f}us, max {lateness[-1]:.0f}us")
        print("最狠的 8 拍:")
        for work_us, k, tag, gap in sorted(late, reverse=True)[:8]:
            print(f"   work={work_us:7.0f}us 跳{k}拍 上游空闲{gap:6.0f}us  {tag}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())