Call a twisted protocol method from another thread

Question

I have made a Home Security program in Python that uses Raspberry Pi's GPIOs to sense movement and actuate the siren. The users activate/deactivate the system using a NFC Tag to a nfc reder connected also to raspberry pi.

For this I need to constantly check for nfc tags in a non blocking manner and at the same time constantly check the sensors for movement also non blocking. I need some more parallel stuff to do but I think these two are enough to make my point.

Right now I use threads which I start/stop like this - Stopping a thread after a certain amount of time - I'm not sure if this is the optimal way but as of now the system works fine.

Now I want to extend its functionality to offer notifications through websockets. I found that this can be done with Twisted but I am confused..

Here is an example code of how I am trying to do it:

from twisted.internet import reactor
from autobahn.websocket import WebSocketServerFactory, \
                               WebSocketServerProtocol, \
                               listenWS


def thread1(stop_event):
    while(not stop_event.is_set()):
        stop_event.wait(4)
        print "checking sensor"
        # sensor_state = GPIO.input(11)
        if sensor_state == 1:
            # how can I call send_m("sensor detected movement")  #<---
            t1_stop_event.set()

t1_stop_event = Event()
t1 = Thread(target=thread1, args=(t1_stop_event,))

class EchoServerProtocol(WebSocketServerProtocol):
   def onMessage(self, msg, binary):
    print "received: "+msg
    print "stopping thread1"
    t1_stop_event.set()

   def send_m(self, msg):
    self.sendMessage(msg)

if __name__ == '__main__':
   t1.start()
   factory = WebSocketServerFactory("ws://localhost:9000")
   factory.protocol = EchoServerProtocol
   listenWS(factory)
   reactor.run()

So how can I call the send method of the server protocol from a thread like the thread1?

Answer 1

As is often the case, the answer to your question about threads and Twisted is "don't use threads".

The reason you're starting a thread here appears to be so you can repeatedly check a GPIO sensor. Does checking the sensor block? I'm guessing not, since if it's a GPIO it's locally available hardware and its results will be available immediately. But I'll give you the answer both ways.

The main thing you are using threads for here is to do something repeatedly. If you want to do something repeatedly in Twisted, there is never a reason to use threads :). Twisted includes a great API for recurring tasks: LoopingCall. Your example, re-written to use LoopingCall (again, assuming that the GPIO call does not block) would look like this:

from somewhere import GPIO

from twisted.internet import reactor, task
from autobahn.websocket import WebSocketServerFactory, \
                               WebSocketServerProtocol, \
                               listenWS

class EchoServerProtocol(WebSocketServerProtocol):

    def check_movement(self):
        print "checking sensor"
        sensor_state = GPIO.input(11)
        if sensor_state == 1:
            self.send_m("sensor detected movement")

    def connectionMade(self):
        WebSocketServerProtocol.connectionMade(self)
        self.movement_checker = task.LoopingCall(self.check_movement)
        self.movement_checker.start(4)

    def onMessage(self, msg, binary):
        self.movement_checker.stop()

    def send_m(self, msg):
        self.sendMessage(msg)

if __name__ == '__main__':
   factory = WebSocketServerFactory("ws://localhost:9000")
   factory.protocol = EchoServerProtocol
   listenWS(factory)
   reactor.run()

Of course, there is one case where you still need to use threads: if the GPIO checker (or whatever your recurring task is) needs to run in a thread because it is a potentially blocking operation in a library that can't be modified to make better use of Twisted, and you don't want to block the main loop.

In that case, you still want to use LoopingCall, and take advantage of another one of its features: if you return a Deferred from the function that LoopingCall is calling, then it won't call that function again until the Deferred fires. This means you can shuttle a task off to a thread and not worry about the main loop piling up queries for that thread: you can just resume the loop on the main thread automatically when the thread completes.

To give you a more concrete idea of what I mean, here's the check_movement function modified to work with a long-running blocking call that's run in a thread, instead of a quick polling call that can be run on the main loop:

def check_movement(self):
    from twisted.internet.threads import deferToThread
    def get_input():
        # this is run in a thread
        return GPIO.input(11)
    def check_input(sensor_state):
        # this is back on the main thread, and can safely call send_m
        if sensor_state == 1:
            self.send_m("sensor movement detected")
    return deferToThread(get_input).addCallback(check_input)

Everything else about the above example stays exactly the same.