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
32 from random
import randint
, seed
37 def check_o_n_valid(dut
, val
):
38 o_n_valid
= yield dut
.n
.o_valid
39 assert o_n_valid
== val
41 def check_o_n_valid2(dut
, val
):
42 o_n_valid
= yield dut
.n
.o_valid
43 assert o_n_valid
== val
47 #yield dut.i_p_rst.eq(1)
48 yield dut
.n
.i_ready
.eq(0)
49 yield dut
.p
.o_ready
.eq(0)
52 #yield dut.i_p_rst.eq(0)
53 yield dut
.n
.i_ready
.eq(1)
54 yield dut
.p
.i_data
.eq(5)
55 yield dut
.p
.i_valid
.eq(1)
58 yield dut
.p
.i_data
.eq(7)
59 yield from check_o_n_valid(dut
, 0) # effects of i_p_valid delayed
61 yield from check_o_n_valid(dut
, 1) # ok *now* i_p_valid effect is felt
63 yield dut
.p
.i_data
.eq(2)
65 yield dut
.n
.i_ready
.eq(0) # begin going into "stall" (next stage says ready)
66 yield dut
.p
.i_data
.eq(9)
68 yield dut
.p
.i_valid
.eq(0)
69 yield dut
.p
.i_data
.eq(12)
71 yield dut
.p
.i_data
.eq(32)
72 yield dut
.n
.i_ready
.eq(1)
74 yield from check_o_n_valid(dut
, 1) # buffer still needs to output
76 yield from check_o_n_valid(dut
, 1) # buffer still needs to output
78 yield from check_o_n_valid(dut
, 0) # buffer outputted, *now* we're done.
83 #yield dut.p.i_rst.eq(1)
84 yield dut
.n
.i_ready
.eq(0)
85 #yield dut.p.o_ready.eq(0)
88 #yield dut.p.i_rst.eq(0)
89 yield dut
.n
.i_ready
.eq(1)
90 yield dut
.p
.i_data
.eq(5)
91 yield dut
.p
.i_valid
.eq(1)
94 yield dut
.p
.i_data
.eq(7)
95 yield from check_o_n_valid2(dut
, 0) # effects of i_p_valid delayed 2 clocks
97 yield from check_o_n_valid2(dut
, 0) # effects of i_p_valid delayed 2 clocks
99 yield dut
.p
.i_data
.eq(2)
101 yield from check_o_n_valid2(dut
, 1) # ok *now* i_p_valid effect is felt
102 yield dut
.n
.i_ready
.eq(0) # begin going into "stall" (next stage says ready)
103 yield dut
.p
.i_data
.eq(9)
105 yield dut
.p
.i_valid
.eq(0)
106 yield dut
.p
.i_data
.eq(12)
108 yield dut
.p
.i_data
.eq(32)
109 yield dut
.n
.i_ready
.eq(1)
111 yield from check_o_n_valid2(dut
, 1) # buffer still needs to output
113 yield from check_o_n_valid2(dut
, 1) # buffer still needs to output
115 yield from check_o_n_valid2(dut
, 1) # buffer still needs to output
117 yield from check_o_n_valid2(dut
, 0) # buffer outputted, *now* we're done.
124 def __init__(self
, dut
, resultfn
):
126 self
.resultfn
= resultfn
128 for i
in range(num_tests
):
129 #data.append(randint(0, 1<<16-1))
130 self
.data
.append(i
+1)
135 while self
.o
!= len(self
.data
):
136 send_range
= randint(0, 3)
137 for j
in range(randint(1,10)):
141 send
= randint(0, send_range
) != 0
142 o_p_ready
= yield self
.dut
.p
.o_ready
146 if send
and self
.i
!= len(self
.data
):
147 yield self
.dut
.p
.i_valid
.eq(1)
148 yield self
.dut
.p
.i_data
.eq(self
.data
[self
.i
])
151 yield self
.dut
.p
.i_valid
.eq(0)
155 while self
.o
!= len(self
.data
):
156 stall_range
= randint(0, 3)
157 for j
in range(randint(1,10)):
158 stall
= randint(0, stall_range
) != 0
159 yield self
.dut
.n
.i_ready
.eq(stall
)
161 o_n_valid
= yield self
.dut
.n
.o_valid
162 i_n_ready
= yield self
.dut
.n
.i_ready_test
163 if not o_n_valid
or not i_n_ready
:
165 o_data
= yield self
.dut
.n
.o_data
166 self
.resultfn(o_data
, self
.data
[self
.o
], self
.i
, self
.o
)
168 if self
.o
== len(self
.data
):
171 def test3_resultfn(o_data
, expected
, i
, o
):
172 assert o_data
== expected
+ 1, \
173 "%d-%d data %x not match %x\n" \
174 % (i
, o
, o_data
, expected
)
176 def data_placeholder():
178 for i
in range(num_tests
):
180 d
.src1
= randint(0, 1<<16-1)
181 d
.src2
= randint(0, 1<<16-1)
187 for i
in range(num_tests
):
188 data
.append({'src1': randint(0, 1<<16-1),
189 'src2': randint(0, 1<<16-1)})
194 def __init__(self
, dut
, resultfn
, data
=None, stage_ctl
=False):
196 self
.resultfn
= resultfn
197 self
.stage_ctl
= stage_ctl
202 for i
in range(num_tests
):
203 self
.data
.append((randint(0, 1<<16-1), randint(0, 1<<16-1)))
208 while self
.o
!= len(self
.data
):
209 send_range
= randint(0, 3)
210 for j
in range(randint(1,10)):
214 send
= randint(0, send_range
) != 0
216 o_p_ready
= yield self
.dut
.p
.o_ready
220 if send
and self
.i
!= len(self
.data
):
221 yield self
.dut
.p
.i_valid
.eq(1)
222 for v
in self
.dut
.set_input(self
.data
[self
.i
]):
226 yield self
.dut
.p
.i_valid
.eq(0)
230 while self
.o
!= len(self
.data
):
231 stall_range
= randint(0, 3)
232 for j
in range(randint(1,10)):
233 ready
= randint(0, stall_range
) != 0
235 yield self
.dut
.n
.i_ready
.eq(ready
)
237 o_n_valid
= yield self
.dut
.n
.o_valid
238 i_n_ready
= yield self
.dut
.n
.i_ready_test
239 if not o_n_valid
or not i_n_ready
:
241 if isinstance(self
.dut
.n
.o_data
, Record
):
243 dod
= self
.dut
.n
.o_data
244 for k
, v
in dod
.fields
.items():
247 o_data
= yield self
.dut
.n
.o_data
248 self
.resultfn(o_data
, self
.data
[self
.o
], self
.i
, self
.o
)
250 if self
.o
== len(self
.data
):
253 def test5_resultfn(o_data
, expected
, i
, o
):
254 res
= expected
[0] + expected
[1]
255 assert o_data
== res
, \
256 "%d-%d data %x not match %s\n" \
257 % (i
, o
, o_data
, repr(expected
))
261 for i
in range(num_tests
):
262 #data.append(randint(0, 1<<16-1))
267 stall
= randint(0, 3) != 0
268 send
= randint(0, 5) != 0
269 yield dut
.n
.i_ready
.eq(stall
)
270 o_p_ready
= yield dut
.p
.o_ready
272 if send
and i
!= len(data
):
273 yield dut
.p
.i_valid
.eq(1)
274 yield dut
.p
.i_data
.eq(data
[i
])
277 yield dut
.p
.i_valid
.eq(0)
279 o_n_valid
= yield dut
.n
.o_valid
280 i_n_ready
= yield dut
.n
.i_ready_test
281 if o_n_valid
and i_n_ready
:
282 o_data
= yield dut
.n
.o_data
283 assert o_data
== data
[o
] + 2, "%d-%d data %x not match %x\n" \
284 % (i
, o
, o_data
, data
[o
])
289 ######################################################################
291 ######################################################################
293 class ExampleBufPipe2(ControlBase
):
294 """ Example of how to do chained pipeline stages.
297 def elaborate(self
, platform
):
300 pipe1
= ExampleBufPipe()
301 pipe2
= ExampleBufPipe()
303 m
.submodules
.pipe1
= pipe1
304 m
.submodules
.pipe2
= pipe2
306 m
.d
.comb
+= self
.connect([pipe1
, pipe2
])
311 ######################################################################
313 ######################################################################
315 class ExampleBufPipeChain2(BufferedHandshake
):
316 """ connects two stages together as a *single* combinatorial stage.
319 stage1
= ExampleStageCls()
320 stage2
= ExampleStageCls()
321 combined
= StageChain([stage1
, stage2
])
322 BufferedHandshake
.__init
__(self
, combined
)
327 for i
in range(num_tests
):
328 data
.append(randint(0, 1<<16-2))
332 def test9_resultfn(o_data
, expected
, i
, o
):
334 assert o_data
== res
, \
335 "%d-%d received data %x not match expected %x\n" \
336 % (i
, o
, o_data
, res
)
339 ######################################################################
341 ######################################################################
344 def __init__(self
, width
, signed
):
346 self
.src1
= Signal((width
, signed
), name
="src1")
347 self
.src2
= Signal((width
, signed
), name
="src2")
348 self
.output
= Signal(width
, name
="out")
350 def elaborate(self
, platform
):
351 self
.m
.d
.comb
+= self
.output
.eq(Mux(self
.src1
< self
.src2
, 1, 0))
355 class LTStage(StageCls
):
356 """ module-based stage example
359 self
.slt
= SetLessThan(16, True)
362 return (Signal(16, name
="sig1"), Signal(16, "sig2"))
365 return Signal(16, "out")
367 def setup(self
, m
, i
):
369 m
.submodules
.slt
= self
.slt
370 m
.d
.comb
+= self
.slt
.src1
.eq(i
[0])
371 m
.d
.comb
+= self
.slt
.src2
.eq(i
[1])
372 m
.d
.comb
+= self
.o
.eq(self
.slt
.output
)
374 def process(self
, i
):
378 class LTStageDerived(SetLessThan
, StageCls
):
379 """ special version of a nmigen module where the module is also a stage
381 shows that you don't actually need to combinatorially connect
382 to the outputs, or add the module as a submodule: just return
383 the module output parameter(s) from the Stage.process() function
387 SetLessThan
.__init
__(self
, 16, True)
390 return (Signal(16), Signal(16))
395 def setup(self
, m
, i
):
396 m
.submodules
.slt
= self
397 m
.d
.comb
+= self
.src1
.eq(i
[0])
398 m
.d
.comb
+= self
.src2
.eq(i
[1])
400 def process(self
, i
):
404 class ExampleLTPipeline(UnbufferedPipeline
):
405 """ an example of how to use the unbuffered pipeline.
410 UnbufferedPipeline
.__init
__(self
, stage
)
413 class ExampleLTBufferedPipeDerived(BufferedHandshake
):
414 """ an example of how to use the buffered pipeline.
418 stage
= LTStageDerived()
419 BufferedHandshake
.__init
__(self
, stage
)
422 def test6_resultfn(o_data
, expected
, i
, o
):
423 res
= 1 if expected
[0] < expected
[1] else 0
424 assert o_data
== res
, \
425 "%d-%d data %x not match %s\n" \
426 % (i
, o
, o_data
, repr(expected
))
429 ######################################################################
431 ######################################################################
433 class ExampleAddRecordStage(StageCls
):
434 """ example use of a Record
437 record_spec
= [('src1', 16), ('src2', 16)]
439 """ returns a Record using the specification
441 return Record(self
.record_spec
)
444 return Record(self
.record_spec
)
446 def process(self
, i
):
447 """ process the input data, returning a dictionary with key names
448 that exactly match the Record's attributes.
450 return {'src1': i
.src1
+ 1,
453 ######################################################################
455 ######################################################################
457 class ExampleAddRecordPlaceHolderStage(StageCls
):
458 """ example use of a Record, with a placeholder as the processing result
461 record_spec
= [('src1', 16), ('src2', 16)]
463 """ returns a Record using the specification
465 return Record(self
.record_spec
)
468 return Record(self
.record_spec
)
470 def process(self
, i
):
471 """ process the input data, returning a PlaceHolder class instance
472 with attributes that exactly match those of the Record.
480 class PlaceHolder
: pass
483 class ExampleAddRecordPipe(UnbufferedPipeline
):
484 """ an example of how to use the combinatorial pipeline.
488 stage
= ExampleAddRecordStage()
489 UnbufferedPipeline
.__init
__(self
, stage
)
492 def test7_resultfn(o_data
, expected
, i
, o
):
493 res
= (expected
['src1'] + 1, expected
['src2'] + 1)
494 assert o_data
['src1'] == res
[0] and o_data
['src2'] == res
[1], \
495 "%d-%d data %s not match %s\n" \
496 % (i
, o
, repr(o_data
), repr(expected
))
499 class ExampleAddRecordPlaceHolderPipe(UnbufferedPipeline
):
500 """ an example of how to use the combinatorial pipeline.
504 stage
= ExampleAddRecordPlaceHolderStage()
505 UnbufferedPipeline
.__init
__(self
, stage
)
508 def test11_resultfn(o_data
, expected
, i
, o
):
509 res1
= expected
.src1
+ 1
510 res2
= expected
.src2
+ 1
511 assert o_data
['src1'] == res1
and o_data
['src2'] == res2
, \
512 "%d-%d data %s not match %s\n" \
513 % (i
, o
, repr(o_data
), repr(expected
))
516 ######################################################################
518 ######################################################################
521 class Example2OpClass
:
522 """ an example of a class used to store 2 operands.
523 requires an eq function, to conform with the pipeline stage API
527 self
.op1
= Signal(16)
528 self
.op2
= Signal(16)
531 return [self
.op1
.eq(i
.op1
), self
.op2
.eq(i
.op2
)]
534 class ExampleAddClassStage(StageCls
):
535 """ an example of how to use the buffered pipeline, as a class instance
539 """ returns an instance of an Example2OpClass.
541 return Example2OpClass()
544 """ returns an output signal which will happen to contain the sum
549 def process(self
, i
):
550 """ process the input data (sums the values in the tuple) and returns it
555 class ExampleBufPipeAddClass(BufferedHandshake
):
556 """ an example of how to use the buffered pipeline, using a class instance
560 addstage
= ExampleAddClassStage()
561 BufferedHandshake
.__init
__(self
, addstage
)
565 """ the eq function, called by set_input, needs an incoming object
566 that conforms to the Example2OpClass.eq function requirements
567 easiest way to do that is to create a class that has the exact
568 same member layout (self.op1, self.op2) as Example2OpClass
570 def __init__(self
, op1
, op2
):
575 def test8_resultfn(o_data
, expected
, i
, o
):
576 res
= expected
.op1
+ expected
.op2
# these are a TestInputAdd instance
577 assert o_data
== res
, \
578 "%d-%d data %x not match %s\n" \
579 % (i
, o
, o_data
, repr(expected
))
583 for i
in range(num_tests
):
584 data
.append(TestInputAdd(randint(0, 1<<16-1), randint(0, 1<<16-1)))
588 ######################################################################
590 ######################################################################
592 class ExampleStageDelayCls(StageCls
):
593 """ an example of how to use the buffered pipeline, in a static class
597 def __init__(self
, valid_trigger
=2):
598 self
.count
= Signal(2)
599 self
.valid_trigger
= valid_trigger
602 return Signal(16, name
="example_input_signal")
605 return Signal(16, name
="example_output_signal")
609 return (self
.count
== 1)# | (self.count == 3)
612 def d_valid(self
, i_ready
):
613 return self
.count
== self
.valid_trigger
616 def process(self
, i
):
617 """ process the input data and returns it (adds 1)
621 def elaborate(self
, platform
):
623 m
.d
.sync
+= self
.count
.eq(self
.count
+ 1)
627 class ExampleBufDelayedPipe(BufferedHandshake
):
630 stage
= ExampleStageDelayCls(valid_trigger
=2)
631 BufferedHandshake
.__init
__(self
, stage
, stage_ctl
=True)
633 def elaborate(self
, platform
):
634 m
= BufferedHandshake
.elaborate(self
, platform
)
635 m
.submodules
.stage
= self
.stage
641 for i
in range(num_tests
):
642 data
.append(1<<((i
*3)%15))
643 #data.append(randint(0, 1<<16-2))
644 #print (hex(data[-1]))
648 def test12_resultfn(o_data
, expected
, i
, o
):
650 assert o_data
== res
, \
651 "%d-%d data %x not match %x\n" \
652 % (i
, o
, o_data
, res
)
655 ######################################################################
657 ######################################################################
659 class ExampleUnBufDelayedPipe(BufferedHandshake
):
662 stage
= ExampleStageDelayCls(valid_trigger
=3)
663 BufferedHandshake
.__init
__(self
, stage
, stage_ctl
=True)
665 def elaborate(self
, platform
):
666 m
= BufferedHandshake
.elaborate(self
, platform
)
667 m
.submodules
.stage
= self
.stage
670 ######################################################################
672 ######################################################################
674 class ExampleBufModeAdd1Pipe(SimpleHandshake
):
677 stage
= ExampleStageCls()
678 SimpleHandshake
.__init
__(self
, stage
)
681 ######################################################################
683 ######################################################################
685 class ExampleBufModeUnBufPipe(ControlBase
):
687 def elaborate(self
, platform
):
688 m
= ControlBase
._elaborate
(self
, platform
)
690 pipe1
= ExampleBufModeAdd1Pipe()
691 pipe2
= ExampleBufAdd1Pipe()
693 m
.submodules
.pipe1
= pipe1
694 m
.submodules
.pipe2
= pipe2
696 m
.d
.comb
+= self
.connect([pipe1
, pipe2
])
700 ######################################################################
702 ######################################################################
704 class ExampleUnBufAdd1Pipe2(UnbufferedPipeline2
):
707 stage
= ExampleStageCls()
708 UnbufferedPipeline2
.__init__(self
, stage
)
711 ######################################################################
713 ######################################################################
715 class PassThroughTest(PassThroughHandshake
):
718 return Signal(16, "out")
721 stage
= PassThroughStage(self
.iospecfn
)
722 PassThroughHandshake
.__init
__(self
, stage
)
724 def test_identical_resultfn(o_data
, expected
, i
, o
):
726 assert o_data
== res
, \
727 "%d-%d data %x not match %x\n" \
728 % (i
, o
, o_data
, res
)
731 ######################################################################
733 ######################################################################
735 class ExampleBufPipe3(ControlBase
):
736 """ Example of how to do delayed pipeline, where the stage signals
740 def elaborate(self
, platform
):
741 m
= ControlBase
._elaborate
(self
, platform
)
743 pipe1
= ExampleBufDelayedPipe()
744 pipe2
= ExampleBufPipe()
746 m
.submodules
.pipe1
= pipe1
747 m
.submodules
.pipe2
= pipe2
749 m
.d
.comb
+= self
.connect([pipe1
, pipe2
])
753 ######################################################################
754 # Test 999 - XXX FAILS
755 # http://bugs.libre-riscv.org/show_bug.cgi?id=57
756 ######################################################################
758 class ExampleBufAdd1Pipe(BufferedHandshake
):
761 stage
= ExampleStageCls()
762 BufferedHandshake
.__init
__(self
, stage
)
765 class ExampleUnBufAdd1Pipe(UnbufferedPipeline
):
768 stage
= ExampleStageCls()
769 UnbufferedPipeline
.__init
__(self
, stage
)
772 class ExampleBufUnBufPipe(ControlBase
):
774 def elaborate(self
, platform
):
775 m
= ControlBase
._elaborate
(self
, platform
)
777 # XXX currently fails: any other permutation works fine.
778 # p1=u,p2=b ok p1=u,p2=u ok p1=b,p2=b ok
779 # also fails using UnbufferedPipeline as well
780 #pipe1 = ExampleUnBufAdd1Pipe()
781 #pipe2 = ExampleBufAdd1Pipe()
782 pipe1
= ExampleBufAdd1Pipe()
783 pipe2
= ExampleUnBufAdd1Pipe()
785 m
.submodules
.pipe1
= pipe1
786 m
.submodules
.pipe2
= pipe2
788 m
.d
.comb
+= self
.connect([pipe1
, pipe2
])
793 ######################################################################
795 ######################################################################
799 if __name__
== '__main__':
801 dut
= ExampleBufPipe()
802 run_simulation(dut
, testbench(dut
), vcd_name
="test_bufpipe.vcd")
805 dut
= ExampleBufPipe2()
806 run_simulation(dut
, testbench2(dut
), vcd_name
="test_bufpipe2.vcd")
807 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
808 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
809 [dut
.p
.i_data
] + [dut
.n
.o_data
]
810 vl
= rtlil
.convert(dut
, ports
=ports
)
811 with
open("test_bufpipe2.il", "w") as f
:
816 dut
= ExampleBufPipe()
817 test
= Test3(dut
, test3_resultfn
)
818 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufpipe3.vcd")
821 dut
= ExamplePipeline()
822 test
= Test3(dut
, test3_resultfn
)
823 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_combpipe3.vcd")
826 dut
= ExampleBufPipe2()
827 run_simulation(dut
, testbench4(dut
), vcd_name
="test_bufpipe4.vcd")
830 dut
= ExampleBufPipeAdd()
831 test
= Test5(dut
, test5_resultfn
, stage_ctl
=True)
832 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufpipe5.vcd")
835 dut
= ExampleLTPipeline()
836 test
= Test5(dut
, test6_resultfn
)
837 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_ltcomb6.vcd")
839 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
840 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
841 list(dut
.p
.i_data
) + [dut
.n
.o_data
]
842 vl
= rtlil
.convert(dut
, ports
=ports
)
843 with
open("test_ltcomb_pipe.il", "w") as f
:
847 dut
= ExampleAddRecordPipe()
849 test
= Test5(dut
, test7_resultfn
, data
=data
)
850 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_addrecord.vcd")
852 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
853 dut
.n
.o_valid
, dut
.p
.o_ready
,
854 dut
.p
.i_data
.src1
, dut
.p
.i_data
.src2
,
855 dut
.n
.o_data
.src1
, dut
.n
.o_data
.src2
]
856 vl
= rtlil
.convert(dut
, ports
=ports
)
857 with
open("test_recordcomb_pipe.il", "w") as f
:
861 dut
= ExampleBufPipeAddClass()
863 test
= Test5(dut
, test8_resultfn
, data
=data
)
864 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufpipe8.vcd")
867 dut
= ExampleBufPipeChain2()
868 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
869 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
870 [dut
.p
.i_data
] + [dut
.n
.o_data
]
871 vl
= rtlil
.convert(dut
, ports
=ports
)
872 with
open("test_bufpipechain2.il", "w") as f
:
876 test
= Test5(dut
, test9_resultfn
, data
=data
)
877 run_simulation(dut
, [test
.send
, test
.rcv
],
878 vcd_name
="test_bufpipechain2.vcd")
881 dut
= ExampleLTBufferedPipeDerived()
882 test
= Test5(dut
, test6_resultfn
)
883 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_ltbufpipe10.vcd")
884 vl
= rtlil
.convert(dut
, ports
=ports
)
885 with
open("test_ltbufpipe10.il", "w") as f
:
889 dut
= ExampleAddRecordPlaceHolderPipe()
890 data
=data_placeholder()
891 test
= Test5(dut
, test11_resultfn
, data
=data
)
892 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_addrecord.vcd")
896 dut
= ExampleBufDelayedPipe()
898 test
= Test5(dut
, test12_resultfn
, data
=data
)
899 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufpipe12.vcd")
900 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
901 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
902 [dut
.p
.i_data
] + [dut
.n
.o_data
]
903 vl
= rtlil
.convert(dut
, ports
=ports
)
904 with
open("test_bufpipe12.il", "w") as f
:
908 dut
= ExampleUnBufDelayedPipe()
910 test
= Test5(dut
, test12_resultfn
, data
=data
)
911 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_unbufpipe13.vcd")
912 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
913 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
914 [dut
.p
.i_data
] + [dut
.n
.o_data
]
915 vl
= rtlil
.convert(dut
, ports
=ports
)
916 with
open("test_unbufpipe13.il", "w") as f
:
920 dut
= ExampleBufModeAdd1Pipe()
922 test
= Test5(dut
, test12_resultfn
, data
=data
)
923 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufunbuf15.vcd")
924 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
925 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
926 [dut
.p
.i_data
] + [dut
.n
.o_data
]
927 vl
= rtlil
.convert(dut
, ports
=ports
)
928 with
open("test_bufunbuf15.il", "w") as f
:
932 dut
= ExampleBufModeUnBufPipe()
934 test
= Test5(dut
, test9_resultfn
, data
=data
)
935 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufunbuf16.vcd")
936 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
937 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
938 [dut
.p
.i_data
] + [dut
.n
.o_data
]
939 vl
= rtlil
.convert(dut
, ports
=ports
)
940 with
open("test_bufunbuf16.il", "w") as f
:
944 dut
= ExampleUnBufAdd1Pipe2()
946 test
= Test5(dut
, test12_resultfn
, data
=data
)
947 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_unbufpipe17.vcd")
948 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
949 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
950 [dut
.p
.i_data
] + [dut
.n
.o_data
]
951 vl
= rtlil
.convert(dut
, ports
=ports
)
952 with
open("test_unbufpipe17.il", "w") as f
:
956 dut
= PassThroughTest()
958 test
= Test5(dut
, test_identical_resultfn
, data
=data
)
959 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_passthru18.vcd")
960 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
961 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
962 [dut
.p
.i_data
] + [dut
.n
.o_data
]
963 vl
= rtlil
.convert(dut
, ports
=ports
)
964 with
open("test_passthru18.il", "w") as f
:
967 print ("test 998 (fails, bug)")
968 dut
= ExampleBufPipe3()
970 test
= Test5(dut
, test9_resultfn
, data
=data
)
971 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufpipe14.vcd")
972 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
973 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
974 [dut
.p
.i_data
] + [dut
.n
.o_data
]
975 vl
= rtlil
.convert(dut
, ports
=ports
)
976 with
open("test_bufpipe14.il", "w") as f
:
979 print ("test 999 (expected to fail, which is a bug)")
980 dut
= ExampleBufUnBufPipe()
982 test
= Test5(dut
, test9_resultfn
, data
=data
)
983 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufunbuf999.vcd")
984 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
985 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
986 [dut
.p
.i_data
] + [dut
.n
.o_data
]
987 vl
= rtlil
.convert(dut
, ports
=ports
)
988 with
open("test_bufunbuf999.il", "w") as f
: