[Pkg-nagios-changes] [SCM] UNNAMED PROJECT branch, debian/master, updated. 810edbdd3feedbfe37f4a65bee50b57b2f60fa2a

Naparuba naparuba at gmail.com
Tue Feb 28 22:08:28 UTC 2012


The following commit has been merged in the debian/master branch:
commit 16cebd27c2f0adcb69a304aaaa0e1cf7a7235faf
Author: Naparuba <naparuba at gmail.com>
Date:   Mon Dec 12 16:36:52 2011 +0100

    Enh : send broks in external modules as a biglist instead of n broks. It's more efficient (X3 perf for this, can be huge time for huge conf)

diff --git a/shinken/daemons/brokerdaemon.py b/shinken/daemons/brokerdaemon.py
index f792175..4c5403f 100644
--- a/shinken/daemons/brokerdaemon.py
+++ b/shinken/daemons/brokerdaemon.py
@@ -556,12 +556,18 @@ class Broker(BaseSatellite):
         # and for external queues
         # REF: doc/broker-modules.png (3)
         # We put to external queues broks that was not already send
+        t0 = time.time()
+        # We are sending broks are a biglist, most efficient than one by one
         queues = self.modules_manager.get_external_to_queues()
-        for b in (b for b in self.broks if getattr(b, 'need_send_to_ext', True)):
-            # if b.type != 'log':
-            for q in queues:
-                q.put(b)
-                b.need_send_to_ext = False
+        to_send = [b for b in self.broks if getattr(b, 'need_send_to_ext', True)]
+
+        for q in queues:
+            q.put(to_send)
+
+        # No more need to send them
+        for b in to_send:
+            b.need_send_to_ext = False
+        print "DBG: Time to send %s broks" % len(self.broks), time.time() - t0
 
         # We must had new broks at the end of the list, so we reverse the list
         self.broks.reverse()
diff --git a/shinken/modules/livestatus_broker/livestatus_broker.py b/shinken/modules/livestatus_broker/livestatus_broker.py
index 117f4e3..0a83dee 100644
--- a/shinken/modules/livestatus_broker/livestatus_broker.py
+++ b/shinken/modules/livestatus_broker/livestatus_broker.py
@@ -832,8 +832,9 @@ class Livestatus_broker(BaseModule):
  
         while not self.interrupted:
             try:
-                b = self.to_q.get(True, .01)  # do not block indefinitely
-                self.manage_brok(b)
+                l = self.to_q.get(True, .01)
+                for b in l:
+                    self.manage_brok(b)
             #We do not ware about Empty queue
             except Queue.Empty:
                 pass
diff --git a/shinken/modules/npcdmod_broker.py b/shinken/modules/npcdmod_broker.py
index fc06371..672ba90 100644
--- a/shinken/modules/npcdmod_broker.py
+++ b/shinken/modules/npcdmod_broker.py
@@ -214,8 +214,9 @@ class Npcd_broker(BaseModule):
         self.rotate()
         last_rotated = time.time()
         while not self.interrupted:
-            b = self.to_q.get() # can block here :)
-            self.manage_brok(b)
+            l = self.to_q.get() # can block here :)
+            for b in l:
+                self.manage_brok(b)
             if time.time() - last_rotated > self.sleep_time:
                 self.rotate()
                 last_rotated = time.time()
diff --git a/shinken/modules/status_dat_broker/status_dat_broker.py b/shinken/modules/status_dat_broker/status_dat_broker.py
index 2f19d63..e65922a 100644
--- a/shinken/modules/status_dat_broker/status_dat_broker.py
+++ b/shinken/modules/status_dat_broker/status_dat_broker.py
@@ -398,8 +398,9 @@ class Status_dat_broker(BaseModule):
 
         while not self.interrupted:
             try:
-                b = self.to_q.get(True, 5)
-                self.manage_brok(b)
+                l = self.to_q.get(True, 5)
+                for b in l:
+                    self.manage_brok(b)
             except IOError, e:
                 if e.errno != os.errno.EINTR:
                     raise
