Index: /trunk/FACT++/src/EventBuilder.c =================================================================== --- /trunk/FACT++/src/EventBuilder.c (revision 11281) +++ /trunk/FACT++/src/EventBuilder.c (revision 11282) @@ -35,5 +35,5 @@ #define MIN_LEN 32 // min #bytes needed to interpret FADheader -#define MAX_LEN 64*1024 // size of read-buffer per socket +#define MAX_LEN 256*1024 // size of read-buffer per socket extern FileHandle_t runOpen(uint32_t irun, RUN_HEAD *runhd, size_t len ) ; @@ -41,19 +41,21 @@ extern int runClose(FileHandle_t fileHd , RUN_TAIL *runth, size_t len ) ; extern void factOut(int severity, int err, char* message ) ; -extern void factStat(int64_t *array , int len ) ; + +extern void factStat(GUI_STAT gj) ; + +extern void factStatNew(EVT_STAT gi) ; extern int eventCheck( PEVNT_HEADER *fadhd, EVENT *event) ; - -extern void debugHead(int i, void *buf); - -extern void debugRead(int isock, int ibyte, int32_t event,int32_t ftmevt, int32_t runnr, int state, - uint32_t tsec, uint32_t tusec ) ; +extern void debugHead(int i, int j, void *buf); + +extern void debugRead(int isock, int ibyte, int32_t event,int32_t ftmevt, + int32_t runnr, int state, uint32_t tsec, uint32_t tusec ) ; extern void debugStream(int isock, void *buf, int len) ; -int g_actTime = 0 ; -int g_runStat = 40 ; +int g_actTime ; +int g_runStat ; size_t g_maxMem ; //maximum memory allowed for buffer @@ -66,27 +68,26 @@ -int gi_runStat ; -int gp_runStat ; -int gw_runStat ; - - -uint gi_SecRate[MAX_SOCK] ; -uint gi_S10Rate[MAX_SOCK] ; -uint gi_MinRate[MAX_SOCK] ; -uint gi_ErrCnt[MAX_SOCK] ; + int gi_runStat ; + int gp_runStat ; + int gw_runStat ; + int32_t gi_myRun ; + + uint gi_NumConnect[NBOARDS]; //4 crates * 10 boards -uint gi_SecTime, gi_S10Time, gi_MinTime ; -uint gi_EvtStart= 0 ; -uint gi_EvtRead = 0 ; -uint gi_EvtBad = 0 ; -uint gi_EvtTot = 0 ; -size_t gi_usedMem = 0 ; - -uint gw_EvtTot = 0 ; -uint gp_EvtTot = 0 ; +//uint gi_EvtStart= 0 ; +//uint gi_EvtRead = 0 ; +//uint gi_EvtBad = 0 ; +//uint gi_EvtTot = 0 ; +//size_t gi_usedMem = 0 ; + +//uint gw_EvtTot = 0 ; +//uint gp_EvtTot = 0 ; PIX_MAP g_pixMap[NPIX] ; + +EVT_STAT gi ; +GUI_STAT gj ; EVT_CTRL evtCtrl ; //control of events during processing @@ -305,5 +306,5 @@ headmem = NBOARDS* sizeof(PEVNT_HEADER) ; - if ( gi_usedMem + needmem + headmem > g_maxMem) { + if ( gj.usdMem + needmem + headmem > g_maxMem) { snprintf(str,MXSTR,"no memory left to keep event %d",evID) ; factOut(kError,882, str ) ; @@ -349,5 +350,6 @@ mBuffer[i].evtLen = needmem ; - gi_usedMem += needmem + headmem; + gj.usdMem += needmem + headmem; + if (gj.usdMem > gj.maxMem ) gj.maxMem = gj.usdMem ; //register event in 'active list (reading)' @@ -357,4 +359,9 @@ evtCtrl.pcTime[ evtCtrl.lastPtr] = g_actTime ; evtIdx[i] = evtCtrl.lastPtr ; + + gj.readEvt++ ; //#events read + gj.evtBuf++ ; + + snprintf(str,MXSTR,"%5d start new evt %8d %8d %2d",evID,i,evtCtrl.lastPtr,0); factOut(kDebug,-11, str ) ; @@ -363,7 +370,5 @@ - - - gi_EvtStart++ ; + gi.evtGet++ ; //check if runId already registered in runCtrl @@ -379,4 +384,17 @@ } else { runCtrl[evFree].runId = runID ; + runCtrl[evFree].fileId = -2 ; + runCtrl[evFree].nextEvt= 0; + runCtrl[evFree].actEvt = 0; + runCtrl[evFree].maxEvt = 999999999 ; //max number events allowed + runCtrl[evFree].lastTime=g_actTime ; + runCtrl[evFree].closeTime=g_actTime + 3600*24 ; //max time allowed + runCtrl[evFree].lastTime = 0 ; + + runTail[evFree].nEventsOk = + runTail[evFree].nEventsRej = + runTail[evFree].nEventsBad = + runTail[evFree].PCtime0 = + runTail[evFree].PCtimeX = 0 ; } @@ -384,4 +402,6 @@ } /*-----------------------------------------------------------------*/ + + @@ -405,6 +425,6 @@ mBuffer[i].evNum = mBuffer[i].runNum = mBuffer[i].nRoi= -1; - gi_usedMem = gi_usedMem - freemem - headmem; - + gj.usdMem = gj.usdMem - freemem - headmem; + gj.evtBuf-- ; return 0 ; @@ -413,5 +433,36 @@ - /*-----------------------------------------------------------------*/ +void resetEvtStat() { + int i ; + + for (i=0; i=0) { //loop until global variable g_runStat claims stop - gi_runStat = g_runStat ; + gi_runStat = g_runStat; + gj.readStat= g_runStat; + gj.totMem = g_maxMem ; +// if (gi.totMem < 50000000 ) gi.totMem = 500000000 ; int b,p,p0,s0,nch; @@ -519,4 +574,5 @@ nch++ ; gi_NumConnect[ b ] = 0 ; //must close all connections + gi.numConn[ b ] = 0; if ( sockDef[b] == 0) s0= 0 ; //sockets to be defined and opened else if (g_port[b].sockDef == 0) s0=-1 ; //sockets to be destroyed @@ -542,8 +598,5 @@ - g_actTime = time(NULL) ; - nokCnt[numok]++; - - loopCnt++ ; + numok = 0 ; //count number of succesfull actions @@ -555,5 +608,5 @@ gettimeofday( tv, NULL); - tsec = atv.tv_sec ; + g_actTime = tsec = atv.tv_sec ; tusec= atv.tv_usec ; @@ -572,6 +625,6 @@ rd[i].skip = 0 ; // start empty gi_NumConnect[ b ]++ ; - numok++ ; //make sure next round will execute - snprintf(str,MXSTR,"+++connect %d %d",b,gi_NumConnect[ b ]); + gi.numConn[ b ]++ ; + snprintf(str,MXSTR,"+++connect %d %d",b,gi.numConn[ b ]); factOut(kInfo,-1, str ) ; } @@ -581,5 +634,4 @@ if (rd[i].bufLen > 0) { //might be nothing to read [buffer full] numok++ ; - sokCnt[i]++; jrd=recv(rd[i].socket,&rd[i].rBuf->B[ rd[i].bufPos], rd[i].bufLen, MSG_DONTWAIT); @@ -593,6 +645,7 @@ factOut(kInfo,441, str ) ; GenSock(s0, i, 0,NULL, &rd[i]) ; - gi_ErrCnt[i]++ ; + gi.gotErr[ b ]++ ; gi_NumConnect[ b ]-- ; + gi.numConn[ b ]-- ; } else if ( jrd<0 ) { //did not read anything @@ -600,9 +653,11 @@ snprintf(str,MXSTR,"Error Reading from %d | %m",i); factOut(kError,442, str ) ; - gi_ErrCnt[i]++ ; + gi.gotErr[ b ]++ ; } else numok-- ; //else nothing waiting to be read jrd = 0 ; } } else jrd = 0 ; //did read nothing as requested + + gi.gotByte[ b ] += jrd ; if ( rd[i].bufTyp <0 ) { // we are skipping this board ... @@ -619,5 +674,5 @@ if ( rd[i].rBuf->B[ rd[i].bufPos-1] != stop.B[0] && rd[i].rBuf->B[ rd[i].bufPos ] != stop.B[1]) { - gi_ErrCnt[i]++ ; + gi.evtErr++ ; snprintf(str,MXSTR,"wrong end of buffer found %d",rd[i].bufPos); factOut(kError,301, str ) ; @@ -628,6 +683,4 @@ //we have a complete buffer, copy to WORK area - gi_SecRate[i]++ ; - roi = ntohs(rd[i].rBuf->S[ head_len/2 + 2 ]) ; //get index into mBuffer for this event (create if needed) @@ -651,7 +704,4 @@ snprintf(str,MXSTR,"wrong Board ID %d %d %d",fadCrate,fadBoard%256,boardId) ; if (errcnt0++ < 99 ) factOut(kWarn,301, str ) ; //print only few times -// } else { -// snprintf(str,MXSTR,"correct Board ID %d %d %d",fadCrate,fadBoard%256,boardId) ; -// if (errcnt0++ < 99 ) factOut(kWarn,301, str ) ; //print only few times } if ( mBuffer[evID].board[ boardId ] != -1) { @@ -703,4 +753,5 @@ } else { snprintf(str,MXSTR,"wrong roi %d %d %d %d",px,pixR,roi,src-2); + gi.evtErr++ ; factOut(kError,202, str ) ; goto EndBuf ; @@ -708,4 +759,5 @@ } }// now we have stored a new board contents into Event structure + mBuffer[evID].board[ boardId ] = boardId ; evtCtrl.evtStat[ iDx ]++ ; @@ -717,6 +769,5 @@ //complete event read ---> flag for next processing evtCtrl.evtStat[ iDx ] = 99; - gi_EvtRead++ ; - gi_EvtTot++ ; + gi.evtTot++ ; } @@ -763,9 +814,10 @@ rd[i].ftmID = ntohl(rd[i].rBuf->I[5]) ; //(FTMevt) rd[i].runID = ntohl(rd[i].rBuf->I[11]) ; - if (rd[i].runID ==0 ) rd[i].runID = myRun ; + if (rd[i].runID ==0 ) rd[i].runID = gi_myRun ; rd[i].bufTyp = 1 ; //ready to read full record rd[i].bufLen = rd[i].fadLen - rd[i].bufPos ; if (rd[i].bufLen <=0 ) rd[i].bufLen = 100000 ; //? - debugHead(i,rd[i].rBuf); + int fadBoard = ntohs(rd[i].rBuf->S[12] ) ; + debugHead(i,fadBoard,rd[i].rBuf); debugRead(i,jrd,rd[i].evtID,rd[i].ftmID,rd[i].runID,-1,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid;-1=start event } else { @@ -780,4 +832,6 @@ } //finished trying to read all sockets + gi.numRead[ numok ] ++ ; + int qwait=0, qdel=0, qskip=0 ; g_actTime = time(NULL) ; @@ -806,6 +860,7 @@ factOut(kWarn,601, str ) ; evtCtrl.evtStat[k0] = 91 ; //timeout for incomplete events - gi_EvtBad++ ; - gi_EvtTot++ ; + gi.evtSkp++ ; + gi.evtTot++ ; + gj.skipEvt++; qskip++; } @@ -831,15 +886,19 @@ int ib ; for (ib=0; ib0) - gi_NumConnect[ i/7 ]-- ; + if (rd[i].sockStat ==0 ) { + GenSock(-1, i, 0, NULL, &rd[i]) ; //close and destroy socket + gi_NumConnect[ i/7 ]-- ; + gi.numConn[ i/7 ]-- ; + } } @@ -894,4 +955,5 @@ nanosleep( &xwait , NULL ) ; gi_runStat = -11 ; //inform all that no update to happen any more + gj.readStat= -11 ; //inform all that no update to happen any more @@ -930,4 +992,7 @@ factOut(kInfo,-1, str ) ; gi_runStat = -99 ; + gj.readStat= -99 ; + factStat(gj); + factStatNew(gi) ; return 0; @@ -1009,10 +1074,13 @@ int i=eventCheck(mBuffer[id].FADhead,mBuffer[id].fEvent) ; - - if (i<0) evtCtrl.evtStat[k0] = 999 ; //flag event to be skipped - + gj.procEvt++ ; + gi.procTot++ ; numProc++ ; evtCtrl.evtStat[k0] = 520 ; - gp_EvtTot++ ; + + if (i<0) { + evtCtrl.evtStat[k0] = 999 ; //flag event to be skipped + gi.procErr++ ; + } } else if ( evtCtrl.evtStat[k0] >=0 && evtCtrl.evtStat[k0] < 90 ) { numWait++ ; @@ -1020,8 +1088,9 @@ } - if ( gi_runStat < -10 && numWait == 0) { //nothing left to do + if ( gj.readStat < -10 && numWait == 0) { //nothing left to do snprintf(str,MXSTR,"Exit Processing Process ..."); factOut(kInfo,-1, str ) ; gp_runStat = -22 ; //==> we should exit + gj.procStat= -22 ; //==> we should exit return 0 ; } @@ -1034,4 +1103,5 @@ } gp_runStat = gi_runStat ; + gj.procStat= gj.readStat ; } @@ -1054,4 +1124,5 @@ gp_runStat = -99 ; + gj.procStat= -99 ; return 0; @@ -1059,46 +1130,29 @@ } /*-----------------------------------------------------------------*/ -int CloseRunFile(uint32_t runId, uint32_t closeTime) { +int CloseRunFile(uint32_t runId, uint32_t closeTime, uint32_t maxEvt) { /* close run runId (all all runs if runId=0) */ /* return: 0=close scheduled / >0 already closed / <0 does not exist */ - int i, j ; - - if (runId == 0 ) { + int j ; + + + if ( runId == 0 ) { for ( j=0; j 0 ) { + if (runCtrl[j].fileId != 0 ) { snprintf(str,MXSTR,"W no open file for this run %d",irun) ; - factOut(kDebug,123,str) ; + factOut(kWarn,123,str) ; evtCtrl.evtStat[k0] = 902 ; + gi.wrtErr++ ; } else { - int i=runWrite(runCtrl[j].fileHd, mBuffer[id].fEvent, sizeof(mBuffer[id]) ); - if (i<0) { + i=runWrite(runCtrl[j].fileHd, mBuffer[id].fEvent, sizeof(mBuffer[id]) ); + if ( i>=0 ) { + runCtrl[j].lastTime = g_actTime; + runCtrl[j].actEvt++ ; + evtCtrl.evtStat[k0] = 901 ; + snprintf(str,MXSTR,"%5d successfully wrote for run %d id %5d",ievt,irun,k0); + factOut(kDebug,504, str ) ; + gj.writEvt++ ; + } else { snprintf(str,MXSTR,"W error writing event for run %d",irun) ; factOut(kError,503, str ) ; evtCtrl.evtStat[k0] = 901 ; - //close run + gi.wrtErr++ ; + } + + if ( i < 0 + || runCtrl[j].closeTime < g_actTime + || runCtrl[j].lastTime < g_actTime-300 + || runCtrl[j].maxEvt < runCtrl[j].actEvt ) { +int ii =0 ; +if ( i < 0 ) ii=1 ; +else if (runCtrl[j].closeTime < g_actTime ) ii=2 ; +else if (runCtrl[j].lastTime < g_actTime-300 ) ii=3 ; +else if (runCtrl[j].maxEvt < runCtrl[j].actEvt ) ii=4 ; + + + + //close run for whatever reason + if (runCtrl[j].runId == gi_myRun) gi_myRun = g_actTime ; i=runClose(runCtrl[j].fileHd, &runTail[j], sizeof(runTail[j]) ); if (i<0) { - snprintf(str,MXSTR,"W error closing run %d",irun) ; + snprintf(str,MXSTR,"error closing run %d %d AAA",runCtrl[j].runId,i) ; factOut(kError,503, str ) ; + runCtrl[j].fileId = 9001 ; } else { - snprintf(str,MXSTR,"W closed run %d because of write error",irun) ; + snprintf(str,MXSTR,"W closed run %d AAA %d",irun,ii) ; factOut(kInfo,503, str ) ; + runCtrl[j].fileId = 901 ; } - runCtrl[j].fileId = 9999 ; - } else { - runCtrl[j].lastTime = g_actTime; - evtCtrl.evtStat[k0] = 901 ; - snprintf(str,MXSTR,"%5d successfully wrote for run %d id %5d",ievt,irun,k0); - factOut(kDebug,504, str ) ; } } @@ -1221,18 +1296,27 @@ } - //check if we should close a run ... + //check if we should close a run (mainly when no event pending) for ( j=0; j we should exit + gj.writStat= -22 ; //==> we should exit goto closerun ; } gw_runStat = gi_runStat ; + gj.writStat= gj.readStat ; } @@ -1264,17 +1350,24 @@ for ( j=0; j