Index: /trunk/FACT++/src/EventBuilder.c =================================================================== --- /trunk/FACT++/src/EventBuilder.c (revision 11133) +++ /trunk/FACT++/src/EventBuilder.c (revision 11134) @@ -40,8 +40,11 @@ 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, const char* message ) ; -extern void factStat(int severity, int err, const char* message ) ; +extern void factOut(int severity, int err, char* message ) ; +extern void factStat(int64_t *array , int len ) ; extern int eventCheck( PEVNT_HEADER *fadhd, EVENT *event) ; + + +extern void debugHead(int i, void *buf); extern void debugRead(int isock, int ibyte, int32_t event, int state, @@ -118,4 +121,5 @@ 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 @@ -271,5 +275,5 @@ snprintf(str,MXSTR,"illegal nRoi: %d",nRoi) ; factOut(kError, 1, str ) ; - return 99 ; + return -9999 ; } @@ -289,5 +293,5 @@ //event does not yet exist; create if (evFree < 0 ) { //no space available in ctrl - snprintf(str,MXSTR,"no control slot to keep event...") ; + snprintf(str,MXSTR,"no control slot to keep event %d",evID) ; factOut(kError,881, str ) ; return -1 ; @@ -301,11 +305,13 @@ if ( gi_usedMem + needmem + headmem > g_maxMem) { - snprintf(str,MXSTR,"no memory left to keep event...") ; + snprintf(str,MXSTR,"no memory left to keep event %d",evID) ; factOut(kError,882, str ) ; - return -1 ; + 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; } @@ -313,4 +319,6 @@ 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 ; @@ -367,5 +375,5 @@ if (evFree <0 ) { snprintf(str,MXSTR,"not able to register the new run %d",runID); - factOut(kError,883, str ) ; + factOut(kFatal,883, str ) ; } else { runCtrl[evFree].runId = runID ; @@ -416,6 +424,6 @@ void *readFAD( void *ptr ) { /* *** main loop reading FAD data and sorting them to complete events */ - int head_len,frst_len,numok,numok2,dest,evID,i,j,k ; - int actBoards = 0; + int head_len,frst_len,numok,numok2,dest,evID,i,k ; + int actBoards = 0, minLen ; int32_t jrd ; int32_t myRun ; @@ -484,6 +492,8 @@ head_len = sizeof(PEVNT_HEADER) ; //frst_len = head_len + 36 * 12 ; //fad_header plus 36*pix_header - frst_len = head_len ; //fad_header only, so each event must be longer, even for roi=0 - + frst_len = head_len ; //max #bytes to read first: fad_header only, so each event must be longer, even for roi=0 + +//minLen = MIN_LEN ; //min #bytes needed to check header + minLen = head_len ; //min #bytes needed to check header: full header for debug numok = numok2 = 0 ; @@ -540,4 +550,6 @@ for (i=0; i 0) s0=+1 ; + else s0=-1 ; gettimeofday( tv, NULL); @@ -553,8 +565,9 @@ rd[i].bufLen = frst_len ; // max size to read at begining } else { - rd[i].bufTyp = -1 ; // data to be skipped + rd[i].bufTyp = -1 ; // full data to be skipped rd[i].bufLen = sizeof(CNV_FACT) ; //huge for skipping } rd[i].bufPos = 0 ; // no byte read so far + rd[i].skip = 0 ; // start empty gi_NumConnect[ b ]++ ; numok++ ; //make sure next round will execute @@ -564,29 +577,33 @@ } - if (rd[i].sockStat ==0) { //we have a connection ==> try to read - numok++ ; - sokCnt[i]++; - jrd=recv(rd[i].socket,&rd[i].rBuf->B[ rd[i].bufPos], rd[i].bufLen, MSG_DONTWAIT); - - if (jrd >0 ) { - qread+=jrd ; - debugStream(i,&rd[i].rBuf->B[ rd[i].bufPos],jrd) ; - } - - if (jrd == 0) { //connection has closed ... - snprintf(str,MXSTR,"Socket %d closed by FAD",i); - factOut(kInfo,441, str ) ; - GenSock(1, i, 0,NULL, &rd[i]) ; - gi_ErrCnt[i]++ ; - gi_NumConnect[ 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_ErrCnt[i]++ ; - } else numok-- ; //else nothing waiting to be read - - } else if ( rd[i].bufTyp <0 ) { // we are skipping this board ... + 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++ ; + sokCnt[i]++; + jrd=recv(rd[i].socket,&rd[i].rBuf->B[ rd[i].bufPos], rd[i].bufLen, MSG_DONTWAIT); + + if (jrd >0 ) { + qread+=jrd ; + debugStream(i,&rd[i].rBuf->B[ rd[i].bufPos],jrd) ; + } + + 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_ErrCnt[i]++ ; + gi_NumConnect[ 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_ErrCnt[i]++ ; + } else numok-- ; //else nothing waiting to be read + jrd = 0 ; + } + } else jrd = 0 ; //did read nothing as requested + + if ( rd[i].bufTyp <0 ) { // we are skipping this board ... // just do nothing @@ -597,5 +614,5 @@ debugRead(i,jrd,rd[i].evtID, 0,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; 0=reading data } else { //full dataset read - rd[i].bufLen = rd[i].bufPos + j ; + rd[i].bufLen = 0 ; rd[i].bufPos = rd[i].fadLen ; if ( rd[i].rBuf->B[ rd[i].bufPos-1] != stop.B[0] @@ -616,9 +633,12 @@ evID = mBufEvt( rd[i].evtID, rd[i].runID, roi ) ; + if (evID <-9000) goto EndBuf ; //illegal event, skip it ... if (evID < 0) { - snprintf(str,MXSTR,"no space left ...%d",evID) ; - factOut(kError,201, str ) ; - goto EndBuf ; //--> skip event (and hope it will improve) + xwait.tv_sec = 0; + xwait.tv_nsec= 20000000 ; // sleep for ~20 msec + nanosleep( &xwait , NULL ) ; + goto EndBuf1 ; //hope there is free space next round } + //we have a valid entry in mBuffer[]; fill it @@ -704,4 +724,6 @@ rd[i].bufLen = frst_len ; rd[i].bufPos = 0 ; +EndBuf1: + ; } @@ -709,5 +731,5 @@ rd[i].bufPos += jrd ; rd[i].bufLen -= jrd ; - if ( rd[i].bufPos > MIN_LEN ){ //sufficient data to take action + if ( rd[i].bufPos >= minLen ){ //sufficient data to take action //check if startflag correct; else shift block .... for (k=0; kB[k+1] == start.B[0] ) break ; } - - - 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 - snprintf(str,MXSTR,"no start of header on port%d", i ) ; - factOut(kWarn,666, str ) ; +// snprintf(str,MXSTR,"no start of header on port%d", i ) ; +// factOut(kWarn,666, str ) ; rd[i].bufPos = 0 ; @@ -733,5 +750,10 @@ memcpy(&rd[i].rBuf->B[0], &rd[i].rBuf->B[k], rd[i].bufPos ) ; } - if ( rd[i].bufPos > MIN_LEN ) { + 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 ; @@ -739,8 +761,9 @@ rd[i].evtID = ntohl(rd[i].rBuf->I[4]) ; //(FADevt) rd[i].runID = ntohl(rd[i].rBuf->I[11]) ; -if (rd[i].runID ==0 ) rd[i].runID = myRun ; + if (rd[i].runID ==0 ) rd[i].runID = myRun ; rd[i].bufTyp = 1 ; //ready to read full record rd[i].bufLen = rd[i].fadLen - rd[i].bufPos ; if (rd[i].bufLen <=0 ) rd[i].bufLen = 100000 ; //? + debugHead(i,rd[i].rBuf); debugRead(i,jrd,rd[i].evtID,-1,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid;-1=start event } else { @@ -759,5 +782,4 @@ if ( g_actTime > gi_SecTime ) { gi_SecTime = g_actTime ; -// PrintRate() ; @@ -807,7 +829,14 @@ int ib ; for (ib=0; ib0) - gi_NumConnect[ i/7 ]-- ; + gi_NumConnect[ i/7 ]-- ; } @@ -973,5 +1001,5 @@ } else { mBuffer[id].fEvent->BoardTime[ib] = - ntohl(mBuffer[id].FADhead[ib].time) ; + ntohl(mBuffer[id].FADhead[ib].time) ; } } @@ -1364,6 +1392,6 @@ - /* + FileHandle_t runOpen(uint32_t irun, RUN_HEAD *runhd, size_t len ) { return 1; } ; @@ -1389,6 +1417,7 @@ -void factStat(int severity, int err, char* message ) { - printf("%3d %3d : %s\n",severity,err,message) ; +void factStat(int64_t *array, int len ) { + printf("stat: bfr%5lu skp%4lu free%4lu (tot%7lu) mem%12lu rd%12lu %3lu\n", + array[0],array[1],array[2],array[3],array[4],array[5],array[6]); } @@ -1401,4 +1430,7 @@ void debugStream(int isock, void *buf, int len) { +} + +void debugHead(int i, void *buf) { } @@ -1450,5 +1482,5 @@ //} // -//version for PC-test +//version for PC-test for (c=0; c<4; c++) { for (b=0; b<10; b++) { @@ -1478,3 +1510,3 @@ } -*/ +*/