Configuring the Simulator

YAQS draws a sharp line between what you simulate and how it runs:

Layer	Role
State, Hamiltonian, AnalogSimParams / StrongSimParams / WeakSimParams	The physics: initial state, operator, time grid, observables, trajectory count, truncation, noise.
Simulator	The execution: parallel vs. serial trajectories, worker count, progress reporting, multiprocessing start method, and retry policy for transient worker errors.

This page walks through every option on the Simulator class so you can tune execution without touching the physics.

from mqt.yaqs import Result, Simulator
from mqt.yaqs.core.data_structures.hamiltonian import Hamiltonian
from mqt.yaqs.core.data_structures.simulation_parameters import AnalogSimParams, Observable
from mqt.yaqs.core.data_structures.state import State
from mqt.yaqs.core.libraries.gate_library import Z

A small reusable analog problem we will simulate throughout:

L = 4
H = Hamiltonian.ising(L, J=1.0, g=0.5)


def make_params(num_traj: int = 8) -> AnalogSimParams:
    """A short Ising evolution measuring `<Z>` on every site."""
    return AnalogSimParams(
        observables=[Observable(Z(), site) for site in range(L)],
        elapsed_time=0.2,
        dt=0.05,
        num_traj=num_traj,
        max_bond_dim=8,
        svd_threshold=1e-9,
        sample_timesteps=False,
        random_seed=0,
    )


state = State(L, initial="zeros")

Quick start: defaults

Calling Simulator() with no arguments gives you parallel execution across most of your CPU cores, a tqdm progress bar, an "auto" multiprocessing context, and a generous retry policy.

sim = Simulator()
print("parallel:        ", sim.parallel)
print("max_workers:     ", sim.max_workers)
print("show_progress:   ", sim.show_progress)
print("mp_context:      ", sim.mp_context)
print("max_retries:     ", sim.max_retries)
print("retry_exceptions:", sim.retry_exceptions)

parallel:         True
max_workers:      1
show_progress:    True
mp_context:       auto
max_retries:      10
retry_exceptions: (<class 'concurrent.futures._base.CancelledError'>, <class 'TimeoutError'>, <class 'OSError'>)

Every option is keyword-only, so you can override one without specifying the others:

quiet_sim = Simulator(show_progress=False)

Reusing one Simulator across runs

A Simulator instance is stateless with respect to the physics; the same instance can drive arbitrarily many run() calls. This is the recommended pattern in scripts and notebooks because it keeps execution configuration in one place.

sim = Simulator(show_progress=False)

for noise_strength in (0.0, 0.05, 0.1):
    params = make_params()
    # ... build a NoiseModel from `noise_strength` here in real code ...
    result = sim.run(state, H, params, noise_model=None)
    print(f"gamma={noise_strength}: <Z_0>={float(result.expectation_values[0][0]):+.4f}")

gamma=0.0: <Z_0>=+0.9803
gamma=0.05: <Z_0>=+0.9803
gamma=0.1: <Z_0>=+0.9803

/tmp/ipykernel_2687/2485894359.py:7: ComplexWarning: Casting complex values to real discards the imaginary part
  print(f"gamma={noise_strength}: <Z_0>={float(result.expectation_values[0][0]):+.4f}")

Each call constructs a short-lived ProcessPoolExecutor when parallel=True; pools are not persisted across run() calls, so you can safely change sim.max_workers (or replace sim entirely) between calls.

parallel: process-pool vs. in-process execution

parallel=True (the default) runs trajectories in worker processes via concurrent.futures.ProcessPoolExecutor. parallel=False runs every trajectory in the calling process, which is useful for:

• Debugging (full tracebacks, no pickling, breakpoints work).

• Very small jobs where the pool startup cost dominates.

• Notebook cells where you want to share state with the caller.

Both modes produce identical results for a fixed random_seed:

import numpy as np

params_serial = make_params()
sim_serial = Simulator(parallel=False, show_progress=False)
result_serial = sim_serial.run(state, H, params_serial)

params_parallel = make_params()
sim_parallel = Simulator(parallel=True, max_workers=2, show_progress=False)
result_parallel = sim_parallel.run(state, H, params_parallel)

for vals_s, vals_p in zip(result_serial.expectation_values, result_parallel.expectation_values, strict=True):
    np.testing.assert_allclose(vals_s, vals_p, atol=1e-10)
print("Serial and parallel results match.")

Serial and parallel results match.

Note

For runs with num_traj == 1 (e.g. noise-free analog/circuit dynamics, Lindblad), the simulator automatically takes the in-process path even with parallel=True. The pool is only spun up when there is more than one trajectory to dispatch.

max_workers and how the default is chosen

When max_workers is left as None, the simulator picks max(1, available_cpus() - 1) to leave one core free for the parent process and the OS:

from mqt.yaqs.simulator import available_cpus

print("available CPUs YAQS would use:", available_cpus())
print("default max_workers:          ", Simulator().max_workers)

available CPUs YAQS would use: 2
default max_workers:           1

available_cpus() (re-exported as available_cpus()) is deliberately cgroup- and scheduler-aware. In priority order it honours:

1. YAQS_MAX_WORKERS (explicit user override; positive integer).

2. PYTEST_XDIST_WORKER (returns 1 to avoid nested parallelism in tests).

3. SLURM hints — SLURM_CPUS_PER_TASK then SLURM_CPUS_ON_NODE.

4. Linux os.sched_getaffinity(0) (respects taskset, containers, cgroups).

5. os.cpu_count() as a final fallback.

Override the resolution either by setting the environment variable…

# In a shell: export YAQS_MAX_WORKERS=4

…or by passing max_workers explicitly:

sim_four = Simulator(max_workers=4, show_progress=False)
print("fixed max_workers:", sim_four.max_workers)

fixed max_workers: 4

show_progress: tqdm bars

show_progress=True (default) shows a tqdm bar labelled “Running trajectories” (or “Running unitary ensemble” for the deterministic ensemble path). Set show_progress=False to silence it — useful in test suites, batch scripts, and CI logs:

silent = Simulator(show_progress=False)
silent.run(state, H, make_params(num_traj=4))

Result(sim_params=<mqt.yaqs.core.data_structures.simulation_parameters.AnalogSimParams object at 0x7cfee00319a0>, observables=[<mqt.yaqs.core.data_structures.simulation_parameters.Observable object at 0x7cfedcc4e420>, <mqt.yaqs.core.data_structures.simulation_parameters.Observable object at 0x7cfedcc4e660>, <mqt.yaqs.core.data_structures.simulation_parameters.Observable object at 0x7cfedcc4e720>, <mqt.yaqs.core.data_structures.simulation_parameters.Observable object at 0x7cfedcc4e7e0>], expectation_values=[array([0.98033079+0.j]), array([0.98110104+0.j]), array([0.98110104+0.j]), array([0.98033079+0.j])], trajectories=[array([[0.98033079+0.j]]), array([[0.98110104+0.j]]), array([[0.98110104+0.j]]), array([[0.98033079+0.j]])], times=array([0.2]), runtime_cost=array([24.]), max_bond=array([2.]), total_bond=array([6.]), noise_model=None, output_state=None, multi_time_times=None, multi_time_results=None, measurements=[], counts=None)

The bar is suppressed regardless of parallel, so the same flag also silences 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.

from mqt.yaqs.parallel_utils import get_parallel_context

for choice in ("auto", "fork", "spawn"):
    try:
        ctx = get_parallel_context(choice)
    except ValueError as exc:
        print(f"{choice}: not available on this platform ({exc})")
    else:
        print(f"{choice}: start_method={ctx.get_start_method()}")

auto: start_method=fork
fork: start_method=fork
spawn: start_method=spawn

If you mix YAQS with GPU libraries or anything that does not survive fork(), force mp_context="spawn".

max_retries and retry_exceptions

Long parallel runs occasionally encounter transient worker failures: a worker is cancelled by the OS, a TimeoutError is raised, or a transient OSError (e.g. a temporary file system hiccup) propagates out of a backend. By default, the simulator retries each failing trajectory up to 10 times for the following exception types:

• concurrent.futures.CancelledError

• TimeoutError

• OSError

