transactron.core package

Submodules

transactron.core.keys module

class transactron.core.keys.TransactionManagerKey

Bases: SimpleKey[TransactionManager]

__init__() → None

transactron.core.manager module

class transactron.core.manager.TransactionComponent

Bases: TransactionModule, Component

Top-level component for Transactron projects.

The TransactronComponent is a wrapper on Component classes, which adds Transactron support for the wrapped class. The use case is to wrap a top-level module of the project, and pass the wrapped module for simulation, HDL generation or synthesis. The ports of the wrapped component are forwarded to the wrapper.

It extends the functionality of TransactionModule.

__init__(component: AbstractComponent, dependency_manager: Optional[DependencyManager] = None, transaction_manager: Optional[TransactionManager] = None)

Parameters

component: Component: The Component which should be wrapped to add support for transactions and methods.
dependency_manager: DependencyManager, optional: The DependencyManager to use inside the transaction component. If omitted, a new one is created.
transaction_manager: TransactionManager, optional: The TransactionManager to use inside the transaction component. If omitted, a new one is created.

class transactron.core.manager.TransactionManager

Bases: Elaboratable

Transaction manager

This module is responsible for granting Transactions and running Methods. It takes care that two conflicting Transactions are never granted in the same clock cycle.

__init__(cc_scheduler: ~typing.Callable[[~transactron.core.manager.MethodMap, dict[transactron.core.transaction.Transaction, set[transactron.core.transaction.Transaction]], set[transactron.core.transaction.Transaction], dict[transactron.core.transaction.Transaction, int]], ~amaranth.hdl._dsl.Module] = <function eager_deterministic_cc_scheduler>)

add_transaction(transaction: Transaction)

debug_signals() → amaranth.hdl._ast.Signal | amaranth.hdl._rec.Record | amaranth.lib.data.View | collections.abc.Iterable[amaranth.hdl._ast.Signal | amaranth.hdl._rec.Record | amaranth.lib.data.View | collections.abc.Iterable[SignalBundle] | collections.abc.Mapping[str, SignalBundle]] | collections.abc.Mapping[str, amaranth.hdl._ast.Signal | amaranth.hdl._rec.Record | amaranth.lib.data.View | collections.abc.Iterable[SignalBundle] | collections.abc.Mapping[str, SignalBundle]]

print_info(cgr: dict[transactron.core.transaction.Transaction, set[transactron.core.transaction.Transaction]], porder: dict[transactron.core.transaction.Transaction, int], ccs: list[set[transactron.core.transaction.Transaction]], method_map: MethodMap)

visual_graph(fragment)

class transactron.core.manager.TransactionModule

Bases: Elaboratable

TransactionModule is used as wrapper on Elaboratable classes, which adds support for transactions. It creates a TransactionManager which will handle transaction scheduling and can be used in definition of Methods and Transactions. The TransactionManager is stored in a DependencyManager.

__init__(elaboratable: HasElaborate, dependency_manager: Optional[DependencyManager] = None, transaction_manager: Optional[TransactionManager] = None)

Parameters

elaboratable: HasElaborate: The Elaboratable which should be wrapped to add support for transactions and methods.
dependency_manager: DependencyManager, optional: The DependencyManager to use inside the transaction module. If omitted, a new one is created.
transaction_manager: TransactionManager, optional: The TransactionManager to use inside the transaction module. If omitted, a new one is created.

context() → DependencyContext

transactron.core.method module

class transactron.core.method.Method

Bases: TransactionBase

Transactional method.

A Method serves to interface a module with external Transactions or Methods. It can be called by at most once in a given clock cycle. When a given Method is required by multiple Transactions (either directly, or indirectly via another Method) simultenaously, at most one of them is granted by the TransactionManager, and the rest of them must wait. (Non-exclusive methods are an exception to this behavior.) Calling a Method always takes a single clock cycle.

Data is combinationally transferred between to and from Methods using Amaranth structures (View with a StructLayout). The transfer can take place in both directions at the same time: from the called Method to the caller (data_out) and from the caller to the called Method (data_in).

A module which defines a Method should use body or def_method to describe the method’s effect on the module state.

Attributes

