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.

Answer 2

There are a few factors at play in your example. Short answer: study this documentation on threads in Twisted.

• While you don't have to use Twisted's reactor to use protocol classes (threading and protocol implementation are decoupled), you have called reactor.run so all of the below I consider applicable to you.
• Let Twisted create threads for you. Going outside the framework can get you in trouble. There are no "public" APIs for IPC messaging with the reactor (I think), so if you use Twisted, you pretty much need to go all the way.
• By default, Twisted does not switch threads to call your callbacks. To delegate to a worker thread from the main reactor thread (i.e. to perform blocking I/O), you don't have to create a thread yourself, you use reactor.callInThread and it will run in a worker thread. If you never do this, everything runs in the main reactor thread, meaning for example any I/O operations will block the reactor thread and you can't receive any events until your I/O completes.
• Code running in worker threads should use reactor.callFromThread to do anything that is not thread-safe. Provide a callback, which will run in the main reactor thread. You're better safe than sorry here, trust me.
• All of the above applies to Deferred processing also. So don't be afraid to use partial(reactor.callFromThread, mycallback) or partial(reactor.callInThread, mycallback) instead of simply mycallback when setting up callbacks. I learned that the hard way; without that, I found that any blocking I/O that I might do in deferred callbacks was either erroring out (due to thread safety issues) or blocking the main thread.

If you're just starting out in Twisted, it's a bit of a "trust fall". Learn to let go of managing your own threads and passing messages via Queue objects, etc. Once you figure out how Deferred and the reactor work (it's called "Twisted" for a reason!), it will seem perfectly natural to you. Twisted does force you to decouple and separate concerns in a functional programming style, but once you're over that, I've found that it's very clean and works well.

One tip: I wrote some decorators to use on all my callback functions so that I didn't have to be constantly calling callInThread and callFromThread and setting up Deferred for exception handling callbacks throughout the code; my decorators enable that behavior for me. It's likely prevented bugs from forgetting to do that, and it's certainly made Twisted development more pleasant for me.