diff --git a/bin/linspector b/bin/linspector index d915f7a..9c64213 100755 --- a/bin/linspector +++ b/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, - help="maximum number of scheduler threads (default: 3500)") + parser.add_argument("-t", "--threads", default=1000, type=int, + 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() \ No newline at end of file diff --git a/linspector/core/worker.py b/linspector/core/worker.py new file mode 100644 index 0000000..bff94a9 --- /dev/null +++ b/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 . +""" + +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) \ No newline at end of file