diff options
Diffstat (limited to 'tools')
| -rwxr-xr-x | tools/cuda-ab/push_battery.sh | 22 | ||||
| -rw-r--r-- | tools/guest-diag/guest_pin_stream.py | 983 | ||||
| -rw-r--r-- | tools/guest-diag/pin_gr_stream.py | 465 | ||||
| -rwxr-xr-x | tools/guest-diag/pin_supervisor.sh | 169 | ||||
| -rwxr-xr-x | tools/guest-diag/semfix_uvmt.py | 547 | ||||
| -rw-r--r-- | tools/guest-diag/semfix_watch.py | 106 |
6 files changed, 2134 insertions, 158 deletions
diff --git a/tools/cuda-ab/push_battery.sh b/tools/cuda-ab/push_battery.sh new file mode 100755 index 0000000..644beba --- /dev/null +++ b/tools/cuda-ab/push_battery.sh @@ -0,0 +1,22 @@ +#!/bin/bash +# push_battery.sh - push every CUDA battery binary from tools/cuda-ab/run/ +# to the running guest via scp over the SSH hostfwd (port 2222). +# Usage: push_battery.sh [guest_dir] (default /tmp) +set -u +HERE="$(cd "$(dirname "$0")" && pwd)" +RUN="$HERE/run" +GUEST_DIR="${1:-/tmp}" +CLIENTS="victory_mail smiley vecadd_loop streams2 memcpy_roundtrip multikernel victory_saxpy victory_reduce victory_matmul victory_strided victory_memset2d victory_atomics victory_streams_events victory_math vector_add" +export SSHPASS="${VGPU_GUEST_PASS:-ubuntu}" +FILES="" +for c in $CLIENTS; do [ -x "$RUN/$c" ] && FILES="$FILES $RUN/$c"; done +[ -n "$FILES" ] || { echo "no binaries in $RUN"; exit 1; } +# shellcheck disable=SC2086 +sshpass -e scp -P "${VGPU_GUEST_PORT:-2222}" \ + -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null \ + -o PubkeyAuthentication=no -o PreferredAuthentications=password \ + -o ConnectTimeout=15 $FILES "ubuntu@127.0.0.1:$GUEST_DIR/" && \ +sshpass -e ssh -p "${VGPU_GUEST_PORT:-2222}" \ + -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null \ + -o PubkeyAuthentication=no -o PreferredAuthentications=password \ + "ubuntu@127.0.0.1" "chmod +x $GUEST_DIR/* 2>/dev/null; ls $GUEST_DIR | wc -l; echo BATTERY_PUSHED" diff --git a/tools/guest-diag/guest_pin_stream.py b/tools/guest-diag/guest_pin_stream.py index 35cbfb9..bb253a2 100644 --- a/tools/guest-diag/guest_pin_stream.py +++ b/tools/guest-diag/guest_pin_stream.py @@ -266,11 +266,114 @@ def _vmalloc_pages_for(kc, va): return 0 -def resolve_mem_pins(kc, mem, out, default_pages=16, va_lo=0, va_hi=0): +def _nv_vmap_areas(kc): + """nv_vm_map_pages vmalloc areas: [(lo, hi, pages)] (5 s cache). + + Every CE push-buffer rm_mem is a vmap created by nv_vm_map_pages; the + demand pinner walks these areas to find the page holding a stalled + push VA by content, without needing any kprobe event or object chain. + """ + now = time.time() + if getattr(_nv_vmap_areas, "ts", 0) > now - 5.0: + return _nv_vmap_areas.areas + areas = [] + try: + for line in open("/proc/vmallocinfo"): + if "nv_vm_map_pages" not in line: + continue + p = line.split() + if len(p) < 2: + continue + try: + lo, hi = p[0].split("-") + lo = int(lo, 16) + hi = int(hi, 16) + except ValueError: + continue + pages = (hi - lo) >> 12 + if pages > 0: + areas.append((lo, hi, pages)) + except OSError: + pass + _nv_vmap_areas.areas = areas + _nv_vmap_areas.ts = now + return areas + + +# Learned page-offset per nv_vm_map_pages area for the demand content match +# (2026-10-16b): the first successful resolution records which page index the +# VA maps to, and later pushes in the same allocation retry that offset first +# instead of walking the whole +/-16 neighbourhood. +_area_off = {} + + +def resolve_va_by_content(kc, va, out): + """Demand pin: find the page carrying GPU VA 'va' by content match. + + The host reports the CE push-buffer VAs whose GPFIFO entries it cannot + read (dmesg 'stalled VA #N: 0x...'). The guest walks every + nv_vm_map_pages vmap area and tests the candidate leaf page at the VA's + in-window offset (plus a neighbourhood, for the per-mem header-page + offset ambiguity) against the host's method-stream prefix check. On a + match it emits the exact pair, so the host can translate and consume the + entry - no kprobe event, no object-chain discovery, no resolution + lottery. Only the requested VA is pinned, so a wrong candidate cannot + poison other VAs. + + 2026-10-16b: the offset search window was +/-1 page. Pushes of a client + whose pushbuffer vmap carries a larger per-allocation header (observed + with multi-MB CE staging: pushes at 0x120000f58+ never resolved and the + demand loop re-sent the same VA for minutes while the client crawled) + fall outside it. Widen to +/-16 pages and LEARN the winning offset per + vmap area, because consecutive pushes advance sequentially inside the + same allocation and share the offset. + """ + if not (PB_CE_LO <= va < PB_CE_HI): + return False + off = (va - PB_CE_LO) & 0xFFFFFF + k = off >> 12 + areas = _nv_vmap_areas(kc) + for (lo, _hi, pages) in areas: + learned = _area_off.get(lo) + if learned is not None: + adjs = [learned] + else: + adjs = [0, 1, -1] + adjs += [d for m in (2, 4, 8, 16) for d in (m, -m)] + for adj in adjs: + idx = k + adj + if idx < 0 or idx >= pages: + continue + cpu = lo + idx * 0x1000 + d = kc.read(cpu, 64) + if not d or not any(d[:32]): + continue + if not _push_prefix_ok(d): + continue + pa = kc.va_to_pa(cpu) + if pa is None: + continue + _area_off[lo] = adj + out.append("0x%x,0x%x,1\n" % (va & ~0xFFF, pa)) + dbg("demand-va: 0x%x -> pa=0x%x (area off=%d adj=%+d)" + % (va & ~0xFFF, pa, idx, adj)) + return True + dbg("demand-va: 0x%x not found by content (%d areas)" + % (va & ~0xFFF, len(areas))) + return False + + +def resolve_mem_pins(kc, mem, out, default_pages=16, va_lo=0, va_hi=0, + page_off=0): """uvm_rm_mem kptr -> GPU-VA/GPA runs (direct map or page-table walk). va_lo/va_hi optionally restrict the emitted GPU VA range; the kcore - chain resolver targets the CE push-buffer window only. + chain resolver targets the CE push-buffer window only. page_off shifts + the vmap page paired with GPU data page k: the CE push-buffer vmap is + one page longer than its GPU data window on some allocations (a leading + header page), so GPU data page k lives at vmap page k + page_off. + page_off is derived per allocation (mem_page_off): a fixed offset is + wrong (r116/r117 regressed client #1 with a hard +1). """ if not is_kern_ptr(mem): return 0 @@ -281,7 +384,7 @@ def resolve_mem_pins(kc, mem, out, default_pages=16, va_lo=0, va_hi=0): and (not va_hi or v < va_hi)] if not cpu or not gpus: return 0 - size = kc.read64(mem + 0x60) or 0 + size = _mem_size(kc, mem) dm = _direct_map_pa(kc, cpu) emitted = 0 if dm is not None: @@ -290,14 +393,15 @@ def resolve_mem_pins(kc, mem, out, default_pages=16, va_lo=0, va_hi=0): out.append("0x%x,0x%x,%d\n" % (gv, dm, npages)) emitted += 1 return emitted + area = _vmalloc_pages_for(kc, cpu) npages = (max(1, min(size >> 12, 4098)) if size - else _vmalloc_pages_for(kc, cpu)) - if not npages: - npages = 1 + else area or default_pages) + if area and page_off + npages > area: + npages = max(1, area - page_off) runs = [] cur = None for k in range(npages): - pa = kc.va_to_pa(cpu + k * 0x1000) + pa = kc.va_to_pa(cpu + (k + page_off) * 0x1000) if pa is None: if cur: runs.append(cur) @@ -318,6 +422,196 @@ def resolve_mem_pins(kc, mem, out, default_pages=16, va_lo=0, va_hi=0): return emitted +def _mem_size(kc, mem): + """uvm_rm_mem size (bytes) from the 570.144 layout, with a fallback. + + The struct is type(4)+pad(4)+mask(8)+vas[33] (264 B)+proxy_vas+ + gpu_owner+size, so size lives at +0x128; +0x60 (inside vas[]) reads 0 + for unused processors and is kept only as a fallback. + """ + for off in (0x128, 0x60): + v = kc.read64(mem + off) or 0 + if 0 < v <= (256 << 20): + return v + return 0 + + +def _push_prefix_ok(buf): + """First-4KB method-stream plausibility, mirroring the host's check.""" + off = 0 + while off + 4 <= len(buf): + hdr = struct.unpack_from('<I', buf, off)[0] + if hdr: + break + off += 4 + if off + 4 > len(buf): + return True # all zeros: not refutable + op = (hdr >> 29) & 0x7 + if op in (4, 7): + return True + if op not in (1, 3, 5): + return False + return ((hdr >> 16) & 0xfff) != 0 + + +def _page_content(kc, va): + pa = kc.va_to_pa(va & ~0xFFF) + if pa is None: + return None + return kc.read(kc.page_offset_base + pa, 64) + + +def mem_page_off(kc, mem): + """Per-allocation vmap offset for a uvm_rm_mem's GPU data pages. + + Derived, never fixed: the vmap area's page count (vmallocinfo) minus + the rm_mem's data page count (size at +0x128). A 4097-page area with + 4096 data pages means one leading header page (offset 1); an area that + exactly matches the data means offset 0. When the size is unreadable, + the two candidate offsets are content-probed with the host's method- + stream plausibility test and the decisively-passing one wins; an + ambiguous result keeps 0 (the status-quo pairing). + """ + vas = [kc.read64(mem + 0x10 + i * 8) for i in range(16)] + cpu = next((v for v in vas if is_kern_ptr(v)), None) + if not cpu or _direct_map_pa(kc, cpu) is not None: + return 0 + area = _vmalloc_pages_for(kc, cpu) + if area <= 1: + return 0 + size = _mem_size(kc, mem) + if size: + data = min(size >> 12, 4098) + if 0 <= area - data <= 2: + return area - data + return 0 + g = [] + for off in (0, 1): + if off >= area: + g.append(0) + continue + buf = _page_content(kc, cpu + off * 0x1000) + g.append(1 if (buf is not None and _push_prefix_ok(buf)) else 0) + if g[0] != g[1]: + return 0 if g[0] else 1 + return 0 + + +def _ce_mem_pages(kc, mem): + """CE mem walk parameters: (gpus, cpu, off, npages) or None. + + Walks the same page set resolve_mem_pins used: the whole vmalloc area + (minus the header offset) when known, else size at +0x128, else the + default. The area governs: the live pushes are written sequentially and + may run past the size field, so a size-derived prefix stopped one page + short of the write frontier (r144: last page 0x120030000 emitted, the + live page 0x120031000 never in any pin set). all CE-window GPU VAs share + the CPU mapping. + """ + vas = [kc.read64(mem + 0x10 + i * 8) for i in range(16)] + cpu = next((v for v in vas if is_kern_ptr(v)), None) + gpus = [v for v in vas if v and PB_CE_LO <= v < PB_CE_HI] + if not cpu or not gpus: + return None + if _direct_map_pa(kc, cpu) is not None: + return None # never a vmap: not the push-buffer shape + off = mem_page_off(kc, mem) + area = _vmalloc_pages_for(kc, cpu) + size = _mem_size(kc, mem) + size_pages = max(1, min(size >> 12, 4098)) if size else 0 + area_pages = area - off if area and area > off else 0 + npages = min(max(size_pages, area_pages) or 64, 4098) + return (gpus, cpu, off, npages) + + +def _emit_ce_mem_delta(kc, st, out, fresh_map=False): + """Emit the CE mem's live {va,pa} pairs as merged runs. + + Two emission modes share one merge step: + + * full re-assert (KC_FULLREM_S cadence, and every fresh_map pass): every + page that translates is emitted, content-independent. This is the + fix for the host's session flush: at a client switch the host wipes + its whole sema map when the previous client's last channel closes + ("session flush (all channels closed)"), and a pin batch that raced + in for the new client is lost. The old content-gated delta never + re-emitted those unchanged pairs, so the new client's CE pushes were + permanently unreadable ("NO ENTRY" in sema-probe) and every GPFIFO + entry was skipped after the 8 s grace (r144/r145 second-client stall, + both mail and matmul). Re-asserting the full pair set every couple of + seconds heals any number of flush wipes; merged contiguous runs keep + the traffic to a few lines per pass. Zero-content pages are pinned + deliberately: the pin carries the translation, the host reads the + live content, and the client writes it later. + + * delta (between full passes): content-gated, emits pages whose PA is + new or changed, so a remap or a freshly-faulted page is picked up + without waiting for the next full pass. Never drops a previously + emitted pair (the host's last-store-wins pin sees the truth, and a + failed burst never clears the whole map). + """ + gpus, cpu, off, npages = st["gpus"], st["cpu"], st["page_off"], st["npages"] + last = st["last"] + nowm = time.monotonic() + full = nowm >= st.get("full_next", 0.0) + if full: + age = nowm - st.get("resolved_at", nowm) + st["full_next"] = nowm + (KC_FULLREM_S if age < KC_FULLREM_FAST_S + else KC_FULLREM_SLOW_S) + if fresh_map: + for k in range(npages): + kc._pa_cache.pop(cpu + (k + off) * 0x1000, None) + changed = [] + if full: + # Re-walk pages that did not translate on an earlier pass: the + # vmap's leaf sizes are huge (2 MB/1 GB), so most pages translate + # immediately; a straggler (the vmap populated lazily) is retried + # with its negative PA-cache entry popped, bounded per pass so the + # sweep stays cheap. + miss = st.setdefault("miss", set()) + retry = 0 + for k in range(npages): + page = cpu + (k + off) * 0x1000 + pa = kc.va_to_pa(page) + if pa is None: + if k in miss and retry < 512: + kc._pa_cache.pop(page, None) + retry += 1 + else: + miss.add(k) + continue + miss.discard(k) + for gv in gpus: + changed.append((gv + k * 0x1000, pa)) + else: + for k in range(npages): + d = kc.read(cpu + (k + off) * 0x1000, 32) + if not d or not any(d): + continue # zero/unreadable: nothing to pin yet + pa = kc.va_to_pa(cpu + (k + off) * 0x1000) + if pa is None: + continue + for gv in gpus: + if last.get(gv + k * 0x1000) != pa: + changed.append((gv + k * 0x1000, pa)) + changed.sort() + merged = [] + for va, pa in changed: + if merged and va == merged[-1][0] + merged[-1][2] * 0x1000 and \ + pa == merged[-1][1] + merged[-1][2] * 0x1000: + merged[-1][2] += 1 + else: + merged.append([va, pa, 1]) + for va, pa, n in merged: + out.append("0x%x,0x%x,%d\n" % (va, pa, n)) + for va, pa in changed: + last[va] = pa + st["emit_total"] = st.get("emit_total", 0) + len(merged) + if full: + st["full_total"] = st.get("full_total", 0) + 1 + return len(merged) + + def _emit_ring_pins(kc, gpu_va, entries_cpu, entries, out): """Pin the GPFIFO ring pages (and the pages after it, where the GPPUT buffer usually sits) from the ring's CPU mapping.""" @@ -440,11 +734,55 @@ def resolve_sem_pins(kc, sem, out): return resolve_mem_pins(kc, mem, out, default_pages=16) +def resolve_pool_va(kc, va, out, sems): + """Pool/waiter VA -> PA via tracker-sem -> page -> mem; NO content match. + + The word at POOL base + 4*idx is a slot in a UVM semaphore-pool page: a + sysmem uvm_rm_mem whose PA is only reachable through its kernel CPU + alias (the client pagemap has no present entry for it). Same kcore walk + the CE pushbuffer uses; the chain is shorter (sem+0 = page, page+0 = + mem). The page sits 16 MB above the CE push buffer base, outside + resolve_sem_pins' default_pages window, which is why the host saw + 'pending release starved' for it on every client's first seconds. + """ + page = va & ~0xFFF + if not (POOL_SEM_LO <= page < POOL_SEM_HI): + return False + for sem in list(sems): + if not is_kern_ptr(sem): + continue + try: + pg = kc.read64(sem) + if not is_kern_ptr(pg): + continue + mem = kc.read64(pg) + if not is_kern_ptr(mem): + continue + got = [] + resolve_mem_pins(kc, mem, got, default_pages=1, + va_lo=page, va_hi=page + 0x1000) + except Exception: + continue + for line in got: + try: + lva = int(line.split(',')[0], 16) + except Exception: + continue + if lva == page: + out.append(line) + dbg('pool-va: 0x%x -> %s (mem=0x%x)' + % (page, line.strip(), mem)) + return True + dbg('pool-va: 0x%x unresolved (%d known sems)' % (page, len(sems))) + return False + + def resolve_pb_pins(kc, pb, out): mem = kc.read64(pb + 8) if not mem: return 0 - return resolve_mem_pins(kc, mem, out, default_pages=64) + return resolve_mem_pins(kc, mem, out, default_pages=64, + page_off=mem_page_off(kc, mem)) # The CE push-buffer window is the one UVM region the client never maps into @@ -458,12 +796,49 @@ def resolve_pb_pins(kc, pb, out): # channel->pool->manager->pushbuffer->rm_mem chain is walkable in kernel # memory, so resolve the CE push buffer directly with kcore: no trace event, # nothing to lose. +# +# The kptr-keyed one-shot latches below are NOT object identity: UVM frees +# and re-creates these rm_mems and channels during a session, and the slab +# reuses the same addresses. A resolution taken before such a move pins +# pages that are stale by the time the host reads them (observed r125: the +# pinned CE pages read all-zero while the channel's real pushes lived at +# new vmap pages of the same rm_mem kptr; every ring entry then stalled in +# the host's 8 s grace since no release could be decoded from zeros). So a +# resolved mem is periodically re-validated against the channel's own ring +# entries and re-resolved when it stops matching (KC_PB_STATE/_kc_pb_fresh), +# and when several mem candidates are reachable the one whose pages actually +# hold the ring-entry pushes is chosen (content anchor, _mem_content_ok). PB_CE_LO = 0x120000000 PB_CE_HI = 0x121000000 +# m165 (D-F1a): UVM semaphore-pool page range. The pool is a 4 KB sysmem +# uvm_rm_mem placed by the UVM arena right after the 16 MB CE push buffer; +# the client's tracker words (0x121000000 + 4*idx) live in its first page. +# The client pagemap has no present entry for it, so only the kcore +# tracker-sem -> page -> mem walk can translate it. +POOL_SEM_LO = 0x121000000 +POOL_SEM_HI = 0x122000000 +KC_POOL_REEMIT_S = 2.0 # re-assert the pool page pair this often too KC_CHAIN_BACK = 0x2000 KC_CHAIN_SPAN = 0xA000 KC_CHAIN_MAX = 400 RESOLVED_KC_PB = set() +KC_PB_STATE = {} # sem kptr -> resolved-mem state (fingerprint + pins) +KC_REFRESH_S = 2.0 # re-validate a resolved CE mem at most this often +KC_REEMIT_S = 2.0 # sweep the CE mem's live pages this often +KC_REMAP_S = 30.0 # full re-translation sweep (catches remaps) +# Full re-assert cadence: every pass at least this far apart re-emits the +# whole translatable pair set regardless of the delta, healing host-side +# session-flush wipes (see _emit_ce_mem_delta). Must stay well under the +# host's gsp_sema_ce_grace_ms (8 s) so a wiped-and-needed page is restored +# inside the hold window. The fast cadence applies to a mem's first +# KC_FULLREM_FAST_S seconds (the flush race lives at the client switch); +# after that the re-assert backs off to KC_FULLREM_SLOW_S - a fully +# fragmented pair set (descending PAs) emits one page per line, and 4 k +# lines every 2 s saturated a guest under multi-client load (r148: load 19, +# three scanners from an install race, clients wedged). +KC_FULLREM_S = float(os.environ.get("STREAM_FULLREM_S", "1.5")) +KC_FULLREM_FAST_S = float(os.environ.get("STREAM_FULLREM_FAST_S", "45")) +KC_FULLREM_SLOW_S = float(os.environ.get("STREAM_FULLREM_SLOW_S", "10.0")) def _chain_mems_at(kc, w): @@ -484,13 +859,96 @@ def _chain_mems_at(kc, w): return mems +def _mem_gpu_cpu(kc, mem): + """The CE-window GPU VA and CPU alias of an rm_mem (None/None when absent).""" + vas = [kc.read64(mem + 0x10 + i * 8) for i in range(16)] + cpu = next((v for v in vas if is_kern_ptr(v)), None) + gpu = next((v for v in vas + if v and PB_CE_LO <= v < PB_CE_HI), None) + return gpu, cpu + + +def _mem_content_score(kc, mem, gpu): + """Rough 'is this the live CE push buffer' score for a mem candidate. + + The channel's CE rings live in (emulated) video memory, so the guest + cannot read the ring entries to anchor a push VA; but the CE push + buffer's guest-RAM pages are CPU-written method streams during channel + setup and client work. Count nonzero u32 pairs across the first 128 + pages of the candidate's window: a live mem scores high, a freed page + range that a reused kptr still describes reads zeros and scores 0. + """ + cpu = None + for i in range(16): + v = kc.read64(mem + 0x10 + i * 8) + if is_kern_ptr(v): + cpu = v + break + if cpu is None: + return -1 + score = 0 + for k in range(128): + buf = kc.read(cpu + k * 0x1000, 0x1000) + if not buf: + continue + for off in range(0, 0x1000, 16): + if buf[off:off + 8] != b"\x00" * 8: + score += 1 + return score + + +def _kc_pb_fresh(kc, sem): + """The pinned CE mem still matches the channel's live backing. + + Cheap check: the mem's own {cpu,gpu} fields and the guest PA of its + first window page must still match the recorded resolution. A vmap area + never moves while the mem object lives, so either check failing means + the object was freed and (possibly) re-created at the same kptr - the + caller then re-resolves and re-emits the current pages. + """ + st = KC_PB_STATE.get(sem) + if not st: + return False + mem = st.get("mem") + if not mem or not is_kern_ptr(mem): + return False + if (kc.read64(mem + 0x30) or 0) != st.get("cpu") or \ + (kc.read64(mem + 0x38) or 0) != st.get("gpu"): + return False + if not st.get("page_va"): + return True + if not st.get("page_pa"): + # Unknown PA at store time: the manual page-table walk could not + # translate the vmap page (it may have been read before the kernel + # page tables were walkable). Treat the record as fresh instead of + # failing every check - a permanently-failing freshness test made + # the scanner drop the latch and re-resolve in a tight loop, and + # each failed re-resolution left the host with no pins at all + # (observed r131: "kc-chain stale: re-resolving" every second for + # five sems while smiley sat in cudaMalloc). + return True + pa = kc.va_to_pa(st["page_va"]) + if pa is None or pa != st.get("page_pa"): + return False + return True + + def resolve_pb_chain(kc, sem, out): """Tracking-sem kptr -> CE push-buffer rm_mem pins (kcore chain walk). Returns True when a CE push-buffer mem was found for this sem, so the caller marks it resolved even when the mem was already pinned; new pin lines are appended to out. Bounded: one window read plus a few reads per - candidate, and each rm_mem is resolved at most once per session. + candidate, and each rm_mem is resolved at most once per session (until a + freshness failure drops it again, see _kc_pb_fresh). + + When several mem candidates are reachable, the content score decides: a + stale candidate (freed memory read through a reused kptr) reads zeros + and is skipped in favour of the live push-buffer pages, so it cannot + shadow them on the host's last-store-wins pin table. An all-zero + candidate set emits nothing: pinning zeros is what made the host read + zero pushes and stall every CE entry; the drift retries (retry_at) and + the periodic refresh pick the pages up as soon as content appears. """ if not is_kern_ptr(sem): return False @@ -499,26 +957,92 @@ def resolve_pb_chain(kc, sem, out): return False words = struct.unpack_from("<%dQ" % (len(data) // 8), data, 0) cands = list(dict.fromkeys(w for w in words if is_kern_ptr(w))) - found = False + # Channel-anchored expansion (r134 live dump): the tracking sem is a + # member of a uvm_channel_t table; the channel records sit in the sem's + # own window and carry their CE ring GPU VA ([0x121010000, 0x121400000)) + # plus back-to-back 32-bit object handles. The pool/manager/pushbuffer + # pointers hang off those records' targets, which the bare sem-window + # scan never reaches - the second-client stall is exactly this: the only + # live uvmt event is the spinning sem, and no mem candidate is found + # around it. For each ring-VA word, scan the surrounding record for + # kptrs, take those targets, and try the chain shape on both. Additive + # and bounded; content scoring still decides, so a wrong candidate + # cannot shadow the live pages. + extra = [] + for i, w in enumerate(words): + if not w or not (0x121010000 <= w < 0x121400000): + continue + lo = max(0, i - 0x60) + hi = min(len(words), i + 0x60) + for c in words[lo:hi]: + if not is_kern_ptr(c): + continue + extra.append(c) + tgt = kc.read(c, 0x100) + if tgt: + tw = struct.unpack_from("<%dQ" % (len(tgt) // 8), + tgt, 0) + for w2 in tw: + if is_kern_ptr(w2): + extra.append(w2) + if len(extra) >= 512: + break + if extra: + head = [] + for w in extra: + if w not in head and w not in cands: + head.append(w) + cands = head + cands + mems = [] for w in cands[:KC_CHAIN_MAX]: for mem in _chain_mems_at(kc, w): - vas = [kc.read64(mem + 0x10 + i * 8) for i in range(16)] - if not any(v and PB_CE_LO <= v < PB_CE_HI for v in vas): - continue - found = True - if mem in RESOLVED_KC_PB: - continue - RESOLVED_KC_PB.add(mem) - n = resolve_mem_pins(kc, mem, out, default_pages=64, - va_lo=PB_CE_LO, va_hi=PB_CE_HI) - try: - with open('/tmp/guest_pin_stream.log', 'a') as f: - f.write('%s kc-chain CE push-buffer mem=0x%x sem=0x%x ' - 'pin-runs=%d\n' - % (time.strftime('%T'), mem, sem, n)) - except OSError: - pass - return found + if mem not in mems: + mems.append(mem) + found = False + scored = [] + for mem in mems: + gpu, cpu = _mem_gpu_cpu(kc, mem) + if not gpu or not cpu: + continue + found = True + score = _mem_content_score(kc, mem, gpu) + scored.append((score, mem, gpu, cpu)) + if not scored: + return found + best = max(scored, key=lambda c: c[0]) + if best[0] <= 0: + dbg("kc-chain all candidates zero (sem=0x%x, n=%d): deferring" + % (sem, len(scored))) + return found + if len(scored) > 1: + dbg("kc-chain scores sem=0x%x: %s" % + (sem, ' '.join('%#x=%d' % (m, s) for s, m, _g, _c in scored))) + _score, mem, gpu, cpu = best + if mem in RESOLVED_KC_PB: + return True + RESOLVED_KC_PB.add(mem) + params = _ce_mem_pages(kc, mem) + if not params: + return found + gpus, cpu, off, npages = params + st = {"mem": mem, "gpu": gpu, "cpu": cpu, "gpus": gpus, + "page_va": cpu, "page_pa": kc.va_to_pa(cpu) or 0, + "page_off": off, "npages": npages, "last": {}, + "resolved_at": time.monotonic()} + # Fresh translation for the first sweep: the Kcore PA cache may hold + # pairs from before this mem's vmap area was (re)populated. + kc._pa_cache.clear() + n = _emit_ce_mem_delta(kc, st, out) + KC_PB_STATE[sem] = st + try: + with open('/tmp/guest_pin_stream.log', 'a') as f: + f.write('%s kc-chain CE push-buffer mem=0x%x sem=0x%x ' + 'pin-runs=%d off=%d npages=%d size=%d score=%d\n' + % (time.strftime('%T'), mem, sem, n, off, npages, + _mem_size(kc, mem), _score)) + except OSError: + pass + return True # UvmGpuChannelInfo field offsets (nv_uvm_types.h, x86_64): @@ -613,15 +1137,72 @@ def resolve_channel_pins(kc, ch, out): -def clear_trace(): - """Drop stale ring events so the fallback tail reader cannot re-learn a - previous client's freed sems after a client switch. The new client's - events are delivered fresh through the consuming trace_pipe reader.""" +def _read_trace_snapshot(max_bytes=1 << 20, budget=0.5): + """Best-effort non-blocking read of the trace ring file. + + Spawning `tail` here left uninterruptible D-state processes when the + seq_file read blocked under load (observed r127: dozens of stuck tails, + load 26); a direct os.read with O_NONBLOCK returns whatever is ready and + never accumulates processes. + """ try: - with open(TRACE, "w") as f: - f.write("") + fd = os.open(TRACE, os.O_RDONLY | os.O_NONBLOCK) except OSError: - pass + return b"" + chunks = [] + got = 0 + t0 = time.time() + try: + while got < (8 << 20) and time.time() - t0 < budget: + try: + d = os.read(fd, 262144) + except BlockingIOError: + break + except OSError: + break + if not d: + break + chunks.append(d) + got += len(d) + finally: + os.close(fd) + return b"".join(chunks)[-max_bytes:] + + +def seed_events_from_trace(seen_sems, seen_pbs, seen_chs): + """One-shot tail scan of the trace ring after a pid switch / start. + + The uvmt/uvbp/uvch kprobe events are one-shot per object, and the + trace_pipe reader only sees events that arrive after it attaches: a + client whose channels are registered in the instant around the scanner's + pid switch would otherwise never appear in the sets (observed r126: + vecadd_loop ran for minutes with sem=0 while its CE entries paid the + 8 s grace). The ring keeps recent events, so scan its tail once and add + every distinct kptr. Stale previous-client pointers that are also + re-learned are harmless: their chains no longer resolve to content + (the mem candidates are scored and deferred). + """ + try: + out = _read_trace_snapshot() + except Exception: + return 0 + n = 0 + for m in re.finditer(rb"sem=0x([0-9a-fA-F]+)", out): + v = int(m.group(1), 16) + if v not in seen_sems: + seen_sems.add(v) + n += 1 + for m in re.finditer(rb"pb=0x([0-9a-fA-F]+)", out): + v = int(m.group(1), 16) + if v not in seen_pbs: + seen_pbs.add(v) + n += 1 + for m in re.finditer(rb"ch=0x([0-9a-fA-F]+)", out): + v = int(m.group(1), 16) + if v not in seen_chs: + seen_chs.add(v) + n += 1 + return n class TraceReader: @@ -724,6 +1305,75 @@ def new_sems(seen): return found +# --------------------------------------------------------------- scrubber +# PMA scrubber channel discovery (third-client scrub freeze). The guest's +# PMA scrubber is an internal CeUtils CE channel (4096 GPFIFO entries, +# pbGpuVA 0x120000000, ring at pbGpuVA+0x64000, finish payload at +# pbGpuVA+0x6c004). It lives in a kernel vmap area, not in any user pagemap, +# and its GPU VAs collide numerically with the client CE push-buffer window +# in the host's global translation map - so the host cannot resolve it and +# must be told the finish/host-semaphore word PAs directly. Resolve them +# here (vmallocinfo + manual kernel page-table walk) and emit one +# #scrub <fin_pa>,<hsem_pa> +# line; pin_gr_stream.py registers them through sema_scrubbuf_add. +SCRUB_AREA_SIZES = (0x6e000, 0x6d000, 0x6c000, 0x6b000, 0x6f000, + 0x70000, 0x71000, 0x72000) +_scrub_sent = {} + + +def _scrub_ring_ok(kc, ring_pa): + """A 906F GPFIFO ring page: all-zero (fresh) or entries whose GET names a + push inside the channel's own 0x120000000 window (encoded as + push_va - 0x100000000, 4-byte aligned).""" + buf = kc.read(kc.page_offset_base + ring_pa, 0x1000) + if not buf: + return False + for off in range(0, 0x100, 8): + e0, _e1 = struct.unpack_from('<II', buf, off) + if e0 == 0: + continue + if not (0x20000000 <= e0 < 0x20080000) or (e0 & 3): + return False + return True + + +def scrub_scan(kc, lines): + """Emit the scrubber buffer words' PAs once (re-emitted if they move).""" + try: + f = open('/proc/vmallocinfo') + except OSError: + return + with f: + for line in f: + if 'nvidia' not in line or 'ioremap' in line or 'vmap' not in line: + continue + parts = line.split() + try: + a, b = parts[0].split('-') + lo, hi = int(a, 16), int(b, 16) + except (ValueError, IndexError): + continue + if (hi - lo) not in SCRUB_AREA_SIZES: + continue + fin_pa = kc.va_to_pa(lo + 0x6c000) + ring_pa = kc.va_to_pa(lo + 0x64000) + if fin_pa is None or ring_pa is None: + continue + if not _scrub_ring_ok(kc, ring_pa): + dbg('scrub candidate 0x%x rejected (ring shape)' % lo) + continue + fin = fin_pa + 4 + hsem = fin_pa + key = (fin, hsem) + if _scrub_sent.get('key') == key: + return + _scrub_sent['key'] = key + lines.append('#scrub 0x%x,0x%x\n' % (fin, hsem)) + dbg('scrub buffer emitted lo=0x%x fin=0x%x hsem=0x%x' + % (lo, fin, hsem)) + return + + def nvidia_fds(pid): """CUDA device fds this pid holds: 'uvm', 'gpu', or None. @@ -748,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) @@ -954,6 +1632,9 @@ def main(): resolved_chs = set() resolved_kc_pb = set() retry_at = {} + kc_refresh_at = {} + kc_reemit_at = {} + kc_remap_at = {} tracer = TraceReader() kc = Kcore() kc_ok = kc.load() @@ -964,13 +1645,34 @@ def main(): except OSError: pass last_pid = None + last_start = None last_log = 0.0 + scrub_next = 0.0 + demand_next = 0.0 + # m165 (D-F1a): pool page VAs learned from demand, re-asserted every + # KC_POOL_REEMIT_S. Boot-stable (UVM frees pool pages only at GPU + # deinit), so the set deliberately persists across client switches. + pool_pages = {} + pool_reemit_next = 0.0 + _want_state = {"mtime": None, "done": set()} loop = 0 while True: 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() @@ -979,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() @@ -989,17 +1692,27 @@ def main(): resolved_chs = set() resolved_kc_pb = set() retry_at = {} + kc_refresh_at = {} + kc_reemit_at = {} + kc_remap_at = {} # Client switch: the old client's channel/put pointers are freed # memory. Clearing the #put map stops stale reads from being # emitted as if they were the new client's counters. CHAN_PUTS.clear() PUT_LAST.clear() - clear_trace() tracer.reset() RESOLVED_KC_PB.clear() + KC_PB_STATE.clear() + # One-shot: recover the new client's one-shot channel events + # from the trace ring (they can fall in the switch instant and + # be invisible to the pipe reader; the old clear_trace() made + # that permanent - part of the r126 stall class). + _seeded = seed_events_from_trace(seen_sems, seen_pbs, seen_chs) + if _seeded: + 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 @@ -1033,6 +1746,84 @@ def main(): 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: @@ -1058,6 +1849,103 @@ def main(): added += len(got) else: retry_at[sem] = now_r + 1.0 + # CE push-buffer pin freshness: a resolved mem is re-probed + # against the channel's live ring content. UVM frees and + # re-creates these rm_mems/channels mid-session and the slab + # reuses the same kptrs, so a one-shot resolution can pin + # pages that are stale (observed reading all zeros) while + # the live pushes sit at new pages; dropping the latches + # makes the resolution below re-emit the current truth. + if sem in resolved_kc_pb and \ + now_r >= kc_refresh_at.get(sem, 0): + kc_refresh_at[sem] = now_r + KC_REFRESH_S + try: + _fresh = _kc_pb_fresh(kc, sem) + except Exception: + _fresh = True + if not _fresh: + # Re-resolve, but never lose the latch on a failed + # attempt: a dropped latch means the host keeps no + # pins for this channel at all, which is worse than + # a possibly-stale pin (the host content-checks the + # first page and now rejects all-FF continuation + # chunks, so stale pages are bounded). On failure + # keep the old state and defer the next attempt. + _st = KC_PB_STATE.get(sem) + if _st and _st.get("mem"): + RESOLVED_KC_PB.discard(_st["mem"]) + got = [] + try: + _found2 = resolve_pb_chain(kc, sem, got) + except Exception: + _found2 = False + got = [] + if _found2 and got: + _new = KC_PB_STATE.get(sem) + _newmem = _new.get("mem") if _new else None + dbg("kc-chain stale: re-resolved sem=0x%x " + "new-mem=%s runs=%d" + % (sem, "yes" if _newmem != + (_st.get("mem") if _st else None) + else "no", len(got))) + lines.extend(got) + added += len(got) + resolved_kc_pb.add(sem) + else: + if _st and _st.get("mem"): + RESOLVED_KC_PB.add(_st["mem"]) + if sem not in KC_PB_STATE: + KC_PB_STATE[sem] = _st + kc_refresh_at[sem] = now_r + 5.0 + dbg("kc-chain stale: keep old latch sem=0x%x" + % sem) + else: + # CE mem delta sweep (second-client pins): the one-shot + # emission pinned the mem once; pages the client writes + # later (or whose vmap pairing moved) were never + # re-emitted, so the host kept zeros for the other CE + # ranks' pushes. Sweep every KC_REEMIT_S, emitting only + # new/changed nonzero pairs; burst the Kcore PA cache + # every KC_REMAP_S so a remap is not shadowed. + _st = KC_PB_STATE.get(sem) + if _st is not None and \ + now_r >= kc_reemit_at.get(sem, 0): + _fresh_map = now_r >= kc_remap_at.get(sem, 0) + if _fresh_map: + kc_remap_at[sem] = now_r + KC_REMAP_S + kc_reemit_at[sem] = now_r + KC_REEMIT_S + got = [] + try: + _n2 = _emit_ce_mem_delta(kc, _st, got, + fresh_map=_fresh_map) + except Exception: + _n2 = 0 + got = [] + if got: + lines.extend(got) + added += len(got) + dbg("ce-delta sem=0x%x runs=%d mem=0x%x" + % (sem, _n2, _st.get("mem", 0))) + # CE push-buffer chain (second-client stall fix): resolve + # pool -> manager -> pushbuffer -> rm_mem directly, so the + # push-buffer pages never depend on the racy one-shot uvbp + # event. Independent of the ring/pool gates below (the + # channel gate's continue must not starve this path). + if sem not in resolved_kc_pb and \ + now_r >= retry_at.get(("pk", sem), 0): + got = [] + try: + _found = resolve_pb_chain(kc, sem, got) + except Exception: + _found = False + got = [] + if _found: + resolved_kc_pb.add(sem) + if got: + lines.extend(got) + added += len(got) + else: + retry_at[("pk", sem)] = now_r + 2.0 # ISSUE-0002: the pool pin (mem) and the channel ring are # resolved independently. The sem is marked resolved as soon # as the pool page pins, but the channel_info anchor may only @@ -1077,26 +1965,6 @@ def main(): added += len(got) else: retry_at[("cy", sem)] = now_r + 2.0 - # CE push-buffer chain (second-client stall fix): resolve - # pool -> manager -> pushbuffer -> rm_mem directly, so the - # push-buffer pages never depend on the racy one-shot uvbp - # event. Independent of the ring/pool gates above. - if sem in resolved_kc_pb or \ - now_r < retry_at.get(("pk", sem), 0): - continue - got = [] - try: - _found = resolve_pb_chain(kc, sem, got) - except Exception: - _found = False - got = [] - if _found: - resolved_kc_pb.add(sem) - if got: - lines.extend(got) - added += len(got) - else: - retry_at[("pk", sem)] = now_r + 2.0 for pb in list(seen_pbs): if pb in resolved_pbs or now_r < retry_at.get(pb, 0): continue @@ -1199,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 b864ab5..6ba51ee 100644 --- a/tools/guest-diag/pin_gr_stream.py +++ b/tools/guest-diag/pin_gr_stream.py @@ -16,6 +16,7 @@ import re import socket import subprocess import sys +import threading import time HERE = os.path.dirname(os.path.abspath(__file__)) @@ -31,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.""" @@ -56,18 +89,50 @@ 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, - capture_output=True, text=True).stdout + capture_output=True, text=True, + timeout=10).stdout prefixes = os.environ.get("CLIENT_PREFIXES", "cdt,va,victory") vma_every = os.environ.get("STREAM_VMA_EVERY", "1") - return ("echo '%s' | base64 -d > /tmp/guest_pin_stream.py; " - "if [ -f /tmp/guest_pin_stream.pid ]; then " - "p=$(cat /tmp/guest_pin_stream.pid 2>/dev/null); " - "[ -n \"$p\" ] && grep -q guest_pin_stream " - "/proc/$p/cmdline 2>/dev/null && sudo kill $p 2>/dev/null; " - "fi; " + # 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 + # leave the first scanner alive, and multiple scanners each re-assert + # the whole CE pair set - r148 ran three and drove the guest to load 19 + # (clients wedged in userspace spins). The pattern is anchored + # (^python3) and assembled at runtime via printf: an unanchored pattern + # or a literal one would also match this command's own shell (its + # cmdline contains the script path), and pkill would kill the shell + # before it could start the new scanner - every start attempt then + # 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' " @@ -82,6 +147,7 @@ def install_and_start(): # or a stale qga client); retry a few times instead of giving up, because # a boot without the scanner streams no pins and the client then hangs. for attempt in range(5): + _heartbeat() out = vexec(_start_cmd(), 60) if any(ch.isdigit() for ch in out): return True @@ -93,78 +159,297 @@ 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) + return False + + +def scrub_register(fin_pa, hsem_pa): + """Register the PMA scrubber buffer words (guest-resolved PAs) with the + service so it can author the scrub progress (third-client scrub freeze).""" + cmd = ("printf '0x%x,0x%x' > %s/sema_scrubbuf_add" + % (fin_pa, hsem_pa, PARAM)) + for _ in range(60): + 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 time.sleep(0.05) + log("scrub register FAILED fin=0x%x hsem=0x%x" % (fin_pa, hsem_pa)) return False +def _demand_loop(): + """Push the host's unreadable CE push VAs to the guest scanner. + + The service logs each distinct stalled VA once per boot ('stalled VA + #N: 0x...'). Each sighting is tracked and re-sent every 20 s for five + minutes: the guest resolves the VA by content against the + nv_vm_map_pages areas (event-independent, no object chain), so a page + 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. + """ + track = {} # va -> [first_seen, last_sent] + while True: + time.sleep(3) + 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+|pending release starved va)=(0x[0-9a-f]+)", + (r.stdout or "")[-200000:]): + va = int(m.group(1), 16) + track.setdefault(va, [now, 0.0]) + for va in list(track): + first, last = track[va] + if now - first > 300.0: + del track[va] + todo = [va for va, (f, l) in track.items() if now - l > 20.0] + if not todo: + continue + lines = "".join("0x%x\n" % va for va in todo) + try: + vexec("printf '%%s' '%s' >> /tmp/guest_want_vas.txt" % lines, 30) + except Exception: + continue + for va in todo: + track[va][1] = now + log("demand: sent %d stalled VA(s) to guest" % len(todo)) + + +_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)") + 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: + 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 + 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: + runs.append((int(m.group(1), 16), + int(m.group(2), 16), + int(m.group(3)) if m.group(3) else 1)) if runs: - _flush(runs, seen) + novel = _novel_pages(runs, seen) + pages = _flush(runs, seen) + pin_total += pages + pins_since_client += pages + if novel: + _activity() 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) - 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 = [] + # 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() @@ -172,19 +457,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: - if seen.get(va) == pa and n == 1: + # 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) 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 @@ -202,6 +550,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) @@ -235,4 +584,12 @@ def _flush(runs, seen): if __name__ == "__main__": - main() + # main() loops forever; only an unexpected exception exits it. Restart + # instead of dying silently - a dead listener means no pins and every + # later CUDA client hangs at first cudaMalloc. + while True: + try: + main() + except Exception as e: + log("FATAL: main() exited: %r - restarting in 2s" % (e,)) + time.sleep(2) diff --git a/tools/guest-diag/pin_supervisor.sh b/tools/guest-diag/pin_supervisor.sh new file mode 100755 index 0000000..b87d5c1 --- /dev/null +++ b/tools/guest-diag/pin_supervisor.sh @@ -0,0 +1,169 @@ +#!/bin/bash +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2026 Open-IOV contributors +# pin_supervisor.sh - supervise the host-side pin tools (pin_gr_stream.py and +# semfix_uvmt.py) so a silent death cannot strand every later CUDA client at +# first cudaMalloc. +# +# Both tools write a heartbeat file; this script checks, every 5 s, that each +# process is alive (exact script path) and that its heartbeat mtime is < 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) +# LOGDIR/12-pin_gr_stream-TAG.log, LOGDIR/12-semfix-TAG.log (tools) +set -uo pipefail + +LOGDIR="${1:?usage: pin_supervisor.sh LOGDIR TAG}" +TAG="${2:?usage: pin_supervisor.sh LOGDIR TAG}" + +HERE=$(cd "$(dirname "$0")" && pwd) +PIN="$HERE/pin_gr_stream.py" +SEMFIX="$HERE/semfix_uvmt.py" +PIN_HB="${PIN_GR_HEARTBEAT:-/tmp/opencode/pin_gr_heartbeat}" +SEMFIX_HB="${SEMFIX_HEARTBEAT_FILE:-/tmp/opencode/semfix_heartbeat}" +SUPLOG="$LOGDIR/12-pin-supervisor-$TAG.log" +PIN_LOG="$LOGDIR/12-pin_gr_stream-$TAG.log" +SEMFIX_LOG="$LOGDIR/12-semfix-$TAG.log" +WINDOW=3600 +CAP=10 +STALE=75 +GRACE=120 +mkdir -p "$LOGDIR" /tmp/opencode + +log() { echo "[$(date +%H:%M:%S)] $*" | tee -a "$SUPLOG"; } + +tool_alive() { pgrep -f "python3 $1" >/dev/null 2>&1; } + +hb_age() { # age in seconds of the heartbeat file, 9999 if missing + local f="$1" m + m=$(stat -c %Y "$f" 2>/dev/null) || { echo 9999; return; } + echo $(( $(date +%s) - m )) +} + +kill_tool() { + pkill -f "python3 $1" 2>/dev/null + sleep 1 + if pgrep -f "python3 $1" >/dev/null 2>&1; then + pkill -9 -f "python3 $1" 2>/dev/null + sleep 1 + fi + return 0 +} + +start_pin() { + if tool_alive "$PIN"; then return 0; fi + touch "$PIN_HB" 2>/dev/null || true + ( setsid nohup env SEMFIX_COUNTER_RETRY=10 \ + python3 "$PIN" >> "$PIN_LOG" 2>&1 & ) + pin_started=$(date +%s) + log "START pin_gr_stream" +} + +start_semfix() { + # pgrep exact-count check before start: if an instance already exists + # (ours or a leftover), never start a second one. + if tool_alive "$SEMFIX"; then + log "semfix already running - not starting another" + return 0 + fi + touch "$SEMFIX_HB" 2>/dev/null || true + ( setsid nohup env SEMFIX_SKIP_PB=0 SEMFIX_COUNTER_RETRY=10 \ + python3 "$SEMFIX" 0.5 >> "$SEMFIX_LOG" 2>&1 & ) + semfix_started=$(date +%s) + log "START semfix_uvmt" +} + +prune() { # $1 now, $2 list -> list with entries older than WINDOW dropped + local now="$1" t out="" + for t in $2; do + [ $(( now - t )) -lt "$WINDOW" ] && out="$out $t" + done + printf '%s' "$out" +} + +count_list() { # $1 list -> element count + local n=0 t + for t in $1; do n=$(( n + 1 )); done + echo "$n" +} + +restarts_pin="" +restarts_semfix="" +pin_started=$(date +%s) +semfix_started=$(date +%s) + +restart() { # $1 label, $2 script path, $3 start function + local now list n + now=$(date +%s) + if [ "$1" = pin_gr_stream ]; then + restarts_pin=$(prune "$now" "$restarts_pin") + n=$(count_list "$restarts_pin") + else + restarts_semfix=$(prune "$now" "$restarts_semfix") + n=$(count_list "$restarts_semfix") + fi + if [ "$n" -ge "$CAP" ]; then + log "GIVEUP $1: $n restarts within ${WINDOW}s" + exit 1 + fi + kill_tool "$2" + if [ "$1" = pin_gr_stream ]; then + restarts_pin="$restarts_pin $now" + else + restarts_semfix="$restarts_semfix $now" + fi + log "RESTART $(( n + 1 )) $1 $(date +%Y-%m-%dT%H:%M:%S)" + "$3" +} + +check_tool() { # $1 label, $2 path, $3 heartbeat file, $4 start function, + # $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 "$STALE" ]; 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 + log "$1 not running (heartbeat age ${age}s) - restarting" + fi + restart "$1" "$2" "$4" +} + +log "pin_supervisor up (LOGDIR=$LOGDIR TAG=$TAG)" +start_pin +start_semfix +while true; do + sleep 5 + check_tool pin_gr_stream "$PIN" "$PIN_HB" start_pin "$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 cf54445..11c191c 100755 --- a/tools/guest-diag/semfix_uvmt.py +++ b/tools/guest-diag/semfix_uvmt.py @@ -53,14 +53,47 @@ 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). +SEMFIX_HB_FILE = os.environ.get("SEMFIX_HEARTBEAT_FILE", + "/tmp/opencode/semfix_heartbeat") +# 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(): + try: + with open(SEMFIX_HB_FILE, "w") as f: + f.write("%.3f\n" % time.time()) + except OSError: + pass 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): @@ -70,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 @@ -80,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: @@ -90,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: @@ -111,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): @@ -196,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", @@ -205,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(): @@ -224,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. @@ -268,10 +462,87 @@ 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 + 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 = {} + first_unreadable_t = {} + skip_until = {} + # Failed pool-pin tracking: retry with a 15 s backoff after 2 failures. + sem_pin_fail = {} + sem_pin_skip_until = {} + qmp_next_try = 0.0 stream = Stream(SEMSTREAM) + 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) @@ -293,40 +564,104 @@ def main(): while True: loop += 1 + _hb() sems = set() 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 not None and p != last_pid: - stream.reset() - known_sems.clear() - seen_sem_pin.clear() - last_state.clear() - last_seen_t.clear() - pinned_ranges.clear() - seen_ch.clear() - seen_pb.clear() - counter_pids.clear() - log("semfix: new client pid %d - reset per-client pin state" % p) - 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: t_sems, t_pbs, t_chans = set(), set(), set() + _hb() sems.update(t_sems) pbs.update(t_pbs) chans.update(t_chans) @@ -336,16 +671,54 @@ def main(): for sem in sorted(sems - known_sems): known_sems.add(sem) - for sem in sorted(known_sems): + for hb_i, sem in enumerate(sorted(known_sems)): + if hb_i % 16 == 0: + # A full pass can run 60-90 s (thousands of QMP reads); keep + # the heartbeat fresh inside it (r203: 79 s gap -> supervisor + # restarted the tool mid-pass). + _hb() + now = time.time() + if now < skip_until.get(sem, 0): + continue try: st = uvmt_state(q, sem) except Exception: st = None + if st is None: + # Stale-sem hygiene: the trace ring replays dead clients' + # kernel pointers forever; without eviction every loop burns + # QMP reads on them and the log floods with "unreadable". + cnt = unread_count.get(sem, 0) + 1 + unread_count[sem] = cnt + first_unreadable_t.setdefault(sem, now) + if cnt >= 3: + skip_until[sem] = now + 30.0 + if cnt == 3: + log("semfix: sem=0x%x unreadable x3 - retrying every " + "30s" % sem) + age = now - first_unreadable_t[sem] + if age >= 60.0 and sem not in stream.sems: + known_sems.discard(sem) + unread_count.pop(sem, None) + first_unreadable_t.pop(sem, None) + skip_until.pop(sem, None) + last_state.pop(sem, None) + last_seen_t.pop(sem, None) + seen_sem_pin.discard(sem) + sem_pin_fail.pop(sem, None) + sem_pin_skip_until.pop(sem, None) + log("semfix: evicting stale sem=0x%x (unreadable %.0fs, " + "absent from stream)" % (sem, age)) + continue + else: + unread_count.pop(sem, None) + first_unreadable_t.pop(sem, None) + skip_until.pop(sem, None) key = state_key(st) - now = time.time() if key != last_state.get(sem): last_state[sem] = key last_seen_t[sem] = now + 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 @@ -354,18 +727,34 @@ def main(): # Pin the pool page once per (sem, word) resolution. The payload # word is what the releasing push writes; pinning the page makes # the host able to apply that release. - if st and sem not in seen_sem_pin: + if (st and sem not in seen_sem_pin + and now >= sem_pin_skip_until.get(sem, 0)): try: pairs = resolve_sem_pool(q, sem) except Exception as e: pairs = [] log("semfix: sem=0x%x resolve failed: %s" % (sem, e)) if pairs: - seen_sem_pin.add(sem) + ok_any = False for gv, gpa in pairs: ok = pin(gv, gpa) + ok_any = ok_any or ok log("semfix: sem=0x%x pool va=0x%x -> gpa=0x%x pin=%s" % (sem, gv, gpa, "ok" if ok else "FAILED")) + if not ok_any: + # Mark pinned only after a successful pin(): a failed + # pin used to be recorded as done and never retried. + fails = sem_pin_fail.get(sem, 0) + 1 + sem_pin_fail[sem] = fails + if fails >= 2: + sem_pin_skip_until[sem] = now + 15.0 + log("semfix: sem=0x%x pool pin failed x%d - retry " + "in 15s" % (sem, fails)) + continue + seen_sem_pin.add(sem) + sem_pin_fail.pop(sem, None) + sem_pin_skip_until.pop(sem, None) + progress_t = time.time() if SKIP_PB: continue d = {} @@ -385,6 +774,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: @@ -393,7 +783,22 @@ def main(): else: log("semfix: sem=0x%x pool unresolved (retry)" % sem) + if getattr(q, "_fails", 0) >= 5: + # QMP went away (QEMU restart, monitor died): all reads have been + # empty for 5 consecutive replies. Recreate the client loudly + # instead of reading empty forever ("uvmt(hb) ... unreadable"). + now = time.time() + if now >= qmp_next_try: + log("semfix: QMP unreadable (%d consecutive empty replies) - " + "recreating connection" % q._fails) + try: + q = Qmp(QMP_SOCK) + except Exception as e: + log("semfix: QMP recreate failed: %s" % e) + qmp_next_try = now + 5.0 + for ch in sorted(chans - seen_ch): + _hb() # a resolve can block for tens of seconds d = {} try: res = resolve_from_channel(q, ch, d) @@ -403,10 +808,12 @@ 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))) for pb in sorted(pbs - seen_pb): + _hb() # a resolve can block for tens of seconds d = {} try: res = resolve_from_pushbuffer(q, pb, d) @@ -416,12 +823,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 @@ -430,24 +840,89 @@ def main(): if loop % 8 == 0: for st in list(pinned_ranges.values()): if not st["done"] and st["tries"] > 0: + _hb() # pin_range_once can walk thousands of QMP reads 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() - if last_pid: - log("counter: client pid %d (agent)" % last_pid) + p = cdt_pid() + _hb() + 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() counter_next_try[last_pid] = now + COUNTER_RETRY_S time.sleep(interval) diff --git a/tools/guest-diag/semfix_watch.py b/tools/guest-diag/semfix_watch.py index 5124805..4b60fd6 100644 --- a/tools/guest-diag/semfix_watch.py +++ b/tools/guest-diag/semfix_watch.py @@ -58,12 +58,47 @@ CHAN_MAX_CANDS = int(os.environ.get("CHAN_MAX_CANDS", "256")) class Qmp: def __init__(self, path): + self.path = path + self._cache = {} + self._fails = 0 + self._last_reconnect = 0.0 + self._reconnecting = False self.s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) self.s.connect(path) self.s.settimeout(15) self._rd() self.cmd("qmp_capabilities") - self._cache = {} + + def reconnect(self): + """Re-establish the QMP connection after the socket died. + + QEMU's QMP socket accepts a single client and closes when the guest + reboots; cmd() used to swallow every exception and return {}, so all + reads stayed empty forever ("uvmt(hb) sem=... unreadable") and + pinning stopped. Rate-limited to one attempt per second so a failing + monitor does not turn every read into a connect storm.""" + if self._reconnecting or time.time() - self._last_reconnect < 1.0: + return + self._reconnecting = True + self._last_reconnect = time.time() + try: + try: + self.s.close() + except Exception: + pass + self.s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.s.connect(self.path) + self.s.settimeout(15) + self._rd() + self.cmd("qmp_capabilities") + self._cache = {} + self._ncpu = 0 + self._cpu = 0 + self._fails = 0 + except Exception: + pass + finally: + self._reconnecting = False def _rd(self): """Read one QMP reply (JSON object ends with a newline). Poll with a @@ -91,8 +126,17 @@ class Qmp: def cmd(self, name, **args): try: self.s.sendall(json.dumps({"execute": name, "arguments": args}).encode() + b"\n") - return json.loads(self._rd() or "{}") + reply = self._rd() + if not reply: + raise OSError("empty QMP reply") + self._fails = 0 + return json.loads(reply) except Exception: + # Broken socket (QEMU restart, monitor hiccup): try exactly one + # reconnect, rate-limited inside reconnect(). + self._fails = getattr(self, "_fails", 0) + 1 + if not getattr(self, "_reconnecting", False): + self.reconnect() return {} def hmp(self, line): @@ -143,8 +187,14 @@ class Qmp: break m = re.search(r":(.*)$", out, re.S) if not m: + if not out.strip(): + # Empty monitor output: the socket is broken, not the address. + self.reconnect() return [] vals = [int(x, 16) for x in re.findall(r"0x([0-9a-fA-F]+)", m.group(1))][:n] + if not vals: + self.reconnect() + return [] if cache: self._cache[key] = vals return vals @@ -189,18 +239,46 @@ def guest_trace(): Read *before* clearing: the create-time probes fire once per client and would be wiped by a clear that races them. + + 2026-10-16 robustness fix (the cascade root): this used to run a bare + `tail` over the guest agent. A tracefs read that blocks under a busy + guest left *uninterruptible* tail processes behind (observed dozens in + D state), the guest-agent exec then blocked for its full 90 s timeout + every iteration, and the pool/tracker pins this tool supplies stopped + flowing - after ~5 clients every new client froze at DEVICE waiting on + its UVM tracker with `pending release starved ... pool/waiter page`. + Now the read is bounded remotely (`timeout -s KILL 12`) so no tail can + outlive its call, and SSH (which stays healthy when the guest agent + wedges) is preferred with the agent as fallback. """ - inner = ("tail -c 131072 /sys/kernel/tracing/trace " + inner = ("timeout -s KILL 12 tail -c 131072 /sys/kernel/tracing/trace " "| grep -E \"uvmt|uvch|uvbp|uvpbf\" " "| tail -400") if not os.environ.get("KEEP_TRACE"): inner += "; echo > /sys/kernel/tracing/trace" - cmd = "sudo sh -c '%s'" % inner + outer = "sudo sh -c '%s'" % inner + out = "" + ssh = os.environ.get("SEMFIX_SSH_PORT", "2222") + ssh_cmd = [ + "sshpass", "-p", os.environ.get("VGPU_GUEST_PASS", "ubuntu"), + "ssh", "-o", "StrictHostKeyChecking=no", + "-o", "UserKnownHostsFile=/dev/null", + "-o", "PubkeyAuthentication=no", "-o", "ConnectTimeout=6", + "-p", ssh, "ubuntu@127.0.0.1", outer, + ] try: - out = subprocess.run([VMEXEC, "exec", cmd], capture_output=True, - text=True, timeout=90).stdout + r = subprocess.run(ssh_cmd, capture_output=True, text=True, + timeout=25) + out = r.stdout or "" except Exception: - return set(), set(), set() + out = "" + if not out: + try: + out = subprocess.run([VMEXEC, "exec", outer], + capture_output=True, text=True, + timeout=25).stdout or "" + except Exception: + return set(), set(), set() sems = {int(m, 16) for m in re.findall(r"sem=0x([0-9a-fA-F]+)", out)} pbs = {int(m, 16) for m in re.findall(r"pb=0x([0-9a-fA-F]+)", out)} chans = {int(m, 16) for m in re.findall(r"ch=0x([0-9a-fA-F]+)", out)} @@ -369,8 +447,11 @@ def resolve_pushbuffer_wide(q, sem, d): def pin(va, pa): cmd = "printf '0x%x,0x%x' > %s/sema_map_add" % (va, pa, PARAM) for _ in range(40): - r = subprocess.run(["sudo", "-n", "sh", "-c", cmd], - capture_output=True, text=True) + try: + r = subprocess.run(["sudo", "-n", "sh", "-c", cmd], + capture_output=True, text=True, timeout=5) + except subprocess.TimeoutExpired: + continue if r.returncode == 0 and "timed out" not in (r.stderr or ""): return True time.sleep(0.05) # async queue full: drain and retry @@ -380,8 +461,11 @@ def pin(va, pa): def pin_bulk(va, pa, nbytes): cmd = "printf '0x%x,0x%x,0x%x' > %s/sema_map_add" % (va, pa, nbytes, PARAM) for _ in range(40): - r = subprocess.run(["sudo", "-n", "sh", "-c", cmd], - capture_output=True, text=True) + try: + r = subprocess.run(["sudo", "-n", "sh", "-c", cmd], + capture_output=True, text=True, timeout=5) + except subprocess.TimeoutExpired: + continue if r.returncode == 0 and "timed out" not in (r.stderr or ""): return True time.sleep(0.05) |