src/soc/experiment/alu_fsm.py

   1 """Simple example of a FSM-based ALU
   2
   3 This demonstrates a design that follows the valid/ready protocol of the
   4 ALU, but with a FSM implementation, instead of a pipeline.  It is also
   5 intended to comply with both the CompALU API and the nmutil Pipeline API
   6 (Liskov Substitution Principle)
   7
   8 The basic rules are:
   9
  10 1) p.ready_o is asserted on the initial ("Idle") state, otherwise it keeps low.
  11 2) n.valid_o is asserted on the final ("Done") state, otherwise it keeps low.
  12 3) The FSM stays in the Idle state while p.valid_i is low, otherwise
  13    it accepts the input data and moves on.
  14 4) The FSM stays in the Done state while n.ready_i is low, otherwise
  15    it releases the output data and goes back to the Idle state.
  16
  17 """
  18
  19 from nmigen import Elaboratable, Signal, Module, Cat
  20 from nmigen.back.pysim import Simulator
  21 from nmigen.cli import rtlil
  22 from soc.fu.cr.cr_input_record import CompCROpSubset
  23 from math import log2
  24 from nmutil.iocontrol import PrevControl, NextControl
  25
  26
  27 class Dummy:
  28     pass
  29
  30
  31 class Shifter(Elaboratable):
  32     """Simple sequential shifter
  33
  34     Prev port data:
  35     * p.data_i.data:  value to be shifted
  36     * p.data_i.shift: shift amount
  37     *                 When zero, no shift occurs.
  38     *                 On POWER, range is 0 to 63 for 32-bit,
  39     *                 and 0 to 127 for 64-bit.
  40     *                 Other values wrap around.
  41     * p.data_i.dir:   shift direction (0 = left, 1 = right)
  42
  43     Next port data:
  44     * n.data_o.data: shifted value
  45     """
  46     class PrevData:
  47         def __init__(self, width):
  48             self.data = Signal(width, name="p_data_i")
  49             self.shift = Signal(width, name="p_shift_i")
  50             self.dir = Signal(name="p_dir_i")
  51             self.ctx = Dummy() # comply with CompALU API
  52
  53         def _get_data(self):
  54             return [self.data, self.shift]
  55
  56     class NextData:
  57         def __init__(self, width):
  58             self.data = Signal(width, name="n_data_o")
  59
  60         def _get_data(self):
  61             return [self.data]
  62
  63     def __init__(self, width):
  64         self.width = width
  65         self.p = PrevControl()
  66         self.n = NextControl()
  67         self.p.data_i = Shifter.PrevData(width)
  68         self.n.data_o = Shifter.NextData(width)
  69
  70         # more pieces to make this example class comply with the CompALU API
  71         self.op = CompCROpSubset()
  72         self.p.data_i.ctx.op = self.op
  73         self.i = self.p.data_i._get_data()
  74         self.out = self.n.data_o._get_data()
  75
  76     def elaborate(self, platform):
  77         m = Module()
  78
  79         m.submodules.p = self.p
  80         m.submodules.n = self.n
  81
  82         # Note:
  83         # It is good practice to design a sequential circuit as
  84         # a data path and a control path.
  85
  86         # Data path
  87         # ---------
  88         # The idea is to have a register that can be
  89         # loaded or shifted (left and right).
  90
  91         # the control signals
  92         load = Signal()
  93         shift = Signal()
  94         direction = Signal()
  95         # the data flow
  96         shift_in = Signal(self.width)
  97         shift_left_by_1 = Signal(self.width)
  98         shift_right_by_1 = Signal(self.width)
  99         next_shift = Signal(self.width)
 100         # the register
 101         shift_reg = Signal(self.width, reset_less=True)
 102         # build the data flow
 103         m.d.comb += [
 104             # connect input and output
 105             shift_in.eq(self.p.data_i.data),
 106             self.n.data_o.data.eq(shift_reg),
 107             # generate shifted views of the register
 108             shift_left_by_1.eq(Cat(0, shift_reg[:-1])),
 109             shift_right_by_1.eq(Cat(shift_reg[1:], 0)),
 110         ]
 111         # choose the next value of the register according to the
 112         # control signals
 113         # default is no change
 114         m.d.comb += next_shift.eq(shift_reg)
 115         with m.If(load):
 116             m.d.comb += next_shift.eq(shift_in)
 117         with m.Elif(shift):
 118             with m.If(direction):
 119                 m.d.comb += next_shift.eq(shift_right_by_1)
 120             with m.Else():
 121                 m.d.comb += next_shift.eq(shift_left_by_1)
 122
 123         # register the next value
 124         m.d.sync += shift_reg.eq(next_shift)
 125
 126         # Control path
 127         # ------------
 128         # The idea is to have a SHIFT state where the shift register
 129         # is shifted every cycle, while a counter decrements.
 130         # This counter is loaded with shift amount in the initial state.
 131         # The SHIFT state is left when the counter goes to zero.
 132
 133         # Shift counter
 134         shift_width = int(log2(self.width)) + 1
 135         next_count = Signal(shift_width)
 136         count = Signal(shift_width, reset_less=True)
 137         m.d.sync += count.eq(next_count)
 138
 139         with m.FSM():
 140             with m.State("IDLE"):
 141                 m.d.comb += [
 142                     # keep p.ready_o active on IDLE
 143                     self.p.ready_o.eq(1),
 144                     # keep loading the shift register and shift count
 145                     load.eq(1),
 146                     next_count.eq(self.p.data_i.shift),
 147                 ]
 148                 # capture the direction bit as well
 149                 m.d.sync += direction.eq(self.p.data_i.dir)
 150                 with m.If(self.p.valid_i):
 151                     # Leave IDLE when data arrives
 152                     with m.If(next_count == 0):
 153                         # short-circuit for zero shift
 154                         m.next = "DONE"
 155                     with m.Else():
 156                         m.next = "SHIFT"
 157             with m.State("SHIFT"):
 158                 m.d.comb += [
 159                     # keep shifting, while counter is not zero
 160                     shift.eq(1),
 161                     # decrement the shift counter
 162                     next_count.eq(count - 1),
 163                 ]
 164                 with m.If(next_count == 0):
 165                     # exit when shift counter goes to zero
 166                     m.next = "DONE"
 167             with m.State("DONE"):
 168                 # keep n.valid_o active while the data is not accepted
 169                 m.d.comb += self.n.valid_o.eq(1)
 170                 with m.If(self.n.ready_i):
 171                     # go back to IDLE when the data is accepted
 172                     m.next = "IDLE"
 173
 174         return m
 175
 176     def __iter__(self):
 177         yield self.p.data_i.data
 178         yield self.p.data_i.shift
 179         yield self.p.data_i.dir
 180         yield self.p.valid_i
 181         yield self.p.ready_o
 182         yield self.n.ready_i
 183         yield self.n.valid_o
 184         yield self.n.data_o.data
 185
 186     def ports(self):
 187         return list(self)
 188
 189
 190 def test_shifter():
 191     m = Module()
 192     m.submodules.shf = dut = Shifter(8)
 193     print("Shifter port names:")
 194     for port in dut:
 195         print("-", port.name)
 196     # generate RTLIL
 197     # try "proc; show" in yosys to check the data path
 198     il = rtlil.convert(dut, ports=dut.ports())
 199     with open("test_shifter.il", "w") as f:
 200         f.write(il)
 201     sim = Simulator(m)
 202     sim.add_clock(1e-6)
 203
 204     def send(data, shift, direction):
 205         # present input data and assert valid_i
 206         yield dut.p.data_i.data.eq(data)
 207         yield dut.p.data_i.shift.eq(shift)
 208         yield dut.p.data_i.dir.eq(direction)
 209         yield dut.p.valid_i.eq(1)
 210         yield
 211         # wait for p.ready_o to be asserted
 212         while not (yield dut.p.ready_o):
 213             yield
 214         # clear input data and negate p.valid_i
 215         yield dut.p.valid_i.eq(0)
 216         yield dut.p.data_i.data.eq(0)
 217         yield dut.p.data_i.shift.eq(0)
 218         yield dut.p.data_i.dir.eq(0)
 219
 220     def receive(expected):
 221         # signal readiness to receive data
 222         yield dut.n.ready_i.eq(1)
 223         yield
 224         # wait for n.valid_o to be asserted
 225         while not (yield dut.n.valid_o):
 226             yield
 227         # read result
 228         result = yield dut.n.data_o.data
 229         # negate n.ready_i
 230         yield dut.n.ready_i.eq(0)
 231         # check result
 232         assert result == expected
 233
 234     def producer():
 235         # 13 >> 2
 236         yield from send(13, 2, 1)
 237         # 3 << 4
 238         yield from send(3, 4, 0)
 239         # 21 << 0
 240         yield from send(21, 0, 0)
 241
 242     def consumer():
 243         # the consumer is not in step with the producer, but the
 244         # order of the results are preserved
 245         # 13 >> 2 = 3
 246         yield from receive(3)
 247         # 3 << 4 = 48
 248         yield from receive(48)
 249         # 21 << 0 = 21
 250         yield from receive(21)
 251
 252     sim.add_sync_process(producer)
 253     sim.add_sync_process(consumer)
 254     sim_writer = sim.write_vcd(
 255         "test_shifter.vcd",
 256         "test_shifter.gtkw",
 257         traces=dut.ports()
 258     )
 259     with sim_writer:
 260         sim.run()
 261
 262
 263 if __name__ == "__main__":
 264     test_shifter()