Merge branch 'master' into lekcyjna/transactron-multi-method-request

coreblocks/fu/div_unit.py

@@ -1,6 +1,6 @@
 from dataclasses import KW_ONLY, dataclass
 from enum import IntFlag, auto
-from typing import Sequence, Tuple
+from collections.abc import Sequence
 
 from amaranth import *
 
@@ -34,7 +34,7 @@ def get_instructions(self) -> Sequence[tuple]:
         ]
 
 
-def get_input(arg: Record) -> Tuple[Value, Value]:
+def get_input(arg: Record) -> tuple[Value, Value]:
     return arg.s1_val, Mux(arg.imm, arg.imm, arg.s2_val)
 
 

coreblocks/fu/mul_unit.py

@@ -1,5 +1,5 @@
 from enum import IntFlag, IntEnum, auto
-from typing import Sequence, Tuple
+from collections.abc import Sequence
 from dataclasses import KW_ONLY, dataclass
 
 from amaranth import *
@@ -45,7 +45,7 @@ def get_instructions(self) -> Sequence[tuple]:
         ]
 
 
-def get_input(arg: Record) -> Tuple[Value, Value]:
+def get_input(arg: Record) -> tuple[Value, Value]:
     """
     Operation of getting two input values.
 
@@ -56,7 +56,7 @@ def get_input(arg: Record) -> Tuple[Value, Value]:
 
     Returns
     -------
-    return : Tuple[Value, Value]
+    return : tuple[Value, Value]
         Two input values.
     """
     return arg.s1_val, Mux(arg.imm, arg.imm, arg.s2_val)

coreblocks/scheduler/scheduler.py

@@ -1,4 +1,4 @@
-from typing import Sequence
+from collections.abc import Sequence
 
 from amaranth import *
 

coreblocks/stages/func_blocks_unifier.py

@@ -1,4 +1,4 @@
-from typing import Iterable
+from collections.abc import Iterable
 
 from amaranth import *
 

coreblocks/structs_common/rs.py

@@ -1,4 +1,5 @@
-from typing import Iterable, Optional
+from collections.abc import Iterable
+from typing import Optional
 from amaranth import *
 from amaranth.lib.coding import PriorityEncoder
 from transactron import Method, def_method, TModule

coreblocks/utils/_typing.py

@@ -1,18 +1,14 @@
 from typing import (
-    ContextManager,
     Generic,
     NoReturn,
     Optional,
     Protocol,
-    Sequence,
-    Tuple,
-    Type,
     TypeAlias,
-    Iterable,
-    Mapping,
     TypeVar,
     runtime_checkable,
 )
+from collections.abc import Iterable, Mapping, Sequence
+from contextlib import AbstractContextManager
 from enum import Enum
 from amaranth import *
 from amaranth.lib.data import View
@@ -21,16 +17,18 @@
 from amaranth.hdl.rec import Direction, Layout
 
 # Types representing Amaranth concepts
-FragmentLike = Fragment | Elaboratable
-ValueLike = Value | int | Enum | ValueCastable
-ShapeLike = Shape | ShapeCastable | int | range | Type[Enum]
+FragmentLike: TypeAlias = Fragment | Elaboratable
+ValueLike: TypeAlias = Value | int | Enum | ValueCastable
+ShapeLike: TypeAlias = Shape | ShapeCastable | int | range | type[Enum]
 StatementLike: TypeAlias = Statement | Iterable["StatementLike"]
-LayoutLike = Layout | Sequence[Tuple[str, ShapeLike | "LayoutLike"] | Tuple[str, ShapeLike | "LayoutLike", Direction]]
+LayoutLike: TypeAlias = (
+    Layout | Sequence[tuple[str, "ShapeLike | LayoutLike"] | tuple[str, "ShapeLike | LayoutLike", Direction]]
+)
 SwitchKey: TypeAlias = str | int | Enum
 
 # Internal Coreblocks types
 SignalBundle: TypeAlias = Signal | Record | View | Iterable["SignalBundle"] | Mapping[str, "SignalBundle"]
-LayoutList = list[Tuple[str, ShapeLike | "LayoutList"]]
+LayoutList: TypeAlias = list[tuple[str, "ShapeLike | LayoutList"]]
 
 
 class _ModuleBuilderDomainsLike(Protocol):
@@ -55,28 +53,30 @@ class ModuleLike(Protocol, Generic[_T_ModuleBuilderDomains]):
     domains: _ModuleBuilderDomainSet
     d: _T_ModuleBuilderDomains
 
-    def If(self, cond: ValueLike) -> ContextManager[None]:  # noqa: N802
+    def If(self, cond: ValueLike) -> AbstractContextManager[None]:  # noqa: N802
         ...
 
-    def Elif(self, cond: ValueLike) -> ContextManager[None]:  # noqa: N802
+    def Elif(self, cond: ValueLike) -> AbstractContextManager[None]:  # noqa: N802
         ...
 
-    def Else(self) -> ContextManager[None]:  # noqa: N802
+    def Else(self) -> AbstractContextManager[None]:  # noqa: N802
         ...
 
-    def Switch(self, test: ValueLike) -> ContextManager[None]:  # noqa: N802
+    def Switch(self, test: ValueLike) -> AbstractContextManager[None]:  # noqa: N802
         ...
 
-    def Case(self, *patterns: SwitchKey) -> ContextManager[None]:  # noqa: N802
+    def Case(self, *patterns: SwitchKey) -> AbstractContextManager[None]:  # noqa: N802
         ...
 
-    def Default(self) -> ContextManager[None]:  # noqa: N802
+    def Default(self) -> AbstractContextManager[None]:  # noqa: N802
         ...
 
-    def FSM(self, reset: Optional[str] = ..., domain: str = ..., name: str = ...) -> ContextManager[FSM]:  # noqa: N802
+    def FSM(  # noqa: N802
+        self, reset: Optional[str] = ..., domain: str = ..., name: str = ...
+    ) -> AbstractContextManager[FSM]:
         ...
 
-    def State(self, name: str) -> ContextManager[None]:  # noqa: N802
+    def State(self, name: str) -> AbstractContextManager[None]:  # noqa: N802
         ...
 
     @property

coreblocks/utils/utils.py

@@ -1,6 +1,7 @@
 from contextlib import contextmanager
 from enum import Enum
-from typing import Iterable, Literal, Mapping, Optional, TypeAlias, cast, overload
+from typing import Literal, Optional, TypeAlias, cast, overload
+from collections.abc import Iterable, Mapping
 from amaranth import *
 from amaranth.hdl.ast import Assign, ArrayProxy
 from amaranth.lib import data

requirements.txt

@@ -1,2 +1,2 @@
-amaranth-yosys==0.10.0.dev46
+amaranth-yosys==0.25.0.0.post77
 git+https://github.com/amaranth-lang/amaranth@ccf7aaf00db54c7647b2f0f0cfdf34835c16fa8f

test/regression/cocotb.py

@@ -147,7 +147,7 @@ async def run(self, mem_model: CoreMemoryModel, timeout_cycles: int = 5000) -> b
         instr_wb = WishboneSlave(self.dut, "wb_instr", self.dut.clk, mem_model, is_instr_bus=True)
         cocotb.start_soon(instr_wb.start())
 
-        data_wb = WishboneSlave(self.dut, "wb_data", self.dut.clk, mem_model, is_instr_bus=True)
+        data_wb = WishboneSlave(self.dut, "wb_data", self.dut.clk, mem_model, is_instr_bus=False)
         cocotb.start_soon(data_wb.start())
 
         res = await with_timeout(self.finish_event.wait(), timeout_cycles, "ns")

test/regression/memory.py

@@ -137,7 +137,7 @@ def write(self, req: WriteRequest) -> WriteReply:
             return WriteReply(status=ReplyStatus.ERROR)
 
 
-def load_segments_from_elf(file_path: str) -> list[RandomAccessMemory]:
+def load_segments_from_elf(file_path: str, *, disable_write_protection: bool = False) -> list[RandomAccessMemory]:
     segments: list[RandomAccessMemory] = []
 
     with open(file_path, "rb") as f:
@@ -160,11 +160,11 @@ def align_down(n: int) -> int:
             data = b"\x00" * (paddr - seg_start) + segment.data() + b"\x00" * (seg_end - (paddr + len(segment.data())))
 
             flags = SegmentFlags(0)
-            if flags_raw & P_FLAGS.PF_R == flags_raw & P_FLAGS.PF_R:
+            if flags_raw & P_FLAGS.PF_R:
                 flags |= SegmentFlags.READ
-            if flags_raw & P_FLAGS.PF_W == flags_raw & P_FLAGS.PF_W:
+            if flags_raw & P_FLAGS.PF_W or disable_write_protection:
                 flags |= SegmentFlags.WRITE
-            if flags_raw & P_FLAGS.PF_X == flags_raw & P_FLAGS.PF_X:
+            if flags_raw & P_FLAGS.PF_X:
                 flags |= SegmentFlags.EXECUTABLE
 
             segments.append(RandomAccessMemory(range(seg_start, seg_end), flags, data))

test/regression/test.py

@@ -7,6 +7,9 @@
 test_dir = Path(__file__).parent.parent
 riscv_tests_dir = test_dir.joinpath("external/riscv-tests")
 
+# disable write protection for specific tests with writes to .text section
+exclude_write_protection = ["rv32uc-rvc"]
+
 
 class MMIO(MemorySegment):
     def __init__(self, on_finish: Callable[[], None]):
@@ -31,7 +34,10 @@ async def run_test(sim_backend: SimulationBackend, test_name: str):
     mmio = MMIO(lambda: sim_backend.stop())
 
     mem_segments: list[MemorySegment] = []
-    mem_segments += load_segments_from_elf(str(riscv_tests_dir.joinpath("test-" + test_name)))
+    mem_segments += load_segments_from_elf(
+        str(riscv_tests_dir.joinpath("test-" + test_name)),
+        disable_write_protection=test_name in exclude_write_protection,
+    )
     mem_segments.append(mmio)
 
     mem_model = CoreMemoryModel(mem_segments)

