Merge pull request #1046 from qiboteam/execute-circuits

Execute multiple circuits in parallel

src/qibo/backends/abstract.py

@@ -194,6 +194,13 @@ def execute_circuit(
         """Execute a :class:`qibo.models.circuit.Circuit`."""
         raise_error(NotImplementedError)
 
+    @abc.abstractmethod
+    def execute_circuits(
+        self, circuits, initial_states=None, nshots=None
+    ):  # pragma: no cover
+        """Execute multiple :class:`qibo.models.circuit.Circuit`s in parallel."""
+        raise_error(NotImplementedError)
+
     @abc.abstractmethod
     def execute_circuit_repeated(
         self, circuit, initial_state=None, nshots=None

src/qibo/backends/numpy.py

@@ -434,6 +434,15 @@ def execute_circuit(
                 "different one using ``qibo.set_device``.",
             )
 
+    def execute_circuits(
+        self, circuits, initial_states=None, nshots=1000, processes=None
+    ):
+        from qibo.parallel import parallel_circuits_execution
+
+        return parallel_circuits_execution(
+            circuits, initial_states, nshots, processes, backend=self
+        )
+
     def execute_circuit_repeated(self, circuit, initial_state=None, nshots=None):
         if nshots is None:
             nshots = 1

src/qibo/parallel.py

@@ -1,6 +1,12 @@
 """
 Resources for parallel circuit evaluation.
 """
+from typing import Iterable
+
+from joblib import Parallel, delayed
+
+from qibo.backends import GlobalBackend, set_threads
+from qibo.config import raise_error
 
 
 def parallel_execution(circuit, states, processes=None, backend=None):
@@ -10,7 +16,6 @@ def parallel_execution(circuit, states, processes=None, backend=None):
         .. code-block:: python
 
             import qibo
-            original_backend = qibo.get_backend()
             qibo.set_backend('qibojit')
             from qibo import models, set_threads
             from qibo.parallel import parallel_execution
@@ -24,7 +29,6 @@ def parallel_execution(circuit, states, processes=None, backend=None):
             set_threads(1)
             # execute in parallel
             results = parallel_execution(circuit, states, processes=2)
-            qibo.set_backend(original_backend)
 
     Args:
         circuit (qibo.models.Circuit): the input circuit.
@@ -35,27 +39,86 @@ def parallel_execution(circuit, states, processes=None, backend=None):
         Circuit evaluation for input states.
     """
     if backend is None:  # pragma: no cover
-        from qibo.backends import GlobalBackend
-
         backend = GlobalBackend()
 
     if states is None or not isinstance(states, list):  # pragma: no cover
-        from qibo.config import raise_error
-
-        raise_error(RuntimeError, "states must be a list.")
+        raise_error(TypeError, "states must be a list.")
 
     def operation(state, circuit):
+        backend.set_threads(backend.nthreads)
         return backend.execute_circuit(circuit, state)
 
-    from joblib import Parallel, delayed
-
     results = Parallel(n_jobs=processes, prefer="threads")(
         delayed(operation)(state, circuit) for state in states
     )
 
     return results
 
 
+def parallel_circuits_execution(
+    circuits, states=None, nshots=1000, processes=None, backend=None
+):
+    """Execute multiple circuits
+
+    Example:
+        .. code-block:: python
+
+            import qibo
+            qibo.set_backend('qibojit')
+            from qibo import models, set_threads
+            from qibo.parallel import parallel_circuits_execution
+            import numpy as np
+            # create different circuits
+            circuits = [models.QFT(n) for n in range(5, 16)]
+            # set threads to 1 per process (optional, requires tuning)
+            set_threads(1)
+            # execute in parallel
+            results = parallel_circuits_execution(circuits, processes=2)
+
+    Args:
+        circuits (list): list of circuits to execute.
+        states (optional, list): list of states to use as initial for each circuit.
+            Must have the same length as ``circuits``.
+            If one state is given it is used on all circuits.
+            If not given the default initial state on all circuits.
+        nshots (int): Number of shots when performing measurements, same for all circuits.
+        processes (int): number of processes for parallel evaluation.
+
+    Returns:
+        Circuit evaluation for input states.
+    """
+    if backend is None:  # pragma: no cover
+        backend = GlobalBackend()
+
+    if not isinstance(circuits, Iterable):  # pragma: no cover
+        raise_error(TypeError, "circuits must be iterable.")
+
+    if (
+        isinstance(states, (list, tuple))
+        and isinstance(circuits, (list, tuple))
+        and len(states) != len(circuits)
+    ):
+        raise_error(ValueError, "states must have the same length as circuits.")
+    elif states is not None and not isinstance(states, Iterable):
+        raise_error(TypeError, "states must be iterable.")
+
+    def operation(circuit, state):
+        backend.set_threads(backend.nthreads)
+        return backend.execute_circuit(circuit, state, nshots)
+
+    if states is None or isinstance(states, backend.tensor_types):
+        results = Parallel(n_jobs=processes, prefer="threads")(
+            delayed(operation)(circuit, states) for circuit in circuits
+        )
+    else:
+        results = Parallel(n_jobs=processes, prefer="threads")(
+            delayed(operation)(circuit, state)
+            for circuit, state in zip(circuits, states)
+        )
+
+    return results
+
+
 def parallel_parametrized_execution(
     circuit, parameters, initial_state=None, processes=None, backend=None
 ):
@@ -65,7 +128,6 @@ def parallel_parametrized_execution(
         .. code-block:: python
 
             import qibo
-            original_backend = qibo.get_backend()
             qibo.set_backend('qibojit')
             from qibo import models, gates, set_threads
             from qibo.parallel import parallel_parametrized_execution
@@ -88,35 +150,33 @@ def parallel_parametrized_execution(
             set_threads(1)
             # execute in parallel
             results = parallel_parametrized_execution(circuit, parameters, processes=2)
-            qibo.set_backend(original_backend)
 
     Args:
         circuit (qibo.models.Circuit): the input circuit.
         parameters (list): list of parameters for the circuit evaluation.
         initial_state (np.array): initial state for the circuit evaluation.
         processes (int): number of processes for parallel evaluation.
+            This corresponds to the number of threads, if a single thread is used
+            for each circuit evaluation. If more threads are used for each circuit
+            evaluation then some tuning may be required to obtain optimal performance.
+            Default is ``None`` which corresponds to a single thread.
 
     Returns:
         Circuit evaluation for input parameters.
     """
     if backend is None:  # pragma: no cover
-        from qibo.backends import GlobalBackend
-
         backend = GlobalBackend()
 
     if not isinstance(parameters, list):  # pragma: no cover
-        from qibo.config import raise_error
-
-        raise_error(RuntimeError, "parameters must be a list.")
+        raise_error(TypeError, "parameters must be a list.")
 
     def operation(params, circuit, state):
+        backend.set_threads(backend.nthreads)
         if state is not None:
             state = backend.cast(state, copy=True)
         circuit.set_parameters(params)
         return backend.execute_circuit(circuit, state)
 
-    from joblib import Parallel, delayed
-
     results = Parallel(n_jobs=processes, prefer="threads")(
         delayed(operation)(param, circuit.copy(deep=True), initial_state)
         for param in parameters

tests/test_parallel.py

@@ -9,11 +9,15 @@
 import qibo
 from qibo import gates
 from qibo.models import QFT, Circuit
-from qibo.parallel import parallel_execution, parallel_parametrized_execution
+from qibo.parallel import (
+    parallel_circuits_execution,
+    parallel_execution,
+    parallel_parametrized_execution,
+)
 
 
 @pytest.mark.skipif(sys.platform == "darwin", reason="Mac tests")
-def test_parallel_circuit_evaluation(backend):
+def test_parallel_states_evaluation(backend):
     """Evaluate circuit for multiple input states."""
     nqubits = 10
     np.random.seed(0)
@@ -31,6 +35,53 @@ def test_parallel_circuit_evaluation(backend):
     backend.assert_allclose(r1, r2)
 
 
+@pytest.mark.skipif(sys.platform == "darwin", reason="Mac tests")
+@pytest.mark.parametrize("use_execute_circuits", [False, True])
+def test_parallel_circuit_evaluation(backend, use_execute_circuits):
+    """Evaluate multiple circuits in parallel."""
+    circuits = [QFT(n) for n in range(1, 11)]
+
+    r1 = []
+    for circuit in circuits:
+        r1.append(backend.execute_circuit(circuit))
+
+    if use_execute_circuits:
+        r2 = backend.execute_circuits(circuits, processes=2)
+    else:
+        r2 = parallel_circuits_execution(circuits, processes=2, backend=backend)
+
+    for x, y in zip(r1, r2):
+        target = x.state(numpy=True)
+        final = y.state(numpy=True)
+        backend.assert_allclose(final, target)
+
+
+@pytest.mark.skipif(sys.platform == "darwin", reason="Mac tests")
+def test_parallel_circuit_states_evaluation(backend):
+    """Evaluate multiple circuits in parallel with different initial states."""
+    circuits = [QFT(n) for n in range(1, 11)]
+    states = [np.random.random(2**n) for n in range(1, 11)]
+
+    with pytest.raises(TypeError):
+        r2 = parallel_circuits_execution(
+            circuits, states=1, processes=2, backend=backend
+        )
+    with pytest.raises(ValueError):
+        r2 = parallel_circuits_execution(
+            circuits, states[:3], processes=2, backend=backend
+        )
+
+    r1 = []
+    for circuit, state in zip(circuits, states):
+        r1.append(backend.execute_circuit(circuit, state))
+
+    r2 = parallel_circuits_execution(circuits, states, processes=2, backend=backend)
+    for x, y in zip(r1, r2):
+        target = x.state(numpy=True)
+        final = y.state(numpy=True)
+        backend.assert_allclose(final, target)
+
+
 @pytest.mark.skipif(sys.platform == "darwin", reason="Mac tests")
 def test_parallel_parametrized_circuit(backend):
     """Evaluate circuit for multiple parameters."""