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

Timestamp:

06/21/11 15:40:04 (13 years ago)

Author:

tbretz

Message:

Changed bahaviour of opening closing sockets

File:

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

trunk/FACT++/src/EventBuilder.c

@@ -52 +52 @@
 int g_actTime   =  0 ;
 int g_runStat   = 40 ;
-int g_actBoards = 40 ;
 size_t g_maxMem  ;  //maximum memory allowed for buffer
 
-int     g_maxBoards ;    //maximum number of boards to be initialized
+//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"
 
@@ -64 +67 @@
 
 
-int  gi_maxSocks = 0 ;  
 uint gi_SecRate[MAX_SOCK] ;
 uint gi_S10Rate[MAX_SOCK] ;
@@ -157 +159 @@
 
 
-int GenSock(int flag, int port, struct sockaddr_in *sockAddr,  READ_STRUCT *rd) {
+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 only close and redo the socket
+*** if flag==0 generate socket and buffer
+***         <0 destroy socket and buffer
+***         >0 close and redo socket
+***
+*** sid : board*7 + port id
  */
 
-
-  rd->sockStat = -1 ;
-
-
-  if (flag !=0 ) {
-     close(rd->socket) ;
-     if ( (rd->socket = socket (PF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0)) <= 0) {
-        snprintf(str,MXSTR,"Could not generate socket | %m");
+  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 ) ;
-        return -2 ;
+     } else {
+        snprintf(str,MXSTR,"Succesfully closed socket %d",sid);
+        factOut(kInfo,771, str ) ;
      }
+  }
+
+
+  if (flag < 0) {
+     free(rd->rBuf) ;   //and never open again
+     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;
@@ -183 +198 @@
      rd->SockAddr.sin_addr = sockAddr->sin_addr ;
 
-     if ( (rd->socket = socket (PF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0)) <= 0) {
-        snprintf(str,MXSTR,"Could not generate socket | %m");
-        factOut(kFatal,773, str ) ;
-        return -2 ;
-     } else {
-       rd->rBuf = malloc(sizeof(CNV_FACT) ) ;
-       if ( rd->rBuf == NULL ) {
-          snprintf(str,MXSTR,"Could not create local buffer");
-          factOut(kFatal,774, str ) ;
-          return -3 ;
-       }
+     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 ;
 
@@ -272 +295 @@
 
    
-   needmem = sizeof(EVENT) + NPIX*nRoi*2 + NTMARK*nRoi*2;
+   needmem = sizeof(EVENT) + NPIX*nRoi*2 + NTMARK*nRoi*2 ;
 
    headmem = NBOARDS* sizeof(PEVNT_HEADER) ;
@@ -393 +416 @@
 /* *** 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;
   int32_t jrd ;
   int32_t myRun ;
@@ -405 +429 @@
   int nokCnt[MAX_SOCK],loopCnt=0;
   int sokCnt[MAX_SOCK];
+  int sockDef[NBOARDS];
 
   struct timeval  *tv, atv;
@@ -422 +447 @@
    cpu = 7 ;
    CPU_SET( cpu, &mask );
-// cpu = 6 ;
-// CPU_SET( cpu, &mask );
 
 /* sched_setaffinity returns 0 in success */
@@ -431 +454 @@
    }
 
-
-  gi_maxSocks = 0 ;
 
   //make sure all sockets are preallocated as 'not exist'
@@ -439 +460 @@
      rd[i].sockStat = 99 ;
   }
-
-  int b,p,p0 ;
-  k = 0 ;
-  for (b=0; b<NBOARDS; b++ ) {
-     if ( g_port[b].sockDef >=0 ) {
-        p0=ntohs(g_port[b].sockAddr.sin_port);
-        for (p=p0+1; p<p0+8; p++) {
-           j = GenSock(0,p, &g_port[b].sockAddr, &rd[k]) ;
-           if ( j != 0 ) {
-              snprintf(str,MXSTR,"problem with Address board %d port %d",b,p);
-              factOut(kFatal,101, str ) ;
-           } else {
-              rd[k].board = b ;
-              k++ ;
-              gi_maxSocks++ ;
-           }
-        }
-     }
-  }
+  for (i=0; i<NBOARDS; i++) sockDef[i]= 0 ; 
+
 
   g_actTime = time(NULL) ;
@@ -469 +473 @@
   gi_SecTime= gi_S10Time= gi_MinTime= g_actTime ;
 
-
-
   mBufInit() ;    //initialize buffers
 
@@ -494 +496 @@
 
 
-  while (g_runStat >=0) {           //loop until global variable g_stop is set
+  while (g_runStat >=0) {           //loop until global variable g_runStat claims stop
 
     gi_runStat = g_runStat ;
+
+    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
+          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++ ;
+       }
+    }
+
 
     g_actTime = time(NULL) ;
@@ -505 +537 @@
     numok = 0 ;                       //count number of succesfull actions
 
