Merge pull request #4536 from YosysHQ/functional

Functional Backend

tests/functional/.gitignore

@@ -0,0 +1,6 @@
+my_module_cxxrtl.cc
+my_module_functional_cxx.cc
+vcd_harness
+*.vcd
+*.smt2
+*.rkt

tests/functional/README.md

@@ -0,0 +1,19 @@
+Tests for the functional backend use pytest as a testrunner.
+
+Run with `pytest -v`
+
+Pytest options you might want:
+
+- `-v`: More progress indication.
+
+- `--basetemp tmp`: Store test files (including vcd results) in tmp. 
+  CAREFUL: contents of tmp will be deleted
+
+- `-k <pattern>`: Run only tests that contain the pattern, e.g.
+  `-k cxx` or `-k smt` or `-k demux` or `-k 'cxx[demux`
+ 
+- `-s`: Don't hide stdout/stderr from the test code.
+
+Custom options for functional backend tests:
+
+- `--per-cell N`: Run only N tests for each cell.

tests/functional/conftest.py

@@ -0,0 +1,34 @@
+import pytest
+from rtlil_cells import generate_test_cases
+import random
+
+random_seed = random.getrandbits(32)
+
+def pytest_configure(config):
+    config.addinivalue_line("markers", "smt: test uses smtlib/z3")
+    config.addinivalue_line("markers", "rkt: test uses racket/rosette")
+
+def pytest_addoption(parser):
+    parser.addoption("--per-cell", type=int, default=None, help="run only N tests per cell")
+    parser.addoption("--steps", type=int, default=1000, help="run each test for N steps")
+    parser.addoption("--seed", type=int, default=random_seed, help="seed for random number generation, use random seed if unspecified")
+
+def pytest_collection_finish(session):
+    print('random seed: {}'.format(session.config.getoption("seed")))
+
+@pytest.fixture
+def num_steps(request):
+    return request.config.getoption("steps")
+
+@pytest.fixture
+def rnd(request):
+    seed1 = request.config.getoption("seed")
+    return lambda seed2: random.Random('{}-{}'.format(seed1, seed2))
+
+def pytest_generate_tests(metafunc):
+    if "cell" in metafunc.fixturenames:
+        per_cell = metafunc.config.getoption("per_cell", default=None)
+        seed1 = metafunc.config.getoption("seed")
+        rnd = lambda seed2: random.Random('{}-{}'.format(seed1, seed2))
+        names, cases = generate_test_cases(per_cell, rnd)
+        metafunc.parametrize("cell,parameters", cases, ids=names)

tests/functional/picorv32_tb.v

@@ -0,0 +1,62 @@
+module gold(
+	input wire clk,
+	output reg resetn,
+	output wire trap,
+	output wire mem_valid,
+	output wire mem_instr,
+	output wire mem_ready,
+	output wire [31:0] mem_addr,
+	output wire [31:0] mem_wdata,
+	output wire [31:0] mem_wstrb,
+	output wire [31:0] mem_rdata,
+);
+
+	initial resetn = 1'b0;
+	always @(posedge clk) resetn <= 1'b1;
+
+	reg [31:0] rom[0:15];
+
+	initial begin
+		rom[0]  = 32'h00200093;
+		rom[1]  = 32'h00200113;
+		rom[2]  = 32'h00111863;
+		rom[3]  = 32'h00102023;
+		rom[4]  = 32'h00108093;
+		rom[5]  = 32'hfe0008e3;
+		rom[6]  = 32'h00008193;
+		rom[7]  = 32'h402181b3;
+		rom[8]  = 32'hfe304ee3;
+		rom[9]  = 32'hfe0186e3;
+		rom[10] = 32'h00110113;
+		rom[11] = 32'hfc000ee3;
+	end
+
+	assign mem_ready = 1'b1;
+	assign mem_rdata = rom[mem_addr[5:2]];
+
+	wire pcpi_wr = 1'b0;
+	wire [31:0] pcpi_rd = 32'b0;
+	wire pcpi_wait = 1'b0;
+	wire pcpi_ready = 1'b0;
+
+	wire [31:0] irq = 32'b0;
+
+	picorv32 picorv32_i(
+		.clk(clk),
+		.resetn(resetn),
+		.trap(trap),
+		.mem_valid(mem_valid),
+		.mem_instr(mem_instr),
+		.mem_ready(mem_ready),
+		.mem_addr(mem_addr),
+		.mem_wdata(mem_wdata),
+		.mem_wstrb(mem_wstrb),
+		.mem_rdata(mem_rdata),
+		.pcpi_wr(pcpi_wr),
+		.pcpi_rd(pcpi_rd),
+		.pcpi_wait(pcpi_wait),
+		.pcpi_ready(pcpi_ready),
+		.irq(irq)
+	);
+
+endmodule

tests/functional/rkt_vcd.py

