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_v1(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     style_debug = GTKWColor.red
 216     with open(base_name + ".gtkw", "wt") as gtkw_file:
 217         gtkw = GTKWSave(gtkw_file)
 218         gtkw.comment("Auto-generated by " + loc)
 219         gtkw.dumpfile(base_name + ".vcd")
 220         # set a reasonable zoom level
 221         # also, move the marker to an interesting place
 222         gtkw.zoom_markers(-22.9, 10500000)
 223         gtkw.trace(dut + "clk")
 224         # place a comment in the signal names panel
 225         gtkw.blank("Shifter Demonstration")
 226         with gtkw.group("prev port"):
 227             gtkw.trace(dut + "op__sdir", color=style_input)
 228             # demonstrates using decimal base (default is hex)
 229             gtkw.trace(dut + "p_data_i[7:0]", color=style_input,
 230                        datafmt='dec')
 231             gtkw.trace(dut + "p_shift_i[7:0]", color=style_input,
 232                        datafmt='dec')
 233             gtkw.trace(dut + "p_valid_i", color=style_input)
 234             gtkw.trace(dut + "p_ready_o", color=style_output)
 235         with gtkw.group("debug"):
 236             gtkw.blank("Some debug statements")
 237             # change the displayed name in the panel
 238             gtkw.trace("top.zero", alias='zero delay shift',
 239                        color=style_debug)
 240             gtkw.trace("top.interesting", color=style_debug)
 241             gtkw.trace("top.test_case", alias="test case", color=style_debug)
 242             gtkw.trace("top.msg", color=style_debug)
 243         with gtkw.group("internal"):
 244             gtkw.trace(dut + "fsm_state")
 245             gtkw.trace(dut + "count[3:0]")
 246             gtkw.trace(dut + "shift_reg[7:0]", datafmt='dec')
 247         with gtkw.group("next port"):
 248             gtkw.trace(dut + "n_data_o[7:0]", color=style_output,
 249                        datafmt='dec')
 250             gtkw.trace(dut + "n_valid_o", color=style_output)
 251             gtkw.trace(dut + "n_ready_i", color=style_input)
 252
 253
 254 def write_gtkw(gtkw_name, vcd_name, gtkw_dom, gtkw_style=None,
 255                module=None, loc=None, color=None, base=None,
 256                zoom=-22.9, marker=-1):
 257     """ Write a GTKWave document according to the supplied style and DOM.
 258
 259     :param gtkw_name: name of the generated GTKWave document
 260     :param vcd_name: name of the waveform file
 261     :param gtkw_dom: DOM style description for the trace pane
 262     :param gtkw_style: style for signals, classes and groups
 263     :param module: default module
 264     :param color: default trace color
 265     :param base: default numerical base
 266     :param loc: source code location to include as a comment
 267     :param zoom: initial zoom level, in GTKWave format
 268     :param marker: initial location of a marker
 269
 270     **gtkw_style format**
 271
 272     Syntax: ``{selector: {attribute: value, ...}, ...}``
 273
 274     "selector" can be a signal, class or group
 275
 276     Signal groups propagate most attributes to their children
 277
 278     Attribute choices:
 279
 280     * module: instance path, for prepending to the signal name
 281     * color: trace color
 282     * base: numerical base for value display
 283     * display: alternate text to display in the signal pane
 284     * comment: comment to display in the signal pane
 285
 286     **gtkw_dom format**
 287
 288     Syntax: ``[signal, (signal, class), (group, [children]), comment, ...]``
 289
 290     The DOM is a list of nodes.
 291
 292     Nodes are signals, signal groups or comments.
 293
 294     * signals are strings, or tuples: ``(signal name, class, class, ...)``
 295     * signal groups are tuples: ``(group name, class, class, ..., [nodes])``
 296     * comments are: ``{'comment': 'comment string'}``
 297
 298     In place of a class name, an inline class description can be used.
 299     ``(signal, {attribute: value, ...}, ...)``
 300     """
 301     colors = {
 302         'blue': GTKWColor.blue,
 303         'cycle': GTKWColor.cycle,
 304         'green': GTKWColor.green,
 305         'indigo': GTKWColor.indigo,
 306         'normal': GTKWColor.normal,
 307         'orange': GTKWColor.orange,
 308         'red': GTKWColor.red,
 309         'violet': GTKWColor.violet,
 310         'yellow': GTKWColor.yellow,
 311     }
 312
 313     with open(gtkw_name, "wt") as gtkw_file:
 314         gtkw = GTKWSave(gtkw_file)
 315         if loc is not None:
 316             gtkw.comment("Auto-generated by " + loc)
 317         gtkw.dumpfile(vcd_name)
 318         # set a reasonable zoom level
 319         # also, move the marker to an interesting place
 320         gtkw.zoom_markers(zoom, marker)
 321
 322         # create an empty style, if needed
 323         if gtkw_style is None:
 324             gtkw_style = dict()
 325
 326         # create an empty root selector, if needed
 327         root_style = gtkw_style.get('', dict())
 328
 329         # apply styles to the root selector, if provided
 330         if module is not None:
 331             root_style['module'] = module
 332         if color is not None:
 333             root_style['color'] = color
 334         if base is not None:
 335             root_style['base'] = base
 336         # base cannot be None, use 'hex' by default
 337         if root_style.get('base') is None:
 338             root_style['base'] = 'hex'
 339
 340         # recursively walk the DOM
 341         def walk(dom, style):
 342             for node in dom:
 343                 node_name = None
 344                 children = None
 345                 # copy the style from the parent
 346                 node_style = style.copy()
 347                 # node is a signal name string
 348                 if isinstance(node, str):
 349                     node_name = node
 350                     # apply style from node name, if specified
 351                     if node_name in gtkw_style:
 352                         node_style.update(gtkw_style[node_name])
 353                 # node is a tuple
 354                 # could be a signal or a group
 355                 elif isinstance(node, tuple):
 356                     node_name = node[0]
 357                     # collect styles from the selectors
 358                     # order goes from the most specific to most generic
 359                     # which means earlier selectors override later ones
 360                     for selector in reversed(node):
 361                         # update the node style from the selector
 362                         if isinstance(selector, str):
 363                             if selector in gtkw_style:
 364                                 node_style.update(gtkw_style[selector])
 365                         # apply an inline style description
 366                         elif isinstance(selector, dict):
 367                             node_style.update(selector)
 368                     # node is a group if it has a child list
 369                     if isinstance(node[-1], list):
 370                         children = node[-1]
 371                 # comment
 372                 elif isinstance(node, dict):
 373                     if 'comment' in node:
 374                         gtkw.blank(node['comment'])
 375                 # emit the group delimiters and walk over the child list
 376                 if children is not None:
 377                     gtkw.begin_group(node_name)
 378                     # pass on the group style to its children
 379                     walk(children, node_style)
 380                     gtkw.end_group(node_name)
 381                 # emit a trace, if the node is a signal
 382                 elif node_name is not None:
 383                     signal_name = node_name
 384                     # prepend module name to signal
 385                     if 'module' in node_style:
 386                         node_module = node_style['module']
 387                         if node_module is not None:
 388                             signal_name = node_module + '.' + signal_name
 389                     node_color = colors.get(node_style.get('color'))
 390                     node_base = node_style.get('base')
 391                     display = node_style.get('display')
 392                     gtkw.trace(signal_name, color=node_color,
 393                                datafmt=node_base, alias=display)
 394
 395         walk(gtkw_dom, root_style)
 396
 397
 398 def test_shifter():
 399     m = Module()
 400     m.submodules.shf = dut = Shifter(8)
 401     print("Shifter port names:")
 402     for port in dut:
 403         print("-", port.name)
 404     # generate RTLIL
 405     # try "proc; show" in yosys to check the data path
 406     il = rtlil.convert(dut, ports=dut.ports())
 407     with open("test_shifter.il", "w") as f:
 408         f.write(il)
 409
 410     # Write the GTKWave project file
 411     write_gtkw_v1("test_shifter", "top.shf", __file__)
 412
 413     # Describe a GTKWave document
 414
 415     # Style for signals, classes and groups
 416     gtkwave_style = {
 417         # Root selector. Gives default attributes for every signal.
 418         '': {'base': 'dec'},
 419         # color the traces, according to class
 420         # class names are not hardcoded, they are just strings
 421         'in': {'color': 'orange'},
 422         'out': {'color': 'yellow'},
 423         # signals in the debug group have a common color and module path
 424         'debug': {'module': 'top', 'color': 'red'},
 425         # display a different string replacing the signal name
 426         'test_case': {'display': 'test case'},
 427     }
 428
 429     # DOM style description for the trace pane
 430     gtkwave_desc = [
 431         # simple signal, without a class
 432         # even so, it inherits the top-level root attributes
 433         'clk',
 434         # comment
 435         {'comment': 'Shifter Demonstration'},
 436         # collapsible signal group
 437         ('prev port', [
 438             # attach a class style for each signal
 439             ('op__sdir', 'in'),
 440             ('p_data_i[7:0]', 'in'),
 441             ('p_shift_i[7:0]', 'in'),
 442             ('p_valid_i', 'in'),
 443             ('p_ready_o', 'out'),
 444         ]),
 445         # Signals in a signal group inherit the group attributes.
 446         # In this case, a different module path and color.
 447         ('debug', [
 448             {'comment': 'Some debug statements'},
 449             # inline attributes, instead of a class name
 450             ('zero', {'display': 'zero delay shift'}),
 451             'interesting',
 452             'test_case',
 453             'msg',
 454         ]),
 455         ('internal', [
 456             'fsm_state',
 457             'count[3:0]',
 458             'shift_reg[7:0]',
 459         ]),
 460         ('next port', [
 461             ('n_data_o[7:0]', 'out'),
 462             ('n_valid_o', 'out'),
 463             ('n_ready_i', 'in'),
 464         ]),
 465     ]
 466
 467     write_gtkw("test_shifter_v2.gtkw", "test_shifter.vcd",
 468                gtkwave_desc,  gtkwave_style,
 469                module="top.shf", loc=__file__, marker=10500000)
 470
 471     sim = Simulator(m)
 472     sim.add_clock(1e-6)
 473
 474     # demonstrates adding extra debug signal traces
 475     # they end up in the top module
 476     #
 477     zero = Signal()  # mark an interesting place
 478     #
 479     # demonstrates string traces
 480     #
 481     # display a message when the signal is high
 482     # the low level is just an horizontal line
 483     interesting = Signal(decoder=lambda v: 'interesting!' if v else '')
 484     # choose between alternate strings based on numerical value
 485     test_cases = ['', '13>>2', '3<<4', '21<<0']
 486     test_case = Signal(8, decoder=lambda v: test_cases[v])
 487     # hack to display arbitrary strings, like debug statements
 488     msg = Signal(decoder=lambda _: msg.str)
 489     msg.str = ''
 490
 491     def send(data, shift, direction):
 492         # present input data and assert valid_i
 493         yield dut.p.data_i.data.eq(data)
 494         yield dut.p.data_i.shift.eq(shift)
 495         yield dut.op.sdir.eq(direction)
 496         yield dut.p.valid_i.eq(1)
 497         yield
 498         # wait for p.ready_o to be asserted
 499         while not (yield dut.p.ready_o):
 500             yield
 501         # show current operation operation
 502         if direction:
 503             msg.str = f'{data}>>{shift}'
 504         else:
 505             msg.str = f'{data}<<{shift}'
 506         # force dump of the above message by toggling the
 507         # underlying signal
 508         yield msg.eq(0)
 509         yield msg.eq(1)
 510         # clear input data and negate p.valid_i
 511         yield dut.p.valid_i.eq(0)
 512         yield dut.p.data_i.data.eq(0)
 513         yield dut.p.data_i.shift.eq(0)
 514         yield dut.op.sdir.eq(0)
 515
 516     def receive(expected):
 517         # signal readiness to receive data
 518         yield dut.n.ready_i.eq(1)
 519         yield
 520         # wait for n.valid_o to be asserted
 521         while not (yield dut.n.valid_o):
 522             yield
 523         # read result
 524         result = yield dut.n.data_o.data
 525         # negate n.ready_i
 526         yield dut.n.ready_i.eq(0)
 527         # check result
 528         assert result == expected
 529         # finish displaying the current operation
 530         msg.str = ''
 531         yield msg.eq(0)
 532         yield msg.eq(1)
 533
 534     def producer():
 535         # 13 >> 2
 536         yield from send(13, 2, 1)
 537         # 3 << 4
 538         yield from send(3, 4, 0)
 539         # 21 << 0
 540         # use a debug signal to mark an interesting operation
 541         # in this case, it is a shift by zero
 542         yield interesting.eq(1)
 543         yield from send(21, 0, 0)
 544         yield interesting.eq(0)
 545
 546     def consumer():
 547         # the consumer is not in step with the producer, but the
 548         # order of the results are preserved
 549         # 13 >> 2 = 3
 550         yield test_case.eq(1)
 551         yield from receive(3)
 552         # 3 << 4 = 48
 553         yield test_case.eq(2)
 554         yield from receive(48)
 555         # 21 << 0 = 21
 556         yield test_case.eq(3)
 557         # you can look for the rising edge of this signal to quickly
 558         # locate this point in the traces
 559         yield zero.eq(1)
 560         yield from receive(21)
 561         yield zero.eq(0)
 562         yield test_case.eq(0)
 563
 564     sim.add_sync_process(producer)
 565     sim.add_sync_process(consumer)
 566     sim_writer = sim.write_vcd(
 567         "test_shifter.vcd",
 568         # include additional signals in the trace dump
 569         traces=[zero, interesting, test_case, msg],
 570     )
 571     with sim_writer:
 572         sim.run()
 573
 574
 575 if __name__ == "__main__":
 576     test_shifter()