Ray:一个分布式应用框架

下一代AI应用程序将持续与环境交互并从这些交互中学习。这些应用程序在性能和灵活性方面都提出了新的和苛刻的系统要求。在本文中，我们考虑这些要求并提出Ray ----- 一个分布式系统来解决它们。Ray实现了一个统一的接口，可以表示任务并行和基于actor的计算，由单个动态执行引擎支持。为了满足性能要求，Ray采用分布式调度程序和分布式容错存储来管理系统的控制状态。在我们的实验中，我们展示了超过每秒180万个任务的规模，并且比几个现有的具有挑战性的强化学习应用程序的性能好。

1.介绍

在过去的20年里，许多研究组织一直在收集和利用不断增长的数据。这导致开发了大量用于分布式数据分析的框架，包括批处理、流处理和图处理系统。这些框架的成功使得研究组织能够将分析大型数据集作为其业务或科学战略的核心部分，并引领了大数据时代。

最近，以数据为中心的应用范围已扩大到包括更复杂的人工智能(AI)或机器学习(ML)技术。典型的例子是监督学习，其中数据点伴随着标签，而将数据点映射到标签的主要技术是由深度神经网络提供的。这些深度神经网络的复杂性引发了另一波框架研究热潮，这些框架研究的重点是深度神经网络的训练及其在预测中的应用。这些框架通常利用专门的硬件(例如gpu和TPUs)，目的是减少批处理设置中的训练时间。例如TensorFlow、MXNet和PyTorch。

然而，人工智能的前景远比经典的监督学习要广阔。新兴的人工智能应用程序必须越来越多地在动态环境中运行，对环境变化作出反应，并采取一系列行动来实现长期目标。他们不仅要利用收集到的数据，而且要探索可能的行动空间。这些更广泛的需求自然被构建在强化学习（RL）的范式中。RL处理基于延迟和有限反馈的学习在不确定环境中持续运行。基于RL的系统已经取得了显著的成果，如谷歌的AlphaGo击败了人类世界冠军，并开始进入对话系统、无人机和机器人操作。

RL应用程序的核心目标是学习策略----从环境状态映射到随着时间的推移产生有效性能的操作选择，例如，赢得游戏或驾驶无人机。在大型应用程序中寻找有效的策略需要三个主要功能。首先，RL方法通常依赖于模拟来评估策略。模拟可以探索许多不同的动作序列选择，并了解这些选择的长期后果。其次，与有监督学习算法一样，RL算法需要执行分布式训练，以基于模拟生成的数据或与物理环境的交互来改进策略。第三，策略是为控制问题提供解决方案，因此在交互式闭环和开环控制场景中为策略服务是必要的。

这些特性驱动了新的系统需求:RL必须支持细粒度计算的系统(例如,渲染行为与现实世界交互时以毫秒级为单位,并进行大量的模拟),必须支持异构性在时间(例如,模拟可能需要毫秒或小时)和资源使用(如训练gpu和cpu仿真),必须支持动态执行,模拟的结果或与环境的相互作用可以改变未来的计算。因此，我们需要一个动态计算框架，以毫秒级的延迟每秒处理数百万个异构任务。

为大数据工作负载或有监督的学习工作负载开发的现有框架无法满足这些新的RL需求。批量同步并行系统（如mapreduce、apache spark和dryad不支持细粒度模拟或策略服务。任务并行系统，如ciel和dask几乎不支持分布式培训和服务。同样适用于流媒体系统，如Naiad和Storm。像TensorFlow和MXnet这样的分布式深度学习框架自然不支持模拟和服务。最后，诸如TensorFlowServing和Clipper等模型服务系统既不支持训练也不支持模拟。

虽然原则上可以通过将几个现有的系统(例如用于分布式培训的Horovod、用于服务的Clipper和用于模拟的CIEL)拼接在一起来开发端到端解决方案，但实际上由于应用程序中这些组件的紧密耦合，这种方法是站不住脚的。因此，今天的研究人员和实践者为专门的RL应用程序构建一次性的系统。这种方法通过将调度、容错和数据移动等标准系统挑战推到每个应用程序上，给分布式应用程序的开发带来了巨大的系统工程负担。

为了满足性能需求，Ray分布了两个通常集中在现有框架中的组件:(1)任务调度程序和(2)元数据存储，元数据存储维护计算沿袭（lineage）和数据对象目录。这允许Ray以毫秒级的延迟每秒调度数百万个任务。此外，Ray还为任务和参与者提供了基于沿袭（lineage）的容错，并为元数据存储提供了基于复制的容错。

虽然Ray支持在RL应用程序环境中提供服务、训练和模拟，但这并不意味着它应该被看作是在其他环境中为这些工作负载提供解决方案的系统的替代品。特别是，Ray的目标不是替代Clipper和TensorFlow这样的服务系统，因为这些系统解决了部署模型时面临的更广泛的挑战，包括模型管理、测试和模型组合。同样，尽管Ray具有灵活性，但它并不能替代一般的数据并行框架(如Spark)，因为它目前缺乏这些框架提供的丰富功能和api(例如，straggler mitigation、query optimization)。

我们做出以下贡献：

我们设计并构建了一个分布式框架，将训练、模拟和服务于新兴RL应用程序的必要组件结合起来。
为了支持这些工作负载，我们在动态任务执行引擎之上统一了参与者(actor)和任务(task)并行抽象。
为了实现可扩展性和容错性，我们提出了一种系统设计原则，其中控制状态存储在分片元数据存储中，而所有其他系统组件都是无状态的。
为了实现可扩展性，我们提出了自下而上的分布式调度策略。

2.动机和要求

我们首先考虑RL系统的基本组件，并充实Ray的关键需求。如图1所示，在RL设置中，代理与环境反复交互。代理的目标是学习一种使奖励最大化的策略。策略是从环境状态到操作选择的映射。环境、代理、状态、操作和奖励的精确定义是特定于应用程序的。

图二典型的强化学习的学习策略伪代码

为了学习策略，代理(agent)通常采用两个步骤:(1)策略评估和(2)策略改进。为了评估策略，代理与环境交互(例如，与环境的模拟)来生成轨迹，其中一个轨迹由当前策略生成的一系列(状态、奖励)元组组成。然后，代理利用这些轨迹来改进策略;即，以使奖励最大化的梯度方向更新策略。图2显示了代理用于学习策略的伪代码示例。这个伪代码通过调用rollout (environment, policy)来生成轨迹来评估策略。train policy()之后使用这些轨迹通过 policy. update (trajectories)改新当前的策略。这个过程重复进行，直到策略收敛为止。

因此，RL应用程序的框架必须为训练、服务和模拟提供有效的支持(图1)。接下来，我们简要描述这些工作负载。

训练–通常在分布式设置中运行随机梯度下降(SGD)，以更新策略。分布式SGD通常依赖于allreduce聚合步骤或参数服务器

服务使用经过训练的策略根据环境的当前状态呈现操作。服务系统旨在最小化延迟，并最大化每秒的决策数量。为了进行扩展，负载通常在为策略提供服务的多个节点之间进行平衡。

最后，大多数现有的RL应用程序使用模拟来评估策略，当前的RL算法的采样效率不足以完全依赖于与物理世界交互获得的数据。这些模拟在复杂性上差别很大，他们可能需要几毫秒（例如，模拟象棋游戏中的移动）到几分钟（例如，模拟自动驾驶汽车的真实环境）。

与监督学习不同，在监督学习中，训练和服务可以由不同的系统单独处理，在RL中，所有三个工作负载都紧密耦合在一个应用程序中，它们之间有严格的延迟要求。目前，没有框架支持工作负载的这种耦合。理论上，可以将多个专门的框架组合在一起以提供总体功能，但在实践中，在RL环境中，所产生的系统间数据移动和延迟是禁止的。因此，研究人员和实践者一直在构建自己的一次性系统。

这种情况需要为RL开发新的分布式框架，以有效地支持训练、服务和模拟。特别是，这种框架应满足以下要求：

细粒度、异构计算。计算的持续时间可以从毫秒（例如，采样行动）到小时（例如，训练一个复杂的策略）。此外，训练通常需要异构硬件（如CPU、GPU或TPU）。

灵活的计算模型。RL应用程序需要无状态和有状态计算。无状态计算可以在系统中的任何节点上执行，这使得在需要时很容易实现负载平衡和计算到数据的移动。因此，无状态计算非常适合于细粒度模拟和数据处理，例如从图像或视频中提取特征。相反，状态计算非常适合于实现参数服务器、对支持GPU的数据执行重复计算或运行不公开其状态的第三方模拟器。

动态执行。RL应用程序的几个组件需要动态执行，因为计算完成的顺序并不总是预先知道（例如，模拟完成的顺序），计算结果可以决定未来的计算（例如，模拟结果将决定我们是否需要执行更多的模拟）。

通过上边的分析得出两个结论。首先，为了在大型集群中实现高利用率，这种框架必须每秒处理数百万个任务；其次，这种框架不必从头开始实现深层神经网络或复杂模拟器。相反，它应该能够与现有模拟器和深度学习框架无缝集成。

名称	描述
futures = f.remote(args)	远程执行函数f.remote()可以接受对象或期货作为输入，并返回一个或多个期货。这是非阻塞的
objects = ray.get(futures)	返回与一个或多个期货关联的值。这是阻塞。
ready futures = ray.wait(futures,k,timeout)	一旦k完成或超时，返回相应任务已完成的期货。
actor = Class.remote(args)或futures = actor.method.remote(args)	将class类实例化为远程参与者，并返回它的句柄。调用远程参与者上的方法并返回一个或多个期货。两者都是阻塞。

表1: Ray API
注：个人更愿意把futures翻译为对象id。

3.规划计算模型

Ray实现了一个动态任务图计算模型，也就是说，它将应用程序建模为在执行过程中演化的依赖任务图。在这个模型的基础上，Ray提供了一个参与者和一个任务并行编程抽象。这种统一将ray与相关系统（如ciel，它只提供任务并行抽象）以及Orleans或Akka区分开来，后者主要提供参与者抽象。

3.1 编程模型

任务。任务表示在无状态工作机上执行远程函数。当调用远程函数时，将立即返回表示任务结果的期货（ID）。可以使用 ray.get() 检索，并将其作为参数传递到其他远程函数，而无需等待其结果。这允许用户在捕获数据依赖项时表示并行性。表1显示了Ray的API。.

远程函数在不可变的对象上运行，并且应该是无状态和无副作用的：它们的输出完全由它们的输入决定。这意味着等幂性，它通过在失败时重新执行函数来简化容错。

Actors。参与者表示有状态的计算。每个参与者公开可以远程调用并串行执行的方法。方法执行类似于任务，因为它是远程执行并返回期货（ID），但不同的是它在有状态工作线程上执行。一个参与者的句柄可以传递给其他参与者或任务，使他们可以在该参与者上调用方法。

任务（无状态）	Actor（有状态）
细粒度负载平衡	粗粒度负载平衡
支持对象位置	地方支持差
小更新的高开销	小更新开销低
高效的故障处理	检查点开销

表2：任务与参与者权衡

表2总结了任务和参与者的属性。任务通过在任务粒度、输入数据位置利用负载感知调度实现细粒度负载平衡，因为每个任务都可以在存储其输入的节点上调度，并且恢复开销较低，因为不需要检查点和恢复中间状态。相反，参与者提供更高效的细粒度更新，因为这些更新是在内部而不是外部状态上执行的，通常需要序列化和反序列化。例如，可以使用参与者来实现参数服务器和基于GPU的迭代计算（例如，训练training）。此外，参与者可以用来包装第三方模拟器和其他难以序列化的不透明句柄。

为了满足异构性和灵活性的需求（第2节），我们以三种方式增强API。首先，为了处理具有异构持续时间的并发任务，我们引入ray.wait()，它等待前k个可用结果，而不是等待所有结果(如 ray.get())。第二，为了处理资源异构任务，我们允许开发人员指定资源需求，以便Ray调度程序能够有效地管理资源。第三，为了提高灵活性，我们启用了嵌套的远程函数，这意味着远程函数可以调用其他远程函数。这对于实现高可伸缩性也是至关重要的（第4节），因为它允许多个进程以分布式方式调用远程函数。

3.2 计算模型

Ray采用了动态任务图计算模型，其中远程函数和actor方法的执行在其输入可用时由系统自动触发。在本节中，我们将描述如何从用户程序（图3）构建计算图（图4）。这个程序使用表1中的API来实现图2中的伪代码。

首先，忽略参与者，计算图中有两种类型的节点:数据对象和远程函数调用或任务。还有两种类型的边:数据边和控制边。数据边捕获数据对象和任务之间的依赖关系。更精确地说，如果数据对象D是任务T的输出，那么我们将添加一条从T到D的数据边缘。类似地，如果D是T的输入，我们添加一条从D到T的数据边。控制边捕获嵌套远程函数产生的计算依赖关系(第3.1节):如果task t1调用task t2，那么我们将从t1添加一个控制边到t2。

图3:Ray中实现图2中示例的python代码。请注意，@ray.remote表示远程函数和参与者。远程函数和actor方法的调用返回该远程函数或actor对象的ID，可以将返回的对象ID传递给后续的远程函数或actor方法，以对任务依赖项进行编码。每个参与者都有一个在其所有方法之间共享的环境对象self.env。

Actor方法调用也表示为计算图中的节点。它们与任务相同，但有一个关键区别。为了捕获同一参与者上后续方法调用之间的状态依赖关系，我们添加了第三种类型的边:状态边。如果在同一参与者上方法Mi之后立即调用方法Mj，那么我们将一条有状态的边从Mi添加到Mj。因此，在同一个actor对象上调用的所有方法形成一个由有状态边连接的链(图4)。

图4：与图3中的train-policy.remote()调用相对应的任务图。远程函数调用和actor方法调用对应于任务图中的任务。这个图显示了两个actor。每个参与者（标记为A1i和A2i的任务）的方法调用在它们之间具有状态边，表明它们共享可变的参与者状态。从训练策略到它调用的任务都有控制边。为了并行地训练多个策略，我们可以多次调用train-policy.remote()。

有状态边帮助我们将参与者嵌入到其他无状态任务图中，因为它们捕获共享参与者内部状态的连续方法调用之间的隐式数据依赖关系。有状态的边也使我们能够维持沿袭。与其他数据流系统一样，我们跟踪数据沿袭以支持重建。通过在沿袭图中显式地包含状态边，我们可以很容易地重建丢失的数据，无论是由远程函数还是actor方法生成的（第4.2.3节）。

4 结构

Ray的体系结构包括（1）实现API的应用层，（2）提供高可扩展性和容错性的系统层。

4.1 应用层

应用层由三种类型的过程组成：

driver ：执行用户程序的进程。
worker ：执行由驱动程序或其他worker程序调用的任务(远程函数)的无状态进程。worker自动启动，并由系统层分配任务。当声明远程函数时，该函数将自动发布给所有worker。worker连续执行任务，任务之间不维护任何本地状态。
Actor ：一个有状态的进程，在调用时只执行它公开的方法。与worker不同，参与者是由worker或driver显式实例化的。与worker一样，参与者也连续地执行方法，只是每个方法都依赖于上一个方法执行产生的状态。

图5:Ray的体系结构由两部分组成：应用层和系统层。应用层实现API和第3节描述的计算模型，系统层实现任务调度和数据管理，以满足性能和容错要求。

4.2 系统层

系统层由三个主要组件组成：全局控制存储(Global Control Store 简称 GCS)、分布式调度和分布式对象存储。所有组件都具有水平扩展性和容错性。

4.2.1 全局控制存储（GCS）

全局控制存储（GCS）维护系统的整个控制状态，这是我们设计的一个独特特征。GCS的核心是一个具有pubsub功能的键值（ key-value）存储。我们使用切分来实现规模，每个切分链复制提供容错。GCS及其设计的主要原因是保持系统的容错性和低延迟，该系统每秒可动态生成数百万个任务。

节点故障时的容错的解决方案需要来维护沿袭信息。现有的基于沿袭的解决方案侧重于粗粒度的并行性，因此可以使用单个节点（如master、driver）来存储沿袭，而不会影响性能。然而，这种设计对于细粒度和动态工作负载之类的模拟是不可扩展的。因此，我们将持久沿袭存储与其他系统组件分离，允许每个组件独立扩展。保持低延迟需要最小化任务调度中的开销，这涉及到选择执行位置，以及随后的任务调度，这还涉及到从其他节点检索远程输入。许多现有的数据流系统通过将对象的位置和大小存储在一个集中的调度程序中，这是调度程序不是瓶颈时的自然设计。但是，Ray目标的规模和粒度要求将集中式调度程序远离关键路径。对于像allreduce这样的分布式训练非常重要的原语，在每个对象传输中都包含使用调度器是非常昂贵的，这既需要通信量，也需要对延迟敏感。因此，我们将对象元数据存储在GCS中而不是调度程序中，从而完全将任务分发与任务调度分离开来。

总之，GCS显著简化了Ray的总体设计，因为它使系统中的每个组件都是无状态的。这不仅简化了对容错的支持（即，在发生故障时，组件只需重新启动并从GCS读取沿袭），而且还可以轻松地独立地扩展分布式对象存储和调度程序，因为所有组件都通过GCS共享所需的状态。另外一个好处是调试、分析和可视化工具的简单开发。

4.2.2 自下而上的分布式调度程序

如第2节所述，Ray需要动态地调度每秒数百万个任务，这些任务可能只需要几毫秒。我们所知道的集群调度程序都不满足这些要求。大多数集群计算框架，如spark、ciel和dryad都实现了一个集中式调度程序，它可以提供局部性，但在几十毫秒的时间内会出现延迟。分布式调度程序，如获取工作(work stealing)、sparrow和canary都可以实现高规模，但它们要么不考虑数据局部性，要么承担任务。属于独立工作，或者假设计算图已知。

为了满足上述要求，我们设计了一个由全局调度程序和每个节点本地调度程序组成的两级层次调度程序。为了避免全局调度程序过载，在节点上创建的任务首先提交给节点的本地调度程序。本地调度程序在本地调度任务，除非节点过载（即，其本地任务队列超过预定义的阈值），或者它不能满足任务的要求（例如，缺少GPU）。如果本地调度程序决定不在本地调度任务，它会将其转发给全局调度程序。由于这个调度程序首先尝试在本地调度任务（即在调度层次结构的左侧），所以我们称它为自底向上调度程序。

图6：自下而上的分布式调度程序。任务从驱动程序和worker程序自下而上提交给本地调度程序，并仅在需要时转发给全局调度程序（第4.2.2节）。每个箭头的厚度与其请求速率成正比。

全局调度程序考虑每个节点的负载和任务的约束，以做出调度决策。更准确地说，全局调度程序标识具有任务所请求类型的足够资源的节点集，并且在这些节点中，选择提供最低估计等待时间的节点。在给定节点上，此时间是（i）任务将在该节点排队的估计时间（即，任务队列大小乘以平均任务执行）和（ii）任务的远程输入的估计传输时间（即，远程输入的总大小除以平均带宽）的总和。全局调度程序通过监控获取每个节点的队列大小和节点资源可用性，以及任务输入的位置及其在GCS中的大小。此外，全局调度程序使用简单的指数平均计算平均任务执行和平均传输带宽。如果全局调度程序成为瓶颈，我们可以实例化更多的副本，所有副本都通过GCS共享相同的信息。这使得我们的调度程序架构具有高度的可扩展性。

4.2.3 内存中的分布式对象存储

为了最小化任务延迟，我们实现了一个内存中的分布式存储系统来存储每个任务或无状态计算的输入和输出。在每个节点上，我们通过共享内存实现对象存储。这允许在同一节点上运行的任务之间零拷贝数据共享。作为数据格式，我们使用Apache Arrow。

如果任务的输入不是本地的，则在执行之前将输入复制到本地对象存储区。此外，任务将其输出写入本地对象存储。复制消除了由于热数据对象导致的潜在瓶颈，并将任务执行时间最小化，因为任务只从本地内存读取/写入数据。这增加了计算绑定工作负载的吞吐量，许多人工智能应用程序共享的配置文件。对于低延迟，我们将对象完全保存在内存中，并根据需要使用LRU策略将其逐出磁盘。

与现有的集群计算框架（如spark和dryad）一样，对象存储仅限于不可变的数据。这避免了对复杂一致性协议的需要（因为对象没有更新），并简化了对容错的支持。在节点故障的情况下，Ray通过沿袭重新执行来恢复任何需要的对象。存储在GCS中的沿袭在初始执行期间跟踪无状态任务和有状态参与者；我们使用前者在存储中重建对象。

为了简单起见，我们的对象存储不支持分布式对象，即每个对象都适合一个节点。像大型矩阵或树这样的分布式对象可以作为未来的集合在应用程序级别实现。

4.2.4 实施

Ray是加州大学伯克利分校开发的一个活跃的开源项目。 Ray完全集成了Python环境，只需运行pip install ray即可轻松安装。该实现包括约40 K行代码（LoC），72％用于系统层的C ++，28％用于应用层。 GCS每个分片使用一个Redis键值存储，完全是单键操作。 GCS表由对象和任务ID进行分片以进行缩放，并且每个分片都经过链式复制以实现容错。我们将本地和全局调度程序实现为事件驱动的单线程进程。在内部，本地调度程序维护本地对象元数据的缓存状态，等待输入的任务以及准备分派给工作程序的任务。为了在不同的对象存储之间传输大对象，我们跨多个TCP连接对对象进行条带化。

4.3 把所有的东西放在一起

图7用一个简单的例子说明了Ray如何端到端地工作，这个例子添加了两个对象a和b，可以是标量或矩阵，并返回结果c。远程函数add()在初始化时自动注册到GCS，并分发给系统中的每个worker（图7a中的步骤0）。

图7a显示了由调用add.remote(a,b)的驱动程序触发的逐步操作，其中a和b分别存储在节点N1和N2上。driver提交(a, b)添加到本地调度程序(步骤1),将它转发到一个全局调度器(步骤2)。接下来,全球调度器在GCS上查找add(a, b) 参数的位置(步骤3),决定安排任务到存储参数b的节点N2上(步骤4)。节点N2的当地调度器检查是否本地对象存储包含add(a, b)的参数(步骤5)。由于本地存储没有a的对象,就到GCS上寻找a的位置(步骤6)。了解到a存储在N1，N2的对象存储在本地复制它（步骤7）。由于add()的所有参数现在都存储在本地，本地调度程序在本地工作程序（步骤8）调用add()，后者通过共享内存访问参数（步骤9）。

图7：添加a和b并返回c的端到端示例。实线是数据平面操作，虚线是控制平面操作。（a）函数add() 由节点1（N1）向GCS注册，在N1上调用，在N2上执行。（b）N1使用ray.get() 获取add() 的结果。c的对象条目在步骤4中创建，并在步骤6中将c复制到N1后更新（更新GCS注册表）

图7b分别显示了在N1执行ray.get() 和在N2执行add() 所触发的逐步操作。在ray.get(idc) 调用时，驱动程序使用add()返回的期货（ID）idc检查本地对象存储的值c（步骤1）。因为本地对象存储不存储c，所以它在GCS中查找其位置。此时，在GCS中c没有条目，因为C尚未创建。因此，N1的对象存储向对象表注册回调，以便在创建c的条目时触发该回调（步骤2）。同时，在N2，add()完成其执行，将结果c存储在本地对象存储区（步骤3），然后将c的条目添加到GCS（步骤4）。因此，GCS使用c的条目触发对N1对象存储的回调（步骤5）。接下来，N1从N2复制c（步骤6），并返回c到ray.get()（步骤7），这最终完成了任务。

虽然本例涉及大量RPC，但在许多情况下，这个数字要小得多，因为大多数任务都在本地调度，而GCS响应由全局和本地调度程序缓存。

5 评价

在我们的评估中，我们研究了以下问题：

Ray在多大程度上满足了第2节中列出的延迟、可扩展性和容错要求？（第5.1节）
对使用Ray的API编写的分布式原语（如AllReduce）施加了哪些开销？（第5.1节）
在RL工作负载的背景下，Ray如何与专门的训练、服务和模拟系统进行比较？（第5.2节）
与定制系统相比，Ray为RL应用程序提供了哪些优势？（第5.3节）

所有的实验都是在AmazonWeb服务上运行的。除非另有说明，我们使用M4.16XLarge CPU实例和P3.16XLarge GPU实例。

5.1 微基准（Microbenchmarks）

位置感知任务放置（Locality-aware task placement）。细粒度负载平衡和位置感知放置是Ray中任务的主要优势。参与者一旦被放置，就不能将它们的计算移动到其他远程节点上，而任务可以。在图8a中，放置的任务没有数据位置意识（与actor方法一样），在10-100MB的输入数据大小下，延迟会增加1-2个数量级。Ray通过共享对象存储来统一任务和参与者，允许开发人员使用任务，例如对模拟器（参与者）生成的输出进行昂贵的后处理。

图8：（a）任务利用位置感知放置。1000个具有随机对象依赖关系的任务被调度到两个节点之一。使用位置感知策略，任务延迟与任务输入的大小无关，而不是增长1-2个数量级。（b）利用GCS和自下而上的分布式调度程序实现近线性可扩展性。Ray通过60个节点每秒可完成100万个任务。X∈{70,80,90}因成本而省略。

端到端的可扩展性(End-to-end scalability)。其主要优点之一是全局控制存储（GCS）和自下而上的分布式调度程序能够水平扩展系统以支持高吞吐量的细粒度任务，同时保持容错性和低延迟任务调度。在图8b中，我们在空任务的并行工作负载上评估这种能力，从而增加X轴上的集群大小。我们在逐步增加任务吞吐量时观察到近乎完美的线性关系。Ray在60个节点的吞吐量超过每秒100万个任务，并且在100个节点的吞吐量继续线性扩展到每秒180万个任务以上。最右边的数据点显示，Ray可以在不到一分钟（54s）的时间内处理1亿个任务，并且变异性最小。正如预期的那样，增加任务持续时间会按平均任务持续时间的比例减少吞吐量，但是总体的可伸缩性仍然是线性的。由于对象依赖性和应用程序并行性固有的限制，许多实际的工作负载可能表现出更有限的可伸缩性，这表明了我们在高负载下的整体架构的可伸缩性。

对象存储性能（Object store performance）。为了评估对象存储的性能（第4.2.3节），我们跟踪两个指标：IOPS（对于小对象）和写吞吐量（对于大对象）。在图9中，随着对象大小的增加，单个客户机的写入吞吐量超过15GB/s。对于较大的对象，memcpy（字符拷贝）控制对象创建时间。对于较小的对象，主要开销在客户端和对象存储之间的序列化和IPC中。

图9：对象存储写入吞吐量和IOPS。对于单个客户机，16核实例（M4.4XL）上的大对象吞吐量超过15GB/s（红色），小对象吞吐量超过18K IOPS（青色）。它使用8个线程来复制大于0.5MB的对象，对于小对象使用1个线程。条形图报告1、2、4、8、16线程的吞吐量。结果平均超过5次。

GCS容错（GCS fault tolerance）。为了在提供强一致性和容错性的同时保持低延迟，我们在Redis之上构建了一个轻量级的链复制层。图10a模拟了向GCS记录Ray任务和从GCS读取任务，其中键为25字节，值为512字节。客户机以最快的速度发送请求，一次最多有一个正在运行的请求。失败将从客户机(接收到显式错误，或在重试时超时)或链中的任何服务器(接收到显式错误)报告给链主。总的来说，重新配置导致客户机观察到的最大延迟小于30ms(这包括故障检测和恢复延迟)。

（a）从提交任务的客户机上查看的GCS读写延迟时间线。链从2个副本开始。我们手动触发如下重新配置。在t≈4.2s时，链成员被杀死；紧接着，一个新的链成员加入，启动状态转移，并将链恢复为双向复制。尽管进行了重新配置，但客户机观察到的最大延迟仍低于30毫秒。

（b）Ray GCS通过GCS刷新保持恒定的内存占用。如果没有GCS刷新，内存占用将达到最大容量，并且工作负载无法在预定的时间内完成（由红叉号示意）。

图10：Ray GCS容错和冲洗。

GCS刷新(GCS flushing)。Ray可以定期将GCS的内容刷新到磁盘上。在图10b中，我们按顺序提交5000万个空任务，并监视GCS内存消耗。正如预期的那样，它会随着跟踪的任务数量线性增长，最终达到系统的内存容量。此时，系统将停止工作，工作负载无法在合理的时间内完成。通过定期刷新GCS，我们实现了两个目标。首先，内存占用被限制在一个用户可配置的级别（在微基准中，我们采用了一种积极的策略，在这种策略中，消耗的内存尽可能低）。第二，刷新机制为长时间运行的Ray应用程序提供了一种自然的快照沿袭到磁盘的方法。

从任务失败中恢复（Recovering from task failures）。在图11a中，我们演示了Ray 使用持久的GCS沿袭存储透明地从worker节点故障恢复和弹性伸缩的能力。工作负载，在m4.xlarge上运行实例，由驱动程序提交的100ms任务的线性链组成。当节点被删除(在25s、50s、100s)时，本地调度程序将重构链中先前的结果，以便继续执行。整个过程中，每个节点的总体吞吐量保持稳定。

图11：Ray故障容限。（A）Ray在删除节点（虚线）时重建丢失的任务依赖项，并在重新添加节点时恢复到原始吞吐量。每个任务都是100毫秒，并且依赖于由以前提交的任务生成的对象。（b）actor从最后一个检查点重新构建。在t=200s时，我们杀死10个节点中的2个，导致集群中2000个参与者中的400个在剩余节点上恢复（t=200-270s）。

