1. 论文基本信息

1.1. 标题

Ray: A Distributed Framework for Emerging AI Applications (Ray：面向新兴人工智能应用的分布式框架)

1.2. 作者

Philipp Moritz, Robert Nishihara, Stephanie Wang, Alexey Tumanov, Richard Liaw, Eric Liang, Melih Elibol, Zongheng Yang, William Paul, Michael I. Jordan, Ion Stoica

• 背景分析： 作者团队主要来自加州大学伯克利分校（UC Berkeley）的 RISELab。这是一个在分布式系统领域极具影响力的实验室，之前孵化了著名的 Apache Spark。Ion Stoica 和 Michael I. Jordan 都是计算机科学和人工智能领域的泰斗级人物。

1.3. 发表期刊/会议

虽然原文元数据仅提供了 ArXiv 发布时间，但这篇论文正式发表于 USENIX OSDI 2018 (13th USENIX Symposium on Operating Systems Design and Implementation)。

• 影响力： OSDI 是计算机系统领域最顶级的会议之一，以录用率低、影响力大著称。Google File System (GFS)、MapReduce、BigTable 等划时代的技术均首发于此。

1.4. 发表年份

2018年 (ArXiv 版本发布于 2017 年 12 月)

1.5. 摘要

论文指出下一代人工智能（AI）应用（特别是强化学习）将不断与环境交互并从中学习。这些应用对系统提出了极高的性能（Performance）和灵活性（Flexibility）要求。为此，作者提出了 Ray——一种分布式系统。Ray 实现了一个统一的接口，能够通过单一的动态执行引擎同时支持任务并行（Task-parallel）和基于演员（Actor-based）的计算。为了满足性能需求，Ray 采用了分布式调度器和分布式容错存储（即 GCS）来管理系统的控制状态。实验表明，Ray 能够扩展到每秒处理超过 180 万个任务，并在多个具有挑战性的强化学习应用中表现优于现有的专用系统。

1.6. 原文链接

• arXiv 链接: https://arxiv.org/abs/1712.05889

• PDF 链接: https://arxiv.org/pdf/1712.05889v2.pdf

  ---

2. 整体概括

2.1. 研究背景与动机

• 核心问题： 传统的分布式计算框架（如 Spark, MapReduce）擅长处理大规模静态数据的批处理，深度学习框架（如 TensorFlow, PyTorch）擅长加速神经网络的训练。然而，新兴的强化学习（RL）应用需要在一个紧密的闭环中同时进行模拟（Simulation）、训练（Training）和服务（Serving）。
• 现有挑战（Gap）：

  1. 细粒度与异构性： RL 任务的持续时间差异巨大（从几毫秒的模拟步到数小时的训练），且资源需求不同（CPU 用于模拟，GPU 用于训练）。

  2. 动态执行： 计算图不是预先定义的，而是根据模拟结果动态生成的（例如，模拟结果决定下一步动作）。

  3. 现有方案的不足： 将多个专用系统（如用 Horovod 做训练，Clipper 做服务，MPI 做模拟）拼接在一起，会导致系统极其复杂，且数据传输延迟过高，无法满足 RL 对毫秒级延迟的要求。

     下图（原文 Figure 1）展示了一个典型的 RL 系统闭环，包括环境交互、策略评估和策略改进，这种紧密耦合的流程正是 Ray 试图优化的目标。

     该图像是一个示意图，展示了强化学习（RL）系统的结构。左侧为智能体（Agent），包括训练和服务两部分，培训中使用如SGD的策略改进方法；右侧为环境（Environment），呈现动作、状态和奖励的交互机制。轨迹表示为 $s_0, (s_1, t_1), \, \ldots, (s_n, t_n)$。

2.2. 核心贡献

1. 统一的编程模型： Ray 首次在一个框架内统一了任务（Task，无状态）和演员（Actor，有状态）两种抽象，使得开发者可以灵活地构建复杂的 AI 应用。

2. 创新的系统架构：

   • 全局控制存储（Global Control Store, GCS）： 提出将系统的控制状态（元数据、任务谱系等）从调度器中剥离，存入一个分片的、容错的存储系统中。这使得系统的其他组件（如调度器）可以变成无状态的，从而极大地提升了水平扩展能力和容错性。
   • 自底向上的分布式调度（Bottom-Up Distributed Scheduling）： 为了实现毫秒级低延迟，任务首先尝试在本地调度，只有在资源不足时才转发给全局调度器。

3. 高性能表现： 实现了线性扩展能力，在 100 个节点上达到每秒 180 万个任务的吞吐量，且延迟极低。

   ---

3. 预备知识与相关工作

3.1. 基础概念

为了理解 Ray，初学者需要掌握以下概念：

• 强化学习 (Reinforcement Learning, RL): 机器学习的一个分支，智能体 (Agent) 通过与 环境 (Environment) 交互（执行动作 Action，观察状态 State，获得奖励 Reward）来学习策略 (Policy)。
  • 推演/模拟 (Rollout): 在 RL 中，指使用当前策略在仿真环境中运行一段时间，生成一系列（状态，动作，奖励）数据的过程。这是 RL 中最耗时的步骤之一。
• 无状态 (Stateless) vs. 有状态 (Stateful):
  • 无状态 (Task): 就像一个数学函数 y = f(x)，只要输入 $x$ 相同，输出 $y$ 永远相同，且不依赖于之前的计算历史。适合大规模并行处理。
  • 有状态 (Actor): 就像一个 Python 对象（Class Instance），内部保存了变量（如神经网络的权重）。调用它的方法会改变其内部状态。适合模拟环境或参数服务器。
• 谱系 (Lineage): 在分布式系统中，指一个数据对象是如何产生的“家谱”。如果某个数据丢失了，系统可以通过查看谱系，重新执行生成该数据的任务来恢复它。
• 幂等性 (Idempotence): 指一个操作无论执行多少次，结果都是一样的。这对于系统故障后的自动重试非常重要。

3.2. 前人工作与差异

• 数据流系统 (Dataflow Systems): 如 MapReduce, Spark, Dryad。
  • 特点： 采用批量同步并行（BSP）模型。
  • 局限： 难以支持细粒度、动态生成的任务；缺乏对“有状态计算”（Actor）的原生支持。Ray 比它们更灵活，延迟更低。
• 深度学习框架 (DL Frameworks): 如 TensorFlow, MXNet。
  • 特点： 针对静态计算图和 GPU 进行了极致优化。
  • 局限： 主要关注训练环节，难以高效处理 RL 中的大规模并发模拟和服务任务。Ray 可以与这些框架无缝集成，负责分布式调度。
• 演员模型系统 (Actor Systems): 如 Erlang, Akka, Orleans。

  • 特点： 高并发，支持有状态计算。

  • 局限： 通常缺乏自动的数据恢复机制（Lineage-based fault tolerance）。Ray 结合了 Actor 的灵活性和数据流系统的容错性。

    ---

4. 方法论

4.1. 方法原理：动态任务图

Ray 的核心计算模型是动态任务图 (Dynamic Task Graph)。与 TensorFlow（早期版本）的静态图不同，Ray 的计算图是在程序运行时动态构建的。

• 节点 (Nodes): 图中的节点代表任务 (Task)（远程函数调用）或对象 (Data Object)。
• 边 (Edges):

  • 数据边 (Data Edge): 表示数据的依赖关系（任务 A 的输出是任务 B 的输入）。

  • 控制边 (Control Edge): 表示嵌套调用关系（任务 A 内部调用了任务 B）。

  • 状态边 (Stateful Edge): 这是 Ray 的创新。为了支持 Actor，Ray 在同一个 Actor 的连续方法调用之间添加状态边，确保它们按顺序执行，并共享内部状态。

    下图（原文 Figure 2）展示了如何用 Python 代码定义 Ray 的 Task 和 Actor；下图（原文 Figure 4）展示了其对应的任务图结构，其中可以看到 Actor 的方法调用之间存在状态依赖（Stateful Edge）。

    该图像是图表，展示了调用 train_policy.remote() 后的任务图。图中包括多个角色和方法的调用，表示任务之间的数据依赖和状态共享关系。其中，控制边和状态边分别用不同的线条表示，反映了任务执行的逻辑结构。

4.2. API 设计 (应用层)

Ray 提供了极其简洁的 Python API，主要包含两个核心抽象：

