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