Port ContentAddressableMemory from #395

test/common/__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

test/common/input_generation.py

@@ -0,0 +1,17 @@
+import random
+from typing import Optional
+from transactron.utils import SimpleLayout
+
+
+def generate_based_on_layout(layout: SimpleLayout, *, max_bits: Optional[int] = None):
+    d = {}
+    for elem in layout:
+        if isinstance(elem[1], int):
+            if max_bits is None:
+                max_val = 2 ** elem[1]
+            else:
+                max_val = 2 ** min(max_bits, elem[1])
+            d[elem[0]] = random.randrange(max_val)
+        else:
+            d[elem[0]] = generate_based_on_layout(elem[1])
+    return d

test/transactions/test_transactron_lib_storage.py

@@ -0,0 +1,50 @@
+from test.common import *
+import random
+from transactron.lib.storage import ContentAddressableMemory
+
+
+class TestContentAddressableMemory(TestCaseWithSimulator):
+    def setUp(self):
+        random.seed(14)
+        self.test_number = 50
+        self.addr_width = 4
+        self.content_width = 5
+        self.entries_count = 8
+        self.addr_layout = data_layout(self.addr_width)
+        self.content_layout = data_layout(self.content_width)
+
+        self.circ = SimpleTestCircuit(
+            ContentAddressableMemory(self.addr_layout, self.content_layout, self.entries_count)
+        )
+
+        self.memory = {}
+
+    def input_process(self):
+        for _ in range(self.test_number):
+            while True:
+                addr = generate_based_on_layout(self.addr_layout)
+                frozen_addr = frozenset(addr.items())
+                if frozen_addr not in self.memory:
+                    break
+            content = generate_based_on_layout(self.content_layout)
+            yield from self.circ.push.call(addr=addr, data=content)
+            yield Settle()
+            self.memory[frozen_addr] = content
+
+    def output_process(self):
+        yield Passive()
+        while True:
+            addr = generate_based_on_layout(self.addr_layout)
+            res = yield from self.circ.pop.call(addr=addr)
+            frozen_addr = frozenset(addr.items())
+            if frozen_addr in self.memory:
+                self.assertEqual(res["not_found"], 0)
+                self.assertEqual(res["data"], self.memory[frozen_addr])
+                self.memory.pop(frozen_addr)
+            else:
+                self.assertEqual(res["not_found"], 1)
+
+    def test_random(self):
+        with self.run_simulation(self.circ) as sim:
+            sim.add_sync_process(self.input_process)
+            sim.add_sync_process(self.output_process)

test/utils/test_amaranth_ext.py

@@ -0,0 +1,40 @@
+from test.common import *
+import random
+from transactron.utils.amaranth_ext import MultiPriorityEncoder
+
+
+class TestMultiPriorityEncoder(TestCaseWithSimulator):
+    def setUp(self):
+        random.seed(14)
+        self.test_number = 50
+        self.input_width = 16
+        self.output_count = 4
+
+        self.circ = MultiPriorityEncoder(self.input_width, self.output_count)
+
+    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:
+                    self.assertEqual((yield valid), 0)
+                else:
+                    self.assertEqual((yield valid), 1)
+                    self.assertEqual((yield real), ex)
+            yield Delay(1e-7)
+
+    def test_random(self):
+        with self.run_simulation(self.circ) as sim:
+            sim.add_process(self.process)

transactron/lib/storage.py

@@ -1,12 +1,12 @@
 from amaranth import *
 from amaranth.utils import *
 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
+from transactron.utils import assign, AssignType, LayoutLike
 from .reqres import ArgumentsToResultsZipper
 
-__all__ = ["MemoryBank"]
+__all__ = ["MemoryBank", "ContentAddressableMemory"]
 
 
 class MemoryBank(Elaboratable):
@@ -133,3 +133,76 @@ def _(arg):
             m.d.comb += assign(write_args, arg, fields=AssignType.ALL)
 
         return m
