From 716047af4a1e90a1b9c4542d60c20f3e6afadaa7 Mon Sep 17 00:00:00 2001 From: Aaron Sander <61705296+aaronleesander@users.noreply.github.com> Date: Thu, 8 Oct 2026 00:14:00 +0200 Subject: [PATCH 1/5] added explicit worker check to avoid spawning parallel processes in serial runs --- CHANGELOG.md | 3 + src/mqt/yaqs/simulator.py | 47 +++++++-------- tests/analog/test_ensemble.py | 4 +- tests/test_simulator.py | 108 ++++++++++++++++++++++++++++++++++ 4 files changed, 138 insertions(+), 24 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 01a5bdc9b..2d5052454 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -73,6 +73,8 @@ releases may include breaking changes. ### Fixed +- avoided process startup for single-worker simulations ([PR pending]) + ([**@aaronleesander**]) - fixed noise characterization iteration limits and initial loss ([#621]) ([**@aaronleesander**]) - fixed Schmidt spectrum simulation results ([#620]) @@ -292,6 +294,7 @@ for previous changelogs._ +[PR pending]: https://github.com/munich-quantum-toolkit/yaqs/pulls [#621]: https://github.com/munich-quantum-toolkit/yaqs/pull/621 [#620]: https://github.com/munich-quantum-toolkit/yaqs/pull/620 [#609]: https://github.com/munich-quantum-toolkit/yaqs/pull/609 diff --git a/src/mqt/yaqs/simulator.py b/src/mqt/yaqs/simulator.py index 80a8c55fe..6643918d3 100644 --- a/src/mqt/yaqs/simulator.py +++ b/src/mqt/yaqs/simulator.py @@ -1073,22 +1073,23 @@ class Simulator: A :class:`Simulator` owns the execution-side configuration: how trajectories are parallelized, how many workers to use, whether to display a progress bar, which multiprocessing context to use, and the retry policy for transient - worker errors. The physics inputs (initial state, operator, simulation + errors in process pool workers. The physics inputs (initial state, operator, simulation parameters, optional noise model) are passed per call to :meth:`run`. - Multiple :meth:`run` calls share the same configuration. Each call constructs - its own short-lived process pool when ``parallel=True``; the pool is not - persisted across runs in the current implementation. + Multiple :meth:`run` calls share the same configuration. A call constructs + a short-lived process pool when ``parallel=True`` and both the job count + and ``max_workers`` exceed one. Other calls run in the current process. + A worker cap of one with ``parallel=True`` also limits numerical threads to one. Attributes: - parallel: Whether to execute trajectories in parallel via a process pool. + parallel: Whether to use a process pool for multiple jobs and workers. max_workers: Maximum number of worker processes when ``parallel=True``. Defaults to ``max(1, available_cpus() - 1)``. show_progress: Whether to display a tqdm progress bar. mp_context: Multiprocessing context: ``"auto"`` (default), ``"fork"``, or ``"spawn"``. ``"auto"`` selects ``"fork"`` on Linux and ``"spawn"`` elsewhere. - max_retries: Maximum retry attempts for transient worker errors. - retry_exceptions: Exception types that trigger a retry. + max_retries: Maximum retry attempts for transient errors in process pool workers. + retry_exceptions: Exception types that trigger a retry in a process pool. """ def __init__( @@ -1104,13 +1105,13 @@ def __init__( """Initialize the simulator with execution-side configuration. Args: - parallel: Boolean that enables a process pool for multi-trajectory runs. + parallel: Enable a process pool when both the job count and worker cap exceed one. max_workers: Positive worker-process cap. ``None`` (default) resolves to - ``max(1, available_cpus() - 1)``. + ``max(1, available_cpus() - 1)``. A cap of one runs in the current process. show_progress: Boolean that controls the tqdm progress bar. mp_context: Multiprocessing start method (``"auto"``, ``"fork"``, or ``"spawn"``). - max_retries: Non-negative maximum retries for transient worker errors. - retry_exceptions: Exception types that trigger a retry. + max_retries: Non-negative maximum retries for transient errors in process pool workers. + retry_exceptions: Exception types that trigger a retry in a process pool. """ self._execution = ExecutionConfig( parallel=parallel, @@ -1123,7 +1124,7 @@ def __init__( @property def parallel(self) -> bool: - """Whether parallel execution is enabled.""" + """Whether process pools are enabled for multiple jobs and workers.""" return self._execution.parallel @parallel.setter @@ -1132,7 +1133,7 @@ def parallel(self, value: bool) -> None: @property def max_workers(self) -> int: - """Effective worker count for parallel execution.""" + """Effective worker cap; one keeps execution in the current process.""" return self._execution.resolved_max_workers() @max_workers.setter @@ -1159,7 +1160,7 @@ def mp_context(self, value: MPContext) -> None: @property def max_retries(self) -> int: - """Maximum retries per job in parallel execution.""" + """Maximum retries per job in a process pool.""" return self._execution.max_retries @max_retries.setter @@ -1168,7 +1169,7 @@ def max_retries(self, value: int) -> None: @property def retry_exceptions(self) -> tuple[type[BaseException], ...]: - """Exception types that trigger a parallel job retry.""" + """Exception types that trigger a retry in a process pool.""" return self._execution.retry_exceptions @retry_exceptions.setter @@ -1394,7 +1395,7 @@ def consume(traj_index: int, trajectory: _ProgramTrajectory) -> None: if trajectory_final is not None: final_mps = trajectory_final - if self.parallel and effective_num_traj > 1: + if self.parallel and effective_num_traj > 1 and self.max_workers > 1: for traj_index, trajectory in run_backend_parallel( worker_fn=_program_worker, payload=payload, @@ -1419,7 +1420,7 @@ def consume(traj_index: int, trajectory: _ProgramTrajectory) -> None: _program_worker, traj_index, payload, - n_threads=available_cpus(), + n_threads=1 if self.parallel and self.max_workers == 1 else available_cpus(), ) consume(traj_index, trajectory) @@ -1629,7 +1630,7 @@ def _run_analog( final_psi: np.ndarray | None = None final_rho: np.ndarray | None = None - if self.parallel and effective_num_traj > 1: + if self.parallel and effective_num_traj > 1 and self.max_workers > 1: for i, traj_payload in run_backend_parallel( worker_fn=worker_fn, payload=payload, @@ -1653,7 +1654,7 @@ def _run_analog( else: final_mps = cast("MPS", traj_final) else: - n_threads = available_cpus() + n_threads = 1 if self.parallel and self.max_workers == 1 else available_cpus() args: list[Any] if state_rep == "vector": @@ -1810,7 +1811,7 @@ def _consume( final_mps = traj_final try: - if self.parallel and effective_num_traj > 1: + if self.parallel and effective_num_traj > 1 and self.max_workers > 1: for i, traj_payload in run_backend_parallel( worker_fn=_digital_worker, payload=payload, @@ -1825,7 +1826,7 @@ def _consume( traj_data, traj_diag, shot_counts, traj_final = traj_payload _consume(i, traj_data, traj_diag, shot_counts, traj_final) else: - n_threads = available_cpus() + n_threads = 1 if self.parallel and self.max_workers == 1 else available_cpus() iterator = tqdm( range(effective_num_traj), desc="Running trajectories", @@ -1967,7 +1968,7 @@ def _run_ensemble( "operator": operator, } - if self.parallel and len(initial_states) > 1: + if self.parallel and len(initial_states) > 1 and self.max_workers > 1: for i, (obs_result, traj_diag, multi_time_result) in run_backend_parallel( worker_fn=_ensemble_worker, payload=payload, @@ -1985,7 +1986,7 @@ def _run_ensemble( assert multi_time_result is not None multi_time_matrix[i] = multi_time_result else: - n_threads = available_cpus() + n_threads = 1 if self.parallel and self.max_workers == 1 else available_cpus() args = [(i, initial_states[i], worker_params, operator) for i in range(len(initial_states))] iterator = tqdm(args, desc="Running unitary ensemble", ncols=80, disable=not self.show_progress) for i, arg in enumerate(iterator): diff --git a/tests/analog/test_ensemble.py b/tests/analog/test_ensemble.py index 50a95ef99..85a7c57f9 100644 --- a/tests/analog/test_ensemble.py +++ b/tests/analog/test_ensemble.py @@ -258,7 +258,9 @@ def test_list_mps_unitary_ensemble_parallel_worker_path() -> None: dt=0.05, multi_time_observables=[(z0, z0), (z0, z1)], ) - result = Simulator(parallel=True, show_progress=False).run(states, hamiltonian, sim_params, noise_model=None) + result = Simulator(parallel=True, max_workers=2, show_progress=False).run( + states, hamiltonian, sim_params, noise_model=None + ) assert result.expectation_values[0] is not None assert result.multi_time_results is not None diff --git a/tests/test_simulator.py b/tests/test_simulator.py index 7ae0838ba..d5c2a704a 100644 --- a/tests/test_simulator.py +++ b/tests/test_simulator.py @@ -36,6 +36,7 @@ NoiseModel, Observable, Result, + SimulationProgram, Simulator, State, simulator, @@ -177,6 +178,113 @@ def _build_params() -> AnalogSimParams: np.testing.assert_allclose(serial_vals, parallel_vals, atol=1e-10) +@pytest.mark.parametrize("workflow", ["analog-mps", "analog-vector", "digital", "program", "ensemble"]) +@pytest.mark.parametrize("worker_mode", ["explicit", "xdist"]) +def test_simulator_one_worker_runs_inline( + workflow: str, + worker_mode: str, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """One worker preserves seeded results and inputs without creating a pool.""" + if worker_mode == "xdist": + monkeypatch.delenv("YAQS_MAX_WORKERS", raising=False) + monkeypatch.setenv("PYTEST_XDIST_WORKER", "gw0") + parallel_runner = Simulator(parallel=True, show_progress=False) + serial_threads = 1 + else: + monkeypatch.setattr(simulator, "available_cpus", lambda: 4) + parallel_runner = Simulator(parallel=True, max_workers=1, show_progress=False) + serial_threads = 4 + assert parallel_runner.max_workers == 1 + + representation: Representation = "vector" if workflow == "analog-vector" else "mps" + states = [State(2, initial="zeros", representation=representation)] + if workflow == "ensemble": + states.append(State(2, initial="ones")) + initial_vectors = [state.vector.copy() if representation == "vector" else state.mps.to_vec() for state in states] + initial_centers = [state.mps.orthogonality_center for state in states] if representation == "mps" else [] + hamiltonian = Hamiltonian.ising(2, J=0.2, g=0.3) + circuit = QuantumCircuit(2) + circuit.ry(0.7, 0) + circuit.cx(0, 1) + noise = NoiseModel([{"name": "pauli_x", "sites": [0], "strength": 0.5}]) + num_jobs = 2 if workflow == "ensemble" else 4 + + def run(runner: Simulator) -> Result: + observables = [Observable("z", site) for site in range(2)] + if workflow == "digital": + params = DigitalSimParams(observables=observables, num_traj=4, random_seed=YAQS_TEST_SEED) + return runner.run(states[0], circuit, params, noise) + if workflow == "program": + program = SimulationProgram( + [(circuit, DigitalSimParams()), (hamiltonian, AnalogSimParams(elapsed_time=0.2, dt=0.1))], + observables=observables, + num_traj=4, + random_seed=YAQS_TEST_SEED, + ) + return runner.run(states[0], program, noise_model=noise) + analog_params = AnalogSimParams( + observables=observables, + elapsed_time=0.2, + dt=0.1, + num_traj=4, + random_seed=YAQS_TEST_SEED, + ) + if workflow == "ensemble": + return runner.run(states, hamiltonian, analog_params) + return runner.run(states[0], hamiltonian, analog_params, noise) + + with ( + patch.object(simulator, "run_backend_parallel", side_effect=AssertionError("Unexpected process pool")) as pool, + patch.object(simulator, "call_serial_capped", wraps=simulator.call_serial_capped) as inline, + ): + serial_result = run(Simulator(parallel=False, show_progress=False)) + assert inline.call_count == num_jobs + assert all(call.kwargs["n_threads"] == serial_threads for call in inline.call_args_list) + inline.reset_mock() + parallel_result = run(parallel_runner) + assert inline.call_count == num_jobs + assert all(call.kwargs["n_threads"] == 1 for call in inline.call_args_list) + pool.assert_not_called() + + serial_segments = serial_result.segment_results or [serial_result] + parallel_segments = parallel_result.segment_results or [parallel_result] + for serial, parallel in zip(serial_segments, parallel_segments, strict=True): + assert len(serial.expectation_values) == len(parallel.expectation_values) == 2 + for serial_mean, parallel_mean, serial_traj, parallel_traj in zip( + serial.expectation_values, + parallel.expectation_values, + serial.trajectories, + parallel.trajectories, + strict=True, + ): + assert serial_traj.shape[0] == parallel_traj.shape[0] == num_jobs + np.testing.assert_allclose(parallel_traj, serial_traj, atol=1e-12) + np.testing.assert_allclose(parallel_mean, serial_mean, atol=1e-12) + for state, initial_vector in zip(states, initial_vectors, strict=True): + vector = state.vector if representation == "vector" else state.mps.to_vec() + np.testing.assert_array_equal(vector, initial_vector) + if representation == "mps": + assert [state.mps.orthogonality_center for state in states] == initial_centers + captured = capsys.readouterr() + assert not captured.out + assert not captured.err + + +def test_simulator_one_worker_propagates_backend_failure() -> None: + """Inline execution propagates a backend failure without process-pool retries.""" + params = AnalogSimParams(observables=[Observable("z", 0)], elapsed_time=0.1, dt=0.1, num_traj=2) + noise = NoiseModel([{"name": "pauli_x", "sites": [0], "strength": 0.5}]) + runner = Simulator(parallel=True, max_workers=1, max_retries=3, show_progress=False) + with ( + patch.object(simulator, "analog_tjm_1", side_effect=OSError("Backend failed")) as backend, + pytest.raises(OSError, match="Backend failed"), + ): + runner.run(State(1, initial="zeros"), Hamiltonian.ising(1, J=0.0, g=0.0), params, noise) + backend.assert_called_once() + + def test_simulator_show_progress_disabled(capsys: pytest.CaptureFixture[str]) -> None: """``show_progress=False`` suppresses the tqdm bar.""" num_qubits = 2 From 801279d8d951aa48941d838e35f7de4420e9165d Mon Sep 17 00:00:00 2001 From: Aaron Sander <61705296+aaronleesander@users.noreply.github.com> Date: Thu, 8 Oct 2026 00:20:19 +0200 Subject: [PATCH 2/5] Document inline single-worker execution and pool retries --- UPGRADING.md | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/UPGRADING.md b/UPGRADING.md index 74d0e21ea..11c014ade 100644 --- a/UPGRADING.md +++ b/UPGRADING.md @@ -6,6 +6,16 @@ of changes including minor and patch releases, please refer to the ## [Unreleased] +### Changed: single-worker simulations run in the calling process + +`Simulator(parallel=True)` uses a process pool only when both the job count and +the resolved `max_workers` are greater than one. Runs with one worker execute in +the calling process and keep the one-thread numerical limit used by pool +workers. + +`max_retries` and `retry_exceptions` apply only to process workers. Exceptions +from single-worker runs propagate directly, as they do with `parallel=False`. + ### Breaking: physical sites use one spatial ordering YAQS now uses one public spatial dense-basis order. Site 0 is the From 7d1e5c3a269eda3eacfc736b11c9b6df7d134a30 Mon Sep 17 00:00:00 2001 From: Aaron Sander <61705296+aaronleesander@users.noreply.github.com> Date: Thu, 8 Oct 2026 00:23:22 +0200 Subject: [PATCH 3/5] Remove single-worker execution upgrade note --- UPGRADING.md | 10 ---------- 1 file changed, 10 deletions(-) diff --git a/UPGRADING.md b/UPGRADING.md index 11c014ade..74d0e21ea 100644 --- a/UPGRADING.md +++ b/UPGRADING.md @@ -6,16 +6,6 @@ of changes including minor and patch releases, please refer to the ## [Unreleased] -### Changed: single-worker simulations run in the calling process - -`Simulator(parallel=True)` uses a process pool only when both the job count and -the resolved `max_workers` are greater than one. Runs with one worker execute in -the calling process and keep the one-thread numerical limit used by pool -workers. - -`max_retries` and `retry_exceptions` apply only to process workers. Exceptions -from single-worker runs propagate directly, as they do with `parallel=False`. - ### Breaking: physical sites use one spatial ordering YAQS now uses one public spatial dense-basis order. Site 0 is the From 83015fcba597acc52e6dfa476f76115c92bd67c4 Mon Sep 17 00:00:00 2001 From: Aaron Sander <61705296+aaronleesander@users.noreply.github.com> Date: Thu, 8 Oct 2026 01:39:42 +0200 Subject: [PATCH 4/5] switched linux default to forkserver for stability --- docs/examples/simulator_initialization.md | 21 +++++---- src/mqt/yaqs/core/parallel_utils.py | 4 +- src/mqt/yaqs/simulator.py | 2 +- .../noise/optimization/test_trajectories.py | 28 +++++++++++ tests/core/data_structures/test_mps.py | 6 ++- tests/core/test_parallel_utils.py | 47 ++++++++++++++++++- tests/test_equivalence_checker.py | 5 +- 7 files changed, 97 insertions(+), 16 deletions(-) diff --git a/docs/examples/simulator_initialization.md b/docs/examples/simulator_initialization.md index 3d1789cc9..d8a5f89c6 100644 --- a/docs/examples/simulator_initialization.md +++ b/docs/examples/simulator_initialization.md @@ -172,14 +172,19 @@ serial runs. ## `mp_context`: multiprocessing start method -`mp_context` controls how worker processes are spawned. The default `"auto"` -picks the best option per OS: - -| Value | Behaviour | -| --------- | --------------------------------------------------------------------------------------------------------------- | -| `"auto"` | `"fork"` on Linux, `"spawn"` everywhere else. | -| `"fork"` | Fastest worker startup; reuses Python state from the parent. Safe in YAQS because BLAS/OpenMP pools are capped. | -| `"spawn"` | Fresh interpreter per worker. Required on Windows/macOS; slower startup but more isolated. | +`mp_context` controls how worker processes start. The default `"auto"` selects a +start method per OS: + +| Value | Behaviour | +| --------- | --------------------------------------------------------------------------- | +| `"auto"` | `"forkserver"` on Linux, `"spawn"` everywhere else. | +| `"fork"` | Copies the parent process. Avoid when the parent has active threads. | +| `"spawn"` | Fresh interpreter per worker. Used on Windows/macOS and available on Linux. | + +On Linux, `"auto"` creates workers through a separate server process. The first +pool has extra startup cost, but workers do not fork the application's threaded +process. Numerical thread limits control resource use; they do not make an +explicit `"fork"` safe when the parent has active threads. ```{code-cell} ipython3 automatic_context = Simulator(mp_context="auto") diff --git a/src/mqt/yaqs/core/parallel_utils.py b/src/mqt/yaqs/core/parallel_utils.py index 2b8d8cc05..e386baa28 100644 --- a/src/mqt/yaqs/core/parallel_utils.py +++ b/src/mqt/yaqs/core/parallel_utils.py @@ -99,7 +99,7 @@ def get_parallel_context(mp_context: MPContext = "auto") -> multiprocessing.cont """Return a multiprocessing context for worker processes. Args: - mp_context: Start method selector. ``"auto"`` uses ``"fork"`` on Linux and + mp_context: Start method selector. ``"auto"`` uses ``"forkserver"`` on Linux and ``"spawn"`` elsewhere; ``"fork"`` or ``"spawn"`` select that method explicitly. Returns: @@ -107,7 +107,7 @@ def get_parallel_context(mp_context: MPContext = "auto") -> multiprocessing.cont """ if mp_context == "auto": if sys.platform == "linux": - return multiprocessing.get_context("fork") + return multiprocessing.get_context("forkserver") return multiprocessing.get_context("spawn") return multiprocessing.get_context(mp_context) diff --git a/src/mqt/yaqs/simulator.py b/src/mqt/yaqs/simulator.py index 328a74840..3cce3c076 100644 --- a/src/mqt/yaqs/simulator.py +++ b/src/mqt/yaqs/simulator.py @@ -1095,7 +1095,7 @@ class Simulator: Defaults to ``max(1, available_cpus() - 1)``. show_progress: Whether to display trajectory and shot-readout progress bars. mp_context: Multiprocessing context: ``"auto"`` (default), ``"fork"``, - or ``"spawn"``. ``"auto"`` selects ``"fork"`` on Linux and ``"spawn"`` elsewhere. + or ``"spawn"``. ``"auto"`` selects ``"forkserver"`` on Linux and ``"spawn"`` elsewhere. max_retries: Maximum retry attempts for transient errors in process pool workers. retry_exceptions: Exception types that trigger a retry in a process pool. """ diff --git a/tests/characterization/noise/optimization/test_trajectories.py b/tests/characterization/noise/optimization/test_trajectories.py index f1a85db80..4342721ea 100644 --- a/tests/characterization/noise/optimization/test_trajectories.py +++ b/tests/characterization/noise/optimization/test_trajectories.py @@ -66,6 +66,34 @@ def test_simulate_observable_trajectories_shape() -> None: assert expectations.shape == (len(observables), len(times)) +def test_seeded_observable_trajectories_match_in_serial_and_parallel() -> None: + """Noise-fitting forward trajectories agree with automatic process-pool execution.""" + hamiltonian, init_state, observables, _sim_params, noise = _three_site_problem() + params = AnalogSimParams( + observables=observables, + elapsed_time=0.2, + dt=0.1, + num_traj=8, + random_seed=42, + order=1, + ) + results = [ + simulate_observable_trajectories( + sim_params=params, + hamiltonian=hamiltonian, + init_state=init_state, + noise_model=noise, + observables=observables, + simulator=build_simulator(ExecutionConfig(parallel=parallel, max_workers=2, show_progress=False)), + representation="mps", + ) + for parallel in (False, True) + ] + serial, parallel = results + np.testing.assert_array_equal(parallel[1], serial[1]) + np.testing.assert_allclose(parallel[0], serial[0], atol=1e-12) + + def test_ref_expectations_path_matches_simulation() -> None: """Precomputed expectations are accepted when shapes match the fitting set.""" hamiltonian, init_state, observables, sim_params, reference_model = _three_site_problem() diff --git a/tests/core/data_structures/test_mps.py b/tests/core/data_structures/test_mps.py index 138c02517..d09ac6a87 100644 --- a/tests/core/data_structures/test_mps.py +++ b/tests/core/data_structures/test_mps.py @@ -1892,10 +1892,12 @@ def test_inplace_measure() -> None: assert np.isclose(psi.expect(Observable("x", 2)), 1.0 if outcome == 0 else -1.0) -def test_multi_shot() -> None: +@pytest.mark.parametrize("parallel", [False, True], ids=["serial", "parallel"]) +def test_multi_shot(*, parallel: bool, monkeypatch: pytest.MonkeyPatch) -> None: """Full-chain readout of a basis state counts every shot at the expected integer.""" + monkeypatch.setattr(mps_mod, "available_cpus", lambda: 3 if parallel else 1) psi_mps = MPS(length=3, state="ones") - assert psi_mps.measure_shots(shots=10) == {7: 10} + assert psi_mps.measure_shots(shots=10, show_progress=False) == {7: 10} def test_norm() -> None: diff --git a/tests/core/test_parallel_utils.py b/tests/core/test_parallel_utils.py index af3670b66..ed136953f 100644 --- a/tests/core/test_parallel_utils.py +++ b/tests/core/test_parallel_utils.py @@ -13,6 +13,7 @@ import multiprocessing import os import sys +import threading from types import SimpleNamespace from typing import Any, cast @@ -35,6 +36,50 @@ from mqt.yaqs.simulator import available_cpus as simulator_available_cpus +def _indexed_pool_worker(job_idx: int, payload: dict[str, Any] | None = None) -> tuple[int, int]: + """Return the indexed value and worker PID after both workers reach the barrier.""" + context = resolve_worker_ctx(payload) + if "barrier" in context: + context["barrier"].wait(timeout=30) + return job_idx + context["offset"], os.getpid() + + +@pytest.mark.filterwarnings("error:This process.*is multi-threaded.*:DeprecationWarning") +def test_auto_pool_from_threaded_parent_preserves_results() -> None: + """Automatic pools start safely beside a live thread and preserve indexed results.""" + serial = run_indexed_jobs( + _indexed_pool_worker, + payload={"offset": 10}, + n_jobs=2, + config=ExecutionConfig(parallel=False, show_progress=False), + desc="test", + ) + stop = threading.Event() + thread = threading.Thread(target=stop.wait) + thread.start() + try: + assert thread.is_alive() + context = get_parallel_context() + pooled = run_indexed_jobs( + _indexed_pool_worker, + payload={"offset": 10, "barrier": context.Barrier(2)}, + n_jobs=2, + config=ExecutionConfig(max_workers=2, show_progress=False, max_retries=0), + desc="test", + ) + finally: + stop.set() + thread.join(timeout=5) + + assert not thread.is_alive() + assert {index: value for index, (value, _pid) in pooled.items()} == { + index: value for index, (value, _pid) in serial.items() + } + worker_pids = {pid for _value, pid in pooled.values()} + assert len(worker_pids) == 2 + assert os.getpid() not in worker_pids + + def test_available_cpus_without_slurm(monkeypatch: pytest.MonkeyPatch) -> None: """Without overrides, ``available_cpus`` falls back to affinity or ``cpu_count``.""" monkeypatch.delenv("YAQS_MAX_WORKERS", raising=False) @@ -96,7 +141,7 @@ def test_threading_config() -> None: """Verify correct multiprocessing context and Numba threading configuration.""" ctx = get_parallel_context() if sys.platform == "linux": - assert ctx.get_start_method() == "fork" + assert ctx.get_start_method() == "forkserver" else: assert ctx.get_start_method() == "spawn" diff --git a/tests/test_equivalence_checker.py b/tests/test_equivalence_checker.py index 2c2d285f4..b7926f2e6 100644 --- a/tests/test_equivalence_checker.py +++ b/tests/test_equivalence_checker.py @@ -1408,7 +1408,8 @@ def test_ensemble_trajectory_worker_uses_initialized_context() -> None: assert result["fidelity"] == pytest.approx(1.0, abs=1e-12) -def test_seeded_serial_and_process_pool_ensembles_agree() -> None: +@pytest.mark.parametrize("mp_context", ["auto", "spawn"]) +def test_seeded_serial_and_process_pool_ensembles_agree(mp_context: Literal["auto", "spawn"]) -> None: """Serial and process-pool workers return the same seeded MPO ensemble.""" qc = QuantumCircuit(2) qc.h(0) @@ -1417,7 +1418,7 @@ def test_seeded_serial_and_process_pool_ensembles_agree() -> None: kwargs = {"noise_model": noise, "num_traj": 6, "random_seed": 0, "return_trajectories": True} serial = EquivalenceChecker(representation="mpo", parallel=False).check(qc, qc, **kwargs) - pooled = EquivalenceChecker(representation="mpo", parallel=True, max_workers=2, mp_context="spawn").check( + pooled = EquivalenceChecker(representation="mpo", parallel=True, max_workers=2, mp_context=mp_context).check( qc, qc, **kwargs ) serial_fidelities = [traj["fidelity"] for traj in serial["trajectories"]] From 1875d6b8a9dca8636f01f69896c3376f8d58866a Mon Sep 17 00:00:00 2001 From: Aaron Sander <61705296+aaronleesander@users.noreply.github.com> Date: Thu, 8 Oct 2026 01:42:20 +0200 Subject: [PATCH 5/5] updated changelog --- CHANGELOG.md | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 33548d61e..0ac67bf7f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -73,7 +73,9 @@ releases may include breaking changes. ### Fixed -- avoided process startup for single-worker simulations ([#623]) +- avoided process startup for single-worker simulations ([#624]) + ([**@aaronleesander**]) +- cleaned up tests and fixed MCWF RNG continuity ([#623]) ([**@aaronleesander**]) - fixed shot readout progress bar suppression ([#622]) ([**@aaronleesander**]) @@ -296,6 +298,7 @@ for previous changelogs._ +[#624]: https://github.com/munich-quantum-toolkit/yaqs/pull/624 [#623]: https://github.com/munich-quantum-toolkit/yaqs/pull/623 [#622]: https://github.com/munich-quantum-toolkit/yaqs/pull/622 [#621]: https://github.com/munich-quantum-toolkit/yaqs/pull/621