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):
  15         self.dut = dut
  16         self.fpkls = fpkls
  17         self.fpop = fpop
  18         self.di = {}
  19         self.do = {}
  20         self.tlen = 10
  21         self.width = width
  22         for muxid in range(dut.num_rows):
  23             self.di[muxid] = {}
  24             self.do[muxid] = []
  25             for i in range(self.tlen):
  26                 op1 = randint(0, (1<<self.width)-1)
  27                 op2 = randint(0, (1<<self.width)-1)
  28                 #op1 = 0x513ba448
  29                 #op2 = 0xfff5c7fe
  30                 #op1 = 0xffcaeefa
  31                 #op2 = 0x3f803262
  32                 #op1 = 0xae430313
  33                 #op2 = 0x901c3214
  34                 #op1 = 0xa4504d7
  35                 #op2 = 0xb4658540 # expect 0x8016147c
  36                 #op1 = 0x40900000
  37                 #op2 = 0x40200000
  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             op1, op2 = self.di[muxid][i]
  45             rs = self.dut.p[muxid]
  46             yield rs.valid_i.eq(1)
  47             yield rs.data_i.a.eq(op1)
  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             fop1 = self.fpkls(op1)
  57             fop2 = self.fpkls(op2)
  58             res = self.fpop(fop1, fop2)
  59             print ("send", muxid, i, hex(op1), hex(op2), hex(res.bits),
  60                            fop1, fop2, res)
  61
  62             yield rs.valid_i.eq(0)
  63             # wait random period of time before queueing another value
  64             for i in range(randint(0, 3)):
  65                 yield
  66
  67         yield rs.valid_i.eq(0)
  68         yield
  69
  70         print ("send ended", muxid)
  71
  72         ## wait random period of time before queueing another value
  73         #for i in range(randint(0, 3)):
  74         #    yield
  75
  76         #send_range = randint(0, 3)
  77         #if send_range == 0:
  78         #    send = True
  79         #else:
  80         #    send = randint(0, send_range) != 0
  81
  82     def rcv(self, muxid):
  83         while True:
  84             #stall_range = randint(0, 3)
  85             #for j in range(randint(1,10)):
  86             #    stall = randint(0, stall_range) != 0
  87             #    yield self.dut.n[0].ready_i.eq(stall)
  88             #    yield
  89             n = self.dut.n[muxid]
  90             yield n.ready_i.eq(1)
  91             yield
  92             o_n_valid = yield n.valid_o
  93             i_n_ready = yield n.ready_i
  94             if not o_n_valid or not i_n_ready:
  95                 continue
  96
  97             out_muxid = yield n.data_o.muxid
  98             out_z = yield n.data_o.z
  99
 100             out_i = 0
 101
 102             print ("recv", out_muxid, hex(out_z), "expected",
 103                         hex(self.do[muxid][out_i] ))
 104
 105             # see if this output has occurred already, delete it if it has
 106             assert muxid == out_muxid, "out_muxid %d not correct %d" % \
 107                                        (out_muxid, muxid)
 108             assert self.do[muxid][out_i] == out_z
 109             del self.do[muxid][out_i]
 110
 111             # check if there's any more outputs
 112             if len(self.do[muxid]) == 0:
 113                 break
 114         print ("recv ended", muxid)
 115
 116
 117 def runfp(dut, width, name, fpkls, fpop):
 118     vl = rtlil.convert(dut, ports=dut.ports())
 119     with open("%s.il" % name, "w") as f:
 120         f.write(vl)
 121
 122     test = InputTest(dut, width, fpkls, fpop)
 123     run_simulation(dut, [test.rcv(1), test.rcv(0),
 124                          test.rcv(3), test.rcv(2),
 125                          test.send(0), test.send(1),
 126                          test.send(3), test.send(2),
 127                         ],
 128                    vcd_name="%s.vcd" % name)
 129