diff options
Diffstat (limited to 'pane.py')
| -rw-r--r-- | pane.py | 210 |
1 files changed, 210 insertions, 0 deletions
@@ -0,0 +1,210 @@ +"""pane.py: tmux pane manager for gdb-driver (Milestone 4). + +Reusable core (target-independent shape): spawn a visible tmux pane running +the backend, register it, poll until *ready*, and tear it down without leaking +panes/sockets. Only the spawn command + readiness check are gdb-specific. + +Registry lives at $XDG_RUNTIME_DIR/gdb-driver/panes.json, keyed by socket path. +""" + +import json +import os +import shlex +import socket +import subprocess +import time +import uuid + +HERE = os.path.dirname(os.path.abspath(__file__)) +DRIVER = os.path.join(HERE, "driver.py") + + +def runtime_dir(): + base = os.environ.get("XDG_RUNTIME_DIR") or f"/tmp/gdb-driver-{os.getuid()}" + d = os.path.join(base, "gdb-driver") + os.makedirs(d, mode=0o700, exist_ok=True) + return d + + +def _reg_path(): + return os.path.join(runtime_dir(), "panes.json") + + +def _load(): + try: + with open(_reg_path()) as f: + return json.load(f) + except (FileNotFoundError, json.JSONDecodeError): + return {} + + +def _save(reg): + tmp = _reg_path() + ".tmp" + with open(tmp, "w") as f: + json.dump(reg, f, indent=2) + os.replace(tmp, _reg_path()) + + +# -------------------------------------------------------------------------- +# liveness +# -------------------------------------------------------------------------- + +def _pane_alive(pane_id): + try: + out = subprocess.check_output( + ["tmux", "list-panes", "-a", "-F", "#{pane_id}"], text=True) + except (subprocess.CalledProcessError, FileNotFoundError): + return False + return pane_id in out.split() + + +def _rpc(sock, method, params=None, timeout=1.0): + s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + s.settimeout(timeout) + try: + s.connect(sock) + f = s.makefile("rwb") + f.write((json.dumps({"id": 1, "method": method, + "params": params or {}}) + "\n").encode()) + f.flush() + line = f.readline() + return json.loads(line) if line else None + finally: + s.close() + + +def sock_ready(sock, timeout=0.5): + """True iff the driver answers ping with ready=True.""" + try: + r = _rpc(sock, "ping", timeout=timeout) + except OSError: + return False + return bool(r and r.get("result", {}).get("ready")) + + +# -------------------------------------------------------------------------- +# spawn / stop / list +# -------------------------------------------------------------------------- + +def spawn(binary, args=None, split="-h", ready_timeout=15.0, env=None): + """Split a tmux pane running gdb+driver on a fresh socket; wait for ready. + + `env` is a list of 'KEY=VALUE' strings exported for gdb AND inherited by the + inferior (e.g. LD_LIBRARY_PATH for shim libs, LD_PRELOAD). + Returns the registry entry dict {sock,pane,bin,args,env,created}. + """ + if "TMUX" not in os.environ: + raise RuntimeError("not inside a tmux session") + binary = os.path.abspath(binary) + if not os.path.exists(binary): + raise FileNotFoundError(binary) + args = args or [] + env = env or [] + sock = os.path.join(runtime_dir(), f"gdb-{uuid.uuid4().hex[:8]}.sock") + + env_tokens = [f"GDB_DRIVER_SOCK={shlex.quote(sock)}"] + env_tokens += [shlex.quote(e) for e in env] + inner = "env {envs} gdb -q -x {drv} --args {bin} {args}".format( + envs=" ".join(env_tokens), + drv=shlex.quote(DRIVER), + bin=shlex.quote(binary), + args=" ".join(shlex.quote(a) for a in args), + ).strip() + + pane = subprocess.check_output( + ["tmux", "split-window", split, "-d", "-P", "-F", "#{pane_id}", inner], + text=True).strip() + + entry = {"sock": sock, "pane": pane, "bin": binary, "args": args, + "env": env, "created": time.time()} + reg = _load() + reg[sock] = entry + _save(reg) + + # readiness poll: socket up AND driver reports ready + deadline = time.time() + ready_timeout + while time.time() < deadline: + if sock_ready(sock): + return entry + if not _pane_alive(pane): + _deregister(sock) + raise RuntimeError(f"pane {pane} died before becoming ready") + time.sleep(0.15) + # timed out + stop(sock) + raise TimeoutError(f"gdb did not become ready in {ready_timeout}s") + + +def _deregister(sock): + reg = _load() + reg.pop(sock, None) + _save(reg) + try: + os.unlink(sock) + except OSError: + pass + + +def stop(sock): + """Graceful quit (best-effort) then kill the pane(s); deregister.""" + reg = _load() + entry = reg.get(sock) + try: + _rpc(sock, "quit", timeout=2.0) + except OSError: + pass + if entry: + subprocess.run(["tmux", "kill-pane", "-t", entry["pane"]], + stderr=subprocess.DEVNULL) + _deregister(sock) + return entry is not None + + +# -------------------------------------------------------------------------- +# raw keys (client-side tmux; bypasses the RPC gate so it works even while a +# blocking run-control call is in flight -- e.g. C-c to interrupt) +# -------------------------------------------------------------------------- + +def entry(sock): + return _load().get(sock) + + +def send_keys(pane_id, tokens): + """Send named tmux key tokens (Enter, C-c, Escape, ...) to a pane.""" + subprocess.run(["tmux", "send-keys", "-t", pane_id] + list(tokens), + stderr=subprocess.DEVNULL) + + +def list_all(prune=False): + """Return [{sock,pane,bin,alive,ready}]; optionally drop dead entries.""" + reg = _load() + out = [] + changed = False + for sock, e in list(reg.items()): + alive = _pane_alive(e["pane"]) + ready = sock_ready(sock) if alive else False + if prune and not alive: + _deregister(sock) + changed = True + continue + out.append({"sock": sock, "pane": e["pane"], "bin": e["bin"], + "alive": alive, "ready": ready}) + if changed: + pass + return out + + +def resolve(explicit=None): + """Pick the socket to drive: explicit > env > the single live one.""" + if explicit: + return explicit + env = os.environ.get("GDB_DRIVER_SOCK") + if env: + return env + live = [e for e in list_all() if e["ready"]] + if len(live) == 1: + return live[0]["sock"] + if not live: + raise RuntimeError("no live gdb-driver pane (try: drive spawn <bin>)") + raise RuntimeError("multiple live panes; pass --sock:\n " + + "\n ".join(e["sock"] for e in live)) |