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

Expand API documentation. (#375)

Browse Source 
* Expand API documentation and convert tutorial to rst.

* Fix formatting error in tutorial.

* Address William's comments.

* Address Stephanie's comments.
This commit is contained in:
Robert Nishihara
2017-03-17 16:48:25 -07:00
committed by Philipp Moritz
parent 6b1e8caf2d
commit 964d5cac48
7 changed files with 581 additions and 297 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+9 -4
View File
@@ -3,10 +3,11 @@
/src/common/thirdparty/redis
/numbuf/thirdparty/arrow
# Files generated by flatcc should be ignored
/src/plasma/format/*_builder.h
/src/plasma/format/*_reader.h
/src/plasma/format/*_verifier.h
# Files generated by flatc should be ignored
/src/common/format/*.py
/src/common/format/*_generated.h
/src/plasma/format/*_generated.h
/src/local_scheduler/format/*_generated.h
# Redis temporary files
*dump.rdb
@@ -87,3 +88,7 @@ scripts/nodes.txt
# Datasets from examples
**/MNIST_data/
**/cifar-10-batches-bin/
# Generated documentation files
/doc/_build
README.md
-13
View File
@@ -1,13 +0,0 @@
# Ray
[![Build Status](https://travis-ci.org/ray-project/ray.svg?branch=master)](https://travis-ci.org/ray-project/ray)
[![Documentation Status](https://readthedocs.org/projects/ray/badge/?version=latest)](http://ray.readthedocs.io/en/latest/?badge=latest)
Ray is an experimental distributed execution engine. It is under development and
not ready to be used.
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.
View the [documentation](http://ray.readthedocs.io/en/latest/index.html).
README.rst
+16
View File
@@ -0,0 +1,16 @@
Ray
===
.. image:: https://travis-ci.org/ray-project/ray.svg?branch=master
:target: https://travis-ci.org/ray-project/ray
.. image:: https://readthedocs.org/projects/ray/badge/?version=latest
:target: http://ray.readthedocs.io/en/latest/?badge=latest
|
Ray is a flexible, high-performance distributed execution framework.
View the `documentation`_.
.. _`documentation`: http://ray.readthedocs.io/en/latest/index.html
doc/source/api.rst
+272 -5
View File
@@ -1,10 +1,277 @@
===========
The Ray API
===========
.. autofunction:: ray.put
.. autofunction:: ray.get
.. autofunction:: ray.remote
.. autofunction:: ray.wait
Starting Ray
------------
There are two main ways in which Ray can be used. First, you can start all of
the relevant Ray processes and shut them all down within the scope of a single
script. Second, you can connect to and use an existing Ray cluster.
Starting and stopping a cluster within a script
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
One use case is to start all of the relevant Ray processes when you call
``ray.init`` and shut them down when the script exits. These processes include
local and global schedulers, an object store and an object manager, a redis
server, and more.
**Note:** this approach is limited to a single machine.
This can be done as follows.
.. code-block:: python
ray.init()
If there are GPUs available on the machine, you should specify this with the
``num_gpus`` argument. Similarly, you can also specify the number of CPUs with
``num_cpus``.
.. code-block:: python
ray.init(num_cpus=20, num_gpus=2)
By default, Ray will use ``psutil.cpu_count()`` to determine the number of CPUs,
and by default the number of GPUs will be zero.
Connecting to an existing cluster
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Once a Ray cluster has been started, the only thing you need in order to connect
to it is the address of the Redis server in the cluster. In this case, your
script will not start up or shut down any processes. The cluster and all of its
processes may be shared between multiple scripts and multiple users. To do this,
you simply need to know the address of the cluster's Redis server. This can be
done with a command like the following.
.. code-block:: python
ray.init(redis_address="12.345.67.89:6379")
In this case, you cannot specify ``num_cpus`` or ``num_gpus`` in ``ray.init``
because that information is passed into the cluster when the cluster is started,
not when your script is started.
View the instructions for how to `start a Ray cluster`_ on multiple nodes.
.. _`start a Ray cluster`: http://ray.readthedocs.io/en/latest/using-ray-on-a-cluster.html
.. autofunction:: ray.init
Defining remote functions
-------------------------
Remote functions are used to create tasks. To define a remote function, the
``@ray.remote`` decorator is placed over the function definition.
The function can then be invoked with ``f.remote``. Invoking the function
creates a **task** which will be scheduled on and executed by some worker
process in the Ray cluster. The call will return an **object ID** (essentially a
future) representing the eventual return value of the task. Anyone with the
object ID can retrieve its value, regardless of where the task was executed (see
`Getting values from object IDs`_).
When a task executes, its outputs will be serialized into a string of bytes and
stored in the object store.
Note that arguments to remote functions can be values or object IDs.
.. code-block:: python
@ray.remote
def f(x):
return x + 1
x_id = f.remote(0)
ray.get(x_id) # 1
y_id = f.remote(x_id)
ray.get(y_id) # 2
If you want a remote function to return multiple object IDs, you can do that by
passing the ``num_return_vals`` argument into the remote decorator.
.. code-block:: python
@ray.remote(num_return_vals=2)
def f():
return 1, 2
x_id, y_id = f.remote()
ray.get(x_id) # 1
ray.get(y_id) # 2
.. autofunction:: ray.remote
Getting values from object IDs
------------------------------
Object IDs can be converted into objects by calling ``ray.get`` on the object
ID. Note that ``ray.get`` accepts either a single object ID or a list of object
IDs.
.. code-block:: python
@ray.remote
def f():
return {'key1': ['value']}
# Get one object ID.
ray.get(f.remote()) # {'key1': ['value']}
# Get a list of object IDs.
ray.get([f.remote() for _ in range(2)]) # [{'key1': ['value']}, {'key1': ['value']}]
Numpy arrays
~~~~~~~~~~~~
Numpy arrays are handled more efficiently than other data types, so **use numpy
arrays whenever possible**.
Any numpy arrays that are part of the serialized object will not be copied out
of the object store. They will remain in the object store and the resulting
deserialized object will simply have a pointer to the relevant place in the
object store's memory.
Since objects in the object store are immutable, this means that if you want to
mutate a numpy array that was returned by a remote function, you will have to
first copy it.
.. autofunction:: ray.get
Putting objects in the object store
-----------------------------------
The primary way that objects are placed in the object store is by being returned
by a task. However, it is also possible to directly place objects in the object
store using ``ray.put``.
.. code-block:: python
x_id = ray.put(1)
ray.get(x_id) # 1
The main reason to use ``ray.put`` is that you want to pass the same large
object into a number of tasks. By first doing ``ray.put`` and then passing the
resulting object ID into each of the tasks, the large object is copied into the
object store only once, whereas when we directly pass the object in, it is
copied multiple times.
.. code-block:: python
import numpy as np
@ray.remote
def f(x):
pass
x = np.zeros(10 ** 6)
# Alternative 1: Here, x is copied into the object store 10 times.
[f.remote(x) for _ in range(10)]
# Alternative 2: Here, x is copied into the object store once.
x_id = ray.put(x)
[f.remote(x_id) for _ in range(10)]
Note that ``ray.put`` is called under the hood in a couple situations.
- It is called on the values returned by a task.
- It is called on the arguments to a task, unless the arguments are Python
primitives like integers or short strings, lists, tuples, or dictionaries.
.. autofunction:: ray.put
Waiting for a subset of tasks to finish
---------------------------------------
It is often desirable to adapt the computation being done based on when
different tasks finish. For example, if a bunch of tasks each take a variable
length of time, and their results can be processed in any order, then it makes
sense to simply process the results in the order that they finish. In other
settings, it makes sense to discard straggler tasks whose results may not be
needed.
To do this, we introduce the ``ray.wait`` primitive, which takes a list of
object IDs and returns when a subset of them are available. By default it blocks
until a single object is available, but the ``num_returns`` value can be
specified to wait for a different number. If a ``timeout`` argument is passed
in, it will block for at most that many milliseconds and may return a list with
fewer than ``num_returns`` elements.
The ``ray.wait`` function returns two lists. The first list is a list of object
IDs of available objects (of length at most ``num_returns``), and the second
list is a list of the remaining object IDs, so the combination of these two
lists is equal to the list passed in to ``ray.wait`` (up to ordering).
.. code-block:: python
import time
import numpy as np
@ray.remote
def f(n):
time.sleep(n)
return n
# Start 3 tasks with different durations.
results = [f.remote(i) for i in range(3)]
# Block until 2 of them have finished.
ready_ids, remaining_ids = ray.wait(results, num_returns=2)
# Start 5 tasks with different durations.
results = [f.remote(i) for i in range(3)]
# Block until 4 of them have finished or 2.5 seconds pass.
ready_ids, remaining_ids = ray.wait(results, num_returns=4, timeout=2500)
It is easy to use this construct to create an infinite loop in which multiple
tasks are executing, and whenever one task finishes, a new one is launched.
.. code-block:: python
@ray.remote
def f():
return 1
# Start 5 tasks.
remaining_ids = [f.remote() for i in range(5)]
# Whenever one task finishes, start a new one.
for _ in range(100):
ready_ids, remaining_ids = ray.wait(remaining_ids)
# Get the available object and do something with it.
print(ray.get(ready_ids))
# Start a new task.
remaining_ids.append(f.remote())
.. autofunction:: ray.wait
Viewing errors
--------------
Keeping track of errors that occur in different processes throughout a cluster
can be challenging. There are a couple mechanisms to help with this.
1. If a task throws an exception, that exception will be printed in the
background of the driver process.
2. If ``ray.get`` is called on an object ID whose parent task threw an exception
before creating the object, the exception will be re-raised by ``ray.get``.
The errors will also be accumulated in Redis and can be accessed with
``ray.error_info``. Normally, you shouldn't need to do this, but it is possible.
.. code-block:: python
@ray.remote
def f():
raise Exception("This task failed!!")
f.remote() # An error message will be printed in the background.
# Wait for the error to propagate to Redis.
import time
time.sleep(1)
ray.error_info() # This returns a list containing the error message.
.. autofunction:: ray.error_info
doc/source/index.rst
+3 -5
View File
@@ -1,9 +1,7 @@
===
Ray
===
*Ray is a low-latency distributed execution framework targeted at machine
learning and reinforcement learning applications.*
*Ray is a flexible, high-performance distributed execution framework.*
.. toctree::
:maxdepth: 1
@@ -18,9 +16,9 @@ learning and reinforcement learning applications.*
:maxdepth: 1
:caption: Getting Started
tutorial.md
actors.rst
tutorial.rst
api.rst
actors.rst
.. toctree::
:maxdepth: 1
doc/source/tutorial.md
-270
View File
@@ -1,270 +0,0 @@
# Tutorial
To use Ray, you need to understand the following:
- How Ray uses object IDs to represent immutable remote objects.
- How Ray executes tasks asynchronously to achieve parallelism.
## Overview
Ray is a Python-based distributed execution engine. It can be used on a single
machine to achieve efficient multiprocessing, and it can be used on a cluster
for large computations.
When using Ray, several processes are involved.
- Multiple **worker** processes execute tasks and store results in object
stores. Each worker is a separate process.
- One **object store** per node stores immutable objects in shared memory and
allows workers to efficiently share objects on the same node with minimal
copying and deserialization.
- One **local scheduler** per node assigns tasks to workers on the same node.
- A **global scheduler** receives tasks from local schedulers and assigns them
to other local schedulers.
- A **driver** is the Python process that the user controls. For example, if the
user is running a script or using a Python shell, then the driver is the Python
process that runs the script or the shell. A driver is similar to a worker in
that it can submit tasks to its local scheduler and get objects from the object
store, but it is different in that the local scheduler will not assign tasks to
the driver to be executed.
- A **Redis server** maintains much of the system's state. For example, it keeps
track of which objects live on which machines and of the task specifications
(but not data). It can also be queried directly for debugging purposes.
## Starting Ray
To start Ray, start Python and run the following commands.
```python
import ray
ray.init(num_workers=10)
```
This starts Ray with ten workers. Each of these are distinct processes. They
will be killed when you exit the Python interpreter.
## Immutable remote objects
In Ray, we can create and manipulate objects. We refer to these objects as
**remote objects**, and we use **object IDs** 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 ID** is essentially a unique ID that can be used to refer to a
remote object. If you're familiar with Futures, our object IDs 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 IDs, as shown in the example below.
```python
x = [1, 2, 3]
ray.put(x) # prints ObjectID(b49a32d72057bdcfc4dda35584b3d838aad89f5d)
```
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 ID, which is essentially an ID that
can be used to refer to the newly created remote object. If we save the object
ID in a variable with `x_id = ray.put(x)`, then we can pass `x_id` into remote
functions, and those remote functions will operate on the corresponding remote
object.
The command `ray.get(x_id)` takes an object ID 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 copies the object
from the object store to the worker process's heap. If the remote object
corresponding to the object ID `x_id` does not live on the same node as the
worker that calls `ray.get(x_id)`, then the remote object will first be
transferred from an object store that has it to the object store that needs it.
```python
x_id = ray.put([1, 2, 3])
ray.get(x_id) # prints [1, 2, 3]
```
If the remote object corresponding to the object ID `x_id` has not been created
yet, *the command `ray.get(x_id)` will wait until the remote object has been
created.*
A very common use case of `ray.get` is to get a list of object IDs. In this
case, you can call `ray.get(object_ids)` where `object_ids` is a list of object
IDs.
```python
result_ids = [ray.put(i) for i in range(10)]
ray.get(result_ids) # prints [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
```
## Asynchronous Computation in Ray
Ray enables arbitrary Python functions to be executed asynchronously. This is
done by designating a Python function as a **remote function**.
For example, a normal Python function looks like this.
```python
def add1(a, b):
return a + b
```
A remote function looks like this.
```python
@ray.remote
def add2(a, b):
return a + b
```
### Remote functions
Whereas calling `add1(1, 2)` returns `3` and causes the Python interpreter to
block until the computation has finished, calling `add2.remote(1, 2)`
immediately returns an object ID and creates a **task**. The task will be
scheduled by the system and executed asynchronously (potentially on a different
machine). When the task finishes executing, its return value will be stored in
the object store.
```python
x_id = add2.remote(1, 2)
ray.get(x_id) # prints 3
```
The following simple example demonstrates how asynchronous tasks can be used
to parallelize computation.
```python
import time
def f1():
time.sleep(1)
@ray.remote
def f2():
time.sleep(1)
# The following takes ten seconds.
[f1() for _ in range(10)]
```
```python
# The following takes one second (assuming the system has at least ten workers).
ray.get([f2.remote() for _ in range(10)])
```
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 a local scheduler, and object IDs for the outputs of the task are
immediately returned. However, the task will not be executed until the system
actually schedules the task on a worker. Task execution is **not** done lazily.
**When a task is submitted, each argument may be passed in by value or by object
ID.** For example, these lines have the same behavior.
```python
add2.remote(1, 2)
add2.remote(1, ray.put(2))
add2.remote(ray.put(1), ray.put(2))
```
Remote functions never return actual values, they always return object IDs.
When the remote function is actually executed, it operates on Python objects.
That is, if the remote function was called with any object IDs, the Python
objects corresponding to those object IDs will be retrieved and passed into the
actual execution of the remote function.
Note that a remote function can return multiple object IDs.
```python
@ray.remote(num_return_vals=3)
def return_multiple():
return 1, 2, 3
a_id, b_id, c_id = return_multiple.remote()
```
### Expressing dependencies between tasks
Programmers can express dependencies between tasks by passing the object ID
output of one task as an argument to another task. For example, we can launch
three tasks as follows, each of which depends on the previous task.
```python
@ray.remote
def f(x):
return x + 1
x = f.remote(0)
y = f.remote(x)
z = f.remote(y)
ray.get(z) # prints 3
```
The second task above will not execute until the first has finished, and the
third will not execute until the second has finished. In this example, there are
no opportunities for parallelism.
The ability to compose tasks makes it easy to express interesting dependencies.
Consider the following implementation of a tree reduce.
```python
import numpy as np
@ray.remote
def generate_data():
return np.random.normal(size=1000)
@ray.remote
def aggregate_data(x, y):
return x + y
# Generate some random data. This launches 100 tasks that will be scheduled on
# various nodes. The resulting data will be distributed around the cluster.
data = [generate_data.remote() for _ in range(100)]
# Perform a tree reduce.
while len(data) > 1:
data.append(aggregate_data.remote(data.pop(0), data.pop(0)))
# Fetch the result.
ray.get(data)
```
### Remote Functions Within Remote Functions
So far, we have been calling remote functions only from the driver. But worker
processes can also call remote functions. To illustrate this, consider the
following example.
```python
@ray.remote
def sub_experiment(i, j):
# Run the jth sub-experiment for the ith experiment.
return i + j
@ray.remote
def run_experiment(i):
sub_results = []
# Launch tasks to perform 10 sub-experiments in parallel.
for j in range(10):
sub_results.append(sub_experiment.remote(i, j))
# Return the sum of the results of the sub-experiments.
return sum(ray.get(sub_results))
results = [run_experiment.remote(i) for i in range(5)]
ray.get(results) # prints [45, 55, 65, 75, 85]
```
When the remote function `run_experiment` is executed on a worker, it calls the
remote function `sub_experiment` a number of times. This is an example of how
multiple experiments, each of which takes advantage of parallelism internally,
can all be run in parallel.
doc/source/tutorial.rst
+281
View File
@@ -0,0 +1,281 @@
Tutorial
========
To use Ray, you need to understand the following:
- How Ray executes tasks asynchronously to achieve parallelism.
- How Ray uses object IDs to represent immutable remote objects.
Overview
--------
Ray is a Python-based distributed execution engine. The same code can be run on
a single machine to achieve efficient multiprocessing, and it can be used on a
cluster for large computations.
When using Ray, several processes are involved.
- Multiple **worker** processes execute tasks and store results in object
stores. Each worker is a separate process.
- One **object store** per node stores immutable objects in shared memory and
allows workers to efficiently share objects on the same node with minimal
copying and deserialization.
- One **local scheduler** per node assigns tasks to workers on the same node.
- A **global scheduler** receives tasks from local schedulers and assigns them
to other local schedulers.
- A **driver** is the Python process that the user controls. For example, if the
user is running a script or using a Python shell, then the driver is the Python
process that runs the script or the shell. A driver is similar to a worker in
that it can submit tasks to its local scheduler and get objects from the object
store, but it is different in that the local scheduler will not assign tasks to
the driver to be executed.
- A **Redis server** maintains much of the system's state. For example, it keeps
track of which objects live on which machines and of the task specifications
(but not data). It can also be queried directly for debugging purposes.
Starting Ray
------------
To start Ray, start Python and run the following commands.
.. code-block:: python
import ray
ray.init()
This starts Ray.
Immutable remote objects
------------------------
In Ray, we can create and compute on objects. We refer to these objects as
**remote objects**, and we use **object IDs** 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 ID** is essentially a unique ID that can be used to refer to a
remote object. If you're familiar with Futures, our object IDs 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 IDs, as shown in the example below.
.. code-block:: python
x = [1, 2, 3]
ray.put(x) # ObjectID(b49a32d72057bdcfc4dda35584b3d838aad89f5d)
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 ID, which is essentially an ID that
can be used to refer to the newly created remote object. If we save the object
ID in a variable with ``x_id = ray.put(x)``, then we can pass ``x_id`` into remote
functions, and those remote functions will operate on the corresponding remote
object.
The command ``ray.get(x_id)`` takes an object ID 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 copies the object
from the object store to the worker process's heap. If the remote object
corresponding to the object ID ``x_id`` does not live on the same node as the
worker that calls ``ray.get(x_id)``, then the remote object will first be
transferred from an object store that has it to the object store that needs it.
.. code-block:: python
x_id = ray.put([1, 2, 3])
ray.get(x_id) # [1, 2, 3]
If the remote object corresponding to the object ID ``x_id`` has not been created
yet, the command ``ray.get(x_id)`` will wait until the remote object has been
created.
A very common use case of ``ray.get`` is to get a list of object IDs. In this
case, you can call ``ray.get(object_ids)`` where ``object_ids`` is a list of object
IDs.
.. code-block:: python
result_ids = [ray.put(i) for i in range(10)]
ray.get(result_ids) # [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
Asynchronous Computation in Ray
-------------------------------
Ray enables arbitrary Python functions to be executed asynchronously. This is
done by designating a Python function as a **remote function**.
For example, a normal Python function looks like this.
.. code-block:: python
def add1(a, b):
return a + b
A remote function looks like this.
.. code-block:: python
@ray.remote
def add2(a, b):
return a + b
Remote functions
~~~~~~~~~~~~~~~~
Whereas calling ``add1(1, 2)`` returns ``3`` and causes the Python interpreter to
block until the computation has finished, calling ``add2.remote(1, 2)``
immediately returns an object ID and creates a **task**. The task will be
scheduled by the system and executed asynchronously (potentially on a different
machine). When the task finishes executing, its return value will be stored in
the object store.
.. code-block:: python
x_id = add2.remote(1, 2)
ray.get(x_id) # 3
The following simple example demonstrates how asynchronous tasks can be used
to parallelize computation.
.. code-block:: python
import time
def f1():
time.sleep(1)
@ray.remote
def f2():
time.sleep(1)
# The following takes ten seconds.
[f1() for _ in range(10)]
# The following takes one second (assuming the system has at least ten CPUs).
ray.get([f2.remote() for _ in range(10)])
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 a local scheduler, and object IDs for the outputs of the task are
immediately returned. However, the task will not be executed until the system
actually schedules the task on a worker. Task execution is **not** done lazily.
The system moves the input data to the task, and the task will execute as soon
as its input dependencies are available and there are enough resources for the
computation.
**When a task is submitted, each argument may be passed in by value or by object
ID.** For example, these lines have the same behavior.
.. code-block:: python
add2.remote(1, 2)
add2.remote(1, ray.put(2))
add2.remote(ray.put(1), ray.put(2))
Remote functions never return actual values, they always return object IDs.
When the remote function is actually executed, it operates on Python objects.
That is, if the remote function was called with any object IDs, the system will
retrieve the corresponding objects from the object store.
Note that a remote function can return multiple object IDs.
.. code-block:: python
@ray.remote(num_return_vals=3)
def return_multiple():
return 1, 2, 3
a_id, b_id, c_id = return_multiple.remote()
Expressing dependencies between tasks
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Programmers can express dependencies between tasks by passing the object ID
output of one task as an argument to another task. For example, we can launch
three tasks as follows, each of which depends on the previous task.
.. code-block:: python
@ray.remote
def f(x):
return x + 1
x = f.remote(0)
y = f.remote(x)
z = f.remote(y)
ray.get(z) # 3
The second task above will not execute until the first has finished, and the
third will not execute until the second has finished. In this example, there are
no opportunities for parallelism.
The ability to compose tasks makes it easy to express interesting dependencies.
Consider the following implementation of a tree reduce.
.. code-block:: python
import numpy as np
@ray.remote
def generate_data():
return np.random.normal(size=1000)
@ray.remote
def aggregate_data(x, y):
return x + y
# Generate some random data. This launches 100 tasks that will be scheduled on
# various nodes. The resulting data will be distributed around the cluster.
data = [generate_data.remote() for _ in range(100)]
# Perform a tree reduce.
while len(data) > 1:
data.append(aggregate_data.remote(data.pop(0), data.pop(0)))
# Fetch the result.
ray.get(data)
Remote Functions Within Remote Functions
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
So far, we have been calling remote functions only from the driver. But worker
processes can also call remote functions. To illustrate this, consider the
following example.
.. code-block:: python
@ray.remote
def sub_experiment(i, j):
# Run the jth sub-experiment for the ith experiment.
return i + j
@ray.remote
def run_experiment(i):
sub_results = []
# Launch tasks to perform 10 sub-experiments in parallel.
for j in range(10):
sub_results.append(sub_experiment.remote(i, j))
# Return the sum of the results of the sub-experiments.
return sum(ray.get(sub_results))
results = [run_experiment.remote(i) for i in range(5)]
ray.get(results) # [45, 55, 65, 75, 85]
When the remote function ``run_experiment`` is executed on a worker, it calls the
remote function ``sub_experiment`` a number of times. This is an example of how
multiple experiments, each of which takes advantage of parallelism internally,
can all be run in parallel.