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

// // // #define EVTDEBUG

#define NUMSOCK   1             //set to 7 for old configuration
#define MAXREAD  65536          //64kB wiznet buffer
#define MIN_LEN  32             // min #bytes needed to interpret FADheader
#define MAX_LEN 256*1024        // size of read-buffer per socket

#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 <netinet/in.h>
#include <netinet/tcp.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
};


void *gi_procPtr;


//#define nanosleep(x,y)

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 int runFinish (uint32_t runnr);



extern void * runStart (uint32_t irun, RUN_HEAD * runhd, size_t len);

extern int  runEnd (uint32_t, void * runPtr );

extern void factOut (int severity, int err, char *message);

extern void gotNewRun (int runnr, PEVNT_HEADER * headers);


extern void factStat (GUI_STAT gj);

extern void factStatNew (EVT_STAT gi);

extern int eventCheck (uint32_t runNr, PEVNT_HEADER * fadhd, EVENT * event,
                       int mboard);

extern int subProcEvt (int threadID, PEVNT_HEADER * fadhd, EVENT * event,
                       int16_t mboard, void * buffer);

extern void debugHead (int i, int j, 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 CloseRunFile (uint32_t runId, uint32_t closeTime, uint32_t maxEvt);


int g_maxProc;
int g_maxSize;
int gi_maxSize =0;
int gi_maxProc;
uint32_t gi_actRun;
void *gi_runPtr;

uint g_actTime;
uint g_actUsec;
int g_runStat;
int g_reset;
int g_useFTM;

int gi_reset, gi_resetR, gi_resetS, gi_resetW, gi_resetX;
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;

int gi_memStat = +1;

uint32_t gi_myRun = 0;
uint32_t actrun = 0;


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

//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_STAT gi;
GUI_STAT gj;

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];
   int16_t S[MAX_LEN / 2];
   int32_t I[MAX_LEN / 4];
   int64_t L[MAX_LEN / 8];
} 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
// size_t bufLen;               //number of bytes left to read size_t might be better
   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 ftmTyp;                  // trigger type
   int board;                   // boardID (softwareID: 0..40 )
   int Port;

   CNV_FACT *rBuf;

#ifdef EVTDEBUG
   CNV_FACT *xBuf;              //a copy of rBuf (temporary for debuging) 
#endif

} 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
runFinish1 (uint32_t runnr)
{
   snprintf (str, MXSTR, "Should finish run %d (but not yet possible)",
             runnr);
   factOut (kInfo, 173, str);   //but continue anyhow
   return 0;
}


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;
   int optval = 1;              //activate keepalive
   socklen_t optlen = sizeof (optval);

   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
#ifdef EVTDEBUG
      free (rd->xBuf);          //and never open again
#endif
      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;

#ifdef EVTDEBUG
      rd->xBuf = malloc (sizeof (CNV_FACT));
