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
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
|
import os.path
import re
import shlex
import subprocess
import time
import pexpect
# Note that gdb comes with its own testsuite. I was unable to figure out how to
# run that testsuite against the spike simulator.
def find_file(path):
for directory in (os.getcwd(), os.path.dirname(__file__)):
fullpath = os.path.join(directory, path)
if os.path.exists(fullpath):
return fullpath
return None
def compile(args, xlen=32): # pylint: disable=redefined-builtin
cc = os.path.expandvars("$RISCV/bin/riscv%d-unknown-elf-gcc" % xlen)
cmd = [cc, "-g"]
for arg in args:
found = find_file(arg)
if found:
cmd.append(found)
else:
cmd.append(arg)
cmd = " ".join(cmd)
result = os.system(cmd)
assert result == 0, "%r failed" % cmd
def unused_port():
# http://stackoverflow.com/questions/2838244/get-open-tcp-port-in-python/2838309#2838309
import socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(("", 0))
port = s.getsockname()[1]
s.close()
return port
class Spike(object):
logname = "spike.log"
def __init__(self, cmd, binary=None, halted=False, with_gdb=True,
timeout=None, xlen=64):
"""Launch spike. Return tuple of its process and the port it's running
on."""
if cmd:
cmd = shlex.split(cmd)
else:
cmd = ["spike"]
if xlen == 32:
cmd += ["--isa", "RV32"]
if timeout:
cmd = ["timeout", str(timeout)] + cmd
if halted:
cmd.append('-H')
if with_gdb:
self.port = unused_port()
cmd += ['--gdb-port', str(self.port)]
cmd.append("-m32")
cmd.append('pk')
if binary:
cmd.append(binary)
logfile = open(self.logname, "w")
logfile.write("+ %s\n" % " ".join(cmd))
logfile.flush()
self.process = subprocess.Popen(cmd, stdin=subprocess.PIPE,
stdout=logfile, stderr=logfile)
def __del__(self):
try:
self.process.kill()
self.process.wait()
except OSError:
pass
def wait(self, *args, **kwargs):
return self.process.wait(*args, **kwargs)
class VcsSim(object):
def __init__(self, simv=None, debug=False):
if simv:
cmd = shlex.split(simv)
else:
cmd = ["simv"]
cmd += ["+jtag_vpi_enable"]
if debug:
cmd[0] = cmd[0] + "-debug"
cmd += ["+vcdplusfile=output/gdbserver.vpd"]
logfile = open("simv.log", "w")
logfile.write("+ %s\n" % " ".join(cmd))
logfile.flush()
listenfile = open("simv.log", "r")
listenfile.seek(0, 2)
self.process = subprocess.Popen(cmd, stdin=subprocess.PIPE,
stdout=logfile, stderr=logfile)
done = False
while not done:
line = listenfile.readline()
if not line:
time.sleep(1)
if "Listening on port 5555" in line:
done = True
def __del__(self):
try:
self.process.kill()
self.process.wait()
except OSError:
pass
class Openocd(object):
logname = "openocd.log"
def __init__(self, cmd=None, config=None, debug=False, otherProcess=None):
# keep handles to other processes -- don't let them be
# garbage collected yet.
self.otherProcess = otherProcess
if cmd:
cmd = shlex.split(cmd)
else:
cmd = ["openocd"]
if config:
cmd += ["-f", find_file(config)]
if debug:
cmd.append("-d")
logfile = open(Openocd.logname, "w")
logfile.write("+ %s\n" % " ".join(cmd))
logfile.flush()
self.process = subprocess.Popen(cmd, stdin=subprocess.PIPE,
stdout=logfile, stderr=logfile)
# TODO: Pick a random port
self.port = 3333
# Wait for OpenOCD to have made it through riscv_examine(). When using
# OpenOCD to communicate with a simulator this may take a long time,
# and gdb will time out when trying to connect if we attempt too early.
start = time.time()
messaged = False
while True:
log = open(Openocd.logname).read()
if "Examined RISCV core" in log:
break
if not self.process.poll() is None:
raise Exception("OpenOCD exited before completing riscv_examine()")
if not messaged and time.time() - start > 1:
messaged = True
print "Waiting for OpenOCD to examine RISCV core..."
def __del__(self):
try:
self.process.kill()
self.process.wait()
except OSError:
pass
class CannotAccess(Exception):
def __init__(self, address):
Exception.__init__(self)
self.address = address
class Gdb(object):
def __init__(self,
cmd=os.path.expandvars("$RISCV/bin/riscv64-unknown-elf-gdb")):
self.child = pexpect.spawn(cmd)
self.child.logfile = open("gdb.log", "w")
self.child.logfile.write("+ %s\n" % cmd)
self.wait()
self.command("set confirm off")
self.command("set width 0")
self.command("set height 0")
# Force consistency.
self.command("set print entry-values no")
def wait(self):
"""Wait for prompt."""
self.child.expect(r"\(gdb\)")
def command(self, command, timeout=-1):
self.child.sendline(command)
self.child.expect("\n", timeout=timeout)
self.child.expect(r"\(gdb\)", timeout=timeout)
return self.child.before.strip()
def c(self, wait=True):
if wait:
output = self.command("c")
assert "Continuing" in output
return output
else:
self.child.sendline("c")
self.child.expect("Continuing")
def interrupt(self):
self.child.send("\003")
self.child.expect(r"\(gdb\)", timeout=60)
return self.child.before.strip()
def x(self, address, size='w'):
output = self.command("x/%s %s" % (size, address))
value = int(output.split(':')[1].strip(), 0)
return value
def p(self, obj):
output = self.command("p/x %s" % obj)
m = re.search("Cannot access memory at address (0x[0-9a-f]+)", output)
if m:
raise CannotAccess(int(m.group(1), 0))
value = int(output.split('=')[-1].strip(), 0)
return value
def p_string(self, obj):
output = self.command("p %s" % obj)
value = shlex.split(output.split('=')[-1].strip())[1]
return value
def stepi(self):
output = self.command("stepi")
return output
def load(self):
output = self.command("load", timeout=60)
assert "failed" not in output
assert "Transfer rate" in output
def b(self, location):
output = self.command("b %s" % location)
assert "not defined" not in output
assert "Breakpoint" in output
return output
def hbreak(self, location):
output = self.command("hbreak %s" % location)
assert "not defined" not in output
assert "Hardware assisted breakpoint" in output
return output
|