Cluster backend

CONTRIBUTING.md

@@ -738,6 +738,45 @@ meta = dict['metadata']
 myrank = meta['mpi_rank']
 ```
 
+### Running with Threadpool and DASK
+
+Qiskit Aer can parallelize the simulation on a circuit bases with ThreadpoolExecutor or DASK as a custom executor. When user gives dask client as the executor, Aer can execute a simulation on the distributed machines like HPC clusters.
+If you want to install dask client at the same time as Qiskit Aer, please add `dask` option as follows. This option installs Aer, dask, and  distributed packages. 
+```
+pip install .[dask]
+```
+
+To use Threadpool or DASK as an executor, you need to set `executor` and `max_job_size` by `set_options` function. If both `executor` (default None) and `max_job_size` (default None) are set, Aer splits the multiple circuits to some chunk of circuits and submits them to the executor. `max_job_size` can control the number of splitting circuits. When `max_job_size` is set to 1, multiple circuits are split into one circuit and distributed to the executor. If user executes 60 circuits with the executor and `max_job_size=1`, Aer splits it to 1 circuit x 60 jobs. 
+If 60 circuits and `max_job_size=2`, Aer splits it to 2 circuits x 30 jobs. 
+
+Example Usage:
+Threadpool execution:
+
+```
+from concurrent.futures import ThreadPoolExecutor
+
+exc = ThreadPoolExecutor(max_workers=2)
+qbackend = Aer.get_backend('qasm_simulator')
+qbackend.set_options(executor=exc)
+qbackend.set_options(max_job_size=3)
+result = qbackend.run(circuits, seed_simulator=54321).result()
+```
+
+Dask execution:
+
+Dask client creates multi-processes so you need to gurad it by if __name__ == "__main__": block. 
+```
+from dask.distributed import Client
+
+def dask_exec():
+    exc = Client(address=LocalCluster(n_workers=1, processes=True))
+    qbackend.set_options(executor=exc)
+    qbackend.set_options(max_job_size=3)
+    result = qbackend.run(circuits, seed_simulator=54321).result()
+
+if __name__ == "__main__":
+    dask_exec()
+```
 
 ### Building a statically linked wheel
 

qiskit/providers/aer/__init__.py

@@ -44,12 +44,13 @@
     StatevectorSimulator
     UnitarySimulator
 
-Job Class
-=========
+Job Classes
+===========
 .. autosummary::
    :toctree: ../stubs/
 
    AerJob
+   AerJobSet
 
 Exceptions
 ==========
@@ -70,7 +71,7 @@
 
 # pylint: disable=wrong-import-position
 from .aerprovider import AerProvider
-from .aerjob import AerJob
+from .jobs import AerJob, AerJobSet
 from .aererror import AerError
 from .backends import *
 from . import library

qiskit/providers/aer/backends/aer_simulator.py

@@ -158,6 +158,14 @@ class AerSimulator(AerBackend):
       certain simulation methods to either ``"single"`` or ``"double"``
       precision (default: ``"double"``).
 
+    * ``executor`` (futures.Executor or None): Set a custom executor for
+      asynchronous running of simulation jobs (Default: None).
+
+    * ``max_job_size`` (int or None): If the number of run circuits
+      exceeds this value simulation will be run as a set of of sub-jobs
+      on the executor. If ``None`` simulation of all circuits are submitted
+      to the executor as a single job (Default: None).
+
     * ``zero_threshold`` (double): Sets the threshold for truncating
       small values to zero in the result data (Default: 1e-10).
 
@@ -496,6 +504,8 @@ def _default_options(cls):
             method='automatic',
             device='CPU',
             precision="double",
+            executor=None,
+            max_job_size=None,
             zero_threshold=1e-10,
             validation_threshold=None,
             max_parallel_threads=None,

qiskit/providers/aer/backends/aerbackend.py

@@ -30,9 +30,10 @@
 from qiskit.qobj import QasmQobj, PulseQobj
 from qiskit.compiler import assemble
 
-from ..aerjob import AerJob
+from ..jobs import AerJob, AerJobSet, split_qobj
 from ..aererror import AerError
 
+
 # Logger
 logger = logging.getLogger(__name__)
 
@@ -60,6 +61,7 @@ def default(self, obj):
 
 class AerBackend(Backend, ABC):
     """Qiskit Aer Backend class."""
+
     def __init__(self,
                  configuration,
                  properties=None,
@@ -98,6 +100,7 @@ def __init__(self,
         self._options_configuration = {}
         self._options_defaults = {}
         self._options_properties = {}
+        self._executor = None
 
         # Set available methods
         self._available_methods = [] if available_methods is None else available_methods
@@ -126,6 +129,9 @@ def run(self,
         Additional Information:
             kwarg options specified in ``run_options`` will temporarily override
             any set options of the same name for the current run.
+
+        Raises:
+            ValueError: if run is not implemented
         """
         if isinstance(circuits, (QasmQobj, PulseQobj)):
             warnings.warn('Using a qobj for run() is deprecated and will be '
@@ -148,17 +154,18 @@ def run(self,
         else:
             qobj = assemble(circuits, self)
 
-        # Add backend options to the Job qobj
-        self._add_options_to_qobj(qobj, **run_options)
-
-        # Optional validation
-        if validate:
-            self._validate(qobj)
+        # Add submit args for the job
+        experiments, executor = self._get_job_submit_args(qobj, validate=validate, **run_options)
+        executor = executor or self._executor
 
         # Submit job
         job_id = str(uuid.uuid4())
-        aer_job = AerJob(self, job_id, self._run, qobj)
+        if isinstance(experiments, list):
+            aer_job = AerJobSet(self, job_id, self._run, experiments, executor)
+        else:
+            aer_job = AerJob(self, job_id, self._run, experiments, executor)
         aer_job.submit()
+        self._executor = executor
         return aer_job
 
     def configuration(self):
@@ -319,6 +326,7 @@ def set_option(self, key, value):
                 setattr(self._options, key, getattr(self._default_options(), key))
 
     def set_options(self, **fields):
+        """Set the simulator options"""
         for key, value in fields.items():
             self.set_option(key, value)
 
@@ -343,9 +351,16 @@ def _set_defaults_option(self, key, value):
         elif key in self._options_defaults:
             self._options_defaults.pop(key)
 
-    def _add_options_to_qobj(self, qobj,
-                             **run_options):
+    def _get_job_submit_args(self, qobj, validate=False, **run_options):
         """Return execution sim config dict from backend options."""
+        # Get executor
+        executor = None
+        if hasattr(self._options, 'executor'):
+            executor = getattr(self._options, 'executor')
+            # We need to remove the executor from the qobj config
+            # since it can't be serialized though JSON/Pybind.
+            delattr(self._options, 'executor')
+
         # Add options to qobj config overriding any existing fields
         config = qobj.config
 
@@ -358,7 +373,14 @@ def _add_options_to_qobj(self, qobj,
         for key, val in run_options.items():
             setattr(config, key, val)
 
-        return qobj
+        # Optional validation
+        if validate:
+            self._validate(qobj)
+
+        # Split circuits for sub-jobs
+        experiments = split_qobj(
+            qobj, max_size=getattr(qobj.config, 'max_job_size', None))
+        return experiments, executor
 
     def __repr__(self):
         """String representation of an AerBackend."""

qiskit/providers/aer/backends/pulse_simulator.py

@@ -119,7 +119,13 @@ class PulseSimulator(AerBackend):
       ``|1>``. Defaults to ``2``.
     * ``meas_return``: Measurement type, ``'single'`` or ``'avg'``. Defaults to ``'avg'``.
     * ``shots``: Number of shots per experiment. Defaults to ``1024``.
+    * ``executor``: Set a custom executor for asynchronous running of simulation
+    * ``max_job_size`` (int or None): If the number of run schedules
+      exceeds this value simulation will be run as a set of of sub-jobs
+      on the executor. If ``None`` simulation of all schedules are submitted
+      to the executor as a single job (Default: None).
 
+      jobs (Default: None).
 
     **Simulation details**
 
@@ -187,7 +193,9 @@ def _default_options(cls):
             qubit_freq_est=inf,
             q_level_meas=1,
             noise_model=None,
-            initial_state=None)
+            initial_state=None,
+            executor=None,
+            max_job_size=None)
 
     # pylint: disable=arguments-differ, missing-param-doc
     @deprecate_arguments({'qobj': 'schedules'})

qiskit/providers/aer/backends/qasm_simulator.py

@@ -116,6 +116,14 @@ class QasmSimulator(AerBackend):
       certain simulation methods to either ``"single"`` or ``"double"``
       precision (default: ``"double"``).
 
+    * ``executor`` (futures.Executor): Set a custom executor for
+      asynchronous running of simulation jobs (Default: None).
+
+    * ``max_job_size`` (int or None): If the number of run circuits
+      exceeds this value simulation will be run as a set of of sub-jobs
+      on the executor. If ``None`` simulation of all circuits are submitted
+      to the executor as a single job (Default: None).
+
     * ``zero_threshold`` (double): Sets the threshold for truncating
       small values to zero in the result data (Default: 1e-10).
 
@@ -385,6 +393,8 @@ def _default_options(cls):
             shots=1024,
             method=None,
             precision="double",
+            executor=None,
+            max_job_size=None,
             zero_threshold=1e-10,
             validation_threshold=None,
             max_parallel_threads=None,

qiskit/providers/aer/backends/statevector_simulator.py

@@ -63,6 +63,14 @@ class StatevectorSimulator(AerBackend):
       certain simulation methods to either ``"single"`` or ``"double"``
       precision (default: ``"double"``).
 
+    * ``executor`` (futures.Executor): Set a custom executor for
+      asynchronous running of simulation jobs (Default: None).
+
+    * ``max_job_size`` (int or None): If the number of run circuits
+      exceeds this value simulation will be run as a set of of sub-jobs
+      on the executor. If ``None`` simulation of all circuits aer submitted
+      to the executor as a single job (Default: None).
+
     * ``zero_threshold`` (double): Sets the threshold for truncating
       small values to zero in the result data (Default: 1e-10).
 
@@ -183,6 +191,8 @@ def _default_options(cls):
             shots=1024,
             method="automatic",
             precision="double",
+            executor=None,
+            max_job_size=None,
             zero_threshold=1e-10,
             validation_threshold=None,
             max_parallel_threads=None,

qiskit/providers/aer/backends/unitary_simulator.py

@@ -65,6 +65,14 @@ class UnitarySimulator(AerBackend):
       certain simulation methods to either ``"single"`` or ``"double"``
       precision (default: ``"double"``).
 
+    * ``executor`` (futures.Executor): Set a custom executor for
+      asynchronous running of simulation jobs (Default: None).
+
+    * ``max_job_size`` (int or None): If the number of run circuits
+      exceeds this value simulation will be run as a set of of sub-jobs
+      on the executor. If ``None`` simulation of all circuits are submitted
+      to the executor as a single job (Default: None).
+
     * ``"initial_unitary"`` (matrix_like): Sets a custom initial unitary
       matrix for the simulation instead of identity (Default: None).
 
@@ -180,6 +188,8 @@ def _default_options(cls):
             shots=1024,
             method="automatic",
             precision="double",
+            executor=None,
+            max_job_size=None,
             zero_threshold=1e-10,
             seed_simulator=None,
             validation_threshold=None,

qiskit/providers/aer/jobs/__init__.py

@@ -0,0 +1,20 @@
+# -*- coding: utf-8 -*-
+
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2018, 2019.
+#
+# This code is licensed under the Apache License, Version 2.0. You may
+# obtain a copy of this license in the LICENSE.txt file in the root directory
+# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
+#
+# Any modifications or derivative works of this code must retain this
+# copyright notice, and modified files need to carry a notice indicating
+# that they have been altered from the originals.
+"""
+Aer simulator job management.
+"""
+
+from .aerjob import AerJob
+from .aerjobset import AerJobSet
+from .utils import split_qobj