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