Change API for remote function declaration, actor instantiation, and actor method invocation. (ray-project#541)

* Direction substitution of @ray.remote -> @ray.task.

* Changes to make '@ray.task' work.

* Instantiate actors with Class.remote() instead of Class().

* Convert actor instantiation in tests and examples from Class() to Class.remote().

* Change actor method invocation from object.method() to object.method.remote().

* Update tests and examples to invoke actor methods with .remote().

* Fix bugs in jenkins tests.

* Fix example applications.

* Change @ray.task back to @ray.remote.

* Changes to make @ray.actor -> @ray.remote work.

* Direct substitution of @ray.actor -> @ray.remote.

* Fixes.

* Raise exception if @ray.actor decorator is used.

* Simplify ActorMethod class.

Author: robertnishihara

doc/source/actors.rst

@@ -27,7 +27,7 @@ An actor can be defined as follows.
 
   import gym
 
-  @ray.actor
+  @ray.remote
   class GymEnvironment(object):
     def __init__(self, name):
       self.env = gym.make(name)
@@ -63,27 +63,28 @@ We can use the actor by calling one of its methods.
 
 .. code-block:: python
 
-  a1.step(0)
-  a2.step(0)
+  a1.step.remote(0)
+  a2.step.remote(0)
 
-When ``a1.step(0)`` is called, a task is created and scheduled on the first
-actor. This scheduling procedure bypasses the global scheduler, and is assigned
-directly to the local scheduler responsible for the actor by the driver's local
-scheduler. Since the method call is a task, ``a1.step(0)`` returns an object ID.
-We can call `ray.get` on the object ID to retrieve the actual value.
+When ``a1.step.remote(0)`` is called, a task is created and scheduled on the
+first actor. This scheduling procedure bypasses the global scheduler, and is
+assigned directly to the local scheduler responsible for the actor by the
+driver's local scheduler. Since the method call is a task, ``a1.step(0)``
+returns an object ID. We can call `ray.get` on the object ID to retrieve the
+actual value.
 
-The call to ``a2.step(0)`` generates a task which is scheduled on the second
-actor. Since these two tasks run on different actors, they can be executed in
-parallel (note that only actor methods will be scheduled on actor workers, not
-regular remote functions).
+The call to ``a2.step.remote(0)`` generates a task which is scheduled on the
+second actor. Since these two tasks run on different actors, they can be
+executed in parallel (note that only actor methods will be scheduled on actor
+workers, not regular remote functions).
 
 On the other hand, methods called on the same actor are executed serially and
 share in the order that they are called and share state with one another. We
 illustrate this with a simple example.
 
 .. code-block:: python
 
-  @ray.actor
+  @ray.remote
   class Counter(object):
     def __init__(self):
       self.value = 0
@@ -96,12 +97,12 @@ illustrate this with a simple example.
 
   # Increment each counter once and get the results. These tasks all happen in
   # parallel.
-  results = ray.get([c.increment() for c in counters])
+  results = ray.get([c.increment.remote() for c in counters])
   print(results)  # prints [1, 1, 1, 1, 1, 1, 1, 1, 1, 1]
 
   # Increment the first counter five times. These tasks are executed serially
   # and share state.
-  results = ray.get([counters[0].increment() for _ in range(5)])
+  results = ray.get([counters[0].increment.remote() for _ in range(5)])
   print(results)  # prints [2, 3, 4, 5, 6]
 
 Using GPUs on actors
@@ -136,8 +137,8 @@ We can then define an actor for this network as follows.
   import os
 
   # Define an actor that runs on GPUs. If there are no GPUs, then simply use
-  # ray.actor without any arguments and no parentheses.
-  @ray.actor(num_gpus=1)
+  # 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
@@ -154,15 +155,15 @@ We can then define an actor for this network as follows.
           self.sess.run(init)
 
 To indicate that an actor requires one GPU, we pass in ``num_gpus=1`` to
-``ray.actor``. Note that in order for this to work, Ray must have been started
+``ray.remote``. Note that in order for this to work, Ray must have been started
 with some GPUs, e.g., via ``ray.init(num_gpus=2)``. Otherwise, when you try to
-instantiate the GPU version with ``NeuralNetOnGPU()``, an exception will be
-thrown saying that there aren't enough GPUs in the system.
+instantiate the GPU version with ``NeuralNetOnGPU.remote()``, an exception will
+be thrown saying that there aren't enough GPUs in the system.
 
 When the actor is created, it will have access to a list of the IDs of the GPUs
 that it is allowed to use via ``ray.get_gpu_ids()``. This is a list of integers,
 like ``[]``, or ``[1]``, or ``[2, 5, 6]``. Since we passed in
-``ray.actor(num_gpus=1)``, this list will have length one.
+``ray.remote(num_gpus=1)``, this list will have length one.
 
 We can put this all together as follows.
 
@@ -190,7 +191,7 @@ We can put this all together as follows.
 
     return x, y_, train_step, accuracy
 
-  @ray.actor(num_gpus=1)
+  @ray.remote(num_gpus=1)
   class NeuralNetOnGPU(object):
     def __init__(self, mnist_data):
       self.mnist = mnist_data
@@ -223,9 +224,9 @@ We can put this all together as follows.
   ray.register_class(type(mnist.train))
 
   # Create the actor.
-  nn = NeuralNetOnGPU(mnist)
+  nn = NeuralNetOnGPU.remote(mnist)
 
   # Run a few steps of training and print the accuracy.
-  nn.train(100)
-  accuracy = ray.get(nn.get_accuracy())
+  nn.train.remote(100)
+  accuracy = ray.get(nn.get_accuracy.remote())
   print("Accuracy is {}.".format(accuracy))

doc/source/example-a3c.rst

@@ -73,7 +73,7 @@ We use a Ray Actor to simulate the environment.
   import numpy as np
   import ray
 
-  @ray.actor
+  @ray.remote
   class Runner(object):
     """Actor object to start running simulation on workers.
         Gradient computation is also executed on this object."""
@@ -127,7 +127,7 @@ global model parameters. The main training script looks like the following.
 
     # Start gradient calculation tasks on each actor
     parameters = policy.get_weights()
-    gradient_list = [agent.compute_gradient(parameters) for agent in agents]
+    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
@@ -147,6 +147,12 @@ global model parameters. The main training script looks like the following.
 Benchmarks and Visualization
 ----------------------------
 
-For the :code:`PongDeterministic-v3` 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.
+For the :code:`PongDeterministic-v3` 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:`./results/`. If you are running multiple experiments, be sure to vary the directory to which Tensorflow saves its progress (found in :code:`driver.py`).
+You can visualize performance by running
+:code:`tensorboard --logdir [directory]` in a separate screen, where
+:code:`[directory]` is defaulted to :code:`./results/`. If you are running
+multiple experiments, be sure to vary the directory to which Tensorflow saves
+its progress (found in :code:`driver.py`).

doc/source/example-resnet.rst

@@ -53,7 +53,7 @@ The core of the script is the actor definition.
 
 .. code-block:: python
 
-  @ray.actor(num_gpus=1)
+  @ray.remote(num_gpus=1)
   class ResNetTrainActor(object):
     def __init__(self, path, num_gpus):
       # Set the CUDA_VISIBLE_DEVICES environment variable in order to restrict
@@ -78,7 +78,7 @@ The main script first creates one actor for each GPU.
 
 .. code-block:: python
 
-  train_actors = [ResNetTrainActor(train_data, num_gpus) for _ in range(num_gpus)]
+  train_actors = [ResNetTrainActor.remote(train_data, num_gpus) for _ in range(num_gpus)]
 
 Then after initializing the actors with the same weights, the main loop performs
 updates on each model, averages the updates, and puts the new weights in the
@@ -87,7 +87,7 @@ object store.
 .. code-block:: python
 
   while True:
-    all_weights = ray.get([actor.compute_steps(weight_id) for actor in train_actors])
+    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)
 

doc/source/example-rl-pong.rst

@@ -58,7 +58,7 @@ the actor.
 
 .. code-block:: python
 
-  @ray.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
@@ -93,7 +93,7 @@ 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(model_id)
+    action_id = actors[i].compute_gradient.remote(model_id)
     actions.append(action_id)
 
 

doc/source/using-ray-with-tensorflow.rst

@@ -158,11 +158,11 @@ complex Python objects.
   x_test, y_test = ray.get(generate_fake_x_y_data.remote(BATCH_SIZE, seed=NUM_BATCHES))
 
   # Create actors to store the networks.
-  remote_network = ray.actor(Network)
-  actor_list = [remote_network(x_ids[i], y_ids[i]) for i in range(NUM_BATCHES)]
+  remote_network = ray.remote(Network)
+  actor_list = [remote_network.remote(x_ids[i], y_ids[i]) for i in range(NUM_BATCHES)]
 
   # Get initial weights of some actor.
-  weights = ray.get(actor_list[0].get_weights())
+  weights = ray.get(actor_list[0].get_weights.remote())
 
   # Do some steps of training.
   for iteration in range(NUM_ITERS):
@@ -173,7 +173,7 @@ complex Python objects.
     # more efficient.
     weights_id = ray.put(weights)
     # Call the remote function multiple times in parallel.
