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