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