Ray 并发模式实战用 Async Actorasyncio让 Actor 方法并发执行【免费下载链接】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 的 Actor 在单个线程中串行执行方法调用一个长时间运行的方法会阻塞其后的所有调用。本指南基于 Ray 官方 Patterns 文档讲解如何利用 asyncio 将 Actor 改造为 Async Actor通过await主动让出控制权让长轮询、I/O 密集型方法与其他查询类方法在同一进程内并发执行。读完本文你将掌握同步 Actor 的阻塞问题诊断、Async Actor 的改造方法、max_concurrency并发控制以及它与 Threaded Actor 的选型取舍。问题背景单线程 Actor 的顺序执行Ray 的普通 Actor详见 Actor 基础文档默认运行在单个线程中其方法调用严格按照提交顺序串行执行。这意味着一个执行时间很长的方法会阻塞所有后续提交的方法即使后续方法只是简单的状态查询也必须等待长任务结束后才能运行单个 Actor 内部无法通过多方法调用获得任何并发能力。这种模型的好处是状态访问天然线程安全但代价是当 Actor 内部存在永不返回的长运行方法时整个 Actor 将失去响应能力。示例场景长轮询取任务 实时查询进度原文档给出的典型场景是一个 Actor 内部有长轮询方法持续不断地从远端存储获取任务并执行与此同时用户希望随时查询该 Actor 已经执行的任务数量。在默认的同步 Actor 下长轮询方法一旦启动就占据整个线程查询方法永远得不到执行机会。仓库中的完整示例代码位于 pattern_async_actor.py。同步版本ray.get阻塞导致的方法饿死先看同步实现。TaskStore负责产出任务TaskExecutor负责拉取并执行同时维护一个执行计数import ray ray.remote class TaskStore: def get_next_task(self): return task ray.remote class TaskExecutor: def __init__(self, task_store): self.task_store task_store self.num_executed_tasks 0 def run(self): while True: task ray.get(self.task_store.get_next_task.remote()) self._execute_task(task) def _execute_task(self, task): # Executing the task self.num_executed_tasks self.num_executed_tasks 1 def get_num_executed_tasks(self): return self.num_executed_tasks task_store TaskStore.remote() task_executor TaskExecutor.remote(task_store) task_executor.run.remote() try: # This will timeout since task_executor.run occupies the entire actor thread # and get_num_executed_tasks cannot run. ray.get(task_executor.get_num_executed_tasks.remote(), timeout5) except ray.exceptions.GetTimeoutError: print(get_num_executed_tasks didnt finish in 5 seconds)这里的问题非常明确TaskExecutor.run中的while True循环永远运行且ray.get(self.task_store.get_next_task.remote())是阻塞调用整个 Actor 线程被它独占。get_num_executed_tasks提交后永远无法获得执行机会ray.get(..., timeout5)最终抛出ray.exceptions.GetTimeoutError——这正好验证了默认 Actor 方法严格串行的行为。这段带超时的代码本身也是一个实用诊断技巧用timeout参数探测方法是否被阻塞。异步版本用await让出控制权解决思路是把 Actor 改造成Async Actor将方法定义为async def并把阻塞的ray.get替换为await一个 ObjectRef。await在等待远端结果期间会让出控制权给事件循环使其他方法得以插队执行ray.remote class AsyncTaskExecutor: def __init__(self, task_store): self.task_store task_store self.num_executed_tasks 0 async def run(self): while True: # Here we use await instead of ray.get() to # wait for the next task and it will yield # the control while waiting. task await self.task_store.get_next_task.remote() self._execute_task(task) def _execute_task(self, task): # Executing the task self.num_executed_tasks self.num_executed_tasks 1 def get_num_executed_tasks(self): return self.num_executed_tasks async_task_executor AsyncTaskExecutor.remote(task_store) async_task_executor.run.remote() # We are able to run get_num_executed_tasks while run method is running. num_executed_tasks ray.get(async_task_executor.get_num_executed_tasks.remote()) print(fnum of executed tasks so far: {num_executed_tasks})改造点只有两处但效果是质的改变def run(self)→async def run(self)让 Ray 把该 Actor 识别为 Async Actortask ray.get(...)→task await self.task_store.get_next_task.remote()在等待 ObjectRef 期间让出事件循环而不是阻塞线程。现在AsyncTaskExecutor.run虽然在无限循环中持续运行但每次等待任务到达时都会通过await释放控制权因此get_num_executed_tasks可以随时并发执行并返回当前进度。主动让出控制权await asyncio.sleep(0)await通常发生在方法执行I/O 操作如网络请求、读取远端存储的时候这是让出控制权最常见也最自然的时机。但如果你希望在没有真实 I/O 等待的代码段中显式让出控制权可以使用await asyncio.sleep(0)。asyncio.sleep(0)会立即返回但它会触发一次事件循环调度把执行机会交给其他排队的协程是 asyncio 中标准的让出 CPU惯用法。原理Ray 如何识别 Async ActorRay 并不是通过显式声明来区分 Async Actor 的而是自动检测类中是否存在异步方法。相关实现在 python/ray/_private/async_compat.pydef is_async_func(func) - bool: Return True if the function is an async or async generator method. return inspect.iscoroutinefunction(func) or inspect.isasyncgenfunction(func) lru_cache(maxsize2**10) def has_async_methods(cls: object) - bool: Return True if the class has any async methods. return len(inspect.getmembers(cls, predicateis_async_func)) 0在创建 Actor 时python/ray/actor.py 会调用has_async_methods判定类型并据此设置默认并发参数is_asyncio has_async_methods(meta.modified_class) if actor_options.get(max_concurrency) is None: actor_options[max_concurrency] ( ... 1000 # for asyncio execution ... )也就是说只要类中存在至少一个async def方法Ray 就会把该 Actor 视为 Async Actor其内部方法将运行在同一个 asyncio 事件循环上。底层 C 端src/ray/core_worker/context.cc也会在 actor 任务执行时记录current_actor_is_asyncio_与current_actor_max_concurrency_用于调度层面的并发控制。Async Actor 的关键语义详见 AsyncIO / Concurrency for Actors所有方法运行在单个 Python 事件循环中只有一个线程同一时刻只有一个任务在真正执行任务之间通过await进行多路复用multiplexed在 async 方法内禁止使用阻塞的ray.get或ray.wait因为它们会卡住整个事件循环导致所有方法失去响应。并发上限max_concurrency选项Async Actor 默认允许最多1000个方法调用同时排队运行实际执行仍是事件循环交替进行的。你可以通过.options(max_concurrency...)限制并发数这常被用来控制资源占用或实现批处理语义。以官方文档中的批处理示例为参考actor AsyncActor.options(max_concurrency2).remote(2) # Only 2 tasks will run concurrently. # Once 2 finish, the next 2 should run. ray.get([actor.run_task.remote() for _ in range(8)])max_concurrency2时8 个任务会以每批 2 个的方式进入并发执行。关于max_concurrencypython/ray/actor.py 的 API 文档明确说明了几条重要约束它只对direct actor call直连调用生效默认值threaded 执行为1asyncio 执行为1000当max_concurrency 1时执行顺序不再保证使用多线程max_concurrency 1或 Async Actor 时allow_out_of_order_execution必须为True默认即如此因为并发执行天然会打乱提交顺序。进阶ObjectRef 与 asyncio.Future 的互操作Async Actor 场景下你还可以把 ObjectRef 直接当作 asyncio 可等待对象使用这在已有异步代码中集成 Ray 时非常方便参考 async_api.rst 中的完整示例import asyncio import ray ray.remote def some_task(): return 1 async def await_obj_ref(): await some_task.remote() await asyncio.wait([some_task.remote()])在 Python 3.11 上还可以把 ObjectRef 包装成标准的asyncio.Future对象async def convert_to_asyncio_future(): ref some_task.remote() fut: asyncio.Future asyncio.wrap_future(ref.future()) print(await fut)与 Threaded Actor 的选型对比原文档特别提示你同样可以使用 Threaded Actor 实现并发详见 Threaded Actors。两者适用场景不同维度Async ActorThreaded Actor实现方式方法定义为async def靠await让出控制权普通同步方法 .options(max_concurrencyn)底层模型单线程 单事件循环任务多路复用线程池线程数由max_concurrency决定适用场景I/O 密集、等待型任务轮询、网络请求、远端存储计算密集且无法用await让出控制权的代码并发默认值10001状态安全单线程无竞争条件多线程访问共享状态需自行加锁关键判断依据如果方法内存在无法通过await让出控制权的计算密集段Async Actor 反而会被拖慢——因为事件循环只有这一个任务在跑其他任务全部饿死。此时应改用 Threaded Actor让长计算运行在独立线程中。ray.remote class ThreadedActor: def task_1(self): print(Im running in a thread!) def task_2(self): print(Im running in another thread!) a ThreadedActor.options(max_concurrency2).remote() ray.get([a.task_1.remote(), a.task_2.remote()])注意一个容易踩坑的规则只要 Actor 中存在一个async def方法Ray 就会把它识别为 Async Actor 而非 Threaded Actor因此不要混用两种模式。另外需要清醒认识 Python 的 GIL 限制无论 Async Actor 还是 Threaded Actor同一时刻只有一个线程能执行 Python 字节码详见 async_api.rst 的说明。只有当代码调用 NumPy、Cython、TensorFlow、PyTorch 等会释放 GIL的原生库时才能真正获得并行加速。这两种并发模型的价值在于避免阻塞、提升吞吐与响应性而非突破 GIL。补充remote task 不支持 asyncio需要特别区分的是Ray 的remote task无状态任务不支持 asyncio直接定义ray.remote async def f()会失败。如果确实需要在任务中运行异步代码可以包一层同步 wrapperasync def f(): pass ray.remote def wrapper(): import asyncio asyncio.run(f())也就是说asyncio 集成是 Actor 专属能力与普通任务不同。总结本文核心模式可归纳为三步识别阻塞点默认同步 Actor 中长运行方法尤其是while True轮询 阻塞ray.get会饿死后续方法异步化改造将方法改为async def用await object_ref替代ray.get在等待 I/O 时让出控制权需要主动让出时使用await asyncio.sleep(0)按需控制并发用max_concurrency调节并发上限并在计算密集场景改用 Threaded Actor。这套模式非常适合后台持续干活 前台随时查询的 Actor 设计例如任务队列消费者、监控探针、模型批处理 worker 等。完整的可运行示例与测试代码见 pattern_async_actor.py更系统的 asyncio/并发说明可继续阅读 AsyncIO / Concurrency for Actors。【免费下载链接】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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Ray 并发模式实战：用 Async Actor（asyncio）让 Actor 方法并发执行
Ray 并发模式实战用 Async Actorasyncio让 Actor 方法并发执行【免费下载链接】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 的 Actor 在单个线程中串行执行方法调用一个长时间运行的方法会阻塞其后的所有调用。本指南基于 Ray 官方 Patterns 文档讲解如何利用 asyncio 将 Actor 改造为 Async Actor通过await主动让出控制权让长轮询、I/O 密集型方法与其他查询类方法在同一进程内并发执行。读完本文你将掌握同步 Actor 的阻塞问题诊断、Async Actor 的改造方法、max_concurrency并发控制以及它与 Threaded Actor 的选型取舍。问题背景单线程 Actor 的顺序执行Ray 的普通 Actor详见 Actor 基础文档默认运行在单个线程中其方法调用严格按照提交顺序串行执行。这意味着一个执行时间很长的方法会阻塞所有后续提交的方法即使后续方法只是简单的状态查询也必须等待长任务结束后才能运行单个 Actor 内部无法通过多方法调用获得任何并发能力。这种模型的好处是状态访问天然线程安全但代价是当 Actor 内部存在永不返回的长运行方法时整个 Actor 将失去响应能力。示例场景长轮询取任务 实时查询进度原文档给出的典型场景是一个 Actor 内部有长轮询方法持续不断地从远端存储获取任务并执行与此同时用户希望随时查询该 Actor 已经执行的任务数量。在默认的同步 Actor 下长轮询方法一旦启动就占据整个线程查询方法永远得不到执行机会。仓库中的完整示例代码位于 pattern_async_actor.py。同步版本ray.get阻塞导致的方法饿死先看同步实现。TaskStore负责产出任务TaskExecutor负责拉取并执行同时维护一个执行计数import ray ray.remote class TaskStore: def get_next_task(self): return task ray.remote class TaskExecutor: def __init__(self, task_store): self.task_store task_store self.num_executed_tasks 0 def run(self): while True: task ray.get(self.task_store.get_next_task.remote()) self._execute_task(task) def _execute_task(self, task): # Executing the task self.num_executed_tasks self.num_executed_tasks 1 def get_num_executed_tasks(self): return self.num_executed_tasks task_store TaskStore.remote() task_executor TaskExecutor.remote(task_store) task_executor.run.remote() try: # This will timeout since task_executor.run occupies the entire actor thread # and get_num_executed_tasks cannot run. ray.get(task_executor.get_num_executed_tasks.remote(), timeout5) except ray.exceptions.GetTimeoutError: print(get_num_executed_tasks didnt finish in 5 seconds)这里的问题非常明确TaskExecutor.run中的while True循环永远运行且ray.get(self.task_store.get_next_task.remote())是阻塞调用整个 Actor 线程被它独占。get_num_executed_tasks提交后永远无法获得执行机会ray.get(..., timeout5)最终抛出ray.exceptions.GetTimeoutError——这正好验证了默认 Actor 方法严格串行的行为。这段带超时的代码本身也是一个实用诊断技巧用timeout参数探测方法是否被阻塞。异步版本用await让出控制权解决思路是把 Actor 改造成Async Actor将方法定义为async def并把阻塞的ray.get替换为await一个 ObjectRef。await在等待远端结果期间会让出控制权给事件循环使其他方法得以插队执行ray.remote class AsyncTaskExecutor: def __init__(self, task_store): self.task_store task_store self.num_executed_tasks 0 async def run(self): while True: # Here we use await instead of ray.get() to # wait for the next task and it will yield # the control while waiting. task await self.task_store.get_next_task.remote() self._execute_task(task) def _execute_task(self, task): # Executing the task self.num_executed_tasks self.num_executed_tasks 1 def get_num_executed_tasks(self): return self.num_executed_tasks async_task_executor AsyncTaskExecutor.remote(task_store) async_task_executor.run.remote() # We are able to run get_num_executed_tasks while run method is running. num_executed_tasks ray.get(async_task_executor.get_num_executed_tasks.remote()) print(fnum of executed tasks so far: {num_executed_tasks})改造点只有两处但效果是质的改变def run(self)→async def run(self)让 Ray 把该 Actor 识别为 Async Actortask ray.get(...)→task await self.task_store.get_next_task.remote()在等待 ObjectRef 期间让出事件循环而不是阻塞线程。现在AsyncTaskExecutor.run虽然在无限循环中持续运行但每次等待任务到达时都会通过await释放控制权因此get_num_executed_tasks可以随时并发执行并返回当前进度。主动让出控制权await asyncio.sleep(0)await通常发生在方法执行I/O 操作如网络请求、读取远端存储的时候这是让出控制权最常见也最自然的时机。但如果你希望在没有真实 I/O 等待的代码段中显式让出控制权可以使用await asyncio.sleep(0)。asyncio.sleep(0)会立即返回但它会触发一次事件循环调度把执行机会交给其他排队的协程是 asyncio 中标准的让出 CPU惯用法。原理Ray 如何识别 Async ActorRay 并不是通过显式声明来区分 Async Actor 的而是自动检测类中是否存在异步方法。相关实现在 python/ray/_private/async_compat.pydef is_async_func(func) - bool: Return True if the function is an async or async generator method. return inspect.iscoroutinefunction(func) or inspect.isasyncgenfunction(func) lru_cache(maxsize2**10) def has_async_methods(cls: object) - bool: Return True if the class has any async methods. return len(inspect.getmembers(cls, predicateis_async_func)) 0在创建 Actor 时python/ray/actor.py 会调用has_async_methods判定类型并据此设置默认并发参数is_asyncio has_async_methods(meta.modified_class) if actor_options.get(max_concurrency) is None: actor_options[max_concurrency] ( ... 1000 # for asyncio execution ... )也就是说只要类中存在至少一个async def方法Ray 就会把该 Actor 视为 Async Actor其内部方法将运行在同一个 asyncio 事件循环上。底层 C 端src/ray/core_worker/context.cc也会在 actor 任务执行时记录current_actor_is_asyncio_与current_actor_max_concurrency_用于调度层面的并发控制。Async Actor 的关键语义详见 AsyncIO / Concurrency for Actors所有方法运行在单个 Python 事件循环中只有一个线程同一时刻只有一个任务在真正执行任务之间通过await进行多路复用multiplexed在 async 方法内禁止使用阻塞的ray.get或ray.wait因为它们会卡住整个事件循环导致所有方法失去响应。并发上限max_concurrency选项Async Actor 默认允许最多1000个方法调用同时排队运行实际执行仍是事件循环交替进行的。你可以通过.options(max_concurrency...)限制并发数这常被用来控制资源占用或实现批处理语义。以官方文档中的批处理示例为参考actor AsyncActor.options(max_concurrency2).remote(2) # Only 2 tasks will run concurrently. # Once 2 finish, the next 2 should run. ray.get([actor.run_task.remote() for _ in range(8)])max_concurrency2时8 个任务会以每批 2 个的方式进入并发执行。关于max_concurrencypython/ray/actor.py 的 API 文档明确说明了几条重要约束它只对direct actor call直连调用生效默认值threaded 执行为1asyncio 执行为1000当max_concurrency 1时执行顺序不再保证使用多线程max_concurrency 1或 Async Actor 时allow_out_of_order_execution必须为True默认即如此因为并发执行天然会打乱提交顺序。进阶ObjectRef 与 asyncio.Future 的互操作Async Actor 场景下你还可以把 ObjectRef 直接当作 asyncio 可等待对象使用这在已有异步代码中集成 Ray 时非常方便参考 async_api.rst 中的完整示例import asyncio import ray ray.remote def some_task(): return 1 async def await_obj_ref(): await some_task.remote() await asyncio.wait([some_task.remote()])在 Python 3.11 上还可以把 ObjectRef 包装成标准的asyncio.Future对象async def convert_to_asyncio_future(): ref some_task.remote() fut: asyncio.Future asyncio.wrap_future(ref.future()) print(await fut)与 Threaded Actor 的选型对比原文档特别提示你同样可以使用 Threaded Actor 实现并发详见 Threaded Actors。两者适用场景不同维度Async ActorThreaded Actor实现方式方法定义为async def靠await让出控制权普通同步方法 .options(max_concurrencyn)底层模型单线程 单事件循环任务多路复用线程池线程数由max_concurrency决定适用场景I/O 密集、等待型任务轮询、网络请求、远端存储计算密集且无法用await让出控制权的代码并发默认值10001状态安全单线程无竞争条件多线程访问共享状态需自行加锁关键判断依据如果方法内存在无法通过await让出控制权的计算密集段Async Actor 反而会被拖慢——因为事件循环只有这一个任务在跑其他任务全部饿死。此时应改用 Threaded Actor让长计算运行在独立线程中。ray.remote class ThreadedActor: def task_1(self): print(Im running in a thread!) def task_2(self): print(Im running in another thread!) a ThreadedActor.options(max_concurrency2).remote() ray.get([a.task_1.remote(), a.task_2.remote()])注意一个容易踩坑的规则只要 Actor 中存在一个async def方法Ray 就会把它识别为 Async Actor 而非 Threaded Actor因此不要混用两种模式。另外需要清醒认识 Python 的 GIL 限制无论 Async Actor 还是 Threaded Actor同一时刻只有一个线程能执行 Python 字节码详见 async_api.rst 的说明。只有当代码调用 NumPy、Cython、TensorFlow、PyTorch 等会释放 GIL的原生库时才能真正获得并行加速。这两种并发模型的价值在于避免阻塞、提升吞吐与响应性而非突破 GIL。补充remote task 不支持 asyncio需要特别区分的是Ray 的remote task无状态任务不支持 asyncio直接定义ray.remote async def f()会失败。如果确实需要在任务中运行异步代码可以包一层同步 wrapperasync def f(): pass ray.remote def wrapper(): import asyncio asyncio.run(f())也就是说asyncio 集成是 Actor 专属能力与普通任务不同。总结本文核心模式可归纳为三步识别阻塞点默认同步 Actor 中长运行方法尤其是while True轮询 阻塞ray.get会饿死后续方法异步化改造将方法改为async def用await object_ref替代ray.get在等待 I/O 时让出控制权需要主动让出时使用await asyncio.sleep(0)按需控制并发用max_concurrency调节并发上限并在计算密集场景改用 Threaded Actor。这套模式非常适合后台持续干活 前台随时查询的 Actor 设计例如任务队列消费者、监控探针、模型批处理 worker 等。完整的可运行示例与测试代码见 pattern_async_actor.py更系统的 asyncio/并发说明可继续阅读 AsyncIO / Concurrency for Actors。【免费下载链接】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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考