Merge lp:~rvb/maas/rework-threadpool into lp:~maas-committers/maas/trunk

[Raphaël Badin]

Revert 3819 and 3815: use the default threadpool size and re-introduce the webcoket threadpool. An unbounded threadpool would end up exhausting the available DB connections.

[Blake Rouse (blake-rouse)]

review: Approve

=== modified file 'src/maasserver/plugin.py'
--- src/maasserver/plugin.py	2015-04-14 08:27:39 +0000
+++ src/maasserver/plugin.py	2015-04-16 07:48:27 +0000
@@ -38,10 +38,6 @@
 
     def makeService(self, options):
         """Construct a service."""
-        from twisted.internet import reactor
-        # This is a workaround until we implement proper unbounded threadpool
-        # with a cleanup method.
-        reactor.suggestThreadPoolSize(1024 * 10)
         # Get something going with the logs.
         from provisioningserver import logger
         logger.basicConfig()

=== modified file 'src/maasserver/tests/test_plugin.py'
--- src/maasserver/tests/test_plugin.py	2015-04-14 08:27:39 +0000
+++ src/maasserver/tests/test_plugin.py	2015-04-16 07:48:27 +0000
@@ -25,7 +25,6 @@
 from provisioningserver import logger
 from provisioningserver.utils.twisted import asynchronous
 from twisted.application.service import MultiService
-from twisted.internet import reactor
 
 
 class TestOptions(MAASTestCase):
@@ -73,7 +72,6 @@
             "rpc-advertise",
             "web",
         ]
-        self.assertEqual(1024 * 10, reactor.getThreadPool().max)
         self.assertItemsEqual(expected_services, service.namedServices)
         self.assertEqual(
             len(service.namedServices), len(service.services),

=== modified file 'src/maasserver/websockets/protocol.py'
--- src/maasserver/websockets/protocol.py	2015-04-14 20:37:30 +0000
+++ src/maasserver/websockets/protocol.py	2015-04-16 07:48:27 +0000
@@ -37,13 +37,15 @@
 from maasserver.websockets.listener import PostgresListener
 from maasserver.websockets.websockets import STATUSES
 from provisioningserver.utils.twisted import synchronous
+from twisted.internet import reactor
 from twisted.internet.defer import inlineCallbacks
 from twisted.internet.protocol import (
     Factory,
     Protocol,
 )
-from twisted.internet.threads import deferToThread
+from twisted.internet.threads import deferToThreadPool
 from twisted.python import log
+from twisted.python.threadpool import ThreadPool
 from twisted.web.server import NOT_DONE_YET
 
 
@@ -175,7 +177,9 @@
                 "Error authenticating user: %s" % failure.getErrorMessage())
             return None
 
-        d = deferToThread(self.getUserFromSessionId, session_id)
+        d = deferToThreadPool(
+            reactor, self.factory.threadpool,
+            self.getUserFromSessionId, session_id)
         d.addCallbacks(got_user, got_user_error)
 
         return d
@@ -254,7 +258,8 @@
         # performed. The execution of this method is defered to a thread
         # because it interacts with the database which is blocking.
         transactional_execute = transactional(handler.execute)
