本章目标：把「Ray 跑得慢」这件事变成可测量、可归因、可修的工程问题。 读完你应该能做到三件事：知道开销花在哪四类地方、认出十个最常见的反模式、 用 State API 自己测出优化前后的差别。
前提：你已经写过真实能跑的 Ray 代码（第 4、5、9 章），并且知道 ObjectRef 是什么。
18.1 先量化，再优化：Ray 的四个开销来源
分布式的性能问题之所以难调，是因为开销的来源是分层的，而每一层的量级差了两个数量级 以上。先看这张表，它决定了你该往哪里看：
前两行的数字来自 Ray 官方文档的 Compiled Graph 页面
（ray-core/compiled-graph/ray-compiled-graph）：文档给出的对照是，经典 Ray Core API
调用一次远程方法的系统开销约 1ms，而 compiled graph 的 graph.execute() 路径
低于 50µs（Anyscale 博客 Announcing Compiled Graphs 的表述是 1–2ms 对约 50µs）。
这是官方口径，不是本书实测；你自己的机器上会不一样。
一个用来做心算的模型
总耗时 ≈ 提交开销 × 任务个数
+ 序列化开销 × 传给每个任务的参数字节数
+ 传输开销 × 跨节点搬运的字节数
+ 启动开销 × 冷启动的 worker 个数
+ 真正干活的 CPU/GPU 时间
这个模型最重要的推论是：前三项与"任务个数"成正比，最后一项与"真正的工作量"成正比。 当你把一个大任务切成 N 个小任务时，最后一项不变，前三项乘以 N。 所以「切分粒度」是调优的第一个旋钮，而它有一个明确的最优区间。
吞吐
▲
│ ┌───────────────┐
│ ╱ ╲
│ ╱ ╲ ← 切太细：调度/序列化开销吃掉收益
│ ╱ ╲
│ ╱ ← 切太粗：并行度不够，机器闲着
│ ╱
└───┴───────┴───────────────┴────────▶ 每任务计算量
<1ms ~10ms-1s >1min
(亏) (甜点区) (考虑再切)
经验规则：每个任务的纯计算时间在 10ms 到 1s 之间时，Ray 的开销占比通常是可接受的； 低于 1ms 就要认真考虑批量化或 compiled graph，高于几十秒则要考虑任务内部再做并行 （否则一个节点挂了就损失几十秒的计算）。
18.2 反模式目录
Ray 官方文档有一组 anti-pattern 页面（ray-core/patterns/anti-patterns，以及各条
ray-core/patterns/* 的 pattern 页）。本节按症状 → 原因 → 正确做法重写它们，
并补上几本实战中高频、但文档里没单独成页的坑。
反模式 1：任务太小（每任务 < 1ms 的计算）
症状：把 100 万个元素拆成 100 万个任务，跑完比单进程还慢；CPU 利用率上不去，
但 state.list_tasks() 里任务个数是百万级。
原因：每个任务要付约 1ms 的提交/调度成本（见 18.1）。100 万个任务 = 约 1000 秒的纯调度开销，而这些开销分布在 driver 与 raylet 上，很多时候是串行的。
正确做法：
- 批量化：一次处理一批而不是一个元素。这是 Ray Data map_batches 的核心理念
（map_batches 给你的是一个 batch，而不是一行）。
- 需要更细粒度、又确实是"反复执行同一张图"的场景（LLM 推理、RL rollout）， 用 compiled graph 把调度开销降到 <50µs。
# 反模式:一百万个 1 元素任务
refs = [process.remote(x) for x in items] # 100 万次提交
# 正确:切成 1000 个批次,每批 1000 个元素
def chunks(seq, size):
for i in range(0, len(seq), size):
yield seq[i:i + size]
refs = [process_batch.remote(batch) for batch in chunks(items, 1000)]
反模式 2：在任务里传大对象，而不用 ObjectRef
症状：提交任务本身就很慢；ray.put 一次之后再用却还是慢；集群网络流量异常高。
原因：f.remote(big_array) 中，big_array 会被按值序列化进任务描述。
虽然 Ray 对 numpy 数组有特殊处理（见 18.5），但任务描述会随任务一起被复制到
每个 worker，且不享受对象存储的引用计数、溢出、本地性优化。
正确做法：ray.put 一次，传 ref。
import numpy as np, ray
big = np.zeros((10_000, 10_000)) # 约 800MB
# 反模式:每个任务都把 800MB 序列化一遍
refs = [f.remote(big, i) for i in range(100)]
# 正确:放一次,之后传的都是 8 字节级的 ObjectRef
big_ref = ray.put(big)
refs = [f.remote(big_ref, i) for i in range(100)]
判定标准：如果你在提交任务时手动序列化出来的字节数 × 任务数超过了
数据本身的规模，就该用 ray.put。参数超过几 MB 就该考虑。
反模式 3：在 driver 里循环 ray.get 单个 ref（把并行吃成串行）
症状：任务明明都在并行跑（state 里 RUNNING 一大堆），但 driver 的墙钟时间
差不多等于每个任务耗时之和。
原因：ray.get 是阻塞的。循环里逐个 ray.get 会让 driver 变成串行执行器，
更糟的是——它还会改变调度行为：当 driver 阻塞在 ray.get(ref_i) 时，
后续任务的依赖关系无法被及时推进。
# 反模式:并行提交、串行收
refs = [f.remote(i) for i in range(1000)]
for ref in refs:
results.append(ray.get(ref)) # 第一个慢,后面全等着
# 正确:一次收完
results = ray.get(refs)
# 更好:需要边出结果边处理时,用 ray.wait 做流式
pending = list(refs)
while pending:
ready, pending = ray.wait(pending, num_returns=1)
results.append(ray.get(ready[0]))
反模式 4：闭包里捕获巨大对象
症状：代码看起来"只是引用了一个外部变量"，但每个任务的内存占用都多出几百 MB。
原因：cloudpickle 会把闭包里捕获的对象按值序列化进任务。也就是说：
# 反模式:每个任务都被塞了一份 lookup_table
lookup_table = load_huge_table() # 500MB
@ray.remote
def lookup(key):
return lookup_table[key] # 闭包捕获 → 每个 worker 一份 500MB
Ray 官方把它单独列为一条反模式（ray-core/patterns/closure-capture-large-objects）。
正确做法：
- 用 ray.put 把大对象放进对象存储，任务参数传 ref；
- 或者把它放到 actor 里当成员变量（见反模式 7），让一份副本长期驻留；
- 或者干脆在 worker 内部按需加载（配合 runtime_env 的 working_dir）。
怎么发现：ray memory 会把 CAPTURED_IN_OBJECT 这类引用类型列出来，
这就是"闭包/嵌套对象里夹带的 ObjectRef"的指纹。
反模式 5：在 actor 方法里 ray.get 另一个 actor
症状：跑着跑着全卡住，CPU 掉到 0，没有任何报错，超时后才各种失败。
原因：这是分布式系统里的经典死锁。actor A 的方法阻塞在等 actor B 的结果， 而 actor B 的方法又阻塞在等 A——两边的邮箱都被"正在执行的那个调用"占着， 谁都不肯让出执行权。默认并发度为 1 的同步 actor 尤其容易踩。
actor A (并发度 1) actor B (并发度 1)
method_a() ── ray.get ──▶ B.method_b()
B 里 ── ray.get ──▶ A.method_c()
A 的邮箱被 method_a 占着 ◀──── 永远排队,永远等不到
正确做法：
- 改成数据流：把依赖表达成 ObjectRef 的传递，让调度器去满足依赖， 而不是在 actor 内部同步等待；
- 确实需要互调时，把等待放到 driver 侧；
- 或者给 actor 设置 max_concurrency > 1（async actor 默认并发度是 1000，
sync actor 是 1）——这能缓解，但不能消除循环依赖，只是把死锁变成死锁得更慢。
反模式 6：无界提交（对象存储爆 / OOM）
症状：提交 100 万个任务后进程被 OOM killer 干掉；
或者 ObjectStoreFullError；或者 driver 内存持续上涨。
原因：每次 f.remote(...) 都会在 driver 侧留下一个 ObjectRef，
每个任务还要在 raylet 的队列里留一条记录。任务产生的结果对象会堆在对象存储里等
driver 来取。提交速度超过消费速度 = 无界队列 = 迟早爆。
正确做法：ray.wait 背压。这是 Ray 官方文档里的标准模式
（ray-core/patterns/limit-pending-tasks，doc code 在
doc/source/ray-core/doc_code/limit_pending_tasks.py）：
import ray
MAX_NUM_PENDING_TASKS = 100 # 在飞任务上限
result_refs = []
for i in range(NUM_TASKS):
if len(result_refs) > MAX_NUM_PENDING_TASKS:
# 至少回收一个完成的,保持队列有界
ready_refs, result_refs = ray.wait(result_refs, num_returns=1)
ray.get(ready_refs)
result_refs.append(worker.heavy_compute.remote(i))
ray.get(result_refs)
两个必须说清楚的细节：
- 不要写 RAY_max_pending_tasks。这个环境变量在 Ray 2.58 里不存在，
本书在 2.58 的配置清单里也没有找到它（很多中文博客里的这个名字来自很老的版本）。
官方给的方案是应用层背压，不是某个配置项。
- 官方文档明确提醒：这个模式的本意是限制在飞任务数，而不是"限制并发度"。
想限制并发度，应该改每个任务的资源声明（比如把 num_cpus 从 1 改成 2），
让调度器自己去算——用 ray.wait 限并发会损害调度性能。
反模式 7：每个任务都重新加载模型
症状：每个任务里 from_pretrained(...) 一次，GPU 利用率极低，
90% 时间花在加载权重上。
原因：任务（task）是无状态、用完即弃的执行单元。每个任务在自己的 worker 进程里跑，worker 可能被回收、可能换一个，模型只能重新加载。 一个 7B 模型加载一次是秒级到十几秒级，乘以任务数就是灾难。
正确做法：
- actor + 常驻模型：把模型放在 actor 的 __init__ 里，方法只做前向：
@ray.remote(num_gpus=1)
class Inferencer:
def __init__(self, model_path):
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
self.tok = AutoTokenizer.from_pretrained(model_path)
self.model = AutoModelForCausalLM.from_pretrained(model_path, device_map="cuda")
def generate(self, prompts):
... # 只做前向,不再加载
# 每张卡一个常驻 actor
engines = [Inferencer.remote(path) for _ in range(num_gpus)]
- ray.util.ActorPool：一组 actor + 自动背压的批处理，是"同构 actor 池"的标准工具。
mini-ray 里实现的是同一个 API（miniray/util/actor_pool.py），关键点也是背压：
每个 actor 手里最多一个任务，拿到结果再补下一个。
from ray.util.actor_pool import ActorPool
pool = ActorPool(engines)
for out in pool.map_unordered(lambda a, p: a.generate.remote(p), prompts):
... # 谁先算完先返回谁
反模式 8：需要状态却用任务（拿 ray.put 传状态）
症状：每轮迭代都要把整个状态 ray.put 一遍再传给下一个任务；
状态越大越慢，且每轮之间无法并行。
原因：任务是无状态的。用「任务 + 把状态当参数传」来模拟有状态计算， 等价于每轮做一次「全量序列化 + 全量传输」。这正是 MapReduce 式框架在迭代算法上的 老问题，也是 Ray 论文当年批评的对象。
正确做法：用 actor。状态留在 actor 的 Python 对象里，方法只传增量。
# 反模式:状态在数据面来回搬
@ray.remote
def step(state, batch):
return update(state, batch)
state = init()
for batch in batches:
state = ray.get(step.remote(state, batch)) # 每轮全量复制状态
# 正确:状态在 actor 里,只搬增量
@ray.remote
class Trainer:
def __init__(self):
self.state = init()
def step(self, batch):
self.state = update(self.state, batch) # 原地更新
return summarize(self.state)
trainer = Trainer.remote()
for batch in batches:
ray.get(trainer.step.remote(batch))
参数服务器（第 09 章 §9.7 模式一，mini-ray 侧对应 examples/10_parameter_server.py）、
RL 的 rollout worker、仿真的环境池，都是这个模式。
反模式 9：把 actor 当无状态函数用
症状：起了 1000 个 actor，每个只处理一个请求；集群资源被占满但吞吐没变；
ray status 里 GPU 全被"预留"但利用率很低。
原因：actor 的资源是终身持有的——从创建到销毁一直占着它声明的 CPU/GPU， 哪怕它 99% 的时间在邮箱里等消息。它的定位是"常驻的有状态组件"， 不是"能并发的函数"。把无状态函数写成 actor，等于用最贵的方式做最便宜的事。
正确做法：
- 无状态并行 → 用 task（@ray.remote 装饰函数）；
- 需要固定并发上限的 IO 密集场景 → 用 task + num_cpus 声明（甚至 num_cpus=0.1）；
- 确实需要常驻（模型、连接、缓存）→ 才用 actor，并且数量要按资源算： N 张卡就起 N 个 actor，而不是 N×10 个。
顺便记住一个默认值：actor 在调度时默认占 1 个逻辑 CPU，但在运行时
默认占 0 个 CPU。这是历史原因造成的（官方文档建议总是显式给 actor 写 num_cpus）。
推论是：如果你不写资源声明，可以起近乎无限个 actor，直到系统崩掉。
反模式 10：命名 actor 泄漏，忘了 ray.kill
症状：任务跑完了，集群里的 actor 还在；第二次跑同一个脚本时
ValueError: The actor with name=X already exists；GPU 一直被占着。
原因：lifetime="detached" 的命名 actor 会活过 driver。这是它的设计目的
（跨 job 复用），也是它的代价：没人回收它。普通 actor 会随 driver 退出而销毁，
但 detached actor 不会。
正确做法：
# 1) 用完显式杀
ray.kill(handle) # 默认 no_restart=True
# 2) 用 get_actor 做幂等获取,避免重复创建
try:
handle = ray.get_actor("serving")
except ValueError:
handle = Serving.options(name="serving", lifetime="detached").remote()
# 3) 定期盘点
from ray.util.state import list_actors
for a in list_actors(filters=[("state", "=", "ALIVE")]):
print(a.name, a.actor_id, a.num_restarts)
mini-ray 的对应实现（签名已对源码核对）：
miniray.get_actor(name, namespace=None)、
miniray.kill(actor, *, no_restart=True) 与 Ray 对齐；
⚠️ 但 miniray.list_actors() 不接受任何参数 —— 它没有 Ray 那边的
filters=[...]（docstring 只说"对齐 ray.util.list_actors 的返回形状"）。
所以上面那段 list_actors(filters=[("state", "=", "ALIVE")]) 是真 Ray 的写法，
搬到 mini-ray 上会 TypeError。
另外 detached actor 只在同一进程生命周期内有效（因为 mini-ray 没有独立的
raylet 进程，见 mini-ray/README.md 的「已知取舍」第 5 条）。
18.3 内存与 OOM
Ray 里"内存"至少有四个不同的池子，混着看必然调错：
对象存储的容量
- 默认占可用内存的 30%，上限 200GB，由常量
DEFAULT_OBJECT_STORE_MAX_MEMORY_BYTES 设定（ray_constants.py）。
要突破 200GB，需要在 Python 进程启动前设置 RAY_DEFAULT_OBJECT_STORE_MAX_MEMORY_BYTES（它是 import 时读的）。
- 环境变量名就是这个长的；网上常见的 DEFAULT_OBJECT_STORE_MAX_MEMORY_BYTES （去掉 RAY_ 前缀）是错的。
- 最小值 75MiB —— 这条是确定的（本书早期版本曾标为"未确认"，现已核实）：
ray/_private/ray_constants.py 里有 OBJECT_STORE_MINIMUM_MEMORY_BYTES = 75 * 1024 * 1024，
注释写明"所允许的对象存储内存下限，必须大于 MEMORY_RESOURCE_UNIT_BYTES"。
显式 object_store_memory= 一个更小的值会撞到这个下限。
- Linux 上对象存储落在 /dev/shm（共享内存）；macOS 上没有共享内存语义，
落到 /tmp（磁盘），性能明显更差。这是「为什么 Mac 上跑同样的代码更慢」的
最常见答案。而且 Mac 上不只是"慢一点"：常量 MAC_DEGRADED_PERF_MMAP_SIZE_LIMIT = 2 * 2**30 把容量主动压到 2GB
（超过 2GB 性能退化，见 issue #20388），所以 Mac 上的默认容量远小于"内存的 30%"。
- 当配置的对象存储容量超过 /dev/shm 的实际大小时，Ray 会直接报错：
ValueError: The configured object store size (10.24 GB) exceeds /dev/shm size (8.0 GB).
This will harm performance. Consider deleting files in /dev/shm or increasing its
size with --shm-size in Docker. To ignore this warning, set
RAY_OBJECT_STORE_ALLOW_SLOW_STORAGE=1.
能生效的变量名是 RAY_OBJECT_STORE_ALLOW_SLOW_STORAGE，不是
RAY_ALLOW_SLOW_STORAGE。⚠️ 但说"短名查无依据"不准确：短名确实出现在
ray_constants.py 紧跟 REQUIRE_SHM_SIZE_THRESHOLD 的注释里，
只是那行注释陈旧、services.py 读的是长名。用长名。 Docker/K8s 里对应的是
--shm-size / shm 卷大小。
- 关于"超过约 10GB 的对象默认报错"：不存在"单对象 10GB 上限"这条规则 ——
但 10GB 这个数字确实存在，它是
REQUIRE_SHM_SIZE_THRESHOLD = 10**10：当对象存储的容量超过 10GB、
而 /dev/shm 装不下时，Ray 默认拒绝启动（就是上面那条报错），
除非设 RAY_OBJECT_STORE_ALLOW_SLOW_STORAGE=1。
它与"单个对象多大"无关。真正会遇到的失败通常来自 ① 对象存储本身太小或塞满了
pinned 对象（ObjectStoreFullError，报错文本里会给出对象大小并提示用 ray memory 排查），
② 序列化路径上 PyArrow/Plasma 出现过 2GB 级别的 Maximum size exceeded (2GB) 报错（社区帖，未确认是否仍是 2.58 的普遍限制）。
对象溢出（spilling）
对象存储满了不是立刻失败，而是先把对象溢出到磁盘。相关开关（来源：
官方 Object Spilling 文档 ray-core/objects/object-spilling 与
ray_config_def.h）：
这些是系统配置，通过 ray.init(_system_config={...}) 或
ray start --system-config='{...}' 传入，也可以写成 RAY_<name> 环境变量
（例如 RAY_object_spilling_threshold=<你确认过的取值>）。
溢出的代价：溢出后的对象再次被 ray.get 时会从磁盘读回来，
延迟从微秒级变成毫秒级甚至更高。把溢出当成安全网，不要当成容量方案。
排查内存问题的正确姿势
# 1) 谁在占对象存储:按对象大小排序,看 call site
ray memory --sort-by=OBJECT_SIZE
# 2) 按调用栈聚合,找"哪一行代码泄漏"
ray memory --group-by=STACK_TRACE
# 3) 只看对象/任务/actor 的汇总(等价物是 State API)
ray summary objects
ray summary tasks
ray memory 的五个引用类型是排查的关键，它们分别代表对象被"谁"钉住：
注意：官方文档里明确写到的是 --sort-by 与 --group-by 两个选项。
ray memory --stats-only（只打印对象存储的聚合统计、不逐条列对象）
是真实存在的 flag（本书早期版本曾标为"未确认"，现已核实：
ray/scripts/scripts.py 里定义为 is_flag=True, help="Display plasma store stats only."）。
新版 Ray 里 State API 的 ray summary objects 是更"结构化"的替代品。
另外两个实战技巧：
- Ray 的 memory monitor 会先动手。Ray 2.2+ 有应用级内存监控，
在 Linux OOM killer 之前杀掉 worker，异常类型是
ray.exceptions.OutOfMemoryError（定义在 python/ray/exceptions.py）。
⚠️ 文案在 2.55 改过，别照抄老博客去搜（本轮核对结果）：
也就是 2.55 起从"Task"改成了带计数的"worker(s)"，而且后面还拼了
kill details 与建议（src/ray/raylet/node_manager.cc 里 worker_exit_message 那段）。如果你按旧文案 grep 日志，在 2.55+ 上一行都搜不到 ——
正确的抓法是搜 were killed due to the node running low on memory。
这条消息意味着"某个 worker 的堆内存涨太快"，而不是对象存储满了。
它同时会以 Out of Memory 事件推给 dashboard（RAY_EVENT_EVERY_MS(ERROR, ...)）。
- --num-cpus=0 的 head 节点是官方推荐的做法（KubeRay 里给 head pod 设 num-cpus: "0"），目的就是别让 worker 跑到 head 上跟 GCS 抢内存。
18.4 调度与并发调优
Ray 的默认调度策略叫 hybrid（混合）：先按资源可用量挑出一小撮候选节点，
再在候选里随机选一个。三个参数决定它的行为（来源：ray_config_def.h，
defaults 已由官方文档 PR #65264 "Add a scheduling overview with defaults" 对源核对过）：
候选集大小 k = max(节点数 × scheduler_top_k_fraction, scheduler_top_k_absolute)。
k 的意义：k 越大越倾向于"全局最优但慢"，k 越小越倾向于"随机但快"。
超大集群（几千节点）里调小 scheduler_top_k_fraction 能显著降低 GCS 上的调度压力。
写成环境变量是 RAY_scheduler_spread_threshold / RAY_scheduler_top_k_fraction
这种形式（RAY_ 前缀 + 配置名）。
0-CPU 任务的语义
两个必须知道的副作用：
- Ray 会用 num_cpus 设置 OMP_NUM_THREADS：给了 num_cpus=2 意味着 OMP_NUM_THREADS=2；不给则设为 1（避免多 worker 的线程争抢，见 issue #6998）。
所以如果你的 numpy/MKL 代码莫名只用 1 个线程，就是这里。
- 0 CPU 的任务没有真正的并发上限。论坛里的原话是"0 CPU 的话你可以起无限个 actor
直到把系统搞崩"。要限流就显式给小数资源，或者用 18.2 的 ray.wait 背压。
worker 复用与空闲回收
让 worker 活着比反复拉起便
本文由 AI图鉴自动抓取并翻译，版权归原作者所有；如需查看原文请点下方链接。
The goal of this chapter is to turn "Ray slow" into a measurable, imputable, repairable engineering problem. You should be able to read three things: know where the spending is in four places, recognize ten of the most common reverse patterns, and use State API to detect the difference between optimization and optimization.
Precedent: You've written Ray code (chapters 4, 5, 9) and you know what ObjectRef is.
18.1 First quantization, then optimize: the four sources of Ray's expenses
The problem of distributed performance is difficult to reconcile because the source of the expenses is stratified, and each layer has two orders of magnitude and above. Look at this watch. It determines where you're going:
The numbers for the first two lines are from the Compiled Graph page of the Ray official document
(ray-core/compiled-graph/ray-compiled-graph): The contrast given in the document is the classic Ray Core API
The system cost of calling a remote method is about 1 ms, and the graph.execute() path of the coded graph
Less than 50 meters (Annyscale blog Announcing Compiled Graphs says 1-2 ms is about 50 ms).
It's an official caliber, not a book; your own machine will be different.
It's a model for mind calculations.
Total time-consuming ≈ Submit cost x number of tasks
+ Sequenced cost x Number of byte parameters transmitted to each task
+ Transfer cost x Number of bytees moved across nodes
+ Start the cost x the number of workkers initiated by cold
+ really working CPU/GPU time
The most important corollary of the model is that the first three are proportional to the number of tasks and the last to the "real workload". When you cut a big task into a small one, the last one is the same, the first three times the N. So "slice-spectrum" is the first dome to be fine-tuned, and it has a clear and optimal range.
Swallow.
▲
│ ┌───────────────┐
│ ╱ ╲
│ ╱ ← 切 Quantities too small: control/sequencing costs to eat
│ ╱ ╲
Sliced too thick: not enough parallels, machines idle
│ ╱
└───┴───────┴───────────────┴────────▶ 每任务计算量
<1ms ~10ms-1s >1min
(deficit)
Empirical rule: Ray's share of expenses is generally acceptable when the net calculation time for each task is between 10 ms and 1 s;
18.2 Anti-Model Directory
Ray official documents have a set of anti-pattern pages (ray-core/patters/anti-patters, and articles)
ra ra ra ra )。 )。 This section rewrites them according to symptoms and causes and correct practices.
A few live high frequency pits were added, but there were no separate pages in the file.
Counter mode 1: Too small (calculation < 1ms per task)
Symptoms: Dismantling 1 million elements into 1 million missions is slower than a single process; CPU is not working.
But the number of jobs in state.list tasks() is a million.
Reason: Approximately 1 ms of submission/dispatch costs per mission (see 18.1). One million missions = about 1,000 seconds of pure dispatch expenses, which are distributed on driver and Raylet, often in parallel.
Correct practice:
- Batch of quantization: Processing a batch at a time instead of an element. This is the core idea of Ray Data Map Batches.
(map batches gives you a watch, not a line).
- Need for more finer, and indeed "repeat the same image" (LLLM reasoning, RL rollout), to reduce dispatch costs to < 50 ms by using copiled graph.
# Counter mode: 1 million 1 element jobs
# 1 million submissions
# Correct: cut to 1000 batches with 1,000 elements per batch
def chunks(seq, size):
for i in range(0, len(seq), size):
yield seq[i:i + size]
refs = [process_batch.remote(batch) for batch in chunks(items, 1000)]
Counter mode 2: Spread objects in the task without using ObjectiveRef
Symptoms: The submission of the task itself is slow; Ray.put is slow to use again after once; cluster network traffic is exceptionally high.
Reason: In f.remote (big array), big array will be sequenced into task descriptions by value.
Although Ray has a special treatment for Numpy arrays (see 18.5), the description of the mission will be copied along with the task.
Each worker does not enjoy the number of references stored by the object, spills, local optimization.
Correct: Ray.put once, pass ref.
import numpy as np, ray
# About 800MB
# Counter-mode: every mission sequences 800MB
refs = [f.remote(big, i) for i in range(100)]
# Correct: Play it once, and then it's all 8Z byte-grade Objectref
big_ref = ray.put(big)
refs = [f.remote(big_ref, i) for i in range(100)]
Criteria for determination: If you manually sequenced the number of byte tasks when submitting the task, the number of tasks x exceeds
The size of the data itself should be used. Parameters exceeding a few MBs should be considered.
Inverse mode 3: cycle Ray.get individually ref in driver
Symptoms: While missions run side by side (RUNNG piles in state), driver's wall clock time
Almost equals the sum of each task.
Reason: Ray.get is blocked. One by one in the cycle, Ray.get turns driver into a serial implementer.
Even worse -- it also changes movement behaviour: when driver is stuck in Ray.get (ref i), it’s not the only thing that makes sense.
Dependency relationships for follow-up missions cannot be advanced in a timely manner.
# Counter-module: parallel submission, serial receipt
refs = [f.remote(i) for i in range(1000)]
for ref in refs:
# First slow, all wait
# That's right: one round
results = ray.get(refs)
# Better: use ray.wait to flow when you need to process the results
pending = list(refs)
while pending:
ready, pending = ray.wait(pending, num_returns=1)
results.append(ray.get(ready[0]))
Counter mode 4: Capture a huge object in a closed package
Symptoms: The code looks like "just a reference to an external variable," but there are hundreds more MBs per mission.
Reason: cloudpickle will sequence objects captured in a closed bag into a task. That means:
# Counter-mode: each mission is stuck with a lookup table
lookup_table = load_huge_table() # 500MB
@ray.remote
def lookup(key):
Turn lookup table[key]
Ray Official singled it out as a counter-model (ray-core/patterns/closure-capture-large-objects).
Correct practice:
- Put large objects in object storage with Ray.put and pass task parameters;
- Or put it in an actual member variable (see reverse mode 7) to keep a copy permanently;
- Or simply load on demand inside the worker.
How did you find out that Ray memory would list CAPTURED IN OBJECT such a type of reference?
This is the fingerprints of the "Objectref" taped in the "closed/destination object".
Counter mode 5: in actor method Ray.get another actor
Symptoms: running all stuck, CPU fell to 0, no mistakes, and timeout failed.
Reason: This is the classic dead lock in the distribution system. The actor A method is blocking the outcome of the actor B, and the actor B method is blocking the A -- the mailbox on both sides is occupied by "the call that is being made," and no one will give up enforcement. The default simultaneous actor at 1 is particularly easy to step on.
Actor A (System 1) Actor B (System 1)
method_a() ── ray.get ──▶ B.method_b()
Bri - Ray.get - A. Method c()
A-boxes are being held by a method a--- never waiting in line.
Correct practice:
- Replaced with data stream: achieve the transfer of the relying table to the objectref and allow the dispatcher to satisfy the dependency, instead of synchronizing within the actual;
- When there is a genuine need for interchange, place the wait on the side of the driver;
- Or set max concurrence > 1 (async actor default parallel of 1000,
Sync actor is 1) - It eases, but it doesn't eliminate cycling dependence, it just turns dead locks into dead locks.
Counter mode 6: Unlimited submission (object storage explosion / OOM)
Symptoms: 1 million post-mission processes were eliminated by OOM killer;
Or objectStoreFullError; or driver memory continues to rise.
Reason: Every time f.remote(...) leaves an Objectref on the side of driver,
Each mission also keeps a record in the line of Raylet. The result objects generated by the task are stacked in the object's storage.
Driver to pick up. Submission faster than consumption = borderless queue = sooner or later explosion.
Correct: Ray.wait back. This is the standard pattern in Ray's official document.
(ray-core/papers/lib-ping-tasks, doc code)
doc/source/ray-core/doc_code/limit_pending_tasks.py）：
import ray
MAX NUM PENDING TASKS = 100 # In flight ceiling
result_refs = []
for i in range(NUM_TASKS):
if len(result_refs) > MAX_NUM_PENDING_TASKS:
# Recover at least one completed, keep the line open
ready_refs, result_refs = ray.wait(result_refs, num_returns=1)
ray.get(ready_refs)
result_refs.append(worker.heavy_compute.remote(i))
ray.get(result_refs)
Two details must be made clear:
- Don't write ray max pending tasks. This environment variable does not exist in Ray 2.58.
The book was not found in the configuration list of 2.58 (the name in many Chinese blogs comes from a very old version).
The official offer is applied layer back pressure, not a configuration item.
- The official document clearly reminds us that this pattern is intended to be limited to the number of flying tasks, not to "limit and hair".
If you want to limit the frequency, you should change the resource statement for each mission (e. g. change num cpus from 1 to 2),
Let the scheduler calculate itself - use the Ray.wait limit and collide to compromise movement performance.
Counter mode 7: Reload the model for each task
Symptoms: once in each mission from pretrained(...) GPU is extremely low.
90% of the time spent on load weight.
Reason: Task (task) is a non-state, obsolete and abandoned implementation module. Each task runs in its own workingrer process, which can be recycled, possibly replaced by one, and the model can only reload. A 7B model is loaded once in seconds to a dozen seconds, multiplied by the number of tasks is disaster.
Correct practice:
- Actor + Permanent Model: Place the model in the init of the actor and do only forward:
@ray.remote(num_gpus=1)
class Inferencer:
def __init__(self, model_path):
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
self.tok = AutoTokenizer.from_pretrained(model_path)
self.model = AutoModelForCausalLM.from_pretrained(model_path, device_map="cuda")
def generate(self, prompts):
# Just do it forward, no more loading
# Every card is permanent
engines = [Inferencer.remote(path) for _ in range(num_gpus)]
- Ray.util.ActorPool: a set of actor + automatic backpressed batch processing is a standard tool for "coconstructor pool".
The key point is the back pressure:
Each actor has up to one task and gets the results and next.
from ray.util.actor_pool import ActorPool
pool = ActorPool(engines)
for out in pool.map_unordered(lambda a, p: a.generate.remote(p), prompts):
# Who's going back first?
Counter mode 8: Need status but task (take Ray.put pass)
Symptoms: The whole state of Ray.put is transmitted over and over again to the next mission;
The larger the situation, the slower it is, and it is impossible to run between rounds.
Reason: The mission is non-existent. Simulates the status calculation by "task + pass state as parameters" for an equal value of one full serialization + full transfer per round. This is the old problem of the MapReduce framework's iterative algorithm, and it was the subject of criticism in Ray's paper.
Good practice: use an actual. The status remains in the Python object of the actor, the method being the increment only.
# Reverse mode: status moving back and forth on the data
@ray.remote
def step(state, batch):
return update(state, batch)
state = init()
for batch in batches:
# Full copy status per round
# Correct: status in actor, only incremental
@ray.remote
class Trainer:
def __init__(self):
self.state = init()
def step(self, batch):
# Update in situ
return summarize(self.state)
trainer = Trainer.remote()
for batch in batches:
ray.get(trainer.step.remote(batch))
Parameter Server (Chapter 09 § 9.7 Mode I, xamples/10 parameter server.py),
RL's rollout worker, the real environment pool, is this pattern.
Inverse mode 9: Use an actor as no function
Symptoms: 1,000 inactors, each processing only one request; cluster resources are occupied but the same;
All GPUs in Ray status are "set aside" but have low utilization.
Reason: Actor's resources are held for life - the CPU/GPU that it has claimed from creation to destruction, even though it's 99% of the time in the mailbox. It's positioned as a "resident state component", not a "composite function". Writes the no-state function as an actual, equals doing the cheapest thing in the most expensive way.
Correct practice:
- No state of parallel task (@ray.remote decoration function);
- The IO density scenario needs to be fixed and capped → with a task + num cpus declaration (even num cpus=0.1);
- There is a real need for a permanent presence (models, connections, cache) in order to use an inactor, and the number is based on resources: N cards start N, not N x 10.
Also remember a default value: actor defaults to 1 logical CPU during schedule, but while running
Default is 0 CPUs. This is the result of historical reasons (official documents suggest that it is always visible to an actor to write num cpus).
The inference is that if you don't write a resource statement, you can start at almost unlimited anor until the system collapses.
Counter mode 10: Name actor leak, forget Ray.kill
Symptoms: Mission's over, and the team's actualor is still there; the second one runs the same script.
ValueError: The actor with name=X already exists; GPU has been occupied.
Reason: Livetime = "detached" an actor will survive. That's what it was designed for.
It's the price of it: no one recovers it. An ordinary actor will destroy it with the exit of the driver.
But detached actor won't.
Correct practice:
# 1) Exhaustion Kill!
# Default no restart=True
#2) Access to get actor etc. to avoid creating
try:
handle = ray.get_actor("serving")
except ValueError:
handle = Serving.options(name="serving", lifetime="detached").remote()
#3) Periodic inventory
from ray.util.state import list_actors
for a in list_actors(filters=[("state", "=", "ALIVE")]):
print(a.name, a.actor_id, a.num_restarts)
Mini-ray matching (signature checked for source code):
miniray.get_actor(name, namespace=None)、
Miniray.kill (actor, *, no restart=True) aligned with Ray;
⚠️ But fewray.list actors() does not accept any parameters -- it doesn't have Ray's.
= [...]).
So the list actors above is true Ray,
Move to Mini-ray to meet TypeError.
And detached actor only works during the life cycle of the same process.
For the Raylet process see Article 5 of the known trade-off of Mini-ray/README.md).
Memory and OOM
Ray's memory has at least four different pools.
The capacity of objects to store
- Default 30% of available memory, cap 200GB, constant
DEFAULT OBJECT STORE MAX MAMORY BYTES set (ray constants.py).
To break 200GB, you need to set up RAY DEFAUT OBJECT STORE MAX MEMORY BYTES before the Python process starts.
- The environmental variable is this long; the common web-based DEFAULT OBJECT STORE MAX EMORY BYTES is wrong.
- Minimal value 75MiB - This is certain (the earlier version of the book was marked "unrecognised" and has been verified):
EY/ private/ray constants.py contains OBJECT STORE MINIMUM MEMORY BYTS = 75 * 1024 * 1024
The note states that "the allowed lower limit for the storage of objects shall be greater than the MEMORY RESOURCE UNT BYTES".
Visible object store memoory = a smaller value will hit this lower limit.
- Linux objects stored on /dev/shm (shared memory); macOS does not share memory syntax,
/tmp (disk), which is significantly worse. That's "Why Mac is running the same code slower"?
The most common answer. And Mac is not just "slower": constant MAC DEMRAD PERF MMAP SIZE LIMIT = 2* 2**30, pressing the capacity to 2GB
(over 2GB performance degraded, see issue #20388), so the default capacity on Mac is much smaller than "30% memory".
- Ray will report the error directly when the configured object storage capacity exceeds the actual hours of /dev/shm:
ValueError: The configured object store size (10.24 GB) exceeds /dev/shm size (8.0 GB).
This will harm performance. Consider deleting files in /dev/shm or increasing its
size with --shm-size in Docker. To ignore this warning, set
RAY_OBJECT_STORE_ALLOW_SLOW_STORAGE=1.
The valid variable is RAY OBJECT STORE ALLOW SLOW STORAGE, no.
RAY ALLOW SLOW STORAGE. But it's not accurate to say that short names are unsupported: short names do appear.
ra Q Q Q,,,,,,,,,,,,,,,,,,,,,,,,,,,,,,,,,,,,,,,,,
It's just that the old note, servicees.py reads a long name. With a long name. Docker/K8s.
--shm-size / shm volume size.
- On the rule "Default by default of objects exceeding approximately 10GB": there is no "one object 10GB cap" -
But the 10GB figure does exist, it is.
REQUIRE SHM SIZE THRESHOLD = 10**10: when the object has a storage capacity exceeding 10GB,
And when /dev/shm failed, Ray defaulted to start (that's the error above),
Unless you set up RAY OBJECT STORE ALLOW SLOW STORAGE=1.
It has nothing to do with "the size of a single object." The real failure usually comes from the fact that object storage is too small or stuffed.
Pinned object (ObjectStoreFullError, which gives the size of the object in the wrong text and suggests that it be sorted out with a Ray memory),
2 There was a 2GB-grade Maximu size error (2GB) in PyArrow/Plasma on the serialization path (community post, unconfirmed as still a general restriction of 2.58).
Object Spilling
The object is stored not as an immediate failure, but as a first spill to disk. Related switches (source:
Official Object Spiling Document Ray-core/Objects/Object-Schilling
ray_config_def.h）：
These are system configurations, by Ray.init system config=...} or
Ray start --system-config='...}'into, or can be written as RAY name> environment variable
(e.g. RAY object spilling threshold= <you confirmed > )
The cost of spilling: the object after spilling is read back on disk once again by Ray.get.
Delays change from microsecond to millisecond or even higher. Use spills as safety nets and not as capacity options.
Checking memory problem right.
#1) Who stores the object: Sort by object size, see call site
ray memory --sort-by=OBJECT_SIZE
#2) Press the indentation, find "what line leaks."
ray memory --group-by=STACK_TRACE
#3) read only the aggregation of objects/tasks/actor (Equivalent is State API)
ray summary objects
ray summary tasks
The five types of references to Ray memory are key to the search, and they represent "who" nailed the object:
Note: Both options --sort-by and --group-by are clearly stated in the official document.
Ray memory --stats-only (printing only the aggregate statistics stored by objects, without entries)
It's a real flag. (The early version of the book, which was marked as "unconfirmed", has been verified:
Ray/scripts/scripts.py is flag=True, help= "Display plasma store positions only.")
The new version of Ray's State API's Ray Summary objects is a more "structured" alternative.
Two other skills:
- Ray's memory monitor will do it first. Ray 2.2+ has application level memory monitoring,
Kill the worker before Linux OOM killer.
ra.exceptions.OutOfMemoryError (defined in python/ray/exceptions.py).
The file was modified at 2.55. Do not copy the old blog and search it.
That's 2.55 from "Task" to "worker" with numbers.
Kill articles and recommendations (src/ray/raylet/node manager.cc) If you follow the old grep log, you can't find it in the previous line of 2.55+...
were killed due to the Node running low on memory.
This message means "some worker's memory goes up too fast", not the object's full.
It also pushes to dashboard (RAY EVENT Every MS (ERROR,...)).
- --num-cpus=0 head node is officially recommended (KubeRay gives head pud num-cpus: "0"), the purpose of which is not to let worker run to head with GCS.
18.4 Movement control and co-ordination
Ray's default dispatch strategy is called hybrid (mix): First select a small number of candidate points by resource availability.
One more random candidate. Three parameters determine its behaviour (source: Ray config def.h).
Reforms have been checked by the official document PR #65264 "Add a scheduling overview with data":
The candidate size k = max (node x scheduler top k section, scheduler top k absolute).
Meaning: k, the bigger the k, the better but the slower it is, the smaller the k, the more random but the faster.
Smallening the scheduler top k section in a mega-cluster (thousands of points) can significantly reduce the movement pressure on GCS.
The environment variable is RAY scheduler spread threshold / RAY scheduler top k section
This form (RAY prefix + configuration name).
Semantics for 0-CPU tasks
Two side effects that must be known:
- Ray will set OMP NUM THREADS with num cpus=2 which means OMP NUM THREADS=2; or set it to 1 (to avoid multiple worker lines, see issue #6998).
So if your numpy/MKL code is anonymously one thread, here it is.
- 0 CPU missions have no real simultaneous caps. The word in the forum is "0 CPU" and you can start an infinite number.
Until the system is destroyed." If rate limit is required, it will give visible resources to decimals or use 18.2's ray.wait back.
Worker reuse and free recycling
Let the worker live more than shit.
Auto-fetched and translated by AI Atlas. All rights belong to the original author.