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
, BufferedPipeline
26 from example_buf_pipe
import StageChain
, ControlBase
, StageCls
28 from random
import randint
31 def check_o_n_valid(dut
, val
):
32 o_n_valid
= yield dut
.n
.o_valid
33 assert o_n_valid
== val
35 def check_o_n_valid2(dut
, val
):
36 o_n_valid
= yield dut
.n
.o_valid
37 assert o_n_valid
== val
41 #yield dut.i_p_rst.eq(1)
42 yield dut
.n
.i_ready
.eq(0)
43 yield dut
.p
.o_ready
.eq(0)
46 #yield dut.i_p_rst.eq(0)
47 yield dut
.n
.i_ready
.eq(1)
48 yield dut
.p
.i_data
.eq(5)
49 yield dut
.p
.i_valid
.eq(1)
52 yield dut
.p
.i_data
.eq(7)
53 yield from check_o_n_valid(dut
, 0) # effects of i_p_valid delayed
55 yield from check_o_n_valid(dut
, 1) # ok *now* i_p_valid effect is felt
57 yield dut
.p
.i_data
.eq(2)
59 yield dut
.n
.i_ready
.eq(0) # begin going into "stall" (next stage says ready)
60 yield dut
.p
.i_data
.eq(9)
62 yield dut
.p
.i_valid
.eq(0)
63 yield dut
.p
.i_data
.eq(12)
65 yield dut
.p
.i_data
.eq(32)
66 yield dut
.n
.i_ready
.eq(1)
68 yield from check_o_n_valid(dut
, 1) # buffer still needs to output
70 yield from check_o_n_valid(dut
, 1) # buffer still needs to output
72 yield from check_o_n_valid(dut
, 0) # buffer outputted, *now* we're done.
77 #yield dut.p.i_rst.eq(1)
78 yield dut
.n
.i_ready
.eq(0)
79 #yield dut.p.o_ready.eq(0)
82 #yield dut.p.i_rst.eq(0)
83 yield dut
.n
.i_ready
.eq(1)
84 yield dut
.p
.i_data
.eq(5)
85 yield dut
.p
.i_valid
.eq(1)
88 yield dut
.p
.i_data
.eq(7)
89 yield from check_o_n_valid2(dut
, 0) # effects of i_p_valid delayed 2 clocks
91 yield from check_o_n_valid2(dut
, 0) # effects of i_p_valid delayed 2 clocks
93 yield dut
.p
.i_data
.eq(2)
95 yield from check_o_n_valid2(dut
, 1) # ok *now* i_p_valid effect is felt
96 yield dut
.n
.i_ready
.eq(0) # begin going into "stall" (next stage says ready)
97 yield dut
.p
.i_data
.eq(9)
99 yield dut
.p
.i_valid
.eq(0)
100 yield dut
.p
.i_data
.eq(12)
102 yield dut
.p
.i_data
.eq(32)
103 yield dut
.n
.i_ready
.eq(1)
105 yield from check_o_n_valid2(dut
, 1) # buffer still needs to output
107 yield from check_o_n_valid2(dut
, 1) # buffer still needs to output
109 yield from check_o_n_valid2(dut
, 1) # buffer still needs to output
111 yield from check_o_n_valid2(dut
, 0) # buffer outputted, *now* we're done.
118 def __init__(self
, dut
, resultfn
):
120 self
.resultfn
= resultfn
122 for i
in range(num_tests
):
123 #data.append(randint(0, 1<<16-1))
124 self
.data
.append(i
+1)
129 while self
.o
!= len(self
.data
):
130 send_range
= randint(0, 3)
131 for j
in range(randint(1,10)):
135 send
= randint(0, send_range
) != 0
136 o_p_ready
= yield self
.dut
.p
.o_ready
140 if send
and self
.i
!= len(self
.data
):
141 yield self
.dut
.p
.i_valid
.eq(1)
142 yield self
.dut
.p
.i_data
.eq(self
.data
[self
.i
])
145 yield self
.dut
.p
.i_valid
.eq(0)
149 while self
.o
!= len(self
.data
):
150 stall_range
= randint(0, 3)
151 for j
in range(randint(1,10)):
152 stall
= randint(0, stall_range
) != 0
153 yield self
.dut
.n
.i_ready
.eq(stall
)
155 o_n_valid
= yield self
.dut
.n
.o_valid
156 i_n_ready
= yield self
.dut
.n
.i_ready
157 if not o_n_valid
or not i_n_ready
:
159 o_data
= yield self
.dut
.n
.o_data
160 self
.resultfn(o_data
, self
.data
[self
.o
], self
.i
, self
.o
)
162 if self
.o
== len(self
.data
):
165 def test3_resultfn(o_data
, expected
, i
, o
):
166 assert o_data
== expected
+ 1, \
167 "%d-%d data %x not match %x\n" \
168 % (i
, o
, o_data
, expected
)
170 def data_placeholder():
172 for i
in range(num_tests
):
174 d
.src1
= randint(0, 1<<16-1)
175 d
.src2
= randint(0, 1<<16-1)
181 for i
in range(num_tests
):
182 data
.append({'src1': randint(0, 1<<16-1),
183 'src2': randint(0, 1<<16-1)})
188 def __init__(self
, dut
, resultfn
, data
=None):
190 self
.resultfn
= resultfn
195 for i
in range(num_tests
):
196 self
.data
.append((randint(0, 1<<16-1), randint(0, 1<<16-1)))
201 while self
.o
!= len(self
.data
):
202 send_range
= randint(0, 3)
203 for j
in range(randint(1,10)):
207 send
= randint(0, send_range
) != 0
209 o_p_ready
= yield self
.dut
.p
.o_ready
213 if send
and self
.i
!= len(self
.data
):
214 yield self
.dut
.p
.i_valid
.eq(1)
215 for v
in self
.dut
.set_input(self
.data
[self
.i
]):
219 yield self
.dut
.p
.i_valid
.eq(0)
223 while self
.o
!= len(self
.data
):
224 stall_range
= randint(0, 3)
225 for j
in range(randint(1,10)):
226 stall
= randint(0, stall_range
) != 0
227 yield self
.dut
.n
.i_ready
.eq(stall
)
229 o_n_valid
= yield self
.dut
.n
.o_valid
230 i_n_ready
= yield self
.dut
.n
.i_ready
231 if not o_n_valid
or not i_n_ready
:
233 if isinstance(self
.dut
.n
.o_data
, Record
):
235 dod
= self
.dut
.n
.o_data
236 for k
, v
in dod
.fields
.items():
239 o_data
= yield self
.dut
.n
.o_data
240 self
.resultfn(o_data
, self
.data
[self
.o
], self
.i
, self
.o
)
242 if self
.o
== len(self
.data
):
245 def test5_resultfn(o_data
, expected
, i
, o
):
246 res
= expected
[0] + expected
[1]
247 assert o_data
== res
, \
248 "%d-%d data %x not match %s\n" \
249 % (i
, o
, o_data
, repr(expected
))
253 for i
in range(num_tests
):
254 #data.append(randint(0, 1<<16-1))
259 stall
= randint(0, 3) != 0
260 send
= randint(0, 5) != 0
261 yield dut
.n
.i_ready
.eq(stall
)
262 o_p_ready
= yield dut
.p
.o_ready
264 if send
and i
!= len(data
):
265 yield dut
.p
.i_valid
.eq(1)
266 yield dut
.p
.i_data
.eq(data
[i
])
269 yield dut
.p
.i_valid
.eq(0)
271 o_n_valid
= yield dut
.n
.o_valid
272 i_n_ready
= yield dut
.n
.i_ready
273 if o_n_valid
and i_n_ready
:
274 o_data
= yield dut
.n
.o_data
275 assert o_data
== data
[o
] + 2, "%d-%d data %x not match %x\n" \
276 % (i
, o
, o_data
, data
[o
])
281 ######################################################################
283 ######################################################################
285 class ExampleBufPipe2(ControlBase
):
286 """ Example of how to do chained pipeline stages.
289 def elaborate(self
, platform
):
292 pipe1
= ExampleBufPipe()
293 pipe2
= ExampleBufPipe()
295 m
.submodules
.pipe1
= pipe1
296 m
.submodules
.pipe2
= pipe2
298 m
.d
.comb
+= self
.connect([pipe1
, pipe2
])
303 ######################################################################
305 ######################################################################
307 class ExampleBufPipeChain2(BufferedPipeline
):
308 """ connects two stages together as a *single* combinatorial stage.
311 stage1
= ExampleStageCls()
312 stage2
= ExampleStageCls()
313 combined
= StageChain([stage1
, stage2
])
314 BufferedPipeline
.__init
__(self
, combined
)
319 for i
in range(num_tests
):
320 data
.append(randint(0, 1<<16-2))
324 def test9_resultfn(o_data
, expected
, i
, o
):
326 assert o_data
== res
, \
327 "%d-%d data %x not match %s\n" \
328 % (i
, o
, o_data
, repr(expected
))
331 ######################################################################
333 ######################################################################
336 def __init__(self
, width
, signed
):
338 self
.src1
= Signal((width
, signed
), name
="src1")
339 self
.src2
= Signal((width
, signed
), name
="src2")
340 self
.output
= Signal(width
, name
="out")
342 def elaborate(self
, platform
):
343 self
.m
.d
.comb
+= self
.output
.eq(Mux(self
.src1
< self
.src2
, 1, 0))
347 class LTStage(StageCls
):
348 """ module-based stage example
351 self
.slt
= SetLessThan(16, True)
354 return (Signal(16, name
="sig1"), Signal(16, "sig2"))
357 return Signal(16, "out")
359 def setup(self
, m
, i
):
361 m
.submodules
.slt
= self
.slt
362 m
.d
.comb
+= self
.slt
.src1
.eq(i
[0])
363 m
.d
.comb
+= self
.slt
.src2
.eq(i
[1])
364 m
.d
.comb
+= self
.o
.eq(self
.slt
.output
)
366 def process(self
, i
):
370 class LTStageDerived(SetLessThan
, StageCls
):
371 """ special version of a nmigen module where the module is also a stage
373 shows that you don't actually need to combinatorially connect
374 to the outputs, or add the module as a submodule: just return
375 the module output parameter(s) from the Stage.process() function
379 SetLessThan
.__init
__(self
, 16, True)
382 return (Signal(16), Signal(16))
387 def setup(self
, m
, i
):
388 m
.submodules
.slt
= self
389 m
.d
.comb
+= self
.src1
.eq(i
[0])
390 m
.d
.comb
+= self
.src2
.eq(i
[1])
392 def process(self
, i
):
396 class ExampleLTPipeline(UnbufferedPipeline
):
397 """ an example of how to use the unbuffered pipeline.
402 UnbufferedPipeline
.__init
__(self
, stage
)
405 class ExampleLTBufferedPipeDerived(BufferedPipeline
):
406 """ an example of how to use the buffered pipeline.
410 stage
= LTStageDerived()
411 BufferedPipeline
.__init
__(self
, stage
)
414 def test6_resultfn(o_data
, expected
, i
, o
):
415 res
= 1 if expected
[0] < expected
[1] else 0
416 assert o_data
== res
, \
417 "%d-%d data %x not match %s\n" \
418 % (i
, o
, o_data
, repr(expected
))
421 ######################################################################
423 ######################################################################
425 class ExampleAddRecordStage(StageCls
):
426 """ example use of a Record
429 record_spec
= [('src1', 16), ('src2', 16)]
431 """ returns a Record using the specification
433 return Record(self
.record_spec
)
436 return Record(self
.record_spec
)
438 def process(self
, i
):
439 """ process the input data, returning a dictionary with key names
440 that exactly match the Record's attributes.
442 return {'src1': i
.src1
+ 1,
445 ######################################################################
447 ######################################################################
449 class ExampleAddRecordPlaceHolderStage(StageCls
):
450 """ example use of a Record, with a placeholder as the processing result
453 record_spec
= [('src1', 16), ('src2', 16)]
455 """ returns a Record using the specification
457 return Record(self
.record_spec
)
460 return Record(self
.record_spec
)
462 def process(self
, i
):
463 """ process the input data, returning a PlaceHolder class instance
464 with attributes that exactly match those of the Record.
472 class PlaceHolder
: pass
475 class ExampleAddRecordPipe(UnbufferedPipeline
):
476 """ an example of how to use the combinatorial pipeline.
480 stage
= ExampleAddRecordStage()
481 UnbufferedPipeline
.__init
__(self
, stage
)
484 def test7_resultfn(o_data
, expected
, i
, o
):
485 res
= (expected
['src1'] + 1, expected
['src2'] + 1)
486 assert o_data
['src1'] == res
[0] and o_data
['src2'] == res
[1], \
487 "%d-%d data %s not match %s\n" \
488 % (i
, o
, repr(o_data
), repr(expected
))
491 class ExampleAddRecordPlaceHolderPipe(UnbufferedPipeline
):
492 """ an example of how to use the combinatorial pipeline.
496 stage
= ExampleAddRecordPlaceHolderStage()
497 UnbufferedPipeline
.__init
__(self
, stage
)
500 def test11_resultfn(o_data
, expected
, i
, o
):
501 res1
= expected
.src1
+ 1
502 res2
= expected
.src2
+ 1
503 assert o_data
['src1'] == res1
and o_data
['src2'] == res2
, \
504 "%d-%d data %s not match %s\n" \
505 % (i
, o
, repr(o_data
), repr(expected
))
508 ######################################################################
510 ######################################################################
513 class Example2OpClass
:
514 """ an example of a class used to store 2 operands.
515 requires an eq function, to conform with the pipeline stage API
519 self
.op1
= Signal(16)
520 self
.op2
= Signal(16)
523 return [self
.op1
.eq(i
.op1
), self
.op2
.eq(i
.op2
)]
526 class ExampleAddClassStage(StageCls
):
527 """ an example of how to use the buffered pipeline, as a class instance
531 """ returns an instance of an Example2OpClass.
533 return Example2OpClass()
536 """ returns an output signal which will happen to contain the sum
541 def process(self
, i
):
542 """ process the input data (sums the values in the tuple) and returns it
547 class ExampleBufPipeAddClass(BufferedPipeline
):
548 """ an example of how to use the buffered pipeline, using a class instance
552 addstage
= ExampleAddClassStage()
553 BufferedPipeline
.__init
__(self
, addstage
)
557 """ the eq function, called by set_input, needs an incoming object
558 that conforms to the Example2OpClass.eq function requirements
559 easiest way to do that is to create a class that has the exact
560 same member layout (self.op1, self.op2) as Example2OpClass
562 def __init__(self
, op1
, op2
):
567 def test8_resultfn(o_data
, expected
, i
, o
):
568 res
= expected
.op1
+ expected
.op2
# these are a TestInputAdd instance
569 assert o_data
== res
, \
570 "%d-%d data %x not match %s\n" \
571 % (i
, o
, o_data
, repr(expected
))
575 for i
in range(num_tests
):
576 data
.append(TestInputAdd(randint(0, 1<<16-1), randint(0, 1<<16-1)))
580 ######################################################################
582 ######################################################################
584 class ExampleStageDelayCls(StageCls
):
585 """ an example of how to use the buffered pipeline, in a static class
590 self
.count
= Signal(2)
593 return Signal(16, name
="example_input_signal")
596 return Signal(16, name
="example_output_signal")
600 return self
.count
== 0
606 def process(self
, i
):
607 """ process the input data and returns it (adds 1)
611 def elaborate(self
, platform
):
613 m
.d
.sync
+= self
.count
.eq(self
.count
+ 1)
617 class ExampleBufDelayedPipe(BufferedPipeline
):
618 """ an example of how to use the buffered pipeline.
622 stage
= ExampleStageDelayCls()
623 BufferedPipeline
.__init
__(self
, stage
, stage_ctl
=True)
625 def elaborate(self
, platform
):
626 m
= BufferedPipeline
.elaborate(self
, platform
)
627 m
.submodules
.stage
= self
.stage
631 class ExampleBufPipe3(ControlBase
):
632 """ Example of how to do delayed pipeline, where the stage signals
636 def elaborate(self
, platform
):
637 m
= ControlBase
._elaborate
(self
, platform
)
639 pipe1
= ExampleBufPipe()
640 pipe2
= ExampleBufDelayedPipe()
642 m
.submodules
.pipe1
= pipe1
643 m
.submodules
.pipe2
= pipe2
645 m
.d
.comb
+= self
.connect([pipe1
, pipe2
])
651 for i
in range(num_tests
):
652 data
.append(randint(0, 1<<16-2))
656 def test12_resultfn(o_data
, expected
, i
, o
):
658 assert o_data
== res
, \
659 "%d-%d data %x not match %s\n" \
660 % (i
, o
, o_data
, repr(expected
))
666 if __name__
== '__main__':
668 dut
= ExampleBufPipe()
669 run_simulation(dut
, testbench(dut
), vcd_name
="test_bufpipe.vcd")
672 dut
= ExampleBufPipe2()
673 run_simulation(dut
, testbench2(dut
), vcd_name
="test_bufpipe2.vcd")
674 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
675 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
676 [dut
.p
.i_data
] + [dut
.n
.o_data
]
677 vl
= rtlil
.convert(dut
, ports
=ports
)
678 with
open("test_bufpipe2.il", "w") as f
:
683 dut
= ExampleBufPipe()
684 test
= Test3(dut
, test3_resultfn
)
685 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufpipe3.vcd")
688 dut
= ExamplePipeline()
689 test
= Test3(dut
, test3_resultfn
)
690 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_combpipe3.vcd")
693 dut
= ExampleBufPipe2()
694 run_simulation(dut
, testbench4(dut
), vcd_name
="test_bufpipe4.vcd")
697 dut
= ExampleBufPipeAdd()
698 test
= Test5(dut
, test5_resultfn
)
699 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufpipe5.vcd")
702 dut
= ExampleLTPipeline()
703 test
= Test5(dut
, test6_resultfn
)
704 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_ltcomb6.vcd")
706 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
707 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
708 list(dut
.p
.i_data
) + [dut
.n
.o_data
]
709 vl
= rtlil
.convert(dut
, ports
=ports
)
710 with
open("test_ltcomb_pipe.il", "w") as f
:
714 dut
= ExampleAddRecordPipe()
716 test
= Test5(dut
, test7_resultfn
, data
=data
)
717 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_addrecord.vcd")
719 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
720 dut
.n
.o_valid
, dut
.p
.o_ready
,
721 dut
.p
.i_data
.src1
, dut
.p
.i_data
.src2
,
722 dut
.n
.o_data
.src1
, dut
.n
.o_data
.src2
]
723 vl
= rtlil
.convert(dut
, ports
=ports
)
724 with
open("test_recordcomb_pipe.il", "w") as f
:
728 dut
= ExampleBufPipeAddClass()
730 test
= Test5(dut
, test8_resultfn
, data
=data
)
731 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufpipe8.vcd")
734 dut
= ExampleBufPipeChain2()
735 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
736 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
737 [dut
.p
.i_data
] + [dut
.n
.o_data
]
738 vl
= rtlil
.convert(dut
, ports
=ports
)
739 with
open("test_bufpipechain2.il", "w") as f
:
743 test
= Test5(dut
, test9_resultfn
, data
=data
)
744 run_simulation(dut
, [test
.send
, test
.rcv
],
745 vcd_name
="test_bufpipechain2.vcd")
748 dut
= ExampleLTBufferedPipeDerived()
749 test
= Test5(dut
, test6_resultfn
)
750 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_ltbufpipe10.vcd")
751 vl
= rtlil
.convert(dut
, ports
=ports
)
752 with
open("test_ltbufpipe10.il", "w") as f
:
756 dut
= ExampleAddRecordPlaceHolderPipe()
757 data
=data_placeholder()
758 test
= Test5(dut
, test11_resultfn
, data
=data
)
759 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_addrecord.vcd")
763 #dut = ExampleBufPipe3()
764 dut
= ExampleBufDelayedPipe()
766 test
= Test5(dut
, test12_resultfn
, data
=data
)
767 run_simulation(dut
, [test
.send
, test
.rcv
], vcd_name
="test_bufpipe12.vcd")
768 ports
= [dut
.p
.i_valid
, dut
.n
.i_ready
,
769 dut
.n
.o_valid
, dut
.p
.o_ready
] + \
770 [dut
.p
.i_data
] + [dut
.n
.o_data
]
771 vl
= rtlil
.convert(dut
, ports
=ports
)
772 with
open("test_bufpipe12.il", "w") as f
: