| 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: |
| 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() |
|
|
