IRON API (aie.iron)

IRON is the high-level Python interface for programming AMD Ryzen™ AI NPUs. You describe a design as Python objects — tiles, workers, data movement, and a host runtime — and IRON compiles it to an optimized xclbin + instruction stream via the MLIR-AIE toolchain.

Every object on this page is resolvable: it lowers to one or more MLIR operations when the design is compiled. That is what makes it the high-level layer. For the direct MLIR op wrappers that IRON lowers to, see the Dialect op wrappers page.

All symbols documented here are importable directly from the iron namespace (e.g. from iron import Worker, ObjectFifo, Runtime).

---

Core design abstractions

The objects most designs are built from.

Program

Program

Program(device: Device, rt: Runtime)

A Program represents all design information needed to run the design on a device.

Note

MLIR verification (ctx.module.operation.verify()) is performed inside resolve_program, not during construction.

Parameters:

Name	Type	Description	Default
device	Device	The device used to generate the final MLIR for the design.	required
rt	Runtime	The runtime object for the design.	required

Source code in python/iron/program.py

def __init__(
    self,
    device: Device,
    rt: Runtime,
):
    """A Program represents all design information needed to run the design on a device.

    !!! note
        MLIR verification (`ctx.module.operation.verify()`) is performed inside
        [`resolve_program`][iron.program.Program.resolve_program], not during construction.

    Args:
        device (Device): The device used to generate the final MLIR for the design.
        rt (Runtime): The runtime object for the design.
    """
    self._device = device
    self._rt = rt

resolve_program

resolve_program(device_name='main')

This method resolves the program components in order to generate MLIR.

Tiles are emitted as aie.logical_tile ops. The --aie-place-tiles pass in the compilation pipeline converts them to aie.tile ops.

Returns:

Name	Type	Description
module	Module	The module containing the MLIR context information.

Source code in python/iron/program.py

def resolve_program(self, device_name="main"):
    """This method resolves the program components in order to generate MLIR.

    Tiles are emitted as aie.logical_tile ops. The --aie-place-tiles pass
    in the compilation pipeline converts them to aie.tile ops.

    Returns:
        module (Module): The module containing the MLIR context information.
    """
    with mlir_mod_ctx() as ctx:
        # Create a fresh device instance of the same type to avoid stale MLIR operations
        # This preserves the device configuration while ensuring clean state
        device_type = type(self._device)
        # For dynamically created device classes, the constructor takes no arguments
        self._device = device_type()  # pyright: ignore[reportCallIssue]

        # Resolve parameters at module scope (before the aie.device).
        # aiex.scratchpad_parameter ops are global across all devices because the
        # scratchpad is a single hardware resource shared by all PDIs.
        for w in self._rt.workers:
            for arg in w.fn_args:
                if isinstance(arg, ScratchpadParameter):
                    arg.resolve()
        for p in self._rt._scratchpad_parameters:
            p.resolve()

        @device(self._device.resolve(), sym_name=device_name)
        def device_body():
            # Collect all fifos
            all_fifos = set()
            all_fifos.update(self._rt.fifos)
            for w in self._rt.workers:
                all_fifos.update(w.fifos)

            # Sort fifos for deterministic resolve
            all_fifos = sorted(all_fifos, key=lambda obj: obj.name)

            # Collect all tiles. Two workers landing on the same compute
            # tile (pinned or after placement) is caught by the aie.device
            # verifier's one-core-per-tile check, so no Python-side guard.
            all_tiles = []
            for w in self._rt.workers:
                all_tiles.append(w.tile)
                # Generic: any user-side Resolvable in fn_args may declare
                # additional tile dependencies via tiles(). Default is [].
                for arg in w.fn_args:
                    if isinstance(arg, Resolvable):
                        all_tiles.extend(arg.tiles())
            for f in all_fifos:
                all_tiles.extend([e.tile for e in f.all_of_endpoints()])
                # Shared-memory delegate tile (ObjectFifo.delegate_tile kwarg)
                # may not appear in any prod/cons endpoint, so pick it up
                # explicitly so resolve_tile() runs on it before fifo resolution.
                if f._object_fifo._delegate_tile is not None:
                    all_tiles.append(f._object_fifo._delegate_tile)
            # Lower-level: explicit Flow / TileDma / Lock primitives
            # contribute tiles too.
            for fl in self._rt.flows:
                all_tiles.extend(fl.all_tiles())
            for td in self._rt.tile_dmas:
                all_tiles.extend(td.all_tiles())
            for lk in self._rt.locks:
                all_tiles.append(lk.tile)

            # Resolve tiles
            for t in all_tiles:
                self._device.resolve_tile(t)

            # Generate fifos
            for f in all_fifos:
                f.resolve()

            # Generate explicit Flows (peers of ObjectFifo)
            for fl in self._rt.flows:
                fl.resolve()

            # Generate explicit Locks (must come before TileDma + Worker
            # bodies that reference them; Buffers attached to worker
            # fn_args are still resolved in the worker loop below).
            for lk in self._rt.locks:
                lk.resolve()

            # Resolve any Buffers referenced by explicit TileDma programs
            # (those aren't reached via worker.fn_args).
            for td in self._rt.tile_dmas:
                bufs, _ = td.all_buffers_and_locks()
                for b in bufs:
                    if b.tile is None:
                        b._tile = td.tile
                    b.resolve()

            # generate functions - this may call resolve() more than once on the same fifo, but that's ok
            for w in self._rt.workers:
                for arg in w.fn_args:
                    if isinstance(arg, FuncBase):
                        arg.emit()
                    elif isinstance(arg, Resolvable):
                        arg.resolve()

            # Generate core programs
            for w in self._rt.workers:
                w.resolve()

            # Emit aie.cascade_flow ops for each Worker's outgoing edges.
            # Must run after worker.resolve() so both tiles are placed.
            for w in self._rt.workers:
                for cf in w._outgoing_cascades:
                    cf.resolve()

            # Generate explicit per-tile DMA programs (lower-level peers
            # of ObjectFifo, paired with Flow + Lock).
            for td in self._rt.tile_dmas:
                td.resolve()

            # Generate trace routes
            # TODO Need to iterate over all tiles or workers & fifos to make list of tiles to trace
            #      Alternatively, we merge the mechanism for packet routed objfifos so we use unique
            #      route IDs for trace as well

            # Scan workers and build list of tiles to trace
            tiles_to_trace = []
            if self._rt._trace_workers is not None:
                for w in self._rt._trace_workers:
                    tiles_to_trace.append(w.tile.op)
            else:
                for w in self._rt._workers:
                    if w.trace is not None:
                        tiles_to_trace.append(w.tile.op)
            if self._rt._trace_size is not None and self._rt._trace_size > 0:
                trace_utils.configure_trace(
                    tiles_to_trace,
                    coretile_events=self._rt._coretile_events,
                    coremem_events=self._rt._coremem_events,
                    memtile_events=self._rt._memtile_events,
                    shimtile_events=self._rt._shimtile_events,
                )

            # In/Out Sequence
            self._rt.resolve()

        self._print_verify(ctx)
        return ctx.module

Worker

Worker and WorkerRuntimeBarrier: compute-core tasks and runtime synchronization primitives.

Worker

Worker(
    core_fn: Callable | None,
    fn_args: list | None = None,
    tile: Tile = AnyComputeTile,
    while_true: bool = True,
    stack_size: int | None = None,
    allocation_scheme: str | None = None,
    trace: int | None = None,
    trace_events: list | None = None,
    dynamic_objfifo_lowering: bool | None = None,
)

Bases: ObjectFifoEndpoint

A task to be run on an AIE compute core.

A Worker takes a core_fn callable and the arguments it needs (ObjectFIFO handles, Buffers, Kernels, etc.). Each Worker is placed on a single compute tile, either explicitly via tile or automatically by the --aie-place-tiles compiler pass.

Construct a Worker

Parameters:

Name	Type	Description	Default
core_fn	Callable | None	The task to run on a core. If None, a busy-loop (while(true): pass) core will be generated.	required
fn_args	list | None	Pointers to arguments, which should include all context the core_fn needs to run. Defaults to None (empty list).	None
tile	Tile	The compute tile for the Worker. Also accepts None (treated as AnyComputeTile). Defaults to AnyComputeTile.	AnyComputeTile
while_true	bool	If true, will wrap the core_fn in a while(true) loop to ensure it runs until reconfiguration. Defaults to True.	True
stack_size	int	The stack_size in bytes to be allocated for the worker. Defaults to 1024 bytes.	None
allocation_scheme	str	The memory allocation scheme to use for the Worker, either 'basic-sequential' or 'bank-aware'. If None, defaults to bank-aware. Will override any allocation scheme set on the tile.	None
trace	int	If >0, enable tracing for this worker.	None
trace_events	list | None	Custom list of trace events for this worker. Defaults to None.	None
dynamic_objfifo_lowering	bool | None	Per-core override for the aie-objectFifo-stateful-transform pass's lowering choice. True forces dynamic (loop-preserving) lowering for this core; False forces static LCM-based unrolling. None (default) leaves the choice to the compiler's global --dynamic-objFifos flag. Note: the per-core attribute is only honored when the global flag is false; when global is true the attribute is ignored. Defaults to None.	None

Raises:

Type	Description
ValueError	Parameters are validated.

Source code in python/iron/worker.py

def __init__(
    self,
    core_fn: Callable | None,
    fn_args: list | None = None,
    tile: Tile = AnyComputeTile,
    while_true: bool = True,
    stack_size: int | None = None,
    allocation_scheme: str | None = None,
    trace: int | None = None,
    trace_events: list | None = None,
    dynamic_objfifo_lowering: bool | None = None,
):
    """Construct a Worker

    Args:
        core_fn (Callable | None): The task to run on a core. If None, a busy-loop (`while(true): pass`) core will be generated.
        fn_args (list | None, optional): Pointers to arguments, which should include all context the core_fn needs to run. Defaults to None (empty list).
        tile (Tile, optional): The compute tile for the Worker. Also accepts None (treated as AnyComputeTile). Defaults to AnyComputeTile.
        while_true (bool, optional): If true, will wrap the core_fn in a while(true) loop to ensure it runs until reconfiguration. Defaults to True.
        stack_size (int, optional): The stack_size in bytes to be allocated for the worker. Defaults to 1024 bytes.
        allocation_scheme (str, optional): The memory allocation scheme to use for the Worker, either 'basic-sequential' or 'bank-aware'. If None, defaults to bank-aware.
            Will override any allocation scheme set on the tile.
        trace (int, optional): If >0, enable tracing for this worker.
        trace_events (list | None, optional): Custom list of trace events for this worker. Defaults to None.
        dynamic_objfifo_lowering (bool | None, optional): Per-core override for the
            ``aie-objectFifo-stateful-transform`` pass's lowering choice. ``True`` forces
            dynamic (loop-preserving) lowering for this core; ``False`` forces static
            LCM-based unrolling. ``None`` (default) leaves the choice to the compiler's
            global ``--dynamic-objFifos`` flag. Note: the per-core attribute is only
            honored when the global flag is ``false``; when global is ``true`` the
            attribute is ignored. Defaults to None.

    Raises:
        ValueError: Parameters are validated.
    """
    if tile is None:
        tile = AnyComputeTile
    if tile.tile_type is not None and tile.tile_type != AIETileType.CoreTile:
        raise ValueError(
            f"Worker requires a compute tile, but got tile_type={tile.tile_type}"
        )
    # Store the user's Tile directly when it is already typed as CoreTile
    # and no allocation_scheme override is needed. This preserves Python
    # object identity so a Buffer and a Worker that share the same Tile
    # object resolve to a single LogicalTileOp. When we need a fresh copy
    # (untyped tile, singleton default, or allocation_scheme override) use
    # with_type() — it always returns a new object.
    if (
        tile.tile_type == AIETileType.CoreTile
        and allocation_scheme is None
        and tile is not AnyComputeTile
    ):
        self._tile = tile
    else:
        self._tile = tile.with_type(
            AIETileType.CoreTile, allocation_scheme=allocation_scheme
        )
    self._while_true = while_true
    self.stack_size = stack_size
    self.allocation_scheme = allocation_scheme
    self._dynamic_objfifo_lowering = dynamic_objfifo_lowering
    self.trace = trace
    self.trace_events = trace_events

    # If no core_fn is given, make a simple while(true) loop.
    if core_fn is None:

        def do_nothing_core_fun(*args) -> None:
            for _ in range_(sys.maxsize):
                pass

        self.core_fn = do_nothing_core_fun
    else:
        self.core_fn = core_fn
    self.fn_args = fn_args if fn_args is not None else []
    self._fifos = []
    self._buffers = []
    self._barriers = []
    # CascadeFlow objects whose source is this Worker. Populated by
    # CascadeFlow(src, dst).__init__ and consumed by Program.resolve()
    # to emit aie.cascade_flow ops after worker placement.
    self._outgoing_cascades: list = []

    # Check arguments to the core. Some information is saved for resolution.
    for arg in self.fn_args:
        if isinstance(arg, ObjectFifoHandle):
            arg.endpoint = self
            self._fifos.append(arg)
        elif isinstance(arg, Buffer):
            # A Buffer pinned to an EXPLICIT tile may legitimately be shared
            # across Workers: AIE compute tiles can read a neighbor tile's L1
            # directly, so a producer core's output buffer can be an input to a
            # consumer core on an adjacent tile. In that case the FIRST worker that
            # references it "owns"/places it and later workers are non-owning
            # readers. We only forbid sharing for AUTO-PLACED buffers (no explicit
            # tile), where two owners would race to pin it to different tiles.
            # Note: ``_tile`` alone is not a reliable signal — the owning Worker
            # auto-pins ``_tile`` to its own tile below — so we key off
            # ``_explicit_tile``, which records the user's construction-time intent.
            if arg._owner_worker is not None and arg._owner_worker is not self:
                if not arg._explicit_tile:
                    raise ValueError(
                        f"Buffer '{arg._name}' has no explicit tile and is shared "
                        f"across Workers; pin it to a tile (Buffer(tile=...)) so "
                        f"placement is unambiguous."
                    )
                # shared reader: keep original owner, just record the reference.
                self._buffers.append(arg)
            else:
                arg._owner_worker = self
                self._buffers.append(arg)
                # If the Buffer has no tile, pin it to the Worker's tile as a
                # convenience.  If the user pinned it explicitly to a neighbor
                # tile (AIE compute tiles can read N/S/E/W neighbors' L1
                # directly), honor that placement — Program.resolve discovers
                # the neighbor tile via Buffer.tiles().
                if arg._tile is None:
                    arg._tile = self._tile
        elif isinstance(arg, ScratchpadParameter):
            pass  # ScratchpadParameters are device-level symbols; no tile placement needed
        elif isinstance(arg, ObjectFifo):
            # This is an easy error to make, so we catch it early
            raise ValueError(
                "Cannot give an ObjectFifo directly to a worker; "
                "must give an ObjectFifoHandle obtained through "
                "ObjectFifo.prod() or ObjectFifo.cons()"
            )
        elif isinstance(arg, WorkerRuntimeBarrier):
            self._barriers.append(arg)

tile property

tile: Tile

The compute tile this Worker is placed on.

fifos property

fifos: list[ObjectFifoHandle]

Returns a list of ObjectFifoHandles given to the Worker via fn_args.

Returns:

Type	Description
list[ObjectFifoHandle]	list[ObjectFifoHandle]: ObjectFifoHandles used by the Worker.

buffers property

buffers: list[Buffer]

Returns a list of Buffers given to the Worker via fn_args.

Returns:

Type	Description
list[Buffer]	list[Buffer]: Buffer used by the Worker.

grid staticmethod

grid(
    rows: int,
    cols: int,
    factory: Callable[[int, int], Worker],
) -> list[list[Worker]]

Build a 2D grid of Workers; factory(r, c) returns one Worker.

Replaces the common pattern::

ws = [Worker(...) for i in range(R) for j in range(C)]
ws[i * C + j]  # 1-D index arithmetic

with::

ws = Worker.grid(R, C, lambda r, c: Worker(...))
ws[i][j]       # natural 2-D access

Parameters:

Name	Type	Description	Default
rows	int	Outer-dimension count (e.g. column index).	required
cols	int	Inner-dimension count (e.g. channel index).	required
factory	Callable[[int, int], Worker]	Called once per cell with (r, c); must return a Worker.	required

Returns:

Type	Description
list[list[Worker]]	rows-by-cols nested list of Worker instances.

Source code in python/iron/worker.py

@staticmethod
def grid(
    rows: int,
    cols: int,
    factory: Callable[[int, int], "Worker"],
) -> list[list["Worker"]]:
    """Build a 2D grid of Workers; ``factory(r, c)`` returns one Worker.

    Replaces the common pattern::

        ws = [Worker(...) for i in range(R) for j in range(C)]
        ws[i * C + j]  # 1-D index arithmetic

    with::

        ws = Worker.grid(R, C, lambda r, c: Worker(...))
        ws[i][j]       # natural 2-D access

    Args:
        rows: Outer-dimension count (e.g. column index).
        cols: Inner-dimension count (e.g. channel index).
        factory: Called once per cell with ``(r, c)``; must return a Worker.

    Returns:
        ``rows``-by-``cols`` nested list of Worker instances.
    """
    return [[factory(r, c) for c in range(cols)] for r in range(rows)]

WorkerRuntimeBarrier

WorkerRuntimeBarrier(initial_value: int = 0)

A barrier allowing individual workers to synchronize with the runtime sequence.

Initialize a WorkerRuntimeBarrier.

Parameters:

Name	Type	Description	Default
initial_value	int	The initial lock value. Defaults to 0.	0

Source code in python/iron/worker.py

def __init__(self, initial_value: int = 0):
    """Initialize a WorkerRuntimeBarrier.

    Args:
        initial_value (int, optional): The initial lock value. Defaults to 0.
    """
    self.initial_value = initial_value
    self.worker_locks = []

wait_for_value

wait_for_value(value: int)

Should be called from inside a core function. Wait for the barrier to be set to value.

Parameters:

Name	Type	Description	Default
value	int	The value to wait for.	required

Source code in python/iron/worker.py

def wait_for_value(self, value: int):
    """
    Should be called from inside a core function.
    Wait for the barrier to be set to `value`.

    Args:
        value (int): The value to wait for.
    """
    # Here this is assuming that the we are currently placing the last added lock
    # And therefore that wait_for_value operations are placed just after their corresponding Worker...
    # This is a pretty bad assumption, think about an alternative way to solve this
    if len(self.worker_locks) == 0:
        raise ValueError(
            "No workers have been registered for this barrier. Need to pass the barrier as an argument to the worker."
        )
    use_lock(self.worker_locks[-1], LockAction.Acquire, value=value)

release_with_value

release_with_value(value: int)

Release and decrement the barrier by value inside the core.

Parameters:

Name	Type	Description	Default
value	int	The value to decrement by in Release.	required

Source code in python/iron/worker.py

def release_with_value(self, value: int):
    """
    Release and decrement the barrier by `value` inside the core.

    Args:
        value (int): The value to decrement by in Release.
    """
    if len(self.worker_locks) == 0:
        raise ValueError(
            "No workers have been registered for this barrier. Need to pass the barrier as an argument to the worker."
        )
    use_lock(self.worker_locks[-1], LockAction.Release, value=value)

ObjectFifo

ObjectFifo

ObjectFifo(
    obj_type: type[ndarray],
    *,
    depth: int | None = 2,
    name: str | None = None,
    dims_to_stream: StreamDims | None = None,
    dims_from_stream_per_cons: StreamDims | None = None,
    plio: bool = False,
    pad_dimensions: PadDims | None = None,
    disable_synchronization: bool = False,
    repeat_count: int | None = None,
    delegate_tile: Tile | None = None,
    via_DMA: bool = False,
    init_values: list[ndarray] | None = None,
    consumer_obj_type: type[ndarray] | None = None,
    aie_stream: tuple[int, int] | None = None
)

Bases: Resolvable

A synchronized, explicit dataflow channel between IRON program components such as Workers and the Runtime.

Internally, an ObjectFifo is a circular buffer with a given depth and element type. Its users are explicitly either a producer or a consumer, and each user holds an ObjectFifoHandle carrying its (possibly unplaced) tile.

Example

of = ObjectFifo(np.ndarray[(1024,), np.dtype[np.int32]], name="in")
producer = of.prod()   # one producer handle
consumer = of.cons()   # one or more consumer handles

Construct an ObjectFifo.

Parameters:

Name	Type	Description	Default
obj_type	type[ndarray]	The type of each buffer in the ObjectFifo	required
depth	int | None	The default depth of the ObjectFifo endpoints. Defaults to 2.	2
name	str | None	The name of the ObjectFifo. If None is given, a unique name will be generated. Defaults to None.	None
dims_to_stream	StreamDims | None	Data layout transformations applied when data is pushed onto the AXI stream, described as pairs of (size, stride) from highest to lowest dimension. Defaults to None.	None
dims_from_stream_per_cons	StreamDims | None	List of data layout transformations applied by each consumer when data is read from the AXI stream, described as pairs of (size, stride) from highest to lowest dimension. Defaults to None.	None
plio	bool	Whether the ObjectFifo uses PLIO connections. Defaults to False.	False
disable_synchronization	bool	When True, disables lock-based synchronization on the ObjectFifo. Defaults to False.	False
repeat_count	int | None	If set, causes the MemTile DMA to replay the buffer descriptor this many times without a new DMA transfer from L3. Distinct from iter_count (BD-chain iteration count). Defaults to None.	None
delegate_tile	Tile | None	Shared-memory delegate tile. When set, the ObjectFifo's underlying buffer pool is allocated on this tile's memory module instead of the default placement. Lowers to aie.objectfifo.allocate. Only valid when both producer and consumer have shared-memory access to the delegate tile (e.g. self-loop fifos where prod == cons, or fifos between adjacent tiles spilling to a neighboring MemTile). The delegate is the storage location, not a producer- or consumer-side concept; the underlying op verifier rejects this if either endpoint cannot share memory with the delegate. Defaults to None.	None
init_values	list[ndarray] | None	Per-buffer static initial values for the producer endpoint. One ndarray per producer-side buffer; the producer tile must be able to hold static data at design startup (e.g. a MemTile). Lowers to the initValues attribute on the underlying aie.objectfifo op. Defaults to None.	None
consumer_obj_type	type[ndarray] | None	Consumer element type for asymmetric transfer granularity. When set, the producer sends obj_type-sized transfers and the consumer receives consumer_obj_type-sized transfers. Producer element count must be an integer multiple of consumer element count. Defaults to None.	None
aie_stream	tuple[int, int] | None	Mark the fifo as a direct AIE-stream connection by stamping the aie_stream / aie_stream_port attributes (end, port) on the underlying aie.objectfifo op. Use with kernels that emit on the wire via put_ms() instead of going through an L1 buffer. Defaults to None.	None

Raises:

Type	Description
ValueError	If depth is provided and is less than 1.

Source code in python/iron/dataflow/objectfifo.py

def __init__(
    self,
    obj_type: type[np.ndarray],
    *,
    depth: int | None = 2,
    name: str | None = None,
    dims_to_stream: StreamDims | None = None,
    dims_from_stream_per_cons: StreamDims | None = None,
    plio: bool = False,
    pad_dimensions: PadDims | None = None,
    disable_synchronization: bool = False,
    repeat_count: int | None = None,
    delegate_tile: Tile | None = None,
    via_DMA: bool = False,
    init_values: list[np.ndarray] | None = None,
    consumer_obj_type: type[np.ndarray] | None = None,
    aie_stream: tuple[int, int] | None = None,
):
    """Construct an ObjectFifo.

    Args:
        obj_type (type[np.ndarray]): The type of each buffer in the ObjectFifo
        depth (int | None, optional): The default depth of the ObjectFifo endpoints. Defaults to 2.
        name (str | None, optional): The name of the ObjectFifo. If None is given, a unique name will be generated. Defaults to None.
        dims_to_stream (StreamDims | None, optional): Data layout transformations applied when data is pushed onto the AXI stream, described as pairs of (size, stride) from highest to lowest dimension. Defaults to None.
        dims_from_stream_per_cons (StreamDims | None, optional): List of data layout transformations applied by each consumer when data is read from the AXI stream, described as pairs of (size, stride) from highest to lowest dimension. Defaults to None.
        plio (bool, optional): Whether the ObjectFifo uses PLIO connections. Defaults to False.
        disable_synchronization (bool, optional): When True, disables lock-based synchronization on the ObjectFifo. Defaults to False.
        repeat_count (int | None, optional): If set, causes the MemTile DMA to replay the buffer descriptor this many times without a new DMA transfer from L3. Distinct from ``iter_count`` (BD-chain iteration count). Defaults to None.
        delegate_tile (Tile | None, optional): Shared-memory delegate tile. When set, the ObjectFifo's underlying buffer pool is allocated on this tile's memory module instead of the default placement. Lowers to ``aie.objectfifo.allocate``. *Only valid when both producer and consumer have shared-memory access to the delegate tile* (e.g. self-loop fifos where prod == cons, or fifos between adjacent tiles spilling to a neighboring MemTile). The delegate is the storage location, not a producer- or consumer-side concept; the underlying op verifier rejects this if either endpoint cannot share memory with the delegate. Defaults to None.
        init_values (list[np.ndarray] | None, optional): Per-buffer static initial values for the producer endpoint. One ndarray per producer-side buffer; the producer tile must be able to hold static data at design startup (e.g. a MemTile). Lowers to the ``initValues`` attribute on the underlying ``aie.objectfifo`` op. Defaults to None.
        consumer_obj_type (type[np.ndarray] | None, optional): Consumer element type for asymmetric transfer granularity. When set, the producer sends obj_type-sized transfers and the consumer receives consumer_obj_type-sized transfers. Producer element count must be an integer multiple of consumer element count. Defaults to None.
        aie_stream (tuple[int, int] | None, optional): Mark the fifo as a direct AIE-stream connection by stamping the ``aie_stream`` / ``aie_stream_port`` attributes ``(end, port)`` on the underlying ``aie.objectfifo`` op. Use with kernels that emit on the wire via ``put_ms()`` instead of going through an L1 buffer. Defaults to None.

    Raises:
        ValueError: If ``depth`` is provided and is less than 1.
    """
    self._depth = depth
    if self._depth is not None and self._depth < 1:
        raise ValueError(
            f"Default ObjectFifo depth must be > 0, but got {self._depth}"
        )
    self._obj_type = obj_type
    self._dims_to_stream = dims_to_stream
    self._dims_from_stream_per_cons = dims_from_stream_per_cons
    self._plio = plio
    self._pad_dimensions = pad_dimensions
    if name is None:
        self.name = f"of{next(ObjectFifo._of_index)}"
    else:
        self.name = name
    self._op: ObjectFifoCreateOp | None = None
    self._prod: ObjectFifoHandle | None = None
    self._cons: list[ObjectFifoHandle] = []
    self._resolving = False
    self._iter_count: int | None = None
    self._repeat_count: int | None = repeat_count
    self._disable_synchronization: bool = disable_synchronization
    # Delegate tile for shared-memory buffer placement (lowers to aie.objectfifo.allocate).
    # Must be resolved before resolve() runs — Program.resolve() picks this up via
    # ObjectFifo._delegate_tile when collecting tiles to assign MLIR ops to.
    self._delegate_tile: Tile | None = delegate_tile
    self._via_DMA: bool = via_DMA
    self._init_values: list[np.ndarray] | None = init_values
    self._consumer_obj_type: type[np.ndarray] | None = consumer_obj_type
    self._aie_stream: tuple[int, int] | None = aie_stream

depth property

depth: int | None

The default depth of the ObjectFifo. This may be overridden by an ObjectFifoHandle upon construction.

dims_from_stream_per_cons property

dims_from_stream_per_cons: StreamDims | None

The default dimensions from stream per consumer value. This may be overridden by an ObjectFifoHandle of type consumer.

dims_to_stream property

dims_to_stream: StreamDims | None

The dimensions to stream value. This will be shared by the ObjectFifoHandle of type producer.

shape property

shape: Sequence[int]

The shape of each buffer belonging to the ObjectFifo

dtype property

dtype: NpuDType

The per-element data type of each element in each buffer belonging to the ObjectFifo

obj_type property

obj_type: type[ndarray]

The tensor type of each buffer belonging to the ObjectFifo

set_iter_count

set_iter_count(iter_count: int)

Set iteration count for DMA BD (Buffer Descriptor) chaining on MemTile for the ObjectFifo.

Parameters:

Name	Type	Description	Default
iter_count	int	Number of forward chain iterations. - Must be in range [1, 256]: Forward chain with specified number of iterations	required

Raises:

Type	Description
ValueError	If iter_count is outside the valid range [1, 256]

Source code in python/iron/dataflow/objectfifo.py

def set_iter_count(self, iter_count: int):
    """Set iteration count for DMA BD (Buffer Descriptor) chaining on MemTile for the ObjectFifo.

    Args:
        iter_count (int): Number of forward chain iterations.
            - Must be in range [1, 256]: Forward chain with specified number of iterations

    Raises:
        ValueError: If iter_count is outside the valid range [1, 256]
    """
    if not iter_count or iter_count < 1 or iter_count > 256:
        raise ValueError("Iter count must be in [1, 256] range.")

    self._iter_count = iter_count

prod

prod(
    depth: int | None = None, channel: int | None = None
) -> ObjectFifoHandle

Returns an ObjectFifoHandle of type producer. Each ObjectFifo may have only one producer handle, so if one already exists, a new reference to this handle will be returned.

Parameters:

Name	Type	Description	Default
depth	int | None	The depth of the buffers at the endpoint corresponding to the producer handle. Defaults to None.	None
channel	int | None	Pin the producer endpoint's DMA channel instead of first-free assignment. Defaults to None (auto-assign).	None

Raises:

Type	Description
ValueError	Arguments are validated
ValueError	If depth was not specified on ObjectFifo construction, depth must be specified here.

Returns:

Name	Type	Description
ObjectFifoHandle	ObjectFifoHandle	The producer handle to this ObjectFifo.

Source code in python/iron/dataflow/objectfifo.py

def prod(
    self, depth: int | None = None, channel: int | None = None
) -> ObjectFifoHandle:
    """Returns an ObjectFifoHandle of type producer. Each ObjectFifo may have only one producer
    handle, so if one already exists, a new reference to this handle will be returned.

    Args:
        depth (int | None, optional): The depth of the buffers at the endpoint corresponding to the producer handle. Defaults to None.
        channel (int | None, optional): Pin the producer endpoint's DMA channel instead of first-free assignment. Defaults to None (auto-assign).

    Raises:
        ValueError: Arguments are validated
        ValueError: If depth was not specified on ObjectFifo construction, depth must be specified here.

    Returns:
        ObjectFifoHandle: The producer handle to this ObjectFifo.
    """
    if self._prod:
        if depth is None:
            if self._depth is None:
                raise ValueError(f"If depth is None, then depth must be specified.")
            else:
                depth = self._depth
        elif depth < 1:
            raise ValueError(f"Depth must be > 1, but got {depth}")
        if channel is not None and self._prod.channel != channel:
            raise ValueError(
                f"Producer handle for {self.name} already pinned to channel "
                f"{self._prod.channel}, cannot re-pin to {channel}."
            )
    else:
        self._prod = ObjectFifoHandle(self, True, depth, channel=channel)
    return self._prod

cons

cons(
    depth: int | None = None,
    dims_from_stream: StreamDims | None = None,
    channel: int | None = None,
) -> ObjectFifoHandle

Returns an ObjectFifoHandle of type consumer. Each ObjectFifo may have multiple consumers, so this will return a new consumer handle every time it is called.

Parameters:

Name	Type	Description	Default
depth	int | None	The depth of the buffers at the endpoint corresponding to this consumer handle. Defaults to None.	None
dims_from_stream	StreamDims | None	Dimensions from stream for this consumer. Defaults to None.	None
channel	int | None	Pin this consumer endpoint's DMA channel instead of first-free assignment. Defaults to None (auto-assign).	None

Raises:

Type	Description
ValueError	Arguments are validated

Returns:

Name	Type	Description
ObjectFifoHandle	ObjectFifoHandle	A consumer handle to this ObjectFifo.

Source code in python/iron/dataflow/objectfifo.py

