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