diff options
| -rw-r--r-- | tools/guest-diag/guest_pin_stream.py | 47 | ||||
| -rw-r--r-- | tools/guest-diag/pin_gr_stream.py | 214 | ||||
| -rwxr-xr-x | tools/guest-diag/pin_supervisor.sh | 57 | ||||
| -rwxr-xr-x | tools/guest-diag/semfix_uvmt.py | 487 |
4 files changed, 705 insertions, 100 deletions
diff --git a/tools/guest-diag/guest_pin_stream.py b/tools/guest-diag/guest_pin_stream.py index 4bfaaf8..bb253a2 100644 --- a/tools/guest-diag/guest_pin_stream.py +++ b/tools/guest-diag/guest_pin_stream.py @@ -1398,6 +1398,34 @@ def nvidia_fds(pid): return kind +def pid_starttime(pid): + """Starttime (clock ticks, field 22) of /proc/<pid>, or None. + + Part of the client identity (pid, starttime): pids are recycled, so a + per-client reset keyed on the pid alone can mis-fire on a recycled pid. + Parsed after the last ')' so comm names with spaces/parens are safe.""" + try: + with open('/proc/%d/stat' % pid) as f: + st = f.read() + rp = st.rfind(')') + if rp < 0: + return None + toks = st[rp + 2:].split() + if len(toks) < 20: # starttime = field 22 overall = index 19 + return None + return int(toks[19]) + except (OSError, ValueError): + return None + + +def pid_line(tag, pid, start): + """'#<tag> <pid> [starttime]' - the optional starttime keeps the line + backward compatible with older hosts (they parse the pid only).""" + if start is None: + return "#%s %d\n" % (tag, pid) + return "#%s %d %d\n" % (tag, pid, start) + + def cdt_pid(): prefixes = tuple(p for p in os.environ.get("CLIENT_PREFIXES", "cdt,va,victory").split(",") if p) @@ -1617,6 +1645,7 @@ def main(): except OSError: pass last_pid = None + last_start = None last_log = 0.0 scrub_next = 0.0 demand_next = 0.0 @@ -1631,7 +1660,19 @@ def main(): pid = cdt_pid() loop += 1 if pid is None: + if last_pid is not None and sock is not None: + # In-band client-death signal: the host tools purge the dead + # client's per-client state on this line, so the NEXT client + # works without a VM cycle (the kill cascade). Best-effort - + # if the socket is gone the host still has its own probe. + try: + sock.sendall(pid_line("nopid", last_pid, + last_start).encode()) + except OSError: + sock.close() + sock = None last_pid = None + last_start = None seen = {} seen_sems = set() seen_pbs = set() @@ -1640,6 +1681,7 @@ def main(): continue if pid != last_pid: last_pid = pid + last_start = pid_starttime(pid) seen = {} seen_sems = set() seen_pbs = set() @@ -1670,7 +1712,7 @@ def main(): dbg("switch seed: +%d event kptr(s)" % _seeded) if sock is not None: try: - sock.sendall(("#pid %d\n" % pid).encode()) + sock.sendall(pid_line("pid", pid, last_start).encode()) except OSError: sock.close() sock = None @@ -2025,7 +2067,8 @@ def main(): sock.connect((CID, PORT)) sock.settimeout(None) if last_pid is not None: - sock.sendall(("#pid %d\n" % last_pid).encode()) + sock.sendall(pid_line("pid", last_pid, + last_start).encode()) except OSError: if sock: sock.close() diff --git a/tools/guest-diag/pin_gr_stream.py b/tools/guest-diag/pin_gr_stream.py index 0072b20..2f78f2a 100644 --- a/tools/guest-diag/pin_gr_stream.py +++ b/tools/guest-diag/pin_gr_stream.py @@ -32,8 +32,20 @@ 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") +# 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): @@ -48,6 +60,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 +89,30 @@ 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 _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.""" + lines = "".join("0x%x\n" % va for va in vas) + try: + vexec("printf '%%s' '%s' >> /tmp/guest_want_vas.txt" % lines, 30) + except Exception: + pass + + def _start_cmd(): with open(GUEST_SCRIPT) as f: b64 = subprocess.run(["base64", "-w0"], stdin=f, @@ -187,11 +233,19 @@ def _demand_loop(): _DEMAND_STARTED = False +_HB_STARTED = False +_REASSERT_REQ_SEEN = 0.0 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: @@ -211,6 +265,25 @@ def main(): pin_total = 0 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() @@ -218,9 +291,20 @@ 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(): + 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: @@ -244,8 +328,43 @@ def main(): chunk = conn.recv(1 << 20) except socket.timeout: if runs: - _flush(runs, seen) + pages = _flush(runs, seen) + pin_total += pages + pins_since_client += pages + if pages: + _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 except OSError: break @@ -259,21 +378,45 @@ def main(): if not text: continue if text.startswith("#pid"): - m = re.search(r"(\d+)", text) + m = re.match(r"#pid\s+(\d+)(?:\s+(\d+))?", text) seen = {} - log("guest reports cdt pid %s" - % (m.group(1) if m else "?")) + 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]+)", @@ -290,7 +433,11 @@ def main(): int(m.group(2), 16), int(m.group(3)) if m.group(3) else 1)) if runs: - pin_total += _flush(runs, seen) + pages = _flush(runs, seen) + pin_total += pages + pins_since_client += pages + if pages: + _activity() runs = [] except Exception as e: # A transient error on one connection must not kill the process: @@ -303,15 +450,52 @@ 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 _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) and the UVM semaphore # pool page range (0x121000000..0x122000000) are re-asserted by the @@ -324,9 +508,13 @@ 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: + for va, pa, n in _coalesce_seen(seen): + todo[(va, pa, n)] = True + 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..b87d5c1 100755 --- a/tools/guest-diag/pin_supervisor.sh +++ b/tools/guest-diag/pin_supervisor.sh @@ -6,11 +6,24 @@ # 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). +# process is alive (exact script path) and that its heartbeat mtime is < 75 s. +# A startup grace (120 s, skipped early once the tool log shows it has done +# first useful 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). +# +# Liveness policy (do not tighten without re-checking the soak evidence): +# * the tools refresh their heartbeats from dedicated loops/threads even +# while idle (pin_gr_stream: 5 s thread; semfix: per loop + around every +# blocking guest-agent/dmesg call), so 75 s never trips on a healthy +# client between phases; +# * a DEAD process is restarted on the next 5 s tick regardless of grace; +# * dead-pid/state recovery (the kill cascade) is handled INSIDE the tools +# (in-band #nopid + idle liveness probe + stall watchdog), NOT by this +# supervisor - a log-based restart here could fire mid-run of a healthy +# client (the m178 lesson). # # Usage: pin_supervisor.sh LOGDIR TAG # Logs: LOGDIR/12-pin-supervisor-TAG.log (supervisor) @@ -30,6 +43,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 +72,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 +86,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 +106,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 +133,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 +162,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..f28efd3 100755 --- a/tools/guest-diag/semfix_uvmt.py +++ b/tools/guest-diag/semfix_uvmt.py @@ -53,13 +53,24 @@ 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 -# 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() +# Dead-pid/stall watchdog tuning (seconds). The tracked client is probed in +# the guest only when it has been idle this long, on a cadence, so live +# clients are never probed and a killed client is detected within ~1 min. +LIVENESS_IDLE_S = float(os.environ.get("SEMFIX_LIVENESS_IDLE", "15")) +LIVENESS_EVERY_S = float(os.environ.get("SEMFIX_LIVENESS_EVERY", "25")) +# Pre-capture stall: client alive, counter page not pinned, no progress for +# this long -> recovery attempt; escalate loudly if it persists. +STALL_PRECAP_S = float(os.environ.get("SEMFIX_STALL_PRECAP", "60")) +STALL_RETRY_S = float(os.environ.get("SEMFIX_STALL_RETRY", "30")) +STALL_ESCALATE_S = float(os.environ.get("SEMFIX_STALL_ESCALATE", "180")) +# Inter-tool re-assert request: written here, consumed by pin_gr_stream.py +# (full ping re-assert of every pair it has ever pinned). +REASSERT_REQ = os.environ.get("PIN_REASSERT_FILE", + "/tmp/opencode/pin_reassert_request") def _hb(): @@ -73,9 +84,16 @@ def _hb(): class Stream: """Tail the guest scanner's control lines mirrored by pin_gr_stream.py. - The guest emits "#pid <n>" and "#sem 0x<k>" over vsock; the listener - appends them here. This replaces per-loop guest-agent trace reads, which - were too slow to catch short-lived tracking semaphores. + The guest emits "#pid <n> [starttime]" and "#sem 0x<k>" over vsock; the + listener appends them here. This replaces per-loop guest-agent trace + reads, which were too slow to catch short-lived tracking semaphores. + + Replay safety (the m178 regression): the file persists across tool + restarts, so a fresh process re-reads the whole history. Lines written + before this process started (byte offset < init_size) are absorbed as + HISTORY - they seed the dedupe/identity state and the sem/pb lists but + can never trigger a per-client reset. Only lines appended afterwards + are live. This makes a supervisor restart mid-run harmless. """ def __init__(self, path): @@ -85,6 +103,15 @@ class Stream: self.pids = [] self.pbs = [] self.new_pbs = [] + self.new_nopids = [] + self.history_pids = [] # (pid, start) identities from history + self.pid_keys = set() # (pid, start) dedupe, per generation + try: + # Everything already in the file when this process starts is + # history (a restart replays it); never reset on it. + self._init_size = os.path.getsize(path) + except OSError: + self._init_size = 0 def reset(self): """Forget everything learned about the previous client. A new client @@ -95,6 +122,16 @@ class Stream: self.pbs = [] self.pids = [] self.new_pbs = [] + self.new_nopids = [] + self.pid_keys = set() + + @staticmethod + def _ident(line): + """Parse '#pid 1234' or '#pid 1234 5678900' -> (pid, start|None).""" + nums = re.findall(r"\d+", line) + if not nums: + return None, None + return int(nums[0]), (int(nums[1]) if len(nums) > 1 else None) def poll(self): try: @@ -105,12 +142,19 @@ class Stream: self.off = 0 f.seek(self.off) data = f.read() + base = self.off self.off = f.tell() except OSError: - return [], [] + return [], [], [] new_sems, new_pids = [], [] self.new_pbs = [] + self.new_nopids = [] + pos = base for line in data.splitlines(): + # A line is live when it ends past the startup boundary; history + # lines are absorbed (identity/dedupe) but never emitted as live. + live = pos + len(line) + 1 > self._init_size + pos += len(line) + 1 if line.startswith("#sem"): m = re.search(r"0x([0-9a-fA-F]+)", line) if m: @@ -126,13 +170,116 @@ class Stream: self.pbs.append(v) self.new_pbs.append(v) elif line.startswith("#pid"): - m = re.search(r"(\d+)", line) - if m: - v = int(m.group(1)) - if v not in self.pids: - self.pids.append(v) - new_pids.append(v) - return new_sems, new_pids + pid, start = self._ident(line) + if pid is not None and (pid, start) not in self.pid_keys: + self.pid_keys.add((pid, start)) + self.pids.append(pid) + if live: + new_pids.append((pid, start)) + else: + self.history_pids.append((pid, start)) + elif line.startswith("#nopid"): + # In-band client-death signal (new guest scanner; older ones + # simply never send it and the host probe covers the case). + pid, start = self._ident(line) + if live and pid is not None: + self.new_nopids.append((pid, start)) + return new_sems, new_pids, self.new_nopids + + +class ClientTracker: + """Monotonic, replay-proof client-identity tracking. + + A client's identity is (pid, starttime). Rules, in priority order: + * an identity adopted once is never reset a second time (duplicate + lines, trace/stream replays, supervisor restarts); + * a backwards pid, or the same pid with an older starttime, is a + replay - never reset; + * a pid recycled with a NEWER starttime is a real new client - reset + immediately (the previous holder is definitionally gone); + * a newer unseen identity only resets the per-client state once the + CURRENT client is CONFIRMED gone - a reset fired while the previous + client was still active is the m178 third-client hang regression. + + Pure logic: the caller feeds guest liveness results in and performs the + actual state purge when an action says so. + """ + + def __init__(self): + self.seen = set() + self.current = None + self.pending = None + self.dead = set() + + @staticmethod + def _start_match(a, b): + # Unknown starttimes match conservatively (legacy scanner lines). + return a[1] == b[1] or a[1] is None or b[1] is None + + def observe_history(self, pid, start): + """Absorb an identity found in pre-restart history: the process has + no per-client state yet, so history only seeds identity tracking and + the freshest identity becomes current - no reset is ever requested + for history lines.""" + key = (pid, start) + if key in self.seen: + return + self.seen.add(key) + self.current = key + + def note_dead(self, pid, start): + """A #nopid or a successful dead probe: identity is gone.""" + key = (pid, start) + self.dead.add(key) + if self.current and self.current[0] == pid \ + and self._start_match(self.current, key): + self.current = None + if self.pending and self.pending[0] == pid \ + and self._start_match(self.pending, key): + self.pending = None + + def is_current(self, pid, start): + """True when (pid, start) is the currently tracked identity (used to + decide whether a death signal may purge - a stale #nopid for a + recycled pid must never purge the new holder).""" + return self.current is not None and self.current[0] == pid \ + and self._start_match(self.current, (pid, start)) + + def decide(self, pid, start): + """Classify a live #pid line: adopt | skip | skip_backwards | + maybe_reset | reset.""" + key = (pid, start) + if key in self.seen: + return "skip" + self.seen.add(key) + cur = self.current + if cur is None: + self.current = key + return "adopt" + if pid == cur[0] and self._start_match(cur, key): + return "skip" + if pid == cur[0] and start is not None and cur[1] is not None \ + and start > cur[1]: + # Recycled pid, newer process: safe to reset now. + self.current = key + return "reset" + if pid < cur[0] or (pid == cur[0] and start is not None + and cur[1] is not None and start < cur[1]): + return "skip_backwards" + self.pending = key + return "maybe_reset" + + def resolve(self, old_dead): + """Caller learned whether the CURRENT client is dead. Returns + 'reset' (purge + adopt the pending client) or 'defer'.""" + if self.pending is None: + return None + if old_dead: + key = self.pending + self.pending = None + self.current = key + return "reset" + return "defer" def log(msg): @@ -211,7 +358,9 @@ def dmesg_releases(seen): Only lines after the last "sema-svc: thread started" marker count: dmesg keeps the previous module load's lines and the old client's releases. + Returns the number of newly printed lines (used as watchdog progress). """ + n = 0 try: r = subprocess.run( ["sudo", "-n", "sh", "-c", @@ -220,13 +369,15 @@ def dmesg_releases(seen): "| grep 'sema-push' | tail -300"], capture_output=True, text=True, timeout=10) except Exception: - return + return 0 for line in r.stdout.splitlines(): line = line.strip() if not line or line in seen: continue seen.add(line) + n += 1 log("push: %s" % line.replace("vgsp: ", "")) + return n def cdt_pid(): @@ -239,6 +390,34 @@ def cdt_pid(): return int(pids[-1]) if pids else None +def _pid_alive_cmd(pid): + """Guest shell that reports ALIVE / DEAD for a pid. Zombie-aware: a + killed (SIGKILLed) client lingers as a defunct /proc entry, and a plain + `test -d /proc/<pid>` would latch it as alive forever (the m178 + dead-pid latch).""" + return ("f=/proc/%d/stat; if [ -r \"$f\" ]; then " + "s=$(sed -e 's/^[^)]*) //' -e 's/ .*//' \"$f\"); " + "if [ \"$s\" = Z ] || [ -z \"$s\" ]; then echo DEAD; " + "else echo ALIVE; fi; else echo DEAD; fi" % pid) + + +def pid_alive(pid): + """Guest liveness of a client pid: True alive, False dead/zombie/gone, + None unknown (agent error). Callers must treat None as alive (never + purge a client on an inconclusive probe).""" + try: + r = subprocess.run([VMEXEC, "exec", _pid_alive_cmd(pid)], + capture_output=True, text=True, timeout=20) + except Exception: + return None + out = r.stdout or "" + if "ALIVE" in out: + return True + if "DEAD" in out: + return False + return None + + def pin_counter_page(pid, pinned_pids): """Pin the client's completion-counter page from its own pagemap. @@ -283,9 +462,10 @@ def main(): dseen = set() counter_pids = set() counter_next_try = {} + # Current client pid/starttime, mirrored from the tracker (starttime + # disambiguates recycled pids). last_pid = None - pending_pid = None - pending_count = 0 + last_start = None # 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 = {} @@ -296,8 +476,73 @@ def main(): sem_pin_skip_until = {} qmp_next_try = 0.0 stream = Stream(SEMSTREAM) + tracker = ClientTracker() + # Dead-pid/backwards-transition proofing and the stall watchdog. + history_absorbed = False + adopted_from_history = False + probe_next = 0.0 # next allowed idle-liveness probe + recover_next = 0.0 # next watchdog recovery attempt + stall_since = None # first observation of the pre-capture stall + escalated = False + progress_t = time.time() # last host-side progress for the client loop = 0 + def _clear_client_state(): + """Purge every per-client latch. ONLY valid when the previous client + is CONFIRMED dead (or at fresh process start): the m178 third-client + hang came from wiping the state of an ACTIVE client on a replayed + (#pid) identity.""" + 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() + counter_next_try.clear() + sem_pin_fail.clear() + sem_pin_skip_until.clear() + unread_count.clear() + first_unreadable_t.clear() + skip_until.clear() + + def _adopt(pid, start, why): + nonlocal last_pid, last_start, progress_t, adopted_from_history + nonlocal stall_since, escalated + last_pid, last_start = pid, start + progress_t = time.time() + adopted_from_history = False + stall_since = None + escalated = False + log("semfix: tracking client pid %d%s (%s)" + % (pid, " start=%s" % start if start is not None else "", why)) + + def _watchdog_reset(): + nonlocal stall_since, escalated + stall_since = None + escalated = False + + def _watchdog_recover(): + """Idempotent pre-capture stall recovery: re-arm every sem latch so + the next loop re-resolves and re-pins, force an immediate counter + retry, and request a full pin re-assert from pin_gr_stream.""" + for sem in set(known_sems) | set(stream.sems): + skip_until.pop(sem, None) + sem_pin_skip_until.pop(sem, None) + sem_pin_fail.pop(sem, None) + unread_count.pop(sem, None) + first_unreadable_t.pop(sem, None) + seen_sem_pin.discard(sem) + if last_pid is not None: + counter_next_try[last_pid] = 0.0 + try: + with open(REASSERT_REQ, "w") as f: + f.write("%.3f\n" % time.time()) + except OSError: + pass + def pin_range_once(gpu_vas, cpu, gpa, size=0): key = tuple(gpu_vas), cpu, size st = pinned_ranges.get(key) @@ -324,64 +569,94 @@ def main(): pbs = set() chans = set() - new_sems, new_pids = stream.poll() + new_sems, new_pids, new_nopids = stream.poll() for s in new_sems: log("stream: new uvmt sem 0x%x" % s) - for p in new_pids: - log("stream: client pid %d" % 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 + if new_sems and last_pid is not None: + progress_t = time.time() + + # In-band client-death signals (new guest scanner): purge the dead + # client's per-client state immediately so the NEXT client works + # without a VM cycle (the kill cascade). + for p, st in new_nopids: + log("stream: client gone (%d%s)" + % (p, " start=%s" % st if st is not None else "")) + was_current = tracker.is_current(p, st) + tracker.note_dead(p, st) + if was_current and last_pid == p: + _clear_client_state() + last_pid = last_start = None + _watchdog_reset() + log("semfix: purged per-client state of exited pid %d" % p) + # The old client is gone: adopt any client whose reset was deferred + # while it was still (assumed) alive. + if last_pid is None and tracker.pending is not None \ + and tracker.current is None: + if tracker.resolve(old_dead=True) == "reset": + p, st = tracker.current + _clear_client_state() + _adopt(p, st, "deferred reset") + # Pre-restart history (a supervisor restart re-reads the file): absorb + # identities WITHOUT any reset - the fresh process already has empty + # state; the freshest identity is the client that was running. + if stream.history_pids and not history_absorbed: + for p, st in stream.history_pids: + tracker.observe_history(p, st) + history_absorbed = True + if tracker.current is not None and last_pid is None: + p, st = tracker.current + _adopt(p, st, "resumed from pre-restart history") + adopted_from_history = True + + for p, st in new_pids: + cur_before = tracker.current + action = tracker.decide(p, st) + tag = " start=%s" % st if st is not None else "" + if action == "skip": + log("stream: client pid %d%s seen before - no reset" % (p, tag)) + elif action == "skip_backwards": + log("semfix: client pid %d%s backwards/replay - no reset" + % (p, tag)) + elif action == "adopt": + log("stream: client pid %d%s" % (p, tag)) + _adopt(p, st, "new client") + elif action == "reset": + # Recycled pid, newer starttime: the previous holder of this + # pid is definitionally gone, safe to reset now. + log("semfix: pid %d recycled (start=%s) - reset per-client " + "pin state" % (p, st)) + _clear_client_state() + _adopt(p, st, "pid recycled") + elif action == "maybe_reset": + old = cur_before + alive = pid_alive(old[0]) if old else False + if alive is False: + log("semfix: new client pid %d%s (pid %d gone) - reset " + "per-client pin state" + % (p, tag, old[0] if old else -1)) + _clear_client_state() + tracker.resolve(old_dead=True) + _adopt(p, st, "client switch") + else: + # NEVER clear the active client's state on the account of + # a newer pid: the m178 hang. Deferred until the old pid + # is confirmed gone (periodic idle probe below). + log("semfix: client pid %d%s appeared while pid %s is " + "alive%s - reset deferred" + % (p, tag, old[0] if old else "?", + "" if alive else " (probe unknown)")) for pb in getattr(stream, "new_pbs", []): log("stream: new pushbuffer 0x%x" % pb) pbs.add(pb) + if getattr(stream, "new_pbs", []) and last_pid is not None: + progress_t = time.time() sems.update(stream.sems) # The guest-agent trace read is slow under client load; use it only # 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: @@ -438,6 +713,7 @@ def main(): if key != last_state.get(sem): last_state[sem] = key last_seen_t[sem] = now + progress_t = time.time() log("uvmt %s" % state_str(sem, st)) elif now - last_seen_t.get(sem, 0) >= HEARTBEAT: last_seen_t[sem] = now @@ -473,6 +749,7 @@ def main(): seen_sem_pin.add(sem) sem_pin_fail.pop(sem, None) sem_pin_skip_until.pop(sem, None) + progress_t = time.time() if SKIP_PB: continue d = {} @@ -492,6 +769,7 @@ def main(): d["exception"] = str(e)[:40] if res: pin_range_once(*res) + progress_t = time.time() log("semfix: sem=0x%x pushbuf -> %s" % (sem, diag_str(d))) else: @@ -524,6 +802,7 @@ def main(): if res: seen_ch.add(ch) pin_range_once(*res) + progress_t = time.time() log("semfix: ch=0x%x -> %s" % (ch, diag_str(d))) else: log("semfix: ch=0x%x pushbuf miss [%s]" % (ch, diag_str(d))) @@ -537,12 +816,15 @@ def main(): if res: seen_pb.add(pb) pin_range_once(*res) + progress_t = time.time() log("semfix: pb=0x%x -> %s" % (pb, diag_str(d))) else: log("semfix: pb=0x%x pushbuf miss [%s]" % (pb, diag_str(d))) if DLOG: - dmesg_releases(dseen) + if dmesg_releases(dseen): + progress_t = time.time() + _hb() # Retry incomplete push-buffer pins: gva2gpa of the kernel vmalloc # alias fails while the sampled vCPU runs in user mode (PTI), so a @@ -553,22 +835,83 @@ def main(): if not st["done"] and st["tries"] > 0: pin_range_once(*st["args"]) + # Dead-pid detection and the pre-capture stall watchdog. The guest is + # probed ONLY when the tracked client has been quiet for a while (a + # live, pushing client is never probed), so the agent is not burdened + # during healthy runs. A confirmed-dead client is purged immediately: + # the NEXT client then starts with clean state - no VM cycle needed + # (the kill cascade). + now = time.time() + if adopted_from_history and last_pid is not None: + # The client resumed from pre-restart history may have died while + # this process was down; do not latch a dead pid into the counter + # retries. + alive = pid_alive(last_pid) + adopted_from_history = False + if alive is False: + log("semfix: resumed pid %d already gone - dropping" % last_pid) + tracker.note_dead(last_pid, last_start) + last_pid = last_start = None + _watchdog_reset() + if last_pid is not None and now >= probe_next \ + and now - progress_t >= LIVENESS_IDLE_S: + probe_next = now + LIVENESS_EVERY_S + alive = pid_alive(last_pid) + if alive is False: + log("semfix: client pid %d gone (probe) - purging per-client " + "state" % last_pid) + tracker.note_dead(last_pid, last_start) + _clear_client_state() + last_pid = last_start = None + _watchdog_reset() + if tracker.pending is not None and tracker.current is None \ + and tracker.resolve(old_dead=True) == "reset": + p, st = tracker.current + log("semfix: adopting deferred client pid %d" % p) + _adopt(p, st, "deferred reset") + elif alive is True and last_pid not in counter_pids: + # Pre-capture stall: the client is alive but has no pinned + # counter page and nothing progressed. Recover (idempotent), + # then escalate loudly if the stall persists - the goal is + # that a first cudaMalloc / first GR capture never stalls + # silently. + idle = now - progress_t + if idle >= STALL_PRECAP_S: + if stall_since is None: + stall_since = now + if now >= recover_next: + recover_next = now + STALL_RETRY_S + log("STALL WATCHDOG: client pid %d alive, no " + "pre-capture progress for %.0fs - recovery " + "attempt (re-arm sems, re-pin counter, request " + "pin re-assert)" % (last_pid, idle)) + _watchdog_recover() + if not escalated and now - stall_since >= STALL_ESCALATE_S: + escalated = True + log("STALL WATCHDOG: recovery FAILED for pid %d " + "after %.0fs (alive, counter page not pinned) - " + "pre-capture hang persists, needs manual " + "attention" % (last_pid, now - stall_since)) + # Pin the completion-counter page of a newly started client, before its # first GR pushes decode their releases. The pid comes from the guest # scanner; the guest-agent fallback is only a safety net. The exec can # 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: - last_pid = cdt_pid() + p = cdt_pid() _hb() - if last_pid: - seen_pids.add(last_pid) - log("counter: client pid %d (agent)" % last_pid) + if p: + if tracker.current is None and (p, None) not in tracker.seen: + tracker.observe_history(p, None) + last_pid, last_start = p, None + log("counter: client pid %d (agent)" % p) 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) + if pin_counter_page(last_pid, counter_pids): + progress_t = time.time() except Exception as e: log("counter: pin failed: %s" % e) _hb() |