"swap followed by measure" optimization pass

qiskit/transpiler/passes/__init__.py

@@ -21,6 +21,7 @@
 from .decompose import Decompose
 from .unroll_3q_or_more import Unroll3qOrMore
 from .commutation_analysis import CommutationAnalysis
+from .optimize_swap_before_measure import OptimizeSwapBeforeMeasure
 from .mapping.barrier_before_final_measurements import BarrierBeforeFinalMeasurements
 from .mapping.check_map import CheckMap
 from .mapping.check_cnot_direction import CheckCnotDirection

qiskit/transpiler/passes/optimize_swap_before_measure.py

@@ -0,0 +1,50 @@
+# -*- coding: utf-8 -*-
+
+# Copyright 2019, IBM.
+#
+# This source code is licensed under the Apache License, Version 2.0 found in
+# the LICENSE.txt file in the root directory of this source tree.
+
+
+"""
+Transpiler pass to remove swaps in front of measurements by re-targeting the classical bit
+ of the measure instruction.
+"""
+
+from qiskit.circuit import Measure
+from qiskit.extensions.standard import SwapGate
+from qiskit.transpiler.basepasses import TransformationPass
+from qiskit.dagcircuit import DAGCircuit
+
+
+class OptimizeSwapBeforeMeasure(TransformationPass):
+    """Remove the swaps followed by measurement (and adapt the measurement)"""
+
+    def run(self, dag):
+        """Return a new circuit that has been optimized."""
+        swaps = dag.op_nodes(SwapGate)
+        for swap in swaps:
+            final_successor = []
+            for successor in dag.successors(swap):
+                final_successor.append(successor.type == 'out' or (successor.type == 'op' and
+                                                                   successor.op.name == 'measure'))
+            if all(final_successor):
+                # the node swap needs to be removed and, if a measure follows, needs to be adapted
+                swap_qargs = swap.qargs
+                measure_layer = DAGCircuit()
+                for qreg in dag.qregs.values():
+                    measure_layer.add_qreg(qreg)
+                for creg in dag.cregs.values():
+                    measure_layer.add_creg(creg)
+                for successor in dag.successors(swap):
+                    if successor.type == 'op' and successor.op.name == 'measure':
+                        # replace measure node with a new one, where qargs is set with the "other"
+                        # swap qarg.
+                        dag.remove_op_node(successor)
+                        old_measure_qarg = successor.qargs[0]
+                        new_measure_qarg = swap_qargs[swap_qargs.index(old_measure_qarg) - 1]
+                        measure_layer.apply_operation_back(Measure(), [new_measure_qarg],
+                                                           [successor.cargs[0]])
+                dag.extend_back(measure_layer)
+                dag.remove_op_node(swap)
+        return dag

test/python/transpiler/test_optimize_swap_before_measure.py

@@ -0,0 +1,154 @@
+# -*- coding: utf-8 -*-
+
+# Copyright 2019, IBM.
+#
+# This source code is licensed under the Apache License, Version 2.0 found in
+# the LICENSE.txt file in the root directory of this source tree.
+
+"""Test OptimizeSwapBeforeMeasure pass"""
+
+import unittest
+
+from qiskit import QuantumRegister, QuantumCircuit, ClassicalRegister
+from qiskit.transpiler import PassManager, transpile_dag
+from qiskit.transpiler.passes import OptimizeSwapBeforeMeasure, DAGFixedPoint
+from qiskit.converters import circuit_to_dag
+from qiskit.test import QiskitTestCase
+
+
+class TestOptimizeSwapBeforeMeasure(QiskitTestCase):
+    """ Test swap-followed-by-measure optimizations. """
+
+    def test_optimize_1swap_1measure(self):
+        """ Remove a single swap
+            qr0:--X--m--       qr0:----
+                  |  |
+            qr1:--X--|--  ==>  qr1:--m-
+                     |               |
+            cr0:-----.--       cr0:--.-
+        """
+        qr = QuantumRegister(2, 'qr')
+        cr = ClassicalRegister(1, 'cr')
+        circuit = QuantumCircuit(qr, cr)
+        circuit.swap(qr[0], qr[1])
+        circuit.measure(qr[0], cr[0])
+        dag = circuit_to_dag(circuit)
+
+        expected = QuantumCircuit(qr, cr)
+        expected.measure(qr[1], cr[0])
+
+        pass_ = OptimizeSwapBeforeMeasure()
+        after = pass_.run(dag)
+
+        self.assertEqual(circuit_to_dag(expected), after)
+
+    def test_optimize_1swap_2measure(self):
+        """ Remove a single swap affecting two measurements
+            qr0:--X--m--         qr0:--m----
+                  |  |                 |
+            qr1:--X--|--m   ==>  qr1:--|--m-
+                     |  |              |  |
+            cr0:-----.--|--      cr0:--|--.-
+            cr1:--------.--      cr1:--.----
+        """
+        qr = QuantumRegister(2, 'qr')
+        cr = ClassicalRegister(2, 'cr')
+        circuit = QuantumCircuit(qr, cr)
+        circuit.swap(qr[0], qr[1])
+        circuit.measure(qr[0], cr[0])
+        circuit.measure(qr[1], cr[1])
+        dag = circuit_to_dag(circuit)
+
+        expected = QuantumCircuit(qr, cr)
+        expected.measure(qr[1], cr[0])
+        expected.measure(qr[0], cr[1])
+
+        pass_ = OptimizeSwapBeforeMeasure()
+        after = pass_.run(dag)
+
+        self.assertEqual(circuit_to_dag(expected), after)
+
+    def test_cannot_optimize(self):
+        """ Cannot optimize when swap is not at the end in all of the successors
+            qr0:--X-----m--
+                  |     |
+            qr1:--X-[H]-|--
+                        |
+            cr0:--------.--
+        """
+        qr = QuantumRegister(2, 'qr')
+        cr = ClassicalRegister(1, 'cr')
+        circuit = QuantumCircuit(qr, cr)
+        circuit.swap(qr[0], qr[1])
+        circuit.h(qr[1])
+        circuit.measure(qr[0], cr[0])
+        dag = circuit_to_dag(circuit)
+
+        pass_ = OptimizeSwapBeforeMeasure()
+        after = pass_.run(dag)
+
+        self.assertEqual(circuit_to_dag(circuit), after)
+
+
+class TestOptimizeSwapBeforeMeasureFixedPoint(QiskitTestCase):
+    """ Test swap-followed-by-measure optimizations in a transpiler, using fixed point. """
+
+    def test_optimize_undone_swap(self):
+        """ Remove redundant swap
+            qr0:--X--X--m--       qr0:--m---
+                  |  |  |               |
+            qr1:--X--X--|--  ==>  qr1:--|--
+                        |               |
+            cr0:--------.--       cr0:--.--
+        """
+        qr = QuantumRegister(2, 'qr')
+        cr = ClassicalRegister(1, 'cr')
+        circuit = QuantumCircuit(qr, cr)
+        circuit.swap(qr[0], qr[1])
+        circuit.swap(qr[0], qr[1])
+        circuit.measure(qr[0], cr[0])
+        dag = circuit_to_dag(circuit)
+
+        expected = QuantumCircuit(qr, cr)
+        expected.measure(qr[0], cr[0])
+
+        pass_manager = PassManager()
+        pass_manager.append(
+            [OptimizeSwapBeforeMeasure(), DAGFixedPoint()],
+            do_while=lambda property_set: not property_set['dag_fixed_point'])
+        after = transpile_dag(dag, pass_manager=pass_manager)
+
+        self.assertEqual(circuit_to_dag(expected), after)
+
+    def test_optimize_overlap_swap(self):
+        """ Remove two swaps that overlap
+            qr0:--X--------       qr0:--m--
+                  |                     |
+            qr1:--X--X-----       qr1:--|--
+                     |       ==>        |
+            qr2:-----X--m--       qr2:--|--
+                        |               |
+            cr0:--------.--       cr0:--.--
+        """
+        qr = QuantumRegister(3, 'qr')
+        cr = ClassicalRegister(1, 'cr')
+        circuit = QuantumCircuit(qr, cr)
+        circuit.swap(qr[0], qr[1])
+        circuit.swap(qr[1], qr[2])
+        circuit.measure(qr[2], cr[0])
+        dag = circuit_to_dag(circuit)
+
+        expected = QuantumCircuit(qr, cr)
+        expected.measure(qr[0], cr[0])
+
+        pass_manager = PassManager()
+        pass_manager.append(
+            [OptimizeSwapBeforeMeasure(), DAGFixedPoint()],
+            do_while=lambda property_set: not property_set['dag_fixed_point'])
+        after = transpile_dag(dag, pass_manager=pass_manager)
+
+        self.assertEqual(circuit_to_dag(expected), after)
+
+
+if __name__ == '__main__':
+    unittest.main()