Adding CollectAndCollapse transpiler pass

qiskit/converters/dagdependency_to_circuit.py

@@ -36,7 +36,7 @@ def dagdependency_to_circuit(dagdependency):
 
     circuit.calibrations = dagdependency.calibrations
 
-    for node in dagdependency.get_nodes():
+    for node in dagdependency.topological_nodes():
         circuit._append(CircuitInstruction(node.op.copy(), node.qargs, node.cargs))
 
     return circuit

qiskit/converters/dagdependency_to_dag.py

@@ -37,7 +37,7 @@ def dagdependency_to_dag(dagdependency):
     for register in dagdependency.cregs.values():
         dagcircuit.add_creg(register)
 
-    for node in dagdependency.get_nodes():
+    for node in dagdependency.topological_nodes():
         # Get arguments for classical control (if any)
         inst = node.op.copy()
         dagcircuit.apply_operation_back(inst, node.qargs, node.cargs)

qiskit/dagcircuit/collect_blocks.py

@@ -0,0 +1,277 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2022.
+#
+# This code is licensed under the Apache License, Version 2.0. You may
+# obtain a copy of this license in the LICENSE.txt file in the root directory
+# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
+#
+# Any modifications or derivative works of this code must retain this
+# copyright notice, and modified files need to carry a notice indicating
+# that they have been altered from the originals.
+
+
+"""Various ways to divide a DAG into blocks of nodes, to split blocks of nodes
+into smaller sub-blocks, and to consolidate blocks."""
+
+from qiskit.circuit import QuantumCircuit, CircuitInstruction
+from . import DAGOpNode, DAGCircuit, DAGDependency
+from .exceptions import DAGCircuitError
+
+
+class BlockCollector:
+    """This class implements various strategies of dividing a DAG (direct acyclic graph)
+    into blocks of nodes that satisfy certain criteria. It works both with the
+    :class:`~qiskit.dagcircuit.DAGCircuit` and
+    :class:`~qiskit.dagcircuit.DAGDependency` representations of a DAG, where
+    DagDependency takes into account commutativity between nodes.
+
+    Collecting nodes from DAGDependency generally leads to more optimal results, but is
+    slower, as it requires to construct a DAGDependency beforehand. Thus, DAGCircuit should
+    be used with lower transpiler settings, and DAGDependency should be used with higher
+    transpiler settings.
+
+    In general, there are multiple ways to collect maximal blocks. The approaches used
+    here are of the form 'starting from the input nodes of a DAG, greedily collect
+    the largest block of nodes that match certain criteria'. For additional details,
+    see https://github.com/Qiskit/qiskit-terra/issues/5775.
+    """
+
+    def __init__(self, dag):
+        """
+        Args:
+            dag (Union[DAGCircuit, DAGDependency]): The input DAG.
+
+        Raises:
+            DAGCircuitError: the input object is not a DAG.
+        """
+
+        self.dag = dag
+        self.pending_nodes = None
+        self.in_degree = None
+
+        if isinstance(dag, DAGCircuit):
+            self.is_dag_dependency = False
+
+        elif isinstance(dag, DAGDependency):
+            self.is_dag_dependency = True
+
+        else:
+            raise DAGCircuitError("not a DAG.")
+
+    def setup_in_degrees(self):
+        """For an efficient implementation, we compute and keep updating the in_degree
+        for every node, that is the number of node's immediate predecessors.
+        A node is a leaf (aka input) node iff its in_degree is 0.
+        When a node is (marked as) collected, the in_degrees of its immediate
+        successors are updated by subtracting 1.
+        Additionally, pending_nodes explicitly keeps the list of nodes with in_degree 0.
+        """
+        self.pending_nodes = []
+        self.in_degree = dict()
+        for node in self._op_nodes():
+            deg = len(self._direct_preds(node))
+            self.in_degree[node] = deg
+            if deg == 0:
+                self.pending_nodes.append(node)
+
+    def _op_nodes(self):
+        """Returns DAG nodes."""
+        if not self.is_dag_dependency:
+            return self.dag.op_nodes()
+        else:
+            return self.dag.get_nodes()
+
+    def _direct_preds(self, node):
+        """Returns direct predecessors of a node."""
+        if not self.is_dag_dependency:
+            return [pred for pred in self.dag.predecessors(node) if isinstance(pred, DAGOpNode)]
+        else:
+            return [
+                self.dag.get_node(pred_id) for pred_id in self.dag.direct_predecessors(node.node_id)
+            ]
+
+    def _direct_succs(self, node):
+        """Returns direct successors of a node."""
+        if not self.is_dag_dependency:
+            return [succ for succ in self.dag.successors(node) if isinstance(succ, DAGOpNode)]
+        else:
+            return [
+                self.dag.get_node(succ_id) for succ_id in self.dag.direct_successors(node.node_id)
+            ]
+
+    def have_uncollected_nodes(self):
+        """Returns whether there are uncollected (pending) nodes"""
+        return len(self.pending_nodes) > 0
+
+    def collect_matching_block(self, filter_fn):
+        """Iteratively collects the largest block of input (aka in_degree=0) nodes that match a
+        given filtering function. Examples of this include collecting blocks of swap gates,
+        blocks of linear gates (CXs and SWAPs), blocks of Clifford gates, blocks of single-qubit gates,
+        blocks of two-qubit gates, etc.  Here 'iteratively' means that once a node is collected,
+        the in_degrees of its immediate successors are decreased by 1, allowing more nodes to become
+        input and to be eligible for collecting into the current block.
+        Returns the block of collected nodes.
+        """
+        current_block = []
+        unprocessed_pending_nodes = self.pending_nodes
+        self.pending_nodes = []
+
+        # Iteratively process unprocessed_pending_nodes:
+        # - any node that does not match filter_fn is added to pending_nodes
+        # - any node that match filter_fn is added to the current_block,
+        #   and some of its successors may be moved to unprocessed_pending_nodes.
+        while unprocessed_pending_nodes:
+            new_pending_nodes = []
+            for node in unprocessed_pending_nodes:
+                if filter_fn(node):
+                    current_block.append(node)
+
+                    # update the in_degree of node's successors
+                    for suc in self._direct_succs(node):
+                        self.in_degree[suc] -= 1
+                        if self.in_degree[suc] == 0:
+                            new_pending_nodes.append(suc)
+                else:
+                    self.pending_nodes.append(node)
+            unprocessed_pending_nodes = new_pending_nodes
+
+        return current_block
+
+    def collect_all_matching_blocks(self, filter_fn, split_blocks=True, min_block_size=2):
+        """Collects all blocks that match a given filtering function filter_fn.
+        This iteratively finds the largest block that does not match filter_fn,
+        then the largest block that matches filter_fn, and so on, until no more uncollected
+        nodes remain. Intuitively, finding larger blocks of non-matching nodes helps to
+        find larger blocks of matching nodes later on.
+
+        The option ``split_blocks`` allows to collected blocks into sub-blocks over
+        disjoint qubit subsets. The option ``min_block_size``specifies the minimum number
+        of gates in the block for the block to be collected.
+
+        Returns the list of matching blocks only.
+        """
+
+        def not_filter_fn(node):
+            """Returns the opposite of filter_fn."""
+            return not filter_fn(node)
+
+        self.setup_in_degrees()
+
+        # Iteratively collect non-matching and matching blocks.
+        matching_blocks = []
+        while self.have_uncollected_nodes():
+            self.collect_matching_block(not_filter_fn)
+            matching_block = self.collect_matching_block(filter_fn)
+            if matching_block:
+                matching_blocks.append(matching_block)
+
+        # If the option split_blocks is set, refine blocks by splitting them into sub-blocks over
+        # disconnected qubit subsets.
+        if split_blocks:
+            split_blocks = []
+            for block in matching_blocks:
+                split_blocks.extend(BlockSplitter().run(block))
+            matching_blocks = split_blocks
+
+        # Keep only blocks with at least min_block_sizes.
+        matching_blocks = [block for block in matching_blocks if len(block) >= min_block_size]
+
+        return matching_blocks
+
+
+class BlockSplitter:
+    """Splits a block of nodes into sub-blocks over disjoint qubits.
+    The implementation is based on the Disjoint Set Union data structure."""
+
+    def __init__(self):
+        self.leader = {}  # qubit's group leader
+        self.group = {}  # qubit's group
+
+    def find_leader(self, index):
+        """Find in DSU."""
+        if index not in self.leader:
+            self.leader[index] = index
+            self.group[index] = []
+            return index
+        if self.leader[index] == index:
+            return index
+        self.leader[index] = self.find_leader(self.leader[index])
+        return self.leader[index]
+
+    def union_leaders(self, index1, index2):
+        """Union in DSU."""
+        leader1 = self.find_leader(index1)
+        leader2 = self.find_leader(index2)
+        if leader1 == leader2:
+            return
+        if len(self.group[leader1]) < len(self.group[leader2]):
+            leader1, leader2 = leader2, leader1
+
+        self.leader[leader2] = leader1
+        self.group[leader1].extend(self.group[leader2])
+        self.group[leader2].clear()
+
+    def run(self, block):
+        """Splits block of nodes into sub-blocks over disjoint qubits."""
+        for node in block:
+            indices = node.qargs
+            if not indices:
+                continue
+            first = indices[0]
+            for index in indices[1:]:
+                self.union_leaders(first, index)
+            self.group[self.find_leader(first)].append(node)
+
+        blocks = []
+        for index in self.leader:
+            if self.leader[index] == index:
+                blocks.append(self.group[index])
+
+        return blocks
+
+
+class BlockCollapser:
+    """This class implements various strategies of consolidating blocks of nodes
+     in a DAG (direct acyclic graph). It works both with the
+    :class:`~qiskit.dagcircuit.DAGCircuit` and
+    :class:`~qiskit.dagcircuit.DAGDependency` DAG representations.
+    """
+
+    def __init__(self, dag):
+        """
+        Args:
+            dag (Union[DAGCircuit, DAGDependency]): The input DAG.
+        """
+
+        self.dag = dag
+
+    def collapse_to_operation(self, blocks, collapse_fn):
+        """For each block, constructs a quantum circuit containing instructions in the block,
+        then uses collapse_fn to collapse this circuit into a single operation.
+        """
+        global_index_map = {wire: idx for idx, wire in enumerate(self.dag.qubits)}
+        for block in blocks:
+            # Find the set of qubits used in this block (which might be much smaller than
+            # the set of all qubits).
+            cur_qubits = set()
+            for node in block:
+                cur_qubits.update(node.qargs)
+
+            # For reproducibility, order these qubits compatibly with the global order.
+            sorted_qubits = sorted(cur_qubits, key=lambda x: global_index_map[x])
+
+            # Construct a quantum circuit from the nodes in the block, remapping the qubits.
+            wire_pos_map = dict((qb, ix) for ix, qb in enumerate(sorted_qubits))
+            qc = QuantumCircuit(len(cur_qubits))
+            for node in block:
+                remapped_qubits = [wire_pos_map[qarg] for qarg in node.qargs]
+                qc.append(CircuitInstruction(node.op, remapped_qubits, node.cargs))
+
+            # Collapse this quantum circuit into an operation.
+            op = collapse_fn(qc)
+
+            # Replace the block of nodes in the DAG by the constructed operation
+            # (the function replace_block_with_op is implemented both in DAGCircuit and DAGDependency).
+            self.dag.replace_block_with_op(block, op, wire_pos_map, cycle_check=False)
+        return self.dag