name: str: Name of this Method.
ready: Signal, in: Signals that the method is ready to run in the current cycle. Typically defined by calling body.
run: Signal, out: Signals that the method is called in the current cycle by some Transaction. Defined by the TransactionManager.
data_in: MethodStruct, out: Contains the data passed to the Method by the caller (a Transaction or another Method).
data_out: MethodStruct, in: Contains the data passed from the Method to the caller (a Transaction or another Method). Typically defined by calling body.

__init__(*, name: Optional[str] = None, i: amaranth.lib.data.StructLayout | collections.abc.Iterable[tuple[str, 'ShapeLike | LayoutList']] = (), o: amaranth.lib.data.StructLayout | collections.abc.Iterable[tuple[str, 'ShapeLike | LayoutList']] = (), nonexclusive: bool = False, combiner: Optional[Callable[[Module, Sequence[View[StructLayout]], Value], amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable | collections.abc.Mapping[str, AssignArg] | collections.abc.Mapping[int, AssignArg] | collections.abc.Sequence[AssignArg]]] = None, single_caller: bool = False, src_loc: int | tuple[str, int] = 0)

Parameters

name: str or None: Name hint for this Method. If None (default) the name is inferred from the variable name this Method is assigned to.
i: method layout: The format of data_in.
o: method layout: The format of data_out.
nonexclusive: bool: If true, the method is non-exclusive: it can be called by multiple transactions in the same clock cycle. If such a situation happens, the method still is executed only once, and each of the callers receive its output. Nonexclusive methods cannot have inputs.
combiner: (Module, Sequence[MethodStruct], Value) -> AssignArg: If nonexclusive is true, the combiner function combines the arguments from multiple calls to this method into a single argument, which is passed to the method body. The third argument is a bit vector, whose n-th bit is 1 if the n-th call is active in a given cycle.
single_caller: bool: If true, this method is intended to be called from a single transaction. An error will be thrown if called from multiple transactions.
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

Define method body

The body context manager can be used to define the actions performed by a Method when it’s run. Each assignment added to a domain under body is guarded by the run signal. Combinational assignments which do not need to be guarded by run can be added to m.d.av_comb or m.d.top_comb instead of m.d.comb. Method calls can be performed under body.

Parameters

mTModule: Module in which operations on signals should be executed, body uses the combinational domain only.
readySignal, in: Signal to indicate if the method is ready to be run. By default it is Const(1), so the method is always ready. Assigned combinationially to the ready attribute.
outValue, in: Data generated by the Method, which will be passed to the caller (a Transaction or another Method). Assigned combinationally to the data_out attribute.
validate_arguments: Optional[Callable[…, ValueLike]]: Function that takes input arguments used to call the method and checks whether the method can be called with those arguments. It instantiates a combinational circuit for each method caller. By default, there is no function, so all arguments are accepted.

Returns

data_inRecord, out: Data passed from the caller (a Transaction or another Method) to this Method.

Examples

m = Module()
my_sum_method = Method(i = Layout([("arg1",8),("arg2",8)]))
sum = Signal(16)
with my_sum_method.body(m, out = sum) as data_in:
    m.d.comb += sum.eq(data_in.arg1 + data_in.arg2)

debug_signals() → amaranth.hdl._ast.Signal | amaranth.hdl._rec.Record | amaranth.lib.data.View | collections.abc.Iterable[amaranth.hdl._ast.Signal | amaranth.hdl._rec.Record | amaranth.lib.data.View | collections.abc.Iterable[SignalBundle] | collections.abc.Mapping[str, SignalBundle]] | collections.abc.Mapping[str, amaranth.hdl._ast.Signal | amaranth.hdl._rec.Record | amaranth.lib.data.View | collections.abc.Iterable[SignalBundle] | collections.abc.Mapping[str, SignalBundle]]

property layout_in

property layout_out

static like(other: Method, *, name: Optional[str] = None, src_loc: int | tuple[str, int] = 0) → Method

Constructs a new Method based on another.

The returned Method has the same input/output data layouts as the other Method.

Parameters

otherMethod: The Method which serves as a blueprint for the new Method.
namestr, optional: Name of the new Method.
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

Returns

Method: The freshly constructed Method.

proxy(m: TModule, method: Method)

Define as a proxy for another method.

