source: trunk/FACT++/src/EventBuilder.c@ 11185

#define PX8   99  //simulator does not create double-length roi for pixel 8  
                  //for real data, set PX8 = 8 ==> ask for double roi=TM



#include <stdlib.h>
#include <stdint.h>
#include <unistd.h>
#include <stdio.h>
#include <sys/time.h>
#include <arpa/inet.h>
#include <string.h>
#include <math.h>
#include <error.h>
#include <errno.h>
#include <unistd.h>
#include <sys/types.h> 
#include <sys/socket.h>
#include <pthread.h>
#include <sched.h>

#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  64*1024   // size of read-buffer per socket
   
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 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,int32_t ftmevt, int32_t runnr, int state, 
                      uint32_t tsec, uint32_t tusec ) ;
extern void debugStream(int isock, void *buf, int len) ;



int g_actTime   =  0 ;
int g_runStat   = 40 ;
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 ;


uint gi_SecRate[MAX_SOCK] ;
uint gi_S10Rate[MAX_SOCK] ;
uint gi_MinRate[MAX_SOCK] ;
uint gi_ErrCnt[MAX_SOCK] ;

uint gi_NumConnect[NBOARDS];   //4 crates * 10 boards

uint gi_SecTime, gi_S10Time, gi_MinTime ;
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_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/8];
  int16_t S[MAX_LEN/4];
  int32_t I[MAX_LEN/2];
  int64_t L[MAX_LEN  ];
} 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  board ;           // boardID (softwareID: 0..40 )
  int  Port ;

  CNV_FACT *rBuf ;

} 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
     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 ;

     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<MAX_EVT*MAX_RUN; i++) {
      mBuffer[i].evNum = mBuffer[i].runNum = -1;

      evtCtrl.evtBuf[  i] = -1 ;
      evtCtrl.evtStat[ i] = -1 ;
      evtCtrl.pcTime[  i] = actime ;  //initiate to far future

   }


   actRun.FADhead = malloc( NBOARDS* sizeof(PEVNT_HEADER) ) ;

   evtCtrl.frstPtr = 0 ;
   evtCtrl.lastPtr = 0 ;

   return 0 ;

} /*-----------------------------------------------------------------*/




int mBufEvt(uint evID, uint runID, uint nRoi) {
// generate a new Event into mBuffer:   
// make sure only complete Event are possible, so 'free' will always work
// returns index into mBuffer[], or negative value in case of error


   int i, k, evFree ;
   int headmem=0 ;
   size_t needmem = 0 ;

   if (nRoi <=0 || nRoi > 1024) {
      snprintf(str,MXSTR,"illegal nRoi: %d",nRoi) ;
      factOut(kError, 1, str ) ;
      return -9999 ;
   }

   i = evID % MAX_EVT ;
   evFree = -1 ;

   for ( k=0; k<MAX_RUN; k++) {
      if ( mBuffer[i].evNum == evID
        && mBuffer[i].runNum== runID ) {
         return i ;  
      }
      if ( evFree < 0 && mBuffer[i].evNum < 0 ) evFree = i ;
      i += MAX_EVT ;
   }


   //event does not yet exist; create
   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 ...

   
   needmem = sizeof(EVENT) + NPIX*nRoi*2 + NTMARK*nRoi*2; //

   headmem = NBOARDS* sizeof(PEVNT_HEADER) ;

   if ( gi_usedMem + needmem + headmem > g_maxMem) {
        snprintf(str,MXSTR,"no memory left to keep event %d",evID) ;
        factOut(kError,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; k<NBOARDS; k++ ) {
      mBuffer[i].board[k] = -1;
   }

   //flag all pixels as unused
   for (k=0; k<NPIX; k++ ) {
      mBuffer[i].fEvent->StartPix[k] = -1 ;
   }

   //flag all TMark as unused
   for (k=0; k<NTMARK; k++ ) {
      mBuffer[i].fEvent->StartTM[k] = -1 ;
   }

   mBuffer[i].pcTime  = g_actTime ;
   mBuffer[i].nRoi    = nRoi ;
   mBuffer[i].evNum   = evID  ;
   mBuffer[i].runNum  = runID ;
   mBuffer[i].evtLen  = needmem ;

   gi_usedMem += needmem + headmem;

   //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 start new evt  %8d %8d %2d",evID,i,evtCtrl.lastPtr,0);
factOut(kDebug,-11, str ) ;
   evtCtrl.lastPtr++ ;
   if (evtCtrl.lastPtr == MAX_EVT*MAX_RUN ) evtCtrl.lastPtr = 0;




   gi_EvtStart++ ;

   //check if runId already registered in runCtrl
   evFree = -1 ;
   for (k=0; k<MAX_RUN; k++) {
      if (runCtrl[k].runId == runID ) return i ;//run exists already
      else if (evFree < 0 && runCtrl[k].runId == 0 ) evFree = k ;
   }
   
   if (evFree <0 ) {
      snprintf(str,MXSTR,"not able to register the new run %d",runID);
      factOut(kFatal,883, str ) ;
   } else {
      runCtrl[evFree].runId = runID ;
   }

   return i ;
 
} /*-----------------------------------------------------------------*/


int mBufFree(int i) {
//delete entry [i] from mBuffer:
//(and make sure multiple calls do no harm ....)

   int headmem=0 ;
   size_t freemem = 0 ;

   if ( mBuffer[i].nRoi > 0) {      //have an fEvent structure generated ...
      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].runNum = mBuffer[i].nRoi= -1;

   gi_usedMem = gi_usedMem - freemem - headmem;


   return 0 ;
 
} /*-----------------------------------------------------------------*/


  /*-----------------------------------------------------------------*/



void initReadFAD() {
return ;
} /*-----------------------------------------------------------------*/



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,k ;
  int actBoards = 0, minLen ;
  int32_t jrd ;
  int32_t myRun ;
  int boardId, roi,drs,px,src,pixS,pixH,pixC,pixR,tmS ;
  uint qtot = 0, qread = 0, qconn = 0 ;
  int errcnt0 = 0 ;

  int goodhed=0;

  struct timespec xwait ;

  int nokCnt[MAX_SOCK],loopCnt=0;
  int sokCnt[MAX_SOCK];
  int sockDef[NBOARDS];

  struct timeval  *tv, atv;
  tv=&atv;
  uint32_t tsec, tusec ;


  snprintf(str,MXSTR,"start initializing");
  factOut(kInfo,-1, str ) ;

  int cpu = 7 ;   //read thread
  cpu_set_t mask;

/* CPU_ZERO initializes all the bits in the mask to zero. */
   CPU_ZERO( &mask );
/* CPU_SET sets only the bit corresponding to cpu. */
   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 ;
  }
  for (i=0; i<NBOARDS; i++) sockDef[i]= 0 ; 


  g_actTime = time(NULL) ;
  for (k=0; k<MAX_SOCK; k++)
     gi_SecRate[k]=gi_S10Rate[k]=gi_MinRate[k]=gi_ErrCnt[k] = 0 ;

  for (k=0; k<NBOARDS; k++)
     gi_NumConnect[k]=0;


  gi_SecTime= gi_S10Time= gi_MinTime= g_actTime ;

  mBufInit() ;    //initialize buffers

  snprintf(str,MXSTR,"end   initializing");
  factOut(kInfo,-1, str ) ;


  for (k=0; k<MAX_SOCK; k++) sokCnt[k]=nokCnt[k]=0 ;

  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;

  myRun = g_actTime ;

  gi_runStat = g_runStat ;


  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) ;
    nokCnt[numok]++;

    loopCnt++ ;

    numok = 0 ;                       //count number of succesfull actions

    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);
      tsec = atv.tv_sec ;
      tusec= atv.tv_usec ; 

      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 = 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
          snprintf(str,MXSTR,"+++connect %d %d",b,gi_NumConnect[ 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++ ;
          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

        } 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  ] != stop.B[1]) {
                gi_ErrCnt[i]++ ;
                snprintf(str,MXSTR,"wrong end of buffer found %d",rd[i].bufPos);
                factOut(kError,301, str ) ;
                goto EndBuf ;

             } 
             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
             gi_SecRate[i]++ ;

             roi = ntohs(rd[i].rBuf->S[ head_len/2 + 2 ]) ; 
             //get index into mBuffer for this event (create if needed)
             evID = mBufEvt( rd[i].evtID, rd[i].runID, roi ) ;

             if (evID <-9000) goto EndBuf ;  //illegal event, skip it ...
             if (evID < 0) {
                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

             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) ;
                if (errcnt0++ < 99 ) factOut(kWarn,301, str ) ;  //print only few times
