diff --git a/README.md b/README.md index bb631349b..cf0e9ac53 100644 --- a/README.md +++ b/README.md @@ -23,3 +23,42 @@ For a description of our design decisions, see 3. git clone https://github.com/amplab/ray.git 4. cd ray 5. ./setup.sh + +## Installing Ray on a cluster + +These instructions work on EC2, but they may require some modifications to run +on your own cluster. In particular, on EC2, running `sudo` does not require a +password, and we currently don't handle the case where a password is needed. + +1. Create a file `nodes.txt` of the IP addresses of the nodes in the cluster. +For example + + 52.50.28.103 + 52.51.210.207 +2. Make sure that the nodes can all communicate with one another. On EC2, this +can be done by creating a new security group and adding the inbound rule "all +traffic" and adding the outbound rule "all traffic". Then add all of the nodes +in your cluster to that security group. + +3. Run something like + ``` + python scripts/cluster.py --nodes nodes.txt \ + --key-file key.pem \ + --username ubuntu \ + --installation-directory /home/ubuntu/ + ``` +where you replace `nodes.txt`, `key.pem`, `ubuntu`, and `/home/ubuntu/` by the +appropriate values. This assumes that you can connect to each IP address in +`nodes.txt` with the command + ``` + ssh -i key.pem ubuntu@ + ``` +4. The previous command should open a Python interpreter. To install Ray on the +cluster, run `install_ray(node_addresses)` in the interpreter. The interpreter +should block until the installation has completed. +5. To check that the installation succeeded, you can ssh to each node, cd into +the directory `ray/test/`, and run the tests (e.g., `python runtest.py`). +6. Now that Ray has been installed, you can start the cluster (the scheduler, +object stores, and workers) with the command `start_ray(node_addresses, +"/home/ubuntu/ray/test/test_worker.py")`, where the second argument is the path +on each node in the cluster to the worker code that you would like to use. diff --git a/scripts/cluster.py b/scripts/cluster.py new file mode 100644 index 000000000..44b7fae64 --- /dev/null +++ b/scripts/cluster.py @@ -0,0 +1,83 @@ +# This script can be used to start Ray on an existing cluster. +# +# How to use it: Create a file "nodes.txt" that contains a list of the IP +# addresses of the nodes in the cluster. Put the head node first. This node will +# host the driver and the scheduler. + +import os +import subprocess +import socket +import argparse +import threading +import IPython + +parser = argparse.ArgumentParser(description="Parse information about the cluster.") +parser.add_argument("--nodes", type=str, required=True, help="Test file with node IP addresses, one line per address.") +parser.add_argument("--key-file", type=str, required=True, help="Path to the file that contains the private key.") +parser.add_argument("--username", type=str, required=True, help="User name for logging in.") +parser.add_argument("--installation-directory", type=str, required=True, help="The directory in which to install Ray.") + +def run_command_over_ssh(node_ip_address, username, key_file, command): + full_command = "ssh -i {} {}@{} '{}'".format(key_file, username, node_ip_address, command) + subprocess.call([full_command], shell=True) + print "Finished running command '{}' on {}@{}.".format(command, username, node_ip_address) + +def install_ray_multi_node(node_ip_addresses, username, key_file, installation_directory): + def install_ray_over_ssh(node_ip_address, username, key_file, installation_directory): + install_ray_command = "sudo apt-get update; sudo apt-get -y install git; mkdir -p {}; cd {}; git clone https://github.com/amplab/ray; cd ray; ./setup.sh".format(installation_directory, installation_directory) + run_command_over_ssh(node_ip_address, username, key_file, install_ray_command) + threads = [] + for node_ip_address in node_ip_addresses: + t = threading.Thread(target=install_ray_over_ssh, args=(node_ip_address, username, key_file, installation_directory)) + t.start() + threads.append(t) + for t in threads: + t.join() + +def start_ray_multi_node(node_ip_addresses, username, key_file, worker_path, installation_directory): + build_directory = os.path.join(installation_directory, "ray/build") + start_scheduler_command = "cd {}; nohup ./scheduler {}:10001 > scheduler.out 2> scheduler.err < /dev/null &".format(build_directory, node_ip_addresses[0]) + run_command_over_ssh(node_ip_addresses[0], username, key_file, start_scheduler_command) + + for i, node_ip_address in enumerate(node_ip_addresses): + scripts_directory = os.path.join(installation_directory, "ray/scripts") + start_workers_command = "cd {}; python start_workers.py --scheduler-address={}:10001 --node-ip={} --worker-path={} > start_workers.out 2> start_workers.err < /dev/null &".format(scripts_directory, node_ip_addresses[0], node_ip_addresses[i], worker_path) + run_command_over_ssh(node_ip_address, username, key_file, start_workers_command) + + print "cluster started; you can start the shell on the head node with:" + shell_script_path = os.path.join(args.installation_directory, "ray/scripts/shell.py") + print "python {} --scheduler-address={}:10001 --objstore-address={}:20001 --worker-address={}:30001".format(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): + for node_ip_address in node_ip_addresses: + kill_cluster_command = "killall scheduler objstore python > /dev/null 2> /dev/null" + run_command_over_ssh(node_ip_address, username, key_file, kill_cluster_command) + +# Returns true if address is a valid IPv4 address and false otherwise. +def is_valid_ip(ip_address): + try: + socket.inet_aton(ip_address) + return True + except socket.error: + return False + +if __name__ == "__main__": + args = parser.parse_args() + username = args.username + key_file = args.key_file + installation_directory = args.installation_directory + node_ip_addresses = map(lambda s: str(s.strip()), open(args.nodes).readlines()) + for index, node_ip_address in enumerate(node_ip_addresses): + if not is_valid_ip(node_ip_address): + print "\nWARNING: The string '{}' from line {} in the file {} is not a valid IP address.\n".format(node_ip_address, index + 1, args.nodes) + + def install_ray(node_ip_addresses): + install_ray_multi_node(node_ip_addresses, username, key_file, installation_directory) + + def start_ray(node_ip_addresses, worker_path): + start_ray_multi_node(node_ip_addresses, username, key_file, worker_path, installation_directory) + + def stop_ray(node_ip_addresses): + stop_ray_multi_node(node_ip_addresses, username, key_file) + + IPython.embed() diff --git a/scripts/shell.py b/scripts/shell.py new file mode 100644 index 000000000..c867efb1a --- /dev/null +++ b/scripts/shell.py @@ -0,0 +1,21 @@ +import argparse +import numpy as np + +import ray +import ray.services as services +import ray.worker as worker + +import ray.arrays.remote as ra +import ray.arrays.distributed as da + +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:30001", 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) + + import IPython + IPython.embed() diff --git a/scripts/start_workers.py b/scripts/start_workers.py new file mode 100644 index 000000000..52d4ef30d --- /dev/null +++ b/scripts/start_workers.py @@ -0,0 +1,15 @@ +import argparse +from ray.services import start_node +import time + +parser = argparse.ArgumentParser(description="Starting workers on a node of the cluster (invoked locally on the node).") +parser.add_argument("--scheduler-address", type=str, help="Address of the scheduler running on the head node (ip + port).") +parser.add_argument("--node-ip", type=str, help="IP address of the current worker.") +parser.add_argument("--num-workers", type=int, default=20, help="Number of workers to be started on the node.") +parser.add_argument("--worker-path", type=str, help="Path to the worker file.") + +if __name__ == "__main__": + args = parser.parse_args() + start_node(args.scheduler_address, args.node_ip, args.num_workers, worker_path=args.worker_path) + + time.sleep(1000000000) # TODO(pcm): Figure out why object store file handle is closed if we don't do this