Fault tolerance#
Ray is a distributed system, so failures can happen. Generally, Ray classifies failures into two classes:
Application-level failures, which bugs in user-level code or external system failures trigger.
System-level failures, which node failures, network failures, or bugs in Ray trigger.
The following sections describe the mechanisms Ray provides for applications to recover from failures.
To handle application-level failures, Ray provides mechanisms to catch errors, retry failed code, and handle misbehaving code. See the pages for task and actor fault tolerance for more information on these mechanisms.
Ray also provides several mechanisms to automatically recover from internal system-level failures, such as node failures. In particular, Ray can automatically recover from some failures in the distributed object store.
How to write fault-tolerant Ray applications#
Follow these recommendations to make your Ray applications fault tolerant.
First, if Ray’s fault-tolerance mechanisms don’t work for you, you can always catch the exceptions that failures cause and recover manually.
@ray.remote
class Actor:
def read_only(self):
import sys
import random
rand = random.random()
if rand < 0.2:
return 2 / 0
elif rand < 0.3:
sys.exit(1)
return 2
actor = Actor.remote()
# Manually retry the actor task.
while True:
try:
print(ray.get(actor.read_only.remote()))
break
except ZeroDivisionError:
pass
except ray.exceptions.RayActorError:
# Manually restart the actor
actor = Actor.remote()
Second, avoid letting an object ref outlive its owner task or actor. The owner is the task or actor that creates the initial object ref by calling ray.put() or foo.remote(). As long as references to an object still exist, the object’s owner worker keeps running, even after the corresponding task or actor finishes. If the owner worker fails, Ray can’t recover the object automatically for any caller that tries to access it. For example, returning an object ref that ray.put() created from a task creates an object that outlives its owner:
import ray
# Non-fault tolerant version:
@ray.remote
def a():
x_ref = ray.put(1)
return x_ref
x_ref = ray.get(a.remote())
# Object x outlives its owner task A.
try:
# If owner of x (i.e. the worker process running task A) dies,
# the application can no longer get value of x.
print(ray.get(x_ref))
except ray.exceptions.OwnerDiedError:
pass
In the preceding example, object x outlives its owner task a. If the worker process running task a fails, calling ray.get on x_ref afterward results in an OwnerDiedError exception.
The following fault-tolerant version returns x directly. In this example, the driver owns x and you only access it within the lifetime of the driver. If x is lost, Ray can automatically recover it through lineage reconstruction. See Anti-pattern: Returning ray.put() ObjectRefs from a task harms performance and fault tolerance for details.
# Fault tolerant version:
@ray.remote
def a():
# Here we return the value directly instead of calling ray.put() first.
return 1
# The owner of x is the driver
# so x is accessible and can be auto recovered
# during the entire lifetime of the driver.
x_ref = a.remote()
print(ray.get(x_ref))
Third, avoid custom resource requirements that only particular nodes can satisfy. If that node fails, Ray can’t retry the running tasks or actors on other nodes.
@ray.remote
def b():
return 1
# If the node with ip 127.0.0.3 fails while task b is running,
# Ray cannot retry the task on other nodes.
b.options(resources={"node:127.0.0.3": 1}).remote()
If you want a task to run on a particular node, use the NodeAffinitySchedulingStrategy. With this strategy, you can specify the affinity as a soft constraint, so even if the target node fails, Ray can still retry the task on other nodes.
# Prefer running on the particular node specified by node id
# but can also run on other nodes if the target node fails.
b.options(
scheduling_strategy=ray.util.scheduling_strategies.NodeAffinitySchedulingStrategy(
node_id=ray.get_runtime_context().get_node_id(), soft=True
)
).remote()