def cons(
    self,
    depth: int | None = None,
    dims_from_stream: StreamDims | None = None,
    channel: int | None = None,
) -> ObjectFifoHandle:
    """Returns an ObjectFifoHandle of type consumer. Each ObjectFifo may have multiple consumers, so this
    will return a new consumer handle every time it is called.

    Args:
        depth (int | None, optional): The depth of the buffers at the endpoint corresponding to this consumer handle. Defaults to None.
        dims_from_stream (StreamDims | None, optional): Dimensions from stream for this consumer. Defaults to None.
        channel (int | None, optional): Pin this consumer endpoint's DMA channel instead of first-free assignment. Defaults to None (auto-assign).

    Raises:
        ValueError: Arguments are validated

    Returns:
        ObjectFifoHandle: A consumer handle to this ObjectFifo.
    """
    if depth is None:
        if self._depth is None:
            raise ValueError(f"If depth is None, then depth must be specified.")
        else:
            depth = self._depth

    if dims_from_stream is None:
        dims_from_stream = self._dims_from_stream_per_cons
    self._cons.append(
        ObjectFifoHandle(
            self,
            is_prod=False,
            depth=depth,
            dims_from_stream=dims_from_stream,
            channel=channel,
        )
    )
    return self._cons[-1]

tiles

tiles(cons_only: bool = False) -> list[Tile]

The list of placement tiles corresponding to the endpoints of all handles of this ObjectFifo

Raises:

Type	Description
ValueError	A producer handle must be constructed.
ValueError	At least one consumer handle must be constructed.

Returns:

Type	Description
list[Tile]	list[Tile]: A list of tiles of the endpoints of this ObjectFifo.

Source code in python/iron/dataflow/objectfifo.py

def tiles(self, cons_only: bool = False) -> list[Tile]:
    """The list of placement tiles corresponding to the endpoints of all handles of this ObjectFifo

    Raises:
        ValueError: A producer handle must be constructed.
        ValueError: At least one consumer handle must be constructed.

    Returns:
        list[Tile]: A list of tiles of the endpoints of this ObjectFifo.
    """
    tiles = []
    if not cons_only:
        if self._prod == None:
            raise ValueError(
                "Cannot return prod.tile.op because prod was not created."
            )
        if self._prod.endpoint is None:
            raise ValueError(f"Prod endpoint not set for {self}")
        assert self._prod.endpoint.tile is not None
        tiles += [self._prod.endpoint.tile]
    if self._cons == []:
        raise ValueError("Cannot return cons tiles because cons were not created.")
    for cons in self._cons:
        if cons.endpoint is None:
            raise ValueError(f"Cons endpoint not set for {self}")
        assert cons.endpoint.tile is not None
        tiles.append(cons.endpoint.tile)
    return tiles

ObjectFifoHandle

ObjectFifoHandle(
    of: ObjectFifo,
    is_prod: bool,
    depth: int | None = None,
    dims_from_stream: StreamDims | None = None,
    channel: int | None = None,
)

Bases: Resolvable

A handle to an ObjectFifo, of type producer or consumer.

Producer and consumer handles are what Worker core functions call acquire and release on to move data through the fifo. Obtain them via ObjectFifo.prod() and ObjectFifo.cons().

Construct an ObjectFifoHandle

Parameters:

Name	Type	Description	Default
of	ObjectFifo	The ObjectFifo to construct the handle for.	required
is_prod	bool	Whether the handle should be producer or consumer handle.	required
depth	int | None	The depth of the ObjectFifo at this endpoint. Defaults to None.	None
dims_from_stream	StreamDims | None	A unique dimensions from stream. This is only valid for consumer handles. Defaults to None.	None
channel	int | None	Pin this endpoint's DMA channel instead of first-free assignment. Defaults to None (auto-assign).	None

Raises:

Type	Description
ValueError	Arguments are validated.

Source code in python/iron/dataflow/objectfifo.py

def __init__(
    self,
    of: ObjectFifo,
    is_prod: bool,
    depth: int | None = None,
    dims_from_stream: StreamDims | None = None,
    channel: int | None = None,
):
    """Construct an ObjectFifoHandle

    Args:
        of (ObjectFifo): The ObjectFifo to construct the handle for.
        is_prod (bool): Whether the handle should be producer or consumer handle.
        depth (int | None, optional): The depth of the ObjectFifo at this endpoint. Defaults to None.
        dims_from_stream (StreamDims | None, optional): A unique dimensions from stream. This is only valid for consumer handles. Defaults to None.
        channel (int | None, optional): Pin this endpoint's DMA channel instead of first-free assignment. Defaults to None (auto-assign).

    Raises:
        ValueError: Arguments are validated.
    """
    if depth is None:
        if of.depth:
            depth = of.depth
        else:
            raise ValueError(
                "Must specify either ObjectFifoHandle depth or ObjectFifo default depth; both are None."
            )
    if depth < 1:
        raise ValueError(f"Depth must be > 0 but is {depth}")
    self._port: ObjectFifoPort = (
        ObjectFifoPort.Produce if is_prod else ObjectFifoPort.Consume
    )
    if is_prod and dims_from_stream:
        raise ValueError("Can only specify dims_from_stream for cons handles")
    elif not is_prod and not dims_from_stream:
        dims_from_stream = of.dims_from_stream_per_cons

    self._is_prod = is_prod
    self._object_fifo = of
    self._depth = depth
    self._channel = channel
    self._endpoint = None
    self._dims_from_stream = dims_from_stream

name property

name: str

The name of the ObjectFifo

channel property

channel: int | None

The pinned DMA channel for this handle's endpoint, or None to auto-assign.

obj_type property

obj_type: type[ndarray]

The per-buffer type of the ObjectFifo

shape property

shape: Sequence[int]

The per-buffer shape of the ObjectFifo

dtype property

dtype: NpuDType

The per-element datatype of the ObjectFifo

handle_type property

handle_type: str

A string referencing the type of this ObjectFifoHandle.

depth property

depth: int

The depth of this ObjectFifoHandle

dims_from_stream property

dims_from_stream: StreamDims | None

The dimensions from stream of a consumer ObjectFifoHandle

endpoint property writable

endpoint: ObjectFifoEndpoint | None

The endpoint of this ObjectFifoHandle

acquire

acquire(num_elem: int) -> list

Acquire access to some elements of the ObjectFifo, using ObjectFifo synchronization to moderate access.

Parameters:

Name	Type	Description	Default
num_elem	int	Number of elements to acquire. If some elements are already acquired, only the additional elements needed to reach a total of num_elem are acquired.	required

Raises:

Type	Description
ValueError	Number of elements cannot exceed ObjectFifo depth.

Returns:

Type	Description
list	An indexable handle to the acquired elements: a single element when num_elem == 1, or an indexable view when num_elem > 1.

Source code in python/iron/dataflow/objectfifo.py

def acquire(
    self,
    num_elem: int,
) -> list:
    """Acquire access to some elements of the ObjectFifo, using ObjectFifo synchronization to moderate access.

    Args:
        num_elem (int): Number of elements to acquire. If some elements are already acquired, only the additional elements needed to reach a total of ``num_elem`` are acquired.

    Raises:
        ValueError: Number of elements cannot exceed ObjectFifo depth.

    Returns:
        An indexable handle to the acquired elements: a single element when ``num_elem == 1``, or an indexable view when ``num_elem > 1``.
    """
    if self._depth < num_elem:
        raise ValueError(
            f"Number of elements to acquire {num_elem} must be smaller than depth {self._depth}"
        )
    return self._object_fifo._acquire(self._port, num_elem)

release

release(num_elem: int) -> None

Release access to some elements of the ObjectFifo, allowing the other endpoint of the ObjectFifo to acquire them.

Parameters:

Name	Type	Description	Default
num_elem	int	Number of elements to release.	required

Raises:

Type	Description
ValueError	Number of elements cannot exceed ObjectFifo depth.

Source code in python/iron/dataflow/objectfifo.py

def release(
    self,
    num_elem: int,
) -> None:
    """Release access to some elements of the ObjectFifo, allowing the other endpoint of the ObjectFifo to acquire them.

    Args:
        num_elem (int): Number of elements to release.

    Raises:
        ValueError: Number of elements cannot exceed ObjectFifo depth.
    """
    if self._depth < num_elem:
        raise ValueError(
            f"Number of elements to release {num_elem} must be smaller than depth {self._depth}"
        )
    self._object_fifo._release(self._port, num_elem)

all_of_endpoints

all_of_endpoints() -> list[ObjectFifoEndpoint]

All endpoints belonging to an ObjectFifo

Source code in python/iron/dataflow/objectfifo.py

def all_of_endpoints(self) -> list[ObjectFifoEndpoint]:
    """All endpoints belonging to an ObjectFifo"""
    return self._object_fifo._get_endpoint(
        is_prod=True
    ) + self._object_fifo._get_endpoint(is_prod=False)

join

join(
    offsets: list[int],
    tile: Tile = AnyMemTile,
    depths: list[int] | None = None,
    obj_types: list[type[ndarray]] | None = None,
    names: list[str] | None = None,
    dims_to_stream: list[StreamDims] | None = None,
    dims_from_stream: list[StreamDims] | None = None,
    plio: bool = False,
    repeat_counts: list[int | None] | None = None,
) -> list[ObjectFifo]

Construct multiple ObjectFifos which feed data into a ObjectFifoHandle. Note that this function is only valid for producer ObjectFifoHandles.

Parameters:

Name	Type	Description	Default
offsets	list[int]	Offsets into the current producer, each corresponding to a new consumer.	required
tile	Tile	The tile where the Join operation occurs. Defaults to AnyMemTile.	AnyMemTile
depths	list[int] | None	The depth of each new ObjectFifo. Defaults to None.	None
obj_types	list[type[ndarray]]	The type of the buffers corresponding to each new ObjectFifo. Defaults to None.	None
names	list[str] | None	The name of each new ObjectFifo. If not given, unique names will be generated. Defaults to None.	None
dims_to_stream	list[list[Sequence[int] | None]] | None	The dimensionsToStream to assign to each new ObjectFifo. Defaults to None.	None
dims_from_stream	list[list[Sequence[int] | None]] | None	The dimensionsFromStream to assign to each new ObjectFifo consumer. Defaults to None.	None
plio	bool	Set plio on each new ObjectFifo. Defaults to False.	False
repeat_counts	list[int | None] | None	Per-sub-fifo MemTile DMA repeat count (see ObjectFifo.repeat_count). Defaults to None.	None

Raises:

Type	Description
ValueError	Arguments are validated

Returns:

Type	Description
list[ObjectFifo]	list[ObjectFifo]: A list of newly constructed ObjectFifos whose consumers are used in this join() operation.

Source code in python/iron/dataflow/objectfifo.py

def join(
    self,
    offsets: list[int],
    tile: Tile = AnyMemTile,
    depths: list[int] | None = None,
    obj_types: list[type[np.ndarray]] | None = None,
    names: list[str] | None = None,
    dims_to_stream: list[StreamDims] | None = None,
    dims_from_stream: list[StreamDims] | None = None,
    plio: bool = False,
    repeat_counts: list[int | None] | None = None,
) -> list[ObjectFifo]:
    """Construct multiple ObjectFifos which feed data into a ObjectFifoHandle.
    Note that this function is only valid for producer ObjectFifoHandles.

    Args:
        offsets (list[int]): Offsets into the current producer, each corresponding to a new consumer.
        tile (Tile, optional): The tile where the Join operation occurs. Defaults to AnyMemTile.
        depths (list[int] | None, optional): The depth of each new ObjectFifo. Defaults to None.
        obj_types (list[type[np.ndarray]], optional): The type of the buffers corresponding to each new ObjectFifo. Defaults to None.
        names (list[str] | None, optional): The name of each new ObjectFifo. If not given, unique names will be generated. Defaults to None.
        dims_to_stream (list[list[Sequence[int]  |  None]] | None, optional): The dimensionsToStream to assign to each new ObjectFifo. Defaults to None.
        dims_from_stream (list[list[Sequence[int]  |  None]] | None, optional): The dimensionsFromStream to assign to each new ObjectFifo consumer. Defaults to None.
        plio (bool, optional): Set plio on each new ObjectFifo. Defaults to False.
        repeat_counts (list[int | None] | None, optional): Per-sub-fifo MemTile DMA repeat count (see ObjectFifo.repeat_count). Defaults to None.

    Raises:
        ValueError: Arguments are validated

    Returns:
        list[ObjectFifo]: A list of newly constructed ObjectFifos whose consumers are used in this join() operation.
    """
    if not self._is_prod:
        raise ValueError(f"Cannot join() a {self.handle_type} ObjectFifoHandle")
    num_subfifos = len(offsets)
    if depths is None:
        depths = [self.depth] * num_subfifos
    elif len(depths) != num_subfifos:
        raise ValueError("Number of depths does not match number of offsets")

    if obj_types is None:
        obj_types = [self._object_fifo.obj_type] * num_subfifos
    elif len(obj_types) != num_subfifos:
        raise ValueError("Number of obj_types does not match number of offsets")

    if names is None:
        names = [self._object_fifo.name + f"_join{i}" for i in range(num_subfifos)]
    elif len(names) != num_subfifos:
        raise ValueError("Number of names does not match number of offsets")

    if dims_to_stream is None:
        dims_to_stream = [[]] * num_subfifos
    elif len(dims_to_stream) != num_subfifos:
        raise ValueError(
            "Number of dims to stream does not match number of offsets"
        )

    if dims_from_stream is None:
        dims_from_stream = [[]] * num_subfifos
    elif dims_from_stream and len(dims_from_stream) != num_subfifos:
        raise ValueError(
            "Number of dims_from_stream does not match number of offsets"
        )

    if repeat_counts is None:
        repeat_counts = [None for _ in range(num_subfifos)]
    elif len(repeat_counts) != num_subfifos:
        raise ValueError("Number of repeat_counts does not match number of offsets")

    # Create subfifos
    subfifos = []
    for i in range(num_subfifos):
        subfifos.append(
            ObjectFifo(
                obj_types[i],
                name=names[i],
                depth=depths[i],
                dims_to_stream=dims_to_stream[i],
                plio=plio,
                repeat_count=repeat_counts[i],
            )
        )

    subfifo_cons = [
        s.cons(depth=depths[i], dims_from_stream=dims_from_stream[i])
        for i, s in enumerate(subfifos)
    ]
    _ = ObjectFifoLink(subfifo_cons, self, tile, offsets, [])
    return subfifos

split

split(
    offsets: list[int],
    tile: Tile = AnyMemTile,
    depths: list[int] | None = None,
    obj_types: list[type[ndarray]] | None = None,
    names: list[str] | None = None,
    dims_to_stream: list[StreamDims] | None = None,
    dims_from_stream: list[StreamDims] | None = None,
    plio: bool = False,
    repeat_counts: list[int | None] | None = None,
) -> list[ObjectFifo]

Split the data from an ObjectFifoConsumer handle by sending it to producers in N newly constructed ObjectFifos. Note this operation is only valid for ObjectFifoHandles of type consumer.

Parameters:

Name	Type	Description	Default
offsets	list[int]	The offset into the current consumer for each new ObjectFifo producer.	required
tile	Tile	The tile where the Split operation takes place. Defaults to AnyMemTile.	AnyMemTile
depths	list[int] | None	The depth of each new ObjectFifo. Defaults to None.	None
obj_types	list[type[ndarray]]	The buffer type of each new ObjectFifo. Defaults to None.	None
names	list[str] | None	The name of each new ObjectFifo. If not given, a unique name will be generated. Defaults to None.	None
dims_to_stream	list[StreamDims] | None	The dimensions to stream for each new ObjectFifo. Defaults to None.	None
dims_from_stream	list[StreamDims] | None	The dimensions from stream for each new ObjectFifo. Defaults to None.	None
plio	bool	Set plio on each new ObjectFifo. Defaults to False.	False
repeat_counts	list[int | None] | None	Per-sub-fifo MemTile DMA repeat count (see ObjectFifo.repeat_count). Defaults to None.	None

Raises:

Type	Description
ValueError	Arguments are validated.

Returns:

Type	Description
list[ObjectFifo]	list[ObjectFifo]: A list of newly constructed ObjectFifos whose producers are used in this split() operation.

Source code in python/iron/dataflow/objectfifo.py

