Hardware Logging

.github/workflows/main.yml

@@ -244,7 +244,7 @@ jobs:
           sudo apt-get install -y binutils-riscv64-unknown-elf
 
       - name: Run tests
-        run: ./scripts/run_tests.py --verbose
+        run: ./scripts/run_tests.py --verbosity info
 
       - name: Check traces and profiles
         run: ./scripts/run_tests.py -t -p -c 1 TestCore

coreblocks/core.py

@@ -48,8 +48,8 @@ def __init__(self, *, gen_params: GenParams, wb_instr_bus: WishboneBus, wb_data_
         self.wb_instr_bus = wb_instr_bus
         self.wb_data_bus = wb_data_bus
 
-        self.wb_master_instr = WishboneMaster(self.gen_params.wb_params)
-        self.wb_master_data = WishboneMaster(self.gen_params.wb_params)
+        self.wb_master_instr = WishboneMaster(self.gen_params.wb_params, "instr")
+        self.wb_master_data = WishboneMaster(self.gen_params.wb_params, "data")
 
         self.bus_master_instr_adapter = WishboneMasterAdapter(self.wb_master_instr)
         self.bus_master_data_adapter = WishboneMasterAdapter(self.wb_master_data)

coreblocks/fu/jumpbranch.py

@@ -7,8 +7,8 @@
 from transactron import *
 from transactron.core import def_method
 from transactron.lib import *
+from transactron.lib import logging
 from transactron.utils import DependencyManager
-
 from coreblocks.params import *
 from coreblocks.params.keys import AsyncInterruptInsertSignalKey, BranchVerifyKey
 from transactron.utils import OneHotSwitch
@@ -19,6 +19,9 @@
 __all__ = ["JumpBranchFuncUnit", "JumpComponent"]
 
 
+log = logging.HardwareLogger("backend.fu.jumpbranch")
+
+
 class JumpBranchFn(DecoderManager):
     class Fn(IntFlag):
         JAL = auto()
@@ -209,6 +212,14 @@ def _(arg):
 
             with m.If(~is_auipc):
                 self.fifo_branch_resolved.write(m, from_pc=jb.in_pc, next_pc=jump_result, misprediction=misprediction)
+                log.debug(
+                    m,
+                    True,
+                    "jumping from 0x{:08x} to 0x{:08x}; misprediction: {}",
+                    jb.in_pc,
+                    jump_result,
+                    misprediction,
+                )
 
         return m
 

coreblocks/peripherals/wishbone.py

@@ -10,6 +10,7 @@
 from transactron.utils import OneHotSwitchDynamic, assign, RoundRobin
 from transactron.lib.connectors import Forwarder
 from transactron.utils.transactron_helpers import make_layout
+from transactron.lib import logging
 
 
 class WishboneParameters:
@@ -114,6 +115,8 @@ class WishboneMaster(Elaboratable):
     ----------
     wb_params: WishboneParameters
         Parameters for bus generation.
+    name: str, optional
+        Name of this bus. Used for logging.
 
     Attributes
     ----------
@@ -129,7 +132,8 @@ class WishboneMaster(Elaboratable):
         Returns state of request (error or success) and data (in case of read request) as `result_layout`.
     """
 
-    def __init__(self, wb_params: WishboneParameters):
+    def __init__(self, wb_params: WishboneParameters, name: str = ""):
+        self.name = name
         self.wb_params = wb_params
         self.wb_layout = WishboneLayout(wb_params).wb_layout
         self.wb_master = Record(self.wb_layout)
@@ -142,6 +146,11 @@ def __init__(self, wb_params: WishboneParameters):
         # latched input signals
         self.txn_req = Signal(self.method_layouts.request_layout)
 
+        logger_name = "bus.wishbone"
+        if name != "":
+            logger_name += f".{name}"
+        self.log = logging.HardwareLogger(logger_name)
+
     def elaborate(self, platform):
         m = TModule()
 
@@ -194,14 +203,34 @@ def FSMWBCycStart(request):  # noqa: N802
 
         @def_method(m, self.result)
         def _():
-            return result.read(m)
+            ret = result.read(m)
+
+            self.log.debug(
+                m,
+                True,
+                "response data=0x{:x} err={}",
+                ret.data,
+                ret.err,
+            )
+
+            return ret
 
         @def_method(m, self.request, ready=request_ready & result.write.ready)
         def _(arg):
             m.d.sync += assign(self.txn_req, arg)
             # do WBCycStart state in the same clock cycle
             FSMWBCycStart(arg)
 
+            self.log.debug(
+                m,
+                True,
+                "request addr=0x{:x} data=0x{:x} sel=0x{:x} write={}",
+                arg.addr,
+                arg.data,
+                arg.sel,
+                arg.we,
+            )
+
         result.write.schedule_before(self.request)
         result.read.schedule_before(self.request)
 

pytest.ini

@@ -6,3 +6,4 @@ norecursedirs = '*.egg', '.*', 'build', 'dist', 'venv', '__traces__', '__pycache
 filterwarnings =
     ignore:cannot collect test class 'TestbenchIO':pytest.PytestCollectionWarning
     ignore:No files were found in testpaths:pytest.PytestConfigWarning:
+log_cli=true

scripts/run_benchmarks.py

@@ -74,17 +74,18 @@ def run_benchmarks_with_cocotb(benchmarks: list[str], traces: bool) -> bool:
     return res.returncode == 0
 
 
-def run_benchmarks_with_pysim(benchmarks: list[str], traces: bool, verbose: bool) -> bool:
+def run_benchmarks_with_pysim(
+    benchmarks: list[str],
+    traces: bool,
+) -> bool:
     suite = unittest.TestSuite()
 
     def _gen_test(test_name: str):
         def test_fn():
             traces_file = None
             if traces:
                 traces_file = "benchmark." + test_name
-            asyncio.run(
-                test.regression.benchmark.run_benchmark(PySimulation(verbose, traces_file=traces_file), test_name)
-            )
+            asyncio.run(test.regression.benchmark.run_benchmark(PySimulation(traces_file=traces_file), test_name))
 
         test_fn.__name__ = test_name
         test_fn.__qualname__ = test_name
@@ -94,17 +95,17 @@ def test_fn():
     for test_name in benchmarks:
         suite.addTest(unittest.FunctionTestCase(_gen_test(test_name)))
 
-    runner = unittest.TextTestRunner(verbosity=(2 if verbose else 1))
+    runner = unittest.TextTestRunner(verbosity=2)
     result = runner.run(suite)
 
     return result.wasSuccessful()
 
 
-def run_benchmarks(benchmarks: list[str], backend: Literal["pysim", "cocotb"], traces: bool, verbose: bool) -> bool:
+def run_benchmarks(benchmarks: list[str], backend: Literal["pysim", "cocotb"], traces: bool) -> bool:
     if backend == "cocotb":
         return run_benchmarks_with_cocotb(benchmarks, traces)
     elif backend == "pysim":
-        return run_benchmarks_with_pysim(benchmarks, traces, verbose)
+        return run_benchmarks_with_pysim(benchmarks, traces)
     return False
 
 
@@ -144,8 +145,9 @@ def main():
     parser = argparse.ArgumentParser()
     parser.add_argument("-l", "--list", action="store_true", help="List all benchmarks")
     parser.add_argument("-t", "--trace", action="store_true", help="Dump waveforms")
+    parser.add_argument("--log-level", default="WARNING", action="store", help="Level of messages to display.")
+    parser.add_argument("--log-filter", default=".*", action="store", help="Regexp used to filter out logs.")
     parser.add_argument("-p", "--profile", action="store_true", help="Write execution profiles")
-    parser.add_argument("-v", "--verbose", action="store_true", help="Verbose output")
     parser.add_argument("-b", "--backend", default="cocotb", choices=["cocotb", "pysim"], help="Simulation backend")
     parser.add_argument(
         "-o",
@@ -164,6 +166,9 @@ def main():
             print(name)
         return
 
+    os.environ["__TRANSACTRON_LOG_LEVEL"] = args.log_level
+    os.environ["__TRANSACTRON_LOG_FILTER"] = args.log_filter
+
     if args.benchmark_name:
         pattern = re.compile(args.benchmark_name)
         benchmarks = [name for name in benchmarks if pattern.search(name)]
@@ -175,7 +180,7 @@ def main():
     if args.profile:
         os.environ["__TRANSACTRON_PROFILE"] = "1"
 
-    success = run_benchmarks(benchmarks, args.backend, args.trace, args.verbose)
+    success = run_benchmarks(benchmarks, args.backend, args.trace)
     if not success:
         print("Benchmark execution failed")
         sys.exit(1)

scripts/run_signature.py

@@ -45,32 +45,31 @@ def run_with_cocotb(test_name: str, traces: bool, output: str) -> bool:
     return os.path.isfile(output)  # completed successfully if signature file was created
 
 
-def run_with_pysim(test_name: str, traces: bool, verbose: bool, output: str) -> bool:
+def run_with_pysim(test_name: str, traces: bool, output: str) -> bool:
     traces_file = None
     if traces:
         traces_file = os.path.basename(test_name)
     try:
-        asyncio.run(
-            test.regression.signature.run_test(PySimulation(verbose, traces_file=traces_file), test_name, output)
-        )
+        asyncio.run(test.regression.signature.run_test(PySimulation(traces_file=traces_file), test_name, output))
     except RuntimeError as e:
         print("RuntimeError:", e)
         return False
     return True
 
 
-def run_test(test: str, backend: Literal["pysim", "cocotb"], traces: bool, verbose: bool, output: str) -> bool:
+def run_test(test: str, backend: Literal["pysim", "cocotb"], traces: bool, output: str) -> bool:
     if backend == "cocotb":
         return run_with_cocotb(test, traces, output)
     elif backend == "pysim":
-        return run_with_pysim(test, traces, verbose, output)
+        return run_with_pysim(test, traces, output)
     return False
 
 
 def main():
     parser = argparse.ArgumentParser()
     parser.add_argument("-t", "--trace", action="store_true", help="Dump waveforms")
-    parser.add_argument("-v", "--verbose", action="store_true", help="Verbose output")
+    parser.add_argument("--log-level", default="WARNING", action="store", help="Level of messages to display.")
+    parser.add_argument("--log-filter", default=".*", action="store", help="Regexp used to filter out logs.")
     parser.add_argument("-b", "--backend", default="pysim", choices=["cocotb", "pysim"], help="Simulation backend")
     parser.add_argument("-o", "--output", default=None, help="Selects output file to write test signature to")
     parser.add_argument("path")
@@ -79,7 +78,10 @@ def main():
 
     output = args.output if args.output else args.path + ".signature"
 
-    success = run_test(args.path, args.backend, args.trace, args.verbose, output)
+    os.environ["__TRANSACTRON_LOG_LEVEL"] = args.log_level
+    os.environ["__TRANSACTRON_LOG_FILTER"] = args.log_filter
+
+    success = run_test(args.path, args.backend, args.trace, output)
     if not success:
         print(f"{args.path}: Program execution failed")
 

test/conftest.py

@@ -28,6 +28,7 @@ def pytest_addoption(parser: pytest.Parser):
         type=int,
         help="Number of tests to start. If less than number of all selected tests, then starts only subset of them.",
     )
+    group.addoption("--coreblocks-log-filter", default=".*", action="store", help="Regexp used to filter out logs.")
 
 
 def generate_unittestname(item: pytest.Item) -> str:
@@ -104,3 +105,6 @@ def pytest_runtest_setup(item: pytest.Item):
 
     if item.config.getoption("--coreblocks-profile", False):  # type: ignore
         os.environ["__TRANSACTRON_PROFILE"] = "1"
+
+    os.environ["__TRANSACTRON_LOG_FILTER"] = item.config.getoption("--coreblocks-log-filter", ".*")  # type: ignore
+    os.environ["__TRANSACTRON_LOG_LEVEL"] = item.config.getoption("--log-level", "WARNING")  # type: ignore

test/regression/cocotb.py

@@ -1,5 +1,6 @@
 from decimal import Decimal
 import inspect
+import re
 import os
 from typing import Any
 from collections.abc import Coroutine
@@ -87,10 +88,6 @@ async def start(self):
 
             sig_s = WishboneSlaveSignals()
             if sig_m.we:
-                cocotb.logging.debug(
-                    f"Wishbone bus '{self.name}' write request: "
-                    f"addr=0x{addr:x} data=0x{int(sig_m.dat_w):x} sel={sig_m.sel}"
-                )
                 resp = self.model.write(
                     WriteRequest(
                         addr=addr,
@@ -100,7 +97,6 @@ async def start(self):
                     )
                 )
             else:
-                cocotb.logging.debug(f"Wishbone bus '{self.name}' read request: addr=0x{addr:x} sel={sig_m.sel}")
                 resp = self.model.read(
                     ReadRequest(
                         addr=addr,
@@ -123,11 +119,6 @@ async def start(self):
                         raise ValueError("Bus doesn't support rty")
                     sig_s.rty = 1
 
-            cocotb.logging.debug(
-                f"Wishbone bus '{self.name}' response: "
-                f"ack={sig_s.ack} err={sig_s.err} rty={sig_s.rty} data={int(sig_s.dat_r):x}"
-            )
-
             for _ in range(self.delay):
                 await clock_edge_event  # type: ignore
 
@@ -148,6 +139,11 @@ def __init__(self, dut):
 
         self.gen_info = GenerationInfo.decode(gen_info_path)
 
+        self.log_level = os.environ["__TRANSACTRON_LOG_LEVEL"]
+        self.log_filter = os.environ["__TRANSACTRON_LOG_FILTER"]
+
+        cocotb.logging.getLogger().setLevel(self.log_level)
+
     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"
@@ -189,14 +185,40 @@ async def profile_handler(self, clock, profile: Profile):
 
             await clock_edge_event  # type: ignore
 
-    async def assert_handler(self, clock):
+    async def logging_handler(self, clock):
         clock_edge_event = FallingEdge(clock)
 
+        log_level = cocotb.logging.getLogger().level
+
+        logs = [
+            (rec, self.get_cocotb_handle(rec.trigger_location))
+            for rec in self.gen_info.logs
+            if rec.level >= log_level and re.search(self.log_filter, rec.logger_name)
+        ]
+
         while True:
-            for assert_info in self.gen_info.asserts:
-                assert_val = self.get_cocotb_handle(assert_info.location)
-                n, i = assert_info.src_loc
-                assert assert_val.value, f"Assertion at {n}:{i}"
+            for rec, trigger_handle in logs:
+                if not trigger_handle.value:
+                    continue
+
+                values: list[int] = []
+                for field in rec.fields_location:
+                    values.append(int(self.get_cocotb_handle(field).value))
+
+                formatted_msg = rec.format(*values)
+
+                cocotb_log = cocotb.logging.getLogger(rec.logger_name)
+
+                cocotb_log.log(
+                    rec.level,
+                    "%s:%d] %s",
+                    rec.location[0],
+                    rec.location[1],
+                    formatted_msg,
+                )
+
+                if rec.level >= cocotb.logging.ERROR:
+                    assert False, f"Assertion failed at {rec.location[0], rec.location[1]}: {formatted_msg}"
 
             await clock_edge_event  # type: ignore
 
@@ -220,7 +242,7 @@ async def run(self, mem_model: CoreMemoryModel, timeout_cycles: int = 5000) -> S
             profile.transactions_and_methods = self.gen_info.profile_data.transactions_and_methods
             cocotb.start_soon(self.profile_handler(self.dut.clk, profile))
 
-        cocotb.start_soon(self.assert_handler(self.dut.clk))
+        cocotb.start_soon(self.logging_handler(self.dut.clk))
 
         success = True
         try:
@@ -237,7 +259,7 @@ async def run(self, mem_model: CoreMemoryModel, timeout_cycles: int = 5000) -> S
             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}")
+                cocotb.logging.info(f"Metric {metric_name}/{reg_name}={value}")
 
         return result