src/add/example_buf_pipe.py

   1 """ nmigen implementation of buffered pipeline stage, based on zipcpu:
   2     https://zipcpu.com/blog/2017/08/14/strategies-for-pipelining.html
   3 """
   4 from nmigen import Signal, Cat, Const, Mux, Module
   5 from nmigen.compat.sim import run_simulation
   6 from nmigen.cli import verilog, rtlil
   7
   8 class BufPipe:
   9     def __init__(self):
  10         # input
  11         self.i_p_stb = Signal()    # >>in - comes in from PREVIOUS stage
  12         self.i_n_busy = Signal()   # in<< - comes in from the NEXT stage
  13         self.i_data = Signal(32) # >>in - comes in from the PREVIOUS stage
  14         #self.i_rst = Signal()
  15
  16         # buffered
  17         self.r_data = Signal(32)
  18
  19         # output
  20         self.o_n_stb = Signal()    # out>> - goes out to the NEXT stage
  21         self.o_p_busy = Signal()   # <<out - goes out to the PREVIOUS stage
  22         self.o_data = Signal(32) # out>> - goes out to the NEXT stage
  23
  24     def pre_process(self, d_in):
  25         return d_in
  26
  27     def process(self, d_in):
  28         return d_in + 1
  29
  30     def elaborate(self, platform):
  31         m = Module()
  32
  33         i_p_stb_o_p_busyn = Signal(reset_less=True)
  34         m.d.comb += i_p_stb_o_p_busyn.eq(self.i_p_stb & (~self.o_p_busy))
  35
  36         with m.If(~self.i_n_busy): # previous stage is not busy
  37             with m.If(~self.o_p_busy): # not stalled
  38                 # nothing in buffer: send input direct to output
  39                 m.d.sync += self.o_n_stb.eq(self.i_p_stb)
  40                 m.d.sync += self.o_data.eq(self.process(self.i_data))
  41             with m.Else(): # o_p_busy is true, and something is in our buffer.
  42                 # Flush the buffer to the output port.
  43                 m.d.sync += self.o_n_stb.eq(1)
  44                 m.d.sync += self.o_data.eq(self.r_data)
  45                 # ignore input, since o_p_busy is also true.
  46             # also clear stall condition, declare register to be empty.
  47             m.d.sync += self.o_p_busy.eq(0)
  48
  49         # (i_n_busy) is true here: previous stage is busy
  50         with m.Elif(~self.o_n_stb): # next stage being told "not busy"
  51             m.d.sync += self.o_n_stb.eq(self.i_p_stb)
  52             m.d.sync += self.o_p_busy.eq(0) # Keep the buffer empty
  53             # Apply the logic to the input data, and set the output data
  54             m.d.sync += self.o_data.eq(self.process(self.i_data))
  55
  56         # (i_n_busy) and (o_n_stb) both true:
  57         with m.Elif(i_p_stb_o_p_busyn):
  58             # If next stage *is* busy, and not stalled yet, accept requested
  59             # input and store in temporary
  60             m.d.sync += self.o_p_busy.eq(self.i_p_stb & self.o_n_stb)
  61             with m.If(~self.o_n_stb):
  62                 m.d.sync += self.r_data.eq(self.i_data)
  63
  64         with m.If(~self.o_p_busy): # not stalled
  65             m.d.sync += self.r_data.eq(self.pre_process(self.i_data))
  66
  67         return m
  68
  69     def ports(self):
  70         return [self.i_p_stb, self.i_n_busy, self.i_data,
  71                 self.r_data,
  72                 self.o_n_stb, self.o_p_busy, self.o_data
  73                ]
  74
  75
  76 def testbench(dut):
  77     yield dut.i_data.eq(5)
  78     yield dut.i_p_stb.eq(1)
  79     yield
  80     yield dut.i_data.eq(7)
  81     yield
  82     yield dut.i_data.eq(2)
  83     yield
  84     yield dut.i_n_busy.eq(1)
  85     yield dut.i_data.eq(9)
  86     yield
  87     yield dut.i_data.eq(12)
  88     yield
  89     yield dut.i_n_busy.eq(0)
  90     yield
  91     yield
  92     yield
  93
  94
  95 if __name__ == '__main__':
  96     dut = BufPipe()
  97     vl = rtlil.convert(dut, ports=dut.ports())
  98     with open("test_bufpipe.il", "w") as f:
  99         f.write(vl)
 100     run_simulation(dut, testbench(dut), vcd_name="test_bufpipe.vcd")
 101