Changeset 18839 for trunk/ppMerge/src/ppMergeReadChunk.c

Timestamp:

Jul 31, 2008, 2:13:59 PM (18 years ago)

Author:

eugene

Message:

adding multithread capability

File:

• trunk/ppMerge/src/ppMergeReadChunk.c (modified) (4 diffs)

trunk/ppMerge/src/ppMergeReadChunk.c

@@ -1 +1 @@
 # include "ppMerge.h"
 
-ppMergeFileGroup *ppMergeReadChunk (psArray *fileGroups, pmConfig *config, int numChunk) {
+ppMergeFileGroup *ppMergeReadChunk (bool *status, psArray *fileGroups, pmConfig *config, int numChunk) {
+
+    *status = true;
+
+    bool mdok;
+    bool haveMasks = psMetadataLookupBool(&mdok, config->arguments, "INPUTS.MASKS"); // Do we have masks?
+    bool haveWeights = psMetadataLookupBool(&mdok, config->arguments, "INPUTS.WEIGHTS"); // Do we have weights?
+    int rows = psMetadataLookupS32(NULL, config->arguments, "ROWS"); // Number of rows to read per chunk
 
     // select an available fileGroup
- 
-    bool haveMasks = psMetadataLookupBool(&mdok, arguments, "INPUTS.MASKS"); // Do we have masks?
-    bool haveWeights = psMetadataLookupBool(&mdok, arguments, "INPUTS.WEIGHTS"); // Do we have weights?
-
     while (1) {
         // check for any fileGroups which can read data
@@ -14 +17 @@
             if (fileGroup->read) continue;
 
+            // find max last scan so far
+            int lastScan = 0;
+            for (int i = 0; i < fileGroups->n; i++) {
+                ppMergeFileGroup *fileGroup = fileGroups->data[i];
+                lastScan = PS_MAX (fileGroup->lastScan, lastScan);
+            }
+            fileGroup->firstScan = lastScan;
+            fileGroup->lastScan = lastScan + rows;
+
             psArray *readouts = fileGroup->readouts;
+
+            psTimerStart ("ppMergeReadChunk");
 
             psTrace("ppStack", 2, "Reading data for chunk %d into fileGroup %d....n", numChunk, j);
@@ -20 +34 @@
                 pmReadout *inRO = readouts->data[i]; // Input readout
 
+                // override the recorded last scan
+                inRO->thisImageScan  = fileGroup->firstScan;
+                inRO->thisWeightScan = fileGroup->firstScan;
+                inRO->thisMaskScan   = fileGroup->firstScan;
+
                 // Read a chunk from a file
                 pmFPAfile *file = pmFPAfileSelectSingle(config->files, "PPMERGE.INPUT", i); 
-                if (!pmReadoutReadChunk(inRO, file->fits, 0, rows, 0, config)) { 
-                    psError(PS_ERR_IO, false, "Unable to read chunk %d for file PPMERGE.INPUT %d", numChunk, i);
-                    return NULL;
-                }                                                       
 
-                if (haveMasks) {
-                    pmFPAfile *file = pmFPAfileSelectSingle(config->files, "PPMERGE.INPUT.MASK", i); 
-                    if (!pmReadoutReadChunkMask(inRO, file->fits, 0, rows, 0, config)) { 
-                        psError(PS_ERR_IO, false, "Unable to read chunk %d for file PPMERGE.INPUT.MASK %d", numChunk, NAME, i);
+                bool keepReading = false;
+                if (pmReadoutMore(inRO, file->fits, 0, rows, config)) {
+                    keepReading = true;
+                    if (!pmReadoutReadChunk(inRO, file->fits, 0, rows, 0, config)) { 
+                        psError(PS_ERR_IO, false, "Unable to read chunk %d for file PPMERGE.INPUT %d", numChunk, i);
+                        *status = false;
                         return NULL;
                     }                                                   
                 }
 
-                if (haveWeights) {
-                    pmFPAfile *file = pmFPAfileSelectSingle(config->files, "PPMERGE.INPUT.WEIGHT", i); 
-                    if (!pmReadoutReadChunkWeight(inRO, file->fits, 0, rows, 0, config)) {
-                        psError(PS_ERR_IO, false, "Unable to read chunk %d for file PPMERGE.INPUT.WEIGHT %d", numChunk, NAME, i);
+                if (haveMasks && pmReadoutMoreMask(inRO, file->fits, 0, rows, config)) {
+                    keepReading = true;
+                    pmFPAfile *file = pmFPAfileSelectSingle(config->files, "PPMERGE.INPUT.MASK", i); 
+                    if (!pmReadoutReadChunkMask(inRO, file->fits, 0, rows, 0, config)) { 
+                        psError(PS_ERR_IO, false, "Unable to read chunk %d for file PPMERGE.INPUT.MASK %d", numChunk, i);
+                        *status = false;
                         return NULL;
                     }                                                   
                 }
+
+                if (haveWeights && pmReadoutMoreWeight(inRO, file->fits, 0, rows, config)) {
+                    keepReading = true;
+                    pmFPAfile *file = pmFPAfileSelectSingle(config->files, "PPMERGE.INPUT.WEIGHT", i); 
+                    if (!pmReadoutReadChunkWeight(inRO, file->fits, 0, rows, 0, config)) {
+                        psError(PS_ERR_IO, false, "Unable to read chunk %d for file PPMERGE.INPUT.WEIGHT %d", numChunk, i);
+                        *status = false;
+                        return NULL;
+                    }                                                   
+                }
+                if (!keepReading) {
+                    return NULL;
+                }
             }
+
             fileGroup->read = fileGroup->busy = true;
             return fileGroup;
@@ -48 +81 @@
 
         // check for any fileGroups which are done processing
-        bool wait = true;
-        bool more = true;
+        bool wait = false;
         for (int j = 0; j < fileGroups->n; j++) {
             ppMergeFileGroup *fileGroup = fileGroups->data[j];
             if (!fileGroup->read || fileGroup->busy) continue;
-            
-            wait = false;
-            psArray *readouts = fileGroup->readouts;
-            // any more data to be read?
-            for (int i = 0; i < readouts->n && more; i++) {
-                pmReadout *inRO = readouts->data[i];
-
-                pmFPAfile *file = pmFPAfileSelectSingle(config->files, "PPMERGE.INPUT", i);
-                more &= pmReadoutMore(inRO, file->fits, 0, rows, config);
-
-                if (haveMasks) {
-                    pmFPAfile *file = pmFPAfileSelectSingle(config->files, "PPMERGE.INPUT.MASK", i);
-                    more &= pmReadoutMoreMask(inRO, file->fits, 0, rows, config);
-                }
-                if (haveWeights) {
-                    pmFPAfile *file = pmFPAfileSelectSingle(config->files, "PPMERGE.INPUT.WEIGHT", i);
-                    more &= pmReadoutMoreWeight(inRO, file->fits, 0, rows, config);
-                }
-            }
             fileGroup->read = false;
+            wait = true;
         }
-        if (!more) return NULL;
-
         if (wait) usleep (10000);
     }