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, workers: list | None = None
)

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
workers	list[Worker] | None	The Workers to run on the device. Defaults to None (no workers). Workers are passed here explicitly rather than started from within the runtime sequence.	None

Source code in python/iron/program.py

def __init__(
    self,
    device: Device,
    rt: Runtime,
    workers: "list | None" = None,
):
    """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.
        workers (list[Worker] | None, optional): The Workers to run on the
            device. Defaults to None (no workers). Workers are passed here
            explicitly rather than started from within the runtime sequence.
    """
    self._device = device
    self._rt = rt
    self._workers = list(workers) if workers is not None else []
    self._trace_size = None
    self._trace_workers = None
    self._reuse_output_buffer = False
    self._egress_shim_col = 0
    self._coretile_events = None
    self._coremem_events = None
    self._memtile_events = None
    self._shimtile_events = None

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. Lives on Program (not Runtime) because it configures both the traced Workers' tiles and the Runtime's trace-buffer sequencing.

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/program.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.
    Lives on Program (not Runtime) because it configures both the traced
    Workers' tiles and the Runtime's trace-buffer sequencing.

    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

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 known up front (Worker fn_args) at module
        # scope now. 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._workers:
            for arg in w.flat_fn_args:
                if isinstance(arg, ScratchpadParameter):
                    arg.resolve()

        @device(self._device.resolve(), sym_name=device_name)
        def device_body():
            # Collect all fifos. Runtime-driven fifos already have their shim
            # endpoints bound (Runtime registered its fn_args at construction),
            # so they resolve here with both ends known -- the sequence body
            # itself is emitted LAST (self._rt.resolve() below), after workers,
            # so body verbs that read worker-side state (barrier locks, worker
            # Buffer placement) see it resolved.
            all_fifos = set()
            all_fifos.update(self._rt.fifos)
            for w in self._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._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.flat_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._workers:
                for arg in w.flat_fn_args:
                    if isinstance(arg, FuncBase):
                        arg.emit()
                    elif isinstance(arg, Resolvable):
                        arg.resolve()

            # Generate core programs
            for w in self._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._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._trace_workers is not None:
                for w in self._trace_workers:
                    tiles_to_trace.append(w.tile.op)
            else:
                for w in self._workers:
                    if w.trace is not None:
                        tiles_to_trace.append(w.tile.op)
            if self._trace_size is not None and self._trace_size > 0:
                trace_utils.configure_trace(
                    tiles_to_trace,
                    coretile_events=self._coretile_events,
                    coremem_events=self._coremem_events,
                    memtile_events=self._memtile_events,
                    shimtile_events=self._shimtile_events,
                )

            # Emit the runtime sequence body LAST: workers, their locks, and
            # worker Buffers are now resolved, so body verbs that read that
            # state (barrier.set, inline_ops over a worker Buffer) are valid.
            # Its shim DMAs reference fifos by symbol name (forward ref), so
            # emitting after the fifo ops is fine.
            #
            # On the full-ELF path the runtime sequence must load its own
            # PDI (no xclbin configures the device), so pass the device
            # symbol as the load_pdi reference. The flag is injected into
            # the compile context by CompilableDesign.
            load_pdi_device_ref = (
                device_name if get_compile_arg("_iron_full_elf") else None
            )
            self._rt.resolve(
                trace_size=self._trace_size,
                reuse_output_buffer=self._reuse_output_buffer,
                egress_shim_col=self._egress_shim_col,
                load_pdi_device_ref=load_pdi_device_ref,
            )

        # Resolve parameters only discoverable once the sequence body has
        # traced (offset_parameter= passed directly to fill()/drain(),
        # rather than declared up front via Worker fn_args). device_body's
        # own insertion point is scoped to its @device region, so by now
        # the ambient insertion point is back to module scope -- the same
        # place the fn_args-declared parameters above were resolved.
        for p in self._rt._scratchpad_parameters:
            p.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.
    # fn_args may nest lists (e.g. one fifo per column); iterate the flattened
    # leaves for registration while the core_fn still receives the structure.
    for arg in flatten_fn_args(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.

flat_fn_args property

flat_fn_args: list

fn_args with any nested lists/tuples flattened to their leaves.

Use this (not fn_args) when iterating to register/resolve individual arguments; fn_args keeps its structure for the core_fn call.

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)

set

set(value: int)

Set the barrier to value from within a runtime sequence body.

Parameters:

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

Source code in python/iron/worker.py