def split(
    self,
    offsets: list[int],
    tile: Tile = AnyMemTile,
    depths: list[int] | None = None,
    obj_types: list[type[np.ndarray]] | None = None,
    names: list[str] | None = None,
    dims_to_stream: list[StreamDims] | None = None,
    dims_from_stream: list[StreamDims] | None = None,
    plio: bool = False,
    repeat_counts: list[int | None] | None = None,
) -> list[ObjectFifo]:
    """Split the data from an ObjectFifoConsumer handle by sending it to producers in N newly constructed ObjectFifos.
    Note this operation is only valid for ObjectFifoHandles of type consumer.

    Args:
        offsets (list[int]): The offset into the current consumer for each new ObjectFifo producer.
        tile (Tile, optional): The tile where the Split operation takes place. Defaults to AnyMemTile.
        depths (list[int] | None, optional): The depth of each new ObjectFifo. Defaults to None.
        obj_types (list[type[np.ndarray]], optional): The buffer type of each new ObjectFifo. Defaults to None.
        names (list[str] | None, optional): The name of each new ObjectFifo. If not given, a unique name will be generated. Defaults to None.
        dims_to_stream (list[StreamDims] | None, optional): The dimensions to stream for each new ObjectFifo. Defaults to None.
        dims_from_stream (list[StreamDims] | None, optional): The dimensions from stream for each new ObjectFifo. Defaults to None.
        plio (bool, optional): Set plio on each new ObjectFifo. Defaults to False.
        repeat_counts (list[int | None] | None, optional): Per-sub-fifo MemTile DMA repeat count (see ObjectFifo.repeat_count). Defaults to None.

    Raises:
        ValueError: Arguments are validated.

    Returns:
        list[ObjectFifo]: A list of newly constructed ObjectFifos whose producers are used in this split() operation.
    """
    if self._is_prod:
        raise ValueError(f"Cannot split() a {self.handle_type} ObjectFifoHandle")
    num_subfifos = len(offsets)
    if depths is None:
        depths = [self.depth] * num_subfifos
    elif len(depths) != num_subfifos:
        raise ValueError("Number of depths does not match number of offsets")

    if obj_types is None:
        obj_types = [self._object_fifo.obj_type] * num_subfifos
    elif len(obj_types) != num_subfifos:
        raise ValueError("Number of obj_types does not match number of offsets")

    if names is None:
        names = [self._object_fifo.name + f"_split{i}" for i in range(num_subfifos)]
    elif len(names) != num_subfifos:
        raise ValueError("Number of names does not match number of offsets")

    if dims_to_stream is None:
        dims_to_stream = [[]] * num_subfifos
    elif len(dims_to_stream) != num_subfifos:
        raise ValueError(
            "Number of dims_to_stream arrays does not match number of offsets"
        )

    if dims_from_stream is None:
        dims_from_stream = [[]] * num_subfifos
    elif len(dims_from_stream) != num_subfifos:
        raise ValueError(
            "Number of dims_from_stream arrays does not match number of offsets"
        )

    if repeat_counts is None:
        repeat_counts = [None for _ in range(num_subfifos)]
    elif len(repeat_counts) != num_subfifos:
        raise ValueError("Number of repeat_counts does not match number of offsets")

    # Create subfifos
    subfifos = []
    for i in range(num_subfifos):
        subfifos.append(
            ObjectFifo(
                obj_types[i],
                name=names[i],
                depth=depths[i],
                dims_to_stream=dims_to_stream[i],
                dims_from_stream_per_cons=dims_from_stream[i],
                plio=plio,
                repeat_count=repeat_counts[i],
            )
        )

    # Create link and set it as endpoints
    subfifo_prods = [s.prod() for s in subfifos]
    _ = ObjectFifoLink(self, subfifo_prods, tile, [], offsets)
    return subfifos

forward

forward(
    tile: Tile = AnyMemTile,
    obj_type: type[ndarray] | None = None,
    depth: int | None = None,
    name: str | None = None,
    dims_to_stream: StreamDims | None = None,
    dims_from_stream: StreamDims | None = None,
    plio: bool = False,
    repeat_count: int | None = None,
) -> ObjectFifo

This is a special case of the split() operation where an ObjectFifoHandle of type consumer is forwarded to the producer of a newly-constructed ObjectFifo.

Parameters:

Name	Type	Description	Default
tile	Tile	The tile for the Forward operation. Defaults to AnyMemTile.	AnyMemTile
obj_type	type[ndarray] | None	The object type of the new ObjectFifo. Defaults to None.	None
depth	int | None	The depth of the new ObjectFifo. Defaults to None.	None
name	str | None	The name of the new ObjectFifo. If None is given, a unique name will be generated. Defaults to None.	None
dims_to_stream	StreamDims | None	The dimensions to stream for the new ObjectFifo. Defaults to None.	None
dims_from_stream	StreamDims | None	The dimensions from stream for the new ObjectFifo. Defaults to None.	None
plio	bool	Set plio on each new ObjectFifo. Defaults to False.	False
repeat_count	int | None	MemTile DMA repeat count for the new ObjectFifo (see ObjectFifo.repeat_count). Defaults to None.	None

Raises:

Type	Description
ValueError	Arguments are Validated

Returns:

Name	Type	Description
ObjectFifo	ObjectFifo	A newly constructed ObjectFifo whose producer used in this forward() operation.

Source code in python/iron/dataflow/objectfifo.py

def forward(
    self,
    tile: Tile = AnyMemTile,
    obj_type: type[np.ndarray] | None = None,
    depth: int | None = None,
    name: str | None = None,
    dims_to_stream: StreamDims | None = None,
    dims_from_stream: StreamDims | None = None,
    plio: bool = False,
    repeat_count: int | None = None,
) -> ObjectFifo:
    """This is a special case of the split() operation where an ObjectFifoHandle of type consumer
    is forwarded to the producer of a newly-constructed ObjectFifo.

    Args:
        tile (Tile, optional): The tile for the Forward operation. Defaults to AnyMemTile.
        obj_type (type[np.ndarray] | None, optional): The object type of the new ObjectFifo. Defaults to None.
        depth (int | None, optional): The depth of the new ObjectFifo. Defaults to None.
        name (str | None, optional): The name of the new ObjectFifo. If None is given, a unique name will be generated. Defaults to None.
        dims_to_stream (StreamDims | None, optional): The dimensions to stream for the new ObjectFifo. Defaults to None.
        dims_from_stream (StreamDims | None, optional): The dimensions from stream for the new ObjectFifo. Defaults to None.
        plio (bool, optional): Set plio on each new ObjectFifo. Defaults to False.
        repeat_count (int | None, optional): MemTile DMA repeat count for the new ObjectFifo (see ObjectFifo.repeat_count). Defaults to None.

    Raises:
        ValueError: Arguments are Validated

    Returns:
        ObjectFifo: A newly constructed ObjectFifo whose producer used in this forward() operation.
    """
    if self._is_prod:
        raise ValueError(f"Cannot forward a {self.handle_type} ObjectFifoHandle")
    obj_types = [obj_type] if obj_type else None
    depths = [depth] if depth else None
    names = [name] if name else [self._object_fifo.name + "_fwd"]
    dims_to_stream_arg = [dims_to_stream] if dims_to_stream else None
    dims_from_stream_arg = [dims_from_stream] if dims_from_stream else None

    forward_fifo = self.split(
        [0],
        tile=tile,
        obj_types=obj_types,
        depths=depths,
        names=names,
        dims_to_stream=dims_to_stream_arg,
        dims_from_stream=dims_from_stream_arg,
        plio=plio,
        repeat_counts=[repeat_count] if repeat_count is not None else None,
    )
    return forward_fifo[0]

ObjectFifoLink

ObjectFifoLink(
    srcs: list[ObjectFifoHandle] | ObjectFifoHandle,
    dsts: list[ObjectFifoHandle] | ObjectFifoHandle,
    tile: Tile = AnyMemTile,
    src_offsets: list[int] | None = None,
    dst_offsets: list[int] | None = None,
)

Bases: ObjectFifoEndpoint, Resolvable

This is an object used internally by split(), join() and forward() operations.

Construct an ObjectFifoLink. This is either a many-to-one, one-to-many, or one-to-one operation.

Parameters:

Name	Type	Description	Default
srcs	list[ObjectFifoHandle] | ObjectFifoHandle	A list of consumer ObjectFifoHandles to link.	required
dsts	list[ObjectFifoHandle] | ObjectFifoHandle	A list of producer ObjectFifoHandles to link.	required
tile	Tile	The tile where the link occurs. Also accepts None (treated as AnyMemTile). Defaults to AnyMemTile.	AnyMemTile
src_offsets	list[int] | None	If many sources, one offset per source is required to split the destination. Defaults to None (empty list).	None
dst_offsets	list[int] | None	If many destinations, one offset per destination is required to split the source. Defaults to None (empty list).	None

Raises:

Type	Description
ValueError	Arguments are validated.

Source code in python/iron/dataflow/objectfifo.py

def __init__(
    self,
    srcs: list[ObjectFifoHandle] | ObjectFifoHandle,
    dsts: list[ObjectFifoHandle] | ObjectFifoHandle,
    tile: Tile = AnyMemTile,
    src_offsets: list[int] | None = None,
    dst_offsets: list[int] | None = None,
):
    """Construct an ObjectFifoLink. This is either a many-to-one, one-to-many, or one-to-one operation.

    Args:
        srcs (list[ObjectFifoHandle] | ObjectFifoHandle): A list of consumer ObjectFifoHandles to link.
        dsts (list[ObjectFifoHandle] | ObjectFifoHandle): A list of producer ObjectFifoHandles to link.
        tile (Tile, optional): The tile where the link occurs. Also accepts None (treated as AnyMemTile). Defaults to AnyMemTile.
        src_offsets (list[int] | None, optional): If many sources, one offset per source is required to split the destination. Defaults to None (empty list).
        dst_offsets (list[int] | None, optional): If many destinations, one offset per destination is required to split the source. Defaults to None (empty list).

    Raises:
        ValueError: Arguments are validated.
    """
    self._srcs = single_elem_or_list_to_list(srcs)
    self._dsts = single_elem_or_list_to_list(dsts)
    self._src_offsets = src_offsets if src_offsets is not None else []
    self._dst_offsets = dst_offsets if dst_offsets is not None else []
    self._resolving = False

    if len(self._srcs) < 1:
        raise ValueError("An ObjectFifoLink must have at least one source")
    if len(self._dsts) < 1:
        raise ValueError("An ObjectFifoLink must have at least one destination")
    if len(self._srcs) != 1 and len(self._dsts) != 1:
        raise ValueError(
            "An ObjectFifoLink may only have > 1 of either sources or destinations, but not both"
        )
    if len(self._src_offsets) > 0 and len(self._src_offsets) != len(self._srcs):
        raise ValueError(
            "The number of source offsets does not match the number of sources"
        )
    if len(self._dst_offsets) > 0 and len(self._dst_offsets) != len(self._dsts):
        raise ValueError(
            "The number of destination offsets does not match the number of destinations"
        )
    self._op = None
    for s in self._srcs:
        s.endpoint = self
    for d in self._dsts:
        d.endpoint = self
    if tile is None:
        tile = AnyMemTile
    # A link normally lives on a mem tile, but forward() documents
    # forwarding through a compute tile as a valid override, so preserve
    # an explicitly-set tile_type and only default when unset.
    default_type = (
        tile.tile_type if tile.tile_type is not None else AIETileType.MemTile
    )
    ObjectFifoEndpoint.__init__(self, tile.with_type(default_type))

Runtime

The host-side orchestration entry point that declares fill / drain / start operations via rt.sequence(...).

Runtime: orchestrates host-side data movement and worker execution for an IRON program.

IronRuntimeError

Bases: Exception

Raised by the IRON Runtime when resolution encounters an unrecoverable state.

Runtime

Runtime(strict_task_groups: bool = True)

Bases: Resolvable

The host-side sequence of data-movement and worker-start operations that execute an IRON design.

A Runtime describes what the host does at runtime: filling input ObjectFifos with data, starting Workers, and draining results back to host buffers. Operations are declared inside a sequence context.

Initialize a runtime object.

Parameters:

Name	Type	Description	Default
strict_task_groups	bool	Disallows mixing the default group and explicit task groups during resolution. This can catch common errors, but can be set to False to disable the checks.	True

Source code in python/iron/runtime/runtime.py

def __init__(
    self,
    strict_task_groups: bool = True,
) -> None:
    """Initialize a runtime object.

    Args:
        strict_task_groups (bool): Disallows mixing the default group and explicit task groups during resolution.
            This can catch common errors, but can be set to False to disable the checks.

    """
    self._rt_data = []
    self._tasks: list[Resolvable] = []
    self._fifos: set[ObjectFifoHandle] = set()
    self._workers: list[Worker] = []
    # Lower-level explicit-routing primitives (peers of ObjectFifo for
    # designs that hand-wire flows + DMA programs instead of letting
    # ObjectFifo manage them).
    self._flows = []
    self._locks = []
    self._tile_dmas = []
    self._scratchpad_parameters: list[ScratchpadParameter] = []
    self._open_task_groups = []
    self._trace_size = None
    self._trace_workers = None
    self._strict_task_groups = strict_task_groups
    self._task_group_index = itertools.count()
    self._reuse_output_buffer = False
    self._egress_shim_col = 0

workers property

workers: list[Worker]

The workers associated with the Runtime by calls to start()

fifos property

fifos: list[ObjectFifoHandle]

The ObjectFifoHandles associated with the Runtime by calls to fill() and drain()

add_flow

add_flow(flow) -> None

Register an explicit Flow (or PacketFlow) so the Program resolves it alongside the ObjectFifos.

Source code in python/iron/runtime/runtime.py

def add_flow(self, flow) -> None:
    """Register an explicit [`Flow`][iron.Flow] (or
    [`PacketFlow`][iron.PacketFlow]) so the Program resolves it alongside
    the ObjectFifos."""
    self._flows.append(flow)

add_lock

add_lock(lock) -> None

Register an explicit Lock shared between a Worker and a TileDma.

Source code in python/iron/runtime/runtime.py

def add_lock(self, lock) -> None:
    """Register an explicit [`Lock`][iron.Lock] shared between a Worker and
    a [`TileDma`][iron.TileDma]."""
    self._locks.append(lock)

add_tile_dma

add_tile_dma(tile_dma) -> None

Register an explicit TileDma program.

Source code in python/iron/runtime/runtime.py

def add_tile_dma(self, tile_dma) -> None:
    """Register an explicit [`TileDma`][iron.TileDma] program."""
    self._tile_dmas.append(tile_dma)

sequence

sequence(
    *input_types: type[ndarray],
) -> Iterator[RuntimeData | tuple[RuntimeData, ...]]

A RuntimeSequence is a sequence of operations that are performed in support of a program. Common operations include input and output data movement.

Raises:

Type	Description
ValueError	Arguments are validated.
ValueError	If task groups are not finished within the sequence() context, and error will be raised.

Yields:

Type	Description
RuntimeData | tuple[RuntimeData, ...]	RuntimeData | tuple[RuntimeData, ...]: Handles to the runtime buffers matching the declared input types.

Source code in python/iron/runtime/runtime.py

@contextmanager
def sequence(
    self, *input_types: type[np.ndarray]
) -> Iterator[RuntimeData | tuple[RuntimeData, ...]]:
    """A RuntimeSequence is a sequence of operations that are performed in
    support of a program. Common operations include input and output data movement.

    Raises:
        ValueError: Arguments are validated.
        ValueError: If task groups are not finished within the sequence() context, and error will be raised.

    Yields:
        RuntimeData | tuple[RuntimeData, ...]: Handles to the runtime buffers matching the declared input types.
    """
    try:
        self._rt_data = list(map(RuntimeData, input_types))
        if len(self._rt_data) == 1:
            yield self._rt_data[0]
        else:
            yield tuple(self._rt_data.copy())
    finally:
        if len(self._open_task_groups) != 0:
            tgs_str = ", ".join([str(t) for t in self._open_task_groups])
            raise ValueError(f"Failed to close task groups: {tgs_str}")
        for of_handle in self._fifos:
            # It's very easy to accidentally generate multiple (identical)
            # consumers in the runtime. This bit of code prunes out duplicates.
            if not of_handle._is_prod:
                fifo_obj = of_handle._object_fifo
                runtime_cons = None
                to_remove = []
                for c in fifo_obj._cons:
                    if isinstance(c.endpoint, RuntimeEndpoint):
                        if not runtime_cons:
                            runtime_cons = c
                        else:
                            if (
                                c.depth == runtime_cons.depth
                                and c.dims_from_stream
                                == runtime_cons.dims_from_stream
                            ):
                                to_remove.append(c)
                            else:
                                raise ValueError(
                                    f"Found two different RuntimeEndpoints for consumers of the same ObjectFifo: {fifo_obj}"
                                )
                for r in to_remove:
                    fifo_obj._cons.remove(r)