@@ -0,0 +1,115 @@
+from __future__ import annotations
+from pathlib import Path
+import re
+import subprocess
+from typing import Dict, List, Tuple
+from random import Random
+
+vcd_version = "Yosys/tests/functional/rkt_vcd.py"
+StepList = List[Tuple[int, str]]
+SignalStepMap = Dict[str, StepList]
+SignalWidthMap = Dict[str, int]
+
+def write_vcd(filename: Path, signals: SignalStepMap, timescale='1 ns', date='today'):
+    with open(filename, 'w') as f:
+        # Write the header
+        f.write(f"$date\n    {date}\n$end\n")
+        f.write(f"$timescale {timescale} $end\n")
+        
+        # Declare signals
+        f.write("$scope module gold $end\n")
+        for signal_name, changes in signals.items():
+            signal_size = len(changes[0][1])
+            f.write(f"$var wire {signal_size - 1} {signal_name} {signal_name} $end\n")
+        f.write("$upscope $end\n")
+        f.write("$enddefinitions $end\n")
+        
+        # Collect all unique timestamps
+        timestamps = sorted(set(time for changes in signals.values() for time, _ in changes))
+        
+        # Write initial values
+        f.write("#0\n")
+        for signal_name, changes in signals.items():
+            for time, value in changes:
+                if time == 0:
+                    f.write(f"{value} {signal_name}\n")
+        
+        # Write value changes
+        for time in timestamps:
+            if time != 0:
+                f.write(f"#{time}\n")
+                for signal_name, changes in signals.items():
+                    for change_time, value in changes:
+                        if change_time == time:
+                            f.write(f"{value} {signal_name}\n")
+
+def simulate_rosette(rkt_file_path: Path, vcd_path: Path, num_steps: int, rnd: Random):
+    signals: dict[str, list[str]] = {}
+    inputs: SignalWidthMap = {}
+    outputs: SignalWidthMap = {}
+
+    current_struct_name: str = ""
+    with open(rkt_file_path, 'r') as rkt_file:
+        for line in rkt_file:
+            m = re.search(r'gold_(Inputs|Outputs|State)', line)
+            if m:
+                current_struct_name = m.group(1)
+                if current_struct_name == "State": break
+            elif not current_struct_name: continue # skip lines before structs
+            m = re.search(r'; (.+?)\b \(bitvector (\d+)\)', line)
+            if not m: continue # skip non matching lines (probably closing the struct)
+            signal = m.group(1)
+            width = int(m.group(2))
+            if current_struct_name == "Inputs":
+                inputs[signal] = width
+            elif current_struct_name == "Outputs":
+                outputs[signal] = width
+
+    for signal, width in inputs.items():
+        step_list: list[int] = []
+        for step in range(num_steps):
+            value = rnd.getrandbits(width)
+            binary_string = format(value, '0{}b'.format(width))
+            step_list.append(binary_string)
+        signals[signal] = step_list
+
+    test_rkt_file_path = rkt_file_path.with_suffix('.tst.rkt')
+    with open(test_rkt_file_path, 'w') as test_rkt_file:
+        test_rkt_file.writelines([
+            '#lang rosette\n',
+            f'(require "{rkt_file_path.name}")\n',
+        ])
+
+        for step in range(num_steps):
+            this_step = f"step_{step}"
+            value_list: list[str] = []
+            for signal, width in inputs.items():
+                value = signals[signal][step]
+                value_list.append(f"(bv #b{value} {width})")
+            gold_Inputs = f"(gold_Inputs {' '.join(value_list)})"
+            gold_State = f"(cdr step_{step-1})" if step else "gold_initial"
+            test_rkt_file.write(f"(define {this_step} (gold {gold_Inputs} {gold_State})) (car {this_step})\n")
+
+    cmd = ["racket", test_rkt_file_path]
+    status = subprocess.run(cmd, capture_output=True)
+    assert status.returncode == 0, f"{cmd[0]} failed"
+
+    for signal in outputs.keys():
+        signals[signal] = []
+
+    for line in status.stdout.decode().splitlines():
+        m = re.match(r'\(gold_Outputs( \(bv \S+ \d+\))+\)', line)
+        assert m, f"Incomplete output definition {line!r}"
+        for output, (value, width) in zip(outputs.keys(), re.findall(r'\(bv (\S+) (\d+)\)', line)):
+            assert isinstance(value, str), f"Bad value {value!r}"
+            assert value.startswith(('#b', '#x')), f"Non-binary value {value!r}"
+            assert int(width) == outputs[output], f"Width mismatch for output {output!r} (got {width}, expected {outputs[output]})"
+            int_value = int(value[2:], 16 if value.startswith('#x') else 2)
+            binary_string = format(int_value, '0{}b'.format(width))
+            signals[output].append(binary_string)
+
+    vcd_signals: SignalStepMap = {}
+    for signal, steps in signals.items():
+        vcd_signals[signal] = [(time, f"b{value}") for time, value in enumerate(steps)]
+
+    write_vcd(vcd_path, vcd_signals)

tests/functional/rtlil_cells.py