def set(self, value: int):
    """Set the barrier to ``value`` from within a runtime sequence body.

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

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,
    tile: Tile | 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
tile	Tile | None	When this handle drives an ObjectFifo from the runtime (passed in Runtime fn_args), the shim tile its host-side DMA binds to. Defaults to None (any available shim tile).	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,
    tile: Tile | 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).
        tile (Tile | None, optional): When this handle drives an ObjectFifo
            from the runtime (passed in ``Runtime`` ``fn_args``), the shim tile
            its host-side DMA binds to. Defaults to None (any available shim tile).

    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}."
            )
        if tile is not None and not _same_shim_pin(self._prod._shim_tile, tile):
            raise ValueError(
                f"Producer handle for {self.name} already pinned to shim tile "
                f"{self._prod._shim_tile}, cannot re-pin to {tile}."
            )
    else:
        self._prod = ObjectFifoHandle(self, True, depth, channel=channel, tile=tile)
    return self._prod

cons

cons(
    depth: int | None = None,
    dims_from_stream: StreamDims | None = None,
    channel: int | None = None,
    tile: Tile | 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
tile	Tile | None	When this handle drains an ObjectFifo to the runtime (passed in Runtime fn_args), the shim tile its host-side DMA binds to. Defaults to None (any available shim tile).	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,
    tile: Tile | 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).
        tile (Tile | None, optional): When this handle drains an ObjectFifo to
            the runtime (passed in ``Runtime`` ``fn_args``), the shim tile its
            host-side DMA binds to. Defaults to None (any available shim tile).

    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,
            tile=tile,
        )
    )
    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,
    tile: Tile | 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
tile	Tile | None	Shim tile for a runtime-driven endpoint (see prod()/cons()). Defaults to None.	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,
    tile: Tile | 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).
        tile (Tile | None, optional): Shim tile for a runtime-driven endpoint (see prod()/cons()). Defaults to None.

    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._shim_tile = tile
    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)

fill

fill(
    source,
    tap=None,
    wait: bool = False,
    packet: tuple[int, int] | None = None,
    offset_parameter=None,
    group=None,
    sizes=None,
    strides=None,
    offset=None,
    transfer_len=None,
    managed: bool = True,
)

Fill this producer ObjectFifo with data from the source runtime buffer.

Call from within a Runtime sequence body on a producer handle. See _emit_transfer for the shared arguments; returns a Task handle to the transfer.

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

def fill(
    self,
    source,
    tap=None,
    wait: bool = False,
    packet: tuple[int, int] | None = None,
    offset_parameter=None,
    group=None,
    sizes=None,
    strides=None,
    offset=None,
    transfer_len=None,
    managed: bool = True,
):
    """Fill this producer ObjectFifo with data from the ``source`` runtime buffer.

    Call from within a [`Runtime`][iron.Runtime] sequence body on a producer
    handle. See ``_emit_transfer`` for the shared arguments; returns a
    [`Task`][iron.runtime.dmataskhandle.Task] handle to the transfer.
    """
    if not self._is_prod:
        raise ValueError("fill() is only valid on a producer ObjectFifoHandle")

    return self._emit_transfer(
        source,
        tap,
        wait,
        packet,
        offset_parameter,
        group,
        sizes,
        strides,
        offset,
        transfer_len,
        managed,
    )

drain

drain(
    dest,
    tap=None,
    wait: bool = False,
    packet: tuple[int, int] | None = None,
    offset_parameter=None,
    group=None,
    sizes=None,
    strides=None,
    offset=None,
    transfer_len=None,
    managed: bool = True,
)

Drain this consumer ObjectFifo, writing data to the dest runtime buffer.

Call from within a Runtime sequence body on a consumer handle. See _emit_transfer for the shared arguments; returns a Task handle to the transfer.

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

def drain(
    self,
    dest,
    tap=None,
    wait: bool = False,
    packet: tuple[int, int] | None = None,
    offset_parameter=None,
    group=None,
    sizes=None,
    strides=None,
    offset=None,
    transfer_len=None,
    managed: bool = True,
):
    """Drain this consumer ObjectFifo, writing data to the ``dest`` runtime buffer.

    Call from within a [`Runtime`][iron.Runtime] sequence body on a consumer
    handle. See ``_emit_transfer`` for the shared arguments; returns a
    [`Task`][iron.runtime.dmataskhandle.Task] handle to the transfer.
    """
    if self._is_prod:
        raise ValueError("drain() is only valid on a consumer ObjectFifoHandle")

    return self._emit_transfer(
        dest,
        tap,
        wait,
        packet,
        offset_parameter,
        group,
        sizes,
        strides,
        offset,
        transfer_len,
        managed,
    )

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. Its fill / drain operations are declared in the sequence body passed to Runtime(seq, fn_args); Workers are passed to Program(workers=...) rather than started from the body.

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

The runtime sequence is written as a callback body -- a plain Python function whose parameters are the runtime I/O buffers (and, optionally, runtime scalars). The body runs eagerly inside @runtime_sequence at resolve time, mirroring how Worker.core_fn runs inside @core. Because the body executes with live MLIR values in scope, it can use native range_/if_ control flow with fill/drain verbs nested inside -- the dynamic path lowers these to scf.for/scf.if (EmitC C++ TXN), and the static path (Python range/int bounds) elaborates to a flat binary sequence.

IronRuntimeError

Bases: Exception

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

ActiveSequence

ActiveSequence(runtime: 'Runtime')

The state of a runtime sequence body while it is being emitted.

The body's data-movement verbs (fifo.fill/fifo.drain) and TaskGroup reach this object through the active-sequence ContextVar (see _context) rather than a threaded rt reference, so the body signature carries only the runtime buffers.

The body runs exactly once, inside the runtime_sequence op: each verb both binds its ObjectFifo's shim endpoint and emits the shim DMA. The DMA references the fifo by symbol name (a legal MLIR forward reference), so it does not require the fifo to be resolved yet -- the Program resolves fifos and cores afterward, with every runtime endpoint already bound.

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

def __init__(self, runtime: "Runtime"):
    self._runtime = runtime
    # The implicit group for fill/drain calls that pass no explicit group.
    self._default_task_group = TaskGroup(next(runtime._task_group_index))
    self._open_task_groups: list[TaskGroup] = []
    self._used_default = False
    self._used_explicit = False

note_fifo

note_fifo(handle: ObjectFifoHandle) -> None

Record that handle is driven from the runtime (its shim endpoint).

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

def note_fifo(self, handle: ObjectFifoHandle) -> None:
    """Record that ``handle`` is driven from the runtime (its shim endpoint)."""
    self._runtime._fifos.add(handle)

finish_task_group

finish_task_group(tg: TaskGroup) -> None

Close a task group: await its waited tasks, then free the rest.

Waits are ordered before frees within the group, matching the hardware-safe order the old flat-list runtime used.

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

def finish_task_group(self, tg: TaskGroup) -> None:
    """Close a task group: await its waited tasks, then free the rest.

    Waits are ordered before frees within the group, matching the
    hardware-safe order the old flat-list runtime used.
    """
    if tg in self._open_task_groups:
        self._open_task_groups.remove(tg)
    actions = tg._actions
    if not actions:
        return
    wait_tasks = [(fn, a) for (fn, a) in actions if fn == dma_await_task]
    free_tasks = [(fn, a) for (fn, a) in actions if fn == dma_free_task]
    if len(wait_tasks) + len(free_tasks) != len(actions):
        unknown = [
            (fn, a)
            for (fn, a) in actions
            if fn != dma_await_task and fn != dma_free_task
        ]
        raise IronRuntimeError(
            f"Unknown action type detected: {','.join(str(a) for a in unknown)}"
        )
    for fn, a in wait_tasks + free_tasks:
        fn(*a)
    tg._actions = []

emit_transfer

emit_transfer(
    task: DMATask, task_group: TaskGroup | None
) -> None

Emit a DMA transfer and record its await/free action(s) for group close.

A waited transfer is both awaited and freed: the dynamic BD-pool pass only returns an id to the runtime free-list on an explicit free (an await is a pure TCT sync), so a waited-but-never-freed task would leak a pool slot on every rolled-loop iteration. Matches the static path's long-standing "await implies release" convention -- it just does so with an explicit free instead of folding it into the await.

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

def emit_transfer(self, task: DMATask, task_group: TaskGroup | None) -> None:
    """Emit a DMA transfer and record its await/free action(s) for group close.

    A waited transfer is both awaited and freed: the dynamic BD-pool pass
    only returns an id to the runtime free-list on an explicit free (an
    await is a pure TCT sync), so a waited-but-never-freed task would leak
    a pool slot on every rolled-loop iteration. Matches the static path's
    long-standing "await implies release" convention -- it just does so
    with an explicit free instead of folding it into the await.
    """
    task.resolve()
    if task_group is not None:
        self._used_explicit = True
        group = task_group
    else:
        self._used_default = True
        group = self._default_task_group
    if task.will_wait():
        group._actions.append((dma_await_task, [task.task]))
    group._actions.append((dma_free_task, [task.task]))

finalize

finalize() -> None

Close bookkeeping after the body runs.

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

def finalize(self) -> None:
    """Close bookkeeping after the body runs."""
    explicit_open = [
        tg for tg in self._open_task_groups if tg is not self._default_task_group
    ]
    if explicit_open:
        tgs = ", ".join(str(t) for t in explicit_open)
        raise IronRuntimeError(f"Failed to close task groups: {tgs}")
    if (
        self._runtime._strict_task_groups
        and self._used_default
        and self._used_explicit
    ):
        raise IronRuntimeError(
            "Mixing explicit task groups and the default task group is "
            "prohibited. Please assign all tasks to a task group."
        )
    # Flush any transfers left in the default group (no explicit finish).
    if self._default_task_group._actions:
        self.finish_task_group(self._default_task_group)

Runtime

Runtime(
    seq_fn: Callable,
    fn_args: "Sequence | None" = None,
    *,
    strict_task_groups: bool = True
)

Bases: Resolvable

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

A Runtime describes what the host does at runtime: filling input ObjectFifos with data and draining results back to host buffers. The sequence is the seq_fn callback passed to the constructor; its body reads the runtime buffers as parameters and moves data with fifo.fill(...) / fifo.drain(...).

Create a runtime from its sequence body and fn_args.

Mirrors Worker(core_fn, fn_args): seq_fn runs inside @runtime_sequence at resolve time, called as seq_fn(*fn_args), and can use native range_/if_ and fifo.fill/fifo.drain. Objects in fn_args (ObjectFifoHandles, Buffers, ...) are registered eagerly at construction -- fifo shim endpoints bind now (from prod(tile=)/cons(tile=)) so the Program resolves fifos and cores before the body emits, letting body verbs read resolved worker state (e.g. barrier.set).

Each fn_args entry is one of:

• a type (a tensor type or a scalar type like np.int32): declares a runtime input and is replaced with a live SSA value bound to a new runtime_sequence block arg -- a tensor type becomes a RuntimeData (fill/drain target), a scalar type becomes the bare SSA value (scf survives to the dynamic EmitC path).
• a concrete int value: also declares a runtime input, but is folded into an arith.constant instead of a block arg (constant-bound range_/if_ unrolls to the static binary path). One body thus serves both lowerings depending on whether the caller passes a type or an int here.
• any other object (ObjectFifoHandle, Buffer, Kernel, ScratchpadParameter, WorkerRuntimeBarrier, ...): passed through to the body unchanged, as with Worker.fn_args.

Parameters:

Name	Type	Description	Default
seq_fn	Callable	The sequence body; params bound to fn_args in order.	required
fn_args	Sequence | None	Types/ints (runtime inputs) and shared objects, in the order seq_fn expects them. Defaults to None (empty list).	None
strict_task_groups	bool	Disallow mixing the default and explicit task groups. Defaults to True.	True

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

def __init__(
    self,
    seq_fn: Callable,
    fn_args: "Sequence | None" = None,
    *,
    strict_task_groups: bool = True,
) -> None:
    """Create a runtime from its sequence body and fn_args.

    Mirrors [`Worker`][iron.Worker]``(core_fn, fn_args)``: ``seq_fn`` runs
    inside ``@runtime_sequence`` at resolve time, called as
    ``seq_fn(*fn_args)``, and can use native ``range_``/``if_`` and
    ``fifo.fill``/``fifo.drain``. Objects in ``fn_args`` (ObjectFifoHandles,
    Buffers, ...) are registered eagerly at construction -- fifo shim
    endpoints bind now (from ``prod(tile=)``/``cons(tile=)``) so the Program
    resolves fifos and cores before the body emits, letting body verbs read
    resolved worker state (e.g. ``barrier.set``).

    Each ``fn_args`` entry is one of:

    * a **type** (a tensor type or a scalar type like ``np.int32``): declares
      a runtime input and is replaced with a live SSA value bound to a new
      ``runtime_sequence`` block arg -- a tensor type becomes a
      ``RuntimeData`` (``fill``/``drain`` target), a scalar type becomes the
      bare SSA value (``scf`` survives to the dynamic EmitC path).
    * a concrete **int value**: also declares a runtime input, but is folded
      into an ``arith.constant`` instead of a block arg (constant-bound
      ``range_``/``if_`` unrolls to the static binary path). One body thus
      serves both lowerings depending on whether the caller passes a type or
      an int here.
    * any other object (ObjectFifoHandle, Buffer, Kernel, ScratchpadParameter,
      WorkerRuntimeBarrier, ...): passed through to the body unchanged, as
      with ``Worker.fn_args``.

    Args:
        seq_fn (Callable): The sequence body; params bound to ``fn_args`` in order.
        fn_args (Sequence | None): Types/ints (runtime inputs) and shared objects,
            in the order ``seq_fn`` expects them. Defaults to None (empty list).
        strict_task_groups (bool): Disallow mixing the default and explicit task groups. Defaults to True.
    """
    self._seq_fn: Callable = seq_fn
    self._fn_args = list(fn_args) if fn_args is not None else []
    # A concrete int entry is a folded constant; a type/generic-alias entry
    # is a runtime input; anything else passes through as an object fn_arg.
    self._const_inputs: list[int | np.integer | None] = [
        v if isinstance(v, (int, np.integer)) and not isinstance(v, bool) else None
        for v in self._fn_args
    ]
    self._rt_data: list["RuntimeData | None"] = [
        (
            RuntimeData(arg)
            if c is None
            and (isinstance(arg, type) or get_origin(arg) is np.ndarray)
            else None
        )
        for c, arg in zip(self._const_inputs, self._fn_args)
    ]
    self._fifos: set[ObjectFifoHandle] = set()
    self._register_fn_args()
    # 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._strict_task_groups = strict_task_groups
    self._task_group_index = itertools.count()

fifos property

fifos: list[ObjectFifoHandle]

The ObjectFifoHandles driven from the runtime by fill()/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)

resolve

resolve(
    loc: Location | None = None,
    ip: InsertionPoint | None = None,
    *,
    trace_size: int | None = None,
    reuse_output_buffer: bool = False,
    egress_shim_col: int = 0,
    load_pdi_device_ref: str | None = None
) -> None

Build the runtime_sequence op and run the sequence body inside it.

The body runs exactly once. Each fill/drain verb binds its ObjectFifo's shim endpoint and emits the shim DMA (referencing the fifo by symbol name, a forward reference). The Program calls this before it resolves ObjectFifos and cores, so by the time a fifo is resolved every runtime endpoint -- including those on link siblings -- is already bound.

Parameters:

Name	Type	Description	Default
trace_size	int | None	Forwarded from Program.enable_trace; see there. None/0 disables tracing.	None
reuse_output_buffer	bool	Forwarded from Program.enable_trace; see there.	False
egress_shim_col	int	Forwarded from Program.enable_trace; see there.	0
load_pdi_device_ref	str | None	On the full-ELF path (no xclbin configures the device), the device symbol to load via npu_load_pdi as the first op in the sequence. None on the xclbin path.	None

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

def resolve(
    self,
    loc: ir.Location | None = None,
    ip: ir.InsertionPoint | None = None,
    *,
    trace_size: int | None = None,
    reuse_output_buffer: bool = False,
    egress_shim_col: int = 0,
    load_pdi_device_ref: str | None = None,
) -> None:
    """Build the ``runtime_sequence`` op and run the sequence body inside it.

    The body runs exactly once. Each ``fill``/``drain`` verb binds its
    ObjectFifo's shim endpoint and emits the shim DMA (referencing the fifo
    by symbol name, a forward reference). The Program calls this before it
    resolves ObjectFifos and cores, so by the time a fifo is resolved every
    runtime endpoint -- including those on link siblings -- is already bound.

    Args:
        trace_size: Forwarded from
            [`Program.enable_trace`][iron.program.Program.enable_trace]; see
            there. ``None``/``0`` disables tracing.
        reuse_output_buffer: Forwarded from
            [`Program.enable_trace`][iron.program.Program.enable_trace]; see
            there.
        egress_shim_col: Forwarded from
            [`Program.enable_trace`][iron.program.Program.enable_trace]; see
            there.
        load_pdi_device_ref: On the full-ELF path (no xclbin configures the
            device), the device symbol to load via ``npu_load_pdi`` as the
            first op in the sequence. ``None`` on the xclbin path.
    """
    # A runtime_sequence block arg per runtime (type) input; folded-constant
    # inputs contribute no block arg.
    rt_dtypes = [
        try_convert_np_type_to_mlir_type(rt_data.arr_type)
        for rt_data in self._rt_data
        if rt_data is not None
    ]
    active = ActiveSequence(self)

    seq_op = RuntimeSequenceOp(sym_name="sequence")
    entry_block = seq_op.body.blocks.append(*rt_dtypes)
    with ir.InsertionPoint(entry_block):
        # Full-ELF designs configure the device themselves: no xclbin
        # pre-loads the PDI, so the sequence must start by loading it.
        if load_pdi_device_ref is not None:
            npu_load_pdi(device_ref=load_pdi_device_ref)

        block_args = iter(entry_block.arguments)
        for rt_data in self._rt_data:
            if rt_data is not None:
                rt_data.op = next(block_args)

        if trace_size is not None and trace_size > 0:
            trace_utils.start_trace(
                trace_size=trace_size,
                reuse_output_buffer=reuse_output_buffer,
                routing="single",
                egress_shim_col=egress_shim_col,
            )

        # Build the body's positional args, one per fn_args entry, in order:
        #   * folded-constant entry -> an arith.constant of that value;
        #   * scalar type entry     -> its live SSA value (used in arithmetic
        #                              and range_/if_ bounds);
        #   * tensor type entry     -> its RuntimeData handle (fill/drain);
        #   * anything else         -> passed through unchanged (fifos, etc).
        body_args = []
        for arg, const_val, rt_data in zip(
            self._fn_args, self._const_inputs, self._rt_data
        ):
            if const_val is not None:
                # i32 to mirror the dynamic np.int32 scalar path, so the same
                # body's arithmetic (extsi to i64, etc.) lowers identically.
                body_args.append(
                    constant(int(const_val), np_dtype_to_mlir_type(np.int32))
                )
            elif rt_data is not None:
                body_args.append(rt_data.op if rt_data.is_scalar else rt_data)
            else:
                body_args.append(arg)

        with active_sequence_scope(active):
            self._seq_fn(*body_args)
            active.finalize()

    self._dedup_runtime_consumers()

sync_parameters

sync_parameters() -> None

Emit aiex.sync_scratchpad_parameters_from_host in the sequence body.

Call after all scratchpad 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() -> None:
    """Emit ``aiex.sync_scratchpad_parameters_from_host`` in the sequence body.

    Call after all scratchpad parameters have been written on the host side and
    before starting workers that read them.
    """
    active_sequence()  # ensure we're inside a sequence body
    _SyncParametersTask().resolve()

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,
    *,
    link_with_mode: str | 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.

link_with_mode selects how that artifact is consumed: the default (None) object-links it, while "merge" asks aiecc to llvm-link it into the core's LLVM module before codegen. The mode is explicit metadata -- it is never inferred from the file suffix.

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
link_with_mode	str | None	Optional link policy emitted alongside link_with. "merge" routes the artifact through aiecc's llvm-link merge path; None (the default) object-links it.	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,
    *,
    link_with_mode: str | 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).
        link_with_mode: Optional link policy emitted alongside
            ``link_with``.  ``"merge"`` routes the artifact through aiecc's
            ``llvm-link`` merge path; None (the default) object-links it.
    """
    super().__init__(name, arg_types)
    self._object_file_name = object_file_name
    self._link_with_mode = link_with_mode

