src/add/example_buf_pipe.py

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