对象

在 Ray 中，任务和 actor 创建并计算对象。我们将这些对象称为 远程对象，因为它们可以存储在 Ray 集群中的任何位置，并且我们使用 对象引用 来指代它们。远程对象缓存在 Ray 的分布式 共享内存 对象存储 中，集群中的每个节点都有一个对象存储。在集群设置中，一个远程对象可以驻留在一个或多个节点上，无论谁持有对象引用。

对象引用 本质上是一个指针或唯一的 ID，可用于指代远程对象而无需查看其值。如果你熟悉 futures，Ray 对象引用在概念上是相似的。

对象引用可以通过两种方式创建。

1. 它们由远程函数调用返回。

2. 它们由 ray.put() 返回。

Python

import ray

# Put an object in Ray's object store.
y = 1
object_ref = ray.put(y)

Java

// Put an object in Ray's object store.
int y = 1;
ObjectRef<Integer> objectRef = Ray.put(y);

C++

// Put an object in Ray's object store.
int y = 1;
ray::ObjectRef<int> object_ref = ray::Put(y);

注意

远程对象是不可变的。也就是说，它们的值在创建后无法更改。这使得远程对象可以在多个对象存储中复制，而无需同步副本。

获取对象数据

您可以使用 ray.get() 方法从对象引用中获取远程对象的结果。如果当前节点的对象存储不包含该对象，则会下载该对象。

Python

如果对象是 numpy 数组 或 numpy 数组的集合，get 调用是零拷贝的，并返回由共享对象存储内存支持的数组。否则，我们会将对象数据反序列化为一个 Python 对象。

import ray
import time

# Get the value of one object ref.
obj_ref = ray.put(1)
assert ray.get(obj_ref) == 1

# Get the values of multiple object refs in parallel.
assert ray.get([ray.put(i) for i in range(3)]) == [0, 1, 2]

# You can also set a timeout to return early from a ``get``
# that's blocking for too long.
from ray.exceptions import GetTimeoutError
# ``GetTimeoutError`` is a subclass of ``TimeoutError``.

@ray.remote
def long_running_function():
    time.sleep(8)

obj_ref = long_running_function.remote()
try:
    ray.get(obj_ref, timeout=4)
except GetTimeoutError:  # You can capture the standard "TimeoutError" instead
    print("`get` timed out.")

`get` timed out.

Java

// Get the value of one object ref.
ObjectRef<Integer> objRef = Ray.put(1);
Assert.assertTrue(objRef.get() == 1);
// You can also set a timeout(ms) to return early from a ``get`` that's blocking for too long.
Assert.assertTrue(objRef.get(1000) == 1);

// Get the values of multiple object refs in parallel.
List<ObjectRef<Integer>> objectRefs = new ArrayList<>();
for (int i = 0; i < 3; i++) {
  objectRefs.add(Ray.put(i));
}
List<Integer> results = Ray.get(objectRefs);
Assert.assertEquals(results, ImmutableList.of(0, 1, 2));

// Ray.get timeout example: Ray.get will throw an RayTimeoutException if time out.
public class MyRayApp {
  public static int slowFunction() throws InterruptedException {
    TimeUnit.SECONDS.sleep(10);
    return 1;
  }
}
Assert.assertThrows(RayTimeoutException.class,
  () -> Ray.get(Ray.task(MyRayApp::slowFunction).remote(), 3000));

C++

// Get the value of one object ref.
ray::ObjectRef<int> obj_ref = ray::Put(1);
assert(*obj_ref.Get() == 1);

// Get the values of multiple object refs in parallel.
std::vector<ray::ObjectRef<int>> obj_refs;
for (int i = 0; i < 3; i++) {
  obj_refs.emplace_back(ray::Put(i));
}
auto results = ray::Get(obj_refs);
assert(results.size() == 3);
assert(*results[0] == 0);
assert(*results[1] == 1);
assert(*results[2] == 2);

传递对象参数

Ray 对象引用可以在 Ray 应用中自由传递。这意味着它们可以作为参数传递给任务、actor 方法，甚至存储在其他对象中。对象通过 分布式引用计数 进行跟踪，一旦对对象的所有引用都被删除，其数据将自动释放。

