summaryrefslogtreecommitdiffstats
path: root/tools/guest-diag/pin_gr_stream.py
diff options
context:
space:
mode:
Diffstat (limited to 'tools/guest-diag/pin_gr_stream.py')
-rw-r--r--tools/guest-diag/pin_gr_stream.py178
1 files changed, 121 insertions, 57 deletions
diff --git a/tools/guest-diag/pin_gr_stream.py b/tools/guest-diag/pin_gr_stream.py
index dff2d2d..0072b20 100644
--- a/tools/guest-diag/pin_gr_stream.py
+++ b/tools/guest-diag/pin_gr_stream.py
@@ -32,12 +32,22 @@ LO = os.environ.get("PINWIN_LO", "0x200000000")
HI = os.environ.get("PINWIN_HI", "0x206000000")
PAMAX = os.environ.get("PINWIN_PA_MAX", "0x180000000")
SEMSTREAM = os.environ.get("SEM_STREAM_FILE", "/tmp/opencode/sem_stream.txt")
+# Liveness file checked by pin_supervisor.sh (mtime must stay < 30 s).
+HEARTBEAT_FILE = os.environ.get("PIN_GR_HEARTBEAT", "/tmp/opencode/pin_gr_heartbeat")
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 _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."""
@@ -60,7 +70,8 @@ def vexec(cmd, timeout=40):
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")
# Kill EVERY existing scanner, not just the pid-file one: a retry after
@@ -90,6 +101,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
@@ -101,8 +113,12 @@ def install_and_start():
def inject(va, pa, ln):
cmd = "printf '0x%x,0x%x,0x%x,1' > %s/sema_map_add" % (va, pa, ln, PARAM)
for _ in range(60):
- r = subprocess.run(["sudo", "-n", "sh", "-c", cmd],
- capture_output=True, text=True)
+ try:
+ r = subprocess.run(["sudo", "-n", "sh", "-c", cmd],
+ capture_output=True, text=True, timeout=10)
+ except subprocess.TimeoutExpired:
+ time.sleep(0.05)
+ continue
if r.returncode == 0:
return True
time.sleep(0.05)
@@ -115,8 +131,12 @@ def scrub_register(fin_pa, hsem_pa):
cmd = ("printf '0x%x,0x%x' > %s/sema_scrubbuf_add"
% (fin_pa, hsem_pa, PARAM))
for _ in range(60):
- r = subprocess.run(["sudo", "-n", "sh", "-c", cmd],
- capture_output=True, text=True)
+ try:
+ r = subprocess.run(["sudo", "-n", "sh", "-c", cmd],
+ capture_output=True, text=True, timeout=10)
+ except subprocess.TimeoutExpired:
+ time.sleep(0.05)
+ continue
if r.returncode == 0:
log("scrub registered fin=0x%x hsem=0x%x" % (fin_pa, hsem_pa))
return True
@@ -166,81 +186,116 @@ def _demand_loop():
log("demand: sent %d stalled VA(s) to guest" % len(todo))
+_DEMAND_STARTED = False
+
+
def main():
+ global _DEMAND_STARTED
+ _heartbeat()
if not install_and_start():
log("WARN: could not start guest scanner (will retry via vsock wait)")
- try:
- threading.Thread(target=_demand_loop, daemon=True).start()
- except Exception as e:
- log("WARN: demand thread failed to start: %s" % e)
+ if not _DEMAND_STARTED:
+ _DEMAND_STARTED = True
+ try:
+ threading.Thread(target=_demand_loop, daemon=True).start()
+ except Exception as e:
+ log("WARN: demand thread failed to start: %s" % e)
srv = socket.socket(socket.AF_VSOCK, socket.SOCK_STREAM)
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind((socket.VMADDR_CID_ANY, PORT))
+ srv.settimeout(10)
srv.listen(4)
log("listening on vsock port %d (window %s..%s)" % (PORT, LO, HI))
seen = {}
pin_total = 0
- buf = b""
+ last_conn = time.time()
+ last_reinstall = 0.0
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).
+ # Re-run install_and_start() at most once per 15 s.
+ if now - last_conn > 30.0 and now - last_reinstall >= 15.0:
+ last_reinstall = now
+ log("reinstall attempt: no guest connection for %.0fs"
+ % (now - last_conn))
+ try:
+ install_and_start()
+ except Exception as e:
+ log("reinstall failed: %s" % e)
+ continue
except OSError as e:
log("accept failed: %s" % e)
time.sleep(1)
continue
+ last_conn = time.time()
+ _heartbeat()
conn.settimeout(5)
log("guest connected")
buf = b""
runs = []
- while True:
- try:
- chunk = conn.recv(1 << 20)
- except socket.timeout:
- if runs:
- _flush(runs, seen)
- runs = []
- continue
- except OSError:
- break
- if not chunk:
- break
- buf += chunk
- while b"\n" in buf:
- line, buf = buf.split(b"\n", 1)
- text = line.decode(errors="replace").strip()
- if not text:
- continue
- if text.startswith("#pid"):
- m = re.search(r"(\d+)", text)
- seen = {}
- log("guest reports cdt pid %s" % (m.group(1) if m else "?"))
- _mirror(text)
- continue
- if text.startswith("#sem"):
- _mirror(text)
- continue
- if text.startswith("#pb"):
- _mirror(text)
- continue
- if text.startswith("#put"):
- log(text)
- _mirror(text)
+ try:
+ while True:
+ try:
+ chunk = conn.recv(1 << 20)
+ except socket.timeout:
+ if runs:
+ _flush(runs, seen)
+ runs = []
continue
- if text.startswith("#scrub"):
- m = re.search(r"0x([0-9a-fA-F]+),0x([0-9a-fA-F]+)", text)
- _mirror(text)
+ except OSError:
+ break
+ if not chunk:
+ break
+ _heartbeat()
+ buf += chunk
+ while b"\n" in buf:
+ line, buf = buf.split(b"\n", 1)
+ text = line.decode(errors="replace").strip()
+ if not text:
+ continue
+ if text.startswith("#pid"):
+ m = re.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
+ if text.startswith("#scrub"):
+ m = re.search(r"0x([0-9a-fA-F]+),0x([0-9a-fA-F]+)",
+ text)
+ _mirror(text)
+ if m:
+ scrub_register(int(m.group(1), 16),
+ int(m.group(2), 16))
+ continue
+ m = re.match(r"^(0x[0-9a-f]+),(0x[0-9a-f]+)(?:,(\d+))?$",
+ text)
if m:
- scrub_register(int(m.group(1), 16),
- int(m.group(2), 16))
- continue
- m = re.match(r"^(0x[0-9a-f]+),(0x[0-9a-f]+)(?:,(\d+))?$", text)
- if m:
- runs.append((int(m.group(1), 16), int(m.group(2), 16),
- int(m.group(3)) if m.group(3) else 1))
- if runs:
- pin_total += _flush(runs, seen)
- runs = []
+ runs.append((int(m.group(1), 16),
+ int(m.group(2), 16),
+ int(m.group(3)) if m.group(3) else 1))
+ if runs:
+ pin_total += _flush(runs, seen)
+ runs = []
+ 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()
@@ -289,6 +344,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)
@@ -322,4 +378,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)
generated by cgit v1.2.3 (git 2.39.1) at 2026-10-25 08:16:08 +0000