Function run_profile

tools/loadtest.py:679–729

(args: argparse.Namespace)

def run_profile(args: argparse.Namespace) -> dict:
    import os

    os.environ["DB_PATH"] = args.db
    from explainshell.store import Store
    from explainshell.web.views import explain_cmd

    store = Store(args.db)

    rng = random.Random(args.seed)
    cmds = [gen_bot_cmd(rng) for _ in range(args.n)]
    distro_preference = [("ubuntu", "26.04"), ("arch", "rolling")]

    for c in cmds[:3]:
        try:
            explain_cmd(c, store, distro_preference=distro_preference)
        except Exception as e:
            print(f"warmup error: {type(e).__name__}: {e}")

    pr = cProfile.Profile()
    per_call: list[float] = []
    errors = 0
    pr.enable()
    for c in cmds:
        t0 = time.perf_counter()
        try:
            explain_cmd(c, store, distro_preference=distro_preference)
            per_call.append((time.perf_counter() - t0) * 1000.0)
        except Exception:
            errors += 1
    pr.disable()

    st = pcts(per_call)
    print(f"explain_cmd calls: n={len(per_call)} errors={errors}")
    if st:
        print(
            f"wall ms: p50={st['p50']:.1f}  p95={st['p95']:.1f}  "
            f"p99={st['p99']:.1f}  max={st['max']:.1f}  mean={st['mean']:.1f}"
        )

    s = io.StringIO()
    pstats.Stats(pr, stream=s).sort_stats("cumulative").print_stats(40)
    print("\n=== cProfile top (cumulative) ===\n")
    print(s.getvalue())

    s = io.StringIO()
    pstats.Stats(pr, stream=s).sort_stats("tottime").print_stats(30)
    print("\n=== cProfile top (self time) ===\n")
    print(s.getvalue())

    return {"n": len(per_call), "errors": errors, "latency_ms": st}


# --- CLI --------------------------------------------------------------------