#endif
      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;
   }
   optval = 1;
   if (setsockopt (rd->socket, SOL_SOCKET, SO_KEEPALIVE, &optval, optlen) < 0) {
      snprintf (str, MXSTR, "Could not set keepalive %d | %m", sid);
      factOut (kInfo, 173, str);        //but continue anyhow
   }
   optval = 10;                 //start after 10 seconds
   if (setsockopt (rd->socket, SOL_TCP, TCP_KEEPIDLE, &optval, optlen) < 0) {
      snprintf (str, MXSTR, "Could not set keepidle %d | %m", sid);
      factOut (kInfo, 173, str);        //but continue anyhow
   }
   optval = 10;                 //do every 10 seconds
   if (setsockopt (rd->socket, SOL_TCP, TCP_KEEPINTVL, &optval, optlen) < 0) {
      snprintf (str, MXSTR, "Could not set keepintvl %d | %m", sid);
      factOut (kInfo, 173, str);        //but continue anyhow
   }
   optval = 2;                  //close after 2 unsuccessful tries
   if (setsockopt (rd->socket, SOL_TCP, TCP_KEEPCNT, &optval, optlen) < 0) {
      snprintf (str, MXSTR, "Could not set keepalive probes %d | %m", sid);
      factOut (kInfo, 173, str);        //but continue anyhow
   }



   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].nRoi = -1;
      mBuffer[i].runNum = 0;

      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 (int evID, uint runID, int nRoi[], int sk,
         int fadlen, int trgTyp, int trgNum, int fadNum)
{
// 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
// error: <-9000 if roi screwed up (not consistent with run)
//        <-8000                   (not consistent with event)
//        <-7000                   (not consistent with board)
//        < 0    if no space left

   struct timeval *tv, atv;
   tv = &atv;
   uint32_t tsec, tusec;
   uint oldest;
   int jold;

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


   b = sk / 7;

   if (nRoi[0] < 0 || nRoi[0] > 1024) {
      snprintf (str, MXSTR, "illegal nRoi[0]: %d", nRoi[0]);
      factOut (kError, 999, str);
      gj.badRoiR++;
      gj.badRoi[b]++;
      return -9999;
   }

   for (jr = 1; jr < 8; jr++) {
      if (nRoi[jr] != nRoi[0]) {
         snprintf (str, MXSTR, "wrong nRoi[%d]: %d %d", jr, nRoi[jr],
                   nRoi[0]);
         factOut (kError, 711, str);
         gj.badRoiB++;
         gj.badRoi[b]++;
         return -7101;
      }
   }
   if (nRoi[8] < nRoi[0]) {
      snprintf (str, MXSTR, "wrong nRoi_TM: %d %d", nRoi[8], nRoi[0]);
      factOut (kError, 712, str);
      gj.badRoiB++;
      gj.badRoi[b]++;
      return -7102;
   }


   i = evID % MAX_EVT;
   evFree = -1;

   for (k = 0; k < MAX_RUN; k++) {
      if (mBuffer[i].evNum == evID && mBuffer[i].runNum == runID) {     //event is already registered;
         // is it ok ????
         if (mBuffer[i].nRoi != nRoi[0]
             || mBuffer[i].nRoiTM != nRoi[8]) {
            snprintf (str, MXSTR, "illegal evt_roi %d %d ; %d %d",
                      nRoi[0], nRoi[8], mBuffer[i].nRoi, mBuffer[i].nRoiTM);
            factOut (kError, 821, str);
            gj.badRoiE++;
            gj.badRoi[b]++;
            return -8201;
         }
//       count for inconsistencies

         if (mBuffer[i].trgNum != trgNum)
            mBuffer[i].Errors[0]++;
         if (mBuffer[i].fadNum != fadNum)
            mBuffer[i].Errors[1]++;
         if (mBuffer[i].trgTyp != trgTyp)
            mBuffer[i].Errors[2]++;

         //everything seems fine so far ==> use this slot ....
         return i;
      }
      if (evFree < 0 && mBuffer[i].evNum < 0)
         evFree = i;
      i += MAX_EVT;
   }


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

   gettimeofday (tv, NULL);
   tsec = atv.tv_sec;
   tusec = atv.tv_usec;

   //check if runId already registered in runCtrl
   evFree = -1;
   oldest = g_actTime + 1000;
   jold = -1;
   for (k = 0; k < MAX_RUN; k++) {
      if (runCtrl[k].runId == runID) {
//         if (runCtrl[k].procId > 0) {   //run is closed -> reject
//            snprintf (str, MXSTR, "skip event since run %d already closed", runID);
//            factOut (kInfo, 931, str);
//            return -21;
//         }

         if (runCtrl[k].roi0 != nRoi[0]
             || runCtrl[k].roi8 != nRoi[8]) {
            snprintf (str, MXSTR, "illegal run_roi %d %d ; %d %d",
                      nRoi[0], nRoi[8], runCtrl[k].roi0, runCtrl[k].roi8);
            factOut (kError, 931, str);
            gj.badRoiR++;
            gj.badRoi[b]++;
            return -9301;
         }
         goto RUNFOUND;
      } else if (evFree < 0 && runCtrl[k].fileId < 0) { //not yet used
         evFree = k;
      } else if (evFree < 0 && runCtrl[k].fileId > 0) { //already closed
         if (runCtrl[k].closeTime < oldest) {
            oldest = runCtrl[k].closeTime;
            jold = k;
         }
      }
   }

   if (evFree < 0 && jold < 0) {
      snprintf (str, MXSTR, "not able to register the new run %d", runID);
      factOut (kFatal, 883, str);
      return -1001;
   } else {
      if (evFree < 0)
         evFree = jold;
      snprintf (str, MXSTR, "register new run %d(%d) roi: %d %d", runID,
                evFree, nRoi[0], nRoi[8]);
      factOut (kInfo, 503, str);
      runCtrl[evFree].runId = runID;
      runCtrl[evFree].roi0 = nRoi[0];
      runCtrl[evFree].roi8 = nRoi[8];
      runCtrl[evFree].fileId = -2;
      runCtrl[evFree].procId = -2;
      runCtrl[evFree].lastEvt = -1;
      runCtrl[evFree].nextEvt = 0;
      runCtrl[evFree].actEvt = 0;
      runCtrl[evFree].procEvt = 0;
      runCtrl[evFree].maxEvt = 999999999;       //max number events allowed
      runCtrl[evFree].firstUsec = tusec;
      runCtrl[evFree].firstTime = runCtrl[evFree].lastTime = tsec;
      runCtrl[evFree].closeTime = tsec + 3600 * 24;     //max time allowed

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

 RUNFOUND:

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

   headmem = NBOARDS * sizeof (PEVNT_HEADER);

   if (gj.usdMem + needmem + headmem + gi_maxSize > g_maxMem) {
      gj.maxMem = gj.usdMem + needmem + headmem + gi_maxSize;
      if (gi_memStat > 0) {
         gi_memStat = -99;
         snprintf (str, MXSTR, "no memory left to keep event %6d sock %3d",
                   evID, sk);
         factOut (kError, 882, str);
      } else {
         snprintf (str, MXSTR, "no memory left to keep event %6d sock %3d",
                   evID, sk);
         factOut (kDebug, 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].FADhead);
      mBuffer[i].FADhead = NULL;
      return -22;
   }

// if (gi_maxSize > 0) {
//    mBuffer[i].buffer = malloc (gi_maxSize);
//    if (mBuffer[i].buffer == NULL) {
//       snprintf (str, MXSTR, "malloc buffer failed for event %d", evID);
//       factOut (kError, 882, str);
//       free (mBuffer[i].FADhead);
//       mBuffer[i].FADhead = NULL;
//       free (mBuffer[i].fEvent);
//       mBuffer[i].fEvent = NULL;
//       return -32;
//    }
// } else {
//    mBuffer[i].buffer = NULL;
// }

   //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].fEvent->NumBoards = 0;
   mBuffer[i].fEvent->PCUsec = tusec;
   mBuffer[i].fEvent->PCTime = mBuffer[i].pcTime = tsec;
   mBuffer[i].nRoi = nRoi[0];
   mBuffer[i].nRoiTM = nRoi[8];
   mBuffer[i].evNum = evID;
   mBuffer[i].runNum = runID;
   mBuffer[i].fadNum = fadNum;
   mBuffer[i].trgNum = trgNum;
   mBuffer[i].trgTyp = trgTyp;
   mBuffer[i].evtLen = needmem;
   mBuffer[i].Errors[0] =
      mBuffer[i].Errors[1] = mBuffer[i].Errors[2] = mBuffer[i].Errors[3] = 0;

   gj.usdMem += needmem + headmem + gi_maxSize;
   if (gj.usdMem > gj.maxMem)
      gj.maxMem = gj.usdMem;

   gj.bufTot++;
   if (gj.bufTot > gj.maxEvt)
      gj.maxEvt = gj.bufTot;

   gj.rateNew++;

   //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 %8d start new evt  %8d %8d sock %3d len %5d t %10d", evID,
             runID, i, evtCtrl.lastPtr, sk, fadlen, mBuffer[i].pcTime);
   factOut (kDebug, -11, str);
   evtCtrl.lastPtr++;
   if (evtCtrl.lastPtr == MAX_EVT * MAX_RUN)
      evtCtrl.lastPtr = 0;

   gi.evtGet++;

   return i;

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




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

   int headmem = 0;
   int evid;
   size_t freemem = 0;

   evid = mBuffer[i].evNum;
   freemem = mBuffer[i].evtLen;

   free (mBuffer[i].fEvent);
   mBuffer[i].fEvent = NULL;

   free (mBuffer[i].FADhead);
   mBuffer[i].FADhead = NULL;

// if (gi_maxSize > 0) {
//    free (mBuffer[i].buffer);
//    mBuffer[i].buffer = NULL;
// }

   headmem = NBOARDS * sizeof (PEVNT_HEADER);
   mBuffer[i].evNum = mBuffer[i].nRoi = -1;
   mBuffer[i].runNum = 0;

   gj.usdMem = gj.usdMem - freemem - headmem - gi_maxSize;
   gj.bufTot--;

   if (gi_memStat < 0) {
      if (gj.usdMem <= 0.75 * gj.maxMem)
         gi_memStat = +1;
   }


   return 0;

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


void
resetEvtStat ()
{
   int i;

   for (i = 0; i < MAX_SOCK; i++)
      gi.numRead[i] = 0;

   for (i = 0; i < NBOARDS; i++) {
      gi.gotByte[i] = 0;
      gi.gotErr[i] = 0;

   }

   gi.evtGet = 0;               //#new Start of Events read
   gi.evtTot = 0;               //#complete Events read
   gi.evtErr = 0;               //#Events with Errors
   gi.evtSkp = 0;               //#Events incomplete (timeout)

   gi.procTot = 0;              //#Events processed
   gi.procErr = 0;              //#Events showed problem in processing
   gi.procTrg = 0;              //#Events accepted by SW trigger
   gi.procSkp = 0;              //#Events rejected by SW trigger

   gi.feedTot = 0;              //#Events used for feedBack system
   gi.feedErr = 0;              //#Events rejected by feedBack

   gi.wrtTot = 0;               //#Events written to disk
   gi.wrtErr = 0;               //#Events with write-error

   gi.runOpen = 0;              //#Runs opened
   gi.runClose = 0;             //#Runs closed
   gi.runErr = 0;               //#Runs with open/close errors

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



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, numokx, dest, evID, i, k;
   int actBoards = 0, minLen;
   int32_t jrd;
   uint gi_SecTime;             //time in seconds
   int boardId, roi[9], drs, px, src, pixS, pixH, pixC, pixR, tmS;

   int goodhed = 0;

   char str[MXSTR];
   int sockDef[NBOARDS];        //internal state of sockets
   int jrdx;


   struct timespec xwait;


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


   head_len = sizeof (PEVNT_HEADER);
   frst_len = head_len;         //max #bytes to read first: fad_header only, so each event must be longer, even for roi=0
   minLen = head_len;           //min #bytes needed to check header: full header for debug

   start.S = 0xFB01;
   stop.S = 0x04FE;

/* initialize run control logics */
   for (i = 0; i < MAX_RUN; i++) {
      runCtrl[i].runId = 0;
      runCtrl[i].fileId = -2;
      runCtrl[i].procId = -2;
   }
   gi_resetS = gi_resetR = 9;
   for (i = 0; i < NBOARDS; i++)
      sockDef[i] = 0;

   gi_actRun = 0;


 START:
   gettimeofday (tv, NULL);
   g_actTime = tsec = atv.tv_sec;
   g_actUsec = tusec = atv.tv_usec;
   gi_myRun = g_actTime;
   evtCtrl.frstPtr = 0;
   evtCtrl.lastPtr = 0;

   gi_SecTime = g_actTime;
   gi_runStat = g_runStat;
   gj.readStat = g_runStat;
   numok = numok2 = 0;

   int cntsock = 8 - NUMSOCK;

   if (gi_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 (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;
      }

   }


   if (gi_resetR > 0) {
      resetEvtStat ();
      gj.bufTot = gj.maxEvt = gj.xxxEvt = 0;
      gj.usdMem = gj.maxMem = gj.xxxMem = 0;
      gj.totMem = g_maxMem;
      gj.bufNew = gj.bufEvt = 0;
      gj.badRoiE = gj.badRoiR = gj.badRoiB =
         gj.evtSkip = gj.evtWrite = gj.evtErr = 0;

      int b, p;
      for (b = 0; b < NBOARDS; b++)
         gj.badRoi[b] = 0;

      mBufInit ();              //initialize buffers

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


   gi_reset = gi_resetR = gi_resetS = gi_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;
      gettimeofday (tv, NULL);
      g_actTime = tsec = atv.tv_sec;
      g_actUsec = tusec = atv.tv_usec;


      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
            gi.numConn[b] = 0;
            gj.numConn[b] = 0;
            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++) {
               if (p < NUMSOCK)
                  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++;
         }
      }


      jrdx = 0;
      numokx = 0;
      numok = 0;                //count number of succesfull actions

      for (i = 0; i < MAX_SOCK; i++) {  //check all sockets if something to read
         b = i / 7;
         p = i % 7;

         if (p >= NUMSOCK) {;
         } else {
            if (sockDef[b] > 0)
               s0 = +1;
            else
               s0 = -1;

            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 = MAX_LEN;    //huge for skipping
                  }
                  rd[i].bufPos = 0;     //  no byte read so far
                  rd[i].skip = 0;       //  start empty
                  gi_NumConnect[b] += cntsock;

                  gi.numConn[b]++;
                  gj.numConn[b]++;
                  snprintf (str, MXSTR, "+++connect %d %d", b, gi.numConn[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++;
                  size_t maxread = rd[i].bufLen;
                  if (maxread > MAXREAD)
                     maxread = MAXREAD;

                  jrd =
                     recv (rd[i].socket, &rd[i].rBuf->B[rd[i].bufPos],
                           maxread, MSG_DONTWAIT);

                  if (jrd > 0) {
                     debugStream (i, &rd[i].rBuf->B[rd[i].bufPos], jrd);
#ifdef EVTDEBUG
                     memcpy (&rd[i].xBuf->B[rd[i].bufPos],
                             &rd[i].rBuf->B[rd[i].bufPos], jrd);
                     snprintf (str, MXSTR,
                               "read sock %3d bytes %5d len %5d first %d %d",
                               i, jrd, rd[i].bufLen,
                               rd[i].rBuf->B[rd[i].bufPos],
                               rd[i].rBuf->B[rd[i].bufPos + 1]);
                     factOut (kDebug, 301, str);
#endif
                  }

                  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.gotErr[b]++;
                     gi_NumConnect[b] -= cntsock;
                     gi.numConn[b]--;
                     gj.numConn[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.gotErr[b]++;
                     } else
                        numok--;        //else nothing waiting to be read
                     jrd = 0;
                  }
               } else {
                  jrd = 0;      //did read nothing as requested
                  snprintf (str, MXSTR, "do not read from socket %d  %d", i,
                            rd[i].bufLen);
                  factOut (kDebug, 301, str);
               }

               gi.gotByte[b] += jrd;
               gj.rateBytes[b] += jrd;

               if (jrd > 0) {
                  numokx++;
                  jrdx += jrd;
               }


               if (rd[i].bufTyp < 0) {  // we are skipping this board ...
                  //         just do nothing
#ifdef EVTDEBUG
                  snprintf (str, MXSTR, "skipping %d bytes on socket %d", jrd,
                            i);
                  factOut (kInfo, 301, str);
#endif

               } 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 - 2] != stop.B[1]) {
                        gi.evtErr++;
                        snprintf (str, MXSTR,
                                  "wrong end of buffer found sock %3d ev %4d len %5d %3d %3d - %3d %3d ",
                                  i, rd[i].fadLen, rd[i].evtID,
                                  rd[i].rBuf->B[0], rd[i].rBuf->B[1],
                                  rd[i].rBuf->B[rd[i].bufPos - 2],
                                  rd[i].rBuf->B[rd[i].bufPos - 1]);
                        factOut (kError, 301, str);
                        goto EndBuf;

#ifdef EVTDEBUG
                     } else {
                        snprintf (str, MXSTR,
                                  "good  end of buffer found sock %3d len %5d %d %d : %d %d - %d %d : %d %d",
                                  i, rd[i].fadLen, rd[i].rBuf->B[0],
                                  rd[i].rBuf->B[1], start.B[1], start.B[0],
                                  rd[i].rBuf->B[rd[i].bufPos - 2],
                                  rd[i].rBuf->B[rd[i].bufPos - 1], stop.B[1],
                                  stop.B[0]);
                        factOut (kDebug, 301, str);
#endif
                     }

                     if (jrd > 0)
                        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
                     int jr;
                     roi[0] = ntohs (rd[i].rBuf->S[head_len / 2 + 2]);
                     for (jr = 0; jr < 9; jr++) {
                        roi[jr] =
                           ntohs (rd[i].rBuf->S[head_len / 2 + 2 +
                                          jr * (roi[0] + 4)]);
                     }
                     //get index into mBuffer for this event (create if needed)

                     int actid;
                     if (g_useFTM > 0)
                        actid = rd[i].evtID;
                     else
                        actid = rd[i].ftmID;

                     evID = mBufEvt (rd[i].evtID, rd[i].runID, roi, i,
                                     rd[i].fadLen, rd[i].ftmTyp, rd[i].ftmID,
                                     rd[i].evtID);

                     if (evID < -1000) {
                        goto EndBuf;    //not usable board/event/run --> skip it
                     }
                     if (evID < 0) {    //no space left, retry later
#ifdef EVTDEBUG
                        if (rd[i].bufLen != 0) {
                           snprintf (str, MXSTR, "something screwed up");
                           factOut (kFatal, 1, str);
                        }
#endif
                        xwait.tv_sec = 0;
                        xwait.tv_nsec = 10000000;       // sleep for ~10 msec
                        nanosleep (&xwait, NULL);
                        goto EndBuf1;   //hope there is free space next round
                     }
                     //we have a valid entry in mBuffer[]; fill it

