Fix reuse of ConsolidateBlocks instances (Qiskit#15258) (Qiskit#15266)

* Fix reuse of `ConsolidateBlocks` instances

The `_qubit_map` state added to `ConsolidateBlocks` in Qiskit#15083 (cb103d9)
was attempting to get round an awkward internal `PassManager` use to
pass the mapping into recursive calls.  Unfortunately, the state was not
cleared on exit to the pass, so re-use of the pass object would be
invalid, such as by calling the `PassManager.run` method on the result
of `generate_preset_pass_manager` more than once on circuits with
different sizes.

Rather than trying to plaster over the poor use of state, this commit
instead moves the runtime logic to be in the `PropertySet`, so re-use
troubles are limited to single executions.  It's still technically
possible for the `PropertySet` to become polluted if `ConsolidateBlocks`
throws an internal exception that is forcibly suppressed such that the
pipeline continues, but this is completely non-standard, would require
significant effort to achieve, and most likely there are large numbers
of transpiler passes that are not completely exception-safe anyway.

* Add test of `PassManager.run` over a list

On some platforms this will be multiprocessing, but it doesn't hurt to
try it for the platforms that aren't.

* Document use of property-set key

(cherry picked from commit 4019fd9)

Co-authored-by: Jake Lishman <[email protected]>

Author: mergify[bot]

qiskit/transpiler/passes/optimization/consolidate_blocks.py

@@ -62,12 +62,20 @@ class ConsolidateBlocks(TransformationPass):
     the same qubits into a Unitary node, to be resynthesized later,
     to a potentially more optimal subcircuit.
 
+    This pass reads the :class:`.PropertySet` key ``ConsolidateBlocks_qubit_map`` which it uses to
+    communicate with recursive worker instances of itself for control-flow operations.  The key
+    should never be observable in a user-facing :class:`.PassManager` pipeline (it is only set in
+    internal :class:`.PassManager` instances), but the pass may return incorrect results or error if
+    another pass sets this key.
+
     Notes:
         This pass assumes that the 'blocks_list' property that it reads is
         given such that blocks are in topological order. The blocks are
         collected by a previous pass, such as `Collect2qBlocks`.
     """
 
+    _QUBIT_MAP_KEY = "ConsolidateBlocks_qubit_map"
+
     def __init__(
         self,
         kak_basis_gate=None,
@@ -119,9 +127,6 @@ def __init__(
         else:
             self.decomposer = TwoQubitBasisDecomposer(CXGate())
             self.basis_gate_name = "cx"
-        # Statefully used within the control-flow block recursion to map back to top-level hardware
-        # qubits.
-        self._qubit_map = None
 
     def run(self, dag):
         """Run the ConsolidateBlocks pass on `dag`.
@@ -139,9 +144,9 @@ def run(self, dag):
         if runs is not None:
             runs = [[node._node_id for node in run] for run in runs]
 
-        if self._qubit_map is None:
-            self._qubit_map = list(range(dag.num_qubits()))
-
+        qubit_map = self.property_set.get(self._QUBIT_MAP_KEY, None)
+        if qubit_map is None:
+            qubit_map = list(range(dag.num_qubits()))
         consolidate_blocks(
             dag,
             self.decomposer._inner_decomposer,
@@ -151,9 +156,9 @@ def run(self, dag):
             basis_gates=self.basis_gates,
             blocks=blocks,
             runs=runs,
-            qubit_map=self._qubit_map,
+            qubit_map=qubit_map,
         )
-        dag = self._handle_control_flow_ops(dag)
+        dag = self._handle_control_flow_ops(dag, qubit_map)
 
         # Clear collected blocks and runs as they are no longer valid after consolidation
         if "run_list" in self.property_set:
@@ -163,7 +168,7 @@ def run(self, dag):
 
         return dag
 
-    def _handle_control_flow_ops(self, dag):
+    def _handle_control_flow_ops(self, dag, qubit_map):
         """
         This is similar to transpiler/passes/utils/control_flow.py except that the
         collect blocks is redone for the control flow blocks.
@@ -176,12 +181,10 @@ def _handle_control_flow_ops(self, dag):
         pass_manager.append(self)
 
         for node in dag.control_flow_op_nodes():
-            old_qubit_map, self._qubit_map = self._qubit_map, [
-                self._qubit_map[dag.find_bit(q).index] for q in node.qargs
-            ]
-            try:
-                new_op = node.op.replace_blocks(pass_manager.run(block) for block in node.op.blocks)
-            finally:
-                self._qubit_map = old_qubit_map
+            inner_qubit_map = [qubit_map[dag.find_bit(q).index] for q in node.qargs]
+            new_op = node.op.replace_blocks(
+                pass_manager.run(block, property_set={self._QUBIT_MAP_KEY: inner_qubit_map})
+                for block in node.op.blocks
+            )
             dag.substitute_node(node, new_op)
         return dag

releasenotes/notes/fix-consolidate-blocks-reuse-a18828999cef56ee.yaml

@@ -0,0 +1,12 @@
+---
+fixes:
+  - |
+    Fixed re-use of the same :class:`.ConsolidateBlocks` instance on multiple circuits, including
+    calls to :func:`.transpile` with more than one circuit and no process-based parallelization.  A
+    bug introduced in Qiskit 2.2.2 caused the pass to panic or produce invalid output if the same
+    instance was re-used on differing circuits.
+upgrade:
+  - |
+    :class:`.ConsolidateBlocks` now reads a :class:`.PropertySet` key ``ConsolidateBlocks_qubit_map``
+    on entry.  This key and its value are not public and should not be read or written to by other
+    passes.

test/python/compiler/test_transpiler.py

@@ -2333,6 +2333,55 @@ def test_single_qubit_circuit_deterministic_output(self, optimization_level):
         for i in range(10):
             self.assertEqual(isa_circs[0], isa_circs[i])
 
+    @data(0, 1, 2, 3)
+    def test_reuse_on_differing_circuits(self, optimization_level):
+        """Test that re-using the same `PassManager` instance on two different circuits is the same
+        as using new instances of the `PassManager`."""
+
+        def make_pm():
+            num_qubits = 12
+            target = Target()
+            target.add_instruction(
+                SXGate(),
+                {(i,): InstructionProperties(error=1e-5 * (i + 1)) for i in range(num_qubits)},
+            )
+            target.add_instruction(
+                RZGate(Parameter("a")), {(i,): InstructionProperties() for i in range(num_qubits)}
+            )
+            target.add_instruction(
+                Measure(),
+                {
+                    (i,): InstructionProperties(error=1e-3 * (num_qubits - i))
+                    for i in range(num_qubits)
+                },
+            )
+            target.add_instruction(
+                CZGate(),
+                {
+                    pair: InstructionProperties(error=1e-3 * sum(pair))
+                    for pair in CouplingMap.from_line(num_qubits)
+                },
+            )
+            return generate_preset_pass_manager(
+                optimization_level=optimization_level, target=target, seed_transpiler=2025_10_27
+            )
+
+        def ghz(num_qubits):
+            qc = QuantumCircuit(num_qubits, num_qubits)
+            qc.h(0)
+            for i in range(1, num_qubits):
+                qc.cx(0, i)
+            qc.measure(qc.qubits, qc.clbits)
+            return qc
+
+        circuits = [ghz(5), ghz(8), ghz(10)]
+        shared_pm = make_pm()
+        shared_circuits = [shared_pm.run(circuit) for circuit in circuits]
+        own_circuits = [make_pm().run(circuit) for circuit in circuits]
+        self.assertEqual(shared_circuits, own_circuits)
+        together_circuits = make_pm().run(circuits)
+        self.assertEqual(together_circuits, own_circuits)
+
 
 @ddt
 class TestPostTranspileIntegration(QiskitTestCase):

test/python/transpiler/test_consolidate_blocks.py

@@ -728,6 +728,24 @@ def test_collection_inside_control_flow(self):
         self.assertEqual(actual_outer.data[0].operation.blocks[0], block)
         self.assertEqual(actual_outer.data[1].operation.blocks[0].data[0].name, "unitary")
 
+    def test_pass_reuse(self):
+        """Test that the pass can be used more than once on different circuits."""
+        # It matters that these have different numbers of qubits.
+        hadamard = QuantumCircuit(1, 1)
+        hadamard.h(0)
+        hadamard.measure(0, 0)
+        bell = QuantumCircuit(2, 2)
+        bell.h(0)
+        bell.cx(0, 1)
+        bell.measure([0, 1], [0, 1])
+
+        def make_pass():
+            return ConsolidateBlocks(force_consolidate=True)
+
+        shared = make_pass()
+        self.assertEqual(shared(hadamard), make_pass()(hadamard))
+        self.assertEqual(shared(bell), make_pass()(bell))
+
     def test_invalid_python_data_does_not_panic(self):
         """If a user feeds in invalid/old data, Rust space shouldn't panic."""
         # It doesn't really matter _what_ the failure mode is, just that we should return a regular