tools: stream UVM trace via trace_pipe in the guest pin scanner (kill the second-client stall race)

The guest scanner's TraceReader tailed the last 256 KB of the trace ring
every 0.3 s. A one-shot uvmt sem= event could age out of that window
before the next poll (the client's spin floods the ring), so a second
client in the same VM was never discovered: semfix re-pinned the previous
client's freed sem while the new client's push page stayed untranslated
(host nomap, drain held, guest spins forever).

- guest_pin_stream.py: consume /sys/kernel/tracing/trace_pipe instead
  (each event delivered exactly once; no ring-window race), with
  reconnect on read error. The old ring-tail read is kept only as
  semfix's fallback.
- clear the trace ring on every client switch so the fallback cannot
  re-learn the previous client's stale events.

1 files changed, 46 insertions, 17 deletions

diff --git a/tools/guest-diag/guest_pin_stream.py b/tools/guest-diag/guest_pin_stream.py
index f89ed4a..f0f07d8 100644
--- a/tools/guest-diag/guest_pin_stream.py
+++ b/tools/guest-diag/guest_pin_stream.py
@@ -60,21 +60,32 @@ VMA_PATHS = tuple(p for p in os.environ.get(
 TRACE = ("/sys/kernel/tracing/trace"
          if os.path.exists("/sys/kernel/tracing/trace")
          else "/sys/kernel/debug/tracing/trace")
+TRACE_PIPE = ("/sys/kernel/tracing/trace_pipe"
+              if os.path.exists("/sys/kernel/tracing/trace_pipe")
+              else "/sys/kernel/debug/tracing/trace_pipe")
 SEM_EVERY = int(os.environ.get("STREAM_SEM_EVERY", "4"))
 
 
-class TraceReader:
-    """Tail the guest trace for new uvmt sems / uvbp pushbuffers in a thread.
+def clear_trace():
+    """Drop stale ring events so the fallback tail reader cannot re-learn a
+    previous client's freed sems after a client switch. The new client's
+    events are delivered fresh through the consuming trace_pipe reader."""
+    try:
+        with open(TRACE, "w") as f:
+            f.write("")
+    except OSError:
+        pass
 
-    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.
+class TraceReader:
+    """Stream uvmt sems / uvbp pushbuffers from the guest trace via trace_pipe.
+
+    trace_pipe is a consuming stream: every event is delivered exactly once,
+    with no ring-window race. The old tail-of-trace reader polled the last
+    256 KB every 0.3 s, so a one-shot sem= event could age out of the window
+    when the client's spin flooded the ring before the next poll - that is
+    the second-client stall class (the previous client's freed sem was
+    re-learned from the stale ring while the new client's sem was missed).
     """
 
     def __init__(self):
@@ -82,29 +93,46 @@ class TraceReader:
         self.seen = set()
         self.lock = threading.Lock()
         self.stop = False
+        self.fd = None
         self.t = threading.Thread(target=self._run, daemon=True)
         self.t.start()
 
+    def _open(self):
+        try:
+            self.fd = os.open(TRACE_PIPE, os.O_RDONLY)
+        except OSError:
+            self.fd = None
+
     def _run(self):
         while not self.stop:
+            if self.fd is None:
+                self._open()
+                if self.fd is None:
+                    time.sleep(1)
+                    continue
             try:
-                out = subprocess.run(["tail", "-c", "262144", TRACE],
-                                     capture_output=True, timeout=10).stdout
-            except Exception:
-                time.sleep(1)
+                data = os.read(self.fd, 65536)
+            except OSError:
+                try:
+                    os.close(self.fd)
+                except OSError:
+                    pass
+                self.fd = None
+                time.sleep(0.5)
+                continue
+            if not data:
                 continue
             with self.lock:
-                for m in re.finditer(rb"sem=0x([0-9a-fA-F]+)", out):
+                for m in re.finditer(rb"sem=0x([0-9a-fA-F]+)", data):
                     v = int(m.group(1), 16)
                     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):
+                for m in re.finditer(rb"pb=0x([0-9a-fA-F]+)", data):
                     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:
@@ -307,6 +335,7 @@ def main():
             last_pid = pid
             seen = {}
             seen_sems = set()
+            clear_trace()
             if sock is not None:
                 try:
                     sock.sendall(("#pid %d\n" % pid).encode())