We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[tune] Missing keep_checkpoints_num arg in tune
Hi everybody,

Thanks a lot for this lib, I've been using for the past few weeks with TF2 and it works pretty well!

In my current use case, I'm checkpointing regularly to ensure a minimal loss in the face of OOM. I do control TF checkpoints manually to avoid filling my disk space but I still need to dump some ray metadata (iterations count, etc.).
I've noticed that ray create folders like `checkpoint_{iter}` to store those but from `tune` there is no way to tell him that I don't need to keep dozens of folder.
I also noticed that `experiment`, and `trial` classes have the perfect argument: `keep_checkpoints_num`.

Sadly this argument is missing from the `tune` config. 
Is there any reasons why it is missing from `tune` arguments?

Have a nice day.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `python/ray/tune/tune.py`
Content:
```
1 from __future__ import absolute_import
2 from __future__ import division
3 from __future__ import print_function
4 
5 import logging
6 import time
7 
8 from ray.tune.error import TuneError
9 from ray.tune.experiment import convert_to_experiment_list, Experiment
10 from ray.tune.analysis import ExperimentAnalysis
11 from ray.tune.suggest import BasicVariantGenerator
12 from ray.tune.trial import Trial, DEBUG_PRINT_INTERVAL
13 from ray.tune.ray_trial_executor import RayTrialExecutor
14 from ray.tune.syncer import wait_for_sync
15 from ray.tune.trial_runner import TrialRunner
16 from ray.tune.schedulers import (HyperBandScheduler, AsyncHyperBandScheduler,
17                                  FIFOScheduler, MedianStoppingRule)
18 from ray.tune.web_server import TuneServer
19 
20 logger = logging.getLogger(__name__)
21 
22 _SCHEDULERS = {
23     "FIFO": FIFOScheduler,
24     "MedianStopping": MedianStoppingRule,
25     "HyperBand": HyperBandScheduler,
26     "AsyncHyperBand": AsyncHyperBandScheduler,
27 }
28 
29 
30 def _make_scheduler(args):
31     if args.scheduler in _SCHEDULERS:
32         return _SCHEDULERS[args.scheduler](**args.scheduler_config)
33     else:
34         raise TuneError("Unknown scheduler: {}, should be one of {}".format(
35             args.scheduler, _SCHEDULERS.keys()))
36 
37 
38 def run(run_or_experiment,
39         name=None,
40         stop=None,
41         config=None,
42         resources_per_trial=None,
43         num_samples=1,
44         local_dir=None,
45         upload_dir=None,
46         trial_name_creator=None,
47         loggers=None,
48         sync_to_cloud=None,
49         sync_to_driver=None,
50         checkpoint_freq=0,
51         checkpoint_at_end=False,
52         global_checkpoint_period=10,
53         export_formats=None,
54         max_failures=3,
55         restore=None,
56         search_alg=None,
57         scheduler=None,
58         with_server=False,
59         server_port=TuneServer.DEFAULT_PORT,
60         verbose=2,
61         resume=False,
62         queue_trials=False,
63         reuse_actors=True,
64         trial_executor=None,
65         raise_on_failed_trial=True,
66         return_trials=True,
67         ray_auto_init=True,
68         sync_function=None):
69     """Executes training.
70 
71     Args:
72         run_or_experiment (function|class|str|Experiment): If
73             function|class|str, this is the algorithm or model to train.
74             This may refer to the name of a built-on algorithm
75             (e.g. RLLib's DQN or PPO), a user-defined trainable
76             function or class, or the string identifier of a
77             trainable function or class registered in the tune registry.
78             If Experiment, then Tune will execute training based on
79             Experiment.spec.
80         name (str): Name of experiment.
81         stop (dict): The stopping criteria. The keys may be any field in
82             the return result of 'train()', whichever is reached first.
83             Defaults to empty dict.
84         config (dict): Algorithm-specific configuration for Tune variant
85             generation (e.g. env, hyperparams). Defaults to empty dict.
86             Custom search algorithms may ignore this.
87         resources_per_trial (dict): Machine resources to allocate per trial,
88             e.g. ``{"cpu": 64, "gpu": 8}``. Note that GPUs will not be
89             assigned unless you specify them here. Defaults to 1 CPU and 0
90             GPUs in ``Trainable.default_resource_request()``.
91         num_samples (int): Number of times to sample from the
92             hyperparameter space. Defaults to 1. If `grid_search` is
93             provided as an argument, the grid will be repeated
94             `num_samples` of times.
95         local_dir (str): Local dir to save training results to.
96             Defaults to ``~/ray_results``.
97         upload_dir (str): Optional URI to sync training results
98             to (e.g. ``s3://bucket``).
99         trial_name_creator (func): Optional function for generating
100             the trial string representation.
101         loggers (list): List of logger creators to be used with
102             each Trial. If None, defaults to ray.tune.logger.DEFAULT_LOGGERS.
103             See `ray/tune/logger.py`.
104         sync_to_cloud (func|str): Function for syncing the local_dir to and
105             from upload_dir. If string, then it must be a string template
106             that includes `{source}` and `{target}` for the syncer to run.
107             If not provided, the sync command defaults to standard
108             S3 or gsutil sync comamnds.
109         sync_to_driver (func|str): Function for syncing trial logdir from
110             remote node to local. If string, then it must be a string template
111             that includes `{source}` and `{target}` for the syncer to run.
112             If not provided, defaults to using rsync.
113         checkpoint_freq (int): How many training iterations between
114             checkpoints. A value of 0 (default) disables checkpointing.
115         checkpoint_at_end (bool): Whether to checkpoint at the end of the
116             experiment regardless of the checkpoint_freq. Default is False.
117         global_checkpoint_period (int): Seconds between global checkpointing.
118             This does not affect `checkpoint_freq`, which specifies frequency
119             for individual trials.
120         export_formats (list): List of formats that exported at the end of
121             the experiment. Default is None.
122         max_failures (int): Try to recover a trial from its last
123             checkpoint at least this many times. Only applies if
124             checkpointing is enabled. Setting to -1 will lead to infinite
125             recovery retries. Defaults to 3.
126         restore (str): Path to checkpoint. Only makes sense to set if
127             running 1 trial. Defaults to None.
128         search_alg (SearchAlgorithm): Search Algorithm. Defaults to
129             BasicVariantGenerator.
130         scheduler (TrialScheduler): Scheduler for executing
131             the experiment. Choose among FIFO (default), MedianStopping,
132             AsyncHyperBand, and HyperBand.
133         with_server (bool): Starts a background Tune server. Needed for
134             using the Client API.
135         server_port (int): Port number for launching TuneServer.
136         verbose (int): 0, 1, or 2. Verbosity mode. 0 = silent,
137             1 = only status updates, 2 = status and trial results.
138         resume (str|bool): One of "LOCAL", "REMOTE", "PROMPT", or bool.
139             LOCAL/True restores the checkpoint from the local_checkpoint_dir.
140             REMOTE restores the checkpoint from remote_checkpoint_dir.
141             PROMPT provides CLI feedback. False forces a new
142             experiment. If resume is set but checkpoint does not exist,
143             ValueError will be thrown.
144         queue_trials (bool): Whether to queue trials when the cluster does
145             not currently have enough resources to launch one. This should
146             be set to True when running on an autoscaling cluster to enable
147             automatic scale-up.
148         reuse_actors (bool): Whether to reuse actors between different trials
149             when possible. This can drastically speed up experiments that start
150             and stop actors often (e.g., PBT in time-multiplexing mode). This
151             requires trials to have the same resource requirements.
152         trial_executor (TrialExecutor): Manage the execution of trials.
153         raise_on_failed_trial (bool): Raise TuneError if there exists failed
154             trial (of ERROR state) when the experiments complete.
155         ray_auto_init (bool): Automatically starts a local Ray cluster
156             if using a RayTrialExecutor (which is the default) and
157             if Ray is not initialized. Defaults to True.
158         sync_function: Deprecated. See `sync_to_cloud` and
159             `sync_to_driver`.
160 
161     Returns:
162         List of Trial objects.
163 
164     Raises:
165         TuneError if any trials failed and `raise_on_failed_trial` is True.
166 
167     Examples:
168         >>> tune.run(mytrainable, scheduler=PopulationBasedTraining())
169 
170         >>> tune.run(mytrainable, num_samples=5, reuse_actors=True)
171 
172         >>> tune.run(
173                 "PG",
174                 num_samples=5,
175                 config={
176                     "env": "CartPole-v0",
177                     "lr": tune.sample_from(lambda _: np.random.rand())
178                 }
179             )
180     """
181     trial_executor = trial_executor or RayTrialExecutor(
182         queue_trials=queue_trials,
183         reuse_actors=reuse_actors,
184         ray_auto_init=ray_auto_init)
185     experiment = run_or_experiment
186     if not isinstance(run_or_experiment, Experiment):
187         run_identifier = Experiment._register_if_needed(run_or_experiment)
188         experiment = Experiment(
189             name=name,
190             run=run_identifier,
191             stop=stop,
192             config=config,
193             resources_per_trial=resources_per_trial,
194             num_samples=num_samples,
195             local_dir=local_dir,
196             upload_dir=upload_dir,
197             sync_to_driver=sync_to_driver,
198             trial_name_creator=trial_name_creator,
199             loggers=loggers,
200             checkpoint_freq=checkpoint_freq,
201             checkpoint_at_end=checkpoint_at_end,
202             export_formats=export_formats,
203             max_failures=max_failures,
204             restore=restore,
205             sync_function=sync_function)
206     else:
207         logger.debug("Ignoring some parameters passed into tune.run.")
208 
209     if sync_to_cloud:
210         assert experiment.remote_checkpoint_dir, (
211             "Need `upload_dir` if `sync_to_cloud` given.")
212 
213     runner = TrialRunner(
214         search_alg=search_alg or BasicVariantGenerator(),
215         scheduler=scheduler or FIFOScheduler(),
216         local_checkpoint_dir=experiment.checkpoint_dir,
217         remote_checkpoint_dir=experiment.remote_checkpoint_dir,
218         sync_to_cloud=sync_to_cloud,
219         checkpoint_period=global_checkpoint_period,
220         resume=resume,
221         launch_web_server=with_server,
222         server_port=server_port,
223         verbose=bool(verbose > 1),
224         trial_executor=trial_executor)
225 
226     runner.add_experiment(experiment)
227 
228     if verbose:
229         print(runner.debug_string(max_debug=99999))
230 
231     last_debug = 0
232     while not runner.is_finished():
233         runner.step()
234         if time.time() - last_debug > DEBUG_PRINT_INTERVAL:
235             if verbose:
236                 print(runner.debug_string())
237             last_debug = time.time()
238 
239     if verbose:
240         print(runner.debug_string(max_debug=99999))
241 
242     wait_for_sync()
243 
244     errored_trials = []
245     for trial in runner.get_trials():
246         if trial.status != Trial.TERMINATED:
247             errored_trials += [trial]
248 
249     if errored_trials:
250         if raise_on_failed_trial:
251             raise TuneError("Trials did not complete", errored_trials)
252         else:
253             logger.error("Trials did not complete: %s", errored_trials)
254 
255     if return_trials:
256         return runner.get_trials()
257     return ExperimentAnalysis(experiment.checkpoint_dir)
258 
259 
260 def run_experiments(experiments,
261                     search_alg=None,
262                     scheduler=None,
263                     with_server=False,
264                     server_port=TuneServer.DEFAULT_PORT,
265                     verbose=2,
266                     resume=False,
267                     queue_trials=False,
268                     reuse_actors=False,
269                     trial_executor=None,
270                     raise_on_failed_trial=True):
271     """Runs and blocks until all trials finish.
272 
273     Examples:
274         >>> experiment_spec = Experiment("experiment", my_func)
275         >>> run_experiments(experiments=experiment_spec)
276 
277         >>> experiment_spec = {"experiment": {"run": my_func}}
278         >>> run_experiments(experiments=experiment_spec)
279 
280         >>> run_experiments(
281         >>>     experiments=experiment_spec,
282         >>>     scheduler=MedianStoppingRule(...))
283 
284         >>> run_experiments(
285         >>>     experiments=experiment_spec,
286         >>>     search_alg=SearchAlgorithm(),
287         >>>     scheduler=MedianStoppingRule(...))
288 
289     Returns:
290         List of Trial objects, holding data for each executed trial.
291 
292     """
293     # This is important to do this here
294     # because it schematize the experiments
295     # and it conducts the implicit registration.
296     experiments = convert_to_experiment_list(experiments)
297 
298     trials = []
299     for exp in experiments:
300         trials += run(
301             exp,
302             search_alg=search_alg,
303             scheduler=scheduler,
304             with_server=with_server,
305             server_port=server_port,
306             verbose=verbose,
307             resume=resume,
308             queue_trials=queue_trials,
309             reuse_actors=reuse_actors,
310             trial_executor=trial_executor,
311             raise_on_failed_trial=raise_on_failed_trial)
312     return trials
313
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/python/ray/tune/tune.py b/python/ray/tune/tune.py
--- a/python/ray/tune/tune.py
+++ b/python/ray/tune/tune.py
@@ -49,6 +49,8 @@
         sync_to_driver=None,
         checkpoint_freq=0,
         checkpoint_at_end=False,
+        keep_checkpoints_num=None,
+        checkpoint_score_attr=None,
         global_checkpoint_period=10,
         export_formats=None,
         max_failures=3,
@@ -114,6 +116,13 @@
             checkpoints. A value of 0 (default) disables checkpointing.
         checkpoint_at_end (bool): Whether to checkpoint at the end of the
             experiment regardless of the checkpoint_freq. Default is False.
+        keep_checkpoints_num (int): Number of checkpoints to keep. A value of
+            `None` keeps all checkpoints. Defaults to `None`. If set, need
+            to provide `checkpoint_score_attr`.
+        checkpoint_score_attr (str): Specifies by which attribute to rank the
+            best checkpoint. Default is increasing order. If attribute starts
+            with `min-` it will rank attribute in decreasing order, i.e.
+            `min-validation_loss`.
         global_checkpoint_period (int): Seconds between global checkpointing.
             This does not affect `checkpoint_freq`, which specifies frequency
             for individual trials.
@@ -199,6 +208,8 @@
             loggers=loggers,
             checkpoint_freq=checkpoint_freq,
             checkpoint_at_end=checkpoint_at_end,
+            keep_checkpoints_num=keep_checkpoints_num,
+            checkpoint_score_attr=checkpoint_score_attr,
             export_formats=export_formats,
             max_failures=max_failures,
             restore=restore,

{"golden_diff": "diff --git a/python/ray/tune/tune.py b/python/ray/tune/tune.py\n--- a/python/ray/tune/tune.py\n+++ b/python/ray/tune/tune.py\n@@ -49,6 +49,8 @@\n         sync_to_driver=None,\n         checkpoint_freq=0,\n         checkpoint_at_end=False,\n+        keep_checkpoints_num=None,\n+        checkpoint_score_attr=None,\n         global_checkpoint_period=10,\n         export_formats=None,\n         max_failures=3,\n@@ -114,6 +116,13 @@\n             checkpoints. A value of 0 (default) disables checkpointing.\n         checkpoint_at_end (bool): Whether to checkpoint at the end of the\n             experiment regardless of the checkpoint_freq. Default is False.\n+        keep_checkpoints_num (int): Number of checkpoints to keep. A value of\n+            `None` keeps all checkpoints. Defaults to `None`. If set, need\n+            to provide `checkpoint_score_attr`.\n+        checkpoint_score_attr (str): Specifies by which attribute to rank the\n+            best checkpoint. Default is increasing order. If attribute starts\n+            with `min-` it will rank attribute in decreasing order, i.e.\n+            `min-validation_loss`.\n         global_checkpoint_period (int): Seconds between global checkpointing.\n             This does not affect `checkpoint_freq`, which specifies frequency\n             for individual trials.\n@@ -199,6 +208,8 @@\n             loggers=loggers,\n             checkpoint_freq=checkpoint_freq,\n             checkpoint_at_end=checkpoint_at_end,\n+            keep_checkpoints_num=keep_checkpoints_num,\n+            checkpoint_score_attr=checkpoint_score_attr,\n             export_formats=export_formats,\n             max_failures=max_failures,\n             restore=restore,\n", "issue": "[tune] Missing keep_checkpoints_num arg in tune\nHi everybody,\r\n\r\nThanks a lot for this lib, I've been using for the past few weeks with TF2 and it works pretty well!\r\n\r\nIn my current use case, I'm checkpointing regularly to ensure a minimal loss in the face of OOM. I do control TF checkpoints manually to avoid filling my disk space but I still need to dump some ray metadata (iterations count, etc.).\r\nI've noticed that ray create folders like `checkpoint_{iter}` to store those but from `tune` there is no way to tell him that I don't need to keep dozens of folder.\r\nI also noticed that `experiment`, and `trial` classes have the perfect argument: `keep_checkpoints_num`.\r\n\r\nSadly this argument is missing from the `tune` config. \r\nIs there any reasons why it is missing from `tune` arguments?\r\n\r\nHave a nice day.\n", "before_files": [{"content": "from __future__ import absolute_import\nfrom __future__ import division\nfrom __future__ import print_function\n\nimport logging\nimport time\n\nfrom ray.tune.error import TuneError\nfrom ray.tune.experiment import convert_to_experiment_list, Experiment\nfrom ray.tune.analysis import ExperimentAnalysis\nfrom ray.tune.suggest import BasicVariantGenerator\nfrom ray.tune.trial import Trial, DEBUG_PRINT_INTERVAL\nfrom ray.tune.ray_trial_executor import RayTrialExecutor\nfrom ray.tune.syncer import wait_for_sync\nfrom ray.tune.trial_runner import TrialRunner\nfrom ray.tune.schedulers import (HyperBandScheduler, AsyncHyperBandScheduler,\n                                 FIFOScheduler, MedianStoppingRule)\nfrom ray.tune.web_server import TuneServer\n\nlogger = logging.getLogger(__name__)\n\n_SCHEDULERS = {\n    \"FIFO\": FIFOScheduler,\n    \"MedianStopping\": MedianStoppingRule,\n    \"HyperBand\": HyperBandScheduler,\n    \"AsyncHyperBand\": AsyncHyperBandScheduler,\n}\n\n\ndef _make_scheduler(args):\n    if args.scheduler in _SCHEDULERS:\n        return _SCHEDULERS[args.scheduler](**args.scheduler_config)\n    else:\n        raise TuneError(\"Unknown scheduler: {}, should be one of {}\".format(\n            args.scheduler, _SCHEDULERS.keys()))\n\n\ndef run(run_or_experiment,\n        name=None,\n        stop=None,\n        config=None,\n        resources_per_trial=None,\n        num_samples=1,\n        local_dir=None,\n        upload_dir=None,\n        trial_name_creator=None,\n        loggers=None,\n        sync_to_cloud=None,\n        sync_to_driver=None,\n        checkpoint_freq=0,\n        checkpoint_at_end=False,\n        global_checkpoint_period=10,\n        export_formats=None,\n        max_failures=3,\n        restore=None,\n        search_alg=None,\n        scheduler=None,\n        with_server=False,\n        server_port=TuneServer.DEFAULT_PORT,\n        verbose=2,\n        resume=False,\n        queue_trials=False,\n        reuse_actors=True,\n        trial_executor=None,\n        raise_on_failed_trial=True,\n        return_trials=True,\n        ray_auto_init=True,\n        sync_function=None):\n    \"\"\"Executes training.\n\n    Args:\n        run_or_experiment (function|class|str|Experiment): If\n            function|class|str, this is the algorithm or model to train.\n            This may refer to the name of a built-on algorithm\n            (e.g. RLLib's DQN or PPO), a user-defined trainable\n            function or class, or the string identifier of a\n            trainable function or class registered in the tune registry.\n            If Experiment, then Tune will execute training based on\n            Experiment.spec.\n        name (str): Name of experiment.\n        stop (dict): The stopping criteria. The keys may be any field in\n            the return result of 'train()', whichever is reached first.\n            Defaults to empty dict.\n        config (dict): Algorithm-specific configuration for Tune variant\n            generation (e.g. env, hyperparams). Defaults to empty dict.\n            Custom search algorithms may ignore this.\n        resources_per_trial (dict): Machine resources to allocate per trial,\n            e.g. ``{\"cpu\": 64, \"gpu\": 8}``. Note that GPUs will not be\n            assigned unless you specify them here. Defaults to 1 CPU and 0\n            GPUs in ``Trainable.default_resource_request()``.\n        num_samples (int): Number of times to sample from the\n            hyperparameter space. Defaults to 1. If `grid_search` is\n            provided as an argument, the grid will be repeated\n            `num_samples` of times.\n        local_dir (str): Local dir to save training results to.\n            Defaults to ``~/ray_results``.\n        upload_dir (str): Optional URI to sync training results\n            to (e.g. ``s3://bucket``).\n        trial_name_creator (func): Optional function for generating\n            the trial string representation.\n        loggers (list): List of logger creators to be used with\n            each Trial. If None, defaults to ray.tune.logger.DEFAULT_LOGGERS.\n            See `ray/tune/logger.py`.\n        sync_to_cloud (func|str): Function for syncing the local_dir to and\n            from upload_dir. If string, then it must be a string template\n            that includes `{source}` and `{target}` for the syncer to run.\n            If not provided, the sync command defaults to standard\n            S3 or gsutil sync comamnds.\n        sync_to_driver (func|str): Function for syncing trial logdir from\n            remote node to local. If string, then it must be a string template\n            that includes `{source}` and `{target}` for the syncer to run.\n            If not provided, defaults to using rsync.\n        checkpoint_freq (int): How many training iterations between\n            checkpoints. A value of 0 (default) disables checkpointing.\n        checkpoint_at_end (bool): Whether to checkpoint at the end of the\n            experiment regardless of the checkpoint_freq. Default is False.\n        global_checkpoint_period (int): Seconds between global checkpointing.\n            This does not affect `checkpoint_freq`, which specifies frequency\n            for individual trials.\n        export_formats (list): List of formats that exported at the end of\n            the experiment. Default is None.\n        max_failures (int): Try to recover a trial from its last\n            checkpoint at least this many times. Only applies if\n            checkpointing is enabled. Setting to -1 will lead to infinite\n            recovery retries. Defaults to 3.\n        restore (str): Path to checkpoint. Only makes sense to set if\n            running 1 trial. Defaults to None.\n        search_alg (SearchAlgorithm): Search Algorithm. Defaults to\n            BasicVariantGenerator.\n        scheduler (TrialScheduler): Scheduler for executing\n            the experiment. Choose among FIFO (default), MedianStopping,\n            AsyncHyperBand, and HyperBand.\n        with_server (bool): Starts a background Tune server. Needed for\n            using the Client API.\n        server_port (int): Port number for launching TuneServer.\n        verbose (int): 0, 1, or 2. Verbosity mode. 0 = silent,\n            1 = only status updates, 2 = status and trial results.\n        resume (str|bool): One of \"LOCAL\", \"REMOTE\", \"PROMPT\", or bool.\n            LOCAL/True restores the checkpoint from the local_checkpoint_dir.\n            REMOTE restores the checkpoint from remote_checkpoint_dir.\n            PROMPT provides CLI feedback. False forces a new\n            experiment. If resume is set but checkpoint does not exist,\n            ValueError will be thrown.\n        queue_trials (bool): Whether to queue trials when the cluster does\n            not currently have enough resources to launch one. This should\n            be set to True when running on an autoscaling cluster to enable\n            automatic scale-up.\n        reuse_actors (bool): Whether to reuse actors between different trials\n            when possible. This can drastically speed up experiments that start\n            and stop actors often (e.g., PBT in time-multiplexing mode). This\n            requires trials to have the same resource requirements.\n        trial_executor (TrialExecutor): Manage the execution of trials.\n        raise_on_failed_trial (bool): Raise TuneError if there exists failed\n            trial (of ERROR state) when the experiments complete.\n        ray_auto_init (bool): Automatically starts a local Ray cluster\n            if using a RayTrialExecutor (which is the default) and\n            if Ray is not initialized. Defaults to True.\n        sync_function: Deprecated. See `sync_to_cloud` and\n            `sync_to_driver`.\n\n    Returns:\n        List of Trial objects.\n\n    Raises:\n        TuneError if any trials failed and `raise_on_failed_trial` is True.\n\n    Examples:\n        >>> tune.run(mytrainable, scheduler=PopulationBasedTraining())\n\n        >>> tune.run(mytrainable, num_samples=5, reuse_actors=True)\n\n        >>> tune.run(\n                \"PG\",\n                num_samples=5,\n                config={\n                    \"env\": \"CartPole-v0\",\n                    \"lr\": tune.sample_from(lambda _: np.random.rand())\n                }\n            )\n    \"\"\"\n    trial_executor = trial_executor or RayTrialExecutor(\n        queue_trials=queue_trials,\n        reuse_actors=reuse_actors,\n        ray_auto_init=ray_auto_init)\n    experiment = run_or_experiment\n    if not isinstance(run_or_experiment, Experiment):\n        run_identifier = Experiment._register_if_needed(run_or_experiment)\n        experiment = Experiment(\n            name=name,\n            run=run_identifier,\n            stop=stop,\n            config=config,\n            resources_per_trial=resources_per_trial,\n            num_samples=num_samples,\n            local_dir=local_dir,\n            upload_dir=upload_dir,\n            sync_to_driver=sync_to_driver,\n            trial_name_creator=trial_name_creator,\n            loggers=loggers,\n            checkpoint_freq=checkpoint_freq,\n            checkpoint_at_end=checkpoint_at_end,\n            export_formats=export_formats,\n            max_failures=max_failures,\n            restore=restore,\n            sync_function=sync_function)\n    else:\n        logger.debug(\"Ignoring some parameters passed into tune.run.\")\n\n    if sync_to_cloud:\n        assert experiment.remote_checkpoint_dir, (\n            \"Need `upload_dir` if `sync_to_cloud` given.\")\n\n    runner = TrialRunner(\n        search_alg=search_alg or BasicVariantGenerator(),\n        scheduler=scheduler or FIFOScheduler(),\n        local_checkpoint_dir=experiment.checkpoint_dir,\n        remote_checkpoint_dir=experiment.remote_checkpoint_dir,\n        sync_to_cloud=sync_to_cloud,\n        checkpoint_period=global_checkpoint_period,\n        resume=resume,\n        launch_web_server=with_server,\n        server_port=server_port,\n        verbose=bool(verbose > 1),\n        trial_executor=trial_executor)\n\n    runner.add_experiment(experiment)\n\n    if verbose:\n        print(runner.debug_string(max_debug=99999))\n\n    last_debug = 0\n    while not runner.is_finished():\n        runner.step()\n        if time.time() - last_debug > DEBUG_PRINT_INTERVAL:\n            if verbose:\n                print(runner.debug_string())\n            last_debug = time.time()\n\n    if verbose:\n        print(runner.debug_string(max_debug=99999))\n\n    wait_for_sync()\n\n    errored_trials = []\n    for trial in runner.get_trials():\n        if trial.status != Trial.TERMINATED:\n            errored_trials += [trial]\n\n    if errored_trials:\n        if raise_on_failed_trial:\n            raise TuneError(\"Trials did not complete\", errored_trials)\n        else:\n            logger.error(\"Trials did not complete: %s\", errored_trials)\n\n    if return_trials:\n        return runner.get_trials()\n    return ExperimentAnalysis(experiment.checkpoint_dir)\n\n\ndef run_experiments(experiments,\n                    search_alg=None,\n                    scheduler=None,\n                    with_server=False,\n                    server_port=TuneServer.DEFAULT_PORT,\n                    verbose=2,\n                    resume=False,\n                    queue_trials=False,\n                    reuse_actors=False,\n                    trial_executor=None,\n                    raise_on_failed_trial=True):\n    \"\"\"Runs and blocks until all trials finish.\n\n    Examples:\n        >>> experiment_spec = Experiment(\"experiment\", my_func)\n        >>> run_experiments(experiments=experiment_spec)\n\n        >>> experiment_spec = {\"experiment\": {\"run\": my_func}}\n        >>> run_experiments(experiments=experiment_spec)\n\n        >>> run_experiments(\n        >>>     experiments=experiment_spec,\n        >>>     scheduler=MedianStoppingRule(...))\n\n        >>> run_experiments(\n        >>>     experiments=experiment_spec,\n        >>>     search_alg=SearchAlgorithm(),\n        >>>     scheduler=MedianStoppingRule(...))\n\n    Returns:\n        List of Trial objects, holding data for each executed trial.\n\n    \"\"\"\n    # This is important to do this here\n    # because it schematize the experiments\n    # and it conducts the implicit registration.\n    experiments = convert_to_experiment_list(experiments)\n\n    trials = []\n    for exp in experiments:\n        trials += run(\n            exp,\n            search_alg=search_alg,\n            scheduler=scheduler,\n            with_server=with_server,\n            server_port=server_port,\n            verbose=verbose,\n            resume=resume,\n            queue_trials=queue_trials,\n            reuse_actors=reuse_actors,\n            trial_executor=trial_executor,\n            raise_on_failed_trial=raise_on_failed_trial)\n    return trials\n", "path": "python/ray/tune/tune.py"}], "after_files": [{"content": "from __future__ import absolute_import\nfrom __future__ import division\nfrom __future__ import print_function\n\nimport logging\nimport time\n\nfrom ray.tune.error import TuneError\nfrom ray.tune.experiment import convert_to_experiment_list, Experiment\nfrom ray.tune.analysis import ExperimentAnalysis\nfrom ray.tune.suggest import BasicVariantGenerator\nfrom ray.tune.trial import Trial, DEBUG_PRINT_INTERVAL\nfrom ray.tune.ray_trial_executor import RayTrialExecutor\nfrom ray.tune.syncer import wait_for_sync\nfrom ray.tune.trial_runner import TrialRunner\nfrom ray.tune.schedulers import (HyperBandScheduler, AsyncHyperBandScheduler,\n                                 FIFOScheduler, MedianStoppingRule)\nfrom ray.tune.web_server import TuneServer\n\nlogger = logging.getLogger(__name__)\n\n_SCHEDULERS = {\n    \"FIFO\": FIFOScheduler,\n    \"MedianStopping\": MedianStoppingRule,\n    \"HyperBand\": HyperBandScheduler,\n    \"AsyncHyperBand\": AsyncHyperBandScheduler,\n}\n\n\ndef _make_scheduler(args):\n    if args.scheduler in _SCHEDULERS:\n        return _SCHEDULERS[args.scheduler](**args.scheduler_config)\n    else:\n        raise TuneError(\"Unknown scheduler: {}, should be one of {}\".format(\n            args.scheduler, _SCHEDULERS.keys()))\n\n\ndef run(run_or_experiment,\n        name=None,\n        stop=None,\n        config=None,\n        resources_per_trial=None,\n        num_samples=1,\n        local_dir=None,\n        upload_dir=None,\n        trial_name_creator=None,\n        loggers=None,\n        sync_to_cloud=None,\n        sync_to_driver=None,\n        checkpoint_freq=0,\n        checkpoint_at_end=False,\n        keep_checkpoints_num=None,\n        checkpoint_score_attr=None,\n        global_checkpoint_period=10,\n        export_formats=None,\n        max_failures=3,\n        restore=None,\n        search_alg=None,\n        scheduler=None,\n        with_server=False,\n        server_port=TuneServer.DEFAULT_PORT,\n        verbose=2,\n        resume=False,\n        queue_trials=False,\n        reuse_actors=True,\n        trial_executor=None,\n        raise_on_failed_trial=True,\n        return_trials=True,\n        ray_auto_init=True,\n        sync_function=None):\n    \"\"\"Executes training.\n\n    Args:\n        run_or_experiment (function|class|str|Experiment): If\n            function|class|str, this is the algorithm or model to train.\n            This may refer to the name of a built-on algorithm\n            (e.g. RLLib's DQN or PPO), a user-defined trainable\n            function or class, or the string identifier of a\n            trainable function or class registered in the tune registry.\n            If Experiment, then Tune will execute training based on\n            Experiment.spec.\n        name (str): Name of experiment.\n        stop (dict): The stopping criteria. The keys may be any field in\n            the return result of 'train()', whichever is reached first.\n            Defaults to empty dict.\n        config (dict): Algorithm-specific configuration for Tune variant\n            generation (e.g. env, hyperparams). Defaults to empty dict.\n            Custom search algorithms may ignore this.\n        resources_per_trial (dict): Machine resources to allocate per trial,\n            e.g. ``{\"cpu\": 64, \"gpu\": 8}``. Note that GPUs will not be\n            assigned unless you specify them here. Defaults to 1 CPU and 0\n            GPUs in ``Trainable.default_resource_request()``.\n        num_samples (int): Number of times to sample from the\n            hyperparameter space. Defaults to 1. If `grid_search` is\n            provided as an argument, the grid will be repeated\n            `num_samples` of times.\n        local_dir (str): Local dir to save training results to.\n            Defaults to ``~/ray_results``.\n        upload_dir (str): Optional URI to sync training results\n            to (e.g. ``s3://bucket``).\n        trial_name_creator (func): Optional function for generating\n            the trial string representation.\n        loggers (list): List of logger creators to be used with\n            each Trial. If None, defaults to ray.tune.logger.DEFAULT_LOGGERS.\n            See `ray/tune/logger.py`.\n        sync_to_cloud (func|str): Function for syncing the local_dir to and\n            from upload_dir. If string, then it must be a string template\n            that includes `{source}` and `{target}` for the syncer to run.\n            If not provided, the sync command defaults to standard\n            S3 or gsutil sync comamnds.\n        sync_to_driver (func|str): Function for syncing trial logdir from\n            remote node to local. If string, then it must be a string template\n            that includes `{source}` and `{target}` for the syncer to run.\n            If not provided, defaults to using rsync.\n        checkpoint_freq (int): How many training iterations between\n            checkpoints. A value of 0 (default) disables checkpointing.\n        checkpoint_at_end (bool): Whether to checkpoint at the end of the\n            experiment regardless of the checkpoint_freq. Default is False.\n        keep_checkpoints_num (int): Number of checkpoints to keep. A value of\n            `None` keeps all checkpoints. Defaults to `None`. If set, need\n            to provide `checkpoint_score_attr`.\n        checkpoint_score_attr (str): Specifies by which attribute to rank the\n            best checkpoint. Default is increasing order. If attribute starts\n            with `min-` it will rank attribute in decreasing order, i.e.\n            `min-validation_loss`.\n        global_checkpoint_period (int): Seconds between global checkpointing.\n            This does not affect `checkpoint_freq`, which specifies frequency\n            for individual trials.\n        export_formats (list): List of formats that exported at the end of\n            the experiment. Default is None.\n        max_failures (int): Try to recover a trial from its last\n            checkpoint at least this many times. Only applies if\n            checkpointing is enabled. Setting to -1 will lead to infinite\n            recovery retries. Defaults to 3.\n        restore (str): Path to checkpoint. Only makes sense to set if\n            running 1 trial. Defaults to None.\n        search_alg (SearchAlgorithm): Search Algorithm. Defaults to\n            BasicVariantGenerator.\n        scheduler (TrialScheduler): Scheduler for executing\n            the experiment. Choose among FIFO (default), MedianStopping,\n            AsyncHyperBand, and HyperBand.\n        with_server (bool): Starts a background Tune server. Needed for\n            using the Client API.\n        server_port (int): Port number for launching TuneServer.\n        verbose (int): 0, 1, or 2. Verbosity mode. 0 = silent,\n            1 = only status updates, 2 = status and trial results.\n        resume (str|bool): One of \"LOCAL\", \"REMOTE\", \"PROMPT\", or bool.\n            LOCAL/True restores the checkpoint from the local_checkpoint_dir.\n            REMOTE restores the checkpoint from remote_checkpoint_dir.\n            PROMPT provides CLI feedback. False forces a new\n            experiment. If resume is set but checkpoint does not exist,\n            ValueError will be thrown.\n        queue_trials (bool): Whether to queue trials when the cluster does\n            not currently have enough resources to launch one. This should\n            be set to True when running on an autoscaling cluster to enable\n            automatic scale-up.\n        reuse_actors (bool): Whether to reuse actors between different trials\n            when possible. This can drastically speed up experiments that start\n            and stop actors often (e.g., PBT in time-multiplexing mode). This\n            requires trials to have the same resource requirements.\n        trial_executor (TrialExecutor): Manage the execution of trials.\n        raise_on_failed_trial (bool): Raise TuneError if there exists failed\n            trial (of ERROR state) when the experiments complete.\n        ray_auto_init (bool): Automatically starts a local Ray cluster\n            if using a RayTrialExecutor (which is the default) and\n            if Ray is not initialized. Defaults to True.\n        sync_function: Deprecated. See `sync_to_cloud` and\n            `sync_to_driver`.\n\n    Returns:\n        List of Trial objects.\n\n    Raises:\n        TuneError if any trials failed and `raise_on_failed_trial` is True.\n\n    Examples:\n        >>> tune.run(mytrainable, scheduler=PopulationBasedTraining())\n\n        >>> tune.run(mytrainable, num_samples=5, reuse_actors=True)\n\n        >>> tune.run(\n                \"PG\",\n                num_samples=5,\n                config={\n                    \"env\": \"CartPole-v0\",\n                    \"lr\": tune.sample_from(lambda _: np.random.rand())\n                }\n            )\n    \"\"\"\n    trial_executor = trial_executor or RayTrialExecutor(\n        queue_trials=queue_trials,\n        reuse_actors=reuse_actors,\n        ray_auto_init=ray_auto_init)\n    experiment = run_or_experiment\n    if not isinstance(run_or_experiment, Experiment):\n        run_identifier = Experiment._register_if_needed(run_or_experiment)\n        experiment = Experiment(\n            name=name,\n            run=run_identifier,\n            stop=stop,\n            config=config,\n            resources_per_trial=resources_per_trial,\n            num_samples=num_samples,\n            local_dir=local_dir,\n            upload_dir=upload_dir,\n            sync_to_driver=sync_to_driver,\n            trial_name_creator=trial_name_creator,\n            loggers=loggers,\n            checkpoint_freq=checkpoint_freq,\n            checkpoint_at_end=checkpoint_at_end,\n            keep_checkpoints_num=keep_checkpoints_num,\n            checkpoint_score_attr=checkpoint_score_attr,\n            export_formats=export_formats,\n            max_failures=max_failures,\n            restore=restore,\n            sync_function=sync_function)\n    else:\n        logger.debug(\"Ignoring some parameters passed into tune.run.\")\n\n    if sync_to_cloud:\n        assert experiment.remote_checkpoint_dir, (\n            \"Need `upload_dir` if `sync_to_cloud` given.\")\n\n    runner = TrialRunner(\n        search_alg=search_alg or BasicVariantGenerator(),\n        scheduler=scheduler or FIFOScheduler(),\n        local_checkpoint_dir=experiment.checkpoint_dir,\n        remote_checkpoint_dir=experiment.remote_checkpoint_dir,\n        sync_to_cloud=sync_to_cloud,\n        checkpoint_period=global_checkpoint_period,\n        resume=resume,\n        launch_web_server=with_server,\n        server_port=server_port,\n        verbose=bool(verbose > 1),\n        trial_executor=trial_executor)\n\n    runner.add_experiment(experiment)\n\n    if verbose:\n        print(runner.debug_string(max_debug=99999))\n\n    last_debug = 0\n    while not runner.is_finished():\n        runner.step()\n        if time.time() - last_debug > DEBUG_PRINT_INTERVAL:\n            if verbose:\n                print(runner.debug_string())\n            last_debug = time.time()\n\n    if verbose:\n        print(runner.debug_string(max_debug=99999))\n\n    wait_for_sync()\n\n    errored_trials = []\n    for trial in runner.get_trials():\n        if trial.status != Trial.TERMINATED:\n            errored_trials += [trial]\n\n    if errored_trials:\n        if raise_on_failed_trial:\n            raise TuneError(\"Trials did not complete\", errored_trials)\n        else:\n            logger.error(\"Trials did not complete: %s\", errored_trials)\n\n    if return_trials:\n        return runner.get_trials()\n    return ExperimentAnalysis(experiment.checkpoint_dir)\n\n\ndef run_experiments(experiments,\n                    search_alg=None,\n                    scheduler=None,\n                    with_server=False,\n                    server_port=TuneServer.DEFAULT_PORT,\n                    verbose=2,\n                    resume=False,\n                    queue_trials=False,\n                    reuse_actors=False,\n                    trial_executor=None,\n                    raise_on_failed_trial=True):\n    \"\"\"Runs and blocks until all trials finish.\n\n    Examples:\n        >>> experiment_spec = Experiment(\"experiment\", my_func)\n        >>> run_experiments(experiments=experiment_spec)\n\n        >>> experiment_spec = {\"experiment\": {\"run\": my_func}}\n        >>> run_experiments(experiments=experiment_spec)\n\n        >>> run_experiments(\n        >>>     experiments=experiment_spec,\n        >>>     scheduler=MedianStoppingRule(...))\n\n        >>> run_experiments(\n        >>>     experiments=experiment_spec,\n        >>>     search_alg=SearchAlgorithm(),\n        >>>     scheduler=MedianStoppingRule(...))\n\n    Returns:\n        List of Trial objects, holding data for each executed trial.\n\n    \"\"\"\n    # This is important to do this here\n    # because it schematize the experiments\n    # and it conducts the implicit registration.\n    experiments = convert_to_experiment_list(experiments)\n\n    trials = []\n    for exp in experiments:\n        trials += run(\n            exp,\n            search_alg=search_alg,\n            scheduler=scheduler,\n            with_server=with_server,\n            server_port=server_port,\n            verbose=verbose,\n            resume=resume,\n            queue_trials=queue_trials,\n            reuse_actors=reuse_actors,\n            trial_executor=trial_executor,\n            raise_on_failed_trial=raise_on_failed_trial)\n    return trials\n", "path": "python/ray/tune/tune.py"}]}