guest-diag: demand pinning for stalled CE VAs (event-independent hole filler)

The one-shot #sem kprobe event that drives the CE push-buffer chain
resolution can be lost (trace rotation, scanner restart, slab kptr reuse),
and the chain resolver then never learns a client - every GPFIFO entry of
its CE channels is grace-skipped and the client spins on its tracker
semaphore (the copy-heavy stall class: streams2/memcpy/multikernel/saxpy
in r148; smiley as a later client in r147 took ~90 s to recover).

New demand channel, both ends here:
- pin_gr_stream: _demand_loop scans the host dmesg for 'stalled VA #N:
  0x...' (the service logs each distinct unreadable push VA once), tracks
  each sighting, and re-sends every 20 s for up to 5 min to
  /tmp/guest_want_vas.txt in the guest (via the guest agent).
- guest_pin_stream: the main loop reads new VAs from that file and resolves
  each by CONTENT against the nv_vm_map_pages vmap areas
  (resolve_va_by_content: candidate leaf page at the VA's in-window offset,
  +1/-1 neighbours for the header-page offset ambiguity, host-plausible
  method-stream prefix). On a match the exact pair is emitted; only the
  requested VA is pinned, so a wrong candidate cannot poison other VAs.

Also fix the scanner start: the kill-first pattern must be anchored
(^python3 /tmp/guest_pin_...) - the unanchored pattern matched the deploy
shell's own cmdline and every start attempt killed its own shell before
launching the scanner (r149: no scanner for the whole boot).

2 files changed, 160 insertions, 4 deletions

diff --git a/tools/guest-diag/guest_pin_stream.py b/tools/guest-diag/guest_pin_stream.py
index e80751a..ce98d5b 100644
--- a/tools/guest-diag/guest_pin_stream.py
+++ b/tools/guest-diag/guest_pin_stream.py
@@ -266,6 +266,80 @@ def _vmalloc_pages_for(kc, va):
     return 0
 
 
+def _nv_vmap_areas(kc):
+    """nv_vm_map_pages vmalloc areas: [(lo, hi, pages)] (5 s cache).
+
+    Every CE push-buffer rm_mem is a vmap created by nv_vm_map_pages; the
+    demand pinner walks these areas to find the page holding a stalled
+    push VA by content, without needing any kprobe event or object chain.
+    """
+    now = time.time()
+    if getattr(_nv_vmap_areas, "ts", 0) > now - 5.0:
+        return _nv_vmap_areas.areas
+    areas = []
+    try:
+        for line in open("/proc/vmallocinfo"):
+            if "nv_vm_map_pages" not in line:
+                continue
+            p = line.split()
+            if len(p) < 2:
+                continue
+            try:
+                lo, hi = p[0].split("-")
+                lo = int(lo, 16)
+                hi = int(hi, 16)
+            except ValueError:
+                continue
+            pages = (hi - lo) >> 12
+            if pages > 0:
+                areas.append((lo, hi, pages))
+    except OSError:
+        pass
+    _nv_vmap_areas.areas = areas
+    _nv_vmap_areas.ts = now
+    return areas
+
+
+def resolve_va_by_content(kc, va, out):
+    """Demand pin: find the page carrying GPU VA 'va' by content match.
+
+    The host reports the CE push-buffer VAs whose GPFIFO entries it cannot
+    read (dmesg 'stalled VA #N: 0x...'). The guest walks every
+    nv_vm_map_pages vmap area and tests the candidate leaf page at the VA's
+    in-window offset (plus one neighbour on either side, for the per-mem
+    header-page offset ambiguity) against the host's method-stream prefix
+    check. On a match it emits the exact pair, so the host can translate
+    and consume the entry - no kprobe event, no object-chain discovery, no
+    resolution lottery. Only the requested VA is pinned, so a wrong
+    candidate cannot poison other VAs.
+    """
+    if not (PB_CE_LO <= va < PB_CE_HI):
+        return False
+    off = (va - PB_CE_LO) & 0xFFFFF
+    k = off >> 12
+    for (lo, _hi, pages) in _nv_vmap_areas(kc):
+        for adj in (0, 1, -1):
+            idx = k + adj
+            if idx < 0 or idx >= pages:
+                continue
+            cpu = lo + idx * 0x1000
+            d = kc.read(cpu, 64)
+            if not d or not any(d[:32]):
+                continue
+            if not _push_prefix_ok(d):
+                continue
+            pa = kc.va_to_pa(cpu)
+            if pa is None:
+                continue
+            out.append("0x%x,0x%x,1\n" % (va & ~0xFFF, pa))
+            dbg("demand-va: 0x%x -> pa=0x%x (area off=%d adj=%+d)"
+                % (va & ~0xFFF, pa, idx, adj))
+            return True
+    dbg("demand-va: 0x%x not found by content (%d areas)"
+        % (va & ~0xFFF, len(_nv_vmap_areas(kc))))
+    return False
+
+
 def resolve_mem_pins(kc, mem, out, default_pages=16, va_lo=0, va_hi=0,
                      page_off=0):
     """uvm_rm_mem kptr -> GPU-VA/GPA runs (direct map or page-table walk).
@@ -1471,6 +1545,8 @@ def main():
     last_pid = None
     last_log = 0.0
     scrub_next = 0.0
+    demand_next = 0.0
+    _want_state = {"mtime": None, "done": set()}
     loop = 0
     while True:
         pid = cdt_pid()
@@ -1549,6 +1625,37 @@ def main():
         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 = []
+                        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
         # 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:
diff --git a/tools/guest-diag/pin_gr_stream.py b/tools/guest-diag/pin_gr_stream.py
index 2e45bbf..e1f5195 100644
--- a/tools/guest-diag/pin_gr_stream.py
+++ b/tools/guest-diag/pin_gr_stream.py
@@ -16,6 +16,7 @@ import re
 import socket
 import subprocess
 import sys
+import threading
 import time
 
 HERE = os.path.dirname(os.path.abspath(__file__))
@@ -66,10 +67,13 @@ def _start_cmd():
     # a partially-failed first attempt (pid file not written yet) used to
     # leave the first scanner alive, and multiple scanners each re-assert
     # the whole CE pair set - r148 ran three and drove the guest to load 19
-    # (clients wedged in userspace spins). The pattern is assembled at
-    # runtime via printf so this command's own cmdline (which contains the
-    # literal file path) cannot match the pkill.
-    return ("P=$(printf 'guest_pin_%%s.py' stream); "
+    # (clients wedged in userspace spins). The pattern is anchored
+    # (^python3) and assembled at runtime via printf: an unanchored pattern
+    # or a literal one would also match this command's own shell (its
+    # cmdline contains the script path), and pkill would kill the shell
+    # before it could start the new scanner - every start attempt then
+    # failed silently (r149).
+    return ("P=$(printf '^python3 /tmp/guest_pin_%%s.py' stream); "
             "sudo pkill -f \"$P\" 2>/dev/null; sleep 1; "
             "echo '%s' | base64 -d > /tmp/guest_pin_stream.py; "
             "sudo rm -f /tmp/guest_pin_stream.pid; "
@@ -121,9 +125,54 @@ def scrub_register(fin_pa, hsem_pa):
     return False
 
 
+def _demand_loop():
+    """Push the host's unreadable CE push VAs to the guest scanner.
+
+    The service logs each distinct stalled VA once per boot ('stalled VA
+    #N: 0x...'). Each sighting is tracked and re-sent every 20 s for five
+    minutes: the guest resolves the VA by content against the
+    nv_vm_map_pages areas (event-independent, no object chain), so a page
+    that was not yet written when first asked is picked up on a retry.
+    This closes the stall class where the one-shot #sem kprobe event was
+    lost and the chain resolver never learns the client.
+    """
+    track = {}			# va -> [first_seen, last_sent]
+    while True:
+        time.sleep(3)
+        try:
+            r = subprocess.run(["sudo", "-n", "dmesg"],
+                               capture_output=True, text=True, timeout=20)
+        except Exception:
+            continue
+        now = time.time()
+        for m in re.finditer(r"stalled VA #\d+: (0x[0-9a-f]+)",
+                             (r.stdout or "")[-200000:]):
+            va = int(m.group(1), 16)
+            track.setdefault(va, [now, 0.0])
+        for va in list(track):
+            first, last = track[va]
+            if now - first > 300.0:
+                del track[va]
+        todo = [va for va, (f, l) in track.items() if now - l > 20.0]
+        if not todo:
+            continue
+        lines = "".join("0x%x\n" % va for va in todo)
+        try:
+            vexec("printf '%%s' '%s' >> /tmp/guest_want_vas.txt" % lines, 30)
+        except Exception:
+            continue
+        for va in todo:
+            track[va][1] = now
+        log("demand: sent %d stalled VA(s) to guest" % len(todo))
+
+
 def main():
     if not install_and_start():
         log("WARN: could not start guest scanner (will retry via vsock wait)")
+    try:
+        threading.Thread(target=_demand_loop, daemon=True).start()
+    except Exception as e:
+        log("WARN: demand thread failed to start: %s" % e)
     srv = socket.socket(socket.AF_VSOCK, socket.SOCK_STREAM)
     srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
     srv.bind((socket.VMADDR_CID_ANY, PORT))