"""pytest plugin: record or replay autotune configs across benchmark runs. When comparing two runs (e.g. two compiler builds), replay the first run's recorded configs in the second: the autotuner's choice is pinned, so the latency delta reflects the change under test rather than the autotuner picking a different config under different IR. Modes (mutually exclusive, selected by env var; op identity from FLAGGEMS_PERF_CURRENT_OP): - FLAGGEMS_PERF_AUTOTUNE_RECORD_DIR: run full autotune, dump the chosen Config per (kernel, key) to /.json at session end. - FLAGGEMS_PERF_AUTOTUNE_REPLAY_DIR: pre-populate Autotuner.cache from /.json, skipping the sweep. Missing key / non-compiling config / missing json fall back to full autotune and emit an AUTOTUNE_REPLAY_FALLBACK marker into the log. """ from __future__ import annotations import atexit import json import os import sys import threading from pathlib import Path from typing import Any, Dict, Optional, Tuple _RECORD_DIR_ENV = "FLAGGEMS_PERF_AUTOTUNE_RECORD_DIR" _REPLAY_DIR_ENV = "FLAGGEMS_PERF_AUTOTUNE_REPLAY_DIR" _OP_ENV = "FLAGGEMS_PERF_CURRENT_OP" # Module-global aggregates, accumulated as run() is called and dumped at exit. # Lock guards merges in case a kernel uses threads internally. _record_map: Dict[str, Dict[str, Dict[str, Any]]] = {} _record_lock = threading.Lock() _replay_map: Dict[str, Dict[str, Dict[str, Any]]] = {} _dump_done = False def _op_name() -> str: return os.environ.get(_OP_ENV, "").strip() or "default" def _record_path(record_dir: str) -> Path: return Path(record_dir) / f"{_op_name()}.json" def _kernel_id(tuner: Any) -> str: # Stable across runs: source module + qualified name. Shapes/keys split # by the inner key dict. name = ( getattr(tuner, "__name__", None) or getattr(getattr(tuner, "base_fn", None), "__name__", None) or getattr(getattr(tuner, "fn", None), "__name__", None) or "unknown_kernel" ) module = getattr(getattr(tuner, "fn", None), "__module__", "") or "" return f"{module}::{name}" def _serialize_key(key: Tuple[Any, ...]) -> str: # Per-element repr, JSON-encoded into a flat string usable as a JSON key; # stable across runs given seeded RNG + yaml-fixed shapes. return json.dumps([repr(x) for x in key]) def _serialize_config(cfg: Any) -> Optional[Dict[str, Any]]: # Drop configs with a pre_hook (un-serializable callable) rather than replay # without it and compute incorrectly. if getattr(cfg, "pre_hook", None) is not None: return None try: return { "kwargs": dict(cfg.kwargs), "num_warps": cfg.num_warps, "num_stages": cfg.num_stages, "num_ctas": cfg.num_ctas, "maxnreg": cfg.maxnreg, } except Exception: return None def _deserialize_config(entry: Dict[str, Any]) -> Any: # Late import (triton loads during collection, after plugin registration). from triton import Config return Config( kwargs=dict(entry["kwargs"]), num_warps=entry.get("num_warps", 4), num_stages=entry.get("num_stages", 3), num_ctas=entry.get("num_ctas", 1), maxnreg=entry.get("maxnreg"), ) def _compute_key(tuner: Any, args: Tuple[Any, ...], kwargs: Dict[str, Any]) -> Optional[Tuple[Any, ...]]: # Reproduce Autotuner/LibTuner key computation; return None on any internals # shift (caller then just loses replay for that call). try: arg_names = tuner.arg_names nargs = dict(zip(arg_names, args)) all_args = {**nargs, **kwargs} _args = {k: v for k, v in all_args.items() if k in arg_names} if hasattr(tuner, "get_key"): return tuner.get_key(_args) key = [_args[name] for name in tuner.keys if name in _args] for _, arg in _args.items(): if hasattr(arg, "dtype"): key.append(str(arg.dtype)) return tuple(key) except Exception: return None def _emit_marker(reason: str) -> None: # Marker lands in run.log (grep it to audit replay coverage); flush so it # appears before the kernel runs. Yellow on the live terminal only — # run_pytest.sh strips ANSI from run.log. from _term_style import YELLOW, paint print(paint(f"AUTOTUNE_REPLAY_FALLBACK reason={reason}", YELLOW), flush=True) def _record_run(original): # Snapshot self.cache before/after run() to capture the chosen config (works # for both Autotuner and LibTuner). Single-config kernels skip the cache # write, so record configs[0] explicitly for uniform replay. def runner(self, *args, **kwargs): try: keys_before = set(self.cache.keys()) if hasattr(self.cache, "keys") else set() except Exception: keys_before = set() result = original(self, *args, **kwargs) try: keys_after = set(self.cache.keys()) if hasattr(self.cache, "keys") else set() except Exception: keys_after = set() new_keys = keys_after - keys_before if not new_keys: # No new entry: record configs[0] for single-config kernels (cache # write bypassed); otherwise nothing to record (disk-cache hit). if len(getattr(self, "configs", []) or []) == 1: key = _compute_key(self, args, kwargs) if key is not None and key not in self.cache: cfg = self.configs[0] entry = _serialize_config(cfg) if entry is not None: kid = _kernel_id(self) with _record_lock: bucket = _record_map.setdefault(kid, {}) bucket[_serialize_key(key)] = entry return result kid = _kernel_id(self) with _record_lock: bucket = _record_map.setdefault(kid, {}) for k in new_keys: try: cfg = self.cache[k] except Exception: continue entry = _serialize_config(cfg) if entry is None: continue bucket[_serialize_key(k)] = entry return result return runner def _replay_run(original): # Inject recorded configs before benchmarking; on key-missing or a recorded # config failing to compile/launch, fall back to full autotune (+ marker). def runner(self, *args, **kwargs): injected_key = None if len(getattr(self, "configs", []) or []) > 1: kid = _kernel_id(self) bucket = _replay_map.get(kid) if bucket: key = _compute_key(self, args, kwargs) if key is not None and key not in self.cache: rec = bucket.get(_serialize_key(key)) if rec is None: _emit_marker("key_missing") else: try: cfg = _deserialize_config(rec) self.cache[key] = cfg injected_key = key except Exception as exc: _emit_marker(f"deserialize_{type(exc).__name__}") try: return original(self, *args, **kwargs) except Exception as exc: if injected_key is not None: # Recorded config failed under the current build: drop it and # let run() autotune. try: del self.cache[injected_key] except Exception: pass _emit_marker(f"compile_{type(exc).__name__}") return original(self, *args, **kwargs) raise return runner def _load_replay_dir(replay_dir: str) -> int: path = _record_path(replay_dir) if not path.is_file(): _emit_marker("record_missing") return 0 try: data = json.loads(path.read_text()) except Exception as exc: _emit_marker(f"load_{type(exc).__name__}") return 0 if not isinstance(data, dict): return 0 count = 0 for kid, bucket in data.items(): if not isinstance(bucket, dict): continue _replay_map[kid] = dict(bucket) count += len(bucket) return count def _dump_record(record_dir: str) -> None: global _dump_done if _dump_done: return _dump_done = True path = _record_path(record_dir) try: path.parent.mkdir(parents=True, exist_ok=True) with _record_lock: payload = {kid: dict(bucket) for kid, bucket in _record_map.items()} path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n") except Exception as exc: # Don't break the run if dump fails; the perf row is still produced. from _term_style import tag print(f"{tag('[autotune-record-plugin]')} dump failed: {exc}", file=sys.stderr, flush=True) def pytest_configure(config): from _term_style import tag record_dir = os.environ.get(_RECORD_DIR_ENV, "").strip() replay_dir = os.environ.get(_REPLAY_DIR_ENV, "").strip() if record_dir and replay_dir: # Mutually exclusive: recording while replaying is pointless. Surface loudly. print(f"{tag('[autotune-record-plugin]')} error: both RECORD_DIR and " "REPLAY_DIR set; ignoring both (no-op)", file=sys.stderr, flush=True) return if not record_dir and not replay_dir: return import triton.runtime.autotuner as _autotuner wrap = _record_run if record_dir else _replay_run _autotuner.Autotuner.run = wrap(_autotuner.Autotuner.run) patched = ["Autotuner"] try: from flag_gems.utils.libentry import LibTuner if "run" in LibTuner.__dict__: LibTuner.run = wrap(LibTuner.__dict__["run"]) patched.append("LibTuner") except Exception: pass if record_dir: # atexit (not sessionfinish): persist whatever was recorded even if an op # crash kills the session; a later replay run falls back for missing keys. atexit.register(_dump_record, record_dir) print(f"{tag('[autotune-record-plugin]')} recording autotune configs to " f"{_record_path(record_dir)} ({'+'.join(patched)})", file=sys.stderr, flush=True) else: loaded = _load_replay_dir(replay_dir) print(f"{tag('[autotune-record-plugin]')} replaying {loaded} recorded entries from " f"{_record_path(replay_dir)} ({'+'.join(patched)})", file=sys.stderr, flush=True)