task_group

task_group() -> RuntimeTaskGroup

Generate a handle to a RuntimeTaskGroup. This should be called within a Runtime.sequence() context.

Returns:

Name	Type	Description
RuntimeTaskGroup	RuntimeTaskGroup	The new RuntimeTaskGroup

Source code in python/iron/runtime/runtime.py

def task_group(self) -> RuntimeTaskGroup:
    """Generate a handle to a RuntimeTaskGroup.
    This should be called within a Runtime.sequence() context.

    Returns:
        RuntimeTaskGroup: The new RuntimeTaskGroup
    """
    tg = RuntimeTaskGroup(next(self._task_group_index))
    self._open_task_groups.append(tg)
    return tg

finish_task_group

finish_task_group(task_group: RuntimeTaskGroup)

Close out a RuntimeTaskGroup. This should be called within a Runtime.sequence() context.

Parameters:

Name	Type	Description	Default
task_group	RuntimeTaskGroup	The task group to close. All associated tasks will be awaited or freed.	required

Source code in python/iron/runtime/runtime.py

def finish_task_group(self, task_group: RuntimeTaskGroup):
    """Close out a RuntimeTaskGroup.
    This should be called within a Runtime.sequence() context.

    Args:
        task_group (RuntimeTaskGroup): The task group to close. All associated tasks will be awaited or freed.
    """
    self._open_task_groups.remove(task_group)
    self._tasks.append(FinishTaskGroupTask(task_group))

fill

fill(
    in_fifo: ObjectFifoHandle,
    source: RuntimeData,
    tap: TensorAccessPattern | None = None,
    task_group: RuntimeTaskGroup | None = None,
    wait: bool = False,
    tile: Tile = AnyShimTile,
    packet: tuple[int, int] | None = None,
    offset_parameter: "ScratchpadParameter | str | None" = None,
) -> None

Conceptually fill an ObjectFifoHandle (of type producer) with data from a runtime buffer. This should be called within a Runtime.sequence() context.

Parameters:

Name	Type	Description	Default
in_fifo	ObjectFifoHandle	The producer ObjectFifoHandle.	required
source	RuntimeData	The input Runtime data buffer.	required
tap	TensorAccessPattern | None	A way of specifying how data in the buffer is accessed when sending it to the in_fifo. If None is given, this will default to a linear transfer containing all data in the source buffer. Defaults to None.	None
task_group	RuntimeTaskGroup | None	A TaskGroup to associate this task with. Defaults to None.	None
wait	bool	Whether this Task should be awaited on or not. If not, it will be freed when the task group is finished. Defaults to False.	False
tile	Tile | None	The Shim tile to associate the data transfer with. Defaults to AnyShimTile.	AnyShimTile
packet	tuple[int, int] | None	Stamp the shim DMA's BD with a packet header (pkt_type, pkt_id). Pairs with downstream packet-switched routing (e.g. ObjectFifos lowered with --packet-sw-objFifos or an explicit PacketFlow). Defaults to None.	None
offset_parameter	ScratchpadParameter | str | None	A ScratchpadParameter (or its name) whose value is used as the element offset for this DMA transfer. Defaults to None.	None

Raises:

Type	Description
ValueError	Arguments are validated.

Source code in python/iron/runtime/runtime.py

def fill(
    self,
    in_fifo: ObjectFifoHandle,
    source: RuntimeData,
    tap: TensorAccessPattern | None = None,
    task_group: RuntimeTaskGroup | None = None,
    wait: bool = False,
    tile: Tile = AnyShimTile,
    packet: tuple[int, int] | None = None,
    offset_parameter: "ScratchpadParameter | str | None" = None,
) -> None:
    """Conceptually fill an ObjectFifoHandle (of type producer) with data from a runtime buffer.
    This should be called within a Runtime.sequence() context.

    Args:
        in_fifo (ObjectFifoHandle): The producer ObjectFifoHandle.
        source (RuntimeData): The input Runtime data buffer.
        tap (TensorAccessPattern | None, optional): A way of specifying how data in the buffer is accessed when sending it to the in_fifo.
            If None is given, this will default to a linear transfer containing all data in the source buffer. Defaults to None.
        task_group (RuntimeTaskGroup | None, optional): A TaskGroup to associate this task with. Defaults to None.
        wait (bool, optional): Whether this Task should be awaited on or not. If not, it will be freed when the task group is finished. Defaults to False.
        tile (Tile | None, optional): The Shim tile to associate the data transfer with. Defaults to AnyShimTile.
        packet (tuple[int, int] | None, optional): Stamp the shim DMA's BD
            with a packet header `(pkt_type, pkt_id)`. Pairs with
            downstream packet-switched routing (e.g. ObjectFifos lowered
            with `--packet-sw-objFifos` or an explicit
            [`PacketFlow`][iron.PacketFlow]). Defaults to None.
        offset_parameter (ScratchpadParameter | str | None, optional): A ScratchpadParameter (or its name) whose value is used as the element offset for this DMA transfer. Defaults to None.

    Raises:
        ValueError: Arguments are validated.
    """
    if source not in self._rt_data:
        raise ValueError(
            f"Source {source} is not a RuntimeData object generated by sequence()"
        )
    rt_endpoint = RuntimeEndpoint(tile)

    if tap is None:
        tap = source.default_tap()

    offset_param_name = None
    if offset_parameter is not None:
        if isinstance(offset_parameter, ScratchpadParameter):
            offset_param_name = offset_parameter.name
            if offset_parameter not in self._scratchpad_parameters:
                self._scratchpad_parameters.append(offset_parameter)
        else:
            offset_param_name = offset_parameter

    in_fifo.endpoint = rt_endpoint
    self._fifos.add(in_fifo)
    self._tasks.append(
        DMATask(
            in_fifo,
            source,
            tap,
            task_group,
            wait,
            offset_param_name,
            packet=packet,
        )
    )

drain

drain(
    out_fifo: ObjectFifoHandle,
    dest: RuntimeData,
    tap: TensorAccessPattern | None = None,
    task_group: RuntimeTaskGroup | None = None,
    wait: bool = False,
    tile: Tile = AnyShimTile,
    packet: tuple[int, int] | None = None,
    offset_parameter: "ScratchpadParameter | str | None" = None,
) -> None

Conceptually drain an ObjectFifoHandle (of type consumer): read data from the ObjectFifo and write it to a runtime buffer. This should be called within a Runtime.sequence() context.

Parameters:

Name	Type	Description	Default
out_fifo	ObjectFifoHandle	The consumer ObjectFifoHandle.	required
dest	RuntimeData	The output Runtime data buffer.	required
tap	TensorAccessPattern | None	A way of specifying how data in the buffer is accessed when reading from the out_fifo. If None is given, this will default to a linear transfer containing all data in the destination buffer. Defaults to None.	None
task_group	RuntimeTaskGroup | None	A TaskGroup to associate this task with. Defaults to None.	None
wait	bool	Whether this Task should be awaited on or not. If not, it will be freed when the task group is finished. Defaults to False.	False
tile	Tile | None	The Shim tile to associate the data transfer with. Defaults to AnyShimTile.	AnyShimTile
packet	tuple[int, int] | None	Stamp the shim DMA's BD with a packet header (pkt_type, pkt_id). Pairs with downstream packet-switched routing (e.g. ObjectFifos lowered with --packet-sw-objFifos or an explicit PacketFlow). Defaults to None.	None
offset_parameter	ScratchpadParameter | str | None	A ScratchpadParameter (or its name) whose value is used as the element offset for this DMA transfer. Defaults to None.	None

Raises:

Type	Description
ValueError	Arguments are validated.

Source code in python/iron/runtime/runtime.py

def drain(
    self,
    out_fifo: ObjectFifoHandle,
    dest: RuntimeData,
    tap: TensorAccessPattern | None = None,
    task_group: RuntimeTaskGroup | None = None,
    wait: bool = False,
    tile: Tile = AnyShimTile,
    packet: tuple[int, int] | None = None,
    offset_parameter: "ScratchpadParameter | str | None" = None,
) -> None:
    """Conceptually drain an ObjectFifoHandle (of type consumer): read data from the ObjectFifo and write it to a runtime buffer.
    This should be called within a Runtime.sequence() context.

    Args:
        out_fifo (ObjectFifoHandle): The consumer ObjectFifoHandle.
        dest (RuntimeData): The output Runtime data buffer.
        tap (TensorAccessPattern | None, optional): A way of specifying how data in the buffer is accessed when reading from the out_fifo.
            If None is given, this will default to a linear transfer containing all data in the destination buffer. Defaults to None.
        task_group (RuntimeTaskGroup | None, optional): A TaskGroup to associate this task with. Defaults to None.
        wait (bool, optional): Whether this Task should be awaited on or not. If not, it will be freed when the task group is finished. Defaults to False.
        tile (Tile | None, optional): The Shim tile to associate the data transfer with. Defaults to AnyShimTile.
        packet (tuple[int, int] | None, optional): Stamp the shim DMA's BD
            with a packet header `(pkt_type, pkt_id)`. Pairs with
            downstream packet-switched routing (e.g. ObjectFifos lowered
            with `--packet-sw-objFifos` or an explicit
            [`PacketFlow`][iron.PacketFlow]). Defaults to None.
        offset_parameter (ScratchpadParameter | str | None, optional): A ScratchpadParameter (or its name) whose value is used as the element offset for this DMA transfer. Defaults to None.

    Raises:
        ValueError: Arguments are validated.
    """
    if dest not in self._rt_data:
        raise ValueError(
            f"Destination {dest} is not a RuntimeData object generated by sequence()"
        )
    rt_endpoint = RuntimeEndpoint(tile)

    if tap is None:
        tap = dest.default_tap()

    offset_param_name = None
    if offset_parameter is not None:
        if isinstance(offset_parameter, ScratchpadParameter):
            offset_param_name = offset_parameter.name
            if offset_parameter not in self._scratchpad_parameters:
                self._scratchpad_parameters.append(offset_parameter)
        else:
            offset_param_name = offset_parameter

    out_fifo.endpoint = rt_endpoint
    self._fifos.add(out_fifo)
    self._tasks.append(
        DMATask(
            out_fifo,
            dest,
            tap,
            task_group,
            wait,
            offset_param_name,
            packet=packet,
        )
    )

start

start(*args: Worker)

A placeholder operation to indicate that one or more Worker should be started on the device. This should be called within a Runtime.sequence() context.

Parameters:

Name	Type	Description	Default
*args	Worker	One or more Workers. If more than one is given, they will be started in order.	()

Raises:

Type	Description
ValueError	Arguments are validated.

Source code in python/iron/runtime/runtime.py

def start(self, *args: Worker):
    """A placeholder operation to indicate that one or more Worker should be started on the device.
    This should be called within a Runtime.sequence() context.

    Args:
        *args: One or more Workers. If more than one is given, they will be started in order.

    Raises:
        ValueError: Arguments are validated.
    """
    for worker in args:
        if not isinstance(worker, Worker):
            raise ValueError("Runtime can only start Worker objects")
        self._workers.append(worker)
        self._tasks.append(RuntimeStartTask(worker))

inline_ops

inline_ops(inline_func: Callable, inline_args: list)

Insert an InlineOpRuntimeTask into the runtime. This should be called within a Runtime.sequence() context.

Parameters:

Name	Type	Description	Default
inline_func	Callable	The function to execute within an MLIR context.	required
inline_args	list	The state the function needs to execute. Any ObjectFifoHandle passed here is registered with the Runtime (so the Program resolves its shim allocation) and, if it has no endpoint yet, is bound to a shim tile -- an inline op driving a fifo from the runtime sequence is a host-side (shim) endpoint. This mirrors how InlineOpRuntimeTask resolves Buffer args.	required

Source code in python/iron/runtime/runtime.py

def inline_ops(self, inline_func: Callable, inline_args: list):
    """Insert an InlineOpRuntimeTask into the runtime.
     This should be called within a Runtime.sequence() context.

    Args:
        inline_func (Callable): The function to execute within an MLIR context.
        inline_args (list): The state the function needs to execute. Any
            ObjectFifoHandle passed here is registered with the Runtime (so
            the Program resolves its shim allocation) and, if it has no
            endpoint yet, is bound to a shim tile -- an inline op driving a
            fifo from the runtime sequence is a host-side (shim) endpoint.
            This mirrors how InlineOpRuntimeTask resolves Buffer args.
    """
    for arg in _iter_flat(inline_args):
        if isinstance(arg, ObjectFifoHandle):
            if arg.endpoint is None:
                arg.endpoint = RuntimeEndpoint(AnyShimTile)
            self._fifos.add(arg)
    self._tasks.append(InlineOpRuntimeTask(inline_func, inline_args))

enable_trace

enable_trace(
    trace_size: int | None = None,
    workers: list | None = None,
    reuse_output_buffer: bool = False,
    coretile_events: list | None = None,
    coremem_events: list | None = None,
    memtile_events: list | None = None,
    shimtile_events: list | None = None,
    egress_shim_col: int = 0,
)

Enable hardware tracing for this program.

Configures the AIE trace units and routes trace packets to DDR via the shim DMA. Should be called within a sequence context before data movement operations.

Parameters:

Name	Type	Description	Default
trace_size	int	Size of the trace buffer in bytes.	None
workers	list[Worker] | None	Specific workers to trace. If None, all workers with trace set will be traced. Defaults to None.	None
reuse_output_buffer	bool	When False (default), trace lowering appends a dedicated trace-buffer argument to the runtime_sequence; it lands at the tail so enabling trace never perturbs the data arguments' indices. When True, trace data is written into the tail of the last output buffer, saving a host buffer. Defaults to False.	False
coretile_events	list | None	List of up to 8 core tile trace events. See the AIEX dialect reference for available events under (type)EventAIE such as CoreEventAIE. Defaults to None (uses hardware defaults).	None
coremem_events	list | None	List of up to 8 core memory trace events. Defaults to None (uses hardware defaults).	None
memtile_events	list | None	List of up to 8 mem tile trace events. Defaults to None (uses hardware defaults).	None
shimtile_events	list | None	List of up to 8 shim tile trace events. Defaults to None (uses hardware defaults).	None
egress_shim_col	int	Column of the shim tile used to egress trace packets to DDR. Defaults to 0.	0

Source code in python/iron/runtime/runtime.py

def enable_trace(
    self,
    trace_size: int | None = None,
    workers: list | None = None,
    reuse_output_buffer: bool = False,
    coretile_events: list | None = None,
    coremem_events: list | None = None,
    memtile_events: list | None = None,
    shimtile_events: list | None = None,
    egress_shim_col: int = 0,
):
    """Enable hardware tracing for this program.

    Configures the AIE trace units and routes trace packets to DDR via the shim DMA.
    Should be called within a [`sequence`][iron.runtime.runtime.Runtime.sequence] context before data movement operations.

    Args:
        trace_size (int): Size of the trace buffer in bytes.
        workers (list[Worker] | None, optional): Specific workers to trace. If None,
            all workers with ``trace`` set will be traced. Defaults to None.
        reuse_output_buffer (bool, optional): When False (default), trace
            lowering appends a dedicated trace-buffer argument to the
            runtime_sequence; it lands at the tail so enabling trace never
            perturbs the data arguments' indices. When True, trace data is
            written into the tail of the last output buffer, saving a host
            buffer. Defaults to False.
        coretile_events (list | None, optional): List of up to 8 core tile trace events.
            See [the AIEX dialect reference](../AIEXDialect.md) for available
            events under (type)EventAIE such as CoreEventAIE.
            Defaults to None (uses hardware defaults).
        coremem_events (list | None, optional): List of up to 8 core memory trace events.
            Defaults to None (uses hardware defaults).
        memtile_events (list | None, optional): List of up to 8 mem tile trace events.
            Defaults to None (uses hardware defaults).
        shimtile_events (list | None, optional): List of up to 8 shim tile trace events.
            Defaults to None (uses hardware defaults).
        egress_shim_col (int, optional): Column of the shim tile used to
            egress trace packets to DDR. Defaults to 0.
    """
    self._trace_size = trace_size
    self._trace_workers = workers
    self._reuse_output_buffer = reuse_output_buffer
    self._coretile_events = coretile_events
    self._coremem_events = coremem_events
    self._memtile_events = memtile_events
    self._shimtile_events = shimtile_events
    self._egress_shim_col = egress_shim_col

