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

Timestamp:

07/11/11 11:08:16 (13 years ago)

Author:

tbretz

Message:

Added g_reset and some other updates and minor fixes.

File:

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

trunk/FACT++/src/EventBuilder.c

@@ -54 +54 @@
 extern void debugStream(int isock, void *buf, int len) ;
 
-
-
-int g_actTime   ;
-int g_runStat   ;
+int CloseRunFile(uint32_t runId, uint32_t closeTime, uint32_t maxEvt);
+
+
+uint g_actTime   ;
+int  g_runStat   ;
+int  g_reset     ;
 size_t g_maxMem  ;  //maximum memory allowed for buffer
 
@@ -251 +253 @@
       evtCtrl.evtStat[ i] = -1 ;
       evtCtrl.pcTime[  i] = actime ;  //initiate to far future
-
    }
 
@@ -488 +489 @@
   int actBoards = 0, minLen ;
   int32_t jrd ;
+  uint gi_SecTime ;        //time in seconds
   int boardId, roi,drs,px,src,pixS,pixH,pixC,pixR,tmS ;
 
   int goodhed = 0 ;
   int errcnt0 = 0 ;
+
+  int reset, resetR, resetS, resetW, resetX ;
+  int sockDef[NBOARDS];    //internal state of sockets
+
+
 
   struct timespec xwait ;
@@ -508 +515 @@
 
 /* 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 = 7 ;
-   CPU_SET( cpu, &mask );
+  cpu = 7 ;
+  CPU_SET( cpu, &mask );
 
 /* sched_setaffinity returns 0 in success */
-   if ( sched_setaffinity( 0, sizeof(mask), &mask ) == -1 ) {
-      snprintf(str,MXSTR,"W ---> can not create affinity to %d",cpu);
-      factOut(kWarn,-1, str ) ;
-   }
-
-
-  //make sure all sockets are preallocated as 'not exist'
-  for (i=0; i<MAX_SOCK; i++) {
-     rd[i].socket   = -1 ; 
-     rd[i].sockStat = 99 ;
+  if ( sched_setaffinity( 0, sizeof(mask), &mask ) == -1 ) {
+     snprintf(str,MXSTR,"W ---> can not create affinity to %d",cpu);
+     factOut(kWarn,-1, str ) ;
   }
 
-  int sockDef[NBOARDS];    //internal state of sockets
-  for (i=0; i<NBOARDS; i++) sockDef[i]= 0 ; 
-
-
-  resetEvtStat(); 
-  gj.usdMem = gj.maxMem = gj.xxxMem = 0 ;
-  gj.totMem = g_maxMem ;
-  gj.bufNew = gj.bufEvt = 0 ;
-  gj.evtSkip= gj.evtWrite = gj.evtErr = 0 ;
-
-  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;
-  }
-
-
-  uint gi_SecTime ;        //time in seconds
-  gi_SecTime= g_actTime ;
-
-  mBufInit() ;    //initialize buffers
-
-  snprintf(str,MXSTR,"end   initializing");
-  factOut(kInfo,-1, str ) ;
-
-
-
   head_len = sizeof(PEVNT_HEADER) ;
-//frst_len = head_len + 36 * 12 ;   //fad_header plus 36*pix_header
   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 ;
 
   start.S=0xFB01;
   stop.S= 0x04FE;
 
-  gi_myRun = g_actTime ;
+/* initialize run control logics */
+  for (i=0; i<MAX_RUN; i++) {
+     runCtrl[i].runId = 0 ;
+     runCtrl[i].fileId = -2 ;
+  }
+  resetS = resetR = 9; 
+
+
+START:
+  evtCtrl.frstPtr = 0 ;
+  evtCtrl.lastPtr = 0 ;
+
+  gi_myRun   = g_actTime ;
+  gi_SecTime = g_actTime ;
   gi_runStat = g_runStat ;
   gj.readStat= g_runStat ;
