diff --git a/python/ray/tune/syncer.py b/python/ray/tune/syncer.py index 5b0beb312..9b97fc31e 100644 --- a/python/ray/tune/syncer.py +++ b/python/ray/tune/syncer.py @@ -7,6 +7,7 @@ from shlex import quote from ray import ray_constants from ray import services +from ray.util.debug import log_once from ray.tune.cluster_info import get_ssh_key, get_ssh_user from ray.tune.sync_client import (CommandBasedClient, get_sync_client, get_cloud_sync_client, NOOP) @@ -43,7 +44,8 @@ def log_sync_template(options=""): unavailable. """ if not distutils.spawn.find_executable("rsync"): - logger.error("Log sync requires rsync to be installed.") + with log_once("tune:rsync"): + logger.error("Log sync requires rsync to be installed.") return None global _log_sync_warned ssh_key = get_ssh_key() diff --git a/python/ray/tune/tests/test_trial_runner_2.py b/python/ray/tune/tests/test_trial_runner_2.py index 0cf5d434f..67b18c084 100644 --- a/python/ray/tune/tests/test_trial_runner_2.py +++ b/python/ray/tune/tests/test_trial_runner_2.py @@ -216,6 +216,30 @@ class TrialRunnerTest2(unittest.TestCase): self.assertEqual(trials[0].status, Trial.ERROR) self.assertRaises(TuneError, lambda: runner.step()) + def testFailFastRaise(self): + ray.init(num_cpus=1, num_gpus=1) + runner = TrialRunner(fail_fast=TrialRunner.RAISE) + kwargs = { + "resources": Resources(cpu=1, gpu=1), + "checkpoint_freq": 1, + "max_failures": 0, + "config": { + "mock_error": True, + "persistent_error": True, + }, + } + runner.add_trial(Trial("__fake", **kwargs)) + runner.add_trial(Trial("__fake", **kwargs)) + trials = runner.get_trials() + + runner.step() # Start trial + self.assertEqual(trials[0].status, Trial.RUNNING) + runner.step() # Process result, dispatch save + self.assertEqual(trials[0].status, Trial.RUNNING) + runner.step() # Process save + with self.assertRaises(Exception): + runner.step() # Error + def testCheckpointing(self): ray.init(num_cpus=1, num_gpus=1) runner = TrialRunner() diff --git a/python/ray/tune/trial_runner.py b/python/ray/tune/trial_runner.py index dfdf4462f..2cf8b1ca3 100644 --- a/python/ray/tune/trial_runner.py +++ b/python/ray/tune/trial_runner.py @@ -104,7 +104,10 @@ class TrialRunner: resume (str|False): see `tune.py:run`. sync_to_cloud (func|str): See `tune.py:run`. server_port (int): Port number for launching TuneServer. - fail_fast (bool): Finishes as soon as a trial fails if True. + fail_fast (bool | str): Finishes as soon as a trial fails if True. + If fail_fast='raise' provided, Tune will automatically + raise the exception received by the Trainable. fail_fast='raise' + can easily leak resources and should be used with caution. verbose (bool): Flag for verbosity. If False, trial results will not be output. checkpoint_period (int): Trial runner checkpoint periodicity in @@ -114,6 +117,7 @@ class TrialRunner: CKPT_FILE_TMPL = "experiment_state-{}.json" VALID_RESUME_TYPES = [True, "LOCAL", "REMOTE", "PROMPT"] + RAISE = "RAISE" def __init__(self, search_alg=None, @@ -141,6 +145,18 @@ class TrialRunner: self._iteration = 0 self._has_errored = False self._fail_fast = fail_fast + if isinstance(self._fail_fast, str): + self._fail_fast = self._fail_fast.upper() + if self._fail_fast == TrialRunner.RAISE: + logger.warning( + "fail_fast='raise' detected. Be careful when using this " + "mode as resources (such as Ray processes, " + "file descriptors, and temporary files) may not be " + "cleaned up properly. To use " + "a safer mode, use fail_fast=True.") + else: + raise ValueError("fail_fast must be one of {bool, RAISE}. " + f"Got {self._fail_fast}.") self._verbose = verbose self._server = None @@ -531,6 +547,8 @@ class TrialRunner: self._execute_action(trial, decision) except Exception: logger.exception("Trial %s: Error processing event.", trial) + if self._fail_fast == TrialRunner.RAISE: + raise self._process_trial_failure(trial, traceback.format_exc()) def _process_trial_save(self, trial): @@ -548,6 +566,8 @@ class TrialRunner: checkpoint_value = self.trial_executor.fetch_result(trial) except Exception: logger.exception("Trial %s: Error processing result.", trial) + if self._fail_fast == TrialRunner.RAISE: + raise self._process_trial_failure(trial, traceback.format_exc()) if checkpoint_value: @@ -558,6 +578,8 @@ class TrialRunner: except Exception: logger.exception("Trial %s: Error handling checkpoint %s", trial, checkpoint_value) + if self._fail_fast == TrialRunner.RAISE: + raise trial.saving_to = None decision = self._cached_trial_decisions.pop(trial.trial_id, None) @@ -579,6 +601,8 @@ class TrialRunner: self.trial_executor.continue_training(trial) except Exception: logger.exception("Trial %s: Error processing restore.", trial) + if self._fail_fast == TrialRunner.RAISE: + raise self._process_trial_failure(trial, traceback.format_exc()) def _process_trial_failure(self, trial, error_msg): diff --git a/python/ray/tune/tune.py b/python/ray/tune/tune.py index f1182dae6..062e65d0f 100644 --- a/python/ray/tune/tune.py +++ b/python/ray/tune/tune.py @@ -190,7 +190,11 @@ def run(run_or_experiment, Ray will recover from the latest checkpoint if present. Setting to -1 will lead to infinite recovery retries. Setting to 0 will disable retries. Defaults to 3. - fail_fast (bool): Whether to fail upon the first error. + fail_fast (bool | str): Whether to fail upon the first error. + If fail_fast='raise' provided, Tune will automatically + raise the exception received by the Trainable. fail_fast='raise' + can easily leak resources and should be used with caution (it + is best used with `ray.init(local_mode=True)`). restore (str): Path to checkpoint. Only makes sense to set if running 1 trial. Defaults to None. search_alg (Searcher): Search algorithm for optimization.