#ifdef EVTDEBUG
                     int xchk = memcmp (&rd[i].xBuf->B[0], &rd[i].rBuf->B[0],
                                        rd[i].fadLen);
                     if (xchk != 0) {
                        snprintf (str, MXSTR,
                                  "ERROR OVERWRITE %d %d on port %d", xchk,
                                  rd[i].fadLen, i);
                        factOut (kFatal, 1, str);

                        uint iq;
                        for (iq = 0; iq < rd[i].fadLen; iq++) {
                           if (rd[i].rBuf->B[iq] != rd[i].xBuf->B[iq]) {
                              snprintf (str, MXSTR, "ERROR %4d %4d %x %x", i,
                                        iq, rd[i].rBuf->B[iq],
                                        rd[i].xBuf->B[iq]);
                              factOut (kFatal, 1, str);
                           }
                        }
                     }
#endif
                     int qncpy = 0;
                     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);
                        factOut (kWarn, 301, str);
                     }
                     if (mBuffer[evID].board[boardId] != -1) {
                        snprintf (str, MXSTR,
                                  "double board: ev %5d, b %3d, %3d;  len %5d %3d %3d - %3d %3d ",
                                  evID, boardId, i, rd[i].fadLen,
                                  rd[i].rBuf->B[0], rd[i].rBuf->B[1],
                                  rd[i].rBuf->B[rd[i].bufPos - 2],
                                  rd[i].rBuf->B[rd[i].bufPos - 1]);
                        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);
                     qncpy += head_len;

                     src = head_len / 2;
                     for (px = 0; px < 9; px++) {       //different sort in FAD board.....
                        for (drs = 0; drs < 4; drs++) {
                           pixH = ntohs (rd[i].rBuf->S[src++]); // ID
                           pixC = ntohs (rd[i].rBuf->S[src++]); // start-cell
                           pixR = ntohs (rd[i].rBuf->S[src++]); // roi
//here we should check if pixH is correct ....

                           pixS = boardId * 36 + drs * 9 + px;
                           src++;


                           mBuffer[evID].fEvent->StartPix[pixS] = pixC;
                           dest = pixS * roi[0];
                           memcpy (&mBuffer[evID].fEvent->Adc_Data[dest],
                                   &rd[i].rBuf->S[src], roi[0] * 2);
                           qncpy += roi[0] * 2;
                           src += pixR;

                           if (px == 8) {
                              tmS = boardId * 4 + drs;
                              if (pixR > roi[0]) {      //and we have additional TM info
                                 dest = tmS * roi[0] + NPIX * roi[0];
                                 int srcT = src - roi[0];
                                 mBuffer[evID].fEvent->StartTM[tmS] =
                                    (pixC + pixR - roi[0]) % 1024;
                                 memcpy (&mBuffer[evID].fEvent->
                                         Adc_Data[dest], &rd[i].rBuf->S[srcT],
                                         roi[0] * 2);
                                 qncpy += roi[0] * 2;
                              } else {
                                 mBuffer[evID].fEvent->StartTM[tmS] = -1;
                              }
                           }
                        }
                     }          // now we have stored a new board contents into Event structure

                     mBuffer[evID].fEvent->NumBoards++;
                     mBuffer[evID].board[boardId] = boardId;
                     evtCtrl.evtStat[iDx]++;
                     evtCtrl.pcTime[iDx] = g_actTime;

                     if (++mBuffer[evID].nBoard >= actBoards) {
                        int qnrun = 0;
                        if (mBuffer[evID].runNum != actrun) {   // have we already reported first event of this run ???
                           actrun = mBuffer[evID].runNum;
                           int ir;
                           for (ir = 0; ir < MAX_RUN; ir++) {
                              qnrun++;
                              if (runCtrl[ir].runId == actrun) {
                                 if (++runCtrl[ir].lastEvt == 0) {
                                    gotNewRun (actrun, mBuffer[evID].FADhead);
                                    snprintf (str, MXSTR,
                                              "gotNewRun %d (ev %d)",
                                              mBuffer[evID].runNum,
                                              mBuffer[evID].evNum);
                                    factOut (kInfo, 1, str);
                                    break;
                                 }
                              }
                           }
                        }
                        snprintf (str, MXSTR,
                                  "%5d complete event roi %4d roiTM %d cpy %8d %5d",
                                  mBuffer[evID].evNum, roi[0],
                                  roi[8] - roi[0], qncpy, qnrun);
                        factOut (kDebug, -1, str);

                        //complete event read ---> flag for next processing
                        evtCtrl.evtStat[iDx] = 99;
                        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
                        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);
