From f4dc04dcdca4ea424cdf0d45684d7d877b3f4900 Mon Sep 17 00:00:00 2001 From: PhMueller Date: Wed, 28 Jul 2021 11:23:42 +0200 Subject: [PATCH 1/3] Continue a optimization run To do so, we store basically all important inforamtion (population, fitness values, bracket states, ..) as a pickle object. By reloading this object, we can continue training from this state. A thing to discuss is the current bracket state. I am not quite certain if there is another way to make sure that the brackets are correct. Also, this functionality is more about continuing and not warmstarting. The difference for me between warmstarting and restarting is that for warmstarting we should think about reloading multiple states and thus combining the populations per budget might not be so simple. However, restarting or continuing is more easy, because we just have to restore the old state. Happy to discuss this difference. Also, I have added a short logger that writes the results similar to Hpbandsters BOHB to file. (TODO: config.json is still missing). --- dehb/optimizers/dehb.py | 106 ++++++++++++++++++++++++++++++++++++---- 1 file changed, 97 insertions(+), 9 deletions(-) diff --git a/dehb/optimizers/dehb.py b/dehb/optimizers/dehb.py index e3d9651..5d4cc9d 100644 --- a/dehb/optimizers/dehb.py +++ b/dehb/optimizers/dehb.py @@ -8,9 +8,12 @@ from copy import deepcopy from loguru import logger from distributed import Client +import pickle +import json +from typing import Dict, Union +from pathlib import Path -from dehb.optimizers import DE, AsyncDE -from dehb.utils import SHBracketManager +from dehb import AsyncDE, SHBracketManager logger.configure(handlers=[{"sink": sys.stdout, "level": "INFO"}]) @@ -157,7 +160,10 @@ class DEHB(DEHBBase): def __init__(self, cs=None, f=None, dimensions=None, mutation_factor=0.5, crossover_prob=0.5, strategy='rand1_bin', min_budget=None, max_budget=None, eta=3, min_clip=None, max_clip=None, configspace=True, - boundary_fix_type='random', max_age=np.inf, n_workers=None, client=None, **kwargs): + boundary_fix_type='random', max_age=np.inf, n_workers=None, client=None, + client_resources: Union[Dict, None] = None, + checkpoint_file: Union[Path, None] = None, restore_checkpoint: bool = False, + **kwargs): super().__init__(cs=cs, f=f, dimensions=dimensions, mutation_factor=mutation_factor, crossover_prob=crossover_prob, strategy=strategy, min_budget=min_budget, max_budget=max_budget, eta=eta, min_clip=min_clip, max_clip=max_clip, @@ -198,6 +204,11 @@ def __init__(self, cs=None, f=None, dimensions=None, mutation_factor=0.5, self.gpu_usage = None self.single_node_with_gpus = None + self.client_resources = client_resources + self.checkpoint_file = checkpoint_file + if restore_checkpoint: + self.restore_checkpoint() + def __getstate__(self): """ Allows the object to picklable while having Dask client as a class attribute. """ @@ -563,9 +574,11 @@ def submit_job(self, job_info, **kwargs): if self.single_node_with_gpus: # managing GPU allocation for the job to be submitted job_info.update({"gpu_devices": self._get_gpu_id_with_low_load()}) - self.futures.append( - self.client.submit(self._f_objective, job_info) - ) + + if isinstance(self.client_resources, Dict) and len(self.client_resources) > 0: + self.futures.append(self.client.submit(self._f_objective, job_info, resources=self.client_resources)) + else: + self.futures.append(self.client.submit(self._f_objective, job_info)) else: # skipping scheduling to Dask worker to avoid added overheads in the synchronous case self.futures.append(self._f_objective(job_info)) @@ -624,10 +637,19 @@ def _fetch_results_from_workers(self): ) # book-keeping self._update_trackers( - traj=self.inc_score, runtime=cost, history=( - config.tolist(), float(fitness), float(cost), float(budget), info - ) + traj=self.inc_score, + runtime=cost, + history=(bracket_id, config.tolist(), float(fitness), float(cost), float(budget), info) ) + + new_result = json.dumps([ + self.iteration_counter, bracket_id, + {'configuration': config.tolist(), 'fidelity': float(budget), + 'fitness': float(fitness), 'cost': float(cost), 'info': info}], sort_keys=True + ) + with open(Path(self.output_path) / 'results.json', 'a+') as fh: + fh.write(new_result) + # remove processed future self.futures = np.delete(self.futures, [i for i, _ in done_list]).tolist() @@ -705,6 +727,72 @@ def _verbosity_runtime(self, fevals, brackets, total_cost): self.logger.info( "{}/{} {}".format(remaining[0], remaining[1], remaining[2]) ) + def save_checkpoint(self, checkpoint_file: Union[str, Path, None] = None): + + assert checkpoint_file is not None or self.checkpoint_file is not None + checkpoint_file = checkpoint_file or self.checkpoint_file + checkpoint_file = Path(checkpoint_file) + + checkpoint = {'de': {budget: [value.population, value.fitness, value.parent_counter] + for budget, value in self.de.items()}, + 'state': {'iteration_counter': self.iteration_counter, + 'inc_score': self.inc_score, + 'inc_config': self.inc_config, + 'inc_info': self.inc_info, + 'start_at': self.start, + 'save_at': time.time(), + }, + 'trajectory': self.traj, + 'history': self.history, + 'run_time': self.runtime, + } + + with checkpoint_file.open('wb') as fh: + pickle.dump(checkpoint, fh) + + def load_checkpoint(self, checkpoint_file: Path): + with checkpoint_file.open('rb') as fh: + checkpoint = pickle.load(fh) + return checkpoint + + def restore_checkpoint(self): + assert self.checkpoint_file is not None + checkpoint = self.load_checkpoint(self.checkpoint_file) + + for budget, values in checkpoint['de'].items(): + self.de[budget].population = values[0] + self.de[budget].fitness = values[1] + self.de[budget].parent_counter = values[2] + + self.inc_score = checkpoint['state']['inc_score'] + self.inc_config = checkpoint['state']['inc_config'] + self.inc_info = checkpoint['state']['inc_info'] + self.start = time.time() - (checkpoint['state']['save_at'] - checkpoint['state']['start_at']) + self.traj = checkpoint['trajectory'] + self.history = checkpoint['history'] + self.run_time = checkpoint['run_time'] + + already_seen_brackets = set() + for run in checkpoint['history']: + bracket_id, config, fitness, cost, budget, info = run + bracket = None + + if bracket_id not in already_seen_brackets: + bracket = self._start_new_bracket() + already_seen_brackets.add(bracket_id) + else: + for cand_bracket in self.active_brackets: + if cand_bracket.bracket_id == bracket_id: + bracket = cand_bracket + break + + assert bracket is not None and bracket_id == bracket.bracket_id + bracket.register_job(int(budget)) + bracket.complete_job(int(budget)) + + # TODO: Iteration counter should match the checkpoint iteration counter after + # "replaying" the brackets + assert self.iteration_counter == checkpoint['state']['iteration_counter'] @logger.catch def run(self, fevals=None, brackets=None, total_cost=None, single_node_with_gpus=False, From ea2651b6929707db78d3ee9423dbfec5c9d8f426 Mon Sep 17 00:00:00 2001 From: PhMueller Date: Wed, 28 Jul 2021 12:11:30 +0200 Subject: [PATCH 2/3] Add examples. --- dehb/__init__.py | 6 + examples/04_restart_an_experiment_local.py | 83 +++++++++++++ .../05_restart_an_experiment_with_dask.py | 114 ++++++++++++++++++ 3 files changed, 203 insertions(+) create mode 100644 examples/04_restart_an_experiment_local.py create mode 100644 examples/05_restart_an_experiment_with_dask.py diff --git a/dehb/__init__.py b/dehb/__init__.py index d2f68de..c2c16a4 100644 --- a/dehb/__init__.py +++ b/dehb/__init__.py @@ -1,3 +1,9 @@ +from loguru import logger +import sys + +logger.configure(handlers=[{"sink": sys.stdout, "level": "INFO"}]) + + from .optimizers import DE, AsyncDE from .optimizers import DEHB from .utils import SHBracketManager diff --git a/examples/04_restart_an_experiment_local.py b/examples/04_restart_an_experiment_local.py new file mode 100644 index 0000000..235d049 --- /dev/null +++ b/examples/04_restart_an_experiment_local.py @@ -0,0 +1,83 @@ +import time +import logging +logging.basicConfig(level=logging.INFO) + +from pathlib import Path +from dehb.optimizers.dehb_with_warmstart import DEHB + + +from hpobench.container.benchmarks.surrogates.paramnet_benchmark import ParamNetPokerOnTimeBenchmark + + +def objective_function(config, budget, **kwargs): + + start = time.time() + socket_id = kwargs.get('socket_id') + benchmark = ParamNetPokerOnTimeBenchmark(socket_id=socket_id) + result_dict = benchmark.objective_function(configuration=config, fidelity={'budget': int(budget)}) + finish = time.time() + return {'fitness': result_dict['function_value'], 'cost': result_dict['cost'], + 'info': {'res_info': result_dict['info'], 'time': float(finish - start)}} + + +def main(result_path: str, seed=0): + + result_path = Path(result_path) + result_path.mkdir(exist_ok=True, parents=True) + + checkpoint_file = result_path / 'checkpoint.pkl' + + benchmark = ParamNetPokerOnTimeBenchmark() + + dehb = DEHB(f=objective_function, + cs=benchmark.get_configuration_space(seed=seed), + dimensions=len(benchmark.get_configuration_space().get_hyperparameters()), + min_budget=81, # Those are the budgets used by the benchmark. + max_budget=2187, + eta=3, + output_path=result_path / 'dehb_logs', + n_workers=1) + + try: + traj, runtime, history = dehb.run(total_cost=20, # Let the procedure run for 20 seconds. + verbose=True, + save_intermediate=True, + # arguments below are part of **kwargs shared across workers + eta=3, + result_path=result_path, + socket_id=benchmark.socket_id) + except Exception: + # One could this a try-except to save the intermediate results in case of an error. + dehb.save_checkpoint(checkpoint_file) + + # Call this function to save the checkpoint to disk. + dehb.save_checkpoint(checkpoint_file) + + # To restart now the optimization procedure, create a new object. + # Set the parameters `checkpoint_file` and `restore_checkpooint`. Then, the checkpoint will be automatically loaded. + dehb_2 = DEHB(f=objective_function, + cs=benchmark.get_configuration_space(seed=seed), + dimensions=len(benchmark.get_configuration_space().get_hyperparameters()), + min_budget=81, + max_budget=2187, + eta=3, + output_path=result_path / 'dehb_logs', + n_workers=1, + checkpoint_file=result_path / 'checkpoint.pkl', + restore_checkpoint=True) + + # NOTE: Make sure to increase the time limit!! + traj2, runtime2, history2 = dehb_2.run(total_cost=20 + 20, + verbose=True, + save_intermediate=True, + # This parameter is needed for the HPOBench-Benchmark Object. + socket_id=benchmark.socket_id) + + +if __name__ == '__main__': + import argparse + parser = argparse.ArgumentParser() + parser.add_argument('--result_path', type=str) + args = parser.parse_args() + + main(result_path=args.result_path, seed=0) \ No newline at end of file diff --git a/examples/05_restart_an_experiment_with_dask.py b/examples/05_restart_an_experiment_with_dask.py new file mode 100644 index 0000000..41215e3 --- /dev/null +++ b/examples/05_restart_an_experiment_with_dask.py @@ -0,0 +1,114 @@ +""" +Before calling this event start the scheduler with + +`dask-scheduler --scheduler-file /scheduler_file.txt` + +and then the worker with + +`dask-worker --scheduler-file /scheduler_file.txt` --name 1 --resources "limit_proc=1" --no-nanny + +Note that we give the worker a resource. By doing so, only a single task can be executed per worker. + +""" + +import time +import logging +logging.basicConfig(level=logging.INFO) + +from distributed import Client +from pathlib import Path +from dehb.optimizers.dehb_with_warmstart import DEHB + +from hpobench.container.benchmarks.surrogates.paramnet_benchmark import ParamNetPokerOnTimeBenchmark + + +def objective_function(config, budget, **kwargs): + + start = time.time() + socket_id = kwargs.get('socket_id') + benchmark = ParamNetPokerOnTimeBenchmark(socket_id=socket_id) + result_dict = benchmark.objective_function(configuration=config, fidelity={'budget': int(budget)}) + finish = time.time() + return {'fitness': result_dict['function_value'], 'cost': result_dict['cost'], + 'info': {'res_info': result_dict['info'], 'time': float(finish - start)}} + + +def main(result_path: str, seed=0): + + result_path = Path(result_path) + result_path.mkdir(exist_ok=True, parents=True) + + checkpoint_file = result_path / 'checkpoint.pkl' + scheduler_file = result_path / 'scheduler_file.txt' + + benchmark = ParamNetPokerOnTimeBenchmark() + + client = Client(scheduler_file=scheduler_file) + + dehb = DEHB(f=objective_function, + cs=benchmark.get_configuration_space(seed=seed), + dimensions=len(benchmark.get_configuration_space().get_hyperparameters()), + min_budget=81, # Those are the budgets used by the benchmark. + max_budget=2187, + eta=3, + output_path=result_path / 'dehb_logs', + client=client, + # Limit the tasks per worker by starting the worker with the same resource! See above. + client_resources={'limit_proc': 1}, + ) + + try: + traj, runtime, history = dehb.run(total_cost=20, # Let the procedure run for 20 seconds. + verbose=True, + save_intermediate=True, + # arguments below are part of **kwargs shared across workers + eta=3, + result_path=result_path, + socket_id=benchmark.socket_id) + except Exception: + # One could this a try-except to save the intermediate results in case of an error. + dehb.save_checkpoint(checkpoint_file) + + # Call this function to save the checkpoint to disk. + dehb.save_checkpoint(checkpoint_file) + + dehb_2 = DEHB(f=objective_function, + cs=benchmark.get_configuration_space(seed=seed), + dimensions=len(benchmark.get_configuration_space().get_hyperparameters()), + min_budget=81, + max_budget=2187, + eta=3, + output_path=result_path / 'dehb_logs', + client=client, + client_resources={'limit_proc': 1}, # Also the new object needs this limit. + checkpoint_file=result_path / 'checkpoint.pkl', + restore_checkpoint=True, + ) + + traj2, runtime2, history2 = dehb_2.run(total_cost=20 + 20, + verbose=True, + save_intermediate=True, + # arguments below are part of **kwargs shared across workers + socket_id=benchmark.socket_id) + + + from matplotlib import pyplot as plt + import numpy as np + + f = plt.figure() + plt.plot(np.arange(len(traj2)), traj2, label='restarted') + plt.plot(np.arange(len(traj)), traj, label='first run') + plt.yscale('log') + plt.legend() + plt.savefig(result_path / 'traj.png') + plt.close() + + +if __name__ == '__main__': + + import argparse + parser = argparse.ArgumentParser() + parser.add_argument('--result_path', type=str) + args = parser.parse_args() + + main(result_path=args.result_path, seed=0) \ No newline at end of file From fc5fd98509b872358c752095341fed55671d7f6e Mon Sep 17 00:00:00 2001 From: PhMueller Date: Wed, 28 Jul 2021 14:43:56 +0200 Subject: [PATCH 3/3] Make dehb installable via `pip install .` --- dehb/__version__.py | 1 + setup.py | 48 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 49 insertions(+) create mode 100644 dehb/__version__.py create mode 100644 setup.py diff --git a/dehb/__version__.py b/dehb/__version__.py new file mode 100644 index 0000000..b8023d8 --- /dev/null +++ b/dehb/__version__.py @@ -0,0 +1 @@ +__version__ = '0.0.1' diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..21310f3 --- /dev/null +++ b/setup.py @@ -0,0 +1,48 @@ +# -*- encoding: utf-8 -*- +import setuptools + + +def read_file(file_name): + with open(file_name, encoding='utf-8') as fh: + text = fh.read() + return text + + +setuptools.setup( + name='dehb', + author_email='{awad, mallik, fh}@cs.uni-freiburg.de', + description='Evolutionary Hyberband for Scalable, Robust and Efficient ' + 'Hyperparameter Optimization', + long_description=read_file('README.md'), + long_description_content_type='text/markdown', + license='Apache-2.0', + url='https://www.automl.org/automl/', + project_urls={ + 'Documentation': 'https://github.com/automl/dehb', + 'Source Code': 'https://github.com/automl/dehb' + }, + version=read_file('dehb/__version__.py').split()[-1].strip('\''), + packages=setuptools.find_packages(exclude=['*.tests', '*.tests.*', + 'tests.*', 'tests'],), + python_requires='>3.5, <=3.9', + install_requires=read_file('./requirements.txt').split('\n'), + # extras_require=get_extra_requirements(), # We dont need this currently. + test_suite='pytest', + platforms=['Linux'], + classifiers=[ + 'Programming Language :: Python :: 3.5', + 'Programming Language :: Python :: 3.6', + 'Programming Language :: Python :: 3.7', + 'Development Status :: 3 - Alpha', + 'Natural Language :: English', + 'Environment :: Console', + 'Intended Audience :: Developers', + 'Intended Audience :: Education', + 'Intended Audience :: Science/Research', + 'License :: OSI Approved :: Apache Software License', + 'Operating System :: POSIX :: Linux', + 'Topic :: Scientific/Engineering :: Artificial Intelligence', + 'Topic :: Scientific/Engineering', + 'Topic :: Software Development', + ] +)