Actor 实现

Mini Ray 的 Actor 用一个长期存活的线程和一个队列实现。

创建 Actor

用户代码：

counter = Counter.remote()

内部路径：

RemoteActorClass.remote()
  -> Runtime.create_actor()
    -> 创建 inbox Queue
    -> 创建 ActorHandle
    -> 启动 actor_loop 线程

ActorHandle 是 Driver 看到的对象。真正的 Counter 实例在 Actor 线程里。

Actor mailbox

Actor 方法调用不会直接执行，而是入队：

handle._inbox.put((method_name, args, kwargs, result_ref))

Actor 线程循环消费：

while True:
    item = inbox.get()
    if item is None:
        break
    method_name, method_args, method_kwargs, result_ref = item
    method = getattr(actor, method_name)
    store.set_result(result_ref, method(*args, **kwargs))

这就是 Actor 默认串行语义的来源。

方法调用也返回 ObjectRef

ref = counter.incr.remote()

每个方法调用都会先 ObjectStore.reserve()，再把 result_ref 放入 Actor 队列。方法执行完后，结果写入这个 ref。

Actor 初始化失败

如果 Actor 构造函数失败，Mini Ray 会把错误保存在 errors 列表里。后续方法调用会生成失败对象，ray.get 时抛出异常。

真实 Ray 的 Actor 生命周期更复杂，包括重启、命名 Actor、所有权、节点故障和方法重试。Mini Ray 只保留错误传播的最小语义。

资源占用

Mini Ray 在 Actor 线程启动时获取资源，线程结束时释放资源。这个模型表示“Actor 持有资源”。

@ray.remote(num_gpus=1)
class ModelServer:
    ...

模型服务 Actor 通常应该这样声明，因为 GPU 会被这个长期服务占用。

小练习

实现一个 Actor Pool：

class ActorPool:
    def __init__(self, actors):
        self.actors = actors
        self.index = 0

    def submit(self, method_name, *args):
        actor = self.actors[self.index % len(self.actors)]
        self.index += 1
        return getattr(actor, method_name).remote(*args)

然后创建多个 Counter 或 ModelWorker，观察请求如何分散。