src/ieee754/fpcommon/test/fpmux.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 nmigen.compat.sim import run_simulation
  10 from nmigen.cli import verilog, rtlil
  11
  12
  13 class InputTest:
  14     def __init__(self, dut, width, fpkls, fpop, vals, single_op):
  15         self.dut = dut
  16         self.fpkls = fpkls
  17         self.fpop = fpop
  18         self.single_op = single_op
  19         self.di = {}
  20         self.do = {}
  21         self.tlen = len(vals) // dut.num_rows
  22         self.width = width
  23         for muxid in range(dut.num_rows):
  24             self.di[muxid] = {}
  25             self.do[muxid] = []
  26             for i in range(self.tlen):
  27                 if self.single_op:
  28                     (op1, ) = vals.pop(0)
  29                     res = self.fpop(self.fpkls(op1))
  30                     self.di[muxid][i] = (op1, )
  31                 else:
  32                     (op1, op2, ) = vals.pop(0)
  33                     print ("test", hex(op1), hex(op2))
  34                     res = self.fpop(self.fpkls(op1), self.fpkls(op2))
  35                     self.di[muxid][i] = (op1, op2)
  36                 self.do[muxid].append(res.bits)
  37
  38     def send(self, muxid):
  39         for i in range(self.tlen):
  40             if self.single_op:
  41                 op1, = self.di[muxid][i]
  42             else:
  43                 op1, op2 = self.di[muxid][i]
  44             rs = self.dut.p[muxid]
  45             yield rs.valid_i.eq(1)
  46             yield rs.data_i.a.eq(op1)
  47             if not self.single_op:
  48                 yield rs.data_i.b.eq(op2)
  49             yield rs.data_i.muxid.eq(muxid)
  50             yield
  51             o_p_ready = yield rs.ready_o
  52             while not o_p_ready:
  53                 yield
  54                 o_p_ready = yield rs.ready_o
  55
  56             if self.single_op:
  57                 fop1 = self.fpkls(op1)
  58                 res = self.fpop(fop1)
  59                 print ("send", muxid, i, hex(op1), hex(res.bits),
  60                                fop1, res)
  61             else:
  62                 fop1 = self.fpkls(op1)
  63                 fop2 = self.fpkls(op2)
  64                 res = self.fpop(fop1, fop2)
  65                 print ("send", muxid, i, hex(op1), hex(op2), hex(res.bits),
  66                                fop1, fop2, res)
  67
  68             yield rs.valid_i.eq(0)
  69             # wait random period of time before queueing another value
  70             for i in range(randint(0, 3)):
  71                 yield
  72
  73         yield rs.valid_i.eq(0)
  74         yield
  75
  76         print ("send ended", muxid)
  77
  78         ## wait random period of time before queueing another value
  79         #for i in range(randint(0, 3)):
  80         #    yield
  81
  82         #send_range = randint(0, 3)
  83         #if send_range == 0:
  84         #    send = True
  85         #else:
  86         #    send = randint(0, send_range) != 0
  87
  88     def rcv(self, muxid):
  89         while True:
  90             #stall_range = randint(0, 3)
  91             #for j in range(randint(1,10)):
  92             #    stall = randint(0, stall_range) != 0
  93             #    yield self.dut.n[0].ready_i.eq(stall)
  94             #    yield
  95             n = self.dut.n[muxid]
  96             yield n.ready_i.eq(1)
  97             yield
  98             o_n_valid = yield n.valid_o
  99             i_n_ready = yield n.ready_i
 100             if not o_n_valid or not i_n_ready:
 101                 continue
 102
 103             out_muxid = yield n.data_o.muxid
 104             out_z = yield n.data_o.z
 105
 106             out_i = 0
 107
 108             print ("recv", out_muxid, hex(out_z), "expected",
 109                         hex(self.do[muxid][out_i] ))
 110
 111             # see if this output has occurred already, delete it if it has
 112             assert muxid == out_muxid, "out_muxid %d not correct %d" % \
 113                                        (out_muxid, muxid)
 114             assert self.do[muxid][out_i] == out_z
 115             del self.do[muxid][out_i]
 116
 117             # check if there's any more outputs
 118             if len(self.do[muxid]) == 0:
 119                 break
 120         print ("recv ended", muxid)
 121
 122
 123 def create_random(num_rows, width, single_op=False, n_vals=10):
 124     vals = []
 125     for muxid in range(num_rows):
 126         for i in range(n_vals):
 127             if single_op:
 128                 op1 = randint(0, (1<<width)-1)
 129                 #op1 = 0x40900000
 130                 #op1 = 0x94607b66
 131                 #op1 = 0x889cd8c
 132                 #op1 = 0xe98646d7
 133                 #op1 = 0x3340f2a7
 134                 #op1 = 0xfff13f05
 135                 #op1 = 0x453eb000
 136                 #op1 = 0x3a05de50
 137                 #op1 = 0xc27ff989
 138                 #op1 = 0x41689000
 139                 #op1 = 0xbbc0edec
 140                 #op1 = 0x2EDBE6FF
 141                 #op1 = 0x358637BD
 142                 #op1 = 0x3340f2a7
 143                 #op1 = 0x33D6BF95
 144                 #op1 = 0x9885020648d8c0e8
 145                 vals.append((op1,))
 146             else:
 147                 op1 = randint(0, (1<<width)-1)
 148                 op2 = randint(0, (1<<width)-1)
 149                 vals.append((op1, op2,))
 150     return vals
 151
 152
 153 def repeat(num_rows, vals):
 154     """ bit of a hack: repeats the last value to create a list
 155         that will be accepted by the muxer, all mux lists to be
 156         of equal length
 157     """
 158     vals = list(vals)
 159     n_to_repeat = len(vals) % num_rows
 160     print (vals, vals[-1])
 161     return vals + [vals[-1]] * n_to_repeat
 162
 163
 164 def runfp(dut, width, name, fpkls, fpop, single_op=False, n_vals=10, vals=None):
 165     vl = rtlil.convert(dut, ports=dut.ports())
 166     with open("%s.il" % name, "w") as f:
 167         f.write(vl)
 168
 169     if vals is None:
 170         vals = create_random(dut.num_rows, width, single_op, n_vals)
 171
 172     test = InputTest(dut, width, fpkls, fpop, vals, single_op)
 173     run_simulation(dut, [test.rcv(1), test.rcv(0),
 174                          test.rcv(3), test.rcv(2),
 175                          test.send(0), test.send(1),
 176                          test.send(3), test.send(2),
 177                         ],
 178                    vcd_name="%s.vcd" % name)
 179