exir/backend/utils.py

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

import logging
import operator
from collections import defaultdict
from functools import lru_cache
from typing import Dict, Iterable, List, Optional, Set, Tuple, Union

import torch
from executorch.exir.backend.backend_details import ExportedProgram
from executorch.exir.backend.canonical_partitioners.duplicate_constant_node_pass import (
    duplicate_constant_node,
)
from executorch.exir.common import setting_python_recursive_limit
from executorch.exir.delegate import executorch_call_delegate
from executorch.exir.dialects._ops import ops as exir_ops

from executorch.exir.lowered_backend_module import create_submodule_from_nodes
from torch._export.utils import is_buffer, is_lifted_tensor_constant, is_param
from torch.fx.node import Node
from torch.fx.passes.utils.source_matcher_utils import SourcePartition

T_QuantPerTensor = exir_ops.edge.quantized_decomposed.quantize_per_tensor.default
T_DQuantPerTensor = exir_ops.edge.quantized_decomposed.dequantize_per_tensor.default


# NB: Set this to None to handle validation from MobileBert
@lru_cache(maxsize=None)
def is_same_node(
    node_left: Iterable[torch.fx.Node],
    node_right: Iterable[torch.fx.Node],
) -> bool:
    # two nodes are the same if they have the same target and op
    # same for their args
    if isinstance(node_left, torch.fx.Node) and isinstance(node_right, torch.fx.Node):
        if not (
            (node_left.target == node_right.target)
            and (node_left.op == node_right.op)
            and (len(node_left.all_input_nodes) == len(node_right.all_input_nodes))
            and all(
                is_same_node(arg_left, arg_right)
                for arg_left, arg_right in zip(
                    node_left.all_input_nodes, node_right.all_input_nodes
                )
            )
        ):
            return False
    else:
        if len(list(node_left)) != len(list(node_right)):
            return False
        for n_left, n_right in zip(node_left, node_right):
            if not is_same_node(n_left, n_right):
                return False
    return True


def is_identical_graph(
    graph_left: torch.fx.GraphModule, graph_right: torch.fx.GraphModule
) -> bool:
    # two graph are the same if they have the same nodes and op. The order of nodes also
    # matters in this function is more strict. Two graph are not considered as the same
    # if the topological order of the nodes is the same in this function but the order of nodes
    # is not the same.
    if len(list(graph_left.graph.nodes)) != len(list(graph_right.graph.nodes)):
        return False
    with setting_python_recursive_limit(30000):
        for node_left, node_right in zip(
            graph_left.graph.nodes, graph_right.graph.nodes
        ):
            if not (is_same_node(node_left, node_right)):
                return False
    return True


def remove_first_quant_and_last_dequant(
    graph_module: torch.fx.GraphModule,
) -> None:
    for node in graph_module.graph.nodes:
        if node.target == T_QuantPerTensor:
            if node.args[0].op == "placeholder":
                node_users = list(node.users.keys())
                for dequant_node in node_users:
                    # point the dequant arg to the placeholder
                    dequant_node.args = (node.args[0],) + dequant_node.args[1:]
        elif node.target == T_DQuantPerTensor:
            node_users = list(node.users.keys())
            if node_users[0].op == "output":
                # point the output arg to the quant node
                output_node = node_users[0]
                output_node.args = ([node.args[0]],)
    # Remove the quant/dequant nodes as they don't have users
    graph_module.graph.eliminate_dead_code()
    graph_module.recompile()


# TODO - use edge ops
def replace_quantized_partition_with_op(
    graph_module: torch.fx.GraphModule,
    partition: SourcePartition,
    replacement_op: torch._ops.OpOverloadPacket,
) -> Tuple[torch.fx.Node, List[torch.fx.Node], List[torch.fx.Node]]:
    """
    Replaces partition with the op specified by replacement_op. It's also expected that
    the nodes contained in partition are sourced from a quantized module as this function
    searches for the quantization pattern to consume along with the nodes in the partition,
    to be then replaced by replacement_op.

    Args:
        graph_module: The graph module from which this partition was sourced.
        partition: Partition to be replaced.
        replacement_op: The op to replace paritition with.
    Returns:
        Tuple: First element in the tuple is the new replaced module. The second and third
        node lists in the returned tuple consist of the dq and q nodes that were consumed
        along with this partition to be replaced by the replacement_op.
    """

    dequant_nodes = []
    quant_nodes = []
    input_nodes = []
    output_nodes = []

    partition_nodes = [node for node in partition.nodes if node not in partition.params]

    # We recreate our input nodes and output nodes list instead of using partition.input_nodes
    # and partition.output_nodes as the ordering of the nodes in those lists is not deterministic,
    # whereas for the quant fusion pass we expect deterministic ordering.
    for node in partition.nodes:
        for arg in node.args:
            if isinstance(arg, torch.fx.Node) and (arg not in partition.nodes):
                input_nodes.append(arg)

        for user in node.users.keys():
            if user not in partition.nodes:
                output_nodes.append(node)

    # Try to find all the dq nodes that are feeding into this module.
    for node in input_nodes:
        if node.target == T_DQuantPerTensor:
            dequant_nodes += [node]

    # Try to find all the q nodes that this module is feeding out into.
    for node in output_nodes:
        for user in node.users.keys():
            if user.target == T_QuantPerTensor:
                quant_nodes += [user]

    assert len(dequant_nodes) >= 1, "Dequant nodes missing in node list to be replaced."
    assert len(quant_nodes) >= 1, "Quant nodes missing in node list to be replaced."

    # After this, node list will essentially contain all the nodes in the
    # dq->op->q pattern that we will want to replace with a custom backend op.
    node_list = dequant_nodes + partition_nodes + quant_nodes

    submodule, call_module_node = create_submodule_from_nodes(
        graph_module, node_list, "to_be_replaced", skip_legalize_graph=True
    )

    # Update the replaced op so that we have all the latest args and kwargs.
    with graph_module.graph.inserting_before(call_module_node):
        replaced_op = graph_module.graph.call_function(
            replacement_op,
            call_module_node.args,
            kwargs=call_module_node.kwargs,
        )
        call_module_node.replace_all_uses_with(replaced_op)
        graph_module.graph.erase_node(call_module_node)
        replaced_op.meta = call_module_node.meta
    graph_module.recompile()

    return (replaced_op, dequant_nodes, quant_nodes)


def _assign_new_tag(
    tagged_exported_program: ExportedProgram,
    copied_nodes: Set[str],
):
    """
    Assign new tag to the copied nodes.

    Before the pass
    constant_0 (tag_10) ------------------> op_b (tag_10)
    constant_0_copy (tag_10) -------------> op_a (tag_11)

    After the pass
    constant_0 (tag_10) ------------------> op_b (tag_10)
    constant_0_copy (tag_11) -------------> op_a (tag_11)

    """
    for node in tagged_exported_program.graph.nodes:
        if node.op == "placeholder":
            if node.name in copied_nodes:
                users_tag = set()
                for user in node.users:
                    users_tag.add(user.meta.get("delegation_tag", None))
                # Assign the tag to the copy constant node the same as their users.
                if len(users_tag) == 1:
                    node.meta["delegation_tag"] = users_tag.pop()


def _maybe_duplicate_constant_nodes(
    tagged_exported_program: ExportedProgram,
    tag: str,
) -> None:
    """
    If the constants node is shared by different tagged nodes, like
    constant_0 ----> op_b (tag_10)
    |-------------> op_a (tag_11)

    we make default as constant_0 is duplicated to constant_0_1, constant_0_2, unless the node is tagged with "no_copy"
    constant_0 ------------------> op_b (tag_10)
    constant_0_copy -------------> op_a (tag_11)

    backend can estimate how much they want to duplicate the constant node, either error out or default to duplicate
    """
    candidate_nodes = set()
    for node in tagged_exported_program.graph.nodes:
        if node.meta.get("delegation_tag", "") == tag:
            if node.op == "placeholder":
                for user in node.users:
                    users_tag = user.meta.get("delegation_tag", None)
                    if users_tag != tag:
                        # If the node is tagged with "no_copy", we stop duplicating it and throw an error
                        if node.meta.get("no_copy", False):
                            raise RuntimeError(
                                f"constant data node ({node}) is tagged with ({tag}) but has user ({user}) which has tag ({users_tag})"
                            )
                        else:
                            candidate_nodes.add(node.name)
    copied_nodes = set()
    for candidate_node in candidate_nodes:
        # Both tagged exported program and the owning program need to go through the same duplication pass
        copied_nodes = copied_nodes.union(
            duplicate_constant_node(tagged_exported_program, candidate_node)
        )
    candidate_node_with_copies = candidate_nodes.union(copied_nodes)
    _assign_new_tag(tagged_exported_program, candidate_node_with_copies)


def _get_item_from_executorch_call_delegate(node: torch.fx.Node) -> bool:
    """
    Check if the node is the getitem followed by executorch_call_delegate node. These getitems node
    are just for getting the result from delegate because the input/output to delegates are flattened
    """
    return (
        node.target == operator.getitem
        and len(node.args) == 2
        and node.args[0].target == executorch_call_delegate  # pyre-ignore
        and isinstance(node.args[1], int)
    )


def get_non_lowered_nodes(graph: torch.fx.Graph) -> List[torch.fx.Node]:
    """
    Returns a list of non lowered nodes in the graph module.
    """
    return [
        node
        for node in graph.nodes
        if node.op == "call_function"
        and node.target != executorch_call_delegate
        and (not _get_item_from_executorch_call_delegate(node))
    ]


def get_delegates(graph: torch.fx.Graph) -> List[torch.fx.Node]:
    """
    Returns the list of delegates from the graph.
    """
    return [
        node
        for node in graph.nodes
        if node.op == "get_attr" and node.name.startswith("lowered_module_")
    ]


def print_delegated_graph(graph_module: torch.fx.GraphModule) -> None:
    """
    Print the formatted graph string.
    """
    print(format_delegated_graph(graph_module))


def format_delegated_graph(graph_module: torch.fx.GraphModule) -> str:
    """
    Return the formatted graph string of including lowered_module (both backend id and original graph) together with the graph module. Example output:
    graph():
        %arg0_1 : [num_users=2] = placeholder[target=arg0_1]
        %arg1_1 : [num_users=2] = placeholder[target=arg1_1]
        %arg2_1 : [num_users=2] = placeholder[target=arg2_1]
        %lowered_module_0 : [num_users=1] = get_attr[target=lowered_module_0]
            backend_id: BackendWithCompilerDemo
            lowered graph():
                %arg0_1 : [num_users=1] = placeholder[target=arg0_1]
                %arg1_1 : [num_users=1] = placeholder[target=arg1_1]
                %arg2_1 : [num_users=1] = placeholder[target=arg2_1]
                %aten_mm_default : [num_users=1] = call_function[target=executorch.exir.dialects.edge._ops.aten.mm.default](args = (%arg0_1, %arg1_1), kwargs = {})
                %aten_add_tensor : [num_users=1] = call_function[target=executorch.exir.dialects.edge._ops.aten.add.Tensor](args = (%aten_mm_default, %arg2_1), kwargs = {})
                return [aten_add_tensor]
        %executorch_call_delegate : [num_users=1] = call_function[target=torch.ops.higher_order.executorch_call_delegate](args = (%lowered_module_0, %arg0_1, %arg1_1, %arg2_1), kwargs = {})
        %getitem : [num_users=1] = call_function[target=operator.getitem](args = (%executorch_call_delegate, 0), kwargs = {})
        %aten_sub_tensor : [num_users=1] = call_function[target=executorch.exir.dialects.edge._ops.aten.sub.Tensor](args = (%getitem, %arg0_1), kwargs = {})
        %lowered_module_1 : [num_users=1] = get_attr[target=lowered_module_1]
            backend_id: BackendWithCompilerDemo
            lowered graph():
                %aten_sub_tensor : [num_users=1] = placeholder[target=aten_sub_tensor]
                %arg1_1 : [num_users=1] = placeholder[target=arg1_1]
                %arg2_1 : [num_users=1] = placeholder[target=arg2_1]
                %aten_mm_default_1 : [num_users=1] = call_function[target=executorch.exir.dialects.edge._ops.aten.mm.default](args = (%aten_sub_tensor, %arg1_1), kwargs = {})
                %aten_add_tensor_1 : [num_users=1] = call_function[target=executorch.exir.dialects.edge._ops.aten.add.Tensor](args = (%aten_mm_default_1, %arg2_1), kwargs = {})
                return [aten_add_tensor_1]
        %executorch_call_delegate_1 : [num_users=1] = call_function[target=torch.ops.higher_order.executorch_call_delegate](args = (%lowered_module_1, %aten_sub_tensor, %arg1_1, %arg2_1), kwargs = {})
        %getitem_1 : [num_users=1] = call_function[target=operator.getitem](args = (%executorch_call_delegate_1, 0), kwargs = {})
        return [getitem_1]
    """
    lowered_module_dict = {
        node.name: getattr(graph_module, node.name)
        for node in graph_module.graph.nodes
        if node.op == "get_attr" and node.name.startswith("lowered_module_")
    }
    indent = "  "
    graph_format_str = "graph():\n"
    for node in graph_module.graph.nodes:
        graph_format_str += f"{indent}{node.format_node()}\n"
        if node.op == "get_attr" and node.name.startswith("lowered_module_"):
            lowered_module = lowered_module_dict[node.name]
            graph_format_str += f"{indent * 2}backend_id: {lowered_module.backend_id}\n"
            graph_format_str += f"{indent * 2}lowered graph():\n"
            for node_in_lowered_module in lowered_module.original_module.graph.nodes:
                graph_format_str += (
                    f"{indent * 3}{node_in_lowered_module.format_node()}\n"
                )
    return graph_format_str


