diff options
| -rw-r--r-- | kernel/gsp_rm_emulation.c | 8 | ||||
| -rw-r--r-- | kernel/gsp_sema_service.c | 297 | ||||
| -rw-r--r-- | kernel/gsp_sema_service.h | 4 | ||||
| -rw-r--r-- | tools/guest-diag/pin_gr_stream.py | 178 | ||||
| -rwxr-xr-x | tools/guest-diag/pin_supervisor.sh | 138 | ||||
| -rwxr-xr-x | tools/guest-diag/semfix_uvmt.py | 152 | ||||
| -rw-r--r-- | tools/guest-diag/semfix_watch.py | 68 |
7 files changed, 751 insertions, 94 deletions
diff --git a/kernel/gsp_rm_emulation.c b/kernel/gsp_rm_emulation.c index 8cd82d1..f6ca2c1 100644 --- a/kernel/gsp_rm_emulation.c +++ b/kernel/gsp_rm_emulation.c @@ -3812,6 +3812,14 @@ void gsp_rm_emu_reset(struct gsp_emu_context *ctx) "vgsp-rm: reset dropped %u stale event record(s)", stale); + /* + * m177: RM reload means a new guest boot: the guest's direct-map base + * and every content-scanned translation may have changed. Raise the + * boot-scoped wipe (executed on the sema svc thread; client session + * flushes intentionally keep that state - see svc_session_flush). + */ + gsp_sema_service_guest_reset(); + if (!rm->parsed) return; diff --git a/kernel/gsp_sema_service.c b/kernel/gsp_sema_service.c index 8f23178..7b12ecc 100644 --- a/kernel/gsp_sema_service.c +++ b/kernel/gsp_sema_service.c @@ -483,6 +483,7 @@ static u32 svc_ring_n; * guest's last-channel close is also the global empty transition. */ static u8 svc_sess_flush; +static u8 svc_guest_reset_req; /* m177: guest RM reload; wiped on svc thread */ static u8 svc_sess_active; /* a channel of a session has registered (until flush) */ static u32 svc_sess_gen; /* m176: bumped by every executed session flush; * stamps channels/scrubbers so per-session learned @@ -1975,13 +1976,133 @@ static void svc_pd_discover(struct nv_vgpu_device *vgpu) svc_pd_cands, svc_pd_best, n); } +/* + * m177c: a SECOND discovered-directory slot for the UVM SYS VA space (the + * 0x2xxxxxxx family: GR rings/pushes, QMD windows, counter pages). The + * primary slot (above) is validated against rm_mem-content pairs - the + * 0x1xxxxxxx pool family - and cannot walk the UVM space, so without pins + * the counter page (0x20400fXXX) never resolved: the guest's KMD/UVM + * epoch waited on completions that could never be written and re-ran its + * ~103 s retry loop forever (r194/r195: 'rel-apply ... slot pending' x8, + * then silence). Samples for this slot are content-VERIFIED + * (push_va -> push_pa) pairs from the GR content discovery and the CE + * push pairing: the push bytes themselves are the proof, and they live in + * exactly the space this directory maps. + */ +#define SVC_PD_RING_SAMPLES 8 +static u64 svc_pd_r_v[SVC_PD_RING_SAMPLES]; +static u64 svc_pd_r_pa[SVC_PD_RING_SAMPLES]; +static u32 svc_pd_r_n; +static u64 svc_pd_r_base; +static int svc_pd_r_ap; +static u8 svc_pd_r_found; +static u8 svc_pd_r_attempts; +static u64 svc_pd_r_ev_hash; + +static void svc_pd_ring_note(u64 va, u64 pa) +{ + u32 i; + + if (!va || !pa) + return; + va &= ~0xfffULL; + pa &= ~0xfffULL; + for (i = 0; i < svc_pd_r_n; i++) + if (svc_pd_r_v[i] == va) + return; + if (svc_pd_r_n < SVC_PD_RING_SAMPLES) { + svc_pd_r_v[svc_pd_r_n] = va; + svc_pd_r_pa[svc_pd_r_n] = pa; + svc_pd_r_n++; + } +} + +static void svc_pd_ring_discover(struct nv_vgpu_device *vgpu) +{ + u64 chunk = 64 * 1024; + u8 *buf; + u64 gpa; + u64 ev = svc_pd_r_n; + u32 i; + + if (svc_pd_r_found || svc_pd_r_attempts >= SVC_PD_ATTEMPTS_MAX) + return; + if (svc_pd_r_n < SVC_PD_MIN_HITS) + return; + for (i = 0; i < svc_pd_r_n; i++) + ev = (ev + svc_pd_r_v[i] + (svc_pd_r_pa[i] << 12)) * 1000003; + if (svc_pd_r_attempts && ev == svc_pd_r_ev_hash) + return; + svc_pd_r_ev_hash = ev; + svc_pd_r_attempts++; + + buf = kmalloc(chunk, GFP_KERNEL); + if (!buf) + return; + for (gpa = 0; gpa < SEMA_PA_MAX && !svc_pd_r_found; gpa += chunk) { + u32 len = (u32)min_t(u64, chunk, SEMA_PA_MAX - gpa); + u64 off; + + if (svc_read(vgpu, gpa, buf, len)) + continue; + for (off = 0; off + 8 <= len && !svc_pd_r_found; + off += SEMA_PAGE_SIZE) { + u64 e; + u64 a = 0; + int eap = 0; + u32 hits = 1; + + memcpy(&e, buf + off, sizeof(e)); + if ((e & 1) || !((e >> 1) & 0x3)) + continue; + /* First sample rejects almost every candidate. */ + if (svc_gmmu_walk(vgpu, gpa + off, 4, svc_pd_r_v[0], + SVC_APERTURE_SYS, &a, &eap, false) || + (a & ~0xfffULL) != svc_pd_r_pa[0]) + continue; + for (i = 1; i < svc_pd_r_n; i++) { + if (!svc_gmmu_walk(vgpu, gpa + off, 4, + svc_pd_r_v[i], + SVC_APERTURE_SYS, &a, &eap, + false) && + (a & ~0xfffULL) == svc_pd_r_pa[i]) + hits++; + } + if (hits >= SVC_PD_MIN_HITS) { + svc_pd_r_base = gpa + off; + svc_pd_r_ap = SVC_APERTURE_SYS; + svc_pd_r_found = 1; + pr_info("sema-svc: discovered UVM ring-space page directory 0x%llx (ram), %u/%u content-verified samples\n", + svc_pd_r_base, hits, svc_pd_r_n); + } + } + } + kfree(buf); +} + static int svc_walk_discovered(struct nv_vgpu_device *vgpu, u64 va, u64 *addr, int *ap) { - if (!svc_pd_found) - return -ENOENT; - return svc_gmmu_walk(vgpu, svc_pd_base, svc_pd_entries, va, - svc_pd_ap, addr, ap, false); + int rc = -ENOENT; + + if (svc_pd_found) + rc = svc_gmmu_walk(vgpu, svc_pd_base, svc_pd_entries, va, + svc_pd_ap, addr, ap, false); + if (rc && svc_pd_r_found) + rc = svc_gmmu_walk(vgpu, svc_pd_r_base, 4, va, + svc_pd_r_ap, addr, ap, false); + /* + * m177c: the ring-space directory discovery runs on new samples + * only (fingerprint + attempt cap inside). A first call with + * enough verified ring/push samples performs one bounded scan; + * after that this is a flag check. + */ + if (rc && !svc_pd_r_found && svc_pd_r_n >= SVC_PD_MIN_HITS) + svc_pd_ring_discover(vgpu); + if (rc && svc_pd_r_found) + rc = svc_gmmu_walk(vgpu, svc_pd_r_base, 4, va, + svc_pd_r_ap, addr, ap, false); + return rc; } /* Learned translations: a walk costs a handful of guest reads, so cache the @@ -5836,13 +5957,13 @@ static struct { u64 pa_page; } svc_cnt_der[SVC_CNT_DER_MAX]; static u32 svc_cnt_der_n; +static u32 svc_cnt_der_log; static int svc_counter_derive(struct nv_vgpu_device *vgpu, u64 va, u64 *pa_out) { u64 page = va & ~(u64)(SEMA_PAGE_SIZE - 1); u64 off = va & (SEMA_PAGE_SIZE - 1); u64 pa = 0; - int ap = SVC_APERTURE_SYS; u32 probe = 0; u32 i; bool fresh = true; @@ -5855,13 +5976,57 @@ static int svc_counter_derive(struct nv_vgpu_device *vgpu, u64 va, u64 *pa_out) } } if (fresh) { - if (svc_walk_vaspaces(vgpu, page, &pa, &ap) && - svc_walk_discovered(vgpu, page, &pa, &ap) && - svc_targeted_resolve(vgpu, page, &pa)) - return -ENOENT; - if (ap != SVC_APERTURE_SYS) + u64 p = 0; + int a = SVC_APERTURE_SYS; + bool ok = false; + const char *src = "none"; + + /* + * m177b: first READABLE source wins. The previous chain kept + * the first source that merely PRODUCED a translation and then + * failed -EIO on an unreadable page without trying the rest: + * the published directory can yield a stale/incorrect chain + * for a counter VA while the rm_mem-content scan would resolve + * it (r194: the counter page 0x20400f000 never resolved; its + * walk hit probed unreadable and targeted was never reached). + */ + if (!svc_walk_vaspaces(vgpu, page, &p, &a) && + a == SVC_APERTURE_SYS && + !svc_read_ap(vgpu, + (p & ~(u64)(SEMA_PAGE_SIZE - 1)) | off, + SVC_APERTURE_SYS, &probe, sizeof(probe))) { + ok = true; + src = "vaspaces"; + } + if (!ok && !svc_walk_discovered(vgpu, page, &p, &a) && + a == SVC_APERTURE_SYS && + !svc_read_ap(vgpu, + (p & ~(u64)(SEMA_PAGE_SIZE - 1)) | off, + SVC_APERTURE_SYS, &probe, sizeof(probe))) { + ok = true; + src = "discovered"; + } + if (!ok && !svc_targeted_resolve(vgpu, page, &p) && + !svc_read_ap(vgpu, + (p & ~(u64)(SEMA_PAGE_SIZE - 1)) | off, + SVC_APERTURE_SYS, &probe, sizeof(probe))) { + ok = true; + src = "targeted"; + } + if (!ok) { + if (svc_cnt_der_log < 16) { + svc_cnt_der_log++; + pr_info("sema-chan: counter page 0x%llx unreadable via vaspaces/discovered/targeted; slot stays pending\n", + page); + } return -ENOENT; - pa = (pa & ~(u64)(SEMA_PAGE_SIZE - 1)) | off; + } + pa = (p & ~(u64)(SEMA_PAGE_SIZE - 1)) | off; + if (svc_cnt_der_log < 16) { + svc_cnt_der_log++; + pr_info("sema-chan: counter page 0x%llx resolved via %s (pa=0x%llx)\n", + page, src, pa); + } } /* Read the slot before trusting the mapping: an unreadable page is * not registered, so the caller keeps the "slot pending" fallback. */ @@ -7717,7 +7882,26 @@ static void svc_session_flush(void) svc_pd_best = 0; svc_pd_cands = 0; svc_pd_ev_hash = 0; - svc.direct_base = 0; + /* + * m177c: the UVM ring-space directory is per-client guest memory + * (and its samples come from this session's rings): reset with the + * session like the primary slot. + */ + svc_pd_r_found = 0; + svc_pd_r_attempts = 0; + svc_pd_r_base = 0; + svc_pd_r_ev_hash = 0; + svc_pd_r_n = 0; + /* + * m177: svc.direct_base is NOT cleared here. It is the guest kernel's + * direct-map base - a per-BOOT constant, identical for every client + * session of the same guest boot - and it is the gate and input of + * the pin-free resolution paths (svc_targeted_resolve). Clearing it + * per session forced every client to wait for a rebuild scan while + * its first cudaMalloc polled, and made the module depend on the + * host pin tools entirely. It is wiped on a guest reboot by + * svc_guest_reset_wipe() (gsp_sema_service_guest_reset()). + */ svc_pending_n = 0; svc_pending_next = 0; /* @@ -7875,6 +8059,52 @@ static void svc_session_flush(void) } /* + * m177: boot-scoped wipe for a GUEST boot (RM reload). Unlike the + * per-client session flush this clears the state that is valid for one + * guest boot only: the direct-map base and every content-scanned table. + * Runs on the svc thread (svc_guest_reset_wipe is invoked from the thread + * loop below), the only writer of this state. + */ +static void svc_guest_reset_wipe(void) +{ + svc.direct_base = 0; + svc.map_n = 0; + svc_pd_found = 0; + svc_pd_attempts = 0; + svc_pd_best = 0; + svc_pd_cands = 0; + svc_pd_ev_hash = 0; + svc_pd_r_found = 0; + svc_pd_r_attempts = 0; + svc_pd_r_base = 0; + svc_pd_r_ev_hash = 0; + svc_pd_r_n = 0; + svc_pb_n = 0; + svc_pool_boot_n = 0; + svc_pool_boot_next = 0; + svc_pool_pa_n = 0; + svc_pool_pa_next = 0; + memset(svc_pool_pa, 0, sizeof(svc_pool_pa)); + memset(tr_failed, 0, sizeof(tr_failed)); + tr_fail_next = 0; + memset(svc.lat, 0, sizeof(svc.lat)); + pr_info("sema-svc: guest reset: boot-scoped discovery state wiped (direct_base, map, caches)\n"); +} + +/* + * m177: called by gsp_rm_emu_reset() when the guest's GSP-RM is (re)loaded + * - a new guest boot, where every VA->PA relation can have changed. The + * request is latched and executed on the svc thread, which owns the + * discovery state (a cross-thread wipe here would race svc_rebuild). + */ +void gsp_sema_service_guest_reset(void) +{ + WRITE_ONCE(svc_guest_reset_req, 1); + if (svc.task) + wake_up_process(svc.task); +} + +/* * Loud, bounded telemetry for a starved deferred release (ISSUE-0001). * Called from the two "no translation yet, keep" paths of * svc_apply_pending() with an incremented try count; logs at 4/64/1024 @@ -9497,6 +9727,9 @@ static bool svc_gr_content_pin(struct nv_vgpu_device *vgpu, u32 rank, svc_pin_store(push_va, push_pa, SVC_APERTURE_SYS); svc_lat_store(push_va, push_pa, SVC_APERTURE_SYS); svc_mark_hot(push_pa); + /* m177c: a content-verified UVM-space pair - feed the ring-space + * directory discovery (counter pages resolve through it). */ + svc_pd_ring_note(push_va, push_pa); if (ring_ap == SVC_APERTURE_FB) pr_info("sema-chan: content-discovered GR ring rank=%u va=0x%llx -> fb 0x%llx (entry0 va=0x%llx len=%u; push pa=0x%llx; userd rc=%d put=%u get=%u)\n", rank, ring_va, ring_pa, push_va, len, push_pa, prc, @@ -10367,6 +10600,9 @@ static void svc_pb_pair(struct nv_vgpu_device *vgpu, struct svc_channel *ch) } svc_pb[hit_i].va = va; set_bit(ch->ce_index, &svc_pb_done); + /* m177c: the base page was content-read (svc_chan_entry0) - a + * verified UVM-space sample for the ring-space directory. */ + svc_pd_ring_note(va, hit_pa); /* * The pinned pairs are verified samples of UVM's own page directory; @@ -16214,10 +16450,20 @@ static void svc_pump_channels(struct nv_vgpu_device *vgpu) ch->defer_since = 0; ch->unread_va = 0; ch->unread_until = 0; - if (gsp_sema_fake_rel || - gsp_sema_rel_recover) - svc_replay_note(ch, idx, - put); + /* + * m177: deliver any + * withheld set even + * on the GR stale-skip + * path. GR channels + * have no rescue and + * their consume-time + * delivery gate is + * below; without this + * the slot strands + * until the sweep. + */ + svc_replay_note(ch, idx, + put); ch->last_get = (idx + 1) % ch->gpfifo_entries; svc.gpfifo_skipped++; @@ -16253,7 +16499,14 @@ static void svc_pump_channels(struct nv_vgpu_device *vgpu) } } - if (ch->engine_type && svc_rel_count != rel_n_iter) + /* + * m177: GR channels (engine_type == 0) have no rescue and + * no CE-shaped delivery branches above; deliver any + * withheld set for this entry whenever the entry is + * consumed so a held release can never strand until the + * 10 s sweep (evtest3's st[2] sync hang). + */ + if (svc_rel_count != rel_n_iter || !ch->engine_type) svc_replay_note(ch, idx, put); ch->last_get = (idx + 1) % ch->gpfifo_entries; ch->rescue_fail = 0; @@ -17197,6 +17450,16 @@ static int svc_thread(void *arg) } /* + * m177: a guest RM (re)load means a new guest boot: wipe the + * boot-scoped discovery state here, on the thread that owns + * it. Raised by gsp_sema_service_guest_reset(). + */ + if (READ_ONCE(svc_guest_reset_req)) { + svc_guest_reset_req = 0; + svc_guest_reset_wipe(); + } + + /* * STALL FIX (r189): expire withheld release sets whose entry * stopped being re-decoded (the deadline flush is only * reachable from the entry path). Forward-only delivery to a diff --git a/kernel/gsp_sema_service.h b/kernel/gsp_sema_service.h index e0f1587..b8c4dfb 100644 --- a/kernel/gsp_sema_service.h +++ b/kernel/gsp_sema_service.h @@ -81,6 +81,10 @@ struct sema_vaspace_desc { void gsp_sema_service_vaspace_add(const struct sema_vaspace_desc *desc); void gsp_sema_service_vaspace_del(u32 h_vaspace); +/* m177: guest RM reload = new guest boot; wipe boot-scoped discovery state + * (direct-map base, content-scanned tables) on the svc thread. Client + * session flushes keep that state - see svc_session_flush. */ +void gsp_sema_service_guest_reset(void); int gsp_sema_service_start(void); void gsp_sema_service_stop(void); diff --git a/tools/guest-diag/pin_gr_stream.py b/tools/guest-diag/pin_gr_stream.py index dff2d2d..0072b20 100644 --- a/tools/guest-diag/pin_gr_stream.py +++ b/tools/guest-diag/pin_gr_stream.py @@ -32,12 +32,22 @@ LO = os.environ.get("PINWIN_LO", "0x200000000") HI = os.environ.get("PINWIN_HI", "0x206000000") PAMAX = os.environ.get("PINWIN_PA_MAX", "0x180000000") SEMSTREAM = os.environ.get("SEM_STREAM_FILE", "/tmp/opencode/sem_stream.txt") +# Liveness file checked by pin_supervisor.sh (mtime must stay < 30 s). +HEARTBEAT_FILE = os.environ.get("PIN_GR_HEARTBEAT", "/tmp/opencode/pin_gr_heartbeat") def log(msg): print("[%s] %s" % (time.strftime("%H:%M:%S"), msg), flush=True) +def _heartbeat(): + try: + with open(HEARTBEAT_FILE, "w") as f: + f.write("%.3f\n" % time.time()) + except OSError: + pass + + def _mirror(line): """Append guest control lines (#pid/#sem) to the shared host file so the single-QMP watcher can consume them without guest-agent round trips.""" @@ -60,7 +70,8 @@ def vexec(cmd, timeout=40): def _start_cmd(): with open(GUEST_SCRIPT) as f: b64 = subprocess.run(["base64", "-w0"], stdin=f, - capture_output=True, text=True).stdout + capture_output=True, text=True, + timeout=10).stdout prefixes = os.environ.get("CLIENT_PREFIXES", "cdt,va,victory") vma_every = os.environ.get("STREAM_VMA_EVERY", "1") # Kill EVERY existing scanner, not just the pid-file one: a retry after @@ -90,6 +101,7 @@ def install_and_start(): # or a stale qga client); retry a few times instead of giving up, because # a boot without the scanner streams no pins and the client then hangs. for attempt in range(5): + _heartbeat() out = vexec(_start_cmd(), 60) if any(ch.isdigit() for ch in out): return True @@ -101,8 +113,12 @@ def install_and_start(): def inject(va, pa, ln): cmd = "printf '0x%x,0x%x,0x%x,1' > %s/sema_map_add" % (va, pa, ln, PARAM) for _ in range(60): - r = subprocess.run(["sudo", "-n", "sh", "-c", cmd], - capture_output=True, text=True) + try: + r = subprocess.run(["sudo", "-n", "sh", "-c", cmd], + capture_output=True, text=True, timeout=10) + except subprocess.TimeoutExpired: + time.sleep(0.05) + continue if r.returncode == 0: return True time.sleep(0.05) @@ -115,8 +131,12 @@ def scrub_register(fin_pa, hsem_pa): cmd = ("printf '0x%x,0x%x' > %s/sema_scrubbuf_add" % (fin_pa, hsem_pa, PARAM)) for _ in range(60): - r = subprocess.run(["sudo", "-n", "sh", "-c", cmd], - capture_output=True, text=True) + try: + r = subprocess.run(["sudo", "-n", "sh", "-c", cmd], + capture_output=True, text=True, timeout=10) + except subprocess.TimeoutExpired: + time.sleep(0.05) + continue if r.returncode == 0: log("scrub registered fin=0x%x hsem=0x%x" % (fin_pa, hsem_pa)) return True @@ -166,81 +186,116 @@ def _demand_loop(): log("demand: sent %d stalled VA(s) to guest" % len(todo)) +_DEMAND_STARTED = False + + def main(): + global _DEMAND_STARTED + _heartbeat() if not install_and_start(): log("WARN: could not start guest scanner (will retry via vsock wait)") - try: - threading.Thread(target=_demand_loop, daemon=True).start() - except Exception as e: - log("WARN: demand thread failed to start: %s" % e) + if not _DEMAND_STARTED: + _DEMAND_STARTED = True + try: + threading.Thread(target=_demand_loop, daemon=True).start() + except Exception as e: + log("WARN: demand thread failed to start: %s" % e) srv = socket.socket(socket.AF_VSOCK, socket.SOCK_STREAM) srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) srv.bind((socket.VMADDR_CID_ANY, PORT)) + srv.settimeout(10) srv.listen(4) log("listening on vsock port %d (window %s..%s)" % (PORT, LO, HI)) seen = {} pin_total = 0 - buf = b"" + last_conn = time.time() + last_reinstall = 0.0 while True: try: conn, _ = srv.accept() + except socket.timeout: + now = time.time() + # No guest scanner connection for 30 s: it died or never started + # (a boot without it streams no pins and every client then hangs). + # Re-run install_and_start() at most once per 15 s. + if now - last_conn > 30.0 and now - last_reinstall >= 15.0: + last_reinstall = now + log("reinstall attempt: no guest connection for %.0fs" + % (now - last_conn)) + try: + install_and_start() + except Exception as e: + log("reinstall failed: %s" % e) + continue except OSError as e: log("accept failed: %s" % e) time.sleep(1) continue + last_conn = time.time() + _heartbeat() conn.settimeout(5) log("guest connected") buf = b"" runs = [] - while True: - try: - chunk = conn.recv(1 << 20) - except socket.timeout: - if runs: - _flush(runs, seen) - runs = [] - continue - except OSError: - break - if not chunk: - break - buf += chunk - while b"\n" in buf: - line, buf = buf.split(b"\n", 1) - text = line.decode(errors="replace").strip() - if not text: - continue - if text.startswith("#pid"): - m = re.search(r"(\d+)", text) - seen = {} - log("guest reports cdt pid %s" % (m.group(1) if m else "?")) - _mirror(text) - continue - if text.startswith("#sem"): - _mirror(text) - continue - if text.startswith("#pb"): - _mirror(text) - continue - if text.startswith("#put"): - log(text) - _mirror(text) + try: + while True: + try: + chunk = conn.recv(1 << 20) + except socket.timeout: + if runs: + _flush(runs, seen) + runs = [] continue - if text.startswith("#scrub"): - m = re.search(r"0x([0-9a-fA-F]+),0x([0-9a-fA-F]+)", text) - _mirror(text) + except OSError: + break + if not chunk: + break + _heartbeat() + buf += chunk + while b"\n" in buf: + line, buf = buf.split(b"\n", 1) + text = line.decode(errors="replace").strip() + if not text: + continue + if text.startswith("#pid"): + m = re.search(r"(\d+)", text) + seen = {} + log("guest reports cdt pid %s" + % (m.group(1) if m else "?")) + _mirror(text) + continue + if text.startswith("#sem"): + _mirror(text) + continue + if text.startswith("#pb"): + _mirror(text) + continue + if text.startswith("#put"): + log(text) + _mirror(text) + continue + if text.startswith("#scrub"): + m = re.search(r"0x([0-9a-fA-F]+),0x([0-9a-fA-F]+)", + text) + _mirror(text) + if m: + scrub_register(int(m.group(1), 16), + int(m.group(2), 16)) + continue + m = re.match(r"^(0x[0-9a-f]+),(0x[0-9a-f]+)(?:,(\d+))?$", + text) if m: - scrub_register(int(m.group(1), 16), - int(m.group(2), 16)) - 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), - int(m.group(3)) if m.group(3) else 1)) - if runs: - pin_total += _flush(runs, seen) - runs = [] + runs.append((int(m.group(1), 16), + int(m.group(2), 16), + int(m.group(3)) if m.group(3) else 1)) + if runs: + pin_total += _flush(runs, seen) + runs = [] + except Exception as e: + # A transient error on one connection must not kill the process: + # log it, drop the connection and go back to accept(). + log("connection handler error: %s (continuing)" % e) log("guest disconnected (pins this session: %d)" % pin_total) try: conn.close() @@ -289,6 +344,7 @@ def _flush(runs, seen): CHUNK = 200 for start in range(0, len(script), CHUNK): batch = script[start:start + CHUNK] + _heartbeat() try: r = subprocess.run(["sudo", "-n", "sh", "-c", "\n".join(batch)], capture_output=True, text=True, timeout=60) @@ -322,4 +378,12 @@ def _flush(runs, seen): if __name__ == "__main__": - main() + # main() loops forever; only an unexpected exception exits it. Restart + # instead of dying silently - a dead listener means no pins and every + # later CUDA client hangs at first cudaMalloc. + while True: + try: + main() + except Exception as e: + log("FATAL: main() exited: %r - restarting in 2s" % (e,)) + time.sleep(2) diff --git a/tools/guest-diag/pin_supervisor.sh b/tools/guest-diag/pin_supervisor.sh new file mode 100755 index 0000000..5bbab0c --- /dev/null +++ b/tools/guest-diag/pin_supervisor.sh @@ -0,0 +1,138 @@ +#!/bin/bash +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2026 Open-IOV contributors +# pin_supervisor.sh - supervise the host-side pin tools (pin_gr_stream.py and +# semfix_uvmt.py) so a silent death cannot strand every later CUDA client at +# first cudaMalloc. +# +# Both tools write a heartbeat file; this script checks, every 5 s, that each +# process is alive (exact script path) and that its heartbeat mtime is < 30 s. +# On failure it kills the tool (exact path match) and restarts it, appending +# RESTART lines to the supervisor log. After 10 restarts in one hour for a +# tool it logs GIVEUP and exits 1. Only one semfix_uvmt instance is ever +# started (QEMU's QMP socket accepts a single client). +# +# Usage: pin_supervisor.sh LOGDIR TAG +# Logs: LOGDIR/12-pin-supervisor-TAG.log (supervisor) +# LOGDIR/12-pin_gr_stream-TAG.log, LOGDIR/12-semfix-TAG.log (tools) +set -uo pipefail + +LOGDIR="${1:?usage: pin_supervisor.sh LOGDIR TAG}" +TAG="${2:?usage: pin_supervisor.sh LOGDIR TAG}" + +HERE=$(cd "$(dirname "$0")" && pwd) +PIN="$HERE/pin_gr_stream.py" +SEMFIX="$HERE/semfix_uvmt.py" +PIN_HB="${PIN_GR_HEARTBEAT:-/tmp/opencode/pin_gr_heartbeat}" +SEMFIX_HB="${SEMFIX_HEARTBEAT_FILE:-/tmp/opencode/semfix_heartbeat}" +SUPLOG="$LOGDIR/12-pin-supervisor-$TAG.log" +PIN_LOG="$LOGDIR/12-pin_gr_stream-$TAG.log" +SEMFIX_LOG="$LOGDIR/12-semfix-$TAG.log" +WINDOW=3600 +CAP=10 +mkdir -p "$LOGDIR" /tmp/opencode + +log() { echo "[$(date +%H:%M:%S)] $*" | tee -a "$SUPLOG"; } + +tool_alive() { pgrep -f "python3 $1" >/dev/null 2>&1; } + +hb_age() { # age in seconds of the heartbeat file, 9999 if missing + local f="$1" m + m=$(stat -c %Y "$f" 2>/dev/null) || { echo 9999; return; } + echo $(( $(date +%s) - m )) +} + +kill_tool() { + pkill -f "python3 $1" 2>/dev/null + sleep 1 + if pgrep -f "python3 $1" >/dev/null 2>&1; then + pkill -9 -f "python3 $1" 2>/dev/null + sleep 1 + fi + return 0 +} + +start_pin() { + if tool_alive "$PIN"; then return 0; fi + touch "$PIN_HB" 2>/dev/null || true + ( setsid nohup env SEMFIX_COUNTER_RETRY=10 \ + python3 "$PIN" >> "$PIN_LOG" 2>&1 & ) + log "START pin_gr_stream" +} + +start_semfix() { + # pgrep exact-count check before start: if an instance already exists + # (ours or a leftover), never start a second one. + if tool_alive "$SEMFIX"; then + log "semfix already running - not starting another" + return 0 + fi + touch "$SEMFIX_HB" 2>/dev/null || true + ( setsid nohup env SEMFIX_SKIP_PB=0 SEMFIX_COUNTER_RETRY=10 \ + python3 "$SEMFIX" 0.5 >> "$SEMFIX_LOG" 2>&1 & ) + log "START semfix_uvmt" +} + +prune() { # $1 now, $2 list -> list with entries older than WINDOW dropped + local now="$1" t out="" + for t in $2; do + [ $(( now - t )) -lt "$WINDOW" ] && out="$out $t" + done + printf '%s' "$out" +} + +count_list() { # $1 list -> element count + local n=0 t + for t in $1; do n=$(( n + 1 )); done + echo "$n" +} + +restarts_pin="" +restarts_semfix="" + +restart() { # $1 label, $2 script path, $3 start function + local now list n + now=$(date +%s) + if [ "$1" = pin_gr_stream ]; then + restarts_pin=$(prune "$now" "$restarts_pin") + n=$(count_list "$restarts_pin") + else + restarts_semfix=$(prune "$now" "$restarts_semfix") + n=$(count_list "$restarts_semfix") + fi + if [ "$n" -ge "$CAP" ]; then + log "GIVEUP $1: $n restarts within ${WINDOW}s" + exit 1 + fi + kill_tool "$2" + if [ "$1" = pin_gr_stream ]; then + restarts_pin="$restarts_pin $now" + else + restarts_semfix="$restarts_semfix $now" + fi + log "RESTART $(( n + 1 )) $1 $(date +%Y-%m-%dT%H:%M:%S)" + "$3" +} + +check_tool() { # $1 label, $2 path, $3 heartbeat file, $4 start function + local age + age=$(hb_age "$3") + if tool_alive "$2" && [ "$age" -le 30 ]; then + return 0 + fi + if ! tool_alive "$2"; then + log "$1 not running (heartbeat age ${age}s) - restarting" + else + log "$1 heartbeat stale (${age}s) - restarting" + fi + restart "$1" "$2" "$4" +} + +log "pin_supervisor up (LOGDIR=$LOGDIR TAG=$TAG)" +start_pin +start_semfix +while true; do + sleep 5 + check_tool pin_gr_stream "$PIN" "$PIN_HB" start_pin + check_tool semfix_uvmt "$SEMFIX" "$SEMFIX_HB" start_semfix +done diff --git a/tools/guest-diag/semfix_uvmt.py b/tools/guest-diag/semfix_uvmt.py index cf54445..fa3ae01 100755 --- a/tools/guest-diag/semfix_uvmt.py +++ b/tools/guest-diag/semfix_uvmt.py @@ -53,6 +53,21 @@ SKIP_PB = os.environ.get("SEMFIX_SKIP_PB", "0") == "1" # Seconds between counter-page pin attempts (the guest-agent exec blocks for # its full timeout when the client saturates the guest). COUNTER_RETRY_S = float(os.environ.get("SEMFIX_COUNTER_RETRY", "10")) +# Liveness file checked by pin_supervisor.sh (mtime must stay < 30 s). +SEMFIX_HB_FILE = os.environ.get("SEMFIX_HEARTBEAT_FILE", + "/tmp/opencode/semfix_heartbeat") +# Every client pid ever observed. Module-level on purpose: the per-client +# reset must NOT clear it, so a replayed ring entry for an old pid can never +# masquerade as a new client and wipe the pin state. +seen_pids = set() + + +def _hb(): + try: + with open(SEMFIX_HB_FILE, "w") as f: + f.write("%.3f\n" % time.time()) + except OSError: + pass class Stream: @@ -269,6 +284,17 @@ def main(): counter_pids = set() counter_next_try = {} last_pid = None + pending_pid = None + pending_count = 0 + # Stale-sem tracking: consecutive unreadable reads, first-unreadable time, + # and the per-sem read skip deadline once a sem has failed 3 times. + unread_count = {} + first_unreadable_t = {} + skip_until = {} + # Failed pool-pin tracking: retry with a 15 s backoff after 2 failures. + sem_pin_fail = {} + sem_pin_skip_until = {} + qmp_next_try = 0.0 stream = Stream(SEMSTREAM) loop = 0 @@ -293,6 +319,7 @@ def main(): while True: loop += 1 + _hb() sems = set() pbs = set() chans = set() @@ -302,17 +329,49 @@ def main(): log("stream: new uvmt sem 0x%x" % s) for p in new_pids: log("stream: client pid %d" % p) - if last_pid is not None and p != last_pid: - stream.reset() - known_sems.clear() - seen_sem_pin.clear() - last_state.clear() - last_seen_t.clear() - pinned_ranges.clear() - seen_ch.clear() - seen_pb.clear() - counter_pids.clear() - log("semfix: new client pid %d - reset per-client pin state" % p) + if last_pid is None: + seen_pids.add(p) + last_pid = p + pending_pid = None + continue + if p == last_pid: + seen_pids.add(p) + continue + if p in seen_pids or p < last_pid: + # Replayed ring/stream entry for a client we already know, or + # a backwards transition: never reset on those. + log("semfix: pid %d seen before/backwards - no reset" % p) + seen_pids.add(p) + continue + if pending_pid != p: + pending_pid = p + pending_count = 1 + log("semfix: pid change %d -> %d (debounce; reset if it " + "repeats)" % (last_pid, p)) + continue + pending_count += 1 + if pending_count < 2: + continue + # Same unseen pid reported on two consecutive polls: real new + # client. seen_pids keeps old pids from ever triggering this. + seen_pids.add(p) + pending_pid = None + pending_count = 0 + stream.reset() + known_sems.clear() + seen_sem_pin.clear() + last_state.clear() + last_seen_t.clear() + pinned_ranges.clear() + seen_ch.clear() + seen_pb.clear() + counter_pids.clear() + sem_pin_fail.clear() + sem_pin_skip_until.clear() + unread_count.clear() + first_unreadable_t.clear() + skip_until.clear() + log("semfix: new client pid %d - reset per-client pin state" % p) last_pid = p for pb in getattr(stream, "new_pbs", []): log("stream: new pushbuffer 0x%x" % pb) @@ -327,6 +386,7 @@ def main(): t_sems, t_pbs, t_chans = guest_trace() except Exception: t_sems, t_pbs, t_chans = set(), set(), set() + _hb() sems.update(t_sems) pbs.update(t_pbs) chans.update(t_chans) @@ -337,12 +397,44 @@ def main(): known_sems.add(sem) for sem in sorted(known_sems): + now = time.time() + if now < skip_until.get(sem, 0): + continue try: st = uvmt_state(q, sem) except Exception: st = None + if st is None: + # Stale-sem hygiene: the trace ring replays dead clients' + # kernel pointers forever; without eviction every loop burns + # QMP reads on them and the log floods with "unreadable". + cnt = unread_count.get(sem, 0) + 1 + unread_count[sem] = cnt + first_unreadable_t.setdefault(sem, now) + if cnt >= 3: + skip_until[sem] = now + 30.0 + if cnt == 3: + log("semfix: sem=0x%x unreadable x3 - retrying every " + "30s" % sem) + age = now - first_unreadable_t[sem] + if age >= 60.0 and sem not in stream.sems: + known_sems.discard(sem) + unread_count.pop(sem, None) + first_unreadable_t.pop(sem, None) + skip_until.pop(sem, None) + last_state.pop(sem, None) + last_seen_t.pop(sem, None) + seen_sem_pin.discard(sem) + sem_pin_fail.pop(sem, None) + sem_pin_skip_until.pop(sem, None) + log("semfix: evicting stale sem=0x%x (unreadable %.0fs, " + "absent from stream)" % (sem, age)) + continue + else: + unread_count.pop(sem, None) + first_unreadable_t.pop(sem, None) + skip_until.pop(sem, None) key = state_key(st) - now = time.time() if key != last_state.get(sem): last_state[sem] = key last_seen_t[sem] = now @@ -354,18 +446,33 @@ def main(): # Pin the pool page once per (sem, word) resolution. The payload # word is what the releasing push writes; pinning the page makes # the host able to apply that release. - if st and sem not in seen_sem_pin: + if (st and sem not in seen_sem_pin + and now >= sem_pin_skip_until.get(sem, 0)): try: pairs = resolve_sem_pool(q, sem) except Exception as e: pairs = [] log("semfix: sem=0x%x resolve failed: %s" % (sem, e)) if pairs: - seen_sem_pin.add(sem) + ok_any = False for gv, gpa in pairs: ok = pin(gv, gpa) + ok_any = ok_any or ok log("semfix: sem=0x%x pool va=0x%x -> gpa=0x%x pin=%s" % (sem, gv, gpa, "ok" if ok else "FAILED")) + if not ok_any: + # Mark pinned only after a successful pin(): a failed + # pin used to be recorded as done and never retried. + fails = sem_pin_fail.get(sem, 0) + 1 + sem_pin_fail[sem] = fails + if fails >= 2: + sem_pin_skip_until[sem] = now + 15.0 + log("semfix: sem=0x%x pool pin failed x%d - retry " + "in 15s" % (sem, fails)) + continue + seen_sem_pin.add(sem) + sem_pin_fail.pop(sem, None) + sem_pin_skip_until.pop(sem, None) if SKIP_PB: continue d = {} @@ -393,6 +500,20 @@ def main(): else: log("semfix: sem=0x%x pool unresolved (retry)" % sem) + if getattr(q, "_fails", 0) >= 5: + # QMP went away (QEMU restart, monitor died): all reads have been + # empty for 5 consecutive replies. Recreate the client loudly + # instead of reading empty forever ("uvmt(hb) ... unreadable"). + now = time.time() + if now >= qmp_next_try: + log("semfix: QMP unreadable (%d consecutive empty replies) - " + "recreating connection" % q._fails) + try: + q = Qmp(QMP_SOCK) + except Exception as e: + log("semfix: QMP recreate failed: %s" % e) + qmp_next_try = now + 5.0 + for ch in sorted(chans - seen_ch): d = {} try: @@ -439,7 +560,9 @@ def main(): # cadence: hammering the agent here starves it and wedges the run. if last_pid is None and loop % TRACE_EVERY == 0: last_pid = cdt_pid() + _hb() if last_pid: + seen_pids.add(last_pid) log("counter: client pid %d (agent)" % last_pid) if last_pid and last_pid not in counter_pids: now = time.time() @@ -448,6 +571,7 @@ def main(): pin_counter_page(last_pid, counter_pids) except Exception as e: log("counter: pin failed: %s" % e) + _hb() counter_next_try[last_pid] = now + COUNTER_RETRY_S time.sleep(interval) diff --git a/tools/guest-diag/semfix_watch.py b/tools/guest-diag/semfix_watch.py index a4b2fc0..4b60fd6 100644 --- a/tools/guest-diag/semfix_watch.py +++ b/tools/guest-diag/semfix_watch.py @@ -58,12 +58,47 @@ CHAN_MAX_CANDS = int(os.environ.get("CHAN_MAX_CANDS", "256")) class Qmp: def __init__(self, path): + self.path = path + self._cache = {} + self._fails = 0 + self._last_reconnect = 0.0 + self._reconnecting = False 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 reconnect(self): + """Re-establish the QMP connection after the socket died. + + QEMU's QMP socket accepts a single client and closes when the guest + reboots; cmd() used to swallow every exception and return {}, so all + reads stayed empty forever ("uvmt(hb) sem=... unreadable") and + pinning stopped. Rate-limited to one attempt per second so a failing + monitor does not turn every read into a connect storm.""" + if self._reconnecting or time.time() - self._last_reconnect < 1.0: + return + self._reconnecting = True + self._last_reconnect = time.time() + try: + try: + self.s.close() + except Exception: + pass + self.s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.s.connect(self.path) + self.s.settimeout(15) + self._rd() + self.cmd("qmp_capabilities") + self._cache = {} + self._ncpu = 0 + self._cpu = 0 + self._fails = 0 + except Exception: + pass + finally: + self._reconnecting = False def _rd(self): """Read one QMP reply (JSON object ends with a newline). Poll with a @@ -91,8 +126,17 @@ class Qmp: def cmd(self, name, **args): try: self.s.sendall(json.dumps({"execute": name, "arguments": args}).encode() + b"\n") - return json.loads(self._rd() or "{}") + reply = self._rd() + if not reply: + raise OSError("empty QMP reply") + self._fails = 0 + return json.loads(reply) except Exception: + # Broken socket (QEMU restart, monitor hiccup): try exactly one + # reconnect, rate-limited inside reconnect(). + self._fails = getattr(self, "_fails", 0) + 1 + if not getattr(self, "_reconnecting", False): + self.reconnect() return {} def hmp(self, line): @@ -143,8 +187,14 @@ class Qmp: break m = re.search(r":(.*)$", out, re.S) if not m: + if not out.strip(): + # Empty monitor output: the socket is broken, not the address. + self.reconnect() return [] vals = [int(x, 16) for x in re.findall(r"0x([0-9a-fA-F]+)", m.group(1))][:n] + if not vals: + self.reconnect() + return [] if cache: self._cache[key] = vals return vals @@ -397,8 +447,11 @@ def resolve_pushbuffer_wide(q, sem, 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) + try: + r = subprocess.run(["sudo", "-n", "sh", "-c", cmd], + capture_output=True, text=True, timeout=5) + except subprocess.TimeoutExpired: + continue if r.returncode == 0 and "timed out" not in (r.stderr or ""): return True time.sleep(0.05) # async queue full: drain and retry @@ -408,8 +461,11 @@ def pin(va, pa): 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) + try: + r = subprocess.run(["sudo", "-n", "sh", "-c", cmd], + capture_output=True, text=True, timeout=5) + except subprocess.TimeoutExpired: + continue if r.returncode == 0 and "timed out" not in (r.stderr or ""): return True time.sleep(0.05) |