@@ -0,0 +1,381 @@
+from itertools import chain
+import random
+
+def write_rtlil_cell(f, cell_type, inputs, outputs, parameters):
+    f.write('autoidx 1\n')
+    f.write('module \\gold\n')
+    idx = 1
+    for name, width in inputs.items():
+        f.write(f'\twire width {width} input {idx} \\{name}\n')
+        idx += 1
+    for name, width in outputs.items():
+        f.write(f'\twire width {width} output {idx} \\{name}\n')
+        idx += 1
+    f.write(f'\tcell ${cell_type} \\UUT\n')
+    for (name, value) in parameters.items():
+        if value >= 2**32:
+            f.write(f'\t\tparameter \\{name} {value.bit_length()}\'{value:b}\n')
+        else:
+            f.write(f'\t\tparameter \\{name} {value}\n')
+    for name in chain(inputs.keys(), outputs.keys()):
+        f.write(f'\t\tconnect \\{name} \\{name}\n')
+    f.write(f'\tend\nend\n')
+
+class BaseCell:
+    def __init__(self, name, parameters, inputs, outputs, test_values):
+        self.name = name
+        self.parameters = parameters
+        self.inputs = inputs
+        self.outputs = outputs
+        self.test_values = test_values
+    def get_port_width(self, port, parameters):
+        def parse_specifier(spec):
+            if isinstance(spec, int):
+                return spec
+            if isinstance(spec, str):
+                return parameters[spec]
+            if callable(spec):
+                return spec(parameters)
+            assert False, "expected int, str or lambda"
+        if port in self.inputs:
+            return parse_specifier(self.inputs[port])
+        elif port in self.outputs:
+            return parse_specifier(self.outputs[port])
+        else:
+            assert False, "expected input or output"
+    def generate_tests(self, rnd):
+        def print_parameter(v):
+            if isinstance(v, bool):
+                return "S" if v else "U"
+            else:
+                return str(v)
+        for values in self.test_values:
+            if isinstance(values, int):
+                values = [values]
+            name = '-'.join([print_parameter(v) for v in values])
+            parameters = {parameter: int(values[i]) for i, parameter in enumerate(self.parameters)}
+            if self.is_test_valid(values):
+                yield (name, parameters)
+    def write_rtlil_file(self, path, parameters):
+        inputs = {port: self.get_port_width(port, parameters) for port in self.inputs}
+        outputs = {port: self.get_port_width(port, parameters) for port in self.outputs}
+        with open(path, 'w') as f:
+            write_rtlil_cell(f, self.name, inputs, outputs, parameters)
+    def is_test_valid(self, values):
+        return True
+
+class UnaryCell(BaseCell):
+    def __init__(self, name, values):
+        super().__init__(name, ['A_WIDTH', 'Y_WIDTH', 'A_SIGNED'], {'A': 'A_WIDTH'}, {'Y': 'Y_WIDTH'}, values)
+
+class BinaryCell(BaseCell):
+    def __init__(self, name, values):
+        super().__init__(name, ['A_WIDTH', 'B_WIDTH', 'Y_WIDTH', 'A_SIGNED', 'B_SIGNED'], {'A': 'A_WIDTH', 'B': 'B_WIDTH'}, {'Y': 'Y_WIDTH'}, values)
+
+class ShiftCell(BaseCell):
+    def __init__(self, name, values):
+        super().__init__(name,  ['A_WIDTH', 'B_WIDTH', 'Y_WIDTH', 'A_SIGNED', 'B_SIGNED'], {'A': 'A_WIDTH', 'B': 'B_WIDTH'}, {'Y': 'Y_WIDTH'}, values)
+    def is_test_valid(self, values):
+        (a_width, b_width, y_width, a_signed, b_signed) = values
+        if not self.name in ('shift', 'shiftx') and b_signed: return False
+        if self.name == 'shiftx' and a_signed: return False
+        return True
+
+class MuxCell(BaseCell):
+    def __init__(self, name, values):
+        super().__init__(name, ['WIDTH'], {'A': 'WIDTH', 'B': 'WIDTH', 'S': 1}, {'Y': 'WIDTH'}, values)
+
+class BWCell(BaseCell):
+    def __init__(self, name, values):
+        inputs = {'A': 'WIDTH', 'B': 'WIDTH'}
+        if name == "bwmux": inputs['S'] = 'WIDTH'
+        super().__init__(name, ['WIDTH'], inputs, {'Y': 'WIDTH'}, values)
+
+class PMuxCell(BaseCell):
+    def __init__(self, name, values):
+        b_width = lambda par: par['WIDTH'] * par['S_WIDTH']
+        super().__init__(name, ['WIDTH', 'S_WIDTH'], {'A': 'WIDTH', 'B': b_width, 'S': 'S_WIDTH'}, {'Y': 'WIDTH'}, values)
+
+class BMuxCell(BaseCell):
+    def __init__(self, name, values):
+        a_width = lambda par: par['WIDTH'] << par['S_WIDTH']
+        super().__init__(name, ['WIDTH', 'S_WIDTH'], {'A': a_width, 'S': 'S_WIDTH'}, {'Y': 'WIDTH'}, values)
+
+class DemuxCell(BaseCell):
+    def __init__(self, name, values):
+        y_width = lambda par: par['WIDTH'] << par['S_WIDTH']
+        super().__init__(name, ['WIDTH', 'S_WIDTH'], {'A': 'WIDTH', 'S': 'S_WIDTH'}, {'Y': y_width}, values)
+
+class LUTCell(BaseCell):
+    def __init__(self, name, values):
+        super().__init__(name, ['WIDTH', 'LUT'], {'A': 'WIDTH'}, {'Y': 1}, values)
+    def generate_tests(self, rnd):
+        for width in self.test_values:
+            lut = rnd(f'lut-{width}').getrandbits(2**width)
+            yield (f'{width}', {'WIDTH' : width, 'LUT' : lut})
+
+class ConcatCell(BaseCell):
+    def __init__(self, name, values):
+        y_width = lambda par: par['A_WIDTH'] + par['B_WIDTH']
+        super().__init__(name, ['A_WIDTH', 'B_WIDTH'], {'A': 'A_WIDTH', 'B': 'B_WIDTH'}, {'Y': y_width}, values)
+
+class SliceCell(BaseCell):
+    def __init__(self, name, values):
+        super().__init__(name, ['A_WIDTH', 'OFFSET', 'Y_WIDTH'], {'A': 'A_WIDTH'}, {'Y': 'Y_WIDTH'}, values)
+
+class FACell(BaseCell):
+    def __init__(self, name, values):
+        super().__init__(name, ['WIDTH'], {'A': 'WIDTH', 'B': 'WIDTH', 'C': 'WIDTH'}, {'X': 'WIDTH', 'Y': 'WIDTH'}, values)
+        self.sim_preprocessing = "techmap" # because FA is not implemented in yosys sim
+
+class LCUCell(BaseCell):
+    def __init__(self, name, values):
+        super().__init__(name, ['WIDTH'], {'P': 'WIDTH', 'G': 'WIDTH', 'CI': 1}, {'CO': 'WIDTH'}, values)
+        self.sim_preprocessing = "techmap" # because LCU is not implemented in yosys sim
+
+class ALUCell(BaseCell):
+    def __init__(self, name, values):
+        super().__init__(name, ['A_WIDTH', 'B_WIDTH', 'Y_WIDTH', 'A_SIGNED', 'B_SIGNED'], {'A': 'A_WIDTH', 'B': 'B_WIDTH', 'CI': 1, 'BI': 1}, {'X': 'Y_WIDTH', 'Y': 'Y_WIDTH', 'CO': 'Y_WIDTH'}, values)
+        self.sim_preprocessing = "techmap" # because ALU is not implemented in yosys sim
+
+class FailCell(BaseCell):
+    def __init__(self, name):
+        super().__init__(name, [], {}, {})
+    def generate_tests(self, rnd):
+        yield ('', {})
+    def write_rtlil_file(self, path, parameters):
+        raise Exception(f'\'{self.name}\' cell unimplemented in test generator')
+
+class FFCell(BaseCell):
+    def __init__(self, name, values):
+        super().__init__(name, ['WIDTH'], ['D'], ['Q'], values)
+    def write_rtlil_file(self, path, parameters):
+        from test_functional import yosys_synth
+        verilog_file = path.parent / 'verilog.v'
+        with open(verilog_file, 'w') as f:
+            width = parameters['WIDTH']
+            f.write(f"""
+module gold(
+    input wire clk,
+    input wire [{width-1}:0] D,
+    output reg [{width-1}:0] Q
+);
+    initial Q = {width}'b{("101" * width)[:width]};
+    always @(posedge clk)
+        Q <= D;
+endmodule""")
+        yosys_synth(verilog_file, path)
+
+class MemCell(BaseCell):
+    def __init__(self, name, values):
+        super().__init__(name, ['DATA_WIDTH', 'ADDR_WIDTH'], {'WA': 'ADDR_WIDTH', 'RA': 'ADDR_WIDTH', 'WD': 'DATA_WIDTH'}, {'RD': 'DATA_WIDTH'}, values)
+    def write_rtlil_file(self, path, parameters):
+        from test_functional import yosys_synth
+        verilog_file = path.parent / 'verilog.v'
+        with open(verilog_file, 'w') as f:
+            f.write("""
+module gold(
+    input wire clk,
+    input wire [{1}:0] WA,
+    input wire [{0}:0] WD,
+    input wire [{1}:0] RA,
+    output reg [{0}:0] RD
+);
+    reg [{0}:0] mem[0:{2}];
+    integer i;
+    initial
+        for(i = 0; i <= {2}; i = i + 1)
+            mem[i] = 9192 * (i + 1);
+    always @(*)
+        RD = mem[RA];
+    always @(posedge clk)
+        mem[WA] <= WD;
+endmodule""".format(parameters['DATA_WIDTH'] - 1, parameters['ADDR_WIDTH'] - 1, 2**parameters['ADDR_WIDTH'] - 1))
+        yosys_synth(verilog_file, path)
+
+class MemDualCell(BaseCell):
+    def __init__(self, name, values):
+        super().__init__(name, ['DATA_WIDTH', 'ADDR_WIDTH'],
+                         {'WA1': 'ADDR_WIDTH', 'WA2': 'ADDR_WIDTH',
+                          'RA1': 'ADDR_WIDTH', 'RA2': 'ADDR_WIDTH',
+                          'WD1': 'DATA_WIDTH', 'WD2': 'DATA_WIDTH'},
+                         {'RD1': 'DATA_WIDTH', 'RD2': 'DATA_WIDTH'}, values)
+        self.sim_preprocessing = "memory_map" # issue #4496 in yosys -sim prevents this example from working without memory_map
+    def write_rtlil_file(self, path, parameters):
+        from test_functional import yosys_synth
+        verilog_file = path.parent / 'verilog.v'
+        with open(verilog_file, 'w') as f:
+            f.write("""
+module gold(
+    input wire clk,
+    input wire [{1}:0] WA1,
+    input wire [{0}:0] WD1,
+    input wire [{1}:0] WA2,
+    input wire [{0}:0] WD2,
+    input wire [{1}:0] RA1,
+    input wire [{1}:0] RA2,
+    output reg [{0}:0] RD1,
+    output reg [{0}:0] RD2
+);
+    reg [{0}:0] mem[0:{2}];
+    integer i;
+    initial
+        for(i = 0; i <= {2}; i = i + 1)
+            mem[i] = 9192 * (i + 1);
+    always @(*)
+        RD1 = mem[RA1];
+    always @(*)
+        RD2 = mem[RA2];
+    always @(posedge clk) begin
+        mem[WA1] <= WD1;
+        mem[WA2] <= WD2;
+    end
+endmodule""".format(parameters['DATA_WIDTH'] - 1, parameters['ADDR_WIDTH'] - 1, 2**parameters['ADDR_WIDTH'] - 1))
+        yosys_synth(verilog_file, path)
+
+class PicorvCell(BaseCell):
+    def __init__(self):
+        super().__init__("picorv", [], {}, {}, [()])
+        self.smt_max_steps = 50 # z3 is too slow for more steps
+    def write_rtlil_file(self, path, parameters):
+        from test_functional import yosys, base_path, quote
+        tb_file = base_path / 'tests/functional/picorv32_tb.v'
+        cpu_file = base_path / 'tests/functional/picorv32.v'
+        yosys(f"read_verilog {quote(tb_file)} {quote(cpu_file)}; prep -top gold; flatten; write_rtlil {quote(path)}")
+
+binary_widths = [
+    # try to cover extending A operand, extending B operand, extending/truncating result
+    (16, 32, 48, True, True),
+    (16, 32, 48, False, False),
+    (32, 16, 48, True, True),
+    (32, 16, 48, False, False),
+    (32, 32, 16, True, True),
+    (32, 32, 16, False, False),
+    # have at least one test that checks small inputs, which will exercise the cornercases more
+    (4, 4, 8, True, True),
+    (4, 4, 8, False, False)
+]
+
+unary_widths = [
+    (6, 12, True),
+    (6, 12, False),
+    (32, 16, True),
+    (32, 16, False)
+]
+
+# note that meaningless combinations of signednesses are eliminated,
+# like e.g. most shift operations don't take signed shift amounts
+shift_widths = [
+    # one set of tests that definitely checks all possible shift amounts
+    # with a bigger result width to make sure it's not truncated
+    (32, 6, 64, True, False),
+    (32, 6, 64, False, False),
+    (32, 6, 64, True, True),
+    (32, 6, 64, False, True),
+    # one set that checks very oversized shifts
+    (32, 32, 64, True, False),
+    (32, 32, 64, False, False),
+    (32, 32, 64, True, True),
+    (32, 32, 64, False, True),
+    # at least one test where the result is going to be truncated
+    (32, 6, 16, False, False),
+    # since 1-bit shifts are special cased
+    (1, 4, 1, False, False),
+    (1, 4, 1, True, False),
+]
+
+rtlil_cells = [
+    UnaryCell("not", unary_widths),
+    UnaryCell("pos", unary_widths),
+    UnaryCell("neg", unary_widths),
+    BinaryCell("and", binary_widths),
+    BinaryCell("or", binary_widths),
+    BinaryCell("xor", binary_widths),
+    BinaryCell("xnor", binary_widths),
+    UnaryCell("reduce_and", unary_widths),
+    UnaryCell("reduce_or", unary_widths),
+    UnaryCell("reduce_xor", unary_widths),
+    UnaryCell("reduce_xnor", unary_widths),
+    UnaryCell("reduce_bool", unary_widths),
+    ShiftCell("shl", shift_widths),
+    ShiftCell("shr", shift_widths),
+    ShiftCell("sshl", shift_widths),
+    ShiftCell("sshr", shift_widths),
+    ShiftCell("shift", shift_widths),
+    ShiftCell("shiftx", shift_widths),
+    FACell("fa", [8, 20]),
+    LCUCell("lcu", [1, 10]),
+    ALUCell("alu", binary_widths),
+    BinaryCell("lt", binary_widths),
+    BinaryCell("le", binary_widths),
+    BinaryCell("eq", binary_widths),
+    BinaryCell("ne", binary_widths),
+    BinaryCell("eqx", binary_widths),
+    BinaryCell("nex", binary_widths),
+    BinaryCell("ge", binary_widths),
+    BinaryCell("gt", binary_widths),
+    BinaryCell("add", binary_widths),
+    BinaryCell("sub", binary_widths),
+    BinaryCell("mul", binary_widths),
+#    BinaryCell("macc"),
+    BinaryCell("div", binary_widths),
+    BinaryCell("mod", binary_widths),
+    BinaryCell("divfloor", binary_widths),
+    BinaryCell("modfloor", binary_widths),
+    BinaryCell("pow", binary_widths),
+    UnaryCell("logic_not", unary_widths),
+    BinaryCell("logic_and", binary_widths),
+    BinaryCell("logic_or", binary_widths),
+    SliceCell("slice", [(32, 10, 15), (8, 0, 4), (10, 0, 10)]),
+    ConcatCell("concat", [(16, 16), (8, 14), (20, 10)]),
+    MuxCell("mux", [10, 16, 40]),
+    BMuxCell("bmux", [(10, 1), (10, 2), (10, 4)]),
+    PMuxCell("pmux", [(10, 1), (10, 4), (20, 4)]),
+    DemuxCell("demux", [(10, 1), (32, 2), (16, 4)]),
+    LUTCell("lut", [4, 6, 8]),
+#    ("sop", ["A", "Y"]),
+#    ("tribuf", ["A", "EN", "Y"]),
+#    ("specify2", ["EN", "SRC", "DST"]),
+#    ("specify3", ["EN", "SRC", "DST", "DAT"]),
+#    ("specrule", ["EN_SRC", "EN_DST", "SRC", "DST"]),
+    BWCell("bweqx", [10, 16, 40]),
+    BWCell("bwmux", [10, 16, 40]),
+    FFCell("ff", [10, 20, 40]),
+    MemCell("mem", [(16, 4)]),
+    MemDualCell("mem-dual", [(16, 4)]),
+#    ("assert", ["A", "EN"]),
+#    ("assume", ["A", "EN"]),
+#    ("live", ["A", "EN"]),
+#    ("fair", ["A", "EN"]),
+#    ("cover", ["A", "EN"]),
+#    ("initstate", ["Y"]),
+#    ("anyconst", ["Y"]),
+#    ("anyseq", ["Y"]),
+#    ("anyinit", ["D", "Q"]),
+#    ("allconst", ["Y"]),
+#    ("allseq", ["Y"]),
+#    ("equiv", ["A", "B", "Y"]),
+#    ("print", ["EN", "TRG", "ARGS"]),
+#    ("check", ["A", "EN", "TRG", "ARGS"]),
+#    ("set_tag", ["A", "SET", "CLR", "Y"]),
+#    ("get_tag", ["A", "Y"]),
+#    ("overwrite_tag", ["A", "SET", "CLR"]),
+#    ("original_tag", ["A", "Y"]),
+#    ("future_ff", ["A", "Y"]),
+#    ("scopeinfo", []),
+    PicorvCell()
+]
+
+def generate_test_cases(per_cell, rnd):
+    tests = []
+    names = []
+    for cell in rtlil_cells:
+        seen_names = set()
+        for (name, parameters) in cell.generate_tests(rnd):
+            if not name in seen_names:
+                seen_names.add(name)
+                tests.append((cell, parameters))
+                names.append(f'{cell.name}-{name}' if name != '' else cell.name)
+                if per_cell is not None and len(seen_names) >= per_cell:
+                    break
+    return (names, tests)

