演员

Actor 将 Ray API 从函数（任务）扩展到类。Actor 本质上是一个有状态的工作者（或服务）。当一个新的 Actor 被实例化时，会创建一个新的工作者，并且 Actor 的方法会被调度到该特定工作者上，并且可以访问和修改该工作者的状态。

Python

ray.remote 装饰器表示 Counter 类的实例将是演员。每个演员在其自己的 Python 进程中运行。

import ray

@ray.remote
class Counter:
    def __init__(self):
        self.value = 0

    def increment(self):
        self.value += 1
        return self.value

    def get_counter(self):
        return self.value

# Create an actor from this class.
counter = Counter.remote()

Java

Ray.actor 用于从常规的 Java 类创建 actor。

// A regular Java class.
public class Counter {

  private int value = 0;

  public int increment() {
    this.value += 1;
    return this.value;
  }
}

// Create an actor from this class.
// `Ray.actor` takes a factory method that can produce
// a `Counter` object. Here, we pass `Counter`'s constructor
// as the argument.
ActorHandle<Counter> counter = Ray.actor(Counter::new).remote();

C++

ray::Actor 用于从常规的 C++ 类创建演员。

// A regular C++ class.
class Counter {

private:
    int value = 0;

public:
  int Increment() {
    value += 1;
    return value;
  }
};

// Factory function of Counter class.
static Counter *CreateCounter() {
    return new Counter();
};

RAY_REMOTE(&Counter::Increment, CreateCounter);

// Create an actor from this class.
// `ray::Actor` takes a factory method that can produce
// a `Counter` object. Here, we pass `Counter`'s factory function
// as the argument.
auto counter = ray::Actor(CreateCounter).Remote();

使用 State API 中的 ray list actors 查看演员状态：

# This API is only available when you install Ray with `pip install "ray[default]"`.
ray list actors

======== List: 2023-05-25 10:10:50.095099 ========
Stats:
------------------------------
Total: 1

Table:
------------------------------
    ACTOR_ID                          CLASS_NAME    STATE      JOB_ID  NAME    NODE_ID                                                     PID  RAY_NAMESPACE
 0  9e783840250840f87328c9f201000000  Counter       ALIVE    01000000          13a475571662b784b4522847692893a823c78f1d3fd8fd32a2624923  38906  ef9de910-64fb-4575-8eb5-50573faa3ddf

指定所需资源

你也可以在角色中指定资源需求（更多详情请参见 资源需求 。）

Python

# Specify required resources for an actor.
@ray.remote(num_cpus=2, num_gpus=0.5)
class Actor:
    pass

Java

// Specify required resources for an actor.
Ray.actor(Counter::new).setResource("CPU", 2.0).setResource("GPU", 0.5).remote();

C++

// Specify required resources for an actor.
ray::Actor(CreateCounter).SetResource("CPU", 2.0).SetResource("GPU", 0.5).Remote();

调用演员

我们可以通过使用 remote 操作符调用其方法与actor进行交互。然后，我们可以在对象引用上调用 get 来检索实际值。

Python

# Call the actor.
obj_ref = counter.increment.remote()
print(ray.get(obj_ref))

1

Java

// Call the actor.
ObjectRef<Integer> objectRef = counter.task(&Counter::increment).remote();
Assert.assertTrue(objectRef.get() == 1);

C++

// Call the actor.
auto object_ref = counter.Task(&Counter::increment).Remote();
assert(*object_ref.Get() == 1);

在不同参与者上调用的方法可以并行执行，而在同一参与者上调用的方法则按调用顺序串行执行。同一参与者上的方法将共享状态，如下所示。

Python

# Create ten Counter actors.
counters = [Counter.remote() for _ in range(10)]

# Increment each Counter once and get the results. These tasks all happen in
# parallel.
results = ray.get([c.increment.remote() for c in counters])
print(results)

# Increment the first Counter five times. These tasks are executed serially
# and share state.
results = ray.get([counters[0].increment.remote() for _ in range(5)])
print(results)

[1, 1, 1, 1, 1, 1, 1, 1, 1, 1]
[2, 3, 4, 5, 6]

Java

