aiogevent event loop "fails" to track greenlets

Question

I recently stumbled onto a problem of mixing up gevent and asyncio-based code, because some synchronous libraries work just fine when I monkey-patch them with gevent.monkey.patch_all(). I found the aiogevent library, which seems to help by implementing the PEP 3156, and replace asyncio event loop with another implementation of your choice (in this case it's gevent). Last significant commits to git repository I found was made 4 years ago. After fixing setup.py, I managed to successfully install it, but problem is that it doesn't pass all the test.

One of those tests is test_soon, which spawns greenlet that should do operation and stop the loop. This test hangs forever, because loop.stop() doesn't have any effect on the loop, which is expected to stop when all tasks are finished. I wrote two snippets to check if it happens with traditional coroutines, and another with greenlets via gevent.spawn.

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

And with 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

And the question: what could possibly go wrong here? I clearly see that greenlet runs after I start the loop, but it is "untracked" by the loop? I can't find the exact line in asyncio sources which corresponds to this mechanism, and the same for gevent - I'm not quite familiar with internals of those modules and searching through them is confusing, but I want to know what is different and what changes to aiogevent's event loop must be done in order to pass the tests.

upd1: to emphasize the problem, gevent.hub.Hub doesn't have "public" handles to stop the loop, only those that should destroy it completely (gevent.hub.get_hub().destroy() currently has no effect, and trying to join the hub greenlet fails if not called within MAIN greenlet). It does have internal exception that is raised somewhere when loop exits (gevent.exceptions.LoopExit). My thought was to find the way of how to catch this exception, and tie it with run_forever, but no results yet.

Answer 1

Taking a guess at this, but I think you're using the event loop wrong for asyncio. I don't know enough about gevent to understand what type of object it's creating to know if this will work. The documentation would indicate that the order in which stop() and run_forever() are called is relevant to the operational call stack, reading as follows:

If stop() is called before run_forever() is called, the loop will poll the I/O selector once with a timeout of zero, run all callbacks scheduled in response to I/O events (and those that were already scheduled), and then exit.

If stop() is called while run_forever() is running, the loop will run the current batch of callbacks and then exit. Note that new callbacks scheduled by callbacks will not run in this case; instead, they will run the next time run_forever() or run_until_complete() is called.

I'm assuming that this option is being used to control the status of the asynchronous event, but you may find that run_until_complete() may be the better option for you and it may be something like this:

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()

I added shutdown_asyncgens() incase you are awaiting a yield response which would cause the close() method to hang/fail. I don't have any of these installed to test with, so let me know if this was helpful.

Answer 2

Why don't you just use a while loop like this:

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

This will print bloop wait one second then print it again (because 1 = 1), please forgive me if I'm wrong, and you have to use the function you were using, but I gave it my best shot.