//           } else {
//              snprintf(str,MXSTR,"correct Board ID %d %d %d",fadCrate,fadBoard%256,boardId) ;
//              if (errcnt0++ < 99 ) factOut(kWarn,301, str ) ;  //print only few times
             }
             if ( mBuffer[evID].board[ boardId ] != -1) {   
                snprintf(str,MXSTR,"double board %d for event %d",boardId,evID) ;
                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) ;
             roi  = mBuffer[evID].nRoi ;

             pixS = boardId*36 -1 ;   //
             tmS  = boardId*4  -1 ;   //
             src  = head_len/2 ;
             for ( drs=0; drs<4; drs++ ) {
                for ( px=0; px<9; px++ ) {
                   pixH= ntohs(rd[i].rBuf->S[src++]) ;
                   pixC= ntohs(rd[i].rBuf->S[src++]) ;
                   pixR= ntohs(rd[i].rBuf->S[src++]) ;

                   src++  ;
                   pixS++ ; //pixS = pixH2S[pixH] ;
                   if ( ( px < PX8 && pixR == roi )
                     || ( px ==PX8 && pixR == 2*roi )
                     || ( px ==PX8 && pixR == roi && roi > 512 ) ) {
                       // correct roi
                      mBuffer[evID].fEvent->StartPix[pixS] =pixC;
                      dest= pixS * roi ;
                      memcpy(
                           &mBuffer[evID].fEvent->Adc_Data[dest],
                           &rd[i].rBuf->S[src],  roi * 2) ;
                      src+= roi ;
                      if ( px==PX8 ) {
                         tmS++; // tmS = tmH2S[pixH]
                         dest= tmS * roi + NPIX* roi ;
                         if ( roi <=512 ) {
                            mBuffer[evID].fEvent->StartTM[tmS] =(pixC+roi)%1024 ;
                            memcpy(
                              &mBuffer[evID].fEvent->Adc_Data[dest],
                              &rd[i].rBuf->S[src],  roi * 2) ;
                            src+=roi ;
                         } else {
                             mBuffer[evID].fEvent->StartTM[tmS] = -1 ;
                         }
                      }
                   } else {
                      snprintf(str,MXSTR,"wrong roi %d %d %d %d",px,pixR,roi,src-2);
                      factOut(kError,202, str ) ;
                      goto EndBuf ;
                   }
                }
             }// now we have stored a new board contents into Event structure
             mBuffer[evID].board[ boardId ] = boardId ;
             evtCtrl.evtStat[ iDx ]++ ;
             evtCtrl.pcTime[ iDx ] = g_actTime ;

             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 ) ;
                //complete event read ---> flag for next processing
                evtCtrl.evtStat[ iDx ] = 99;
                gi_EvtRead++ ;
                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; 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 ) ;

                 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 ) ;
              }
              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].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]) ;
                 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,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

    int qwait=0, qdel=0, qskip=0 ;
    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) ;

       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] < 90 ) {

           qwait++; 

           if( 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_EvtBad++ ;
             gi_EvtTot++ ;    
             qskip++; 
           } 


          } 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; 
              qdel++; 
              qtot++;
          }

          if ( k0 == evtCtrl.frstPtr && evtCtrl.evtStat[k0] <0 ) {
             evtCtrl.frstPtr = (evtCtrl.frstPtr+1) % (MAX_EVT*MAX_RUN) ;
          }
       }

qconn=0 ;
int ib ;
for (ib=0; ib<NBOARDS; ib++) qconn+=gi_NumConnect[ib] ;
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 ;
    }




    if (numok > 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 ...

 //must quit eventbuilding
 snprintf(str,MXSTR,"stop reading ...");
 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_EvtBad++ ;
       gi_EvtTot++ ;    
    }
 }

 //must close all open sockets ...
 snprintf(str,MXSTR,"close all sockets ...");
 factOut(kInfo,-1, str ) ;
 for (i=0; i<MAX_SOCK; i++) {  
    GenSock(-1, i, 0, NULL, &rd[i]) ; //close and destroy socket   
    gi_NumConnect[ 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


 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 ;
    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 ) ;
 }

 snprintf(str,MXSTR,"Exit read Process ...");
 factOut(kInfo,-1, str ) ;
 gi_runStat = -99 ;
 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) {
           int      id   = evtCtrl.evtBuf[k0] ;
           int      ievt = mBuffer[id].evNum ;
           int      roi  = mBuffer[id].nRoi ;
//         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
           int ip,it,dest,ib;
           for (ip=0; ip<NPIX; ip++) {
             if (mBuffer[id].fEvent->StartPix[ip] == -1 ) {
                dest= ip*roi ;
                bzero( &mBuffer[id].fEvent->Adc_Data[dest], roi*2) ;
             }
           }
           for (it=0; it<NTMARK; it++) {
             if (mBuffer[id].fEvent->StartTM[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->EventNum = ievt ;
           mBuffer[id].fEvent->TriggerType = 0 ; // TBD
           mBuffer[id].fEvent->SoftTrig = 0 ;
           for (ib=0; ib<NBOARDS; ib++) {
              if (mBuffer[id].board[ib] == -1 ) {  //board is not read
                 mBuffer[id].FADhead[ib].start_package_flag = 0 ;
                 mBuffer[id].fEvent->BoardTime[ib] = 0 ;
              } else {
                 mBuffer[id].fEvent->BoardTime[ib] = 
                   ntohl(mBuffer[id].FADhead[ib].time) ;
              }
           }

           int i=eventCheck(mBuffer[id].FADhead,mBuffer[id].fEvent) ;
           
           if (i<0) evtCtrl.evtStat[k0] = 999 ; //flag event to be skipped

           numProc++ ;
           evtCtrl.evtStat[k0] = 520 ;
           gp_EvtTot++ ;
        } else if ( evtCtrl.evtStat[k0] >=0 && evtCtrl.evtStat[k0] < 90 ) {
           numWait++ ;
        }
     }

     if ( gi_runStat < -10 && numWait == 0) {  //nothing left to do
        snprintf(str,MXSTR,"Exit Processing Process ...");
        factOut(kInfo,-1, str ) ;
        gp_runStat = -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 ;

  }

  //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 ;

  return 0;
  
} /*-----------------------------------------------------------------*/

int CloseRunFile(uint32_t runId, uint32_t closeTime) {
/* close run runId (all all runs if runId=0) */
/* return: 0=close scheduled / >0 already closed / <0 does not exist */
  int i, j ;

  if (runId == 0 ) {
     for ( j=0; j<MAX_RUN; j++) {
        if ( runCtrl[j].fileId == 0 ) {  //run is open
           i=runClose(runCtrl[j].fileHd, &runTail[j], sizeof(runTail[j]) );
           if (i<0) {
              snprintf(str,MXSTR,"error closing run %d %d",runCtrl[j].runId,i) ;
              factOut(kError,506, str ) ;
              runCtrl[j].fileId = 888 ;
           } else {
              snprintf(str,MXSTR,"closing run %d ok  AAA",runCtrl[j].runId);
              factOut(kInfo,507, str ) ;
              runCtrl[j].fileId = 7777 ;
           }
           runCtrl[j].closeTime = closeTime ;
        }
     }
     return 0 ;
  }


  for ( j=0; j<MAX_RUN; j++) {
     if ( runCtrl[j].runId == runId ) {
        if ( runCtrl[j].fileId == 0 ) {  //run is open
           i=runClose(runCtrl[j].fileHd, &runTail[j], sizeof(runTail[j]) );
           if (i<0) {
              snprintf(str,MXSTR,"error closing run %d %d",runCtrl[j].runId,i) ;
              factOut(kError,506, str ) ;
              runCtrl[j].fileId = 888 ;
           } else {
              snprintf(str,MXSTR,"closing run %d ok  AAA",runCtrl[j].runId);
              factOut(kInfo,507, str ) ;
              runCtrl[j].fileId = 7777 ;
           }
           runCtrl[j].closeTime = closeTime ;
           return 0;
        } else if ( runCtrl[j].fileId <0 ) { //run not yet opened
           runCtrl[j].closeTime = closeTime ;
           return +1;
        } else {     // run already closed
           return +2;
        }
     }
  }   //we only reach here if the run was never created
  return -1;

} /*-----------------------------------------------------------------*/


void *writeEvt( void *ptr ) {
/* *** main loop writing event (including opening and closing run-files */

  int  numWrite, numWait ;
  int k,j ;
  struct timespec xwait ;
  char str[MXSTR] ;

  cpu_set_t mask;
  int cpu = 3 ;   //write thread

  snprintf(str,MXSTR,"Starting write-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,"W ---> 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) {
           int      id   = evtCtrl.evtBuf[k0] ;
           uint32_t irun = mBuffer[id].runNum ;
           int      ievt = mBuffer[id].evNum ;

           if (runCtrl[lastRun].runId == irun ) {
              j = lastRun ;
           } else {
              //check which fileID to use (or open if needed)
              for ( j=0; j<MAX_RUN; j++) {
                 if ( runCtrl[j].runId == irun ) break ;
              }
              if ( 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 ) ;
for ( j=0; j<MAX_RUN; j++) printf("j %d   run.j %d   run %d\n",j,runCtrl[j].runId,irun );
exit(111);
              }
              lastRun = j ;
           }

           if (runCtrl[j].fileId < 0 ) {
              actRun.Version =  1 ;
              actRun.RunType = -1 ;
              actRun.NBoard  = NBOARDS ;
              actRun.NPix    = NPIX ;
              actRun.NTm     = NTMARK ;
              actRun.Nroi    = mBuffer[id].nRoi ;
//            actRun.FADhead = mBuffer[id].FADhead ;  //to be corrected
              runCtrl[j].nextEvt= 0;
              runCtrl[j].lastTime=g_actTime ;
              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 = 99 ;
              } else {
                 snprintf(str,MXSTR,"W opened new run_file %d",irun) ;
                 factOut(kInfo,-1, str ) ;
                 runCtrl[j].fileId = 0 ;
              }

           }

           if (runCtrl[j].fileId > 0 ) {
              snprintf(str,MXSTR,"W no open file for this run %d",irun) ;
              factOut(kDebug,123,str) ;
              evtCtrl.evtStat[k0] = 902 ;
           } else {
              int i=runWrite(runCtrl[j].fileHd, mBuffer[id].fEvent, sizeof(mBuffer[id]) );
              if (i<0) {
                 snprintf(str,MXSTR,"W error writing event for run %d",irun) ;
                 factOut(kError,503, str ) ;
                 evtCtrl.evtStat[k0] = 901 ;
                 //close run
                 i=runClose(runCtrl[j].fileHd, &runTail[j], sizeof(runTail[j]) );
                 if (i<0) {
                    snprintf(str,MXSTR,"W error closing run %d",irun) ;
                    factOut(kError,503, str ) ;
                 } else {
                    snprintf(str,MXSTR,"W closed run %d because of write error",irun) ;
                    factOut(kInfo,503, str ) ;
                 }
                 runCtrl[j].fileId = 9999 ;
              } else {
                 runCtrl[j].lastTime = g_actTime; 
                 evtCtrl.evtStat[k0] = 901 ;
                 snprintf(str,MXSTR,"%5d successfully wrote for run %d id %5d",ievt,irun,k0);
                 factOut(kDebug,504, str ) ;
              }
           }
        } else if (evtCtrl.evtStat[k0] > 0 ) numWait++ ;
     }

     //check if we should close a run ...
     for ( j=0; j<MAX_RUN; j++) {
        if ( runCtrl[j].fileId==0    
          && (  runCtrl[j].closeTime < g_actTime  
              ||runCtrl[j].lastTime  < g_actTime-120) ) {
           int i=runClose(runCtrl[j].fileHd, &runTail[j], sizeof(runTail[j]) );
           if (i<0) {
              snprintf(str,MXSTR,"error closing run %d %d",runCtrl[j].runId,i) ;
              factOut(kError,506, str ) ;
              runCtrl[j].fileId = 888 ;
           } else {
              snprintf(str,MXSTR,"closing run %d ok  BBB",runCtrl[j].runId);
              factOut(kInfo,507, str ) ;
              runCtrl[j].fileId = 7777 ;
           }
        }
     }

     if (numWrite == 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 ) ;
     }

     if ( gi_runStat < -10 && numWait == 0) {  //nothing left to do
        snprintf(str,MXSTR,"Finish Write Process ...");
        factOut(kInfo,-1, str ) ;
        gw_runStat = -22 ;                     //==> we should exit
        goto closerun ;
     }
     gw_runStat = gi_runStat ;

  }

  //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<MAX_RUN; j++) 
     if ( runCtrl[j].fileId ==0 ) {
        int i=runClose(runCtrl[j].fileHd, &runTail[j], sizeof(runTail[j]) );
        if (i<0) {
           snprintf(str,MXSTR,"error closing run %d %d",runCtrl[j].runId,i) ;
           factOut(kError,506, str ) ;
           runCtrl[j].fileId = 888 ;
        } else {
           snprintf(str,MXSTR,"closing run %d ok  AAA",runCtrl[j].runId);
           factOut(kInfo,507, str ) ;
           runCtrl[j].fileId = 7777 ;
        }
     }

  gw_runStat = -99; 
  snprintf(str,MXSTR,"Exit Writing Process ...");
  factOut(kInfo,-1, str ) ;
  return 0;


  

} /*-----------------------------------------------------------------*/




void StartEvtBuild() {

  int i,j,imax,status,th_ret[50] ;
  pthread_t thread[50] ;
  struct timespec xwait ;
  uint32_t actime ;

  gi_runStat = gp_runStat = gw_runStat = 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].lastTime = 0 ;
      runCtrl[i].closeTime = time(NULL) + 3600*24*7; 

      runCtrl[i].nextEvt = 0 ;
      runCtrl[i].fileId = -2 ;

      runTail[i].nEventsOk =
      runTail[i].nEventsRej =
      runTail[i].nEventsBad =
      runTail[i].PCtime0 =
      runTail[i].PCtimeX = 0 ;
   }

//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
     nanosleep( &xwait , NULL ) ;
  }

  i=0 ;
  th_ret[i] = pthread_create( &thread[i], NULL, readFAD,  NULL );
  i++;
  th_ret[i] = pthread_create( &thread[i], NULL, procEvt,  NULL );
  i++;
  th_ret[i] = pthread_create( &thread[i], NULL, writeEvt, NULL );
  i++;
  imax=i ;






     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) ;

     xwait.tv_sec = 5;;
     xwait.tv_nsec= 0 ;  // sleep for ~20sec
     nanosleep( &xwait , NULL ) ;

     printf("setting g_runstat to -1\n");

     g_runStat = -1 ;












