diff --git a/README.md b/README.md index a0c64081e..7302c53ae 100644 --- a/README.md +++ b/README.md @@ -5,83 +5,4 @@ Ray is an experimental distributed execution framework with a Python-like programming model. It is under development and not ready for general use. -## Example Code - -### Loading ImageNet -TODO: fill this out. - -## Design Decisions - -For a description of our design decisions, see - -- [Reference Counting](doc/reference-counting.md) -- [Aliasing](doc/aliasing.md) -- [Scheduler](doc/scheduler.md) - -## Setup - -### Linux, Mac, and other Unix-based systems - -After running these instruction, add the line `source "$RAY_ROOT/setup-env.sh"` in your `~/.bashrc` file manually, where "$RAY_ROOT" is the path of the directory containing `setup-env.sh`. - -1. sudo apt-get update -2. sudo apt-get install git -3. git clone https://github.com/amplab/ray.git -4. cd ray -5. ./setup.sh -6. ./build.sh -7. source setup-env.sh - -### Windows - -**Note:** A batch file is provided that clones any missing third-party libraries and applies patches to them. -Do not attempt to open the solution before the batch file applies the patches; otherwise, if the projects have been modified, the patches may be rejected, and you may be forced to revert your changes before re-running the batch file. - -1. Install Microsoft Visual Studio 2015 -2. Install Git -3. git clone https://github.com/amplab/ray.git -4. ray\thirdparty\download_thirdparty.bat - -## 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()` 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("/home/ubuntu/ray/scripts/default_worker.py")`, where the argument is -the path on each node in the cluster to the worker code that you would like to -use. The workers can be restarted with -`restart_workers("/home/ubuntu/ray/scripts/default_worker.py")`, for example if -you wish to update the application code running on the workers. The cluster -processes (the scheduler, the object stores, and the workers) can be stopped -with `stop_ray()`. +Read this [introduction to Ray](doc/introduction.md). diff --git a/doc/about-the-system.md b/doc/about-the-system.md new file mode 100644 index 000000000..52b7eb164 --- /dev/null +++ b/doc/about-the-system.md @@ -0,0 +1,32 @@ +## 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/basic-usage.md b/doc/basic-usage.md new file mode 100644 index 000000000..feeeb09e2 --- /dev/null +++ b/doc/basic-usage.md @@ -0,0 +1,207 @@ +## Basic Usage + +To use Ray, you need to understand the following: + +- How Ray uses object references to represent immutable remote objects. +- How Ray constructs computation graphs using remote functions. + +### Immutable remote objects + +In Ray, we can create and manipulate objects. We refer to these objects as +**remote objects**, and we use **object references** to refer to them. Remote +objects are stored in **object stores**, and there is one object store per node +in the cluster. In the cluster setting, we may not actually know which machine +each object lives on. + +An **object reference** is essentially a unique ID that can be used to refer to +a remote object. If you're familiar with Futures, our object references are +conceptually similar. + +We assume that remote objects are immutable. That is, their values cannot be +changed after creation. This allows remote objects to be replicated in multiple +object stores without needing to synchronize the copies. + +#### Put and Get + +The commands `ray.get` and `ray.put` can be used to convert between Python +objects and object references, as shown in the example below. +```python +>>> x = [1, 2, 3] +>>> ray.put(x) + +``` + +The command `ray.put(x)` would be run by a worker process or by the driver +process (the driver process is the one running your script). It takes a Python +object and copies it to the local object store (here *local* means *on the same +node*). Once the object has been stored in the object store, its value cannot be +changed. + +In addition, `ray.put(x)` returns an object reference, which is essentially an +ID that can be used to refer to the newly created remote object. If we save the +object reference in a variable with `ref = ray.put(x)`, then we can pass `ref` +into remote functions, and those remote functions will operate on the +corresponding remote object. + +The command `ray.get(ref)` takes an object reference and creates a Python object +from the corresponding remote object. For some objects like arrays, we can use +shared memory and avoid copying the object. For other objects, this currently +copies the object from the object store into the memory of the worker process. +If the remote object corresponding to the object reference `ref` does not live +on the same node as the worker that calls `ray.get(ref)`, then the remote object +will first be copied from an object store that has it to the object store that +needs it. +```python +>>> ref = ray.put([1, 2, 3]) +>>> ray.get(ref) +[1, 2, 3] +``` + +If the remote object corresponding to the object reference `ref` has not been +created yet, the command `ray.get(ref)` will wait until the remote object has +been created. + +### Computation graphs in Ray + +Ray represents computation with a directed acyclic graph of tasks. Tasks are +added to this graph by calling **remote functions**. + +For example, a normal Python function looks like this. +```python +def add(a, b): + return a + b +``` +A remote function in Ray looks like this. +```python +@ray.remote([int, int], [int]) +def add(a, b): + return a + b +``` + +The information passed to the `@ray.remote` decorator includes type information +for the arguments and for the return values of the function. Because of the +distinction that we make between *submitting a task* and *executing the task*, +we require type information so that we can catch type errors when the remote +function is called instead of catching them when the task is actually executed. + +However, the only piece of information that is fundamentally required by the +system is the number of return values (because the system must assign the +correct number of object references to the outputs before the function has +actually executed and produced any outputs). + +#### Remote functions + +Whereas in regular Python, calling `add(1, 2)` would return `3`, in Ray, calling +`add(1, 2)` does not actually execute the task. Instead, it adds a task to the +computation graph and immediately returns an object reference to the output of +the computation. + +```python +>>> ref = add(1, 2) +>>> ray.get(ref) +3 +``` + +There is a sharp distinction between *submitting a task* and *executing the +task*. When a remote function is called, the task of executing that function is +submitted to the scheduler, and the scheduler immediately returns object +references for the outputs of the task. However, the task will not be executed +until the scheduler actually schedules the task on a worker. + +When a task is submitted, each argument may be passed in by value or by object +reference. For example, these lines have the same behavior. + +```python +>>> add(1, 2) +>>> add(1, ray.put(2)) +>>> add(ray.put(1), ray.put(2)) +``` + +Remote functions never return actual values, they always return object +references. + +When the remote function is actually executed, it operates on Python objects. +That is, if the remote function was called with any object references, the +Python objects corresponding to those object references will be retrieved and +passed into the actual execution of the remote function. + +#### Blocking computation + +In a regular Python script, the specification of a computation is intimately +linked to the actual execution of the code. For example, consider the following +code. +```python +result = [] +for i in range(10): + result.append(np.zeros(size=[100, 100])) +``` + +At the core of the above script, there are 10 separate tasks, each of which +generates a 100x100 matrix of zeros. These tasks do not depend on each other, so +in principle, they could be executed in parallel. However, in the above +implementation, they will be executed serially. + +Ray gets around this by representing computation as a graph of tasks, where some +tasks depend on the outputs of other tasks and where tasks can be executed once +their dependencies have been executed. + +For example, suppose we define the remote function `zeros` to be a wrapper +around `np.zeros`. +```python +from typing import List +import numpy as np + +@ray.remote([List[int]], [np.ndarray]) +def zeros(shape): + return np.zeros(shape) +``` +Then we can write +```python +result_refs = [] +for i in range(10): + result.append(zeros([100, 100])) +``` +This adds 10 tasks to the computation graph, with no dependencies between the +tasks. + +The computation graph looks like this. + +

