r5404 - trunk/mint/python/mint
tmckay at fedoraproject.org
tmckay at fedoraproject.org
Wed Jun 6 17:33:57 UTC 2012
Author: tmckay
Date: 2012-06-06 17:33:56 +0000 (Wed, 06 Jun 2012)
New Revision: 5404
Modified:
trunk/mint/python/mint/expire.py
trunk/mint/python/mint/main.py
trunk/mint/python/mint/update.py
trunk/mint/python/mint/util.py
trunk/mint/python/mint/vacuum.py
Log:
Use joins and condition variables to make thread shutdown more reliable.
Modified: trunk/mint/python/mint/expire.py
===================================================================
--- trunk/mint/python/mint/expire.py 2012-06-06 16:04:27 UTC (rev 5403)
+++ trunk/mint/python/mint/expire.py 2012-06-06 17:33:56 UTC (rev 5404)
@@ -24,9 +24,20 @@
def run(self):
if self.interval >= 60 * 60:
- time.sleep(0.25 * self.interval)
+ interval = 0.25 * self.interval
else:
- time.sleep(self.interval)
+ interval = self.interval
+
+ try:
+ self._condition.acquire()
+ if not self.stop_requested:
+ self._condition.wait(interval)
+ if self.stop_requested:
+ log.debug("Periodic process %s exited" % self.getName())
+ return
+ finally:
+ self._condition.release()
+
super(ExpireThread, self).run()
def process(self):
@@ -49,7 +60,7 @@
count = cursor.rowcount
# For timestamp persistent classes, we get rid of the
- # objects in additoin to the samples
+ # objects in addition to the samples
if hasattr(cls, "sql_timestamp_delete") and \
cls.sql_timestamp_delete is not None:
cls.sql_timestamp_delete.execute(cursor, values)
@@ -63,10 +74,14 @@
cursor = conn.cursor()
if len(self.classes) > 0:
for cls in self.classes:
+ if self.stop_requested:
+ break
count += loop_body(cls, cursor)
else:
for pkg in self.packages:
for cls in pkg._classes:
+ if self.stop_requested:
+ break
count += loop_body(cls, cursor)
finally:
conn.close()
Modified: trunk/mint/python/mint/main.py
===================================================================
--- trunk/mint/python/mint/main.py 2012-06-06 16:04:27 UTC (rev 5403)
+++ trunk/mint/python/mint/main.py 2012-06-06 17:33:56 UTC (rev 5404)
@@ -89,14 +89,19 @@
log.info("Stopping %s", self)
self.update_thread.stop()
+ log.info("Update thread stopped")
- self.session.stop()
-
if self.expire_enabled:
self.expire_thread.stop()
+ log.info("Expire thread stopped")
if self.vacuum_enabled:
self.vacuum_thread.stop()
+ log.info("Vacuum thread stopped")
+
+ self.session.stop()
+ log.info("Session stopped")
+
def __repr__(self):
return self.__class__.__name__
Modified: trunk/mint/python/mint/update.py
===================================================================
--- trunk/mint/python/mint/update.py 2012-06-06 16:04:27 UTC (rev 5403)
+++ trunk/mint/python/mint/update.py 2012-06-06 17:33:56 UTC (rev 5404)
@@ -75,11 +75,12 @@
while True:
if self.stop_requested:
break
-
try:
update = self.updates.get(True, 1)
except Empty:
continue
+ if self.stop_requested:
+ break
self.stats.dequeued += 1
Modified: trunk/mint/python/mint/util.py
===================================================================
--- trunk/mint/python/mint/util.py 2012-06-06 16:04:27 UTC (rev 5403)
+++ trunk/mint/python/mint/util.py 2012-06-06 17:33:56 UTC (rev 5404)
@@ -12,7 +12,7 @@
from qmf.console import ObjectId
from random import sample
from tempfile import mkstemp
-from threading import Thread, Lock, RLock
+from threading import Thread, Lock, RLock, Condition
from time import clock, sleep
from traceback import print_exc
@@ -29,19 +29,40 @@
self.app = app
self.name = self.__class__.__name__
-
self.stop_requested = False
self.setDaemon(True)
-
+
def init(self):
pass
- def stop(self):
- assert self.stop_requested is False
+ def stop(self, timeout=5):
+ if self.stop_requested:
+ return
self.stop_requested = True
+ if self.isAlive():
+ self.join(timeout)
-class MintPeriodicProcessThread(MintDaemonThread):
+class MintBlockingDaemonThread(MintDaemonThread):
+ def __init__(self, app):
+ super(MintBlockingDaemonThread, self).__init__(app)
+
+ self._lock = Lock()
+ self._condition = Condition(self._lock)
+
+ def stop(self, timeout=5):
+ try:
+ self._condition.acquire()
+ if self.stop_requested:
+ return
+ self.stop_requested = True
+ self._condition.notify()
+ finally:
+ self._condition.release()
+ if self.isAlive():
+ self.join(timeout)
+
+class MintPeriodicProcessThread(MintBlockingDaemonThread):
def __init__(self, app, interval):
super(MintPeriodicProcessThread, self).__init__(app)
@@ -63,10 +84,18 @@
delta = elapsed % self.interval
then = datetime.now() + timedelta(seconds=delta)
- log.debug("Sleeping until %s", then.strftime("%H:%M:%S"))
+ try:
+ self._condition.acquire()
+ if not self.stop_requested:
+ log.debug("Periodic process %s sleeping until %s" \
+ % (self.getName(), then.strftime("%H:%M:%S")))
+ self._condition.wait(delta)
+ if self.stop_requested:
+ log.debug("Periodic process %s exiting" % self.getName())
+ break
+ finally:
+ self._condition.release()
- time.sleep(delta)
-
def process(self):
pass
Modified: trunk/mint/python/mint/vacuum.py
===================================================================
--- trunk/mint/python/mint/vacuum.py 2012-06-06 16:04:27 UTC (rev 5403)
+++ trunk/mint/python/mint/vacuum.py 2012-06-06 17:33:56 UTC (rev 5404)
@@ -14,9 +14,20 @@
def run(self):
if self.interval >= 60 * 60:
- time.sleep(0.25 * self.interval)
+ interval = 0.25 * self.interval
else:
- time.sleep(self.interval)
+ interval = self.interval
+
+ try:
+ self._condition.acquire()
+ if not self.stop_requested:
+ self._condition.wait(interval)
+ if self.stop_requested:
+ log.debug("Periodic process %s exited" % self.getName())
+ return
+ finally:
+ self._condition.release()
+
super(VacuumThread, self).run()
def process(self):
@@ -27,7 +38,6 @@
try:
cursor = conn.cursor()
-
cursor.execute("vacuum analyze")
finally:
conn.close()
More information about the cumin-developers
mailing list