aiogevent 事件循环“未能”跟踪 greenlets

Question

我最近偶然发现了混淆基于 gevent 和 asyncio 的代码的问题，因为当我使用 gevent.monkey.patch_all() 对它们进行猴子补丁时，一些同步库工作得很好。我找到了aiogevent 库，它似乎对实现 PEP 3156 有所帮助，并将 asyncio 事件循环替换为您选择的另一个实现（在本例中是 gevent）。我发现对 git 存储库的最后一次重大提交是在 4 年前进行的。修复 setup.py 后，我成功安装了它，但问题是它没有通过所有测试。

其中一个测试是test_soon，它会生成应该执行操作并停止循环的greenlet。这个测试永远挂起，因为loop.stop() 对循环没有任何影响，预计在所有任务完成后会停止。我写了两个 sn-ps 来检查它是否发生在传统协程中，另一个是通过 gevent.spawn 使用 greenlets。

import gevent
import aiogevent
import asyncio

asyncio.set_event_loop_policy(aiogevent.EventLoopPolicy())

loop = asyncio.get_event_loop()

async def func():
    print('bloop')
    loop.stop()

loop.create_task(func())
loop.run_forever() # works alright and stops as soon as func finish

还有gevent.spawn:

import gevent
import aiogevent
import asyncio

asyncio.set_event_loop_policy(aiogevent.EventLoopPolicy())

loop = asyncio.get_event_loop()

def func():
    print('bloop')
    loop.stop()

g = gevent.spawn(func)
loop.run_forever() # func is executed as soon as loop runs, but loop.stop() is ignored

还有一个问题：这里可能出现什么问题？我清楚地看到greenlet在我开始循环后运行，但它被循环“未跟踪”？我在 asyncio 源代码中找不到与此机制相对应的确切行，对于 gevent 也是如此 - 我对这些模块的内部不太熟悉，搜索它们很混乱，但我想知道有什么不同和必须对aiogevent 的事件循环进行哪些更改才能通过测试。

upd1：为了强调这个问题，gevent.hub.Hub 没有“公共”句柄来停止循环，只有那些应该完全破坏它的句柄（gevent.hub.get_hub().destroy() 目前没有效果，并且如果未在 MAIN greenlet 中调用，则尝试加入 hub greenlet 失败）。它确实有在循环退出时引发的内部异常 (gevent.exceptions.LoopExit)。我的想法是找到如何捕获这个异常的方法，并将其与 run_forever 绑定，但还没有结果。

Answer 1

对此进行猜测，但我认为您将事件循环用于 asyncio 是错误的。 我对 gevent 的了解还不够，无法理解它正在创建什么类型的对象以知道这是否会起作用。documentation 将表明 stop() 和 run_forever() 的调用顺序与操作调用栈相关，内容如下：

如果在调用 run_forever() 之前调用了 stop()，则循环将轮询 I/O 选择器一次，超时时间为零，运行所有为响应 I/O 事件而调度的回调（以及那些已经调度的回调） )，然后退出。

如果在 run_forever() 运行时调用了 stop()，则循环将运行当前批次的回调然后退出。请注意，回调安排的新回调在这种情况下不会运行；相反，它们将在下次调用 run_forever() 或 run_until_complete() 时运行。

我假设此选项用于控制异步事件的状态，但您可能会发现run_until_complete() 可能是您更好的选择，它可能是这样的：

import gevent
import aiogevent
import asyncio

asyncio.set_event_loop_policy(aiogevent.EventLoopPolicy())

loop = asyncio.get_running_loop()

async def func():
    await print('bloop') # await keyword not needed, just demonstrating
    loop.stop()

try:
    loop.run_until_complete(gevent.spawn(func))
finally:
    loop.run_until_complete(loop.shutdown_asyncgens())
    loop.close()

我添加了shutdown_asyncgens()，以防您正在等待导致close() 方法挂起/失败的yield 响应。我没有安装任何这些来测试，所以让我知道这是否有帮助。

Answer 2

为什么不直接使用这样的 while 循环：

from time import sleep
while(1 == 1):
  print('bloop')
  sleep(1)

这将打印 bloop 等待一秒钟然后再次打印（因为 1 = 1），如果我错了，请原谅我，你必须使用你正在使用的功能，但我已经尽力了。