Measure how much local LLM speed your Mac loses to heat over 30 minutes. Declares its definitions for burst, retention and onset so your numbers are comparable to someone else's.

A developer released sustained_bench.py, an open-source Python script that measures how much local LLM inference throughput on Apple silicon decays over a 30-minute sustained run due to thermal throttling. The tool writes one JSON record per fixed-token generation window to a JSONL file, discarding a warmup generation and rotating four fixed prompts to isolate heat effects from cache pressure, and supports MLX, Ollama, and mock backends. It captures chip, macOS build, Low Power Mode, and AC state at run start, and explicitly declares its definitions for burst, retention, and onset so results are comparable across machines.

| | /usr/bin/env python3 | | | """ | | | sustained bench.py - measure sustained local-inference throughput on Apple silicon. | | | | | | Writes one JSON record per generation window to a JSONL file, so the decay curve | | | survives intact instead of being collapsed into a summary. | | | | | | Design commitments these are the reason the output is comparable to someone else's : | | | A warmup generation runs first and is DISCARDED. It contains lazy Metal kernel | | | compilation and model page-in, which otherwise manufactures a fake decay curve. | | | Windows are fixed at N TOKENS, never N seconds. Time-boxed windows silently | | | change what throughput means once the machine starts throttling. | | | Four fixed prompts rotate. This holds work-per-window constant while preventing | | | unbounded KV-cache growth, so you measure heat rather than cache pressure. | | | Environment is captured at run start: chip, macOS build, Low Power Mode, AC state. | | | Ambient room temperature is the one thing it cannot capture. Write it down. | | | | | | Do NOT run this under sudo. thermal log.sh needs root; this does not, and running | | | the measured workload as root changes scheduling behaviour. | | | | | | Usage: | | | ./sustained bench.py --backend mlx --model mlx-community/Qwen2.5-7B-Instruct-4bit \\ | | | --duration 1800 --out run-plugged-lidopen.jsonl --label "plugged/lid-open/hard" | | | | | | ./sustained bench.py --backend ollama --model qwen2.5:7b --duration 1800 \\ | | | --out run-unplugged.jsonl --label "unplugged/lid-open/hard" | | | | | | ./sustained bench.py --backend mock --duration 1800 --out synthetic.jsonl | | | """ | | | | | | from future import annotations | | | | | | import argparse | | | import json | | | import math | | | import platform | | | import subprocess | | | import sys | | | import time | | | | | | Four fixed prompts. Rotated, not extended. Roughly matched in length so that | | | prefill cost per window is near-constant. | | | PROMPTS = | | | "Explain how a B-tree index speeds up a range query, step by step.", | | | "Describe the trade-offs between mutexes and channels for shared state.", | | | "Walk through how a TLS handshake establishes a shared secret.", | | | "Explain why floating point addition is not associative, with an example.", | | | | | | | | | | | | def sh cmd: list str - str: | | | """Run a command, return stdout stripped, empty string on any failure.""" | | | try: | | | return subprocess.run | | | cmd, capture output=True, text=True, timeout=10, check=False | | | .stdout.strip | | | except Exception: | | | return "" | | | | | | | | | def environment - dict: | | | """Capture everything about the machine that changes the answer.""" | | | env = { | | | "platform": platform.platform , | | | "python": platform.python version , | | | "chip": sh "sysctl", "-n", "machdep.cpu.brand string" , | | | "cores": sh "sysctl", "-n", "hw.ncpu" , | | | "memory bytes": sh "sysctl", "-n", "hw.memsize" , | | | "os product": sh "sw vers", "-productVersion" , | | | "os build": sh "sw vers", "-buildVersion" , | | | } | | | pmset = sh "pmset", "-g" | | | env "lowpowermode" = "unknown" | | | for line in pmset.splitlines : | | | if "lowpowermode" in line: | | | env "lowpowermode" = line.split -1 | | | ps = sh "pmset", "-g", "ps" | | | if "AC Power" in ps: | | | env "power source" = "ac" | | | elif "Battery Power" in ps: | | | env "power source" = "battery" | | | else: | | | env "power source" = "unknown" | | | return env | | | | | | | | | class MLXBackend: | | | """mlx lm. The usual choice on Apple silicon.""" | | | | | | virtual = False | | | | | | def init self, model: str : | | | from mlx lm import load imported late so other backends need no MLX | | | | | | self.model, self.tokenizer = load model | | | self.name = f"mlx:{model}" | | | | | | def generate self, prompt: str, max tokens: int - tuple int, float : | | | from mlx lm import generate | | | | | | messages = {"role": "user", "content": prompt} | | | text = self.tokenizer.apply chat template | | | messages, add generation prompt=True, tokenize=False | | | | | | t0 = time.monotonic | | | out = generate | | | self.model, self.tokenizer, prompt=text, | | | max tokens=max tokens, verbose=False, | | | | | | elapsed = time.monotonic - t0 | | | return len self.tokenizer.encode out , elapsed | | | | | | | | | class OllamaBackend: | | | """Ollama HTTP API. Wall-clock timed here, not trusted from eval duration.""" | | | | | | virtual = False | | | | | | def init self, model: str, host: str = "http://127.0.0.1:11434" : | | | self.model, self.host = model, host | | | self.name = f"ollama:{model}" | | | | | | def generate self, prompt: str, max tokens: int - tuple int, float : | | | import urllib.request | | | | | | payload = json.dumps { | | | "model": self.model, | | | "prompt": prompt, | | | "stream": False, | | | "options": {"num predict": max tokens}, | | | } .encode | | | req = urllib.request.Request | | | f"{self.host}/api/generate", data=payload, | | | headers={"Content-Type": "application/json"}, | | | | | | t0 = time.monotonic | | | with urllib.request.urlopen req, timeout=600 as r: | | | body = json.loads r.read | | | elapsed = time.monotonic - t0 | | | return int body.get "eval count", 0 , elapsed | | | | | | | | | class MockBackend: | | | """ | | | Synthetic thermal curve. No model, no sleeping, virtual clock. | | | | | | Exists so the harness and the analyzer can be validated against a decay whose | | | true onset is known exactly. That is the only way to measure how far the | | | detector lags the physics, which is the central claim of the article this | | | harness accompanies. | | | """ | | | | | | virtual = True | | | | | | def init self, burst: float = 40.0, floor frac: float = 0.60, | | | onset s: float = 90.0, tau s: float = 120.0 : | | | self.burst, self.floor frac = burst, floor frac | | | self.onset s, self.tau s = onset s, tau s | | | self.t = 0.0 | | | self.name = f"mock:burst={burst}tok/s,onset={onset s}s," | | | f"tau={tau s}s,floor={floor frac}" | | | | | | def rate at self, t: float - float: | | | if t < self.onset s: | | | return self.burst | | | decayed = self.floor frac + 1 - self.floor frac math.exp | | | - t - self.onset s / self.tau s | | | | | | return self.burst decayed | | | | | | def generate self, prompt: str, max tokens: int - tuple int, float : | | | Integrate the rate across the window rather than sampling it once, so the | | | window's reported rate is the true average over its span. | | | | | | Fixed step, not an adaptive one. An earlier version used | | | step = min 0.05, remaining / rate , which collapses the step size as | | | remaining approaches zero and then grinds on floating-point residue, | | | exiting only via the guard below. Bounded iteration counts only. | | | STEP = 0.01 | | | produced, dt, t = 0.0, 0.0, self.t | | | while produced < max tokens and dt < 3600: | | | produced += self.rate at t STEP | | | t += STEP | | | dt += STEP | | | self.t += dt | | | return max tokens, dt | | | | | | | | | def build backend args - object: | | | if args.backend == "mlx": | | | return MLXBackend args.model | | | if args.backend == "ollama": | | | return OllamaBackend args.model | | | return MockBackend | | | burst=args.mock burst, floor frac=args.mock floor, | | | onset s=args.mock onset, tau s=args.mock tau, | | | | | | | | | | | | def main - int: | | | p = argparse.ArgumentParser description= doc , | | | formatter class=argparse.RawDescriptionHelpFormatter | | | p.add argument "--backend", choices= "mlx", "ollama", "mock" , default="mock" | | | p.add argument "--model", default="" | | | p.add argument "--duration", type=float, default=1800, | | | help="run length in seconds, excluding warmup default 1800 " | | | p.add argument "--tokens-per-window", type=int, default=256, | | | help="FIXED tokens per window. Never time-box a window." | | | p.add argument "--out", default="run.jsonl" | | | p.add argument "--label", default="", | | | help='condition, e.g. "plugged/lid-open/hard-surface"' | | | p.add argument "--ambient-c", type=float, default=None, | | | help="room temperature in C. The harness cannot measure this." | | | p.add argument "--mock-burst", type=float, default=40.0 | | | p.add argument "--mock-floor", type=float, default=0.60 | | | p.add argument "--mock-onset", type=float, default=90.0 | | | p.add argument "--mock-tau", type=float, default=120.0 | | | args = p.parse args | | | | | | if args.backend in "mlx", "ollama" and not args.model: | | | p.error f"--model is required for backend {args.backend}" | | | | | | backend = build backend args | | | | | | header = { | | | "record": "header", | | | "schema": 1, | | | "started utc": time.strftime "%Y-%m-%dT%H:%M:%SZ", time.gmtime , | | | "backend": backend.name, | | | "virtual clock": backend.virtual, | | | "tokens per window": args.tokens per window, | | | "target duration s": args.duration, | | | "label": args.label, | | | "ambient c": args.ambient c, | | | "prompts": PROMPTS, | | | "env": environment , | | | } | | | | | | with open args.out, "w" as fh: | | | fh.write json.dumps header + "\n" | | | fh.flush | | | | | | Warmup. Discarded on purpose. Recorded only so the file proves it happened. | | | w tokens, w elapsed = backend.generate PROMPTS 0 , args.tokens per window | | | fh.write json.dumps { | | | "record": "warmup", "tokens": w tokens, "wall s": round w elapsed, 4 , | | | "tok s": round w tokens / w elapsed, 3 if w elapsed else None, | | | "note": "discarded from all statistics", | | | } + "\n" | | | fh.flush | | | print f"warmup discarded: {w tokens} tok in {w elapsed:.1f}s", file=sys.stderr | | | | | | clock = lambda: backend.t if backend.virtual else None | | | t0 = time.monotonic | | | elapsed total = 0.0 | | | i = 0 | | | if backend.virtual: | | | backend.t = 0.0 | | | | | | while elapsed total < args.duration: | | | prompt = PROMPTS i % len PROMPTS | | | t start = elapsed total | | | tokens, wall = backend.generate prompt, args.tokens per window | | | elapsed total = clock if clock else time.monotonic - t0 | | | rec = { | | | "record": "window", | | | "i": i, | | | "prompt idx": i % len PROMPTS , | | | "t start s": round t start, 3 , | | | "t end s": round elapsed total, 3 , | | | "wall s": round wall, 4 , | | | "tokens": tokens, | | | "tok s": round tokens / wall, 3 if wall else None, | | | } | | | fh.write json.dumps rec + "\n" | | | fh.flush | | | print f" window {i: 3} t={t start:7.1f}s {rec 'tok s' : 7.2f} tok/s", | | | file=sys.stderr | | | i += 1 | | | | | | fh.write json.dumps { | | | "record": "footer", "windows": i, | | | "total s": round elapsed total, 3 , | | | } + "\n" | | | | | | print f"\n{i} windows over {elapsed total:.0f}s written to {args.out}", | | | file=sys.stderr | | | print f"next: ./analyze.py {args.out}", file=sys.stderr | | | return 0 | | | | | | | | | if name == " main ": | | | sys.exit main |