-
-
-  while (g_runStat >=0) {           //loop until global variable g_runStat claims stop
+  numok = numok2   = 0 ;
+
+  if ( 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 (i=0; i<NBOARDS; i++) sockDef[i]= 0 ; 
+
+  }
+
+
+  if ( resetR > 0) {
+     resetEvtStat(); 
+     gj.usdMem = gj.maxMem = gj.xxxMem = 0 ;
+     gj.totMem = g_maxMem ;
+     gj.bufNew = gj.bufEvt = 0 ;
+     gj.evtSkip= gj.evtWrite = gj.evtErr = 0 ;
+
+     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;
+     }
+
+     mBufInit() ;    //initialize buffers
+
+     snprintf(str,MXSTR,"end   initializing");
+     factOut(kInfo,-1, str ) ;
+  }
+
+
+  reset = resetR = resetS = 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;
-//  if (gi.totMem < 50000000 ) gi.totMem = 500000000 ;
 
     int b,p,p0,s0,nch; 
@@ -867 +881 @@
 
           if (evtCtrl.evtStat[k0] > 0
-           && evtCtrl.evtStat[k0] < 90 ) {
+           && evtCtrl.evtStat[k0] < 92 ) {    
              gj.bufNew++ ;   //incomplete event in Buffer
-
-             if ( evtCtrl.pcTime[k0] < g_actTime-10 ) {
+             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]);
@@ -881 +895 @@
           } else if (evtCtrl.evtStat[k0] >= 900 ) {
 
-              int id =evtCtrl.evtBuf[k0] ;
-              snprintf(str,MXSTR,"%5d free event buffer (written) %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] >= 900 ) {
+             int id =evtCtrl.evtBuf[k0] ;
+             snprintf(str,MXSTR,"%5d free event buffer (written) %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
           }
@@ -901 +915 @@
 
        int b;
-       for ( b=0; b<NBOARDS; b++) gj.totBytes[k] +=gj.rateBytes[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 ;
@@ -911 +925 @@
        for ( b=0; b<NBOARDS; b++) gj.rateBytes[b] =0 ;
     }
-
-
-
 
     if (numok > 0 ) numok2=0;
@@ -929 +940 @@
  } //and do next loop over all sockets ...
 
- //must quit eventbuilding
- snprintf(str,MXSTR,"stop reading ...");
+
+ snprintf(str,MXSTR,"stop reading ... RESET=%d",g_reset);
  factOut(kInfo,-1, str ) ;
 
- //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++ ;    
+ if (g_reset >0 ) {
+   reset  = g_reset ;
+   resetR = reset%10 ;        //shall we stop reading ?
+   resetS = (reset/10)%10 ;   //shall we close sockets ?
+   resetW = (reset/100)%10 ;  //shall we close files ?
+   resetX = reset/1000 ;      //shall we simply wait resetX seconds ?
+   g_reset= 0;
+ } else {
+   reset  = 0;
+   if ( g_runStat==-1 ) resetR = 1 ;
+   else                 resetR = 7 ;
+   resetS = 7 ; //close all sockets
+   resetW = 7 ; //close all files
+   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 (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 socket   
+          gi_NumConnect[ i/7 ]-- ;
+          gi.numConn[ i/7 ]-- ;
+          gj.numConn[ i/7 ]-- ;
+       }
     }
  }
 
