Tornado & Thread-safe websocket handler

Question

I'm developping a web socket application using Tornado. I need to handle simultenous connections and be able to process requests in parallel.

Yes, by default Tornado isn't thread-safe, so I have to do it by myself in my application. The documentation doesn't say a lot on thread-safe websocket handlers, so I'd like to ask some feed-back from you who have already experimented with it.

UPDATE
Here is a minimal piece of code describing the problem:

handler.py

# -*- coding: utf-8 -*-
import re, zmq, json, sys, time
import tornado.ioloop
import tornado.web
import tornado.websocket

from subprocess import Popen, PIPE, STDOUT
from tornado.options import define, options


define("port", default=9000, help="run on the given port", type=int)

clients = []


class JobProgressHandler(tornado.websocket.WebSocketHandler):
    def open(self, *args):
        if self not in clients:
            clients.append(self)

        self.stream.set_nodelay(True)        

    def on_message(self, script_uid):
        print 'Starting verbose for : ', script_uid
        self.send_progress(script_uid)

    def on_close(self):
        print 'Closing: ', self
        if self in clients:
            clients.remove(self)

    def send_progress(self, script_uid):
        """
        Send a fake job progress
        """
        for x in xrange(10):
            self.write_message(json.dumps({script_uid: (x+1)*10}))
            #time.sleep(1)
            print script_uid, (x+1)*10

server.py

# -*- coding: utf-8 -*-
import tornado.web
from tornado.options import options, parse_command_line
from handler import JobProgressHandler


app = tornado.web.Application([
    (r'/jobprogress', JobProgressHandler),
])


if __name__ == '__main__':
    parse_command_line()
    app.listen(options.port)
    tornado.ioloop.IOLoop.instance().start()

test.js (tests in browser's console)

function websock(script_name)
{
    var ws = new WebSocket("ws://localhost:9000/jobprogress");
    ws.onopen = function(){ws.send(script_name); console.log("Socket opened");};
    ws.onmessage = function (evt){console.log(evt.data);};
    ws.onclose = function(){console.log("Connection is closed ...");};
}


websock('diagnostics.eth0');
websock('diagnostics.eth1');

Answer 1

Why do you need threads? The reason tornado is not thread-safe is because it is designed for event-driven concurrency within a single thread; you don't need threads to support many simultaneous connections. If you do combine threads with tornado you need to be very careful about the handoff between threads since nearly all tornado methods can only be safely called from the IOLoop thread.

Answer 2

In the code below, you add a callback in the IOLoop telling it to run your method, when it can. It will then update the WS client, and add another callback for itself to execute 1 second later. As far as I know the time specified may not be the exact time taken.

What it essentially does is it tells the IOLoop "go ahead and process other events, and in about 1 second, can you call my function". The other events I mention above could be anything. It could be servicing new open connection requests from other WS clients, it could be other callbacks to update their handlers, etc.

On a side note, I strongly suggest you read up a little on Python async concepts and programming.

Note, the code below wasn't tested, but it should work fine. Let me know if you have to make alterations to get it working.

from functools import partial
import datetime

def send_update(self, script_uid, max=10):
    self.write_message(json.dumps({script_uid: (script_uid+1)*10}))
    if script_uid > max:
        return
    io_loop.add_timeout(datetime.timedelta(seconds=1), partial(self.send_update, script_uid + 1))