Repository: Epistimio/orion

Issue:
Experiment cannot reserve trial of parent experiment
This is very problematic as non-completed trials of parents cannot be execute anymore unless the environment state is reverted to the one used for parent experiment (ex: resetting code). It should look for executable trials across the EVC tree.

From @hbertrand 

Discussion:

Patch:
diff --git a/src/orion/client/experiment.py b/src/orion/client/experiment.py
index f0a17485..a7054104 100644
--- a/src/orion/client/experiment.py
+++ b/src/orion/client/experiment.py
@@ -294,6 +294,15 @@ class ExperimentClient:
             status, with_evc_tree=with_evc_tree
         )
 
+    def fetch_pending_trials(self, with_evc_tree=False):
+        """Fetch all trials with status new, interrupted or suspended
+
+        Trials are sorted based on ``Trial.submit_time``
+
+        :return: list of :class:`orion.core.worker.trial.Trial` objects
+        """
+        return self._experiment.fetch_pending_trials(with_evc_tree=with_evc_tree)
+
     def fetch_noncompleted_trials(self, with_evc_tree=False):
         """Fetch non-completed trials of this `Experiment` instance.
 
diff --git a/src/orion/core/worker/experiment.py b/src/orion/core/worker/experiment.py
index 8cbe623a..094e90d0 100644
--- a/src/orion/core/worker/experiment.py
+++ b/src/orion/core/worker/experiment.py
@@ -16,6 +16,7 @@ import pandas
 
 from orion.core.evc.adapters import BaseAdapter
 from orion.core.evc.experiment import ExperimentNode
+from orion.core.io.database import DuplicateKeyError
 from orion.core.utils.exceptions import UnsupportedOperation
 from orion.core.utils.flatten import flatten
 from orion.core.utils.singleton import update_singletons
@@ -240,6 +241,8 @@ class Experiment:
 
         self.fix_lost_trials()
 
+        self.duplicate_pending_trials()
+
         selected_trial = self._storage.reserve_trial(self)
         log.debug("reserved trial (trial: %s)", selected_trial)
         return selected_trial
@@ -265,6 +268,43 @@ class Experiment:
             except FailedUpdate:
                 log.debug("failed")
 
+    def duplicate_pending_trials(self):
+        """Find pending trials in EVC and duplicate them in current experiment.
+
+        An experiment cannot execute trials from parent experiments otherwise some trials
+        may have been executed in different environements of different experiment although they
+        belong to the same experiment. Instead, trials that are pending in parent and child
+        experiment are copied over to current experiment so that it can be reserved and executed.
+        The parent or child experiment will only see their original copy of the trial, and
+        the current experiment will only see the new copy of the trial.
+        """
+        self._check_if_writable()
+        evc_pending_trials = self._select_evc_call(
+            with_evc_tree=True, function="fetch_pending_trials"
+        )
+        exp_pending_trials = self._select_evc_call(
+            with_evc_tree=False, function="fetch_pending_trials"
+        )
+
+        exp_trials_ids = set(
+            trial.compute_trial_hash(trial, ignore_experiment=True)
+            for trial in exp_pending_trials
+        )
+
+        for trial in evc_pending_trials:
+            if (
+                trial.compute_trial_hash(trial, ignore_experiment=True)
+                in exp_trials_ids
+            ):
+                continue
+
+            trial.experiment = self.id
+            # Danger danger, race conditions!
+            try:
+                self._storage.register_trial(trial)
+            except DuplicateKeyError:
+                log.debug("Race condition while trying to duplicate trial %s", trial.id)
+
     # pylint:disable=unused-argument
     def update_completed_trial(self, trial, results_file=None):
         """Inform database about an evaluated `trial` with results.
@@ -354,6 +394,15 @@ class Experiment:
         """
         return self._select_evc_call(with_evc_tree, "fetch_trials_by_status", status)
 
+    def fetch_pending_trials(self, with_evc_tree=False):
+        """Fetch all trials with status new, interrupted or suspended
+
+        Trials are sorted based on `Trial.submit_time`
+
+        :return: list of `Trial` objects
+        """
+        return self._select_evc_call(with_evc_tree, "fetch_pending_trials")
+
     def fetch_noncompleted_trials(self, with_evc_tree=False):
         """Fetch non-completed trials of this `Experiment` instance.
 
diff --git a/src/orion/core/worker/strategy.py b/src/orion/core/worker/strategy.py
index 1b2a26ea..7a9df269 100644
--- a/src/orion/core/worker/strategy.py
+++ b/src/orion/core/worker/strategy.py
@@ -142,9 +142,11 @@ class MaxParallelStrategy(BaseParallelStrategy):
     def observe(self, points, results):
         """See BaseParallelStrategy.observe"""
         super(MaxParallelStrategy, self).observe(points, results)
-        self.max_result = max(
+        results = [
             result["objective"] for result in results if result["objective"] is not None
-        )
+        ]
+        if results:
+            self.max_result = max(results)
 
     def lie(self, trial):
         """See BaseParallelStrategy.lie"""
@@ -175,9 +177,10 @@ class MeanParallelStrategy(BaseParallelStrategy):
         objective_values = [
             result["objective"] for result in results if result["objective"] is not None
         ]
-        self.mean_result = sum(value for value in objective_values) / float(
-            len(objective_values)
-        )
+        if objective_values:
+            self.mean_result = sum(value for value in objective_values) / float(
+                len(objective_values)
+            )
 
     def lie(self, trial):
         """See BaseParallelStrategy.lie"""


Test patch:
diff --git a/src/orion/testing/evc.py b/src/orion/testing/evc.py
new file mode 100644
index 00000000..74f2876c
--- /dev/null
+++ b/src/orion/testing/evc.py
@@ -0,0 +1,85 @@
+import contextlib
+import copy
+
+from orion.client import build_experiment, get_experiment
+
+
+@contextlib.contextmanager
+def disable_duplication(monkeypatch):
+    def stub(self):
+        pass
+
+    with monkeypatch.context() as m:
+        m.setattr(
+            "orion.core.worker.experiment.Experiment.duplicate_pending_trials", stub
+        )
+
+        yield
+
+
+def generate_trials(exp, trials):
+    """Generate trials for each item in trials.
+
+    Items of trials can be either dictionary of valid hyperparameters based on exp.space and status
+    or `None`.
+
+    If status not provided, 'new' is used by default.
+
+    For items that are `None`, trials are suggested with exp.suggest().
+    """
+    for trial_config in trials:
+        trial_config = copy.deepcopy(trial_config)
+        status = trial_config.pop("status", None) if trial_config else None
+        if trial_config:
+            trial = exp.insert(params=trial_config)
+        else:
+            with exp.suggest() as trial:
+                # Releases suggested trial when leaving with-clause.
+                pass
+
+        if status is not None:
+            print(status)
+            exp._experiment._storage.set_trial_status(trial, status)
+
+        print([trial.status for trial in exp.fetch_trials()])
+
+
+def build_root_experiment(space=None, trials=None):
+    """Build a root experiment and generate trials."""
+    if space is None:
+        space = {"x": "uniform(0, 100)", "y": "uniform(0, 100)", "z": "uniform(0, 100)"}
+    if trials is None:
+        trials = [{"x": i, "y": i * 2, "z": i ** 2} for i in range(4)]
+
+    root = build_experiment(name="root", max_trials=len(trials), space=space)
+
+    generate_trials(root, trials)
+
+
+def build_child_experiment(space=None, trials=None, name="child", parent="root"):
+    """Build a child experiment by branching from `parent` and generate trials."""
+    if trials is None:
+        trials = [None for i in range(6)]
+
+    max_trials = get_experiment(parent).max_trials + len(trials)
+
+    child = build_experiment(
+        name=name,
+        space=space,
+        max_trials=max_trials,
+        branching={"branch_from": parent, "enable": True},
+    )
+    assert child.name == name
+    assert child.version == 1
+
+    generate_trials(child, trials)
+
+
+def build_grand_child_experiment(space=None, trials=None):
+    """Build a grand-child experiment by branching from `child` and generate trials."""
+    if trials is None:
+        trials = [None for i in range(5)]
+
+    build_child_experiment(
+        space=space, trials=trials, name="grand-child", parent="child"
+    )
diff --git a/tests/functional/branching/test_branching.py b/tests/functional/branching/test_branching.py
index d4e8bae2..2f7e965d 100644
--- a/tests/functional/branching/test_branching.py
+++ b/tests/functional/branching/test_branching.py
@@ -831,15 +831,15 @@ def test_run_entire_full_x_full_y(init_entire):
 
     orion.core.cli.main(
         (
-            "-vv hunt --max-trials 20 --pool-size 1 -n full_x_full_y "
+            "-vv hunt --max-trials 30 --pool-size 1 -n full_x_full_y "
             "./black_box_with_y.py "
             "-x~uniform(-10,10) "
             "-y~uniform(-10,10,default_value=1)"
         ).split(" ")
     )
 
-    assert len(experiment.fetch_trials(with_evc_tree=True)) == 39
-    assert len(experiment.fetch_trials()) == 20
+    assert len(experiment.fetch_trials(with_evc_tree=True)) == 30
+    assert len(experiment.fetch_trials(with_evc_tree=False)) == 30
 
 
 def test_run_entire_full_x_full_y_no_args(init_entire):
@@ -850,11 +850,11 @@ def test_run_entire_full_x_full_y_no_args(init_entire):
     assert len(experiment.fetch_trials()) == 4
 
     orion.core.cli.main(
-        ("-vv hunt --max-trials 20 --pool-size 1 -n full_x_full_y").split(" ")
+        ("-vv hunt --max-trials 30 --pool-size 1 -n full_x_full_y").split(" ")
     )
 
-    assert len(experiment.fetch_trials(with_evc_tree=True)) == 39
-    assert len(experiment.fetch_trials()) == 20
+    assert len(experiment.fetch_trials(with_evc_tree=True)) == 30
+    assert len(experiment.fetch_trials(with_evc_tree=False)) == 30
 
 
 def test_new_algo(init_full_x_new_algo):
@@ -872,8 +872,8 @@ def test_new_algo(init_full_x_new_algo):
         ("-vv hunt --max-trials 20 --pool-size 1 -n full_x_new_algo").split(" ")
     )
 
-    assert len(experiment.fetch_trials(with_evc_tree=True)) == 21
-    assert len(experiment.fetch_trials()) == 20
+    assert len(experiment.fetch_trials(with_evc_tree=True)) == 20
+    assert len(experiment.fetch_trials(with_evc_tree=False)) == 20
 
 
 def test_new_algo_not_resolved(init_full_x, capsys):
@@ -1002,8 +1002,8 @@ def test_new_cli(init_full_x_new_cli):
         ("-vv hunt --max-trials 20 --pool-size 1 -n full_x_new_cli").split(" ")
     )
 
-    assert len(experiment.fetch_trials(with_evc_tree=True)) == 21
-    assert len(experiment.fetch_trials()) == 20
+    assert len(experiment.fetch_trials(with_evc_tree=True)) == 20
+    assert len(experiment.fetch_trials(with_evc_tree=False)) == 20
 
 
 @pytest.mark.usefixtures("init_full_x")
