杂项

本页将介绍 Ray 中的一些杂项主题。

动态远程参数

您可以使用 .options 在执行期间动态调整 ray.remote 的资源需求或返回值。

例如，在这里我们实例化了许多具有不同资源需求的相同 actor 的副本。请注意，要成功创建这些 actor，Ray 需要以足够的 CPU 资源和相关的自定义资源启动。

import ray

@ray.remote(num_cpus=4)
class Counter(object):
    def __init__(self):
        self.value = 0

    def increment(self):
        self.value += 1
        return self.value

a1 = Counter.options(num_cpus=1, resources={"Custom1": 1}).remote()
a2 = Counter.options(num_cpus=2, resources={"Custom2": 1}).remote()
a3 = Counter.options(num_cpus=3, resources={"Custom3": 1}).remote()

您可以为任务指定不同的资源需求（但不能为 actor 方法指定）

ray.init(num_cpus=1, num_gpus=1)

@ray.remote
def g():
    return ray.get_gpu_ids()

object_gpu_ids = g.remote()
assert ray.get(object_gpu_ids) == []

dynamic_object_gpu_ids = g.options(num_cpus=1, num_gpus=1).remote()
assert ray.get(dynamic_object_gpu_ids) == [0]

并更改任务（以及 actor 方法）的返回值数量

@ray.remote
def f(n):
    return list(range(n))

id1, id2 = f.options(num_returns=2).remote(2)
assert ray.get(id1) == 0
assert ray.get(id2) == 1

并在任务提交时为任务（以及 actor 方法）指定一个名称

import psutil

@ray.remote
def f(x):
   assert psutil.Process().cmdline()[0] == "ray::special_f"
   return x + 1

obj = f.options(name="special_f").remote(3)
assert ray.get(obj) == 4

此名称将显示在仪表板的机器视图中作为任务名称，在执行此任务时（如果是 Python 任务）将显示为工作进程名称，并在日志中显示为任务名称。

重载函数

Ray Java API 支持远程调用重载的 Java 函数。但是，由于 Java 编译器类型推断的限制，必须显式地将方法引用转换为正确的函数类型。例如，考虑以下内容。

重载普通任务调用

public static class MyRayApp {

  public static int overloadFunction() {
    return 1;
  }

  public static int overloadFunction(int x) {
    return x;
  }
}

// Invoke overloaded functions.
Assert.assertEquals((int) Ray.task((RayFunc0<Integer>) MyRayApp::overloadFunction).remote().get(), 1);
Assert.assertEquals((int) Ray.task((RayFunc1<Integer, Integer>) MyRayApp::overloadFunction, 2).remote().get(), 2);

重载 actor 任务调用

public static class Counter {
  protected int value = 0;

  public int increment() {
    this.value += 1;
    return this.value;
  }
}

public static class CounterOverloaded extends Counter {
  public int increment(int diff) {
    super.value += diff;
    return super.value;
  }

  public int increment(int diff1, int diff2) {
    super.value += diff1 + diff2;
    return super.value;
  }
}

ActorHandle<CounterOverloaded> a = Ray.actor(CounterOverloaded::new).remote();
// Call an overloaded actor method by super class method reference.
Assert.assertEquals((int) a.task(Counter::increment).remote().get(), 1);
// Call an overloaded actor method, cast method reference first.
a.task((RayFunc1<CounterOverloaded, Integer>) CounterOverloaded::increment).remote();
a.task((RayFunc2<CounterOverloaded, Integer, Integer>) CounterOverloaded::increment, 10).remote();
a.task((RayFunc3<CounterOverloaded, Integer, Integer, Integer>) CounterOverloaded::increment, 10, 10).remote();
Assert.assertEquals((int) a.task(Counter::increment).remote().get(), 33);

检查集群状态

基于 Ray 构建的应用程序通常需要有关集群的一些信息或诊断信息。一些常见的问题包括：

1. 我的自动伸缩集群中有多少节点？

2. 我的集群当前可用资源是多少，包括已用和总资源？

3. 我的集群中当前有哪些对象？

为此，您可以使用全局状态 API。

节点信息

要获取有关集群中当前节点的信息，您可以使用 ray.nodes()。

ray.nodes()

获取集群中节点列表（仅用于调试）。

返回：