#ifdef EVTDEBUG
                        memcpy (&rd[i].xBuf->B[0], &rd[i].xBuf->B[k],
                                rd[i].bufPos);
#endif
                     }
                     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].ftmTyp = ntohl (rd[i].rBuf->S[5]);
                        rd[i].ftmID = ntohl (rd[i].rBuf->I[3]); //(FTMevt)
                        rd[i].evtID = ntohl (rd[i].rBuf->I[4]); //(FADevt)
                        rd[i].runID = ntohl (rd[i].rBuf->I[11]);
                        rd[i].bufTyp = 1;       //ready to read full record
                        rd[i].bufLen = rd[i].fadLen - rd[i].bufPos;

                        int fadboard = ntohs (rd[i].rBuf->S[12]);
                        int fadcrate = fadboard / 256;
                        fadboard = (fadcrate * 10 + fadboard % 256);
#ifdef EVTDEBUG
                        snprintf (str, MXSTR,
                                  "sk %3d head: %5d %5d %5d %10d %4d %6d", i,
                                  rd[i].fadLen, rd[i].evtID, rd[i].ftmID,
                                  rd[i].runID, fadboard, jrd);
                        factOut (kDebug, 1, str);
#endif

                        if (rd[i].runID == 0)
                           rd[i].runID = gi_myRun;

                        if (rd[i].bufLen <= head_len
                            || rd[i].bufLen > MAX_LEN) {
                           snprintf (str, MXSTR,
                                     "illegal event-length on port %d", i);
                           factOut (kFatal, 881, str);
                           rd[i].bufLen = 100000;       //?
                        }
                        int fadBoard = ntohs (rd[i].rBuf->S[12]);
                        debugHead (i, fadBoard, 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

#ifdef EVTDEBUG
      snprintf (str, MXSTR, "Loop ---- %3d --- %8d", numokx, jrdx);
      factOut (kDebug, -1, str);
#endif

      gi.numRead[numok]++;

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

         gj.bufNew = gj.bufEvt = 0;
         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] < 92) {
               gj.bufNew++;     //incomplete event in Buffer
               if (evtCtrl.evtStat[k0] < 90
                   && evtCtrl.pcTime[k0] < g_actTime - 10) {
                  int id = evtCtrl.evtBuf[k0];
                  int ic,ib,ik,jb;
                  snprintf (str, MXSTR, "%5d skip short evt %8d %8d %2d",
                            mBuffer[id].evNum, evtCtrl.evtBuf[k0], k0,
                            evtCtrl.evtStat[k0]);
                  factOut (kWarn, 601, str);

                  ik=0;
                  for (ib=0; ib<NBOARDS; ib++) {
                     if (ib%10==0) {
                        snprintf (&str[ik], MXSTR, "'");
                        ik++;
                     }
                     jb = mBuffer[id].board[ib];
                     if ( jb<0 ) {
                        snprintf (&str[ik], MXSTR, ".");
                        ik++;
                     } else {
                        snprintf (&str[ik], MXSTR, "%d",jb%10);
                        ik+=1;
                     }
                  }
                  snprintf (&str[ik], MXSTR, "'");
                  factOut (kWarn, 601, str);

                  evtCtrl.evtStat[k0] = 93;     //timeout for incomplete events
                  gi.evtSkp++;
                  gi.evtTot++;
                  gj.evtSkip++;
               }
            } else if (evtCtrl.evtStat[k0] >= 9000      //'delete'
                       || evtCtrl.evtStat[k0] == 0) {   //'useless'

               int id = evtCtrl.evtBuf[k0];
               snprintf (str, MXSTR, "%5d free event buffer, nb=%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
            }

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


         gj.deltaT = 1000;      //temporary, must be improved

         int 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;
         if (gj.maxEvt > gj.xxxEvt)
            gj.xxxEvt = gj.maxEvt;

         factStat (gj);
         factStatNew (gi);
         gj.rateNew = gj.rateWrite = 0;
         gj.maxMem = gj.usdMem;
         gj.maxEvt = gj.bufTot;
         for (b = 0; b < NBOARDS; b++)
            gj.rateBytes[b] = 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 ...


   snprintf (str, MXSTR, "stop reading ... RESET=%d", g_reset);
   factOut (kInfo, -1, str);

   if (g_reset > 0) {
      gi_reset = g_reset;
      gi_resetR = gi_reset % 10;        //shall we stop reading ?
      gi_resetS = (gi_reset / 10) % 10; //shall we close sockets ?
      gi_resetW = (gi_reset / 100) % 10;        //shall we close files ?
      gi_resetX = gi_reset / 1000;      //shall we simply wait resetX seconds ?
      g_reset = 0;
   } else {
      gi_reset = 0;
      if (g_runStat == -1)
         gi_resetR = 1;
      else
         gi_resetR = 7;
      gi_resetS = 7;            //close all sockets
      gi_resetW = 7;            //close all files
      gi_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 (gi_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 open socket   
            if (i % 7 == 0) {
               gi_NumConnect[i / 7] -= cntsock;
               gi.numConn[i / 7]--;
               gj.numConn[i / 7]--;
               sockDef[i / 7] = 0;      //flag ro recreate the sockets ...
               rd[i / 7].sockStat = -1; //and try to open asap
            }
         }
      }
   }


   if (gi_resetR > 0) {
      //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++;
         }
      }

      xwait.tv_sec = 0;
      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 (gi_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];
               snprintf (str, MXSTR, "ev %5d free event buffer, nb=%3d",
                         mBuffer[id].evNum, mBuffer[id].nBoard);
               factOut (kDebug, -1, str);
               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 (gi_reset > 0) {
      if (gi_resetW > 0) {
         CloseRunFile (0, 0, 0);        //ask all Runs to be closed
      }
      if (gi_resetX > 0) {
         xwait.tv_sec = gi_resetX;
         xwait.tv_nsec = 0;
         nanosleep (&xwait, NULL);
      }

      snprintf (str, MXSTR, "Continue read Process ...");
      factOut (kInfo, -1, str);
      gi_reset = 0;
      goto START;
   }



   snprintf (str, MXSTR, "Exit read Process ...");
   factOut (kInfo, -1, str);
   gi_runStat = -99;
   gj.readStat = -99;
   factStat (gj);
   factStatNew (gi);
   return 0;

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


