人工智能分布式训练强化学习任务调度模型推理服务【免费下载链接】rayRay is an AI compute engine. Ray consists of a core distributed runtime and a set of AI Libraries for accelerating ML workloads.项目地址https://gitcode.com/gh_mirrors/ra/ray点击查看免费下载导读在 Ray 分布式运行时中任务Task可能因为应用层错误如 Python 异常或系统级故障如机器宕机、Worker 进程崩溃而失败。本文基于 Ray 官方文档 tasks.rst 与其配套示例系统讲解应用开发者应对任务失败的三类核心机制捕获应用级失败RayTaskError 异常体系、配置失败任务自动重试max_retries / retry_exceptions、取消卡死任务ray.cancel / max_calls。读完本文你将掌握用try/except正确捕获远程任务异常、按需控制系统级与应用级重试策略、以及用 State API 排查任务失败根因的完整实战能力。概述任务失败的两种类型Ray 把任务失败划分为两类处理机制截然不同失败类型典型场景Ray 的默认行为应用级错误Application-level errors任务内部抛出 Python 异常异常被包装为RayTaskError作为任务返回值通过ray.get或依赖该结果的后续任务抛给调用方默认不重试系统级故障System-level failuresWorker 进程崩溃、机器宕机Ray 自动重新调度执行直到成功或超过max_retries上限默认 3 次两类机制分别对应下文「捕获应用级失败」与「重试失败任务」两节而「取消行为异常的任务」则是针对任务挂起hanging时的主动干预手段。捕获应用级失败RayTaskError 异常体系当远程 Worker 或 Actor 上的任务因 Python 异常失败时Ray 会把原始异常包装成RayTaskError源码定义于 python/ray/exceptions.py并作为任务的返回值存入对象存储。此后任何调用ray.get获取该结果、或执行依赖该对象的其他任务的 Worker都会收到这个包装后的异常。以下示例来自 doc/source/ray-core/doc_code/task_exceptions.pyimport ray ray.remote def f(): raise Exception(the real error) ray.remote def g(x): return try: ray.get(f.remote()) except ray.exceptions.RayTaskError as e: print(e) # ray::f() (pid71867, ipXXX.XX.XXX.XX) # File errors.py, line 5, in f # raise Exception(the real error) # Exception: the real error try: ray.get(g.remote(f.remote())) except ray.exceptions.RayTaskError as e: print(e) # ray::g() (pid73085, ip128.32.132.47) # At least one of the input arguments for this task could not be computed: # ray.exceptions.RayTaskError: ray::f() (pid73085, ipXXX.XX.XXX.XX) # File errors.py, line 5, in f # raise Exception(the real error) # Exception: the real error从输出可见直接调用ray.get时异常信息包含函数名、进程 PID、IP 与完整堆栈而当任务g的入参依赖失败的任务f的结果时g的任务异常会携带“至少一个输入参数无法计算”的说明并透传底层f的RayTaskError形成异常链。可子类化的用户异常直接 try-catch 用户类型为了提供更友好的捕获体验Ray 在内部通过动态子类化让抛出的异常同时是RayTaskError与用户异常类型的实例该逻辑实现在RayTaskError.as_instanceof_cause()见 python/ray/exceptions.py。只要用户异常类型可被子类化就可以直接用用户异常类型捕获class MyException(Exception): ... ray.remote def raises_my_exc(): raise MyException(a user exception) try: ray.get(raises_my_exc.remote()) except MyException as e: print(e) # ray::raises_my_exc() (pid15329, ip127.0.0.1) # File $PWD/task_exceptions.py, line 45, in raises_my_exc # raise MyException(a user exception) # MyException: a user exception也可以仍然按ray.exceptions.RayTaskError捕获两种写法等价因为该异常实例同时属于两个类型。不可子类化的用户异常通过 cause 字段访问若用户异常类型不可被子类化例如用__init_subclass__禁止派生Ray 会打印一条警告并只抛出纯RayTaskError此时需要通过其cause字段访问原始用户异常class MyFinalException(Exception): def __init_subclass__(cls, /, *args, **kwargs): raise TypeError(Cannot subclass this little exception class.) ray.remote def raises_my_final_exc(): raise MyFinalException(a *final* user exception) try: ray.get(raises_my_final_exc.remote()) except ray.exceptions.RayTaskError as e: assert isinstance(e.cause, MyFinalException) print(e) # 2024-04-08 21:11:47,417 WARNING exceptions.py:177 -- User exception type # class __main__.MyFinalException in RayTaskError can not be subclassed! # This exception will be raised as RayTaskError only. You can use # ray_task_error.cause to access the user exception. print(type(e.cause)) # class __main__.MyFinalException print(e.cause) # a *final* user exception无法序列化的异常降级为 RayErrorRay 在构造RayTaskError时会尝试序列化原始异常见 python/ray/exceptions.py 中对pickle.dumps(cause)的调用。如果原始异常内部持有不可序列化的对象如下例中的threading.Lock序列化会失败Ray 便把cause覆盖为一个ray.exceptions.RayError并在消息中说明原始异常类型import threading class UnserializableException(Exception): def __init__(self): self.lock threading.Lock() ray.remote def raise_unserializable_error(): raise UnserializableException try: ray.get(raise_unserializable_error.remote()) except ray.exceptions.RayTaskError as e: print(e) # ray::raise_unserializable_error() (pid328577, ip172.31.5.154) # File .../main.py, line 25, in raise_unserializable_error # raise UnserializableException # UnserializableException print(type(e.cause)) # class ray.exceptions.RayError print(e.cause) # The original cause of the RayTaskError (class __main__.UnserializableException) # isnt serializable: cannot pickle _thread.lock object. # Overwriting the cause to a RayError.这提醒我们自定义异常应保持“可序列化”只携带简单字段否则将丢失原始异常类型信息只能拿到通用的RayError。用 State API 查询任务退出详情Ray 提供了 State API CLI 来查询任务失败明细该 API 需要pip install ray[default]才可用ray list tasks输出示例节选关键列 List: 2023-05-26 10:32:00.962610 Stats: ------------------------------ Total: 3 Table: ------------------------------ TASK_ID ATTEMPT_NUMBER NAME STATE JOB_ID ACTOR_ID TYPE FUNC_OR_CLASS_NAME PARENT_TASK_ID NODE_ID WORKER_ID ERROR_TYPE 0 16310a0f0a45af5cffffffffffffffffffffffff01000000 0 f FAILED 01000000 NORMAL_TASK f ffffffffffffffffffffffffffffffffffffffff01000000 767bd47b72efb83f33dda1b661621cce9b969b4ef00788140ecca8ad b39e3c523629ab6976556bd46be5dbfbf319f0fce79a664122eb39a9 TASK_EXECUTION_EXCEPTION 1 c2668a65bda616c1ffffffffffffffffffffffff01000000 0 g FAILED 01000000 NORMAL_TASK g ffffffffffffffffffffffffffffffffffffffff01000000 767bd47b72efb83f33dda1b661621cce9b969b4ef00788140ecca8ad b39e3c523629ab6976556bd46be5dbfbf319f0fce79a664122eb39a9 TASK_EXECUTION_EXCEPTION 2 c8ef45ccd0112571ffffffffffffffffffffffff01000000 0 f FAILED 01000000 NORMAL_TASK f ffffffffffffffffffffffffffffffffffffffff01000000 767bd47b72efb83f33dda1b661621cce9b969b4ef00788140ecca8ad b39e3c523629ab6976556bd46be5dbfbf319f0fce79a664122eb39a9 TASK_EXECUTION_EXCEPTIONERROR_TYPE列为TASK_EXECUTION_EXCEPTION即表示应用级异常失败而系统级失败Worker 崩溃会显示WORKER_DIED见下文重试一节。ray list tasks的完整说明可参考文档 State API 概览该文档位于 fault_tolerance 同级目录之外的 ray-core 模块下可沿doc/source/ray-core/目录继续查阅。重试失败任务max_retries 与 retry_exceptions系统级失败自动重执行当执行任务的 Worker 因进程崩溃或机器故障意外死亡时Ray 会自动重新调度该任务直到任务成功或超过最大重试次数。默认重试次数为 3可通过ray.remote装饰器中的max_retries覆盖max_retries3默认值最多重试 3 次max_retries-1无限重试max_retries0完全禁用重试。该选项在 python/ray/_common/ray_option_utils.py 中定义其默认值取自ray_constants.DEFAULT_TASK_MAX_RETRIES。若想为所有提交的任务覆盖默认重试次数可设置 OS 环境变量RAY_TASK_MAX_RETRIES例如通过驱动脚本的环境变量或在 runtime environments 中配置env_vars。其实现位于 python/ray/remote_function.py每次调用远程函数时Ray 会读取os.environ.get(RAY_TASK_MAX_RETRIES, 默认值)作为max_retries的默认值。仓库测试 python/ray/tests/test_reconstruction_2.py 中也演示了通过 runtime_env 与os.environ两种方式注入该变量。官方配套的体验示例见 doc/source/ray-core/doc_code/tasks_fault_tolerance.py它通过os._exit(0)模拟 Worker 崩溃import numpy as np import os import ray import time ray.init(ignore_reinit_errorTrue) ray.remote(max_retries1) def potentially_fail(failure_probability): time.sleep(0.2) if np.random.random() failure_probability: os._exit(0) return 0 for _ in range(3): try: # 如果该任务崩溃Ray 最多再重试 1 次。 # 只要其中一次尝试成功ray.get 就会正常返回 # 否则将抛出异常。 ray.get(potentially_fail.remote(0.5)) print(SUCCESS) except ray.exceptions.WorkerCrashedError: print(FAILURE)注意这里捕获的是ray.exceptions.WorkerCrashedError——当所有重试耗尽、任务因 Worker 崩溃而失败时ray.get抛出该异常与应用级异常抛出的RayTaskError相区分。对象丢失时的自动重建任务完成后其结果对象存储在 Ray 对象存储中。若对象在任务已结束之后丢失例如节点宕机导致对象数据丢失Ray 也会尝试自动重建即重新执行创建该对象的任务恢复丢失的对象。该行为同样由上述max_retries选项控制。更多细节见 Ray 的对象容错文档 objects.rst。应用级错误重试retry_exceptions默认情况下Ray不会因应用代码抛出异常而重试任务——因为很多异常如参数校验错误重试多少次结果都一样。但你可以通过retry_exceptions参数显式控制retry_exceptionsFalse默认值应用级异常不重试retry_exceptionsTrue任何应用级异常都触发重试retry_exceptions[ExceptionClassA, ...]仅当抛出的异常属于列表中的类型时触发重试白名单机制。该选项在 python/ray/_common/ray_option_utils.py 定义可以写在ray.remote装饰器中也可以用.options(...)在调用时临时覆盖。官方示例import numpy as np import os import ray import time ray.init(ignore_reinit_errorTrue) class RandomError(Exception): pass ray.remote(max_retries1, retry_exceptionsTrue) def potentially_fail(failure_probability): if failure_probability 0 or failure_probability 1: raise ValueError( failure_probability must be between 0 and 1, but got: f{failure_probability} ) time.sleep(0.2) if np.random.random() failure_probability: raise RandomError(Failed!) return 0 for _ in range(3): try: ray.get(potentially_fail.remote(0.5)) print(SUCCESS) except RandomError: print(FAILURE)再看按异常类型白名单重试的进阶用法——这里的关键差异在于未被列入白名单的异常如下例的ValueError不会被重试而是立刻抛出# 仅对 RandomError 类型的异常启用重试 retry_on_exception potentially_fail.options(retry_exceptions[RandomError]) try: # failure_probability-1 会触发 ValueError不在白名单内 # 因此不会被重试直接抛出。 ray.get(retry_on_exception.remote(-1)) except ValueError: print(FAILED AS EXPECTED) else: raise RuntimeError(An exception should be raised so this shouldnt be reached.) # 而 RandomError 会被重试。 for _ in range(3): try: ray.get(retry_on_exception.remote(0.5)) print(SUCCESS) except RandomError: print(FAILURE AFTER RETRIES)这一白名单机制非常适合“瞬时故障可重试、确定性错误立即失败”的生产实践例如网络抖动导致的TimeoutError可以重试而参数非法导致的ValueError应直接失败并暴露给上层。观察重试过程ATTEMPT_NUMBER重试会使同一个逻辑任务产生多次执行尝试attempt。用 State API 按任务 ID 过滤即可看到每次尝试的状态# 该 API 仅在通过 pip install ray[default] 安装 Ray 时可用 ray list tasks -f task_id16310a0f0a45af5cffffffffffffffffffffffff01000000输出示例清楚地展示了「第 0 次尝试因 Worker 崩溃失败WORKER_DIED第 1 次尝试成功FINISHED」的完整重试轨迹 List: 2023-05-26 10:38:08.809127 Stats: ------------------------------ Total: 2 Table: ------------------------------ TASK_ID ATTEMPT_NUMBER NAME STATE JOB_ID ACTOR_ID TYPE FUNC_OR_CLASS_NAME PARENT_TASK_ID NODE_ID WORKER_ID ERROR_TYPE 0 16310a0f0a45af5cffffffffffffffffffffffff01000000 0 potentially_fail FAILED 01000000 NORMAL_TASK potentially_fail ffffffffffffffffffffffffffffffffffffffff01000000 94909e0958e38d10d668aa84ed4143d0bf2c23139ae1a8b8d6ef8d9d b36d22dbf47235872ad460526deaf35c178c7df06cee5aa9299a9255 WORKER_DIED 1 16310a0f0a45af5cffffffffffffffffffffffff01000000 1 potentially_fail FINISHED 01000000 NORMAL_TASK potentially_fail ffffffffffffffffffffffffffffffffffffffff01000000 94909e0958e38d10d668aa84ed4143d0bf2c23139ae1a8b8d6ef8d9d 22df7f2a9c68f3db27498f2f435cc18582de991fbcaf49ce0094ddb0同一TASK_ID对应两次尝试ATTEMPT_NUMBER分别为 0 和 1ERROR_TYPE从WORKER_DIED变为空成功这与max_retries1的配置完全吻合。由此可以直观区分两类失败ERROR_TYPE为WORKER_DIED是系统级崩溃为TASK_EXECUTION_EXCEPTION则是应用级异常。取消行为异常的任务ray.cancel 与 max_calls主动取消挂起任务如果任务一直挂起hanging没有返回可以对其返回的ObjectRef调用ray.cancel来取消任务从而继续推进工作流默认行为若任务正在执行中向任务所在 Worker 发送KeyboardInterrupt即让 Python 解释器在任务内部抛出中断异常ray.cancel(obj_ref, forceTrue)强制退出该 Worker 进程适用于任务不响应KeyboardInterrupt例如阻塞在不可中断的系统调用或 C 扩展中的场景。ray.cancel的详细 API 说明可参考 ray.cancel 的 API 参考文档 中的函数签名与参数说明该文档位于doc/source/ray-core/api/目录下可通过 API 索引 定位。需要特别注意的是目前 Ray 不会自动重试已被取消的任务。因此在取消后应用代码需要自行决定后续策略如重新提交、降级或终止整个流程。应对内存泄漏max_calls还有一种隐蔽的任务异常场景某些应用级代码尤其是第三方库的 bug会在 Worker 上重复执行任务后造成内存泄漏。此时即使任务逻辑正确内存也会随执行次数增长而被耗尽。解决办法是在ray.remote装饰器中设置max_callsray.remote(max_calls100) def leaky_task(): # 有内存泄漏风险的第三方库调用 ...一旦某个 Worker 执行该远程函数的次数达到max_callsWorker 会自动退出随后 Ray 会启动新 Worker 承接后续任务从而周期性回收泄漏的内存。该选项定义于 python/ray/_common/ray_option_utils.py。max_calls的默认值规则值得注意从源码_counting_option的默认值及文档说明可以确认CPU 任务默认为无限max_calls0表示不限制Worker 常驻复用GPU 任务默认为 1每次执行后 Worker 即退出以避免 GPU 显存等状态在多次任务间残留或泄漏。因此对 CPU 任务而言max_calls是一个按需开启的“内存兜底”手段而对 GPU 任务默认的一次一换已经是内置的显存卫生策略。小结任务容错的配置决策矩阵综合全文针对不同失败场景的推荐配置可归纳如下场景推荐配置说明Worker 崩溃/机器故障max_retries默认 3-1 无限0 禁用全局默认用RAY_TASK_MAX_RETRIES系统级失败自动重执行应用级异常retry_exceptionsTrue或retry_exceptions[ExType]默认不重试白名单只重试指定异常结果对象丢失任务已完成后复用max_retries配置通过重执行创建任务来重建对象见 objects.rst任务挂起ray.cancel(obj_ref)/ray.cancel(obj_ref, forceTrue)取消后不会被自动重试Worker 内存泄漏max_callsN执行满 N 次后 Worker 自动退出GPU 任务默认 1失败排查ray list tasks/ray list tasks -f task_id...通过ERROR_TYPE区分WORKER_DIED与TASK_EXECUTION_EXCEPTION故障排查的入口是 State APIray list tasks异常捕获的入口是RayTaskError及其cause字段重试策略的入口是max_retries、retry_exceptions与RAY_TASK_MAX_RETRIES主动干预的入口是ray.cancel与max_calls——掌握这四类工具即可在生产环境中从容应对绝大多数任务级故障。同一容错专题下Actor 的故障恢复actors.rst、GCS 容错gcs.rst与节点故障nodes.rst可一并参阅构建完整的 Ray 容错知识体系。赞分享人工智能分布式训练强化学习任务调度模型推理服务【免费下载链接】rayRay is an AI compute engine. Ray consists of a core distributed runtime and a set of AI Libraries for accelerating ML workloads.项目地址https://gitcode.com/gh_mirrors/ra/ray点击查看免费下载相关推荐qinglong任务容错处理异常捕获与自动恢复qinglong任务容错处理异常捕获与自动恢复 引言定时任务的可靠性挑战 在自动化运维和脚本调度场景中定时任务的稳定性直接关系到业务连续性。qinglon任务调度后端前端Friend 任务捕获不变量 INV-TASK-2 深度解析捕获只提议、绝不代写任务Friend 任务捕获不变量 INV TASK 2 深度解析捕获只提议、绝不代写任务 本篇文章围绕 FriendOmi开源仓库中 product/inva人工智能AI 应用语音移动开发后端桌面应用智能硬件MCP 服务Android 3D模型查看器终极指南如何在手机上免费查看STL、OBJ、PLY文件Android 3D模型查看器终极指南如何在手机上免费查看STL、OBJ、PLY文件 还在为无法在手机上查看3D模型而烦恼吗ModelViewer3D是一款移动开发3D渲染图形学创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考