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

Documentation for using Ray on a cluster. (#165)

Browse Source
This commit is contained in:
Robert Nishihara
2016-12-30 00:29:03 -08:00
committed by Philipp Moritz
parent 13ee0ef366
commit 84296c8905
9 changed files with 109 additions and 632 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1
View File
@@ -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
doc/about-the-system.md
-32
View File
@@ -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.
doc/cluster-api.rst
-10
View File
@@ -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
doc/install-on-macosx.md
+9 -8
View File
@@ -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.
doc/install-on-ubuntu.md
+9 -8
View File
@@ -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
doc/reference-counting.md
-68
View File
@@ -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.
doc/scheduler.md
-25
View File
@@ -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).
doc/using-ray-on-a-cluster.md
+90 -153
View File
@@ -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
`<redis-address>` 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 <keypair> -i <key-file> -s <num-slaves> launch
<cluster-name>`, where `<keypair>` is the name of your EC2 key pair (that you
gave it when you created it), `<key-file>` is the private key file for your key
pair, `<num-slaves>` is the number of slave nodes to launch (try 1 at first),
and `<cluster-name>` is the name to give to your cluster.
```
./ray/scripts/start_ray.sh --redis-address <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=<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=<ec2-region>` specifies an EC2 region in which to launch instances.
The default region is `us-east-1`.
- `--zone=<ec2-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=<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=<num-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="<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 `<public-ip-address>, <private-ip-address>` 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 `<ip-address>` in
`nodes.txt` with the command
```
ssh -i <key-file> <username>@<ip-address>
```
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="<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)])
```
scripts/cluster.py
-328
View File
@@ -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()