Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
NoureldinYosri merged 9 commits into from
Mar 31, 2026
Merged

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

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
e98da30
use concurrent.futures.Executor instead of multiprocessing pool to re…
NoureldinYosri Mar 7, 2026
9ebdd7e
Merge branch 'main' into z_phase_threads
NoureldinYosri Mar 7, 2026
23aaef5
lint
NoureldinYosri Mar 7, 2026
2d92d28
ruff
NoureldinYosri Mar 7, 2026
3fc08ba
Merge branch 'main' into z_phase_threads
NoureldinYosri Mar 11, 2026
762dabc
coverage
NoureldinYosri Mar 11, 2026
c13e5b9
address comments
NoureldinYosri Mar 26, 2026
10f8183
Merge branch 'main' into z_phase_threads
NoureldinYosri Mar 26, 2026
23a175e
Merge branch 'main' into z_phase_threads
NoureldinYosri Mar 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions cirq-core/cirq/experiments/two_qubit_xeb.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

from __future__ import annotations

import concurrent.futures as cf
Comment thread
NoureldinYosri marked this conversation as resolved.
Outdated
import functools
import itertools
from collections.abc import Mapping, Sequence
Expand Down Expand Up @@ -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,
Expand All @@ -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.
Expand Down
11 changes: 6 additions & 5 deletions cirq-core/cirq/experiments/xeb_fitting.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.

Expand Down Expand Up @@ -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.
Expand All @@ -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(
Expand Down Expand Up @@ -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.
Expand All @@ -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()
Expand Down
3 changes: 2 additions & 1 deletion cirq-core/cirq/experiments/xeb_simulation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

from __future__ import annotations

import concurrent.futures as cf
from collections.abc import Sequence
from dataclasses import dataclass
from typing import Any, TYPE_CHECKING
Expand Down Expand Up @@ -85,7 +86,7 @@ def simulate_2q_xeb_circuits(
circuits: Sequence[cirq.Circuit],
cycle_depths: Sequence[int],
param_resolver: cirq.ParamResolverOrSimilarType = None,
pool: multiprocessing.pool.Pool | None = None,
pool: multiprocessing.pool.Pool | cf.Executor | None = None,
simulator: cirq.SimulatesIntermediateState | None = None,
) -> pd.DataFrame:
"""Simulate two-qubit XEB circuits.
Expand Down
18 changes: 9 additions & 9 deletions cirq-core/cirq/experiments/z_phase_calibration.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@

from __future__ import annotations

import multiprocessing
import concurrent.futures as cf
import multiprocessing.pool
from collections.abc import Sequence
from typing import Any, TYPE_CHECKING
Expand Down Expand Up @@ -46,7 +46,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,

@pavoljuhas pavoljuhas Mar 11, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

AFAICT, the code later needs only the Pool.map or Executor.map functions.

Would it be possible to change this to accept either an int for a number of workers or a parallel-map function?

@NoureldinYosri NoureldinYosri Mar 26, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this is too much indirection, also everywhere else in experiments/ expects either a pool or number of workers

pairs: Sequence[tuple[cirq.GridQubit, cirq.GridQubit]] | None = None,
tags: Sequence[Any] = (),
) -> tuple[xeb_fitting.XEBCharacterizationResult, pd.DataFrame]:
Expand Down Expand Up @@ -81,7 +81,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.
Expand All @@ -92,12 +92,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):
if isinstance(num_workers_or_pool, (multiprocessing.pool.Pool, cf.Executor)):
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)

@pavoljuhas pavoljuhas Mar 11, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ThreadPoolExecutor is subject to GIL. Unless the mapped function spends a lot of time in numpy calls or waiting for IO, the execution would be the same as in a serial call or worse due to thread-switching overhead.

I made a quick test with a many-term sums computed in series or in parallel with multiprocessing.Pool.map vs ThreadPoolExecutor.map. The ThreadPoolExecutor took about 2.5 times longer than a serial evaluation.

example timing code 
def partial_sum(start_end: tuple[int, int]) -> float:
    total = 0
    for i in range(*start_end, 3):
        total += (-1) ** i * 1.0 / i
    return total


def tedious_sum(terms_count: int, mapfunc) -> float:
    total = sum(mapfunc(partial_sum, ((start, terms_count) for start in (1, 2, 3))))
    return total

# %timeit tedious_sum(10_000_000, map)
# 2.93 s ± 51.8 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

# pool = multiprocessing.Pool(3)
# %timeit tedious_sum(10_000_000, pool.map)
# 1.01 s ± 25.7 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

# tpx = concurrent.futures.ThreadPoolExecutor(3)
# %timeit tedious_sum(10_000_000, tpx.map)
# 8.46 s ± 1.11 s per loop (mean ± std. dev. of 7 runs, 1 loop each)

Can you make a quick comparison of the z_phase_calibration_workflow duration with ThreadPoolExecutor compared to a serial run?

If comparable I'd suggest to make it a default to do a serial evaluation.

@NoureldinYosri NoureldinYosri Mar 26, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

see https://colab.sandbox.google.com/drive/17eskpU9OH_L-mLIIcrFNkafhsZwgT4BN#scrollTo=GH8YAuLgFZ__

serial: 4m35s
with ThreadPoolExecuter(): 3m54s

@pavoljuhas pavoljuhas Mar 27, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is a bit hard to isolate the duration of z_phase_calibration_workflow calls in that colab.
If I patch the test here which runs that function with 2 workers to run it in serial, ie,

diff --git a/cirq-core/cirq/experiments/z_phase_calibration_test.py b/cirq-core/cirq/experiments/z_phase_calibration>
index 48db85ad2..309651e50 100644
--- a/cirq-core/cirq/experiments/z_phase_calibration_test.py
+++ b/cirq-core/cirq/experiments/z_phase_calibration_test.py
@@ -223,3 +223,3 @@ def test_calibrate_z_phases_workflow_no_options_no_pool(angles, error) -> None:
         random_state=_SEED,
-        num_workers_or_pool=2,
+        num_workers_or_pool=0,
     )

the test duration goes down from ~1.6 to ~1.4 s - measured with

pytest -p no:randomly -Wignore --durations=1 \
"cirq-core/cirq/experiments/z_phase_calibration_test.py::test_calibrate_z_phases_workflow_no_options_no_pool[angles1-error1]"

@NoureldinYosri NoureldinYosri Mar 27, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the comparison I did was between serial and num_workers=None (i.e. os.cpu_count()) which on my device=24

local_pool = True

fids_df_0, circuits, sampled_df = parallel_xeb_workflow(
Expand Down Expand Up @@ -143,8 +143,8 @@ def z_phase_calibration_workflow(
)

if local_pool:
assert isinstance(pool, multiprocessing.pool.Pool)
pool.close()
assert isinstance(pool, cf.Executor)
pool.shutdown()
Comment thread
NoureldinYosri marked this conversation as resolved.
Outdated
return result, before_after


Expand All @@ -159,7 +159,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]:
Expand Down
Loading