print("default max_retries:    ", Simulator().max_retries)
print("default retry_exceptions:", Simulator().retry_exceptions)

default max_retries:     10
default retry_exceptions: (<class 'concurrent.futures._base.CancelledError'>, <class 'TimeoutError'>, <class 'OSError'>)

Tighten the policy for fail-fast development (e.g. when bisecting an error):

strict = Simulator(max_retries=0, show_progress=False)

Or broaden it for unreliable environments:

import concurrent.futures

resilient = Simulator(
    max_retries=20,
    retry_exceptions=(concurrent.futures.CancelledError, TimeoutError, OSError, ConnectionError),
    show_progress=False,
)

Permanent errors (e.g. ValueError from your physics setup, AssertionError from invariants) are not retried — they propagate after the first failure regardless of max_retries.

Inspecting the return value: Result

run() returns a Result that holds every simulation output through a small, stable surface. The AnalogSimParams you passed in is referenced unchanged at result.sim_params:

sim = Simulator(show_progress=False)
params = make_params()
result = sim.run(state, H, params)

print("type:                 ", type(result).__name__)
print("len(observables):     ", len(result.observables))
print("len(expectation_values):", len(result.expectation_values))
print("len(trajectories):    ", len(result.trajectories))
print("times:                ", result.times)
print("noise_model:          ", result.noise_model)
print("output_state:         ", result.output_state)
print("counts (weak only):   ", result.counts)
print("multi_time_times:     ", result.multi_time_times)
print("multi_time_results:   ", result.multi_time_results)
print("runtime_cost:         ", result.runtime_cost)
print("max_bond:             ", result.max_bond)
print("total_bond:           ", result.total_bond)

type:                  Result
len(observables):      4
len(expectation_values): 4
len(trajectories):     4
times:                 [0.2]
noise_model:           None
output_state:          None
counts (weak only):    None
multi_time_times:      None
multi_time_results:    None
runtime_cost:          [24.]
max_bond:              [2.]
total_bond:            [6.]

The properties that don’t apply to your simulation kind return None (or an empty list for observables in weak simulations), so you can branch on them safely. The full set is:

Property	Populated for
observables	Analog and strong digital runs. Empty list for weak digital.
expectation_values	Aggregated expectation per observable (parallel to observables).
trajectories	Per-trajectory data per observable (parallel to observables).
times	Shared analog time grid; None for digital circuits.
runtime_cost	MPS-backed analog and strong digital runs (contraction-cost heuristic over time).
max_bond	MPS-backed analog and strong digital runs (maximum bond dimension over time).
total_bond	MPS-backed analog and strong digital runs (sum of internal bond dimensions).
noise_model	Any run that was given a NoiseModel; otherwise None.
output_state	Runs with get_state=True on AnalogSimParams or StrongSimParams (no noise).
multi_time_times, multi_time_results	Analog deterministic ensembles with multi_time_observables set.
counts	Weak digital simulations (the dict[int, int] of aggregated measurement outcomes).

Result (and its wrapped sim_params) is pickleable, so you can checkpoint and resume analysis from disk:

import pickle

blob = pickle.dumps(result)
restored: Result = pickle.loads(blob)  # noqa: S301
print("restored observable count:", len(restored.observables))

restored observable count: 4

Choosing settings for common scenarios

Scenario	Recommended Simulator(...)
Quick local run, want to see a progress bar	Simulator()
Notebook / docs build / CI logs	Simulator(show_progress=False)
Debugging a physics setup	Simulator(parallel=False, show_progress=False)
Single-process benchmark, all cores in the worker	Simulator(parallel=False) and let BLAS/OpenMP use all threads
Fixed core budget (e.g. SLURM job step)	YAQS_MAX_WORKERS=N in the environment, or Simulator(max_workers=N)
Mixing with GPU / non-fork-safe code	Simulator(mp_context="spawn")
Long unattended run on a flaky cluster	Simulator(max_retries=20) with broadened retry_exceptions

For physics-side settings (num_traj, max_bond_dim, svd_threshold, random_seed, sample_timesteps, observables, noise), see Noisy Analog Simulation, Representation Comparison, and Initializing quantum states.