Changeset 32461

Timestamp:

Sep 29, 2011, 3:23:06 PM (15 years ago)

Author:

rhenders

Message:

added a method to lock the batch table; reworked methid to get a new batch ID - this is a critical section and is called by clients after locking the table; new method for getting a list of consistently failed IDs; a new class field for the max number of failures before we give up on a batch; new method processingNow() for checking if another client is processing this batch now; some name changes for methods that get lists of batch IDs

File:

• trunk/ippToPsps/jython/ipptopspsdb.py (modified) (11 diffs)

trunk/ippToPsps/jython/ipptopspsdb.py

@@ -20 +20 @@
         super(IppToPspsDb, self).__init__(logger, doc, "ipptopspsdatabase")
 
-    '''
-    Creates a new batch
-    '''
-    def createNewBatch(self, batchType, stageID, survey, dvoDb, datastoreProduct):
-
-        sql = "INSERT INTO batch ( \
-               batch_type, \
-               stage_id, \
-               survey, \
-               dvo_db, \
-               datastore_product \
-               ) VALUES ( \
-               '" + batchType + "', \
-               " + str(stageID) + ", \
-               '" + survey + "', \
-               '" + dvoDb + "', \
-               '" + datastoreProduct + "' \
-               )"
-
-        self.execute(sql)
-
-        sql = "SELECT MAX(batch_id) FROM batch"
-
-        batchID = -1;
-
-        try:
-            rs = self.executeQuery(sql)
-            rs.first()
-            batchID = rs.getInt(1)
-        except:
-            self.logger.exception("Unable to get batch ID")
-
-        self.logger.debug("Created new batch in ippToPsps database with batchID = %d" % batchID)
-
-        return batchID;
+        self.MAX_FAILS = 3
 
     '''
@@ -132 +98 @@
     Returns a list of processed batch IDs that have not yet been loaded to the ODM
     '''
-    def getBatchIDsUnloadedToODM(self, epoch, dvoGpc1Label):
+    def getUnloadedBatchIDs(self, epoch, dvoGpc1Label, batchType):
 
         sql = "SELECT DISTINCT batch_id \
@@ -139 +105 @@
                AND dvo_db = '" + dvoGpc1Label + "' \
                AND loaded_to_datastore = 1 \
+               AND batch_type = '" + batchType + "' \
                AND loaded_to_ODM = 0"
 
@@ -157 +124 @@
 
     '''
-    Returns a list of processed batch IDs that have not failed to load and are not yet merged, for this epoch and dvo label
-    '''
-    def getUnfinishedBatchIDs(self, epoch, dvoGpc1Label):
+    Returns a list of processed batch IDs that are merge_worthy, but not yet merged, for this epoch and dvo label
+    '''
+    def getUnmergedBatchIDs(self, epoch, dvoGpc1Label, batchType):
 
         sql = "SELECT DISTINCT batch_id \
@@ -165 +132 @@
                WHERE timestamp > '" + epoch + "' \
                AND dvo_db = '" + dvoGpc1Label + "' \
-               AND loaded_to_datastore = 1 \
-               AND merged != 1 \
-               AND loaded_to_ODM != -1"
-
-
-        ids = []
-        try:
-            rs = self.executeQuery(sql)
-            while (rs.next()):
-                ids.append(rs.getInt(1))
-        except:
-            self.logger.exception("Can't query for unfinished batch ids in ipptopsps Db")
-
-        rs.close()
-
-        self.logger.debug("Found %d unfinished items" % len(ids))
+               AND loaded_to_ODM = 1 \
+               AND batch_type = '" + batchType + "' \
+               AND merged != 1"
+
+        ids = []
+        try:
+            rs = self.executeQuery(sql)
+            while (rs.next()):
+                ids.append(rs.getInt(1))
+        except:
+            self.logger.exception("Can't query for unmerged batch ids in ipptopsps Db")
+
+        rs.close()
+
+        self.logger.debug("Found %d un-merged items" % len(ids))
 
         return ids
@@ -300 +266 @@
     '''
     def getStages(self):
-        return ['processed', 'loaded_to_datastore', 'loaded_to_ODM', 'merge_worthy', 'merged', 'deleted_datastore', 'deleted_dxlayer', 'deleted_local']
+        return ['processed', 'loaded_to_datastore', 'loaded_to_ODM', 'merge_worthy', 'deleted_datastore', 'merged', 'deleted_dxlayer', 'deleted_local']
 
     '''
@@ -328 +294 @@
     def getTimeOfLastBatchPublished(self, batchType, epoch, dvoGpc1Label):
 
-        minutes = None
+        seconds = None
 
         sql = "SELECT TIMESTAMPDIFF(SECOND, timestamp, now()) \
@@ -427 +393 @@
 
     '''