def tag_constant_data(edge_program: ExportedProgram) -> None:
    """
    Util function for partitioners. This function tags the const/param/buffers nodes
    whose users all belong within the same partition. This should be called after tagging all other nodes.
    Any const/param/buffer which is used as input to a subgraph, will be tagged with the same tag as that
    subgraph. Throw error when const/param/buffers is used across different partitions. That is the
    underlying data will be owned by multiple delegates.
    """
    mutated_buffer = set()
    for node in edge_program.graph.nodes:
        if node.op == "placeholder" and (
            is_param(edge_program, node)
            or is_buffer(edge_program, node)
            or is_lifted_tensor_constant(edge_program, node)
        ):
            for node_user in node.users:
                if node_user.name in edge_program.graph_signature.buffers_to_mutate:
                    logging.info(
                        "The buffer node is a mutated buffer node, which is not constant."
                    )
                    mutated_buffer.add(node)

    for node in edge_program.graph.nodes:
        # go through const/param/buffer nodes, if all users of const/param/buffer nodes are partitioned then partition
        if node.op == "placeholder" and (
            is_param(edge_program, node)
            or is_buffer(edge_program, node)
            or is_lifted_tensor_constant(edge_program, node)
        ):
            if node not in mutated_buffer:
                user_tags = set()
                for user in node.users:
                    user_tag = user.meta.get("delegation_tag", None)
                    if user_tag is not None:
                        user_tags.add(user_tag)
                if len(user_tags) > 1:
                    logging.info(
                        f"The data node is used across multiple partitions, including {user_tags}. "
                        "If the data is too large and it's not preferred to copy, please tag the "
                        "constant node like node.['no_copy'] = True and they won't be copied."
                    )
                # tag the data node with the same tag as the last user
                if len(user_tags) > 0:
                    node.meta["delegation_tag"] = user_tags.pop()


def tag_mutated_buffer(edge_program: ExportedProgram) -> None:
    """
    Util function for partitioners. This function tags the mutated buffer nodes
    whose users all belong within the same partition. This should be called after tagging all other nodes.
    Any buffer which is used as input to a subgraph, will be tagged with the same tag as that
    subgraph. Throw error when buffers is used across different partitions. That is the
    underlying data will be owned by multiple delegates.
    """
    for node in edge_program.graph.nodes:
        # Determine whether this node is a mutated buffer
        is_mutated_buffer_node = False
        if node.op == "placeholder" and is_buffer(edge_program, node):
            for node_user in node.users:
                if node_user.name in edge_program.graph_signature.buffers_to_mutate:
                    is_mutated_buffer_node = True
                    break
        # This node is mutated buffer, tag it
        if is_mutated_buffer_node:
            user_tags = set()
            for user in node.users:
                user_tag = user.meta.get("delegation_tag", None)
                if user_tag is not None:
                    user_tags.add(user_tag)
            if len(user_tags) > 1:
                logging.info(
                    f"The data node is used across multiple partitions, including {user_tags}. "
                    "If the data is too large and it's not preferred to copy, please tag the "
                    "constant node like node.['no_copy'] = True and they won't be copied."
                )
            # tag the data node with the same tag as the last user
            if len(user_tags) > 0:
                node.meta["delegation_tag"] = user_tags.pop()