+
+
+class ContentAddressableMemory(Elaboratable):
+    """Content addresable memory
+
+    This module implements a transactorn interface for the content addressable memory.
+
+    .. warning::
+       Current implementation has critical path O(entries_number). If needed we can
+       optimise it in future to have O(log(entries_number)).
+
+
+    Attributes
+    ----------
+    pop : Method
+        Looks for the data in memory and, if found, returns it and removes it.
+    push : Method
+        Inserts new data.
+    """
+
+    def __init__(self, address_layout: LayoutLike, data_layout: LayoutLike, 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 = address_layout
+        self.data_layout = data_layout
+        self.entries_number = entries_number
+
+        self.pop = Method(i=[("addr", self.address_layout)], o=[("data", self.data_layout), ("not_found", 1)])
+        self.push = Method(i=[("addr", self.address_layout), ("data", self.data_layout)])
+
+    def elaborate(self, platform) -> TModule:
+        m = TModule()
+
+        address_array = Array([Record(self.address_layout) for _ in range(self.entries_number)])
+        data_array = Array([Record(self.data_layout) for _ in range(self.entries_number)])
+        valids = Signal(self.entries_number, name="valids")
+
+        m.submodules.encoder_addr = encoder_addr = MultiPriorityEncoder(self.entries_number, 1)
+        m.submodules.encoder_valids = encoder_valids = MultiPriorityEncoder(self.entries_number, 1)
+        m.d.comb += encoder_valids.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.comb += id.eq(encoder_valids.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)
+
+        if_addr = Signal(self.entries_number, name="if_addr")
+        data_to_send = Record(self.data_layout)
+
+        @def_method(m, self.pop)
+        def _(addr):
+            m.d.top_comb += if_addr.eq(Cat([addr == stored_addr for stored_addr in address_array]) & valids)
+            id = encoder_addr.outputs[0]
+            with m.If(if_addr.any()):
+                m.d.comb += data_to_send.eq(data_array[id])
+                m.d.sync += valids.bit_select(id, 1).eq(0)
+
+            return {"data": data_to_send, "not_found": ~if_addr.any()}
+
+        m.d.comb += encoder_addr.input.eq(if_addr)
+
+        return m

transactron/utils/_typing.py

@@ -1,4 +1,5 @@
 from typing import (
+    Tuple,
     Callable,
     Concatenate,
     Generic,
@@ -27,6 +28,7 @@
     "ValueLike",
     "StatementLike",
     "LayoutLike",
+    "SimpleLayout",
     "SwitchKey",
     "MethodLayout",
     "SrcLoc",
@@ -61,6 +63,7 @@
 SignalBundle: TypeAlias = Signal | Record | View | Iterable["SignalBundle"] | Mapping[str, "SignalBundle"]
 LayoutListField: TypeAlias = tuple[str, "ShapeLike | LayoutList"]
 LayoutList: TypeAlias = list[LayoutListField]
+SimpleLayout = list[Tuple[str, Union[int, "SimpleLayout"]]]
 
 RecordIntDict: TypeAlias = Mapping[str, Union[int, "RecordIntDict"]]
 RecordIntDictRet: TypeAlias = Mapping[str, Any]  # full typing hard to work with

transactron/utils/amaranth_ext/elaboratables.py

@@ -11,6 +11,7 @@
     "ModuleConnector",
     "Scheduler",
     "RoundRobin",
+    "MultiPriorityEncoder",
 ]
 
 
@@ -239,3 +240,63 @@ def elaborate(self, platform):
         m.d.sync += self.valid.eq(self.requests.any())
 
         return m
+
+
+class MultiPriorityEncoder(Elaboratable):
+    """Priority encoder with more outputs
+
+    This is an extension of the `PriorityEncoder` from amaranth that supports
+    more than one output from an input signal. In other words
+    it decodes multi-hot encoded signal into lists of signals in binary
+    format, each with the index of a different high bit in the input.
+
+    Attributes
+    ----------
+    input_width : int
+        Width of the input signal
+    outputs_count : int
+        Number of outputs to generate at once.
+    input : Signal, in
+        Signal with 1 on `i`-th bit if `i` can be selected by encoder
+    outputs : list[Signal], out
+        Signals with selected indicies, sorted in ascending order,
+        if the number of ready signals is less than `outputs_count`
+        then valid signals are at the beginning of the list.
+    valids : list[Signals], out
+        One bit for each output signal, indicating whether the output is valid or not.
+    """
+
+    def __init__(self, input_width: int, outputs_count: int):
+        self.input_width = input_width
+        self.outputs_count = outputs_count
+
+        self.input = Signal(self.input_width)
+        self.outputs = [Signal(range(self.input_width), name="output") for _ in range(self.outputs_count)]
+        self.valids = [Signal(name="valid") for _ in range(self.outputs_count)]
+
+    def elaborate(self, platform):
+        m = Module()
+
+        current_outputs = [Signal(range(self.input_width)) for _ in range(self.outputs_count)]
+        current_valids = [Signal() for _ in range(self.outputs_count)]
+        for j in reversed(range(self.input_width)):
+            new_current_outputs = [Signal(range(self.input_width)) for _ in range(self.outputs_count)]
+            new_current_valids = [Signal() for _ in range(self.outputs_count)]
+            with m.If(self.input[j]):
+                m.d.comb += new_current_outputs[0].eq(j)
+                m.d.comb += new_current_valids[0].eq(1)
+                for k in range(self.outputs_count - 1):
+                    m.d.comb += new_current_outputs[k + 1].eq(current_outputs[k])
+                    m.d.comb += new_current_valids[k + 1].eq(current_valids[k])
+            with m.Else():
+                for k in range(self.outputs_count):
+                    m.d.comb += new_current_outputs[k].eq(current_outputs[k])
+                    m.d.comb += new_current_valids[k].eq(current_valids[k])
+            current_outputs = new_current_outputs
+            current_valids = new_current_valids
+
+        for k in range(self.outputs_count):
+            m.d.comb += self.outputs[k].eq(current_outputs[k])
+            m.d.comb += self.valids[k].eq(current_valids[k])
+
+        return m

transactron/utils/data_repr.py

@@ -1,5 +1,5 @@
 from collections.abc import Iterable, Mapping
-from ._typing import LayoutList, ShapeLike, LayoutLike
+from ._typing import LayoutList, SimpleLayout
 from typing import Any, Sized
 from statistics import fmean
 
@@ -77,7 +77,7 @@ def bits_from_int(num: int, lower: int, length: int):
     return (num >> lower) & ((1 << (length)) - 1)
 
 
-def data_layout(val: ShapeLike) -> LayoutLike:
+def data_layout(val: int) -> SimpleLayout:
     return [("data", val)]