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