Sylpheed/tools/re-capture/probeharness.py

#!/usr/bin/env python3
"""Shared harness for live-memory probes.

Written because the same four lessons were re-learned in four separate one-off
probes, each time by losing a run:

  * **stall witness** — flat output is indistinguishable from a frozen guest.
    Four probes shipped without one; three produced runs that could not be
    interpreted (guest-stalls.md, remaining-ob-hunt.md).
  * **incremental save** — two probes deferred results to the end and a turn
    timeout destroyed them, the second time four iterations after the lesson was
    written down.
  * **baseline discard** — a run whose roster count is not the stage's member
    count is a different stage and must not be interpreted
    (mission-per-record-strength.md).
  * **witness first** — a run's stall state must be read before its numbers.

Anything reading live guest memory should start from `Probe`, not from a blank
file.

    p = Probe(baseline=116)          # None = accept whatever loads
    if not p.ok: sys.exit(p.why)
    p.log('t\\tvalue')               # opens an incrementally-flushed TSV
    while p.tick(every=15):
        p.emit('%d\\t%d' % (p.elapsed, measure()))
        print(p.status())            # includes the stall verdict
"""
import collections, os, struct, sys, time
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import gmem, gworld, entities2
import numpy as np

ROSTER_VT = struct.pack('>I', 0x820AF030)
DELTA, WIN, LINK, HULL = 0x130, 0x400, 0x08, 0x154
HEAP_LO, HEAP_HI = 0xBC000000, 0xBE000000
ENT_LO, ENT_HI = entities2.ENT_VA_LO, entities2.ENT_VA_HI


class Probe:
    def __init__(self, baseline=None, tsv=None, witness=True, rescans=5):
        self.w = gworld.World()
        self.fd = self.w.fd
        self.t0 = time.time()
        self.ok, self.why = False, ''
        self.stalled_samples = 0
        self._f = None
        self.defs = entities2.definitions(self.w)
        if not self.defs:
            self.why = 'NOT IN A MISSION (no unit definitions)'
            return
        self.roster = self._scan_vt(ROSTER_VT)
        for _ in range(rescans):
            if baseline is None or len(self.roster) == baseline:
                break
            time.sleep(10)
            self.roster = self._scan_vt(ROSTER_VT)
        if baseline is not None and len(self.roster) != baseline:
            self.why = ('DISCARD: roster settled at %d, not the baseline %d '
                        '(a different stage loaded)' % (len(self.roster), baseline))
            return
        self.want = {}
        for o in self.roster:
            va = gmem.primary_va(o)
            if va is not None:
                self.want[va + LINK] = o
        self.ticks, self.rate, self._last = [], None, None
        if witness:
            self._calibrate()
        if tsv:
            self.log_path = tsv
        self.ok = True

    # ---- guest access -------------------------------------------------
    def _scan_vt(self, vt):
        out = []
        for a, b in gmem.extents(self.fd, self.w.size):
            pos = a
            while pos < b:
                m = min(1 << 24, b - pos)
                blob = os.pread(self.fd, m, pos)
                i = blob.find(vt)
                while i != -1:
                    if (pos + i) % 4 == 0:
                        out.append(pos + i)
                    i = blob.find(vt, i + 1)
                pos += m
        return sorted(out)

    def heap(self):
        """Raw bytes of the game heap."""
        lo, hi = gmem.va_to_off(HEAP_LO), gmem.va_to_off(HEAP_HI)
        out, pos = bytearray(), lo
        while pos < hi:
            n = min(1 << 24, hi - pos)
            out += os.pread(self.fd, n, pos)
            pos += n
        return bytes(out), lo

    def craft(self):
        """[(base_offset, unit_name, roster_offset|None)] — one per live craft."""
        lo, hi = gmem.va_to_off(ENT_LO), gmem.va_to_off(ENT_HI)
        out, pos = [], lo
        while pos < hi:
            m = min(1 << 24, hi - pos)
            blob = os.pread(self.fd, m, pos)
            for needle, nm in self.defs.items():
                i = blob.find(needle)
                while i != -1:
                    if (pos + i) % 4 == 0:
                        base = pos + i - DELTA
                        own = None
                        head = os.pread(self.fd, WIN, base)
                        for j in range(0, len(head) - 3, 4):
                            (p,) = struct.unpack_from('>I', head, j)
                            if p in self.want:
                                own = self.want[p]
                                break
                        out.append((base, nm, own))
                    i = blob.find(needle, i + 1)
            pos += m
        return out

    def alive(self, base):
        try:
            return struct.unpack('>f', os.pread(self.fd, 4, base + HULL))[0] > 0
        except Exception:
            return False

    def strengths(self):
        """craft alive per roster record."""
        per = collections.Counter()
        for base, _, own in self.craft():
            if own is not None and self.alive(base):
                per[own] += 1
        return per

    # ---- witness ------------------------------------------------------
    def _calibrate(self, dt=3.0):
        a, _ = self.heap()
        time.sleep(dt)
        b, _ = self.heap()
        A = np.frombuffer(a, dtype='>u4').astype(np.int64)
        B = np.frombuffer(b, dtype='>u4').astype(np.int64)
        d = B - A
        rates = collections.Counter((d[(d > 15) & (d < 600)] // int(dt)).tolist())
        band = [r for r in rates if 8 <= r <= 40]
        pick = max(band, key=lambda r: rates[r]) if band else None
        if pick is None:
            return
        self.rate = pick
        self.ticks = np.nonzero(d // int(dt) == pick)[0][:32]
        self._last = B[self.ticks]

    def stalled(self):
        """True if the guest did not advance since the previous check."""
        if not len(self.ticks):
            return None                      # unknown: no witness
        b, _ = self.heap()
        now = np.frombuffer(b, dtype='>u4').astype(np.int64)[self.ticks]
        moved = int((now > self._last).sum())
        self._last = now
        return moved == 0

    # ---- loop / output ------------------------------------------------
    @property
    def elapsed(self):
        return round(time.time() - self.t0)

    def tick(self, every, secs):
        if time.time() - self.t0 >= secs:
            return False
        time.sleep(every)
        self._stall = self.stalled()
        if self._stall:
            self.stalled_samples += 1
        return True

    def status(self):
        s = getattr(self, '_stall', None)
        if s is None:
            return '  (UNVALIDATED: no witness)'
        return '  *** GUEST STALLED ***' if s else ''

    def log(self, header, path=None):
        self._f = open(path or getattr(self, 'log_path', '/tmp/probe.tsv'), 'w')
        self._f.write(header.rstrip('\n') + '\n')
        self._f.flush()

    def emit(self, line):
        """Write AND flush — never defer results to the end of a run."""
        if self._f:
            self._f.write(line.rstrip('\n') + '\n')
            self._f.flush()

    def summary(self):
        w = ('%d witnesses at %s/s' % (len(self.ticks), self.rate)
             if len(self.ticks) else 'NO WITNESS - run UNVALIDATED')
        return ('roster %d, definitions %d, %s, stalled samples %d'
                % (len(self.roster), len(self.defs), w, self.stalled_samples))