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.cli import rtlil
  21 from math import log2
  22
  23 from nmutil.iocontrol import PrevControl, NextControl
  24
  25 from soc.fu.base_input_record import CompOpSubsetBase
  26
  27 from nmutil.gtkw import write_gtkw
  28 from nmutil.sim_tmp_alternative import (Simulator, is_engine_pysim)
  29
  30
  31 class CompFSMOpSubset(CompOpSubsetBase):
  32     def __init__(self, name=None):
  33         layout = (('sdir', 1),
  34                   )
  35
  36         super().__init__(layout, name=name)
  37
  38
  39 class Dummy:
  40     pass
  41
  42
  43 class Shifter(Elaboratable):
  44     """Simple sequential shifter
  45
  46     Prev port data:
  47     * p.data_i.data:  value to be shifted
  48     * p.data_i.shift: shift amount
  49     *                 When zero, no shift occurs.
  50     *                 On POWER, range is 0 to 63 for 32-bit,
  51     *                 and 0 to 127 for 64-bit.
  52     *                 Other values wrap around.
  53
  54     Operation type
  55     * op.sdir:       shift direction (0 = left, 1 = right)
  56
  57     Next port data:
  58     * n.data_o.data: shifted value
  59     """
  60     class PrevData:
  61         def __init__(self, width):
  62             self.data = Signal(width, name="p_data_i")
  63             self.shift = Signal(width, name="p_shift_i")
  64             self.ctx = Dummy()  # comply with CompALU API
  65
  66         def _get_data(self):
  67             return [self.data, self.shift]
  68
  69     class NextData:
  70         def __init__(self, width):
  71             self.data = Signal(width, name="n_data_o")
  72
  73         def _get_data(self):
  74             return [self.data]
  75
  76     def __init__(self, width):
  77         self.width = width
  78         self.p = PrevControl()
  79         self.n = NextControl()
  80         self.p.data_i = Shifter.PrevData(width)
  81         self.n.data_o = Shifter.NextData(width)
  82
  83         # more pieces to make this example class comply with the CompALU API
  84         self.op = CompFSMOpSubset(name="op")
  85         self.p.data_i.ctx.op = self.op
  86         self.i = self.p.data_i._get_data()
  87         self.out = self.n.data_o._get_data()
  88
  89     def elaborate(self, platform):
  90         m = Module()
  91
  92         m.submodules.p = self.p
  93         m.submodules.n = self.n
  94
  95         # Note:
  96         # It is good practice to design a sequential circuit as
  97         # a data path and a control path.
  98
  99         # Data path
 100         # ---------
 101         # The idea is to have a register that can be
 102         # loaded or shifted (left and right).
 103
 104         # the control signals
 105         load = Signal()
 106         shift = Signal()
 107         direction = Signal()
 108         # the data flow
 109         shift_in = Signal(self.width)
 110         shift_left_by_1 = Signal(self.width)
 111         shift_right_by_1 = Signal(self.width)
 112         next_shift = Signal(self.width)
 113         # the register
 114         shift_reg = Signal(self.width, reset_less=True)
 115         # build the data flow
 116         m.d.comb += [
 117             # connect input and output
 118             shift_in.eq(self.p.data_i.data),
 119             self.n.data_o.data.eq(shift_reg),
 120             # generate shifted views of the register
 121             shift_left_by_1.eq(Cat(0, shift_reg[:-1])),
 122             shift_right_by_1.eq(Cat(shift_reg[1:], 0)),
 123         ]
 124         # choose the next value of the register according to the
 125         # control signals
 126         # default is no change
 127         m.d.comb += next_shift.eq(shift_reg)
 128         with m.If(load):
 129             m.d.comb += next_shift.eq(shift_in)
 130         with m.Elif(shift):
 131             with m.If(direction):
 132                 m.d.comb += next_shift.eq(shift_right_by_1)
 133             with m.Else():
 134                 m.d.comb += next_shift.eq(shift_left_by_1)
 135
 136         # register the next value
 137         m.d.sync += shift_reg.eq(next_shift)
 138
 139         # Control path
 140         # ------------
 141         # The idea is to have a SHIFT state where the shift register
 142         # is shifted every cycle, while a counter decrements.
 143         # This counter is loaded with shift amount in the initial state.
 144         # The SHIFT state is left when the counter goes to zero.
 145
 146         # Shift counter
 147         shift_width = int(log2(self.width)) + 1
 148         next_count = Signal(shift_width)
 149         count = Signal(shift_width, reset_less=True)
 150         m.d.sync += count.eq(next_count)
 151
 152         with m.FSM():
 153             with m.State("IDLE"):
 154                 m.d.comb += [
 155                     # keep p.ready_o active on IDLE
 156                     self.p.ready_o.eq(1),
 157                     # keep loading the shift register and shift count
 158                     load.eq(1),
 159                     next_count.eq(self.p.data_i.shift),
 160                 ]
 161                 # capture the direction bit as well
 162                 m.d.sync += direction.eq(self.op.sdir)
 163                 with m.If(self.p.valid_i):
 164                     # Leave IDLE when data arrives
 165                     with m.If(next_count == 0):
 166                         # short-circuit for zero shift
 167                         m.next = "DONE"
 168                     with m.Else():
 169                         m.next = "SHIFT"
 170             with m.State("SHIFT"):
 171                 m.d.comb += [
 172                     # keep shifting, while counter is not zero
 173                     shift.eq(1),
 174                     # decrement the shift counter
 175                     next_count.eq(count - 1),
 176                 ]
 177                 with m.If(next_count == 0):
 178                     # exit when shift counter goes to zero
 179                     m.next = "DONE"
 180             with m.State("DONE"):
 181                 # keep n.valid_o active while the data is not accepted
 182                 m.d.comb += self.n.valid_o.eq(1)
 183                 with m.If(self.n.ready_i):
 184                     # go back to IDLE when the data is accepted
 185                     m.next = "IDLE"
 186
 187         return m
 188
 189     def __iter__(self):
 190         yield self.op.sdir
 191         yield self.p.data_i.data
 192         yield self.p.data_i.shift
 193         yield self.p.valid_i
 194         yield self.p.ready_o
 195         yield self.n.ready_i
 196         yield self.n.valid_o
 197         yield self.n.data_o.data
 198
 199     def ports(self):
 200         return list(self)
 201
 202
 203 def test_shifter():
 204     m = Module()
 205     m.submodules.shf = dut = Shifter(8)
 206     print("Shifter port names:")
 207     for port in dut:
 208         print("-", port.name)
 209     # generate RTLIL
 210     # try "proc; show" in yosys to check the data path
 211     il = rtlil.convert(dut, ports=dut.ports())
 212     with open("test_shifter.il", "w") as f:
 213         f.write(il)
 214
 215     gtkwave_style = {
 216         'in': {'color': 'orange'},
 217         'out': {'color': 'yellow'},
 218     }
 219
 220     gtkwave_desc = [
 221         'clk',
 222         {'comment': 'Shifter Demonstration'},
 223         ('prev port', [
 224             ('op__sdir', 'in'),
 225             ('p_data_i[7:0]', 'in'),
 226             ('p_shift_i[7:0]', 'in'),
 227             ({'submodule': 'p'}, [
 228                 ('p_valid_i', 'in'),
 229                 ('p_ready_o', 'out')])]),
 230         ('internal', [
 231             'fsm_state' if is_engine_pysim() else 'fsm_state[1:0]',
 232             'count[3:0]',
 233             'shift_reg[7:0]']),
 234         ('next port', [
 235             ('n_data_o[7:0]', 'out'),
 236             ({'submodule': 'n'}, [
 237                 ('n_valid_o', 'out'),
 238                 ('n_ready_i', 'in')])])]
 239
 240     write_gtkw("test_shifter.gtkw", "test_shifter.vcd",
 241                gtkwave_desc,  gtkwave_style,
 242                module='top.shf', loc=__file__, base='dec')
 243
 244     sim = Simulator(m)
 245     sim.add_clock(1e-6)
 246
 247     def send(data, shift, direction):
 248         # present input data and assert valid_i
 249         yield dut.p.data_i.data.eq(data)
 250         yield dut.p.data_i.shift.eq(shift)
 251         yield dut.op.sdir.eq(direction)
 252         yield dut.p.valid_i.eq(1)
 253         yield
 254         # wait for p.ready_o to be asserted
 255         while not (yield dut.p.ready_o):
 256             yield
 257         # clear input data and negate p.valid_i
 258         yield dut.p.valid_i.eq(0)
 259         yield dut.p.data_i.data.eq(0)
 260         yield dut.p.data_i.shift.eq(0)
 261         yield dut.op.sdir.eq(0)
 262
 263     def receive(expected):
 264         # signal readiness to receive data
 265         yield dut.n.ready_i.eq(1)
 266         yield
 267         # wait for n.valid_o to be asserted
 268         while not (yield dut.n.valid_o):
 269             yield
 270         # read result
 271         result = yield dut.n.data_o.data
 272         # negate n.ready_i
 273         yield dut.n.ready_i.eq(0)
 274         # check result
 275         assert result == expected
 276
 277     def producer():
 278         # 13 >> 2
 279         yield from send(13, 2, 1)
 280         # 3 << 4
 281         yield from send(3, 4, 0)
 282         # 21 << 0
 283         yield from send(21, 0, 0)
 284
 285     def consumer():
 286         # the consumer is not in step with the producer, but the
 287         # order of the results are preserved
 288         # 13 >> 2 = 3
 289         yield from receive(3)
 290         # 3 << 4 = 48
 291         yield from receive(48)
 292         # 21 << 0 = 21
 293         yield from receive(21)
 294
 295     sim.add_sync_process(producer)
 296     sim.add_sync_process(consumer)
 297     sim_writer = sim.write_vcd("test_shifter.vcd")
 298     with sim_writer:
 299         sim.run()
 300
 301
 302 if __name__ == "__main__":
 303     test_shifter()