src/add/test_buf_pipe.py

   1 from nmigen import Module, Signal
   2 from nmigen.compat.sim import run_simulation
   3 from example_buf_pipe import BufPipe
   4 from random import randint
   5
   6
   7 def check_o_n_stb(dut, val):
   8     o_n_stb = yield dut.o.n_stb
   9     assert o_n_stb == val
  10
  11 def check_o_n_stb2(dut, val):
  12     o_n_stb = yield dut.o_n_stb
  13     assert o_n_stb == val
  14
  15
  16 def testbench(dut):
  17     #yield dut.i_p_rst.eq(1)
  18     yield dut.i.n_busy.eq(1)
  19     yield dut.o.p_busy.eq(1)
  20     yield
  21     yield
  22     #yield dut.i_p_rst.eq(0)
  23     yield dut.i.n_busy.eq(0)
  24     yield dut.stage.i_data.eq(5)
  25     yield dut.i.p_stb.eq(1)
  26     yield
  27
  28     yield dut.stage.i_data.eq(7)
  29     yield from check_o_n_stb(dut, 0) # effects of i_p_stb delayed
  30     yield
  31     yield from check_o_n_stb(dut, 1) # ok *now* i_p_stb effect is felt
  32
  33     yield dut.stage.i_data.eq(2)
  34     yield
  35     yield dut.i.n_busy.eq(1) # begin going into "stall" (next stage says busy)
  36     yield dut.stage.i_data.eq(9)
  37     yield
  38     yield dut.i.p_stb.eq(0)
  39     yield dut.stage.i_data.eq(12)
  40     yield
  41     yield dut.stage.i_data.eq(32)
  42     yield dut.i.n_busy.eq(0)
  43     yield
  44     yield from check_o_n_stb(dut, 1) # buffer still needs to output
  45     yield
  46     yield from check_o_n_stb(dut, 1) # buffer still needs to output
  47     yield
  48     yield from check_o_n_stb(dut, 0) # buffer outputted, *now* we're done.
  49     yield
  50
  51
  52 def testbench2(dut):
  53     #yield dut.i.p_rst.eq(1)
  54     yield dut.i_n_busy.eq(1)
  55     #yield dut.o.p_busy.eq(1)
  56     yield
  57     yield
  58     #yield dut.i.p_rst.eq(0)
  59     yield dut.i_n_busy.eq(0)
  60     yield dut.i_data.eq(5)
  61     yield dut.i_p_stb.eq(1)
  62     yield
  63
  64     yield dut.i_data.eq(7)
  65     yield from check_o_n_stb2(dut, 0) # effects of i_p_stb delayed 2 clocks
  66     yield
  67     yield from check_o_n_stb2(dut, 0) # effects of i_p_stb delayed 2 clocks
  68
  69     yield dut.i_data.eq(2)
  70     yield
  71     yield from check_o_n_stb2(dut, 1) # ok *now* i_p_stb effect is felt
  72     yield dut.i_n_busy.eq(1) # begin going into "stall" (next stage says busy)
  73     yield dut.i_data.eq(9)
  74     yield
  75     yield dut.i_p_stb.eq(0)
  76     yield dut.i_data.eq(12)
  77     yield
  78     yield dut.i_data.eq(32)
  79     yield dut.i_n_busy.eq(0)
  80     yield
  81     yield from check_o_n_stb2(dut, 1) # buffer still needs to output
  82     yield
  83     yield from check_o_n_stb2(dut, 1) # buffer still needs to output
  84     yield
  85     yield from check_o_n_stb2(dut, 1) # buffer still needs to output
  86     yield
  87     yield from check_o_n_stb2(dut, 0) # buffer outputted, *now* we're done.
  88     yield
  89     yield
  90     yield
  91
  92
  93 class Test3:
  94     def __init__(self, dut):
  95         self.dut = dut
  96         self.data = []
  97         for i in range(10000):
  98             #data.append(randint(0, 1<<16-1))
  99             self.data.append(i+1)
 100         self.i = 0
 101         self.o = 0
 102
 103     def send(self):
 104         while self.o != len(self.data):
 105             send_range = randint(0, 3)
 106             for j in range(randint(1,10)):
 107                 if send_range == 0:
 108                     send = True
 109                 else:
 110                     send = randint(0, send_range) != 0
 111                 o_p_busy = yield self.dut.o.p_busy
 112                 if o_p_busy:
 113                     yield
 114                     continue
 115                 if send and self.i != len(self.data):
 116                     yield self.dut.i.p_stb.eq(1)
 117                     yield self.dut.stage.i_data.eq(self.data[self.i])
 118                     self.i += 1
 119                 else:
 120                     yield self.dut.i.p_stb.eq(0)
 121                 yield
 122
 123     def rcv(self):
 124         while self.o != len(self.data):
 125             stall_range = randint(0, 3)
 126             for j in range(randint(1,10)):
 127                 stall = randint(0, stall_range) == 0
 128                 yield self.dut.i.n_busy.eq(stall)
 129                 yield
 130                 o_n_stb = yield self.dut.o.n_stb
 131                 i_n_busy = yield self.dut.i.n_busy
 132                 if not o_n_stb or i_n_busy:
 133                     continue
 134                 o_data = yield self.dut.stage.o_data
 135                 assert o_data == self.data[self.o] + 1, \
 136                             "%d-%d data %x not match %x\n" \
 137                             % (self.i, self.o, o_data, self.data[self.o])
 138                 self.o += 1
 139                 if self.o == len(self.data):
 140                     break
 141
 142
 143 def testbench4(dut):
 144     data = []
 145     for i in range(10000):
 146         #data.append(randint(0, 1<<16-1))
 147         data.append(i+1)
 148     i = 0
 149     o = 0
 150     while True:
 151         stall = randint(0, 3) == 0
 152         send = randint(0, 5) != 0
 153         yield dut.i_n_busy.eq(stall)
 154         o_p_busy = yield dut.o_p_busy
 155         if not o_p_busy:
 156             if send and i != len(data):
 157                 yield dut.i_p_stb.eq(1)
 158                 yield dut.i_data.eq(data[i])
 159                 i += 1
 160             else:
 161                 yield dut.i_p_stb.eq(0)
 162         yield
 163         o_n_stb = yield dut.o_n_stb
 164         i_n_busy = yield dut.i_n_busy
 165         if o_n_stb and not i_n_busy:
 166             o_data = yield dut.o_data
 167             assert o_data == data[o] + 2, "%d-%d data %x not match %x\n" \
 168                                         % (i, o, o_data, data[o])
 169             o += 1
 170             if o == len(data):
 171                 break
 172
 173
 174 class BufPipe2:
 175     """
 176         connect these:  ------|---------------|
 177                               v               v
 178         i_p_stb  >>in  pipe1 o_n_stb  out>> i_p_stb  >>in  pipe2
 179         o_p_busy <<out pipe1 i_n_busy <<in  o_p_busy <<out pipe2
 180         stage.i_data   >>in  pipe1 o_data   out>> stage.i_data   >>in  pipe2
 181     """
 182     def __init__(self):
 183         self.pipe1 = BufPipe()
 184         self.pipe2 = BufPipe()
 185
 186         # input
 187         self.i_p_stb = Signal()    # >>in - comes in from PREVIOUS stage
 188         self.i_n_busy = Signal()   # in<< - comes in from the NEXT stage
 189         self.i_data = Signal(32) # >>in - comes in from the PREVIOUS stage
 190
 191         # output
 192         self.o_n_stb = Signal()    # out>> - goes out to the NEXT stage
 193         self.o_p_busy = Signal()   # <<out - goes out to the PREVIOUS stage
 194         self.o_data = Signal(32) # out>> - goes out to the NEXT stage
 195
 196     def elaborate(self, platform):
 197         m = Module()
 198         m.submodules.pipe1 = self.pipe1
 199         m.submodules.pipe2 = self.pipe2
 200
 201         # connect inter-pipe input/output stb/busy/data
 202         m.d.comb += self.pipe2.i.p_stb.eq(self.pipe1.o.n_stb)
 203         m.d.comb += self.pipe1.i.n_busy.eq(self.pipe2.o.p_busy)
 204         m.d.comb += self.pipe2.stage.i_data.eq(self.pipe1.stage.o_data)
 205
 206         # inputs/outputs to the module: pipe1 connections here (LHS)
 207         m.d.comb += self.pipe1.i.p_stb.eq(self.i_p_stb)
 208         m.d.comb += self.o_p_busy.eq(self.pipe1.o.p_busy)
 209         m.d.comb += self.pipe1.stage.i_data.eq(self.i_data)
 210
 211         # now pipe2 connections (RHS)
 212         m.d.comb += self.o_n_stb.eq(self.pipe2.o.n_stb)
 213         m.d.comb += self.pipe2.i.n_busy.eq(self.i_n_busy)
 214         m.d.comb += self.o_data.eq(self.pipe2.stage.o_data)
 215
 216         return m
 217
 218 if __name__ == '__main__':
 219     print ("test 1")
 220     dut = BufPipe()
 221     run_simulation(dut, testbench(dut), vcd_name="test_bufpipe.vcd")
 222
 223     print ("test 2")
 224     dut = BufPipe2()
 225     run_simulation(dut, testbench2(dut), vcd_name="test_bufpipe2.vcd")
 226
 227     print ("test 3")
 228     dut = BufPipe()
 229     test = Test3(dut)
 230     run_simulation(dut, [test.send, test.rcv], vcd_name="test_bufpipe3.vcd")
 231
 232     print ("test 4")
 233     dut = BufPipe2()
 234     run_simulation(dut, testbench4(dut), vcd_name="test_bufpipe4.vcd")