Local Socket + Tornadio Multi-Threading

[Zach Taylor]

I'm writing an application that connects to a microcontroller over a
TCP/IP socket and then ideally would broadcast the responses from that
device to a website in realtime. I'm having trouble connecting the
socket thread to the broadcast (main) thread.

The basic program is based on:
- tornadio's chatroom example for broadcasting and
- a separate thread that connects to the device and reads data from
it as it comes in.

The code is below. I would be up for moving the Arduino-reading code
into the ioloop but I can't figure out how. I tried making the
participants set global but that ran into threading exceptions when
iterating through it to send. I tried locking before accessing the
participants set when it was global, but that didn't help. How can I
get the info from the ArduinoProcessor class out to the tornadio
connections? Unfortunately, the Arduino stream is only local, so this
might be tricky to test.

Thanks!

Tornado Code:

import logging
import tornado.httpserver
import tornado.ioloop
import tornado.web
from tornado.options import define, options
import os

import threading
import simplejson as json

import tornadio
import tornadio.router
import tornadio.server

# Define options that can be changed as we run this via the command
line
define("port", default=7777, help="Run server on a specific port",
type=int)

participantLock = threading.Lock()

class ArduinoProcessor(threading.Thread):
def run(self):
import socket
import simplejson

logging.info("Connecting to Arduino")

#The Arduino microcontroller at this address sends a constant
stream of JSON, kind of like Twitter
host = '135.124.11.2'
port = 80

sock = socket.socket()
sock.connect((host,port))

file = sock.makefile("rb")

while True:
datum = file.readline()
if len(datum) > 2:
t = json.loads(datum)
if t:
#I want to send this data to a function like
BroadcastConnection.emit
logging.info("[arduino] %s" % t)

#code doesn't really get here
logging.info("Closing the firehose")
sock.close()

class IndexHandler(tornado.web.RequestHandler):
"""Regular HTTP handler to serve the chatroom page"""
def get(self):
self.render("templates/chat.html")

class BroadcastConnection(tornadio.SocketConnection):
#I tried making participants a global variable but when iterating
through it to send, I ran into errors
participants = set()

def on_open(self, *args, **kwargs):
participantLock.acquire()
self.participants.add(self)
participantLock.release()

#on_message broadcasts out messages like in a normal chatroom
def on_message(self, message):
logging.info("[echo] %s" % message)

participantLock.acquire()
for p in self.participants:
p.send("[echo] %s" % message)
participantLock.release()

def on_close(self):
participantLock.acquire()
self.participants.remove(self)
participantLock.release()
logging.info("Closing Socket.IO Client for protocol 'x'")

#this never gets called
def emit(self, msg):
logging.info(msg)
participantLock.acquire()
for p in self.participants:
p.send(msg)
participantLock.release()

local_static_path = os.path.join(os.path.dirname(__file__), "static")

broadcastRouter = tornadio.get_router(BroadcastConnection)

application = tornado.web.Application([
(r"/", IndexHandler), broadcastRouter.route()],
static_path=local_static_path, socket_io_port=7777,
enabled_protocols=['websocket', 'xhr-multipart', 'xhr-
polling', 'jsonp-polling', 'htmlfile'])

if __name__ == "__main__":

http_server = tornado.httpserver.HTTPServer(application)
tornado.options.parse_command_line()
http_server.listen(options.port)

ap = ArduinoProcessor()
ap.start()

ioloop = tornado.ioloop.IOLoop().instance()
ioloop.start()

-------------------------------------

chat.html script using jquery and socket.io:

window.onload = function() {

var s = new io.Socket("172.16.10.65", {port: 7777});
s.connect();

s.addEvent('connect', function() {
s.send('Connected');
});

s.addEvent('message', function(data) {
$("#chat").prepend("<div>" + data + "</div>");
});

//send the message when submit is clicked
$('#chatform').submit(function (evt) {
var line = $('#chatform [type=text]').val()
$('#chatform [type=text]').val('')
s.send(line);
return false;
});
};

[Sree]

Hi Zach,
I am working on the same concept and having the same issues. If you have
got the solution then please help me out. I would be grateful to you. Thanks in
advance.

Thanks,
Sree

[Zach Taylor]

Hi Sree,

I wound up restructuring the program so I wouldn't have to use multiple threads.  It was turning into too much of a headache.  Good luck though, and I'd love to hear if you make more progress.

--Zach

Zach Taylor | Digital Creatologist | mcgarrybowen | 601 W 26th Street, 11th Floor | New York, NY 10001 | T: 212.979.4989 | mcgarrybowen.com | facebook.com/mcgarrybowen | twitter.com/mcgarrybowen

----------------------------------------------------------------------------------------------
The contents of this message may be privileged and confidential. Therefore, if this message
has been received in error, please delete it without reading it. All contents of the message,
including any attachments, are the copyright property of the sender. This message cannot in
any way bind Dentsu McGarry Bowen, L.L.C. to any contract or other obligation.

[Serge S. Koval]

Hi,

 There's no problem to use threads in the Tornado (or Tornadio), but you have to follow some rules:

https://github.com/facebook/tornado/wiki/Threading-and-concurrency 

 Anyway, if you want to use thread to poll your device, that's fine, but you have to separate polling thread from your Tornado application by using io_loop.add_callback() to send data to main thread.

 So, you can do something like:

1. Declare your TornadIO connection class:

class MyConnection(SocketConnection):

    participants = set()

    ...

    @classmethod

    def broadcast(cls, msg):

        for p in cls.participants:

            p.send(msg)

    @classmethod

    def controller_msg(cls, msg):

        cls.broadcast(msg)

2. In your device polling thread, do something like:

    while True: 

            datum = file.readline() 

            if len(datum) > 2: 

                t = json.loads(datum) 

                ...

                def callback():

                    MyConnection.controller_msg(t)

                io_loop.add_callback(callback)

3. ...

4. Profit!

Basically, callback() will be executed in Tornado main thread and will call TornadIO handler class, which will then broadcast message to everyone else.

Hope this helps.

Serge.