Add a kwarg for num_processes in the transpiler (#11554)

* Modify transpiler to parse all parameters before starting to run circuits. Create new function to run circuits and pass it to parallel map.

* input num_processes argument and allow None values

* Revert Changes to resolve conficts

* Allow value none for num_processes

* fix import for merged changes from main

* add reno and Tests

* remove unused import

* pass num_processes argument through the pass manager instead of through the transpiler. updated documentation

qiskit/compiler/transpiler.py

@@ -67,6 +67,7 @@ def transpile(  # pylint: disable=too-many-return-statements
     init_method: Optional[str] = None,
     optimization_method: Optional[str] = None,
     ignore_backend_supplied_default_methods: bool = False,
+    num_processes: Optional[int] = None,
 ) -> _CircuitT:
     """Transpile one or more circuits, according to some desired transpilation targets.
 
@@ -257,6 +258,11 @@ def callback_func(**kwargs):
             to support custom compilation target-specific passes/plugins which support
             backend-specific compilation techniques. If you'd prefer that these defaults were
             not used this option is used to disable those backend-specific defaults.
+        num_processes: The maximum number of parallel processes to launch for this call to
+            transpile if parallel execution is enabled. This argument overrides
+            ``num_processes`` in the user configuration file, and the ``QISKIT_NUM_PROCS``
+            environment variable. If set to ``None`` the system default or local user configuration
+            will be used.
 
     Returns:
         The transpiled circuit(s).
@@ -390,7 +396,7 @@ def callback_func(**kwargs):
                     optimization_method=optimization_method,
                     _skip_target=_skip_target,
                 )
-                out_circuits.append(pm.run(circuit, callback=callback))
+                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()

qiskit/passmanager/passmanager.py

@@ -171,6 +171,7 @@ def run(
         self,
         in_programs: Any | list[Any],
         callback: Callable = None,
+        num_processes: int = None,
         **kwargs,
     ) -> Any:
         """Run all the passes on the specified ``in_programs``.
@@ -204,6 +205,10 @@ def callback_func(**kwargs):
                         running_time = kwargs['running_time']
                         count = kwargs['count']
                         ...
+            num_processes: The maximum number of parallel processes to launch if parallel
+                execution is enabled. This argument overrides ``num_processes`` in the user
+                configuration file, and the ``QISKIT_NUM_PROCS`` environment variable. If set
+                to ``None`` the system default or local user configuration will be used.
 
             kwargs: Arbitrary arguments passed to the compiler frontend and backend.
 
@@ -240,6 +245,7 @@ def callback_func(**kwargs):
             _run_workflow_in_new_process,
             values=in_programs,
             task_kwargs={"pass_manager_bin": dill.dumps(self)},
+            num_processes=num_processes,
         )
 
     def to_flow_controller(self) -> FlowControllerLinear:

qiskit/transpiler/passmanager.py

@@ -129,6 +129,7 @@ def run(
         circuits: _CircuitsT,
         output_name: str | None = None,
         callback: Callable = None,
+        num_processes: int = None,
     ) -> _CircuitsT:
         """Run all the passes on the specified ``circuits``.
 
@@ -167,6 +168,10 @@ def callback_func(**kwargs):
                         property_set = kwargs['property_set']
                         count = kwargs['count']
                         ...
+            num_processes: The maximum number of parallel processes to launch if parallel
+                execution is enabled. This argument overrides ``num_processes`` in the user
+                configuration file, and the ``QISKIT_NUM_PROCS`` environment variable. If set
+                to ``None`` the system default or local user configuration will be used.
 
         Returns:
             The transformed circuit(s).
@@ -178,6 +183,7 @@ def callback_func(**kwargs):
             in_programs=circuits,
             callback=callback,
             output_name=output_name,
+            num_processes=num_processes,
         )
 
     def draw(self, filename=None, style=None, raw=False):
@@ -385,9 +391,10 @@ def run(
         circuits: _CircuitsT,
         output_name: str | None = None,
         callback: Callable | None = None,
+        num_processes: int = None,
     ) -> _CircuitsT:
         self._update_passmanager()
-        return super().run(circuits, output_name, callback)
+        return super().run(circuits, output_name, callback, num_processes=num_processes)
 
     def to_flow_controller(self) -> FlowControllerLinear:
         self._update_passmanager()

qiskit/utils/parallel.py

@@ -145,6 +145,8 @@ def func(_):
                     return 0
             parallel_map(func, list(range(10)));
     """
+    if num_processes is None:
+        num_processes = CPU_COUNT
     if len(values) == 0:
         return []
     if len(values) == 1:

releasenotes/notes/add-num-processes-kwarg-to-transpiler-3cb7f3457b54a535.yaml

@@ -0,0 +1,6 @@
+---
+features:
+  - |
+    Add new keyword argument ``num_processes`` to :func:`~qiskit.compiler.transpiler`.
+    Allows for overriding user configurations file entry ``num_processes`` and
+    the environment variable ``QISKIT_NUM_PROCS`` on a per transpile basis.

test/python/compiler/test_transpiler.py

@@ -153,6 +153,19 @@ def test_pass_manager_none(self):
         )
         self.assertEqual(circuit2, circuit3)
 
+    @data(0, 1, 2, 3)
+    def test_num_processes_kwarg_concurrent_default(self, num_processes):
+        """Test that num_processes kwarg works when the system default parallel is false"""
+        qc = QuantumCircuit(2)
+        qc.h(0)
+        qc.cx(0, 1)
+        qc.measure_all()
+        target = FakeMumbaiV2().target
+        res = transpile([qc] * 3, target=target, num_processes=num_processes)
+        self.assertIsInstance(res, list)
+        for circ in res:
+            self.assertIsInstance(circ, QuantumCircuit)
+
     def test_transpile_basis_gates_no_backend_no_coupling_map(self):
         """Verify transpile() works with no coupling_map or backend."""
         qr = QuantumRegister(2, "qr")
@@ -2210,6 +2223,19 @@ def test_parallel_with_target(self, opt_level):
         for circ in res:
             self.assertIsInstance(circ, QuantumCircuit)
 
+    @data(0, 1, 2, 3)
+    def test_parallel_num_processes_kwarg(self, num_processes):
+        """Test that num_processes kwarg works when the system default parallel is true"""
+        qc = QuantumCircuit(2)
+        qc.h(0)
+        qc.cx(0, 1)
+        qc.measure_all()
+        target = FakeMumbaiV2().target
+        res = transpile([qc] * 3, target=target, num_processes=num_processes)
+        self.assertIsInstance(res, list)
+        for circ in res:
+            self.assertIsInstance(circ, QuantumCircuit)
+
     @data(0, 1, 2, 3)
     def test_parallel_dispatch(self, opt_level):
         """Test that transpile in parallel works for all optimization levels."""