Allow passing asyncio_loop argument to AsyncIOLoop

[minrk]

allows patterns of creating and explicitly passing the asyncio loop before instantiating the IOLoop.

For example: creating a loop with the non-default event loop policy without having to set the current policy

This is actually related to #3156 where in some of our explicit background thread event loop setup, we've been creating the asyncio event loop prior to the tornado IOLoop. To avoid relying on the deprecated asyncio.get_event_loop(), we need to pass the asyncio loop directly to the tornado IOLoop.

[minrk]

Not sure if you want this in 6.2 or hold for 6.3. I think it's likely to be useful for folks like me trying to deal with the new deprecations, though.

[bdarnell]

This should definitely go into 6.2 if we do it at all.

[graingert]

this opens up a situation where you can have two IOLoop instances for the same asyncio loop - which causes IOLoop.current() to change value inside a running loop

import asyncio
from tornado.ioloop import IOLoop


async def main():
    loop = asyncio.get_running_loop()
    io_loop = IOLoop.current()
    other_loop = IOLoop(asyncio_loop=loop, make_current=False)
    print(f"{loop=}\n{io_loop=}\n{other_loop=}\n{IOLoop.current()=}")


asyncio.run(main())

loop=<_UnixSelectorEventLoop running=True closed=False debug=False>
io_loop=<tornado.platform.asyncio.AsyncIOMainLoop object at 0x7fac54b4b940>
other_loop=<tornado.platform.asyncio.AsyncIOLoop object at 0x7fac549efc40>
IOLoop.current()=<tornado.platform.asyncio.AsyncIOLoop object at 0x7fac549efc40>

I think this should raise if there's already a IOLoop in the IOLoop._ioloop_for_asyncio dictionary for the loop passed into asyncio_loop, but I'm not sure if this can be made threadsafe

[minrk]

That condition already exists if you instantiate AsyncIOMainLoop multiple times:

import asyncio
from tornado.platform.asyncio import AsyncIOMainLoop

async def main():
    loop1 = AsyncIOMainLoop()
    loop2 = AsyncIOMainLoop()
    assert loop1.asyncio_loop is asyncio.get_running_loop()
    assert loop1.asyncio_loop is loop2.asyncio_loop
    assert loop1 is not loop2


if __name__ == "__main__":
    asyncio.run()

I'm not sure such the check needs to be perfect, or indeed threadsafe, since it would require user code to explicitly pass a single asyncio loop to two IOLoop constructors in different threads. The most likely case is multiple paths to initialization, which accidentally get taken twice, which should be handled by the check I've added here.

[bdarnell]

The way I solved this in tornado/testing.py was to use asyncio_loop.run_until_complete to activate the asyncio loop and construct a tornado loop with IOLoop.current().

This does seem a little nicer - the run_until_complete solution is cumbersome and non-obvious. But on the other hand run_until_complete avoids questions about creating multiple tornado loops for the same asyncio loop and who closes which loop.

What do you think about the run_until_complete alternative? Do you still think it's worth creating a new variation of the constructor?

[bdarnell]

This should definitely go into 6.2 if we do it at all.

[graingert]

The way I solved this in tornado/testing.py was to use asyncio_loop.run_until_complete to activate the asyncio loop and construct a tornado loop with IOLoop.current().

This does seem a little nicer - the run_until_complete solution is cumbersome and non-obvious. But on the other hand run_until_complete avoids questions about creating multiple tornado loops for the same asyncio loop and who closes which loop.

What do you think about the run_until_complete alternative? Do you still think it's worth creating a new variation of the constructor?

This would work if there was a public way to reset the threading ident

def reset_threading_ident():
    asyncio.get_running_loop()
    IOLoop.curent()._threading_identity = threading.get_ident()

async def call(fn):
    return fn()

class IOLoopInThread(Thread):
    def __init__(self):
        runner = asyncio.Runner()
        self._event = asyncio.Event() # LoopBoundMixin in 3.10
        self._io_loop = runner.run(call(IOLoop.current))

    def run(self):
        with self._runner as runner:
            runner.run(call(reset_threading_ident))
            runner.run(self._event.wait())

    def stop(self):
        self._runner.get_loop().call_soon_threadsafe(self._event.set)
        self.join()

[bdarnell]

Ah, right, the thread imprinting is a good reason to avoid run_until_complete, so this PR looks like the right solution to me.