Re-raise actor initialization errors on method invocation (#2843)

If an actor constructor fails, save that error and re-raise it on any subsequent attempts to interact with the actor. Related to https://github.com/ray-project/ray/issues/282 and https://github.com/ray-project/ray/issues/1093.

python/ray/tune/trainable.py

@@ -57,7 +57,6 @@ class Trainable(object):
                 object. If unspecified, a default logger is created.
         """
 
-        self._initialize_ok = False
         self._experiment_id = uuid.uuid4().hex
         self.config = config or {}
 
@@ -80,7 +79,6 @@ class Trainable(object):
         self._iterations_since_restore = 0
         self._restored = False
         self._setup()
-        self._initialize_ok = True
         self._local_ip = ray.services.get_node_ip_address()
 
     @classmethod
@@ -138,10 +136,6 @@ class Trainable(object):
             A dict that describes training progress.
         """
 
-        if not self._initialize_ok:
-            raise ValueError(
-                "Trainable initialization failed, see previous errors")
-
         start = time.time()
         result = self._train()
         result = result.copy()
@@ -282,9 +276,8 @@ class Trainable(object):
     def stop(self):
         """Releases all resources used by this trainable."""
 
-        if self._initialize_ok:
-            self._result_logger.close()
-            self._stop()
+        self._result_logger.close()
+        self._stop()
 
     def _train(self):
         """Subclasses should override this to implement train().

python/ray/worker.py

@@ -233,6 +233,7 @@ class Worker(object):
         self.cached_remote_functions_and_actors = []
         self.cached_functions_to_run = []
         self.fetch_and_register_actor = None
+        self.actor_init_error = None
         self.make_actor = None
         self.actors = {}
         self.actor_task_counter = 0
@@ -254,6 +255,17 @@ class Worker(object):
         # Identity of the driver that this worker is processing.
         self.task_driver_id = None
 
+    def mark_actor_init_failed(self, error):
+        """Called to mark this actor as failed during initialization."""
+
+        self.actor_init_error = error
+
+    def reraise_actor_init_error(self):
+        """Raises any previous actor initialization error."""
+
+        if self.actor_init_error is not None:
+            raise self.actor_init_error
+
     def get_serialization_context(self, driver_id):
         """Get the SerializationContext of the driver that this worker is processing.
 
@@ -907,6 +919,8 @@ class Worker(object):
 
         # Get task arguments from the object store.
         try:
+            if function_name != "__ray_terminate__":
+                self.reraise_actor_init_error()
             with profiling.profile("task:deserialize_arguments", worker=self):
                 arguments = self._get_arguments_for_execution(
                     function_name, args)
@@ -973,6 +987,9 @@ class Worker(object):
                 "function_id": function_id.id(),
                 "function_name": function_name
             })
+        # Mark the actor init as failed
+        if self.actor_id != NIL_ACTOR_ID and function_name == "__init__":
+            self.mark_actor_init_failed(error)
 
     def _become_actor(self, task):
         """Turn this worker into an actor.

test/actor_test.py

@@ -31,6 +31,24 @@ def shutdown_only():
     ray.shutdown()
 
 
+def test_actor_init_error_propagated(ray_start_regular):
+    @ray.remote
+    class Actor(object):
+        def __init__(self, error=False):
+            if error:
+                raise Exception("oops")
+
+        def foo(self):
+            return "OK"
+
+    actor = Actor.remote(error=False)
+    ray.get(actor.foo.remote())
+
+    actor = Actor.remote(error=True)
+    with pytest.raises(Exception, match=".*oops.*"):
+        ray.get(actor.foo.remote())
+
+
 def test_keyword_args(ray_start_regular):
     @ray.remote
     class Actor(object):

test/failure_test.py

@@ -213,9 +213,6 @@ def test_failed_actor_init(ray_start_regular):
         def __init__(self):
             raise Exception(error_message1)
 
-        def get_val(self):
-            return 1
-
         def fail_method(self):
             raise Exception(error_message2)
 
@@ -230,7 +227,27 @@ def test_failed_actor_init(ray_start_regular):
     a.fail_method.remote()
     wait_for_errors(ray_constants.TASK_PUSH_ERROR, 2)
     assert len(ray.error_info()) == 2
-    assert error_message2 in ray.error_info()[1]["message"]
+    assert error_message1 in ray.error_info()[1]["message"]
+
+
+def test_failed_actor_method(ray_start_regular):
+    error_message2 = "actor method failed"
+
+    @ray.remote
+    class FailedActor(object):
+        def __init__(self):
+            pass
+
+        def fail_method(self):
+            raise Exception(error_message2)
+
+    a = FailedActor.remote()
+
+    # Make sure that we get errors from a failed method.
+    a.fail_method.remote()
+    wait_for_errors(ray_constants.TASK_PUSH_ERROR, 1)
+    assert len(ray.error_info()) == 1
+    assert error_message2 in ray.error_info()[0]["message"]
 
 
 def test_incorrect_method_calls(ray_start_regular):