集群中 Ray 客户端的信息。

DeveloperAPI: 此 API 在 Ray 的次要版本之间可能会发生变化。

import ray

ray.init()
print(ray.nodes())

  [{'NodeID': '2691a0c1aed6f45e262b2372baf58871734332d7',
    'Alive': True,
    'NodeManagerAddress': '192.168.1.82',
    'NodeManagerHostname': 'host-MBP.attlocal.net',
    'NodeManagerPort': 58472,
    'ObjectManagerPort': 52383,
    'ObjectStoreSocketName': '/tmp/ray/session_2020-08-04_11-00-17_114725_17883/sockets/plasma_store',
    'RayletSocketName': '/tmp/ray/session_2020-08-04_11-00-17_114725_17883/sockets/raylet',
    'MetricsExportPort': 64860,
    'alive': True,
    'Resources': {'CPU': 16.0, 'memory': 100.0, 'object_store_memory': 34.0, 'node:192.168.1.82': 1.0}}]

以上信息包括：

• NodeID：raylet 的唯一标识符。

• alive：节点是否仍然存活。

• NodeManagerAddress：raylet 所在的节点的私有 IP 地址。

• Resources：节点上的总资源容量。

• MetricsExportPort：通过 Prometheus 端点 导出指标的端口号。

资源信息

要获取有关集群当前总资源容量的信息，您可以使用 ray.cluster_resources()。

ray.cluster_resources()

获取当前的集群总资源。

请注意，随着节点的添加或移除，此信息可能会过时。

返回：

一个映射资源名称到集群中该资源的

总数量的字典。

DeveloperAPI: 此 API 在 Ray 的次要版本之间可能会发生变化。

要获取有关集群当前可用资源容量的信息，您可以使用 ray.available_resources()。

ray.available_resources()

获取当前的集群可用资源。

这与 cluster_resources 不同，因为它返回的是空闲（可用）资源而不是总资源。

请注意，随着任务的开始和完成，此信息可能会过时。

返回：

一个映射资源名称到集群中该资源的

集群中的资源。请注意，如果某个资源（例如，“CPU”）当前不可用（即数量为 0），则它不会包含在此字典中。

DeveloperAPI: 此 API 在 Ray 的次要版本之间可能会发生变化。

运行大型 Ray 集群

以下是一些运行拥有 1k+ 节点的 Ray 集群的技巧。当使用如此多节点运行 Ray 时，可能需要调整几个系统设置以实现如此多机器之间的通信。

调整操作系统设置

由于所有节点和工作进程都连接到 GCS，因此会创建许多网络连接，操作系统必须支持这些连接的数量。

最大打开文件数

操作系统必须配置为支持打开许多 TCP 连接，因为每个工作进程和 raylet 都连接到 GCS。在 POSIX 系统中，可以使用 ulimit -n 检查当前限制，如果限制很小，则应根据操作系统手册进行增加。

ARP 缓存

另一个需要配置的是 ARP 缓存。在一个大型集群中，所有工作节点都连接到 head 节点，这会向 ARP 表添加许多条目。确保 ARP 缓存大小足够大以处理这么多节点。否则，head 节点将挂起。发生这种情况时，dmesg 会显示类似 neighbor table overflow message 的错误。

在 Ubuntu 中，可以通过在 /etc/sysctl.conf 中增加 net.ipv4.neigh.default.gc_thresh1 - net.ipv4.neigh.default.gc_thresh3 的值来调整 ARP 缓存大小。有关更多详细信息，请参阅操作系统手册。

基准测试

机器设置

• 1 个 head 节点：m5.4xlarge（16 vCPU/64GB 内存）

• 2000 个工作节点：m5.large（2 vCPU/8GB 内存）

操作系统设置

• 将最大打开文件数设置为 1048576

• 增加 ARP 缓存大小

  • net.ipv4.neigh.default.gc_thresh1=2048

  • net.ipv4.neigh.default.gc_thresh2=4096

  • net.ipv4.neigh.default.gc_thresh3=8192

Ray 设置

• RAY_event_stats=false

测试工作负载

• 测试脚本：代码

基准测试结果

Actor 数量	Actor 启动时间	Actor 就绪时间	总时间
20k（每个节点 10 个 actor）	14.5 秒	136.1 秒	150.7 秒