src/add/test_buf_pipe.py

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