//wait for all threads to finish
  for (i=0; i<imax; i++) {
     j = pthread_join ( thread[i], (void **)&status) ;
  }

} /*-----------------------------------------------------------------*/





  /*-----------------------------------------------------------------*/
  /*-----------------------------------------------------------------*/
  /*-----------------------------------------------------------------*/
  /*-----------------------------------------------------------------*/
  /*-----------------------------------------------------------------*/
  /*-----------------------------------------------------------------*/
  /*-----------------------------------------------------------------*/
  /*-----------------------------------------------------------------*/
  /*-----------------------------------------------------------------*/
  /*-----------------------------------------------------------------*/

  
/*

FileHandle_t  runOpen(uint32_t irun, RUN_HEAD *runhd, size_t len ) 
{ return 1; } ;

int  runWrite(FileHandle_t fileHd ,  EVENT    *event, size_t len ) 
{ return 1; } ;

int  runClose(FileHandle_t fileHd ,  RUN_TAIL *runth, size_t len ) 
{ return 1; } ;



int  eventCheck( PEVNT_HEADER *fadhd, EVENT *event) 
{
   int i=0;

   printf("------------%d\n",ntohl(fadhd[7].fad_evt_counter) );
   for (i=0; i<NBOARDS; i++) {
      printf("b=%2d,=%5d\n",i,fadhd[i].board_id);
   }
   return 0;
}


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]);
}


void debugRead(int isock, int ibyte, int32_t event, int32_t ftmevt, int32_t runnr, int state, uint32_t tsec, uint32_t tusec ) {
//  printf("%3d %5d %9d %3d %12d\n",isock, ibyte, event, state, tusec) ;
}



void debugStream(int isock, void *buf, int len) {
}

void debugHead(int i, void *buf) { 
}


void factOut(int severity, int err, char* message ) {
static FILE * fd ;
static int  file=0 ;

  if (file==0) {
     printf("open file\n");
     fd=fopen("x.out","w+") ;
     file=999; 
  }

  fprintf(fd,"%3d %3d | %s \n",severity,err,message) ;

  if (severity != kDebug) 
  printf("%3d %3d | %s\n",severity,err,message) ;
}



int main() {
  int i,b,c,p ;
  char     ipStr[100] ;
  struct in_addr IPaddr ;

  g_maxMem = 1024*1024    ;        //MBytes
  g_maxMem = g_maxMem * 1024 *10 ; //10GBytes


  g_runStat = 40 ;

  i=0 ;

// version for standard crates
//for (c=0; c<4,c++) {
//   for (b=0; b<10; b++) {
//      sprintf(ipStr,"10.0.%d.%d",128+c,128+b)
//
//      inet_pton(PF_INET, ipStr, &IPaddr) ;
//
//      g_port[i].sockAddr.sin_family = PF_INET;
//      g_port[i].sockAddr.sin_port = htons(5000) ;
//      g_port[i].sockAddr.sin_addr = IPaddr ;
//      g_port[i].sockDef = 1 ;
//      i++ ;
//   }
//}
//
//version for PC-test 
  for (c=0; c<4; c++) {
     for (b=0; b<10; b++) {
        sprintf(ipStr,"10.0.%d.11",128+c) ;
        if (c==0) sprintf(ipStr,"10.0.100.11") ;

        inet_pton(PF_INET, ipStr, &IPaddr) ;
        p = 31919+100*c+10*b; 


        g_port[i].sockAddr.sin_family = PF_INET;
        g_port[i].sockAddr.sin_port = htons(p) ;
        g_port[i].sockAddr.sin_addr = IPaddr ;
        g_port[i].sockDef = 1 ;

        i++ ;
     }
  }


//g_port[17].sockDef =-1 ;
//g_actBoards-- ; 

  StartEvtBuild() ;

  return 0;

}
*/