对象

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

对象引用 本质上是一个指针或唯一ID，可以用来引用远程对象而无需查看其值。如果你熟悉期货，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

对象的闭包捕获

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

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 对象

• 序列化
• 对象溢出