src/nmutil/test/test_reservation_stations.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, Elaboratable
  11 from nmigen.compat.sim import run_simulation
  12 from nmigen.cli import verilog, rtlil
  13
  14 from nmutil.concurrentunit import ReservationStations2
  15 from nmutil.singlepipe import SimpleHandshake, RecordObject, Object
  16
  17
  18 class PassData2(RecordObject):
  19     def __init__(self):
  20         RecordObject.__init__(self)
  21         self.muxid = Signal(2, reset_less=True)
  22         self.idx = Signal(8, reset_less=True)
  23         self.data = Signal(16, reset_less=True)
  24
  25
  26 class PassData(Object):
  27     def __init__(self, name=None):
  28         Object.__init__(self)
  29         if name is None:
  30             name = ""
  31         self.muxid = Signal(2, name="muxid"+name, reset_less=True)
  32         self.idx = Signal(8, name="idx"+name, reset_less=True)
  33         self.data = Signal(16, name="data"+name, reset_less=True)
  34
  35
  36
  37 class PassThroughStage:
  38     def ispec(self, name=None):
  39         return PassData(name=name)
  40     def ospec(self, name=None):
  41         return self.ispec(name) # same as ospec
  42
  43     def process(self, i):
  44         return i # pass-through
  45
  46
  47
  48 class PassThroughPipe(SimpleHandshake):
  49     def __init__(self):
  50         SimpleHandshake.__init__(self, PassThroughStage())
  51
  52
  53 class InputTest:
  54     def __init__(self, dut):
  55         self.dut = dut
  56         self.di = {}
  57         self.do = {}
  58         self.tlen = 100
  59         for muxid in range(dut.num_rows):
  60             self.di[muxid] = {}
  61             self.do[muxid] = {}
  62             for i in range(self.tlen):
  63                 self.di[muxid][i] = randint(0, 255) + (muxid<<8)
  64                 self.do[muxid][i] = self.di[muxid][i]
  65
  66     def send(self, muxid):
  67         for i in range(self.tlen):
  68             op2 = self.di[muxid][i]
  69             rs = self.dut.p[muxid]
  70             yield rs.i_valid.eq(1)
  71             yield rs.i_data.data.eq(op2)
  72             yield rs.i_data.idx.eq(i)
  73             yield rs.i_data.muxid.eq(muxid)
  74             yield
  75             o_p_ready = yield rs.o_ready
  76             while not o_p_ready:
  77                 yield
  78                 o_p_ready = yield rs.o_ready
  79
  80             print ("send", muxid, i, hex(op2))
  81
  82             yield rs.i_valid.eq(0)
  83             # wait random period of time before queueing another value
  84             for i in range(randint(0, 3)):
  85                 yield
  86
  87         yield rs.i_valid.eq(0)
  88         yield
  89
  90         print ("send ended", muxid)
  91
  92         ## wait random period of time before queueing another value
  93         #for i in range(randint(0, 3)):
  94         #    yield
  95
  96         #send_range = randint(0, 3)
  97         #if send_range == 0:
  98         #    send = True
  99         #else:
 100         #    send = randint(0, send_range) != 0
 101
 102     def rcv(self, muxid):
 103         while True:
 104             #stall_range = randint(0, 3)
 105             #for j in range(randint(1,10)):
 106             #    stall = randint(0, stall_range) != 0
 107             #    yield self.dut.n[0].i_ready.eq(stall)
 108             #    yield
 109             n = self.dut.n[muxid]
 110             yield n.i_ready.eq(1)
 111             yield
 112             o_n_valid = yield n.o_valid
 113             i_n_ready = yield n.i_ready
 114             if not o_n_valid or not i_n_ready:
 115                 continue
 116
 117             out_muxid = yield n.o_data.muxid
 118             out_i = yield n.o_data.idx
 119             out_v = yield n.o_data.data
 120
 121             print ("recv", out_muxid, out_i, hex(out_v))
 122
 123             # see if this output has occurred already, delete it if it has
 124             assert muxid == out_muxid, \
 125                     "out_muxid %d not correct %d" % (out_muxid, muxid)
 126             assert out_i in self.do[muxid], "out_i %d not in array %s" % \
 127                                           (out_i, repr(self.do[muxid]))
 128             assert self.do[muxid][out_i] == out_v # pass-through data
 129             del self.do[muxid][out_i]
 130
 131             # check if there's any more outputs
 132             if len(self.do[muxid]) == 0:
 133                 break
 134         print ("recv ended", muxid)
 135
 136
 137 class TestALU(Elaboratable):
 138     def __init__(self):
 139         self.pipe1 = PassThroughPipe()              # stage 1 (clock-sync)
 140         self.pipe2 = PassThroughPipe()              # stage 2 (clock-sync)
 141
 142         self.p = self.pipe1.p
 143         self.n = self.pipe2.n
 144         self._ports = self.pipe1.ports() + self.pipe2.ports()
 145
 146     def elaborate(self, platform):
 147         m = Module()
 148         m.submodules.pipe1 = self.pipe1
 149         m.submodules.pipe2 = self.pipe2
 150
 151         m.d.comb += self.pipe1.connect_to_next(self.pipe2)
 152
 153         return m
 154
 155     def new_specs(self, name):
 156         return self.pipe1.ispec(name), self.pipe2.ospec(name)
 157
 158     def ports(self):
 159         return self._ports
 160
 161
 162 def test1():
 163     alu = TestALU()
 164     dut = ReservationStations2(alu, num_rows=4)
 165     vl = rtlil.convert(dut, ports=dut.ports())
 166     with open("test_reservation_stations.il", "w") as f:
 167         f.write(vl)
 168     #run_simulation(dut, testbench(dut), vcd_name="test_inputgroup.vcd")
 169
 170     test = InputTest(dut)
 171     run_simulation(dut, [test.rcv(1), test.rcv(0),
 172                          test.rcv(3), test.rcv(2),
 173                          test.send(0), test.send(1),
 174                          test.send(3), test.send(2),
 175                         ],
 176                    vcd_name="test_reservation_stations.vcd")
 177
 178 if __name__ == '__main__':
 179     test1()