src/nmutil/test/test_outmux_pipe.py

   1 from random import randint
   2 from math import log
   3 from nmigen import Module, Signal, Cat, Elaboratable
   4 from nmigen.compat.sim import run_simulation
   5 from nmigen.cli import verilog, rtlil
   6
   7 from nmutil.multipipe import CombMuxOutPipe
   8 from nmutil.singlepipe import SimpleHandshake, PassThroughHandshake, RecordObject
   9
  10
  11 class PassInData(RecordObject):
  12     def __init__(self):
  13         RecordObject.__init__(self)
  14         self.muxid = Signal(2, reset_less=True)
  15         self.data = Signal(16, reset_less=True)
  16
  17
  18 class PassThroughStage:
  19
  20     def ispec(self):
  21         return PassInData()
  22
  23     def ospec(self, name):
  24         return Signal(16, name="%s_dout" % name, reset_less=True)
  25
  26     def process(self, i):
  27         return i.data
  28
  29
  30 class PassThroughDataStage:
  31     def ispec(self):
  32         return PassInData()
  33     def ospec(self):
  34         return self.ispec() # same as ospec
  35
  36     def process(self, i):
  37         return i # pass-through
  38
  39
  40
  41 class PassThroughPipe(PassThroughHandshake):
  42     def __init__(self):
  43         PassThroughHandshake.__init__(self, PassThroughDataStage())
  44
  45
  46 class OutputTest:
  47     def __init__(self, dut):
  48         self.dut = dut
  49         self.di = []
  50         self.do = {}
  51         self.tlen = 10
  52         for i in range(self.tlen * dut.num_rows):
  53             if i < dut.num_rows:
  54                 muxid = i
  55             else:
  56                 muxid = randint(0, dut.num_rows-1)
  57             data = randint(0, 255) + (muxid<<8)
  58             if muxid not in self.do:
  59                 self.do[muxid] = []
  60             self.di.append((data, muxid))
  61             self.do[muxid].append(data)
  62
  63     def send(self):
  64         for i in range(self.tlen * self.dut.num_rows):
  65             op2 = self.di[i][0]
  66             muxid = self.di[i][1]
  67             rs = self.dut.p
  68             yield rs.i_valid.eq(1)
  69             yield rs.i_data.data.eq(op2)
  70             yield rs.i_data.muxid.eq(muxid)
  71             yield
  72             o_p_ready = yield rs.o_ready
  73             while not o_p_ready:
  74                 yield
  75                 o_p_ready = yield rs.o_ready
  76
  77             print ("send", muxid, i, hex(op2))
  78
  79             yield rs.i_valid.eq(0)
  80             # wait random period of time before queueing another value
  81             for i in range(randint(0, 3)):
  82                 yield
  83
  84         yield rs.i_valid.eq(0)
  85
  86     def rcv(self, muxid):
  87         out_i = 0
  88         count = 0
  89         stall_range = randint(0, 3)
  90         while out_i != len(self.do[muxid]):
  91             count += 1
  92             assert count != 2000, "timeout: too long"
  93             n = self.dut.n[muxid]
  94             yield n.i_ready.eq(1)
  95             yield
  96             o_n_valid = yield n.o_valid
  97             i_n_ready = yield n.i_ready
  98             if not o_n_valid or not i_n_ready:
  99                 continue
 100
 101             out_v = yield n.o_data
 102
 103             print ("recv", muxid, out_i, hex(out_v))
 104
 105             assert self.do[muxid][out_i] == out_v # pass-through data
 106
 107             out_i += 1
 108
 109             if randint(0, 5) == 0:
 110                 stall_range = randint(0, 3)
 111             stall = randint(0, stall_range) != 0
 112             if stall:
 113                 yield n.i_ready.eq(0)
 114                 for i in range(stall_range):
 115                     yield
 116
 117
 118 class TestPriorityMuxPipe(CombMuxOutPipe):
 119     def __init__(self, num_rows):
 120         self.num_rows = num_rows
 121         stage = PassThroughStage()
 122         CombMuxOutPipe.__init__(self, stage, n_len=self.num_rows)
 123
 124
 125 class TestSyncToPriorityPipe(Elaboratable):
 126     def __init__(self):
 127         self.num_rows = 4
 128         self.pipe = PassThroughPipe()
 129         self.muxpipe = TestPriorityMuxPipe(self.num_rows)
 130
 131         self.p = self.pipe.p
 132         self.n = self.muxpipe.n
 133
 134     def elaborate(self, platform):
 135         m = Module()
 136         m.submodules.pipe = self.pipe
 137         m.submodules.muxpipe = self.muxpipe
 138         m.d.comb += self.pipe.n.connect_to_next(self.muxpipe.p)
 139         return m
 140
 141     def ports(self):
 142         res = [self.p.i_valid, self.p.o_ready] + \
 143                 self.p.i_data.ports()
 144         for i in range(len(self.n)):
 145             res += [self.n[i].i_ready, self.n[i].o_valid] + \
 146                     [self.n[i].o_data]
 147                     #self.n[i].o_data.ports()
 148         return res
 149
 150
 151 def test1():
 152     dut = TestSyncToPriorityPipe()
 153     vl = rtlil.convert(dut, ports=dut.ports())
 154     with open("test_outmux_pipe.il", "w") as f:
 155         f.write(vl)
 156
 157     test = OutputTest(dut)
 158     run_simulation(dut, [test.rcv(1), test.rcv(0),
 159                          test.rcv(3), test.rcv(2),
 160                          test.send()],
 161                    vcd_name="test_outmux_pipe.vcd")
 162
 163 if __name__ == '__main__':
 164     test1()