+    Is someone processing this item right now? Checks whether this was attempted within the last 2 HOUR
+    '''
+    def processingNow(self, batchType, stage_id, epoch, dvoGpc1Label):
+
+        sql = "SELECT COUNT(*) \
+               FROM batch \
+               WHERE stage_id = " + str(stage_id) + " \
+               AND batch_type = '" + batchType + "' \
+               AND dvo_db = '" + dvoGpc1Label + "' \
+               AND timestamp BETWEEN now() - INTERVAL 2 HOUR AND now()"
+
+        try:
+            rs = self.executeQuery(sql)
+            rs.first()
+            if rs.getInt(1) > 0: 
+                self.logger.errorPair(str(stage_id) + " is already being processed", "skipping")
+                return True
+            else: 
+                return False
+        except:
+            self.logger.exception("Unable to check whether this batch is being processed")
+
+    '''
     Have we already processed and published this batch?
     '''
@@ -454 +443 @@
     Has this stage_id consistently failed to process?
     '''
-    def consistentlyFailed(self, batchType, stage_id, epoch, dvoGpc1Label, count):
+    def consistentlyFailed(self, batchType, stage_id, epoch, dvoGpc1Label):
 
         sql = "SELECT COUNT(*) \
@@ -467 +456 @@
             rs = self.executeQuery(sql)
             rs.first()
-            if rs.getInt(1) >= count: 
-                self.logger.errorPair(str(stage_id) + " has failed %d times" % count, "skipping")
+            if rs.getInt(1) >= self.MAX_FAILS: 
+                self.logger.errorPair(str(stage_id) + " has failed %d times" % self.MAX_FAILS, "skipping")
                 return True
             else: 
@@ -475 +464 @@
             self.logger.exception("Unable to check whether this batch has consistently failed")
 
+    '''
+    Returns a list of stage_ids that have failed the max number of times
+    '''
+    def getConsistentlyFailedIDs(self, batchType, epoch, dvoGpc1Label):
+
+        sql = "SELECT DISTINCT stage_id, COUNT(stage_id) AS numoccur \
+               FROM batch \
+               WHERE batch_type = '" + batchType + "' \
+               AND timestamp > '" + epoch + "' \
+               AND dvo_db = '" + dvoGpc1Label + "' \
+               AND processed = -1 \
+               GROUP BY stage_id HAVING numoccur >= " + str(self.MAX_FAILS)
+
+        ids = []
+        try:
+            rs = self.executeQuery(sql)
+            while (rs.next()):
+                ids.append(rs.getInt(1))
+        except:
+            self.logger.exception("Can't query for consistantly failed ids in ipptopsps Db")
+
+        rs.close()
+
+        self.logger.debug("Found %d consistantly failed stage_ids for batch type='%s'" % (len(ids), batchType))
+
+        return ids
+
+    '''
+    Locks the batch table. This will wait if the lock is held by someone else
+    '''
+    def lockBatchTable(self):
+        self.lockTable("batch")
+
+    '''
+    Creates a new batch. This is done with a locked table in order that no two clients can start work on the same item
+    '''
+    def createNewBatch(self, batchType, stageID, survey, epoch, dvoDb, datastoreProduct, force):
+
+        batchID = -1;
+
+        if force or \
+            (not self.alreadyProcessed(batchType, stageID, epoch, dvoDb) \
+            and not self.processingNow(batchType, stageID, epoch, dvoDb) \
+            and not self.consistentlyFailed(batchType, stageID, epoch, dvoDb)): 
+
+            sql = "INSERT INTO batch ( \
+                   batch_type, \
+                   stage_id, \
+                   survey, \
+                   dvo_db, \
+                   datastore_product \
+                   ) VALUES ( \
+                       '" + batchType + "', \
+                       " + str(stageID) + ", \
+                       '" + survey + "', \
+                       '" + dvoDb + "', \
+                       '" + datastoreProduct + "' \
+                       )"
+
+            self.execute(sql)
+
+            sql = "SELECT MAX(batch_id) FROM batch"
+
+            try:
+                rs = self.executeQuery(sql)
+                rs.first()
+                batchID = rs.getInt(1)
+                self.logger.debug("Created new batch in ippToPsps database with batchID = %d" % batchID)
+            except:
+                self.logger.exception("Unable to get batch ID")
+                batchID = -1
+
+        return batchID;