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