Prohibit returning actor handles and also update actor documentation. (#1246)

* Prohibit returning actor handles and also update actor documentation.

* Clarify documentation.

doc/source/actors.rst

@@ -256,3 +256,68 @@ We can put this all together as follows.
   nn.train.remote(100)
   accuracy = ray.get(nn.get_accuracy.remote())
   print("Accuracy is {}.".format(accuracy))
+
+Passing Around Actor Handles (Experimental)
+-------------------------------------------
+
+Actor handles can be passed into other tasks. To see an example of this, take a
+look at the `asynchronous parameter server example`_. To illustrate this with
+a simple example, consider a simple actor definition. This functionality is
+currently **experimental** and subject to the limitations described below.
+
+.. code-block:: python
+
+  @ray.remote
+  class Counter(object):
+      def __init__(self):
+          self.counter = 0
+
+      def inc(self):
+          self.counter += 1
+
+      def get_counter(self):
+          return self.counter
+
+We can define remote functions (or actor methods) that use actor handles.
+
+.. code-block:: python
+
+  @ray.remote
+  def f(counter):
+      while True:
+          counter.inc.remote()
+
+If we instantiate an actor, we can pass the handle around to various tasks.
+
+.. code-block:: python
+
+  counter = Counter.remote()
+
+  # Start some tasks that use the actor.
+  [f.remote(counter) for _ in range(4)]
+
+  # Print the counter value.
+  for _ in range(10):
+      print(ray.get(counter.get_counter.remote()))
+
+Current Actor Limitations
+-------------------------
+
+We are working to address the following issues.
+
+1. **Actor lifetime management:** Currently, when the original actor handle for
+   an actor goes out of scope, a task is scheduled on that actor that kills the
+   actor process (this new task will run once all previous tasks have finished
+   running). This could be an issue if the original actor handle goes out of
+   scope, but the actor is still being used by tasks that have been passed the
+   actor handle.
+2. **Returning actor handles:** Actor handles currently cannot be returned from
+   a remote function or actor method. Similarly, ``ray.put`` cannot be called on
+   an actor handle.
+3. **Reconstruction evicted actor objects:** If ``ray.get`` is called on an
+   evicted object that was created by an actor method, Ray currently will not
+   reconstruct the object. For more information, see the documentation on
+   `fault tolerance`_.
+
+.. _`asynchronous parameter server example`: http://ray.readthedocs.io/en/latest/example-parameter-server.html
+.. _`fault tolerance`: http://ray.readthedocs.io/en/latest/fault-tolerance.html

python/ray/worker.py

@@ -347,6 +347,11 @@ class Worker(object):
                             "do this, you can wrap the ObjectID in a list and "
                             "call 'put' on it (or return it).")
 
+        if isinstance(value, ray.actor.ActorHandleParent):
+            raise Exception("Calling 'put' on an actor handle is currently "
+                            "not allowed (similarly, returning an actor "
+                            "handle from a remote function is not allowed).")
+
         # Serialize and put the object in the object store.
         try:
             self.store_and_register(object_id, value)

test/actor_test.py

@@ -1563,6 +1563,34 @@ class DistributedActorHandles(unittest.TestCase):
         # through both the original handle and the forked handle.
         self.assertEqual(ray.get(actor.inc.remote()), y + 1)
 
+    def testCallingPutOnActorHandle(self):
+        ray.worker.init(num_workers=1)
+
+        @ray.remote
+        class Counter(object):
+            pass
+
+        @ray.remote
+        def f():
+            return Counter.remote()
+
+        @ray.remote
+        def g():
+            return [Counter.remote()]
+
+        with self.assertRaises(Exception):
+            ray.put(Counter.remote())
+
+        with self.assertRaises(Exception):
+            ray.get(f.remote())
+
+        # The below test is commented out because it currently does not behave
+        # properly. The call to g.remote() does not raise an exception because
+        # even though the actor handle cannot be pickled, pyarrow attempts to
+        # serialize it as a dictionary of its fields which kind of works.
+        # self.assertRaises(Exception):
+        #     ray.get(g.remote())
+
 
 if __name__ == "__main__":
     unittest.main(verbosity=2)