Debug logs

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 log
 from transactron.utils import DependencyManager
-
 from coreblocks.params import *
 from coreblocks.params.keys import AsyncInterruptInsertSignalKey, BranchVerifyKey
 from transactron.utils import OneHotSwitch
@@ -209,6 +209,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
 

test/regression/pysim.py

@@ -8,6 +8,8 @@
 from .common import SimulationBackend, SimulationExecutionResult
 
 from transactron.testing import PysimSimulator, TestGen
+from transactron.testing.log import *
+from transactron.lib import log
 from transactron.utils.dependencies import DependencyContext, DependencyManager
 from transactron.lib.metrics import HardwareMetricsManager
 from ..peripherals.test_wishbone import WishboneInterfaceWrapper
@@ -46,17 +48,11 @@ def f():
 
                 resp_data = 0
 
-                bus_name = "instr" if is_instr_bus else "data"
-
                 if (yield wb_ctrl.wb.we):
-                    if self.verbose:
-                        print(f"Wishbone '{bus_name}' bus write request: addr=0x{addr:x} data={dat_w:x} sel={sel:b}")
                     resp = mem_model.write(
                         WriteRequest(addr=addr, data=dat_w, byte_count=word_width_bytes, byte_sel=sel)
                     )
                 else:
-                    if self.verbose:
-                        print(f"Wishbone '{bus_name}' bus read request: addr=0x{addr:x} sel={sel:b}")
                     resp = mem_model.read(
                         ReadRequest(
                             addr=addr,
@@ -67,9 +63,6 @@ def f():
                     )
                     resp_data = resp.data
 
-                    if self.verbose:
-                        print(f"Wishbone '{bus_name}' bus read response: data=0x{resp.data:x}")
-
                 ack = err = rty = 0
                 match resp.status:
                     case ReplyStatus.OK:
@@ -141,6 +134,7 @@ async def run(self, mem_model: CoreMemoryModel, timeout_cycles: int = 5000) -> S
             sim = PysimSimulator(core, max_cycles=timeout_cycles, traces_file=self.traces_file)
             sim.add_sync_process(self._wishbone_slave(mem_model, wb_instr_ctrl, is_instr_bus=True))
             sim.add_sync_process(self._wishbone_slave(mem_model, wb_data_ctrl, is_instr_bus=False))
+            sim.add_sync_process(make_log_process(log.DEBUG))
 
             metric_values: dict[str, dict[str, int]] = {}
 

transactron/lib/log.py

@@ -0,0 +1,95 @@
+import operator
+from functools import reduce
+from dataclasses import dataclass, field
+from dataclasses_json import dataclass_json
+from typing import TypeAlias
+
+from amaranth import *
+from amaranth.tracer import get_src_loc
+
+from transactron.utils import SrcLoc
+from transactron.utils._typing import ModuleLike, ValueLike
+from transactron.utils.dependencies import DependencyContext, ListKey
+
+LogLevel: TypeAlias = int
+
+DEBUG: LogLevel = 10
+INFO: LogLevel = 20
+WARNING: LogLevel = 30
+ERROR: LogLevel = 40
+
+_level_to_name = {
+    ERROR: "ERROR",
+    WARNING: "WARN",
+    INFO: "INFO",
+    DEBUG: "DEBUG",
+}
+
+
+def get_level_name(level: LogLevel) -> str:
+    if level in _level_to_name:
+        return _level_to_name[level]
+    return "Level {}".format(level)
+
+
+@dataclass_json
+@dataclass(frozen=True)
+class LogEntryDescription:
+    level: LogLevel
+    format_str: str
+    location: SrcLoc
+
+    def format(self, *args) -> str:
+        return self.format_str.format(*args)
+
+
+@dataclass(frozen=True)
+class LogEntry(LogEntryDescription):
+    trigger: Signal
+    fields: list[Signal] = field(default_factory=list)
+
+
+@dataclass(frozen=True)
+class LogKey(ListKey[LogEntry]):
+    pass
+
+
+def log(m: ModuleLike, level: LogLevel, format: str, trigger: ValueLike, *args, src_loc_at: int = 0):
+    """
+
+    Parameters
+    ----------
+    src_loc_at : int, optional
+        How many stack frames below to look for the source location, used to
+        identify the failing assertion.
+    """
+
+    # TODO: make the location relative to the root of the project
+    src_loc = get_src_loc(src_loc_at + 1)
+
+    trigger_signal = Signal()
+    m.d.comb += trigger_signal.eq(trigger)
+
+    entry = LogEntry(level=level, format_str=format, location=src_loc, trigger=trigger_signal)
+
+    for arg in args:
+        sig = Signal.like(arg)
+        m.d.top_comb += sig.eq(arg)
+        entry.fields.append(sig)
+
+    dependencies = DependencyContext.get()
+    dependencies.add_dependency(LogKey(), entry)
+
+
+def debug(m: ModuleLike, trigger: ValueLike, format: str, *args, **kwargs):
+    log(m, DEBUG, format, trigger, *args, **kwargs)
+
+
+def get_logs(log_level: LogLevel) -> list[LogEntry]:
+    dependencies = DependencyContext.get()
+    all_logs = dependencies.get_dependency(LogKey())
+    return [entry for entry in all_logs if entry.level >= log_level]
+
+
+def get_trigger_bit(log_level: LogLevel) -> Value:
+    return reduce(operator.and_, [entry.trigger for entry in get_logs(log_level)], C(1))

transactron/testing/infrastructure.py

@@ -14,9 +14,11 @@
 from .profiler import profiler_process, Profile
 from .functions import TestGen
 from .assertion import make_assert_handler
+from .log import make_log_process
 from .gtkw_extension import write_vcd_ext
 from transactron import Method
 from transactron.lib import AdapterTrans
+from transactron.lib import log
 from transactron.core import TransactionManagerKey, TransactionModule
 from transactron.utils import ModuleConnector, HasElaborate, auto_debug_signals, HasDebugSignals
 
@@ -264,6 +266,7 @@ def run_simulation(self, module: HasElaborate, max_cycles: float = 10e4, add_tra
             )
 
         sim.add_sync_process(make_assert_handler(self.assertTrue))
+        sim.add_sync_process(make_log_process(log.DEBUG))
 
         res = sim.run()
 

transactron/testing/log.py

@@ -0,0 +1,40 @@
+from amaranth.sim import Passive, Tick
+from transactron.lib import log
+
+
+__all__ = ["make_log_process"]
+
+
+def make_log_process(log_level: log.LogLevel):
+    combined_trigger = log.get_trigger_bit(log_level)
+    cycle = 0
+
+    def handle_logs():
+        if not (yield combined_trigger):
+            return
+
+        for entry in log.get_logs(log_level):
+            if not (yield entry.trigger):
+                return
+
+            values: list[int] = []
+            for field in entry.fields:
+                values.append((yield field))
+            formatted_msg = entry.format(*values)
+            print(
+                "{} {} {}:{}] {}".format(
+                    log.get_level_name(entry.level), cycle, entry.location[0], entry.location[1], formatted_msg
+                )
+            )
+
+    def log_process():
+        nonlocal cycle
+
+        yield Passive()
+        while True:
+            yield Tick("sync_neg")
+            yield from handle_logs()
+            yield
+            cycle += 1
+
+    return log_process