set_barrier

set_barrier(barrier: WorkerRuntimeBarrier, value: int)

Set the value of a worker barrier. This should be called within a Runtime.sequence() context.

Parameters:

Name	Type	Description	Default
barrier	WorkerRuntimeBarrier	The WorkerRuntimeBarrier to set.	required
value	int	The value to set the barrier to.	required

Source code in python/iron/runtime/runtime.py

def set_barrier(self, barrier: WorkerRuntimeBarrier, value: int):
    """Set the value of a worker barrier.
    This should be called within a Runtime.sequence() context.

    Args:
        barrier (WorkerRuntimeBarrier): The WorkerRuntimeBarrier to set.
        value (int): The value to set the barrier to.
    """
    self._tasks.append(_BarrierSetOp(barrier, value))

sync_parameters

sync_parameters()

Emit aiex.sync_scratchpad_parameters_from_host in the runtime sequence.

Call this within a sequence context after all parameters have been written on the host side and before starting workers that read them.

Source code in python/iron/runtime/runtime.py

def sync_parameters(self):
    """Emit `aiex.sync_scratchpad_parameters_from_host` in the runtime sequence.

    Call this within a [`sequence`][iron.runtime.runtime.Runtime.sequence] context after all
    parameters have been written on the host side and before starting
    workers that read them.
    """
    self._tasks.append(_SyncParametersTask())

Buffer

Named memory region accessible by both Workers and the Runtime.

Buffer

Buffer(
    type: type[ndarray] | None = None,
    initial_value: ndarray | None = None,
    name: str | None = None,
    tile: Tile | None = None,
    use_write_rtp: bool = False,
    address: int | None = None,
)

Bases: Resolvable

A buffer that is available both to Workers and to the Runtime for operations. This is often used for Runtime Parameters.

A Buffer is a memory region declared at the top-level of the design, allowing it to be accessed by both Workers and the Runtime.

Parameters:

Name	Type	Description	Default
type	type[ndarray] | None	The type of the buffer. Defaults to None.	None
initial_value	ndarray | None	An initial value to set the buffer to. Should be of same datatype and shape as the buffer. Defaults to None.	None
name	str | None	The name of the buffer. If none is given, a unique name will be generated. Defaults to None.	None
tile	Tile | None	The tile for the buffer. Automatically set to the Worker's tile when the buffer is passed in the Worker's fn_args list. Defaults to None.	None
use_write_rtp	bool	If use_write_rtp, write_rtp/read_rtp operations will be generated. Otherwise, traditional write/read operations will be used. Defaults to False.	False
address	int | None	Pin the buffer to a fixed L1 address. Needed for host-written RTP buffers the runtime pokes at a hardcoded address. Defaults to None (compiler-assigned).	None

Raises:

Type	Description
ValueError	If neither type nor initial_value is provided.

Source code in python/iron/buffer.py

def __init__(
    self,
    type: type[np.ndarray] | None = None,
    initial_value: np.ndarray | None = None,
    name: str | None = None,
    tile: Tile | None = None,
    use_write_rtp: bool = False,
    address: int | None = None,
):
    """A Buffer is a memory region declared at the top-level of the design, allowing it to
    be accessed by both Workers and the Runtime.

    Args:
        type (type[np.ndarray] | None, optional): The type of the buffer. Defaults to None.
        initial_value (np.ndarray | None, optional): An initial value to set the buffer to. Should be of same datatype and shape as the buffer. Defaults to None.
        name (str | None, optional): The name of the buffer. If none is given, a unique name will be generated. Defaults to None.
        tile (Tile | None, optional): The tile for the buffer. Automatically set to the Worker's tile when the buffer is passed in the Worker's fn_args list. Defaults to None.
        use_write_rtp (bool, optional): If use_write_rtp, write_rtp/read_rtp operations will be generated. Otherwise, traditional write/read operations will be used. Defaults to False.
        address (int | None, optional): Pin the buffer to a fixed L1 address. Needed for host-written RTP buffers the runtime pokes at a hardcoded address. Defaults to None (compiler-assigned).

    Raises:
        ValueError: If neither ``type`` nor ``initial_value`` is provided.
    """
    if type is None and initial_value is None:
        raise ValueError("Must provide either type, initial value, or both.")
    if type is None:
        assert initial_value is not None
        type = np.ndarray[initial_value.shape, np.dtype[initial_value.dtype.type]]
    self._initial_value = initial_value
    self._name = name
    self._op = None
    self._arr_type = type
    if not self._name:
        self._name = f"buf_{next(Buffer._gbuf_index)}"
    self._use_write_rtp = use_write_rtp
    self._address = address
    self._tile = tile
    # Whether the user pinned this Buffer to an explicit tile at
    # construction.  A Worker may auto-pin _tile later as a
    # convenience, so `_tile is not None` is not a reliable signal of
    # user intent; this flag is.  Only explicitly-placed Buffers may be
    # shared (read) across Workers — see Worker.
    self._explicit_tile = tile is not None
    self._owner_worker: "Worker | None" = None

tile property

tile: Tile | None

The tile this buffer is on.

shape property

shape: Sequence[int]

The shape of the buffer

dtype property

dtype: NpuDType

The per-element datatype of the buffer.

tiles

tiles() -> list

Tile dependency for Program.resolve tile discovery.

