allow alias SimdSignal<-PartitionedSignal to allow tracking back through
[ieee754fpu.git] / src / add / test_inout_mux_pipe.py
1 """ key strategic example showing how to do multi-input fan-in into a
2 multi-stage pipeline, then multi-output fanout.
3
4 the multiplex ID from the fan-in is passed in to the pipeline, preserved,
5 and used as a routing ID on the fanout.
6 """
7
8 from random import randint
9 from math import log
10 from nmigen import Module, Signal, Cat, Value
11 from nmigen.compat.sim import run_simulation
12 from nmigen.cli import verilog, rtlil
13
14 from multipipe import CombMultiOutPipeline
15 from multipipe import CombMultiInPipeline, InputPriorityArbiter
16 from singlepipe import UnbufferedPipeline
17
18
19 class PriorityUnbufferedPipeline(CombMultiInPipeline):
20 def __init__(self, stage, p_len):
21 p_mux = InputPriorityArbiter(self, p_len)
22 CombMultiInPipeline.__init__(self, stage, p_len=p_len, p_mux=p_mux)
23
24 def ports(self):
25 return self.p_mux.ports()
26 #return UnbufferedPipeline.ports(self) + self.p_mux.ports()
27
28
29 class MuxUnbufferedPipeline(CombMultiOutPipeline):
30 def __init__(self, stage, n_len):
31 # HACK: stage is also the n-way multiplexer
32 CombMultiOutPipeline.__init__(self, stage, n_len=n_len, n_mux=stage)
33
34 # HACK: n-mux is also the stage... so set the muxid equal to input mid
35 stage.m_id = self.p.i_data.mid
36
37 def ports(self):
38 return self.p_mux.ports()
39
40
41 class PassData: # (Value):
42 def __init__(self):
43 self.mid = Signal(2, reset_less=True)
44 self.idx = Signal(8, reset_less=True)
45 self.data = Signal(16, reset_less=True)
46
47 def _rhs_signals(self):
48 return self.ports()
49
50 def shape(self):
51 bits, sign = 0, False
52 for elem_bits, elem_sign in (elem.shape() for elem in self.ports()):
53 bits = max(bits, elem_bits + elem_sign)
54 sign = max(sign, elem_sign)
55 return bits, sign
56
57 def eq(self, i):
58 return [self.mid.eq(i.mid), self.idx.eq(i.idx), self.data.eq(i.data)]
59
60 def ports(self):
61 return [self.mid, self.idx, self.data]
62
63
64 class PassThroughStage:
65 def ispec(self):
66 return PassData()
67 def ospec(self):
68 return self.ispec() # same as ospec
69
70 def process(self, i):
71 return i # pass-through
72
73
74
75 class PassThroughPipe(UnbufferedPipeline):
76 def __init__(self):
77 UnbufferedPipeline.__init__(self, PassThroughStage())
78
79
80 class InputTest:
81 def __init__(self, dut):
82 self.dut = dut
83 self.di = {}
84 self.do = {}
85 self.tlen = 100
86 for mid in range(dut.num_rows):
87 self.di[mid] = {}
88 self.do[mid] = {}
89 for i in range(self.tlen):
90 self.di[mid][i] = randint(0, 255) + (mid<<8)
91 self.do[mid][i] = self.di[mid][i]
92
93 def send(self, mid):
94 for i in range(self.tlen):
95 op2 = self.di[mid][i]
96 rs = dut.p[mid]
97 yield rs.i_valid.eq(1)
98 yield rs.i_data.data.eq(op2)
99 yield rs.i_data.idx.eq(i)
100 yield rs.i_data.mid.eq(mid)
101 yield
102 o_p_ready = yield rs.o_ready
103 while not o_p_ready:
104 yield
105 o_p_ready = yield rs.o_ready
106
107 print ("send", mid, i, hex(op2))
108
109 yield rs.i_valid.eq(0)
110 # wait random period of time before queueing another value
111 for i in range(randint(0, 3)):
112 yield
113
114 yield rs.i_valid.eq(0)
115 yield
116
117 print ("send ended", mid)
118
119 ## wait random period of time before queueing another value
120 #for i in range(randint(0, 3)):
121 # yield
122
123 #send_range = randint(0, 3)
124 #if send_range == 0:
125 # send = True
126 #else:
127 # send = randint(0, send_range) != 0
128
129 def rcv(self, mid):
130 while True:
131 #stall_range = randint(0, 3)
132 #for j in range(randint(1,10)):
133 # stall = randint(0, stall_range) != 0
134 # yield self.dut.n[0].i_ready.eq(stall)
135 # yield
136 n = self.dut.n[mid]
137 yield n.i_ready.eq(1)
138 yield
139 o_n_valid = yield n.o_valid
140 i_n_ready = yield n.i_ready
141 if not o_n_valid or not i_n_ready:
142 continue
143
144 out_mid = yield n.o_data.mid
145 out_i = yield n.o_data.idx
146 out_v = yield n.o_data.data
147
148 print ("recv", out_mid, out_i, hex(out_v))
149
150 # see if this output has occurred already, delete it if it has
151 assert mid == out_mid, "out_mid %d not correct %d" % (out_mid, mid)
152 assert out_i in self.do[mid], "out_i %d not in array %s" % \
153 (out_i, repr(self.do[mid]))
154 assert self.do[mid][out_i] == out_v # pass-through data
155 del self.do[mid][out_i]
156
157 # check if there's any more outputs
158 if len(self.do[mid]) == 0:
159 break
160 print ("recv ended", mid)
161
162
163 class TestPriorityMuxPipe(PriorityUnbufferedPipeline):
164 def __init__(self, num_rows):
165 self.num_rows = num_rows
166 stage = PassThroughStage()
167 PriorityUnbufferedPipeline.__init__(self, stage, p_len=self.num_rows)
168
169 def ports(self):
170 res = []
171 for i in range(len(self.p)):
172 res += [self.p[i].i_valid, self.p[i].o_ready] + \
173 self.p[i].i_data.ports()
174 res += [self.n.i_ready, self.n.o_valid] + \
175 self.n.o_data.ports()
176 return res
177
178
179
180 class OutputTest:
181 def __init__(self, dut):
182 self.dut = dut
183 self.di = []
184 self.do = {}
185 self.tlen = 100
186 for i in range(self.tlen * dut.num_rows):
187 if i < dut.num_rows:
188 mid = i
189 else:
190 mid = randint(0, dut.num_rows-1)
191 data = randint(0, 255) + (mid<<8)
192
193 def send(self):
194 for i in range(self.tlen * dut.num_rows):
195 op2 = self.di[i][0]
196 mid = self.di[i][1]
197 rs = dut.p
198 yield rs.i_valid.eq(1)
199 yield rs.i_data.data.eq(op2)
200 yield rs.i_data.mid.eq(mid)
201 yield
202 o_p_ready = yield rs.o_ready
203 while not o_p_ready:
204 yield
205 o_p_ready = yield rs.o_ready
206
207 print ("send", mid, i, hex(op2))
208
209 yield rs.i_valid.eq(0)
210 # wait random period of time before queueing another value
211 for i in range(randint(0, 3)):
212 yield
213
214 yield rs.i_valid.eq(0)
215
216
217 class TestMuxOutPipe(MuxUnbufferedPipeline):
218 def __init__(self, num_rows):
219 self.num_rows = num_rows
220 stage = PassThroughStage()
221 MuxUnbufferedPipeline.__init__(self, stage, n_len=self.num_rows)
222
223 def ports(self):
224 res = [self.p.i_valid, self.p.o_ready] + \
225 self.p.i_data.ports()
226 for i in range(len(self.n)):
227 res += [self.n[i].i_ready, self.n[i].o_valid] + \
228 self.n[i].o_data.ports()
229 return res
230
231
232 class TestInOutPipe:
233 def __init__(self, num_rows=4):
234 self.num_rows = num_rows
235 self.inpipe = TestPriorityMuxPipe(num_rows)
236 self.pipe1 = PassThroughPipe()
237 self.pipe2 = PassThroughPipe()
238 self.outpipe = TestMuxOutPipe(num_rows)
239
240 self.p = self.inpipe.p
241 self.n = self.outpipe.n
242 self._ports = self.inpipe.ports() + self.outpipe.ports()
243
244 def elaborate(self, platform):
245 m = Module()
246 m.submodules.inpipe = self.inpipe
247 m.submodules.pipe1 = self.pipe1
248 m.submodules.pipe2 = self.pipe2
249 m.submodules.outpipe = self.outpipe
250
251 m.d.comb += self.inpipe.n.connect_to_next(self.pipe1.p)
252 m.d.comb += self.pipe1.connect_to_next(self.pipe2)
253 m.d.comb += self.pipe1.connect_to_next(self.outpipe)
254
255 return m
256
257 def ports(self):
258 return self._ports
259
260
261 if __name__ == '__main__':
262 dut = TestInOutPipe()
263 vl = rtlil.convert(dut, ports=dut.ports())
264 with open("test_inoutmux_pipe.il", "w") as f:
265 f.write(vl)
266 #run_simulation(dut, testbench(dut), vcd_name="test_inputgroup.vcd")
267
268 test = InputTest(dut)
269 run_simulation(dut, [test.rcv(1), test.rcv(0),
270 test.rcv(3), test.rcv(2),
271 test.send(0), test.send(1),
272 test.send(3), test.send(2),
273 ],
274 vcd_name="test_inoutmux_pipe.vcd")
275