AsyncIO / 演员的并发性#

在一个单一的actor进程中，可以执行并发线程。

Ray 在 actor 中提供两种类型的并发：

异步执行

线程

请记住，Python 的全局解释器锁 (GIL) 将只允许一个 Python 代码线程同时运行。

这意味着如果你只是并行化 Python 代码，你不会得到真正的并行性。如果你调用 Numpy、Cython、Tensorflow 或 PyTorch 代码，这些库在调用 C/C++ 函数时会释放 GIL。

无论是 线程化角色还是异步角色 模型，都无法让你绕过 GIL。

Actor 的 AsyncIO#

自Python 3.5起，可以使用 async/await 语法编写并发代码。Ray 原生集成了 asyncio。你可以将 ray 与流行的异步框架如 aiohttp、aioredis 等一起使用。

import ray
import asyncio

@ray.remote
class AsyncActor:
    # multiple invocation of this method can be running in
    # the event loop at the same time
    async def run_concurrent(self):
        print("started")
        await asyncio.sleep(2) # concurrent workload here
        print("finished")

actor = AsyncActor.remote()

# regular ray.get
ray.get([actor.run_concurrent.remote() for _ in range(4)])

# async ray.get
async def async_get():
    await actor.run_concurrent.remote()
asyncio.run(async_get())

(AsyncActor pid=40293) started
(AsyncActor pid=40293) started
(AsyncActor pid=40293) started
(AsyncActor pid=40293) started
(AsyncActor pid=40293) finished
(AsyncActor pid=40293) finished
(AsyncActor pid=40293) finished
(AsyncActor pid=40293) finished

...

ObjectRefs 作为 asyncio.Futures#

ObjectRefs 可以转换为 asyncio.Futures。这一特性使得在现有的并发应用程序中可以 await ray futures。

而不是：

import ray

@ray.remote
def some_task():
    return 1

ray.get(some_task.remote())
ray.wait([some_task.remote()])

你可以这样做：

import ray
import asyncio

@ray.remote
def some_task():
    return 1

async def await_obj_ref():
    await some_task.remote()
    await asyncio.wait([some_task.remote()])

asyncio.run(await_obj_ref())

更多关于 asyncio 的模式，包括超时和 asyncio.gather，请参阅 asyncio 文档。

如果你需要直接访问未来对象，你可以调用：

import asyncio

async def convert_to_asyncio_future():
    ref = some_task.remote()
    fut: asyncio.Future = asyncio.wrap_future(ref.future())
    print(await fut)
asyncio.run(convert_to_asyncio_future())

ObjectRefs 作为 concurrent.futures.Futures#

ObjectRefs 也可以被包装成 concurrent.futures.Future 对象。这对于与现有的 concurrent.futures API 接口非常有用：

import concurrent

refs = [some_task.remote() for _ in range(4)]
futs = [ref.future() for ref in refs]
for fut in concurrent.futures.as_completed(futs):
    assert fut.done()
    print(fut.result())

定义一个异步角色#

通过使用 async 方法定义，Ray 将自动检测一个 actor 是否支持 async 调用。

import asyncio

@ray.remote
class AsyncActor:
    async def run_task(self):
        print("started")
        await asyncio.sleep(2) # Network, I/O task here
        print("ended")

actor = AsyncActor.remote()
# All 5 tasks should start at once. After 2 second they should all finish.
# they should finish at the same time
ray.get([actor.run_task.remote() for _ in range(5)])

(AsyncActor pid=3456) started
(AsyncActor pid=3456) started
(AsyncActor pid=3456) started
(AsyncActor pid=3456) started
(AsyncActor pid=3456) started
(AsyncActor pid=3456) ended
(AsyncActor pid=3456) ended
(AsyncActor pid=3456) ended
(AsyncActor pid=3456) ended
(AsyncActor pid=3456) ended

在底层，Ray 在单个 Python 事件循环中运行所有方法。请注意，不允许在异步 actor 方法中运行阻塞的 ray.get 或 ray.wait，因为 ray.get 会阻塞事件循环的执行。

在异步角色中，任何时候只能运行一个任务（尽管任务可以多路复用）。AsyncActor 中将只有一个线程！如果你想要一个线程池，请参阅线程化角色。

在异步执行者中设置并发#

你可以使用 max_concurrency 标志设置一次运行的“并发”任务数量。默认情况下，可以同时运行1000个任务。

import asyncio

@ray.remote
class AsyncActor:
    async def run_task(self):
        print("started")
        await asyncio.sleep(1) # Network, I/O task here
        print("ended")

actor = AsyncActor.options(max_concurrency=2).remote()

# Only 2 tasks will be running concurrently. Once 2 finish, the next 2 should run.
ray.get([actor.run_task.remote() for _ in range(8)])

(AsyncActor pid=5859) started
(AsyncActor pid=5859) started
(AsyncActor pid=5859) ended
(AsyncActor pid=5859) ended
(AsyncActor pid=5859) started
(AsyncActor pid=5859) started
(AsyncActor pid=5859) ended
(AsyncActor pid=5859) ended
(AsyncActor pid=5859) started
(AsyncActor pid=5859) started
(AsyncActor pid=5859) ended
(AsyncActor pid=5859) ended
(AsyncActor pid=5859) started
(AsyncActor pid=5859) started
(AsyncActor pid=5859) ended
(AsyncActor pid=5859) ended

线程化角色#

有时，asyncio 并不是你角色的理想解决方案。例如，你可能有一个方法执行一些计算密集型任务，同时阻塞事件循环，不通过 await 放弃控制。这会损害异步角色的性能，因为异步角色一次只能执行一个任务，并且依赖 await 进行上下文切换。

相反，你可以使用 max_concurrency 演员选项，而无需任何异步方法，从而实现线程并发（如线程池）。

警告

当actor定义中至少有一个 async def 方法时，Ray 会将该actor识别为 AsyncActor 而不是 ThreadedActor。

@ray.remote
class ThreadedActor:
    def task_1(self): print("I'm running in a thread!")
    def task_2(self): print("I'm running in another thread!")

a = ThreadedActor.options(max_concurrency=2).remote()
ray.get([a.task_1.remote(), a.task_2.remote()])

(ThreadedActor pid=4822) I'm running in a thread!
(ThreadedActor pid=4822) I'm running in another thread!

每个线程化角色的调用都将在一个线程池中运行。线程池的大小受 max_concurrency 值的限制。

远程任务的AsyncIO#

我们不支持远程任务的 asyncio。以下代码片段将会失败：

@ray.remote
async def f():
    pass

相反，你可以用一个包装器包裹 async 函数来同步运行任务：

async def f():
    pass

@ray.remote
def wrapper():
    import asyncio
    asyncio.run(f())