6bf690c24e81703afd1fa11b47f160f3172a7b2f
1 """ Unit tests for Buffered and Unbuffered pipelines
3 contains useful worked examples of how to use the Pipeline API,
6 * Combinatorial Stage "Chaining"
7 * class-based data stages
8 * nmigen module-based data stages
9 * special nmigen module-based data stage, where the stage *is* the module
10 * Record-based data stages
11 * static-class data stages
12 * multi-stage pipelines (and how to connect them)
13 * how to *use* the pipelines (see Test5) - how to get data in and out
17 from nmigen
import Module
, Signal
, Mux
, Const
18 from nmigen
.hdl
.rec
import Record
19 from nmigen
.compat
.sim
import run_simulation
20 from nmigen
.cli
import verilog
, rtlil
22 from example_buf_pipe
import ExampleBufPipe
, ExampleBufPipeAdd
23 from example_buf_pipe
import ExamplePipeline
, UnbufferedPipeline
24 from example_buf_pipe
import ExampleStageCls
25 from example_buf_pipe
import PrevControl
, NextControl
, BufferedHandshake
26 from example_buf_pipe
import StageChain
, ControlBase
, StageCls
27 from singlepipe
import UnbufferedPipeline2
28 from singlepipe
import SimpleHandshake
29 from singlepipe
import PassThroughHandshake
30 from singlepipe
import PassThroughStage
31 from singlepipe
import FIFOtest
33 from random
import randint
, seed
38 def check_o_n_valid(dut
, val
):
39 o_n_valid
= yield dut
.n
.o_valid
40 assert o_n_valid
== val
42 def check_o_n_valid2(dut
, val
):
43 o_n_valid
= yield dut
.n
.o_valid
44 assert o_n_valid
== val
48 #yield dut.i_p_rst.eq(1)
49 yield dut
.n
.i_ready
.eq(0)
50 yield dut
.p
.o_ready
.eq(0)
53 #yield dut.i_p_rst.eq(0)
54 yield dut
.n
.i_ready
.eq(1)
55 yield dut
.p
.i_data
.eq(5)
56 yield dut
.p
.i_valid
.eq(1)
59 yield dut
.p
.i_data
.eq(7)
60 yield from check_o_n_valid(dut
, 0) # effects of i_p_valid delayed
62 yield from check_o_n_valid(dut
, 1) # ok *now* i_p_valid effect is felt
64 yield dut
.p
.i_data
.eq(2)
66 yield dut
.n
.i_ready
.eq(0) # begin going into "stall" (next stage says ready)
67 yield dut
.p
.i_data
.eq(9)
69 yield dut
.p
.i_valid
.eq(0)
70 yield dut
.p
.i_data
.eq(12)
72 yield dut
.p
.i_data
.eq(32)
73 yield dut
.n
.i_ready
.eq(1)
75 yield from check_o_n_valid(dut
, 1) # buffer still needs to output
77 yield from check_o_n_valid(dut
, 1) # buffer still needs to output
79 yield from check_o_n_valid(dut
, 0) # buffer outputted, *now* we're done.
84 #yield dut.p.i_rst.eq(1)
85 yield dut
.n
.i_ready
.eq(0)
86 #yield dut.p.o_ready.eq(0)
89 #yield dut.p.i_rst.eq(0)
90 yield dut
.n
.i_ready
.eq(1)
91 yield dut
.p
.i_data
.eq(5)
92 yield dut
.p
.i_valid
.eq(1)
95 yield dut
.p
.i_data
.eq(7)
96 yield from check_o_n_valid2(dut
, 0) # effects of i_p_valid delayed 2 clocks
98 yield from check_o_n_valid2(dut
, 0) # effects of i_p_valid delayed 2 clocks
100 yield dut
.p
.i_data
.eq(2)
102 yield from check_o_n_valid2(dut
, 1) # ok *now* i_p_valid effect is felt
103 yield dut
.n
.i_ready
.eq(0) # begin going into "stall" (next stage says ready)
104 yield dut
.p
.i_data
.eq(9)
106 yield dut
.p
.i_valid
.eq(0)
107 yield dut
.p
.i_data
.eq(12)
109 yield dut
.p
.i_data
.eq(32)
110 yield dut
.n
.i_ready
.eq(1)
112 yield from check_o_n_valid2(dut
, 1) # buffer still needs to output
114 yield from check_o_n_valid2(dut
, 1) # buffer still needs to output
116 yield from check_o_n_valid2(dut
, 1) # buffer still needs to output
118 yield from check_o_n_valid2(dut
, 0) # buffer outputted, *now* we're done.
125 def __init__(self
, dut
, resultfn
):
127 self
.resultfn
= resultfn
129 for i
in range(num_tests
):
130 #data.append(randint(0, 1<<16-1))
131 self
.data
.append(i
+1)
136 while self
.o
!= len(self
.data
):
137 send_range
= randint(0, 3)
138 for j
in range(randint(1,10)):
142 send
= randint(0, send_range
) != 0
143 o_p_ready
= yield self
.dut
.p
.o_ready
147 if send
and self
.i
!= len(self
.data
):
148 yield self
.dut
.p
.i_valid
.eq(1)
149 yield self
.dut
.p
.i_data
.eq(self
.data
[self
.i
])
152 yield self
.dut
.p
.i_valid
.eq(0)
156 while self
.o
!= len(self
.data
):
157 stall_range
= randint(0, 3)
158 for j
in range(randint(1,10)):
159 stall
= randint(0, stall_range
) != 0
160 yield self
.dut
.n
.i_ready
.eq(stall
)
162 o_n_valid
= yield self
.dut
.n
.o_valid
163 i_n_ready
= yield self
.dut
.n
.i_ready_test
164 if not o_n_valid
or not i_n_ready
:
166 o_data
= yield self
.dut
.n
.o_data
167 self
.resultfn(o_data
, self
.data
[self
.o
], self
.i
, self
.o
)
169 if self
.o
== len(self
.data
):
172 def test3_resultfn(o_data
, expected
, i
, o
):
173 assert o_data
== expected
+ 1, \
174 "%d-%d data %x not match %x\n" \
175 % (i
, o
, o_data
, expected
)
177 def data_placeholder():
179 for i
in range(num_tests
):
181 d
.src1
= randint(0, 1<<16-1)
182 d
.src2
= randint(0, 1<<16-1)
188 for i
in range(num_tests
):
189 data
.append({'src1': randint(0, 1<<16-1),
190 'src2': randint(0, 1<<16-1)})
195 def __init__(self
, dut
, resultfn
, data
=None, stage_ctl
=False):
197 self
.resultfn
= resultfn
198 self
.stage_ctl
= stage_ctl
203 for i
in range(num_tests
):
204 self
.data
.append((randint(0, 1<<16-1), randint(0, 1<<16-1)))
209 while self
.o
!= len(self
.data
):
210 send_range
= randint(0, 3)
211 for j
in range(randint(1,10)):
215 send
= randint(0, send_range
) != 0
217 o_p_ready
= yield self
.dut
.p
.o_ready
221 if send
and self
.i
!= len(self
.data
):
222 yield self
.dut
.p
.i_valid
.eq(1)
223 for v
in self
.dut
.set_input(self
.data
[self
.i
]):
227 yield self
.dut
.p
.i_valid
.eq(0)
231 while self
.o
!= len(self
.data
):
232 stall_range
= randint(0, 3)
233 for j
in range(randint(1,10)):
234 ready
= randint(0, stall_range
) != 0
236 yield self
.dut
.n
.i_ready
.eq(ready
)
238 o_n_valid
= yield self
.dut
.n
.o_valid
239 i_n_ready
= yield self
.dut
.n
.i_ready_test
240 if not o_n_valid
or not i_n_ready
:
242 if isinstance(self
.dut
.n
.o_data
, Record
):
244 dod
= self
.dut
.n
.o_data
245 for k
, v
in dod
.fields
.items():
248 o_data
= yield self
.dut
.n
.o_data
249 self
.resultfn(o_data
, self
.data
[self
.o
], self
.i
, self
.o
)
251 if self
.o
== len(self
.data
):
254 def test5_resultfn(o_data
, expected
, i
, o
):
255 res
= expected
[0] + expected
[1]
256 assert o_data
== res
, \
257 "%d-%d data %x not match %s\n" \
258 % (i
, o
, o_data
, repr(expected
))
262 for i
in range(num_tests
):
263 #data.append(randint(0, 1<<16-1))
268 stall
= randint(0, 3) != 0
269 send
= randint(0, 5) != 0
270 yield dut
.n
.i_ready
.eq(stall
)
271 o_p_ready
= yield dut
.p
.o_ready
273 if send
and i
!= len(data
):
274 yield dut
.p
.i_valid
.eq(1)
275 yield dut
.p
.i_data
.eq(data
[i
])
278 yield dut
.p
.i_valid
.eq(0)
280 o_n_valid
= yield dut
.n
.o_valid
281 i_n_ready
= yield dut
.n
.i_ready_test
282 if o_n_valid
and i_n_ready
:
283 o_data
= yield dut
.n
.o_data
284 assert o_data
== data
[o
] + 2, "%d-%d data %x not match %x\n" \
285 % (i
, o
, o_data
, data
[o
])
290 ######################################################################
292 ######################################################################
294 class ExampleBufPipe2(ControlBase
):
295 """ Example of how to do chained pipeline stages.
298 def elaborate(self
, platform
):
301 pipe1
= ExampleBufPipe()
302 pipe2
= ExampleBufPipe()
304 m
.submodules
.pipe1
= pipe1
305 m
.submodules
.pipe2
= pipe2
307 m
.d
.comb
+= self
.connect([pipe1
, pipe2
])
312 ######################################################################
314 ######################################################################
316 class ExampleBufPipeChain2(BufferedHandshake
):
317 """ connects two stages together as a *single* combinatorial stage.
320 stage1
= ExampleStageCls()
321 stage2
= ExampleStageCls()
322 combined
= StageChain([stage1
, stage2
])
323 BufferedHandshake
.__init
__(self
, combined
)
328 for i
in range(num_tests
):
329 data
.append(randint(0, 1<<16-2))
333 def test9_resultfn(o_data
, expected
, i
, o
):
335 assert o_data
== res
, \
336 "%d-%d received data %x not match expected %x\n" \
337 % (i
, o
, o_data
, res
)
340 ######################################################################
342 ######################################################################
345 def __init__(self
, width
, signed
):
347 self
.src1
= Signal((width
, signed
), name
="src1")
348 self
.src2
= Signal((width
, signed
), name
="src2")
349 self
.output
= Signal(width
, name
="out")
351 def elaborate(self
, platform
):
352 self
.m
.d
.comb
+= self
.output
.eq(Mux(self
.src1
< self
.src2
, 1, 0))
356 class LTStage(StageCls
):
357 """ module-based stage example
360 self
.slt
= SetLessThan(16, True)
363 return (Signal(16, name
="sig1"), Signal(16, "sig2"))
366 return Signal(16, "out")
368 def setup(self
, m
, i
):
370 m
.submodules
.slt
= self
.slt
371 m
.d
.comb
+= self
.slt
.src1
.eq(i
[0])
372 m
.d
.comb
+= self
.slt
.src2
.eq(i
[1])
373 m
.d
.comb
+= self
.o
.eq(self
.slt
.output
)
375 def process(self
, i
):
379 class LTStageDerived(SetLessThan
, StageCls
):
380 """ special version of a nmigen module where the module is also a stage
382 shows that you don't actually need to combinatorially connect
383 to the outputs, or add the module as a submodule: just return
384 the module output parameter(s) from the Stage.process() function
388 SetLessThan
.__init
__(self
, 16, True)
391 return (Signal(16), Signal(16))
396 def setup(self
, m
, i
):
397 m
.submodules
.slt
= self
398 m
.d
.comb
+= self
.src1
.eq(i
[0])
399 m
.d
.comb
+= self
.src2
.eq(i
[1])
401 def process(self
, i
):
405 class ExampleLTPipeline(UnbufferedPipeline
):
406 """ an example of how to use the unbuffered pipeline.
411 UnbufferedPipeline
.__init
__(self
, stage
)
414 class ExampleLTBufferedPipeDerived(BufferedHandshake
):
415 """ an example of how to use the buffered pipeline.
419 stage
= LTStageDerived()
420 BufferedHandshake
.__init
__(self
, stage
)
423 def test6_resultfn(o_data
, expected
, i
, o
):
424 res
= 1 if expected
[0] < expected
[1] else 0
425 assert o_data
== res
, \
426 "%d-%d data %x not match %s\n" \
427 % (i
, o
, o_data
, repr(expected
))
430 ######################################################################
432 ######################################################################
434 class ExampleAddRecordStage(StageCls
):
435 """ example use of a Record
438 record_spec
= [('src1', 16), ('src2', 16)]
440 """ returns a Record using the specification
442 return Record(self
.record_spec
)
445 return Record(self
.record_spec
)
447 def process(self
, i
):
448 """ process the input data, returning a dictionary with key names
449 that exactly match the Record's attributes.
451 return {'src1': i
.src1
+ 1,
454 ######################################################################
456 ######################################################################
458 class ExampleAddRecordPlaceHolderStage(StageCls
):
459 """ example use of a Record, with a placeholder as the processing result
462 record_spec
= [('src1', 16), ('src2', 16)]
464 """ returns a Record using the specification
466 return Record(self
.record_spec
)
469 return Record(self
.record_spec
)
471 def process(self
, i
):
472 """ process the input data, returning a PlaceHolder class instance
473 with attributes that exactly match those of the Record.
481 class PlaceHolder
: pass
484 class ExampleAddRecordPipe(UnbufferedPipeline
):
485 """ an example of how to use the combinatorial pipeline.
489 stage
= ExampleAddRecordStage()
490 UnbufferedPipeline
.__init
__(self
, stage
)
493 def test7_resultfn(o_data
, expected
, i
, o
):
494 res
= (expected
['src1'] + 1, expected
['src2'] + 1)
495 assert o_data
['src1'] == res
[0] and o_data
['src2'] == res
[1], \
496 "%d-%d data %s not match %s\n" \
497 % (i
, o
, repr(o_data
), repr(expected
))
500 class ExampleAddRecordPlaceHolderPipe(UnbufferedPipeline
):
501 """ an example of how to use the combinatorial pipeline.
505 stage
= ExampleAddRecordPlaceHolderStage()
506 UnbufferedPipeline
.__init
__(self
, stage
)
509 def test11_resultfn(o_data
, expected
, i
, o
):
510 res1
= expected
.src1
+ 1
511 res2
= expected
.src2
+ 1
512 assert o_data
['src1'] == res1
and o_data
['src2'] == res2
, \
513 "%d-%d data %s not match %s\n" \
514 % (i
, o
, repr(o_data
), repr(expected
))
517 ######################################################################
519 ######################################################################
522 class Example2OpClass
:
523 """ an example of a class used to store 2 operands.
524 requires an eq function, to conform with the pipeline stage API
528 self
.op1
= Signal(16)
529 self
.op2
= Signal(16)
532 return [self
.op1
.eq(i
.op1
), self
.op2
.eq(i
.op2
)]
535 class ExampleAddClassStage(StageCls
):
536 """ an example of how to use the buffered pipeline, as a class instance
540 """ returns an instance of an Example2OpClass.
542 return Example2OpClass()
545 """ returns an output signal which will happen to contain the sum
550 def process(self
, i
):
551 """ process the input data (sums the values in the tuple) and returns it
556 class ExampleBufPipeAddClass(BufferedHandshake
):
557 """ an example of how to use the buffered pipeline, using a class instance
561 addstage
= ExampleAddClassStage()
562 BufferedHandshake
.__init
__(self
, addstage
)
566 """ the eq function, called by set_input, needs an incoming object
567 that conforms to the Example2OpClass.eq function requirements
568 easiest way to do that is to create a class that has the exact
569 same member layout (self.op1, self.op2) as Example2OpClass
571 def __init__(self
, op1
, op2
):
576 def test8_resultfn(o_data
, expected
, i
, o
):
577 res
= expected
.op1
+ expected
.op2
# these are a TestInputAdd instance
578 assert o_data
== res
, \
579 "%d-%d data %x not match %s\n" \
580 % (i
, o
, o_data
, repr(expected
))
584 for i
in range(num_tests
):
585 data
.append(TestInputAdd(randint(0, 1<<16-1), randint(0, 1<<16-1)))
589 ######################################################################
591 ######################################################################
593 class ExampleStageDelayCls(StageCls
):
594 """ an example of how to use the buffered pipeline, in a static class
598 def __init__(self
, valid_trigger
=2):
599 self
.count
= Signal(2)
600 self
.valid_trigger
= valid_trigger
603 return Signal(16, name
="example_input_signal")
606 return Signal(16, name
="example_output_signal")
610 return (self
.count
== 1)# | (self.count == 3)
613 def d_valid(self
, i_ready
):
614 return self
.count
== self
.valid_trigger
617 def process(self
, i
):
618 """ process the input data and returns it (adds 1)
622 def elaborate(self
, platform
):
624 m
.d
.sync
+= self
.count
.eq(self
.count
+ 1)
628 class ExampleBufDelayedPipe(BufferedHandshake
):
631 stage
= ExampleStageDelayCls(valid_trigger
=2)
632 BufferedHandshake
.__init
__(self
, stage
, stage_ctl
=True)
634 def elaborate(self
, platform
):
635 m
= BufferedHandshake
.elaborate(self
, platform
)
636 m
.submodules
.stage
= self
.stage
642 for i
in range(num_tests
):
643 data
.append(1<<((i
*3)%15))
644 #data.append(randint(0, 1<<16-2))
645 #print (hex(data[-1]))
649 def test12_resultfn(o_data
, expected
, i
, o
):
651 assert o_data
== res
, \
652 "%d-%d data %x not match %x\n" \
653 % (i
, o
, o_data
, res
)
656 ######################################################################
658 ######################################################################
660 class ExampleUnBufDelayedPipe(BufferedHandshake
):
663 stage
= ExampleStageDelayCls(valid_trigger
=3)
664 BufferedHandshake
.__init
__(self
, stage
, stage_ctl
=True)
666 def elaborate(self
, platform
):
667 m
= BufferedHandshake
.elaborate(self
, platform
)
668 m
.submodules
.stage
= self
.stage
671 ######################################################################
673 ######################################################################
675 class ExampleBufModeAdd1Pipe(SimpleHandshake
):
678 stage
= ExampleStageCls()
679 SimpleHandshake
.__init
__(self
, stage
)
682 ######################################################################
684 ######################################################################
686 class ExampleBufModeUnBufPipe(ControlBase
):
688 def elaborate(self
, platform
):
689 m
= ControlBase
._elaborate
(self
, platform
)
691 pipe1
= ExampleBufModeAdd1Pipe()
692 pipe2
= ExampleBufAdd1Pipe()
694 m
.submodules
.pipe1
= pipe1
695 m
.submodules
.pipe2
= pipe2
697 m
.d
.comb
+= self
.connect([pipe1
, pipe2
])
701 ######################################################################
703 ######################################################################
705 class ExampleUnBufAdd1Pipe2(UnbufferedPipeline2
):
708 stage
= ExampleStageCls()
709 UnbufferedPipeline2
.__init__(self
, stage
)
712 ######################################################################
714 ######################################################################
716 class PassThroughTest(PassThroughHandshake
):
719 return Signal(16, "out")
722 stage
= PassThroughStage(self
.iospecfn
)
723 PassThroughHandshake
.__init
__(self
, stage
)
725 def test_identical_resultfn(o_data
, expected
, i
, o
):
727 assert o_data
== res
, \
728 "%d-%d data %x not match %x\n" \
729 % (i
, o
, o_data
, res
)
732 ######################################################################
734 ######################################################################
736 class ExamplePassAdd1Pipe(PassThroughHandshake
):
739 stage
= ExampleStageCls()
740 PassThroughHandshake
.__init
__(self
, stage
)
743 class ExampleBufPassThruPipe(ControlBase
):
745 def elaborate(self
, platform
):
746 m
= ControlBase
._elaborate
(self
, platform
)
748 # XXX currently fails: any other permutation works fine.
749 # p1=u,p2=b ok p1=u,p2=u ok p1=b,p2=b ok
750 # also fails using UnbufferedPipeline as well
751 pipe1
= ExampleBufModeAdd1Pipe()
752 pipe2
= ExamplePassAdd1Pipe()
754 m
.submodules
.pipe1
= pipe1
755 m
.submodules
.pipe2
= pipe2
757 m
.d
.comb
+= self
.connect([pipe1
, pipe2
])
762 ######################################################################
764 ######################################################################
767 return Signal(16, name
="din")
769 class FIFOTest16(FIFOtest
):
773 FIFOtest
.__init
__(self
, iospecfn
, 16, 2)
776 ######################################################################
778 ######################################################################
780 class ExampleFIFOPassThruPipe1(ControlBase
):
782 def elaborate(self
, platform
):
783 m
= ControlBase
._elaborate
(self
, platform
)
786 pipe2
= ExamplePassAdd1Pipe()
788 m
.submodules
.pipe1
= pipe1
789 m
.submodules
.pipe2
= pipe2
791 m
.d
.comb
+= self
.connect([pipe1
, pipe2
])
796 ######################################################################
798 ######################################################################
800 class ExampleBufPassThruPipe2(ControlBase
):
802 def elaborate(self
, platform
):
803 m
= ControlBase
._elaborate
(self
, platform
)
805 # XXX currently fails: any other permutation works fine.
806 # p1=u,p2=b ok p1=u,p2=u ok p1=b,p2=b ok
807 # also fails using UnbufferedPipeline as well
808 #pipe1 = ExampleUnBufAdd1Pipe()
809 #pipe2 = ExampleBufAdd1Pipe()
810 pipe1
= ExampleBufAdd1Pipe()
811 pipe2
= ExamplePassAdd1Pipe()
813 m
.submodules
.pipe1
= pipe1
814 m
.submodules
.pipe2
= pipe2
816 m
.d
.comb
+= self
.connect([pipe1
, pipe2
])
821 ######################################################################
823 ######################################################################
825 class ExampleBufPipe3(ControlBase
):
826 """ Example of how to do delayed pipeline, where the stage signals
830 def elaborate(self
, platform
):
831 m
= ControlBase
._elaborate
(self
, platform
)
833 pipe1
= ExampleBufDelayedPipe()
834 pipe2
= ExampleBufPipe()
836 m
.submodules
.pipe1
= pipe1
837 m
.submodules
.pipe2
= pipe2
839 m
.d
.comb
+= self
.connect([pipe1
, pipe2
])
843 ######################################################################
844 # Test 999 - XXX FAILS
845 # http://bugs.libre-riscv.org/show_bug.cgi?id=57
846 ######################################################################
848 class ExampleBufAdd1Pipe(BufferedHandshake
):
851 stage
= ExampleStageCls()
852 BufferedHandshake
.__init
__(self
, stage
)
855 class ExampleUnBufAdd1Pipe(UnbufferedPipeline
):
858 stage
= ExampleStageCls()
859 UnbufferedPipeline
.__init
__(self
, stage
)
862 class ExampleBufUnBufPipe(ControlBase
):
864 def elaborate(self
, platform
):
865 m
= ControlBase
._elaborate
(self
, platform
)
867 # XXX currently fails: any other permutation works fine.
868 # p1=u,p2=b ok p1=u,p2=u ok p1=b,p2=b ok
869 # also fails using UnbufferedPipeline as well
870 #pipe1 = ExampleUnBufAdd1Pipe()
871 #pipe2 = ExampleBufAdd1Pipe()
872 pipe1
= ExampleBufAdd1Pipe()
873 pipe2
= ExampleUnBufAdd1Pipe()
875 m
.submodules
.pipe1
= pipe1
876 m
.submodules
.pipe2
= pipe2
878 m
.d
.comb
+= self
.connect([pipe1
, pipe2
])
883 ######################################################################
885 ######################################################################
889 if __name__
== '__main__':
891 dut
= ExampleBufPipe()
892 run_simulation(dut
, testbench(dut
), vcd_name
="test_bufpipe.vcd")
895 dut
= ExampleBufPipe2()
896 run_simulation(dut
, testbench2(dut
), vcd_name
="test_bufpipe2.vcd")
897 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
898 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
899 [dut
.p
.i_data
] + [dut
.n
.o_data
]
900 vl
= rtlil
.convert(dut
, ports
=ports
)
901 with
open("test_bufpipe2.il", "w") as f
:
906 dut
= ExampleBufPipe()
907 test
= Test3(dut
, test3_resultfn
)
908 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufpipe3.vcd")
911 dut
= ExamplePipeline()
912 test
= Test3(dut
, test3_resultfn
)
913 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_combpipe3.vcd")
916 dut
= ExampleBufPipe2()
917 run_simulation(dut
, testbench4(dut
), vcd_name
="test_bufpipe4.vcd")
920 dut
= ExampleBufPipeAdd()
921 test
= Test5(dut
, test5_resultfn
, stage_ctl
=True)
922 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufpipe5.vcd")
925 dut
= ExampleLTPipeline()
926 test
= Test5(dut
, test6_resultfn
)
927 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_ltcomb6.vcd")
929 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
930 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
931 list(dut
.p
.i_data
) + [dut
.n
.o_data
]
932 vl
= rtlil
.convert(dut
, ports
=ports
)
933 with
open("test_ltcomb_pipe.il", "w") as f
:
937 dut
= ExampleAddRecordPipe()
939 test
= Test5(dut
, test7_resultfn
, data
=data
)
940 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_addrecord.vcd")
942 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
943 dut
.n
.o_valid
, dut
.p
.o_ready
,
944 dut
.p
.i_data
.src1
, dut
.p
.i_data
.src2
,
945 dut
.n
.o_data
.src1
, dut
.n
.o_data
.src2
]
946 vl
= rtlil
.convert(dut
, ports
=ports
)
947 with
open("test_recordcomb_pipe.il", "w") as f
:
951 dut
= ExampleBufPipeAddClass()
953 test
= Test5(dut
, test8_resultfn
, data
=data
)
954 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufpipe8.vcd")
957 dut
= ExampleBufPipeChain2()
958 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
959 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
960 [dut
.p
.i_data
] + [dut
.n
.o_data
]
961 vl
= rtlil
.convert(dut
, ports
=ports
)
962 with
open("test_bufpipechain2.il", "w") as f
:
966 test
= Test5(dut
, test9_resultfn
, data
=data
)
967 run_simulation(dut
, [test
.send
, test
.rcv
],
968 vcd_name
="test_bufpipechain2.vcd")
971 dut
= ExampleLTBufferedPipeDerived()
972 test
= Test5(dut
, test6_resultfn
)
973 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_ltbufpipe10.vcd")
974 vl
= rtlil
.convert(dut
, ports
=ports
)
975 with
open("test_ltbufpipe10.il", "w") as f
:
979 dut
= ExampleAddRecordPlaceHolderPipe()
980 data
=data_placeholder()
981 test
= Test5(dut
, test11_resultfn
, data
=data
)
982 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_addrecord.vcd")
986 dut
= ExampleBufDelayedPipe()
988 test
= Test5(dut
, test12_resultfn
, data
=data
)
989 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufpipe12.vcd")
990 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
991 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
992 [dut
.p
.i_data
] + [dut
.n
.o_data
]
993 vl
= rtlil
.convert(dut
, ports
=ports
)
994 with
open("test_bufpipe12.il", "w") as f
:
998 dut
= ExampleUnBufDelayedPipe()
1000 test
= Test5(dut
, test12_resultfn
, data
=data
)
1001 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_unbufpipe13.vcd")
1002 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
1003 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
1004 [dut
.p
.i_data
] + [dut
.n
.o_data
]
1005 vl
= rtlil
.convert(dut
, ports
=ports
)
1006 with
open("test_unbufpipe13.il", "w") as f
:
1010 dut
= ExampleBufModeAdd1Pipe()
1011 data
= data_chain1()
1012 test
= Test5(dut
, test12_resultfn
, data
=data
)
1013 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufunbuf15.vcd")
1014 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
1015 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
1016 [dut
.p
.i_data
] + [dut
.n
.o_data
]
1017 vl
= rtlil
.convert(dut
, ports
=ports
)
1018 with
open("test_bufunbuf15.il", "w") as f
:
1022 dut
= ExampleBufModeUnBufPipe()
1023 data
= data_chain1()
1024 test
= Test5(dut
, test9_resultfn
, data
=data
)
1025 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufunbuf16.vcd")
1026 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
1027 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
1028 [dut
.p
.i_data
] + [dut
.n
.o_data
]
1029 vl
= rtlil
.convert(dut
, ports
=ports
)
1030 with
open("test_bufunbuf16.il", "w") as f
:
1034 dut
= ExampleUnBufAdd1Pipe2()
1035 data
= data_chain1()
1036 test
= Test5(dut
, test12_resultfn
, data
=data
)
1037 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_unbufpipe17.vcd")
1038 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
1039 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
1040 [dut
.p
.i_data
] + [dut
.n
.o_data
]
1041 vl
= rtlil
.convert(dut
, ports
=ports
)
1042 with
open("test_unbufpipe17.il", "w") as f
:
1046 dut
= PassThroughTest()
1047 data
= data_chain1()
1048 test
= Test5(dut
, test_identical_resultfn
, data
=data
)
1049 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_passthru18.vcd")
1050 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
1051 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
1052 [dut
.p
.i_data
] + [dut
.n
.o_data
]
1053 vl
= rtlil
.convert(dut
, ports
=ports
)
1054 with
open("test_passthru18.il", "w") as f
:
1058 dut
= ExampleBufPassThruPipe()
1059 data
= data_chain1()
1060 test
= Test5(dut
, test9_resultfn
, data
=data
)
1061 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufpass19.vcd")
1062 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
1063 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
1064 [dut
.p
.i_data
] + [dut
.n
.o_data
]
1065 vl
= rtlil
.convert(dut
, ports
=ports
)
1066 with
open("test_bufpass19.il", "w") as f
:
1071 data
= data_chain1()
1072 test
= Test5(dut
, test_identical_resultfn
, data
=data
)
1073 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_fifo20.vcd")
1074 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
1075 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
1076 [dut
.p
.i_data
] + [dut
.n
.o_data
]
1077 vl
= rtlil
.convert(dut
, ports
=ports
)
1078 with
open("test_fifo20.il", "w") as f
:
1082 dut
= ExampleFIFOPassThruPipe1()
1083 data
= data_chain1()
1084 test
= Test5(dut
, test12_resultfn
, data
=data
)
1085 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_fifopass21.vcd")
1086 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
1087 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
1088 [dut
.p
.i_data
] + [dut
.n
.o_data
]
1089 vl
= rtlil
.convert(dut
, ports
=ports
)
1090 with
open("test_fifopass21.il", "w") as f
:
1094 dut
= ExampleBufPassThruPipe2()
1095 data
= data_chain1()
1096 test
= Test5(dut
, test9_resultfn
, data
=data
)
1097 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufpass997.vcd")
1098 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
1099 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
1100 [dut
.p
.i_data
] + [dut
.n
.o_data
]
1101 vl
= rtlil
.convert(dut
, ports
=ports
)
1102 with
open("test_bufpass997.il", "w") as f
:
1105 print ("test 998 (fails, bug)")
1106 dut
= ExampleBufPipe3()
1107 data
= data_chain1()
1108 test
= Test5(dut
, test9_resultfn
, data
=data
)
1109 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufpipe14.vcd")
1110 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
1111 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
1112 [dut
.p
.i_data
] + [dut
.n
.o_data
]
1113 vl
= rtlil
.convert(dut
, ports
=ports
)
1114 with
open("test_bufpipe14.il", "w") as f
:
1117 print ("test 999 (expected to fail, which is a bug)")
1118 dut
= ExampleBufUnBufPipe()
1119 data
= data_chain1()
1120 test
= Test5(dut
, test9_resultfn
, data
=data
)
1121 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufunbuf999.vcd")
1122 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
1123 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
1124 [dut
.p
.i_data
] + [dut
.n
.o_data
]
1125 vl
= rtlil
.convert(dut
, ports
=ports
)
1126 with
open("test_bufunbuf999.il", "w") as f
: