1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
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))
|
