Merge branch 'master' into nonexclusive-wrapper

pyproject.toml

@@ -40,6 +40,9 @@ dev = [
     "pre-commit == 2.16.0"
 ]
 
+[project.scripts]
+transactron-prof = "transactron.cmd.tprof:main"
+
 [tool.setuptools_scm]
 
 [tool.pyright]

test/lib/test_connectors.py

@@ -1,20 +1,17 @@
 import pytest
 import random
-from operator import and_
-from functools import reduce
 from typing import TypeAlias
+from collections import defaultdict
 
 from amaranth import *
 from transactron import *
 from transactron.utils._typing import MethodLayout
-from transactron.lib.adapters import Adapter
 from transactron.lib.connectors import *
 from transactron.testing.testbenchio import CallTrigger
 from transactron.testing import (
     SimpleTestCircuit,
     TestCaseWithSimulator,
     data_layout,
-    TestbenchIO,
     TestbenchContext,
 )
 
@@ -124,26 +121,16 @@ def test_pipelining(self):
 
 
 class ManyToOneConnectTransTestCircuit(Elaboratable):
-    def __init__(self, count: int, lay: MethodLayout):
-        self.count = count
-        self.lay = lay
-        self.inputs: list[TestbenchIO] = []
+    inputs: Required[list[Method]]
+    output: Required[Method]
+
+    def __init__(self, count: int, layout: MethodLayout):
+        self.inputs = [Method(o=layout) for _ in range(count)]
+        self.output = Method(i=layout)
 
     def elaborate(self, platform):
         m = TModule()
-
-        get_results = []
-        for i in range(self.count):
-            input = TestbenchIO(Adapter.create(o=self.lay))
-            get_results.append(input.adapter.iface)
-            m.submodules[f"input_{i}"] = input
-            self.inputs.append(input)
-
-        # Create ManyToOneConnectTrans, which will serialize results from different inputs
-        output = TestbenchIO(Adapter.create(i=self.lay))
-        m.submodules.output = self.output = output
-        m.submodules.fu_arbitration = ManyToOneConnectTrans(get_results=get_results, put_result=output.adapter.iface)
-
+        m.submodules.fu_arbitration = ManyToOneConnectTrans(get_results=self.inputs, put_result=self.output)
         return m
 
 
@@ -153,13 +140,13 @@ def initialize(self):
         f2_size = 3
         self.lay = [("field1", f1_size), ("field2", f2_size)]
 
-        self.m = ManyToOneConnectTransTestCircuit(self.count, self.lay)
+        self.m = SimpleTestCircuit(ManyToOneConnectTransTestCircuit(self.count, self.lay))
         random.seed(14)
 
         self.inputs = []
         # Create list with info if we processed all data from inputs
         self.producer_end = [False for i in range(self.count)]
-        self.expected_output = {}
+        self.expected_output = defaultdict(int)
         self.max_wait = 4
 
         # Prepare random results for inputs
@@ -168,45 +155,30 @@ def initialize(self):
             input_size = random.randint(20, 30)
             for j in range(input_size):
                 t = (
-                    random.randint(0, 2**f1_size),
-                    random.randint(0, 2**f2_size),
+                    random.randrange(0, 2**f1_size),
+                    random.randrange(0, 2**f2_size),
                 )
                 data.append(t)
-                if t in self.expected_output:
-                    self.expected_output[t] += 1
-                else:
-                    self.expected_output[t] = 1
+                self.expected_output[t] += 1
             self.inputs.append(data)
 
     def generate_producer(self, i: int):
-        """
-        This is an helper function, which generates a producer process,
-        which will simulate an FU. Producer will insert in random intervals new
-        results to its output FIFO. This records will be next serialized by FUArbiter.
-        """
-
         async def producer(sim: TestbenchContext):
             inputs = self.inputs[i]
             for field1, field2 in inputs:
-                self.m.inputs[i].call_init(sim, field1=field1, field2=field2)
+                await self.m.inputs[i].call(sim, field1=field1, field2=field2)
                 await self.random_wait(sim, self.max_wait)
             self.producer_end[i] = True
 
         return producer
 
     async def consumer(self, sim: TestbenchContext):
-        # TODO: this test doesn't test anything, needs to be fixed!
-        while reduce(and_, self.producer_end, True):
-            result = await self.m.output.call_do(sim)
-
-            assert result is not None
-
-            t = (result["field1"], result["field2"])
-            assert t in self.expected_output
-            if self.expected_output[t] == 1:
-                del self.expected_output[t]
-            else:
-                self.expected_output[t] -= 1
+        while not all(self.producer_end):
+            result = await self.m.output.call(sim)
+
+            t = (result.field1, result.field2)
+            assert self.expected_output[t]
+            self.expected_output[t] -= 1
             await self.random_wait(sim, self.max_wait)
 
     @pytest.mark.parametrize("count", [1, 4])

test/lib/test_simultaneous.py