The calls to this method will be forwarded to method.

Parameters

mTModule: Module in which operations on signals should be executed, proxy uses the combinational domain only.
methodMethod: Method for which this method is a proxy for.

transactron.core.schedulers module

transactron.core.schedulers.eager_deterministic_cc_scheduler(method_map: MethodMap, gr: TransactionGraph, cc: TransactionGraphCC, porder: PriorityOrder) → Module

This function generates an eager scheduler for the transaction subsystem. It isn’t fair, because it starts transactions using transaction index in cc as a priority. Transaction with the lowest index has the highest priority.

If there are two different transactions which have no conflicts then they will be started concurrently.

Parameters

managerTransactionManager: TransactionManager which uses this instance of scheduler for arbitrating which agent should get a grant signal.
grTransactionGraph: Graph of conflicts between transactions, where vertices are transactions and edges are conflicts.
ccSet[Transaction]: Connected components of the graph gr for which scheduler should be generated.
porderPriorityOrder: Linear ordering of transactions which is consistent with priority constraints.

transactron.core.schedulers.trivial_roundrobin_cc_scheduler(method_map: MethodMap, gr: TransactionGraph, cc: TransactionGraphCC, porder: PriorityOrder) → Module

This function generates a simple round-robin scheduler for the transaction subsystem. In a one cycle there will be at most one transaction granted (in a given connected component of the conflict graph), even if there is another ready, non-conflicting, transaction. It is mainly for testing purposes.

Parameters

managerTransactionManager: TransactionManager which uses this instance of scheduler for arbitrating which agent should get grant signal.
grTransactionGraph: Graph of conflicts between transactions, where vertices are transactions and edges are conflicts.
ccSet[Transaction]: Connected components of the graph gr for which scheduler should be generated.
porderPriorityOrder: Linear ordering of transactions which is consistent with priority constraints.

transactron.core.sugar module

