如何在事件循环之外运行协程



通常,您可以通过执行以下操作来获取协程的结果:

async def coro():
await asycnio.sleep(3)
return 'a value'
loop = asyncio.get_event_loop()
value = loop.run_until_complete(coro())

出于好奇,在不使用事件循环的情况下获得该值的最简单方法是什么?

[编辑]

我认为更简单的方法是:

async def coro():
...
value = asyncio.run(coro())  # Python 3.7+

但是有什么方法可以像在JS中一样在全球范围内yield from(或await)coro()?如果没有,为什么?

这里有两个问题:一个是关于"在顶层"等待协程,或者更具体地说是在开发环境中。另一个是关于在没有事件循环的情况下运行协程。

关于第一个问题,这在 Python 中当然是可能的,就像在 Chrome Canary 开发工具中一样 - 通过工具通过自己与事件循环的集成来处理它。事实上,IPython 7.0 及更高版本原生支持 asyncio,您可以按预期在顶级使用await coro()

关于第二个问题,在没有事件循环的情况下驱动单个协程很容易,但它不是很有用。让我们来看看为什么。

调用协程函数时,它会返回协程对象。此对象通过调用其send()方法启动和恢复。当协程决定挂起时(因为它await阻塞的东西),send()将返回。当协程决定返回时(因为它已经到达终点或因为它遇到了显式return),它将引发一个StopIteration异常,并将value属性设置为返回值。考虑到这一点,单个协程的最小驱动程序可能如下所示:

def drive(c):
while True:
try:
c.send(None)
except StopIteration as e:
return e.value

这将非常适合简单的协程:

>>> async def pi():
...     return 3.14
... 
>>> drive(pi())
3.14

甚至对于更复杂的:

>>> async def plus(a, b):
...     return a + b
... 
>>> async def pi():
...     val = await plus(3, 0.14)
...     return val
... 
>>> drive(pi())
3.14

但是仍然缺少一些东西 - 上述协程都不会暂停其执行。当协程挂起时,它允许其他协程运行,这使事件循环能够(看起来)一次执行多个协程。例如,asyncio 有一个sleep()协程,当等待时,它会在指定的时间段内暂停执行:

async def wait(s):
await asyncio.sleep(1)
return s
>>> asyncio.run(wait("hello world"))
'hello world'      # printed after a 1-second pause

但是,drive无法执行此协程以完成:

>>> drive(wait("hello world"))
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "<stdin>", line 4, in drive
File "<stdin>", line 2, in wait
File "/usr/lib/python3.7/asyncio/tasks.py", line 564, in sleep
return await future
RuntimeError: await wasn't used with future

发生的情况是,sleep()通过生成一个特殊的"未来"对象来与事件循环进行通信。等待未来的协程只有在设置未来后才能恢复。"真实"事件循环将通过运行其他协程来实现,直到将来完成。

为了解决这个问题,我们可以编写自己的sleep实现,与我们的迷你事件循环配合使用。为此,我们需要使用迭代器来实现可等待的:

class my_sleep:
def __init__(self, d):
self.d = d
def __await__(self):
yield 'sleep', self.d

我们生成一个元组,该元组不会被协程调用方看到,但会告诉drive(我们的事件循环)该做什么。drivewait现在看起来像这样:

def drive(c):
while True:
try:
susp_val = c.send(None)
if susp_val is not None and susp_val[0] == 'sleep':
time.sleep(susp_val[1])
except StopIteration as e:
return e.value
async def wait(s):
await my_sleep(1)
return s

使用此版本,wait按预期工作:

>>> drive(wait("hello world"))
'hello world'

这仍然不是很有用,因为驱动我们的协程的唯一方法是调用drive(),它再次支持单个协程。因此,我们不妨编写一个同步函数,该函数只需调用time.sleep()并调用一天。为了使我们的协程支持异步编程的用例,drive()需要:

  • 支持多个协程的运行和挂起
  • 在驱动循环中实现新协程的生成
  • 允许协程在与 IO 相关的事件上注册唤醒,例如文件描述符变得可读或可写 - 同时支持多个此类事件而不会降低性能

这就是 asyncio 事件循环带来的,以及许多其他功能。David Beazley 在这次演讲中出色地展示了从头开始构建事件循环,他在现场观众面前实现了功能性事件循环。

因此,经过一番挖掘,我想我找到了全局执行协程的最简单解决方案。

如果您>>> dir(coro)Python 将打印出以下属性:

['__await__', '__class__', '__del__', '__delattr__', '__dir__', '__doc__', '__eq__', '__format__', '__ge__', '__getattribute__', '__gt__', '__hash__', '__init__', '__init_subclass__', '__le__', '__lt__', '__name__', '__ne__', '__new__', '__qualname__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', '__str__', '__subclasshook__', 'close', 'cr_await', 'cr_code', 'cr_frame', 'cr_origin', 'cr_running', 'send', 'throw']

有几个属性很突出,即:

[
'__await__',
'close',
'cr_await',
'cr_code',
'cr_frame',
'cr_origin',
'cr_running',
'send',
'throw'
]

在阅读了yield(yield)的作用?以及生成器通常如何工作之后,我认为send方法一定是关键。

所以我尝试:

>>> the_actual_coro = coro()
<coroutine object coro at 0x7f5afaf55348> 
>>>the_actual_coro.send(None)

它引发了一个有趣的错误:

Original exception was:
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
StopIteration: a value

它实际上在异常中给了我返回值!

所以我认为一个非常基本的循环,嗯,它更像是一个运行器,可以这样实现:

def run(coro):
try:
coro.send(None)
except StopIteration as e:
return e.value

现在,我可以在同步函数中甚至全局运行协程,而不是建议这样做。但是,了解运行协程的最简单和最低级别很有趣

>>> run(coro())
'a value'

但是,当coro有等待时,这会返回None(这实际上是协程的本质)。

我认为这可能是因为事件循环通过将协程(coro.cr_frame.f_locals)分配给期货并单独处理它们来处理其协程()的可伸缩性? 我的简单run函数显然没有提供。在这方面,我可能是错的。因此,如果我错了,请有人纠正我。

最新更新