guest-diag: content-validated delta emission for the CE push-buffer memHEADmaster

The one-shot CE mem emission pinned the whole 16 MB window including
pages whose content was zero at the time; the host then read zeros for
the other CE ranks' pushes. Sweep the mem every 2 s, skip zero pages,
emit only new/changed {va,pa} pairs, and burst the Kcore PA cache every
30 s so a remap is re-emitted instead of shadowed by the cache.

1 files changed, 118 insertions, 9 deletions

diff --git a/tools/guest-diag/guest_pin_stream.py b/tools/guest-diag/guest_pin_stream.py
index 49974c9..6607ead 100644
--- a/tools/guest-diag/guest_pin_stream.py
+++ b/tools/guest-diag/guest_pin_stream.py
@@ -400,6 +400,77 @@ def mem_page_off(kc, mem):
     return 0
 
 
+def _ce_mem_pages(kc, mem):
+    """CE mem walk parameters: (gpus, cpu, off, npages) or None.
+
+    Mirrors resolve_mem_pins' derivation so the delta sweeper emits the same
+    page set the one-shot resolver used: size at +0x128 when readable, else
+    the vmalloc area length; all CE-window GPU VAs share the CPU mapping.
+    """
+    vas = [kc.read64(mem + 0x10 + i * 8) for i in range(16)]
+    cpu = next((v for v in vas if is_kern_ptr(v)), None)
+    gpus = [v for v in vas if v and PB_CE_LO <= v < PB_CE_HI]
+    if not cpu or not gpus:
+        return None
+    if _direct_map_pa(kc, cpu) is not None:
+        return None		# never a vmap: not the push-buffer shape
+    off = mem_page_off(kc, mem)
+    area = _vmalloc_pages_for(kc, cpu)
+    size = _mem_size(kc, mem)
+    npages = (max(1, min(size >> 12, 4098)) if size
+              else area or 64)
+    if area and off + npages > area:
+        npages = max(1, area - off)
+    return (gpus, cpu, off, npages)
+
+
+def _emit_ce_mem_delta(kc, st, out, fresh_map=False):
+    """Emit the CE mem's live {va,pa} pairs as content-validated runs.
+
+    The one-shot emission this replaces pinned the whole 16 MB window
+    including pages whose content was zero at the time and pages whose vmap
+    pairing had drifted; the host then read zeros for the other CE ranks'
+    pushes (the second-client stall class). The sweep now skips zero pages
+    (a zero page cannot carry a push), re-derives every PA against the live
+    page tables (optionally bursting the Kcore cache first), and emits only
+    pairs that are new or changed since the last sweep - so a page the
+    client writes later, or a mem whose vmap pages moved, is re-emitted
+    within KC_REEMIT_S and the host's last-store-wins pin sees the truth.
+    Never drops a previously emitted pair (a stale pin is corrected by the
+    changed pair, and a failed burst never clears the whole map).
+    """
+    gpus, cpu, off, npages = st["gpus"], st["cpu"], st["page_off"], st["npages"]
+    last = st["last"]
+    if fresh_map:
+        for k in range(npages):
+            kc._pa_cache.pop(cpu + (k + off) * 0x1000, None)
+    changed = []
+    for k in range(npages):
+        d = kc.read(cpu + (k + off) * 0x1000, 32)
+        if not d or not any(d):
+            continue		# zero/unreadable: nothing to pin yet
+        pa = kc.va_to_pa(cpu + (k + off) * 0x1000)
+        if pa is None:
+            continue
+        for gv in gpus:
+            if last.get(gv + k * 0x1000) != pa:
+                changed.append((gv + k * 0x1000, pa))
+    changed.sort()
+    merged = []
+    for va, pa in changed:
+        if merged and va == merged[-1][0] + merged[-1][2] * 0x1000 and \
+                pa == merged[-1][1] + merged[-1][2] * 0x1000:
+            merged[-1][2] += 1
+        else:
+            merged.append([va, pa, 1])
+    for va, pa, n in merged:
+        out.append("0x%x,0x%x,%d\n" % (va, pa, n))
+    for va, pa in changed:
+        last[va] = pa
+    st["emit_total"] = st.get("emit_total", 0) + len(merged)
+    return len(merged)
+
+
 def _emit_ring_pins(kc, gpu_va, entries_cpu, entries, out):
     """Pin the GPFIFO ring pages (and the pages after it, where the GPPUT
     buffer usually sits) from the ring's CPU mapping."""
@@ -559,8 +630,10 @@ KC_CHAIN_BACK = 0x2000
 KC_CHAIN_SPAN = 0xA000
 KC_CHAIN_MAX = 400
 RESOLVED_KC_PB = set()