diff --git a/shinken/modules/thrift_broker/thrift_broker.py b/shinken/modules/thrift_broker/thrift_broker.py
index efa2d42..ce9b1a7 100644
--- a/shinken/modules/thrift_broker/thrift_broker.py
+++ b/shinken/modules/thrift_broker/thrift_broker.py
@@ -963,9 +963,10 @@ class Thrift_broker(BaseModule):
     def manage_broks(self,*args):
         while True:
             try:
-                b = self.to_q.get(True, .01)
-                print b
-                self.manage_brok(b)
+                l = self.to_q.get(True, .01)
+                for b in l:
+                    print b
+                    self.manage_brok(b)
             except Queue.Empty:
                 pass
             except IOError, e:
diff --git a/shinken/modules/webui_broker/webui_broker.py b/shinken/modules/webui_broker/webui_broker.py
index 513e14c..4a23224 100644
--- a/shinken/modules/webui_broker/webui_broker.py
+++ b/shinken/modules/webui_broker/webui_broker.py
@@ -214,36 +214,38 @@ class Webui_broker(BaseModule, Daemon):
     def manage_brok_thread(self):
         print "Data thread started"
         while True:
-           b = self.to_q.get()
-           # For updating, we cannot do it while
-           # answer queries, so wait for no readers
-           self.wait_for_no_readers()
-           try:
+           l = self.to_q.get()
+           
+           for b in l:
+              # For updating, we cannot do it while
+              # answer queries, so wait for no readers
+              self.wait_for_no_readers()
+              try:
                #print "Got data lock, manage brok"
-               self.rg.manage_brok(b)
-               for mod in self.modules_manager.get_internal_instances():
-                   try:
-                       mod.manage_brok(b)
-                   except Exception , exp:
-                       print exp.__dict__
-                       logger.log("[%s] Warning : The mod %s raise an exception: %s, I'm tagging it to restart later" % (self.name, mod.get_name(),str(exp)))
-                       logger.log("[%s] Exception type : %s" % (self.name, type(exp)))
-                       logger.log("Back trace of this kill: %s" % (traceback.format_exc()))
-                       self.modules_manager.set_to_restart(mod)
-           except Exception, exp:            
-               msg = Message(id=0, type='ICrash', data={'name' : self.get_name(), 'exception' : exp, 'trace' : traceback.format_exc()})
-               self.from_q.put(msg)
-               # wait 2 sec so we know that the broker got our message, and die
-               time.sleep(2)
-               # No need to raise here, we are in a thread, exit!
-               os._exit(2)
-           finally:
-               # We can remove us as a writer from now. It's NOT an atomic operation
-               # so we REALLY not need a lock here (yes, I try without and I got
-               # a not so accurate value there....)
-               self.global_lock.acquire()
-               self.nb_writers -= 1
-               self.global_lock.release()
+                  self.rg.manage_brok(b)
+                  for mod in self.modules_manager.get_internal_instances():
+                      try:
+                          mod.manage_brok(b)
+                      except Exception , exp:
+                          print exp.__dict__
+                          logger.log("[%s] Warning : The mod %s raise an exception: %s, I'm tagging it to restart later" % (self.name, mod.get_name(),str(exp)))
+                          logger.log("[%s] Exception type : %s" % (self.name, type(exp)))
+                          logger.log("Back trace of this kill: %s" % (traceback.format_exc()))
+                          self.modules_manager.set_to_restart(mod)
+              except Exception, exp:            
+                  msg = Message(id=0, type='ICrash', data={'name' : self.get_name(), 'exception' : exp, 'trace' : traceback.format_exc()})
+                  self.from_q.put(msg)
+                  # wait 2 sec so we know that the broker got our message, and die
+                  time.sleep(2)
+                  # No need to raise here, we are in a thread, exit!
+                  os._exit(2)
+              finally:
+                  # We can remove us as a writer from now. It's NOT an atomic operation
+                  # so we REALLY not need a lock here (yes, I try without and I got
+                  # a not so accurate value there....)
+                  self.global_lock.acquire()
+                  self.nb_writers -= 1
+                  self.global_lock.release()
 
 
     # Here we will load all plugins (pages) under the webui/plugins

-- 
UNNAMED PROJECT



More information about the Pkg-nagios-changes mailing list