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