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
|
import os.path
import pexpect
import shlex
import subprocess
import tempfile
import testlib
import unittest
# 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(testlib.__file__)):
fullpath = os.path.join(directory, path)
if os.path.exists(fullpath):
return fullpath
return None
def compile(args, xlen=32):
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):
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('pk')
if binary:
cmd.append(binary)
logfile = open("spike.log", "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 Openocd(object):
def __init__(self, cmd=None, config=None, debug=False):
if cmd:
cmd = shlex.split(cmd)
else:
cmd = ["openocd"]
if config:
cmd += ["-f", find_file(config)]
if debug:
cmd.append("-d")
logfile = open("openocd.log", "w")
logfile.write("+ %s\n" % " ".join(cmd))
self.process = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=logfile,
stderr=logfile)
# TODO: Pick a random port
self.port = 3333
def __del__(self):
try:
self.process.kill()
self.process.wait()
except OSError:
pass
class Gdb(object):
def __init__(self,
path=os.path.expandvars("$RISCV/bin/riscv64-unknown-elf-gdb")):
self.child = pexpect.spawn(path)
self.child.logfile = file("gdb.log", "w")
self.child.logfile.write("+ %s\n" % path)
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("\(gdb\)")
def command(self, command, timeout=-1):
self.child.sendline(command)
self.child.expect("\n", timeout=timeout)
self.child.expect("\(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("\(gdb\)")
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)
value = int(output.split('=')[-1].strip(), 0)
return value
def stepi(self):
output = self.command("stepi")
assert "Cannot" not in output
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
|
