diff options
Diffstat (limited to 'tools')
| -rw-r--r-- | tools/guest-diag/guest_pin_stream.py | 213 | ||||
| -rw-r--r-- | tools/guest-diag/pin_gr_stream.py | 91 | ||||
| -rwxr-xr-x | tools/guest-diag/pin_supervisor.sh | 44 | ||||
| -rwxr-xr-x | tools/guest-diag/semfix_uvmt.py | 70 |
4 files changed, 274 insertions, 144 deletions
diff --git a/tools/guest-diag/guest_pin_stream.py b/tools/guest-diag/guest_pin_stream.py index 4bfaaf8..f645f76 100644 --- a/tools/guest-diag/guest_pin_stream.py +++ b/tools/guest-diag/guest_pin_stream.py @@ -1627,15 +1627,126 @@ def main(): pool_reemit_next = 0.0 _want_state = {"mtime": None, "done": set()} loop = 0 + + def _send(lines): + """Flush control/pin lines to the host, connecting the vsock if needed. + + Returns True when the lines were sent (or there were none to send).""" + nonlocal sock + if not lines: + return True + payload = ''.join(lines).encode() + if sock is None: + try: + sock = socket.socket(socket.AF_VSOCK, socket.SOCK_STREAM) + sock.settimeout(5) + sock.connect((CID, PORT)) + sock.settimeout(None) + if last_pid is not None: + sock.sendall(("#pid %d\n" % last_pid).encode()) + except OSError: + if sock: + sock.close() + sock = None + return False + try: + sock.sendall(payload) + except OSError: + sock.close() + sock = None + return False + return True + while True: pid = cdt_pid() loop += 1 + lines = [] + # Pid-independent work first: the scrubber words are kernel objects + # that exist before any client, and the pool/demand re-asserts heal + # the host pin table between clients. Gating these on a live pid kept + # the host unregistered/unhealed whenever no client was present. + if kc_ok and time.time() >= demand_next: + demand_next = time.time() + 1.0 + try: + st_d = os.stat("/tmp/guest_want_vas.txt") + if st_d.st_mtime != _want_state.get("mtime"): + with open("/tmp/guest_want_vas.txt") as fw: + wvas = [int(t, 16) for t in fw.read().split() + if t.startswith("0x")] + _want_state["mtime"] = st_d.st_mtime + for wv in wvas[-64:]: + if wv in _want_state["done"]: + continue + got = [] + if POOL_SEM_LO <= (wv & ~0xFFF) < POOL_SEM_HI: + # m165 (D-F1a): pool/waiter VAs have no method + # stream content and the CE offset math does not + # apply; route them to the structural tracker-sem + # walk instead. m168: the tracker sems are + # long-lived objects (allocated once) whose page + # pointer follows the pool across clients, so try + # every sem we have ever resolved a chain for - + # seen_sems alone can miss the live generation. + _psems = set(seen_sems) | set(KC_PB_STATE.keys()) \ + | set(resolved_kc_pb) + try: + ok = resolve_pool_va(kc, wv, got, + _psems) + except Exception: + ok = False + got = [] + dbg('demand-pool-va: 0x%x %s' + % (wv, 'ok' if ok else 'pending')) + if ok and got: + pool_pages.setdefault(wv & ~0xFFF, True) + else: + try: + ok = resolve_va_by_content(kc, wv, got) + except Exception: + ok = False + got = [] + if ok and got: + lines.extend(got) + _want_state["done"].add(wv) + except OSError: + pass + # m165 (D-F1a): re-assert the pool page pair every KC_POOL_REEMIT_S. + # The pool page is boot-stable (UVM frees it only at GPU deinit) but + # the host pin table is wiped at every session flush, so a one-shot + # pin leaves later clients starving until the slow structural walks + # answer; the periodic re-assert heals the wipe within seconds (the + # same argument as the CE mem sweep above). + if kc_ok and pool_pages and time.time() >= pool_reemit_next: + pool_reemit_next = time.time() + KC_POOL_REEMIT_S + for pv in list(pool_pages): + got = [] + try: + ok = resolve_pool_va(kc, pv, got, seen_sems) + except Exception: + ok = False + got = [] + if ok and got: + lines.extend(got) + # PMA scrubber buffer discovery (throttled): resolve the finish/host- + # semaphore word PAs and stream them for host registration. This runs + # with or without a client (the words are kernel objects). + if kc_ok and time.time() >= scrub_next: + scrub_next = time.time() + 5.0 + try: + scrub_scan(kc, lines) + except Exception as e: + dbg('scrub scan failed: %s' % e) if pid is None: - last_pid = None + if last_pid is not None: + last_pid = None + # Notify the host so it drops its per-pair dedupe map; the + # next client's pairs must be injected unconditionally. + lines.append('#nopid\n') seen = {} seen_sems = set() seen_pbs = set() seen_chs = set() + _send(lines) time.sleep(INTERVAL) continue if pid != last_pid: @@ -1703,85 +1814,6 @@ def main(): cur[0x20400f000] = cpa new = [(va, pa) for va, pa in cur.items() if seen.get(va) != pa] pending_cur = cur - lines = [] - # Demand pins (host -> this scanner): the host appends the CE - # push-buffer VAs of GPFIFO entries it cannot read to - # /tmp/guest_want_vas.txt (written via the guest agent by - # pin_gr_stream); resolve each by content against the - # nv_vm_map_pages areas and emit the pair. This is the - # event-independent hole filler for the stall class where the - # one-shot #sem kprobe event was lost and the object-chain - # resolver never learns the client. - if kc_ok and time.time() >= demand_next: - demand_next = time.time() + 1.0 - try: - st_d = os.stat("/tmp/guest_want_vas.txt") - if st_d.st_mtime != _want_state.get("mtime"): - with open("/tmp/guest_want_vas.txt") as fw: - wvas = [int(t, 16) for t in fw.read().split() - if t.startswith("0x")] - _want_state["mtime"] = st_d.st_mtime - for wv in wvas[-64:]: - if wv in _want_state["done"]: - continue - got = [] - if POOL_SEM_LO <= (wv & ~0xFFF) < POOL_SEM_HI: - # m165 (D-F1a): pool/waiter VAs have no method - # stream content and the CE offset math does not - # apply; route them to the structural tracker-sem - # walk instead. m168: the tracker sems are - # long-lived objects (allocated once) whose page - # pointer follows the pool across clients, so try - # every sem we have ever resolved a chain for - - # seen_sems alone can miss the live generation. - _psems = set(seen_sems) | set(KC_PB_STATE.keys()) \ - | set(resolved_kc_pb) - try: - ok = resolve_pool_va(kc, wv, got, - _psems) - except Exception: - ok = False - got = [] - dbg('demand-pool-va: 0x%x %s' - % (wv, 'ok' if ok else 'pending')) - if ok and got: - pool_pages.setdefault(wv & ~0xFFF, True) - else: - try: - ok = resolve_va_by_content(kc, wv, got) - except Exception: - ok = False - got = [] - if ok and got: - lines.extend(got) - _want_state["done"].add(wv) - except OSError: - pass - # m165 (D-F1a): re-assert the pool page pair every KC_POOL_REEMIT_S. - # The pool page is boot-stable (UVM frees it only at GPU deinit) but - # the host pin table is wiped at every session flush, so a one-shot - # pin leaves later clients starving until the slow structural walks - # answer; the periodic re-assert heals the wipe within seconds (the - # same argument as the CE mem sweep above). - if kc_ok and pool_pages and time.time() >= pool_reemit_next: - pool_reemit_next = time.time() + KC_POOL_REEMIT_S - for pv in list(pool_pages): - got = [] - try: - ok = resolve_pool_va(kc, pv, got, seen_sems) - except Exception: - ok = False - got = [] - if ok and got: - lines.extend(got) - # PMA scrubber buffer discovery (throttled): resolve the finish/host- - # semaphore word PAs and stream them for host registration. - if kc_ok and time.time() >= scrub_next: - scrub_next = time.time() + 5.0 - try: - scrub_scan(kc, lines) - except Exception as e: - dbg('scrub scan failed: %s' % e) for ctl in tracer.drain(): parts = ctl.split() if ctl.startswith("#sem") and len(parts) >= 2: @@ -2017,27 +2049,8 @@ def main(): lines.append('0x%x,0x%x,%d\n' % (va0, pa0, n)) va0, pa0, n = va, pa, 1 lines.append('0x%x,0x%x,%d\n' % (va0, pa0, n)) - payload = ''.join(lines).encode() - if sock is None: - try: - sock = socket.socket(socket.AF_VSOCK, socket.SOCK_STREAM) - sock.settimeout(5) - sock.connect((CID, PORT)) - sock.settimeout(None) - if last_pid is not None: - sock.sendall(("#pid %d\n" % last_pid).encode()) - except OSError: - if sock: - sock.close() - sock = None - time.sleep(INTERVAL) - continue - try: - sock.sendall(payload) + if _send(lines): seen = pending_cur - except OSError: - sock.close() - sock = None time.sleep(INTERVAL) diff --git a/tools/guest-diag/pin_gr_stream.py b/tools/guest-diag/pin_gr_stream.py index 0072b20..1cd3eb1 100644 --- a/tools/guest-diag/pin_gr_stream.py +++ b/tools/guest-diag/pin_gr_stream.py @@ -32,7 +32,8 @@ 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). +# 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") @@ -48,6 +49,16 @@ def _heartbeat(): 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.""" @@ -67,6 +78,17 @@ def vexec(cmd, timeout=40): return "" +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 _start_cmd(): with open(GUEST_SCRIPT) as f: b64 = subprocess.run(["base64", "-w0"], stdin=f, @@ -187,11 +209,18 @@ def _demand_loop(): _DEMAND_STARTED = False +_HB_STARTED = False def main(): - global _DEMAND_STARTED + 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)") if not _DEMAND_STARTED: @@ -218,9 +247,16 @@ def main(): 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: + # 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(): + _heartbeat() + log("no guest connection for %.0fs but scanner alive; " + "not reinstalling" % (now - last_conn)) + continue log("reinstall attempt: no guest connection for %.0fs" % (now - last_conn)) try: @@ -243,8 +279,9 @@ def main(): try: chunk = conn.recv(1 << 20) except socket.timeout: - if runs: - _flush(runs, seen) + # Run even with no pending runs: the periodic full + # re-assert inside _flush needs a quiet-period tick too. + if _flush(runs, seen): runs = [] continue except OSError: @@ -265,6 +302,14 @@ def main(): % (m.group(1) if m else "?")) _mirror(text) continue + if text.startswith("#nopid"): + # Client exited: drop the per-pair dedupe map so the + # next client's pairs are injected unconditionally + # (the guest emits this on the pid -> None edge). + seen = {} + log("guest reports no client pid") + _mirror(text) + continue if text.startswith("#sem"): _mirror(text) continue @@ -303,6 +348,10 @@ def main(): pass +_FULL_REASSERT_S = 5.0 +_full_reassert_next = 0.0 + + def _flush(runs, seen): """Inject run lines (one batched sudo pass), return number of pages pinned. @@ -311,7 +360,18 @@ def _flush(runs, seen): deadline. The batch writes all runs in a single shell, retrying only the ones the kernel queue rejected (ENOSPC) individually. """ - todo = [] + global _full_reassert_next + now = time.time() + # Every _FULL_REASSERT_S re-assert every pair pinned so far, not only the + # ones in this batch. The guest streams deltas only and the kernel session + # flush wipes the pin table when a client's channels close, so without + # this a flush while the same pid continues (a second context in one + # process) strands the GR-window pages for good. Writes are idempotent; + # contiguous pages are coalesced into one run line. + full = now >= _full_reassert_next + if full: + _full_reassert_next = now + _FULL_REASSERT_S + todo = {} for va, pa, n in runs: # The CE window (0x120000000..0x121000000) and the UVM semaphore # pool page range (0x121000000..0x122000000) are re-asserted by the @@ -324,9 +384,22 @@ def _flush(runs, seen): # pool/tracker page (pending release starved, every client's first # seconds). ce_win = 0x120000000 <= va < 0x122000000 - if not ce_win and seen.get(va) == pa and n == 1: + 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: + items = sorted(seen.items()) + 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 + todo[(va, pa, n)] = True + i += n + todo = list(todo) if not todo: return 0 diff --git a/tools/guest-diag/pin_supervisor.sh b/tools/guest-diag/pin_supervisor.sh index 5bbab0c..f1f4eae 100755 --- a/tools/guest-diag/pin_supervisor.sh +++ b/tools/guest-diag/pin_supervisor.sh @@ -6,10 +6,12 @@ # 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 +# process is alive (exact script path) and that its heartbeat mtime is < 75 s. +# A startup grace (120 s since the tool started, skipped early when the tool +# log already shows first successful work) keeps a slow cold start from being +# restarted. 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 @@ -30,6 +32,8 @@ PIN_LOG="$LOGDIR/12-pin_gr_stream-$TAG.log" SEMFIX_LOG="$LOGDIR/12-semfix-$TAG.log" WINDOW=3600 CAP=10 +STALE=75 +GRACE=120 mkdir -p "$LOGDIR" /tmp/opencode log() { echo "[$(date +%H:%M:%S)] $*" | tee -a "$SUPLOG"; } @@ -57,6 +61,7 @@ start_pin() { touch "$PIN_HB" 2>/dev/null || true ( setsid nohup env SEMFIX_COUNTER_RETRY=10 \ python3 "$PIN" >> "$PIN_LOG" 2>&1 & ) + pin_started=$(date +%s) log "START pin_gr_stream" } @@ -70,6 +75,7 @@ start_semfix() { 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 & ) + semfix_started=$(date +%s) log "START semfix_uvmt" } @@ -89,6 +95,8 @@ count_list() { # $1 list -> element count restarts_pin="" restarts_semfix="" +pin_started=$(date +%s) +semfix_started=$(date +%s) restart() { # $1 label, $2 script path, $3 start function local now list n @@ -114,16 +122,26 @@ restart() { # $1 label, $2 script path, $3 start function "$3" } -check_tool() { # $1 label, $2 path, $3 heartbeat file, $4 start function - local age +check_tool() { # $1 label, $2 path, $3 heartbeat file, $4 start function, + # $5 tool start TS, $6 tool log file, $7 first-work pattern + local age runtime age=$(hb_age "$3") - if tool_alive "$2" && [ "$age" -le 30 ]; then + if tool_alive "$2" && [ "$age" -le "$STALE" ]; then return 0 fi - if ! tool_alive "$2"; then - log "$1 not running (heartbeat age ${age}s) - restarting" - else + if tool_alive "$2"; then + # Startup grace: a cold tool can spend its first ~2 min installing the + # guest scanner or connecting QMP with a lagging heartbeat; only + # enforce staleness once the grace has passed or the log shows the + # tool has already done successful work. + runtime=$(( $(date +%s) - $5 )) + if [ "$runtime" -lt "$GRACE" ] && \ + ! grep -q -- "$7" "$6" 2>/dev/null; then + return 0 + fi log "$1 heartbeat stale (${age}s) - restarting" + else + log "$1 not running (heartbeat age ${age}s) - restarting" fi restart "$1" "$2" "$4" } @@ -133,6 +151,8 @@ 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 + check_tool pin_gr_stream "$PIN" "$PIN_HB" start_pin "$pin_started" \ + "$PIN_LOG" "listening on vsock" + check_tool semfix_uvmt "$SEMFIX" "$SEMFIX_HB" start_semfix \ + "$semfix_started" "$SEMFIX_LOG" "semfix_uvmt: watching" done diff --git a/tools/guest-diag/semfix_uvmt.py b/tools/guest-diag/semfix_uvmt.py index fa3ae01..812076e 100755 --- a/tools/guest-diag/semfix_uvmt.py +++ b/tools/guest-diag/semfix_uvmt.py @@ -53,7 +53,8 @@ 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). +# Liveness file checked by pin_supervisor.sh (mtime must stay < 75 s; the +# supervisor grants a 120 s startup grace on top). 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 @@ -239,6 +240,23 @@ def cdt_pid(): return int(pids[-1]) if pids else None +def pid_alive(pid): + """Probe /proc/<pid> in the guest before a blocking exec on it. + + A killed client's pid stays latched in the watcher state; retrying its + counter pin every COUNTER_RETRY_S ran a full 25 s agent exec per attempt + and starved the scanner (the post-kill agent hammering). Assume alive on + a failed probe so a transient agent error cannot drop a live client. + """ + try: + r = subprocess.run( + [VMEXEC, "exec", "test -d /proc/%d && echo ALIVE" % pid], + capture_output=True, text=True, timeout=15) + except Exception: + return True + return "ALIVE" in (r.stdout or "") + + def pin_counter_page(pid, pinned_pids): """Pin the client's completion-counter page from its own pagemap. @@ -282,10 +300,9 @@ def main(): last_seen_t = {} dseen = set() counter_pids = set() + counter_dead = 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 = {} @@ -332,7 +349,6 @@ def main(): if last_pid is None: seen_pids.add(p) last_pid = p - pending_pid = None continue if p == last_pid: seen_pids.add(p) @@ -343,20 +359,13 @@ def main(): 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. + # Any unseen pid above the current one is a real new client. + # Stream.poll dedupes pids against its own list, so each pid + # arrives in exactly one poll batch and never repeats - the old + # two-poll debounce therefore never fired after the first client + # (the post-kill counter latch bug). seen_pids still guards + # replays of old pids; reset immediately instead. seen_pids.add(p) - pending_pid = None - pending_count = 0 stream.reset() known_sems.clear() seen_sem_pin.clear() @@ -366,6 +375,7 @@ def main(): seen_ch.clear() seen_pb.clear() counter_pids.clear() + counter_dead.clear() sem_pin_fail.clear() sem_pin_skip_until.clear() unread_count.clear() @@ -382,6 +392,7 @@ def main(): # as a fallback for channel/pushbuffer discovery and to recover sems # the scanner may have missed. if loop % TRACE_EVERY == 0: + _hb() try: t_sems, t_pbs, t_chans = guest_trace() except Exception: @@ -542,6 +553,7 @@ def main(): log("semfix: pb=0x%x pushbuf miss [%s]" % (pb, diag_str(d))) if DLOG: + _hb() dmesg_releases(dseen) # Retry incomplete push-buffer pins: gva2gpa of the kernel vmalloc @@ -559,20 +571,32 @@ def main(): # block for its whole timeout under client load, so retry on a slow # cadence: hammering the agent here starves it and wedges the run. if last_pid is None and loop % TRACE_EVERY == 0: + _hb() 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: + if last_pid and last_pid not in counter_pids \ + and last_pid not in counter_dead: now = time.time() if now >= counter_next_try.get(last_pid, 0): - try: - 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 + if not pid_alive(last_pid): + # Killed clients stay latched as last_pid forever; probe + # first so the 10 s cadence does not block the agent for a + # full 25 s exec on every retry (the post-kill hammering + # that starved the scanner). A new client's #pid resets + # counter_dead and takes over below. + log("counter: pid %d is gone - dropping" % last_pid) + counter_dead.add(last_pid) + else: + try: + 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) |