Index: trunk/FACT++/src/EventBuilder.c =================================================================== --- trunk/FACT++/src/EventBuilder.c (revision 15482) +++ trunk/FACT++/src/EventBuilder.c (revision 15483) @@ -69,6 +69,6 @@ int CloseRunFile (uint32_t runId, uint32_t closeTime, uint32_t maxEvt); -int evtCtrl_frstPtr; -int evtCtrl_lastPtr; +int evtCtrl_frstPtr; // First event in queue +int evtCtrl_lastPtr; // pointer to next free slot int g_maxProc; @@ -105,5 +105,8 @@ GUI_STAT gj; -EVT_CTRL evtCtrl[MAX_EVT * MAX_RUN]; //control of events during processing +#define MAX_EVT 65536 // ( 300s @ 220Hz; 16GB = 5000 evt @ roi=1024 (27s) ; 18000 evt @ roi = 300 ) +#define MAX_RUN 8 // Number of concurrent runs + +EVT_CTRL evtCtrl[MAX_EVT]; //control of events during processing void factPrintf(int severity, int id, const char *fmt, ...) @@ -377,12 +380,4 @@ int mBufEvt(const READ_STRUCT *rs) { -// generate a new Event into mBuffer: -// make sure only complete Event are possible, so 'free' will always work -// returns index into mBuffer[], or negative value in case of error -// error: <-9000 if roi screwed up (not consistent with run) -// <-8000 (not consistent with event) -// <-7000 (not consistent with board) -// < 0 if no space left - int nRoi[9]; if (!checkRoiConsistency(rs->rBuf, nRoi)) @@ -395,8 +390,8 @@ const int fadNum = rs->evtID; - const int beg = (evtCtrl_lastPtr + MAX_EVT*MAX_RUN - 1) % (MAX_EVT*MAX_RUN); - const int end = (evtCtrl_frstPtr + MAX_EVT*MAX_RUN - 1) % (MAX_EVT*MAX_RUN); - - for (int k=beg; k!=end; k=(k+MAX_EVT*MAX_RUN-1)%(MAX_EVT*MAX_RUN)) + const int beg = (evtCtrl_lastPtr + MAX_EVT - 1) % MAX_EVT; + const int end = (evtCtrl_frstPtr + MAX_EVT - 1) % MAX_EVT; + + for (int k=beg; k!=end; k=(k+MAX_EVT-1)%MAX_EVT) { // If the run is different, go on searching. @@ -437,5 +432,5 @@ // Check if the control structure still has space left - if (end-beg==1 || (end==0 && beg==MAX_EVT*MAX_RUN-1)) + if (end-beg==1 || (end==0 && beg==MAX_EVT-1)) { factPrintf(kError, 881, "No control slot to keep event %d (run %d) %d %d", evID, runID, beg, end); @@ -537,4 +532,13 @@ evtCtrl[k].FADhead = NULL; + // -1: kInValid + // 0: kValid + // 1-40: kIncomplete + // 90: kIncompleteReported + // 100: kCompleteEventInBuffer + // 1000+x: kToBeProcessedByThreadX + // 5000: kToBeWritten + // 10000: kToBeDeleted + evtCtrl[k].evtStat = 0; @@ -543,5 +547,5 @@ // in two instructions. Must be done only _after_ the contents // have been initialized - evtCtrl_lastPtr = (evtCtrl_lastPtr+1) % (MAX_EVT*MAX_RUN); + evtCtrl_lastPtr = (evtCtrl_lastPtr+1) % MAX_EVT; return k; @@ -638,8 +642,8 @@ }*/ /*-----------------------------------------------------------------*/ -void reportIncomplete(int id) -{ - factPrintf(kWarn, 601, "%5d skip incomplete evt %8d", - evtCtrl[id].evNum, id); +uint64_t reportIncomplete(int id, const char *txt) +{ + factPrintf(kWarn, 601, "skip incomplete evt (run=%d, evt=%d, %s)", + evtCtrl[id].runNum, evtCtrl[id].evNum, txt); uint64_t report = 0; @@ -660,5 +664,7 @@ } - // FIXME: Is that really 'b' or should that be 'ib' ? + // FIXME: This is not synchronous... it reports + // accoridng to the current connection status, not w.r.t. to the + // one when the event was taken. if (gi_NumConnect[ib]<=0) // board not connected { @@ -677,5 +683,5 @@ factOut(kWarn, 601, str); - factReportIncomplete(report); + return report; } @@ -808,5 +814,5 @@ // initialize mBuffer (mark all entries as unused\empty) - for (int i = 0; i < MAX_EVT * MAX_RUN; i++) + for (int i = 0; i < MAX_EVT; i++) { evtCtrl[i].evNum = evtCtrl[i].nRoi = -1; @@ -1201,6 +1207,33 @@ } + const int beg = (idx + MAX_EVT - 1) % MAX_EVT; + const int end = (evtCtrl_frstPtr + MAX_EVT - 1) % MAX_EVT; + + // we have just completed an event... so all previous events + // must have been completed already. If they are not, there + // is no need to wait for the timeout, because they will never + // get completed. We can just ensure that if we check for the previous + // event to be complete every time we receive a new complete event. + // If we find an incomplete one, we remove all consecutive + // incomplete ones. + for (int k=beg; k!=end; k=(k+MAX_EVT-1)%MAX_EVT) + { + // We are done if we find a complete or fully processed event + if (evtCtrl[k].evtStat>=100) + break; + + // we do not call factReportIncomplete here, because by starting a new + // run and having received the first complete event from that run + // the user has expressed that the old events are obsolste now + // and the run will be closed anyway + if (evtCtrl[k].evtStat>=0 && evtCtrl[k].evtStat<90) + { + reportIncomplete(k, "expired"); + evtCtrl[k].evtStat = 90; + } + } + // Flag that the event is ready for processing - evtCtrl[idx].evtStat = 99; + evtCtrl[idx].evtStat = 100; } // end for loop over all sockets @@ -1219,15 +1252,15 @@ //delete those that are written to disk .... - const int count = (evtCtrl_lastPtr-evtCtrl_frstPtr+MAX_EVT*MAX_RUN)%(MAX_EVT*MAX_RUN); - - for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT*MAX_RUN) + const int count = (evtCtrl_lastPtr-evtCtrl_frstPtr+MAX_EVT)%MAX_EVT; + + for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT) { // Check the more likely case first: incomplete events - if (evtCtrl[k0].evtStat>=0 && evtCtrl[k0].evtStat<92) + if (evtCtrl[k0].evtStat>=0 && evtCtrl[k0].evtStat<100) { gj.bufNew++; //incomplete event in Buffer // Event has not yet timed out or was reported already - if (evtCtrl[k0].evtStat>=90 || evtCtrl[k0].pcTime[0]/*evtCtrl[k0].lastRecv*/>=g_actTime - 30) + if (evtCtrl[k0].evtStat==90 || evtCtrl[k0].pcTime[0]/*evtCtrl[k0].lastRecv*/>=g_actTime - 30) continue; @@ -1235,33 +1268,30 @@ // It doesn't matter if that takes comparably long, // because we have to stop the run anyway. - reportIncomplete(k0); + const uint64_t rep = reportIncomplete(k0, "timeout"); + factReportIncomplete(rep); //timeout for incomplete events - evtCtrl[k0].evtStat = 91; + evtCtrl[k0].evtStat = 90; gj.evtSkip++; continue; } - - // complete event in Buffer - //if (evtCtrl[k0].evtStat >= 95) - // gj.bufEvt++; // Check the less likely case: 'useless' or 'delete' // evtState==0 can happen if the event was initialized (some data received) // but the data did not make sense (e.g. inconsistent rois) - if (evtCtrl[k0].evtStat==0 || evtCtrl[k0].evtStat >= 9000) + if (evtCtrl[k0].evtStat==0 || evtCtrl[k0].evtStat == 10000) { mBufFree(k0); //event written--> free memory evtCtrl[k0].evtStat = -1; - if (k0==evtCtrl_frstPtr) - evtCtrl_frstPtr = (evtCtrl_frstPtr+1) % (MAX_EVT*MAX_RUN); - else - factPrintf(kError, -1, "Freed a non-first slot"); - gj.evtWrite++; gj.rateWrite++; - + } + + // Remove leading invalidated slots from queue + if (evtCtrl[evtCtrl_frstPtr].evtStat==-1 && k0==evtCtrl_frstPtr) + { + evtCtrl_frstPtr = (evtCtrl_frstPtr+1) % MAX_EVT; continue; } @@ -1271,5 +1301,6 @@ // first complete events which processing is pending, we can stop. // All other events (if everything works well) must have the same state. - break; + // FIXME: This only works if we start from lastPtr and go down to frstPtr + //break; } @@ -1353,14 +1384,14 @@ // flag incomplete events as 'read finished' - if (evtCtrl[k0].evtStat>0 && evtCtrl[k0].evtStat < 90) - evtCtrl[k0].evtStat = 91; - - if (evtCtrl[k0].evtStat==0 || evtCtrl[k0].evtStat>900) + // We cannot just detele all events, because some might currently being processed, + // so we have to wait until the processing thread currently processing the event + // signals that the event can be deleted. (Note, that there are currently never + // two threads processing the same event at the same time) + if ((evtCtrl[k0].evtStat>0 && evtCtrl[k0].evtStat<90) || evtCtrl[k0].evtStat==10000) { mBufFree(k0); //event written--> free memory + evtCtrl_frstPtr = (evtCtrl_frstPtr+1) % MAX_EVT; evtCtrl[k0].evtStat = -1; - - evtCtrl_frstPtr = (evtCtrl_frstPtr+1) % (MAX_EVT*MAX_RUN); - } + } usleep(1); @@ -1410,7 +1441,6 @@ { int numWait = 0; - int numProc = 0; - - for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT*MAX_RUN) + + for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT) { if (evtCtrl[k0].evtStat != 1000 + threadID) @@ -1422,25 +1452,32 @@ } - int jret = 9100; // flag to be deleted (gi_resetR>1 : flush buffers asap) - - if (gi_resetR<=1) + /* + // gi_resetR>1 : flush buffers asap + if (gi_resetR>1) { - jret = subProcEvt(threadID, evtCtrl[k0].FADhead, - evtCtrl[k0].fEvent, NULL/*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[k].evtStat = 9000; // flag as 'to be deleted' + continue; + + }*/ + + const int jret = subProcEvt(threadID, evtCtrl[k0].FADhead, + evtCtrl[k0].fEvent, NULL/*mBuffer[id].buffer*/); + + if (jret>0 && jret<=threadID) + factPrintf(kError, -1, "Process %ld wants to send event to process %d... not allowed.", threadID, jret); + + if (jret<=threadID) + { + evtCtrl[k0].evtStat = 10000; // flag as 'to be deleted' + continue; } - evtCtrl[k0].evtStat = jret; - numProc++; + if (jret>=gi_maxProc) + { + evtCtrl[k0].evtStat = 5000; // flag as 'to be written' + continue; + } + + evtCtrl[k0].evtStat = 1000 + jret; // flag for next proces } @@ -1450,7 +1487,5 @@ } - //seems we have nothing to do, so sleep a little - if (numProc == 0) - usleep(1); + usleep(1); } @@ -1507,9 +1542,9 @@ int numProc = 0; - for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT*MAX_RUN) + for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT) { - if (evtCtrl[k0].evtStat <= 90 || evtCtrl[k0].evtStat >= 1000) + if (evtCtrl[k0].evtStat<90 || evtCtrl[k0].evtStat>=1000) { - if (evtCtrl[k0].evtStat >= 0 && evtCtrl[k0].evtStat< 90) + if (/*evtCtrl[k0].evtStat>=0 &&*/ evtCtrl[k0].evtStat<90) numWait++; @@ -1517,10 +1552,11 @@ } + /* //we are asked to flush buffers asap if (gi_resetR > 1) { - evtCtrl[k0].evtStat = 9991; + evtCtrl[k0].evtStat = 9000; continue; - } + }*/ //-------- it is better to open the run already here, so call can be used to initialize @@ -1611,5 +1647,5 @@ if (runCtrl[lastRun].procId != 0) { - evtCtrl[k0].evtStat = 9091; + evtCtrl[k0].evtStat = 10000; // flag 'to be deleted' continue; } @@ -1645,10 +1681,10 @@ if (rc < 0) { - evtCtrl[k0].evtStat = 9999; //flag event to be skipped + evtCtrl[k0].evtStat = 10000; // flag event to be deleted //gi.procErr++; } else { - evtCtrl[k0].evtStat = 1000; + evtCtrl[k0].evtStat = 1000; // flag 'start processing' runCtrl[lastRun].procEvt++; } @@ -1680,9 +1716,10 @@ } - for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT*MAX_RUN) + /* + for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT) { if (evtCtrl[k0].evtStat >= 0 && evtCtrl[k0].evtStat < 1000) - evtCtrl[k0].evtStat = 9800; //flag event as 'processed' - } + evtCtrl[k0].evtStat = 9000; //flag event as 'processed' + }*/ gp_runStat = -99; @@ -1800,9 +1837,11 @@ int numWait = 0; - for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT*MAX_RUN) + // Note that the current loop does not at all gurantee that + // the events are written in the correct order. + for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT) { - if (evtCtrl[k0].evtStat <= 5000 || evtCtrl[k0].evtStat >= 9000) + if (evtCtrl[k0].evtStat<5000 || evtCtrl[k0].evtStat >= 10000) { - if (evtCtrl[k0].evtStat > 0 && evtCtrl[k0].evtStat < 9000) + if (/*evtCtrl[k0].evtStat > 0 &&*/ evtCtrl[k0].evtStat < 5000) numWait++; @@ -1811,9 +1850,10 @@ //we must drain the buffer asap + /* if (gi_resetR > 1) { - evtCtrl[k0].evtStat = 9904; + evtCtrl[k0].evtStat = 9000; continue; - } + }*/ const uint32_t irun = evtCtrl[k0].runNum; @@ -1868,16 +1908,16 @@ } + /* if (runCtrl[lastRun].fileId > 0) { // There is an event but file is already closed - /* - if (runCtrl[j].fileId < 100) - { - factPrintf(kWarn, 123, "writeEvt: File for run %d is closed", irun); - runCtrl[j].fileId += 100; - }*/ - - evtCtrl[k0].evtStat = 9903; - } + //if (runCtrl[j].fileId < 100) + //{ + // factPrintf(kWarn, 123, "writeEvt: File for run %d is closed", irun); + // runCtrl[j].fileId += 100; + //} + + evtCtrl[k0].evtStat = 9000; + }*/ // File is open @@ -1891,15 +1931,12 @@ runCtrl[lastRun].lastTime = g_actTime; runCtrl[lastRun].actEvt++; - - evtCtrl[k0].evtStat = 9901; } else - { factPrintf(kError, 503, "writeEvt: Writing event for run %d failed (runWrite)", irun); - evtCtrl[k0].evtStat = 9902; - } checkAndCloseRun(lastRun, irun, rc<0, 1); } + + evtCtrl[k0].evtStat = 10000; // event written (or has to be discarded) -> delete } /* Index: trunk/FACT++/src/FAD.h =================================================================== --- trunk/FACT++/src/FAD.h (revision 15482) +++ trunk/FACT++/src/FAD.h (revision 15483) @@ -177,7 +177,4 @@ //--------------------------------------------------------------- - -#define MAX_RUN 8 -#define MAX_EVT 32768 //don't worry, for events is MAX_RUN*MAX_EVT typedef void* FileHandle_t ;