# Copyright (c) Meta Platforms, Inc. and affiliates.
# All rights reserved.
#
# This source code is licensed under the BSD-style license found in the
# LICENSE file in the root directory of this source tree.
"""
Example usage:
.. code-block:: python
from pathlib import Path
import torch
from executorch.runtime import Runtime, Program, Method
et_runtime: Runtime = Runtime.get()
program: Program = et_runtime.load_program(Path("/tmp/program.pte"))
print("Program methods:", program.method_names)
forward: Method = program.load_method("forward")
inputs = (torch.ones(2, 2), torch.ones(2, 2))
outputs = forward.execute(inputs)
print(f"Ran forward({inputs})")
print(f" outputs: {outputs}")
Example output:
.. code-block:: text
Program methods: {'forward'}
Ran forward((tensor([[1., 1.],
[1., 1.]]), tensor([[1., 1.],
[1., 1.]])))
outputs: [tensor([[2., 2.],
[2., 2.]])]
Example usage with ETDump generation:
Note: ETDump requires building ExecuTorch with event tracing enabled
(CMake option ``EXECUTORCH_ENABLE_EVENT_TRACER=ON``).
.. code-block:: python
from pathlib import Path
import os
import torch
from executorch.runtime import Runtime, Program, Method
# Create program with etdump generation enabled
et_runtime: Runtime = Runtime.get()
program: Program = et_runtime.load_program(
Path("/tmp/program.pte"),
enable_etdump=True,
debug_buffer_size=int(1e7), # 10MB buffer to capture all debug info
)
# Load method and execute
forward: Method = program.load_method("forward")
inputs = (torch.ones(2, 2), torch.ones(2, 2))
outputs = forward.execute(inputs)
# Write etdump result to file
etdump_file = "/tmp/etdump_output.etdp"
debug_file = "/tmp/debug_output.bin"
program.write_etdump_result_to_file(etdump_file, debug_file)
# Check that files were created
print(f"ETDump file created: {os.path.exists(etdump_file)}")
print(f"Debug file created: {os.path.exists(debug_file)}")
print("Directory contents:", os.listdir("/tmp"))
Example output:
.. code-block:: text
ETDump file created: True
Debug file created: True
Directory contents: ['program.pte', 'etdump_output.etdp', 'debug_output.bin']
Example usage with backend and operator introspection:
.. code-block:: python
from executorch.runtime import Runtime
runtime = Runtime.get()
# Check available backends
backends = runtime.backend_registry.registered_backend_names
print(f"Available backends: {backends}")
# Check if a specific backend is available
if runtime.backend_registry.is_available("XnnpackBackend"):
print("XNNPACK backend is available")
# List all registered operators
operators = runtime.operator_registry.operator_names
print(f"Number of registered operators: {len(operators)}")
Example output:
.. code-block:: text
Available backends: ['XnnpackBackend', ...] # Depends on your build configuration
XNNPACK backend is available
Number of registered operators: 247 # Depends on linked kernels
"""
import functools
from pathlib import Path
from types import ModuleType
from typing import (
Any,
BinaryIO,
Dict,
List,
Optional,
Sequence,
Set,
TYPE_CHECKING,
Union,
)
if TYPE_CHECKING:
import torch
try:
from executorch.extension.pybindings.portable_lib import ( # type: ignore[import-not-found] # noqa: F401
ExecuTorchMethod,
ExecuTorchProgram,
ExecuTorchResult,
MethodMeta,
Verification,
)
except ModuleNotFoundError as e:
raise ModuleNotFoundError(
"The prebuilt extension module executorch.extension.pybindings._C is not "
"found. Please reinstall ExecuTorch from pip."
) from e
[docs]
class Method:
"""An ExecuTorch method, loaded from a Program.
This can be used to execute the method with inputs.
"""
def __init__(self, method: ExecuTorchMethod) -> None:
self._method = method
[docs]
def execute(
self, inputs: Sequence[Any], clone_outputs: bool = True
) -> Sequence[Any]:
"""Executes the method with the given inputs.
Outputs are copies, except for an output bound with
:meth:`set_output`, which is returned as the bound tensor and
overwritten by later calls. An input that is not memory planned (e.g.
exported with ``alloc_graph_input=False``) is used in place, so a
mutation of it is written into the tensor passed in for it, while an
output that is that input is still returned as a copy. A memory planned
input is copied into the method, so the tensor passed in for it is not
updated.
Args:
inputs: A sequence of input values, typically torch.Tensor objects.
Without PyTorch, pass NumPy arrays, other buffers such as
memoryview, DLPack objects such as CuPy arrays, or
ExecuTorchResult values. Scalar inputs take Python numbers.
clone_outputs: If False, the outputs share the method's memory and
a later execution may overwrite them, which avoids a copy.
Returns:
A list of output values, typically torch.Tensor objects. Without
PyTorch, tensor outputs are ExecuTorchResult objects.
"""
return self._method(inputs, clone_outputs=clone_outputs)
[docs]
def set_output(self, tensor: "torch.Tensor", index: int) -> None:
"""Binds output ``index`` to ``tensor``'s storage for every later call.
The method then writes that output straight into ``tensor`` and returns
it uncloned (viewed at the written shape when that differs) instead of
a copy of its own storage. Every later call overwrites ``tensor``, so
clone a returned output to keep it.
The output must not be memory planned (export with
``alloc_graph_output=False``), must not be returned at another output
index too, and must not be an input, which already lives in the tensor
passed in for it. That last case is only known once inputs have been
set: a binding made before the first call is removed by that call,
which raises. ``tensor`` must have the output's dtype and device, be
contiguous, hold at least as many bytes as the output's largest shape,
and not share memory with another bound output or with an input. The
output itself must be contiguous, not e.g. channels-last. A rejected
binding raises ``ValueError``. The ATen bindings don't support it.
Args:
tensor: The tensor that stores the output from now on.
index: The index of the output to bind.
"""
self._method.set_output(tensor, index)
@property
def metadata(self) -> MethodMeta:
"""Gets the metadata for the method.
The metadata includes information about input and output specifications,
such as tensor shapes, data types, and memory requirements.
Returns:
The MethodMeta object containing method specifications.
"""
return self._method.method_meta()
[docs]
class Program:
"""An ExecuTorch program, loaded from binary PTE data.
This can be used to load the methods/models defined by the program.
"""
def __init__(self, program: ExecuTorchProgram, data: Optional[bytes]) -> None:
# Hold the data so the program is not freed.
self._data = data
self._program = program
self._methods: Dict[str, Optional[Method]] = {}
# The names of the methods are preemptively added to the dictionary,
# but only map to None until they are loaded.
for method_idx in range(self._program.num_methods()):
self._methods[self._program.get_method_name(method_idx)] = None
@property
def method_names(self) -> Set[str]:
"""Returns method names of the Program as a set of strings."""
return set(self._methods.keys())
[docs]
def load_method(self, name: str) -> Optional[Method]:
"""Loads a method from the program.
Memory-planned buffers are allocated differently depending on where the
program placed them. A method whose buffers are all on the host shares
one set of arenas with every other host-only method of this program,
sized to the largest of them, so two such methods overwrite each
other's intermediate values. By default execute copies its outputs out,
so that sharing does not change what it returns. With
``clone_outputs=False`` the outputs share that memory, so the next
execution of any method of this program may overwrite them. A method
with any buffer placed on an accelerator gets its own arenas instead, claimed
when that method is first loaded and owned by that method until the
method itself is released. This program caches every method it loads,
so in normal use those arenas live as long as the program does. A
missing or exhausted accelerator therefore fails that one load rather
than the whole program, but whatever it does claim stays claimed for
every method loaded after it.
A program exported with ``share_mutable_buffers=True`` relies on every
method reading its shared mutable state from one allocation. A method
with accelerator-placed buffers gets its own arenas instead, so it also
gets its own copy of that state and will not observe writes made through
another method. Export without ``share_mutable_buffers`` if a method
needs both accelerator memory and shared state.
Args:
name: The name of the method to load.
Returns:
The loaded method.
"""
method = self._methods[name]
if method is None:
method = Method(self._program.load_method(name))
self._methods[name] = method
return method
def metadata(self, method_name: str) -> MethodMeta:
"""Gets the metadata for the specified method without loading it.
Args:
method_name: The name of the method.
Returns:
The metadata for the method, including input/output specifications.
"""
return self._program.method_meta(method_name)
def write_etdump_result_to_file(
self, etdump_path: str, debug_buffer_path: str
) -> None:
"""Writes the etdump and debug result to a file.
Args:
etdump_path: The path to the etdump file.
debug_buffer_path: The path to the debug buffer file.
"""
self._program.write_etdump_result_to_file(etdump_path, debug_buffer_path)
[docs]
class BackendRegistry:
"""The registry of backends that are available to the runtime."""
def __init__(self, legacy_module: ModuleType) -> None:
# TODO: Expose the kernel callables to Python.
self._legacy_module = legacy_module
@property
def registered_backend_names(self) -> List[str]:
"""Returns the names of all registered backends as a list of strings."""
return self._legacy_module._get_registered_backend_names()
[docs]
def is_available(self, backend_name: str) -> bool:
"""Checks if a specific backend is available in the runtime.
Args:
backend_name: The name of the backend to check (e.g., "XnnpackBackend").
Returns:
True if the backend is available, False otherwise.
"""
return self._legacy_module._is_available(backend_name)
[docs]
def set_option(
self, backend_name: str, options: Dict[str, Union[bool, int, str]]
) -> None:
"""Sets process-wide options of a backend, for example
``set_option("XnnpackBackend", {"weight_cache_enabled": True})``."""
self._legacy_module._set_option(backend_name, options)
[docs]
def get_option(
self, backend_name: str, options: Dict[str, Union[bool, int, str]]
) -> Dict[str, Union[bool, int, str]]:
"""Returns the backend's current values for the keys of ``options``,
whose values set the expected types."""
return self._legacy_module._get_option(backend_name, options)
[docs]
class OperatorRegistry:
"""The registry of operators that are available to the runtime."""
def __init__(self, legacy_module: ModuleType) -> None:
# TODO: Expose the kernel callables to Python.
self._legacy_module = legacy_module
@property
def operator_names(self) -> Set[str]:
"""Returns the names of all registered operators as a set of strings."""
return set(self._legacy_module._get_operator_names())
[docs]
class Runtime:
"""An instance of the ExecuTorch runtime environment.
This can be used to concurrently load and execute any number of ExecuTorch
programs and methods.
Attributes:
backend_registry: Registry for querying available hardware backends.
operator_registry: Registry for querying available operators/kernels.
"""
[docs]
@staticmethod
@functools.lru_cache(maxsize=1)
def get() -> "Runtime":
"""Gets the Runtime singleton."""
import executorch.extension.pybindings.portable_lib as legacy_module # type: ignore[import-not-found]
return Runtime(legacy_module=legacy_module)
def __init__(self, *, legacy_module: ModuleType) -> None:
# Public attributes.
self.backend_registry = BackendRegistry(legacy_module)
self.operator_registry = OperatorRegistry(legacy_module)
# Private attributes.
self._legacy_module = legacy_module
[docs]
def load_program(
self,
data: Union[bytes, bytearray, BinaryIO, Path, str],
*,
verification: Verification = Verification.InternalConsistency,
enable_etdump: bool = False,
debug_buffer_size: int = 0,
data_path: Optional[Union[Path, str]] = None,
backend_options: Optional[Dict[str, Dict[str, Union[bool, int, str]]]] = None,
) -> Program:
"""Loads an ExecuTorch program from a PTE binary.
Args:
data: The binary program data to load. Can be a file path (str or Path),
bytes/bytearray, or a file-like object.
verification: Level of program verification to perform (Minimal or InternalConsistency).
Default is InternalConsistency.
enable_etdump: If True, enables ETDump profiling for runtime performance analysis.
Default is False.
debug_buffer_size: Size of the debug buffer in bytes for ETDump data.
Only used when enable_etdump=True. Default is 0.
data_path: Path to a .ptd file holding data the program keeps outside the PTE,
such as the weights a CUDA export writes to a separate file.
backend_options: Load-time options for each backend, keyed by backend name,
used by every method loaded from this program.
Returns:
The loaded Program instance.
"""
if isinstance(data, (Path, str)):
p = self._legacy_module._load_program(
str(data),
enable_etdump=enable_etdump,
debug_buffer_size=debug_buffer_size,
program_verification=verification,
data_path=str(data_path) if data_path is not None else None,
backend_options=backend_options or {},
)
return Program(p, data=None)
elif isinstance(data, bytes):
data_bytes = data
elif isinstance(data, bytearray):
data_bytes = bytes(data)
elif hasattr(data, "read"):
# File-like object with read() method
data_bytes = data.read()
else:
raise TypeError(
f"Expected data to be bytes, bytearray, a path to a .pte file, or a file-like object, but got {type(data).__name__}."
)
p = self._legacy_module._load_program_from_buffer(
data_bytes,
enable_etdump=enable_etdump,
debug_buffer_size=debug_buffer_size,
program_verification=verification,
data_path=str(data_path) if data_path is not None else None,
backend_options=backend_options or {},
)
return Program(p, data=data_bytes)
