src/nmutil/test/test_prioritymux_pipe.py

   1 from random import randint
   2 from math import log
   3 from nmigen import Module, Signal, Cat
   4 from nmigen.compat.sim import run_simulation
   5 from nmigen.cli import verilog, rtlil
   6
   7 from nmutil.singlepipe import PassThroughStage
   8 from nmutil.multipipe import (CombMultiInPipeline, PriorityCombMuxInPipe)
   9
  10
  11 class PassData:
  12     def __init__(self):
  13         self.muxid = Signal(2, reset_less=True)
  14         self.idx = Signal(6, reset_less=True)
  15         self.data = Signal(16, reset_less=True)
  16
  17     def eq(self, i):
  18         return [self.muxid.eq(i.muxid), self.idx.eq(i.idx), self.data.eq(i.data)]
  19
  20     def ports(self):
  21         return [self.muxid, self.idx, self.data]
  22
  23
  24 def tbench(dut):
  25     stb = yield dut.out_op.stb
  26     assert stb == 0
  27     ack = yield dut.out_op.ack
  28     assert ack == 0
  29
  30     # set row 1 input 0
  31     yield dut.rs[1].in_op[0].eq(5)
  32     yield dut.rs[1].stb.eq(0b01) # strobe indicate 1st op ready
  33     #yield dut.rs[1].ack.eq(1)
  34     yield
  35
  36     # check row 1 output (should be inactive)
  37     decode = yield dut.rs[1].out_decode
  38     assert decode == 0
  39     if False:
  40         op0 = yield dut.rs[1].out_op[0]
  41         op1 = yield dut.rs[1].out_op[1]
  42         assert op0 == 0 and op1 == 0
  43
  44     # output should be inactive
  45     out_stb = yield dut.out_op.stb
  46     assert out_stb == 1
  47
  48     # set row 0 input 1
  49     yield dut.rs[1].in_op[1].eq(6)
  50     yield dut.rs[1].stb.eq(0b11) # strobe indicate both ops ready
  51
  52     # set acknowledgement of output... takes 1 cycle to respond
  53     yield dut.out_op.ack.eq(1)
  54     yield
  55     yield dut.out_op.ack.eq(0) # clear ack on output
  56     yield dut.rs[1].stb.eq(0) # clear row 1 strobe
  57
  58     # output strobe should be active, MID should be 0 until "ack" is set...
  59     out_stb = yield dut.out_op.stb
  60     assert out_stb == 1
  61     out_muxid = yield dut.muxid
  62     assert out_muxid == 0
  63
  64     # ... and output should not yet be passed through either
  65     op0 = yield dut.out_op.v[0]
  66     op1 = yield dut.out_op.v[1]
  67     assert op0 == 0 and op1 == 0
  68
  69     # wait for out_op.ack to activate...
  70     yield dut.rs[1].stb.eq(0b00) # set row 1 strobes to zero
  71     yield
  72
  73     # *now* output should be passed through
  74     op0 = yield dut.out_op.v[0]
  75     op1 = yield dut.out_op.v[1]
  76     assert op0 == 5 and op1 == 6
  77
  78     # set row 2 input
  79     yield dut.rs[2].in_op[0].eq(3)
  80     yield dut.rs[2].in_op[1].eq(4)
  81     yield dut.rs[2].stb.eq(0b11) # strobe indicate 1st op ready
  82     yield dut.out_op.ack.eq(1) # set output ack
  83     yield
  84     yield dut.rs[2].stb.eq(0) # clear row 2 strobe
  85     yield dut.out_op.ack.eq(0) # set output ack
  86     yield
  87     op0 = yield dut.out_op.v[0]
  88     op1 = yield dut.out_op.v[1]
  89     assert op0 == 3 and op1 == 4, "op0 %d op1 %d" % (op0, op1)
  90     out_muxid = yield dut.muxid
  91     assert out_muxid == 2
  92
  93     # set row 0 and 3 input
  94     yield dut.rs[0].in_op[0].eq(9)
  95     yield dut.rs[0].in_op[1].eq(8)
  96     yield dut.rs[0].stb.eq(0b11) # strobe indicate 1st op ready
  97     yield dut.rs[3].in_op[0].eq(1)
  98     yield dut.rs[3].in_op[1].eq(2)
  99     yield dut.rs[3].stb.eq(0b11) # strobe indicate 1st op ready
 100
 101     # set acknowledgement of output... takes 1 cycle to respond
 102     yield dut.out_op.ack.eq(1)
 103     yield
 104     yield dut.rs[0].stb.eq(0) # clear row 1 strobe
 105     yield
 106     out_muxid = yield dut.muxid
 107     assert out_muxid == 0, "out muxid %d" % out_muxid
 108
 109     yield
 110     yield dut.rs[3].stb.eq(0) # clear row 1 strobe
 111     yield dut.out_op.ack.eq(0) # clear ack on output
 112     yield
 113     out_muxid = yield dut.muxid
 114     assert out_muxid == 3, "out muxid %d" % out_muxid
 115
 116
 117 class InputTest:
 118     def __init__(self, dut):
 119         self.dut = dut
 120         self.di = {}
 121         self.do = {}
 122         self.tlen = 10
 123         for muxid in range(dut.num_rows):
 124             self.di[muxid] = {}
 125             self.do[muxid] = {}
 126             for i in range(self.tlen):
 127                 self.di[muxid][i] = randint(0, 100) + (muxid<<8)
 128                 self.do[muxid][i] = self.di[muxid][i]
 129
 130     def send(self, muxid):
 131         for i in range(self.tlen):
 132             op2 = self.di[muxid][i]
 133             rs = self.dut.p[muxid]
 134             yield rs.valid_i.eq(1)
 135             yield rs.data_i.data.eq(op2)
 136             yield rs.data_i.idx.eq(i)
 137             yield rs.data_i.muxid.eq(muxid)
 138             yield
 139             o_p_ready = yield rs.ready_o
 140             while not o_p_ready:
 141                 yield
 142                 o_p_ready = yield rs.ready_o
 143
 144             print ("send", muxid, i, hex(op2))
 145
 146             yield rs.valid_i.eq(0)
 147             # wait random period of time before queueing another value
 148             for i in range(randint(0, 3)):
 149                 yield
 150
 151         yield rs.valid_i.eq(0)
 152         ## wait random period of time before queueing another value
 153         #for i in range(randint(0, 3)):
 154         #    yield
 155
 156         #send_range = randint(0, 3)
 157         #if send_range == 0:
 158         #    send = True
 159         #else:
 160         #    send = randint(0, send_range) != 0
 161
 162     def rcv(self):
 163         while True:
 164             #stall_range = randint(0, 3)
 165             #for j in range(randint(1,10)):
 166             #    stall = randint(0, stall_range) != 0
 167             #    yield self.dut.n[0].ready_i.eq(stall)
 168             #    yield
 169             n = self.dut.n
 170             yield n.ready_i.eq(1)
 171             yield
 172             o_n_valid = yield n.valid_o
 173             i_n_ready = yield n.ready_i
 174             if not o_n_valid or not i_n_ready:
 175                 continue
 176
 177             muxid = yield n.data_o.muxid
 178             out_i = yield n.data_o.idx
 179             out_v = yield n.data_o.data
 180
 181             print ("recv", muxid, out_i, hex(out_v))
 182
 183             # see if this output has occurred already, delete it if it has
 184             assert out_i in self.do[muxid], "out_i %d not in array %s" % \
 185                                           (out_i, repr(self.do[muxid]))
 186             assert self.do[muxid][out_i] == out_v # pass-through data
 187             del self.do[muxid][out_i]
 188
 189             # check if there's any more outputs
 190             zerolen = True
 191             for (k, v) in self.do.items():
 192                 if v:
 193                     zerolen = False
 194             if zerolen:
 195                 break
 196
 197
 198 class TestPriorityMuxPipe(PriorityCombMuxInPipe):
 199     def __init__(self):
 200         self.num_rows = 4
 201         def iospecfn(): return PassData()
 202         stage = PassThroughStage(iospecfn)
 203         PriorityCombMuxInPipe.__init__(self, stage, p_len=self.num_rows)
 204
 205 def test1():
 206     dut = TestPriorityMuxPipe()
 207     vl = rtlil.convert(dut, ports=dut.ports())
 208     with open("test_inputgroup_multi.il", "w") as f:
 209         f.write(vl)
 210     #run_simulation(dut, tbench(dut), vcd_name="test_inputgroup.vcd")
 211
 212     test = InputTest(dut)
 213     run_simulation(dut, [test.send(1), test.send(0),
 214                          test.send(3), test.send(2),
 215                          test.rcv()],
 216                    vcd_name="test_inputgroup_multi.vcd")
 217
 218 if __name__ == '__main__':
 219     test1()