vgsp: m178 GR launch reliability + diagnostics; tools: pin latching robustness

Kernel: authorship instrumentation + swrel gate for in-flight GR release VAs;
first-poll shortening + fast extended re-arm (params gr_retry_fast=1,
gr_first_poll_ms=2500); per-attempt win0 bank recompute + shadow recopy in the
outcome-retry loop; poll-break/deposit-readback diagnostics; C1-C10 sema
diagnostics (crel-entry, relx, opdef-clear, walk, pub-adv, hold-fallback,
busy-defer, CAND-OVERWRITE, inline recs); busy-defer hold instead of
software-release consume; counter fast path; QMD word 48/49 skip.
Tools: heartbeat thread, scanner-alive probe on reinstall, full periodic
re-assert, #nopid handling, per-client reset fix (debounce), dead-pid drop,
supervisor grace+threshold.

1 files changed, 113 insertions, 100 deletions

diff --git a/tools/guest-diag/guest_pin_stream.py b/tools/guest-diag/guest_pin_stream.py
index 4bfaaf8..f645f76 100644
--- a/tools/guest-diag/guest_pin_stream.py
+++ b/tools/guest-diag/guest_pin_stream.py
@@ -1627,15 +1627,126 @@ def main():
     pool_reemit_next = 0.0
     _want_state = {"mtime": None, "done": set()}
     loop = 0
+
+    def _send(lines):
+        """Flush control/pin lines to the host, connecting the vsock if needed.
+
+        Returns True when the lines were sent (or there were none to send)."""
+        nonlocal sock
+        if not lines:
+            return True
+        payload = ''.join(lines).encode()
+        if sock is None:
+            try:
+                sock = socket.socket(socket.AF_VSOCK, socket.SOCK_STREAM)
+                sock.settimeout(5)
+                sock.connect((CID, PORT))
+                sock.settimeout(None)
+                if last_pid is not None:
+                    sock.sendall(("#pid %d\n" % last_pid).encode())
+            except OSError:
+                if sock:
+                    sock.close()
+                sock = None
+                return False
+        try:
+            sock.sendall(payload)
+        except OSError:
+            sock.close()
+            sock = None
+            return False
+        return True
+
     while True:
         pid = cdt_pid()
         loop += 1
+        lines = []
+        # Pid-independent work first: the scrubber words are kernel objects
+        # that exist before any client, and the pool/demand re-asserts heal
+        # the host pin table between clients. Gating these on a live pid kept
+        # the host unregistered/unhealed whenever no client was present.
+        if kc_ok and time.time() >= demand_next:
+            demand_next = time.time() + 1.0
+            try:
+                st_d = os.stat("/tmp/guest_want_vas.txt")
+                if st_d.st_mtime != _want_state.get("mtime"):
+                    with open("/tmp/guest_want_vas.txt") as fw:
+                        wvas = [int(t, 16) for t in fw.read().split()
+                                if t.startswith("0x")]
+                    _want_state["mtime"] = st_d.st_mtime
+                    for wv in wvas[-64:]:
+                        if wv in _want_state["done"]:
+                            continue
+                        got = []
+                        if POOL_SEM_LO <= (wv & ~0xFFF) < POOL_SEM_HI:
+                            # m165 (D-F1a): pool/waiter VAs have no method
+                            # stream content and the CE offset math does not
+                            # apply; route them to the structural tracker-sem
+                            # walk instead. m168: the tracker sems are
+                            # long-lived objects (allocated once) whose page
+                            # pointer follows the pool across clients, so try
+                            # every sem we have ever resolved a chain for -
+                            # seen_sems alone can miss the live generation.
+                            _psems = set(seen_sems) | set(KC_PB_STATE.keys()) \
+                                | set(resolved_kc_pb)
+                            try:
+                                ok = resolve_pool_va(kc, wv, got,
+                                                     _psems)
+                            except Exception:
+                                ok = False
+                                got = []
+                            dbg('demand-pool-va: 0x%x %s'
+                                % (wv, 'ok' if ok else 'pending'))
+                            if ok and got:
+                                pool_pages.setdefault(wv & ~0xFFF, True)
+                        else:
+                            try:
+                                ok = resolve_va_by_content(kc, wv, got)
+                            except Exception:
+                                ok = False
+                                got = []
+                        if ok and got:
+                            lines.extend(got)
+                            _want_state["done"].add(wv)
+            except OSError:
+                pass
+        # m165 (D-F1a): re-assert the pool page pair every KC_POOL_REEMIT_S.
+        # The pool page is boot-stable (UVM frees it only at GPU deinit) but
+        # the host pin table is wiped at every session flush, so a one-shot
+        # pin leaves later clients starving until the slow structural walks
+        # answer; the periodic re-assert heals the wipe within seconds (the
+        # same argument as the CE mem sweep above).
+        if kc_ok and pool_pages and time.time() >= pool_reemit_next:
+            pool_reemit_next = time.time() + KC_POOL_REEMIT_S
+            for pv in list(pool_pages):
+                got = []
+                try:
+                    ok = resolve_pool_va(kc, pv, got, seen_sems)
+                except Exception:
+                    ok = False
+                    got = []
+                if ok and got:
+                    lines.extend(got)
+        # PMA scrubber buffer discovery (throttled): resolve the finish/host-
+        # semaphore word PAs and stream them for host registration. This runs
+        # with or without a client (the words are kernel objects).
+        if kc_ok and time.time() >= scrub_next:
+            scrub_next = time.time() + 5.0
+            try:
+                scrub_scan(kc, lines)
+            except Exception as e:
+                dbg('scrub scan failed: %s' % e)
         if pid is None:
