from __future__ import annotations import json import subprocess import threading import time from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path ROOT = Path(__file__).resolve().parent RESULTS = ROOT / ".work" / "results" def process_rss() -> dict[str, dict[int, int]]: output = subprocess.run( ["ps", "-axo", "pid=,rss=,command="], check=True, capture_output=True, text=True, ).stdout result: dict[str, dict[int, int]] = {"renderer": {}, "gpu": {}} for line in output.splitlines(): pieces = line.strip().split(maxsplit=2) if len(pieces) != 3: continue pid_text, rss_text, command = pieces if "T3 Code (Alpha)" not in command: continue if "--type=renderer" in command and "Helper (Renderer)" in command: result["renderer"][int(pid_text)] = int(rss_text) * 1024 elif "--type=gpu-process" in command and "user-data-dir=/Users/admin/Library/Application Support/t3code" in command: result["gpu"][int(pid_text)] = int(rss_text) * 1024 return result class ProcessSampler: def __init__(self, run_id: str) -> None: self.run_id = run_id self.started_at = time.time() self.phase = "startup" self.samples: list[dict] = [] self.stop_event = threading.Event() self.thread = threading.Thread(target=self._run, daemon=True) self.thread.start() def _run(self) -> None: while not self.stop_event.is_set(): try: self.samples.append( {"time": time.time(), "phase": self.phase, **process_rss()} ) except Exception as error: # diagnostic sampling must not break OCR self.samples.append({"time": time.time(), "error": str(error)}) self.stop_event.wait(0.05) def mark(self, phase: str) -> None: self.phase = phase def finish(self) -> dict: self.stop_event.set() self.thread.join(timeout=2) usable = [sample for sample in self.samples if "renderer" in sample] if not usable: return {"samples": len(self.samples), "error": "no usable process samples"} baseline = usable[0] def summarize(kind: str) -> dict: baseline_by_pid = baseline[kind] peak_by_pid: dict[int, int] = {} first_by_pid: dict[int, int] = dict(baseline_by_pid) for sample in usable: for pid, rss in sample[kind].items(): first_by_pid.setdefault(pid, rss) peak_by_pid[pid] = max(peak_by_pid.get(pid, 0), rss) deltas = { pid: peak - first_by_pid.get(pid, peak) for pid, peak in peak_by_pid.items() } selected_pid = max(deltas, key=deltas.get) if deltas else None phases: dict[str, dict[str, int]] = {} if selected_pid: for sample in usable: rss = sample[kind].get(selected_pid) if rss is None: continue phase = sample.get("phase", "unknown") phase_summary = phases.setdefault( phase, {"startBytes": rss, "endBytes": rss, "peakBytes": rss}, ) phase_summary["endBytes"] = rss phase_summary["peakBytes"] = max(phase_summary["peakBytes"], rss) for phase_summary in phases.values(): phase_summary["peakDeltaBytes"] = ( phase_summary["peakBytes"] - phase_summary["startBytes"] ) return { "selectedPid": selected_pid, "baselineBytes": first_by_pid.get(selected_pid, 0) if selected_pid else 0, "peakBytes": peak_by_pid.get(selected_pid, 0) if selected_pid else 0, "peakDeltaBytes": deltas.get(selected_pid, 0) if selected_pid else 0, "allPeakDeltas": deltas, "phases": phases, } return { "scope": "T3 renderer with largest RSS delta plus shared T3 GPU process", "samples": len(usable), "durationSeconds": time.time() - self.started_at, "renderer": summarize("renderer"), "gpu": summarize("gpu"), } active_sampler: ProcessSampler | None = None sampler_lock = threading.Lock() class Handler(SimpleHTTPRequestHandler): def do_POST(self) -> None: global active_sampler length = int(self.headers.get("Content-Length", "0")) if length <= 0 or length > 2_000_000: self.send_error(400, "invalid request size") return payload = json.loads(self.rfile.read(length)) if self.path == "/benchmark-start": run_id = str(payload.get("runId", "")) if not run_id or "/" in run_id or ".." in run_id: self.send_error(400, "invalid run id") return with sampler_lock: if active_sampler: active_sampler.finish() active_sampler = ProcessSampler(run_id) self.send_response(204) self.end_headers() return if self.path == "/benchmark-mark": run_id = str(payload.get("runId", "")) phase = str(payload.get("phase", "")) if not phase or len(phase) > 80: self.send_error(400, "invalid benchmark phase") return with sampler_lock: sampler = active_sampler if not sampler or sampler.run_id != run_id: self.send_error(409, "benchmark sampler mismatch") return sampler.mark(phase) self.send_response(204) self.end_headers() return if self.path == "/benchmark-result": run_id = str(payload.get("runId", "")) with sampler_lock: sampler = active_sampler active_sampler = None if not sampler or sampler.run_id != run_id: self.send_error(409, "benchmark sampler mismatch") return payload["processMemory"] = sampler.finish() RESULTS.mkdir(parents=True, exist_ok=True) destination = RESULTS / f"{run_id}.json" destination.write_text( json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) response = json.dumps(payload["processMemory"]).encode() self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(response))) self.end_headers() self.wfile.write(response) return self.send_error(404) def end_headers(self) -> None: self.send_header("Cross-Origin-Opener-Policy", "same-origin") self.send_header("Cross-Origin-Embedder-Policy", "require-corp") self.send_header("Cross-Origin-Resource-Policy", "same-origin") self.send_header("Cache-Control", "no-store") super().end_headers() if __name__ == "__main__": ThreadingHTTPServer(("127.0.0.1", 8765), Handler).serve_forever()