transactron/_utils.py

@@ -1,7 +1,8 @@
 import itertools
 import sys
 from inspect import Parameter, signature
-from typing import Callable, Iterable, Optional, TypeAlias, TypeVar, Mapping
+from typing import Optional, TypeAlias, TypeVar
+from collections.abc import Callable, Iterable, Mapping
 from amaranth import *
 from coreblocks.utils._typing import LayoutLike
 from coreblocks.utils import OneHotSwitchDynamic

transactron/core.py

@@ -2,7 +2,18 @@
 from collections.abc import Sequence, Iterable, Callable, Mapping, Iterator
 from contextlib import contextmanager
 from enum import Enum, auto
-from typing import ClassVar, NoReturn, TypeAlias, TypedDict, Union, Optional, Tuple
+from typing import (
+    ClassVar,
+    NoReturn,
+    TypeAlias,
+    TypedDict,
+    Union,
+    Optional,
+    Tuple,
+    TypeVar,
+    Protocol,
+    runtime_checkable,
+)
 from graphlib import TopologicalSorter
 from typing_extensions import Self
 from amaranth import *
@@ -38,6 +49,7 @@
 TransactionScheduler: TypeAlias = Callable[["MethodMap", TransactionGraph, TransactionGraphCC, PriorityOrder], Module]
 RecordDict: TypeAlias = ValueLike | Mapping[str, "RecordDict"]
 TransactionOrMethod: TypeAlias = Union["Transaction", "Method"]
+TransactionOrMethodBound = TypeVar("TransactionOrMethodBound", "Transaction", "Method")
 
 
 class Priority(Enum):
@@ -675,15 +687,20 @@ def elaborate(self, platform):
         return self.main_module
 
 
-class TransactionBase(Owned):
+@runtime_checkable
+class TransactionBase(Owned, Protocol):
     stack: ClassVar[list[Union["Transaction", "Method"]]] = []
     def_counter: ClassVar[count] = count()
     def_order: int
     defined: bool = False
     name: str
+    method_uses: dict["Method", Tuple[Record, ValueLike]]
+    relations: list[RelationBase]
+    simultaneous_list: list[TransactionOrMethod]
+    independent_list: list[TransactionOrMethod]
 
     def __init__(self):
-        self.method_uses: dict[Method, Tuple[Record, ValueLike]] = dict()
+        self.method_uses: dict["Method", Tuple[Record, ValueLike]] = dict()
         self.relations: list[RelationBase] = []
         self.simultaneous_list: list[TransactionOrMethod] = []
         self.independent_list: list[TransactionOrMethod] = []
@@ -774,9 +791,7 @@ def _independent(self, *others: TransactionOrMethod) -> None:
         self.independent_list += others
 
     @contextmanager
-    def context(self, m: TModule) -> Iterator[Self]:
-        assert isinstance(self, Transaction) or isinstance(self, Method)  # for typing
-
+    def context(self: TransactionOrMethodBound, m: TModule) -> Iterator[TransactionOrMethodBound]:
         parent = TransactionBase.peek()
         if parent is not None:
             parent.schedule_before(self)

transactron/graph.py

@@ -3,15 +3,14 @@
 """
 
 from enum import IntFlag
-from abc import ABC
 from collections import defaultdict
-from typing import Literal, Optional
+from typing import Literal, Optional, Protocol
 
 from amaranth.hdl.ir import Elaboratable, Fragment
 from .tracing import TracingFragment
 
 
-class Owned(ABC):
+class Owned(Protocol):
     name: str
     owner: Optional[Elaboratable]
 

transactron/lib/transformers.py

@@ -1,7 +1,8 @@
 from amaranth import *
 from ..core import *
 from ..core import RecordDict
-from typing import Optional, Callable, Tuple
+from typing import Optional
+from collections.abc import Callable
 from coreblocks.utils import ValueLike, assign, AssignType
 from .connectors import Forwarder, ManyToOneConnectTrans, ConnectTrans
 
@@ -35,8 +36,8 @@ def __init__(
         self,
         target: Method,
         *,
-        i_transform: Optional[Tuple[MethodLayout, Callable[[TModule, Record], RecordDict]]] = None,
-        o_transform: Optional[Tuple[MethodLayout, Callable[[TModule, Record], RecordDict]]] = None,
+        i_transform: Optional[tuple[MethodLayout, Callable[[TModule, Record], RecordDict]]] = None,
+        o_transform: Optional[tuple[MethodLayout, Callable[[TModule, Record], RecordDict]]] = None,
     ):
         """
         Parameters
@@ -132,7 +133,7 @@ class MethodProduct(Elaboratable):
     def __init__(
         self,
         targets: list[Method],
-        combiner: Optional[Tuple[MethodLayout, Callable[[TModule, list[Record]], RecordDict]]] = None,
+        combiner: Optional[tuple[MethodLayout, Callable[[TModule, list[Record]], RecordDict]]] = None,
     ):
         """Method product.