# TODO - style: use templated types
class DelegateMappingBuilder:
    """
    Profiling helper class for building Delegate Mappings.
    Delegate Mappings are mappings from delegate debug identifiers to node
    debug handles. Specifically this is used to log within backend delegates

    Args:
        generated_identifiers (bool, optional): Whether identifier keys are
            generated automatically. Defaults to False.
    """

    def __init__(self, generated_identifiers: bool = False):
        self._generated_identifiers = generated_identifiers

        # Note that the internal struct has a Set value, while the getter
        # function returns the values as a tuple
        self._debug_handle_map: Union[Dict[int, Set[int]], Dict[str, Set[int]]] = (
            defaultdict(set)
        )
        self._next_index: int = 0

    def get_delegate_mapping(
        self,
    ) -> Union[Dict[int, Tuple[int]], Dict[str, Tuple[int]]]:
        """
        Returns:
           Union[Dict[int, Tuple[int]], Dict[str, Tuple[int]]]:
                A map of delegate debug identifier to a list of debug handles
                The keys (identifier) are either integers or strings
                The values are a sorted tuple of integer debug handles
        """
        # pyre-ignore Warning between Union[Dict[K, V], Dict[K2, V]] vs Dict[Union[K, K2], V]
        return {k: tuple(sorted(v)) for k, v in self._debug_handle_map.items()}

    def insert_delegate_mapping_entry(
        self,
        nodes: Optional[Union[Node, List[Node]]] = None,
        handles: Optional[Union[int, List[Optional[int]]]] = None,
        identifier: Optional[Union[int, str]] = None,
    ) -> Union[int, str]:
        """
        Add a new delegate mapping entry

        If self._generated_identifiers = False:
            - A new identifier must be provided, else an exception is thrown

        If self._generated_identifiers = True:
            - New identifiers are generated incrementally, 0 indexed
            - Identifiers cannot be manually provided, else an exception is thrown

        Args:
            nodes (Union[Node, List[Node]]): A (list of) Node(s)
            handles (Union[int, List[Optional[int]]]): A (list of) debug handle(s)
            identifier (Optional[Union[int, str]]):
                Debug identifier corresponding to the Node(s)

        Note: Exactly one of nodes and handles must be provided
        Note: If a debug handle is missing or None, it is skipped

        Returns:
            Union[int, str]:
                Delegate debug identifier inserted
        """

        # Check for manual addition of identifier (with generated identifiers enabled)
        if self._generated_identifiers and identifier is not None:
            raise Exception(
                f"Builders using generated identifiers can't manually add identifiers: {identifier}. Failed to add or update entry"
            )

        if identifier is not None and identifier in self._debug_handle_map:
            raise Exception(
                "This delegate debug identifier was already inserted. Duplicate delegate debug identifiers are not allowed."
            )

        # Check for exactly one of nodes and handles being populated
        if not ((nodes is not None) ^ (handles is not None)):
            raise Exception(
                "Only one of nodes or handles must be provided. Either both were provided or neither were provided. Failed to add or update entry."
            )

        # Resolve Identifier
        if identifier is None:
            if self._generated_identifiers:
                identifier = self._next_index
                self._next_index += 1
            else:
                raise Exception(
                    "No identifier provided. Failed to add or update entry."
                )

        # Collect debug handles
        if nodes is not None:
            new_debug_handles = {
                node.meta.get("debug_handle")
                for node in (nodes if isinstance(nodes, List) else [nodes])
            }
        else:
            new_debug_handles = (
                handles if isinstance(handles, (tuple, List)) else [handles]
            )

        # Filter for empty debug handles
        filtered_debug_handles = {
            handle for handle in new_debug_handles if handle is not None
        }
        if len(filtered_debug_handles) == 0:
            raise Exception("No valid debug handles found. Failed to add entry.")

        # pyre-ignore Warning from Union[int, st] keys
        self._debug_handle_map[identifier] = filtered_debug_handles
        return identifier


class WhyNoPartition:
    """
    Simple helper class for partitioners to log why a node was not lowered.

    Example usage:

        # In your backend partitioner file(s)
        why = WhyNoPartition(logger=your_backend_logger)

        # hypothetical function that checks if a node can be lowered
        if not can_be_lowered(node):
            why(node, "This node was not lowered because ...")
    """

    def __init__(self, logger: logging.Logger):
        self.logger = logger
        self.node: Optional[torch.fx.Node] = None
        self.reason: str = ""

    def __call__(self, node: torch.fx.Node, reason: str) -> None:
        self.node = node
        self.reason = reason
        self.logger.debug(self)

    def __str__(self) -> str:
        return f"WhyNoPartition: Node {self.node} was not partitioned because {self.reason}."