-    for (i=0; i<gi_maxSocks; i++) {      //check all sockets if something to read
-
-gettimeofday( tv, NULL);
-tsec = atv.tv_sec ;
-tusec= atv.tv_usec ; 
+    for (i=0; i<MAX_SOCK; i++) {         //check all sockets if something to read
+      b = i / 7 ;
+
+      gettimeofday( tv, NULL);
+      tsec = atv.tv_sec ;
+      tusec= atv.tv_usec ; 
 
       if (rd[i].sockStat <0 ) {         //try to connect if not yet done
@@ -515 +548 @@
             (struct sockaddr*) &rd[i].SockAddr, sizeof(rd[i].SockAddr)) ;
         if (rd[i].sockStat ==0 ) {      //successfull ==>
-          rd[i].bufTyp = 0 ;            //  expect a header
-          rd[i].bufLen = frst_len ;     //  max size to read at begining
+          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 ;           //  data to be skipped
+             rd[i].bufLen = sizeof(CNV_FACT) ; //huge for skipping
+          }
           rd[i].bufPos = 0 ;            //  no byte read so far
-          gi_NumConnect[ rd[i].board ]++ ;
+          gi_NumConnect[ b ]++ ;
           numok++ ;                     //make sure next round will execute
-          snprintf(str,MXSTR,"+++connect %d %d",rd[i].board,gi_NumConnect[ rd[i].board ]);
+          snprintf(str,MXSTR,"+++connect %d %d",b,gi_NumConnect[ b ]);
           factOut(kInfo,-1, str ) ;
         }
@@ -530 +568 @@
         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) ;
-}
-
-
+        if (jrd >0 ) {
+           qread+=jrd ;
+           debugStream(i,&rd[i].rBuf->B[ rd[i].bufPos],jrd) ;
+        }
 
         if (jrd == 0) {                 //connection has closed ...
-           rd[i].sockStat = -1 ;        //flag (try to reopen next round)
            snprintf(str,MXSTR,"Socket %d closed by FAD",i);
            factOut(kInfo,441, str ) ;
-           j = GenSock(1,0,NULL, &rd[i]) ;
+           j = GenSock(1, i, 0,NULL, &rd[i]) ;
            gi_ErrCnt[i]++ ;
-           gi_NumConnect[ rd[i].board ]-- ;
+           gi_NumConnect[ b ]-- ;
+
         } else if ( jrd<0 ) {           //did not read anything
            if (errno != EAGAIN && errno != EWOULDBLOCK ) {
@@ -551 +587 @@
            } else  numok-- ;            //else nothing waiting to be read
 
+        } else if ( rd[i].bufTyp <0 ) { // we are skipping this board ...
+//         just do nothing
+
         } else if ( rd[i].bufTyp >0 ) { // we are reading data ...
-           qread+=jrd ;
            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, 0,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; 0=reading data
+             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 ;
@@ -568 +606 @@
 
              } 
-debugRead(i,jrd,rd[i].evtID, 1,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; 1=finished event
+             debugRead(i,jrd,rd[i].evtID, 1,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; 1=finished event
 
              //we have a complete buffer, copy to WORK area
@@ -585 +623 @@
              //we have a valid entry in mBuffer[]; fill it
 
-             boardId = rd[i].board  ;
+             boardId = b ;
              int fadBoard = ntohs(rd[i].rBuf->S[12] ) ;
              int fadCrate = fadBoard/256 ;
@@ -652 +690 @@
              evtCtrl.pcTime[ iDx ] = g_actTime ;
 
-             if (++mBuffer[evID].nBoard == g_actBoards ) {
+             if (++mBuffer[evID].nBoard >= actBoards ) {
                 snprintf(str,MXSTR,"%5d complete event %8d %8d %2d",mBuffer[evID].evNum,evtCtrl.evtBuf[iDx],iDx,evtCtrl.evtStat[ iDx ]);
                 factOut(kDebug,-1, str ) ;
@@ -668 +706 @@
 
         } else {                        //we are reading event header
-           qread+=jrd ;
            rd[i].bufPos += jrd ;
            rd[i].bufLen -= jrd ;
@@ -704 +741 @@
                  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 ;
-debugRead(i,jrd,rd[i].evtID,-1,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid;-1=start event
+                 if (rd[i].bufLen <=0 ) rd[i].bufLen = 100000 ;    //?
+                 debugRead(i,jrd,rd[i].evtID,-1,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid;-1=start event
+              } else {
+                 debugRead(i,jrd,0,-2,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; -2=start event, unknown id yet
               }
-  else {
-debugRead(i,jrd,0,-2,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; -2=start event, unknown id yet
-  }
-
+           } else {
+              debugRead(i,jrd,0,-2,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; -2=start event, unknown id yet
            }
-
-  else {
-debugRead(i,jrd,0,-2,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; -2=start event, unknown id yet
-  }
 
         } //end interpreting last read
@@ -721 +754 @@
     } //finished trying to read all sockets
 
-int qwait=0, qdel=0, qskip=0 ;
+    int qwait=0, qdel=0, qskip=0 ;
     g_actTime = time(NULL) ;
     if ( g_actTime > gi_SecTime ) {
@@ -827 +860 @@
        }
        rd[i].sockStat = -1 ;        //flag (try to reopen next round)
-       gi_NumConnect[ rd[i].board ]-- ;
+       gi_NumConnect[ i/7 ]-- ;
     }
 
@@ -1325 +1358 @@
   /*-----------------------------------------------------------------*/
 
+  
+
 /*
-
-
 FileHandle_t  runOpen(uint32_t irun, RUN_HEAD *runhd, size_t len ) 
 { return 1; } ;
@@ -1379 +1412 @@
   g_maxMem = g_maxMem * 1024 *10 ; //10GBytes
 
-  g_maxBoards = 40 ;
-
-  g_actBoards = g_maxBoards; 
 
   g_runStat = 40 ;
@@ -1430 +1460 @@
 
 }
-   
-   */
+*/