Rework handling of instruction durations in preset pass managers (#12183

)

* Rework use of instruction durations, move logic from transpile function to individual passes.

* Apply review feedback on reno

qiskit/compiler/transpiler.py

@@ -337,7 +337,7 @@ def callback_func(**kwargs):
 
     _skip_target = False
     _given_inst_map = bool(inst_map)  # check before inst_map is overwritten
-    # If a target is specified have it override any implicit selections from a backend
+    # If a target is specified, have it override any implicit selections from a backend
     if target is not None:
         if coupling_map is None:
             coupling_map = target.build_coupling_map()
@@ -354,7 +354,7 @@ def callback_func(**kwargs):
         if backend_properties is None:
             backend_properties = target_to_backend_properties(target)
     # If target is not specified and any hardware constraint object is
-    # manually specified then do not use the target from the backend as
+    # manually specified, do not use the target from the backend as
     # it is invalidated by a custom basis gate list, custom coupling map,
     # custom dt or custom instruction_durations
     elif (
@@ -379,6 +379,7 @@ def callback_func(**kwargs):
     _check_circuits_coupling_map(circuits, coupling_map, backend)
 
     timing_constraints = _parse_timing_constraints(backend, timing_constraints)
+    instruction_durations = _parse_instruction_durations(backend, instruction_durations, dt)
 
     if _given_inst_map and inst_map.has_custom_gate() and target is not None:
         # Do not mutate backend target
@@ -391,51 +392,6 @@ def callback_func(**kwargs):
         if translation_method is None and hasattr(backend, "get_translation_stage_plugin"):
             translation_method = backend.get_translation_stage_plugin()
 
-    if instruction_durations or dt:
-        # If durations are provided and there is more than one circuit
-        # we need to serialize the execution because the full durations
-        # is dependent on the circuit calibrations which are per circuit
-        if len(circuits) > 1:
-            out_circuits = []
-            for circuit in circuits:
-                instruction_durations = _parse_instruction_durations(
-                    backend, instruction_durations, dt, circuit
-                )
-                pm = generate_preset_pass_manager(
-                    optimization_level,
-                    backend=backend,
-                    target=target,
-                    basis_gates=basis_gates,
-                    inst_map=inst_map,
-                    coupling_map=coupling_map,
-                    instruction_durations=instruction_durations,
-                    backend_properties=backend_properties,
-                    timing_constraints=timing_constraints,
-                    initial_layout=initial_layout,
-                    layout_method=layout_method,
-                    routing_method=routing_method,
-                    translation_method=translation_method,
-                    scheduling_method=scheduling_method,
-                    approximation_degree=approximation_degree,
-                    seed_transpiler=seed_transpiler,
-                    unitary_synthesis_method=unitary_synthesis_method,
-                    unitary_synthesis_plugin_config=unitary_synthesis_plugin_config,
-                    hls_config=hls_config,
-                    init_method=init_method,
-                    optimization_method=optimization_method,
-                    _skip_target=_skip_target,
-                )
-                out_circuits.append(pm.run(circuit, callback=callback, num_processes=num_processes))
-            for name, circ in zip(output_name, out_circuits):
-                circ.name = name
-                end_time = time()
-            _log_transpile_time(start_time, end_time)
-            return out_circuits
-        else:
-            instruction_durations = _parse_instruction_durations(
-                backend, instruction_durations, dt, circuits[0]
-            )
-
     pm = generate_preset_pass_manager(
         optimization_level,
         backend=backend,
@@ -460,7 +416,7 @@ def callback_func(**kwargs):
         optimization_method=optimization_method,
         _skip_target=_skip_target,
     )
-    out_circuits = pm.run(circuits, callback=callback)
+    out_circuits = pm.run(circuits, callback=callback, num_processes=num_processes)
     for name, circ in zip(output_name, out_circuits):
         circ.name = name
     end_time = time()
@@ -535,32 +491,20 @@ def _parse_initial_layout(initial_layout):
     return initial_layout
 
 
-def _parse_instruction_durations(backend, inst_durations, dt, circuit):
+def _parse_instruction_durations(backend, inst_durations, dt):
     """Create a list of ``InstructionDuration``s. If ``inst_durations`` is provided,
     the backend will be ignored, otherwise, the durations will be populated from the
-    backend. If any circuits have gate calibrations, those calibration durations would
-    take precedence over backend durations, but be superceded by ``inst_duration``s.
+    backend.
     """
+    final_durations = InstructionDurations()
     if not inst_durations:
         backend_durations = InstructionDurations()
         if backend is not None:
             backend_durations = backend.instruction_durations
-
-    circ_durations = InstructionDurations()
-    if not inst_durations:
-        circ_durations.update(backend_durations, dt or backend_durations.dt)
-
-    if circuit.calibrations:
-        cal_durations = []
-        for gate, gate_cals in circuit.calibrations.items():
-            for (qubits, parameters), schedule in gate_cals.items():
-                cal_durations.append((gate, qubits, parameters, schedule.duration))
-        circ_durations.update(cal_durations, circ_durations.dt)
-
-    if inst_durations:
-        circ_durations.update(inst_durations, dt or getattr(inst_durations, "dt", None))
-
-    return circ_durations
+        final_durations.update(backend_durations, dt or backend_durations.dt)
+    else:
+        final_durations.update(inst_durations, dt or getattr(inst_durations, "dt", None))
+    return final_durations
 
 
 def _parse_approximation_degree(approximation_degree):

qiskit/transpiler/passes/scheduling/dynamical_decoupling.py

@@ -1,6 +1,6 @@
 # This code is part of Qiskit.
 #
-# (C) Copyright IBM 2021.
+# (C) Copyright IBM 2021, 2024.
 #
 # 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
@@ -20,6 +20,7 @@
 from qiskit.dagcircuit import DAGOpNode, DAGInNode
 from qiskit.quantum_info.operators.predicates import matrix_equal
 from qiskit.synthesis.one_qubit import OneQubitEulerDecomposer
+from qiskit.transpiler import InstructionDurations
 from qiskit.transpiler.passes.optimization import Optimize1qGates
 from qiskit.transpiler.basepasses import TransformationPass
 from qiskit.transpiler.exceptions import TranspilerError
@@ -168,6 +169,8 @@ def run(self, dag):
         if dag.duration is None:
             raise TranspilerError("DD runs after circuit is scheduled.")
 
+        durations = self._update_inst_durations(dag)
+
         num_pulses = len(self._dd_sequence)
         sequence_gphase = 0
         if num_pulses != 1:
@@ -208,7 +211,7 @@ def run(self, dag):
             for index, gate in enumerate(self._dd_sequence):
                 gate = gate.to_mutable()
                 self._dd_sequence[index] = gate
-                gate.duration = self._durations.get(gate, physical_qubit)
+                gate.duration = durations.get(gate, physical_qubit)
 
                 dd_sequence_duration += gate.duration
             index_sequence_duration_map[physical_qubit] = dd_sequence_duration
@@ -277,6 +280,26 @@ def run(self, dag):
 
         return new_dag
 
+    def _update_inst_durations(self, dag):
+        """Update instruction durations with circuit information. If the dag contains gate
+        calibrations and no instruction durations were provided through the target or as a
+        standalone input, the circuit calibration durations will be used.
+        The priority order for instruction durations is: target > standalone > circuit.
+        """
+        circ_durations = InstructionDurations()
+
+        if dag.calibrations:
+            cal_durations = []
+            for gate, gate_cals in dag.calibrations.items():
+                for (qubits, parameters), schedule in gate_cals.items():
+                    cal_durations.append((gate, qubits, parameters, schedule.duration))
+            circ_durations.update(cal_durations, circ_durations.dt)
+
+        if self._durations is not None:
+            circ_durations.update(self._durations, getattr(self._durations, "dt", None))
+
+        return circ_durations
+
     def __gate_supported(self, gate: Gate, qarg: int) -> bool:
         """A gate is supported on the qubit (qarg) or not."""
         if self._target is None or self._target.instruction_supported(gate.name, qargs=(qarg,)):

qiskit/transpiler/passes/scheduling/padding/dynamical_decoupling.py

@@ -1,6 +1,6 @@
 # This code is part of Qiskit.
 #
-# (C) Copyright IBM 2021.
+# (C) Copyright IBM 2021, 2024.
 #
 # 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
@@ -179,9 +179,31 @@ def __init__(
                         f"{gate.name} in dd_sequence is not supported in the target"
                     )
 
+    def _update_inst_durations(self, dag):
+        """Update instruction durations with circuit information. If the dag contains gate
+        calibrations and no instruction durations were provided through the target or as a
+        standalone input, the circuit calibration durations will be used.
+        The priority order for instruction durations is: target > standalone > circuit.
+        """
+        circ_durations = InstructionDurations()
+
+        if dag.calibrations:
+            cal_durations = []
+            for gate, gate_cals in dag.calibrations.items():
+                for (qubits, parameters), schedule in gate_cals.items():
+                    cal_durations.append((gate, qubits, parameters, schedule.duration))
+            circ_durations.update(cal_durations, circ_durations.dt)
+
+        if self._durations is not None:
+            circ_durations.update(self._durations, getattr(self._durations, "dt", None))
+
+        return circ_durations
+
     def _pre_runhook(self, dag: DAGCircuit):
         super()._pre_runhook(dag)
 
+        durations = self._update_inst_durations(dag)
+
         num_pulses = len(self._dd_sequence)
 
         # Check if physical circuit is given
@@ -245,7 +267,7 @@ def _pre_runhook(self, dag: DAGCircuit):
                             f"is not acceptable in {self.__class__.__name__} pass."
                         )
                 except KeyError:
-                    gate_length = self._durations.get(gate, physical_index)
+                    gate_length = durations.get(gate, physical_index)
                 sequence_lengths.append(gate_length)
                 # Update gate duration. This is necessary for current timeline drawer, i.e. scheduled.
                 gate = gate.to_mutable()

qiskit/transpiler/passes/scheduling/time_unit_conversion.py

@@ -1,6 +1,6 @@
 # This code is part of Qiskit.
 #
-# (C) Copyright IBM 2021.
+# (C) Copyright IBM 2021, 2024.
 #
 # 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
@@ -51,6 +51,7 @@ def __init__(self, inst_durations: InstructionDurations = None, target: Target =
         self.inst_durations = inst_durations or InstructionDurations()
         if target is not None:
             self.inst_durations = target.durations()
+        self._durations_provided = inst_durations is not None or target is not None
 
     def run(self, dag: DAGCircuit):
         """Run the TimeUnitAnalysis pass on `dag`.
@@ -64,8 +65,11 @@ def run(self, dag: DAGCircuit):
         Raises:
             TranspilerError: if the units are not unifiable
         """
+
+        inst_durations = self._update_inst_durations(dag)
+
         # Choose unit
-        if self.inst_durations.dt is not None:
+        if inst_durations.dt is not None:
             time_unit = "dt"
         else:
             # Check what units are used in delays and other instructions: dt or SI or mixed
@@ -75,7 +79,7 @@ def run(self, dag: DAGCircuit):
                     "Fail to unify time units in delays. SI units "
                     "and dt unit must not be mixed when dt is not supplied."
                 )
-            units_other = self.inst_durations.units_used()
+            units_other = inst_durations.units_used()
             if self._unified(units_other) == "mixed":
                 raise TranspilerError(
                     "Fail to unify time units in instruction_durations. SI units "
@@ -96,7 +100,7 @@ def run(self, dag: DAGCircuit):
         # Make units consistent
         for node in dag.op_nodes():
             try:
-                duration = self.inst_durations.get(
+                duration = inst_durations.get(
                     node.op, [dag.find_bit(qarg).index for qarg in node.qargs], unit=time_unit
                 )
             except TranspilerError:
@@ -108,6 +112,26 @@ def run(self, dag: DAGCircuit):
         self.property_set["time_unit"] = time_unit
         return dag
 
+    def _update_inst_durations(self, dag):
+        """Update instruction durations with circuit information. If the dag contains gate
+        calibrations and no instruction durations were provided through the target or as a
+        standalone input, the circuit calibration durations will be used.
+        The priority order for instruction durations is: target > standalone > circuit.
+        """
+        circ_durations = InstructionDurations()
+
+        if dag.calibrations:
+            cal_durations = []
+            for gate, gate_cals in dag.calibrations.items():
+                for (qubits, parameters), schedule in gate_cals.items():
+                    cal_durations.append((gate, qubits, parameters, schedule.duration))
+            circ_durations.update(cal_durations, circ_durations.dt)
+
+        if self._durations_provided:
+            circ_durations.update(self.inst_durations, getattr(self.inst_durations, "dt", None))
+
+        return circ_durations
+
     @staticmethod
     def _units_used_in_delays(dag: DAGCircuit) -> Set[str]:
         units_used = set()

releasenotes/notes/rework-inst-durations-passes-28c78401682e22c0.yaml

@@ -0,0 +1,15 @@
+---
+fixes:
+  - |
+    The internal handling of custom circuit calibrations and :class:`.InstructionDurations`
+    has been offloaded from the :func:`.transpile` function to the individual transpiler passes: 
+    :class:`qiskit.transpiler.passes.scheduling.DynamicalDecoupling`,
+    :class:`qiskit.transpiler.passes.scheduling.padding.DynamicalDecoupling`. Before, 
+    instruction durations from circuit calibrations would not be taken into account unless 
+    they were manually incorporated into `instruction_durations` input argument, but the passes
+    that need it now analyze the circuit and pick the most relevant duration value according 
+    to the following priority order: target > custom input > circuit calibrations.
+
+  - |
+    Fixed a bug in :func:`.transpile` where the ``num_processes`` argument would only be used
+    if ``dt`` or ``instruction_durations`` were provided.