diff --git a/README.md b/README.md index 013f13761..bac386836 100644 --- a/README.md +++ b/README.md @@ -55,6 +55,7 @@ estimate of pi (waiting until the computation has finished if necessary). - [Serialization in the Object Store](doc/serialization.md) - [Reusable Variables](doc/reusable-variables.md) - [Using Ray with TensorFlow](doc/using-ray-with-tensorflow.md) + - [Using Ray on a Cluster](doc/using-ray-on-a-cluster.md) ## Example Applications diff --git a/doc/about-the-system.md b/doc/about-the-system.md deleted file mode 100644 index 9bc2611a5..000000000 --- a/doc/about-the-system.md +++ /dev/null @@ -1,32 +0,0 @@ -# About the System - -This document describes the current architecture of Ray. However, some of these -decisions are likely to change. - -## Components - -A Ray cluster consists of several components. - -- One scheduler -- Multiple workers per node -- One object store per node -- One (or more) drivers - -### The scheduler - -The scheduler assigns tasks to the workers. - -### The workers - -The workers execute tasks and submit tasks to the scheduler. - -### The object store - -The object store shares objects between the worker processes on the same node so -that the workers don't need to each have their own copies of the objects. - -### The driver - -The driver submits tasks to the scheduler. If you use Ray in a script, the -Python process running the script is the driver. If you use Ray interactively -through a shell, the shell process is the driver. diff --git a/doc/cluster-api.rst b/doc/cluster-api.rst deleted file mode 100644 index cf11ee6b0..000000000 --- a/doc/cluster-api.rst +++ /dev/null @@ -1,10 +0,0 @@ -=============== -The Cluster API -=============== - -.. automethod:: cluster.RayCluster.install_ray -.. automethod:: cluster.RayCluster.start_ray -.. automethod:: cluster.RayCluster.stop_ray -.. automethod:: cluster.RayCluster.restart_workers -.. automethod:: cluster.RayCluster.update_ray -.. automethod:: cluster.RayCluster.run_command_over_ssh_on_all_nodes_in_parallel diff --git a/doc/install-on-macosx.md b/doc/install-on-macosx.md index 481cb0df4..93efc2f77 100644 --- a/doc/install-on-macosx.md +++ b/doc/install-on-macosx.md @@ -1,6 +1,6 @@ # Installation on Mac OS X -Ray should work with Python 2. We have tested Ray on OS X 10.11. +Ray should work with Python 2 and Python 3. We have tested Ray on OS X 10.11. ## Dependencies @@ -19,13 +19,7 @@ pip install --upgrade --verbose "git+git://github.com/ray-project/ray.git#egg=nu # Install Ray -Ray can be installed with pip as follows. - -``` -pip install --upgrade --verbose "git+git://github.com/ray-project/ray.git#egg=ray&subdirectory=lib/python" -``` - -Alternatively, Ray can be built from the repository as follows. +Ray can be built from the repository as follows. ``` git clone https://github.com/ray-project/ray.git @@ -33,6 +27,13 @@ cd lib/python python setup.py install ``` +Alternatively, Ray can be installed with pip as follows. However, this is +slightly less likely to succeed. + +``` +pip install --upgrade --verbose "git+git://github.com/ray-project/ray.git#egg=ray&subdirectory=lib/python" +``` + ## Test if the installation succeeded To test if the installation was successful, try running some tests. This assumes that you've cloned the git repository. diff --git a/doc/install-on-ubuntu.md b/doc/install-on-ubuntu.md index 4ab6468ff..f4d702923 100644 --- a/doc/install-on-ubuntu.md +++ b/doc/install-on-ubuntu.md @@ -1,6 +1,6 @@ # Installation on Ubuntu -Ray should work with Python 2. We have tested Ray on Ubuntu 14.04 +Ray should work with Python 2 and Python 3. We have tested Ray on Ubuntu 14.04 and Ubuntu 16.04 ## Dependencies @@ -19,13 +19,7 @@ pip install --upgrade --verbose "git+git://github.com/ray-project/ray.git#egg=nu # Install Ray -Ray can be installed with pip as follows. - -``` -pip install --upgrade --verbose "git+git://github.com/ray-project/ray.git#egg=ray&subdirectory=lib/python" -``` - -Alternatively, Ray can be built from the repository as follows. +Ray can be built from the repository as follows. ``` git clone https://github.com/ray-project/ray.git @@ -33,6 +27,13 @@ cd lib/python python setup.py install ``` +Alternatively, Ray can be installed with pip as follows. However, this is +slightly less likely to succeed. + +``` +pip install --upgrade --verbose "git+git://github.com/ray-project/ray.git#egg=ray&subdirectory=lib/python" +``` + ## Test if the installation succeeded To test if the installation was successful, try running some tests. This assumes diff --git a/doc/reference-counting.md b/doc/reference-counting.md deleted file mode 100644 index 3731f4554..000000000 --- a/doc/reference-counting.md +++ /dev/null @@ -1,68 +0,0 @@ -# Reference Counting - -In Ray, each object is assigned a globally unique object ID by the -scheduler (starting with 0 and incrementing upward). The objects are stored in -object stores. In order to avoid running out of memory, the object stores must -know when it is ok to deallocate an object. Since a worker on one node may have -an object ID for an object that lives in an object store on a different -node, knowing when we can safely deallocate an object requires cluster-wide -information. - -## Reference Counting - -Two approaches to reclaiming memory are garbage collection and reference -counting. We choose to use a reference counting approach in Ray. There are a -couple of reasons for this. Reference counting allows us to reclaim memory as -early as possible. It also avoids pausing the system for garbage collection. We -also note that implementing reference counting at the cluster level plays nicely -with worker processes that use reference counting internally (currently our -worker processes are Python processes). However, this could be made to work with -worker processes that use garbage collection, for example, if each worker -process is a Java Virtual Machine. - -At a high level, the scheduler keeps track of the number of object IDs -that exist on the cluster for each object ID. When the number of object -references reaches 0 for a particular object, the scheduler notifies all of the -object stores that contain that object to deallocate it. - -object IDs can exist in several places. - -1. They can be Python objects on a worker. -2. They can be serialized within an object in an object store. -3. They can be in a message being sent between processes (e.g., as an argument -to a remote procedure call). - -## When to Increment and Decrement the Reference Count - -We handle these three cases by calling the SchedulerService methods -`IncrementRefCount` and `DecrementRefCount` as follows: - -1. To handle the first case, we increment in the ObjectID constructor and -decrement in the ObjectID destructor. -2. To handle the second case, when an object is written to an object store with -a call to `put_object`, we call `IncrementRefCount` for each object ID -that is contained internally in the serialized object (for example, if we -serialize a `DistArray`, we increment the reference counts for its blocks). This -will notify the scheduler that those object IDs are in the object store. -Then when the scheduler deallocates the object, we call `DecrementRefCount` for -the object IDs that it holds internally (the scheduler keeps track of -these internal object IDs in the `contained_objectids_` data structure). -3. To handle the third case, we increment in the `serialize_task` method and -decrement in the `deserialize_task` method. - -## Complications -The following problem has not yet been resolved. In the following code, the -result `x` will be garbage. -```python -x = ray.get(ra.zeros([10, 10], "float")) -``` -When `ra.zeros` is called, a worker will create an array of zeros and store -it in an object store. An object ID to the output is returned. The call -to `ray.get` will not copy data from the object store process to the worker -process, but will instead give the worker process a pointer to shared memory. -After the `ray.get` call completes, the object ID returned by -`ra.zeros` will go out of scope, and the object it refers to will be -deallocated from the object store. This will cause the memory that `x` points to -to be garbage. - -This problem is currently unresolved. diff --git a/doc/scheduler.md b/doc/scheduler.md deleted file mode 100644 index 97187619e..000000000 --- a/doc/scheduler.md +++ /dev/null @@ -1,25 +0,0 @@ -# Scheduler - -The scheduling strategies currently implemented in Ray are fairly basic and -all use a central scheduler. - -* The naive scheduler assigns tasks to workers just taking into account -dependencies between tasks (no other information like data locality). It is -supposed to be an example for how to write a scheduler. We do not recommend -its use and it only works well for single node setups. Tasks are assigned in the -following way: For each idle worker, we iterate over the tasks in the task -queue. The first task that has all its requirements satisfied will be scheduled -on the worker. - -* The locality aware scheduler is more suited for multi node setups, but still -inappropriate for very large clusters. This is because the computational -overhead for each scheduling decision is O(mn) where m is the number of idle -workers and n is the number of tasks in the task queue. For each idle worker, -all tasks in the task queue are considered and the one that requires the -smallest number of objects to be shipped will be executed. - -We expect to implement more refined scheduling strategies in the future, -including more computationally efficient location aware scheduling, -scheduling that takes into account sizes of the shipped objects, and strategies -that do not require a central scheduler (which is a bottleneck for large -clusters). diff --git a/doc/using-ray-on-a-cluster.md b/doc/using-ray-on-a-cluster.md index 623534061..a9d426c25 100644 --- a/doc/using-ray-on-a-cluster.md +++ b/doc/using-ray-on-a-cluster.md @@ -1,185 +1,122 @@ # Using Ray on a cluster -Running Ray on a cluster is still experimental. +Deploying Ray on a cluster currently requires a bit of manual work. -Ray can be used in several ways. In addition to running on a single machine, Ray -is designed to run on a cluster of machines. This document is about how to use -Ray on a cluster. +## Deploying Ray on a cluster. -## Launching a cluster on EC2 +This section assumes that you have a cluster running and that the node in the +cluster can communicate with each other. It also assumes that Ray is installed +on each machine. To install Ray, follow the instructions for [installation on +Ubuntu](install-on-ubuntu.md). -This section describes how to start a cluster on EC2. These instructions are -copied and adapted from https://github.com/amplab/spark-ec2. +### Starting Ray on each machine. -### Before you start +On the head node (just choose some node to be the head node), run the following. -- Create an Amazon EC2 key pair for yourself. This can be done by logging into -your Amazon Web Services account through the [AWS -console](http://aws.amazon.com/console/), clicking Key Pairs on the left -sidebar, and creating and downloading a key. Make sure that you set the -permissions for the private key file to `600` (i.e. only you can read and write -it) so that `ssh` will work. -- Whenever you want to use the `ec2.py` script, set the environment variables -`AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY` to your Amazon EC2 access key ID -and secret access key. These can be generated from the [AWS -homepage](http://aws.amazon.com/) by clicking My Account > Security Credentials > -Access Keys, or by [creating an IAM user](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_users_create.html). +``` +./ray/scripts/start_ray.sh --head +``` -### Launching a Cluster +This will print out the address of the Redis server that was started (and some +other address information). -- Install the required dependencies on the machine you will be using to run the -cluster launch scripts. - ``` - sudo pip install --upgrade boto - ``` +Then on all of the other nodes, run the following. Make sure to replace +`` with the value printed by the command on the head node (it +should look something like `123.45.67.89:12345`). -- Go into the `ray/scripts` directory. -- Run `python ec2.py -k -i -s launch -`, where `` is the name of your EC2 key pair (that you -gave it when you created it), `` is the private key file for your key -pair, `` is the number of slave nodes to launch (try 1 at first), -and `` is the name to give to your cluster. +``` +./ray/scripts/start_ray.sh --redis-address +``` - For example: +Now we've started all of the Ray processes on each node Ray. This includes - ```bash - export AWS_SECRET_ACCESS_KEY=AaBbCcDdEeFGgHhIiJjKkLlMmNnOoPpQqRrSsTtU - export AWS_ACCESS_KEY_ID=ABCDEFG1234567890123 - python ec2.py --key-pair=awskey \ - --identity-file=awskey.pem \ - --region=us-west-1 \ - --instance-type=c4.4xlarge \ - --spot-price=2.50 \ - --slaves=1 \ - launch my-ray-cluster - ``` +- Some worker processes on each machine. +- An object store on each machine. +- A local scheduler on each machine. +- One Redis server (on the head node). +- One global scheduler (on the head node). -The following options are worth pointing out: +Later when you want to stop the Ray processes, run `./ray/scripts/stop_ray.sh` +on each node. -- `--instance-type=` can be used to specify an EC2 instance type -to use. For now, the script only supports 64-bit instance types, and the default -type is `m3.large` (which has 2 cores and 7.5 GB RAM). -- `--region=` specifies an EC2 region in which to launch instances. -The default region is `us-east-1`. -- `--zone=` can be used to specify an EC2 availability zone to launch -instances in. Sometimes, you will get an error because there is not enough -capacity in one zone, and you should try to launch in another. -- `--spot-price=` will launch the worker nodes as [Spot -Instances](http://aws.amazon.com/ec2/spot-instances/), bidding for the given -maximum price (in dollars). -- `--slaves=` will launch a cluster with `(1 + num_slaves)` instances. -The first instance is the head node, which in addition to hosting workers runs the -Ray scheduler and application driver programs. +That should start up all of the Ray processes. To run some commands, start up +Python on one of the nodes in the cluster, and do the following. -## Getting started with Ray on a cluster +```python +import ray +ray.init(redis_address="") +``` -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. +Now you can define remote functions and execute tasks. For example: -1. If you launched a cluster using the `ec2.py` script from the previous -section, then the file `ray/scripts/nodes.txt` will already have been created. -Otherwise, create a file `nodes.txt` of the IP addresses of the nodes in the -cluster. For example +```python +@ray.remote +def f(x): + return x - 12.34.56.789 - 12.34.567.89 -The first node in the file is the "head" node. The scheduler will be started on -the head node, and the driver should run on the head node as well. If the nodes -have public and private IP addresses (as in the case of EC2 instances), you can -list the `, ` in `nodes.txt` like +ray.get([f.remote(f.remote(f.remote(0))) for _ in range(1000)]) +``` - 12.34.56.789, 98.76.54.321 - 12.34.567.89, 98.76.543.21 -The `cluster.py` administrative script will use the public IP addresses to ssh -to the nodes. Ray will use the private IP addresses to send messages between the -nodes during execution. +### Copying Application Files to Other Nodes (Experimental) -2. Make sure that the nodes can all communicate with one another. On EC2, this -can be done by creating a new security group with the appropriate inbound and -outbound rules and adding all of the nodes in your cluster to that security -group. This is done automatically by the `ec2.py` script. If you have used the -`ec2.py` script you can log into the hosts with the username `ubuntu`. +If you're running an application that imports Python files that are present +locally but not on the other machines in the cluster, you may first want to copy +those files to the other machines. One way to do that is through Ray (this is +experimental). Suppose you're directory structure looks -3. From the `ray/scripts` directory, run something like +``` +application_files/ + __init__.py + example.py +``` - ``` - python cluster.py --nodes=nodes.txt \ - --key-file=awskey.pem \ - --username=ubuntu - ``` -where you replace `nodes.txt`, `key.pem`, and `ubuntu` by the appropriate -values. This assumes that you can connect to each IP address `` in -`nodes.txt` with the command - ``` - ssh -i @ - ``` -4. The previous command should open a Python interpreter. To install Ray on the -cluster, run `cluster.install_ray()` in the interpreter. The interpreter should -block until the installation has completed. The standard output from the nodes -will be redirected to your terminal. -5. To check that the installation succeeded, you can ssh to each node and run -the tests. - ``` - cd $HOME/ray/ - source setup-env.sh # Add Ray to your Python path. - python test/runtest.py # This tests basic functionality. - python test/array_test.py # This tests some array libraries. - ``` +And suppose `example.py` defines the following functions. -6. Start the cluster with `cluster.start_ray()`. The `cluster.start_ray` command -will start the Ray scheduler, object stores, and workers, and before finishing -it will print instructions for connecting to the cluster via ssh. +```python +import ray -7. To connect to the cluster (either with a Python shell or with a script), ssh -to the cluster's head node (as described by the output of the -`cluster.start_ray` command. E.g., - ``` - The cluster has been started. You can attach to the cluster by sshing to the head node with the following command. +def example_helper(x): + return x - ssh -i awskey.pem ubuntu@12.34.56.789 +@ray.remote +def example_function(x): + return example_helper(x) +``` - Then run the following commands. +If you simply run - source $HOME/ray/setup-env.sh # Add Ray to your Python path. +```python +from application_files import example +``` - Then within a Python interpreter, run the following commands. +An error message will be printed like the following. This indicates that one of +the workers was unable to register the remote function. - import ray - ray.init(node_ip_address="98.76.54.321", scheduler_address="98.76.54.321:10001") - ``` +``` +Traceback (most recent call last): + File "/home/ubuntu/ray/lib/python/ray/worker.py", line 813, in fetch_and_register_remote_function + function = pickling.loads(serialized_function) +ImportError: No module named 'application_files' +``` -8. If you would like to run the example applications on the cluster. You will -need to install a few more Python packages. This can be done, within -`cluster.py`, by running the following. - ```python - install_example_dependencies_command = """ - # Install TensorFlow - sudo pip install --upgrade https://storage.googleapis.com/tensorflow/linux/cpu/tensorflow-0.9.0-cp27-none-linux_x86_64.whl; - # Install SciPy - sudo apt-get -y install libatlas-base-dev gfortran; - sudo pip install scipy; - # Install Gym - sudo apt-get -y install zlib1g-dev libjpeg-dev xvfb libav-tools xorg-dev python-opengl libsdl2-dev swig wget; - sudo pip install gym[atari] - """ - cluster.run_command_over_ssh_on_all_nodes_in_parallel(install_example_dependencies_command) - ``` +To make this work, you need to copy your application files to all of the nodes. +The following command will do that through Ray, and will add the files to Python +path of each worker. This functionality is experimental. You may be able to do +something like the following. -9. Note that there are several more commands that can be run from within -`cluster.py`. +```python +import ray - - `cluster.install_ray()` - This pulls the Ray source code on each node, - builds all of the third party libraries, and builds the project itself. - - `cluster.start_ray(num_workers_per_node=10)` - This starts a scheduler - process on the head node, and it starts an object store and some workers - on each node. - - `cluster.stop_ray()` - This shuts down the cluster (killing all of the - processes). - - `cluster.copy_code_to_cluster(user_source_directory)` - This copies the - contents of `user_source_directory` locally to the cluster under - `~/ray_source_files/`. - - `cluster.update_ray()` - This pulls the latest Ray source code and builds - it. - - `cluster.run_command_over_ssh_on_all_nodes_in_parallel(command)` - This - will ssh to each node in the cluster and run a command. +ray.init(redis_address="") + +ray.experimental.copy_directory("application_files/") + +# Now the import should work. +from application_files import example +``` + +Now you should be able to run the following command. + +```python +ray.get([example.example_function.remote(0) for _ in range(1000)]) +``` diff --git a/scripts/cluster.py b/scripts/cluster.py deleted file mode 100644 index 6be24e71f..000000000 --- a/scripts/cluster.py +++ /dev/null @@ -1,328 +0,0 @@ -from __future__ import absolute_import -from __future__ import division -from __future__ import print_function - -import os -import subprocess -import socket -import argparse -import threading -import IPython -import numpy as np - -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.") - -class RayCluster(object): - """A class for setting up, starting, and stopping Ray on a cluster. - - Attributes: - node_ip_addresses (List[str]): A list of the ip addresses of the nodes in - the cluster. The first element is the head node and will host the - scheduler process. - username (str): The username used to ssh to nodes in the cluster. - key_file (str): The path to the key used to ssh to nodes in the cluster. - installation_directory (str): The path on the nodes in the cluster to the - directory in which Ray should be installed. - """ - - def __init__(self, node_ip_addresses, node_private_ip_addresses, username, key_file, installation_directory): - """Initialize the RayCluster object. - - Args: - node_ip_addresses (List[str]): A list of the ip addresses of the nodes in - the cluster. The first element is the head node and will host the - scheduler process. - node_private_ip_addresses (List[str]): A list of the ip addresses that the - nodes use internally to connect to one another. We include this because - on EC2 communication within a security group must be done over private - ip addresses. - username (str): The username used to ssh to nodes in the cluster. - key_file (str): The path to the key used to ssh to nodes in the cluster. - installation_directory (str): The path on the nodes in the cluster to the - directory in which Ray should be installed. - - Raises: - Exception: An exception is raised by check_ip_addresses if one of the ip - addresses is not a valid ip address. - """ - _check_ip_addresses(node_ip_addresses) - self.node_ip_addresses = node_ip_addresses - self.node_private_ip_addresses = node_private_ip_addresses - self.username = username - self.key_file = key_file - self.installation_directory = installation_directory - - def _run_command_over_ssh(self, node_ip_address, command): - """Run a command over ssh. - - Args: - node_ip_address (str): The ip address of the node to ssh to. - command (str): The command to run over ssh, currently this command is not - allowed to have any single quotes. - """ - if "'" in command: - raise Exception("Commands run over ssh must not contain the single quote character. This command does: {}".format(command)) - full_command = "ssh -o StrictHostKeyChecking=no -i {} {}@{} '{}'".format(self.key_file, self.username, node_ip_address, command) - subprocess.call([full_command], shell=True) - print("Finished running command '{}' on {}@{}.".format(command, self.username, node_ip_address)) - - def _run_parallel_functions(self, functions, inputs): - """Run functions in parallel. - - This will run each function in functions in a separate thread. This method - blocks until all of the functions have finished. - - Args: - functions (List[Callable]): The functions to execute in parallel. - inputs (List[Tuple]): The inputs to the functions. - """ - threads = [] - for i in range(len(self.node_ip_addresses)): - t = threading.Thread(target=functions[i], args=inputs[i]) - t.start() - threads.append(t) - for t in threads: - t.join() - - def run_command_over_ssh_on_all_nodes_in_parallel(self, command): - """Run a command over ssh on all nodes in the cluster in parallel. - - Args: - command: This is either a single command to run on every node in the - cluster over ssh, or it is a list of commands of the same length as - node_ip_addresses, in which case the ith command will be run on the ith - element of node_ip_addresses. Currently this command is not allowed to - have any single quotes. - - Raises: - Exception: An exception is raised if command is not a string or is not a - list with the same length as node_ip_addresses. - """ - if isinstance(command, str): - # If there is only one command, then run this command on every node in the - # cluster. - commands = len(self.node_ip_addresses) * [command] - else: - # Otherwise, there is a list of one command for each node in the cluster. - commands = command - # Make sure we have one command for each node. - if len(commands) != len(self.node_ip_addresses): - raise Exception("The number of commands must match the number of nodes.") - # Make sure that the commands do not contain any single quotes. - for command in commands: - if "'" in command: - raise Exception("Commands run over ssh must not contain the single quote character. This command does: {}".format(command)) - functions = [] - inputs = [] - def function(node_ip_address, command): - self._run_command_over_ssh(node_ip_address, command) - inputs = zip(node_ip_addresses, commands) - self._run_parallel_functions(len(self.node_ip_addresses) * [function], inputs) - print("Finished running commands {} on all nodes.".format(inputs)) - - def install_ray(self): - """Install Ray on every node in the cluster. - - This method will ssh to each node, clone the Ray repository, install the - dependencies, build the third-party libraries, and build Ray. - """ - install_ray_command = """ - sudo apt-get update && - sudo apt-get -y install git && - mkdir -p "{}" && - cd "{}" && - git clone "https://github.com/ray-project/ray"; - cd ray; - ./install-dependencies.sh; - ./setup.sh; - ./build.sh - """.format(self.installation_directory, self.installation_directory) - self.run_command_over_ssh_on_all_nodes_in_parallel(install_ray_command) - - def start_ray(self, num_workers_per_node=10): - """Start Ray on a cluster. - - This method is used to start Ray on a cluster. It will ssh to the head node, - that is, the first node in the list node_ip_addresses, and it will start the - scheduler. Then it will ssh to each node and start an object store and some - workers. - - Args: - num_workers_per_node (int): The number workers to start on each node. - """ - scripts_directory = os.path.join(self.installation_directory, "ray/scripts") - # Start the scheduler - # The triple backslashes are used for two rounds of escaping, something like \\\" -> \" -> " - start_scheduler_command = """ - cd "{}"; - source ../setup-env.sh; - python -c "import ray; ray.services.start_scheduler(\\\"{}:10001\\\", cleanup=False)" > start_scheduler.out 2> start_scheduler.err < /dev/null & - """.format(scripts_directory, self.node_private_ip_addresses[0]) - self._run_command_over_ssh(self.node_ip_addresses[0], start_scheduler_command) - - # Start the workers on each node - # The triple backslashes are used for two rounds of escaping, something like \\\" -> \" -> " - start_workers_commands = [] - for i, node_ip_address in enumerate(self.node_ip_addresses): - start_workers_command = """ - cd "{}"; - source ../setup-env.sh; - python -c "import ray; ray.services.start_node(\\\"{}:10001\\\", \\\"{}\\\", {})" > start_workers.out 2> start_workers.err < /dev/null & - """.format(scripts_directory, self.node_private_ip_addresses[0], self.node_private_ip_addresses[i], num_workers_per_node) - start_workers_commands.append(start_workers_command) - self.run_command_over_ssh_on_all_nodes_in_parallel(start_workers_commands) - - setup_env_path = os.path.join(self.installation_directory, "ray/setup-env.sh") - print(""" - The cluster has been started. You can attach to the cluster by sshing to the head node with the following command. - - ssh -i {} {}@{} - - Then run the following commands. - - source {} # Add Ray to your Python path. - - Then within a Python interpreter or script, run the following commands. - - import ray - ray.init(node_ip_address="{}", scheduler_address="{}:10001") - """.format(self.key_file, self.username, self.node_ip_addresses[0], setup_env_path, self.node_private_ip_addresses[0], self.node_private_ip_addresses[0])) - - def stop_ray(self): - """Kill all of the processes in the Ray cluster. - - This method is used for stopping a Ray cluster. It will ssh to each node and - kill every schedule, object store, and Python process. - """ - kill_cluster_command = "killall scheduler objstore python > /dev/null 2> /dev/null" - self.run_command_over_ssh_on_all_nodes_in_parallel(kill_cluster_command) - - def update_ray(self, branch=None): - """Pull the latest Ray source code and rebuild Ray. - - This method is used for updating the Ray source code on a Ray cluster. It - will ssh to each node, will pull the latest source code from the Ray - repository, and will rerun the build script (though currently it will not - rebuild the third party libraries). - - Args: - branch (Optional[str]): The branch to check out. If omitted, then stay on - the current branch. - """ - ray_directory = os.path.join(self.installation_directory, "ray") - change_branch_command = "git checkout -f {}".format(branch) if branch is not None else "" - update_cluster_command = """ - cd "{}" && - git fetch && - {} - git reset --hard "@{{upstream}}" -- && - (make -C "./build" clean || rm -rf "./build") && - ./build.sh - """.format(ray_directory, change_branch_command) - self.run_command_over_ssh_on_all_nodes_in_parallel(update_cluster_command) - - def copy_code_to_cluster(self, user_source_directory): - """Update the user's source code on each node in the cluster. - - This method is used to copy the user's source code on each node in the - cluster. The local user_source_directory will be copied under - ray_source_files in the home directory on the worker node. For example, if - we call copy_code_to_cluster("~/a/b/c"), then the contents of "~/a/b/c" on - the local machine will be copied to "~/ray_source_files/c" on each node in - the cluster. - - Args: - user_source_directory (str): The path on the local machine to the directory - that contains the worker code. - - Returns: - A string with the path to the source code of the worker on the remote - nodes. - """ - user_source_directory = os.path.expanduser(user_source_directory) - if not os.path.isdir(user_source_directory): - raise Exception("Directory {} does not exist.".format(user_source_directory)) - # If user_source_directory is "/a/b/c", then local_directory_name is "c". - local_directory_name = os.path.split(os.path.realpath(user_source_directory))[1] - remote_directory = os.path.join(self.installation_directory, "ray_source_files", local_directory_name) - # Remove and recreate the directory on the node. - recreate_directory_command = """ - rm -r "{}"; - mkdir -p "{}" - """.format(remote_directory, remote_directory) - self.run_command_over_ssh_on_all_nodes_in_parallel(recreate_directory_command) - # Copy the files from the local machine to the node. - def copy_function(node_ip_address): - copy_command = """ - scp -r -i {} {}/* {}@{}:{}/ - """.format(self.key_file, user_source_directory, self.username, node_ip_address, remote_directory) - subprocess.call([copy_command], shell=True) - inputs = [(node_ip_address,) for node_ip_address in node_ip_addresses] - self._run_parallel_functions(len(self.node_ip_addresses) * [copy_function], inputs) - # Return the source directory path on the remote nodes - return remote_directory - -def _is_valid_ip(ip_address): - """Check if ip_addess is a valid IPv4 address. - - Args: - ip_address (str): The ip address to check. - - Returns: - True if the address is a valid IPv4 address and False otherwise. - """ - try: - socket.inet_aton(ip_address) - return True - except socket.error: - return False - -def _check_ip_addresses(node_ip_addresses): - """Check if a list of ip addresses are all valid IPv4 addresses. - - This method checks if all of the addresses in a list are valid IPv4 address. - It prints an error message for each invalid address. - - Args: - node_ip_addresses (List[str]): The list of ip addresses to check. - - Raises: - Exception: An exception is raisd if one of the addresses is not a valid IPv4 - address. - """ - for i, node_ip_address in enumerate(node_ip_addresses): - if not _is_valid_ip(node_ip_address): - raise Exception("node_ip_addresses[{}] is '{}', which is not a valid IP address.".format(i, node_ip_address)) - -if __name__ == "__main__": - args = parser.parse_args() - username = args.username - key_file = args.key_file - node_ip_addresses = [] - node_private_ip_addresses = [] - # Check if the IP addresses in the nodes file are valid. - for line in open(args.nodes).readlines(): - parts = line.split(",") - ip_address = str(parts[0].strip()) - if len(parts) == 1: - private_ip_address = ip_address - elif len(parts) == 2: - private_ip_address = str(parts[1].strip()) - else: - raise Exception("Each line in the nodes file must have either one or two ip addresses.") - node_ip_addresses.append(ip_address) - node_private_ip_addresses.append(private_ip_address) - # This command finds the home directory on the cluster. That directory will be - # used for installing Ray. Note that single quotes around 'echo $HOME' are - # important. If you use double quotes, then the $HOME environment variable - # will be expanded locally instead of remotely. - echo_home_command = "ssh -o StrictHostKeyChecking=no -i {} {}@{} 'echo $HOME'".format(key_file, username, node_ip_addresses[0]) - installation_directory = subprocess.check_output(echo_home_command, shell=True).strip() - print("Using '{}' as the home directory on the cluster.".format(installation_directory)) - # Create the Raycluster object. - cluster = RayCluster(node_ip_addresses, node_private_ip_addresses, username, key_file, installation_directory) - # Drop into an IPython shell. - IPython.embed()