mirror of
https://github.com/wassname/ray.git
synced 2026-07-19 11:27:32 +08:00
[tune] Refactor search algorithms (#7037)
* start refactoring of search algorithms * format * needs tests * fix * suggestions * Fix PBT * lint * refactoring * hyperopt_working * dragonfly * hyperopt * change_half_of_algs * save * code-removed * remove_lots_of_unneccessary * changes * formatting * suggest * reset * rm * tests * search-change * exception * refactor-doc * search * py * moredocs * Update doc/source/tune-searchalg.rst * concurrency * max * tune * betterwarning * bohb * tests * test-change Co-authored-by: ujvl <misraujval@gmail.com>
This commit is contained in:
Richard Liaw
@@ -15,7 +15,7 @@ from ray.tune.trial_runner import TrialRunner
|
||||
from ray.tune.resources import Resources, json_to_resources, resources_to_json
|
||||
from ray.tune.suggest.repeater import Repeater
|
||||
from ray.tune.suggest.suggestion import (_MockSuggestionAlgorithm,
|
||||
SuggestionAlgorithm)
|
||||
SearchGenerator, Searcher)
|
||||
|
||||
|
||||
class TrialRunnerTest3(unittest.TestCase):
|
||||
@@ -30,11 +30,11 @@ class TrialRunnerTest3(unittest.TestCase):
|
||||
def on_step_begin(self, trialrunner):
|
||||
self._update_avail_resources()
|
||||
cnt = self.pre_step if hasattr(self, "pre_step") else 0
|
||||
setattr(self, "pre_step", cnt + 1)
|
||||
self.pre_step = cnt + 1
|
||||
|
||||
def on_step_end(self, trialrunner):
|
||||
cnt = self.pre_step if hasattr(self, "post_step") else 0
|
||||
setattr(self, "post_step", 1 + cnt)
|
||||
self.post_step = 1 + cnt
|
||||
|
||||
import types
|
||||
runner.trial_executor.on_step_begin = types.MethodType(
|
||||
@@ -101,9 +101,10 @@ class TrialRunnerTest3(unittest.TestCase):
|
||||
ray.init(num_cpus=4, num_gpus=2)
|
||||
experiment_spec = {"run": "__fake", "stop": {"training_iteration": 2}}
|
||||
experiments = [Experiment.from_json("test", experiment_spec)]
|
||||
searcher = _MockSuggestionAlgorithm(max_concurrent=10)
|
||||
searcher.add_configurations(experiments)
|
||||
runner = TrialRunner(search_alg=searcher)
|
||||
search_alg = _MockSuggestionAlgorithm()
|
||||
searcher = search_alg.searcher
|
||||
search_alg.add_configurations(experiments)
|
||||
runner = TrialRunner(search_alg=search_alg)
|
||||
runner.step()
|
||||
trials = runner.get_trials()
|
||||
self.assertEqual(trials[0].status, Trial.RUNNING)
|
||||
@@ -122,7 +123,7 @@ class TrialRunnerTest3(unittest.TestCase):
|
||||
ray.init(num_cpus=4, num_gpus=2)
|
||||
experiment_spec = {"run": "__fake", "stop": {"training_iteration": 1}}
|
||||
experiments = [Experiment.from_json("test", experiment_spec)]
|
||||
searcher = _MockSuggestionAlgorithm(max_concurrent=10)
|
||||
searcher = _MockSuggestionAlgorithm()
|
||||
searcher.add_configurations(experiments)
|
||||
runner = TrialRunner(search_alg=searcher)
|
||||
runner.step()
|
||||
@@ -147,7 +148,7 @@ class TrialRunnerTest3(unittest.TestCase):
|
||||
ray.init(num_cpus=4, num_gpus=2)
|
||||
experiment_spec = {"run": "__fake", "stop": {"training_iteration": 2}}
|
||||
experiments = [Experiment.from_json("test", experiment_spec)]
|
||||
searcher = _MockSuggestionAlgorithm(max_concurrent=10)
|
||||
searcher = _MockSuggestionAlgorithm()
|
||||
searcher.add_configurations(experiments)
|
||||
runner = TrialRunner(search_alg=searcher, scheduler=_MockScheduler())
|
||||
runner.step()
|
||||
@@ -162,30 +163,6 @@ class TrialRunnerTest3(unittest.TestCase):
|
||||
self.assertTrue(searcher.is_finished())
|
||||
self.assertTrue(runner.is_finished())
|
||||
|
||||
def testSearchAlgSchedulerEarlyStop(self):
|
||||
"""Early termination notif to Searcher can be turned off."""
|
||||
|
||||
class _MockScheduler(FIFOScheduler):
|
||||
def on_trial_result(self, *args, **kwargs):
|
||||
return TrialScheduler.STOP
|
||||
|
||||
ray.init(num_cpus=4, num_gpus=2)
|
||||
experiment_spec = {"run": "__fake", "stop": {"training_iteration": 2}}
|
||||
experiments = [Experiment.from_json("test", experiment_spec)]
|
||||
searcher = _MockSuggestionAlgorithm(use_early_stopped_trials=True)
|
||||
searcher.add_configurations(experiments)
|
||||
runner = TrialRunner(search_alg=searcher, scheduler=_MockScheduler())
|
||||
runner.step()
|
||||
runner.step()
|
||||
self.assertEqual(len(searcher.final_results), 1)
|
||||
|
||||
searcher = _MockSuggestionAlgorithm(use_early_stopped_trials=False)
|
||||
searcher.add_configurations(experiments)
|
||||
runner = TrialRunner(search_alg=searcher, scheduler=_MockScheduler())
|
||||
runner.step()
|
||||
runner.step()
|
||||
self.assertEqual(len(searcher.final_results), 0)
|
||||
|
||||
def testSearchAlgStalled(self):
|
||||
"""Checks that runner and searcher state is maintained when stalled."""
|
||||
ray.init(num_cpus=4, num_gpus=2)
|
||||
@@ -197,9 +174,10 @@ class TrialRunnerTest3(unittest.TestCase):
|
||||
}
|
||||
}
|
||||
experiments = [Experiment.from_json("test", experiment_spec)]
|
||||
searcher = _MockSuggestionAlgorithm(max_concurrent=1)
|
||||
searcher.add_configurations(experiments)
|
||||
runner = TrialRunner(search_alg=searcher)
|
||||
search_alg = _MockSuggestionAlgorithm(max_concurrent=1)
|
||||
search_alg.add_configurations(experiments)
|
||||
searcher = search_alg.searcher
|
||||
runner = TrialRunner(search_alg=search_alg)
|
||||
runner.step()
|
||||
trials = runner.get_trials()
|
||||
self.assertEqual(trials[0].status, Trial.RUNNING)
|
||||
@@ -219,7 +197,7 @@ class TrialRunnerTest3(unittest.TestCase):
|
||||
self.assertEqual(len(searcher.live_trials), 0)
|
||||
|
||||
self.assertTrue(all(trial.is_finished() for trial in trials))
|
||||
self.assertFalse(searcher.is_finished())
|
||||
self.assertFalse(search_alg.is_finished())
|
||||
self.assertFalse(runner.is_finished())
|
||||
|
||||
searcher.stall = False
|
||||
@@ -232,25 +210,27 @@ class TrialRunnerTest3(unittest.TestCase):
|
||||
runner.step()
|
||||
self.assertEqual(trials[2].status, Trial.TERMINATED)
|
||||
self.assertEqual(len(searcher.live_trials), 0)
|
||||
self.assertTrue(searcher.is_finished())
|
||||
self.assertTrue(search_alg.is_finished())
|
||||
self.assertTrue(runner.is_finished())
|
||||
|
||||
def testSearchAlgFinishes(self):
|
||||
"""Empty SearchAlg changing state in `next_trials` does not crash."""
|
||||
|
||||
class FinishFastAlg(SuggestionAlgorithm):
|
||||
class FinishFastAlg(_MockSuggestionAlgorithm):
|
||||
_index = 0
|
||||
|
||||
def next_trials(self):
|
||||
spec = self._experiment.spec
|
||||
trials = []
|
||||
if self._index < spec["num_samples"]:
|
||||
trial = Trial(
|
||||
spec.get("run"), stopping_criterion=spec.get("stop"))
|
||||
trials.append(trial)
|
||||
self._index += 1
|
||||
|
||||
for trial in self._trial_generator:
|
||||
trials += [trial]
|
||||
break
|
||||
|
||||
if self._index > 4:
|
||||
self.set_finished()
|
||||
|
||||
return trials
|
||||
|
||||
def suggest(self, trial_id):
|
||||
@@ -406,7 +386,7 @@ class TrialRunnerTest3(unittest.TestCase):
|
||||
tmpdir = tempfile.mkdtemp()
|
||||
runner = TrialRunner(local_checkpoint_dir=tmpdir, checkpoint_period=0)
|
||||
runner.add_trial(trial)
|
||||
for i in range(5):
|
||||
for _ in range(5):
|
||||
runner.step()
|
||||
# force checkpoint
|
||||
runner.checkpoint()
|
||||
@@ -427,14 +407,14 @@ class TrialRunnerTest3(unittest.TestCase):
|
||||
tmpdir = tempfile.mkdtemp()
|
||||
runner = TrialRunner(local_checkpoint_dir=tmpdir, checkpoint_period=0)
|
||||
runner.add_trial(trial)
|
||||
for i in range(5):
|
||||
for _ in range(5):
|
||||
runner.step()
|
||||
# force checkpoint
|
||||
runner.checkpoint()
|
||||
self.assertEquals(count_checkpoints(tmpdir), 1)
|
||||
|
||||
runner2 = TrialRunner(resume="LOCAL", local_checkpoint_dir=tmpdir)
|
||||
for i in range(5):
|
||||
for _ in range(5):
|
||||
runner2.step()
|
||||
self.assertEquals(count_checkpoints(tmpdir), 2)
|
||||
|
||||
@@ -473,50 +453,64 @@ class SearchAlgorithmTest(unittest.TestCase):
|
||||
_register_all()
|
||||
|
||||
def testNestedSuggestion(self):
|
||||
class TestSuggestion(SuggestionAlgorithm):
|
||||
class TestSuggestion(Searcher):
|
||||
def suggest(self, trial_id):
|
||||
return {"a": {"b": {"c": {"d": 4, "e": 5}}}}
|
||||
|
||||
alg = TestSuggestion()
|
||||
searcher = TestSuggestion()
|
||||
alg = SearchGenerator(searcher)
|
||||
alg.add_configurations({"test": {"run": "__fake"}})
|
||||
trial = alg.next_trials()[0]
|
||||
self.assertTrue("e=5" in trial.experiment_tag)
|
||||
self.assertTrue("d=4" in trial.experiment_tag)
|
||||
|
||||
def _test_repeater(self, repeat):
|
||||
def _test_repeater(self, num_samples, repeat):
|
||||
ray.init(num_cpus=4)
|
||||
|
||||
class TestSuggestion(SuggestionAlgorithm):
|
||||
count = 0
|
||||
class TestSuggestion(Searcher):
|
||||
index = 0
|
||||
|
||||
def suggest(self, trial_id):
|
||||
return {"test_variable": 5}
|
||||
self.index += 1
|
||||
return {"test_variable": 5 + self.index}
|
||||
|
||||
def on_trial_complete(self, *args, **kwargs):
|
||||
self.count += 1
|
||||
return
|
||||
|
||||
alg = TestSuggestion(metric="episode_reward_mean")
|
||||
repeat_alg = Repeater(alg, repeat=repeat, set_index=False)
|
||||
searcher = TestSuggestion(metric="episode_reward_mean")
|
||||
repeat_searcher = Repeater(searcher, repeat=repeat, set_index=False)
|
||||
alg = SearchGenerator(repeat_searcher)
|
||||
experiment_spec = {
|
||||
"run": "__fake",
|
||||
"num_samples": 1,
|
||||
"num_samples": num_samples,
|
||||
"stop": {
|
||||
"training_iteration": 1
|
||||
}
|
||||
}
|
||||
repeat_alg.add_configurations({"test": experiment_spec})
|
||||
runner = TrialRunner(search_alg=repeat_alg)
|
||||
for i in range(repeat * 2):
|
||||
alg.add_configurations({"test": experiment_spec})
|
||||
runner = TrialRunner(search_alg=alg)
|
||||
while not runner.is_finished():
|
||||
runner.step()
|
||||
|
||||
trials = runner.get_trials()
|
||||
self.assertEquals(len(trials), repeat)
|
||||
return runner.get_trials()
|
||||
|
||||
def testRepeat1(self):
|
||||
self._test_repeater(repeat=1)
|
||||
trials = self._test_repeater(num_samples=2, repeat=1)
|
||||
self.assertEquals(len(trials), 2)
|
||||
parameter_set = {t.evaluated_params["test_variable"] for t in trials}
|
||||
self.assertEquals(len(parameter_set), 2)
|
||||
|
||||
def testRepeat4(self):
|
||||
self._test_repeater(repeat=4)
|
||||
trials = self._test_repeater(num_samples=12, repeat=4)
|
||||
self.assertEquals(len(trials), 12)
|
||||
parameter_set = {t.evaluated_params["test_variable"] for t in trials}
|
||||
self.assertEquals(len(parameter_set), 3)
|
||||
|
||||
def testOddRepeat(self):
|
||||
trials = self._test_repeater(num_samples=11, repeat=5)
|
||||
self.assertEquals(len(trials), 11)
|
||||
parameter_set = {t.evaluated_params["test_variable"] for t in trials}
|
||||
self.assertEquals(len(parameter_set), 3)
|
||||
|
||||
|
||||
class ResourcesTest(unittest.TestCase):
|
||||
|
||||
@@ -643,6 +643,7 @@ class BOHBSuite(unittest.TestCase):
|
||||
sched = HyperBandForBOHB(max_t=3, reduction_factor=3)
|
||||
runner = _MockTrialRunner(sched)
|
||||
runner._search_alg = MagicMock()
|
||||
runner._search_alg.searcher = MagicMock()
|
||||
trials = [Trial("__fake") for i in range(3)]
|
||||
for t in trials:
|
||||
runner.add_trial(t)
|
||||
@@ -656,8 +657,8 @@ class BOHBSuite(unittest.TestCase):
|
||||
decision = sched.on_trial_result(runner, trials[-1], spy_result)
|
||||
self.assertEqual(decision, TrialScheduler.STOP)
|
||||
sched.choose_trial_to_run(runner)
|
||||
self.assertEqual(runner._search_alg.on_pause.call_count, 2)
|
||||
self.assertEqual(runner._search_alg.on_unpause.call_count, 1)
|
||||
self.assertEqual(runner._search_alg.searcher.on_pause.call_count, 2)
|
||||
self.assertEqual(runner._search_alg.searcher.on_unpause.call_count, 1)
|
||||
self.assertTrue("hyperband_info" in spy_result)
|
||||
self.assertEquals(spy_result["hyperband_info"]["budget"], 1)
|
||||
|
||||
@@ -668,6 +669,7 @@ class BOHBSuite(unittest.TestCase):
|
||||
sched = HyperBandForBOHB(max_t=3, reduction_factor=3, mode="min")
|
||||
runner = _MockTrialRunner(sched)
|
||||
runner._search_alg = MagicMock()
|
||||
runner._search_alg.searcher = MagicMock()
|
||||
trials = [Trial("__fake") for i in range(3)]
|
||||
for t in trials:
|
||||
runner.add_trial(t)
|
||||
@@ -681,7 +683,7 @@ class BOHBSuite(unittest.TestCase):
|
||||
decision = sched.on_trial_result(runner, trials[-1], spy_result)
|
||||
self.assertEqual(decision, TrialScheduler.CONTINUE)
|
||||
sched.choose_trial_to_run(runner)
|
||||
self.assertEqual(runner._search_alg.on_pause.call_count, 2)
|
||||
self.assertEqual(runner._search_alg.searcher.on_pause.call_count, 2)
|
||||
self.assertTrue("hyperband_info" in spy_result)
|
||||
self.assertEquals(spy_result["hyperband_info"]["budget"], 1)
|
||||
|
||||
|
||||
@@ -13,6 +13,7 @@ import ray
|
||||
from ray import tune
|
||||
from ray.test_utils import recursive_fnmatch
|
||||
from ray.rllib import _register_all
|
||||
from ray.tune.suggest import ConcurrencyLimiter
|
||||
from ray.tune.suggest.hyperopt import HyperOptSearch
|
||||
from ray.tune.suggest.bayesopt import BayesOptSearch
|
||||
from ray.tune.suggest.skopt import SkOptSearch
|
||||
@@ -132,7 +133,7 @@ class AutoInitTest(unittest.TestCase):
|
||||
|
||||
class AbstractWarmStartTest:
|
||||
def setUp(self):
|
||||
ray.init(local_mode=True)
|
||||
ray.init(num_cpus=1, local_mode=True)
|
||||
self.tmpdir = tempfile.mkdtemp()
|
||||
|
||||
def tearDown(self):
|
||||
@@ -146,20 +147,26 @@ class AbstractWarmStartTest:
|
||||
def run_exp_1(self):
|
||||
np.random.seed(162)
|
||||
search_alg, cost = self.set_basic_conf()
|
||||
results_exp_1 = tune.run(cost, num_samples=5, search_alg=search_alg)
|
||||
search_alg = ConcurrencyLimiter(search_alg, 1)
|
||||
results_exp_1 = tune.run(
|
||||
cost, num_samples=5, search_alg=search_alg, verbose=0)
|
||||
self.log_dir = os.path.join(self.tmpdir, "warmStartTest.pkl")
|
||||
search_alg.save(self.log_dir)
|
||||
return results_exp_1
|
||||
|
||||
def run_exp_2(self):
|
||||
search_alg2, cost = self.set_basic_conf()
|
||||
search_alg2 = ConcurrencyLimiter(search_alg2, 1)
|
||||
search_alg2.restore(self.log_dir)
|
||||
return tune.run(cost, num_samples=5, search_alg=search_alg2)
|
||||
return tune.run(cost, num_samples=5, search_alg=search_alg2, verbose=0)
|
||||
|
||||
def run_exp_3(self):
|
||||
print("FULL RUN")
|
||||
np.random.seed(162)
|
||||
search_alg3, cost = self.set_basic_conf()
|
||||
return tune.run(cost, num_samples=10, search_alg=search_alg3)
|
||||
search_alg3 = ConcurrencyLimiter(search_alg3, 1)
|
||||
return tune.run(
|
||||
cost, num_samples=10, search_alg=search_alg3, verbose=0)
|
||||
|
||||
def testWarmStart(self):
|
||||
results_exp_1 = self.run_exp_1()
|
||||
@@ -185,10 +192,10 @@ class HyperoptWarmStartTest(AbstractWarmStartTest, unittest.TestCase):
|
||||
|
||||
search_alg = HyperOptSearch(
|
||||
space,
|
||||
max_concurrent=1,
|
||||
metric="loss",
|
||||
mode="min",
|
||||
random_state_seed=5)
|
||||
random_state_seed=5,
|
||||
n_initial_points=1)
|
||||
return search_alg, cost
|
||||
|
||||
|
||||
@@ -201,7 +208,6 @@ class BayesoptWarmStartTest(AbstractWarmStartTest, unittest.TestCase):
|
||||
|
||||
search_alg = BayesOptSearch(
|
||||
space,
|
||||
max_concurrent=1,
|
||||
metric="loss",
|
||||
mode="min",
|
||||
utility_kwargs={
|
||||
@@ -223,7 +229,6 @@ class SkoptWarmStartTest(AbstractWarmStartTest, unittest.TestCase):
|
||||
|
||||
search_alg = SkOptSearch(
|
||||
optimizer, ["width", "height"],
|
||||
max_concurrent=1,
|
||||
metric="loss",
|
||||
mode="min",
|
||||
points_to_evaluate=previously_run_params,
|
||||
@@ -242,11 +247,7 @@ class NevergradWarmStartTest(AbstractWarmStartTest, unittest.TestCase):
|
||||
mean_loss=(space["height"] - 14)**2 - abs(space["width"] - 3))
|
||||
|
||||
search_alg = NevergradSearch(
|
||||
optimizer,
|
||||
parameter_names,
|
||||
max_concurrent=1,
|
||||
metric="mean_loss",
|
||||
mode="min")
|
||||
optimizer, parameter_names, metric="mean_loss", mode="min")
|
||||
return search_alg, cost
|
||||
|
||||
|
||||
@@ -305,7 +306,6 @@ class ZOOptWarmStartTest(AbstractWarmStartTest, unittest.TestCase):
|
||||
algo="Asracos", # only support ASRacos currently
|
||||
budget=200,
|
||||
dim_dict=dim_dict,
|
||||
max_concurrent=1,
|
||||
metric="loss",
|
||||
mode="min")
|
||||
|
||||
|
||||
@@ -8,9 +8,7 @@ from ray.rllib import _register_all
|
||||
|
||||
from ray import tune
|
||||
from ray.tune.result import DEFAULT_RESULTS_DIR
|
||||
from ray.tune.experiment import Experiment
|
||||
from ray.tune.suggest import grid_search, BasicVariantGenerator
|
||||
from ray.tune.suggest.suggestion import _MockSuggestionAlgorithm
|
||||
from ray.tune.suggest.variant_generator import (RecursiveDependencyError,
|
||||
resolve_nested_dict)
|
||||
|
||||
@@ -301,36 +299,7 @@ class VariantGeneratorTest(unittest.TestCase):
|
||||
except RecursiveDependencyError as e:
|
||||
assert "`foo` recursively depends on" in str(e), e
|
||||
else:
|
||||
assert False
|
||||
|
||||
def testMaxConcurrentSuggestions(self):
|
||||
"""Checks that next_trials() supports throttling."""
|
||||
experiment_spec = {
|
||||
"run": "PPO",
|
||||
"num_samples": 6,
|
||||
}
|
||||
experiments = [Experiment.from_json("test", experiment_spec)]
|
||||
|
||||
searcher = _MockSuggestionAlgorithm(max_concurrent=4)
|
||||
searcher.add_configurations(experiments)
|
||||
trials = searcher.next_trials()
|
||||
self.assertEqual(len(trials), 4)
|
||||
self.assertEqual(searcher.next_trials(), [])
|
||||
|
||||
finished_trial = trials.pop()
|
||||
searcher.on_trial_complete(finished_trial.trial_id)
|
||||
self.assertEqual(len(searcher.next_trials()), 1)
|
||||
|
||||
finished_trial = trials.pop()
|
||||
searcher.on_trial_complete(finished_trial.trial_id)
|
||||
|
||||
finished_trial = trials.pop()
|
||||
searcher.on_trial_complete(finished_trial.trial_id)
|
||||
|
||||
finished_trial = trials.pop()
|
||||
searcher.on_trial_complete(finished_trial.trial_id)
|
||||
self.assertEqual(len(searcher.next_trials()), 1)
|
||||
self.assertEqual(len(searcher.next_trials()), 0)
|
||||
raise
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -79,8 +79,7 @@ space = {
|
||||
"momentum": hp.uniform("momentum", 0.1, 0.9),
|
||||
}
|
||||
|
||||
hyperopt_search = HyperOptSearch(
|
||||
space, max_concurrent=2, reward_attr="mean_accuracy")
|
||||
hyperopt_search = HyperOptSearch(space, metric="mean_accuracy", mode="max")
|
||||
|
||||
analysis = tune.run(train_mnist, num_samples=10, search_alg=hyperopt_search)
|
||||
# __run_searchalg_end__
|
||||
|
||||
Reference in New Issue
Block a user