Index: /trunk/FACT++/src/EventBuilder.c =================================================================== --- /trunk/FACT++/src/EventBuilder.c (revision 15345) +++ /trunk/FACT++/src/EventBuilder.c (revision 15346) @@ -866,6 +866,5 @@ #endif evtCtrl.lastPtr++; - if (evtCtrl.lastPtr == MAX_EVT * MAX_RUN) - evtCtrl.lastPtr = 0; + evtCtrl.lastPtr %= MAX_EVT * MAX_RUN; gi.evtGet++; @@ -977,4 +976,48 @@ return; } /*-----------------------------------------------------------------*/ + +void reportIncomplete(int k0) +{ + const int id = evtCtrl.evtBuf[k0]; + + factPrintf(kWarn, 601, "%5d skip incomplete evt %8d %8d %2d", + mBuffer[id].evNum, evtCtrl.evtBuf[k0], k0, evtCtrl.evtStat[k0]); + + uint64_t report = 0; + + char str[1000]; + + int ik=0; + for (int ib=0; ib=0) // data received from that board + { + str[ik++] = '0'+(jb%10); + continue; + } + + // FIXME: Is that really 'b' or should that be 'ib' ? + if (gi_NumConnect[ib]<=0) // board not connected + { + str[ik++] = 'x'; + continue; + } + + // data from this board lost + str[ik++] = '.'; + report |= ((uint64_t)1)< 0 && evtCtrl.evtStat[k0] < 92) { - gj.bufNew++; //incomplete event in Buffer - if (evtCtrl.evtStat[k0] < 90 - && evtCtrl.pcTime[k0] < g_actTime - 30) { - int id = evtCtrl.evtBuf[k0]; - factPrintf(kWarn, 601, "%5d skip incomplete evt %8d %8d %2d", - mBuffer[id].evNum, evtCtrl.evtBuf[k0], k0, evtCtrl.evtStat[k0]); - - uint64_t report = 0; - - char str[1000]; - - int ik,ib,jb; - ik=0; - for (ib=0; ib0 ) { - str[ik++] = '.'; - report |= ((uint64_t)1)<= 9000 //'delete' - || evtCtrl.evtStat[k0] == 0) { //'useless' - - int id = evtCtrl.evtBuf[k0]; + gi_SecTime = g_actTime; + + + //loop over all active events and flag those older than read-timeout + //delete those that are written to disk .... + + int kd = evtCtrl.lastPtr - evtCtrl.frstPtr; + if (kd < 0) + kd += (MAX_EVT * MAX_RUN); + + gj.bufNew = gj.bufEvt = 0; + + const int k1 = evtCtrl.frstPtr; + for (int k=k1; k0 && evtCtrl.evtStat[k0]<92) + { + gj.bufNew++; //incomplete event in Buffer + + // Event has not yet timed out or was reported already + if (evtCtrl.evtStat[k0]>=90 || evtCtrl.pcTime[k0]>=g_actTime - 30) + continue; + + reportIncomplete(k0); + + evtCtrl.evtStat[k0] = 91; //timeout for incomplete events + gi.evtSkp++; + gi.evtTot++; + gj.evtSkip++; + + continue; + } + + // complete event in Buffer + if (evtCtrl.evtStat[k0] >= 95) + gj.bufEvt++; + + // Check the less likely case: 'useless' or 'delete' + if (evtCtrl.evtStat[k0]==0 || evtCtrl.evtStat[k0] >= 9000) + { + const int id = evtCtrl.evtBuf[k0]; #ifdef EVTDEBUG - factPrintf(kDebug, -1, "%5d free event buffer, nb=%3d", mBuffer[id].evNum, mBuffer[id].nBoard); -#endif - mBufFree (id); //event written--> free memory - evtCtrl.evtStat[k0] = -1; - gj.evtWrite++; - gj.rateWrite++; - } else if (evtCtrl.evtStat[k0] >= 95) { - gj.bufEvt++; //complete event in Buffer - } - - if (k0 == evtCtrl.frstPtr && evtCtrl.evtStat[k0] < 0) { - evtCtrl.frstPtr = (evtCtrl.frstPtr + 1) % (MAX_EVT * MAX_RUN); - } - } + factPrintf(kDebug, -1, "%5d free event buffer, nb=%3d", mBuffer[id].evNum, mBuffer[id].nBoard); +#endif + mBufFree (id); //event written--> free memory + evtCtrl.evtStat[k0] = -1; + gj.evtWrite++; + gj.rateWrite++; + } + + } @@ -1880,23 +1910,22 @@ - if (gi_resetR > 0) { - //flag all events as 'read finished' - int kd = evtCtrl.lastPtr - evtCtrl.frstPtr; - if (kd < 0) - kd += (MAX_EVT * MAX_RUN); - - int k1 = evtCtrl.frstPtr; - for (k = k1; k < (k1 + kd); k++) { - int k0 = k % (MAX_EVT * MAX_RUN); - if (evtCtrl.evtStat[k0] > 0 && evtCtrl.evtStat[k0] < 90) { - evtCtrl.evtStat[k0] = 91; - gi.evtSkp++; - gi.evtTot++; - } - } - - xwait.tv_sec = 0; - xwait.tv_nsec = 1000; // sleep for ~1 usec - nanosleep (&xwait, NULL); + if (gi_resetR > 0) + { + //flag all events as 'read finished' + int kd = evtCtrl.lastPtr - evtCtrl.frstPtr; + if (kd < 0) + kd += (MAX_EVT * MAX_RUN); + + const int k1 = evtCtrl.frstPtr; + for (int k=k1; k 0 && evtCtrl.evtStat[k0] < 90) + { + evtCtrl.evtStat[k0] = 91; + gi.evtSkp++; + gi.evtTot++; + } + } //and clear all buffers (might have to wait until all others are done) @@ -1911,30 +1940,37 @@ int numclear = 1; - while (numclear > 0) { + while (numclear > 0) + { numclear = 0; + int kd = evtCtrl.lastPtr - evtCtrl.frstPtr; if (kd < 0) - kd += (MAX_EVT * MAX_RUN); - - int k1 = evtCtrl.frstPtr; - for (k = k1; k < (k1 + kd); k++) { - int k0 = k % (MAX_EVT * MAX_RUN); - if (evtCtrl.evtStat[k0] > minclear) { - int id = evtCtrl.evtBuf[k0]; + kd += (MAX_EVT * MAX_RUN); + + const int k1 = evtCtrl.frstPtr; + for (int k=k1; k minclear) + { + const int id = evtCtrl.evtBuf[k0]; #ifdef EVTDEBUG - factPrintf(kDebug, -1, "ev %5d free event buffer, nb=%3d", mBuffer[id].evNum, mBuffer[id].nBoard); -#endif - mBufFree (id); //event written--> free memory - evtCtrl.evtStat[k0] = -1; - } else if (evtCtrl.evtStat[k0] > 0) - numclear++; //writing is still ongoing... - - if (k0 == evtCtrl.frstPtr && evtCtrl.evtStat[k0] < 0) - evtCtrl.frstPtr = (evtCtrl.frstPtr + 1) % (MAX_EVT * MAX_RUN); + factPrintf(kDebug, -1, "ev %5d free event buffer, nb=%3d", mBuffer[id].evNum, mBuffer[id].nBoard); +#endif + mBufFree (id); //event written--> free memory + evtCtrl.evtStat[k0] = -1; + } + else + if (evtCtrl.evtStat[k0] > 0) + numclear++; //writing is still ongoing... + + if (k0 == evtCtrl.frstPtr && evtCtrl.evtStat[k0] < 0) + { + evtCtrl.frstPtr++; + evtCtrl.frstPtr %= MAX_EVT * MAX_RUN; + } } - xwait.tv_sec = 0; - xwait.tv_nsec = 1000; // sleep for ~1 umsec - nanosleep (&xwait, NULL); + usleep(1); } } @@ -1972,70 +2008,75 @@ -void * -subProc (void *thrid) -{ - int64_t threadID; - int numWait, numProc, k, jret; - struct timespec xwait; - -// int32_t cntr ; - - threadID = (int64_t)thrid; - - factPrintf(kInfo, -1, "Starting sub-process-thread %ld", threadID); - - while (g_runStat > -2) { //in case of 'exit' we still must process pending events - numWait = numProc = 0; - int kd = evtCtrl.lastPtr - evtCtrl.frstPtr; - if (kd < 0) - kd += (MAX_EVT * MAX_RUN); - - int k1 = evtCtrl.frstPtr; - for (k = k1; k < (k1 + kd); k++) { - int k0 = k % (MAX_EVT * MAX_RUN); - - if (evtCtrl.evtStat[k0] == 1000 + threadID) { - if (gi_resetR > 1) { //we are asked to flush buffers asap - jret = 9100; //flag to be deleted - } else { - int id = evtCtrl.evtBuf[k0]; - - - jret = - subProcEvt (threadID, mBuffer[id].FADhead, - mBuffer[id].fEvent, mBuffer[id].buffer); - - - if (jret <= threadID) { - factPrintf(kError, -1, "Process %ld wants to send event to process %d... not allowed.", threadID, jret); - jret = 5300; - } else if (jret <= 0) - jret = 9200 + threadID; //flag as 'to be deleted' - else if (jret >= gi_maxProc) - jret = 5200 + threadID; //flag as 'to be written' - else - jret = 1000 + jret; //flag for next proces +void *subProc(void *thrid) +{ + const int64_t threadID = (int64_t)thrid; + + factPrintf(kInfo, -1, "Starting sub-process-thread %ld", threadID); + + while (g_runStat > -2) //in case of 'exit' we still must process pending events + { + int numWait = 0; + int numProc = 0; + + int kd = evtCtrl.lastPtr - evtCtrl.frstPtr; + if (kd < 0) + kd += (MAX_EVT * MAX_RUN); + + const int k1 = evtCtrl.frstPtr; + for (int k=k1; k1 : flush buffers asap) + + if (gi_resetR<=1) + { + const int id = evtCtrl.evtBuf[k0]; + + jret = subProcEvt(threadID, mBuffer[id].FADhead, + mBuffer[id].fEvent, mBuffer[id].buffer); + + + if (jret <= threadID) { + factPrintf(kError, -1, "Process %ld wants to send event to process %d... not allowed.", threadID, jret); + jret = 5300; + } else if (jret <= 0) + jret = 9200 + threadID; // flag as 'to be deleted' + else if (jret >= gi_maxProc) + jret = 5200 + threadID; // flag as 'to be written' + else + jret = 1000 + jret; // flag for next proces + } + evtCtrl.evtStat[k0] = jret; numProc++; - } else if (evtCtrl.evtStat[k0] < 1000 + threadID) - numWait++; - } - - if (gj.readStat < -10 && numWait == 0) { //nothing left to do - factPrintf(kInfo, -1, "Exit subProcessing in process %ld", threadID); - return 0; - } - if (numProc == 0) { - //seems we have nothing to do, so sleep a little - xwait.tv_sec = 0; - xwait.tv_nsec = 1000; // sleep for ~1 usec - nanosleep (&xwait, NULL); - } - } - - factPrintf(kInfo, -1, "Ending sub-process-thread %ld", threadID); - return 0; -} /*-----------------------------------------------------------------*/ + } + + if (gj.readStat < -10 && numWait == 0) { //nothing left to do + factPrintf(kInfo, -1, "Exit subProcessing in process %ld", threadID); + return 0; + } + + //seems we have nothing to do, so sleep a little + if (numProc == 0) + usleep(1); + } + + factPrintf(kInfo, -1, "Ending sub-process-thread %ld", threadID); + + return 0; +} + +/*-----------------------------------------------------------------*/ @@ -2044,7 +2085,5 @@ { /* *** main loop processing file, including SW-trigger */ - int numProc, numWait; int status, j; - struct timespec xwait; int lastRun = 0; //usually run from last event still valid @@ -2080,188 +2119,211 @@ } - while (g_runStat > -2) { //in case of 'exit' we still must process pending events - - numWait = numProc = 0; - int kd = evtCtrl.lastPtr - evtCtrl.frstPtr; - if (kd < 0) - kd += (MAX_EVT * MAX_RUN); - - int k1 = evtCtrl.frstPtr; - for (int k = k1; k < (k1 + kd); k++) { - int k0 = k % (MAX_EVT * MAX_RUN); -//would be better to use bitmaps for evtStat (allow '&' instead of multi-if) - if (evtCtrl.evtStat[k0] > 90 && evtCtrl.evtStat[k0] < 1000) { - - if (gi_resetR > 1) { //we are asked to flush buffers asap + // in case of 'exit' we still must process pending events + while (g_runStat > -2) + { + int numWait = 0; + int numProc = 0; + + int kd = evtCtrl.lastPtr - evtCtrl.frstPtr; + if (kd < 0) + kd += (MAX_EVT * MAX_RUN); + + const int k1 = evtCtrl.frstPtr; + for (int k=k1; k 90 && evtCtrl.evtStat[k0] < 1000)) + { + if (evtCtrl.evtStat[k0] >= 0 && evtCtrl.evtStat[k0] < 90) + numWait++; + + continue; + } + + /*** if (evtCtrl.evtStat[k0] > 90 && evtCtrl.evtStat[k0] < 1000) ***/ + + //we are asked to flush buffers asap + if (gi_resetR > 1) + { evtCtrl.evtStat[k0] = 9991; - } else { - -//-------- it is better to open the run already here, so call can be used to initialize -//-------- buffers etc. needed to interprete run (e.g. DRS calibration) - int id = evtCtrl.evtBuf[k0]; - uint32_t irun = mBuffer[id].runNum; - int32_t ievt = mBuffer[id].evNum; - if (runCtrl[lastRun].runId == irun) { - j = lastRun; - } else { - //check which fileID to use (or open if needed) - for (j = 0; j < MAX_RUN; j++) { - if (runCtrl[j].runId == irun) - break; - } - if (j >= MAX_RUN) { - factPrintf(kFatal, 901, "procEvt: Can not find run %d for event %d in %d", irun, ievt, id); - } - lastRun = j; + continue; + } + + //-------- it is better to open the run already here, so call can be used to initialize + //-------- buffers etc. needed to interprete run (e.g. DRS calibration) + const int id = evtCtrl.evtBuf[k0]; + + const uint32_t irun = mBuffer[id].runNum; + const int32_t ievt = mBuffer[id].evNum; + + int j = lastRun; + if (runCtrl[lastRun].runId != irun) + { + //check which fileID to use (or open if needed) + for (j = 0; j < MAX_RUN; j++) { + if (runCtrl[j].runId == irun) + break; } - - if (runCtrl[j].fileId < 0) { -//---- we need to open a new run ==> make sure all older runs are -//---- finished and marked to be closed .... - int j1; - for (j1 = 0; j1 < MAX_RUN; j1++) { - if (runCtrl[j1].fileId == 0) { - runCtrl[j1].procId = 2; //--> do no longer accept events for processing -//---- problem: processing still going on ==> must wait for closing .... - factPrintf(kInfo, -1, "procEvt: Finished run since new one opened %d", runCtrl[j1].runId); - runFinish1 (runCtrl[j1].runId); - } - - } - - actRun.Version = 1; - actRun.RunType = -1; //to be adapted - - actRun.Nroi = runCtrl[j].roi0; - actRun.NroiTM = runCtrl[j].roi8; -//ETIENNE don't reset it to zero as it is taken care of in DataWriteFits -// if (actRun.Nroi == actRun.NroiTM) -// actRun.NroiTM = 0; - actRun.RunTime = runCtrl[j].firstTime; - actRun.RunUsec = runCtrl[j].firstTime; - actRun.NBoard = NBOARDS; - actRun.NPix = NPIX; - actRun.NTm = NTMARK; - actRun.Nroi = mBuffer[id].nRoi; - memcpy (actRun.FADhead, mBuffer[id].FADhead, - NBOARDS * sizeof (PEVNT_HEADER)); - - runCtrl[j].fileHd = - runOpen (irun, &actRun, sizeof (actRun)); - if (runCtrl[j].fileHd == NULL) { - factPrintf(kError, 502, "procEvt: Could not open a file for run %d (runOpen failed)", irun); - runCtrl[j].fileId = 91; - runCtrl[j].procId = 91; - } else { - factPrintf(kInfo, -1, "procEvt: Opened new file for run %d (evt=%d)", irun, ievt); - runCtrl[j].fileId = 0; - runCtrl[j].procId = 0; - } - + if (j >= MAX_RUN) { + factPrintf(kFatal, 901, "procEvt: Can not find run %d for event %d in %d", irun, ievt, id); } -//-------- also check if run shall be closed (==> skip event, but do not close the file !!! ) - if (runCtrl[j].procId == 0) { - if (runCtrl[j].closeTime < g_actTime - || runCtrl[j].lastTime < g_actTime - 300 - || runCtrl[j].maxEvt <= runCtrl[j].procEvt) { - factPrintf(kInfo, 502, "procEvt: Reached end of run condition for run %d", irun); - runFinish1 (runCtrl[j].runId); - runCtrl[j].procId = 1; - } + lastRun = j; + } + + if (runCtrl[j].fileId < 0) + { + //---- we need to open a new run ==> make sure all older runs are + //---- finished and marked to be closed .... + int j1; + for (j1 = 0; j1 < MAX_RUN; j1++) { + if (runCtrl[j1].fileId == 0) { + runCtrl[j1].procId = 2; //--> do no longer accept events for processing + //---- problem: processing still going on ==> must wait for closing .... + factPrintf(kInfo, -1, "procEvt: Finished run since new one opened %d", runCtrl[j1].runId); + runFinish1(runCtrl[j1].runId); + } } - if (runCtrl[j].procId != 0) { + + actRun.Version = 1; + actRun.RunType = -1; //to be adapted + + actRun.Nroi = runCtrl[j].roi0; + actRun.NroiTM = runCtrl[j].roi8; + //ETIENNE don't reset it to zero as it is taken care of in DataWriteFits + // if (actRun.Nroi == actRun.NroiTM) + // actRun.NroiTM = 0; + actRun.RunTime = runCtrl[j].firstTime; + actRun.RunUsec = runCtrl[j].firstTime; + actRun.NBoard = NBOARDS; + actRun.NPix = NPIX; + actRun.NTm = NTMARK; + actRun.Nroi = mBuffer[id].nRoi; + + memcpy(actRun.FADhead, mBuffer[id].FADhead, NBOARDS*sizeof(PEVNT_HEADER)); + + runCtrl[j].fileHd = runOpen (irun, &actRun, sizeof (actRun)); + if (runCtrl[j].fileHd == NULL) + { + factPrintf(kError, 502, "procEvt: Could not open a file for run %d (runOpen failed)", irun); + runCtrl[j].fileId = 91; + runCtrl[j].procId = 91; + } + else + { + factPrintf(kInfo, -1, "procEvt: Opened new file for run %d (evt=%d)", irun, ievt); + runCtrl[j].fileId = 0; + runCtrl[j].procId = 0; + } + } + + //-------- also check if run shall be closed (==> skip event, but do not close the file !!! ) + if (runCtrl[j].procId == 0) + { + if (runCtrl[j].closeTime < g_actTime || + runCtrl[j].lastTime < g_actTime - 300 || + runCtrl[j].maxEvt <= runCtrl[j].procEvt) + { + factPrintf(kInfo, 502, "procEvt: Reached end of run condition for run %d", irun); + runFinish1 (runCtrl[j].runId); + runCtrl[j].procId = 1; + } + } + if (runCtrl[j].procId != 0) { #ifdef EVTDEBUG - fcatPrintf(kDebug, 502, "procEvt: Skip event %d because no active run %d", ievt, irun); -#endif - evtCtrl.evtStat[k0] = 9091; - } else { -//-------- -//-------- - id = evtCtrl.evtBuf[k0]; - int itevt = mBuffer[id].trgNum; - int itrg = mBuffer[id].trgTyp; - int roi = mBuffer[id].nRoi; - int roiTM = mBuffer[id].nRoiTM; - -//make sure unused pixels/tmarks are cleared to zero -//ETIENNE don't reset it to zero as it is taken care of in DataWriteFits -// if (roiTM == roi) -// roiTM = 0; - int ip, it, dest, ib; - for (ip = 0; ip < NPIX; ip++) { - if (mBuffer[id].fEvent->StartPix[ip] == -1) { - dest = ip * roi; - memset (&mBuffer[id].fEvent->Adc_Data[dest], 0, roi * 2); - } - } - - for (it = 0; it < NTMARK; it++) { - if (mBuffer[id].fEvent->StartTM[it] == -1) { - dest = it * roi + NPIX * roi; - memset (&mBuffer[id].fEvent->Adc_Data[dest], 0, roi * 2); - } - } - - -//and set correct event header ; also check for consistency in event (not yet) - mBuffer[id].fEvent->Roi = roi; - mBuffer[id].fEvent->RoiTM = roiTM; - mBuffer[id].fEvent->EventNum = ievt; - mBuffer[id].fEvent->TriggerNum = itevt; - mBuffer[id].fEvent->TriggerType = itrg; - mBuffer[id].fEvent->Errors[0] = mBuffer[id].Errors[0]; - mBuffer[id].fEvent->Errors[1] = mBuffer[id].Errors[1]; - mBuffer[id].fEvent->Errors[2] = mBuffer[id].Errors[2]; - mBuffer[id].fEvent->Errors[3] = mBuffer[id].Errors[3]; - mBuffer[id].fEvent->SoftTrig = 0; - - - for (ib = 0; ib < NBOARDS; ib++) { - if (mBuffer[id].board[ib] == -1) { //board is not read - mBuffer[id].FADhead[ib].start_package_flag = 0; - mBuffer[id].fEvent->BoardTime[ib] = 0; - } else { - mBuffer[id].fEvent->BoardTime[ib] = - mBuffer[id].FADhead[ib].time; - } - } - int i = eventCheck (mBuffer[id].runNum, mBuffer[id].FADhead, - mBuffer[id].fEvent); - gi.procTot++; - numProc++; - - if (i < 0) { - evtCtrl.evtStat[k0] = 9999; //flag event to be skipped - gi.procErr++; - } else { - evtCtrl.evtStat[k0] = 1000; - runCtrl[j].procEvt++; - } + factPrintf(kDebug, 502, "procEvt: Skip event %d because no active run %d", ievt, irun); +#endif + evtCtrl.evtStat[k0] = 9091; + continue; + } + + //-------- + //-------- + + const int roi = mBuffer[id].nRoi; + //const int roiTM = mBuffer[id].nRoiTM; + + //make sure unused pixels/tmarks are cleared to zero + //ETIENNE don't reset it to zero as it is taken care of in DataWriteFits + // if (roiTM == roi) + // roiTM = 0; + for (int ip=0; ipStartPix[ip] == -1) + { + const int dest = ip * roi; + memset(&mBuffer[id].fEvent->Adc_Data[dest], 0, roi * 2); } - } - } else if (evtCtrl.evtStat[k0] >= 0 && evtCtrl.evtStat[k0] < 90) { - numWait++; - } - } - - if (gj.readStat < -10 && numWait == 0) { //nothing left to do - factPrintf(kInfo, -1, "Exit Processing Process ..."); - gp_runStat = -22; //==> we should exit - gj.procStat = -22; //==> we should exit - return 0; - } - - if (numProc == 0) { - //seems we have nothing to do, so sleep a little - xwait.tv_sec = 0; - xwait.tv_nsec = 1000; // sleep for ~1 usec - nanosleep (&xwait, NULL); - } - gp_runStat = gi_runStat; - gj.procStat = gj.readStat; - - } - - //we are asked to abort asap ==> must flag all remaining events + } + + for (int it=0; itStartTM[it] == -1) + { + const int dest = it * roi + NPIX * roi; + memset(&mBuffer[id].fEvent->Adc_Data[dest], 0, roi * 2); + } + } + + //and set correct event header ; also check for consistency in event (not yet) + mBuffer[id].fEvent->Roi = mBuffer[id].nRoi; + mBuffer[id].fEvent->RoiTM = mBuffer[id].nRoiTM; + mBuffer[id].fEvent->EventNum = mBuffer[id].evNum; + mBuffer[id].fEvent->TriggerNum = mBuffer[id].trgNum; + mBuffer[id].fEvent->TriggerType = mBuffer[id].trgTyp; + mBuffer[id].fEvent->Errors[0] = mBuffer[id].Errors[0]; + mBuffer[id].fEvent->Errors[1] = mBuffer[id].Errors[1]; + mBuffer[id].fEvent->Errors[2] = mBuffer[id].Errors[2]; + mBuffer[id].fEvent->Errors[3] = mBuffer[id].Errors[3]; + mBuffer[id].fEvent->SoftTrig = 0; + + + for (int ib=0; ibBoardTime[ib] = 0; + } + else + { + mBuffer[id].fEvent->BoardTime[ib] = mBuffer[id].FADhead[ib].time; + } + } + + const int rc = eventCheck(mBuffer[id].runNum, mBuffer[id].FADhead, + mBuffer[id].fEvent); + gi.procTot++; + numProc++; + + if (rc < 0) + { + evtCtrl.evtStat[k0] = 9999; //flag event to be skipped + gi.procErr++; + } + else + { + evtCtrl.evtStat[k0] = 1000; + runCtrl[j].procEvt++; + } + } + + if (gj.readStat < -10 && numWait == 0) { //nothing left to do + factPrintf(kInfo, -1, "Exit Processing Process ..."); + gp_runStat = -22; //==> we should exit + gj.procStat = -22; //==> we should exit + return 0; + } + + //seems we have nothing to do, so sleep a little + if (numProc == 0) + usleep(1); + + gp_runStat = gi_runStat; + gj.procStat = gj.readStat; + + } + + //we are asked to abort asap ==> must flag all remaining events // when gi_runStat claims that all events are in the buffer... @@ -2275,10 +2337,10 @@ } - int k1 = evtCtrl.frstPtr; - for (int k = k1; k < (k1 + kd); k++) { - int k0 = k % (MAX_EVT * MAX_RUN); - if (evtCtrl.evtStat[k0] >= 0 && evtCtrl.evtStat[k0] < 1000) { - evtCtrl.evtStat[k0] = 9800; //flag event as 'processed' - } + const int k1 = evtCtrl.frstPtr; + for (int k=k1; k= 0 && evtCtrl.evtStat[k0] < 1000) + evtCtrl.evtStat[k0] = 9800; //flag event as 'processed' } @@ -2332,5 +2394,5 @@ runCtrl[j].closeTime >= g_actTime && runCtrl[j].lastTime >= g_actTime - 300 && - runCtrl[j].maxEvt >= runCtrl[j].actEvt) + runCtrl[j].maxEvt > runCtrl[j].actEvt) return; @@ -2375,12 +2437,7 @@ -void * -writeEvt (void *ptr) +void *writeEvt (void *ptr) { /* *** main loop writing event (including opening and closing run-files */ - - int numWrite, numWait; - int k, j ; - struct timespec xwait; // cpu_set_t mask; @@ -2400,110 +2457,127 @@ int lastRun = 0; //usually run from last event still valid - while (g_runStat > -2) { - - numWait = numWrite = 0; - int kd = evtCtrl.lastPtr - evtCtrl.frstPtr; - if (kd < 0) - kd += (MAX_EVT * MAX_RUN); - - int k1 = evtCtrl.frstPtr; - for (k = k1; k < (k1 + kd); k++) { - int k0 = k % (MAX_EVT * MAX_RUN); -//would be better to use bitmaps for evtStat (allow '&' instead of multi-if) - if (evtCtrl.evtStat[k0] > 5000 && evtCtrl.evtStat[k0] < 9000) { - - if (gi_resetR > 1) { //we must drain the buffer asap + while (g_runStat > -2) + { + int numWrite = 0; + int numWait = 0; + + int kd = evtCtrl.lastPtr - evtCtrl.frstPtr; + if (kd < 0) + kd += (MAX_EVT * MAX_RUN); + + const int k1 = evtCtrl.frstPtr; + for (int k=k1; k= 9000) + { + if (evtCtrl.evtStat[k0] > 0 && evtCtrl.evtStat[k0] < 9000) + numWait++; + + continue; + } + + /*** if (evtCtrl.evtStat[k0] > 5000 && evtCtrl.evtStat[k0] < 9000) ***/ + + //we must drain the buffer asap + if (gi_resetR > 1) + { evtCtrl.evtStat[k0] = 9904; - } else { - - - int id = evtCtrl.evtBuf[k0]; - uint32_t irun = mBuffer[id].runNum; - int32_t ievt = mBuffer[id].evNum; - - gi.wrtTot++; - if (runCtrl[lastRun].runId == irun) { - j = lastRun; + continue; + } + + const int id = evtCtrl.evtBuf[k0]; + + const uint32_t irun = mBuffer[id].runNum; + const int32_t ievt = mBuffer[id].evNum; + + gi.wrtTot++; + + int j = lastRun; + + if (runCtrl[lastRun].runId != irun) + { + //check which fileID to use (or open if needed) + for (j = 0; j < MAX_RUN; j++) + { + if (runCtrl[j].runId == irun) + break; + } + + if (j >= MAX_RUN) + { + factPrintf(kFatal, 901, "writeEvt: Can not find run %d for event %d in %d", irun, ievt, id); + gi.wrtErr++; + } + + lastRun = j; + } + + if (runCtrl[j].fileId < 0) { + actRun.Version = 1; + actRun.RunType = -1; //to be adapted + + actRun.Nroi = runCtrl[j].roi0; + actRun.NroiTM = runCtrl[j].roi8; + //ETIENNE don't reset it to zero as it is taken care of in DataWriteFits + // if (actRun.Nroi == actRun.NroiTM) + // actRun.NroiTM = 0; + actRun.RunTime = runCtrl[j].firstTime; + actRun.RunUsec = runCtrl[j].firstTime; + actRun.NBoard = NBOARDS; + actRun.NPix = NPIX; + actRun.NTm = NTMARK; + actRun.Nroi = mBuffer[id].nRoi; + memcpy (actRun.FADhead, mBuffer[id].FADhead, + NBOARDS * sizeof (PEVNT_HEADER)); + + runCtrl[j].fileHd = + runOpen (irun, &actRun, sizeof (actRun)); + if (runCtrl[j].fileHd == NULL) { + factPrintf(kError, 502, "writeEvt: Could not open a file for run %d (runOpen failed)", irun); + runCtrl[j].fileId = 91; } else { - //check which fileID to use (or open if needed) - for (j = 0; j < MAX_RUN; j++) { - if (runCtrl[j].runId == irun) - break; - } - if (j >= MAX_RUN) { - factPrintf(kFatal, 901, "writeEvt: Can not find run %d for event %d in %d", irun, ievt, id); - gi.wrtErr++; - } - lastRun = j; + factPrintf(kInfo, -1, "writeEvt: Opened new file for run %d (evt %d)", irun, ievt); + runCtrl[j].fileId = 0; } + } + + if (runCtrl[j].fileId != 0) { if (runCtrl[j].fileId < 0) { - actRun.Version = 1; - actRun.RunType = -1; //to be adapted - - actRun.Nroi = runCtrl[j].roi0; - actRun.NroiTM = runCtrl[j].roi8; -//ETIENNE don't reset it to zero as it is taken care of in DataWriteFits -// if (actRun.Nroi == actRun.NroiTM) -// actRun.NroiTM = 0; - actRun.RunTime = runCtrl[j].firstTime; - actRun.RunUsec = runCtrl[j].firstTime; - actRun.NBoard = NBOARDS; - actRun.NPix = NPIX; - actRun.NTm = NTMARK; - actRun.Nroi = mBuffer[id].nRoi; - memcpy (actRun.FADhead, mBuffer[id].FADhead, - NBOARDS * sizeof (PEVNT_HEADER)); - - runCtrl[j].fileHd = - runOpen (irun, &actRun, sizeof (actRun)); - if (runCtrl[j].fileHd == NULL) { - factPrintf(kError, 502, "writeEvt: Could not open a file for run %d (runOpen failed)", irun); - runCtrl[j].fileId = 91; - } else { - factPrintf(kInfo, -1, "writeEvt: Opened new file for run %d (evt %d)", irun, ievt); - runCtrl[j].fileId = 0; - } - + factPrintf(kError, 123, "writeEvt: Never opened file for run %d", irun); + } else if (runCtrl[j].fileId < 100) { + factPrintf(kWarn, 123, "writeEvt: File for run %d is closed", irun); + runCtrl[j].fileId += 100; + } else { +#ifdef EVTDEBUG + factPrintf(kDebug, 123, "writeEvt: File for run %d is closed", irun); +#endif } - - if (runCtrl[j].fileId != 0) { - if (runCtrl[j].fileId < 0) { - factPrintf(kError, 123, "writeEvt: Never opened file for run %d", irun); - } else if (runCtrl[j].fileId < 100) { - factPrintf(kWarn, 123, "writeEvt: File for run %d is closed", irun); - runCtrl[j].fileId += 100; - } else { + evtCtrl.evtStat[k0] = 9903; + gi.wrtErr++; + } else { + // snprintf (str, MXSTR,"write event %d size %d",ievt,sizeof (mBuffer[id])); + // factOut (kInfo, 504, str); + const int rc = runWrite (runCtrl[j].fileHd, mBuffer[id].fEvent, + sizeof (mBuffer[id])); + if (rc >= 0) { + runCtrl[j].lastTime = g_actTime; + runCtrl[j].actEvt++; + evtCtrl.evtStat[k0] = 9901; #ifdef EVTDEBUG - factPrintf(kDebug, 123, "writeEvt: File for run %d is closed", irun); -#endif - } - evtCtrl.evtStat[k0] = 9903; - gi.wrtErr++; + factPrintf(kDebug, 504, "%5d successfully wrote for run %d id %5d", ievt, irun, k0); +#endif + // gj.writEvt++ ; } else { -// snprintf (str, MXSTR,"write event %d size %d",ievt,sizeof (mBuffer[id])); -// factOut (kInfo, 504, str); - const int rc = runWrite (runCtrl[j].fileHd, mBuffer[id].fEvent, - sizeof (mBuffer[id])); - if (rc >= 0) { - runCtrl[j].lastTime = g_actTime; - runCtrl[j].actEvt++; - evtCtrl.evtStat[k0] = 9901; -#ifdef EVTDEBUG - factPrintf(kDebug, 504, "%5d successfully wrote for run %d id %5d", ievt, irun, k0); -#endif -// gj.writEvt++ ; - } else { - factPrintf(kError, 503, "writeEvt: Writing event for run %d failed (runWrite)", irun); - evtCtrl.evtStat[k0] = 9902; - gi.wrtErr++; - } - - checkAndCloseRun(j, irun, rc<0, 1); + factPrintf(kError, 503, "writeEvt: Writing event for run %d failed (runWrite)", irun); + evtCtrl.evtStat[k0] = 9902; + gi.wrtErr++; } - } - } else if (evtCtrl.evtStat[k0] > 0 && evtCtrl.evtStat[k0] < 9000) - numWait++; - } + + checkAndCloseRun(j, irun, rc<0, 1); + } + } //check if we should close a run (mainly when no event pending) @@ -2516,5 +2590,5 @@ if (actrun != 0) {//If we have an active run, look for its start time - for (j=0;j