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.

Structural design objects are resolvable: they lower to MLIR operations when the design is compiled. This page also includes host-side utilities, type markers, and runtime task handles. For the direct MLIR op wrappers that IRON lowers to, see the Dialect op wrappers page.

Import the public design abstractions from aie.iron (e.g. from aie.iron import Worker, ObjectFifo, Runtime), or use import aie.iron as iron for decorators and tensor factories. Supporting types are also documented below; not all are re-exported at package level.

Prefer ObjectFifo for synchronized streaming, Worker for compute, and Runtime(sequence, fn_args) for host-side transfers. Pass runtime buffer types and the ObjectFifo handles used by the sequence through fn_args, and pass workers to Program(workers=...). Use Flow, TileDma, and Lock when explicit routing or DMA control is the teaching goal; use raw dialect operations only where those abstractions do not expose the required operation.

---

Core design abstractions

The objects most designs are built from.

Program

Program

Program(
    device: Device | None,
    rt: Runtime,
    workers: list | None = None,
    *,
    expand_load_pdis: bool | None = None,
    reconfig_teardown: bool | None = None,
    reconfig_init_entry: bool | None = None
)

One compilation unit and host-visible execution graph.

A Program emits one MLIR module. It composes one or more DeviceConfiguration images and selects the entry Runtime that orchestrates one run. Each device configuration owns the resources inside one aie.device operation; Program owns module creation, entry selection, and cross-configuration validation.

Construct a Program with 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. Accepts the Device | None returned by iron.get_current_device directly and raises if no device has been selected, so callers need not narrow it first.	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
expand_load_pdis	bool | None	The value passed to aiecc's --expand-load-pdis option during full-ELF compilation. None omits the option.	None
reconfig_teardown	bool | None	The value passed to aiecc's --reconfig-teardown option. None omits the option.	None
reconfig_init_entry	bool | None	The value passed to aiecc's --reconfig-init-entry option. None omits the option.	None

Raises:

Type	Description
ValueError	If device is None (no NPU device was selected/detected).

Source code in python/iron/program.py

def __init__(
    self,
    device: Device | None,
    rt: Runtime,
    workers: "list | None" = None,
    *,
    expand_load_pdis: bool | None = None,
    reconfig_teardown: bool | None = None,
    reconfig_init_entry: bool | None = None,
):
    """Construct a Program with 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.
            Accepts the ``Device | None`` returned by ``iron.get_current_device``
            directly and raises if no device has been selected, so callers need
            not narrow it first.
        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.
        expand_load_pdis: The value passed to aiecc's
            ``--expand-load-pdis`` option during full-ELF compilation.
            ``None`` omits the option.
        reconfig_teardown: The value passed to aiecc's
            ``--reconfig-teardown`` option. ``None`` omits the option.
        reconfig_init_entry: The value passed to aiecc's
            ``--reconfig-init-entry`` option. ``None`` omits the option.

    Raises:
        ValueError: If ``device`` is None (no NPU device was selected/detected).
    """
    if device is None:
        raise ValueError(
            "Program requires a device, but none was selected. Pass an explicit "
            "Device, or ensure an NPU runtime is available for "
            "iron.get_current_device()."
        )
    configuration = DeviceConfiguration(
        "main", device, workers=workers or (), runtimes=[rt]
    )
    self._configurations = [configuration]
    self._entry = rt
    self._implicit_configuration = True
    self._reconfiguration_options = {
        "expand_load_pdis": expand_load_pdis,
        "reconfig_teardown": reconfig_teardown,
        "reconfig_init_entry": reconfig_init_entry,
    }
    self._site = SourceSite.capture()
    self._assign_names()

compose classmethod

compose(
    configurations: list[DeviceConfiguration],
    *,
    entry: Runtime,
    expand_load_pdis: bool | None = None,
    reconfig_teardown: bool | None = None,
    reconfig_init_entry: bool | None = None
) -> Program

Compose configurations into one Program run from entry.

Parameters:

Name	Type	Description	Default
configurations	list[DeviceConfiguration]	The device configurations, each one aie.device.	required
entry	Runtime	The runtime that orchestrates one run.	required
expand_load_pdis	bool | None	The value passed to aiecc's --expand-load-pdis option. None omits the option.	None
reconfig_teardown	bool | None	The value passed to aiecc's --reconfig-teardown option. None omits the option.	None
reconfig_init_entry	bool | None	The value passed to aiecc's --reconfig-init-entry option. None omits the option.	None

Source code in python/iron/program.py

@classmethod
def compose(
    cls,
    configurations: "list[DeviceConfiguration]",
    *,
    entry: Runtime,
    expand_load_pdis: bool | None = None,
    reconfig_teardown: bool | None = None,
    reconfig_init_entry: bool | None = None,
) -> "Program":
    """Compose configurations into one Program run from ``entry``.

    Args:
        configurations: The device configurations, each one ``aie.device``.
        entry: The runtime that orchestrates one run.
        expand_load_pdis: The value passed to aiecc's
            ``--expand-load-pdis`` option. ``None`` omits the option.
        reconfig_teardown: The value passed to aiecc's
            ``--reconfig-teardown`` option. ``None`` omits the option.
        reconfig_init_entry: The value passed to aiecc's
            ``--reconfig-init-entry`` option. ``None`` omits the option.
    """
    program = cls.__new__(cls)
    program._configurations = list(configurations)
    program._entry = entry
    program._implicit_configuration = False
    program._reconfiguration_options = {
        "expand_load_pdis": expand_load_pdis,
        "reconfig_teardown": reconfig_teardown,
        "reconfig_init_entry": reconfig_init_entry,
    }
    program._site = SourceSite.capture()
    program._assign_names()
    program._validate_composition()
    return program

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,
    core_trace_mode=EventTime,
)

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
core_trace_mode	TraceMode	Trace mode for core tiles. Defaults to Event-Time.	EventTime

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,
    core_trace_mode=TraceMode.EventTime,
):
    """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.
        core_trace_mode (TraceMode, optional): Trace mode for core tiles.
            Defaults to Event-Time.
    """
    configuration = self._entry.configuration
    assert configuration is not None
    configuration.enable_trace(
        trace_size=trace_size,
        workers=workers,
        reuse_output_buffer=reuse_output_buffer,
        coretile_events=coretile_events,
        coremem_events=coremem_events,
        memtile_events=memtile_events,
        shimtile_events=shimtile_events,
        egress_shim_col=egress_shim_col,
        core_trace_mode=core_trace_mode,
    )

resolve_program

resolve_program(
    device_name: str = "main",
    context: Context | None = None,
) -> Module

Resolve 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.

Parameters:

Name	Type	Description	Default
device_name	str	The symbol of the generated aie.device.	'main'
context	Context | None	The context to build the module in, so a caller combining several programs' devices can move them rather than print and parse them; a new one when None.	None

Returns:

Type	Description
Module	The module holding the design's aie.device.

Source code in python/iron/program.py

def resolve_program(
    self, device_name: str = "main", context: ir.Context | None = None
) -> ir.Module:
    """Resolve 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.

    Args:
        device_name: The symbol of the generated ``aie.device``.
        context: The context to build the module in, so a caller combining
            several programs' devices can move them rather than print and
            parse them; a new one when None.

    Returns:
        The module holding the design's ``aie.device``.
    """
    try:
        return self._resolve_program(device_name, context)
    except Exception as exc:
        raise design_error(exc)

DeviceConfiguration

DeviceConfiguration

DeviceConfiguration(
    name: str | Device | None = None,
    device: Device | None = None,
    *,
    workers: Sequence = (),
    runtimes: Sequence[Runtime] = ()
)

Bases: DeviceResources

A device image and the runtime sequences that use it.

Source code in python/iron/configuration.py

def __init__(
    self,
    name: str | Device | None = None,
    device: Device | None = None,
    *,
    workers: Sequence = (),
    runtimes: Sequence[Runtime] = (),
) -> None:
    super().__init__()
    if isinstance(name, Device):
        if device is not None:
            raise TypeError("DeviceConfiguration received two devices.")
        device = name
        name = None
    if device is None:
        raise TypeError("DeviceConfiguration requires a device.")
    if name == "":
        raise ValueError("DeviceConfiguration name must not be empty.")
    self._name = name
    self._device = device
    self._workers = list(workers)
    self._runtimes = list(runtimes)
    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
    self._core_trace_mode = TraceMode.EventTime
    self._owners = None
    self._prebuilt: str | None = None
    self._scratchpad_parameters: list[ScratchpadParameter] = []

    runtime_names = [
        runtime.name for runtime in self._runtimes if runtime.name is not None
    ]
    duplicates = sorted(
        name for name in set(runtime_names) if runtime_names.count(name) > 1
    )
    if duplicates:
        raise ValueError(
            f"DeviceConfiguration {self._name!r} has duplicate runtime names: {duplicates}."
        )
    for runtime in self._runtimes:
        runtime._bind_configuration(self)

scratchpad_parameters property

scratchpad_parameters: list[ScratchpadParameter]

The parameters a prebuilt device declares beside it, used or not by a sequence.

from_module classmethod

from_module(
    name: str, module: Module
) -> DeviceConfiguration

Return a configuration that emits module's one aie.device verbatim.

Each runtime sequence in the device becomes a Runtime of the same name and argument types. The aiex.scratchpad_parameter declarations beside the device are declared again with all their attributes, so Program.compose takes it like any other.

Parameters:

Name	Type	Description	Default
name	str	The configuration's name; the device is renamed to it.	required
module	Module	The module holding the device and its declarations.	required

Returns:

Type	Description
DeviceConfiguration	The configuration.

Raises:

Type	Description
ValueError	module holds no aie.device, or several; or the device needs renaming and an op inside it has its name, which the rename would rewrite too.
TypeError	A sequence argument has no NumPy dtype or a dynamic shape, or a scratchpad parameter has no NumPy dtype.

Source code in python/iron/configuration.py

@classmethod
def from_module(cls, name: str, module: ir.Module) -> DeviceConfiguration:
    """Return a configuration that emits ``module``'s one ``aie.device`` verbatim.

    Each runtime sequence in the device becomes a Runtime of the same name
    and argument types. The ``aiex.scratchpad_parameter`` declarations
    beside the device are declared again with all their attributes, so
    ``Program.compose`` takes it like any other.

    Args:
        name: The configuration's name; the device is renamed to it.
        module: The module holding the device and its declarations.

    Returns:
        The configuration.

    Raises:
        ValueError: ``module`` holds no ``aie.device``, or several; or the
            device needs renaming and an op inside it has its name,
            which the rename would rewrite too.
        TypeError: A sequence argument has no NumPy dtype or a dynamic
            shape, or a scratchpad parameter has no NumPy dtype.
    """
    devices = [op for op in module.body.operations if isinstance(op, DeviceOp)]
    if len(devices) != 1:
        raise ValueError(
            f"DeviceConfiguration.from_module needs a module holding one "
            f"aie.device, not {len(devices)}."
        )
    (device_op,) = devices
    old = ir.StringAttr(device_op.attributes["sym_name"]).value
    if old != name and any(
        "sym_name" in op.attributes
        and ir.StringAttr(op.attributes["sym_name"]).value == old
        for op in device_op.body_region.blocks[0].operations
    ):
        raise ValueError(
            f"aie.device @{old} holds an op named @{old}; renaming the "
            f"device to @{name} would rewrite references to that symbol too."
        )
    with module.context:
        declarations = [
            op
            for op in module.body.operations
            if isinstance(op, ScratchpadParameterOp)
        ]
        parameters = [
            ScratchpadParameter.from_declaration(op) for op in declarations
        ]
        runtimes = []
        for sequence in device_op.body_region.blocks[0].operations:
            if not isinstance(sequence, RuntimeSequenceOp):
                continue
            types = []
            for arg in sequence.body.blocks[0].arguments:
                if not isinstance(arg.type, ir.MemRefType):
                    dtype = mlir_type_to_np_dtype(arg.type)
                    if dtype is None:
                        raise TypeError(f"No NumPy dtype for argument {arg.type}.")
                    types.append(dtype)
                    continue
                memref = ir.MemRefType(arg.type)
                if not memref.has_static_shape:
                    raise TypeError(
                        f"Argument {memref} has a dynamic shape; a Runtime "
                        "argument's shape is fixed."
                    )
                dtype = mlir_type_to_np_dtype(memref.element_type)
                if dtype is None:
                    raise TypeError(f"No NumPy dtype for the elements of {memref}.")
                types.append(np.ndarray[tuple(memref.shape), np.dtype[dtype]])
            runtime = Runtime(
                lambda *_: None,
                types,
                name=ir.StringAttr(sequence.attributes["sym_name"]).value,
            )
            for parameter in parameters:
                runtime.register_parameter(parameter)
            runtimes.append(runtime)
        kind = AIEDevice(ir.IntegerAttr(device_op.attributes["device"]).value)
        # The device's reads verify against the declarations, so the text
        # parsed again at resolution keeps them; only the device is cloned.
        text = "\n".join([*(str(op) for op in declarations), str(device_op)])
    configuration = cls(name, NamedDevice.of(kind), runtimes=runtimes)
    configuration._prebuilt = text
    configuration._scratchpad_parameters = parameters
    return configuration

resolve

resolve(
    *,
    device_name: str,
    loc: Location,
    entry: Runtime,
    owners: dict[int, "DeviceConfiguration"]
) -> None

Emit this configuration as one aie.device operation.

Source code in python/iron/configuration.py

def resolve(
    self,
    *,
    device_name: str,
    loc: ir.Location,
    entry: Runtime,
    owners: dict[int, "DeviceConfiguration"],
) -> None:
    """Emit this configuration as one ``aie.device`` operation."""
    # The remaining emission order matches the dependencies between device ops.
    self._owners = owners
    if self._prebuilt is not None:
        prebuilt = ir.Module.parse(self._prebuilt)  # alive while cloned
        (op,) = [o for o in prebuilt.body.operations if isinstance(o, DeviceOp)]
        clone = op.clone(ip=ir.InsertionPoint.current)
        old = ir.StringAttr(clone.attributes["sym_name"]).value
        clone.attributes["sym_name"] = ir.StringAttr.get(device_name)
        ir.SymbolTable.replace_all_symbol_uses(old, device_name, clone.operation)
        if entry in self._runtimes and get_compile_arg("_iron_full_elf"):
            (sequence,) = [
                o
                for o in clone.body_region.blocks[0].operations
                if isinstance(o, RuntimeSequenceOp)
                and ir.StringAttr(o.attributes["sym_name"]).value == entry.name
            ]
            entry.load_pdi_first(sequence.body.blocks[0], device_name)
        return
    self._name_unnamed()
    device_type = type(self._device)
    self._device = device_type()  # pyright: ignore[reportCallIssue]
    current_device = self._device

    for runtime in self._runtimes:
        self._claim(runtime, owners)
    for worker in self._workers:
        self._claim(worker, owners)

    @device(current_device.resolve(), sym_name=device_name, loc=loc)
    def device_body():
        all_fifos = set()
        all_pools = set()
        for runtime in self._runtimes:
            all_fifos.update(runtime.fifos)
        for worker in self._workers:
            for handle in worker.fifos:
                # A pool handle states one end only, so it has no fifo graph
                # to walk and no delegate tile to follow.
                if isinstance(handle, ObjectFifoPoolHandle):
                    all_pools.add(handle)
                else:
                    all_fifos.add(handle)
        all_fifos = sorted(all_fifos, key=lambda obj: obj.name)
        all_pools = sorted(all_pools, key=lambda obj: obj.name or "")

        all_tiles = []
        for worker in self._workers:
            all_tiles.append(worker.tile)
            for barrier in worker._barriers:
                self._claim(barrier, owners)
            for arg in worker.flat_fn_args:
                if isinstance(arg, Resolvable):
                    self._claim(arg, owners)
                    all_tiles.extend(arg.tiles())
        reached = []
        for handle in all_fifos:
            self._claim_fifo_graph(handle, owners, reached=reached)
            all_tiles.extend(
                [endpoint.tile for endpoint in handle.all_of_endpoints()]
            )
        # A fifo reached only through links, as the middle of a chain of
        # forwards is, has endpoints no worker or runtime names.
        for fifo in reached:
            for candidate in [fifo._prod, *fifo._cons]:
                if candidate and candidate.endpoint is not None:
                    all_tiles.append(candidate.endpoint.tile)
            delegate = fifo._delegate_tile
            if delegate is not None:
                all_tiles.append(delegate)
        for handle in all_pools:
            self._claim(handle.pool, owners)
            all_tiles.extend(handle.tiles())
            # The pool names these but does not own them, so nothing else
            # would claim one a hand-written DMA program never mentions.
            for resource in handle.pool.resources():
                self._claim(resource, owners)
                all_tiles.extend(resource.tiles())
        for flow in self._flows:
            self._claim(flow, owners)
            all_tiles.extend(flow.all_tiles())
        for tile_dma in self._tile_dmas:
            self._claim(tile_dma, owners)
            all_tiles.extend(tile_dma.all_tiles())
            buffers, locks = tile_dma.all_buffers_and_locks()
            for resource in [*buffers, *locks]:
                self._claim(resource, owners)
        for lock in self._locks:
            self._claim(lock, owners)
            all_tiles.append(lock.tile)

        for tile in all_tiles:
            self._claim(tile, owners)
            current_device.resolve_tile(tile)

        for handle in all_fifos:
            handle.resolve()
        for handle in all_pools:
            for resource in handle.pool.resources():
                if isinstance(resource, Buffer):
                    resource.place(handle.pool.tile)
                resource.resolve()
            handle.resolve()
        for lock in self._locks:
            lock.resolve()
        for tile_dma in self._tile_dmas:
            buffers, locks = tile_dma.all_buffers_and_locks()
            for lock in locks:
                lock.resolve()
            for buffer in buffers:
                buffer.place(tile_dma.tile)
                buffer.resolve()

        for worker in self._workers:
            for arg in worker.flat_fn_args:
                if isinstance(arg, FuncBase):
                    arg.emit()
                elif isinstance(arg, Resolvable):
                    if arg not in self._flows and arg not in self._tile_dmas:
                        arg.resolve()

        for worker in self._workers:
            worker.resolve()
        for worker in self._workers:
            for cascade in worker._outgoing_cascades:
                if cascade._dst not in self._workers:
                    raise ValueError(
                        "CascadeFlow endpoints must belong to the same device "
                        f"configuration {self.name!r}."
                    )
                self._claim(cascade, owners)
                self._claim(cascade._dst, owners)
                cascade.resolve()

        self.resolve_tile_dmas()

        if self._trace_workers is not None:
            tiles_to_trace = [worker.tile.op for worker in self._trace_workers]
        else:
            tiles_to_trace = [
                worker.tile.op
                for worker in self._workers
                if worker.trace is not None
            ]
        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,
                core_trace_mode=self._core_trace_mode,
            )

        for runtime in self._runtimes:
            implicit_configure_device_ref = (
                device_name
                if runtime is entry and get_compile_arg("_iron_full_elf")
                else None
            )
            runtime.resolve(
                trace_size=self._trace_size,
                reuse_output_buffer=self._reuse_output_buffer,
                egress_shim_col=self._egress_shim_col,
                implicit_configure_device_ref=implicit_configure_device_ref,
                device=current_device,
            )

        for flow in self._flows:
            flow.resolve()

Worker

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

Worker

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

Bases: ObjectFifoEndpoint, PerDeviceConfiguration

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 for the worker. None lets aiecc reserve the core's measured stack requirement, never less than AIETargetModel::getDefaultCoreStackSize() (currently 1024 bytes). Set it where aiecc cannot measure the core, as in a Chess build. Defaults to None.	None
data_size	int	Bytes of data memory to reserve for this core's compiled sections (.data/.rodata/.bss), beyond the stack. The buffer allocator packs the tile's buffers around the reservation. None leaves the core whatever contiguous run the buffers leave. Defaults to None.	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
wait_for_parameters	bool	If true, each iteration of the Worker's loop first waits for the current dispatch's ScratchpadParameter values (aiex.wait_scratchpad_parameters), so the reads in core_fn see this dispatch's values. Set it when one iteration of a while_true Worker is one dispatch; without it the Worker waits only once, before its first iteration. Requires a ScratchpadParameter in fn_args. Defaults to False.	False

Raises:

Type	Description
ValueError	Parameters are validated.
TypeError	An argument is computed per call.

Source code in python/iron/worker.py

def __init__(
    self,
    core_fn: Callable | None,
    fn_args: list | None = None,
    tile: Tile | None = AnyComputeTile,
    while_true: bool = True,
    stack_size: int | None = None,
    data_size: int | None = None,
    trace: int | None = None,
    trace_events: list | None = None,
    dynamic_objfifo_lowering: bool | None = None,
    wait_for_parameters: bool = False,
):
    """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 for the worker.
            None lets aiecc reserve the core's measured stack requirement, never
            less than AIETargetModel::getDefaultCoreStackSize() (currently 1024
            bytes). Set it where aiecc cannot measure the core, as in a Chess
            build. Defaults to None.
        data_size (int, optional): Bytes of data memory to reserve for this
            core's compiled sections (.data/.rodata/.bss), beyond the stack. The
            buffer allocator packs the tile's buffers around the reservation. None
            leaves the core whatever contiguous run the buffers leave.
            Defaults to None.
        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.
        wait_for_parameters (bool, optional): If true, each iteration of the
            Worker's loop first waits for the current dispatch's
            ScratchpadParameter values (``aiex.wait_scratchpad_parameters``),
            so the reads in ``core_fn`` see this dispatch's values. Set it when
            one iteration of a ``while_true`` Worker is one dispatch; without
            it the Worker waits only once, before its first iteration.
            Requires a ScratchpadParameter in ``fn_args``. Defaults to False.

    Raises:
        ValueError: Parameters are validated.
        TypeError: An argument is computed per call.
    """
    self._site = SourceSite.capture()
    for arg in flatten_fn_args(
        [
            fn_args or [],
            tile,
            while_true,
            stack_size,
            data_size,
            trace,
            trace_events,
            dynamic_objfifo_lowering,
        ]
    ):
        if isinstance(arg, _DispatchParameter):
            arg.refuse()
        if isinstance(arg, Affine):
            raise TypeError(
                f"A Worker cannot take {arg}, which is computed per call; use "
                "CompileTime or specialize"
            )
    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}"
        )
    if stack_size is not None:
        if not isinstance(stack_size, int) or isinstance(stack_size, bool):
            raise ValueError(
                f"Worker stack_size must be an int, but got "
                f"{type(stack_size).__name__}"
            )
        if stack_size < 1:
            raise ValueError(
                f"Worker stack_size must be >= 1, but got {stack_size}"
            )
    if data_size is not None:
        if not isinstance(data_size, int) or isinstance(data_size, bool):
            raise ValueError(
                f"Worker data_size must be an int, but got "
                f"{type(data_size).__name__}"
            )
        if data_size < 0:
            raise ValueError(
                f"Worker data_size must be >= 0, but got " f"{data_size}"
            )
    # Store the user's Tile directly when it is already typed as CoreTile.
    # This preserves Python object identity so a Buffer and a Worker that
    # share the same Tile object resolve to a single LogicalTileOp. When a
    # fresh copy is needed (untyped tile or the singleton default) use
    # with_type() — it always returns a new object.
    if tile.tile_type == AIETileType.CoreTile and tile is not AnyComputeTile:
        self._tile = tile
    else:
        self._tile = tile.with_type(AIETileType.CoreTile)
    self._while_true = while_true
    self.stack_size = stack_size
    self.data_size = data_size
    self._dynamic_objfifo_lowering = dynamic_objfifo_lowering
    if wait_for_parameters and not any(
        isinstance(arg, ScratchpadParameter)
        for arg in flatten_fn_args(fn_args or [])
    ):
        raise ValueError(
            "Worker(wait_for_parameters=True) needs a ScratchpadParameter "
            "in fn_args: there are no parameters to wait for."
        )
    self._wait_for_parameters = wait_for_parameters
    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 forever_():
                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 registers each outgoing edge here for configuration discovery.
    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, ObjectFifoPoolHandle)):
            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 or arg._arr_type} 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. DeviceConfiguration discovers
                # the neighbor tile via Buffer.tiles().
                arg.place(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)
        # Kernel/ExternalFunction instances are valid fn_args — they resolve to
        # func.call ops when invoked inside core_fn and carry link_with on their
        # func.func declaration. Other unrecognized args are assumed to be
        # metaprogramming values (Python scalars, etc.).

    # A library kernel's contract may name a setup kernel, such as the
    # rounding mode its bf16 stores assume; a fresh core boots in floor.
    # Run each once before the loop, unless fn_args already hands it over
    # for core_fn to call.
    kernels = [a for a in flatten_fn_args(self.fn_args) if isinstance(a, Kernel)]
    handed = {k.name for k in kernels}
    self._setup_kernels = []
    for k in kernels:
        if k.contract is None or k.contract.setup is None:
            continue
        setup = k.contract.setup()
        if setup.name not in handed:
            handed.add(setup.name)
            self._setup_kernels.append(setup)

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. The setup kernels the Worker runs for its kernels' contracts come last.

fifos property

fifos: list[ObjectFifoHandle | ObjectFifoPoolHandle]

Returns the fifo and pool handles given to the Worker via fn_args.

Returns:

Type	Description
list[ObjectFifoHandle | ObjectFifoPoolHandle]	list[ObjectFifoHandle | ObjectFifoPoolHandle]: handles the Worker acquires and releases objects through.

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:

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

    with:

    ```python
    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)

Bases: PerDeviceConfiguration

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, *, greater_equal: bool = False)

Wait for the barrier to be set to value.

Should be called from inside a core function. The wait leaves the barrier at value; a worker that loops over dispatches calls release_with_value after reading its runtime parameters so the next iteration waits for the next set.

Parameters:

Name	Type	Description	Default
value	int	The value to wait for.	required
greater_equal	bool	Use a consuming greater-than-or-equal lock wait.	False

Source code in python/iron/worker.py

def wait_for_value(self, value: int, *, greater_equal: bool = False):
    """Wait for the barrier to be set to `value`.

    Should be called from inside a core function. The wait leaves the
    barrier at ``value``; a worker that loops over dispatches calls
    ``release_with_value`` after reading its runtime parameters so the
    next iteration waits for the next ``set``.

    Args:
        value (int): The value to wait for.
        greater_equal (bool): Use a consuming greater-than-or-equal lock wait.
    """
    action = LockAction.AcquireGreaterEqual if greater_equal else LockAction.Acquire
    use_lock(self._get_worker_lock(), action, 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 the barrier, adding value to it, inside the core.

Parameters:

Name	Type	Description	Default
value	int	The value to add.	required

Source code in python/iron/worker.py

def release_with_value(self, value: int):
    """Release the barrier, adding ``value`` to it, inside the core.

    Args:
        value (int): The value to add.
    """
    use_lock(self._get_worker_lock(), LockAction.Release, value=value)

ObjectFifo

ObjectFifo

ObjectFifo(
    obj_type: type[ndarray] | In | Out | InOut,
    *,
    depth: int | None = 2,
    name: str | None = None,
    to_stream: TensorAccessPattern | None = None,
    from_stream_per_cons: TensorAccessPattern | None = None,
    plio: bool = False,
    pad_value: int = 0,
    disable_synchronization: bool = False,
    repeat_count: int | None = None,
    delegate_tile: Tile | None = None,
    transport: Transport | str | None = None,
    init_values: list[ndarray] | None = None,
    consumer_obj_type: (
        type[ndarray] | In | Out | InOut | None
    ) = None
)

Bases: PerDeviceConfigurationResolvable

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] | In | Out | InOut	The type of each buffer in the ObjectFifo, or a generator's tensor operand.	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, its DeviceConfiguration names it. Defaults to None.	None
to_stream	TensorAccessPattern | None	How the producer's DMA walks each object onto the AXI stream: a pattern over a tensor the size of one object (of each input's object, for a join's output), at offset 0. A padded one (from .pad()) also pads the stream, on a MemTile. A forwarded fifo's pattern may instead walk the whole stream of input objects (see forward()). Defaults to None (a linear walk).	None
from_stream_per_cons	TensorAccessPattern | None	How each consumer's DMA writes the stream into its object (into each output's object, for a distribute's input), as for to_stream but unpadded. Defaults to None (a linear walk).	None
plio	bool	Whether the ObjectFifo uses PLIO connections. Defaults to False.	False
pad_value	int	Per-element constant value used to fill the padding of a padded to_stream. Packed into the raw 32-bit CONSTANT_PAD_VALUE register using this fifo's element width. Defaults to 0.	0
disable_synchronization	bool	When True, disables lock-based synchronization on the ObjectFifo. Defaults to False.	False
repeat_count	int | None	If set, the sending end replicates each object this many times before moving on to the next one. The receiving end covers a whole batch in one acquire, so its lock initializers scale to match. 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
transport	Transport | str | None	Which hardware path carries the objects. Transport.dma() forces the DMAs even when producer and consumer share memory, and Transport.shared_mem() insists on shared memory. Transport.dma(packet=Packet()) routes the fifo as an aie.packet_flow sharing the stream with other packet flows, and Packet(id=7) pins the 5-bit header for designs that route on it. None leaves the choice to the lowering, which is the same as Transport.auto(). What drives each end, a DMA channel or a core stream port, is the port of prod and cons. Lowers to the transport attribute on the underlying aie.objectfifo op. 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

Raises:

Type	Description
TypeError	If a stream walk is not a TensorAccessPattern.
ValueError	If depth is provided and is less than 1, or a stream walk is staged, has a nonzero offset, or pads a consumer.

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

def __init__(
    self,
    obj_type: type[np.ndarray] | In | Out | InOut,
    *,
    depth: int | None = 2,
    name: str | None = None,
    to_stream: TensorAccessPattern | None = None,
    from_stream_per_cons: TensorAccessPattern | None = None,
    plio: bool = False,
    pad_value: int = 0,
    disable_synchronization: bool = False,
    repeat_count: int | None = None,
    delegate_tile: Tile | None = None,
    transport: Transport | str | None = None,
    init_values: list[np.ndarray] | None = None,
    consumer_obj_type: type[np.ndarray] | In | Out | InOut | None = None,
):
    """Construct an ObjectFifo.

    Args:
        obj_type (type[np.ndarray] | In | Out | InOut): The type of each buffer in the
            ObjectFifo, or a generator's tensor operand.
        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, its DeviceConfiguration names it. Defaults to None.
        to_stream (TensorAccessPattern | None, optional): How the producer's DMA
            walks each object onto the AXI stream: a pattern over a tensor the
            size of one object (of each input's object, for a join's output),
            at offset 0. A padded one (from ``.pad()``) also pads the stream,
            on a MemTile. A forwarded fifo's pattern may instead walk the whole
            stream of input objects (see ``forward()``). Defaults to None (a
            linear walk).
        from_stream_per_cons (TensorAccessPattern | None, optional): How each
            consumer's DMA writes the stream into its object (into each
            output's object, for a distribute's input), as for ``to_stream``
            but unpadded. Defaults to None (a linear walk).
        plio (bool, optional): Whether the ObjectFifo uses PLIO connections. Defaults to False.
        pad_value (int, optional): Per-element constant value used to fill the padding
            of a padded ``to_stream``. Packed into the raw 32-bit CONSTANT_PAD_VALUE
            register using this fifo's element width. Defaults to 0.
        disable_synchronization (bool, optional): When True, disables lock-based
            synchronization on the ObjectFifo. Defaults to False.
        repeat_count (int | None, optional): If set, the sending end replicates
            each object this many times before moving on to the next one. The
            receiving end covers a whole batch in one acquire, so its lock
            initializers scale to match. 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.
        transport (Transport | str | None, optional): Which hardware path carries the
            objects. ``Transport.dma()`` forces the DMAs even when producer and consumer
            share memory, and ``Transport.shared_mem()`` insists on shared memory.
            ``Transport.dma(packet=Packet())`` routes the fifo as an
            ``aie.packet_flow`` sharing the stream with other packet flows, and
            ``Packet(id=7)`` pins the 5-bit header for designs that route on it. None
            leaves the choice to the lowering, which is the same as ``Transport.auto()``.
            What drives each end, a DMA channel or a core stream port, is the ``port``
            of [`prod`][iron.dataflow.objectfifo.ObjectFifo.prod] and
            [`cons`][iron.dataflow.objectfifo.ObjectFifo.cons].
            Lowers to the ``transport`` attribute on the underlying ``aie.objectfifo`` op.
            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.

    Raises:
        TypeError: If a stream walk is not a ``TensorAccessPattern``.
        ValueError: If ``depth`` is provided and is less than 1, or a stream
            walk is staged, has a nonzero offset, or pads a consumer.
    """
    self._site = SourceSite.capture()
    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 = np_ndarray_type_of(obj_type)
    self._consumer_obj_type: type[np.ndarray] | None = (
        None if consumer_obj_type is None else np_ndarray_type_of(consumer_obj_type)
    )
    self._to_stream = _object_walk(to_stream, "to_stream", can_pad=True)
    self._from_stream_per_cons = _object_walk(
        from_stream_per_cons, "from_stream_per_cons"
    )
    self._plio = plio
    self._pad_value = pad_value
    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; DeviceConfiguration finds it via
    # ObjectFifo._delegate_tile when collecting tiles to assign MLIR ops to.
    self._delegate_tile: Tile | None = delegate_tile
    self._transport: Transport | None = Transport.coerce(transport)
    self._init_values: list[np.ndarray] | None = init_values

depth property

depth: int | None

The default depth of the ObjectFifo's endpoints; prod() and cons() may override it.

from_stream_per_cons property

from_stream_per_cons: TensorAccessPattern | None

How each consumer's DMA writes the stream into its object, unless cons() overrides it; None for a linear walk.

to_stream property

to_stream: TensorAccessPattern | None

How the producer's DMA walks each object onto the stream; None for a linear walk.

shape property

shape: Sequence[int]

The shape of each buffer belonging to the ObjectFifo.

dtype property

dtype: type[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.

consumer_obj_type property

consumer_obj_type: type[ndarray]

The tensor type of each buffer at a consumer: obj_type unless the transfer is asymmetric.

set_iter_count

set_iter_count(iter_count: int)

Set how many times each end of the ObjectFifo cycles through its buffers.

Parameters:

Name	Type	Description	Default
iter_count	int	Passes each end's BD chain makes before it stops. Both ends carry depth * repeat_count * iter_count objects, so a sending end replicating each object repeat_count times makes that many fewer passes than the receiving end opposite it. - Must be in range [1, 256]	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 how many times each end of the ObjectFifo cycles through its buffers.

    Args:
        iter_count (int): Passes each end's BD chain makes before it stops.
            Both ends carry ``depth * repeat_count * iter_count`` objects, so
            a sending end replicating each object ``repeat_count`` times
            makes that many fewer passes than the receiving end opposite it.
            - Must be in range [1, 256]

    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,
    port: Port | None = None,
    tile: Tile | None = None,
) -> ObjectFifoHandle

Return 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
port	Port | None	What drives the producer end: Port.dma(1) pins DMA channel 1, and Port.stream(0) wires it to the core's stream port 0. Defaults to None (a DMA channel allocation picks).	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,
    port: Port | None = None,
    tile: Tile | None = None,
) -> ObjectFifoHandle:
    """Return 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.
        port (Port | None, optional): What drives the producer end:
            ``Port.dma(1)`` pins DMA channel 1, and ``Port.stream(0)`` wires it to
            the core's stream port 0. Defaults to None (a DMA channel allocation picks).
        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("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 port is not None and self._prod.port != port:
            raise ValueError(
                f"Producer handle for {self.name} already pinned to "
                f"{self._prod.port}, cannot re-pin to {port}."
            )
        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, port=port, tile=tile)
    return self._prod

cons

cons(
    depth: int | None = None,
    from_stream: TensorAccessPattern | None = None,
    port: Port | None = None,
    tile: Tile | None = None,
) -> ObjectFifoHandle

Return 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
from_stream	TensorAccessPattern | None	How this consumer's DMA writes the stream into its object (see from_stream_per_cons). Defaults to None.	None
port	Port | None	What drives this consumer end: Port.dma(1) pins DMA channel 1, and Port.stream(0) wires it to the core's stream port 0. Defaults to None (a DMA channel allocation picks).	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,
    from_stream: TensorAccessPattern | None = None,
    port: Port | None = None,
    tile: Tile | None = None,
) -> ObjectFifoHandle:
    """Return 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.
        from_stream (TensorAccessPattern | None, optional): How this consumer's DMA writes
            the stream into its object (see ``from_stream_per_cons``). Defaults to None.
        port (Port | None, optional): What drives this consumer end:
            ``Port.dma(1)`` pins DMA channel 1, and ``Port.stream(0)`` wires it to
            the core's stream port 0. Defaults to None (a DMA channel allocation picks).
        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("If depth is None, then depth must be specified.")
        else:
            depth = self._depth

    self._cons.append(
        ObjectFifoHandle(
            self,
            is_prod=False,
            depth=depth,
            from_stream=from_stream,
            port=port,
            tile=tile,
        )
    )
    return self._cons[-1]

tiles

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

Return the 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]:
    """Return the 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 is 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,
    from_stream: TensorAccessPattern | None = None,
    port: Port | None = None,
    tile: Tile | None = None,
)

Bases: PerDeviceConfigurationResolvable

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
from_stream	TensorAccessPattern | None	This consumer's own walk into its object. Only valid for consumer handles. Defaults to None.	None
port	Port | None	What drives this end (see prod()/cons()). Defaults to None.	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,
    from_stream: TensorAccessPattern | None = None,
    port: Port | 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.
        from_stream (TensorAccessPattern | None, optional): This consumer's own walk into its object. Only valid for consumer handles. Defaults to None.
        port (Port | None, optional): What drives this end (see prod()/cons()). Defaults to None.
        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._fifo_port: ObjectFifoPort = (
        ObjectFifoPort.Produce if is_prod else ObjectFifoPort.Consume
    )
    from_stream = _object_walk(from_stream, "from_stream")
    if is_prod and from_stream is not None:
        raise ValueError("Can only specify from_stream for cons handles")
    elif not is_prod and from_stream is None:
        from_stream = of.from_stream_per_cons

    self._is_prod = is_prod
    self._object_fifo = of
    self._depth = depth
    self._end_port = port
    self._shim_tile = tile
    self._endpoint = None
    self._from_stream = from_stream

name property

name: str | None

The name, or None until its DeviceConfiguration assigns one.

port property

port: Port | None

What drives this handle's end, or None for a DMA channel allocation picks.

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: type[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.

from_stream property

from_stream: TensorAccessPattern | None

How this consumer's DMA writes the stream into its object; None for a linear walk.

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._fifo_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._fifo_port, num_elem)

fill

fill(
    source,
    tap=None,
    wait: bool = False,
    packet: tuple[int, int] | None = None,
    offset_parameter: (
        ScratchpadParameter | str | None
    ) = None,
    group=None,
    managed: bool = True,
    length_parameter: (
        ScratchpadParameter | str | None
    ) = None,
    length_unit: int | None = None,
    repeat_parameter: (
        ScratchpadParameter | str | None
    ) = None,
)

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: ScratchpadParameter | str | None = None,
    group=None,
    managed: bool = True,
    length_parameter: ScratchpadParameter | str | None = None,
    length_unit: int | None = None,
    repeat_parameter: ScratchpadParameter | str | None = None,
):
    """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,
        managed,
        length_parameter,
        length_unit,
        repeat_parameter,
    )

drain

drain(
    dest,
    tap=None,
    wait: bool = False,
    packet: tuple[int, int] | None = None,
    offset_parameter: (
        ScratchpadParameter | str | None
    ) = None,
    group=None,
    managed: bool = True,
    length_parameter: (
        ScratchpadParameter | str | None
    ) = None,
    length_unit: int | None = None,
    repeat_parameter: (
        ScratchpadParameter | str | None
    ) = None,
)

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: ScratchpadParameter | str | None = None,
    group=None,
    managed: bool = True,
    length_parameter: ScratchpadParameter | str | None = None,
    length_unit: int | None = None,
    repeat_parameter: ScratchpadParameter | str | None = None,
):
    """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,
        managed,
        length_parameter,
        length_unit,
        repeat_parameter,
    )

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 | None = AnyMemTile,
    depths: list[int] | None = None,
    obj_types: list[type[ndarray]] | None = None,
    names: Sequence[str | None] | None = None,
    to_stream: (
        Sequence[TensorAccessPattern | None] | None
    ) = None,
    from_stream: (
        Sequence[TensorAccessPattern | None] | None
    ) = None,
    plio: bool = False,
    repeat_counts: Sequence[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. Also accepts None (treated as AnyMemTile). 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	Sequence[str | None] | None	The name of each new ObjectFifo. If not given, each is named after this one, or by the Program if this one is unnamed. Defaults to None.	None
to_stream	Sequence[TensorAccessPattern | None] | None	Each new ObjectFifo's to_stream. Defaults to None.	None
from_stream	Sequence[TensorAccessPattern | None] | None	Each new ObjectFifo consumer's from_stream. Defaults to None.	None
plio	bool	Set plio on each new ObjectFifo. Defaults to False.	False
repeat_counts	Sequence[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 | None = AnyMemTile,
    depths: list[int] | None = None,
    obj_types: list[type[np.ndarray]] | None = None,
    names: Sequence[str | None] | None = None,
    to_stream: Sequence[TensorAccessPattern | None] | None = None,
    from_stream: Sequence[TensorAccessPattern | None] | None = None,
    plio: bool = False,
    repeat_counts: Sequence[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. Also accepts None (treated as AnyMemTile). 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 (Sequence[str | None] | None, optional): The name of each new ObjectFifo. If not given,
            each is named after this one, or by the Program if this one is unnamed.
            Defaults to None.
        to_stream (Sequence[TensorAccessPattern | None] | None, optional): Each new ObjectFifo's
            ``to_stream``. Defaults to None.
        from_stream (Sequence[TensorAccessPattern | None] | None, optional): Each new
            ObjectFifo consumer's ``from_stream``. Defaults to None.
        plio (bool, optional): Set plio on each new ObjectFifo. Defaults to False.
        repeat_counts (Sequence[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._derived_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 to_stream is None:
        to_stream = [None] * num_subfifos
    elif len(to_stream) != num_subfifos:
        raise ValueError(
            "Number of dims to stream does not match number of offsets"
        )

    if from_stream is None:
        from_stream = [None] * num_subfifos
    elif len(from_stream) != num_subfifos:
        raise ValueError("Number of 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],
                to_stream=to_stream[i],
                plio=plio,
                repeat_count=repeat_counts[i],
            )
        )

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

merge

merge(
    num_inputs: int,
    tile: Tile | None = AnyMemTile,
    depths: list[int] | None = None,
    names: Sequence[str | None] | None = None,
    transports: (
        Sequence[Transport | str | None] | None
    ) = None,
) -> list[ObjectFifo]

Construct ObjectFifos that take turns feeding whole objects into this producer.

Unlike join(), where each input fills a slice of every object, each input here sends whole objects of this fifo's type, and they share one channel at the link tile through one packet-switched fan-in route. Objects arrive in whatever order the inputs send them. Only valid for producer ObjectFifoHandles.

Parameters:

Name	Type	Description	Default
num_inputs	int	How many ObjectFifos take turns; at least two.	required
tile	Tile	The tile where the merge occurs. Also accepts None (treated as AnyMemTile). Defaults to AnyMemTile.	AnyMemTile
depths	list[int] | None	The depth of each new ObjectFifo. Defaults to this handle's depth.	None
names	Sequence[str | None] | None	The name of each new ObjectFifo. Defaults to this one's name with _merge{i}.	None
transports	Sequence[Transport | str | None] | None	Each new ObjectFifo's transport, for example to pin the packet header they share. Defaults to None.	None

Raises:

Type	Description
ValueError	Arguments are validated.

Returns:

Type	Description
list[ObjectFifo]	list[ObjectFifo]: The new ObjectFifos whose consumers feed this merge.

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

def merge(
    self,
    num_inputs: int,
    tile: Tile | None = AnyMemTile,
    depths: list[int] | None = None,
    names: Sequence[str | None] | None = None,
    transports: Sequence[Transport | str | None] | None = None,
) -> list[ObjectFifo]:
    """Construct ObjectFifos that take turns feeding whole objects into this producer.

    Unlike join(), where each input fills a slice of every object, each
    input here sends whole objects of this fifo's type, and they share one
    channel at the link tile through one packet-switched fan-in route.
    Objects arrive in whatever order the inputs send them. Only valid for
    producer ObjectFifoHandles.

    Args:
        num_inputs (int): How many ObjectFifos take turns; at least two.
        tile (Tile, optional): The tile where the merge occurs. Also accepts
            None (treated as AnyMemTile). Defaults to AnyMemTile.
        depths (list[int] | None, optional): The depth of each new
            ObjectFifo. Defaults to this handle's depth.
        names (Sequence[str | None] | None, optional): The name of each new
            ObjectFifo. Defaults to this one's name with ``_merge{i}``.
        transports (Sequence[Transport | str | None] | None, optional): Each
            new ObjectFifo's transport, for example to pin the packet header
            they share. Defaults to None.

    Raises:
        ValueError: Arguments are validated.

    Returns:
        list[ObjectFifo]: The new ObjectFifos whose consumers feed this merge.
    """
    if not self._is_prod:
        raise ValueError(f"Cannot merge() a {self.handle_type} ObjectFifoHandle")
    subfifos = self._turn_fifos("merge", num_inputs, depths, names, transports)
    _ = ObjectFifoLink(
        [s.cons() for s in subfifos],
        self,
        tile,
        mode=ObjectFifoLinkMode.Time,
    )
    return subfifos

dispatch

dispatch(
    num_outputs: int,
    tile: Tile | None = AnyMemTile,
    depths: list[int] | None = None,
    names: Sequence[str | None] | None = None,
    transports: (
        Sequence[Transport | str | None] | None
    ) = None,
) -> list[ObjectFifo]

Construct ObjectFifos that take turns receiving this consumer's objects.

Unlike split(), where each output takes a slice of every object, object k goes whole to output k mod num_outputs, over one channel at the link tile and one packet-switched route per output. Each output has one consumer. Only valid for consumer ObjectFifoHandles.

Parameters:

Name	Type	Description	Default
num_outputs	int	How many ObjectFifos take turns; at least two.	required
tile	Tile	The tile where the dispatch occurs. Also accepts None (treated as AnyMemTile). Defaults to AnyMemTile.	AnyMemTile
depths	list[int] | None	The depth of each new ObjectFifo. Defaults to this handle's depth.	None
names	Sequence[str | None] | None	The name of each new ObjectFifo. Defaults to this one's name with _dispatch{i}.	None
transports	Sequence[Transport | str | None] | None	Each new ObjectFifo's transport, for example to pin its turn's packet header. Defaults to None.	None

Raises:

Type	Description
ValueError	Arguments are validated.

Returns:

Type	Description
list[ObjectFifo]	list[ObjectFifo]: The new ObjectFifos whose producers take the turns.

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

def dispatch(
    self,
    num_outputs: int,
    tile: Tile | None = AnyMemTile,
    depths: list[int] | None = None,
    names: Sequence[str | None] | None = None,
    transports: Sequence[Transport | str | None] | None = None,
) -> list[ObjectFifo]:
    """Construct ObjectFifos that take turns receiving this consumer's objects.

    Unlike split(), where each output takes a slice of every object, object
    k goes whole to output k mod `num_outputs`, over one channel at the
    link tile and one packet-switched route per output. Each output has one
    consumer. Only valid for consumer ObjectFifoHandles.

    Args:
        num_outputs (int): How many ObjectFifos take turns; at least two.
        tile (Tile, optional): The tile where the dispatch occurs. Also
            accepts None (treated as AnyMemTile). Defaults to AnyMemTile.
        depths (list[int] | None, optional): The depth of each new
            ObjectFifo. Defaults to this handle's depth.
        names (Sequence[str | None] | None, optional): The name of each new
            ObjectFifo. Defaults to this one's name with ``_dispatch{i}``.
        transports (Sequence[Transport | str | None] | None, optional): Each
            new ObjectFifo's transport, for example to pin its turn's packet
            header. Defaults to None.

    Raises:
        ValueError: Arguments are validated.

    Returns:
        list[ObjectFifo]: The new ObjectFifos whose producers take the turns.
    """
    if self._is_prod:
        raise ValueError(f"Cannot dispatch() a {self.handle_type} ObjectFifoHandle")
    subfifos = self._turn_fifos("dispatch", num_outputs, depths, names, transports)
    _ = ObjectFifoLink(
        self,
        [s.prod() for s in subfifos],
        tile,
        mode=ObjectFifoLinkMode.Time,
    )
    return subfifos

split

split(
    offsets: list[int],
    tile: Tile | None = AnyMemTile,
    depths: list[int] | None = None,
    obj_types: list[type[ndarray]] | None = None,
    names: Sequence[str | None] | None = None,
    to_stream: (
        Sequence[TensorAccessPattern | None] | None
    ) = None,
    from_stream: (
        Sequence[TensorAccessPattern | None] | None
    ) = None,
    plio: bool = False,
    repeat_counts: Sequence[int | None] | None = None,
    pad_value: list[int] | None = None,
    ports: Sequence[Port | 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. Also accepts None (treated as AnyMemTile). 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	Sequence[str | None] | None	The name of each new ObjectFifo. If not given, each is named after this one, or by the Program if this one is unnamed. Defaults to None.	None
to_stream	Sequence[TensorAccessPattern | None] | None	Each new ObjectFifo's to_stream; a padded one pads that output. A split into one fifo at offset 0 is a forward(), whose to_stream may also walk the whole stream (overlapping windows; see forward()). Defaults to None.	None
from_stream	Sequence[TensorAccessPattern | None] | None	Each new ObjectFifo's from_stream_per_cons. Defaults to None.	None
plio	bool	Set plio on each new ObjectFifo. Defaults to False.	False
repeat_counts	Sequence[int | None] | None	Per-sub-fifo MemTile DMA repeat count (see ObjectFifo.repeat_count). Defaults to None.	None
pad_value	list[int] | None	Per-sub-fifo per-element pad fill value (see ObjectFifo.pad_value). Defaults to None.	None
ports	Sequence[Port | None] | None	What drives each output ObjectFifo's producer end, one per output, such as Port.dma(1). split() builds those producer handles itself, so this is the only place to say it. Defaults to None (all compiler-assigned).	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 | None = AnyMemTile,
    depths: list[int] | None = None,
    obj_types: list[type[np.ndarray]] | None = None,
    names: Sequence[str | None] | None = None,
    to_stream: Sequence[TensorAccessPattern | None] | None = None,
    from_stream: Sequence[TensorAccessPattern | None] | None = None,
    plio: bool = False,
    repeat_counts: Sequence[int | None] | None = None,
    pad_value: list[int] | None = None,
    ports: Sequence[Port | 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. Also accepts None (treated as AnyMemTile). 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 (Sequence[str | None] | None, optional): The name of each new ObjectFifo. If not given,
            each is named after this one, or by the Program if this one is unnamed.
            Defaults to None.
        to_stream (Sequence[TensorAccessPattern | None] | None, optional): Each new ObjectFifo's
            ``to_stream``; a padded one pads that output. A split into one
            fifo at offset 0 is a ``forward()``, whose ``to_stream`` may
            also walk the whole stream (overlapping windows; see
            ``forward()``). Defaults to None.
        from_stream (Sequence[TensorAccessPattern | None] | None, optional): Each new
            ObjectFifo's ``from_stream_per_cons``. Defaults to None.
        plio (bool, optional): Set plio on each new ObjectFifo. Defaults to False.
        repeat_counts (Sequence[int | None] | None, optional): Per-sub-fifo MemTile DMA repeat count (see ObjectFifo.repeat_count). Defaults to None.
        pad_value (list[int] | None, optional): Per-sub-fifo per-element pad fill value (see ObjectFifo.pad_value). Defaults to None.

        ports (Sequence[Port | None] | None, optional): What drives each output
            ObjectFifo's producer end, one per output, such as ``Port.dma(1)``.
            split() builds those producer handles itself, so this is the
            only place to say it. Defaults to None (all compiler-assigned).

    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._derived_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 to_stream is None:
        to_stream = [None] * num_subfifos
    elif len(to_stream) != num_subfifos:
        raise ValueError(
            "Number of to_stream arrays does not match number of offsets"
        )

    if from_stream is None:
        from_stream = [None] * num_subfifos
    elif len(from_stream) != num_subfifos:
        raise ValueError(
            "Number of 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")

    if pad_value is None:
        pad_value = [0 for _ in range(num_subfifos)]
    elif len(pad_value) != num_subfifos:
        raise ValueError("Number of pad_value 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],
                to_stream=to_stream[i],
                from_stream_per_cons=from_stream[i],
                plio=plio,
                repeat_count=repeat_counts[i],
                pad_value=pad_value[i],
            )
        )

    # Create link and set it as endpoints
    pinned: list[Port | None] = (
        [None] * len(subfifos) if ports is None else list(ports)
    )
    if len(pinned) != len(subfifos):
        raise ValueError(
            f"split() got {len(pinned)} ports for {len(subfifos)} "
            "outputs; give one per output or none at all."
        )
    # A subfifo's producer handle is built here, so a caller wanting its
    # port pinned has nowhere else to say it -- prod() refuses to re-pin.
    subfifo_prods = [s.prod(port=p) for s, p in zip(subfifos, pinned)]
    _ = ObjectFifoLink(self, subfifo_prods, tile, [], offsets)
    return subfifos

forward

forward(
    tile: Tile | None = AnyMemTile,
    obj_type: type[ndarray] | None = None,
    depth: int | None = None,
    name: str | None = None,
    to_stream: TensorAccessPattern | None = None,
    from_stream: TensorAccessPattern | None = None,
    plio: bool = False,
    repeat_count: int | None = None,
    pad_value: int = 0,
    port: Port | None = None,
) -> ObjectFifo

Forward an ObjectFifoHandle of type consumer to a newly-constructed ObjectFifo.

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

Parameters:

Name	Type	Description	Default
tile	Tile	The tile for the Forward operation. Also accepts None (treated as AnyMemTile). 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, it is named after this one, or by the Program if this one is unnamed. Defaults to None.	None
to_stream	TensorAccessPattern | None	The new ObjectFifo's to_stream; a padded one pads the forwarded stream. On a MemTile it may walk the whole stream of this fifo's objects, e.g. overlapping windows of rows: its tensor is the stream (a whole number of objects), the outer dimension enumerates the new fifo's objects and steps whole input objects, and the inner dimensions are one new object, at most one of them stepping by one input object. Padding only the outer dimension, by (before, after), adds that many zero objects ahead of and behind the stream; walk coordinates start at the first of them. The MemTile then holds the whole stream. Defaults to None.	None
from_stream	TensorAccessPattern | None	The new ObjectFifo's from_stream_per_cons. 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
pad_value	int	Per-element constant fill value for a padded to_stream (see ObjectFifo.pad_value). Defaults to 0.	0
port	Port | None	What drives the forwarded ObjectFifo's producer end, such as Port.dma(1). forward() builds that producer handle itself, so this is the only place to say it. Defaults to None (assigned by the compiler).	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 | None = AnyMemTile,
    obj_type: type[np.ndarray] | None = None,
    depth: int | None = None,
    name: str | None = None,
    to_stream: TensorAccessPattern | None = None,
    from_stream: TensorAccessPattern | None = None,
    plio: bool = False,
    repeat_count: int | None = None,
    pad_value: int = 0,
    port: Port | None = None,
) -> ObjectFifo:
    """Forward an ObjectFifoHandle of type consumer to a newly-constructed ObjectFifo.

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

    Args:
        tile (Tile, optional): The tile for the Forward operation. Also accepts None (treated as AnyMemTile). 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, it
            is named after this one, or by the Program if this one is unnamed. Defaults to None.
        to_stream (TensorAccessPattern | None, optional): The new ObjectFifo's ``to_stream``;
            a padded one pads the forwarded stream. On a MemTile it may walk
            the whole stream of this fifo's objects, e.g. overlapping windows
            of rows: its tensor is the stream (a whole number of objects), the
            outer dimension enumerates the new fifo's objects and steps whole
            input objects, and the inner dimensions are one new object, at
            most one of them stepping by one input object. Padding only the
            outer dimension, by ``(before, after)``, adds that many zero
            objects ahead of and behind the stream; walk coordinates start at
            the first of them. The MemTile then holds the whole stream.
            Defaults to None.
        from_stream (TensorAccessPattern | None, optional): The new ObjectFifo's
            ``from_stream_per_cons``. 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.
        pad_value (int, optional): Per-element constant fill value for a padded
            ``to_stream`` (see ObjectFifo.pad_value). Defaults to 0.
        port (Port | None, optional): What drives the forwarded ObjectFifo's
            producer end, such as ``Port.dma(1)``. forward() builds that
            producer handle itself, so this is the only place to say it.
            Defaults to None (assigned by the compiler).

    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 or self._derived_name("_fwd")]
    to_stream_arg = [to_stream] if to_stream is not None else None
    from_stream_arg = [from_stream] if from_stream is not None else None

    forward_fifo = self.split(
        [0],
        tile=tile,
        obj_types=obj_types,
        depths=depths,
        names=names,
        to_stream=to_stream_arg,
        from_stream=from_stream_arg,
        plio=plio,
        repeat_counts=[repeat_count] if repeat_count is not None else None,
        pad_value=[pad_value] if pad_value else None,
        ports=[port] if port is not None else None,
    )
    return forward_fifo[0]

ObjectFifoLink

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

Bases: ObjectFifoEndpoint, PerDeviceConfigurationResolvable

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
mode	ObjectFifoLinkMode	Space splits every object between the many side's fifos at the offsets; Time has them take turns with whole objects (merge() and dispatch()). Defaults to Space.	Space

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 | None = AnyMemTile,
    src_offsets: list[int] | None = None,
    dst_offsets: list[int] | None = None,
    mode: ObjectFifoLinkMode = ObjectFifoLinkMode.Space,
):
    """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).
        mode (ObjectFifoLinkMode, optional): ``Space`` splits every object
            between the many side's fifos at the offsets; ``Time`` has them
            take turns with whole objects (merge() and dispatch()). Defaults
            to ``Space``.

    Raises:
        ValueError: Arguments are validated.
    """
    self._site = SourceSite.capture()
    self._mode = mode
    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"
        )
    if mode == ObjectFifoLinkMode.Time and (self._src_offsets or self._dst_offsets):
        raise ValueError("A time link moves whole objects and takes no offsets")
    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
    # Isolate singleton defaults, but retain user tiles shared with Workers
    # or other links so they resolve to the same logical tile.
    if any(
        tile is default for default in (AnyMemTile, AnyComputeTile, AnyShimTile)
    ):
        tile = tile.copy()
    # Respect explicit types and let the device infer fully placed tiles.
    placed = tile.col is not None and tile.row is not None
    if tile.tile_type is None and not placed:
        tile.tile_type = AIETileType.MemTile
    ObjectFifoEndpoint.__init__(self, tile)

Runtime

The host-side orchestration entry point. Calls to producer-handle fill and consumer-handle drain are declared in the sequence body passed to Runtime(seq, fn_args); Workers are passed to Program(workers=...).

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",
    seq_op: RuntimeSequenceOp,
    device=None,
)

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. DeviceConfiguration resolves fifos and cores afterward, with every runtime endpoint already bound.

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

def __init__(self, runtime: "Runtime", seq_op: RuntimeSequenceOp, device=None):
    self._runtime = runtime
    self._seq_op = seq_op
    self._device = device
    # 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

runtime property

runtime: 'Runtime'

The Runtime whose sequence body is running.

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[handle] = None

resolve_in_device

resolve_in_device(resolvable) -> None

Resolve a Buffer or Lock the body reaches first, ahead of the sequence.

DeviceConfiguration resolves discoverable objects before the body runs. This method places an object named only by a runtime task at device scope.

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

def resolve_in_device(self, resolvable) -> None:
    """Resolve a Buffer or Lock the body reaches first, ahead of the sequence.

    DeviceConfiguration resolves discoverable objects before the body runs.
    This method places an object named only by a runtime task at device scope.
    """
    configuration = self._runtime.configuration
    if configuration is not None:
        configuration.claim_runtime_resource(resolvable)
    with ir.InsertionPoint(self._seq_op):
        if self._device is not None:
            self._device.resolve_tile(resolvable.tile)
        resolvable.resolve()

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:
        with a[0].location:
            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(strict: bool | None = None) -> None

Close bookkeeping after the body runs, or after one merged step's body.

A fresh default group then takes the next step's transfers.

Parameters:

Name	Type	Description	Default
strict	bool | None	That body's strict_task_groups; the runtime's when None.	None

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

def finalize(self, strict: bool | None = None) -> None:
    """Close bookkeeping after the body runs, or after one merged step's body.

    A fresh default group then takes the next step's transfers.

    Args:
        strict: That body's ``strict_task_groups``; the runtime's when None.
    """
    if strict is None:
        strict = self._runtime._strict_task_groups
    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 strict 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),
    # and close it even when empty: a merged step makes a fresh one.
    self.finish_task_group(self._default_task_group)
    self._default_task_group = TaskGroup(next(self._runtime._task_group_index))
    self._used_default = self._used_explicit = False

Runtime

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

Bases: DeviceResources, 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:

• an unbound DispatchTime parameter: replaced with its live SSA scalar. Forward each parameter once, in any order; parameter identity determines the host binding, not its position in fn_args.
• a type (a tensor type, a generator's In/Out/InOut operand, 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 or NumPy integer value: also declares a runtime input, but is folded into a 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 integer here. NumPy integers retain their scalar dtype; plain Python integers use i32 for compatibility with the common np.int32 path.
• 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
name	str | None	The runtime-sequence symbol name. Program assigns a unique name when this value is None.	None
strict_task_groups	bool	Disallow mixing the default and explicit task groups. Defaults to True.	True
implicit_configure	bool	Configure the full-ELF entry sequence's device when the body contains no aiex.configure or aiex.npu.load_pdi operation. Defaults to True.	True

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

def __init__(
    self,
    seq_fn: Callable,
    fn_args: "Sequence | None" = None,
    *,
    name: str | None = None,
    strict_task_groups: bool = True,
    implicit_configure: 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:

    * an unbound **DispatchTime parameter**: replaced with its live SSA
      scalar. Forward each parameter once, in any order; parameter identity
      determines the host binding, not its position in ``fn_args``.
    * a **type** (a tensor type, a generator's ``In``/``Out``/``InOut``
      operand, 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 or NumPy integer value**: also declares a runtime input,
      but is folded into a 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 integer here. NumPy integers retain their scalar dtype; plain Python
      integers use i32 for compatibility with the common ``np.int32`` path.
    * 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).
        name (str | None): The runtime-sequence symbol name. Program assigns
            a unique name when this value is None.
        strict_task_groups (bool): Disallow mixing the default and explicit task groups. Defaults to True.
        implicit_configure (bool): Configure the full-ELF entry sequence's
            device when the body contains no ``aiex.configure`` or
            ``aiex.npu.load_pdi`` operation. Defaults to True.
    """
    super().__init__()
    if name == "":
        raise ValueError("Runtime name must not be empty.")
    self._site = SourceSite.capture()
    self._name = name
    self._configuration: DeviceConfiguration | None = None
    self._seq_fn: Callable = seq_fn
    self._fn_args = [np_ndarray_type_of(arg) for arg in fn_args or ()]
    self._dispatch_binding = object()
    dispatch_args = [
        arg for arg in self._fn_args if isinstance(arg, _DispatchParameter)
    ]
    if len({id(arg) for arg in dispatch_args}) != len(dispatch_args):
        raise TypeError(
            "Forward each DispatchTime parameter exactly once to Runtime."
        )
    if len({id(arg.owner) for arg in dispatch_args}) > 1:
        raise TypeError(
            "Runtime cannot mix DispatchTime parameters from different designs."
        )
    for container in self._fn_args:
        if isinstance(container, (list, tuple)):
            for arg in flatten_fn_args(container):
                if isinstance(arg, _DispatchParameter):
                    raise TypeError(
                        f"DispatchTime parameter {arg.name!r} must be a direct Runtime fn_args entry."
                    )
    # 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.scalar_type if isinstance(arg, _DispatchParameter) else arg
            )
            if c is None
            and (
                isinstance(arg, (type, _DispatchParameter))
                or get_origin(arg) is np.ndarray
            )
            else None
        )
        for c, arg in zip(self._const_inputs, self._fn_args)
    ]
    # Keep the host scalar ABI in signature order, while the callback and
    # tensor arguments retain fn_args order. Only dispatch slots move.
    dispatch_data = iter(
        data
        for _, data in sorted(
            (
                (arg.position, data)
                for arg, data in zip(self._fn_args, self._rt_data)
                if isinstance(arg, _DispatchParameter)
            ),
            key=lambda item: item[0],
        )
    )
    self._block_data = []
    for arg, data in zip(self._fn_args, self._rt_data):
        if isinstance(arg, _DispatchParameter):
            self._block_data.append(next(dispatch_data))
        elif data is not None:
            if dispatch_args and data.is_scalar:
                raise TypeError(
                    "Runtime cannot mix bare scalar types with DispatchTime parameters."
                )
            self._block_data.append(data)
    self._fifos: dict[ObjectFifoHandle, None] = {}
    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._scratchpad_parameters: list[ScratchpadParameter] = []
    self._strict_task_groups = strict_task_groups
    self._implicit_configure = implicit_configure
    self._task_group_index = itertools.count()

arg_types property

arg_types: list[type]

The sequence's argument types, in the order it takes them.

fifos property

fifos: list[ObjectFifoHandle]

The ObjectFifoHandles driven from the runtime by fill()/drain().

scratchpad_parameters property

scratchpad_parameters: list[ScratchpadParameter]

The ScratchpadParameters the sequence's transfers registered.

merge classmethod

merge(
    runtimes: Sequence[Runtime], name: str | None = None
) -> Runtime

Return one Runtime whose sequence runs the bodies of runtimes in order.

For steps that run back to back on one configuration: they issue one instruction stream with no aiex.configure or aiex.run between them, which an xclbin can carry. Its fn_args are theirs, in order, with a DispatchTime parameter several steps forward kept once, and it registers their flows, locks, tile DMAs and scratchpad parameters.

Parameters:

Name	Type	Description	Default
runtimes	Sequence[Runtime]	The steps, in the order their bodies run.	required
name	str | None	The merged Runtime's name, or None to have one given.	None

Returns:

Type	Description
Runtime	The merged Runtime.

Raises:

Type	Description
ValueError	One of runtimes already belongs to a configuration.
TypeError	The steps forward DispatchTime parameters of different designs.

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

@classmethod
def merge(cls, runtimes: Sequence[Runtime], name: str | None = None) -> Runtime:
    """Return one Runtime whose sequence runs the bodies of ``runtimes`` in order.

    For steps that run back to back on one configuration: they issue one
    instruction stream with no ``aiex.configure`` or ``aiex.run`` between
    them, which an xclbin can carry. Its ``fn_args`` are theirs, in order,
    with a DispatchTime parameter several steps forward kept once, and it
    registers their flows, locks, tile DMAs and scratchpad parameters.

    Args:
        runtimes: The steps, in the order their bodies run.
        name: The merged Runtime's name, or None to have one given.

    Returns:
        The merged Runtime.

    Raises:
        ValueError: One of ``runtimes`` already belongs to a configuration.
        TypeError: The steps forward DispatchTime parameters of different
            designs.
    """
    runtimes = list(runtimes)
    for runtime in runtimes:
        if runtime.configuration is not None:
            raise ValueError(
                f"Runtime {runtime.name!r} belongs to configuration "
                f"{runtime.configuration.name!r}; merge it before configuring."
            )
    fn_args: list = []
    slots: list[list[int]] = []
    for runtime in runtimes:
        slots.append([])
        for arg in runtime._fn_args:
            shared = [
                i
                for i, kept in enumerate(fn_args)
                if isinstance(arg, _DispatchParameter) and kept is arg
            ]
            slots[-1].append(shared[0] if shared else len(fn_args))
            if not shared:
                fn_args.append(arg)

    def body(*args):
        for runtime, slot in zip(runtimes, slots):
            runtime._seq_fn(*(args[i] for i in slot))
            # Closing the step's task groups awaits its wait=True
            # transfers before the next step issues; the rest are freed.
            active_sequence().finalize(runtime._strict_task_groups)

    merged = cls(
        body,
        fn_args,
        name=name,
        implicit_configure=all(rt._implicit_configure for rt in runtimes),
    )
    for runtime in runtimes:
        for flow in runtime.flows:
            merged.add_flow(flow)
        for lock in runtime.locks:
            merged.add_lock(lock)
        for tile_dma in runtime.tile_dmas:
            merged.add_tile_dma(tile_dma)
        for parameter in runtime.scratchpad_parameters:
            merged.register_parameter(parameter)
    return merged

add_flow

add_flow(flow) -> None

Register a flow for resolution alongside the ObjectFifos.

Accepts a Flow or PacketFlow.

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

def add_flow(self, flow) -> None:
    """Register a flow for resolution alongside the ObjectFifos.

    Accepts a [`Flow`][iron.Flow] or [`PacketFlow`][iron.PacketFlow].
    """
    super().add_flow(flow)
    if self._configuration is not None:
        self._configuration.add_flow(flow)

add_lock

add_lock(lock) -> None

Register an explicit Lock no TileDma, task or Worker reaches.

Locks a TileDma's or a task's Bds use are found from them, and a Worker's from its fn_args.

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

def add_lock(self, lock) -> None:
    """Register an explicit [`Lock`][iron.Lock] no TileDma, task or Worker reaches.

    Locks a [`TileDma`][iron.TileDma]'s or a task's Bds use are found
    from them, and a Worker's from its ``fn_args``.
    """
    super().add_lock(lock)
    if self._configuration is not None:
        self._configuration.add_lock(lock)

add_tile_dma

add_tile_dma(tile_dma) -> None

Register a TileDma; channels sharing a Tile are combined at resolution.

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

def add_tile_dma(self, tile_dma) -> None:
    """Register a TileDma; channels sharing a Tile are combined at resolution."""
    if self._configuration is not None:
        self._configuration.add_tile_dma(tile_dma)
    super().add_tile_dma(tile_dma)

resolve_tile_dmas

resolve_tile_dmas() -> None

Validate and emit one DMA region per Tile without changing registrations.

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

def resolve_tile_dmas(self) -> None:
    """Validate and emit one DMA region per Tile without changing registrations."""
    if self._configuration is not None:
        self._configuration.resolve_tile_dmas()
        return
    super().resolve_tile_dmas()

register_parameter

register_parameter(
    param: ScratchpadParameter | str | None,
) -> str | None

Record a ScratchpadParameter a transfer uses, for the Program to declare.

Parameters:

Name	Type	Description	Default
param	ScratchpadParameter | str | None	The parameter, the name of one the Program declares anyway (e.g. from a Worker's fn_args), or None.	required

Returns:

Type	Description
str | None	The parameter's name, or None for None.

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

def register_parameter(self, param: ScratchpadParameter | str | None) -> str | None:
    """Record a ScratchpadParameter a transfer uses, for the Program to declare.

    Args:
        param: The parameter, the name of one the Program declares anyway
            (e.g. from a Worker's `fn_args`), or None.

    Returns:
        The parameter's name, or None for None.
    """
    if isinstance(param, ScratchpadParameter):
        if param not in self._scratchpad_parameters:
            self._scratchpad_parameters.append(param)
        return param.name
    return param

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,
    implicit_configure_device_ref: str | None = None,
    device: Device | 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
loc	Location | None	Optional MLIR source location for the generated ops.	None
ip	InsertionPoint | None	Optional MLIR insertion point for the generated ops.	None
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
implicit_configure_device_ref	str | None	The enclosing device symbol for an implicit full-ELF aiex.npu.load_pdi. None on the xclbin path.	None
device	Device | None	The Device that places tiles the body reaches first, such as a buffer only a runtime task names.	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,
    implicit_configure_device_ref: str | None = None,
    device: Device | 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:
        loc: Optional MLIR source location for the generated ops.
        ip: Optional MLIR insertion point for the generated ops.
        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.
        implicit_configure_device_ref: The enclosing device symbol for an
            implicit full-ELF ``aiex.npu.load_pdi``. ``None`` on the xclbin path.
        device: The [`Device`][iron.Device] that places tiles the body
            reaches first, such as a buffer only a runtime task names.
    """
    # 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._block_data
        if rt_data is not None
    ]
    loc = loc or self._site.location()
    seq_op = RuntimeSequenceOp(sym_name=self._name, loc=loc, ip=ip)
    active = ActiveSequence(self, seq_op, device)
    entry_block = seq_op.body.blocks.append(
        *rt_dtypes, arg_locs=[loc] * len(rt_dtypes)
    )
    with ir.InsertionPoint(entry_block), seq_op.location:
        block_args = iter(entry_block.arguments)
        for rt_data in self._block_data:
            if rt_data is not None:
                rt_data.op = next(block_args)
        for arg, rt_data in zip(self._fn_args, self._rt_data):
            if isinstance(arg, _DispatchParameter):
                assert rt_data is not None
                arg._bind(self._dispatch_binding, rt_data.op)

        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:
                dtype = (
                    type(const_val)
                    if isinstance(const_val, np.integer)
                    else np.int32
                )
                scalar_type = np_dtype_to_mlir_type(dtype)
                value = int(const_val)
                # IntegerAttr's C API accepts int64_t, including the bit
                # pattern of a uint64/index value above INT64_MAX.
                if value > np.iinfo(np.int64).max:
                    value -= 1 << 64
                if (
                    isinstance(scalar_type, ir.IntegerType)
                    and ir.IntegerType(scalar_type).is_unsigned
                ):
                    # arith.constant requires signless integers. EmitC's
                    # ConstantLike op preserves unsigned types and folds.
                    from ...dialects.emitc import (  # pyright: ignore[reportMissingImports]
                        ConstantOp,
                    )

                    body_args.append(
                        ConstantOp(
                            scalar_type, ir.IntegerAttr.get(scalar_type, value)
                        ).result
                    )
                else:
                    body_args.append(constant(value, scalar_type))
            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):
            with traced_body(self._seq_fn):
                self._seq_fn(*body_args)
            active.finalize()

    if implicit_configure_device_ref is not None:
        self.load_pdi_first(entry_block, implicit_configure_device_ref)

    self._dedup_runtime_consumers()

load_pdi_first

load_pdi_first(entry_block: Block, device_ref: str) -> None

Load device_ref's PDI at the start of entry_block, the full-ELF entry's body.

Nothing is added when this Runtime was made with implicit_configure=False or the body already configures a device.

Parameters:

Name	Type	Description	Default
entry_block	Block	The entry runtime sequence's body.	required
device_ref	str	The symbol of the device whose PDI it loads.	required

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

def load_pdi_first(self, entry_block: ir.Block, device_ref: str) -> None:
    """Load ``device_ref``'s PDI at the start of ``entry_block``, the full-ELF entry's body.

    Nothing is added when this Runtime was made with
    ``implicit_configure=False`` or the body already configures a device.

    Args:
        entry_block: The entry runtime sequence's body.
        device_ref: The symbol of the device whose PDI it loads.
    """
    if not self._implicit_configure:
        return

    def contains_configuration(block: ir.Block) -> bool:
        for operation in block.operations:
            if operation.name in ("aiex.configure", "aiex.npu.load_pdi"):
                return True
            for region in operation.regions:
                if any(
                    contains_configuration(nested_block)
                    for nested_block in region.blocks
                ):
                    return True
        return False

    if not contains_configuration(entry_block):
        operations = list(entry_block.operations)
        with ir.InsertionPoint(entry_block):
            load_pdi = npu_load_pdi(device_ref=device_ref)
        if operations:
            load_pdi.operation.move_before(operations[0])

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

Use a Buffer for local scratch storage shared by sequential kernel calls in one Worker, as in the edge-detection example. Passing it in the Worker's fn_args associates it with that Worker's tile. Unlike an ObjectFifo, a Buffer does not provide producer/consumer synchronization.

Named memory region accessible by both Workers and the Runtime.

Buffer

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

Bases: PerDeviceConfigurationResolvable

A buffer that is available both to Workers and to the Runtime for operations.

This is often used for Runtime Parameters.

Declare a memory region at the top-level of the design.

The buffer is accessible by both Workers and the Runtime.

Parameters:

Name	Type	Description	Default
type	type[ndarray] | In | Out | InOut | None	The type of the buffer, or a generator's tensor operand. 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, the buffer is left unnamed, or its DeviceConfiguration names it if the runtime writes it. 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
mem_bank	int | None	Pin the buffer to a specific L1 memory bank. The pin is a hard constraint: if the bank cannot hold the buffer, the compiler reports an error rather than placing it elsewhere. Defaults to None (compiler-assigned).	None

Raises:

Type	Description
ValueError	If neither type nor initial_value is provided, or if address/mem_bank are provided but are not a non-negative int.

Source code in python/iron/buffer.py

def __init__(
    self,
    type: "type[np.ndarray] | In | Out | InOut | 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,
    mem_bank: int | None = None,
):
    """Declare a memory region at the top-level of the design.

    The buffer is accessible by both Workers and the Runtime.

    Args:
        type (type[np.ndarray] | In | Out | InOut | None, optional): The type of the buffer,
            or a generator's tensor operand. 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, the
            buffer is left unnamed, or its DeviceConfiguration names it if
            the runtime writes it. 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).
        mem_bank (int | None, optional): Pin the buffer to a specific L1 memory bank.
            The pin is a hard constraint: if the bank cannot hold the buffer, the
            compiler reports an error rather than placing it elsewhere. Defaults to
            None (compiler-assigned).

    Raises:
        ValueError: If neither ``type`` nor ``initial_value`` is provided, or if
            ``address``/``mem_bank`` are provided but are not a non-negative int.
    """
    self._site = SourceSite.capture()
    if type is None and initial_value is None:
        raise ValueError("Must provide either type, initial value, or both.")
    if address is not None:
        if not isinstance(address, int) or isinstance(address, bool):
            raise ValueError(
                f"Buffer address must be an int, but got "
                f"{address.__class__.__name__}"
            )
        if address < 0:
            raise ValueError(f"Buffer address must be >= 0, but got {address}")
    if mem_bank is not None:
        if not isinstance(mem_bank, int) or isinstance(mem_bank, bool):
            raise ValueError(
                f"Buffer mem_bank must be an int, but got "
                f"{mem_bank.__class__.__name__}"
            )
        if mem_bank < 0:
            raise ValueError(f"Buffer mem_bank must be >= 0, but got {mem_bank}")
    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 = np_ndarray_type_of(type)
    self._use_write_rtp = use_write_rtp
    self._address = address
    self._mem_bank = mem_bank
    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.

name property

name: str | None

The buffer's symbol, which another op refers to it by.

shape property

shape: Sequence[int]

The shape of the buffer.

dtype property

dtype: type[NpuDType]

The per-element datatype of the buffer.

place

place(tile: Tile) -> Tile

Put this buffer on tile unless it already has one.

Parameters:

Name	Type	Description	Default
tile	Tile	The tile to place an unplaced buffer on.	required

Returns:

Type	Description
Tile	The tile the buffer is on, which differs from tile if it was
Tile	already placed elsewhere.

Source code in python/iron/buffer.py

def place(self, tile: Tile) -> Tile:
    """Put this buffer on ``tile`` unless it already has one.

    Args:
        tile: The tile to place an unplaced buffer on.

    Returns:
        The tile the buffer is on, which differs from ``tile`` if it was
        already placed elsewhere.
    """
    if self._tile is None:
        self._tile = tile
    return self._tile

tiles

tiles() -> list

Tile dependency for DeviceConfiguration 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 exposes any extra cross-tile placements.

Source code in python/iron/buffer.py

def tiles(self) -> list:
    """Tile dependency for DeviceConfiguration 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 exposes any extra cross-tile
    placements.
    """
    return [self._tile] if self._tile is not None else []

Kernels

Kernel binds a function symbol; KernelObject owns the shared link artifact. Both are exported from aie.iron. Pass a KernelObject("shared.o") to several Kernel constructors to bind symbols from one precompiled object; ObjectFile("shared.o", symbol_prefix=...) is the same for a prebuilt object whose symbols were renamed under a prefix. For C++ source, ExternalFunction creates the owner, exposed as fn.object_file; fn.object_file.bind(symbol, arg_types) binds another entry point to that owner, applying its symbol prefix. External functions with the same explicit output filename and identical source recipes also share ownership. Conflicting recipes for one output filename are rejected. Resolving a source-backed binding registers its artifact for compilation without requiring the original ExternalFunction to remain alive. Independent operations such as kernels.zero(...) own their own objects.

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

KernelObject dataclass

KernelObject(
    name: str,
    link_with_mode: str | None = None,
    _source: _KernelSource | None = None,
    _compiled_dirs: set[str] = set(),
    *,
    _symbol_prefix: str | None = None,
    _built_for_archs: set[str] = set()
)

One link artifact, shared by every kernel that binds one of its symbols.

A prebuilt object only needs a filename and link policy. ExternalFunction supplies the immutable source recipe; compilation state belongs here, not to any particular exported function.

object_file_name property

object_file_name: str

Filename of the linked artifact.

symbol_prefix property

symbol_prefix: str | None

Symbol namespace shared by all bindings of this artifact.

resolve_symbol

resolve_symbol(name: str) -> str

Return name qualified into this object's symbol namespace.

Source code in python/iron/kernel.py

def resolve_symbol(self, name: str) -> str:
    """Return ``name`` qualified into this object's symbol namespace."""
    if not name:
        raise ValueError("Kernel name cannot be empty.")
    return f"{self.symbol_prefix}_{name}" if self.symbol_prefix else name

bind

bind(
    name: str,
    arg_types: list[ArgType] | None = None,
    *,
    link_with_mode: str | None = None,
    stack_size_override: int | None = None
) -> Kernel

Bind a source-level symbol while retaining this artifact's ownership.

Source code in python/iron/kernel.py

def bind(
    self,
    name: str,
    arg_types: list[ArgType] | None = None,
    *,
    link_with_mode: str | None = None,
    stack_size_override: int | None = None,
) -> "Kernel":
    """Bind a source-level symbol while retaining this artifact's ownership."""
    return Kernel(
        self.resolve_symbol(name),
        self,
        arg_types,
        link_with_mode=link_with_mode,
        stack_size_override=stack_size_override,
    )

ObjectFile

ObjectFile(
    object_file_name: str,
    *,
    symbol_prefix: str | None = None,
    link_with_mode: str | None = None
)

Bases: KernelObject

A prebuilt KernelObject with an optional symbol namespace.

Source code in python/iron/kernel.py

def __init__(
    self,
    object_file_name: str,
    *,
    symbol_prefix: str | None = None,
    link_with_mode: str | None = None,
) -> None:
    super().__init__(object_file_name, link_with_mode, _symbol_prefix=symbol_prefix)

Kernel

Kernel(
    name: str,
    object_file_name: str | KernelObject,
    arg_types: list[ArgType] | None = None,
    *,
    link_with_mode: str | None = None,
    stack_size_override: int | None = None
)

Bases: Resolvable

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.

Construct a Kernel backed by a pre-compiled object file.

Parameters:

Name	Type	Description	Default
name	str	Symbol name of the function as it appears in the object file.	required
object_file_name	str | KernelObject	Filename (e.g. "add_one.o") or shared KernelObject of the pre-compiled object file. Must be on the linker search path at compile time.	required
arg_types	list[ArgType] | 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
stack_size_override	int | None	Declared upper bound, in bytes, on the stack that this kernel's call subtree uses. Set it for recursion, for an indirect call, or for a link_with_mode="merge" kernel, which aiecc's stack analysis reads as part of the core rather than as a separate object. See Kernel.stack_size_override.	None

Source code in python/iron/kernel.py

def __init__(
    self,
    name: str,
    object_file_name: str | KernelObject,
    arg_types: list[ArgType] | None = None,
    *,
    link_with_mode: str | None = None,
    stack_size_override: int | None = None,
) -> None:
    """Construct a Kernel backed by a pre-compiled object file.

    Args:
        name: Symbol name of the function as it appears in the object file.
        object_file_name: Filename (e.g. ``"add_one.o"``) or shared
            ``KernelObject`` of the pre-compiled object file. 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.
        stack_size_override: Declared upper bound, in bytes, on the stack
            that this kernel's call subtree uses. Set it for recursion, for
            an indirect call, or for a ``link_with_mode="merge"`` kernel,
            which aiecc's stack analysis reads as part of the core rather
            than as a separate object. See
            [`Kernel.stack_size_override`][iron.kernel.Kernel.stack_size_override].
    """
    self._init_identity(name, arg_types)
    if isinstance(object_file_name, KernelObject):
        if (
            link_with_mode is not None
            and link_with_mode != object_file_name.link_with_mode
        ):
            raise ValueError(
                "link_with_mode conflicts with the shared KernelObject"
            )
        self._object_file = object_file_name
    else:
        self._object_file = KernelObject(object_file_name, link_with_mode)
    self._stack_size_override = stack_size_override

object_file property

object_file: KernelObject

The artifact owner, also shared by sibling symbol bindings.

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.

stack_size_override property

stack_size_override: int | None

Declared upper bound on the stack that this kernel's call subtree uses.

With None, aiecc's analysis computes the bound. An explicit value replaces that computed bound, even when it is smaller: it is a declaration, and 0 is legal. See Core Data Memory.

name property

name: str

Symbol name of the function as it appears in the object file.

resolve

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

Declare this kernel in the enclosing symbol table, once.

The declaration lives in the IR, not on the kernel, so one kernel can be resolved into any number of designs and contexts. A kernel equal to one already declared reuses that declaration; one that shares only the symbol name is rejected.

Source code in python/iron/kernel.py

def resolve(
    self,
    loc: ir.Location | None = None,
    ip: ir.InsertionPoint | None = None,
) -> None:
    """Declare this kernel in the enclosing symbol table, once.

    The declaration lives in the IR, not on the kernel, so one kernel can
    be resolved into any number of designs and contexts. A kernel equal to
    one already declared reuses that declaration; one that shares only the
    symbol name is rejected.
    """
    if self.object_file._source is not None:
        # A generation collects kernels in a registry of its own, so one made
        # before it registers again here, as does one only a sibling binds.
        ExternalFunction._register_object(self)
    point = ip if ip is not None else ir.InsertionPoint.current
    table = _enclosing_symbol_table(point)
    if self._name in table:
        self._check_declaration(table[self._name])
        return
    with point:
        external_func(
            self._name,
            inputs=self._arg_types,
            link_with=self._object_file_name,
            link_with_mode=self._link_with_mode,
            stack_size_override=self._stack_size_override,
            loc=loc or self._site.location(self._name),
            ip=ip,
        )

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 _as_dtype(dt_args[0] if dt_args is not None else dt)
    dtype = getattr(arg, "dtype", None)
    if dtype is not None:
        return _as_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.Kernel.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 the argument types as declared: np.ndarray[shape, dtype] / scalars.

A copy, and stable: resolving the kernel builds MLIR types from these without replacing them, so a memoized kernel describes itself the same way before and after a build.

Source code in python/iron/kernel.py

def arg_types(self) -> list:
    """Return the argument types as declared: ``np.ndarray[shape, dtype]`` / scalars.

    A copy, and stable: resolving the kernel builds MLIR types from these
    without replacing them, so a memoized kernel describes itself the same
    way before and after a build.
    """
    return self._arg_types.copy()

ExternalFunction

ExternalFunction(
    name: str,
    object_file_name: str | None = None,
    source_file: str | None = None,
    source_string: str | None = None,
    arg_types: list[ArgType] | 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,
    stack_size_override: int | None = None,
    contract: Any = None
)

Bases: Kernel

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

Each instance is registered at construction time so that the @jit decorator can discover and compile all source files before invoking the MLIR compilation pipeline. A generation in progress registers into its own set, so registrations from a previous (possibly failed) run stay out of it; outside one, instances go to the process-wide _instances.

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

Construct an ExternalFunction compiled from C/C++ source at JIT time.

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[ArgType] | 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). Relative paths resolve against the current directory at construction. 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
stack_size_override	int | None	Declared upper bound, in bytes, on the stack that this kernel's call subtree uses. See Kernel.stack_size_override. With inline=True, the merged kernel has no separate object, so this bound is the one input aiecc's stack analysis reads for this kernel.	None
contract	Any	What the kernel computes, as an aie.iron.kernels.KernelContract: argument roles, a host reference, a tolerance and the operand layouts. The library factories always give one; a hand-built kernel may leave it None and then cannot be built or judged generically.	None

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[ArgType] | 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,
    stack_size_override: int | None = None,
    contract: Any = None,
) -> None:
    """Construct an ExternalFunction compiled from C/C++ source at JIT time.

    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``).
            Relative paths resolve against the current directory at construction.
            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).
        stack_size_override: Declared upper bound, in bytes, on the stack
            that this kernel's call subtree uses. See
            [`Kernel.stack_size_override`][iron.kernel.Kernel.stack_size_override].
            With ``inline=True``, the merged kernel has no separate object,
            so this bound is the one input aiecc's stack analysis reads for
            this kernel.
        contract: What the kernel computes, as an
            ``aie.iron.kernels.KernelContract``: argument roles, a host
            reference, a tolerance and the operand layouts. The library
            factories always give one; a hand-built kernel may leave it
            ``None`` and then cannot be built or judged generically.
    """
    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."
        )

    self._original_name = name
    self.contract = contract
    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,
        stack_size_override=stack_size_override,
    )

    if source_file is None and source_string is None:
        raise ValueError("source_file or source_string must be provided.")
    if source_file is not None:
        try:
            source_bytes = Path(source_file).read_bytes()
        except OSError:
            source_bytes = f"<unreadable:{source_file}>".encode()
    else:
        assert source_string is not None
        source_bytes = source_string.encode()
    self._object_file = KernelObject(
        object_file_name,
        "merge" if inline else None,
        _KernelSource(
            str(Path(source_file).resolve()) if source_file is not None else None,
            source_string if source_file is None else None,
            hashlib.sha256(source_bytes).hexdigest(),
            tuple(str(Path(d).absolute()) for d in include_dirs or ()),
            tuple(compile_flags or ()),
            use_chess,
            symbol_prefix,
            name if inline else None,
        ),
    )
    self._cached_digest: str | None = None

    # A translation unit may export several symbols, but an output path
    # must have only one recipe. Auto-suffix default names on conflict;
    # explicit names are never silently changed.
    registry = get_compile_arg("_iron_kernel_registry", ExternalFunction._instances)
    for existing in registry:
        if (
            existing.object_file_name == object_file_name
            and existing.object_file._source != self.object_file._source
        ):
            if object_file_name_explicit:
                raise ValueError(
                    f"ExternalFunction '{effective_name}' would collide with "
                    f"an already-registered instance: same "
                    f"explicit object_file_name='{object_file_name}' but "
                    f"different compile_flags / source.  Distinguish them "
                    f"by passing a distinct `object_file_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 = replace(self.object_file, name=object_file_name)
            self._cached_digest = None
            break
    for existing in registry:
        if existing.object_file_name == self.object_file_name:
            if existing.object_file._source != self.object_file._source:
                raise ValueError(
                    f"ExternalFunction '{effective_name}' would collide on '{self.object_file_name}'"
                )
            self._object_file = existing.object_file
            break
    registry.add(self)

