use concurrent.futures.Executor instead of multiprocessing pool to resolve conflict with duet

Author: NoureldinYosri

cirq-core/cirq/experiments/two_qubit_xeb.py

@@ -16,6 +16,7 @@
 
 from __future__ import annotations
 
+import concurrent.futures as cf
 import functools
 import itertools
 from collections.abc import Mapping, Sequence
@@ -409,7 +410,7 @@ def parallel_xeb_workflow(
     random_state: cirq.RANDOM_STATE_OR_SEED_LIKE = None,
     ax: plt.Axes | None = None,
     pairs: Sequence[tuple[cirq.GridQubit, cirq.GridQubit]] | None = None,
-    pool: multiprocessing.pool.Pool | None = None,
+    pool: multiprocessing.pool.Pool | cf.Executor | None = None,
     batch_size: int = 9,
     tags: Sequence[Any] = (),
     **plot_kwargs,
@@ -428,7 +429,7 @@ def parallel_xeb_workflow(
         ax: the plt.Axes to plot the device layout on. If not given,
             no plot is created.
         pairs: Pairs to use. If not specified, use all pairs between adjacent qubits.
-        pool: An optional multiprocessing pool.
+        pool: An optional pool.
         batch_size: We call `run_batch` on the sampler, which can speed up execution in certain
             environments. The number of (circuit, cycle_depth) tasks to be run in each batch
             is given by this number.

cirq-core/cirq/experiments/xeb_fitting.py

@@ -16,6 +16,7 @@
 
 from __future__ import annotations
 
+import concurrent.futures as cf
 import dataclasses
 from abc import ABC, abstractmethod
 from collections.abc import Iterable, Sequence
@@ -49,7 +50,7 @@ def benchmark_2q_xeb_fidelities(
     circuits: Sequence[cirq.Circuit],
     cycle_depths: Sequence[int] | None = None,
     param_resolver: cirq.ParamResolverOrSimilarType = None,
-    pool: multiprocessing.pool.Pool | None = None,
+    pool: multiprocessing.pool.Pool | cf.Executor | None = None,
 ) -> pd.DataFrame:
     """Simulate and benchmark two-qubit XEB circuits.
 
@@ -451,7 +452,7 @@ def characterize_phased_fsim_parameters_with_xeb(
     xatol: float = 1e-3,
     fatol: float = 1e-3,
     verbose: bool = True,
-    pool: multiprocessing.pool.Pool | None = None,
+    pool: multiprocessing.pool.Pool | cf.Executor | None = None,
 ) -> XEBCharacterizationResult:
     """Run a classical optimization to fit phased fsim parameters to experimental data, and
     thereby characterize PhasedFSim-like gates.
@@ -470,7 +471,7 @@ def characterize_phased_fsim_parameters_with_xeb(
         fatol: The `fatol` argument for Nelder-Mead. This is the absolute error for convergence
             in the function evaluation.
         verbose: Whether to print progress updates.
-        pool: An optional multiprocessing pool to execute circuit simulations in parallel.
+        pool: An optional pool to execute circuit simulations in parallel.
     """
     (pair,) = sampled_df['pair'].unique()
     initial_simplex, names = options.get_initial_simplex_and_names(
@@ -546,7 +547,7 @@ def characterize_phased_fsim_parameters_with_xeb_by_pair(
     initial_simplex_step_size: float = 0.1,
     xatol: float = 1e-3,
     fatol: float = 1e-3,
-    pool: multiprocessing.pool.Pool | None = None,
+    pool: multiprocessing.pool.Pool | cf.Executor | None = None,
 ) -> XEBCharacterizationResult:
     """Run a classical optimization to fit phased fsim parameters to experimental data, and
     thereby characterize PhasedFSim-like gates grouped by pairs.
@@ -570,7 +571,7 @@ def characterize_phased_fsim_parameters_with_xeb_by_pair(
             in the parameters.
         fatol: The `fatol` argument for Nelder-Mead. This is the absolute error for convergence
             in the function evaluation.
-        pool: An optional multiprocessing pool to execute pair optimization in parallel. Each
+        pool: An optional pool to execute pair optimization in parallel. Each
             optimization (and the simulations therein) runs serially.
     """
     pairs = sampled_df['pair'].unique()

cirq-core/cirq/experiments/z_phase_calibration.py

@@ -16,6 +16,7 @@
 
 from __future__ import annotations
 
+import concurrent.futures as cf
 import multiprocessing
 import multiprocessing.pool
 from collections.abc import Sequence
@@ -46,7 +47,7 @@ def z_phase_calibration_workflow(
     cycle_depths: Sequence[int] = tuple(np.arange(3, 100, 20)),
     random_state: cirq.RANDOM_STATE_OR_SEED_LIKE = None,
     atol: float = 1e-3,
-    num_workers_or_pool: int | multiprocessing.pool.Pool = -1,
+    num_workers_or_pool: int | multiprocessing.pool.Pool | cf.Executor = -1,
     pairs: Sequence[tuple[cirq.GridQubit, cirq.GridQubit]] | None = None,
     tags: Sequence[Any] = (),
 ) -> tuple[xeb_fitting.XEBCharacterizationResult, pd.DataFrame]:
@@ -81,7 +82,7 @@ def z_phase_calibration_workflow(
         cycle_depths: The cycle depths to use.
         random_state: The random state to use.
         atol: Absolute tolerance to be used by the minimizer.
-        num_workers_or_pool: An optional multi-processing pool or number of workers.
+        num_workers_or_pool: An optional pool or number of workers.
             A zero value means no multiprocessing.
             A positive integer value will create a pool with the given number of workers.
             A negative value will create pool with maximum number of workers.
@@ -92,12 +93,12 @@ def z_phase_calibration_workflow(
         - A `pd.DataFrame` comparing the before and after fidelities.
     """
 
-    pool: multiprocessing.pool.Pool | None = None
+    pool: multiprocessing.pool.Pool | cf.Executor | None = None
     local_pool = False
     if isinstance(num_workers_or_pool, multiprocessing.pool.Pool):
         pool = num_workers_or_pool  # pragma: no cover
     elif num_workers_or_pool != 0:
-        pool = multiprocessing.Pool(num_workers_or_pool if num_workers_or_pool > 0 else None)
+        pool = cf.ThreadPoolExecutor(num_workers_or_pool if num_workers_or_pool > 0 else None)
         local_pool = True
 
     fids_df_0, circuits, sampled_df = parallel_xeb_workflow(
@@ -143,8 +144,8 @@ def z_phase_calibration_workflow(
     )
 
     if local_pool:
-        assert isinstance(pool, multiprocessing.pool.Pool)
-        pool.close()
+        assert isinstance(pool, cf.Executor)
+        pool.shutdown()
     return result, before_after
 
 
@@ -159,7 +160,7 @@ def calibrate_z_phases(
     cycle_depths: Sequence[int] = tuple(np.arange(3, 100, 20)),
     random_state: cirq.RANDOM_STATE_OR_SEED_LIKE = None,
     atol: float = 1e-3,
-    num_workers_or_pool: int | multiprocessing.pool.Pool = -1,
+    num_workers_or_pool: int | multiprocessing.pool.Pool | cf.Executor = -1,
     pairs: Sequence[tuple[cirq.GridQubit, cirq.GridQubit]] | None = None,
     tags: Sequence[Any] = (),
 ) -> dict[tuple[cirq.Qid, cirq.Qid], cirq.PhasedFSimGate]: