wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-20 12:40:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

[tune] [rllib] Automatically determine RLlib resources and add queueing mechanism for autoscaling (#1848)

Browse Source
This commit is contained in:
Eric Liang
2018-04-16 16:58:15 -07:00
committed by Richard Liaw
parent 2e25972d4d
commit 7ab890f4a1
39 changed files with 286 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/ray/tune/config_parser.py
+8 -11
View File
@@ -11,6 +11,8 @@ from ray.tune.trial import Resources
def json_to_resources(data):
if data is None or data == "null":
return None
if type(data) is str:
data = json.loads(data)
for k in data:
@@ -29,7 +31,7 @@ def json_to_resources(data):
def resources_to_json(resources):
if resources is None:
resources = Resources(cpu=1, gpu=0)
return None
return {
"cpu": resources.cpu,
"gpu": resources.gpu,
@@ -70,19 +72,14 @@ def make_parser(**kwargs):
type=json.loads,
help="Algorithm-specific configuration (e.g. env, hyperparams), "
"specified in JSON.")
parser.add_argument(
"--resources",
help="Deprecated, use --trial-resources.",
type=lambda v: _tune_error("The `resources` argument is no longer "
"supported. Use `trial_resources` or "
"--trial-resources instead."))
parser.add_argument(
"--trial-resources",
default='{"cpu": 1}',
default=None,
type=json_to_resources,
help="Machine resources to allocate per trial, e.g. "
help="Override the machine resources to allocate per trial, e.g. "
"'{\"cpu\": 64, \"gpu\": 8}'. Note that GPUs will not be assigned "
"unless you specify them here.")
"unless you specify them here. For RLlib, you probably want to "
"leave this alone and use RLlib configs to control parallelism.")
parser.add_argument(
"--repeat",
default=1,
@@ -115,7 +112,7 @@ def make_parser(**kwargs):
"--scheduler",
default="FIFO",
type=str,
help="FIFO (default), MedianStopping, AsyncHyperBand,"
help="FIFO (default), MedianStopping, AsyncHyperBand, "
"HyperBand, or HyperOpt.")
parser.add_argument(
"--scheduler-config",
python/ray/tune/examples/pbt_example.py
-4
View File
@@ -86,10 +86,6 @@ if __name__ == "__main__":
"training_iteration": 2 if args.smoke_test else 99999
},
"repeat": 10,
"trial_resources": {
"cpu": 1,
"gpu": 0
},
"config": {
"factor_1": 4.0,
"factor_2": 1.0,
python/ray/tune/examples/pbt_ppo_example.py
-4
View File
@@ -51,10 +51,6 @@ if __name__ == "__main__":
"run": "PPO",
"env": "Humanoid-v1",
"repeat": 8,
"trial_resources": {
"cpu": 4,
"gpu": 1
},
"config": {
"kl_coeff":
1.0,
python/ray/tune/test/trial_runner_test.py
+50 -3
View File
@@ -12,7 +12,7 @@ from ray.rllib import _register_all
from ray.tune import Trainable, TuneError
from ray.tune import register_env, register_trainable, run_experiments
from ray.tune.registry import _default_registry, TRAINABLE_CLASS
from ray.tune.result import DEFAULT_RESULTS_DIR
from ray.tune.result import DEFAULT_RESULTS_DIR, TrainingResult
from ray.tune.util import pin_in_object_store, get_pinned_object
from ray.tune.experiment import Experiment
from ray.tune.trial import Trial, Resources
@@ -23,7 +23,7 @@ from ray.tune.variant_generator import generate_trials, grid_search, \
class TrainableFunctionApiTest(unittest.TestCase):
def setUp(self):
ray.init()
ray.init(num_cpus=4, num_gpus=0)
def tearDown(self):
ray.worker.cleanup()
@@ -76,6 +76,46 @@ class TrainableFunctionApiTest(unittest.TestCase):
self.assertRaises(TypeError, lambda: register_trainable("foo", B()))
self.assertRaises(TypeError, lambda: register_trainable("foo", A))
def testBuiltInTrainableResources(self):
class B(Trainable):
@classmethod
def default_resource_request(cls, config):
return Resources(cpu=config["cpu"], gpu=config["gpu"])
def _train(self):
return TrainingResult(timesteps_this_iter=1, done=True)
register_trainable("B", B)
def f(cpus, gpus, queue_trials):
return run_experiments(
{
"foo": {
"run": "B",
"config": {
"cpu": cpus,
"gpu": gpus,
},
}
},
queue_trials=queue_trials)[0]
# Should all succeed
self.assertEqual(f(0, 0, False).status, Trial.TERMINATED)
self.assertEqual(f(1, 0, True).status, Trial.TERMINATED)
self.assertEqual(f(1, 0, True).status, Trial.TERMINATED)
# Infeasible even with queueing enabled (no gpus)
self.assertRaises(TuneError, lambda: f(1, 1, True))
# Too large resource request
self.assertRaises(TuneError, lambda: f(100, 100, False))
self.assertRaises(TuneError, lambda: f(0, 100, False))
self.assertRaises(TuneError, lambda: f(100, 0, False))
# TODO(ekl) how can we test this is queued (hangs)?
# f(100, 0, True)
def testRewriteEnv(self):
def train(config, reporter):
reporter(timesteps_total=1)
@@ -357,6 +397,13 @@ class RunExperimentTest(unittest.TestCase):
class VariantGeneratorTest(unittest.TestCase):
def setUp(self):
ray.init()
def tearDown(self):
ray.worker.cleanup()
_register_all() # re-register the evicted objects
def testParseToTrials(self):
trials = generate_trials({
"run": "PPO",
@@ -531,7 +578,7 @@ class TrialRunnerTest(unittest.TestCase):
def testTrialErrorOnStart(self):
ray.init()
_default_registry.register(TRAINABLE_CLASS, "asdf", None)
trial = Trial("asdf")
trial = Trial("asdf", resources=Resources(1, 0))
try:
trial.start()
except Exception as e:
python/ray/tune/test/trial_scheduler_test.py
+29
View File
@@ -6,6 +6,7 @@ import random
import unittest
import numpy as np
import ray
from ray.tune.hyperband import HyperBandScheduler
from ray.tune.async_hyperband import AsyncHyperBandScheduler
from ray.tune.pbt import PopulationBasedTraining, explore
@@ -24,6 +25,13 @@ def result(t, rew):
class EarlyStoppingSuite(unittest.TestCase):
def setUp(self):
ray.init()
def tearDown(self):
ray.worker.cleanup()
_register_all() # re-register the evicted objects
def basicSetup(self, rule):
t1 = Trial("PPO") # mean is 450, max 900, t_max=10
t2 = Trial("PPO") # mean is 450, max 450, t_max=5
@@ -184,6 +192,13 @@ class _MockTrialRunner():
class HyperbandSuite(unittest.TestCase):
def setUp(self):
ray.init()
def tearDown(self):
ray.worker.cleanup()
_register_all() # re-register the evicted objects
def schedulerSetup(self, num_trials):
"""Setup a scheduler and Runner with max Iter = 9
@@ -538,6 +553,13 @@ class _MockTrial(Trial):
class PopulationBasedTestingSuite(unittest.TestCase):
def setUp(self):
ray.init()
def tearDown(self):
ray.worker.cleanup()
_register_all() # re-register the evicted objects
def basicSetup(self, resample_prob=0.0, explore=None):
pbt = PopulationBasedTraining(
time_attr="training_iteration",
@@ -751,6 +773,13 @@ class PopulationBasedTestingSuite(unittest.TestCase):
class AsyncHyperBandSuite(unittest.TestCase):
def setUp(self):
ray.init()
def tearDown(self):
ray.worker.cleanup()
_register_all() # re-register the evicted objects
def basicSetup(self, scheduler):
t1 = Trial("PPO") # mean is 450, max 900, t_max=10
t2 = Trial("PPO") # mean is 450, max 450, t_max=5
python/ray/tune/trainable.py
+17
View File
@@ -17,6 +17,7 @@ import ray
from ray.tune import TuneError
from ray.tune.logger import UnifiedLogger
from ray.tune.result import DEFAULT_RESULTS_DIR
from ray.tune.trial import Resources
class Trainable(object):
@@ -90,6 +91,22 @@ class Trainable(object):
self._initialize_ok = True
self._local_ip = ray.services.get_node_ip_address()
@classmethod
def default_resource_request(cls, config):
"""Returns the resource requirement for the given configuration.
This can be overriden by sub-classes to set the correct trial resource
allocation, so the user does not need to.
"""
return Resources(cpu=1, gpu=0)
@classmethod
def resource_help(cls, config):
"""Returns a help string for configuring this trainable's resources."""
return ""
def train(self):
"""Runs one logical iteration of training.
python/ray/tune/trial.py
+9 -5
View File
@@ -82,7 +82,7 @@ class Trial(object):
config=None,
local_dir=DEFAULT_RESULTS_DIR,
experiment_tag="",
resources=Resources(cpu=1, gpu=0),
resources=None,
stopping_criterion=None,
checkpoint_freq=0,
restore_path=None,
@@ -112,7 +112,9 @@ class Trial(object):
self.config = config or {}
self.local_dir = local_dir
self.experiment_tag = experiment_tag
self.resources = resources
self.resources = (
resources
or self._get_trainable_cls().default_resource_request(self.config))
self.stopping_criterion = stopping_criterion or {}
self.checkpoint_freq = checkpoint_freq
self.upload_dir = upload_dir
@@ -350,11 +352,9 @@ class Trial(object):
def _setup_runner(self):
self.status = Trial.RUNNING
trainable_cls = ray.tune.registry.get_registry().get(
ray.tune.registry.TRAINABLE_CLASS, self.trainable_name)
cls = ray.remote(
num_cpus=self.resources.cpu,
num_gpus=self.resources.gpu)(trainable_cls)
num_gpus=self.resources.gpu)(self._get_trainable_cls())
if not self.result_logger:
if not os.path.exists(self.local_dir):
os.makedirs(self.local_dir)
@@ -380,6 +380,10 @@ class Trial(object):
registry=ray.tune.registry.get_registry(),
logger_creator=logger_creator)
def _get_trainable_cls(self):
return ray.tune.registry.get_registry().get(
ray.tune.registry.TRAINABLE_CLASS, self.trainable_name)
def set_verbose(self, verbose):
self.verbose = verbose
python/ray/tune/trial_runner.py
+41 -8
View File
@@ -42,7 +42,8 @@ class TrialRunner(object):
scheduler=None,
launch_web_server=False,
server_port=TuneServer.DEFAULT_PORT,
verbose=True):
verbose=True,
queue_trials=False):
"""Initializes a new TrialRunner.
Args:
@@ -51,6 +52,10 @@ class TrialRunner(object):
server_port (int): Port number for launching TuneServer
verbose (bool): Flag for verbosity. If False, trial results
will not be output.
queue_trials (bool): Whether to queue trials when the cluster does
not currently have enough resources to launch one. This should
be set to True when running on an autoscaling cluster to enable
automatic scale-up.
"""
self._scheduler_alg = scheduler or FIFOScheduler()
@@ -70,6 +75,7 @@ class TrialRunner(object):
self._server = TuneServer(self, server_port)
self._stop_queue = []
self._verbose = verbose
self._queue_trials = queue_trials
def is_finished(self):
"""Returns whether all trials have finished running."""
@@ -102,9 +108,14 @@ class TrialRunner(object):
raise TuneError(
("Insufficient cluster resources to launch trial: "
"trial requested {} but the cluster only has {} "
"available.").format(
"available. Pass `queue_trials=True` in "
"ray.tune.run_experiments() or on the command "
"line to queue trials until the cluster scales "
"up. {}").format(
trial.resources.summary_string(),
self._avail_resources.summary_string()))
self._avail_resources.summary_string(),
trial._get_trainable_cls().resource_help(
trial.config)))
elif trial.status == Trial.PAUSED:
raise TuneError(
"There are paused trials, but no more pending "
@@ -177,9 +188,10 @@ class TrialRunner(object):
messages = ["== Status =="]
messages.append(self._scheduler_alg.debug_string())
if self._resources_initialized:
messages.append("Resources used: {}/{} CPUs, {}/{} GPUs".format(
self._committed_resources.cpu, self._avail_resources.cpu,
self._committed_resources.gpu, self._avail_resources.gpu))
messages.append(
"Resources requested: {}/{} CPUs, {}/{} GPUs".format(
self._committed_resources.cpu, self._avail_resources.cpu,
self._committed_resources.gpu, self._avail_resources.gpu))
return messages
def has_resources(self, resources):
@@ -187,8 +199,29 @@ class TrialRunner(object):
cpu_avail = self._avail_resources.cpu - self._committed_resources.cpu
gpu_avail = self._avail_resources.gpu - self._committed_resources.gpu
return (resources.cpu_total() <= cpu_avail
and resources.gpu_total() <= gpu_avail)
have_space = (resources.cpu_total() <= cpu_avail
and resources.gpu_total() <= gpu_avail)
if have_space:
return True
can_overcommit = self._queue_trials
if ((resources.cpu_total() > 0 and cpu_avail <= 0)
or (resources.gpu_total() > 0 and gpu_avail <= 0)):
can_overcommit = False # requested resource is already saturated
if can_overcommit:
print("WARNING:tune:allowing trial to start even though the "
"cluster does not have enough free resources. Trial actors "
"may appear to hang until enough resources are added to the "
"cluster (e.g., via autoscaling). You can disable this "
"behavior by specifying `queue_trials=False` in "
"ray.tune.run_experiments().")
return True
return False
def _get_next_trial(self):
self._update_avail_resources()
python/ray/tune/tune.py
+8 -2
View File
@@ -37,7 +37,8 @@ def run_experiments(experiments,
scheduler=None,
with_server=False,
server_port=TuneServer.DEFAULT_PORT,
verbose=True):
verbose=True,
queue_trials=False):
"""Tunes experiments.
Args:
@@ -49,6 +50,10 @@ def run_experiments(experiments,
using the Client API.
server_port (int): Port number for launching TuneServer.
verbose (bool): How much output should be printed for each trial.
queue_trials (bool): Whether to queue trials when the cluster does
not currently have enough resources to launch one. This should
be set to True when running on an autoscaling cluster to enable
automatic scale-up.
"""
if scheduler is None:
@@ -58,7 +63,8 @@ def run_experiments(experiments,
scheduler,
launch_web_server=with_server,
server_port=server_port,
verbose=verbose)
verbose=verbose,
queue_trials=queue_trials)
exp_list = experiments
if isinstance(experiments, Experiment):
exp_list = [experiments]
python/ray/tune/variant_generator.py
+5 -1
View File
@@ -53,12 +53,16 @@ def generate_trials(unresolved_spec, output_path=''):
else:
experiment_tag = str(i)
i += 1
if "trial_resources" in spec:
resources = json_to_resources(spec["trial_resources"])
else:
resources = None
yield Trial(
trainable_name=spec["run"],
config=spec.get("config", {}),
local_dir=os.path.join(args.local_dir, output_path),
experiment_tag=experiment_tag,
resources=json_to_resources(spec.get("trial_resources", {})),
resources=resources,
stopping_criterion=spec.get("stop", {}),
checkpoint_freq=args.checkpoint_freq,
restore_path=spec.get("restore"),