Fix bug in which actor classes are not exported multiple times. (ray-…

…project#4838)

python/ray/actor.py

@@ -186,8 +186,9 @@ class ActorClass(object):
             task.
         _resources: The default resources required by the actor creation task.
         _actor_method_cpus: The number of CPUs required by actor method tasks.
-        _exported: True if the actor class has been exported and false
-            otherwise.
+        _last_export_session: The index of the last session in which the remote
+            function was exported. This is used to determine if we need to
+            export the remote function again.
         _actor_methods: The actor methods.
         _method_decorators: Optional decorators that should be applied to the
             method invocation function before invoking the actor methods. These
@@ -208,7 +209,7 @@ def __init__(self, modified_class, class_id, max_reconstructions, num_cpus,
         self._num_cpus = num_cpus
         self._num_gpus = num_gpus
         self._resources = resources
-        self._exported = False
+        self._last_export_session = None
 
         self._actor_methods = inspect.getmembers(
             self._modified_class, ray.utils.is_function_or_method)
@@ -341,10 +342,14 @@ def _remote(self,
                 *copy.deepcopy(args), **copy.deepcopy(kwargs))
         else:
             # Export the actor.
-            if not self._exported:
+            if (self._last_export_session is None
+                    or self._last_export_session < worker._session_index):
+                # If this actor class was exported in a previous session, we
+                # need to export this function again, because current GCS
+                # doesn't have it.
+                self._last_export_session = worker._session_index
                 worker.function_actor_manager.export_actor_class(
                     self._modified_class, self._actor_method_names)
-                self._exported = True
 
             resources = ray.utils.resources_from_resource_arguments(
                 cpus_to_use, self._num_gpus, self._resources, num_cpus,

python/ray/remote_function.py

@@ -43,6 +43,9 @@ class RemoteFunction(object):
             return the resulting ObjectIDs. For an example, see
             "test_decorated_function" in "python/ray/tests/test_basic.py".
         _function_signature: The function signature.
+        _last_export_session: The index of the last session in which the remote
+            function was exported. This is used to determine if we need to
+            export the remote function again.
     """
 
     def __init__(self, function, num_cpus, num_gpus, resources,
@@ -68,7 +71,6 @@ def __init__(self, function, num_cpus, num_gpus, resources,
 
         # Export the function.
         worker = ray.worker.get_global_worker()
-        # In which session this function was exported last time.
         self._last_export_session = worker._session_index
         worker.function_actor_manager.export(self)
 

python/ray/tests/test_basic.py

@@ -2942,3 +2942,29 @@ def get_postprocessor(object_ids, values):
 
     assert ray.get(
         [ray.put(i) for i in [0, 1, 3, 5, -1, -3, 4]]) == [1, 3, 5, 4]
+
+
+def test_export_after_shutdown(ray_start_regular):
+    # This test checks that we can use actor and remote function definitions
+    # across multiple Ray sessions.
+
+    @ray.remote
+    def f():
+        pass
+
+    @ray.remote
+    class Actor(object):
+        def method(self):
+            pass
+
+    ray.get(f.remote())
+    a = Actor.remote()
+    ray.get(a.method.remote())
+
+    ray.shutdown()
+
+    # Start Ray and use the remote function and actor again.
+    ray.init(num_cpus=1)
+    ray.get(f.remote())
+    a = Actor.remote()
+    ray.get(a.method.remote())