From 8c4089d7e33a3eae0a53613b1fe491d7808a9ba3 Mon Sep 17 00:00:00 2001 From: Tim Laszlo Date: Fri, 4 Jun 2010 19:20:16 +0000 Subject: Added ThreadedStatistics plugin class. Migrated DBStats to use ThreadedStatistics. git-svn-id: https://svn.mcs.anl.gov/repos/bcfg/trunk/bcfg2@5889 ce84e21b-d406-0410-9b95-82705330c041 --- src/lib/Server/Plugin.py | 31 +++++++++++++++++++++++++++++++ 1 file changed, 31 insertions(+) (limited to 'src/lib/Server/Plugin.py') diff --git a/src/lib/Server/Plugin.py b/src/lib/Server/Plugin.py index f3b97b336..ff87c15df 100644 --- a/src/lib/Server/Plugin.py +++ b/src/lib/Server/Plugin.py @@ -7,6 +7,8 @@ import lxml.etree import os import posixpath import re +import Queue +import threading from lxml.etree import XML, XMLSyntaxError @@ -141,6 +143,35 @@ class Statistics(object): def process_statistics(self, client, xdata): pass +class ThreadedStatistics(Statistics, + threading.Thread): + '''Threaded statistics handling capability''' + def __init__(self, core, datastore): + Statistics.__init__(self) + threading.Thread.__init__(self) + # Event from the core signaling an exit + self.terminate = core.terminate + self.work_queue = Queue.Queue() + self.start() + + def run(self): + while not (self.terminate.isSet() and self.work_queue.empty()): + try: + (xdata, client) = self.work_queue.get(block=True, timeout=5) + except Queue.Empty: + continue + except Exception, e: + logger.error("ThreadedStatistics: %s" % e) + continue + self.handle_statistic(xdata, client) + + def process_statistics(self, metadata, data): + self.work_queue.put((metadata, copy.deepcopy(data))) + + def handle_statistics(self, metadata, data): + '''Handle stats here''' + pass + class PullSource(object): def GetExtra(self, client): return [] -- cgit v1.2.3-1-g7c22