Concurrency improvements: Don’t use the main reactor thread for MQTT message processing. Use a different thread pool instead.

Author: amotl

CHANGES.rst

@@ -7,6 +7,13 @@ in progress
 ===========
 
 
+.. _kotori-0.13.0:
+
+2017-03-20 0.13.0
+=================
+- Concurrency improvements: Don’t use the main reactor thread for MQTT message processing. Use a different thread pool instead.
+
+
 .. _kotori-0.12.3:
 
 2017-03-20 0.12.3

kotori/daq/services/mig.py

@@ -3,10 +3,12 @@
 import time
 import json
 from bunch import Bunch
+from kotori.thimble import Thimble
 from twisted.logger import Logger, LogLevel
 from twisted.internet import reactor, threads
 from twisted.internet.task import LoopingCall
 from twisted.application.service import MultiService
+from twisted.python.threadpool import ThreadPool
 from kotori.configuration import read_list
 from kotori.daq.services import MultiServiceMixin
 from kotori.daq.intercom.mqtt import MqttAdapter
@@ -43,6 +45,10 @@ def setupService(self):
 
         self.registerService(self.mqtt_service)
 
+        # Perform MQTT message processing using a different thread pool
+        self.threadpool = ThreadPool()
+        self.thimble = Thimble(reactor, self.threadpool, self, ["process_message"])
+
     def startService(self):
         self.setupService()
         self.log(log.info, u'Starting')
@@ -66,7 +72,11 @@ def mqtt_receive(self, topic=None, payload=None, **kwargs):
             #return self.process_message(topic, payload, **kwargs)
 
             # Asynchronous message processing
-            deferred = threads.deferToThread(self.process_message, topic, payload, **kwargs)
+            #deferred = threads.deferToThread(self.process_message, topic, payload, **kwargs)
+
+            # Asynchronous message processing using different thread pool
+            deferred = self.thimble.process_message(topic, payload, **kwargs)
+
             deferred.addErrback(self.mqtt_receive_error, topic)
             return deferred
 

kotori/thimble.py

@@ -0,0 +1,61 @@
+"""Implementation of a Twisted-friendly thread pool wrapper."""
+from functools import partial
+from twisted.internet.threads import deferToThreadPool
+from twisted.internet.defer import fail
+from twisted.internet.error import ReactorNotRunning
+
+# Shamelessly stolen from:
+# https://github.com/lvh/thimble
+# https://pypi.python.org/pypi/thimble
+
+class Thimble(object):
+
+    """A Twisted thread-pool wrapper for a blocking API."""
+
+    def __init__(self, reactor, pool, wrapped, blocking_methods):
+        """Initialize a :class:`Thimble`.
+        :param reactor: The reactor that will handle events.
+        :type reactor: :class:`twisted.internet.interfaces.IReactorThreads` and
+            :class:`twisted.internet.interfaces.IReactorCore`. Pretty much any
+            real reactor implementation will do.
+        :param pool: The thread pool to defer to.
+        :type pool: :class:`twisted.python.threadpool.ThreadPool`
+        :param wrapped: The blocking implementation being wrapped.
+        :param blocking_methods: The names of the methods that will be wrapped
+            and executed in the thread pool.
+        :type blocking_methods: ``list`` of native ``str``
+        """
+        self._reactor = reactor
+        self._pool = pool
+        self._wrapped = wrapped
+        self._blocking_methods = blocking_methods
+
+    def _deferToThreadPool(self, f, *args, **kwargs):
+        """Defer execution of ``f(*args, **kwargs)`` to the thread pool.
+        This returns a deferred which will callback with the result of
+        that expression, or errback with a failure wrapping the raised
+        exception.
+        """
+        if self._pool.joined:
+            return fail(
+                ReactorNotRunning("This thimble's threadpool already stopped.")
+            )
+        if not self._pool.started:
+            self._pool.start()
+            self._reactor.addSystemEventTrigger(
+                'during', 'shutdown', self._pool.stop)
+
+        return deferToThreadPool(self._reactor, self._pool, f, *args, **kwargs)
+
+    def __getattr__(self, attr):
+        """Get and maybe wraps an attribute from the wrapped object.
+        If the attribute is blocking, it will be wrapped so that
+        calling it will return a Deferred and the actual function will
+        be ran in a thread pool.
+        """
+        value = getattr(self._wrapped, attr)
+
+        if attr in self._blocking_methods:
+            value = partial(self._deferToThreadPool, value)
+
+        return value