add control flow to CommutativeCancellation pass

qiskit/circuit/commutation_checker.py

@@ -17,6 +17,7 @@
 import numpy as np
 
 from qiskit.circuit.operation import Operation
+from qiskit.circuit.controlflow import ControlFlowOp
 from qiskit.quantum_info.operators import Operator
 
 
@@ -88,6 +89,11 @@ def commute(
         ):
             return False
 
+        # Commutation of ControlFlow gates also not supported yet. This may be
+        # pending a control flow graph.
+        if isinstance(op1, ControlFlowOp) or isinstance(op2, ControlFlowOp):
+            return False
+
         # These lines are adapted from dag_dependency and say that two gates over
         # different quantum and classical bits necessarily commute. This is more
         # permissive that the check from commutation_analysis, as for example it

qiskit/transpiler/passes/optimization/commutative_cancellation.py

@@ -18,12 +18,14 @@
 from qiskit.circuit.quantumregister import QuantumRegister
 from qiskit.transpiler.exceptions import TranspilerError
 from qiskit.transpiler.basepasses import TransformationPass
+from qiskit.transpiler.passmanager import PassManager
 from qiskit.transpiler.passes.optimization.commutation_analysis import CommutationAnalysis
 from qiskit.dagcircuit import DAGCircuit, DAGInNode, DAGOutNode
 from qiskit.circuit.library.standard_gates.u1 import U1Gate
 from qiskit.circuit.library.standard_gates.rx import RXGate
 from qiskit.circuit.library.standard_gates.p import PhaseGate
 from qiskit.circuit.library.standard_gates.rz import RZGate
+from qiskit.circuit import ControlFlowOp
 
 
 _CUTOFF_PRECISION = 1e-5
@@ -97,7 +99,6 @@ def run(self, dag):
         #  - For 2qbit gates the key: (gate_type, first_qbit, sec_qbit, first commutation_set_id,
         #    sec_commutation_set_id), the value is the list gates that share the same gate type,
         #    qubits and commutation sets.
-
         for wire in dag.wires:
             wire_commutation_set = self.property_set["commutation_set"][wire]
 
@@ -186,4 +187,21 @@ def run(self, dag):
                 if np.mod(total_angle, (2 * np.pi)) < _CUTOFF_PRECISION:
                     dag.remove_op_node(run[0])
 
+        dag = self._handle_control_flow_ops(dag)
+
+        return dag
+
+    def _handle_control_flow_ops(self, dag):
+        """
+        This is similar to transpiler/passes/utils/control_flow.py except that the
+        commutation analysis is redone for the control flow blocks.
+        """
+
+        pass_manager = PassManager([CommutationAnalysis(), self])
+        for node in dag.op_nodes(ControlFlowOp):
+            mapped_blocks = []
+            for block in node.op.blocks:
+                new_circ = pass_manager.run(block)
+                mapped_blocks.append(new_circ)
+            node.op = node.op.replace_blocks(mapped_blocks)
         return dag

releasenotes/notes/add-control-flow-to-commutative-cancellation-pass-85fe310d911d9a00.yaml

@@ -0,0 +1,8 @@
+---
+features:
+  - |
+    Enabled performing the :class:`qiskit.transpiler.passes.CommutativeCancellation` pass inside the
+    blocks of :class:`qiskit.circuit.ControlFlowOp`. This pass reorders some commuting gates and
+    reduces resulting pairs of self-inverse gates. Previously, the blocks in control flow operations
+    were skipped by this pass. The new feature operates recursively, that is, it will act on control
+    flow operations inside blocks.

test/python/transpiler/test_commutative_cancellation.py

@@ -654,6 +654,114 @@ def test_basic_classical_wires(self):
         transpiled = PassManager([CommutativeCancellation()]).run(original)
         self.assertEqual(original, transpiled)
 