built_for_arch property writable

built_for_arch: str | None

The arch a kernels/ factory built this for, or None when unknown.

Stored on the shared KernelObject, so a sibling binding from object_file.bind(...) carries it as well. An artifact that factories built for more than one arch reads as unknown.

source_file property

source_file: str | None

Path to the C/C++ source on disk, or None for inline source.

source_string property

source_string: str | None

Inline C/C++ source text, or None when compiled from a file.

include_dirs property

include_dirs: list[str]

Copy of the extra -I directories passed to the compiler.

compile_flags property

compile_flags: list[str]

Copy of the extra flags passed verbatim to the compiler.

use_chess property

use_chess: bool

True when this kernel's object is built with xchesscc, not Peano.

recipe_key property

recipe_key: str

The same string on any machine that builds this kernel from the same recipe.

It covers the symbol, the source's name and content, the argument types, the compile flags and the toolchain choice (chess, inline), but not the headers the source includes, where any of them live, or the symbol prefix, which a design adds. A path in a flag under the kernel sources, the runtime library or the installed headers counts relative to that root, so -I and -DAIE_LUT_KERNEL_SOURCE="..." flags name no installation; any other path counts whole. The nightly kernel timings carry it, so a design that calls a kernel can find that kernel's timed row.

unregistered staticmethod

unregistered() -> AbstractContextManager[dict]

Return a context whose new kernels no build compiles.

A kernel made only to be read, for its frame_bytes or its contract, otherwise joins the registry a build in progress compiles.

Returns:

Type	Description
AbstractContextManager[dict]	The context manager.

Source code in python/iron/kernel.py

@staticmethod
def unregistered() -> AbstractContextManager[dict]:
    """Return a context whose new kernels no build compiles.

    A kernel made only to be read, for its ``frame_bytes`` or its
    contract, otherwise joins the registry a build in progress compiles.

    Returns:
        The context manager.
    """
    return compile_context(_iron_kernel_registry=set())

check_target_arch

check_target_arch(target_arch: str) -> None

Raise unless this kernel may compile for target_arch.

A kernels/ factory picks its source, flags and contract for the arch of the device bound when it runs, so compiling the result for another arch fails here rather than in Peano. A kernel no factory built accepts any.

Raises:

Type	Description
ValueError	When the kernel was built for another architecture, which happens when its factory ran before the device was bound.

Source code in python/iron/kernel.py

def check_target_arch(self, target_arch: str) -> None:
    """Raise unless this kernel may compile for ``target_arch``.

    A kernels/ factory picks its source, flags and contract for the arch of
    the device bound when it runs, so compiling the result for another arch
    fails here rather than in Peano. A kernel no factory built accepts any.

    Raises:
        ValueError: When the kernel was built for another architecture,
            which happens when its factory ran before the device was bound.
    """
    built = self.built_for_arch
    if built is not None and built != target_arch:
        raise ValueError(
            f"kernel {self.name} was built for {built} but this design "
            f"compiles for {target_arch}: its factory ran with an {built} "
            "device bound (or none, which reads as aie2). Call the factory "
            "inside the design, or bind the device first with "
            "iron.set_current_device()"
        )

build_for

build_for(arch: str | None = None)

Return the Build of this kernel that ran on arch.

The arch is the device's, by default the bound one's, not the one this kernel's factory saw: a kernel returned before any device was bound describes the kernel a design built for the device it then ran on. The compile flags say whether it was built portable.

Parameters:

Name	Type	Description	Default
arch	str | None	The architecture the output came from.	None

Raises:

Type	Description
ValueError	When arch is None and no device is bound.

Source code in python/iron/kernel.py

def build_for(self, arch: str | None = None):
    """Return the ``Build`` of this kernel that ran on ``arch``.

    The arch is the device's, by default the bound one's, not the one this
    kernel's factory saw: a kernel returned before any device was bound
    describes the kernel a design built for the device it then ran on. The
    compile flags say whether it was built portable.

    Args:
        arch: The architecture the output came from.

    Raises:
        ValueError: When ``arch`` is None and no device is bound.
    """
    # kernels._common imports this module.
    from .kernels._common import Build, _bound_arch

    arch = arch or _bound_arch()
    if arch is None:
        raise ValueError(
            f"kernel '{self.name}' is judged by the build that produced its "
            "output; bind the device it ran on, or pass arch="
        )
    return Build(arch, "-DAIE_KERNELS_PORTABLE" in self.compile_flags)

contract_for

contract_for(arch: str | None = None)

Return the contract as it judges this kernel's output from arch.

A contract with per_build rows is resolved for the build that produced the output: arch, by default the bound device's, and this kernel's own portable flag. Any other contract is returned as is.

Parameters:

Name	Type	Description	Default
arch	str | None	The architecture the output came from.	None

Raises:

Type	Description
ValueError	When the kernel has no contract, or its contract is per build and neither arch nor a bound device names one.

Source code in python/iron/kernel.py

def contract_for(self, arch: str | None = None):
    """Return the contract as it judges this kernel's output from ``arch``.

    A contract with ``per_build`` rows is resolved for the build that
    produced the output: ``arch``, by default the bound device's, and this
    kernel's own portable flag. Any other contract is returned as is.

    Args:
        arch: The architecture the output came from.

    Raises:
        ValueError: When the kernel has no contract, or its contract is
            per build and neither ``arch`` nor a bound device names one.
    """
    c = self._require_contract()
    return c.for_build(self.build_for(arch)) if c.per_build else c

param_values

param_values(inputs: list) -> list

Pick the Param arrays out of one logical input list.

inputs is one array per unbound In/tensor Param in argument order. A design bakes tensor Param arguments into core buffers rather than streaming them, so it needs them separately.

Source code in python/iron/kernel.py

def param_values(self, inputs: list) -> list:
    """Pick the ``Param`` arrays out of one logical input list.

    ``inputs`` is one array per unbound ``In``/tensor ``Param`` in argument
    order. A design bakes tensor ``Param`` arguments into core buffers
    rather than streaming them, so it needs them separately.
    """
    from .kernels._common import Param, _is_tensor_type

    c = self._require_contract()
    types = self.arg_types()
    c.validate_types(types)
    positions = [i for i in c.reference_indices() if _is_tensor_type(types[i])]
    if len(inputs) != len(positions):
        raise ValueError(f"{self.name}: expected {len(positions)} input arrays")
    return [np.asarray(a) for a, i in zip(inputs, positions) if c.roles[i] is Param]

input_limit

input_limit(
    dtype, *, reduction: int | None = None
) -> int | None

Largest integer magnitude an input may take without overflowing.

From the contract's acc_dtype and reduction (reduction overrides the per-call value, e.g. with the full K of a tiled matmul): with two or more multiplied inputs every product of two limits summed reduction times must fit the accumulator with a factor-4 margin; with one input the sum of reduction limits must. None for float inputs, or when the contract declares no accumulator.

The output dtype does not bound this. What a kernel does when a result leaves the output range is its reference's to model, and clipping inputs to the output range would leave a requantizing kernel's data near zero.

Source code in python/iron/kernel.py