@@ -89,3 +89,38 @@ async def process(sim: TestbenchContext):
 
         with self.run_simulation(m) as sim:
             sim.add_testbench(process)
+
+
+class UncalledMethodTestCircuit(Elaboratable):
+    def __init__(self):
+        self.sig = Signal()
+
+    def elaborate(self, platform):
+        m = TModule()
+
+        uncalled_method = Method()
+        internal_method = Method()
+
+        @def_method(m, uncalled_method)
+        def _():
+            with condition(m, nonblocking=True) as branch:
+                with branch(True):
+                    internal_method(m)
+
+        @def_method(m, internal_method)
+        def _():
+            m.d.comb += self.sig.eq(1)
+
+        return m
+
+
+class TestUncalledMethod(TestCaseWithSimulator):
+    def test_uncalled_method(self):
+        circ = UncalledMethodTestCircuit()
+
+        async def process(sim: TestbenchContext):
+            _, _, val = await sim.tick().sample(circ.sig)
+            assert val == 0
+
+        with self.run_simulation(circ) as sim:
+            sim.add_testbench(process)

test/lib/test_storage.py

@@ -5,8 +5,11 @@
 from collections import deque
 from datetime import timedelta
 from hypothesis import given, settings, Phase
+import amaranth.lib.memory as memory
+import amaranth_types.memory as amemory
 from transactron.testing import *
 from transactron.lib.storage import *
+from transactron.utils.amaranth_ext.memory import MultiportXORMemory, MultiportILVTMemory
 from transactron.utils.transactron_helpers import make_layout
 
 
@@ -150,6 +153,7 @@ class TestMemoryBank(TestCaseWithSimulator):
     @pytest.mark.parametrize("transparent", [False, True])
     @pytest.mark.parametrize("read_ports", [1, 2])
     @pytest.mark.parametrize("write_ports", [1, 2])
