Merge lp:~free.ekanayaka/landscape-client/clones-fixes into lp:~landscape/landscape-client/trunk

[Free Ekanayaka]

This branch:

- makes the watchdog increase the system's nofile limit of open file descriptors when using the --clones option, allowing the child processes to open a possibly high number of unix sockets to communicate with each other (there will be several instances of Broker, Manager and Monitor in each process)

- makes the TwistedReactor.call_in_thread method use Twisted's deferToThread, which will use a threadpool instead of spawning a thread each time, this should be both more efficient and safe (when running clones the number of threads will be kept under control)

- increases the MethodCallProtocol timeout when using --clones, as RPCs between processes can lag a bit in that case

[Geoff Teale (tealeg)]

+1 All good.

review: Approve

[Alberto Donato (ack)]

Great, +1!

review: Approve

=== modified file 'landscape/reactor.py'
--- landscape/reactor.py	2011-12-02 08:35:25 +0000
+++ landscape/reactor.py	2012-02-03 10:35:22 +0000
@@ -1,4 +1,3 @@
-import thread
 import time
 import sys
 import logging
@@ -8,6 +7,7 @@
 from twisted.test.proto_helpers import FakeDatagramTransport
 from twisted.internet.defer import succeed, fail
 from twisted.internet.error import DNSLookupError
+from twisted.internet.threads import deferToThread
 
 from landscape.log import format_object
 
@@ -120,8 +120,20 @@
         @note: Both C{callback} and C{errback} will be executed in the
             the parent thread.
         """
-        thread.start_new_thread(self._in_thread,
-                                (callback, errback, f, args, kwargs))
+        def on_success(result):
+            if callback:
+                return callback(result)
+
+        def on_failure(failure):
+            exc_info = (failure.type, failure.value, failure.tb)
+            if errback:
+                errback(*exc_info)
+            else:
+                logging.error(exc_info[1], exc_info=exc_info)
+
+        deferred = deferToThread(f, *args, **kwargs)
+        deferred.addCallback(on_success)
+        deferred.addErrback(on_failure)
 
     def _in_thread(self, callback, errback, f, args, kwargs):
         try:

=== modified file 'landscape/service.py'
--- landscape/service.py	2011-12-01 13:38:58 +0000
+++ landscape/service.py	2012-02-03 10:35:22 +0000
@@ -8,6 +8,7 @@
 from landscape.log import rotate_logs
 from landscape.reactor import TwistedReactor
 from landscape.deployment import get_versioned_persist, init_logging
+from landscape.amp import ComponentProtocol
 
 
 class LandscapeService(Service, object):
@@ -78,6 +79,9 @@
 
     if configuration.clones > 0:
 
+        # Increase the timeout of AMP's MethodCalls
+        ComponentProtocol.timeout = 300
+
         # Create clones here because TwistedReactor.__init__ would otherwise
         # cancel all scheduled delayed calls
         clones = []
@@ -93,8 +97,9 @@
         configuration.is_clone = False
 
         def start_clones():
-            # Spawn instances over 25-30 minutes.
-            delay = configuration.start_clones_over / configuration.clones
+            # Spawn instances over the given time window
+            start_clones_over = float(configuration.start_clones_over)
+            delay = start_clones_over / configuration.clones
 
             for i, clone in enumerate(clones):
 

=== modified file 'landscape/tests/test_reactor.py'
--- landscape/tests/test_reactor.py	2011-07-05 05:09:11 +0000
+++ landscape/tests/test_reactor.py	2012-02-03 10:35:22 +0000
@@ -243,9 +243,6 @@
 
         reactor.call_in_thread(None, None, f, 1, 2, c=3)
 
-        while not called:
-            pass
-
         reactor.call_later(0.7, reactor.stop)
         reactor.run()
 
@@ -274,9 +271,6 @@
 
         reactor.call_in_thread(callback, errback, f)
 
-        while not called:
-            pass
-
         reactor.call_later(0.7, reactor.stop)
         reactor.run()
 
@@ -301,9 +295,6 @@
 
         reactor.call_in_thread(callback, errback, f)
 
-        while not called:
-            pass
-
         reactor.call_later(0.7, reactor.stop)
         reactor.run()
 
@@ -329,9 +320,6 @@
 
         reactor.call_in_thread(callback, None, f)
 
-        while not called:
-            pass
-
         reactor.call_later(0.7, reactor.stop)
         reactor.run()
 
@@ -355,9 +343,6 @@
 
         reactor.call_in_thread(None, None, f)
 
-        while not called:
-            pass
-
         reactor.call_later(0.7, reactor.stop)
         reactor.run()
 

=== modified file 'landscape/watchdog.py'
--- landscape/watchdog.py	2011-12-02 08:35:25 +0000
+++ landscape/watchdog.py	2012-02-03 10:35:22 +0000
@@ -13,6 +13,7 @@
 import time
 
 from logging import warning, info, error
+from resource import setrlimit, RLIMIT_NOFILE
 
 from twisted.internet import reactor
 from twisted.internet.defer import Deferred, succeed
@@ -27,6 +28,7 @@
 from landscape.lib.bootstrap import (BootstrapList, BootstrapFile,
                                      BootstrapDirectory)
 from landscape.log import rotate_logs
+from landscape.amp import ComponentProtocol
 from landscape.broker.amp import (
     RemoteBrokerConnector, RemoteMonitorConnector, RemoteManagerConnector)
 from landscape.reactor import TwistedReactor
@@ -490,10 +492,21 @@
         Service.startService(self)
         bootstrap_list.bootstrap(data_path=self._config.data_path,
                                  log_dir=self._config.log_dir)
-        for i in range(self._config.clones):
-            suffix = "-clone-%d" % i
-            bootstrap_list.bootstrap(data_path=self._config.data_path + suffix,
-                                     log_dir=self._config.log_dir + suffix)
+        if self._config.clones > 0:
+
+            # Let clones open an appropriate number of fds
+            setrlimit(RLIMIT_NOFILE, (self._config.clones * 100,
+                                      self._config.clones * 200))
+
+            # Increase the timeout of AMP's MethodCalls
+            ComponentProtocol.timeout = 300
+
+            # Create clones log and data directories
+            for i in range(self._config.clones):
+                suffix = "-clone-%d" % i
+                bootstrap_list.bootstrap(
+                    data_path=self._config.data_path + suffix,
+                    log_dir=self._config.log_dir + suffix)
 
         result = self.watchdog.check_running()