likewise replace data_o with o_data and data_i with i_data
[nmutil.git] / src / nmutil / test / 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, Elaboratable
11 from nmigen.compat.sim import run_simulation
12 from nmigen.cli import verilog, rtlil
13
14 from nmutil.multipipe import CombMultiOutPipeline, CombMuxOutPipe
15 from nmutil.multipipe import PriorityCombMuxInPipe
16 from nmutil.singlepipe import SimpleHandshake, RecordObject, Object
17
18
19 class PassData2(RecordObject):
20 def __init__(self):
21 RecordObject.__init__(self)
22 self.muxid = Signal(2, reset_less=True)
23 self.idx = Signal(8, reset_less=True)
24 self.data = Signal(16, reset_less=True)
25
26
27 class PassData(Object):
28 def __init__(self):
29 Object.__init__(self)
30 self.muxid = Signal(2, reset_less=True)
31 self.idx = Signal(8, reset_less=True)
32 self.data = Signal(16, reset_less=True)
33
34
35
36 class PassThroughStage:
37 def ispec(self):
38 return PassData()
39 def ospec(self):
40 return self.ispec() # same as ospec
41
42 def process(self, i):
43 return i # pass-through
44
45
46
47 class PassThroughPipe(SimpleHandshake):
48 def __init__(self):
49 SimpleHandshake.__init__(self, PassThroughStage())
50
51
52 class InputTest:
53 def __init__(self, dut):
54 self.dut = dut
55 self.di = {}
56 self.do = {}
57 self.tlen = 100
58 for muxid in range(dut.num_rows):
59 self.di[muxid] = {}
60 self.do[muxid] = {}
61 for i in range(self.tlen):
62 self.di[muxid][i] = randint(0, 255) + (muxid<<8)
63 self.do[muxid][i] = self.di[muxid][i]
64
65 def send(self, muxid):
66 for i in range(self.tlen):
67 op2 = self.di[muxid][i]
68 rs = self.dut.p[muxid]
69 yield rs.valid_i.eq(1)
70 yield rs.data_i.data.eq(op2)
71 yield rs.data_i.idx.eq(i)
72 yield rs.data_i.muxid.eq(muxid)
73 yield
74 o_p_ready = yield rs.ready_o
75 while not o_p_ready:
76 yield
77 o_p_ready = yield rs.ready_o
78
79 print ("send", muxid, i, hex(op2))
80
81 yield rs.valid_i.eq(0)
82 # wait random period of time before queueing another value
83 for i in range(randint(0, 3)):
84 yield
85
86 yield rs.valid_i.eq(0)
87 yield
88
89 print ("send ended", muxid)
90
91 ## wait random period of time before queueing another value
92 #for i in range(randint(0, 3)):
93 # yield
94
95 #send_range = randint(0, 3)
96 #if send_range == 0:
97 # send = True
98 #else:
99 # send = randint(0, send_range) != 0
100
101 def rcv(self, muxid):
102 while True:
103 #stall_range = randint(0, 3)
104 #for j in range(randint(1,10)):
105 # stall = randint(0, stall_range) != 0
106 # yield self.dut.n[0].ready_i.eq(stall)
107 # yield
108 n = self.dut.n[muxid]
109 yield n.ready_i.eq(1)
110 yield
111 o_n_valid = yield n.valid_o
112 i_n_ready = yield n.ready_i
113 if not o_n_valid or not i_n_ready:
114 continue
115
116 out_muxid = yield n.data_o.muxid
117 out_i = yield n.data_o.idx
118 out_v = yield n.data_o.data
119
120 print ("recv", out_muxid, out_i, hex(out_v))
121
122 # see if this output has occurred already, delete it if it has
123 assert muxid == out_muxid, \
124 "out_muxid %d not correct %d" % (out_muxid, muxid)
125 assert out_i in self.do[muxid], "out_i %d not in array %s" % \
126 (out_i, repr(self.do[muxid]))
127 assert self.do[muxid][out_i] == out_v # pass-through data
128 del self.do[muxid][out_i]
129
130 # check if there's any more outputs
131 if len(self.do[muxid]) == 0:
132 break
133 print ("recv ended", muxid)
134
135
136 class TestPriorityMuxPipe(PriorityCombMuxInPipe):
137 def __init__(self, num_rows):
138 self.num_rows = num_rows
139 stage = PassThroughStage()
140 PriorityCombMuxInPipe.__init__(self, stage, p_len=self.num_rows)
141
142
143 class OutputTest:
144 def __init__(self, dut):
145 self.dut = dut
146 self.di = []
147 self.do = {}
148 self.tlen = 100
149 for i in range(self.tlen * dut.num_rows):
150 if i < dut.num_rows:
151 muxid = i
152 else:
153 muxid = randint(0, dut.num_rows-1)
154 data = randint(0, 255) + (muxid<<8)
155
156 def send(self):
157 for i in range(self.tlen * dut.num_rows):
158 op2 = self.di[i][0]
159 muxid = self.di[i][1]
160 rs = dut.p
161 yield rs.valid_i.eq(1)
162 yield rs.data_i.data.eq(op2)
163 yield rs.data_i.muxid.eq(muxid)
164 yield
165 o_p_ready = yield rs.ready_o
166 while not o_p_ready:
167 yield
168 o_p_ready = yield rs.ready_o
169
170 print ("send", muxid, i, hex(op2))
171
172 yield rs.valid_i.eq(0)
173 # wait random period of time before queueing another value
174 for i in range(randint(0, 3)):
175 yield
176
177 yield rs.valid_i.eq(0)
178
179
180 class TestMuxOutPipe(CombMuxOutPipe):
181 def __init__(self, num_rows):
182 self.num_rows = num_rows
183 stage = PassThroughStage()
184 CombMuxOutPipe.__init__(self, stage, n_len=self.num_rows)
185
186
187 class TestInOutPipe(Elaboratable):
188 def __init__(self, num_rows=4):
189 self.num_rows = num_rows
190 self.inpipe = TestPriorityMuxPipe(num_rows) # fan-in (combinatorial)
191 self.pipe1 = PassThroughPipe() # stage 1 (clock-sync)
192 self.pipe2 = PassThroughPipe() # stage 2 (clock-sync)
193 self.outpipe = TestMuxOutPipe(num_rows) # fan-out (combinatorial)
194
195 self.p = self.inpipe.p # kinda annoying,
196 self.n = self.outpipe.n # use pipe in/out as this class in/out
197 self._ports = self.inpipe.ports() + self.outpipe.ports()
198
199 def elaborate(self, platform):
200 m = Module()
201 m.submodules.inpipe = self.inpipe
202 m.submodules.pipe1 = self.pipe1
203 m.submodules.pipe2 = self.pipe2
204 m.submodules.outpipe = self.outpipe
205
206 m.d.comb += self.inpipe.n.connect_to_next(self.pipe1.p)
207 m.d.comb += self.pipe1.connect_to_next(self.pipe2)
208 m.d.comb += self.pipe2.connect_to_next(self.outpipe)
209
210 return m
211
212 def ports(self):
213 return self._ports
214
215
216 def test1():
217 dut = TestInOutPipe()
218 vl = rtlil.convert(dut, ports=dut.ports())
219 with open("test_inoutmux_pipe.il", "w") as f:
220 f.write(vl)
221 #run_simulation(dut, testbench(dut), vcd_name="test_inputgroup.vcd")
222
223 test = InputTest(dut)
224 run_simulation(dut, [test.rcv(1), test.rcv(0),
225 test.rcv(3), test.rcv(2),
226 test.send(0), test.send(1),
227 test.send(3), test.send(2),
228 ],
229 vcd_name="test_inoutmux_pipe.vcd")
230
231 if __name__ == '__main__':
232 test1()