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.

2 files changed, 118 insertions, 49 deletions

diff --git a/kernel/gsp_sema_service.c b/kernel/gsp_sema_service.c
index ac1849f..b9e7db3 100644
--- a/kernel/gsp_sema_service.c
+++ b/kernel/gsp_sema_service.c
@@ -2050,10 +2050,12 @@ static bool svc_method_stream_plausible(const u8 *buf, u32 len)
  * not match the channel rank).
  */
 static u64 svc_rel_va_seen;
+static u32 svc_dbg_streams;
 
 static void svc_decode_stream(struct nv_vgpu_device *vgpu, const u8 *buf,
 			      u32 len, u64 page_pa, bool allow_ops,
-			      u32 ch_h, u32 ch_eng, u64 push_va)
+			      u32 ch_h, u32 ch_eng, u64 push_va,
+			      bool trusted)
 {
 	struct svc_push_state st;
 	/*
@@ -2075,13 +2077,26 @@ static void svc_decode_stream(struct nv_vgpu_device *vgpu, const u8 *buf,
 	 * addr 0, sane count). Releases from such a page still apply, so a
 	 * genuinely shifted stream is not lost.
 	 */
-	if (allow_ops && !svc_method_stream_plausible(buf, len)) {
+	if (allow_ops && !trusted && !svc_method_stream_plausible(buf, len)) {
 		svc.ops_denied++;
 		allow_ops = false;
 	}
 
 	memset(&st, 0, sizeof(st));
 	svc.pages_decoded++;
+	if (ch_eng && svc_dbg_streams < 24) {
+		svc_dbg_streams++;
+		pr_info("sema-dbg: stream h=0x%x eng=%u va=0x%llx len=%u allow=%d w=%08x %08x %08x %08x %08x %08x %08x %08x\n",
+			ch_h, ch_eng, push_va, len, allow_ops,
+			len >= 4 ? *(const u32 *)(buf + 0) : 0,
+			len >= 8 ? *(const u32 *)(buf + 4) : 0,
+			len >= 12 ? *(const u32 *)(buf + 8) : 0,
+			len >= 16 ? *(const u32 *)(buf + 12) : 0,
+			len >= 20 ? *(const u32 *)(buf + 16) : 0,
+			len >= 24 ? *(const u32 *)(buf + 20) : 0,
+			len >= 28 ? *(const u32 *)(buf + 24) : 0,
+			len >= 32 ? *(const u32 *)(buf + 28) : 0);
+	}
 
 	while (off + 4 <= len) {
 		u32 hdr = 0, op, count, subch, addr;
@@ -2595,7 +2610,8 @@ static u32 svc_apply_page_data(struct nv_vgpu_device *vgpu, const u8 *buf,
 	 * fabricated "pushes" executing memsets); work execution is limited
 	 * to pushes the pump read through a trusted ring translation.
 	 */
-	svc_decode_stream(vgpu, buf, SEMA_PAGE_SIZE, pa_page, false, 0, 0, 0);
+	svc_decode_stream(vgpu, buf, SEMA_PAGE_SIZE, pa_page, false, 0, 0, 0,
+			  false);
 
 	for (off = 0; off + 16 <= SEMA_PAGE_SIZE; off += 4) {
 		u32 hdr, hi, lo, payload;
@@ -3898,9 +3914,7 @@ static void svc_pump_channels(struct nv_vgpu_device *vgpu)
 									    &iap) == 0);
 						if (!svc_method_stream_plausible(pb, len)) {
 							svc.false_trans++;
-							if (injected)
-								allow_ops = false;
-							else
+							if (!injected)
 								unreadable = true;
 						}
 					}
@@ -3923,7 +3937,8 @@ static void svc_pump_channels(struct nv_vgpu_device *vgpu)
 									  allow_ops,
 									  ch->h_object,
 									  ch->engine_type,
-									  va);
+									  va,
+									  injected);
 							if (svc_rel_va_seen != rel_before)
 								ch->last_rel_va =
 									svc_rel_va_seen;
@@ -3990,7 +4005,8 @@ static void svc_pump_channels(struct nv_vgpu_device *vgpu)
 											  false,
 											  ch->h_object,
 											  ch->engine_type,
-											  va);
+											  va,
+											  false);
 									if (svc_rel_va_seen != rel_before)
 										ch->last_rel_va =
 											svc_rel_va_seen;
@@ -4380,10 +4396,12 @@ static DEFINE_SPINLOCK(svc_map_add_lock);
 static void svc_handle_map_add(void)
 {
 	unsigned long flags;
+	bool rewound = false;
+	u32 i;
 
 	for (;;) {
 		struct svc_map_add_req req;
-		u32 ap, i;
+		u32 ap;
 		u64 off;
 
 		spin_lock_irqsave(&svc_map_add_lock, flags);
@@ -4402,17 +4420,25 @@ static void svc_handle_map_add(void)
 			svc_ipin_store(req.va + off, req.pa + off, ap);
 			svc_lat_store(req.va + off, req.pa + off, ap);
 		}
-		/*
-		 * A translation can be replaced after the pump already
-		 * consumed pushes that release through it (a new client maps
-		 * new physical pages at the same VA, or the first translation
-		 * was wrong): the releases resolved against the old page,
-		 * compared as stale against its values, and were dropped.
-		 * Rewind every channel a few ring entries so those pushes are
-		 * re-read under the new translation; the release/fill
-		 * application paths are forward-progress guarded, so
-		 * re-executing them is idempotent.
-		 */
+		rewound = true;
+		pr_info_ratelimited("sema-svc: injected mapping va=0x%llx -> pa=0x%llx len=0x%llx (%s)\n",
+			req.va, req.pa, req.len,
+			ap == SVC_APERTURE_FB ? "vidmem" : "sysmem");
+	}
+
+	/*
+	 * Rewind once per drained batch, not once per mapping: a push-buffer
+	 * pin injects thousands of single pages, and rewinding the channels
+	 * on every one of them pinned the pump's last_get in place and
+	 * re-read the same ring entries for the whole injection (observed:
+	 * 3645 "submissions" with zero executed ops). A translation can be
+	 * replaced after the pump consumed pushes that released through it
+	 * (new client at the same VA, or a first translation that was
+	 * wrong), so the batch still replays a few entries; release/fill
+	 * application is forward-progress guarded, so re-executing is
+	 * idempotent.
+	 */
+	if (rewound) {
 		spin_lock_irqsave(&svc.chan_lock, flags);
 		for (i = 0; i < SVC_MAX_CHANNELS; i++) {
 			struct svc_channel *ch = &svc.chan[i];
@@ -4440,10 +4466,6 @@ static void svc_handle_map_add(void)
 		 * stall.
 		 */
 		svc.fast_force_until = jiffies + 10 * HZ;
-		pr_info("sema-svc: injected mapping va=0x%llx -> pa=0x%llx len=0x%llx (%s), rewound channels by %u\n",
-			req.va, req.pa, req.len,
-			ap == SVC_APERTURE_FB ? "vidmem" : "sysmem",
-			SVC_PIN_REWIND);
 	}
 }
 
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: