Port ContentAddressableMemory from #395

.gitignore

@@ -28,6 +28,7 @@ test/__profiles__/*.json
 pytestdebug.log
 _coreblocks_regression.lock
 _coreblocks_regression.counter
+.hypothesis
 
 # cocotb build
 /test/regression/cocotb/build

coreblocks/params/genparams.py

@@ -25,7 +25,7 @@ def __init__(self, cfg: CoreConfiguration):
         ext_partial, ext_full = extensions_supported(self.func_units_config, cfg.embedded, cfg.compressed)
         extensions = ext_partial if cfg.allow_partial_extensions else ext_full
         if not cfg.allow_partial_extensions and ext_partial != ext_full:
-            raise RuntimeError(f"Extensions {ext_partial&~ext_full!r} are only partially supported")
+            raise RuntimeError(f"Extensions {ext_partial & ~ext_full!r} are only partially supported")
 
         extensions |= cfg._implied_extensions
         self.isa_str = gen_isa_string(extensions, cfg.xlen)

requirements-dev.txt

@@ -1,7 +1,7 @@
 -r requirements.txt
-black==24.3.0
+black==24.4.2
 docutils==0.15.2
-flake8==6.0.0
+flake8==7.0.0
 pep8-naming==0.13.3
 markupsafe==2.0.1
 myst-parser==0.18.0
@@ -19,3 +19,4 @@ pytest-xdist==3.5.0
 pyelftools==0.29
 tabulate==0.9.0
 filelock==3.13.1
+hypothesis==6.99.6

test/frontend/test_decode_stage.py

@@ -12,7 +12,7 @@
 
 class TestDecode(TestCaseWithSimulator):
     @pytest.fixture(autouse=True)
-    def setup(self, configure_dependency_context):
+    def setup(self, fixture_initialize_testing_env):
         self.gen_params = GenParams(test_core_config.replace(start_pc=24))
 
         fifo_in = FIFO(self.gen_params.get(FetchLayouts).raw_instr, depth=2)

test/frontend/test_fetch.py

@@ -57,7 +57,7 @@ class TestFetchUnit(TestCaseWithSimulator):
     with_rvc: bool
 
     @pytest.fixture(autouse=True)
-    def setup(self, configure_dependency_context):
+    def setup(self, fixture_initialize_testing_env):
         self.pc = 0
         self.gen_params = GenParams(
             test_core_config.replace(
@@ -416,7 +416,7 @@ class TestPredictionChecker(TestCaseWithSimulator):
     with_rvc: bool
 
     @pytest.fixture(autouse=True)
-    def setup(self, configure_dependency_context):
+    def setup(self, fixture_initialize_testing_env):
         self.gen_params = GenParams(
             test_core_config.replace(compressed=self.with_rvc, fetch_block_bytes_log=self.fetch_block_log)
         )

test/func_blocks/fu/functional_common.py

@@ -95,7 +95,7 @@ def compute_result(i1: int, i2: int, i_imm: int, pc: int, fn: _T, xlen: int) ->
         raise NotImplementedError
 
     @pytest.fixture(autouse=True)
-    def setup(self, configure_dependency_context):
+    def setup(self, fixture_initialize_testing_env):
         self.gen_params = GenParams(test_core_config)
 
         self.report_mock = TestbenchIO(Adapter(i=self.gen_params.get(ExceptionRegisterLayouts).report))

test/func_blocks/lsu/test_dummylsu.py

@@ -151,7 +151,7 @@ def generate_instr(self, max_reg_val, max_imm_val):
                     "addr": word_addr,
                     "mask": mask,
                     "sign": signess,
-                    "rnd_bytes": bytes.fromhex(f"{random.randint(0,2**32-1):08x}"),
+                    "rnd_bytes": bytes.fromhex(f"{random.randint(0, 2**32-1):08x}"),
                     "misaligned": misaligned,
                     "err": bus_err,
                 }
@@ -262,7 +262,7 @@ def generate_instr(self, max_reg_val, max_imm_val):
         wish_data = {
             "addr": (s1_val + imm) >> 2,
             "mask": 0xF,
-            "rnd_bytes": bytes.fromhex(f"{random.randint(0,2**32-1):08x}"),
+            "rnd_bytes": bytes.fromhex(f"{random.randint(0, 2**32-1):08x}"),
         }
         return instr, wish_data
 

test/transactron/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/transactron/utils/test_amaranth_ext.py

@@ -0,0 +1,40 @@
+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(14)
+        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)

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 feed in runtime
+    as keys (smimlar 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