Adding support for scoreboard filtering functions (#20)

When you register a monitor, you can now provide a scoreboard_filter
argument in order to transform/drop a transaction:

```python
from copy import copy

class Testbench:
    def __init__(self, dut):
        ...
        self.register(
            "x_mon",
            StreamMonitor(...),
            scoreboard_filter=self.filter_x_mon
        )

    def filter_x_mon(self, component, event, transaction):
        masked = copy(transaction)
        masked.data &= 0x0000_FFFF
        return masked
```

You can also entirely drop a transaction by returning `None` from the
filter function.

This PR also fixes an issue where if no channels of a filter funnel
exist but all contain values, the scoreboard did not detect the early
failure and just timed out.

example/testbench/testbench.py

@@ -17,6 +17,7 @@
 from forastero.bench import BaseBench
 from forastero.driver import DriverEvent
 from forastero.io import IORole
+from forastero.monitor import MonitorEvent
 
 from .stream import (
     StreamInitiator,
@@ -57,6 +58,7 @@ def __init__(self, dut: HierarchyObject) -> None:
             "x_mon",
             StreamMonitor(self, x_io, self.clk, self.rst),
             scoreboard_queues=["a", "b"],
+            scoreboard_filter=self.filter_x_mon,
         )
         # Register callbacks to the model
         self.a_init.subscribe(DriverEvent.POST_DRIVE, self.model)
@@ -65,9 +67,25 @@ def __init__(self, dut: HierarchyObject) -> None:
     def model(
         self, driver: StreamInitiator, event: DriverEvent, obj: StreamTransaction
     ) -> None:
+        """
+        Demonstration model that forwards transactions seen on interfaces A & B
+        and sets bit 32 (to match the filtering behaviour below)
+        """
         assert driver in (self.a_init, self.b_init)
         assert event == DriverEvent.POST_DRIVE
+        exp = StreamTransaction(data=obj.data | (1 << 32))
         if driver is self.a_init:
-            self.scoreboard.channels["x_mon"].push_reference("a", obj)
+            self.scoreboard.channels["x_mon"].push_reference("a", exp)
         else:
-            self.scoreboard.channels["x_mon"].push_reference("b", obj)
+            self.scoreboard.channels["x_mon"].push_reference("b", exp)
+
+    def filter_x_mon(
+        self, monitor: StreamMonitor, event: MonitorEvent, obj: StreamTransaction
+    ) -> StreamTransaction:
+        """
+        Demonstration filter function that modifies the data captured from the
+        X interface by always setting bit 32
+        """
+        assert monitor is self.x_mon
+        assert event is MonitorEvent.CAPTURE
+        return StreamTransaction(data=obj.data | (1 << 32))

forastero/bench.py

@@ -205,6 +205,7 @@ def register(
         scoreboard: bool = True,
         scoreboard_verbose: bool = False,
         scoreboard_queues: list[str] | None = None,
+        scoreboard_filter: Callable | None = None,
     ) -> Component | Coroutine:
         """
         Register a driver, monitor, or coroutine with the testbench. Drivers and
@@ -220,6 +221,11 @@ def register(
         :param scoreboard_verbose: Only applies to scoreboarded monitors,
                                    controls whether to log each transaction,
                                    even when they don't mismatch
+        :param scoreboard_queues:  A list of named queues used when a funnel
+                                   type scoreboard channel is required
+        :param scoreboard_filter:  A function that can filter or modify items
+                                   recorded by the monitor before they are passed
+                                   to the scoreboard
         """
         assert isinstance(name, str), f"Name must be a string '{name}'"
         if asyncio.iscoroutine(comp_or_coro):
@@ -234,7 +240,10 @@ def register(
             comp_or_coro.seed(self.random)
             if scoreboard and isinstance(comp_or_coro, BaseMonitor):
                 self.scoreboard.attach(
-                    comp_or_coro, verbose=scoreboard_verbose, queues=scoreboard_queues
+                    comp_or_coro,
+                    verbose=scoreboard_verbose,
+                    filter_fn=scoreboard_filter,
+                    queues=scoreboard_queues,
                 )
         else:
             raise TypeError(f"Unsupported object: {comp_or_coro}")

forastero/scoreboard.py

@@ -92,15 +92,19 @@ class Channel:
     both queues and tested for equivalence. Any mismatches are reported to the
     scoreboard.
 
-    :param name:    Name of this scoreboard channel
-    :param monitor: Handle to the monitor capturing traffic from the DUT
-    :param log:     Handle to the scoreboard log
+    :param name:      Name of this scoreboard channel
+    :param monitor:   Handle to the monitor capturing traffic from the DUT
+    :param log:       Handle to the scoreboard log
+    :param filter_fn: Function to filter or modify captured transactions
     """
 
-    def __init__(self, name: str, monitor: BaseMonitor, log: SimLog) -> None:
+    def __init__(
+        self, name: str, monitor: BaseMonitor, log: SimLog, filter_fn: Callable | None
+    ) -> None:
         self.name = name
         self.monitor = monitor
         self.log = log
+        self.filter_fn = filter_fn
         self._q_mon = Queue()
         self._q_ref = Queue()
         self._lock = Lock()
@@ -109,7 +113,12 @@ def __init__(self, name: str, monitor: BaseMonitor, log: SimLog) -> None:
 
         def _sample(mon: BaseMonitor, evt: MonitorEvent, obj: BaseTransaction) -> None:
             if mon is self.monitor and evt is MonitorEvent.CAPTURE:
-                self.push_monitor(obj)
+                # If a filter function was provided, apply it
+                if self.filter_fn is not None:
+                    obj = self.filter_fn(mon, evt, obj)
+                # A filter can drop the transaction, so test for None
+                if obj is not None:
+                    self.push_monitor(obj)
 
         self.monitor.subscribe(MonitorEvent.CAPTURE, _sample)
 
@@ -241,12 +250,23 @@ class FunnelChannel(Channel):
     is not strictly defined, often due to the hardware interleaving different
     streams. Reference data can be pushed into one or more named queues and when
     monitor packets arrive any queue head is valid.
+
+    :param name:       Name of this scoreboard channel
+    :param monitor:    Handle to the monitor capturing traffic from the DUT
+    :param log:        Handle to the scoreboard log
+    :param filter_fn:  Function to filter or modify captured transactions
+    :param ref_queues: List of reference queue names
     """
 
     def __init__(
-        self, name: str, monitor: BaseMonitor, log: SimLog, ref_queues: list[str]
+        self,
+        name: str,
+        monitor: BaseMonitor,
+        log: SimLog,
+        filter_fn: Callable | None,
+        ref_queues: list[str],
     ) -> None:
-        super().__init__(name, monitor, log)
+        super().__init__(name, monitor, log, filter_fn)
         self._q_ref = {x: Queue() for x in ref_queues}
 
     @property
@@ -281,11 +301,19 @@ async def _dequeue(self) -> tuple[BaseTransaction, BaseTransaction]:
         # Peek at the front of all of the queues
         while True:
             next_mon = self._q_mon.peek()
+            any_empty = False
             for queue in self._q_ref.values():
+                any_empty = any_empty or (queue.level == 0)
                 if queue.level > 0 and queue.peek() == next_mon:
                     await self._q_mon.pop()
                     next_ref = await queue.pop()
                     return next_mon, next_ref
+            # If all queues contain objects but none matched, this is a mismatch!
+            # NOTE: This will just pop the final queue in order to report miscompare
+            if not any_empty:
+                await self._q_mon.pop()
+                next_ref = await queue.pop()
+                return next_mon, next_ref
             # Wait for a reference object to be pushed to any queue
             await First(*(x.on_push_event.wait() for x in self._q_ref.values()))
 
@@ -354,24 +382,38 @@ def __init__(self, tb: "BaseBench", fail_fast: bool = False):  # noqa: F821
         self.channels: dict[str, Channel] = {}
 
     def attach(
-        self, monitor: BaseMonitor, verbose=False, queues: list[str] | None = None
+        self,
+        monitor: BaseMonitor,
+        verbose=False,
+        filter_fn: Callable | None = None,
+        queues: list[str] | None = None,
     ) -> None:
         """
         Attach a monitor to the scoreboard, creating and scheduling a new
         channel in the process.
 
-        :param monitor: The monitor to attach
-        :param verbose: Whether to tabulate matches as well as mismatches
-        :param queues:  List of reference queue names
+        :param monitor:   The monitor to attach
+        :param verbose:   Whether to tabulate matches as well as mismatches
+        :param queues:    List of reference queue names, this causes a funnel
+                          type scoreboard channel to be used
+        :param filter_fn: A filter function that can either drop or modify items
+                          recorded by the monitor prior to scoreboarding
         """
         assert monitor.name not in self.channels, f"Monitor known for '{monitor.name}'"
         if isinstance(queues, list) and len(queues) > 0:
             channel = FunnelChannel(
-                monitor.name, monitor, self.tb.fork_log("channel", monitor.name), queues
+                monitor.name,
+                monitor,
+                self.tb.fork_log("channel", monitor.name),
+                filter_fn,
+                queues,
             )
         else:
             channel = Channel(
-                monitor.name, monitor, self.tb.fork_log("channel", monitor.name)
+                monitor.name,
+                monitor,
+                self.tb.fork_log("channel", monitor.name),
+                filter_fn,
             )
         self.channels[channel.name] = channel
         if verbose: