From 611259b2c70262716771b9bacde97d1cb9cf0278 Mon Sep 17 00:00:00 2001 From: Eric Liang Date: Mon, 10 Sep 2018 10:51:19 -0700 Subject: [PATCH] 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 | 11 ++--------- python/ray/worker.py | 17 +++++++++++++++++ test/actor_test.py | 18 ++++++++++++++++++ test/failure_test.py | 25 +++++++++++++++++++++---- 4 files changed, 58 insertions(+), 13 deletions(-) diff --git a/python/ray/tune/trainable.py b/python/ray/tune/trainable.py index 898938d5e..bbdfb69ae 100644 --- a/python/ray/tune/trainable.py +++ b/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(). diff --git a/python/ray/worker.py b/python/ray/worker.py index 2ac5d931f..a076c02a6 100644 --- a/python/ray/worker.py +++ b/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. diff --git a/test/actor_test.py b/test/actor_test.py index 70ea474f4..49d51835c 100644 --- a/test/actor_test.py +++ b/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): diff --git a/test/failure_test.py b/test/failure_test.py index 6370c4633..149d4c74d 100644 --- a/test/failure_test.py +++ b/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):