Btukfyl

I am writing a Tornado webserver in Python 3.7 to display the status of processes run by the multiprocessing library.

The following code works, but I'd like to be able to do it using Tornado's built-in library instead of hacking in the threading library. I haven't figured out how to do it without blocking Tornado during queue.get. I think the correct solution is to wrap the get calls in some sort of future. I've tried for hours, but haven't figured out how to do this.

Inside of my multiprocessing script:

class ProcessToMonitor(multiprocessing.Process)



def __init__(self):

    multiprocessing.Process.__init__(self)

    self.queue = multiprocessing.Queue()



def run():

    while True:

        # do stuff

        self.queue.put(value)

Then, in my Tornado script

class MyWebSocket(tornado.websocket.WebSocketHandler):

    connections = set()



    def open(self):

        self.connections.add(self)



    def close(self):

        self.connections.remove(self)



    @classmethod

    def emit(self, message):

        [client.write_message(message) for client in self.connections]



def worker():

    ptm = ProcessToMonitor()

    ptm.start()

    while True:

        message = ptm.queue.get()

        MyWebSocket.emit(message)



if __name__ == '__main__':

    app = tornado.web.Application([

        (r'/', MainHandler), # Not shown

        (r'/websocket', MyWebSocket)

    ])

    app.listen(8888)



    threading.Thread(target=worker)



    ioloop = tornado.ioloop.IOLoop.current()

    ioloop.start()

queue.get isn't a blocking function, it just waits until there's an item in the queue in case the queue is empty. I can see from your code that queue.get fits perfectly for you use case inside a while loop.

I think you're probably using it incorrectly. You'll have to make the worker function a coroutine (async/await syntax):

async def worker():

    ...

    while True:

        message = await queue.get()

        ...

However, if you don't want to wait for an item and would like to proceed immediately, its alternative is queue.get_nowait.

One thing to note here is thatqueue.get_nowait will raise an exception called QueueEmpty if the queue is empty. So, you'll need to handle that exception.

Example:

while True:

    try:

        message = queue.get_nowait()

    except QueueEmpty:

        # wait for some time before

        # next iteration

        # otherwise this loop will

        # keep running for no reason



    MyWebSocket.emit(message)

As you can see, you'll have to use pause the while loop for some time if the queue is empty to prevent it from overwhelming the system.

So why not use queue.get in the first place?