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

[docs] Convert Examples to Gallery (#5414)

Browse Source
This commit is contained in:
Richard Liaw
2019-09-24 15:46:56 -07:00
committed by GitHub
parent ea9376c9ce
commit 10f21fa313
28 changed files with 966 additions and 885 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doc/source/_static/img/thumbnails/default.png
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 26 KiB

doc/source/conf.py
+37 -7
View File
@@ -12,10 +12,13 @@
# All configuration values have a default; values that are commented out
# serve to show the default.
import glob
import shutil
import sys
import os
import urllib
import shlex
sys.path.insert(0, os.path.abspath('.'))
from custom_directives import CustomGalleryItemDirective
# These lines added to enable Sphinx to work without installing Ray.
import mock
@@ -67,13 +70,33 @@ sys.path.insert(0, os.path.abspath("../../python/"))
# extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
# ones.
extensions = [
'sphinx.ext.autodoc',
'sphinx.ext.viewcode',
'sphinx.ext.napoleon',
'sphinx_click.ext',
'sphinx-jsonschema',
'sphinx.ext.autodoc', 'sphinx.ext.viewcode', 'sphinx.ext.napoleon',
'sphinx_click.ext', 'sphinx-jsonschema', 'sphinx_gallery.gen_gallery'
]
sphinx_gallery_conf = {
"examples_dirs": ["../examples"], # path to example scripts
"gallery_dirs": ["auto_examples"], # path where to save generated examples
"ignore_pattern": "../examples/doc_code/",
"plot_gallery": "False",
# "filename_pattern": "tutorial.py",
"backreferences_dir": False
# "show_memory': False,
# 'min_reported_time': False
}
for i in range(len(sphinx_gallery_conf["examples_dirs"])):
gallery_dir = sphinx_gallery_conf["gallery_dirs"][i]
source_dir = sphinx_gallery_conf["examples_dirs"][i]
try:
os.mkdir(gallery_dir)
except OSError:
pass
# Copy rst files from source dir to gallery dir.
for f in glob.glob(os.path.join(source_dir, '*.rst')):
shutil.copy(f, gallery_dir)
# Add any paths that contain templates here, relative to this directory.
templates_path = ['_templates']
@@ -95,7 +118,7 @@ master_doc = 'index'
# General information about the project.
project = u'Ray'
copyright = u'2016, The Ray Team'
copyright = u'2019, The Ray Team'
author = u'The Ray Team'
# The version info for the project you're documenting, acts as replacement for
@@ -123,6 +146,8 @@ language = None
# List of patterns, relative to source directory, that match files and
# directories to ignore when looking for source files.
exclude_patterns = ['_build']
exclude_patterns += sphinx_gallery_conf['examples_dirs']
exclude_patterns += ["*/README.rst"]
# The reST default role (used for this markup: `text`) to use for all
# documents.
@@ -354,5 +379,10 @@ def update_context(app, pagename, templatename, context, doctree):
pagename)
# see also http://searchvoidstar.tumblr.com/post/125486358368/making-pdfs-from-markdown-on-readthedocsorg-using
def setup(app):
app.connect('html-page-context', update_context)
# Custom directives
app.add_directive('customgalleryitem', CustomGalleryItemDirective)
doc/source/custom_directives.py
+94
View File
@@ -0,0 +1,94 @@
# Originally from:
# github.com/pytorch/tutorials/blob/60d6ef365e36f3ba82c2b61bf32cc40ac4e86c7b/custom_directives.py # noqa
from docutils.parsers.rst import Directive, directives
from docutils.statemachine import StringList
from docutils import nodes
import os
import sphinx_gallery
try:
FileNotFoundError
except NameError:
FileNotFoundError = IOError
GALLERY_TEMPLATE = """
.. raw:: html
<div class="sphx-glr-thumbcontainer" tooltip="{tooltip}">
.. only:: html
.. figure:: {thumbnail}
{description}
.. raw:: html
</div>
"""
class CustomGalleryItemDirective(Directive):
"""Create a sphinx gallery style thumbnail.
tooltip and figure are self explanatory. Description could be a link to
a document like in below example.
Example usage:
.. customgalleryitem::
:tooltip: I am writing this tutorial to focus specifically on NLP.
:figure: /_static/img/thumbnails/babel.jpg
:description: :doc:`/beginner/deep_learning_nlp_tutorial`
If figure is specified, a thumbnail will be made out of it and stored in
_static/thumbs. Therefore, consider _static/thumbs as a "built" directory.
"""
required_arguments = 0
optional_arguments = 0
final_argument_whitespace = True
option_spec = {
"tooltip": directives.unchanged,
"figure": directives.unchanged,
"description": directives.unchanged
}
has_content = False
add_index = False
def run(self):
# Cutoff the `tooltip` after 195 chars.
if "tooltip" in self.options:
tooltip = self.options["tooltip"]
if len(self.options["tooltip"]) > 195:
tooltip = tooltip[:195] + "..."
else:
raise ValueError("Need to provide :tooltip: under "
"`.. customgalleryitem::`.")
# Generate `thumbnail` used in the gallery.
if "figure" in self.options:
env = self.state.document.settings.env
rel_figname, figname = env.relfn2path(self.options["figure"])
thumbnail = os.path.join("_static/thumbs/",
os.path.basename(figname))
os.makedirs("_static/thumbs", exist_ok=True)
sphinx_gallery.gen_rst.scale_image(figname, thumbnail, 400, 280)
else:
thumbnail = "/_static/img/thumbnails/default.png"
if "description" in self.options:
description = self.options["description"]
else:
raise ValueError("Need to provide :description: under "
"`customgalleryitem::`.")
thumbnail_rst = GALLERY_TEMPLATE.format(
tooltip=tooltip, thumbnail=thumbnail, description=description)
thumbnail = StringList(thumbnail_rst.split("\n"))
thumb = nodes.paragraph()
self.state.nested_parse(thumbnail, self.content_offset, thumb)
return [thumb]
doc/source/example-a3c.rst
-162
View File
@@ -1,162 +0,0 @@
Asynchronous Advantage Actor Critic (A3C)
=========================================
This document walks through `A3C`_, a state-of-the-art reinforcement learning
algorithm. In this example, we adapt the OpenAI `Universe Starter Agent`_
implementation of A3C to use Ray.
View the `code for this example`_.
.. _`A3C`: https://arxiv.org/abs/1602.01783
.. _`Universe Starter Agent`: https://github.com/openai/universe-starter-agent
.. _`code for this example`: https://github.com/ray-project/ray/tree/master/rllib/agents/a3c
.. note::
For an overview of Ray's reinforcement learning library, see `RLlib <http://ray.readthedocs.io/en/latest/rllib.html>`__.
To run the application, first install **ray** and then some dependencies:
.. code-block:: bash
pip install tensorflow
pip install six
pip install gym[atari]
pip install opencv-python-headless
pip install scipy
You can run the code with
.. code-block:: bash
rllib train --env=Pong-ram-v4 --run=A3C --config='{"num_workers": N}'
Reinforcement Learning
----------------------
Reinforcement Learning is an area of machine learning concerned with **learning
how an agent should act in an environment** so as to maximize some form of
cumulative reward. Typically, an agent will observe the current state of the
environment and take an action based on its observation. The action will change
the state of the environment and will provide some numerical reward (or penalty)
to the agent. The agent will then take in another observation and the process
will repeat. **The mapping from state to action is a policy**, and in
reinforcement learning, this policy is often represented with a deep neural
network.
The **environment** is often a simulator (for example, a physics engine), and
reinforcement learning algorithms often involve trying out many different
sequences of actions within these simulators. These **rollouts** can often be
done in parallel.
Policies are often initialized randomly and incrementally improved via
simulation within the environment. To improve a policy, gradient-based updates
may be computed based on the sequences of states and actions that have been
observed. The gradient calculation is often delayed until a termination
condition is reached (that is, the simulation has finished) so that delayed
rewards have been properly accounted for. However, in the Actor Critic model, we
can begin the gradient calculation at any point in the simulation rollout by
predicting future rewards with a Value Function approximator.
In our A3C implementation, each worker, implemented as a Ray actor, continuously
simulates the environment. The driver will create a task that runs some steps
of the simulator using the latest model, computes a gradient update, and returns
the update to the driver. Whenever a task finishes, the driver will use the
gradient update to update the model and will launch a new task with the latest
model.
There are two main parts to the implementation - the driver and the worker.
Worker Code Walkthrough
-----------------------
We use a Ray Actor to simulate the environment.
.. code-block:: python
import numpy as np
import ray
@ray.remote
class Runner(object):
"""Actor object to start running simulation on workers.
Gradient computation is also executed on this object."""
def __init__(self, env_name, actor_id):
# starts simulation environment, policy, and thread.
# Thread will continuously interact with the simulation environment
self.env = env = create_env(env_name)
self.id = actor_id
self.policy = LSTMPolicy()
self.runner = RunnerThread(env, self.policy, 20)
self.start()
def start(self):
# starts the simulation thread
self.runner.start_runner()
def pull_batch_from_queue(self):
# Implementation details removed - gets partial rollout from queue
return rollout
def compute_gradient(self, params):
self.policy.set_weights(params)
rollout = self.pull_batch_from_queue()
batch = process_rollout(rollout, gamma=0.99, lambda_=1.0)
gradient = self.policy.compute_gradients(batch)
info = {"id": self.id,
"size": len(batch.a)}
return gradient, info
Driver Code Walkthrough
-----------------------
The driver manages the coordination among workers and handles updating the
global model parameters. The main training script looks like the following.
.. code-block:: python
import numpy as np
import ray
def train(num_workers, env_name="PongDeterministic-v4"):
# Setup a copy of the environment
# Instantiate a copy of the policy - mainly used as a placeholder
env = create_env(env_name, None, None)
policy = LSTMPolicy(env.observation_space.shape, env.action_space.n, 0)
obs = 0
# Start simulations on actors
agents = [Runner.remote(env_name, i) for i in range(num_workers)]
# Start gradient calculation tasks on each actor
parameters = policy.get_weights()
gradient_list = [agent.compute_gradient.remote(parameters) for agent in agents]
while True: # Replace with your termination condition
# wait for some gradient to be computed - unblock as soon as the earliest arrives
done_id, gradient_list = ray.wait(gradient_list)
# get the results of the task from the object store
gradient, info = ray.get(done_id)[0]
obs += info["size"]
# apply update, get the weights from the model, start a new task on the same actor object
policy.apply_gradients(gradient)
parameters = policy.get_weights()
gradient_list.extend([agents[info["id"]].compute_gradient(parameters)])
return policy
Benchmarks and Visualization
----------------------------
For the :code:`PongDeterministic-v4` and an Amazon EC2 m4.16xlarge instance, we
are able to train the agent with 16 workers in around 15 minutes. With 8
workers, we can train the agent in around 25 minutes.
You can visualize performance by running
:code:`tensorboard --logdir [directory]` in a separate screen, where
:code:`[directory]` is defaulted to :code:`~/ray_results/`. If you are running
multiple experiments, be sure to vary the directory to which Tensorflow saves
its progress (found in :code:`a3c.py`).
doc/source/example-lbfgs.rst
-160
View File
@@ -1,160 +0,0 @@
Batch L-BFGS
============
This document provides a walkthrough of the L-BFGS example. To run the
application, first install these dependencies.
.. code-block:: bash
pip install tensorflow
pip install scipy
You can view the `code for this example`_.
.. _`code for this example`: https://github.com/ray-project/ray/tree/master/doc/examples/lbfgs
Then you can run the example as follows.
.. code-block:: bash
python ray/doc/examples/lbfgs/driver.py
Optimization is at the heart of many machine learning algorithms. Much of
machine learning involves specifying a loss function and finding the parameters
that minimize the loss. If we can compute the gradient of the loss function,
then we can apply a variety of gradient-based optimization algorithms. L-BFGS is
one such algorithm. It is a quasi-Newton method that uses gradient information
to approximate the inverse Hessian of the loss function in a computationally
efficient manner.
The serial version
------------------
First we load the data in batches. Here, each element in ``batches`` is a tuple
whose first component is a batch of ``100`` images and whose second component is a
batch of the ``100`` corresponding labels. For simplicity, we use TensorFlow's
built in methods for loading the data.
.. code-block:: python
from tensorflow.examples.tutorials.mnist import input_data
mnist = input_data.read_data_sets("MNIST_data/", one_hot=True)
batch_size = 100
num_batches = mnist.train.num_examples // batch_size
batches = [mnist.train.next_batch(batch_size) for _ in range(num_batches)]
Now, suppose we have defined a function which takes a set of model parameters
``theta`` and a batch of data (both images and labels) and computes the loss for
that choice of model parameters on that batch of data. Similarly, suppose we've
also defined a function that takes the same arguments and computes the gradient
of the loss for that choice of model parameters.
.. code-block:: python
def loss(theta, xs, ys):
# compute the loss on a batch of data
return loss
def grad(theta, xs, ys):
# compute the gradient on a batch of data
return grad
def full_loss(theta):
# compute the loss on the full data set
return sum([loss(theta, xs, ys) for (xs, ys) in batches])
def full_grad(theta):
# compute the gradient on the full data set
return sum([grad(theta, xs, ys) for (xs, ys) in batches])
Since we are working with a small dataset, we don't actually need to separate
these methods into the part that operates on a batch and the part that operates
on the full dataset, but doing so will make the distributed version clearer.
Now, if we wish to optimize the loss function using L-BFGS, we simply plug these
functions, along with an initial choice of model parameters, into
``scipy.optimize.fmin_l_bfgs_b``.
.. code-block:: python
theta_init = 1e-2 * np.random.normal(size=dim)
result = scipy.optimize.fmin_l_bfgs_b(full_loss, theta_init, fprime=full_grad)
The distributed version
-----------------------
In this example, the computation of the gradient itself can be done in parallel
on a number of workers or machines.
First, let's turn the data into a collection of remote objects.
.. code-block:: python
batch_ids = [(ray.put(xs), ray.put(ys)) for (xs, ys) in batches]
We can load the data on the driver and distribute it this way because MNIST
easily fits on a single machine. However, for larger data sets, we will need to
use remote functions to distribute the loading of the data.
Now, lets turn ``loss`` and ``grad`` into methods of an actor that will contain our network.
.. code-block:: python
class Network(object):
def __init__():
# Initialize network.
def loss(theta, xs, ys):
# compute the loss
return loss
def grad(theta, xs, ys):
# compute the gradient
return grad
Now, it is easy to speed up the computation of the full loss and the full
gradient.
.. code-block:: python
def full_loss(theta):
theta_id = ray.put(theta)
loss_ids = [actor.loss(theta_id) for actor in actors]
return sum(ray.get(loss_ids))
def full_grad(theta):
theta_id = ray.put(theta)
grad_ids = [actor.grad(theta_id) for actor in actors]
return sum(ray.get(grad_ids)).astype("float64") # This conversion is necessary for use with fmin_l_bfgs_b.
Note that we turn ``theta`` into a remote object with the line ``theta_id =
ray.put(theta)`` before passing it into the remote functions. If we had written
.. code-block:: python
[actor.loss(theta_id) for actor in actors]
instead of
.. code-block:: python
theta_id = ray.put(theta)
[actor.loss(theta_id) for actor in actors]
then each task that got sent to the scheduler (one for every element of
``batch_ids``) would have had a copy of ``theta`` serialized inside of it. Since
``theta`` here consists of the parameters of a potentially large model, this is
inefficient. *Large objects should be passed by object ID to remote functions
and not by value*.
We use remote actors and remote objects internally in the implementation of
``full_loss`` and ``full_grad``, but the user-facing behavior of these methods is
identical to the behavior in the serial version.
We can now optimize the objective with the same function call as before.
.. code-block:: python
theta_init = 1e-2 * np.random.normal(size=dim)
result = scipy.optimize.fmin_l_bfgs_b(full_loss, theta_init, fprime=full_grad)
doc/source/example-newsreader.rst
-29
View File
@@ -1,29 +0,0 @@
News Reader
===========
This document shows how to implement a simple news reader using Ray. The reader
consists of a simple Vue.js `frontend`_ and a backend consisting of a Flask
server and a Ray actor. View the `code for this example`_.
To run this example, you will need to install NPM and a few python dependencies.
.. code-block:: bash
pip install atoma
pip install flask
To use this example you need to
* In the ``ray/doc/examples/newsreader`` directory, start the server with
``python server.py``.
* Clone the client code with ``git clone https://github.com/ray-project/qreader``
* Start the client with ``cd qreader; npm install; npm run dev``
* You can now add a channel by clicking "Add channel" and for example pasting
``http://news.ycombinator.com/rss`` into the field.
* Star some of the articles and dump the database by running
``sqlite3 newsreader.db`` in a terminal in the ``ray/doc/examples/newsreader``
directory and entering ``SELECT * FROM news;``.
.. _`frontend`: https://github.com/saqueib/qreader
.. _`code for this example`: https://github.com/ray-project/ray/tree/master/doc/examples/newsreader
doc/source/example-parameter-server.rst
-127
View File
@@ -1,127 +0,0 @@
Parameter Server
================
This document walks through how to implement simple synchronous and asynchronous
parameter servers using actors. To run the application, first install some
dependencies.
.. code-block:: bash
pip install tensorflow
You can view the `code for this example`_.
.. _`code for this example`: https://github.com/ray-project/ray/tree/master/doc/examples/parameter_server
The examples can be run as follows.
.. code-block:: bash
# Run the asynchronous parameter server.
python ray/doc/examples/parameter_server/async_parameter_server.py --num-workers=4
# Run the synchronous parameter server.
python ray/doc/examples/parameter_server/sync_parameter_server.py --num-workers=4
Note that this examples uses distributed actor handles, which are still
considered experimental.
Asynchronous Parameter Server
-----------------------------
The asynchronous parameter server itself is implemented as an actor, which
exposes the methods ``push`` and ``pull``.
.. code-block:: python
@ray.remote
class ParameterServer(object):
def __init__(self, keys, values):
values = [value.copy() for value in values]
self.weights = dict(zip(keys, values))
def push(self, keys, values):
for key, value in zip(keys, values):
self.weights[key] += value
def pull(self, keys):
return [self.weights[key] for key in keys]
We then define a worker task, which take a parameter server as an argument and
submits tasks to it. The structure of the code looks as follows.
.. code-block:: python
@ray.remote
def worker_task(ps):
while True:
# Get the latest weights from the parameter server.
weights = ray.get(ps.pull.remote(keys))
# Compute an update.
...
# Push the update to the parameter server.
ps.push.remote(keys, update)
Then we can create a parameter server and initiate training as follows.
.. code-block:: python
ps = ParameterServer.remote(keys, initial_values)
worker_tasks = [worker_task.remote(ps) for _ in range(4)]
Synchronous Parameter Server
----------------------------
The parameter server is implemented as an actor, which exposes the
methods ``apply_gradients`` and ``get_weights``. A constant linear scaling
rule is applied by scaling the learning rate by the number of workers.
.. code-block:: python
@ray.remote
class ParameterServer(object):
def __init__(self, learning_rate):
self.net = model.SimpleCNN(learning_rate=learning_rate)
def apply_gradients(self, *gradients):
self.net.apply_gradients(np.mean(gradients, axis=0))
return self.net.variables.get_flat()
def get_weights(self):
return self.net.variables.get_flat()
Workers are actors which expose the method ``compute_gradients``.
.. code-block:: python
@ray.remote
class Worker(object):
def __init__(self, worker_index, batch_size=50):
self.worker_index = worker_index
self.batch_size = batch_size
self.mnist = input_data.read_data_sets("MNIST_data", one_hot=True,
seed=worker_index)
self.net = model.SimpleCNN()
def compute_gradients(self, weights):
self.net.variables.set_flat(weights)
xs, ys = self.mnist.train.next_batch(self.batch_size)
return self.net.compute_gradients(xs, ys)
Training alternates between computing the gradients given the current weights
from the parameter server and updating the parameter server's weights with the
resulting gradients.
.. code-block:: python
while True:
gradients = [worker.compute_gradients.remote(current_weights)
for worker in workers]
current_weights = ps.apply_gradients.remote(*gradients)
Both of these examples implement the parameter server using a single actor,
however they can be easily extended to **split the parameters across multiple
actors**.
doc/source/example-resnet.rst
-103
View File
@@ -1,103 +0,0 @@
ResNet
======
This code uses ResNet to do data parallel training
across multiple GPUs using Ray. View the `code for this example`_.
To run the example, you will need to install `TensorFlow`_ (at
least version ``1.0.0``). Then you can run the example as follows.
First download the CIFAR-10 or CIFAR-100 dataset.
.. code-block:: bash
# Get the CIFAR-10 dataset.
curl -o cifar-10-binary.tar.gz https://www.cs.toronto.edu/~kriz/cifar-10-binary.tar.gz
tar -xvf cifar-10-binary.tar.gz
# Get the CIFAR-100 dataset.
curl -o cifar-100-binary.tar.gz https://www.cs.toronto.edu/~kriz/cifar-100-binary.tar.gz
tar -xvf cifar-100-binary.tar.gz
Then run the training script that matches the dataset you downloaded.
.. code-block:: bash
# Train Resnet on CIFAR-10.
python ray/doc/examples/resnet/resnet_main.py \
--eval_dir=/tmp/resnet-model/eval \
--train_data_path=cifar-10-batches-bin/data_batch* \
--eval_data_path=cifar-10-batches-bin/test_batch.bin \
--dataset=cifar10 \
--num_gpus=1
# Train Resnet on CIFAR-100.
python ray/doc/examples/resnet/resnet_main.py \
--eval_dir=/tmp/resnet-model/eval \
--train_data_path=cifar-100-binary/train.bin \
--eval_data_path=cifar-100-binary/test.bin \
--dataset=cifar100 \
--num_gpus=1
To run the training script on a cluster with multiple machines, you will need
to also pass in the flag ``--address=<address>``, where
``<address>`` is the address of the Redis server on the head node.
The script will print out the IP address that the log files are stored on. In
the single-node case, you can ignore this and run tensorboard on the current
machine.
.. code-block:: bash
python -m tensorflow.tensorboard --logdir=/tmp/resnet-model
If you are running Ray on multiple nodes, you will need to go to the node at the
IP address printed, and run the command.
The core of the script is the actor definition.
.. code-block:: python
@ray.remote(num_gpus=1)
class ResNetTrainActor(object):
def __init__(self, data, dataset, num_gpus):
# data is the preprocessed images and labels extracted from the dataset.
# Thus, every actor has its own copy of the data.
# Set the CUDA_VISIBLE_DEVICES environment variable in order to restrict
# which GPUs TensorFlow uses. Note that this only works if it is done before
# the call to tf.Session.
os.environ['CUDA_VISIBLE_DEVICES'] = ','.join([str(i) for i in ray.get_gpu_ids()])
with tf.Graph().as_default():
with tf.device('/gpu:0'):
# We omit the code here that actually constructs the residual network
# and initializes it. Uses the definition in the Tensorflow Resnet Example.
def compute_steps(self, weights):
# This method sets the weights in the network, runs some training steps,
# and returns the new weights. self.model.variables is a TensorFlowVariables
# class that we pass the train operation into.
self.model.variables.set_weights(weights)
for i in range(self.steps):
self.model.variables.sess.run(self.model.train_op)
return self.model.variables.get_weights()
The main script first creates one actor for each GPU, or a single actor if
``num_gpus`` is zero.
.. code-block:: python
train_actors = [ResNetTrainActor.remote(train_data, dataset, num_gpus) for _ in range(num_gpus)]
Then the main loop passes the same weights to every model, performs
updates on each model, averages the updates, and puts the new weights in the
object store.
.. code-block:: python
while True:
all_weights = ray.get([actor.compute_steps.remote(weight_id) for actor in train_actors])
mean_weights = {k: sum([weights[k] for weights in all_weights]) / num_gpus for k in all_weights[0]}
weight_id = ray.put(mean_weights)
.. _`TensorFlow`: https://www.tensorflow.org/install/
.. _`code for this example`: https://github.com/ray-project/ray/tree/master/doc/examples/resnet
doc/source/example-rl-pong.rst
-118
View File
@@ -1,118 +0,0 @@
Learning to Play Pong
=====================
In this example, we'll train a **very simple** neural network to play Pong using
the OpenAI Gym. This application is adapted, with minimal modifications, from
Andrej Karpathy's `code`_ (see the accompanying `blog post`_).
You can view the `code for this example`_.
To run the application, first install some dependencies.
.. code-block:: bash
pip install gym[atari]
Then you can run the example as follows.
.. code-block:: bash
python ray/doc/examples/rl_pong/driver.py --batch-size=10
To run the example on a cluster, simply pass in the flag
``--address=<address>``.
At the moment, on a large machine with 64 physical cores, computing an update
with a batch of size 1 takes about 1 second, a batch of size 10 takes about 2.5
seconds. A batch of size 60 takes about 3 seconds. On a cluster with 11 nodes,
each with 18 physical cores, a batch of size 300 takes about 10 seconds. If the
numbers you see differ from these by much, take a look at the
**Troubleshooting** section at the bottom of this page and consider `submitting
an issue`_.
.. _`code`: https://gist.github.com/karpathy/a4166c7fe253700972fcbc77e4ea32c5
.. _`blog post`: http://karpathy.github.io/2016/05/31/rl/
.. _`code for this example`: https://github.com/ray-project/ray/tree/master/doc/examples/rl_pong
.. _`submitting an issue`: https://github.com/ray-project/ray/issues
**Note** that these times depend on how long the rollouts take, which in turn
depends on how well the policy is doing. For example, a really bad policy will
lose very quickly. As the policy learns, we should expect these numbers to
increase.
The distributed version
-----------------------
At the core of Andrej's `code`_, a neural network is used to define a "policy"
for playing Pong (that is, a function that chooses an action given a state). In
the loop, the network repeatedly plays games of Pong and records a gradient from
each game. Every ten games, the gradients are combined together and used to
update the network.
This example is easy to parallelize because the network can play ten games in
parallel and no information needs to be shared between the games.
We define an **actor** for the Pong environment, which includes a method for
performing a rollout and computing a gradient update. Below is pseudocode for
the actor.
.. code-block:: python
@ray.remote
class PongEnv(object):
def __init__(self):
# Tell numpy to only use one core. If we don't do this, each actor may try
# to use all of the cores and the resulting contention may result in no
# speedup over the serial version. Note that if numpy is using OpenBLAS,
# then you need to set OPENBLAS_NUM_THREADS=1, and you probably need to do
# it from the command line (so it happens before numpy is imported).
os.environ["MKL_NUM_THREADS"] = "1"
self.env = gym.make("Pong-v0")
def compute_gradient(self, model):
# Reset the game.
observation = self.env.reset()
while not done:
# Choose an action using policy_forward.
# Take the action and observe the new state of the world.
# Compute a gradient using policy_backward. Return the gradient and reward.
return [gradient, reward_sum]
We then create a number of actors, so that we can perform rollouts in parallel.
.. code-block:: python
actors = [PongEnv() for _ in range(batch_size)]
Calling this remote function inside of a for loop, we launch multiple tasks to
perform rollouts and compute gradients in parallel.
.. code-block:: python
model_id = ray.put(model)
actions = []
# Launch tasks to compute gradients from multiple rollouts in parallel.
for i in range(batch_size):
action_id = actors[i].compute_gradient.remote(model_id)
actions.append(action_id)
Troubleshooting
---------------
If you are not seeing any speedup from Ray (and assuming you're using a
multicore machine), the problem may be that numpy is trying to use multiple
threads. When many processes are each trying to use multiple threads, the result
is often no speedup. When running this example, try opening up ``top`` and
seeing if some python processes are using more than 100% CPU. If yes, then this
is likely the problem.
The example tries to set ``MKL_NUM_THREADS=1`` in the actor. However, that only
works if the numpy on your machine is actually using MKL. If it's using
OpenBLAS, then you'll need to set ``OPENBLAS_NUM_THREADS=1``. In fact, you may
have to do this **before** running the script (it may need to happen before
numpy is imported).
.. code-block:: python
export OPENBLAS_NUM_THREADS=1
doc/source/example-streaming.rst
-151
View File
@@ -1,151 +0,0 @@
Streaming MapReduce
===================
This document walks through how to implement a simple streaming application
using Ray's actor capabilities. It implements a streaming MapReduce which
computes word counts on wikipedia articles.
You can view the `code for this example`_.
.. _`code for this example`: https://github.com/ray-project/ray/tree/master/doc/examples/streaming
To run the example, you need to install the dependencies
.. code-block:: bash
pip install wikipedia
and then execute the script as follows:
.. code-block:: bash
python ray/doc/examples/streaming/streaming.py
For each round of articles read, the script will output
the top 10 words in these articles together with their word count:
.. code-block:: text
article index = 0
the 2866
of 1688
and 1448
in 1101
to 593
a 553
is 509
as 325
are 284
by 261
article index = 1
the 3597
of 1971
and 1735
in 1429
to 670
a 623
is 578
as 401
by 293
for 285
article index = 2
the 3910
of 2123
and 1890
in 1468
to 658
a 653
is 488
as 364
by 362
for 297
article index = 3
the 2962
of 1667
and 1472
in 1220
a 546
to 538
is 516
as 307
by 253
for 243
article index = 4
the 3523
of 1866
and 1690
in 1475
to 645
a 583
is 572
as 352
by 318
for 306
...
Note that this examples uses `distributed actor handles`_, which are still
considered experimental.
.. _`distributed actor handles`: http://ray.readthedocs.io/en/latest/actors.html
There is a ``Mapper`` actor, which has a method ``get_range`` used to retrieve
word counts for words in a certain range:
.. code-block:: python
@ray.remote
class Mapper(object):
def __init__(self, title_stream):
# Constructor, the title stream parameter is a stream of wikipedia
# article titles that will be read by this mapper
def get_range(self, article_index, keys):
# Return counts of all the words with first
# letter between keys[0] and keys[1] in the
# articles that haven't been read yet with index
# up to article_index
The ``Reducer`` actor holds a list of mappers, calls ``get_range`` on them
and accumulates the results.
.. code-block:: python
@ray.remote
class Reducer(object):
def __init__(self, keys, *mappers):
# Constructor for a reducer that gets input from the list of mappers
# in the argument and accumulates word counts for words with first
# letter between keys[0] and keys[1]
def next_reduce_result(self, article_index):
# Get articles up to article_index that haven't been read yet,
# accumulate the word counts and return them
On the driver, we then create a number of mappers and reducers and run the
streaming MapReduce:
.. code-block:: python
streams = # Create list of num_mappers streams
keys = # Partition the keys among the reducers.
# Create a number of mappers.
mappers = [Mapper.remote(stream) for stream in streams]
# Create a number of reduces, each responsible for a different range of keys.
# This gives each Reducer actor a handle to each Mapper actor.
reducers = [Reducer.remote(key, *mappers) for key in keys]
article_index = 0
while True:
counts = ray.get([reducer.next_reduce_result.remote(article_index)
for reducer in reducers])
article_index += 1
The actual example reads a list of articles and creates a stream object which
produces an infinite stream of articles from the list. This is a toy example
meant to illustrate the idea. In practice we would produce a stream of
non-repeating items for each mapper.
doc/source/images/param_actor.png
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 19 KiB

doc/source/index.rst
+9 -7
View File
@@ -254,13 +254,15 @@ Getting Involved
:maxdepth: -1
:caption: Examples
example-rl-pong.rst
example-parameter-server.rst
example-newsreader.rst
example-resnet.rst
example-a3c.rst
example-lbfgs.rst
example-streaming.rst
auto_examples/overview.rst
auto_examples/plot_lbfgs.rst
auto_examples/plot_newsreader.rst
auto_examples/plot_hyperparameter.rst
auto_examples/plot_pong_example.rst
auto_examples/plot_resnet.rst
auto_examples/plot_streaming.rst
auto_examples/plot_parameter_server.rst
auto_examples/plot_example-a3c.rst
using-ray-with-tensorflow.rst
using-ray-with-pytorch.rst