Add option to no pad idle qubits in scheduling passes

qiskit_ibm_provider/transpiler/passes/scheduling/block_base_padder.py

@@ -62,7 +62,7 @@ class BlockBasePadder(TransformationPass):
     which may result in violation of hardware alignment constraints.
     """
 
-    def __init__(self) -> None:
+    def __init__(self, schedule_idle_qubits: bool = False) -> None:
         self._node_start_time = None
         self._node_block_dags = None
         self._idle_after: Optional[Dict[Qubit, int]] = None
@@ -84,7 +84,8 @@ def __init__(self) -> None:
 
         self._dirty_qubits: Set[Qubit] = set()
         # Qubits that are dirty in the circuit.
-
+        self._schedule_idle_qubits = schedule_idle_qubits
+        self._idle_qubits: Set[Qubit] = set()
         super().__init__()
 
     def run(self, dag: DAGCircuit) -> DAGCircuit:
@@ -100,6 +101,10 @@ def run(self, dag: DAGCircuit) -> DAGCircuit:
             TranspilerError: When a particular node is not scheduled, likely some transform pass
                 is inserted before this node is called.
         """
+        if not self._schedule_idle_qubits:
+            self._idle_qubits = set(
+                wire for wire in dag.idle_wires() if isinstance(wire, Qubit)
+            )
         self._pre_runhook(dag)
 
         self._init_run(dag)
@@ -155,10 +160,13 @@ def _empty_dag_like(
         # not be equivalent to one written manually as bits will not
         # be defined on registers like in the test case.
 
-        source_wire_dag = self._root_dag if pad_wires else dag
+        if not self._schedule_idle_qubits:
+            source_wire_dag = dag
+        else:
+            source_wire_dag = self._root_dag if pad_wires else dag
 
         # trivial wire map if not provided, or if the top-level dag is used
-        if not wire_map or pad_wires:
+        if not wire_map or (pad_wires and self._schedule_idle_qubits):
             wire_map = {wire: wire for wire in source_wire_dag.wires}
         if dag.qregs:
             for qreg in source_wire_dag.qregs.values():
@@ -306,11 +314,20 @@ def _add_block_terminating_barrier(
 
         if needs_terminating_barrier:
             # Terminate with a barrier to ensure topological ordering does not slide past
+            if self._schedule_idle_qubits:
+                barrier = Barrier(self._block_dag.num_qubits())
+                qubits = self._block_dag.qubits
+            else:
+                barrier = Barrier(self._block_dag.num_qubits() - len(self._idle_qubits))
+                qubits = [
+                    x for x in self._block_dag.qubits if x not in self._idle_qubits
+                ]
+
             barrier_node = self._apply_scheduled_op(
                 block_idx,
                 time,
-                Barrier(self._block_dag.num_qubits()),
-                self._block_dag.qubits,
+                barrier,
+                qubits,
                 [],
             )
             barrier_node.op.duration = 0
@@ -420,7 +437,7 @@ def _visit_control_flow_op(self, node: DAGNode) -> None:
         self._add_block_terminating_barrier(block_idx, t0, node)
 
         # Only pad non-fast path nodes
-        fast_path_node = node in self._fast_path_nodes
+        fast_path_node = node in self._fast_path_nodes or not self._schedule_idle_qubits
 
         # TODO: This is a hack required to tie nodes of control-flow
         # blocks across the scheduler and block_base_padder. This is
@@ -503,6 +520,8 @@ def _visit_generic(self, node: DAGNode) -> None:
         self._block_duration = max(self._block_duration, t1)
 
         for bit in self._map_wires(node.qargs):
+            if bit in self._idle_qubits:
+                continue
             # Fill idle time with some sequence
             if t0 - self._idle_after.get(bit, 0) > 0:
                 # Find previous node on the wire, i.e. always the latest node on the wire
@@ -549,6 +568,8 @@ def _terminate_block(self, block_duration: int, block_idx: int) -> None:
     def _pad_until_block_end(self, block_duration: int, block_idx: int) -> None:
         # Add delays until the end of circuit.
         for bit in self._block_dag.qubits:
+            if bit in self._idle_qubits:
+                continue
             idle_after = self._idle_after.get(bit, 0)
             if block_duration - idle_after > 0:
                 node = self._block_dag.output_map[bit]

qiskit_ibm_provider/transpiler/passes/scheduling/dynamical_decoupling.py

@@ -121,6 +121,7 @@ def __init__(
         insert_multiple_cycles: bool = False,
         coupling_map: CouplingMap = None,
         alt_spacings: Optional[Union[List[List[float]], List[float]]] = None,
+        schedule_idle_qubits: bool = False,
     ):
         """Dynamical decoupling initializer.
 
@@ -173,6 +174,9 @@ def __init__(
             alt_spacings: A list of lists of spacings between the DD gates, for the second subcircuit,
                 as determined by the coupling map. If None, a balanced spacing that is staggered with
                 respect to the first subcircuit will be used [d, d, d, ..., d, d, 0].
+            schedule_idle_qubits: Set to true if you'd like a delay inserted on idle qubits.
+                This is useful for timeline visualizations, but may cause issues
+                for execution on large backends.
         Raises:
             TranspilerError: When invalid DD sequence is specified.
             TranspilerError: When pulse gate with the duration which is

qiskit_ibm_provider/transpiler/passes/scheduling/pad_delay.py

@@ -50,13 +50,16 @@ class PadDelay(BlockBasePadder):
     See :class:`BlockBasePadder` pass for details.
     """
 
-    def __init__(self, fill_very_end: bool = True):
+    def __init__(self, fill_very_end: bool = True, schedule_idle_qubits: bool = False):
         """Create new padding delay pass.
 
         Args:
             fill_very_end: Set ``True`` to fill the end of circuit with delay.
+            schedule_idle_qubits: Set to true if you'd like a delay inserted on idle qubits.
+                This is useful for timeline visualizations, but may cause issues for execution
+                on large backends.
         """
-        super().__init__()
+        super().__init__(schedule_idle_qubits=schedule_idle_qubits)
         self.fill_very_end = fill_very_end
 
     def _pad(

releasenotes/notes/scheduling-no-idle-bbadd1e95d2a71b8.yaml

@@ -0,0 +1,20 @@
+---
+features:
+  - |
+    Added a new flag, ``schedule_idle_qubits`` to the constructor for the
+    :class:`.PadDelay` and :class:`.PadDynamicalDecoupling` passes. This
+    flag when set to ``True`` will have the scheduling passes insert a full
+    circuit duration delay on any idle qubits in the circuit.
+upgrade:
+  - |
+    The default behavior of the :class:`.PadDelay` and
+    :class:`.PadDynamicalDecoupling` passes for idle qubits in the circuit
+    have changed. Previously, by default the passes would schedule any idle
+    qubits in the circuit by inserting a delay equal to the full circuit
+    duration. This has been changed so by default only active qubits are
+    scheduled. This change was made because the extra delays were additional
+    overhead in the job payload that were effectively a no-op so they added
+    extra overhead to job submission for no gain. If you need to restore
+    the previous behavior you can instantiate :class:`.PadDelay` or
+    :class:`.PadDynamicalDecoupling` with the keyword argument
+    ``schedule_idle_qubits=True`` which will restore the previous behavior.

test/unit/test_account.py

@@ -275,7 +275,6 @@ def test_list(self):
                 ).to_saved_format(),
             }
         ), self.subTest("filtered list of accounts"):
-
             accounts = list(AccountManager.list(channel="ibm_quantum").keys())
             self.assertEqual(len(accounts), 2)
             self.assertListEqual(accounts, ["key2", _DEFAULT_ACCOUNT_NAME_IBM_QUANTUM])
@@ -321,6 +320,7 @@ def test_delete_auth(self):
     "urls": {"https": "127.0.0.1", "username_ntlm": "", "password_ntlm": ""}
 }
 
+
 # TODO: update and reenable test cases to work with qiskit-ibm-provider
 # NamedTemporaryFiles not supported in Windows
 @skipIf(os.name == "nt", "Test not supported in Windows")