1. Tasks (任务): 使用 @ray.remote 装饰的函数。调用时返回一个 Future（未来对象），是非阻塞的。
   • 代码示例：z_future = add.remote(x, y)。
2. Actors (演员): 使用 @ray.remote 装饰的类。实例化后是一个远程进程，维护自己的状态。

   • 代码示例：counter = Counter.remote(); c_future = counter.increment.remote()。

     下表（原文 Table 2）清晰地总结了 Task 和 Actor 的权衡：

Table 2: Tasks vs. actors tradeoffs.

4.3. 系统架构 (系统层)

Ray 的系统架构设计是为了同时满足低延迟和高吞吐。下图（原文 Figure 5）展示了其整体架构。

该图像是图示，展示了Ray的体系结构，包括应用层和系统层。应用层实现API和计算模型，系统层负责任务调度和数据管理，以满足性能和容错要求。

4.3.1. 全局控制存储 (GCS - Global Control Store)

这是 Ray 最具创新性的设计。

• 传统做法： 调度器（Scheduler）通常在其内存中存储任务队列、对象位置等元数据。这使得调度器成为单点故障，且难以扩展。
• Ray 的做法： 将所有控制状态（任务表、对象表、函数表、事件日志）推送到一个中心化的存储系统 GCS 中（底层使用 Redis 实现分片存储）。
• 优势：
  1. 组件无状态化： 调度器和工作节点（Worker）都变成了无状态的，这意味着它们可以随意崩溃并重启，只需从 GCS 读取状态即可恢复。
  2. 独立扩展： 可以独立地增加 GCS 的分片来提升元数据吞吐量。

4.3.2. 自底向上的分布式调度器 (Bottom-Up Distributed Scheduler)

Ray 需要每秒调度数百万个任务，中心化调度器（如 Spark Driver）是瓶颈。Ray 设计了两级调度结构：

1. 本地调度器 (Local Scheduler): 每个节点都有一个。当任务在某节点生成时，首先提交给本地调度器。如果本地资源（如 CPU/GPU）充足，直接在本地执行。这保证了极低的调度延迟。

2. 全局调度器 (Global Scheduler): 如果本地节点过载，任务会被转发给全局调度器。全局调度器根据 GCS 中的负载信息，将任务分配给其他合适的节点。

   下图（原文 Figure 6）直观地展示了这种自底向上的流程。粗箭头表示大部分任务都在本地消化了。

   该图像是一个图表，展示了底向上分布式调度器的结构。任务从驱动程序和工作节点提交到本地调度器，必要时再转发至全局调度器。箭头的粗细与请求速率成正比。

4.3.3. 内存分布式对象存储 (In-Memory Distributed Object Store)

• 实现： 每个节点有一个共享内存对象存储（使用 Apache Arrow 格式）。
• 零拷贝 (Zero-Copy): 同一个节点上的多个任务（进程）可以读取同一个对象，而无需在内存中复制数据。这对于传输大的 NumPy 数组或神经网络权重至关重要。
• 数据传输： 当任务需要的对象不在本地时，对象存储负责在节点间自动复制数据。

4.3.4. 端到端执行流程

让我们通过一个例子 add(a, b) 来看 Ray 是如何工作的（结合原文 Figure 7）：

1. 提交： 驱动程序调用 add.remote(a, b)。

2. 调度： 本地调度器检查参数 $a$ 和 $b$ 的位置。如果决定在远程节点执行，任务被发送过去。

3. 获取依赖： 远程节点的对象存储检查本地是否有 $a$ 和 $b$。如果没有，它会查询 GCS 找到它们的位置，并将其拉取（Replicate）到本地。

4. 执行： 一旦数据就绪，Worker 执行函数，结果存入本地对象存储。

5. 返回： 结果的元数据写入 GCS，通知调用方。

   该图像是示意图，展示了一个远程任务执行和结果返回的过程。图中分为两部分：(a) 在节点 N1 上注册并调用函数 add()，然后在节点 N2 上执行该函数；(b) 节点 N1 通过 c = \text{ray.get}(id_c) 获取 add() 的结果。实线表示数据平面操作，虚线表示控制平面操作。此过程涉及全局控制存储（GCS）和对象存储的交互。

---

5. 实验设置

5.1. 实验环境

• 硬件平台： Amazon Web Services (AWS)。
• 实例类型：
  • CPU 实例：m4.16xlarge（用于通用计算）。
  • GPU 实例：p3.16xlarge（用于深度学习训练）。
• 网络： 25Gbps 以太网。

5.2. 评估指标

1. 吞吐量 (Throughput):
   • 定义： 系统单位时间内能完成的任务数量。
   • 单位： Tasks/second (任务/秒)。
2. 延迟 (Latency):
   • 定义： 从任务提交到开始执行或完成的时间间隔。
   • 关注点： 调度延迟对于细粒度任务至关重要。
3. IOPS (Input/Output Operations Per Second):
   • 定义： 对象存储每秒能处理的读写操作次数。对于小对象传输非常关键。
4. 完成时间 (Time to Completion):
   • 定义： 训练一个 RL 模型达到特定分数所需的时间。

5.3. 对比基线 (Baselines)

• OpenMPI: 代表高性能计算（HPC）的标准，速度快但难编程，容错差。

• Horovod: 基于 MPI 的分布式深度学习训练工具，当时的最先进水平（SOTA）。

• Clipper: 专用的机器学习模型服务系统。

• 专用系统 (Special-purpose systems): 针对特定 RL 算法（如 ES, PPO）定制开发的系统。

  ---

6. 实验结果与分析

6.1. 可扩展性与延迟 (Microbenchmarks)

• 线性扩展： Ray 展现了惊人的扩展能力。如下图（原文 Figure 8b）所示，随着节点数从 0 增加到 60，吞吐量线性增长，超过了 100 万任务/秒。在 100 个节点时达到 180 万任务/秒。

• 本地化感知的优势： 如下图（原文 Figure 8a）所示，Ray 的调度器能够感知数据位置。如果忽略数据位置（Random placement），随着数据量增大，延迟会增加 1-2 个数量级。Ray 通过将任务调度到数据所在节点，保持了低延迟。

  该图像是图表，展示了Ray的任务调度和可扩展性的性能。左侧(a)显示了在不同对象大小下，本地敏感调度与无感调度的平均任务延迟对比(单位为秒)，而右侧(b)则展示了随着节点数增加，Ray能够达到的任务处理能力，以每秒百万任务为单位。

• 对象存储性能： 如下图（原文 Figure 9）所示，对于大对象，单客户端写入吞吐量超过 15GB/s（接近内存带宽极限）；对于小对象，IOPS 超过 18K。这证明了共享内存架构的高效性。

  该图像是图表，展示了对象存储的写入吞吐量和每秒输入输出操作数（IOPS）。随着对象大小的变化，吞吐量超过15GB/s，IOPS达到18K，图中使用的线程数分别为1、2、4、8和16，结果为5次试验的平均值。

6.2. 容错性 (Fault Tolerance)

• GCS 的作用： 下图（原文 Figure 10）展示了 GCS 的稳定性。即使在人为触发节点故障和重配置时，客户端观察到的延迟波动也控制在 30ms 以内。
• 任务与 Actor 恢复： 下图（原文 Figure 11）显示了 Ray 在节点被移除时（虚线处）的恢复能力。

  • (a) 任务恢复： 吞吐量短暂下降后迅速恢复，因为 Ray 根据 GCS 中的谱系自动重新调度了丢失的任务。

  • (b) Actor 恢复： 同样能够从检查点（Checkpoint）恢复 Actor 状态，确保持续运行。

    该图像是图表，展示了 Ray 框架在任务重构和演员重构中的容错能力。图 (a) 显示在节点被移除时，任务依赖关系的恢复过程，而图 (b) 展示了演员的重建情况。图中任务的吞吐量以每秒任务数表示，常规任务和重新执行的任务分别用不同颜色区分。

6.3. RL 应用性能对比

Ray 在三个典型的 RL 场景中均表现出色：

6.3.1. 分布式训练 (Allreduce)

在分布式 SGD 训练中，Ray 实现的 Allreduce（全归约）算法性能惊人。如下图（原文 Figure 12a）所示，在 16 个节点上，Ray 完成 100MB 数据的 Allreduce 仅需约 200ms，比 OpenMPI 快 1.5 倍。这得益于 Ray 能够利用多线程进行网络传输，而 OpenMPI 通常是单线程的。

该图像是图表，展示了 Ray 与 OpenMPI 的迭代时间比较（a）以及调度器延迟对 Ray 环的影响（b）。图表左侧部分显示不同对象大小下的迭代时间，右侧部分则反映了添加调度器延迟对迭代时间的影响。

6.3.2. 服务 (Serving)

下表（原文 Table 3）对比了 Ray 和 Clipper 在嵌入式服务场景下的表现。

• 结果分析： Ray 的吞吐量远高于 Clipper，特别是在小输入场景下。这是因为 Clipper 是为通过 REST API 服务外部请求设计的，而 Ray 针对的是同一集群内的低延迟调用（使用共享内存），避免了昂贵的序列化和网络开销。

Table 3: Throughput comparisons for Clipper and Ray. (以下是原文 Table 3 的数据转录)

6.3.3. 真实 RL 算法 (ES & PPO)

• 进化策略 (Evolution Strategies, ES): 如下图（原文 Figure 14a）所示，Ray 的实现可以扩展到 8192 个核心，中位完成时间仅为 3.7 分钟，比之前发布的最佳结果快两倍以上。相比之下，专用系统在 2048 核时就因控制节点过载而崩溃。
• 近端策略优化 (PPO): 如下图（原文 Figure 14b）所示，Ray 的 PPO 实现不仅速度超过了优化的 MPI 实现，而且成本更低。

  • 原因： Ray 支持异构调度。MPI 实现要求每个 CPU 进程都配一个 GPU（即使某些任务只用 CPU），导致 GPU 闲置。Ray 允许只给需要 GPU 的任务（训练）分配 GPU，给只需要 CPU 的任务（模拟）分配 CPU，从而大幅节省了昂贵的 GPU 资源。

    该图像是图表，展示了在Humanoidv1任务中，不同CPU数量下Ray ES与参考ES的平均解决时间，以及不同GPU配置下Ray PPO与MPI PPO的平均解决时间。左侧(a)显示Evolution Strategies的结果，右侧(b)显示PPO的结果，均以分钟为单位。

---

7. 总结与思考

7.1. 结论总结

这篇论文介绍的 Ray 是一个里程碑式的分布式框架。它通过引入 GCS (全局控制存储) 和 Task/Actor 统一抽象，成功解决了 AI 应用（特别是强化学习）中对动态性、低延迟和高吞吐的苛刻要求。实验证明，Ray 不仅在微基准测试中展现了线性扩展能力，而且在真实的 RL 算法中击败了高度优化的专用系统。

7.2. 局限性与未来工作

• 调度器的复杂性： 论文提到，为了实现通用的调度，很难针对特定工作负载做极致优化。由于调度器不掌握全局计算图的全貌（因为是动态生成的），某些全局优化可能无法实现。
• 垃圾回收 (GC): 存储完整的任务谱系会导致 GCS 数据量无限增长。论文提到正在开发基于谱系的垃圾回收策略，以限制存储成本。
• API 级别： 当前 API 相对底层（Task/Actor），虽然灵活但对数据科学家有一定门槛。未来工作包括在 Ray 之上构建高级库（后来确实发展出了 RLlib, Ray Tune 等）。

7.3. 个人启发与批判

• 架构的胜利： Ray 的成功再次印证了“控制平面与数据平面分离”这一系统设计原则的强大。GCS 的设计让复杂的分布式状态管理变得可行，这种思想在 SDN（软件定义网络）中也很常见。
• 对 AI 系统的启示： 传统的“大数据系统”和“高性能计算系统”正在融合。AI 应用既需要大数据的容错和易用性，又需要 HPC 的低延迟和高性能。Ray 填补了这一空白。
• 思考： 虽然 GCS 解决了扩展性问题，但 Redis 本身是否会成为瓶颈？虽然论文中展示了很好的扩展性，但在超大规模（数万节点）集群下，中心化的 GCS（即使分片）的网络带宽压力仍然值得关注。此外，Ray 的编程模型要求开发者显式地管理状态（Actor）和无状态逻辑，这对习惯了单机 Python 的开发者来说仍有一定的心智负担。