IPP Software Navigation Tools IPP Links Communication Pan-STARRS Links

Changeset 33490


Ignore:
Timestamp:
Mar 13, 2012, 10:04:15 AM (14 years ago)
Author:
rhenders
Message:

no need to pass config object to methods as it's a class field; now using temporary table to store all pending stuff before cutting up into squares; clients now registers with one 'stripe' of RA

File:
1 edited

Legend:

Unmodified
Added
Removed

  • trunk/ippToPsps/jython/ipptopspsdb.py

    r33354 r33490  
    141141    want to load a test batch to PSPS
    142142
    143     We also check that these batches are more than 4 HOURs old, so that we don't attempt to
     143    We also check that these batches are more than 6 HOURs old, so that we don't attempt to
    144144    publish something simulataneously with the client that it producing it
    145145    '''
    … …  
    154154               AND loaded_to_datastore != 1 \
    155155               AND timestamp > '2011-10-27' \
    156                AND timestamp < now() -  INTERVAL 4 HOUR"
     156               AND timestamp < now() -  INTERVAL 6 HOUR"
    157157
    158158        ids = []
    … …  
    161161            while (rs.next()):
    162162                ids.append(rs.getInt(1))
     163            rs.close()
    163164        except:
    164165            self.logger.exception("Can't query for processed batch ids not loaded to datastore ipptopsps Db")
    165 
    166         rs.close()
    167166
    168167        self.logger.debug("Found %d processed but-not-on-datastore items" % len(ids))
    … …  
    274273        except:
    275274            self.logger.exception("Can't query for batch ids with " + column + " = " + str(value) + " in ipptopsps Db")
    276 
    277         rs.close()
     275        finally:
     276            rs.close()
    278277
    279278        self.logger.debug("Found %d batches with %s = %d" % (len(ids), column, value))
    … …  
    284283    Returns the total detections published to the datastore for this epoch, dvo label and batch type
    285284    '''
    286     def getTotalDetectionsPublished(self, batchType, epoch, dvoGpc1Label):
     285    def getTotalDetectionsPublished(self, batchType, epoch, dvoGpc1Label, interval="10 YEAR"):
    287286
    288287        sql = "SELECT SUM(total_detections) \
    … …  
    292291               AND dvo_db = '" + dvoGpc1Label + "' \
    293292               AND processed = 1 \
    294                AND loaded_to_datastore = 1"
     293               AND loaded_to_datastore = 1 \
     294               AND timestamp BETWEEN now() - INTERVAL " + interval + " AND now()"
    295295
    296296        total = -1
    … …  
    301301        except:
    302302            self.logger.exception("Can't query for total detections published")
    303 
    304         rs.close()
     303        finally:
     304            rs.close()
    305305
    306306        self.logger.debug("Found %d detections" % total)
    … …  
    372372    '''
    373373    def getStages(self):
    374         return ['processed', 'loaded_to_datastore', 'loaded_to_ODM', 'merge_worthy', 'deleted_datastore', 'deleted_dxlayer', 'merged', 'deleted_local' ]
     374        return [
     375        'processed',
     376        'loaded_to_datastore',
     377        'loaded_to_ODM',
     378        'merge_worthy',
     379        'deleted_datastore',
     380        'deleted_dxlayer',
     381        'merged',
     382        'deleted_local' ]
    375383
    376384    '''
    377385    Returns a count of batches processed in the specified time period.
    378386    '''
    379     def countBatchesInLastPeriod(self, batchType, epoch, dvoGpc1Label, interval):
     387    def countBatches(self, batchType, epoch, dvoGpc1Label, interval="10 YEAR"):
    380388
    381389        sql = "SELECT COUNT(batch_id) \
    … …  
    393401            self.logger.exception("Unable to count batches in interval")
    394402            return 0
     403        finally:
     404            rs.close()
    395405
    396406    '''
    … …  
    497507
    498508    '''
    499     Is someone processing this item right now? Checks whether this was attempted within the last 2 HOUR
    500     '''
    501     def processingNow(self, batchType, stage_id, epoch, dvoGpc1Label):
     509    Is someone processing this item right now? Checks whether this was attempted within the last 6 HOUR
     510    '''
     511    def processingNow(self, batchType, stage_id):
    502512
    503513        sql = "SELECT COUNT(*) \
    … …  
    505515               WHERE stage_id = " + str(stage_id) + " \
    506516               AND batch_type = '" + batchType + "' \
    507                AND dvo_db = '" + dvoGpc1Label + "' \
    508                AND timestamp BETWEEN now() - INTERVAL 4 HOUR AND now()"
     517               AND dvo_db = '" + self.config.dvoLabel + "' \
     518               AND batch_type != 'IN' \
     519               AND timestamp BETWEEN now() - INTERVAL 6 HOUR AND now()"
    509520
    510521        try:
    … …  
    541552    Have we already processed this stage_id?
    542553    '''
    543     def alreadyProcessed(self, batchType, stage_id, epoch, dvoGpc1Label):
     554    def alreadyProcessed(self, batchType, stage_id):
    544555
    545556        sql = "SELECT COUNT(*) \
    … …  
    547558               WHERE stage_id = " + str(stage_id) + " \
    548559               AND batch_type = '" + batchType + "' \
    549                AND timestamp > '" + epoch + "' \
    550                AND dvo_db = '" + dvoGpc1Label + "' \
     560               AND timestamp > '" + self.config.epoch + "' \
     561               AND dvo_db = '" + self.config.dvoLabel + "' \
     562               AND batch_type != 'IN' \
    551563               AND processed = 1"
    552564
    … …  
    565577    Has this stage_id consistently failed to process?
    566578    '''
    567     def consistentlyFailed(self, batchType, stage_id, epoch, dvoGpc1Label):
     579    def consistentlyFailed(self, batchType, stage_id):
    568580
    569581        sql = "SELECT COUNT(*) \
    … …  
    571583               WHERE stage_id = " + str(stage_id) + " \
    572584               AND batch_type = '" + batchType + "' \
    573                AND timestamp > '" + epoch + "' \
    574                AND dvo_db = '" + dvoGpc1Label + "' \
     585               AND timestamp > '" + self.config.epoch + "' \
     586               AND dvo_db = '" + self.config.dvoLabel + "' \
    575587               AND processed = -1"
    576588
    … …  
    620632
    621633    '''
    622     Creates a new batch. This is done with a locked table in order that no two clients can start work on the same item
    623     '''
    624     def createNewBatch(self, batchType, stageID, config):
     634    Creates a new batch. This is done with a locked table in order that no two clients can start work on the same item, i.e. this is a crital section
     635    '''
     636    def createNewBatch(self, batchType, stageID):
    625637
    626638        batchID = -1;
    627639
    628         if config.force or \
    629             (not self.alreadyProcessed(batchType, stageID, config.epoch, config.dvoLabel) \
    630             and not self.processingNow(batchType, stageID, config.epoch, config.dvoLabel) \
    631             and not self.consistentlyFailed(batchType, stageID, config.epoch, config.dvoLabel)):
     640        if self.config.force or \
     641            (not self.alreadyProcessed(batchType, stageID) \
     642            and not self.processingNow(batchType, stageID) \
     643            and not self.consistentlyFailed(batchType, stageID)):
    632644
    633645            sql = "INSERT INTO batch ( \
    … …  
    640652                       '" + batchType + "', \
    641653                       " + str(stageID) + ", \
    642                        '" + config.survey + "', \
    643                        '" + config.dvoLabel + "', \
    644                        '" + config.datastoreProduct + "' \
     654                       '" + self.config.survey + "', \
     655                       '" + self.config.dvoLabel + "', \
     656                       '" + self.config.datastoreProduct + "' \
    645657                       )"
    646658
    … …  
    712724
    713725    '''
     726    Pause all loading clients
     727    '''
     728    def pauseAllLoaders(self, bool):
     729        self.setColForAllLoaders("paused", bool)
     730
     731    '''
     732    Kill all loading clients
     733    '''
     734    def killAllLoaders(self, bool):
     735        self.setColForAllLoaders("killed", bool)
     736
     737    '''
     738    Sets the value for a given column in the clients table
     739    '''
     740    def setColForAllLoaders(self, col, bool):
     741        if bool: value = 1
     742        else: value = 0
     743        self.execute("UPDATE clients SET " + col + " = " + str(value) + " WHERE type = 'loader.py' AND config = '" + self.config.name + "'")
     744
     745    '''
    714746    Remove client
    715747    '''
    … …  
    720752    Insert client
    721753    '''
    722     def insertClient(self, type, host, pid, config):
    723         self.execute("INSERT INTO clients (timestamp, type, host, pid, config) VALUES (now(), '" + type + "', '" + host + "', " + str(pid) + ", '" + config + "')")
     754    def insertClient(self, type, host, pid):
     755        self.execute("INSERT INTO clients (timestamp, type, host, pid, config) VALUES (now(), '" + type + "', '" + host + "', " + str(pid) + ", '" + self.config.name + "')")
    724756
    725757    '''
    726758    Update clients, or inserts it for the first time
    727759    '''
    728     def updateClient(self, type, host, pid, config):
    729 
    730         try:
    731             self.insertClient(type, host, pid, config)
     760    def updateClient(self, type, host, pid):
     761
     762        try:
     763            self.insertClient(type, host, pid)
    732764        except:
    733765            self.execute("UPDATE clients SET timestamp = now() WHERE host = '" + host + "' AND pid = " + str(pid))
    … …  
    763795
    764796        return False
    765 
    766797
    767798    '''
    … …  
    851882    Gets id of box with these coords
    852883    '''
    853     def getBoxId(self, config, ra, dec, boxSide):
     884    def getBoxId(self, ra, dec):
    854885
    855886        sql = "SELECT id FROM box \
    856                WHERE config  = '" + config + "' \
     887               WHERE config  = '" + self.config.name + "' \
    857888               AND ra_center = " + str(ra) + " \
    858889               AND dec_center = " + str(dec) + " \
    859                AND box_side = " + str(boxSide)
     890               AND box_side = " + str(self.config.boxSize)
    860891
    861892        id = -1
    … …  
    869900        return id
    870901
    871 
    872902    '''
    873903    Remove all boxes and associated ids in pending table
    874904    '''
    875     def removeAllBoxes(self, config):
    876         self.execute("DELETE FROM box WHERE config = '" + config + "'")
    877 
    878     '''
    879     Removes a box and it's associated ids in the pending table
    880     '''
    881     def removeBox(self, config, ra, dec, boxSide):
    882         self.execute("DELETE FROM box \
    883                 WHERE config = '" + config + "' \
    884                 AND ra_center = " + str(ra) + " \
    885                 AND dec_center = " + str(dec) + " \
    886                 AND box_side = " + str(boxSide))
     905    def removeAllBoxes(self):
     906        self.execute("DELETE FROM box WHERE config = '" + self.config.name + "'")
    887907
    888908    '''
    889909    Inserts new box into box table. If it already exisits, it returns existing box id
    890910    '''
    891     def insertBox(self, config, ra, dec, boxSide):
    892 
    893         id = self.getBoxId(config, ra, dec, boxSide)
     911    def insertBox(self, ra, dec):
     912
     913        id = self.getBoxId(ra, dec)
    894914
    895915        # does not already exists so insert
    … …  
    899919                   (config, ra_center, dec_center, box_side) \
    900920                   VALUES \
    901                    ('" + config + "', " + str(ra) + ", " + str(dec) + ", " + str(boxSide) + ")"
     921                   ('" + self.config.name + "', " + str(ra) + ", " + str(dec) + ", " + str(self.config.boxSize) + ")"
    902922
    903923            self.execute(sql)
    904             id = self.getBoxId(config, ra, dec, boxSide)
     924            id = self.getBoxId(ra, dec)
    905925
    906926        return id
    … …  
    912932
    913933        # first delete old pending items
    914         self.execute("DELETE FROM pending WHERE box_id = " + str(box_id))
     934        self.execute("DELETE FROM pending \
     935                WHERE box_id = " + str(box_id) + " \
     936                AND batch_type = '" + batchType + "'")
    915937
    916938        for id in ids:
    … …  
    923945
    924946    '''
    925     Returns ids for all boxes with pending items
    926     '''
    927     def getBoxIds(self, config):
    928 
     947    Returns ids for all boxes with pending items in a stripe of RA not already being processed
     948    by another client. If none are available, then it uses the stripe already used by another
     949    client that has the most pending items
     950    '''
     951    def getStripeBoxIds(self, host, pid, batchType):
     952
     953        self.execute("LOCK TABLES box READ, pending READ, stripe WRITE, clients WRITE")
     954
     955        ids = []
     956        raCenter = -1.0
     957        sql = "SELECT ra_center \
     958               FROM box \
     959               JOIN pending ON (id = box_id) \
     960               WHERE config = '" + self.config.name + "' \
     961               AND batch_type = '" + batchType + "' \
     962               AND ra_center NOT IN \
     963               (SELECT ra_center \
     964                FROM stripe \
     965                WHERE config = '" + self.config.name + "') \
     966               GROUP BY ra_center LIMIT 1"
     967
     968        try:
     969            rs = self.executeQuery(sql)
     970            rs.first()
     971            raCenter = rs.getFloat(1)
     972            rs.close()
     973        except:
     974            self.logger.errorPair("No available stripes", "sharing with other client")
     975            sql = "SELECT ra_center, COUNT(*) AS numPending\
     976                   FROM box \
     977                   JOIN pending ON (id = box_id) \
     978                   WHERE config = '" + self.config.name + "' \
     979                   AND batch_type = '" + batchType + "' \
     980                   GROUP BY ra_center \
     981                   ORDER BY numPending \
     982                   DESC LIMIT 1"
     983
     984            try:
     985                rs = self.executeQuery(sql)
     986                rs.first()
     987                raCenter = rs.getFloat(1)
     988                rs.close()
     989            except:
     990                self.logger.errorPair("Still no available stripes", "")
     991
     992        self.logger.infoPair("Using RA stripe", "%f" % raCenter)
    929993        sql = "SELECT DISTINCT id \
    930994               FROM box \
    931                JOIN pending ON (id = box_id)  \
    932                WHERE config = '" + config + "' \
    933                ORDER BY ra_center, dec_center"
    934 
    935         ids = []
     995               JOIN pending ON (id = box_id) \
     996               WHERE ra_center = " + str(raCenter) + " \
     997               AND batch_type = '" + batchType + "' \
     998               AND config = '" + self.config.name + "'"
     999
    9361000        try:
    9371001            rs = self.executeQuery(sql)
    … …  
    9391003            rs.close()
    9401004        except:
    941             self.logger.errorPair("Can't get boxes", sql)
    942 
    943         return ids
     1005            self.logger.errorPair("Can't get stripe boxes", sql)
     1006
     1007        self.registerStripe(host, pid, raCenter)
     1008
     1009        self.unlockTables()
     1010
     1011        return ids
     1012
     1013    '''
     1014    Inserts row into strip table
     1015    '''
     1016    def registerStripe(self, host, pid, raCenter):
     1017       
     1018        # first, delete any existing record
     1019        try:
     1020            self.execute("DELETE FROM stripe \
     1021                    WHERE client_id = \
     1022                    (SELECT id FROM clients \
     1023                     WHERE host = '" + host + "' AND pid = " + str(pid) + ")")
     1024        except:
     1025            self.logger.errorPair("Can't delete from stripe table", sql)
     1026
     1027
     1028        # now insert new stripe entry
     1029        sql = "INSERT INTO stripe \
     1030               (client_id, config, ra_center) \
     1031               SELECT clients.id, clients.config, " + str(raCenter) + " \
     1032               FROM clients \
     1033               WHERE host = '" + host + "' \
     1034               AND pid = " + str(pid)
     1035        try:
     1036            self.execute(sql)
     1037        except:
     1038            self.logger.errorPair("Can't insert into stripe", sql)
     1039
    9441040
    9451041    '''
    … …  
    9611057
    9621058    '''
    963     Returns ids for pending items in this box for this batch type
    964     '''
    965     def getPendingIds(self, boxId, batchType):
    966 
    967         sql = "SELECT stage_id FROM pending WHERE batch_type = '" + batchType + "' AND box_id = " + str(boxId)
     1059    Returns ids for pending items for this config
     1060    '''
     1061    def getPendingIds(self):
     1062
     1063        sql = "SELECT DISTINCT stage_id FROM pending JOIN box ON (box_id = id) WHERE config = '" + self.config.name + "'"
    9681064
    9691065        ids = []
    … …  
    9731069            rs.close()
    9741070        except:
     1071            self.logger.errorPair("Can't get pending ids for this config", sql)
     1072
     1073        return ids
     1074
     1075    '''
     1076    Returns ids for pending items in this box for this batch type
     1077    '''
     1078    def getPendingIdsForThisBox(self, boxId, batchType):
     1079
     1080        sql = "SELECT stage_id FROM pending WHERE batch_type = '" + batchType + "' AND box_id = " + str(boxId)
     1081
     1082        ids = []
     1083        try:
     1084            rs = self.executeQuery(sql)
     1085            while (rs.next()): ids.append(rs.getInt(1))
     1086            rs.close()
     1087        except:
    9751088            self.logger.errorPair("Can't get ids from this box", sql)
    9761089
    … …  
    9781091
    9791092    '''
    980     Returns max number of pensing items in a box for this config
    981     '''
    982     def getMaxPendingInBox(self, config):
    983 
    984         sql = "SELECT MAX(pending) \
    985                FROM (SELECT COUNT(stage_id) AS pending \
     1093    Deletes an item from the pending table
     1094    '''
     1095    def deletePendingItem(self, batchType, stageID):
     1096
     1097        sql = "DELETE pending \
     1098               FROM pending \
     1099               JOIN box ON (id = box_id) \
     1100               WHERE stage_id = " + str(stageID) + " \
     1101               AND batch_type = '" + batchType + "' \
     1102               AND config = '" + self.config.name + "'"
     1103
     1104        self.execute(sql)
     1105
     1106    '''
     1107    Writes density-plot data to file for pending stuff over sky for given config
     1108    '''
     1109    def createPendingDensityPlotData(self, batchType, DATFILE):
     1110
     1111        sql = "SELECT id, ra_center, dec_center \
     1112               FROM box \
     1113               WHERE config = '" + self.config.name + "' \
     1114               ORDER BY ra_center, dec_center"
     1115
     1116        maxPending = 0
     1117        try:
     1118            rs = self.executeQuery(sql)
     1119            while (rs.next()):
     1120               
     1121                boxId = rs.getInt(1)
     1122
     1123                sql = "SELECT COUNT(stage_id) \
    9861124                       FROM pending \
    987                        JOIN box ON (id = box_id) \
    988                        WHERE config = '" + config + "' \
    989                        GROUP BY box_id) AS t"
    990 
    991         try:
    992             rs = self.executeQuery(sql)
    993             rs.first()
    994             return rs.getInt(1)
    995         except:
    996             self.logger.errorPair("Could not count max pending", sql)
    997             return 100
    998 
    999     '''
    1000     Writes density-plot data to file for pending stuff over sky for given config
    1001     '''
    1002     def createPendingDensityPlotData(self, config, DATFILE):
    1003 
    1004         sql = "SELECT ra_center, dec_center, COUNT(stage_id) \
    1005                FROM box \
    1006                LEFT JOIN pending ON (id = box_id) \
    1007                WHERE config = '" + config + "' \
    1008                GROUP BY id \
    1009                ORDER BY ra_center, dec_center"
    1010 
    1011         try:
    1012             rs = self.executeQuery(sql)
    1013             while (rs.next()):
    1014                 print >> DATFILE, rs.getFloat(1), rs.getFloat(2), rs.getInt(3)
     1125                       WHERE box_id = " + str(boxId) + " \
     1126                       AND batch_type = '" + batchType + "'"
     1127                rs2 = self.executeQuery(sql)
     1128                rs2.first()
     1129                max = rs2.getInt(1)
     1130                rs2.close()
     1131                print >> DATFILE, rs.getFloat(2), rs.getFloat(3), max
     1132                if max > maxPending: maxPending = max
    10151133            rs.close()
    10161134        except:
    10171135            self.logger.errorPair("Can't get density plot data", sql)
     1136
     1137        return maxPending
     1138
     1139    '''
     1140    Creates a temporary table and shoves it full of stage_ids and ra/dec coords
     1141    '''
     1142    def storeAllItems(self, rows):
     1143
     1144        try:
     1145            self.execute("DROP TABLE all_pending")
     1146        except:
     1147            pass
     1148
     1149        self.execute("CREATE TEMPORARY TABLE all_pending (stage_id bigint(20), ra_bore float, dec_bore float)")
     1150        count = 0
     1151        for row in rows:
     1152
     1153            try:
     1154                sql = "INSERT INTO all_pending (stage_id, ra_bore, dec_bore) \
     1155                       VALUES (%d, %f, %f)" % (row[0], row[1], row[2])
     1156                self.execute(sql)
     1157                count += 1
     1158            except: continue
     1159
     1160        self.logger.infoPair("All items written to Db", "%d" % count)
     1161
     1162    '''
     1163    Gets all items in the all_pending temporary table within the bounds of this box
     1164    '''
     1165    def getItemsInThisThisBox(self, ra, dec):
     1166
     1167        halfSide = self.config.boxSize/2.0
     1168        minRa =  ra-halfSide
     1169        maxRa = ra+halfSide
     1170        minDec = dec-halfSide
     1171        maxDec = dec+halfSide
     1172
     1173        ids = []
     1174
     1175        sql = "SELECT DISTINCT stage_id \
     1176               FROM all_pending \
     1177               WHERE ra_bore BETWEEN " + str(minRa) + " AND " + str(maxRa) + " \
     1178               AND dec_bore BETWEEN " + str(minDec) + " AND " + str(maxDec)
     1179
     1180        try:
     1181            rs = self.executeQuery(sql)
     1182            while (rs.next()): ids.append(rs.getInt(1))
     1183        except:
     1184            self.logger.errorPair("Can't get items in this box", sql)
     1185
     1186        return ids
    10181187
    10191188    '''
Note: See TracChangeset for help on using the changeset viewer.