+    def test_simple_if_else(self):
+        """Test that the pass is not confused by if-else."""
+        base_test1 = QuantumCircuit(3, 3)
+        base_test1.x(1)
+        base_test1.cx(0, 1)
+        base_test1.x(1)
+
+        base_test2 = QuantumCircuit(3, 3)
+        base_test2.rz(0.1, 1)
+        base_test2.rz(0.1, 1)
+
+        test = QuantumCircuit(3, 3)
+        test.h(0)
+        test.x(0)
+        test.rx(0.2, 0)
+        test.measure(0, 0)
+        test.x(0)
+        test.if_else(
+            (test.clbits[0], True), base_test1.copy(), base_test2.copy(), test.qubits, test.clbits
+        )
+
+        expected = QuantumCircuit(3, 3)
+        expected.h(0)
+        expected.rx(np.pi + 0.2, 0)
+        expected.measure(0, 0)
+        expected.x(0)
+
+        expected_test1 = QuantumCircuit(3, 3)
+        expected_test1.cx(0, 1)
+
+        expected_test2 = QuantumCircuit(3, 3)
+        expected_test2.rz(0.2, 1)
+
+        expected.if_else(
+            (expected.clbits[0], True),
+            expected_test1.copy(),
+            expected_test2.copy(),
+            expected.qubits,
+            expected.clbits,
+        )
+
+        passmanager = PassManager([CommutationAnalysis(), CommutativeCancellation()])
+        new_circuit = passmanager.run(test)
+        self.assertEqual(new_circuit, expected)
+
+    def test_nested_control_flow(self):
+        """Test that the pass does not add barrier into nested control flow."""
+        level2_test = QuantumCircuit(2, 1)
+        level2_test.cz(0, 1)
+        level2_test.cz(0, 1)
+        level2_test.cz(0, 1)
+        level2_test.measure(0, 0)
+
+        level1_test = QuantumCircuit(2, 1)
+        level1_test.for_loop((0,), None, level2_test.copy(), level1_test.qubits, level1_test.clbits)
+        level1_test.h(0)
+        level1_test.h(0)
+        level1_test.measure(0, 0)
+
+        test = QuantumCircuit(2, 1)
+        test.while_loop((test.clbits[0], True), level1_test.copy(), test.qubits, test.clbits)
+        test.measure(0, 0)
+
+        level2_expected = QuantumCircuit(2, 1)
+        level2_expected.cz(0, 1)
+        level2_expected.measure(0, 0)
+
+        level1_expected = QuantumCircuit(2, 1)
+        level1_expected.for_loop(
+            (0,), None, level2_expected.copy(), level1_expected.qubits, level1_expected.clbits
+        )
+        level1_expected.measure(0, 0)
+
+        expected = QuantumCircuit(2, 1)
+        expected.while_loop(
+            (expected.clbits[0], True), level1_expected.copy(), expected.qubits, expected.clbits
+        )
+        expected.measure(0, 0)
+
+        passmanager = PassManager([CommutationAnalysis(), CommutativeCancellation()])
+        new_circuit = passmanager.run(test)
+        self.assertEqual(new_circuit, expected)
+
+    def test_cancellation_not_crossing_block_boundary(self):
+        """Test that the pass does cancel gates across control flow op block boundaries."""
+        test1 = QuantumCircuit(2, 2)
+        test1.x(1)
+        with test1.if_test((0, False)):
+            test1.cx(0, 1)
+            test1.x(1)
+
+        passmanager = PassManager([CommutationAnalysis(), CommutativeCancellation()])
+        new_circuit = passmanager.run(test1)
+        self.assertEqual(new_circuit, test1)
+
+    def test_cancellation_not_crossing_between_blocks(self):
+        """Test that the pass does cancel gates in different control flow ops."""
+        test2 = QuantumCircuit(2, 2)
+        with test2.if_test((0, True)):
+            test2.x(1)
+        with test2.if_test((0, True)):
+            test2.cx(0, 1)
+            test2.x(1)
+
+        passmanager = PassManager([CommutationAnalysis(), CommutativeCancellation()])
+        new_circuit = passmanager.run(test2)
+        self.assertEqual(new_circuit, test2)
+
 
 if __name__ == "__main__":
     unittest.main()