Pinned Buffers (e.g. a compute Worker reading a neighbor tile's L1 directly) need their tile registered with the Device before resolve runs. Worker-attached Buffers without an explicit placement get pinned to the Worker's tile in Worker's constructor, which is already discoverable via Worker.tile; this method just exposes any extra (cross-tile) placements.

Source code in python/iron/buffer.py

def tiles(self) -> list:
    """Tile dependency for Program.resolve tile discovery.

    Pinned Buffers (e.g. a compute [`Worker`][iron.Worker] reading a
    neighbor tile's L1 directly) need their tile registered with the
    Device before `resolve` runs. Worker-attached Buffers without an
    explicit placement get pinned to the Worker's tile in
    [`Worker`][iron.Worker]'s constructor, which is already discoverable
    via `Worker.tile`; this method just exposes any extra (cross-tile)
    placements.
    """
    return [self._tile] if self._tile is not None else []

Kernels

Kernel and ExternalFunction: wrappers for pre-compiled and C++ AIE compute kernels.

BaseKernel

BaseKernel(
    name: str,
    arg_types: list[type[ndarray] | dtype] | None = None,
)

Bases: Resolvable

Base class for AIE core functions that resolve to a func.func declaration.

Subclasses

Kernel: wraps a pre-compiled object file. ExternalFunction: compiles C/C++ source at JIT time.

Parameters:

Name	Type	Description	Default
name	str	Symbol name of the function.	required
arg_types	list[type[ndarray] | dtype] | None	Type signature of the function arguments. Defaults to None (empty list).	None

Source code in python/iron/kernel.py

def __init__(
    self,
    name: str,
    arg_types: list[type[np.ndarray] | np.dtype] | None = None,
):
    """
    Args:
        name: Symbol name of the function.
        arg_types: Type signature of the function arguments.  Defaults to None (empty list).
    """
    if not name:
        raise ValueError("Kernel name cannot be empty.")
    self._name = name
    self._arg_types = arg_types if arg_types is not None else []
    self._op: FuncOp | None = None

arg_shape

arg_shape(arg_index: int = 0) -> tuple[int, ...]

Return the shape tuple of the array argument at arg_index.

Works for both np.ndarray[(...,), np.dtype[T]] parameterized types (the canonical IRON kernel signature) and MLIR MemRefType operands.

Parameters:

Name	Type	Description	Default
arg_index	int	Index into arg_types. Defaults to 0.	0

Raises:

Type	Description
ValueError	When arg_index is out of range or the argument at that index is not an array type.

Source code in python/iron/kernel.py

def arg_shape(self, arg_index: int = 0) -> tuple[int, ...]:
    """Return the shape tuple of the array argument at `arg_index`.

    Works for both `np.ndarray[(...,), np.dtype[T]]` parameterized
    types (the canonical IRON kernel signature) and MLIR MemRefType
    operands.

    Args:
        arg_index: Index into `arg_types`. Defaults to 0.

    Raises:
        ValueError: When `arg_index` is out of range or the
            argument at that index is not an array type.
    """
    arg = self._resolve_arg(arg_index)
    type_args = getattr(arg, "__args__", None)
    if type_args is not None and len(type_args) > 0:
        shape_arg = type_args[0]
        if isinstance(shape_arg, tuple):
            return shape_arg
    shape = getattr(arg, "shape", None)
    if shape is not None:
        return tuple(shape)
    raise ValueError(
        f"Argument {arg_index} does not have a shape or is not an array type."
    )

arg_dtype

arg_dtype(arg_index: int = 0)

Return the numpy dtype of the array argument at arg_index.

Parameters:

Name	Type	Description	Default
arg_index	int	Index into arg_types. Defaults to 0.	0

Raises:

Type	Description
ValueError	When arg_index is out of range or the argument at that index is not an array type.

Source code in python/iron/kernel.py

def arg_dtype(self, arg_index: int = 0):
    """Return the numpy dtype of the array argument at `arg_index`.

    Args:
        arg_index: Index into `arg_types`. Defaults to 0.

    Raises:
        ValueError: When `arg_index` is out of range or the
            argument at that index is not an array type.
    """
    arg = self._resolve_arg(arg_index)
    type_args = getattr(arg, "__args__", None)
    if type_args is not None and len(type_args) >= 2:
        dt = type_args[1]
        dt_args = getattr(dt, "__args__", None)
        return np.dtype(dt_args[0]) if dt_args is not None else np.dtype(dt)
    dtype = getattr(arg, "dtype", None)
    if dtype is not None:
        return np.dtype(dtype)
    raise ValueError(
        f"Argument {arg_index} does not have a dtype or is not an array type."
    )

tile_size

tile_size(arg_index: int = 0) -> int

Return the first dimension of the array argument at arg_index.

Convenience wrapper over arg_shape for the common case of a 1-D buffer argument. tile_size(i) is equivalent to arg_shape(i)[0].

Parameters:

Name	Type	Description	Default
arg_index	int	Index into arg_types. Defaults to 0.	0

Source code in python/iron/kernel.py

def tile_size(self, arg_index: int = 0) -> int:
    """Return the first dimension of the array argument at `arg_index`.

    Convenience wrapper over
    [`arg_shape`][iron.kernel.BaseKernel.arg_shape] for the common case of
    a 1-D buffer argument. `tile_size(i)` is equivalent to
    `arg_shape(i)[0]`.

    Args:
        arg_index: Index into `arg_types`. Defaults to 0.
    """
    shape = self.arg_shape(arg_index)
    if len(shape) == 0:
        raise ValueError(
            f"Argument {arg_index} does not have a shape or is not an array type."
        )
    return shape[0]

arg_types

arg_types() -> list

Return a copy of the argument type list.

Source code in python/iron/kernel.py

def arg_types(self) -> list:
    """Return a copy of the argument type list."""
    return self._arg_types.copy()

Kernel

Kernel(
    name: str,
    object_file_name: str,
    arg_types: list[type[ndarray] | dtype] | None = None,
)

Bases: BaseKernel

An AIE core function backed by a pre-compiled object file.

Use ExternalFunction instead when you want to compile from C/C++ source at JIT time.

resolve() emits a func.func private declaration with a link_with attribute naming object_file_name. The aie-assign-core-link-files pass propagates this into the CoreOp's link_files attribute so the linker knows which file to include.

Parameters:

Name	Type	Description	Default
name	str	Symbol name of the function as it appears in the object file.	required
object_file_name	str	Filename of the pre-compiled object file (e.g. "add_one.o"). Must be on the linker search path at compile time.	required
arg_types	list[type[ndarray] | dtype] | None	Type signature of the function arguments. Defaults to None (empty list).	None

Source code in python/iron/kernel.py

def __init__(
    self,
    name: str,
    object_file_name: str,
    arg_types: list[type[np.ndarray] | np.dtype] | None = None,
) -> None:
    """
    Args:
        name: Symbol name of the function as it appears in the object file.
        object_file_name: Filename of the pre-compiled object file
            (e.g. ``"add_one.o"``).  Must be on the linker search path
            at compile time.
        arg_types: Type signature of the function arguments.  Defaults to None (empty list).
    """
    super().__init__(name, arg_types)
    self._object_file_name = object_file_name

object_file_name property

object_file_name: str

Filename of the compiled object file.

ExternalFunction

ExternalFunction(
    name: str,
    object_file_name: str | None = None,
    source_file: str | None = None,
    source_string: str | None = None,
    arg_types: list[type[ndarray] | dtype] | None = None,
    include_dirs: list[str] | None = None,
    compile_flags: list[str] | None = None,
    *,
    symbol_prefix: str | None = None,
    use_chess: bool = False
)

Bases: Kernel

An AIE core function compiled from C/C++ source at JIT time.

Each instance is registered in _instances at construction time so that the @jit decorator can discover and compile all source files before invoking the MLIR compilation pipeline. _instances is cleared at the start of each @jit call to prevent stale registrations from a previous (possibly failed) run.

Use the base Kernel class instead when you have a pre-built object file.

Parameters:

Name	Type	Description	Default
name	str	Symbol name of the function as it will appear in the object file.	required
object_file_name	str | None	Output object file name. Defaults to <effective_name>.o.	None
source_file	str | None	Path to a C/C++ source file on disk. Mutually exclusive with source_string.	None
source_string	str | None	Inline C/C++ source code. Mutually exclusive with source_file.	None
arg_types	list[type[ndarray] | dtype] | None	Type signature of the function arguments. Defaults to None (empty list).	None
include_dirs	list[str] | None	Additional -I directories passed to the chosen compiler (Peano by default; xchesscc when use_chess=True). Defaults to None (empty list).	None
compile_flags	list[str] | None	Additional flags passed verbatim to the chosen compiler. Defaults to None (empty list).	None
symbol_prefix	str | None	Optional prefix for the exported symbol name. When set, the effective symbol name becomes <symbol_prefix>_<name> and the object file is named accordingly. The original name is preserved in _original_name for source file naming.	None
use_chess	bool	When True, this ExternalFunction's source is compiled with xchesscc_wrapper instead of Peano's clang++. The JIT compile orchestration auto-detects the design-level toolchain from the registered EFs and switches aiecc's front-end accordingly; mixing chess + peano EFs in one design is rejected loudly because aiecc only invokes one front-end per compile.	False

Source code in python/iron/kernel.py

def __init__(
    self,
    name: str,
    object_file_name: str | None = None,
    source_file: str | None = None,
    source_string: str | None = None,
    arg_types: list[type[np.ndarray] | np.dtype] | None = None,
    include_dirs: list[str] | None = None,
    compile_flags: list[str] | None = None,
    *,
    symbol_prefix: str | None = None,
    use_chess: bool = False,
) -> None:
    """
    Args:
        name: Symbol name of the function as it will appear in the object
            file.
        object_file_name: Output object file name.  Defaults to
            ``<effective_name>.o``.
        source_file: Path to a C/C++ source file on disk.  Mutually
            exclusive with ``source_string``.
        source_string: Inline C/C++ source code.  Mutually exclusive with
            ``source_file``.
        arg_types: Type signature of the function arguments.  Defaults to
            None (empty list).
        include_dirs: Additional ``-I`` directories passed to the chosen
            compiler (Peano by default; xchesscc when ``use_chess=True``).
            Defaults to None (empty list).
        compile_flags: Additional flags passed verbatim to the chosen
            compiler.  Defaults to None (empty list).
        symbol_prefix: Optional prefix for the exported symbol name.  When
            set, the effective symbol name becomes ``<symbol_prefix>_<name>``
            and the object file is named accordingly.  The original name is
            preserved in ``_original_name`` for source file naming.
        use_chess: When ``True``, this ExternalFunction's source is
            compiled with ``xchesscc_wrapper`` instead of Peano's
            ``clang++``.  The JIT compile orchestration auto-detects the
            design-level toolchain from the registered EFs and switches
            aiecc's front-end accordingly; mixing chess + peano EFs in
            one design is rejected loudly because aiecc only invokes one
            front-end per compile.
    """
    self._original_name = name
    self._symbol_prefix = symbol_prefix
    effective_name = f"{symbol_prefix}_{name}" if symbol_prefix else name
    object_file_name_explicit = object_file_name is not None
    if not object_file_name:
        object_file_name = f"{effective_name}.o"
    super().__init__(effective_name, object_file_name, arg_types)

    if source_file is not None:
        self._source_file = source_file
        self._source_string = None
    elif source_string is not None:
        self._source_file = None
        self._source_string = source_string
    else:
        raise ValueError("source_file or source_string must be provided.")

    self._include_dirs = include_dirs if include_dirs is not None else []
    self._compile_flags = compile_flags if compile_flags is not None else []
    self._use_chess = use_chess
    self._compiled = False
    self._cached_digest: str | None = None

    # Two same-name EFs with default object_file_name would collide on
    # the same .o path. Auto-suffix defaulted names with a content digest;
    # raise on explicit names so silent renames don't surprise the caller.
    for existing in ExternalFunction._instances:
        if (
            existing._name == effective_name
            and existing._object_file_name == object_file_name
            and existing._content_digest() != self._content_digest()
        ):
            if object_file_name_explicit:
                raise ValueError(
                    f"ExternalFunction '{effective_name}' would collide with "
                    f"an already-registered instance: same name and "
                    f"explicit object_file_name='{object_file_name}' but "
                    f"different compile_flags / source.  Distinguish them "
                    f"by passing a distinct `object_file_name=...` or "
                    f"`name=...`."
                )
            suffix = self._content_digest()[:8]
            object_file_name = f"{effective_name}_{suffix}.o"
            self._object_file_name = object_file_name
            break
    ExternalFunction._instances.add(self)

ScratchpadParameter

ScratchpadParameter: a named runtime value set from the host and read by Workers.

ScratchpadParameter

ScratchpadParameter(name: str, dtype: NpuDType)

Bases: Resolvable

A named runtime parameter communicated from host to AIE cores via the scratchpad mechanism.

Declare a ScratchpadParameter at design time. Pass it to a Worker via fn_args and call read inside the core_fn to obtain its current value. The --aie-lower-scratchpad-parameters pass automatically inserts the necessary lock and scratchpad-sync preamble ops.

Example

import numpy as np
from aie.iron import ScratchpadParameter, Worker, Runtime, Program

seq_len = ScratchpadParameter("seq_len", np.int32)

def core_body(p):
    v = p.read()
    ...

worker = Worker(core_body, [seq_len])

rt = Runtime()
with rt.sequence(output_type) as out:
    # The compiler automatically inserts the parameter-sync preamble.
    ...

Create a ScratchpadParameter.

Parameters:

Name	Type	Description	Default
name	str	Symbol name for the parameter (must be unique within the device).	required
dtype	NpuDType	The numpy scalar type (e.g. np.int32, np.int16, bfloat16). np.float32 is not supported -- the scratchpad encoding zeroes the top 2 bits of the value, which clobbers the sign and top exponent bits of an f32.	required

Source code in python/iron/scratchpad_parameter.py

def __init__(self, name: str, dtype: NpuDType):
    """Create a ScratchpadParameter.

    Args:
        name: Symbol name for the parameter (must be unique within the
              device).
        dtype: The numpy scalar type (e.g. ``np.int32``, ``np.int16``,
               ``bfloat16``).  ``np.float32`` is not supported -- the
               scratchpad encoding zeroes the top 2 bits of the value,
               which clobbers the sign and top exponent bits of an f32.
    """
    self._name = name
    self._dtype = dtype
    self._resolved = False

name property

name: str

The symbol name of this parameter.

dtype property

dtype: NpuDType

The numpy scalar type of this parameter.

read

read() -> Value

Emit aiex.read_scratchpad_parameter inside a core body.

Must be called within an active MLIR insertion point (i.e. inside a Worker's core_fn).

Returns:

Type	Description
Value	An MLIR SSA value of the parameter's type.

Source code in python/iron/scratchpad_parameter.py

def read(self) -> "ir.Value":
    """Emit `aiex.read_scratchpad_parameter` inside a core body.

    Must be called within an active MLIR insertion point (i.e. inside a
    Worker's `core_fn`).

    Returns:
        An MLIR SSA value of the parameter's type.
    """
    mlir_type = np_dtype_to_mlir_type(self._dtype)
    return aiex.read_scratchpad_parameter(self._name, mlir_type)

resolve

resolve(
    loc: Location | None = None,
    ip: InsertionPoint | None = None,
) -> None

Emit aiex.scratchpad_parameter @name : type at module scope.

Source code in python/iron/scratchpad_parameter.py

def resolve(
    self,
    loc: ir.Location | None = None,
    ip: ir.InsertionPoint | None = None,
) -> None:
    """Emit ``aiex.scratchpad_parameter @name : type`` at module scope."""
    if not self._resolved:
        mlir_type = np_dtype_to_mlir_type(self._dtype)
        aiex.scratchpad_parameter(  # pyright: ignore[reportAttributeAccessIssue]
            self._name, mlir_type, loc=loc, ip=ip
        )
        self._resolved = True

---

Compile-time & JIT

Decorators and markers for JIT-compiling a design and injecting compile-time constants. These are re-exported into iron from aie.utils.

Note

The JIT entry point and tensor factories below are thin re-exports from the compiled aie.utils package. Their full signatures and source are available in the running package; the summaries here describe the public contract.

Symbol	Kind	Summary
iron.jit	decorator	JIT-compile a design and run it on the attached NPU (Triton-style). The first call compiles to an xclbin + instruction stream; later calls hit a cache.
iron.CompilableDesign	class	Bundle a design generator with its compile-time configuration.
iron.CallableDesign	class	A compiled, callable design produced from a CompilableDesign.
iron.compileconfig	decorator	Attach compile-time configuration to a design generator.
iron.get_compile_arg	function	Dynamically inject a compile-time argument (advanced).
iron.In / iron.Out / iron.InOut	markers	Type-annotation markers for design inputs/outputs.
iron.CompileTime	marker	Type-annotation marker for a compile-time constant argument.

See the Programming Guide for worked examples of @iron.jit.

---

Tensor factories

NumPy-like helpers that allocate NPU-accessible host tensors. Re-exported into iron from aie.utils.

Symbol	Summary
iron.tensor	Wrap existing data as an NPU-accessible tensor.
iron.arange	NPU-accessible analogue of numpy.arange.
iron.zeros / iron.ones / iron.full	Allocate a tensor filled with 0, 1, or a constant.
iron.zeros_like	Allocate a zero tensor matching another's shape/dtype.
iron.rand / iron.randint	Allocate a tensor of random floats / integers.

Device management

Symbol	Summary
iron.get_current_device	Return the currently selected NPU device.
iron.set_current_device	Select the NPU device for subsequent allocations.
iron.ensure_current_device	Raise if no device is currently selected.

---

Data type helpers

Utilities for converting between short string names and numpy dtype objects.

str_to_dtype

str_to_dtype(dtype_str: str) -> type

Convert a string representation of a data type to its corresponding dtype object.

Parameters:

Name	Type	Description	Default
dtype_str	str	The string representation of the data type.	required

Returns:

Type	Description
type	The corresponding numpy/ml_dtypes type object.

Source code in python/iron/dtype.py

def str_to_dtype(dtype_str: str) -> type:
    """
    Convert a string representation of a data type to its corresponding dtype object.

    Args:
        dtype_str: The string representation of the data type.

    Returns:
        The corresponding numpy/ml_dtypes type object.
    """

    value = None
    try:
        value = dtype_map[dtype_str]
    except KeyError:
        raise ValueError(f"Unrecognized dtype: {dtype_str}")
    return value

dtype_to_str

dtype_to_str(dtype: type) -> str

Convert a dtype object to its string representation.

Parameters:

Name	Type	Description	Default
dtype	type	The dtype object to convert.	required

Returns:

Type	Description
str	The string representation of the dtype.

Source code in python/iron/dtype.py

def dtype_to_str(dtype: type) -> str:
    """
    Convert a dtype object to its string representation.

    Args:
        dtype: The dtype object to convert.

    Returns:
        The string representation of the dtype.
    """

    for key, value in dtype_map.items():
        if value == dtype:
            return key
    raise ValueError(f"Unrecognized dtype: {dtype}")

---

Advanced primitives

Still part of the high-level aie.iron API — every object here is resolvable — but reach for these only when the managed ObjectFifo abstraction is not enough and you need explicit control over routing, DMA descriptors, and locks.

Flow / PacketFlow

Circuit-switched (Flow) and packet-switched (PacketFlow) stream connections, plus the PacketDest endpoint descriptor.

IRON-level circuit- and packet-switched route primitives.

Two classes live here: Flow (circuit-switched) and PacketFlow (packet-switched, with explicit packet IDs), plus the small PacketDest dataclass PacketFlow uses for its destination list. They share a private _emit_shim_dma_alloc helper and are treated as a sibling pair by dataflow/__init__.py; splitting them across two modules would either duplicate the helper or require a third file to hold it.

Both are peers of ObjectFifo in the dataflow namespace. ObjectFifo wraps route + buffers + locks + DMA into one circular-buffer abstraction; Flow / PacketFlow are the lower-level "just declare the route" primitives, paired with explicit TileDma programs (and Buffer / Lock shared state) for designs that need direct control.

Flow

Flow(
    src: Tile,
    dst: Tile,
    *,
    src_port: WireBundle = DMA,
    src_channel: int = 0,
    dst_port: WireBundle = DMA,
    dst_channel: int = 0,
    shim_symbol: str | None = None
)

Bases: Resolvable

An explicit AXI-stream route between (src_tile, src_port, src_channel) and (dst_tile, dst_port, dst_channel).

Lowers to a single aie.flow op. The user is responsible for arranging matching TileDma channels on the producer and consumer ends.

Construct a Flow.

Parameters:

Name	Type	Description	Default
src	Tile	The source tile.	required
dst	Tile	The destination tile.	required
src_port	WireBundle	The source port bundle. Defaults to DMA.	DMA
src_channel	int	The source channel. Defaults to 0.	0
dst_port	WireBundle	The destination port bundle. Defaults to DMA.	DMA
dst_channel	int	The destination channel. Defaults to 0.	0
shim_symbol	str | None	When this Flow has a shim endpoint that will be driven from the runtime sequence (via shim_dma_single_bd_task("symbol", ...)), provide the symbol name here and the Flow will emit a matching aie.shim_dma_allocation at the device level. Direction is inferred: shim-as-source → MM2S, shim-as-dest → S2MM.	None

Source code in python/iron/dataflow/flow.py

def __init__(
    self,
    src: Tile,
    dst: Tile,
    *,
    src_port: WireBundle = WireBundle.DMA,
    src_channel: int = 0,
    dst_port: WireBundle = WireBundle.DMA,
    dst_channel: int = 0,
    shim_symbol: str | None = None,
):
    """Construct a Flow.

    Args:
        src (Tile): The source tile.
        dst (Tile): The destination tile.
        src_port (WireBundle): The source port bundle.  Defaults to DMA.
        src_channel (int): The source channel.  Defaults to 0.
        dst_port (WireBundle): The destination port bundle.  Defaults to DMA.
        dst_channel (int): The destination channel.  Defaults to 0.
        shim_symbol (str | None): When this Flow has a shim endpoint that
            will be driven from the runtime sequence (via
            ``shim_dma_single_bd_task("symbol", ...)``), provide the
            symbol name here and the Flow will emit a matching
            ``aie.shim_dma_allocation`` at the device level.  Direction
            is inferred: shim-as-source → MM2S, shim-as-dest → S2MM.
    """
    self._src = src
    self._dst = dst
    self._src_port = src_port
    self._src_channel = src_channel
    self._dst_port = dst_port
    self._dst_channel = dst_channel
    self._shim_symbol = shim_symbol
    self._op = None

all_tiles

all_tiles()

The tiles this Flow touches — Program uses this to resolve them.

Source code in python/iron/dataflow/flow.py

def all_tiles(self):
    """The tiles this Flow touches — Program uses this to resolve them."""
    return [self._src, self._dst]

PacketDest dataclass

PacketDest(
    tile: Tile, port: WireBundle = DMA, channel: int = 0
)

One destination endpoint of a PacketFlow. Held as a small dataclass so the PacketFlow constructor's destination list reads cleanly when there are multiple sinks (uncommon, but the underlying op supports it).

PacketFlow

PacketFlow(
    pkt_id: int,
    src: Tile,
    dst: Tile,
    *,
    src_port: WireBundle = DMA,
    src_channel: int = 0,
    dst_port: WireBundle = DMA,
    dst_channel: int = 0,
    extra_dsts: Sequence[PacketDest] = (),
    keep_pkt_header: bool = False,
    shim_symbol: str | None = None
)

Bases: Resolvable

An explicit packet-switched route with caller-controlled pkt_id.

Peer of Flow for the packet-switched case. Unlike the --packet-sw-objFifos global lowering (which auto-assigns sequential packet IDs to every ObjectFifo in the design), PacketFlow exposes the packet ID directly so the same ID can be reused across stages and used as a routing decision (e.g. memtile dispatch by pkt_id to one of several compute cores).

Lowers to a single aie.packetflow op containing one aie.packet_source and one or more aie.packet_dest ops in its region.

Construct a PacketFlow.

Parameters:

Name	Type	Description	Default
pkt_id	int	The packet ID — the same byte the routing fabric uses to dispatch. Caller controls the value (often reused across stages so a memtile can re-emit packets keeping the original ID for downstream routing).	required
src	Tile	Source tile.	required
dst	Tile	Primary destination tile.	required
src_port	WireBundle	Source port bundle (as for Flow).	DMA
src_channel	int	Source channel (as for Flow).	0
dst_port	WireBundle	Destination port bundle (as for Flow).	DMA
dst_channel	int	Destination channel (as for Flow).	0
extra_dsts	Sequence[PacketDest]	Additional destination endpoints if this packet needs to fan out. Each is a PacketDest.	()
keep_pkt_header	bool	If True, downstream tile receives the 4-byte packet header alongside the payload (useful when the receiver needs to re-emit with the same pkt_id). Defaults to False.	False
shim_symbol	str | None	Same meaning as on Flow — auto-emit a matching aie.shim_dma_allocation when one endpoint is a shim tile.	None

Source code in python/iron/dataflow/flow.py

def __init__(
    self,
    pkt_id: int,
    src: Tile,
    dst: Tile,
    *,
    src_port: WireBundle = WireBundle.DMA,
    src_channel: int = 0,
    dst_port: WireBundle = WireBundle.DMA,
    dst_channel: int = 0,
    extra_dsts: Sequence[PacketDest] = (),
    keep_pkt_header: bool = False,
    shim_symbol: str | None = None,
):
    """Construct a PacketFlow.

    Args:
        pkt_id: The packet ID — the same byte the routing fabric uses to
            dispatch.  Caller controls the value (often reused across
            stages so a memtile can re-emit packets keeping the original
            ID for downstream routing).
        src: Source tile.
        dst: Primary destination tile.
        src_port: Source port bundle (as for [`Flow`][iron.Flow]).
        src_channel: Source channel (as for [`Flow`][iron.Flow]).
        dst_port: Destination port bundle (as for [`Flow`][iron.Flow]).
        dst_channel: Destination channel (as for [`Flow`][iron.Flow]).
        extra_dsts: Additional destination endpoints if this packet needs
            to fan out. Each is a [`PacketDest`][iron.PacketDest].
        keep_pkt_header: If `True`, downstream tile receives the 4-byte
            packet header alongside the payload (useful when the receiver
            needs to re-emit with the same pkt_id). Defaults to `False`.
        shim_symbol: Same meaning as on [`Flow`][iron.Flow] — auto-emit a
            matching `aie.shim_dma_allocation` when one endpoint is a
            shim tile.
    """
    self._pkt_id = pkt_id
    self._src = src
    self._dst = dst
    self._src_port = src_port
    self._src_channel = src_channel
    self._dst_port = dst_port
    self._dst_channel = dst_channel
    self._extra_dsts: list[PacketDest] = list(extra_dsts)
    self._keep_pkt_header = keep_pkt_header
    self._shim_symbol = shim_symbol
    self._op = None

CascadeFlow

Directed cascade-stream connection between two adjacent Workers.

CascadeFlow: a directed cascade stream connection between two Workers.

CascadeFlow

CascadeFlow(src: 'Worker', dst: 'Worker')

Bases: Resolvable

A directed cascade stream connection from one Worker to another.

Construct one of these for each cascade edge in your design::

CascadeFlow(producer_worker, consumer_worker)

Lowers to aie.cascade_flow(producer.tile, consumer.tile) after both Workers are placed. The kernel functions are responsible for using the put_mcd / get_scd intrinsics to actually drive/read the cascade stream — this object only declares the directed topology edge.

Hardware constraints (enforced by the underlying op verifier):

• Source and destination tiles must be cardinal-adjacent.
• Each compute tile has at most one cascade input (from N or W) and one cascade output (to S or E). Multiple cascade outputs from the same tile will fail at lowering, not at construction.
• ShimTiles and MemTiles do not have cascade interfaces.

Discovery: each newly-constructed CascadeFlow registers itself on its source Worker's _outgoing_cascades list. Program.resolve() walks the runtime's workers and resolves each worker's outgoing cascades after placement — no global registry, no drain step.

Construct a CascadeFlow.

Parameters:

Name	Type	Description	Default
src	'Worker'	Source Worker whose tile drives the cascade stream.	required
dst	'Worker'	Destination Worker whose tile reads the cascade stream.	required

Source code in python/iron/dataflow/cascadeflow.py

def __init__(self, src: "Worker", dst: "Worker"):
    """Construct a CascadeFlow.

    Args:
        src: Source `Worker` whose tile drives the cascade stream.
        dst: Destination `Worker` whose tile reads the cascade stream.
    """
    self._src = src
    self._dst = dst
    # Self-register on the source Worker so Program.resolve() can find
    # us by walking its workers (the same walk it already does).
    src._outgoing_cascades.append(self)

resolve

resolve(loc=None, ip=None) -> None

Emit aie.cascade_flow(src.tile, dst.tile).

Source code in python/iron/dataflow/cascadeflow.py

def resolve(self, loc=None, ip=None) -> None:
    """Emit ``aie.cascade_flow(src.tile, dst.tile)``."""
    _cascade_flow_op(self._src.tile.op, self._dst.tile.op)

TileDma / DmaChannel / Bd

Explicit tile DMA programs: TileDma, DmaChannel, buffer descriptors (Bd), and the Acquire / Release lock actions.

IRON-level explicit per-tile DMA program.

Peer of Worker (which describes the compute body of a tile). A TileDma describes the DMA engine program for the same (or a different) tile — what each hardware DMA channel does, which buffers it reads/writes, and how it synchronizes with the compute side via locks.

Used together with Flow / PacketFlow (which describe the AXI-stream routes) and explicit Buffer + Lock declarations, for designs where ObjectFifo would hide too much to be useful.

Acquire dataclass

Acquire(
    lock: Lock, value: int = 1, greater_equal: bool = True
)

An aie.use_lock(..., AcquireGreaterEqual|Acquire) op at the start of a BD.

Release dataclass

Release(lock: Lock, value: int = 1)

An aie.use_lock(..., Release) op at the end of a BD.

Bd dataclass

Bd(
    buffer: Buffer,
    offset: int = 0,
    length: int | None = None,
    acquires: list[Acquire] = list(),
    releases: list[Release] = list(),
    next: int | str | None = "self",
    packet: tuple[int, int] | None = None,
    sizes: list = list(),
    strides: list = list(),
)

A single buffer-descriptor entry in a DmaChannel's chain.

Lowers to one basic block containing acquires + aie.dma_bd + releases + an aie.next_bd. The next field selects what the next_bd points at:

• "self" (default) — the BD loops to itself (the common "keep streaming" pattern).
• an int i — point at the i-th BD in this channel's bds list (zero-based). Useful for explicit cycles in a multi-BD chain.
• None — emit no next_bd (rarely useful; this leaves the basic block without a terminator).

DmaChannel dataclass

DmaChannel(
    direction: DMAChannelDir, channel: int, bds: list[Bd]
)

One hardware DMA channel on a tile, with its BD chain.

Parameters:

Name	Type	Description	Default
direction	DMAChannelDir	DMAChannelDir.S2MM (host→tile) or DMAChannelDir.MM2S (tile→host).	required
channel	int	hardware channel index.	required
bds	list[Bd]	ordered list of Bd entries that form the chain.	required

TileDma

TileDma(tile: Tile, channels: Iterable[DmaChannel])

Bases: Resolvable

Per-tile DMA program — lowers to an aie.mem (compute tile), aie.memtile_dma (memtile), or aie.shim_dma (shim tile) region based on the tile's type.

Parameters:

Name	Type	Description	Default
tile	Tile	the tile whose DMA hardware this program targets.	required
channels	Iterable[DmaChannel]	ordered list of DmaChannel entries.	required

Source code in python/iron/dataflow/tile_dma.py

def __init__(self, tile: Tile, channels: Iterable[DmaChannel]):
    self._tile = tile
    self._channels: list[DmaChannel] = list(channels)
    self._resolved = False

all_buffers_and_locks

all_buffers_and_locks()

Iterate every Buffer + Lock this program touches — Program uses this to make sure they're all resolved before us.

Source code in python/iron/dataflow/tile_dma.py

def all_buffers_and_locks(self):
    """Iterate every Buffer + Lock this program touches — Program uses
    this to make sure they're all resolved before us."""
    seen_buffers: list[Buffer] = []
    seen_locks: list[Lock] = []
    for ch in self._channels:
        for bd in ch.bds:
            if bd.buffer not in seen_buffers:
                seen_buffers.append(bd.buffer)
            for use in (*bd.acquires, *bd.releases):
                if use.lock not in seen_locks:
                    seen_locks.append(use.lock)
    return seen_buffers, seen_locks

Lock

IRON-level Lock primitive — a named aie.lock on a specific tile.

Pairs with Buffer for designs that wire DMA / compute synchronization explicitly (via TileDma and Flow) instead of letting ObjectFifo manage it.

Lock

Lock(
    tile: Tile,
    lock_id: int | None = None,
    init: int = 0,
    name: str | None = None,
)

Bases: Resolvable

A named hardware lock on a specific tile.

Construct a Lock.

Parameters:

Name	Type	Description	Default
tile	Tile	The tile that owns this lock.	required
lock_id	int | None	Hardware lock ID; passed straight through to the underlying aie.lock op. If None (the default), the lowering pass picks one.	None
init	int	Initial lock value at design startup. Defaults to 0.	0
name	str | None	Symbol name for the lock. A unique name is generated if not provided.	None

Source code in python/iron/lock.py

def __init__(
    self,
    tile: Tile,
    lock_id: int | None = None,
    init: int = 0,
    name: str | None = None,
):
    """Construct a Lock.

    Args:
        tile (Tile): The tile that owns this lock.
        lock_id (int | None): Hardware lock ID; passed straight through to
            the underlying `aie.lock` op. If `None` (the default),
            the lowering pass picks one.
        init (int): Initial lock value at design startup. Defaults to 0.
        name (str | None): Symbol name for the lock. A unique name is
            generated if not provided.
    """
    self._tile = tile
    self._lock_id = lock_id
    self._init = init
    self._name = name or f"lock_{next(Lock._glock_index)}"
    self._op = None

acquire

acquire(value: int = 1) -> None

Emit aie.use_lock(self, AcquireGreaterEqual, value=value).

The default AcquireGreaterEqual mode matches what almost every ObjectFifo / DMA-driven design wants; use acquire_exact for the rarer Acquire (exact-equality) mode.

Source code in python/iron/lock.py

def acquire(self, value: int = 1) -> None:
    """Emit `aie.use_lock(self, AcquireGreaterEqual, value=value)`.

    The default `AcquireGreaterEqual` mode matches what almost every
    ObjectFifo / DMA-driven design wants; use
    [`acquire_exact`][iron.lock.Lock.acquire_exact] for the rarer
    `Acquire` (exact-equality) mode.
    """
    _use_lock(self.op, LockAction.AcquireGreaterEqual, value=value)

acquire_exact

acquire_exact(value: int = 1) -> None

Emit aie.use_lock(self, Acquire, value=value) (exact match).

Source code in python/iron/lock.py

def acquire_exact(self, value: int = 1) -> None:
    """Emit `aie.use_lock(self, Acquire, value=value)` (exact match)."""
    _use_lock(self.op, LockAction.Acquire, value=value)

release

release(value: int = 1) -> None

Emit aie.use_lock(self, Release, value=value).

Source code in python/iron/lock.py

def release(self, value: int = 1) -> None:
    """Emit `aie.use_lock(self, Release, value=value)`."""
    _use_lock(self.op, LockAction.Release, value=value)

Runtime tasks

Lower-level runtime task types scheduled by the Runtime.

RuntimeTask

RuntimeTask(task_group: RuntimeTaskGroup | None = None)

Bases: Resolvable

A RuntimeTask is a task to be performed during runtime. A task may be synchronous or asynchronous.

Construct a RuntimeTask. It may be associated with a RuntimeTaskGroup.

Parameters:

Name	Type	Description	Default
task_group	RuntimeTaskGroup | None	The TaskGroup associated with this task. Defaults to None.	None

Source code in python/iron/runtime/task.py

def __init__(self, task_group: RuntimeTaskGroup | None = None):
    """Construct a RuntimeTask. It may be associated with a RuntimeTaskGroup.

    Args:
        task_group (RuntimeTaskGroup | None, optional): The TaskGroup associated with this task. Defaults to None.
    """
    self._task_group = task_group

task_group property

task_group: RuntimeTaskGroup | None

The RuntimeTaskGroup associated with this task.

FinishTaskGroupTask

FinishTaskGroupTask(task_group: RuntimeTaskGroup)

Bases: RuntimeTask

A Task indicating the task_group should be finished(), which generally means it's tasks should be waited on and freed.

Source code in python/iron/runtime/task.py

def __init__(self, task_group: RuntimeTaskGroup):
    RuntimeTask.__init__(self, task_group)

RuntimeStartTask

RuntimeStartTask(
    worker: Worker,
    task_group: RuntimeTaskGroup | None = None,
)

Bases: RuntimeTask

A StartTask is a placeholder to indicated that a Worker should be started

Source code in python/iron/runtime/task.py

def __init__(self, worker: Worker, task_group: RuntimeTaskGroup | None = None):
    self._worker = worker
    RuntimeTask.__init__(self, task_group)

worker property

worker: Worker

The worker associated with this RuntimeStartTask

InlineOpRuntimeTask

InlineOpRuntimeTask(
    fn: Callable,
    args: list,
    task_group: RuntimeTaskGroup | None = None,
)

Bases: RuntimeTask

An InlineOpRuntimeTask is a way of submitting arbitrary operations to a runtime that are defined in a lower-level style of IRON. This can be especially useful for tracing.

Construct an InlineOpRuntimeTask.

Parameters:

Name	Type	Description	Default
fn	Callable	The function that will generate ops. It will be run within an MLIR module context.	required
args	list	The arguments for the task: this should included objects such as Buffers used by the function.	required
task_group	RuntimeTaskGroup | None	The TaskGroup to associated these operation with. Defaults to None.	None

Source code in python/iron/runtime/task.py

def __init__(
    self, fn: Callable, args: list, task_group: RuntimeTaskGroup | None = None
):
    """Construct an InlineOpRuntimeTask.

    Args:
        fn (Callable): The function that will generate ops. It will be run within an MLIR module context.
        args (list): The arguments for the task: this should included objects such as Buffers used by the function.
        task_group (RuntimeTaskGroup | None, optional): The TaskGroup to associated these operation with. Defaults to None.
    """
    # TODO: should validate somehow?
    self._fn = fn
    self._args = args
    RuntimeTask.__init__(self, task_group)

RuntimeTaskGroup: a tag for grouping related RuntimeTasks for concurrent execution.

RuntimeTaskGroup

RuntimeTaskGroup(id: int)

A RuntimeTaskGroup is a structured tag to indicate groupings of RuntimeTasks.

Construct a RuntimeTaskGroup

Parameters:

Name	Type	Description	Default
id	int	The id of the task group. The id should be unique to tasks groups within a Runtime.	required

Source code in python/iron/runtime/taskgroup.py

def __init__(self, id: int):
    """Construct a RuntimeTaskGroup

    Args:
        id (int): The id of the task group. The id should be unique to tasks groups within a Runtime.
    """
    self._group_id = id

group_id property

group_id: int

The id of the task group.

DMATask: a RuntimeTask that generates a shim DMA transfer operation.

DMATask

DMATask(
    object_fifo: ObjectFifoHandle,
    rt_data: RuntimeData,
    tap: TensorAccessPattern,
    task_group: RuntimeTaskGroup | None = None,
    wait: bool = False,
    offset_parameter: str | None = None,
    packet: tuple[int, int] | None = None,
)

Bases: RuntimeTask

A RuntimeTask that will resolve to a DMA Operation.

Parameters:

Name	Type	Description	Default
object_fifo	ObjectFifoHandle	The ObjectFifoHandle associated with the operation	required
rt_data	RuntimeData	The Runtime buffer associated with the operation.	required
tap	TensorAccessPattern	The access pattern associated with the operation.	required
task_group	RuntimeTaskGroup | None	The task group associated with the operation. Defaults to None.	None
wait	bool	Whether this task should conclude with a call to await or a call to free. Defaults to False.	False
offset_parameter	str | None	Name of a ScratchpadParameter whose value is used as the element offset for this DMA transfer. Defaults to None.	None
packet	tuple[int, int] | None	Stamp the shim DMA's BD with a packet header (pkt_type, pkt_id). Pairs with downstream packet-switched routing (e.g. ObjectFifos lowered with --packet-sw-objFifos or an explicit PacketFlow). Defaults to None.	None

Source code in python/iron/runtime/dmatask.py

def __init__(
    self,
    object_fifo: ObjectFifoHandle,
    rt_data: RuntimeData,
    tap: TensorAccessPattern,
    task_group: RuntimeTaskGroup | None = None,
    wait: bool = False,
    offset_parameter: str | None = None,
    packet: tuple[int, int] | None = None,
):
    """A RuntimeTask that will resolve to a DMA Operation.

    Args:
        object_fifo (ObjectFifoHandle): The ObjectFifoHandle associated with the operation
        rt_data (RuntimeData): The Runtime buffer associated with the operation.
        tap (TensorAccessPattern): The access pattern associated with the operation.
        task_group (RuntimeTaskGroup | None, optional): The task group associated with the operation. Defaults to None.
        wait (bool, optional): Whether this task should conclude with a call to await or a call to free. Defaults to False.
        offset_parameter (str | None, optional): Name of a ScratchpadParameter whose value is used as the element offset for this DMA transfer. Defaults to None.
        packet (tuple[int, int] | None, optional): Stamp the shim DMA's
            BD with a packet header `(pkt_type, pkt_id)`. Pairs with
            downstream packet-switched routing (e.g. ObjectFifos
            lowered with `--packet-sw-objFifos` or an explicit
            [`PacketFlow`][iron.PacketFlow]). Defaults to None.
    """
    self._object_fifo = object_fifo
    self._rt_data = rt_data
    self._tap = tap
    self._wait = wait
    self._offset_parameter = offset_parameter
    self._packet = packet
    self._task = None
    RuntimeTask.__init__(self, task_group)

fifo property

fifo: ObjectFifoHandle

The ObjectFifoHandle associated with this task.

task property

task

The MLIR op generated by resolve(). This handle is used after resolution for await or free operations as dictated by a RuntimeTaskGroup Finish operation

will_wait

will_wait() -> bool

Whether this task should conclude with an await operation.

Source code in python/iron/runtime/dmatask.py

def will_wait(self) -> bool:
    """Whether this task should conclude with an await operation."""
    return self._wait