transactron.core.sugar.def_method(m: TModule, method: Method, ready: amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable = (const 1'd1), validate_arguments: ~typing.Optional[~collections.abc.Callable[[...], amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable]] = None)

Define a method.

This decorator allows to define transactional methods in an elegant way using Python’s def syntax. Internally, def_method uses Method.body.

The decorated function should take keyword arguments corresponding to the fields of the method’s input layout. The **kwargs syntax is supported. Alternatively, it can take one argument named arg, which will be a structure with input signals.

The returned value can be either a structure with the method’s output layout or a dictionary of outputs.

Parameters

m: TModule: Module in which operations on signals should be executed.
method: Method: The method whose body is going to be defined.
ready: Signal: Signal to indicate if the method is ready to be run. By default it is Const(1), so the method is always ready. Assigned combinationally to the ready attribute.
validate_arguments: Optional[Callable[…, ValueLike]]: Function that takes input arguments used to call the method and checks whether the method can be called with those arguments. It instantiates a combinational circuit for each method caller. By default, there is no function, so all arguments are accepted.

Examples

m = Module()
my_sum_method = Method(i=[("arg1",8),("arg2",8)], o=[("res",8)])
@def_method(m, my_sum_method)
def _(arg1, arg2):
    return arg1 + arg2

Alternative syntax (keyword args in dictionary):

@def_method(m, my_sum_method)
def _(**args):
    return args["arg1"] + args["arg2"]

Alternative syntax (arg structure):

@def_method(m, my_sum_method)
def _(arg):
    return {"res": arg.arg1 + arg.arg2}

transactron.core.sugar.def_methods(m: TModule, methods: ~collections.abc.Sequence[Method], ready: ~collections.abc.Callable[[int], amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable] = <function <lambda>>, validate_arguments: ~typing.Optional[~collections.abc.Callable[[...], amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable]] = None)

Decorator for defining similar methods

This decorator is a wrapper over def_method, which allows you to easily define multiple similar methods in a loop.

The function over which this decorator is applied, should always expect at least one argument, as the index of the method will be passed as the first argument to the function.

This is a syntax sugar equivalent to:

for i in range(len(my_methods)):
    @def_method(m, my_methods[i])
    def _(arg):
        ...

Parameters

m: TModule: Module in which operations on signals should be executed.
methods: Sequence[Method]: The methods whose body is going to be defined.
ready: Callable[[int], Value]: A Callable that takes the index in the form of an int of the currently defined method and produces a Value describing whether the method is ready to be run. When omitted, each defined method is always ready. Assigned combinationally to the ready attribute.
validate_arguments: Optional[Callable[Concatenate[int, …], ValueLike]]: Function that takes input arguments used to call the method and checks whether the method can be called with those arguments. It instantiates a combinational circuit for each method caller. By default, there is no function, so all arguments are accepted.

Examples

Define three methods with the same body:

m = TModule()
my_sum_methods = [Method(i=[("arg1",8),("arg2",8)], o=[("res",8)]) for _ in range(3)]
@def_methods(m, my_sum_methods)
def _(_, arg1, arg2):
    return arg1 + arg2

Define three methods with different bodies parametrized with the index of the method:

m = TModule()
my_sum_methods = [Method(i=[("arg1",8),("arg2",8)], o=[("res",8)]) for _ in range(3)]
@def_methods(m, my_sum_methods)
def _(index : int, arg1, arg2):
    return arg1 + arg2 + index

Define three methods with different ready signals:

@def_methods(m, my_filter_read_methods, ready_list=lambda i: fifo.head == i)
def _(_):
    return fifo.read(m)

transactron.core.tmodule module

class transactron.core.tmodule.TModule

Bases: ModuleLike, Elaboratable

Extended Amaranth module for use with transactions.

It includes three different combinational domains:

comb domain, works like the comb domain in plain Amaranth modules. Statements in comb are guarded by every condition, including AvoidedIf. This means they are guarded by transaction and method bodies: they don’t execute if the given transaction/method is not run.
av_comb is guarded by all conditions except AvoidedIf. This means they are not guarded by transaction and method bodies. This allows to reduce the amount of useless multplexers due to transaction use, while still allowing the use of conditions in transaction/method bodies.
top_comb is unguarded: statements added to this domain always execute. It can be used to reduce combinational path length due to multplexers while keeping related combinational and synchronous statements together.

AvoidedIf(cond: amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable)

Case(*patterns: str | int | enum.Enum)

Default()

Elif(cond)

Else()

FSM(init: Optional[str] = None, domain: str = 'sync', name: str = 'fsm')

If(cond: amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable)

State(name: str)

Switch(test: amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable)

__init__()

property ctrl_path

property next: NoReturn

transactron.core.transaction module

class transactron.core.transaction.Transaction

Bases: TransactionBase

Transaction.

A Transaction represents a task which needs to be regularly done. Execution of a Transaction always lasts a single clock cycle. A Transaction signals readiness for execution by setting the request signal. If the conditions for its execution are met, it can be granted by the TransactionManager.

A Transaction can, as part of its execution, call a number of Methods. A Transaction can be granted only if every Method it runs is ready.

A Transaction cannot execute concurrently with another, conflicting Transaction. Conflicts between Transactions are either explicit or implicit. An explicit conflict is added using the add_conflict method. Implicit conflicts arise between pairs of Transactions which use the same Method.

A module which defines a Transaction should use body to describe used methods and the transaction’s effect on the module state. The used methods should be called inside the body’s with block.

Attributes

name: str: Name of this Transaction.
request: Signal, in: Signals that the transaction wants to run. If omitted, the transaction is always ready. Defined in the constructor.
runnable: Signal, out: Signals that all used methods are ready.
grant: Signal, out: Signals that the transaction is granted by the TransactionManager, and all used methods are called.

__init__(*, name: Optional[str] = None, manager: Optional[TransactionManager] = None, src_loc: int | tuple[str, int] = 0)

Parameters

name: str or None: Name hint for this Transaction. If None (default) the name is inferred from the variable name this Transaction is assigned to. If the Transaction was not assigned, the name is inferred from the class name where the Transaction was constructed.
manager: TransactionManager: The TransactionManager controlling this Transaction. If omitted, the manager is received from TransactionContext.
src_loc: int | SrcLoc: How many stack frames deep the source location is taken from. Alternatively, the source location to use instead of the default.

body(m: TModule, *, request: amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable = (const 1'd1)) → Iterator[Transaction]

Defines the Transaction body.

This context manager allows to conveniently define the actions performed by a Transaction when it’s granted. Each assignment added to a domain under body is guarded by the grant signal. Combinational assignments which do not need to be guarded by grant can be added to m.d.top_comb or m.d.av_comb instead of m.d.comb. Method calls can be performed under body.

Parameters

m: TModule: The module where the Transaction is defined.
request: Signal: Indicates that the Transaction wants to be executed. By default it is Const(1), so it wants to be executed in every clock cycle.

debug_signals() → amaranth.hdl._ast.Signal | amaranth.hdl._rec.Record | amaranth.lib.data.View | collections.abc.Iterable[amaranth.hdl._ast.Signal | amaranth.hdl._rec.Record | amaranth.lib.data.View | collections.abc.Iterable[SignalBundle] | collections.abc.Mapping[str, SignalBundle]] | collections.abc.Mapping[str, amaranth.hdl._ast.Signal | amaranth.hdl._rec.Record | amaranth.lib.data.View | collections.abc.Iterable[SignalBundle] | collections.abc.Mapping[str, SignalBundle]]

transactron.core.transaction_base module

class transactron.core.transaction_base.Priority

Bases: Enum

LEFT = 2: Left transaction/method is prioritized over the right one.

RIGHT = 3: Right transaction/method is prioritized over the left one.

UNDEFINED = 1: Conflicting transactions/methods don’t have a priority order.

class transactron.core.transaction_base.TransactionBase

Bases: Owned, Protocol

__init__(*, src_loc: int | tuple[str, int])

add_conflict(end: Union[Transaction, Method], priority: Priority = Priority.UNDEFINED) → None

Registers a conflict.

Record that that the given Transaction or Method cannot execute simultaneously with this Method or Transaction. Typical reason is using a common resource (register write or memory port).

Parameters

end: Transaction or Method: The conflicting Transaction or Method
priority: Priority, optional: Is one of conflicting Transactions or Methods prioritized? Defaults to undefined priority relation.

context(m: TModule) → Iterator[TransactionOrMethodBound]

ctrl_path: CtrlPath = CtrlPath(module=-1, path=[])

def_counter: ClassVar[count] = count(0)

def_order: int

defined: bool = False

classmethod get() → Self

independent_list: list[typing.Union[ForwardRef('Transaction'), ForwardRef('Method')]]

method_calls: defaultdict[Method, list[tuple[transactron.core.tmodule.CtrlPath, 'View[StructLayout]', amaranth.hdl._ast.Value | int | enum.Enum | amaranth.hdl._ast.ValueCastable]]]

method_uses: dict['Method', tuple['View[StructLayout]', amaranth.hdl._ast.Signal]]

name: str

property owned_name

classmethod peek() → Optional[Self]

relations: list[transactron.core.transaction_base.RelationBase]

schedule_before(end: Union[Transaction, Method]) → None

Adds a priority relation.

Record that that the given Transaction or Method needs to be scheduled before this Method or Transaction, without adding a conflict. Typical reason is data forwarding.

Parameters

end: Transaction or Method: The other Transaction or Method

simultaneous(*others: Union[Transaction, Method]) → None

Adds simultaneity relations.

The given Transactions or Method`s will execute simultaneously (in the same clock cycle) with this Transaction or Method.

Parameters

*others: Transaction or Method: The Transactions or Methods to be executed simultaneously.

simultaneous_alternatives(*others: Union[Transaction, Method]) → None

Adds exclusive simultaneity relations.

Each of the given Transactions or Method`s will execute simultaneously (in the same clock cycle) with this Transaction or Method. However, each of the given Transactions or Methods will be separately considered for execution.

Parameters

*others: Transaction or Method: The Transactions or Methods to be executed simultaneously, but mutually exclusive, with this Transaction or Method.

simultaneous_list: list[typing.Union[ForwardRef('Transaction'), ForwardRef('Method')]]

src_loc: tuple[str, int]

stack: Union[ForwardRef('Transaction'), ForwardRef('Method')]]] = []

transactron.core package

Submodules

transactron.core.keys module

transactron.core.manager module

transactron.core.method module

transactron.core.schedulers module

transactron.core.sugar module

transactron.core.tmodule module

transactron.core.transaction module

transactron.core.transaction_base module

Module contents