Changeset 36

Timestamp:

04/04/05 16:05:26 (19 years ago)

Author:

bastiaans

Message:

daemon/togad.py:

• Rewrote multithreaded
• Error occurs when trying to update a RRD with multiple values

File:

• trunk/daemon/togad.py (modified) (8 diffs)

trunk/daemon/togad.py

@@ -11 +11 @@
 import time
 import re
+import threading
+import mutex
 
 # Specify debugging level here;
@@ -181 +183 @@
                 "Connect, and return a file object"
 
-                if not self.s:
-                        self.connect()
+                self.disconnect()
+                self.connect()
 
                 return self.s.makefile( 'r' )
@@ -230 +232 @@
                 sys.stdin.close()
                 sys.stdout.close()
-                sys.stderr.close()
+                #sys.stderr.close()
 
                 os.open('/dev/null', 0)
@@ -279 +281 @@
                 "Main thread"
 
-                # Daemonized not working yet
-                if DAEMONIZE:
-                        pid = os.fork()
-
-                        # Handle XML grabbing in Child
-                        if pid == 0:
-
-                                while( 1 ):
-                                        self.grabXML()
-
-                        # Do scheduled RRD storing in Parent
-                        #elif pid > ):
-
-                else:
-                        #self.grabXML()
-                        self.processXML()
-                        self.storeMetrics()
+                #self.processXML()
+                #self.storeMetrics()
+
+                #time.sleep( 20 )
+
+                #self.processXML()
+                #self.storeMetrics()
+
+                #sys.exit(1)
+
+                xmlthread = threading.Thread( None, self.processXML, 'xmlthread' )
+                storethread = threading.Thread( None, self.storeMetrics, 'storethread' )
+
+                while( 1 ):
+
+                        if not xmlthread.isAlive():
+                                # Gather XML at the same interval as gmetad
+
+                                # threaded call to: self.processXML()
+                                #
+                                xmlthread = threading.Thread( None, self.processXML, 'xmlthread' )
+                                debug_msg( 7, self.printTime() + ' - mainthread() - xmlthread() started' )
+                                xmlthread.start()
+
+                        if not storethread.isAlive():
+                                # Store metrics every .. sec
+
+                                # threaded call to: self.storeMetrics()
+                                #
+                                storethread = threading.Thread( None, self.storeMetrics, 'storethread' )
+                                debug_msg( 7, self.printTime() + ' - mainthread() - storethread() started' )
+                                storethread.start()
+                
+                        # Just sleep a sec here, to prevent daemon from going mad. We're all threads here anyway
+                        time.sleep( 1 ) 
 
         def storeMetrics( self ):
                 "Store metrics retained in memory to disk"
 
-                self.myHandler.storeMetrics()
+                STORE_INTERVAL = 30
+
+                debug_msg( 7, self.printTime() + ' - storethread() - Storing data..' )
+
+                # threaded call to: self.myHandler.storeMetrics()
+                #
+                storethread = threading.Thread( None, self.myHandler.storeMetrics(), 'storemetricthread' )
+                storethread.start()
+
+                debug_msg( 7, self.printTime() + ' - storethread() - Sleeping.. (%ss)' %STORE_INTERVAL )
+                time.sleep( STORE_INTERVAL )
+                debug_msg( 7, self.printTime() + ' - storethread() - Done sleeping.' )
+
+                if storethread.isAlive():
+
+                        debug_msg( 7, self.printTime() + ' - storethread() - storemetricthread() still running, waiting to finish..' )
+                        parsethread.join( 180 ) # Maximum time is 3 minutes for storing thread to finish - more then enough
+                        debug_msg( 7, self.printTime() + ' - storethread() - storemetricthread() finished.' )
+
+                debug_msg( 7, self.printTime() + ' - storethread() finished' )
+
+                return 0
 
         def processXML( self ):
                 "Process XML"
 
-                self.myParser.parse( self.myXMLGatherer.getFileObject() )
+                debug_msg( 7, self.printTime() + ' - xmlthread() - Parsing..' )
+
+                # threaded call to: self.myParser.parse( self.myXMLGatherer.getFileObject() )
+                #
+                parsethread = threading.Thread( None, self.myParser.parse, 'parsethread', [ self.myXMLGatherer.getFileObject() ] )
+                parsethread.start()
+
+                debug_msg( 7, self.printTime() + ' - xmlthread() - Sleeping.. (%ss)' %self.config.getLowestInterval() )
+                time.sleep( float( self.config.getLowestInterval() ) )  
+                debug_msg( 7, self.printTime() + ' - xmlthread() - Done sleeping.' )
+
+                if parsethread.isAlive():
+
+                        debug_msg( 7, self.printTime() + ' - xmlthread() - parsethread() still running, waiting to finish..' )
+                        parsethread.join( 180 ) # Maximum time is 3 minutes for XML thread to finish - more then enough
+                        debug_msg( 7, self.printTime() + ' - xmlthread() - parsethread() finished.' )
+
+                debug_msg( 7, self.printTime() + ' - xmlthread() finished.' )
+
+                return 0
 
 class GangliaConfigParser:
@@ -382 +442 @@
 
         myMetrics = { }
+        slot = None
 
         def __init__( self, config, cluster ):
                 self.cluster = cluster
                 self.config = config
+                self.slot = mutex.mutex()
 
         def getClusterName( self ):
@@ -408 +470 @@
                         self.myMetrics[ host ][ metric['name'] ] = [ ]
 
-
+                self.slot.testandset()
                 self.myMetrics[ host ][ metric['name'] ].append( metric )
+                self.slot.unlock()
 
         def makeUpdateString( self, host, metricname ):
@@ -429 +492 @@
                                 mytime = self.makeTimeSerial()
                                 self.createCheck( hostname, metricname, mytime )        
-                                update_okay = self.update( hostname, metricname, mytime )
-
-                                if not update_okay:
-
+                                update_ret = self.update( hostname, metricname, mytime )
+
+                                if update_ret == 0:
+
+                                        self.slot.testandset()
                                         del self.myMetrics[ hostname ][ metricname ]
+                                        self.slot.unlock()
                                         debug_msg( 9, 'stored metric %s for %s' %( hostname, metricname ) )
                                 else:
                                         debug_msg( 9, 'metric update failed' )
 
-                                sys.exit(1)
+                                return 1
 
         def makeTimeSerial( self ):
@@ -590 +655 @@
                 debug_msg( 9, 'updated rrd %s with %s' %( str(rrd_file), update_string ) )
 
+                return 0
+
 def main():
         "Program startup"