Support to configure executor for benchmark

src/orion/benchmark/__init__.py

@@ -9,7 +9,9 @@
 
 from tabulate import tabulate
 
+import orion.core
 from orion.client import create_experiment
+from orion.executor.base import Executor
 
 
 class Benchmark:
@@ -49,15 +51,22 @@ class Benchmark:
 
     storage: dict, optional
         Configuration of the storage backend.
+    executor: `orion.executor.base.Executor`, optional
+        Executor to run the benchmark experiments
     """
 
-    def __init__(self, name, algorithms, targets, storage=None):
+    def __init__(self, name, algorithms, targets, storage=None, executor=None):
         self._id = None
         self.name = name
         self.algorithms = algorithms
         self.targets = targets
         self.metadata = {}
         self.storage_config = storage
+        self.executor = executor or Executor(
+            orion.core.config.worker.executor,
+            n_workers=orion.core.config.worker.n_workers,
+            **orion.core.config.worker.executor_configuration,
+        )
 
         self.studies = []
 
@@ -319,6 +328,7 @@ def setup_experiments(self):
                     algorithms=algorithm.experiment_algorithm,
                     max_trials=max_trials,
                     storage=self.benchmark.storage_config,
+                    executor=self.benchmark.executor,
                 )
                 self.experiments_info.append((task_index, experiment))
 

src/orion/benchmark/benchmark_client.py

@@ -18,7 +18,7 @@
 
 
 def get_or_create_benchmark(
-    name, algorithms=None, targets=None, storage=None, debug=False
+    name, algorithms=None, targets=None, storage=None, executor=None, debug=False
 ):
     """
     Create or get a benchmark object.
