#!/usr/bin/env python3 # semfix_watch.py - keep UVM VA->PA translations fresh from guest truth # (host-side dev tool, no guest changes). # # Two jobs: # 1. Semaphore pool: the emulation's content-scanned map can hold a *stale* # translation for a tracking-semaphore pool page (same VA, previous # client's page), so GPU-tracking-semaphore waits never clear. The # tracking semaphore object (kprobe `uvmt` gives its kernel pointer) has # `semaphore.page->memory->vas[]` with the pool page's CPU and GPU VAs; # QEMU turns the CPU VA into a GPA (`gva2gpa`) which we pin. # 2. CE pushbuffer: the pushes live in a UVM rm_mem with GPU VA # `0x120000000` whose CPU VA is vmalloc-mapped guest RAM. Pinning the # range lets the host pump read the pushes directly and execute the PTE # memsets/copies instead of only acknowledging their releases. # # Pushbuffer chain (uvm 570.144 layouts): # uvm_channel_t +0x00 = pool # uvm_channel_pool_t +0x00 = manager # uvm_channel_manager_t +0x08 = pushbuffer # uvm_pushbuffer_t +0x08 = memory (uvm_rm_mem_t) # uvm_rm_mem_t +0x10 = vas[] (CPU @ +0x30, GPU @ +0x38 observed) # The channel is reached three ways: kprobe on # __uvm_push_begin_acquire_on_channel_with_info (`uvch` -> ch), kprobe on # uvm_pushbuffer_begin_push (`uvbp` -> pb), or the tracking-sem kptr `sem` # from `uvmt` (sem sits inside the channel struct at a fixed offset). # # Sources: guest kprobes, QEMU QMP `x/..gx` reads and `gva2gpa`. # # Usage: tools/guest-diag/semfix_watch.py [interval-seconds] # Runs until killed; prints one line per newly resolved/pinned page and a # structured diagnostic line whenever a chain resolution misses. import json import os import re import select import socket import subprocess import sys import time HERE = os.path.dirname(os.path.abspath(__file__)) VMEXEC = os.path.normpath(os.path.join(HERE, "..", "..", "scripts", "vgpu_vm.py")) PARAM = "/sys/module/vgsp/parameters" 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", "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")) class Qmp: def __init__(self, path): self.s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) self.s.connect(path) self.s.settimeout(15) self._rd() self.cmd("qmp_capabilities") self._cache = {} def _rd(self): """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: self.s.sendall(json.dumps({"execute": name, "arguments": args}).encode() + b"\n") return json.loads(self._rd() or "{}") except Exception: return {} def hmp(self, line): return self.cmd("human-monitor-command", **{"command-line": line}).get("return", "") def ncpu(self): if getattr(self, "_ncpu", 0): return self._ncpu n = 0 for line in self.hmp("info cpus").splitlines(): if line.strip().startswith("CPU #") or line.strip().startswith("* CPU #"): n += 1 self._ncpu = n or 4 return self._ncpu def _set_cpu(self, n): """Select the monitor's current CPU. HMP `x`/`gva2gpa` translate with the selected CPU's CR3: while that CPU runs user code under PTI the kernel half is hidden, so a translation of a kernel VA fails. Cycling CPUs until one is in kernel mode makes the reads work during active clients (the client spins/sleeps in kernel code on at least one CPU).""" if getattr(self, "_cpu", 0) == n: return self.hmp("cpu %d" % n) self._cpu = n def _cpu_order(self): cur = getattr(self, "_cpu", 0) return [cur] + [c for c in range(self.ncpu()) if c != cur] def read64(self, addr, n=1, cache=True): key = (addr, n) if cache and key in self._cache: return self._cache[key] out = "" for c in self._cpu_order(): self._set_cpu(c) out = self.hmp("x/%dgx 0x%x" % (n, addr)) m = re.search(r":(.*)$", out, re.S) if m and re.findall(r"0x([0-9a-fA-F]+)", m.group(1))[:n]: break m = re.search(r":(.*)$", out, re.S) if not m: return [] vals = [int(x, 16) for x in re.findall(r"0x([0-9a-fA-F]+)", m.group(1))][:n] if cache: self._cache[key] = vals return vals def gva2gpa(self, va): for c in self._cpu_order(): self._set_cpu(c) out = self.hmp("gva2gpa 0x%x" % va) m = re.search(r"gpa: 0x([0-9a-fA-F]+)", out) if m: return int(m.group(1), 16) return None def is_kern(v): return v and (v >> 48) == 0xFFFF def is_gpu(v): return v and GPU_VA_LO <= v < GPU_VA_HI def is_pushbuf_gpu(v): return v and PUSHBUF_VA_LO <= v < PUSHBUF_VA_HI def is_vmalloc(v): # pushbuffer rm_mems are vmalloc-mapped (the sem pools are direct-map) return v and 0xFFFFC90000000000 <= v < 0xFFFFE90000000000 def diag_add(d, k, n=1): d[k] = d.get(k, 0) + n def diag_str(d): return " ".join("%s=%s" % (k, v) for k, v in sorted(d.items())) def guest_trace(): """(sem kptrs, pushbuffer kptrs, channel kptrs) from the trace, then clear. Read *before* clearing: the create-time probes fire once per client and would be wiped by a clear that races them. """ inner = ("tail -c 131072 /sys/kernel/tracing/trace " "| grep -E \"uvmt|uvch|uvbp|uvpbf\" " "| 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 except Exception: return set(), set(), set() sems = {int(m, 16) for m in re.findall(r"sem=0x([0-9a-fA-F]+)", out)} pbs = {int(m, 16) for m in re.findall(r"pb=0x([0-9a-fA-F]+)", out)} chans = {int(m, 16) for m in re.findall(r"ch=0x([0-9a-fA-F]+)", out)} return sems, pbs, chans def vas_from_mem(q, mem, d): """uvm_rm_mem_t pointer -> (gpu_vas, cpu, gpa) or None.""" if not is_kern(mem): diag_add(d, "badmem") return None vas = q.read64(mem + 0x10, 16) if not vas: diag_add(d, "empty_vas") return None gpu_vas = [v for v in vas if is_gpu(v)] cpu = next((v for v in vas if is_vmalloc(v)), None) if not gpu_vas: diag_add(d, "no_gpuva") return None if not cpu: diag_add(d, "no_vmalloc") return None gpa = q.gva2gpa(cpu) if not gpa: diag_add(d, "no_gpa") return None size = q.read64(mem + 0x60) return gpu_vas, cpu, gpa, (size[0] if size else 0) def pushbuffer_of_manager(q, mgr, d): pb = q.read64(mgr + 8) if not pb or not is_kern(pb[0]): diag_add(d, "stop_manager") return None mem = q.read64(pb[0] + 8) if not mem or not is_kern(mem[0]): diag_add(d, "stop_pushbuffer") return None return vas_from_mem(q, mem[0], d) def pushbuffer_of_pool(q, pool, d): mgr = q.read64(pool) if not mgr or not is_kern(mgr[0]): diag_add(d, "stop_pool") return None return pushbuffer_of_manager(q, mgr[0], d) def best_candidate(found, d): """Prefer a chain whose rm_mem has a GPU VA in the CE pushbuffer range.""" if not found: return None for res in found: if any(is_pushbuf_gpu(v) for v in res[0]): return res gpu_vas = found[0][0] diag_add(d, "range_miss") d["_near"] = "0x%x" % gpu_vas[0] return found[0] def chains_from_window(q, words, d): """Interpret kernel pointers in the scanned window as channel/pool/ manager/pushbuffer pointers and walk the chain.""" cands = list(dict.fromkeys(w for w in words if is_kern(w))) d["cands"] = len(cands) found = [] for w in cands[:CHAN_MAX_CANDS]: res = pushbuffer_of_pool(q, w, d) if res: found.append(res) continue ch = q.read64(w) if ch and is_kern(ch[0]): res = pushbuffer_of_pool(q, ch[0], d) if res: found.append(res) continue res = pushbuffer_of_manager(q, w, d) if res: found.append(res) continue pb = q.read64(w + 8) if pb and is_kern(pb[0]): res = vas_from_mem(q, pb[0], d) if res: found.append(res) return best_candidate(found, d) def resolve_from_channel(q, ch, d): """uvm_channel_t * -> pool -> manager -> pushbuffer -> rm_mem.""" d["ch"] = 1 pool = q.read64(ch) if not pool or not is_kern(pool[0]): diag_add(d, "stop_channel") return None res = pushbuffer_of_pool(q, pool[0], d) return best_candidate([res] if res else [], d) def resolve_from_pushbuffer(q, pb, d): d["pb"] = 1 mem = q.read64(pb + 8) if not mem or not is_kern(mem[0]): diag_add(d, "stop_pushbuffer") return None res = vas_from_mem(q, mem[0], d) return best_candidate([res] if res else [], d) def resolve_from_sem(q, sem, d): """Scan the pages around the channel's tracking_sem for the pool pointer (channel+0 = pool) and walk pool -> manager -> pushbuffer -> rm_mem.""" d["sem"] = 1 page = sem & ~0xFFF words = [] nback = max(1, CHAN_WINDOW_BACK // 0x1000) for i in range(nback, -2, -1): chunk = q.read64(page - i * 0x1000, 512) if not chunk: diag_add(d, "empty_chunks") continue words += chunk diag_add(d, "chunks") d["words"] = len(words) if not words: return None return chains_from_window(q, words, d) def resolve_pushbuffer_wide(q, sem, d): """Legacy 256 KB window scan (kept as a last resort).""" base = sem & ~0xFFF words = [] for i in range(64): chunk = q.read64(base - 0x40000 + i * 0x1000, 512) if not chunk: diag_add(d, "empty_wide_chunks") continue words += chunk d["wide_words"] = len(words) d["wide_empty"] = d.get("empty_wide_chunks", 0) if not words: return None cands = list(dict.fromkeys(w for w in words if is_kern(w) and (w >> 32) == (sem >> 32))) d["wide_cands"] = len(cands) found = [] for M in cands[:400]: res = pushbuffer_of_manager(q, M, d) if res: found.append(res) continue pb = q.read64(M + 8) if pb and is_kern(pb[0]): res = vas_from_mem(q, pb[0], d) if res: found.append(res) return best_candidate(found, d) def pin(va, pa): cmd = "printf '0x%x,0x%x' > %s/sema_map_add" % (va, pa, 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) # async queue full: drain and retry 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 sem_state(q, sem): """(index, queued, payload) for a tracking-semaphore kptr, or None. payload is the u32 the client's update_completed_value reads: the pool page at the semaphore's index, through the CPU mapping the struct names. Printed when it changes so the last stall before a failure is visible. """ vals = q.read64(sem, 11) if len(vals) < 2 or not is_kern(vals[0]): return None page = vals[0] pv = q.read64(page) if not pv or not is_kern(pv[0]): return None mem = pv[0] vas = q.read64(mem + 0x10, 16) cpu = next((v for v in vas if is_kern(v)), None) idx = vals[1] & 0xffff w10 = vals[2] if len(vals) > 2 else 0 queued = vals[5] if len(vals) > 5 else 0 payload = None if cpu: w = q.read64(cpu + idx * 4) payload = w[0] if w else None return (idx, w10, queued, payload) def resolve_sem_pool(q, sem): page = q.read64(sem) if not page or not page[0]: return [] mem = q.read64(page[0]) if not mem or not mem[0]: return [] vas = q.read64(mem[0] + 0x10, 16) cpu = next((v for v in vas if is_kern(v)), None) gpu_vas = [v for v in vas if is_gpu(v)] if not cpu or not gpu_vas: return [] gpa = q.gva2gpa(cpu) if not gpa: return [] return [(gv, gpa) for gv in gpu_vas] def pin_pushbuffer(q, gpu_vas, cpu, gpa, size=0, retries=8): """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). gva2gpa of a kernel vmalloc VA intermittently fails while the sampled vCPU executes in user mode (page-table isolation hides the kernel half), so a pass can miss a run of pages. Retry the missing pages in-call (bounded, with a short sleep) before reporting; the caller also retries incomplete pins on later loops. """ pages = min(max(size >> 12, 1) if size else PUSHBUF_MAX_PAGES, PUSHBUF_MAX_PAGES) pinned = 0 for gv in gpu_vas: pending = list(range(pages)) attempts = 0 while pending and attempts < retries: attempts += 1 rest = [] i = 0 while i < len(pending): page = pending[i] g = q.gva2gpa(cpu + page * 0x1000) if not g: rest.append(page) i += 1 continue n = 1 last = g while i + n < len(pending): npg = pending[i + n] if npg != page + n: break gn = q.gva2gpa(cpu + npg * 0x1000) if gn != last + 0x1000: break last = gn n += 1 if not pin_bulk(gv + page * 0x1000, g, n * 0x1000): print("semfix: pushbuf va=0x%x: bulk pin failed at page %d" % (gv, page), flush=True) return pinned pinned += n i += n if rest: time.sleep(0.1) pending = rest if pending: print("semfix: pushbuf va=0x%x: %d/%d pages unresolved after %d " "attempts (wrong cpu base or PTI)" % (gv, len(pending), pages, attempts), flush=True) 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 def main(): interval = float(sys.argv[1]) if len(sys.argv) > 1 else 3.0 q = Qmp(QMP_SOCK) seen = set() seen_pb = set() seen_ch = set() pinned_ranges = set() sem_last = {} 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, size) print("semfix: watching (interval %.1fs)" % interval, flush=True) while True: sems, pbs, chans = guest_trace() for sem in sems: try: st = sem_state(q, sem) except Exception: st = None if st != sem_last.get(sem): sem_last[sem] = st if st: print("semfix: sem=0x%x idx=%d +10=0x%x +28=0x%x payload=%s" % (sem, st[0], st[1], st[2], "0x%x" % st[3] if st[3] is not None else "?"), flush=True) for ch in chans - seen_ch: d = {} try: res = resolve_from_channel(q, ch, d) except Exception as e: res = None d["exception"] = str(e)[:40] if res: seen_ch.add(ch) pin_range_once(*res) print("semfix: ch=0x%x -> %s" % (ch, diag_str(d)), flush=True) else: print("semfix: ch=0x%x pushbuf miss [%s]" % (ch, diag_str(d)), flush=True) for pb in pbs - seen_pb: d = {} try: res = resolve_from_pushbuffer(q, pb, d) except Exception as e: res = None d["exception"] = str(e)[:40] if res: seen_pb.add(pb) pin_range_once(*res) print("semfix: pb=0x%x -> %s" % (pb, diag_str(d)), flush=True) else: print("semfix: pb=0x%x pushbuf miss [%s]" % (pb, diag_str(d)), flush=True) for sem in sems - seen: d = {} try: pairs = resolve_sem_pool(q, sem) except Exception as e: print("semfix: sem=0x%x resolve failed: %s" % (sem, e), flush=True) continue if not pairs: print("semfix: sem=0x%x unresolved (retry)" % sem, flush=True) continue seen.add(sem) for gv, gpa in pairs: ok = pin(gv, gpa) print("semfix: sem=0x%x va=0x%x -> gpa=0x%x pin=%s" % (sem, gv, gpa, "ok" if ok else "FAILED"), flush=True) d = {} try: res = resolve_from_sem(q, sem, d) except Exception as e: res = None d["exception"] = str(e)[:40] if not res: for k in ("chunks", "words", "cands", "stop_pool", "no_gpuva", "no_vmalloc", "empty_vas"): d.pop(k, None) try: res = resolve_pushbuffer_wide(q, sem, d) except Exception as e: res = None d["exception"] = str(e)[:40] if res: pin_range_once(*res) print("semfix: sem=0x%x pushbuf -> %s" % (sem, diag_str(d)), flush=True) else: print("semfix: sem=0x%x pushbuf not found [%s]" % (sem, diag_str(d)), flush=True) time.sleep(interval) if __name__ == "__main__": main()