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