-    new_weights_ids = [actor.step(weights_id) for actor in actor_list]
+    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
@@ -288,8 +288,8 @@ For reference, the full code is below:
   x_test, y_test = ray.get(generate_fake_x_y_data.remote(BATCH_SIZE, seed=NUM_BATCHES))
 
   # Create actors to store the networks.
-  remote_network = ray.actor(Network)
-  actor_list = [remote_network(x_ids[i], y_ids[i]) for i in range(NUM_BATCHES)]
+  remote_network = ray.remote(Network)
+  actor_list = [remote_network.remote(x_ids[i], y_ids[i]) for i in range(NUM_BATCHES)]
   local_network = Network(x_test, y_test)
 
   # Get initial weights of local network.
@@ -304,7 +304,7 @@ For reference, the full code is below:
     # more efficient.
     weights_id = ray.put(weights)
     # Call the remote function multiple times in parallel.
-    gradients_ids = [actor.step(weights_id) for actor in actor_list]
+    gradients_ids = [actor.step.remote(weights_id) for actor in actor_list]
     # Get all of the weights.
     gradients_list = ray.get(gradients_ids)
 

examples/a3c/driver.py

@@ -15,7 +15,7 @@
 from misc import timestamp, time_string
 from envs import create_env
 
-@ray.actor
+@ray.remote
 class Runner(object):
     """Actor object to start running simulation on workers.
         Gradient computation is also executed from this object."""
@@ -58,9 +58,9 @@ def compute_gradient(self, params):
 def train(num_workers, env_name="PongDeterministic-v3"):
     env = create_env(env_name)
     policy = LSTMPolicy(env.observation_space.shape, env.action_space.n, 0)
-    agents = [Runner(env_name, i) for i in range(num_workers)]
+    agents = [Runner.remote(env_name, i) for i in range(num_workers)]
     parameters = policy.get_weights()
-    gradient_list = [agent.compute_gradient(parameters) for agent in agents]
+    gradient_list = [agent.compute_gradient.remote(parameters) for agent in agents]
     steps = 0
     obs = 0
     while True:
@@ -70,7 +70,7 @@ def train(num_workers, env_name="PongDeterministic-v3"):
         parameters = policy.get_weights()
         steps += 1
         obs += info["size"]
-        gradient_list.extend([agents[info["id"]].compute_gradient(parameters)])
+        gradient_list.extend([agents[info["id"]].compute_gradient.remote(parameters)])
     return policy
 
 if __name__ == '__main__':

examples/lbfgs/driver.py

@@ -60,7 +60,7 @@ def grad(self, xs, ys):
     """Computes the gradients of the network."""
     return self.sess.run(self.cross_entropy_grads, feed_dict={self.x: xs, self.y_: ys})
 
-@ray.actor
+@ray.remote
 class NetActor(object):
   def __init__(self, xs, ys):
     os.environ["CUDA_VISIBLE_DEVICES"] = ""
@@ -88,13 +88,13 @@ def get_flat_size(self):
 # Compute the loss on the entire dataset.
 def full_loss(theta):
   theta_id = ray.put(theta)
-  loss_ids = [actor.loss(theta_id) for actor in actors]
+  loss_ids = [actor.loss.remote(theta_id) for actor in actors]
   return sum(ray.get(loss_ids))
 
 # Compute the gradient of the loss on the entire dataset.
 def full_grad(theta):
   theta_id = ray.put(theta)
