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)
@@ -138,6 +143,7 @@ def _empty_dag_like(
         dag: DAGCircuit,
         pad_wires: bool = True,
         wire_map: Optional[Dict[Qubit, Qubit]] = None,
+        ignore_idle: bool = False,
     ) -> DAGCircuit:
         """Create an empty dag like the input dag."""
         new_dag = DAGCircuit()
@@ -160,11 +166,17 @@ def _empty_dag_like(
         # trivial wire map if not provided, or if the top-level dag is used
         if not wire_map or pad_wires:
             wire_map = {wire: wire for wire in source_wire_dag.wires}
-        if dag.qregs:
+        if dag.qregs and self._schedule_idle_qubits or not ignore_idle:
             for qreg in source_wire_dag.qregs.values():
                 new_dag.add_qreg(qreg)
         else:
-            new_dag.add_qubits([wire_map[qubit] for qubit in source_wire_dag.qubits])
+            new_dag.add_qubits(
+                [
+                    wire_map[qubit]
+                    for qubit in source_wire_dag.qubits
+                    if qubit not in self._idle_qubits or not ignore_idle
+                ]
+            )
 
         # Don't add root cargs as these will not be padded.
         # Just focus on current block dag.
@@ -306,17 +318,30 @@ 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
 
     def _visit_block(
-        self, block: DAGCircuit, wire_map: Dict[Qubit, Qubit], pad_wires: bool = True
+        self,
+        block: DAGCircuit,
+        wire_map: Dict[Qubit, Qubit],
+        pad_wires: bool = True,
+        ignore_idle: bool = False,
     ) -> DAGCircuit:
         # Push the previous block dag onto the stack
         prev_node = self._prev_node
@@ -325,7 +350,7 @@ def _visit_block(
 
         prev_block_dag = self._block_dag
         self._block_dag = new_block_dag = self._empty_dag_like(
-            block, pad_wires, wire_map=wire_map
+            block, pad_wires, wire_map=wire_map, ignore_idle=ignore_idle
         )
 
         self._block_duration = 0
@@ -443,7 +468,10 @@ def _visit_control_flow_op(self, node: DAGNode) -> None:
             }
             new_node_block_dags.append(
                 self._visit_block(
-                    block_dag, pad_wires=not fast_path_node, wire_map=inner_wire_map
+                    block_dag,
+                    pad_wires=not fast_path_node,
+                    wire_map=inner_wire_map,
+                    ignore_idle=True,
                 )
             )
 
@@ -455,11 +483,19 @@ def _visit_control_flow_op(self, node: DAGNode) -> None:
         # Enforce that this control-flow operation contains all wires since it has now been padded
         # such that each qubit is scheduled within each block. Don't added all cargs as these will not
         # be padded.
+        if fast_path_node:
+            padded_qubits = node.qargs
+        elif not self._schedule_idle_qubits:
+            padded_qubits = [
+                q for q in self._block_dag.qubits if q not in self._idle_qubits
+            ]
+        else:
+            padded_qubits = self._block_dag.qubits
         self._apply_scheduled_op(
             block_idx,
             t0,
             new_control_flow_op,
-            self._map_wires(node.qargs) if fast_path_node else self._block_dag.qubits,
+            padded_qubits,
             self._map_wires(node.cargs),
         )
 
@@ -503,6 +539,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 +587,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,14 +174,17 @@ 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
                 non-multiple of the alignment constraint value is found.
             TranspilerError: When the coupling map is not supported (i.e., if degree > 3)
         """
 
-        super().__init__()
+        super().__init__(schedule_idle_qubits=schedule_idle_qubits)
         self._durations = durations
 
         # Enforce list of DD sequences

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

@@ -230,7 +230,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])
@@ -258,6 +257,7 @@ def test_delete(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")