From 8fc7dc3ed4a6caa7d504c1a39440902013256eb6 Mon Sep 17 00:00:00 2001
From: Crystal <crystalzyan@berkeley.edu>
Date: Sun, 16 Jul 2017 22:19:33 -0700
Subject: [PATCH] Change Python examples in documentation to use 4 space
 indentation. (#736)

* Ray doc - changed python indentation to 4 spaces in documentation files actors.rst, api.rst, and example-*.rst

* Ray documentation - changed Python to 4 space indentation for files install-*.rst, installation-troubleshooting.rst, internals-overview.rst, serialization.rst, troubleshootin.rst, tutorial.rst, using-ray-*.rst
---
 doc/source/actors.rst                       | 132 ++++++------
 doc/source/api.rst                          |  31 +--
 doc/source/example-a3c.rst                  |  94 ++++----
 doc/source/example-evolution-strategies.rst |  52 ++---
 doc/source/example-hyperopt.rst             | 106 ++++-----
 doc/source/example-lbfgs.rst                |  44 ++--
 doc/source/example-resnet.rst               |  44 ++--
 doc/source/example-rl-pong.rst              |  36 ++--
 doc/source/internals-overview.rst           |   8 +-
 doc/source/serialization.rst                |   6 +-
 doc/source/troubleshooting.rst              |  12 +-
 doc/source/tutorial.rst                     |  34 +--
 doc/source/using-ray-on-a-cluster.rst       |   4 +-
 doc/source/using-ray-on-a-large-cluster.rst |   4 +-
 doc/source/using-ray-with-gpus.rst          |  26 +--
 doc/source/using-ray-with-tensorflow.rst    | 226 ++++++++++----------
 16 files changed, 430 insertions(+), 429 deletions(-)

diff --git a/doc/source/actors.rst b/doc/source/actors.rst
index dd78e7aef..d1ee3cf85 100644
--- a/doc/source/actors.rst
+++ b/doc/source/actors.rst
@@ -28,12 +28,12 @@ that instances of the ``Counter`` class will be actors.
 
   @ray.remote
   class Counter(object):
-    def __init__(self):
-      self.value = 0
+      def __init__(self):
+          self.value = 0
 
-    def increment(self):
-      self.value += 1
-      return self.value
+      def increment(self):
+          self.value += 1
+          return self.value
 
 To actually create an actor, we can instantiate this class by calling
 ``Counter.remote()``.
@@ -115,15 +115,15 @@ encapsulate the state of these simulators.
 
   @ray.remote
   class GymEnvironment(object):
-    def __init__(self, name):
-      self.env = gym.make(name)
-      self.env.reset()
+      def __init__(self, name):
+          self.env = gym.make(name)
+          self.env.reset()
 
-    def step(self, action):
-      return self.env.step(action)
+      def step(self, action):
+          return self.env.step(action)
 
-    def reset(self):
-      self.env.reset()
+      def reset(self):
+          self.env.reset()
 
 We can then instantiate an actor and schedule a task on that actor as follows.
 
@@ -144,19 +144,19 @@ a neural net.
   import tensorflow as tf
 
   def construct_network():
-    x = tf.placeholder(tf.float32, [None, 784])
-    y_ = tf.placeholder(tf.float32, [None, 10])
+      x = tf.placeholder(tf.float32, [None, 784])
+      y_ = tf.placeholder(tf.float32, [None, 10])
 
-    W = tf.Variable(tf.zeros([784, 10]))
-    b = tf.Variable(tf.zeros([10]))
-    y = tf.nn.softmax(tf.matmul(x, W) + b)
+      W = tf.Variable(tf.zeros([784, 10]))
+      b = tf.Variable(tf.zeros([10]))
+      y = tf.nn.softmax(tf.matmul(x, W) + b)
 
-    cross_entropy = tf.reduce_mean(-tf.reduce_sum(y_ * tf.log(y), reduction_indices=[1]))
-    train_step = tf.train.GradientDescentOptimizer(0.5).minimize(cross_entropy)
-    correct_prediction = tf.equal(tf.argmax(y,1), tf.argmax(y_,1))
-    accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))
+      cross_entropy = tf.reduce_mean(-tf.reduce_sum(y_ * tf.log(y), reduction_indices=[1]))
+      train_step = tf.train.GradientDescentOptimizer(0.5).minimize(cross_entropy)
+      correct_prediction = tf.equal(tf.argmax(y,1), tf.argmax(y_,1))
+      accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))
 
-    return x, y_, train_step, accuracy
+      return x, y_, train_step, accuracy
 
 We can then define an actor for this network as follows.
 
@@ -168,19 +168,19 @@ We can then define an actor for this network as follows.
   # ray.remote without any arguments and no parentheses.
   @ray.remote(num_gpus=1)
   class NeuralNetOnGPU(object):
-    def __init__(self):
-      # Set an environment variable to tell TensorFlow which GPUs to use. Note
-      # that this must be 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"):
-          self.x, self.y_, self.train_step, self.accuracy = construct_network()
-          # Allow this to run on CPUs if there aren't any GPUs.
-          config = tf.ConfigProto(allow_soft_placement=True)
-          self.sess = tf.Session(config=config)
-          # Initialize the network.
-          init = tf.global_variables_initializer()
-          self.sess.run(init)
+      def __init__(self):
+          # Set an environment variable to tell TensorFlow which GPUs to use. Note
+          # that this must be 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"):
+                  self.x, self.y_, self.train_step, self.accuracy = construct_network()
+                  # Allow this to run on CPUs if there aren't any GPUs.
+                  config = tf.ConfigProto(allow_soft_placement=True)
+                  self.sess = tf.Session(config=config)
+                  # Initialize the network.
+                  init = tf.global_variables_initializer()
+                  self.sess.run(init)
 
 To indicate that an actor requires one GPU, we pass in ``num_gpus=1`` to
 ``ray.remote``. Note that in order for this to work, Ray must have been started
