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 def testbench3(dut):
  90     data = []
  91     for i in range(1000):
  92         #data.append(randint(0, 1<<16-1))
  93         data.append(i+1)
  94     i = 0
  95     o = 0
  96     while True:
  97         stall = randint(0, 3) == 0
  98         send = randint(0, 5) != 0
  99         yield dut.i_n_busy.eq(stall)
 100         o_p_busy = yield dut.o_p_busy
 101         if not o_p_busy:
 102             if send and i != len(data):
 103                 yield dut.i_p_stb.eq(1)
 104                 yield dut.stage.i_data.eq(data[i])
 105                 i += 1
 106             else:
 107                 yield dut.i_p_stb.eq(0)
 108         yield
 109         o_n_stb = yield dut.o_n_stb
 110         i_n_busy = yield dut.i_n_busy
 111         if o_n_stb and not i_n_busy:
 112             o_data = yield dut.stage.o_data
 113             assert o_data == data[o] + 1, "%d-%d data %x not match %x\n" \
 114                                         % (i, o, o_data, data[o])
 115             o += 1
 116             if o == len(data):
 117                 break
 118
 119
 120 def testbench4(dut):
 121     data = []
 122     for i in range(1000):
 123         #data.append(randint(0, 1<<16-1))
 124         data.append(i+1)
 125     i = 0
 126     o = 0
 127     while True:
 128         stall = randint(0, 3) == 0
 129         send = randint(0, 5) != 0
 130         yield dut.i_n_busy.eq(stall)
 131         o_p_busy = yield dut.o_p_busy
 132         if not o_p_busy:
 133             if send and i != len(data):
 134                 yield dut.i_p_stb.eq(1)
 135                 yield dut.i_data.eq(data[i])
 136                 i += 1
 137             else:
 138                 yield dut.i_p_stb.eq(0)
 139         yield
 140         o_n_stb = yield dut.o_n_stb
 141         i_n_busy = yield dut.i_n_busy
 142         if o_n_stb and not i_n_busy:
 143             o_data = yield dut.o_data
 144             assert o_data == data[o] + 2, "%d-%d data %x not match %x\n" \
 145                                         % (i, o, o_data, data[o])
 146             o += 1
 147             if o == len(data):
 148                 break
 149
 150
 151 class BufPipe2:
 152     """
 153         connect these:  ------|---------------|
 154                               v               v
 155         i_p_stb  >>in  pipe1 o_n_stb  out>> i_p_stb  >>in  pipe2
 156         o_p_busy <<out pipe1 i_n_busy <<in  o_p_busy <<out pipe2
 157         stage.i_data   >>in  pipe1 o_data   out>> stage.i_data   >>in  pipe2
 158     """
 159     def __init__(self):
 160         self.pipe1 = BufPipe()
 161         self.pipe2 = BufPipe()
 162
 163         # input
 164         self.i_p_stb = Signal()    # >>in - comes in from PREVIOUS stage
 165         self.i_n_busy = Signal()   # in<< - comes in from the NEXT stage
 166         self.i_data = Signal(32) # >>in - comes in from the PREVIOUS stage
 167
 168         # output
 169         self.o_n_stb = Signal()    # out>> - goes out to the NEXT stage
 170         self.o_p_busy = Signal()   # <<out - goes out to the PREVIOUS stage
 171         self.o_data = Signal(32) # out>> - goes out to the NEXT stage
 172
 173     def elaborate(self, platform):
 174         m = Module()
 175         m.submodules.pipe1 = self.pipe1
 176         m.submodules.pipe2 = self.pipe2
 177
 178         # connect inter-pipe input/output stb/busy/data
 179         m.d.comb += self.pipe2.i_p_stb.eq(self.pipe1.o_n_stb)
 180         m.d.comb += self.pipe1.i_n_busy.eq(self.pipe2.o_p_busy)
 181         m.d.comb += self.pipe2.stage.i_data.eq(self.pipe1.stage.o_data)
 182
 183         # inputs/outputs to the module: pipe1 connections here (LHS)
 184         m.d.comb += self.pipe1.i_p_stb.eq(self.i_p_stb)
 185         m.d.comb += self.o_p_busy.eq(self.pipe1.o_p_busy)
 186         m.d.comb += self.pipe1.stage.i_data.eq(self.i_data)
 187
 188         # now pipe2 connections (RHS)
 189         m.d.comb += self.o_n_stb.eq(self.pipe2.o_n_stb)
 190         m.d.comb += self.pipe2.i_n_busy.eq(self.i_n_busy)
 191         m.d.comb += self.o_data.eq(self.pipe2.stage.o_data)
 192
 193         return m
 194
 195 if __name__ == '__main__':
 196     dut = BufPipe()
 197     run_simulation(dut, testbench(dut), vcd_name="test_bufpipe.vcd")
 198
 199     dut = BufPipe2()
 200     run_simulation(dut, testbench2(dut), vcd_name="test_bufpipe2.vcd")
 201
 202     dut = BufPipe()
 203     run_simulation(dut, testbench3(dut), vcd_name="test_bufpipe3.vcd")
 204
 205     dut = BufPipe2()
 206     run_simulation(dut, testbench4(dut), vcd_name="test_bufpipe4.vcd")