438 lines
18 KiB
Python
Executable File
438 lines
18 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Drives the C++ and Rust debug builds under gdb, and the Python build
|
|
under debugpy, to their equivalent breakpoint (right after the 30/20/10
|
|
list is built), evaluates the JSON-producing debug-graph helper exactly as
|
|
VS Code's "Debug Visualizer" extension would (via -data-evaluate-expression
|
|
for gdb, an `evaluate` DAP request for debugpy), and replays the
|
|
extension's own parseEvaluationResultFromGenericDebugAdapter algorithm on
|
|
the raw result -- so we can tell, without going through the VS Code UI,
|
|
whether our JSON will actually be recognized or whether the extension will
|
|
fall back to its generic byte-walk visualization.
|
|
|
|
The parsing port below mirrors:
|
|
external/vscode-debug-visualizer/extension/src/VisualizationBackend/
|
|
parseEvaluationResultFromGenericDebugAdapter.ts
|
|
Keep it in sync if that file changes upstream.
|
|
|
|
Writes debug_cpp.json / debug_rust.json / debug_python.json into the
|
|
output directory (default ~/lg, matching the style of the extension's own
|
|
captured dumps): either the resolved data object on success, or a
|
|
{"PARSE_ERROR": ..., "raw": ...} diagnostic on failure.
|
|
|
|
For the C++ build specifically, the breakpoint sits inside
|
|
`if (const int* top = list.peek_front())`, so `top` is a live local pointing
|
|
at the head node's `data` field (the node's first member, so its address
|
|
equals the node's own address, but its *static type* is just `const int*` --
|
|
gdb has no way to know it's secretly interior to a `Node<int>`). VS Code's
|
|
"debugVisualizer.debugAdapterConfigurations" expressionTemplate wraps
|
|
whatever you type in the Watch box as $debug_graph(${expr}), so typing
|
|
plain `top` evaluates $debug_graph(top) -- not $debug_graph(list, ...): we
|
|
replay that exact call and verify it still delves into the full list. This
|
|
relies on debuggers/gdb_debug_graph.py's auto-discovery of sibling locals (here,
|
|
`list`, a struct type it *can* walk) merging into the same node graph as
|
|
`top`'s address, rather than requiring `top` to be passed as an explicit
|
|
watched root.
|
|
|
|
The Python build has no pointers -- `top` there is just a plain int, not a
|
|
value interior to any struct -- but debuggers/py_debug_graph.py's own
|
|
auto-discovery of sibling locals (here `lst`) means evaluating
|
|
debug_graph(top) still delves into the whole list for the same underlying
|
|
reason: verify_top_delves_into_list_py exercises that.
|
|
"""
|
|
import ast
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
REPO = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def run_gdb(gdb_bin, binary, break_file, break_line, expr="$debug_graph(list)"):
|
|
commands = [
|
|
"-gdb-set print elements 0",
|
|
"-gdb-set print characters 0",
|
|
"source %s" % (REPO / "debuggers" / "gdb_debug_graph.py"),
|
|
"-break-insert %s:%d" % (break_file, break_line),
|
|
"-exec-run",
|
|
'-data-evaluate-expression "%s"' % expr.replace('"', '\\"'),
|
|
]
|
|
proc = subprocess.run(
|
|
[gdb_bin, "--interpreter=mi", binary],
|
|
input="\n".join(commands).encode(),
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
timeout=15,
|
|
cwd=str(REPO),
|
|
)
|
|
out = proc.stdout.decode(errors="replace")
|
|
matches = re.findall(r'\^done,value="(.*)"\s*$', out, re.M)
|
|
errors = re.findall(r'\^error,msg="(.*)"\s*$', out, re.M)
|
|
if not matches:
|
|
raise RuntimeError("no ^done,value=... in gdb output; errors=%r\n---\n%s" % (errors, out))
|
|
return matches[-1]
|
|
|
|
|
|
def is_enclosed_with(s, ch):
|
|
return s.startswith(ch) and s.endswith(ch) and len(s) >= 2
|
|
|
|
|
|
def parse_like_extension(mi_escaped_value):
|
|
"""Port of parseEvaluationResultFromGenericDebugAdapter, applied to the
|
|
raw MI `value="..."` payload (itself one layer of MI's own
|
|
string-escaping around whatever gdb's evaluate actually reported)."""
|
|
result_text = json.loads('"' + mi_escaped_value + '"') # undo MI's escaping
|
|
json_data = result_text.strip()
|
|
|
|
try:
|
|
try:
|
|
if is_enclosed_with(json_data, '"') or is_enclosed_with(json_data, "'"):
|
|
json_data2 = json_data[1:-1]
|
|
else:
|
|
json_data2 = json_data
|
|
result_obj = json.loads(json_data2)
|
|
except Exception:
|
|
# "in case of C++": the whole thing is itself a JSON string
|
|
# literal; unwrap it once, then parse *that* as JSON.
|
|
s = json.loads(json_data)
|
|
result_obj = json.loads(s)
|
|
return {"ok": True, "data": result_obj}
|
|
except Exception as e:
|
|
return {"ok": False, "error": str(e), "raw": result_text}
|
|
|
|
|
|
def find_debugpy_python():
|
|
"""Locates a Python interpreter with `debugpy` importable: the debugpy
|
|
analog of picking `gdb`/`rust-gdb` off the PATH above. Checked in order:
|
|
an explicit override, the interpreter running this script, then a
|
|
couple of common places a dev might have installed it, since debugpy
|
|
isn't always on the system Python (this repo's system python3 is on a
|
|
version too new for the Debian-packaged debugpy, so it typically lives
|
|
in a venv)."""
|
|
candidates = []
|
|
override = os.environ.get("DUMP_DEBUG_GRAPH_PYTHON")
|
|
if override:
|
|
candidates.append(override)
|
|
candidates.append(sys.executable)
|
|
candidates.append(str(Path.home() / "py314" / "bin" / "python3"))
|
|
which = shutil.which("python3")
|
|
if which:
|
|
candidates.append(which)
|
|
|
|
for candidate in candidates:
|
|
if not candidate:
|
|
continue
|
|
try:
|
|
subprocess.run(
|
|
[candidate, "-c", "import debugpy"],
|
|
check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10,
|
|
)
|
|
except Exception:
|
|
continue
|
|
return candidate
|
|
raise RuntimeError(
|
|
"no python interpreter with debugpy found; set DUMP_DEBUG_GRAPH_PYTHON=/path/to/python"
|
|
)
|
|
|
|
|
|
class _DapClient:
|
|
"""Minimal Debug Adapter Protocol client: just enough of the wire
|
|
protocol (Content-Length-framed JSON over a TCP socket) to drive
|
|
debugpy the same way -interpreter=mi drives gdb above -- no IDE, no
|
|
`debugpy` package needed on this side, since debugpy itself only needs
|
|
to be importable by the *debuggee* interpreter."""
|
|
|
|
def __init__(self, sock):
|
|
self._sock = sock
|
|
self._buf = b""
|
|
self._seq = 0
|
|
|
|
def send(self, command, arguments=None):
|
|
self._seq += 1
|
|
msg = {"seq": self._seq, "type": "request", "command": command}
|
|
if arguments is not None:
|
|
msg["arguments"] = arguments
|
|
body = json.dumps(msg).encode("utf-8")
|
|
self._sock.sendall(("Content-Length: %d\r\n\r\n" % len(body)).encode("ascii") + body)
|
|
return self._seq
|
|
|
|
def _read_message(self):
|
|
while b"\r\n\r\n" not in self._buf:
|
|
chunk = self._sock.recv(4096)
|
|
if not chunk:
|
|
raise EOFError("debugpy closed the connection")
|
|
self._buf += chunk
|
|
header, _, rest = self._buf.partition(b"\r\n\r\n")
|
|
length = int(header.split(b":")[1].strip())
|
|
while len(rest) < length:
|
|
chunk = self._sock.recv(4096)
|
|
if not chunk:
|
|
raise EOFError("debugpy closed the connection")
|
|
rest += chunk
|
|
body, self._buf = rest[:length], rest[length:]
|
|
return json.loads(body.decode("utf-8"))
|
|
|
|
def wait_for(self, predicate, timeout=15):
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
msg = self._read_message()
|
|
if predicate(msg):
|
|
return msg
|
|
raise TimeoutError("timed out waiting for a matching DAP message")
|
|
|
|
|
|
def run_debugpy(python_bin, script_path, break_line, expr):
|
|
"""Launches `script_path` under debugpy (paused via --wait-for-client),
|
|
connects a plain DAP client, sets one breakpoint, waits for it to hit,
|
|
and evaluates `expr` in that frame -- the debugpy equivalent of
|
|
run_gdb()'s -data-evaluate-expression. `context: "watch"` mirrors what
|
|
VS Code sends for a Watch-panel entry (rather than "repl"), matching
|
|
how the Debug Visualizer extension's expressionTemplate is actually
|
|
invoked. Returns the raw `result` string from the evaluate response
|
|
(debugpy's repr() of whatever the expression returned)."""
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
|
|
probe.bind(("127.0.0.1", 0))
|
|
port = probe.getsockname()[1]
|
|
|
|
env = dict(os.environ, PYTHONPATH=str(REPO / "debuggers"))
|
|
proc = subprocess.Popen(
|
|
[python_bin, "-m", "debugpy", "--listen", "127.0.0.1:%d" % port, "--wait-for-client", str(script_path)],
|
|
cwd=str(Path(script_path).parent),
|
|
env=env,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
)
|
|
try:
|
|
sock = None
|
|
for _ in range(50):
|
|
try:
|
|
sock = socket.create_connection(("127.0.0.1", port), timeout=1)
|
|
break
|
|
except OSError:
|
|
time.sleep(0.2)
|
|
if sock is None:
|
|
raise RuntimeError("debugpy never opened its listen socket")
|
|
|
|
with sock:
|
|
client = _DapClient(sock)
|
|
client.send("initialize", {
|
|
"clientID": "dump-debug-graph", "adapterID": "debugpy", "pathFormat": "path",
|
|
"linesStartAt1": True, "columnsStartAt1": True,
|
|
})
|
|
client.wait_for(lambda m: m.get("type") == "response" and m.get("command") == "initialize")
|
|
|
|
client.send("attach", {"justMyCode": False})
|
|
client.wait_for(lambda m: m.get("type") == "event" and m.get("event") == "initialized")
|
|
|
|
client.send("setBreakpoints", {
|
|
"source": {"path": str(script_path)},
|
|
"breakpoints": [{"line": break_line}],
|
|
})
|
|
client.wait_for(lambda m: m.get("type") == "response" and m.get("command") == "setBreakpoints")
|
|
|
|
client.send("configurationDone")
|
|
client.wait_for(lambda m: m.get("type") == "response" and m.get("command") == "configurationDone")
|
|
client.wait_for(lambda m: m.get("type") == "response" and m.get("command") == "attach")
|
|
|
|
stopped = client.wait_for(lambda m: m.get("type") == "event" and m.get("event") == "stopped")
|
|
thread_id = stopped["body"]["threadId"]
|
|
|
|
st_seq = client.send("stackTrace", {"threadId": thread_id})
|
|
frames = client.wait_for(lambda m: m.get("type") == "response" and m.get("request_seq") == st_seq)
|
|
frame_id = frames["body"]["stackFrames"][0]["id"]
|
|
|
|
ev_seq = client.send("evaluate", {"expression": expr, "frameId": frame_id, "context": "watch"})
|
|
evaluated = client.wait_for(lambda m: m.get("type") == "response" and m.get("request_seq") == ev_seq)
|
|
if not evaluated.get("success"):
|
|
raise RuntimeError("evaluate failed: %r" % (evaluated.get("message"),))
|
|
result = evaluated["body"]["result"]
|
|
|
|
client.send("continue", {"threadId": thread_id})
|
|
return result
|
|
finally:
|
|
try:
|
|
proc.wait(timeout=10)
|
|
except subprocess.TimeoutExpired:
|
|
proc.kill()
|
|
proc.wait(timeout=10)
|
|
|
|
|
|
def parse_python_result(raw_result):
|
|
"""debugpy's evaluate response reports a string return value as
|
|
Python's own repr() of it (e.g. `'{"kind": ...}'`), not raw text --
|
|
ast.literal_eval() is the exact inverse of that repr(), so it recovers
|
|
our debug_graph()-produced JSON string before the normal json.loads()."""
|
|
try:
|
|
json_text = ast.literal_eval(raw_result)
|
|
result_obj = json.loads(json_text)
|
|
return {"ok": True, "data": result_obj}
|
|
except Exception as e:
|
|
return {"ok": False, "error": str(e), "raw": raw_result}
|
|
|
|
|
|
def verify_top_delves_into_list_py(data):
|
|
"""Python analog of verify_top_delves_into_list: checks that evaluating
|
|
bare `top` (a plain int with no reference graph of its own) still shows
|
|
the whole 30/20/10 list because `lst`, a sibling local at the same
|
|
breakpoint, gets auto-discovered and walked too."""
|
|
nodes = data.get("nodes", [])
|
|
edges = data.get("edges", [])
|
|
roots_by_name = {r.get("name"): r for r in data.get("roots", [])}
|
|
|
|
node_shaped = [
|
|
n for n in nodes
|
|
if [f.get("name") for f in n.get("fields", [])] == ["data", "next"]
|
|
]
|
|
if not node_shaped:
|
|
print(" VERIFY FAIL: debug_graph(top) produced no data+next Node-shaped node -- "
|
|
"nodes=%r" % [n.get("id") for n in nodes])
|
|
return False
|
|
|
|
next_edges = [e for e in edges if e.get("label") == "next"]
|
|
if len(next_edges) < 2:
|
|
print(" VERIFY FAIL: expected 2 'next' edges chaining the 30/20/10 nodes, found %d "
|
|
"(edges=%r)" % (len(next_edges), edges))
|
|
return False
|
|
|
|
if "lst" not in roots_by_name:
|
|
print(" VERIFY FAIL: 'lst' was not auto-discovered as a sibling local alongside "
|
|
"top (roots=%r)" % data.get("roots"))
|
|
return False
|
|
|
|
print(" VERIFY OK: debug_graph(top) delves into the full list -- %d nodes, %d 'next' "
|
|
"edges, 'lst' auto-discovered as a %r root"
|
|
% (len(nodes), len(next_edges), roots_by_name["lst"].get("kind")))
|
|
return True
|
|
|
|
|
|
def verify_top_delves_into_list(data):
|
|
"""Checks that evaluating bare `top` -- i.e. $debug_graph(top), exactly
|
|
what VS Code's expressionTemplate produces for a Watch box containing
|
|
just "top" -- still shows the whole 30/20/10 list rather than stopping
|
|
at a lone scalar leaf. Returns True/False and prints why."""
|
|
nodes = data.get("nodes", [])
|
|
edges = data.get("edges", [])
|
|
roots_by_name = {r.get("name"): r for r in data.get("roots", [])}
|
|
|
|
# Other auto-discovered locals (e.g. a `char*` local) can legitimately
|
|
# produce their own bare scalar-leaf nodes -- only care whether *some*
|
|
# node looks like a real Node<int> (data+next), i.e. top's address got
|
|
# upgraded into the full struct rather than stopping at a lone int.
|
|
node_shaped = [
|
|
n for n in nodes
|
|
if [f.get("name") for f in n.get("fields", [])] == ["data", "next"]
|
|
]
|
|
if not node_shaped:
|
|
print(" VERIFY FAIL: $debug_graph(top) produced no data+next Node-shaped node -- "
|
|
"top's address was never upgraded to the full Node it coincides with "
|
|
"(nodes=%r)" % [n.get("id") for n in nodes])
|
|
return False
|
|
|
|
next_edges = [e for e in edges if e.get("label") == "next"]
|
|
if len(next_edges) < 2:
|
|
print(" VERIFY FAIL: expected 2 'next' edges chaining the 30/20/10 nodes, found %d "
|
|
"(edges=%r)" % (len(next_edges), edges))
|
|
return False
|
|
|
|
if "list" not in roots_by_name:
|
|
print(" VERIFY FAIL: 'list' was not auto-discovered as a sibling local alongside "
|
|
"top (roots=%r)" % data.get("roots"))
|
|
return False
|
|
|
|
print(" VERIFY OK: $debug_graph(top) delves into the full list -- %d nodes, %d 'next' "
|
|
"edges, 'list' auto-discovered as a %r root"
|
|
% (len(nodes), len(next_edges), roots_by_name["list"].get("kind")))
|
|
return True
|
|
|
|
|
|
def _write_dump(label, out_dir, result):
|
|
out_path = out_dir / ("debug_%s.json" % label)
|
|
out_path.write_text(json.dumps(result, indent=4))
|
|
print("Wrote %s (%s)" % (out_path, "OK" if "PARSE_ERROR" not in result else "PARSE_ERROR"))
|
|
|
|
|
|
def dump(label, gdb_bin, binary, break_file, break_line, out_dir, expr="$debug_graph(list)", verify=None):
|
|
ok = True
|
|
try:
|
|
raw_value = run_gdb(gdb_bin, binary, break_file, break_line, expr)
|
|
except Exception as e:
|
|
result = {"PARSE_ERROR": "gdb invocation failed", "detail": str(e)}
|
|
ok = False
|
|
else:
|
|
parsed = parse_like_extension(raw_value)
|
|
if parsed["ok"]:
|
|
result = parsed["data"]
|
|
if verify is not None:
|
|
ok = verify(result)
|
|
else:
|
|
result = {
|
|
"PARSE_ERROR": parsed["error"],
|
|
"raw": parsed["raw"],
|
|
"note": "This is what the Debug Visualizer extension would fail to parse, "
|
|
"triggering its generic byte-walk fallback (constructGraphFromVariablesReference).",
|
|
}
|
|
ok = False
|
|
|
|
_write_dump(label, out_dir, result)
|
|
return ok
|
|
|
|
|
|
def dump_python(label, python_bin, script_path, break_line, out_dir,
|
|
expr="__import__('py_debug_graph').debug_graph(lst)", verify=None):
|
|
ok = True
|
|
try:
|
|
raw_result = run_debugpy(python_bin, script_path, break_line, expr)
|
|
except Exception as e:
|
|
result = {"PARSE_ERROR": "debugpy invocation failed", "detail": str(e)}
|
|
ok = False
|
|
else:
|
|
parsed = parse_python_result(raw_result)
|
|
if parsed["ok"]:
|
|
result = parsed["data"]
|
|
if verify is not None:
|
|
ok = verify(result)
|
|
else:
|
|
result = {
|
|
"PARSE_ERROR": parsed["error"],
|
|
"raw": parsed["raw"],
|
|
"note": "This is what the Debug Visualizer extension would fail to parse, "
|
|
"triggering its generic byte-walk fallback (constructGraphFromVariablesReference).",
|
|
}
|
|
ok = False
|
|
|
|
_write_dump(label, out_dir, result)
|
|
return ok
|
|
|
|
|
|
def main():
|
|
out_dir = Path(sys.argv[1]) if len(sys.argv) > 1 else Path.home() / "lg"
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
ok_cpp = dump("cpp", "gdb", str(REPO / "cpp" / "list_example_cpp"), "cpp/main.cpp", 70, out_dir)
|
|
ok_cpp_top = dump(
|
|
"cpp_top", "gdb", str(REPO / "cpp" / "list_example_cpp"), "cpp/main.cpp", 70, out_dir,
|
|
expr="$debug_graph(top)",
|
|
verify=verify_top_delves_into_list,
|
|
)
|
|
ok_rust = dump("rust", "rust-gdb", str(REPO / "target" / "debug" / "list_example"), "src/main.rs", 51, out_dir)
|
|
|
|
python_bin = find_debugpy_python()
|
|
python_script = REPO / "python" / "main.py"
|
|
ok_python = dump_python("python", python_bin, python_script, 54, out_dir)
|
|
ok_python_top = dump_python(
|
|
"python_top", python_bin, python_script, 54, out_dir,
|
|
expr="__import__('py_debug_graph').debug_graph(top)",
|
|
verify=verify_top_delves_into_list_py,
|
|
)
|
|
|
|
if not (ok_cpp and ok_cpp_top and ok_rust and ok_python and ok_python_top):
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|