tests/functional/run-test.sh

@@ -0,0 +1,2 @@
+#!/usr/bin/env bash
+pytest -v -m "not smt and not rkt" "$@"

tests/functional/smt_vcd.py

@@ -0,0 +1,188 @@
+import sys
+import argparse
+import os
+import smtio
+import re
+
+class SExprParserError(Exception):
+    pass
+
+class SExprParser:
+    def __init__(self):
+        self.peekbuf = None
+        self.stack = [[]]
+        self.atom_pattern = re.compile(r'[a-zA-Z0-9~!@$%^&*_\-+=<>.?/#]+')
+    def parse_line(self, line):
+        ptr = 0
+        while ptr < len(line):
+            if line[ptr].isspace():
+                ptr += 1
+            elif line[ptr] == ';':
+                break
+            elif line[ptr] == '(':
+                ptr += 1
+                self.stack.append([])
+            elif line[ptr] == ')':
+                ptr += 1
+                assert len(self.stack) > 1, "too many closed parentheses"
+                v = self.stack.pop()
+                self.stack[-1].append(v)
+            else:
+                match = self.atom_pattern.match(line, ptr)
+                if match is None:
+                    raise SExprParserError(f"invalid character '{line[ptr]}' in line '{line}'")
+                start, ptr = match.span()
+                self.stack[-1].append(line[start:ptr])
+    def finish(self):
+        assert len(self.stack) == 1, "too many open parentheses"
+    def retrieve(self):
+        rv, self.stack[0] = self.stack[0], []
+        return rv
+
+def simulate_smt_with_smtio(smt_file_path, vcd_path, smt_io, num_steps, rnd):
+    inputs = {}
+    outputs = {}
+    states = {}
+
+    def handle_datatype(lst):
+        print(lst)
+        datatype_name = lst[1]
+        declarations = lst[2][0][1:]  # Skip the first item (e.g., 'mk_inputs')
+        if datatype_name.endswith("_Inputs"):
+            for declaration in declarations:
+                input_name = declaration[0]
+                bitvec_size = declaration[1][2]
+                assert input_name.startswith("gold_Inputs_")
+                inputs[input_name[len("gold_Inputs_"):]] = int(bitvec_size)
+        elif datatype_name.endswith("_Outputs"):
+            for declaration in declarations:
+                output_name = declaration[0]
+                bitvec_size = declaration[1][2]
+                assert output_name.startswith("gold_Outputs_")
+                outputs[output_name[len("gold_Outputs_"):]] = int(bitvec_size)
+        elif datatype_name.endswith("_State"):
+            for declaration in declarations:
+                state_name = declaration[0]
+                assert state_name.startswith("gold_State_")
+                if declaration[1][0] == "_":
+                    states[state_name[len("gold_State_"):]] = int(declaration[1][2])
+                else:
+                    states[state_name[len("gold_State_"):]] = (declaration[1][1][2], declaration[1][2][2])
+
+    parser = SExprParser()
+    with open(smt_file_path, 'r') as smt_file:
+        for line in smt_file:
+            parser.parse_line(line)
+            for expr in parser.retrieve():
+                smt_io.write(smt_io.unparse(expr))
+                if expr[0] == "declare-datatype":
+                    handle_datatype(expr)
+                    
+    parser.finish()
+    assert smt_io.check_sat() == 'sat'
+
+    def set_step(inputs, step):
+        # This function assumes 'inputs' is a dictionary like {"A": 5, "B": 4}
+        # and 'input_values' is a dictionary like {"A": 5, "B": 13} specifying the concrete values for each input.
+        
+        mk_inputs_parts = []
+        for input_name, width in inputs.items():
+            value = rnd.getrandbits(width)  # Generate a random number up to the maximum value for the bit size
+            binary_string = format(value, '0{}b'.format(width))  # Convert value to binary with leading zeros
+            mk_inputs_parts.append(f"#b{binary_string}")
+
+        mk_inputs_call = "gold_Inputs " + " ".join(mk_inputs_parts)
+        return [
+            f"(define-const test_inputs_step_n{step} gold_Inputs ({mk_inputs_call}))\n",
+            f"(define-const test_results_step_n{step} (Pair gold_Outputs gold_State) (gold test_inputs_step_n{step} test_state_step_n{step}))\n",
+            f"(define-const test_outputs_step_n{step} gold_Outputs (first test_results_step_n{step}))\n",
+            f"(define-const test_state_step_n{step+1} gold_State (second test_results_step_n{step}))\n",
+        ]
+
+    smt_commands = [f"(define-const test_state_step_n0 gold_State gold-initial)\n"]
+    for step in range(num_steps):
+        for step_command in set_step(inputs, step):
+            smt_commands.append(step_command)
+
+    for command in smt_commands:
+        smt_io.write(command)
+
+    assert smt_io.check_sat() == 'sat'
+
+    # Store signal values
+    signals = {name: [] for name in list(inputs.keys()) + list(outputs.keys())}
+    # Retrieve and print values for each state
+    def hex_to_bin(value):
+        if value.startswith('x'):
+            hex_value = value[1:]  # Remove the 'x' prefix
+            bin_value = bin(int(hex_value, 16))[2:]  # Convert to binary and remove the '0b' prefix
+            return f'b{bin_value.zfill(len(hex_value) * 4)}'  # Add 'b' prefix and pad with zeros
+        return value
+
+    combined_assertions = []
+    for step in range(num_steps):
+        print(f"Values for step {step + 1}:")
+        for input_name, width in inputs.items():
+            value = smt_io.get(f'(gold_Inputs_{input_name} test_inputs_step_n{step})')
+            value = hex_to_bin(value[1:])
+            print(f"  {input_name}: {value}")        
+            signals[input_name].append((step, value))
+        for output_name, width in outputs.items():
+            value = smt_io.get(f'(gold_Outputs_{output_name} test_outputs_step_n{step})')
+            value = hex_to_bin(value[1:])
+            print(f"  {output_name}: {value}")
+            signals[output_name].append((step, value))
+            combined_assertions.append(f'(= (gold_Outputs_{output_name} test_outputs_step_n{step}) #{value})')
+    # Create a single assertion covering all timesteps
+    combined_condition = " ".join(combined_assertions)
+    smt_io.write(f'(assert (not (and {combined_condition})))')
+
+    # Check the combined assertion
+    assert smt_io.check_sat(["unsat"]) == "unsat"
+
+    def write_vcd(filename, signals, timescale='1 ns', date='today'):
+        with open(filename, 'w') as f:
+            # Write the header
+            f.write(f"$date\n    {date}\n$end\n")
+            f.write(f"$timescale {timescale} $end\n")
+            
+            # Declare signals
+            f.write("$scope module gold $end\n")
+            for signal_name, changes in signals.items():
+                signal_size = len(changes[0][1])
+                f.write(f"$var wire {signal_size - 1} {signal_name} {signal_name} $end\n")
+            f.write("$upscope $end\n")
+            f.write("$enddefinitions $end\n")
+            
+            # Collect all unique timestamps
+            timestamps = sorted(set(time for changes in signals.values() for time, _ in changes))
+            
+            # Write initial values
+            f.write("#0\n")
+            for signal_name, changes in signals.items():
+                for time, value in changes:
+                    if time == 0:
+                        f.write(f"{value} {signal_name}\n")
+            
+            # Write value changes
+            for time in timestamps:
+                if time != 0:
+                    f.write(f"#{time}\n")
+                    for signal_name, changes in signals.items():
+                        for change_time, value in changes:
+                            if change_time == time:
+                                f.write(f"{value} {signal_name}\n")
+
+
+    write_vcd(vcd_path, signals)
+
+def simulate_smt(smt_file_path, vcd_path, num_steps, rnd):
+    so = smtio.SmtOpts()
+    so.solver = "z3"
+    so.logic = "ABV"
+    so.debug_print = True
+    smt_io = smtio.SmtIo(opts=so)
+    try:
+        simulate_smt_with_smtio(smt_file_path, vcd_path, smt_io, num_steps, rnd)
+    finally:
+        smt_io.p_close()

tests/functional/test_functional.py

@@ -0,0 +1,94 @@
+import subprocess
+import pytest
+import sys
+import shlex
+from pathlib import Path
+
+base_path = Path(__file__).resolve().parent.parent.parent
+
+# quote a string or pathlib path so that it can be used by bash or yosys
+# TODO: is this really appropriate for yosys?
+def quote(path):
+    return shlex.quote(str(path))
+
+# run a shell command and require the return code to be 0
+def run(cmd, **kwargs):
+    print(' '.join([quote(x) for x in cmd]))
+    status = subprocess.run(cmd, **kwargs)
+    assert status.returncode == 0, f"{cmd[0]} failed"
+
+def yosys(script):
+    run([base_path / 'yosys', '-Q', '-p', script])
+
+def compile_cpp(in_path, out_path, args):
+    run(['g++', '-g', '-std=c++17'] + args + [str(in_path), '-o', str(out_path)])
+
+def yosys_synth(verilog_file, rtlil_file):
+    yosys(f"read_verilog {quote(verilog_file)} ; prep ; write_rtlil {quote(rtlil_file)}")
+
+# simulate an rtlil file with yosys, comparing with a given vcd file, and writing out the yosys simulation results into a second vcd file
+def yosys_sim(rtlil_file, vcd_reference_file, vcd_out_file, preprocessing = ""):
+    try:
+        yosys(f"read_rtlil {quote(rtlil_file)}; {preprocessing}; sim -r {quote(vcd_reference_file)} -scope gold -vcd {quote(vcd_out_file)} -timescale 1us -sim-gold -fst-noinit")
+    except:
+        # if yosys sim fails it's probably because of a simulation mismatch
+        # since yosys sim aborts on simulation mismatch to generate vcd output
+        # we have to re-run with a different set of flags
+        # on this run we ignore output and return code, we just want a best-effort attempt to get a vcd
+        subprocess.run([base_path / 'yosys', '-Q', '-p',
+            f'read_rtlil {quote(rtlil_file)}; sim -vcd {quote(vcd_out_file)} -a -r {quote(vcd_reference_file)} -scope gold -timescale 1us -fst-noinit'],
+            capture_output=True, check=False)
+        raise
+
+def test_cxx(cell, parameters, tmp_path, num_steps, rnd):
+    rtlil_file = tmp_path / 'rtlil.il'
+    vcdharness_cc_file = base_path / 'tests/functional/vcd_harness.cc'
+    cc_file = tmp_path / 'my_module_functional_cxx.cc'
+    vcdharness_exe_file = tmp_path / 'a.out'
+    vcd_functional_file = tmp_path / 'functional.vcd'
+    vcd_yosys_sim_file = tmp_path / 'yosys.vcd'
+
+    cell.write_rtlil_file(rtlil_file, parameters)
+    yosys(f"read_rtlil {quote(rtlil_file)} ; clk2fflogic ; write_functional_cxx {quote(cc_file)}")
+    compile_cpp(vcdharness_cc_file, vcdharness_exe_file, ['-I', tmp_path, '-I', str(base_path / 'backends/functional/cxx_runtime')])
+    seed = str(rnd(cell.name + "-cxx").getrandbits(32))
+    run([str(vcdharness_exe_file.resolve()), str(vcd_functional_file), str(num_steps), str(seed)])
+    yosys_sim(rtlil_file, vcd_functional_file, vcd_yosys_sim_file, getattr(cell, 'sim_preprocessing', ''))
+
+@pytest.mark.smt
+def test_smt(cell, parameters, tmp_path, num_steps, rnd):
+    import smt_vcd
+
+    rtlil_file = tmp_path / 'rtlil.il'
+    smt_file = tmp_path / 'smtlib.smt'
+    vcd_functional_file = tmp_path / 'functional.vcd'
+    vcd_yosys_sim_file = tmp_path / 'yosys.vcd'
+
+    if hasattr(cell, 'smt_max_steps'):
+        num_steps = min(num_steps, cell.smt_max_steps)
+
+    cell.write_rtlil_file(rtlil_file, parameters)
+    yosys(f"read_rtlil {quote(rtlil_file)} ; clk2fflogic ; write_functional_smt2 {quote(smt_file)}")
+    run(['z3', smt_file]) # check if output is valid smtlib before continuing
+    smt_vcd.simulate_smt(smt_file, vcd_functional_file, num_steps, rnd(cell.name + "-smt"))
+    yosys_sim(rtlil_file, vcd_functional_file, vcd_yosys_sim_file, getattr(cell, 'sim_preprocessing', ''))
+
+@pytest.mark.rkt
+def test_rkt(cell, parameters, tmp_path, num_steps, rnd):
+    import rkt_vcd
+
+    rtlil_file = tmp_path / 'rtlil.il'
+    rkt_file = tmp_path / 'smtlib.rkt'
+    vcd_functional_file = tmp_path / 'functional.vcd'
+    vcd_yosys_sim_file = tmp_path / 'yosys.vcd'
+
+    cell.write_rtlil_file(rtlil_file, parameters)
+    yosys(f"read_rtlil {quote(rtlil_file)} ; clk2fflogic ; write_functional_rosette -provides {quote(rkt_file)}")
+    rkt_vcd.simulate_rosette(rkt_file, vcd_functional_file, num_steps, rnd(cell.name + "-rkt"))
+    yosys_sim(rtlil_file, vcd_functional_file, vcd_yosys_sim_file, getattr(cell, 'sim_preprocessing', ''))
+
+def test_print_graph(tmp_path):
+    tb_file = base_path / 'tests/functional/picorv32_tb.v'
+    cpu_file = base_path / 'tests/functional/picorv32.v'
+    # currently we only check that we can print the graph without getting an error, not that it prints anything sensibl
+    yosys(f"read_verilog {quote(tb_file)} {quote(cpu_file)}; prep -top gold; flatten; clk2fflogic; test_generic")

