Add StagedPassManager class for pass manager with defined stages (Qis…

…kit#6403)

* Add FullPassManager class for pass manager with defined stages

This commit adds a new PassManager subclass, FullPassManager. This class
is used to have a PassManager with a defined structure and stages for
the normal transpile workflow. The preset pass managers are then updated
to be FullPassManager objects they conform to the fixed structure. Having
a class with defined phases gives us flexibility in the future for making
the transpiler pluggable with external plugins (similar to what's done in
PR Qiskit#6124) and also have backend hook points before or after different
phases of the transpile.

Fixes Qiskit#5978

* Add docs

* Deduplicate preset passmanager construction

* Update docs

* Add dedicated scheduling stage to FullPassManager

* Add missing new UnitarySynthesis kwargs after rebase

* Use basis_translator as default method instead of basis

Co-authored-by: Kevin Krsulich <[email protected]>

* Rename FullPassManager StructuredPassManager

* Rename generate_scheduling_post_opt() generate_scheduling()

* Fix missing and incorrect arguments

* Fix more rebase issues

* Fix even more rebase issues

* Only run unroll3q on level 0-2 if coupling map is set

To preserve the behavior prior to this reorganization this commit makes
the Unroll3qorMore pass run if we have a coupling map set. This is
because this pass is only needed to be run for these optimization levels
so that the layout and routing passes can function (as they only work
with 2q gates). If we're not running these passes we shouldn't be
unrolling gates. This will fix the last 4 QAOA test failures post the
recent rebase as that was failing because we were trying to unroll an
unbound 4q hamiltonian gate when weren't before.

* Rework StructuredPassManager as a more dynamic StagedPassManager

This commit redefines the previous StructuredPassManager class into a
more dynamic StagedPassmanager. The StagedPassManager doesn't have fixed
hard coded stages anymore but instead lets users define their own
stages. It also adds implicit 'pre_' and 'post_' hook points for each
listed stage. This lets users dynamically define the stages based on a
particular use case.

* Fix docs

* Update internal pass set on each access

This commit updates the api for the StagedPassManager to refresh the
internal pass list on each access. This adds a little overhead but with
the tradeoff of making in place modifications to a stage's pass manager
reflected without needing to manually call an update method.

* Rename phases attribute to stages

* Fix lint

* Explicitly set name in qpy compat tests

The rework of the preset passmanager construction changes the import
order slightly and the number of circuits constructed in a session prior
to the qpy compat deserialization side generating equivalent circuits
for comparision has changed. This is causing all the deserialization
side numbers to change and the tests are now failing because the circuit
names are not equivalent. Since the exact name is a side effect of the
import order (based on the number of unnamed circuits created in the
session priort) it's not part of what the qpy tests are validating.
We're trying to assert that the naem is preserved loading a qpy file
across a version boundary. To fix this issues this commit adds an
explicit name to the generation for all the circuits to ensure that we
have a deterministic name for each circuit.

* Apply suggestions from code review

Co-authored-by: Luciano Bello <[email protected]>

* Run black

* Update type hint

* Remove out of date docstring note

* Update copyright header date in qiskit/transpiler/preset_passmanagers/common.py

Co-authored-by: Luciano Bello <[email protected]>

* Add check for invalid stage names

* Add backwards compatibility note

* Add docs on using StagedPassManager features with preset passmanagers

Co-authored-by: Kevin Krsulich <[email protected]>
Co-authored-by: Luciano Bello <[email protected]>
Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

qiskit/transpiler/__init__.py

@@ -595,6 +595,74 @@
 
       <br>
 
+.. dropdown:: Working with preset :class:`~.PassManager`
+   :animate: fade-in-slide-down
+
+   By default Qiskit includes functions to build preset :class:`~.PassManager` objects.
+   These preset passmangers are what get used by the :func:`~.transpile` function
+   for each optimization level. If you'd like to work directly with a
+   preset pass manager you can use the :func:`~.generate_preset_pass_manager`
+   function to easily generate one. For example:
+
+   .. code-block:: python
+
+       from qiskit.transpiler.preset_passmanager import generate_preset_pass_manager
+       from qiskit.providers.fake_provider import FakeLagosV2
+
+       backend = FakeLagosV2()
+       pass_manager = generate_preset_pass_manager(3, backend)
+
+   which will generate a :class:`~.StagedPassManager` object for optimization level 3
+   targetting the :class:`~.FakeLagosV2` backend (equivalent to what is used internally
+   by :func:`~.transpile` with ``backend=FakeLagosV2()`` and ``optimization_level=3``).
+   You can use this just like working with any other :class:`~.PassManager`. However,
+   because it is a :class:`~.StagedPassManager` it also makes it easy to compose and/or
+   replace stages of the pipeline. For example, if you wanted to run a custom scheduling
+   stage using dynamical decoupling (via the :class:`~.PadDynamicalDecoupling` pass) and
+   also add initial logical optimization prior to routing you would do something like
+   (building off the previous example):
+
+   .. code-block:: python
+
+       from qiskit.circuit.library import XGate, HGate, RXGate, PhaseGate, TGate, TdgGate
+       from qiskit.transpiler import PassManager
+       from qiskit.transpiler.passes import ALAPScheduleAnalysis, PadDynamicalDecoupling
+       from qiskit.transpiler.passes import CXCancellation, InverseCancellation
+
+       backend_durations = backend.target.durations()
+       dd_sequence = [XGate(), XGate()]
+       scheduling_pm = PassManager([
+           ALAPScheduleAnalysis(backend_durations),
+           PadDynamicalDecoupling(backend_durations, dd_sequence),
+       ])
+       inverse_gate_list = [
+           HGate(),
+           (RXGate(np.pi / 4), RXGate(-np.pi / 4)),
+           (PhaseGate(np.pi / 4), PhaseGate(-np.pi / 4)),
+           (TGate(), TdgGate()),
+
+       ])
+       logical_opt = PassManager([
+           CXCancellation(),
+           InverseCancellation([HGate(), (RXGate(np.pi / 4), RXGate(-np.pi / 4))
+       ])
+
+
+       # Add pre-layout stage to run extra logical optimization
+       pass_manager.pre_layout = logical_opt
+       # Set scheduling stage to custom pass manager
+       pass_manager.scheduling = scheduling_pm
+
+
+   Then when :meth:`~.StagedPassManager.run` is called on ``pass_manager`` the
+   ``logical_opt`` :class:`~.PassManager` will be called prior to the ``layout`` stage
+   and for the ``scheduling`` stage our custom :class:`~.PassManager`
+   ``scheduling_pm`` will be used.
+
+   .. raw:: html
+
+      <br>
+
 
 Transpiler API
 ==============
@@ -614,6 +682,7 @@
 .. autosummary::
    :toctree: ../stubs/
 
+   StagedPassManager
    PassManager
    PassManagerConfig
    PropertySet
@@ -669,6 +738,7 @@
 from .runningpassmanager import FlowController, ConditionalController, DoWhileController
 from .passmanager import PassManager
 from .passmanager_config import PassManagerConfig
+from .passmanager import StagedPassManager
 from .propertyset import PropertySet
 from .exceptions import TranspilerError, TranspilerAccessError
 from .fencedobjs import FencedDAGCircuit, FencedPropertySet

qiskit/transpiler/passes/optimization/consolidate_blocks.py

@@ -117,7 +117,7 @@ def run(self, dag):
         # If 1q runs are collected before consolidate those too
         runs = self.property_set["run_list"] or []
         for run in runs:
-            if run[0] in all_block_gates:
+            if any(gate in all_block_gates for gate in run):
                 continue
             if len(run) == 1 and not self._check_not_in_basis(
                 run[0].name, run[0].qargs, global_index_map
@@ -135,6 +135,11 @@ def run(self, dag):
                     continue
                 unitary = UnitaryGate(operator)
                 dag.replace_block_with_op(run, unitary, {qubit: 0}, cycle_check=False)
+        # Clear collected blocks and runs as they are no longer valid after consolidation
+        if "run_list" in self.property_set:
+            del self.property_set["run_list"]
+        if "block_list" in self.property_set:
+            del self.property_set["block_list"]
         return dag
 
     def _check_not_in_basis(self, gate_name, qargs, global_index_map):

qiskit/transpiler/passmanager.py

@@ -12,6 +12,8 @@
 
 """Manager for a set of Passes and their scheduling during transpilation."""
 
+import io
+import re
 from typing import Union, List, Callable, Dict, Any
 
 import dill
@@ -319,3 +321,185 @@ def passes(self) -> List[Dict[str, BasePass]]:
                 item["flow_controllers"] = {}
             ret.append(item)
         return ret
+
+
+class StagedPassManager(PassManager):
+    """A Pass manager pipeline built up of individual stages
+
+    This class enables building a compilation pipeline out of fixed stages.
+    Each ``StagedPassManager`` defines a list of stages which are executed in
+    a fixed order, and each stage is defined as a standalone :class:`~.PassManager`
+    instance. There are also ``pre_`` and ``post_`` stages for each defined stage.
+    This enables easily composing and replacing different stages and also adding
+    hook points to enable programmtic modifications to a pipeline. When using a staged
+    pass manager you are not able to modify the individual passes and are only able
+    to modify stages.
+
+    By default instances of StagedPassManager define a typical full compilation
+    pipeline from an abstract virtual circuit to one that is optimized and
+    capable of running on the specified backend. The default pre-defined stages are:
+
+    #. ``init`` - any initial passes that are run before we start embedding the circuit to the backend
+    #. ``layout`` - This stage runs layout and maps the virtual qubits in the
+       circuit to the physical qubits on a backend
+    #. ``routing`` - This stage runs after a layout has been run and will insert any
+       necessary gates to move the qubit states around until it can be run on
+       backend's compuling map.
+    #. ``translation`` - Perform the basis gate translation, in other words translate the gates
+       in the circuit to the target backend's basis set
+    #. ``optimization`` - The main optimization loop, this will typically run in a loop trying to
+       optimize the circuit until a condtion (such as fixed depth) is reached.
+    #. ``scheduling`` - Any hardware aware scheduling passes
+
+    .. note::
+
+        For backwards compatibility the relative positioning of these default
+        stages will remain stable moving forward. However, new stages may be
+        added to the default stage list in between current stages. For example,
+        in a future release a new phase, something like ``logical_optimization``, could be added
+        immediately after the existing ``init`` stage in the default stage list.
+        This would preserve compatibility for pre-existing ``StagedPassManager``
+        users as the relative positions of the stage are preserved so the behavior
+        will not change between releases.
+
+    These stages will be executed in order and any stage set to ``None`` will be skipped. If
+    a :class:`~qiskit.transpiler.PassManager` input is being used for more than 1 stage here
+    (for example in the case of a :class:`~.Pass` that covers both Layout and Routing) you will want
+    to set that to the earliest stage in sequence that it covers.
+    """
+
+    invalid_stage_regex = re.compile(
+        r"\s|\+|\-|\*|\/|\\|\%|\<|\>|\@|\!|\~|\^|\&|\:|\[|\]|\{|\}|\(|\)"
+    )
+
+    def __init__(self, stages=None, **kwargs):
+        """Initialize a new StagedPassManager object
+
+        Args:
+            stages (List[str]): An optional list of stages to use for this
+                instance. If this is not specified the default stages list
+                ``['init', 'layout', 'routing', 'translation', 'optimization', 'scheduling']`` is
+                used
+            kwargs: The initial :class:`~.PassManager` values for any stages
+                defined in ``stages``. If a argument is not defined the
+                stages will default to ``None`` indicating an empty/undefined
+                stage.
+
+        Raises:
+            AttributeError: If a stage in the input keyword arguments is not defined.
+            ValueError: If an invalid stage name is specified.
+        """
+        if stages is None:
+            self.stages = [
+                "init",
+                "layout",
+                "routing",
+                "translation",
+                "optimization",
+                "scheduling",
+            ]
+        else:
+            invalid_stages = [
+                stage for stage in stages if self.invalid_stage_regex.search(stage) is not None
+            ]
+            if invalid_stages:
+                with io.StringIO() as msg:
+                    msg.write(f"The following stage names are not valid: {invalid_stages[0]}")
+                    for invalid_stage in invalid_stages[1:]:
+                        msg.write(f", {invalid_stage}")
+                    raise ValueError(msg.getvalue())
+
+            self.stages = stages
+        super().__init__()
+        for stage in self.stages:
+            pre_stage = "pre_" + stage
+            post_stage = "post_" + stage
+            setattr(self, pre_stage, None)
+            setattr(self, stage, None)
+            setattr(self, post_stage, None)
+        for stage, pm in kwargs.items():
+            if (
+                stage not in self.stages
+                and not (stage.startswith("pre_") and stage[4:] in self.stages)
+                and not (stage.startswith("post_") and stage[5:] in self.stages)
+            ):
+                raise AttributeError(f"{stage} is not a valid stage.")
+            setattr(self, stage, pm)
+        self._update_passmanager()
+
+    def _update_passmanager(self):
+        self._pass_sets = []
+        for stage in self.stages:
+            # Add pre-stage PM
+            pre_stage_pm = getattr(self, "pre_" + stage, None)
+            if pre_stage_pm is not None:
+                self._pass_sets.extend(pre_stage_pm._pass_sets)
+            # Add stage PM
+            stage_pm = getattr(self, stage, None)
+            if stage_pm is not None:
+                self._pass_sets.extend(stage_pm._pass_sets)
+            # Add post-stage PM
+            post_stage_pm = getattr(self, "post_" + stage, None)
+            if post_stage_pm is not None:
+                self._pass_sets.extend(post_stage_pm._pass_sets)
+
+    def __setattr__(self, attr, value):
+        super().__setattr__(attr, value)
+        if (
+            attr in self.stages
+            or (attr.startswith("pre_") and attr[4:] not in self.stages)
+            or (attr.startswith("post_") and attr[5:] not in self.stages)
+        ):
+            self._update_passmanager()
+
+    def append(
+        self,
+        passes: Union[BasePass, List[BasePass]],
+        max_iteration: int = None,
+        **flow_controller_conditions: Any,
+    ) -> None:
+        raise NotImplementedError
+
+    def replace(
+        self,
+        index: int,
+        passes: Union[BasePass, List[BasePass]],
+        max_iteration: int = None,
+        **flow_controller_conditions: Any,
+    ) -> None:
+        raise NotImplementedError
+
+    # Raise NotImplemntedError on individual pass manipulation
+    def remove(self, index: int) -> None:
+        raise NotImplementedError
+
+    def __getitem(self, index):
+        self._update_passmanager()
+        return super().__getitem__(index)
+
+    def __len__(self):
+        self._update_passmanager()
+        return super().__len__()
+
+    def __setitem__(self, index, item):
+        raise NotImplementedError
+
+    def __add__(self, other):
+        raise NotImplementedError
+
+    def _create_running_passmanager(self) -> RunningPassManager:
+        self._update_passmanager()
+        return super()._create_running_passmanager()
+
+    def passes(self) -> List[Dict[str, BasePass]]:
+        self._update_passmanager()
+        return super().passes()
+
+    def run(
+        self,
+        circuits: Union[QuantumCircuit, List[QuantumCircuit]],
+        output_name: str = None,
+        callback: Callable = None,
+    ) -> Union[QuantumCircuit, List[QuantumCircuit]]:
+        self._update_passmanager()
+        return super().run(circuits, output_name, callback)

qiskit/transpiler/preset_passmanagers/__init__.py

@@ -136,7 +136,7 @@ def generate_preset_pass_manager(
             to use this option.
 
     Returns:
-        PassManager: The preset pass manager for the given options
+        StagedPassManager: The preset pass manager for the given options
 
     Raises:
         ValueError: if an invalid value for ``optimization_level`` is passed in.