Index: trunk/FACT++/src/EventBuilder.c =================================================================== --- trunk/FACT++/src/EventBuilder.c (revision 11765) +++ trunk/FACT++/src/EventBuilder.c (revision 11766) @@ -13,5 +13,5 @@ #include #include -#include +#include #include #include @@ -24,74 +24,79 @@ enum Severity { - kMessage = 10, ///< Just a message, usually obsolete - kInfo = 20, ///< An info telling something which can be interesting to know - kWarn = 30, ///< A warning, things that somehow might result in unexpected or unwanted bahaviour - kError = 40, ///< Error, something unexpected happened, but can still be handled by the program - kFatal = 50, ///< An error which cannot be handled at all happend, the only solution is program termination - kDebug = 99, ///< A message used for debugging only + kMessage = 10, ///< Just a message, usually obsolete + kInfo = 20, ///< An info telling something which can be interesting to know + kWarn = 30, ///< A warning, things that somehow might result in unexpected or unwanted bahaviour + kError = 40, ///< Error, something unexpected happened, but can still be handled by the program + kFatal = 50, ///< An error which cannot be handled at all happend, the only solution is program termination + kDebug = 99, ///< A message used for debugging only }; -#define MIN_LEN 32 // min #bytes needed to interpret FADheader -#define MAX_LEN 256*1024 // size of read-buffer per socket +#define MIN_LEN 32 // min #bytes needed to interpret FADheader +#define MAX_LEN 256*1024 // size of read-buffer per socket //#define nanosleep(x,y) -extern FileHandle_t runOpen(uint32_t irun, RUN_HEAD *runhd, size_t len ) ; -extern int runWrite(FileHandle_t fileHd , EVENT *event, size_t len ) ; -extern int runClose(FileHandle_t fileHd , RUN_TAIL *runth, size_t len ) ; -extern void factOut(int severity, int err, char* message ) ; -extern void gotNewRun( int runnr, PEVNT_HEADER *headers ); - - -extern void factStat(GUI_STAT gj) ; - -extern void factStatNew(EVT_STAT gi) ; - -extern int eventCheck( uint32_t runNr, PEVNT_HEADER *fadhd, EVENT *event) ; - -extern int subProcEvt(int threadID, PEVNT_HEADER *fadhd, EVENT *event, int8_t *buffer) ; - -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 CloseRunFile(uint32_t runId, uint32_t closeTime, uint32_t maxEvt); - - -int g_maxProc ; -int g_maxSize ; -int gi_maxSize ; -int gi_maxProc ; - -uint g_actTime ; -uint g_actUsec ; -int g_runStat ; -int g_reset ; -int g_useFTM ; - -int gi_reset, gi_resetR, gi_resetS, gi_resetW, gi_resetX ; -size_t g_maxMem ; //maximum memory allowed for buffer +extern FileHandle_t runOpen (uint32_t irun, RUN_HEAD * runhd, size_t len); +extern int runWrite (FileHandle_t fileHd, EVENT * event, size_t len); +extern int runClose (FileHandle_t fileHd, RUN_TAIL * runth, size_t len); +//extern int runFinish (uint32_t runnr); + +extern void factOut (int severity, int err, char *message); + +extern void gotNewRun (int runnr, PEVNT_HEADER * headers); + + +extern void factStat (GUI_STAT gj); + +extern void factStatNew (EVT_STAT gi); + +extern int eventCheck (uint32_t runNr, PEVNT_HEADER * fadhd, EVENT * event); + +extern int subProcEvt (int threadID, PEVNT_HEADER * fadhd, EVENT * event, + int8_t * buffer); + +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 CloseRunFile (uint32_t runId, uint32_t closeTime, uint32_t maxEvt); + + +int g_maxProc; +int g_maxSize; +int gi_maxSize; +int gi_maxProc; + +uint g_actTime; +uint g_actUsec; +int g_runStat; +int g_reset; +int g_useFTM; + +int gi_reset, gi_resetR, gi_resetS, gi_resetW, gi_resetX; +size_t g_maxMem; //maximum memory allowed for buffer //no longer needed ... - int g_maxBoards ; //maximum number of boards to be initialized - int g_actBoards ; +int g_maxBoards; //maximum number of boards to be initialized +int g_actBoards; // -FACT_SOCK g_port[NBOARDS] ; // .addr=string of IP-addr in dotted-decimal "ddd.ddd.ddd.ddd" - - - int gi_runStat ; - int gp_runStat ; - int gw_runStat ; - - int gi_memStat = +1 ; - - uint32_t gi_myRun = 0 ; - uint32_t actrun = 0 ; - - -uint gi_NumConnect[NBOARDS]; //4 crates * 10 boards +FACT_SOCK g_port[NBOARDS]; // .addr=string of IP-addr in dotted-decimal "ddd.ddd.ddd.ddd" + + +int gi_runStat; +int gp_runStat; +int gw_runStat; + +int gi_memStat = +1; + +uint32_t gi_myRun = 0; +uint32_t actrun = 0; + + +uint gi_NumConnect[NBOARDS]; //4 crates * 10 boards //uint gi_EvtStart= 0 ; @@ -104,22 +109,22 @@ //uint gp_EvtTot = 0 ; -PIX_MAP g_pixMap[NPIX] ; - -EVT_STAT gi ; -GUI_STAT gj ; - -EVT_CTRL evtCtrl ; //control of events during processing -int evtIdx[MAX_EVT*MAX_RUN] ; //index from mBuffer to evtCtrl - -WRK_DATA mBuffer[MAX_EVT*MAX_RUN]; //local working space - - - - -RUN_HEAD actRun ; - -RUN_CTRL runCtrl[MAX_RUN] ; - -RUN_TAIL runTail[MAX_RUN] ; +PIX_MAP g_pixMap[NPIX]; + +EVT_STAT gi; +GUI_STAT gj; + +EVT_CTRL evtCtrl; //control of events during processing +int evtIdx[MAX_EVT * MAX_RUN]; //index from mBuffer to evtCtrl + +WRK_DATA mBuffer[MAX_EVT * MAX_RUN]; //local working space + + + + +RUN_HEAD actRun; + +RUN_CTRL runCtrl[MAX_RUN]; + +RUN_TAIL runTail[MAX_RUN]; @@ -129,54 +134,57 @@ -typedef union { - int8_t B[MAX_LEN ]; - int16_t S[MAX_LEN/2]; - int32_t I[MAX_LEN/4]; - int64_t L[MAX_LEN/8]; -} CNV_FACT ; - -typedef struct { - int bufTyp ; //what are we reading at the moment: 0=header 1=data -1=skip ... - int32_t bufPos ; //next byte to read to the buffer next - int32_t bufLen ; //number of bytes left to read - int32_t skip ; //number of bytes skipped before start of event - - int sockStat ; //-1 if socket not yet connected , 99 if not exist - int socket ; //contains the sockets - struct sockaddr_in SockAddr ; //IP for each socket - - int evtID ; // event ID of event currently read - int runID ; // run " - int ftmID ; // event ID from FTM - uint fadLen ; // FADlength of event currently read - int fadVers ; // Version of FAD - int ftmTyp ; // trigger type - int board ; // boardID (softwareID: 0..40 ) - int Port ; - - CNV_FACT *rBuf ; +typedef union +{ + int8_t B[MAX_LEN]; + int16_t S[MAX_LEN / 2]; + int32_t I[MAX_LEN / 4]; + int64_t L[MAX_LEN / 8]; +} CNV_FACT; + +typedef struct +{ + int bufTyp; //what are we reading at the moment: 0=header 1=data -1=skip ... + int32_t bufPos; //next byte to read to the buffer next + int32_t bufLen; //number of bytes left to read + int32_t skip; //number of bytes skipped before start of event + + int sockStat; //-1 if socket not yet connected , 99 if not exist + int socket; //contains the sockets + struct sockaddr_in SockAddr; //IP for each socket + + int evtID; // event ID of event currently read + int runID; // run " + int ftmID; // event ID from FTM + uint fadLen; // FADlength of event currently read + int fadVers; // Version of FAD + int ftmTyp; // trigger type + int board; // boardID (softwareID: 0..40 ) + int Port; + + CNV_FACT *rBuf; #ifdef EVTDEBUG - CNV_FACT *xBuf ; //a copy of rBuf (temporary for debuging) + CNV_FACT *xBuf; //a copy of rBuf (temporary for debuging) #endif -} READ_STRUCT ; - - -typedef union { - int8_t B[2]; - int16_t S ; -} SHORT_BYTE ; - - - - - -#define MXSTR 1000 -char str[MXSTR] ; - -SHORT_BYTE start, stop; - -READ_STRUCT rd[MAX_SOCK] ; //buffer to read IP and afterwards store in mBuffer +} READ_STRUCT; + + +typedef union +{ + int8_t B[2]; + int16_t S; +} SHORT_BYTE; + + + + + +#define MXSTR 1000 +char str[MXSTR]; + +SHORT_BYTE start, stop; + +READ_STRUCT rd[MAX_SOCK]; //buffer to read IP and afterwards store in mBuffer @@ -187,7 +195,18 @@ /*-----------------------------------------------------------------*/ - - -int GenSock(int flag, int sid, int port, struct sockaddr_in *sockAddr, READ_STRUCT *rd) { +int +runFinish1 (uint32_t runnr) +{ + snprintf (str, MXSTR, "Should finish run %d (but not yet possible)", + runnr); + factOut (kInfo, 173, str); //but continue anyhow + return 0; +} + + +int +GenSock (int flag, int sid, int port, struct sockaddr_in *sockAddr, + READ_STRUCT * rd) +{ /* *** generate Address, create sockets and allocates readbuffer for it @@ -200,83 +219,83 @@ */ - int j ; - int optval = 1 ; //activate keepalive - socklen_t optlen = sizeof(optval); - - if (rd->sockStat ==0 ) { //close socket if open - j=close(rd->socket) ; - if (j>0) { - snprintf(str,MXSTR,"Error closing socket %d | %m",sid); - factOut(kFatal,771, str ) ; - } else { - snprintf(str,MXSTR,"Succesfully closed socket %d",sid); - factOut(kInfo,771, str ) ; - } - } - - - if (flag < 0) { - free(rd->rBuf) ; //and never open again + int j; + int optval = 1; //activate keepalive + socklen_t optlen = sizeof (optval); + + if (rd->sockStat == 0) { //close socket if open + j = close (rd->socket); + if (j > 0) { + snprintf (str, MXSTR, "Error closing socket %d | %m", sid); + factOut (kFatal, 771, str); + } else { + snprintf (str, MXSTR, "Succesfully closed socket %d", sid); + factOut (kInfo, 771, str); + } + } + + + if (flag < 0) { + free (rd->rBuf); //and never open again #ifdef EVTDEBUG - free(rd->xBuf) ; //and never open again + free (rd->xBuf); //and never open again #endif - rd->rBuf = NULL ; - rd->sockStat = 99 ; - return 0 ; - } - - - if (flag == 0) { //generate address and buffer ... - rd->Port = port ; - rd->SockAddr.sin_family = sockAddr->sin_family; - rd->SockAddr.sin_port = htons(port) ; - rd->SockAddr.sin_addr = sockAddr->sin_addr ; + rd->rBuf = NULL; + rd->sockStat = 99; + return 0; + } + + + if (flag == 0) { //generate address and buffer ... + rd->Port = port; + rd->SockAddr.sin_family = sockAddr->sin_family; + rd->SockAddr.sin_port = htons (port); + rd->SockAddr.sin_addr = sockAddr->sin_addr; #ifdef EVTDEBUG - rd->xBuf = malloc(sizeof(CNV_FACT) ) ; + rd->xBuf = malloc (sizeof (CNV_FACT)); #endif - rd->rBuf = malloc(sizeof(CNV_FACT) ) ; - if ( rd->rBuf == NULL ) { - snprintf(str,MXSTR,"Could not create local buffer %d",sid); - factOut(kFatal,774, str ) ; - rd->sockStat = 77 ; - return -3 ; - } - } - - - if ( (rd->socket = socket (PF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0)) <= 0) { - snprintf(str,MXSTR,"Could not generate socket %d | %m",sid); - factOut(kFatal,773, str ) ; - rd->sockStat = 88 ; - return -2 ; - } - optval=1; - if ( setsockopt(rd->socket, SOL_SOCKET, SO_KEEPALIVE, &optval, optlen) < 0) { - snprintf(str,MXSTR,"Could not set keepalive %d | %m",sid); - factOut(kInfo,173, str ) ; //but continue anyhow - } - optval=10; //start after 10 seconds - if ( setsockopt(rd->socket, SOL_TCP, TCP_KEEPIDLE, &optval, optlen) < 0) { - snprintf(str,MXSTR,"Could not set keepidle %d | %m",sid); - factOut(kInfo,173, str ) ; //but continue anyhow - } - optval=10; //do every 10 seconds - if ( setsockopt(rd->socket, SOL_TCP, TCP_KEEPINTVL, &optval, optlen) < 0) { - snprintf(str,MXSTR,"Could not set keepintvl %d | %m",sid); - factOut(kInfo,173, str ) ; //but continue anyhow - } - optval=2; //close after 2 unsuccessful tries - if ( setsockopt(rd->socket, SOL_TCP, TCP_KEEPCNT, &optval, optlen) < 0) { - snprintf(str,MXSTR,"Could not set keepalive probes %d | %m",sid); - factOut(kInfo,173, str ) ; //but continue anyhow - } - - - - snprintf(str,MXSTR,"Successfully generated socket %d ",sid); - factOut(kInfo,773, str ) ; - rd->sockStat = -1 ; //try to (re)open socket - return 0 ; + rd->rBuf = malloc (sizeof (CNV_FACT)); + if (rd->rBuf == NULL) { + snprintf (str, MXSTR, "Could not create local buffer %d", sid); + factOut (kFatal, 774, str); + rd->sockStat = 77; + return -3; + } + } + + + if ((rd->socket = socket (PF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0)) <= 0) { + snprintf (str, MXSTR, "Could not generate socket %d | %m", sid); + factOut (kFatal, 773, str); + rd->sockStat = 88; + return -2; + } + optval = 1; + if (setsockopt (rd->socket, SOL_SOCKET, SO_KEEPALIVE, &optval, optlen) < 0) { + snprintf (str, MXSTR, "Could not set keepalive %d | %m", sid); + factOut (kInfo, 173, str); //but continue anyhow + } + optval = 10; //start after 10 seconds + if (setsockopt (rd->socket, SOL_TCP, TCP_KEEPIDLE, &optval, optlen) < 0) { + snprintf (str, MXSTR, "Could not set keepidle %d | %m", sid); + factOut (kInfo, 173, str); //but continue anyhow + } + optval = 10; //do every 10 seconds + if (setsockopt (rd->socket, SOL_TCP, TCP_KEEPINTVL, &optval, optlen) < 0) { + snprintf (str, MXSTR, "Could not set keepintvl %d | %m", sid); + factOut (kInfo, 173, str); //but continue anyhow + } + optval = 2; //close after 2 unsuccessful tries + if (setsockopt (rd->socket, SOL_TCP, TCP_KEEPCNT, &optval, optlen) < 0) { + snprintf (str, MXSTR, "Could not set keepalive probes %d | %m", sid); + factOut (kInfo, 173, str); //but continue anyhow + } + + + + snprintf (str, MXSTR, "Successfully generated socket %d ", sid); + factOut (kInfo, 773, str); + rd->sockStat = -1; //try to (re)open socket + return 0; } /*-----------------------------------------------------------------*/ @@ -287,28 +306,30 @@ -int mBufInit() { +int +mBufInit () +{ // initialize mBuffer (mark all entries as unused\empty) - int i ; - uint32_t actime ; - - actime = g_actTime + 50000000 ; - - for (i=0; i 1024) { - snprintf(str,MXSTR,"illegal nRoi[0]: %d",nRoi[0]) ; - factOut(kError, 999, str ) ; - gj.badRoiR++ ; - gj.badRoi[b]++ ; - return -9999 ; - } - - for (jr=1; jr<8; jr++) { - if ( nRoi[jr] != nRoi[0] ) { - snprintf(str,MXSTR,"wrong nRoi[%d]: %d %d",jr,nRoi[jr],nRoi[0]) ; - factOut(kError,711, str ) ; - gj.badRoiB++ ; - gj.badRoi[b]++ ; - return -7101 ; - } - } - if ( nRoi[8] < nRoi[0] ) { - snprintf(str,MXSTR,"wrong nRoi_TM: %d %d",nRoi[8],nRoi[0]) ; - factOut(kError,712, str ) ; - gj.badRoiB++ ; - gj.badRoi[b]++ ; - return -7102 ; - } - - - i = evID % MAX_EVT ; - evFree = -1 ; - - for ( k=0; k 1024) { + snprintf (str, MXSTR, "illegal nRoi[0]: %d", nRoi[0]); + factOut (kError, 999, str); + gj.badRoiR++; + gj.badRoi[b]++; + return -9999; + } + + for (jr = 1; jr < 8; jr++) { + if (nRoi[jr] != nRoi[0]) { + snprintf (str, MXSTR, "wrong nRoi[%d]: %d %d", jr, nRoi[jr], + nRoi[0]); + factOut (kError, 711, str); + gj.badRoiB++; + gj.badRoi[b]++; + return -7101; + } + } + if (nRoi[8] < nRoi[0]) { + snprintf (str, MXSTR, "wrong nRoi_TM: %d %d", nRoi[8], nRoi[0]); + factOut (kError, 712, str); + gj.badRoiB++; + gj.badRoi[b]++; + return -7102; + } + + + i = evID % MAX_EVT; + evFree = -1; + + for (k = 0; k < MAX_RUN; k++) { + if (mBuffer[i].evNum == evID && mBuffer[i].runNum == runID) { //event is already registered; + // is it ok ???? + if (mBuffer[i].nRoi != nRoi[0] + || mBuffer[i].nRoiTM != nRoi[8]) { + snprintf (str, MXSTR, "illegal evt_roi %d %d ; %d %d", + nRoi[0], nRoi[8], mBuffer[i].nRoi, mBuffer[i].nRoiTM); + factOut (kError, 821, str); + gj.badRoiE++; + gj.badRoi[b]++; + return -8201; } - // count for inconsistencies - if ( mBuffer[i].trgNum != trgNum ) mBuffer[i].Errors[0]++ ; - if ( mBuffer[i].fadNum != fadNum ) mBuffer[i].Errors[1]++ ; - if ( mBuffer[i].trgTyp != trgTyp ) mBuffer[i].Errors[2]++ ; + if (mBuffer[i].trgNum != trgNum) + mBuffer[i].Errors[0]++; + if (mBuffer[i].fadNum != fadNum) + mBuffer[i].Errors[1]++; + if (mBuffer[i].trgTyp != trgTyp) + mBuffer[i].Errors[2]++; //everything seems fine so far ==> use this slot .... - return i ; - } - if ( evFree < 0 && mBuffer[i].evNum < 0 ) evFree = i ; - i += MAX_EVT ; + return i; + } + if (evFree < 0 && mBuffer[i].evNum < 0) + evFree = i; + i += MAX_EVT; } //event does not yet exist; create it - if (evFree < 0 ) { //no space available in ctrl - snprintf(str,MXSTR,"no control slot to keep event %d",evID) ; - factOut(kError,881, str ) ; - return -1 ; - } - i = evFree ; //found free entry; use it ... - - gettimeofday( tv, NULL); - tsec = atv.tv_sec ; - tusec= atv.tv_usec ; + if (evFree < 0) { //no space available in ctrl + snprintf (str, MXSTR, "no control slot to keep event %d", evID); + factOut (kError, 881, str); + return -1; + } + i = evFree; //found free entry; use it ... + + gettimeofday (tv, NULL); + tsec = atv.tv_sec; + tusec = atv.tv_usec; //check if runId already registered in runCtrl - evFree = -1 ; - for (k=0; k 0) { //run is closed -> reject + snprintf (str, MXSTR, "skip event since run %d finished", runID); + factOut (kInfo, 931, str); + return -21; } - goto RUNFOUND ; - } - else if (evFree < 0 && runCtrl[k].runId == 0 ) evFree = k ; - } - - if (evFree <0 ) { - snprintf(str,MXSTR,"not able to register the new run %d",runID); - factOut(kFatal,883, str ) ; - return -1001 ; + + if (runCtrl[k].roi0 != nRoi[0] + || runCtrl[k].roi8 != nRoi[8]) { + snprintf (str, MXSTR, "illegal run_roi %d %d ; %d %d", + nRoi[0], nRoi[8], runCtrl[k].roi0, runCtrl[k].roi8); + factOut (kError, 931, str); + gj.badRoiR++; + gj.badRoi[b]++; + return -9301; + } + goto RUNFOUND; + } else if (evFree < 0 && runCtrl[k].fileId < 0) { //not yet used + evFree = k; + } else if (evFree < 0 && runCtrl[k].fileId > 0) { //already closed + if (runCtrl[k].closeTime < oldest) { + oldest = runCtrl[k].closeTime; + jold = k; + } + } + } + + if (evFree < 0 && jold < 0) { + snprintf (str, MXSTR, "not able to register the new run %d", runID); + factOut (kFatal, 883, str); + return -1001; } else { - snprintf(str,MXSTR,"register new run %d roi: %d %d",runID,nRoi[0],nRoi[8]) ; - factOut(kInfo,503, str ) ; - runCtrl[evFree].runId = runID ; - runCtrl[evFree].roi0 = nRoi[0] ; - runCtrl[evFree].roi8 = nRoi[8] ; - runCtrl[evFree].fileId = -2 ; - runCtrl[evFree].lastEvt= -1 ; - runCtrl[evFree].nextEvt= 0; + if (evFree < 0) + evFree = jold; + snprintf (str, MXSTR, "register new run %d(%d) roi: %d %d", runID, + evFree, nRoi[0], nRoi[8]); + factOut (kInfo, 503, str); + runCtrl[evFree].runId = runID; + runCtrl[evFree].roi0 = nRoi[0]; + runCtrl[evFree].roi8 = nRoi[8]; + runCtrl[evFree].fileId = -2; + runCtrl[evFree].procId = -2; + runCtrl[evFree].lastEvt = -1; + runCtrl[evFree].nextEvt = 0; runCtrl[evFree].actEvt = 0; - runCtrl[evFree].maxEvt = 999999999 ; //max number events allowed - runCtrl[evFree].firstUsec=tusec ; - runCtrl[evFree].firstTime= - runCtrl[evFree].lastTime=tsec ; - runCtrl[evFree].closeTime=tsec + 3600*24 ; //max time allowed - runCtrl[evFree].lastTime = 0 ; + runCtrl[evFree].procEvt = 0; + runCtrl[evFree].maxEvt = 999999999; //max number events allowed + runCtrl[evFree].firstUsec = tusec; + runCtrl[evFree].firstTime = runCtrl[evFree].lastTime = tsec; + runCtrl[evFree].closeTime = tsec + 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 ; - } - -RUNFOUND: - - needmem = sizeof(EVENT) + NPIX*nRoi[0]*2 + NTMARK*nRoi[0]*2; // - - headmem = NBOARDS* sizeof(PEVNT_HEADER) ; - - if ( gj.usdMem + needmem + headmem + gi_maxSize > g_maxMem) { - gj.maxMem = gj.usdMem + needmem + headmem + gi_maxSize ; - if (gi_memStat>0 ) { - gi_memStat = -99 ; - snprintf(str,MXSTR,"no memory left to keep event %6d sock %3d",evID,sk) ; - factOut(kError,882, str ) ; - } else { - snprintf(str,MXSTR,"no memory left to keep event %6d sock %3d",evID,sk) ; - factOut(kDebug,882, str ) ; - } - return -11 ; - } - - mBuffer[i].FADhead = malloc( headmem ) ; + runTail[evFree].nEventsRej = + runTail[evFree].nEventsBad = + runTail[evFree].PCtime0 = runTail[evFree].PCtimeX = 0; + } + + RUNFOUND: + + needmem = sizeof (EVENT) + NPIX * nRoi[0] * 2 + NTMARK * nRoi[0] * 2; // + + headmem = NBOARDS * sizeof (PEVNT_HEADER); + + if (gj.usdMem + needmem + headmem + gi_maxSize > g_maxMem) { + gj.maxMem = gj.usdMem + needmem + headmem + gi_maxSize; + if (gi_memStat > 0) { + gi_memStat = -99; + snprintf (str, MXSTR, "no memory left to keep event %6d sock %3d", + evID, sk); + factOut (kError, 882, str); + } else { + snprintf (str, MXSTR, "no memory left to keep event %6d sock %3d", + evID, sk); + factOut (kDebug, 882, str); + } + return -11; + } + + mBuffer[i].FADhead = malloc (headmem); if (mBuffer[i].FADhead == NULL) { - snprintf(str,MXSTR,"malloc header failed for event %d",evID) ; - factOut(kError,882, str ) ; + snprintf (str, MXSTR, "malloc header failed for event %d", evID); + factOut (kError, 882, str); return -12; } - mBuffer[i].fEvent = malloc( needmem ) ; - if (mBuffer[i].fEvent == NULL) { - snprintf(str,MXSTR,"malloc data failed for event %d",evID) ; - factOut(kError,882, str ) ; - free(mBuffer[i].FADhead) ; - mBuffer[i].FADhead = NULL ; + mBuffer[i].fEvent = malloc (needmem); + if (mBuffer[i].fEvent == NULL) { + snprintf (str, MXSTR, "malloc data failed for event %d", evID); + factOut (kError, 882, str); + free (mBuffer[i].FADhead); + mBuffer[i].FADhead = NULL; return -22; } - mBuffer[i].buffer = malloc( gi_maxSize ) ; + mBuffer[i].buffer = malloc (gi_maxSize); if (mBuffer[i].buffer == NULL) { - snprintf(str,MXSTR,"malloc buffer failed for event %d",evID) ; - factOut(kError,882, str ) ; - free(mBuffer[i].FADhead) ; - mBuffer[i].FADhead = NULL ; - free(mBuffer[i].fEvent) ; - mBuffer[i].fEvent = NULL ; + snprintf (str, MXSTR, "malloc buffer failed for event %d", evID); + factOut (kError, 882, str); + free (mBuffer[i].FADhead); + mBuffer[i].FADhead = NULL; + free (mBuffer[i].fEvent); + mBuffer[i].fEvent = NULL; return -32; } - - - //flag all boards as unused - mBuffer[i].nBoard = 0 ; - for (k=0; kStartPix[k] = -1 ; + for (k = 0; k < NPIX; k++) { + mBuffer[i].fEvent->StartPix[k] = -1; } //flag all TMark as unused - for (k=0; kStartTM[k] = -1 ; - } - - mBuffer[i].fEvent->NumBoards = 0 ; - mBuffer[i].fEvent->PCUsec = tusec ; - mBuffer[i].fEvent->PCTime = - mBuffer[i].pcTime = tsec ; - mBuffer[i].nRoi = nRoi[0] ; - mBuffer[i].nRoiTM = nRoi[8] ; - mBuffer[i].evNum = evID ; - mBuffer[i].runNum = runID ; - mBuffer[i].fadNum = fadNum; - mBuffer[i].trgNum = trgNum; - mBuffer[i].trgTyp = trgTyp ; - mBuffer[i].evtLen = needmem ; + for (k = 0; k < NTMARK; k++) { + mBuffer[i].fEvent->StartTM[k] = -1; + } + + mBuffer[i].fEvent->NumBoards = 0; + mBuffer[i].fEvent->PCUsec = tusec; + mBuffer[i].fEvent->PCTime = mBuffer[i].pcTime = tsec; + mBuffer[i].nRoi = nRoi[0]; + mBuffer[i].nRoiTM = nRoi[8]; + mBuffer[i].evNum = evID; + mBuffer[i].runNum = runID; + mBuffer[i].fadNum = fadNum; + mBuffer[i].trgNum = trgNum; + mBuffer[i].trgTyp = trgTyp; + mBuffer[i].evtLen = needmem; mBuffer[i].Errors[0] = - mBuffer[i].Errors[1] = - mBuffer[i].Errors[2] = - mBuffer[i].Errors[3] = 0 ; - - gj.usdMem += needmem + headmem + gi_maxSize ; - if (gj.usdMem > gj.maxMem ) gj.maxMem = gj.usdMem ; - - gj.bufTot++ ; - if (gj.bufTot > gj.maxEvt ) gj.maxEvt = gj.bufTot ; - - gj.rateNew++ ; + mBuffer[i].Errors[1] = mBuffer[i].Errors[2] = mBuffer[i].Errors[3] = 0; + + gj.usdMem += needmem + headmem + gi_maxSize; + if (gj.usdMem > gj.maxMem) + gj.maxMem = gj.usdMem; + + gj.bufTot++; + if (gj.bufTot > gj.maxEvt) + gj.maxEvt = gj.bufTot; + + gj.rateNew++; //register event in 'active list (reading)' - evtCtrl.evtBuf[ evtCtrl.lastPtr] = i ; - evtCtrl.evtStat[ evtCtrl.lastPtr] = 0 ; - evtCtrl.pcTime[ evtCtrl.lastPtr] = g_actTime ; - evtIdx[i] = evtCtrl.lastPtr ; - - - snprintf(str,MXSTR,"%5d %8d start new evt %8d %8d sock %3d len %5d t %10d", - evID,runID,i,evtCtrl.lastPtr,sk,fadlen,mBuffer[i].pcTime); - factOut(kDebug,-11, str ) ; - evtCtrl.lastPtr++ ; - if (evtCtrl.lastPtr == MAX_EVT*MAX_RUN ) evtCtrl.lastPtr = 0; - - gi.evtGet++ ; - - return i ; - + evtCtrl.evtBuf[evtCtrl.lastPtr] = i; + evtCtrl.evtStat[evtCtrl.lastPtr] = 0; + evtCtrl.pcTime[evtCtrl.lastPtr] = g_actTime; + evtIdx[i] = evtCtrl.lastPtr; + + + snprintf (str, MXSTR, + "%5d %8d start new evt %8d %8d sock %3d len %5d t %10d", evID, + runID, i, evtCtrl.lastPtr, sk, fadlen, mBuffer[i].pcTime); + factOut (kDebug, -11, str); + evtCtrl.lastPtr++; + if (evtCtrl.lastPtr == MAX_EVT * MAX_RUN) + evtCtrl.lastPtr = 0; + + gi.evtGet++; + + return i; + } /*-----------------------------------------------------------------*/ @@ -563,1269 +608,1510 @@ -int mBufFree(int i) { +int +mBufFree (int i) +{ //delete entry [i] from mBuffer: //(and make sure multiple calls do no harm ....) - int headmem=0 ; - int evid ; - size_t freemem = 0 ; - - evid = mBuffer[i].evNum ; - freemem = mBuffer[i].evtLen ; - - free(mBuffer[i].fEvent ) ; - mBuffer[i].fEvent = NULL ; - - free(mBuffer[i].FADhead ) ; - mBuffer[i].FADhead = NULL ; - - free(mBuffer[i].buffer ) ; - mBuffer[i].buffer = NULL ; - - headmem = NBOARDS* sizeof(PEVNT_HEADER) ; - mBuffer[i].evNum = mBuffer[i].nRoi= -1; - mBuffer[i].runNum = 0; - - gj.usdMem = gj.usdMem - freemem - headmem - gi_maxSize ; - gj.bufTot-- ; - - if (gi_memStat < 0 ) { - if (gj.usdMem <= 0.75 * gj.maxMem ) gi_memStat = +1 ; - } - - - return 0 ; - + int headmem = 0; + int evid; + size_t freemem = 0; + + evid = mBuffer[i].evNum; + freemem = mBuffer[i].evtLen; + + free (mBuffer[i].fEvent); + mBuffer[i].fEvent = NULL; + + free (mBuffer[i].FADhead); + mBuffer[i].FADhead = NULL; + + free (mBuffer[i].buffer); + mBuffer[i].buffer = NULL; + + headmem = NBOARDS * sizeof (PEVNT_HEADER); + mBuffer[i].evNum = mBuffer[i].nRoi = -1; + mBuffer[i].runNum = 0; + + gj.usdMem = gj.usdMem - freemem - headmem - gi_maxSize; + gj.bufTot--; + + if (gi_memStat < 0) { + if (gj.usdMem <= 0.75 * gj.maxMem) + gi_memStat = +1; + } + + + return 0; + } /*-----------------------------------------------------------------*/ -void resetEvtStat() { - int i ; - - for (i=0; i can not create affinity to %d",cpu); - factOut(kWarn,-1, str ) ; - } - - - head_len = sizeof(PEVNT_HEADER) ; - frst_len = head_len ; //max #bytes to read first: fad_header only, so each event must be longer, even for roi=0 - minLen = head_len ; //min #bytes needed to check header: full header for debug - - start.S=0xFB01; - stop.S= 0x04FE; + if (sched_setaffinity (0, sizeof (mask), &mask) == -1) { + snprintf (str, MXSTR, "W ---> can not create affinity to %d", cpu); + factOut (kWarn, -1, str); + } + + + head_len = sizeof (PEVNT_HEADER); + frst_len = head_len; //max #bytes to read first: fad_header only, so each event must be longer, even for roi=0 + minLen = head_len; //min #bytes needed to check header: full header for debug + + start.S = 0xFB01; + stop.S = 0x04FE; /* initialize run control logics */ - for (i=0; i 0) { - //make sure all sockets are preallocated as 'not exist' - for (i=0; i 0) { - resetEvtStat(); - gj.bufTot = gj.maxEvt = gj.xxxEvt = 0 ; - gj.usdMem = gj.maxMem = gj.xxxMem = 0 ; - gj.totMem = g_maxMem ; - gj.bufNew = gj.bufEvt = 0 ; - gj.badRoiE = gj.badRoiR = gj.badRoiB = - gj.evtSkip = gj.evtWrite = gj.evtErr = 0 ; - - int b ; - for (b=0; b=0 && g_reset ==0 ) { //loop until global variable g_runStat claims stop - - gi_runStat = g_runStat; - gj.readStat= g_runStat; - gettimeofday( tv, NULL); - g_actTime = tsec = atv.tv_sec ; - g_actUsec = tusec= atv.tv_usec ; - - - int b,p,p0,s0,nch; - nch = 0 ; - for (b=0; b 0 ) { - actBoards = 0 ; - for (b=0; b 0 ) actBoards++ ; - } - } - - - jrdx = 0; - numokx= 0; - numok = 0 ; //count number of succesfull actions - - for (i=0; i 0) s0= +1 ; - else s0= -1 ; - - if (rd[i].sockStat <0 ) { //try to connect if not yet done - rd[i].sockStat=connect(rd[i].socket, - (struct sockaddr*) &rd[i].SockAddr, sizeof(rd[i].SockAddr)) ; - if (rd[i].sockStat ==0 ) { //successfull ==> - if (sockDef[b] > 0) { - rd[i].bufTyp = 0 ; // expect a header - rd[i].bufLen = frst_len ; // max size to read at begining - } else { - rd[i].bufTyp = -1 ; // full data to be skipped - rd[i].bufLen = MAX_LEN ; //huge for skipping - } - rd[i].bufPos = 0 ; // no byte read so far - rd[i].skip = 0 ; // start empty - gi_NumConnect[ b ]++ ; - gi.numConn[ b ]++ ; - gj.numConn[ b ]++ ; - snprintf(str,MXSTR,"+++connect %d %d",b,gi.numConn[ b ]); - factOut(kInfo,-1, str ) ; - } - } - - if (rd[i].sockStat ==0 ) { //we have a connection ==> try to read - if (rd[i].bufLen > 0) { //might be nothing to read [buffer full] - numok++ ; - jrd=recv(rd[i].socket,&rd[i].rBuf->B[ rd[i].bufPos], rd[i].bufLen, MSG_DONTWAIT); - - if (jrd >0 ) { - debugStream(i,&rd[i].rBuf->B[ rd[i].bufPos],jrd) ; + for (i = 0; i < MAX_RUN; i++) { + runCtrl[i].runId = 0; + runCtrl[i].fileId = -2; + runCtrl[i].procId = -2; + } + gi_resetS = gi_resetR = 9; + for (i = 0; i < NBOARDS; i++) + sockDef[i] = 0; + + START: + gettimeofday (tv, NULL); + g_actTime = tsec = atv.tv_sec; + g_actUsec = tusec = atv.tv_usec; + gi_myRun = g_actTime; + evtCtrl.frstPtr = 0; + evtCtrl.lastPtr = 0; + + gi_SecTime = g_actTime; + gi_runStat = g_runStat; + gj.readStat = g_runStat; + numok = numok2 = 0; + + if (gi_resetS > 0) { + //make sure all sockets are preallocated as 'not exist' + for (i = 0; i < MAX_SOCK; i++) { + rd[i].socket = -1; + rd[i].sockStat = 99; + } + + for (k = 0; k < NBOARDS; k++) { + gi_NumConnect[k] = 0; + gi.numConn[k] = 0; + gj.numConn[k] = 0; + gj.errConn[k] = 0; + gj.rateBytes[k] = 0; + gj.totBytes[k] = 0; + } + + } + + + if (gi_resetR > 0) { + resetEvtStat (); + gj.bufTot = gj.maxEvt = gj.xxxEvt = 0; + gj.usdMem = gj.maxMem = gj.xxxMem = 0; + gj.totMem = g_maxMem; + gj.bufNew = gj.bufEvt = 0; + gj.badRoiE = gj.badRoiR = gj.badRoiB = + gj.evtSkip = gj.evtWrite = gj.evtErr = 0; + + int b; + for (b = 0; b < NBOARDS; b++) + gj.badRoi[b] = 0; + + mBufInit (); //initialize buffers + + snprintf (str, MXSTR, "end initializing"); + factOut (kInfo, -1, str); + } + + + gi_reset = gi_resetR = gi_resetS = gi_resetW = 0; + + while (g_runStat >= 0 && g_reset == 0) { //loop until global variable g_runStat claims stop + + gi_runStat = g_runStat; + gj.readStat = g_runStat; + gettimeofday (tv, NULL); + g_actTime = tsec = atv.tv_sec; + g_actUsec = tusec = atv.tv_usec; + + + int b, p, p0, s0, nch; + nch = 0; + for (b = 0; b < NBOARDS; b++) { + k = b * 7; + if (g_port[b].sockDef != sockDef[b]) { //something has changed ... + nch++; + gi_NumConnect[b] = 0; //must close all connections + gi.numConn[b] = 0; + gj.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 + else + s0 = +1; //sockets to be closed and reopened + + if (s0 == 0) + p0 = ntohs (g_port[b].sockAddr.sin_port); + else + p0 = 0; + + for (p = p0 + 1; p < p0 + 8; p++) { + GenSock (s0, k, p, &g_port[b].sockAddr, &rd[k]); //generate address and socket + k++; + } + sockDef[b] = g_port[b].sockDef; + } + } + + if (nch > 0) { + actBoards = 0; + for (b = 0; b < NBOARDS; b++) { + if (sockDef[b] > 0) + actBoards++; + } + } + + + jrdx = 0; + numokx = 0; + numok = 0; //count number of succesfull actions + + for (i = 0; i < MAX_SOCK; i++) { //check all sockets if something to read + b = i / 7; + if (sockDef[b] > 0) + s0 = +1; + else + s0 = -1; + + if (rd[i].sockStat < 0) { //try to connect if not yet done + rd[i].sockStat = connect (rd[i].socket, + (struct sockaddr *) &rd[i].SockAddr, + sizeof (rd[i].SockAddr)); + if (rd[i].sockStat == 0) { //successfull ==> + if (sockDef[b] > 0) { + rd[i].bufTyp = 0; // expect a header + rd[i].bufLen = frst_len; // max size to read at begining + } else { + rd[i].bufTyp = -1; // full data to be skipped + rd[i].bufLen = MAX_LEN; //huge for skipping + } + rd[i].bufPos = 0; // no byte read so far + rd[i].skip = 0; // start empty + gi_NumConnect[b]++; + gi.numConn[b]++; + gj.numConn[b]++; + snprintf (str, MXSTR, "+++connect %d %d", b, gi.numConn[b]); + factOut (kInfo, -1, str); + } + } + + if (rd[i].sockStat == 0) { //we have a connection ==> try to read + if (rd[i].bufLen > 0) { //might be nothing to read [buffer full] + numok++; + jrd = + recv (rd[i].socket, &rd[i].rBuf->B[rd[i].bufPos], + rd[i].bufLen, MSG_DONTWAIT); + + if (jrd > 0) { + debugStream (i, &rd[i].rBuf->B[rd[i].bufPos], jrd); #ifdef EVTDEBUG - memcpy(&rd[i].xBuf->B[ rd[i].bufPos], &rd[i].rBuf->B[ rd[i].bufPos], jrd) ; - snprintf(str,MXSTR,"read sock %3d bytes %5d len %5d first %d %d",i,jrd,rd[i].bufLen, - rd[i].rBuf->B[ rd[i].bufPos ], - rd[i].rBuf->B[ rd[i].bufPos +1] ); - factOut(kDebug,301, str ) ; + memcpy (&rd[i].xBuf->B[rd[i].bufPos], + &rd[i].rBuf->B[rd[i].bufPos], jrd); + snprintf (str, MXSTR, + "read sock %3d bytes %5d len %5d first %d %d", i, + jrd, rd[i].bufLen, rd[i].rBuf->B[rd[i].bufPos], + rd[i].rBuf->B[rd[i].bufPos + 1]); + factOut (kDebug, 301, str); #endif - } - - if (jrd == 0) { //connection has closed ... - snprintf(str,MXSTR,"Socket %d closed by FAD",i); - factOut(kInfo,441, str ) ; - GenSock(s0, i, 0,NULL, &rd[i]) ; - gi.gotErr[ b ]++ ; - gi_NumConnect[ b ]-- ; - gi.numConn[ b ]-- ; - gj.numConn[ b ]-- ; - - } else if ( jrd<0 ) { //did not read anything - if (errno != EAGAIN && errno != EWOULDBLOCK ) { - snprintf(str,MXSTR,"Error Reading from %d | %m",i); - factOut(kError,442, str ) ; - gi.gotErr[ b ]++ ; - } else numok-- ; //else nothing waiting to be read - jrd = 0 ; - } - } else { - jrd = 0 ; //did read nothing as requested - snprintf(str,MXSTR,"do not read from socket %d %d",i,rd[i].bufLen ) ; - factOut(kDebug,301, str ) ; - } - - gi.gotByte[ b ] += jrd ; - gj.rateBytes[b] += jrd ; - - if (jrd>0) {numokx++ ; jrdx+= jrd; } - - - if ( rd[i].bufTyp <0 ) { // we are skipping this board ... + } + + if (jrd == 0) { //connection has closed ... + snprintf (str, MXSTR, "Socket %d closed by FAD", i); + factOut (kInfo, 441, str); + GenSock (s0, i, 0, NULL, &rd[i]); + gi.gotErr[b]++; + gi_NumConnect[b]--; + gi.numConn[b]--; + gj.numConn[b]--; + + } else if (jrd < 0) { //did not read anything + if (errno != EAGAIN && errno != EWOULDBLOCK) { + snprintf (str, MXSTR, "Error Reading from %d | %m", i); + factOut (kError, 442, str); + gi.gotErr[b]++; + } else + numok--; //else nothing waiting to be read + jrd = 0; + } + } else { + jrd = 0; //did read nothing as requested + snprintf (str, MXSTR, "do not read from socket %d %d", i, + rd[i].bufLen); + factOut (kDebug, 301, str); + } + + gi.gotByte[b] += jrd; + gj.rateBytes[b] += jrd; + + if (jrd > 0) { + numokx++; + jrdx += jrd; + } + + + if (rd[i].bufTyp < 0) { // we are skipping this board ... // just do nothing #ifdef EVTDEBUG - snprintf(str,MXSTR,"skipping %d bytes on socket %d",jrd,i) ; - factOut(kInfo,301, str ) ; + snprintf (str, MXSTR, "skipping %d bytes on socket %d", jrd, + i); + factOut (kInfo, 301, str); #endif - } else if ( rd[i].bufTyp >0 ) { // we are reading data ... - if ( jrd < rd[i].bufLen ) { //not yet all read - rd[i].bufPos += jrd ; //==> prepare for continuation - rd[i].bufLen -= jrd ; - debugRead(i,jrd,rd[i].evtID,rd[i].ftmID,rd[i].runID, 0,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; 0=reading data - } else { //full dataset read - rd[i].bufLen = 0 ; - rd[i].bufPos = rd[i].fadLen ; - if ( rd[i].rBuf->B[ rd[i].bufPos-1] != stop.B[0] - || rd[i].rBuf->B[ rd[i].bufPos-2] != stop.B[1]) { - gi.evtErr++ ; - snprintf(str,MXSTR,"wrong end of buffer found sock %3d ev %4d len %5d %3d %3d - %3d %3d ", - i,rd[i].fadLen, rd[i].evtID, rd[i].rBuf->B[0], rd[i].rBuf->B[1], rd[i].rBuf->B[ rd[i].bufPos-2], - rd[i].rBuf->B[ rd[i].bufPos-1]); - factOut(kError,301, str ) ; - goto EndBuf ; + } else if (rd[i].bufTyp > 0) { // we are reading data ... + if (jrd < rd[i].bufLen) { //not yet all read + rd[i].bufPos += jrd; //==> prepare for continuation + rd[i].bufLen -= jrd; + debugRead (i, jrd, rd[i].evtID, rd[i].ftmID, rd[i].runID, 0, tsec, tusec); // i=socket; jrd=#bytes; ievt=eventid; 0=reading data + } else { //full dataset read + rd[i].bufLen = 0; + rd[i].bufPos = rd[i].fadLen; + if (rd[i].rBuf->B[rd[i].bufPos - 1] != stop.B[0] + || rd[i].rBuf->B[rd[i].bufPos - 2] != stop.B[1]) { + gi.evtErr++; + snprintf (str, MXSTR, + "wrong end of buffer found sock %3d ev %4d len %5d %3d %3d - %3d %3d ", + i, rd[i].fadLen, rd[i].evtID, rd[i].rBuf->B[0], + rd[i].rBuf->B[1], + rd[i].rBuf->B[rd[i].bufPos - 2], + rd[i].rBuf->B[rd[i].bufPos - 1]); + factOut (kError, 301, str); + goto EndBuf; #ifdef EVTDEBUG - } else { - snprintf(str,MXSTR,"good end of buffer found sock %3d len %5d %d %d : %d %d - %d %d : %d %d", - i,rd[i].fadLen, - rd[i].rBuf->B[ 0 ], rd[i].rBuf->B[ 1 ], start.B[1],start.B[0], - rd[i].rBuf->B[ rd[i].bufPos-2], rd[i].rBuf->B[ rd[i].bufPos-1], stop.B[1], stop.B[0]); - factOut(kDebug,301, str ) ; + } else { + snprintf (str, MXSTR, + "good end of buffer found sock %3d len %5d %d %d : %d %d - %d %d : %d %d", + i, rd[i].fadLen, rd[i].rBuf->B[0], + rd[i].rBuf->B[1], start.B[1], start.B[0], + rd[i].rBuf->B[rd[i].bufPos - 2], + rd[i].rBuf->B[rd[i].bufPos - 1], stop.B[1], + stop.B[0]); + factOut (kDebug, 301, str); #endif - } - - if (jrd>0) debugRead(i,jrd,rd[i].evtID,rd[i].ftmID,rd[i].runID, 1,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; 1=finished event - - //we have a complete buffer, copy to WORK area - int jr ; - roi[0] = ntohs(rd[i].rBuf->S[ head_len/2 + 2 ]) ; - for (jr=0; jr<9; jr++) { - roi[jr] = ntohs(rd[i].rBuf->S[ head_len/2 + 2 + jr*(roi[0]+4) ]) ; - } - //get index into mBuffer for this event (create if needed) - - int actid; - if (g_useFTM >0) actid = rd[i].evtID ; - else actid = rd[i].ftmID ; - - evID = mBufEvt( rd[i].evtID, rd[i].runID, roi, i, - rd[i].fadLen, rd[i].ftmTyp, rd[i].ftmID, rd[i].evtID ) ; - - if (evID <-1000) { - goto EndBuf ; //not usable board/event/run --> skip it - } - if (evID < 0) { //no space left, retry later + } + + if (jrd > 0) + debugRead (i, jrd, rd[i].evtID, rd[i].ftmID, rd[i].runID, 1, tsec, tusec); // i=socket; jrd=#bytes; ievt=eventid; 1=finished event + + //we have a complete buffer, copy to WORK area + int jr; + roi[0] = ntohs (rd[i].rBuf->S[head_len / 2 + 2]); + for (jr = 0; jr < 9; jr++) { + roi[jr] = + ntohs (rd[i]. + rBuf->S[head_len / 2 + 2 + jr * (roi[0] + 4)]); + } + //get index into mBuffer for this event (create if needed) + + int actid; + if (g_useFTM > 0) + actid = rd[i].evtID; + else + actid = rd[i].ftmID; + + evID = mBufEvt (rd[i].evtID, rd[i].runID, roi, i, + rd[i].fadLen, rd[i].ftmTyp, rd[i].ftmID, + rd[i].evtID); + + if (evID < -1000) { + goto EndBuf; //not usable board/event/run --> skip it + } + if (evID < 0) { //no space left, retry later #ifdef EVTDEBUG - if ( rd[i].bufLen != 0) { - snprintf(str,MXSTR,"something screwed up"); - factOut(kFatal, 1, str ) ; - } + if (rd[i].bufLen != 0) { + snprintf (str, MXSTR, "something screwed up"); + factOut (kFatal, 1, str); + } #endif - xwait.tv_sec = 0; - xwait.tv_nsec= 10000000 ; // sleep for ~10 msec - nanosleep( &xwait , NULL ) ; - goto EndBuf1 ; //hope there is free space next round - } - - - //we have a valid entry in mBuffer[]; fill it + xwait.tv_sec = 0; + xwait.tv_nsec = 10000000; // sleep for ~10 msec + nanosleep (&xwait, NULL); + goto EndBuf1; //hope there is free space next round + } + //we have a valid entry in mBuffer[]; fill it #ifdef EVTDEBUG - int xchk = memcmp(&rd[i].xBuf->B[0], &rd[i].rBuf->B[0], rd[i].fadLen ) ; - if (xchk != 0) { - snprintf(str,MXSTR,"ERROR OVERWRITE %d %d on port %d",xchk,rd[i].fadLen,i) ; - factOut(kFatal, 1, str ) ; - - uint iq; - for (iq=0; iq < rd[i].fadLen ; iq++) { - if (rd[i].rBuf->B[iq] != rd[i].xBuf->B[iq] ) { - snprintf(str,MXSTR,"ERROR %4d %4d %x %x",i,iq,rd[i].rBuf->B[iq], rd[i].xBuf->B[iq]); - factOut(kFatal, 1, str ) ; - } + int xchk = memcmp (&rd[i].xBuf->B[0], &rd[i].rBuf->B[0], + rd[i].fadLen); + if (xchk != 0) { + snprintf (str, MXSTR, "ERROR OVERWRITE %d %d on port %d", + xchk, rd[i].fadLen, i); + factOut (kFatal, 1, str); + + uint iq; + for (iq = 0; iq < rd[i].fadLen; iq++) { + if (rd[i].rBuf->B[iq] != rd[i].xBuf->B[iq]) { + snprintf (str, MXSTR, "ERROR %4d %4d %x %x", i, iq, + rd[i].rBuf->B[iq], rd[i].xBuf->B[iq]); + factOut (kFatal, 1, str); + } + } + } +#endif + int qncpy = 0; + boardId = b; + int fadBoard = ntohs (rd[i].rBuf->S[12]); + int fadCrate = fadBoard / 256; + if (boardId != (fadCrate * 10 + fadBoard % 256)) { + snprintf (str, MXSTR, "wrong Board ID %d %d %d", + fadCrate, fadBoard % 256, boardId); + factOut (kWarn, 301, str); + } + if (mBuffer[evID].board[boardId] != -1) { + snprintf (str, MXSTR, + "double board: ev %5d, b %3d, %3d; len %5d %3d %3d - %3d %3d ", + evID, boardId, i, rd[i].fadLen, + rd[i].rBuf->B[0], rd[i].rBuf->B[1], + rd[i].rBuf->B[rd[i].bufPos - 2], + rd[i].rBuf->B[rd[i].bufPos - 1]); + factOut (kWarn, 501, str); + goto EndBuf; //--> skip Board + } + + int iDx = evtIdx[evID]; //index into evtCtrl + + memcpy (&mBuffer[evID].FADhead[boardId].start_package_flag, + &rd[i].rBuf->S[0], head_len); + qncpy += head_len; + + src = head_len / 2; + for (px = 0; px < 9; px++) { //different sort in FAD board..... + for (drs = 0; drs < 4; drs++) { + pixH = ntohs (rd[i].rBuf->S[src++]); // ID + pixC = ntohs (rd[i].rBuf->S[src++]); // start-cell + pixR = ntohs (rd[i].rBuf->S[src++]); // roi +//here we should check if pixH is correct .... + + pixS = boardId * 36 + drs * 9 + px; + src++; + + + mBuffer[evID].fEvent->StartPix[pixS] = pixC; + dest = pixS * roi[0]; + memcpy (&mBuffer[evID].fEvent->Adc_Data[dest], + &rd[i].rBuf->S[src], roi[0] * 2); + qncpy += roi[0] * 2; + src += pixR; + + if (px == 8) { + tmS = boardId * 4 + drs; + if (pixR > roi[0]) { //and we have additional TM info + dest = tmS * roi[0] + NPIX * roi[0]; + int srcT = src - roi[0]; + mBuffer[evID].fEvent->StartTM[tmS] = + (pixC + pixR - roi[0]) % 1024; + memcpy (&mBuffer[evID].fEvent->Adc_Data[dest], + &rd[i].rBuf->S[srcT], roi[0] * 2); + qncpy += roi[0] * 2; + } else { + mBuffer[evID].fEvent->StartTM[tmS] = -1; + } + } + } + } // now we have stored a new board contents into Event structure + + mBuffer[evID].fEvent->NumBoards++; + mBuffer[evID].board[boardId] = boardId; + evtCtrl.evtStat[iDx]++; + evtCtrl.pcTime[iDx] = g_actTime; + + if (++mBuffer[evID].nBoard >= actBoards) { + int qnrun = 0; + if (mBuffer[evID].runNum != actrun) { // have we already reported first event of this run ??? + actrun = mBuffer[evID].runNum; + int ir; + for (ir = 0; ir < MAX_RUN; ir++) { + qnrun++; + if (runCtrl[ir].runId == actrun) { + if (++runCtrl[ir].lastEvt == 0) { + gotNewRun (actrun, mBuffer[evID].FADhead); + snprintf (str, MXSTR, "gotNewRun %d (ev %d)", + mBuffer[evID].runNum, + mBuffer[evID].evNum); + factOut (kInfo, 1, str); + break; + } + } + } + } + snprintf (str, MXSTR, + "%5d complete event roi %4d roiTM %d cpy %8d %5d", + mBuffer[evID].evNum, roi[0], roi[8] - roi[0], + qncpy, qnrun); + factOut (kDebug, -1, str); + + //complete event read ---> flag for next processing + evtCtrl.evtStat[iDx] = 99; + gi.evtTot++; + } + + EndBuf: + rd[i].bufTyp = 0; //ready to read next header + rd[i].bufLen = frst_len; + rd[i].bufPos = 0; + EndBuf1: + ; } - } + + } else { //we are reading event header + rd[i].bufPos += jrd; + rd[i].bufLen -= jrd; + if (rd[i].bufPos >= minLen) { //sufficient data to take action + //check if startflag correct; else shift block .... + for (k = 0; k < rd[i].bufPos - 1; k++) { + if (rd[i].rBuf->B[k] == start.B[1] + && rd[i].rBuf->B[k + 1] == start.B[0]) + break; + } + rd[i].skip += k; + + if (k >= rd[i].bufPos - 1) { //no start of header found + rd[i].bufPos = 0; + rd[i].bufLen = head_len; + } else if (k > 0) { + rd[i].bufPos -= k; + rd[i].bufLen += k; + memcpy (&rd[i].rBuf->B[0], &rd[i].rBuf->B[k], + rd[i].bufPos); +#ifdef EVTDEBUG + memcpy (&rd[i].xBuf->B[0], &rd[i].xBuf->B[k], + rd[i].bufPos); #endif - int qncpy = 0 ; - boardId = b ; - int fadBoard = ntohs(rd[i].rBuf->S[12] ) ; - int fadCrate = fadBoard/256 ; - if (boardId != (fadCrate*10 + fadBoard%256) ) { - snprintf(str,MXSTR,"wrong Board ID %d %d %d",fadCrate,fadBoard%256,boardId) ; - factOut(kWarn,301, str ) ; - } - if ( mBuffer[evID].board[ boardId ] != -1) { - snprintf(str,MXSTR,"double board: ev %5d, b %3d, %3d; len %5d %3d %3d - %3d %3d ", - evID,boardId,i,rd[i].fadLen, rd[i].rBuf->B[0], - rd[i].rBuf->B[1], rd[i].rBuf->B[ rd[i].bufPos-2], rd[i].rBuf->B[ rd[i].bufPos-1]); - factOut(kWarn,501, str ) ; - goto EndBuf ; //--> skip Board - } - - int iDx = evtIdx[evID] ; //index into evtCtrl - - memcpy( &mBuffer[evID].FADhead[boardId].start_package_flag, - &rd[i].rBuf->S[0], head_len) ; - qncpy+=head_len ; - - src = head_len/2 ; - for ( px=0; px<9; px++ ) { //different sort in FAD board..... - for ( drs=0; drs<4; drs++ ) { - pixH= ntohs(rd[i].rBuf->S[src++]) ; // ID - pixC= ntohs(rd[i].rBuf->S[src++]) ; // start-cell - pixR= ntohs(rd[i].rBuf->S[src++]) ; // roi -//here we should check if pixH is correct .... - - pixS = boardId*36 + drs*9 + px ; - src++ ; - - - mBuffer[evID].fEvent->StartPix[pixS] =pixC; - dest= pixS * roi[0] ; - memcpy( - &mBuffer[evID].fEvent->Adc_Data[dest], - &rd[i].rBuf->S[src], roi[0] * 2) ; - qncpy+=roi[0]*2 ; - src+= pixR ; - - if ( px==8 ) { - tmS =boardId*4 + drs ; - if ( pixR > roi[0]) { //and we have additional TM info - dest= tmS * roi[0] + NPIX* roi[0] ; - int srcT= src - roi[0] ; - mBuffer[evID].fEvent->StartTM[tmS] = (pixC+pixR-roi[0])%1024 ; - memcpy( - &mBuffer[evID].fEvent->Adc_Data[dest], - &rd[i].rBuf->S[srcT], roi[0] * 2) ; - qncpy+=roi[0]*2 ; - } else { - mBuffer[evID].fEvent->StartTM[tmS] = -1 ; - } - } - } - }// now we have stored a new board contents into Event structure - - mBuffer[evID].fEvent->NumBoards++ ; - mBuffer[evID].board[ boardId ] = boardId ; - evtCtrl.evtStat[ iDx ]++ ; - evtCtrl.pcTime[ iDx ] = g_actTime ; - - if (++mBuffer[evID].nBoard >= actBoards ) { - int qnrun =0 ; - if (mBuffer[evID].runNum != actrun ) { // have we already reported first event of this run ??? - actrun = mBuffer[evID].runNum ; - int ir ; - for ( ir=0; ir flag for next processing - evtCtrl.evtStat[ iDx ] = 99; - gi.evtTot++ ; - } - -EndBuf: - rd[i].bufTyp = 0 ; //ready to read next header - rd[i].bufLen = frst_len ; - rd[i].bufPos = 0 ; -EndBuf1: - ; - } - - } else { //we are reading event header - rd[i].bufPos += jrd ; - rd[i].bufLen -= jrd ; - if ( rd[i].bufPos >= minLen ){ //sufficient data to take action - //check if startflag correct; else shift block .... - for (k=0; kB[k ] == start.B[1] - && rd[i].rBuf->B[k+1] == start.B[0] ) break ; - } - rd[i].skip += k ; - - if (k >= rd[i].bufPos-1 ) { //no start of header found - rd[i].bufPos = 0 ; - rd[i].bufLen = head_len ; - } else if ( k>0 ) { - rd[i].bufPos -= k ; - rd[i].bufLen += k ; - memcpy(&rd[i].rBuf->B[0], &rd[i].rBuf->B[k], rd[i].bufPos ) ; + } + if (rd[i].bufPos >= minLen) { + if (rd[i].skip > 0) { + snprintf (str, MXSTR, "skipped %d bytes on port%d", + rd[i].skip, i); + factOut (kInfo, 666, str); + rd[i].skip = 0; + } + goodhed++; + rd[i].fadLen = ntohs (rd[i].rBuf->S[1]) * 2; + rd[i].fadVers = ntohs (rd[i].rBuf->S[2]); + rd[i].ftmTyp = ntohl (rd[i].rBuf->S[5]); + rd[i].ftmID = ntohl (rd[i].rBuf->I[3]); //(FTMevt) + rd[i].evtID = ntohl (rd[i].rBuf->I[4]); //(FADevt) + rd[i].runID = ntohl (rd[i].rBuf->I[11]); + rd[i].bufTyp = 1; //ready to read full record + rd[i].bufLen = rd[i].fadLen - rd[i].bufPos; + + int fadboard = ntohs (rd[i].rBuf->S[12]); + int fadcrate = fadboard / 256; + fadboard = (fadcrate * 10 + fadboard % 256); #ifdef EVTDEBUG - memcpy(&rd[i].xBuf->B[0], &rd[i].xBuf->B[k], rd[i].bufPos ) ; + snprintf (str, MXSTR, + "sk %3d head: %5d %5d %5d %10d %4d %6d", i, + rd[i].fadLen, rd[i].evtID, rd[i].ftmID, + rd[i].runID, fadboard, jrd); + factOut (kDebug, 1, str); #endif - } - if ( rd[i].bufPos >= minLen ) { - if ( rd[i].skip >0 ) { - snprintf(str,MXSTR,"skipped %d bytes on port%d", rd[i].skip, i ) ; - factOut(kInfo,666, str ) ; - rd[i].skip = 0 ; - } - goodhed++; - rd[i].fadLen = ntohs(rd[i].rBuf->S[1])*2 ; - rd[i].fadVers= ntohs(rd[i].rBuf->S[2]) ; - rd[i].ftmTyp = ntohl(rd[i].rBuf->S[5]) ; - rd[i].ftmID = ntohl(rd[i].rBuf->I[3]) ; //(FTMevt) - rd[i].evtID = ntohl(rd[i].rBuf->I[4]) ; //(FADevt) - rd[i].runID = ntohl(rd[i].rBuf->I[11]) ; - rd[i].bufTyp = 1 ; //ready to read full record - rd[i].bufLen = rd[i].fadLen - rd[i].bufPos ; - - int fadboard = ntohs(rd[i].rBuf->S[12] ) ; - int fadcrate = fadboard/256 ; - fadboard = (fadcrate*10 + fadboard%256) ; + + if (rd[i].runID == 0) + rd[i].runID = gi_myRun; + + if (rd[i].bufLen <= head_len || rd[i].bufLen > MAX_LEN) { + snprintf (str, MXSTR, + "illegal event-length on port %d", i); + factOut (kFatal, 881, str); + rd[i].bufLen = 100000; //? + } + 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 { + debugRead (i, jrd, 0, 0, 0, -2, tsec, tusec); // i=socket; jrd=#bytes; ievt=eventid; -2=start event, unknown id yet + } + } else { + debugRead (i, jrd, 0, 0, 0, -2, tsec, tusec); // i=socket; jrd=#bytes; ievt=eventid; -2=start event, unknown id yet + } + + } //end interpreting last read + } //end of successful read anything + } //finished trying to read all sockets + #ifdef EVTDEBUG - snprintf(str,MXSTR,"sk %3d head: %5d %5d %5d %10d %4d %6d",i,rd[i].fadLen,rd[i].evtID,rd[i].ftmID,rd[i].runID,fadboard,jrd) ; - factOut(kDebug,1, str ) ; + snprintf (str, MXSTR, "Loop ---- %3d --- %8d", numokx, jrdx); + factOut (kDebug, -1, str); #endif - if (rd[i].runID ==0 ) rd[i].runID = gi_myRun ; - - if (rd[i].bufLen <=head_len || rd[i].bufLen > MAX_LEN ) { - snprintf(str,MXSTR,"illegal event-length on port %d",i) ; - factOut(kFatal,881, str ) ; - rd[i].bufLen = 100000 ; //? - } - 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 { - debugRead(i,jrd,0,0,0,-2,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; -2=start event, unknown id yet - } - } else { - debugRead(i,jrd,0,0,0,-2,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; -2=start event, unknown id yet - } - - } //end interpreting last read - } //end of successful read anything - } //finished trying to read all sockets - -#ifdef EVTDEBUG - snprintf(str,MXSTR,"Loop ---- %3d --- %8d",numokx,jrdx); - factOut(kDebug,-1, str ) ; -#endif - - gi.numRead[ numok ] ++ ; - - g_actTime = time(NULL) ; - if ( g_actTime > gi_SecTime ) { - 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 ; - int k1=evtCtrl.frstPtr; - for ( k=k1; k<(k1+kd); k++ ) { - int k0 = k % (MAX_EVT*MAX_RUN) ; + gi.numRead[numok]++; + + g_actTime = time (NULL); + if (g_actTime > gi_SecTime) { + 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; + 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] > 0 - && evtCtrl.evtStat[k0] < 92 ) { - gj.bufNew++ ; //incomplete event in Buffer - if ( evtCtrl.evtStat[k0] < 90 - && evtCtrl.pcTime[k0] < g_actTime-10 ) { - int id =evtCtrl.evtBuf[k0] ; - snprintf(str,MXSTR,"%5d skip short evt %8d %8d %2d",mBuffer[id].evNum,evtCtrl.evtBuf[k0],k0 ,evtCtrl.evtStat[k0]); - factOut(kWarn,601, str ) ; - evtCtrl.evtStat[k0] = 91 ; //timeout for incomplete events - gi.evtSkp++ ; - gi.evtTot++ ; - gj.evtSkip++; - } - } else if (evtCtrl.evtStat[k0] >= 9000 //'delete' - || evtCtrl.evtStat[k0] == 0 ) { //'useless' - - int id =evtCtrl.evtBuf[k0] ; - snprintf(str,MXSTR,"%5d free event buffer, nb=%3d", mBuffer[id].evNum, mBuffer[id].nBoard ) ; - factOut(kDebug,-1, str ) ; - 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) ; - } - } - - - gj.deltaT = 1000 ; //temporary, must be improved - - int b; - for ( b=0; b gj.xxxMem) gj.xxxMem = gj.maxMem ; - if (gj.maxEvt > gj.xxxEvt) gj.xxxEvt = gj.maxEvt ; - - factStat(gj); - factStatNew(gi) ; - gj.rateNew = gj.rateWrite = 0 ; - gj.maxMem = gj.usdMem ; - gj.maxEvt = gj.bufTot ; - for ( b=0; b 0 ) numok2=0; - else if (numok2++ > 3) { - if (g_runStat == 1) { - xwait.tv_sec = 1; - xwait.tv_nsec= 0 ; // hibernate for 1 sec - } else { - xwait.tv_sec = 0; - xwait.tv_nsec= 2000000 ; // sleep for ~2 msec - } - nanosleep( &xwait , NULL ) ; - } - - } //and do next loop over all sockets ... - - - snprintf(str,MXSTR,"stop reading ... RESET=%d",g_reset); - factOut(kInfo,-1, str ) ; - - if (g_reset >0 ) { - gi_reset = g_reset ; - gi_resetR = gi_reset%10 ; //shall we stop reading ? - gi_resetS = (gi_reset/10)%10 ; //shall we close sockets ? - gi_resetW = (gi_reset/100)%10 ; //shall we close files ? - gi_resetX = gi_reset/1000 ; //shall we simply wait resetX seconds ? - g_reset= 0; - } else { - gi_reset = 0; - if ( g_runStat== -1 ) gi_resetR = 1 ; - else gi_resetR = 7 ; - gi_resetS = 7 ; //close all sockets - gi_resetW = 7 ; //close all files - gi_resetX = 0 ; - - //inform others we have to quit .... - gi_runStat = -11 ; //inform all that no update to happen any more - gj.readStat= -11 ; //inform all that no update to happen any more - } - - if (gi_resetS > 0) { - //must close all open sockets ... - snprintf(str,MXSTR,"close all sockets ..."); - factOut(kInfo,-1, str ) ; - for (i=0; i 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= 2000000 ; // sleep for ~2 msec - nanosleep( &xwait , NULL ) ; - - //and clear all buffers (might have to wait until all others are done) - int minclear ; - if (gi_resetR == 1) { - minclear = 900 ; - snprintf(str,MXSTR,"drain all buffers ..."); - } else { - minclear = 0 ; - snprintf(str,MXSTR,"flush all buffers ..."); - } - factOut(kInfo,-1, str ) ; - - int numclear=1 ; - 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] ; - snprintf(str,MXSTR,"ev %5d free event buffer, nb=%3d", mBuffer[id].evNum, mBuffer[id].nBoard ) ; - factOut(kDebug,-1, str ) ; - 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) ; - } - - xwait.tv_sec = 0; - xwait.tv_nsec= 2000000 ; // sleep for ~2 msec - nanosleep( &xwait , NULL ) ; - } - } - - if (gi_reset > 0) { - if (gi_resetW > 0) { - CloseRunFile(0,0,0) ; //ask all Runs to be closed - } - if (gi_resetX > 0) { - xwait.tv_sec = gi_resetX; - xwait.tv_nsec= 0 ; - nanosleep( &xwait , NULL ) ; - } - - snprintf(str,MXSTR,"Continue read Process ..."); - factOut(kInfo,-1, str ) ; - gi_reset = 0 ; - goto START ; - } - - - - snprintf(str,MXSTR,"Exit read Process ..."); - factOut(kInfo,-1, str ) ; - gi_runStat = -99 ; - gj.readStat= -99 ; - factStat(gj); - factStatNew(gi) ; - return 0; + if (evtCtrl.evtStat[k0] > 0 && evtCtrl.evtStat[k0] < 92) { + gj.bufNew++; //incomplete event in Buffer + if (evtCtrl.evtStat[k0] < 90 + && evtCtrl.pcTime[k0] < g_actTime - 10) { + int id = evtCtrl.evtBuf[k0]; + snprintf (str, MXSTR, "%5d skip short evt %8d %8d %2d", + mBuffer[id].evNum, evtCtrl.evtBuf[k0], k0, + evtCtrl.evtStat[k0]); + factOut (kWarn, 601, str); + evtCtrl.evtStat[k0] = 91; //timeout for incomplete events + gi.evtSkp++; + gi.evtTot++; + gj.evtSkip++; + } + } else if (evtCtrl.evtStat[k0] >= 9000 //'delete' + || evtCtrl.evtStat[k0] == 0) { //'useless' + + int id = evtCtrl.evtBuf[k0]; + snprintf (str, MXSTR, "%5d free event buffer, nb=%3d", + mBuffer[id].evNum, mBuffer[id].nBoard); + factOut (kDebug, -1, str); + 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); + } + } + + + gj.deltaT = 1000; //temporary, must be improved + + int b; + for (b = 0; b < NBOARDS; b++) + gj.totBytes[b] += gj.rateBytes[b]; + gj.totMem = g_maxMem; + if (gj.maxMem > gj.xxxMem) + gj.xxxMem = gj.maxMem; + if (gj.maxEvt > gj.xxxEvt) + gj.xxxEvt = gj.maxEvt; + + factStat (gj); + factStatNew (gi); + gj.rateNew = gj.rateWrite = 0; + gj.maxMem = gj.usdMem; + gj.maxEvt = gj.bufTot; + for (b = 0; b < NBOARDS; b++) + gj.rateBytes[b] = 0; + } + + if (numok > 0) + numok2 = 0; + else if (numok2++ > 3) { + if (g_runStat == 1) { + xwait.tv_sec = 1; + xwait.tv_nsec = 0; // hibernate for 1 sec + } else { + xwait.tv_sec = 0; + xwait.tv_nsec = 2000000; // sleep for ~2 msec + } + nanosleep (&xwait, NULL); + } + + } //and do next loop over all sockets ... + + + snprintf (str, MXSTR, "stop reading ... RESET=%d", g_reset); + factOut (kInfo, -1, str); + + if (g_reset > 0) { + gi_reset = g_reset; + gi_resetR = gi_reset % 10; //shall we stop reading ? + gi_resetS = (gi_reset / 10) % 10; //shall we close sockets ? + gi_resetW = (gi_reset / 100) % 10; //shall we close files ? + gi_resetX = gi_reset / 1000; //shall we simply wait resetX seconds ? + g_reset = 0; + } else { + gi_reset = 0; + if (g_runStat == -1) + gi_resetR = 1; + else + gi_resetR = 7; + gi_resetS = 7; //close all sockets + gi_resetW = 7; //close all files + gi_resetX = 0; + + //inform others we have to quit .... + gi_runStat = -11; //inform all that no update to happen any more + gj.readStat = -11; //inform all that no update to happen any more + } + + if (gi_resetS > 0) { + //must close all open sockets ... + snprintf (str, MXSTR, "close all sockets ..."); + factOut (kInfo, -1, str); + for (i = 0; i < MAX_SOCK; i++) { + if (rd[i].sockStat == 0) { + GenSock (-1, i, 0, NULL, &rd[i]); //close and destroy open socket + if (i % 7 == 0) { + gi_NumConnect[i / 7]--; + gi.numConn[i / 7]--; + gj.numConn[i / 7]--; + sockDef[i / 7] = 0; //flag ro recreate the sockets ... + rd[i / 7].sockStat = -1; //and try to open asap + } + } + } + } + + + 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 = 2000000; // sleep for ~2 msec + nanosleep (&xwait, NULL); + + //and clear all buffers (might have to wait until all others are done) + int minclear; + if (gi_resetR == 1) { + minclear = 900; + snprintf (str, MXSTR, "drain all buffers ..."); + } else { + minclear = 0; + snprintf (str, MXSTR, "flush all buffers ..."); + } + factOut (kInfo, -1, str); + + int numclear = 1; + 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]; + snprintf (str, MXSTR, "ev %5d free event buffer, nb=%3d", + mBuffer[id].evNum, mBuffer[id].nBoard); + factOut (kDebug, -1, str); + 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); + } + + xwait.tv_sec = 0; + xwait.tv_nsec = 2000000; // sleep for ~2 msec + nanosleep (&xwait, NULL); + } + } + + if (gi_reset > 0) { + if (gi_resetW > 0) { + CloseRunFile (0, 0, 0); //ask all Runs to be closed + } + if (gi_resetX > 0) { + xwait.tv_sec = gi_resetX; + xwait.tv_nsec = 0; + nanosleep (&xwait, NULL); + } + + snprintf (str, MXSTR, "Continue read Process ..."); + factOut (kInfo, -1, str); + gi_reset = 0; + goto START; + } + + + + snprintf (str, MXSTR, "Exit read Process ..."); + factOut (kInfo, -1, str); + gi_runStat = -99; + gj.readStat = -99; + factStat (gj); + factStatNew (gi); + return 0; } /*-----------------------------------------------------------------*/ -void *subProc( void *thrid ) { - int threadID,status,numWait,numProc,kd,k1,k0,k,jret; - struct timespec xwait ; - - threadID= (int) thrid; - - snprintf(str,MXSTR,"Starting sub-process-thread %d",threadID); - factOut(kInfo,-1, str ) ; - - 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) { - snprintf(str,MXSTR,"process %d wants to send to process %d",threadID,jret) ; - factOut(kError,-1, str ) ; - 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 - snprintf(str,MXSTR,"Exit subProcessing Process %d",threadID); - factOut(kInfo,-1, str ) ; - return 0 ; - } - if (numProc == 0) { - //seems we have nothing to do, so sleep a little - xwait.tv_sec = 0; - xwait.tv_nsec= 2000000 ; // sleep for ~2 msec - nanosleep( &xwait , NULL ) ; - } - } - - snprintf(str,MXSTR,"Ending sub-process-thread %d",threadID); - factOut(kInfo,-1, str ) ; - return ; +void * +subProc (void *thrid) +{ + int threadID, status, numWait, numProc, kd, k1, k0, k, jret; + struct timespec xwait; + + threadID = (int) thrid; + + snprintf (str, MXSTR, "Starting sub-process-thread %d", threadID); + factOut (kInfo, -1, str); + + 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) { + snprintf (str, MXSTR, + "process %d wants to send to process %d", + threadID, jret); + factOut (kError, -1, str); + 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 + snprintf (str, MXSTR, "Exit subProcessing Process %d", threadID); + factOut (kInfo, -1, str); + return 0; + } + if (numProc == 0) { + //seems we have nothing to do, so sleep a little + xwait.tv_sec = 0; + xwait.tv_nsec = 2000000; // sleep for ~2 msec + nanosleep (&xwait, NULL); + } + } + + snprintf (str, MXSTR, "Ending sub-process-thread %d", threadID); + factOut (kInfo, -1, str); + return; } /*-----------------------------------------------------------------*/ -void *procEvt( void *ptr ) { +void * +procEvt (void *ptr) +{ /* *** main loop processing file, including SW-trigger */ - int numProc, numWait ; - int k, status ; - struct timespec xwait ; - char str[MXSTR] ; - - - - - cpu_set_t mask; - int cpu = 1 ; //process thread (will be several in final version) - - snprintf(str,MXSTR,"Starting process-thread with %d subprocess",gi_maxProc); - factOut(kInfo,-1, str ) ; + int numProc, numWait; + int k, status, j; + struct timespec xwait; + char str[MXSTR]; + + + + int lastRun = 0; //usually run from last event still valid + + cpu_set_t mask; + int cpu = 1; //process thread (will be several in final version) + + snprintf (str, MXSTR, "Starting process-thread with %d subprocess", + gi_maxProc); + factOut (kInfo, -1, str); /* CPU_ZERO initializes all the bits in the mask to zero. */ - CPU_ZERO( &mask ); + CPU_ZERO (&mask); /* CPU_SET sets only the bit corresponding to cpu. */ // CPU_SET( 0 , &mask ); leave for system // CPU_SET( 1 , &mask ); used by write process - CPU_SET( 2 , &mask ); - CPU_SET( 3 , &mask ); - CPU_SET( 4 , &mask ); - CPU_SET( 5 , &mask ); - CPU_SET( 6 , &mask ); + CPU_SET (2, &mask); + CPU_SET (3, &mask); + CPU_SET (4, &mask); + CPU_SET (5, &mask); + CPU_SET (6, &mask); // CPU_SET( 7 , &mask ); used by read process /* sched_setaffinity returns 0 in success */ - if ( sched_setaffinity( 0, sizeof(mask), &mask ) == -1 ) { - snprintf(str,MXSTR,"P ---> can not create affinity to %d",cpu); - factOut(kWarn,-1, str ) ; - } - - - pthread_t thread[100] ; - int th_ret[100]; - - for (k=0; k < gi_maxProc; k++) { - th_ret[k] = pthread_create( &thread[k], NULL, subProc, (void *)k ); - } - - 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 (sched_setaffinity (0, sizeof (mask), &mask) == -1) { + snprintf (str, MXSTR, "P ---> can not create affinity to %d", cpu); + factOut (kWarn, -1, str); + } + + + pthread_t thread[100]; + int th_ret[100]; + + for (k = 0; k < gi_maxProc; k++) { + th_ret[k] = pthread_create (&thread[k], NULL, subProc, (void *) k); + } + + 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); //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 - evtCtrl.evtStat[k0] = 9991 ; - } else { - - int id = evtCtrl.evtBuf[k0] ; - int ievt = mBuffer[id].evNum ; - int itevt= mBuffer[id].trgNum ; - int itrg = mBuffer[id].trgTyp ; - int roi = mBuffer[id].nRoi ; - int roiTM= mBuffer[id].nRoiTM ; -// uint32_t irun = mBuffer[id].runNum ; -//snprintf(str,MXSTR,"P processing %d %d %d %d",ievt,k,id,evtCtrl.evtStat[k0]) ; -//factOut(kDebug,-1, str ) ; + if (evtCtrl.evtStat[k0] > 90 && evtCtrl.evtStat[k0] < 1000) { + + if (gi_resetR > 1) { //we are asked to flush buffers asap + 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) { + snprintf (str, MXSTR, + "P error: can not find run %d for event %d in %d", + irun, ievt, id); + factOut (kFatal, 901, str); + } + 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 .... + snprintf (str, MXSTR, + "P finish 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; + 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) { + snprintf (str, MXSTR, + "P could not open a file for run %d", irun); + factOut (kError, 502, str); + runCtrl[j].fileId = 91; + runCtrl[j].procId = 91; + } else { + snprintf (str, MXSTR, "P opened new run_file %d evt %d", + irun, ievt); + factOut (kInfo, -1, str); + 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) { + snprintf (str, MXSTR, + "P reached end of run condition for run %d", + irun); + factOut (kInfo, 502, str); + runFinish1 (runCtrl[j].runId); + runCtrl[j].procId = 1; + } + } + if (runCtrl[j].procId != 0) { + snprintf (str, MXSTR, + "P skip event %d because no active run %d", ievt, + irun); + factOut (kInfo, 502, str); + evtCtrl.evtStat[k0] = 9091; + } else { +//-------- +//-------- + int 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 - if (roiTM == roi) roiTM=0 ; - int ip,it,dest,ib; - for (ip=0; ipStartPix[ip] == -1 ) { - dest= ip*roi ; - bzero( &mBuffer[id].fEvent->Adc_Data[dest], roi*2) ; - } - } - for (it=0; itStartTM[it] == -1 ) { - dest= it*roi + NPIX*roi ; - bzero( &mBuffer[id].fEvent->Adc_Data[dest], roi*2) ; - } - } + 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; + bzero (&mBuffer[id].fEvent->Adc_Data[dest], roi * 2); + } + } + for (it = 0; it < NTMARK; it++) { + if (mBuffer[id].fEvent->StartTM[it] == -1) { + dest = it * roi + NPIX * roi; + bzero (&mBuffer[id].fEvent->Adc_Data[dest], 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; ibBoardTime[ib] = 0 ; - } else { - mBuffer[id].fEvent->BoardTime[ib] = - ntohl(mBuffer[id].FADhead[ib].time) ; - } - } - - int i=eventCheck(mBuffer[id].runNum, mBuffer[id].FADhead, - mBuffer[id].fEvent) ; -// gj.procEvt++ ; - gi.procTot++ ; - numProc++ ; - - if (i<0) { - evtCtrl.evtStat[k0] = 9999 ; //flag event to be skipped - gi.procErr++ ; - } else { - evtCtrl.evtStat[k0] = 1000 ; - } + 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] = + ntohl (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++; + } + } + } + } else if (evtCtrl.evtStat[k0] >= 0 && evtCtrl.evtStat[k0] < 90) { + numWait++; } - } else if ( evtCtrl.evtStat[k0] >=0 && evtCtrl.evtStat[k0] < 90 ) { - numWait++ ; - } - } - - 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 ; - } - - if (numProc == 0) { - //seems we have nothing to do, so sleep a little - xwait.tv_sec = 0; - xwait.tv_nsec= 2000000 ; // sleep for ~2 msec - nanosleep( &xwait , NULL ) ; - } - 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... - - snprintf(str,MXSTR,"Abort Processing Process ..."); - factOut(kInfo,-1, str ) ; - int kd = evtCtrl.lastPtr - evtCtrl.frstPtr ; - if ( kd < 0 ) kd+= (MAX_EVT*MAX_RUN) ; - - for (k=0; k=0 && evtCtrl.evtStat[k0] <1000) { - evtCtrl.evtStat[k0] = 9800 ; //flag event as 'processed' - } - } - - gp_runStat = -99 ; - gj.procStat= -99 ; - - return 0; - + } + + 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; + } + + if (numProc == 0) { + //seems we have nothing to do, so sleep a little + xwait.tv_sec = 0; + xwait.tv_nsec = 2000000; // sleep for ~2 msec + nanosleep (&xwait, NULL); + } + 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... + + snprintf (str, MXSTR, "Abort Processing Process ..."); + factOut (kInfo, -1, str); + int kd = evtCtrl.lastPtr - evtCtrl.frstPtr; + if (kd < 0) + kd += (MAX_EVT * MAX_RUN); + + for (k = 0; k < gi_maxProc; k++) { + pthread_join (thread[k], (void **) &status); + } + + 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] < 1000) { + evtCtrl.evtStat[k0] = 9800; //flag event as 'processed' + } + } + + gp_runStat = -99; + gj.procStat = -99; + + return 0; + } /*-----------------------------------------------------------------*/ -int CloseRunFile(uint32_t runId, uint32_t closeTime, uint32_t maxEvt) { +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 j ; - - - if ( runId == 0 ) { - for ( j=0; j can not create affinity to %d",cpu); - } - - 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) ; + if (sched_setaffinity (0, sizeof (mask), &mask) == -1) { + snprintf (str, MXSTR, "W ---> can not create affinity to %d", cpu); + } + + 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] < 9900) { - - if ( gi_resetR > 1 ) { //we must drain the buffer asap - evtCtrl.evtStat[k0] = 9904 ; + if (evtCtrl.evtStat[k0] > 5000 && evtCtrl.evtStat[k0] < 9900) { + + if (gi_resetR > 1) { //we must drain the buffer asap + 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; + } 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) { + snprintf (str, MXSTR, + "W error: can not find run %d for event %d in %d", + irun, ievt, id); + factOut (kFatal, 901, str); + 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; + 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) { + snprintf (str, MXSTR, + "W could not open a file for run %d", irun); + factOut (kError, 502, str); + runCtrl[j].fileId = 91; + } else { + snprintf (str, MXSTR, "W opened new run_file %d evt %d", + irun, ievt); + factOut (kInfo, -1, str); + runCtrl[j].fileId = 0; + } + + } + + if (runCtrl[j].fileId != 0) { + if (runCtrl[j].fileId < 0) { + snprintf (str, MXSTR, + "W never opened file for this run %d", irun); + factOut (kError, 123, str); + } else if (runCtrl[j].fileId < 100) { + snprintf (str, MXSTR, "W.file for this run is closed %d", + irun); + factOut (kWarn, 123, str); + runCtrl[j].fileId += 100; + } else { + snprintf (str, MXSTR, "W:file for this run is closed %d", + irun); + factOut (kDebug, 123, str); + } + evtCtrl.evtStat[k0] = 9903; + gi.wrtErr++; + } else { + 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] = 9901; + 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] = 9902; + gi.wrtErr++; + } + + if (i < 0 + || runCtrl[j].lastTime < g_actTime - 300 + || runCtrl[j].closeTime < g_actTime + || 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; + + if (runCtrl[j].procId == 0) { + runFinish1 (runCtrl[j].runId); + runCtrl[j].procId = 92; + } + + runCtrl[j].closeTime = g_actTime - 1; + i = runClose (runCtrl[j].fileHd, &runTail[j], + sizeof (runTail[j])); + if (i < 0) { + snprintf (str, MXSTR, "error closing run %d %d AAA", + runCtrl[j].runId, i); + factOut (kError, 503, str); + runCtrl[j].fileId = 92; + } else { + snprintf (str, MXSTR, "W closed run %d AAA %d", irun, + ii); + factOut (kInfo, 503, str); + runCtrl[j].fileId = 93; + } + } + } + } + } else if (evtCtrl.evtStat[k0] > 0 && evtCtrl.evtStat[k0] < 9000) + numWait++; + } + + //check if we should close a run (mainly when no event pending) + for (j = 0; j < MAX_RUN; j++) { + if (runCtrl[j].fileId == 0 + && (runCtrl[j].closeTime < g_actTime + || runCtrl[j].lastTime < g_actTime - 300 + || runCtrl[j].maxEvt <= runCtrl[j].actEvt)) { + if (runCtrl[j].runId == gi_myRun) + gi_myRun = g_actTime; + int ii = 0; + 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; + + if (runCtrl[j].procId == 0) { + runFinish1 (runCtrl[j].runId); + runCtrl[j].procId = 92; + } + + runCtrl[j].closeTime = g_actTime - 1; + i = runClose (runCtrl[j].fileHd, &runTail[j], + sizeof (runTail[j])); + if (i < 0) { + snprintf (str, MXSTR, "error closing run %d %d BBB", + runCtrl[j].runId, i); + factOut (kError, 506, str); + runCtrl[j].fileId = 94; + } else { + snprintf (str, MXSTR, "W closed run %d BBB %d", + runCtrl[j].runId, ii); + factOut (kInfo, 507, str); + runCtrl[j].fileId = 95; + } + } + } + + if (numWrite == 0) { + //seems we have nothing to do, so sleep a little + xwait.tv_sec = 0; + xwait.tv_nsec = 2000000; // sleep for ~2 msec + nanosleep (&xwait, NULL); + } + + if (gj.readStat < -10 && numWait == 0) { //nothing left to do + snprintf (str, MXSTR, "Finish Write Process ..."); + factOut (kInfo, -1, str); + gw_runStat = -22; //==> we should exit + gj.writStat = -22; //==> we should exit + goto closerun; + } + gw_runStat = gi_runStat; + gj.writStat = gj.readStat; + + } + + //must close all open files .... + snprintf (str, MXSTR, "Abort Writing Process ..."); + factOut (kInfo, -1, str); + + closerun: + snprintf (str, MXSTR, "Close all open files ..."); + factOut (kInfo, -1, str); + for (j = 0; j < MAX_RUN; j++) + if (runCtrl[j].fileId == 0) { + if (runCtrl[j].runId == gi_myRun) + gi_myRun = g_actTime; + + if (runCtrl[j].procId == 0) { + runFinish (runCtrl[j].runId); + runCtrl[j].procId = 92; + } + + runCtrl[j].closeTime = g_actTime - 1; + i = runClose (runCtrl[j].fileHd, &runTail[j], sizeof (runTail[j])); + int ii = 0; + 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; + if (i < 0) { + snprintf (str, MXSTR, "error closing run %d %d CCC", + runCtrl[j].runId, i); + factOut (kError, 506, str); + runCtrl[j].fileId = 96; } else { - - - int id = evtCtrl.evtBuf[k0] ; - uint32_t irun = mBuffer[id].runNum ; - int ievt = mBuffer[id].evNum ; - - gi.wrtTot++ ; - if (runCtrl[lastRun].runId == irun ) { - j = lastRun ; - } else { - //check which fileID to use (or open if needed) - for ( j=0; j= MAX_RUN ) { - snprintf(str,MXSTR,"W error: can not find run %d for event %d in %d", irun,ievt,id); - factOut(kFatal,901, str ) ; - 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 ; - 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 ) { - snprintf(str,MXSTR,"W could not open a file for run %d",irun); - factOut(kError,502, str ) ; - runCtrl[j].fileId = 91 ; - } else { - snprintf(str,MXSTR,"W opened new run_file %d evt %d",irun,ievt) ; - factOut(kInfo,-1, str ) ; - runCtrl[j].fileId = 0 ; - } - - } - - if (runCtrl[j].fileId != 0 ) { - if (runCtrl[j].fileId < 0 ) { - snprintf(str,MXSTR,"W never opened file for this run %d",irun) ; - factOut(kError,123,str) ; - } else if (runCtrl[j].fileId < 100 ) { - snprintf(str,MXSTR,"W.file for this run is closed %d",irun) ; - factOut(kWarn,123,str) ; - runCtrl[j].fileId += 100 ; - } else { - snprintf(str,MXSTR,"W:file for this run is closed %d",irun) ; - factOut(kDebug,123,str) ; - } - evtCtrl.evtStat[k0] = 9903 ; - gi.wrtErr++ ; - } else { - 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] = 9901 ; - 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] = 9902 ; - gi.wrtErr++ ; - } - - if ( i < 0 - || runCtrl[j].lastTime < g_actTime-300 - || runCtrl[j].closeTime < g_actTime - || 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,"error closing run %d %d AAA",runCtrl[j].runId,i) ; - factOut(kError,503, str ) ; - runCtrl[j].fileId = 92 ; - } else { - snprintf(str,MXSTR,"W closed run %d AAA %d",irun,ii) ; - factOut(kInfo,503, str ) ; - runCtrl[j].fileId = 93 ; - } - } - } + snprintf (str, MXSTR, "W closed run %d CCC %d", runCtrl[j].runId, + ii); + factOut (kInfo, 507, str); + runCtrl[j].fileId = 97; } - } else if (evtCtrl.evtStat[k0] > 0 - && evtCtrl.evtStat[k0] < 9000 ) numWait++ ; - } - - //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 ; - - } - - //must close all open files .... - snprintf(str,MXSTR,"Abort Writing Process ..."); - factOut(kInfo,-1, str ) ; - -closerun: - snprintf(str,MXSTR,"Close all open files ..."); - factOut(kInfo,-1, str ) ; - for ( j=0; j90) { - snprintf(str,MXSTR,"illegal number of processes %d",gi_maxProc ) ; - factOut(kFatal,301, str ) ; - gi_maxProc=1; - } - + gi_maxSize = g_maxSize; + if (gi_maxSize <= 0) + gi_maxSize = 1; + + gi_maxProc = g_maxProc; + if (gi_maxProc <= 0 || gi_maxProc > 90) { + snprintf (str, MXSTR, "illegal number of processes %d", gi_maxProc); + factOut (kFatal, 301, str); + gi_maxProc = 1; + } //partially initialize event control logics - evtCtrl.frstPtr = 0 ; - evtCtrl.lastPtr = 0 ; + evtCtrl.frstPtr = 0; + evtCtrl.lastPtr = 0; //start all threads (more to come) when we are allowed to .... - while (g_runStat == 0 ) { - xwait.tv_sec = 0; - xwait.tv_nsec= 10000000 ; // sleep for ~10 msec - nanosleep( &xwait , NULL ) ; - } - - i=0 ; - th_ret[i] = pthread_create( &thread[i], NULL, readFAD, NULL ); - i++; - th_ret[i] = pthread_create( &thread[i], NULL, procEvt, NULL ); - i++; - th_ret[i] = pthread_create( &thread[i], NULL, writeEvt, NULL ); - i++; - imax=i ; + while (g_runStat == 0) { + xwait.tv_sec = 0; + xwait.tv_nsec = 10000000; // sleep for ~10 msec + nanosleep (&xwait, NULL); + } + + i = 0; + th_ret[i] = pthread_create (&thread[i], NULL, readFAD, NULL); + i++; + th_ret[i] = pthread_create (&thread[i], NULL, procEvt, NULL); + i++; + th_ret[i] = pthread_create (&thread[i], NULL, writeEvt, NULL); + i++; + imax = i; #ifdef BILAND - xwait.tv_sec = 30;; - xwait.tv_nsec= 0 ; // sleep for ~20sec - nanosleep( &xwait , NULL ) ; - - printf("close all runs in 2 seconds\n"); - - CloseRunFile( 0, time(NULL)+2, 0) ; - - xwait.tv_sec = 1;; - xwait.tv_nsec= 0 ; // sleep for ~20sec - nanosleep( &xwait , NULL ) ; - - printf("setting g_runstat to -1\n"); - - g_runStat = -1 ; + xwait.tv_sec = 30;; + xwait.tv_nsec = 0; // sleep for ~20sec + nanosleep (&xwait, NULL); + + printf ("close all runs in 2 seconds\n"); + + CloseRunFile (0, time (NULL) + 2, 0); + + xwait.tv_sec = 1;; + xwait.tv_nsec = 0; // sleep for ~20sec + nanosleep (&xwait, NULL); + + printf ("setting g_runstat to -1\n"); + + g_runStat = -1; #endif //wait for all threads to finish - for (i=0; i0 closed + int16_t fileId ; //<0 never opened, 0=open, >0 closed + int16_t procId ; //processing <0 never opened, 0=open, >0 closed int16_t roi0 ; //roi for normal pixels int16_t roi8 ; //roi for pixels8