-  grad_ids = [actor.grad(theta_id) for actor in actors]
+  grad_ids = [actor.grad.remote(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.
 
 if __name__ == "__main__":
@@ -117,9 +117,9 @@ def full_grad(theta):
   batch_size = mnist.train.num_examples // num_batches
   batches = [mnist.train.next_batch(batch_size) for _ in range(num_batches)]
   print("Putting MNIST in the object store.")
-  actors = [NetActor(xs, ys) for (xs, ys) in batches]
+  actors = [NetActor.remote(xs, ys) for (xs, ys) in batches]
   # Initialize the weights for the network to the vector of all zeros.
-  dim = ray.get(actors[0].get_flat_size())
+  dim = ray.get(actors[0].get_flat_size.remote())
   theta_init = 1e-2 * np.random.normal(size=dim)
 
   # Use L-BFGS to minimize the loss function.

examples/policy_gradient/examples/example.py

@@ -47,15 +47,15 @@
     preprocessor = AtariPixelPreprocessor()
 
   print("Using the environment {}.".format(mdp_name))
-  agents = [RemoteAgent(mdp_name, 1, preprocessor, config, False) for _ in range(5)]
+  agents = [RemoteAgent.remote(mdp_name, 1, preprocessor, config, False) for _ in range(5)]
   agent = Agent(mdp_name, 1, preprocessor, config, True)
 
   kl_coeff = config["kl_coeff"]
 
   for j in range(1000):
     print("== iteration", j)
     weights = ray.put(agent.get_weights())
-    [a.load_weights(weights) for a in agents]
+    [a.load_weights.remote(weights) for a in agents]
     trajectory, total_reward, traj_len_mean = collect_samples(agents, config["timesteps_per_batch"], 0.995, 1.0, 2000)
     print("total reward is ", total_reward)
     print("trajectory length mean is ", traj_len_mean)

examples/policy_gradient/reinforce/agent.py

@@ -40,4 +40,4 @@ def compute_trajectory(self, gamma, lam, horizon):
     add_advantage_values(trajectory, gamma, lam, self.reward_filter)
     return trajectory
 
-RemoteAgent = ray.actor(Agent)
+RemoteAgent = ray.remote(Agent)

examples/policy_gradient/reinforce/rollout.py

@@ -79,7 +79,7 @@ def collect_samples(agents, num_timesteps, gamma, lam, horizon, observation_filt
   total_rewards = []
   traj_len_means = []
   while num_timesteps_so_far < num_timesteps:
-    trajectory_batch = ray.get([agent.compute_trajectory(gamma, lam, horizon) for agent in agents])
+    trajectory_batch = ray.get([agent.compute_trajectory.remote(gamma, lam, horizon) for agent in agents])
     trajectory = concatenate(trajectory_batch)
     total_rewards.append(trajectory["raw_rewards"].sum(axis=0).mean() / len(agents))
     trajectory = flatten(trajectory)

examples/resnet/resnet_main.py

@@ -44,7 +44,7 @@ def get_data(path, size, dataset):
           images[int(2 * size / 3):, :],
           labels)
 
-@ray.actor(num_gpus=use_gpu)
+@ray.remote(num_gpus=use_gpu)
 class ResNetTrainActor(object):
   def __init__(self, data, dataset, num_gpus):
     if num_gpus > 0:
@@ -89,7 +89,7 @@ def compute_steps(self, weights):
   def get_weights(self):
     return self.model.variables.get_weights()
 
-@ray.actor
+@ray.remote
 class ResNetTestActor(object):
   def __init__(self, data, dataset, eval_batch_count, eval_dir):
     hps = resnet_model.HParams(batch_size=100,
@@ -162,25 +162,25 @@ def train():
   train_data = get_data.remote(FLAGS.train_data_path, 50000, FLAGS.dataset)
   test_data = get_data.remote(FLAGS.eval_data_path, 10000, FLAGS.dataset)
   if num_gpus > 0:
-    train_actors = [ResNetTrainActor(train_data, FLAGS.dataset, num_gpus) for _ in range(num_gpus)]
+    train_actors = [ResNetTrainActor.remote(train_data, FLAGS.dataset, num_gpus) for _ in range(num_gpus)]
   else:
-    train_actors = [ResNetTrainActor(train_data, num_gpus)]
-  test_actor = ResNetTestActor(test_data, FLAGS.dataset, FLAGS.eval_batch_count, FLAGS.eval_dir)
-  print('The log files for tensorboard are stored at ip {}.'.format(ray.get(test_actor.get_ip_addr())))
+    train_actors = [ResNetTrainActor.remote(train_data, num_gpus, 0)]
+  test_actor = ResNetTestActor.remote(test_data, FLAGS.dataset, FLAGS.eval_batch_count, FLAGS.eval_dir)
+  print('The log files for tensorboard are stored at ip {}.'.format(ray.get(test_actor.get_ip_addr.remote())))
   step = 0
-  weight_id = train_actors[0].get_weights()
-  acc_id = test_actor.accuracy(weight_id, step)
+  weight_id = train_actors[0].get_weights.remote()
+  acc_id = test_actor.accuracy.remote(weight_id, step)
   if num_gpus == 0:
     num_gpus = 1
   print("Starting computation.")
   while True:
-    all_weights = ray.get([actor.compute_steps(weight_id) for actor in train_actors])
+    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)
     step += 10
     if step % 200 == 0:
       acc = ray.get(acc_id)
-      acc_id = test_actor.accuracy(weight_id, step)
+      acc_id = test_actor.accuracy.remote(weight_id, step)
       print('Step {0}: {1:.6f}'.format(step - 200, acc))
 
 def main(_):