Ray --内部运行机制、对象存储中对象的存储和容错

本篇主要介绍了分布式框架 Ray 的内部运行机制，对象的存储过程和容错能力。内部运行机制主要分为Ray的连接，远程函数的定义和使用以及获取对象ID的运行机制。对象存储主要是对象放入的序列化和从对象调出的反序列化，以及特殊的numpy数组的Apache Arrow 化存储。容错主要是进程和对象的恢复，丢失的actor不能重建，以及那些进程和对象不能重建问题。

1.内部运行机制

本节部分，将详细地跟踪在进行某些API调用时在系统级发生的情况。

1.1 连接到Ray

有两种方法可以启动Ray脚本。它可以以独立的方式运行，也可以连接到现有的Ray集群。

（1）独立的运行
Ray可以通过在脚本中调用ray .init()独立使用。当调用ray.init()时，将启动所有相关的进程。其中包括一个raylet、一个对象存储和管理器、一个Redis服务器和一些worker进程。
当脚本退出时，这些进程将被终止。
注意: 这种方法仅限于一台机器。

（2）连接到现有的Ray集群上

要连接到现有的Ray集群，只需将Redis服务器的参数地址作为redis_address = keyword参数传递给ray.init。在这种情况下(keyword 为集群头结点的IP地址)，调用ray.init时不会启动任何新进程，类似地，当脚本退出时，进程将继续运行。在这种情况下，除了与actor对应的worker之外的所有进程在不同的驱动程序进程之间共享。

1.2 定义远程函数

该系统的一个核心部件是集中控制平面。这是使用一个或多个Redis服务器实现的。数据在Redis的内存中以键值的形式存储。

使用的集中控制平面有两种方式。首先，作为系统控制状态的持久性存储。其次，作为进程之间通信的消息总线(使用Redis的发布-订阅功能)。

现在，考虑一个如下所示的远程函数定义。

@ray.remote
def f(x):return x + 1

当远程函数定义为如上所示时，将立即对该函数进行pickle，分配一个惟一的ID，并将其存储在Redis服务器中。可以在集中控制平面中查看此远程功能。

TODO: Fill this in.

每个工作进程都有一个单独的线程在后台运行，该线程侦听添加到集中控制状态的远程函数。当添加一个新的远程函数时，线程获取pickle的远程函数，unpickle它，然后可以执行该函数。

远程函数定义的注意事项

由于远程函数一旦定义就会立即导出，这意味着远程函数不能关闭在远程函数定义之后定义的变量。例如，下面的代码给出了一个错误。

@ray.remote
def f(x):return helper(x)def helper(x):return x + 1

当调用f.remote(),就会报一下错误：

Traceback (most recent call last):File "<ipython-input-3-12a5beeb2306>", line 3, in f
NameError: name 'helper' is not defined

另一方面，如果helper函数是在f之前定义的，那么它就可以工作。

1.3 调用远程函数

当驱动程序或worker调用远程函数时，会发生许多事情。

首先，创建一个task对象。任务对象包括以下内容。
- 被调用函数的ID。
- 函数参数的ID或值。像整数或短字符串这样的Python原语将被pickle并作为任务对象的一部分包含在内。通过对ray.put的内部调用，将更大或更复杂的对象放入对象库中，并将结果ID包含在任务对象中。直接作为参数传递的对象ID也包含在任务对象中。
- 任务的ID。这是从上面的内容中唯一生成的。（对应上边生成内容的id）。
- 任务返回值的id。这些都是由上面的内容惟一生成的。
然后，任务对象被发送到与驱动程序或worker程序相同节点上的raylet。
raylet决定是在本地调度任务，还是将任务传递给另一个raylet。
- 如果任务的所有对象依赖项都存在于本地对象存储中，并且有足够的CPU和GPU资源可用来执行任务，那么本地raylet将把任务分配给它的一个可用worker。
- 如果不满足这些条件，任务将被转发给另一个raylet。这是通过raylet之间的对等连接完成的。任务表可以检查如下。
```
TODO: Fill this in.
```
一旦一个任务被调度到一个raylet中，raylet就会对任务进行排队等待执行。当有足够的资源可用并且对象依赖项在本地可用时，按先进先出的顺序将任务分配给worker。
将任务分配给worker后，worker将执行该任务并将任务的返回值放入对象存储中。然后，对象存储将更新对象表，该对象表是集中控制状态的一部分，以反映它包含新创建的对象的事实。对象表可以如下查看。
```
TODO: Fill this in.
```
当将任务的返回值放入对象存储中时，首先使用Apache Arrow数据布局将它们序列化为一个连续的字节团，这有助于在使用共享内存的进程之间高效地共享数据。

调用远程函数的注意事项

当特定节点上的对象存储器填满时，它将以最近最少使用的方式开始驱逐（删除）对象。如果稍后需要的对象被驱逐，则对该对象的ray.get的调用将启动对象的重建。 raylet将尝试通过重放其任务沿袭来重建对象。

TODO: Limitations on reconstruction.

1.4 获取对象ID的值

当一个driver或worker在一个对象ID上调用ray.get时有一下事情发生。

ray.get(x_id)

driver或worker进入同一节点上的对象存储并请求相关对象。每个对象存储由两个组件组成，一个共享内存键值存储的不可变对象，以及一个管理器来协调节点之间的对象传输。