-        d = deferToThread(
+        d = deferToThreadPool(
+            reactor, self.factory.threadpool,
             transactional_execute, method, message.get("params", {}))
         d.addCallbacks(
             partial(self.sendResult, request_id),
@@ -300,22 +305,31 @@
 
 
 class WebSocketFactory(Factory):
+    """Factory for WebSocketProtocol.
+
+    :ivar threadpool: The thread-pool used for servicing websocket
+        requests.
+    """
 
     handlers = {}
     clients = []
 
     def __init__(self):
+        self.threadpool = ThreadPool(name=self.__class__.__name__)
         self.listener = PostgresListener()
         self.cacheHandlers()
         self.registerNotifiers()
 
     def startFactory(self):
-        """Start the listener."""
+        """Start the thread pool and the listener."""
+        self.threadpool.start()
         return self.listener.start()
 
     def stopFactory(self):
-        """Stop the listener."""
-        return self.listener.stop()
+        """Stop the thread pool and the listener."""
+        stopped = self.listener.stop()
+        self.threadpool.stop()
+        return stopped
 
     def getSessionEngine(self):
         """Returns the session engine being used by Django.
@@ -359,7 +373,8 @@
     def onNotify(self, handler_class, channel, action, obj_id):
         for client in self.clients:
             handler = handler_class(client.user, client.cache)
-            data = yield deferToThread(
+            data = yield deferToThreadPool(
+                reactor, self.threadpool,
                 self.processNotify, handler, channel, action, obj_id)
             if data is not None:
                 (name, data) = data

=== modified file 'src/maasserver/websockets/tests/test_protocol.py'
--- src/maasserver/websockets/tests/test_protocol.py	2015-04-14 20:37:30 +0000
+++ src/maasserver/websockets/tests/test_protocol.py	2015-04-16 07:48:27 +0000
@@ -47,12 +47,14 @@
 from testtools.matchers import (
     Equals,
     Is,
+    IsInstance,
 )
 from twisted.internet.defer import (
     fail,
     inlineCallbacks,
 )
 from twisted.internet.threads import deferToThread
+from twisted.python.threadpool import ThreadPool
 from twisted.web.server import NOT_DONE_YET
 
 
@@ -61,6 +63,8 @@
     def make_protocol(self, patch_authenticate=True, transport_uri=''):
         self.patch(protocol_module, "PostgresListener")
         factory = WebSocketFactory()
+        factory.startFactory()
+        self.addCleanup(factory.stopFactory)
         protocol = factory.buildProtocol(None)
         protocol.transport = MagicMock()
         protocol.transport.uri = transport_uri
@@ -521,9 +525,9 @@
         self.addCleanup(self.clean_node, node)
         protocol, factory = self.make_protocol()
         protocol.user = MagicMock()
-        mock_deferToThread = self.patch_autospec(
-            protocol_module, "deferToThread")
-        mock_deferToThread.return_value = fail(
+        mock_deferToThreadPool = self.patch_autospec(
+            protocol_module, "deferToThreadPool")
+        mock_deferToThreadPool.return_value = fail(
             maas_factory.make_exception("error"))
         message = {
             "type": MSG_TYPE.REQUEST,
@@ -561,6 +565,8 @@
 
     def make_protocol_with_factory(self, user=None):
         factory = WebSocketFactory()
+        factory.startFactory()
+        self.addCleanup(factory.stopFactory)
         protocol = factory.buildProtocol(None)
         protocol.transport = MagicMock()
         if user is None:
@@ -618,12 +624,31 @@
 
     @wait_for_reactor
     @inlineCallbacks
+    def test_startFactory_starts_threadpool(self):
+        factory = WebSocketFactory()
+        yield factory.startFactory()
+        try:
+            self.assertThat(factory.threadpool, IsInstance(ThreadPool))
+            self.expectThat(factory.threadpool.started, Equals(True))
+        finally:
+            yield factory.stopFactory()
+
+    @wait_for_reactor
+    @inlineCallbacks
     def test_stopFactory_stops_listener(self):
         factory = WebSocketFactory()
         yield factory.startFactory()
         yield factory.stopFactory()
         self.expectThat(factory.listener.connected(), Equals(False))
 
+    @wait_for_reactor
+    @inlineCallbacks
+    def test_stopFactory_stops_threadpool(self):
+        factory = WebSocketFactory()
+        yield factory.startFactory()
+        yield factory.stopFactory()
+        self.assertEqual([], factory.threadpool.threads)
+
     def test_registerNotifiers_registers_all_notifiers(self):
         factory = WebSocketFactory()
         self.assertItemsEqual(