可以将对象传递给 Ray 任务或方法，有两种不同的方式。根据对象传递的方式，Ray 将决定是否在任务执行之前 解引用 该对象。

将对象作为顶级参数传递：当对象直接作为顶级参数传递给任务时，Ray 将解引用该对象。这意味着 Ray 将获取所有顶级对象引用参数的基础数据，直到对象数据完全可用时才执行任务。

import ray


@ray.remote
def echo(a: int, b: int, c: int):
    """This function prints its input values to stdout."""
    print(a, b, c)


# Passing the literal values (1, 2, 3) to `echo`.
echo.remote(1, 2, 3)
# -> prints "1 2 3"

# Put the values (1, 2, 3) into Ray's object store.
a, b, c = ray.put(1), ray.put(2), ray.put(3)

# Passing an object as a top-level argument to `echo`. Ray will de-reference top-level
# arguments, so `echo` will see the literal values (1, 2, 3) in this case as well.
echo.remote(a, b, c)
# -> prints "1 2 3"

将对象作为嵌套参数传递：当对象作为嵌套对象的一部分传递时，例如在 Python 列表中，Ray 将 不会 解引用它。这意味着任务需要对引用调用 ray.get() 来获取具体的值。但是，如果任务从未调用 ray.get()，那么对象值就不需要传输到任务运行的机器上。我们建议尽可能将对象作为顶级参数传递，但嵌套参数对于将对象传递给其他任务而无需查看数据时很有用。

import ray


@ray.remote
def echo_and_get(x_list):  # List[ObjectRef]
    """This function prints its input values to stdout."""
    print("args:", x_list)
    print("values:", ray.get(x_list))


# Put the values (1, 2, 3) into Ray's object store.
a, b, c = ray.put(1), ray.put(2), ray.put(3)

# Passing an object as a nested argument to `echo_and_get`. Ray does not
# de-reference nested args, so `echo_and_get` sees the references.
echo_and_get.remote([a, b, c])
# -> prints args: [ObjectRef(...), ObjectRef(...), ObjectRef(...)]
#           values: [1, 2, 3]

顶级参数与非顶级参数的传递约定也适用于 actor 构造函数和 actor 方法调用

@ray.remote
class Actor:
  def __init__(self, arg):
    pass

  def method(self, arg):
    pass

obj = ray.put(2)

# Examples of passing objects to actor constructors.
actor_handle = Actor.remote(obj)  # by-value
actor_handle = Actor.remote([obj])  # by-reference

# Examples of passing objects to actor method calls.
actor_handle.method.remote(obj)  # by-value
actor_handle.method.remote([obj])  # by-reference

对象的闭包捕获

您还可以通过 闭包捕获 将对象传递给任务。当您有一个大型对象希望在许多任务或 actor 之间原样共享，并且不想重复作为参数传递时，这会很方便。但请注意，定义一个对对象引用进行闭包的任务将通过引用计数来固定对象，因此在作业完成之前，该对象不会被驱逐。

import ray

# Put the values (1, 2, 3) into Ray's object store.
a, b, c = ray.put(1), ray.put(2), ray.put(3)


@ray.remote
def print_via_capture():
    """This function prints the values of (a, b, c) to stdout."""
    print(ray.get([a, b, c]))


# Passing object references via closure-capture. Inside the `print_via_capture`
# function, the global object refs (a, b, c) can be retrieved and printed.
print_via_capture.remote()
# -> prints [1, 2, 3]

嵌套对象

Ray 还支持嵌套对象引用。这允许您构建复合对象，这些复合对象本身又持有指向进一步子对象的引用。

# Objects can be nested within each other. Ray will keep the inner object
# alive via reference counting until all outer object references are deleted.
object_ref_2 = ray.put([object_ref])

容错

Ray 可以通过 血缘关系重建 自动从对象数据丢失中恢复，但不能从 所有者 故障中恢复。有关更多详细信息，请参阅 Ray 容错。

更多关于 Ray 对象的信息

• 序列化
• 对象溢出