Refactor executor and job splitting

Separate logic for a custom executor and job splitting so both can be changed separately.
* Adds `executor` option to simulators for specifying a custom executor
* Adds `max_job_size` option to simulators for specifying the max number of circuits per job when submitting to executor. If less than the number of circuits this will result in a job set.
* Add max_size options to splitting code to control job sizes
* Clean up splitting code in aerbackend so there are not so many conditionals

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).
 
@@ -491,6 +499,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

@@ -32,10 +32,8 @@
 from qiskit.qobj import QasmQobj, PulseQobj
 from qiskit.compiler import assemble
 
-from ..aerjob import AerJob
-from ..aererror import AerError
-from .cluster.utils import split
-from .cluster.aerjobset import AerJobSet
+from qiskit.providers.aer.jobs import AerJob, AerJobSet, split_qobj
+from qiskit.providers.aer.aererror import AerError
 
 
 # Logger
@@ -65,6 +63,7 @@ def default(self, obj):
 
 class AerBackend(Backend, ABC):
     """Qiskit Aer Backend class."""
+
     def __init__(self,
                  configuration,
                  properties=None,
@@ -115,107 +114,46 @@ def __init__(self,
     @deprecate_arguments({'qobj': 'circuits'})
     def run(self,
             circuits,
-            backend_options=None,  # DEPRECATED
             validate=False,
             **run_options):
         """Run a qobj on the backend.
 
         Args:
             circuits (QuantumCircuit or list): The QuantumCircuit (or list
                 of QuantumCircuit objects) to run
-            backend_options (dict or None): DEPRECATED dictionary of backend options
-                                            for the execution (default: None).
             validate (bool): validate the Qobj before running (default: False).
             run_options (kwargs): additional run time backend options.
 
         Returns:
             AerJob: The simulation job.
 
         Additional Information:
-            * kwarg options specified in ``run_options`` will temporarily override
-              any set options of the same name for the current run.
-
-            * The entries in the ``backend_options`` will be combined with
-              the ``Qobj.config`` dictionary with the values of entries in
-              ``backend_options`` taking precedence. This kwarg is deprecated
-              and direct kwarg's should be used for options to pass them to
-              ``run_options``.
+            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
         """
-        # DEPRECATED
-        if backend_options is not None:
-            warnings.warn(
-                'Using `backend_options` kwarg has been deprecated as of'
-                ' qiskit-aer 0.7.0 and will be removed no earlier than 3'
-                ' months from that release date. Runtime backend options'
-                ' should now be added directly using kwargs for each option.',
-                DeprecationWarning,
-                stacklevel=3)
-
-        executor = None
-        if backend_options and "executor" in backend_options:
-            executor = backend_options["executor"]
-            del backend_options["executor"]
-
-        if "executor" in run_options:
-            executor = run_options["executor"]
-            del run_options["executor"]
-
-        if executor:
-            if isinstance(circuits, (QasmQobj, PulseQobj)):
-                experiments = split(circuits)
-            elif isinstance(circuits, (QuantumCircuit, Schedule)):
-                experiments = [assemble(circuits, self)]
-            elif (
-                    isinstance(circuits, list) and
-                    all(isinstance(circ, QuantumCircuit) for circ in circuits) or
-                    isinstance(circuits, list) and
-                    all(isinstance(circ, Schedule) for circ in circuits)
-            ):
-                experiments = [assemble(circ, self) for circ in circuits]
-            else:
-                raise ValueError(
-                    "run() is not implemented for this "
-                    "type of experiment ({})".format(str(type(circuits))))
-
-            for experiment in experiments:
-                self._add_options_to_qobj(experiment,
-                                          backend_options=backend_options,
-                                          **run_options)
-                # Optional validation
-                if validate:
-                    self._validate(experiment)
-
-            job_id = str(uuid.uuid4())
-            aer_job_set = AerJobSet(self, job_id, self._run, experiments, executor)
-            aer_job_set.submit()
-            return aer_job_set
-
+        if isinstance(circuits, (QasmQobj, PulseQobj)):
+            warnings.warn('Using a qobj for run() is deprecated and will be '
+                            'removed in a future release.',
+                            PendingDeprecationWarning,
+                            stacklevel=2)
+            qobj = circuits
         else:
-            if isinstance(circuits, (QasmQobj, PulseQobj)):
-                warnings.warn('Using a qobj for run() is deprecated and will be '
-                              'removed in a future release.',
-                              PendingDeprecationWarning,
-                              stacklevel=2)
-                qobj = circuits
-            else:
-                qobj = assemble(circuits, self)
+            qobj = assemble(circuits, self)
 
-            # Add backend options to the Job qobj
-            self._add_options_to_qobj(
-                qobj, backend_options=backend_options, **run_options)
+        # Add submit args for the job
+        experiments, executor = self._get_job_submit_args(qobj, validate=validate, **run_options)
 
-            # Optional validation
-            if validate:
-                self._validate(qobj)
-
-            # Submit job
-            job_id = str(uuid.uuid4())
-            aer_job = AerJob(self, job_id, self._run, qobj)
-            aer_job.submit()
-            return aer_job
+        # Submit job
+        job_id = str(uuid.uuid4())
+        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()
+        return aer_job
 
     def configuration(self):
         """Return the simulator backend configuration.
@@ -284,26 +222,6 @@ def status(self):
             pending_jobs=0,
             status_msg='')
 
-    def _run_job(self, job_id, qobj, backend_options=None, noise_model=None, validate=False):
-        """Run a qobj job"""
-        warnings.warn(
-            'The `_run_job` method has been deprecated. Use `_run` instead.',
-            DeprecationWarning)
-        if validate:
-            warnings.warn(
-                'The validate arg of `_run_job` has been removed. Use '
-                'validate=True in the `run` method instead.',
-                DeprecationWarning)
-
-        # The new function swaps positional args qobj and job id so we do a
-        # type check to swap them back
-        if not isinstance(job_id, str) and isinstance(qobj, str):
-            job_id, qobj = qobj, job_id
-        self._add_options_to_qobj(qobj,
-                                  backend_options=backend_options,
-                                  noise_model=noise_model)
-        return self._run(qobj, job_id)
-
     def _dummy_job(self):
         return
 
@@ -423,9 +341,7 @@ 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,
-                             backend_options=None,  # DEPRECATED
-                             **run_options):
+    def _get_job_submit_args(self, qobj, validate=False, **run_options):
         """Return execution sim config dict from backend options."""
         # Add options to qobj config overriding any existing fields
         config = qobj.config
@@ -435,16 +351,26 @@ def _add_options_to_qobj(self, qobj,
             if val is not None and not hasattr(config, key):
                 setattr(config, key, val)
 
-        # DEPRECATED backend options
-        if backend_options is not None:
-            for key, val in backend_options.items():
-                setattr(config, key, val)
-
         # Override with run-time options
         for key, val in run_options.items():
             setattr(config, key, val)
 
-        return qobj
+        # Get executor
+        executor = None
+        if hasattr(qobj.config, 'executor'):
+            executor = getattr(qobj.config, 'executor')
+            # We need to remove the executor from the qobj config
+            # since it can't be serialized though JSON/Pybind.
+            delattr(qobj.config, 'executor')
+
+        # 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).
 
@@ -380,6 +388,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/backends/cluster/__init__.py → qiskit/providers/aer/jobs/__init__.py

@@ -12,20 +12,9 @@
 # copyright notice, and modified files need to carry a notice indicating
 # that they have been altered from the originals.
 """
-===========================================================
-Cluster Backend (:mod:`qiskit.providers.aer.backends.cluster`)
-===========================================================
-
-.. currentmodule:: qiskit.providers.aer.backends.cluster
-
-High level mechanism for handling cluster jobs.
-
-Classes
-==========================
-.. autosummary::
-   :toctree: ../stubs/
-
-   ClusterBackend
+Aer simulator job management.
 """
 
+from .aerjob import AerJob
 from .aerjobset import AerJobSet
+from .utils import split_qobj