-KC_PB_STATE = {}		# sem kptr -> resolved-mem fingerprint (freshness)
+KC_PB_STATE = {}		# sem kptr -> resolved-mem state (fingerprint + pins)
 KC_REFRESH_S = 2.0		# re-validate a resolved CE mem at most this often
+KC_REEMIT_S = 2.0		# delta-sweep the CE mem's live pages this often
+KC_REMAP_S = 30.0		# full re-translation sweep (catches remaps)
 
 
 def _chain_mems_at(kc, w):
@@ -743,18 +816,23 @@ def resolve_pb_chain(kc, sem, out):
     if mem in RESOLVED_KC_PB:
         return True
     RESOLVED_KC_PB.add(mem)
-    off = mem_page_off(kc, mem)
-    n = resolve_mem_pins(kc, mem, out, default_pages=64,
-                         va_lo=PB_CE_LO, va_hi=PB_CE_HI,
-                         page_off=off)
-    st = {"mem": mem, "gpu": gpu, "cpu": cpu,
-          "page_va": cpu, "page_pa": kc.va_to_pa(cpu) or 0, "page_off": 0}
+    params = _ce_mem_pages(kc, mem)
+    if not params:
+        return found
+    gpus, cpu, off, npages = params
+    st = {"mem": mem, "gpu": gpu, "cpu": cpu, "gpus": gpus,
+          "page_va": cpu, "page_pa": kc.va_to_pa(cpu) or 0,
+          "page_off": off, "npages": npages, "last": {}}
+    # Fresh translation for the first sweep: the Kcore PA cache may hold
+    # pairs from before this mem's vmap area was (re)populated.
+    kc._pa_cache.clear()
+    n = _emit_ce_mem_delta(kc, st, out)
     KC_PB_STATE[sem] = st
     try:
         with open('/tmp/guest_pin_stream.log', 'a') as f:
             f.write('%s kc-chain CE push-buffer mem=0x%x sem=0x%x '
-                    'pin-runs=%d off=%d size=%d score=%d\n'
-                    % (time.strftime('%T'), mem, sem, n, off,
+                    'pin-runs=%d off=%d npages=%d size=%d score=%d\n'
+                    % (time.strftime('%T'), mem, sem, n, off, npages,
                        _mem_size(kc, mem), _score))
     except OSError:
         pass
@@ -1252,6 +1330,8 @@ def main():
     resolved_kc_pb = set()
     retry_at = {}
     kc_refresh_at = {}
+    kc_reemit_at = {}
+    kc_remap_at = {}
     tracer = TraceReader()
     kc = Kcore()
     kc_ok = kc.load()
@@ -1288,6 +1368,8 @@ def main():
             resolved_kc_pb = set()
             retry_at = {}
             kc_refresh_at = {}
+            kc_reemit_at = {}
+            kc_remap_at = {}
             # Client switch: the old client's channel/put pointers are freed
             # memory. Clearing the #put map stops stale reads from being
             # emitted as if they were the new client's counters.
@@ -1414,6 +1496,33 @@ def main():
                             kc_refresh_at[sem] = now_r + 5.0
                             dbg("kc-chain stale: keep old latch sem=0x%x"
                                 % sem)
+                    else:
+                        # CE mem delta sweep (second-client pins): the one-shot
+                        # emission pinned the mem once; pages the client writes
+                        # later (or whose vmap pairing moved) were never
+                        # re-emitted, so the host kept zeros for the other CE
+                        # ranks' pushes. Sweep every KC_REEMIT_S, emitting only
+                        # new/changed nonzero pairs; burst the Kcore PA cache
+                        # every KC_REMAP_S so a remap is not shadowed.
+                        _st = KC_PB_STATE.get(sem)
+                        if _st is not None and \
+                                now_r >= kc_reemit_at.get(sem, 0):
+                            _fresh_map = now_r >= kc_remap_at.get(sem, 0)
+                            if _fresh_map:
+                                kc_remap_at[sem] = now_r + KC_REMAP_S
+                            kc_reemit_at[sem] = now_r + KC_REEMIT_S
+                            got = []
+                            try:
+                                _n2 = _emit_ce_mem_delta(kc, _st, got,
+                                                         fresh_map=_fresh_map)
+                            except Exception:
+                                _n2 = 0
+                                got = []
+                            if got:
+                                lines.extend(got)
+                                added += len(got)
+                                dbg("ce-delta sem=0x%x runs=%d mem=0x%x"
+                                    % (sem, _n2, _st.get("mem", 0)))
                 # CE push-buffer chain (second-client stall fix): resolve
                 # pool -> manager -> pushbuffer -> rm_mem directly, so the
                 # push-buffer pages never depend on the racy one-shot uvbp