From a0df13b14f7b87186a58494ffb2ccf450aea90ec Mon Sep 17 00:00:00 2001 From: Philipp Moritz Date: Mon, 27 Jun 2016 16:33:12 -0700 Subject: [PATCH] Implement launching cluster with shell --- scripts/cluster.py | 2 +- scripts/default_worker.py | 28 ++++++++++++++++++++++++++++ scripts/shell.py | 20 ++++++++++++++++++-- 3 files changed, 47 insertions(+), 3 deletions(-) create mode 100644 scripts/default_worker.py diff --git a/scripts/cluster.py b/scripts/cluster.py index 868fb84f0..78bf015fb 100644 --- a/scripts/cluster.py +++ b/scripts/cluster.py @@ -65,7 +65,7 @@ def start_ray_multi_node(node_ip_addresses, username, key_file, worker_path, ins shell_script_path = os.path.join(args.installation_directory, "ray/scripts/shell.py") print """ source "{}"; - python "{}" --scheduler-address={}:10001 --objstore-address={}:20001 --worker-address={}:30001 + python "{}" --scheduler-address={}:10001 --objstore-address={}:20001 --worker-address={}:30001 --attach """.format(setup_env_path, shell_script_path, node_ip_addresses[0], node_ip_addresses[0], node_ip_addresses[0]) def stop_ray_multi_node(node_ip_addresses, username, key_file): diff --git a/scripts/default_worker.py b/scripts/default_worker.py new file mode 100644 index 000000000..0876d5479 --- /dev/null +++ b/scripts/default_worker.py @@ -0,0 +1,28 @@ +import sys +import argparse +import numpy as np + +import ray.arrays.remote as ra +import ray.arrays.distributed as da + +import ray +import ray.services as services +import ray.worker as worker + +parser = argparse.ArgumentParser(description="Parse addresses for the worker to connect to.") +parser.add_argument("--scheduler-address", default="127.0.0.1:10001", type=str, help="the scheduler's address") +parser.add_argument("--objstore-address", default="127.0.0.1:20001", type=str, help="the objstore's address") +parser.add_argument("--worker-address", default="127.0.0.1:40001", type=str, help="the worker's address") + +if __name__ == "__main__": + args = parser.parse_args() + worker.connect(args.scheduler_address, args.objstore_address, args.worker_address) + + ray.register_module(ra) + ray.register_module(ra.random) + ray.register_module(ra.linalg) + ray.register_module(da) + ray.register_module(da.random) + ray.register_module(da.linalg) + + worker.main_loop() diff --git a/scripts/shell.py b/scripts/shell.py index 14cbc4849..39005ca40 100644 --- a/scripts/shell.py +++ b/scripts/shell.py @@ -1,3 +1,4 @@ +import os import argparse import numpy as np @@ -8,14 +9,29 @@ import ray.worker as worker import ray.array.remote as ra import ray.array.distributed as da -parser = argparse.ArgumentParser(description='Parse addresses for the worker to connect to.') +DEFAULT_NUM_WORKERS = 10 +DEFAULT_WORKER_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "default_worker.py") + +parser = argparse.ArgumentParser(description="Parse shell options") parser.add_argument("--scheduler-address", default="127.0.0.1:10001", type=str, help="the scheduler's address") parser.add_argument("--objstore-address", default="127.0.0.1:20001", type=str, help="the objstore's address") parser.add_argument("--worker-address", default="127.0.0.1:30001", type=str, help="the worker's address") +parser.add_argument("--attach", action="store_true", help="If true, attach the shell to an already running cluster. If false, start a new cluster.") +parser.add_argument("--worker-path", help="Path to the worker script") +parser.add_argument("--num-workers", help="Number of workers to start") if __name__ == "__main__": args = parser.parse_args() - worker.connect(args.scheduler_address, args.objstore_address, args.worker_address) + if args.attach: + assert args.worker_path is None, "when attaching, no new worker can be started" + assert args.num_workers is None, "when attaching, no new worker can be started" + worker.connect(args.scheduler_address, args.objstore_address, args.worker_address) + else: + services.start_ray_local(num_workers=args.num_workers if not args.num_workers is None else DEFAULT_NUM_WORKERS, + worker_path=args.worker_path if not args.num_workers is None else DEFAULT_WORKER_PATH) import IPython IPython.embed() + + if not args.attach: + services.cleanup()