Ray: A Distributed Framework for Emerging AI Applications

Philipp Moritz, Robert Nishihara, Stephanie Wang, et al., arXiv:1712.05889v2, 2018

Ray 面向的不是单一的模型训练，而是强化学习（Reinforcement Learning，RL）中紧密耦合的仿真（simulation）、训练（training）和推理服务（serving）。这三类工作负载在执行时间、资源需求和状态管理方式上差异很大，并且会相互动态触发：

  • 仿真通常是大量、细粒度、无状态或状态可重建的 CPU 任务；
  • 训练通常是长时间、有状态、使用 GPU 的迭代计算；
  • 推理服务需要在环境交互的闭环中提供低延迟决策；
  • 仿真结果决定下一步要执行哪些任务，任务图不能在运行前完全确定。

传统 BSP 数据处理框架不擅长细粒度、异构和动态计算，深度学习框架又主要优化静态计算图内 的数值计算。Ray 的核心贡献，是在同一个动态执行引擎中统一 Task 和 Actor 两种抽象， 并通过分布式控制面支撑百万级任务吞吐。

一句话总结

Ray 把无状态计算表示为可重放的 Task，把有状态计算表示为串行执行方法的 Actor，并将二者 放进同一张动态任务图；系统再以 GCS 保存全局控制状态，以本地优先的两级调度降低调度开销， 以共享内存对象存储完成数据传递，从而同时获得灵活性、低延迟、水平扩展能力和基于 lineage 的容错。

问题背景：为什么已有框架不够用

一个典型 RL 训练循环先让当前策略与环境交互，生成若干 trajectory，再用这些数据更新策略：

当前策略
   │
   ├── Serving：根据状态产生动作
   │       │
   │       ▼
   ├── Simulation：环境执行动作，产生状态和奖励
   │       │
   │       ▼
   └── Training：汇总 trajectories，更新策略
           │
           └──────────► 下一版策略

这类应用给系统提出了四项关键要求：

  1. 细粒度与高吞吐：任务可能只运行几毫秒，大集群需要每秒调度数百万个任务；
  2. 异构性：任务耗时从毫秒到小时不等，资源可能是 CPU、GPU 或其他加速器；
  3. 同时支持无状态与有状态计算：仿真和数据处理适合无状态任务，模型副本、参数服务器及 第三方模拟器则需要保留状态；
  4. 动态执行：任务完成顺序不可预知，已有计算的结果还会决定后续要生成哪些任务。

MapReduce、Spark 一类 BSP 系统存在阶段屏障和较高的单任务开销；TensorFlow、MXNet 主要 解决计算图内部的分布式训练；CIEL、Dask 虽支持动态任务图，但论文当时认为它们缺少 Actor 和可水平扩展的控制面。将仿真、训练和服务分别放进多个专用系统，又会引入额外的数据搬运、 延迟和工程复杂度。

编程模型：Task 与 Actor 的统一

Ray 使用很小的一组异步 API 表达并行性和依赖关系：

API语义
f.remote(args)异步执行远程函数，立即返回 future
ray.get(futures)阻塞并取得 future 对应的对象
ray.wait(futures, k, timeout)在最先完成的 k 个结果就绪或超时后返回
Class.remote(args)创建远程 Actor，返回 Actor handle
actor.method.remote(args)异步调用 Actor 方法，返回 future

future 可以直接作为另一个远程调用的参数。因此，Driver 不必先 get 出结果再提交下游任务，运行时可以从 future 之间的关系直接得到数据依赖。远程函数还可以继续创建远程函数，即nested tasks；动态任务图由多个进程并行生成，而不是只能由单一 Driver 展开。

Task：无状态计算

Task 是无状态 Worker 对远程函数的一次执行。远程函数操作不可变对象，论文要求它无副作用，且输出只由输入决定，因此可以通过重新执行实现容错。

Task 的优势是：

  • 可以在任务粒度做细粒度负载均衡；
  • 可以把计算调度到输入数据所在节点，利用数据局部性；
  • 中间结果丢失后可根据 lineage 重算，不需要保存 Worker 状态。

它的代价是频繁的小更新需要序列化和传输外部状态。

Actor：有状态计算

Actor 是一个显式创建的有状态进程。对同一个 Actor 的方法调用按顺序执行，后一个方法能看到前一个方法留下的内部状态。Actor handle 可以传给其他 Task 或 Actor。

Actor 适合：

  • 参数服务器和模型副本；
  • 反复访问驻留在 GPU 上的数据；
  • 初始化成本高昂、内部状态难以序列化的第三方模拟器；
  • 需要对小块内部状态频繁更新的计算。

Actor 的负载均衡粒度较粗，创建后也难以像 Task 一样迁移到大对象所在节点；发生故障时还需要依赖 checkpoint 控制恢复成本。

二者并非竞争关系，而是同一应用中的互补工具：

维度TaskActor
状态无状态，输入输出不可变进程内保留可变状态
负载均衡细粒度粗粒度
数据局部性可逐任务选择节点Actor 固定后较弱
小更新序列化开销较高更新内部状态，开销低
故障恢复根据 lineage 重算checkpoint 后重放方法

