removed singleton and passed instance of TaskExecutor to workers

bin/linspector

@@ -114,7 +114,7 @@ def main():
         logger.error(msg)
         exit()
 
-    task = TaskExecutor.Instance()
+    executor = TaskExecutor()
     #logger.debug("task executor instance :%s", str(task))
 
     job_count = 0
@@ -153,7 +153,7 @@ def main():
                         new_start_date = start_date + datetime.timedelta(seconds=time_delta)
 
                         worker = workers[instance]
-                        worker.create_job(jobs, service, host, hostgroup, core, period, start_date=new_start_date)
+                        worker.create_job(jobs, service, host, hostgroup, core, period, start_date=new_start_date, executor=executor)
 
                         instance += 1
                         if instance == args.instances:

linspector/core/job.py

@@ -39,7 +39,7 @@ logger = getLogger(__name__)
 
 
 class LinspectorJob:
-    def __init__(self, service, host, members, core, hostgroup):
+    def __init__(self, service, host, members, core, hostgroup, executor):
         self.service = service
         self.host = host
         self.members = members
@@ -49,6 +49,7 @@ class LinspectorJob:
         self.enabled = True
         self.scheduler_job = None
         self.job_id = self.hex_string()
+        self.executor = executor
         """
         NONE     job was not executed
         OK       when everything is fine
@@ -117,7 +118,7 @@ class LinspectorJob:
                     logger.debug("Executing Task of type: " + self.status)
                     try:
 
-                        TaskExecutor.Instance().schedule_task(job_information, task)
+                        self.executor.schedule_task(job_information, task)
                     except Exception, e:
                         logger.error("Error while executing: " + str(e))
     def handle_call(self):

linspector/core/worker.py

@@ -53,8 +53,8 @@ class LinspectorWorker(Process):
     def shutdown(self, wait=True):
         self.scheduler.shutdown(wait=wait)
 
-    def create_job(self, jobs, service, host, hostgroup, core, period, start_date=None):
+    def create_job(self, jobs, service, host, hostgroup, core, period, start_date=None, executor=None):
-        job = LinspectorJob(service, host, hostgroup.get_members(), core, hostgroup)
+        job = LinspectorJob(service, host, hostgroup.get_members(), core, hostgroup, executor)
         scheduler_job = period.createJob(self.scheduler, job, self.handle_job, start_date=start_date)
         if scheduler_job is not None:
             job.set_job(scheduler_job)

linspector/tasks/task.py

@@ -21,6 +21,7 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
 
 from logging import getLogger
 from threading import Event, Thread
+from time import sleep
 from Queue import Queue
 from linspector.utils.singleton import Singleton
 
@@ -75,7 +76,6 @@ class Task(object):
             raise e
 
 
-@Singleton
 class TaskExecutor(object):
     def __init__(self):
         logger.debug("init taskExecutor")
@@ -90,31 +90,14 @@ class TaskExecutor(object):
     def _run_worker_thread(self):
         logger.debug("start running taskExcecutor worker Thread")
         while self.is_running() or not self.is_instant_end():
-            '''
-            if len(self.taskInfos) == 0:
-                logger.debug("waiting for jobs to add")
-                self.event.clear()
-                self.event.wait()
-                logger.debug("Task Executor waked up")
-
             try:
-                logger.debug("trying to get task information")
+                logger.debug("try to get queued task from %s", self.taskInfos)
-                msg, task = self.taskInfos[0]
-                del self.taskInfos[0]
-                if task:
-                    logger.debug("Starting Task Execution of task: %s", str(task))
-                    task.execute(msg)
-
-            except Exception, e:
-                logger.error("Error " + str(e))
-            '''
-            try:
-                logger.debug("try to get queued task")
                 msg, task = self.taskInfos.get()
                 logger.debug("got task %s for msg: %s", str(task), str(msg))
                 task.execute(msg)
+                self.taskInfos.task_done()
             except Exception, e:
-                logger.error("Error " + str(e))
+                logger.error("Error: %s", e.message)
         logger.debug("shutting down TaskExecutor!")
 
     def is_instant_end(self):
@@ -125,18 +108,16 @@ class TaskExecutor(object):
 
     def stop(self):
         self._running = False
-        self.event.set()
 
     def stop_immediately(self):
         self._running = False
         self._instantEnd = True
-        self.event.set()
 
     def schedule_task(self, msg, task):
         try:
             logger.debug("appending task '%s' for msg: %s", str(task), str(msg))
             self.taskInfos.put((msg, task))
+            logger.debug("into queue: %s ", str(self.taskInfos))
         except Exception, e:
             logger.debug("queue is probably full: %s", str(e))
-        #logger.debug("%s task Queued", str(len(self.taskInfos)))
+
-        #self.event.set()