Index: trunk/ppMerge/src/Makefile.am =================================================================== --- trunk/ppMerge/src/Makefile.am (revision 18756) +++ trunk/ppMerge/src/Makefile.am (revision 18757) @@ -11,4 +11,6 @@ ppMergeScaleZero.c \ ppMergeLoop.c \ + ppMergeThreadLauncher.c \ + ppMergeFileGroup.c \ ppMergeMask.c Index: trunk/ppMerge/src/ppMerge.c =================================================================== --- trunk/ppMerge/src/ppMerge.c (revision 18756) +++ trunk/ppMerge/src/ppMerge.c (revision 18757) @@ -1,14 +1,3 @@ -#ifdef HAVE_CONFIG_H -#include -#endif - -#include -#include -#include -#include - #include "ppMerge.h" - -//#include "ppMem.h" // Yet to do: Index: trunk/ppMerge/src/ppMerge.h =================================================================== --- trunk/ppMerge/src/ppMerge.h (revision 18756) +++ trunk/ppMerge/src/ppMerge.h (revision 18757) @@ -1,4 +1,15 @@ #ifndef PP_MERGE_H #define PP_MERGE_H + +#ifdef HAVE_CONFIG_H +#include +#endif + +#include +#include +#include +#include +#include +#include #define TIMERNAME "ppMerge" // Name for timer @@ -22,4 +33,10 @@ PPMERGE_FILES_OUTPUT // Output files } ppMergeFiles; + +typedef struct { + psArray *readouts; + bool read; + bool busy; +} ppMergeFileGroup; // Parse command-line arguments and recipe @@ -80,3 +97,8 @@ ); + +ppMergeFileGroup *ppMergeFileGroupAlloc(); +ppMergeFileGroup *ppMergeReadChunk (psArray *fileGroups, pmConfig *config, int numChunk); +void *ppMergeThreadLauncher (void *data); + #endif Index: trunk/ppMerge/src/ppMergeArguments.c =================================================================== --- trunk/ppMerge/src/ppMergeArguments.c (revision 18756) +++ trunk/ppMerge/src/ppMergeArguments.c (revision 18757) @@ -1,12 +1,2 @@ -#ifdef HAVE_CONFIG_H -#include -#endif - -#include -#include -#include -#include -#include - #include "ppMerge.h" @@ -179,4 +169,16 @@ psMetadataAddStr (config->arguments, PS_LIST_TAIL, "CELL_SELECTIONS", PS_DATA_STRING, "", argv[argnum]); psArgumentRemove (argnum, &argc, argv); + } + + // Number of threads + if ((argnum = psArgumentGet(argc, argv, "-threads"))) { + psArgumentRemove(argnum, &argc, argv); + int nThreads = atoi(argv[argnum]); + psMetadataAddS32(config->arguments, PS_LIST_TAIL, "NTHREADS", 0, "number of warp threads", nThreads); + psArgumentRemove(argnum, &argc, argv); + + // 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); } Index: trunk/ppMerge/src/ppMergeFileGroup.c =================================================================== --- trunk/ppMerge/src/ppMergeFileGroup.c (revision 18757) +++ trunk/ppMerge/src/ppMergeFileGroup.c (revision 18757) @@ -0,0 +1,20 @@ +# include "ppMerge.h" + +void ppMergeFileGroupFree (ppMergeFileGroup *group) { + + if (!group) return; + + psFree (group->readouts); + return; +} + +ppMergeFileGroup *ppMergeFileGroupAlloc() { + + ppMergeFileGroup *group = (ppMergeFileGroup *) psAlloc(sizeof(ppMergeFileGroup)); + psMemSetDeallocator(group, (psFreeFunc) ppMergeFileGroupFree); + + group->readouts = NULL; + group->read = false; + group->busy = false; + return group; +} Index: trunk/ppMerge/src/ppMergeLoop.c =================================================================== --- trunk/ppMerge/src/ppMergeLoop.c (revision 18756) +++ trunk/ppMerge/src/ppMergeLoop.c (revision 18757) @@ -1,13 +1,2 @@ -#ifdef HAVE_CONFIG_H -#include -#endif - -#include -#include - -#include -#include -#include - #include "ppMerge.h" Index: trunk/ppMerge/src/ppMergeLoop_Threaded.c =================================================================== --- trunk/ppMerge/src/ppMergeLoop_Threaded.c (revision 18757) +++ trunk/ppMerge/src/ppMergeLoop_Threaded.c (revision 18757) @@ -0,0 +1,403 @@ +#ifdef HAVE_CONFIG_H +#include +#endif + +#include +#include + +#include +#include +#include + +#include "ppMerge.h" + +bool ppMergeLoop(pmConfig *config) +{ + assert(config); + + psMetadata *arguments = config->arguments; // Arguments + ppMergeType type = psMetadataLookupS32(NULL, config->arguments, "TYPE"); // Type of frame + int numFiles = psMetadataLookupS32(NULL, arguments, "INPUTS.NUM"); // Number of input files + bool mdok; // Status of MD lookup + bool haveMasks = psMetadataLookupBool(&mdok, arguments, "INPUTS.MASKS"); // Do we have masks? + bool haveWeights = psMetadataLookupBool(&mdok, arguments, "INPUTS.WEIGHTS"); // Do we have weights? + + psArray *inputs = ppMergeFileDataLevel(config, "PPMERGE.INPUT", PM_FPA_LEVEL_READOUT); // Input images + psArray *masks = NULL, *weights = NULL; // Input masks and weights + if (haveMasks) { + masks = ppMergeFileDataLevel(config, "PPMERGE.INPUT.MASK", PM_FPA_LEVEL_READOUT); + } + if (haveWeights) { + weights = ppMergeFileDataLevel(config, "PPMERGE.INPUT.WEIGHT", PM_FPA_LEVEL_READOUT); + } + + int nThreads = psMetadataLookupS32 (&mdok, arguments, "NTHREADS"); + if (!mdok) nThreads = 0; + + // 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 + float fraclow = psMetadataLookupF32(NULL, arguments, "FRACLOW"); // Reject fraction of low pixels + float frachigh = psMetadataLookupF32(NULL, arguments, "FRACHIGH"); // Reject fraction of hi pixels + int nKeep = psMetadataLookupS32(NULL, arguments, "NKEEP"); // Minimum number of values to keep + psStatsOptions combineStat = psMetadataLookupS32(NULL, arguments, "COMBINE"); // Combination statistic + bool useWeights = psMetadataLookupBool(NULL, arguments, "WEIGHTS"); // Use weights? + + // Fringe parameters + int fringeNum = psMetadataLookupS32(NULL, arguments, "FRINGE.NUM"); // Number of fringe points + int fringeSize = psMetadataLookupS32(NULL, arguments, "FRINGE.SIZE"); // Size of fringe regions + int fringeSmoothX = psMetadataLookupS32(NULL, arguments, "FRINGE.XSMOOTH"); // Smoothing regions in x + int fringeSmoothY = psMetadataLookupS32(NULL, arguments, "FRINGE.YSMOOTH"); // Smoothing regions in y + + // set the mask and mark bit values based on the named masks + psMaskType maskVal; + psMaskType markVal; + if (!pmConfigMaskSetBits (&maskVal, &markVal, config)) { + psError (PS_ERR_UNKNOWN, true, "Unable to define the mask bit values"); + return false; + } + + pmCombineParams *combination = pmCombineParamsAlloc(combineStat); // Combination parameters + combination->maskVal = maskVal; + combination->blank = pmConfigMaskGet("BLANK", config); + combination->nKeep = nKeep; + combination->fracHigh = frachigh; + combination->fracLow = fraclow; + combination->iter = iter; + combination->rej = rej; + combination->weights = useWeights; + + psMetadata *stats = NULL; // Statistics for output + if (psMetadataLookup(config->arguments, "STATS.NAME")) { + stats = psMetadataAlloc(); + psMetadataAddMetadata(config->arguments, PS_LIST_TAIL, "STATS.DATA", 0, "Statistics output", stats); + } + + pmFPAview *view = pmFPAviewAlloc(0); // View to component of interest + + // Retrieve data placed on analysis + psVector *scales = NULL, *zeros = NULL; // Scale and zeroes for combination + psArray *shutters = NULL; // Shutter correction data + switch (type) { + case PPMERGE_TYPE_FRINGE: + zeros = psMetadataLookupPtr(NULL, arguments, "ZEROS"); + if (!zeros) { + psError(PS_ERR_UNEXPECTED_NULL, false, "Unable to find ZEROS"); + goto ERROR; + } + // Flow through + case PPMERGE_TYPE_FLAT: + scales = psMetadataLookupPtr(NULL, arguments, "SCALES"); + if (!scales) { + psError(PS_ERR_UNEXPECTED_NULL, false, "Unable to find SCALES"); + goto ERROR; + } + break; + case PPMERGE_TYPE_SHUTTER: + shutters = psMetadataLookupPtr(NULL, arguments, "SHUTTER"); + if (!shutters) { + psError(PS_ERR_UNEXPECTED_NULL, false, "Unable to find SHUTTER"); + goto ERROR; + } + break; + case PPMERGE_TYPE_MASK: + case PPMERGE_TYPE_BIAS: + case PPMERGE_TYPE_DARK: + break; + default: + psAbort("Should never get here."); + } + + // Dark parameters + psArray *darkOrdinates = psMetadataLookupPtr(NULL, arguments, "DARK.ORDINATES"); // Dark info + psString darkNorm = psMetadataLookupStr(&mdok, arguments, "DARK.NORM"); // Dark normalisation + + + if (!ppMergeFileActivate(config, PPMERGE_FILES_ALL, true)) { + psError(PS_ERR_UNKNOWN, false, "Unable to activate files."); + goto ERROR; + } + psString outName = ppMergeOutputFile(config); // Name of output file + pmFPAfile *output = psMetadataLookupPtr(NULL, config->files, outName); // Output file + psFree(outName); + assert(output && output->fpa); + pmFPA *outFPA = output->fpa; // Output FPA + int cellNum = 0; // Index of cell + if (!pmFPAfileIOChecks(config, view, PM_FPA_BEFORE)) { + goto ERROR; + } + pmChip *outChip; // Chip of interest + while ((outChip = pmFPAviewNextChip(view, outFPA, 1))) { + if (!pmFPAfileIOChecks(config, view, PM_FPA_BEFORE)) { + goto ERROR; + } + pmCell *outCell; // Cell of interest + while ((outCell = pmFPAviewNextCell(view, outFPA, 1))) { + if (!pmFPAfileIOChecks(config, view, PM_FPA_BEFORE)) { + goto ERROR; + } + + pmHDU *hdu = pmHDUGetLowest(outFPA, outChip, outCell); // HDU for cell + if (!hdu || hdu->blankPHU) { + // No data here + continue; + } + + float shutterRef = NAN; // Reference shutter correction + if (type == PPMERGE_TYPE_SHUTTER) { + shutterRef = pmShutterCorrectionReference(shutters->data[cellNum]); + } + + pmReadout *outRO = pmReadoutAlloc(outCell); + + // open the input files (we need to do the work ourselves) + for (int i = 0; i < numFiles; i++) { + if (!ppMergeFileOpenInput(config, view, i)) { + psError(PS_ERR_UNKNOWN, false, "Unable to open file %d", i); + goto ERROR; + } + } + + psArray *fileGroups = psArrayAlloc (nThreads + 1); + + // generate readouts for each input file in each file group + for (int i = 0; i < fileGroups->n; i++) { + psArray *readouts = psArrayAlloc(numFiles); // Input readouts + for (int j = 0; j < numFiles; j++) { + pmFPAfile *input = pmFPAfileSelectSingle(config->files, "PPMERGE.INPUT", i); + pmCell *inCell = pmFPAviewThisCell(view, input->fpa); // Input cell + readouts->data[j] = pmReadoutAlloc(inCell); + } + + ppMergeFileGroup *fileGroup = ppMergeFileGroupAlloc(); + fileGroup->readouts = readouts; + fileGroup->read = false; + fileGroup->busy = false; + fileGroups->data[i] = fileGroup; + } + + // Read input data by chunks + for (int numChunk = 0; true; numChunk++) { + + ppMergeFileGroup *fileGroup = ppMergeReadChunk (&status, fileGroups, config, numChunk); + if (!status) goto ERROR; + if (!fileGroup) break; + + 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; + } + 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; + } + break; + case PPMERGE_TYPE_BIAS: + 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; + } + break; + default: + psAbort("Should never get here."); + } + } + + // 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); + } + } + + // 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); + } + if (!pmConceptsAverageCells(outCell, inCells, NULL, NULL, true)) { + psError(PS_ERR_UNKNOWN, false, "Unable to average cell concepts."); + psFree(inCells); + psFree(outRO); + goto ERROR; + } + psFree(inCells); + + psFree(fileGroups); + + // Plug supplementary images into their own FPAs + { + pmCell *countsCell = pmFPAfileThisCell(config->files, view, "PPMERGE.OUTPUT.COUNT"); + pmReadout *countsRO = pmReadoutAlloc(countsCell); // Readout with count of inputs per pixel + psImage *counts = psMetadataLookupPtr(NULL, outRO->analysis, PM_READOUT_STACK_ANALYSIS_COUNT); + countsRO->image = psImageCopy(countsRO->image, counts, PS_TYPE_F32); + psMetadataRemoveKey(outRO->analysis, PM_READOUT_STACK_ANALYSIS_COUNT); + countsRO->data_exists = countsCell->data_exists = countsCell->parent->data_exists = true; + psFree(countsRO); + + pmCell *sigmaCell = pmFPAfileThisCell(config->files, view, "PPMERGE.OUTPUT.SIGMA"); + pmReadout *sigmaRO = pmReadoutAlloc(sigmaCell); // Readout with stdev per pixel + psImage *sigma = psMetadataLookupPtr(NULL, outRO->analysis, PM_READOUT_STACK_ANALYSIS_SIGMA); + sigmaRO->image = psImageCopy(sigmaRO->image, sigma, PS_TYPE_F32); + psMetadataRemoveKey(outRO->analysis, PM_READOUT_STACK_ANALYSIS_SIGMA); + sigmaRO->data_exists = sigmaCell->data_exists = sigmaCell->parent->data_exists = true; + psFree(sigmaRO); + } + + // Measure the fringes for this cell + // + // XXX Need to deal with multiple components: we will do this by building up the components + // Read the existing fringe measurements + // Use existing regions to measure fringe statistics + // Add the new fringe measurements to the existing fringe measurements. + // Write the appended fringe measurements. + // Read in the "output" file to get the existing components. + // Put the new readout into the cell after the existing readouts. + if (type == PPMERGE_TYPE_FRINGE && outRO) { + pmFringeRegions *regions = pmFringeRegionsAlloc(fringeNum, fringeSize, fringeSize, + fringeSmoothX, fringeSmoothY); + pmFringeStats *fringe = pmFringeStatsMeasure(regions, outRO, maskVal); + psFree(regions); + if (!fringe) { + psError(PS_ERR_UNKNOWN, false, "Unable to measure fringe statistics.\n"); + psFree(outRO); + goto ERROR; + } + + psArray *fringes = psArrayAlloc(1); // Array of fringes + fringes->data[0] = fringe; + + pmFringesFormat(outCell, NULL, fringes); + psFree(fringes); // Drop reference + } + + if (!ppStatsFPA(stats, outFPA, view, maskVal, config)) { + psError(PS_ERR_UNKNOWN, true, "Unable to generate stats for image."); + goto ERROR; + } + + psFree(outRO); + cellNum++; + + if (!pmFPAfileIOChecks(config, view, PM_FPA_AFTER)) { + goto ERROR; + } + } + + if (outChip->data_exists) { + psList *inChips = psListAlloc(NULL); + for (int i=0; i < numFiles; i++) { + pmChip *chip = pmFPAviewThisChip(view, ((pmFPAfile *)inputs->data[i])->fpa); + psListAdd(inChips, PS_LIST_TAIL, chip); + } + if (!pmConceptsAverageChips(outChip, inChips, true)) { + psError(PS_ERR_UNKNOWN, false, "Unable to average Chip concepts."); + psFree(inChips); + goto ERROR; + } + psFree(inChips); + } + if (!pmFPAfileIOChecks(config, view, PM_FPA_AFTER)) { + goto ERROR; + } + } + + // Get list of FPAs for concepts averaging + psList *inFPAs = psListAlloc(NULL); // List of FPAs + for (int i = 0; i < numFiles; i++) { + pmFPAfile *input = inputs->data[i]; // Input file + psListAdd(inFPAs, PS_LIST_TAIL, input->fpa); + } + if (!pmConceptsAverageFPAs(output->fpa, inFPAs)) { + psError(PS_ERR_UNKNOWN, false, "Unable to average FPA concepts."); + psFree(inFPAs); + goto ERROR; + } + psFree(inFPAs); + + + if (!pmFPAfileIOChecks(config, view, PM_FPA_AFTER)) { + goto ERROR; + } + + psFree(view); + psFree(combination); + psFree(inputs); + psFree(masks); + psFree(weights); + psFree(stats); + return true; + +ERROR: + psFree(view); + psFree(combination); + psFree(inputs); + psFree(masks); + psFree(weights); + psFree(stats); + return false; +} + Index: trunk/ppMerge/src/ppMergeReadChunk.c =================================================================== --- trunk/ppMerge/src/ppMergeReadChunk.c (revision 18757) +++ trunk/ppMerge/src/ppMergeReadChunk.c (revision 18757) @@ -0,0 +1,81 @@ +# include "ppMerge.h" + +ppMergeFileGroup *ppMergeReadChunk (psArray *fileGroups, pmConfig *config, int numChunk) { + + // 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 + for (int j = 0; j < fileGroups->n; j++) { + ppMergeFileGroup *fileGroup = fileGroups->data[j]; + if (fileGroup->read) continue; + + psArray *readouts = fileGroup->readouts; + + psTrace("ppStack", 2, "Reading data for chunk %d into fileGroup %d....n", numChunk, j); + for (int i = 0; i < readouts->n; i++) { + pmReadout *inRO = readouts->data[i]; // Input readout + + // 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); + 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); + return NULL; + } + } + } + fileGroup->read = fileGroup->busy = true; + return fileGroup; + } + + // check for any fileGroups which are done processing + bool wait = true; + bool more = true; + 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; + } + if (!more) return NULL; + + if (wait) usleep (10000); + } + return NULL; +} Index: trunk/ppMerge/src/ppMergeThreadLauncher.c =================================================================== --- trunk/ppMerge/src/ppMergeThreadLauncher.c (revision 18757) +++ trunk/ppMerge/src/ppMergeThreadLauncher.c (revision 18757) @@ -0,0 +1,93 @@ +# include "ppMerge.h" + +// each thread runs this function, starting a new job when it finished with an old one +// it is called with a (void *) pointer to its own thread pointer +void *ppMergeThreadLauncher (void *data) { + + psThread *self = data; + psThreadJob *job = NULL; + + while (1) { + + // if we get an error, just wait until we are cleared or killed + while (self->fault) { + usleep (10000); + } + + // request a new job, if there are none available, sleep a bit + // we have to lock here so the job queue cannot be empty yet no threads busy + psThreadLock(); + while ((job = psThreadJobGetPending ()) == NULL) { + psThreadUnlock(); + usleep (10000); + } + self->busy = true; + psThreadUnlock(); + + // list all allowed job types here + + // pmReadoutCombine + if (!strcmp (job->type, "PPMERGE_READOUT_COMBINE")) { + psAssert (job->args->n == 5, "invalid number of arguments to pmReadoutCombine"); + + 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); + if (!status) { + self->fault = true; + } + // we do not have to lock here because this transition is not tied to the job queue + fileGroup->busy = false; + self->busy = false; + continue; + } + + // pmDarkCombine + if (!strcmp (job->type, "PPMERGE_DARK_COMBINE")) { + psAssert (job->args->n == 7, "invalid number of arguments to pmDarkCombine"); + + 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); + if (!status) { + self->fault = true; + } + // we do not have to lock here because this transition is not tied to the job queue + fileGroup->busy = false; + self->busy = false; + continue; + } + + // pmShutterCorrectionGenerate + if (!strcmp (job->type, "PPMERGE_SHUTTER_CORRECTION")) { + psAssert (job->args->n == 7, "invalid number of arguments to pmDarkCombine"); + + 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); + if (!status) { + self->fault = true; + } + // we do not have to lock here because this transition is not tied to the job queue + fileGroup->busy = false; + self->busy = false; + continue; + } + } +}