-            last_pid = None
+            if last_pid is not None:
+                last_pid = None
+                # Notify the host so it drops its per-pair dedupe map; the
+                # next client's pairs must be injected unconditionally.
+                lines.append('#nopid\n')
             seen = {}
             seen_sems = set()
             seen_pbs = set()
             seen_chs = set()
+            _send(lines)
             time.sleep(INTERVAL)
             continue
         if pid != last_pid:
@@ -1703,85 +1814,6 @@ def main():
                 cur[0x20400f000] = cpa
         new = [(va, pa) for va, pa in cur.items() if seen.get(va) != pa]
         pending_cur = cur
-        lines = []
-        # Demand pins (host -> this scanner): the host appends the CE
-        # push-buffer VAs of GPFIFO entries it cannot read to
-        # /tmp/guest_want_vas.txt (written via the guest agent by
-        # pin_gr_stream); resolve each by content against the
-        # nv_vm_map_pages areas and emit the pair. This is the
-        # event-independent hole filler for the stall class where the
-        # one-shot #sem kprobe event was lost and the object-chain
-        # resolver never learns the client.
-        if kc_ok and time.time() >= demand_next:
-            demand_next = time.time() + 1.0
-            try:
-                st_d = os.stat("/tmp/guest_want_vas.txt")
-                if st_d.st_mtime != _want_state.get("mtime"):
-                    with open("/tmp/guest_want_vas.txt") as fw:
-                        wvas = [int(t, 16) for t in fw.read().split()
-                                if t.startswith("0x")]
-                    _want_state["mtime"] = st_d.st_mtime
-                    for wv in wvas[-64:]:
-                        if wv in _want_state["done"]:
-                            continue
-                        got = []
-                        if POOL_SEM_LO <= (wv & ~0xFFF) < POOL_SEM_HI:
-                            # m165 (D-F1a): pool/waiter VAs have no method
-                            # stream content and the CE offset math does not
-                            # apply; route them to the structural tracker-sem
-                            # walk instead. m168: the tracker sems are
-                            # long-lived objects (allocated once) whose page
-                            # pointer follows the pool across clients, so try
-                            # every sem we have ever resolved a chain for -
-                            # seen_sems alone can miss the live generation.
-                            _psems = set(seen_sems) | set(KC_PB_STATE.keys()) \
-                                | set(resolved_kc_pb)
-                            try:
-                                ok = resolve_pool_va(kc, wv, got,
-                                                     _psems)
-                            except Exception:
-                                ok = False
-                                got = []
-                            dbg('demand-pool-va: 0x%x %s'
-                                % (wv, 'ok' if ok else 'pending'))
-                            if ok and got:
-                                pool_pages.setdefault(wv & ~0xFFF, True)
-                        else:
-                            try:
-                                ok = resolve_va_by_content(kc, wv, got)
-                            except Exception:
-                                ok = False
-                                got = []
-                        if ok and got:
-                            lines.extend(got)
-                            _want_state["done"].add(wv)
-            except OSError:
-                pass
-        # m165 (D-F1a): re-assert the pool page pair every KC_POOL_REEMIT_S.
-        # The pool page is boot-stable (UVM frees it only at GPU deinit) but
-        # the host pin table is wiped at every session flush, so a one-shot
-        # pin leaves later clients starving until the slow structural walks
-        # answer; the periodic re-assert heals the wipe within seconds (the
-        # same argument as the CE mem sweep above).
-        if kc_ok and pool_pages and time.time() >= pool_reemit_next:
-            pool_reemit_next = time.time() + KC_POOL_REEMIT_S
-            for pv in list(pool_pages):
-                got = []
-                try:
-                    ok = resolve_pool_va(kc, pv, got, seen_sems)
-                except Exception:
-                    ok = False
-                    got = []
-                if ok and got:
-                    lines.extend(got)
-        # PMA scrubber buffer discovery (throttled): resolve the finish/host-
-        # semaphore word PAs and stream them for host registration.
-        if kc_ok and time.time() >= scrub_next:
-            scrub_next = time.time() + 5.0
-            try:
-                scrub_scan(kc, lines)
-            except Exception as e:
-                dbg('scrub scan failed: %s' % e)
         for ctl in tracer.drain():
             parts = ctl.split()
             if ctl.startswith("#sem") and len(parts) >= 2:
@@ -2017,27 +2049,8 @@ def main():
                 lines.append('0x%x,0x%x,%d\n' % (va0, pa0, n))
                 va0, pa0, n = va, pa, 1
             lines.append('0x%x,0x%x,%d\n' % (va0, pa0, n))
-        payload = ''.join(lines).encode()
-        if sock is None:
-            try:
-                sock = socket.socket(socket.AF_VSOCK, socket.SOCK_STREAM)
-                sock.settimeout(5)
-                sock.connect((CID, PORT))
-                sock.settimeout(None)
-                if last_pid is not None:
-                    sock.sendall(("#pid %d\n" % last_pid).encode())
-            except OSError:
-                if sock:
-                    sock.close()
-                sock = None
-                time.sleep(INTERVAL)
-                continue
-        try:
-            sock.sendall(payload)
+        if _send(lines):
             seen = pending_cur
-        except OSError:
-            sock.close()
-            sock = None
         time.sleep(INTERVAL)