@@ -38,6 +38,8 @@ def get_or_create_benchmark(
             Task objects
     storage: dict, optional
         Configuration of the storage backend.
+    executor: `orion.executor.base.Executor`, optional
+        Executor to run the benchmark experiments
     debug: bool, optional
         If using in debug mode, the storage config is overrided with legacy:EphemeralDB.
         Defaults to False.
@@ -66,7 +68,9 @@ def get_or_create_benchmark(
             "algorithms and targets space was not defined.".format(name)
         )
 
-    benchmark = _create_benchmark(name, algorithms, targets, storage=storage)
+    benchmark = _create_benchmark(
+        name, algorithms, targets, storage=storage, executor=executor
+    )
 
     if input_configure and input_benchmark.configuration != benchmark.configuration:
         logger.warn(
@@ -84,7 +88,7 @@ def get_or_create_benchmark(
                 "Benchmark registration failed. This is likely due to a race condition. "
                 "Now rolling back and re-attempting building it."
             )
-            get_or_create_benchmark(name, algorithms, targets, storage, debug)
+            get_or_create_benchmark(name, algorithms, targets, storage, executor, debug)
 
     return benchmark
 
@@ -126,9 +130,9 @@ def _resolve_db_config(db_config):
     return benchmark_id, algorithms, targets
 
 
-def _create_benchmark(name, algorithms, targets, storage):
+def _create_benchmark(name, algorithms, targets, storage, executor):
 
-    benchmark = Benchmark(name, algorithms, targets, storage)
+    benchmark = Benchmark(name, algorithms, targets, storage, executor)
     benchmark.setup_studies()
 
     return benchmark

src/orion/client/__init__.py

@@ -55,6 +55,7 @@ def build_experiment(
     heartbeat=None,
     working_dir=None,
     debug=False,
+    executor=None,
 ):
     """Build an experiment to be executable
 
@@ -176,6 +177,8 @@ def build_experiment(
         config_change_type: str, optional
             How to resolve config change automatically. Must be one of 'noeffect', 'unsure' or
             'break'.  Defaults to 'break'.
+    executor: `orion.executor.base.Executor`, optional
+        Executor to run the experiment
 
     Raises
     ------
@@ -240,7 +243,7 @@ def build_experiment(
 
     producer = Producer(experiment, max_idle_time)
 
-    return ExperimentClient(experiment, producer, heartbeat)
+    return ExperimentClient(experiment, producer, executor, heartbeat)
 
 
 def get_experiment(name, version=None, mode="r", storage=None):

src/orion/client/experiment.py

@@ -693,8 +693,9 @@ def workon(
         """
         self._check_if_executable()
 
+        # TODO: should this n_workers always be smaller than executor configured n_workers?
         if n_workers is None:
-            n_workers = orion.core.config.worker.n_workers
+            n_workers = self.executor.n_workers
 
         if max_trials is None:
             max_trials = self.max_trials

tests/unittests/benchmark/test_benchmark_client.py

@@ -15,6 +15,7 @@
 from orion.core.io.database.pickleddb import PickledDB
 from orion.core.utils.exceptions import NoConfigurationError
 from orion.core.utils.singleton import SingletonNotInstantiatedError, update_singletons
+from orion.executor.joblib_backend import Joblib
 from orion.storage.base import get_storage
 from orion.storage.legacy import Legacy
 from orion.testing.state import OrionState
@@ -306,3 +307,60 @@ def insert_race_condition(*args, **kwargs):
 
             assert bm.configuration == benchmark_config
             assert count_benchmarks() == 1
+
+    def test_create_with_executor(self, benchmark_config, benchmark_config_py):
+
+        with OrionState():
+            config = copy.deepcopy(benchmark_config_py)
+            bm1 = get_or_create_benchmark(**config)
+
+            assert bm1.configuration == benchmark_config
+            assert bm1.executor.n_workers == orion.core.config.worker.n_workers
+
+            executor = Joblib(n_workers=2, backend="threading")
+            config["executor"] = executor
+            bm2 = get_or_create_benchmark(**config)
+
+            assert bm2.configuration == benchmark_config
+            assert bm2.executor.n_workers == executor.n_workers
+
+    def test_experiments_parallel(
+        self, benchmark_config_py, benchmark_algorithms, experiment_config, monkeypatch
+    ):
+        from orion.benchmark import Study
+        from orion.testing import create_experiment
+
+        def foo(x):
+            return [dict(name="result", type="objective", value=x * 2)]
+
+        def optimize(*args, **kwargs):
+            optimize.count += 1
+            return 1
+
+        with create_experiment(
+            exp_config=experiment_config, trial_config={}, statuses=[]
+        ) as (
+            cfg,
+            experiment,
+            client,
+        ):
+            monkeypatch.setattr(client, "_optimize", optimize)
+
+            config = copy.deepcopy(benchmark_config_py)
+            executor = Joblib(n_workers=5, backend="threading")
+            config["executor"] = executor
+            bm1 = get_or_create_benchmark(**config)
+            study = Study(
+                bm1, benchmark_algorithms, AverageResult(2), RosenBrock(25, dim=3)
+            )
+
+            study.experiments_info = [(0, client)]
+            bm1.studies = [study]
+
+            optimize.count = 0
+            bm1.process(n_workers=2)
+            assert optimize.count == 2
+
+            optimize.count = 0
+            bm1.process(n_workers=3)
+            assert optimize.count == 3

tests/unittests/client/test_experiment_client.py

@@ -18,6 +18,7 @@
 )
 from orion.core.worker.trial import Trial
 from orion.executor.base import Executor
+from orion.executor.joblib_backend import Joblib
 from orion.storage.base import get_storage
 from orion.testing import create_experiment, mock_space_iterate
 
@@ -1168,3 +1169,9 @@ def optimize(*args, **kwargs):
             with client.tmp_executor("joblib", n_workers=5, backend="threading"):
                 client.workon(foo, max_trials=5, n_workers=3)
             assert optimize.count == 3
+
+            optimize.count = 0
+            executor = Joblib(n_workers=5, backend="threading")
+            client.executor = executor
+            client.workon(foo, max_trials=5, n_workers=4)
+            assert optimize.count == 4