+ +

+ +In this figure, boxes are tasks and ovals are objects. + +The box that says `op-root` in it just refers to the overall script itself. The +dotted lines indicate that the script launched 10 tasks (tasks are denoted by +rectangular boxes). The solid lines indicate that each task produces one output +(represented by an oval). + +It is clear from the computation graph that these ten tasks can be executed in +parallel. + +Computation graphs encode dependencies. For example, suppose we define +```python +ray.remote([np.ndarray, np.ndarray], [np.ndarray]) +def dot(a, b): + return np.dot(a, b) +``` +Then we run +```python +aref = zeros([10, 10]) +bref = zeros([10, 10]) +cref = dot(aref, bref) +``` +The corresponding computation graph looks like this. + +

+ +

+ + +The three dashed lines indicate that the script launched three tasks (the two +`zeros` tasks and the one `dot` task). Each task produces a single output, and +the `dot` task depends on the outputs of the two `zeros` tasks. + +This makes it clear that the two `zeros` tasks can execute in parallel but that +the `dot` task must wait until the two `zeros` tasks have finished. diff --git a/doc/download-and-setup.md b/doc/download-and-setup.md new file mode 100644 index 000000000..2090abadb --- /dev/null +++ b/doc/download-and-setup.md @@ -0,0 +1,53 @@ +## Download and Setup + +Ray must currently be built from source. + +### Clone the Ray repository + +``` +git clone https://github.com/amplab/ray.git +``` +These instructions will install the latest master branch for Ray. + +### Installation for Ubuntu and Mac OS X + +For convenience, we provide a setup script that pulls the necessary +dependencies. + +``` +cd ray +./setup.sh # This builds all necessary third party libraries (e.g., gRPC and Apache Arrow). It will require a sudo password. +./build.sh # This builds Ray. +source setup-env.sh # This adds Ray to your Python path. +``` + +For convenience, you may also want to add the line `source +"$RAY_ROOT/setup-env.sh"` to your `~/.bashrc` file manually, where `$RAY_ROOT` +is the Ray directory (e.g., `/home/ubuntu/ray`). + +To test if the installation was successful, try running some tests. + +### Installation for Windows + +Ray currently does not run on Windows. However, it can be compiled with the +following instructions. + +**Note:** A batch file is provided that clones any missing third-party libraries +and applies patches to them. Do not attempt to open the solution before the +batch file applies the patches; otherwise, if the projects have been modified, +the patches may be rejected, and you may be forced to revert your changes before +re-running the batch file. + +1. Install Microsoft Visual Studio 2015 +2. Install Git +3. `git clone https://github.com/amplab/ray.git` +4. `ray\thirdparty\download_thirdparty.bat` + +### Test if the installation succeeded + +Try running some tests. + +``` +python test/runtest.py # This tests basic functionality. +python test/array_test.py # This tests some array libraries. +``` diff --git a/doc/figures/compgraph1.png b/doc/figures/compgraph1.png new file mode 100644 index 000000000..9abd43a36 Binary files /dev/null and b/doc/figures/compgraph1.png differ diff --git a/doc/figures/compgraph2.png b/doc/figures/compgraph2.png new file mode 100644 index 000000000..f05f7eeb5 Binary files /dev/null and b/doc/figures/compgraph2.png differ diff --git a/doc/figures/compgraph3.png b/doc/figures/compgraph3.png new file mode 100644 index 000000000..923096304 Binary files /dev/null and b/doc/figures/compgraph3.png differ diff --git a/doc/introduction.md b/doc/introduction.md new file mode 100644 index 000000000..d1900b9b4 --- /dev/null +++ b/doc/introduction.md @@ -0,0 +1,49 @@ +## Introduction + +The goal of Ray is to make it easy to write machine learning applications that +run on a cluster while providing the development and debugging experience of +working on a single machine. + +Before jumping into the details, here's a simple Python example for doing a +Monte Carlo estimation of pi (using multiple cores or potentially multiple +machines). +```python +import ray +import functions # See definition below + +results = [] +for _ in range(10): + results.append(functions.estimate_pi(100)) +estimate = np.mean([ray.get(ref) for ref in results]) +print "Pi is approximately {}.".format(estimate) +``` + +This assumes that we've defined the file `functions.py` as below. +```python +import ray +import numpy as np + +@ray.remote([int], [float]) +def estimate_pi(n): + x = np.random.uniform(size=n) + y = np.random.uniform(size=n) + return 4 * np.mean(x ** 2 + y ** 2 < 1) +``` + +Within the for loop, each call to `functions.estimate_pi(100)` sends a message +to the scheduler asking it to schedule the task of running +`functions.estimate_pi` with the argument `100`. This call returns right away +without waiting for the actual estimation of pi to take place. Instead of +returning a float, it returns an **object reference**, which represents the +eventual output of the computation. + +The call to `ray.get(ref)` takes an object reference and returns the actual +estimate of pi (waiting until the computation has finished if necessary). + +## Next Steps + +- [Download and Setup](download-and-setup.md) +- [Basic Usage](basic-usage.md) +- [Tutorial](tutorial.md) +- [About the System](about-the-system.md) +- [Using Ray on a Cluster](using-ray-on-a-cluster.md) diff --git a/doc/tutorial.md b/doc/tutorial.md new file mode 100644 index 000000000..4feb74f16 --- /dev/null +++ b/doc/tutorial.md @@ -0,0 +1,118 @@ +## Tutorial + +This section assumes that Ray has been built. See the [instructions for +installing Ray](download-and-setup.md) + +### Trying it out + +Start a shell by running this command. +``` +python scripts/shell.py +``` +By default, this will start up several things + +- 1 scheduler (for assigning tasks to workers) +- 1 object store (for sharing objects between worker processes) +- 10 workers (for executing tasks) +- 1 driver (for submitting tasks to the scheduler) + +Each of the above items, and each worker, is its own process. + +The shell that you just started is the driver process. + +You can take a Python object and store it in the object store using `ray.put`. +This turns it into a **remote object** (we are currently on a single machine, +but the terminology makes sense if we are on a cluster), and allows it to be +shared among the worker processes. The function `ray.put` returns an object +reference that is used to identify this remote object. + +```python +>>> xref = ray.put([1, 2, 3]) +``` + +We can use `ray.get` to retrieve the object corresponding to an object +reference. + +```python +>>> ray.get(xref) +[1, 2, 3] +``` +We can call a remote function. +```python +>>> ref = example_functions.increment(1) +>>>ray.get(ref) +2 +``` + +Note that `example_functions.increment` is defined in +[`scripts/example_functions.py`](../scripts/example_functions.py) as + +```python +@ray.remote([int], [int]) +def increment(x): + return x + 1 +``` + +Note that, we can pass arguments into remote functions either by value or by +object reference. That is, these two lines have the same behavior. + +```python +>>> ray.get(example_functions.increment(1)) +2 +>>> ray.get(example_functions.increment(ray.put(1))) +2 +``` +This is convenient for chaining remote functions together, for example, +```python +>>> ref = example_functions.increment(1) +>>> ref = example_functions.increment(ref) +>>> ref = example_functions.increment(ref) +>>> ray.get(ref) +4 +``` + +### Visualize the computation graph + +At any point, we can visualize the computation graph by running +```python +>>> ray.visualize_computation_graph(view=True) +``` +This will display an image like the following one. + +

