Allow returning an actor handle from a remote call (#13476)

python/ray/_raylet.pyx

@@ -1525,27 +1525,23 @@ cdef class CoreWorker:
         serialized_objects = []
         for i in range(len(outputs)):
             return_id, output = return_ids[i], outputs[i]
-            if isinstance(output, ray.actor.ActorHandle):
-                raise Exception("Returning an actor handle from a remote "
-                                "function is not allowed).")
-            else:
-                context = worker.get_serialization_context()
-                serialized_object = context.serialize(output)
-                data_sizes.push_back(serialized_object.total_bytes)
-                metadata = serialized_object.metadata
-                if ray.worker.global_worker.debugger_get_breakpoint:
-                    breakpoint = (
-                        ray.worker.global_worker.debugger_get_breakpoint)
-                    metadata += (
-                        b"," + ray_constants.OBJECT_METADATA_DEBUG_PREFIX +
-                        breakpoint.encode())
-                    # Reset debugging context of this worker.
-                    ray.worker.global_worker.debugger_get_breakpoint = b""
-                metadatas.push_back(string_to_buffer(metadata))
-                serialized_objects.append(serialized_object)
-                contained_ids.push_back(
-                    ObjectRefsToVector(serialized_object.contained_object_refs)
-                )
+            context = worker.get_serialization_context()
+            serialized_object = context.serialize(output)
+            data_sizes.push_back(serialized_object.total_bytes)
+            metadata = serialized_object.metadata
+            if ray.worker.global_worker.debugger_get_breakpoint:
+                breakpoint = (
+                    ray.worker.global_worker.debugger_get_breakpoint)
+                metadata += (
+                    b"," + ray_constants.OBJECT_METADATA_DEBUG_PREFIX +
+                    breakpoint.encode())
+                # Reset debugging context of this worker.
+                ray.worker.global_worker.debugger_get_breakpoint = b""
+            metadatas.push_back(string_to_buffer(metadata))
+            serialized_objects.append(serialized_object)
+            contained_ids.push_back(
+                ObjectRefsToVector(serialized_object.contained_object_refs)
+            )
 
         with nogil:
             check_status(CCoreWorkerProcess.GetCoreWorker()

python/ray/tests/test_actor.py

@@ -955,5 +955,24 @@ def test_atexit_handler(ray_start_regular_shared, exit_condition):
         wait_for_condition(check_file_written)
 
 
+def test_return_actor_handle_from_actor(ray_start_regular_shared):
+    @ray.remote
+    class Inner:
+        def ping(self):
+            return "pong"
+
+    @ray.remote
+    class Outer:
+        def __init__(self):
+            self.inner = Inner.remote()
+
+        def get_ref(self):
+            return self.inner
+
+    outer = Outer.remote()
+    inner = ray.get(outer.get_ref.remote())
+    assert ray.get(inner.ping.remote()) == "pong"
+
+
 if __name__ == "__main__":
     sys.exit(pytest.main(["-v", __file__]))

python/ray/tests/test_actor_advanced.py

@@ -597,10 +597,6 @@ def test_calling_put_on_actor_handle(ray_start_regular):
     def f():
         return Counter.remote()
 
-    @ray.remote
-    def g():
-        return [Counter.remote()]
-
     # Currently, calling ray.put on an actor handle is allowed, but is
     # there a good use case?
     counter = Counter.remote()
@@ -610,11 +606,7 @@ def test_calling_put_on_actor_handle(ray_start_regular):
     assert ray.get(counter.inc.remote()) == 2
     assert ray.get(new_counter.inc.remote()) == 3
 
-    with pytest.raises(Exception):
-        ray.get(f.remote())
-
-    # The below test works, but do we want to disallow this usage?
-    ray.get(g.remote())
+    ray.get(f.remote())
 
 
 def test_named_but_not_detached(ray_start_regular):