Index: trunk/ippToPsps/jython/loader.py =================================================================== --- trunk/ippToPsps/jython/loader.py (revision 34661) +++ trunk/ippToPsps/jython/loader.py (revision 35097) @@ -34,13 +34,18 @@ super(Loader, self).__init__(argv, 1, 1) - # create gpc1 database objects + if self.skychunk.parallel: + print "PARALLEL dvo" + else: + print "SERIAL dvo" + + # connect to the gpc1 database self.gpc1Db = Gpc1Db(self.logger, self.config) # connect to scratch database - scratchDbs = ['ipptopsps_scratch', 'ipptopsps_scratch2', 'ipptopsps_scratch3'] - for dbName in scratchDbs: - self.scratchDb = ScratchDb(self.logger, self.config, dbName) - if self.scratchDb.anyOtherConnections(): - self.logger.errorPair("This scratch Db is already in use", dbName) + scratchDbs = ['1', '2', '3'] + for dbVersion in scratchDbs: + self.scratchDb = ScratchDb(self.logger, self.config, dbVersion) + if not self.config.test and self.scratchDb.anyOtherConnections(): + self.logger.errorPair("This scratch Db is already in use", self.scratchDb.dbName) self.scratchDb.disconnect() continue @@ -52,5 +57,5 @@ # create Datastore objects - self.datastore = Datastore(self.logger, self.config, self.ippToPspsDb) + self.datastore = Datastore(self.logger, self.skychunk, self.ippToPspsDb) # if an IN batch is requested, create and quit @@ -60,4 +65,5 @@ batch = InitBatch(self.logger, self.config, + self.skychunk, self.gpc1Db, self.ippToPspsDb, @@ -72,32 +78,37 @@ + # if the 'once' option is passed, we do not loop multiple times in 'run' + self.onePassOnly = 0 + if len(sys.argv) > 2 and sys.argv[2] == "once": + self.onePassOnly = 1 + # set a poll time of about 1 minute self.parsePollTimeArg("0.0166") - self.config.printAll() + self.skychunk.printAll() ''' Overrides base-class version so we can ensure our scratch Db is using the right DVO Db ''' - def refreshConfig(self): - - ret = super(Loader, self).refreshConfig() + def refreshSkychunk(self): + + ret = super(Loader, self).refreshSkychunk() try: self.scratchDb except: return ret - self.dvoDetections = DvoDetections(self.logger, self.config, self.scratchDb.dbName) + self.dvoDetections = DvoDetections(self.logger, self.config, self.skychunk, self.ippToPspsDb, self.scratchDb.dbName) return ret ''' - Overrides base_class version so we can break out of processing loops if the config changes + Overrides base_class version so we can break out of processing loops if the skychunk changes ''' def checkClientStatus(self): - oldConfigName = self.config.name + oldSkychunkName = self.skychunk.name super(Loader, self).checkClientStatus() - if oldConfigName != self.config.name: - self.logger.infoPair("Config changed", "from '" + oldConfigName + "' to '" + self.config.name + "'") - self.config.printAll() + if oldSkychunkName != self.skychunk.name: + self.logger.infoPair("Skychunk changed", "from '" + oldSkychunkName + "' to '" + self.skychunk.name + "'") + self.skychunk.printAll() return False @@ -115,5 +126,5 @@ abort = False - for batchType in self.config.batchTypes: + for batchType in self.skychunk.batchTypes: # get a stripe's worth of box IDs @@ -141,4 +152,6 @@ ids = self.ippToPspsDb.getPendingIdsForThisBox(boxId, batchType) + # ids are the stage_ids for items to be processed (eg, stack_id for a stack CMF, cam_id for P2 smf) + if len(ids) < 1: self.logger.debugPair("No " + batchType + " items found in this box", "skipping") @@ -149,25 +162,54 @@ self.logger.infoPair("Box dimensions", "%.1f / %.1f / %.1f" % (boxDim['RA'], boxDim['DEC'], boxDim['SIDE'])) self.logger.infoPair(batchType + " items found in this box", "%d" % len(ids)) - boxSizeWithBorder = boxDim['SIDE'] + (self.config.BORDER * 2) + boxSizeWithBorder = boxDim['SIDE'] + (self.skychunk.BORDER * 2) + boxSizeSansBorder = boxDim['SIDE'] self.logger.infoPair("got here", "ok") useFullTables = 0 if batchType != "OB": - # look in DVO for this box (with extra border) - self.logger.infoPair("Querying DVO for this sky area", "") - self.dvoDetections.setSkyAreaAsBox(boxDim['RA'], boxDim['DEC'], boxSizeWithBorder) - #self.dvoDetections.setSkyArea() - sizeToBeIngested = self.dvoDetections.getDiskSizeOfRegionsToBeIngested() - if sizeToBeIngested == 0.0: smfsPerGB = 999999999 - else: smfsPerGB = len(ids)/sizeToBeIngested - self.logger.infoPair("DVO to be ingested", "%.1f GB" % sizeToBeIngested) - self.logger.infoPair("smfs-per-GB", "%.1f" % smfsPerGB) - # should do we pre-ingest stuff from DVO? - if batchType == 'P2' and smfsPerGB > 30: - if not self.dvoDetections.sync(): - self.logger.errorPair("Could not sync DVO with MySQL", "skipping") - continue - + + if (self.dvoDetections.useStilts): + # look in DVO for this box (with extra border) + self.logger.infoPair("Querying DVO for this sky area", "") + self.dvoDetections.setSkyAreaAsBox(boxDim['RA'], boxDim['DEC'], boxSizeWithBorder) + #self.dvoDetections.setSkyArea() + sizeToBeIngested = self.dvoDetections.getDiskSizeOfRegionsToBeIngested() + if sizeToBeIngested == 0.0: smfsPerGB = 999999999 + else: smfsPerGB = len(ids)/sizeToBeIngested + self.logger.infoPair("DVO to be ingested", "%7.1e GB" % sizeToBeIngested) + self.logger.infoPair("smfs-per-GB", "%.1f" % smfsPerGB) + + # should do we pre-ingest stuff from DVO? + # NOTE EAM : this skychunk loads the dvo detections into the mysql db + # XXXX EAM : this should happen for both P2 and Stack detection + # XXXX EAM : in parallel model, this should happen for all P2 & Stack batches + # + if (batchType == 'P2' or batchType == 'ST') and smfsPerGB > 30: + if not self.dvoDetections.sync(): + self.logger.errorPair("Could not sync DVO with MySQL", "skipping") + continue + + useFullTables = 1 + else: + # XXX EAM : this is not currently an + # option. either remove the non-native + # ingest code or make it an option + # if we are using the native loader, always use it + # need to work out a good boundary / region strategy in coordination with + # impact of mysql insertion + if not self.ippToPspsDb.isBoxIngested(boxId): + self.dvoDetections.nativeIngestDetections(boxId, boxDim['RA'], boxDim['DEC'], boxSizeSansBorder) + useFullTables = 1 + ## else: + ## # if we are using the native loader, always use it + ## # need to work out a good boundary / region strategy in coordination with + ## # impact of mysql insertion + ## self.dvoObjects.nativeIngestObjects(boxDim['RA'], boxDim['DEC'], boxSizeSansBorder) + ## useFullTables = 1 + + ''' + NOTE EAM : "TheseItems" refers to each of the stack CMFs, camera SMFs, or object batches + ''' self.logger.infoBool("Using pre-ingested DVO data?", useFullTables) if not self.processTheseItems(batchType, ids, useFullTables): @@ -175,7 +217,9 @@ break self.logger.infoPair("processed","ok") + if abort or not self.checkClientStatus(): abort = True elif numAttempts > 1 and not self.waitForPollTime(): break + if self.onePassOnly: self.exitProgram("one pass completed") ''' @@ -210,4 +254,5 @@ batch = DetectionBatch(self.logger, self.config, + self.skychunk, self.gpc1Db, self.ippToPspsDb, @@ -219,4 +264,5 @@ batch = StackBatch(self.logger, self.config, + self.skychunk, self.gpc1Db, self.ippToPspsDb, @@ -229,4 +275,5 @@ batch = ObjectBatch(self.logger, self.config, + self.skychunk, self.gpc1Db, self.ippToPspsDb, @@ -260,5 +307,5 @@ ''' def printUsage(self): - super(Cleanup, self).printUsage("<[init]>") + super(Loader, self).printUsage("<[init]>")