Index: trunk/ppMerge/src/Makefile.am =================================================================== --- trunk/ppMerge/src/Makefile.am (revision 18758) +++ trunk/ppMerge/src/Makefile.am (revision 18839) @@ -10,11 +10,11 @@ ppMergeFiles.c \ ppMergeScaleZero.c \ - ppMergeLoop.c \ ppMergeFileGroup.c \ + ppMergeReadChunk.c \ + ppMergeLoop_Threaded.c \ + ppMergeSetThreads.c \ ppMergeMask.c -# ppMergeLoop_Threaded.c -# ppMergeThreadLauncher.c - +# ppMergeLoop.c noinst_HEADERS = \ Index: trunk/ppMerge/src/ppMerge.c =================================================================== --- trunk/ppMerge/src/ppMerge.c (revision 18758) +++ trunk/ppMerge/src/ppMerge.c (revision 18839) @@ -10,5 +10,4 @@ { psLibInit(NULL); - psMemSetThreadSafety(false); psTimerStart(TIMERNAME); Index: trunk/ppMerge/src/ppMerge.h =================================================================== --- trunk/ppMerge/src/ppMerge.h (revision 18758) +++ trunk/ppMerge/src/ppMerge.h (revision 18839) @@ -15,5 +15,5 @@ #define TIMERNAME "ppMerge" // Name for timer #define PPMERGE_RECIPE "PPMERGE" // Recipe name -#define THREADED 0 +#define THREADED 1 // Type of frame to merge @@ -39,4 +39,6 @@ bool read; bool busy; + int firstScan; + int lastScan; } ppMergeFileGroup; @@ -100,6 +102,8 @@ ppMergeFileGroup *ppMergeFileGroupAlloc(); -ppMergeFileGroup *ppMergeReadChunk (psArray *fileGroups, pmConfig *config, int numChunk); +ppMergeFileGroup *ppMergeReadChunk (bool *status, psArray *fileGroups, pmConfig *config, int numChunk); void *ppMergeThreadLauncher (void *data); +bool ppMergeSetThreads (); + #endif Index: trunk/ppMerge/src/ppMergeArguments.c =================================================================== --- trunk/ppMerge/src/ppMergeArguments.c (revision 18758) +++ trunk/ppMerge/src/ppMergeArguments.c (revision 18839) @@ -171,5 +171,4 @@ } -# if (THREADED) // Number of threads if ((argnum = psArgumentGet(argc, argv, "-threads"))) { @@ -181,7 +180,7 @@ // create the thread pool with number of desired threads, supplying our thread launcher function // XXX need to determine the number of threads from the config data - psThreadPoolInit (nThreads, &ppMergeThreadLauncher); - } -# endif + psThreadPoolInit (nThreads); + } + ppMergeSetThreads(); if (argc == 1 || !psArgumentParse(arguments, &argc, argv) || argc != 3) { Index: trunk/ppMerge/src/ppMergeLoop_Threaded.c =================================================================== --- trunk/ppMerge/src/ppMergeLoop_Threaded.c (revision 18758) +++ trunk/ppMerge/src/ppMergeLoop_Threaded.c (revision 18839) @@ -36,5 +36,4 @@ // General combination parameters - int rows = psMetadataLookupS32(NULL, arguments, "ROWS"); // Number of rows to read per chunk int iter = psMetadataLookupS32(NULL, arguments, "ITER"); // Number of rejection iterations float rej = psMetadataLookupF32(NULL, arguments, "REJ"); // Rejection level @@ -175,57 +174,72 @@ fileGroup->read = false; fileGroup->busy = false; + fileGroup->lastScan = 0; + fileGroup->firstScan = 0; fileGroups->data[i] = fileGroup; } + // call the init functions + switch (type) { + case PPMERGE_TYPE_BIAS: + case PPMERGE_TYPE_FLAT: + case PPMERGE_TYPE_FRINGE: + psAssert (fileGroups->n > 0, "no valid file groups defined"); + ppMergeFileGroup *fileGroup = fileGroups->data[0]; + if (!pmReadoutCombinePrepare(outRO, fileGroup->readouts, combination)) { + goto ERROR; + } + break; + + default: + fprintf (stderr, "not yet ready"); + goto ERROR; + } + // Read input data by chunks + // psTimerStart ("ppMergeLoop"); for (int numChunk = 0; true; numChunk++) { - + + bool status = false; ppMergeFileGroup *fileGroup = ppMergeReadChunk (&status, fileGroups, config, numChunk); if (!status) goto ERROR; if (!fileGroup) break; + psThreadJob *job = NULL; + switch (type) { case PPMERGE_TYPE_SHUTTER: - if (nThreads) { - // allocate a job - psThreadJob *job = psThreadJobAlloc ("PPMERGE_SHUTTER_CORRECTION", 0); - - // construct the arguments for this job - psArrayAdd (job->args, 1, outRO); - psArrayAdd (job->args, 1, fileGroup); - psArrayAdd (job->args, 1, psScalarAlloc(shutterRef, PS_TYPE_F32)) - psArrayAdd (job->args, 1, shutters->data[cellNum]); - psArrayAdd (job->args, 1, psScalarAlloc(iter, PS_TYPE_S32)); - psArrayAdd (job->args, 1, psScalarAlloc(rej, PS_TYPE_F32)); - psArrayAdd (job->args, 1, psScalarAlloc(maskVal, PS_TYPE_U8)); - - psThreadJobAddPending (job); - } else { - if (!pmShutterCorrectionGenerate(outRO, NULL, fileGroup->readouts, shutterRef, shutters->data[cellNum], iter, rej, maskVal)) { - goto ERROR; - } - fileGroup->busy = false; + // allocate a job + job = psThreadJobAlloc ("PPMERGE_SHUTTER_CORRECTION"); + + // construct the arguments for this job + psArrayAdd (job->args, 1, outRO); + psArrayAdd (job->args, 1, fileGroup); + psArrayAdd (job->args, 1, psScalarAlloc(shutterRef, PS_TYPE_F32)); + psArrayAdd (job->args, 1, shutters->data[cellNum]); + psArrayAdd (job->args, 1, psScalarAlloc(iter, PS_TYPE_S32)); + psArrayAdd (job->args, 1, psScalarAlloc(rej, PS_TYPE_F32)); + psArrayAdd (job->args, 1, psScalarAlloc(maskVal, PS_TYPE_U8)); + + // call: pmShutterCorrectionGenerate(outRO, NULL, fileGroup->readouts, shutterRef, shutters->data[cellNum], iter, rej, maskVal) + if (!psThreadJobAddPending (job)) { + goto ERROR; } break; case PPMERGE_TYPE_DARK: - if (nThreads) { - // allocate a job - psThreadJob *job = psThreadJobAlloc ("PPMERGE_DARK_COMBINE", 0); - - // construct the arguments for this job - psArrayAdd (job->args, 1, outCell); - psArrayAdd (job->args, 1, fileGroup); - psArrayAdd (job->args, 1, darkOrdinates); - psArrayAdd (job->args, 1, darkNorm); - psArrayAdd (job->args, 1, psScalarAlloc(iter, PS_TYPE_S32)); - psArrayAdd (job->args, 1, psScalarAlloc(rej, PS_TYPE_F32)); - psArrayAdd (job->args, 1, psScalarAlloc(maskVal, PS_TYPE_U8)); - - psThreadJobAddPending (job); - } else { - if (!pmDarkCombine(outCell, fileGroup->readouts, darkOrdinates, darkNorm, iter, rej, maskVal)) { - goto ERROR; - } - fileGroup->busy = false; + // allocate a job + job = psThreadJobAlloc ("PPMERGE_DARK_COMBINE"); + + // construct the arguments for this job + psArrayAdd (job->args, 1, outCell); + psArrayAdd (job->args, 1, fileGroup); + psArrayAdd (job->args, 1, darkOrdinates); + psArrayAdd (job->args, 1, darkNorm); + psArrayAdd (job->args, 1, psScalarAlloc(iter, PS_TYPE_S32)); + psArrayAdd (job->args, 1, psScalarAlloc(rej, PS_TYPE_F32)); + psArrayAdd (job->args, 1, psScalarAlloc(maskVal, PS_TYPE_U8)); + + // call: pmDarkCombine(outCell, fileGroup->readouts, darkOrdinates, darkNorm, iter, rej, maskVal); + if (!psThreadJobAddPending (job)) { + goto ERROR; } break; @@ -233,21 +247,17 @@ case PPMERGE_TYPE_FLAT: case PPMERGE_TYPE_FRINGE: - if (nThreads) { - // allocate a job - psThreadJob *job = psThreadJobAlloc ("PPMERGE_READOUT_COMBINE", 0); - - // construct the arguments for this job - psArrayAdd (job->args, 1, outRO); - psArrayAdd (job->args, 1, fileGroup); - psArrayAdd (job->args, 1, zeros); - psArrayAdd (job->args, 1, scales); - psArrayAdd (job->args, 1, combination); - - psThreadJobAddPending (job); - } else { - if (!pmReadoutCombine(outRO, fileGroup->readouts, zeros, scales, combination)) { - goto ERROR; - } - fileGroup->busy = false; + // allocate a job + job = psThreadJobAlloc ("PPMERGE_READOUT_COMBINE"); + + // construct the arguments for this job + psArrayAdd (job->args, 1, outRO); + psArrayAdd (job->args, 1, fileGroup); + psArrayAdd (job->args, 1, zeros); + psArrayAdd (job->args, 1, scales); + psArrayAdd (job->args, 1, combination); + + // call: pmReadoutCombine(outRO, fileGroup->readouts, zeros, scales, combination); + if (!psThreadJobAddPending (job)) { + goto ERROR; } break; @@ -258,24 +268,23 @@ // wait for the threads to finish and manage results - if (nThreads) { - // wait here for the threaded jobs to finish - if (!psThreadPoolWait ()) { - psError(PS_ERR_UNKNOWN, false, "Unable to interpolate image."); - return false; - } - fprintf (stderr, "success for threaded jobs\n"); - - // we don't care about the results, just dump the done queue jobs - psThreadJob *job = NULL; - while ((job = psThreadJobGetDone()) != NULL) { - psFree (job); - } + if (!psThreadPoolWait ()) { + psError(PS_ERR_UNKNOWN, false, "Unable to combine images."); + return false; } + + // we don't care about the results, just dump the done queue jobs + psThreadJob *job = NULL; + while ((job = psThreadJobGetDone()) != NULL) { + psFree (job); + } + + psFree(fileGroups); // Get list of cells for concepts averaging psList *inCells = psListAlloc(NULL); // List of cells for (int i = 0; i < numFiles; i++) { - pmReadout *readout = readouts->data[i]; // Readout of interest - psListAdd(inCells, PS_LIST_TAIL, readout->parent); + pmFPAfile *input = pmFPAfileSelectSingle(config->files, "PPMERGE.INPUT", i); + pmCell *inCell = pmFPAviewThisCell(view, input->fpa); // Input cell + psListAdd(inCells, PS_LIST_TAIL, inCell); } if (!pmConceptsAverageCells(outCell, inCells, NULL, NULL, true)) { @@ -286,6 +295,5 @@ } psFree(inCells); - - psFree(fileGroups); + // fprintf (stdout, "done ppMergeLoop for cell : %f\n", psTimerMark ("ppMergeLoop")); // Plug supplementary images into their own FPAs Index: trunk/ppMerge/src/ppMergeReadChunk.c =================================================================== --- trunk/ppMerge/src/ppMergeReadChunk.c (revision 18758) +++ trunk/ppMerge/src/ppMergeReadChunk.c (revision 18839) @@ -1,11 +1,14 @@ # 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,5 +17,16 @@ 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,27 +34,46 @@ 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,32 +81,11 @@ // 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); } Index: trunk/ppMerge/src/ppMergeSetThreads.c =================================================================== --- trunk/ppMerge/src/ppMergeSetThreads.c (revision 18839) +++ trunk/ppMerge/src/ppMergeSetThreads.c (revision 18839) @@ -0,0 +1,70 @@ +# include "ppMerge.h" + +// "PPMERGE_READOUT_COMBINE", 5 +bool ppMergeThread_pmReadoutCombine (psThreadJob *job) { + + pmReadout *output = job->args->data[0]; + ppMergeFileGroup *fileGroup = job->args->data[1]; + psVector *zero = job->args->data[2]; + psVector *scale = job->args->data[3]; + pmCombineParams *params = job->args->data[4]; + + bool status = pmReadoutCombine (output, fileGroup->readouts, zero, scale, params); + + // after we are done, tell the I/O system that this file group is done + fileGroup->busy = false; + return status; +} + +bool ppMergeThread_pmDarkCombine (psThreadJob *job) { + + pmCell *outCell = job->args->data[0]; + ppMergeFileGroup *fileGroup = job->args->data[1]; + psArray *darkOrdinates = job->args->data[2]; + psString darkNorm = job->args->data[3]; + psScalar *iter = job->args->data[4]; + psScalar *rej = job->args->data[5]; + psScalar *maskVal = job->args->data[6]; + + bool status = pmDarkCombine(outCell, fileGroup->readouts, darkOrdinates, darkNorm, iter->data.S32, rej->data.F32, maskVal->data.U8); + + // after we are done, tell the I/O system that this file group is done + fileGroup->busy = false; + return status; +} + +bool ppMergeThread_pmShuttderCorrectionGenerate (psThreadJob *job) { + + pmReadout *output = job->args->data[0]; + ppMergeFileGroup *fileGroup = job->args->data[1]; + psScalar *shutterRef = job->args->data[2]; + pmShutterCorrectionData *data = job->args->data[3]; + psScalar *iter = job->args->data[4]; + psScalar *rej = job->args->data[5]; + psScalar *maskVal = job->args->data[6]; + + bool status = pmShutterCorrectionGenerate(output, NULL, fileGroup->readouts, shutterRef->data.F32, data, iter->data.S32, rej->data.F32, maskVal->data.U8); + + // after we are done, tell the I/O system that this file group is done + fileGroup->busy = false; + return status; +} + +bool ppMergeSetThreads () { + + psThreadTask *task = NULL; + + task = psThreadTaskAlloc ("PPMERGE_READOUT_COMBINE", 5); + task->function = &ppMergeThread_pmReadoutCombine; + psThreadTaskAdd (task); + + task = psThreadTaskAlloc ("PPMERGE_DARK_COMBINE", 7); + task->function = &ppMergeThread_pmReadoutCombine; + psThreadTaskAdd (task); + + task = psThreadTaskAlloc ("PPMERGE_SHUTTER_CORRECTION", 7); + task->function = &ppMergeThread_pmReadoutCombine; + psThreadTaskAdd (task); + + return true; +} Index: trunk/ppMerge/src/ppMergeThreadLauncher.c =================================================================== --- trunk/ppMerge/src/ppMergeThreadLauncher.c (revision 18758) +++ trunk/ppMerge/src/ppMergeThreadLauncher.c (revision 18839) @@ -41,4 +41,5 @@ self->fault = true; } + // we do not have to lock here because this transition is not tied to the job queue fileGroup->busy = false;