From 964d5cac48b3754d6059f047f6a7892b56e8fd7a Mon Sep 17 00:00:00 2001 From: Robert Nishihara Date: Fri, 17 Mar 2017 16:48:25 -0700 Subject: [PATCH] Expand API documentation. (#375) * Expand API documentation and convert tutorial to rst. * Fix formatting error in tutorial. * Address William's comments. * Address Stephanie's comments. --- .gitignore | 13 +- README.md | 13 -- README.rst | 16 +++ doc/source/api.rst | 277 ++++++++++++++++++++++++++++++++++++++- doc/source/index.rst | 8 +- doc/source/tutorial.md | 270 -------------------------------------- doc/source/tutorial.rst | 281 ++++++++++++++++++++++++++++++++++++++++ 7 files changed, 581 insertions(+), 297 deletions(-) delete mode 100644 README.md create mode 100644 README.rst delete mode 100644 doc/source/tutorial.md create mode 100644 doc/source/tutorial.rst diff --git a/.gitignore b/.gitignore index 666380f61..f51683b36 100644 --- a/.gitignore +++ b/.gitignore @@ -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 diff --git a/README.md b/README.md deleted file mode 100644 index 9e9a43d64..000000000 --- a/README.md +++ /dev/null @@ -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). diff --git a/README.rst b/README.rst new file mode 100644 index 000000000..bf7e2a89a --- /dev/null +++ b/README.rst @@ -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 diff --git a/doc/source/api.rst b/doc/source/api.rst index 76632ff15..fadbac890 100644 --- a/doc/source/api.rst +++ b/doc/source/api.rst @@ -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 diff --git a/doc/source/index.rst b/doc/source/index.rst index 4e806d6a5..fe26fd598 100644 --- a/doc/source/index.rst +++ b/doc/source/index.rst @@ -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 diff --git a/doc/source/tutorial.md b/doc/source/tutorial.md deleted file mode 100644 index 85c2a97b1..000000000 --- a/doc/source/tutorial.md +++ /dev/null @@ -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. diff --git a/doc/source/tutorial.rst b/doc/source/tutorial.rst new file mode 100644 index 000000000..b9382e3b8 --- /dev/null +++ b/doc/source/tutorial.rst @@ -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.