@@ -205,45 +205,45 @@ We can put this all together as follows.
   ray.init(num_gpus=8)
 
   def construct_network():
-    x = tf.placeholder(tf.float32, [None, 784])
-    y_ = tf.placeholder(tf.float32, [None, 10])
+      x = tf.placeholder(tf.float32, [None, 784])
+      y_ = tf.placeholder(tf.float32, [None, 10])
 
-    W = tf.Variable(tf.zeros([784, 10]))
-    b = tf.Variable(tf.zeros([10]))
-    y = tf.nn.softmax(tf.matmul(x, W) + b)
+      W = tf.Variable(tf.zeros([784, 10]))
+      b = tf.Variable(tf.zeros([10]))
+      y = tf.nn.softmax(tf.matmul(x, W) + b)
 
-    cross_entropy = tf.reduce_mean(-tf.reduce_sum(y_ * tf.log(y), reduction_indices=[1]))
-    train_step = tf.train.GradientDescentOptimizer(0.5).minimize(cross_entropy)
-    correct_prediction = tf.equal(tf.argmax(y,1), tf.argmax(y_,1))
-    accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))
+      cross_entropy = tf.reduce_mean(-tf.reduce_sum(y_ * tf.log(y), reduction_indices=[1]))
+      train_step = tf.train.GradientDescentOptimizer(0.5).minimize(cross_entropy)
+      correct_prediction = tf.equal(tf.argmax(y,1), tf.argmax(y_,1))
+      accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))
 
-    return x, y_, train_step, accuracy
+      return x, y_, train_step, accuracy
 
   @ray.remote(num_gpus=1)
   class NeuralNetOnGPU(object):
-    def __init__(self, mnist_data):
-      self.mnist = mnist_data
-      # Set an environment variable to tell TensorFlow which GPUs to use. Note
-      # that this must be 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"):
-          self.x, self.y_, self.train_step, self.accuracy = construct_network()
-          # Allow this to run on CPUs if there aren't any GPUs.
-          config = tf.ConfigProto(allow_soft_placement=True)
-          self.sess = tf.Session(config=config)
-          # Initialize the network.
-          init = tf.global_variables_initializer()
-          self.sess.run(init)
+      def __init__(self, mnist_data):
+          self.mnist = mnist_data
+          # Set an environment variable to tell TensorFlow which GPUs to use. Note
+          # that this must be 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"):
+                  self.x, self.y_, self.train_step, self.accuracy = construct_network()
+                  # Allow this to run on CPUs if there aren't any GPUs.
+                  config = tf.ConfigProto(allow_soft_placement=True)
+                  self.sess = tf.Session(config=config)
+                  # Initialize the network.
+                  init = tf.global_variables_initializer()
+                  self.sess.run(init)
 
-    def train(self, num_steps):
-      for _ in range(num_steps):
-        batch_xs, batch_ys = self.mnist.train.next_batch(100)
-        self.sess.run(self.train_step, feed_dict={self.x: batch_xs, self.y_: batch_ys})
+      def train(self, num_steps):
+          for _ in range(num_steps):
+              batch_xs, batch_ys = self.mnist.train.next_batch(100)
+              self.sess.run(self.train_step, feed_dict={self.x: batch_xs, self.y_: batch_ys})
 
-    def get_accuracy(self):
-      return self.sess.run(self.accuracy, feed_dict={self.x: self.mnist.test.images,
-                                                     self.y_: self.mnist.test.labels})
+      def get_accuracy(self):
+          return self.sess.run(self.accuracy, feed_dict={self.x: self.mnist.test.images,
+                                                         self.y_: self.mnist.test.labels})
 
 
   # Load the MNIST dataset and tell Ray how to serialize the custom classes.
diff --git a/doc/source/api.rst b/doc/source/api.rst
index 9004cc4e7..799fd21ce 100644
--- a/doc/source/api.rst
+++ b/doc/source/api.rst
@@ -51,9 +51,9 @@ 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
+.. code-block:: python
 
-     ray.init(redis_address="12.345.67.89:6379")
+  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,
@@ -87,7 +87,7 @@ Note that arguments to remote functions can be values or object IDs.
 
   @ray.remote
   def f(x):
-    return x + 1
+      return x + 1
 
   x_id = f.remote(0)
   ray.get(x_id)  # 1
@@ -102,7 +102,7 @@ passing the ``num_return_vals`` argument into the remote decorator.
 
   @ray.remote(num_return_vals=2)
   def f():
-    return 1, 2
+      return 1, 2
 
   x_id, y_id = f.remote()
   ray.get(x_id)  # 1
@@ -121,10 +121,11 @@ IDs.
 
   @ray.remote
   def f():
-    return {'key1': ['value']}
+      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']}]
 
@@ -169,7 +170,7 @@ copied multiple times.
 
   @ray.remote
   def f(x):
-    pass
+      pass
 
   x = np.zeros(10 ** 6)
 
@@ -217,8 +218,8 @@ lists is equal to the list passed in to ``ray.wait`` (up to ordering).
 
   @ray.remote
   def f(n):
-    time.sleep(n)
-    return n
+      time.sleep(n)
+      return n
 
   # Start 3 tasks with different durations.
   results = [f.remote(i) for i in range(3)]
@@ -237,17 +238,17 @@ tasks are executing, and whenever one task finishes, a new one is launched.
 
   @ray.remote
   def f():
-    return 1
+      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())
+      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
 
@@ -270,7 +271,7 @@ The errors will also be accumulated in Redis and can be accessed with
 
   @ray.remote
   def f():
-    raise Exception("This task failed!!")
+      raise Exception("This task failed!!")
 
   f.remote()  # An error message will be printed in the background.
 
diff --git a/doc/source/example-a3c.rst b/doc/source/example-a3c.rst
index 3b6397e0e..160986fff 100644
--- a/doc/source/example-a3c.rst
+++ b/doc/source/example-a3c.rst
@@ -75,33 +75,33 @@ We use a Ray Actor to simulate the environment.
 
   @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()