object_file_name property

object_file_name: str

Filename of the compiled object file.

link_with_mode property

link_with_mode: str | None

Link policy emitted with link_with, or None for object linking.

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,
    inline: 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 artifact name. Defaults to <effective_name>.o, or <effective_name>.ll with inline=True. With inline=True an explicit name must end in .ll (textual LLVM IR) or .bc (bitcode) -- that suffix selects the emitted format -- and is otherwise rejected.	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
inline	bool	When True, compile the kernel to alwaysinline LLVM IR (.ll) and declare it with link_with_mode = "merge" so aiecc llvm-links it into the core and inlines it, instead of object-linking a separate .o. Removes the func.call boundary and the separate object. Peano path only (the Chess/xchesscc toolchain cannot llvm-link).	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,
    inline: bool = False,
) -> None:
    """
    Args:
        name: Symbol name of the function as it will appear in the object
            file.
        object_file_name: Output artifact name. Defaults to
            ``<effective_name>.o``, or ``<effective_name>.ll`` with
            ``inline=True``. With ``inline=True`` an explicit name must end
            in ``.ll`` (textual LLVM IR) or ``.bc`` (bitcode) -- that suffix
            selects the emitted format -- and is otherwise rejected.
        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.
        inline: When True, compile the kernel to ``alwaysinline`` LLVM IR
            (``.ll``) and declare it with ``link_with_mode = "merge"`` so
            aiecc llvm-links it into the core and inlines it, instead of
            object-linking a separate ``.o``. Removes the ``func.call``
            boundary and the separate object. Peano path only (the
            Chess/xchesscc toolchain cannot llvm-link).
    """
    if inline and use_chess:
        raise ValueError(
            f"ExternalFunction '{name}': inline=True requires the Peano "
            "toolchain and cannot be combined with use_chess=True."
        )
    if inline and symbol_prefix:
        raise NotImplementedError(
            f"ExternalFunction '{name}': inline=True combined with symbol_prefix is "
            "not supported (an inline kernel is emitted as LLVM IR and cannot be "
            "symbol-renamed). Use inline without a symbol_prefix, or drop inline for "
            "this kernel."
        )

    # Must precede the collision scan below: it registers this instance in
    # `_instances`, whose hash/eq run `_content_digest()` -- which reads
    # `_inline`.  `_original_name` / `_symbol_prefix` are likewise read by
    # compile_external_kernel (source naming and the objcopy symbol rename).
    self._original_name = name
    self._symbol_prefix = symbol_prefix
    self._inline = inline
    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}.ll" if inline else f"{effective_name}.o"
        )
    elif inline and Path(object_file_name).suffix.lower() not in (".ll", ".bc"):
        # An inline kernel is emitted as LLVM IR, and the suffix picks the
        # format (textual vs bitcode), so a wrong one has no valid reading.
        # Reject it instead of silently renaming the caller's artifact --
        # aiecc routes on the `link_with_mode` attribute, not the suffix, so
        # a rename would buy nothing.  Compared case-insensitively, matching
        # compile_cxx_core_function's own suffix check; the caller's exact
        # spelling is preserved either way.
        raise ValueError(
            f"ExternalFunction '{name}': inline=True emits LLVM IR, so "
            f"object_file_name must end in '.ll' (textual LLVM IR) or "
            f"'.bc' (bitcode); got '{object_file_name}'."
        )
    super().__init__(
        effective_name,
        object_file_name,
        arg_types,
        link_with_mode="merge" if inline else None,
    )

    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 artifact 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]
            output_path = Path(object_file_name)
            object_file_name = str(
                output_path.with_name(
                    f"{output_path.stem}_{suffix}{output_path.suffix}"
                )
            )
            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])

def sequence(out):
    # The compiler automatically inserts the parameter-sync preamble.
    ...

rt = Runtime(sequence, [output_type])

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: TaskGroup | 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 TaskGroup.

Parameters:

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

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

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

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

task_group property

task_group: TaskGroup | None

The TaskGroup associated with this task.

TaskGroup: groups related runtime transfers so they are awaited/freed together.

TaskGroup

TaskGroup(id: int | None = None)

A grouping of runtime data transfers awaited and freed together.

Construct one inside a runtime sequence body and pass it as the group= argument to fifo.fill(...) / fifo.drain(...). Call finish to await the group's waited transfers and free the rest (waits are ordered before frees).

def seq(A, C):
    tg = TaskGroup()
    inA.prod().fill(A, group=tg)
    outC.cons().drain(C, wait=True, group=tg)
    tg.finish()

Construct a TaskGroup, registering it with the active runtime sequence.

Parameters:

Name	Type	Description	Default
id	int | None	Group id, unique within a Runtime. Defaults to the active sequence's next id. Passing an explicit id is only needed for the runtime's internal default group.	None

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

def __init__(self, id: int | None = None):
    """Construct a TaskGroup, registering it with the active runtime sequence.

    Args:
        id (int | None): Group id, unique within a Runtime. Defaults to the
            active sequence's next id. Passing an explicit id is only needed
            for the runtime's internal default group.
    """
    # Actions accumulated for this group: (dma_await_task | dma_free_task, [task]).
    self._actions: list = []
    # Lazy import to avoid a cycle (runtime -> taskgroup -> _context).
    from ._context import _active_sequence

    active = _active_sequence.get()
    if id is None:
        if active is None:
            raise RuntimeError(
                "TaskGroup() must be constructed within the function passed "
                "to Runtime(seq_fn, fn_args)."
            )
        id = next(active._runtime._task_group_index)
    self._group_id = id
    if active is not None and id is not None:
        active.register_task_group(self)