// Create ten Counter actors.
List<ActorHandle<Counter>> counters = new ArrayList<>();
for (int i = 0; i < 10; i++) {
    counters.add(Ray.actor(Counter::new).remote());
}

// Increment each Counter once and get the results. These tasks all happen in
// parallel.
List<ObjectRef<Integer>> objectRefs = new ArrayList<>();
for (ActorHandle<Counter> counterActor : counters) {
    objectRefs.add(counterActor.task(Counter::increment).remote());
}
// prints [1, 1, 1, 1, 1, 1, 1, 1, 1, 1]
System.out.println(Ray.get(objectRefs));

// Increment the first Counter five times. These tasks are executed serially
// and share state.
objectRefs = new ArrayList<>();
for (int i = 0; i < 5; i++) {
    objectRefs.add(counters.get(0).task(Counter::increment).remote());
}
// prints [2, 3, 4, 5, 6]
System.out.println(Ray.get(objectRefs));

C++

// Create ten Counter actors.
std::vector<ray::ActorHandle<Counter>> counters;
for (int i = 0; i < 10; i++) {
    counters.emplace_back(ray::Actor(CreateCounter).Remote());
}

// Increment each Counter once and get the results. These tasks all happen in
// parallel.
std::vector<ray::ObjectRef<int>> object_refs;
for (ray::ActorHandle<Counter> counter_actor : counters) {
    object_refs.emplace_back(counter_actor.Task(&Counter::Increment).Remote());
}
// prints 1, 1, 1, 1, 1, 1, 1, 1, 1, 1
auto results = ray::Get(object_refs);
for (const auto &result : results) {
    std::cout << *result;
}

// Increment the first Counter five times. These tasks are executed serially
// and share state.
object_refs.clear();
for (int i = 0; i < 5; i++) {
    object_refs.emplace_back(counters[0].Task(&Counter::Increment).Remote());
}
// prints 2, 3, 4, 5, 6
results = ray::Get(object_refs);
for (const auto &result : results) {
    std::cout << *result;
}

传递演员句柄

角色句柄可以传递到其他任务中。我们可以定义使用角色句柄的远程函数（或角色方法）。

Python

import time

@ray.remote
def f(counter):
    for _ in range(10):
        time.sleep(0.1)
        counter.increment.remote()

Java

public static class MyRayApp {

  public static void foo(ActorHandle<Counter> counter) throws InterruptedException {
    for (int i = 0; i < 1000; i++) {
      TimeUnit.MILLISECONDS.sleep(100);
      counter.task(Counter::increment).remote();
    }
  }
}

C++

void Foo(ray::ActorHandle<Counter> counter) {
    for (int i = 0; i < 1000; i++) {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        counter.Task(&Counter::Increment).Remote();
    }
}

如果我们实例化一个角色，我们可以将句柄传递给各种任务。

Python

counter = Counter.remote()

# Start some tasks that use the actor.
[f.remote(counter) for _ in range(3)]

# Print the counter value.
for _ in range(10):
    time.sleep(0.1)
    print(ray.get(counter.get_counter.remote()))

0
3
8
10
15
18
20
25
30
30

Java

ActorHandle<Counter> counter = Ray.actor(Counter::new).remote();

// Start some tasks that use the actor.
for (int i = 0; i < 3; i++) {
  Ray.task(MyRayApp::foo, counter).remote();
}

// Print the counter value.
for (int i = 0; i < 10; i++) {
  TimeUnit.SECONDS.sleep(1);
  System.out.println(counter.task(Counter::getCounter).remote().get());
}

C++

auto counter = ray::Actor(CreateCounter).Remote();

// Start some tasks that use the actor.
for (int i = 0; i < 3; i++) {
  ray::Task(Foo).Remote(counter);
}

// Print the counter value.
for (int i = 0; i < 10; i++) {
  std::this_thread::sleep_for(std::chrono::seconds(1));
  std::cout << *counter.Task(&Counter::GetCounter).Remote().Get() << std::endl;
}

生成器

Ray 兼容 Python 生成器语法。更多详情请参见 Ray 生成器。

取消演员任务

通过调用 ray.cancel() 对返回的 ObjectRef 取消演员任务。

Python

import ray
import asyncio
import time


@ray.remote
class Actor:
    async def f(self):
        try:
            await asyncio.sleep(5)
        except asyncio.CancelledError:
            print("Actor task canceled.")


