psxrecomp/tools/load_probe.py

#!/usr/bin/env python3
"""load_probe.py — E0 load-window decomposition (LOAD_TIME_ZERO L1.0).

Passive ring-buffer consumer: samples always-on counters while the game
free-runs, then locates completed CD load bursts (cdrom_bursts ring) inside
the monitored span and reports, per burst window:

  guest side: cycles, effective Mcyc/s + multiplier vs 1x (33.8688 Mcyc/s),
              sectors delivered, sector rate, disc divisor, frames
  host side:  phase_profile shares (interp / native-overlay / static / gpu /
              host-other / exception) over the covering seconds,
              phase_hot top overlay funcs (delta), dirty_ram per_pc top (delta)

No arming, no pausing, no stepping — everything read from rings after the
fact. Usage:

  python tools/load_probe.py --port 4470 run --secs 90
      Monitor for N seconds (trigger loads by playing / TCP input in
      parallel), then report every burst that completed in the span.

  python tools/load_probe.py --port 4470 snap
      One-shot dump of the counters this probe uses (sanity check).
"""
import argparse, json, socket, sys, time

X1_MCYC_PER_S = 33.8688  # NTSC PSX CPU clock, guest Mcyc per real second at 1x


class Client:
    """One-shot connection per command (the server aborts a second request
    on the same socket — matches debug_client.py's connect/send/close)."""

    def __init__(self, host, port, timeout=8.0):
        self.host, self.port, self.timeout = host, port, timeout
        self.next_id = 1

    def cmd(self, name, **params):
        req = {"cmd": name, "id": self.next_id}
        req.update(params)
        self.next_id += 1
        self.sock = socket.create_connection((self.host, self.port),
                                             timeout=self.timeout)
        self.buf = b""
        self.sock.sendall((json.dumps(req) + "\n").encode())
        # Responses may span multiple lines (e.g. cdrom_bursts emits one
        # record per line): accumulate until the buffer parses as one JSON
        # object, not until the first newline.
        while True:
            text = self.buf.decode(errors="replace").strip()
            if text:
                try:
                    obj, _end = json.JSONDecoder().raw_decode(text)
                    self.sock.close()
                    return obj
                except json.JSONDecodeError:
                    pass
            chunk = self.sock.recv(65536)
            if not chunk:
                raise ConnectionError("server closed")
            self.buf += chunk


def phot_map(resp):
    return {e["addr"]: e["samples"] for e in resp.get("top", [])}


def perpc_map(resp):
    return {e["pc"]: e["insns"] for e in resp.get("per_pc", [])}


def delta_top(before, after, k=10):
    d = {a: after[a] - before.get(a, 0) for a in after}
    items = [(a, v) for a, v in d.items() if v > 0]
    items.sort(key=lambda kv: -kv[1])
    return items[:k]


def snap(c, retries=3):
    # freeze_check occasionally returns a short/err response; retry rather
    # than kill a long monitoring run over one bad sample.
    for attempt in range(retries):
        try:
            fz = c.cmd("freeze_check", window=1)
            return {
                "t": time.monotonic(),
                "wall": time.time(),
                "cycles": int(fz["psx_cycle_count"]),
                "frame": int(fz["frame_count"]),
                "dirty_insns": int(fz["dirty_ram_insns"]),
            }
        except (KeyError, ValueError, json.JSONDecodeError):
            if attempt == retries - 1:
                raise
            time.sleep(0.1)


def run(c, secs, tick, out):
    print(f"[load_probe] monitoring {secs}s (tick {tick}s) — trigger loads now",
          flush=True)
    t0 = time.monotonic()
    bursts0 = c.cmd("cdrom_bursts", count=1).get("total", 0)
    phot0 = phot_map(c.cmd("phase_hot", top=64))
    stat0 = phot_map(c.cmd("phase_hot", set="static", top=64))
    perpc0 = perpc_map(c.cmd("dirty_ram_stats"))
    series = [snap(c)]
    while time.monotonic() - t0 < secs:
        time.sleep(tick)
        series.append(snap(c))
    phot1 = phot_map(c.cmd("phase_hot", top=64))
    stat1 = phot_map(c.cmd("phase_hot", set="static", top=64))
    perpc1 = perpc_map(c.cmd("dirty_ram_stats"))
    bursts = c.cmd("cdrom_bursts", count=64)
    span_s = series[-1]["t"] - series[0]["t"]

    # Whole-span summary (context for the per-burst numbers).
    dcyc = series[-1]["cycles"] - series[0]["cycles"]
    rate = dcyc / span_s / 1e6
    print(f"\n=== span: {span_s:.1f}s  guest {dcyc/1e6:.0f} Mcyc  "
          f"{rate:.1f} Mcyc/s  ({rate / X1_MCYC_PER_S:.2f}x)  "
          f"new bursts: {bursts.get('total', 0) - bursts0}")

    # Frame -> monotonic time mapping from the series (piecewise linear).
    def frame_to_t(fr):
        for a, b in zip(series, series[1:]):
            if a["frame"] <= fr <= b["frame"]:
                if b["frame"] == a["frame"]:
                    return a["t"]
                f = (fr - a["frame"]) / (b["frame"] - a["frame"])
                return a["t"] + f * (b["t"] - a["t"])
        return None

    def series_at(tm):  # nearest-sample counters at monotonic time tm
        return min(series, key=lambda s: abs(s["t"] - tm))

    report = {"span_s": span_s, "span_mcyc_per_s": rate,
              "bursts": [], "phot_delta": delta_top(phot0, phot1),
              "static_delta": delta_top(stat0, stat1, 15),
              "perpc_delta": delta_top(perpc0, perpc1)}

    now_t = time.monotonic()
    for b in bursts.get("bursts", []):
        ta, tb = frame_to_t(b["start_frame"]), frame_to_t(b["end_frame"])
        if ta is None or tb is None:
            continue  # burst outside the monitored span
        sa, sb = series_at(ta), series_at(tb)
        wall = tb - ta
        gcyc = sb["cycles"] - sa["cycles"]
        gint = sb["dirty_insns"] - sa["dirty_insns"]
        eff = gcyc / wall / 1e6 if wall > 0 else 0.0
        # phase_profile ring is per-second, 64s deep: query the window of
        # seconds that covers this burst, measured back from "now".
        back_end = int(now_t - tb)
        need = int(wall) + 2
        ph = c.cmd("phase_profile", window=min(back_end + need, 62))
        rec = {
            "frames": b["frames"], "sectors": b["sectors"],
            "rate_sps": b["rate"], "divisor": b["divisor"],
            "wall_s": round(wall, 2),
            "guest_mcyc": round(gcyc / 1e6, 1),
            "eff_mcyc_per_s": round(eff, 1),
            "multiplier_vs_1x": round(eff / X1_MCYC_PER_S, 2),
            "guest_s_at_1x": round(gcyc / (X1_MCYC_PER_S * 1e6), 2),
            "interp_insns": gint,
            "phase_window": ph,
        }
        report["bursts"].append(rec)
        print(f"\n--- burst: {b['sectors']} sectors, {b['frames']} frames, "
              f"{wall:.2f}s wall")
        print(f"    guest {gcyc/1e6:.0f} Mcyc = {rec['guest_s_at_1x']}s at 1x"
              f" -> {rec['multiplier_vs_1x']}x effective")
        print(f"    interp insns in window: {gint}")
        shares = {k.replace("_share", ""): v for k, v in ph.items()
                  if k.endswith("_share")}
        print(f"    phase shares (covering secs, approximate): {shares}")

    if report["static_delta"]:
        tot = sum(v for _, v in report["static_delta"]) or 1
        print("\ntop STATIC funcs by wall samples (span delta):")
        for a, v in report["static_delta"]:
            print(f"    {a}  {v}  ({100.0*v/tot:.1f}% of top)")
    if report["phot_delta"]:
        print("\ntop native-overlay funcs by wall samples (span delta):")
        for a, v in report["phot_delta"]:
            print(f"    {a}  {v}")
    if report["perpc_delta"]:
        print("\ntop interp PCs by insns (span delta):")
        for a, v in report["perpc_delta"]:
            print(f"    {a}  {v}")

    if out:
        with open(out, "w") as f:
            json.dump(report, f, indent=1)
        print(f"\n[load_probe] report -> {out}")


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--host", default="127.0.0.1")
    ap.add_argument("--port", type=int, default=4470)
    sub = ap.add_subparsers(dest="mode", required=True)
    r = sub.add_parser("run")
    r.add_argument("--secs", type=float, default=90)
    r.add_argument("--tick", type=float, default=0.25)
    r.add_argument("--out", default=None)
    sub.add_parser("snap")
    args = ap.parse_args()

    c = Client(args.host, args.port)
    if args.mode == "snap":
        for cmd, kw in (("freeze_check", {"window": 1}),
                        ("cdrom_bursts", {"count": 4}),
                        ("phase_profile", {"window": 10}),
                        ("phase_hot", {"top": 10})):
            resp = c.cmd(cmd, **kw)
            print(f"== {cmd}: {json.dumps(resp)[:600]}")
    else:
        run(c, args.secs, args.tick, args.out)


if __name__ == "__main__":
    main()