内存不足预防#

如果应用任务或 actor 消耗大量堆空间，可能会导致节点内存不足 (OOM)。发生这种情况时，操作系统会开始杀死 worker 或 raylet 进程，从而中断应用。OOM 还可能导致指标停滞，如果发生在头部节点上，可能会导致 dashboard 或其他控制进程停滞，并导致集群不可用。

在本节中，我们将介绍

  • 内存监控器是什么以及它的工作原理

  • 如何启用和配置它

  • 如何使用内存监控器检测和解决内存问题

另请参阅 调试内存不足，了解如何排查内存不足问题。

内存监控器是什么？#

内存监控器是运行在每个节点 raylet 进程中的一个组件。它会定期检查内存使用情况，包括 worker 堆、对象存储和 raylet，如 内存管理 中所述。如果总使用量超过可配置的阈值，raylet 将杀死一个任务或 actor 进程以释放内存并防止 Ray 失败。

它可在 Linux 上使用，并在使用 cgroup v1/v2 的容器中运行的 Ray 上进行了测试。如果在容器外部运行内存监控器时遇到问题，请提交问题或提问。

如何禁用内存监控器？#

内存监控器默认启用，可以通过在 Ray 启动时将环境变量 RAY_memory_monitor_refresh_ms 设置为零来禁用（例如，RAY_memory_monitor_refresh_ms=0 ray start …）。

如何配置内存监控器？#

内存监控器由以下环境变量控制

  • RAY_memory_monitor_refresh_ms (int，默认为 250) 是检查内存使用并根据需要杀死任务或 actor 的间隔。当该值为 0 时，禁用任务杀死。内存监控器一次选择并杀死一个任务，并在选择下一个任务之前等待其被杀死，无论内存监控器运行频率如何。

  • RAY_memory_usage_threshold (float，默认为 0.95) 是节点超出内存容量的阈值。如果内存使用量超过此比例，它将开始杀死进程以释放内存。范围为 [0, 1]。

使用内存监控器#

重试策略#

当任务或 actor 被内存监控器杀死时，它将以指数退避方式重试。重试延迟有上限，为 60 秒。如果任务被内存监控器杀死，它会无限重试（不遵守 max_retries）。如果 actor 被内存监控器杀死，它不会无限期地重新创建 actor（它遵守 max_restarts，默认为 0）。

Worker 杀死策略#

内存监控器通过确保每个节点上的每个调用者至少有一个任务能够运行来避免任务无限重试的循环。如果无法保证这一点，工作负载将因 OOM 错误而失败。请注意，这仅是任务的问题，因为内存监控器不会无限期地重试 actor。如果工作负载失败，请参阅 如何解决内存问题，了解如何调整工作负载以使其通过。有关代码示例，请参见下面的最后一个任务示例。

当需要杀死 worker 时，策略首先优先杀死可重试的任务，即当 max_retries 或 max_restarts > 0 时。这样做是为了最大程度地减少工作负载失败。Actor 默认不可重试，因为 max_restarts 默认为 0。因此，默认情况下，在选择要先杀死哪个进程时，任务优先于 actor。

当有多个调用者创建了任务时，策略将从运行任务数量最多的调用者中选择一个任务。如果两个调用者拥有相同数量的任务，它将选择其最早任务启动时间较晚的调用者。这样做是为了确保公平性并允许每个调用者取得进展。

在共享相同调用者的任务中，最新启动的任务将首先被杀死。

下面是一个示例来演示该策略。在此示例中，我们有一个创建两个任务的脚本，这两个任务又分别创建了四个任务。任务按颜色着色，以便每种颜色形成一个任务“组”，它们属于同一个调用者。

Initial state of the task graph

如果此时节点内存不足，它将从运行任务数量最多的调用者中选择一个任务，并杀死其最后启动的任务

Initial state of the task graph

如果此时节点仍然内存不足，该过程将重复

Initial state of the task graph
示例：如果调用者的最后一个任务被杀死，工作负载会失败

让我们创建一个应用 oom.py，它运行一个任务，该任务需要的内存超过可用内存。通过将 max_retries 设置为 -1，将其设置为无限重试。

worker 杀死策略会看到它是调用者的最后一个任务，当它杀死该任务时，即使该任务被设置为永远重试，工作负载也会失败，因为它就是调用者的最后一个任务。

import ray

@ray.remote(max_retries=-1)
def leaks_memory():
    chunks = []
    bits_to_allocate = 8 * 100 * 1024 * 1024  # ~100 MiB
    while True:
        chunks.append([0] * bits_to_allocate)


try:
    ray.get(leaks_memory.remote())
except ray.exceptions.OutOfMemoryError as ex:
    print("task failed with OutOfMemoryError, which is expected")

设置 RAY_event_stats_print_interval_ms=1000，以便每秒打印 worker 杀死摘要，因为默认情况下是每分钟打印。

RAY_event_stats_print_interval_ms=1000 python oom.py

(raylet) node_manager.cc:3040: 1 Workers (tasks / actors) killed due to memory pressure (OOM), 0 Workers crashed due to other reasons at node (ID: 2c82620270df6b9dd7ae2791ef51ee4b5a9d5df9f795986c10dd219c, IP: 172.31.183.172) over the last time period. To see more information about the Workers killed on this node, use `ray logs raylet.out -ip 172.31.183.172`
(raylet)
(raylet) Refer to the documentation on how to address the out of memory issue: https://docs.rayai.org.cn/en/latest/ray-core/scheduling/ray-oom-prevention.html. Consider provisioning more memory on this node or reducing task parallelism by requesting more CPUs per task. To adjust the kill threshold, set the environment variable `RAY_memory_usage_threshold` when starting Ray. To disable worker killing, set the environment variable `RAY_memory_monitor_refresh_ms` to zero.
        task failed with OutOfMemoryError, which is expected
        Verify the task was indeed executed twice via ``task_oom_retry``:
示例：内存监控器优先杀死可重试的任务

让我们首先启动 ray 并指定内存阈值。

RAY_memory_usage_threshold=0.4 ray start --head

让我们创建一个应用 two_actors.py，它提交两个 actor，其中第一个是可重试的，第二个是不可重试的。

from math import ceil
import ray
from ray._private.utils import (
    get_system_memory,
)  # do not use outside of this example as these are private methods.
from ray._private.utils import (
    get_used_memory,
)  # do not use outside of this example as these are private methods.


# estimates the number of bytes to allocate to reach the desired memory usage percentage.
def get_additional_bytes_to_reach_memory_usage_pct(pct: float) -> int:
    used = get_used_memory()
    total = get_system_memory()
    bytes_needed = int(total * pct) - used
    assert (
        bytes_needed > 0
    ), "memory usage is already above the target. Increase the target percentage."
    return bytes_needed


@ray.remote
class MemoryHogger:
    def __init__(self):
        self.allocations = []

    def allocate(self, bytes_to_allocate: float) -> None:
        # divide by 8 as each element in the array occupies 8 bytes
        new_list = [0] * ceil(bytes_to_allocate / 8)
        self.allocations.append(new_list)


first_actor = MemoryHogger.options(
    max_restarts=1, max_task_retries=1, name="first_actor"
).remote()
second_actor = MemoryHogger.options(
    max_restarts=0, max_task_retries=0, name="second_actor"
).remote()

# each task requests 0.3 of the system memory when the memory threshold is 0.4.
allocate_bytes = get_additional_bytes_to_reach_memory_usage_pct(0.3)

first_actor_task = first_actor.allocate.remote(allocate_bytes)
second_actor_task = second_actor.allocate.remote(allocate_bytes)

error_thrown = False
try:
    ray.get(first_actor_task)
except ray.exceptions.OutOfMemoryError as ex:
    error_thrown = True
    print("First started actor, which is retriable, was killed by the memory monitor.")
assert error_thrown

ray.get(second_actor_task)
print("Second started actor, which is not-retriable, finished.")

运行该应用，可以看到只有第一个 actor 被杀死。

$ python two_actors.py

First started actor, which is retriable, was killed by the memory monitor.
Second started actor, which is not-retriable, finished.

解决内存问题#

当应用因 OOM 失败时，请考虑减少任务和 actor 的内存使用，增加节点的内存容量，或限制同时运行的任务数量。

问题或议题？#

您可以通过以下渠道发布问题、议题或反馈

  1. 讨论区：用于 Ray 使用问题或功能请求。

  2. GitHub Issues：用于bug 报告。

  3. Ray Slack：用于联系 Ray 维护者。

  4. StackOverflow：使用 [ray] 标签提问 Ray 相关问题。