diff options
Diffstat (limited to 'tools/guest-diag/pin_gr_stream.py')
| -rw-r--r-- | tools/guest-diag/pin_gr_stream.py | 585 |
1 files changed, 504 insertions, 81 deletions
diff --git a/tools/guest-diag/pin_gr_stream.py b/tools/guest-diag/pin_gr_stream.py index e1f5195..35c2ea3 100644 --- a/tools/guest-diag/pin_gr_stream.py +++ b/tools/guest-diag/pin_gr_stream.py @@ -32,12 +32,44 @@ 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 < 75 s; the +# supervisor grants a 120 s startup grace on top). +HEARTBEAT_FILE = os.environ.get("PIN_GR_HEARTBEAT", "/tmp/opencode/pin_gr_heartbeat") +# Inter-tool re-assert request (written by semfix_uvmt.py's stall watchdog): +# when its mtime moves, re-assert every pair this tool has ever pinned. +REASSERT_REQ = os.environ.get("PIN_REASSERT_FILE", + "/tmp/opencode/pin_reassert_request") +# Stall watchdog: a known client with no pin/data activity for this long gets +# an automatic recovery (full re-assert + demand of the fixed counter/pool +# VAs). Pre-capture (no pin yet for this client) uses the tighter limit. +STALL_PRE_S = float(os.environ.get("PIN_STALL_PRE", "60")) +STALL_POST_S = float(os.environ.get("PIN_STALL_POST", "120")) +STALL_RETRY_S = float(os.environ.get("PIN_STALL_RETRY", "30")) +STALL_ESCALATE_S = float(os.environ.get("PIN_STALL_ESCALATE", "180")) 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 _hb_loop(): + """Refresh the heartbeat on a fixed cadence, independent of data-path + activity: recv can be legally idle for minutes between client phases and + the supervisor used to read that silence as a hung tool (r198 restarts at + 34 s staleness while injection bursts were running).""" + while True: + _heartbeat() + time.sleep(5) + + 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.""" @@ -57,11 +89,73 @@ def vexec(cmd, timeout=40): return "" -def _start_cmd(): +def _scanner_alive(): + """True when the in-guest scanner process still exists. + + install_and_start() pkill's and restarts the scanner, which resets every + in-guest latch; probing first avoids killing a healthy scanner that is + merely silent because no client exists yet (r198 pre-client churn).""" + out = vexec("sudo sh -c 'kill -0 $(cat /tmp/guest_pin_stream.pid) " + "2>/dev/null && echo ALIVE'") + return "ALIVE" in (out or "") + + +def _demand_write_cmd(vas): + """Build a transport-proof append command for the guest demand file. + + r582: the old `printf '%s' '<lines with real LF>'` form was mangled by + the guest-agent transport under load - the newlines arrived as LITERAL + backslash-n text and merged VAs onto one line, and the guest scanner's + parser then crashed on `int('0x20400f000\\\\n0x121000000...', 16)` + (ValueError in main; r581 boot scanner log). The scanner's death stopped + ALL new pins while the host only re-asserted known pairs - the observed + pool-pin-latency (0x121000000 never arrived while other injections + flowed). One `echo <va> >> file` per VA carries no newlines and no + backslashes at all, so no transport layer can escape them.""" + return "; ".join("echo 0x%x >> /tmp/guest_want_vas.txt" % va + for va in vas) + + +def _demand_send(vas): + """Append VA hints to the guest scanner's demand file (best-effort). + + Same channel pin_gr_stream._demand_loop uses for host-dmesg stalls; the + stall watchdog uses it to ask the scanner to re-resolve the fixed + pre-capture VAs (counter page / UVM pool) by content/pagemap.""" + cmd = _demand_write_cmd(vas) + if not cmd: + return + try: + vexec(cmd, 30) + except Exception: + pass + + +def _push_guest_script(): + """Push the scanner into the guest in argv-safe chunks. + + SESSION 153-CONTINUED-64: the one-shot `echo <b64> | base64 -d >` form + put the whole ~135 KB base64 of the (grown) scanner on vgpu_vm.py's + argv, and the host's MAX_ARG_STRLEN (128 KB) rejected the exec before + it ever reached the guest ('OSError: Argument list too long'), so no + scanner started and the boot streamed zero pins. Push 60 KB chunks and + print the guest-side md5 for verification. + """ with open(GUEST_SCRIPT) as f: b64 = subprocess.run(["base64", "-w0"], stdin=f, - capture_output=True, text=True).stdout - prefixes = os.environ.get("CLIENT_PREFIXES", "cdt,va,victory") + capture_output=True, text=True, + timeout=10).stdout.strip() + vexec("rm -f /tmp/gps.b64", 30) + for off in range(0, len(b64), 60000): + vexec("printf %%s '%s' >> /tmp/gps.b64" % b64[off:off + 60000], 30) + return vexec("base64 -d /tmp/gps.b64 > /tmp/guest_pin_stream.py && " + "rm -f /tmp/gps.b64 && md5sum /tmp/guest_pin_stream.py " + "&& echo PUSH_OK", 60) or "" + + +def _start_cmd(): + prefixes = os.environ.get("CLIENT_PREFIXES", + "cdt,va,victory,Xorg,X,glxgears,gzdoom") vma_every = os.environ.get("STREAM_VMA_EVERY", "1") # Kill EVERY existing scanner, not just the pid-file one: a retry after # a partially-failed first attempt (pid file not written yet) used to @@ -75,14 +169,13 @@ def _start_cmd(): # failed silently (r149). return ("P=$(printf '^python3 /tmp/guest_pin_%%s.py' stream); " "sudo pkill -f \"$P\" 2>/dev/null; sleep 1; " - "echo '%s' | base64 -d > /tmp/guest_pin_stream.py; " "sudo rm -f /tmp/guest_pin_stream.pid; " "sudo setsid nohup env CLIENT_PREFIXES='%s' " "STREAM_VMA_EVERY='%s' " "python3 /tmp/guest_pin_stream.py %d %d %s %s %s " ">/tmp/guest_pin_stream.log 2>&1 & sleep 2; " "sudo cat /tmp/guest_pin_stream.pid 2>/dev/null" - % (b64, prefixes, vma_every, 2, PORT, LO, HI, PAMAX)) + % (prefixes, vma_every, 2, PORT, LO, HI, PAMAX)) def install_and_start(): @@ -90,6 +183,15 @@ 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() + try: + out_push = _push_guest_script() + if "PUSH_OK" not in (out_push or ""): + log("WARN: scanner push attempt %d incomplete: %s" + % (attempt + 1, (out_push or "")[-120:])) + except Exception as e: + log("WARN: scanner push attempt %d failed: %s" + % (attempt + 1, e)) out = vexec(_start_cmd(), 60) if any(ch.isdigit() for ch in out): return True @@ -101,8 +203,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 +221,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 @@ -135,112 +245,361 @@ def _demand_loop(): that was not yet written when first asked is picked up on a retry. This closes the stall class where the one-shot #sem kprobe event was lost and the chain resolver never learns the client. + + r908: the GR ring VA itself ('GR ring h=0x... ... (va=0x...)', the + path-B "RAM content scan found no candidate" and "not at fb" rows) is + harvested too - the ring page is a kernel allocation the userspace + pagemap cannot cover, so the guest's content resolver is the only + source, and the pump needs the ring readable to decode its entries. + Same bounded/deduped tracking (distinct VAs, 20 s resend, 300 s + expiry) as the stalled-VA class. """ track = {} # va -> [first_seen, last_sent] + # SESSION 153-CONTINUED-64: strict demand anchors (entry va -> [first, + # last_sent]) from the anchored debugfs lines. The guest validates the + # exact [entry_va, +len) read window with the host's full stream parse + # before pinning, so the setup-push pin can never land on a page whose + # bytes the host's read will reject. Resent fast (2 s) while the hold + # is up; the loose cadence (20 s) is too slow for the client's churn. + strict_track = {} + it = 0 while True: - time.sleep(3) - try: - r = subprocess.run(["sudo", "-n", "dmesg"], - capture_output=True, text=True, timeout=20) - except Exception: - continue + # SESSION 153-CONTINUED-57: the 3 s pass was the losing half of the + # r1096 race - the beef setup-hold's demanded pages resolved in the + # guest ~2-4 s after the arm, while the pump re-read the entry + # every ~110 ms and moved on. Poll fast (0.75 s) so a newly armed + # demand goes to the guest within a second; the expensive dmesg + # harvest stays on a slower cadence (every 4th pass) because the + # debugfs setup-demand set is the time-critical source. + time.sleep(0.75) + it += 1 + r = None + if it % 4 == 1: + try: + r = subprocess.run(["sudo", "-n", "dmesg"], + capture_output=True, text=True, timeout=20) + except Exception: + continue now = time.time() - for m in re.finditer(r"stalled VA #\d+: (0x[0-9a-f]+)", - (r.stdout or "")[-200000:]): - va = int(m.group(1), 16) + for m in re.finditer( + r"(?:stalled VA #\d+[:=]\s*" + r"|pending release starved va=" + r"|GR ring h=0x[0-9a-f]+ rank=\d+[^\n]*?\(va=)" + r"(0x[0-9a-f]+)" + # r914: an unreadable GR push names its pushbuffer VA - + # the GL client's push page is a kernel allocation the + # userspace pagemap cannot cover, so the guest's content + # resolver is the only source (same class as the ring VA + # demand above). Harvest it so the guest resolves and + # injects the page. + r"|sema-pushdec: h=0x[0-9a-f]+ va=(0x[0-9a-f]+)" + r"[^\n]*unreadable=1" + # SESSION 153-CONTINUED-46: a big GL push whose trusted + # chain answered an all-zero page is the beef draw that + # never captures ("sema-pushzero: ... rescued=0"). The + # promote rescue inside the read may heal it; when it + # cannot, the guest content resolver is the last source - + # harvest the VA so it resolves and injects the real page. + r"|sema-pushzero: h=0x[0-9a-f]+ va=(0x[0-9a-f]+)", + (r.stdout or "")[-2000000:] if r else ""): + va = int(m.group(1) or m.group(2) or m.group(3), 16) track.setdefault(va, [now, 0.0]) + # SESSION 153-CONTINUED-51: the pushzero line now carries the + # push length; a zero-head WINDOW push spans 1-2 pages, and the + # demand of a single page leaves the capture zero - expand the + # range and demand every page so the guest pagemap resolves the + # whole push at once (bulk demand instead of one page per cycle). + for m in re.finditer( + r"sema-pushzero: h=0x[0-9a-f]+ va=(0x[0-9a-f]+) len=(\d+)", + (r.stdout or "")[-2000000:] if r else ""): + zva = int(m.group(1), 16) + zlen = int(m.group(2)) + for zoff in range(0, min(zlen, 0x8000), 0x1000): + track.setdefault(zva + zoff, [now, 0.0]) + # SESSION 153-CONTINUED-56: the armed setup-hold's demand VA set is + # exposed by the service as STATE (debugfs), not as a dmesg row, so + # it cannot race the ring rotation under vgsp-sema spam (r1090 lost + # the setup-push pages that way). Read it every pass and feed the + # same track/resend machinery: new VAs are sent this pass (last=0), + # existing ones keep the 20 s resend / 300 s expiry cadence. + try: + rr = subprocess.run( + ["sudo", "-n", "cat", + "/sys/kernel/debug/vgsp/sema_setup_demand"], + capture_output=True, text=True, timeout=10) + if rr.returncode == 0: + for line in (rr.stdout or "").splitlines(): + ma = re.match( + r"\s*0x([0-9a-f]+)\s+0x([0-9a-f]+)\s+(\d+)\s*$", + line) + if ma: + ev = int(ma.group(1), 16) + off = int(ma.group(2), 16) + ln = int(ma.group(3)) + if ev and 0 < ln <= 0x40000: + strict_track.setdefault(ev, [now, 0.0, + off, ln]) + continue + for m in re.finditer(r"0x[0-9a-f]+", line): + sva = int(m.group(0), 16) & ~0xfff + if sva: + track.setdefault(sva, [now, 0.0]) + except Exception: + pass for va in list(track): first, last = track[va] if now - first > 300.0: del track[va] + for ev in list(strict_track): + first, last, off, ln = strict_track[ev] + if now - first > 300.0: + del strict_track[ev] + cmds = [] todo = [va for va, (f, l) in track.items() if now - l > 20.0] - if not todo: + if todo: + todo = todo[:48] + cmds.append(_demand_write_cmd(todo)) + stodo = [ev for ev, (f, l, o, ln) in strict_track.items() + if now - l > 2.0] + if stodo: + stodo = stodo[:16] + cmds.append("; ".join( + "echo S 0x%x 0x%x %d >> /tmp/guest_want_vas.txt" + % (ev, strict_track[ev][2], strict_track[ev][3]) + for ev in stodo)) + if not cmds: continue - lines = "".join("0x%x\n" % va for va in todo) try: - vexec("printf '%%s' '%s' >> /tmp/guest_want_vas.txt" % lines, 30) + vexec("; ".join(cmds), 30) except Exception: continue for va in todo: track[va][1] = now - log("demand: sent %d stalled VA(s) to guest" % len(todo)) + for ev in stodo: + strict_track[ev][1] = now + log("demand: sent %d stalled VA(s)%s to guest" % + (len(todo), + " + %d strict anchor(s)" % len(stodo) if stodo else "")) + + +_DEMAND_STARTED = False +_HB_STARTED = False +_REASSERT_REQ_SEEN = 0.0 def main(): + global _DEMAND_STARTED, _HB_STARTED + _heartbeat() + if not _HB_STARTED: + _HB_STARTED = True + try: + threading.Thread(target=_hb_loop, daemon=True).start() + except Exception as e: + log("WARN: heartbeat thread failed to start: %s" % e) 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 + # Stall-watchdog state for the connected client. client_pid is the + # scanner-reported pid (None until a client appears); last_activity marks + # the last pin/data/control event; pins_since_client counts pages pinned + # for the current client (0 = pre-capture). + client_pid = None + client_start = None + last_activity = time.time() + pins_since_client = 0 + stall_since = None + recover_next = 0.0 + escalated = False + no_conn_warn_next = 0.0 + + def _activity(): + nonlocal last_activity, stall_since, escalated + last_activity = time.time() + stall_since = None + escalated = False + 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). + # Probe the scanner first instead of reinstalling blindly: + # install_and_start() pkill's a live scanner and resets all + # in-guest latches, and pre-client silence is normal (r198 churn). + if now - last_conn > 30.0 and now - last_reinstall >= 60.0: + last_reinstall = now + if _scanner_alive(): + log("no guest connection for %.0fs but scanner alive; " + "not reinstalling" % (now - last_conn)) + if now - last_conn > 180.0 and now >= no_conn_warn_next: + no_conn_warn_next = now + 60.0 + log("STALL WATCHDOG: no guest scanner connection for " + "%.0fs (scanner process alive) - pins/demand may " + "be stranded" % (now - last_conn)) + continue + 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: + novel = _novel_pages(runs, seen) + pages = _flush(runs, seen) + pin_total += pages + pins_since_client += pages + if novel: + _activity() + runs = [] + _maybe_reassert_request(seen) + now = time.time() + # Stall watchdog: a known client with no pin/data + # activity gets a bounded, idempotent recovery (re-assert + # everything + demand the fixed counter/pool VAs), then a + # loud escalation if the stall persists. This is the + # "first cudaMalloc / first GR capture must not stall + # silently" guard. recover_next rate-limits attempts (the + # loop wakes every 5 s on the recv timeout). + if client_pid is not None: + limit = STALL_PRE_S if pins_since_client == 0 \ + else STALL_POST_S + idle = now - last_activity + if idle >= limit and now >= recover_next: + recover_next = now + STALL_RETRY_S + if stall_since is None: + stall_since = now + log("STALL WATCHDOG: client pid %d - no pin " + "activity for %.0fs (pre-capture=%s) - " + "re-asserting all known pairs + demanding " + "counter/pool VAs" + % (client_pid, idle, pins_since_client == 0)) + if seen: + _flush([], seen, full=True) + _demand_send([0x20400f000, 0x121000000]) + if not escalated and stall_since is not None \ + and now - stall_since >= STALL_ESCALATE_S: + escalated = True + log("STALL WATCHDOG: recovery FAILED for pid %d " + "after %.0fs - stall persists, needs manual " + "attention" % (client_pid, now - stall_since)) 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.match(r"#pid\s+(\d+)(?:\s+(\d+))?", text) + seen = {} + client_pid = int(m.group(1)) if m else None + client_start = (int(m.group(2)) if m and m.group(2) + else None) + pins_since_client = 0 + _activity() + log("guest reports cdt pid %s%s" + % (client_pid if client_pid is not None else "?", + (" start=%s" % client_start + if client_start is not None else ""))) + _mirror(text) + continue + if text.startswith("#nopid"): + # In-band client death: the guest scanner tells us the + # client is gone (with its identity). Drop the per-pair + # dedupe map so the next client's pairs are injected + # unconditionally; semfix_uvmt consumes the mirrored + # line and purges its per-client state too. + client_pid = None + client_start = None + seen = {} + pins_since_client = 0 + _activity() + log("guest reports client gone (%s)" % text) + _mirror(text) + continue + if text.startswith("#sem"): + _mirror(text) + _activity() + continue + if text.startswith("#pb"): + _mirror(text) + _activity() + continue + if text.startswith("#put"): + log(text) + _mirror(text) + _activity() + 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: + novel = _novel_pages(runs, seen) + pages = _flush(runs, seen) + pin_total += pages + pins_since_client += pages + if novel: + _activity() + runs = [] + # Poll the semfix re-assert request on every recv cycle, not + # only on recv-timeouts: sustained scanner traffic starves the + # timeout branch, so the request fired post-mortem in r203 + # (119 s late, `seen` already cleared by #nopid). + _maybe_reassert_request(seen) + 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() @@ -248,27 +607,82 @@ def main(): pass -def _flush(runs, seen): +def _maybe_reassert_request(seen): + """Honour a full-re-assert request written by semfix_uvmt's watchdog.""" + global _REASSERT_REQ_SEEN + try: + m = os.stat(REASSERT_REQ).st_mtime + except OSError: + return + if m > _REASSERT_REQ_SEEN: + _REASSERT_REQ_SEEN = m + if seen: + log("semfix requested a full pin re-assert - re-asserting %d " + "pair(s)" % len(seen)) + _flush([], seen, full=True) + + +def _coalesce_seen(seen): + """Coalesce the seen {va: pa} per-page map into (va, pa, npages) runs.""" + items = sorted(seen.items()) + runs = [] + i = 0 + while i < len(items): + va, pa = items[i] + n = 1 + while (i + n < len(items) + and items[i + n][0] == va + n * 0x1000 + and items[i + n][1] == pa + n * 0x1000): + n += 1 + runs.append((va, pa, n)) + i += n + return runs + + +def _novel_pages(runs, seen): + """Count pages in runs whose translation is new or changed. + + Progress accounting for the stall watchdog: the guest re-streams the CE + window and UVM pool every couple of seconds even when nothing moves, so a + pure repeat of an already-known pair must not count as pin activity + (r203: those repeats masked the watchdog through every hang).""" + return sum(1 for va, pa, n in runs for i in range(n) + if seen.get(va + i * 0x1000) != pa + i * 0x1000) + + +def _flush(runs, seen, full=False): """Inject run lines (one batched sudo pass), return number of pages pinned. The old implementation spawned one sudo process per run; with ~400 runs that cost ~4 s, which landed the GR pins after the client's context-init deadline. The batch writes all runs in a single shell, retrying only the ones the kernel queue rejected (ENOSPC) individually. + + full=True re-asserts every pair ever pinned (coalesced), not just this + batch's runs: the kernel session flush wipes the pin table when a client's + channels close, and the stall watchdog / semfix re-assert request use this + to heal it without touching the guest. Writes are idempotent. """ - todo = [] + todo = {} for va, pa, n in runs: - # The CE window (0x120000000..0x121000000) is re-asserted by the + # The CE window (0x120000000..0x121000000) and the UVM semaphore + # pool page range (0x121000000..0x122000000) are re-asserted by the # guest every couple of seconds on purpose: the host's session flush # wipes the whole pin table when the previous client's last channel # closes, and a pin batch that raced in for the new client must be # allowed to land again even when the pair is unchanged. Skipping # repeats there stranded the second client's CE pushes (the r144/ - # r145 stall: sema-probe "NO ENTRY" for the live push pages). - ce_win = 0x120000000 <= va < 0x121000000 - if not ce_win and seen.get(va) == pa and n == 1: + # r145 stall: sema-probe "NO ENTRY" for the live push pages) and the + # pool/tracker page (pending release starved, every client's first + # seconds). + ce_win = 0x120000000 <= va < 0x122000000 + if not full and not ce_win and seen.get(va) == pa and n == 1: continue - todo.append((va, pa, n)) + todo[(va, pa, n)] = True + if full: + for va, pa, n in _coalesce_seen(seen): + todo[(va, pa, n)] = True + todo = list(todo) if not todo: return 0 @@ -286,6 +700,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) @@ -319,4 +734,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) |