计算模型：动态任务图

Ray 把 Task、Actor 方法调用及其结果统一表示在一张运行时持续增长的图中。图中有三类边：

  1. data edge：Task 与其输入、输出对象之间的数据依赖；
  2. control edge：父 Task 创建 nested task 时产生的控制依赖；
  3. stateful edge：同一 Actor 上相邻方法调用之间的状态依赖。

stateful edge 是统一两种抽象的关键。它把 Actor 隐含的可变状态变化显式纳入 lineage，使 Actor 方法也能使用任务图的依赖追踪和重放机制。换言之，Actor 不是一套独立运行时，而是被嵌入无状态任务图中的一条有序状态链。

系统架构

论文中的 Ray 分为应用层与系统层：

应用层： Driver        Stateless Workers        Stateful Actors
             │                 │                       │
             └─────────────────┴───────────────────────┘
                               │
系统层：              Per-node Local Scheduler
                               │ 仅在本地过载或资源不满足时上送
                               ▼
                         Global Scheduler(s)
                               │
              ┌────────────────┴────────────────┐
              ▼                                 ▼
     Global Control Store              Distributed Object Store
     lineage、对象位置、任务等           每节点共享内存、节点间复制

系统层由 GCS、分布式调度器和分布式对象存储三个核心组件组成。

Global Control Store：把状态与逻辑解耦

GCS 是带 pub-sub 能力的分片键值存储，保存系统完整的控制状态，包括：

  • Task 和 Actor 的 lineage；
  • 对象位置与大小；
  • 注册的远程函数；
  • 节点负载、事件日志等元数据。

论文实现中每个分片使用一个 Redis，并用 chain replication 提供容错。GCS 的关键不是简单地设置一个“中心数据库”，而是将控制状态的存储与使用这些状态作决策的组件解耦：

  • lineage 不再依赖单个 Driver 或 Scheduler 的内存；
  • Scheduler 不保存对象目录，数据传输不必经过 Scheduler；
  • Scheduler、对象存储等组件自身可以保持无状态，失败后从 GCS 重建；
  • 元数据存储和调度逻辑可以分别通过增加分片或副本水平扩展；
  • 调试、性能分析和可视化工具可以查询同一份全局状态。

这是论文最重要的系统设计原则：逻辑上集中控制状态，物理上分片和复制；让其余组件尽量无状态化。它同时改善扩展性、容错性和可观测性。

Bottom-up 两级调度

Ray 使用每节点一个 Local Scheduler，加上一个或多个 Global Scheduler。新任务先提交给创建它的节点：

  1. Local Scheduler 在节点资源足够且未过载时直接本地执行；
  2. 节点缺少所需资源（例如 GPU）或本地队列超过阈值时，才把任务上送 Global Scheduler；
  3. Global Scheduler 从满足资源约束的节点中选择预计完成时间最短者。

其估价近似为：

预计等待时间 = 本地排队时间 + 远端输入传输时间
             ≈ 队列长度 × 平均任务耗时
               + 远端输入总大小 ÷ 平均网络带宽

节点队列与资源信息通过 heartbeat 获得，对象位置和大小从 GCS 获取，任务耗时与带宽用指数平均估计。大部分任务走本地短路径，只有需要全局协调的任务才进入 Global Scheduler，因此这种设计被称为 bottom-up。Global Scheduler 也可增加共享 GCS 状态的副本来扩展吞吐。

该设计在集中式调度的全局决策能力与纯 work stealing 的低开销之间取了折中：保留异构资源与数据局部性决策，同时避免让全局调度器成为每个任务的必经路径。

分布式内存对象存储

Task 的输入输出都放在分布式对象存储中。每个节点的对象存储使用共享内存和 Apache Arrow 格式，同节点 Worker 可以零拷贝读取对象；输入不在本地时，执行前把它复制到本地对象存储。

这里有几个重要设计选择：

  • 对象不可变：无需处理更新冲突和复杂一致性协议，也便于 lineage 重算；
  • 计算只访问本地对象：远端输入先复制，执行阶段不依赖远程读；
  • 复制热点对象：避免单一数据源成为瓶颈；
  • 内存优先：内存不足时按 LRU 淘汰到磁盘；
  • 单对象不跨节点：一个对象必须能放入单节点内存，大矩阵等需由应用拆成多个 future。

对象目录存放在 GCS，而非调度器中，因此调度决策和数据传输彼此解耦。ray.get() 找不到本地对象时，可通过 GCS 的 pub-sub 等待对象位置出现，再从对应节点复制结果。

容错机制

Task 容错

GCS 持久保存 Task 的 lineage。当节点或对象丢失时，Ray 递归重新执行产生该对象的远程函数。这依赖无状态 Task 具有确定性、幂等性和无副作用；若用户在 Task 中直接修改外部系统，重放可能破坏这一假设。

Actor 容错

