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