Port ContentAddressableMemory from kuznia-rdzeni/coreblocks#395 (kuznia-rdzeni/coreblocks#573)

Author: lekcyjna123

test/test_transactron_lib_storage.py

@@ -0,0 +1,135 @@
+from datetime import timedelta
+from hypothesis import given, settings, Phase
+from transactron.testing import *
+from transactron.lib.storage import ContentAddressableMemory
+
+
+class TestContentAddressableMemory(TestCaseWithSimulator):
+    addr_width = 4
+    content_width = 5
+    test_number = 30
+    nop_number = 3
+    addr_layout = data_layout(addr_width)
+    content_layout = data_layout(content_width)
+
+    def setUp(self):
+        self.entries_count = 8
+
+        self.circ = SimpleTestCircuit(
+            ContentAddressableMemory(self.addr_layout, self.content_layout, self.entries_count)
+        )
+
+        self.memory = {}
+
+    def generic_process(
+        self,
+        method,
+        input_lst,
+        behaviour_check=None,
+        state_change=None,
+        input_verification=None,
+        settle_count=0,
+        name="",
+    ):
+        def f():
+            while input_lst:
+                # wait till all processes will end the previous cycle
+                yield from self.multi_settle(4)
+                elem = input_lst.pop()
+                if isinstance(elem, OpNOP):
+                    yield
+                    continue
+                if input_verification is not None and not input_verification(elem):
+                    yield
+                    continue
+                response = yield from method.call(**elem)
+                yield from self.multi_settle(settle_count)
+                if behaviour_check is not None:
+                    # Here accesses to circuit are allowed
+                    ret = behaviour_check(elem, response)
+                    if isinstance(ret, Generator):
+                        yield from ret
+                if state_change is not None:
+                    # It is standard python function by purpose to don't allow accessing circuit
+                    state_change(elem, response)
+                yield
+
+        return f
+
+    def push_process(self, in_push):
+        def verify_in(elem):
+            return not (frozenset(elem["addr"].items()) in self.memory)
+
+        def modify_state(elem, response):
+            self.memory[frozenset(elem["addr"].items())] = elem["data"]
+
+        return self.generic_process(
+            self.circ.push,
+            in_push,
+            state_change=modify_state,
+            input_verification=verify_in,
+            settle_count=3,
+            name="push",
+        )
+
+    def read_process(self, in_read):
+        def check(elem, response):
+            addr = elem["addr"]
+            frozen_addr = frozenset(addr.items())
+            if frozen_addr in self.memory:
+                assert response["not_found"] == 0
+                assert response["data"] == self.memory[frozen_addr]
+            else:
+                assert response["not_found"] == 1
+
+        return self.generic_process(self.circ.read, in_read, behaviour_check=check, settle_count=0, name="read")
+
+    def remove_process(self, in_remove):
+        def modify_state(elem, response):
+            if frozenset(elem["addr"].items()) in self.memory:
+                del self.memory[frozenset(elem["addr"].items())]
+
+        return self.generic_process(self.circ.remove, in_remove, state_change=modify_state, settle_count=2, name="remv")
+
+    def write_process(self, in_write):
+        def verify_in(elem):
+            ret = frozenset(elem["addr"].items()) in self.memory
+            return ret
+
+        def check(elem, response):
+            assert response["not_found"] == int(frozenset(elem["addr"].items()) not in self.memory)
+
+        def modify_state(elem, response):
+            if frozenset(elem["addr"].items()) in self.memory:
+                self.memory[frozenset(elem["addr"].items())] = elem["data"]
+
+        return self.generic_process(
+            self.circ.write,
+            in_write,
+            behaviour_check=check,
+            state_change=modify_state,
+            input_verification=None,
+            settle_count=1,
+            name="writ",
+        )
+
+    @settings(
+        max_examples=10,
+        phases=(Phase.explicit, Phase.reuse, Phase.generate, Phase.shrink),
+        derandomize=True,
+        deadline=timedelta(milliseconds=500),
+    )
+    @given(
+        generate_process_input(test_number, nop_number, [("addr", addr_layout), ("data", content_layout)]),
+        generate_process_input(test_number, nop_number, [("addr", addr_layout), ("data", content_layout)]),
+        generate_process_input(test_number, nop_number, [("addr", addr_layout)]),
+        generate_process_input(test_number, nop_number, [("addr", addr_layout)]),
+    )
+    def test_random(self, in_push, in_write, in_read, in_remove):
+        with self.reinitialize_fixtures():
+            self.setUp()
+            with self.run_simulation(self.circ, max_cycles=500) as sim:
+                sim.add_sync_process(self.push_process(in_push))
+                sim.add_sync_process(self.read_process(in_read))
+                sim.add_sync_process(self.write_process(in_write))
+                sim.add_sync_process(self.remove_process(in_remove))

test/utils/test_amaranth_ext.py

@@ -0,0 +1,92 @@
+from transactron.testing import *
+import random
+from transactron.utils.amaranth_ext import MultiPriorityEncoder
+
+
+class TestMultiPriorityEncoder(TestCaseWithSimulator):
+    def get_expected(self, input):
+        places = []
+        for i in range(self.input_width):
+            if input % 2:
+                places.append(i)
+            input //= 2
+        places += [None] * self.output_count
+        return places
+
+    def process(self):
+        for _ in range(self.test_number):
+            input = random.randrange(2**self.input_width)
+            yield self.circ.input.eq(input)
+            yield Settle()
+            expected_output = self.get_expected(input)
+            for ex, real, valid in zip(expected_output, self.circ.outputs, self.circ.valids):
+                if ex is None:
+                    assert (yield valid) == 0
+                else:
+                    assert (yield valid) == 1
+                    assert (yield real) == ex
+            yield Delay(1e-7)
+
+    @pytest.mark.parametrize("input_width", [1, 5, 16, 23, 24])
+    @pytest.mark.parametrize("output_count", [1, 3, 4])
+    def test_random(self, input_width, output_count):
+        random.seed(input_width + output_count)
+        self.test_number = 50
+        self.input_width = input_width
+        self.output_count = output_count
+        self.circ = MultiPriorityEncoder(self.input_width, self.output_count)
+
+        with self.run_simulation(self.circ) as sim:
+            sim.add_process(self.process)
+
+    @pytest.mark.parametrize("name", ["prio_encoder", None])
+    def test_static_create_simple(self, name):
+        random.seed(14)
+        self.test_number = 50
+        self.input_width = 7
+        self.output_count = 1
+
+        class DUT(Elaboratable):
+            def __init__(self, input_width, output_count, name):
+                self.input = Signal(input_width)
+                self.output_count = output_count
+                self.input_width = input_width
+                self.name = name
+
+            def elaborate(self, platform):
+                m = Module()
+                out, val = MultiPriorityEncoder.create_simple(m, self.input_width, self.input, name=self.name)
+                # Save as a list to use common interface in testing
+                self.outputs = [out]
+                self.valids = [val]
+                return m
+
+        self.circ = DUT(self.input_width, self.output_count, name)
+
+        with self.run_simulation(self.circ) as sim:
+            sim.add_process(self.process)
+
+    @pytest.mark.parametrize("name", ["prio_encoder", None])
+    def test_static_create(self, name):
+        random.seed(14)
+        self.test_number = 50
+        self.input_width = 7
+        self.output_count = 2
+
+        class DUT(Elaboratable):
+            def __init__(self, input_width, output_count, name):
+                self.input = Signal(input_width)
+                self.output_count = output_count
+                self.input_width = input_width
+                self.name = name
+
+            def elaborate(self, platform):
+                m = Module()
+                out = MultiPriorityEncoder.create(m, self.input_width, self.input, self.output_count, name=self.name)
+                self.outputs, self.valids = list(zip(*out))
+                return m
+
+        self.circ = DUT(self.input_width, self.output_count, name)
+
+        with self.run_simulation(self.circ) as sim:
+            sim.add_process(self.process)

transactron/lib/storage.py

@@ -3,12 +3,12 @@
 
 from transactron.utils.transactron_helpers import from_method_layout, make_layout
 from ..core import *
-from ..utils import SrcLoc, get_src_loc
+from ..utils import SrcLoc, get_src_loc, MultiPriorityEncoder
 from typing import Optional
-from transactron.utils import assign, AssignType, LayoutList
+from transactron.utils import assign, AssignType, LayoutList, MethodLayout
 from .reqres import ArgumentsToResultsZipper
 
-__all__ = ["MemoryBank", "AsyncMemoryBank"]
+__all__ = ["MemoryBank", "ContentAddressableMemory", "AsyncMemoryBank"]
 
 
 class MemoryBank(Elaboratable):
@@ -37,7 +37,7 @@ def __init__(
         elem_count: int,
         granularity: Optional[int] = None,
         safe_writes: bool = True,
-        src_loc: int | SrcLoc = 0
+        src_loc: int | SrcLoc = 0,
     ):
         """
         Parameters
@@ -138,6 +138,103 @@ def _(arg):
         return m
 
 
+class ContentAddressableMemory(Elaboratable):
+    """Content addresable memory
+
+    This module implements a content-addressable memory (in short CAM) with Transactron interface.
+    CAM is a type of memory where instead of predefined indexes there are used values fed in runtime
+    as keys (similar as in python dictionary). To insert new entry a pair `(key, value)` has to be
+    provided. Such pair takes an free slot which depends on internal implementation. To read value
+    a `key` has to be provided. It is compared with every valid key stored in CAM. If there is a hit,
+    a value is read. There can be many instances of the same key in CAM. In such case it is undefined
+    which value will be read.
+
+
+    .. warning::
+        Pushing the value with index already present in CAM is an undefined behaviour.
+
+    Attributes
+    ----------
+    read : Method
+        Nondestructive read
+    write : Method
+        If index present - do update
+    remove : Method
+        Remove
+    push : Method
+        Inserts new data.
+    """
+
+    def __init__(self, address_layout: MethodLayout, data_layout: MethodLayout, entries_number: int):
+        """
+        Parameters
+        ----------
+        address_layout : LayoutLike
+            The layout of the address records.
+        data_layout : LayoutLike
+            The layout of the data.
+        entries_number : int
+            The number of slots to create in memory.
+        """
+        self.address_layout = from_method_layout(address_layout)
+        self.data_layout = from_method_layout(data_layout)
+        self.entries_number = entries_number
+
+        self.read = Method(i=[("addr", self.address_layout)], o=[("data", self.data_layout), ("not_found", 1)])
+        self.remove = Method(i=[("addr", self.address_layout)])
+        self.push = Method(i=[("addr", self.address_layout), ("data", self.data_layout)])
+        self.write = Method(i=[("addr", self.address_layout), ("data", self.data_layout)], o=[("not_found", 1)])
+
+    def elaborate(self, platform) -> TModule:
+        m = TModule()
+
+        address_array = Array(
+            [Signal(self.address_layout, name=f"address_array_{i}") for i in range(self.entries_number)]
+        )
+        data_array = Array([Signal(self.data_layout, name=f"data_array_{i}") for i in range(self.entries_number)])
+        valids = Signal(self.entries_number, name="valids")
+
+        m.submodules.encoder_read = encoder_read = MultiPriorityEncoder(self.entries_number, 1)
+        m.submodules.encoder_write = encoder_write = MultiPriorityEncoder(self.entries_number, 1)
+        m.submodules.encoder_push = encoder_push = MultiPriorityEncoder(self.entries_number, 1)
+        m.submodules.encoder_remove = encoder_remove = MultiPriorityEncoder(self.entries_number, 1)
+        m.d.top_comb += encoder_push.input.eq(~valids)
+
+        @def_method(m, self.push, ready=~valids.all())
+        def _(addr, data):
+            id = Signal(range(self.entries_number), name="id_push")
+            m.d.top_comb += id.eq(encoder_push.outputs[0])
+            m.d.sync += address_array[id].eq(addr)
+            m.d.sync += data_array[id].eq(data)
+            m.d.sync += valids.bit_select(id, 1).eq(1)
+
+        @def_method(m, self.write)
+        def _(addr, data):
+            write_mask = Signal(self.entries_number, name="write_mask")
+            m.d.top_comb += write_mask.eq(Cat([addr == stored_addr for stored_addr in address_array]) & valids)
+            m.d.top_comb += encoder_write.input.eq(write_mask)
+            with m.If(write_mask.any()):
+                m.d.sync += data_array[encoder_write.outputs[0]].eq(data)
+            return {"not_found": ~write_mask.any()}
+
+        @def_method(m, self.read)
+        def _(addr):
+            read_mask = Signal(self.entries_number, name="read_mask")
+            m.d.top_comb += read_mask.eq(Cat([addr == stored_addr for stored_addr in address_array]) & valids)
+            m.d.top_comb += encoder_read.input.eq(read_mask)
+            return {"data": data_array[encoder_read.outputs[0]], "not_found": ~read_mask.any()}
+
+        @def_method(m, self.remove)
+        def _(addr):
+            rm_mask = Signal(self.entries_number, name="rm_mask")
+            m.d.top_comb += rm_mask.eq(Cat([addr == stored_addr for stored_addr in address_array]) & valids)
+            m.d.top_comb += encoder_remove.input.eq(rm_mask)
+            with m.If(rm_mask.any()):
+                m.d.sync += valids.bit_select(encoder_remove.outputs[0], 1).eq(0)
+
+        return m
+
+
 class AsyncMemoryBank(Elaboratable):
     """AsyncMemoryBank module.
 

transactron/testing/__init__.py

@@ -1,3 +1,4 @@
+from .input_generation import *  # noqa: F401
 from .functions import *  # noqa: F401
 from .infrastructure import *  # noqa: F401
 from .sugar import *  # noqa: F401