Generalize CXCancellation optimization pass to generic gate-inverse pairs

qiskit/transpiler/passes/__init__.py

@@ -173,6 +173,7 @@
 from .optimization import CrosstalkAdaptiveSchedule
 from .optimization import HoareOptimizer
 from .optimization import TemplateOptimization
+from .optimization import InverseCancellation
 
 # circuit analysis
 from .analysis import ResourceEstimation

qiskit/transpiler/passes/optimization/__init__.py

@@ -25,3 +25,4 @@
 from .crosstalk_adaptive_schedule import CrosstalkAdaptiveSchedule
 from .hoare_opt import HoareOptimizer
 from .template_optimization import TemplateOptimization
+from .inverse_cancellation import InverseCancellation

qiskit/transpiler/passes/optimization/inverse_cancellation.py

@@ -0,0 +1,137 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2021.
+#
+# 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.
+
+"""
+A generic InverseCancellation pass for any set of gate-inverse pairs.
+"""
+from typing import List, Tuple, Union
+
+from qiskit.circuit import Gate
+from qiskit.dagcircuit import DAGCircuit
+from qiskit.transpiler.basepasses import TransformationPass
+from qiskit.transpiler.exceptions import TranspilerError
+
+
+class InverseCancellation(TransformationPass):
+    """Cancel specific Gates which are inverses of each other when they occur back-to-
+    back."""
+
+    def __init__(self, gates_to_cancel: List[Union[Gate, Tuple[Gate, Gate]]]):
+        """Initialize InverseCancellation pass.
+
+        Args:
+            gates_to_cancel: list of gates to cancel
+
+        Raises:
+            TranspilerError:
+                Initalization raises an error when the input is not a self-inverse gate
+                or a two-tuple of inverse gates.
+        """
+
+        for gates in gates_to_cancel:
+            if isinstance(gates, Gate):
+                if gates != gates.inverse():
+                    raise TranspilerError("Gate {} is not self-inverse".format(gates.name))
+            elif isinstance(gates, tuple):
+                if len(gates) != 2:
+                    raise TranspilerError(
+                        "Too many or too few inputs: {}. Only two are allowed.".format(gates)
+                    )
+                if gates[0] != gates[1].inverse():
+                    raise TranspilerError(
+                        "Gate {} and {} are not inverse.".format(gates[0].name, gates[1].name)
+                    )
+            else:
+                raise TranspilerError(
+                    "InverseCancellation pass does not take input type {}. Input must be"
+                    " a Gate.".format(type(gates))
+                )
+
+        self.self_inverse_gates = []
+        self.inverse_gate_pairs = []
+
+        for gates in gates_to_cancel:
+            if isinstance(gates, Gate):
+                self.self_inverse_gates.append(gates)
+            else:
+                self.inverse_gate_pairs.append(gates)
+
+        super().__init__()
+
+    def run(self, dag: DAGCircuit):
+        """Run the InverseCancellation pass on `dag`.
+
+        Args:
+            dag: the directed acyclic graph to run on.
+
+        Returns:
+            DAGCircuit: Transformed DAG.
+        """
+
+        dag = self._run_on_self_inverse(dag, self.self_inverse_gates)
+        return self._run_on_inverse_pairs(dag, self.inverse_gate_pairs)
+
+    def _run_on_self_inverse(self, dag: DAGCircuit, self_inverse_gates: List[Gate]):
+        """
+        Run self-inverse gates on `dag`.
+
+        Args:
+            dag: the directed acyclic graph to run on.
+            self_inverse_gates: list of gates who cancel themeselves in pairs
+
+        Returns:
+            DAGCircuit: Transformed DAG.
+        """
+        # Sets of gate runs by name, for instance: [{(H 0, H 0), (H 1, H 1)}, {(X 0, X 0}]
+        gate_runs_sets = [dag.collect_runs([gate.name]) for gate in self_inverse_gates]
+        for gate_runs in gate_runs_sets:
+            for gate_cancel_run in gate_runs:
+                partitions = []
+                chunk = []
+                for i in range(len(gate_cancel_run) - 1):
+                    chunk.append(gate_cancel_run[i])
+                    if gate_cancel_run[i].qargs != gate_cancel_run[i + 1].qargs:
+                        partitions.append(chunk)
+                        chunk = []
+                chunk.append(gate_cancel_run[-1])
+                partitions.append(chunk)
+                # Remove an even number of gates from each chunk
+                for chunk in partitions:
+                    if len(chunk) % 2 == 0:
+                        dag.remove_op_node(chunk[0])
+                    for node in chunk[1:]:
+                        dag.remove_op_node(node)
+        return dag
+
+    def _run_on_inverse_pairs(self, dag: DAGCircuit, inverse_gate_pairs: List[Tuple[Gate, Gate]]):
+        """
+        Run inverse gate pairs on `dag`.
+
+        Args:
+            dag: the directed acyclic graph to run on.
+            inverse_gate_pairs: list of gates with inverse angles that cancel each other.
+
+        Returns:
+            DAGCircuit: Transformed DAG.
+        """
+        for pair in inverse_gate_pairs:
+            gate_cancel_runs = dag.collect_runs([pair[0].name])
+            for dag_nodes in gate_cancel_runs:
+                for i in range(len(dag_nodes) - 1):
+                    if dag_nodes[i].op == pair[0] and dag_nodes[i + 1].op == pair[1]:
+                        dag.remove_op_node(dag_nodes[i])
+                        dag.remove_op_node(dag_nodes[i + 1])
+                    elif dag_nodes[i].op == pair[1] and dag_nodes[i + 1].op == pair[0]:
+                        dag.remove_op_node(dag_nodes[i])
+                        dag.remove_op_node(dag_nodes[i + 1])
+
+        return dag

