diff --git a/harness/loadtest.py b/harness/loadtest.py new file mode 100644 index 0000000..a1dfdce --- /dev/null +++ b/harness/loadtest.py @@ -0,0 +1,144 @@ +"""MCP parallelism test: N clients call read-only tools for D seconds; reports calls/s, latency and lock hits. + + python3 -m harness.loadtest [--levels 1 2 4 6 8] [--seconds 20] + +No object is created or changed. A "Concurrent call detected" answer is counted as a lock hit (no hidden retry), +then the client waits 0.3 s and calls again. Other runs on the same A4H change the numbers: read the result with that in mind. +""" +import argparse +import json +import os +import statistics +import threading +import time + +from .adt_client import load_env +from .mcp_client import McpClient + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +CALLS = [("sap_search_object", {"query": "CL_ABAP_CHAR_UTIL*", "objType": "CLAS"}), + ("sap_pull_source", {"objectType": "INTF", "objectName": "IF_SERIALIZABLE_OBJECT"}), + ("sap_object_structure", {"objectType": "CLAS", "objectName": "CL_ABAP_CHAR_UTILITIES"})] + + +HEAVY = [("sap_atc_run", {"objectType": "CLAS", "objectName": "CL_ABAP_CHAR_UTILITIES"}), + ("sap_syntax_check", {"objectType": "CLAS", "objectName": "CL_ABAP_CHAR_UTILITIES"})] + + +def write_worker(idx, end, out): + """create + push (activates) + delete a tiny class; latency = create + push.""" + from .runner import delete_uris + lat, locks, errors = [], 0, 0 + m = McpClient(timeout=300).open() + n = 0 + + def call(tool, args): + nonlocal locks + for _ in range(60): + body = {"jsonrpc": "2.0", "id": 1000 + n, "method": "tools/call", "params": {"name": tool, "arguments": args}} + res = m._post(body) + text = "\n".join(c.get("text", "") for c in res["result"].get("content", [])) + if "Concurrent call detected" in text: + locks += 1 + time.sleep(0.3) + continue + return text + return "lock retries exhausted" + try: + while time.time() < end: + n += 1 + name = f"ZLT_LOAD_W{idx}_{n:03d}" + src = (f"CLASS {name.lower()} DEFINITION PUBLIC FINAL CREATE PUBLIC.\n PUBLIC SECTION.\n" + " METHODS run RETURNING VALUE(rv) TYPE i.\nENDCLASS.\n" + f"CLASS {name.lower()} IMPLEMENTATION.\n METHOD run.\n rv = 1.\n ENDMETHOD.\nENDCLASS.\n") + t0 = time.time() + try: + c = call("sap_create_object", {"objectType": "CLAS", "objectName": name, "packageName": "$TMP", + "description": "load test"}) + p = call("sap_push_source", {"objectType": "CLAS", "objectName": name, "source": src}) + ok = '"success":true' in p.replace(" ", "") + (lat.append(time.time() - t0) if ok else None) + errors += 0 if ok else 1 + except Exception: # noqa: BLE001 + errors += 1 + try: + delete_uris([f"/sap/bc/adt/oo/classes/{name.lower()}"]) + except Exception: # noqa: BLE001 + pass + finally: + m.close() + out[idx] = {"calls": len(lat), "lat": lat, "locks": locks, "errors": errors} + + +def worker(idx, end, out): + if MODE == "write": + return write_worker(idx, end, out) + lat, locks, errors = [], 0, 0 + try: + m = McpClient(timeout=120).open() + except Exception as e: # noqa: BLE001 session limit (8) or server down + out[idx] = {"calls": 0, "lat": [], "locks": 0, "errors": 1, "open_error": repr(e)[:120]} + return + n = 0 + try: + while time.time() < end: + calls = HEAVY if MODE == "heavy" else CALLS + tool, args = calls[n % len(calls)] + n += 1 + t0 = time.time() + body = {"jsonrpc": "2.0", "id": n + 1, "method": "tools/call", "params": {"name": tool, "arguments": args}} + try: + res = m._post(body) + text = "\n".join(c.get("text", "") for c in res["result"].get("content", [])) if "result" in res else str(res) + except Exception: # noqa: BLE001 + errors += 1 + continue + if "Concurrent call detected" in text: + locks += 1 + time.sleep(0.3) + continue + lat.append(time.time() - t0) + finally: + m.close() + out[idx] = {"calls": len(lat), "lat": lat, "locks": locks, "errors": errors} + + +def level(n, seconds): + out, end = {}, time.time() + seconds + ts = [threading.Thread(target=worker, args=(i, end, out)) for i in range(n)] + t0 = time.time() + for t in ts: + t.start() + time.sleep(0.2) + for t in ts: + t.join() + wall = time.time() - t0 + lat = sorted(x for r in out.values() for x in r["lat"]) + calls = len(lat) + return {"clients": n, "calls": calls, "calls_per_s": round(calls / wall, 2), + "lat_median_s": round(statistics.median(lat), 2) if lat else None, + "lat_p95_s": round(lat[int(0.95 * len(lat))], 2) if lat else None, + "lock_hits": sum(r["locks"] for r in out.values()), "errors": sum(r["errors"] for r in out.values()), + "open_errors": [r["open_error"] for r in out.values() if r.get("open_error")]} + + +MODE = "light" + + +def main(): + load_env(os.path.join(ROOT, ".env")) + ap = argparse.ArgumentParser() + ap.add_argument("--levels", type=int, nargs="*", default=[1, 2, 4, 6, 8]) + ap.add_argument("--seconds", type=int, default=20) + ap.add_argument("--heavy", action="store_true", help="ATC and syntax check instead of light reads") + ap.add_argument("--write", action="store_true", help="create + activate + delete a tiny class (creates objects ZLT_LOAD_*)") + a = ap.parse_args() + global MODE + MODE = "write" if a.write else ("heavy" if a.heavy else "light") + for n in a.levels: + print(json.dumps(level(n, a.seconds)), flush=True) + time.sleep(3) + + +if __name__ == "__main__": + main()