qiskit/dagcircuit/dagcircuit.py

@@ -491,7 +491,7 @@ def _add_op_node(self, op, qargs, cargs):
         """Add a new operation node to the graph and assign properties.
 
         Args:
-            op (qiskit.circuit.Instruction): the operation associated with the DAG node
+            op (qiskit.circuit.Operation): the operation associated with the DAG node
             qargs (list[Qubit]): list of quantum wires to attach to.
             cargs (list[Clbit]): list of classical wires to attach to.
         Returns:
@@ -547,7 +547,7 @@ def apply_operation_back(self, op, qargs=(), cargs=()):
         """Apply an operation to the output of the circuit.
 
         Args:
-            op (qiskit.circuit.Instruction): the operation associated with the DAG node
+            op (qiskit.circuit.Operation): the operation associated with the DAG node
             qargs (tuple[Qubit]): qubits that op will be applied to
             cargs (tuple[Clbit]): cbits that op will be applied to
         Returns:
@@ -583,7 +583,7 @@ def apply_operation_front(self, op, qargs=(), cargs=()):
         """Apply an operation to the input of the circuit.
 
         Args:
-            op (qiskit.circuit.Instruction): the operation associated with the DAG node
+            op (qiskit.circuit.Operation): the operation associated with the DAG node
             qargs (tuple[Qubit]): qubits that op will be applied to
             cargs (tuple[Clbit]): cbits that op will be applied to
         Returns:
@@ -1069,7 +1069,7 @@ def topological_op_nodes(self, key=None):
         return (nd for nd in self.topological_nodes(key) if isinstance(nd, DAGOpNode))
 
     def replace_block_with_op(self, node_block, op, wire_pos_map, cycle_check=True):
-        """Replace a block of nodes with a single.
+        """Replace a block of nodes with a single node.
 
         This is used to consolidate a block of DAGOpNodes into a single
         operation. A typical example is a block of gates being consolidated
@@ -1079,14 +1079,14 @@ def replace_block_with_op(self, node_block, op, wire_pos_map, cycle_check=True):
         Args:
             node_block (List[DAGNode]): A list of dag nodes that represents the
                 node block to be replaced
-            op (qiskit.circuit.Instruction): The instruction to replace the
+            op (qiskit.circuit.Operation): The operation to replace the
                 block with
             wire_pos_map (Dict[Qubit, int]): The dictionary mapping the qarg to
                 the position. This is necessary to reconstruct the qarg order
                 over multiple gates in the combined single op node.
             cycle_check (bool): When set to True this method will check that
                 replacing the provided ``node_block`` with a single node
-                would introduce a a cycle (which would invalidate the
+                would introduce a cycle (which would invalidate the
                 ``DAGCircuit``) and will raise a ``DAGCircuitError`` if a cycle
                 would be introduced. This checking comes with a run time
                 penalty. If you can guarantee that your input ``node_block`` is
@@ -1316,24 +1316,24 @@ def edge_weight_map(wire):
         return {k: self._multi_graph[v] for k, v in node_map.items()}
 
     def substitute_node(self, node, op, inplace=False):
-        """Replace an DAGOpNode with a single instruction. qargs, cargs and
-        conditions for the new instruction will be inferred from the node to be
-        replaced. The new instruction will be checked to match the shape of the
-        replaced instruction.
+        """Replace an DAGOpNode with a single operation. qargs, cargs and
+        conditions for the new operation will be inferred from the node to be
+        replaced. The new operation will be checked to match the shape of the
+        replaced operation.
 
         Args:
             node (DAGOpNode): Node to be replaced
-            op (qiskit.circuit.Instruction): The :class:`qiskit.circuit.Instruction`
+            op (qiskit.circuit.Operation): The :class:`qiskit.circuit.Operation`
                 instance to be added to the DAG
             inplace (bool): Optional, default False. If True, existing DAG node
                 will be modified to include op. Otherwise, a new DAG node will
                 be used.
 
         Returns:
-            DAGOpNode: the new node containing the added instruction.
+            DAGOpNode: the new node containing the added operation.
 
         Raises:
-            DAGCircuitError: If replacement instruction was incompatible with
+            DAGCircuitError: If replacement operation was incompatible with
             location of target node.
         """
 
@@ -1343,7 +1343,7 @@ def substitute_node(self, node, op, inplace=False):
         if node.op.num_qubits != op.num_qubits or node.op.num_clbits != op.num_clbits:
             raise DAGCircuitError(
                 "Cannot replace node of width ({} qubits, {} clbits) with "
-                "instruction of mismatched width ({} qubits, {} clbits).".format(
+                "operation of mismatched width ({} qubits, {} clbits).".format(
                     node.op.num_qubits, node.op.num_clbits, op.num_qubits, op.num_clbits
                 )
             )
@@ -1419,7 +1419,7 @@ def op_nodes(self, op=None, include_directives=True):
         """Get the list of "op" nodes in the dag.
 
         Args:
-            op (Type): :class:`qiskit.circuit.Instruction` subclass op nodes to
+            op (Type): :class:`qiskit.circuit.Operation` subclass op nodes to
                 return. If None, return all op nodes.
             include_directives (bool): include `barrier`, `snapshot` etc.