+      """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 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 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.get_gradients(batch)
-      info = {"id": self.id,
-              "size": len(batch.a)}
-      return gradient, info
+      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.get_gradients(batch)
+          info = {"id": self.id,
+                  "size": len(batch.a)}
+          return gradient, info
 
 Driver Code Walkthrough
 -----------------------
@@ -116,32 +116,32 @@ global model parameters. The main training script looks like the following.
   import ray
 
   def train(num_workers, env_name="PongDeterministic-v3"):
-    # 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
+      # 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(env_name, i) for i in range(num_workers)]
+      # Start simulations on actors
+      agents = [Runner(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.model_update(gradient)
+      # Start gradient calculation tasks on each actor
       parameters = policy.get_weights()
-      gradient_list.extend([agents[info["id"]].compute_gradient(parameters)])
-    return policy
+      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.model_update(gradient)
+          parameters = policy.get_weights()
+          gradient_list.extend([agents[info["id"]].compute_gradient(parameters)])
+      return policy
 
 
 Benchmarks and Visualization
diff --git a/doc/source/example-evolution-strategies.rst b/doc/source/example-evolution-strategies.rst
index c807500c4..71a6f7ab0 100644
--- a/doc/source/example-evolution-strategies.rst
+++ b/doc/source/example-evolution-strategies.rst
@@ -28,23 +28,23 @@ perturbed policies in a given environment.
 
   @ray.remote
   class Worker(object):
-    def __init__(self, config, policy_params, env_name, noise):
-      self.env = # Initialize environment.
-      self.policy = # Construct policy.
-      # Details omitted.
+      def __init__(self, config, policy_params, env_name, noise):
+          self.env = # Initialize environment.
+          self.policy = # Construct policy.
+          # Details omitted.
 
-    def do_rollouts(self, params):
-      # Set the network weights.
-      self.policy.set_trainable_flat(params)
-      perturbation = # Generate a random perturbation to the policy.
+      def do_rollouts(self, params):
+          # Set the network weights.
+          self.policy.set_trainable_flat(params)
+          perturbation = # Generate a random perturbation to the policy.
 
-      self.policy.set_trainable_flat(params + perturbation)
-      # Do rollout with the perturbed policy.
+          self.policy.set_trainable_flat(params + perturbation)
+          # Do rollout with the perturbed policy.
 
-      self.policy.set_trainable_flat(params - perturbation)
-      # Do rollout with the perturbed policy.
+          self.policy.set_trainable_flat(params - perturbation)
+          # Do rollout with the perturbed policy.
 
-      # Return the rewards.
+          # Return the rewards.
 
 In the main loop, we create a number of actors with this class.
 
@@ -59,17 +59,17 @@ and use the rewards from the rollouts to update the policy.
 .. code-block:: python
 
   while True:
-    # Get the current policy weights.
-    theta = policy.get_trainable_flat()
-    # Put the current policy weights in the object store.
-    theta_id = ray.put(theta)
-    # Use the actors to do rollouts, note that we pass in the ID of the policy
-    # weights.
-    rollout_ids = [worker.do_rollouts.remote(theta_id), for worker in workers]
-    # Get the results of the rollouts.
-    results = ray.get(rollout_ids)
-    # Update the policy.
-    optimizer.update(...)
+      # Get the current policy weights.
+      theta = policy.get_trainable_flat()
+      # Put the current policy weights in the object store.
+      theta_id = ray.put(theta)
+      # Use the actors to do rollouts, note that we pass in the ID of the policy
+      # weights.
+      rollout_ids = [worker.do_rollouts.remote(theta_id), for worker in workers]
+      # Get the results of the rollouts.
+      results = ray.get(rollout_ids)
+      # Update the policy.
+      optimizer.update(...)
 
 In addition, note that we create a large object representing a shared block of
 random noise. We then put the block in the object store so that each ``Worker``
@@ -79,8 +79,8 @@ actor can use it without creating its own copy.
 
   @ray.remote
   def create_shared_noise():
-    noise = np.random.randn(250000000)
-    return noise
+      noise = np.random.randn(250000000)
+      return noise
 
   noise_id = create_shared_noise.remote()
 
diff --git a/doc/source/example-hyperopt.rst b/doc/source/example-hyperopt.rst
index f8cbbb2c0..950281c19 100644
--- a/doc/source/example-hyperopt.rst
+++ b/doc/source/example-hyperopt.rst
@@ -66,9 +66,9 @@ returns the accuracy of the trained model on a validation set.
                                      train_labels,
                                      validation_images,
                                      validation_labels):
-    # Construct a deep network, train it, and return the accuracy on the
-    # validation data.
-    return np.random.uniform(0, 1)
+      # Construct a deep network, train it, and return the accuracy on the
+      # validation data.
+      return np.random.uniform(0, 1)
 
 Basic random search
 -------------------
@@ -80,11 +80,11 @@ hyperparameter configurations.
 .. code-block:: python
 
   def generate_hyperparameters():
-    # Randomly choose values for the hyperparameters.
-    return {"learning_rate": 10 ** np.random.uniform(-5, 5),
-            "batch_size": np.random.randint(1, 100),
-            "dropout": np.random.uniform(0, 1),
-            "stddev": 10 ** np.random.uniform(-5, 5)}
+      # Randomly choose values for the hyperparameters.
+      return {"learning_rate": 10 ** np.random.uniform(-5, 5),
+              "batch_size": np.random.randint(1, 100),
+              "dropout": np.random.uniform(0, 1),
+              "stddev": 10 ** np.random.uniform(-5, 5)}
 
 In addition, let's assume that we've started Ray and loaded some data.
 
@@ -113,11 +113,11 @@ bunch of experiments, and we get the results.
   # Launch some experiments.
   results = []
   for hyperparameters in hyperparameter_configurations:
-    results.append(train_cnn_and_compute_accuracy.remote(hyperparameters,
-                                                         train_images,
-                                                         train_labels,
-                                                         validation_images,
-                                                         validation_labels))
+      results.append(train_cnn_and_compute_accuracy.remote(hyperparameters,
+                                                           train_images,
+                                                           train_labels,
+                                                           validation_images,
+                                                           validation_labels))
 
   # Get the results.
   accuracies = ray.get(results)
@@ -145,25 +145,25 @@ detail in driver.py_.
   # Launch some experiments.
   remaining_ids = []
   for hyperparameters in hyperparameter_configurations:
-    remaining_ids.append(train_cnn_and_compute_accuracy.remote(hyperparameters,
-                                                               train_images,
-                                                               train_labels,
-                                                               validation_images,
-                                                               validation_labels))
+      remaining_ids.append(train_cnn_and_compute_accuracy.remote(hyperparameters,
+                                                                 train_images,
+                                                                 train_labels,
+                                                                 validation_images,
+                                                                 validation_labels))
 
   # Whenever a new experiment finishes, print the value and start a new
   # experiment.
   for i in range(100):
-    ready_ids, remaining_ids = ray.wait(remaining_ids, num_returns=1)
-    accuracy = ray.get(ready_ids[0])
-    print("Accuracy is {}".format(accuracy))
-    # Start a new experiment.
-    new_hyperparameters = generate_hyperparameters()
-    remaining_ids.append(train_cnn_and_compute_accuracy.remote(new_hyperparameters,
-                                                               train_images,
-                                                               train_labels,
-                                                               validation_images,
-                                                               validation_labels))
+      ready_ids, remaining_ids = ray.wait(remaining_ids, num_returns=1)
+      accuracy = ray.get(ready_ids[0])
+      print("Accuracy is {}".format(accuracy))
+      # Start a new experiment.
+      new_hyperparameters = generate_hyperparameters()
+      remaining_ids.append(train_cnn_and_compute_accuracy.remote(new_hyperparameters,
+                                                                 train_images,
+                                                                 train_labels,
+                                                                 validation_images,
+                                                                 validation_labels))
 
 .. _driver.py: https://github.com/ray-project/ray/blob/master/examples/hyperopt/driver.py
 
@@ -191,12 +191,12 @@ model and to return the updated model.
 
   @ray.remote
   def train_cnn_and_compute_accuracy(hyperparameters, model=None):
-    # Construct a deep network, train it, and return the accuracy on the
-    # validation data as well as the latest version of the model. If the model
-    # argument is not None, this will continue training an existing model.
-    validation_accuracy = np.random.uniform(0, 1)
-    new_model = model
-    return validation_accuracy, new_model
+      # Construct a deep network, train it, and return the accuracy on the
+      # validation data as well as the latest version of the model. If the model
+      # argument is not None, this will continue training an existing model.
+      validation_accuracy = np.random.uniform(0, 1)
+      new_model = model
+      return validation_accuracy, new_model
 
 Here's a different variant that uses the same principles. Divide each training
 session into a series of shorter training sessions. Whenever a short session
@@ -208,33 +208,33 @@ doing well, then terminate it and start a new experiment.
   import numpy as np
 
   def is_promising(model):
-    # Return true if the model is doing well and false otherwise. In practice,
-    # this function will want more information than just the model.
-    return np.random.choice([True, False])
+      # Return true if the model is doing well and false otherwise. In practice,
+      # this function will want more information than just the model.
+      return np.random.choice([True, False])
 
   # Start 10 experiments.
   remaining_ids = []
   for _ in range(10):
-    experiment_id = train_cnn_and_compute_accuracy.remote(hyperparameters, model=None)
-    remaining_ids.append(experiment_id)
+      experiment_id = train_cnn_and_compute_accuracy.remote(hyperparameters, model=None)
+      remaining_ids.append(experiment_id)
 
   accuracies = []
   for i in range(100):
-    # Whenever a segment of an experiment finishes, decide if it looks promising
-    # or not.
-    ready_ids, remaining_ids = ray.wait(remaining_ids, num_returns=1)
-    experiment_id = ready_ids[0]
-    current_accuracy, current_model = ray.get(experiment_id)
-    accuracies.append(current_accuracy)
+      # Whenever a segment of an experiment finishes, decide if it looks promising
+      # or not.
+      ready_ids, remaining_ids = ray.wait(remaining_ids, num_returns=1)
+      experiment_id = ready_ids[0]
+      current_accuracy, current_model = ray.get(experiment_id)
+      accuracies.append(current_accuracy)
 
-    if is_promising(experiment_id):
-      # Continue running the experiment.
-      experiment_id = train_cnn_and_compute_accuracy.remote(hyperparameters,
-                                                            model=current_model)
-    else:
-      # Start a new experiment.
-      experiment_id = train_cnn_and_compute_accuracy.remote(hyperparameters)
+      if is_promising(experiment_id):
+          # Continue running the experiment.
+          experiment_id = train_cnn_and_compute_accuracy.remote(hyperparameters,
+                                                                model=current_model)
+      else:
+          # Start a new experiment.
+          experiment_id = train_cnn_and_compute_accuracy.remote(hyperparameters)
 
-    remaining_ids.append(experiment_id)
+      remaining_ids.append(experiment_id)
 
 .. _Hyperband: https://arxiv.org/abs/1603.06560
diff --git a/doc/source/example-lbfgs.rst b/doc/source/example-lbfgs.rst
index 93e9fad78..8de2efe7f 100644
--- a/doc/source/example-lbfgs.rst
+++ b/doc/source/example-lbfgs.rst
@@ -53,20 +53,20 @@ 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
+      # 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
+      # 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])
+      # 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])
+      # 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
@@ -102,16 +102,16 @@ Now, lets turn ``loss`` and ``grad`` into methods of an actor that will contain
 .. code-block:: python
 
   class Network(object):
-    def __init__():
-      # Initialize network.
+      def __init__():
+          # Initialize network.
 
-    def loss(theta, xs, ys):
-      # compute the loss
-      return loss
+      def loss(theta, xs, ys):
+          # compute the loss
+          return loss
 
-    def grad(theta, xs, ys):
-      # compute the gradient
-      return grad
+      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.
@@ -119,14 +119,14 @@ 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))
+      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.
+      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
diff --git a/doc/source/example-resnet.rst b/doc/source/example-resnet.rst
index 0aff3c88b..272bbcd47 100644
--- a/doc/source/example-resnet.rst
+++ b/doc/source/example-resnet.rst
@@ -55,26 +55,26 @@ The core of the script is the actor definition.
 
   @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 __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()
+      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.
 
@@ -89,9 +89,9 @@ 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)
+      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 ResNet example`: https://github.com/tensorflow/models/tree/master/resnet
 .. _`TensorFlow`: https://www.tensorflow.org/install/
diff --git a/doc/source/example-rl-pong.rst b/doc/source/example-rl-pong.rst
index babd24166..d8da3a5c2 100644
--- a/doc/source/example-rl-pong.rst
+++ b/doc/source/example-rl-pong.rst
@@ -60,23 +60,23 @@ the actor.
 
   @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 __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]
+      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.
 
@@ -93,8 +93,8 @@ perform rollouts and compute gradients in parallel.
   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)
+      action_id = actors[i].compute_gradient.remote(model_id)
+      actions.append(action_id)
 
 
 Troubleshooting
diff --git a/doc/source/internals-overview.rst b/doc/source/internals-overview.rst
index 50464f591..69ac1895a 100644
--- a/doc/source/internals-overview.rst
+++ b/doc/source/internals-overview.rst
@@ -51,7 +51,7 @@ Now, consider a remote function definition as below.
 
   @ray.remote
   def f(x):
-    return x + 1
+      return x + 1
 
 When the remote function is defined as above, the function is immediately
 pickled, assigned a unique ID, and stored in a Redis server. You can view the
@@ -77,17 +77,17 @@ Notes and limitations
 
     @ray.remote
     def f(x):
-      return helper(x)
+        return helper(x)
 
     def helper(x):
-      return x + 1
+        return x + 1
 
   If you call ``f.remote(0)``, it will give an error of the form.
 
   .. code-block:: python
 
     Traceback (most recent call last):
-      File "<ipython-input-3-12a5beeb2306>", line 3, in f
+        File "<ipython-input-3-12a5beeb2306>", line 3, in f
     NameError: name 'helper' is not defined
 
   On the other hand, if ``helper`` is defined before ``f``, then it will work.
diff --git a/doc/source/serialization.rst b/doc/source/serialization.rst
index 79993deb4..6f6d4f8c8 100644
--- a/doc/source/serialization.rst
+++ b/doc/source/serialization.rst
@@ -116,9 +116,9 @@ calling pickle by hand).
 
   @ray.remote
   def f(complicated_object):
-    # Deserialize the object manually.
-    obj = pickle.loads(complicated_object)
-    return "Successfully passed {} into f.".format(obj)
+      # Deserialize the object manually.
+      obj = pickle.loads(complicated_object)
+      return "Successfully passed {} into f.".format(obj)
 
   # Define a complicated object.
   l = []
diff --git a/doc/source/troubleshooting.rst b/doc/source/troubleshooting.rst
index b26ece658..ff4b3039e 100644
--- a/doc/source/troubleshooting.rst
+++ b/doc/source/troubleshooting.rst
@@ -135,11 +135,11 @@ the newest version.
 
   @ray.remote
   def f():
-    return 1
+      return 1
 
   @ray.remote
   def f():
-    return 2
+      return 2
 
   ray.get(f.remote())  # This should be 2.
 
@@ -182,7 +182,7 @@ Ray).
   .. code-block:: python
 
     def h():
-      return 1
+        return 1
 
   And you define remote function ``f`` as
 
@@ -190,7 +190,7 @@ Ray).
 
     @ray.remote
     def f():
-      return file.h()
+        return file.h()
 
   You can redefine ``f`` as follows.
 
@@ -198,8 +198,8 @@ Ray).
 
     @ray.remote
     def f():
-      reload(file)
-      return file.h()
+        reload(file)
+        return file.h()
 
   This forces the reload to happen on the workers as needed. Note that in
   Python 3, you need to do ``from importlib import reload``.
diff --git a/doc/source/tutorial.rst b/doc/source/tutorial.rst
index 6db9cbd00..0493b6916 100644
--- a/doc/source/tutorial.rst
+++ b/doc/source/tutorial.rst
@@ -122,7 +122,7 @@ For example, a normal Python function looks like this.
 .. code-block:: python
 
   def add1(a, b):
-    return a + b
+      return a + b
 
 A remote function looks like this.
 
@@ -130,7 +130,7 @@ A remote function looks like this.
 
   @ray.remote
   def add2(a, b):
-    return a + b
+      return a + b
 
 Remote functions
 ~~~~~~~~~~~~~~~~
@@ -155,11 +155,11 @@ to parallelize computation.
   import time
 
   def f1():
-    time.sleep(1)
+      time.sleep(1)
 
   @ray.remote
   def f2():
-    time.sleep(1)
+      time.sleep(1)
 
   # The following takes ten seconds.
   [f1() for _ in range(10)]
@@ -197,7 +197,7 @@ Note that a remote function can return multiple object IDs.
 
   @ray.remote(num_return_vals=3)
   def return_multiple():
-    return 1, 2, 3
+      return 1, 2, 3
 
   a_id, b_id, c_id = return_multiple.remote()
 
@@ -212,7 +212,7 @@ three tasks as follows, each of which depends on the previous task.
 
   @ray.remote
   def f(x):
-    return x + 1
+      return x + 1
 
   x = f.remote(0)
   y = f.remote(x)
@@ -232,11 +232,11 @@ Consider the following implementation of a tree reduce.
 
   @ray.remote
   def generate_data():
-    return np.random.normal(size=1000)
+      return np.random.normal(size=1000)
 
   @ray.remote
   def aggregate_data(x, y):
-    return 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.
@@ -244,7 +244,7 @@ Consider the following implementation of a tree reduce.
 
   # Perform a tree reduce.
   while len(data) > 1:
-    data.append(aggregate_data.remote(data.pop(0), data.pop(0)))
+      data.append(aggregate_data.remote(data.pop(0), data.pop(0)))
 
   # Fetch the result.
   ray.get(data)
@@ -260,17 +260,17 @@ following example.
 
   @ray.remote
   def sub_experiment(i, j):
-    # Run the jth sub-experiment for the ith experiment.
-    return 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))
+      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]
diff --git a/doc/source/using-ray-on-a-cluster.rst b/doc/source/using-ray-on-a-cluster.rst
index a25b702e1..4e57e3bda 100644
--- a/doc/source/using-ray-on-a-cluster.rst
+++ b/doc/source/using-ray-on-a-cluster.rst
@@ -68,8 +68,8 @@ following.
 
   @ray.remote
   def f():
-    time.sleep(0.01)
-    return ray.services.get_node_ip_address()
+      time.sleep(0.01)
+      return ray.services.get_node_ip_address()
 
   # Get a list of the IP addresses of the nodes that have joined the cluster.
   set(ray.get([f.remote() for _ in range(1000)]))
diff --git a/doc/source/using-ray-on-a-large-cluster.rst b/doc/source/using-ray-on-a-large-cluster.rst
index c68ae0b03..0d31c4c9b 100644
--- a/doc/source/using-ray-on-a-large-cluster.rst
+++ b/doc/source/using-ray-on-a-large-cluster.rst
@@ -174,8 +174,8 @@ following.
 
   @ray.remote
   def f():
-    time.sleep(0.01)
-    return ray.services.get_node_ip_address()
+      time.sleep(0.01)
+      return ray.services.get_node_ip_address()
 
   # Get a list of the IP addresses of the nodes that have joined the cluster.
   set(ray.get([f.remote() for _ in range(1000)]))
diff --git a/doc/source/using-ray-with-gpus.rst b/doc/source/using-ray-with-gpus.rst
index 6dea3a6e6..1a063a957 100644
--- a/doc/source/using-ray-with-gpus.rst
+++ b/doc/source/using-ray-with-gpus.rst
@@ -42,7 +42,7 @@ remote decorator.
 
   @ray.remote(num_gpus=1)
   def gpu_method():
-    return "This function is allowed to use GPUs {}.".format(ray.get_gpu_ids())
+      return "This function is allowed to use GPUs {}.".format(ray.get_gpu_ids())
 
 Inside of the remote function, a call to ``ray.get_gpu_ids()`` will return a
 list of integers indicating which GPUs the remote function is allowed to use.
@@ -62,10 +62,10 @@ TensorFlow.
 
   @ray.remote(num_gpus=1)
   def gpu_method():
-    os.environ["CUDA_VISIBLE_DEVICES"] = ",".join(map(str, ray.get_gpu_ids()))
-    # Create a TensorFlow session. TensorFlow will restrict itself to use the
-    # GPUs specified by the CUDA_VISIBLE_DEVICES environment variable.
-    tf.Session()
+      os.environ["CUDA_VISIBLE_DEVICES"] = ",".join(map(str, ray.get_gpu_ids()))
+      # Create a TensorFlow session. TensorFlow will restrict itself to use the
+      # GPUs specified by the CUDA_VISIBLE_DEVICES environment variable.
+      tf.Session()
 
 **Note:** It is certainly possible for the person implementing ``gpu_method`` to
 ignore ``ray.get_gpu_ids`` and to use all of the GPUs on the machine. Ray does
@@ -84,8 +84,8 @@ instance requires in the ``ray.remote`` decorator.
 
   @ray.remote(num_gpus=1)
   class GPUActor(object):
-    def __init__(self):
-      return "This actor is allowed to use GPUs {}.".format(ray.get_gpu_ids())
+      def __init__(self):
+          return "This actor is allowed to use GPUs {}.".format(ray.get_gpu_ids())
 
 When the actor is created, GPUs will be reserved for that actor for the lifetime
 of the actor.
@@ -101,12 +101,12 @@ The following is an example of how to use GPUs in an actor through TensorFlow.
 
   @ray.remote(num_gpus=1)
   class GPUActor(object):
-    def __init__(self):
-      self.gpu_ids = ray.get_gpu_ids()
-      os.environ["CUDA_VISIBLE_DEVICES"] = ",".join(map(str, self.gpu_ids))
-      # The call to tf.Session() will restrict TensorFlow to use the GPUs
-      # specified in the CUDA_VISIBLE_DEVICES environment variable.
-      self.sess = tf.Session()
+      def __init__(self):
+          self.gpu_ids = ray.get_gpu_ids()
+          os.environ["CUDA_VISIBLE_DEVICES"] = ",".join(map(str, self.gpu_ids))
+          # The call to tf.Session() will restrict TensorFlow to use the GPUs
+          # specified in the CUDA_VISIBLE_DEVICES environment variable.
+          self.sess = tf.Session()
 
 Troubleshooting
 ---------------
diff --git a/doc/source/using-ray-with-tensorflow.rst b/doc/source/using-ray-with-tensorflow.rst
index 39caf7724..429e11461 100644
--- a/doc/source/using-ray-with-tensorflow.rst
+++ b/doc/source/using-ray-with-tensorflow.rst
@@ -105,50 +105,50 @@ complex Python objects.
   NUM_ITERS = 201
 
   class Network(object):
-    def __init__(self, x, y):
-      # Seed TensorFlow to make the script deterministic.
-      tf.set_random_seed(0)
-      # Define the inputs.
-      self.x_data = tf.constant(x, dtype=tf.float32)
-      self.y_data = tf.constant(y, dtype=tf.float32)
-      # Define the weights and computation.
-      w = tf.Variable(tf.random_uniform([1], -1.0, 1.0))
-      b = tf.Variable(tf.zeros([1]))
-      y = w * self.x_data + b
-      # Define the loss.
-      self.loss = tf.reduce_mean(tf.square(y - self.y_data))
-      optimizer = tf.train.GradientDescentOptimizer(0.5)
-      self.grads = optimizer.compute_gradients(self.loss)
-      self.train = optimizer.apply_gradients(self.grads)
-      # Define the weight initializer and session.
-      init = tf.global_variables_initializer()
-      self.sess = tf.Session()
-      # Additional code for setting and getting the weights
-      self.variables = ray.experimental.TensorFlowVariables(self.loss, self.sess)
-      # Return all of the data needed to use the network.
-      self.sess.run(init)
+      def __init__(self, x, y):
+          # Seed TensorFlow to make the script deterministic.
+          tf.set_random_seed(0)
+          # Define the inputs.
+          self.x_data = tf.constant(x, dtype=tf.float32)
+          self.y_data = tf.constant(y, dtype=tf.float32)
+          # Define the weights and computation.
+          w = tf.Variable(tf.random_uniform([1], -1.0, 1.0))
+          b = tf.Variable(tf.zeros([1]))
+          y = w * self.x_data + b
+          # Define the loss.
+          self.loss = tf.reduce_mean(tf.square(y - self.y_data))
+          optimizer = tf.train.GradientDescentOptimizer(0.5)
+          self.grads = optimizer.compute_gradients(self.loss)
+          self.train = optimizer.apply_gradients(self.grads)
+          # Define the weight initializer and session.
+          init = tf.global_variables_initializer()
+          self.sess = tf.Session()
+          # Additional code for setting and getting the weights
+          self.variables = ray.experimental.TensorFlowVariables(self.loss, self.sess)
+          # Return all of the data needed to use the network.
+          self.sess.run(init)
 
-    # Define a remote function that trains the network for one step and returns the
-    # new weights.
-    def step(self, weights):
-      # Set the weights in the network.
-      self.variables.set_weights(weights)
-      # Do one step of training.
-      self.sess.run(self.train)
-      # Return the new weights.
-      return self.variables.get_weights()
+      # Define a remote function that trains the network for one step and returns the
+      # new weights.
+      def step(self, weights):
+          # Set the weights in the network.
+          self.variables.set_weights(weights)
+          # Do one step of training.
+          self.sess.run(self.train)
+          # Return the new weights.
+          return self.variables.get_weights()
 
-    def get_weights(self):
-      return self.variables.get_weights()
+      def get_weights(self):
+          return self.variables.get_weights()
 
   # Define a remote function for generating fake data.
   @ray.remote(num_return_vals=2)
   def generate_fake_x_y_data(num_data, seed=0):
-    # Seed numpy to make the script deterministic.
-    np.random.seed(seed)
-    x = np.random.rand(num_data)
-    y = x * 0.1 + 0.3
-    return x, y
+      # Seed numpy to make the script deterministic.
+      np.random.seed(seed)
+      x = np.random.rand(num_data)
+      y = x * 0.1 + 0.3
+      return x, y
 
   # Generate some training data.
   batch_ids = [generate_fake_x_y_data.remote(BATCH_SIZE, seed=i) for i in range(NUM_BATCHES)]
@@ -166,24 +166,24 @@ complex Python objects.
 
   # Do some steps of training.
   for iteration in range(NUM_ITERS):
-    # Put the weights in the object store. This is optional. We could instead pass
-    # the variable weights directly into step.remote, in which case it would be
-    # placed in the object store under the hood. However, in that case multiple
-    # copies of the weights would be put in the object store, so this approach is
-    # more efficient.
-    weights_id = ray.put(weights)
-    # Call the remote function multiple times in parallel.
-    new_weights_ids = [actor.step.remote(weights_id) for actor in actor_list]
-    # Get all of the weights.
-    new_weights_list = ray.get(new_weights_ids)
-    # Add up all the different weights. Each element of new_weights_list is a dict
-    # of weights, and we want to add up these dicts component wise using the keys
-    # of the first dict.
-    weights = {variable: sum(weight_dict[variable] for weight_dict in new_weights_list) / NUM_BATCHES for variable in new_weights_list[0]}
-    # Print the current weights. They should converge to roughly to the values 0.1
-    # and 0.3 used in generate_fake_x_y_data.
-    if iteration % 20 == 0:
-      print("Iteration {}: weights are {}".format(iteration, weights))
+      # Put the weights in the object store. This is optional. We could instead pass
+      # the variable weights directly into step.remote, in which case it would be
+      # placed in the object store under the hood. However, in that case multiple
+      # copies of the weights would be put in the object store, so this approach is
+      # more efficient.
+      weights_id = ray.put(weights)
+      # Call the remote function multiple times in parallel.
+      new_weights_ids = [actor.step.remote(weights_id) for actor in actor_list]
+      # Get all of the weights.
+      new_weights_list = ray.get(new_weights_ids)
+      # Add up all the different weights. Each element of new_weights_list is a dict
+      # of weights, and we want to add up these dicts component wise using the keys
+      # of the first dict.
+      weights = {variable: sum(weight_dict[variable] for weight_dict in new_weights_list) / NUM_BATCHES for variable in new_weights_list[0]}
+      # Print the current weights. They should converge to roughly to the values 0.1
+      # and 0.3 used in generate_fake_x_y_data.
+      if iteration % 20 == 0:
+          print("Iteration {}: weights are {}".format(iteration, weights))
 
 How to Train in Parallel using Ray
 ----------------------------------
@@ -236,49 +236,49 @@ For reference, the full code is below:
   NUM_ITERS = 201
 
   class Network(object):
-    def __init__(self, x, y):
-      # Seed TensorFlow to make the script deterministic.
-      tf.set_random_seed(0)
-      # Define the inputs.
-      x_data = tf.constant(x, dtype=tf.float32)
-      y_data = tf.constant(y, dtype=tf.float32)
-      # Define the weights and computation.
-      w = tf.Variable(tf.random_uniform([1], -1.0, 1.0))
-      b = tf.Variable(tf.zeros([1]))
-      y = w * x_data + b
-      # Define the loss.
-      self.loss = tf.reduce_mean(tf.square(y - y_data))
-      optimizer = tf.train.GradientDescentOptimizer(0.5)
-      self.grads = optimizer.compute_gradients(self.loss)
-      self.train = optimizer.apply_gradients(self.grads)
-      # Define the weight initializer and session.
-      init = tf.global_variables_initializer()
-      self.sess = tf.Session()
-      # Additional code for setting and getting the weights
-      self.variables = ray.experimental.TensorFlowVariables(self.loss, self.sess)
-      # Return all of the data needed to use the network.
-      self.sess.run(init)
+      def __init__(self, x, y):
+          # Seed TensorFlow to make the script deterministic.
+          tf.set_random_seed(0)
+          # Define the inputs.
+          x_data = tf.constant(x, dtype=tf.float32)
+          y_data = tf.constant(y, dtype=tf.float32)
+          # Define the weights and computation.
+          w = tf.Variable(tf.random_uniform([1], -1.0, 1.0))
+          b = tf.Variable(tf.zeros([1]))
+          y = w * x_data + b
+          # Define the loss.
+          self.loss = tf.reduce_mean(tf.square(y - y_data))
+          optimizer = tf.train.GradientDescentOptimizer(0.5)
+          self.grads = optimizer.compute_gradients(self.loss)
+          self.train = optimizer.apply_gradients(self.grads)
+          # Define the weight initializer and session.
+          init = tf.global_variables_initializer()
+          self.sess = tf.Session()
+          # Additional code for setting and getting the weights
+          self.variables = ray.experimental.TensorFlowVariables(self.loss, self.sess)
+          # Return all of the data needed to use the network.
+          self.sess.run(init)
 
-    # Define a remote function that trains the network for one step and returns the
-    # new weights.
-    def step(self, weights):
-      # Set the weights in the network.
-      self.variables.set_weights(weights)
-      # Do one step of training. We only need the actual gradients so we filter over the list.
-      actual_grads = self.sess.run([grad[0] for grad in self.grads])
-      return actual_grads
+      # Define a remote function that trains the network for one step and returns the
+      # new weights.
+      def step(self, weights):
+          # Set the weights in the network.
+          self.variables.set_weights(weights)
+          # Do one step of training. We only need the actual gradients so we filter over the list.
+          actual_grads = self.sess.run([grad[0] for grad in self.grads])
+          return actual_grads
 
-    def get_weights(self):
-      return self.variables.get_weights()
+      def get_weights(self):
+          return self.variables.get_weights()
 
   # Define a remote function for generating fake data.
   @ray.remote(num_return_vals=2)
   def generate_fake_x_y_data(num_data, seed=0):
-    # Seed numpy to make the script deterministic.
-    np.random.seed(seed)
-    x = np.random.rand(num_data)
-    y = x * 0.1 + 0.3
-    return x, y
+      # Seed numpy to make the script deterministic.
+      np.random.seed(seed)
+      x = np.random.rand(num_data)
+      y = x * 0.1 + 0.3
+      return x, y
 
   # Generate some training data.
   batch_ids = [generate_fake_x_y_data.remote(BATCH_SIZE, seed=i) for i in range(NUM_BATCHES)]
@@ -297,26 +297,26 @@ For reference, the full code is below:
 
   # Do some steps of training.
   for iteration in range(NUM_ITERS):
-    # Put the weights in the object store. This is optional. We could instead pass
-    # the variable weights directly into step.remote, in which case it would be
-    # placed in the object store under the hood. However, in that case multiple
-    # copies of the weights would be put in the object store, so this approach is
-    # more efficient.
-    weights_id = ray.put(weights)
-    # Call the remote function multiple times in parallel.
-    gradients_ids = [actor.step.remote(weights_id) for actor in actor_list]
-    # Get all of the weights.
-    gradients_list = ray.get(gradients_ids)
+      # Put the weights in the object store. This is optional. We could instead pass
+      # the variable weights directly into step.remote, in which case it would be
+      # placed in the object store under the hood. However, in that case multiple
+      # copies of the weights would be put in the object store, so this approach is
+      # more efficient.
+      weights_id = ray.put(weights)
+      # Call the remote function multiple times in parallel.
+      gradients_ids = [actor.step.remote(weights_id) for actor in actor_list]
+      # Get all of the weights.
+      gradients_list = ray.get(gradients_ids)
 
-    # Take the mean of the different gradients. Each element of gradients_list is a list
-    # of gradients, and we want to take the mean of each one.
-    mean_grads = [sum([gradients[i] for gradients in gradients_list]) / len(gradients_list) for i in range(len(gradients_list[0]))]
+      # Take the mean of the different gradients. Each element of gradients_list is a list
+      # of gradients, and we want to take the mean of each one.
+      mean_grads = [sum([gradients[i] for gradients in gradients_list]) / len(gradients_list) for i in range(len(gradients_list[0]))]
 
-    feed_dict = {grad[0]: mean_grad for (grad, mean_grad) in zip(local_network.grads, mean_grads)}
-    local_network.sess.run(local_network.train, feed_dict=feed_dict)
-    weights = local_network.get_weights()
+      feed_dict = {grad[0]: mean_grad for (grad, mean_grad) in zip(local_network.grads, mean_grads)}
+      local_network.sess.run(local_network.train, feed_dict=feed_dict)
+      weights = local_network.get_weights()
 
-    # Print the current weights. They should converge to roughly to the values 0.1
-    # and 0.3 used in generate_fake_x_y_data.
-    if iteration % 20 == 0:
-      print("Iteration {}: weights are {}".format(iteration, weights))
+      # Print the current weights. They should converge to roughly to the values 0.1
+      # and 0.3 used in generate_fake_x_y_data.
+      if iteration % 20 == 0:
+          print("Iteration {}: weights are {}".format(iteration, weights))