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
|
"""risc-v-gsp-rm-debugger plugin: RM host-oracle harness commands.
Registers REPL commands that run the read-only Open-IOV RM tools through the
``rmharness`` wrapper scripts and record each run into the active session:
risc-v-gsp-rm-debugger> rmtool gpu list
risc-v-gsp-rm-debugger> rmtool mem stats --format=json
risc-v-gsp-rm-debugger> rmtop --once
risc-v-gsp-rm-debugger> rmq
Every run is recorded as a ``rm_harness_run`` event with a captured
stdout+stderr blob (via ``ge.recorder.blob``); write/ownership operations are
refused and recorded as ``rm_harness_skip``. The plugin is stdlib-only and
never touches kernel modules, VFIO, mdevs or the live GPU state.
"""
from __future__ import annotations
import hashlib
import os
import subprocess
import sys
import time
_TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
_HARNESS = os.path.join(_TOOL_ROOT, "rmharness")
_WRAPPERS = {
"rmtool": os.path.join(_HARNESS, "rmtool.sh"),
"rmtop": os.path.join(_HARNESS, "rmtop.sh"),
"rmq": os.path.join(_HARNESS, "rmq.sh"),
}
#: arguments that would name a write/ownership operation on any tool.
_BLOCKED = frozenset({
"alloc", "allocate", "map", "unmap", "pin", "free", "create",
"destroy", "submit", "bind", "unbind", "reset", "start", "stop",
"load", "unload", "reboot", "write", "poke", "set",
})
def _blocked_arg(args):
for arg in args:
token = arg.split("=", 1)[0].lstrip("-").lower()
if token in _BLOCKED:
return arg
return None
def _record(ge, tool, args, proc, duration):
stdout = proc.stdout or ""
stderr = proc.stderr or ""
payload = (stdout + stderr).encode("utf-8", "replace")
blob = None
try:
blob = ge.recorder.blob(
f"rmharness/{tool}-{time.time_ns()}.txt", payload)
except Exception:
blob = None
fields = dict(
plugin="rmharness",
tool=tool,
argv=list(args),
exit=proc.returncode,
duration_s=round(duration, 6),
stdout_sha256=hashlib.sha256(
stdout.encode("utf-8", "replace")).hexdigest(),
stderr_sha256=hashlib.sha256(
stderr.encode("utf-8", "replace")).hexdigest(),
stdout_bytes=len(stdout.encode("utf-8", "replace")),
stderr_bytes=len(stderr.encode("utf-8", "replace")),
blob=blob,
readonly=True,
)
ge.recorder.event("rm_harness_run", **fields)
def _run(ge, tool, args):
wrapper = _WRAPPERS[tool]
if not os.path.isfile(wrapper):
print(f" rmharness: wrapper missing: {wrapper}", file=sys.stderr)
ge.recorder.event("rm_harness_error", plugin="rmharness",
tool=tool, error="wrapper missing")
return 2
bad = _blocked_arg(args)
if bad is not None:
print(f" rmharness: refused '{bad}': read-only policy "
f"(see docs/17-RM-HARNESS.md)", file=sys.stderr)
ge.recorder.event("rm_harness_skip", plugin="rmharness", tool=tool,
argv=list(args), reason=f"blocked argument {bad!r}")
return 2
start = time.monotonic()
proc = subprocess.run([wrapper, *args], capture_output=True, text=True)
duration = time.monotonic() - start
if proc.stdout:
print(proc.stdout, end="")
if proc.stderr:
print(proc.stderr, end="", file=sys.stderr)
_record(ge, tool, args, proc, duration)
directory = getattr(ge.recorder, "dir", None)
if directory:
print(f" rmharness: recorded {tool} run -> {directory}")
return proc.returncode
def register(ge):
ge.log("rmharness plugin loaded; tools=" + ",".join(sorted(_WRAPPERS)))
def cmd_rmtool(session, args):
return _run(ge, "rmtool", args)
def cmd_rmtop(session, args):
return _run(ge, "rmtop", args)
def cmd_rmq(session, args):
return _run(ge, "rmq", args)
ge.add_command("rmtool", cmd_rmtool,
help="rmtool <args>: read-only RM oracle (gpu/engine/mem)")
ge.add_command("rmtop", cmd_rmtop,
help="rmtop <args>: dashboard; use --once for capture")
ge.add_command("rmq", cmd_rmq,
help="rmq: existing tools/rmq host oracle (read-only)")
|
