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