diff --git a/doc/source/fault-tolerance.rst b/doc/source/fault-tolerance.rst index f27c5d596..b62becaf2 100644 --- a/doc/source/fault-tolerance.rst +++ b/doc/source/fault-tolerance.rst @@ -1,50 +1,90 @@ Fault Tolerance =============== -This document describes the handling of failures in Ray. +This document describes how Ray handles machine and process failures. -Machine and Process Failures ----------------------------- +Tasks +----- -Each **raylet** (the scheduler process) sends heartbeats to a **monitor** -process. If the monitor does not receive any heartbeats from a given raylet for -some period of time (about ten seconds), then it will mark that process as dead. +When a worker is executing a task, if the worker dies unexpectedly, either +because the process crashed or because the machine failed, Ray will rerun +the task (after a delay of several seconds) until either the task succeeds +or the maximum number of retries is exceeded. The default number of retries +is 4. -Lost Objects ------------- +You can experiment with this behavior by running the following code. + +.. code-block:: python + + import numpy as np + import os + import ray + import time + + ray.init(ignore_reinit_error=True) + + @ray.remote(max_retries=1) + def potentially_fail(failure_probability): + time.sleep(0.2) + if np.random.random() < failure_probability: + os._exit(0) + return 0 + + for _ in range(3): + try: + # If this task crashes, Ray will retry it up to one additional + # time. If either of the attempts succeeds, the call to ray.get + # below will return normally. Otherwise, it will raise an + # exception. + ray.get(potentially_fail.remote(0.5)) + print('SUCCESS') + except ray.exceptions.RayWorkerError: + print('FAILURE') -If an object is needed but is lost or was never created, then the task that -created the object will be re-executed to create the object. If necessary, tasks -needed to create the input arguments to the task being re-executed will also be -re-executed. This is the standard *lineage-based fault tolerance* strategy used -by other systems like Spark. Actors ------ -When an actor dies (either because the actor process crashed or because the node -that the actor was on died), by default any attempt to get an object from that -actor that cannot be created will raise an exception. Subsequent releases will -include an option for automatically restarting actors. +If an actor process crashes unexpectedly, Ray will attempt to reconstruct the +actor process up to a maximum number of times. This value can be specified with +the ``max_reconstructions`` keyword, which by default is ``0``. If the maximum +number of reconstructions has been used up, then subsequent actor methods will +raise exceptions. -Current Limitations -------------------- +When an actor is reconstructed, its state will be recreated by rerunning its +constructor. -At the moment, Ray does not handle all failure scenarios. We are working on -addressing these known problems. +You can experiment with this behavior by running the following code. -Process Failures -~~~~~~~~~~~~~~~~ +.. code-block:: python -1. Ray does not recover from the failure of any of the following processes: - any of the Redis servers and the monitor process. -2. If a driver fails, that driver will not be restarted and the job will not - complete. + import os + import ray + import time -Lost Objects -~~~~~~~~~~~~ + ray.init(ignore_reinit_error=True) -1. If an object is constructed by a call to ``ray.put`` on the driver, is then - evicted, and is later needed, Ray will not reconstruct this object. -2. If an object is constructed by an actor method, is then evicted, and is later - needed, Ray will not reconstruct this object. + @ray.remote(max_reconstructions=5) + class Actor: + def __init__(self): + self.counter = 0 + + def increment_and_possibly_fail(self): + self.counter += 1 + time.sleep(0.2) + if self.counter == 10: + os._exit(0) + return self.counter + + actor = Actor.remote() + + # The actor will be reconstructed up to 5 times. After that, methods will + # raise exceptions. The actor is reconstructed by rerunning its + # constructor. Methods that were executing when the actor died will also + # raise exceptions. + for _ in range(100): + try: + counter = ray.get(actor.increment_and_possibly_fail.remote()) + print(counter) + except ray.exceptions.RayActorError: + print('FAILURE') diff --git a/java/api/src/main/java/org/ray/api/options/ActorCreationOptions.java b/java/api/src/main/java/org/ray/api/options/ActorCreationOptions.java index cc8a126eb..de50c0895 100644 --- a/java/api/src/main/java/org/ray/api/options/ActorCreationOptions.java +++ b/java/api/src/main/java/org/ray/api/options/ActorCreationOptions.java @@ -9,7 +9,7 @@ import java.util.Map; public class ActorCreationOptions extends BaseTaskOptions { public static final int NO_RECONSTRUCTION = 0; - public static final int INFINITE_RECONSTRUCTIONS = (int) Math.pow(2, 30); + public static final int INFINITE_RECONSTRUCTION = (int) Math.pow(2, 30); // DO NOT set this environment variable. It's only used for test purposes. // Please use `setUseDirectCall` instead. public static final boolean DEFAULT_USE_DIRECT_CALL = "1" diff --git a/python/ray/worker.py b/python/ray/worker.py index 07c338127..38e9d4745 100644 --- a/python/ray/worker.py +++ b/python/ray/worker.py @@ -1727,7 +1727,12 @@ def remote(*args, **kwargs): number of times that the actor should be reconstructed when it dies unexpectedly. The minimum valid value is 0 (default), which indicates that the actor doesn't need to be reconstructed. And the maximum valid - value is ray.ray_constants.INFINITE_RECONSTRUCTIONS. + value is ray.ray_constants.INFINITE_RECONSTRUCTION. + * **max_retries**: Only for *remote functions*. This specifies the maximum + number of times that the remote function should be rerun when the worker + process executing it crashes unexpectedly. The minimum valid value is 0, + the default is 4 (default), and the maximum valid value is + ray.ray_constants.INFINITE_RECONSTRUCTION. This can be done as follows: