Tornado: Spawn a thread and communicate between the thread and tornado app?

[shamrin]

My Tornado app needs to have bidirectional communicate with the worker thread spawned by .run_in_executor. Starting a worker thread takes time. It needs to initialise an expensive network connection before it can do the work.

Is there a native Tornado approach to solve this problem?

Approach 1: tornado.queue

I've looked into tornado.queues module, but I couldn't figure out how to use it to communicate between sync and async worlds. Documentation suggests wrapping operations with IOLoop.add_callback:

https://github.com/tornadoweb/tornado/blob/18b653cf93fe870cbc630fb48595e1b92fc1e06c/tornado/queues.py#L21-L24

However, .add_callback is useful only for .put and .put_nowait methods, and not for .get and .get_nowait. There's no way to get the result back from .add_callback.

Approach 2: standard queue library

Next option is queue from Python standard library. It should work, but all queue operation needs to be wrapped with .run_in_executor. Seems excessive and not pretty.

Approach 3: janus library

There's janus - library for mixed sync/async queue. Looks good, but it's not small and it's an extra dependency to maintain.

Approach 4: Tornado native approach?

Is there Tornado-native approach? For example something similar to trio.from_thread.run. It's the same as IOLoop.add_callback, but Trio's run also sends back the result of the operation. Thus run can safely wrap Trio queue methods (called "memory channel" in Trio). Example from Trio documentation:

def thread_fn(receive_from_trio, send_to_trio):
    while True:
        # Since we're in a thread, we can't call methods on Trio
        # objects directly -- so we use trio.from_thread to call them.
        try:
            request = trio.from_thread.run(receive_from_trio.receive)
        except trio.EndOfChannel:
            trio.from_thread.run(send_to_trio.aclose)
            return
        else:
            response = request + 1
            trio.from_thread.run(send_to_trio.send, response)

[mivade]

Using queues is definitely an option, but as you have described it can be tricky to synchronize between the sync and async versions (if you look at how Janus works, it's pretty straightforward but requires a bunch of synchronization primitives to ensure thread safety). One option you are missing that would fit naturally with a Tornado application is to use sockets rather than queues. Something like this:

from socketserver import TCPServer, StreamRequestHandler

from tornado.ioloop import IOLoop
from tornado.tcpclient import TCPClient


class EchoHandler(StreamRequestHandler):
    def handle(self):
        data = self.rfile.readline()
        print(f"sync_worker received {data}")
        self.wfile.write(data)


def sync_worker(port):
    print("starting sync_worker...")

    with TCPServer(("127.0.0.1", port), EchoHandler) as server:
        server.handle_request()

    print("exiting sync_worker")


async def main():
    port = 8975
    client = TCPClient()
    future = IOLoop.current().run_in_executor(None, sync_worker, port)
    stream = await client.connect("127.0.0.1", port, timeout=1)
    await stream.write(b"echo this\n")
    result = await stream.read_until(b"\n")
    print(f"main received {result}")
    await future


if __name__ == "__main__":
    loop = IOLoop.current()
    loop.run_sync(main)

You may also want to look into something like ZeroMQ which makes a lot of messaging patterns easy and natively supports Tornado/asyncio.

[ploxiln]

If the python-stdlib (thread-type) queue is un-bounded, then putting a message to it will not block, so you can put a message to that queue from the tornado ioloop context, and receive it from the other thread.

So, you could use the python-stdlib queue to put from tornado context to the worker thread, and you could use the tornado queue with IOLoop.add_callback() to put from the worker thread back to the tornado ioloop context. One queue for each way, different types.

[ploxiln]

(IOLoop.run_in_executor() re-uses threads from a pool, so it might not be so bad to use it for each potentially-blocking put or get of a python-stdlib queue from ioloop context. So that's another option, if you need blocking puts to bounded queues either way.)

[bdarnell]

However, .add_callback is useful only for .put and .put_nowait methods, and not for .get and .get_nowait. There's no way to get the result back from .add_callback.

That's true, but it shouldn't be a problem because each queue would only be consumed from one side of the divide. You'd use a Tornado queue to send things from sync to async (using add_callback(q.put_nowait), and a threaded queue.Queue to send from async to sync (using q.put_nowait()). This assumes that you're OK with unbounded queues; bounding them gets more complicated (this is what janus does, and why it's "not small")

It should work, but all queue operation needs to be wrapped with .run_in_executor. Seems excessive and not pretty.

Only blocking operations (put and get) need to be run in an executor; the nowait versions can be called without this.

For example something similar to trio.from_thread.run. It's the same as IOLoop.add_callback, but Trio's run also sends back the result of the operation

That's easy to do (assuming the function is a coroutine that returns a future):

def run_on_ioloop_from_thread(io_loop, f, *a, **kw):
    concurrent_future = concurrent.futures.Future()
    async def wrapper():
        try:
            result = await f(*a, **kw)
        except Exception as e:
            concurrent_future.set_exception(e)
        else:
            concurrent_future.set_result(result)
    io_loop.add_callback(wrapper)
    return concurrent_future.result()

So I think all of these options are viable, and it largely comes down to whether you want to treat the sync or async worlds as primary, and whether your work is best modeled as passing messages over queues or calling functions that return values.