spb/localvm-research Public License

1#!/usr/bin/env python32# =============================================================================3#  Project   : localvm-research4#  File      : experiments/micro/expH_ssd_feasibility/benchmark.py5#  Purpose   : Measure real SSD read behavior on macOS/APFS (F_NOCACHE, block6#              sizes, random vs sequential, threads, concurrent Metal load)7#  Author    : Simon-Pierre Boucher8#  Contact   : contact@spboucher.ai9#  Created   : 2026-08-1110#  Modified  : 2026-08-1111#  Platform  : macOS / Apple Silicon (arm64)12#  License   : All rights reserved (research code)13# =============================================================================14"""Experiment H — SSD feasibility (charter §9.H).1516Measures actual (not theoretical) read throughput of the internal Apple NVMe17at block sizes 4 KB … 4 MB, sequential vs random, with and without the page18cache (F_NOCACHE — macOS has no O_DIRECT), 1/4/8 threads, and optionally19under concurrent Metal GPU compute (MLX matmul loop).2021Writes results JSON (with embedded hardware manifest) to22results/expH_ssd_feasibility/<timestamp>/results.json.2324Usage:25    python3 benchmark.py [--file-gib 8] [--repeats 3] [--quick]26"""2728from __future__ import annotations2930import argparse31import fcntl32import json33import os34import statistics35import sys36import threading37import time38from datetime import datetime, timezone39from pathlib import Path4041REPO_ROOT = Path(__file__).resolve().parents[3]42sys.path.insert(0, str(REPO_ROOT / "benchmarks"))43from hardware_manifest import collect_manifest  # noqa: E4024445F_NOCACHE = 48  # from <sys/fcntl.h> on macOS46BLOCK_SIZES = [4 << 10, 16 << 10, 64 << 10, 256 << 10, 1 << 20, 4 << 20]47THREAD_COUNTS = [1, 4, 8]484950def create_test_file(path: Path, size_bytes: int) -> None:51    """Write an incompressible test file WITHOUT populating the page cache.5253    Crucial macOS detail (discovered in the first run of this experiment):54    F_NOCACHE on a *read* fd does not bypass pages already resident in the55    unified page cache — and writing the file normally makes every page56    resident. The file must therefore be written with F_NOCACHE set on the57    write fd, so 'cold' reads afterwards genuinely hit the SSD.58    """59    if path.exists():60        path.unlink()  # always recreate: an old file may have cached pages61    import numpy as np6263    rng = np.random.default_rng(42)64    chunk = rng.integers(0, 256, size=64 << 20, dtype=np.uint8).tobytes()65    written = 066    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o644)67    try:68        fcntl.fcntl(fd, F_NOCACHE, 1)69        while written < size_bytes:70            n = min(len(chunk), size_bytes - written)71            os.write(fd, chunk[:n])72            written += n73        os.fsync(fd)74    finally:75        os.close(fd)767778def _reader(path: Path, offsets: list[int], block: int, nocache: bool,79            out: dict, idx: int) -> None:80    fd = os.open(path, os.O_RDONLY)81    try:82        if nocache:83            fcntl.fcntl(fd, F_NOCACHE, 1)84        total = 085        t0 = time.perf_counter()86        for off in offsets:87            total += len(os.pread(fd, block, off))88        dt = time.perf_counter() - t089        out[idx] = (total, dt)90    finally:91        os.close(fd)929394def measure(path: Path, file_size: int, block: int, pattern: str,95            nocache: bool, threads: int, budget_bytes: int, seed: int) -> dict:96    """One measurement cell. Returns MB/s and IOPS."""97    import random as _random9899    n_blocks_total = max(1, budget_bytes // block)100    max_off = file_size - block101    rng = _random.Random(seed)102    if pattern == "random":103        all_offsets = [rng.randrange(0, max_off // block) * block for _ in range(n_blocks_total)]104    else:105        stride = max(block, (max_off // n_blocks_total) // block * block) if n_blocks_total > 1 else block106        all_offsets = [(i * block) % (max_off + 1) for i in range(n_blocks_total)]107        del stride108109    per_thread = [all_offsets[i::threads] for i in range(threads)]110    out: dict = {}111    ts = [112        threading.Thread(target=_reader, args=(path, per_thread[i], block, nocache, out, i))113        for i in range(threads)114    ]115    t0 = time.perf_counter()116    for t in ts:117        t.start()118    for t in ts:119        t.join()120    wall = time.perf_counter() - t0121    total_bytes = sum(v[0] for v in out.values())122    return {123        "mb_per_s": total_bytes / wall / 1e6,124        "iops": total_bytes / block / wall,125        "bytes": total_bytes,126        "wall_s": wall,127    }128129130def warm_cache(path: Path) -> None:131    with open(path, "rb") as f:132        while f.read(64 << 20):133            pass134135136def adaptive_budget(path: Path, file_size: int, block: int, pattern: str,137                    nocache: bool, threads: int, floor: int,138                    target_s: float = 3.0) -> int:139    """Choose a per-cell byte budget so each repeat runs ~target_s of wall140    time (short cells are dominated by thread-start overhead and timer141    granularity — the first quick run produced impossible 48 GB/s cells at142    ~10 ms wall)."""143    probe = measure(path, file_size, block, pattern, nocache, threads,144                    budget_bytes=max(floor // 8, 32 << 20), seed=7)145    rate = probe["mb_per_s"] * 1e6146    budget = int(rate * target_s)147    return max(floor, min(budget, 8 << 30))148149150class IostatLogger:151    """Log `iostat -d -w 1` for the duration of the run — ground truth for152    whether bytes actually came from the disk controller vs the page cache."""153154    def __init__(self, out_path: Path) -> None:155        self.out_path = out_path156        self.proc: object = None157158    def __enter__(self) -> "IostatLogger":159        import subprocess160161        self._fh = open(self.out_path, "w")162        self.proc = subprocess.Popen(163            ["iostat", "-d", "-w", "1"], stdout=self._fh, stderr=subprocess.DEVNULL164        )165        return self166167    def __exit__(self, *exc: object) -> None:168        self.proc.terminate()169        self._fh.close()170171172class GpuLoad:173    """Background MLX matmul loop to contend for unified memory bandwidth."""174175    def __init__(self) -> None:176        self.stop = threading.Event()177        self.iterations = 0178        self.thread: threading.Thread | None = None179        self.available = False180        try:181            import mlx.core as mx182183            self._mx = mx184            self.available = True185        except ImportError:186            pass187188    def _loop(self) -> None:189        mx = self._mx190        a = mx.random.normal((4096, 4096), dtype=mx.float16)191        b = mx.random.normal((4096, 4096), dtype=mx.float16)192        while not self.stop.is_set():193            c = mx.matmul(a, b)194            mx.eval(c)195            self.iterations += 1196197    def __enter__(self) -> "GpuLoad":198        if self.available:199            self.thread = threading.Thread(target=self._loop, daemon=True)200            self.thread.start()201            time.sleep(0.5)  # let the GPU ramp202        return self203204    def __exit__(self, *exc: object) -> None:205        self.stop.set()206        if self.thread:207            self.thread.join(timeout=10)208209210def main() -> None:211    ap = argparse.ArgumentParser()212    ap.add_argument("--file-gib", type=float, default=8.0)213    ap.add_argument("--repeats", type=int, default=3)214    ap.add_argument("--budget-mib", type=int, default=1024,215                    help="bytes read per measurement cell")216    ap.add_argument("--quick", action="store_true",217                    help="1 repeat, 512 MiB budget, threads {1,8}")218    args = ap.parse_args()219    if args.quick:220        args.repeats, args.budget_mib = 1, 512221        thread_counts = [1, 8]222    else:223        thread_counts = THREAD_COUNTS224225    file_size = int(args.file_gib * (1 << 30))226    test_file = Path(__file__).parent / "results" / "testfile.bin"227    test_file.parent.mkdir(exist_ok=True)228    print(f"creating {args.file_gib} GiB test file (once)…", flush=True)229    create_test_file(test_file, file_size)230231    manifest = collect_manifest()232    results: list[dict] = []233    budget = args.budget_mib << 20234235    ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")236    out_dir = REPO_ROOT / "results" / "expH_ssd_feasibility" / ts237    out_dir.mkdir(parents=True)238    iostat = IostatLogger(out_dir / "iostat.log")239    iostat.__enter__()240241    def run_cells(gpu_loaded: bool) -> None:242        for nocache in (True, False):243            if not nocache:244                warm_cache(test_file)245            for pattern in ("random", "sequential"):246                for block in BLOCK_SIZES:247                    for threads in thread_counts:248                        if gpu_loaded and (pattern != "random" or not nocache or threads != 8):249                            continue  # GPU-load condition: key cells only250                        cell_budget = (251                            budget if args.quick else adaptive_budget(252                                test_file, file_size, block, pattern, nocache,253                                threads, floor=budget)254                        )255                        runs = [256                            measure(test_file, file_size, block, pattern, nocache,257                                    threads, cell_budget, seed=1000 + r)258                            for r in range(args.repeats)259                        ]260                        mbps = [r["mb_per_s"] for r in runs]261                        cell = {262                            "block_bytes": block,263                            "pattern": pattern,264                            "nocache": nocache,265                            "threads": threads,266                            "gpu_load": gpu_loaded,267                            "repeats": args.repeats,268                            "budget_bytes": cell_budget,269                            "mb_per_s_mean": statistics.mean(mbps),270                            "mb_per_s_median": statistics.median(mbps),271                            "mb_per_s_std": statistics.stdev(mbps) if len(mbps) > 1 else 0.0,272                            "iops_mean": statistics.mean(r["iops"] for r in runs),273                        }274                        results.append(cell)275                        print(276                            f"{'GPU+' if gpu_loaded else '    '}"277                            f"{pattern:>10} {block >> 10:>5} KiB "278                            f"nocache={int(nocache)} t={threads}: "279                            f"{cell['mb_per_s_mean']:9.1f} MB/s",280                            flush=True,281                        )282283    run_cells(gpu_loaded=False)284285    gpu = GpuLoad()286    gpu_iters = 0287    if gpu.available:288        # the warm-cache phase above populated the page cache; recreate the289        # file (uncached write) so the GPU-load cells are genuinely cold290        print("recreating test file to evict cached pages before GPU-load phase…", flush=True)291        create_test_file(test_file, file_size)292        print("re-running key cells under concurrent Metal (MLX) matmul load…", flush=True)293        with gpu:294            run_cells(gpu_loaded=True)295        gpu_iters = gpu.iterations296    else:297        print("MLX not available — skipping GPU-load condition", flush=True)298299    iostat.__exit__()300    payload = {301        "experiment": "expH_ssd_feasibility",302        "author": "Simon-Pierre Boucher",303        "contact": "contact@spboucher.ai",304        "manifest": manifest,305        "config": vars(args),306        "test_file_bytes": file_size,307        "gpu_load_matmul_iterations": gpu_iters,308        "thermal_level_after": collect_manifest()["thermal_level_at_collect"],309        "cells": results,310    }311    out_path = out_dir / "results.json"312    out_path.write_text(json.dumps(payload, indent=2))313    print(f"\nwrote {out_path}")314315316if __name__ == "__main__":317    main()318