序列化#

由于 Ray 进程不共享内存空间，因此工作进程和节点之间传输的数据需要被序列化和反序列化。Ray 使用 Plasma 对象存储来高效地在不同进程和节点之间传输对象。对象存储中的 Numpy 数组在同一节点上的工作进程之间共享（零拷贝反序列化）。

概述#

Ray 决定使用定制的 Pickle 协议版本 5 反向移植来替换原有的 PyArrow 序列化器。这消除了之前的一些限制（例如无法序列化递归对象）。

Ray 当前与 Pickle 协议版本 5 兼容，同时借助 cloudpickle 支持更广泛的对象序列化（例如 lambda & 嵌套函数，动态类）。

Plasma 对象存储#

Plasma 是一个内存对象存储。它最初是 Apache Arrow 的一部分。在 Ray 1.0.0 版本发布之前，Ray 将 Arrow 的 Plasma 代码分叉到 Ray 的代码库中，以便根据 Ray 的架构和性能需求进行解耦并继续开发。

Plasma 用于高效地在不同进程和节点之间传输对象。Plasma 对象存储中的所有对象都是不可变的并保存在共享内存中。这样可以使同一节点上的许多工作进程高效地访问它们。

每个节点都有自己的对象存储。当数据放入对象存储时，它不会自动广播到其他节点。数据会保留在写入者本地，直到被另一个节点上的任务或 Actor 请求。

序列化 ObjectRefs#

明确使用 ObjectRefs 使用 ray.cloudpickle 序列化 ObjectRefs 应作为最后的手段。推荐的方法是通过 Ray 任务参数和返回值传递 ObjectRefs。

Ray ObjectRefs 可以使用 ray.cloudpickle 进行序列化。然后可以使用 ray.get() 对 ObjectRef 进行反序列化并访问。请注意，必须使用 ray.cloudpickle；其他 pickle 工具不保证有效。此外，反序列化 ObjectRef 的进程必须是序列化它的同一个 Ray 集群的一部分。

序列化后，ObjectRef 的值将保留在 Ray 的共享内存对象存储中。必须通过调用 ray._private.internal_api.free(obj_ref) 来显式释放对象。

警告

ray._private.internal_api.free(obj_ref) 是一个私有 API，在未来的 Ray 版本中可能会改变。

此代码示例演示了如何序列化一个 ObjectRef，将其存储到外部存储，反序列化并使用它，最后释放其对象。

import ray
from ray import cloudpickle

FILE = "external_store.pickle"

ray.init()

my_dict = {"hello": "world"}

obj_ref = ray.put(my_dict)
with open(FILE, "wb+") as f:
    cloudpickle.dump(obj_ref, f)

# ObjectRef remains pinned in memory because
# it was serialized with ray.cloudpickle.
del obj_ref

with open(FILE, "rb") as f:
    new_obj_ref = cloudpickle.load(f)

# The deserialized ObjectRef works as expected.
assert ray.get(new_obj_ref) == my_dict

# Explicitly free the object.
ray._private.internal_api.free(new_obj_ref)

Numpy 数组#

Ray 通过使用带有带外数据的 Pickle 协议版本 5 对 numpy 数组进行优化。numpy 数组存储为只读对象，同一节点上的所有 Ray 工作进程都可以零拷贝读取对象存储中的 numpy 数组（零拷贝读取）。工作进程中的每个 numpy 数组对象都持有一个指向共享内存中相关数组的指针。对只读对象的任何写入都需要用户先将其复制到本地进程内存中。

提示

您通常可以通过仅使用原生类型（例如，numpy 数组或 numpy 数组及其他原生类型的列表/字典），或者通过使用 Actor 来持有无法序列化的对象，从而避免序列化问题。

修复“assignment destination is read-only”（赋值目标是只读的）#

由于 Ray 将 numpy 数组放入对象存储，当它们在远程函数中被反序列化为参数时，将成为只读的。例如，以下代码片段将会崩溃

import ray
import numpy as np


@ray.remote
def f(arr):
    # arr = arr.copy()  # Adding a copy will fix the error.
    arr[0] = 1


try:
    ray.get(f.remote(np.zeros(100)))
except ray.exceptions.RayTaskError as e:
    print(e)
# ray.exceptions.RayTaskError(ValueError): ray::f()
#   File "test.py", line 6, in f
#     arr[0] = 1
# ValueError: assignment destination is read-only

为了避免这个问题，如果需要修改数组，可以在目的地手动复制数组 (arr = arr.copy())。请注意，这实际上相当于禁用了 Ray 提供的零拷贝反序列化功能。

序列化注意事项#

  • Ray 当前使用 Pickle 协议版本 5。大多数 python 发行版使用的默认 pickle 协议是版本 3。对于较大的对象，协议 4 和 5 比协议 3 更高效。

  • 对于非原生对象，即使在一个对象中被多次引用，Ray 也只会保留一个副本

    import ray
import numpy as np

obj = [np.zeros(42)] * 99
l = ray.get(ray.put(obj))
assert l[0] is l[1]  # no problem!

  • 如果可能，请使用 numpy 数组或 numpy 数组的 Python 集合，以获得最佳性能。

  • Lock 对象大多是不可序列化的，因为复制一个锁没有意义，并且可能导致严重的并发问题。如果您的对象包含锁，您可能需要找到一个变通方法。

自定义序列化#

有时您可能希望自定义序列化过程，因为 Ray 使用的默认序列化器 (pickle5 + cloudpickle) 不适合您（无法序列化某些对象，对某些对象来说太慢等）。

至少有 3 种方法可以定义您的自定义序列化过程

  1. 如果您想自定义某种类型对象的序列化，并且可以访问代码，您可以在相应的类中定义 __reduce__ 函数。大多数 Python 库通常都这样做。示例代码

    import ray
import sqlite3

class DBConnection:
    def __init__(self, path):
        self.path = path
        self.conn = sqlite3.connect(path)

    # without '__reduce__', the instance is unserializable.
    def __reduce__(self):
        deserializer = DBConnection
        serialized_data = (self.path,)
        return deserializer, serialized_data

original = DBConnection("/tmp/db")
print(original.conn)

copied = ray.get(ray.put(original))
print(copied.conn)

<sqlite3.Connection object at ...>
<sqlite3.Connection object at ...>

  1. 如果您想自定义某种类型对象的序列化，但无法访问或修改相应的类，可以使用您使用的序列化器注册该类

    import ray
import threading

class A:
    def __init__(self, x):
        self.x = x
        self.lock = threading.Lock()  # could not be serialized!

try:
  ray.get(ray.put(A(1)))  # fail!
except TypeError:
  pass

def custom_serializer(a):
    return a.x

def custom_deserializer(b):
    return A(b)

# Register serializer and deserializer for class A:
ray.util.register_serializer(
  A, serializer=custom_serializer, deserializer=custom_deserializer)
ray.get(ray.put(A(1)))  # success!

# You can deregister the serializer at any time.
ray.util.deregister_serializer(A)
try:
  ray.get(ray.put(A(1)))  # fail!
except TypeError:
  pass

# Nothing happens when deregister an unavailable serializer.
ray.util.deregister_serializer(A)

    注意：序列化器在每个 Ray 工作进程中都是本地管理的。因此，对于每个 Ray 工作进程，如果您想使用该序列化器，需要注册该序列化器。取消注册序列化器也只在本地生效。

    如果您为一个类注册新的序列化器，新的序列化器将立即替换工作进程中的旧序列化器。此 API 也是幂等的，重复注册相同的序列化器不会产生副作用。

  2. 如果您想自定义特定对象的序列化，我们也为您提供了一个示例

    import threading

class A:
    def __init__(self, x):
        self.x = x
        self.lock = threading.Lock()  # could not serialize!

try:
   ray.get(ray.put(A(1)))  # fail!
except TypeError:
   pass

class SerializationHelperForA:
    """A helper class for serialization."""
    def __init__(self, a):
        self.a = a

    def __reduce__(self):
        return A, (self.a.x,)

ray.get(ray.put(SerializationHelperForA(A(1))))  # success!
# the serializer only works for a specific object, not all A
# instances, so we still expect failure here.
try:
   ray.get(ray.put(A(1)))  # still fail!
except TypeError:
   pass

故障排除#

使用 ray.util.inspect_serializability 来识别棘手的 pickling 问题。此函数可用于跟踪任何 Python 对象（无论是函数、类还是对象实例）中的潜在不可序列化对象。

下面，我们演示了对一个包含不可序列化对象（线程锁）的函数的这种行为

from ray.util import inspect_serializability
import threading

lock = threading.Lock()

def test():
    print(lock)

inspect_serializability(test, name="test")

结果输出为

  =============================================================
  Checking Serializability of <function test at 0x7ff130697e50>
  =============================================================
  !!! FAIL serialization: cannot pickle '_thread.lock' object
  Detected 1 global variables. Checking serializability...
      Serializing 'lock' <unlocked _thread.lock object at 0x7ff1306a9f30>...
      !!! FAIL serialization: cannot pickle '_thread.lock' object
      WARNING: Did not find non-serializable object in <unlocked _thread.lock object at 0x7ff1306a9f30>. This may be an oversight.
  =============================================================
  Variable:

      FailTuple(lock [obj=<unlocked _thread.lock object at 0x7ff1306a9f30>, parent=<function test at 0x7ff130697e50>])

  was found to be non-serializable. There may be multiple other undetected variables that were non-serializable.
  Consider either removing the instantiation/imports of these variables or moving the instantiation into the scope of the function/class.
  =============================================================
  Check https://docs.rayai.org.cn/en/master/ray-core/objects/serialization.html#troubleshooting for more information.
  If you have any suggestions on how to improve this error message, please reach out to the Ray developers on github.com/ray-project/ray/issues/
  =============================================================

要获取更详细的信息，请在导入 Ray 之前设置环境变量 RAY_PICKLE_VERBOSE_DEBUG='2'。这将启用基于 Python 后端的序列化而不是 C-Pickle，以便您可以在序列化过程中调试 Python 代码。然而，这将大大降低序列化速度。

已知问题#

在使用某些 python3.8 和 3.9 版本时，用户可能会遇到内存泄漏。这是由于 python pickle 模块中的一个错误。

Python 3.8.2rc1、Python 3.9.0 alpha 4 或更高版本已解决了这个问题。