void *
subProc (void *thrid)
{
   char str[MXSTR];
   int threadID, status, numWait, numProc, kd, k1, k0, k, jret;
   struct timespec xwait;

   threadID = (int) thrid;

   snprintf (str, MXSTR, "Starting sub-process-thread %d", threadID);
   factOut (kInfo, -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);

         if (evtCtrl.evtStat[k0] == 1000 + threadID) {
            if (gi_resetR > 1) {        //we are asked to flush buffers asap
               jret = 9100;     //flag to be deleted
            } else {
               int id = evtCtrl.evtBuf[k0];
               jret =
                  subProcEvt (threadID, mBuffer[id].FADhead,
                              mBuffer[id].fEvent, mBuffer[id].mBoard, gi_runPtr );

               if (jret <= threadID) {
                  snprintf (str, MXSTR,
                            "process %d wants to send to process %d",
                            threadID, jret);
                  factOut (kError, -1, str);
                  jret = 5300;
               } else if (jret <= 0)
                  jret = 9200 + threadID;       //flag as 'to be deleted'
               else if (jret >= gi_maxProc)
                  jret = 5200 + threadID;       //flag as 'to be written'
               else
                  jret = 1000 + jret;   //flag for next proces
            }
            evtCtrl.evtStat[k0] = jret;
            numProc++;
         } else if (evtCtrl.evtStat[k0] < 1000 + threadID)
            numWait++;
      }

      if (gj.readStat < -10 && numWait == 0) {  //nothing left to do
         snprintf (str, MXSTR, "Exit subProcessing Process %d", threadID);
         factOut (kInfo, -1, str);
         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);
      }
   }

   snprintf (str, MXSTR, "Ending sub-process-thread %d", threadID);
   factOut (kInfo, -1, str);
   return;
} /*-----------------------------------------------------------------*/


void *
procEvt (void *ptr)
{
/* *** main loop processing file, including SW-trigger */
   int numProc, numWait, iskip;
   int k, status, j;
   struct timespec xwait;
   char str[MXSTR];



   int lastRun = 0;             //usually run from last event still valid

   cpu_set_t mask;
   int cpu = 1;                 //process thread  (will be several in final version)

   snprintf (str, MXSTR, "Starting process-thread with %d subprocess",
             gi_maxProc);
   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(  0 , &mask );  leave for system
// CPU_SET(  1 , &mask );  used by write process
//   CPU_SET (2, &mask);
//   CPU_SET (3, &mask);
//   CPU_SET (4, &mask);
//   CPU_SET (5, &mask);
//   CPU_SET (6, &mask);
// CPU_SET(  7 , &mask );  used by read process
/* 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);
//   }


   pthread_t thread[100];
   int th_ret[100];

   for (k = 0; k < gi_maxProc; k++) {
      th_ret[k] = pthread_create (&thread[k], NULL, subProc, (void *) k);
   }

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

            if (gi_resetR > 1) {        //we are asked to flush buffers asap
               evtCtrl.evtStat[k0] = 9991;
            } else {

               int id = evtCtrl.evtBuf[k0];
               uint32_t irun = mBuffer[id].runNum;
               int32_t ievt = mBuffer[id].evNum;
               iskip = 0 ;
               if (irun != gi_actRun) { // we should open a new run ?
                  if (irun < gi_actRun) { //  very old event; skip it
                     iskip = -1 ;         //  but anyhow checkEvent() will be done
                     snprintf (str, MXSTR,
                               "P event %d from old run???", ievt,irun);
                     factOut (kWarn, 901, str);
                  } else if (gi_actRun != 0) { //    but there is still old one open ?

//  // // // // // //MUST WAIT UNTIL PENDING OLD EVENTS PROCESSED ...
//  loop over all waiting events; and decide what to do ....
                     int q,q0 ;
                     int qcnt=0; 
                     for (q=k1; q < (k1 + kd); q++) {
                        q0 = q % (MAX_EVT * MAX_RUN);
                        if (evtCtrl.evtStat[q0] > 0 
                         && evtCtrl.evtStat[q0] < 9000) {
                           //this event is still active in the queue
                           int qd = evtCtrl.evtBuf[q0];
                           uint32_t qrun = mBuffer[qd].runNum;
                           if (qrun < irun ) {
                              if (evtCtrl.evtStat[q0] < 92) {
                                 //incomplete old event -> reduce timeout 
                                 evtCtrl.pcTime[q0] = 0 ;
                              }
                              qcnt++ ;
                           }
                        }
                     }
                     if (qcnt == 0 ) {
                        runEnd (gi_actRun, gi_runPtr );
                        snprintf (str, MXSTR,
                                  "P run %d ended", gi_actRun);
                        factOut (kWarn, 901, str);
                        gi_actRun = 0;  //there are no events from old runs left
                     }
                  }


                  if (gi_actRun == 0) { //    now we can try to open new run
                     for (j = 0; j < MAX_RUN; j++) {
                        if (runCtrl[j].runId == irun)
                           break;
                     }
                     if (j >= MAX_RUN) {
                        snprintf (str, MXSTR,
                                  "P error: can not find run %d for event %d in %d",
                                  irun, ievt, id);
                        factOut (kFatal, 901, str);
                        j = 0;
                     }

                     if (runCtrl[j].fileId == 0) {
                        snprintf (str, MXSTR,
                                  "P run %d is already open ???", irun);
                        factOut (kWarn, 901, str);
                     } else if (runCtrl[j].fileId > 0) {
                        snprintf (str, MXSTR,
                                  "P run %d is already closed ???", irun);
                        factOut (kWarn, 901, str);
                     } else {   // now open new run

                        actRun.Version = 1;
                        actRun.RunType = -1;    //to be adapted

                        actRun.Nroi = runCtrl[j].roi0;
                        actRun.NroiTM = runCtrl[j].roi8;
                        if (actRun.Nroi == actRun.NroiTM)
                           actRun.NroiTM = 0;
                        actRun.RunTime = runCtrl[j].firstTime;
                        actRun.RunUsec = runCtrl[j].firstTime;
                        actRun.NBoard = NBOARDS;
                        actRun.NPix = NPIX;
                        actRun.NTm = NTMARK;
                        actRun.Nroi = mBuffer[id].nRoi;
                        memcpy (actRun.FADhead, mBuffer[id].FADhead,
                                NBOARDS * sizeof (PEVNT_HEADER));

                        gi_actRun = irun;
                        gi_runPtr =
                           runStart (irun, &actRun, sizeof (actRun));
                        if (gi_runPtr == NULL) {
                           snprintf (str, MXSTR,
                                     "P could not start run %d", irun);
                           factOut (kFatal, 502, str);
                        } else {
                           snprintf (str, MXSTR, "P started new run %d",
                                     irun);
                           factOut (kInfo, -1, str);
                        }

                        runCtrl[j].fileHd =
                           runOpen (irun, &actRun, sizeof (actRun));
                        if (runCtrl[j].fileHd == NULL) {
                           snprintf (str, MXSTR,
                                     "P could not open a file for run %d",
                                     irun);
                           factOut (kError, 502, str);
                           runCtrl[j].fileId = 91;
                           runCtrl[j].procId = 91;
                        } else {
                           snprintf (str, MXSTR,
                                     "P opened new run_file %d evt %d", irun,
                                     ievt);
                           factOut (kInfo, -1, str);
                           runCtrl[j].fileId = 0;
                           runCtrl[j].procId = 0;
                        }
                     }
                  }
               }

               if (irun == gi_actRun) { 
//                now we are sure the run is ready and we can start processing the event
                  id = evtCtrl.evtBuf[k0];
                  int itevt = mBuffer[id].trgNum;
                  int itrg = mBuffer[id].trgTyp;
                  int roi = mBuffer[id].nRoi;
                  int roiTM = mBuffer[id].nRoiTM;

                  //make sure unused pixels/tmarks are cleared to zero
                  if (roiTM == roi)
                     roiTM = 0;
                  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 
                  mBuffer[id].fEvent->Roi = roi;
                  mBuffer[id].fEvent->RoiTM = roiTM;
                  mBuffer[id].fEvent->EventNum = ievt;
                  mBuffer[id].fEvent->TriggerNum = itevt;
                  mBuffer[id].fEvent->TriggerType = itrg;
                  mBuffer[id].fEvent->Errors[0] = mBuffer[id].Errors[0];
                  mBuffer[id].fEvent->Errors[1] = mBuffer[id].Errors[1];
                  mBuffer[id].fEvent->Errors[2] = mBuffer[id].Errors[2];
                  mBuffer[id].fEvent->Errors[3] = mBuffer[id].Errors[3];
                  mBuffer[id].fEvent->SoftTrig = 0;

                  int mboard = -1;
                  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);
                        if (mboard < 0)
                           mboard = ib;
                     }
                  }
                  mBuffer[id].mBoard = mboard; 
                  int i = eventCheck (mBuffer[id].runNum, mBuffer[id].FADhead,
                                      mBuffer[id].fEvent, mboard);
                  gi.procTot++;
                  numProc++;

                  if (i < 0 || iskip < 0 ) {
                     evtCtrl.evtStat[k0] = 9999;   //flag event to be skipped
                     gi.procErr++;
                  } else {
                     evtCtrl.evtStat[k0] = 1000;   //flag event to be processed
                     runCtrl[j].procEvt++;
                     if (runCtrl[j].procId != 0)
                        mBuffer[id].toWrite = -1;  //flag event to be processed but now written     
                     else
                        mBuffer[id].toWrite = 0;
                  }
               }
            }
         } else if (evtCtrl.evtStat[k0] >= 0 && evtCtrl.evtStat[k0] < 90) {
            numWait++;
         }
      }

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

   }

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

   for (k = 0; k < gi_maxProc; k++) {
      pthread_join (thread[k], (void **) &status);
   }

   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] < 1000) {
         evtCtrl.evtStat[k0] = 9800;    //flag event as 'processed'
      }
   }

   gp_runStat = -99;
   gj.procStat = -99;

   return 0;

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

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


   if (runId == 0) {
      for (j = 0; j < MAX_RUN; j++) {
         if (runCtrl[j].fileId == 0) {  //run is open
            runCtrl[j].closeTime = closeTime;
            runCtrl[j].maxEvt = maxEvt;
         }
      }
      return 0;
   }

   for (j = 0; j < MAX_RUN; j++) {
      if (runCtrl[j].runId == runId) {
         if (runCtrl[j].fileId == 0) {  //run is open
            runCtrl[j].closeTime = closeTime;
            runCtrl[j].maxEvt = maxEvt;
            return 0;
         } else if (runCtrl[j].fileId < 0) {    //run not yet opened
            runCtrl[j].closeTime = closeTime;
            runCtrl[j].maxEvt = maxEvt;
            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, i;
   struct timespec xwait;
   char str[MXSTR];

   cpu_set_t mask;
   int cpu = 1;                 //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] > 5000 && evtCtrl.evtStat[k0] < 9900) {

            if (gi_resetR > 1) {        //we must drain the buffer asap
               evtCtrl.evtStat[k0] = 9904;
            } else {


               int id = evtCtrl.evtBuf[k0];
               uint32_t irun = mBuffer[id].runNum;
               int32_t ievt = mBuffer[id].evNum;

               gi.wrtTot++;
               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);
                     gi.wrtErr++;
                  }
                  lastRun = j;
               }

               if (runCtrl[j].fileId < 0) {
                  actRun.Version = 1;
                  actRun.RunType = -1;  //to be adapted

                  actRun.Nroi = runCtrl[j].roi0;
                  actRun.NroiTM = runCtrl[j].roi8;
                  if (actRun.Nroi == actRun.NroiTM)
                     actRun.NroiTM = 0;
                  actRun.RunTime = runCtrl[j].firstTime;
                  actRun.RunUsec = runCtrl[j].firstTime;
                  actRun.NBoard = NBOARDS;
                  actRun.NPix = NPIX;
                  actRun.NTm = NTMARK;
                  actRun.Nroi = mBuffer[id].nRoi;
                  memcpy (actRun.FADhead, mBuffer[id].FADhead,
                          NBOARDS * sizeof (PEVNT_HEADER));

                  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 = 91;
                  } else {
                     snprintf (str, MXSTR, "W opened new run_file %d evt %d",
                               irun, ievt);
                     factOut (kWarn, -1, str);
                     runCtrl[j].fileId = 0;
                  }

               }

               if (runCtrl[j].fileId != 0 || mBuffer[id].toWrite < 0) {
                  if (runCtrl[j].fileId < 0) {
                     snprintf (str, MXSTR,
                               "W never opened file for this run %d", irun);
                     factOut (kError, 123, str);
                  } else if (runCtrl[j].fileId < 100) {
                     snprintf (str, MXSTR, "W.file for this run is closed %d",
                               irun);
                     factOut (kWarn, 123, str);
                     runCtrl[j].fileId += 100;
                  } else {
                     snprintf (str, MXSTR, "W:file for this run is closed %d",
                               irun);
                     factOut (kDebug, 123, str);
                  }
                  evtCtrl.evtStat[k0] = 9903;
                  gi.wrtErr++;
               } else {
                  i = runWrite (runCtrl[j].fileHd, mBuffer[id].fEvent,
                                sizeof (mBuffer[id]));
                  if (i >= 0) {
                     runCtrl[j].lastTime = g_actTime;
                     runCtrl[j].actEvt++;
                     evtCtrl.evtStat[k0] = 9901;
                     snprintf (str, MXSTR,
                               "%5d successfully wrote for run %d id %5d",
                               ievt, irun, k0);
                     factOut (kDebug, 504, str);
                  } else {
                     snprintf (str, MXSTR, "W error writing event for run %d",
                               irun);
                     factOut (kError, 503, str);
                     evtCtrl.evtStat[k0] = 9902;
                     gi.wrtErr++;
                  }

                  if (i < 0
                      || runCtrl[j].lastTime < g_actTime - 300
                      || runCtrl[j].closeTime < g_actTime
                      || runCtrl[j].maxEvt < runCtrl[j].actEvt) {
                     int ii = 0;
                     if (i < 0)
                        ii = 1;
                     else 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;



                     //close run for whatever reason
                     if (runCtrl[j].runId == gi_myRun)
                        gi_myRun = g_actTime;

                     if (runCtrl[j].procId == 0) {
                        runFinish1 (runCtrl[j].runId);
                        runCtrl[j].procId = 92;
                     }

                     runCtrl[j].closeTime = g_actTime - 1;
                     i = runClose (runCtrl[j].fileHd, &runTail[j],
                                   sizeof (runTail[j]));
                     if (i < 0) {
                        snprintf (str, MXSTR, "error closing run %d %d AAA",
                                  runCtrl[j].runId, i);
                        factOut (kError, 503, str);
                        runCtrl[j].fileId = 92;
                     } else {
                        snprintf (str, MXSTR, "W closed run %d AAA %d", irun,
                                  ii);
                        factOut (kInfo, 503, str);
                        runCtrl[j].fileId = 93;
                     }
                  }
               }
            }
         } else if (evtCtrl.evtStat[k0] > 0 && evtCtrl.evtStat[k0] < 9000)
            numWait++;
      }

      //check if we should close a run (mainly when no event pending)
      for (j = 0; j < MAX_RUN; j++) {
         if (runCtrl[j].fileId == 0
             && (runCtrl[j].closeTime < g_actTime
                 || runCtrl[j].lastTime < g_actTime - 300
                 || runCtrl[j].maxEvt <= runCtrl[j].actEvt)) {
            if (runCtrl[j].runId == gi_myRun)
               gi_myRun = g_actTime;
            int ii = 0;
            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;

            if (runCtrl[j].procId == 0) {
               runFinish1 (runCtrl[j].runId);
               runCtrl[j].procId = 92;
            }

            runCtrl[j].closeTime = g_actTime - 1;
            i = runClose (runCtrl[j].fileHd, &runTail[j],
                          sizeof (runTail[j]));
            if (i < 0) {
               snprintf (str, MXSTR, "error closing run %d %d BBB",
                         runCtrl[j].runId, i);
               factOut (kError, 506, str);
               runCtrl[j].fileId = 94;
            } else {
               snprintf (str, MXSTR, "W closed run %d BBB %d",
                         runCtrl[j].runId, ii);
               factOut (kInfo, 507, str);
               runCtrl[j].fileId = 95;
            }
         }
      }

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

   }

   //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) {
         if (runCtrl[j].runId == gi_myRun)
            gi_myRun = g_actTime;

         if (runCtrl[j].procId == 0) {
            runFinish1 (runCtrl[j].runId);
            runCtrl[j].procId = 92;
         }

         runCtrl[j].closeTime = g_actTime - 1;
         i = runClose (runCtrl[j].fileHd, &runTail[j], sizeof (runTail[j]));
         int ii = 0;
         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;
         if (i < 0) {
            snprintf (str, MXSTR, "error closing run %d %d CCC",
                      runCtrl[j].runId, i);
            factOut (kError, 506, str);
            runCtrl[j].fileId = 96;
         } else {
            snprintf (str, MXSTR, "W closed run %d CCC %d", runCtrl[j].runId,
                      ii);
            factOut (kInfo, 507, str);
            runCtrl[j].fileId = 97;
         }
      }

   gw_runStat = -99;
   gj.writStat = -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;

   gi_runStat = gp_runStat = gw_runStat = 0;
   gj.readStat = gj.procStat = gj.writStat = 0;

   snprintf (str, MXSTR, "Starting EventBuilder V15.07 A");
   factOut (kInfo, -1, str);

//initialize run control logics 
   for (i = 0; i < MAX_RUN; i++) {
      runCtrl[i].runId = 0;
      runCtrl[i].fileId = -2;
   }

//prepare for subProcesses
//   gi_maxSize = g_maxSize;
//   if (gi_maxSize < 0)
      gi_maxSize = 0;

   gi_maxProc = g_maxProc;
   if (gi_maxProc <= 0 || gi_maxProc > 90) {
      snprintf (str, MXSTR, "illegal number of processes %d", gi_maxProc);
      factOut (kFatal, 301, str);
      gi_maxProc = 1;
   }
//partially initialize event control logics
   evtCtrl.frstPtr = 0;
   evtCtrl.lastPtr = 0;

//start all threads (more to come) when we are allowed to ....
   while (g_runStat == 0) {
      xwait.tv_sec = 0;
      xwait.tv_nsec = 10000000; // sleep for ~10 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;


#ifdef BILAND
   xwait.tv_sec = 30;;
   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 = 1;;
   xwait.tv_nsec = 0;           // sleep for ~20sec
   nanosleep (&xwait, NULL);

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

   g_runStat = -1;
#endif


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

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















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

#ifdef BILAND

int
subProcEvt (int threadID, PEVNT_HEADER * fadhd, EVENT * event,
            int16_t mboard, void * buffer)
{
   printf ("called subproc %d\n", threadID);
   return threadID + 1;
}




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




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;
   usleep (10000);
   return 1;
}


//{ return 1; } ;

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




int
eventCheck (uint32_t runNr, PEVNT_HEADER * fadhd, EVENT * event, int mboard)
{
   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
factStatNew (EVT_STAT gi)
{
   int i;

//for (i=0;i<MAX_SOCK;i++) {
//   printf("%4d",gi.numRead[i]);
//   if (i%20 == 0 ) printf("\n");
//}
}

void
gotNewRun (int runnr, PEVNT_HEADER * headers)
{
   printf ("got new run %d\n", runnr);
   return;
}

void
factStat (GUI_STAT gj)
{
//  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, int j, 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_maxMem = g_maxMem * 200;   //100MBytes

   g_maxProc = 20;
   g_maxSize = 0;

   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 < 2)
            sprintf (ipStr, "10.0.%d.11", 128);
         else
            sprintf (ipStr, "10.0.%d.11", 131);
//      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;

}
#endif