如果对象不存于在对象存储中，则管理器将检查对象表，以查看其他对象存储是否具有该对象。如果对象存在，它通过其管理器直接从其中一个对象库中请求对象。如果对象不存在，集中控制状态将在创建对象时通知请求管理器（请求管理器时请求这个对象的管理器）。如果对象不存在于任何地方，因为它已被从所有对象存储中删除，则worker还将请求从raylet中重构对象。这些检查会周期性地重复，直到对象在本地对象存储中可用为止，可能是通过重构或者是通过对象传输。
一旦对象在本地对象存储中可用，driver或worker将把内存的相关区域映射到它自己的地址空间(以避免复制对象)，并将字节反序列化到Python对象中。注意，作为对象一部分的任何numpy数组都不会被复制。

2. 对象存储中的序列化（Serialization in the Object Store）

本节主要介绍Python对象怎么通过序列化存储在ray的对象存储空间中的。一旦对象被放入对象存储，它就是不可变的。

Ray把对象放入对象存储中的几种情况：

远程函数的返回值。
调用ray.put(x)中的值x。
远程函数的参数(除了像int或float这样的简单参数)。

Python对象可以有任意数量的指针，指针的深度可以任意嵌套。要将对象放入对象存储区或在进程之间发送对象，必须首先将其转换为连续的字节字符串。这个过程称为序列化。将字节字符串转换回Python对象的过程称为反序列化。序列化和反序列化常常是分布式计算的瓶颈。

Pickle是Python中用于序列化和反序列化的库的一个例子。

Pickle(以及我们使用的变体cloudpickle)是通用的。它可以序列化各种各样的Python对象。然而，对于数值工作负载，pickle和反pickle可能效率不高。例如，如果多个进程希望访问一个由numpy数组组成的Python列表，则每个进程必须反序列化（unpickle ）该列表，并创建自己的新数组副本。即使所有进程都是只读的，并且可以很容易地共享内存，这也可能导致高内存开销。

在Ray中，我们使用Apache Arrow数据格式对numpy数组进行优化。当我们从对象存储区反序列化一个numpy数组列表时，我们仍然创建一个Python的numpy数组对象列表。但是，不是复制每个numpy数组，而是每个numpy数组对象持有一个指向共享内存中保存的相关数组的指针。这种序列化形式有一些优点。

反序列化可以非常快。
内存在进程之间共享，因此worker进程可以读取相同的数据，而不必复制它。

2.1 Ray可以处理的对象

Ray目前不支持任意Python对象的序列化。Ray可以使用Apache Arrow序列化的Python对象集包括以下内容。

基本类型:int、float、long、bools、string、unicode和numpy数组。
可以被Ray序列化的元素：列表、字典或元组。

对于更一般的对象，Ray将首先尝试通过将对象解包（unpacking）为其字段的字典来序列化该对象。这种行为并非在所有情况下都是正确的。如果Ray不能将对象序列化为其字段的字典，Ray将回到使用pickle。然而，使用pickle可能效率不高。

2.2 注意事项

根据Python语义，我们目前处理某些模式不正确。例如，包含相同列表的两个副本的列表将被序列化，就好像这两个列表是不同的一样。

    l1 = [0]l2 = [l1, l1]l3 = ray.get(ray.put(l2))l2[0] is l2[1]  # True.l3[0] is l3[1]  # False.

由于与上面的示例类似的原因，我们目前也不处理递归包含它们自己的对象(这在类似图形的数据结构中可能很常见)。

 l = []l.append(l)# Try to put this list that recursively contains itself in the object store.ray.put(l)

这将抛出一个异常，其消息如下所示。

This object exceeds the maximum recursion depth. It may contain itself recursively.

这种情况可以使用numpy数组进行处理。

2.3自定义序列化

编写定制的序列化和反序列化代码(例如，手工调用pickle)。

import pickle@ray.remote
def f(complicated_object):# Deserialize the object manually.obj = pickle.loads(complicated_object)return "Successfully passed {} into f.".format(obj)# Define a complicated object.
l = []
l.append(l)# Manually serialize the object and pass it in as a string.
ray.get(f.remote(pickle.dumps(l)))  # prints 'Successfully passed [[...]] into f.'

注：如果pckle不熟悉，也可以使用cloudpickle。

3. Ray 的容错

本节主要介绍Ray的故障处理。

3.1 机器和进程的故障

每个raylet(调度程序进程)都将进程状态信息发送到监视器进程。如果监视器在一段时间内(大约10秒)没有接收到来自给定raylet的任何进程状态信息，那么它将把该进程标记为死进程。

3.2 丢失对象

如果需要一个对象，但是丢失了或者从来没有创建过，那么创建该对象的任务将被重新执行以创建该对象。如果需要，为正在重新执行的任务创建输入参数所需的任务也将重新执行。这是Spark等其他系统使用的基于线性的标准容错策略。

3.3 Actors

当一个参与者（Actor）死亡时(要么是因为参与者进程崩溃，要么是因为参与者所在的节点死亡)，在默认情况下，任何试图从该参与者获取无法创建的对象的尝试都会引发异常。后续版本将包含一个自动重启参与者的选项。调用不存在的actor会引发异常。

3.4 Ray 容错限制

目前，Ray还没有能力处理所有的故障场景。正在努力解决这些已知的问题。

进程失败

Ray无法恢复的进程：Redis服务器进程和监视器进程。
如果驱动程序（driver）失败，该驱动程序将不会重新启动，作业也不会完成

丢失对象

如果一个对象在driver上是通过调用Ray.put构造的，然后被驱逐，并且稍后需要，Ray将不会重构这个对象。
如果一个对象是由一个actor方法构造的，然后被驱逐，并且稍后需要，Ray将不会重构这个对象。