从参与者失败中恢复(Recovering from actor failures)。通过将actor方法直接编码为依赖关系图中的状态边缘，我们可以重用图11a中的相同对象重建机制，为状态计算提供透明的容错性。Ray还利用用户定义的检查点函数来限定参与者的重建时间（图11b）。使用最小的开销，检查点只允许重新执行500个方法，而不需要检查点就可以重新执行10000个方法。在未来，我们希望进一步减少参与者重建时间，例如，允许用户注释不改变状态的方法。

Allreduce。Allreduce是一个分布式通信原语，对于许多机器学习工作负载非常重要。在这里，我们评估Ray是否能够原生地支持一个环形allreduce实现，并提供足够低的开销来匹配现有的实现。我们发现Ray在200ms和1200ms中分别在100MB和1GB上完成了16个节点上的allreduce，这两个节点的性能分别比流行的MPI实现OpenMPI (v1.10)好1.5和2(图12a)。我们将Ray s的性能归因于它使用多个线程进行网络工作传输，充分利用了AWS上节点之间25Gbps的连接，而OpenMPI则在单个线程上发送和接收数据。对于较小的对象，OpenMPI通过切换到一个更低开销的算法来超越ray，这是一个我们计划在未来实现的优化。

图12：（a）16个M4.16XL节点上AllReduce的平均执行时间。每个工人都在不同的节点上运行。Ray*将Ray限制为1个用于发送的线程和1个用于接收的线程。（b）Ray的低延迟调度对于AllReduce至关重要。

Ray的调度器性能对于实现AllReduce等原语至关重要。在图12b中，我们注入了人工任务执行延迟，并显示性能下降了近2倍，只需要几毫秒的额外延迟。具有诸如spark和ciel等集中式调度程序的系统通常在几十毫秒内就有调度程序开销，这使得这种工作负载不实用。调度程序吞吐量也成为一个瓶颈，因为环所需的任务数量与参与者的数量成正比地减少。

5.2 Building blocks

端到端应用程序（例如，AlphaGo）需要训练、服务和模拟的紧密耦合。在本节中，我们将这些工作负载中的每一个隔离到一个说明典型RL应用程序需求的设置中。由于以RL为目标的灵活编程模型和设计用于支持此编程模型的系统，Ray会匹配这些单独工作负载，有时会超过专用系统的性能。

5.2.1 分布式培训
我们利用Ray参与者抽象来表示模型副本，实现了数据并行同步SGD。模型权重通过AllReduce（5.1）或参数服务器进行同步，这两个服务器都在Ray API之上实现。

在图13中，我们使用相同的TensorFlow模型和每个实验的合成数据生成器，针对最先进的实现评估Ray（同步）参数服务器SGD实现的性能。我们仅将其与基于TensorFlow的系统进行比较，以准确测量Ray施加的开销，而不是深层次学习框架之间的差异。在每次迭代中，模型副本参与者并行计算渐变，将渐变发送到一个切分的参数服务器，然后从参数服务器读取下一次迭代的渐变总和。

图13：分发resnet-101张量流模型（来自官方TF基准）的训练时每秒到达的图像。所有实验都是在通过25Gbps以太网连接的P3.16XL实例上运行的，工作人员按照Horovod的做法为每个节点分配了4个GPU。我们注意到一些与之前报告的测量偏差，可能是由于硬件差异和最近的TensorFlow性能改进造成的。我们使用openMPI 3.0、TF1.8和NCCL2进行所有运行。

图13显示了Ray与在分布张量流的10%以内。（在分布式复制模式下）。这是因为能够在Ray的通用API中表示这些专用系统中的相同应用程序级优化。一个关键的优化是在一次迭代中对梯度计算、传递和求和进行流水线操作。为了将GPU计算与网络传输重叠，我们使用一个定制的TensorFlow操作符将张量直接写入到Ray的对象存储中。

5.2.2 服务

模型服务是端到端应用程序的重要组成部分。Ray主要关注在同一动态任务图（例如，在Ray上的RL应用程序中）中运行的模拟器的模型的嵌入式服务。相反，像Clipper这样的系统专注于为外部客户提供预测服务。

表3:Clipper的吞吐量比较，这是一个专用的服务系统，以及两个嵌入式服务工作负载的ray。我们使用一个剩余网络和一个完全连接的小网络，分别用10毫秒和5毫秒进行评估。客户机查询服务器，每个发送状态的大小分别为4KB和100KB，每批64个。

在此设置中，低延迟对于实现高利用率至关重要。为了说明这一点，在表3中，我们比较了使用Ray actor服务策略与使用开源Clipper系统在REST上实现的服务器吞吐量。在这里，客户机和服务器进程都在同一台机器(一个3.8xlarge实例)上进行协作。这通常是RL应用程序的情况，但不适用于Clipper等系统处理的一般web服务工作负载。由于它的低开销串行化和共享内存抽象，Ray为一个小的完全连接的策略模型实现了一个数量级的更高吞吐量，该策略模型接受大量输入，并且也是在更昂贵的剩余网络策略模型(类似于AlphaGo Zero中使用的模型)上运行更快，需要更小的输入。

5.2.3 模拟

RL中使用的模拟器产生长度可变的结果（“时间步”），由于训练的紧密循环，必须在可用时立即使用。任务异质性和及时性要求使得在BSP风格的系统中很难有效地支持仿真。为了证明这一点，我们将（1）一个MPI实现与（2）一个Ray程序进行了比较，该程序在3轮中在n个核上提交3n个并行模拟运行，在两轮之间有一个全局屏障，在同时将模拟结果收集回驱动程序的同时发出相同的3n任务。表4显示两个系统的规模都很好，但是Ray的吞吐量达到了1.8倍。这将激发一个编程模型，该模型可以动态生成和收集细粒度模拟任务的结果。

表4:openAI gym 摆锤-v0(Pendulum-v0)模拟器的每秒时间步数。Ray允许在大规模运行异构模拟时获得更好的利用率。

5.3 RL应用

如果没有一个能够紧密结合训练、模拟和服务步骤的系统，今天的强化学习算法将作为一次性解决方案来实现，这使得很难合并优化，例如，需要不同的计算结构或使用不同的架构。因此，通过在Ray中实现两个具有代表性的强化学习应用程序，我们能够匹配甚至优于专门为这些算法构建的自定义系统。主要原因是Ray的编程模型的灵活性，它可以表示应用程序级的优化，这需要大量的工程工作才能移植到定制的系统，但是Ray的动态任务图执行引擎显然支持这种优化。

