如何在Ray框架Actor类中正确使用ray.wait()等待收集任务结果

问题原因 #

ray.wait() 接口不会自动修改传入的待等待任务ID列表：每次调用时传入全量任务ID列表，接口只会返回当前已完成的任务ID和未完成的任务ID，不会帮你把已完成ID从原列表删除。
原代码中从未更新runs列表，列表长度始终为3，while循环永远无法退出；且首轮执行后已完成的ID一直存在于列表中，后续调用ray.wait会反复返回同一个已完成ID，导致无限重复打印result:0。

修复方案 #

每次获取到已完成的任务ID、取完执行结果后，手动将对应ID从runs待处理列表中移除即可，修复后的完整代码如下：

import time
import ray


@ray.remote
class Tester:
    def __init__(self, param):
        self.param = param

    def run(self):
        return self.param

params = [0,1,2]
testers = []
for p in params:
    tester = Tester.remote(p)
    testers.append(tester)

runs = []
for i, tester in enumerate(testers):
    runs.append(tester.run.remote())

while len(runs):
    done_ids, pending_ids = ray.wait(runs)
    for done_id in done_ids:
        result = ray.get(done_id)
        print('result:{}'.format(result))
        # 手动移除已处理完成的任务ID
        runs.remove(done_id)
    time.sleep(1)

说明 #

• 代码中将原返回值变量result_ids重命名为pending_ids，更贴合Ray接口的实际返回语义：第二个返回值是传入任务列表中尚未完成的任务ID集合，不会对原传入的runs列表做任何修改。
• 遍历done_ids处理结果的写法兼容ray.wait的num_returns参数调整，即使配置一次返回多个已完成任务，代码也能正常运行。待所有任务ID都从runs列表移除后，循环自动终止，输出结果符合预期。

内容的提问来源于stack exchange，提问作者tompal18