added LinspectorWorker to startup more scheduler instances for multiporcessing of jobs; does not work but is on the right way

bin/linspector

@@ -29,13 +29,13 @@ import logging.handlers
 import os
 import os.path as path
 import sys
-import multiprocessing
+
-from multiprocessing import Pool
+from multiprocessing import cpu_count
 
 from linspector.config.parser import FullConfigParser
 from linspector.core.interface import LinspectorInterface
 from linspector.core.job import LinspectorJob
-from linspector.core.scheduler import LinspectorScheduler
+from linspector.core.worker import LinspectorWorker
 from linspector.frontends.lish import LishFrontend
 from linspector.tasks.task import TaskExecutor
 
@@ -63,11 +63,15 @@ def parse_args():
     parser.add_argument("-c", "--logcount", default=5, type=int,
                         help="maximum number of logfiles in rotation (default: 5)")
 
+    parser.add_argument("-i", "--instances", default=cpu_count(), type=int,
+                        help="number of scheduler instances (default: number of cpu cores available (" +
+                              str(cpu_count()) + "))")
+
     parser.add_argument("-m", "--logsize", default=10485760, type=int,
                         help="maximum logfile size in bytes (default: 10485760)")
 
-    parser.add_argument("-t", "--threads", default=3500, type=int,
+    parser.add_argument("-t", "--threads", default=1000, type=int,
-                        help="maximum number of scheduler threads (default: 3500)")
+                        help="maximum number of scheduler threads (default: 1000)")
 
     parser.add_argument("-k", "--corethreads", default=0, type=int,
                         help="number of scheduler core threads (default: 0)")
@@ -91,14 +95,6 @@ def parse_args():
     return parser.parse_args()
 
 
-pool = Pool(processes=16)
-print 'cpu_count() = %d\n' % multiprocessing.cpu_count()
-
-
-def handle_job(job):
-    result = pool.apply_async(job.handle_call())
-
-
 def main():
     global lin_conf
     args = parse_args()
@@ -122,9 +118,6 @@ def main():
         logger.error(msg)
         exit()
 
-    scheduler = LinspectorScheduler({"apscheduler.threadpool.core_threads": args.corethreads,
-                           "apscheduler.threadpool.max_threads": args.threads})
-    scheduler.start()
     TaskExecutor.Instance()
 
     job_count = 0
@@ -132,11 +125,23 @@ def main():
         for hostgroup in layout.get_hostgroups():
             job_count += (hostgroup.get_services().__len__() * hostgroup.get_hosts().__len__())
 
+    workers = []
+    for i in range(0, args.instances):
+        name = "LinspectorWorker-"+str(i)
+        process = LinspectorWorker(name, args.corethreads, args.threads)
+        workers.append(process)
+        process.daemon = True
+        process.start()
+
+    for worker in workers:
+        print "RESULT is %s" % worker.get_scheduler_name()
+
     start_date = datetime.datetime.now()
     time_delta = 0
     jobs = []
     count = 0
     percent = 0
+    instance = 0
     for layout in lin_conf.get_enabled_layouts():
         for hostgroup in layout.get_hostgroups():
             for service in hostgroup.get_services():
@@ -159,14 +164,22 @@ def main():
                                             hostgroup.get_members(),
                                             core,
                                             hostgroup)
-                        scheduler_job = period.createJob(scheduler, job, handle_job, start_date=new_start_date)
+
+                        worker = workers[instance]
+                        scheduler_job = period.createJob(worker.get_scheduler(), job, worker.handle_job,
+                                                         start_date=new_start_date)
                         if scheduler_job is not None:
                             job.set_job(scheduler_job)
                             jobs.append(job)
 
+                        instance += 1
+                        if instance == args.instances:
+                            instance = 0
+
     print("\nScheduled " + str(job_count) + " jobs")
 
-    interface = LinspectorInterface(jobs, scheduler, lin_conf, root_logger, __version__)
+    #TODO: add a list of workers and not only the last one; just a hack to make it run for testing
+    interface = LinspectorInterface(jobs, worker.get_scheduler(), lin_conf, root_logger, __version__)
 
     if "jsonrpc_backend" in core and core["jsonrpc_backend"]:
         from linspector.backends.jsonrpc import JsonrpcBackend
@@ -184,13 +197,15 @@ def main():
     if "shutdown_wait" in core:
         shutdown_wait = core["shutdown_wait"]
 
-    scheduler.shutdown(wait=shutdown_wait)
+    for worker in workers:
+        worker.shutdown(shutdown_wait)
+        worker.terminate()
+
     logging.shutdown()
     if shutdown_wait:
         TaskExecutor.Instance().stop()
     else:
         TaskExecutor.Instance().stop_immediately()
 
-
 if __name__ == "__main__":
     main()

linspector/core/worker.py

@@ -0,0 +1,50 @@
+"""
+Copyright (c) 2011-2013 by Johannes Findeisen and Rafael Timmerberg
+
+This file is part of Linspector (http://linspector.org).
+
+Linspector is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as
+published by the Free Software Foundation, either version 3 of the
+License, or (at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program. If not, see <http://www.gnu.org/licenses/>.
+"""
+
+from logging import getLogger
+from multiprocessing import Process
+
+from linspector.core.scheduler import LinspectorScheduler
+
+logger = getLogger(__name__)
+
+
+class LinspectorWorker(Process):
+    def __init__(self, name, core_threads, max_threads):
+        super(LinspectorWorker, self).__init__()
+
+        self._name = name
+        self.core_threads = core_threads
+        self.max_threads = max_threads
+
+        self.scheduler = LinspectorScheduler({"apscheduler.threadpool.core_threads": self.core_threads,
+                                              "apscheduler.threadpool.max_threads": self.max_threads})
+        self.scheduler.start()
+
+    def handle_job(self, job):
+        job.handle_call()
+
+    def get_scheduler_name(self):
+        return "Process returned %s" % self.name + " " + str(self.scheduler)
+
+    def get_scheduler(self):
+        return self.scheduler
+
+    def shutdown(self, wait=True):
+        self.scheduler.shutdown(wait=wait)