+    @pytest.mark.parametrize("memory_type", [memory.Memory, MultiportXORMemory, MultiportILVTMemory])
     @pytest.mark.parametrize("shape,to_shape,from_shape", bank_shapes)
     def test_mem(
         self,
@@ -161,6 +165,7 @@ def test_mem(
         transparent: bool,
         read_ports: int,
         write_ports: int,
+        memory_type: amemory.AbstractMemoryConstructor[ShapeLike, Value],
         shape: ShapeLike,
         to_shape: Callable,
         from_shape: Callable,
@@ -174,11 +179,13 @@ def test_mem(
                 transparent=transparent,
                 read_ports=read_ports,
                 write_ports=write_ports,
+                memory_type=memory_type,
             ),
         )
 
         data: list[int] = [0 for _ in range(max_addr)]
         read_req_queues = [deque() for _ in range(read_ports)]
+        address_lock = [False] * max_addr
 
         random.seed(seed)
 
@@ -187,9 +194,19 @@ async def process(sim: TestbenchContext):
                 for cycle in range(test_count):
                     d = random.randrange(2 ** Shape.cast(shape).width)
                     a = random.randrange(max_addr)
+
+                    # one address shouldn't be written by multiple ports at the same time
+                    while address_lock[a]:
+                        a = random.randrange(max_addr)
+                    address_lock[a] = True
+
                     await m.write[i].call(sim, data=to_shape(d), addr=a)
+
                     await sim.delay(1e-9 * (i + 2 if not transparent else i))
                     data[a] = d
+
+                    address_lock[a] = False
+
                     await self.random_wait(sim, writer_rand)
 
             return process

scripts/tprof.py → transactron/cmd/tprof.py

@@ -1,19 +1,10 @@
-#!/usr/bin/env python3
-
 import argparse
-import sys
 import re
-from pathlib import Path
 from typing import Optional
 from collections.abc import Callable, Iterable
 from tabulate import tabulate
 from dataclasses import asdict
-
-topdir = Path(__file__).parent.parent
-sys.path.insert(0, str(topdir))
-
-
-from transactron.profiler import Profile, RunStat, RunStatNode  # noqa: E402
+from transactron.profiler import Profile, RunStat, RunStatNode
 
 
 def process_stat_tree(

transactron/core/manager.py

@@ -261,13 +261,21 @@ def _simultaneous(self):
         simultaneous = set[frozenset[TBody]]()
 
         for elem in method_map.methods_and_transactions:
+            pruned_sim = False
             for sim_elem in elem.simultaneous_list:
+                if sim_elem not in method_map.methods_and_transactions:
+                    if sim_elem.independent_list:
+                        raise RuntimeError("Pruned method with independent list not supported")
+                    pruned_sim = True
+                    continue
                 for tr1, tr2 in product(method_map.transactions_for(elem), method_map.transactions_for(sim_elem)):
                     if tr1 in independents[tr2]:
                         raise RuntimeError(
                             f"Unsatisfiable simultaneity constraints for '{elem.name}' and '{sim_elem.name}'"
                         )
                     simultaneous.add(frozenset({tr1, tr2}))
+            if pruned_sim and elem in method_map.transactions:
+                elem.ready = Signal()
 
         # step 2: transitivity computation
         tr_simultaneous = set[frozenset[TBody]]()

transactron/core/method.py

@@ -325,23 +325,25 @@ def debug_signals(self) -> SignalBundle:
 class Methods(Sequence[Method]):
     @type_self_add_1pos_kwargs_as(Method.__init__)
     def __init__(self, count: int, **kwargs):
-        if count <= 0:
-            raise ValueError("count should be at least 1")
+        if count < 0:
+            raise ValueError("count should be at least 0")
         _, owner_name = get_caller_class_name(default="$method")
         self.name = kwargs["name"] if "name" in kwargs else tracer.get_var_name(depth=2, default=owner_name)
         if "src_loc" not in kwargs:
             kwargs["src_loc"] = 0
         if isinstance(kwargs["src_loc"], int):
             kwargs["src_loc"] += 1
         self._methods = [Method(**{**kwargs, "name": f"{self.name}{i}"}) for i in range(count)]
+        self._layout_in = from_method_layout(kwargs["i"] if "i" in kwargs else ())
+        self._layout_out = from_method_layout(kwargs["o"] if "o" in kwargs else ())
 
     @property
     def layout_in(self):
-        return self._methods[0].layout_in
+        return self._layout_in
 
     @property
     def layout_out(self):
-        return self._methods[0].layout_out
+        return self._layout_out
 
     def __call__(
         self, m: TModule, arg: Optional[AssignArg] = None, enable: ValueLike = C(1), /, **kwargs: AssignArg

transactron/core/transaction_base.py

@@ -102,6 +102,8 @@ def simultaneous(self, *others: _T) -> None:
             The `Transaction`\\s or `Method`\\s to be executed simultaneously.
         """
         self.simultaneous_list += others
+        for other in others:
+            other.simultaneous_list.append(self)  # type: ignore
 
     def simultaneous_alternatives(self, *others: _T) -> None:
         """Adds exclusive simultaneity relations.

transactron/utils/amaranth_ext/elaboratables.py

@@ -3,6 +3,7 @@
 from typing import Literal, Optional, overload
 from collections.abc import Iterable
 from amaranth import *
+from amaranth import ValueCastable
 from amaranth.lib.data import ArrayLayout
 from amaranth_types import ShapeLike
 from transactron.utils._typing import HasElaborate, ModuleLike, ValueLike
@@ -610,3 +611,68 @@ def elaborate(self, platform):
         m.d.comb += self.output_cnt.eq(total_cnt)
 
         return m
+
+
+class OneHotMux(Elaboratable):
+    """One-hot multiplexer.
+
+    If all select bits are 0, the `output` signal is set to `default_input`.
+    In the other case, the `output` signal is set to the input which
+    select bit is set. It is assumed that at most one `select` bit is set.
+
+    Attributes
+    ----------
+    inputs: Signal(ArrayLayout(shape, inputs_count)), in
+        Input signals.
+    select: Signal(inputs_count), in
+        Selection signal. When one of the select bits is set,
+        the corresponding input is assigned to `output`.
+    default_input: Signal(shape), in
+        Default input signal.
+    output: Signal(shape), out
+        Output signal. It is set to `default_input` or one of `inputs`
+        depending on `select`.
+    """
+
+    def __init__(self, shape: ShapeLike, inputs_count: int):
+        self.inputs = Signal(ArrayLayout(shape, inputs_count))
+        self.select = Signal(inputs_count)
+        self.default_input = Signal(shape)
+        self.output = Signal(shape)
+
+    @staticmethod
+    def create(m: ModuleLike, inputs: Iterable[tuple[ValueLike, ValueLike]], default_input: ValueLike) -> ValueLike:
+        """Syntax sugar for creating a `OneHotMux`.
+
+        Parameters
+        ----------
+        m: Module
+            Module to add the `OneHotMux` to.
+        default_input: ValueLike
+            Default input.
+        forward_inputs: Iterable[tuple[ValueLike, ValueLike]]
+            Select bits and corresponding inputs.
+        """
+        if isinstance(default_input, ValueCastable):
+            input_shape = default_input.shape()
+        else:
+            input_shape = Value.cast(default_input).shape()
+        inputs = list(inputs)
+        fw_net = OneHotMux(input_shape, len(inputs))
+        m.submodules += fw_net
+        m.d.comb += Value.cast(fw_net.default_input).eq(default_input)
+        for i, (sel_bit, input) in enumerate(inputs):
+            m.d.comb += fw_net.select[i].eq(sel_bit)
+            m.d.comb += Value.cast(fw_net.inputs[i]).eq(input)
+        return fw_net.output
+
+    def elaborate(self, platform):
+        m = Module()
+
+        for i in OneHotSwitchDynamic(m, self.select, default=True):
+            if i is None:
+                m.d.comb += Value.cast(self.output).eq(self.default_input)
+            else:
+                m.d.comb += Value.cast(self.output).eq(self.inputs[i])
+
+        return m