Changeset 18839 for trunk/ppMerge
- Timestamp:
- Jul 31, 2008, 2:13:59 PM (18 years ago)
- Location:
- trunk/ppMerge/src
- Files:
-
- 1 added
- 7 edited
-
Makefile.am (modified) (1 diff)
-
ppMerge.c (modified) (1 diff)
-
ppMerge.h (modified) (3 diffs)
-
ppMergeArguments.c (modified) (2 diffs)
-
ppMergeLoop_Threaded.c (modified) (5 diffs)
-
ppMergeReadChunk.c (modified) (4 diffs)
-
ppMergeSetThreads.c (added)
-
ppMergeThreadLauncher.c (modified) (1 diff)
Legend:
- Unmodified
- Added
- Removed
-
trunk/ppMerge/src/Makefile.am
r18758 r18839 10 10 ppMergeFiles.c \ 11 11 ppMergeScaleZero.c \ 12 ppMergeLoop.c \13 12 ppMergeFileGroup.c \ 13 ppMergeReadChunk.c \ 14 ppMergeLoop_Threaded.c \ 15 ppMergeSetThreads.c \ 14 16 ppMergeMask.c 15 17 16 # ppMergeLoop_Threaded.c 17 # ppMergeThreadLauncher.c 18 18 # ppMergeLoop.c 19 19 20 20 noinst_HEADERS = \ -
trunk/ppMerge/src/ppMerge.c
r18757 r18839 10 10 { 11 11 psLibInit(NULL); 12 psMemSetThreadSafety(false);13 12 psTimerStart(TIMERNAME); 14 13 -
trunk/ppMerge/src/ppMerge.h
r18758 r18839 15 15 #define TIMERNAME "ppMerge" // Name for timer 16 16 #define PPMERGE_RECIPE "PPMERGE" // Recipe name 17 #define THREADED 017 #define THREADED 1 18 18 19 19 // Type of frame to merge … … 39 39 bool read; 40 40 bool busy; 41 int firstScan; 42 int lastScan; 41 43 } ppMergeFileGroup; 42 44 … … 100 102 101 103 ppMergeFileGroup *ppMergeFileGroupAlloc(); 102 ppMergeFileGroup *ppMergeReadChunk ( psArray *fileGroups, pmConfig *config, int numChunk);104 ppMergeFileGroup *ppMergeReadChunk (bool *status, psArray *fileGroups, pmConfig *config, int numChunk); 103 105 void *ppMergeThreadLauncher (void *data); 104 106 107 bool ppMergeSetThreads (); 108 105 109 #endif -
trunk/ppMerge/src/ppMergeArguments.c
r18758 r18839 171 171 } 172 172 173 # if (THREADED)174 173 // Number of threads 175 174 if ((argnum = psArgumentGet(argc, argv, "-threads"))) { … … 181 180 // create the thread pool with number of desired threads, supplying our thread launcher function 182 181 // XXX need to determine the number of threads from the config data 183 psThreadPoolInit (nThreads , &ppMergeThreadLauncher);184 } 185 # endif 182 psThreadPoolInit (nThreads); 183 } 184 ppMergeSetThreads(); 186 185 187 186 if (argc == 1 || !psArgumentParse(arguments, &argc, argv) || argc != 3) { -
trunk/ppMerge/src/ppMergeLoop_Threaded.c
r18757 r18839 36 36 37 37 // General combination parameters 38 int rows = psMetadataLookupS32(NULL, arguments, "ROWS"); // Number of rows to read per chunk39 38 int iter = psMetadataLookupS32(NULL, arguments, "ITER"); // Number of rejection iterations 40 39 float rej = psMetadataLookupF32(NULL, arguments, "REJ"); // Rejection level … … 175 174 fileGroup->read = false; 176 175 fileGroup->busy = false; 176 fileGroup->lastScan = 0; 177 fileGroup->firstScan = 0; 177 178 fileGroups->data[i] = fileGroup; 178 179 } 179 180 181 // call the init functions 182 switch (type) { 183 case PPMERGE_TYPE_BIAS: 184 case PPMERGE_TYPE_FLAT: 185 case PPMERGE_TYPE_FRINGE: 186 psAssert (fileGroups->n > 0, "no valid file groups defined"); 187 ppMergeFileGroup *fileGroup = fileGroups->data[0]; 188 if (!pmReadoutCombinePrepare(outRO, fileGroup->readouts, combination)) { 189 goto ERROR; 190 } 191 break; 192 193 default: 194 fprintf (stderr, "not yet ready"); 195 goto ERROR; 196 } 197 180 198 // Read input data by chunks 199 // psTimerStart ("ppMergeLoop"); 181 200 for (int numChunk = 0; true; numChunk++) { 182 201 202 bool status = false; 183 203 ppMergeFileGroup *fileGroup = ppMergeReadChunk (&status, fileGroups, config, numChunk); 184 204 if (!status) goto ERROR; 185 205 if (!fileGroup) break; 186 206 207 psThreadJob *job = NULL; 208 187 209 switch (type) { 188 210 case PPMERGE_TYPE_SHUTTER: 189 if (nThreads) { 190 // allocate a job 191 psThreadJob *job = psThreadJobAlloc ("PPMERGE_SHUTTER_CORRECTION", 0); 192 193 // construct the arguments for this job 194 psArrayAdd (job->args, 1, outRO); 195 psArrayAdd (job->args, 1, fileGroup); 196 psArrayAdd (job->args, 1, psScalarAlloc(shutterRef, PS_TYPE_F32)) 197 psArrayAdd (job->args, 1, shutters->data[cellNum]); 198 psArrayAdd (job->args, 1, psScalarAlloc(iter, PS_TYPE_S32)); 199 psArrayAdd (job->args, 1, psScalarAlloc(rej, PS_TYPE_F32)); 200 psArrayAdd (job->args, 1, psScalarAlloc(maskVal, PS_TYPE_U8)); 201 202 psThreadJobAddPending (job); 203 } else { 204 if (!pmShutterCorrectionGenerate(outRO, NULL, fileGroup->readouts, shutterRef, shutters->data[cellNum], iter, rej, maskVal)) { 205 goto ERROR; 206 } 207 fileGroup->busy = false; 211 // allocate a job 212 job = psThreadJobAlloc ("PPMERGE_SHUTTER_CORRECTION"); 213 214 // construct the arguments for this job 215 psArrayAdd (job->args, 1, outRO); 216 psArrayAdd (job->args, 1, fileGroup); 217 psArrayAdd (job->args, 1, psScalarAlloc(shutterRef, PS_TYPE_F32)); 218 psArrayAdd (job->args, 1, shutters->data[cellNum]); 219 psArrayAdd (job->args, 1, psScalarAlloc(iter, PS_TYPE_S32)); 220 psArrayAdd (job->args, 1, psScalarAlloc(rej, PS_TYPE_F32)); 221 psArrayAdd (job->args, 1, psScalarAlloc(maskVal, PS_TYPE_U8)); 222 223 // call: pmShutterCorrectionGenerate(outRO, NULL, fileGroup->readouts, shutterRef, shutters->data[cellNum], iter, rej, maskVal) 224 if (!psThreadJobAddPending (job)) { 225 goto ERROR; 208 226 } 209 227 break; 210 228 case PPMERGE_TYPE_DARK: 211 if (nThreads) { 212 // allocate a job 213 psThreadJob *job = psThreadJobAlloc ("PPMERGE_DARK_COMBINE", 0); 214 215 // construct the arguments for this job 216 psArrayAdd (job->args, 1, outCell); 217 psArrayAdd (job->args, 1, fileGroup); 218 psArrayAdd (job->args, 1, darkOrdinates); 219 psArrayAdd (job->args, 1, darkNorm); 220 psArrayAdd (job->args, 1, psScalarAlloc(iter, PS_TYPE_S32)); 221 psArrayAdd (job->args, 1, psScalarAlloc(rej, PS_TYPE_F32)); 222 psArrayAdd (job->args, 1, psScalarAlloc(maskVal, PS_TYPE_U8)); 223 224 psThreadJobAddPending (job); 225 } else { 226 if (!pmDarkCombine(outCell, fileGroup->readouts, darkOrdinates, darkNorm, iter, rej, maskVal)) { 227 goto ERROR; 228 } 229 fileGroup->busy = false; 229 // allocate a job 230 job = psThreadJobAlloc ("PPMERGE_DARK_COMBINE"); 231 232 // construct the arguments for this job 233 psArrayAdd (job->args, 1, outCell); 234 psArrayAdd (job->args, 1, fileGroup); 235 psArrayAdd (job->args, 1, darkOrdinates); 236 psArrayAdd (job->args, 1, darkNorm); 237 psArrayAdd (job->args, 1, psScalarAlloc(iter, PS_TYPE_S32)); 238 psArrayAdd (job->args, 1, psScalarAlloc(rej, PS_TYPE_F32)); 239 psArrayAdd (job->args, 1, psScalarAlloc(maskVal, PS_TYPE_U8)); 240 241 // call: pmDarkCombine(outCell, fileGroup->readouts, darkOrdinates, darkNorm, iter, rej, maskVal); 242 if (!psThreadJobAddPending (job)) { 243 goto ERROR; 230 244 } 231 245 break; … … 233 247 case PPMERGE_TYPE_FLAT: 234 248 case PPMERGE_TYPE_FRINGE: 235 if (nThreads) { 236 // allocate a job 237 psThreadJob *job = psThreadJobAlloc ("PPMERGE_READOUT_COMBINE", 0); 238 239 // construct the arguments for this job 240 psArrayAdd (job->args, 1, outRO); 241 psArrayAdd (job->args, 1, fileGroup); 242 psArrayAdd (job->args, 1, zeros); 243 psArrayAdd (job->args, 1, scales); 244 psArrayAdd (job->args, 1, combination); 245 246 psThreadJobAddPending (job); 247 } else { 248 if (!pmReadoutCombine(outRO, fileGroup->readouts, zeros, scales, combination)) { 249 goto ERROR; 250 } 251 fileGroup->busy = false; 249 // allocate a job 250 job = psThreadJobAlloc ("PPMERGE_READOUT_COMBINE"); 251 252 // construct the arguments for this job 253 psArrayAdd (job->args, 1, outRO); 254 psArrayAdd (job->args, 1, fileGroup); 255 psArrayAdd (job->args, 1, zeros); 256 psArrayAdd (job->args, 1, scales); 257 psArrayAdd (job->args, 1, combination); 258 259 // call: pmReadoutCombine(outRO, fileGroup->readouts, zeros, scales, combination); 260 if (!psThreadJobAddPending (job)) { 261 goto ERROR; 252 262 } 253 263 break; … … 258 268 259 269 // wait for the threads to finish and manage results 260 if (nThreads) { 261 // wait here for the threaded jobs to finish 262 if (!psThreadPoolWait ()) { 263 psError(PS_ERR_UNKNOWN, false, "Unable to interpolate image."); 264 return false; 265 } 266 fprintf (stderr, "success for threaded jobs\n"); 267 268 // we don't care about the results, just dump the done queue jobs 269 psThreadJob *job = NULL; 270 while ((job = psThreadJobGetDone()) != NULL) { 271 psFree (job); 272 } 270 if (!psThreadPoolWait ()) { 271 psError(PS_ERR_UNKNOWN, false, "Unable to combine images."); 272 return false; 273 273 } 274 275 // we don't care about the results, just dump the done queue jobs 276 psThreadJob *job = NULL; 277 while ((job = psThreadJobGetDone()) != NULL) { 278 psFree (job); 279 } 280 281 psFree(fileGroups); 274 282 275 283 // Get list of cells for concepts averaging 276 284 psList *inCells = psListAlloc(NULL); // List of cells 277 285 for (int i = 0; i < numFiles; i++) { 278 pmReadout *readout = readouts->data[i]; // Readout of interest 279 psListAdd(inCells, PS_LIST_TAIL, readout->parent); 286 pmFPAfile *input = pmFPAfileSelectSingle(config->files, "PPMERGE.INPUT", i); 287 pmCell *inCell = pmFPAviewThisCell(view, input->fpa); // Input cell 288 psListAdd(inCells, PS_LIST_TAIL, inCell); 280 289 } 281 290 if (!pmConceptsAverageCells(outCell, inCells, NULL, NULL, true)) { … … 286 295 } 287 296 psFree(inCells); 288 289 psFree(fileGroups); 297 // fprintf (stdout, "done ppMergeLoop for cell : %f\n", psTimerMark ("ppMergeLoop")); 290 298 291 299 // Plug supplementary images into their own FPAs -
trunk/ppMerge/src/ppMergeReadChunk.c
r18757 r18839 1 1 # include "ppMerge.h" 2 2 3 ppMergeFileGroup *ppMergeReadChunk (psArray *fileGroups, pmConfig *config, int numChunk) { 3 ppMergeFileGroup *ppMergeReadChunk (bool *status, psArray *fileGroups, pmConfig *config, int numChunk) { 4 5 *status = true; 6 7 bool mdok; 8 bool haveMasks = psMetadataLookupBool(&mdok, config->arguments, "INPUTS.MASKS"); // Do we have masks? 9 bool haveWeights = psMetadataLookupBool(&mdok, config->arguments, "INPUTS.WEIGHTS"); // Do we have weights? 10 int rows = psMetadataLookupS32(NULL, config->arguments, "ROWS"); // Number of rows to read per chunk 4 11 5 12 // select an available fileGroup 6 7 bool haveMasks = psMetadataLookupBool(&mdok, arguments, "INPUTS.MASKS"); // Do we have masks?8 bool haveWeights = psMetadataLookupBool(&mdok, arguments, "INPUTS.WEIGHTS"); // Do we have weights?9 10 13 while (1) { 11 14 // check for any fileGroups which can read data … … 14 17 if (fileGroup->read) continue; 15 18 19 // find max last scan so far 20 int lastScan = 0; 21 for (int i = 0; i < fileGroups->n; i++) { 22 ppMergeFileGroup *fileGroup = fileGroups->data[i]; 23 lastScan = PS_MAX (fileGroup->lastScan, lastScan); 24 } 25 fileGroup->firstScan = lastScan; 26 fileGroup->lastScan = lastScan + rows; 27 16 28 psArray *readouts = fileGroup->readouts; 29 30 psTimerStart ("ppMergeReadChunk"); 17 31 18 32 psTrace("ppStack", 2, "Reading data for chunk %d into fileGroup %d....n", numChunk, j); … … 20 34 pmReadout *inRO = readouts->data[i]; // Input readout 21 35 36 // override the recorded last scan 37 inRO->thisImageScan = fileGroup->firstScan; 38 inRO->thisWeightScan = fileGroup->firstScan; 39 inRO->thisMaskScan = fileGroup->firstScan; 40 22 41 // Read a chunk from a file 23 42 pmFPAfile *file = pmFPAfileSelectSingle(config->files, "PPMERGE.INPUT", i); 24 if (!pmReadoutReadChunk(inRO, file->fits, 0, rows, 0, config)) {25 psError(PS_ERR_IO, false, "Unable to read chunk %d for file PPMERGE.INPUT %d", numChunk, i);26 return NULL;27 }28 43 29 if (haveMasks) { 30 pmFPAfile *file = pmFPAfileSelectSingle(config->files, "PPMERGE.INPUT.MASK", i); 31 if (!pmReadoutReadChunkMask(inRO, file->fits, 0, rows, 0, config)) { 32 psError(PS_ERR_IO, false, "Unable to read chunk %d for file PPMERGE.INPUT.MASK %d", numChunk, NAME, i); 44 bool keepReading = false; 45 if (pmReadoutMore(inRO, file->fits, 0, rows, config)) { 46 keepReading = true; 47 if (!pmReadoutReadChunk(inRO, file->fits, 0, rows, 0, config)) { 48 psError(PS_ERR_IO, false, "Unable to read chunk %d for file PPMERGE.INPUT %d", numChunk, i); 49 *status = false; 33 50 return NULL; 34 51 } 35 52 } 36 53 37 if (haveWeights) { 38 pmFPAfile *file = pmFPAfileSelectSingle(config->files, "PPMERGE.INPUT.WEIGHT", i); 39 if (!pmReadoutReadChunkWeight(inRO, file->fits, 0, rows, 0, config)) { 40 psError(PS_ERR_IO, false, "Unable to read chunk %d for file PPMERGE.INPUT.WEIGHT %d", numChunk, NAME, i); 54 if (haveMasks && pmReadoutMoreMask(inRO, file->fits, 0, rows, config)) { 55 keepReading = true; 56 pmFPAfile *file = pmFPAfileSelectSingle(config->files, "PPMERGE.INPUT.MASK", i); 57 if (!pmReadoutReadChunkMask(inRO, file->fits, 0, rows, 0, config)) { 58 psError(PS_ERR_IO, false, "Unable to read chunk %d for file PPMERGE.INPUT.MASK %d", numChunk, i); 59 *status = false; 41 60 return NULL; 42 61 } 43 62 } 63 64 if (haveWeights && pmReadoutMoreWeight(inRO, file->fits, 0, rows, config)) { 65 keepReading = true; 66 pmFPAfile *file = pmFPAfileSelectSingle(config->files, "PPMERGE.INPUT.WEIGHT", i); 67 if (!pmReadoutReadChunkWeight(inRO, file->fits, 0, rows, 0, config)) { 68 psError(PS_ERR_IO, false, "Unable to read chunk %d for file PPMERGE.INPUT.WEIGHT %d", numChunk, i); 69 *status = false; 70 return NULL; 71 } 72 } 73 if (!keepReading) { 74 return NULL; 75 } 44 76 } 77 45 78 fileGroup->read = fileGroup->busy = true; 46 79 return fileGroup; … … 48 81 49 82 // check for any fileGroups which are done processing 50 bool wait = true; 51 bool more = true; 83 bool wait = false; 52 84 for (int j = 0; j < fileGroups->n; j++) { 53 85 ppMergeFileGroup *fileGroup = fileGroups->data[j]; 54 86 if (!fileGroup->read || fileGroup->busy) continue; 55 56 wait = false;57 psArray *readouts = fileGroup->readouts;58 // any more data to be read?59 for (int i = 0; i < readouts->n && more; i++) {60 pmReadout *inRO = readouts->data[i];61 62 pmFPAfile *file = pmFPAfileSelectSingle(config->files, "PPMERGE.INPUT", i);63 more &= pmReadoutMore(inRO, file->fits, 0, rows, config);64 65 if (haveMasks) {66 pmFPAfile *file = pmFPAfileSelectSingle(config->files, "PPMERGE.INPUT.MASK", i);67 more &= pmReadoutMoreMask(inRO, file->fits, 0, rows, config);68 }69 if (haveWeights) {70 pmFPAfile *file = pmFPAfileSelectSingle(config->files, "PPMERGE.INPUT.WEIGHT", i);71 more &= pmReadoutMoreWeight(inRO, file->fits, 0, rows, config);72 }73 }74 87 fileGroup->read = false; 88 wait = true; 75 89 } 76 if (!more) return NULL;77 78 90 if (wait) usleep (10000); 79 91 } -
trunk/ppMerge/src/ppMergeThreadLauncher.c
r18757 r18839 41 41 self->fault = true; 42 42 } 43 43 44 // we do not have to lock here because this transition is not tied to the job queue 44 45 fileGroup->busy = false;
Note:
See TracChangeset
for help on using the changeset viewer.