- //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 socket   
-       gi_NumConnect[ i/7 ]-- ;
-       gi.numConn[ i/7 ]-- ;
-       gj.numConn[ i/7 ]-- ;
-    }
- }
-
- xwait.tv_sec = 0;
- xwait.tv_nsec= 2000000 ;  // sleep for ~2 msec
- nanosleep( &xwait , NULL ) ;
- gi_runStat = -11 ;  //inform all that no update to happen any more
- gj.readStat= -11 ;  //inform all that no update to happen any more
-
-
- int minclear = 900 ; //usually wait until writing finished (stat 900)
- if (g_runStat <-1 ) minclear = 0 ;  //in case of abort clear all
-
-
- //and clear all buffers (might have to wait until all others are done)
- snprintf(str,MXSTR,"clear all buffers ...");
- factOut(kInfo,-1, str ) ;
- int numclear=1 ;
- while (numclear > 0 ) {
-    numclear = 0 ;
+
+ if (resetR > 0) {
+    //flag all events as 'read finished'
     int kd = evtCtrl.lastPtr - evtCtrl.frstPtr ;
     if ( kd < 0 ) kd+= (MAX_EVT*MAX_RUN) ;
@@ -984 +987 @@
     for ( k=k1; k<(k1+kd); k++ ) {
        int k0 = k % (MAX_EVT*MAX_RUN) ;
-       if (evtCtrl.evtStat[k0] > minclear ) {
-         int id =evtCtrl.evtBuf[k0] ;
-          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) ;
+       if (evtCtrl.evtStat[k0] > 0 
+        && evtCtrl.evtStat[k0] < 90 ) {
+          evtCtrl.evtStat[k0] = 91 ;    
+          gi.evtSkp++ ;
+          gi.evtTot++ ;    
+       }
     }
 
@@ -997 +998 @@
     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 (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] ;
+             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 (reset > 0) {
+    if (resetW > 0) {
+       CloseRunFile(0,0,0) ; //ask all Runs to be closed
+    }
+    if (resetX > 0) {
+       xwait.tv_sec = resetX;
+       xwait.tv_nsec= 0 ;        
+       nanosleep( &xwait , NULL ) ;
+    }
+
+    snprintf(str,MXSTR,"Continue read Process ...");
+    factOut(kInfo,-1, str ) ;
+    reset = 0 ; 
+    goto START ;
+ }
+
+
 
  snprintf(str,MXSTR,"Exit read Process ...");
@@ -1314 +1368 @@
            if (runCtrl[j].runId == gi_myRun) gi_myRun = g_actTime ;
 int ii =0 ;
-if ( i < 0 ) ii=1 ;
-else if (runCtrl[j].closeTime < g_actTime ) ii=2 ; 
+     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 ; 
@@ -1363 +1416 @@
         i=runClose(runCtrl[j].fileHd, &runTail[j], sizeof(runTail[j]) );
 int ii =0 ;
-if ( i < 0 ) ii=1 ;
-else if (runCtrl[j].closeTime < g_actTime ) ii=2 ; 
+     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 ; 
@@ -1397 +1449 @@
   pthread_t thread[50] ;
   struct timespec xwait ;
-  uint32_t actime ;
 
   gi_runStat = gp_runStat = gw_runStat = 0 ;
   gj.readStat= gj.procStat= gj.writStat= 0 ;
 
-
   snprintf(str,MXSTR,"Starting EventBuilder");
   factOut(kInfo,-1, str ) ;
-
-
-   evtCtrl.frstPtr = 0 ;
-   evtCtrl.lastPtr = 0 ;
-
-   actime = g_actTime + 50000000 ;
-/* initialize run control logics */
-   for (i=0; i<MAX_RUN; i++) {
-      runCtrl[i].runId = 0 ;
-      runCtrl[i].fileId = -2 ;
-   }
-
 
 //start all threads (more to come) when we are allowed to ....
   while (g_runStat == 0 ) {
      xwait.tv_sec = 0;
-     xwait.tv_nsec= 2000000 ;  // sleep for ~2 msec
+     xwait.tv_nsec= 10000000 ;  // sleep for ~10 msec
      nanosleep( &xwait , NULL ) ;
   }
@@ -1434 +1472 @@
   imax=i ;
 
+
 #ifdef BILAND
-     xwait.tv_sec = 20;;
-     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 = 5;;
-     xwait.tv_nsec= 0 ;  // sleep for ~20sec
-     nanosleep( &xwait , NULL ) ;
-
-     printf("setting g_runstat to -1\n");
-
-     g_runStat = -1 ;
+         xwait.tv_sec = 20;;
+         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 = 5;;
+         xwait.tv_nsec= 0 ;  // sleep for ~20sec
+         nanosleep( &xwait , NULL ) ;
+
+         printf("setting g_runstat to -1\n");
+
+         g_runStat = -1 ;
 #endif