Tornado: AssertionError: assert loop is None or isinstance(loop, AbstractEventLoop)

[loretoparisi]

I get this error when running Tornado in a thread and using AnyThreadEventLoopPolicy.

class WebServer(threading.Thread):
    def run(self):
        PORT = os.getenv('PORT', 8888)

        #asyncio.set_event_loop(asyncio.new_event_loop())
        asyncio.set_event_loop(AnyThreadEventLoopPolicy())

        app = application()
        app.listen(PORT)
        tornado.ioloop.IOLoop.instance().start()
WebServer.start()

Full stack trace:

Exception in thread Thread-1:
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/threading.py", line 926, in _bootstrap_inner
    self.run()
  File "tornadoaas.py", line 101, in run
    asyncio.set_event_loop(AnyThreadEventLoopPolicy())
  File "/usr/local/lib/python3.7/asyncio/events.py", line 757, in set_event_loop
    get_event_loop_policy().set_event_loop(loop)
  File "/usr/local/lib/python3.7/asyncio/unix_events.py", line 1119, in set_event_loop
    super().set_event_loop(loop)
  File "/usr/local/lib/python3.7/asyncio/events.py", line 651, in set_event_loop
    assert loop is None or isinstance(loop, AbstractEventLoop)
AssertionError

[bdarnell]

There are two separate functions: asyncio.set_event_loop takes an EventLoop, asyncio.set_event_loop_policy takes a policy. You need to use the latter with AnyThreadEventLoopPolicy.

[bdarnell]

But after seeing the comments you've posted on some other issues, you don't actually want AnyThreadEventLoopPolicy. That thread makes it possible to have event loops on multiple threads, but it's very rare that that's actually a good idea. When mixing threads and async code, you have to make a distinction between the event loop thread (almost always just one) and the executor's threads. You can't call tornado methods from executor threads, which means you can't generally use something like @run_on_decorator on the get() method.

When using Tornado, you should do as much as possible on the event loop thread, and move only small pieces of code to other threads with executors. (conversely, if you prefer to use multiple threads, you should probably be using a framework other than Tornado).

[loretoparisi]

@bdarnell thank you! fixed with

class WebServer(threading.Thread):
    def run(self):
        PORT = os.getenv('PORT', 8888)
        asyncio.set_event_loop_policy(AnyThreadEventLoopPolicy())
        app = application()
        app.listen(PORT)
        tornado.ioloop.IOLoop.instance().start()
WebServer.start()

Thanks for the clarification! In fact I have realized that and I have fixed it using something like

class MyHeavyHandler(tornado.web.RequestHandler):
    executor = BoundedThreadPoolExecutor(max_workers=2)
    @tornado.concurrent.run_on_executor
    def get(self, *args):
       # cpu bound task

where a BoundedThreadPoolExecutor customization is available here.

Before that, I had that the MyHeavyHandler wit the @tornado.web.asynchronous decorator, and this was blocking the main thread, so any other lightweight calls were simple hanging and waiting it came to end i.e. until MyHeavyHandler handler called the self.finish().

Finally, my last attempt, that was successful on the stack Docker/macOS (host) but not working on Docker/Debian was the following base handler:

class BaseHandler(tornado.web.RequestHandler):

    @tornado.gen.coroutine
    def post_async(self, *args):
        resp_dict = yield ioloop.IOLoop.current().run_in_executor(None, self.post_handler, args)
        return resp_dict

    @tornado.web.asynchronous
    @tornado.gen.coroutine
    def post(self, *args):

        response = yield self.post_async(args)

        self.set_header("Content-Type", "application/json; charset=UTF-8")

        status_code = response.getField('status_code', 200)
        self.set_status(status_code)
        self.write(json.dumps(response.getData()))
        self.finish()

    @tornado.gen.coroutine
    def get_async(self, *args):
        resp_dict = yield ioloop.IOLoop.current().run_in_executor(None, self.get_handler, args)
        return resp_dict

    @tornado.web.asynchronous
    @tornado.gen.coroutine
    def get(self, *args):

        response = yield self.get_async(args)

        self.set_header("Content-Type", "application/json; charset=UTF-8")

        status_code = response.getField('status_code', 200)
        self.set_status(status_code)
        self.write(json.dumps(response.getData()))
        self.finish()

With this approach, that I have adapted from another issue on this repo, I was then able to do like

class MyHeavyHandler(tornado.web.RequestHandler):
    def get_handler(self, *args):
       # cpu bound task

having the call stack BasicHandler#get -> BasicHandler#get_async -(executor)-> MyHeavyHandler#get_handler)->finish()

This approach, when Docker/macOS was actually working pretty fine for all kind of handlers (cpu bound or I/O bound or lightweight ones), with any blocking issues. So far I don't get why on Docker/Debian it does not work (I don't get any response from the get method of the BasicHandler so the get_handler is not called at all. Maybe you have a tip (thank you in advance).

[ploxiln]

For your purposes, you don't really want to use the AnyThreadEventLoopPolicy or @tornado.web.asynchronous - just use @tornado.gen.coroutine.

I would simplify to something like this:

from concurrent.futures import ThreadPoolExecutor

import tornado

class MyHeavyHandler(tornado.web.RequestHandler):
    executor = ThreadPoolExecutor(2)

    @tornado.concurrent.run_on_executor
    def get_work(args):
        # cpu bound task, do not use any tornado objects or functions

    @tornado.gen.coroutine
    def get(self, *args):
        response = yield self.get_work(args)
        self.write(response)

EDIT: By the way, calling self.write() with a python dict will result in json output and content-type header automatically, and you don't need self.finish() if you do not use @tornado.web.asynchronous

EDIT2: I see that BoundedThreadPoolExecutor is your own creation - might as well use a plain ThreadPoolExecutor, it is also bounded