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                     res = self.fpop(self.fpkls(op1), self.fpkls(op2))
  34                     self.di[muxid][i] = (op1, op2)
  35                 self.do[muxid].append(res.bits)
  36
  37     def send(self, muxid):
  38         for i in range(self.tlen):
  39             if self.single_op:
  40                 op1, = self.di[muxid][i]
  41             else:
  42                 op1, op2 = self.di[muxid][i]
  43             rs = self.dut.p[muxid]
  44             yield rs.valid_i.eq(1)
  45             yield rs.data_i.a.eq(op1)
  46             if not self.single_op:
  47                 yield rs.data_i.b.eq(op2)
  48             yield rs.data_i.muxid.eq(muxid)
  49             yield
  50             o_p_ready = yield rs.ready_o
  51             while not o_p_ready:
  52                 yield
  53                 o_p_ready = yield rs.ready_o
  54
  55             if self.single_op:
  56                 fop1 = self.fpkls(op1)
  57                 res = self.fpop(fop1)
  58                 print ("send", muxid, i, hex(op1), hex(res.bits),
  59                                fop1, res)
  60             else:
  61                 fop1 = self.fpkls(op1)
  62                 fop2 = self.fpkls(op2)
  63                 res = self.fpop(fop1, fop2)
  64                 print ("send", muxid, i, hex(op1), hex(op2), hex(res.bits),
  65                                fop1, fop2, res)
  66
  67             yield rs.valid_i.eq(0)
  68             # wait random period of time before queueing another value
  69             for i in range(randint(0, 3)):
  70                 yield
  71
  72         yield rs.valid_i.eq(0)
  73         yield
  74
  75         print ("send ended", muxid)
  76
  77         ## wait random period of time before queueing another value
  78         #for i in range(randint(0, 3)):
  79         #    yield
  80
  81         #send_range = randint(0, 3)
  82         #if send_range == 0:
  83         #    send = True
  84         #else:
  85         #    send = randint(0, send_range) != 0
  86
  87     def rcv(self, muxid):
  88         while True:
  89             #stall_range = randint(0, 3)
  90             #for j in range(randint(1,10)):
  91             #    stall = randint(0, stall_range) != 0
  92             #    yield self.dut.n[0].ready_i.eq(stall)
  93             #    yield
  94             n = self.dut.n[muxid]
  95             yield n.ready_i.eq(1)
  96             yield
  97             o_n_valid = yield n.valid_o
  98             i_n_ready = yield n.ready_i
  99             if not o_n_valid or not i_n_ready:
 100                 continue
 101
 102             out_muxid = yield n.data_o.muxid
 103             out_z = yield n.data_o.z
 104
 105             out_i = 0
 106
 107             print ("recv", out_muxid, hex(out_z), "expected",
 108                         hex(self.do[muxid][out_i] ))
 109
 110             # see if this output has occurred already, delete it if it has
 111             assert muxid == out_muxid, "out_muxid %d not correct %d" % \
 112                                        (out_muxid, muxid)
 113             assert self.do[muxid][out_i] == out_z
 114             del self.do[muxid][out_i]
 115
 116             # check if there's any more outputs
 117             if len(self.do[muxid]) == 0:
 118                 break
 119         print ("recv ended", muxid)
 120
 121
 122 class InputTestRandom(InputTest):
 123     def __init__(self, dut, width, fpkls, fpop, single_op=False, n_vals=10):
 124         vals = []
 125         for muxid in range(dut.num_rows):
 126             for i in range(n_vals):
 127                 if single_op:
 128                     op1 = randint(0, (1<<width)-1)
 129                     vals.append((op1,))
 130                 else:
 131                     op1 = randint(0, (1<<width)-1)
 132                     op2 = randint(0, (1<<width)-1)
 133                     vals.append((op1, op2,))
 134
 135         InputTest.__init__(self, dut, width, fpkls, fpop, vals, single_op)
 136
 137
 138 def runfp(dut, width, name, fpkls, fpop, single_op=False, n_vals=10):
 139     vl = rtlil.convert(dut, ports=dut.ports())
 140     with open("%s.il" % name, "w") as f:
 141         f.write(vl)
 142
 143     test = InputTestRandom(dut, width, fpkls, fpop, single_op, n_vals)
 144     run_simulation(dut, [test.rcv(1), test.rcv(0),
 145                          test.rcv(3), test.rcv(2),
 146                          test.send(0), test.send(1),
 147                          test.send(3), test.send(2),
 148                         ],
 149                    vcd_name="%s.vcd" % name)
 150