tests/functional/vcd_harness.cc

@@ -0,0 +1,146 @@
+#include <cstdio>
+#include <iostream>
+#include <fstream>
+#include <random>
+#include <ctype.h>
+#include <vector>
+#include <unordered_map>
+
+#include "my_module_functional_cxx.cc"
+
+class VcdFile {
+	std::ofstream &ofs;
+	std::string code_alloc = "!";
+	std::unordered_map<std::string, std::string> codes;
+	std::string name_mangle(std::string name) {
+		std::string ret = name;
+		bool escape = ret.empty() || !isalpha(ret[0]) && ret[0] != '_';
+		for(size_t i = 0; i < ret.size(); i++) {
+			if(isspace(ret[i])) ret[i] = '_';
+			if(!isalnum(ret[i]) && ret[i] != '_' && ret[i] != '$')
+			escape = true;
+		}
+		if(escape)
+			return "\\" + ret;
+		else
+			return ret;
+	}
+	std::string allocate_code() {
+		std::string ret = code_alloc;
+		for (size_t i = 0; i < code_alloc.size(); i++)
+			if (code_alloc[i]++ == '~')
+				code_alloc[i] = '!';
+			else
+				return ret;
+		code_alloc.push_back('!');
+		return ret;
+	}
+public:
+	VcdFile(std::ofstream &ofs) : ofs(ofs) {}
+	struct DumpHeader {
+		VcdFile *file;
+		explicit DumpHeader(VcdFile *file) : file(file) {}
+		template <size_t n> void operator()(const char *name, Signal<n> value)
+		{
+			auto it = file->codes.find(name);
+			if(it == file->codes.end())
+				it = file->codes.emplace(name, file->allocate_code()).first;
+			file->ofs << "$var wire " << n << " " << it->second << " " << file->name_mangle(name) << " $end\n";
+		}
+		template <size_t n, size_t m> void operator()(const char *name, Memory<n, m> value) {}
+	};
+	struct Dump {
+		VcdFile *file;
+		explicit Dump(VcdFile *file) : file(file) {}
+		template <size_t n> void operator()(const char *name, Signal<n> value)
+		{
+			if (n == 1) {
+				file->ofs << (value[0] ? '1' : '0');
+				file->ofs << file->codes.at(name) << "\n";
+			} else {
+				file->ofs << "b";
+				for (size_t i = n; i-- > 0;)
+					file->ofs << (value[i] ? '1' : '0');
+				file->ofs << " " << file->codes.at(name) << "\n";
+			}
+		}
+		template <size_t n, size_t m> void operator()(const char *name, Memory<n, m> value) {}
+	};
+	void begin_header() {
+		constexpr int number_timescale = 1;
+		const std::string units_timescale = "us";
+		ofs << "$timescale " << number_timescale << " " << units_timescale << " $end\n";
+		ofs << "$scope module gold $end\n";
+	}
+	void end_header() {
+		ofs << "$enddefinitions $end\n$dumpvars\n";
+	}
+	template<typename... Args> void header(Args ...args) {
+		begin_header();
+		DumpHeader d(this);
+		(args.visit(d), ...);
+		end_header();
+	}
+	void begin_data(int step) {
+		ofs << "#" << step << "\n";
+	}
+	template<typename... Args> void data(int step, Args ...args) {
+		begin_data(step);
+		Dump d(this);
+		(args.visit(d), ...);
+	}
+	DumpHeader dump_header() { return DumpHeader(this); }
+	Dump dump() { return Dump(this); }
+};
+
+template <size_t n> Signal<n> random_signal(std::mt19937 &gen)
+{
+	std::uniform_int_distribution<uint32_t> dist;
+	std::array<uint32_t, (n + 31) / 32> words;
+	for (auto &w : words)
+		w = dist(gen);
+	return Signal<n>::from_array(words);
+}
+
+struct Randomize {
+	std::mt19937 &gen;
+	Randomize(std::mt19937 &gen) : gen(gen) {}
+
+	template <size_t n> void operator()(const char *, Signal<n> &signal) { signal = random_signal<n>(gen); }
+};
+
+int main(int argc, char **argv)
+{
+	if (argc != 4) {
+		std::cerr << "Usage: " << argv[0] << " <functional_vcd_filename> <steps> <seed>\n";
+		return 1;
+	}
+
+	const std::string functional_vcd_filename = argv[1];
+	const int steps = atoi(argv[2]);
+	const uint32_t seed = atoi(argv[3]);
+
+	gold::Inputs inputs;
+	gold::Outputs outputs;
+	gold::State state;
+	gold::State next_state;
+
+	std::ofstream vcd_file(functional_vcd_filename);
+	VcdFile vcd(vcd_file);
+	vcd.header(inputs, outputs, state);
+
+	std::mt19937 gen(seed);
+
+	gold::initialize(state);
+
+	for (int step = 0; step < steps; ++step) {
+		inputs.visit(Randomize(gen));
+
+		gold::eval(inputs, outputs, state, next_state);
+		vcd.data(step, inputs, outputs, state);
+
+		state = next_state;
+	}
+
+	return 0;
+}