Changeset 32468 for trunk/ippToPsps

Timestamp:

Sep 29, 2011, 3:33:14 PM (15 years ago)

Author:

rhenders

Message:

big changes to support multi-client loading; now locks batch table when looking for new item to process, then releases it to perform the processing; gets a list of consistently failing batches and removes them from master list to save time during main processing loop;

File:

• trunk/ippToPsps/jython/load.py (modified) (8 diffs)

trunk/ippToPsps/jython/load.py

@@ -7 +7 @@
 import time
 import sys
+import os
+import socket
 import logging.config
 from xml.etree.ElementTree import ElementTree, Element, tostring
@@ -35 +37 @@
         logger.infoPair("Batch name", batchName)
 
-        Batch.publishToDatastore(datastore, ippToPspsDb, batchID, batchName, subDir, tarballFile)
-
+        Batch.publishToDatastore(datastore, batchID, batchName, subDir, tarballFile)
 
 '''
@@ -67 +68 @@
     else:
         processedIDs = ippToPspsDb.getProcessedIDs(batchType, EPOCH, DVOGPC1LABEL)
-        logger.infoPair("Processed %s items" % batchType, "%d" % len(processedIDs))
-        ids = list(set(allIDs) - set(processedIDs))
-        logger.infoPair("Unprocessed %s items" % batchType, "%d" % len(ids))
+        consistaentlyFailedIDs = ippToPspsDb.getConsistentlyFailedIDs(batchType, EPOCH, DVOGPC1LABEL)
+        ids = list(set(allIDs) - set(processedIDs) - set(consistaentlyFailedIDs))
+        logger.infoPair("Processed items", "%d" % len(processedIDs))
+        logger.infoPair("Consistently failing items", "%d" % len(consistaentlyFailedIDs))
+        logger.infoPair("Unprocessed items", "%d" % len(ids))
 
     logger.infoPair("%s items queued" % batchType, "%d" % len(ids))
 
     # loop round IDs of all items to be processed
+    ippToPspsDb.lockBatchTable()
     for id in ids:
 
-        if not FORCE and ippToPspsDb.alreadyProcessed(batchType, id, EPOCH, DVOGPC1LABEL): continue
-        if not FORCE and ippToPspsDb.consistentlyFailed(batchType, id, EPOCH, DVOGPC1LABEL, 3): continue
+        batchID = ippToPspsDb.createNewBatch(
+                batchType,
+                id,
+                SURVEY,
+                EPOCH,
+                DVOGPC1LABEL,
+                datastore.product,
+                FORCE)
+        
+        if batchID < 0: continue;
+
+        ippToPspsDb.unlockTables()
 
         if batchType == "P2":
@@ -85 +99 @@
                     gpc1Db,
                     ippToPspsDb,
-                    id)
+                    id,
+                    batchID)
         elif batchType == "ST":
             batch = StackBatch(logger,
@@ -92 +107 @@
                     gpc1Db,
                     ippToPspsDb,
-                    id)
+                    id,
+                    batchID)
 
         batch.run()
+
+        ippToPspsDb.lockBatchTable()
+
+        logger.infoSeparator()
 
         # if in TEST mode, then quit after one batch
         if TEST: break
 
+    ippToPspsDb.unlockTables()
 
 '''
@@ -122 +143 @@
 logger = logging.getLogger("ippToPspsLog")
 
+# get hostnamee and PID for unique log naming
+HOST = socket.gethostname()
+PID = os.getpid()
+
 # if in test mode, print log to screen, otherwise, only to file
-logger.setup(configDoc, "ippToPsps", TEST, not TEST)
+logger.setup(configDoc, "ippToPsps", TEST, not TEST, HOST, PID)
 
 # create database objects
 gpc1Db = Gpc1Db(logger, configDoc)
 ippToPspsDb = IppToPspsDb(logger, configDoc)
-datastore = Datastore(logger, configDoc)
+datastore = Datastore(logger, configDoc, ippToPspsDb)
 
 # check we connected ok
@@ -143 +168 @@
 QUEUE_P2 = int(configDoc.find("options/queueP2").text)
 QUEUE_ST = int(configDoc.find("options/queueST").text)
+SURVEY = configDoc.find("options/survey").text
 
 # prompt user: FORCE and PUBLISH is a dangerous combination
@@ -180 +206 @@
     # queue all items for all batch types, sequentially
     if QUEUE_IN:
-       batch = InitBatch(logger, CONFIG, configDoc, gpc1Db, ippToPspsDb)
-       batch.run()
+       batchID = ippToPspsDb.createNewBatch(
+                 "IN",
+                 0,
+                 SURVEY,
+                 EPOCH,
+                 DVOGPC1LABEL,
+                 datastore.product,
+                 1)
+       if batchID > 0:
+           batch = InitBatch(logger, CONFIG, configDoc, gpc1Db, ippToPspsDb, batchID)
+           batch.run()
        break;