+ +

+ +### Restart workers + +During development, suppose that you want to change the implementation of +`example_functions.increment`, but you've already done a bunch of work in the +shell loading and preprocessing data, and you don't want to have to recompute +all of that work. + +We can simply restart the workers. + +First, change the code, for example, modify the function +`example_functions.increment` in +[`scripts/example_functions.py`](../scripts/example_functions.py) to add 10 +instead of 1. + +```python +@ray.remote([int], [int]) +def increment(x): + return x + 10 +``` +Then from the shell, restart the workers like this. +```python +>>> ray.restart_workers("scripts/example_worker.py") # This should be the correct relative path to the example_worker.py code +``` +We can check that the code has been updated by running. +```python +>>> ray.get(example_functions.increment(1)) +11 +``` + +Note that it is not as simple as running `reload(example_functions)` because we +need to reload the Python module on all of the workers as well, and the workers +are separate Python processes. Calling `reload(example_functions)` would only +reload the module on the driver. diff --git a/doc/using-ray-on-a-cluster.md b/doc/using-ray-on-a-cluster.md new file mode 100644 index 000000000..e20dabcfa --- /dev/null +++ b/doc/using-ray-on-a-cluster.md @@ -0,0 +1,79 @@ +## Using Ray on a cluster + +Running Ray on a cluster is still experimental. + +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. + +### Getting started with 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 + + 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. + +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 @ + ``` +4. The previous command should open a Python interpreter. To install Ray on the +cluster, run `install_ray()` 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("/home/ubuntu/ray/scripts/default_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. After completing successfully, this command will print out a +command that can be run on the head node to attach a shell (the driver) to the +cluster. For example, + + ``` + source "$RAY_HOME/setup-env.sh"; + python "$RAY_HOME/scripts/shell.py" --scheduler-address=52.50.28.103:10001 --objstore-address=52.50.28.103:20001 --worker-address=52.50.28.103:30001 --attach + ``` + +7. Note that there are several more commands that can be run from within +`cluster.py`. + + - `install_ray()` - This pulls the Ray source code on each node, builds all + of the third party libraries, and builds the project itself. + - `start_ray(worker_path, 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. + - `stop_ray()` - This shuts down the cluster (killing all of the processes). + - `restart_workers(worker_path, num_workers_per_node=10)` - This kills the + current workers and starts new workers using the worker code from the + given file. Currently, this can only run when there are no tasks currently + executing on any of the workers. + - `update_ray()` - This pulls the latest Ray source code and builds it. + +### Running Ray on a cluster + +Once you've started a Ray cluster using the above instructions, to actually use +Ray, ssh to the head node (the first node listed in the `nodes.txt` file) and +attach a shell to the already running cluster. diff --git a/scripts/example_functions.py b/scripts/example_functions.py index e3e43f24f..f04e55a60 100644 --- a/scripts/example_functions.py +++ b/scripts/example_functions.py @@ -12,6 +12,10 @@ def estimate_pi(n): def increment(x): return x + 1 +@ray.remote([int, int], [int]) +def add(a, b): + return a + b + @ray.remote([List[int]], [np.ndarray]) def zeros(shape): return np.zeros(shape)