kernel: batch the map-add channel rewind; trust injected streams for ops

Two fixes for the CE data path. (1) svc_handle_map_add rewound every
channel by 8 ring entries on every injected mapping; a 16 MB push-buffer pin
is thousands of page injections, so the pump's last_get was pinned in place
and it re-read the same entries for the whole batch (3645 submissions, zero
ops). The rewind now runs once per drained batch. (2) The push-start
heuristic (op=INC_METHOD at address 0) rejects UVM's real CE memop pushes,
which start with an IMM method at address 8 (observed 0x60020002, PTE data
words) or a release method at 0x700 (0x200381c0); a buffer read through a
host-injected (ipin) translation is now decoded fully trusted, ops included.
Result on the next boot: memset=2718/0/0 ringops=1794 (first sustained CE
work), false 12k->1.9k. semfix_watch: bulk-pin contiguous runs via
sema_map_add len, read rm_mem size, 16 MB (4096 page) cap, non-blocking QMP
read (the fixed 30 ms sleep made 4096 gva2gpa calls take two minutes), and
KEEP_TRACE=1 to preserve the guest trace for post-mortems.

1 files changed, 72 insertions, 25 deletions

diff --git a/tools/guest-diag/semfix_watch.py b/tools/guest-diag/semfix_watch.py
index 7730e86..400fafe 100644
--- a/tools/guest-diag/semfix_watch.py
+++ b/tools/guest-diag/semfix_watch.py
@@ -34,6 +34,7 @@
 import json
 import os
 import re
+import select
 import socket
 import subprocess
 import sys
@@ -47,8 +48,8 @@ QMP_SOCK = os.environ.get("QMP_SOCK", "/tmp/qemu-qmp.sock")
 GPU_VA_LO = 0x100000000
 GPU_VA_HI = 0x2000000000
 PUSHBUF_VA_LO = int(os.environ.get("PUSHBUF_VA_LO", "0x120000000"), 16)
-PUSHBUF_VA_HI = int(os.environ.get("PUSHBUF_VA_HI", "0x120400000"), 16)
-PUSHBUF_MAX_PAGES = 1024
+PUSHBUF_VA_HI = int(os.environ.get("PUSHBUF_VA_HI", "0x124000000"), 16)
+PUSHBUF_MAX_PAGES = int(os.environ.get("PUSHBUF_MAX_PAGES", "4096"))
 CHAN_WINDOW_BACK = int(os.environ.get("CHAN_WINDOW_BACK", "0x4000"), 16)
 CHAN_MAX_CANDS = int(os.environ.get("CHAN_MAX_CANDS", "256"))
 
@@ -63,11 +64,27 @@ class Qmp:
         self._cache = {}
 
     def _rd(self):
-        time.sleep(0.03)
-        try:
-            return self.s.recv(1 << 20).decode()
-        except Exception:
-            return ""
+        """Read one QMP reply (JSON object ends with a newline). Poll with a
+        short timeout instead of a fixed sleep: the fixed 30 ms made batch
+        page translations (4096 per pushbuffer) take two minutes."""
+        buf = b""
+        deadline = time.time() + 10
+        while time.time() < deadline:
+            r, _, _ = select.select([self.s], [], [], 0.05)
+            if not r:
+                if buf:
+                    break
+                continue
+            try:
+                chunk = self.s.recv(1 << 20)
+            except Exception:
+                break
+            if not chunk:
+                break
+            buf += chunk
+            if buf.endswith(b"\n"):
+                break
+        return buf.decode()
 
     def cmd(self, name, **args):
         try:
@@ -128,8 +145,11 @@ def guest_trace():
     Read *before* clearing: the create-time probes fire once per client and
     would be wiped by a clear that races them.
     """
-    cmd = ("sudo sh -c 'grep -E \"uvmt|uvch|uvbp|uvpbf\" /sys/kernel/tracing/trace "
-           "| tail -400; echo > /sys/kernel/tracing/trace'")
+    inner = ("grep -E \"uvmt|uvch|uvbp|uvpbf\" /sys/kernel/tracing/trace "
+             "| tail -400")
+    if not os.environ.get("KEEP_TRACE"):
+        inner += "; echo > /sys/kernel/tracing/trace"
+    cmd = "sudo sh -c '%s'" % inner
     try:
         out = subprocess.run([VMEXEC, "exec", cmd], capture_output=True,
                              text=True, timeout=90).stdout
@@ -162,7 +182,8 @@ def vas_from_mem(q, mem, d):
     if not gpa:
         diag_add(d, "no_gpa")
         return None
-    return gpu_vas, cpu, gpa
+    size = q.read64(mem + 0x60)
+    return gpu_vas, cpu, gpa, (size[0] if size else 0)
 
 
 def pushbuffer_of_manager(q, mgr, d):
@@ -310,6 +331,17 @@ def pin(va, pa):
     return False
 
 
+def pin_bulk(va, pa, nbytes):
+    cmd = "printf '0x%x,0x%x,0x%x' > %s/sema_map_add" % (va, pa, nbytes, PARAM)
+    for _ in range(40):
+        r = subprocess.run(["sudo", "-n", "sh", "-c", cmd],
+                           capture_output=True, text=True)
+        if r.returncode == 0 and "timed out" not in (r.stderr or ""):
+            return True
+        time.sleep(0.05)
+    return False
+
+
 def resolve_sem_pool(q, sem):
     page = q.read64(sem)
     if not page or not page[0]:
@@ -328,28 +360,43 @@ def resolve_sem_pool(q, sem):
     return [(gv, gpa) for gv in gpu_vas]
 
 
-def pin_pushbuffer(q, gpu_vas, cpu, gpa):
-    """Pin the whole pushbuffer with *per-page* translations (the layout is
-    not linear across the allocation: page 32 jumped on a real client)."""
+def pin_pushbuffer(q, gpu_vas, cpu, gpa, size=0):
+    """Pin the pushbuffer pages (the UVM allocation is 16 x 1 MB chunks =
+    16 MB = 4096 pages; the page layout is only piecewise contiguous, so
+    group ascending-contiguous runs into one bulk injection each)."""
+    pages = min(max(size >> 12, 1) if size else PUSHBUF_MAX_PAGES,
+                PUSHBUF_MAX_PAGES)
     pinned = 0
     for gv in gpu_vas:
         miss = 0
-        for i in range(PUSHBUF_MAX_PAGES):
+        i = 0
+        while i < pages:
             g = q.gva2gpa(cpu + i * 0x1000)
             if not g:
                 miss += 1
-                if miss >= 32:
-                    print("semfix: pushbuf va=0x%x: 32 unmapped pages at %d, stop (%d pinned)"
-                          % (gv, i, pinned), flush=True)
+                if miss >= 64:
+                    print("semfix: pushbuf va=0x%x: %d unmapped pages at %d, stop (%d pinned)"
+                          % (gv, miss, i, pinned), flush=True)
                     return pinned
+                i += 1
                 continue
             miss = 0
-            if not pin(gv + i * 0x1000, g):
-                print("semfix: pushbuf va=0x%x: pin failed at page %d" % (gv, i), flush=True)
+            j = i + 1
+            last = g
+            while j < pages:
+                gj = q.gva2gpa(cpu + j * 0x1000)
+                if gj != last + 0x1000:
+                    break
+                last = gj
+                j += 1
+            n = j - i
+            if not pin_bulk(gv + i * 0x1000, g, n * 0x1000):
+                print("semfix: pushbuf va=0x%x: bulk pin failed at page %d" % (gv, i), flush=True)
                 return pinned
-            pinned += 1
-        print("semfix: pushbuf va=0x%x cpu=0x%x gpa=0x%x: pinned %d page(s)"
-              % (gv, cpu, gpa, pinned), flush=True)
+            pinned += n
+            i = j
+        print("semfix: pushbuf va=0x%x cpu=0x%x gpa=0x%x size=0x%x: pinned %d page(s)"
+              % (gv, cpu, gpa, size, pinned), flush=True)
     return pinned
 
 
@@ -361,12 +408,12 @@ def main():
     seen_ch = set()
     pinned_ranges = set()
 
-    def pin_range_once(gpu_vas, cpu, gpa):
-        key = tuple(gpu_vas), cpu
+    def pin_range_once(gpu_vas, cpu, gpa, size=0):
+        key = tuple(gpu_vas), cpu, size
         if key in pinned_ranges:
             return
         pinned_ranges.add(key)
-        pin_pushbuffer(q, gpu_vas, cpu, gpa)
+        pin_pushbuffer(q, gpu_vas, cpu, gpa, size)
 
     print("semfix: watching (interval %.1fs)" % interval, flush=True)
     while True: