Changeset 11134 for trunk/FACT++/src/EventBuilder.c

Timestamp:

06/23/11 15:54:33 (13 years ago)

Author:

tbretz

Message:

Latest Update.

File:

• trunk/FACT++/src/EventBuilder.c (modified) (28 diffs)

trunk/FACT++/src/EventBuilder.c

@@ -40 +40 @@
 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 +121 @@
   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 +275 @@
       snprintf(str,MXSTR,"illegal nRoi: %d",nRoi) ;
       factOut(kError, 1, str ) ;
-      return 99 ;
+      return -9999 ;
    }
 
@@ -289 +293 @@
    //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 +305 @@
 
    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 +319 @@
    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 +375 @@
    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 +424 @@
 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 +492 @@
   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 +550 @@
     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 ;
 
       gettimeofday( tv, NULL);
@@ -553 +565 @@
              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 +577 @@
       }
 
-      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 +614 @@
              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 +633 @@
              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 +724 @@
              rd[i].bufLen = frst_len ;
              rd[i].bufPos = 0 ;         
+EndBuf1:
+             ;
            }
 
@@ -709 +731 @@
            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; k<rd[i].bufPos -1 ; k++) {
@@ -715 +737 @@
                   && rd[i].rBuf->B[k+1] == start.B[0] ) break ;
               }
-
-
-              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
-                 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 +750 @@
                  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 +761 @@
                  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 +782 @@
     if ( g_actTime > gi_SecTime ) {
          gi_SecTime = g_actTime ;
-//       PrintRate() ;
 
 
@@ -807 +829 @@
 int ib ;
 for (ib=0; ib<NBOARDS; ib++) qconn+=gi_NumConnect[ib] ;
-
-snprintf(str,MXSTR,"bfr%5d skp%4d free%4d (tot%7d) mem%9lu rd%10d %3d",qwait,qskip,qdel,qtot,gi_usedMem,qread,qconn);
-factStat(kInfo,-1, str ) ;
+int64_t stat[9] ;
+        stat[0]= qwait;
+        stat[1]= qskip;
+        stat[2]= qdel ;
+        stat[3]= qtot ;
+        stat[4]= gi_usedMem ;
+        stat[5]= qread;
+        stat[6]= qconn;
+
+factStat(stat,7);
 qread=0 ;
     }
@@ -855 +884 @@
  for (i=0; i<MAX_SOCK; i++) {  
     GenSock(-1, i, 0, NULL, &rd[i]) ; //close and destroy socket   
-    if (gi_NumConnect[ i/7 ]>0)
-        gi_NumConnect[ i/7 ]-- ;
+    gi_NumConnect[ i/7 ]-- ;
  }
 
@@ -973 +1001 @@
               } else {
                  mBuffer[id].fEvent->BoardTime[ib] = 
-                 ntohl(mBuffer[id].FADhead[ib].time) ;
+                   ntohl(mBuffer[id].FADhead[ib].time) ;
               }
            }
@@ -1364 +1392 @@
 
   
-
 /*
+
 FileHandle_t  runOpen(uint32_t irun, RUN_HEAD *runhd, size_t len ) 
 { return 1; } ;
@@ -1389 +1417 @@
 
 
-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 +1430 @@
 
 void debugStream(int isock, void *buf, int len) {
+}
+
+void debugHead(int i, void *buf) { 
 }
 
@@ -1450 +1482 @@
 //}
 //
-//version for PC-test
+//version for PC-test 
   for (c=0; c<4; c++) {
      for (b=0; b<10; b++) {
@@ -1478 +1510 @@
 
 }
-*/
+*/