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