Prototype named actors.

python/ray/experimental/__init__.py

@@ -7,10 +7,13 @@
     flush_redis_unsafe, flush_task_and_object_metadata_unsafe,
     flush_finished_tasks_unsafe, flush_evicted_objects_unsafe,
     _flush_finished_tasks_unsafe_shard, _flush_evicted_objects_unsafe_shard)
+from .named_actors import (
+    get_actor, register_actor
+)
 
 __all__ = [
     "TensorFlowVariables", "flush_redis_unsafe",
     "flush_task_and_object_metadata_unsafe", "flush_finished_tasks_unsafe",
     "flush_evicted_objects_unsafe", "_flush_finished_tasks_unsafe_shard",
-    "_flush_evicted_objects_unsafe_shard"
+    "_flush_evicted_objects_unsafe_shard", "get_actor", "register_actor"
 ]

python/ray/experimental/named_actors.py

@@ -0,0 +1,34 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import ray
+import ray.cloudpickle as pickle
+"""
+This file contains functions intended to implement the named actor
+"""
+
+def _calculate_key_(name):
+    return b"Actor:" + str.encode(name)
+
+def get_actor(name):
+    worker = ray.worker.get_global_worker()
+    actor_hash = _calculate_key_(name)
+    pickled_state = worker.redis_client.hmget(actor_hash, name)
+    assert len(pickled_state) == 1, \
+        "Error: Multiple actors under this name."
+    assert pickled_state[0] is not None, \
+        "Error: actor with name {} doesn't exist".format(name)
+    handle = pickle.loads(pickled_state[0])
+    return handle
+
+def register_actor(name, actor_handle):
+    worker = ray.worker.get_global_worker()
+    actor_hash = _calculate_key_(name)
+    assert type(actor_handle) == ray.actor.ActorHandle, \
+        "Error: you could only store named-actors."
+    is_existed = worker.redis_client.hexists(actor_hash, name)
+    assert not is_existed, \
+        "Error: the actor with name={} already exists".format(name)
+    pickled_state = pickle.dumps(actor_handle)
+    worker.redis_client.hmset(actor_hash, {name: pickled_state})

test/actor_test.py

@@ -1906,6 +1906,36 @@ def method(self):
         # we should also test this from a different driver.
         ray.get(new_f.method.remote())
 
+    def testRegisterAndGetActorHandle(self):
+        # TODO(heyucongtom): We may want to test this from another driver?
+        # One viable way might be setting up another ray process here, connecting to the
+        # redis_address, sending the objectID back, and compare
+
+        ray.worker.init(num_workers=1)
+
+        @ray.remote
+        class Foo(object):
+            def method(self):
+                pass
+
+        f1 = Foo.remote()
+        # Test saving f
+        ray.experimental.register_actor("f1", f1)
+        # Test getting f
+        f2 = ray.experimental.get_actor("f1")
+        self.assertEqual(f1._actor_id, f2._actor_id)
+
+        # Test same name register shall raise error
+        with self.assertRaises(AssertionError):
+            ray.experimental.register_actor("f1", f2)
+
+        # Test register with wrong object type
+        with self.assertRaises(AssertionError):
+            ray.experimental.register_actor("f3", 1)
+
+        # Test getting an unexist actor
+        with self.assertRaises(AssertionError):
+            err = ray.experimental.get_actor("unexisted")
 
 class ActorPlacementAndResources(unittest.TestCase):
     def tearDown(self):