src/add/test_buf_pipe.py

   1 from nmigen import Module, Signal, Mux
   2 from nmigen.compat.sim import run_simulation
   3 from nmigen.cli import verilog, rtlil
   4
   5 from example_buf_pipe import ExampleBufPipe, ExampleBufPipeAdd
   6 from example_buf_pipe import ExampleCombPipe, CombPipe
   7 from example_buf_pipe import PrevControl, NextControl
   8 from random import randint
   9
  10
  11 def check_o_n_valid(dut, val):
  12     o_n_valid = yield dut.n.o_valid
  13     assert o_n_valid == val
  14
  15
  16 def testbench(dut):
  17     #yield dut.i_p_rst.eq(1)
  18     yield dut.n.i_ready.eq(0)
  19     yield dut.p.o_ready.eq(0)
  20     yield
  21     yield
  22     #yield dut.i_p_rst.eq(0)
  23     yield dut.n.i_ready.eq(1)
  24     yield dut.p.i_data.eq(5)
  25     yield dut.p.i_valid.eq(1)
  26     yield
  27
  28     yield dut.p.i_data.eq(7)
  29     yield from check_o_n_valid(dut, 0) # effects of i_p_valid delayed
  30     yield
  31     yield from check_o_n_valid(dut, 1) # ok *now* i_p_valid effect is felt
  32
  33     yield dut.p.i_data.eq(2)
  34     yield
  35     yield dut.n.i_ready.eq(0) # begin going into "stall" (next stage says ready)
  36     yield dut.p.i_data.eq(9)
  37     yield
  38     yield dut.p.i_valid.eq(0)
  39     yield dut.p.i_data.eq(12)
  40     yield
  41     yield dut.p.i_data.eq(32)
  42     yield dut.n.i_ready.eq(1)
  43     yield
  44     yield from check_o_n_valid(dut, 1) # buffer still needs to output
  45     yield
  46     yield from check_o_n_valid(dut, 1) # buffer still needs to output
  47     yield
  48     yield from check_o_n_valid(dut, 0) # buffer outputted, *now* we're done.
  49     yield
  50
  51
  52 def testbench2(dut):
  53     #yield dut.p.i_rst.eq(1)
  54     yield dut.n.i_ready.eq(0)
  55     #yield dut.p.o_ready.eq(0)
  56     yield
  57     yield
  58     #yield dut.p.i_rst.eq(0)
  59     yield dut.n.i_ready.eq(1)
  60     yield dut.p.i_data.eq(5)
  61     yield dut.p.i_valid.eq(1)
  62     yield
  63
  64     yield dut.p.i_data.eq(7)
  65     yield from check_o_n_valid(dut, 0) # effects of i_p_valid delayed 2 clocks
  66     yield
  67     yield from check_o_n_valid(dut, 0) # effects of i_p_valid delayed 2 clocks
  68
  69     yield dut.p.i_data.eq(2)
  70     yield
  71     yield from check_o_n_valid(dut, 1) # ok *now* i_p_valid effect is felt
  72     yield dut.n.i_ready.eq(0) # begin going into "stall" (next stage says ready)
  73     yield dut.p.i_data.eq(9)
  74     yield
  75     yield dut.p.i_valid.eq(0)
  76     yield dut.p.i_data.eq(12)
  77     yield
  78     yield dut.p.i_data.eq(32)
  79     yield dut.n.i_ready.eq(1)
  80     yield
  81     yield from check_o_n_valid(dut, 1) # buffer still needs to output
  82     yield
  83     yield from check_o_n_valid(dut, 1) # buffer still needs to output
  84     yield
  85     yield from check_o_n_valid(dut, 1) # buffer still needs to output
  86     yield
  87     yield from check_o_n_valid(dut, 0) # buffer outputted, *now* we're done.
  88     yield
  89     yield
  90     yield
  91
  92
  93 class Test3:
  94     def __init__(self, dut, resultfn):
  95         self.dut = dut
  96         self.resultfn = resultfn
  97         self.data = []
  98         for i in range(num_tests):
  99             #data.append(randint(0, 1<<16-1))
 100             self.data.append(i+1)
 101         self.i = 0
 102         self.o = 0
 103
 104     def send(self):
 105         while self.o != len(self.data):
 106             send_range = randint(0, 3)
 107             for j in range(randint(1,10)):
 108                 if send_range == 0:
 109                     send = True
 110                 else:
 111                     send = randint(0, send_range) != 0
 112                 o_p_ready = yield self.dut.p.o_ready
 113                 if not o_p_ready:
 114                     yield
 115                     continue
 116                 if send and self.i != len(self.data):
 117                     yield self.dut.p.i_valid.eq(1)
 118                     yield self.dut.p.i_data.eq(self.data[self.i])
 119                     self.i += 1
 120                 else:
 121                     yield self.dut.p.i_valid.eq(0)
 122                 yield
 123
 124     def rcv(self):
 125         while self.o != len(self.data):
 126             stall_range = randint(0, 3)
 127             for j in range(randint(1,10)):
 128                 stall = randint(0, stall_range) != 0
 129                 yield self.dut.n.i_ready.eq(stall)
 130                 yield
 131                 o_n_valid = yield self.dut.n.o_valid
 132                 i_n_ready = yield self.dut.n.i_ready
 133                 if not o_n_valid or not i_n_ready:
 134                     continue
 135                 o_data = yield self.dut.n.o_data
 136                 self.resultfn(o_data, self.data[self.o], self.i, self.o)
 137                 self.o += 1
 138                 if self.o == len(self.data):
 139                     break
 140
 141 def test3_resultfn(o_data, expected, i, o):
 142     assert o_data == expected + 1, \
 143                 "%d-%d data %x not match %x\n" \
 144                 % (i, o, o_data, expected)
 145
 146 class Test5:
 147     def __init__(self, dut, resultfn):
 148         self.dut = dut
 149         self.resultfn = resultfn
 150         self.data = []
 151         for i in range(num_tests):
 152             self.data.append((randint(0, 1<<16-1), randint(0, 1<<16-1)))
 153         self.i = 0
 154         self.o = 0
 155
 156     def send(self):
 157         while self.o != len(self.data):
 158             send_range = randint(0, 3)
 159             for j in range(randint(1,10)):
 160                 if send_range == 0:
 161                     send = True
 162                 else:
 163                     send = randint(0, send_range) != 0
 164                 o_p_ready = yield self.dut.p.o_ready
 165                 if not o_p_ready:
 166                     yield
 167                     continue
 168                 if send and self.i != len(self.data):
 169                     yield self.dut.p.i_valid.eq(1)
 170                     for v in self.dut.set_input(self.data[self.i]):
 171                         yield v
 172                     self.i += 1
 173                 else:
 174                     yield self.dut.p.i_valid.eq(0)
 175                 yield
 176
 177     def rcv(self):
 178         while self.o != len(self.data):
 179             stall_range = randint(0, 3)
 180             for j in range(randint(1,10)):
 181                 stall = randint(0, stall_range) != 0
 182                 yield self.dut.n.i_ready.eq(stall)
 183                 yield
 184                 o_n_valid = yield self.dut.n.o_valid
 185                 i_n_ready = yield self.dut.n.i_ready
 186                 if not o_n_valid or not i_n_ready:
 187                     continue
 188                 o_data = yield self.dut.n.o_data
 189                 self.resultfn(o_data, self.data[self.o], self.i, self.o)
 190                 self.o += 1
 191                 if self.o == len(self.data):
 192                     break
 193
 194 def test5_resultfn(o_data, expected, i, o):
 195     res = expected[0] + expected[1]
 196     assert o_data == res, \
 197                 "%d-%d data %x not match %s\n" \
 198                 % (i, o, o_data, repr(expected))
 199
 200 def testbench4(dut):
 201     data = []
 202     for i in range(num_tests):
 203         #data.append(randint(0, 1<<16-1))
 204         data.append(i+1)
 205     i = 0
 206     o = 0
 207     while True:
 208         stall = randint(0, 3) != 0
 209         send = randint(0, 5) != 0
 210         yield dut.n.i_ready.eq(stall)
 211         o_p_ready = yield dut.p.o_ready
 212         if o_p_ready:
 213             if send and i != len(data):
 214                 yield dut.p.i_valid.eq(1)
 215                 yield dut.p.i_data.eq(data[i])
 216                 i += 1
 217             else:
 218                 yield dut.p.i_valid.eq(0)
 219         yield
 220         o_n_valid = yield dut.n.o_valid
 221         i_n_ready = yield dut.n.i_ready
 222         if o_n_valid and i_n_ready:
 223             o_data = yield dut.n.o_data
 224             assert o_data == data[o] + 2, "%d-%d data %x not match %x\n" \
 225                                         % (i, o, o_data, data[o])
 226             o += 1
 227             if o == len(data):
 228                 break
 229
 230
 231 class ExampleBufPipe2:
 232     """
 233         connect these:  ------|---------------|
 234                               v               v
 235         i_p_valid >>in  pipe1 o_n_valid out>> i_p_valid >>in  pipe2
 236         o_p_ready <<out pipe1 i_n_ready <<in  o_p_ready <<out pipe2
 237         p_i_data  >>in  pipe1 p_i_data  out>> n_o_data  >>in  pipe2
 238     """
 239     def __init__(self):
 240         self.pipe1 = ExampleBufPipe()
 241         self.pipe2 = ExampleBufPipe()
 242
 243         # input
 244         self.p = PrevControl()
 245         self.p.i_data = Signal(32) # >>in - comes in from the PREVIOUS stage
 246
 247         # output
 248         self.n = NextControl()
 249         self.n.o_data = Signal(32) # out>> - goes out to the NEXT stage
 250
 251     def elaborate(self, platform):
 252         m = Module()
 253         m.submodules.pipe1 = self.pipe1
 254         m.submodules.pipe2 = self.pipe2
 255
 256         # connect inter-pipe input/output valid/ready/data
 257         m.d.comb += self.pipe1.connect_to_next(self.pipe2)
 258
 259         # inputs/outputs to the module: pipe1 connections here (LHS)
 260         m.d.comb += self.pipe1.connect_in(self)
 261
 262         # now pipe2 connections (RHS)
 263         m.d.comb += self.pipe2.connect_out(self)
 264
 265         return m
 266
 267 class SetLessThan:
 268     def __init__(self, width, signed):
 269         self.src1 = Signal((width, signed))
 270         self.src2 = Signal((width, signed))
 271         self.output = Signal(width)
 272
 273     def elaborate(self, platform):
 274         m = Module()
 275         m.d.comb += self.output.eq(Mux(self.src1 < self.src2, 1, 0))
 276         return m
 277
 278
 279 class LTStage:
 280     def __init__(self):
 281         self.slt = SetLessThan(16, True)
 282
 283     def ispec(self):
 284         return (Signal(16), Signal(16))
 285
 286     def ospec(self):
 287         return Signal(16)
 288
 289     def setup(self, m, i):
 290         self.o = Signal(16)
 291         m.submodules.slt = self.slt
 292         m.d.comb += self.slt.src1.eq(i[0])
 293         m.d.comb += self.slt.src2.eq(i[1])
 294         m.d.comb += self.o.eq(self.slt.output)
 295
 296     def process(self, i):
 297         return self.o
 298
 299
 300 class ExampleLTCombPipe(CombPipe):
 301     """ an example of how to use the combinatorial pipeline.
 302     """
 303
 304     def __init__(self):
 305         stage = LTStage()
 306         CombPipe.__init__(self, stage)
 307
 308
 309 def test6_resultfn(o_data, expected, i, o):
 310     res = 1 if expected[0] < expected[1] else 0
 311     assert o_data == res, \
 312                 "%d-%d data %x not match %s\n" \
 313                 % (i, o, o_data, repr(expected))
 314
 315
 316 num_tests = 1000
 317
 318 if __name__ == '__main__':
 319     print ("test 1")
 320     dut = ExampleBufPipe()
 321     run_simulation(dut, testbench(dut), vcd_name="test_bufpipe.vcd")
 322
 323     print ("test 2")
 324     dut = ExampleBufPipe2()
 325     run_simulation(dut, testbench2(dut), vcd_name="test_bufpipe2.vcd")
 326
 327     print ("test 3")
 328     dut = ExampleBufPipe()
 329     test = Test3(dut, test3_resultfn)
 330     run_simulation(dut, [test.send, test.rcv], vcd_name="test_bufpipe3.vcd")
 331
 332     print ("test 3.5")
 333     dut = ExampleCombPipe()
 334     test = Test3(dut, test3_resultfn)
 335     run_simulation(dut, [test.send, test.rcv], vcd_name="test_combpipe3.vcd")
 336
 337     print ("test 4")
 338     dut = ExampleBufPipe2()
 339     run_simulation(dut, testbench4(dut), vcd_name="test_bufpipe4.vcd")
 340
 341     print ("test 5")
 342     dut = ExampleBufPipeAdd()
 343     test = Test5(dut, test5_resultfn)
 344     run_simulation(dut, [test.send, test.rcv], vcd_name="test_bufpipe5.vcd")
 345
 346     print ("test 6")
 347     dut = ExampleLTCombPipe()
 348     test = Test5(dut, test6_resultfn)
 349     run_simulation(dut, [test.send, test.rcv], vcd_name="test_ltcomb6.vcd")
 350
 351     ports = [dut.p.i_valid, dut.n.i_ready,
 352              dut.n.o_valid, dut.p.o_ready] + \
 353              list(dut.p.i_data) + [dut.n.o_data]
 354     vl = rtlil.convert(dut, ports=ports)
 355     with open("test_ltcomb_pipe.il", "w") as f:
 356         f.write(vl)
 357