group_id property

group_id: int

The id of the task group.

finish

finish() -> None

Await this group's waited transfers, then free the rest.

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

def finish(self) -> None:
    """Await this group's waited transfers, then free the rest."""
    from ._context import active_sequence

    active_sequence().finish_task_group(self)

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

DMATask

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

Bases: RuntimeTask

A RuntimeTask that will resolve to a DMA Operation.

Provide the access pattern either as a static tap (TensorAccessPattern) or as explicit sizes/strides/offset/ transfer_len lists whose entries may be runtime SSA values (for the dynamic lowering). The two forms are mutually exclusive.

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 | None	The static access pattern. Mutually exclusive with sizes/strides/offset/transfer_len.	None
task_group	TaskGroup | 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
sizes	optional	Explicit access-pattern sizes whose entries may be runtime SSA values. Used instead of tap for the dynamic path.	None
strides	optional	Explicit access-pattern strides, paired with sizes. Used instead of tap for the dynamic path.	None
offset	optional	Explicit access-pattern offset. Used instead of tap for the dynamic path.	None
transfer_len	optional	Explicit access-pattern transfer length. Used instead of tap for the dynamic path.	None

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

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

    Provide the access pattern either as a static ``tap``
    (TensorAccessPattern) or as explicit ``sizes``/``strides``/``offset``/
    ``transfer_len`` lists whose entries may be runtime SSA values (for the
    dynamic lowering). The two forms are mutually exclusive.

    Args:
        object_fifo (ObjectFifoHandle): The ObjectFifoHandle associated with the operation
        rt_data (RuntimeData): The Runtime buffer associated with the operation.
        tap (TensorAccessPattern | None, optional): The static access pattern. Mutually exclusive with sizes/strides/offset/transfer_len.
        task_group (TaskGroup | 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.
        sizes (optional): Explicit access-pattern sizes whose entries may be
            runtime SSA values. Used instead of ``tap`` for the dynamic path.
        strides (optional): Explicit access-pattern strides, paired with
            ``sizes``. Used instead of ``tap`` for the dynamic path.
        offset (optional): Explicit access-pattern offset. Used instead of
            ``tap`` for the dynamic path.
        transfer_len (optional): Explicit access-pattern transfer length.
            Used instead of ``tap`` for the dynamic path.
    """
    if tap is not None and any(
        v is not None for v in (sizes, strides, offset, transfer_len)
    ):
        raise ValueError(
            "DMATask: tap and sizes/strides/offset/transfer_len are mutually "
            "exclusive access-pattern specifications; pass only one."
        )
    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._sizes = sizes
    self._strides = strides
    self._offset = offset
    self._transfer_len = transfer_len
    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 TaskGroup 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

Task: a handle to an in-flight shim DMA transfer.

Returned by fifo.fill/fifo.drain. Its handle is the transfer's !index SSA value, so a Task can be carried across scf.for iterations as a range_ iter_args entry for software-pipelined transfers -- and it carries .free() / .await_() verbs so the loop body does not need to reach for the raw aiex.dma_free_task / aiex.dma_await_task dialect ops.

A Task returned by an unmanaged transfer (managed=False) is not enrolled in a TaskGroup's automatic await/free, so the caller owns its lifetime with these verbs -- exactly what a hand-rolled ping-pong needs.

Task

Task(handle)

A handle to a submitted shim DMA transfer.

Wraps the transfer's !index SSA value (handle). Pass a Task as a range_ iter_args entry to carry an in-flight transfer across loop iterations; range_ unwraps it to its handle for the scf.for iter_arg and re-wraps the block argument as a Task for the body.

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

def __init__(self, handle):
    self._handle = handle

handle property

handle

The transfer's !index SSA value (the scf iter_arg payload).

free

free() -> None

Return this transfer's buffer descriptor to the pool (dma_free_task).

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

def free(self) -> None:
    """Return this transfer's buffer descriptor to the pool (``dma_free_task``)."""
    dma_free_task(self._handle)

await_

await_() -> None

Block until this transfer completes (dma_await_task).

The transfer must have been issued with wait=True so it carries a completion token.

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

def await_(self) -> None:
    """Block until this transfer completes (``dma_await_task``).

    The transfer must have been issued with ``wait=True`` so it carries a
    completion token.
    """
    dma_await_task(self._handle)