// // // #define EVTDEBUG #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include "EventBuilder.h" 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 }; #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( PEVNT_HEADER *fadhd, EVENT *event) ; 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); 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 ; // 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 ; //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 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] ; /* *** Definition of rdBuffer to read in IP packets; keep it global !!!! */ 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) #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 /*-----------------------------------------------------------------*/ /*-----------------------------------------------------------------*/ int GenSock(int flag, int sid, int port, struct sockaddr_in *sockAddr, READ_STRUCT *rd) { /* *** generate Address, create sockets and allocates readbuffer for it *** *** if flag==0 generate socket and buffer *** <0 destroy socket and buffer *** >0 close and redo socket *** *** sid : board*7 + port id */ int j ; 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 #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 ; #ifdef EVTDEBUG 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 ; } snprintf(str,MXSTR,"Successfully generated socket %d ",sid); factOut(kInfo,773, str ) ; rd->sockStat = -1 ; //try to (re)open socket return 0 ; } /*-----------------------------------------------------------------*/ /*-----------------------------------------------------------------*/ 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 use this slot .... 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 ; //check if runId already registered in runCtrl evFree = -1 ; for (k=0; k g_maxMem) { gj.maxMem = gj.usdMem + needmem + headmem ; 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 ) ; 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].fEvent) ; mBuffer[i].fEvent = NULL ; return -22; } //flag all boards as unused mBuffer[i].nBoard = 0 ; for (k=0; kStartPix[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 ; mBuffer[i].Errors = 0 ; gj.usdMem += needmem + headmem; 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 ; } /*-----------------------------------------------------------------*/ 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 ; headmem = NBOARDS* sizeof(PEVNT_HEADER) ; mBuffer[i].evNum = mBuffer[i].nRoi= -1; mBuffer[i].runNum = 0; gj.usdMem = gj.usdMem - freemem - headmem; 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; /* 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) ; #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 ) ; #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 ... // just do nothing #ifdef EVTDEBUG 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 ; #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 ) ; #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 #ifdef EVTDEBUG 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 #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 ) ; } } } #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 ) ; #ifdef EVTDEBUG memcpy(&rd[i].xBuf->B[0], &rd[i].xBuf->B[k], rd[i].bufPos ) ; #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].evtID = ntohl(rd[i].rBuf->I[4]) ; //(FADevt) rd[i].ftmID = ntohl(rd[i].rBuf->I[5]) ; //(FTMevt) 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 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].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) ; //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] >= 900 //'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; } /*-----------------------------------------------------------------*/ void *procEvt( void *ptr ) { /* *** main loop processing file, including SW-trigger */ int numProc, numWait ; int k ; struct timespec xwait ; char str[MXSTR] ; cpu_set_t mask; int cpu = 5 ; //process thread (will be several in final version) snprintf(str,MXSTR,"Starting process-thread"); factOut(kInfo,-1, str ) ; /* CPU_ZERO initializes all the bits in the mask to zero. */ CPU_ZERO( &mask ); /* CPU_SET sets only the bit corresponding to cpu. */ CPU_SET( cpu, &mask ); /* 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 ) ; } 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] <500) { if ( gi_resetR > 1 ) { //we are asked to flush buffers asap evtCtrl.evtStat[k0] = 991 ; } 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 ; int Errors=mBuffer[id].Errors ; // 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 ) ; //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) ; } } //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 = Errors ; 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].FADhead,mBuffer[id].fEvent) ; // gj.procEvt++ ; gi.procTot++ ; numProc++ ; if (i<0) { evtCtrl.evtStat[k0] = 999 ; //flag event to be skipped gi.procErr++ ; } else { evtCtrl.evtStat[k0] = 520 ; } } } 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) ; 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] <500) { evtCtrl.evtStat[k0] = 555 ; //flag event as 'processed' } } gp_runStat = -99 ; gj.procStat= -99 ; return 0; } /*-----------------------------------------------------------------*/ 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) ; //would be better to use bitmaps for evtStat (allow '&' instead of multi-if) if (evtCtrl.evtStat[k0] > 500 && evtCtrl.evtStat[k0] < 900) { if ( gi_resetR > 1 ) { //we must drain the buffer asap evtCtrl.evtStat[k0] = 904 ; } 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] = 903 ; 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] = 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] = 902 ; 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 ; } } } } } else if (evtCtrl.evtStat[k0] > 0 && evtCtrl.evtStat[k0] < 900 ) 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; j