Actor 方法调用通过 stateful edge 记录在 lineage 中。Actor 故障后，Ray 从最近的用户定义 checkpoint 恢复状态，再按顺序重放后续方法。checkpoint 并非为了正确性所必需，但它决定了恢复时需要重放多少工作。

论文报告，在 2000 个 Actor 中故障恢复 400 个 Actor 时，checkpoint 将重放量从约 10000 次方法调用降到 500 次。相比 Task 的透明重算，Actor 仍把 checkpoint 策略的一部分责任留给应用开发者。

GCS 容错与空间控制

GCS 分片通过 chain replication 容忍节点故障。实验中，两副本链在成员退出、加入和状态传输期间，客户端观察到的最大延迟低于 30 ms。lineage 会随任务数持续增长，因此系统需要把旧记录定期 flush 到磁盘；论文也承认，对 lineage 进行安全垃圾回收仍是当时尚未完整解决的问题。

实验结论

论文的 benchmark 主要说明通用抽象没有牺牲关键路径性能：

实验论文结果说明
调度扩展性100 节点超过 180 万 Task/s；1 亿空 Task 用时 54 s接近线性扩展，但属于高度并行的空任务上限测试
数据局部性10–100 MB 输入下，无局部性感知的延迟高 1–2 个数量级Task 可以移动到数据所在节点
对象存储大对象写吞吐超过 15 GB/s；小对象约 18K IOPS单客户端、16 核实例上的结果
AllReduce16 节点、100 MB 约 200 ms，1 GB 约 1.2 s分别比当时 OpenMPI 快约 1.5× 和 2×
分布式训练ResNet-101 吞吐与 Horovod 相当，比 Distributed TF 低 10% 以内Ray 负责外层编排，数值计算仍由 TensorFlow 完成
内嵌 Serving大输入吞吐 6900 states/s，Clipper 为 290 states/s客户端与服务同机，发挥共享内存优势，并非通用 Web Serving 对比
仿真256 CPU 上 4.03M timesteps/s，BSP MPI 为 2.16M异步收集减少异构时长任务造成的屏障等待
Evolution Strategies扩展到 8192 核，中位求解时间 3.7 分钟nested tasks 与 Actor 聚合树避免 Driver 瓶颈
PPO所有规模均快于专用 MPI 实现，并减少 GPU 使用Task/Actor 级资源声明支持非对称 CPU/GPU 集群

其中 AllReduce 实验很能说明低延迟调度的重要性：人为给调度增加几毫秒延迟，迭代时间就接近翻倍。对于由许多短小、存在依赖关系的任务组成的通信原语，仅有高吞吐还不够，逐轮调度延迟同样会直接累积到端到端时间。

应用实验则体现了 Ray 的另一层价值：性能提升不只来自运行时更快，也来自编程模型允许开发者方便地改变应用结构。例如 ES 使用 Actor 构成分层聚合树，PPO 把 CPU 仿真和 GPU 训练放到非对称资源上；在固定架构的专用实现中，这类优化通常需要重写通信协议。

论文的边界与不足

  1. Ray 不是深度学习算子框架。它编排 TensorFlow 等系统完成训练，也不负责实现模拟器内部逻辑；
  2. Ray 不是 Spark 的替代品。论文版本缺少查询优化、straggler mitigation 和丰富的数据处理 API；
  3. Ray 的内嵌 Serving 不等于完整模型服务平台。模型版本管理、测试和模型组合仍是Clipper、TensorFlow Serving 一类系统的职责；
  4. 动态性限制了全局优化。Scheduler 无法预先看到完整计算图，需要依赖运行时统计做近似决策；
  5. lineage 有空间成本。长生命周期、超高任务量应用需要 flush 和垃圾回收；
  6. Actor 弱化了位置灵活性。状态驻留提升了小更新效率，也使数据局部性和故障恢复更复杂；
  7. 部分对比有特定前提。例如 Clipper 实验是同机内嵌调用，MPI 仿真对比使用带屏障的 BSP 实现，180 万 Task/s 则来自空任务，不能直接外推到所有真实负载。

此外，这篇笔记讨论的是 2018 年论文中的架构。论文中的 Redis 分片、chain replication 和具体组件边界不应直接等同于后续版本的 Ray 实现。

我的理解

这篇论文真正有影响力的地方，不是单独发明了 future、Actor、lineage 或两级调度，而是根据 AI 应用的组合型负载，把这些已有思想放进了一套一致的执行语义中：

  • Task 提供可移动、可重算的函数式计算；
  • Actor 提供状态驻留和低成本迭代；
  • future 与动态任务图负责组合二者；
  • 对象存储承担数据面；
  • GCS 承担可扩展、可查询的控制状态；
  • bottom-up 调度让常见路径保持本地化，只在必要时进行全局协调。

因此，Ray 的核心抽象并不局限于 RL。任何同时包含细粒度并行、长生命周期状态、动态控制流和异构资源的工作流，都可能从这种 Task + Actor 的组合中受益。论文留下的长期设计启示是：

面对复杂分布式应用，与其为每种子负载拼接一个专用系统，不如先寻找能够统一表达这些负载、同时仍保留关键性能特征的最小抽象集合。