reworked file processing manager and thread to shutdown cleanly and avoid busy wait

src/python/dm/common/processing/fileProcessingManager.py

@@ -58,7 +58,7 @@ class FileProcessingManager(threading.Thread,Singleton):
                 (moduleName,className,constructor) = cm.getModuleClassConstructorTuple(value)
                 self.logger.debug('Creating file processor instance of class %s' % className)
                 fileProcessor = ObjectUtility.createObjectInstance(moduleName, className, constructor)
-                self.logger.debug('Configring file processor %s' % fileProcessor)
+                self.logger.debug('Configuring file processor %s' % fileProcessor)
                 fileProcessor.setNumberOfRetries(self.defaultNumberOfRetries)
                 fileProcessor.setRetryWaitPeriodInSeconds(self.defaultRetryWaitPeriodInSeconds)
                 fileProcessor.configure()
@@ -67,8 +67,8 @@ class FileProcessingManager(threading.Thread,Singleton):
         self.fileProcessorKeyList.sort()
         self.logger.debug('File processor keys: %s' % self.fileProcessorKeyList)
 
-    def processObservedFile(self, observedFile):
-        self.fileProcessingQueue.push(observedFile)
+    def processFile(self, fileInfo):
+        self.fileProcessingQueue.push(fileInfo)
         self.eventFlag.set()
 
     def start(self):
@@ -77,24 +77,23 @@ class FileProcessingManager(threading.Thread,Singleton):
             self.logger.debug('Starting file processing threads')
             for i in range(0, self.nProcessingThreads):
                 tName = 'FileProcessingThread-%s' % i
-                t = FileProcessingThread(tName, self.eventFlag, self.fileProcessorDict, self.fileProcessorKeyList, self.fileProcessingQueue, self.processedFileDict, self.unprocessedFileDict)
+                t = FileProcessingThread(tName, self, self.fileProcessorDict, self.fileProcessorKeyList, self.fileProcessingQueue, self.processedFileDict, self.unprocessedFileDict)
                 t.start()
                 self.fileProcessingThreadDict[tName] = t
         finally:
             self.lock.release()
 
     def stop(self):
+        self.logger.debug('Stopping file processing threads')
+        for (tName, t) in self.fileProcessingThreadDict.items():
+            t.stop()
         self.lock.acquire()
         try:
-            self.logger.debug('Stopping file processing threads')
-            for (tName, t) in self.fileProcessingThreadDict.items():
-                t.stop()
             self.eventFlag.set()
-            for (tName, t) in self.fileProcessingThreadDict.items():
-                t.join()
         finally:
              self.lock.release()
-
+        for (tName, t) in self.fileProcessingThreadDict.items():
+            t.join()
 
     def setEvent(self):
         self.lock.acquire()
@@ -110,6 +109,9 @@ class FileProcessingManager(threading.Thread,Singleton):
         finally:
             self.lock.release()
 
+    def waitOnEvent(self, timeoutInSeconds=None):
+        self.eventFlag.wait(timeoutInSeconds)
+
 ####################################################################
 # Testing
 

src/python/dm/common/processing/fileProcessingThread.py

@@ -6,12 +6,14 @@ from dm.common.utility.loggingManager import LoggingManager
 
 class FileProcessingThread(threading.Thread):
 
-    def __init__ (self, name, eventFlag, fileProcessorDict, fileProcessorKeyList, fileProcessingQueue, processedFileDict, unprocessedFileDict):
+    THREAD_EVENT_TIMEOUT_IN_SECONDS = 10.0
+
+    def __init__ (self, name, fileProcessingManager, fileProcessorDict, fileProcessorKeyList, fileProcessingQueue, processedFileDict, unprocessedFileDict):
     
         threading.Thread.__init__(self)
         self.setName(name)
         self.exitFlag = False
-        self.eventFlag = eventFlag
+        self.fileProcessingManager = fileProcessingManager
         self.fileProcessorDict = fileProcessorDict
         self.fileProcessorKeyList = fileProcessorKeyList
         self.fileProcessingQueue = fileProcessingQueue
