Support for core metrics in cocotb (#588)

.github/workflows/main.yml

@@ -40,7 +40,9 @@ jobs:
       - uses: actions/upload-artifact@v3
         with:
           name: "verilog-full-core"
-          path: core.v
+          path: |
+            core.v
+            core.v.json
 
 
   build-riscof-tests:

.gitignore

@@ -15,6 +15,9 @@ venv.bak/
 # Verilog files
 *.v
 
+# Verilog generation debug files
+*.v.json
+
 # Waveform dumps
 *.vcd
 *.gtkw

scripts/gen_verilog.py

@@ -4,10 +4,11 @@
 import sys
 import argparse
 
+from amaranth import *
 from amaranth.build import Platform
-from amaranth.back import verilog
 from amaranth import Module, Elaboratable
 
+
 if __name__ == "__main__":
     parent = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
     sys.path.insert(0, parent)
@@ -16,8 +17,8 @@
 from coreblocks.peripherals.wishbone import WishboneBus
 from coreblocks.core import Core
 from transactron import TransactionModule
-from transactron.utils import flatten_signals
-from transactron.utils.dependencies import DependencyManager, DependencyContext
+from transactron.utils import flatten_signals, DependencyManager, DependencyContext
+from transactron.utils.gen import generate_verilog
 
 from coreblocks.params.configurations import *
 
@@ -44,14 +45,17 @@ def elaborate(self, platform: Platform):
         return tm
 
 
-def gen_verilog(core_config: CoreConfiguration, output_path):
+def gen_verilog(core_config: CoreConfiguration, output_path: str):
     with DependencyContext(DependencyManager()):
-        top = Top(GenParams(core_config))
+        gp = GenParams(core_config)
+        top = Top(gp)
+        ports = list(flatten_signals(top.wb_instr)) + list(flatten_signals(top.wb_data))
 
-        with open(output_path, "w") as f:
-            signals = list(flatten_signals(top.wb_instr)) + list(flatten_signals(top.wb_data))
+        verilog_text, gen_info = generate_verilog(top, ports)
 
-            f.write(verilog.convert(top, ports=signals, strip_internal_attrs=True))
+        gen_info.encode(f"{output_path}.json")
+        with open(output_path, "w") as f:
+            f.write(verilog_text)
 
 
 def main():

scripts/run_benchmarks.py

@@ -61,7 +61,10 @@ def run_benchmarks_with_cocotb(benchmarks: list[str], traces: bool) -> bool:
     arglist += [f"TESTCASE={test_cases}"]
 
     verilog_code = topdir.joinpath("core.v")
+    gen_info_path = f"{verilog_code}.json"
+
     arglist += [f"VERILOG_SOURCES={verilog_code}"]
+    arglist += [f"_COREBLOCKS_GEN_INFO={gen_info_path}"]
 
     if traces:
         arglist += ["TRACES=1"]

scripts/run_signature.py

@@ -32,7 +32,10 @@ def run_with_cocotb(test_name: str, traces: bool, output: str) -> bool:
     arglist += [f"OUTPUT={output}"]
 
     verilog_code = f"{parent}/core.v"
+    gen_info_path = f"{verilog_code}.json"
+
     arglist += [f"VERILOG_SOURCES={verilog_code}"]
+    arglist += [f"_COREBLOCKS_GEN_INFO={gen_info_path}"]
 
     if traces:
         arglist += ["TRACES=1"]

scripts/run_tests.py

@@ -61,7 +61,10 @@ def run_regressions_with_cocotb(tests: list[str], traces: bool) -> bool:
     arglist += [f"TESTCASE={test_cases}"]
 
     verilog_code = topdir.joinpath("core.v")
+    gen_info_path = f"{verilog_code}.json"
+
     arglist += [f"VERILOG_SOURCES={verilog_code}"]
+    arglist += [f"_COREBLOCKS_GEN_INFO={gen_info_path}"]
 
     if traces:
         arglist += ["TRACES=1"]

test/regression/cocotb.py

@@ -1,5 +1,6 @@
 from decimal import Decimal
 import inspect
+import os
 from typing import Any
 from collections.abc import Coroutine
 from dataclasses import dataclass
@@ -14,6 +15,8 @@
 from .memory import *
 from .common import SimulationBackend, SimulationExecutionResult
 
+from transactron.utils.gen import GenerationInfo
+
 
 @dataclass
 class WishboneMasterSignals:
@@ -137,6 +140,23 @@ def __init__(self, dut):
         self.dut = dut
         self.finish_event = Event()
 
+        try:
+            gen_info_path = os.environ["_COREBLOCKS_GEN_INFO"]
+        except KeyError:
+            raise RuntimeError("No core generation info provided")
+
+        self.gen_info = GenerationInfo.decode(gen_info_path)
+
+    def get_cocotb_handle(self, path_components: list[str]) -> ModifiableObject:
+        obj = self.dut
+        # Skip the first component, as it is already referenced in "self.dut"
+        for component in path_components[1:]:
+            # As the component may start with '_' character, we need to use '_id'
+            # function instead of 'getattr' - this is required by cocotb.
+            obj = obj._id(component, extended=False)
+
+        return obj
+
     async def run(self, mem_model: CoreMemoryModel, timeout_cycles: int = 5000) -> SimulationExecutionResult:
         clk = Clock(self.dut.clk, 1, "ns")
         cocotb.start_soon(clk.start())
@@ -157,7 +177,16 @@ async def run(self, mem_model: CoreMemoryModel, timeout_cycles: int = 5000) -> S
         except SimTimeoutError:
             success = False
 
-        return SimulationExecutionResult(success)
+        result = SimulationExecutionResult(success)
+
+        for metric_name, metric_loc in self.gen_info.metrics_location.items():
+            result.metric_values[metric_name] = {}
+            for reg_name, reg_loc in metric_loc.regs.items():
+                value = int(self.get_cocotb_handle(reg_loc))
+                result.metric_values[metric_name][reg_name] = value
+                cocotb.logging.debug(f"Metric {metric_name}/{reg_name}={value}")
+
+        return result
 
     def stop(self):
         self.finish_event.set()

transactron/utils/gen.py

@@ -0,0 +1,124 @@
+from dataclasses import dataclass, field
+from dataclasses_json import dataclass_json
+
+from amaranth import *
+from amaranth.back import verilog
+from amaranth.hdl import ir
+from amaranth.hdl.ast import SignalDict
+
+from transactron.lib.metrics import HardwareMetricsManager
+
+
+__all__ = [
+    "MetricLocation",
+    "GenerationInfo",
+    "generate_verilog",
+]
+
+
+@dataclass_json
+@dataclass
+class MetricLocation:
+    """Information about the location of a metric in the generated Verilog code.
+
+    Attributes
+    ----------
+    regs : dict[str, list[str]]
+        The location of each register of that metric. The location is a list of
+        Verilog identifiers that denote a path consiting of modules names
+        (and the signal name at the end) leading to the register wire.
+    """
+
+    regs: dict[str, list[str]] = field(default_factory=dict)
+
+
+@dataclass_json
+@dataclass
+class GenerationInfo:
+    """Various information about the generated circuit.
+
+    Attributes
+    ----------
+    metrics_location : dict[str, MetricInfo]
+        Mapping from a metric name to an object storing Verilog locations
+        of its registers.
+    """
+
+    metrics_location: dict[str, MetricLocation] = field(default_factory=dict)
+
+    def encode(self, file_name: str):
+        """
+        Encodes the generation information as JSON and saves it to a file.
+        """
+        with open(file_name, "w") as fp:
+            fp.write(self.to_json())  # type: ignore
+
+    @staticmethod
+    def decode(file_name: str) -> "GenerationInfo":
+        """
+        Loads the generation information from a JSON file.
+        """
+        with open(file_name, "r") as fp:
+            return GenerationInfo.from_json(fp.read())  # type: ignore
+
+
+def escape_verilog_identifier(identifier: str) -> str:
+    """
+    Escapes a Verilog identifier according to the language standard.
+
+    From IEEE Std 1364-2001 (IEEE Standard Verilog® Hardware Description Language)
+
+    "2.7.1 Escaped identifiers
+
+    Escaped identifiers shall start with the backslash character and end with white
+    space (space, tab, newline). They provide a means of including any of the printable ASCII
+    characters in an identifier (the decimal values 33 through 126, or 21 through 7E in hexadecimal)."
+    """
+
+    # The standard says how to escape a identifier, but not when. So this is
+    # a non-exhaustive list of characters that Yosys escapes (it is used
+    # by Amaranth when generating Verilog code).
+    characters_to_escape = [".", "$"]
+
+    for char in characters_to_escape:
+        if char in identifier:
+            # Note the intentional space at the end.
+            return f"\\{identifier} "
+
+    return identifier
+
+
+def get_signal_location(signal: Signal, name_map: SignalDict) -> list[str]:
+    raw_location = name_map[signal]
+
+    # Amaranth escapes identifiers when generating Verilog code, but returns non-escaped identifiers
+    # in the name map, so we need to escape it manually.
+    return [escape_verilog_identifier(component) for component in raw_location]
+
+
+def collect_metric_locations(name_map: SignalDict) -> dict[str, MetricLocation]:
+    metrics_location: dict[str, MetricLocation] = {}
+
+    # Collect information about the location of metric registers in the generated code.
+    metrics_manager = HardwareMetricsManager()
+    for metric_name, metric in metrics_manager.get_metrics().items():
+        metric_loc = MetricLocation()
+        for reg_name in metric.regs:
+            metric_loc.regs[reg_name] = get_signal_location(
+                metrics_manager.get_register_value(metric_name, reg_name), name_map
+            )
+
+        metrics_location[metric_name] = metric_loc
+
+    return metrics_location
+
+
+def generate_verilog(
+    top_module: Elaboratable, ports: list[Signal], top_name: str = "top"
+) -> tuple[str, GenerationInfo]:
+    fragment = ir.Fragment.get(top_module, platform=None).prepare(ports=ports)
+    verilog_text, name_map = verilog.convert_fragment(fragment, name=top_name, emit_src=True, strip_internal_attrs=True)
+
+    gen_info = GenerationInfo(metrics_location=collect_metric_locations(name_map))  # type: ignore
+
+    return verilog_text, gen_info