diff options
| -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, 144 insertions, 274 deletions
diff --git a/tools/guest-diag/guest_pin_stream.py b/tools/guest-diag/guest_pin_stream.py index f645f76..4bfaaf8 100644 --- a/tools/guest-diag/guest_pin_stream.py +++ b/tools/guest-diag/guest_pin_stream.py @@ -1627,126 +1627,15 @@ 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: - 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') + last_pid = None seen = {} seen_sems = set() seen_pbs = set() seen_chs = set() - _send(lines) time.sleep(INTERVAL) continue if pid != last_pid: @@ -1814,6 +1703,85 @@ 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: @@ -2049,8 +2017,27 @@ 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)) - if _send(lines): + 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) 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 1cd3eb1..0072b20 100644 --- a/tools/guest-diag/pin_gr_stream.py +++ b/tools/guest-diag/pin_gr_stream.py @@ -32,8 +32,7 @@ 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). +# 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") @@ -49,16 +48,6 @@ 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.""" @@ -78,17 +67,6 @@ 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, @@ -209,18 +187,11 @@ def _demand_loop(): _DEMAND_STARTED = False -_HB_STARTED = False def main(): - global _DEMAND_STARTED, _HB_STARTED + global _DEMAND_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: @@ -247,16 +218,9 @@ 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). - # 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: + # 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 - 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: @@ -279,9 +243,8 @@ def main(): try: chunk = conn.recv(1 << 20) except socket.timeout: - # Run even with no pending runs: the periodic full - # re-assert inside _flush needs a quiet-period tick too. - if _flush(runs, seen): + if runs: + _flush(runs, seen) runs = [] continue except OSError: @@ -302,14 +265,6 @@ 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 @@ -348,10 +303,6 @@ 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. @@ -360,18 +311,7 @@ def _flush(runs, seen): deadline. The batch writes all runs in a single shell, retrying only the ones the kernel queue rejected (ENOSPC) individually. """ - 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 = {} + 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 @@ -384,22 +324,9 @@ def _flush(runs, seen): # 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: + if not ce_win and seen.get(va) == pa and n == 1: continue - 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) + todo.append((va, pa, n)) if not todo: return 0 diff --git a/tools/guest-diag/pin_supervisor.sh b/tools/guest-diag/pin_supervisor.sh index f1f4eae..5bbab0c 100755 --- a/tools/guest-diag/pin_supervisor.sh +++ b/tools/guest-diag/pin_supervisor.sh @@ -6,12 +6,10 @@ # 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 < 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 +# 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 @@ -32,8 +30,6 @@ 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"; } @@ -61,7 +57,6 @@ 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" } @@ -75,7 +70,6 @@ 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" } @@ -95,8 +89,6 @@ 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 @@ -122,26 +114,16 @@ restart() { # $1 label, $2 script path, $3 start function "$3" } -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 +check_tool() { # $1 label, $2 path, $3 heartbeat file, $4 start function + local age age=$(hb_age "$3") - if tool_alive "$2" && [ "$age" -le "$STALE" ]; then + if tool_alive "$2" && [ "$age" -le 30 ]; then return 0 fi - 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 + 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" } @@ -151,8 +133,6 @@ start_pin start_semfix while true; do sleep 5 - 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" + 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 812076e..fa3ae01 100755 --- a/tools/guest-diag/semfix_uvmt.py +++ b/tools/guest-diag/semfix_uvmt.py @@ -53,8 +53,7 @@ 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 < 75 s; the -# supervisor grants a 120 s startup grace on top). +# 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 @@ -240,23 +239,6 @@ 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. @@ -300,9 +282,10 @@ 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 = {} @@ -349,6 +332,7 @@ 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) @@ -359,13 +343,20 @@ def main(): log("semfix: pid %d seen before/backwards - no reset" % p) seen_pids.add(p) continue - # 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. + 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() @@ -375,7 +366,6 @@ 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() @@ -392,7 +382,6 @@ 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: @@ -553,7 +542,6 @@ 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 @@ -571,32 +559,20 @@ 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 \ - and last_pid not in counter_dead: + if last_pid and last_pid not in counter_pids: 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() - 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 + counter_next_try[last_pid] = now + COUNTER_RETRY_S time.sleep(interval) |