guest-diag: stream uvbp pushbuffer kptrs for immediate pinning

The first CE channel's push-buffer was only pinned ~1 minute after the
client's failure: uvbp fired at t+4s but semfix's trace read (guest agent,
every 5 loops) picked it up at t+70s, so the tracker wait for value 3 spun
for 16 s. The guest scanner now emits '#pb 0x<kptr>' from the uvbp trace
lines as soon as they appear; the host mirrors them into the sem stream and
semfix resolves/pins the push-buffer through the UVM chain on the next loop
(sub-second).

1 files changed, 21 insertions, 10 deletions

diff --git a/tools/guest-diag/guest_pin_stream.py b/tools/guest-diag/guest_pin_stream.py
index ac4f72b..2f99bbd 100644
--- a/tools/guest-diag/guest_pin_stream.py
+++ b/tools/guest-diag/guest_pin_stream.py
@@ -43,16 +43,21 @@ SEM_EVERY = int(os.environ.get("STREAM_SEM_EVERY", "4"))
 
 
 class TraceReader:
-    """Tail the guest trace for new uvmt semaphore kptrs in a thread.
+    """Tail the guest trace for new uvmt sems / uvbp pushbuffers in a thread.
 
     Reading the trace can block for seconds under client load; doing it in
     the scan loop delayed the first GR-window delta (observed a 3 s gap
     right when the client's channels appeared). The thread appends to a
     queue the main loop drains without blocking.
+
+    Push-buffer kptrs are emitted as soon as the client begins its first
+    push: the host resolves them through the UVM chain and pins the pages,
+    which is what makes the first CE channel's releases applicable without
+    waiting for the content sweep.
     """
 
     def __init__(self):
-        self.sems = []
+        self.lines = []
         self.seen = set()
         self.lock = threading.Lock()
         self.stop = False
@@ -70,14 +75,19 @@ class TraceReader:
             with self.lock:
                 for m in re.finditer(rb"sem=0x([0-9a-fA-F]+)", out):
                     v = int(m.group(1), 16)
-                    if v not in self.seen:
-                        self.seen.add(v)
-                        self.sems.append(v)
-            time.sleep(0.5)
+                    if ("s", v) not in self.seen:
+                        self.seen.add(("s", v))
+                        self.lines.append("#sem 0x%x\n" % v)
+                for m in re.finditer(rb"pb=0x([0-9a-fA-F]+)", out):
+                    v = int(m.group(1), 16)
+                    if ("p", v) not in self.seen:
+                        self.seen.add(("p", v))
+                        self.lines.append("#pb 0x%x\n" % v)
+            time.sleep(0.3)
 
     def drain(self):
         with self.lock:
-            out, self.sems = self.sems, []
+            out, self.lines = self.lines, []
         return out
 
 
@@ -222,9 +232,10 @@ def main():
         new = [(va, pa) for va, pa in cur.items() if seen.get(va) != pa]
         seen = cur
         lines = []
-        for sem in tracer.drain():
-            seen_sems.add(sem)
-            lines.append('#sem 0x%x\n' % sem)
+        for ctl in tracer.drain():
+            if ctl.startswith("#sem"):
+                seen_sems.add(int(ctl.split()[1], 16))
+            lines.append(ctl)
         now = time.time()
         if now - last_log > 1.0:
             last_log = now