Actor 容错#
Actor 可能会因 actor 进程死亡或 actor 的拥有者死亡而失败。Actor 的拥有者是最初通过调用 ActorClass.remote() 创建 actor 的 worker。 分离式 actor 没有拥有者进程，并在 Ray 集群销毁时进行清理。
Actor 进程故障#
Ray 可以自动重启意外崩溃的 actor。此行为由 max_restarts 控制，该参数设置 actor 将被重启的最大次数。 max_restarts 的默认值为 0，表示 actor 不会被重启。如果设置为 -1，actor 将被无限次重启。当 actor 被重启时，其状态将通过重新运行其构造函数来重建。在指定的重启次数之后，后续的 actor 方法将引发 RayActorError。
默认情况下，actor 任务以最多一次（at-most-once）语义执行（在 @ray.remote 装饰器 中 max_task_retries=0）。这意味着，如果 actor 任务被提交到一个不可达的 actor，Ray 将会通过 RayActorError 报告错误，这是一种在对任务返回的 future 调用 ray.get 时抛出的 Python 级别异常。请注意，即使任务确实成功执行，也可能抛出此异常。例如，如果 actor 在执行完任务后立即死亡，则可能会发生这种情况。
Ray 还为 actor 任务提供了至少一次（at-least-once）执行语义（max_task_retries=-1 或 max_task_retries > 0）。这意味着，如果 actor 任务被提交到一个不可达的 actor，系统将自动重试该任务。使用此选项，只有在发生以下任一情况时，系统才会向应用程序抛出 RayActorError：（1）actor 的 max_restarts 限制已超出，actor 无法再重启；或（2）该特定任务的 max_task_retries 限制已超出。请注意，如果 actor 在提交任务时正在重启，这将算作一次重试。可以通过 max_task_retries = -1 将重试限制设置为无限。
您可以通过运行以下代码来尝试此行为。
import os
import ray
ray.init()
# This actor kills itself after executing 10 tasks.
@ray.remote(max_restarts=4, max_task_retries=-1)
class Actor:
def __init__(self):
self.counter = 0
def increment_and_possibly_fail(self):
# Exit after every 10 tasks.
if self.counter == 10:
os._exit(0)
self.counter += 1
return self.counter
actor = Actor.remote()
# The actor will be reconstructed up to 4 times, so we can execute up to 50
# tasks successfully. The actor is reconstructed by rerunning its constructor.
# Methods that were executing when the actor died will be retried and will not
# raise a `RayActorError`. Retried methods may execute twice, once on the
# failed actor and a second time on the restarted actor.
for _ in range(50):
counter = ray.get(actor.increment_and_possibly_fail.remote())
print(counter) # Prints the sequence 1-10 5 times.
# After the actor has been restarted 4 times, all subsequent methods will
# raise a `RayActorError`.
for _ in range(10):
try:
counter = ray.get(actor.increment_and_possibly_fail.remote())
print(counter) # Unreachable.
except ray.exceptions.RayActorError:
print("FAILURE") # Prints 10 times.
对于至少一次（at-least-once）的 actor，系统仍然会根据初始提交顺序保证执行顺序。例如，任何提交给已失败 actor 任务的后续任务，在已失败的 actor 任务成功重试之前不会在 actor 上执行。系统不会尝试重新执行在故障前已成功执行的任何任务（除非 max_task_retries 非零且任务需要用于 对象重建）。
注意
对于 异步或线程 actor，任务可能会乱序执行。在 actor 重启后，系统只会重试未完成的任务。先前已完成的任务将不会被重新执行。
至少一次（at-least-once）执行最适合只读 actor 或状态短暂且在故障后不需要重建的 actor。对于具有关键状态的 actor，应用程序负责恢复状态，例如通过定期进行检查点并在 actor 重启时从检查点恢复。
Actor 检查点#
max_restarts 会自动重启崩溃的 actor，但不会自动恢复 actor 中的应用程序级别状态。相反，您应该手动检查点 actor 的状态并在 actor 重启时恢复。
对于手动重启的 actor，actor 的创建者应管理检查点，并在失败时手动重启和恢复 actor。如果您希望创建者决定何时重启 actor，以及/或创建者正在协调 actor 与其他执行的检查点，则推荐此做法。
import os
import sys
import ray
import json
import tempfile
import shutil
@ray.remote(num_cpus=1)
class Worker:
def __init__(self):
self.state = {"num_tasks_executed": 0}
def execute_task(self, crash=False):
if crash:
sys.exit(1)
# Execute the task
# ...
# Update the internal state
self.state["num_tasks_executed"] = self.state["num_tasks_executed"] + 1
def checkpoint(self):
return self.state
def restore(self, state):
self.state = state
class Controller:
def __init__(self):
self.worker = Worker.remote()
self.worker_state = ray.get(self.worker.checkpoint.remote())
def execute_task_with_fault_tolerance(self):
i = 0
while True:
i = i + 1
try:
ray.get(self.worker.execute_task.remote(crash=(i % 2 == 1)))
# Checkpoint the latest worker state
self.worker_state = ray.get(self.worker.checkpoint.remote())
return
except ray.exceptions.RayActorError:
print("Actor crashes, restarting...")
# Restart the actor and restore the state
self.worker = Worker.remote()
ray.get(self.worker.restore.remote(self.worker_state))
controller = Controller()
controller.execute_task_with_fault_tolerance()
controller.execute_task_with_fault_tolerance()
assert ray.get(controller.worker.checkpoint.remote())["num_tasks_executed"] == 2
或者，如果您正在使用 Ray 的自动 actor 重启，actor 可以手动检查点并在构造函数中从检查点恢复。
@ray.remote(max_restarts=-1, max_task_retries=-1)
class ImmortalActor:
def __init__(self, checkpoint_file):
self.checkpoint_file = checkpoint_file
if os.path.exists(self.checkpoint_file):
# Restore from a checkpoint
with open(self.checkpoint_file, "r") as f:
self.state = json.load(f)
else:
self.state = {}
def update(self, key, value):
import random
if random.randrange(10) < 5:
sys.exit(1)
self.state[key] = value
# Checkpoint the latest state
with open(self.checkpoint_file, "w") as f:
json.dump(self.state, f)
def get(self, key):
return self.state[key]
checkpoint_dir = tempfile.mkdtemp()
actor = ImmortalActor.remote(os.path.join(checkpoint_dir, "checkpoint.json"))
ray.get(actor.update.remote("1", 1))
ray.get(actor.update.remote("2", 2))
assert ray.get(actor.get.remote("1")) == 1
shutil.rmtree(checkpoint_dir)
注意
如果检查点已保存到外部存储，请确保整个集群都可以访问它，因为 actor 可以在不同的节点上重启。例如，将检查点保存到云存储（例如 S3）或共享目录（例如通过 NFS）。
Actor 创建者故障#
对于 非分离式 actor，actor 的拥有者是创建它的 worker，即调用 ActorClass.remote() 的 worker。与 对象 类似，如果 actor 的拥有者死亡，那么 actor 也将与拥有者共享命运。Ray 不会自动恢复拥有者已死的 actor，即使它有非零的 max_restarts。
由于 分离式 actor 没有拥有者，即使它们的原始创建者死亡，Ray 仍会重启它们。分离式 actor 将继续自动重启，直到超出最大重启次数、actor 被销毁，或 Ray 集群被销毁。
您可以在以下代码中尝试此行为。
import ray
import os
import signal
ray.init()
@ray.remote(max_restarts=-1)
class Actor:
def ping(self):
return "hello"
@ray.remote
class Parent:
def generate_actors(self):
self.child = Actor.remote()
self.detached_actor = Actor.options(name="actor", lifetime="detached").remote()
return self.child, self.detached_actor, os.getpid()
parent = Parent.remote()
actor, detached_actor, pid = ray.get(parent.generate_actors.remote())
os.kill(pid, signal.SIGKILL)
try:
print("actor.ping:", ray.get(actor.ping.remote()))
except ray.exceptions.RayActorError as e:
print("Failed to submit actor call", e)
# Failed to submit actor call The actor died unexpectedly before finishing this task.
# class_name: Actor
# actor_id: 56f541b178ff78470f79c3b601000000
# namespace: ea8b3596-7426-4aa8-98cc-9f77161c4d5f
# The actor is dead because because all references to the actor were removed.
try:
print("detached_actor.ping:", ray.get(detached_actor.ping.remote()))
except ray.exceptions.RayActorError as e:
print("Failed to submit detached actor call", e)
# detached_actor.ping: hello
强制终止行为不当的 actor#
有时应用程序级别的代码会导致 actor 挂起或泄漏资源。在这些情况下，Ray 允许您通过手动终止 actor 来从故障中恢复。您可以通过对 actor 的任何句柄调用 ray.kill 来实现此目的。请注意，它不需要是 actor 的原始句柄。
如果设置了 max_restarts，您还可以通过将 no_restart=False 传递给 ray.kill 来允许 Ray 自动重启 actor。
Actor 方法异常#
有时您希望在 actor 方法引发异常时进行重试。使用 max_task_retries 和 retry_exceptions 来启用此功能。
请注意，默认情况下，重试用户引发的异常是禁用的。要启用它，请确保该方法是幂等的，即多次调用它应该等同于只调用一次。
您可以在 @ray.method(retry_exceptions=...) 装饰器中设置 retry_exceptions，或在方法调用中使用 .options(retry_exceptions=...)。
重试行为取决于您为 retry_exceptions 设置的值
False（默认）：用户异常不重试。True：Ray 会在用户异常时最多重试该方法max_task_retries次。异常列表：Ray 会在用户异常时最多重试该方法
max_task_retries次，仅当该方法引发这些特定类的异常时。
max_task_retries 适用于异常和 actor 崩溃。Ray actor 可以设置此选项以应用于其所有方法。方法也可以为其自身设置一个覆盖选项。Ray 按以下顺序查找 max_task_retries 的第一个非默认值：
方法调用的值，例如
actor.method.options(max_task_retries=2)。如果您未设置此值，Ray 将忽略它。方法定义的 ist，例如
@ray.method(max_task_retries=2)。如果您未设置此值，Ray 将忽略它。actor 创建调用的 ist，例如
Actor.options(max_task_retries=2)。如果您未设置此值，Ray 将忽略它。Actor 类定义的 ist，例如
@ray.remote(max_task_retries=2)装饰器。如果您未设置此值，Ray 将忽略它。默认值，
0。
例如，如果一个方法设置了 max_task_retries=5 和 retry_exceptions=True，而 actor 设置了 max_restarts=2，Ray 将执行该方法最多 6 次：一次是初始调用，另外 5 次是重试。这 6 次调用可能包括 2 次 actor 崩溃。在第 6 次调用后，对结果 Ray ObjectRef 的 ray.get 调用将引发最后一次调用的异常，或者如果 actor 在最后一次调用中崩溃，则为 ray.exceptions.RayActorError。