Add option to no pad idle qubits in scheduling passes (#725)

* Add option to no pad idle qubits in scheduling passes

This commit adds a new flag to the scheduling padding passes,
schedule_idle_qubits, which is used to opt-in to scheduling idle qubits.
By default these passes will only schedule qubits which are active. This
is a change in behavior from before where the previous default is all
qubits were scheduled. This was undesireable because it's adding
uneceesary instructions to the job payload which will need to be
processed which are effectively a no-op. The only real use case for
adding delays to idle wires is visualization, so it's left as an option
to re-enable the previous default.

Fixes #723

Co-authored-by: Thomas Alexander <[email protected]>

* Add missing inline type hint

* Don't put barriers on idle qubits either

* Fix formatting

* Add schedule_idle_qubits to parent constructor

* Reneable padding in control flow

* Omit idle qubits in control flow

* Run black

---------

Co-authored-by: Thomas Alexander <[email protected]>
Co-authored-by: Diego Ristè <[email protected]>
Co-authored-by: Kevin Tian <[email protected]>

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")