Changeset 35175 for trunk/ippToPsps

Timestamp:

Feb 16, 2013, 3:36:57 PM (13 years ago)

Author:

eugene

Message:

swap inner and outer loops: do boxes then batches (so a single dvopsps load can be used for both P2 and ST)

Location:

trunk/ippToPsps/jython

Files:

• dvo.py (modified) (1 diff)
• ipptopspsdb.py (modified) (3 diffs)
• loader.py (modified) (2 diffs)

trunk/ippToPsps/jython/dvo.py

@@ -640 +640 @@
         p.wait()
 
+        ## dvopsps can take a very long time. try 2 or 3 times to re-establish communication with the db before proceeding?
+
         self.ippToPspsDb.setIngestedBox(boxId)
 

trunk/ippToPsps/jython/ipptopspsdb.py

@@ -1168 +1168 @@
             rs.close()
         except:
-            self.logger.errorPair("No free stripes", "trying to share with other client...")
+            self.logger.errorPair("No free stripes", "trying to steal boxes from other client...")
             sql = "SELECT ra_center, COUNT(*) AS numPending\
                    FROM box \
@@ -1184 +1184 @@
                 rs.close()
             except:
-                self.logger.errorPair("No available stripes to share", "no stripe obtained")
+                self.logger.errorPair("No outstanding boxes to steal", "no stripe obtained")
 
         self.logger.infoPair("Using RA stripe", "%f" % raCenter)
@@ -1203 +1203 @@
 
         self.registerStripe(host, pid, raCenter)
+
+        self.unlockTables()
+
+        return ids
+
+    '''
+    Returns ids for all boxes with pending items in a stripe of RA not already being processed
+    by another client. If none are available, then it uses the stripe already used by another 
+    client that has the most pending items
+    '''
+    def getStripeBoxIdsAnyBatch(self, host, pid):
+
+        self.execute("LOCK TABLES box READ, pending READ, stripe WRITE, clients WRITE")
+
+        ids = []
+        raCenter = -1.0
+        sql = "SELECT ra_center \
+               FROM box \
+               JOIN pending ON (id = box_id) \
+               WHERE skychunk = '" + self.skychunk.name + "' \
+               AND ra_center NOT IN \
+               (SELECT ra_center \
+                FROM stripe \
+                WHERE skychunk = '" + self.skychunk.name + "') \
+               GROUP BY ra_center LIMIT 1"
+
+        try:
+            rs = self.executeQuery(sql)
+            rs.first()
+            raCenter = rs.getFloat(1)
+            rs.close()
+        except:
+            self.logger.errorPair("No free stripes", "trying to steal boxes from other client...")
+            sql = "SELECT ra_center, COUNT(*) AS numPending\
+                   FROM box \
+                   JOIN pending ON (id = box_id) \
+                   WHERE skychunk = '" + self.skychunk.name + "' \
+                   GROUP BY ra_center \
+                   ORDER BY numPending \
+                   DESC LIMIT 1"
+
+            try:
+                rs = self.executeQuery(sql)
+                rs.first()
+                raCenter = rs.getFloat(1)
+                rs.close()
+            except:
+                self.logger.errorPair("No outstanding boxes to steal", "no stripe obtained")
+
+        self.logger.infoPair("Using RA stripe", "%f" % raCenter)
+        sql = "SELECT DISTINCT id \
+               FROM box \
+               JOIN pending ON (id = box_id) \
+               WHERE ra_center = " + str(raCenter) + " \
+               AND skychunk = '" + self.skychunk.name + "' \
+               ORDER BY dec_center DESC"
+
+        try:
+            rs = self.executeQuery(sql)
+            while (rs.next()): ids.append(rs.getInt(1))
+            rs.close()
+        except:
+            self.logger.errorPair("Can't get stripe boxes", sql)
 
         self.unlockTables()

trunk/ippToPsps/jython/loader.py

@@ -125 +125 @@
         while True:
 
-            abort = False
-            for batchType in self.skychunk.batchTypes:
-
-                # get a stripe's worth of box IDs
-                boxIds = self.ippToPspsDb.getStripeBoxIds(self.HOST, self.PID, batchType)
-
-                numAttempts += 1
-
-                self.logger.infoPair("Boxes with unprocessed data", "%d" % len(boxIds))
-                self.logger.infoPair("numAttempts", str(numAttempts))
-                if len(boxIds) > 0: numAttempts = 0
-
-                for boxId in boxIds:
+            # get a stripe's worth of box IDs
+            boxIds = self.ippToPspsDb.getStripeBoxIdsAnyBatch(self.HOST, self.PID)
+            self.logger.infoPair("Boxes with unprocessed data", "%d" % len(boxIds))
+            self.logger.infoPair("numAttempts", str(numAttempts))
+
+            if len(boxIds) > 0: numAttempts = 0
+            numAttempts += 1
+
+            for boxId in boxIds:
+
+                # get box info. if boxes have changed, break and start again
+                try:
+                    boxDim = self.ippToPspsDb.getBoxDimensions(boxId)
+                except:
+                    self.logger.infoPair("Can't get details for this box", "%d" % boxId)
+                    break
+                self.logger.infoPair("client status", "ok")
+
+                abort = False
+                for batchType in self.skychunk.batchTypes:
+
                     self.logger.infoPair("Check","client status")
                     if not self.checkClientStatus():
@@ -143 +151 @@
                         break
                     self.logger.infoPair("client status", "ok")
-                    # get box info. if boxes have changed, break and start again
-                    try:
-                        boxDim = self.ippToPspsDb.getBoxDimensions(boxId)
-                    except:
-                        self.logger.infoPair("Can't get details for this box", "%d" % boxId)
-                        break
-                    self.logger.infoPair("client status", "ok")
+
                     ids = self.ippToPspsDb.getPendingIdsForThisBox(boxId, batchType)