diff --git a/tests/functional/core/worker/test_experiment_functional.py b/tests/functional/core/worker/test_experiment_functional.py
new file mode 100644
index 00000000..27fc75b6
--- /dev/null
+++ b/tests/functional/core/worker/test_experiment_functional.py
@@ -0,0 +1,147 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""Collection of functional tests for :mod:`orion.core.worker.experiment`."""
+import logging
+
+from orion.client import build_experiment, get_experiment
+from orion.core.io.database import DuplicateKeyError
+from orion.core.worker.trial import Trial
+from orion.testing.evc import (
+    build_child_experiment,
+    build_grand_child_experiment,
+    build_root_experiment,
+    disable_duplication,
+)
+
+
+def test_fix_lost_trials_in_evc():
+    pass
+
+
+SPACE = {"x": "uniform(0, 100)"}
+N_PENDING = 3  # new, interrupted and suspended
+
+
+def generate_trials_list(level, stati=Trial.allowed_stati):
+    return [
+        {"status": trial_status, "x": i + len(stati) * level}
+        for i, trial_status in enumerate(stati)
+    ]
+
+
+status = []
+
+
+def build_evc_tree(levels):
+    build_root_experiment(space=SPACE, trials=generate_trials_list(levels[0]))
+    names = ["root", "parent", "experiment", "child", "grand-child"]
+    for level, (parent, name) in zip(levels[1:], zip(names, names[1:])):
+        build_child_experiment(
+            space=SPACE, name=name, parent=parent, trials=generate_trials_list(level)
+        )
+
+
+def test_duplicate_pending_trials(storage, monkeypatch):
+    """Test that only pending trials are duplicated"""
+    with disable_duplication(monkeypatch):
+        build_evc_tree(list(range(5)))
+
+    for exp in ["root", "parent", "experiment", "child", "grand-child"]:
+        assert len(get_experiment(name=exp).fetch_trials(with_evc_tree=False)) == len(
+            Trial.allowed_stati
+        )
+
+    experiment = build_experiment(name="experiment")
+    experiment._experiment.duplicate_pending_trials()
+
+    for exp in ["root", "parent", "child", "grand-child"]:
+        assert len(get_experiment(name=exp).fetch_trials(with_evc_tree=False)) == len(
+            Trial.allowed_stati
+        )
+
+    assert (
+        len(experiment.fetch_trials(with_evc_tree=False))
+        == len(Trial.allowed_stati) + N_PENDING * 4
+    )
+
+
+def test_duplicate_closest_duplicated_pending_trials(storage, monkeypatch):
+    """Test that only closest duplicated pending trials are duplicated"""
+    with disable_duplication(monkeypatch):
+        build_evc_tree([0, 0, 1, 2, 2])
+
+    for exp in ["root", "parent", "experiment", "child", "grand-child"]:
+        assert len(get_experiment(name=exp).fetch_trials(with_evc_tree=False)) == len(
+            Trial.allowed_stati
+        )
+
+    experiment = build_experiment(name="experiment")
+    experiment._experiment.duplicate_pending_trials()
+
+    for exp in ["root", "parent", "child", "grand-child"]:
+        assert len(get_experiment(name=exp).fetch_trials(with_evc_tree=False)) == len(
+            Trial.allowed_stati
+        )
+
+    assert (
+        len(experiment.fetch_trials(with_evc_tree=False))
+        == len(Trial.allowed_stati) + N_PENDING * 2
+    )
+
+
+def test_duplicate_only_once(storage, monkeypatch):
+    """Test that trials may not be duplicated twice"""
+    with disable_duplication(monkeypatch):
+        build_evc_tree(list(range(5)))
+
+    for exp in ["root", "parent", "experiment", "child", "grand-child"]:
+        assert len(get_experiment(name=exp).fetch_trials(with_evc_tree=False)) == len(
+            Trial.allowed_stati
+        )
+
+    experiment = build_experiment(name="experiment")
+    experiment._experiment.duplicate_pending_trials()
+
+    for exp in ["root", "parent", "child", "grand-child"]:
+        assert len(get_experiment(name=exp).fetch_trials(with_evc_tree=False)) == len(
+            Trial.allowed_stati
+        )
+
+    assert (
+        len(experiment.fetch_trials(with_evc_tree=False))
+        == len(Trial.allowed_stati) + N_PENDING * 4
+    )
+
+    experiment._experiment.duplicate_pending_trials()
+
+    for exp in ["root", "parent", "child", "grand-child"]:
+        assert len(get_experiment(name=exp).fetch_trials(with_evc_tree=False)) == len(
+            Trial.allowed_stati
+        )
+
+    assert (
+        len(experiment.fetch_trials(with_evc_tree=False))
+        == len(Trial.allowed_stati) + N_PENDING * 4
+    )
+
+
+def test_duplicate_race_conditions(storage, monkeypatch, caplog):
+    """Test that duplication does not raise an error during race conditions."""
+    with disable_duplication(monkeypatch):
+        build_evc_tree(list(range(2)))
+
+    experiment = build_experiment(name="parent")
+
+    def register_race_condition(trial):
+        raise DuplicateKeyError("Race condition!")
+
+    monkeypatch.setattr(
+        experiment._experiment._storage, "register_trial", register_race_condition
+    )
+
+    assert len(experiment.fetch_trials(with_evc_tree=False)) == len(Trial.allowed_stati)
+
+    with caplog.at_level(logging.DEBUG):
+        experiment._experiment.duplicate_pending_trials()
+
+        assert "Race condition while trying to duplicate trial" in caplog.text
diff --git a/tests/unittests/client/test_experiment_client.py b/tests/unittests/client/test_experiment_client.py
index e502a7f4..c518f6f2 100644
--- a/tests/unittests/client/test_experiment_client.py
+++ b/tests/unittests/client/test_experiment_client.py
@@ -104,6 +104,12 @@ def test_experiment_fetch_trials_by_status():
         )
 
 
+def test_experiment_fetch_pending_trials():
+    """Test compliance of client and experiment `fetch_pending_trials()`"""
+    with create_experiment(config, base_trial) as (cfg, experiment, client):
+        compare_trials(experiment.fetch_pending_trials(), client.fetch_pending_trials())
+
+
 def test_experiment_fetch_non_completed_trials():
     """Test compliance of client and experiment `fetch_noncompleted_trials()`"""
     with create_experiment(config, base_trial) as (cfg, experiment, client):
diff --git a/tests/unittests/core/evc/test_experiment_tree.py b/tests/unittests/core/evc/test_experiment_tree.py
index 1774b1cd..4c2b11b6 100644
--- a/tests/unittests/core/evc/test_experiment_tree.py
+++ b/tests/unittests/core/evc/test_experiment_tree.py
@@ -7,63 +7,12 @@ import pytest
 from orion.client import build_experiment, get_experiment
 from orion.core.evc.adapters import Adapter, CodeChange
 from orion.core.evc.experiment import ExperimentNode
-
-
-def generate_trials(exp, params):
-    """Generate trials for each item in params.
-
-    Items of params can be either dictionary of valid hyperparameters based on exp.space or `None`.
-    For items that are `None`, trials are suggested with exp.suggest().
-    """
-    for trial_params in params:
-        if trial_params is None:
-            with exp.suggest() as trial:
-                # Releases suggested trial when leaving with-clause.
-                pass
-        else:
-            exp.insert(params=trial_params)
-
-
-def build_root_experiment(space=None, trials=None):
-    """Build a root experiment and generate trials."""
-    if space is None:
-        space = {"x": "uniform(0, 100)", "y": "uniform(0, 100)", "z": "uniform(0, 100)"}
-    if trials is None:
-        trials = [{"x": i, "y": i * 2, "z": i ** 2} for i in range(4)]
-
-    root = build_experiment(name="root", max_trials=len(trials), space=space)
-
-    generate_trials(root, trials)
-
-
-def build_child_experiment(space=None, trials=None, name="child", parent="root"):
-    """Build a child experiment by branching from `parent` and generate trials."""
-    if trials is None:
-        trials = [None for i in range(6)]
-
-    max_trials = get_experiment(parent).max_trials + len(trials)
-
-    child = build_experiment(
-        name=name,
-        space=space,
-        max_trials=max_trials,
-        branching={"branch_from": parent, "enable": True},
-    )
-    assert child.name == name
-    assert child.version == 1
-
-    generate_trials(child, trials)
-
-
-def build_grand_child_experiment(space=None, trials=None):
-    """Build a grand-child experiment by branching from `child` and generate trials."""
-    if trials is None:
-        trials = [None for i in range(5)]
-
-    build_child_experiment(
-        space=space, trials=trials, name="grand-child", parent="child"
-    )
-
+from orion.testing.evc import (
+    build_child_experiment,
+    build_grand_child_experiment,
+    build_root_experiment,
+    disable_duplication,
+)
 
 ROOT_SPACE_WITH_DEFAULTS = {
     "x": "uniform(0, 100, default_value=0)",
@@ -590,10 +539,13 @@ parametrization = {
     list(parametrization.values()),
     ids=list(parametrization.keys()),
 )
-def test_evc_fetch_adapters(storage, root, child, grand_child, test_kwargs):
+def test_evc_fetch_adapters(
+    monkeypatch, storage, root, child, grand_child, test_kwargs
+):
     """Test the recursive fetch of trials in the EVC tree."""
-    build_root_experiment(**root)
-    build_child_experiment(**child)
-    if grand_child is not None:
-        build_grand_child_experiment(**grand_child)
+    with disable_duplication(monkeypatch):
+        build_root_experiment(**root)
+        build_child_experiment(**child)
+        if grand_child is not None:
+            build_grand_child_experiment(**grand_child)
     generic_tree_test(**test_kwargs)
diff --git a/tests/unittests/core/test_strategy.py b/tests/unittests/core/test_strategy.py
index d679a12e..609815fd 100644
--- a/tests/unittests/core/test_strategy.py
+++ b/tests/unittests/core/test_strategy.py
@@ -52,6 +52,7 @@ strategies = [
 def test_handle_corrupted_trials(caplog, strategy, corrupted_trial):
     """Verify that corrupted trials are handled properly"""
     with caplog.at_level(logging.WARNING, logger="orion.core.worker.strategy"):
+        Strategy(strategy).observe([corrupted_trial], [{"objective": 1}])
         lie = Strategy(strategy).lie(corrupted_trial)
 
     match = "Trial `{}` has an objective but status is not completed".format(
@@ -64,9 +65,10 @@ def test_handle_corrupted_trials(caplog, strategy, corrupted_trial):
 
 
 @pytest.mark.parametrize("strategy", strategies)
-def test_handle_uncorrupted_trials(caplog, strategy, incomplete_trial):
+def test_handle_uncompleted_trials(caplog, strategy, incomplete_trial):
     """Verify that no warning is logged if trial is valid"""
     with caplog.at_level(logging.WARNING, logger="orion.core.worker.strategy"):
+        Strategy(strategy).observe([incomplete_trial], [{"objective": None}])
         Strategy(strategy).lie(incomplete_trial)
 
     assert "Trial `{}` has an objective but status is not completed" not in caplog.text
diff --git a/tests/unittests/core/worker/test_experiment.py b/tests/unittests/core/worker/test_experiment.py
index f44d652a..176a2126 100644
--- a/tests/unittests/core/worker/test_experiment.py
+++ b/tests/unittests/core/worker/test_experiment.py
@@ -428,6 +428,22 @@ def test_fetch_all_trials():
         assert trials == cfg.trials
 
 
+def test_fetch_pending_trials():
+    """Fetch a list of the trials that are pending
+
+    trials.status in ['new', 'interrupted', 'suspended']
+    """
+    pending_stati = ["new", "interrupted", "suspended"]
+    stati = pending_stati + ["completed", "broken", "reserved"]
+    with OrionState(trials=generate_trials(stati)) as cfg:
+        exp = Experiment("supernaekei", mode="x")
+        exp._id = cfg.trials[0]["experiment"]
+
+        trials = exp.fetch_pending_trials()
+        assert len(trials) == 3
+        assert set(trial.status for trial in trials) == set(pending_stati)
+
+
 def test_fetch_non_completed_trials():
     """Fetch a list of the trials that are not completed
 
@@ -571,6 +587,7 @@ def test_experiment_pickleable():
 read_only_methods = [
     "algorithms",
     "configuration",
+    "fetch_pending_trials",
     "fetch_noncompleted_trials",
     "fetch_trials",
     "fetch_trials_by_status",
@@ -598,6 +615,7 @@ read_write_only_methods = [
     "register_trial",
     "set_trial_status",
     "update_completed_trial",
+    "duplicate_pending_trials",
 ]
 execute_only_methods = [
     "reserve_trial",