#!/usr/bin/env python3 """EPM strict-mode runner: enforce an 8 GiB / no-swap ceiling on one process. Why this exists --------------- The repository's headline deployment claim is "MiniCPM-V-4.7-35B-A3B decodes inside 8 GiB of RAM with swap disabled". Past runs only ever set ``RLIMIT_DATA`` (which caps anonymous data/heap, ignores file-backed mappings, and does not touch swap) and then *observed* ``VmSwap: 0``. That is a pager-sizing number, not a guarantee. cgroups would be the clean way to express this — ``memory.max=8G`` plus ``memory.swap.max=0`` — but a writable cgroup2 mount needs ``CAP_SYS_ADMIN``. Where that is unavailable this harness reaches the same two guarantees: no-swap Every anonymous mapping the process creates is ``mlock``ed by ``tools/epq_lock.so``. Locked pages can never be written to a swap device, so no-swap is enforced by the kernel for the whole run rather than inferred from a single reading afterwards. File-backed mappings are deliberately *not* locked: clean page cache is reclaimable, so a real 8 GiB machine evicts it rather than swapping, and a cgroup would reclaim it too. ``VmLck`` in the record shows how much was actually locked. 8 GiB ``RLIMIT_DATA`` on the child is a hard kernel cap on anonymous memory: past it, ``brk``/``mmap(MAP_ANONYMOUS)`` fail instead of quietly growing. A supervisor samples ``/proc//status`` every few milliseconds and ``SIGKILL``s on a breach, so transient spikes cannot slip through. Enforcement and evidence come from kernel counters, not from the sampler: ``RssAnon``/``VmHWM``/``VmSwap``/``VmLck`` are exact, so the sampler only has to be fast, not lucky. ``--metric anon`` (default) caps anonymous RSS, which is the quantity that binds on real hardware: 8 GiB of RAM must hold the arena, the KV cache, the compute buffers and the runtime, and the kernel can always reclaim the rest. ``--metric total`` caps total RSS instead, which is strictly harder than a cgroup because it forbids even reclaimable cache. Use it when you want the pessimistic number; use ``anon`` for the number that matches a cgroup. Usage ----- python3 tools/strict8g.py --label decode -- epq-chat -m model.gguf ... python3 tools/strict8g.py --limit-mb 8192 --metric total --record r.jsonl -- cmd Exit status is the child's own, except 124 when the ceiling was breached or the run timed out (matching ``timeout``), and 125 for harness errors. """ from __future__ import annotations import argparse import json import os import resource import signal import subprocess import sys import tempfile import threading import time from pathlib import Path REPO = Path(__file__).resolve().parent.parent LOCK_SRC = REPO / "tools" / "epq_lock.so.c" LOCK_SO = REPO / "tools" / "epq_lock.so" MIB = 1024 * 1024 def build_lock_helper() -> Path: """Compile the anonymous-memory locker once; it is one small C file.""" if not LOCK_SRC.exists(): raise SystemExit(f"[strict8g] 缺少 {LOCK_SRC}") cc = os.environ.get("CC", "cc") if (not LOCK_SO.exists()) or LOCK_SO.stat().st_mtime < LOCK_SRC.stat().st_mtime: subprocess.run([cc, "-O2", "-fPIC", "-shared", "-o", str(LOCK_SO), str(LOCK_SRC)], check=True) return LOCK_SO _WANTED = ("RssAnon", "RssFile", "VmRSS", "VmHWM", "VmSwap", "VmLck", "VmSize", "VmData") def read_status(pid: int) -> dict: """Read one process's exact memory counters, in bytes.""" out = dict.fromkeys(_WANTED, 0) try: with open(f"/proc/{pid}/status", "r") as fh: for line in fh: name = line.split(":", 1)[0] if name in out: out[name] = int(line.split()[1]) * 1024 except (FileNotFoundError, ProcessLookupError, ValueError, IndexError): pass return out class Sampler(threading.Thread): """Polls the child's memory counters and kills it on a ceiling breach.""" def __init__(self, pid: int, limit: int, metric: str, interval: float): super().__init__(daemon=True) self.pid = pid self.limit = limit self.metric = metric self.interval = interval self.peak = dict.fromkeys(_WANTED, 0) self.samples = 0 self.swap_samples = 0 self.breach = 0 self.stop = threading.Event() def run(self) -> None: while not self.stop.is_set(): st = read_status(self.pid) self.samples += 1 for k in _WANTED: self.peak[k] = max(self.peak[k], st[k]) if st["VmSwap"] > 0: self.swap_samples += 1 cur = st[self.metric] if cur > self.limit: self.breach = cur try: os.kill(self.pid, signal.SIGKILL) except ProcessLookupError: pass return time.sleep(self.interval) def _mib(x: int) -> float: return round(x / MIB, 1) def main() -> int: ap = argparse.ArgumentParser(description="Run one command under a strict 8 GiB / no-swap ceiling") ap.add_argument("--limit-mb", type=int, default=8192, help="hard ceiling in MiB") ap.add_argument("--data-mb", type=int, default=0, help="RLIMIT_DATA in MiB; default is --limit-mb plus VA slack " "(RLIMIT_DATA bounds anonymous *virtual* memory, which is " "stricter than physical RAM, so it is used as a runaway guard)") ap.add_argument("--no-rlimit", action="store_true", help="do not set RLIMIT_DATA at all") ap.add_argument("--metric", choices=["anon", "total"], default="anon", help="cap anonymous RSS (default, cgroup-equivalent) or total RSS") ap.add_argument("--interval-ms", type=float, default=4.0, help="sampling interval") ap.add_argument("--label", default="strict") ap.add_argument("--record", default="", help="append a JSON record to this file") ap.add_argument("--timeout", type=float, default=0, help="wall-clock limit in seconds (0 = none)") ap.add_argument("--no-lock", action="store_true", help="skip mlock (no-swap then rests on VmSwap sampling alone)") ap.add_argument("--quiet", action="store_true") ap.add_argument("cmd", nargs=argparse.REMAINDER) args = ap.parse_args() cmd = args.cmd[1:] if args.cmd and args.cmd[0] == "--" else args.cmd if not cmd: ap.error("missing command; put it after `--`") limit = args.limit_mb * MIB # RLIMIT_DATA caps anonymous *virtual* memory; llama.cpp reserves graph and # compute buffers that are mapped but never touched, so charging them against # a physical 8 GiB budget costs ~1 GiB of expert arena for no physical # benefit. Physical residency is bounded by the sampler instead, exactly as a # cgroup memory.max bounds it. RLIMIT_DATA stays as an independent kernel # guard against a runaway, sized with slack for untouched mappings. VA_SLACK_MB = 1536 data_mb = args.data_mb or (args.limit_mb + VA_SLACK_MB) metric = "RssAnon" if args.metric == "anon" else "VmRSS" try: helper = build_lock_helper() if not args.no_lock else None except (subprocess.CalledProcessError, OSError) as exc: raise SystemExit(f"[strict8g] 無法編譯 mlock helper：{exc}") fd, status_path = tempfile.mkstemp(prefix="epq-mlock-", suffix=".txt") os.close(fd) try: env = dict(os.environ) if helper: env["LD_PRELOAD"] = str(helper) + (":" + env["LD_PRELOAD"] if env.get("LD_PRELOAD") else "") env["EPM_MLOCK_STATUS"] = status_path def child_setup() -> None: # mlock() is bounded by RLIMIT_MEMLOCK, so raise it before main runs. try: resource.setrlimit(resource.RLIMIT_MEMLOCK, (resource.RLIM_INFINITY, resource.RLIM_INFINITY)) except (ValueError, OSError): pass # Kernel guard on anonymous virtual memory (see VA_SLACK_MB above). if not args.no_rlimit: cap = data_mb * MIB try: resource.setrlimit(resource.RLIMIT_DATA, (cap, cap)) except (ValueError, OSError): pass # A multi-GiB core dump would fill the disk and mask the failure we # are trying to measure. try: resource.setrlimit(resource.RLIMIT_CORE, (0, 0)) except (ValueError, OSError): pass os.setpgrp() t0 = time.time() try: proc = subprocess.Popen(cmd, env=env, preexec_fn=child_setup, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) except OSError as exc: print(f"[strict8g] 無法執行 {cmd[0]}: {exc}", file=sys.stderr) return 125 sampler = Sampler(proc.pid, limit, metric, args.interval_ms / 1000.0) sampler.start() timed_out = False try: stdout, stderr = proc.communicate(timeout=args.timeout or None) except subprocess.TimeoutExpired: timed_out = True proc.kill() stdout, stderr = proc.communicate() finally: sampler.stop.set() sampler.join(timeout=2.0) wall = time.time() - t0 lock_note = "" if helper: try: lock_note = Path(status_path).read_text().strip().replace("\n", " ") except OSError: pass finally: try: os.unlink(status_path) except OSError: pass rc = proc.returncode overflow = sampler.breach > 0 swapped = sampler.peak["VmSwap"] > 0 locked = sampler.peak["VmLck"] # VmHWM is a kernel high-water mark: the exact peak even if it lasted well # under one sampling interval. ok = (rc == 0 and not overflow and not swapped and not timed_out and (args.no_lock or locked > 0)) rec = { "label": args.label, "cmd": cmd, "returncode": rc, "wall_s": round(wall, 3), "limit_mb": args.limit_mb, "metric": args.metric, "rlimit_data_mb": data_mb, "interval_ms": args.interval_ms, "samples": sampler.samples, "peak_anon_mb": _mib(sampler.peak["RssAnon"]), "peak_file_mb": _mib(sampler.peak["RssFile"]), "peak_rss_mb": _mib(sampler.peak["VmRSS"]), "vm_hwm_mb": _mib(sampler.peak["VmHWM"]), "peak_swap_mb": _mib(sampler.peak["VmSwap"]), "swap_samples_over_zero": sampler.swap_samples, "vm_lck_mb": _mib(locked), "peak_vmsize_mb": _mib(sampler.peak["VmSize"]), "peak_vmdata_mb": _mib(sampler.peak["VmData"]), "overflow_killed": overflow, "breach_mb": _mib(sampler.breach) if overflow else None, "timed_out": timed_out, "lock": lock_note, "strict_ok": ok, } if args.record: p = Path(args.record) p.parent.mkdir(parents=True, exist_ok=True) with p.open("a", encoding="utf-8") as fh: fh.write(json.dumps(rec, ensure_ascii=False) + "\n") if not args.quiet: print(f"[strict8g] {'PASS' if ok else 'FAIL'} {args.label}: rc={rc} wall={wall:.1f}s " f"| anon={rec['peak_anon_mb']} file={rec['peak_file_mb']} " f"total={rec['peak_rss_mb']} hwm={rec['vm_hwm_mb']} MiB " f"vmdata={rec['peak_vmdata_mb']} " f"| cap {args.limit_mb} MiB ({args.metric}) swap={rec['peak_swap_mb']} " f"lck={rec['vm_lck_mb']} samples={sampler.samples}", file=sys.stderr) if overflow: print(f"[strict8g] 超標：{metric} 到 {rec['breach_mb']} MiB 時 SIGKILL", file=sys.stderr) if swapped: print(f"[strict8g] 偵測到 swap：{rec['peak_swap_mb']} MiB", file=sys.stderr) if locked <= 0 and not args.no_lock: print(f"[strict8g] 沒有任何匿名記憶體被鎖定，no-swap 無法成立" f"（{lock_note or 'helper 未回報'}）", file=sys.stderr) elif not ok and not overflow and not swapped and not timed_out: print(f"[strict8g] 子行程自行失敗 rc={rc}", file=sys.stderr) sys.stdout.write(stdout or "") sys.stderr.write(stderr or "") if overflow or timed_out: return 124 return rc if __name__ == "__main__": sys.exit(main())