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