5.3.1 进化策略(Evolution Strategies)

为了在大规模RL工作负载上评估Ray，我们实现了Evolution Strategies（ES）算法，并与传统ES（Reference ES）实现进行了比较，这是一个专门为该算法构建的系统，它依赖于Redis进行消息传递，以及用于数据共享的低级多处理库。该算法周期性地将新策略广播到一组worker，并聚合大约10000个任务的结果（每个任务执行10到1000个模拟步骤）。

如图14a所示，Ray上的一个实现可以扩展到8192个内核。将可用核心加倍，平均完成时间加快1.6倍。相反，专用系统无法在2048核上完成，系统中的工作超过了应用程序驱动程序的处理能力。为了避免这个问题，Ray实现使用了参与者的聚合树，平均时间达到3.7分钟，比最佳发布结果（10分钟）快两倍多。

使用ray的串行实现的初始并行化只需要修改7行代码。使用Ray对嵌套任务和参与者的支持，通过分层聚合可以很容易地实现性能改进。相比之下，引用实现有几百行代码专门用于在worker之间通信任务和数据的协议，并且需要进一步的工程来支持诸如分层聚合之类的优化。

图14:Humanoidv1任务中达到6000分的时间。（a）Ray ES实现可扩展到8192个内核，并且实现了3.7分钟的中值时间，是最佳发布结果的两倍多。专用系统无法运行超过1024个核心。在这个基准上，ES比PPO快，但显示出更大的运行时差异。（b）Ray PPO实现的性能优于专门的MPI实现，GPU更少，只是成本的一小部分。MPI实现每8个CPU需要1个GPU，而Ray版本最多需要8个GPU（每8个CPU永远不超过1个GPU）。

5.3.2 近端策略优化(Proximal Policy Optimization)

我们在Ray中实现了近似策略优化(PPO)，并与使用OpenMPI通信原语的高度优化的参考实现进行了比较。该算法是一种异步的分散收集，当仿真参与者返回到驱动程序时，将新任务分配给仿真参与者。提交任务，直到收集到32万个模拟步骤(每个任务生成10到1000个步骤)。策略更新执行批大小为32768的SGD的20个步骤。本例中的模型参数大约为350KB。这些实验使用的是p2.16xlarge (GPU)和m4.16xlarge (high CPU)实例。

如图14b所示，在所有实验中，Ray实现优于优化的MPI实现，而使用的是GPU的一部分。原因是，Ray具有异构性意识，允许用户通过以任务或参与者的粒度表示资源需求来使用非对称架构。然后，Ray实现可以利用TensorFlow的单进程多GPU支持，并在可能的情况下将对象固定在GPU内存中。由于需要将卷展异步收集到单个GPU进程，因此无法轻松将此优化移植到MPI。实际上，包含了两个自定义的PPO实现，一个对大型集群使用MPI，另一个为GPU优化，但仅限于单个节点。Ray允许实现适用于这两种情况。

Ray处理资源异构性的能力也将PPO的成本降低了4.5倍，因为只有CPU的任务可以安排在更便宜的高CPU实例上。与此相反，MPI应用程序通常呈现对称的体系结构，其中所有进程运行相同的代码并需要相同的资源，在这种情况下，防止使用仅CPU的机器进行扩展。此外，MPI实现需要按需实例，因为它不能透明地处理故障。假设有4个更便宜的现场实例，Ray的容错性和资源感知调度一起减少了18倍的成本。

6 相关工作

动态任务图(Dynamic task graphs)。Ray与CIEL和DASK密切相关。这三个都支持带有嵌套任务的动态任务图，并实现了期货(ID)抽象。CIEL还提供了基于沿袭的容错功能，而DASK和Ray一样，完全与Python集成。然而，Ray在两个方面存在差异，这两个方面具有重要的性能影响。首先，Ray使用参与者抽象扩展了任务模型。这对于分布式训练和服务中的有效状态计算，保持模型数据与计算的一致性是必要的。其次，Ray采用了一个完全分布式和解耦的控制平面和调度程序，而不是依赖于存储所有元数据的单个主控系统。这对于有效地支持诸如allreduce之类的原语而不进行系统修改至关重要。在16个节点上100MB的峰值性能时，AllReduce on Ray（第5.1节）在200毫秒内提交32轮16个任务。同时，DASK报告512核上的最大调度程序吞吐量为3K个任务/秒。对于集中式调度程序，每一轮allreduce都会产生至少～5ms的调度延迟，这会导致最长2×更差的完成时间（图12b）。即使使用分散式调度程序，将控制平面信息与调度程序耦合，也会使后者处于数据传输的关键路径上，从而为每一轮allreduce增加一个额外的往返路径。

数据流系统(Dataflow systems)。流行的数据流系统，如MapReduce、Spark和Dryad已广泛应用于分析和ML工作负载，但它们的计算模型对于细粒度和动态模拟工作负载来说过于限制。Spark和MapReduce实现了BSP执行模型，该模型假定同一阶段中的任务执行相同的计算，所花费的时间大致相同。Dryad放宽了这一限制，但缺少对动态任务图的支持。此外，这些系统都不提供参与者抽象，也不实现分布式可扩展控制平面和调度程序。最后，NAIAD是一个数据流系统，它为某些工作负载提供了改进的可伸缩性，但只支持静态任务图。

机器学习框架(Machine learning frameworks)。 TensorFlow和MXnet针对深度学习工作负载，并有效地利用CPU和GPU。虽然它们在由线性代数操作的静态DAG组成的训练工作负载上取得了很好的性能，但它们对将训练与模拟和嵌入式服务紧密结合所需的更通用计算的支持有限。MXNet提供了对动态任务图的支持，以及对其内部C++ API的支持，但在执行任务进度、任务完成时间或故障时，都不完全支持在执行过程中修改DAG的能力。TensorFlow和MXnet原则上通过允许程序员模拟低级消息传递和同步原语来实现通用性，但是这种情况下的陷阱和用户体验与MPI类似。OpenMPI可以实现高性能，但是编程相对困难，因为它需要显式的协调来处理异构和动态的任务图。此外，它强制程序员显式地处理容错。

参与者系统(Actor systems)。Orleans和Akka是两个非常适合开发高可用性和并发分布式系统的参与者框架。但是，与Ray相比，它们提供的数据丢失恢复支持更少。为了恢复有状态的参与者，Orleans开发人员必须显式地检查参与者状态和中间响应。新Orleans的无状态的参与者可以被复制以扩大规模，因此可以作为任务，但与ray不同，他们没有血统。同样，尽管Akka明确支持跨故障持久化actor状态，但它不为无状态计算（即任务）提供有效的容错。对于消息传递，Orleans至少提供一次，Akka最多提供一次语义。相比之下，Ray提供了透明的容错性和一次性语义，因为每个方法调用都记录在GCS中，并且参数和结果都是不可变的。我们发现，在实践中，这些限制不会影响应用程序的性能。Erlang和C++参与者框架，另外两个基于参与者的系统，对容错的支持同样有限。

全局控制存储和调度(Global control store and scheduling)。以前在软件定义网络（SDN）、分布式文件系统（如GFS）、资源管理（如Omega）和分布式框架（如MapReduce、Boom）中提出了逻辑集中控制平面的概念。Ray从这些开拓性的努力中获得灵感，但也提供了显著的改进。与SDN、Boom和GFS不同，Ray将控制平面信息（如GCS）的存储与逻辑实现（如调度程序）分离。这使得存储层和计算层都可以独立扩展，这是实现可扩展性目标的关键。Omega使用分布式体系结构，其中调度程序通过全局共享状态进行协调。在这种体系结构中，Ray添加了全局调度程序来平衡本地调度程序之间的负载，并以MS级别（而不是第二级别）为目标任务调度。

Ray实现了一个独特的分布式自底向上调度程序，该调度程序具有水平可伸缩性，可以处理动态构建的任务图。与Ray不同，大多数现有的集群计算系统使用集中式调度程序体系结构。虽然Sparrow是分散的，但其调度程序会做出独立的决策，限制可能的调度策略，并且作业的所有任务都由同一全局调度程序处理。Mesos实现了一个两级层次调度程序，但它的顶层调度程序管理框架，而不是单个任务。Canary通过让每个调度程序实例处理任务图的一部分，但不处理动态计算图，实现了令人印象深刻的性能。

Cilk是一种并行编程语言，它的工作窃取调度程序可以有效地实现动态任务图的负载平衡。然而，由于没有像Ray的全局调度程序那样的中央协调器，这种完全并行的设计也很难扩展到支持分布式环境中的数据位置和资源异构性。

7 讨论和经验

建造Ray是一段漫长的旅程。它开始于两年前，在Spark库中执行分布式训练和模拟。然而，BSP模型的相对不灵活、每个任务的高开销和缺少参与者抽象导致我们开发了一个新的系统。自从我们大约一年前发布了Ray，数百人使用了它，有几家公司正在生产中运行它。在这里，我们将讨论我们开发和使用Ray的经验，以及一些早期用户反馈。

应用程序编程接口.在设计API时，我们强调了极简主义。最初我们从一个基本的任务抽象开始。稍后，我们添加了wait（）原语以适应具有异构持续时间的卷展，并添加了actor抽象以适应第三方模拟器，并分摊昂贵初始化的开销。虽然最终得到的API级别相对较低，但已经证明它既强大又易于使用。我们已经使用这个API在Ray上实现了许多最先进的RL算法，包括A3C、PPO、DQN、ES、DDPG和APE-X。在大多数情况下，我们只需要几十行代码就可以将这些算法移植到Ray。基于早期的用户反馈，我们正在考虑增强API以包含更高级别的原语和库，这也可以通知调度决策。

局限性。考虑到工作负载的一般性，专门的优化很难实现。例如，我们必须在不完全了解计算图的情况下做出调度决策。在Ray中调度优化可能需要更复杂的运行时分析。此外，为每个任务存储沿袭需要在GCS中实现垃圾收集策略以绑定存储成本，这是我们正在积极开发的功能。

容错。我们经常被问到，人工智能应用是否真的需要容错。毕竟，由于许多人工智能算法的统计特性，人们可以简单地忽略失败的推广。根据我们的经验，我们的答案是“是”。首先，忽略失败的能力使应用程序更容易编写和解释。第二，我们通过确定性重播实现的容错功能大大简化了调试，因为它允许我们轻松地重现大多数错误。这一点尤为重要，因为由于人工智能算法的随机性，其调试难度众所周知。第三，容错有助于节省资金，因为它允许我们使用便宜的资源，如AWS上的现货实例。当然，这是以一些日常开支为代价的。但是，对于我们的目标工作负载，我们发现这种开销是最小的。

GCS和水平可扩展性。GCS大大简化了Ray的开发和调试。它使我们能够在调试Ray本身时查询整个系统状态，而不必手动公开内部组件状态。此外，GCS也是时间线可视化工具的后端，用于应用程序级调试。

GCS也有助于Ray的水平扩展。在第5节中，我们可以通过在GCS成为瓶颈时添加更多碎片来进行扩展。GCS还通过简单地添加更多副本使全局调度程序能够进行扩展。由于这些优点，我们认为集中控制状态将是未来分布式系统的一个关键设计组件。

8 结论

如今，没有一个通用系统能够有效地支持训练、服务和模拟的紧密循环。为了表达这些核心构建块并满足新兴的人工智能应用程序的需求，Ray在单个动态任务图中统一了任务并行和参与者编程模型，并采用了由全局控制存储和自底向上分布式调度程序启用的可扩展架构。该体系结构同时实现的编程灵活性、高吞吐量和低延迟对于新兴的人工智能工作负载尤其重要，这些工作负载产生的任务在资源需求、持续时间和功能上各不相同。我们的评估显示，线性可扩展性高达每秒180万个任务，透明的容错性，以及对当前几个RL工作负载的实质性性能改进。因此，Ray为未来的人工智能应用程序的开发提供了灵活性、性能和易用性的强大组合。