diff --git a/qiskit/transpiler/passes/__init__.py b/qiskit/transpiler/passes/__init__.py index b185c7a03d28..0374c8251fa2 100644 --- a/qiskit/transpiler/passes/__init__.py +++ b/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 diff --git a/qiskit/transpiler/passes/optimization/__init__.py b/qiskit/transpiler/passes/optimization/__init__.py index 8a418aa0df6f..e487946b2118 100644 --- a/qiskit/transpiler/passes/optimization/__init__.py +++ b/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 diff --git a/qiskit/transpiler/passes/optimization/inverse_cancellation.py b/qiskit/transpiler/passes/optimization/inverse_cancellation.py new file mode 100644 index 000000000000..65b752ae8afc --- /dev/null +++ b/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 diff --git a/releasenotes/notes/cx-cancellation-pass-generalization-538fb7cfe49b3fd5.yaml b/releasenotes/notes/cx-cancellation-pass-generalization-538fb7cfe49b3fd5.yaml new file mode 100644 index 000000000000..2db152e3ac4f --- /dev/null +++ b/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) diff --git a/test/python/transpiler/test_inverse_cancellation.py b/test/python/transpiler/test_inverse_cancellation.py new file mode 100644 index 000000000000..c5cb5639b089 --- /dev/null +++ b/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)