@@ -22,55 +24,58 @@ class FileProcessingThread(threading.Thread):
     def run(self):
         self.logger.debug('Starting thread: %s' % self.getName())
         while True:
+            self.fileProcessingManager.clearEvent()
             if self.exitFlag:
-                self.logger.debug('Exit flag set, %s done' % self.getName())
+                self.logger.debug('Exit flag is set')
                 break
 
             while True:
-                file = self.fileProcessingQueue.pop()
-                if file is None:
+                fileInfo = self.fileProcessingQueue.pop()
+                if fileInfo is None:
                     break
-                filePath = file.getFilePath()
-                daqPath = file.getDaqPath()
-                experiment = file.getExperiment()
+                filePath = fileInfo.get('filePath')
                 
                 try:
                     for processorKey in self.fileProcessorKeyList: 
                         processor = self.fileProcessorDict.get(processorKey)
                         processorName = processor.__class__.__name__
-                        fileProcessedByDict = file.get('processedByDict', {})
+                        fileProcessedByDict = fileInfo.get('processedByDict', {})
+                        fileInfo['processedByDict'] = fileProcessedByDict
+
                         if fileProcessedByDict.has_key(processorName):
-                            self.logger.debug('%s has already been processed by %s ' % (file, processorName))
+                            self.logger.debug('%s has already been processed by %s ' % (fileInfo, processorName))
                             continue
 
-                        self.logger.debug('%s is about to process file %s ' % (processorName, file))
+                        self.logger.debug('%s is about to process file %s ' % (processorName, fileInfo))
                         try:
-                            processor.processFile(filePath, daqPath, experiment)
+                            processor.processFile(fileInfo)
                             fileProcessedByDict[processorName] = True
-                            file['processedByDict'] = fileProcessedByDict
-                            self.logger.debug('%s processed file %s ' % (processorName, file))
+                            self.logger.debug('%s processed file at path %s ' % (processorName, filePath))
                         except Exception, ex:
                             self.logger.exception(ex)
-                            self.logger.debug('%s processing failed for file %s ' % (processorName, file))
-                            fileProcessingDict = file.get('processingDict', {})
-                            file['processingDict'] = fileProcessingDict
+                            self.logger.debug('%s processing failed for file at path %s ' % (processorName, filePath))
+                            fileProcessingDict = fileInfo.get('processingDict', {})
+                            fileInfo['processingDict'] = fileProcessingDict
                             processorDict = fileProcessingDict.get(processorName, {}) 
-                            processorDict['lastError'] = ex
+                            fileProcessingDict[processorName] = processorDict
+
+                            processorDict['lastError'] = str(ex)
                             nRetriesLeft = processorDict.get('numberOfRetriesLeft', processor.getNumberOfRetries()) 
-                            self.logger.debug('Number of %s retries left for file %s: %s' % (processorName, file, nRetriesLeft))
+                            self.logger.debug('Number of %s retries left for file %s: %s' % (processorName, filePath, nRetriesLeft))
                             processorDict['numberOfRetriesLeft'] = nRetriesLeft - 1
                             if nRetriesLeft <= 0:
-                                self.logger.debug('No more %s retries left for file %s' % (processorName, file))
-                                self.unprocessedFileDict[filePath] = file
+                                self.logger.debug('No more %s retries left for file %s' % (processorName, fileInfo))
+                                self.unprocessedFileDict[filePath] = fileInfo
                             else:
                                 retryWaitPeriod = processor.getRetryWaitPeriodInSeconds()
-                                self.logger.debug('%s will retry processing file %s in %s seconds' % (processorName, file, retryWaitPeriod))
-                                self.fileProcessingQueue.push(file, retryWaitPeriod)
+                                self.logger.debug('%s will retry processing file %s in %s seconds' % (processorName, filePath, retryWaitPeriod))
+                                self.fileProcessingQueue.push(fileInfo, retryWaitPeriod)
 
                 except Exception, ex:
                     self.logger.exception(ex)
 
-            self.eventFlag.wait()
+            self.fileProcessingManager.waitOnEvent(self.THREAD_EVENT_TIMEOUT_IN_SECONDS)
+        self.logger.debug('%s is done' % self.getName())
 
     def stop(self):
         self.exitFlag = True