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, ExampleStageCls
   8 from example_buf_pipe import PrevControl, NextControl, BufferedPipeline
   9 from example_buf_pipe import StageChain
  10
  11 from random import randint
  12
  13
  14 def check_o_n_valid(dut, val):
  15     o_n_valid = yield dut.n.o_valid
  16     assert o_n_valid == val
  17
  18
  19 def testbench(dut):
  20     #yield dut.i_p_rst.eq(1)
  21     yield dut.n.i_ready.eq(0)
  22     yield dut.p.o_ready.eq(0)
  23     yield
  24     yield
  25     #yield dut.i_p_rst.eq(0)
  26     yield dut.n.i_ready.eq(1)
  27     yield dut.p.i_data.eq(5)
  28     yield dut.p.i_valid.eq(1)
  29     yield
  30
  31     yield dut.p.i_data.eq(7)
  32     yield from check_o_n_valid(dut, 0) # effects of i_p_valid delayed
  33     yield
  34     yield from check_o_n_valid(dut, 1) # ok *now* i_p_valid effect is felt
  35
  36     yield dut.p.i_data.eq(2)
  37     yield
  38     yield dut.n.i_ready.eq(0) # begin going into "stall" (next stage says ready)
  39     yield dut.p.i_data.eq(9)
  40     yield
  41     yield dut.p.i_valid.eq(0)
  42     yield dut.p.i_data.eq(12)
  43     yield
  44     yield dut.p.i_data.eq(32)
  45     yield dut.n.i_ready.eq(1)
  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, 1) # buffer still needs to output
  50     yield
  51     yield from check_o_n_valid(dut, 0) # buffer outputted, *now* we're done.
  52     yield
  53
  54
  55 def testbench2(dut):
  56     #yield dut.p.i_rst.eq(1)
  57     yield dut.n.i_ready.eq(0)
  58     #yield dut.p.o_ready.eq(0)
  59     yield
  60     yield
  61     #yield dut.p.i_rst.eq(0)
  62     yield dut.n.i_ready.eq(1)
  63     yield dut.p.i_data.eq(5)
  64     yield dut.p.i_valid.eq(1)
  65     yield
  66
  67     yield dut.p.i_data.eq(7)
  68     yield from check_o_n_valid(dut, 0) # effects of i_p_valid delayed 2 clocks
  69     yield
  70     yield from check_o_n_valid(dut, 0) # effects of i_p_valid delayed 2 clocks
  71
  72     yield dut.p.i_data.eq(2)
  73     yield
  74     yield from check_o_n_valid(dut, 1) # ok *now* i_p_valid effect is felt
  75     yield dut.n.i_ready.eq(0) # begin going into "stall" (next stage says ready)
  76     yield dut.p.i_data.eq(9)
  77     yield
  78     yield dut.p.i_valid.eq(0)
  79     yield dut.p.i_data.eq(12)
  80     yield
  81     yield dut.p.i_data.eq(32)
  82     yield dut.n.i_ready.eq(1)
  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, 1) # buffer still needs to output
  89     yield
  90     yield from check_o_n_valid(dut, 0) # buffer outputted, *now* we're done.
  91     yield
  92     yield
  93     yield
  94
  95
  96 class Test3:
  97     def __init__(self, dut, resultfn):
  98         self.dut = dut
  99         self.resultfn = resultfn
 100         self.data = []
 101         for i in range(num_tests):
 102             #data.append(randint(0, 1<<16-1))
 103             self.data.append(i+1)
 104         self.i = 0
 105         self.o = 0
 106
 107     def send(self):
 108         while self.o != len(self.data):
 109             send_range = randint(0, 3)
 110             for j in range(randint(1,10)):
 111                 if send_range == 0:
 112                     send = True
 113                 else:
 114                     send = randint(0, send_range) != 0
 115                 o_p_ready = yield self.dut.p.o_ready
 116                 if not o_p_ready:
 117                     yield
 118                     continue
 119                 if send and self.i != len(self.data):
 120                     yield self.dut.p.i_valid.eq(1)
 121                     yield self.dut.p.i_data.eq(self.data[self.i])
 122                     self.i += 1
 123                 else:
 124                     yield self.dut.p.i_valid.eq(0)
 125                 yield
 126
 127     def rcv(self):
 128         while self.o != len(self.data):
 129             stall_range = randint(0, 3)
 130             for j in range(randint(1,10)):
 131                 stall = randint(0, stall_range) != 0
 132                 yield self.dut.n.i_ready.eq(stall)
 133                 yield
 134                 o_n_valid = yield self.dut.n.o_valid
 135                 i_n_ready = yield self.dut.n.i_ready
 136                 if not o_n_valid or not i_n_ready:
 137                     continue
 138                 o_data = yield self.dut.n.o_data
 139                 self.resultfn(o_data, self.data[self.o], self.i, self.o)
 140                 self.o += 1
 141                 if self.o == len(self.data):
 142                     break
 143
 144 def test3_resultfn(o_data, expected, i, o):
 145     assert o_data == expected + 1, \
 146                 "%d-%d data %x not match %x\n" \
 147                 % (i, o, o_data, expected)
 148
 149 def data_dict():
 150         data = []
 151         for i in range(num_tests):
 152             data.append({'src1': randint(0, 1<<16-1),
 153                          'src2': randint(0, 1<<16-1)})
 154         return data
 155
 156
 157 class Test5:
 158     def __init__(self, dut, resultfn, data=None):
 159         self.dut = dut
 160         self.resultfn = resultfn
 161         if data:
 162             self.data = data
 163         else:
 164             self.data = []
 165             for i in range(num_tests):
 166                 self.data.append((randint(0, 1<<16-1), randint(0, 1<<16-1)))
 167         self.i = 0
 168         self.o = 0
 169
 170     def send(self):
 171         while self.o != len(self.data):
 172             send_range = randint(0, 3)
 173             for j in range(randint(1,10)):
 174                 if send_range == 0:
 175                     send = True
 176                 else:
 177                     send = randint(0, send_range) != 0
 178                 o_p_ready = yield self.dut.p.o_ready
 179                 if not o_p_ready:
 180                     yield
 181                     continue
 182                 if send and self.i != len(self.data):
 183                     yield self.dut.p.i_valid.eq(1)
 184                     for v in self.dut.set_input(self.data[self.i]):
 185                         yield v
 186                     self.i += 1
 187                 else:
 188                     yield self.dut.p.i_valid.eq(0)
 189                 yield
 190
 191     def rcv(self):
 192         while self.o != len(self.data):
 193             stall_range = randint(0, 3)
 194             for j in range(randint(1,10)):
 195                 stall = randint(0, stall_range) != 0
 196                 yield self.dut.n.i_ready.eq(stall)
 197                 yield
 198                 o_n_valid = yield self.dut.n.o_valid
 199                 i_n_ready = yield self.dut.n.i_ready
 200                 if not o_n_valid or not i_n_ready:
 201                     continue
 202                 if isinstance(self.dut.n.o_data, Record):
 203                     o_data = {}
 204                     dod = self.dut.n.o_data
 205                     for k, v in dod.fields.items():
 206                         o_data[k] = yield v
 207                 else:
 208                     o_data = yield self.dut.n.o_data
 209                 self.resultfn(o_data, self.data[self.o], self.i, self.o)
 210                 self.o += 1
 211                 if self.o == len(self.data):
 212                     break
 213
 214 def test5_resultfn(o_data, expected, i, o):
 215     res = expected[0] + expected[1]
 216     assert o_data == res, \
 217                 "%d-%d data %x not match %s\n" \
 218                 % (i, o, o_data, repr(expected))
 219
 220 def testbench4(dut):
 221     data = []
 222     for i in range(num_tests):
 223         #data.append(randint(0, 1<<16-1))
 224         data.append(i+1)
 225     i = 0
 226     o = 0
 227     while True:
 228         stall = randint(0, 3) != 0
 229         send = randint(0, 5) != 0
 230         yield dut.n.i_ready.eq(stall)
 231         o_p_ready = yield dut.p.o_ready
 232         if o_p_ready:
 233             if send and i != len(data):
 234                 yield dut.p.i_valid.eq(1)
 235                 yield dut.p.i_data.eq(data[i])
 236                 i += 1
 237             else:
 238                 yield dut.p.i_valid.eq(0)
 239         yield
 240         o_n_valid = yield dut.n.o_valid
 241         i_n_ready = yield dut.n.i_ready
 242         if o_n_valid and i_n_ready:
 243             o_data = yield dut.n.o_data
 244             assert o_data == data[o] + 2, "%d-%d data %x not match %x\n" \
 245                                         % (i, o, o_data, data[o])
 246             o += 1
 247             if o == len(data):
 248                 break
 249
 250
 251 class ExampleBufPipe2:
 252     """
 253         connect these:  ------|---------------|
 254                               v               v
 255         i_p_valid >>in  pipe1 o_n_valid out>> i_p_valid >>in  pipe2
 256         o_p_ready <<out pipe1 i_n_ready <<in  o_p_ready <<out pipe2
 257         p_i_data  >>in  pipe1 p_i_data  out>> n_o_data  >>in  pipe2
 258     """
 259     def __init__(self):
 260         self.pipe1 = ExampleBufPipe()
 261         self.pipe2 = ExampleBufPipe()
 262
 263         # input
 264         self.p = PrevControl()
 265         self.p.i_data = Signal(32) # >>in - comes in from the PREVIOUS stage
 266
 267         # output
 268         self.n = NextControl()
 269         self.n.o_data = Signal(32) # out>> - goes out to the NEXT stage
 270
 271     def elaborate(self, platform):
 272         m = Module()
 273         m.submodules.pipe1 = self.pipe1
 274         m.submodules.pipe2 = self.pipe2
 275
 276         # connect inter-pipe input/output valid/ready/data
 277         m.d.comb += self.pipe1.connect_to_next(self.pipe2)
 278
 279         # inputs/outputs to the module: pipe1 connections here (LHS)
 280         m.d.comb += self.pipe1.connect_in(self)
 281
 282         # now pipe2 connections (RHS)
 283         m.d.comb += self.pipe2.connect_out(self)
 284
 285         return m
 286
 287
 288 class ExampleBufPipeChain2(BufferedPipeline):
 289     """ connects two stages together as a *single* combinatorial stage.
 290     """
 291     def __init__(self):
 292         stage1 = ExampleStageCls()
 293         stage2 = ExampleStageCls()
 294         combined = StageChain([stage1, stage2])
 295         BufferedPipeline.__init__(self, combined)
 296
 297
 298 def data_chain2():
 299         data = []
 300         for i in range(num_tests):
 301             data.append(randint(0, 1<<16-2))
 302         return data
 303
 304
 305 def test9_resultfn(o_data, expected, i, o):
 306     res = expected + 2
 307     assert o_data == res, \
 308                 "%d-%d data %x not match %s\n" \
 309                 % (i, o, o_data, repr(expected))
 310
 311
 312 class SetLessThan:
 313     def __init__(self, width, signed):
 314         self.src1 = Signal((width, signed))
 315         self.src2 = Signal((width, signed))
 316         self.output = Signal(width)
 317
 318     def elaborate(self, platform):
 319         m = Module()
 320         m.d.comb += self.output.eq(Mux(self.src1 < self.src2, 1, 0))
 321         return m
 322
 323
 324 class LTStage:
 325     def __init__(self):
 326         self.slt = SetLessThan(16, True)
 327
 328     def ispec(self):
 329         return (Signal(16), Signal(16))
 330
 331     def ospec(self):
 332         return Signal(16)
 333
 334     def setup(self, m, i):
 335         self.o = Signal(16)
 336         m.submodules.slt = self.slt
 337         m.d.comb += self.slt.src1.eq(i[0])
 338         m.d.comb += self.slt.src2.eq(i[1])
 339         m.d.comb += self.o.eq(self.slt.output)
 340
 341     def process(self, i):
 342         return self.o
 343
 344
 345 class LTStageDerived(SetLessThan):
 346
 347     def __init__(self):
 348         SetLessThan.__init__(self, 16, True)
 349
 350     def ispec(self):
 351         return (Signal(16), Signal(16))
 352
 353     def ospec(self):
 354         return Signal(16)
 355
 356     def setup(self, m, i):
 357         self.o = Signal(16)
 358         m.submodules.slt = self
 359         m.d.comb += self.src1.eq(i[0])
 360         m.d.comb += self.src2.eq(i[1])
 361         m.d.comb += self.o.eq(self.output)
 362
 363     def process(self, i):
 364         return self.o
 365
 366
 367 class ExampleLTCombPipe(CombPipe):
 368     """ an example of how to use the combinatorial pipeline.
 369     """
 370
 371     def __init__(self):
 372         stage = LTStage()
 373         CombPipe.__init__(self, stage)
 374
 375
 376 class ExampleLTBufferedPipeDerived(CombPipe):
 377     """ an example of how to use the combinatorial pipeline.
 378     """
 379
 380     def __init__(self):
 381         stage = LTStageDerived()
 382         CombPipe.__init__(self, stage)
 383
 384
 385 def test6_resultfn(o_data, expected, i, o):
 386     res = 1 if expected[0] < expected[1] else 0
 387     assert o_data == res, \
 388                 "%d-%d data %x not match %s\n" \
 389                 % (i, o, o_data, repr(expected))
 390
 391
 392 class ExampleAddRecordStage:
 393     """ example use of a Record
 394     """
 395
 396     record_spec = [('src1', 16), ('src2', 16)]
 397     def ispec(self):
 398         """ returns a tuple of input signals which will be the incoming data
 399         """
 400         return Record(self.record_spec)
 401
 402     def ospec(self):
 403         return Record(self.record_spec)
 404
 405     def process(self, i):
 406         """ process the input data (sums the values in the tuple) and returns it
 407         """
 408         return {'src1': i.src1 + 1,
 409                 'src2': i.src2 + 1}
 410
 411
 412 class ExampleAddRecordPipe(CombPipe):
 413     """ an example of how to use the combinatorial pipeline.
 414     """
 415
 416     def __init__(self):
 417         stage = ExampleAddRecordStage()
 418         CombPipe.__init__(self, stage)
 419
 420
 421 def test7_resultfn(o_data, expected, i, o):
 422     res = (expected['src1'] + 1, expected['src2'] + 1)
 423     assert o_data['src1'] == res[0] and o_data['src2'] == res[1], \
 424                 "%d-%d data %s not match %s\n" \
 425                 % (i, o, repr(o_data), repr(expected))
 426
 427
 428 class Example2OpClass:
 429     """ an example of a class used to store 2 operands.
 430         requires an eq function, to conform with the pipeline stage API
 431     """
 432
 433     def __init__(self):
 434         self.op1 = Signal(16)
 435         self.op2 = Signal(16)
 436
 437     def eq(self, i):
 438         return [self.op1.eq(i.op1), self.op2.eq(i.op2)]
 439
 440
 441 class ExampleAddClassStage:
 442     """ an example of how to use the buffered pipeline, as a class instance
 443     """
 444
 445     def ispec(self):
 446         """ returns an instance of an Example2OpClass.
 447         """
 448         return Example2OpClass()
 449
 450     def ospec(self):
 451         """ returns an output signal which will happen to contain the sum
 452             of the two inputs
 453         """
 454         return Signal(16)
 455
 456     def process(self, i):
 457         """ process the input data (sums the values in the tuple) and returns it
 458         """
 459         return i.op1 + i.op2
 460
 461
 462 class ExampleBufPipeAddClass(BufferedPipeline):
 463     """ an example of how to use the buffered pipeline, using a class instance
 464     """
 465
 466     def __init__(self):
 467         addstage = ExampleAddClassStage()
 468         BufferedPipeline.__init__(self, addstage)
 469
 470
 471 class TestInputAdd:
 472     """ the eq function, called by set_input, needs an incoming object
 473         that conforms to the Example2OpClass.eq function requirements
 474         easiest way to do that is to create a class that has the exact
 475         same member layout (self.op1, self.op2) as Example2OpClass
 476     """
 477     def __init__(self, op1, op2):
 478         self.op1 = op1
 479         self.op2 = op2
 480
 481
 482 def test8_resultfn(o_data, expected, i, o):
 483     res = expected.op1 + expected.op2 # these are a TestInputAdd instance
 484     assert o_data == res, \
 485                 "%d-%d data %x not match %s\n" \
 486                 % (i, o, o_data, repr(expected))
 487
 488 def data_2op():
 489         data = []
 490         for i in range(num_tests):
 491             data.append(TestInputAdd(randint(0, 1<<16-1), randint(0, 1<<16-1)))
 492         return data
 493
 494
 495 num_tests = 100
 496
 497 if __name__ == '__main__':
 498     print ("test 1")
 499     dut = ExampleBufPipe()
 500     run_simulation(dut, testbench(dut), vcd_name="test_bufpipe.vcd")
 501
 502     print ("test 2")
 503     dut = ExampleBufPipe2()
 504     run_simulation(dut, testbench2(dut), vcd_name="test_bufpipe2.vcd")
 505
 506     print ("test 3")
 507     dut = ExampleBufPipe()
 508     test = Test3(dut, test3_resultfn)
 509     run_simulation(dut, [test.send, test.rcv], vcd_name="test_bufpipe3.vcd")
 510
 511     print ("test 3.5")
 512     dut = ExampleCombPipe()
 513     test = Test3(dut, test3_resultfn)
 514     run_simulation(dut, [test.send, test.rcv], vcd_name="test_combpipe3.vcd")
 515
 516     print ("test 4")
 517     dut = ExampleBufPipe2()
 518     run_simulation(dut, testbench4(dut), vcd_name="test_bufpipe4.vcd")
 519
 520     print ("test 5")
 521     dut = ExampleBufPipeAdd()
 522     test = Test5(dut, test5_resultfn)
 523     run_simulation(dut, [test.send, test.rcv], vcd_name="test_bufpipe5.vcd")
 524
 525     print ("test 6")
 526     dut = ExampleLTCombPipe()
 527     test = Test5(dut, test6_resultfn)
 528     run_simulation(dut, [test.send, test.rcv], vcd_name="test_ltcomb6.vcd")
 529
 530     ports = [dut.p.i_valid, dut.n.i_ready,
 531              dut.n.o_valid, dut.p.o_ready] + \
 532              list(dut.p.i_data) + [dut.n.o_data]
 533     vl = rtlil.convert(dut, ports=ports)
 534     with open("test_ltcomb_pipe.il", "w") as f:
 535         f.write(vl)
 536
 537     print ("test 7")
 538     dut = ExampleAddRecordPipe()
 539     data=data_dict()
 540     test = Test5(dut, test7_resultfn, data=data)
 541     run_simulation(dut, [test.send, test.rcv], vcd_name="test_addrecord.vcd")
 542
 543     ports = [dut.p.i_valid, dut.n.i_ready,
 544              dut.n.o_valid, dut.p.o_ready,
 545              dut.p.i_data.src1, dut.p.i_data.src2,
 546              dut.n.o_data.src1, dut.n.o_data.src2]
 547     vl = rtlil.convert(dut, ports=ports)
 548     with open("test_recordcomb_pipe.il", "w") as f:
 549         f.write(vl)
 550
 551     print ("test 8")
 552     dut = ExampleBufPipeAddClass()
 553     data=data_2op()
 554     test = Test5(dut, test8_resultfn, data=data)
 555     run_simulation(dut, [test.send, test.rcv], vcd_name="test_bufpipe8.vcd")
 556
 557     print ("test 9")
 558     dut = ExampleBufPipeChain2()
 559     ports = [dut.p.i_valid, dut.n.i_ready,
 560              dut.n.o_valid, dut.p.o_ready] + \
 561              [dut.p.i_data] + [dut.n.o_data]
 562     vl = rtlil.convert(dut, ports=ports)
 563     with open("test_bufpipechain2.il", "w") as f:
 564         f.write(vl)
 565
 566     data = data_chain2()
 567     test = Test5(dut, test9_resultfn, data=data)
 568     run_simulation(dut, [test.send, test.rcv],
 569                         vcd_name="test_bufpipechain2.vcd")
 570
 571     print ("test 10")
 572     dut = ExampleLTBufferedPipeDerived()
 573     test = Test5(dut, test6_resultfn)
 574     run_simulation(dut, [test.send, test.rcv], vcd_name="test_ltbufpipe10.vcd")
 575