def input_limit(self, dtype, *, reduction: int | None = None) -> int | None:
    """Largest integer magnitude an input may take without overflowing.

    From the contract's ``acc_dtype`` and ``reduction`` (``reduction``
    overrides the per-call value, e.g. with the full ``K`` of a tiled
    matmul): with two or more multiplied inputs every product of two
    limits summed ``reduction`` times must fit the accumulator with a
    factor-4 margin; with one input the sum of ``reduction`` limits must.
    ``None`` for float inputs, or when the contract declares no
    accumulator.

    The output dtype does not bound this. What a kernel does when a
    result leaves the output range is its reference's to model, and
    clipping inputs to the output range would leave a requantizing
    kernel's data near zero.
    """
    from aie.utils.compile.jit.markers import In

    from .kernels._common import Param, _is_tensor_type

    c = self._require_contract()
    types = self.arg_types()
    c.validate_types(types)
    dt = np.dtype(dtype)
    if not np.issubdtype(dt, np.integer):
        return None
    if c.acc_dtype is None or not np.issubdtype(np.dtype(c.acc_dtype), np.integer):
        return None
    n = reduction or c.reduction or 1
    budget = np.iinfo(c.acc_dtype).max // 4
    n_tensors = sum(
        r in (In, Param) and _is_tensor_type(t) for r, t in zip(c.roles, types)
    )
    limit = int(np.sqrt(budget // n)) if n_tensors >= 2 else budget // n
    return max(1, min(limit, int(np.iinfo(dt).max)))

expected

expected(
    inputs: list,
    *,
    scalars: tuple = (),
    arch: str | None = None
)

Return reference output(s), cast to each output argument's dtype.

arch names the architecture the output to judge came from; see contract_for.

Source code in python/iron/kernel.py

def expected(self, inputs: list, *, scalars: tuple = (), arch: str | None = None):
    """Return reference output(s), cast to each output argument's dtype.

    ``arch`` names the architecture the output to judge came from; see
    [`contract_for`][iron.kernel.ExternalFunction.contract_for].
    """
    c = self.contract_for(arch)
    if c.reference is None:
        raise ValueError(f"{self.name}: contract has no reference")
    result = c.reference(*self._reference_args(inputs, scalars))
    multiple = len(c.out_indices) > 1
    results = result if multiple else (result,)
    if multiple and (
        not isinstance(results, tuple) or len(results) != len(c.out_indices)
    ):
        raise ValueError("reference must return one tuple entry per output")
    outputs = []
    for i, value in zip(c.out_indices, results):
        dt = self.arg_dtype(i)
        outputs.append(
            np.asarray(value).astype(np.float32 if is_block_float(dt) else dt)
        )
    return tuple(outputs) if multiple else outputs[0]

output_dtype

output_dtype(ref_dtype=None)

Host dtype(s) of the device output buffer(s).

Defaults to the declared argument dtypes, with block-floating-point types represented as packed bytes. The optional reference dtype override is retained for compatibility. Multiple outputs return a tuple in argument order.

Source code in python/iron/kernel.py

def output_dtype(self, ref_dtype=None):
    """Host dtype(s) of the device output buffer(s).

    Defaults to the declared argument dtypes, with block-floating-point
    types represented as packed bytes. The optional reference dtype
    override is retained for compatibility. Multiple outputs return a
    tuple in argument order.
    """
    outputs = self._require_contract().out_indices
    multiple = len(outputs) > 1
    dtypes = (
        tuple(self.arg_dtype(i) for i in outputs)
        if ref_dtype is None
        else ref_dtype if multiple else (ref_dtype,)
    )
    if multiple and (
        not isinstance(dtypes, (tuple, list)) or len(dtypes) != len(outputs)
    ):
        raise ValueError("provide one reference dtype per output")
    result = tuple(
        np.uint8 if is_block_float(self.arg_dtype(i)) else dt
        for i, dt in zip(outputs, dtypes)
    )
    return result if multiple else result[0]

judge

judge(
    got,
    ref,
    *,
    calls: int = 1,
    tolerance=None,
    inputs: list | None = None,
    scalars: tuple = (),
    arch: str | None = None
)

Compare a flat device output against a reference under the contract.

Declared layouts decode each output into logical tiles. DMA padding is trimmed per call. One Verdict summarizes all outputs and is false if any output fails; its detail identifies the failing output. Without streamed inputs, a complete one-call reference may be repeated. With no streamed inputs, a one-tile reference is repeated across calls. A tolerance that is a function of the inputs needs the inputs and scalars the reference was given. Without tolerance, the contract's is that of the build arch names; see contract_for.

Source code in python/iron/kernel.py

def judge(
    self,
    got,
    ref,
    *,
    calls: int = 1,
    tolerance=None,
    inputs: list | None = None,
    scalars: tuple = (),
    arch: str | None = None,
):
    """Compare a flat device output against a reference under the contract.

    Declared layouts decode each output into logical tiles. DMA padding
    is trimmed per call. One Verdict summarizes all outputs and is false
    if any output fails; its detail identifies the failing output.
    Without streamed inputs, a complete one-call reference may be repeated.
    With no streamed inputs, a one-tile reference is repeated across calls.
    A tolerance that is a function of the inputs needs the ``inputs`` and
    ``scalars`` the reference was given. Without ``tolerance``, the
    contract's is that of the build ``arch`` names; see
    [`contract_for`][iron.kernel.ExternalFunction.contract_for].
    """
    from aie.utils.compile.jit.markers import In
    from aie.utils.verify import Tolerance, Verdict, compare

    c = self._require_contract()
    multiple = len(c.out_indices) > 1
    actuals, references = (got, ref) if multiple else ((got,), (ref,))
    if (
        not isinstance(actuals, (tuple, list))
        or not isinstance(references, (tuple, list))
        or len(actuals) != len(c.out_indices)
        or len(references) != len(c.out_indices)
    ):
        raise ValueError("provide one actual and reference array per output")
    tol = tolerance or self.contract_for(arch).tolerance
    bounds = (None,) * len(c.out_indices)
    if tol is not None and tol.kind == "bound":
        if inputs is None:
            raise ValueError(
                f"{self.name}: its tolerance is a function of the inputs; "
                "pass the reference's inputs= (and scalars=) to judge"
            )
        bounds = tol.bound(*self._reference_args(inputs, scalars))
        bounds = bounds if multiple else (bounds,)
    verdicts = []
    for i, actual, reference, bound in zip(
        c.out_indices, actuals, references, bounds
    ):
        layout = c.layouts[i] if c.layouts else None
        got = layout.decode(actual, calls=calls) if layout else np.asarray(actual)
        got, ref = got.reshape(calls, -1), np.asarray(reference)
        if c.out_valid is not None:
            got = got[:, : c.out_valid]
        if In not in c.roles and ref.size == got.shape[1]:
            ref = np.broadcast_to(ref.reshape(1, -1), got.shape)
        else:
            ref = ref.reshape(calls, -1)
        verdicts.append(
            compare(
                got,
                ref,
                tol or Tolerance.default_for(ref.dtype),
                range_axis=1,
                bound=None if bound is None else np.reshape(bound, ref.shape),
            )
        )
    if not multiple:
        return verdicts[0]
    first_bad, offset = None, 0
    for result in verdicts:
        if first_bad is None and result.first_bad_index is not None:
            first_bad = offset + result.first_bad_index
        offset += result.n_checked
    ulps = [v.max_ulp_err for v in verdicts if v.max_ulp_err is not None]
    return Verdict(
        ok=all(verdicts),
        n_checked=sum(v.n_checked for v in verdicts),
        n_mismatch=sum(v.n_mismatch for v in verdicts),
        max_abs_err=max(v.max_abs_err for v in verdicts),
        max_ulp_err=max(ulps) if ulps else None,
        first_bad_index=first_bad,
        detail="; ".join(
            f"output {i} (argument {arg}): {v.detail}"
            for i, (arg, v) in enumerate(zip(c.out_indices, verdicts))
        ),
    )

frame_bytes

frame_bytes(
    target_arch: str,
    object_cache: KernelObjectCache | None = None,
    entry: str | None = None,
) -> int

Return the core data memory this kernel takes beyond its arguments.

Parameters:

Name	Type	Description	Default
target_arch	str	"aie2" or "aie2p".	required
object_cache	KernelObjectCache | None	The store it compiles through, so the design's build reuses the object; the one under NPU_CACHE_HOME when None.	None
entry	str | None	A bound Kernel's symbol in this object to take the stack below.	None

Returns:

Type	Description
int	Its object's static data plus its stack, or plus a declared
int	stack_size_override for its own entry; 0 for what measure
int	cannot read.

Source code in python/iron/kernel.py

def frame_bytes(
    self,
    target_arch: str,
    object_cache: KernelObjectCache | None = None,
    entry: str | None = None,
) -> int:
    """Return the core data memory this kernel takes beyond its arguments.

    Args:
        target_arch: ``"aie2"`` or ``"aie2p"``.
        object_cache: The store it compiles through, so the design's build
            reuses the object; the one under ``NPU_CACHE_HOME`` when None.
        entry: A bound ``Kernel``'s symbol in this object to take the stack below.

    Returns:
        Its object's static data plus its stack, or plus a declared
        ``stack_size_override`` for its own entry; 0 for what ``measure``
        cannot read.
    """
    found = self.measure(
        target_arch,
        object_cache
        or KernelObjectCache(
            NPU_CACHE_HOME / "objects", COMPILE_LOCK_TIMEOUT_SECONDS
        ),
        entry=entry,
    )
    data = found.data if found else 0
    if self.stack_size_override is not None and entry is None:
        return data + self.stack_size_override
    return data + ((found.stack if found else None) or 0)

measure

measure(
    target_arch: str,
    object_cache: KernelObjectCache | None = None,
    embed_bitcode: bool = False,
    entry: str | None = None,
) -> Linked | None

Compile this kernel and read its object from its entry.

Parameters:

Name	Type	Description	Default
target_arch	str	"aie2" or "aie2p".	required
object_cache	KernelObjectCache | None	The design's KernelObjectCache; None compiles in a scratch directory.	None
embed_bitcode	bool	Keep the object's LLVM IR, as the design's build does for --check-lut-banks.	False
entry	str | None	A bound Kernel's symbol in this object to read from.	None

Returns:

Type	Description
Linked | None	Its Linked; None for a kernel aiecc cannot measure either: a
Linked | None	Chess or merge-mode kernel, or any kernel when no Peano is
Linked | None	installed.

Raises:

Type	Description
ValueError	Its factory built it for another architecture (see check_target_arch).

Source code in python/iron/kernel.py

def measure(
    self,
    target_arch: str,
    object_cache: KernelObjectCache | None = None,
    embed_bitcode: bool = False,
    entry: str | None = None,
) -> Linked | None:
    """Compile this kernel and read its object from its entry.

    Args:
        target_arch: ``"aie2"`` or ``"aie2p"``.
        object_cache: The design's ``KernelObjectCache``; None compiles
            in a scratch directory.
        embed_bitcode: Keep the object's LLVM IR, as the design's build
            does for ``--check-lut-banks``.
        entry: A bound ``Kernel``'s symbol in this object to read from.

    Returns:
        Its ``Linked``; None for a kernel aiecc cannot measure either: a
        Chess or merge-mode kernel, or any kernel when no Peano is
        installed.

    Raises:
        ValueError: Its factory built it for another architecture (see
            ``check_target_arch``).
    """
    self.check_target_arch(target_arch)
    try:
        config.peano_cxx_path()
    except RuntimeError:
        return None
    if self.use_chess or self.link_with_mode is not None:
        return None
    with tempfile.TemporaryDirectory() as tmp:
        compile_external_kernel(
            self,
            tmp,
            target_arch,
            embed_bitcode=embed_bitcode,
            object_cache=object_cache,
        )
        return linked(Path(tmp) / self.object_file_name, entry or self.name)

ScratchpadParameter

A design that declares ScratchpadParameters and compiles to a full ELF (@iron.jit(full_elf=True)) takes their values per call: design(out, parameters={"foo": bfloat16(3.0), "bar": bfloat16(4.0)}). The values go through the run's ctrl scratchpad, laid out by the build's own params.txt. A design without ScratchpadParameters refuses parameters=, and so do the HSA and HRX runtimes.

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

ScratchpadParameter

ScratchpadParameter(name: str, dtype: type[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	type[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: type[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._site = SourceSite.capture()
    self._name = name
    self._dtype = dtype
    self._resolved = False
    self._attributes: dict[str, str] = {}

name property

name: str

The symbol name of this parameter.

dtype property

dtype: type[NpuDType]

The numpy scalar type of this parameter.

attributes property

attributes: dict[str, str]

The declaration's attributes beyond its name and type, as MLIR text.

from_declaration classmethod

from_declaration(
    op: ScratchpadParameterOp,
) -> ScratchpadParameter

Return the parameter an aiex.scratchpad_parameter op declares.

It is declared again with every attribute of op, so one that --aie-lower-scratchpad-parameters already lowered keeps its kind, state_table_idx, bounds and joint bounds.

Parameters:

Name	Type	Description	Default
op	ScratchpadParameterOp	The aiex.scratchpad_parameter declaration.	required

Returns:

Type	Description
ScratchpadParameter	The parameter, named and typed as op declares it.

Raises:

Type	Description
TypeError	The declared type has no NumPy dtype.

Source code in python/iron/scratchpad_parameter.py

@classmethod
def from_declaration(cls, op: ScratchpadParameterOp) -> "ScratchpadParameter":
    """Return the parameter an ``aiex.scratchpad_parameter`` op declares.

    It is declared again with every attribute of ``op``, so one that
    ``--aie-lower-scratchpad-parameters`` already lowered keeps its
    ``kind``, ``state_table_idx``, bounds and joint bounds.

    Args:
        op: The ``aiex.scratchpad_parameter`` declaration.

    Returns:
        The parameter, named and typed as ``op`` declares it.

    Raises:
        TypeError: The declared type has no NumPy dtype.
    """
    type_ = ir.TypeAttr(op.attributes["type"]).value
    dtype = mlir_type_to_np_dtype(type_)
    if dtype is None:
        raise TypeError(f"No NumPy dtype for scratchpad parameter {type_}.")
    parameter = cls(
        ir.StringAttr(op.attributes["sym_name"]).value, cast(type[NpuDType], dtype)
    )
    parameter._attributes = {
        name: str(op.attributes[name])
        for name in op.attributes
        if name not in ("sym_name", "type")
    }
    return 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)
        op = aiex.scratchpad_parameter(  # pyright: ignore[reportAttributeAccessIssue]
            self._name,
            mlir_type,
            loc=loc or self._site.location(self._name),
            ip=ip,
        )
        for name, text in self._attributes.items():
            op.attributes[name] = ir.Attribute.parse(text)
        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	Compile a design on a cache miss, then run it on the attached NPU.
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	classes	As annotations they mark a design's inputs and outputs; instantiated in an operator's class body, they declare its tensors.
iron.CompileTime	marker	Type-annotation marker for a compile-time constant argument.
iron.DispatchTime	marker	Integer scalar that can vary per call without recompiling the device program.

For dispatch scalar defaults, specialization, and runtime binding, see Dispatch-time scalars in the runtime data-movement guide.

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; under with, restore the previous one on exit.
iron.ensure_current_device	Raise if no device is currently selected.

The Device a Program targets also reports the hardware limits a design is sized against, such as max_lock_value, max_repeat_count, dma_task_queue_depth and get_num_bds(tile_type), so a design can read them instead of hardcoding them.

Device

Device(device: AIEDevice)

Bases: Resolvable

A representation of a device of a specific type.

Provides device metadata and emits aie.logical_tile operations for Tile objects. A DeviceConfiguration reconstructs this descriptor before it emits an aie.device, so the descriptor does not own one configuration's IR.

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

def __init__(self, device: AIEDevice) -> None:
    self._device = device
    self._tm = get_target_model(device)
    self._resolved_tiles: dict[int, LogicalTileOp] = {}

name property

name: str

The device's name in the dialect: "npu2", "npu1_1col", ...

A name, not a family: a four-column NPU2 is "npu2_4col". Ask :attr:arch which architecture a device is.

cols property

cols: int

Number of columns in the device tile array.

rows property

rows: int

Number of rows in the device tile array.

arch property

arch: AIEArch

AIE architecture of the device (AIE1, AIE2, or AIE2p).

can_dispatch_full_elf property

can_dispatch_full_elf: bool

Whether a full ELF (PDIs and control code in one ELF) dispatches here: on AIE2p.

core_memory_bytes property

core_memory_bytes: int

Data memory local to one compute tile.

core_memory_banks property

core_memory_banks: int

Banks a compute tile's data memory is divided into, as the buffer allocator places buffers.

Each is two interleaved hardware banks on AIE2 and AIE2P (AM020, AM027).

core_memory_bank_bytes property

core_memory_bank_bytes: int

Bytes in one of core_memory_banks.

default_core_stack_bytes property

default_core_stack_bytes: int

Stack a Worker gets when nothing it calls declares a larger need.

core_rows property

core_rows: list[int]

Rows of compute tiles, bottom to top.

core_dma_channels_in property

core_dma_channels_in: int

Input DMA channels a compute tile has, and so the most fifos one can be fed.

core_dma_channels_out property

core_dma_channels_out: int

Output DMA channels available on a compute tile.

shim_dma_channels_in property

shim_dma_channels_in: int

DMA channels the shim tiles feed the array through, summed over them.

How many streams from the host the device carries at once.

shim_dma_channels_out property

shim_dma_channels_out: int

DMA channels the shim tiles drain the array through, summed over them.

address_gen_granularity property

address_gen_granularity: int

Address-generation granularity of the device, in bits.

max_lock_value property

max_lock_value: int

Largest value a lock register holds.

max_repeat_count property

max_repeat_count: int

Largest repeat count one DMA task queue push carries (0 = none).

A task started with a larger repeat count is issued as several pushes of the same task.

dma_task_queue_depth property

dma_task_queue_depth: int

How many tasks one DMA channel's task queue holds (0 = no queue).

get_tile_type

get_tile_type(col, row) -> AIETileType

Return the AIETileType for the given device coordinates.

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

def get_tile_type(self, col, row) -> AIETileType:
    """Return the AIETileType for the given device coordinates."""
    self._validate_coordinates(col, row)
    return AIETileType(self._tm.get_tile_type(col, row))

get_dma_bd_wrap_bits

get_dma_bd_wrap_bits(col, row) -> int

Wrap (size) field width, in bits.

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

def get_dma_bd_wrap_bits(self, col, row) -> int:
    """Wrap (size) field width, in bits."""
    self._validate_coordinates(col, row)
    return self._tm.get_dma_bd_wrap_bits(col, row)

get_dma_bd_step_bits

get_dma_bd_step_bits(col, row) -> int

Step (stride) field width, in bits. The field counts address granules.

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

def get_dma_bd_step_bits(self, col, row) -> int:
    """Step (stride) field width, in bits. The field counts address granules."""
    self._validate_coordinates(col, row)
    return self._tm.get_dma_bd_step_bits(col, row)

get_dma_bd_iter_bits

get_dma_bd_iter_bits(col, row) -> int

Return the iteration (repeat) field width, in bits.

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

def get_dma_bd_iter_bits(self, col, row) -> int:
    """Return the iteration (repeat) field width, in bits."""
    self._validate_coordinates(col, row)
    return self._tm.get_dma_bd_iter_bits(col, row)

dma_bd_fits

dma_bd_fits(
    col: int,
    row: int,
    tap: TensorAccessPattern,
    dtype: type,
) -> bool

Whether one DMA buffer descriptor of the tile at (col, row) holds tap.

Decided as the DMA lowerings verify a descriptor: a linear transfer, or a contiguous one on a shim tile, in linear mode past the wrap limits, and any other within the wrap, step and iteration limits; either way, the three inner dimensions within the tile's buffer length field. On a shim tile the tap is laid out as shim_dma_single_bd_task lays it out: coalesced, and a leading re-read kept in the iteration dimension. Elsewhere a tap of fewer than four dimensions is padded in front with unit ones. The descriptor's length is its three inner dimensions, and where that is all of it, its unit dimensions go as aie-normalize-dma-bd-dims drops them, which can move a large one into a wrap field.

Parameters:

Name	Type	Description	Default
col	int	Column of the tile.	required
row	int	Row of the tile.	required
tap	TensorAccessPattern	The access pattern, in elements of dtype.	required
dtype	type	The numpy element type the pattern moves.	required

Returns:

Name	Type	Description
bool	bool	False for more than four dimensions (after coalescing, on a
	bool	shim tile), padding on a shim tile, an offset that is not whole
	bool	address granules, or a pattern past the descriptor's limits.

Raises:

Type	Description
ValueError	The coordinates are outside the device.

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

def dma_bd_fits(
    self, col: int, row: int, tap: TensorAccessPattern, dtype: type
) -> bool:
    """Whether one DMA buffer descriptor of the tile at (`col`, `row`) holds `tap`.

    Decided as the DMA lowerings verify a descriptor: a linear transfer,
    or a contiguous one on a shim tile, in linear mode past the wrap
    limits, and any other within the wrap, step and iteration limits;
    either way, the three inner dimensions within the tile's buffer
    length field. On a shim tile the tap is laid out as
    `shim_dma_single_bd_task` lays it out: coalesced, and a leading
    re-read kept in the iteration dimension. Elsewhere a tap of fewer
    than four dimensions is padded in front with unit ones. The
    descriptor's length is its three inner dimensions, and where that
    is all of it, its unit dimensions go as `aie-normalize-dma-bd-dims`
    drops them, which can move a large one into a wrap field.

    Args:
        col: Column of the tile.
        row: Row of the tile.
        tap: The access pattern, in elements of `dtype`.
        dtype: The numpy element type the pattern moves.

    Returns:
        bool: False for more than four dimensions (after coalescing, on a
        shim tile), padding on a shim tile, an offset that is not whole
        address granules, or a pattern past the descriptor's limits.

    Raises:
        ValueError: The coordinates are outside the device.
    """
    self._validate_coordinates(col, row)
    if self._tm.is_shim_noc_or_pl_tile(col, row):
        if tap.padding is not None:
            return False
        offset, sizes, strides = shim_dims(tap)
    else:
        pad = 4 - len(tap.sizes)
        offset = tap.offset
        sizes = [1] * pad + list(tap.sizes)
        strides = [0] * pad + list(tap.strides)
    if len(sizes) > 4:
        return False
    # A fresh context costs ~4 ms, and resolving a design calls this thousands of times.
    context = Device._bd_context
    if context is None:
        context = Device._bd_context = ir.Context()
    with context, ir.Location.unknown():
        return dma_bd_fits(
            self._device.value,
            col,
            row,
            np_dtype_to_mlir_type(np.dtype(dtype).type),
            offset,
            sizes,
            strides,
            math.prod(sizes[1:]),
        )

get_num_bds

get_num_bds(tile_type: AIETileType) -> int

Return how many DMA buffer descriptors (BDs) a tile of tile_type has.

The BD budget is per tile TYPE, not per coordinate, and it is not uniform: on AIE2, a MemTile has 48 BDs while a CoreTile and a ShimNOCTile both have 16 — the same count for two different roles. Callers must name the tile type they mean rather than hardcode a BD count, since "16" is only right for two of the three types and silently wrong for the third.

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

def get_num_bds(self, tile_type: AIETileType) -> int:
    """Return how many DMA buffer descriptors (BDs) a tile of ``tile_type`` has.

    The BD budget is per tile TYPE, not per coordinate, and it is not
    uniform: on AIE2, a MemTile has 48 BDs while a CoreTile and a
    ShimNOCTile both have 16 — the same count for two different roles.
    Callers must name the tile type they mean rather than hardcode a
    BD count, since "16" is only right for two of the three types and
    silently wrong for the third.
    """
    for row in range(self.rows):
        for col in range(self.cols):
            if self.get_tile_type(col, row) == tile_type:
                return self._tm.get_num_bds(col, row)
    raise ValueError(f"Device has no tile of type {tile_type!r}")

---

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 aie.iron API — 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. A Flow given no DMA channels lets the compiler assign them. endpoint(tile) on either returns the FlowEndpoint on tile, which a DmaChannel takes in place of a channel index and which can run a runtime-sequence task. A Flow or PacketFlow with one shim end has fill / drain; PacketFlow.fill stamps the packet header on the input.

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, and FlowEndpoint, one end of a Flow as the DMA on that end's tile sees it. Flow and PacketFlow share how they name their ends and how the runtime sequence fills or drains a shim end.

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.

FlowEndpoint

FlowEndpoint(flow: _Route, end: int)

Bases: DmaEndpoint

One end of a Flow or PacketFlow, seen from the DMA on that end's tile.

Obtained from Flow.endpoint or PacketFlow.endpoint. The end knows its tile, its direction (MM2S at the source, S2MM at a destination) and its channel: the one the route gives, or else the one the compiler assigns. Like any DmaEndpoint it stands in for a DmaChannel's channel and builds runtime-sequence tasks with task.

It is not an ObjectFifo endpoint: those are the Workers and runtime a fifo attaches to and places, whereas this names one DMA channel of a route between tiles the design already has.

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

def __init__(self, flow: _Route, end: int):
    channel = flow._channels[end]
    super().__init__(
        flow._endpoint_tiles()[end],
        DMAChannelDir.MM2S if end == 0 else DMAChannelDir.S2MM,
        channel.channel if isinstance(channel, FlowEndpoint) else channel,
    )
    self._flow = flow
    self._end = end

symbol property

symbol: str | None

The aie.route_endpoint this end lowers to.

None for a Flow that gives both channels, which lowers to an aie.flow with no endpoint symbols.

tile property

tile: Tile

The tile whose DMA owns this channel.

direction property

direction: DMAChannelDir

S2MM into the tile's memory, MM2S out of it.

channel property

channel: int | None

The channel index, or None if the compiler assigns it.

sync

sync() -> None

Wait until this end's tasks have issued their completion tokens.

Called in a Runtime sequence body, as DmaEndpoint.sync is.

A shim end the sequence waits on is driven at run time, so it gets its aie.shim_dma_allocation, as an end a fill or drain uses does, and no fifo is given the channel. That is the end a route's other runtime traffic programs, such as a gather's descriptors pushed by control packets.

Raises:

Type	Description
ValueError	The route is not registered with the runtime, or the tile is not placed or the channel is not given.

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

def sync(self) -> None:
    """Wait until this end's tasks have issued their completion tokens.

    Called in a [`Runtime`][iron.Runtime] sequence body, as
    ``DmaEndpoint.sync`` is.

    A shim end the sequence waits on is driven at run time, so it gets its
    ``aie.shim_dma_allocation``, as an end a ``fill`` or ``drain`` uses
    does, and no fifo is given the channel. That is the end a route's
    other runtime traffic programs, such as a gather's descriptors pushed
    by control packets.

    Raises:
        ValueError: The route is not registered with the runtime, or the
            tile is not placed or the channel is not given.
    """
    flow = self._flow
    if flow not in active_sequence().runtime.flows:
        raise ValueError(
            f"{type(flow).__name__} must be registered with rt.add_flow(flow) "
            "before the runtime sequence waits on its end."
        )
    super().sync()
    if self.tile.effective_tile_type in _SHIM_TILE_TYPES:
        owner, owner_end = flow._owner(self._end)
        owner._shim_ends.add(owner_end)

task

task(
    *bds: "Bd | Buffer",
    runs=1,
    wait: bool = False,
    out_of_order: bool = False
) -> "TileDmaTask"

Configure a runtime-sequence task that walks bds on this channel.

Call from within a Runtime sequence body on a mem or core tile's channel (a shim channel moves host memory: use fill/drain). The buffer descriptors are written here, at the call; start pushes the task onto the channel queue, as often as needed, and free returns its descriptors after the last start.

load = into.endpoint(mem).task(
    Bd(resident, tap=TensorAccessPattern.full((N,))[:n])
)
load.start()
...
load.start()
load.free()

Each Bd is spelled as in a TileDma, but without next, and its tap may hold dispatch-time values. An iteration takes the descriptor's outermost dimension, so its tap gets one dimension fewer. A buffer that has no tile yet is placed on this one.

Parameters:

Name	Type	Description	Default
*bds	'Bd | Buffer'	The descriptors, walked in order as one task; a bare Buffer stands for Bd(buffer).	()
runs	int | Value	How many times the whole chain runs per start. Defaults to 1.	1
wait	bool	Issue a completion token, so the task can be awaited.	False
out_of_order	bool	Run an S2MM channel in out-of-order mode, as DmaChannel.out_of_order does: a Bd with no bd_id takes its position in bds, and runs counts packets. Needs a channel given by index.	False

Returns:

Type	Description
'TileDmaTask'	The configured TileDmaTask.

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

def task(
    self,
    *bds: "Bd | Buffer",
    runs=1,
    wait: bool = False,
    out_of_order: bool = False,
) -> "TileDmaTask":
    """Configure a runtime-sequence task that walks ``bds`` on this channel.

    Call from within a [`Runtime`][iron.Runtime] sequence body on a mem or
    core tile's channel (a shim channel moves host memory: use
    ``fill``/``drain``). The buffer descriptors are written here, at the
    call; [`start`][iron.runtime.dmataskhandle.Task.start] pushes the task
    onto the channel queue, as often as needed, and
    [`free`][iron.runtime.dmataskhandle.Task.free] returns its descriptors
    after the last start.

    ```python
    load = into.endpoint(mem).task(
        Bd(resident, tap=TensorAccessPattern.full((N,))[:n])
    )
    load.start()
    ...
    load.start()
    load.free()
    ```

    Each [`Bd`][iron.Bd] is spelled as in a [`TileDma`][iron.TileDma], but
    without ``next``, and its ``tap`` may hold dispatch-time values. An
    ``iteration`` takes the descriptor's outermost dimension, so its
    ``tap`` gets one dimension fewer. A buffer that has no tile yet is
    placed on this one.

    Args:
        *bds: The descriptors, walked in order as one task; a bare
            [`Buffer`][iron.Buffer] stands for ``Bd(buffer)``.
        runs (int | Value): How many times the whole chain runs per start.
            Defaults to 1.
        wait: Issue a completion token, so the task can be awaited.
        out_of_order: Run an S2MM channel in out-of-order mode, as
            [`DmaChannel.out_of_order`][iron.DmaChannel] does: a Bd with
            no ``bd_id`` takes its position in ``bds``, and ``runs`` counts
            packets. Needs a channel given by index.

    Returns:
        The configured [`TileDmaTask`][iron.TileDmaTask].
    """
    active = active_sequence()
    if self.tile.effective_tile_type in _SHIM_TILE_TYPES:
        raise ValueError(
            f"{self} is a shim channel, which moves host memory; use the "
            "Flow's fill()/drain() for it."
        )
    if not bds:
        raise ValueError(f"A task on {self} needs at least one Bd.")
    bds = tuple(bd if isinstance(bd, Bd) else Bd(bd) for bd in bds)
    for i, bd in enumerate(bds):
        if bd.next is not None:
            raise ValueError(
                f"Bd {i} of a task on {self} sets next={bd.next!r}; a task "
                "runs its Bds in order and ends after the last."
            )
        if bd.buffer.place(self.tile) != self.tile:
            raise ValueError(
                f"A task on {self} was given a buffer on {bd.buffer.tile}; a "
                "tile's DMA can only address buffers on that tile."
            )
        active.resolve_in_device(bd.buffer)
        for use in (*bd.acquires, *bd.releases):
            active.resolve_in_device(use.lock)
    if out_of_order and self.channel is None:
        raise ValueError(
            f"An out-of-order task needs a channel given by index, not the "
            f"compiler-assigned {self}."
        )

    if isinstance(runs, (int, np.integer)):
        repeat = dict(repeat_count=int(runs) - 1)
    else:
        repeat = dict(repeat_count_val=runs - 1)
    # aiex.dma_configure_task_for has no out_of_order.
    operand = self.channel if out_of_order else self._operand()
    if isinstance(operand, str):
        op = dma_configure_task_for(operand, issue_token=wait, **repeat)
    else:
        op = dma_configure_task(
            self.tile.op,
            self.direction,
            operand,
            issue_token=wait,
            out_of_order=out_of_order or None,
            **repeat,
        )
    with bd_blocks(op) as block:
        for i, bd in enumerate(bds):
            with block[i]:
                bd._emit(i if out_of_order and bd.bd_id is None else bd.bd_id)
                if i + 1 < len(bds):
                    next_bd(block[i + 1])
                else:
                    EndOp()
    return TileDmaTask(self, op.result)

Flow

Flow(
    src: Tile,
    dst: Tile | Sequence[Tile],
    *,
    src_port: WireBundle = DMA,
    src_channel: int | None = None,
    dst_port: WireBundle = DMA,
    dst_channel: int | None = None,
    vias: Sequence[
        tuple[
            Tile,
            tuple[WireBundle, int],
            tuple[WireBundle, int],
        ]
    ] = (),
    shim_symbol: str | None = None,
    name: str | None = None
)

Bases: _Route

An explicit AXI-stream route from a source to one or more destinations.

Connects (src_tile, src_port, src_channel) to (dst_tile, dst_port, dst_channel). With both channels given, it lowers to a single aie.flow op, and the user arranges matching TileDma channels on the producer and consumer ends.

A channel left as None (the default) is assigned by the compiler: the Flow then lowers to one aie.route_endpoint per end and an aie.route, and a DMA program reaches an end through endpoint rather than an index. That is also how a list of destinations (a circuit-switched broadcast) lowers.

Construct a Flow.

Parameters:

Name	Type	Description	Default
src	Tile	The source tile.	required
dst	Tile | Sequence[Tile]	The destination tile, or several to broadcast to.	required
src_port	WireBundle	The source port bundle. Defaults to DMA.	DMA
src_channel	int | None	The source channel. None (default) lets the compiler assign a DMA channel.	None
dst_port	WireBundle	The destination port bundle. Defaults to DMA.	DMA
dst_channel	int | None	The channel at every destination. None (default) lets the compiler assign each a DMA channel.	None
vias	Sequence[tuple[Tile, tuple[WireBundle, int], tuple[WireBundle, int]]]	Exact (tile, ingress, egress) switchbox hops. Each port is a (WireBundle, channel) tuple. Vias require one destination and explicit source and destination channels.	()
shim_symbol	str | None	Name the runtime sequence reaches the shim end by, in place of the one built from name. Only needed to refer to the channel from elsewhere (e.g. a raw shim_dma_single_bd_task("symbol", ...)); fill/drain name it themselves. Direction is inferred: shim-as-source → MM2S, shim-as-dest → S2MM. When both ends are shim tiles it names the source; a drain at the destination uses {name}_dst.	None
name	str | None	Base of the end symbols, {name}_src and {name}_dst ({name}_dst{i} for a broadcast). Left None, the configuration the Flow is registered with names it flow{n} when it resolves, counting in registration order.	None

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

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

    Args:
        src (Tile): The source tile.
        dst (Tile | Sequence[Tile]): The destination tile, or several to
            broadcast to.
        src_port (WireBundle): The source port bundle.  Defaults to DMA.
        src_channel (int | None): The source channel. ``None`` (default)
            lets the compiler assign a DMA channel.
        dst_port (WireBundle): The destination port bundle.  Defaults to DMA.
        dst_channel (int | None): The channel at every destination. ``None``
            (default) lets the compiler assign each a DMA channel.
        vias: Exact ``(tile, ingress, egress)`` switchbox hops. Each port is
            a ``(WireBundle, channel)`` tuple. Vias require one destination
            and explicit source and destination channels.
        shim_symbol (str | None): Name the runtime sequence reaches the
            shim end by, in place of the one built from ``name``. Only
            needed to refer to the channel from elsewhere (e.g. a raw
            ``shim_dma_single_bd_task("symbol", ...)``); ``fill``/``drain``
            name it themselves. Direction is inferred: shim-as-source →
            MM2S, shim-as-dest → S2MM. When both ends are shim tiles it
            names the source; a drain at the destination uses
            ``{name}_dst``.
        name (str | None): Base of the end symbols, ``{name}_src`` and
            ``{name}_dst`` (``{name}_dst{i}`` for a broadcast). Left
            ``None``, the configuration the Flow is registered with names
            it ``flow{n}`` when it resolves, counting in registration order.
    """
    self._site = SourceSite.capture()
    self._broadcast = not isinstance(dst, Tile)
    self._dsts: list[Tile] = [dst] if isinstance(dst, Tile) else list(dst)
    if not self._dsts:
        raise ValueError("Flow needs at least one destination.")
    self._vias = list(vias)
    if self._vias and (
        self._broadcast or src_channel is None or dst_channel is None
    ):
        raise ValueError(
            "Flow vias require one destination and explicit source and "
            "destination channels."
        )
    for port, channel, end in (
        (src_port, src_channel, "src"),
        (dst_port, dst_channel, "dst"),
    ):
        if channel is None and port != WireBundle.DMA:
            raise ValueError(
                f"Flow {end}_port={port} needs an explicit {end}_channel; "
                "the compiler only assigns DMA channels."
            )
    self._src = src
    self._src_port = src_port
    self._dst_port = dst_port
    self._channels = [src_channel, *([dst_channel] * len(self._dsts))]
    self._shim_symbol = shim_symbol
    self._shim_ends: set[int] = set()
    self._name = name
    self._endpoints: dict[int, FlowEndpoint] = {}
    self._op = None

name property

name: str | None

The name the route's end symbols are built from.

None for a route left unnamed until its configuration resolves.

fill

fill(source, **kwargs)

Send data from the source runtime buffer into this route.

Call from within a Runtime sequence body, on a Flow whose source is a shim tile; a destination may be a shim too. See emit_shim_transfer for the keyword arguments; returns a Task handle to the transfer.

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

def fill(self, source, **kwargs):
    """Send data from the ``source`` runtime buffer into this route.

    Call from within a [`Runtime`][iron.Runtime] sequence body, on a Flow
    whose source is a shim tile; a destination may be a shim too. See
    ``emit_shim_transfer`` for the keyword arguments; returns a
    [`Task`][iron.runtime.dmataskhandle.Task] handle to the transfer.
    """
    return self._transfer(source, DMAChannelDir.MM2S, **kwargs)

tiles

tiles() -> list

Tiles this Resolvable depends on for code generation.

Override this in user-side Resolvable subclasses that reference tiles which aren't already discoverable via Workers or ObjectFifos. The DeviceConfiguration resolves these tiles before calling resolve, so tile.op is valid by then. Default: empty list.

Source code in python/iron/resolvable.py

def tiles(self) -> list:
    """Tiles this Resolvable depends on for code generation.

    Override this in user-side Resolvable subclasses that reference tiles
    which aren't already discoverable via Workers or ObjectFifos. The
    DeviceConfiguration resolves these tiles before calling `resolve`, so
    `tile.op` is valid by then. Default: empty list.
    """
    return []

control_tiles

control_tiles() -> list[Tile]

Return the destination tiles whose TileControl port this route writes.

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

def control_tiles(self) -> list[Tile]:
    """Return the destination tiles whose TileControl port this route writes."""
    return [
        tile
        for tile, port in zip(self._dsts, self.dst_ports)
        if port == WireBundle.TileControl
    ]

endpoint

endpoint(tile: Tile) -> FlowEndpoint

Return this route's end on tile.

Pass it as a DmaChannel's channel, or build a runtime-sequence task on it with task. Its channel is the one given to the route, or else the one the compiler assigns. An end that shares another route's channel returns that route's end, so one channel is one endpoint.

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

def endpoint(self, tile: Tile) -> "FlowEndpoint":
    """Return this route's end on ``tile``.

    Pass it as a [`DmaChannel`][iron.DmaChannel]'s ``channel``, or build a
    runtime-sequence task on it with
    [`task`][iron.dataflow.tile_dma.DmaEndpoint.task]. Its channel is the
    one given to the route, or else the one the compiler assigns. An end
    that shares another route's channel returns that route's end, so one
    channel is one endpoint.
    """
    ends = [i for i, t in enumerate(self._endpoint_tiles()) if t == tile]
    if len(ends) != 1:
        raise ValueError(
            f"{tile} is {'not an end' if not ends else 'more than one end'} "
            f"of this {type(self).__name__}."
        )
    owner, end = self._owner(ends[0])
    if end not in owner._endpoints:
        owner._endpoints[end] = FlowEndpoint(owner, end)
    return owner._endpoints[end]

drain

drain(dest, **kwargs)

Receive data from this route into the dest runtime buffer.

Call from within a Runtime sequence body, on a route whose one destination is a shim tile; the source may be a shim too. See emit_shim_transfer for the keyword arguments; returns a Task handle to the transfer.

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

def drain(self, dest, **kwargs):
    """Receive data from this route into the ``dest`` runtime buffer.

    Call from within a [`Runtime`][iron.Runtime] sequence body, on a route
    whose one destination is a shim tile; the source may be a shim too. See
    ``emit_shim_transfer`` for the keyword arguments; returns a
    [`Task`][iron.runtime.dmataskhandle.Task] handle to the transfer.
    """
    return self._transfer(dest, DMAChannelDir.S2MM, **kwargs)

PacketDest dataclass

PacketDest(
    tile: Tile,
    port: WireBundle = DMA,
    channel: int | None | FlowEndpoint = 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 | None | FlowEndpoint = 0,
    dst_port: WireBundle = DMA,
    dst_channel: int | None | FlowEndpoint = 0,
    extra_dsts: Sequence[PacketDest] = (),
    keep_pkt_header: bool = False,
    shim_symbol: str | None = None,
    name: str | None = None
)

Bases: _Route

An explicit packet-switched route from a source to one or more destinations.

Connects (src_tile, src_port, src_channel) to each destination endpoint, tagging the stream with pkt_id. The user arranges matching TileDma channels on the producer and consumer ends, which endpoint names.

Each end's channel is an index, None for the compiler to assign, or another PacketFlow's end on the same tile and side, whose channel this end then shares: several flows merging into one receiving channel, or one channel sending several ids. With every channel an index it lowers to one aie.packet_flow; otherwise to an aie.route as a Flow does.

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 | None | FlowEndpoint	Source channel: an index (default 0), None for the compiler to assign, or another PacketFlow's source end on src to share.	0
dst_port	WireBundle	Destination port bundle (as for Flow).	DMA
dst_channel	int | None | FlowEndpoint	Destination channel, as src_channel is, sharing another PacketFlow's destination end on dst.	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 — emit a matching aie.shim_dma_allocation of that name at the shim source, else at the one shim destination.	None
name	str | None	Same meaning as on Flow; left None, it is named packetflow{n}.	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 | None | FlowEndpoint" = 0,
    dst_port: WireBundle = WireBundle.DMA,
    dst_channel: "int | None | FlowEndpoint" = 0,
    extra_dsts: Sequence[PacketDest] = (),
    keep_pkt_header: bool = False,
    shim_symbol: str | None = None,
    name: 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: an index (default 0), ``None`` for
            the compiler to assign, or another PacketFlow's source end on
            ``src`` to share.
        dst_port: Destination port bundle (as for [`Flow`][iron.Flow]).
        dst_channel: Destination channel, as ``src_channel`` is, sharing
            another PacketFlow's destination end on ``dst``.
        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] — emit a
            matching `aie.shim_dma_allocation` of that name at the shim
            source, else at the one shim destination.
        name: Same meaning as on [`Flow`][iron.Flow]; left ``None``, it is
            named ``packetflow{n}``.
    """
    self._site = SourceSite.capture()
    self._pkt_id = pkt_id
    self._src = src
    self._dsts = [dst, *(d.tile for d in extra_dsts)]
    self._src_port = src_port
    self._dst_port = dst_port
    self._channels = [src_channel, dst_channel, *(d.channel for d in extra_dsts)]
    ports = [src_port, dst_port, *(d.port for d in extra_dsts)]
    for end, (tile, port, channel) in enumerate(
        zip(self._endpoint_tiles(), ports, self._channels)
    ):
        if isinstance(channel, int):
            continue
        if port != WireBundle.DMA:
            raise ValueError(
                f"PacketFlow end on {tile} with port={port} needs a channel "
                "index; the compiler only assigns DMA channels."
            )
        if isinstance(channel, FlowEndpoint) and not (
            isinstance(channel._flow, PacketFlow)
            and channel.tile == tile
            and (channel._end == 0) == (end == 0)
            and channel.channel is None
        ):
            raise ValueError(
                f"PacketFlow end on {tile} can share only another "
                "PacketFlow's end on this same Tile object, on the same "
                f"side, whose channel the compiler assigns, not {channel}."
            )
    self._extra_dsts: list[PacketDest] = list(extra_dsts)
    self._endpoints: dict[int, FlowEndpoint] = {}
    self._keep_pkt_header = keep_pkt_header
    self._route_attrs = dict(
        packet=(0, pkt_id), keep_pkt_header=keep_pkt_header or None
    )
    self._shim_symbol = shim_symbol
    self._shim_ends: set[int] = set()
    self._name = name
    self._op = None

name property

name: str | None

The name the route's end symbols are built from.

None for a route left unnamed until its configuration resolves.

fill

fill(source, pkt_type: int = 0, **kwargs)

Send data from the source runtime buffer into this route.

Call from within a Runtime sequence body, on a PacketFlow whose src is a shim tile. The shim stamps every packet with pkt_type and this route's pkt_id, so several PacketFlows leaving one shim channel each fill their own route. See emit_shim_transfer for the keyword arguments; returns a Task handle to the transfer.

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

def fill(self, source, pkt_type: int = 0, **kwargs):
    """Send data from the ``source`` runtime buffer into this route.

    Call from within a [`Runtime`][iron.Runtime] sequence body, on a
    PacketFlow whose src is a shim tile. The shim stamps every packet with
    ``pkt_type`` and this route's ``pkt_id``, so several PacketFlows leaving
    one shim channel each fill their own route. See ``emit_shim_transfer``
    for the keyword arguments; returns a
    [`Task`][iron.runtime.dmataskhandle.Task] handle to the transfer.
    """
    return self._transfer(
        source, DMAChannelDir.MM2S, packet=(pkt_type, self._pkt_id), **kwargs
    )

tiles

tiles() -> list

Tiles this Resolvable depends on for code generation.

Override this in user-side Resolvable subclasses that reference tiles which aren't already discoverable via Workers or ObjectFifos. The DeviceConfiguration resolves these tiles before calling resolve, so tile.op is valid by then. Default: empty list.

Source code in python/iron/resolvable.py

def tiles(self) -> list:
    """Tiles this Resolvable depends on for code generation.

    Override this in user-side Resolvable subclasses that reference tiles
    which aren't already discoverable via Workers or ObjectFifos. The
    DeviceConfiguration resolves these tiles before calling `resolve`, so
    `tile.op` is valid by then. Default: empty list.
    """
    return []

control_tiles

control_tiles() -> list[Tile]

Return the destination tiles whose TileControl port this route writes.

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

def control_tiles(self) -> list[Tile]:
    """Return the destination tiles whose TileControl port this route writes."""
    return [
        tile
        for tile, port in zip(self._dsts, self.dst_ports)
        if port == WireBundle.TileControl
    ]

all_tiles

all_tiles()

Return the tiles this route touches — Program uses this to resolve them.

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

def all_tiles(self):
    """Return the tiles this route touches — Program uses this to resolve them."""
    return self._endpoint_tiles()

endpoint

endpoint(tile: Tile) -> FlowEndpoint

Return this route's end on tile.

Pass it as a DmaChannel's channel, or build a runtime-sequence task on it with task. Its channel is the one given to the route, or else the one the compiler assigns. An end that shares another route's channel returns that route's end, so one channel is one endpoint.

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

def endpoint(self, tile: Tile) -> "FlowEndpoint":
    """Return this route's end on ``tile``.

    Pass it as a [`DmaChannel`][iron.DmaChannel]'s ``channel``, or build a
    runtime-sequence task on it with
    [`task`][iron.dataflow.tile_dma.DmaEndpoint.task]. Its channel is the
    one given to the route, or else the one the compiler assigns. An end
    that shares another route's channel returns that route's end, so one
    channel is one endpoint.
    """
    ends = [i for i, t in enumerate(self._endpoint_tiles()) if t == tile]
    if len(ends) != 1:
        raise ValueError(
            f"{tile} is {'not an end' if not ends else 'more than one end'} "
            f"of this {type(self).__name__}."
        )
    owner, end = self._owner(ends[0])
    if end not in owner._endpoints:
        owner._endpoints[end] = FlowEndpoint(owner, end)
    return owner._endpoints[end]

drain

drain(dest, **kwargs)

Receive data from this route into the dest runtime buffer.

Call from within a Runtime sequence body, on a route whose one destination is a shim tile; the source may be a shim too. See emit_shim_transfer for the keyword arguments; returns a Task handle to the transfer.

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

def drain(self, dest, **kwargs):
    """Receive data from this route into the ``dest`` runtime buffer.

    Call from within a [`Runtime`][iron.Runtime] sequence body, on a route
    whose one destination is a shim tile; the source may be a shim too. See
    ``emit_shim_transfer`` for the keyword arguments; returns a
    [`Task`][iron.runtime.dmataskhandle.Task] handle to the transfer.
    """
    return self._transfer(dest, DMAChannelDir.S2MM, **kwargs)

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: PerDeviceConfigurationResolvable

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. DeviceConfiguration.resolve() walks its 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._site = SourceSite.capture()
    self._src = src
    self._dst = dst
    # The source registration makes the edge part of its configuration's graph.
    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,
        loc=loc or self._site.location(),
        ip=ip,
    )

TileDma / DmaChannel / Bd / DmaEndpoint

Explicit tile DMA programs: TileDma, DmaChannel, buffer descriptors (Bd), and the Acquire / Release lock actions. A DmaEndpoint names one channel of one tile; endpoint.task(*bds) configures a TileDmaTask on a mem or core tile from the runtime sequence, so its descriptors can change per dispatch.

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.

A DmaEndpoint names one DMA channel of a tile. It is what a DmaChannel runs on when the compiler assigns the channel, and it builds the runtime-sequence TileDmaTasks that reprogram a mem or core tile's DMA from inside the sequence body.

Acquire dataclass

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

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

In a runtime-sequence task, value may be a dispatch-time value; the instruction stream is rejected if it falls outside the lock's range (at least 1 for greater_equal).

Release dataclass

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

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

In a runtime-sequence task, value may be a dispatch-time value.

Bd dataclass

Bd(
    buffer: Buffer,
    tap: TensorAccessPattern | None = None,
    acquires: list[Acquire] = list(),
    releases: list[Release] = list(),
    next: int | str | None = None,
    packet: tuple[int, int] | None = None,
    bd_id: int | None = None,
    iteration: BdIteration | None = None,
    out_of_order_id: int | None = None,
)

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:

• None (default) — follow the channel: the next entry in bds, and from the last entry either back to the head or out of the chain, per DmaChannel.loop.
• "self" — the BD loops to itself, whatever the rest of the chain does.
• 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.

next is ignored on an out-of-order channel because those BDs are chained only for configuration and the hardware selects by header id (see DmaChannel.out_of_order).

DmaEndpoint

DmaEndpoint(
    tile: Tile,
    direction: DMAChannelDir,
    channel: int | None,
)

One DMA channel of a tile: its tile, direction and channel.

DmaEndpoint(tile, direction, channel) pins a channel by index. A Flow's endpoint is one too, whose channel the compiler may assign instead.

Pass one as a DmaChannel's channel, or build a runtime-sequence task on it with task.

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

def __init__(self, tile: Tile, direction: DMAChannelDir, channel: int | None):
    self._tile = tile
    self._direction = direction
    self._channel = channel

tile property

tile: Tile

The tile whose DMA owns this channel.

direction property

direction: DMAChannelDir

S2MM into the tile's memory, MM2S out of it.

channel property

channel: int | None

The channel index, or None if the compiler assigns it.

symbol property

symbol: str | None

The aie.route_endpoint a compiler-assigned channel lowers to.

None when the channel is given by index.

sync

sync() -> None

Wait until this channel's tasks have issued their completion tokens.

Called in a Runtime sequence body, it emits aiex.npu_sync on its tile and channel.

Raises:

Type	Description
ValueError	The tile is not placed or the channel is not given.

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

def sync(self) -> None:
    """Wait until this channel's tasks have issued their completion tokens.

    Called in a [`Runtime`][iron.Runtime] sequence body, it emits
    ``aiex.npu_sync`` on its tile and channel.

    Raises:
        ValueError: The tile is not placed or the channel is not given.
    """
    tile = self.tile
    if tile.col is None or tile.row is None or self.channel is None:
        raise ValueError(
            f"sync() waits on a placed tile's channel by index; {self} is not one."
        )
    npu_sync(tile.col, tile.row, self.direction.value, self.channel)

task

task(
    *bds: "Bd | Buffer",
    runs=1,
    wait: bool = False,
    out_of_order: bool = False
) -> "TileDmaTask"

Configure a runtime-sequence task that walks bds on this channel.

Call from within a Runtime sequence body on a mem or core tile's channel (a shim channel moves host memory: use fill/drain). The buffer descriptors are written here, at the call; start pushes the task onto the channel queue, as often as needed, and free returns its descriptors after the last start.

load = into.endpoint(mem).task(
    Bd(resident, tap=TensorAccessPattern.full((N,))[:n])
)
load.start()
...
load.start()
load.free()

Each Bd is spelled as in a TileDma, but without next, and its tap may hold dispatch-time values. An iteration takes the descriptor's outermost dimension, so its tap gets one dimension fewer. A buffer that has no tile yet is placed on this one.

Parameters:

Name	Type	Description	Default
*bds	'Bd | Buffer'	The descriptors, walked in order as one task; a bare Buffer stands for Bd(buffer).	()
runs	int | Value	How many times the whole chain runs per start. Defaults to 1.	1
wait	bool	Issue a completion token, so the task can be awaited.	False
out_of_order	bool	Run an S2MM channel in out-of-order mode, as DmaChannel.out_of_order does: a Bd with no bd_id takes its position in bds, and runs counts packets. Needs a channel given by index.	False

Returns:

Type	Description
'TileDmaTask'	The configured TileDmaTask.

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

def task(
    self,
    *bds: "Bd | Buffer",
    runs=1,
    wait: bool = False,
    out_of_order: bool = False,
) -> "TileDmaTask":
    """Configure a runtime-sequence task that walks ``bds`` on this channel.

    Call from within a [`Runtime`][iron.Runtime] sequence body on a mem or
    core tile's channel (a shim channel moves host memory: use
    ``fill``/``drain``). The buffer descriptors are written here, at the
    call; [`start`][iron.runtime.dmataskhandle.Task.start] pushes the task
    onto the channel queue, as often as needed, and
    [`free`][iron.runtime.dmataskhandle.Task.free] returns its descriptors
    after the last start.

    ```python
    load = into.endpoint(mem).task(
        Bd(resident, tap=TensorAccessPattern.full((N,))[:n])
    )
    load.start()
    ...
    load.start()
    load.free()
    ```

    Each [`Bd`][iron.Bd] is spelled as in a [`TileDma`][iron.TileDma], but
    without ``next``, and its ``tap`` may hold dispatch-time values. An
    ``iteration`` takes the descriptor's outermost dimension, so its
    ``tap`` gets one dimension fewer. A buffer that has no tile yet is
    placed on this one.

    Args:
        *bds: The descriptors, walked in order as one task; a bare
            [`Buffer`][iron.Buffer] stands for ``Bd(buffer)``.
        runs (int | Value): How many times the whole chain runs per start.
            Defaults to 1.
        wait: Issue a completion token, so the task can be awaited.
        out_of_order: Run an S2MM channel in out-of-order mode, as
            [`DmaChannel.out_of_order`][iron.DmaChannel] does: a Bd with
            no ``bd_id`` takes its position in ``bds``, and ``runs`` counts
            packets. Needs a channel given by index.

    Returns:
        The configured [`TileDmaTask`][iron.TileDmaTask].
    """
    active = active_sequence()
    if self.tile.effective_tile_type in _SHIM_TILE_TYPES:
        raise ValueError(
            f"{self} is a shim channel, which moves host memory; use the "
            "Flow's fill()/drain() for it."
        )
    if not bds:
        raise ValueError(f"A task on {self} needs at least one Bd.")
    bds = tuple(bd if isinstance(bd, Bd) else Bd(bd) for bd in bds)
    for i, bd in enumerate(bds):
        if bd.next is not None:
            raise ValueError(
                f"Bd {i} of a task on {self} sets next={bd.next!r}; a task "
                "runs its Bds in order and ends after the last."
            )
        if bd.buffer.place(self.tile) != self.tile:
            raise ValueError(
                f"A task on {self} was given a buffer on {bd.buffer.tile}; a "
                "tile's DMA can only address buffers on that tile."
            )
        active.resolve_in_device(bd.buffer)
        for use in (*bd.acquires, *bd.releases):
            active.resolve_in_device(use.lock)
    if out_of_order and self.channel is None:
        raise ValueError(
            f"An out-of-order task needs a channel given by index, not the "
            f"compiler-assigned {self}."
        )

    if isinstance(runs, (int, np.integer)):
        repeat = dict(repeat_count=int(runs) - 1)
    else:
        repeat = dict(repeat_count_val=runs - 1)
    # aiex.dma_configure_task_for has no out_of_order.
    operand = self.channel if out_of_order else self._operand()
    if isinstance(operand, str):
        op = dma_configure_task_for(operand, issue_token=wait, **repeat)
    else:
        op = dma_configure_task(
            self.tile.op,
            self.direction,
            operand,
            issue_token=wait,
            out_of_order=out_of_order or None,
            **repeat,
        )
    with bd_blocks(op) as block:
        for i, bd in enumerate(bds):
            with block[i]:
                bd._emit(i if out_of_order and bd.bd_id is None else bd.bd_id)
                if i + 1 < len(bds):
                    next_bd(block[i + 1])
                else:
                    EndOp()
    return TileDmaTask(self, op.result)

TileDmaTask

TileDmaTask(endpoint: DmaEndpoint, handle)

Bases: Task

A runtime-sequence DMA task on one channel of a mem or core tile.

Built by DmaEndpoint.task, already configured. It is a Task: start() pushes it (again), await_() waits for it (needs wait=True), free() returns its descriptors, and it rides a range_ iter_args entry across loop iterations.

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

def __init__(self, endpoint: DmaEndpoint, handle):
    super().__init__(handle)
    self._endpoint = endpoint

endpoint property

endpoint: DmaEndpoint

The channel this task runs on.

DmaChannel dataclass

DmaChannel(
    direction: DMAChannelDir,
    channel: int | DmaEndpoint,
    bds: list[Bd],
    pad_value: int = 0,
    repeat_count: int = 0,
    out_of_order: bool = False,
    loop: bool = True,
)

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 | DmaEndpoint	hardware channel index, or a DmaEndpoint such as Flow.endpoint's, to run on the channel the compiler assigns that end. An endpoint must be on the program's tile and point in direction.	required
bds	list[Bd]	ordered list of Bd entries that form the chain (in-order) or n-way merge (out-of-order).	required
repeat_count	int	extra repeats of the task (0 = run once), where the task is the BD chain (in-order) or a merge round (out-of-order). Only meaningful on a chain that ends -- see loop.	0
loop	bool	whether the last BD chains back to the first (the default), making the chain endless. An endless chain is one task that never completes: it runs for as long as its locks let it, which is how ObjectFifo expresses the same thing, and repeat_count has nothing to count and is ignored. loop=False ends the chain after its last BD, making it a task that completes and can be re-run -- which is what gives repeat_count meaning, and what a design reproducing a specific descriptor layout wants. Note a chain that ends runs exactly repeat_count + 1 times, so a loop=False channel expected to move more than one buffer needs a matching count; left at 0 it moves one and stops.	True
out_of_order	bool	put the channel into out-of-order mode (S2MM only). Each BD receives the packet with bd.bd_id == pkt.out_of_order_id, and the BD chain (next bd) is ignored. Each BD receives its own BdIteration.size packets per merge round. Every BD must be packet-enabled and the ingress flow must set keep_pkt_header=True. Multiple out-of-order channels must have disjoint BD ids. At the hardware level, repeat_count is the total number of packets to accept (0-based). This class converts repeated merge rounds to that total.	False

TileDma

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

Bases: PerDeviceConfigurationResolvable

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._site = SourceSite.capture()
    self._tile = tile
    self._channels: list[DmaChannel] = []
    self.add_channels(channels)
    self._resolved = False

add_channel

add_channel(channel: DmaChannel) -> None

Add a channel to this tile's DMA program.

A tile has one DMA program, so a helper that wires transfers one at a time needs somewhere to put the second channel it wants on a tile it has already reached.

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

def add_channel(self, channel: DmaChannel) -> None:
    """Add a channel to this tile's DMA program.

    A tile has one DMA program, so a helper that wires transfers one at a
    time needs somewhere to put the second channel it wants on a tile it has
    already reached.
    """
    self.add_channels([channel])

add_channels

add_channels(channels: Iterable[DmaChannel]) -> None

Add channels after checking all hardware channel keys.

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

def add_channels(self, channels: Iterable[DmaChannel]) -> None:
    """Add channels after checking all hardware channel keys."""
    channels = list(channels)
    keys = {channel._key for channel in self._channels}
    for channel in channels:
        if isinstance(channel.channel, DmaEndpoint):
            channel.channel._check_on(self._tile, channel.direction)
        key = channel._key
        if key in keys:
            raise ValueError(
                f"TileDma for {self._tile} already has "
                f"{channel.direction} channel {channel.channel}."
            )
        keys.add(key)
    self._channels.extend(channels)

all_tiles

all_tiles()

Return this DMA's tile plus the tiles of the Buffers and Locks it uses.

A mem tile DMA may use a neighbor's lock or buffer, whose tile must be resolved before that lock or buffer is.

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

def all_tiles(self):
    """Return this DMA's tile plus the tiles of the Buffers and Locks it uses.

    A mem tile DMA may use a neighbor's lock or buffer, whose tile must
    be resolved before that lock or buffer is.
    """
    tiles = [self._tile]
    bufs, locks = self.all_buffers_and_locks()
    for b in bufs:
        if b.tile is not None and b.tile not in tiles:
            tiles.append(b.tile)
    for lk in locks:
        if lk.tile not in tiles:
            tiles.append(lk.tile)
    return tiles

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: PerDeviceConfigurationResolvable

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. Defaults to None (unnamed).	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. Defaults to None
            (unnamed).
    """
    self._site = SourceSite.capture()
    self._tile = tile
    self._lock_id = lock_id
    self._init = init
    self._name = name
    self._op = None

acquire

acquire(value: int | Value = 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 | ir.Value = 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 | Value = 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 | ir.Value = 1) -> None:
    """Emit `aie.use_lock(self, Acquire, value=value)` (exact match)."""
    _use_lock(self.op, LockAction.Acquire, value=value)

release

release(value: int | Value = 1) -> None

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

Source code in python/iron/lock.py

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

set

set(value: int | Value) -> None

Emit aiex.set_lock(self, value) from a runtime sequence body.

Overwrites the lock's value from the host side, e.g. to re-arm a producer lock before a runtime-sequence DMA chain starts reusing a buffer. The write is not ordered against anything the array is doing, so pair it with a blocking op (an await, or a lock the core waits on) that makes it safe.

A core cannot assign a lock: its lock instructions only add to or subtract from the value, which is what acquire/release emit. The aiex.set_lock verifier rejects a call outside a runtime sequence and a constant outside [0, Device.max_lock_value]; a dispatch-time value outside that range makes the instruction stream fail to build.

Source code in python/iron/lock.py

def set(self, value: int | ir.Value) -> None:
    """Emit `aiex.set_lock(self, value)` from a runtime sequence body.

    Overwrites the lock's value from the host side, e.g. to re-arm a
    producer lock before a runtime-sequence DMA chain starts reusing a
    buffer. The write is not ordered against anything the array is doing,
    so pair it with a blocking op (an await, or a lock the core waits on)
    that makes it safe.

    A core cannot assign a lock: its lock instructions only add to or
    subtract from the value, which is what `acquire`/`release` emit. The
    `aiex.set_lock` verifier rejects a call outside a runtime sequence and
    a constant outside `[0, Device.max_lock_value]`; a dispatch-time value
    outside that range makes the instruction stream fail to build.
    """
    _set_lock_value(self.op, value)

Runtime tasks

Lower-level runtime task types scheduled by the Runtime.

RuntimeTask

RuntimeTask(task_group: TaskGroup | None = None)

Bases: Resolvable

A sequence-local operation resolved while one Runtime body executes.

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()

A software pipeline keeps a step's transfers in flight while it issues the next step's: pipelined hands out one group per step and finishes each a step later, with the same body whether the step count is a Python int or a dispatch-time value.

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
    # Set once the group's transfers have been handed to a loop (as an
    # iter_arg or a yield); the object is then spent.
    self._carried = False
    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

    if self._carried:
        raise RuntimeError(
            f"{self} was carried into a loop; finish the group the loop "
            "hands back (its body argument or result), not this one"
        )
    here = InsertionPoint.current.block
    for task, _ in self._transfers():
        self._check_reaches(_block_of(task), here)
    active_sequence().finish_task_group(self)

pipelined classmethod

pipelined(
    n, depth: int = 2
) -> Iterator[tuple[Any, TaskGroup]]

Run n steps, each in a fresh group, keeping depth groups in flight.

Each iteration yields (step, group). Put the step's transfers in group; it is finished once the next depth - 1 steps have been issued, and the last groups after the last step. depth=1 finishes each step before the next starts.

for step, tg in TaskGroup.pipelined(n_steps):
    in_h.fill(a, tap=in_tiles[step], group=tg)
    out_h.drain(c, tap=out_tiles[step], group=tg, wait=True)

With a Python int the steps unroll. With a staged n the sequence stays rolled: the first depth - 1 steps run under if_(n > step) with step an int, and the rest in one range_ that carries the groups in flight, with step its index. So the body is traced depth times, and every step must issue the same transfers, waited the same way, in the same order.

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

@classmethod
def pipelined(cls, n, depth: int = 2) -> Iterator[tuple[Any, TaskGroup]]:
    """Run ``n`` steps, each in a fresh group, keeping ``depth`` groups in flight.

    Each iteration yields ``(step, group)``. Put the step's transfers in
    ``group``; it is finished once the next ``depth - 1`` steps have been
    issued, and the last groups after the last step. ``depth=1`` finishes
    each step before the next starts.

    ```python
    for step, tg in TaskGroup.pipelined(n_steps):
        in_h.fill(a, tap=in_tiles[step], group=tg)
        out_h.drain(c, tap=out_tiles[step], group=tg, wait=True)
    ```

    With a Python ``int`` the steps unroll. With a staged ``n`` the
    sequence stays rolled: the first ``depth - 1`` steps run under
    ``if_(n > step)`` with ``step`` an ``int``, and the rest in one
    ``range_`` that carries the groups in flight, with ``step`` its
    index. So the body is traced ``depth`` times, and every step must
    issue the same transfers, waited the same way, in the same order.
    """
    if depth < 1:
        raise ValueError(f"pipelined depth must be at least 1, got {depth}")
    if isinstance(n, (int, np.integer)):
        in_flight: list[TaskGroup] = []
        for step in range(int(n)):
            tg = cls()
            yield step, tg
            in_flight.append(tg)
            if len(in_flight) == depth:
                in_flight.pop(0).finish()
        for tg in in_flight:
            tg.finish()
        return
    if not isinstance(n.type, IndexType):
        n = arith.index_cast(IndexType.get(), n)
    yield from cls._pipelined(n, depth, [])

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

DMATask

DMATask(
    alloc: str,
    rt_data: RuntimeData,
    tap: TensorAccessPattern,
    task_group: TaskGroup | None = None,
    wait: bool = False,
    offset_parameter: str | None = None,
    packet: tuple[int, int] | None = None,
    length_parameter: str | None = None,
    length_unit: int | None = None,
    repeat_parameter: str | None = None,
)

Bases: RuntimeTask

Construct a RuntimeTask that will resolve to a DMA Operation.

Parameters:

Name	Type	Description	Default
alloc	str	Name of the shim DMA allocation this transfer drives, i.e. the symbol naming the shim channel. An ObjectFifo with a shim endpoint generates an allocation of its own name; a Flow generates one for its shim_symbol.	required
rt_data	RuntimeData	The Runtime buffer associated with the operation.	required
tap	TensorAccessPattern	The walk over rt_data; its values may be staged.	required
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
length_parameter	str | None	Name of a ScratchpadParameter n; the transfer moves exactly n passes of tap (per iteration). Defaults to None.	None
length_unit	int | None	Elements per unit of length_parameter, a multiple of 16 bytes. Defaults to the elements one pass of tap moves.	None
repeat_parameter	str | None	Name of a ScratchpadParameter n; the transfer runs n + 1 times, at most as many as tap's outer dimensions give (it stops an iteration over them early). Defaults to None.	None
packet	tuple[int, int] | None	Stamp the shim DMA's BD with a packet header (pkt_type, pkt_id). A packet-switched ObjectFifo needs none, since the transfer takes the header its shim allocation was assigned; this is for an explicit PacketFlow or to override that header. Defaults to None.	None

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

def __init__(
    self,
    alloc: str,
    rt_data: RuntimeData,
    tap: TensorAccessPattern,
    task_group: TaskGroup | None = None,
    wait: bool = False,
    offset_parameter: str | None = None,
    packet: tuple[int, int] | None = None,
    length_parameter: str | None = None,
    length_unit: int | None = None,
    repeat_parameter: str | None = None,
):
    """Construct a RuntimeTask that will resolve to a DMA Operation.

    Args:
        alloc (str): Name of the shim DMA allocation this transfer drives,
            i.e. the symbol naming the shim channel. An
            [`ObjectFifo`][iron.ObjectFifo] with a shim endpoint generates an
            allocation of its own name; a [`Flow`][iron.Flow] generates one
            for its ``shim_symbol``.
        rt_data (RuntimeData): The Runtime buffer associated with the operation.
        tap (TensorAccessPattern): The walk over ``rt_data``; its values may be staged.
        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.
        length_parameter (str | None, optional): Name of a ScratchpadParameter
            n; the transfer moves exactly n passes of ``tap`` (per
            iteration). Defaults to None.
        length_unit (int | None, optional): Elements per unit of
            length_parameter, a multiple of 16 bytes. Defaults to the
            elements one pass of ``tap`` moves.
        repeat_parameter (str | None, optional): Name of a
            ScratchpadParameter n; the transfer runs n + 1 times, at most
            as many as ``tap``'s outer dimensions give (it stops an
            iteration over them early). Defaults to None.
        packet (tuple[int, int] | None, optional): Stamp the shim DMA's
            BD with a packet header `(pkt_type, pkt_id)`. A packet-switched
            [`ObjectFifo`][iron.ObjectFifo] needs none, since the transfer
            takes the header its shim allocation was assigned; this is for
            an explicit [`PacketFlow`][iron.PacketFlow] or to override that
            header. Defaults to None.
    """
    self._alloc = alloc
    self._rt_data = rt_data
    self._tap = tap
    self._wait = wait
    self._offset_parameter = offset_parameter
    self._length_parameter = length_parameter
    self._length_unit = length_unit
    self._repeat_parameter = repeat_parameter
    self._packet = packet
    self._task = None
    RuntimeTask.__init__(self, task_group)

task property

task

Return 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

emit_shim_transfer

emit_shim_transfer(
    alloc: str,
    rt_data,
    tap=None,
    wait: bool = False,
    packet: tuple[int, int] | None = None,
    offset_parameter: (
        ScratchpadParameter | str | None
    ) = None,
    group=None,
    managed: bool = True,
    length_parameter: (
        ScratchpadParameter | str | None
    ) = None,
    length_unit: int | None = None,
    repeat_parameter: (
        ScratchpadParameter | str | None
    ) = None,
) -> Task

Emit one shim DMA transfer on the alloc channel, inside the active sequence.

The body shared by the runtime's data-movement verbs -- ObjectFifoHandle.fill/drain and Flow.fill/drain -- which differ only in how they name the shim channel and in the endpoint bookkeeping they do first.

tap is the walk over the runtime buffer; its offset, sizes and strides may be runtime values. When it is None, the whole buffer moves linearly.

When managed is True (default), the transfer is enrolled in a TaskGroup (explicit group or the sequence's implicit one), which awaits/frees it at group close. When False, the caller owns the transfer's lifetime via the returned Task's .free()/.await_() -- used for hand-rolled software pipelines that carry the task across scf.for iterations.

offset_parameter, length_parameter and repeat_parameter take a ScratchpadParameter (or its name). A length_parameter n makes the transfer move exactly n passes of tap (per iteration), and a repeat_parameter n runs it n + 1 times; see DMATask.

Returns:

Name	Type	Description
Task	Task	A handle to the transfer.

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

def emit_shim_transfer(
    alloc: str,
    rt_data,
    tap=None,
    wait: bool = False,
    packet: tuple[int, int] | None = None,
    offset_parameter: ScratchpadParameter | str | None = None,
    group=None,
    managed: bool = True,
    length_parameter: ScratchpadParameter | str | None = None,
    length_unit: int | None = None,
    repeat_parameter: ScratchpadParameter | str | None = None,
) -> Task:
    """Emit one shim DMA transfer on the ``alloc`` channel, inside the active sequence.

    The body shared by the runtime's data-movement verbs --
    ``ObjectFifoHandle.fill``/``drain`` and ``Flow.fill``/``drain`` -- which
    differ only in how they name the shim channel and in the endpoint
    bookkeeping they do first.

    ``tap`` is the walk over the runtime buffer; its offset, sizes and strides
    may be runtime values. When it is None, the whole buffer moves linearly.

    When ``managed`` is True (default), the transfer is enrolled in a TaskGroup
    (explicit ``group`` or the sequence's implicit one), which awaits/frees it at
    group close. When False, the caller owns the transfer's lifetime via the
    returned Task's ``.free()``/``.await_()`` -- used for hand-rolled software
    pipelines that carry the task across ``scf.for`` iterations.

    ``offset_parameter``, ``length_parameter`` and ``repeat_parameter`` take a
    ScratchpadParameter (or its name). A ``length_parameter`` n makes the
    transfer move exactly n passes of ``tap`` (per iteration), and a
    ``repeat_parameter`` n runs it n + 1 times; see DMATask.

    Returns:
        Task: A handle to the transfer.
    """
    active = active_sequence()
    rt = active.runtime

    if not isinstance(rt_data, RuntimeData):
        raise ValueError(f"Expected a RuntimeData source/dest, got {rt_data}")
    if rt_data not in rt._rt_data:
        raise ValueError(
            f"{rt_data} is not a RuntimeData object declared by sequence()"
        )

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

    if not managed and group is not None:
        raise ValueError(
            "An unmanaged transfer (managed=False) is not part of a TaskGroup; "
            "do not also pass group=."
        )

    task = DMATask(
        alloc,
        rt_data,
        tap=tap,
        task_group=group,
        wait=wait,
        offset_parameter=rt.register_parameter(offset_parameter),
        packet=packet,
        length_parameter=rt.register_parameter(length_parameter),
        length_unit=length_unit,
        repeat_parameter=rt.register_parameter(repeat_parameter),
    )
    if managed:
        active.emit_transfer(task, group)
    else:
        # Emit the BD only; the caller owns await/free via the Task.
        task.resolve()
    # Wrap the transfer's !index result: it is both the scf iter_arg payload
    # and the operand dma_await_task/dma_free_task accept.
    return Task(task.task.result)

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 .start() / .free() / .await_() verbs so the loop body does not need to reach for the raw aiex.dma_start_task / 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
    self._lifetime = _Lifetime()

handle property

handle

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

start

start(
    repeat_count: int | Value | None = None,
    repeat_parameter: (
        ScratchpadParameter | str | None
    ) = None,
) -> "Task"

Push this task onto its channel's queue (dma_start_task).

Its buffer descriptors are written when the task is configured, so each start costs one queue push.

A task carried through a range_ iter_args entry can be started only if every value it can carry comes from the same configure (the loop yields it unchanged), since the push names one head buffer descriptor. A task reconfigured inside the loop is started where it is configured, before it is yielded; compilation otherwise fails.

Parameters:

Name	Type	Description	Default
repeat_count	int | Value | None	Replaces the task's configured count for this start only; a runtime value must fit one queue push.	None
repeat_parameter	ScratchpadParameter | str | None	A scratchpad parameter whose value, written per run, is the count instead, at most the task's own; a full ELF's shim and mem tile channels take one.	None

Returns:

Type	Description
'Task'	This task.

Raises:

Type	Description
RuntimeError	If this task was already freed, since its buffer descriptors may since describe another transfer.

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

def start(
    self,
    repeat_count: int | Value | None = None,
    repeat_parameter: ScratchpadParameter | str | None = None,
) -> "Task":
    """Push this task onto its channel's queue (``dma_start_task``).

    Its buffer descriptors are written when the task is configured, so each
    start costs one queue push.

    A task carried through a ``range_`` ``iter_args`` entry can be started
    only if every value it can carry comes from the same configure (the
    loop yields it unchanged), since the push names one head buffer
    descriptor. A task reconfigured inside the loop is started where it is
    configured, before it is yielded; compilation otherwise fails.

    Args:
        repeat_count: Replaces the task's configured count for this start
            only; a runtime value must fit one queue push.
        repeat_parameter: A scratchpad parameter whose value, written per
            run, is the count instead, at most the task's own; a full
            ELF's shim and mem tile channels take one.

    Returns:
        This task.

    Raises:
        RuntimeError: If this task was already freed, since its buffer
            descriptors may since describe another transfer.
    """
    if self._lifetime.freed:
        raise RuntimeError(
            "Task.start() after Task.free(): the freed buffer descriptors "
            "may already describe another transfer. Free a task after its "
            "last start."
        )
    dma_start_task(
        self._handle,
        repeat_count=repeat_count,
        repeat_parameter=(
            None
            if repeat_parameter is None
            else active_sequence().runtime.register_parameter(repeat_parameter)
        ),
    )
    return self

free

free() -> None

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

Raises:

Type	Description
RuntimeError	If this task was already freed.

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

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

    Raises:
        RuntimeError: If this task was already freed.
    """
    if self._lifetime.freed:
        raise RuntimeError("Task.free() called twice on the same task.")
    dma_free_task(self._handle)
    self._lifetime.freed = True

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)