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).

3 files changed, 37 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
diff --git a/tools/guest-diag/pin_gr_stream.py b/tools/guest-diag/pin_gr_stream.py
index b80d30a..fbe2ee5 100644
--- a/tools/guest-diag/pin_gr_stream.py
+++ b/tools/guest-diag/pin_gr_stream.py
@@ -134,6 +134,9 @@ def main():
                 if text.startswith("#sem"):
                     _mirror(text)
                     continue
+                if text.startswith("#pb"):
+                    _mirror(text)
+                    continue
                 m = re.match(r"^(0x[0-9a-f]+),(0x[0-9a-f]+)(?:,(\d+))?$", text)
                 if m:
                     runs.append((int(m.group(1), 16), int(m.group(2), 16),
diff --git a/tools/guest-diag/semfix_uvmt.py b/tools/guest-diag/semfix_uvmt.py
index fd58299..09cc5f8 100755
--- a/tools/guest-diag/semfix_uvmt.py
+++ b/tools/guest-diag/semfix_uvmt.py
@@ -63,6 +63,8 @@ class Stream:
         self.off = 0
         self.sems = []
         self.pids = []
+        self.pbs = []
+        self.new_pbs = []
 
     def poll(self):
         try:
@@ -77,6 +79,7 @@ class Stream:
         except OSError:
             return [], []
         new_sems, new_pids = [], []
+        self.new_pbs = []
         for line in data.splitlines():
             if line.startswith("#sem"):
                 m = re.search(r"0x([0-9a-fA-F]+)", line)
@@ -85,6 +88,13 @@ class Stream:
                     if v not in self.sems:
                         self.sems.append(v)
                         new_sems.append(v)
+            elif line.startswith("#pb"):
+                m = re.search(r"0x([0-9a-fA-F]+)", line)
+                if m:
+                    v = int(m.group(1), 16)
+                    if v not in self.pbs:
+                        self.pbs.append(v)
+                        self.new_pbs.append(v)
             elif line.startswith("#pid"):
                 m = re.search(r"(\d+)", line)
                 if m:
@@ -265,6 +275,9 @@ def main():
         for p in new_pids:
             log("stream: client pid %d" % p)
             last_pid = p
+        for pb in getattr(stream, "new_pbs", []):
+            log("stream: new pushbuffer 0x%x" % pb)
+            pbs.add(pb)
         sems.update(stream.sems)
 
         # The guest-agent trace read is slow under client load; use it only