我是一个Ray Task

第一回 横空出世

我出生的时候，Driver 正盯着一行代码：

result_ref = slow_double.remote(21)

他敲下回车，我就来了。

我还没看清周围是什么地方，一个戴眼镜的办事员已经把一张卡片塞进我的口袋，上面印着一串长得吓人的 Task ID。

“拿好，这是你的身份证。”

“我是谁？”

“一个 Ray Task。”

“我要干什么？”

“把 21 乘以 2。”

我低头一看，身边还站着一个和我同时出生的家伙。他瘦得像一张纸，胸前也挂着一串编号。

“他又是谁？”

“ObjectRef，你的结果凭证。”

我还想和他打个招呼，他已经被人飞快地送回了 Driver。

“等等！”我喊道，“我还没算呢，他回去干什么？”

办事员白了我一眼：“人家只是凭证，又不是结果。先把凭证给 Driver，等你干完活，他再凭票取货。”

后来我才知道，从 Driver 敲下回车到 ObjectRef 回到他手里，一共只过了 28.504 毫秒。

Driver 拿到凭证，半信半疑地问：“结果有了吗？”

ObjectRef 摊开双手：“我只是证明世界上已经有了这么一件任务。至于结果，您得等那位兄弟干完。”

Driver 不死心，马上用 ray.wait(..., timeout=0) 查了一次。

结果当然是：没好。

Task 已提交，ObjectRef 先返回，但任务尚未开始执行

那时候的我，连 Worker 的门朝哪边开都不知道。

办事员把我领进一间档案室。门牌上写着：CoreWorker。

屋里堆满了函数档案。最上面一份写着 slow_double，旁边还盖着一个 RemoteFunction 的章。

我好奇地问：“这份档案什么时候送来的？”

CoreWorker 一边翻文件一边说：“@ray.remote 出现的时候，只是把普通函数登记成 RemoteFunction。第一次在这个 Job 里真的有人调用，函数定义才会被序列化、导出。以后再来同样的活，就不必每次重新登记了。”

说完，他拿出一本厚厚的册子，开始往里面填东西。

“姓名，Task ID；要调用的函数，slow_double；参数，21；返回值，一个；CPU 需求……”

“这是什么？”

“TaskSpecification，你的任务说明书。你去哪儿、用多少资源、怎么重试、采用什么调度策略，都写在这里。没有它，谁知道该怎么安排你？”

我抱着说明书，觉得自己总算是个有身份、有使命的 Task 了。

可我万万没想到，真正的旅程才刚开始。

第二回 三种包裹

离开档案室，我被带到参数检查站。

这里堆满了大大小小的包裹。有的只有一张纸，有的像小山一样，还有一些包裹里什么都没有，外面只贴着一个 ObjectRef。

检查员拿起我的参数看了看：“一个整数 21，小件，随身带走。”

“参数不是都要先存进对象存储吗？”我问。

旁边几个 Task 听完哈哈大笑。

一个老 Task 说：“新来的吧？这么小的东西，序列化以后直接塞进 RPC 就行。要是传个整数也得先存仓库、再领货，大家一天到晚净办手续了。”

这时，一辆货车慢慢开过来，上面装着一个巨大的数组。检查员看了一眼就挥手：“太大了，送对象存储！”

大数组被卸进仓库，原来的位置换成了一张新的 ObjectRef。

“这种大件如果要给很多 Task 反复使用，最好由 Driver 提前 ray.put() 一次。”老 Task告诉我，“要不然每次都按值传，光搬家就能把人累死。”

小参数、顶层 ObjectRef 和大参数对应三条数据路径

我们正聊着，角落里忽然传来一阵抱怨。

“凭什么不让我走？我的 CPU 和参数都写得清清楚楚！”

我转头看去，一个 Task 正被拦在门口。他的说明书上也贴着一张 ObjectRef。

检查员敲了敲那张凭证：“你的上游还没把货生产出来。拿着一张空凭证，你让 Worker 怎么干活？”

“那 Driver 先 ray.get() 一下，再把结果交给我不就行了？”

“用不着。你就在这里等。上游完成以后，我们自然会让你继续。Ray 既然记住了你们之间的依赖，何必让 Driver 跑回来当传话筒？”

那个 Task 只好坐下。

后来，他等的上游在 2046.546 毫秒时完成，他在 2061.772 毫秒时才进入 Worker。中间没有人去通知 Driver，更没有 Driver 手工把他叫醒。

我这才明白，在我们这里，ObjectRef 不只是一张取结果的票，它还能把两个 Task 的命运连在一起。

第三回 一张租约

参数检查通过以后，我兴冲冲地往前跑，以为马上就能计算 21 乘以 2。

没跑几步，又被人拦住了。

“排队。”

“这次又等什么？”

“等 Worker，等资源。”

这里叫 NormalTaskSubmitter 大厅，所有准备开工的普通 Task 都要从这里出发。墙上挂着一块大屏幕，上面滚动着各种状态：等待依赖、等待节点、等待 Worker、运行中、已完成。

我旁边有两个 Task，长得一模一样，都说自己需要一个 CPU。

第一个很快被叫走了。第二个等了半天，头顶上的牌子一直写着 PENDING_NODE_ASSIGNMENT。

“我明明已经 .remote() 了！”他气得直跺脚，“ObjectRef 都发回去了，怎么还不让我运行？”

管理员指着资源牌说：“这里只剩 1 个逻辑 CPU，已经被前面那位占了。你被提交了，不等于你已经分到资源；有了 ObjectRef，也不等于你已经开始执行。”

资源大厅里一个 Task 运行，另一个 Task 等待 CPU 和 Worker

“那我要等多久？”

“等资源空出来。”

他垂头丧气地坐下了。

管理员拿起电话，开始替我联系 Worker。

“Raylet 吗？这里有个 Task，参数已经齐了，需要一份 Worker lease。”

电话那边沉默了一会儿，像是在翻账本。

Raylet 是每个节点上的资源管家。他要看本节点的逻辑 CPU、GPU 和自定义资源够不够，还要看有没有合适的 Worker。如果本地不合适，他也可能把我们介绍到别的节点。

没多久，一张租约送了过来。

“拿好，Worker 已经租给你了。”管理员说。

我接过租约，忍不住问：“既然 Raylet 权力这么大，为什么不直接让他执行我？”

管理员一口水差点喷出来。

“让 Raylet 跑你的 Python 函数？你想累死他吗？Raylet 管调度、资源和 Worker，真正执行用户代码的是 Worker。以后看日志也别找错人：没分到资源可以先查调度和 Raylet，用户代码报错要去找对应的 Worker。”

我牢牢记住了这句话。

拿到租约后，Owner 把我的 TaskSpecification 装进 PushTask RPC，送往 Worker。

我终于要进车间了。

第四回 Worker 车间

Worker 车间和我想象的不太一样。

这里没有一个人对着代码敲键盘。Worker 拿过我的说明书，像熟练工一样逐项检查。

“引用参数在本地吗？”

“在。”

“函数定义有吗？”

“有。”

“参数能反序列化吗？”

“能。”

Worker 这才卷起袖子：“开工。”

Worker 收到说明书和参数后，才真正执行 Python 函数

从 .remote() 被调用开始算，他在 57.243 毫秒时真正进入了我的用户函数。

我原以为把 21 乘以 2 是件眨眼就能完成的事，谁知函数里赫然写着：

time.sleep(1.5)

“为什么还要睡一觉？”我问。

Worker 打了个哈欠：“实验就是这么安排的。你别吵，等一秒半。”

“Driver 会不会一直在外面等我？”

“他想等就 ray.get()，不想等就先做别的事。你们 Task 是异步提交的，他没必要从你出生起就守在门口。”

我只好跟着睡。

睡醒以后，Worker 算出结果 42，又把自己的 PID、Worker ID、Task ID 和执行时间一起记录下来。

我看了一眼记录：Driver 的 PID 是 81420，Worker 的 PID 是 81497。

“我们果然不在同一个进程。”我说。

“那当然。还有，你在 Driver 那边登记的 Task ID，和我运行时看到的是同一个。这能证明执行的就是你，不是半路掉包的另一个 Task。”

到了 1558.296 毫秒，工作全部完成。

临走前，我看到门外又来了一位同函数的 Task。他的手续办得比我快得多。

“凭什么？”我不服气。

Worker 说：“你是第一次来，函数导出、Worker 启动、模块导入、运行环境准备，都可能赶在你身上。后面的人能复用一些东西，自然轻松。但只跑一次不能当性能基准，回去可别吹嘘什么固定延迟。”

第五回 凭票取货

结果离开 Worker 时，也要看个头。

小结果可以直接跟着 PushTask 的响应回去，放进调用者的本地内存。大结果没那么轻巧，通常要写进执行节点的 Plasma 对象存储。如果 Driver 在另一台机器，等他真的来取时，还得先跨节点搬过去。

我们的结果很小，很快就送到了。

远在 Driver 身边的 ObjectRef 胸前忽然亮起一盏绿灯。

“货到了！”他喊道。

Driver 早已等得不耐烦，立刻执行：

result = ray.get(result_ref)

他拿到 42，满意地点点头。

旁边有人说：“原来 ray.get() 就是去 GCS 下载结果。”

ObjectRef 马上纠正他：“别什么都往 GCS 身上推。GCS 管的是控制面元数据和函数定义。普通 Task 的大对象，主要在各节点的对象存储之间流动。我的这位 Task 兄弟，则是靠 Owner、CoreWorker、Raylet 和 Worker 一路协作才完成的。”

Driver 看着我从 Worker 车间寄回的记录，问：“所以 .remote() 到底做了什么？”

我想了想，说：

“它给我登记身份，装好参数和资源要求，异步把我送进 Ray 的执行系统，再把 ObjectRef 提前交给你。此后我可能等依赖，可能等 CPU，可能等 Worker，也可能正在运行。只有当结果真的产生，ObjectRef 才会变成可取状态。”

“那 ObjectRef 是结果吗？”

“不是，他是结果的身份和凭证。”

“Raylet 会执行 Python 函数吗？”

“不会，那是 Worker 的活。”

“所有参数和结果都会进对象存储吗？”

“也不会。小东西可以内联，大东西才通常需要仓库。”

Driver 终于笑了：“这下，一行 .remote() 不再像魔法了。”

从 Driver 到 ObjectRef 可取的一次完整 remote() 旅程

我的第一次任务就这样结束了。

至于 ObjectRef 为什么不只是普通 Future，为什么他能把上下游 Task 连成一张依赖图，以及当所有人都忘掉他以后，对象会不会跟着消失——

那是他的故事了。