actor = Actor.remote()
ref = actor.f.remote()

# Wait until task is scheduled.
time.sleep(1)
ray.cancel(ref)

try:
    ray.get(ref)
except ray.exceptions.RayTaskError:
    print("Object reference was cancelled.")

在 Ray 中，任务取消行为取决于任务的当前状态：

未调度的任务：如果演员任务尚未被调度，Ray 尝试取消调度。在此阶段成功取消时，调用 ray.get(actor_task_ref) 会产生一个 TaskCancelledError。

运行Actor任务（常规Actor，线程化Actor）：对于分类为单线程Actor或多线程Actor的任务，Ray不提供中断机制。

运行异步Actor任务: 对于分类为 异步Actor <_async-actors> 的任务，Ray 试图取消相关的 asyncio.Task。这种取消方法符合 asyncio 任务取消 中提出的标准。请注意，如果你不在异步函数中 await，asyncio.Task 在执行过程中不会被中断。

取消保证：Ray 尝试以 尽力而为 的方式取消任务，这意味着取消并不总是得到保证。例如，如果取消请求未能传达给执行者，任务可能不会被取消。你可以使用 ray.get(actor_task_ref) 检查任务是否成功取消。

递归取消：Ray 跟踪所有子任务和 Actor 任务。当给出 recursive=True 参数时，它会取消所有子任务和 Actor 任务。

调度

对于每个actor，Ray会选择一个节点来运行它，调度决策基于一些因素，如 actor的资源需求 和 指定的调度策略。更多详情请参见 Ray调度。

容错性

默认情况下，Ray 角色不会被 重启 ，当角色意外崩溃时，角色任务不会被重试。您可以通过在 ray.remote() 和 .options() 中设置 max_restarts 和 max_task_retries 选项来改变这种行为。更多详情请参见 Ray 容错。

常见问题：演员、工人和资源

worker 和 actor 之间有什么区别？

每个“Ray worker”是一个Python进程。

工作者在任务和角色中受到不同的对待。任何“Ray 工作者”要么 1. 用于执行多个 Ray 任务，要么 2. 作为专用 Ray 角色启动。

• 任务：当 Ray 在一台机器上启动时，会自动启动多个 Ray 工作进程（默认情况下每个 CPU 一个）。它们将用于执行任务（类似于进程池）。如果你执行 8 个任务，每个任务使用 num_cpus=2，并且总 CPU 数为 16（ray.cluster_resources()["CPU"] == 16），你最终会有 8 个工作进程闲置。

• Actor: Ray Actor 也是一个 “Ray 工作者”，但在运行时实例化（通过 actor_cls.remote()）。它的所有方法都将在同一进程中运行，使用相同的资源（在定义 Actor 时指定）。请注意，与任务不同，运行 Ray Actor 的 Python 进程不会被重用，当 Actor 被删除时，这些进程将被终止。

为了最大限度地利用您的资源，您希望最大化工人的工作时间。您还希望分配足够的集群资源，以便所有需要的执行者都能运行，并且您定义的任何其他任务也能运行。这也意味着任务的调度更加灵活，并且如果您不需要执行者的有状态部分，您最好使用任务。

任务事件

默认情况下，Ray 会跟踪执行 actor 任务，报告任务状态事件和 Ray Dashboard 以及 State API 使用的分析事件。

你可以通过在 ray.remote() 和 .options() 中将 enable_task_events 选项设置为 False 来禁用角色的任务事件，这减少了任务执行的开销，并减少了发送到 Ray Dashboard 的数据量。

你也可以通过在 ray.remote() 和 .options() 中将 enable_task_events 选项设置为 False 来禁用某些参与者方法的任务事件。方法设置会覆盖参与者设置：

@ray.remote
class FooActor:

    # Disable task events reporting for this method.
    @ray.method(enable_task_events=False)
    def foo(self):
        pass


foo_actor = FooActor.remote()
ray.get(foo_actor.foo.remote())

更多关于 Ray Actors

• 命名角色
• 终止角色
• AsyncIO / 演员的并发性
• 使用并发组限制每方法的并发性
• 实用类
• 带外通信
• 演员任务执行顺序