Investigate host overhead before changing GPU settings

"""Rank actual host operator records without inferring GPU utilization."""

import json, math
from pathlib import Path


def rank_operators(rows):
    if not rows:
        raise ValueError("Operator records are required")
    if any(
        not math.isfinite(r["self_cpu_us"]) or r["self_cpu_us"] < 0 or r["calls"] < 1
        for r in rows
    ):
        raise ValueError("Invalid host operator record")
    return sorted(rows, key=lambda r: r["self_cpu_us"], reverse=True)


def main():
    record = json.loads(Path("profile-run.json").read_text())
    print(
        json.dumps(
            {
                "execution_mode": "offline analysis of actual CPU profiler events around Metal model execution",
                "profiled_wall_ms": record["profiled_wall_ms"],
                "top_host_operators": rank_operators(record["operators"]),
                "limits": "Host operator time is not GPU kernel time, GPU utilization or device bandwidth. Profiling changes execution.",
            },
            indent=2,
        )
    )


if __name__ == "__main__":
    main()

"""Collect actual CPU operator events around synchronized local model forwards on Metal."""

import argparse
import hashlib
import json
import platform
import time
from pathlib import Path


def main():
    p = argparse.ArgumentParser()
    p.add_argument("--model-path", required=True)
    p.add_argument("--output", required=True)
    a = p.parse_args()
    import torch, transformers
    from transformers import AutoTokenizer, AutoModelForCausalLM

    tokenizer = AutoTokenizer.from_pretrained(a.model_path, local_files_only=True)
    model = (
        AutoModelForCausalLM.from_pretrained(
            a.model_path,
            local_files_only=True,
            dtype=torch.float32,
            attn_implementation="eager",
        )
        .eval()
        .to("mps")
    )
    ids = tokenizer(
        "A code assistant retrieves relevant functions before generating a patch.",
        return_tensors="pt",
    )["input_ids"].to("mps")
    with torch.inference_mode():
        model(ids, use_cache=True)
    torch.mps.synchronize()
    start = time.perf_counter()
    with torch.profiler.profile(
        activities=[torch.profiler.ProfilerActivity.CPU], record_shapes=True
    ) as prof:
        with torch.inference_mode():
            result = model(ids, use_cache=True)
            torch.mps.synchronize()
    elapsed = (time.perf_counter() - start) * 1000
    rows = sorted(
        prof.key_averages(), key=lambda x: x.self_cpu_time_total, reverse=True
    )[:12]
    result = {
        "execution_mode": "fresh local inference; CPU operator profiler around Metal execution",
        "environment": {
            "torch": torch.__version__,
            "transformers": transformers.__version__,
            "platform": platform.platform(),
            "python": platform.python_version(),
        },
        "model_id": "Qwen/Qwen2.5-0.5B-Instruct",
        "model_revision": "7ae557604adf67be50417f59c2c2f167def9a775",
        "source_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
        "profiled_wall_ms": round(elapsed, 3),
        "operators": [
            {
                "name": x.key,
                "calls": x.count,
                "self_cpu_us": round(x.self_cpu_time_total, 3),
            }
            for x in rows
        ],
        "limits": "CPU operator durations include dispatch and synchronization overhead, not GPU kernel duration or occupancy. Profiling changes runtime; compare unprofiled repetitions separately.",
    }
    Path(a.output).write_text(json.dumps(result, indent=2) + "\n")


if __name__ == "__main__":
    main()

import unittest
from lab import rank_operators


class ProfileTests(unittest.TestCase):
    def test_rank_uses_actual_host_duration(self):
        rows = [
            {"name": "b", "calls": 2, "self_cpu_us": 3},
            {"name": "a", "calls": 1, "self_cpu_us": 7},
        ]
        self.assertEqual(rank_operators(rows)[0]["name"], "a")

    def test_empty_profile_is_rejected(self):
        with self.assertRaises(ValueError):
            rank_operators([])

    def test_nonfinite_duration_is_rejected(self):
        for value in [float("nan"), float("inf"), -1]:
            with self.assertRaises(ValueError):
                rank_operators([{"name": "a", "calls": 1, "self_cpu_us": value}])


if __name__ == "__main__":
    unittest.main()