path: root/tools/guest-diag/semfix_watch.py

#!/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/open_iov_vfio/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 read64(self, addr, n=1, cache=True):
        key = (addr, n)
        if cache and key in self._cache:
            return self._cache[key]
        out = self.hmp("x/%dgx 0x%x" % (n, addr))
        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):
        out = self.hmp("gva2gpa 0x%x" % va)
        m = re.search(r"gpa: 0x([0-9a-fA-F]+)", out)
        return int(m.group(1), 16) if m else 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):
    """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. Do not stop at the first hole: scan
    the whole range (bailing only when essentially everything misses) and
    return the pinned count, so the caller can retry the remainder later.
    """
    pages = min(max(size >> 12, 1) if size else PUSHBUF_MAX_PAGES,
                PUSHBUF_MAX_PAGES)
    pinned = 0
    for gv in gpu_vas:
        miss = 0
        i = 0
        while i < pages:
            g = q.gva2gpa(cpu + i * 0x1000)
            if not g:
                miss += 1
                if miss >= pages:
                    print("semfix: pushbuf va=0x%x: no page resolved (%d "
                          "tried), wrong cpu base" % (gv, pages), flush=True)
                    return pinned
                i += 1
                continue
            miss = 0
            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 += 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


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