Added term handler to TaskProvider and DataDispatcher

src/sender/DataDispatcher.py

@@ -8,6 +8,7 @@ import logging
 import traceback
 import cPickle
 import shutil
+import signal
 from multiprocessing import Process
 
 try:
@@ -53,7 +54,9 @@ class DataDispatcher():
         self.id              = id
         self.log             = self.__getLogger(logQueue)
 
-        self.currentPID         = os.getpid()
+        signal.signal(signal.SIGTERM, self.signal_term_handler)
+
+        self.currentPID      = os.getpid()
         self.log.debug("DataDispatcher-" + str(self.id) + " started (PID " + str(self.currentPID) + ").")
 
         self.controlConId    = controlConId
@@ -297,6 +300,11 @@ class DataDispatcher():
             self.context = None
 
 
+    def signal_term_handler(self, signal, frame):
+        self.log.debug('got SIGTERM')
+        self.stop()
+
+
     def __exit__ (self):
         self.stop()
 

src/sender/TaskProvider.py

@@ -7,6 +7,8 @@ import logging
 import sys
 import trace
 import cPickle
+import signal
+import errno
 
 try:
     BASE_PATH = os.path.dirname ( os.path.dirname ( os.path.dirname ( os.path.realpath ( __file__ ) )))
@@ -45,6 +47,8 @@ class TaskProvider():
 
         self.log                = self.getLogger(logQueue)
 
+        signal.signal(signal.SIGTERM, self.signal_term_handler)
+
         self.currentPID         = os.getpid()
         self.log.debug("TaskProvider started (PID " + str(self.currentPID) + ").")
 
@@ -158,6 +162,12 @@ class TaskProvider():
                 workloadList = self.eventDetector.getNewEvent()
             except KeyboardInterrupt:
                 break
+            except IOError as e:
+                if e.errno == errno.EINTR:
+                    break
+                else:
+                    self.log.error("Invalid fileEvent message received.", exc_info=True)
+                    continue
             except:
                 self.log.error("Invalid fileEvent message received.", exc_info=True)
                 #skip all further instructions and continue with next iteration
@@ -240,6 +250,11 @@ class TaskProvider():
             self.context = None
 
 
+    def signal_term_handler(self, signal, frame):
+        self.log.debug('got SIGTERM')
+        self.stop()
+
+
     def __exit__ (self):
         self.stop()