swap_cancellation pass

qiskit/transpiler/passes/__init__.py

@@ -68,6 +68,7 @@
    Collect2qBlocks
    ConsolidateBlocks
    CXCancellation
+   SWAPCancellation
    CommutationAnalysis
    CommutativeCancellation
    RemoveDiagonalGatesBeforeMeasure
@@ -156,6 +157,7 @@
 from .optimization import CommutationAnalysis
 from .optimization import CommutativeCancellation
 from .optimization import CXCancellation
+from .optimization import SWAPCancellation
 from .optimization import OptimizeSwapBeforeMeasure
 from .optimization import RemoveResetInZeroState
 from .optimization import RemoveDiagonalGatesBeforeMeasure

qiskit/transpiler/passes/optimization/__init__.py

@@ -19,6 +19,7 @@
 from .commutation_analysis import CommutationAnalysis
 from .commutative_cancellation import CommutativeCancellation
 from .cx_cancellation import CXCancellation
+from .swap_cancellation import SWAPCancellation
 from .optimize_swap_before_measure import OptimizeSwapBeforeMeasure
 from .remove_reset_in_zero_state import RemoveResetInZeroState
 from .remove_diagonal_gates_before_measure import RemoveDiagonalGatesBeforeMeasure

qiskit/transpiler/passes/optimization/swap_cancellation.py

@@ -0,0 +1,54 @@
+# 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.
+
+"""Cancel back-to-back `swap` gates in dag."""
+
+from qiskit.transpiler.basepasses import TransformationPass
+
+
+class SWAPCancellation(TransformationPass):
+    """Cancel back-to-back `swap` gates in dag."""
+
+    def run(self, dag):
+        """Run the SWAPCancellation pass on `dag`.
+
+        Args:
+            dag (DAGCircuit): the directed acyclic graph to run on.
+
+        Returns:
+            DAGCircuit: Transformed DAG.
+        """
+        swap_runs = dag.collect_runs(["swap"])
+        for swap_run in swap_runs:
+            # Partition the swap_run into chunks with equal gate arguments
+            partition = []
+            chunk = []
+            for i in range(len(swap_run) - 1):
+                chunk.append(swap_run[i])
+
+                qargs0 = swap_run[i].qargs
+                qargs1 = swap_run[i + 1].qargs
+
+                if set(qargs0) != set(qargs1):
+                    partition.append(chunk)
+                    chunk = []
+            chunk.append(swap_run[-1])
+            partition.append(chunk)
+            # Simplify each chunk in the partition
+            for chunk in partition:
+                if len(chunk) % 2 == 0:
+                    for n in chunk:
+                        dag.remove_op_node(n)
+                else:
+                    for n in chunk[1:]:
+                        dag.remove_op_node(n)
+        return dag

test/python/transpiler/test_swap_cancellation.py

@@ -0,0 +1,52 @@
+# 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.
+
+"""Tests for pass cancelling 2 consecutive SWAPs on the same qubits."""
+
+from qiskit import QuantumCircuit
+from qiskit.transpiler.passes import SWAPCancellation
+from qiskit.test import QiskitTestCase
+
+
+class TestSWAPCancellation(QiskitTestCase):
+    """Test the SWAPCancellation pass."""
+
+    def test_pass_swap_cancellation_simple(self):
+        """Test a simple swap cancellation pass, symmetric wires """
+        circuit = QuantumCircuit(2)
+        circuit.swap(0, 1)
+        circuit.swap(1, 0)
+
+        out_circuit = SWAPCancellation()(circuit)
+        resources_after = out_circuit.count_ops()
+
+        self.assertNotIn('swap', resources_after)
+
+    def test_pass_swap_cancellation_many(self):
+        """Test the swap cancellation pass.
+
+        It should cancel consecutive swap pairs on same qubits.
+        """
+        circuit = QuantumCircuit(2)
+        circuit.h(0)
+        circuit.h(0)
+        circuit.swap(0, 1)
+        circuit.swap(0, 1)
+        circuit.swap(0, 1)
+        circuit.swap(0, 1)
+        circuit.swap(1, 0)
+        circuit.swap(1, 0)
+
+        out_circuit = SWAPCancellation()(circuit)
+        resources_after = out_circuit.count_ops()
+
+        self.assertNotIn('swap', resources_after)