releasenotes/notes/cx-cancellation-pass-generalization-538fb7cfe49b3fd5.yaml

@@ -0,0 +1,17 @@
+---
+features:
+  - |
+    Introduced a new feature ``InverseCancellation`` that generalizes the ``CXInverseCancellation`` 
+    pass to cancel any self-inverse gates or gate-inverse pairs. It can be used by 
+    initializing ``InverseCancellation`` and passing a gate to cancel, for example::
+
+      from qiskit.transpiler.passes import InverseCancellation
+      from qiskit import QuantumCircuit
+      from qiskit.circuit.library import HGate
+      from qiskit.transpiler import PassManager
+      qc = QuantumCircuit(2, 2)
+      qc.h(0)
+      qc.h(0)
+      pass_ = InverseCancellation([HGate()])
+      pm = PassManager(pass_)
+      new_circ = pm.run(qc)

test/python/transpiler/test_inverse_cancellation.py

@@ -0,0 +1,166 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2020.
+#
+# 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.
+
+"""
+Testing InverseCancellation
+"""
+
+import numpy as np
+
+from qiskit import QuantumCircuit
+from qiskit.transpiler.exceptions import TranspilerError
+from qiskit.transpiler.passes import InverseCancellation
+from qiskit.transpiler import PassManager
+from qiskit.test import QiskitTestCase
+from qiskit.circuit.library import RXGate, HGate, CXGate, PhaseGate, XGate
+
+
+class TestInverseCancellation(QiskitTestCase):
+    """Test the InverseCancellation transpiler pass."""
+
+    def test_basic_self_inverse(self):
+        """Test that a single self-inverse gate as input can be cancelled."""
+        qc = QuantumCircuit(2, 2)
+        qc.h(0)
+        qc.h(0)
+        pass_ = InverseCancellation([HGate()])
+        pm = PassManager(pass_)
+        new_circ = pm.run(qc)
+        gates_after = new_circ.count_ops()
+        self.assertNotIn("h", gates_after)
+
+    def test_odd_number_self_inverse(self):
+        """Test that an odd number of self-inverse gates leaves one gate remaining."""
+        qc = QuantumCircuit(2, 2)
+        qc.h(0)
+        qc.h(0)
+        qc.h(0)
+        pass_ = InverseCancellation([HGate()])
+        pm = PassManager(pass_)
+        new_circ = pm.run(qc)
+        gates_after = new_circ.count_ops()
+        self.assertIn("h", gates_after)
+        self.assertEqual(gates_after["h"], 1)
+
+    def test_basic_cx_self_inverse(self):
+        """Test that a single self-inverse cx gate as input can be cancelled."""
+        qc = QuantumCircuit(2, 2)
+        qc.cx(0, 1)
+        qc.cx(0, 1)
+        pass_ = InverseCancellation([CXGate()])
+        pm = PassManager(pass_)
+        new_circ = pm.run(qc)
+        gates_after = new_circ.count_ops()
+        self.assertNotIn("cx", gates_after)
+
+    def test_basic_gate_inverse(self):
+        """Test that a basic pair of gate inverse can be cancelled."""
+        qc = QuantumCircuit(2, 2)
+        qc.rx(np.pi / 4, 0)
+        qc.rx(-np.pi / 4, 0)
+        pass_ = InverseCancellation([(RXGate(np.pi / 4), RXGate(-np.pi / 4))])
+        pm = PassManager(pass_)
+        new_circ = pm.run(qc)
+        gates_after = new_circ.count_ops()
+        self.assertNotIn("rx", gates_after)
+
+    def test_non_inverse_do_not_cancel(self):
+        """Test that non-inverse gate pairs do not cancel."""
+        qc = QuantumCircuit(2, 2)
+        qc.rx(np.pi / 4, 0)
+        qc.rx(np.pi / 4, 0)
+        pass_ = InverseCancellation([(RXGate(np.pi / 4), RXGate(-np.pi / 4))])
+        pm = PassManager(pass_)
+        new_circ = pm.run(qc)
+        gates_after = new_circ.count_ops()
+        self.assertIn("rx", gates_after)
+        self.assertEqual(gates_after["rx"], 2)
+
+    def test_non_consecutive_gates(self):
+        """Test that only consecutive gates cancel."""
+        qc = QuantumCircuit(2, 2)
+        qc.h(0)
+        qc.h(0)
+        qc.h(0)
+        qc.cx(0, 1)
+        qc.cx(0, 1)
+        qc.h(0)
+        pass_ = InverseCancellation([HGate(), CXGate()])
+        pm = PassManager(pass_)
+        new_circ = pm.run(qc)
+        gates_after = new_circ.count_ops()
+        self.assertNotIn("cx", gates_after)
+        self.assertEqual(gates_after["h"], 2)
+
+    def test_gate_inverse_phase_gate(self):
+        """Test that an inverse pair of a PhaseGate can be cancelled."""
+        qc = QuantumCircuit(2, 2)
+        qc.p(np.pi / 4, 0)
+        qc.p(-np.pi / 4, 0)
+        pass_ = InverseCancellation([(PhaseGate(np.pi / 4), PhaseGate(-np.pi / 4))])
+        pm = PassManager(pass_)
+        new_circ = pm.run(qc)
+        gates_after = new_circ.count_ops()
+        self.assertNotIn("p", gates_after)
+
+    def test_self_inverse_on_different_qubits(self):
+        """Test that self_inverse gates cancel on the correct qubits."""
+        qc = QuantumCircuit(2, 2)
+        qc.h(0)
+        qc.h(1)
+        qc.h(0)
+        qc.h(1)
+        pass_ = InverseCancellation([HGate()])
+        pm = PassManager(pass_)
+        new_circ = pm.run(qc)
+        gates_after = new_circ.count_ops()
+        self.assertNotIn("h", gates_after)
+
+    def test_non_inverse_raise_error(self):
+        """Test that non-inverse gate inputs raise an error."""
+        qc = QuantumCircuit(2, 2)
+        qc.rx(np.pi / 2, 0)
+        qc.rx(np.pi / 4, 0)
+        with self.assertRaises(TranspilerError):
+            InverseCancellation([RXGate(0.5)])
+
+    def test_non_gate_inverse_raise_error(self):
+        """Test that non-inverse gate inputs raise an error."""
+        qc = QuantumCircuit(2, 2)
+        qc.rx(np.pi / 4, 0)
+        qc.rx(np.pi / 4, 0)
+        with self.assertRaises(TranspilerError):
+            InverseCancellation([(RXGate(np.pi / 4))])
+
+    def test_string_gate_error(self):
+        """Test that when gate is passed as a string an error is raised."""
+        qc = QuantumCircuit(2, 2)
+        qc.h(0)
+        qc.h(0)
+        with self.assertRaises(TranspilerError):
+            InverseCancellation(["h"])
+
+    def test_consecutive_self_inverse_h_x_gate(self):
+        """Test that only consecutive self-inverse gates cancel."""
+        qc = QuantumCircuit(2, 2)
+        qc.h(0)
+        qc.h(0)
+        qc.h(0)
+        qc.x(0)
+        qc.x(0)
+        qc.h(0)
+        pass_ = InverseCancellation([HGate(), XGate()])
+        pm = PassManager(pass_)
+        new_circ = pm.run(qc)
+        gates_after = new_circ.count_ops()
+        self.assertNotIn("x", gates_after)
+        self.assertEqual(gates_after["h"], 2)